"""The ``monitor`` command: a live, read-only dashboard over the run's state.
Where ``status`` is a one-shot report, ``monitor`` *watches*: it opens the
state database as an observer (``StateStore(..., read_only=True)`` -- the
engine itself rejects writes, so watching a live run can never corrupt it)
and redraws the five-panel dashboard (run overview, stage pipeline, running
now, recent completions, recent failures) every ``--interval`` seconds until
Ctrl-C. ``--once`` prints a single snapshot and exits -- the
scriptable/CI-friendly leg.
The wiring mirrors the Orchestrator's constructor discipline: this command
hands ``core.monitor.build_snapshot`` plain names and mappings derived from
the config (stage order, stage -> backend name, backend name -> kind), so
the core stays free of config imports. Two probes touch infrastructure, both
throttled: backend health (constructed via ``build_runtime``, asked
``health_check`` through ``core.preflight.check_backends``, cached by
``_HealthCache`` -- once at startup, then at most every
``HEALTH_PROBE_TTL_S`` seconds) and per-job CPU/GPU usage
(``core.cpusample.UsageSampler`` over the in-flight ledger -- local backends
every tick, remote ones on the samplers' own TTL).
Exit codes: 0 = watched and stopped cleanly (Ctrl-C included -- stopping a
dashboard is not a failure), 1 = no run state exists yet, 2 = config or
usage error.
"""
from __future__ import annotations
import time
from pathlib import Path
from typing import TYPE_CHECKING
from matmelt.commands import STATE_DB_FILENAME, resolve_config_path
from matmelt.config import ConfigError, build_runtime, load_config
from matmelt.core.cpusample import UsageSampler
from matmelt.core.monitor import build_snapshot
from matmelt.core.preflight import check_backends
from matmelt.core.stages.registry import FUNNEL_ORDER
from matmelt.core.state import StateStore
from matmelt.ui.console import render_config_error, render_error, render_note
from matmelt.ui.monitor import print_snapshot, render_snapshot
from matmelt.ui.monitor_live import run_live
if TYPE_CHECKING:
from collections.abc import Callable, Mapping
from rich.console import RenderableType
from matmelt.backends.base import ComputeBackend
from matmelt.core.monitor import MonitorSnapshot
#: How long one backend-health probe stays fresh (seconds). Probes can be
#: slow (``ssh_exec`` does a network round-trip), so the dashboard re-probes
#: on this cadence instead of on every refresh tick.
HEALTH_PROBE_TTL_S = 30.0
class _HealthCache:
"""A time-based cache around one backend-health probe.
Without it, a one-second refresh interval would mean an SSH round-trip per
second per remote backend -- the dashboard would generate more load than
the run it is watching.
Parameters
----------
prober
The whole probe, injected as ``() -> {backend_name: healthy}`` so tests
exercise the caching with a counter instead of real backends.
ttl_s
How long a probe result stays fresh.
"""
def __init__(
self, prober: Callable[[], dict[str, bool]], *, ttl_s: float = HEALTH_PROBE_TTL_S
) -> None:
self._prober = prober
self._ttl_s = ttl_s
self._cached: dict[str, bool] | None = None
self._probed_at = 0.0
def get(self, now: float) -> dict[str, bool]:
"""Give the current health mapping, re-probing only when stale.
Parameters
----------
now
The caller's clock reading. Passing it in keeps the cache
deterministic under test and free of hidden time reads.
Returns
-------
dict
``{backend_name: healthy}``.
"""
if self._cached is None or (now - self._probed_at) > self._ttl_s:
self._cached = self._prober()
self._probed_at = now
return self._cached
def _probe_backends(backends: Mapping[str, ComputeBackend]) -> dict[str, bool]:
"""Sweep every backend's health once.
Parameters
----------
backends
The configured backends.
Returns
-------
dict
``{backend_name: healthy}`` -- the shape ``build_snapshot`` takes.
"""
return {health.backend: health.healthy for health in check_backends(backends)}
def _parse_track(track: str | None, track_file: str | None) -> list[str] | int | None:
"""Combine the two tracking options into one ordered list of ids.
Parameters
----------
track
Comma-separated ids, or ``None``.
track_file
A file of ids, or ``None``. Same conventions as ``retry``'s
``--materials-file``: one per line, blanks and ``#`` comments skipped.
Returns
-------
list or int or None
The ids **in the caller's order**, de-duplicated keeping the first
occurrence, so the tracked panel lists rows as the user named them.
``None`` when neither option was given; or the exit code ``2`` when the
file cannot be read or an explicit selection came out empty.
"""
if track is None and track_file is None:
return None
picked = [part.strip() for part in (track or "").split(",") if part.strip()]
if track_file is not None:
try:
lines = Path(track_file).read_text(encoding="utf-8").splitlines()
except OSError as exc:
render_error(f"monitor: cannot read --track-file {track_file}: {exc}")
return 2
picked.extend(
line.strip() for line in lines if line.strip() and not line.lstrip().startswith("#")
)
if not picked:
render_error(
"monitor: --track/--track-file given but the selection is empty -- "
"name at least one material id."
)
return 2
return list(dict.fromkeys(picked)) # de-dupe, keep first-occurrence order
[docs]
def run_monitor(
config_path: str,
*,
interval: float = 2.0,
once: bool = False,
max_ticks: int | None = None,
track: str | None = None,
track_file: str | None = None,
) -> int:
"""Watch a live run on a refreshing dashboard. Read-only.
Parameters
----------
config_path
The config file naming the workspace to watch.
interval
Seconds between refreshes. Must be positive.
once
Print a single snapshot instead of entering the live loop -- what a
script or a CI log wants.
max_ticks
Bound the live loop's updates. The testing knob; ``None`` watches until
Ctrl-C.
track
Comma-separated material ids to follow.
track_file
A file of ids, one per line. Either option switches the third panel
from recent failures to the tracked-materials detail for exactly those
ids, including honest "not found" rows for ids the store does not know.
Returns
-------
int
``0`` on a clean exit, Ctrl-C included; ``1`` when there is no run
state; ``2`` for a config error, a non-positive interval, or an
unusable tracking selection.
Notes
-----
The store is opened **read-only**, enforced by the SQLite engine rather
than by discipline, so watching a run can never disturb it. Backend health
is cached rather than probed per tick, and remote usage sampling is
throttled behind its own TTL.
"""
config_path = resolve_config_path(config_path)
try:
config = load_config(config_path)
except ConfigError as exc:
render_config_error(str(exc))
return 2
if interval <= 0:
render_error(f"monitor: --interval must be a positive number of seconds, got {interval}.")
return 2
tracked_ids = _parse_track(track, track_file)
if isinstance(tracked_ids, int): # error path returned an exit code
return tracked_ids
db_path = Path(config.run.work_dir) / STATE_DB_FILENAME
if not db_path.is_file():
render_error(f"monitor: no run state at {db_path} -- start one with 'matmelt run'.")
return 1
# Plain-data wiring for build_snapshot: names and mappings only, exactly
# like the Orchestrator's constructor -- core never sees the config.
stage_names = [name for name in FUNNEL_ORDER if name in config.stages]
stage_backends = {name: config.stages[name].backend for name in stage_names}
backend_kinds = {name: backend_cfg.kind for name, backend_cfg in config.backends.items()}
try:
runtime = build_runtime(config)
except ConfigError as exc: # backend constructor problems are config problems
render_config_error(str(exc))
return 2
health = _HealthCache(lambda: _probe_backends(runtime.backends))
health.get(time.time()) # probe once at startup, before the first render
# Per-material CPU/GPU: local sampling is a /proc (or one nvidia-smi)
# read -- every tick, TTL 0; anything else costs an SSH round-trip and
# stays on the samplers' default TTL.
local_ttls = {name: 0.0 for name, cfg in config.backends.items() if cfg.kind == "local"}
cpu = UsageSampler(runtime.backends, ttl_by_backend=local_ttls)
gpu = UsageSampler(runtime.backends, hook_name="sample_gpu_percents", ttl_by_backend=local_ttls)
store = StateStore(db_path, read_only=True)
try:
def snapshot() -> MonitorSnapshot:
now = time.time()
jobs = store.list_jobs()
return build_snapshot(
store,
run_name=config.run.name,
stage_names=stage_names,
stage_backends=stage_backends,
backend_kinds=backend_kinds,
backend_health=health.get(now),
stop_after=config.pipeline.stop_after,
cpu_by_job=cpu.sample(jobs, now),
gpu_by_job=gpu.sample(jobs, now),
tracked_ids=tracked_ids,
)
if once:
print_snapshot(snapshot())
return 0
def build() -> RenderableType:
return render_snapshot(snapshot())
try:
run_live(build, interval_s=interval, max_ticks=max_ticks)
except KeyboardInterrupt:
# Stopping a dashboard is the normal way out, not a failure; the
# Live context has already restored the terminal by the time the
# interrupt reaches us (see ui/monitor_live.py).
render_note("monitor: stopped.")
return 0
finally:
store.close()