Coverage for src/metador_core/widget/__init__.py: 51%
102 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-11-02 09:33 +0000
1"""Interface of Metador widget plugins."""
2from __future__ import annotations
4from abc import ABC, abstractmethod
5from typing import (
6 TYPE_CHECKING,
7 Any,
8 ClassVar,
9 Dict,
10 Iterator,
11 List,
12 Optional,
13 Set,
14 Type,
15)
17from overrides import overrides
18from panel.viewable import Viewable
19from pydantic import Field
20from typing_extensions import Annotated, TypeAlias
22from ..container import MetadorDataset, MetadorNode
23from ..plugin import interface as pg
24from ..plugins import schemas
25from ..schema import MetadataSchema
26from ..schema.plugins import PluginRef
27from ..schema.types import SemVerTuple
28from .server import WidgetServer
31class Widget(ABC):
32 """Base class for metador widgets."""
34 _args: Dict[str, Any]
35 """Additional passed arguments (e.g. from the dashboard)."""
37 _node: MetadorNode
38 """Container node passed to the widget."""
40 _meta: MetadataSchema
41 """Metadata object to be used as the starting point (e.g. widget config)."""
43 _server: WidgetServer
44 """Widget-backing server object to use (e.g. to access container files from frontend)."""
46 Plugin: ClassVar[WidgetPlugin]
48 def __init__(
49 self,
50 node: MetadorNode,
51 schema_name: str = "",
52 schema_version: Optional[SemVerTuple] = None,
53 *,
54 server: Optional[WidgetServer] = None,
55 container_id: Optional[str] = None,
56 metadata: Optional[MetadataSchema] = None,
57 max_width: Optional[int] = None,
58 max_height: Optional[int] = None,
59 keep_ratio: bool = False,
60 ):
61 """Instantiate a widget for a node.
63 If no schema name is provided, the widget will try to pick the first metadata object
64 from the node that is an instance of a supported schema, in the listed order.
66 If no server is provided, a stand-alone server is started (e.g. for use in a notebook).
68 If a metadata object is passed explicitly, it will be used instead of trying to
69 retrieve one from the node.
70 """
71 # NOTE: we restrict the node so that widgets don't try to escape their scope
72 self._node = node.restrict(read_only=True, local_only=True)
73 # NOTE: if no container_id is passed, we assume its jupyter mode (based on the metador UUIDs)
74 self._container_id = container_id or str(node.metador.container_uuid)
76 # if no server passed, we're in Jupyter mode - use standalone
77 srv: WidgetServer
78 if server is not None:
79 srv = server
80 else:
81 from .jupyter.standalone import running, widget_server
83 if not running():
84 raise ValueError(
85 "No widget server passed and standalone server not running!"
86 )
87 srv = widget_server()
88 self._server = srv
90 # setup correct metadata
91 if metadata is not None:
92 if not self.supports_meta(metadata):
93 msg = "Passed metadata is not instance of a supported schema!"
94 raise ValueError(msg)
95 self._meta = metadata
96 else:
97 if not schema_name:
98 for schemaref in self.Plugin.supports:
99 if node.meta.get(schemaref.name):
100 schema_name = schemaref.name
101 break
102 if not schema_name:
103 raise ValueError("The node does not contain any suitable metadata!")
105 if metadata := node.meta.get(schema_name, schema_version):
106 self._meta = metadata
107 else:
108 raise ValueError("The node does not contain '{schema_name}' metadata!")
110 # maximum width and height that can be used (if None, unlimited)
111 self._w: Optional[int] = max_width
112 self._h: Optional[int] = max_height
114 # recalibrate maximum width and height of widgets to preserve ration, if possible + desired
115 can_scale = self._meta.width is not None and self._meta.height is not None
116 if keep_ratio and can_scale:
117 scale_factor = min(
118 self._h / self._meta.height.value, self._w / self._meta.width.value
119 )
120 self._w = int(self._meta.width.value * scale_factor)
121 self._h = int(self._meta.height.value * scale_factor)
123 # widget-specific setup hook
124 self.setup()
126 def file_data(self, node: Optional[MetadorDataset] = None) -> bytes:
127 """Return data at passed dataset node as bytes.
129 If no node passed, will use the widget root node (if it is a dataset).
130 """
131 node = node or self._node
132 if not isinstance(node, MetadorDataset):
133 raise ValueError(
134 f"Passed node {node.name} does not look like a dataset node!"
135 )
136 return node[()].tolist()
138 def file_url(self, node: Optional[MetadorNode] = None) -> str:
139 """Return URL resolving to the data at given node.
141 If no node passed, will use the widget root node (if it is a dataset).
142 """
143 node = node or self._node
144 if not isinstance(node, MetadorDataset):
145 raise ValueError(
146 f"Passed node {node.name} does not look like a dataset node!"
147 )
148 return self._server.file_url_for(self._container_id, node)
150 @classmethod
151 def supports(cls, *schemas: PluginRef) -> bool:
152 """Return whether any (exact) schema is supported (version-compatible) by widget."""
153 return any(
154 any(map(lambda sref: sref.supports(schema), cls.Plugin.supports))
155 for schema in schemas
156 )
158 @classmethod
159 def supports_meta(cls, obj: MetadataSchema) -> bool:
160 """Return whether widget supports the specific metadata object.
162 The passed object is assumed to be of one of the supported schema types.
164 Default implementation will just check that the object is of a supported schema.
166 Override to constrain further (e.g. check field values).
168 This method affects the dashboard widget selection process and is used
169 to check a metadata object if directly passed to `__init__`.
170 """
171 return cls.supports(type(obj).Plugin.ref())
173 def setup(self): # noqa: B027 # implementing it is not mandatory
174 """Check that passed node is valid and do preparations.
176 If multiple supported schemas are listed, case splitting based on the
177 schema type should be done here to minimize logic in the rendering.
179 Everything that instances can reuse, especially if it is computationally
180 expensive, should also be done here.
182 In case the widget is not able to work with the given node and metadata,
183 it will raise a `ValueError`.
184 """
186 @abstractmethod
187 def show(self) -> Viewable:
188 """Return a fresh Panel widget representing the node data and/or metadata.
190 If width and height were provided during initialization, the widget is supposed
191 to fit within these dimensions, not exceed them and if possible, usefully
192 fill up the space.
194 This method assumes that the widget is fully initialized and setup is completed.
195 """
196 raise NotImplementedError
199WIDGET_GROUP_NAME = "widget"
201if TYPE_CHECKING:
202 SchemaPluginRef: TypeAlias = PluginRef
203else:
204 SchemaPluginRef = schemas.PluginRef
207class WidgetPlugin(pg.PluginBase):
208 supports: Annotated[List[SchemaPluginRef], Field(min_items=1)] # type: ignore
209 """Return list of schemas supported by this widget."""
211 primary: bool = True
212 """Return whether the widget is a primary choice.
214 If False, will not be used automatically by dashboard.
215 """
218class PGWidget(pg.PluginGroup[Widget]):
219 """Widget plugin group interface."""
221 class Plugin:
222 name = WIDGET_GROUP_NAME
223 version = (0, 1, 0)
225 requires = [PluginRef(group="plugingroup", name="schema", version=(0, 1, 0))]
227 plugin_class = Widget
228 plugin_info_class = WidgetPlugin
230 @overrides
231 def check_plugin(self, ep_name: str, plugin: Type[Widget]):
232 pg.util.check_implements_method(ep_name, plugin, Widget.show)
234 def plugin_deps(self, plugin) -> Set[PluginRef]:
235 return set(plugin.Plugin.supports)
237 def widgets_for(self, schema: PluginRef) -> Iterator[PluginRef]:
238 """Return widgets that support (a parent of) the given schema."""
239 ws = set()
240 p_path = schemas.parent_path(schema.name, schema.version)
241 for s_ref in reversed(p_path): # in decreasing specifity
242 for w_cls in self.values():
243 if w_cls.supports(s_ref) and s_ref not in ws:
244 w_ref = w_cls.Plugin.ref()
245 ws.add(w_ref)
246 yield w_ref
248 # def supported_schemas(self) -> Set[PluginRef]:
249 # """Return union of all schemas supported by all installed widgets.
251 # This includes registered child schemas (subclasses of supported schemas).
252 # """
253 # supported = set()
254 # for w in self.values():
255 # for sref in w.Plugin.supports:
256 # supported.add(sref)
257 # for cs_ref in schemas.children(sref.name, sref.version):
258 # supported.add(cs_ref)
259 # return supported