Motion generation benchmarks

Contents

Motion generation benchmarks#

The bundled embodichain_benchmarks.motion_generation extension evaluates Motion Planning and Atomic Skills suites from another Python project. The benchmark owns case generation, simulation, lifecycle timing, evaluation and artifact persistence. It consumes the simulation, planning and Atomic Action APIs in embodichain and ships in the main wheel with the same version and dependencies. Callers select a suite and provide planner configuration.

Calling from another project#

from embodichain_benchmarks.motion_generation import (
    PlannerSpecCfg,
    load_suite,
    run_benchmark,
)

suite = load_suite("motion_panda_free_space_smoke")
suite.protocol.measured_trials = 5
result = run_benchmark(
    suite,
    planners=[PlannerSpecCfg(id="toppra", adapter="toppra")],
    device="cuda",
    output_root="./evaluation_results",
)
print(result.aggregates)
print(result.report_path)

Installed suite resources resolve independently of the working directory. An external YAML file resolves its component references relative to that file. The caller’s suite and planner specifications are copied. Explicit planners replace the suite’s planner list and are enabled; omitting the argument selects the suite’s enabled algorithms.

Pass adapters={"my_method": MyPlannerAdapter} and a matching PlannerSpecCfg(adapter="my_method", ...) to evaluate a class defined in another repository. The mapping applies to one call, including overrides of registered adapter names. Implement build(), plan(case) and close(); use prepare() for distinct lazy initialization. Declare supported capabilities and return a canonical batched PlanResult. Atomic Skills also require an exposed MotionGenerator and the atomic_action capability.

Calls own their simulation and return structured results after teardown. Run calls serially within a process. Install the selected backend and resolve its robot assets as for ordinary EmbodiChain use. The installed command is embodichain benchmark motion-generation; the module form is python -m embodichain_benchmarks.motion_generation.

Public entry points#

BenchmarkCase

One env-batched planner input frozen before execution.

BenchmarkRunResult

Paths and aggregate data produced by one completed run.

PlannerAdapter

Uniform lifecycle around a motion-planning implementation.

PlannerContext

Runtime objects shared with a planner adapter.

PlannerSpecCfg

Configuration for one registered planner adapter.

SuiteCfg

Resolved benchmark suite configuration.

load_suite([name_or_path])

Load an installed suite or an external YAML file.

run_all_benchmarks([num_waypoints_list, ...])

Resolve configuration and run all selected benchmark tracks.

run_benchmark([suite, planners, adapters, ...])

Evaluate planners against a suite without modifying caller configuration.

embodichain_benchmarks.motion_generation.run_benchmark(suite='motion_panda_free_space_smoke', *, planners=None, adapters=None, device='auto', headless=True, output_root='outputs/benchmarks', video=None, case_manifest=None)[source]#

Evaluate planners against a suite without modifying caller configuration.

Parameters:
  • suite (str | Path | SuiteCfg) – Installed suite name, YAML path, or assembled configuration. Embodiment component paths in YAML resolve relative to that file; component paths in an assembled configuration use the working directory. Adapter-specific paths follow the adapter’s rules; relative NMG model paths use the working directory.

  • planners (Sequence[PlannerSpecCfg] | None) – Algorithms to evaluate, replacing the suite’s planner list. Every explicitly supplied planner is enabled. When omitted, use the enabled planners declared by the suite.

  • adapters (Mapping[str, type[PlannerAdapter]] | None) – PlannerAdapter subclasses indexed by adapter name. Entries override registered adapters for this call only. The runner owns construction, preparation, warmup, evaluation and cleanup.

  • device (str) – Simulation device: auto, cpu, or a Torch device string.

  • headless (bool) – Whether to run without a viewer.

  • output_root (str | Path) – Directory under which a unique run directory is created.

  • video (VideoRecordCfg | None) – Optional Atomic Skills replay recording configuration.

  • case_manifest (str | Path | None) – Hashed case_manifest.json to replay instead of generating new inputs. Cases must match the suite’s tracks, version and robot. Retain the host-written resolved_embodiment identity; replay rejects changed robot/embodiment configurations before creating simulation. This supports fixed quality populations and declared timing subsets.

Return type:

BenchmarkRunResult

Returns:

Raw trial records, aggregate metrics and paths to persisted artifacts.

Raises:
  • ValueError – The suite is invalid or no planners are selected.

  • TypeError – A suite, planner or adapter has the wrong type.

Note

Runs create and destroy their own simulator. Execute calls serially within one process. Backend and model dependencies must be installed separately. Atomic Skills adapters also expose a MotionGenerator.

embodichain_benchmarks.motion_generation.load_suite(name_or_path='motion_panda_free_space_smoke')[source]#

Load an installed suite or an external YAML file.

Parameters:

name_or_path (str | Path) – Installed suite stem or path to an external YAML file.

Return type:

SuiteCfg

Returns:

A validated suite with component paths resolved relative to its file.

Raises:

FileNotFoundError – No file or installed suite matches the request.

class embodichain_benchmarks.motion_generation.BenchmarkRunResult[source]#

Paths and aggregate data produced by one completed run.

Methods:

__init__(run_dir, report_path, trials_path, ...)

__new__(**kwargs)

__init__(run_dir, report_path, trials_path, records, aggregates)#
__new__(**kwargs)#
class embodichain_benchmarks.motion_generation.PlannerAdapter[source]#

Uniform lifecycle around a motion-planning implementation.

Methods:

__init__(spec, context)

atomic_plan_options()

Return planner options for Atomic Skill calls that leave them unset.

availability()

Return whether this adapter can run in the current process.

build()

Construct the underlying planner without preparing lazy backends.

close()

Release backend resources when the implementation exposes them.

plan(case)

Plan one env-batched benchmark case.

prepare(case)

Prepare a lazy backend, or return None when not applicable.

prepare_cases(cases)

Prepare all backend variants required by a supported case set.

require_motion_generator()

Return the adapter-owned MotionGenerator or fail clearly.

resolve_motion_limits(*[, require_timing])

Freeze common budgets outside timing; expose them in run metadata.

supports_case(case)

Return whether one manifest case is representable without mutation.

Attributes:

metadata

Return stable identity, role, configuration, and capabilities.

motion_policy_planner

Return the backend name pinned into Atomic Action motion policies.

separate_prepare

Whether this backend exposes a distinct lazy preparation phase.

timing_is_solved

Whether this backend solves trajectory timing instead of assuming it.

__init__(spec, context)[source]#
atomic_plan_options()[source]#

Return planner options for Atomic Skill calls that leave them unset.

Atomic Skill providers stay planner-neutral and usually build motion policies without backend options. A backend whose own defaults cannot plan such calls returns the options it uses for free-space cases.

Return type:

PlanOptions | None

Returns:

Options applied to every motion policy without plan_opts, or None to keep the backend defaults.

availability()[source]#

Return whether this adapter can run in the current process.

Return type:

tuple[bool, str | None]

abstract build()[source]#

Construct the underlying planner without preparing lazy backends.

Return type:

None

close()[source]#

Release backend resources when the implementation exposes them.

Return type:

None

property metadata: PlannerMetadata#

Return stable identity, role, configuration, and capabilities.

property motion_policy_planner: str#

Return the backend name pinned into Atomic Action motion policies.

abstract plan(case)[source]#

Plan one env-batched benchmark case.

Return type:

PlanResult

prepare(case)[source]#

Prepare a lazy backend, or return None when not applicable.

Return type:

dict[str, object] | None

prepare_cases(cases)[source]#

Prepare all backend variants required by a supported case set.

Return type:

dict[str, object] | None

require_motion_generator()[source]#

Return the adapter-owned MotionGenerator or fail clearly.

Atomic Skill scenarios use this boundary instead of reaching into a backend implementation. Every planner that opts into the atomic_action capability must expose its generator here.

Return type:

MotionGenerator

resolve_motion_limits(*, require_timing=False)[source]#

Freeze common budgets outside timing; expose them in run metadata.

Backend-specific limit overrides must equal the common resolved budget. A stricter private budget changes the experiment and must be moved to the suite protocol instead of silently using a different comparison.

Return type:

dict[str, object]

separate_prepare: bool = False#

Whether this backend exposes a distinct lazy preparation phase.

supports_case(case)[source]#

Return whether one manifest case is representable without mutation.

Return type:

tuple[bool, str | None]

timing_is_solved: bool = False#

Whether this backend solves trajectory timing instead of assuming it.

Defaults to nominal, so a backend that has not declared otherwise is not judged on a clock it may never have solved: its dynamic-limit verdict reads N/A, which is visible, rather than passing or failing silently. A backend sets this only when its timing comes from a solve under limits.

Implementation reference#

The modules below expose the suite schema, extension contracts, evaluators and reporting helpers used by the shared runner. Motion Planning geometry and solved-timing rankings remain separate; Atomic Skills use measured physical success. Unsupported populations have explicit coverage semantics. Raw trials, the frozen case manifest, resolved suite, environment metadata and aggregate schema are persisted together with each report.

embodichain_benchmarks.motion_generation.aggregation#

Aggregate raw planner trials and build a complete benchmark leaderboard.

Functions:

aggregate_results(records, metadata, cases, ...)

Build scenario-owned report tables with separate geometry and timed ranks.

embodichain_benchmarks.motion_generation.aggregation.aggregate_results(records, metadata, cases, measured_trials, *, track_scenarios, confidence_level=0.95)[source]#

Build scenario-owned report tables with separate geometry and timed ranks.

track_scenarios must explicitly assign every case and trial track to a supported scenario family; report semantics are never inferred from names.

Return type:

dict[str, list[dict[str, object]]]

aggregate_generalization(records, metadata, ...)

Aggregate versioned nominal, robustness, and held-out populations.

embodichain_benchmarks.motion_generation.api#

Programmatic evaluation of installed suites and external planner adapters.

run_benchmark([suite, planners, adapters, ...])

Evaluate planners against a suite without modifying caller configuration.

embodichain_benchmarks.motion_generation.artifacts#

Reproducibility and raw-result artifacts for planner benchmarks.

Classes:

TrialJsonlWriter

Append numeric raw trial records to one JSONL artifact.

Functions:

create_run_directory(output_root, suite_name)

Create one timestamped benchmark run directory.

environment_metadata()

Collect software and hardware metadata needed to interpret results.

write_case_manifest(path, cases, *[, suite, ...])

Write the algorithm-independent case manifest.

write_json(path, value)

Write one UTF-8 JSON artifact with stable formatting.

write_resolved_suite(path, suite)

Write the fully resolved suite as YAML.

class embodichain_benchmarks.motion_generation.artifacts.TrialJsonlWriter[source]#

Append numeric raw trial records to one JSONL artifact.

Methods:

__init__(path)

append(record)

Append one record and flush it immediately for recoverability.

__init__(path)[source]#
append(record)[source]#

Append one record and flush it immediately for recoverability.

Return type:

None

embodichain_benchmarks.motion_generation.artifacts.create_run_directory(output_root, suite_name)[source]#

Create one timestamped benchmark run directory.

Return type:

Path

embodichain_benchmarks.motion_generation.artifacts.environment_metadata()[source]#

Collect software and hardware metadata needed to interpret results.

Return type:

dict[str, object]

embodichain_benchmarks.motion_generation.artifacts.write_case_manifest(path, cases, *, suite=None, planner_config_hashes=None, embodiment_resolution=None)[source]#

Write the algorithm-independent case manifest.

Return type:

Path

embodichain_benchmarks.motion_generation.artifacts.write_json(path, value)[source]#

Write one UTF-8 JSON artifact with stable formatting.

Return type:

Path

embodichain_benchmarks.motion_generation.artifacts.write_resolved_suite(path, suite)[source]#

Write the fully resolved suite as YAML.

Return type:

Path

embodichain_benchmarks.motion_generation.atomic_skill#

Case construction contract for Atomic Skill benchmark integrations.

Classes:

AtomicSkillCaseProvider

Generate and ground one Atomic Action without planner-specific logic.

Functions:

atomic_skill_provider_names()

Return registered Atomic Action case providers.

create_atomic_skill_provider(skill_id)

Construct one registered Atomic Action case provider.

register_atomic_skill_provider(skill_id, ...)

Register one Atomic Action case provider.

class embodichain_benchmarks.motion_generation.atomic_skill.AtomicSkillCaseProvider[source]#

Generate and ground one Atomic Action without planner-specific logic.

Methods:

articulation_effect_segment(case)

Return the segment that starts the articulation-displacement diagnostic.

build_invocation(scenario, case, adapter)

Ground the case into one planner-independent action invocation.

frozen_grasps(scenario, case)

Return the grasps a self-sampling skill must use, by gripper part.

generate_case(scenario, suite, track, ...)

Generate one frozen case and independent IK validity evidence.

initial_scene(scenario, case)

Return explicit scene observations the skill plans against.

initial_task_state(scenario, case)

Return an optional symbolic precondition for isolated action testing.

lift_segment_start(compiled)

Return the first lift waypoint that should release object dynamics.

object_id(case)

Return the manipulated object identifier when one exists.

object_rotation_segment(compiled)

Return the waypoint window whose object rotation is measured.

physical_evaluator()

Return a measured-stage evaluator, or delegate to task_result.

prefix_invocations(scenario, case)

Return actions compiled before this skill to establish its precondition.

replay_observation_windows(compiled, case)

Declare raw simulator traces required by the physical evaluator.

task_result(scenario, case, compiled, ...)

Evaluate a custom provider that does not supply a stage evaluator.

Attributes:

requires_gripper

samples_grasps

Whether the skill samples grasps itself and needs a grasp generator.

skill_id

articulation_effect_segment(case)[source]#

Return the segment that starts the articulation-displacement diagnostic.

Return type:

str | None

abstract build_invocation(scenario, case, adapter)[source]#

Ground the case into one planner-independent action invocation.

Return type:

ActionInvocation

frozen_grasps(scenario, case)[source]#

Return the grasps a self-sampling skill must use, by gripper part.

Skills that sample their own grasps, such as OpenDoor or HandOver, receive these manifest grasps instead, so every planner plans the same contacts.

Return type:

Mapping[str, tuple[Tensor, Tensor]]

Returns:

{gripper_part: (target_pose, grasp_pose)} batches; empty when the skill accepts explicit grasps or needs none.

abstract generate_case(scenario, suite, track, config, *, seed, batch_size)[source]#

Generate one frozen case and independent IK validity evidence.

Return type:

BenchmarkCase

initial_scene(scenario, case)[source]#

Return explicit scene observations the skill plans against.

Return type:

SceneSnapshot | None

Returns:

A snapshot that replaces the engine’s rigid-object observations, or None to keep them.

initial_task_state(scenario, case)[source]#

Return an optional symbolic precondition for isolated action testing.

Return type:

TaskState | None

lift_segment_start(compiled)[source]#

Return the first lift waypoint that should release object dynamics.

Return type:

int | None

object_id(case)[source]#

Return the manipulated object identifier when one exists.

Return type:

str | None

object_rotation_segment(compiled)[source]#

Return the waypoint window whose object rotation is measured.

Return type:

tuple[int, int] | None

Returns:

(start, stop) trajectory indices, or None when the skill is not scored on how far it rotates the held object.

physical_evaluator()[source]#

Return a measured-stage evaluator, or delegate to task_result.

Return type:

PhysicalStageEvaluator | None

Returns:

An evaluator that receives no compiled planning state.

prefix_invocations(scenario, case)[source]#

Return actions compiled before this skill to establish its precondition.

A skill that needs a held object, such as Pour, returns the PickUp that acquires it. The prefix is replayed and scored with the skill.

Return type:

tuple[ActionInvocation, ...]

Returns:

Invocations compiled in order before build_invocation().

replay_observation_windows(compiled, case)[source]#

Declare raw simulator traces required by the physical evaluator.

Providers resolve semantic action segments here. The replay adapter stores only measured tensors, so the evaluator remains independent of compiled planning state and the live simulator.

Return type:

tuple[ReplayObservationWindow, ...]

requires_gripper = False#
samples_grasps = False#

Whether the skill samples grasps itself and needs a grasp generator.

skill_id: str#
task_result(scenario, case, compiled, observation, motion_outcomes)[source]#

Evaluate a custom provider that does not supply a stage evaluator.

Return type:

tuple[Tensor, str]

embodichain_benchmarks.motion_generation.atomic_skill.atomic_skill_provider_names()[source]#

Return registered Atomic Action case providers.

Return type:

tuple[str, ...]

embodichain_benchmarks.motion_generation.atomic_skill.create_atomic_skill_provider(skill_id)[source]#

Construct one registered Atomic Action case provider.

Return type:

AtomicSkillCaseProvider

embodichain_benchmarks.motion_generation.atomic_skill.register_atomic_skill_provider(skill_id, provider_type)[source]#

Register one Atomic Action case provider.

Return type:

None

embodichain_benchmarks.motion_generation.cli#

Run Motion Planning and Atomic Skills suites through the shared benchmark.

cuRobo is the default primary baseline. IK interpolation and TOPPRA are optional diagnostic baselines. NMG is loaded from a standalone ONNX policy.

Run: embodichain benchmark motion-generation --suite motion_panda_free_space_smoke or select the Franka + PGI Atomic Skills suite with --suite atomic_franka_pgi_smoke.

Functions:

add_parser_arguments(parser)

Add planner benchmark options to an existing argument parser.

main()

Run the motion-generation CLI using process arguments.

run_all_benchmarks([num_waypoints_list, ...])

Resolve configuration and run all selected benchmark tracks.

run_from_args(args)

Run the benchmark from parsed unified-CLI arguments.

embodichain_benchmarks.motion_generation.cli.add_parser_arguments(parser)[source]#

Add planner benchmark options to an existing argument parser.

Return type:

None

embodichain_benchmarks.motion_generation.cli.main()[source]#

Run the motion-generation CLI using process arguments.

Return type:

None

embodichain_benchmarks.motion_generation.cli.run_all_benchmarks(num_waypoints_list=None, sim_device='auto', headless=True, nmg_onnx_path=None, *, suite_name='motion_panda_free_space_smoke', algorithms=None, extra_baselines=None, batch_sizes=None, path_shapes=None, start_state_bins=None, seeds=None, num_trials=None, warmup_trials=None, sample_interval=None, validation_samples=None, position_threshold_m=None, rotation_threshold_rad=None, nmg_pos_eps=None, nmg_rot_eps=None, output_root='outputs/benchmarks', video=None)[source]#

Resolve configuration and run all selected benchmark tracks.

Return type:

BenchmarkRunResult

embodichain_benchmarks.motion_generation.cli.run_from_args(args)[source]#

Run the benchmark from parsed unified-CLI arguments.

Return type:

BenchmarkRunResult

embodichain_benchmarks.motion_generation.config#

Configuration loading and hashing for motion-generation benchmark suites.

Classes:

AtomicPoseRandomizationCfg

Normalized random-pose settings for an atomic_skill track.

DomainCfg

Explicit provenance for one track's existing case population.

EmbodimentSpecCfg

Official component supplying the runtime robot and endpoint bindings.

FreeSpaceTrackCfg

Case matrix for the free-space-common track.

PlannerSpecCfg

Configuration for one registered planner adapter.

ProtocolCfg

Common timing and external-validation protocol.

RobotSpecCfg

Robot provider selected for every track in one suite run.

SuiteCfg

Resolved benchmark suite configuration.

TrackCfg

One enabled benchmark track and its scenario provider.

Functions:

resolve_atomic_batch_sizes(track)

Return simulator batch sizes for an Atomic Skill track.

