"""Runs beside each other: one metric's curves, and what differs between them. The engine already answers both halves. `GET /runs/series/compare` returns the chart's own series shape for any number of runs, and `GET /runs/export/runs` returns one row per run with every input and final number the selection recorded as its own column — the table this screen keeps the varying half of. So nothing here computes what a route already knows. """ from __future__ import annotations import json from typing import Any from rich.text import Text from textual import work from textual.app import ComposeResult from textual.containers import Horizontal from textual.screen import Screen from textual.widgets import DataTable, Footer, Header, Select, Static from fluksio.sdk.cli import _dur from fluksio.sdk.client import Client from fluksio.tui.chart import MAX_SERIES, Curves, legend #: The axes every comparison offers, before the run's own metrics are added. STEP_AXIS, TIME_AXIS = "step", "time" #: What the columns of `export runs` are called that are not a run's own. PARAM, METRIC = "param.", "metric." #: The statuses that mean a run's numbers are not final yet. RUNNING = frozenset({"queued", "running"}) #: How often a curve is re-read while a run in the selection is still going. #: The engine announces each batch its sink writes (`run_metric`), which is #: what usually wakes this screen; the poll is the heartbeat behind it, and #: what makes the screen live against an engine older than that event. LIVE_S = 1.0 def _cell(value: Any) -> str: """A reading narrow enough for a column, at the precision one reads at. Four significant figures, which is what the browser's legend settles on: the seventeen digits a float prints are a column nobody can scan. """ if isinstance(value, float): return f"{value:.4g}" if isinstance(value, (dict, list)): # An export keeps a record a value now, and python's repr of one is # not what anybody reading a run wrote down. return json.dumps(value) return "" if value is None else str(value) def varying(rows: list[dict[str, Any]], prefix: str) -> list[str]: """The prefixed columns whose value is not the same in every row. What a reader is looking for across a page of runs is where they differ; a sweep over one parameter should not draw twelve identical columns. """ if len(rows) < 2: return sorted(key for key in rows[0] if key.startswith(prefix)) if rows else [] return sorted( key for key in rows[0] if key.startswith(prefix) and len({str(row.get(key)) for row in rows}) > 1 ) class CompareScreen(Screen[None]): """One metric across the picked runs, over the table of what differs.""" CSS = """ #head { padding: 0 1; } #pickers { height: 3; padding: 0 1; } #pickers Select { width: 34; } #legend { padding: 0 1; height: auto; } #diff { height: auto; max-height: 12; } """ BINDINGS = [ ("escape", "app.pop_screen", "back"), ("r", "reload", "reload"), ] def __init__(self, client: Client, ids: list[str]) -> None: super().__init__() self.client = client self.ids = ids self.metric = "" self.x = STEP_AXIS self.names: list[str] = [] #: What the metric picker is currently offering. Rebuilding it on every #: refresh would shut a dropdown under whoever opened it. self.names_shown: list[str] = [] self.live: Any = None def compose(self) -> ComposeResult: yield Header() yield Static(id="head") with Horizontal(id="pickers"): yield Select([], id="metric", prompt="metric", allow_blank=True) yield Select( [("step", STEP_AXIS), ("time (s)", TIME_AXIS)], id="x", value=STEP_AXIS, allow_blank=False, ) yield Curves(id="chart") yield Static(id="legend") table: DataTable[Any] = DataTable(id="diff", cursor_type="row") yield table yield Footer() def on_mount(self) -> None: self.title = f"{len(self.ids)} runs" self.query_one("#head", Static).update("Reading these runs\u2026") self.read_table() def action_reload(self) -> None: self.read_table() def on_select_changed(self, event: Select.Changed) -> None: if event.value is Select.BLANK: return if event.select.id == "metric": self.metric = str(event.value) else: self.x = str(event.value) self.read_curves() # -- reading --------------------------------------------------------------- @work(thread=True, exclusive=True, group="table") def read_table(self) -> None: try: rows = self.client.export_runs(ids=self.ids) names = list(self.client.metric_names(ids=self.ids)) except Exception as exc: # noqa: BLE001 — the screen reports it self.app.call_from_thread(self.show_error, exc) return self.app.call_from_thread(self.show_table, rows, names) @work(thread=True, exclusive=True, group="curves") def read_curves(self) -> None: if not self.metric: return try: answer = self.client.compare(self.ids[:MAX_SERIES], self.metric, self.x) except Exception as exc: # noqa: BLE001 — the screen reports it self.app.call_from_thread(self.show_error, exc) return self.app.call_from_thread(self.show_curves, answer) # -- drawing --------------------------------------------------------------- def show_error(self, exc: Exception) -> None: self.query_one("#head", Static).update( Text(f"{type(exc).__name__}: {exc}", style="red") ) def show_table(self, rows: list[dict[str, Any]], names: list[str]) -> None: self.names = names flow = str(rows[0].get("flow", "")) if rows else "" drawn = ( f" {MAX_SERIES} of {len(self.ids)} drawn" if len(self.ids) > MAX_SERIES else "" ) self.query_one("#head", Static).update( f"{len(self.ids)} runs \u00b7 {flow}{drawn}" ) picker = self.query_one("#metric", Select) if names and names != self.names_shown: self.names_shown = list(names) picker.set_options([(name, name) for name in names]) self.metric = self.metric if self.metric in names else names[0] picker.value = self.metric axes = self.query_one("#x", Select) axes.set_options( [("step", STEP_AXIS), ("time (s)", TIME_AXIS)] + [(name, name) for name in names if name != self.metric] ) axes.value = ( self.x if self.x in {STEP_AXIS, TIME_AXIS, *names} else STEP_AXIS ) table = self.query_one("#diff", DataTable) table.clear(columns=True) params = varying(rows, PARAM) scores = sorted( key for key in (rows[0] if rows else {}) if key.startswith(METRIC) ) # Not a parameter, but it is part of what produced the number, and in # a sweep it is sometimes the only thing that moved. seeds = len({str(row.get("seed")) for row in rows}) > 1 table.add_columns( "run", "status", "took", *(["seed"] if seeds else []), *(key[len(PARAM) :] for key in params), *(key[len(METRIC) :] for key in scores), ) for row in rows: table.add_row( str(row.get("id", ""))[-8:], str(row.get("status", "")), _dur(row.get("duration_ms")), *([_cell(row.get("seed"))] if seeds else []), *(_cell(row.get(key)) for key in params), *(_cell(row.get(key)) for key in scores), ) # A curve only grows while its run does. self.keep_live(any(str(row.get("status")) in RUNNING for row in rows)) self.read_curves() def show_curves(self, answer: dict[str, Any]) -> None: lines = [ (str(line.get("label", "")), line.get("points") or []) for line in answer.get("lines") or [] ] self.query_one("#chart", Curves).show(lines, str(answer.get("x") or "")) self.query_one("#legend", Static).update( legend(label for label, _ in lines[:MAX_SERIES]) ) def refresh_live(self) -> None: """New readings have landed. Draw them. A run opened before it measured anything has no metric picked, and nothing but the table read fills that in — so the whole read is what a first batch needs, and only the curve after that. """ if not self.names: self.read_table() else: self.read_curves() def keep_live(self, running: bool) -> None: if running and self.live is None: self.live = self.set_interval(LIVE_S, self.refresh_live) elif not running and self.live is not None: self.live.stop() self.live = None