Coverage for src/metador_core/widget/dashboard.py: 40%
145 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-11-02 09:33 +0000
1"""Generic container dashboard.
3To **configure** a container dashboard: attach `DashboardConf` metadata to `MetadorContainer` nodes.
5To **show** a container dashboard: create a `Dashboard` instance.
6"""
7from __future__ import annotations
9from functools import partial
10from itertools import groupby
11from typing import TYPE_CHECKING, Dict, Iterable, List, Optional, Tuple
13import panel as pn
14from panel.viewable import Viewable
15from phantom.interval import Inclusive
17from ..container import MetadorContainer, MetadorNode
18from ..plugins import schemas, widgets
19from ..schema import MetadataSchema
20from ..schema.plugins import PluginRef
21from ..schema.types import NonEmptyStr, SemVerTuple
23if TYPE_CHECKING:
24 from . import WidgetServer
27class DashboardPriority(int, Inclusive, low=1, high=10):
28 """Dashboard priority of a widget."""
31class DashboardGroup(int, Inclusive, low=1):
32 """Dashboard group of a widget."""
35class WidgetConf(MetadataSchema):
36 """Configuration of a widget in the dashboard."""
38 priority: Optional[DashboardPriority] = DashboardPriority(1)
39 """Priority of the widget (1-10), higher priority nodes are shown first."""
41 group: Optional[DashboardGroup]
42 """Dashboard group of the widget.
44 Groups are presented in ascending order.
45 Widgets are ordered by priority within a group.
46 All widgets in a group are shown in a single row.
48 Widgets without an assigned group come last.
49 """
51 # ----
53 schema_name: Optional[NonEmptyStr]
54 """Name of schema of an metadata object at the current node that is to be visualized.
56 If not given, any suitable will be selected if possible.
57 """
59 schema_version: Optional[SemVerTuple]
60 """Version of schema to be used.
62 If not given, any suitable will be selected if possible.
63 """
65 widget_name: Optional[str]
66 """Name of widget to be used.
68 If not given, any suitable will be selected if possible.
69 """
71 widget_version: Optional[SemVerTuple]
72 """Version of widget to be used.
74 If not given, any suitable will be selected if possible.
75 """
78class DashboardConf(MetadataSchema):
79 """Schema describing dashboard configuration for a node in a container.
81 Instantiating without passing a list of widget configurations will
82 return an instance that will show an arbitrary suitable widget, i.e.
83 is equivalent to `DashboardConf.show()`
84 """
86 class Plugin:
87 name = "core.dashboard"
88 version = (0, 1, 0)
90 widgets: List[WidgetConf] = [WidgetConf()]
91 """Widgets to present for this node in the dashboard.
93 If left empty, will try present any widget usable for this node.
94 """
96 @staticmethod
97 def widget(**kwargs) -> WidgetConf:
98 """Construct a dashboard widget configuration (see `WidgetConf`)."""
99 # for convenience
100 return WidgetConf(**kwargs)
102 @classmethod
103 def show(cls, _arg: List[WidgetConf] = None, **kwargs):
104 """Construct a dashboard configuration for the widget(s) of one container node.
106 For one widget, pass the widget config (if any) as keyword arguments,
107 e.g. `DashboardConf.show(group=1)`.
109 For multiple widgets, create widget configurations with `widget(...)`,
110 and pass them to `show`, e.g.:
111 `DashboardConf.show([DashboardConf.widget(), DashboardConf.widget(group=2)])`.
112 """
113 if _arg and kwargs:
114 msg = "Pass widget config arguments or list of widget configs - not both!"
115 raise ValueError(msg)
117 if _arg is None:
118 # kwargs have config for a singleton widget
119 widgets = [cls.widget(**kwargs)]
120 else:
121 # multiple widgets, preconfigured
122 widgets = list(_arg)
123 return cls(widgets=widgets)
126# ----
128NodeWidgetPair = Tuple[MetadorNode, WidgetConf]
129"""A container node paired up with a widget configuration."""
131NodeWidgetRow = List[NodeWidgetPair]
132"""Sorted list of NodeWidgetPairs.
134Ordered first by descending priority, then by ascending node path.
135"""
138def sorted_widgets(
139 widgets: Iterable[NodeWidgetPair],
140) -> Tuple[Dict[int, NodeWidgetRow], NodeWidgetRow]:
141 """Return widgets in groups, ordered by priority and node path.
143 Returns tuple with dict of groups and a remainder of ungrouped widgets.
144 """
146 def nwp_group(tup: NodeWidgetPair) -> int:
147 return tup[1].group or 0
149 def nwp_prio(tup: NodeWidgetPair) -> int:
150 return -tup[1].priority or 0 # in descending order of priority
152 def sorted_group(ws: Iterable[NodeWidgetPair]) -> NodeWidgetRow:
153 """Sort first on priority, and for same priority on container node."""
154 return list(sorted(sorted(ws, key=lambda x: x[0].name), key=nwp_prio))
156 # dict, sorted in ascending group order (but ungrouped are 0)
157 ret = dict(
158 sorted(
159 {
160 k: sorted_group(v)
161 for k, v in groupby(sorted(widgets, key=nwp_group), key=nwp_group)
162 }.items()
163 )
164 )
165 ungrp = ret.pop(0, []) # separate out the ungrouped (were mapped to 0)
166 return ret, ungrp
169# ----
172def _resolve_schema(node: MetadorNode, wmeta: WidgetConf) -> PluginRef:
173 """Return usable schema+version pair for the node based on widget metadata.
175 If a schema name or schema version is missing, will complete these values.
177 Usable schema means that:
178 * there exists a compatible installed schema
179 * there exists a compatible metadata object at given node
181 Raises ValueError on failure to find a suitable schema.
182 """
183 if wmeta.schema_name is None:
184 # if no schema selected -> pick any schema for which we have:
185 # * a schema instance at the current node
186 # * installed widget(s) that support it
187 for obj_schema in node.meta.query():
188 if next(widgets.widgets_for(obj_schema), None):
189 return obj_schema
191 msg = f"Cannot find suitable schema for a widget at node: {node.name}"
192 raise ValueError(msg)
194 # check that a node object is compatible with the one requested
195 req_ver = wmeta.schema_version if wmeta.schema_version else "any"
196 req_schema = f"{wmeta.schema_name} ({req_ver})"
197 s_ref = next(node.meta.query(wmeta.schema_name, wmeta.schema_version), None)
198 if s_ref is None:
199 msg = f"Dashboard wants metadata compatible with {req_schema}, but node"
200 if nrf := next(node.meta.query(wmeta.schema_name), None):
201 nobj_schema = f"{nrf.name} {nrf.version}"
202 msg += f"only has incompatible object: {nobj_schema}"
203 else:
204 msg += "has no suitable object"
205 raise ValueError(msg)
207 # if no version is specified, pick the one actually present at the node
208 version = wmeta.schema_version or s_ref.version
209 s_ref = schemas.PluginRef(name=wmeta.schema_name, version=version)
211 # ensure there is an installed schema compatible with the one requested
212 # (NOTE: if child schemas exist, the parents do too - no need to check)
213 installed_schema = schemas.get(s_ref.name, s_ref.version)
214 if installed_schema is None:
215 msg = f"No installed schema is compatible with {req_schema}"
216 raise ValueError(msg)
218 return s_ref
221def _widget_suitable_for(m_obj: MetadataSchema, w_ref: PluginRef) -> bool:
222 """Granular check whether the widget actually works with the metadata object.
224 Assumes that the passed object is known to be one of the supported schemas.
225 """
226 w_cls = widgets._get_unsafe(w_ref.name, w_ref.version)
227 return w_cls.Plugin.primary and w_cls.supports_meta(m_obj)
230def _resolve_widget(
231 node: MetadorNode,
232 s_ref: PluginRef,
233 w_name: Optional[str],
234 w_version: Optional[SemVerTuple],
235) -> PluginRef:
236 """Return suitable widget for the node based on given dashboard metadata."""
237 if w_name is None:
238 # get candidate widgets in alphabetic order (all that claim to work with schema)
239 cand_widgets = sorted(widgets.widgets_for(s_ref))
240 is_suitable = partial(_widget_suitable_for, node.meta[s_ref.name])
241 # filter out the ones that ACTUALLY can handle the object and are eligible
242 if w_ref := next(filter(is_suitable, cand_widgets), None):
243 return w_ref
244 else:
245 msg = f"Could not find suitable widget for {w_name} at node {node.name}"
246 raise ValueError(msg)
248 # now we have a widget name (and possibly version) - check it
249 widget_class = widgets._get_unsafe(w_name, w_version)
250 if widget_class is None:
251 raise ValueError(f"Could not find compatible widget: {w_name} {w_version}")
252 if not widget_class.supports(*schemas.parent_path(s_ref.name, s_ref.version)):
253 msg = f"Widget {widget_class.Plugin.ref()} does not support {s_ref}"
254 raise ValueError(msg)
256 w_ref = widget_class.Plugin.ref()
257 return w_ref
260# ----
263def get_grp_label(idx):
264 """Create and return a styled group label."""
265 return pn.pane.Str(
266 f"Group {idx+1}" if idx is not None else "Ungrouped resources",
267 style={
268 "font-size": "15px",
269 "font-weight": "bold",
270 "text-decoration": "underline",
271 },
272 )
275def add_widgets(w_grp, ui_grp, *, server=None, container_id: Optional[str] = None):
276 """Instantiate and add widget to the flexibly wrapping row that handles the entire group."""
277 w_width, w_height = 500, 500 # max size of a widget tile, arbitrarily set
278 for node, wmeta in w_grp:
279 w_cls = widgets.get(wmeta.widget_name, wmeta.widget_version)
280 label = pn.pane.Str(f"{node.name}:")
282 # instantiating the appropriate widget
283 w_obj = w_cls(
284 node,
285 wmeta.schema_name,
286 wmeta.schema_version,
287 server=server,
288 container_id=container_id,
289 # reset max widget of a widget tile, only if it is for a pdf, text or video file
290 max_width=700
291 if "pdf" in wmeta.widget_name
292 or "text" in wmeta.widget_name
293 or "video" in wmeta.widget_name
294 else w_width,
295 # reset max height of a widget tile, only if it is for a text file
296 max_height=700 if "text" in wmeta.widget_name else w_height,
297 )
299 # adding the new widget to the given row
300 ui_grp.append(
301 pn.Column(
302 label,
303 w_obj.show(),
304 sizing_mode="scale_both",
305 scroll=False
306 if "image" in wmeta.widget_name or "pdf" in wmeta.widget_name
307 else True,
308 )
309 )
310 return ui_grp
313def get_grp_row(
314 *,
315 idx=None,
316 widget_group=None,
317 divider=False,
318 server=None,
319 container_id: Optional[str] = None,
320):
321 """Create a flexible and wrapping row for all widgets within a single group."""
322 return pn.FlexBox(
323 get_grp_label(idx=idx),
324 add_widgets(
325 widget_group,
326 pn.FlexBox(
327 flex_direction="row",
328 justify_content="space-evenly",
329 align_content="space-evenly",
330 align_items="center",
331 sizing_mode="scale_both",
332 ),
333 server=server,
334 container_id=container_id,
335 ),
336 pn.layout.Divider(margin=(100, 0, 20, 0)) if divider else None,
337 flex_direction="column",
338 justify_content="space-evenly",
339 align_content="space-evenly",
340 align_items="center",
341 sizing_mode="scale_both",
342 )
345class Dashboard:
346 """The dashboard presents a view of all marked nodes in a container.
348 To be included in the dashboard, a node must be marked by a `DashboardConf`
349 object configuring at least one widget for that node.
352 Note that the `Dashboard` needs
353 * either a widget server to be passed (embedding in a website),
354 * or the container is wrapped by `metador_core.widget.jupyter.Previewable` (notebook mode)
355 """
357 def __init__(
358 self,
359 container: MetadorContainer,
360 *,
361 server: WidgetServer = None,
362 container_id: Optional[str] = None,
363 ):
364 """Return instance of a dashboard.
366 Args:
367 container: Actual Metador container that is open and readable
368 server: `WidgetServer` to use for the widgets (default: standalone server / Jupyter mode)
369 container_id: Container id usable with the server to get this container (default: container UUID)
370 """
371 self._container: MetadorContainer = container
372 self._server = server
373 self._container_id: str = container_id
375 # figure out what schemas to show and what widgets to use and collect
376 ws: List[NodeWidgetPair] = []
377 for node in self._container.metador.query(DashboardConf):
378 dbmeta = node.meta.get(DashboardConf)
379 restr_node = node.restrict(read_only=True, local_only=True)
380 for wmeta in dbmeta.widgets:
381 ws.append((restr_node, self._resolve_node(node, wmeta)))
383 grps, ungrp = sorted_widgets(ws)
384 self._groups = grps
385 self._ungrouped = ungrp
387 def _resolve_node(self, node: MetadorNode, wmeta: WidgetConf) -> WidgetConf:
388 """Check and resolve widget dashboard metadata for a node."""
389 wmeta = wmeta.copy() # use copy, abandon original
391 s_ref: PluginRef = _resolve_schema(node, wmeta)
392 wmeta.schema_name = s_ref.name
393 wmeta.schema_version = s_ref.version
395 w_ref: PluginRef = _resolve_widget(
396 node, s_ref, wmeta.widget_name, wmeta.widget_version
397 )
398 wmeta.widget_name = w_ref.name
399 wmeta.widget_version = w_ref.version
401 return wmeta
403 def show(self) -> Viewable:
404 """Instantiate widgets for container and return resulting dashboard."""
405 # Outermost element: The Dashboard is a column of widget groups
406 db = pn.FlexBox(
407 flex_direction="column",
408 justify_content="space-evenly",
409 align_content="space-evenly",
410 align_items="center",
411 sizing_mode="scale_both",
412 )
414 # add each widget group within individual, flexibly-wrapping rows
415 for idx, widget_group in enumerate(self._groups.values()):
416 db.append(
417 get_grp_row(
418 idx=idx,
419 widget_group=widget_group,
420 divider=False, # does not work offline with panel >= 1.0?
421 server=self._server,
422 container_id=self._container_id,
423 )
424 )
426 # dump remaining ungrouped widgets into a separate flexibly-wrapping row
427 ungrp_exist = len(self._ungrouped) != 0
428 if ungrp_exist:
429 db.append(
430 get_grp_row(
431 widget_group=self._ungrouped,
432 server=self._server,
433 container_id=self._container_id,
434 )
435 )
436 return db