resolve_atomic_pose_randomization(track)

Resolve shared random-pose controls from an Atomic Skill track.

stable_hash(value)

Return a stable SHA256 hash for JSON-compatible configuration data.

suite_to_dict(suite)

Convert a resolved suite to plain YAML/JSON-compatible values.

class embodichain_benchmarks.motion_generation.config.AtomicPoseRandomizationCfg[source]#

Normalized random-pose settings for an atomic_skill track.

The benchmark represents each random pose as one simulator environment row. Consequently pose_batch_size is both the number of independently perturbed poses sent to a batched Atomic Action goal and the default simulator batch size when a track does not declare batch_sizes.

The nested randomization mapping is intentionally small and limited to strict pose changes. Skill-level legacy keys such as object_position_jitter_m remain supported by the scenario provider; this object only normalizes track-level controls shared by all skills.

Methods:

__init__([enabled, pose_batch_size, mode, ...])

Attributes:

__init__(enabled=False, pose_batch_size=1, mode='translation', sampling_pattern='uniform', object_translation_jitter_m=(0.0, 0.0, 0.0), object_yaw_jitter_rad=0.0, object_roll_pitch_jitter_rad=(0.0, 0.0), target_translation_jitter_m=(0.0, 0.0, 0.0), target_yaw_jitter_rad=0.0, target_roll_pitch_jitter_rad=(0.0, 0.0), support_object_translation_jitter_m=(0.0, 0.0, 0.0), support_object_yaw_jitter_rad=0.0, support_object_roll_pitch_jitter_rad=(0.0, 0.0), task_frame_translation_jitter_m=(0.0, 0.0, 0.0), task_frame_yaw_jitter_rad=0.0, task_frame_roll_pitch_jitter_rad=(0.0, 0.0), seed_stride=1)#
enabled: bool = False#
mode: str = 'translation'#
object_roll_pitch_jitter_rad: tuple[float, float] = (0.0, 0.0)#
object_translation_jitter_m: tuple[float, float, float] = (0.0, 0.0, 0.0)#
object_yaw_jitter_rad: float = 0.0#
pose_batch_size: int = 1#
sampling_pattern: str = 'uniform'#
seed_stride: int = 1#
support_object_roll_pitch_jitter_rad: tuple[float, float] = (0.0, 0.0)#
support_object_translation_jitter_m: tuple[float, float, float] = (0.0, 0.0, 0.0)#
support_object_yaw_jitter_rad: float = 0.0#
target_roll_pitch_jitter_rad: tuple[float, float] = (0.0, 0.0)#
target_translation_jitter_m: tuple[float, float, float] = (0.0, 0.0, 0.0)#
target_yaw_jitter_rad: float = 0.0#
task_frame_roll_pitch_jitter_rad: tuple[float, float] = (0.0, 0.0)#
task_frame_translation_jitter_m: tuple[float, float, float] = (0.0, 0.0, 0.0)#
task_frame_yaw_jitter_rad: float = 0.0#
class embodichain_benchmarks.motion_generation.config.DomainCfg[source]#

Explicit provenance for one track’s existing case population.

This declaration labels cases; it does not generate perturbations or establish that a population was held out during model development.

Methods:

__init__([id, version, kind, overrides, ...])

copy(**kwargs)

Return a new object replacing specified fields with new values.

replace(**kwargs)

Return a new object replacing specified fields with new values.

to_dict()

Convert an object into dictionary recursively.

to_identity()

Validate and snapshot domain provenance for a frozen case.

validate([prefix])

Check the validity of configclass object.

Attributes:

__init__(id=<factory>, version=<factory>, kind=<factory>, overrides=<factory>, required_capabilities=<factory>, applicability=<factory>, objects=<factory>, scene_layout=<factory>, case_generator=<factory>, frozen_cases=<factory>, perturbations=<factory>, metadata=<factory>)#
applicability: list[str]#
case_generator: str | None#
copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

frozen_cases: list[dict[str, Any]]#
id: str#
kind: str#
metadata: dict[str, Any]#
objects: list[str]#
overrides: dict[str, Any]#
perturbations: dict[str, Any]#
replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

required_capabilities: list[str]#
scene_layout: str | None#
to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

to_identity()[source]#

Validate and snapshot domain provenance for a frozen case.

Return type:

EvaluationDomain

Returns:

Immutable domain identity, independent of planner configuration.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:
  • obj (object) – The object to check.

  • prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

version: str#
class embodichain_benchmarks.motion_generation.config.EmbodimentSpecCfg[source]#

Official component supplying the runtime robot and endpoint bindings.

Methods:

__init__([component, resource_id, ...])

copy(**kwargs)

Return a new object replacing specified fields with new values.

replace(**kwargs)

Return a new object replacing specified fields with new values.

to_dict()

Convert an object into dictionary recursively.

validate([prefix])

Check the validity of configclass object.

Attributes:

__init__(component=<factory>, resource_id=<factory>, overrides=<factory>, initial_qpos=<factory>, endpoint_bindings=<factory>, runtime_services=<factory>)#
component: str#
copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

endpoint_bindings: dict[str, Any]#
initial_qpos: list[float]#
overrides: dict[str, Any]#
replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

resource_id: str | None#
runtime_services: dict[str, Any]#
to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:
  • obj (object) – The object to check.

  • prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain_benchmarks.motion_generation.config.FreeSpaceTrackCfg[source]#

Case matrix for the free-space-common track.

Methods:

__init__([batch_sizes, waypoint_counts, ...])

copy(**kwargs)

Return a new object replacing specified fields with new values.

replace(**kwargs)

Return a new object replacing specified fields with new values.

to_dict()

Convert an object into dictionary recursively.

validate([prefix])

Check the validity of configclass object.

Attributes:

batch_modes

batch_sizes

constraint_pattern_design

All patterns, or all singletons plus one rotating pattern per other length.

constraint_patterns

Modality projections of the same screened Cartesian base requests.

distance_bins

distance_unit

Translation ranges in metres or fractions of the declared robot reach.

jacobian_position_scale_m

Characteristic length used to make Jacobian conditioning dimensionless.

joint_threshold_rad

Maximum per-joint error when matching ordered joint targets.

max_generation_attempts

motion_components

near_limit_threshold

near_singular_threshold

path_shapes

population_design

Full factorial geometry or a balanced rotation of shapes and waypoint counts.

reachability_solver

FK-only numerical screening, or explicit legacy robot_ik screening.

robot_reach_m

Explicit reference reach for normalized distances; never inferred from names.

sampling_method

Legacy joint templates or independently screened Cartesian requests.

screening_iterations

screening_restarts

seeds

singularity_candidates

start_state_bins

target_types

Target modalities; omission preserves the legacy Cartesian population.

translation_ranges

Optional near/mid/far bounds in distance_unit; omitted bins use defaults.

waypoint_counts

__init__(batch_sizes=<factory>, waypoint_counts=<factory>, path_shapes=<factory>, start_state_bins=<factory>, seeds=<factory>, target_types=<factory>, joint_threshold_rad=<factory>, sampling_method=<factory>, constraint_patterns=<factory>, constraint_pattern_design=<factory>, batch_modes=<factory>, population_design=<factory>, distance_bins=<factory>, motion_components=<factory>, max_generation_attempts=<factory>, singularity_candidates=<factory>, jacobian_position_scale_m=<factory>, near_singular_threshold=<factory>, near_limit_threshold=<factory>, distance_unit=<factory>, robot_reach_m=<factory>, translation_ranges=<factory>, reachability_solver=<factory>, screening_iterations=<factory>, screening_restarts=<factory>)#
batch_modes: list[str]#
batch_sizes: list[int]#
constraint_pattern_design: str#

All patterns, or all singletons plus one rotating pattern per other length.

constraint_patterns: list[list[str]]#

Modality projections of the same screened Cartesian base requests.

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

distance_bins: list[str]#
distance_unit: str#

Translation ranges in metres or fractions of the declared robot reach.

jacobian_position_scale_m: float#

Characteristic length used to make Jacobian conditioning dimensionless.

joint_threshold_rad: float#

Maximum per-joint error when matching ordered joint targets.

max_generation_attempts: int#
motion_components: list[str]#
near_limit_threshold: float#
near_singular_threshold: float#
path_shapes: list[str]#
population_design: str#

Full factorial geometry or a balanced rotation of shapes and waypoint counts.

reachability_solver: str#

FK-only numerical screening, or explicit legacy robot_ik screening.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

robot_reach_m: float | None#

Explicit reference reach for normalized distances; never inferred from names.

sampling_method: str#

Legacy joint templates or independently screened Cartesian requests.

screening_iterations: int#
screening_restarts: int#
seeds: list[int]#
singularity_candidates: int#
start_state_bins: list[str]#
target_types: list[str]#

Target modalities; omission preserves the legacy Cartesian population.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

translation_ranges: dict[str, list[float]]#

Optional near/mid/far bounds in distance_unit; omitted bins use defaults.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:
  • obj (object) – The object to check.

  • prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

waypoint_counts: list[int]#
class embodichain_benchmarks.motion_generation.config.PlannerSpecCfg[source]#

Configuration for one registered planner adapter.

Methods:

__init__([id, adapter, role, enabled, config])

copy(**kwargs)

Return a new object replacing specified fields with new values.

replace(**kwargs)

Return a new object replacing specified fields with new values.

to_dict()

Convert an object into dictionary recursively.

validate([prefix])

Check the validity of configclass object.

Attributes:

__init__(id=<factory>, adapter=<factory>, role=<factory>, enabled=<factory>, config=<factory>)#
adapter: str#
config: dict[str, Any]#
copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

enabled: bool#
id: str#
replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

role: str#
to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:
  • obj (object) – The object to check.

  • prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain_benchmarks.motion_generation.config.ProtocolCfg[source]#

Common timing and external-validation protocol.

Methods:

__init__([warmup_trials, measured_trials, ...])

copy(**kwargs)

Return a new object replacing specified fields with new values.

replace(**kwargs)

Return a new object replacing specified fields with new values.

to_dict()

Convert an object into dictionary recursively.

validate([prefix])

Check the validity of configclass object.

Attributes:

confidence_level

dynamic_limit_tolerance

Relative slack on dynamic-limit utilization before a violation is flagged.

joint_acceleration_limit_rad_s2

Per-joint or scalar acceleration limit, or None to report N/A.

joint_error_norm

Joint arrival is the maximum absolute controlled-joint error.

joint_jerk_limit_rad_s3

Per-joint or scalar jerk limit, or None to report N/A.

joint_limit_tolerance_rad

joint_velocity_limit_rad_s

Optional per-joint or scalar speed budget, capped by effective robot limits.

measured_trials

position_threshold_m

report_schema_version

rotation_threshold_rad

sample_interval

seeds

start_state_tolerance_rad

Maximum absolute per-joint error at the first trajectory sample.

validation_samples

warmup_trials

__init__(warmup_trials=<factory>, measured_trials=<factory>, sample_interval=<factory>, validation_samples=<factory>, position_threshold_m=<factory>, rotation_threshold_rad=<factory>, joint_error_norm=<factory>, joint_limit_tolerance_rad=<factory>, start_state_tolerance_rad=<factory>, confidence_level=<factory>, report_schema_version=<factory>, seeds=<factory>, joint_velocity_limit_rad_s=<factory>, joint_acceleration_limit_rad_s2=<factory>, joint_jerk_limit_rad_s3=<factory>, dynamic_limit_tolerance=<factory>)#
confidence_level: float#
copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

dynamic_limit_tolerance: float#

Relative slack on dynamic-limit utilization before a violation is flagged.

joint_acceleration_limit_rad_s2: float | list[float] | None#

Per-joint or scalar acceleration limit, or None to report N/A.

URDF carries no acceleration field and Robot exposes no qacc_limits, so this limit has no asset source and must be stated by the suite as an evaluation budget.

joint_error_norm: str#

Joint arrival is the maximum absolute controlled-joint error.

joint_jerk_limit_rad_s3: float | list[float] | None#

Per-joint or scalar jerk limit, or None to report N/A.

This is a suite-owned evaluation budget with no asset fallback.

joint_limit_tolerance_rad: float#
joint_velocity_limit_rad_s: float | list[float] | None#

Optional per-joint or scalar speed budget, capped by effective robot limits.

Planning and validation use the same resolved limit. The explicit budget also supplies a usable limit when a backend reports an unavailable asset limit; it cannot increase a valid robot limit.

measured_trials: int#
position_threshold_m: float#
replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

report_schema_version: str#
rotation_threshold_rad: float#
sample_interval: int#
seeds: list[int]#
start_state_tolerance_rad: float#

Maximum absolute per-joint error at the first trajectory sample.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:
  • obj (object) – The object to check.

  • prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

validation_samples: int#
warmup_trials: int#
class embodichain_benchmarks.motion_generation.config.RobotSpecCfg[source]#

Robot provider selected for every track in one suite run.

Methods:

__init__([id, provider, config])

copy(**kwargs)

Return a new object replacing specified fields with new values.

replace(**kwargs)

Return a new object replacing specified fields with new values.

to_dict()

Convert an object into dictionary recursively.

validate([prefix])

Check the validity of configclass object.

Attributes:

__init__(id=<factory>, provider=<factory>, config=<factory>)#
config: dict[str, Any]#
copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

id: str#
provider: str#
replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:
  • obj (object) – The object to check.

  • prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain_benchmarks.motion_generation.config.SuiteCfg[source]#

Resolved benchmark suite configuration.

Methods:

__init__([schema_version, name, ...])

copy(**kwargs)

Return a new object replacing specified fields with new values.

enabled_tracks()

Return enabled tracks in suite order.

evaluation_tracks()

Expand multi-domain tracks into deterministic evaluation tracks.

from_dict(data)

Build and validate a suite from a YAML-compatible mapping.

replace(**kwargs)

Return a new object replacing specified fields with new values.

to_dict()

Convert an object into dictionary recursively.

validate([prefix])

Check the validity of configclass object.

validate_benchmark()

Validate values that affect benchmark correctness.

Attributes:

embodiment

name

planners

profile

protocol

resolved_embodiment_id

Return the stable embodiment identity used in manifests.

robot

schema_version

suite_version

tracks

__init__(schema_version=<factory>, name=<factory>, suite_version=<factory>, profile=<factory>, planners=<factory>, robot=<factory>, embodiment=<factory>, protocol=<factory>, tracks=<factory>)#
copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

embodiment: EmbodimentSpecCfg | None#
enabled_tracks()[source]#

Return enabled tracks in suite order.

Return type:

list[TrackCfg]

evaluation_tracks()[source]#

Expand multi-domain tracks into deterministic evaluation tracks.

Domain expansion happens at the runner boundary. Existing scenario providers therefore keep their stable TrackCfg input while each domain receives a distinct case population and track identity.

Return type:

list[TrackCfg]

classmethod from_dict(data)[source]#

Build and validate a suite from a YAML-compatible mapping.

Return type:

SuiteCfg

name: str#
planners: list[PlannerSpecCfg]#
profile: str#
protocol: ProtocolCfg#
replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

property resolved_embodiment_id: str#

Return the stable embodiment identity used in manifests.

robot: RobotSpecCfg#
schema_version: int#
suite_version: str#
to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

tracks: list[TrackCfg]#
validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:
  • obj (object) – The object to check.

  • prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

validate_benchmark()[source]#

Validate values that affect benchmark correctness.

Return type:

None

class embodichain_benchmarks.motion_generation.config.TrackCfg[source]#

One enabled benchmark track and its scenario provider.

Methods:

__init__([id, scenario, enabled, config, ...])

copy(**kwargs)

Return a new object replacing specified fields with new values.

evaluation_domains()

Return explicit domains while preserving the legacy singular field.

replace(**kwargs)

Return a new object replacing specified fields with new values.

to_dict()

Convert an object into dictionary recursively.

validate([prefix])

Check the validity of configclass object.

Attributes:

__init__(id=<factory>, scenario=<factory>, enabled=<factory>, config=<factory>, domain=<factory>, domains=<factory>, group_id=<factory>)#
config: dict[str, Any]#
copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

domain: DomainCfg | None#
domains: list[DomainCfg]#
enabled: bool#
evaluation_domains()[source]#

Return explicit domains while preserving the legacy singular field.

Return type:

list[DomainCfg]

group_id: str | None#
id: str#
replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2
Parameters:
  • obj (object) – The object to replace.

  • **kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

scenario: str#
to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:

obj (object) – An instance of a class to convert.

Raises:

ValueError – When input argument is not an object.

Return type:

dict[str, Any]

Returns:

Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:
  • obj (object) – The object to check.

  • prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

embodichain_benchmarks.motion_generation.config.resolve_atomic_batch_sizes(track)[source]#

Return simulator batch sizes for an Atomic Skill track.

Explicit batch_sizes remains authoritative. When omitted, an enabled random-pose configuration uses its pose_batch_size as the single batch, allowing one planner call to evaluate all sampled poses.

Return type:

list[int]

embodichain_benchmarks.motion_generation.config.resolve_atomic_pose_randomization(track)[source]#

Resolve shared random-pose controls from an Atomic Skill track.

Accepted spellings are deliberately compatible with the existing skill-level randomization keys. The canonical nested form is:

randomization:
  enabled: true
  mode: spatial_pose
  sampling_pattern: stratified
  pose_batch_size: 8
  object_translation_jitter_m: [0.03, 0.03, 0.0]
  object_yaw_jitter_rad: 0.5236
  object_roll_pitch_jitter_rad: [0.1745, 0.1745]
  target_translation_jitter_m: [0.03, 0.03, 0.02]
  target_yaw_jitter_rad: 0.5236
  target_roll_pitch_jitter_rad: [0.1745, 0.1745]
  support_object_translation_jitter_m: [0.03, 0.03, 0.0]
  support_object_yaw_jitter_rad: 0.5236
  support_object_roll_pitch_jitter_rad: [0.1745, 0.1745]
  task_frame_translation_jitter_m: [0.03, 0.03, 0.0]
  task_frame_yaw_jitter_rad: 0.1745
  task_frame_roll_pitch_jitter_rad: [0.0873, 0.0873]
  seed_stride: 1

random_pose_count is accepted as a readable alias for pose_batch_size. The helper does not mutate the track mapping, so the resolved suite and case manifest remain faithful to the source YAML. workspace_shell assigns ordered normalized XY offset radii to rows; three-dimensional translation vectors use those radii while scalar and rotation ranges retain deterministic stratification.

Return type:

AtomicPoseRandomizationCfg

embodichain_benchmarks.motion_generation.config.stable_hash(value)[source]#

Return a stable SHA256 hash for JSON-compatible configuration data.

Return type:

str

embodichain_benchmarks.motion_generation.config.suite_to_dict(suite)[source]#

Convert a resolved suite to plain YAML/JSON-compatible values.

Return type:

dict[str, Any]

load_suite([name_or_path])

Load an installed suite or an external YAML file.

embodichain_benchmarks.motion_generation.contracts#

Simulator-independent contracts shared by Atomic Skill benchmark owners.

Classes:

CapabilityDecision

Independent capability preflight result for one benchmark population.

EvaluationDomain

Versioned evaluation population; classification does not generate cases.

ExecutionObservation

Measured execution summary shared with physical evaluators.

PhysicalEvaluation

Ordered stage results for one environment, independent of plan success.

PhysicalStageEvaluator

Evaluate measured scenes without access to a compiled/projected plan.

ReplayObservationTrace

Raw synchronized simulator samples for one requested replay window.

