Record metric inputs and run metadata during generation

autodata_core/algorithms.py

@@ -15,7 +15,8 @@
 
 from __future__ import annotations
 
-from collections.abc import Iterator
+import time
+from collections.abc import Iterable, Iterator
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
@@ -71,6 +72,14 @@ def validate_setup(self, datastream: Datastream) -> None:
         or ``datastream.get_env()`` (escape hatch) to enforce algorithm-specific invariants.
         """
 
+    def attempt_extras(self, eef_states: Iterable[_EEFGenerationState]) -> dict[str, Any]:
+        """Algorithm-specific facts about one finished attempt, written to its attempt-log line. Default: none.
+
+        Args:
+            eef_states: Per-EEF generation state of the attempt.
+        """
+        return {}
+
     def plan_subtask_trajectory(
         self,
         *,
@@ -190,6 +199,14 @@ def __init__(self, motion_planners: dict[int, Any]) -> None:
         assert motion_planners, "SkillGen requires at least one motion planner"
         self.motion_planners = motion_planners
 
+    def attempt_extras(self, eef_states: Iterable[_EEFGenerationState]) -> dict[str, Any]:
+        """Report the attempt's wall-clock motion-planning time [s] and whether a motion plan failed."""
+        states = list(eef_states)
+        return {
+            "planning_time_s": sum(state.planning_time_s for state in states),
+            "planning_failed": any(state.planning_failed for state in states),
+        }
+
     def plan_subtask_trajectory(
         self,
         *,
@@ -238,14 +255,17 @@ def plan_subtask_trajectory(
         )
 
         planner = self.motion_planners[env_id]
+        start_time = time.perf_counter()
         planning_success = planner.update_world_and_plan_motion(
             target_pose=target_pose,
             expected_attached_object=expected_attached_object,
             env_id=env_id,
             step_size=getattr(planner, "step_size", None),
             enable_retiming=(hasattr(planner, "step_size") and planner.step_size is not None),
         )
+        eef_state.planning_time_s += time.perf_counter() - start_time
         if not planning_success:
+            eef_state.planning_failed = True
             return None
 
         mp_waypoints = self._convert_planned_trajectory_to_waypoints(planner, target_passthrough_action)

autodata_core/data_generator.py

@@ -47,10 +47,18 @@ async def _optional_lock(lock: asyncio.Lock | None):
 
 @dataclass(frozen=True)
 class GenerationResult:
-    """Output of one :meth:`DataGenerator.generate` invocation."""
+    """Output of one :meth:`DataGenerator.generate` invocation.
+
+    Attributes:
+        initial_state: Scene state of every env right after this env's reset.
+        success: Whether the task succeeded at any step of the attempt.
+        extras: Algorithm-specific facts about the attempt, from
+            :meth:`~autodata_core.algorithms.GenerationAlgorithm.attempt_extras`.
+    """
 
     initial_state: dict
     success: bool
+    extras: dict[str, Any] = field(default_factory=dict)
 
 
 @dataclass
@@ -76,6 +84,8 @@ class _EEFGenerationState:
     # subtask trajectory here so the follow-up call splices it in.
     pending_subtask_trajectory: WaypointTrajectory | None = None
     is_in_motion_plan: bool = False
+    planning_time_s: float = 0.0
+    planning_failed: bool = False
 
 
 class DataGenerator:
@@ -569,7 +579,11 @@ async def generate(
                     )
                     if result is None:
                         # Planning failure (e.g. SkillGen motion planner). Abort with no success.
-                        return GenerationResult(initial_state=initial_state, success=False)
+                        return GenerationResult(
+                            initial_state=initial_state,
+                            success=False,
+                            extras=self.algorithm.attempt_extras(eef_states.values()),
+                        )
                     next_trajectory, _is_motion_plan = result
                     eef_state.current_trajectory = next_trajectory
                     eef_state.subtask_step_index = 0
@@ -620,6 +634,7 @@ async def generate(
         return GenerationResult(
             initial_state=initial_state,
             success=buffers.success,
+            extras=self.algorithm.attempt_extras(eef_states.values()),
         )
 
     # ------------------------------------------------------------------

autodata_examples/embodiments/franka_ik_rel.yaml

@@ -31,3 +31,16 @@ action_layout:
 # that same frame, so no shift is needed. SkillGen's cube-stack dataset is annotated in the
 # panda_hand frame, so it needs [0, 0, 0.1034] — use franka_cube_stack_embodiment_skillgen.yaml.
 eef_offset: [0.0, 0.0, 0.0]
+
+# Franka FCI datasheet joint limits, keyed by joint-name pattern. Read only by the dataset metrics.
+hardware_limits:
+  velocity: # [rad/s]
+    panda_joint[1-4]: 2.175
+    panda_joint[5-7]: 2.61
+  acceleration: # [rad/s^2]
+    panda_joint1: 15.0
+    panda_joint2: 7.5
+    panda_joint3: 10.0
+    panda_joint4: 12.5
+    panda_joint5: 15.0
+    panda_joint[6-7]: 20.0

autodata_examples/embodiments/franka_ik_rel_skillgen.yaml

@@ -35,3 +35,16 @@ action_layout:
 # approach axis) while the SkillGen dataset is annotated in the panda_hand frame, so the adapter
 # shifts reports back to panda_hand to keep planned transits and recorded skill segments aligned.
 eef_offset: [0.0, 0.0, 0.1034]
+
+# Franka FCI datasheet joint limits, keyed by joint-name pattern. Read only by the dataset metrics.
+hardware_limits:
+  velocity: # [rad/s]
+    panda_joint[1-4]: 2.175
+    panda_joint[5-7]: 2.61
+  acceleration: # [rad/s^2]
+    panda_joint1: 15.0
+    panda_joint2: 7.5
+    panda_joint3: 10.0
+    panda_joint4: 12.5
+    panda_joint5: 15.0
+    panda_joint[6-7]: 20.0

autodata_interfaces/embodiments/absolute_pose_embodiment_adapter.py

@@ -77,6 +77,18 @@ def from_dict(cls, data: dict[str, Any]) -> AbsolutePoseEmbodimentAdapter:
         """Build the adapter from a validated descriptor."""
         return cls(**data)
 
+    def metric_context(self) -> dict[str, Any]:
+        """Each end effector with a ``pose_action_slice`` commands its target pose there: position, then quaternion.
+
+        These entries are not clipped.
+        """
+        pose_action = {
+            name: {"action_slice": list(spec["pose_action_slice"]), "bound": None}
+            for name, spec in self.eefs.items()
+            if spec.get("pose_action_slice") is not None
+        }
+        return {**super().metric_context(), "pose_action": pose_action}
+
     def get_eef_names(self) -> tuple[str, ...]:
         return tuple(self.eefs)
 

autodata_interfaces/embodiments/bimanual_embodiment_adapter.py

@@ -191,6 +191,17 @@ def action_dim(self) -> int:
         channel_ends = [channel_slice[1] for channel_slice in self.passthrough_channels.values()]
         return max([self.hand_joints_slice[1], *channel_ends])
 
+    def metric_context(self) -> dict[str, Any]:
+        """``left_pose_slice`` and ``right_pose_slice`` of the action vector hold the hands' target poses.
+
+        Each holds a position, then a quaternion. These entries are not clipped.
+        """
+        pose_action = {
+            self.left.name: {"action_slice": list(self.left_pose_slice), "bound": None},
+            self.right.name: {"action_slice": list(self.right_pose_slice), "bound": None},
+        }
+        return {**super().metric_context(), "pose_action": pose_action}
+
     def action_to_target_eef_pose(self, action: torch.Tensor) -> dict[str, torch.Tensor]:
         """Convert env action to target EEF poses for both arms.
 

autodata_interfaces/embodiments/embodiment_adapter.py

@@ -50,6 +50,17 @@ class EmbodimentAdapter(ABC):
     robot_asset_name: str = "robot"
     raw_config: dict[str, Any] | None = None
 
+    def metric_context(self) -> dict[str, Any]:
+        """Describe this embodiment for the dataset metrics; the description is saved with every dataset it records.
+
+        The base returns the optional ``hardware_limits`` section of the embodiment config: datasheet joint limits
+        keyed by joint-name pattern. Subclasses add ``pose_action``, which says for each end effector which part of
+        the action vector commands its pose (``action_slice``, a half-open ``[start, end)`` index range) and the
+        value the adapter clips those entries to (``bound``, or None when they are not clipped).
+        """
+        hardware_limits = (self.raw_config or {}).get("hardware_limits")
+        return {"hardware_limits": hardware_limits} if hardware_limits else {}
+
     def get_navigation_eef_name(self) -> str | None:
         """Return the configured navigation channel without requiring a live environment."""
 

autodata_interfaces/embodiments/factory.py

@@ -70,7 +70,8 @@ def embodiment_adapter_from_dict(data: dict[str, Any], env: Any = None) -> Embod
         type_key in EMBODIMENT_TYPE_REGISTRY
     ), f"Unknown embodiment type {type_key!r}. Registered: {sorted(EMBODIMENT_TYPE_REGISTRY)}"
     cls = EMBODIMENT_TYPE_REGISTRY[type_key]
-    payload = {k: v for k, v in data.items() if k != "type"}
+    # ``hardware_limits`` is read from ``raw_config`` by :meth:`EmbodimentAdapter.metric_context`.
+    payload = {k: v for k, v in data.items() if k not in ("type", "hardware_limits")}
     adapter = cls.from_dict(payload)  # type: ignore[attr-defined]
     adapter.raw_config = data  # the full parsed config, for sections no typed adapter field models
     if env is not None:

autodata_interfaces/embodiments/single_arm_embodiment_adapter.py

@@ -133,6 +133,15 @@ class DeltaPoseIKSingleArmAdapter(SingleArmEmbodimentAdapter):
 
     clip_pose_action_to_unit: bool = True
 
+    def metric_context(self) -> dict[str, Any]:
+        """Action entries 0-5 command the end effector's pose change: delta position, then axis-angle rotation.
+
+        They are clipped to ±1 when ``clip_pose_action_to_unit`` is set.
+        """
+        bound = 1.0 if self.clip_pose_action_to_unit else None
+        pose_action = {self.eef_name: {"action_slice": [0, _DELTA_POSE_ACTION_DIM], "bound": bound}}
+        return {**super().metric_context(), "pose_action": pose_action}
+
     @property
     def action_dim(self) -> int:
         """Total width of the env action vector."""

autodata_interfaces/env/__init__.py

@@ -6,6 +6,7 @@
 from autodata_interfaces.env.env_profile import EnvironmentProfile
 from autodata_interfaces.env.isaaclab_env_interface import (
     apply_env_profile,
+    collect_autodata_entry,
     env_loop,
     get_env_name_from_dataset,
     setup_env_config,
@@ -18,5 +19,6 @@
     "get_env_name_from_dataset",
     "setup_output_paths",
     "env_loop",
+    "collect_autodata_entry",
     "setup_env_config",
 ]

autodata_interfaces/env/datagen_info_recorder.py

@@ -0,0 +1,95 @@
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Recorder terms that store the per-step data annotation, generation and the dataset metrics read."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from isaaclab.envs.mdp.recorders.recorders_cfg import ActionStateRecorderManagerCfg
+from isaaclab.managers import RecorderTerm, RecorderTermCfg, TerminationTermCfg
+from isaaclab.utils import configclass
+
+if TYPE_CHECKING:
+    from autodata_interfaces.datastream import Datastream
+
+
+class PreStepDatagenInfoRecorder(RecorderTerm):
+    """Records end-effector and commanded target poses, and optionally object poses, before each step.
+
+    Attributes:
+        datastream: Datastream to read. Set it on the class once the Datastream exists, since Isaac Lab
+            copies recorder configurations when building the env.
+    """
+
+    datastream: Datastream | None = None
+
+    def record_pre_step(self):
+        assert self.datastream is not None, "Set PreStepDatagenInfoRecorder.datastream before stepping the env."
+        datagen_info = {
+            "eef_pose": self.datastream.embodiment_adapter.get_eef_poses(env_ids=None),
+            "target_eef_pose": self.datastream.action_to_target_eef_pose(self._env.action_manager.action),
+        }
+        if self.cfg.record_object_poses:
+            datagen_info["object_pose"] = self.datastream.get_object_poses()
+        return "obs/datagen_info", datagen_info
+
+
+@configclass
+class PreStepDatagenInfoRecorderCfg(RecorderTermCfg):
+    """Configuration for :class:`PreStepDatagenInfoRecorder`."""
+
+    class_type: type[RecorderTerm] = PreStepDatagenInfoRecorder
+    record_object_poses: bool = True
+    """Annotation needs object poses for the source pool; generation leaves them out, as no metric reads them."""
+
+
+class PostStepSuccessRecorder(RecorderTerm):
+    """Records whether the task's success term holds after each step, as ``step_success``."""
+
+    def record_post_step(self):
+        success_term = self.cfg.success_term
+        return "step_success", success_term.func(self._env, **success_term.params)
+
+
+@configclass
+class PostStepSuccessRecorderCfg(RecorderTermCfg):
+    """Configuration for :class:`PostStepSuccessRecorder`."""
+
+    class_type: type[RecorderTerm] = PostStepSuccessRecorder
+    success_term: TerminationTermCfg | None = None
+    """The task's success term; the env itself runs without termination terms."""
+
+
+@configclass
+class DatagenInfoRecorderManagerCfg(ActionStateRecorderManagerCfg):
+    """Action/state recorder plus the per-step ``obs/datagen_info`` term."""
+
+    record_pre_step_datagen_info = PreStepDatagenInfoRecorderCfg()
+
+
+def add_step_success_recorder(recorder_cfg: object, success_term: TerminationTermCfg | None) -> None:
+    """Add the ``step_success`` term to any recorder manager configuration.
+
+    Nothing is added without a success term, since there is then no success to record.
+
+    Args:
+        recorder_cfg: Recorder manager configuration the env will be built with.
+        success_term: The task's success term, or None when the task has none.
+    """
+    if success_term is not None:
+        recorder_cfg.record_post_step_success = PostStepSuccessRecorderCfg(success_term=success_term)
+
+
+def add_pose_recorder(recorder_cfg: object) -> None:
+    """Add the per-step end-effector and commanded target pose term to any recorder manager configuration.
+
+    Object poses are left out, as no metric reads them. A configuration that already has the
+    ``record_pre_step_datagen_info`` term, such as :class:`DatagenInfoRecorderManagerCfg`, keeps its own.
+
+    Args:
+        recorder_cfg: Recorder manager configuration the env will be built with.
+    """
+    if not hasattr(recorder_cfg, "record_pre_step_datagen_info"):
+        recorder_cfg.record_pre_step_datagen_info = PreStepDatagenInfoRecorderCfg(record_object_poses=False)