diff --git a/areno/api/metric_reader.py b/areno/api/metric_reader.py new file mode 100644 index 00000000..915890e4 --- /dev/null +++ b/areno/api/metric_reader.py @@ -0,0 +1,282 @@ +"""Read-side helpers for querying metric history from local run artifacts. + +Issue #254 extracts the dashboard's TensorBoard scalar reading into a set of +pure, side-effect-free functions so the ``areno metrics`` CLI and the dashboard +share one fact source. Reading is CPU-only; the heavy ``tensorboard`` import is +deferred to inside :func:`read_scalar_points`. + +Scope of the first version (issue #254 plan): + - First TLS data source only: ``events.out.tfevents.*`` scalars. + - ``--pid`` selects a run by filename pid suffix. + - jsonl fallback and ``--run `` are follow-ups (tracked separately). + +Reading semantics mirror ``areno/dashboard/server.py``'s +``_load_tensorboard_scalars`` byte-for-byte so a dashboard switch is +behavior-preserving: + - ``EventAccumulator(size_guidance={"scalars": 10000})`` + - ``accumulator.Scalars(tag)[-500:]`` + - NaN values skipped + - ``(name, step, value)`` de-duplication +""" + +from __future__ import annotations + +import datetime as dt +import math +from pathlib import Path +from typing import Any + + +def now() -> str: + """UTC timestamp string, matching ``areno/dashboard/server.py``'s ``now``.""" + return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat() + + +def number_like(value: Any) -> bool: + """True if ``value`` is a usable finite number (mirrors server's check).""" + try: + return not math.isnan(float(value)) + except (TypeError, ValueError): + return False + + +def tensorboard_event_sources(path: Path, pid: int | None) -> list[Path]: + """Locate ``events.out.tfevents.*`` files under ``path``. + + ``pid`` filters by the filename pid suffix when given; ``None`` merges every + event file in the directory. Falls back to ``[path]`` when no event files + exist so a fresh directory degrades to an empty read rather than a crash. + """ + event_files = sorted(path.rglob("events.out.tfevents.*"), key=lambda item: item.stat().st_mtime) + if not event_files: + return [path] + if pid is None: + return event_files + pid_marker = f".{pid}." + return [file for file in event_files if pid_marker in file.name or file.parent.name == f"pid-{pid}"] + + +def read_scalar_points(metrics_dir: str | Path, pid: int | None = None) -> list[dict[str, Any]]: + """Read TensorBoard scalars into de-duplicated ``[{name, value, step, time}]``. + + Iteration order, truncation (:code:`[-500:]`), NaN skip, and + ``(name, step, value)`` dedup match the dashboard read path exactly; callers + that feed these points back through the dashboard's ``_add_metric`` get + byte-identical ``job.metrics`` ordering. + """ + path = Path(metrics_dir) + try: + from tensorboard.backend.event_processing.event_accumulator import EventAccumulator + except Exception: + return [] + + points: list[dict[str, Any]] = [] + seen: set[tuple[str, int, float]] = set() + for accumulator_path in tensorboard_event_sources(path, pid): + try: + accumulator = EventAccumulator(str(accumulator_path), size_guidance={"scalars": 10000}) + accumulator.Reload() + tags = accumulator.Tags().get("scalars", []) + except Exception: + continue + for tag in tags: + try: + events = accumulator.Scalars(tag)[-500:] + except Exception: + continue + for event in events: + step = int(event.step) + value = float(event.value) + if math.isnan(value): + continue + key = (tag, step, value) + if key in seen: + continue + seen.add(key) + points.append({"name": tag, "value": value, "step": step, "time": now()}) + return points + + +def list_available_tags(points: list[dict[str, Any]]) -> list[str]: + """Return the distinct metric names found in ``points``, sorted for display.""" + names: set[str] = set() + for point in points: + name = str(point.get("name") or "") + if name: + names.add(name) + return sorted(names) + + +def summarize_metric( + points: list[dict[str, Any]], name: str, *, recent_n: int = 20 +) -> dict[str, Any]: + """Aggregate one metric into a summary dict. + + Existing contract (unchanged, additive fields below): + ``{name, count, last, min, max, recent, trend}`` plus the new step-aware + ``last_step``/``min_step``/``max_step``/``mean``/``recent_steps``. + + - ``count``: number of finite points for ``name`` (NaN-free, since the reader + already skips NaN). + - ``last`` / ``last_step``: value and step at the highest step. + - ``min``/``max``/``mean``: streaming single-pass over all retained points + (O(1) memory, independent of the ``[-500:]`` truncation). + - ``min_step``/``max_step``: the step range of the **retained window** + (the reader keeps ``accumulator.Scalars(tag)[-500:]``). For runs longer + than 500 steps ``min_step`` is the earliest step still in that tail, **not** + the start of training; ``max_step`` is always the latest available step. + - ``recent`` / ``recent_steps``: the last ``recent_n`` values and the matching + steps, in step order (parallel arrays of equal length). + - ``trend``: ``recent`` normalized to ``[0, 1]`` -- the same window as + ``recent`` -- so ``--limit`` bounds the sparkline length; ``render_table`` + maps it to a UTF-8 sparkline, ``render_json`` returns it verbatim. + """ + series = sorted( + (point for point in points if point.get("name") == name and number_like(point.get("value"))), + key=lambda point: int(point.get("step") or 0), + ) + steps = [int(point.get("step") or 0) for point in series] + values = [float(point.get("value")) for point in series] + count = len(values) + if count == 0: + last = min_v = max_v = mean = None + last_step = min_step = max_step = None + recent: list[float] = [] + recent_steps: list[int] = [] + trend: list[float] = [] + else: + last = values[-1] + min_v = max_v = values[0] + total = 0.0 + for value in values: + if value < min_v: + min_v = value + if value > max_v: + max_v = value + total += value + mean = total / count + last_step = max(steps) + min_step = min(steps) + max_step = last_step + window = values[-recent_n:] if recent_n > 0 else [] + window_steps = steps[-recent_n:] if recent_n > 0 else [] + recent = window + recent_steps = window_steps + trend = _normalize(window) + + return { + "name": name, + "count": count, + "last": last, + "last_step": last_step, + "min": min_v, + "max": max_v, + "mean": mean, + "min_step": min_step, + "max_step": max_step, + "recent": recent, + "recent_steps": recent_steps, + "trend": trend, + } + + +def _normalize(values: list[float]) -> list[float]: + """Scale ``values`` to ``[0, 1]``; a flat series maps to ``0.5`` everywhere.""" + if not values: + return [] + low = min(values) + high = max(values) + if high == low: + return [0.5 for _ in values] + span = high - low + return [(value - low) / span for value in values] + + +_SPARKS = "▁▂▃▄▅▆▇█" + + +def render_table(summary: dict[str, Any]) -> str: + """Render a summary as text: header rows + a hand-written sparkline. + + No external dependency (``rich``/``sparklines``) -- the trend is mapped to the + 8-glyph ``▁▂▃▄▅▆▇█`` rung by normalized value. Step-aware fields are shown so a + human can read training progress (``steps``/``last``/``recent``) at a glance. + """ + name = summary.get("name", "") + count = summary.get("count", 0) + trend = summary.get("trend") or [] + sparkline = _sparkline(trend) + last = _fmt(summary.get("last")) + last_step = summary.get("last_step") + if last_step is not None and summary.get("last") is not None: + last = f"{last} (step {last_step})" + steps = _fmt_range(summary.get("min_step"), summary.get("max_step")) + recent = summary.get("recent") or [] + recent_steps = summary.get("recent_steps") or [] + if recent_steps and len(recent_steps) == len(recent): + recent_str = ", ".join(f"step {s}: {_fmt(v)}" for s, v in zip(recent_steps, recent)) + else: + recent_str = ", ".join(_fmt(value) for value in recent) + lines = [ + f"metric {name}", + f"count {count}", + f"steps {steps}", + f"last {last}", + f"min {_fmt(summary.get('min'))}", + f"max {_fmt(summary.get('max'))}", + f"mean {_fmt(summary.get('mean'))}", + f"trend {sparkline}", + f"recent {recent_str}", + ] + return "\n".join(lines) + + +def _fmt_range(low: Any, high: Any) -> str: + """Format a step range as ``low -> high`` (``-`` when either end is missing).""" + if low is None or high is None: + return "-" + if low == high: + return str(low) + return f"{low} -> {high}" + + +def _sparkline(trend: list[float]) -> str: + if not trend: + return "" + glyphs = [] + for value in trend: + index = min(len(_SPARKS) - 1, max(0, int(value * len(_SPARKS)))) + glyphs.append(_SPARKS[index]) + return "".join(glyphs) + + +def _fmt(value: Any) -> str: + if value is None: + return "-" + if isinstance(value, float): + return f"{value:.6g}" + return str(value) + + +def render_json(summary: dict[str, Any]) -> dict[str, Any]: + """Return the summary as a JSON-serializable dict (trend = normalized array). + + The step-aware fields (``last_step``/``min_step``/``max_step``/``mean``/ + ``recent_steps``) are exposed so machine consumers can read training progress + and pair recent values with their steps (``jq '.recent_steps, .recent'``). + Existing keys keep their shape; the additions are purely append-only. + """ + return { + "name": summary.get("name"), + "count": summary.get("count"), + "last": summary.get("last"), + "last_step": summary.get("last_step"), + "min": summary.get("min"), + "max": summary.get("max"), + "mean": summary.get("mean"), + "min_step": summary.get("min_step"), + "max_step": summary.get("max_step"), + "recent": summary.get("recent"), + "recent_steps": summary.get("recent_steps"), + "trend": summary.get("trend"), + } diff --git a/areno/cli/main.py b/areno/cli/main.py index 8a6dc0bd..775a869b 100644 --- a/areno/cli/main.py +++ b/areno/cli/main.py @@ -21,6 +21,7 @@ class ArenoCli(click.Group): "dashboard": ("areno.cli.dashboard", "dashboard_command", "Start or stop the AReno React dashboard."), "train": ("areno.cli.train", "train_command", "Run SFT, DPO, GSPO, GRPO, or PPO training."), "serve": ("areno.cli.serve", "serve_command", "Serve an OpenAI-compatible chat API."), + "metrics": ("areno.cli.metrics", "metrics_command", "Query metric history from local run artifacts."), } def list_commands(self, ctx: click.Context) -> list[str]: diff --git a/areno/cli/metrics.py b/areno/cli/metrics.py new file mode 100644 index 00000000..e9c4201b --- /dev/null +++ b/areno/cli/metrics.py @@ -0,0 +1,135 @@ +"""``areno metrics`` -- query metric history from local run artifacts. + +Issue #254 adds a read-only CLI that summarizes one metric (last/min/max/ +recent/trend) from the ``events.out.tfevents.*`` artifacts a run writes. The +heavy work lives in the light, pure :mod:`areno.api.metric_reader` module; this +file only wires Click options, renders, and turns not-found cases into a clear +error plus the list of available tags. + +The command never writes or starts anything -- it only reads artifacts. +""" + +from __future__ import annotations + +import json as _json +from pathlib import Path +from typing import Any + +import click + +from areno.api import metric_reader +from areno.api.defaults import DEFAULT_METRICS_LOG_DIR + + +@click.command(name="metrics", context_settings={"help_option_names": ["-h", "--help"]}) +@click.option( + "--metrics-dir", + "metrics_dir", + default=None, + help=( + "Directory holding the run's events.out.tfevents.* artifacts. " + "Omit to use areno's default metrics log dir." + ), +) +@click.option( + "--pid", + type=int, + default=None, + help="Filter event files by the pid suffix in the filename; merge every run when omitted.", +) +@click.option( + "--name", + "name", + default=None, + help="Metric tag to summarize (e.g. rollout/rewards_mean). Omit to list available tags.", +) +@click.option( + "--limit", + default=20, + show_default=True, + help="Number of recent values to include in recent/trend.", +) +@click.option("--json", "as_json", is_flag=True, help="Emit a machine-readable JSON object.") +def metrics_command( + metrics_dir: str | None, + pid: int | None, + name: str | None, + limit: int, + as_json: bool, +) -> None: + """Query metric history (last/min/max/recent/trend) from local run artifacts.""" + if metrics_dir is None: + metrics_dir = DEFAULT_METRICS_LOG_DIR + + # Surface a missing directory early with a clear, located error -- do not let + # the reader silently degrade an empty path into an empty result. + directory = Path(metrics_dir) + if not directory.exists(): + raise click.ClickException(f"metrics dir not found: {metrics_dir} (pid={pid})") + + # Probe tensorboard up front: read_scalar_points silently returns [] when the + # import fails, which would otherwise look identical to "no metrics found". + # Raise a clear, located error so a missing dependency is distinguishable from + # an empty directory. (The dashboard keeps read_scalar_points' graceful + # degrade-to-empty; this probe is CLI-only.) + try: + import tensorboard # noqa: F401 + except ImportError as exc: + raise click.ClickException( + "tensorboard is not installed; run `pip install tensorboard` to read metrics." + ) from exc + + points = metric_reader.read_scalar_points(metrics_dir, pid) + tags = metric_reader.list_available_tags(points) + + # No tag requested: list what's available so the user can pick one. + if not name: + _emit_available(tags, metrics_dir, pid, as_json=as_json) + return + + if name not in tags: + _emit_available(tags, metrics_dir, pid, as_json=as_json, missing=name) + raise click.ClickException(f"metric name not found: {name} (in {metrics_dir}, pid={pid})") + + summary = metric_reader.summarize_metric(points, name, recent_n=limit) + if as_json: + click.echo(_json.dumps(metric_reader.render_json(summary), indent=2, sort_keys=True)) + else: + click.echo(metric_reader.render_table(summary)) + + +def _emit_available( + tags: list[str], + metrics_dir: str, + pid: int | None, + *, + as_json: bool, + missing: str | None = None, +) -> None: + """Print the available metric tags for the chosen run. + + ``missing`` is set when emitting the list as part of a not-found error so + the user sees both the unknown name they asked for and what they can use. + """ + if as_json: + payload: dict[str, Any] = { + "metrics_dir": metrics_dir, + "pid": pid, + "available_tags": tags, + } + if missing is not None: + payload["missing"] = missing + click.echo(_json.dumps(payload, indent=2, sort_keys=True)) + return + + label = ( + f"metric '{missing}' not found in {metrics_dir} (pid={pid}); available tags" + if missing is not None + else f"available metric tags in {metrics_dir} (pid={pid})" + ) + if not tags: + click.echo(f"{label}: (none found)") + return + click.echo(label) + for tag in tags: + click.echo(f" {tag}") diff --git a/areno/dashboard/server.py b/areno/dashboard/server.py index 9bb8b4d4..1ea45597 100644 --- a/areno/dashboard/server.py +++ b/areno/dashboard/server.py @@ -6,7 +6,6 @@ import argparse import datetime as dt import json -import math import os import re import shlex @@ -23,6 +22,8 @@ from typing import Any from uuid import uuid4 +from areno.api.defaults import DEFAULT_METRICS_LOG_DIR +from areno.api.metric_reader import number_like from areno.cli.dashboard_registry import GLOBAL_REGISTRY_FILE from areno.cli.diagnostics import collect_env, run_checks from areno.dashboard.agent_context import agent_system_prompt @@ -31,7 +32,6 @@ ROOT = Path(os.environ.get("ARENO_DASHBOARD_ROOT", Path.cwd())).resolve() STATIC_DIR = Path(__file__).resolve().parent / "dist" STATE_FILE = ROOT / ".areno-dashboard-state.json" -DEFAULT_METRICS_LOG_DIR = "/tmp/areno/tfevent" TIME_SEGMENT_ORDER = [ "rollout", "make_sample", @@ -317,36 +317,27 @@ def _load_tensorboard_scalars(self, job: Job, path: Path) -> None: from tensorboard.backend.event_processing.event_accumulator import EventAccumulator except Exception: return + # Lazy import: read_scalar_points imports TensorBoard's EventAccumulator + # internally, so keep it out of module import time. + from areno.api.metric_reader import read_scalar_points + job.timeperf = [row for row in job.timeperf if row.get("source") != "metrics"] job._timeperf_keys = {int(row.get("step", -1)) for row in job.timeperf} by_step: dict[int, dict[str, float]] = {} - for accumulator_path in tensorboard_event_sources(path, job_pid(job)): - try: - accumulator = EventAccumulator(str(accumulator_path), size_guidance={"scalars": 10000}) - accumulator.Reload() - tags = accumulator.Tags().get("scalars", []) - except Exception: - continue - for tag in tags: - try: - events = accumulator.Scalars(tag)[-500:] - except Exception: - continue - for event in events: - step = int(event.step) - value = float(event.value) - if math.isnan(value): - continue - self._add_metric(job, tag, value, step) - time_name = tensorboard_time_segment_name(tag) - if time_name: - by_step.setdefault(step, {})[time_name] = value - if tag in {"train/step_e2e_time_s", "time/total", "time/e2e"}: - by_step.setdefault(step, {})["total"] = value - elif tag == "train/step_rollout_time_s": - by_step.setdefault(step, {})["rollout"] = value - elif tag in {"train/step_train_time_s", "train/policy_train_wall_time_s"}: - by_step.setdefault(step, {})["train"] = value + for point in read_scalar_points(path, job_pid(job)): + name = str(point.get("name") or "") + step = int(point.get("step") or 0) + value = float(point.get("value")) + self._add_metric(job, name, value, step) + time_name = tensorboard_time_segment_name(name) + if time_name: + by_step.setdefault(step, {})[time_name] = value + if name in {"train/step_e2e_time_s", "time/total", "time/e2e"}: + by_step.setdefault(step, {})["total"] = value + elif name == "train/step_rollout_time_s": + by_step.setdefault(step, {})["rollout"] = value + elif name in {"train/step_train_time_s", "train/policy_train_wall_time_s"}: + by_step.setdefault(step, {})["train"] = value for step, values in sorted(by_step.items()): total = values.pop("total", None) if total is None: @@ -796,24 +787,6 @@ def job_pid(job: Job) -> int | None: return job.pid -def number_like(value: Any) -> bool: - try: - return not math.isnan(float(value)) - except (TypeError, ValueError): - return False - - -def tensorboard_event_sources(path: Path, pid: int | None) -> list[Path]: - event_files = sorted(path.rglob("events.out.tfevents.*"), key=lambda item: item.stat().st_mtime) - if not event_files: - return [path] - if pid is None: - return event_files - pid_marker = f".{pid}." - matched = [file for file in event_files if pid_marker in file.name or file.parent.name == f"pid-{pid}"] - return matched - - def rollout_sample_sources(path: Path, pid: int | None) -> list[Path]: if pid is not None: candidates = [ diff --git a/docs/cli/metrics.rst b/docs/cli/metrics.rst new file mode 100644 index 00000000..25e711ad --- /dev/null +++ b/docs/cli/metrics.rst @@ -0,0 +1,142 @@ +:orphan: + +Metrics CLI reference +===================== + +``areno metrics`` + +Query metric history from local run artifacts -- read the ``events.out.tfevents.*`` +files a training run writes and summarize one metric into ``last`` / ``min`` / +``max`` / ``mean`` / a ``steps`` range / ``recent`` values / a compact ``trend``. +Step-aware fields (``last_step``, ``min_step``, ``max_step``, ``recent_steps``) +let you read training progress at a glance and pair each recent value with its +absolute step. The command is read-only: it never writes, starts, or contacts +training/serving processes. It is the command-line counterpart of the +dashboard's metric view and shares the same read-side code, so the two never +drift. + +.. code-block:: bash + + areno train --ckpt Qwen/Qwen3-0.6B --dataset-path gsm8k:main \ + --reward-fn-path examples/math/math_verify_reward.py --algo gspo \ + --metrics-log-dir /tmp/areno/tfevent + + areno metrics --metrics-dir /tmp/areno/tfevent --name rollout/rewards_mean + +Output (text table with a hand-written UTF-8 sparkline trend): + +.. code-block:: text + + metric rollout/rewards_mean + count 5 + steps 0 -> 4 + last 1 (step 4) + min 0.5 + max 1 + mean 0.75 + trend ▁▃▅▇█ + recent step 0: 0.5, step 1: 0.625, step 2: 0.75, step 3: 0.875, step 4: 1 + +For structured output (piped into ``jq`` or other tools): + +.. code-block:: bash + + areno metrics --metrics-dir /tmp/areno/tfevent \ + --name rollout/rewards_mean --json + +.. code-block:: json + + { + "count": 5, + "last": 1.0, + "last_step": 4, + "max": 1.0, + "max_step": 4, + "mean": 0.75, + "min": 0.5, + "min_step": 0, + "name": "rollout/rewards_mean", + "recent": [0.5, 0.625, 0.75, 0.875, 1.0], + "recent_steps": [0, 1, 2, 3, 4], + "trend": [0.0, 0.25, 0.5, 0.75, 1.0] + } + +areno metrics +------------- + +Summarize one metric from local run artifacts. + +Options: + +``--metrics-dir TEXT`` + Directory holding the run's ``events.out.tfevents.*`` artifacts. Omit to use + areno's default metrics log dir (the same ``--metrics-log-dir`` training + writes to by default). + +``--pid INTEGER`` + Filter event files by the ``pid`` suffix embedded in the filename + (``events.out.tfevents...*``). Omit to merge every run writing + into the same directory. + +``--name TEXT`` + Metric tag to summarize, for example ``rollout/rewards_mean``. Omit to list + the available tags found in the directory. When the tag is not found, the + command exits non-zero and prints the available tags so you can pick one. + +``--limit INTEGER`` + Number of recent values to include in ``recent`` / ``recent_steps`` and + ``trend``. Defaults to ``20``. + +``--json`` + Emit a machine-readable JSON object instead of the text table. The ``trend`` + field is the normalized ``[0, 1]`` series; the text table renders the same + series as a UTF-8 sparkline. JSON also exposes ``last_step`` / ``min_step`` / + ``max_step`` / ``mean`` / ``recent_steps`` so machine consumers can read + training progress and pair each recent value with its step + (``jq '.max_step, (.recent_steps, .recent)'``). + +Input contract +-------------- + +* Source: TensorBoard scalar events (``events.out.tfevents.*``). Each tag is + truncated to its last ``500`` points (``EventAccumulator`` with + ``size_guidance={"scalars": 10000}``), ``NaN`` values are skipped, and + ``(name, step, value)`` triples are de-duplicated. +* ``min`` / ``max`` / ``mean`` are computed in a single streaming pass over the + bounded series, so memory stays bounded regardless of run length. +* ``rollout_samples.*.jsonl`` files are **not** read as metrics; only + ``events.out.tfevents.*`` is. + +Step range and training progress +-------------------------------- + +* ``max_step`` (and the high end of the text ``steps`` line) is the latest step + available in the artifacts -- i.e. how far training has progressed. +* ``min_step`` / ``max_step`` describe the **retained window**, not necessarily + the full training history. Because each tag is truncated to its last ``500`` + points, a run longer than ``500`` steps reports ``min_step`` as the earliest + step still in that tail, **not** the start of training. ``max_step`` is always + the current progress. +* ``recent_steps`` is parallel to ``recent`` (equal length, step order), so each + recent value maps back to its absolute step -- useful when correlating with + training logs. + +Limitations (first version) +--------------------------- + +* Only TensorBoard scalars are read. A jsonl fallback source and a friendly + ``--run `` selector backed by the dashboard jobs registry are planned as + follow-ups; use ``--pid`` today to disambiguate runs sharing one directory. +* The command is read-only and CPU-only; it does not require CUDA. + +Examples +-------- + +List the metrics available from the default log dir:: + + areno metrics + +Query one metric from a specific run identified by pid:: + + areno metrics --metrics-dir /tmp/areno/tfevent --pid 12345 \ + --name rollout/rewards_mean --limit 50 diff --git a/docs/reference/cli.rst b/docs/reference/cli.rst index 7e7b7e88..4e5a9200 100644 --- a/docs/reference/cli.rst +++ b/docs/reference/cli.rst @@ -11,4 +11,5 @@ Command pages: * :doc:`/cli/agent` * :doc:`/cli/dataset_loaders` * :doc:`/cli/observability` +* :doc:`/cli/metrics` * :doc:`/cli/diagnostics` diff --git a/tests/test_cli_metrics_cpu.py b/tests/test_cli_metrics_cpu.py new file mode 100644 index 00000000..655df945 --- /dev/null +++ b/tests/test_cli_metrics_cpu.py @@ -0,0 +1,223 @@ +"""CPU tests for the ``areno metrics`` CLI (issue #254). + +Drives the Click command end-to-end with ``CliRunner`` against real +``events.out.tfevents.*`` fixtures so the matrix runs through the same read +path users hit (option wiring, exit codes, available-tag listing, JSON shape). +""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from click.testing import CliRunner +from torch.utils.tensorboard import SummaryWriter + +from areno.cli.metrics import metrics_command + + +def _writer(metrics_dir: Path) -> SummaryWriter: + metrics_dir.mkdir(parents=True, exist_ok=True) + return SummaryWriter(log_dir=str(metrics_dir)) + + +class MetricsCliTest(unittest.TestCase): + def setUp(self) -> None: + self.runner = CliRunner() + + def _invoke(self, args: list[str]): + return self.runner.invoke(metrics_command, args) + + def test_success_table_and_json(self): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) + writer = _writer(d) + try: + for step in range(5): + writer.add_scalar("rollout/rewards_mean", 0.5 + 0.125 * step, step) + writer.add_scalar("train/loss", 2.0, 0) + finally: + writer.close() + + table = self._invoke(["--metrics-dir", str(d), "--name", "rollout/rewards_mean"]) + self.assertEqual(table.exit_code, 0) + self.assertIn("metric rollout/rewards_mean", table.output) + self.assertIn("count 5", table.output) + self.assertIn("steps 0 -> 4", table.output) + self.assertIn("mean", table.output) + self.assertIn("(step 4)", table.output) + self.assertIn("step 0: 0.5", table.output) + sparkline = table.output.split("trend ", 1)[1].splitlines()[0] + self.assertEqual(len(sparkline), 5) + + jres = self._invoke(["--metrics-dir", str(d), "--name", "rollout/rewards_mean", "--json"]) + self.assertEqual(jres.exit_code, 0) + payload = json.loads(jres.output) + self.assertEqual(payload["name"], "rollout/rewards_mean") + self.assertEqual(payload["count"], 5) + self.assertEqual(payload["min"], 0.5) + self.assertEqual(payload["max"], 1.0) + self.assertEqual(payload["max_step"], 4) + self.assertEqual(payload["last_step"], 4) + self.assertEqual(payload["mean"], 0.75) + self.assertEqual(payload["recent_steps"], [0, 1, 2, 3, 4]) + self.assertEqual(len(payload["trend"]), 5) + self.assertEqual(len(payload["recent"]), 5) + + def test_no_name_lists_available_tags_exit_0(self): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) + writer = _writer(d) + try: + writer.add_scalar("rollout/a", 1.0, 0) + writer.add_scalar("train/b", 2.0, 0) + finally: + writer.close() + + res = self._invoke(["--metrics-dir", str(d)]) + self.assertEqual(res.exit_code, 0) + self.assertIn("available metric tags", res.output) + self.assertIn("rollout/a", res.output) + self.assertIn("train/b", res.output) + + def test_unknown_name_lists_available_tags_exit_1(self): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) + writer = _writer(d) + try: + writer.add_scalar("rollout/a", 1.0, 0) + finally: + writer.close() + + res = self._invoke(["--metrics-dir", str(d), "--name", "does/not/exist"]) + self.assertEqual(res.exit_code, 1) + self.assertIn("metric 'does/not/exist' not found", res.output) + self.assertIn("rollout/a", res.output) + self.assertIn("Error:", res.output) + + def test_unknown_name_json_emits_missing_tag(self): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) + writer = _writer(d) + try: + writer.add_scalar("rollout/a", 1.0, 0) + finally: + writer.close() + + res = self._invoke(["--metrics-dir", str(d), "--name", "x", "--json"]) + # The JSON listing is emitted before the ClickException, which follows + # as an "Error:" line. Assert the payload unconditionally so a dropped + # listing fails the test instead of passing silently. + payload = json.loads(res.output.split("\nError:")[0]) + self.assertEqual(payload["available_tags"], ["rollout/a"]) + self.assertEqual(payload["missing"], "x") + self.assertEqual(res.exit_code, 1) + + def test_empty_directory_reports_none_found(self): + with tempfile.TemporaryDirectory() as tmp: + res = self._invoke(["--metrics-dir", tmp, "--name", "anything"]) + self.assertEqual(res.exit_code, 1) + self.assertIn("available tags: (none found)", res.output) + + def test_directory_with_only_rollout_samples_no_event(self): + # rollout_samples.jsonl must NOT be read as metrics; with no event files + # the run yields no metric points. + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) + (d / "rollout_samples.123.jsonl").write_text('{"name":"x","value":1}\n', encoding="utf-8") + res = self._invoke(["--metrics-dir", str(d), "--name", "x"]) + self.assertEqual(res.exit_code, 1) + self.assertIn("available tags: (none found)", res.output) + + def test_missing_directory_exits_1(self): + res = self._invoke(["--metrics-dir", "/definitely/missing/areno/zzz", "--name", "x"]) + self.assertEqual(res.exit_code, 1) + self.assertIn("metrics dir not found", res.output) + + def test_missing_tensorboard_reports_clear_error(self): + # Simulate a missing tensorboard install by mapping the module to None, + # which makes the CLI's `import tensorboard` probe raise ImportError. + with tempfile.TemporaryDirectory() as tmp: + with patch.dict(sys.modules, {"tensorboard": None}): + res = self._invoke(["--metrics-dir", tmp, "--name", "train/loss"]) + self.assertEqual(res.exit_code, 1) + self.assertIn("tensorboard is not installed", res.output) + + def test_single_point_series(self): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) + writer = _writer(d) + try: + writer.add_scalar("rollout/one", 9.0, 7) + finally: + writer.close() + + res = self._invoke(["--metrics-dir", str(d), "--name", "rollout/one"]) + self.assertEqual(res.exit_code, 0) + self.assertIn("count 1", res.output) + self.assertIn("last 9", res.output) + sparkline = res.output.split("trend ", 1)[1].splitlines()[0] + self.assertEqual(len(sparkline), 1) + + def test_all_nan_series_is_empty(self): + # A tag whose every point is NaN must not appear as available at all. + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) + writer = _writer(d) + try: + writer.add_scalar("rollout/only_nan", float("nan"), 0) + writer.add_scalar("rollout/only_nan", float("nan"), 1) + writer.add_scalar("rollout/real", 1.0, 0) + finally: + writer.close() + + listing = self._invoke(["--metrics-dir", str(d)]) + self.assertNotIn("only_nan", listing.output) + self.assertIn("rollout/real", listing.output) + + res = self._invoke(["--metrics-dir", str(d), "--name", "rollout/only_nan"]) + self.assertEqual(res.exit_code, 1) + self.assertIn("rollout/only_nan", res.output) + + def test_truncation_caps_at_500_points(self): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) + writer = _writer(d) + try: + for step in range(600): + writer.add_scalar("rollout/big", float(step), step) + finally: + writer.close() + + res = self._invoke(["--metrics-dir", str(d), "--name", "rollout/big"]) + self.assertEqual(res.exit_code, 0) + self.assertIn("count 500", res.output) + # --limit (default 20) bounds the sparkline to the recent window; + # count is the full 500, but trend/recent stay bounded. + sparkline = res.output.split("trend ", 1)[1].splitlines()[0] + self.assertLessEqual(len(sparkline), 20) + + def test_pid_filter_excludes_other_run(self): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) + writer = _writer(d) + try: + writer.add_scalar("rollout/a", 1.0, 0) + finally: + writer.close() + + match = self._invoke(["--metrics-dir", str(d), "--pid", str(os.getpid()), "--name", "rollout/a"]) + self.assertEqual(match.exit_code, 0) + + miss = self._invoke(["--metrics-dir", str(d), "--pid", "99999999", "--name", "rollout/a"]) + self.assertEqual(miss.exit_code, 1) + self.assertIn("available tags: (none found)", miss.output) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_metric_reader_cpu.py b/tests/test_metric_reader_cpu.py new file mode 100644 index 00000000..f057d990 --- /dev/null +++ b/tests/test_metric_reader_cpu.py @@ -0,0 +1,230 @@ +"""CPU tests for the pure ``areno.api.metric_reader`` helpers (issue #254).""" + +from __future__ import annotations + +import json +import os +import tempfile +import unittest +from pathlib import Path + +from torch.utils.tensorboard import SummaryWriter + +from areno.api import metric_reader as mr + + +def _write_fixture(metrics_dir: Path) -> int: + """Write real tfevents covering normal / NaN / single-point series. + + Returns the writer process pid so pid-filter tests have a known match. + Values use powers-of-two fractions to survive TensorBoard's float32 + round-trip, so assertions can compare for exact equality. + """ + metrics_dir.mkdir(parents=True, exist_ok=True) + writer = SummaryWriter(log_dir=str(metrics_dir)) + try: + for step in range(5): + writer.add_scalar("rollout/rewards_mean", 0.5 + 0.125 * step, step) + writer.add_scalar("rollout/loss_with_nan", float("nan"), 0) + writer.add_scalar("rollout/loss_with_nan", 1.5, 1) + writer.add_scalar("rollout/single_point", 9.0, 7) + finally: + writer.close() + return os.getpid() + + +class ReadScalarPointsTest(unittest.TestCase): + def test_reads_points_skipping_nan_and_preserving_order(self): + with tempfile.TemporaryDirectory() as tmp: + _write_fixture(Path(tmp)) + points = mr.read_scalar_points(tmp) + + names_steps = [(p["name"], p["step"], p["value"]) for p in points] + self.assertEqual( + names_steps, + [ + ("rollout/rewards_mean", 0, 0.5), + ("rollout/rewards_mean", 1, 0.625), + ("rollout/rewards_mean", 2, 0.75), + ("rollout/rewards_mean", 3, 0.875), + ("rollout/rewards_mean", 4, 1.0), + ("rollout/loss_with_nan", 1, 1.5), + ("rollout/single_point", 7, 9.0), + ], + ) + self.assertTrue(all("time" in p and p["time"] for p in points)) + + def test_empty_directory_returns_empty(self): + with tempfile.TemporaryDirectory() as tmp: + self.assertEqual(mr.read_scalar_points(tmp), []) + self.assertEqual(mr.tensorboard_event_sources(Path(tmp), None), [Path(tmp)]) + + def test_pid_filter_matches_filename_suffix(self): + with tempfile.TemporaryDirectory() as tmp: + pid = _write_fixture(Path(tmp)) + self.assertEqual(len(mr.read_scalar_points(tmp, pid=pid)), 7) + self.assertEqual(mr.read_scalar_points(tmp, pid=99999999), []) + + def test_truncation_keeps_last_500(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) + path.mkdir(parents=True, exist_ok=True) + writer = SummaryWriter(log_dir=str(path)) + try: + for step in range(600): + writer.add_scalar("rollout/big", float(step), step) + finally: + writer.close() + points = mr.read_scalar_points(tmp, pid=None) + self.assertEqual(len(points), 500) + self.assertEqual(points[0]["step"], 100) + self.assertEqual(points[-1]["step"], 599) + + +class SummarizeMetricTest(unittest.TestCase): + @staticmethod + def _points(name: str, values: list[float], *, start_step: int = 0) -> list[dict]: + return [ + {"name": name, "value": value, "step": start_step + i, "time": "t"} + for i, value in enumerate(values) + ] + + def test_normal_series_streaming_min_max_and_trend(self): + points = self._points("m", [1.0, 3.0, 2.0, 6.0, 4.0]) + summary = mr.summarize_metric(points, "m", recent_n=3) + + self.assertEqual(summary["count"], 5) + self.assertEqual(summary["last"], 4.0) + self.assertEqual(summary["last_step"], 4) + self.assertEqual(summary["min_step"], 0) + self.assertEqual(summary["max_step"], 4) + self.assertEqual(summary["mean"], 3.2) + self.assertEqual(summary["min"], 1.0) + self.assertEqual(summary["max"], 6.0) + self.assertEqual(summary["recent"], [2.0, 6.0, 4.0]) + self.assertEqual(summary["recent_steps"], [2, 3, 4]) + # trend shares the recent window, so --limit bounds the sparkline length + self.assertEqual(len(summary["trend"]), 3) + self.assertEqual(summary["trend"], [0.0, 1.0, 0.5]) + + def test_trend_bounded_by_recent_n(self): + points = self._points("m", [float(i) for i in range(10)]) + summary = mr.summarize_metric(points, "m", recent_n=5) + + self.assertEqual(len(summary["recent"]), 5) + self.assertEqual(len(summary["trend"]), 5) + self.assertEqual(summary["trend"], [0.0, 0.25, 0.5, 0.75, 1.0]) + self.assertEqual(summary["recent_steps"], [5, 6, 7, 8, 9]) + self.assertEqual(summary["max_step"], 9) + self.assertEqual(summary["mean"], 4.5) + + def test_flat_series_trend_is_half(self): + points = self._points("m", [2.0, 2.0, 2.0]) + summary = mr.summarize_metric(points, "m") + self.assertEqual(summary["trend"], [0.5, 0.5, 0.5]) + self.assertEqual(summary["min"], 2.0) + self.assertEqual(summary["max"], 2.0) + self.assertEqual(summary["mean"], 2.0) + self.assertEqual(summary["recent_steps"], [0, 1, 2]) + + def test_single_point_series(self): + points = self._points("m", [9.0]) + summary = mr.summarize_metric(points, "m") + self.assertEqual(summary["count"], 1) + self.assertEqual(summary["last"], 9.0) + self.assertEqual(summary["last_step"], 0) + self.assertEqual(summary["min_step"], 0) + self.assertEqual(summary["max_step"], 0) + self.assertEqual(summary["mean"], 9.0) + self.assertEqual(summary["min"], 9.0) + self.assertEqual(summary["max"], 9.0) + self.assertEqual(summary["recent"], [9.0]) + self.assertEqual(summary["recent_steps"], [0]) + self.assertEqual(summary["trend"], [0.5]) + + def test_unknown_name_yields_empty_summary(self): + points = self._points("m", [1.0, 2.0]) + summary = mr.summarize_metric(points, "does/not/exist") + self.assertEqual(summary["count"], 0) + self.assertIsNone(summary["last"]) + self.assertIsNone(summary["last_step"]) + self.assertIsNone(summary["min_step"]) + self.assertIsNone(summary["max_step"]) + self.assertIsNone(summary["mean"]) + self.assertEqual(summary["recent"], []) + self.assertEqual(summary["recent_steps"], []) + self.assertEqual(summary["trend"], []) + + def test_nan_values_filtered_by_number_like(self): + points = [ + {"name": "m", "value": 1.0, "step": 0, "time": "t"}, + {"name": "m", "value": float("nan"), "step": 1, "time": "t"}, + {"name": "m", "value": 3.0, "step": 2, "time": "t"}, + ] + summary = mr.summarize_metric(points, "m") + self.assertEqual(summary["count"], 2) + self.assertEqual(summary["min"], 1.0) + self.assertEqual(summary["max"], 3.0) + self.assertEqual(summary["mean"], 2.0) + self.assertEqual(summary["min_step"], 0) + self.assertEqual(summary["max_step"], 2) + + +class ListAndRenderTest(unittest.TestCase): + def test_list_available_tags_sorted(self): + points = [ + {"name": "c", "value": 1.0, "step": 0, "time": "t"}, + {"name": "a", "value": 1.0, "step": 0, "time": "t"}, + {"name": "b", "value": 1.0, "step": 0, "time": "t"}, + {"name": "", "value": 1.0, "step": 0, "time": "t"}, + ] + self.assertEqual(mr.list_available_tags(points), ["a", "b", "c"]) + + def test_render_table_contains_sparkline_and_fields(self): + points = [ + {"name": "m", "value": 1.0, "step": 0, "time": "t"}, + {"name": "m", "value": 5.0, "step": 1, "time": "t"}, + {"name": "m", "value": 3.0, "step": 2, "time": "t"}, + ] + summary = mr.summarize_metric(points, "m") + rendered = mr.render_table(summary) + + self.assertIn("metric m", rendered) + self.assertIn("count 3", rendered) + self.assertIn("steps 0 -> 2", rendered) + self.assertIn("last 3", rendered) + self.assertIn("(step 2)", rendered) + self.assertIn("mean", rendered) + self.assertIn("step 0: 1", rendered) + self.assertIn("step 2: 3", rendered) + sparkline = rendered.split("trend ", 1)[1].splitlines()[0] + self.assertEqual(len(sparkline), 3) + for glyph in sparkline: + self.assertIn(glyph, "▁▂▃▄▅▆▇█") + + def test_render_table_handles_empty_trend(self): + rendered = mr.render_table(mr.summarize_metric([], "missing")) + self.assertIn("count 0", rendered) + self.assertIn("last -", rendered) + + def test_render_json_round_trips_and_keeps_trend_array(self): + points = [ + {"name": "m", "value": 1.0, "step": 0, "time": "t"}, + {"name": "m", "value": 3.0, "step": 1, "time": "t"}, + ] + summary = mr.summarize_metric(points, "m") + decoded = json.loads(json.dumps(mr.render_json(summary))) + self.assertEqual(decoded["name"], "m") + self.assertEqual(decoded["count"], 2) + self.assertEqual(decoded["last"], 3.0) + self.assertEqual(decoded["last_step"], 1) + self.assertEqual(decoded["min_step"], 0) + self.assertEqual(decoded["max_step"], 1) + self.assertEqual(decoded["mean"], 2.0) + self.assertEqual(decoded["recent_steps"], [0, 1]) + self.assertEqual(decoded["trend"], [0.0, 1.0]) + self.assertEqual(decoded["recent"], [1.0, 3.0]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_metric_reader_equivalence_cpu.py b/tests/test_metric_reader_equivalence_cpu.py new file mode 100644 index 00000000..dd7a0d3b --- /dev/null +++ b/tests/test_metric_reader_equivalence_cpu.py @@ -0,0 +1,148 @@ +"""Equivalence guard: dashboard read-side output before vs after the metric_reader extract (issue #254). + +Locks the current output of ``DashboardState._load_tensorboard_scalars`` +(``job.metrics`` + ``job.timeperf``) against a real +``events.out.tfevents.*`` fixture written via ``SummaryWriter``. After the +extract switches ``_load_tensorboard_scalars`` to call +``areno.api.metric_reader``, this test must keep passing unchanged -- it is +the regression guard that proves the refactor is behavior-preserving. +""" + +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from areno.dashboard.server import DashboardState, Job, ROOT + + +def _write_tfevents_fixture(metrics_dir: Path) -> None: + """Write a small set of real TensorBoard event files. + + Tags cover every branch of ``_load_tensorboard_scalars``: normal scalars, + NaN-skip, single-point series, and the timeperf ``time/`` + + ``train/step_*_time_s`` tags that drive the by_step aggregation. Values + use powers-of-two fractions to survive TensorBoard's float32 round-trip. + """ + from torch.utils.tensorboard import SummaryWriter + + metrics_dir.mkdir(parents=True, exist_ok=True) + writer = SummaryWriter(log_dir=str(metrics_dir)) + try: + for step in range(5): + writer.add_scalar("rollout/rewards_mean", 0.5 + 0.125 * step, step) + writer.add_scalar("rollout/loss_with_nan", float("nan"), 0) + writer.add_scalar("rollout/loss_with_nan", 1.5, 1) + writer.add_scalar("rollout/loss_with_nan", float("nan"), 2) + writer.add_scalar("rollout/loss_with_nan", 2.5, 3) + writer.add_scalar("rollout/single_point", 9.0, 7) + for step in range(3): + writer.add_scalar("time/rollout", 2.0 + step, step) + writer.add_scalar("time/train", 1.0 + step, step) + writer.add_scalar("train/step_e2e_time_s", 3.0 + step, step) + finally: + writer.close() + + +def _load_job_metrics(metrics_dir: str) -> Job: + """Drive the dashboard read path exactly as the server does.""" + state = DashboardState() + job = Job(kind="train", name="probe", command=[], config={}, metrics_dir=metrics_dir) + state.jobs[job.id] = job + path = (ROOT / metrics_dir).resolve() + state._load_tensorboard_scalars(job, path) + return job + + +def _strip_nondeterministic(points: list[dict]) -> list[dict]: + """Drop the ``time`` field, which is stamped with wall-clock at read time.""" + return [ + {"name": p.get("name"), "value": p.get("value"), "step": int(p.get("step") or 0)} + for p in points + ] + + +def _strip_timeperf_nondeterministic(rows: list[dict]) -> list[dict]: + """Drop ``time`` from timeperf rows; keep the deterministic aggregation.""" + return [{k: v for k, v in dict(row).items() if k != "time"} for row in rows] + + +class MetricReaderEquivalenceTest(unittest.TestCase): + def test_locks_metric_points_and_timeperf(self): + with tempfile.TemporaryDirectory() as tmp: + metrics_dir = Path(tmp) + _write_tfevents_fixture(metrics_dir) + job = _load_job_metrics(str(metrics_dir)) + + metrics = _strip_nondeterministic(job.metrics) + timeperf = _strip_timeperf_nondeterministic(job.timeperf) + + self.assertEqual( + metrics, + [ + {"name": "rollout/rewards_mean", "value": 0.5, "step": 0}, + {"name": "rollout/rewards_mean", "value": 0.625, "step": 1}, + {"name": "rollout/rewards_mean", "value": 0.75, "step": 2}, + {"name": "rollout/rewards_mean", "value": 0.875, "step": 3}, + {"name": "rollout/rewards_mean", "value": 1.0, "step": 4}, + {"name": "rollout/loss_with_nan", "value": 1.5, "step": 1}, + {"name": "rollout/loss_with_nan", "value": 2.5, "step": 3}, + {"name": "rollout/single_point", "value": 9.0, "step": 7}, + {"name": "time/rollout", "value": 2.0, "step": 0}, + {"name": "time/rollout", "value": 3.0, "step": 1}, + {"name": "time/rollout", "value": 4.0, "step": 2}, + {"name": "time/train", "value": 1.0, "step": 0}, + {"name": "time/train", "value": 2.0, "step": 1}, + {"name": "time/train", "value": 3.0, "step": 2}, + {"name": "train/step_e2e_time_s", "value": 3.0, "step": 0}, + {"name": "train/step_e2e_time_s", "value": 4.0, "step": 1}, + {"name": "train/step_e2e_time_s", "value": 5.0, "step": 2}, + ], + ) + + self.assertEqual( + timeperf, + [ + { + "step": 0, + "segments": [ + {"name": "rollout", "seconds": 2.0}, + {"name": "train", "seconds": 1.0}, + ], + "rollout_s": 2.0, + "train_s": 1.0, + "other_s": 0.0, + "total_s": 3.0, + "source": "metrics", + }, + { + "step": 1, + "segments": [ + {"name": "rollout", "seconds": 3.0}, + {"name": "train", "seconds": 2.0}, + ], + "rollout_s": 3.0, + "train_s": 2.0, + "other_s": 0.0, + "total_s": 4.0, + "source": "metrics", + }, + { + "step": 2, + "segments": [ + {"name": "rollout", "seconds": 4.0}, + {"name": "train", "seconds": 3.0}, + ], + "rollout_s": 4.0, + "train_s": 3.0, + "other_s": 0.0, + "total_s": 5.0, + "source": "metrics", + }, + ], + ) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file