ReplayObservationWindow

One compiled-coordinate window requested by a physical evaluator.

StageOutcome

One physical stage; unreached and inapplicable stages are not failures.

Functions:

validate_physical_evaluations(results, *, ...)

Validate batch coverage and stage order, then order results by environment.

class embodichain_benchmarks.motion_generation.contracts.CapabilityDecision[source]#

Independent capability preflight result for one benchmark population.

Methods:

__init__(status[, reason])

Attributes:

__init__(status, reason=None)#
reason: str | None = None#
status: Literal['supported', 'unsupported', 'invalid']#
class embodichain_benchmarks.motion_generation.contracts.EvaluationDomain[source]#

Versioned evaluation population; classification does not generate cases.

Methods:

__init__(id, version, kind[, ...])

Attributes:

__init__(id, version, kind, required_capabilities=frozenset({}))#
id: str#
kind: Literal['nominal', 'robustness', 'held_out']#
required_capabilities: frozenset[str] = frozenset({})#
version: str#
class embodichain_benchmarks.motion_generation.contracts.ExecutionObservation[source]#

Measured execution summary shared with physical evaluators.

Optional values and missing replay channels are unavailable evidence, not successful effects. Provider-declared traces carry raw segment samples; no compiled trajectory or projected task context is carried here.

Methods:

__init__(execution_success, final_tcp_pose, ...)

Attributes:

__init__(execution_success, final_tcp_pose, joint_tracking_rmse_rad, execution_time_ms, task_completion_time_s, object_lift_delta_m=None, final_arm_qpos=None, final_object_pose=None, articulation_joint_initial=None, articulation_joint_final=None, articulation_joint_peak_signed_delta=None, articulation_joint_effect_end=None, object_rotation_peak_signed_rad=None, scene_object_final_poses=<factory>, final_gripper_qpos=<factory>, gripper_open_qpos=<factory>, gripper_grasp_qpos=<factory>, replay_traces=<factory>)#
articulation_joint_effect_end: Tensor | None = None#
articulation_joint_final: Tensor | None = None#
articulation_joint_initial: Tensor | None = None#
articulation_joint_peak_signed_delta: Tensor | None = None#
execution_success: Tensor#
execution_time_ms: float#
final_arm_qpos: Tensor | None = None#
final_gripper_qpos: Mapping[str, Tensor]#
final_object_pose: Tensor | None = None#
final_tcp_pose: Tensor#
gripper_grasp_qpos: Mapping[str, Tensor]#
gripper_open_qpos: Mapping[str, Tensor]#
joint_tracking_rmse_rad: Tensor#
object_lift_delta_m: Tensor | None = None#
object_rotation_peak_signed_rad: Tensor | None = None#
replay_traces: Mapping[str, ReplayObservationTrace]#
scene_object_final_poses: Mapping[str, Tensor]#
task_completion_time_s: float#
class embodichain_benchmarks.motion_generation.contracts.PhysicalEvaluation[source]#

Ordered stage results for one environment, independent of plan success.

Methods:

__init__(env_index, stages[, protocol_version])

Attributes:

env_index

failure_code

Return the measured failure or an explicit incomplete-evaluation code.

failure_stage

Return the first failed or unreached stage, if any.

protocol_version

stages

task_success

Return whether at least one applicable stage exists and all passed.

__init__(env_index, stages, protocol_version=None)#
env_index: int#
property failure_code: str | None#

Return the measured failure or an explicit incomplete-evaluation code.

property failure_stage: str | None#

Return the first failed or unreached stage, if any.

protocol_version: str | None = None#
stages: tuple[StageOutcome, ...]#
property task_success: bool#

Return whether at least one applicable stage exists and all passed.

class embodichain_benchmarks.motion_generation.contracts.PhysicalStageEvaluator[source]#

Evaluate measured scenes without access to a compiled/projected plan.

Methods:

__init__(*args, **kwargs)

evaluate(case, observation, motion_outcomes)

Return physical results for every environment in the case.

__init__(*args, **kwargs)#
evaluate(case, observation, motion_outcomes)[source]#

Return physical results for every environment in the case.

Parameters:
  • case (BenchmarkCase) – Frozen planner-independent task inputs.

  • observation (ExecutionObservation) – Measured replay evidence.

  • motion_outcomes (tuple[CaseOutcome, ...]) – Motion validation outcomes in environment order.

Return type:

tuple[PhysicalEvaluation, ...]

Returns:

One result per environment with a common ordered stage schema.

class embodichain_benchmarks.motion_generation.contracts.ReplayObservationTrace[source]#

Raw synchronized simulator samples for one requested replay window.

Every tensor starts with (sample, environment). Optional channels are absent evidence; gripper tensors additionally retain their joint axis.

Methods:

__init__([tcp_pose, object_pose, ...])

Attributes:

__init__(tcp_pose=None, object_pose=None, articulation_joint=None, gripper_qpos=<factory>, scene_object_poses=<factory>)#
articulation_joint: Tensor | None = None#
gripper_qpos: Mapping[str, Tensor]#
object_pose: Tensor | None = None#
scene_object_poses: Mapping[str, Tensor]#
tcp_pose: Tensor | None = None#
class embodichain_benchmarks.motion_generation.contracts.ReplayObservationWindow[source]#

One compiled-coordinate window requested by a physical evaluator.

start and stop are half-open waypoint indices. A terminal-settle request uses start == stop at the trajectory boundary together with include_settle=True. Providers resolve semantic segment names while they still have the compiled trajectory; evaluators receive only the measured trace identified by observation_id.

Methods:

__init__(observation_id, start, stop, channels)

Attributes:

__init__(observation_id, start, stop, channels, include_settle=False, scene_object_ids=())#
channels: frozenset[str]#
include_settle: bool = False#
observation_id: str#
scene_object_ids: tuple[str, ...] = ()#
start: int#
stop: int#
class embodichain_benchmarks.motion_generation.contracts.StageOutcome[source]#

One physical stage; unreached and inapplicable stages are not failures.

Measurements retain numeric values and units in their keys. Evidence must come from execution observations, never projected planning effects.

Methods:

__init__(stage_id, status[, failure_code, ...])

Attributes:

__init__(stage_id, status, failure_code=None, measurements=<factory>, evidence_source=None, observation_window=None)#
evidence_source: str | None = None#
failure_code: str | None = None#
measurements: dict[str, float | bool]#
observation_window: str | None = None#
stage_id: str#
status: Literal['passed', 'failed', 'not_reached', 'not_applicable', 'unavailable']#
embodichain_benchmarks.motion_generation.contracts.validate_physical_evaluations(results, *, batch_size)[source]#

Validate batch coverage and stage order, then order results by environment.

Parameters:
  • results (tuple[PhysicalEvaluation, ...]) – Evaluator outputs in any environment order.

  • batch_size (int) – Number of environments that must be represented exactly once.

Return type:

tuple[PhysicalEvaluation, ...]

Returns:

Results ordered by environment index.

Raises:

ValueError – Rows are missing/duplicated or stage schemas differ.

embodichain_benchmarks.motion_generation.embodiments#

Resolve official embodiment components for benchmark metadata and runtime.

Classes:

EmbodimentResolution

Resolved embodiment metadata used by cases and benchmark artifacts.

YamlEmbodimentProvider

Resolve official embodiment YAML metadata for benchmark preflight.

Functions:

check_embodiment_capabilities(resolution, ...)

Check a skill's declared capabilities against one embodiment profile.

resolve_embodiment(spec, robot, *[, base_dir])

Resolve metadata from an embodiment component or a registered robot provider.

class embodichain_benchmarks.motion_generation.embodiments.EmbodimentResolution[source]#

Resolved embodiment metadata used by cases and benchmark artifacts.

Methods:

__init__(embodiment_id, component_path, ...)

to_metadata()

Return JSON-safe resolved embodiment metadata.

Attributes:

__init__(embodiment_id, component_path, config_hash, capabilities, endpoint_bindings, runtime_services, source, simulation=None, skill_profile=None)#
capabilities: frozenset[str]#
component_path: Path | None#
config_hash: str#
embodiment_id: str#
endpoint_bindings: Mapping[str, Mapping[str, str]]#
runtime_services: Mapping[str, object]#
simulation: Mapping[str, object] | None#
skill_profile: Mapping[str, object] | None#
source: str#
to_metadata()[source]#

Return JSON-safe resolved embodiment metadata.

Return type:

dict[str, object]

class embodichain_benchmarks.motion_generation.embodiments.YamlEmbodimentProvider[source]#

Resolve official embodiment YAML metadata for benchmark preflight.

Methods:

resolve(spec, *[, base_dir])

Load, validate, and hash one embodiment component.

resolve(spec, *, base_dir=None)[source]#

Load, validate, and hash one embodiment component.

Return type:

EmbodimentResolution

embodichain_benchmarks.motion_generation.embodiments.check_embodiment_capabilities(resolution, required_capabilities)[source]#

Check a skill’s declared capabilities against one embodiment profile.

Return type:

CapabilityDecision

embodichain_benchmarks.motion_generation.embodiments.resolve_embodiment(spec, robot, *, base_dir=None)[source]#

Resolve metadata from an embodiment component or a registered robot provider.

Return type:

EmbodimentResolution

embodichain_benchmarks.motion_generation.evaluation#

Versioned Atomic Skill criteria applied to measured, row-local evidence.

This module never reads a planner, simulator, or projected symbolic context. Observation adapters own contact interpretation, geometry and segment sampling.

Classes:

Criterion

One finite measured quantity and its absolute acceptance bound.

StageCriterion

Required observation window, measurements and physical failure reason.

StageEvidence

Measured evidence for one stage of one environment.

Functions:

evaluate_skill_stage_batch(skill_id, ...)

Evaluate row-local stage evidence in stable environment order.

evaluate_skill_stages(skill_id, evidence, *, ...)

Evaluate the declared stage ladder using measured evidence only.

validate_standard_evaluation(skill_id, result)

Reject a claimed v1 result inconsistent with the executable criteria.

class embodichain_benchmarks.motion_generation.evaluation.Criterion[source]#

One finite measured quantity and its absolute acceptance bound.

Methods:

__init__(measurement, operator[, threshold])

accepts(value)

Check a measurement without interpreting missing evidence as zero.

Attributes:

__init__(measurement, operator, threshold=0.0)#
accepts(value)[source]#

Check a measurement without interpreting missing evidence as zero.

Parameters:

value (float | bool) – Measured scalar in the unit encoded in the measurement name.

Return type:

bool

Returns:

Whether the scalar meets this criterion.

measurement: str#
operator: Literal['le', 'ge', 'gt', 'true']#
threshold: float = 0.0#
class embodichain_benchmarks.motion_generation.evaluation.StageCriterion[source]#

Required observation window, measurements and physical failure reason.

Methods:

__init__(stage_id, window, criteria, ...)

Attributes:

__init__(stage_id, window, criteria, failure_code)#
criteria: tuple[Criterion, ...]#
failure_code: str#
stage_id: str#
window: str#
class embodichain_benchmarks.motion_generation.evaluation.StageEvidence[source]#

Measured evidence for one stage of one environment.

A proxy must name its frozen measurement method; it cannot be a projected held-object state. Sample times and raw traces belong in the case artifacts.

Methods:

__init__(window, source, measurements, method)

Attributes:

__init__(window, source, measurements, method)#
measurements: Mapping[str, float | bool]#
method: str#
source: Literal['simulation_state', 'contact_sensor', 'measured_proxy']#
window: str#
embodichain_benchmarks.motion_generation.evaluation.evaluate_skill_stage_batch(skill_id, evidence_by_env)[source]#

Evaluate row-local stage evidence in stable environment order.

This is the bridge between batched replay observations and the standard’s deliberately row-local evaluator. Observation adapters remain responsible for sampling physical state and naming the measurement method; this helper only assigns environment indices and applies the frozen criteria.

Parameters:
  • skill_id (str) – Canonical snake-case Atomic Skill identifier.

  • evidence_by_env (Sequence[Mapping[str, StageEvidence]]) – One stage-evidence mapping per environment row.

Return type:

tuple[PhysicalEvaluation, ...]

Returns:

Physical evaluations ordered by environment index.

embodichain_benchmarks.motion_generation.evaluation.evaluate_skill_stages(skill_id, evidence, *, env_index)[source]#

Evaluate the declared stage ladder using measured evidence only.

Parameters:
  • skill_id (str) – Canonical snake-case Atomic Skill identifier.

  • evidence (Mapping[str, StageEvidence]) – Row-local observations indexed by stage id, sampled by the host.

  • env_index (int) – Environment row associated with every supplied observation.

Return type:

PhysicalEvaluation

Returns:

Versioned results; missing evidence stops the ladder as unavailable.

Raises:

ValueError – Skill, stage identity or observation window is incorrect.

embodichain_benchmarks.motion_generation.evaluation.validate_standard_evaluation(skill_id, result)[source]#

Reject a claimed v1 result inconsistent with the executable criteria.

Parameters:
  • skill_id (str) – Canonical skill identifier for the frozen case.

  • result (PhysicalEvaluation) – Evaluator output, possibly from a legacy custom evaluator.

Raises:

ValueError – Version, ladder, provenance, measurements or verdict disagree.

Return type:

None

embodichain_benchmarks.motion_generation.generalization#

Domain-aware aggregation for configuration-driven benchmark suites.

Functions:

aggregate_generalization(records, metadata, ...)

Aggregate versioned nominal, robustness, and held-out populations.

embodichain_benchmarks.motion_generation.generalization.aggregate_generalization(records, metadata, cases, measured_trials)[source]#

Aggregate versioned nominal, robustness, and held-out populations.

Unsupported capability cases lower coverage and are excluded from the success denominator. Cases that bypass the selected planner are excluded from planner attribution and reported under their effective motion algorithm. Runtime errors and ordinary planner failures remain applicable cases and therefore count as failures. The returned rows are sufficient to recompute the domain summary from raw trial records.

Generalization eligibility requires complete nominal and non-nominal populations for the same track and attributed algorithm. Retention is unavailable until the nominal baseline is complete and nonzero.

Parameters:
  • records (Iterable[TrialRecord]) – Raw lifecycle and measured trial records.

  • metadata (Iterable[PlannerMetadata]) – Planner metadata for every algorithm in the run.

  • cases (Iterable[BenchmarkCase]) – Planner-independent frozen case manifest.

  • measured_trials (int) – Number of measured repeats expected per case.

Return type:

list[dict[str, object]]

Returns:

One row per attributed planner or effective algorithm and evaluation domain, including generalization summary statistics shared by that algorithm’s non-nominal domains.

embodichain_benchmarks.motion_generation.limits#

Shared benchmark motion budgets for planning and independent validation.

Functions:

resolve_dynamic_limit(limit, dof, reference)

Expand a finite positive suite budget to the trajectory joint order.

resolve_robot_velocity_limits(robot, ...[, dof])

Resolve effective robot speeds, including mimic caps and batch minima.

resolve_velocity_limits(asset_limits, ...)

Intersect valid asset caps with suite requests, filling placeholders.

embodichain_benchmarks.motion_generation.limits.resolve_dynamic_limit(limit, dof, reference)[source]#

Expand a finite positive suite budget to the trajectory joint order.

Return type:

Tensor | None

embodichain_benchmarks.motion_generation.limits.resolve_robot_velocity_limits(robot, control_part, stated_limit, reference, *, dof=None)[source]#

Resolve effective robot speeds, including mimic caps and batch minima.

Real robots expose state-order metadata, allowing follower speed limits to constrain their leaders before selecting the requested control-part order. Lightweight callers without that metadata may provide already-effective control-part limits via get_qvel_limits(name=...).

Return type:

Tensor | None

embodichain_benchmarks.motion_generation.limits.resolve_velocity_limits(asset_limits, stated_limit, reference)[source]#

Intersect valid asset caps with suite requests, filling placeholders.

Zero is a stationary-joint cap. Unknown, negative, nonfinite and implausibly large asset values require a suite budget; they never loosen a known cap. Batched inputs resolve to one conservative vector shared by all rows.

Return type:

Tensor | None

embodichain_benchmarks.motion_generation.metrics#

Metric evaluators for motion-generation benchmarks.

TimedCall

Result and resource deltas captured around one callable.

compute_case_outcomes(result, case, robot, ...)

Recompute free-space success and quality from a planner trajectory.

compute_waypoint_errors(trajectory_poses, ...)

Return ordered, same-sample waypoint errors for one trajectory.

get_pose_err(matrix_a, matrix_b, *[, ...])

Return translation (m) and geodesic rotation (rad) pose errors.

make_failure_outcomes(batch_size, ...[, ...])

Create per-env outcomes when validation cannot produce a trajectory.

match_ordered_waypoints(trajectory_poses, ...)

Evaluate ordered arrival and threshold-constrained waypoint errors.

timed_call(callable_fn)

Time only callable_fn and capture CPU/GPU memory deltas.

embodichain_benchmarks.motion_generation.metrics.performance#

Wall-clock and process-memory measurement helpers.

Classes:

TimedCall

Result and resource deltas captured around one callable.

Functions:

timed_call(callable_fn)

Time only callable_fn and capture CPU/GPU memory deltas.

class embodichain_benchmarks.motion_generation.metrics.performance.TimedCall[source]#

Result and resource deltas captured around one callable.

Methods:

__init__(result, cost_time_ms, cpu_delta_mb, ...)

Attributes:

__init__(result, cost_time_ms, cpu_delta_mb, gpu_delta_mb, peak_gpu_mb)#
cost_time_ms: float#
cpu_delta_mb: float#
gpu_delta_mb: float#
peak_gpu_mb: float | None#
result: TypeVar(_T)#
embodichain_benchmarks.motion_generation.metrics.performance.timed_call(callable_fn)[source]#

Time only callable_fn and capture CPU/GPU memory deltas.

Return type:

TimedCall[TypeVar(_T)]

embodichain_benchmarks.motion_generation.metrics.stats#

Shared numeric helpers for benchmark metric aggregation.

Functions:

nearest_rank_percentile(values, percentile)

Return a nearest-rank percentile over finite values, or None if empty.

embodichain_benchmarks.motion_generation.metrics.stats.nearest_rank_percentile(values, percentile)[source]#

Return a nearest-rank percentile over finite values, or None if empty.

Return type:

float | None

embodichain_benchmarks.motion_generation.metrics.trajectory#

Ordered waypoint and free-space trajectory validity metrics.

Functions:

compute_case_outcomes(result, case, robot, ...)

Recompute free-space success and quality from a planner trajectory.

compute_waypoint_errors(trajectory_poses, ...)

Return ordered, same-sample waypoint errors for one trajectory.

get_pose_err(matrix_a, matrix_b, *[, ...])

Return translation (m) and geodesic rotation (rad) pose errors.

make_failure_outcomes(batch_size, ...[, ...])

Create per-env outcomes when validation cannot produce a trajectory.

match_ordered_joint_waypoints(...)

Evaluate ordered joint-waypoint arrival with an L-infinity threshold.

match_ordered_waypoints(trajectory_poses, ...)

Evaluate ordered arrival and threshold-constrained waypoint errors.

embodichain_benchmarks.motion_generation.metrics.trajectory.compute_case_outcomes(result, case, robot, control_part, *, validation_samples, position_threshold_m, rotation_threshold_rad, joint_limit_tolerance_rad, start_state_tolerance_rad=0.001, joint_velocity_limit_rad_s=None, joint_acceleration_limit_rad_s2=None, joint_jerk_limit_rad_s3=None, dynamic_limit_tolerance=0.001, timing_is_solved=True)[source]#

