"""Dashboards: what a wall panel shows, and what its buttons do. A dashboard is its own document, not a set of nodes placed in a flow. Widgets bind to message names — the same names that wire the graph — so a dashboard reads across flows without being part of any of them, and a flow stays the logic it was. Stored beside the flows in the same git repository, under a directory the flow listing ignores. Editing is separated from showing, exactly as it is for flows: the editor writes ``dashboard.draft.json`` and a wall panel reads only the published ``dashboard.json``, so a half-arranged page never reaches the wall. Publishing promotes the draft and removes it; a dashboard directory without one is simply a dashboard with nothing unpublished. A new dashboard starts as a draft alone, so a directory may just as well hold only the draft — a dashboard nobody has published yet, which no panel can be shown. """ from __future__ import annotations import threading from pathlib import Path from typing import Any, Literal from pydantic import BaseModel, Field, field_validator, model_validator from fluksio.flow.messages import DType, qualify from fluksio.flow.schemas import FlowDef, _validate_name from fluksio.flow.store import FlowStore, StaleVersion #: Sibling of the shared-node library, and likewise not a flow. DASHBOARD_DIR = "_dashboards" #: A chart cannot ask for an unbounded series; this is the ceiling. HISTORY_CAP = 5000 #: How many runs one pinned chart puts side by side. The client's own #: ``MAX_SERIES``: the palette is five steps, and a sixth line repeats one. RUN_LINES = 5 #: How many readings one bar draws. Mirrored in the client #: (``ui/core/config.ts``, ``MAX_ROWS``). #: #: It used to be three, and the limit was contrast: the readings were nested #: inside one fill and shared a single token, because no slot of the chart ramp #: cleared 3:1 against the outer one. A bar now draws a row per reading in the #: dashboard's own data colours, so what bounds it is how many tracks stay #: legible stacked in one tile. BAR_ROWS = 8 #: Resolved out here on purpose: the store has a ``list`` method, which #: shadows the builtin for any annotation written inside the class. Bindings = list[dict[str, Any]] WidgetType = Literal[ # Display "stat", "gauge", "chart", "markdown", "agenda", "notification", "bar", "icon", "forecast", "clock", # Input "button", "switch", "slider", "input", "dropdown", "color", ] INPUT_WIDGETS = {"button", "switch", "slider", "input", "dropdown", "color"} #: What a colour widget puts on the wire, by the format it was configured for. #: The default is what the Node-RED installation this ports already sends its #: DMX encoders — ``[h, s, v]``, hue in degrees and the other two in percent — #: and the two alternatives exist because fixtures differ. Mirrored in the #: client (``frontend/src/components/Dashboard/ColorWidget.tsx``). COLOR_DTYPES = {"hsv": "list", "rgb": "list", "hex": "str"} #: What a widget may be pointed at, by payload type. A switch that reads a #: float has nothing to show and nothing safe to send, so the pairing belongs #: to the document rather than to the editor that happened to write it. Types #: missing here take anything. Mirrored in the client #: (``frontend/src/components/Dashboard/widgets.tsx``). WIDGET_DTYPES: dict[str, set[str]] = { "gauge": {"float", "int"}, # A chart reading the engine's ring. One that queries binds a `series` # answer and a `record` request instead, checked separately below. "chart": {"float", "int"}, "slider": {"float", "int"}, "switch": {"bool"}, "agenda": {"list"}, "notification": {"record"}, "bar": {"float", "int"}, "forecast": {"list"}, # Either shape a colour can travel as; which of the two this widget means # is its ``format``, checked against ``COLOR_DTYPES`` below. "color": {"list", "str"}, # An icon maps weather strings, bool hints and numbers alike, and a clock # binds nothing at all, so neither has a row to be held to. } #: What a dashboard-wide setting may be bound to, by payload type. The channel #: is general — a setting is a value plus an optional binding — but the wired #: ones are a closed set, and a name missing here is simply a setting this #: build does not act on. Mirrored in the client #: (``frontend/src/components/Dashboard/settings.tsx``). SETTING_DTYPES: dict[str, str] = { # "system" | "light" | "dark". A panel in a room has no way to set the # device preference the app otherwise inherits. Idle while a palette is # set: that names the ground, so which way it reads is already decided. "theme": "str", # Read-only: the input widgets stop publishing. "locked": "bool", # "fluksio" | "material" | "glass": which component set draws this # dashboard. Anything else is "fluksio", the app's own design. The three # share every feature; only the drawing differs. "look": "str", # The dashboard's own colours, as an ordered list of hex strings. Position # is the role: background, surface, primary, accent, text, and anything # after that is another colour for a chart. Trailing roles may be left off # and are derived from what is there. "palette": "list", # An image drawn under the widgets, by URL. A flow publishing to it is what # a wallpaper that changes looks like here. "background": "str", # Bigger controls and no hover states, for a panel that is touched. "touch": "bool", } class SettingDef(BaseModel): """One dashboard-wide setting: a value, and optionally where it comes from. Unbound — no ``message`` — the setting is simply ``value``, which is what makes a panel that is always dark cost no flow at all. Bound, a flow drives it live and ``value`` is the fallback: what the dashboard uses until something arrives, and whenever the message is silent. A schedule is not a third case. A node publishing to the bound message on a cron *is* the schedule, which is the whole reason this is a channel rather than a switching rule per setting. """ value: Any = None #: The message that drives it, or empty for a setting that is just a value. message: str = "" #: The payload type the editor recorded when it bound that message, so the #: pairing can be judged from the document alone — the rule widget bindings #: are held to. dtype: str = "" class Placement(BaseModel): """Where a widget sits in its section's grid, in grid units.""" x: int = 0 y: int = 0 w: int = 3 h: int = 2 class WidgetDef(BaseModel): """One tile: what it shows or does, and where it sits. ``config`` is per type — a chart names its series, a button names the message it publishes — and is validated against the type below rather than by a schema per class, because the whole set is small and closed. A chart comes in two kinds. The default reads what the engine kept for a message. One with ``source: "query"`` asks instead, and its config is ``{source, request, request_dtype: "record", message, dtype: "series", refresh_s, range_s}``: it publishes ``{range_s, interval_s}`` to ``request`` exactly as a slider publishes a value, and draws the ``series`` a flow answers with on ``message``. A colour widget picks what it publishes with ``format``, because fixtures differ and a change node per tile is not the answer: - ``hsv`` (the default) — ``[h, s, v]``, hue 0-360 degrees, saturation and value 0-100 percent. What the Node-RED installation this ports feeds its 3CH/4CH DMX encoders, which divide by 360 and by 100. - ``rgb`` — ``[r, g, b]``, each 0-255. The conventional range; the reference's own encoders produce it after converting. - ``hex`` — ``"#rrggbb"``, lowercase. Conventional likewise. The first two are a ``list`` message, the third a ``str``, which is what ``COLOR_DTYPES`` records and the check below holds a binding to. """ id: str type: WidgetType title: str = "" #: Keyed by breakpoint (``lg``/``md``/``sm``); missing ones are derived by #: the client from the widest one it has. layout: dict[str, Placement] = Field(default_factory=dict) config: dict[str, Any] = Field(default_factory=dict) @field_validator("id") @classmethod def _check_id(cls, value: str) -> str: return _validate_name(value) @property def _query_chart(self) -> bool: """A chart that asks a flow for its series instead of reading the ring.""" return self.type == "chart" and self.config.get("source") == "query" @property def _runs_chart(self) -> bool: """A chart pinned to runs: it reads the run tables, not the engine. The other way round from opening a dashboard against a run — that is a way of looking at a page, this is a tile that always shows the last few runs of something, which is what a wall panel over a lab bench wants. """ return self.type == "chart" and self.config.get("source") == "runs" @property def inner_bindings(self) -> Bindings: """A bar's nested readings, as documents written before rows carry them. One binding beside ``inner_dtype``, as a bar was written before it stacked, or an ordered list of ``{message, dtype}``. """ inner = self.config.get("inner") if isinstance(inner, list): return [s for s in inner if isinstance(s, dict)] dtype = self.config.get("inner_dtype") # A recorded type with nothing bound is still a type to be held to. if inner or dtype: return [{"message": inner or "", "dtype": dtype}] return [] @property def bar_rows(self) -> Bindings: """The readings a bar draws, in every shape a document carries them. Current documents write ``rows``. Before that a bar drew one reading with up to three nested inside it, which is read here as that reading followed by the nested ones — the same set of messages, drawn as separate tracks — so an older dashboard keeps working without being migrated first. """ rows = self.config.get("rows") if isinstance(rows, list): return [row for row in rows if isinstance(row, dict)] name = self.config.get("message") outer: Bindings = ( [{"message": name, "dtype": self.config.get("dtype")}] if name else [] ) return [*outer, *self.inner_bindings] @property def messages(self) -> list[str]: """Every message name this widget reads.""" if self._runs_chart: # Its metric is a run's recorded series, which no live message # carries — nothing for the engine to route or to keep. return [] if self._query_chart: name = self.config.get("message") return [str(name)] if name else [] if self.type == "chart": return [ str(series.get("message")) for series in self.config.get("series") or [] if series.get("message") ] if self.type == "bar": # A bar draws a row per reading, and each row binds its own. return [ str(row.get("message")) for row in self.bar_rows if row.get("message") ] name = self.config.get("message") return [str(name)] if name else [] @property def target(self) -> str: """The message this widget publishes, if it is an input. A querying chart is one too: its request is a value it puts into the graph, so the canvas draws it as an endpoint like any other control. """ if self._query_chart: return str(self.config.get("request") or "") return str(self.config.get("target") or "") @property def history_points(self) -> int: """How much past this widget needs kept for it. Nothing, for a chart that queries: the answer carries its own past, so asking the engine to keep a ring as well would store it twice. """ if self.type != "chart" or self._query_chart or self._runs_chart: return 0 points = int((self.config.get("history") or {}).get("points") or 0) return min(points, HISTORY_CAP) @property def bound_dtypes(self) -> list[str]: """The payload types this widget was bound to, as the editor recorded. Empty for a document written before the editor kept them, which is why a missing type is never an error. """ if self.type == "chart": return [ str(series.get("dtype") or "") for series in self.config.get("series") or [] ] if self.type == "bar": return [str(row.get("dtype") or "") for row in self.bar_rows] return [str(self.config.get("dtype") or "")] @model_validator(mode="after") def _check_binding(self) -> WidgetDef: """Refuse a widget wired to a message it cannot carry.""" if self._runs_chart: runs = self.config.get("runs") or {} if not runs.get("metric"): raise ValueError("a chart of runs must name the metric it draws") if not (runs.get("ids") or runs.get("group") or runs.get("flow")): raise ValueError( "a chart of runs must say which: a flow, a sweep, or run ids" ) latest = int(runs.get("latest") or 1) if not 1 <= latest <= RUN_LINES: raise ValueError(f"a chart draws between 1 and {RUN_LINES} runs") if len(runs.get("ids") or []) > RUN_LINES: raise ValueError(f"a chart draws at most {RUN_LINES} runs") return self if self._query_chart: for key, want in (("dtype", "series"), ("request_dtype", "record")): bound = str(self.config.get(key) or "") if bound and bound != want: raise ValueError( f"a querying chart's {key} must be '{want}', not '{bound}'" ) return self if self.type == "bar" and len(self.bar_rows) > BAR_ROWS: raise ValueError(f"a bar draws at most {BAR_ROWS} readings") if self.type == "color": # The row above allows both shapes a colour travels as; the format # is what decides which of them this widget actually sends. An # unknown one is read as the default, exactly as the client does. fmt = str(self.config.get("format") or "hsv") want = COLOR_DTYPES.get(fmt, "list") bound = str(self.config.get("dtype") or "") if bound and bound != want: raise ValueError( f"a colour widget sending {fmt} needs a '{want}' message, " f"not a '{bound}'" ) allowed = WIDGET_DTYPES.get(self.type) if not allowed: return self for dtype in self.bound_dtypes: if dtype and dtype not in allowed: raise ValueError( f"a '{self.type}' widget cannot carry a '{dtype}' message" ) return self def _placement(widget: dict[str, Any]) -> dict[str, Any]: """Where a stored widget sits, by the widest breakpoint it names.""" layout = widget.get("layout") or {} for key in ("lg", "md", "sm"): box = layout.get(key) if isinstance(box, dict): return dict(box) return {} def _flatten_pages(pages: list[Any]) -> list[dict[str, Any]]: """The widgets of a document written as pages and sections. Only the first page: no UI ever wrote a second one, and a panel carries several whole dashboards instead. Its sections are stacked into one grid the way the viewer always drew them, so a document that placed its widgets keeps the arrangement it had rather than piling everything at row zero. """ if not pages or not isinstance(pages[0], dict): return [] sections = [s for s in (pages[0].get("sections") or []) if isinstance(s, dict)] lists = [ [w for w in (s.get("widgets") or []) if isinstance(w, dict)] for s in sections ] flat = [w for widgets in lists for w in widgets] placed = any( (_placement(w).get("x") or 0) > 0 or (_placement(w).get("y") or 0) > 0 for w in flat ) if len(sections) < 2 or not placed: return flat stacked: list[dict[str, Any]] = [] offset = 0 for widgets in lists: bottom = 0 for widget in widgets: box = _placement(widget) y = max(0, int(box.get("y") or 0)) bottom = max(bottom, y + max(1, int(box.get("h") or 2))) if offset: widget = { **widget, "layout": { **(widget.get("layout") or {}), "lg": {**box, "y": y + offset}, }, } stacked.append(widget) offset += bottom return stacked class DashboardDef(BaseModel): """A dashboard as stored, and as the API hands it over.""" name: str title: str = "" #: How many columns the grid is cut into, so a dashboard can be matched to #: the panel it will hang on. columns: int = Field(default=12, ge=1, le=48) #: The panel this dashboard is drawn for, in CSS pixels. Both the editor #: and the wall panel scale that surface to fit whatever room they have, so #: an arrangement does not depend on the window it was made in. Zero means #: "unset" and the client falls back to its default. canvas_width: int = Field(default=1920, ge=0, le=7680) canvas_height: int = Field(default=1080, ge=0, le=4320) #: A lucide icon name, drawn on the panel rail; empty falls back to two #: letters of the title. icon: str = "" #: One grid. Pages and sections were in the schema and never in the UI — #: only the first page was ever read and its sections were drawn as one — #: so a dashboard is its widgets, and several dashboards on one device is #: what a panel is for. widgets: list[WidgetDef] = Field(default_factory=list) #: Settings the whole dashboard carries, by name — see ``SettingDef``. The #: one channel a dashboard consumes as a dashboard rather than as a set of #: tiles, so a screen on a wall can be told things nobody standing at it #: could set. settings: dict[str, SettingDef] = Field(default_factory=dict) #: Bumped on every save; a save based on an older one is refused. version: int = 1 #: Whether there are unpublished changes. Reported by the store on read, #: never stored — the draft file's existence is the only record of it. has_draft: bool = False @model_validator(mode="before") @classmethod def _flatten(cls, data: Any) -> Any: """Read a document written as pages and sections as one grid. Stored dashboards live in each installation's git repository, so the old shape is normalised on the way in rather than migrated: an untouched document keeps working, and the next save writes it flat. """ if isinstance(data, dict) and "widgets" not in data and "pages" in data: pages = data.get("pages") or [] data = {k: v for k, v in data.items() if k != "pages"} data["widgets"] = _flatten_pages(pages) return data @field_validator("name") @classmethod def _check_name(cls, value: str) -> str: return _validate_name(value) @model_validator(mode="after") def _check_settings(self) -> DashboardDef: """Refuse a setting driven by a message it cannot carry. Judged from the document alone, exactly as a widget's binding is: the picker records the payload type beside the name, so neither the editor nor a wall panel has to fetch the catalogue to know the wiring is wrong. A name this build does not know is left alone rather than refused — an older installation reading a newer document simply does not act on it. """ for name, setting in self.settings.items(): want = SETTING_DTYPES.get(name) if want and setting.dtype and setting.dtype != want: raise ValueError( f"the '{name}' setting needs a '{want}' message, " f"not a '{setting.dtype}'" ) return self @property def setting_messages(self) -> list[str]: """Every message a bound setting reads. Empty for a static dashboard.""" return [s.message for s in self.settings.values() if s.message] class DashboardSummary(BaseModel): """A dashboard in a list, without its contents.""" name: str title: str = "" widget_count: int = 0 has_draft: bool = False #: Of the working copy, so publishing from a list needs no second read. version: int = 1 class DashboardsPublic(BaseModel): data: list[DashboardSummary] count: int class DashboardNotFound(KeyError): def __init__(self, name: str) -> None: super().__init__(name) self.name = name class DashboardExists(ValueError): def __init__(self, name: str) -> None: super().__init__(name) self.name = name class DashboardStore: """Dashboards in the flow store's repository, invisible to the flow listing. Shares the flow store's write lock and commit, so a dashboard save and a flow save cannot interleave into one confused commit. """ def __init__(self, flows: FlowStore) -> None: self.flows = flows self.root = flows.root / DASHBOARD_DIR # Read-modify-write of the version counter, same as the flow store. self._lock = threading.Lock() def _file(self, name: str) -> Path: return self.root / name / "dashboard.json" def _draft_file(self, name: str) -> Path: return self.root / name / "dashboard.draft.json" @staticmethod def _dump(defn: DashboardDef) -> str: """What goes on disk. ``has_draft`` is the file layout, not a field.""" return defn.model_dump_json(indent=2, exclude={"has_draft"}) def list(self) -> list[DashboardSummary]: """Every dashboard the editor knows, published or not.""" names = {path.parent.name for path in self.root.glob("*/dashboard.json")} names |= {path.parent.name for path in self.root.glob("*/dashboard.draft.json")} summaries = [] for name in sorted(names): try: defn = self.read(name, draft=True) except Exception: continue summaries.append( DashboardSummary( name=defn.name, title=defn.title, widget_count=len(defn.widgets), has_draft=defn.has_draft, version=defn.version, ) ) return summaries def exists(self, name: str) -> bool: return self._file(name).exists() or self._draft_file(name).exists() def is_published(self, name: str) -> bool: """Is there a document a panel can be shown?""" return self._file(name).exists() def has_draft(self, name: str) -> bool: """Are there unpublished changes to this dashboard?""" return self._draft_file(name).exists() def read(self, name: str, draft: bool = False) -> DashboardDef: """The published dashboard, or with ``draft`` the working copy.""" path = self._draft_file(name) if draft else self._file(name) if not path.exists(): path = self._file(name) if not path.exists(): raise DashboardNotFound(name) return DashboardDef.model_validate_json(path.read_text()).model_copy( update={"has_draft": self.has_draft(name)} ) def write( self, defn: DashboardDef, base_version: int | None = None ) -> DashboardDef: """Publish a dashboard directly, skipping the draft. The API never does: it creates a draft and promotes it. This is for a caller that already has the finished document — a test, or a seed. """ with self._lock, self.flows._write_lock: path = self._file(defn.name) current = 0 if path.exists(): current = DashboardDef.model_validate_json(path.read_text()).version if base_version is not None and base_version != current: raise StaleVersion(defn.name, current) saved = defn.model_copy(update={"version": current + 1, "has_draft": False}) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(self._dump(saved)) self.flows._commit(f"Save dashboard '{defn.name}'") return saved def write_draft( self, defn: DashboardDef, base_version: int | None = None ) -> DashboardDef: """Save unpublished changes, refusing to overwrite someone else's. ``base_version`` is the version the editor last saw — of the working copy, which is the draft once there is one, and 0 for a dashboard that does not exist yet: creating one is its first draft. """ with self._lock, self.flows._write_lock: current = 0 if self.exists(defn.name): current = self.read(defn.name, draft=True).version if base_version is not None and base_version != current: raise StaleVersion(defn.name, current) saved = defn.model_copy(update={"version": current + 1, "has_draft": True}) path = self._draft_file(defn.name) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(self._dump(saved)) self.flows._commit(f"Update draft of dashboard '{defn.name}'") return saved def publish(self, name: str, base_version: int | None = None) -> DashboardDef: """Promote the working copy to what the panels show.""" with self._lock, self.flows._write_lock: current = self.read(name, draft=True) if base_version is not None and base_version != current.version: raise StaleVersion(name, current.version) draft = self._draft_file(name) if draft.exists(): self._file(name).write_text(self._dump(current)) draft.unlink() self.flows._commit(f"Publish dashboard '{name}'") return current.model_copy(update={"has_draft": False}) def discard_draft(self, name: str) -> DashboardDef: """Throw the unpublished changes away and go back to what is shown.""" with self._lock, self.flows._write_lock: draft = self._draft_file(name) if draft.exists(): draft.unlink() self.flows._commit(f"Discard draft of dashboard '{name}'") return self.read(name) def delete(self, name: str) -> None: if not self.exists(name): raise DashboardNotFound(name) path = self._file(name) with self.flows._write_lock: path.unlink(missing_ok=True) self._draft_file(name).unlink(missing_ok=True) try: path.parent.rmdir() except OSError: pass self.flows._commit(f"Delete dashboard '{name}'") def rename(self, name: str, new_name: str) -> DashboardDef: defn = self.read(name, draft=True) if self.exists(new_name): raise DashboardExists(new_name) with self.flows._write_lock: renamed = defn.model_copy(update={"name": new_name}) self._file(new_name).parent.mkdir(parents=True, exist_ok=True) # Whichever files the dashboard has move; one nobody published yet # has only the draft, and renaming it must not publish it. if self.is_published(name): published = self.read(name).model_copy(update={"name": new_name}) self._file(new_name).write_text(self._dump(published)) self._file(name).unlink() # An unpublished edit belongs to the dashboard, so it moves too. if self.has_draft(name): self._draft_file(new_name).write_text(self._dump(renamed)) self._draft_file(name).unlink() try: self._file(name).parent.rmdir() except OSError: pass self.flows._commit(f"Rename dashboard '{name}' to '{new_name}'") return renamed def bindings_for(self, flow: str) -> Bindings: """Every widget bound to a message of ``flow``. What the canvas draws as an endpoint: a control that sets one of this flow's messages, or a tile that shows one. Without this a dashboard is an invisible participant — a value changes and nothing on the canvas accounts for it. """ prefix = f"{flow}." found: Bindings = [] for path in sorted(self.root.glob("*/dashboard.json")): try: defn = DashboardDef.model_validate_json(path.read_text()) except Exception: continue # A bound setting is a consumer too — the dashboard itself reading # a message rather than any tile on it — so the canvas accounts for # it the same way. ``widget`` is what the endpoint id is built # from, and no widget id can collide with it: a dot is not a legal # name character. for name, setting in defn.settings.items(): if not setting.message.startswith(prefix): continue found.append( { "dashboard": defn.name, "dashboard_title": defn.title or defn.name, "widget": f"settings.{name}", "title": f"{defn.title or defn.name} {name}", "type": "setting", "provides": "", "requires": [setting.message], } ) for widget in defn.widgets: # A control produces the message; a tile consumes it. produces = widget.target if widget.target.startswith(prefix) else "" consumes = [m for m in widget.messages if m.startswith(prefix)] if not produces and not consumes: continue found.append( { "dashboard": defn.name, "dashboard_title": defn.title or defn.name, "widget": widget.id, "title": widget.title or widget.id, "type": widget.type, "provides": produces, "requires": consumes, } ) return found def history_requirements(self) -> dict[str, int]: """How many points to keep per message, so charts have a past to draw. The deepest chart bound to a message wins; a message no chart reads keeps the default. """ limits: dict[str, int] = {} for path in self.root.glob("*/dashboard.json"): try: defn = DashboardDef.model_validate_json(path.read_text()) except Exception: continue for widget in defn.widgets: points = widget.history_points if not points: continue for message in widget.messages: limits[message] = max(limits.get(message, 0), points) return limits def default_dashboard(name: str) -> DashboardDef: """A new dashboard: one page, one section, nothing in it yet.""" return DashboardDef( name=name, title=name.replace("_", " ").capitalize(), ) #: What the generated dashboard is called, for a flow of this name. def results_name(flow: str) -> str: return f"{flow}_results" #: Numbers a stat or a chart can draw. _NUMERIC = {DType.FLOAT, DType.INT} def results_dashboard(flow: FlowDef) -> DashboardDef: """A results dashboard for a batch flow, from the ports it declares. A streaming output is a curve and gets a chart; a scalar output is a number and gets a stat. Nothing here is specific to runs: the widgets bind to the flow's own message names, which is what makes the same page draw a run live, draw a finished one when opened in a run's context, and stay an ordinary dashboard anyone can edit afterwards. A starting point rather than a finished page — which is the only reason generating one is worth doing at all. """ charts = [ spec for node in flow.nodes for spec in node.provides if spec.stream and spec.dtype in _NUMERIC and spec.name ] # What a run reports. Declared outputs are unqualified names; an empty list # means "everything the flow ends up holding", which is not a set this can # enumerate, so it draws no stats rather than guessing at them. produced = { spec.name.rsplit(".", 1)[-1]: spec for node in flow.nodes for spec in node.provides if spec.name } stats = [ produced[name] for name in flow.outputs if name in produced and not produced[name].stream ] widgets: list[WidgetDef] = [] for index, spec in enumerate(charts): widgets.append( WidgetDef( id=f"chart_{spec.port or index}", type="chart", title=spec.port or spec.name, layout={"lg": Placement(x=0, y=index * 4, w=8, h=4)}, config={ "series": [ { "message": qualify(flow.name, spec.name), "dtype": spec.dtype.value, "label": spec.port or spec.name, } ], "history": {"points": 600}, }, ) ) row = 0 for spec in stats: if spec.dtype is DType.RECORD: kind: WidgetType = "notification" elif spec.dtype in _NUMERIC or spec.dtype is DType.STR: kind = "stat" else: # A list, a series or an artifact has no single reading to show. continue widgets.append( WidgetDef( id=f"out_{spec.port}", type=kind, title=spec.port, layout={"lg": Placement(x=8, y=row * 2, w=4, h=2)}, config={ "message": qualify(flow.name, spec.name), "dtype": spec.dtype.value, }, ) ) row += 1 return DashboardDef( name=results_name(flow.name), title=f"{flow.title or flow.name} results", widgets=widgets, ) __all__ = [ "BAR_ROWS", "RUN_LINES", "COLOR_DTYPES", "DASHBOARD_DIR", "HISTORY_CAP", "INPUT_WIDGETS", "SETTING_DTYPES", "WIDGET_DTYPES", "DashboardDef", "DashboardExists", "DashboardNotFound", "DashboardStore", "DashboardSummary", "DashboardsPublic", "Placement", "SettingDef", "WidgetDef", "default_dashboard", ]