Add Hydra experiment configuration example

Compose typed experiments from YAML and adapt them to the existing eval-runner Job and Namespace APIs.

Signed-off-by: Clemens Volk <cvolk@nvidia.com>

Author: cvolkcvolk

CONTEXT.md

@@ -0,0 +1,30 @@
+# Isaac Lab-Arena
+
+Isaac Lab-Arena defines composable robotics environments and evaluates policies against them.
+
+## Language
+
+**Environment Configuration**:
+Declarative values that specialize one environment provider for an Arena Experiment.
+_Avoid_: Hydra environment, environment wrapper
+
+**Environment Provider**:
+A named recipe that turns an environment configuration into an assembled Arena environment.
+_Avoid_: Runtime environment, Hydra environment
+
+**Arena Environment**:
+An assembled embodiment, scene, and task ready to be passed to an environment builder.
+_Avoid_: Environment configuration, environment provider
+
+**Gym Environment**:
+The instantiated simulation interface used for reset and step operations.
+_Avoid_: Arena environment configuration
+
+**Arena Experiment**:
+A portable, declarative evaluation condition pairing an environment configuration with a policy,
+rollout, variations, and repetition count. It contains no dispatch state or results.
+_Avoid_: Job, runtime execution, simulation application
+
+**Evaluation Job**:
+A runtime work item derived from an Arena Experiment and tracked through its execution lifecycle.
+_Avoid_: Experiment configuration, suite

isaaclab_arena/evaluation/__init__.py

@@ -0,0 +1,6 @@
+# Copyright (c) 2026, The Isaac Lab Arena Project Developers (https://github.com/isaac-sim/IsaacLab-Arena/blob/main/CONTRIBUTORS.md).
+# All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Arena policy-evaluation configuration and runners."""

isaaclab_arena/evaluation/arena_experiment_cfg.py