Recompute free-space success and quality from a planner trajectory.

Return type:

tuple[CaseOutcome, ...]

embodichain_benchmarks.motion_generation.metrics.trajectory.compute_waypoint_errors(trajectory_poses, waypoints, *, position_threshold_m=0.01, rotation_threshold_rad=0.1, rotation_symmetry=None)[source]#

Return ordered, same-sample waypoint errors for one trajectory.

Return type:

dict[str, float]

embodichain_benchmarks.motion_generation.metrics.trajectory.get_pose_err(matrix_a, matrix_b, *, rotation_symmetry=None)[source]#

Return translation (m) and geodesic rotation (rad) pose errors.

Return type:

tuple[float, float]

embodichain_benchmarks.motion_generation.metrics.trajectory.make_failure_outcomes(batch_size, failure_code, *, planner_failure_code=None, planning_success=False, timing_is_solved=None)[source]#

Create per-env outcomes when validation cannot produce a trajectory.

Return type:

tuple[CaseOutcome, ...]

embodichain_benchmarks.motion_generation.metrics.trajectory.match_ordered_joint_waypoints(trajectory_qpos, waypoints_qpos, *, joint_threshold_rad)[source]#

Evaluate ordered joint-waypoint arrival with an L-infinity threshold.

Each waypoint must be reached after the previous arrival by one trajectory sample whose maximum absolute per-joint error is within the task threshold. The function mirrors match_ordered_waypoints() while retaining joint task semantics instead of projecting the targets through FK.

Return type:

dict[str, object]

embodichain_benchmarks.motion_generation.metrics.trajectory.match_ordered_waypoints(trajectory_poses, waypoints, *, position_threshold_m, rotation_threshold_rad, rotation_symmetry=None)[source]#

Evaluate ordered arrival and threshold-constrained waypoint errors.

Success and continuous waypoint errors share the same greedy matching: each waypoint must be hit after the previous arrival by a sample that jointly satisfies the position and rotation thresholds. Reported errors are taken at those arrival samples so motion_valid cannot disagree with waypoint p95/max exceeding the external thresholds.

Return type:

dict[str, object]

resolve_dynamic_limit(limit, dof, reference)

Expand a finite positive suite budget to the trajectory joint order.

resolve_velocity_limits(asset_limits, ...)

Intersect valid asset caps with suite requests, filling placeholders.

embodichain_benchmarks.motion_generation.models#

Typed records shared by the motion-generation benchmark modules.

Classes:

AlgorithmRole

Role an algorithm plays in benchmark comparisons.

BenchmarkCase

One env-batched planner input frozen before execution.

CaseOutcome

External validation result for one environment row in a batch.

PlannerMetadata

Stable planner identity and capability metadata.

TrialPhase

Lifecycle phase represented by a raw trial record.

TrialRecord

Raw lifecycle or planning record written to trials.jsonl.

Functions:

with_physical_evaluation(outcome, ...)

Attach physical results without gating them on motion diagnostics.

class embodichain_benchmarks.motion_generation.models.AlgorithmRole[source]#

Role an algorithm plays in benchmark comparisons.

Attributes:

Methods:

__new__(value)

CANDIDATE = 'candidate'#
DIAGNOSTIC_BASELINE = 'diagnostic_baseline'#
PRIMARY_BASELINE = 'primary_baseline'#
__new__(value)#
class embodichain_benchmarks.motion_generation.models.BenchmarkCase[source]#

One env-batched planner input frozen before execution.

The first fields retain the free-space case contract. The trailing fields describe execution/task tracks without forcing planner adapters to know about a particular robot, atomic skill, or object implementation.

Methods:

__init__(suite_version, track, scenario_id, ...)

Attributes:

__init__(suite_version, track, scenario_id, case_id, seed, batch_size, num_waypoints, path_shape, start_state_bin, start_qpos, target_waypoints, reference_qpos, robot_id='franka_panda', skill_id='N/A', object_id=None, task_difficulty=None, primary_success='motion_valid', full_start_qpos=None, case_parameters=<factory>, domain=None, track_group=None, embodiment_id=None, required_capabilities=frozenset({}))#
batch_size: int#
case_id: str#
case_parameters: dict[str, object]#
domain: EvaluationDomain | None = None#
embodiment_id: str | None = None#
full_start_qpos: Tensor | None = None#
num_waypoints: int#
object_id: str | None = None#
path_shape: str#
primary_success: str = 'motion_valid'#
reference_qpos: Tensor#
required_capabilities: frozenset[str] = frozenset({})#
robot_id: str = 'franka_panda'#
scenario_id: str#
seed: int#
skill_id: str = 'N/A'#
start_qpos: Tensor#
start_state_bin: str#
suite_version: str#
target_waypoints: Tensor#
task_difficulty: str | None = None#
track: str#
track_group: str | None = None#
class embodichain_benchmarks.motion_generation.models.CaseOutcome[source]#

External validation result for one environment row in a batch.

Methods:

__init__(env_index, planning_success, ...[, ...])

Attributes:

acceleration_limit_violation

acceleration_utilization

articulation_joint_delta

articulation_joint_effect_end

articulation_joint_final

articulation_joint_initial

articulation_joint_peak_signed_delta

cartesian_path_length_m

completed_waypoint_ratio

constraint_metrics

Constrained-component arrival errors; unused components are absent.

dynamic_limits_satisfied

Whether every applicable dynamic limit holds, or None when none apply.

dynamic_measurement_status

Whether trajectory dynamics were finite, unbounded, or unavailable.

end_to_end_success

Return overall success; unavailable physical evaluation stays unknown.

env_index

evaluation_protocol

executed_final_rotation_err_deg

executed_final_translation_err_mm

execution_success

failure_code

failure_stage

final_rotation_err_deg

final_translation_err_mm

finite

geometry_valid

Derived path validity, also accepted when reading serialized outcomes.

jerk_limit_violation

jerk_utilization

joint_limit_violation

joint_path_length_rad

joint_tracking_rmse_rad

max_joint_acceleration_rad_s2

max_joint_jerk_rad_s3

max_joint_velocity_rad_s

max_normalized_joint_violation

min_rotation_err_deg

min_translation_err_mm

motion_valid

object_lift_delta_m

object_rotation_peak_signed_rad

ordered_waypoints_reached

path_efficiency

physical_failure_code

physical_skill_success

planner_failure_code

planning_success

reference_path_deviation_mm_max

Sampled translation cross-track error to the requested waypoint polyline.

reference_path_deviation_mm_mean

replan_count

stages

start_joint_error_rad

start_state_valid

Whether the first native sample matches the requested joint state.

task_completion_time_s

task_success

timed_motion_valid

Validity under solved timing; nominal or unknown clocks remain unavailable.

trajectory_moved

velocity_limit_violation

velocity_utilization

waypoint_min_rotation_err_deg

waypoint_min_rotation_err_deg_at_position

waypoint_min_translation_err_mm

waypoint_min_translation_err_mm_at_orientation

waypoint_rotation_err_deg_max

waypoint_rotation_err_deg_mean

waypoint_rotation_err_deg_p95

waypoint_translation_err_mm_max

waypoint_translation_err_mm_mean

waypoint_translation_err_mm_p95

__init__(env_index, planning_success, finite, ordered_waypoints_reached, motion_valid, completed_waypoint_ratio, final_translation_err_mm, final_rotation_err_deg, waypoint_translation_err_mm_mean, waypoint_translation_err_mm_p95, waypoint_translation_err_mm_max, waypoint_rotation_err_deg_mean, waypoint_rotation_err_deg_p95, waypoint_rotation_err_deg_max, joint_limit_violation, max_normalized_joint_violation, joint_path_length_rad, cartesian_path_length_m, path_efficiency, failure_code=None, planner_failure_code=None, execution_success=None, task_success=None, task_completion_time_s=None, joint_tracking_rmse_rad=None, object_lift_delta_m=None, articulation_joint_initial=None, articulation_joint_final=None, articulation_joint_delta=None, articulation_joint_peak_signed_delta=None, articulation_joint_effect_end=None, object_rotation_peak_signed_rad=None, replan_count=None, min_translation_err_mm=None, min_rotation_err_deg=None, trajectory_moved=None, waypoint_min_translation_err_mm=(), waypoint_min_rotation_err_deg=(), waypoint_min_rotation_err_deg_at_position=(), waypoint_min_translation_err_mm_at_orientation=(), executed_final_translation_err_mm=None, executed_final_rotation_err_deg=None, max_joint_velocity_rad_s=None, velocity_utilization=None, velocity_limit_violation=None, max_joint_acceleration_rad_s2=None, acceleration_utilization=None, acceleration_limit_violation=None, max_joint_jerk_rad_s3=None, jerk_utilization=None, jerk_limit_violation=None, dynamic_measurement_status=None, dynamic_limits_satisfied=None, stages=(), failure_stage=None, physical_skill_success=None, physical_failure_code=None, evaluation_protocol=None, timed_motion_valid=None, constraint_metrics=<factory>, geometry_valid=False, start_state_valid=True, start_joint_error_rad=None, reference_path_deviation_mm_mean=None, reference_path_deviation_mm_max=None)#
acceleration_limit_violation: bool | None = None#
acceleration_utilization: float | None = None#
articulation_joint_delta: float | None = None#
articulation_joint_effect_end: float | None = None#
articulation_joint_final: float | None = None#
articulation_joint_initial: float | None = None#
articulation_joint_peak_signed_delta: float | None = None#
cartesian_path_length_m: float | None#
completed_waypoint_ratio: float#
constraint_metrics: dict[str, object]#

Constrained-component arrival errors; unused components are absent.

dynamic_limits_satisfied: bool | None = None#

Whether every applicable dynamic limit holds, or None when none apply.

Acceleration and jerk limits have no asset source, so a suite that leaves them unset yields None rather than a satisfied result.

dynamic_measurement_status: str | None = None#

Whether trajectory dynamics were finite, unbounded, or unavailable.

property end_to_end_success: bool | None#

Return overall success; unavailable physical evaluation stays unknown.

env_index: int#
evaluation_protocol: str | None = None#
executed_final_rotation_err_deg: float | None = None#
executed_final_translation_err_mm: float | None = None#
execution_success: bool | None = None#
failure_code: str | None = None#
failure_stage: str | None = None#
final_rotation_err_deg: float | None#
final_translation_err_mm: float | None#
finite: bool#
geometry_valid: bool = False#

Derived path validity, also accepted when reading serialized outcomes.

jerk_limit_violation: bool | None = None#
jerk_utilization: float | None = None#
joint_limit_violation: bool#
joint_path_length_rad: float | None#
joint_tracking_rmse_rad: float | None = None#
max_joint_acceleration_rad_s2: float | None = None#
max_joint_jerk_rad_s3: float | None = None#
max_joint_velocity_rad_s: float | None = None#
max_normalized_joint_violation: float | None#
min_rotation_err_deg: float | None = None#
min_translation_err_mm: float | None = None#
motion_valid: bool#
object_lift_delta_m: float | None = None#
object_rotation_peak_signed_rad: float | None = None#
ordered_waypoints_reached: bool#
path_efficiency: float | None#
physical_failure_code: str | None = None#
physical_skill_success: bool | None = None#
planner_failure_code: str | None = None#
planning_success: bool#
reference_path_deviation_mm_max: float | None = None#

Sampled translation cross-track error to the requested waypoint polyline.

Diagnostic only; ordered arrival remains the validity contract. Joint goals and orientation-only requests have no translation reference-path metric.

reference_path_deviation_mm_mean: float | None = None#
replan_count: int | None = None#
stages: tuple[StageOutcome, ...] = ()#
start_joint_error_rad: float | None = None#
start_state_valid: bool = True#

Whether the first native sample matches the requested joint state.

task_completion_time_s: float | None = None#
task_success: bool | None = None#
timed_motion_valid: bool | None = None#

Validity under solved timing; nominal or unknown clocks remain unavailable.

trajectory_moved: bool | None = None#
velocity_limit_violation: bool | None = None#
velocity_utilization: float | None = None#
waypoint_min_rotation_err_deg: tuple[float, ...] = ()#
waypoint_min_rotation_err_deg_at_position: tuple[float | None, ...] = ()#
waypoint_min_translation_err_mm: tuple[float, ...] = ()#
waypoint_min_translation_err_mm_at_orientation: tuple[float | None, ...] = ()#
waypoint_rotation_err_deg_max: float | None#
waypoint_rotation_err_deg_mean: float | None#
waypoint_rotation_err_deg_p95: float | None#
waypoint_translation_err_mm_max: float | None#
waypoint_translation_err_mm_mean: float | None#
waypoint_translation_err_mm_p95: float | None#
class embodichain_benchmarks.motion_generation.models.PlannerMetadata[source]#

Stable planner identity and capability metadata.

Methods:

__init__(algorithm_id, algorithm_role, ...)

Attributes:

__init__(algorithm_id, algorithm_role, adapter, config_hash, capabilities, model_revision='N/A', inference_dtype='fp32', timing_is_solved=False, supported_robots=('franka_panda', ), parameters=<factory>)#
adapter: str#
algorithm_id: str#
algorithm_role: AlgorithmRole#
capabilities: frozenset[str]#
config_hash: str#
inference_dtype: str = 'fp32'#
model_revision: str = 'N/A'#
parameters: dict[str, object]#
supported_robots: tuple[str, ...] = ('franka_panda',)#
timing_is_solved: bool = False#

Whether reported dt is solved rather than a nominal constant.

PlanResult requires dt whenever it carries positions, so a planner with no time parameterization still reports intervals. Recording the distinction here keeps solved-timing comparisons from treating a nominal constant as a solved duration; report such rows as N/A.

class embodichain_benchmarks.motion_generation.models.TrialPhase[source]#

Lifecycle phase represented by a raw trial record.

Attributes:

Methods:

__new__(value)

AVAILABILITY = 'availability'#
COLD = 'cold'#
CONSTRUCT = 'construct'#
MEASURED = 'measured'#
PREPARE = 'prepare'#
WARMUP = 'warmup'#
__new__(value)#
class embodichain_benchmarks.motion_generation.models.TrialRecord[source]#

Raw lifecycle or planning record written to trials.jsonl.

Methods:

__init__(suite_version, track, scenario_id, ...)

to_dict()

Return a JSON-serializable mapping while retaining numeric values.

Attributes:

__init__(suite_version, track, scenario_id, case_id, algorithm_id, algorithm_role, model_revision, planner_config_hash, seed, repeat, batch_size, waypoint_count, path_shape, start_state_bin, phase, status='ok', failure_code=None, failure_message=None, cost_time_ms=None, cpu_delta_mb=None, gpu_delta_mb=None, peak_gpu_mb=None, robot_id='franka_panda', skill_id='N/A', object_id=None, task_difficulty=None, primary_success='motion_valid', execution_time_ms=None, end_to_end_time_ms=None, trajectory_duration_s=None, trajectory_waypoints=None, metadata=<factory>, outcomes=(), domain=None, track_group=None, embodiment_id=None)#
algorithm_id: str#
algorithm_role: AlgorithmRole#
batch_size: int#
case_id: str#
cost_time_ms: float | None = None#
cpu_delta_mb: float | None = None#
domain: EvaluationDomain | None = None#
property effective_motion_algorithm: str#

Return the algorithm that actually produced this trial’s motion.

Planner-attributed trials retain their configured planner identity. Scenario-defined motion uses its explicit metadata identity, with a stable fallback for malformed or older non-planner artifacts.

embodiment_id: str | None = None#
end_to_end_time_ms: float | None = None#
execution_time_ms: float | None = None#
failure_code: str | None = None#
failure_message: str | None = None#
gpu_delta_mb: float | None = None#
metadata: dict[str, object]#
model_revision: str#
object_id: str | None = None#
outcomes: tuple[CaseOutcome, ...] = ()#
path_shape: str#
peak_gpu_mb: float | None = None#
phase: TrialPhase#
planner_config_hash: str#
primary_success: str = 'motion_valid'#
repeat: int#
robot_id: str = 'franka_panda'#
scenario_id: str#
seed: int#
skill_id: str = 'N/A'#
start_state_bin: str#
status: str = 'ok'#
suite_version: str#
task_difficulty: str | None = None#
to_dict()[source]#

Return a JSON-serializable mapping while retaining numeric values.

Return type:

dict[str, object]

track: str#
track_group: str | None = None#
trajectory_duration_s: float | None = None#
trajectory_waypoints: int | None = None#
property uses_selected_planner_backend: bool#

Return whether this trial exercised its attributed planner backend.

Older artifacts predate explicit motion attribution, so a missing flag retains their original planner-attributed behavior.

waypoint_count: int#
embodichain_benchmarks.motion_generation.models.with_physical_evaluation(outcome, evaluation, *, skill_id)[source]#

Attach physical results without gating them on motion diagnostics.

Parameters:
  • outcome (CaseOutcome) – Motion and controller diagnostics for the same environment.

  • evaluation (PhysicalEvaluation) – Measured physical stages for that environment.

  • skill_id (str) – Canonical skill id used to validate a versioned result.

Return type:

CaseOutcome

Returns:

Outcome retaining independent physical success and the legacy overall alias.

Raises:

ValueError – Environment rows or claimed standard results disagree.

embodichain_benchmarks.motion_generation.planners#

Built-in planner adapters and their registry side effects.

CuroboAdapter

Run cuRobo with a frozen, empty-world operational configuration.

IkInterpolateAdapter

Use the robot IK solver followed by fixed-count joint interpolation.

NmgOnnxAdapter

Run a standalone NMG ONNX policy through NeuralPlanner.

PlannerAdapter

Uniform lifecycle around a motion-planning implementation.

PlannerContext

Runtime objects shared with a planner adapter.

ToppraAdapter

Pre-interpolate EEF targets through IK, then run TOPPRA.

embodichain_benchmarks.motion_generation.planners.base#

Backend-independent planner adapter contract.

Classes:

PlannerContext

Runtime objects shared with a planner adapter.

class embodichain_benchmarks.motion_generation.planners.base.PlannerContext[source]#

Runtime objects shared with a planner adapter.

Methods:

__init__(robot, control_part, device, ...[, ...])

Attributes:

control_part

device

joint_acceleration_limit_rad_s2

The suite's acceleration limit, or None when it states none.

joint_jerk_limit_rad_s3

The suite's jerk limit, or None when it states none.

joint_velocity_limit_rad_s

Suite speed budget intersected with effective Robot caps.

robot

robot_id

sample_interval

__init__(robot, control_part, device, sample_interval, robot_id='unknown', joint_velocity_limit_rad_s=None, joint_acceleration_limit_rad_s2=None, joint_jerk_limit_rad_s3=None)#
control_part: str#
device: device#
joint_acceleration_limit_rad_s2: float | Sequence[float] | None = None#

The suite’s acceleration limit, or None when it states none.

An adapter that constrains its own output must use the same limits the metrics check it against, or a satisfied result says nothing. Velocity limits come from robot.get_qvel_limits(), which the metrics also use; acceleration has no asset source and is suite-owned.

joint_jerk_limit_rad_s3: float | Sequence[float] | None = None#

The suite’s jerk limit, or None when it states none.

joint_velocity_limit_rad_s: float | Sequence[float] | None = None#

Suite speed budget intersected with effective Robot caps.

robot: Robot#
robot_id: str = 'unknown'#
sample_interval: int#

PlannerAdapter

Uniform lifecycle around a motion-planning implementation.

embodichain_benchmarks.motion_generation.planners.curobo#

