File: C:/Users/fred/anaconda3/Lib/site-packages/panel/custom.py
from __future__ import annotations
import asyncio
import hashlib
import inspect
import os
import pathlib
import sys
import textwrap
from collections import defaultdict
from functools import partial
from typing import (
TYPE_CHECKING, Any, Callable, ClassVar, Literal, Mapping, Optional,
)
import param
from param.parameterized import ParameterizedMetaclass
from .config import config
from .io.datamodel import construct_data_model
from .io.state import state
from .models import (
AnyWidgetComponent as _BkAnyWidgetComponent,
ReactComponent as _BkReactComponent, ReactiveESM as _BkReactiveESM,
)
from .models.esm import DataEvent, ESMEvent
from .models.reactive_html import DOMEvent
from .pane.base import PaneBase # noqa
from .reactive import ( # noqa
Reactive, ReactiveCustomBase, ReactiveHTML, ReactiveMetaBase,
)
from .util import camel_to_kebab, classproperty
from .util.checks import import_available
from .viewable import ( # noqa
Child, Children, Layoutable, Viewable, Viewer, is_viewable_param,
)
from .widgets.base import WidgetBase # noqa
if TYPE_CHECKING:
from bokeh.document import Document
from bokeh.events import Event
from bokeh.model import Model
from pyviz_comms import Comm
ExportSpec = dict[str, list[str | tuple[str, ...]]]
class PyComponent(Viewable, Layoutable):
'''
The `PyComponent` combines the convenience of `Viewer` components
that allow creating custom components by declaring a `__panel__`
method with the ability of controlling layout and styling
related parameters directly on the class. Internally the
`PyComponent` will forward layout parameters to the underlying
object, which is created lazily on render.
Reference: https://panel.holoviz.org/reference/custom_components/PyComponent.html
```python
import panel as pn
import param
pn.extension()
class CounterButton(pn.custom.PyComponent, pn.widgets.WidgetBase):
value = param.Integer(default=0)
def __panel__(self):
return pn.widgets.Button(
name=self._button_name, on_click=self._on_click
)
def _on_click(self, event):
self.value += 1
@param.depends("value")
def _button_name(self):
return f"count is {self.value}"
CounterButton().servable()
```
'''
def __init__(self, **params):
super().__init__(**params)
self._view__ = None
self._changing__ = {}
self.param.watch(self._sync__view, [p for p in Layoutable.param if p != 'name'])
def _sync__view(self, *events):
if not events or self._view__ is None:
return
target = self._view__ if events[0].obj is self else self
params = {
e.name: e.new for e in events if e.name not in self._changing__
or self._changing__[e.name] is not e.new
}
if not params:
return
try:
self._changing__.update(params)
with param.parameterized._syncing(target, list(params)):
target.param.update(params)
finally:
for p in params:
if p in self._changing__:
del self._changing__[p]
def _cleanup(self, root: Model | None = None) -> None:
if self._view__ is None:
return
super()._cleanup(root)
if root and root.ref['id'] in self._models:
del self._models[root.ref['id']]
self._view__._cleanup(root)
def _create__view(self):
from .pane import panel
from .param import ParamMethod
if hasattr(self.__panel__, "_dinfo"):
view = ParamMethod(self.__panel__, lazy=True)
else:
view = panel(self.__panel__())
self._view__ = view
params = view.param.values()
overrides, sync = {}, {}
for p in Layoutable.param:
if p != 'name' and view.param[p].default != params[p]:
overrides[p] = params[p]
elif p != 'name':
sync[p] = getattr(self, p)
view.param.watch(self._sync__view, [p for p in Layoutable.param if p != 'name'])
self.param.update(overrides)
with param.parameterized._syncing(view, list(sync)):
view.param.update(sync)
def _get_model(
self, doc: Document, root: Optional['Model'] = None,
parent: Optional['Model'] = None, comm: Optional[Comm] = None
) -> 'Model':
if self._view__ is None:
self._create__view()
model = self._view__._get_model(doc, root, parent, comm)
root = model if root is None else root
self._models[root.ref['id']] = (model, parent)
return model
def select(
self, selector: Optional[type | Callable[Viewable, bool]] = None
) -> list[Viewable]:
return super().select(selector) + self._view__.select(selector)
class ReactiveESMMetaclass(ReactiveMetaBase):
def __init__(mcs, name: str, bases: tuple[type, ...], dict_: Mapping[str, Any]):
mcs.__original_doc__ = mcs.__doc__
ParameterizedMetaclass.__init__(mcs, name, bases, dict_)
# Create model with unique name
ReactiveMetaBase._name_counter[name] += 1
model_name = f'{name}{ReactiveMetaBase._name_counter[name]}'
ignored = [p for p in Reactive.param if not issubclass(type(mcs.param[p].owner), ReactiveESMMetaclass)]
mcs._data_model = construct_data_model(
mcs, name=model_name, ignore=ignored
)
class ReactiveESM(ReactiveCustomBase, metaclass=ReactiveESMMetaclass):
'''
The `ReactiveESM` classes allow you to create custom Panel
components using HTML, CSS and/ or Javascript and without the
complexities of Javascript build tools.
A `ReactiveESM` subclass provides bi-directional syncing of its
parameters with arbitrary HTML elements, attributes and
properties. The key part of the subclass is the `_esm`
variable. Use this to define a `render` function as shown in the
example below.
```python
import panel as pn
import param
pn.extension()
class CounterButton(pn.custom.ReactiveESM):
value = param.Integer()
_esm = """
export function render({ model }) {
let btn = document.createElement("button");
btn.innerHTML = `count is ${model.value}`;
btn.addEventListener("click", () => {
model.value += 1
});
model.on('value', () => {
btn.innerHTML = `count is ${model.value}`;
})
return btn
}
"""
CounterButton().servable()
```
'''
_bokeh_model = _BkReactiveESM
_bundle: ClassVar[str | os.PathLike | None] = None
_esm: ClassVar[str | os.PathLike] = ""
# Specifies exports to make available to JS in a bundled file
# 1. Default export: "<export>"
# 2. Import all (`* as`): "*<export>"
# 3. Named export (`{ <export>, ... }`): ("<export>", ...)
_exports__: ClassVar[ExportSpec] = {}
_importmap: ClassVar[dict[Literal['imports', 'scopes'], str]] = {}
__abstract = True
def __init__(self, **params):
super().__init__(**params)
self._watching_esm = False
self._event__callbacks = defaultdict(list)
self._msg__callbacks = []
@classproperty
def _bundle_path(cls) -> os.PathLike | None:
if config.autoreload:
return
try:
mod_path = pathlib.Path(inspect.getfile(cls)).parent
except (OSError, TypeError, ValueError):
if not isinstance(cls._bundle, pathlib.PurePath):
return
if cls._bundle:
bundle = cls._bundle
if isinstance(bundle, pathlib.PurePath):
return bundle
elif bundle.endswith('.js'):
bundle_path = mod_path / bundle
if bundle_path.is_file():
return bundle_path
return
else:
raise ValueError(
'Could not resolve {cls.__name__}._bundle. Ensure '
'you provide either a string with a relative or absolute '
'path or a Path object to a .js file extension.'
)
path = mod_path / f'{cls.__name__}.bundle.js'
if path.is_file():
return path
module = cls.__module__
path = mod_path / f'{module}.bundle.js'
if path.is_file():
return path
elif module in sys.modules:
module = os.path.basename(sys.modules[module].__file__).replace('.py', '')
path = mod_path / f'{module}.bundle.js'
return path if path.is_file() else None
return None
@classmethod
def _esm_path(cls, compiled: bool = True) -> os.PathLike | None:
if compiled:
bundle_path = cls._bundle_path
if bundle_path:
return bundle_path
esm = cls._esm
if isinstance(esm, pathlib.PurePath):
return esm
elif not esm.endswith(('.js', '.jsx', '.ts', '.tsx')):
return
try:
if hasattr(cls, '__path__'):
mod_path = cls.__path__
else:
mod_path = pathlib.Path(inspect.getfile(cls)).parent
esm_path = mod_path / esm
if esm_path.is_file():
return esm_path
except (OSError, TypeError, ValueError):
pass
return
@classmethod
def _render_esm(cls, compiled: bool | Literal['compiling'] = True):
if (esm_path:= cls._esm_path(compiled=compiled is True)):
esm = esm_path.read_text(encoding='utf-8')
else:
esm = cls._esm
esm = textwrap.dedent(esm)
return esm
def _cleanup(self, root: Model | None) -> None:
if root:
ref = root.ref['id']
if ref in self._models:
model, _ = self._models[ref]
for child in model.children:
children = getattr(self, child)
if isinstance(children, Viewable):
children = [children]
if isinstance(children, list):
for child in children:
if isinstance(child, Viewable):
child._cleanup(root)
super()._cleanup(root)
if not self._models and self._watching_esm:
self._watching_esm.set()
if self._watching_esm in state._watch_events:
state._watch_events.remove(self._watching_esm)
self._watching_esm = False
async def _watch_esm(self):
import watchfiles
path = self._esm_path(compiled=False)
async for _ in watchfiles.awatch(path, stop_event=self._watching_esm):
self._update_esm()
def _update_esm(self):
esm = self._render_esm(not config.autoreload)
for ref, (model, _) in self._models.copy().items():
if esm == model.esm:
continue
self._apply_update({}, {'esm': esm}, model, ref)
@property
def _linked_properties(self) -> list[str]:
return [p for p in self._data_model.properties() if p not in ('js_property_callbacks',)]
def _init_params(self) -> dict[str, Any]:
cls = type(self)
ignored = [p for p in Reactive.param if not issubclass(cls.param[p].owner, ReactiveESM)]
params = {
p : getattr(self, p) for p in list(Layoutable.param)
if getattr(self, p) is not None and p != 'name'
}
data_params = {}
for k, v in self.param.values().items():
if (
(k in ignored and k != 'name') or
(((p:= self.param[k]).precedence or 0) < 0) or
is_viewable_param(p)
):
continue
if k in params:
params.pop(k)
data_params[k] = v
bundle_path = self._bundle_path
if bundle_path:
bundle_hash = hashlib.sha256(str(bundle_path).encode('utf-8')).hexdigest()
importmap = {}
else:
bundle_hash = None
importmap = self._process_importmap()
data_props = self._process_param_change(data_params)
params.update({
'bundle': bundle_hash,
'class_name': camel_to_kebab(cls.__name__),
'data': self._data_model(**{p: v for p, v in data_props.items() if p not in ignored}),
'dev': config.autoreload or getattr(self, '_debug', False),
'esm': self._render_esm(not config.autoreload),
'importmap': importmap,
'name': cls.__name__
})
return params
@classmethod
def _process_importmap(cls):
return cls._importmap
def _get_children(self, data_model, doc, root, parent, comm):
children = {}
ref = root.ref['id']
for k, v in self.param.values().items():
p = self.param[k]
if not is_viewable_param(p):
continue
if v is None:
children[k] = None
elif isinstance(v, list):
children[k] = [
sv._models[ref][0] if ref in sv._models else sv._get_model(doc, root, parent, comm)
for sv in v
]
elif ref in v._models:
children[k] = v._models[ref][0]
else:
children[k] = v._get_model(doc, root, parent, comm)
return children
def _setup_autoreload(self):
from .config import config
if not ((config.autoreload or getattr(self, '_debug', False)) and import_available('watchfiles')):
return
super()._setup_autoreload()
if (self._esm_path(compiled=False) and not self._watching_esm):
self._watching_esm = event = asyncio.Event()
state._watch_events.append(event)
state.execute(self._watch_esm)
def _get_model(
self, doc: Document, root: Optional[Model] = None,
parent: Optional[Model] = None, comm: Optional[Comm] = None
) -> Model:
model = self._bokeh_model(**self._get_properties(doc))
root = root or model
children = self._get_children(model.data, doc, root, model, comm)
model.data.update(**children)
model.children = list(children)
self._models[root.ref['id']] = (model, parent)
self._link_props(model.data, self._linked_properties, doc, root, comm)
self._register_events('dom_event', 'data_event', model=model, doc=doc, comm=comm)
self._setup_autoreload()
return model
def _process_event(self, event: 'Event') -> None:
if isinstance(event, DataEvent):
for cb in self._msg__callbacks:
state.execute(partial(cb, event), schedule=False)
state.execute(partial(self._handle_msg, event.data), schedule=False)
return
elif not isinstance(event, DOMEvent):
return
if hasattr(self, f'_handle_{event.node}'):
getattr(self, f'_handle_{event.node}')(event)
for cb in self._event__callbacks.get(event.node, []):
state.execute(partial(cb, event), schedule=False)
def _update_model(
self, events: dict[str, param.parameterized.Event], msg: dict[str, Any],
root: Model, model: Model, doc: Document, comm: Optional[Comm]
) -> None:
model_msg, data_msg = {}, {}
for prop, v in list(msg.items()):
if prop in list(Reactive.param)+['esm', 'importmap']:
model_msg[prop] = v
elif prop in model.children:
continue
else:
data_msg[prop] = v
for name, event in events.items():
if name not in model.children:
continue
new = event.new
old_objects = event.old if isinstance(event.old, list) else [event.old]
for old in old_objects:
if old is None or old is new or (isinstance(new, list) and old in new):
continue
old._cleanup(root)
if any(e in model.children for e in events):
children = self._get_children(model.data, doc, root, model, comm)
data_msg.update(children)
model_msg['children'] = list(children)
self._set_on_model(model_msg, root, model)
self._set_on_model(data_msg, root, model.data)
def _handle_msg(self, data: any) -> None:
"""
Message handler for messages sent from the frontend using the
`model.send_msg` API.
Arguments
---------
data: any
Data received from the frontend.
"""
def _send_msg(self, data: any) -> None:
"""
Sends data to the frontend which can be observed on the frontend
with the `model.on_msg("msg:custom", callback)` API.
Arguments
---------
data: any
Data to send to the frontend.
"""
self._send_event(ESMEvent, data=data)
def on_msg(self, callback: Callable) -> None:
"""
Registers a callback to be executed when a message event
containing arbitrary data is received.
Arguments
---------
event: str
Name of the DOM event to add an event listener to.
callback: callable
A callable which will be given the msg data.
"""
self._msg__callbacks.append(callback)
def on_event(self, event: str, callback: Callable) -> None:
"""
Registers a callback to be executed when the specified DOM
event is triggered.
Arguments
---------
event: str
Name of the DOM event to add an event listener to.
callback: callable
A callable which will be given the DOMEvent object.
"""
self._event__callbacks[event].append(callback)
class JSComponent(ReactiveESM):
'''
The `JSComponent` allows you to create custom Panel components
using Javascript and CSS without the complexities of
Javascript build tools.
A `JSComponent` subclass provides bi-directional syncing of its
parameters with arbitrary HTML elements, attributes and
properties. The key part of the subclass is the `_esm`
variable. Use this to define a `render` function as shown in the
example below.
Reference: https://panel.holoviz.org/reference/custom_components/JSComponent.html
```python
import panel as pn
import param
pn.extension()
class CounterButton(pn.custom.JSComponent):
value = param.Integer()
_esm = """
export function render({ model }) {
let btn = document.createElement("button");
btn.innerHTML = `count is ${model.value}`;
btn.addEventListener("click", () => {
model.value += 1
});
model.on('value', () => {
btn.innerHTML = `count is ${model.value}`;
})
return btn
}
"""
CounterButton().servable()
```
'''
__abstract = True
class ReactComponent(ReactiveESM):
'''
The `ReactComponent` allows you to create custom Panel components
using React without the complexities of Javascript build tools.
A `ReactComponent` subclass provides bi-directional syncing of its
parameters with arbitrary HTML elements, attributes and
properties. The key part of the subclass is the `_esm`
variable. Use this to define a `render` function as shown in the
example below.
Reference: https://panel.holoviz.org/reference/custom_components/ReactComponent.html
```python
import panel as pn
import param
class CounterButton(pn.custom.ReactComponent):
value = param.Integer()
_esm = """
export function render({model}) {
const [value, setValue] = model.useState("value");
return (
<button onClick={e => setValue(value+1)}>
count is {value}
</button>
)
}
"""
CounterButton().servable()
```
'''
__abstract = True
_bokeh_model = _BkReactComponent
_react_version = '18.3.1'
@classproperty
def _exports__(cls) -> ExportSpec:
imports = cls._importmap.get('imports', {})
exports = {
"react": ["*React"],
"react-dom/client": [("createRoot",)]
}
if any('@mui' in v for v in imports.values()):
exports.update({
"@emotion/cache": "createCache",
"@emotion/react": ("CacheProvider",)
})
return exports
@classmethod
def _render_esm(cls, compiled: bool | Literal['compiling'] = True):
esm = super()._render_esm(compiled=compiled)
if compiled == 'compiling':
esm = 'import * as React from "react"\n' + esm
return esm
@classmethod
def _process_importmap(cls):
imports = cls._importmap.get('imports', {})
v_react = cls._react_version
if config.autoreload:
pkg_suffix, path_suffix = '?dev', '&dev'
else:
pkg_suffix = path_suffix = ''
imports_with_deps = {
"react": f"https://esm.sh/react@{v_react}{pkg_suffix}",
"react/": f"https://esm.sh/react@{v_react}{path_suffix}/",
"react-dom": f"https://esm.sh/react-dom@{v_react}?deps=react@{v_react}&external=react",
"react-dom/": f"https://esm.sh/react-dom@{v_react}&deps=react@{v_react}{path_suffix}&external=react/"
}
suffix = f'deps=react@{v_react},react-dom@{v_react}&external=react,react-dom'
if any('@mui' in v for v in imports.values()):
suffix += ',react-is,@emotion/react'
imports_with_deps.update({
"react-is": f"https://esm.sh/react-is@{v_react}&external=react",
"@emotion/cache": f"https://esm.sh/@emotion/cache?deps=react@{v_react},react-dom@{v_react}",
"@emotion/react": f"https://esm.sh/@emotion/react?deps=react@{v_react},react-dom@{v_react}&external=react,react-is",
})
for k, v in imports.items():
if '?' not in v and 'esm.sh' in v:
if v.endswith('/'):
v = f'{v[:-1]}?{suffix}&path=/'
else:
v = f'{v}?{suffix}'
imports_with_deps[k] = v
return {
'imports': imports_with_deps,
'scopes': cls._importmap.get('scopes', {})
}
class AnyWidgetComponent(ReactComponent):
'''
The `AnyWidgetComponent` allows you to create custom Panel components
in the style of an AnyWidget component. Specifically this component
type creates shims that make it possible to reuse AnyWidget ESM code
as is, without having to adapt the callbacks to use Bokeh APIs.
Reference: https://panel.holoviz.org/reference/custom_components/AnyWidgetComponent.html
```python
import param
import panel as pn
pn.extension()
class CounterWidget(pn.custom.AnyWidgetComponent):
_esm = """
function render({ model, el }) {
let count = () => model.get("value");
let btn = document.createElement("button");
btn.innerHTML = `count is ${count()}`;
btn.addEventListener("click", () => {
model.set("value", count() + 1);
model.save_changes();
});
model.on("change:value", () => {
btn.innerHTML = `count is ${count()}`;
});
el.appendChild(btn);
}
export default { render };
"""
value = param.Integer()
CounterWidget().servable()
```
'''
__abstract = True
_bokeh_model = _BkAnyWidgetComponent
def send(self, msg: dict):
"""
Sends a custom event containing the provided message to the frontend.
Arguments
---------
msg: dict
"""
self._send_msg(msg)