@@ -0,0 +1,71 @@
+# Copyright (c) 2026, The Isaac Lab Arena Project Developers (https://github.com/isaac-sim/IsaacLab-Arena/blob/main/CONTRIBUTORS.md).
+# All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Typed configuration for Arena experiments."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from isaaclab_arena.environments.arena_environment_cfg import ArenaEnvironmentCfg
+
+
+@dataclass
+class EnvironmentBuilderCfg:
+    """Configure how Arena builds the selected environment."""
+
+    num_envs: int = 1
+    env_spacing: float = 30.0
+    seed: int = 42
+    solve_relations: bool = True
+    placement_seed: int | None = None
+    resolve_on_reset: bool | None = None
+    random_yaw_init: bool = False
+    disable_fabric: bool = False
+    mimic: bool = False
+    presets: str | None = None
+
+    def __post_init__(self) -> None:
+        assert self.num_envs > 0, "num_envs must be greater than zero"
+
+
+@dataclass
+class RolloutCfg:
+    """Configure an evaluation rollout."""
+
+    num_steps: int = 2
+
+    def __post_init__(self) -> None:
+        assert self.num_steps > 0, "num_steps must be greater than zero"
+
+
+@dataclass
+class PolicyCfg:
+    """Select and configure an evaluation policy."""
+
+    type: str = "zero_action"
+    parameters: dict[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        assert self.type, "policy type must not be empty"
+
+
+@dataclass
+class ArenaExperimentCfg:
+    """Configure one portable Arena evaluation experiment."""
+
+    name: str
+    environment: ArenaEnvironmentCfg
+    """Concrete registered environment configuration selected during composition."""
+    environment_builder: EnvironmentBuilderCfg = field(default_factory=EnvironmentBuilderCfg)
+    policy: PolicyCfg = field(default_factory=PolicyCfg)
+    rollout: RolloutCfg = field(default_factory=RolloutCfg)
+    num_rebuilds: int = 1
+    variations: dict[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        assert self.name, "experiment name must not be empty"
+        assert self.num_rebuilds > 0, "num_rebuilds must be greater than zero"

isaaclab_arena/evaluation/eval_runner.py

@@ -212,6 +212,133 @@ def _run_in_chunks(args_cli: argparse.Namespace, master_cfg: dict) -> None:
             sys.exit(returncode)
 
 
+def evaluate_jobs(
+    args_cli: argparse.Namespace,
+    jobs: list[Job | dict],
+    *,
+    environment_loader=None,
+) -> None:
+    """Evaluate jobs inside an active ``SimulationAppContext``.
+
+    Args:
+        args_cli: Evaluation-runner options shared by every job.
+        jobs: Resolved jobs or legacy job dictionaries.
+        environment_loader: Optional callable that builds one job's environment.
+    """
+    job_manager = JobManager(jobs)
+    metrics_logger = MetricsLogger()
+
+    job_manager.print_jobs_info()
+
+    # One reverse-dated run directory shared by all jobs; each job gets a subdirectory within it.
+    # Always dated so every run produces its own report dir, recording or not.
+    # TODO(alexmillane): Currently each chunk produces its own output directory.
+    # We should use the same output directory for all chunks in the future.
+    run_output_dir = timestamped_run_dir(args_cli.output_base_dir)
+
+    if args_cli.record_viewport_video:
+        os.makedirs(run_output_dir, exist_ok=True)
+        print(f"[INFO] Video recording enabled. Videos will be saved to: {run_output_dir}")
+
+    for job in job_manager:
+        if job is None:
+            continue
+        env = None
+        policy = None
+
+        metrics_per_run: list[MetricsDataCollection] = []
+
+        # num_episodes is the total across rebuilds, so split it over the rebuilds.
+        num_episodes_per_rebuild = _split_episodes_across_rebuilds(job.num_episodes, job.num_rebuilds, job.name)
+
+        # Rebuild the environment and re-run the rollout job.num_rebuilds times, then
+        # aggregate the metrics across rebuilds into a single result.
+        for rebuild_idx in range(job.num_rebuilds):
+            try:
+                job_output_dir = os.path.join(run_output_dir, job.name)
+
+                # Per-job video output directory; cameras are tagged with the rebuild index.
+                video_cfg = VideoRecordingCfg(
+                    record_viewport_video=args_cli.record_viewport_video,
+                    record_camera_video=args_cli.record_camera_video,
+                    video_base_dir=job_output_dir,
+                    camera_name_prefix=f"robot-cam-rebuild{rebuild_idx}",
+                )
+                if environment_loader is None:
+                    env = load_env(
+                        job.arena_env_args,
+                        job.name,
+                        variations=job.variations,
+                        render_mode=video_cfg.render_mode,
+                        language_instruction=job.language_instruction,
+                    )
+                else:
+                    env = environment_loader(job, video_cfg.render_mode)
+
+                # Write per-episode results to disk.
+                # TODO: Aggregate the per-episode records across rebuilds into a single file,
+                # as is done for the metrics below.
+                results_path = os.path.join(job_output_dir, f"episode_results_rebuild{rebuild_idx}.jsonl")
+                env.unwrapped.episode_recorder.set_job_name(job.name)
+                env.unwrapped.episode_recorder.set_output_path(results_path)
+
+                policy = get_policy_from_job(job)
+
+                # Episodes allotted to this rebuild (None when the job is length-driven by steps).
+                num_episodes_this_rebuild = num_episodes_per_rebuild[rebuild_idx]
+
+                # Resolve simulation length: num_steps and num_episodes are mutually exclusive.
+                # Priority: job config -> policy length -> CLI default
+                if job.num_steps is None and num_episodes_this_rebuild is None:
+                    if policy.has_length():
+                        job.num_steps = policy.length()
+                    else:
+                        job.num_steps = args_cli.num_steps
+
+                env = wrap_env_for_video(env, video_cfg, job.num_steps, num_episodes_this_rebuild)
+
+                metrics = rollout_policy(
+                    env,
+                    policy,
+                    num_steps=job.num_steps,
+                    num_episodes=num_episodes_this_rebuild,
+                )
+
+                job_manager.complete_job(job, metrics=metrics, status=Status.COMPLETED)
+
+                # users may not specify metrics for a task, although it's not recommended
+                if metrics is not None:
+                    metrics_per_run.append(metrics)
+
+            except Exception as e:
+                job_manager.complete_job(job, metrics={}, status=Status.FAILED)
+                print(f"Job {job.name} failed with error: {e}")
+                print(f"Traceback: {traceback.format_exc()}")
+                if not args_cli.continue_on_error:
+                    raise
+
+            finally:
+                try:
+                    _close_job_resources(policy, env)
+                finally:
+                    policy = None
+                    env = None
+                    collect_garbage_and_clear_cuda_cache()
+
+        # Aggregate the metrics from the different experiments into a single view.
+        if metrics_per_run:
+            aggregated_metrics = aggregate_metrics(metrics_per_run)
+            metrics_logger.append_job_metrics(job.name, aggregated_metrics)
+
+    job_manager.print_jobs_info()
+    metrics_logger.print_metrics()
+
+    # Write HTML report.
+    report_path = build_report(run_output_dir)
+    if args_cli.serve_evaluation_report:
+        serve_until_ctrl_c(report_path.parent, args_cli.evaluation_report_port, report_path.name)
+
+
 def main():
     args_parser = get_isaaclab_arena_cli_parser()
     args_cli, unknown = args_parser.parse_known_args()
@@ -252,115 +379,7 @@ def main():
     enable_cameras_if_required(eval_jobs_config, args_cli)
 
     with SimulationAppContext(args_cli):
-        job_manager = JobManager(eval_jobs_config["jobs"])
-        metrics_logger = MetricsLogger()
-
-        job_manager.print_jobs_info()
-
-        # One reverse-dated run directory shared by all jobs; each job gets a subdirectory within it.
-        # Always dated so every run produces its own report dir, recording or not.
-        # TODO(alexmillane): Currently each chunk produces its own output directory.
-        # We should use the same output directory for all chunks in the future.
-        run_output_dir = timestamped_run_dir(args_cli.output_base_dir)
-
-        if args_cli.record_viewport_video:
-            os.makedirs(run_output_dir, exist_ok=True)
-            print(f"[INFO] Video recording enabled. Videos will be saved to: {run_output_dir}")
-
-        for job in job_manager:
-            if job is None:
-                continue
-            env = None
-            policy = None
-
-            metrics_per_run: list[MetricsDataCollection] = []
-
-            # num_episodes is the total across rebuilds, so split it over the rebuilds.
-            num_episodes_per_rebuild = _split_episodes_across_rebuilds(job.num_episodes, job.num_rebuilds, job.name)
-
-            # Rebuild the environment and re-run the rollout job.num_rebuilds times, then
-            # aggregate the metrics across rebuilds into a single result.
-            for rebuild_idx in range(job.num_rebuilds):
-                try:
-                    job_output_dir = os.path.join(run_output_dir, job.name)
-
-                    # Per-job video output directory; cameras are tagged with the rebuild index.
-                    video_cfg = VideoRecordingCfg(
-                        record_viewport_video=args_cli.record_viewport_video,
-                        record_camera_video=args_cli.record_camera_video,
-                        video_base_dir=job_output_dir,
-                        camera_name_prefix=f"robot-cam-rebuild{rebuild_idx}",
-                    )
-                    env = load_env(
-                        job.arena_env_args,
-                        job.name,
-                        variations=job.variations,
-                        render_mode=video_cfg.render_mode,
-                        language_instruction=job.language_instruction,
-                    )
-
-                    # Write per-episode results to disk.
-                    # TODO: Aggregate the per-episode records across rebuilds into a single file,
-                    # as is done for the metrics below.
-                    results_path = os.path.join(job_output_dir, f"episode_results_rebuild{rebuild_idx}.jsonl")
-                    env.unwrapped.episode_recorder.set_job_name(job.name)
-                    env.unwrapped.episode_recorder.set_output_path(results_path)
-
-                    policy = get_policy_from_job(job)
-
-                    # Episodes allotted to this rebuild (None when the job is length-driven by steps).
-                    num_episodes_this_rebuild = num_episodes_per_rebuild[rebuild_idx]
-
-                    # Resolve simulation length: num_steps and num_episodes are mutually exclusive.
-                    # Priority: job config -> policy length -> CLI default
-                    if job.num_steps is None and num_episodes_this_rebuild is None:
-                        if policy.has_length():
-                            job.num_steps = policy.length()
-                        else:
-                            job.num_steps = args_cli.num_steps
-
-                    env = wrap_env_for_video(env, video_cfg, job.num_steps, num_episodes_this_rebuild)
-
-                    metrics = rollout_policy(
-                        env,
-                        policy,
-                        num_steps=job.num_steps,
-                        num_episodes=num_episodes_this_rebuild,
-                    )
-
-                    job_manager.complete_job(job, metrics=metrics, status=Status.COMPLETED)
-
-                    # users may not specify metrics for a task, although it's not recommended
-                    if metrics is not None:
-                        metrics_per_run.append(metrics)
-
-                except Exception as e:
-                    job_manager.complete_job(job, metrics={}, status=Status.FAILED)
-                    print(f"Job {job.name} failed with error: {e}")
-                    print(f"Traceback: {traceback.format_exc()}")
-                    if not args_cli.continue_on_error:
-                        raise
-
-                finally:
-                    try:
-                        _close_job_resources(policy, env)
-                    finally:
-                        policy = None
-                        env = None
-                        collect_garbage_and_clear_cuda_cache()
-
-            # Aggregate the metrics from the different experiments into a single view.
-            if metrics_per_run:
-                aggregated_metrics = aggregate_metrics(metrics_per_run)
-                metrics_logger.append_job_metrics(job.name, aggregated_metrics)
-
-        job_manager.print_jobs_info()
-        metrics_logger.print_metrics()
-
-        # Write HTML report.
-        report_path = build_report(run_output_dir)
-        if args_cli.serve_evaluation_report:
-            serve_until_ctrl_c(report_path.parent, args_cli.evaluation_report_port, report_path.name)
+        evaluate_jobs(args_cli, eval_jobs_config["jobs"])
 
 
 if __name__ == "__main__":