cuRobo primary-baseline adapter for an empty collision world.

Classes:

CuroboAdapter

Run cuRobo with a frozen, empty-world operational configuration.

class embodichain_benchmarks.motion_generation.planners.curobo.CuroboAdapter[source]#

Run cuRobo with a frozen, empty-world operational configuration.

Methods:

__init__(spec, context)

availability()

Require both CUDA and the optional cuRobo runtime.

build()

Construct MotionGenerator without materializing its lazy backend.

close()

Destroy cached cuRobo graph and planner resources.

plan(case)

Plan all ordered Cartesian or joint waypoints in one call.

prepare(case)

Materialize and warm the backend required by the benchmark case.

prepare_cases(cases)

Prepare every distinct batch-size and planning-mode backend.

Attributes:

capabilities

model_revision

separate_prepare

Whether this backend exposes a distinct lazy preparation phase.

timing_is_solved

cuRobo optimizes the trajectory's timing under its own dynamic limits.

__init__(spec, context)[source]#
availability()[source]#

Require both CUDA and the optional cuRobo runtime.

Return type:

tuple[bool, str | None]

build()[source]#

Construct MotionGenerator without materializing its lazy backend.

Return type:

None

capabilities: frozenset[str] = frozenset({'atomic_action', 'batched', 'eef_waypoint', 'empty_world', 'joint_waypoint'})#
close()[source]#

Destroy cached cuRobo graph and planner resources.

Return type:

None

model_revision: str = 'curobo-v2'#
motion_generator: MotionGenerator | None#
plan(case)[source]#

Plan all ordered Cartesian or joint waypoints in one call.

Return type:

PlanResult

prepare(case)[source]#

Materialize and warm the backend required by the benchmark case.

Return type:

dict[str, object]

prepare_cases(cases)[source]#

Prepare every distinct batch-size and planning-mode backend.

Return type:

dict[str, object]

separate_prepare: bool = True#

Whether this backend exposes a distinct lazy preparation phase.

timing_is_solved: bool = True#

cuRobo optimizes the trajectory’s timing under its own dynamic limits.

embodichain_benchmarks.motion_generation.planners.ik_interpolate#

Sequential IK plus joint interpolation diagnostic adapter.

Classes:

IkInterpolateAdapter

Use the robot IK solver followed by fixed-count joint interpolation.

class embodichain_benchmarks.motion_generation.planners.ik_interpolate.IkInterpolateAdapter[source]#

Use the robot IK solver followed by fixed-count joint interpolation.

Methods:

build()

Freeze shared limits; no planner object is required.

plan(case)

Solve each waypoint sequentially while retaining per-env failures.

Attributes:

build()[source]#

Freeze shared limits; no planner object is required.

Return type:

None

capabilities: frozenset[str] = frozenset({'batched', 'eef_waypoint', 'empty_world', 'joint_waypoint'})#
plan(case)[source]#

Solve each waypoint sequentially while retaining per-env failures.

Return type:

PlanResult

embodichain_benchmarks.motion_generation.planners.nmg_onnx#

NMG ONNX benchmark adapter.

Classes:

NmgOnnxAdapter

Run a standalone NMG ONNX policy through NeuralPlanner.

class embodichain_benchmarks.motion_generation.planners.nmg_onnx.NmgOnnxAdapter[source]#

Run a standalone NMG ONNX policy through NeuralPlanner.

Methods:

__init__(spec, context)

availability()

Require ONNX Runtime and a resolved standalone policy path.

build()

Construct the ONNX-backed MotionGenerator.

close()

Release the adapter-owned MotionGenerator.

plan(case)

Plan all ordered constraints in one closed-loop rollout.

supports_case(case)

Reject constraints incompatible with policy capacity or TCP mapping.

Attributes:

capabilities

motion_policy_planner

Select EmbodiChain's neural MotionGenerator backend.

retime

Whether this row re-parameterizes the rollout under dynamic limits.

timing_is_solved

Whether reported timing is solved rather than nominal.

__init__(spec, context)[source]#
availability()[source]#

Require ONNX Runtime and a resolved standalone policy path.

Return type:

tuple[bool, str | None]

build()[source]#

Construct the ONNX-backed MotionGenerator.

Return type:

None

capabilities: frozenset[str] = frozenset({'atomic_action', 'batched', 'eef_waypoint', 'empty_world', 'joint_waypoint', 'masked_constraints'})#
close()[source]#

Release the adapter-owned MotionGenerator.

Return type:

None

motion_generator: MotionGenerator | None#
property motion_policy_planner: str#

Select EmbodiChain’s neural MotionGenerator backend.

plan(case)[source]#

Plan all ordered constraints in one closed-loop rollout.

Return type:

PlanResult

property retime: bool#

Whether this row re-parameterizes the rollout under dynamic limits.

supports_case(case)[source]#

Reject constraints incompatible with policy capacity or TCP mapping.

Return type:

tuple[bool, str | None]

property timing_is_solved: bool#

Whether reported timing is solved rather than nominal.

Without retiming the policy integrates a joint delta per step and never solves a duration, so its dt carries no executable timing and the row must be reported as N/A. Retiming replaces that constant with a solved parameterization, which makes the duration comparable.

embodichain_benchmarks.motion_generation.planners.toppra#

IK plus TOPPRA time-parameterization diagnostic adapter.

Classes:

ToppraAdapter

Pre-interpolate EEF targets through IK, then run TOPPRA.

class embodichain_benchmarks.motion_generation.planners.toppra.ToppraAdapter[source]#

Pre-interpolate EEF targets through IK, then run TOPPRA.

TOPPRA is the dual-arm Atomic Skill backend: the coordinated skills reject cuRobo, so the adapter also exposes its MotionGenerator to AtomicActionEngine and plans joint-space cases from their frozen reference configurations.

Methods:

__init__(spec, context)

atomic_plan_options()

Use the suite limits and sample count for Atomic Skill calls.

availability()

The in-tree TOPPRA implementation needs no optional external package.

build()

Construct the TOPPRA MotionGenerator.

close()

Drop the motion generator through the common planner lifecycle.

plan(case)

Plan ordered EEF or joint waypoints through the IK-to-TOPPRA pipeline.

Attributes:

capabilities

timing_is_solved

TOPPRA solves the time parameterization under velocity and acceleration limits.

__init__(spec, context)[source]#
atomic_plan_options()[source]#

Use the suite limits and sample count for Atomic Skill calls.

Return type:

ToppraPlanOptions

availability()[source]#

The in-tree TOPPRA implementation needs no optional external package.

Return type:

tuple[bool, str | None]

build()[source]#

Construct the TOPPRA MotionGenerator.

Return type:

None

capabilities: frozenset[str] = frozenset({'atomic_action', 'batched', 'eef_waypoint', 'empty_world', 'joint_waypoint'})#
close()[source]#

Drop the motion generator through the common planner lifecycle.

Return type:

None

motion_generator: MotionGenerator | None#
plan(case)[source]#

Plan ordered EEF or joint waypoints through the IK-to-TOPPRA pipeline.

Return type:

PlanResult

timing_is_solved: bool = True#

TOPPRA solves the time parameterization under velocity and acceleration limits.

embodichain_benchmarks.motion_generation.registry#

Planner and scenario registries used by the generic benchmark runner.

Functions:

create_planner_adapter(spec, context, *[, ...])

Construct an adapter, preferring the current run's explicit mapping.

create_robot_provider(spec)

Construct the robot provider selected by a suite.

create_scenario_provider(name)

Construct the scenario provider selected by a track specification.

planner_adapter_names()

Return registered adapter names in deterministic order.

register_planner_adapter(name, adapter_cls)

Register one adapter class under a stable configuration name.

register_robot_provider(name, provider_cls)

Register one robot provider under a stable suite name.

register_scenario_provider(name, provider_cls)

Register one scenario provider class under a stable configuration name.

robot_provider_names()

Return registered robot-provider names in deterministic order.

scenario_provider_names()

Return registered scenario provider names in deterministic order.

unregister_planner_adapter(name)

Remove one planner adapter registration when present.

embodichain_benchmarks.motion_generation.registry.create_planner_adapter(spec, context, *, adapters=None)[source]#

Construct an adapter, preferring the current run’s explicit mapping.

Parameters:
  • spec (PlannerSpecCfg) – Algorithm identity and backend configuration.

  • context (PlannerContext) – Robot and shared evaluation budgets.

  • adapters (Mapping[str, type[PlannerAdapter]] | None) – Overrides local to this run. The global registry is unchanged.

Return type:

PlannerAdapter

Returns:

A new adapter owned by the caller.

embodichain_benchmarks.motion_generation.registry.create_robot_provider(spec)[source]#

Construct the robot provider selected by a suite.

Return type:

RobotProvider

embodichain_benchmarks.motion_generation.registry.create_scenario_provider(name)[source]#

Construct the scenario provider selected by a track specification.

Return type:

ScenarioProvider

embodichain_benchmarks.motion_generation.registry.planner_adapter_names()[source]#

Return registered adapter names in deterministic order.

Return type:

tuple[str, ...]

embodichain_benchmarks.motion_generation.registry.register_planner_adapter(name, adapter_cls)[source]#

Register one adapter class under a stable configuration name.

Return type:

None

embodichain_benchmarks.motion_generation.registry.register_robot_provider(name, provider_cls)[source]#

Register one robot provider under a stable suite name.

Return type:

None

embodichain_benchmarks.motion_generation.registry.register_scenario_provider(name, provider_cls)[source]#

Register one scenario provider class under a stable configuration name.

Return type:

None

embodichain_benchmarks.motion_generation.registry.robot_provider_names()[source]#

Return registered robot-provider names in deterministic order.

Return type:

tuple[str, ...]

embodichain_benchmarks.motion_generation.registry.scenario_provider_names()[source]#

Return registered scenario provider names in deterministic order.

Return type:

tuple[str, ...]

embodichain_benchmarks.motion_generation.registry.unregister_planner_adapter(name)[source]#

Remove one planner adapter registration when present.

Return type:

None

embodichain_benchmarks.motion_generation.reporting#

Render planning and skill views with explicit workload and difficulty tables.

Functions:

write_markdown_report(path, suite, aggregates)

Write one report, keeping each scenario family's results and ranks separate.

embodichain_benchmarks.motion_generation.reporting.write_markdown_report(path, suite, aggregates, notes=None)[source]#

Write one report, keeping each scenario family’s results and ranks separate.

Return type:

Path

embodichain_benchmarks.motion_generation.robots#

Built-in robot providers for motion-generation benchmarks.

DualArmProvider

Build the tutorial dual-arm robot: two arms with one gripper each.

FrankaPandaProvider

Build the stock Franka Panda used by the free-space track.

FrankaPgiProvider

Build the benchmark-owned Franka + PGI assembly used by the Atomic Skill suites.

RobotProvider

Build one benchmark embodiment behind a stable suite identifier.

UR5PgiProvider

Build the UR5 with the tutorial PGI gripper for Atomic Skills.

UR5Provider

Build the fixed-base UR5 used by the NMG free-space benchmark.

embodichain_benchmarks.motion_generation.robots.base#

Robot-provider contract for planner benchmark suites.

Classes:

EmbodimentRobotProvider

Build an official robot and bind one explicit motion resource.

RobotProvider

Build one benchmark embodiment behind a stable suite identifier.

class embodichain_benchmarks.motion_generation.robots.base.EmbodimentRobotProvider[source]#

Build an official robot and bind one explicit motion resource.

Parameters:
  • resolution (EmbodimentResolution) – Component configuration and bindings resolved for this run.

  • resource_id (str | None) – Selected resource, required when several provide motion.

Methods:

__init__(resolution[, resource_id])

build_cfg()

Return a fresh component-owned robot configuration, including its TCP.

configure_track(track)

Supply component-owned gripper commands without accepting conflicts.

__init__(resolution, resource_id=None)[source]#
build_cfg()[source]#

Return a fresh component-owned robot configuration, including its TCP.

Return type:

RobotCfg

configure_track(track)[source]#

Supply component-owned gripper commands without accepting conflicts.

Parameters:

track (TrackCfg) – Effective track, including any selected domain overrides.

Return type:

None

class embodichain_benchmarks.motion_generation.robots.base.RobotProvider[source]#

Build one benchmark embodiment behind a stable suite identifier.

Methods:

__init__(spec)

add_robot(simulation)

Add the configured robot to a simulation.

build_cfg()

Build the robot configuration without mutating a simulation.

Attributes:

__init__(spec)[source]#
add_robot(simulation)[source]#

Add the configured robot to a simulation.

Return type:

Robot

abstract build_cfg()[source]#

Build the robot configuration without mutating a simulation.

Return type:

RobotCfg

control_part: str = 'arm'#

embodichain_benchmarks.motion_generation.robots.dual_arm#

Dual-arm robot provider for coordinated Atomic Skill benchmarks.

Classes:

DualArmProvider

Build the tutorial dual-arm robot: two arms with one gripper each.

class embodichain_benchmarks.motion_generation.robots.dual_arm.DualArmProvider[source]#

Build the tutorial dual-arm robot: two arms with one gripper each.

The control parts are left_arm/right_arm and left_eef/right_eef. robot.config selects the arm family and the tutorial construction parameters, so each coordinated skill suite can reuse its tutorial’s validated robot.

Methods:

build_cfg()

Build the configured dual-arm robot.

Attributes:

build_cfg()[source]#

Build the configured dual-arm robot.

Return type:

RobotCfg

control_part: str = 'left_arm'#

embodichain_benchmarks.motion_generation.robots.franka#

Built-in Franka robot providers.

Classes:

FrankaPandaProvider

Build the stock Franka Panda used by the free-space track.

FrankaPgiProvider

Build the benchmark-owned Franka + PGI assembly used by the Atomic Skill suites.

class embodichain_benchmarks.motion_generation.robots.franka.FrankaPandaProvider[source]#

Build the stock Franka Panda used by the free-space track.

Methods:

build_cfg()

Build a stock Panda while applying optional suite overrides.

build_cfg()[source]#

Build a stock Panda while applying optional suite overrides.

Return type:

RobotCfg

class embodichain_benchmarks.motion_generation.robots.franka.FrankaPgiProvider[source]#

Build the benchmark-owned Franka + PGI assembly used by the Atomic Skill suites.

The suites were authored with the base rotated 180 degrees, as in the atomic-action tutorial scenes, so the provider keeps that rotation; suite overrides apply on top.

Methods:

build_cfg()

Build the Franka + PGI benchmark configuration.

build_cfg()[source]#

Build the Franka + PGI benchmark configuration.

Return type:

RobotCfg

embodichain_benchmarks.motion_generation.robots.ur5#

UR5 robot provider for motion-generation benchmarks.

Classes:

UR5PgiProvider

Build the UR5 with the tutorial PGI gripper for Atomic Skills.

UR5Provider

Build the fixed-base UR5 used by the NMG free-space benchmark.

class embodichain_benchmarks.motion_generation.robots.ur5.UR5PgiProvider[source]#

Build the UR5 with the tutorial PGI gripper for Atomic Skills.

Methods:

build_cfg()

Use the tutorial UR5 arm, gripper, and TCP configuration.

build_cfg()[source]#

Use the tutorial UR5 arm, gripper, and TCP configuration.

Return type:

RobotCfg

Returns:

The configured UR5 with PGI gripper.

class embodichain_benchmarks.motion_generation.robots.ur5.UR5Provider[source]#

Build the fixed-base UR5 used by the NMG free-space benchmark.

Methods:

build_cfg()

Build a stock UR5 while applying optional suite overrides.

build_cfg()[source]#

Build a stock UR5 while applying optional suite overrides.

Return type:

RobotCfg

Returns:

The configured UR5 robot.

embodichain_benchmarks.motion_generation.runner#

Generic lifecycle runner for motion-generation benchmark tracks.

Classes:

BenchmarkRunner

Generate fixed cases, execute adapters, aggregate, and report results.

Functions:

resolve_device(requested)

Resolve auto while rejecting an unavailable explicit CUDA request.

class embodichain_benchmarks.motion_generation.runner.BenchmarkRunner[source]#

Generate fixed cases, execute adapters, aggregate, and report results.

Methods:

__init__(suite, planner_specs, *[, device, ...])

run()

Run the suite and write all required artifacts.

__init__(suite, planner_specs, *, device='auto', headless=True, output_root='outputs/benchmarks', video=None, adapters=None, case_manifest=None)[source]#
run()[source]#

Run the suite and write all required artifacts.

Return type:

BenchmarkRunResult

embodichain_benchmarks.motion_generation.runner.resolve_device(requested)[source]#

Resolve auto while rejecting an unavailable explicit CUDA request.

Return type:

device

BenchmarkRunResult

Paths and aggregate data produced by one completed run.

embodichain_benchmarks.motion_generation.scenarios#

Scenario providers for motion-generation benchmarks.

AtomicObjectHandle

Simulation object plus its algorithm-independent frozen state.

AtomicSkillCaseProvider

Generate and ground one Atomic Action without planner-specific logic.

AtomicSkillScenario

Run fixed Atomic Actions through an adapter-owned MotionGenerator.

atomic_object_kind_names()

Return registered object kinds in deterministic order.

atomic_skill_provider_names()

Return registered Atomic Action case providers.

create_atomic_object(simulation, ...[, ...])

Create one config-driven object and freeze its settled initial pose.

create_atomic_skill_provider(skill_id)

Construct one registered Atomic Action case provider.

FreeSpaceScenario

register_atomic_object_kind(name, factory)

Register a config-only object shape factory.

register_atomic_skill_provider(skill_id, ...)

Register one Atomic Action case provider.

ScenarioEvaluation

Outcomes and execution metrics produced outside planner timing.

ScenarioProvider

Generate, plan, execute, and evaluate one registered scenario kind.

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils#

Deterministic case identity, pose randomization, and goal construction.

Functions:

apply_seeded_spatial_rotation(pose, *, ...)

Pre-multiply each pose by a deterministic world-frame rotation.

apply_target_rotation_randomization(pose, ...)

Apply configured goal-relative roll, pitch, and yaw to a target pose.

atomic_case_id(track_id, skill_id, name, *, ...)

Return an Atomic Skill identity that distinguishes simulator batches.

canonical_case_qpos(qpos)

Remove insignificant device-level IK noise from frozen case states.

case_generation_seed(seed, *, skill_index, ...)

Derive one stable RNG seed for all stochastic IK work in a case.

case_name(config)

Return and validate a stable case name.

case_pose(case, name, *, device)

Restore one frozen pose tensor from JSON-compatible case parameters.

difficulty(config)

Resolve the explicit, frozen Atomic Skill difficulty label.

float_vector(value, *, name, length[, default])

Resolve a finite numeric vector from a YAML-compatible value.

held_object_state(scenario, case)

Reconstruct the frozen held-object relation for an isolated case.

randomization_parameters(config, *, seed)

Serialize the deterministic randomization contract into a case manifest.

randomized_vector_batch(base, config, *, ...)

Return one independently jittered vector for every environment row.

seeded_jitter(amplitude, *, seed, stream, ...)

Sample deterministic jitter, optionally one row per environment.

seeded_roll_pitch_rotation(*, batch_size, ...)

Return deterministic Ry(pitch) @ Rx(roll) rotations.

seeded_spatial_rotation(*, batch_size, ...)

Return deterministic world-frame Rz @ Ry @ Rx rotations.

seeded_yaw_rotation(*, batch_size, ...[, ...])

Return deterministic world-Z rotation matrices for a pose batch.

slide_direction(value)

