"""Rich rendering for the ``monitor`` command: snapshot in, panels out.
The UI half of the monitor seam: ``core/monitor.py`` computes a
``MonitorSnapshot``; this module turns it into the five stacked panels
(run overview with the done-donut and state bar, stage pipeline, running
now with per-material CPU/GPU meters, recent completions, recent failures)
and nothing else -- no queries, no business logic, no mutation. It prints
through the shared
``console`` from ``ui/console.py``, so the one-console rule holds; it lives
in its own module only because the monitor's rendering is a cohesive block
that would double ``console.py``'s size.
Everything user-data-bearing (ids, stages, failure reasons) is ``escape``-d
before rendering -- reasons carry paths and bracketed text that Rich would
otherwise swallow as markup (the ``render_config_error`` lesson).
"""
from __future__ import annotations
import time
from typing import TYPE_CHECKING
from rich.console import Group, RenderableType
from rich.markup import escape
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from matmelt.core.state import MaterialState
from matmelt.ui.console import console
if TYPE_CHECKING:
from matmelt.core.monitor import (
MonitorSnapshot,
RunningMaterial,
StageSnapshot,
TrackedMaterial,
)
#: Character width of the stage-progress bar.
_BAR_WIDTH = 10
#: Character width of a running row's CPU meter (scaled to one core).
_CPU_BAR_WIDTH = 5
#: Most rows the running-now panel shows (longest-running first); the rest
#: fold into a "+N more" line so a big max_parallel can't scroll the
#: dashboard off screen.
_RUNNING_LIMIT = 15
#: Character width of the stacked state-distribution bar.
_STATE_BAR_WIDTH = 46
#: The order states appear in the distribution bar and legend: the forward
#: funnel left-to-right, then the two terminal off-chain states. Reading the
#: bar tracks a campaign's progress toward the green DONE segment.
_STATE_ORDER: tuple[MaterialState, ...] = (
MaterialState.NEW,
MaterialState.COMPOSED,
MaterialState.STRUCTURED,
MaterialState.SCREENED,
MaterialState.CALCULATED,
MaterialState.EXTRACTED,
MaterialState.DONE,
MaterialState.EXPANDED,
MaterialState.FAILED,
)
#: One color per state, shared by the bar segments and their legend swatches.
#: A cool funnel gradient warms to green at DONE; the terminals stand apart
#: (yellow = retired-after-fan-out, red = failed).
_STATE_STYLE: dict[MaterialState, str] = {
MaterialState.NEW: "grey50",
MaterialState.COMPOSED: "cyan",
MaterialState.STRUCTURED: "blue",
MaterialState.SCREENED: "magenta",
MaterialState.CALCULATED: "bright_cyan",
MaterialState.EXTRACTED: "bright_blue",
MaterialState.DONE: "green",
MaterialState.EXPANDED: "yellow",
MaterialState.FAILED: "red",
}
#: The donut ring's block cells, clockwise from the top arc's left end, on a
#: 5-row x 9-col canvas. Filling the first ``round(fraction * 16)`` cells
#: green sweeps a real proportional arc; the rest stay dim so the ring shape
#: is always legible (even at 0%).
_DONUT_ROWS, _DONUT_COLS = 5, 9
_DONUT_RING: tuple[tuple[int, int], ...] = (
(0, 2),
(0, 3),
(0, 4),
(0, 5),
(0, 6), # top arc, left -> right
(1, 7),
(2, 7),
(3, 7), # right side, top -> bottom
(4, 6),
(4, 5),
(4, 4),
(4, 3),
(4, 2), # bottom arc, right -> left
(3, 1),
(2, 1),
(1, 1), # left side, bottom -> top
)
_DONUT_RING_INDEX: dict[tuple[int, int], int] = {pos: i for i, pos in enumerate(_DONUT_RING)}
#: Truncation width for failure reasons (one line each, detail via ``status``).
_REASON_WIDTH = 60
#: Minimum column width reserved for material ids in the failures / tracked
#: tables. Chosen so realistic funnel ids (``C2Cl3Cu3Fe3O3_10`` -- 16 chars)
#: survive un-mangled at an 80-column terminal; the flexible Reason / Info
#: column absorbs the squeeze instead (regression-tested at width 80).
_MATERIAL_MIN_WIDTH = 16
[docs]
def fmt_duration(seconds: float) -> str:
"""``5085.0`` -> ``"1h 24m"`` -- coarse, human-first wall-clock spans.
Sub-minute spans show seconds (``"42s"``), sub-hour spans minutes and
seconds (``"4m 12s"``), longer spans hours and minutes. Negative or zero
input renders as ``"0s"`` (clock skew must never crash a dashboard).
Parameters
----------
seconds
A wall-clock span.
Returns
-------
str
A coarse, human-first rendering -- one unit of precision is what a
glancing reader can use.
"""
total = max(0, int(seconds))
hours, rest = divmod(total, 3600)
minutes, secs = divmod(rest, 60)
if hours:
return f"{hours}h {minutes:02d}m"
if minutes:
return f"{minutes}m {secs:02d}s"
return f"{secs}s"
[docs]
def fmt_rate(per_min: float) -> str:
"""``2.34`` -> ``"2.3/min"``; a zero rate is a dash, never ``"0.0/min"``.
Parameters
----------
per_min
A rate per minute.
Returns
-------
str
The formatted rate, or a dash when it is zero: an idle stage should
read as idle, not as a precise zero.
"""
return f"{per_min:.1f}/min" if per_min > 0 else "—"
def _fmt_clock(epoch: float) -> str:
"""Epoch seconds -> local ``HH:MM:SS`` (the failures' "failed at" column).
Parameters
----------
epoch
Epoch seconds.
Returns
-------
str
Local ``HH:MM:SS``.
"""
return time.strftime("%H:%M:%S", time.localtime(epoch))
def _progress_cell(stage: StageSnapshot) -> Text:
"""The stage's inline progress bar: done vs everything it has ever seen.
Parameters
----------
stage
The stage to render.
Returns
-------
rich.text.Text
An inline bar of completions against everything the stage has ever
seen.
"""
total = stage.done_total + stage.ready + stage.running + stage.failed
if total == 0:
return Text("waiting", style="dim")
fraction = stage.done_total / total
filled = round(fraction * _BAR_WIDTH)
bar = "█" * filled + "░" * (_BAR_WIDTH - filled)
return Text(f"{bar} {fraction:>4.0%}")
def _stage_name_cell(stage: StageSnapshot) -> Text:
"""Stage name colored by its situation: red failed, green done, yellow busy.
Parameters
----------
stage
The stage to render.
Returns
-------
rich.text.Text
The name, coloured by situation: red when it has failures, green when
it is done, yellow when it is busy.
"""
if stage.failed > 0:
return Text(stage.name, style="red")
if stage.ready == stage.running == 0:
if stage.done_total > 0:
return Text(stage.name, style="green")
return Text(stage.name, style="dim") # nothing seen yet
return Text(stage.name, style="yellow")
def _backend_cell(stage: StageSnapshot) -> Text:
"""``name (kind)``, prefixed with a red ✗ when its health probe failed.
Parameters
----------
stage
The stage whose backend is rendered.
Returns
-------
rich.text.Text
``name (kind)``, marked when its health probe failed. An unprobed
backend is rendered neutrally rather than as a failure.
"""
label = f"{stage.backend_name} ({stage.backend_kind})"
if stage.backend_healthy is False:
return Text(f"✗ {label}", style="bold red")
if stage.backend_healthy is None:
return Text(label, style="dim")
return Text(label)
def _donut(fraction: float) -> list[Text]:
"""A proportional ring gauge for ``fraction`` done, as 5 stacked Texts.
The ring's block cells fill green clockwise from the top; the rest stay a
dim ``░`` so the circle reads even at 0%. The whole-number percentage
sits centered inside. Out-of-range fractions are clamped (clock skew or a
partial store must never throw off the geometry).
Parameters
----------
fraction
The completed fraction, from 0 to 1.
Returns
-------
list
Five stacked ``Text`` rows forming the ring.
"""
fraction = max(0.0, min(1.0, fraction))
filled = round(fraction * len(_DONUT_RING))
label = f"{round(fraction * 100)}%".center(5) # cols 2..6 of the middle row
lines: list[Text] = []
for row in range(_DONUT_ROWS):
line = Text()
for col in range(_DONUT_COLS):
idx = _DONUT_RING_INDEX.get((row, col))
if idx is not None:
line.append(
"█" if idx < filled else "░", style="green" if idx < filled else "grey37"
)
elif row == 2 and 2 <= col <= 6:
line.append(label[col - 2], style="bold green" if filled else "bold")
else:
line.append(" ")
lines.append(line)
return lines
def _pad_to(lines: list[Text], height: int) -> list[Text]:
"""Vertically center ``lines`` within ``height`` rows using blank Texts.
Parameters
----------
lines
The rows to centre.
height
The target row count.
Returns
-------
list
The rows with blank ``Text`` padding above and below.
"""
if len(lines) >= height:
return lines
top = (height - len(lines)) // 2
bottom = height - len(lines) - top
return [Text() for _ in range(top)] + lines + [Text() for _ in range(bottom)]
def _allocate_widths(counts: list[int], width: int) -> list[int]:
"""Split ``width`` cells across ``counts`` proportionally (largest remainder).
Floors each share then hands the leftover cells to the largest fractional
remainders, so the segments always sum to exactly ``width`` and the
biggest states never lose their cell to rounding. A state may still get 0
cells when its share rounds below one -- the legend carries its exact
count regardless.
Parameters
----------
counts
The values to divide the width between.
width
Total cells available.
Returns
-------
list of int
Cell counts summing to ``width``. Largest-remainder apportionment, so
the bar always fills its width exactly and a small-but-nonzero state
still shows.
"""
total = sum(counts)
if total <= 0 or width <= 0:
return [0] * len(counts)
raw = [c / total * width for c in counts]
floors = [int(x) for x in raw]
leftover = width - sum(floors)
order = sorted(range(len(counts)), key=lambda i: raw[i] - floors[i], reverse=True)
for i in order[:leftover]:
floors[i] += 1
return floors
def _state_distribution(snap: MonitorSnapshot) -> list[RenderableType]:
"""The stacked state bar plus its legend, or a dim note on an empty store.
Parameters
----------
snap
The snapshot to render.
Returns
-------
list
The stacked bar and its swatched legend, or a dim note on an empty
store.
"""
if not snap.total:
return [Text("No materials yet", style="dim")]
present = [(state, snap.by_state.get(state, 0)) for state in _STATE_ORDER]
present = [(state, count) for state, count in present if count]
widths = _allocate_widths([count for _, count in present], _STATE_BAR_WIDTH)
bar = Text()
for (state, _), width in zip(present, widths, strict=True):
if width:
bar.append("█" * width, style=_STATE_STYLE[state])
legend = Text()
for i, (state, count) in enumerate(present):
if i:
legend.append(" ")
legend.append("█ ", style=_STATE_STYLE[state])
legend.append(f"{state.value} {count} ({count / snap.total:.0%})")
return [bar, legend]
[docs]
def overview_panel(snap: MonitorSnapshot) -> Panel:
"""Panel 1 -- headline numbers, a done-donut, and the state-distribution bar.
Parameters
----------
snap
The snapshot to render.
Returns
-------
rich.panel.Panel
Headline numbers, a completion donut, and the state-distribution bar.
"""
grid = Table.grid(padding=(0, 2))
grid.add_column(style="bold")
grid.add_column()
done = snap.by_state.get(MaterialState.DONE, 0)
percent = f" ({done / snap.total:.1%})" if snap.total else ""
failed = snap.by_state.get(MaterialState.FAILED, 0)
grid.add_row("Run", escape(snap.run_name))
grid.add_row("Elapsed", fmt_duration(snap.elapsed_s) if snap.started_at else "—")
grid.add_row("Total materials", str(snap.total))
grid.add_row("Done", f"{done}{percent}")
grid.add_row("In flight", str(snap.in_flight))
grid.add_row("Failed", Text(str(failed), style="bold red") if failed else "0")
grid.add_row("Throughput", fmt_rate(snap.throughput_per_min))
grid.add_row("ETA", "—" if snap.eta_s is None else f"~{fmt_duration(snap.eta_s)}")
grid.add_row("Stop stage", escape(snap.stop_after) if snap.stop_after else "(full funnel)")
donut = Group(*_pad_to(_donut(done / snap.total if snap.total else 0.0), height=9))
header = Table.grid(padding=(0, 4))
header.add_column()
header.add_column(justify="center")
header.add_row(grid, donut)
body = Group(header, Text(), *_state_distribution(snap))
return Panel(body, title="Run overview", title_align="left")
[docs]
def pipeline_panel(snap: MonitorSnapshot) -> Panel:
"""Panel 2 -- one row per configured stage, funnel order.
Parameters
----------
snap
The snapshot to render.
Returns
-------
rich.panel.Panel
One row per configured stage, in funnel order.
"""
table = Table(expand=True, pad_edge=False)
table.add_column("Stage", no_wrap=True, overflow="ellipsis")
table.add_column("Backend", no_wrap=True, overflow="ellipsis")
for numeric in ("Ready", "Running", "Done", "Failed"):
table.add_column(numeric, justify="right")
table.add_column("Speed", justify="right")
table.add_column("Progress")
for stage in snap.stages:
table.add_row(
_stage_name_cell(stage),
_backend_cell(stage),
str(stage.ready),
str(stage.running),
str(stage.done_total),
Text(str(stage.failed), style="red" if stage.failed else ""),
fmt_rate(stage.speed_per_min),
_progress_cell(stage),
)
return Panel(table, title="Stage pipeline", title_align="left")
def _usage_cell(percent: float | None) -> Text:
"""A usage meter cell (CPU or GPU): ``███░░ 63%`` scaled to one unit.
``None`` (no sample yet, un-probed remote, SLURM job, no GPU tooling) is
a dim dash -- never a fabricated 0. The meter clamps at 100% while the
number tells the whole truth (a 4-worker job legitimately reads
``400%``). Color speaks load: dim when idle, green while working,
yellow when hot.
Parameters
----------
percent
A usage reading, or ``None`` when the sampler had none.
Returns
-------
rich.text.Text
A meter, or a dash for ``None`` -- unknown is never drawn as zero.
"""
if percent is None:
return Text("—", style="dim")
fraction = min(1.0, percent / 100.0)
filled = round(fraction * _CPU_BAR_WIDTH)
bar = "█" * filled + "░" * (_CPU_BAR_WIDTH - filled)
if percent < 10.0:
style = "dim"
elif percent < 85.0:
style = "green"
else:
style = "yellow"
return Text(f"{bar} {percent:>3.0f}%", style=style)
def _attempt_cell(item: RunningMaterial) -> Text:
"""Which try this run is: ``attempts`` failed tries came before it.
A first try is quiet (dim ``#1``); a retry is yellow -- a glance at the
column shows which materials are struggling.
Parameters
----------
item
The running material.
Returns
-------
rich.text.Text
Which try this is, highlighted once a material has failed before.
"""
n = item.attempts + 1
return Text(f"#{n}", style="yellow" if n > 1 else "dim")
[docs]
def running_panel(snap: MonitorSnapshot) -> Panel:
"""Panel 3 -- every material riding a live job right now.
Longest-running first (the snapshot's order); capped at
``_RUNNING_LIMIT`` rows with an honest "+N more" line. An idle pipeline
shows a dim note, not an empty table.
Parameters
----------
snap
The snapshot to render.
Returns
-------
rich.panel.Panel
Every material riding a live job right now, longest-running first, with
its stage, backend, elapsed time, try count and usage meters.
"""
if not snap.running:
body: RenderableType = Text("Nothing running right now", style="dim", justify="center")
return Panel(body, title="Running now", title_align="left")
table = Table(expand=True, pad_edge=False)
table.add_column("Material", no_wrap=True, overflow="ellipsis", min_width=_MATERIAL_MIN_WIDTH)
table.add_column("Stage", no_wrap=True, overflow="ellipsis")
table.add_column("Backend", no_wrap=True, overflow="ellipsis")
table.add_column("Elapsed", justify="right")
table.add_column("Try", justify="right")
table.add_column("CPU", justify="left", no_wrap=True)
table.add_column("GPU", justify="left", no_wrap=True)
for item in snap.running[:_RUNNING_LIMIT]:
table.add_row(
Text(item.material_id),
escape(item.stage),
escape(item.backend_name),
fmt_duration(item.elapsed_s),
_attempt_cell(item),
_usage_cell(item.cpu_percent),
_usage_cell(item.gpu_percent),
)
hidden = len(snap.running) - _RUNNING_LIMIT
body = table
if hidden > 0:
body = Group(table, Text(f"… and {hidden} more", style="dim"))
return Panel(body, title=f"Running now ({len(snap.running)})", title_align="left")
[docs]
def successes_panel(snap: MonitorSnapshot) -> Panel:
"""Panel 4 -- the newest finished (DONE) materials, the failures' mirror.
``Took`` is the material's whole ride (created -> finished), the number a
screening campaign actually cares about per material.
Parameters
----------
snap
The snapshot to render.
Returns
-------
rich.panel.Panel
The newest finished materials, with how long each ride took.
"""
if not snap.recent_successes:
body: RenderableType = Text("No completed materials yet", style="dim", justify="center")
return Panel(body, title="Recent completions", title_align="left")
table = Table(expand=True, pad_edge=False)
table.add_column("Material", no_wrap=True, overflow="ellipsis", min_width=_MATERIAL_MIN_WIDTH)
table.add_column("Took", justify="right")
table.add_column("Finished at", justify="right")
for record in snap.recent_successes:
table.add_row(
Text(f"✓ {record.material_id}", style="green"),
fmt_duration(record.updated_at - record.created_at),
_fmt_clock(record.updated_at),
)
return Panel(table, title="Recent completions", title_align="left")
[docs]
def failures_panel(snap: MonitorSnapshot) -> Panel:
"""Panel 3 -- the newest terminal failures, or a green all-clear.
Parameters
----------
snap
The snapshot to render.
Returns
-------
rich.panel.Panel
The newest terminal failures, or a green all-clear when there are none.
"""
if not snap.recent_failures:
body: RenderableType = Text("✓ No failures yet", style="green", justify="center")
return Panel(body, title="Recent failures", title_align="left")
table = Table(expand=True, pad_edge=False)
table.add_column("Material", no_wrap=True, overflow="ellipsis", min_width=_MATERIAL_MIN_WIDTH)
table.add_column("Stage", no_wrap=True, overflow="ellipsis")
table.add_column("Reason", overflow="ellipsis", no_wrap=True, ratio=1)
table.add_column("Failed at")
for record in snap.recent_failures:
reason = (record.fail_reason or "?").splitlines()[0]
if len(reason) > _REASON_WIDTH:
reason = reason[: _REASON_WIDTH - 3] + "..."
table.add_row(
escape(record.material_id),
escape(record.fail_stage or "?"),
escape(reason),
_fmt_clock(record.updated_at),
)
return Panel(table, title="Recent failures", title_align="left")
def _tracked_state_cell(item: TrackedMaterial) -> Text:
"""The tracked row's state, colored like the rest of the palette.
Red for ``FAILED``, green for ``DONE``, yellow for everything still live
(matching ``_stage_name_cell``'s red/green/yellow language).
Parameters
----------
item
The tracked material.
Returns
-------
rich.text.Text
Its state in the shared palette, or a dim "not found".
"""
if item.state is MaterialState.FAILED:
return Text(item.state.value, style="red")
if item.state is MaterialState.DONE:
return Text(item.state.value, style="green")
return Text(item.state.value if item.state else "?", style="yellow")
def _tracked_info_cell(item: TrackedMaterial) -> Text:
"""The tracked row's free-text column: job key, failure reason, or a dash.
A material riding a live job shows the job's ledger key (what ``status``
and the logs call it); a ``FAILED`` material shows its (already
first-line) reason truncated to ``_REASON_WIDTH`` with the same ``...``
convention as the failures panel; anything else is a dash.
Parameters
----------
item
The tracked material.
Returns
-------
rich.text.Text
Its live job key, its first failure line, or a dash.
"""
if item.job_key is not None:
return Text(item.job_key)
if item.state is MaterialState.FAILED and item.fail_reason:
reason = item.fail_reason
if len(reason) > _REASON_WIDTH:
reason = reason[: _REASON_WIDTH - 3] + "..."
return Text(reason)
return Text("—", style="dim")
[docs]
def tracked_panel(snap: MonitorSnapshot) -> Panel:
"""Panel 3 (tracked mode) -- one row per ``--track``-ed material.
Rendered instead of the failures panel whenever ``snap.tracked`` is not
``None`` (failures stay reachable via ``status``). A ``found=False`` row
keeps its id visible with a dim-red "not found" -- a typo must be
diagnosable from the dashboard itself. All user-data cells go through
``Text``/``escape`` so ids and reasons never execute as Rich markup (the
module-docstring lesson).
Parameters
----------
snap
The snapshot to render.
Returns
-------
rich.panel.Panel
One row per tracked material, in the order the user named them,
including honest rows for ids the store does not know.
"""
table = Table(expand=True, pad_edge=False)
table.add_column("Material", no_wrap=True, overflow="ellipsis", min_width=_MATERIAL_MIN_WIDTH)
table.add_column("State", no_wrap=True)
table.add_column("Stage", no_wrap=True, overflow="ellipsis")
table.add_column("Backend", no_wrap=True, overflow="ellipsis")
table.add_column("In state", justify="right")
table.add_column("Info", overflow="ellipsis", no_wrap=True, ratio=1)
dash = Text("—", style="dim")
for item in snap.tracked or ():
if not item.found:
table.add_row(
Text(item.material_id, style="dim red"),
Text("not found", style="dim red"),
dash,
dash,
dash,
dash,
)
continue
table.add_row(
Text(item.material_id),
_tracked_state_cell(item),
Text(item.stage) if item.stage else dash,
Text(item.backend_name) if item.backend_name else dash,
fmt_duration(item.in_state_s),
_tracked_info_cell(item),
)
return Panel(table, title="Tracked materials", title_align="left")
[docs]
def render_snapshot(snap: MonitorSnapshot) -> RenderableType:
"""The full dashboard, stacked: overview, pipeline, running now,
recent completions, then the mode-dependent last panel -- tracked
materials when the snapshot carries ``tracked`` rows (the ``--track``
flow, even an empty tuple), recent failures otherwise.
Parameters
----------
snap
The snapshot to render.
Returns
-------
RenderableType
The stacked dashboard. The third panel is the tracked table when
tracking was requested, and recent failures otherwise.
"""
last = failures_panel(snap) if snap.tracked is None else tracked_panel(snap)
return Group(
overview_panel(snap),
pipeline_panel(snap),
running_panel(snap),
successes_panel(snap),
last,
)
[docs]
def print_snapshot(snap: MonitorSnapshot) -> None:
"""The ``--once`` path: render the dashboard to stdout a single time.
Parameters
----------
snap
The snapshot to print.
"""
console.print(render_snapshot(snap))