diff --git a/README.md b/README.md index 00e9355..10339b0 100644 --- a/README.md +++ b/README.md @@ -138,6 +138,80 @@ See [examples/datatool_dashboard.py](examples/datatool_dashboard.py) for a full To regenerate the screenshots after UI changes, see [docs/generate_screenshots.py](docs/generate_screenshots.py). +## ProcessObjectView (Process/Object Dashboard UI) + +Where `DataToolView` is centered on data tools, `ProcessObjectView` is centered +on the **objects** (samples, specimens, ... - any `Item`) that pass through +processes. It answers *"how did measurement X compare across the runs my objects +went through?"* by overlaying repeated runs on a common, time-normalized axis. + +![Process Demo](docs/process_demo.gif) + +*Pick objects (tree 1) and a channel under a process type (tree 2); each run is +overlaid from t=0.* + +Two trees drive the plot: + +1. **Objects** - the `Item` instances to compare. +2. **Process types -> channels** - for each process type the objects went + through, the channels of the data tools used in those runs. + +Tree 2 is **aggregated for selection only** (tick once instead of ticking the +same channel on every run and tool). The aggregation is **co-presence aware**: + +- data tools of the same type that run **together** in a process stay as + **separate** entries (distinct measurement points); +- data tools that only ever appear in **different** runs are treated as drop-in + replacements and **merged** into one `… [n channels]` entry (the actual + channels are listed in its tooltip). + +![Process trees](docs/screenshot_process_trees.png) + +*Evacuation ran both probes together (separate per-instance entries); Heating +swapped the probe between runs (merged `DataTool/… [2 channels]` entries).* + +Selecting an object + a channel entry plots **every real channel that object has +data on** - fanning out across process runs and co-present tools. Each line is +normalized to its own run (first data point at t=0, x-axis in seconds), grouped +by characteristic, and gets a distinct legend entry +(`object / process / data tool / channel`). + +![Process overlay](docs/screenshot_process_overlay.png) + +*One channel selection fans out to two runs of the same sample, overlaid from +t=0 for comparison.* + +![Heating drop-in](docs/screenshot_process_heating.png) + +*Adding `Heating/DataTool/temp` and `…/pressure` on top of the Evacuation +selection overlays both processes, grouped into separate Temperature and Pressure +plots; the merged Heating entries resolve to probe A for Sample 1 and probe B for +Sample 2 (drop-in replacements compared across objects).* + +```python +from opensemantic.base.view import ProcessObjectView +from opensemantic.base.view._config import DashboardConfig + +view = ProcessObjectView( + objects=objects, # list[Item] + processes=processes, # list[Process] (filtered to those with start+end + # time and >=1 DataTool whose data you can load) + controllers=controllers, # list[DataToolController], matched to process tools + config=DashboardConfig(lang="en"), + title="Process / Object Archive View", +) +view.servable() # for panel serve +``` + +`DataToolView` and `ProcessObjectView` share their plot/unit/config machinery via +`BaseDataView`, and both support `embeddable=True` (exposing `sidebar_cards` / +`main_cards`) so a host app can combine them. + +See [examples/process_dashboard.py](examples/process_dashboard.py) for a full +working example, and +[docs/generate_process_screenshots.py](docs/generate_process_screenshots.py) to +regenerate these screenshots. + ## Installation ```bash diff --git a/docs/generate_process_screenshots.py b/docs/generate_process_screenshots.py new file mode 100644 index 0000000..ae484bd --- /dev/null +++ b/docs/generate_process_screenshots.py @@ -0,0 +1,217 @@ +"""Generate screenshots and demo GIF for the ProcessObjectView README. + +Prerequisites: + pip install playwright imageio + playwright install chromium + +Usage: + python docs/generate_process_screenshots.py +""" + +import glob +import io +import os +import subprocess +import sys +import time + +import imageio.v3 as iio +from playwright.sync_api import sync_playwright + +DOCS_DIR = os.path.dirname(os.path.abspath(__file__)) +PACKAGE_DIR = os.path.dirname(DOCS_DIR) +EXAMPLE = os.path.join(PACKAGE_DIR, "examples", "process_dashboard.py") +PORT = 5012 +URL = f"http://localhost:{PORT}/process_dashboard" +VIEWPORT = {"width": 1400, "height": 900} + + +def all_wb_shadows_js(): + """JS that returns all Wunderbaum shadow roots in DOM order.""" + return """ + function allWbShadows(root) { + const out = []; + function rec(r) { + for (const el of r.querySelectorAll('*')) { + if (el.shadowRoot) { + if (el.shadowRoot.querySelectorAll('.wb-row').length > 0) + out.push(el.shadowRoot); + rec(el.shadowRoot); + } + } + } + rec(root); + return out; + } + """ + + +def click_tree_checkbox(page, tree_idx, cb_idx): + """Click the cb_idx-th checkbox of the tree_idx-th Wunderbaum (0=objects).""" + page.evaluate( + f"""() => {{ + {all_wb_shadows_js()} + const shadows = allWbShadows(document); + const sh = shadows[{tree_idx}]; + if (sh) {{ + const cbs = sh.querySelectorAll('i.wb-checkbox'); + if (cbs[{cb_idx}]) cbs[{cb_idx}].click(); + }} + }}""" + ) + + +def switch_temperature_unit(page): + """Switch the Temperature unit dropdown to Celsius.""" + page.evaluate( + """() => { + function findAll(root) { + const sels = root.querySelectorAll('select.bk-input'); + for (const s of sels) { + const label = s.closest('.bk-input-group') + ?.querySelector('label')?.textContent || ''; + if (label.includes('Temperature')) { + for (let i = 0; i < s.options.length; i++) { + const t = s.options[i].text; + if (t.includes('C') && !t.includes('K') && t.length < 5) { + s.value = s.options[i].value; + s.dispatchEvent(new Event('change', {bubbles: true})); + return true; + } + } + } + } + for (const el of root.querySelectorAll('*')) { + if (el.shadowRoot) { + const r = findAll(el.shadowRoot); + if (r) return r; + } + } + return false; + } + return findAll(document); + }""" + ) + + +def capture(page, frames, delay=500): + page.wait_for_timeout(delay) + buf = page.screenshot() + frames.append(iio.imread(io.BytesIO(buf))) + + +def start_server(): + """Start the Panel server, cleaning old archive DBs for fresh data.""" + for db in glob.glob(os.path.join(PACKAGE_DIR, "*_db.sqlite")): + os.remove(db) + proc = subprocess.Popen( + [sys.executable, "-m", "panel", "serve", EXAMPLE, "--port", str(PORT)], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + cwd=PACKAGE_DIR, + ) + # Let the import + first module execution settle. panel serve re-runs the + # script per session (storing the demo data each time), so the first page + # load is slow - handled by a long goto timeout below. + time.sleep(10) + return proc + + +def stop_server(proc): + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + + +def main(): + print("Starting Panel server...") + proc = start_server() + try: + frames = [] + with sync_playwright() as p: + browser = p.chromium.launch(headless=True) + page = browser.new_page(viewport=VIEWPORT) + # panel serve rebuilds the demo data per session -> slow first load. + page.goto(URL, timeout=120000, wait_until="domcontentloaded") + page.wait_for_timeout(12000) + + # Frame 0-1: initial view (both trees, process tree shows + # per-instance Evacuation entries and merged Heating entries). + capture(page, frames, 800) + capture(page, frames, 500) + tree_shot = len(frames) - 1 + + # Select object "Sample 1" (objects tree, checkbox 0). + click_tree_checkbox(page, 0, 0) + capture(page, frames, 1500) + + # Select Evacuation / FurnaceProbe-A / temp + # (process tree: 0=Evac root, 1=A/temp, 2=B/temp, 3=A/press, ...). + click_tree_checkbox(page, 1, 1) + capture(page, frames, 3000) + overlay_shot = len(frames) - 1 # Sample 1: two Evacuation runs + + # Add FurnaceProbe-B / temp -> co-present fan-out. + click_tree_checkbox(page, 1, 2) + capture(page, frames, 2500) + + # Add object "Sample 2" -> compare across objects. + click_tree_checkbox(page, 0, 1) + capture(page, frames, 3000) + fanout_shot = len(frames) - 1 + + # Switch Temperature unit to Celsius. + switch_temperature_unit(page) + capture(page, frames, 2500) + units_shot = len(frames) - 1 + capture(page, frames, 1000) + + # Also add the Heating process so BOTH processes are selected at the + # end. Evacuation stays selected; the merged Heating entry is a + # drop-in pool resolving to probe A for Sample 1 and probe B for + # Sample 2 (process tree: 6 = Heating DataTool/temp). + click_tree_checkbox(page, 1, 6) # check Heating DataTool/temp + capture(page, frames, 3000) + + # Finally add the Heating pressure channel (process tree: 7) -> a + # second plot group (Pressure) appears alongside Temperature. + click_tree_checkbox(page, 1, 7) # check Heating DataTool/pressure + capture(page, frames, 3000) + heating_shot = len(frames) - 1 + capture(page, frames, 1500) + + browser.close() + + gif_path = os.path.join(DOCS_DIR, "process_demo.gif") + iio.imwrite(gif_path, frames, duration=1500, loop=0) + print(f"process_demo.gif: {len(frames)} frames") + + iio.imwrite( + os.path.join(DOCS_DIR, "screenshot_process_trees.png"), frames[tree_shot] + ) + iio.imwrite( + os.path.join(DOCS_DIR, "screenshot_process_overlay.png"), + frames[overlay_shot], + ) + iio.imwrite( + os.path.join(DOCS_DIR, "screenshot_process_fanout.png"), + frames[fanout_shot], + ) + iio.imwrite( + os.path.join(DOCS_DIR, "screenshot_process_units.png"), + frames[units_shot], + ) + iio.imwrite( + os.path.join(DOCS_DIR, "screenshot_process_heating.png"), + frames[heating_shot], + ) + print("Static screenshots saved") + finally: + print("Stopping server...") + stop_server(proc) + + +if __name__ == "__main__": + main() diff --git a/docs/process_demo.gif b/docs/process_demo.gif new file mode 100644 index 0000000..b179ede Binary files /dev/null and b/docs/process_demo.gif differ diff --git a/docs/screenshot_process_fanout.png b/docs/screenshot_process_fanout.png new file mode 100644 index 0000000..5046c7c Binary files /dev/null and b/docs/screenshot_process_fanout.png differ diff --git a/docs/screenshot_process_heating.png b/docs/screenshot_process_heating.png new file mode 100644 index 0000000..74c72e8 Binary files /dev/null and b/docs/screenshot_process_heating.png differ diff --git a/docs/screenshot_process_overlay.png b/docs/screenshot_process_overlay.png new file mode 100644 index 0000000..05d2b49 Binary files /dev/null and b/docs/screenshot_process_overlay.png differ diff --git a/docs/screenshot_process_trees.png b/docs/screenshot_process_trees.png new file mode 100644 index 0000000..17c5968 Binary files /dev/null and b/docs/screenshot_process_trees.png differ diff --git a/docs/screenshot_process_units.png b/docs/screenshot_process_units.png new file mode 100644 index 0000000..2b78d64 Binary files /dev/null and b/docs/screenshot_process_units.png differ diff --git a/examples/process_dashboard.py b/examples/process_dashboard.py new file mode 100644 index 0000000..f1aa83d --- /dev/null +++ b/examples/process_dashboard.py @@ -0,0 +1,254 @@ +"""Example: process/object-centered archive view (``ProcessObjectView``). + +Where ``DataToolView`` is centered on data tools, this view is centered on the +**objects** that pass through processes (samples, specimens, ...). It answers +"how did measurement X compare across the runs my objects went through?" + +Two trees drive the plot: + +1. **Objects** - the ``Item`` instances you want to compare. +2. **Process types -> channels** - for each type of process the objects went + through, the channels of the data tools used in those runs. + +Tree 2 is *aggregated for selection only*: instead of ticking the same channel +on every run and every tool, you tick it once. The aggregation is co-presence +aware: + +- data tools of the same type that run **together** in a process stay as + **separate** entries (distinct measurement points); +- data tools that only ever appear in **different** runs are treated as drop-in + replacements and **merged** into one entry, shown as + ``/ [n channels]`` with the actual channels in its tooltip. + +Selecting an object + a channel entry then plots **every real channel that +object has data on** for it - fanning out across process runs and across +co-present tools. Each line is time-normalized to its own run (first data point +at t=0, x-axis in seconds) and grouped by characteristic, so repeated runs and +different objects overlay for direct comparison. Every line gets a distinct +legend entry (object / process / data tool / channel). + +This example shows both aggregation cases with two probes of the same type: + +- **Evacuation** runs both probes together -> separate per-instance entries + (``FurnaceProbe-A/temp``, ``FurnaceProbe-B/temp``); +- **Heating** swaps the probe between runs -> one merged entry + (``DataTool/temp [2 channels]``). + +Sample 1 ran Evacuation twice, so selecting it + ``temp`` overlays both runs. + +Run with:: + + panel serve examples/process_dashboard.py --dev + +A commented block at the bottom shows how to populate the same view from an +OpenSemanticLab instance instead of the local data built here. +""" + +import asyncio +import datetime as dt +import random +from typing import Optional +from uuid import NAMESPACE_URL, uuid5 + +import nest_asyncio +import panel as pn + +from opensemantic import compute_scoped_uuid +from opensemantic.base.v1 import ( + Database, + DataChannel, + DataTool, + DataToolController, + Process, +) +from opensemantic.base.view import ProcessObjectView +from opensemantic.base.view._config import DashboardConfig, PlotConfig +from opensemantic.characteristics.quantitative.v1 import ( + ForcePerAreaUnit, + Pressure, + Temperature, + TemperatureUnit, +) +from opensemantic.core.v1 import Item, Label + +pn.extension() + +# -- Process-type marker classes (for readable tree-2 group labels) ---------- +# In a real wiki these are proper Category pages; here we register lightweight +# subclasses in the oold type registry so the view can resolve a nice label. + +EVAC_TYPE = "Category:OSW000000000000000000000000000000e1" +HEAT_TYPE = "Category:OSW000000000000000000000000000000e2" + + +class EvacuationProcess(Process): + class Config: + schema_extra = {"title": "Evacuation"} + + type: Optional[list] = [EVAC_TYPE] + + +class HeatingProcess(Process): + class Config: + schema_extra = {"title": "Heating"} + + type: Optional[list] = [HEAT_TYPE] + + +try: + from oold.model import _types as _types_v2 + from oold.model.v1 import _types as _types_v1 + + for _reg in (_types_v1, _types_v2): + _reg[EVAC_TYPE] = EvacuationProcess + _reg[HEAT_TYPE] = HeatingProcess +except ImportError: + pass + + +# -- Build DataTool probes + controllers ------------------------------------- +# Two probes of the same datatool type (same channel names temp/pressure), so +# their channels collapse to one aggregated entry per channel in the treeview. + + +def _make_probe(name): + u = uuid5(NAMESPACE_URL, name) + tool = DataTool( + uuid=u, + name=name, + label=[Label(text=name)], + data_channels=[ + DataChannel( + uuid=str(compute_scoped_uuid(u, "temp")), + osw_id="placeholder", + name="temp", + label=[Label(text="Temperature")], + characteristic=Temperature.get_cls_iri(), + ), + DataChannel( + uuid=str(compute_scoped_uuid(u, "pressure")), + osw_id="placeholder", + name="pressure", + label=[Label(text="Pressure")], + characteristic=Pressure.get_cls_iri(), + ), + ], + storage_locations=[Database(name=name + "_db", label=[Label(text="db")])], + ) + return DataToolController(tool, auto_archive=True) + + +BASE = dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=3) + + +async def setup(): + probe_a = _make_probe("FurnaceProbe-A") + probe_b = _make_probe("FurnaceProbe-B") + + # Store ~130 min of 1-minute data on both probes so every process window has + # data. Values ramp with absolute time, so different windows differ but + # share a shape (clear after normalizing each run to t=0). + for probe in (probe_a, probe_b): + for minute in range(130): + ts = BASE + dt.timedelta(minutes=minute) + await probe.store_channel_data( + DataToolController.StoreChannelDataParams( + channel="temp", + value=Temperature( + value=20.0 + 0.15 * minute + random.uniform(-0.3, 0.3), + unit=TemperatureUnit.Celsius, + ), + timestamp=ts, + ) + ) + await probe.store_channel_data( + DataToolController.StoreChannelDataParams( + channel="pressure", + value=Pressure( + value=1013.0 - 0.2 * minute + random.uniform(-1, 1), + unit=ForcePerAreaUnit.hecto_pascal, + ), + timestamp=ts, + ) + ) + + return probe_a, probe_b + + +# Panel serve runs inside Tornado's event loop, so use nest_asyncio. +nest_asyncio.apply() +probe_a, probe_b = asyncio.run(setup()) +controllers = [probe_a, probe_b] + +# -- Sample objects and their process runs ----------------------------------- + +sample1 = Item(uuid=uuid5(NAMESPACE_URL, "Sample-1"), label=[Label(text="Sample 1")]) +sample2 = Item(uuid=uuid5(NAMESPACE_URL, "Sample-2"), label=[Label(text="Sample 2")]) +objects = [sample1, sample2] + + +def _run(cls, name, sample, tools, start_min, dur_min=30): + start = BASE + dt.timedelta(minutes=start_min) + return cls( + uuid=uuid5(NAMESPACE_URL, name), + label=[Label(text=name)], + input=[sample], + tool=list(tools), + start_date_time=start, + end_date_time=start + dt.timedelta(minutes=dur_min), + ) + + +# Two contrasting cases: +# - Evacuation always runs BOTH probes together (co-present) -> the tree keeps +# them as separate per-instance entries (FurnaceProbe-A/temp, .../B/temp). +# - Heating swaps the probe between runs (A for Sample 1, B for Sample 2, never +# together) -> the tree merges them into one drop-in entry +# (DataTool/temp [2 channels]) whose tooltip lists the actual channels. +processes = [ + _run(EvacuationProcess, "S1 Evacuation #1", sample1, [probe_a, probe_b], 0), + _run(EvacuationProcess, "S1 Evacuation #2", sample1, [probe_a, probe_b], 40), + _run(HeatingProcess, "S1 Heating", sample1, [probe_a], 80), + _run(EvacuationProcess, "S2 Evacuation", sample2, [probe_a, probe_b], 50), + _run(HeatingProcess, "S2 Heating", sample2, [probe_b], 90), +] + +# -- Launch ------------------------------------------------------------------ + +config = DashboardConfig(lang="en", plot=PlotConfig(auto_fetch=True, row_limit=10000)) + +view = ProcessObjectView( + objects=objects, + processes=processes, + controllers=controllers, + config=config, + title="Process / Object Archive View", +) + +view.servable() + + +# -- Loading from OpenSemanticLab instead of building locally ---------------- +# +# from osw.express import OswExpress +# from osw.wiki_tools import SearchParam +# +# osw = OswExpress(domain="your-domain.org", cred_filepath="accounts.pwd.yaml") +# object_titles = ["Item:OSW", "Item:OSW"] +# objects = osw.load_entity(object_titles) +# +# # processes that consumed any of these objects +# process_titles = [] +# for obj in objects: +# process_titles += osw.site.semantic_search( +# SearchParam(query=f"[[HasInput::{obj.get_iri()}]]") +# ) +# processes = osw.load_entity(list(set(process_titles))) +# +# # build a DataToolController per datatool referenced by the processes, +# # wiring each controller's archive_database from its storage_locations +# # (LocalTimeSeriesDatabaseController / PostgrestTimeSeriesDatabaseController). +# controllers = [...] +# +# view = ProcessObjectView(objects, processes, controllers) +# view.servable() diff --git a/src/opensemantic/base/view/__init__.py b/src/opensemantic/base/view/__init__.py index fbd0d3f..c7da6e5 100644 --- a/src/opensemantic/base/view/__init__.py +++ b/src/opensemantic/base/view/__init__.py @@ -16,14 +16,24 @@ PlotConfig, ) from opensemantic.base.view._datatool_dashboard import DataToolView +from opensemantic.base.view._process_dashboard import ProcessObjectView +from opensemantic.base.view._process_utils import ( + build_concrete_tree, + derive_aggregated_channels, + resolve_aggregated_channel, +) __all__ = [ "DataToolView", + "ProcessObjectView", "DashboardConfig", "GroupingMode", "LangCode", "PlotConfig", "build_tree_source", + "build_concrete_tree", + "derive_aggregated_channels", + "resolve_aggregated_channel", "flatten_composite_channels", "get_available_units", "get_display_label", diff --git a/src/opensemantic/base/view/_base_view.py b/src/opensemantic/base/view/_base_view.py new file mode 100644 index 0000000..a5c23a0 --- /dev/null +++ b/src/opensemantic/base/view/_base_view.py @@ -0,0 +1,232 @@ +"""Shared building blocks for archive views. + +``BaseDataView`` is a mixin (no ``__init__``) holding the parts that are +identical across the channel-centered :class:`DataToolView` and the +process/object-centered :class:`ProcessObjectView`: unit-switch controls, the +plot/log/config cards, the value→numeric unit conversion, and the +servable/panel helpers. Subclasses provide the data model and the +view-specific bits (trees, controls, ``_build_figure`` / +``_update_log_console`` / ``_load_and_plot``). + +The mixin only reads attributes the subclass sets in its own ``__init__`` / +build steps (``_config``, ``_groups``, ``_unit_selections``, +``_unit_controls``, ``_plot_col``/``_plot_card``, ``_app`` ...), so it composes +cleanly without participating in ``__init__`` and without disturbing existing +subclasses (e.g. ``LiveDataToolView``). +""" + +import asyncio +import logging +from typing import Any, List, Tuple + +import panel as pn +from bokeh.palettes import Category10_10 +from panelini.panels.jsoneditor import JsonEditor + +from opensemantic.base.view._channel_utils import ( + _get_unit_symbol_map, + _t, + get_available_units, + get_unit_enum, + resolve_characteristic_class, + resolve_characteristic_label, + resolve_value_type, +) + +_logger = logging.getLogger(__name__) + +COLORS = Category10_10 + + +class BaseDataView: + """Mixin with the UI/plot pieces shared by the archive views.""" + + @property + def lang(self) -> str: + return self._config.lang + + # -- Unit-switch controls (one dropdown per quantity group) -- + + def _update_unit_controls(self): + self._unit_controls.clear() + for group_key, channels in self._groups.items(): + if not channels: + continue + sample_ch = channels[0][1] + if resolve_value_type(sample_ch) != "quantity": + continue + units = get_available_units(sample_ch) + if not units: + continue + char_label = resolve_characteristic_label(sample_ch, self.lang) + current = self._unit_selections.get(group_key) + options = {u["symbol"]: u["name"] for u in units} + if current not in options.values(): + current = next(iter(options.values())) + self._unit_selections[group_key] = current + dropdown = pn.widgets.Select( + name=f"{_t('unit', self.lang)}: {char_label}", + options=options, + value=current, + ) + dropdown.param.watch( + lambda event, _key=group_key: self._on_unit_change(_key, event), + ["value"], + ) + self._unit_controls.append(dropdown) + + def _on_unit_change(self, group_key: str, event): + self._unit_selections[group_key] = event.new + self._refresh_plot() + + # -- Plot / log / config cards -- + + def _build_plot(self): + self._plot_col = pn.Column( + sizing_mode="stretch_width", scroll=True, max_height=600 + ) + self._plot_card = pn.Card( + self._plot_col, + title=_t("time_series", self.lang), + sizing_mode="stretch_width", + ) + + def _build_log_console(self): + self._log_pane = pn.pane.HTML( + "", + sizing_mode="stretch_width", + height=200, + styles={"overflow-y": "auto"}, + ) + self._log_card = pn.Card( + self._log_pane, + title=_t("log_console", self.lang), + collapsed=False, + visible=False, + sizing_mode="stretch_width", + ) + + def _build_config_editor(self): + schema = self._config.model_json_schema() + self._config_editor = JsonEditor( + value=self._config.model_dump(), + options={ + "schema": schema, + "no_additional_properties": True, + "disable_edit_json": False, + }, + ) + self._config_editor.param.watch(self._on_config_editor_change, ["value"]) + self._config_card = pn.Card( + pn.Column( + self._config_editor, + sizing_mode="stretch_width", + max_height=1000, + scroll=True, + ), + title=_t("config", self.lang), + collapsed=True, + ) + + # -- Plot helpers -- + + def _refresh_plot(self): + """Rebuild plot and log console from the subclass's loaded data.""" + self._build_figure() + self._update_log_console() + + def _numeric(self, value: Any, channel: Any, target_unit_name: Any) -> Any: + """Convert a value to its display unit and return the numeric scalar. + + Handles typed Characteristic instances (with ``to_unit``), raw dicts + and bare scalars. Returns ``None`` when no value can be extracted. + """ + if target_unit_name and hasattr(value, "to_unit"): + unit_enum = get_unit_enum(channel) + if unit_enum is not None and target_unit_name in unit_enum.__members__: + try: + value = value.to_unit(unit_enum[target_unit_name]) + except Exception: + pass + if hasattr(value, "value"): + return value.value + if isinstance(value, dict): + return value.get("value") + return value + + def _get_axis_label(self, group_key: str) -> str: + """Build y-axis label: characteristic name [unit symbol].""" + channels = self._groups.get(group_key, []) + if not channels: + return "" + sample_ch = channels[0][1] + char_label = resolve_characteristic_label(sample_ch, self.lang) + + unit_name = self._unit_selections.get(group_key) + if unit_name: + for u in get_available_units(sample_ch): + if u["name"] == unit_name: + return f"{char_label} [{u['symbol']}]" + + ch_unit = getattr(sample_ch, "unit", None) + if ch_unit: + cls = resolve_characteristic_class(sample_ch) + symbol_map = _get_unit_symbol_map(cls) + unit_enum = get_unit_enum(sample_ch) + if unit_enum: + for member in unit_enum: + if member.value == ch_unit or member.name == ch_unit: + symbol = symbol_map.get(member.name, member.name) + return f"{char_label} [{symbol}]" + + return char_label + + # -- Data loading entry point (async-context aware) -- + + def _trigger_load(self): + """Start data loading, handling sync vs running-loop contexts.""" + try: + asyncio.get_running_loop() + asyncio.ensure_future(self._load_and_plot()) + except RuntimeError: + asyncio.run(self._load_and_plot()) + + # -- Serving -- + + def servable(self, **kwargs): + if self._app is None: + raise RuntimeError( + f"{type(self).__name__}(embeddable=True) has no servable app; " + "place sidebar_cards / main_cards into a host app instead." + ) + return self._app.servable(**kwargs) + + def panel(self): + """Return the Panelini app for embedding (None when embeddable).""" + return self._app + + # -- Subclass contract (implemented by each view) -- + + def _build_figure(self): # pragma: no cover - overridden + raise NotImplementedError + + def _update_log_console(self): # pragma: no cover - overridden + raise NotImplementedError + + async def _load_and_plot(self): # pragma: no cover - overridden + raise NotImplementedError + + def _on_config_editor_change(self, event): # pragma: no cover - overridden + raise NotImplementedError + + @property + def sidebar_cards(self) -> List[Any]: # pragma: no cover - overridden + raise NotImplementedError + + @property + def main_cards(self) -> List[Any]: # pragma: no cover - overridden + raise NotImplementedError + + +# Re-exported helper type for subclasses that annotate group maps. +GroupList = List[Tuple[Any, Any]] diff --git a/src/opensemantic/base/view/_datatool_dashboard.py b/src/opensemantic/base/view/_datatool_dashboard.py index efd8962..3b53f60 100644 --- a/src/opensemantic/base/view/_datatool_dashboard.py +++ b/src/opensemantic/base/view/_datatool_dashboard.py @@ -13,31 +13,24 @@ view.servable() """ -import asyncio import datetime as dt import logging from typing import Any, Dict, List, Optional, Tuple import panel as pn from bokeh.models import ColumnDataSource, DatetimeTickFormatter -from bokeh.palettes import Category10_10 from bokeh.plotting import figure as bk_figure from panelini import Panelini -from panelini.panels.jsoneditor import JsonEditor from panelini.panels.wunderbaum import Wunderbaum +from opensemantic.base.view._base_view import COLORS, BaseDataView from opensemantic.base.view._channel_utils import ( - _get_unit_symbol_map, _t, build_tree_source, flatten_composite_channels, - get_available_units, get_display_label, get_selected_channels, - get_unit_enum, group_channels_by_characteristic, - resolve_characteristic_class, - resolve_characteristic_label, resolve_value_type, ) from opensemantic.base.view._config import DashboardConfig @@ -54,10 +47,8 @@ def get_unit_enum_from_value(value: Any) -> Any: _logger = logging.getLogger(__name__) -COLORS = Category10_10 - -class DataToolView: +class DataToolView(BaseDataView): """Archive-mode DataTool view. Displays multiple DataToolControllers in a TreeGrid sidebar. @@ -79,10 +70,15 @@ def __init__( controllers: Optional[List[Any]] = None, config: Optional[DashboardConfig] = None, title: str = "DataTool Dashboard", + embeddable: bool = False, ): self._controllers = controllers or [] self._config = config or DashboardConfig() self._title = title + # When embeddable, skip building the internal Panelini app so the cards + # can be placed into a host app via sidebar_cards / main_cards (Panel + # rejects the same model living in two layouts/documents). + self._embeddable = embeddable # Build lookup maps self._channel_map: Dict[str, Any] = {} @@ -107,10 +103,6 @@ def __init__( self._build_config_editor() self._build_layout() - @property - def lang(self) -> str: - return self._config.lang - def _build_lookup_maps(self): for ctrl in self._controllers: for ch in ctrl.get_all_channels(): @@ -239,90 +231,7 @@ def _on_clear_cache(self, event): self._cache.clear_cache() self._cached_data.clear() - def _update_unit_controls(self): - self._unit_controls.clear() - for group_key, channels in self._groups.items(): - if not channels: - continue - sample_ch = channels[0][1] - vtype = resolve_value_type(sample_ch) - if vtype != "quantity": - continue - units = get_available_units(sample_ch) - if not units: - continue - char_label = resolve_characteristic_label(sample_ch, self.lang) - current = self._unit_selections.get(group_key) - options = {u["symbol"]: u["name"] for u in units} - if current not in options.values(): - current = next(iter(options.values())) - self._unit_selections[group_key] = current - dropdown = pn.widgets.Select( - name=f"{_t('unit', self.lang)}: {char_label}", - options=options, - value=current, - ) - dropdown.param.watch( - lambda event, _key=group_key: self._on_unit_change(_key, event), - ["value"], - ) - self._unit_controls.append(dropdown) - - def _on_unit_change(self, group_key: str, event): - self._unit_selections[group_key] = event.new - self._refresh_plot() - - # -- Plot -- - - def _build_plot(self): - self._plot_col = pn.Column( - sizing_mode="stretch_width", scroll=True, max_height=600 - ) - self._plot_card = pn.Card( - self._plot_col, - title=_t("time_series", self.lang), - sizing_mode="stretch_width", - ) - - def _build_log_console(self): - self._log_data = [] - self._log_pane = pn.pane.HTML( - "", - sizing_mode="stretch_width", - height=200, - styles={"overflow-y": "auto"}, - ) - self._log_card = pn.Card( - self._log_pane, - title=_t("log_console", self.lang), - collapsed=False, - visible=False, - sizing_mode="stretch_width", - ) - - # -- Config Editor -- - - def _build_config_editor(self): - schema = self._config.model_json_schema() - self._config_editor = JsonEditor( - value=self._config.model_dump(), - options={ - "schema": schema, - "no_additional_properties": True, - "disable_edit_json": False, - }, - ) - self._config_editor.param.watch(self._on_config_editor_change, ["value"]) - self._config_card = pn.Card( - pn.Column( - self._config_editor, - sizing_mode="stretch_width", - max_height=1000, - scroll=True, - ), - title=_t("config", self.lang), - collapsed=True, - ) + # -- Config editor change handler (view-specific) -- def _on_config_editor_change(self, event): if not event.new or not isinstance(event.new, dict): @@ -383,16 +292,7 @@ def _rebuild_ui_labels(self): # Rebuild unit controls self._update_unit_controls() - # -- Data Loading -- - - def _trigger_load(self): - """Start data loading (handles async context).""" - _logger.debug("_trigger_load called, %d selected", len(self._selected)) - try: - asyncio.get_running_loop() - asyncio.ensure_future(self._load_and_plot()) - except RuntimeError: - asyncio.run(self._load_and_plot()) + # -- Data Loading (_trigger_load comes from BaseDataView) -- async def _load_and_plot(self): """Load data for all selected channels and update plots.""" @@ -432,11 +332,6 @@ async def _load_and_plot(self): self._refresh_plot() - def _refresh_plot(self): - """Rebuild plot and log console from cached data.""" - self._build_figure() - self._update_log_console() - def _build_figure(self): """Build Bokeh figures - one per characteristic group.""" self._plot_col.clear() @@ -517,23 +412,7 @@ def _extract_trace_data(self, ch: Any, group_key: str) -> Tuple[List, List]: if ts.tzinfo is not None: ts = ts.astimezone(tz=None).replace(tzinfo=None) timestamps.append(ts) - - value = pt.value - # Convert to display unit if needed - if target_unit_name and hasattr(value, "to_unit"): - unit_enum = get_unit_enum(ch) - if unit_enum is not None and target_unit_name in unit_enum.__members__: - try: - value = value.to_unit(unit_enum[target_unit_name]) - except Exception: - pass - # Extract numeric value - if hasattr(value, "value"): - values.append(value.value) - elif isinstance(value, dict): - values.append(value.get("value", 0)) - else: - values.append(value) + values.append(self._numeric(pt.value, ch, target_unit_name)) return timestamps, values @@ -595,39 +474,6 @@ def _extract_composite_field( return timestamps, values - def _get_axis_label(self, group_key: str) -> str: - """Build y-axis label: characteristic name [unit symbol]. - - Uses the selected unit if set, otherwise the channel's configured unit. - """ - channels = self._groups.get(group_key, []) - if not channels: - return "" - sample_ch = channels[0][1] - char_label = resolve_characteristic_label(sample_ch, self.lang) - - # Try selected unit first - unit_name = self._unit_selections.get(group_key) - if unit_name: - units = get_available_units(sample_ch) - for u in units: - if u["name"] == unit_name: - return f"{char_label} [{u['symbol']}]" - - # Fall back to channel's configured unit - ch_unit = getattr(sample_ch, "unit", None) - if ch_unit: - cls = resolve_characteristic_class(sample_ch) - symbol_map = _get_unit_symbol_map(cls) - unit_enum = get_unit_enum(sample_ch) - if unit_enum: - for member in unit_enum: - if member.value == ch_unit or member.name == ch_unit: - symbol = symbol_map.get(member.name, member.name) - return f"{char_label} [{symbol}]" - - return char_label - def _update_log_console(self): """Update the log console with text-type channel data.""" log_entries = [] @@ -675,6 +521,10 @@ def _update_log_console(self): # -- Layout -- def _build_layout(self): + if self._embeddable: + # Host app owns the layout; expose cards via sidebar_cards/main_cards. + self._app = None + return self._app = Panelini( title=self._title, sidebar_enabled=True, @@ -693,10 +543,38 @@ def _build_main_area(self): """Set main area content. Override in subclasses for tabs.""" self._app.main_set([self._plot_card, self._log_card]) - def servable(self, **kwargs): - """Make the dashboard servable.""" - return self._app.servable(**kwargs) + @property + def sidebar_cards(self): + """Sidebar cards (channel tree, plot controls, config) for embedding.""" + return [self._tree_card, self._controls_card, self._config_card] + + @property + def main_cards(self): + """Main-area cards (time series plot, log console) for embedding.""" + return [self._plot_card, self._log_card] + + def set_time_range(self, start, end, fetch: bool = True): + """Set an explicit time range and (optionally) reload the data. + + Accepts tz-aware or naive datetimes (naive is treated as UTC). Disables + auto-fetch, which otherwise pins the end to ``now``, so an arbitrary + historical window can be shown. + """ + self._auto_fetch_cb.value = False + self._start_picker.value = self._to_picker_value(start) + self._end_picker.value = self._to_picker_value(end) + if fetch: + self._trigger_load() + + @staticmethod + def _to_picker_value(value): + """Convert a datetime to the naive local value the picker expects. + + The DatetimePicker holds naive local time which _load_and_plot converts + back to UTC, so hand it the local-naive representation of ``value``. + """ + if value.tzinfo is None: + value = value.replace(tzinfo=dt.timezone.utc) + return value.astimezone().replace(tzinfo=None) - def panel(self): - """Return the Panelini app for embedding.""" - return self._app + # servable() / panel() come from BaseDataView. diff --git a/src/opensemantic/base/view/_process_dashboard.py b/src/opensemantic/base/view/_process_dashboard.py new file mode 100644 index 0000000..44de0de --- /dev/null +++ b/src/opensemantic/base/view/_process_dashboard.py @@ -0,0 +1,498 @@ +"""Process/object-centered DataTool view. + +Two Wunderbaum trees: +- Objects (Item instances tracked as process inputs) +- Process types -> aggregated channels (grouped by datatool type, channel name, + characteristic) + +Selected object x aggregated-channel pairs resolve to concrete channels, are +loaded over each process's [start, end] window, time-normalized so the first +data point of each process is t=0, and plotted (seconds on the x-axis) grouped +by characteristic - reusing the plot/group/unit machinery of DataToolView. + +Usage: + from opensemantic.base.view import ProcessObjectView + + view = ProcessObjectView(objects, processes, controllers) + view.servable() +""" + +import datetime as dt +import logging +from typing import Any, Dict, List, Optional, Tuple + +import panel as pn +from bokeh.models import ColumnDataSource +from bokeh.plotting import figure as bk_figure +from panelini import Panelini +from panelini.panels.wunderbaum import Wunderbaum + +from opensemantic.base.view._base_view import COLORS, BaseDataView +from opensemantic.base.view._channel_utils import ( + _t, + get_display_label, + group_channels_by_characteristic, + resolve_value_type, +) +from opensemantic.base.view._config import DashboardConfig +from opensemantic.base.view._data_cache import ChannelDataCache +from opensemantic.base.view._process_utils import ( + build_concrete_tree, + build_object_tree_source, + build_process_tree_source, + derive_aggregated_channels, + entity_iri, + get_selected_keys, + resolve_aggregated_channel, +) + +_logger = logging.getLogger(__name__) + +# Local labels not in the shared _channel_utils translation table. +_PT_STRINGS = { + "objects": {"en": "Objects", "de": "Objekte"}, + "process_channels": { + "en": "Process Types / Channels", + "de": "Prozesstypen / Kanaele", + }, + "rel_time": {"en": "t [s] (relative)", "de": "t [s] (relativ)"}, + "n_processes": {"en": "#", "de": "#"}, +} + + +def _pt(key: str, lang: str = "en") -> str: + entry = _PT_STRINGS.get(key, {}) + return entry.get(lang, entry.get("en", key)) + + +def _as_utc(value: Any) -> Optional[dt.datetime]: + if value is None: + return None + if isinstance(value, str): + value = dt.datetime.fromisoformat(value) + if value.tzinfo is None: + value = value.replace(tzinfo=dt.timezone.utc) + return value + + +class ProcessObjectView(BaseDataView): + """Process/object-centered archive view. + + Parameters + ---------- + objects + Item instances (the entities tracked as process inputs) shown in tree 1. + processes + Process instances to scan. A process qualifies when one of ``objects`` + is among its inputs, it has start+end times, and >=1 DataTool attached. + controllers + DataToolController instances used to load channel data; matched to + process tools by IRI. + config + Dashboard configuration (uses ``lang`` and ``plot.*``). If None, defaults. + title + Dashboard title shown in the Panelini header. + embeddable + If True, skip the internal Panelini app (expose ``sidebar_cards`` / + ``main_cards`` for a host app instead). + """ + + def __init__( + self, + objects: Optional[List[Any]] = None, + processes: Optional[List[Any]] = None, + controllers: Optional[List[Any]] = None, + config: Optional[DashboardConfig] = None, + title: str = "Process Dashboard", + embeddable: bool = False, + ): + self._objects = objects or [] + self._processes = processes or [] + self._controllers = controllers or [] + self._config = config or DashboardConfig() + self._title = title + self._embeddable = embeddable + + # Virtual structure + self._concrete = build_concrete_tree( + self._objects, self._processes, self._controllers, self.lang + ) + self._aggregated = derive_aggregated_channels(self._concrete, self.lang) + self._agg_by_key: Dict[str, Dict[str, Any]] = {} + for grp in self._aggregated.values(): + for agg in grp["channels"].values(): + self._agg_by_key[agg["key"]] = agg + + # State + self._cache = ChannelDataCache(enabled=self._config.plot.cache_enabled) + self._selected_objects: List[Dict[str, Any]] = [] + self._selected_aggs: List[Dict[str, Any]] = [] + self._groups: Dict[str, List[Tuple[Any, Any]]] = {} + self._group_of: Dict[str, str] = {} + self._unit_selections: Dict[str, str] = {} + self._traces: List[Dict[str, Any]] = [] + self._t0: Dict[Tuple[Any, Any], dt.datetime] = {} + + # UI + self._build_object_tree() + self._build_process_tree() + self._build_controls() + self._build_plot() + self._build_log_console() + self._build_config_editor() + self._build_layout() + + # -- Trees -- + + def _build_object_tree(self): + source = build_object_tree_source(self._concrete, self.lang) + self._obj_tree = Wunderbaum( + source=source, + height=220, + columns=[ + {"id": "*", "title": _pt("objects", self.lang), "width": "200px"}, + { + "id": "processes", + "title": _pt("n_processes", self.lang), + "width": "60px", + }, + ], + options={"checkbox": True, "selectMode": "hier"}, + ) + self._obj_tree.param.watch(self._on_change, ["source"]) + self._obj_tree_card = pn.Card( + self._obj_tree, + title=_pt("objects", self.lang), + collapsed=False, + ) + + def _build_process_tree(self): + source = build_process_tree_source(self._aggregated, self.lang) + self._proc_tree = Wunderbaum( + source=source, + height=220, + columns=[ + { + "id": "*", + "title": _pt("process_channels", self.lang), + "width": "220px", + }, + { + "id": "characteristic", + "title": _t("characteristic", self.lang), + "width": "150px", + }, + ], + options={"checkbox": True, "selectMode": "hier"}, + ) + self._proc_tree.param.watch(self._on_change, ["source"]) + self._proc_tree_card = pn.Card( + self._proc_tree, + title=_pt("process_channels", self.lang), + collapsed=False, + ) + + def _on_change(self, *args): + try: + self._recompute_selection() + self._update_unit_controls() + if self._config.plot.auto_fetch: + self._trigger_load() + except Exception as e: + _logger.error("Error in _on_change: %s", e) + + def _recompute_selection(self): + obj_keys = get_selected_keys(self._obj_tree.source) + agg_keys = get_selected_keys(self._proc_tree.source) + self._selected_objects = [ + self._concrete[k] for k in obj_keys if k in self._concrete + ] + self._selected_aggs = [ + self._agg_by_key[k] for k in agg_keys if k in self._agg_by_key + ] + # Unique concrete (controller, channel) pairs for y-axis grouping + pairs: Dict[str, Tuple[Any, Any]] = {} + for obj_entry in self._selected_objects: + for agg in self._selected_aggs: + for ctrl, ch, _proc in resolve_aggregated_channel(obj_entry, agg): + pairs[ch.uuid] = (ctrl, ch) + self._groups = group_channels_by_characteristic( + list(pairs.values()), self._config.plot.grouping + ) + self._group_of = {} + for gkey, chans in self._groups.items(): + for _ctrl, ch in chans: + self._group_of[ch.uuid] = gkey + + # -- Controls -- + + def _build_controls(self): + self._load_button = pn.widgets.Button( + name=_t("load_data", self.lang), button_type="primary" + ) + self._load_button.on_click(lambda e: self._trigger_load()) + + self._auto_fetch_cb = pn.widgets.Checkbox( + name=_t("auto_fetch", self.lang), + value=self._config.plot.auto_fetch, + ) + self._auto_fetch_cb.param.watch(self._on_auto_fetch_change, ["value"]) + + self._row_limit_input = pn.widgets.IntInput( + name=_t("row_limit", self.lang), + value=self._config.plot.row_limit, + start=1, + step=1000, + ) + self._row_limit_input.param.watch(self._on_row_limit_change, ["value"]) + + self._clear_cache_button = pn.widgets.Button( + name=_t("clear_cache", self.lang), button_type="warning" + ) + self._clear_cache_button.on_click(self._on_clear_cache) + + self._unit_controls = pn.Column() + + self._controls_card = pn.Card( + self._load_button, + self._auto_fetch_cb, + self._row_limit_input, + self._clear_cache_button, + self._unit_controls, + title=_t("plot_controls", self.lang), + ) + + def _on_auto_fetch_change(self, event): + self._config.plot.auto_fetch = event.new + + def _on_row_limit_change(self, event): + self._config.plot.row_limit = event.new + + def _on_clear_cache(self, event): + self._cache.clear_cache() + + # -- Config editor change handler (view-specific) -- + + def _on_config_editor_change(self, event): + if not event.new or not isinstance(event.new, dict): + return + try: + new_config = DashboardConfig.model_validate(event.new) + except Exception as e: + _logger.debug("Incomplete config value, skipping: %s", e) + return + old_config = self._config + self._config = new_config + + if old_config.plot.grouping != new_config.plot.grouping: + self._recompute_selection() + self._update_unit_controls() + self._refresh_plot() + if old_config.plot.cache_enabled != new_config.plot.cache_enabled: + self._cache.enabled = new_config.plot.cache_enabled + self._auto_fetch_cb.value = new_config.plot.auto_fetch + self._row_limit_input.value = new_config.plot.row_limit + + # -- Data loading (_trigger_load comes from BaseDataView) -- + + async def _load_and_plot(self): + limit = self._config.plot.row_limit + self._traces = [] + + for obj_entry in self._selected_objects: + for agg in self._selected_aggs: + for ctrl, ch, proc in resolve_aggregated_channel(obj_entry, agg): + start = _as_utc(getattr(proc, "start_date_time", None)) + end = _as_utc(getattr(proc, "end_date_time", None)) + if start is None: + continue + try: + points = await self._cache.get_data(ctrl, ch, start, end, limit) + except Exception as e: + _logger.error( + "Error loading %s/%s: %s", + getattr(ctrl, "name", "?"), + getattr(ch, "name", "?"), + e, + ) + points = [] + self._traces.append( + { + "object": obj_entry["object"], + "object_label": obj_entry["label"], + "process": proc, + "process_label": get_display_label(proc, self.lang), + "controller": ctrl, + "channel": ch, + "points": points, + } + ) + + # t=0 per (object, process) = earliest loaded point across its channels + self._t0 = {} + for tr in self._traces: + if not tr["points"]: + continue + key = (entity_iri(tr["object"]), entity_iri(tr["process"])) + mn = min(_as_utc(p.timestamp) for p in tr["points"]) + cur = self._t0.get(key) + if cur is None or mn < cur: + self._t0[key] = mn + + self._refresh_plot() + + def _build_figure(self): + self._plot_col.clear() + + # Group traces by y-axis group, skipping text channels. + plot_groups: Dict[str, List[Dict[str, Any]]] = {} + for tr in self._traces: + ch = tr["channel"] + if resolve_value_type(ch) == "text": + continue + gkey = self._group_of.get(ch.uuid) + if gkey is None: + continue + plot_groups.setdefault(gkey, []).append(tr) + + if not plot_groups: + return + + color_idx = 0 + for gkey, traces in plot_groups.items(): + axis_label = self._get_axis_label(gkey) + fig = bk_figure( + height=250, + sizing_mode="stretch_width", + x_axis_label=_pt("rel_time", self.lang), + y_axis_label=axis_label, + ) + for tr in traces: + xs, ys = self._extract_trace(tr, gkey) + if not xs: + continue + # One aggregated channel fans out to every real channel each + # sample has data on (across process runs and across datatools + # of the same type). Label each line distinctly by + # object · process / datatool / channel so Bokeh keeps them as + # separate, individually-toggleable legend entries. + label = ( + f"{tr['object_label']} · {tr['process_label']} / " + f"{get_display_label(tr['controller'], self.lang)} / " + f"{get_display_label(tr['channel'], self.lang)}" + ) + src = ColumnDataSource(data={"x": xs, "y": ys}) + fig.line( + "x", + "y", + source=src, + legend_label=label, + color=COLORS[color_idx % len(COLORS)], + line_width=2, + ) + color_idx += 1 + fig.legend.click_policy = "hide" + fig.legend.label_text_font_size = "8pt" + self._plot_col.append(pn.pane.Bokeh(fig, sizing_mode="stretch_width")) + + def _extract_trace(self, tr: Dict[str, Any], group_key: str) -> Tuple[List, List]: + """Relative-seconds x and (unit-converted) numeric y for a trace.""" + points = tr["points"] + if not points: + return [], [] + key = (entity_iri(tr["object"]), entity_iri(tr["process"])) + t0 = self._t0.get(key) + if t0 is None: + return [], [] + + ch = tr["channel"] + target_unit_name = self._unit_selections.get(group_key) + xs: List[float] = [] + ys: List[Any] = [] + + for pt in points: + ts = _as_utc(pt.timestamp) + secs = (ts - t0).total_seconds() + v = self._numeric(pt.value, ch, target_unit_name) + if v is None: + continue + xs.append(secs) + ys.append(v) + + return xs, ys + + def _update_log_console(self): + log_entries = [] + has_text = False + for tr in self._traces: + ch = tr["channel"] + if resolve_value_type(ch) != "text": + continue + has_text = True + key = (entity_iri(tr["object"]), entity_iri(tr["process"])) + t0 = self._t0.get(key) + prefix = f"{tr['object_label']} · {tr['process_label']}" + for pt in tr["points"]: + ts = _as_utc(pt.timestamp) + secs = (ts - t0).total_seconds() if t0 is not None else 0.0 + val = pt.value + if hasattr(val, "value"): + text = str(val.value) + elif isinstance(val, dict): + text = str(val.get("value", val)) + else: + text = str(val) + log_entries.append((secs, prefix, text)) + + self._log_card.visible = has_text + if not log_entries: + self._log_pane.object = "" + return + + log_entries.sort(key=lambda x: x[0]) + html_lines = [] + for secs, prefix, text in log_entries: + html_lines.append( + "
" + f"+{secs:.1f}s " + f"[{prefix}] {text}
" + ) + self._log_pane.object = "\n".join(html_lines) + + # -- Layout -- + + def _build_layout(self): + if self._embeddable: + self._app = None + return + self._app = Panelini( + title=self._title, + sidebar_enabled=True, + sidebars_max_width=400, + ) + self._app.sidebar_set( + [ + self._obj_tree_card, + self._proc_tree_card, + self._controls_card, + self._config_card, + ] + ) + self._app.main_set([self._plot_card, self._log_card]) + + @property + def sidebar_cards(self): + """Sidebar cards (object tree, process tree, controls, config).""" + return [ + self._obj_tree_card, + self._proc_tree_card, + self._controls_card, + self._config_card, + ] + + @property + def main_cards(self): + """Main-area cards (time series plot, log console).""" + return [self._plot_card, self._log_card] + + # servable() / panel() come from BaseDataView. diff --git a/src/opensemantic/base/view/_process_utils.py b/src/opensemantic/base/view/_process_utils.py new file mode 100644 index 0000000..ca3e686 --- /dev/null +++ b/src/opensemantic/base/view/_process_utils.py @@ -0,0 +1,472 @@ +"""Pure logic for the process/object-centered dashboard. + +Builds, from a list of objects (Item instances), the processes they were inputs +to, the DataTool controllers attached to those processes, and a virtual +"aggregated channel" structure grouped by process type. No UI dependencies +(no Panel/Bokeh imports). + +All entity relations are resolved by IRI via ``get_iri`` / ``get_iri_ref`` so the +logic works offline (no backend resolver required): an object qualifies for a +process when its IRI appears in the process's ``input`` references; a process's +``tool`` references are matched against the provided controllers by IRI. +""" + +import logging +from typing import Any, Dict, List, Optional, Tuple + +from opensemantic.base.view._channel_utils import ( + get_characteristic_iri, + get_display_label, + get_display_label_cls, + resolve_characteristic_label, + resolve_value_type, +) + +_logger = logging.getLogger(__name__) + + +# -- IRI / identity helpers (offline-safe) -- + + +def entity_iri(obj: Any) -> Optional[str]: + """Return an entity's IRI string, or None.""" + getter = getattr(obj, "get_iri", None) + if getter is None: + return None + try: + return getter() + except Exception: + return None + + +def iri_refs(obj: Any, field: str) -> List[str]: + """Return the IRI reference string(s) of a (range) field without resolving. + + Tries ``get_iri_ref`` then ``__iris__`` then a resolved-value fallback. + """ + getter = getattr(obj, "get_iri_ref", None) + refs = None + if getter is not None: + try: + refs = getter(field) + except Exception: + refs = None + if refs is None: + iris = getattr(obj, "__iris__", {}) or {} + refs = iris.get(field) + if refs is None: + # Fallback: field may hold inline objects/strings + try: + val = getattr(obj, field, None) + except Exception: + val = None + if val is None: + return [] + if not isinstance(val, list): + val = [val] + out = [] + for v in val: + if isinstance(v, str): + out.append(v) + else: + vi = entity_iri(v) + if vi: + out.append(vi) + return out + if isinstance(refs, str): + return [refs] + return list(refs) + + +def type_key(obj: Any) -> Tuple[str, ...]: + """Return the entity's ``type`` as a hashable tuple of category IRIs.""" + t = getattr(obj, "type", None) or [] + if isinstance(t, str): + t = [t] + return tuple(t) + + +def type_label(type_key_: Tuple[str, ...], lang: str = "en") -> str: + """Human label for a type tuple: resolve the first IRI to a class label.""" + if not type_key_: + return "?" + iri = type_key_[0] + try: + from oold.model import _types + + cls = _types.get(iri) + if cls is None: + from oold.model.v1 import _types as _v1_types + + cls = _v1_types.get(iri) + if cls is not None: + lbl = get_display_label_cls(cls, lang) + if lbl: + return lbl + except ImportError: + pass + return iri.split(":")[-1] if ":" in iri else iri + + +def _controllers_by_iri(controllers: List[Any]) -> Dict[str, Any]: + """Map controller IRIs (and ``Item:`` + osw_id) to controllers.""" + out: Dict[str, Any] = {} + for ctrl in controllers: + iri = entity_iri(ctrl) + if iri: + out[iri] = ctrl + osw_getter = getattr(ctrl, "get_osw_id", None) + if osw_getter is not None: + try: + out["Item:" + ctrl.get_osw_id()] = ctrl + except Exception: + pass + return out + + +# -- Concrete tree: object -> processes -> controllers -- + + +def build_concrete_tree( + objects: List[Any], + processes: List[Any], + controllers: List[Any], + lang: str = "en", +) -> Dict[str, Dict[str, Any]]: + """Build object -> qualifying-processes -> DataTool controllers. + + A process qualifies for an object when the object's IRI is one of the + process ``input`` references, the process has both ``start_date_time`` and + ``end_date_time``, and at least one of its ``tool`` references resolves to a + provided DataTool controller. + + Returns ``{object_iri: {object, iri, label, processes: [{process, iri, + label, controllers}]}}``. + """ + by_iri = _controllers_by_iri(controllers) + tree: Dict[str, Dict[str, Any]] = {} + + for obj in objects: + obj_iri = entity_iri(obj) + if obj_iri is None: + _logger.warning("Object has no IRI, skipping: %r", obj) + continue + entry: Dict[str, Any] = { + "object": obj, + "iri": obj_iri, + "label": get_display_label(obj, lang), + "processes": [], + } + + for proc in processes: + if obj_iri not in iri_refs(proc, "input"): + continue + proc_id = entity_iri(proc) or get_display_label(proc, lang) + start = getattr(proc, "start_date_time", None) + end = getattr(proc, "end_date_time", None) + if start is None or end is None: + _logger.warning("Process %s skipped: missing start/end time", proc_id) + continue + + proc_ctrls = [] + for tiri in iri_refs(proc, "tool"): + ctrl = by_iri.get(tiri) + if ctrl is None: + _logger.info( + "Process %s: tool %s is not a DataTool controller, " "skipping", + proc_id, + tiri, + ) + continue + proc_ctrls.append(ctrl) + + if not proc_ctrls: + _logger.warning("Process %s skipped: no DataTool attached", proc_id) + continue + + entry["processes"].append( + { + "process": proc, + "iri": entity_iri(proc), + "label": get_display_label(proc, lang), + "controllers": proc_ctrls, + } + ) + + tree[obj_iri] = entry + + return tree + + +# -- Virtual structure: aggregated channels grouped by process type -- + + +Signature = Tuple[Tuple[str, ...], Optional[str], Optional[str]] + + +def _sig_str(sig: Signature) -> str: + dt_key, name, char = sig + return "+".join(dt_key) + "::" + str(name) + "::" + str(char) + + +def _agg_key_merged(process_type: Tuple[str, ...], sig: Signature) -> str: + return "+".join(process_type) + "##" + _sig_str(sig) + "##agg" + + +def _agg_key_instance( + process_type: Tuple[str, ...], sig: Signature, datatool_iri: str +) -> str: + return "+".join(process_type) + "##" + _sig_str(sig) + "##inst::" + datatool_iri + + +def process_type_node_key(process_type: Tuple[str, ...]) -> str: + """Stable key for a process-type root node.""" + return "ptype::" + "+".join(process_type) + + +def _plural(n: int) -> str: + return "channel" if n == 1 else "channels" + + +def derive_aggregated_channels( + concrete_tree: Dict[str, Dict[str, Any]], + lang: str = "en", +) -> Dict[Tuple[str, ...], Dict[str, Any]]: + """Group qualifying processes by type and derive treeview channel entries. + + Aggregation is only a treeview convenience (tick once instead of many). + Channels share a *signature* ``(datatool_type, channel_name, + characteristic_iri)``. Within a process type, two datatool instances of the + same signature are merged only when they are **never co-present in the same + process run** (treated as drop-in replacements). Datatool instances that do + co-occur in some run are kept as **separate per-instance entries**, since + they are distinct measurement points. + + So per signature within a process type: + - co-present instances -> one entry each, labelled by datatool instance; + - the remaining (never co-present) instances -> one merged entry labelled + ``/ [n channels]`` whose tooltip lists the actual channels. + + Composite/unknown channels are skipped (info-logged). + + Returns ``{process_type: {process_type, label, channels: {agg_key: agg}}}``. + Each ``agg`` carries ``datatool_iris`` (the datatool instances it + represents), ``aggregated_channels`` (display strings), ``n_channels`` and an + ``aggregated`` flag. + """ + # Pass 1: gather, per process type, the runs (sig -> set of datatool IRIs) + # and per (pt, sig) the instance metadata. + pt_runs: Dict[Tuple[str, ...], List[Dict[Signature, set]]] = {} + pt_sig_inst: Dict[Tuple[str, ...], Dict[Signature, Dict[str, Dict[str, Any]]]] = {} + pt_label: Dict[Tuple[str, ...], str] = {} + + for obj_entry in concrete_tree.values(): + for pe in obj_entry["processes"]: + proc = pe["process"] + pt = type_key(proc) + pt_label.setdefault(pt, type_label(pt, lang)) + run_sig: Dict[Signature, set] = {} + for ctrl in pe["controllers"]: + dt_key = type_key(ctrl) + ctrl_iri = entity_iri(ctrl) + if ctrl_iri is None: + continue + for ch in ctrl.get_all_channels(): + vtype = resolve_value_type(ch) + if vtype in ("composite", "unknown"): + _logger.info( + "Channel %s (%s) excluded from aggregation", + getattr(ch, "name", "?"), + vtype, + ) + continue + name = getattr(ch, "name", None) + char_iri = get_characteristic_iri(ch) + sig: Signature = (dt_key, name, char_iri) + run_sig.setdefault(sig, set()).add(ctrl_iri) + inst = pt_sig_inst.setdefault(pt, {}).setdefault(sig, {}) + if ctrl_iri not in inst: + inst[ctrl_iri] = { + "datatool_label": get_display_label(ctrl, lang), + "channel_name": name, + "value_type": vtype, + "characteristic_label": resolve_characteristic_label( + ch, lang + ), + } + pt_runs.setdefault(pt, []).append(run_sig) + + # Pass 2: per process type + signature, split co-present vs free instances. + groups: Dict[Tuple[str, ...], Dict[str, Any]] = {} + for pt, sig_inst in pt_sig_inst.items(): + grp: Dict[str, Any] = { + "process_type": pt, + "label": pt_label[pt], + "channels": {}, + } + # Instances co-present with another same-signature instance in a run. + copresent: Dict[Signature, set] = {} + for run_sig in pt_runs[pt]: + for sig, iris in run_sig.items(): + if len(iris) >= 2: + copresent.setdefault(sig, set()).update(iris) + + for sig, inst_map in sig_inst.items(): + dt_key, name, char_iri = sig + co = copresent.get(sig, set()) + + # Per-instance entries for co-present datatools. + for iri, meta in inst_map.items(): + if iri not in co: + continue + key = _agg_key_instance(pt, sig, iri) + grp["channels"][key] = { + "key": key, + "process_type": pt, + "datatool_type": dt_key, + "channel_name": name, + "characteristic_iri": char_iri, + "characteristic_label": meta["characteristic_label"], + "value_type": meta["value_type"], + "aggregated": False, + "datatool_iris": [iri], + "n_channels": 1, + "aggregated_channels": [f"{meta['datatool_label']}/{name}"], + "label": f"{meta['datatool_label']}/{name}", + } + + # One merged entry for the remaining (never co-present) instances. + free = [iri for iri in inst_map if iri not in co] + if free: + n = len(free) + chans = [f"{inst_map[i]['datatool_label']}/{name}" for i in free] + meta0 = inst_map[free[0]] + type_lbl = type_label(dt_key, lang) + key = _agg_key_merged(pt, sig) + grp["channels"][key] = { + "key": key, + "process_type": pt, + "datatool_type": dt_key, + "channel_name": name, + "characteristic_iri": char_iri, + "characteristic_label": meta0["characteristic_label"], + "value_type": meta0["value_type"], + "aggregated": True, + "datatool_iris": list(free), + "n_channels": n, + "aggregated_channels": chans, + "label": f"{type_lbl}/{name} [{n} {_plural(n)}]", + } + + groups[pt] = grp + + return groups + + +def resolve_aggregated_channel( + object_entry: Dict[str, Any], + agg: Dict[str, Any], +) -> List[Tuple[Any, Any, Any]]: + """Resolve a treeview entry to concrete (controller, channel, process). + + Each entry represents a specific set of datatool instances + (``agg['datatool_iris']``): one instance for a per-instance entry, or the + pooled drop-in replacements for a merged entry. This returns one tuple per + matching physical channel across the object's processes - so a selection + fans out over the runs (and, for merged entries, the pooled datatools) the + object actually has data on. The view labels each line distinctly by + object / process / datatool / channel. + """ + allowed = set(agg.get("datatool_iris") or []) + out: List[Tuple[Any, Any, Any]] = [] + for pe in object_entry["processes"]: + proc = pe["process"] + if type_key(proc) != agg["process_type"]: + continue + for ctrl in pe["controllers"]: + if entity_iri(ctrl) not in allowed: + continue + for ch in ctrl.get_all_channels(): + if getattr(ch, "name", None) != agg["channel_name"]: + continue + if get_characteristic_iri(ch) != agg["characteristic_iri"]: + continue + out.append((ctrl, ch, proc)) + return out + + +# -- Wunderbaum tree-source builders -- + + +def build_object_tree_source( + concrete_tree: Dict[str, Dict[str, Any]], + lang: str = "en", +) -> List[Dict[str, Any]]: + """Flat checkable list of objects; key = object IRI.""" + source = [] + for obj_iri, entry in concrete_tree.items(): + n_proc = len(entry["processes"]) + source.append( + { + "title": entry["label"] or obj_iri, + "key": obj_iri, + "checkbox": True, + "selected": False, + "processes": str(n_proc), + "tooltip": (f"{entry['label']}\nIRI: {obj_iri}\nProcesses: {n_proc}"), + } + ) + return source + + +def build_process_tree_source( + aggregated: Dict[Tuple[str, ...], Dict[str, Any]], + lang: str = "en", +) -> List[Dict[str, Any]]: + """Process-type roots with aggregated channels as checkable children.""" + source = [] + for pt, grp in aggregated.items(): + children = [] + for agg in grp["channels"].values(): + chans = agg.get("aggregated_channels", []) + chan_lines = "\n".join(f" - {c}" for c in chans) + children.append( + { + "title": agg["label"], + "key": agg["key"], + "checkbox": True, + "selected": False, + "characteristic": agg["characteristic_label"], + "tooltip": ( + f"{agg['label']}\n" + f"Characteristic: {agg['characteristic_label']}\n" + f"Channel: {agg['channel_name']}\n" + f"Aggregated channels ({agg.get('n_channels', len(chans))}):\n" + f"{chan_lines}" + ), + } + ) + source.append( + { + "title": grp["label"], + "key": process_type_node_key(pt), + "expanded": True, + "checkbox": True, + "children": children, + } + ) + return source + + +def get_selected_keys(source: List[Dict[str, Any]]) -> List[str]: + """Collect keys of checked leaf nodes (and checked childless roots).""" + keys = [] + for node in source: + children = node.get("children") + if children: + for child in children: + if child.get("selected"): + keys.append(child.get("key")) + elif node.get("selected"): + keys.append(node.get("key")) + return keys diff --git a/tests/test_process_view.py b/tests/test_process_view.py new file mode 100644 index 0000000..3dd8b35 --- /dev/null +++ b/tests/test_process_view.py @@ -0,0 +1,364 @@ +"""Tests for the process/object-centered dashboard logic (_process_utils). + +Pure-logic tests (no Panel/Bokeh): build in-memory objects, controllers and +processes, then exercise concrete-tree building, aggregation, and resolution. +All relations are matched offline by IRI. +""" + +import datetime as dt +import logging +from uuid import NAMESPACE_URL, uuid5 + +import pytest + +from opensemantic import compute_scoped_uuid +from opensemantic.base.v1 import ( + Database, + DataChannel, + DataTool, + DataToolController, + Process, +) +from opensemantic.base.view._process_utils import ( + build_concrete_tree, + build_object_tree_source, + build_process_tree_source, + derive_aggregated_channels, + entity_iri, + get_selected_keys, + iri_refs, + resolve_aggregated_channel, +) +from opensemantic.characteristics.quantitative.v1 import Pressure, Temperature +from opensemantic.core.v1 import Item, Label + +START = dt.datetime(2024, 1, 1, tzinfo=dt.timezone.utc) +END = dt.datetime(2024, 1, 1, 1, tzinfo=dt.timezone.utc) + +# Two distinct process-type IRIs for grouping tests. +EVAC_TYPE = "Category:OSW000000000000000000000000000000e1" +HEAT_TYPE = "Category:OSW000000000000000000000000000000e2" + + +# -- Fixtures -- + + +def _make_tool(name, channels): + u = uuid5(NAMESPACE_URL, name) + return DataTool( + uuid=u, + name=name, + label=[Label(text=name)], + data_channels=[ + DataChannel( + uuid=str(compute_scoped_uuid(u, cn)), + osw_id="placeholder", + name=cn, + label=[Label(text=cn)], + characteristic=char.get_cls_iri(), + ) + for cn, char in channels + ], + storage_locations=[Database(name=name + "db", label=[Label(text="db")])], + ) + + +@pytest.fixture +def controllers(): + # Two DataTools of the same (default) datatool type. Both have a `temp` + # (Temperature) channel; pressures differ in name (pressure_x / pressure_y). + t1 = _make_tool("ToolA", [("temp", Temperature), ("pressure_x", Pressure)]) + t2 = _make_tool("ToolB", [("temp", Temperature), ("pressure_y", Pressure)]) + return [ + DataToolController(t1, auto_archive=True), + DataToolController(t2, auto_archive=True), + ] + + +@pytest.fixture +def objects(): + return [ + Item(uuid=uuid5(NAMESPACE_URL, "S1"), label=[Label(text="Sample 1")]), + Item(uuid=uuid5(NAMESPACE_URL, "S2"), label=[Label(text="Sample 2")]), + ] + + +def _make_process(name, sample, tools, type_iri=EVAC_TYPE, start=START, end=END): + return Process( + uuid=uuid5(NAMESPACE_URL, name), + label=[Label(text=name)], + type=[type_iri], + input=[sample], + tool=list(tools), + start_date_time=start, + end_date_time=end, + ) + + +# -- IRI matching -- + + +class TestIriMatching: + def test_process_input_and_tool_refs(self, objects, controllers): + s1 = objects[0] + proc = _make_process("P1", s1, [controllers[0]]) + assert entity_iri(s1) in iri_refs(proc, "input") + assert entity_iri(controllers[0]) in iri_refs(proc, "tool") + + def test_iri_refs_empty_for_missing(self, objects): + s1 = objects[0] + assert iri_refs(s1, "tool") == [] + + +# -- Concrete tree -- + + +class TestConcreteTree: + def test_basic_qualification(self, objects, controllers): + s1, s2 = objects + procs = [ + _make_process("P1", s1, controllers), + _make_process("P2", s2, [controllers[0]]), + ] + tree = build_concrete_tree(objects, procs, controllers) + assert len(tree[entity_iri(s1)]["processes"]) == 1 + assert len(tree[entity_iri(s2)]["processes"]) == 1 + # S1's process has both controllers attached + assert len(tree[entity_iri(s1)]["processes"][0]["controllers"]) == 2 + + def test_skip_missing_end_time(self, objects, controllers, caplog): + s1 = objects[0] + proc = _make_process("Pbad", s1, [controllers[0]], end=None) + with caplog.at_level(logging.WARNING): + tree = build_concrete_tree([s1], [proc], controllers) + assert tree[entity_iri(s1)]["processes"] == [] + assert any("missing start/end" in r.message for r in caplog.records) + + def test_skip_missing_start_time(self, objects, controllers): + s1 = objects[0] + proc = _make_process("Pbad", s1, [controllers[0]], start=None) + tree = build_concrete_tree([s1], [proc], controllers) + assert tree[entity_iri(s1)]["processes"] == [] + + def test_non_datatool_tool_skipped(self, objects, controllers, caplog): + """A tool that is not among the provided controllers is skipped.""" + s1 = objects[0] + stranger = Item(uuid=uuid5(NAMESPACE_URL, "Hammer"), label=[Label(text="H")]) + # Process tools = [non-controller, real controller] + proc = _make_process("P1", s1, [stranger, controllers[0]]) + with caplog.at_level(logging.INFO): + tree = build_concrete_tree([s1], [proc], controllers) + pe = tree[entity_iri(s1)]["processes"] + assert len(pe) == 1 + # Only the real controller remains + assert pe[0]["controllers"] == [controllers[0]] + assert any("not a DataTool controller" in r.message for r in caplog.records) + + def test_process_with_no_datatool_skipped(self, objects, controllers, caplog): + s1 = objects[0] + stranger = Item(uuid=uuid5(NAMESPACE_URL, "Hammer"), label=[Label(text="H")]) + proc = _make_process("P1", s1, [stranger]) + with caplog.at_level(logging.WARNING): + tree = build_concrete_tree([s1], [proc], controllers) + assert tree[entity_iri(s1)]["processes"] == [] + assert any("no DataTool attached" in r.message for r in caplog.records) + + def test_object_not_input_excluded(self, objects, controllers): + s1, s2 = objects + proc = _make_process("P1", s1, controllers) # only s1 is input + tree = build_concrete_tree(objects, [proc], controllers) + assert len(tree[entity_iri(s1)]["processes"]) == 1 + assert len(tree[entity_iri(s2)]["processes"]) == 0 + + +# -- Aggregation -- + + +def _by_name(grp, name): + return [a for a in grp["channels"].values() if a["channel_name"] == name] + + +class TestAggregation: + def test_copresent_splits_into_per_instance(self, objects, controllers): + """Two same-type datatools running together -> per-instance entries.""" + s1 = objects[0] + proc = _make_process("P1", s1, controllers) # both tools, both have temp + tree = build_concrete_tree([s1], [proc], controllers) + agg = derive_aggregated_channels(tree) + grp = next(iter(agg.values())) + temps = _by_name(grp, "temp") + assert len(temps) == 2 # one entry per co-present datatool + assert all(not t["aggregated"] for t in temps) + assert all(t["n_channels"] == 1 for t in temps) + # distinct datatool instances + iris = {t["datatool_iris"][0] for t in temps} + assert iris == {entity_iri(c) for c in controllers} + + def test_distinct_name_merges_free(self, objects, controllers): + """A channel present on only one datatool stays a single merged entry.""" + s1 = objects[0] + proc = _make_process("P1", s1, controllers) + tree = build_concrete_tree([s1], [proc], controllers) + agg = derive_aggregated_channels(tree) + grp = next(iter(agg.values())) + px = _by_name(grp, "pressure_x") + assert len(px) == 1 + assert px[0]["aggregated"] is True + assert px[0]["n_channels"] == 1 + + def test_dropin_merge_across_runs(self, objects, controllers): + """Two same-type datatools never run together -> one merged entry.""" + s1 = objects[0] + procs = [ + _make_process("Run1", s1, [controllers[0]]), + _make_process("Run2", s1, [controllers[1]]), + ] + tree = build_concrete_tree([s1], procs, controllers) + agg = derive_aggregated_channels(tree) + grp = next(iter(agg.values())) + temps = _by_name(grp, "temp") + assert len(temps) == 1 # merged drop-in entry + merged = temps[0] + assert merged["aggregated"] is True + assert merged["n_channels"] == 2 + assert set(merged["datatool_iris"]) == {entity_iri(c) for c in controllers} + assert "[2 channels]" in merged["label"] + assert len(merged["aggregated_channels"]) == 2 + + def test_grouped_by_process_type(self, objects, controllers): + s1 = objects[0] + procs = [ + _make_process("Pevac", s1, [controllers[0]], type_iri=EVAC_TYPE), + _make_process("Pheat", s1, [controllers[0]], type_iri=HEAT_TYPE), + ] + tree = build_concrete_tree([s1], procs, controllers) + agg = derive_aggregated_channels(tree) + assert len(agg) == 2 # two process types + + def test_aggregated_channel_fields(self, objects, controllers): + s1 = objects[0] + proc = _make_process("P1", s1, [controllers[0]]) + tree = build_concrete_tree([s1], [proc], controllers) + agg = derive_aggregated_channels(tree) + grp = next(iter(agg.values())) + temp = _by_name(grp, "temp")[0] + assert temp["characteristic_iri"] == Temperature.get_cls_iri() + assert temp["value_type"] == "quantity" + assert "temp" in temp["label"] + + +# -- Resolution -- + + +class TestResolution: + def test_per_instance_resolves_only_its_datatool(self, objects, controllers): + """A per-instance entry resolves only to its own datatool's channel.""" + s1 = objects[0] + proc = _make_process("P1", s1, controllers) # both tools co-present + tree = build_concrete_tree([s1], [proc], controllers) + agg = derive_aggregated_channels(tree) + grp = next(iter(agg.values())) + entry_a = [ + t + for t in _by_name(grp, "temp") + if t["datatool_iris"] == [entity_iri(controllers[0])] + ][0] + resolved = resolve_aggregated_channel(tree[entity_iri(s1)], entry_a) + assert len(resolved) == 1 + ctrl, ch, _proc = resolved[0] + assert entity_iri(ctrl) == entity_iri(controllers[0]) + assert ch.name == "temp" + + def test_per_instance_fans_out_across_runs(self, objects, controllers): + """A co-present datatool used in two runs -> one trace per run.""" + s1 = objects[0] + procs = [ + _make_process("Run1", s1, controllers), + _make_process("Run2", s1, controllers), + ] + tree = build_concrete_tree([s1], procs, controllers) + agg = derive_aggregated_channels(tree) + grp = next(iter(agg.values())) + entry_a = [ + t + for t in _by_name(grp, "temp") + if t["datatool_iris"] == [entity_iri(controllers[0])] + ][0] + resolved = resolve_aggregated_channel(tree[entity_iri(s1)], entry_a) + assert len(resolved) == 2 # two runs, same datatool + assert len({entity_iri(p) for _c, _ch, p in resolved}) == 2 + + def test_merged_resolves_all_free_instances(self, objects, controllers): + """A merged drop-in entry resolves to each datatool in its own run.""" + s1 = objects[0] + procs = [ + _make_process("Run1", s1, [controllers[0]]), + _make_process("Run2", s1, [controllers[1]]), + ] + tree = build_concrete_tree([s1], procs, controllers) + agg = derive_aggregated_channels(tree) + grp = next(iter(agg.values())) + merged = _by_name(grp, "temp")[0] + resolved = resolve_aggregated_channel(tree[entity_iri(s1)], merged) + assert len(resolved) == 2 + assert {entity_iri(c) for c, _ch, _p in resolved} == { + entity_iri(c) for c in controllers + } + + def test_resolve_distinct_channel(self, objects, controllers): + s1 = objects[0] + proc = _make_process("P1", s1, controllers) + tree = build_concrete_tree([s1], [proc], controllers) + agg = derive_aggregated_channels(tree) + grp = next(iter(agg.values())) + px = _by_name(grp, "pressure_x")[0] + resolved = resolve_aggregated_channel(tree[entity_iri(s1)], px) + assert len(resolved) == 1 # only ToolA has pressure_x + + +# -- Tree sources -- + + +class TestTreeSources: + def test_object_tree_source(self, objects, controllers): + s1 = objects[0] + proc = _make_process("P1", s1, controllers) + tree = build_concrete_tree(objects, [proc], controllers) + source = build_object_tree_source(tree) + assert len(source) == 2 + keys = [n["key"] for n in source] + assert entity_iri(s1) in keys + s1_node = [n for n in source if n["key"] == entity_iri(s1)][0] + assert s1_node["processes"] == "1" + assert s1_node["checkbox"] is True + + def test_process_tree_source(self, objects, controllers): + s1 = objects[0] + proc = _make_process("P1", s1, controllers) + tree = build_concrete_tree([s1], [proc], controllers) + agg = derive_aggregated_channels(tree) + source = build_process_tree_source(agg) + assert len(source) == 1 # one process type root + root = source[0] + # temp co-present -> 2 per-instance entries; pressure_x, pressure_y + # each merged -> 4 children total + assert len(root["children"]) == 4 + assert all(c["checkbox"] for c in root["children"]) + + def test_get_selected_keys(self, objects, controllers): + s1 = objects[0] + proc = _make_process("P1", s1, controllers) + tree = build_concrete_tree([s1], [proc], controllers) + agg = derive_aggregated_channels(tree) + proc_source = build_process_tree_source(agg) + # Select first aggregated channel + proc_source[0]["children"][0]["selected"] = True + keys = get_selected_keys(proc_source) + assert len(keys) == 1 + assert keys[0] == proc_source[0]["children"][0]["key"] + + # Object tree (flat, childless nodes) + obj_source = build_object_tree_source(tree) + obj_source[0]["selected"] = True + obj_keys = get_selected_keys(obj_source) + assert obj_keys == [obj_source[0]["key"]]