Validate one Slide direction without coercing arbitrary values.

snapshot_case_qpos(robot, control_part)

Capture one canonical full state and its controlled joints.

twist_arc_poses(target_pose, grasp_pose, ...)

Build the same target-local circular keyframes as the Twist primitive.

workspace_shell_profile(batch_size)

Return ordered normalized radii and near/mid/far workspace labels.

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.apply_seeded_spatial_rotation(pose, *, roll_pitch_amplitude, yaw_amplitude, roll_pitch_name, yaw_name, seed, roll_pitch_stream, yaw_stream, seed_stride, sampling_pattern='uniform')[source]#

Pre-multiply each pose by a deterministic world-frame rotation.

Return type:

None

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.apply_target_rotation_randomization(pose, config, *, seed, stream)[source]#

Apply configured goal-relative roll, pitch, and yaw to a target pose.

Return type:

None

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.atomic_case_id(track_id, skill_id, name, *, batch_size, seed)[source]#

Return an Atomic Skill identity that distinguishes simulator batches.

Return type:

str

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.canonical_case_qpos(qpos)[source]#

Remove insignificant device-level IK noise from frozen case states.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.case_generation_seed(seed, *, skill_index, case_index)[source]#

Derive one stable RNG seed for all stochastic IK work in a case.

Return type:

int

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.case_name(config)[source]#

Return and validate a stable case name.

Return type:

str

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.case_pose(case, name, *, device)[source]#

Restore one frozen pose tensor from JSON-compatible case parameters.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.difficulty(config)[source]#

Resolve the explicit, frozen Atomic Skill difficulty label.

Return type:

str

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.float_vector(value, *, name, length, default=None)[source]#

Resolve a finite numeric vector from a YAML-compatible value.

Return type:

list[float]

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.held_object_state(scenario, case)[source]#

Reconstruct the frozen held-object relation for an isolated case.

Return type:

HeldObjectState

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.randomization_parameters(config, *, seed)[source]#

Serialize the deterministic randomization contract into a case manifest.

Return type:

dict[str, object]

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.randomized_vector_batch(base, config, *, jitter_name, seed, stream, batch_size, dtype, device)[source]#

Return one independently jittered vector for every environment row.

Skill-local jitter keys take precedence. A track-level pose-batch translation range from the typed context is used as a fallback, which keeps suite YAML concise.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.seeded_jitter(amplitude, *, seed, stream, dtype, device, batch_size=1, seed_stride=1, sampling_pattern='uniform')[source]#

Sample deterministic jitter, optionally one row per environment.

The legacy batch_size=1 result remains a one-dimensional vector. For a pose-batch benchmark the returned tensor has shape (B, D). Sampling is stable across devices and independent of planner execution order.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.seeded_roll_pitch_rotation(*, batch_size, amplitude, name, seed, stream, seed_stride, dtype, device, sampling_pattern='uniform')[source]#

Return deterministic Ry(pitch) @ Rx(roll) rotations.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.seeded_spatial_rotation(*, batch_size, roll_pitch_amplitude, yaw_amplitude, roll_pitch_name, yaw_name, seed, roll_pitch_stream, yaw_stream, seed_stride, dtype, device, sampling_pattern='uniform')[source]#

Return deterministic world-frame Rz @ Ry @ Rx rotations.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.seeded_yaw_rotation(*, batch_size, amplitude, name, seed, stream, seed_stride, dtype, device, sampling_pattern='uniform')[source]#

Return deterministic world-Z rotation matrices for a pose batch.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.slide_direction(value)[source]#

Validate one Slide direction without coercing arbitrary values.

Return type:

Literal['pull', 'push']

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.snapshot_case_qpos(robot, control_part)[source]#

Capture one canonical full state and its controlled joints.

Return type:

tuple[Tensor, Tensor]

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.twist_arc_poses(target_pose, grasp_pose, twist_axis, axis_origin, twist_angle, waypoint_count)[source]#

Build the same target-local circular keyframes as the Twist primitive.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_case_utils.workspace_shell_profile(batch_size)[source]#

Return ordered normalized radii and near/mid/far workspace labels.

Return type:

tuple[Tensor, list[str]]

embodichain_benchmarks.motion_generation.scenarios.atomic_contact_cases#

Press, slide, and twist benchmark cases.

Classes:

PressCases

Close the gripper, reach a contact pose, and retract.

SlideCases

Approach, grasp, and physically translate an articulation link.

TwistCases

Approach, grasp, and physically rotate an articulation link.

class embodichain_benchmarks.motion_generation.scenarios.atomic_contact_cases.PressCases[source]#

Close the gripper, reach a contact pose, and retract.

Methods:

articulation_effect_segment(case)

Start button-displacement diagnostics at the pressing segment.

build_invocation(scenario, case, adapter)

Ground the case into one planner-independent action invocation.

generate_case(scenario, suite, track, ...)

Generate one frozen case and independent IK validity evidence.

physical_evaluator()

Return the v1 press-stage evaluator.

replay_observation_windows(compiled, case)

Collect the target-joint trace only during commanded pressing.

Attributes:

articulation_effect_segment(case)[source]#

Start button-displacement diagnostics at the pressing segment.

Return type:

str

build_invocation(scenario, case, adapter)[source]#

Ground the case into one planner-independent action invocation.

Return type:

ActionInvocation

generate_case(scenario, suite, track, config, *, seed, batch_size)[source]#

Generate one frozen case and independent IK validity evidence.

Return type:

BenchmarkCase

physical_evaluator()[source]#

Return the v1 press-stage evaluator.

Return type:

PressEvaluator

replay_observation_windows(compiled, case)[source]#

Collect the target-joint trace only during commanded pressing.

Return type:

tuple[ReplayObservationWindow, ...]

requires_gripper = True#
skill_id: str = 'press'#
class embodichain_benchmarks.motion_generation.scenarios.atomic_contact_cases.SlideCases[source]#

Approach, grasp, and physically translate an articulation link.

Methods:

articulation_effect_segment(case)

Start drawer-displacement diagnostics at the pull or push segment.

build_invocation(scenario, case, adapter)

Ground the case into one planner-independent action invocation.

generate_case(scenario, suite, track, ...)

Generate one frozen case and independent IK validity evidence.

physical_evaluator()

Return the v1 slide-stage evaluator.

replay_observation_windows(compiled, case)

Collect the commanded slide and release windows.

Attributes:

articulation_effect_segment(case)[source]#

Start drawer-displacement diagnostics at the pull or push segment.

Return type:

str

build_invocation(scenario, case, adapter)[source]#

Ground the case into one planner-independent action invocation.

Return type:

ActionInvocation

generate_case(scenario, suite, track, config, *, seed, batch_size)[source]#

Generate one frozen case and independent IK validity evidence.

Return type:

BenchmarkCase

physical_evaluator()[source]#

Return the v1 slide-stage evaluator.

Return type:

SlideEvaluator

replay_observation_windows(compiled, case)[source]#

Collect the commanded slide and release windows.

Return type:

tuple[ReplayObservationWindow, ...]

requires_gripper = True#
skill_id: str = 'slide'#
class embodichain_benchmarks.motion_generation.scenarios.atomic_contact_cases.TwistCases[source]#

Approach, grasp, and physically rotate an articulation link.

Methods:

articulation_effect_segment(case)

Start knob-displacement diagnostics at the rotational segment.

build_invocation(scenario, case, adapter)

Ground the case into one planner-independent action invocation.

generate_case(scenario, suite, track, ...)

Generate one frozen case and independent IK validity evidence.

physical_evaluator()

Return the v1 twist-stage evaluator.

replay_observation_windows(compiled, case)

Collect executed TCP/joint traces and the release boundary.

Attributes:

articulation_effect_segment(case)[source]#

Start knob-displacement diagnostics at the rotational segment.

Return type:

str

build_invocation(scenario, case, adapter)[source]#

Ground the case into one planner-independent action invocation.

Return type:

ActionInvocation

generate_case(scenario, suite, track, config, *, seed, batch_size)[source]#

Generate one frozen case and independent IK validity evidence.

Return type:

BenchmarkCase

physical_evaluator()[source]#

Return the v1 twist-stage evaluator.

Return type:

TwistEvaluator

replay_observation_windows(compiled, case)[source]#

Collect executed TCP/joint traces and the release boundary.

Return type:

tuple[ReplayObservationWindow, ...]

requires_gripper = True#
skill_id: str = 'twist'#

embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation#

Measured physical stage evaluators and observation-only evidence helpers.

Classes:

MoveEndEffectorEvaluator

Score the physically settled TCP position without planner state.

MoveHeldObjectEvaluator

Evaluate transport plus the held object's settled pose and retention.

MoveJointsEvaluator

Score the physically settled arm joints without planner state.

PickUpEvaluator

Evaluate grasp, lift, and terminal retention from replayed state.

PlaceEvaluator

Evaluate release, terminal placement, and post-release stability.

PressEvaluator

Evaluate contact and signed button stroke inside the press segment.

SlideEvaluator

Evaluate grasp proxy, drawer travel, and release at their phase windows.

TwistEvaluator

Evaluate executed TCP twist rather than substituting knob displacement.

Functions:

executed_final_pose_errors(case, final_tcp_pose)

Return final TCP errors under the case's declared rotation symmetry.

gripper_holding(trace, observation, part, *)

Return whether the final sample stopped between open and grasp commands.

gripper_released(trace, observation, part)

Return whether the final window sample is nearer open than grasp.

merge_stage_evidence_batches(*batches)

Merge stage batches while preserving stable environment order.

minimum_height_above_initial_support(trace, case)

Return the minimum object-reference height above its frozen start plane.

motion_valid_mask(motion_outcomes, *, device)

Return the common motion-valid result as one device-local mask.

pose_rotation_error(actual, target)

Return the geodesic rotation error for matching pose batches.

replay_trace(observation, observation_id)

Return one measured replay trace without fabricating missing evidence.

signed_rotation_about_axis(...)

Return the signed angle of current @ reference^T about world_axis.

stage_evidence_batch(*, stage_id, window, ...)

Convert measured batch scalars into row-local standard evidence.

trace_validity(*traces, observation)

Preserve row-local raw-data validity before deriving boolean proxies.

class embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.MoveEndEffectorEvaluator[source]#

Score the physically settled TCP position without planner state.

Methods:

evaluate(case, observation, motion_outcomes)

evaluate(case, observation, motion_outcomes)[source]#
Return type:

tuple[PhysicalEvaluation, ...]

class embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.MoveHeldObjectEvaluator[source]#

Evaluate transport plus the held object’s settled pose and retention.

Methods:

evaluate(case, observation, motion_outcomes)

evaluate(case, observation, motion_outcomes)[source]#
Return type:

tuple[PhysicalEvaluation, ...]

class embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.MoveJointsEvaluator[source]#

Score the physically settled arm joints without planner state.

Methods:

evaluate(case, observation, motion_outcomes)

evaluate(case, observation, motion_outcomes)[source]#
Return type:

tuple[PhysicalEvaluation, ...]

class embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.PickUpEvaluator[source]#

Evaluate grasp, lift, and terminal retention from replayed state.

Methods:

evaluate(case, observation, motion_outcomes)

evaluate(case, observation, motion_outcomes)[source]#
Return type:

tuple[PhysicalEvaluation, ...]

class embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.PlaceEvaluator[source]#

Evaluate release, terminal placement, and post-release stability.

Methods:

evaluate(case, observation, motion_outcomes)

evaluate(case, observation, motion_outcomes)[source]#
Return type:

tuple[PhysicalEvaluation, ...]

class embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.PressEvaluator[source]#

Evaluate contact and signed button stroke inside the press segment.

Methods:

evaluate(case, observation, motion_outcomes)

evaluate(case, observation, motion_outcomes)[source]#
Return type:

tuple[PhysicalEvaluation, ...]

class embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.SlideEvaluator[source]#

Evaluate grasp proxy, drawer travel, and release at their phase windows.

Methods:

evaluate(case, observation, motion_outcomes)

evaluate(case, observation, motion_outcomes)[source]#
Return type:

tuple[PhysicalEvaluation, ...]

class embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.TwistEvaluator[source]#

Evaluate executed TCP twist rather than substituting knob displacement.

Methods:

evaluate(case, observation, motion_outcomes)

evaluate(case, observation, motion_outcomes)[source]#
Return type:

tuple[PhysicalEvaluation, ...]

embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.executed_final_pose_errors(case, final_tcp_pose)[source]#

Return final TCP errors under the case’s declared rotation symmetry.

Return type:

tuple[Tensor, Tensor]

embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.gripper_holding(trace, observation, part, *, gap=0.001)[source]#

Return whether the final sample stopped between open and grasp commands.

Return type:

Tensor | None

embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.gripper_released(trace, observation, part)[source]#

Return whether the final window sample is nearer open than grasp.

Return type:

Tensor | None

embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.merge_stage_evidence_batches(*batches)[source]#

Merge stage batches while preserving stable environment order.

Return type:

tuple[Mapping[str, StageEvidence], ...]

embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.minimum_height_above_initial_support(trace, case)[source]#

Return the minimum object-reference height above its frozen start plane.

Return type:

Tensor | None

embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.motion_valid_mask(motion_outcomes, *, device)[source]#

Return the common motion-valid result as one device-local mask.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.pose_rotation_error(actual, target)[source]#

Return the geodesic rotation error for matching pose batches.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.replay_trace(observation, observation_id)[source]#

Return one measured replay trace without fabricating missing evidence.

Return type:

ReplayObservationTrace | None

embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.signed_rotation_about_axis(reference_rotation, current_rotation, world_axis)[source]#

Return the signed angle of current @ reference^T about world_axis.

The sine comes from the relative rotation’s skew part projected on the axis and the cosine from its trace, so a rotation purely about the axis is recovered exactly over the full (-pi, pi] range.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.stage_evidence_batch(*, stage_id, window, measurements, method, valid_mask=None, source='simulation_state')[source]#

Convert measured batch scalars into row-local standard evidence.

Return type:

tuple[Mapping[str, StageEvidence], ...]

embodichain_benchmarks.motion_generation.scenarios.atomic_evaluation.trace_validity(*traces, observation)[source]#

Preserve row-local raw-data validity before deriving boolean proxies.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_grasps#

Frozen and case-scoped grasp generators for reproducible skill cases.

Classes:

ActiveCaseGraspPoseGenerator

Serve the manifest grasp of the case being compiled.

CaseSampleCountGraspPoseGenerator

Serve each grasp-sampling case from a sampler with its own sample count.

DeterministicCenterGraspPoseGenerator

Return the manifest-defined synthetic handle's center grasp.

FrozenArticulationGraspPoseGenerator

Replay manifest-frozen target-to-grasp transforms for Slide cases.

Functions:

box_surface_mesh(size, *, device)

Build a target-local box mesh for one synthetic graspable handle.

center_grasp_pose(mesh_vertices, obj_poses, ...)

Resolve a deterministic center grasp aligned with the approach axis.

grasp_sample_count(case)

Return the grasp sampler size a self-sampling case was frozen with.

parallel_jaw_sampler(n_sample)

Build the PGI grasp sampler a self-sampling case compiles with.

class embodichain_benchmarks.motion_generation.scenarios.atomic_grasps.ActiveCaseGraspPoseGenerator[source]#

Serve the manifest grasp of the case being compiled.

Skills such as OpenDoor sample their own grasp and accept no explicit one. The scenario activates the case’s frozen (target, grasp) pair around compilation; the target-local grasp is replayed for the observed target pose. Without an active pair the request goes to fallback.

Methods:

__init__([fallback])

activate(target_pose, grasp_pose)

Freeze one case's target-local grasp for the next compilation.

deactivate()

Return to the fallback generator after compilation.

get_best_grasp_poses(*, mesh_vertices, ...)

Return the active frozen grasp for every target row.

get_valid_grasp_poses(*, mesh_vertices, ...)

Return the active frozen grasp as one zero-cost candidate per row.

__init__(fallback=None)[source]#
activate(target_pose, grasp_pose)[source]#

Freeze one case’s target-local grasp for the next compilation.

Return type:

None

deactivate()[source]#

Return to the fallback generator after compilation.

Return type:

None

get_best_grasp_poses(*, mesh_vertices, mesh_triangles, obj_poses, approach_direction)[source]#

Return the active frozen grasp for every target row.

Return type:

tuple[Tensor, Tensor, Tensor]

get_valid_grasp_poses(*, mesh_vertices, mesh_triangles, obj_poses, approach_direction, obj_longest_axis=None, is_positive_part=True)[source]#

Return the active frozen grasp as one zero-cost candidate per row.

Return type:

list[tuple[Tensor, Tensor]]

class embodichain_benchmarks.motion_generation.scenarios.atomic_grasps.CaseSampleCountGraspPoseGenerator[source]#

Serve each grasp-sampling case from a sampler with its own sample count.

Skills such as CoordinatedPickment sample their grasps during compilation and the case manifest froze the waypoints those grasps produced with the case’s grasp_sample_count. The scenario activates that count around compilation so another case in the track cannot change the sampled surface. Without an active count the request goes to default_count.

Methods:

__init__(samplers, *, default_count)

activate(sample_count)

Use the sampler built with sample_count for the next compilation.

deactivate()

Return to the default sampler after compilation.

get_best_grasp_poses(**kwargs)

Forward the request to the active sampler.

get_dual_arm_valid_grasp_poses(**kwargs)

Forward the request to the active sampler.

get_grasp_candidates(**kwargs)

Forward the request to the active sampler.

get_valid_grasp_poses(**kwargs)

Forward the request to the active sampler.

__init__(samplers, *, default_count)[source]#
activate(sample_count)[source]#

Use the sampler built with sample_count for the next compilation.

Return type:

None

deactivate()[source]#

Return to the default sampler after compilation.

Return type:

None

get_best_grasp_poses(**kwargs)[source]#

Forward the request to the active sampler.

Return type:

tuple

get_dual_arm_valid_grasp_poses(**kwargs)[source]#

Forward the request to the active sampler.

Return type:

list

get_grasp_candidates(**kwargs)[source]#

Forward the request to the active sampler.

Return type:

list

get_valid_grasp_poses(**kwargs)[source]#

Forward the request to the active sampler.

Return type:

list

class embodichain_benchmarks.motion_generation.scenarios.atomic_grasps.DeterministicCenterGraspPoseGenerator[source]#

Return the manifest-defined synthetic handle’s center grasp.

Methods:

get_best_grasp_poses(*, mesh_vertices, ...)

Return the deterministic center grasp for every object pose.

get_valid_grasp_poses(*, mesh_vertices, ...)

Return one deterministic zero-cost grasp per object pose.

get_best_grasp_poses(*, mesh_vertices, mesh_triangles, obj_poses, approach_direction)[source]#

Return the deterministic center grasp for every object pose.

Return type:

tuple[Tensor, Tensor, Tensor]

get_valid_grasp_poses(*, mesh_vertices, mesh_triangles, obj_poses, approach_direction, obj_longest_axis=None, is_positive_part=True)[source]#

Return one deterministic zero-cost grasp per object pose.

Return type:

list[tuple[Tensor, Tensor]]

class embodichain_benchmarks.motion_generation.scenarios.atomic_grasps.FrozenArticulationGraspPoseGenerator[source]#

Replay manifest-frozen target-to-grasp transforms for Slide cases.

Methods:

__init__(target_grasps)

get_best_grasp_poses(*, mesh_vertices, ...)

Return the exact grasp selected while building the case manifest.

