diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fb5051c..20ea111 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -62,7 +62,7 @@ jobs: # trials each) are heavy and run in the nightly instead. No marker filter, so # both subprocess and unit tests run. env: - TEST_PATH: "autodata_tests/e2e/ autodata_tests/interfaces/ autodata_tests/core/ autodata_tests/utils/ autodata_tests/retargeting/" + TEST_PATH: "autodata_tests/e2e/ autodata_tests/interfaces/ autodata_tests/core/ autodata_tests/utils/ autodata_tests/retargeting/ autodata_tests/metrics/" steps: - name: nvidia-smi diff --git a/autodata_metrics/__init__.py b/autodata_metrics/__init__.py new file mode 100644 index 0000000..c9d8546 --- /dev/null +++ b/autodata_metrics/__init__.py @@ -0,0 +1,13 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Offline metrics of AutoData datasets and generation runs. + +Producers record evidence; :func:`build_report` computes every registered metric from it without a simulator. +Importing this package registers the built-in metrics. +""" + +from autodata_metrics import viability +from autodata_metrics.evidence import Episode, Evidence +from autodata_metrics.registry import AREAS, REGISTERED_METRICS, MetricSpec, NotApplicable, Value, metric +from autodata_metrics.report import build_report, format_report, gate_violations diff --git a/autodata_metrics/evidence.py b/autodata_metrics/evidence.py new file mode 100644 index 0000000..5f5a8b2 --- /dev/null +++ b/autodata_metrics/evidence.py @@ -0,0 +1,90 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""What a metric reads: a dataset's episodes and ``env_args``, and its generation run's record and attempt log.""" + +from __future__ import annotations + +import h5py +import json +import numpy as np +from collections.abc import Mapping +from typing import Any + + +def _has_path(record: Any, dotted_key: str) -> bool: + """Whether ``record[a][b]`` exists for ``dotted_key`` ``"a.b"``.""" + for part in dotted_key.split("."): + if not isinstance(record, Mapping) or part not in record: + return False + record = record[part] + return True + + +def read_env_args(h5_file: h5py.File) -> dict[str, Any]: + """The dataset's ``env_args``, or empty when they are not a JSON object.""" + raw_env_args = h5_file["data"].attrs.get("env_args", "{}") + env_args = json.loads(raw_env_args.decode() if isinstance(raw_env_args, bytes) else raw_env_args) + return env_args if isinstance(env_args, dict) else {} + + +class Episode: + """One episode of a dataset; arrays are read when accessed.""" + + def __init__(self, group: h5py.Group) -> None: + self.name = group.name.rsplit("/", 1)[-1] + self._group = group + + def __contains__(self, key: str) -> bool: + return key in self._group + + def __getitem__(self, key: str) -> np.ndarray: + return self._group[key][()] + + +class Evidence: + """A dataset, and optionally the generation run that produced it, as metrics read them. + + Attributes: + episodes: The dataset's episodes. + env_args: The dataset's ``env_args``; its ``autodata`` entry was written by the producer. + result: Generation result, or None for a dataset without a run record or an interrupted run. + run: The result's ``run`` block, or empty. + attempts: Attempt-log lines, or None. + source: Evidence of the source dataset the run started from, or None. + """ + + def __init__( + self, + h5_file: h5py.File, + result: Mapping[str, Any] | None = None, + attempts: list[dict[str, Any]] | None = None, + source: Evidence | None = None, + ) -> None: + self.env_args = read_env_args(h5_file) + self.episodes = [Episode(group) for group in h5_file["data"].values()] + self.result = result + self.run: Mapping[str, Any] = (result or {}).get("run", {}) + self.attempts = attempts + self.source = source + + def has(self, key: str) -> bool: + """Whether the evidence key ``key`` is present. + + A key is one of: + + - an HDF5 path present in every episode, such as ``step_success``; + - ``env_args.`` or ``run.``, a dotted path into ``env_args`` or the run block; + - ``attempts``, the attempt log, or ``attempts.``, a dotted path present in every attempt line; + - ``source``, the source dataset, or ``source.``, a key of the source dataset. + """ + head, _, rest = key.partition(".") + if head == "source": + return self.source is not None and (not rest or self.source.has(rest)) + if head == "attempts": + return bool(self.attempts) and (not rest or all(_has_path(line, rest) for line in self.attempts)) + if head == "env_args": + return _has_path(self.env_args, rest) + if head == "run": + return _has_path(self.run, rest) + return bool(self.episodes) and all(key in episode for episode in self.episodes) diff --git a/autodata_metrics/registry.py b/autodata_metrics/registry.py new file mode 100644 index 0000000..c043c6f --- /dev/null +++ b/autodata_metrics/registry.py @@ -0,0 +1,104 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Registry of dataset and run metrics. + +A metric is a function of :class:`~autodata_metrics.evidence.Evidence` registered with :func:`metric`. It declares +the evidence it ``requires``; the runner reports it as not applicable, naming the missing key, instead of calling it +without that evidence. +""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, Literal + +if TYPE_CHECKING: + from autodata_metrics.evidence import Evidence + +AREAS = ("validity", "diversity", "viability", "diagnostics") + + +class NotApplicable(Exception): + """Raised by a metric whose evidence is present but does not let it apply.""" + + +@dataclass(frozen=True) +class Value: + """What a metric computed. + + Attributes: + value: A number, or named numbers such as percentiles. + n: Sample count the value rests on. + details: Context a reader needs to interpret the value. + """ + + value: float | dict[str, float] + n: int + details: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class MetricSpec: + """A registered metric and how to report it.""" + + name: str + compute: Callable[[Evidence], Value] + area: str + signal: str + scope: Literal["dataset", "run"] + unit: str + better: Literal["higher", "lower"] | None + requires: tuple[str, ...] + primary: str | None + version: int + + +REGISTERED_METRICS: dict[str, MetricSpec] = {} + + +def metric( + name: str, + *, + area: str, + signal: str, + scope: Literal["dataset", "run"], + unit: str, + better: Literal["higher", "lower"] | None = None, + requires: tuple[str, ...] | list[str] = (), + primary: str | None = None, + version: int = 1, +) -> Callable[[Callable[[Evidence], Value]], Callable[[Evidence], Value]]: + """Register the decorated function as the metric ``name``. + + Args: + name: Unique metric name, used in reports and gates. + area: One of :data:`AREAS`. + signal: What the metric measures within its area, e.g. ``"task correctness"``. + scope: ``"dataset"`` metrics read episodes; ``"run"`` metrics read attempts and the run record. + unit: Unit of the value. + better: Which direction is better, if either. + requires: Evidence keys; see :meth:`~autodata_metrics.evidence.Evidence.has`. + primary: For a dict value, the entry a gate checks by default. + version: Raise it whenever the definition changes, so gates pinned to the old one fail. + """ + assert area in AREAS, f"Metric {name!r} has area {area!r}; expected one of {AREAS}." + + def register(compute: Callable[[Evidence], Value]) -> Callable[[Evidence], Value]: + assert name not in REGISTERED_METRICS, f"Metric {name!r} is already registered." + REGISTERED_METRICS[name] = MetricSpec( + name=name, + compute=compute, + area=area, + signal=signal, + scope=scope, + unit=unit, + better=better, + requires=tuple(requires), + primary=primary, + version=version, + ) + return compute + + return register diff --git a/autodata_metrics/report.py b/autodata_metrics/report.py new file mode 100644 index 0000000..f8724aa --- /dev/null +++ b/autodata_metrics/report.py @@ -0,0 +1,238 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Compute the registered metrics of a dataset and its run, render them, and check them against a gate.""" + +from __future__ import annotations + +import contextlib +import h5py +import json +import math +import numbers +import yaml +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +from autodata_metrics.evidence import Evidence, read_env_args +from autodata_metrics.registry import AREAS, REGISTERED_METRICS, MetricSpec, NotApplicable +from autodata_utils.generation_result import attempts_file_for, code_revision, input_record + +SCHEMA_VERSION = 1 + + +def read_run( + result_file: str, run_id: str | None +) -> tuple[dict[str, Any] | None, list[dict[str, Any]] | None, list[str]]: + """The generation result and attempt log of run ``run_id``, each None when missing, and any warnings. + + A rerun with the same result file leaves the earlier result in place until it completes, so a result from + another run is set aside and the run is reported from its own attempt log. + """ + result_path = Path(result_file) + result = json.loads(result_path.read_text(encoding="utf-8")) if result_path.exists() else None + warnings = [] + if result is not None and run_id is not None and result.get("run", {}).get("run_id") != run_id: + warnings.append(f"{result_file} belongs to another run, so the run is reported from its attempt log") + result = None + if result is not None and "attempts_file" in result.get("run", {}): + attempts_path = result_path.parent / result["run"]["attempts_file"] + elif run_id is not None: + attempts_path = Path(attempts_file_for(result_file, run_id)) + else: + return result, None, warnings + attempts = None + if attempts_path.exists(): + lines = attempts_path.read_text(encoding="utf-8").splitlines() + attempts = [json.loads(line) for line in lines if line.strip()] + return result, attempts, warnings + + +def check_same_run(evidence: Evidence) -> list[str]: + """Refuse a dataset and run record that come from different runs; return warnings for unverifiable data. + + Run ids alone are not enough: merging datasets keeps the first file's ``env_args``, so the exported episode + count must match too. + + Raises: + ValueError: If the dataset, result and attempt log disagree on their run or their counts. + """ + num_episodes = len(evidence.episodes) + result = evidence.result + if result is not None and num_episodes != result["num_success"]: + raise ValueError( + f"The dataset holds {num_episodes} episodes but the run counts {result['num_success']} successes." + ) + if result is not None and evidence.attempts is not None: + logged = len(evidence.attempts), sum(line["success"] for line in evidence.attempts) + if logged != (result["num_attempts"], result["num_success"]): + raise ValueError( + f"The attempt log holds {logged[0]} attempts and {logged[1]} successes but the result counts" + f" {result['num_attempts']} and {result['num_success']}." + ) + entry = evidence.env_args.get("autodata") + if entry is None: + return ["the dataset has no autodata entry, so it is matched to its run by episode count only"] + if entry["exported_episodes"] != num_episodes: + raise ValueError( + f"The dataset holds {num_episodes} episodes but its run exported {entry['exported_episodes']};" + " it was merged or edited after export." + ) + run_ids = {entry["run_id"]} | {line["run_id"] for line in evidence.attempts or []} + if "run_id" in evidence.run: + run_ids.add(evidence.run["run_id"]) + if len(run_ids) != 1: + raise ValueError(f"The dataset, result and attempt log come from different runs: {sorted(run_ids)}") + return [] + + +def evaluate(spec: MetricSpec, evidence: Evidence) -> dict[str, Any]: + """One report row: the metric's value, or why it is not applicable or failed.""" + row = { + "area": spec.area, + "signal": spec.signal, + "scope": spec.scope, + "unit": spec.unit, + "better": spec.better, + "primary": spec.primary, + "version": spec.version, + } + missing = next((key for key in spec.requires if not evidence.has(key)), None) + if missing is not None: + return {**row, "status": "not_applicable", "reason": f"missing {missing}"} + try: + value = spec.compute(evidence) + except NotApplicable as exc: + return {**row, "status": "not_applicable", "reason": str(exc)} + except Exception as exc: # A failing metric must not sink the report of every other metric. + return {**row, "status": "error", "reason": f"{type(exc).__name__}: {exc}"} + if not _is_finite(value.value): + return {**row, "status": "error", "reason": f"non-finite value {value.value}"} + return {**row, "status": "ok", "value": value.value, "n": value.n, "details": _finite_or_none(value.details)} + + +def _is_finite(value: Any) -> bool: + """Whether ``value`` is a finite number, or a mapping of finite numbers.""" + if isinstance(value, Mapping): + return all(_is_finite(item) for item in value.values()) + return isinstance(value, numbers.Real) and math.isfinite(value) + + +def _finite_or_none(value: Any) -> Any: + """``value`` with every non-finite number replaced by None, so the report stays valid JSON.""" + value = value.tolist() if hasattr(value, "tolist") else value + if isinstance(value, Mapping): + return {key: _finite_or_none(item) for key, item in value.items()} + if isinstance(value, list | tuple): + return [_finite_or_none(item) for item in value] + return None if isinstance(value, float) and not math.isfinite(value) else value + + +def build_report( + dataset: str, + result_file: str | None = None, + source: str | None = None, + metrics: Mapping[str, MetricSpec] | None = None, +) -> dict[str, Any]: + """Compute every metric for a dataset and, when ``result_file`` is given, its generation run. + + Args: + dataset: HDF5 dataset. + result_file: Generation result (``generate_dataset.py --result_file``); its attempt log is read beside it. + source: Source dataset the run started from. + metrics: Metrics to compute; defaults to every registered metric. + + Raises: + ValueError: If the dataset, result, attempt log and source do not all come from one run. + """ + with contextlib.ExitStack() as stack: + h5_file = stack.enter_context(h5py.File(dataset, "r")) + run_id = read_env_args(h5_file).get("autodata", {}).get("run_id") + result, attempts, warnings = read_run(result_file, run_id) if result_file else (None, None, []) + source_evidence = Evidence(stack.enter_context(h5py.File(source, "r"))) if source else None + evidence = Evidence(h5_file, result, attempts, source_evidence) + warnings += check_same_run(evidence) + source_record = input_record(source) + recorded_source = evidence.run.get("inputs", {}).get("source") + if source_record and recorded_source and source_record["sha256"] != recorded_source["sha256"]: + raise ValueError(f"The source {source} is not the one the run started from, {recorded_source['path']}.") + metrics = REGISTERED_METRICS if metrics is None else metrics + rows = {name: evaluate(spec, evidence) for name, spec in metrics.items()} + num_episodes = len(evidence.episodes) + return { + "schema_version": SCHEMA_VERSION, + "inputs": { + "dataset": input_record(dataset), + "result_file": result_file, + "source": source_record, + "run_id": run_id, + "revision": code_revision(), + }, + "num_episodes": num_episodes, + "warnings": warnings, + "metrics": rows, + } + + +def gate_violations(report: Mapping[str, Any], gate_file: str) -> list[str]: + """Bounds in ``gate_file`` that the report violates. + + The gate maps metric names to ``min``/``max`` bounds, an optional ``key`` into a dict value (default: the + metric's primary entry) and an optional ``version``. A gated metric that is missing, not applicable, errored or + at another version is a violation. + + Raises: + ValueError: If a gate entry has a bound that is not a finite number, or bounds a dict value without a + ``key`` and the metric has no primary entry. + """ + violations = [] + for name, bound in yaml.safe_load(Path(gate_file).read_text(encoding="utf-8")).items(): + limits = [bound[side] for side in ("min", "max") if side in bound] + if not all(_is_bound(limit) for limit in limits): + raise ValueError(f"Gate entry {name!r} has a bound that is not a finite number: {limits}.") + row = report["metrics"].get(name) + if row is None or row["status"] != "ok": + violations.append(f"{name}: {'not reported' if row is None else row['status']}") + continue + if "version" in bound and bound["version"] != row["version"]: + violations.append(f"{name}: version {row['version']}, gate written for version {bound['version']}") + continue + key = bound.get("key", row["primary"]) + value = row["value"].get(key) if key else row["value"] + if isinstance(value, Mapping): + raise ValueError(f"Gate entry {name!r} needs a key into {sorted(value)}.") + if not _is_finite(value): + violations.append(f"{name}: {value!r} is not a finite number") + continue + if "min" in bound and value < bound["min"]: + violations.append(f"{name}: {value:.4g} is below {bound['min']}") + if "max" in bound and value > bound["max"]: + violations.append(f"{name}: {value:.4g} is above {bound['max']}") + return violations + + +def _is_bound(limit: Any) -> bool: + """Whether a gate bound is a finite number; YAML reads ``true`` and ``false`` as bools, which are not bounds.""" + return isinstance(limit, int | float) and not isinstance(limit, bool) and math.isfinite(limit) + + +def _format_value(value: Any) -> str: + if isinstance(value, Mapping): + return ", ".join(f"{key}={_format_value(item)}" for key, item in value.items()) or "none" + return f"{value:.3g}" if isinstance(value, float) else str(value) + + +def format_report(report: Mapping[str, Any]) -> str: + """The report as an aligned text table, grouped by area.""" + rows = [("Metric", "Area", "Value", "Unit", "n")] + for name, row in sorted(report["metrics"].items(), key=lambda item: (AREAS.index(item[1]["area"]), item[0])): + if row["status"] == "ok": + rows.append((name, row["area"], _format_value(row["value"]), row["unit"], str(row["n"]))) + else: + rows.append((name, row["area"], f"{row['status']}: {row['reason']}", "", "")) + widths = [max(len(row[column]) for row in rows) for column in range(len(rows[0]))] + lines = [" ".join(cell.ljust(width) for cell, width in zip(row, widths)).rstrip() for row in rows] + lines.insert(1, " ".join("-" * width for width in widths)) + lines += [f"warning: {warning}" for warning in report["warnings"]] + return "\n".join(lines) diff --git a/autodata_metrics/viability.py b/autodata_metrics/viability.py new file mode 100644 index 0000000..a8de7a1 --- /dev/null +++ b/autodata_metrics/viability.py @@ -0,0 +1,164 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Viability metrics: what producing the dataset cost, and diagnostics of why attempts failed.""" + +from __future__ import annotations + +import math +import numpy as np +from collections import Counter +from typing import Any + +from autodata_metrics.evidence import Evidence +from autodata_metrics.registry import NotApplicable, Value, metric + +Z_95 = 1.959963984540054 +MIN_ATTEMPTS_PER_ENV_FOR_STOP_CORRECTION = 3 +"""Below this many logged attempts in some env, the stop correction is itself biased.""" + + +def wilson_interval(successes: int, trials: int, z: float = Z_95) -> tuple[float, float]: + """Wilson score interval for a binomial proportion.""" + assert trials > 0, "trials must be positive" + p = successes / trials + denominator = 1 + z**2 / trials + center = (p + z**2 / (2 * trials)) / denominator + half_width = z * math.sqrt(p * (1 - p) / trials + z**2 / (4 * trials**2)) / denominator + return max(0.0, center - half_width), min(1.0, center + half_width) + + +def attempts_before_stop(attempts: list[dict[str, Any]], num_envs: int) -> list[dict[str, Any]] | None: + """Attempts that started before the first env completed its last logged attempt. + + Attempts running when generation stops are cancelled unlogged, and long attempts are the likelier to be running. + Each env's cancelled attempt started after that env's last completion, so this start-time subset was fully + logged. None when some env logged fewer than :data:`MIN_ATTEMPTS_PER_ENV_FOR_STOP_CORRECTION` attempts. + """ + last_end: dict[int, float] = {} + logged: Counter[int] = Counter() + for attempt in attempts: + last_end[attempt["env_id"]] = max(last_end.get(attempt["env_id"], 0.0), attempt["end_time_s"]) + logged[attempt["env_id"]] += 1 + if len(logged) < num_envs or min(logged.values()) < MIN_ATTEMPTS_PER_ENV_FOR_STOP_CORRECTION: + return None + cutoff = min(last_end.values()) + return [attempt for attempt in attempts if attempt["start_time_s"] < cutoff] + + +def _counted_attempts(ev: Evidence) -> tuple[list[dict[str, Any]], bool]: + """The stop-corrected attempts when the log allows the correction, else all attempts; and which one it is.""" + corrected = attempts_before_stop(ev.attempts, ev.run["num_envs"]) if "num_envs" in ev.run else None + return (corrected, True) if corrected is not None else (ev.attempts, False) + + +@metric( + "generation_success_rate", + area="viability", + signal="generation performance", + scope="run", + unit="fraction", + better="higher", + requires=["attempts"], +) +def generation_success_rate(ev: Evidence) -> Value: + """Successful attempts over attempts, stop-corrected when the attempt log allows, with a Wilson 95% interval.""" + attempts, corrected = _counted_attempts(ev) + successes = sum(attempt["success"] for attempt in attempts) + low, high = wilson_interval(successes, len(attempts)) + raw_rate = sum(attempt["success"] for attempt in ev.attempts) / len(ev.attempts) + details = {"interval_95": [low, high], "raw_rate": raw_rate, "stop_corrected": corrected} + return Value(successes / len(attempts), n=len(attempts), details=details) + + +@metric( + "demos_per_gpu_hour", + area="viability", + signal="generation performance", + scope="run", + unit="demos/GPU-h", + better="higher", + requires=["run.generation_time_s", "run.num_gpus", "env_args.autodata.exported_episodes"], +) +def demos_per_gpu_hour(ev: Evidence) -> Value: + """Exported demonstrations per GPU-hour of generation time.""" + if ev.run["num_gpus"] == 0: + raise NotApplicable("the run used no GPU") + gpu_hours = ev.run["generation_time_s"] / 3600 * ev.run["num_gpus"] + exported = ev.env_args["autodata"]["exported_episodes"] + details = {key: ev.run.get(key) for key in ("gpu", "cpu", "device", "num_envs")} + return Value(exported / gpu_hours, n=exported, details=details) + + +@metric( + "time_to_dataset", + area="viability", + signal="time to dataset", + scope="run", + unit="s", + better="lower", + requires=["run.setup_time_s", "run.generation_time_s"], +) +def time_to_dataset(ev: Evidence) -> Value: + """Wall time of the automated steps: annotation of the source, setup and generation.""" + annotation = ev.source.env_args.get("autodata", {}).get("annotation", {}) if ev.source else {} + parts = { + "annotation_s": annotation.get("time_s"), + "setup_s": ev.run["setup_time_s"], + "generation_s": ev.run["generation_time_s"], + } + details = {**parts, "note": "automated steps only; manual recording and annotation time is not measured"} + return Value(sum(seconds for seconds in parts.values() if seconds is not None), n=1, details=details) + + +@metric( + "annotation_pass_rate", + area="viability", + signal="time to dataset", + scope="run", + unit="fraction", + better="higher", + requires=["source.env_args.autodata.annotation"], +) +def annotation_pass_rate(ev: Evidence) -> Value: + """Share of processed recordings that annotation exported into the source dataset.""" + annotation = ev.source.env_args["autodata"]["annotation"] + if annotation["processed_episodes"] == 0: + raise NotApplicable("annotation processed no episodes") + rate = annotation["annotated_episodes"] / annotation["processed_episodes"] + return Value(rate, n=annotation["processed_episodes"], details={"input_episodes": annotation["input_episodes"]}) + + +@metric( + "planning_failure_rate", + area="diagnostics", + signal="failure causes", + scope="run", + unit="fraction of attempts", + better="lower", + requires=["attempts.extras.planning_failed"], +) +def planning_failure_rate(ev: Evidence) -> Value: + """Share of attempts whose motion planning failed, stop-corrected when the attempt log allows.""" + attempts, corrected = _counted_attempts(ev) + failed = sum(attempt["extras"]["planning_failed"] for attempt in attempts) + return Value(failed / len(attempts), n=len(attempts), details={"stop_corrected": corrected}) + + +@metric( + "planning_time", + area="diagnostics", + signal="generation performance", + scope="run", + unit="s per attempt", + better="lower", + primary="p95", + requires=["attempts.extras.planning_time_s"], +) +def planning_time(ev: Evidence) -> Value: + """p50 and p95 motion-planning time per attempt, and its share of generation time.""" + times = np.array([attempt["extras"]["planning_time_s"] for attempt in ev.attempts]) + p50, p95 = np.percentile(times, [50, 95]) + generation_time = ev.run.get("generation_time_s") + share = float(times.sum() / generation_time) if generation_time else None + return Value({"p50": float(p50), "p95": float(p95)}, n=len(times), details={"share_of_generation_time": share}) diff --git a/autodata_tests/e2e/test_mimicgen_data_generation.py b/autodata_tests/e2e/test_mimicgen_data_generation.py index 0552279..16b27f1 100644 --- a/autodata_tests/e2e/test_mimicgen_data_generation.py +++ b/autodata_tests/e2e/test_mimicgen_data_generation.py @@ -8,10 +8,12 @@ from autodata_tests.utils.constants import TestPaths from autodata_tests.utils.subprocess import run_subprocess -from autodata_tests.utils.utils import assert_recorded_run, assert_valid_dataset +from autodata_tests.utils.utils import assert_ok_metrics, assert_recorded_run, assert_valid_dataset HEADLESS = True GENERATION_NUM_TRIALS = 1 +# Metrics a MimicGen run reported with its result file must compute. +RUN_METRICS = {"generation_success_rate", "demos_per_gpu_hour", "time_to_dataset"} def _run_franka_cube_stack_mimicgen(num_envs: int, device: str) -> None: @@ -52,6 +54,7 @@ def _run_franka_cube_stack_mimicgen(num_envs: int, device: str) -> None: assert_valid_dataset(output_file, min_num_demos=GENERATION_NUM_TRIALS) assert_recorded_run(output_file, result_file) + assert_ok_metrics(output_file, result_file, RUN_METRICS) @pytest.mark.with_subprocess diff --git a/autodata_tests/e2e/test_skillgen_bin_data_generation.py b/autodata_tests/e2e/test_skillgen_bin_data_generation.py index 580d086..bf45200 100644 --- a/autodata_tests/e2e/test_skillgen_bin_data_generation.py +++ b/autodata_tests/e2e/test_skillgen_bin_data_generation.py @@ -17,10 +17,18 @@ from autodata_tests.utils.constants import TestPaths from autodata_tests.utils.subprocess import run_subprocess -from autodata_tests.utils.utils import assert_recorded_run, assert_valid_dataset +from autodata_tests.utils.utils import assert_ok_metrics, assert_recorded_run, assert_valid_dataset HEADLESS = True GENERATION_NUM_TRIALS = 1 +# Metrics a SkillGen run reported with its result file must compute; the planning metrics are SkillGen's own. +RUN_METRICS = { + "generation_success_rate", + "demos_per_gpu_hour", + "time_to_dataset", + "planning_time", + "planning_failure_rate", +} def _run_franka_bin_stack_skillgen(num_envs: int, device: str) -> None: @@ -66,6 +74,7 @@ def _run_franka_bin_stack_skillgen(num_envs: int, device: str) -> None: attempts = assert_recorded_run(output_file, result_file) # Every SkillGen attempt plans at least one transit, and the algorithm reports how long it took. assert all(attempt["extras"]["planning_time_s"] > 0 for attempt in attempts) + assert_ok_metrics(output_file, result_file, RUN_METRICS) with open(result_file, encoding="utf-8") as result_handle: result = json.load(result_handle) diff --git a/autodata_tests/metrics/__init__.py b/autodata_tests/metrics/__init__.py new file mode 100644 index 0000000..57c822a --- /dev/null +++ b/autodata_tests/metrics/__init__.py @@ -0,0 +1,2 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 diff --git a/autodata_tests/metrics/test_report.py b/autodata_tests/metrics/test_report.py new file mode 100644 index 0000000..d5a5b90 --- /dev/null +++ b/autodata_tests/metrics/test_report.py @@ -0,0 +1,139 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests of how the report runs metrics, refuses mismatched runs and gates values.""" + +import h5py +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from autodata_metrics import MetricSpec, Value, build_report, gate_violations + + +def _spec(name: str, compute, requires: tuple[str, ...] = ()) -> MetricSpec: + return MetricSpec( + name=name, + compute=compute, + area="viability", + signal="test", + scope="run", + unit="fraction", + better="higher", + requires=requires, + primary=None, + version=1, + ) + + +def _dataset(path: Path, num_episodes: int = 1, entry: dict | None = None) -> str: + with h5py.File(path, "w") as h5_file: + data = h5_file.create_group("data") + data.attrs["env_args"] = json.dumps({"env_name": "Test-v0", **({"autodata": entry} if entry else {})}) + for index in range(num_episodes): + data.create_group(f"demo_{index}").create_dataset("actions", data=[[0.0, 1.0]]) + return str(path) + + +def _log(tmp_path: Path, run_id: str, successes: list[bool]) -> None: + lines = [json.dumps({"run_id": run_id, "success": success}) for success in successes] + (tmp_path / f"result.{run_id}.attempts.jsonl").write_text("\n".join(lines) + "\n") + + +def _run(tmp_path: Path, run_id: str, source_sha256: str | None = None) -> str: + """A completed run: its result file and attempt log.""" + result_file = tmp_path / "result.json" + run = {"run_id": run_id, "attempts_file": f"result.{run_id}.attempts.jsonl"} + if source_sha256 is not None: + run["inputs"] = {"source": {"path": "source.hdf5", "sha256": source_sha256}} + result_file.write_text(json.dumps({"num_success": 1, "num_attempts": 1, "run": run})) + _log(tmp_path, run_id, [True]) + return str(result_file) + + +def test_missing_evidence_is_not_applicable_and_names_the_key(tmp_path): + report = build_report( + _dataset(tmp_path / "d.hdf5"), metrics={"m": _spec("m", lambda ev: Value(1.0, n=1), ("step_success",))} + ) + + assert report["metrics"]["m"]["status"] == "not_applicable" + assert report["metrics"]["m"]["reason"] == "missing step_success" + + +def test_failing_metric_is_an_error_and_the_rest_are_still_computed(tmp_path): + metrics = {"broken": _spec("broken", lambda ev: 1 / 0), "fine": _spec("fine", lambda ev: Value(0.5, n=2))} + + rows = build_report(_dataset(tmp_path / "d.hdf5"), metrics=metrics)["metrics"] + + assert rows["broken"]["status"] == "error" and "ZeroDivisionError" in rows["broken"]["reason"] + assert rows["fine"]["status"] == "ok" and rows["fine"]["value"] == 0.5 + + +def test_interrupted_rerun_is_reported_from_its_own_attempt_log(tmp_path): + result_file = _run(tmp_path, "run-a") + _log(tmp_path, "run-b", [True, False]) + dataset = _dataset(tmp_path / "d.hdf5", entry={"run_id": "run-b", "exported_episodes": 1}) + count = {"attempts": _spec("attempts", lambda ev: Value(len(ev.attempts), n=1), ("attempts",))} + + report = build_report(dataset, result_file=result_file, metrics=count) + + assert report["metrics"]["attempts"]["value"] == 2 + assert "belongs to another run" in report["warnings"][0] + + +def test_report_refuses_inputs_that_disagree_with_their_run(tmp_path): + entry = {"run_id": "run-a", "exported_episodes": 1} + with pytest.raises(ValueError, match="merged or edited"): + build_report(_dataset(tmp_path / "merged.hdf5", num_episodes=2, entry=entry), metrics={}) + + dataset = _dataset(tmp_path / "d.hdf5", entry=entry) + result_file = _run(tmp_path, "run-a") + _log(tmp_path, "run-a", [True, True]) + with pytest.raises(ValueError, match="attempt log holds 2 attempts"): + build_report(dataset, result_file=result_file, metrics={}) + + result_file = _run(tmp_path, "run-a", source_sha256="0" * 64) + with pytest.raises(ValueError, match="not the one the run started from"): + build_report(dataset, result_file=result_file, source=_dataset(tmp_path / "source.hdf5"), metrics={}) + + +def test_gate_flags_violations_and_refuses_a_non_finite_bound(tmp_path): + row = {"status": "ok", "value": 0.4, "primary": None, "version": 2} + missing = {"status": "ok", "value": {"p50": 1.0}, "primary": "p95", "version": 1} + report = {"metrics": {"low": row, "old": row, "fine": row, "absent": {"status": "not_applicable"}, "gap": missing}} + gate = tmp_path / "gate.yaml" + gate.write_text( + "low: {min: 0.5}\nold: {version: 1}\nfine: {min: 0.3, max: 0.5, version: 2}\nabsent: {min: 0}\ngap: {max: 2}\n" + ) + + violations = gate_violations(report, str(gate)) + + assert [violation.split(":")[0] for violation in violations] == ["low", "old", "absent", "gap"] + for bad_bound in (".nan", "false", "{p95: 1}"): + gate.write_text(f"absent: {{min: {bad_bound}}}\n") + with pytest.raises(ValueError, match="not a finite number"): + gate_violations(report, str(gate)) + + +def test_non_finite_numbers_never_reach_the_report(tmp_path): + metrics = { + "m": _spec("m", lambda ev: Value(float("nan"), n=1)), + "d": _spec("d", lambda ev: Value(1.0, n=1, details={"interval_95": [0.5, float("inf")]})), + } + report = build_report(_dataset(tmp_path / "d.hdf5"), metrics=metrics) + gate = tmp_path / "gate.yaml" + gate.write_text("m: {min: 0}\n") + + assert report["metrics"]["m"]["status"] == "error" + assert gate_violations(report, str(gate)) == ["m: error"] + assert report["metrics"]["d"]["details"] == {"interval_95": [0.5, None]} + + +def test_metrics_import_without_torch_or_isaaclab(): + code = "import sys, autodata_metrics; print(sorted({'torch', 'isaaclab', 'autodata_core'} & set(sys.modules)))" + output = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, check=True).stdout + + assert output.strip() == "[]" diff --git a/autodata_tests/metrics/test_viability.py b/autodata_tests/metrics/test_viability.py new file mode 100644 index 0000000..ecaa78e --- /dev/null +++ b/autodata_tests/metrics/test_viability.py @@ -0,0 +1,28 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Known-answer tests of the viability statistics.""" + +import pytest + +from autodata_metrics.viability import attempts_before_stop, wilson_interval + + +def test_wilson_interval_matches_known_values(): + assert wilson_interval(0, 10) == pytest.approx((0.0, 0.2775), abs=1e-4) + assert wilson_interval(10, 10) == pytest.approx((0.7225, 1.0), abs=1e-4) + + +def _attempt(env_id: int, success: bool, start: float, duration: float) -> dict: + return {"env_id": env_id, "success": success, "start_time_s": start, "end_time_s": start + duration} + + +def test_attempts_before_stop_keeps_attempts_started_before_the_first_env_finished(): + long_successes = [_attempt(0, True, 4.0 * k, 4.0) for k in range(3)] # env 0 last finishes at 12 s + short_failures = [_attempt(1, False, 1.6 * k, 1.6) for k in range(10)] # one runs from 11.2 to 12.8 s + + kept = attempts_before_stop(long_successes + short_failures, num_envs=2) + + assert len(kept) == 11 and sum(attempt["success"] for attempt in kept) == 3 + assert attempts_before_stop(long_successes[:2] + short_failures, num_envs=2) is None + assert attempts_before_stop(short_failures, num_envs=2) is None diff --git a/autodata_tests/utils/test_validate_dataset.py b/autodata_tests/utils/test_validate_dataset.py index 3e1ea73..d109227 100644 --- a/autodata_tests/utils/test_validate_dataset.py +++ b/autodata_tests/utils/test_validate_dataset.py @@ -7,6 +7,8 @@ import json from pathlib import Path +import pytest + from scripts import validate_dataset @@ -45,3 +47,10 @@ def test_main_returns_nonzero_for_missing_file(tmp_path: Path) -> None: """An unreadable input path should produce a failing exit status.""" assert validate_dataset.main([str(tmp_path / "missing.hdf5")]) == 1 + + +def test_report_options_without_report_are_an_error(tmp_path: Path) -> None: + """A gate given without --report would otherwise be silently ignored.""" + + with pytest.raises(SystemExit): + validate_dataset.main([str(tmp_path / "data.hdf5"), "--gate", str(tmp_path / "gate.yaml")]) diff --git a/autodata_tests/utils/utils.py b/autodata_tests/utils/utils.py index 4792836..14a47b6 100644 --- a/autodata_tests/utils/utils.py +++ b/autodata_tests/utils/utils.py @@ -7,6 +7,9 @@ import json import os +from autodata_tests.utils.constants import TestPaths +from autodata_tests.utils.subprocess import run_subprocess + def assert_valid_dataset(output_file: str, min_num_demos: int) -> None: """Assert the run wrote a dataset containing at least min_num_demos entries.""" @@ -87,6 +90,30 @@ def assert_recorded_run(dataset_file: str, result_file: str) -> list[dict]: return attempts +def assert_ok_metrics(dataset_file: str, result_file: str, expected_ok: set[str]) -> None: + """Report a generation run's metrics with ``validate_dataset.py`` and assert exactly ``expected_ok`` computed a value. + + A renamed recorded key or an adapter that stops recording a stream turns a metric not applicable, which + this catches. + """ + + report_file = os.path.join(os.path.dirname(dataset_file), "report.json") + run_subprocess([ + TestPaths.python_path, + os.path.join(TestPaths.scripts_dir, "validate_dataset.py"), + dataset_file, + "--report", + "--result_file", + result_file, + "--report_file", + report_file, + ]) + with open(report_file, encoding="utf-8") as report_handle: + rows = json.load(report_handle)["metrics"] + ok = {name for name, row in rows.items() if row["status"] == "ok"} + assert ok == expected_ok, f"Computed {sorted(ok)}, expected {sorted(expected_ok)}: {rows}" + + def assert_no_metric_inputs(dataset_file: str) -> None: """Assert a dataset recorded without ``--record_metric_inputs`` has no metric inputs, as before they existed.""" diff --git a/docs/_ext/autodata_doc_tools.py b/docs/_ext/autodata_doc_tools.py index feb9f27..72db1fd 100644 --- a/docs/_ext/autodata_doc_tools.py +++ b/docs/_ext/autodata_doc_tools.py @@ -8,7 +8,9 @@ in one place and can switch between internal/external targets on release. """ +import ast import re +from pathlib import Path from typing import Any from sphinx.application import Sphinx @@ -69,8 +71,41 @@ def curobo_replacer(_: Any) -> str: source[0] = re.sub(r":docker_run_curobo:", curobo_replacer, source[0]) +def autodata_metrics_table(app: Sphinx, _: Any, source: list[str]) -> None: + """Replaces ``:autodata_metrics_table:`` with a table of the built-in metrics. + + The rows are read from the ``@metric`` declarations in ``autodata_metrics/`` instead of importing them, so the + docs build needs none of the metrics' dependencies and the table cannot drift from the code. + """ + if ":autodata_metrics_table:" not in source[0]: + return + rows = [] + for path in sorted((Path(app.confdir).parent / "autodata_metrics").glob("*.py")): + for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))): + if not isinstance(node, ast.FunctionDef): + continue + for decorator in node.decorator_list: + if isinstance(decorator, ast.Call) and getattr(decorator.func, "id", None) == "metric": + options = {keyword.arg: ast.literal_eval(keyword.value) for keyword in decorator.keywords} + requires = ", ".join(f"``{key}``" for key in options.get("requires", ())) or "-" + definition = (ast.get_docstring(node) or "").split("\n")[0] + name = ast.literal_eval(decorator.args[0]) + rows.append((options["area"], name, options["unit"], requires, definition)) + table = [ + ".. list-table::", + " :header-rows: 1", + " :widths: 20 12 14 26 28", + "", + " * - Metric\n - Area\n - Unit\n - Requires\n - Definition", + ] + for area, name, unit, requires, definition in sorted(rows): + table.append(f" * - ``{name}``\n - {area}\n - {unit}\n - {requires}\n - {definition}") + source[0] = source[0].replace(":autodata_metrics_table:", "\n".join(table) + "\n") + + def setup(app: Sphinx) -> None: app.connect("source-read", autodata_git_clone_code_block) app.connect("source-read", autodata_code_link) app.connect("source-read", docker_run_command_replacer) + app.connect("source-read", autodata_metrics_table) app.add_config_value("autodata_docs_config", {}, "env") diff --git a/docs/index.rst b/docs/index.rst index 6d4c9bf..853caf5 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -146,5 +146,6 @@ Table of Contents :maxdepth: 1 :caption: References + pages/references/metrics pages/references/troubleshooting pages/references/release_notes diff --git a/docs/pages/references/metrics.rst b/docs/pages/references/metrics.rst new file mode 100644 index 0000000..de230ad --- /dev/null +++ b/docs/pages/references/metrics.rst @@ -0,0 +1,87 @@ +Metrics +======= + +``scripts/validate_dataset.py --report`` computes the metrics of a dataset and, given its result +file, of the generation run that produced it. Metrics are computed from recorded files and need no +simulator. Generate, annotate or retarget with ``--record_metric_inputs`` to record what they read; +without it, most metrics report that their inputs are missing. + +.. code-block:: bash + + python scripts/validate_dataset.py datasets/generated.hdf5 --report \ + --result_file datasets/generated.json \ + --source datasets/annotated_datasets/dataset_franka_annotated.hdf5 \ + --report_file datasets/generated_report.json + +Each metric reports one of three things: + +- a value, with its unit and sample count; +- ``not_applicable``, with the reason, such as a missing recorded input; +- ``error``, when it fails. The other metrics are still computed. + +The report reads the run named by the dataset's run id. A result file from another run, such as the +earlier result of an interrupted rerun, is set aside with a warning, and the run is reported from its own +attempt log. The report refuses a dataset, result file and attempt log whose episode, attempt or success +counts disagree, as they do after datasets are merged. When the result file records the run's source, the +report also refuses a different ``--source``; without that record, the source is not verified. + +Gating +------ + +``--gate`` reads metric bounds from YAML and makes the command fail when one is violated: + +.. code-block:: yaml + + generation_success_rate: {min: 0.3, version: 1} + planning_time: {max: 5.0, key: p95} + +A gated metric also violates the gate if it is not applicable, fails, or reports a different +``version``. A bound that is not a finite number is an error. + +Built-in metrics +---------------- + +:autodata_metrics_table: + +Adding a metric +--------------- + +A metric is a function registered with ``autodata_metrics.metric``. It names the evidence it +requires. Without that evidence, the report says which key is missing and does not call the function. + +.. code-block:: python + + from autodata_metrics import Value, metric + + + @metric( + "slowest_plan", + area="diagnostics", + signal="generation performance", + scope="run", + unit="s", + better="lower", + requires=["attempts.extras.planning_time_s"], + ) + def slowest_plan(ev): + """Longest motion-planning time of any attempt.""" + times = [attempt["extras"]["planning_time_s"] for attempt in ev.attempts] + return Value(max(times), n=len(times)) + +An evidence key is one of: + +- an HDF5 path present in every episode, such as ``step_success``; +- ``env_args.`` or ``run.``; +- ``attempts`` or ``attempts.``; +- ``source`` or ``source.``. + +Built-in metrics live in ``autodata_metrics/``. Load your own with +``--metrics my_package.my_metrics``. Raise a metric's ``version`` whenever its definition changes. + +When you add an algorithm or an embodiment: + +- An algorithm reports per-attempt facts by overriding ``attempt_extras()``. Metrics that require + those facts then apply to its runs. +- An embodiment's adapter describes its pose actions through ``metric_context()``. +- List the workflow's expected metrics in its end-to-end test, so a metric that stops applying fails + the test. diff --git a/scripts/validate_dataset.py b/scripts/validate_dataset.py index a53db2d..8b9bb43 100644 --- a/scripts/validate_dataset.py +++ b/scripts/validate_dataset.py @@ -9,18 +9,22 @@ python scripts/validate_dataset.py [ ...] python scripts/validate_dataset.py directory/*.hdf5 + python scripts/validate_dataset.py --report --result_file [--gate ] -The command exits nonzero if any supplied file is invalid. +The command exits nonzero if any supplied file is invalid or a gated metric violates its bound. """ from __future__ import annotations import argparse import h5py +import importlib import json import os import sys +from autodata_metrics import build_report, format_report, gate_violations + REQUIRED_DEMO_FIELDS = { "actions": h5py.Dataset, "initial_state": h5py.Group, @@ -163,7 +167,20 @@ def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Validate AutoData demonstration HDF5 datasets.") parser.add_argument("files", nargs="+", help="HDF5 dataset file(s) to validate.") + parser.add_argument("--report", action="store_true", help="Also report the metrics of a single dataset.") + parser.add_argument("--result_file", help="Generation result (generate_dataset.py --result_file) of the dataset.") + parser.add_argument("--source", help="Source dataset the generation started from.") + parser.add_argument( + "--metrics", action="append", default=[], help="Module registering more metrics; may be repeated." + ) + parser.add_argument("--gate", help="YAML of metric bounds; a violated bound makes the command fail.") + parser.add_argument("--report_file", help="Also write the report as JSON to this file.") args = parser.parse_args(argv) + if args.report and len(args.files) != 1: + parser.error("--report takes exactly one dataset") + given = [f"--{name}" for name in ("result_file", "source", "metrics", "gate", "report_file") if getattr(args, name)] + if given and not args.report: + parser.error(f"{', '.join(given)} only apply with --report") results = [validate_file(path) for path in args.files] print_results(results) @@ -173,7 +190,28 @@ def main(argv: list[str] | None = None) -> int: print() print(f"All {total} file(s) valid." if num_invalid == 0 else f"{num_invalid} of {total} file(s) invalid.") - return 1 if num_invalid else 0 + # A run whose every attempt failed leaves an empty dataset; its run metrics are still worth reporting. + violations = [] + if args.report and results[0].num_episodes is not None: + for module in args.metrics: + importlib.import_module(module) + report = build_report(args.files[0], result_file=args.result_file, source=args.source) + print() + print(format_report(report)) + if args.report_file: + with open(args.report_file, "w", encoding="utf-8") as report_handle: + json.dump(report, report_handle, indent=2, default=_json_value) + report_handle.write("\n") + if args.gate: + violations = gate_violations(report, args.gate) + print("\nGate passed." if not violations else "\nGate failed:\n " + "\n ".join(violations)) + + return 1 if num_invalid or violations else 0 + + +def _json_value(value: object) -> object: + """JSON form of the numpy scalars and arrays that metric values may hold.""" + return value.tolist() if hasattr(value, "tolist") else str(value) if __name__ == "__main__": diff --git a/setup.py b/setup.py index 4c412ae..35b3a9f 100644 --- a/setup.py +++ b/setup.py @@ -26,6 +26,7 @@ include=[ "autodata_core*", "autodata_interfaces*", + "autodata_metrics*", "autodata_utils*", "autodata_examples*", ]