Source code for panel.pane.idom

from __future__ import annotations

import asyncio
import shutil
import sys

from functools import partial
from queue import Queue as SyncQueue
from threading import Thread
from typing import TYPE_CHECKING, Optional

from packaging.version import Version

from import push_on_root
from import DIST_DIR, LOCAL_DIST
from import state
from ..models import IDOM as _BkIDOM
from ..util.warnings import deprecated
from .base import PaneBase

    from bokeh.document import Document
    from bokeh.model import Model
    from pyviz_comms import Comm

_IDOM_MIN_VER = "0.23"
_IDOM_MAX_VER = "0.24"

def _spawn_threaded_event_loop(coro):
    loop_q = SyncQueue()

    def run_in_thread():
        loop = asyncio.new_event_loop()

    thread = Thread(target=run_in_thread, daemon=True)

    return loop_q.get()

[docs]class IDOM(PaneBase): """ The `IDOM` pane renders any IDOM component both in the notebook and in a deployed server. IDOM defines an API for defining and controlling interactive HTML components directly from Python. Note that in the notebook the IDOM support for loading external modules relies on Panel’s Jupyter serverextension. Reference: :Example: >>> IDOM(ClickCount, width=300) """ priority = None _updates = True _unpack = True _bokeh_model = _BkIDOM def __init__(self, object=None, **params): from idom import __version__ as idom_version if Version(_IDOM_MIN_VER) > Version(idom_version) >= Version(_IDOM_MAX_VER): raise RuntimeError( f"Expected idom>={_IDOM_MIN_VER},<{_IDOM_MAX_VER}, but found {idom_version}" ) else: deprecated('1.0', 'panel.pane.IDOM', extra='It may be reimplemented as a separate package in future.') super().__init__(object, **params) self._idom_loop = None self._idom_model = {}, 'object') def _update_layout(self, *args): self._idom_model = {} if self._idom_loop is None: return self._setup() def _setup(self): if self.object is None: return from idom.core.component import Component from idom.core.layout import Layout if isinstance(self.object, Layout): self._idom_layout = self.object elif isinstance(self.object, Component): self._idom_layout = Layout(self.object) else: self._idom_layout = Layout(self.object()) self._idom_loop = _spawn_threaded_event_loop(self._idom_layout_render_loop()) def _get_model( self, doc: Document, root: Optional[Model] = None, parent: Optional[Model] = None, comm: Optional[Comm] = None ) -> Model: from idom.config import IDOM_CLIENT_IMPORT_SOURCE_URL from idom.core.layout import LayoutUpdate # let the client determine import source location IDOM_CLIENT_IMPORT_SOURCE_URL.set("./") if comm: url = '/panel_dist/idom/build' else: url = '/'+LOCAL_DIST+'idom/build' if self._idom_loop is None: self._setup() update = LayoutUpdate.create_from({}, self._idom_model) props = self._init_params() model = self._bokeh_model( event=[update.path, update.changes], importSourceUrl=url, **props ) if root is None: root = model self._link_props(model, ['msg'], doc, root, comm) if root is None: root = model self._models[root.ref['id']] = (model, parent) return model def _cleanup(self, root: Model | None = None) -> None: super()._cleanup(root) if not self._models: # Clean up loop when no views are shown try: self._idom_loop.stop() finally: self._idom_loop = None self._idom_layout = None def _process_property_change(self, msg): if msg['msg'] is None: return {} from idom.core.layout import LayoutEvent dispatch = self._idom_layout.dispatch(LayoutEvent(**msg['msg'])) asyncio.run_coroutine_threadsafe(dispatch, loop=self._idom_loop) for ref, (m, _) in self._models.items(): m.msg = None push_on_root(ref) return {} async def _idom_layout_render_loop(self): async with self._idom_layout: while True: update = await self._idom_layout.render() self._idom_model = update.apply_to(self._idom_model) for ref, (model, _) in self._models.items(): doc = state._views[ref][2] if doc.session_context: doc.add_next_tick_callback(partial(model.update, event=update)) else: model.event = update push_on_root(ref)
[docs] @classmethod def applies(self, object): if object is None: return None elif 'idom' in sys.modules: from idom.core.component import Component from idom.core.layout import Layout if isinstance(object, (Component, Layout)): return 0.8 elif callable(object): return None return False
[docs] @classmethod def install(cls, packages, ignore_installed=False, fallback=None): """ Installs specified packages into application directory. Arguments --------- packages: list or tuple The packages to install from npm ignored_installed: boolean Whether to ignore if the package was previously installed. fallback: str or idom.component The fallback to display while the component is loading """ import idom from idom.config import IDOM_CLIENT_BUILD_DIR idom_dist_dir = DIST_DIR / "idom" idom_build_dir = idom_dist_dir / "build" if not idom_build_dir.is_dir(): idom_build_dir.mkdir() shutil.copyfile(idom_dist_dir / 'package.json', idom_build_dir / 'package.json') if IDOM_CLIENT_BUILD_DIR.get() != idom_build_dir: IDOM_CLIENT_BUILD_DIR.set(idom_build_dir) # just in case packages were already installed but the build hasn't been # copied over to DIST_DIR yet. ignore_installed = True return idom.install(packages, ignore_installed, fallback)
[docs] @classmethod def use_param(cls, parameter): """ Links parameter to some IDOM state value and returns the linked value. Arguments --------- parameter: param.Parameter The parameter to link to a idom state value. Returns ------- An idom state value which is updated when the parameter changes. """ import idom from ..depends import param_value_if_widget parameter = param_value_if_widget(parameter) initial = getattr(parameter.owner, value, set_value = idom.hooks.use_state(initial) def update(event): set_value(, return value