get_valid_grasp_poses(*, mesh_vertices, ...)

Return one frozen zero-cost grasp candidate per environment.

__init__(target_grasps)[source]#
get_best_grasp_poses(*, mesh_vertices, mesh_triangles, obj_poses, approach_direction)[source]#

Return the exact grasp selected while building the case manifest.

Return type:

tuple[Tensor, Tensor, Tensor]

get_valid_grasp_poses(*, mesh_vertices, mesh_triangles, obj_poses, approach_direction, obj_longest_axis=None, is_positive_part=True)[source]#

Return one frozen zero-cost grasp candidate per environment.

Return type:

list[tuple[Tensor, Tensor]]

embodichain_benchmarks.motion_generation.scenarios.atomic_grasps.box_surface_mesh(size, *, device)[source]#

Build a target-local box mesh for one synthetic graspable handle.

Return type:

tuple[Tensor, Tensor]

embodichain_benchmarks.motion_generation.scenarios.atomic_grasps.center_grasp_pose(mesh_vertices, obj_poses, approach_direction)[source]#

Resolve a deterministic center grasp aligned with the approach axis.

Return type:

Tensor

embodichain_benchmarks.motion_generation.scenarios.atomic_grasps.grasp_sample_count(case)[source]#

Return the grasp sampler size a self-sampling case was frozen with.

Return type:

int

embodichain_benchmarks.motion_generation.scenarios.atomic_grasps.parallel_jaw_sampler(n_sample)[source]#

Build the PGI grasp sampler a self-sampling case compiles with.

Return type:

ParallelJawGraspPoseGenerator

embodichain_benchmarks.motion_generation.scenarios.atomic_manipulation_cases#

Pickup, held-object motion, and placement benchmark cases.

Classes:

HeldObjectCases

Shared isolated held-object precondition for transport and placement.

MoveHeldObjectCases

Move an already held rigid object to one frozen target pose.

PickUpCases

Explicit-grasp PickUp cases that isolate motion-planner performance.

PlaceCases

Place an already held rigid object and release it under physics.

class embodichain_benchmarks.motion_generation.scenarios.atomic_manipulation_cases.HeldObjectCases[source]#

Shared isolated held-object precondition for transport and placement.

Methods:

initial_task_state(scenario, case)

Return an optional symbolic precondition for isolated action testing.

lift_segment_start(compiled)

Return the prepended PickUp lift boundary for composite cases.

Attributes:

initial_task_state(scenario, case)[source]#

Return an optional symbolic precondition for isolated action testing.

Return type:

TaskState

lift_segment_start(compiled)[source]#

Return the prepended PickUp lift boundary for composite cases.

move_held_object and place are compiled as PickUp -> transport/place sequences. Once the first PickUp lift segment begins, the benchmark replay should clear residual rigid-body dynamics just as it does for an isolated PickUp; otherwise gravity and contact impulses can make the cube drift out of the gripper before the second action starts.

Return type:

int | None

requires_gripper = True#
skill_id: str#
class embodichain_benchmarks.motion_generation.scenarios.atomic_manipulation_cases.MoveHeldObjectCases[source]#

Move an already held rigid object to one frozen target pose.

Methods:

build_invocation(scenario, case, adapter)

Ground the case into one planner-independent action invocation.

generate_case(scenario, suite, track, ...)

Generate one frozen case and independent IK validity evidence.

physical_evaluator()

Return the v1 MoveHeldObject evaluator.

replay_observation_windows(compiled, case)

Collect transport plus settled pose and grasp state.

Attributes:

build_invocation(scenario, case, adapter)[source]#

Ground the case into one planner-independent action invocation.

Return type:

ActionInvocation

generate_case(scenario, suite, track, config, *, seed, batch_size)[source]#

Generate one frozen case and independent IK validity evidence.

Return type:

BenchmarkCase

physical_evaluator()[source]#

Return the v1 MoveHeldObject evaluator.

Return type:

MoveHeldObjectEvaluator

replay_observation_windows(compiled, case)[source]#

Collect transport plus settled pose and grasp state.

Return type:

tuple[ReplayObservationWindow, ...]

skill_id: str = 'move_held_object'#
class embodichain_benchmarks.motion_generation.scenarios.atomic_manipulation_cases.PickUpCases[source]#

Explicit-grasp PickUp cases that isolate motion-planner performance.

Methods:

build_invocation(scenario, case, adapter)

Ground the case into one planner-independent action invocation.

generate_case(scenario, suite, track, ...)

Generate one frozen case and independent IK validity evidence.

lift_segment_start(compiled)

Return the lift boundary emitted by PickUp.

physical_evaluator()

Return the v1 PickUp evaluator.

replay_observation_windows(compiled, case)

Collect grasp, lift, and terminal retention evidence.

Attributes:

build_invocation(scenario, case, adapter)[source]#

Ground the case into one planner-independent action invocation.

Return type:

ActionInvocation

generate_case(scenario, suite, track, config, *, seed, batch_size)[source]#

Generate one frozen case and independent IK validity evidence.

Return type:

BenchmarkCase

lift_segment_start(compiled)[source]#

Return the lift boundary emitted by PickUp.

Return type:

int | None

physical_evaluator()[source]#

Return the v1 PickUp evaluator.

Return type:

PickUpEvaluator

replay_observation_windows(compiled, case)[source]#

Collect grasp, lift, and terminal retention evidence.

Return type:

tuple[ReplayObservationWindow, ...]

requires_gripper = True#
skill_id: str = 'pick_up'#
class embodichain_benchmarks.motion_generation.scenarios.atomic_manipulation_cases.PlaceCases[source]#

Place an already held rigid object and release it under physics.

Methods:

build_invocation(scenario, case, adapter)

Ground the case into one planner-independent action invocation.

generate_case(scenario, suite, track, ...)

Generate one frozen case and independent IK validity evidence.

physical_evaluator()

Return the v1 Place evaluator.

replay_observation_windows(compiled, case)

Collect release and terminal stability evidence.

Attributes:

build_invocation(scenario, case, adapter)[source]#

Ground the case into one planner-independent action invocation.

Return type:

ActionInvocation

generate_case(scenario, suite, track, config, *, seed, batch_size)[source]#

Generate one frozen case and independent IK validity evidence.

Return type:

BenchmarkCase

physical_evaluator()[source]#

Return the v1 Place evaluator.

Return type:

PlaceEvaluator

replay_observation_windows(compiled, case)[source]#

Collect release and terminal stability evidence.

Return type:

tuple[ReplayObservationWindow, ...]

skill_id: str = 'place'#

embodichain_benchmarks.motion_generation.scenarios.atomic_motion_cases#

Representative end-effector and joint-motion integration cases.

Classes:

MoveEndEffectorCases

Deterministic robot-relative MoveEndEffector cases.

MoveJointsCases

Deterministic relative joint-space waypoint cases.

class embodichain_benchmarks.motion_generation.scenarios.atomic_motion_cases.MoveEndEffectorCases[source]#

Deterministic robot-relative MoveEndEffector cases.

Methods:

build_invocation(scenario, case, adapter)

Ground the case into one planner-independent action invocation.

generate_case(scenario, suite, track, ...)

Generate one frozen case and independent IK validity evidence.

physical_evaluator()

Evaluate the terminal measured TCP position under the v1 standard.

Attributes:

build_invocation(scenario, case, adapter)[source]#

Ground the case into one planner-independent action invocation.

Return type:

ActionInvocation

generate_case(scenario, suite, track, config, *, seed, batch_size)[source]#

Generate one frozen case and independent IK validity evidence.

Return type:

BenchmarkCase

physical_evaluator()[source]#

Evaluate the terminal measured TCP position under the v1 standard.

Return type:

MoveEndEffectorEvaluator

skill_id: str = 'move_end_effector'#
class embodichain_benchmarks.motion_generation.scenarios.atomic_motion_cases.MoveJointsCases[source]#

Deterministic relative joint-space waypoint cases.

Methods:

build_invocation(scenario, case, adapter)

Ground the case into one planner-independent action invocation.

generate_case(scenario, suite, track, ...)

Generate one frozen case and independent IK validity evidence.

physical_evaluator()

Evaluate the terminal measured joint state under the v1 standard.

Attributes:

build_invocation(scenario, case, adapter)[source]#

Ground the case into one planner-independent action invocation.

Return type:

ActionInvocation

generate_case(scenario, suite, track, config, *, seed, batch_size)[source]#

Generate one frozen case and independent IK validity evidence.

Return type:

BenchmarkCase

physical_evaluator()[source]#

Evaluate the terminal measured joint state under the v1 standard.

Return type:

MoveJointsEvaluator

skill_id: str = 'move_joints'#

embodichain_benchmarks.motion_generation.scenarios.atomic_objects#

Config-driven scene objects for Atomic Skill benchmark cases.

Classes:

AtomicArticulationHandle

Simulation articulation plus its frozen benchmark reset state.

AtomicObjectHandle

Simulation object plus its algorithm-independent frozen state.

Functions:

atomic_object_kind_names()

Return registered object kinds in deterministic order.

create_atomic_articulation(simulation, ...)

Create one asset-backed articulation and freeze its settled reset state.

create_atomic_object(simulation, ...[, ...])

Create one config-driven object and freeze its settled initial pose.

register_atomic_object_kind(name, factory)

Register a config-only object shape factory.

class embodichain_benchmarks.motion_generation.scenarios.atomic_objects.AtomicArticulationHandle[source]#

Simulation articulation plus its frozen benchmark reset state.

Methods:

__init__(object_id, config, entity, ...)

joint_qpos(joint_name)

Return the live position of one named active joint.

link_mesh(link_name)

Return cloned target-local mesh tensors for one articulation link.

link_pose(link_name)

Return a cloned world pose for one articulation link.

park(index)

Reset and move an inactive articulation outside the workspace.

reset(*[, pose, qpos])

Restore root pose, actual joint state, drive target, and dynamics.

Attributes:

__init__(object_id, config, entity, initial_pose, initial_qpos)#
config: dict[str, object]#
entity: Articulation#
initial_pose: Tensor#
initial_qpos: Tensor#
joint_qpos(joint_name)[source]#

Return the live position of one named active joint.

Return type:

Tensor

Return cloned target-local mesh tensors for one articulation link.

Return type:

tuple[Tensor, Tensor]

Return a cloned world pose for one articulation link.

Return type:

Tensor

object_id: str#
park(index)[source]#

Reset and move an inactive articulation outside the workspace.

Return type:

None

reset(*, pose=None, qpos=None)[source]#

Restore root pose, actual joint state, drive target, and dynamics.

Return type:

None

class embodichain_benchmarks.motion_generation.scenarios.atomic_objects.AtomicObjectHandle[source]#

Simulation object plus its algorithm-independent frozen state.

Methods:

__init__(object_id, kind, config, entity, ...)

park(index)

Move an inactive object outside every benchmark workspace.

reset()

Restore the frozen initial pose and clear residual dynamics.

Attributes:

__init__(object_id, kind, config, entity, initial_pose)#
config: dict[str, object]#
entity: RigidObject#
initial_pose: Tensor#
kind: str#
object_id: str#
park(index)[source]#

Move an inactive object outside every benchmark workspace.

Return type:

None

reset()[source]#

Restore the frozen initial pose and clear residual dynamics.

Return type:

None

embodichain_benchmarks.motion_generation.scenarios.atomic_objects.atomic_object_kind_names()[source]#

Return registered object kinds in deterministic order.

Return type:

tuple[str, ...]

embodichain_benchmarks.motion_generation.scenarios.atomic_objects.create_atomic_articulation(simulation, articulation_config, *, initialize=True)[source]#

Create one asset-backed articulation and freeze its settled reset state.

Return type:

AtomicArticulationHandle

embodichain_benchmarks.motion_generation.scenarios.atomic_objects.create_atomic_object(simulation, object_config, *, initialize=True)[source]#

Create one config-driven object and freeze its settled initial pose.

Return type:

AtomicObjectHandle

embodichain_benchmarks.motion_generation.scenarios.atomic_objects.register_atomic_object_kind(name, factory)[source]#

Register a config-only object shape factory.

Return type:

None

embodichain_benchmarks.motion_generation.scenarios.atomic_policy#

Benchmark motion policy, attribution, and shared motion budgets.

Functions:

benchmark_motion_attribution(skill_id)

Describe whether the selected planner backend produced the motion.

benchmark_motion_policy(skill_id, sample_count)

Build the benchmark policy without hiding exact-path planner bypasses.

benchmark_motion_policy_strategy(skill_id)

Return the auditable motion strategy required by one benchmark skill.

policy_motion_budget(protocol)

Map suite requests to the scalar interface of scenario-owned motion.

with_motion_budget(invocation, protocol)

Apply shared scalar budgets to IK paths without overriding backend vectors.

with_plan_options(invocation, plan_opts)

Give a motion-generation invocation planner options it left unset.

embodichain_benchmarks.motion_generation.scenarios.atomic_policy.benchmark_motion_attribution(skill_id)[source]#

Describe whether the selected planner backend produced the motion.

Return type:

dict[str, object]

embodichain_benchmarks.motion_generation.scenarios.atomic_policy.benchmark_motion_policy(skill_id, sample_count)[source]#

Build the benchmark policy without hiding exact-path planner bypasses.

Return type:

MotionPolicy

embodichain_benchmarks.motion_generation.scenarios.atomic_policy.benchmark_motion_policy_strategy(skill_id)[source]#

Return the auditable motion strategy required by one benchmark skill.

Return type:

Literal['motion_gen', 'ik_interp']

embodichain_benchmarks.motion_generation.scenarios.atomic_policy.policy_motion_budget(protocol)[source]#

Map suite requests to the scalar interface of scenario-owned motion.

Return type:

dict[str, float]

embodichain_benchmarks.motion_generation.scenarios.atomic_policy.with_motion_budget(invocation, protocol)[source]#

Apply shared scalar budgets to IK paths without overriding backend vectors.

MotionPolicy accepts scalars. Per-joint suite requests use their minimum for these scenario-owned paths, conservatively satisfying every request; the robot still caps each joint separately. Omitted requests retain the invocation defaults. Selected backends receive the full vector through their existing planner options.

Return type:

ActionInvocation

embodichain_benchmarks.motion_generation.scenarios.atomic_policy.with_plan_options(invocation, plan_opts)[source]#

Give a motion-generation invocation planner options it left unset.

Return type:

ActionInvocation

embodichain_benchmarks.motion_generation.scenarios.atomic_replay#

Physical replay and observation collection for Atomic Skill benchmarks.

Classes:

AtomicReplayRuntime

Replay trajectories against explicitly bound simulator state.

class embodichain_benchmarks.motion_generation.scenarios.atomic_replay.AtomicReplayRuntime[source]#

Replay trajectories against explicitly bound simulator state.

Create one runtime at each evaluation or video boundary so its object and gripper bindings belong to the currently configured scenario. This runtime owns neither scene reset nor simulator teardown.

Parameters:
  • simulation (SimulationManager) – Simulator receiving the replay’s physics steps.

  • robot (Robot) – Robot receiving full-joint position commands.

  • control_part (str) – Arm used for end-effector and final-state observations.

  • track_config (Mapping[str, object]) – Track-owned physics replay settings.

  • objects (Mapping[str, AtomicObjectHandle]) – Configured movable object handles indexed by object ID.

  • articulations (Mapping[str, AtomicArticulationHandle]) – Configured articulation handles indexed by object ID.

  • gripper_states (Mapping[str, tuple[Tensor, Tensor]]) – Open and grasp commands for every configured gripper.

Methods:

__init__(*, simulation, robot, control_part, ...)

execute(compiled, case, provider)

Replay a finite full-robot trajectory and collect physical evidence.

hold_static()

Hold the current scene so a failed-case debug video has frames.

is_recordable(compiled)

Check whether a rollout can produce a diagnostic replay video.

is_replayable(compiled)

Return whether successful rows contain a finite nonempty trajectory.

replay_physics(compiled, case, provider, *, ...)

Replay one compiled trajectory with the common physics contract.

__init__(*, simulation, robot, control_part, track_config, objects, articulations, gripper_states)[source]#
execute(compiled, case, provider)[source]#

Replay a finite full-robot trajectory and collect physical evidence.

Parameters:
  • compiled (CompiledTrajectory) – Full-robot trajectory with one planning result per row.

  • case (BenchmarkCase) – Frozen benchmark inputs and expected initial robot state.

  • provider (AtomicSkillCaseProvider) – Skill-specific observation and effect-window declarations.

Return type:

ExecutionObservation | None

Returns:

Execution observations, or None when no successful finite row can be replayed. Failed rows remain at their frozen start state.

hold_static()[source]#

Hold the current scene so a failed-case debug video has frames.

Return type:

None

static is_recordable(compiled)[source]#

Check whether a rollout can produce a diagnostic replay video.

Parameters:

compiled (CompiledTrajectory) – Full-robot trajectory, including failed planning rows.

Return type:

bool

Returns:

Whether every row is finite and at least one waypoint is present.

static is_replayable(compiled)[source]#

Return whether successful rows contain a finite nonempty trajectory.

Parameters:

compiled (CompiledTrajectory) – Full-robot trajectory and per-row planning results.

Return type:

bool

Returns:

Whether at least one row succeeded and all successful rows are finite.

replay_physics(compiled, case, provider, *, collect_metrics)[source]#

Replay one compiled trajectory with the common physics contract.

Parameters:
  • compiled (CompiledTrajectory) – Full-robot trajectory and per-row planning results.

  • case (BenchmarkCase) – Frozen benchmark inputs, including failed-row hold positions.

  • provider (AtomicSkillCaseProvider | None) – Optional skill-specific observation and effect declarations.

  • collect_metrics (bool) – Accumulate tracking and physical observations when true; video replay uses the same stepping without accumulation.

Return type:

_ReplayMetrics | None

Returns:

Collected replay metrics when requested, otherwise None.

embodichain_benchmarks.motion_generation.scenarios.atomic_skill#

Physics-backed Atomic Skill scenario orchestration.

Classes:

AtomicSkillScenario

Run fixed Atomic Actions through an adapter-owned MotionGenerator.

class embodichain_benchmarks.motion_generation.scenarios.atomic_skill.AtomicSkillScenario[source]#

Run fixed Atomic Actions through an adapter-owned MotionGenerator.

Methods:

__init__()

activate_articulation(object_id)

Reset one articulation and park all inactive scene objects.

activate_object(object_id)

Reset one case object and park every other configured object.

activate_objects(object_ids)

Reset several case objects together and park every other one.

articulation_handle(object_id)

Resolve one configured articulation identifier.

batch_sizes(suite, track)

Return the explicitly configured physical-execution batch sizes.

bind_cases(cases)

Restore skill providers and scene bindings without resampling inputs.

close_planner(adapter)

Drop the engine before its adapter closes the shared generator.

close_runtime()

Release Python references before SimulationManager teardown.

configure_runtime(simulation, robot, suite, ...)

Finalize the pre-created object pool and cache robot command states.

create_runtime_entities(simulation, suite, track)

Add every rigid/articulated asset before GPU physics initialization.

evaluate_case(result, case, robot, ...[, ...])

Validate motion, replay physics, and evaluate physical task success.

failure_outcomes(case, failure_code)

Mark execution/task stages false after runner-level failures.

generate_cases(suite, track, robot, ...)

Generate the algorithm-independent Atomic Skill case manifest.

object_handle(object_id)

Resolve a configured object identifier with an actionable error.

open_grippers()

Command and set every configured gripper to its open state.

plan_case(adapter, case)

