diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 26c5f0c..b4dad6f 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -805,7 +805,7 @@ def _status_screen(client: Client) -> Any: ) + Text( f"{str(row.get('flow', '')):<16}" - f"{(row.get('duration_ms') or 0) / 1000:7.1f}s" + f"{_dur(row.get('duration_ms')):>8}" f" {_ago(row.get('finished_at') or row.get('created_at')):>9}", style="dim", ) @@ -886,6 +886,19 @@ def _declared(client: Client, flows: Iterable[str]) -> dict[str, dict[str, Any]] return known +def _dur(ms: Any) -> str: + """How long something took, in the same notation `_ago` reads an age in. + + A run measured in hours used to print four digits of seconds, because the + three places that formatted a duration each did it their own way. + """ + seconds = max(float(ms or 0.0), 0.0) / 1000 + for span, unit in ((86400, "d"), (3600, "h"), (60, "min")): + if seconds >= span: + return f"{seconds / span:.1f}{unit}" + return f"{seconds:.1f}s" + + def _ago(stamp: Any) -> str: """How long ago something happened, in the notation the screens use. @@ -950,7 +963,7 @@ def cmd_runs(args: argparse.Namespace) -> int: params = params[: room - 3] + "..." _say( f"{row['id']} {_status(row['status'], 9)} {row['flow']:<16} " - f"{row['duration_ms'] / 1000:7.1f}s {_ago(row.get('created_at')):>9} " + f"{_dur(row['duration_ms']):>8} {_ago(row.get('created_at')):>9} " f"{_stamp(row):<22} {params}" ) return 0 @@ -1131,12 +1144,20 @@ def _write_rows(rows: list[dict[str, Any]], fmt: str, out: str, hint: str = "") LIST_SCAN = 10 -def _list_names(client: Client, args: argparse.Namespace) -> int: +def metric_names(client: Client, ids: Iterable[str]) -> list[str]: """The metric names these runs carry, since a name is flow-qualified. `train_loss` is recorded as `train.train_loss`, and asking for the bare one matches nothing — so this is the answer to "what would match". """ + for run_id in list(ids)[:LIST_SCAN]: + names = sorted({point["name"] for point in client.metrics(run_id)}) + if names: + return names + return [] + + +def _list_names(client: Client, args: argparse.Namespace) -> int: filters = _selection(args) if "until" in filters: # The history spells the same bound `before`, where it is also the @@ -1145,12 +1166,8 @@ def _list_names(client: Client, args: argparse.Namespace) -> int: ids = args.run or [ row["id"] for row in client.runs(flow=args.flow, limit=LIST_SCAN, **filters) ] - for run_id in ids[:LIST_SCAN]: - names = sorted({point["name"] for point in client.metrics(run_id)}) - if names: - _say("\n".join(names)) - return 0 - _say("No metrics recorded by these runs.") + names = metric_names(client, ids) + _say("\n".join(names) if names else "No metrics recorded by these runs.") return 0 diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 1f5dc2f..4ddd9ec 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -390,12 +390,17 @@ class Client: query["name"] = name return self._call("GET", f"/runs/{run_id}/metrics", params=query) - def compare(self, ids: Iterable[str], metric: str) -> Any: - return self._call( - "GET", - "/runs/series/compare", - params={"ids": ",".join(ids), "metric": metric}, - ) + def compare(self, ids: Iterable[str], metric: str, x: str = "") -> Any: + """One metric across several runs, in the chart widget's series shape. + + ``x`` is what the readings are plotted against: the step they were + recorded at, "time" for the seconds since each run's own first one, or + another metric of the same runs. + """ + query = {"ids": ",".join(ids), "metric": metric} + if x: + query["x"] = x + return self._call("GET", "/runs/series/compare", params=query) def export_metrics( self, diff --git a/backend/fluksio/sdk/stream.py b/backend/fluksio/sdk/stream.py new file mode 100644 index 0000000..ee4e250 --- /dev/null +++ b/backend/fluksio/sdk/stream.py @@ -0,0 +1,85 @@ +"""The engine's event bus, read the way a browser reads it. + +Polling answers "what is true now", so a run that starts and finishes between +two ticks was only ever a row in the history. This subscribes to the same +websocket the dashboard in a browser does — nothing new is served for it. + +Blocking on purpose: `subscribe` is a thread's whole job, and the caller +brings the thread. Reconnecting is this module's, since a dashboard outlives +the engine it supervises. +""" + +from __future__ import annotations + +import json +import threading +from collections.abc import Callable, Collection +from typing import Any +from urllib.parse import urlsplit, urlunsplit + +#: What a reconnect waits before trying again, growing to a minute — the shape +#: the engine's own supervision backs off with. +BACKOFF = (1.0, 5.0, 30.0, 60.0) + +#: How long a read waits before looking at `stop`. The socket is otherwise +#: silent for as long as the house is, and a dashboard that is closing should +#: not wait for the next event to notice. +POLL_S = 1.0 + + +def socket_url(url: str, token: str) -> str: + """The websocket beside an engine's HTTP address. + + The token goes in the query string because a handshake carries no headers + of its own — the same reason the browser sends it that way. + """ + parts = urlsplit(url) + scheme = "wss" if parts.scheme == "https" else "ws" + return urlunsplit((scheme, parts.netloc, "/api/v1/flows/ws", f"token={token}", "")) + + +def subscribe( + url: str, + token: str, + on_event: Callable[[dict[str, Any]], None], + stop: threading.Event, + kinds: Collection[str] | None = None, + on_error: Callable[[Exception], None] | None = None, +) -> None: + """Hand every published event to `on_event` until `stop` is set. + + `kinds` keeps only the event types named, which is worth doing here rather + than in the caller: `message_value` is most of what the bus carries and a + screen drawing runs wants none of it. + + A failure is reported once per outage. A dashboard is the thing that stops + and starts the engine, so a reconnect loop announcing itself every second + would bury the log pane it shares. + """ + from websockets.sync.client import connect + + attempt = 0 + while not stop.is_set(): + try: + with connect(socket_url(url, token), open_timeout=POLL_S * 5) as socket: + attempt = 0 + while not stop.is_set(): + try: + frame = socket.recv(timeout=POLL_S) + except TimeoutError: + continue + payload = json.loads(frame) + events = ( + payload.get("events") or [] + if payload.get("type") == "batch" + else [payload] + ) + for event in events: + if kinds is None or event.get("type") in kinds: + on_event(event) + return + except Exception as exc: # noqa: BLE001 — any transport failure retries + if attempt == 0 and on_error is not None: + on_error(exc) + stop.wait(BACKOFF[min(attempt, len(BACKOFF) - 1)]) + attempt += 1 diff --git a/backend/fluksio/tui/__init__.py b/backend/fluksio/tui/__init__.py new file mode 100644 index 0000000..5e90887 --- /dev/null +++ b/backend/fluksio/tui/__init__.py @@ -0,0 +1,14 @@ +"""The dashboard `fluksio serve` opens at a terminal. + +A supervisor, not an engine: it starts `serve --plain` as a child and talks to +it over the same HTTP API and the same websocket every other client uses. So +the engine is a process that can be stopped and started under the screen +watching it, an engine somebody else started can be adopted rather than +duplicated, and quitting the dashboard is not the same as stopping the engine. + +Nothing of the engine is imported here — that lives in the child. +""" + +from fluksio.tui.app import ServeApp, child_argv, run_tui + +__all__ = ["ServeApp", "child_argv", "run_tui"] diff --git a/backend/fluksio/tui.py b/backend/fluksio/tui/app.py similarity index 58% rename from backend/fluksio/tui.py rename to backend/fluksio/tui/app.py index a02223a..068bf86 100644 --- a/backend/fluksio/tui.py +++ b/backend/fluksio/tui/app.py @@ -1,12 +1,6 @@ -"""The dashboard `fluksio serve` opens at a terminal. +"""The screen itself: the engine under it, and what it is doing. -A supervisor, not an engine: it starts `serve --plain` as a child and talks to -it over the same HTTP API every other client uses. So the engine is a process -that can be stopped and started under the screen watching it, an engine -somebody else started can be adopted rather than duplicated, and quitting the -dashboard is not the same as stopping the engine. - -Nothing of the engine is imported here — that lives in the child. +See `fluksio.tui` for why the engine is a child process rather than a thread. """ from __future__ import annotations @@ -17,6 +11,7 @@ import os import signal import subprocess import sys +import threading from pathlib import Path from typing import Any @@ -34,13 +29,30 @@ from fluksio.cli import ( probe_engine, read_pidfile, ) -from fluksio.sdk.cli import WATCH_INTERVAL_S, _status_screen +from fluksio.sdk import stream +from fluksio.sdk.cli import WATCH_INTERVAL_S, _dur, _status_screen from fluksio.sdk.client import Client, config_path +from fluksio.tui.compare import CompareScreen #: How long to keep asking the child for the credential it writes on the way #: up, before giving up and letting the panels say so. STARTUP_TRIES = 60 +#: How many runs one comparison holds. The chart draws five of them +#: (`chart.MAX_SERIES`); the table below it carries the rest, which is the +#: split the browser makes for the same reason. +MAX_PICKED = 20 + +#: What the dashboard subscribes to. `message_value` is most of what the bus +#: carries and none of what this screen draws. +KINDS = frozenset( + {"run_started", "run_finished", "flow_paused", "engine_fatal", "flow_changed"} +) + +#: How long an event waits for the ones behind it. A sweep starting twenty +#: runs should cost one read of the engine, not twenty. +COALESCE_S = 0.25 + def child_argv(argv: list[str]) -> list[str]: """The same command, told to serve without a dashboard of its own. @@ -75,6 +87,87 @@ class Enroll(ModalScreen[tuple[str, str] | None]): self.dismiss((code, portal or DEFAULT_PORTAL) if code else None) +class Artifacts(ModalScreen[None]): + """What a run left behind, and one key to fetch it. + + The rows are already on `GET /runs/{id}` — the same list the browser draws + on a run's page — so this is a view of them rather than a second reading. + """ + + BINDINGS = [ + ("escape", "dismiss(None)", "close"), + ("enter", "download", "download"), + ] + + def __init__(self, client: Client, run_id: str) -> None: + super().__init__() + self.client = client + self.run_id = run_id + self.rows: list[dict[str, Any]] = [] + + def compose(self) -> ComposeResult: + with Vertical(id="artifacts"): + yield Label(f"Artifacts of {self.run_id[-8:]}") + table: DataTable[str] = DataTable(id="files", cursor_type="row") + yield table + yield Label("", id="note") + + def on_mount(self) -> None: + table = self.query_one("#files", DataTable) + table.add_columns("name", "node", "type", "bytes") + table.focus() + self.read() + + @work(thread=True) + def read(self) -> None: + try: + detail = self.client.run(self.run_id) + except Exception as exc: # noqa: BLE001 — the modal reports it + self.app.call_from_thread(self.note, f"Could not read the run: {exc}") + return + self.app.call_from_thread(self.show, detail.get("artifacts") or []) + + def show(self, rows: list[dict[str, Any]]) -> None: + self.rows = rows + table = self.query_one("#files", DataTable) + for row in rows: + table.add_row( + str(row.get("name", "")), + str(row.get("node", "")), + str(row.get("media_type", "")), + f"{int(row.get('size') or 0):,}", + key=str(row.get("digest", "")), + ) + self.note("This run produced no artifacts." if not rows else "Enter downloads.") + + def note(self, message: str) -> None: + self.query_one("#note", Label).update(message) + + def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: + self.action_download() + + def action_download(self) -> None: + table = self.query_one("#files", DataTable) + if not table.row_count: + return + digest = table.coordinate_to_cell_key(table.cursor_coordinate).row_key.value + row = next((one for one in self.rows if one.get("digest") == digest), None) + if row is not None: + self.fetch(row) + + @work(thread=True) + def fetch(self, row: dict[str, Any]) -> None: + # The name it was written under reads better than the message's, which + # is chosen for the graph — the same choice `fluksio artifacts` makes. + out = Path(str(row.get("filename") or row.get("name"))) + try: + out.write_bytes(self.client.download(str(row["digest"]))) + except Exception as exc: # noqa: BLE001 — the modal reports it + self.app.call_from_thread(self.note, f"Could not fetch it: {exc}") + return + self.app.call_from_thread(self.note, f"Wrote {out.resolve()}") + + class ServeApp(App[int]): """One screen: how the engine is, what has run, and what it is saying.""" @@ -83,6 +176,8 @@ class ServeApp(App[int]): #runs { height: 2fr; } #log { height: 1fr; border-top: solid $panel; } #enroll { width: 60; height: auto; padding: 1 2; background: $surface; } + #artifacts { width: 80; height: auto; padding: 1 2; background: $surface; } + #artifacts DataTable { height: auto; max-height: 14; } """ BINDINGS = [ @@ -90,6 +185,9 @@ class ServeApp(App[int]): ("s", "stop_start", "stop/start"), ("r", "restart", "restart"), ("c", "cancel_run", "cancel run"), + ("space", "pick", "pick for comparison"), + ("enter", "compare", "compare"), + ("a", "artifacts", "artifacts"), ("e", "enroll", "enroll"), ] @@ -103,6 +201,12 @@ class ServeApp(App[int]): self.adopted: int | None = None self.client: Client | None = None self.url = "" + #: The runs ticked for a comparison, in the order they were ticked. + self.picked: list[str] = [] + #: Set by the socket thread, drained by the coalescing timer — so a + #: cascade of events costs one read of the engine rather than one each. + self.stirred = False + self.stop_stream = threading.Event() # -- layout --------------------------------------------------------------- @@ -117,9 +221,13 @@ class ServeApp(App[int]): def on_mount(self) -> None: self.title = f"fluksio — {self.data_dir}" table = self.query_one("#runs", DataTable) - table.add_columns("run", "status", "flow", "took", "params") + table.add_columns(" ", "run", "status", "flow", "took", "params") + table.focus() self.start_engine(first=True) + # The socket is what makes a run that starts and finishes between two + # ticks visible; this stays as the heartbeat for what is not on the bus. self.set_interval(WATCH_INTERVAL_S, self.refresh_panels) + self.set_interval(COALESCE_S, self.drain) # -- the engine under the screen ------------------------------------------ @@ -201,6 +309,8 @@ class ServeApp(App[int]): self.note(f"No credential to talk to the engine with: {exc}") return self.url = self.client.url + self.stop_stream = threading.Event() + self.watch_engine(self.client.url, self.client.token) self.refresh_panels() def engine_pid(self) -> int | None: @@ -218,8 +328,36 @@ class ServeApp(App[int]): except OSError as exc: self.note(f"Could not stop it: {exc}") self.adopted = None + self.stop_stream.set() self.client = None + # -- the engine's own events ---------------------------------------------- + + @work(thread=True, exclusive=True, group="stream") + def watch_engine(self, url: str, token: str) -> None: + """The bus, over the websocket the browser subscribes to.""" + stream.subscribe( + url, + token, + lambda event: setattr(self, "stirred", True), + self.stop_stream, + kinds=KINDS, + on_error=lambda exc: self.call_from_thread( + self.note, f"Not receiving events: {exc}" + ), + ) + + def drain(self) -> None: + """Whatever the bus said since the last tick, as one read.""" + if not self.stirred: + return + self.stirred = False + self.refresh_panels() + # A run finishing is exactly when a comparison's numbers become final, + # and the screen showing them cannot hear the bus itself. + if isinstance(self.screen, CompareScreen): + self.screen.action_reload() + # -- what the panels show ------------------------------------------------- def refresh_panels(self) -> None: @@ -244,16 +382,24 @@ class ServeApp(App[int]): status.clear() status.write(screen) table = self.query_one("#runs", DataTable) + cursor = table.cursor_row table.clear() for row in rows: + run_id = str(row.get("id", "")) table.add_row( - str(row.get("id", ""))[-8:], + "·" if run_id in self.picked else " ", + run_id[-8:], str(row.get("status", "")), str(row.get("flow", "")), - f"{(row.get('duration_ms') or 0) / 1000:.1f}s", + _dur(row.get("duration_ms")), json.dumps(row.get("params") or {})[:60], - key=str(row.get("id", "")), + key=run_id, ) + # The rows are rewritten wholesale every refresh, and a cursor that + # jumped back to the top on each one would make picking three runs a + # race against the clock. + if 0 <= cursor < table.row_count: + table.move_cursor(row=cursor) def show_offline(self, exc: Exception) -> None: status = self.query_one("#status", RichLog) @@ -273,15 +419,54 @@ class ServeApp(App[int]): self.stop_engine() self.start_engine() - def action_cancel_run(self) -> None: + def cursor_run(self) -> str: + """The whole id of the run the cursor is on. + + The cell holds its tail, which is what fits in a column; the row's key + is the whole of it, which is what the engine is asked about. + """ table = self.query_one("#runs", DataTable) if self.client is None or not table.row_count: - return - # The cell holds the tail of the id, which is what fits; the row's key - # is the whole of it, which is what the engine is asked about. + return "" key = table.coordinate_to_cell_key(table.cursor_coordinate).row_key - if key.value: - self.cancel_run(str(key.value)) + return str(key.value or "") + + def action_cancel_run(self) -> None: + run_id = self.cursor_run() + if run_id: + self.cancel_run(run_id) + + def action_pick(self) -> None: + """Tick the run under the cursor into the comparison, or untick it.""" + run_id = self.cursor_run() + if not run_id: + return + if run_id in self.picked: + self.picked.remove(run_id) + elif len(self.picked) >= MAX_PICKED: + self.note(f"A comparison carries {MAX_PICKED} runs; untick one first.") + return + else: + self.picked.append(run_id) + self.refresh_panels() + + def action_compare(self) -> None: + """The ticked runs, or the one under the cursor if none are ticked.""" + if self.client is None: + return + ids = self.picked or [self.cursor_run()] + if any(ids): + self.push_screen(CompareScreen(self.client, [one for one in ids if one])) + + def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: + """Enter on the focused runs table, which never reaches the binding.""" + if event.data_table.id == "runs": + self.action_compare() + + def action_artifacts(self) -> None: + run_id = self.cursor_run() + if self.client is not None and run_id: + self.push_screen(Artifacts(self.client, run_id)) @work(thread=True) def cancel_run(self, run_id: str) -> None: diff --git a/backend/fluksio/tui/chart.py b/backend/fluksio/tui/chart.py new file mode 100644 index 0000000..12d79c9 --- /dev/null +++ b/backend/fluksio/tui/chart.py @@ -0,0 +1,212 @@ +"""Curves at a terminal, drawn with braille. + +A terminal cell holds 2x4 braille dots, so an 80x20 pane is a 160x80 plot — +enough resolution that a training curve reads as a curve rather than as a bar +chart. Nothing is imported for this: a plotting library would be two +dependencies on a package whose every dependency is argued for in +`pyproject.toml`, for something a screenful of arithmetic does. + +The colours are deliberately not the web's. `--chart-1…5` step one brand hue +by lightness, which is right against a designed surface and unreadable in a +terminal — five steps of one hue collapse to one colour on a 16-colour tty. +Here identity is carried by hue, and the legend names every series anyway. +""" + +from __future__ import annotations + +from collections.abc import Iterable, Iterator, Sequence +from typing import Any + +from rich.text import Text +from textual.widget import Widget + +#: One run's readings: a label, and `[[x, y], ...]` as the engine answers. +Series = tuple[str, Sequence[Sequence[float]]] + +#: The cap the web draws at, for the same reason: past five, a legend is a +#: puzzle. A sweep of twenty wants faceting, not a sixth colour. +MAX_SERIES = 5 + +#: Distinct hues, and the bright half of the ANSI sixteen so they hold up on a +#: light terminal as well as a dark one. +SERIES_STYLES = ( + "bright_cyan", + "bright_magenta", + "bright_yellow", + "bright_green", + "bright_red", +) + +#: `⠀`, and the dot bits of the cell above it, by (column, row) within it. +BRAILLE = 0x2800 +DOTS = ((0x01, 0x02, 0x04, 0x40), (0x08, 0x10, 0x20, 0x80)) + +#: What the y tick labels take, and what the x axis and its labels take below. +GUTTER = 9 +AXIS_ROWS = 2 + + +def _num(value: float) -> str: + """A tick label short enough to sit in the gutter.""" + if value and (abs(value) >= 1e5 or abs(value) < 1e-3): + return f"{value:.1e}" + return f"{value:.4g}" + + +def shorten(label: str) -> str: + """A run's label, with the id cut to the tail that identifies it. + + The engine answers ` (seed 3)`; a terminal wants the eight + characters people actually read it by, and the seed kept. + """ + head, space, rest = label.partition(" ") + return head[-8:] + space + rest + + +def _segment(x0: int, y0: int, x1: int, y1: int) -> Iterator[tuple[int, int]]: + """Every dot along a straight line between two readings. + + Points alone would draw a dotted cloud: forty readings across a + two-hundred dot axis touch one dot in five. + """ + dx, dy = abs(x1 - x0), -abs(y1 - y0) + step_x = 1 if x0 < x1 else -1 + step_y = 1 if y0 < y1 else -1 + error = dx + dy + while True: + yield x0, y0 + if x0 == x1 and y0 == y1: + return + doubled = 2 * error + if doubled >= dy: + error += dy + x0 += step_x + if doubled <= dx: + error += dx + y0 += step_y + + +def _bounds(lines: Sequence[Series]) -> tuple[float, float, float, float]: + xs = [point[0] for _, points in lines for point in points] + ys = [point[1] for _, points in lines for point in points] + return min(xs), max(xs), min(ys), max(ys) + + +def _place(value: float, low: float, high: float, span: int) -> int: + """A reading's dot along an axis of `span` dots. + + A flat curve has no range to divide by and is drawn down the middle, which + is the honest picture of a metric that never moved. + """ + if high <= low: + return (span - 1) // 2 + return round((value - low) / (high - low) * (span - 1)) + + +def _cells( + lines: Sequence[Series], cols: int, rows: int +) -> tuple[dict[tuple[int, int], list[int]], tuple[float, float, float, float]]: + """The braille cells the curves fill, and the bounds they were scaled to.""" + low_x, high_x, low_y, high_y = _bounds(lines) + width, height = cols * 2, rows * 4 + cells: dict[tuple[int, int], list[int]] = {} + for index, (_, points) in enumerate(lines): + previous: tuple[int, int] | None = None + for x_value, y_value in points: + spot = ( + _place(x_value, low_x, high_x, width), + height - 1 - _place(y_value, low_y, high_y, height), + ) + dots = _segment(*previous, *spot) if previous is not None else [spot] + for dot_x, dot_y in dots: + cell = cells.get((dot_y // 4, dot_x // 2)) + bit = DOTS[dot_x % 2][dot_y % 4] + if cell is None: + cells[dot_y // 4, dot_x // 2] = [bit, index] + else: + # ponytail: the first series to reach a cell keeps its + # colour — a terminal cell has one foreground. Half-blocks + # would let two share it, at a quarter of the resolution. + cell[0] |= bit + previous = spot + return cells, (low_x, high_x, low_y, high_y) + + +def _row(cells: dict[tuple[int, int], list[int]], row: int, cols: int) -> Text: + """One line of the plot, as few spans as the colours allow.""" + out = Text() + run, style = "", "" + for col in range(cols): + cell = cells.get((row, col)) + glyph = chr(BRAILLE + cell[0]) if cell else " " + wanted = SERIES_STYLES[cell[1] % len(SERIES_STYLES)] if cell else "" + if wanted != style: + out.append(run, style=style or None) + run, style = "", wanted + run += glyph + out.append(run, style=style or None) + return out + + +def plot(lines: Sequence[Series], width: int, height: int, x_label: str = "") -> Text: + """The curves, scaled to a pane `width` by `height` cells. + + Every series shares one y scale, which is the whole point of drawing them + together: two runs whose losses differ by a factor of ten should look like + it. + """ + drawn = [line for line in lines[:MAX_SERIES] if len(line[1]) > 0] + if not drawn: + return Text("No readings.", style="dim") + cols, rows = width - GUTTER, height - AXIS_ROWS + if cols < 8 or rows < 2: + return Text("Too small to draw.", style="dim") + + cells, (low_x, high_x, low_y, high_y) = _cells(drawn, cols, rows) + #: Top, middle and bottom carry a value; labelling every row would be + #: noise on a chart that is mostly one curve. + labels = {0: high_y, rows // 2: (low_y + high_y) / 2, rows - 1: low_y} + + out = Text(no_wrap=True, overflow="crop") + for row in range(rows): + mark = _num(labels[row]) if row in labels else "" + out.append(f"{mark:>{GUTTER - 2}} ", style="dim") + out.append("┤" if row in labels else "│", style="dim") + out.append_text(_row(cells, row, cols)) + out.append("\n") + out.append(" " * (GUTTER - 1) + "└" + "─" * cols + "\n", style="dim") + left, right = _num(low_x), _num(high_x) + tail = f" {x_label}" if x_label else "" + pad = max(cols - len(left) - len(right) - len(tail), 1) + out.append(" " * GUTTER + left + " " * pad + right + tail, style="dim") + return out + + +def legend(labels: Iterable[str]) -> Text: + """Which colour is which run — never colour alone.""" + out = Text() + for index, label in enumerate(list(labels)[:MAX_SERIES]): + if index: + out.append(" ") + out.append("■ ", style=SERIES_STYLES[index % len(SERIES_STYLES)]) + out.append(shorten(label)) + return out + + +class Curves(Widget): + """The plot, redrawn for whatever size the terminal gives it.""" + + DEFAULT_CSS = "Curves { height: 1fr; }" + + def __init__(self, **kwargs: Any) -> None: + super().__init__(**kwargs) + self.lines: list[Series] = [] + self.x_label = "" + + def show(self, lines: Sequence[Series], x_label: str = "") -> None: + self.lines = list(lines)[:MAX_SERIES] + self.x_label = x_label + self.refresh() + + def render(self) -> Text: + return plot(self.lines, self.size.width, self.size.height, self.x_label) diff --git a/backend/fluksio/tui/compare.py b/backend/fluksio/tui/compare.py new file mode 100644 index 0000000..7e6875e --- /dev/null +++ b/backend/fluksio/tui/compare.py @@ -0,0 +1,231 @@ +"""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 + +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, metric_names +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. +#: Metric points are not published on the event bus — the run's sink writes +#: them — so the socket makes the *table* live and this makes the *curve* live. +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}" + 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 = metric_names(self.client, 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, and metric points are not on + # the event bus — so this is the one thing the screen polls for. + 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 keep_live(self, running: bool) -> None: + if running and self.live is None: + self.live = self.set_interval(LIVE_S, self.read_curves) + elif not running and self.live is not None: + self.live.stop() + self.live = None diff --git a/backend/pyproject.toml b/backend/pyproject.toml index ba3ce16..606d1e7 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -72,6 +72,10 @@ dependencies = [ "rich>=13", # `fluksio serve` opens a dashboard with it at a terminal. "textual>=1.0", + # That dashboard subscribes to the engine's event bus over the same socket + # the browser uses. It arrives under `fastapi[standard]` either way, and is + # named here because this code imports it. + "websockets>=13", # The encrypted-payload half of web push (RFC 8291). It brings only # cryptography, which is already here. `pywebpush` would do the signing and # the request too, at the cost of `requests` *and* `aiohttp` — two HTTP @@ -185,7 +189,7 @@ ignore = [ "fluksio/cli.py" = ["T201"] "fluksio/sdk/cli.py" = ["T201"] # What it prints is the line left in the terminal after the dashboard closes. -"fluksio/tui.py" = ["T201"] +"fluksio/tui/app.py" = ["T201"] # Node functions take `params` whether or not they use it — that is the # contract the engine calls them with. "fluksio/flow/nodes.py" = ["ARG001", "ARG002"] diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index 15fb523..17b0993 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -712,3 +712,26 @@ def test_serve_moves_off_a_port_that_is_taken() -> None: assert _parser().parse_args(["serve"]).port is None assert _parser().parse_args(["serve", "--port", "9000"]).port == 9000 assert DEFAULT_PORT == 8000 + + +def test_a_duration_reads_the_way_an_age_does() -> None: + """One notation for both, rather than one per place that printed one.""" + from fluksio.sdk.cli import _dur + + assert _dur(None) == "0.0s" + assert _dur(1240) == "1.2s" + assert _dur(90_000) == "1.5min" + # A run measured in hours used to print four digits of seconds. + assert _dur(7_200_000) == "2.0h" + assert _dur(172_800_000) == "2.0d" + + +def test_the_dashboard_subscribes_where_the_browser_does() -> None: + """The same socket, and the token where a handshake can carry it.""" + from fluksio.sdk.stream import socket_url + + assert socket_url("http://127.0.0.1:8000", "abc") == ( + "ws://127.0.0.1:8000/api/v1/flows/ws?token=abc" + ) + # TLS on the one side is TLS on the other. + assert socket_url("https://engine.example.com", "t").startswith("wss://") diff --git a/backend/tests/test_tui_chart.py b/backend/tests/test_tui_chart.py new file mode 100644 index 0000000..79642e2 --- /dev/null +++ b/backend/tests/test_tui_chart.py @@ -0,0 +1,79 @@ +"""The braille renderer, which is the one part of the dashboard that is logic. + +The screens around it are wiring the type checker already covers; this is +arithmetic that a wrong sign would silently flip upside down. +""" + +from fluksio.tui.chart import ( + DOTS, + GUTTER, + SERIES_STYLES, + _cells, + _place, + legend, + plot, + shorten, +) + + +def test_a_diagonal_lands_in_the_corners_it_should() -> None: + """Bottom left to top right, on a grid small enough to name every dot.""" + # Two cells across, one down: a 4x4 dot grid. + cells, bounds = _cells([("a", [[0.0, 0.0], [1.0, 1.0]])], cols=2, rows=1) + assert bounds == (0.0, 1.0, 0.0, 1.0) + assert set(cells) == {(0, 0), (0, 1)} + # The low reading sits at the bottom of the left cell... + assert cells[0, 0][0] & DOTS[0][3] + # ...and the high one at the top of the right, which is the y axis being + # the right way up. + assert cells[0, 1][0] & DOTS[1][0] + assert cells[0, 0][1] == cells[0, 1][1] == 0 + + +def test_readings_are_joined_rather_than_dotted() -> None: + """Two readings far apart are a line, not two lonely dots.""" + cells, _ = _cells([("a", [[0.0, 0.0], [1.0, 1.0]])], cols=10, rows=4) + assert len(cells) >= 9 + + +def test_a_flat_curve_is_drawn_down_the_middle() -> None: + """A metric that never moved has no range to divide by.""" + assert _place(5.0, 5.0, 5.0, 8) == 3 + drawn = plot([("a", [[0.0, 5.0], [1.0, 5.0], [2.0, 5.0]])], 40, 6) + assert "5" in drawn.plain + + +def test_each_series_keeps_its_own_colour() -> None: + drawn = plot( + [("a", [[0.0, 0.0], [1.0, 0.0]]), ("b", [[0.0, 9.0], [1.0, 9.0]])], 40, 8 + ) + used = {span.style for span in drawn.spans} + assert SERIES_STYLES[0] in used + assert SERIES_STYLES[1] in used + + +def test_the_gutter_carries_the_real_bounds() -> None: + drawn = plot([("a", [[0.0, 0.5], [10.0, 9.5]])], 60, 8, x_label="step") + assert "9.5" in drawn.plain + assert "0.5" in drawn.plain + # And the x axis says what it is and where it ran. + assert "step" in drawn.plain + assert "10" in drawn.plain + # Nothing wraps: a chart that reflowed would be unreadable. + assert all(len(line) <= 60 for line in drawn.plain.splitlines()) + + +def test_nothing_to_draw_says_so_instead_of_raising() -> None: + assert "No readings" in plot([], 40, 8).plain + assert "No readings" in plot([("a", [])], 40, 8).plain + # One reading is a dot, not a division by zero. + assert plot([("a", [[1.0, 2.0]])], 40, 8).plain.count("\n") == 7 + # And a pane with no room for a gutter says that rather than drawing junk. + assert "Too small" in plot([("a", [[0.0, 1.0]])], GUTTER + 4, 3).plain + + +def test_a_run_is_named_by_the_tail_of_its_id() -> None: + assert ( + shorten("0193ab7c-dead-beef-1234-aabbccddeeff (seed 3)") == "ccddeeff (seed 3)" + ) + assert "■" in legend(["run-aaaaaaaa"]).plain diff --git a/docs/code/cli.md b/docs/code/cli.md index 9f8b0dc..f514019 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -125,6 +125,9 @@ by it. | `s` | stop the engine, or start it again | | `r` | restart it | | `c` | cancel the run the cursor is on | +| `space` | tick the run under the cursor into a comparison | +| `enter` | compare the ticked runs, or draw the one under the cursor | +| `a` | list what the run left behind, and fetch it | | `e` | pair with a portal, without leaving the screen | The engine is a child process rather than a thread, which is what makes those @@ -136,6 +139,37 @@ stopped from here only when it is this installation's own: both the pidfile beside the data and a token this directory's key signed have to agree. Another installation's engine is named and left alone. +The screen subscribes to the engine's event bus over the same websocket a +browser uses, so a run appears the moment it starts rather than at the next +poll — which is what used to make a run that started and finished inside five +seconds visible only in the history. It reconnects on its own, quietly: an +engine stopped from this screen is a normal state, not an error to fill the +log pane with. + +#### Comparing runs + +`enter` opens a comparison of the ticked runs: one metric's curve for each, +over a table of what differs between them. + +The curves are drawn in braille, five to a chart, in five distinct colours — +a deliberate departure from the browser's chart palette, where the five series +step one hue by lightness. Hue is what survives a terminal. Every curve is +named in the legend either way. + +Two pickers sit above the chart. The first is the metric, named as the run +records it — `train.loss` rather than `loss`, since a name is qualified by the +node that published it. The second is what it is plotted against: the step, +`time (s)` measured from each run's own first reading so runs started hours +apart lie on top of each other, or another metric of the same runs. + +The table below keeps only what actually differs — the inputs whose values are +not the same in every run, and the seed when it varies — beside each run's +status, duration and final numbers. A parameter every run shared is not a +column worth scanning. + +A run that is still going has its curve re-read once a second, and the whole +comparison refreshes when the engine says a run finished. + ## `fluksio enroll` Pairs an existing installation with a portal. diff --git a/uv.lock b/uv.lock index ea1b9b8..4c342bb 100644 --- a/uv.lock +++ b/uv.lock @@ -894,6 +894,7 @@ dependencies = [ { name = "sqlmodel" }, { name = "textual" }, { name = "uv" }, + { name = "websockets" }, ] [package.optional-dependencies] @@ -945,6 +946,7 @@ requires-dist = [ { name = "sqlmodel", specifier = ">=0.0.21,<1.0.0" }, { name = "textual", specifier = ">=1.0" }, { name = "uv", specifier = ">=0.5" }, + { name = "websockets", specifier = ">=13" }, ] provides-extras = ["parquet", "server"]