diff --git a/autodata_metrics/__init__.py b/autodata_metrics/__init__.py index c9d8546..8897644 100644 --- a/autodata_metrics/__init__.py +++ b/autodata_metrics/__init__.py @@ -7,7 +7,17 @@ Importing this package registers the built-in metrics. """ -from autodata_metrics import viability +from autodata_metrics import validity, 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 +from autodata_metrics.registry import ( + AREAS, + REGISTERED_CHECKS, + REGISTERED_METRICS, + CheckSpec, + MetricSpec, + NotApplicable, + Value, + check, + metric, +) +from autodata_metrics.report import build_report, format_report, gate_results, load_gate diff --git a/autodata_metrics/evidence.py b/autodata_metrics/evidence.py index 5f5a8b2..3180c95 100644 --- a/autodata_metrics/evidence.py +++ b/autodata_metrics/evidence.py @@ -9,8 +9,11 @@ import json import numpy as np from collections.abc import Mapping +from functools import cached_property from typing import Any +from autodata_metrics.records import InvalidRecord, parse_env_args + def _has_path(record: Any, dotted_key: str) -> bool: """Whether ``record[a][b]`` exists for ``dotted_key`` ``"a.b"``.""" @@ -22,10 +25,18 @@ def _has_path(record: Any, dotted_key: str) -> bool: 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.""" + """The dataset's ``env_args``, with its ``autodata`` entry parsed; empty when it records none. + + Raises: + InvalidRecord: If ``env_args`` is not JSON, or it or its entry lacks the shape the metrics read. + """ + where = f"{h5_file.filename} env_args" 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 {} + try: + env_args = json.loads(raw_env_args.decode() if isinstance(raw_env_args, bytes) else raw_env_args) + except (TypeError, ValueError, RecursionError) as exc: + raise InvalidRecord(f"{where} must be JSON: {exc}") from exc + return parse_env_args(env_args, where) class Episode: @@ -41,12 +52,27 @@ def __contains__(self, key: str) -> bool: def __getitem__(self, key: str) -> np.ndarray: return self._group[key][()] + def children(self, key: str) -> list[str]: + """Names under the group ``key``, such as the end effectors under ``obs/datagen_info/eef_pose``.""" + return list(self._group[key]) + + def datasets(self) -> list[tuple[str, h5py.Dataset]]: + """Every dataset of the episode with its path, unread until accessed.""" + found: list[tuple[str, h5py.Dataset]] = [] + self._group.visititems( + lambda path, item: found.append((path, item)) if isinstance(item, h5py.Dataset) else None + ) + return found + class Evidence: """A dataset, and optionally the generation run that produced it, as metrics read them. Attributes: - episodes: The dataset's episodes. + all_episodes: The dataset's episodes. + episodes: The episodes that passed every episode check; dataset metrics read these. + failures: Per episode check that applied to some episode, the failing episodes' names and reasons. + check_groups: Per episode check considered, its group; a check absent from ``failures`` did not apply. 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. @@ -62,7 +88,10 @@ def __init__( 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.all_episodes = [Episode(group) for group in h5_file["data"].values()] + self.episodes = list(self.all_episodes) + self.failures: dict[str, dict[str, str]] = {} + self.check_groups: dict[str, str] = {} self.result = result self.run: Mapping[str, Any] = (result or {}).get("run", {}) self.attempts = attempts @@ -88,3 +117,13 @@ def has(self, key: str) -> bool: if head == "run": return _has_path(self.run, rest) return bool(self.episodes) and all(key in episode for episode in self.episodes) + + @cached_property + def camera_paths(self) -> frozenset[str]: + """Paths of the camera streams, 4-D ``uint8`` arrays, that any episode records.""" + return frozenset( + path + for episode in self.all_episodes + for path, dataset in episode.datasets() + if dataset.dtype == np.uint8 and dataset.ndim == 4 + ) diff --git a/autodata_metrics/records.py b/autodata_metrics/records.py new file mode 100644 index 0000000..82be6b7 --- /dev/null +++ b/autodata_metrics/records.py @@ -0,0 +1,156 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Shapes of the records a report reads, checked once when a record is read. + +A record that does not match its shape is refused with an error naming the file and the field, so no metric reads a +malformed value. Fields a shape does not list are left unchecked, so producers can add fields that older readers +ignore. +""" + +from __future__ import annotations + +import numbers +import sys +from dataclasses import dataclass +from typing import Any + +from autodata_utils.dataset_metadata import AUTODATA_ENTRY_VERSION + + +class InvalidRecord(ValueError): + """A record without the shape the report reads.""" + + +@dataclass(frozen=True) +class Maybe: + """A field that may be absent or null.""" + + shape: Any + + +@dataclass(frozen=True) +class ListOf: + """A list of values of one shape, of a fixed length when ``length`` is given.""" + + shape: Any + length: int | None = None + + +@dataclass(frozen=True) +class MapOf: + """An object whose values all have one shape.""" + + shape: Any + + +NUMBER = "a finite number" +COUNT = "a non-negative integer" +DURATION = "a non-negative finite number" + +INPUT = {"path": str, "sha256": str} +RUN = { + "run_id": str, + "seed": Maybe(int), + "revision": Maybe(str), + "inputs": Maybe(MapOf(Maybe(INPUT))), + "device": Maybe(str), + "gpu": Maybe(str), + "cpu": Maybe(str), + "num_gpus": Maybe(COUNT), + "num_envs": Maybe(COUNT), + "setup_time_s": Maybe(DURATION), + "generation_time_s": Maybe(DURATION), + "attempts_file": Maybe(str), +} +RESULT = {"num_success": COUNT, "num_failures": Maybe(COUNT), "num_attempts": COUNT, "run": Maybe(RUN)} +ATTEMPT = { + "run_id": str, + "env_id": COUNT, + "success": bool, + "start_time_s": NUMBER, + "end_time_s": NUMBER, + "extras": Maybe({"planning_time_s": Maybe(DURATION), "planning_failed": Maybe(bool)}), +} +ANNOTATION = { + "mode": Maybe(("auto", "manual")), + "input_episodes": COUNT, + "processed_episodes": COUNT, + "annotated_episodes": COUNT, + "setup_time_s": Maybe(DURATION), + "time_s": DURATION, +} +ENTRY = { + "version": COUNT, + "run_id": str, + "exported_episodes": COUNT, + "pose_action": Maybe(MapOf({"action_slice": ListOf(int, 2), "bound": Maybe(NUMBER)})), + "annotation": Maybe(ANNOTATION), +} +ENV_ARGS = {"autodata": Maybe(ENTRY)} + + +def parse(value: Any, shape: Any, where: str) -> Any: + """``value`` checked against ``shape``, with its absent or null optional fields left out. + + A shape is a type, a tuple of allowed values, :data:`NUMBER`, :data:`COUNT`, :data:`DURATION`, a + :class:`Maybe`, :class:`ListOf` or :class:`MapOf`, or a dict of field shapes; a dict's fields are required + unless their shape is a :class:`Maybe`. + + Raises: + InvalidRecord: Naming ``where`` and the field, if ``value`` does not have ``shape``. + """ + if isinstance(shape, Maybe): + return None if value is None else parse(value, shape.shape, where) + if isinstance(shape, dict): + _require(isinstance(value, dict), value, "an object", where) + parsed = dict(value) + for name, field_shape in shape.items(): + _require(name in value or isinstance(field_shape, Maybe), value, f"an object with {name!r}", where) + field = parse(value.get(name), field_shape, f"{where}.{name}") + if field is None: + parsed.pop(name, None) + else: + parsed[name] = field + return parsed + if isinstance(shape, ListOf): + _require(isinstance(value, list) and shape.length in (None, len(value)), value, "a list", where) + return [parse(item, shape.shape, f"{where}[{index}]") for index, item in enumerate(value)] + if isinstance(shape, MapOf): + _require(isinstance(value, dict), value, "an object", where) + parsed = {key: parse(item, shape.shape, f"{where}.{key}") for key, item in value.items()} + return {key: item for key, item in parsed.items() if item is not None} + if isinstance(shape, tuple): + _require(value in shape, value, f"one of {list(shape)}", where) + elif shape in (NUMBER, DURATION): + _require(is_finite_number(value) and (shape is NUMBER or value >= 0), value, shape, where) + elif shape is COUNT: + _require(isinstance(value, int) and not isinstance(value, bool) and value >= 0, value, shape, where) + else: + exact = isinstance(value, shape) and not (shape is int and isinstance(value, bool)) + _require(exact, value, shape.__name__, where) + return value + + +def parse_env_args(env_args: Any, where: str) -> dict[str, Any]: + """A dataset's ``env_args`` with its ``autodata`` entry parsed, refusing an entry from a newer AutoData.""" + entry = env_args.get("autodata") if isinstance(env_args, dict) else None + version = entry.get("version") if isinstance(entry, dict) else None + if isinstance(version, int) and version > AUTODATA_ENTRY_VERSION: + raise InvalidRecord(f"{where}.autodata was written by a newer AutoData (entry version {version}).") + parsed = parse(env_args, ENV_ARGS, where) + annotation = parsed.get("autodata", {}).get("annotation") + if annotation is not None: + counts = [annotation[name] for name in ("annotated_episodes", "processed_episodes", "input_episodes")] + _require(counts == sorted(counts), counts, "annotated <= processed <= input", f"{where}.autodata.annotation") + return parsed + + +def is_finite_number(value: Any) -> bool: + """Whether ``value`` is a real number other than a bool, of a magnitude a float can hold.""" + return isinstance(value, numbers.Real) and not isinstance(value, bool) and abs(value) <= sys.float_info.max + + +def _require(condition: bool, value: Any, expected: str, where: str) -> None: + if not condition: + raise InvalidRecord(f"{where} must be {expected}, not {value!r}.") diff --git a/autodata_metrics/registry.py b/autodata_metrics/registry.py index c043c6f..b076cb5 100644 --- a/autodata_metrics/registry.py +++ b/autodata_metrics/registry.py @@ -10,12 +10,15 @@ from __future__ import annotations +import numbers from collections.abc import Callable, Mapping from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, Literal +from autodata_metrics.records import is_finite_number + if TYPE_CHECKING: - from autodata_metrics.evidence import Evidence + from autodata_metrics.evidence import Episode, Evidence AREAS = ("validity", "diversity", "viability", "diagnostics") @@ -38,6 +41,15 @@ class Value: n: int details: Mapping[str, Any] = field(default_factory=dict) + def __post_init__(self) -> None: + values = self.value.values() if isinstance(self.value, Mapping) else [self.value] + if not all(is_finite_number(item) for item in values): + raise ValueError(f"value {self.value!r} is not a finite number or a mapping of finite numbers") + if not isinstance(self.n, numbers.Integral) or self.n < 0: + raise ValueError(f"sample count {self.n!r} is not a non-negative integer") + if not isinstance(self.details, Mapping): + raise ValueError(f"details {self.details!r} are not a mapping") + @dataclass(frozen=True) class MetricSpec: @@ -47,15 +59,29 @@ class MetricSpec: compute: Callable[[Evidence], Value] area: str signal: str - scope: Literal["dataset", "run"] + scope: Literal["episode", "dataset", "run"] unit: str + sample: str better: Literal["higher", "lower"] | None requires: tuple[str, ...] primary: str | None version: int + baseline: bool = False + + +@dataclass(frozen=True) +class CheckSpec: + """A registered episode check.""" + + name: str + check: Callable[[Episode, Evidence], str | None] + group: str + requires: tuple[str, ...] REGISTERED_METRICS: dict[str, MetricSpec] = {} +REGISTERED_CHECKS: dict[str, CheckSpec] = {} +CHECK_GROUPS = ("correctness", "image") def metric( @@ -63,12 +89,14 @@ def metric( *, area: str, signal: str, - scope: Literal["dataset", "run"], + scope: Literal["episode", "dataset", "run"], unit: str, + sample: str, better: Literal["higher", "lower"] | None = None, requires: tuple[str, ...] | list[str] = (), primary: str | None = None, version: int = 1, + baseline: bool = False, ) -> Callable[[Callable[[Evidence], Value]], Callable[[Evidence], Value]]: """Register the decorated function as the metric ``name``. @@ -76,12 +104,15 @@ def metric( 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. + scope: ``"episode"`` metrics summarize the episode checks over every episode; ``"dataset"`` metrics read + the episodes that passed them; ``"run"`` metrics read attempts and the run record. unit: Unit of the value. + sample: What the value's sample count counts, in the plural, e.g. ``"attempts"`` or ``"episodes"``. 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. + baseline: Also compute the metric on the source dataset. """ assert area in AREAS, f"Metric {name!r} has area {area!r}; expected one of {AREAS}." @@ -94,11 +125,36 @@ def register(compute: Callable[[Evidence], Value]) -> Callable[[Evidence], Value signal=signal, scope=scope, unit=unit, + sample=sample, better=better, requires=tuple(requires), primary=primary, version=version, + baseline=baseline, ) return compute return register + + +def check( + name: str, *, group: str = "correctness", requires: tuple[str, ...] | list[str] = () +) -> Callable[[Callable[[Episode, Evidence], str | None]], Callable[[Episode, Evidence], str | None]]: + """Register the decorated function as the episode check ``name``. + + A check returns None for an episode that passes, a reason for one that fails, or raises :class:`NotApplicable` + for an episode it does not cover. Episodes that fail any check are left out of every dataset metric. + + Args: + name: Unique check name. + group: One of :data:`CHECK_GROUPS`; the pass-rate metric the check counts toward. + requires: Evidence keys the check needs; without them it does not run. + """ + assert group in CHECK_GROUPS, f"Check {name!r} has group {group!r}; expected one of {CHECK_GROUPS}." + + def register(function: Callable[[Episode, Evidence], str | None]) -> Callable[[Episode, Evidence], str | None]: + assert name not in REGISTERED_CHECKS, f"Check {name!r} is already registered." + REGISTERED_CHECKS[name] = CheckSpec(name=name, check=function, group=group, requires=tuple(requires)) + return function + + return register diff --git a/autodata_metrics/report.py b/autodata_metrics/report.py index f8724aa..8e643d6 100644 --- a/autodata_metrics/report.py +++ b/autodata_metrics/report.py @@ -9,14 +9,24 @@ import h5py import json import math -import numbers +import re +import textwrap import yaml -from collections.abc import Mapping +from collections.abc import Hashable, 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_metrics.records import ATTEMPT, RESULT, InvalidRecord, is_finite_number, parse +from autodata_metrics.registry import ( + AREAS, + REGISTERED_CHECKS, + REGISTERED_METRICS, + CheckSpec, + MetricSpec, + NotApplicable, + Value, +) from autodata_utils.generation_result import attempts_file_for, code_revision, input_record SCHEMA_VERSION = 1 @@ -31,8 +41,9 @@ def read_run( 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 = [] + text = _read_text(result_path) + result = None if text is None else parse(_json(text, result_path), RESULT, result_file) + warnings = [] if text is not None else [f"{result_file} does not exist"] 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 @@ -42,13 +53,34 @@ def read_run( attempts_path = Path(attempts_file_for(result_file, run_id)) else: return result, None, warnings + text = _read_text(attempts_path) 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()] + if text is not None: + attempts = [ + parse(_json(line, attempts_path), ATTEMPT, f"{attempts_path} line {number}") + for number, line in enumerate(text.splitlines(), start=1) + if line.strip() + ] return result, attempts, warnings +def _read_text(path: Path) -> str | None: + """The file's text, or None when there is no such file.""" + try: + return path.read_text(encoding="utf-8") + except (FileNotFoundError, IsADirectoryError, NotADirectoryError): + return None + except (OSError, UnicodeDecodeError) as exc: + raise InvalidRecord(f"{path} cannot be read: {exc}") from exc + + +def _json(text: str, path: Path) -> Any: + try: + return json.loads(text) + except (ValueError, RecursionError) as exc: + raise InvalidRecord(f"{path} must be JSON: {exc}") from exc + + def check_same_run(evidence: Evidence) -> list[str]: """Refuse a dataset and run record that come from different runs; return warnings for unverifiable data. @@ -58,7 +90,7 @@ def check_same_run(evidence: Evidence) -> list[str]: Raises: ValueError: If the dataset, result and attempt log disagree on their run or their counts. """ - num_episodes = len(evidence.episodes) + num_episodes = len(evidence.all_episodes) result = evidence.result if result is not None and num_episodes != result["num_success"]: raise ValueError( @@ -72,8 +104,12 @@ def check_same_run(evidence: Evidence) -> list[str]: f" {result['num_attempts']} and {result['num_success']}." ) entry = evidence.env_args.get("autodata") + if entry is None and result is not None: + return [ + "the dataset was recorded without --record_metric_inputs, so it is matched to its run by episode count only" + ] if entry is None: - return ["the dataset has no autodata entry, so it is matched to its run by episode count only"] + return [] if entry["exported_episodes"] != num_episodes: raise ValueError( f"The dataset holds {num_episodes} episodes but its run exported {entry['exported_episodes']};" @@ -87,6 +123,30 @@ def check_same_run(evidence: Evidence) -> list[str]: return [] +def run_checks(evidence: Evidence, checks: Mapping[str, CheckSpec]) -> None: + """Run the episode checks and keep only the episodes that pass all of them in ``evidence.episodes``.""" + for spec in checks.values(): + evidence.check_groups[spec.name] = spec.group + if any(not evidence.has(key) for key in spec.requires): + continue + failed: dict[str, str] = {} + applied = False + for episode in evidence.all_episodes: + try: + reason = spec.check(episode, evidence) + except NotApplicable: + continue + except Exception as exc: # A broken check fails its episodes visibly instead of sinking the report. + reason = f"{type(exc).__name__}: {exc}" + applied = True + if reason is not None: + failed[episode.name] = reason + if applied: + evidence.failures[spec.name] = failed + failing = {name for failed in evidence.failures.values() for name in failed} + evidence.episodes = [episode for episode in evidence.all_episodes if episode.name not in failing] + + 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 = { @@ -94,36 +154,43 @@ def evaluate(spec: MetricSpec, evidence: Evidence) -> dict[str, Any]: "signal": spec.signal, "scope": spec.scope, "unit": spec.unit, + "sample": spec.sample, "better": spec.better, "primary": spec.primary, "version": spec.version, } + if spec.scope == "dataset" and not evidence.episodes: + return {**row, "status": "not_applicable", "reason": "no valid episodes"} 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}"} + head = missing.partition(".")[0] + missing = head if head in ("attempts", "source") and not evidence.has(head) else missing + return {**row, "status": "not_applicable", "reason": f"missing {missing}", "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) + if not isinstance(value, Value): + return {**row, "status": "error", "reason": f"returned {type(value).__name__}, not a Value"} + row = { + **row, + "status": "ok", + "value": _finite_or_none(value.value), + "n": int(value.n), + "details": _finite_or_none(value.details), + } + if spec.baseline and evidence.source is not None: + row["baseline"] = evaluate(spec, evidence.source).get("value") + return row 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()} + return {str(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 @@ -134,6 +201,7 @@ def build_report( result_file: str | None = None, source: str | None = None, metrics: Mapping[str, MetricSpec] | None = None, + checks: Mapping[str, CheckSpec] | None = None, ) -> dict[str, Any]: """Compute every metric for a dataset and, when ``result_file`` is given, its generation run. @@ -142,6 +210,7 @@ def build_report( 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. + checks: Episode checks to run first; defaults to every registered check. Raises: ValueError: If the dataset, result, attempt log and source do not all come from one run. @@ -151,15 +220,21 @@ def build_report( 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 + if source_evidence is not None: + warnings += _set_aside_merged_annotation(source_evidence, source) 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']}.") + for checked in (evidence, source_evidence): + if checked is not None: + run_checks(checked, REGISTERED_CHECKS if checks is None else checks) 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) + checked = _check_summary(evidence) + source_checked = _check_summary(source_evidence) if source_evidence is not None else None return { "schema_version": SCHEMA_VERSION, "inputs": { @@ -169,70 +244,262 @@ def build_report( "run_id": run_id, "revision": code_revision(), }, - "num_episodes": num_episodes, + **checked, + "source_checks": source_checked, "warnings": warnings, "metrics": rows, } -def gate_violations(report: Mapping[str, Any], gate_file: str) -> list[str]: - """Bounds in ``gate_file`` that the report violates. +def _set_aside_merged_annotation(source: Evidence, path: str) -> list[str]: + """Drop the annotation record of a source merged or edited after annotation, since it counts other episodes.""" + entry = source.env_args.get("autodata") + if entry is None or "annotation" not in entry or entry["exported_episodes"] == len(source.all_episodes): + return [] + source.env_args = {**source.env_args, "autodata": {key: item for key, item in entry.items() if key != "annotation"}} + return [ + f"{path} holds {len(source.all_episodes)} episodes but was annotated with {entry['exported_episodes']}, so its" + " annotation record is set aside" + ] + + +def _check_summary(evidence: Evidence) -> dict[str, Any]: + return { + "num_episodes": len(evidence.all_episodes), + "num_valid_episodes": len(evidence.episodes), + "check_failures": { + name: {"failed": len(failed), "examples": dict(list(failed.items())[:3])} + for name, failed in evidence.failures.items() + }, + } + + +_GATE_FIELDS = {"min", "max", "key", "version"} + + +class _GateLoader(yaml.SafeLoader): + """Safe YAML loading that refuses a repeated key, which plain YAML silently resolves to its last value.""" + + def construct_mapping(self, node: yaml.MappingNode, deep: bool = False) -> dict[Any, Any]: + keys = [self.construct_object(key_node, deep=deep) for key_node, _ in node.value] + repeated = sorted({repr(key) for key in keys if isinstance(key, Hashable) and keys.count(key) > 1}) + if repeated: + raise yaml.constructor.ConstructorError(None, None, f"repeated keys {', '.join(repeated)}", node.start_mark) + return super().construct_mapping(node, deep=deep) - 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. + +def load_gate(gate_file: str) -> dict[str, dict[str, Any]]: + """The gate in ``gate_file``: per metric, optional ``min``/``max`` bounds, a ``key`` into a dict value (default: the + metric's primary entry) and a ``version``. 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. + ValueError: If the file does not map metric names to these fields, a bound is not a finite number, the key is + not a name or the version not an integer. """ - 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 + try: + gate = yaml.load(Path(gate_file).read_text(encoding="utf-8"), Loader=_GateLoader) + except (OSError, yaml.YAMLError) as exc: + raise ValueError(f"{gate_file} is not a readable YAML gate: {exc}") from exc + if not isinstance(gate, Mapping): + raise ValueError(f"{gate_file} must map metric names to bounds, such as 'name: {{min: 0.5}}'.") + entries = {} + for name, bound in gate.items(): + bound = {} if bound is None else bound + if not isinstance(bound, Mapping) or not set(bound) <= _GATE_FIELDS: + raise ValueError(f"Gate entry {name!r} must map some of {sorted(_GATE_FIELDS)}, not {bound!r}.") + if not isinstance(bound.get("key", ""), str): + raise ValueError(f"Gate entry {name!r} has a key that is not an entry name such as p95: {bound['key']!r}.") + if isinstance(bound.get("version", 1), bool) or not isinstance(bound.get("version", 1), int): + raise ValueError(f"Gate entry {name!r} has a version that is not an integer: {bound['version']!r}.") + entries[name] = {**bound, **{side: _bound(name, bound[side]) for side in ("min", "max") if side in bound}} + return entries + + +def _bound(name: str, limit: Any) -> float: + """A gate bound as a number: YAML reads ``1e-3`` as a string, and ``true`` as a bool, which is not a bound.""" + number = limit + if isinstance(limit, str): + with contextlib.suppress(ValueError): + number = float(limit) + if not is_finite_number(number): + raise ValueError(f"Gate entry {name!r} has a bound that is not a finite number: {limit!r}.") + return number + + +def gate_results(report: Mapping[str, Any], gate: Mapping[str, Mapping[str, Any]]) -> dict[str, str | None]: + """Per gated metric, None when the report meets its bounds, else why it does not. + + A gated metric that is missing, not applicable, errored or at another version fails. + + Args: + report: A report from :func:`build_report`. + gate: A gate from :func:`load_gate`. + """ + return {name: _gate_failure(report["metrics"].get(name), bound) for name, bound in gate.items()} + + +def _gate_failure(row: Mapping[str, Any] | None, bound: Mapping[str, Any]) -> str | None: + if row is None or row["status"] != "ok": + return "not reported" if row is None else row["status"] + if "version" in bound and bound["version"] != row["version"]: + return f"version {row['version']}, gate written for version {bound['version']}" + value = row["value"] + if isinstance(value, Mapping): 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 + if key not in value: + return f"needs a key into {sorted(value)}" + value = value[key] + elif "key" in bound: + return f"key {bound['key']!r} given for a value without entries" + if "min" in bound and value < bound["min"]: + return f"{value:.4g} is below {bound['min']}" + if "max" in bound and value > bound["max"]: + return f"{value:.4g} is above {bound['max']}" + return None -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) +CONSOLE_WIDTH = 100 +"""Width the console summary wraps its lists of metrics to.""" -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"]] +def format_report(report: Mapping[str, Any], gate: Mapping[str, str | None] | None = None) -> str: + """The report as a console summary: a verdict, the computed metrics by area, and what was not computed and why. + + The JSON report holds every detail; the summary keeps a metric's primary value, its source baseline, its sample + count and, for rates, the 95% interval. + + Args: + report: A report from :func:`build_report`. + gate: Results from :func:`gate_results`, to mark each gated metric as met or failed. + """ + inputs, metrics = report["inputs"], report["metrics"] + computed = {name: row for name, row in metrics.items() if row["status"] == "ok"} + errors = {name: row for name, row in metrics.items() if row["status"] == "error"} + skipped = {name: row for name, row in metrics.items() if row["status"] == "not_applicable"} + + title = [f"AutoData metrics · {Path(inputs['dataset']['path']).name}"] + title += [f"run {inputs['run_id'][:8]}"] if inputs["run_id"] else [] + if inputs["source"]: + valid, total = report["source_checks"]["num_valid_episodes"], report["source_checks"]["num_episodes"] + title.append( + f"source {Path(inputs['source']['path']).name}" + (f" ({valid}/{total} valid)" if valid < total else "") + ) + verdict = [f"{report['num_valid_episodes']}/{report['num_episodes']} episodes valid", f"{len(computed)} computed"] + verdict += [f"{len(skipped)} not computed"] if skipped else [] + verdict += [_count(len(errors), "errors")] if errors else [] + if gate is not None: + failed = sum(reason is not None for reason in gate.values()) + verdict.append( + f"gate failed on {failed} of {len(gate)}" if failed else f"gate passed on {len(gate)} of {len(gate)}" + ) + lines = [" · ".join(title), " · ".join(verdict)] + for name, entry in report["check_failures"].items(): + if entry["failed"]: + lines.append( + f" ✗ {name} failed {_count(entry['failed'], 'episodes')}, e.g. {', '.join(entry['examples'])}" + ) + + rows = [] + for name, row in computed.items(): + mark = " " if gate is None or name not in gate else ("✗" if gate[name] else "✓") + baseline = row.get("baseline") + source = "" if baseline is None else _console_value(baseline, row, labelled=False) + rows.append((row["area"], mark, name, _console_value(row["value"], row), source, _based_on(row))) + if rows: + has_source = any(row[4] for row in rows) + name_width = max(len(row[2]) for row in rows) + value_width = max(len(row[3]) for row in rows + [("", "", "", "Value")]) + source_width = max(len(row[4]) for row in rows + [("", "", "", "", "Source")]) + + def line(mark: str, name: str, value: str, source: str, based_on: str) -> str: + cells = [f"{mark} {name.ljust(name_width)}", value.rjust(value_width)] + cells += [source.rjust(source_width)] if has_source else [] + return " ".join(cells + [based_on]).rstrip() + + lines += ["", line(" ", "", "Value", "Source", "Based on")] + for area in AREAS: + area_rows = sorted((row for row in rows if row[0] == area), key=lambda row: row[2]) + if area_rows: + lines.append(area.upper()) + lines += [line(*row[1:]) for row in area_rows] + if errors: + lines += ["", "ERRORS"] + [f" {name}: {row['reason']}" for name, row in errors.items()] + if skipped: + groups: dict[str, list[str]] = {} + for name, row in skipped.items(): + groups.setdefault(_not_computed_because(row, inputs), []).append(name) + lines += ["", "NOT COMPUTED"] + for reason, names in sorted(groups.items(), key=lambda item: -len(item[1])): + lines += textwrap.wrap( + ", ".join(sorted(names)), CONSOLE_WIDTH, initial_indent=f" {reason}: ", subsequent_indent=" " + ) + if report["warnings"]: + lines += [""] + [f"Warning: {warning}" for warning in report["warnings"]] return "\n".join(lines) + + +def _count(count: int, noun: str) -> str: + return f"{count:,} {noun.removesuffix('s') if count == 1 else noun}" + + +def _based_on(row: Mapping[str, Any]) -> str: + """The sample count, with a rate's 95% interval and a pass rate's check coverage.""" + parts = [_count(row["n"], row["sample"])] + interval, run, skipped = (row["details"].get(key) for key in ("interval_95", "checks_run", "checks_skipped")) + if isinstance(interval, list) and len(interval) == 2 and all(isinstance(bound, int | float) for bound in interval): + low, high = (_console_number(bound, row["unit"]) for bound in interval) + parts.append(f"95% CI {low.removesuffix('%')}–{high}") + if isinstance(run, list) and isinstance(skipped, list) and skipped: + parts.append(f"{len(run)} of {len(run) + len(skipped)} checks") + return " · ".join(parts) + + +def _console_value(value: Any, row: Mapping[str, Any], labelled: bool = True) -> str: + """A value, or a dict value's primary entry (every entry without one), as the console shows it.""" + if not isinstance(value, Mapping): + return _console_number(value, row["unit"]) + keys = [row["primary"]] if row["primary"] in value else list(value) + return ", ".join((f"{key} " if labelled else "") + _console_number(value[key], row["unit"]) for key in keys) + + +def _console_number(value: float, unit: str) -> str: + """Fractions as percentages, seconds as durations; other units follow the number unless they hold a space, which + marks a description of the quantity rather than a unit.""" + if unit.startswith("fraction"): + percent = f"{100 * value:.3g}%" + return ">99.9%" if percent == "100%" and value < 1 else percent + if unit == "s" or unit.startswith("s "): + return _duration(value) + text = f"{value:,.0f}" if abs(value) >= 100 else re.sub(r"e\+?(-?)0*(\d)", r"e\1\2", f"{value:.3g}") + return text if " " in unit else f"{text} {unit}" + + +def _duration(seconds: float) -> str: + if seconds < 120: + return f"{seconds:.3g} s" + minutes, rest = divmod(round(seconds), 60) + if minutes < 120: + return f"{minutes} min {rest:02d} s" + return f"{minutes // 60} h {minutes % 60:02d} min" + + +_METRIC_INPUT_KEYS = ("env_args.autodata", "step_success", "obs/datagen_info/") +"""Evidence keys a generated dataset holds only when recorded with ``--record_metric_inputs``.""" + + +def _not_computed_because(row: Mapping[str, Any], inputs: Mapping[str, Any]) -> str: + """What would make a not-applicable metric computable, or its own reason.""" + missing = row.get("missing") + if missing is None: + return row["reason"] + if missing == "source": + return "needs --source" + if missing.startswith("source."): + return "needs a --source annotated with --record_metric_inputs" + if missing == "attempts" or missing.startswith("run."): + return "needs --result_file" if inputs["result_file"] is None else "needs --record_metric_inputs" + if missing.startswith("attempts.extras."): + return "not reported by this algorithm" + if missing.startswith(_METRIC_INPUT_KEYS): + return "needs --record_metric_inputs" + return row["reason"] diff --git a/autodata_metrics/validity.py b/autodata_metrics/validity.py new file mode 100644 index 0000000..6fa360e --- /dev/null +++ b/autodata_metrics/validity.py @@ -0,0 +1,228 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Validity metrics: whether the recorded data is well formed and the demonstrated task holds.""" + +from __future__ import annotations + +import h5py +import numpy as np + +from autodata_metrics.evidence import Episode, Evidence +from autodata_metrics.registry import NotApplicable, Value, check, metric + +ROTATION_TOLERANCE = 1e-3 +"""Largest deviation from a proper rotation, or from a unit quaternion, a recorded pose may show.""" +BLACK_FRAME_MEAN = 2.0 +"""Mean pixel value, out of 255, below which a camera frame counts as black.""" +FROZEN_FRAMES = 10 +"""Identical consecutive camera frames that count as a frozen stream.""" +FRAME_CHUNK = 64 +"""Steps read at a time, so checks of camera frames and other large arrays stream instead of loading whole episodes.""" +PER_EPISODE_PATHS = ("initial_state", "obs/datagen_info/subtask_term_signals", "obs/datagen_info/subtask_start_signals") +"""Paths whose arrays are not one entry per step: the initial state, and annotation's subtask signals.""" +OBSERVED_POSE_PATHS = ("obs/datagen_info/eef_pose/", "obs/datagen_info/object_pose/") +"""Recorded poses that are observations; commanded targets are measured by ``action_rotation_error`` instead.""" + + +def _image_streams(episode: Episode, ev: Evidence) -> dict[str, h5py.Dataset]: + """The episode's camera streams; not applicable to a dataset that records none.""" + if not ev.camera_paths: + raise NotApplicable("no camera streams") + return {path: dataset for path, dataset in episode.datasets() if path in ev.camera_paths} + + +def _all_finite(dataset: h5py.Dataset) -> bool: + if dataset.ndim == 0: + return bool(np.isfinite(dataset[()])) + return all(np.isfinite(dataset[start : start + FRAME_CHUNK]).all() for start in range(0, len(dataset), FRAME_CHUNK)) + + +@check("finite") +def finite(episode: Episode, ev: Evidence) -> str | None: + """Fails an episode with a NaN or infinite value in any floating-point array.""" + bad = [path for path, dataset in episode.datasets() if dataset.dtype.kind == "f" and not _all_finite(dataset)] + return f"non-finite values in {', '.join(bad)}" if bad else None + + +@check("aligned_lengths") +def aligned_lengths(episode: Episode, ev: Evidence) -> str | None: + """Fails an episode whose per-step arrays do not all hold one entry per action; camera streams are checked apart.""" + num_steps = len(episode["actions"]) if "actions" in episode else 0 + if num_steps == 0: + return "no actions" + bad = [ + path + for path, dataset in episode.datasets() + if dataset.ndim > 0 + and not path.startswith(PER_EPISODE_PATHS) + and path not in ev.camera_paths + and len(dataset) != num_steps + ] + return f"{', '.join(bad)} do not hold {num_steps} steps" if bad else None + + +def _rotation_errors(poses: np.ndarray) -> np.ndarray: + """Per 4x4 pose, how far its rotation is from a proper one: the larger of the largest entry of ``|R^T R - I|`` + and ``|det R - 1|``, which also catches reflections.""" + rotations = poses[..., :3, :3] + orthonormality = np.abs(np.einsum("...ji,...jk->...ik", rotations, rotations) - np.eye(3)).max(axis=(-2, -1)) + return np.maximum(orthonormality, np.abs(np.linalg.det(rotations) - 1.0)) + + +@check("valid_rotations") +def valid_rotations(episode: Episode, ev: Evidence) -> str | None: + """Fails an episode with an observed pose rotation that is not a proper rotation or a root quaternion that is not unit. + + Both tests hold for either quaternion order, so the check needs no convention. + """ + bad, checked = [], False + for path, dataset in episode.datasets(): + if path.startswith(OBSERVED_POSE_PATHS) and dataset.shape[-2:] == (4, 4): + error = _rotation_errors(dataset[()]).max(initial=0.0) + elif path.endswith("root_pose") and dataset.shape[-1:] == (7,): + error = np.abs(np.linalg.norm(dataset[()][..., 3:7], axis=-1) - 1.0).max(initial=0.0) + else: + continue + checked = True + if error > ROTATION_TOLERANCE: + bad.append(path) + if not checked: + raise NotApplicable("no recorded rotations") + return f"invalid rotations in {', '.join(bad)}" if bad else None + + +@check("pose_action_bounds", requires=["env_args.autodata.pose_action"]) +def pose_action_bounds(episode: Episode, ev: Evidence) -> str | None: + """Fails an episode whose pose actions exceed the bound the embodiment clips them to.""" + actions = episode["actions"] + bad = [] + for eef, layout in ev.env_args["autodata"]["pose_action"].items(): + if layout.get("bound") is None: + continue + start, end = layout["action_slice"] + peak = float(np.abs(actions[:, start:end]).max(initial=0.0)) + if peak > layout["bound"] + 1e-6: + bad.append(f"{eef} reaches {peak:.3g}") + return f"pose actions beyond their bound: {', '.join(bad)}" if bad else None + + +@check("one_frame_per_step", group="image") +def one_frame_per_step(episode: Episode, ev: Evidence) -> str | None: + """Fails an episode that lacks a camera stream the dataset records, or whose streams do not hold one frame per action.""" + streams = _image_streams(episode, ev) + missing = sorted(ev.camera_paths - streams.keys()) + if missing: + return f"{', '.join(missing)} not recorded" + num_steps = len(episode["actions"]) + bad = [path for path, dataset in streams.items() if len(dataset) != num_steps] + return f"{', '.join(bad)} do not hold {num_steps} frames" if bad else None + + +@check("no_black_frames", group="image") +def no_black_frames(episode: Episode, ev: Evidence) -> str | None: + """Fails an episode with a camera frame whose mean pixel value is below :data:`BLACK_FRAME_MEAN`.""" + for path, dataset in _image_streams(episode, ev).items(): + for start in range(0, len(dataset), FRAME_CHUNK): + means = dataset[start : start + FRAME_CHUNK].mean(axis=(1, 2, 3)) + if (means < BLACK_FRAME_MEAN).any(): + return f"{path} is black at step {start + int(np.argmax(means < BLACK_FRAME_MEAN))}" + return None + + +@check("no_frozen_frames", group="image") +def no_frozen_frames(episode: Episode, ev: Evidence) -> str | None: + """Fails an episode whose camera repeats one frame :data:`FROZEN_FRAMES` times in a row.""" + for path, dataset in _image_streams(episode, ev).items(): + previous, repeats = None, 1 + for start in range(0, len(dataset), FRAME_CHUNK): + for offset, frame in enumerate(dataset[start : start + FRAME_CHUNK]): + repeats = repeats + 1 if previous is not None and np.array_equal(frame, previous) else 1 + if repeats >= FROZEN_FRAMES: + return f"{path} repeats one frame {FROZEN_FRAMES} times up to step {start + offset}" + previous = frame + return None + + +def _pass_rate(ev: Evidence, group: str, none_apply: str) -> Value: + if not ev.all_episodes: + raise NotApplicable("the dataset has no episodes") + considered = [name for name, check_group in ev.check_groups.items() if check_group == group] + names = [name for name in considered if name in ev.failures] + if not names: + raise NotApplicable(none_apply) + failing = {episode for name in names for episode in ev.failures[name]} + details = {"checks_run": names, "checks_skipped": [name for name in considered if name not in ev.failures]} + return Value(1 - len(failing) / len(ev.all_episodes), n=len(ev.all_episodes), details=details) + + +@metric( + "correctness_pass_rate", + area="validity", + signal="data correctness", + scope="episode", + unit="fraction", + sample="episodes", + better="higher", +) +def correctness_pass_rate(ev: Evidence) -> Value: + """Share of episodes passing every correctness check: finite, aligned, valid rotations, actions in bounds.""" + return _pass_rate(ev, "correctness", "no correctness check applies to this dataset") + + +@metric( + "image_stream_pass_rate", + area="validity", + signal="data correctness", + scope="episode", + unit="fraction", + sample="episodes", + better="higher", +) +def image_stream_pass_rate(ev: Evidence) -> Value: + """Share of episodes whose camera streams hold one frame per step, no black frame and no frozen run.""" + return _pass_rate(ev, "image", "no camera streams") + + +@metric( + "final_success_rate", + area="validity", + signal="task correctness", + scope="dataset", + unit="fraction", + sample="episodes", + better="higher", + requires=["step_success"], + baseline=True, +) +def final_success_rate(ev: Evidence) -> Value: + """Share of valid episodes whose task success holds at their last step.""" + held = [bool(episode["step_success"][-1]) for episode in ev.episodes] + return Value(sum(held) / len(held), n=len(held)) + + +@metric( + "action_rotation_error", + area="validity", + signal="data correctness", + scope="dataset", + unit="max(|R^T R - I|, |det R - 1|)", + sample="poses", + better="lower", + primary="p95", + requires=["obs/datagen_info/target_eef_pose"], + baseline=True, +) +def action_rotation_error(ev: Evidence) -> Value: + """p50 and p95 deviation of the commanded target rotations from a valid rotation, over every step and end effector.""" + errors = [] + for episode in ev.episodes: + for eef in episode.children("obs/datagen_info/target_eef_pose"): + targets = episode[f"obs/datagen_info/target_eef_pose/{eef}"] + # A channel that commands no pose records all-zero targets, which are not rotations. + errors.append(_rotation_errors(targets[np.abs(targets).reshape(len(targets), -1).max(axis=1) > 0])) + errors = np.concatenate(errors) if errors else np.empty(0) + if not len(errors): + raise NotApplicable("no end effector commands a pose") + p50, p95 = np.percentile(errors, [50, 95]) + return Value({"p50": float(p50), "p95": float(p95)}, n=len(errors)) diff --git a/autodata_metrics/viability.py b/autodata_metrics/viability.py index a8de7a1..7f94094 100644 --- a/autodata_metrics/viability.py +++ b/autodata_metrics/viability.py @@ -58,6 +58,7 @@ def _counted_attempts(ev: Evidence) -> tuple[list[dict[str, Any]], bool]: signal="generation performance", scope="run", unit="fraction", + sample="attempts", better="higher", requires=["attempts"], ) @@ -77,6 +78,7 @@ def generation_success_rate(ev: Evidence) -> Value: signal="generation performance", scope="run", unit="demos/GPU-h", + sample="demos", better="higher", requires=["run.generation_time_s", "run.num_gpus", "env_args.autodata.exported_episodes"], ) @@ -96,19 +98,25 @@ def demos_per_gpu_hour(ev: Evidence) -> Value: signal="time to dataset", scope="run", unit="s", + sample="runs", better="lower", requires=["run.setup_time_s", "run.generation_time_s"], + version=2, ) def time_to_dataset(ev: Evidence) -> Value: - """Wall time of the automated steps: annotation of the source, setup and generation.""" + """Wall time of the automated steps, each process timed from its start: annotating the source and generating. + + Annotation that was manual, or recorded without its mode, is reported in ``uncounted_annotation_s`` instead. + """ 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) + parts = {"setup_s": ev.run["setup_time_s"], "generation_s": ev.run["generation_time_s"]} + uncounted = None + if annotation: + automatic = annotation.get("mode") == "auto" + parts["annotation_s"] = annotation.get("setup_time_s", 0.0) + (annotation["time_s"] if automatic else 0.0) + uncounted = None if automatic else annotation["time_s"] + details = {**parts, "annotation_mode": annotation.get("mode"), "uncounted_annotation_s": uncounted} + return Value(sum(parts.values()), n=1, details=details) @metric( @@ -117,6 +125,7 @@ def time_to_dataset(ev: Evidence) -> Value: signal="time to dataset", scope="run", unit="fraction", + sample="recordings", better="higher", requires=["source.env_args.autodata.annotation"], ) @@ -135,6 +144,7 @@ def annotation_pass_rate(ev: Evidence) -> Value: signal="failure causes", scope="run", unit="fraction of attempts", + sample="attempts", better="lower", requires=["attempts.extras.planning_failed"], ) @@ -151,6 +161,7 @@ def planning_failure_rate(ev: Evidence) -> Value: signal="generation performance", scope="run", unit="s per attempt", + sample="attempts", better="lower", primary="p95", requires=["attempts.extras.planning_time_s"], diff --git a/autodata_tests/e2e/test_annotation.py b/autodata_tests/e2e/test_annotation.py index fa93c2e..f8e8991 100644 --- a/autodata_tests/e2e/test_annotation.py +++ b/autodata_tests/e2e/test_annotation.py @@ -41,6 +41,7 @@ def test_automatic_annotation_from_dataset(tmp_path): assert_valid_dataset(str(output_file), min_num_demos=1) entry = assert_recorded_episodes(str(output_file)) assert entry["annotation"]["annotated_episodes"] == entry["exported_episodes"] + assert entry["annotation"]["mode"] == "auto" and entry["annotation"]["setup_time_s"] > 0 with h5py.File(input_file, "r") as source, h5py.File(output_file, "r") as annotated: assert len(annotated["data"]) == len(source["data"]) assert json.loads(annotated["data"].attrs["env_args"])["env_name"] == "IsaacContrib-Stack-Cube-Franka-IK-Rel" diff --git a/autodata_tests/e2e/test_mimicgen_data_generation.py b/autodata_tests/e2e/test_mimicgen_data_generation.py index 16b27f1..e5584cc 100644 --- a/autodata_tests/e2e/test_mimicgen_data_generation.py +++ b/autodata_tests/e2e/test_mimicgen_data_generation.py @@ -12,8 +12,15 @@ 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"} +# Metrics a MimicGen run recorded with --record_metric_inputs must compute; no camera, so no image checks. +METRICS = { + "generation_success_rate", + "demos_per_gpu_hour", + "time_to_dataset", + "correctness_pass_rate", + "final_success_rate", + "action_rotation_error", +} def _run_franka_cube_stack_mimicgen(num_envs: int, device: str) -> None: @@ -54,7 +61,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) + assert_ok_metrics(output_file, result_file, 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 bf45200..8e6ec57 100644 --- a/autodata_tests/e2e/test_skillgen_bin_data_generation.py +++ b/autodata_tests/e2e/test_skillgen_bin_data_generation.py @@ -21,13 +21,16 @@ 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 = { +# Metrics a SkillGen run recorded with --record_metric_inputs must compute; the planning metrics are SkillGen's own. +METRICS = { "generation_success_rate", "demos_per_gpu_hour", "time_to_dataset", "planning_time", "planning_failure_rate", + "correctness_pass_rate", + "final_success_rate", + "action_rotation_error", } @@ -74,7 +77,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) + assert_ok_metrics(output_file, result_file, METRICS) with open(result_file, encoding="utf-8") as result_handle: result = json.load(result_handle) diff --git a/autodata_tests/metrics/test_records.py b/autodata_tests/metrics/test_records.py new file mode 100644 index 0000000..50bc4f0 --- /dev/null +++ b/autodata_tests/metrics/test_records.py @@ -0,0 +1,238 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Generated robustness tests: every malformed record or gate is refused or reported, and no check raises.""" + +import h5py +import json +import numpy as np +import re +from collections.abc import Iterator +from pathlib import Path + +import pytest + +from autodata_metrics import build_report, format_report, gate_results, load_gate +from autodata_metrics import report as report_module +from autodata_metrics.evidence import Evidence +from autodata_metrics.registry import REGISTERED_CHECKS +from autodata_utils.generation_result import input_record + +ABSENT = object() +SOURCE_RECORD = "source record" +"""Stands for the source's path and SHA-256, filled in once the source file is written.""" +WRONG_VALUES = (ABSENT, None, "", "x", -1, 1.5, 10**400, True, float("nan"), [], {}, {"x": 1}) + + +@pytest.fixture(autouse=True) +def _no_git(monkeypatch): + monkeypatch.setattr(report_module, "code_revision", lambda: None) + + +def _write_dataset(path: Path, entry: dict, num_episodes: int = 2) -> str: + pose = np.tile(np.eye(4), (3, 1, 1)) + 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}) + for index in range(num_episodes): + episode = data.create_group(f"demo_{index}") + episode.create_dataset("actions", data=np.zeros((3, 7))) + episode.create_dataset("step_success", data=[False, True, True]) + episode.create_dataset("obs/datagen_info/eef_pose/arm", data=pose) + episode.create_dataset("obs/datagen_info/target_eef_pose/arm", data=pose) + return str(path) + + +def _samples(tmp_path: Path) -> dict[str, dict]: + """Records as AutoData writes them, for a run that exported two episodes from a source of two.""" + annotation = { + "mode": "auto", + "input_episodes": 2, + "processed_episodes": 2, + "annotated_episodes": 2, + "setup_time_s": 10.0, + "time_s": 20.0, + } + source_entry = {"version": 1, "run_id": "annotation", "exported_episodes": 2, "annotation": annotation} + run = { + "run_id": "run-a", + "seed": 1, + "revision": None, + "inputs": {"source": SOURCE_RECORD, "env_profile": None}, + "device": "cuda:0", + "gpu": "GPU", + "cpu": "CPU", + "num_gpus": 1, + "num_envs": 1, + "setup_time_s": 5.0, + "generation_time_s": 30.0, + "attempts_file": "result.run-a.attempts.jsonl", + } + attempt = { + "run_id": "run-a", + "env_id": 0, + "success": True, + "start_time_s": 0.0, + "end_time_s": 1.0, + "initial_object_positions": {"cube": [0.1, 0.2]}, + "extras": {"planning_time_s": 0.5, "planning_failed": False}, + } + return { + "entry": { + "version": 1, + "run_id": "run-a", + "exported_episodes": 2, + "pose_action": {"arm": {"action_slice": [0, 6], "bound": 1.0}}, + }, + "source_entry": source_entry, + "result": {"algorithm": "mimicgen", "num_success": 2, "num_failures": 1, "num_attempts": 3, "run": run}, + "attempt": attempt, + } + + +def _paths(record: dict, prefix: tuple = ()) -> Iterator[tuple]: + """Every field path of a record, nested objects included.""" + for key, value in record.items(): + yield (*prefix, key) + if isinstance(value, dict): + yield from _paths(value, (*prefix, key)) + + +def _replaced(record: dict, path: tuple, value: object) -> dict: + copy = json.loads(json.dumps(record)) + parent = copy + for key in path[:-1]: + parent = parent[key] + if value is ABSENT: + del parent[path[-1]] + else: + parent[path[-1]] = value + return copy + + +def _report_or_refusal(tmp_path: Path, records: dict) -> dict | None: + """Build, summarize and serialize the report of these records; None when the report refuses them.""" + dataset = _write_dataset(tmp_path / "data.hdf5", records["entry"]) + source = _write_dataset(tmp_path / "source.hdf5", records["source_entry"]) + result_file = tmp_path / "result.json" + result_file.write_text( + json.dumps(records["result"]).replace(json.dumps(SOURCE_RECORD), json.dumps(input_record(source))) + ) + first = records["attempt"] + lines = [first, {**first, "start_time_s": 1.0, "end_time_s": 2.0}, {**first, "success": False, "start_time_s": 2.0}] + (tmp_path / "result.run-a.attempts.jsonl").write_text("".join(json.dumps(line) + "\n" for line in lines)) + gate = tmp_path / "gate.yaml" + gate.write_text("generation_success_rate: {min: 0.1}\naction_rotation_error: {max: 0.001}\n") + try: + report = build_report(dataset, result_file=str(result_file), source=source) + except ValueError: + return None + json.dumps(report, allow_nan=False) + format_report(report, gate_results(report, load_gate(str(gate)))) + return report + + +def test_every_malformed_record_field_is_refused_or_reported(tmp_path): + samples = _samples(tmp_path) + report = _report_or_refusal(tmp_path, samples) + assert report is not None and report["metrics"]["time_to_dataset"]["status"] == "ok" + unread = {**samples["attempt"], "initial_object_positions": {"cube": [float("nan"), 0.0]}} + assert _report_or_refusal(tmp_path, {**samples, "attempt": unread}) is not None + negative = {**samples["attempt"], "extras": {"planning_time_s": -5.0, "planning_failed": False}} + assert _report_or_refusal(tmp_path, {**samples, "attempt": negative}) is None + crashes = [] + for kind in ("entry", "source_entry", "result", "attempt"): + for path in _paths(samples[kind]): + for wrong in WRONG_VALUES: + records = {**samples, kind: _replaced(samples[kind], path, wrong)} + try: + _report_or_refusal(tmp_path, records) + except Exception as exc: # Anything but a refusal is a crash this test exists to find. + crashes.append(f"{kind}.{'.'.join(path)} = {wrong!r}: {type(exc).__name__}: {exc}") + assert not crashes, "\n".join(crashes[:20]) + + +def test_unreadable_run_files_are_refused(tmp_path): + samples = _samples(tmp_path) + crashes = [] + for name in ("result.json", "result.run-a.attempts.jsonl"): + for raw in (b"", b"{", b"\xff\xfe", b"[" * 10000 + b"]" * 10000): + _report_or_refusal(tmp_path, samples) + (tmp_path / name).write_bytes(raw) + try: + build_report( + str(tmp_path / "data.hdf5"), + result_file=str(tmp_path / "result.json"), + source=str(tmp_path / "source.hdf5"), + ) + except ValueError: + continue + except Exception as exc: # Anything but a refusal is a crash this test exists to find. + crashes.append(f"{name} = {raw[:8]!r}: {type(exc).__name__}: {exc}") + assert not crashes, "\n".join(crashes) + + +def test_every_malformed_gate_field_is_refused_or_evaluated(tmp_path): + samples = _samples(tmp_path) + _report_or_refusal(tmp_path, samples) + report = build_report( + str(tmp_path / "data.hdf5"), result_file=str(tmp_path / "result.json"), source=str(tmp_path / "source.hdf5") + ) + gate = { + "generation_success_rate": {"min": 0.1, "version": 1}, + "action_rotation_error": {"max": 0.001, "key": "p95"}, + } + crashes = [] + for path in _paths(gate): + for wrong in WRONG_VALUES: + gate_file = tmp_path / "gate.yaml" + gate_file.write_text(json.dumps(_replaced(gate, path, wrong))) + try: + loaded = load_gate(str(gate_file)) + except ValueError: + continue + try: + format_report(report, gate_results(report, loaded)) + except Exception as exc: # Evaluating a loaded gate must never raise. + crashes.append(f"{'.'.join(path)} = {wrong!r}: {type(exc).__name__}: {exc}") + assert not crashes, "\n".join(crashes) + + +def test_checks_never_fail_an_episode_by_raising(tmp_path): + steps = 3 + degenerate = { + "scalar": {"obs/scalar": np.float32(1.0), "obs/robot/root_pose": np.float32(1.0)}, + "strings": {"obs/names": np.array([b"a", b"b", b"c"])}, + "empty_camera": {"obs/camera": np.zeros((0, 4, 4, 3), np.uint8)}, + "float_camera": {"obs/depth": np.zeros((steps, 4, 4, 1), np.float32)}, + "gray_camera": {"obs/gray": np.zeros((steps, 4, 4), np.uint8)}, + "zero_pose": {"obs/datagen_info/eef_pose/arm": np.zeros((steps, 4, 4))}, + "short_success": {"step_success": np.zeros(1, bool)}, + "no_actions": {}, + "empty_actions": {"actions": np.zeros((0, 7))}, + } + path = tmp_path / "d.hdf5" + with h5py.File(path, "w") as h5_file: + data = h5_file.create_group("data") + data.attrs["env_args"] = json.dumps({"env_name": "Test-v0"}) + for name, arrays in degenerate.items(): + episode = data.create_group(name) + if name != "no_actions": + episode.create_dataset("actions", data=arrays.pop("actions", np.zeros((steps, 7)))) + episode.create_group("obs/datagen_info/target_eef_pose") + for key, array in arrays.items(): + episode.create_dataset(key, data=array) + + with h5py.File(path, "r") as h5_file: + evidence = Evidence(h5_file) + report_module.run_checks(evidence, REGISTERED_CHECKS) + raised = { + f"{check} on {episode}: {reason}" + for check, failed in evidence.failures.items() + for episode, reason in failed.items() + if re.match(r"^[A-Z]\w*(Error|Exception):", reason) + } + assert not raised, "\n".join(sorted(raised)) + report = build_report(str(path)) + json.dumps(report, allow_nan=False) + format_report(report) diff --git a/autodata_tests/metrics/test_report.py b/autodata_tests/metrics/test_report.py index d5a5b90..e3b2f23 100644 --- a/autodata_tests/metrics/test_report.py +++ b/autodata_tests/metrics/test_report.py @@ -1,17 +1,20 @@ # 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.""" +"""Tests of how the report runs metrics, refuses mismatched runs, gates values and summarizes them.""" import h5py import json +import numpy as np +import re import subprocess import sys +from dataclasses import replace from pathlib import Path import pytest -from autodata_metrics import MetricSpec, Value, build_report, gate_violations +from autodata_metrics import REGISTERED_METRICS, MetricSpec, Value, build_report, format_report, gate_results, load_gate def _spec(name: str, compute, requires: tuple[str, ...] = ()) -> MetricSpec: @@ -22,6 +25,7 @@ def _spec(name: str, compute, requires: tuple[str, ...] = ()) -> MetricSpec: signal="test", scope="run", unit="fraction", + sample="samples", better="higher", requires=requires, primary=None, @@ -32,14 +36,18 @@ def _spec(name: str, compute, requires: tuple[str, ...] = ()) -> MetricSpec: 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 {})}) + autodata = {"autodata": {"version": 1, **entry}} if entry else {} + data.attrs["env_args"] = json.dumps({"env_name": "Test-v0", **autodata}) 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] + lines = [ + json.dumps({"run_id": run_id, "env_id": 0, "success": success, "start_time_s": step, "end_time_s": step + 1.0}) + for step, success in enumerate(successes) + ] (tmp_path / f"result.{run_id}.attempts.jsonl").write_text("\n".join(lines) + "\n") @@ -84,6 +92,27 @@ def test_interrupted_rerun_is_reported_from_its_own_attempt_log(tmp_path): assert "belongs to another run" in report["warnings"][0] +def test_dataset_without_an_entry_is_flagged_only_with_a_run_record(tmp_path): + dataset = _dataset(tmp_path / "d.hdf5") + + assert build_report(dataset, metrics={})["warnings"] == [] + assert "episode count only" in build_report(dataset, result_file=_run(tmp_path, "run-a"), metrics={})["warnings"][0] + + +def test_a_source_merged_after_annotation_has_its_annotation_set_aside(tmp_path): + annotation = {"input_episodes": 1, "processed_episodes": 1, "annotated_episodes": 1, "time_s": 2.0} + source = _dataset( + tmp_path / "source.hdf5", + num_episodes=3, + entry={"run_id": "a", "exported_episodes": 1, "annotation": annotation}, + ) + + report = build_report(_dataset(tmp_path / "d.hdf5"), source=source) + + assert "set aside" in report["warnings"][0] + assert report["metrics"]["annotation_pass_rate"]["status"] == "not_applicable" + + 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"): @@ -99,23 +128,54 @@ def test_report_refuses_inputs_that_disagree_with_their_run(tmp_path): 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={}) + malformed_entry = json.dumps({"autodata": {"version": 1, "run_id": 5, "exported_episodes": 1}}) + for corrupt, reason in ( + ("{'run_id': 'run-a'}", "must be JSON"), + (malformed_entry, r"autodata\.run_id must be str"), + ): + with h5py.File(dataset, "r+") as h5_file: + h5_file["data"].attrs["env_args"] = corrupt + with pytest.raises(ValueError, match=reason): + build_report(dataset, result_file=_run(tmp_path, "run-a"), metrics={}) + -def test_gate_flags_violations_and_refuses_a_non_finite_bound(tmp_path): +def test_gate_flags_violations_and_refuses_a_malformed_gate(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}} + metrics = { + "low": row, + "old": row, + "fine": row, + "keyed": 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" + "low: {min: 0.5}\nold: {version: 1}\nfine: {min: 0.3, max: 5e-1, version: 2}\nkeyed: {key: p95}\n" + "absent: {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)) + results = gate_results({"metrics": metrics}, load_gate(str(gate))) + + assert [name for name, reason in results.items() if reason is not None] == ["low", "old", "keyed", "absent", "gap"] + assert results["fine"] is None + for malformed in ( + "absent: {min: .nan}", + "absent: {min: false}", + "absent: {min: {p95: 1}}", + "absent: {mn: 0.5}", + "gap: {key: [p95]}", + "old: {version: '1'}", + "low: {min: 0.5}\nlow: {max: 1.0}", + "low: {min: 0.5", + "absent: 0.5", + "[low]", + "", + ): + gate.write_text(malformed + "\n") + with pytest.raises(ValueError): + load_gate(str(gate)) def test_non_finite_numbers_never_reach_the_report(tmp_path): @@ -128,10 +188,66 @@ def test_non_finite_numbers_never_reach_the_report(tmp_path): gate.write_text("m: {min: 0}\n") assert report["metrics"]["m"]["status"] == "error" - assert gate_violations(report, str(gate)) == ["m: error"] + assert gate_results(report, load_gate(str(gate))) == {"m": "error"} assert report["metrics"]["d"]["details"] == {"interval_95": [0.5, None]} +def test_console_summary_marks_the_gate_and_says_how_to_compute_the_rest(tmp_path): + metrics = { + "rate": _spec("rate", lambda ev: Value(0.25, n=4)), + "held": _spec("held", lambda ev: Value(1.0, n=1), ("step_success",)), + "planned": _spec("planned", lambda ev: Value(1.0, n=1), ("attempts.extras.planning_time_s",)), + } + report = build_report(_dataset(tmp_path / "d.hdf5"), metrics=metrics) + gate = tmp_path / "gate.yaml" + gate.write_text("rate: {min: 0.5}\n") + + summary = format_report(report, gate_results(report, load_gate(str(gate)))) + + assert "gate failed on 1 of 1" in summary + assert re.search(r"✗ rate +25% +4 samples", summary) + assert "needs --record_metric_inputs: held" in summary + assert "needs --result_file: planned" in summary + + +def test_degenerate_inputs_are_reported_without_breaking_the_summary_or_json(tmp_path): + path = tmp_path / "d.hdf5" + with h5py.File(path, "w") as h5_file: + data = h5_file.create_group("data") + data.attrs["env_args"] = json.dumps({"env_name": "Test-v0"}) + odd = data.create_group("demo_0") + odd.create_dataset("actions", data=np.zeros((2, 7))) + odd.create_dataset("obs/robot/root_pose", data=1.0) + odd.create_group("obs/datagen_info/target_eef_pose") + data.create_group("demo_1").create_dataset("actions", data=np.zeros((0, 7))) + metrics = { + "action_rotation_error": REGISTERED_METRICS["action_rotation_error"], + "open": _spec("open", lambda ev: Value(0.5, n=2, details={"interval_95": [0.1, float("inf")]})), + "partial": replace(_spec("partial", lambda ev: Value({"p50": 1.0}, n=1)), primary="p95"), + "almost": _spec("almost", lambda ev: Value(0.9995, n=2000, details={np.int64(3): 1})), + "uncounted": _spec("uncounted", lambda ev: Value(1.0, n=None)), + "bare": _spec("bare", lambda ev: 0.5), + "nested": _spec("nested", lambda ev: Value({"left": {"p95": 0.1}}, n=1)), + "loose": _spec("loose", lambda ev: Value(0.5, n=1, details=None)), + } + + report = build_report(str(path), result_file=str(tmp_path / "typo.json"), metrics=metrics) + + summary = format_report(report) + json.dumps(report, allow_nan=False) + assert report["num_valid_episodes"] == 1 and "aligned_lengths" in report["check_failures"] + assert report["metrics"]["action_rotation_error"]["status"] == "not_applicable" + assert re.search(r"open +50% +2 samples$", summary, re.M) and re.search(r"partial +p50 100%", summary) + assert re.search(r"almost +>99.9%", summary) + assert {name for name, row in report["metrics"].items() if row["status"] == "error"} == { + "uncounted", + "bare", + "nested", + "loose", + } + assert "typo.json does not exist" in report["warnings"][0] + + 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 diff --git a/autodata_tests/metrics/test_validity.py b/autodata_tests/metrics/test_validity.py new file mode 100644 index 0000000..2633450 --- /dev/null +++ b/autodata_tests/metrics/test_validity.py @@ -0,0 +1,127 @@ +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Known-answer tests of the episode checks and validity metrics, each on episodes with one known defect.""" + +import h5py +import json +import numpy as np +from pathlib import Path + +import pytest + +from autodata_metrics import REGISTERED_METRICS, build_report +from autodata_metrics.validity import FROZEN_FRAMES + + +def _write(path: Path, episodes: list[dict[str, np.ndarray]], entry: dict | None = None) -> str: + with h5py.File(path, "w") as h5_file: + data = h5_file.create_group("data") + env_args = {"env_name": "Test-v0", **({"autodata": {"version": 1, **entry}} if entry else {})} + data.attrs["env_args"] = json.dumps(env_args) + for index, arrays in enumerate(episodes): + group = data.create_group(f"demo_{index}") + for key, array in arrays.items(): + group.create_dataset(key, data=array) + return str(path) + + +def _metrics(*names: str) -> dict: + return {name: REGISTERED_METRICS[name] for name in names} + + +def test_failing_episodes_are_counted_and_left_out_of_dataset_metrics(tmp_path): + actions = np.zeros((3, 7)) + good = {"actions": actions, "step_success": np.array([False, False, True])} + non_finite = {"actions": np.where(np.eye(3, 7) > 0, np.nan, 0.0), "step_success": np.array([True, True, False])} + short = {"actions": actions, "step_success": np.array([False, False])} + empty = {"actions": np.zeros((0, 7)), "step_success": np.zeros(0, dtype=bool)} + + report = build_report( + _write(tmp_path / "d.hdf5", [good, non_finite, short, empty]), + metrics=_metrics("correctness_pass_rate", "final_success_rate"), + ) + + assert report["metrics"]["correctness_pass_rate"]["value"] == pytest.approx(1 / 4) + assert report["metrics"]["correctness_pass_rate"]["details"]["checks_skipped"] == [ + "valid_rotations", + "pose_action_bounds", + ] + assert report["metrics"]["final_success_rate"]["value"] == 1.0 and report["metrics"]["final_success_rate"]["n"] == 1 + assert {name: entry["failed"] for name, entry in report["check_failures"].items() if entry["failed"]} == { + "finite": 1, + "aligned_lengths": 2, + } + + +def test_rotation_and_bound_checks_flag_their_defects(tmp_path): + pose = np.tile(np.eye(4), (3, 1, 1)) + entry = {"run_id": "r", "exported_episodes": 3, "pose_action": {"arm": {"action_slice": [0, 6], "bound": 1.0}}} + clipped = {"actions": np.zeros((3, 7)), "obs/datagen_info/eef_pose/arm": pose} + beyond_bound = {"actions": np.full((3, 7), 1.5), "obs/datagen_info/eef_pose/arm": pose} + skewed = {"actions": np.zeros((3, 7)), "obs/datagen_info/eef_pose/arm": pose * [[1.1], [1.0], [1.0], [1.0]]} + zero = {"actions": np.zeros((3, 7)), "obs/datagen_info/eef_pose/arm": np.zeros((3, 4, 4))} + reflected = {"actions": np.zeros((3, 7)), "obs/datagen_info/eef_pose/arm": pose * [[-1.0], [1.0], [1.0], [1.0]]} + episodes = [clipped, beyond_bound, skewed, zero, reflected] + entry["exported_episodes"] = len(episodes) + + report = build_report(_write(tmp_path / "d.hdf5", episodes, entry), metrics={}) + + failures = {name: sorted(entry["examples"]) for name, entry in report["check_failures"].items() if entry["failed"]} + assert failures == {"pose_action_bounds": ["demo_1"], "valid_rotations": ["demo_2", "demo_3", "demo_4"]} + + +def test_image_checks_flag_black_frozen_and_missing_frames(tmp_path): + steps = FROZEN_FRAMES + 2 + rng = np.random.default_rng(0) + + def frames(count: int) -> np.ndarray: + return rng.integers(50, 200, size=(count, 4, 4, 3), dtype=np.uint8) + + black = frames(steps) + black[1] = 0 + frozen = np.repeat(frames(1), steps, axis=0) + episodes = [ + {"actions": np.zeros((steps, 7)), "obs/camera": camera} + for camera in (frames(steps), black, frozen, frames(steps - 1)) + ] + episodes.append({"actions": np.zeros((steps, 7))}) + + report = build_report(_write(tmp_path / "d.hdf5", episodes), metrics=_metrics("image_stream_pass_rate")) + + failures = {name: sorted(entry["examples"]) for name, entry in report["check_failures"].items() if entry["failed"]} + assert failures == { + "no_black_frames": ["demo_1"], + "no_frozen_frames": ["demo_2"], + "one_frame_per_step": ["demo_3", "demo_4"], + } + assert report["metrics"]["image_stream_pass_rate"]["value"] == pytest.approx(1 / 5) + + +def test_final_success_rate_reports_the_source_baseline(tmp_path): + ending = [{"actions": np.zeros((2, 7)), "step_success": np.array([True, held])} for held in (True, False)] + broken = {"actions": np.full((2, 7), np.nan), "step_success": np.array([True, False])} + source = _write(tmp_path / "source.hdf5", [ending[0], broken]) + + report = build_report(_write(tmp_path / "d.hdf5", ending), source=source, metrics=_metrics("final_success_rate")) + + row = report["metrics"]["final_success_rate"] + assert (row["value"], row["baseline"]) == (0.5, 1.0) + assert (report["source_checks"]["num_valid_episodes"], report["source_checks"]["num_episodes"]) == (1, 2) + + +def test_action_rotation_error_measures_commanded_rotations(tmp_path): + identity = np.tile(np.eye(4), (4, 1, 1)) + episode = { + "actions": np.zeros((4, 7)), + "obs/datagen_info/eef_pose/arm": identity, + "obs/datagen_info/target_eef_pose/arm": identity * [[1.01], [1.0], [1.0], [1.0]], + # A channel that commands no pose records an all-zero target; it is not a rotation and is not counted. + "obs/datagen_info/target_eef_pose/body": np.zeros((4, 4, 4)), + } + + report = build_report(_write(tmp_path / "d.hdf5", [episode]), metrics=_metrics("action_rotation_error")) + + row = report["metrics"]["action_rotation_error"] + assert row["value"]["p95"] == pytest.approx(1.01**2 - 1) and row["n"] == 4 + assert report["num_valid_episodes"] == 1 diff --git a/autodata_tests/metrics/test_viability.py b/autodata_tests/metrics/test_viability.py index ecaa78e..c0afed0 100644 --- a/autodata_tests/metrics/test_viability.py +++ b/autodata_tests/metrics/test_viability.py @@ -3,9 +3,11 @@ """Known-answer tests of the viability statistics.""" +from types import SimpleNamespace + import pytest -from autodata_metrics.viability import attempts_before_stop, wilson_interval +from autodata_metrics.viability import attempts_before_stop, time_to_dataset, wilson_interval def test_wilson_interval_matches_known_values(): @@ -26,3 +28,20 @@ def test_attempts_before_stop_keeps_attempts_started_before_the_first_env_finish 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 + + +def test_time_to_dataset_counts_only_annotation_recorded_as_automatic(): + run = {"setup_time_s": 20.0, "generation_time_s": 80.0} + + def measured(annotation: dict): + return time_to_dataset( + SimpleNamespace(source=SimpleNamespace(env_args={"autodata": {"annotation": annotation}}), run=run) + ) + + automatic = measured({"mode": "auto", "setup_time_s": 15.0, "time_s": 60.0}) + manual = measured({"mode": "manual", "setup_time_s": 15.0, "time_s": 600.0}) + unknown = measured({"time_s": 60.0}) + + assert automatic.value == 175.0 + assert (manual.value, manual.details["uncounted_annotation_s"]) == (115.0, 600.0) + assert (unknown.value, unknown.details["uncounted_annotation_s"]) == (100.0, 60.0) diff --git a/autodata_tests/utils/test_validate_dataset.py b/autodata_tests/utils/test_validate_dataset.py index d109227..eee0200 100644 --- a/autodata_tests/utils/test_validate_dataset.py +++ b/autodata_tests/utils/test_validate_dataset.py @@ -54,3 +54,14 @@ def test_report_options_without_report_are_an_error(tmp_path: Path) -> None: with pytest.raises(SystemExit): validate_dataset.main([str(tmp_path / "data.hdf5"), "--gate", str(tmp_path / "gate.yaml")]) + + +def test_report_refused_for_an_unreadable_source(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """A source the validator cannot read is refused in one line instead of a traceback.""" + + dataset = tmp_path / "data.hdf5" + with h5py.File(dataset, "w") as h5_file: + h5_file.create_group("data/demo_0").create_dataset("actions", data=[[0.0]]) + + assert validate_dataset.main([str(dataset), "--report", "--source", str(tmp_path / "missing.hdf5")]) == 1 + assert "Report refused" in capsys.readouterr().out diff --git a/autodata_tests/utils/utils.py b/autodata_tests/utils/utils.py index 14a47b6..42bd977 100644 --- a/autodata_tests/utils/utils.py +++ b/autodata_tests/utils/utils.py @@ -7,6 +7,7 @@ import json import os +from autodata_metrics.records import ATTEMPT, RESULT, parse, parse_env_args from autodata_tests.utils.constants import TestPaths from autodata_tests.utils.subprocess import run_subprocess @@ -49,8 +50,8 @@ def read_generation_result(result_file: str) -> tuple[int, int]: def assert_recorded_episodes(dataset_file: str) -> dict: """Assert the per-step recording and the ``autodata`` entry agree with the dataset's episodes. - Every episode has one ``step_success`` value per action, its success attribute equals whether - ``step_success`` ever held, and the entry counts every exported episode. + The entry has the shape the metrics read, every episode has one ``step_success`` value per action, its success + attribute equals whether ``step_success`` ever held, and the entry counts every exported episode. Returns: The dataset's ``autodata`` entry. @@ -58,7 +59,7 @@ def assert_recorded_episodes(dataset_file: str) -> dict: with h5py.File(dataset_file, "r") as h5_file: data = h5_file["data"] - entry = json.loads(data.attrs["env_args"])["autodata"] + entry = parse_env_args(json.loads(data.attrs["env_args"]), dataset_file)["autodata"] assert entry["exported_episodes"] == len(data), f"The entry counts {entry['exported_episodes']} episodes." for name, episode in data.items(): step_success = episode["step_success"][:] @@ -68,7 +69,8 @@ def assert_recorded_episodes(dataset_file: str) -> dict: def assert_recorded_run(dataset_file: str, result_file: str) -> list[dict]: - """Assert a generation run's dataset, result JSON and attempt log describe the same run. + """Assert a generation run's dataset, result JSON and attempt log have the shapes the metrics read and describe + the same run. Returns: The attempt-log lines. @@ -76,10 +78,10 @@ def assert_recorded_run(dataset_file: str, result_file: str) -> list[dict]: entry = assert_recorded_episodes(dataset_file) with open(result_file, encoding="utf-8") as result_handle: - result = json.load(result_handle) + result = parse(json.load(result_handle), RESULT, result_file) attempts_file = os.path.join(os.path.dirname(result_file), result["run"]["attempts_file"]) with open(attempts_file, encoding="utf-8") as attempts_handle: - attempts = [json.loads(line) for line in attempts_handle if line.strip()] + attempts = [parse(json.loads(line), ATTEMPT, attempts_file) for line in attempts_handle if line.strip()] assert entry["exported_episodes"] == result["num_success"], "The dataset and result count different successes." assert len(attempts) == result["num_attempts"], "The attempt log and result count different attempts." assert ( @@ -91,7 +93,8 @@ def assert_recorded_run(dataset_file: str, result_file: str) -> list[dict]: 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. + """Report a generation run's metrics with ``validate_dataset.py``; assert every episode passes the checks and + 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. @@ -109,7 +112,11 @@ def assert_ok_metrics(dataset_file: str, result_file: str, expected_ok: set[str] report_file, ]) with open(report_file, encoding="utf-8") as report_handle: - rows = json.load(report_handle)["metrics"] + report = json.load(report_handle) + assert ( + report["num_valid_episodes"] == report["num_episodes"] + ), f"{report['num_episodes'] - report['num_valid_episodes']} episodes failed checks: {report['check_failures']}" + rows = report["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}" diff --git a/docs/_ext/autodata_doc_tools.py b/docs/_ext/autodata_doc_tools.py index 72db1fd..f44d232 100644 --- a/docs/_ext/autodata_doc_tools.py +++ b/docs/_ext/autodata_doc_tools.py @@ -71,36 +71,62 @@ 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. +def _declarations(app: Sphinx, decorator: str) -> list[tuple[str, dict[str, Any], str]]: + """Name, options and summary line of every function in ``autodata_metrics/`` registered with ``decorator``. - 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. + The declarations are parsed instead of imported, so the docs build needs none of the metrics' dependencies and + the tables cannot drift from the code. In the summary, Sphinx roles become literals that need no reference + target, and pipes are escaped so they do not read as substitutions. """ - if ":autodata_metrics_table:" not in source[0]: - return - rows = [] + found = [] 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 call in node.decorator_list: + if isinstance(call, ast.Call) and getattr(call.func, "id", None) == decorator: + options = {keyword.arg: ast.literal_eval(keyword.value) for keyword in call.keywords} + summary = (ast.get_docstring(node) or "").split("\n")[0] + summary = re.sub(r":\w+:`~?([^`]+)`", r"``\1``", summary).replace("|", "\\|") + found.append((ast.literal_eval(call.args[0]), options, summary)) + return found + + +def _list_table(widths: str, header: tuple[str, ...], rows: list[tuple[str, ...]]) -> str: + lines = [".. list-table::", " :header-rows: 1", f" :widths: {widths}", ""] + lines += [" * - " + "\n - ".join(row) for row in [header, *rows]] + return "\n".join(lines) + "\n" + + +def autodata_metrics_table(app: Sphinx, _: Any, source: list[str]) -> None: + """Replaces ``:autodata_metrics_table:`` with a table of the built-in metrics.""" + if ":autodata_metrics_table:" not in source[0]: + return + rows = [ + ( + f"``{name}``", + options["area"], + options["unit"].replace("|", "\\|"), + ", ".join(f"``{key}``" for key in options.get("requires", ())) or "none", + summary, + ) + for name, options, summary in sorted(_declarations(app, "metric"), key=lambda item: (item[1]["area"], item[0])) + ] + table = _list_table("20 12 14 26 28", ("Metric", "Area", "Unit", "Requires", "Definition"), rows) + source[0] = source[0].replace(":autodata_metrics_table:", table) + + +def autodata_checks_table(app: Sphinx, _: Any, source: list[str]) -> None: + """Replaces ``:autodata_checks_table:`` with a table of the episode checks.""" + if ":autodata_checks_table:" not in source[0]: + return + declarations = [ + (options.get("group", "correctness"), name, summary) for name, options, summary in _declarations(app, "check") ] - 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") + rows = [(f"``{name}``", group, summary) for group, name, summary in sorted(declarations)] + source[0] = source[0].replace( + ":autodata_checks_table:", _list_table("22 14 64", ("Check", "Group", "Definition"), rows) + ) def setup(app: Sphinx) -> None: @@ -108,4 +134,5 @@ def setup(app: Sphinx) -> None: app.connect("source-read", autodata_code_link) app.connect("source-read", docker_run_command_replacer) app.connect("source-read", autodata_metrics_table) + app.connect("source-read", autodata_checks_table) app.add_config_value("autodata_docs_config", {}, "env") diff --git a/docs/pages/concepts/data_generator.rst b/docs/pages/concepts/data_generator.rst index 244451b..525100d 100644 --- a/docs/pages/concepts/data_generator.rst +++ b/docs/pages/concepts/data_generator.rst @@ -172,8 +172,10 @@ The ``run`` block is written only with ``--record_metric_inputs``. With that opt dataset, whether from generation, annotation or retargeting, also carries an ``autodata`` entry in its ``env_args`` attribute. The entry holds the ``run_id`` and the number of exported episodes, the embodiment's pose-action layout, the robot's joint names with their simulator limits, and the -embodiment's optional datasheet ``hardware_limits``. For a dataset generated with ``--result_file``, -the ``run_id`` ties it to that run's attempt log and, once the run completes, its result file. +embodiment's optional datasheet ``hardware_limits``. An annotated dataset's entry also records the +annotation mode (automatic or manual), its episode counts, and its setup and annotation times. For a +dataset generated with ``--result_file``, the ``run_id`` ties it to that run's attempt log and, once the +run completes, its result file. The GenerationAlgorithm Plug-In ------------------------------- diff --git a/docs/pages/references/metrics.rst b/docs/pages/references/metrics.rst index de230ad..e4d451e 100644 --- a/docs/pages/references/metrics.rst +++ b/docs/pages/references/metrics.rst @@ -13,11 +13,36 @@ without it, most metrics report that their inputs are missing. --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 command prints a summary; ``--report_file`` writes the full report, with every detail, as JSON. + +.. code-block:: text + + AutoData metrics · gr1.hdf5 · run e79e37b1 · source dataset_gr1_annotated.hdf5 + 6/6 episodes valid · 7 computed · 3 not computed · gate failed on 2 of 4 + + Value Source Based on + VALIDITY + ✗ action_rotation_error p95 0.0345 3.58e-7 1,612 poses + ✓ correctness_pass_rate 100% 6 episodes + ✓ final_success_rate 100% 6 episodes + image_stream_pass_rate 100% 6 episodes + VIABILITY + demos_per_gpu_hour 65.8 demos/GPU-h 6 demos + ✗ generation_success_rate 21.4% 28 attempts · 95% CI 10.2–39.5% + time_to_dataset 6 min 11 s 1 run + + NOT COMPUTED + not reported by this algorithm: planning_failure_rate, planning_time + needs a --source annotated with --record_metric_inputs: annotation_pass_rate + +- The second line is the verdict: valid episodes, how many metrics were computed, and, with ``--gate``, + the gate, whose metrics are marked ✓ or ✗. +- **Value** is the metric's value; a metric with percentiles shows the one a gate checks by default. + **Source** is the same metric on the ``--source`` dataset's valid episodes. **Based on** is the sample + count, with a rate's 95% interval and, when some checks did not apply, how many checks a pass rate rests + on. +- **Not computed** groups the metrics that do not apply by what would make them apply. A metric that + fails is listed under **Errors**; 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 @@ -25,6 +50,16 @@ attempt log. The report refuses a dataset, result file and attempt log whose epi 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. +What the report trusts +---------------------- + +The report checks every record it reads against the shape it expects before computing anything: the +dataset's ``env_args`` and ``autodata`` entry, the result file, each attempt-log line, and the gate. A +record with a missing field, a wrong type or a negative count is refused with a message naming the file +and the field, so no metric is computed from part of a malformed record. Fields the report does not read +are not checked. A metric loaded with ``--metrics`` runs as trusted code, but its ``Value`` is checked when +it is created: a malformed one makes that metric an error, and the other metrics are still computed. + Gating ------ @@ -36,7 +71,17 @@ Gating 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. +``version``. A gate entry may set ``min``, ``max``, ``key`` and ``version``; any other field, or a bound +that is not a finite number, is an error. + +Episode checks +-------------- + +Before any metric, every episode goes through the checks below. An episode that fails one is counted in +``correctness_pass_rate`` or ``image_stream_pass_rate`` and left out of every dataset metric; the report lists +the failures per check, with examples. + +:autodata_checks_table: Built-in metrics ---------------- @@ -60,6 +105,7 @@ requires. Without that evidence, the report says which key is missing and does n signal="generation performance", scope="run", unit="s", + sample="attempts", better="lower", requires=["attempts.extras.planning_time_s"], ) diff --git a/docs/pages/workflows/franka_cube_stack_mimicgen/step_3_generate_dataset.rst b/docs/pages/workflows/franka_cube_stack_mimicgen/step_3_generate_dataset.rst index 0e15b62..04734e4 100644 --- a/docs/pages/workflows/franka_cube_stack_mimicgen/step_3_generate_dataset.rst +++ b/docs/pages/workflows/franka_cube_stack_mimicgen/step_3_generate_dataset.rst @@ -71,6 +71,9 @@ For dataset-scale generation, run headless (``--viz none``) with parallel enviro Progress is printed after every attempt (successful demonstrations, attempts, and the running success rate). +To report the run's :doc:`metrics <../../references/metrics>` in the next step, add +``--record_metric_inputs`` and ``--result_file ./datasets/generated_dataset_mimicgen_franka.json``. + .. note:: **Upgrade validation:** the 100-environment GPU benchmark achieved 38.0% success diff --git a/docs/pages/workflows/franka_cube_stack_mimicgen/step_4_validate_dataset.rst b/docs/pages/workflows/franka_cube_stack_mimicgen/step_4_validate_dataset.rst index ea1a7d4..7fddc01 100644 --- a/docs/pages/workflows/franka_cube_stack_mimicgen/step_4_validate_dataset.rst +++ b/docs/pages/workflows/franka_cube_stack_mimicgen/step_4_validate_dataset.rst @@ -17,6 +17,21 @@ checked for the required fields (``actions``, ``initial_state``, ``obs``): For unreadable, invalid, or truncated files, see :ref:`troubleshooting-hdf5`. +Metrics +^^^^^^^ + +For a dataset generated with ``--record_metric_inputs`` and ``--result_file``, report its metrics: + +.. code-block:: bash + + python scripts/validate_dataset.py ./datasets/generated_dataset_mimicgen_franka.hdf5 \ + --report \ + --result_file ./datasets/generated_dataset_mimicgen_franka.json \ + --source ./datasets/annotated_datasets/dataset_franka_annotated.hdf5 + +See :doc:`../../references/metrics` for what each metric means and how to gate on it. + + Visual Validation ^^^^^^^^^^^^^^^^^ diff --git a/scripts/annotate_demos.py b/scripts/annotate_demos.py index ee5434f..e9ba4d6 100644 --- a/scripts/annotate_demos.py +++ b/scripts/annotate_demos.py @@ -30,11 +30,14 @@ """Launch Isaac Sim Simulator first.""" import argparse +import time from isaaclab.app import AppLauncher from autodata_utils.cli import callable_path +PROCESS_START = time.monotonic() + parser = argparse.ArgumentParser( description="Manually annotate AutoData source demonstrations with subtask signals.", formatter_class=argparse.RawDescriptionHelpFormatter, @@ -109,7 +112,6 @@ import math # noqa: E402 import os # noqa: E402 import sys # noqa: E402 -import time # noqa: E402 import torch # noqa: E402 import traceback # noqa: E402 import uuid # noqa: E402 @@ -337,9 +339,11 @@ def main() -> int: ) if autodata_entry is not None: autodata_entry["annotation"] = { + "mode": "auto" if args_cli.auto else "manual", "input_episodes": len(episode_names), "processed_episodes": processed_episode_count, "annotated_episodes": exported_episode_count, + "setup_time_s": annotation_start - PROCESS_START, "time_s": time.monotonic() - annotation_start, } print("Exiting the app.") diff --git a/scripts/validate_dataset.py b/scripts/validate_dataset.py index 8b9bb43..81dfbbc 100644 --- a/scripts/validate_dataset.py +++ b/scripts/validate_dataset.py @@ -23,7 +23,7 @@ import os import sys -from autodata_metrics import build_report, format_report, gate_violations +from autodata_metrics import build_report, format_report, gate_results, load_gate REQUIRED_DEMO_FIELDS = { "actions": h5py.Dataset, @@ -181,6 +181,10 @@ def main(argv: list[str] | None = None) -> int: 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") + try: + gate = load_gate(args.gate) if args.gate else None + except ValueError as exc: + parser.error(str(exc)) results = [validate_file(path) for path in args.files] print_results(results) @@ -195,15 +199,24 @@ def main(argv: list[str] | None = None) -> int: 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) + source = validate_file(args.source) if args.source else None + if source is not None and source.num_episodes is None: + print(f"\nReport refused: {args.source}: {'; '.join(source.issues)}") + return 1 + try: + report = build_report(args.files[0], result_file=args.result_file, source=args.source) + except ValueError as exc: + print(f"\nReport refused: {exc}") + return 1 + results = gate_results(report, gate) if gate is not None else None print() - print(format_report(report)) + print(format_report(report, results)) 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) + if results is not None: + violations = [f"{name}: {reason}" for name, reason in results.items() if reason is not None] print("\nGate passed." if not violations else "\nGate failed:\n " + "\n ".join(violations)) return 1 if num_invalid or violations else 0