Compile one Atomic Action with an explicitly pinned motion backend.

plan_contract_error(result)

Accept compiled Atomic Action trajectories instead of raw plans.

prepare_planner(adapter, first_case)

Bind one AtomicActionEngine to the adapter-owned MotionGenerator.

randomize_articulation_pose(handle, config, ...)

Translate and yaw an articulation root in every environment.

randomize_object_pose(handle, config, *, ...)

Install one deterministic position/yaw perturbation for an object.

randomize_robot_start(config, *, seed, stream)

Apply a deterministic, bounded perturbation to the arm start state.

record_replay(result, case, evaluation, *, ...)

Reset the case and record a second untimed physics replay.

reference_motion_generator()

Return the TOPPRA generator used to record a skill's own key poses.

require_engine()

Return the prepared Atomic Action engine.

reset_case(simulation, robot, case, control_part)

Restore full robot and object state before every planner call.

resolve_antipodal_grasp(handle, object_pose, ...)

Freeze one geometry-aware, independently reachable PGI grasp pose.

resolve_articulation_grasp(*, mesh_vertices, ...)

Freeze a geometry-aware PGI grasp that can complete a Slide path.

restore_base_robot()

Restore the robot state captured before scenario case generation.

sample_articulation_geometry(handle, ...)

Resolve and cache deterministic tutorial-style link geometry.

screen_antipodal_grasp(handle, object_pose, ...)

Freeze the first sampled object grasp whose skill poses are reachable.

screen_articulation_grasp(*, mesh_vertices, ...)

Freeze the first sampled link grasp whose skill poses are reachable.

set_robot_start(manipulator_qpos, *, ...[, ...])

Install a manipulator start and optional gripper command.

solve_reference_qpos(start_qpos, ...)

Build independent sequential-IK validity evidence for a case.

trial_metadata(case)

Expose planner attribution before planning or evaluation can fail.

validate_articulation_target(handle, *, ...)

Require the configured joint to be the link's nearest typed ancestor.

Attributes:

__init__()[source]#
activate_articulation(object_id)[source]#

Reset one articulation and park all inactive scene objects.

Return type:

AtomicArticulationHandle

activate_object(object_id)[source]#

Reset one case object and park every other configured object.

Return type:

AtomicObjectHandle

activate_objects(object_ids)[source]#

Reset several case objects together and park every other one.

Return type:

list[AtomicObjectHandle]

articulation_handle(object_id)[source]#

Resolve one configured articulation identifier.

Return type:

AtomicArticulationHandle

batch_sizes(suite, track)[source]#

Return the explicitly configured physical-execution batch sizes.

Return type:

list[int]

bind_cases(cases)[source]#

Restore skill providers and scene bindings without resampling inputs.

Parameters:

cases (list[BenchmarkCase]) – Generated or replayed cases for this configured runtime.

Return type:

None

close_planner(adapter)[source]#

Drop the engine before its adapter closes the shared generator.

Return type:

None

close_runtime()[source]#

Release Python references before SimulationManager teardown.

Return type:

None

configure_runtime(simulation, robot, suite, track, control_part)[source]#

Finalize the pre-created object pool and cache robot command states.

Return type:

None

create_runtime_entities(simulation, suite, track)[source]#

Add every rigid/articulated asset before GPU physics initialization.

Return type:

None

end_effector_part: str | None#
evaluate_case(result, case, robot, control_part, suite, *, planning_time_ms, timing_is_solved=True)[source]#

Validate motion, replay physics, and evaluate physical task success.

Return type:

ScenarioEvaluation

failure_outcomes(case, failure_code)[source]#

Mark execution/task stages false after runner-level failures.

Return type:

tuple[CaseOutcome, ...]

generate_cases(suite, track, robot, control_part, batch_size)[source]#

Generate the algorithm-independent Atomic Skill case manifest.

Return type:

list[BenchmarkCase]

object_handle(object_id)[source]#

Resolve a configured object identifier with an actionable error.

Return type:

AtomicObjectHandle

open_grippers()[source]#

Command and set every configured gripper to its open state.

Return type:

None

plan_case(adapter, case)[source]#

Compile one Atomic Action with an explicitly pinned motion backend.

Return type:

object

plan_contract_error(result)[source]#

Accept compiled Atomic Action trajectories instead of raw plans.

Return type:

str | None

prepare_planner(adapter, first_case)[source]#

Bind one AtomicActionEngine to the adapter-owned MotionGenerator.

Return type:

None

randomize_articulation_pose(handle, config, *, seed, stream)[source]#

Translate and yaw an articulation root in every environment.

The articulation joint state stays frozen. Skill-local position and yaw ranges override the pose-batch track’s object-pose ranges.

Return type:

Tensor

randomize_object_pose(handle, config, *, seed, stream, settle_steps=None)[source]#

Install one deterministic position/yaw perturbation for an object.

When settle_steps is supplied, physics is advanced for exactly that many frames and the live post-settle pose is returned. This is used by PickUp case generation so its frozen grasp target matches a cube that has actually fallen onto the table. Omitting the argument preserves the legacy two-frame write/restore behavior used by the other case providers.

Return type:

Tensor

randomize_robot_start(config, *, seed, stream)[source]#

Apply a deterministic, bounded perturbation to the arm start state.

Return type:

Tensor

record_replay(result, case, evaluation, *, output_dir, algorithm_id, video)[source]#

Reset the case and record a second untimed physics replay.

Return type:

Path | None

reference_motion_generator()[source]#

Return the TOPPRA generator used to record a skill’s own key poses.

Coordinated skills compute their waypoints internally. Providers compile them once with this generator while building the manifest and freeze the poses the skill requests; those poses do not depend on the planner later being benchmarked.

Return type:

object

require_engine()[source]#

Return the prepared Atomic Action engine.

Return type:

AtomicActionEngine

required_capabilities: frozenset[str] = frozenset({'atomic_action', 'eef_waypoint', 'joint_waypoint'})#
reset_case(simulation, robot, case, control_part)[source]#

Restore full robot and object state before every planner call.

Return type:

None

resolve_antipodal_grasp(handle, object_pose, approach_direction, *, seed, start_qpos, pre_grasp_distance, lift_height, n_sample, max_candidates, alignment_max_angle_deg)[source]#

Freeze one geometry-aware, independently reachable PGI grasp pose.

Grasp sampling and sequential IK screening happen once while the case manifest is built, before any planner adapter is evaluated. Every planner therefore receives the same explicit grasp pose and planning latency excludes grasp generation.

Return type:

Tensor

resolve_articulation_grasp(*, mesh_vertices, mesh_triangles, target_pose, approach_direction, start_qpos, direction, approach_distance, translation_distance, seed, n_sample, max_candidates, alignment_max_angle_deg)[source]#

Freeze a geometry-aware PGI grasp that can complete a Slide path.

Return type:

Tensor

restore_base_robot()[source]#

Restore the robot state captured before scenario case generation.

Return type:

None

robot: Robot | None#
sample_articulation_geometry(handle, target_link, *, seed, config)[source]#

Resolve and cache deterministic tutorial-style link geometry.

Return type:

dict[str, Tensor]

screen_antipodal_grasp(handle, object_pose, approach_direction, *, seed, start_qpos, n_sample, max_candidates, alignment_max_angle_deg, waypoint_poses, rank=None)[source]#

Freeze the first sampled object grasp whose skill poses are reachable.

Candidates are sampled once with seed, ordered by rank (grasp cost by default), filtered by approach alignment, and tried with their half-turn variant. The first variant whose waypoint_poses pass sequential IK from start_qpos is returned.

Parameters:
  • handle (AtomicObjectHandle) – Object whose mesh is sampled.

  • object_pose (Tensor) – Frozen object pose batch.

  • approach_direction (Tensor) – World approach axis the grasp must follow.

  • seed (int) – Sampling seed.

  • start_qpos (Tensor) – Arm state the IK chain starts from.

  • n_sample (int) – Generator surface-sample count.

  • max_candidates (int) – Number of ranked candidates to screen.

  • alignment_max_angle_deg (float) – Maximum grasp-axis deviation from the approach axis.

  • waypoint_poses (Callable[[Tensor], Sequence[Tensor]]) – Skill poses that a grasp variant must reach, in order, given that variant with shape (1, 4, 4).

  • rank (Callable[[Tensor, Tensor], Tensor] | None) – Optional ordering of finite candidates from their poses and costs; returns candidate indices, best first.

Return type:

Tensor

Returns:

The selected grasp with shape (1, 4, 4).

Raises:

RuntimeError – No candidate variant reaches every skill pose.

screen_articulation_grasp(*, mesh_vertices, mesh_triangles, target_pose, approach_direction, start_qpos, seed, n_sample, max_candidates, alignment_max_angle_deg, waypoint_poses)[source]#

Freeze the first sampled link grasp whose skill poses are reachable.

All rows share one target-link mesh and differ only by the articulation root pose, so candidates are sampled once and the selected target-local grasp is preserved across the batch.

Parameters:
  • mesh_vertices (Tensor) – Target-link vertices in the link frame.

  • mesh_triangles (Tensor) – Target-link triangle indices.

  • target_pose (Tensor) – Link pose batch with shape (B, 4, 4).

  • approach_direction (Tensor) – World approach axes with shape (B, 3).

  • start_qpos (Tensor) – Arm state batch the IK chain starts from.

  • seed (int) – Sampling seed.

  • n_sample (int) – Generator surface-sample count.

  • max_candidates (int) – Number of lowest-cost candidates to screen.

  • alignment_max_angle_deg (float) – Maximum grasp-axis deviation from the approach axis in every row.

  • waypoint_poses (Callable[[Tensor], Sequence[Tensor]]) – Skill poses that a batched grasp variant must reach, in order.

Return type:

Tensor

Returns:

The selected grasp batch with shape (B, 4, 4).

Raises:

RuntimeError – No candidate variant reaches every skill pose.

set_robot_start(manipulator_qpos, *, open_gripper, grasp_gripper=False)[source]#

Install a manipulator start and optional gripper command.

Return type:

None

simulation: SimulationManager | None#
solve_reference_qpos(start_qpos, target_waypoints)[source]#

Build independent sequential-IK validity evidence for a case.

Return type:

Tensor

suite: SuiteCfg | None#
track: TrackCfg | None#
trial_metadata(case)[source]#

Expose planner attribution before planning or evaluation can fail.

Return type:

dict[str, object]

static validate_articulation_target(handle, *, target_link, target_joint, joint_type)[source]#

Require the configured joint to be the link’s nearest typed ancestor.

Return type:

None

embodichain_benchmarks.motion_generation.scenarios.base#

Scenario provider contract for motion-generation tracks.

Classes:

ScenarioEvaluation

Outcomes and execution metrics produced outside planner timing.

ScenarioProvider

Generate, plan, execute, and evaluate one registered scenario kind.

class embodichain_benchmarks.motion_generation.scenarios.base.ScenarioEvaluation[source]#

Outcomes and execution metrics produced outside planner timing.

Methods:

__init__(outcomes[, execution_time_ms, ...])

Attributes:

__init__(outcomes, execution_time_ms=None, end_to_end_time_ms=None, trajectory_duration_s=None, trajectory_waypoints=None, metadata=<factory>)#
end_to_end_time_ms: float | None = None#
execution_time_ms: float | None = None#
metadata: dict[str, object]#
outcomes: tuple[CaseOutcome, ...]#
trajectory_duration_s: float | None = None#
trajectory_waypoints: int | None = None#
class embodichain_benchmarks.motion_generation.scenarios.base.ScenarioProvider[source]#

Generate, plan, execute, and evaluate one registered scenario kind.

Methods:

batch_sizes(suite, track)

Return simulator batch sizes required by this track.

bind_cases(cases)

Bind runtime resources for generated or loaded immutable cases.

close_planner(adapter)

Release scenario resources that retain a planner adapter.

close_runtime()

Release references to scenario-owned simulation entities.

configure_runtime(simulation, robot, suite, ...)

Finalize scenario runtime state before case generation.

create_runtime_entities(simulation, suite, track)

Add scenario-owned entities before the first physics update.

evaluate_case(result, case, robot, ...[, ...])

Externally validate a planner-only trajectory.

failure_outcomes(case, failure_code)

Create scenario-appropriate outcomes after a runner-level failure.

generate_cases(suite, track, robot, ...)

Build the frozen case manifest for one batch size.

plan_case(adapter, case)

Plan one case through the selected adapter.

plan_contract_error(result)

Return a diagnostic when a planner artifact violates this scenario.

prepare_planner(adapter, first_case)

Bind scenario resources to a built planner outside trial timing.

record_replay(result, case, evaluation, *, ...)

Optionally record a second, untimed replay.

reset_case(simulation, robot, case, control_part)

Restore the frozen robot start state before a planning call.

trial_metadata(case)

Return scenario attribution known before a planning call.

Attributes:

abstract batch_sizes(suite, track)[source]#

Return simulator batch sizes required by this track.

Return type:

list[int]

bind_cases(cases)[source]#

Bind runtime resources for generated or loaded immutable cases.

Parameters:

cases (list[BenchmarkCase]) – Canonical cases for the current track and simulator batch. Bindings must not mutate their tensors or nested parameters.

Return type:

None

close_planner(adapter)[source]#

Release scenario resources that retain a planner adapter.

Return type:

None

close_runtime()[source]#

Release references to scenario-owned simulation entities.

Return type:

None

configure_runtime(simulation, robot, suite, track, control_part)[source]#

Finalize scenario runtime state before case generation.

Return type:

None

create_runtime_entities(simulation, suite, track)[source]#

Add scenario-owned entities before the first physics update.

Return type:

None

evaluate_case(result, case, robot, control_part, suite, *, planning_time_ms, timing_is_solved=True)[source]#

Externally validate a planner-only trajectory.

Return type:

ScenarioEvaluation

failure_outcomes(case, failure_code)[source]#

Create scenario-appropriate outcomes after a runner-level failure.

Return type:

tuple[CaseOutcome, ...]

abstract generate_cases(suite, track, robot, control_part, batch_size)[source]#

Build the frozen case manifest for one batch size.

Return type:

list[BenchmarkCase]

plan_case(adapter, case)[source]#

Plan one case through the selected adapter.

Return type:

object

plan_contract_error(result)[source]#

Return a diagnostic when a planner artifact violates this scenario.

Return type:

str | None

prepare_planner(adapter, first_case)[source]#

Bind scenario resources to a built planner outside trial timing.

Return type:

None

record_replay(result, case, evaluation, *, output_dir, algorithm_id, video)[source]#

Optionally record a second, untimed replay. Default is a no-op.

Parameters:
  • result (object) – Planner or compiled-action artifact from the measured trial.

  • case (BenchmarkCase) – Frozen case identity used for the output filename.

  • evaluation (ScenarioEvaluation | None) – Timed evaluation, or None after a runner-level failure.

  • output_dir (Path) – Directory that should receive the mp4.

  • algorithm_id (str) – Planner id used in the filename.

  • video (VideoRecordCfg) – Recording policy and encoder settings.

Return type:

Path | None

Returns:

Path to a saved video, or None when this scenario does not record.

required_capabilities: frozenset[str] = frozenset({})#
reset_case(simulation, robot, case, control_part)[source]#

Restore the frozen robot start state before a planning call.

Return type:

None

trial_metadata(case)[source]#

Return scenario attribution known before a planning call.

Return type:

dict[str, object]

embodichain_benchmarks.motion_generation.scenarios.free_space#

Deterministic reachable cases for the free-space-common track.

Classes:

class embodichain_benchmarks.motion_generation.scenarios.free_space.FreeSpaceScenario[source]#

Methods:

batch_sizes(suite, track)

Return simulator batch sizes required by this track.

generate_cases(suite, track, robot, ...)

Build the frozen case manifest for one batch size.

Attributes:

batch_sizes(suite, track)[source]#

Return simulator batch sizes required by this track.

Return type:

list[int]

generate_cases(suite, track, robot, control_part, batch_size)[source]#

Build the frozen case manifest for one batch size.

Return type:

list[BenchmarkCase]

required_capabilities: frozenset[str] = frozenset({'batched', 'empty_world'})#

embodichain_benchmarks.motion_generation.video#

Optional Atomic Skill replay recording helpers.

Recording is a second, untimed physics pass. Failures never change trial success and never add a fourth Markdown table.

Classes:

VideoRecordCfg

CLI-resolved recording policy for Atomic Skill measured replays.

Functions:

build_video_path(output_dir, algorithm_id, ...)

Build {algorithm}_{skill}_{case}.mp4 under the run videos directory.

record_with_window(sim, cfg, video_path, ...)

Record one headless replay.

should_record_case(cfg, recorded_count, success)

Return whether one measured Atomic Skill case should emit a video.

summarize_video_recording(cfg, video_paths)

Return report notes describing recording coverage without extra tables.

video_cfg_from_args(args)

Build recording config from the motion-generation CLI namespace.

class embodichain_benchmarks.motion_generation.video.VideoRecordCfg[source]#

CLI-resolved recording policy for Atomic Skill measured replays.

Methods:

__init__([enabled, record_failed, ...])

Attributes:

__init__(enabled=False, record_failed=False, case_limit=0, fps=20, width=640, height=480, max_memory_mb=2048, look_at=((0.7, 1.0, 0.8), (-0.5, 0, 0.3), (0.0, 0.0, 1.0)), output_dir=None)#
case_limit: int = 0#
enabled: bool = False#
fps: int = 20#
height: int = 480#
look_at: tuple[Sequence[float], Sequence[float], Sequence[float]] = ((0.7, 1.0, 0.8), (-0.5, 0, 0.3), (0.0, 0.0, 1.0))#
max_memory_mb: int = 2048#
output_dir: Path | None = None#
record_failed: bool = False#
width: int = 640#
embodichain_benchmarks.motion_generation.video.build_video_path(output_dir, algorithm_id, skill_id, case_id)[source]#

Build {algorithm}_{skill}_{case}.mp4 under the run videos directory.

Return type:

Path

embodichain_benchmarks.motion_generation.video.record_with_window(sim, cfg, video_path, replay_fn)[source]#

Record one headless replay. Failures print a warning and return None.

Return type:

Path | None

embodichain_benchmarks.motion_generation.video.should_record_case(cfg, recorded_count, success)[source]#

Return whether one measured Atomic Skill case should emit a video.

Return type:

bool

embodichain_benchmarks.motion_generation.video.summarize_video_recording(cfg, video_paths)[source]#

Return report notes describing recording coverage without extra tables.

Return type:

list[str]

embodichain_benchmarks.motion_generation.video.video_cfg_from_args(args)[source]#

Build recording config from the motion-generation CLI namespace.

Return type:

VideoRecordCfg

embodichain.cli.benchmark#

Unified benchmark CLI.

Usage examples:

embodichain benchmark rl --tasks push_cube --algorithms ppo --suite default
embodichain benchmark rl --rebuild-report-only
embodichain benchmark robotics-kinematic-solver -s pytorch
embodichain benchmark motion-generation --suite motion_panda_free_space_smoke
embodichain benchmark motion-generation --suite atomic_franka_pgi_smoke
embodichain benchmark atomic-action-profile --smoke
embodichain benchmark grasp-pose-generator --device cuda
embodichain benchmark workspace-analyzer
embodichain benchmark rendering --camera-count 2 --resolution 1280x720

Functions:

main([argv])

Dispatch to the appropriate benchmark sub-command CLI.

embodichain.cli.benchmark.main(argv=None)[source]#

Dispatch to the appropriate benchmark sub-command CLI.

Return type:

None