embodichain.lab.sim.atomic_actions

Returns:

Side-effect-free action plan.

build_plan(request, context, *, success, trajectory, expected_effects=None, effect_candidates=None, effect_verification=None, replannable=True, diagnostics=None, segment_lengths=None, scene_dependency_monitor_until=None, scene_dependency_end_segment=None)[source]

Build a validated action plan for a primitive implementation.

Parameters:

request (ResolvedActionRequest[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]) – Resolved invocation snapshot being planned.
context (PlanningContext) – Planning input used for the plan.
success (bool | Tensor) – Per-environment planning success or scalar planner result.
trajectory (TimedTrajectory) – Full-robot trajectory with explicit timing.
expected_effects (StateDelta | None) – Symbolic effects to verify after execution.
effect_candidates (StateDelta | None) – Planned attachment baselines used by phase gates and in-flight held-object guards without committing task state.
effect_verification (EffectVerificationRequirement | None) – Optional explicit physical-effect boundary. Use this when verification is required without a symbolic task- state delta.
replannable (bool) – Whether the execution runtime may replan this action.
diagnostics (PlannerDiagnostics | None) – Optional retained planner diagnostics.
segment_lengths (Mapping[str, int] | None) – Optional ordered mapping from semantic segment names to waypoint counts. Zero-length entries are omitted.
scene_dependency_monitor_until (Mapping[str, int] | None) – Optional per-entity exclusive waypoint-index upper bound for scene-motion invalidation. An entity is monitored while the current waypoint index is smaller than its bound. 0 disables monitoring immediately; omitted dependencies remain monitored for the full action. Once the bound is reached, all pose changes for that entity are ignored.
scene_dependency_end_segment (str | None) – Optional last segment during which scene motion may invalidate and replan the action for every dependency.

Return type:

Returns:

Side-effect-free action plan.

property default_options: OptionsT: Return an owned copy of the action’s default runtime options.

classmethod descriptor()[source]

Return stable metadata used by registries and Action Agent adapters.

Return type:: SkillDescriptor

property device: device: Return the concrete runtime device associated with the engine.

failed_plan(request, context, *, message=None, failure_code='planning_failed', retryable=True)[source]

Build a failed empty plan without changing task state.

Parameters:

request (ResolvedActionRequest[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]) – Resolved invocation that failed to plan.
context (PlanningContext) – Planning input used for the attempt.
message (str | None) – Optional diagnostic message.
failure_code (str) – Stable machine-readable planning failure code.
retryable (bool) – Whether execution may spend action-retry budget on the failed rows.

Return type:

Returns:

Failed action plan with an empty trajectory.

property is_bound: bool: Whether an engine has supplied this action’s planning resources.

property motion_generator: MotionGenerator: Return the engine-owned motion generator borrowed by this action.

property num_envs: int: Number of environments owned by the bound robot.

open_loop: ClassVar[bool] = False: Whether the skill intentionally declares no verified physical effect.

plan(request, context)[source]

Bind the current collision scene and invoke the skill planner.

Parameters:

request (ResolvedActionRequest[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]) – Immutable, typed, and embodiment-resolved action request.
context (PlanningContext) – Latest observed robot, task, and scene state.

Return type:

Returns:

Scene-bound action plan with expected, uncommitted effects.

property planning_services: ActionPlanningServices

Return the engine-owned services borrowed by this action.

Raises:: RuntimeError – If the action has not been registered or planned by an AtomicActionEngine.

require_goal(request)[source]

Validate a resolved request and return its concrete goal.

Return type:: TypeVar(GoalT)

resolve_request(invocation)[source]

Validate and snapshot an invocation through engine-owned resources.

Parameters:

invocation (ActionInvocation[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]) – Caller-owned invocation to resolve.

Return type:

ResolvedActionRequest[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]

Returns:

Immutable request reused by planning and recovery replans.

Raises:

ValueError – If the stable skill identifier does not match.
TypeError – If the goal or options type is incompatible.
KeyError – If a required binding role is missing.

property robot: Robot: Return the robot associated with the owning engine.

property robot_dof: int: Number of full-robot degrees of freedom.

skill_id: ClassVar[str]: Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.AtomicActionEngine[source]

Own planning resources and coordinate side-effect-free atomic actions.

Methods:

`__init__`(motion_generator[, ...])	Initialize one engine and bind its built-in action implementations.
`bind_control_parts`(skill_id, endpoints, *[, ...])	Build an advanced direct-core binding from control-part names.
`compile`(invocations[, context])	Compile a static sequence of grounded invocations.
`initial_context`(*[, task, scene, timestamp, ...])	Capture the robot state needed to start offline compilation.
`make_invocation`(skill_id, goal, *[, ...])	Construct a grounded invocation while naming the skill only once.
`plan`(invocation[, context])	Plan one registered invocation through the engine-owned backend.
`plan_request`(request[, context])	Plan an already-resolved request without rebuilding its snapshot.
`register`(action, *[, replace])	Register one action instance using its descriptor.
`resolve`(invocation)	Resolve a registered invocation into an engine-owned snapshot.
`start`(invocations[, context, eligible_mask])	Start incremental execution for a grounded invocation sequence.

Attributes:

`actions`	Registered action instances keyed by stable skill identifier.
`binding_owner_id`	Return the opaque owner identity required by action bindings.
`control_profiles`	Semantic command profiles registered for robot control parts.
`device`	Return the concrete planning device used by this engine.
`grasp_pose_generators`	Standalone grasp-pose services installed for endpoint targets.
`motion_generator`	Return the single motion generator owned by this engine.
`planning_services`	Engine-owned resources shared by every bound atomic action.
`robot`	Return the robot controlled by this engine.
`skill_catalog_revision`	Return the monotonic installed Atomic Skill catalog revision.
`skills`	Return explicitly declared, agent-visible installed skill metadata.
`tracking_runtime`	Typed endpoint-feedback runtime used by plans and sessions.

__init__(motion_generator, control_profiles=None, grasp_pose_generators=None, *, load_builtins=True, tracking_runtime=None, scene_provider=None)[source]

Initialize one engine and bind its built-in action implementations.

Parameters:

motion_generator (MotionGenerator) – Engine-owned motion-generation backend.
control_profiles (Optional[Mapping[str, ControlPartCommandProfile]]) – Semantic commands keyed by robot control-part name.
grasp_pose_generators (Optional[Mapping[str, GraspPoseGenerator]]) – Standalone grasp-pose services keyed by the runtime target ID of each grasp endpoint, normally its robot control-part name.
load_builtins (bool) – Whether to instantiate and register every built-in action. Disable this for isolated tests or fully custom engines.
tracking_runtime (TrackingRuntime | None) – Optional exact-version feedback, projector, and metric registries. Built-in joint tracking is installed when omitted.
scene_provider (SceneProvider | None) – Optional default scene-observation source used by initial_context() when the caller does not supply an explicit scene snapshot. The provider is borrowed by reference.

property actions: dict[str, AtomicAction]: Registered action instances keyed by stable skill identifier.

bind_control_parts(skill_id, endpoints, *, task_state_keys=None)[source]

Build an advanced direct-core binding from control-part names.

Parameters:

skill_id (str) – Installed skill ID.
endpoints (Mapping[str, Mapping[str, str]]) – Nested slot_id -> endpoint_id -> control_part mapping.
task_state_keys (Optional[Mapping[str, str]]) – Optional explicit stable task-state key for each resource slot. See ActionPlanningServices.bind_control_parts() for inference rules when omitted.

Return type:

ActionBinding

Returns:

Engine-owned generic endpoint binding.

property binding_owner_id: str: Return the opaque owner identity required by action bindings.

compile(invocations, context=None)[source]

Compile a static sequence of grounded invocations.

Planning is side-effect free. Expected effects are applied only to the returned hypothetical projected_context so following actions can be checked against the expected state. No simulator or observed task state is mutated.

Parameters:

invocations (Iterable[ActionInvocation]) – Grounded action requests in execution order.
context (PlanningContext | None) – Optional initial planning context captured by the caller.

Return type:

CompiledTrajectory

Returns:

Concatenated timed trajectory, individual plans, and projected state.

Raises:

KeyError – If an invocation references an unregistered skill.
ValueError – If context, plan, or trajectory dimensions are incompatible.

property control_profiles: Mapping[str, ControlPartCommandProfile]: Semantic command profiles registered for robot control parts.

property device: device: Return the concrete planning device used by this engine.

property grasp_pose_generators: Mapping[str, GraspPoseGenerator]: Standalone grasp-pose services installed for endpoint targets.

initial_context(*, task=None, scene=None, timestamp=0.0, control_dt=None)[source]

Capture the robot state needed to start offline compilation.

Parameters:

task (TaskState | None) – Optional symbolic task state; an empty state is used otherwise.
scene (SceneSnapshot | None) – Optional explicit scene snapshot. It overrides the engine’s configured scene provider; an empty snapshot is used when both are absent.
timestamp (float) – Timestamp assigned to the captured robot observation.
control_dt (float | None) – Explicit command period for action-owned interpolation.

Return type:

Returns:

Planning context containing owned robot tensors.

make_invocation(skill_id, goal, *, control_parts=None, motion_policy=None, tracking_policy=None, recovery_policy=None, skill_options=None, control_overrides=None, invocation_id=None, revision=0)[source]

Construct a grounded invocation while naming the skill only once.

control_parts uses the advanced direct-core binding path. Profile- based integrations resolve an ActionBinding in the semantic layer and construct ActionInvocation directly.

Parameters:

skill_id (str) – Stable identifier of an installed atomic skill.
goal (TypeVar(GoalT)) – Action-specific typed goal.
control_parts (Optional[Mapping[str, Mapping[str, str]]]) – Direct slot -> endpoint -> control_part mapping.
motion_policy (MotionPolicy | None) – Optional invocation motion policy.
tracking_policy (TrackingPolicy | None) – Optional typed tracking and terminal-acceptance policy.
recovery_policy (RecoveryPolicy | None) – Optional invocation recovery policy.
skill_options (Optional[TypeVar(OptionsT, bound= ActionOptions)]) – Optional action-specific invocation options.
control_overrides (ActionControlOverrides | None) – Optional endpoint-scoped command overrides.
invocation_id (str | None) – Optional correlation identifier.
revision (int) – Monotonic invocation revision.

Return type:

ActionInvocation[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]

Returns:

A standard ActionInvocation accepted by plan, compile, and start.

Raises:

ValueError – If control_parts is omitted.
KeyError – If the skill or control part is unknown.
TypeError – If an invocation field or binding input has an invalid type.

property motion_generator: MotionGenerator: Return the single motion generator owned by this engine.

plan(invocation, context=None)[source]

Plan one registered invocation through the engine-owned backend.

Parameters:

invocation (ActionInvocation) – Grounded request for a registered skill.
context (PlanningContext | None) – Optional latest planning state; captured when omitted.

Return type:

Returns:

Validated action plan.

Raises:

KeyError – If the invocation references an unregistered skill.

plan_request(request, context=None)[source]

Plan an already-resolved request without rebuilding its snapshot.

Return type:: ActionPlan

property planning_services: ActionPlanningServices: Engine-owned resources shared by every bound atomic action.

register(action, *, replace=False)[source]

Register one action instance using its descriptor.

Parameters:

action (AtomicAction) – Configured action instance.
replace (bool) – Whether to replace an implementation already registered under the same stable skill identifier. Replacement is always explicit so extensions cannot silently shadow built-ins.

Raises:

TypeError – If action is not an AtomicAction.
ValueError – If it belongs to another engine or its skill identifier conflicts with an existing action.

Return type:

None

resolve(invocation)[source]

Resolve a registered invocation into an engine-owned snapshot.

Return type:: ResolvedActionRequest

property robot: Robot: Return the robot controlled by this engine.

property skill_catalog_revision: int

Return the monotonic installed Atomic Skill catalog revision.

Installing or replacing an agent-visible implementation advances the revision even when its public descriptor is equal. External binding and compilation layers can therefore reject stale implementation snapshots.

property skills: Mapping[str, SkillDescriptor]

Return explicitly declared, agent-visible installed skill metadata.

Process-wide type discovery, engine installation, and semantic exposure are separate boundaries. Only an action installed in this engine whose concrete class explicitly declares a generic binding contract appears here. Direct-core callers may continue to use every entry in actions.

start(invocations, context=None, *, eligible_mask=None)[source]

Start incremental execution for a grounded invocation sequence.

Parameters:

invocations (Iterable[ActionInvocation]) – Grounded action requests in execution order.
context (PlanningContext | None) – Initial measured state and scene snapshot. The engine captures one when omitted.
eligible_mask (Tensor | None) – Optional per-environment cohort allowed to execute. Ineligible rows remain excluded for the whole session. All rows are eligible when omitted.

Return type:

ExecutionSession

Returns:

Stateful execution session advanced by session.tick(...).

property tracking_runtime: TrackingRuntime: Typed endpoint-feedback runtime used by plans and sessions.

class embodichain.lab.sim.atomic_actions.AxisAlign[source]

Grasp an object and align its local axis to a world axis.

Classes:

`GoalType`	alias of `AxisAlignGoal`
`OptionsType`	alias of `AxisAlignOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`open_loop`	Whether the skill intentionally declares no verified physical effect.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of AxisAlignGoal

OptionsType: alias of AxisAlignOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='primary', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'kinematics.forward', 'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'open': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>, 'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)),), constraints=())

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

open_loop: ClassVar[bool] = True: Whether the skill intentionally declares no verified physical effect.

skill_id: ClassVar[str] = 'axis_align': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.AxisAlignAffordance[source]

Antipodal grasp affordance with an object-local alignment axis.

Methods:

__init__([object_label, custom_config, ...])

Attributes:

internal_axis

Axis expressed in the target object's local frame.

__init__(object_label='', custom_config=<factory>, mesh_vertices=None, mesh_triangles=None, internal_axis=<factory>)

custom_config: dict[str, Any]: User-defined configuration payload.

internal_axis: Tensor: Axis expressed in the target object’s local frame.

class embodichain.lab.sim.atomic_actions.AxisAlignGoal[source]

Object whose local axis should be aligned after an antipodal grasp.

Methods:

__init__(semantics[, grasp_xpos])

Attributes:

grasp_xpos

Optional explicit end-effector grasp pose; omitted poses are sampled.

__init__(semantics, grasp_xpos=None)

grasp_xpos: PoseGoalValue | None: Optional explicit end-effector grasp pose; omitted poses are sampled.

class embodichain.lab.sim.atomic_actions.AxisAlignOptions[source]

Per-invocation grasp-and-axis-alignment behavior.

Methods:

__init__([hand_interp_steps, ...])

Attributes:

target_axis

Desired world-frame axis, shape (3,) or (B, 3).

__init__(hand_interp_steps=5, grasp_settle_steps=0, pick_object_part='center', lift_height=0.1, pre_grasp_distance=0.15, approach_direction=tensor([0., 0., -1.]), approach_alignment_max_angle=None, downstream_object_target_poses=(), obj_upright_direction=None, rotate_upright=None, grasp_frame_to_eef=tensor([[1., 0., 0., 0.], [0., 1., 0., 0.], [0., 0., 1., 0.], [0., 0., 0., 1.]]), fixed_object_to_eef=None, target_axis=tensor([0., 0., 1.]))

target_axis: Tensor: Desired world-frame axis, shape (3,) or (B, 3).

class embodichain.lab.sim.atomic_actions.CommandAckStatus[source]

Outcome reported by a command transport or controller.

Methods:

__new__(value)

__new__(value)

class embodichain.lab.sim.atomic_actions.CommandAcknowledgement[source]

Synchronous acknowledgement returned by a CommandSink.

Methods:

`__init__`(status[, message])
`accepted_ack`([message])	Build an accepted acknowledgement.

Attributes:

`accepted`	Whether the controller accepted the requested operation.
`message`	Human-readable diagnostic intended for logs, not policy branching.
`status`	Transport/controller acknowledgement status.

__init__(status, message='')

property accepted: bool: Whether the controller accepted the requested operation.

classmethod accepted_ack(message='')[source]

Build an accepted acknowledgement.

Parameters:: message (str) – Optional controller diagnostic.
Return type:: CommandAcknowledgement
Returns:: Accepted acknowledgement.

message: str: Human-readable diagnostic intended for logs, not policy branching.

status: CommandAckStatus: Transport/controller acknowledgement status.

class embodichain.lab.sim.atomic_actions.CommandDispatch[source]

Auditable record of one controller operation and acknowledgement.

Methods:

__init__(operation, acknowledgement)

__init__(operation, acknowledgement)

class embodichain.lab.sim.atomic_actions.CommandOperation[source]

Command-sink operation recorded by an execution runner.

Methods:

__new__(value)

__new__(value)

class embodichain.lab.sim.atomic_actions.CommandSink[source]

Controller boundary used by ExecutionRunner.

Methods:

`__init__`(args, *kwargs)
`cancel`(targets, *, timeout)	Cancel any controller-side command that has not completed.
`hold`(targets, context, *, timeout)	Apply transport-specific safe state to the supplied targets.
`send`(command, *, timeout)	Submit one synchronized endpoint-command frame.

__init__(*args, **kwargs)

cancel(targets, *, timeout)[source]

Cancel any controller-side command that has not completed.

Parameters:

targets (tuple[RuntimeEndpointTarget, ...]) – Runtime targets whose queued work must be cancelled.
timeout (float) – Maximum acknowledgement latency in seconds.

Return type:

Returns:

Transport or controller acknowledgement.

hold(targets, context, *, timeout)[source]

Apply transport-specific safe state to the supplied targets.

Parameters:

targets (tuple[RuntimeEndpointTarget, ...]) – Runtime targets that may retain controller state.
context (PlanningContext) – Latest observation used by position-hold transports.
timeout (float) – Maximum acknowledgement latency in seconds.

Return type:

Returns:

Transport or controller acknowledgement.

send(command, *, timeout)[source]

Submit one synchronized endpoint-command frame.

Parameters:

command (RuntimeCommandFrame) – Transport-neutral command frame with an active-row mask. The sink must actively neutralize inactive rows for every addressed target; omission is not a safe state for persistent controllers.
timeout (float) – Maximum acknowledgement latency in seconds.

Return type:

Returns:

Transport or controller acknowledgement.

class embodichain.lab.sim.atomic_actions.CompiledTrajectory[source]

Offline compilation result for a sequence of action invocations.

Methods:

`__init__`(plan_success, trajectory, ...)
`action_waypoint_offset`(action_index)	Return the global waypoint offset of one compiled action.
`segment`(action_index, name)	Return action segment metadata shifted into compiled coordinates.

__init__(plan_success, trajectory, action_plans, projected_context)

action_waypoint_offset(action_index)[source]

Return the global waypoint offset of one compiled action.

Return type:: int

segment(action_index, name)[source]

Return action segment metadata shifted into compiled coordinates.

Return type:: TrajectorySegment

class embodichain.lab.sim.atomic_actions.ControlCommand[source]

Immutable-by-ownership command associated with one control part.

Command subclasses own their payload and must return another owned value from snapshot(). This keeps engine profiles and resolved invocation requests isolated from caller-owned mutable tensors.

Methods:

`equivalent_to`(other)	Return whether `other` has exactly the same command semantics.
`snapshot`()	Return an independently owned copy of this command.

abstract equivalent_to(other)[source]

Return whether other has exactly the same command semantics.

Return type:: bool

abstract snapshot()[source]

Return an independently owned copy of this command.

Return type:: ControlCommand

class embodichain.lab.sim.atomic_actions.ControlPartCommandProfile[source]

Reusable semantic commands for one named robot control part.

Profiles are registered once on AtomicActionEngine, keyed by names from Robot.control_parts. They describe embodiment-specific meanings such as open, grasp or ready without coupling those values to an action implementation.

Methods:

`__init__`([commands])
`joint_positions`(**commands)	Build a profile whose entries are joint-position commands.
`snapshot`()	Return an independently owned profile snapshot.

__init__(commands=<factory>)

classmethod joint_positions(**commands)[source]

Build a profile whose entries are joint-position commands.

Return type:: ControlPartCommandProfile

snapshot()[source]

Return an independently owned profile snapshot.

Return type:: ControlPartCommandProfile

class embodichain.lab.sim.atomic_actions.CoordinatedHeldObjectState[source]

Observed or projected relation for an object held by two manipulators.

Methods:

__init__(semantics, left_object_to_eef, ...)

__init__(semantics, left_object_to_eef, right_object_to_eef, left_grasp_xpos, right_grasp_xpos, env_mask=None)

class embodichain.lab.sim.atomic_actions.CoordinatedPickGoal[source]

Object-centric target for picking and moving one object with two hands.

The left/right grasp poses are not supplied by the caller; they are sampled by the parallel-jaw grasp-pose service at planning time using the dual-arm direction and approach direction declared on CoordinatedPickmentOptions.

Methods:

__init__(semantics, object_target_pose[, ...])

Attributes:

`object_initial_pose`	Optional initial object pose.
`object_target_pose`	Target pose for the shared object, shape `(4, 4)` or `(num_envs, 4, 4)`.

__init__(semantics, object_target_pose, object_initial_pose=None)

object_initial_pose: PoseGoalValue | None

Optional initial object pose.

When omitted, the pose is grounded through the semantic object’s stable scene identity.

object_target_pose: PoseGoalValue: Target pose for the shared object, shape (4, 4) or (num_envs, 4, 4).

class embodichain.lab.sim.atomic_actions.CoordinatedPickment[source]

Pick and move a single object pinched by two hands.

Classes:

`GoalType`	alias of `CoordinatedPickGoal`
`OptionsType`	alias of `CoordinatedPickmentOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of CoordinatedPickGoal

OptionsType: alias of CoordinatedPickmentOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='left', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'kinematics.inverse'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'open': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>, 'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)), SkillResourceSlot(slot_id='right', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'kinematics.inverse'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'open': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>, 'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),))), constraints=(DisjointResourceSlots(slots=('left', 'right')),))

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

skill_id: ClassVar[str] = 'coordinated_pickment': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.CoordinatedPickmentOptions[source]

Per-invocation coordinated pickup behavior.

Left/right grasps are sampled from target-local affordance geometry by the engine’s parallel-jaw grasp-pose service. The dual-arm direction splits the object into left/right grasp regions and the approach direction filters the sampled antipodal pairs.

Methods:

__init__([object_motion_keyframes, ...])

Attributes:

`approach_direction`	World-frame direction used to sample and approach both grasps, shape `(3,)`.
`hand_interp_steps`	Number of waypoints used for the simultaneous hand-close segment.
`hold_steps`	Number of waypoints to hold the final object target pose.
`left_to_right_arm_direction`	World-frame direction from the left arm base to the right arm base, shape `(3,)`.
`lift_height`	World-Z lift distance before moving to the object target pose.
`middle_empty_ratio`	Fraction of the object's left-to-right extent left grasp-free in the middle so the two grippers pinch opposite ends.
`object_motion_keyframes`	Number of object-pose keyframes solved by IK before joint-space interpolation.
`pre_grasp_distance`	World distance to retreat from each grasp pose along negative TCP z.

__init__(object_motion_keyframes=6, pre_grasp_distance=0.1, lift_height=0.08, hand_interp_steps=10, hold_steps=4, approach_direction=tensor([0., 0., -1.]), left_to_right_arm_direction=tensor([1., 0., 0.]), middle_empty_ratio=0.4)

approach_direction: Tensor: World-frame direction used to sample and approach both grasps, shape (3,).

hand_interp_steps: int: Number of waypoints used for the simultaneous hand-close segment.

hold_steps: int: Number of waypoints to hold the final object target pose.

left_to_right_arm_direction: Tensor: World-frame direction from the left arm base to the right arm base, shape (3,). It partitions the object into left/right grasp regions and should be a finite, non-zero vector; it is normalized at planning time.

lift_height: float: World-Z lift distance before moving to the object target pose.

middle_empty_ratio: float: Fraction of the object’s left-to-right extent left grasp-free in the middle so the two grippers pinch opposite ends. Must be in [0, 1].

object_motion_keyframes: int: Number of object-pose keyframes solved by IK before joint-space interpolation.

pre_grasp_distance: float: World distance to retreat from each grasp pose along negative TCP z.

class embodichain.lab.sim.atomic_actions.CoordinatedPlacement[source]

Coordinate two held objects: support object below, placing object above.

Classes:

`GoalType`	alias of `CoordinatedPlacementGoal`
`OptionsType`	alias of `CoordinatedPlacementOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of CoordinatedPlacementGoal

OptionsType: alias of CoordinatedPlacementOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='placing', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'open': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>, 'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)), SkillResourceSlot(slot_id='support', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),))), constraints=(DisjointResourceSlots(slots=('placing', 'support')),))

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

skill_id: ClassVar[str] = 'coordinated_placement': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.CoordinatedPlacementGoal[source]

Object-centric target for dual-arm coordinated placement.

Methods:

__init__(placing_object_target_pose, ...[, ...])

Attributes:

`placing_height_offset`	World-Z offset above the placing object target pose.
`placing_object_target_pose`	Target pose for the object released by the placing arm.
`release`	Whether the placing hand releases.
`support_height_offset`	World-Z offset above the support object target pose.
`support_object_target_pose`	Target pose for the object held by the support arm.

__init__(placing_object_target_pose, support_object_target_pose, placing_height_offset=None, support_height_offset=None, release=None)

placing_height_offset: float | None: World-Z offset above the placing object target pose.

placing_object_target_pose: Tensor | SceneEntityPose: Target pose for the object released by the placing arm.

release: bool | None: Whether the placing hand releases. None uses invocation options.

support_height_offset: float | None: World-Z offset above the support object target pose.

support_object_target_pose: Tensor | SceneEntityPose: Target pose for the object held by the support arm.

class embodichain.lab.sim.atomic_actions.CoordinatedPlacementOptions[source]

Per-invocation coordinated placement behavior.

Methods:

__init__([release, placing_height_offset, ...])

Attributes:

`hand_interp_steps`	Number of waypoints for the placing-hand release interpolation.
`hold_steps`	Number of waypoints to hold alignment before releasing.
`lift_height`	World-Z lift distance for the placing arm after release.
`placing_height_offset`	Default World-Z offset above the placing object target pose.
`release`	Whether to open the placing hand at the aligned placement pose.
`retreat_steps`	Number of waypoints used for the placing-arm lift retreat.
`support_height_offset`	Default World-Z offset above the support object target pose.

__init__(release=True, placing_height_offset=0.0, support_height_offset=0.0, lift_height=0.08, hand_interp_steps=10, hold_steps=4, retreat_steps=16)

hand_interp_steps: int: Number of waypoints for the placing-hand release interpolation.

hold_steps: int: Number of waypoints to hold alignment before releasing.

lift_height: float: World-Z lift distance for the placing arm after release.

placing_height_offset: float: Default World-Z offset above the placing object target pose.

release: bool: Whether to open the placing hand at the aligned placement pose.

retreat_steps: int: Number of waypoints used for the placing-arm lift retreat.

support_height_offset: float: Default World-Z offset above the support object target pose.

class embodichain.lab.sim.atomic_actions.DisjointResourceSlots[source]

Require selected slots to have pairwise-disjoint physical claims.

Methods:

__init__(slots)

__init__(slots)

class embodichain.lab.sim.atomic_actions.DisjointSlotEndpoints[source]

Require selected endpoints within one participant to be disjoint.

Methods:

__init__(endpoint_ids)

__init__(endpoint_ids)

class embodichain.lab.sim.atomic_actions.DynamicCollisionMode[source]

Policy for consuming a live dynamic collision world.

This mode controls scene-snapshot obstacle binding and collision-world revision recovery. It does not enable or disable a planner’s configured static-world or self-collision checks.

Attributes:

`AUTO`	Use live collision entities when the selected motion strategy supports them.
`OFF`	Ignore scene-snapshot collision entities and their revisions.
`REQUIRED`	Require live collision entities and a compatible motion planner.

Methods:

__new__(value)

AUTO = 'auto': Use live collision entities when the selected motion strategy supports them.

OFF = 'off': Ignore scene-snapshot collision entities and their revisions.

REQUIRED = 'required': Require live collision entities and a compatible motion planner.

__new__(value)

class embodichain.lab.sim.atomic_actions.EffectExpectationResult[source]

Current-observation outcome for one physical state expectation.

inverse_satisfied_mask is stronger than contradiction: every clause must have reached its explicit inverse band for the monitor’s complete hysteresis window. It may therefore be used to retain a pre-existing relation during failure reconciliation, while a single contradictory clause may not.

Methods:

`__init__`(expectation_id, satisfied_mask, ...)
`snapshot`()	Return an independently owned expectation outcome.

__init__(expectation_id, satisfied_mask, contradicted_mask, inverse_satisfied_mask)

snapshot()[source]

Return an independently owned expectation outcome.

Return type:: EffectExpectationResult

class embodichain.lab.sim.atomic_actions.EffectVerificationRequest[source]

Typed boundary describing a physical effect awaiting verification.

requested_at and deadline use the same timestamp domain as RobotObservation. Request-mask shrinkage retains both values; only a newly installed plan starts a new attempt deadline. attempt_generation is session-local and remains stable when partial resolution or row deactivation replaces only the request ID. failure_invalidation is a core-owned removal-only delta; verification results may select failed rows on which to apply it but cannot replace it.

Methods:

`__init__`(verification_id, skill_id, ...[, ...])
`snapshot`()	Return a request snapshot with an independently owned row mask.

__init__(verification_id, skill_id, invocation_id, invocation_revision, invocation_index, attempt_generation, terminal_segment, requested_at, deadline, env_mask, expected_effects, effect_verification=None, failure_invalidation=<factory>)

snapshot()[source]

Return a request snapshot with an independently owned row mask.

Return type:: EffectVerificationRequest

class embodichain.lab.sim.atomic_actions.EffectVerificationRequirement[source]

Explicit physical-effect verification independent of symbolic state.

Presence of this value on an ActionPlan forces a terminal effect boundary even when the plan declares no StateDelta. The open kind identifier lets an external runtime select an appropriate verifier without placing backend-specific callbacks in the core plan.

Parameters:: kind (str) – Stable, non-empty discriminator for the physical effect.

Methods:

`__init__`(kind)
`snapshot`()	Return an independently owned requirement value.

__init__(kind)

snapshot()[source]

Return an independently owned requirement value.

Return type:: EffectVerificationRequirement

class embodichain.lab.sim.atomic_actions.EffectVerificationResult[source]

Correlated per-environment update for one effect boundary.

Rows absent from both success_mask and failure_mask remain unresolved. invalidation_mask and retry_mask classify only failed rows: the former selects the request’s core-owned removal delta, while the latter authorizes replay of the same invocation. Failed rows outside the retry mask require external recovery. This lets one shared batch barrier commit verified rows while other rows continue observing the same physical effect.

Methods:

__init__(verification_id, success_mask, ...)

__init__(verification_id, success_mask, failure_mask, invalidation_mask, retry_mask, expectation_results=())

class embodichain.lab.sim.atomic_actions.EndEffectorPoseGoal[source]

End-effector pose goal with optional batched intermediate waypoints.

Methods:

__init__(xpos)

Attributes:

xpos

Homogeneous pose with shape (4,4), (B,4,4) or (B,N,4,4).

__init__(xpos)

xpos: Tensor | SceneEntityPose: Homogeneous pose with shape (4,4), (B,4,4) or (B,N,4,4).

class embodichain.lab.sim.atomic_actions.EndpointBinding[source]

One action-local endpoint resolved to a runtime controller target.

Methods:

`__init__`(slot_id, endpoint_id, resource_id, ...)
`command`(name)	Return one owned semantic-command snapshot.
`joint_positions`(name, *, num_envs, device[, ...])	Resolve a named joint-position command for a planning batch.
`require_target`(target_type)	Return the runtime target after an explicit type check.
`snapshot`()	Return an independently owned endpoint-binding snapshot.
`tracking_channel`(channel_id)	Return one independently owned typed tracking-channel binding.
`with_commands`(overrides)	Return an endpoint snapshot with semantic-command overrides.

Attributes:

`destination_key`	Return the transport-scoped physical destination key.
`key`	Return the action-local `(slot, endpoint)` key.
`task_state_key`	Symbolic task-state key; defaults to `target.target_id`.

__init__(slot_id, endpoint_id, resource_id, adapter_id, target, task_state_key=None, tracking_channels=<factory>, capabilities=frozenset({}), commands=<factory>, claim_tokens=frozenset({}), joint_ids=())

command(name)[source]

Return one owned semantic-command snapshot.

Return type:: ControlCommand

property destination_key: tuple[str, str]: Return the transport-scoped physical destination key.

joint_positions(name, *, num_envs, device, dtype=None)[source]

Resolve a named joint-position command for a planning batch.

Return type:: Tensor

property key: tuple[str, str]: Return the action-local (slot, endpoint) key.

require_target(target_type)[source]

Return the runtime target after an explicit type check.

Return type:: TypeVar(TargetT, bound= RuntimeEndpointTarget)

snapshot()[source]

Return an independently owned endpoint-binding snapshot.

Return type:: EndpointBinding

task_state_key: str | None: Symbolic task-state key; defaults to target.target_id.

tracking_channel(channel_id)[source]

Return one independently owned typed tracking-channel binding.

Return type:: EndpointTrackingChannelBinding

with_commands(overrides)[source]

Return an endpoint snapshot with semantic-command overrides.

Return type:: EndpointBinding

class embodichain.lab.sim.atomic_actions.EndpointCommand[source]

One transport-compatible payload addressed to one runtime target.

Parameters:

target (RuntimeEndpointTarget) – Immutable destination resolved from an action endpoint.
payload (RuntimeCommandPayload) – Batched command value accepted by the target transport.

Methods:

`__init__`(target, payload)
`snapshot`()	Return an independently owned endpoint command.

Attributes:

`batch_size`	Return the payload batch size.
`destination_key`	Return the transport-scoped destination identifier.
`device`	Return the payload device.
`transport_id`	Return the common target and payload transport identifier.

__init__(target, payload)

property batch_size: int: Return the payload batch size.

property destination_key: tuple[str, str]: Return the transport-scoped destination identifier.

property device: device: Return the payload device.

snapshot()[source]

Return an independently owned endpoint command.

Return type:: EndpointCommand

property transport_id: str: Return the common target and payload transport identifier.

class embodichain.lab.sim.atomic_actions.EndpointCommandRouter[source]

Route generic endpoint operations to exact registered transports.

The router implements CommandSink structurally while avoiding a module-load dependency on runner. Acknowledgement types are imported only when an operation is executed, which keeps the transport boundary safe to import while the runner imports this module.

Parameters:

transports (Mapping[str, EndpointCommandTransport] | Iterable[EndpointCommandTransport]) – Either an exact transport_id -> transport mapping or an iterable of transports from which that mapping is built. Mapping keys must exactly equal each value’s declared transport_id.

Raises:

TypeError – If a registration does not implement the transport contract.
ValueError – If an identifier is invalid, a mapping key is not exact, or the same transport identifier is registered more than once.

Methods:

`__init__`(transports)
`cancel`(targets, *, timeout)	Route cancellation by target transport.
`hold`(targets, context, *, timeout)	Route an observed-state hold request by target transport.
`send`(frame, *, timeout)	Route one synchronized command frame by transport identifier.

Attributes:

transports

Return the immutable exact transport registry.

__init__(transports)[source]

cancel(targets, *, timeout)[source]

Route cancellation by target transport.

Parameters:

targets (tuple[RuntimeEndpointTarget, ...]) – Runtime destinations whose outstanding commands are cancelled.
timeout (float) – Maximum acknowledgement latency for each transport.

Return type:

Returns:

Aggregated acknowledgement. It is accepted only when every addressed transport accepts cancellation.

hold(targets, context, *, timeout)[source]

Route an observed-state hold request by target transport.

Parameters:

targets (tuple[RuntimeEndpointTarget, ...]) – Runtime destinations to hold.
context (PlanningContext) – Fresh observation used by each transport to form its hold.
timeout (float) – Maximum acknowledgement latency for each transport.

Return type:

Returns:

Aggregated acknowledgement. It is accepted only when every addressed transport accepts its hold.

send(frame, *, timeout)[source]

Route one synchronized command frame by transport identifier.

Dispatch is preflighted before any transport is called. An unknown transport or incompatible payload therefore rejects the whole frame without creating a partially dispatched operation.

Parameters:

frame (RuntimeCommandFrame) – Generic runtime command frame to split by transport.
timeout (float) – Maximum acknowledgement latency for each transport.

Return type:

Returns:

Aggregated acknowledgement. It is accepted only when every addressed transport accepts its local frame.

property transports: Mapping[str, EndpointCommandTransport]: Return the immutable exact transport registry.

class embodichain.lab.sim.atomic_actions.EndpointCommandTransport[source]

Backend that owns one kind of runtime endpoint command.

Implementations own live simulator entities, device clients, or controller handles. Runtime command values retain only immutable addressing and payload data, so they remain independent of those process-owned resources.

Methods:

`__init__`(args, *kwargs)
`cancel`(targets, *, timeout)	Cancel outstanding commands for transport-local targets.
`hold`(targets, context, *, timeout)	Hold transport-local targets at their observed state.
`send`(frame, *, timeout)	Submit one transport-local command frame.

Attributes:

`payload_type`	Return the runtime payload type accepted by `send()`.
`transport_id`	Return the exact identifier used to register this transport.

__init__(*args, **kwargs)

cancel(targets, *, timeout)[source]

Cancel outstanding commands for transport-local targets.

Return type:: CommandAcknowledgement

hold(targets, context, *, timeout)[source]

Hold transport-local targets at their observed state.

Return type:: CommandAcknowledgement

property payload_type: type[RuntimeCommandPayload]: Return the runtime payload type accepted by send().

send(frame, *, timeout)[source]

Submit one transport-local command frame.

Implementations must actively neutralize every inactive environment row for every addressed target. Silently skipping an inactive row is unsafe for persistent controllers such as base-velocity transports.

Return type:: CommandAcknowledgement

property transport_id: str: Return the exact identifier used to register this transport.

class embodichain.lab.sim.atomic_actions.EndpointTrackingChannelBinding[source]

Resolved source and projector for one endpoint tracking channel.

Methods:

`__init__`(channel_id, source, projector)
`snapshot`()	Return an independently owned channel binding.

Attributes:

route_fingerprint

Return the exact channel, source, and projector route identity.

__init__(channel_id, source, projector)

property route_fingerprint: tuple[str, Hashable, str, str]: Return the exact channel, source, and projector route identity.

snapshot()[source]

Return an independently owned channel binding.

Return type:: EndpointTrackingChannelBinding

class embodichain.lab.sim.atomic_actions.EndpointTrackingFeedbackAddress[source]

Feedback address for one runtime endpoint and open tracking channel.

Methods:

__init__(target, channel_id)

Attributes:

address_fingerprint

Return the endpoint- and channel-scoped address identity.

__init__(target, channel_id)

property address_fingerprint: Hashable: Return the endpoint- and channel-scoped address identity.

class embodichain.lab.sim.atomic_actions.EntityState[source]

Scene entity state addressable by a stable entity identifier.

Methods:

__init__(pose[, confidence])

__init__(pose, confidence=1.0)

class embodichain.lab.sim.atomic_actions.ExecutionClock[source]

Clock abstraction used for deterministic and simulation scheduling.

Methods:

`__init__`(args, *kwargs)
`now`()	Return a monotonic timestamp in seconds.
`sleep`(duration)	Wait or advance the execution backend by `duration` seconds.

__init__(*args, **kwargs)

now()[source]

Return a monotonic timestamp in seconds.

Return type:: float
Returns:: Monotonic timestamp in seconds.

sleep(duration)[source]

Wait or advance the execution backend by duration seconds.

Parameters:: duration (float) – Non-negative duration in seconds.
Return type:: None

class embodichain.lab.sim.atomic_actions.ExecutionEvent[source]

One timestamped execution or recovery event.

Methods:

__init__(kind, timestamp, skill_id, ...[, ...])

__init__(kind, timestamp, skill_id, invocation_id, invocation_revision, invocation_index, env_mask, message='', segment_name=None, failure_code=None, retryable=None)

class embodichain.lab.sim.atomic_actions.ExecutionEventKind[source]

Structured event categories emitted by ExecutionSession.tick().

Methods:

__new__(value)

__new__(value)

class embodichain.lab.sim.atomic_actions.ExecutionPlanAttempt[source]

Owned inspection snapshot for one installed action plan.

Recovery can install several plans for one logical invocation. This value preserves the exact scene/collision revisions and trajectory structure of every installation, correlated with the session-local attempt generation and row-local recovery counters.

Methods:

`__init__`(attempt_generation, event_kind, ...)
`snapshot`()	Return an independently owned plan-attempt trace.

__init__(attempt_generation, event_kind, planned_at, invocation_index, planned_mask, action_retry_counts, replan_counts, request, plan)

snapshot()[source]

Return an independently owned plan-attempt trace.

Return type:: ExecutionPlanAttempt

class embodichain.lab.sim.atomic_actions.ExecutionRunner[source]

Connect an execution session to observation, controller, and time ports.

step() is non-blocking. It observes and advances the session only when the next command is due according to RuntimeCommandFrame.hold_duration. run_until_blocked() supplies the blocking loop for tutorials and simple applications. Controller rejection, timeout, observation failure, and session exceptions all trigger a best-effort cancel-then-hold sequence. Runner methods are designed for serialized event-loop use and are not thread-safe.

Parameters:

session (ExecutionSession) – Stateful atomic-action execution session.
observation_provider (ObservationProvider) – Source of fresh robot and scene observations.
command_sink (CommandSink) – Controller or simulation command boundary.
clock (ExecutionClock | None) – Optional scheduler clock. Defaults to monotonic wall time.
cfg (ExecutionRunnerCfg | None) – Optional acknowledgement, scheduling, and completion policy.

Methods:

`__init__`(session, observation_provider, ...)
`cancel`([reason])	Cancel controller work and hold the latest observed position.
`deactivate_rows`(env_mask, *, reason)	Permanently deactivate environment rows owned by this runner.
`revise_current`(invocation)	Stage a newer revision for the next scheduled observation boundary.
`run_until_blocked`(*[, effect_verifier, ...])	Run with clock-driven waiting until terminal or effect verification blocks.
`step`(*[, effect_result, effect_verifier, ...])	Perform one due observation/session/controller update without sleeping.

Attributes:

`command_count`	Number of active commands accepted by the sink.
`effect_verification_pending`	Whether execution is waiting for an external semantic-effect result.
`session`	Execution session advanced by this runner.
`status`	Current runner lifecycle status.

__init__(session, observation_provider, command_sink, *, clock=None, cfg=None)[source]

cancel(reason='Execution cancelled by caller.')[source]

Cancel controller work and hold the latest observed position.

Parameters:: reason (str) – Human-readable cancellation reason.
Return type:: RunnerStep
Returns:: Terminal runner step. The status is cancelled only when both cancel and hold are acknowledged; otherwise it is failed.

property command_count: int: Number of active commands accepted by the sink.

deactivate_rows(env_mask, *, reason)[source]

Permanently deactivate environment rows owned by this runner.

The runner refreshes its cached effect boundary so a verifier cannot submit a result correlated with a request that deactivation replaced. In-flight controller work is neutralized for those rows by the next due command frame according to the CommandSink contract.

Parameters:

env_mask (Tensor) – Rows requested for deactivation.
reason (str) – Human-readable event message.

Return type:

Tensor

Returns:

Owned mask of rows that changed from eligible to inactive.

Raises:

RuntimeError – If the runner is already terminal.
TypeError – If env_mask is not a tensor.
ValueError – If the mask or reason is invalid.

property effect_verification_pending: bool: Whether execution is waiting for an external semantic-effect result.

revise_current(invocation)[source]

Stage a newer revision for the next scheduled observation boundary.

Staging preserves the active frame deadline. When that deadline is due, step() observes fresh state, atomically plans and installs the replacement, and dispatches its first command. The submitted invocation is resolved into an owned snapshot immediately, so later caller mutation cannot alter the staged revision.

Parameters:

invocation (ActionInvocation) – Strictly newer revision of the active logical call.

Raises:

TypeError – If invocation is not an ActionInvocation.
RuntimeError – If this runner or its session is no longer running, or if a physical effect is awaiting verification.
ValueError – If session-level revision invariants are violated.

Return type:

None

run_until_blocked(*, effect_verifier=None, phase_effect_gate_verifier=None, held_object_guard_verifier=None, on_step=None, max_steps=100000)[source]

Run with clock-driven waiting until terminal or effect verification blocks.

Parameters:

effect_verifier (Callable[[PlanningContext, EffectVerificationRequest], EffectVerificationResult] | None) – Optional synchronous callback used on fresh due-cycle observations while effect verification is pending. Without one, the method returns the running boundary so the caller can verify externally.
phase_effect_gate_verifier (Callable[[PlanningContext, PhaseEffectGateRequest], PhaseEffectGateResult] | None) – Optional synchronous callback used on fresh observations while a trajectory-segment entry is gated.
held_object_guard_verifier (Callable[[PlanningContext, HeldObjectGuardRequest], HeldObjectGuardResult | None] | None) – Optional synchronous phase-aware held-object verifier used before every due command cycle.
on_step (Callable[[RunnerStep], None] | None) – Optional callback for tracing or tutorial visualization.
max_steps (int) – Hard bound on loop iterations.

Return type:

Returns:

Terminal step, or a running step blocked on external verification.

property session: ExecutionSession

Execution session advanced by this runner.

Call revise_current() or deactivate_rows() on the runner, rather than mutating the session directly, while this runner owns scheduling.

property status: RunnerStatus: Current runner lifecycle status.

step(*, effect_result=None, effect_verifier=None, phase_effect_gate_result=None, phase_effect_gate_verifier=None, held_object_guard_verifier=None)[source]

Perform one due observation/session/controller update without sleeping.

Parameters:

effect_result (EffectVerificationResult | None) – Optional correlated effect result. If this call occurs before the next cycle is due, it is not consumed and must be supplied again on a later call.
effect_verifier (Callable[[PlanningContext, EffectVerificationRequest], EffectVerificationResult] | None) – Optional synchronous verifier for the current pending request. It runs after a fresh due-cycle observation and before the session consumes the result. It is not called after the request deadline. Mutually exclusive with effect_result.
phase_effect_gate_result (PhaseEffectGateResult | None) – Optional externally produced result for the current blocking trajectory-segment entry gate.
phase_effect_gate_verifier (Callable[[PlanningContext, PhaseEffectGateRequest], PhaseEffectGateResult] | None) – Optional synchronous verifier for the current gate. It runs on a fresh due-cycle observation and is mutually exclusive with phase_effect_gate_result.
held_object_guard_verifier (Callable[[PlanningContext, HeldObjectGuardRequest], HeldObjectGuardResult | None] | None) – Optional synchronous phase-aware verifier. It receives a fresh observation and the current command-phase request before ExecutionSession.tick() and command dispatch. Returning None means the current phase has no applicable held-object guard.

Return type:

Returns:

Runner status, optional session tick, controller acknowledgements, and time remaining before another update is due.

class embodichain.lab.sim.atomic_actions.ExecutionRunnerCfg[source]

Transport and scheduling policy for an ExecutionRunner.

Methods:

`__init__`([command_timeout, ...])
`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:

`command_timeout`	Maximum time allowed for a command acknowledgement.
`hold_during_effect_verification`	Whether to hold observed state while terminal effects are pending.
`hold_on_completion`	Whether to issue a final hold after the session completes.
`minimum_cycle_time`	Minimum delay between feedback cycles, including passive hold cycles.
`safe_stop_timeout`	Maximum time allowed for each cancel or hold acknowledgement.

__init__(command_timeout=<factory>, safe_stop_timeout=<factory>, minimum_cycle_time=<factory>, hold_on_completion=<factory>, hold_during_effect_verification=<factory>)

command_timeout: float: Maximum time allowed for a command acknowledgement.

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.

hold_during_effect_verification: bool

Whether to hold observed state while terminal effects are pending.

Disable this only for persistent transports whose last accepted command remains active without refresh, such as a position-controlled gripper that must retain contact preload. Failure and cancellation still perform the normal cancel-then-observed-hold safe stop.

hold_on_completion: bool: Whether to issue a final hold after the session completes.

minimum_cycle_time: float: Minimum delay between feedback cycles, including passive hold cycles.

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.

safe_stop_timeout: float: Maximum time allowed for each cancel or hold acknowledgement.

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.lab.sim.atomic_actions.ExecutionSession[source]

Execute grounded invocations incrementally with bounded local recovery.

The session never steps a simulator itself. Each tick() consumes the latest observation and scene snapshot and emits at most one synchronized endpoint-command frame. A declared physical-effect boundary resolves only after the caller supplies a correlated EffectVerificationResult, and non-empty expected symbolic effects are committed for accepted rows only. Higher-level runtimes decide how to produce that result from their configured monitor selection.

Environment eligibility and recovery budgets are tracked per row. The waypoint cursor is batch-synchronized: a recoverable row replans the active cohort from the latest observation and restarts the action trajectory. Calls that mutate the session must be serialized by its owner; the session does not provide thread synchronization.

Methods:

`__init__`(engine, invocations, context, *[, ...])
`deactivate_rows`(env_mask, *, reason)	Permanently remove selected rows from this invocation sequence.
`revise_current`(invocation, *[, context])	Replace and replan the current invocation with a newer revision.
`tick`(context, *[, effect_result, ...])	Advance execution by one observation/command cycle.
`trajectory_segment`(name)	Return named segment metadata for the active action plan.

Attributes:

`active_commands`	Return an owned snapshot of the active action command sequence.
`active_plan`	Return an independently owned snapshot of the active action plan.
`effect_verification_pending`	Whether the current physical effect still requires verification.
`eligible_mask`	Rows still eligible to complete the full invocation sequence.
`held_object_guard_request`	Describe the phase that must be checked before the next command.
`latest_context`	Latest validated context with the session's verified task state.
`pending_effect`	Owned snapshot of the current effect boundary, when present.
`phase_effect_gate_request`	Return the blocking gate at the next trajectory-segment entry.
`plan_attempts`	Return every installed plan in deterministic recovery order.
`status`	Current session status.
`task_state`	Verified symbolic task state accumulated by this session.

__init__(engine, invocations, context, *, eligible_mask=None)[source]

property active_commands: TimedCommandSequence

Return an owned snapshot of the active action command sequence.

This inspection surface is intended for diagnostics and visualization. Mutating the returned tensors cannot affect execution state.

property active_plan: ActionPlan

Return an independently owned snapshot of the active action plan.

This is a read-only diagnostics boundary for runtime metadata, visualization, and tests. Planning and recovery remain session-owned; mutating any tensor in the returned value cannot affect execution.

deactivate_rows(env_mask, *, reason)[source]

Permanently remove selected rows from this invocation sequence.

Deactivation is sticky across action barriers and recovery replans. The next emitted command frame marks those rows inactive so the command sink can apply target-specific safe hold behavior.

Parameters:

env_mask (Tensor) – Rows requested for deactivation.
reason (str) – Human-readable event message.

Return type:

Tensor

Returns:

Owned mask of rows that changed from eligible to inactive.

Raises:

RuntimeError – If the session is already terminal.
ValueError – If reason is empty or the mask shape is invalid.

property effect_verification_pending: bool: Whether the current physical effect still requires verification.

property eligible_mask: Tensor

Rows still eligible to complete the full invocation sequence.

This is deliberately not named success_mask: while the session is running, eligibility does not imply that execution or semantic effects have succeeded.

property held_object_guard_request: HeldObjectGuardRequest | None

Describe the phase that must be checked before the next command.

The request remains available while terminal acceptance is settling, using the final waypoint and segment identity. Once terminal physical effect verification begins, that verifier owns the boundary and this property returns None.

Returns:: Owned phase-aware guard request, or None when no command-phase guard is active.

property latest_context: PlanningContext: Latest validated context with the session’s verified task state.

property pending_effect: EffectVerificationRequest | None: Owned snapshot of the current effect boundary, when present.

property phase_effect_gate_request: PhaseEffectGateRequest | None

Return the blocking gate at the next trajectory-segment entry.

Returns:: Owned request snapshot, or None when the next command is not blocked by a physical-effect gate.

property plan_attempts: tuple[ExecutionPlanAttempt, ...]

Return every installed plan in deterministic recovery order.

The initial plan has generation zero. Each invocation revision, recovery replan, or whole-action retry appends a new generation instead of replacing earlier scene/collision evidence.

revise_current(invocation, *, context=None)[source]

Replace and replan the current invocation with a newer revision.

The replacement is resolved into a new immutable request snapshot from context or the session’s latest observation. Retry and replan budgets restart for the new revision, while verified task state, the current batch barrier, and per-environment eligibility are preserved. Ordinary recovery replans continue to reuse this snapshot until another explicit revision. Once the action owns runtime destinations, the replacement must preserve their exact address fingerprints; changing controllers or safe-hold footprints requires a new invocation.

Parameters:

invocation (ActionInvocation) – Grounded replacement for the currently active skill. Its revision must be strictly greater than the active one, and its skill_id and invocation_id must identify the same logical call.
context (PlanningContext | None) – Optional fresh observation used to ground the replacement. A manually ticked caller may omit it to reuse latest_context. Runner-driven code stages revisions on ExecutionRunner, which supplies a due-time observation.

Raises:

TypeError – If invocation is not an ActionInvocation.
RuntimeError – If the session is no longer running or a physical effect is awaiting verification.
ValueError – If the replacement identifies another invocation or does not advance the revision, or if its plan changes the active runtime target addresses.

Return type:

None

property status: ExecutionStatus: Current session status.

property task_state: TaskState: Verified symbolic task state accumulated by this session.

tick(context, *, effect_result=None, phase_effect_gate_result=None, held_object_guard_result=None)[source]

Advance execution by one observation/command cycle.

Parameters:

context (PlanningContext) – Latest measured robot and versioned scene state. Its task state is replaced by the session’s verified task state.
effect_result (EffectVerificationResult | None) – Optional correlated semantic-effect result for an action waiting at its terminal waypoint.
phase_effect_gate_result (PhaseEffectGateResult | None) – Optional correlated physical-effect decision for a blocked trajectory-segment entry.
held_object_guard_result (HeldObjectGuardResult | None) – Optional correlated in-flight held-object loss result for the current waypoint phase. None means the verifier found no applicable guard for this phase or no result was supplied.

Return type:

ExecutionTick

Returns:

Status, optional command, events, and current verified task state.

trajectory_segment(name)[source]

Return named segment metadata for the active action plan.

Segment ranges are action-local and may change after a replan when a backend preserves its own sample count.

Return type:: TrajectorySegment

class embodichain.lab.sim.atomic_actions.ExecutionStatus[source]

Lifecycle status of an execution session.

Methods:

__new__(value)

__new__(value)

class embodichain.lab.sim.atomic_actions.ExecutionTick[source]

Result returned after one closed-loop execution update.

Methods:

__init__(status, eligible_mask, command, ...)

__init__(status, eligible_mask, command, hold_targets, events, task_state, pending_effect=None, pending_phase_effect_gate=None)

class embodichain.lab.sim.atomic_actions.FeedbackTerminalAcceptance[source]

Terminal acceptance proven by typed endpoint feedback.

Methods:

__init__(metrics[, settle_timeout, ...])

__init__(metrics, settle_timeout=0.0, consecutive_acceptances=1)

class embodichain.lab.sim.atomic_actions.GraspGoal[source]

Pickup target with an affordance-selected or supplied grasp pose.

Methods:

__init__(semantics[, grasp_xpos])

Attributes:

grasp_xpos

Optional end-effector grasp pose.

__init__(semantics, grasp_xpos=None)

grasp_xpos: PoseGoalValue | None

Optional end-effector grasp pose.

When omitted, PickUp uses the configured fixed object-relative grasp when available, otherwise it selects one from the target affordance. An explicit tensor or late-bound SceneEntityPose skips grasp sampling. Late-bound poses also declare the scene dependency used by closed-loop execution recovery.

class embodichain.lab.sim.atomic_actions.HandOver[source]

Pick an object with the nearer arm, hand it over, and place it.

For each environment, the action chooses the arm whose root link is closer to the observed object pose. It samples at most 1000 mesh-surface points and applies SVD in the current object pose to find obj_longest_axis. When that axis is closer to world Z than to the horizontal plane, both grasp approaches point toward the object horizontally and tilt downward by 45 degrees. Otherwise both approaches are world-Z downward.

The first arm grasps the projected end of obj_longest_axis nearest its current TCP; the receiving arm grasps the opposite end at the predicted middle object pose. This keeps the two hands from selecting the same object region regardless of whether a long object is standing or lying down.

After each grasp waypoint, subsequent EEF waypoints preserve that grasp rotation and change translation only. In particular, placement first moves strictly horizontally at the handover height and then lowers to the final target pose before releasing the object.

Classes:

`GoalType`	alias of `HandOverGoal`
`OptionsType`	alias of `HandOverOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`open_loop`	Whether the skill intentionally declares no verified physical effect.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of HandOverGoal

OptionsType: alias of HandOverOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='source', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'kinematics.forward', 'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'open': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>, 'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)), SkillResourceSlot(slot_id='destination', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'kinematics.forward', 'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'open': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>, 'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),))), constraints=(DisjointResourceSlots(slots=('source', 'destination')),))

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

open_loop: ClassVar[bool] = True: Whether the skill intentionally declares no verified physical effect.

skill_id: ClassVar[str] = 'hand_over': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.HandOverGoal[source]

Object to pick and hand over, plus its final object pose.

Methods:

__init__(semantics, target_pose)

Attributes:

target_pose

Final object pose after the receiving arm lowers and releases it.

__init__(semantics, target_pose)

target_pose: PoseGoalValue: Final object pose after the receiving arm lowers and releases it.

class embodichain.lab.sim.atomic_actions.HandOverOptions[source]

Per-invocation pick-up, handover, and placement behavior.

Methods:

__init__([pre_grasp_distance, lift_height, ...])

Attributes:

`hand_interp_steps`	Waypoints used by every gripper open/close interpolation.
`lift_height`	World-Z distance used to lift the object after the first grasp.
`pre_grasp_distance`	Distance from each grasp pose to its approach pose, in metres.

__init__(pre_grasp_distance=0.1, lift_height=0.1, hand_interp_steps=10)

hand_interp_steps: int: Waypoints used by every gripper open/close interpolation.

lift_height: float: World-Z distance used to lift the object after the first grasp.

pre_grasp_distance: float: Distance from each grasp pose to its approach pose, in metres.

class embodichain.lab.sim.atomic_actions.HeldObjectGuardRequest[source]

Describe the next in-flight command boundary for held-object checks.

A request is correlated to one installed action-plan attempt and one next waypoint. The named segment lets an external verifier select phase-aware physical evidence without teaching the execution core skill-specific phases. deadline uses the observation timestamp domain.

Methods:

`__init__`(verification_id, skill_id, ...)
`snapshot`()	Return an independently owned guard request.

__init__(verification_id, skill_id, invocation_id, invocation_revision, invocation_index, attempt_generation, next_waypoint_index, segment_name, env_mask, allowed_held_object_relations, allowed_coordinated_held_object_relations, deadline)

snapshot()[source]

Return an independently owned guard request.

Return type:: HeldObjectGuardRequest
Returns:: Request with an independently owned environment mask.

class embodichain.lab.sim.atomic_actions.HeldObjectGuardResult[source]

Correlated in-flight held-object loss and recovery decision.

state_invalidation may only remove single-resource or coordinated held-object relations. It is applied to failure_mask before recovery planning, so a retry always observes reconciled symbolic state.

Methods:

__init__(verification_id, object_id, ...[, ...])

__init__(verification_id, object_id, attempt_generation, invocation_index, next_waypoint_index, failure_mask, state_invalidation, retry_mask, message='')

class embodichain.lab.sim.atomic_actions.HeldObjectPoseGoal[source]

Desired pose for the object held by this action’s control part.

Methods:

__init__(object_target_pose)

Attributes:

object_target_pose

Target object pose, shape (4, 4) or (num_envs, 4, 4).

__init__(object_target_pose)

object_target_pose: Tensor | SceneEntityPose: Target object pose, shape (4, 4) or (num_envs, 4, 4).

class embodichain.lab.sim.atomic_actions.HeldObjectState[source]

Observed or projected relation between an object and one manipulator.

Methods:

__init__(semantics, object_to_eef, grasp_xpos)

Attributes:

`env_mask`	Environments in which the relation is active.
`grasp_xpos`	End-effector grasp pose.
`object_to_eef`	Object-to-end-effector transform.
`semantics`	Semantics of the held object.

__init__(semantics, object_to_eef, grasp_xpos, env_mask=None)

env_mask: Tensor | None: Environments in which the relation is active.

grasp_xpos: Tensor: End-effector grasp pose.

object_to_eef: Tensor: Object-to-end-effector transform.

semantics: ObjectSemantics: Semantics of the held object.

class embodichain.lab.sim.atomic_actions.InFlightTrackingPolicy[source]

Feedback checks used while a command sequence is still in flight.

Methods:

__init__(metrics[, consecutive_violations, ...])

__init__(metrics, consecutive_violations=1, grace_period=0.0)

class embodichain.lab.sim.atomic_actions.InteractionPoints[source]

Batch of 3D interaction points on an object surface.

Methods:

`__init__`([object_label, custom_config, ...])
`get_approach_direction`(point_idx)	Get recommended approach direction for a given point.
`get_batch_size`()	Return the number of interaction points in this affordance.
`get_points_by_type`(point_type)	Get points by their interaction type.

Attributes:

`normals`	Optional surface normals at each interaction point with shape [B, 3].
`point_types`	Optional labels for each point's interaction type.
`points`	Batch of 3D interaction points with shape [B, 3].

__init__(object_label='', custom_config=<factory>, points=<factory>, normals=None, point_types=<factory>)

custom_config: dict[str, Any]: User-defined configuration payload.

get_approach_direction(point_idx)[source]

Get recommended approach direction for a given point.

Return type:: Tensor

get_batch_size()[source]

Return the number of interaction points in this affordance.

Return type:: int

get_points_by_type(point_type)[source]

Get points by their interaction type.

Return type:: Tensor | None

normals: Tensor | None = None: Optional surface normals at each interaction point with shape [B, 3].

point_types: list[str]: Optional labels for each point’s interaction type.

points: Tensor: Batch of 3D interaction points with shape [B, 3].

class embodichain.lab.sim.atomic_actions.JointPositionCommand[source]

A semantic command represented by one or batched joint positions.

positions has shape (control_dof,) or (num_envs, control_dof). A one-dimensional command is broadcast to the planning batch when resolved.

Methods:

`__init__`(positions)
`equivalent_to`(other)	Return whether `other` owns identical joint positions.
`resolve`(*, num_envs, control_dof, device[, ...])	Validate, move, and broadcast this command for a planning batch.
`snapshot`()	Return an independently owned command snapshot.

Attributes:

positions

Return an owned copy of the command payload.

__init__(positions)[source]

equivalent_to(other)[source]

Return whether other owns identical joint positions.

Return type:: bool

property positions: Tensor: Return an owned copy of the command payload.

resolve(*, num_envs, control_dof, device, dtype=None)[source]

Validate, move, and broadcast this command for a planning batch.

Parameters:

num_envs (int) – Number of selected environments.
control_dof (int) – Joint count of the resolved control part.
device (device | str) – Target planning device.
dtype (dtype | None) – Optional target dtype.

Return type:

Tensor

Returns:

Independently owned tensor with shape (num_envs, control_dof).

Raises:

ValueError – If the command shape does not match the control part or selected environment batch.

snapshot()[source]

Return an independently owned command snapshot.

Return type:: JointPositionCommand

class embodichain.lab.sim.atomic_actions.JointPositionGoal[source]

Explicit or named joint-space goal for a bound robot resource.

Methods:

__init__(target)

Attributes:

target

Joint qpos/waypoints or a named control-part profile command.

__init__(target)

target: Tensor | str: Joint qpos/waypoints or a named control-part profile command.

class embodichain.lab.sim.atomic_actions.JointPositionPayload[source]

Batched joint-position targets for the built-in robot transport.

Parameters:

positions (Tensor) – Joint positions with shape (batch_size, control_dof).
velocities (Tensor | None) – Optional joint velocities with the same shape and device.

Methods:

`__init__`(positions[, velocities])
`snapshot`()	Return an independently owned joint payload.

Attributes:

`batch_size`	Return the number of environment rows.
`device`	Return the tensor device.
`dof`	Return the number of controlled joints.
`transport_id`	Return the built-in joint-position transport identifier.

__init__(positions, velocities=None)

property batch_size: int: Return the number of environment rows.

property device: device: Return the tensor device.

property dof: int: Return the number of controlled joints.

snapshot()[source]

Return an independently owned joint payload.

Return type:: JointPositionPayload

property transport_id: str: Return the built-in joint-position transport identifier.

class embodichain.lab.sim.atomic_actions.JointPositionTarget[source]

Joint-position destination backed by one robot control part.

Methods:

__init__(control_part, joint_ids)

Attributes:

`address_fingerprint`	Return the destination plus the joints that must remain holdable.
`target_id`	Return the robot control-part destination.
`transport_id`	Return the built-in joint-position transport identifier.

__init__(control_part, joint_ids)

property address_fingerprint: Hashable: Return the destination plus the joints that must remain holdable.

property target_id: str: Return the robot control-part destination.

property transport_id: str: Return the built-in joint-position transport identifier.

class embodichain.lab.sim.atomic_actions.JointPositionTrackingEvaluator[source]

Evaluator for JointPositionTrackingMetric.

Classes:

metric_type

alias of JointPositionTrackingMetric

metric_type: alias of JointPositionTrackingMetric

class embodichain.lab.sim.atomic_actions.JointPositionTrackingMetric[source]

Maximum absolute joint-error tolerance.

Methods:

__init__([tolerance])

__init__(tolerance=0.05)

class embodichain.lab.sim.atomic_actions.JointPositionTrackingProjector[source]: Built-in projector for joint-position endpoint commands.

class embodichain.lab.sim.atomic_actions.JointPositionTrackingState[source]

Batched joint positions with shape (B, D).

Methods:

`__init__`(positions)
`snapshot`()	Return an independently owned state snapshot.

Attributes:

`batch_size`	Return the represented environment count.
`device`	Return the tensor device.

__init__(positions)

property batch_size: int: Return the represented environment count.

property device: device: Return the tensor device.

snapshot()[source]

Return an independently owned state snapshot.

Return type:: JointPositionTrackingState

class embodichain.lab.sim.atomic_actions.MonotonicExecutionClock[source]

Wall-clock implementation backed by time.

Methods:

`now`()	Return the current monotonic wall-clock time.
`sleep`(duration)	Sleep for a non-negative wall-clock duration.

now()[source]

Return the current monotonic wall-clock time.

Return type:: float
Returns:: Monotonic wall-clock timestamp in seconds.

sleep(duration)[source]

Sleep for a non-negative wall-clock duration.

Parameters:: duration (float) – Requested duration in seconds.
Return type:: None

class embodichain.lab.sim.atomic_actions.MotionPolicy[source]

Immutable motion-generation policy for one action invocation.

The policy is a runtime value object rather than application configuration. plan_opts is copied on construction so a caller-owned planner config cannot change an invocation after it has been created.

Methods:

`__init__`([strategy, sample_count, ...])
`to_motion_gen_options`(*, start_qpos, ...[, ...])	Translate this atomic policy into motion-generator options.

Attributes:

`dynamic_collision_mode`	How this invocation consumes live scene-snapshot collision entities.
`plan_opts`	Optional typed planner-specific options.
`sample_count`	Requested trajectory sample count when the backend does not preserve samples.
`strategy`	`motion_gen` or `ik_interp`.

__init__(strategy='ik_interp', sample_count=50, dynamic_collision_mode=DynamicCollisionMode.AUTO, plan_opts=None)

dynamic_collision_mode: DynamicCollisionMode: How this invocation consumes live scene-snapshot collision entities.

plan_opts: PlanOptions | None: Optional typed planner-specific options.

sample_count: int: Requested trajectory sample count when the backend does not preserve samples.

strategy: Literal['motion_gen', 'ik_interp']

motion_gen or ik_interp.

Type:: Motion strategy

to_motion_gen_options(*, start_qpos, control_part, sample_count=None, interpolation_dt=None, cartesian_linear=False)[source]

Translate this atomic policy into motion-generator options.

Parameters:

start_qpos (Tensor) – Observed controlled-joint start positions.
control_part (str) – Bound robot control-part name.
sample_count (int | None) – Optional segment-local sample-count override.
interpolation_dt (float | None) – Explicit waypoint interval used only by deterministic interpolation.
cartesian_linear (bool) – Whether every supplied Cartesian keyframe is a required linear-path sample rather than a sparse endpoint.

Return type:

MotionGenOptions

Returns:

Independently owned options for MotionGenerator.

class embodichain.lab.sim.atomic_actions.MoveEndEffector[source]

Plan a free-space move for a bound manipulator.

Classes:

`GoalType`	alias of `EndEffectorPoseGoal`
`OptionsType`	alias of `MoveEndEffectorOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of EndEffectorPoseGoal

OptionsType: alias of MoveEndEffectorOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='primary', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'motion.cartesian_pose'}), required_commands=mappingproxy({})),), constraints=()),), constraints=())

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

skill_id: ClassVar[str] = 'move_end_effector': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.MoveEndEffectorOptions[source]

Per-invocation behavior for MoveEndEffector.

Methods:

__init__()

__init__()

class embodichain.lab.sim.atomic_actions.MoveHeldObject[source]

Move the held object to the exact target object pose with a closed hand.

The requested object orientation is preserved exactly. Callers that need a transport orientation must encode it in HeldObjectPoseGoal; this action never substitutes an implicit end-effector orientation.

Classes:

`GoalType`	alias of `HeldObjectPoseGoal`
`OptionsType`	alias of `MoveHeldObjectOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of HeldObjectPoseGoal

OptionsType: alias of MoveHeldObjectOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='primary', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'kinematics.forward', 'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)),), constraints=())

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

skill_id: ClassVar[str] = 'move_held_object': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.MoveHeldObjectOptions[source]

Per-invocation held-object transport behavior.

Methods:

__init__()

__init__()

class embodichain.lab.sim.atomic_actions.MoveJoints[source]

Plan joint motion from the observed state to one or more waypoints.

Classes:

`GoalType`	alias of `JointPositionGoal`
`OptionsType`	alias of `MoveJointsOptions`

Attributes:

`agent_visible`	Whether an Action Agent should expose this skill by default.
`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of JointPositionGoal

OptionsType: alias of MoveJointsOptions

agent_visible: ClassVar[bool] = False: Whether an Action Agent should expose this skill by default.

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='primary', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'motion.joint_position'}), required_commands=mappingproxy({})),), constraints=()),), constraints=())

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

skill_id: ClassVar[str] = 'move_joints': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.MoveJointsOptions[source]

Per-invocation behavior for MoveJoints.

Methods:

__init__()

__init__()

class embodichain.lab.sim.atomic_actions.ObjectActionGoal[source]

Shared semantic-object goal contract for object-centric skills.

Methods:

__init__(semantics)

Attributes:

semantics

Semantic and geometric description of the object.

__init__(semantics)

semantics: ObjectSemantics: Semantic and geometric description of the object.

class embodichain.lab.sim.atomic_actions.ObjectSemantics[source]

Shallow-frozen semantic information about an interaction object.

Attention

Top-level fields cannot be rebound after construction. Nested affordance and metadata objects may remain mutable but never establish object identity.

Methods:

__init__(affordance, geometry, entity_id[, ...])

Attributes:

`affordance`	Affordance data describing supported interactions.
`entity_id`	Stable scene identifier used by snapshot grounding and object identity.
`geometry`	Non-affordance metadata used to resolve geometry-derived affordance data.
`label`	Semantic object category.
`properties`	Physical properties such as mass and friction.

__init__(affordance, geometry, entity_id, properties=<factory>, label='none')

affordance: Affordance: Affordance data describing supported interactions.

entity_id: str: Stable scene identifier used by snapshot grounding and object identity.

geometry: dict[str, Any]: Non-affordance metadata used to resolve geometry-derived affordance data.

label: str: Semantic object category.

properties: dict[str, Any]: Physical properties such as mass and friction.

class embodichain.lab.sim.atomic_actions.ObservationProvider[source]

Source of fresh planning contexts for feedback-driven execution.

Methods:

`__init__`(args, *kwargs)
`observe`(task_state)	Capture the latest robot and scene state.

__init__(*args, **kwargs)

observe(task_state)[source]

Capture the latest robot and scene state.

Parameters:: task_state (TaskState) – Runner-owned, externally verified symbolic task state.
Return type:: PlanningContext
Returns:: Fresh context with stable, ordered environment IDs.

class embodichain.lab.sim.atomic_actions.ObservedArticulationJointState[source]

Live measured state for one scene articulation joint.

Methods:

`__init__`(position[, valid_mask])
`snapshot`()	Return an independently owned observation value.

Attributes:

`position`	Measured joint position with shape `(J,)` or `(B, J)`.
`valid_mask`	Optional row-validity mask for a batched observation.

__init__(position, valid_mask=None)

position: Tensor: Measured joint position with shape (J,) or (B, J).

snapshot()[source]

Return an independently owned observation value.

Return type:: ObservedArticulationJointState

valid_mask: Tensor | None: Optional row-validity mask for a batched observation.

class embodichain.lab.sim.atomic_actions.OpenDoor[source]

Approach, grasp, rotate a door about its hinge, release, and retract.

Classes:

`GoalType`	alias of `OpenDoorGoal`
`OptionsType`	alias of `OpenDoorOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`open_loop`	Whether the skill intentionally declares no verified physical effect.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of OpenDoorGoal

OptionsType: alias of OpenDoorOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='primary', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'open': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>, 'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)),), constraints=())

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

open_loop: ClassVar[bool] = True: Whether the skill intentionally declares no verified physical effect.

skill_id: ClassVar[str] = 'open_door': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.OpenDoorAffordance[source]

Target-local handle geometry and a resolved hinge axis.

Use from_articulation() to start from a graspable handle link. The factory consumes the articulation’s public parent-joint chain, skips only fixed joints, and automatically accepts one unambiguous active revolute ancestor. Ambiguous mechanisms require an explicit hinge joint name. The resulting affordance owns no simulator entity or live pose.

Methods:

`__init__`([object_label, custom_config, ...])
`from_articulation`(articulation, link_name, *)	Build handle semantics from a parent revolute joint.

Attributes:

`axis_origin`	Resolved point on the hinge axis in the handle-link frame.
`joint_limits`	Optional lower and upper hinge limits in radians.
`joint_name`	Stable name of the resolved parent revolute joint.
`opening_direction`	Joint-coordinate direction from the closed limit toward the open limit.
`rotation_axis`	Resolved hinge axis expressed in the handle-link frame.

__init__(object_label='', custom_config=<factory>, joint_limits=None, opening_direction=1, *, mesh_vertices, mesh_triangles, rotation_axis, axis_origin, joint_name)

axis_origin: tuple[float, float, float]: Resolved point on the hinge axis in the handle-link frame.

custom_config: dict[str, Any]: User-defined configuration payload.

classmethod from_articulation(articulation, link_name, *, hinge_joint_name=None, opening_direction=1)[source]

Build handle semantics from a parent revolute joint.

Automatic resolution skips only fixed joints. It succeeds when the handle chain contains exactly one active ancestor and that joint is revolute. A chain with multiple active ancestors can represent a latch, handle joint, or another mechanism and therefore requires an explicit hinge_joint_name. Hinge geometry is converted from the joint frame into the requested handle-link frame using current public link poses.

Parameters:

articulation (Articulation) – Articulation containing the graspable handle link.
link_name (str) – Graspable handle link from which to start the traversal.
hinge_joint_name (str | None) – Optional explicit revolute ancestor. Required when more than one active ancestor makes automatic resolution ambiguous.
opening_direction (int) – Joint-coordinate direction from the closed legal endpoint toward the open endpoint. Defaults to increasing qpos.

Return type:

OpenDoorAffordance

Returns:

Pure target-local handle and hinge semantics.

Raises:

TypeError – If a supplied name is not a string.
ValueError – If the link or explicit joint is unknown, automatic resolution is ambiguous, the selected joint is not revolute, or the resolved joint geometry is invalid.

joint_limits: tuple[float, float] | None = None: Optional lower and upper hinge limits in radians.

joint_name: str: Stable name of the resolved parent revolute joint.

opening_direction: int = 1: Joint-coordinate direction from the closed limit toward the open limit.

rotation_axis: Tensor: Resolved hinge axis expressed in the handle-link frame.

class embodichain.lab.sim.atomic_actions.OpenDoorGoal[source]

Door handle and desired absolute opening state.

Methods:

__init__(semantics, target_pose, open_fraction)

Attributes:

`open_fraction`	Desired hinge position normalized from its closed to open legal endpoint.
`target_pose`	Handle-link pose snapshot or late-bound scene-entity reference.

__init__(semantics, target_pose, open_fraction)

open_fraction: float | Tensor: Desired hinge position normalized from its closed to open legal endpoint.

target_pose: Tensor | SceneEntityPose: Handle-link pose snapshot or late-bound scene-entity reference.

class embodichain.lab.sim.atomic_actions.OpenDoorOptions[source]

Per-invocation approach, interpolation, release, and retract behavior.

Methods:

__init__([hand_interp_steps, ...])

Attributes:

`approach_distance`	Pre-grasp distance opposite the automatically inferred approach axis.
`door_waypoint_count`	Number of Cartesian keyframes along the handle's circular arc.
`hand_interp_steps`	Number of waypoints used for each close/open hand segment.
`joint_position_tolerance`	Tolerance for legal-limit and already-open comparisons in radians.
`retract_distance`	Post-release retreat distance opposite the rotated approach axis.

__init__(hand_interp_steps=5, door_waypoint_count=20, approach_distance=0.1, retract_distance=0.1, joint_position_tolerance=0.0001)

approach_distance: float: Pre-grasp distance opposite the automatically inferred approach axis.

door_waypoint_count: int: Number of Cartesian keyframes along the handle’s circular arc.

hand_interp_steps: int: Number of waypoints used for each close/open hand segment.

joint_position_tolerance: float: Tolerance for legal-limit and already-open comparisons in radians.

retract_distance: float: Post-release retreat distance opposite the rotated approach axis.

class embodichain.lab.sim.atomic_actions.PhaseEffectGateRequest[source]

Correlate a blocking physical-effect check with a segment entry.

The action’s preceding command remains active while the gate is unresolved. A gate is scoped to the enclosing action attempt and does not create a separate planning, recovery, or timeout budget.

Parameters:

verification_id (int) – Session-local single-use request identity.
gate_id (str) – Invocation-local stable gate identity.
skill_id (str) – Registered action skill identity.
invocation_id (str | None) – Optional logical invocation correlation identity.
invocation_revision (int) – Active invocation revision.
invocation_index (int) – Active invocation position in the session.
attempt_generation (int) – Installed action-plan attempt generation.
next_waypoint_index (int) – First command frame blocked by the gate.
segment_name (str) – Named trajectory segment blocked by the gate.
requested_at (float) – Request creation time in the observation timestamp domain.
deadline (float) – Enclosing action deadline in that same timestamp domain.
env_mask (Tensor) – Active rows that must satisfy the gate together.

Methods:

`__init__`(verification_id, gate_id, skill_id, ...)
`snapshot`()	Return an independently owned gate request.

__init__(verification_id, gate_id, skill_id, invocation_id, invocation_revision, invocation_index, attempt_generation, next_waypoint_index, segment_name, requested_at, deadline, env_mask)

snapshot()[source]

Return an independently owned gate request.

Return type:: PhaseEffectGateRequest

class embodichain.lab.sim.atomic_actions.PhaseEffectGateRequirement[source]

Require physical-effect evidence before one trajectory segment starts.

The requirement carries only stable core correlation data. Semantic integrations own the corresponding observation specification and monitor; the execution session owns blocking, timeout, and action-retry behavior.

Parameters:

gate_id (str) – Invocation-local stable gate identifier.
segment_name (str) – Exact named trajectory segment blocked by this gate.

Methods:

`__init__`(gate_id, segment_name)
`snapshot`()	Return an independently constructed immutable requirement.

__init__(gate_id, segment_name)

snapshot()[source]

Return an independently constructed immutable requirement.

Return type:: PhaseEffectGateRequirement

class embodichain.lab.sim.atomic_actions.PhaseEffectGateResult[source]

Current-observation decision for one blocking segment-entry gate.

Rows absent from both decision masks remain unresolved. retry_mask is a subset of failed rows for which replaying the enclosing action remains valid; no gate outcome mutates verified task state.

Parameters:

verification_id (int) – Identity copied from the consumed gate request.
gate_id (str) – Stable gate identity copied from the request.
attempt_generation (int) – Action attempt copied from the request.
invocation_index (int) – Session invocation index copied from the request.
next_waypoint_index (int) – Blocked waypoint copied from the request.
success_mask (Tensor) – Rows whose current evidence satisfies the gate.
failure_mask (Tensor) – Rows whose current evidence contradicts the gate.
retry_mask (Tensor) – Failed rows allowed to retry the enclosing action.
message (str) – Optional physical-failure diagnostic.

Methods:

__init__(verification_id, gate_id, ...[, ...])

__init__(verification_id, gate_id, attempt_generation, invocation_index, next_waypoint_index, success_mask, failure_mask, retry_mask, message='')

class embodichain.lab.sim.atomic_actions.PickUp[source]

Approach a grasp pose, close the gripper, lift.

Classes:

`GoalType`	alias of `GraspGoal`
`OptionsType`	alias of `PickUpOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of GraspGoal

OptionsType: alias of PickUpOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='primary', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'kinematics.forward', 'kinematics.batch_inverse', 'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'open': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>, 'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)),), constraints=())

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

skill_id: ClassVar[str] = 'pick_up': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.PickUpOptions[source]

Per-invocation pickup behavior.

Methods:

__init__([hand_interp_steps, ...])

Attributes:

`approach_alignment_max_angle`	Optional maximum TCP z-axis deviation from the approach direction.
`approach_direction`	World-frame direction from the pre-grasp pose to the grasp pose.
`downstream_object_target_poses`	Future object poses that must be reachable with the selected grasp.
`fixed_object_to_eef`	Optional object-frame to end-effector SE(3) grasp calibration.
`grasp_frame_to_eef`	Canonical grasp-frame to robot end-effector SE(3) calibration.
`grasp_settle_steps`	Fully closed hold frames before lifting the end-effector.
`hand_interp_steps`	Number of waypoints for the gripper-close interpolation segment.
`lift_height`	Height (m) to lift the end-effector after closing the gripper.
`obj_upright_direction`	Optional object local direction used to choose the upright grasp rotation.
`pick_object_part`	Name of the object part to pick up (used for grasp pose generation).
`pre_grasp_distance`	Distance to offset back from the grasp pose along the approach direction.
`rotate_upright`	Optional rotation (radians) about the grasp x-axis to apply after grasp selection.

__init__(hand_interp_steps=5, grasp_settle_steps=0, pick_object_part='center', lift_height=0.1, pre_grasp_distance=0.15, approach_direction=tensor([0., 0., -1.]), approach_alignment_max_angle=None, downstream_object_target_poses=(), obj_upright_direction=None, rotate_upright=None, grasp_frame_to_eef=tensor([[1., 0., 0., 0.], [0., 1., 0., 0.], [0., 0., 1., 0.], [0., 0., 0., 1.]]), fixed_object_to_eef=None)

approach_alignment_max_angle: float | None: Optional maximum TCP z-axis deviation from the approach direction.

approach_direction: Tensor: World-frame direction from the pre-grasp pose to the grasp pose.

downstream_object_target_poses: tuple[Tensor | SceneEntityPose, ...]: Future object poses that must be reachable with the selected grasp.

fixed_object_to_eef: Tensor | None

Optional object-frame to end-effector SE(3) grasp calibration.

When no explicit goal grasp is supplied, this transform bypasses affordance sampling and the sampled-grasp orientation/calibration adjustments.

grasp_frame_to_eef: Tensor: Canonical grasp-frame to robot end-effector SE(3) calibration.

grasp_settle_steps: int: Fully closed hold frames before lifting the end-effector.

hand_interp_steps: int: Number of waypoints for the gripper-close interpolation segment.

lift_height: float: Height (m) to lift the end-effector after closing the gripper.

obj_upright_direction: Tensor | None: Optional object local direction used to choose the upright grasp rotation.

pick_object_part: str: Name of the object part to pick up (used for grasp pose generation). Currently support [center | top | bottom].

pre_grasp_distance: float: Distance to offset back from the grasp pose along the approach direction.

rotate_upright: float | None: Optional rotation (radians) about the grasp x-axis to apply after grasp selection.

class embodichain.lab.sim.atomic_actions.Place[source]

Lower the held object to a place pose, open the gripper, retract.

The PlaceGoal may carry either a single waypoint (num_envs, 4, 4) (or a broadcastable (4, 4)) or a multi-waypoint trajectory (num_envs, n_waypoint, 4, 4). In the multi-waypoint case the approach segment visits every waypoint in order; approaching from above the first waypoint, descending through each waypoint, then opening the gripper at the final waypoint and retracting to above the last waypoint. Starting joint positions are inherited from PlanningContext.

An AssembleGoal replaces the explicit EEF pose with an assembly affordance: the place pose is derived from the base object’s snapshot pose and assemble_to_base_pose, converted to an EEF pose through the held object’s object_to_eef (read from PlanningContext).

Attributes:

`GoalType`	Concrete goal dataclass or dataclasses accepted by this skill.
`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`skill_id`	Stable registry identifier for this skill.

Classes:

OptionsType

alias of PlaceOptions

GoalType: ClassVar[type | tuple[type, ...]] = (<class 'embodichain.lab.sim.atomic_actions.primitives.place.PlaceGoal'>, <class 'embodichain.lab.sim.atomic_actions.primitives.place.AssembleGoal'>): Concrete goal dataclass or dataclasses accepted by this skill.

OptionsType: alias of PlaceOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='primary', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'kinematics.forward', 'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'open': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>, 'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)),), constraints=())

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

skill_id: ClassVar[str] = 'place': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.PlaceGoal[source]

End-effector release-pose target used by Place.

Methods:

__init__(xpos[, tcp_symmetry])

Attributes:

`tcp_symmetry`	Optional TCP-frame symmetry allowed by the placement semantics.
`xpos`	Target end-effector release pose.

__init__(xpos, tcp_symmetry='none')

tcp_symmetry: Literal['none', 'z_roll_180']

Optional TCP-frame symmetry allowed by the placement semantics.

"none" preserves the pose exactly. "z_roll_180" lets placement choose between the pose and its TCP z-roll 180 equivalent, which flips TCP x/y while preserving TCP z and translation.

xpos: Tensor | SceneEntityPose

Target end-effector release pose.

Accepts (4, 4), (num_envs, 4, 4), or (num_envs, n_waypoint, 4, 4).

class embodichain.lab.sim.atomic_actions.PlaceOptions[source]

Per-invocation placement behavior.

Methods:

__init__([hand_interp_steps, ...])

Attributes:

`cartesian_waypoint_count`	Number of fixed-orientation Cartesian keyframes per translation segment.
`hand_interp_steps`	Number of waypoints for the gripper-open interpolation segment.
`lift_height`	Height (m) to retract the end-effector after opening the gripper.
`max_approach_retract_z`	Optional maximum world-frame TCP z for approach and retract poses (m).
`preserve_current_object_orientation`	Keep the held object's observed world orientation at the place target.
`release_settle_steps`	Fully open hold frames before retracting the end-effector.

__init__(hand_interp_steps=5, release_settle_steps=0, lift_height=0.1, max_approach_retract_z=None, cartesian_waypoint_count=1, preserve_current_object_orientation=False)

cartesian_waypoint_count: int: Number of fixed-orientation Cartesian keyframes per translation segment.

hand_interp_steps: int: Number of waypoints for the gripper-open interpolation segment.

lift_height: float: Height (m) to retract the end-effector after opening the gripper.

max_approach_retract_z: float | None: Optional maximum world-frame TCP z for approach and retract poses (m).

preserve_current_object_orientation: bool: Keep the held object’s observed world orientation at the place target.

release_settle_steps: int: Fully open hold frames before retracting the end-effector.

class embodichain.lab.sim.atomic_actions.PlannerDiagnostics[source]

Planner metadata retained for debugging and recovery decisions.

Methods:

__init__(backend[, messages, metadata, failure])

__init__(backend, messages=(), metadata=<factory>, failure=None)

class embodichain.lab.sim.atomic_actions.PlanningContext[source]

Complete side-effect-free input to AtomicAction.plan().

Methods:

`__init__`(robot, task, scene, env_ids[, ...])
`get_articulation_joint_state`(...)	Return verified state for one canonical articulation joint.
`get_coordinated_held_object`(first_resource, ...)	Return a coordinated held-object relation, if any.
`get_held_object`(resource)	Return the object held by `resource`, if any.
`project`(*, qpos, task)	Create the hypothetical context used to compile a following action.
`require_control_dt`()	Return the explicit command period required for interpolation.

Attributes:

`articulation_joints`	Verified articulation-joint states.
`batch_size`	Number of environments in this planning request.
`control_dt`	Explicit command period used by action-owned interpolation.
`coordinated_held_objects`	Coordinated held-object relations.
`held_objects`	Single-resource held-object relations.
`last_qpos`	Measured joint positions used as the planning start state.

__init__(robot, task, scene, env_ids, control_dt=None)

property articulation_joints: Mapping[tuple[str, str], ArticulationJointState]: Verified articulation-joint states.

property batch_size: int: Number of environments in this planning request.

control_dt: float | None: Explicit command period used by action-owned interpolation.

property coordinated_held_objects: Mapping[tuple[str, str], CoordinatedHeldObjectState]: Coordinated held-object relations.

get_articulation_joint_state(articulation_id, joint_id)[source]

Return verified state for one canonical articulation joint.

Return type:: ArticulationJointState | None

get_coordinated_held_object(first_resource, second_resource)[source]

Return a coordinated held-object relation, if any.

Return type:: CoordinatedHeldObjectState | None

get_held_object(resource)[source]

Return the object held by resource, if any.

Return type:: HeldObjectState | None

property held_objects: Mapping[str, HeldObjectState]: Single-resource held-object relations.

property last_qpos: Tensor: Measured joint positions used as the planning start state.

project(*, qpos, task)[source]

Create the hypothetical context used to compile a following action.

Parameters:

qpos (Tensor) – Projected terminal joint positions.
task (TaskState) – Task state after applying expected effects.

Return type:

Returns:

New context. No measured state or simulator state is mutated.

require_control_dt()[source]

Return the explicit command period required for interpolation.

Raises:: ValueError – If the caller did not provide control_dt.
Return type:: float

class embodichain.lab.sim.atomic_actions.PlanningContextTrackingFeedbackProvider[source]: Built-in provider backed by PlanningContext.robot.

class embodichain.lab.sim.atomic_actions.PlanningFailure[source]

Stable planning-failure classification used by recovery policy.

Parameters:

code (str) – Exact machine-readable failure code.
retryable (bool) – Whether action-level recovery may replan failed rows.

Methods:

__init__(code[, retryable])

__init__(code, retryable=True)

class embodichain.lab.sim.atomic_actions.PoseTrackingEvaluator[source]

Evaluator for PoseTrackingMetric.

Classes:

metric_type

alias of PoseTrackingMetric

metric_type: alias of PoseTrackingMetric

class embodichain.lab.sim.atomic_actions.PoseTrackingMetric[source]

Independent translation and rotation tolerances for base pose.

Methods:

__init__([translation_tolerance, ...])

__init__(translation_tolerance=0.02, rotation_tolerance=0.05)

class embodichain.lab.sim.atomic_actions.PoseTrackingState[source]

Batched homogeneous poses with shape (B, 4, 4).

Methods:

`__init__`(poses)
`snapshot`()	Return an independently owned state snapshot.

Attributes:

`batch_size`	Return the represented environment count.
`device`	Return the tensor device.

__init__(poses)

property batch_size: int: Return the represented environment count.

property device: device: Return the tensor device.

snapshot()[source]

Return an independently owned state snapshot.

Return type:: PoseTrackingState

class embodichain.lab.sim.atomic_actions.Pour[source]

Rotate and return an exclusively held object about its internal axis.

Classes:

`GoalType`	alias of `PourGoal`
`OptionsType`	alias of `PourOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`open_loop`	Whether the skill intentionally declares no verified physical effect.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of PourGoal

OptionsType: alias of PourOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='primary', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'kinematics.forward', 'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)),), constraints=())

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

open_loop: ClassVar[bool] = True: Whether the skill intentionally declares no verified physical effect.

skill_id: ClassVar[str] = 'pour': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.PourGoal[source]

Rotate the object currently held by the bound manipulator.

Methods:

__init__()

__init__()

class embodichain.lab.sim.atomic_actions.PourOptions[source]

Per-invocation pouring behavior.

Methods:

__init__([rotate_angle])

Attributes:

rotate_angle

Signed rotation about the held object's local internal axis, in radians.

__init__(rotate_angle=0.7853981633974483)

rotate_angle: float: Signed rotation about the held object’s local internal axis, in radians.

class embodichain.lab.sim.atomic_actions.Press[source]

Open-loop motion primitive that approaches, presses, and retracts.

Classes:

`GoalType`	alias of `PressGoal`
`OptionsType`	alias of `PressOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`open_loop`	Whether the skill intentionally declares no verified physical effect.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of PressGoal

OptionsType: alias of PressOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='primary', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'kinematics.forward', 'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)),), constraints=())

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

open_loop: ClassVar[bool] = True: Whether the skill intentionally declares no verified physical effect.

skill_id: ClassVar[str] = 'press': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.PressAffordance[source]

Target-local contact point and parent-joint pressing geometry.

Methods:

`__init__`([object_label, custom_config, ...])
`get_press_pose`(target_pose[, press_position])	Construct a press pose at the configured surface point.
`resolve_from_object_geometry`(geometry)	Resolve the target-local prismatic axis, sign, and contact point.

Attributes:

`press_axis`	Parent prismatic-joint axis, signed toward articulation geometry.
`press_position`	Local contact point; inferred from articulation geometry when omitted.

__init__(object_label='', custom_config=<factory>, press_axis=<factory>, *, press_position=None)

custom_config: dict[str, Any]: User-defined configuration payload.

get_press_pose(target_pose, press_position=None)[source]

Construct a press pose at the configured surface point.

The end-effector z-axis follows press_axis in world space. An adaptive reference produces an orthonormal, right-handed frame.

Parameters:

target_pose (Tensor) – Current target world pose with shape (B, 4, 4).
press_position (tuple[float, float, float] | None) – Optional per-call exact local-frame press position. It overrides press_position.

Return type:

Tensor

Returns:

Batched world-frame press poses with shape (B, 4, 4).

Raises:

ValueError – If an input has an invalid shape or value.

press_axis: Tensor: Parent prismatic-joint axis, signed toward articulation geometry.

press_position: tuple[float, float, float] | None = None: Local contact point; inferred from articulation geometry when omitted.

resolve_from_object_geometry(geometry)[source]

Resolve the target-local prismatic axis, sign, and contact point.

Return type:: None

class embodichain.lab.sim.atomic_actions.PressGoal[source]

Target object described by a press affordance.

Methods:

__init__(semantics, target_pose)

Attributes:

target_pose

Target pose snapshot or late-bound stable scene-entity reference.

__init__(semantics, target_pose)

target_pose: Tensor | SceneEntityPose: Target pose snapshot or late-bound stable scene-entity reference.

class embodichain.lab.sim.atomic_actions.PressOptions[source]

Per-invocation pressing behavior.

Methods:

__init__([hand_interp_steps, ...])

Attributes:

`approach_distance`	Distance from the press position opposite the press direction.
`hand_interp_steps`	Number of waypoints used to close the hand.
`press_distance`	Distance traveled into the target along its press axis.
`press_position`	Optional local-frame position overriding the affordance press position.

__init__(hand_interp_steps=5, approach_distance=0.1, press_distance=0.05, press_position=None)

approach_distance: float: Distance from the press position opposite the press direction.

hand_interp_steps: int: Number of waypoints used to close the hand.

press_distance: float: Distance traveled into the target along its press axis.

press_position: tuple[float, float, float] | None: Optional local-frame position overriding the affordance press position.

class embodichain.lab.sim.atomic_actions.PushObject[source]

Close the end effector, contact a rigid object, and push it in-plane.

Classes:

`GoalType`	alias of `PushObjectGoal`
`OptionsType`	alias of `PushObjectOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`open_loop`	Whether the skill intentionally declares no verified physical effect.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of PushObjectGoal

OptionsType: alias of PushObjectOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='primary', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)),), constraints=())

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

open_loop: ClassVar[bool] = True: Whether the skill intentionally declares no verified physical effect.

skill_id: ClassVar[str] = 'push_object': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.PushObjectGoal[source]

Push one rigid object toward a target pose on the target support plane.

Methods:

__init__(semantics, target_pose)

Attributes:

target_pose

Desired object pose or a late-bound scene-entity target reference.

__init__(semantics, target_pose)

target_pose: PoseGoalValue: Desired object pose or a late-bound scene-entity target reference.

class embodichain.lab.sim.atomic_actions.PushObjectOptions[source]

Per-invocation planar pushing behavior.

Methods:

__init__([hand_interp_steps, ...])

Attributes:

`approach_height`	Distance above the contact pose used for the free-space approach.
`completion_tolerance`	Planar target distance at which the action succeeds without moving.
`contact_distance`	Initial planar clearance behind the object along the push direction.
`contact_frame_to_eef`	Contact-frame to robot end-effector SE(3) calibration.
`hand_interp_steps`	Number of waypoints used to close the end effector before approach.
`object_contact_offset`	Object-local point used as the center of the planar contact frame.
`push_overshoot`	Additional end-effector travel beyond the object's target displacement.
`retract_height`	Distance above the pushed pose used for the final retraction.
`support_frame_planar_contact_offset`	Optional target-support-frame override for the contact's planar offset.
`tool_calibrations`	Per-control-part tool-frame overrides for asymmetric robot arms.

__init__(hand_interp_steps=5, approach_height=0.1, retract_height=0.1, contact_distance=0.03, push_overshoot=0.0, completion_tolerance=0.0, object_contact_offset=tensor([0., 0., 0.]), support_frame_planar_contact_offset=None, contact_frame_to_eef=tensor([[1., 0., 0., 0.], [0., 1., 0., 0.], [0., 0., 1., 0.], [0., 0., 0., 1.]]), tool_calibrations=())

approach_height: float: Distance above the contact pose used for the free-space approach.

completion_tolerance: float: Planar target distance at which the action succeeds without moving.

contact_distance: float: Initial planar clearance behind the object along the push direction.

contact_frame_to_eef: Tensor: Contact-frame to robot end-effector SE(3) calibration.

hand_interp_steps: int: Number of waypoints used to close the end effector before approach.

object_contact_offset: Tensor: Object-local point used as the center of the planar contact frame.

push_overshoot: float: Additional end-effector travel beyond the object’s target displacement.

retract_height: float: Distance above the pushed pose used for the final retraction.

support_frame_planar_contact_offset: Tensor | None: Optional target-support-frame override for the contact’s planar offset.

tool_calibrations: tuple[PushObjectToolCalibration, ...]: Per-control-part tool-frame overrides for asymmetric robot arms.

class embodichain.lab.sim.atomic_actions.PushObjectToolCalibration[source]

End-effector calibration selected by a bound motion control part.

Parameters:

control_part (str) – Exact motion control-part identifier that selects this calibration.
contact_frame_to_eef (Tensor) – Contact-frame to robot end-effector SE(3) calibration.
contact_distance (float | None) – Optional tool-specific planar clearance behind the object. None uses PushObjectOptions’ default.

Methods:

__init__(control_part, contact_frame_to_eef)

Attributes:

`contact_distance`	Optional tool-specific planar clearance behind the object.
`contact_frame_to_eef`	Contact-frame to robot end-effector SE(3) calibration.
`control_part`	Exact motion control-part identifier that selects this calibration.

__init__(control_part, contact_frame_to_eef, contact_distance=None)

contact_distance: float | None: Optional tool-specific planar clearance behind the object.

contact_frame_to_eef: Tensor: Contact-frame to robot end-effector SE(3) calibration.

control_part: str: Exact motion control-part identifier that selects this calibration.

class embodichain.lab.sim.atomic_actions.RecoveryPolicy[source]

Bounded local recovery policy used by the execution runtime.

Methods:

__init__([max_replans, max_action_retries, ...])

Attributes:

`action_timeout`	Maximum time for one action attempt, including terminal effect verification.
`goal_rotation_threshold`	Dynamic-goal rotation threshold in radians (five degrees by default).
`goal_translation_threshold`	Dynamic-goal translation threshold in metres.
`max_action_retries`	Maximum whole-action retries after planning, execution, or effect failure.
`max_replans`	Maximum replans within one action attempt.

__init__(max_replans=3, max_action_retries=2, goal_translation_threshold=0.02, goal_rotation_threshold=0.0872664626, action_timeout=30.0)

action_timeout: float: Maximum time for one action attempt, including terminal effect verification.

goal_rotation_threshold: float: Dynamic-goal rotation threshold in radians (five degrees by default).

goal_translation_threshold: float: Dynamic-goal translation threshold in metres.

max_action_retries: int: Maximum whole-action retries after planning, execution, or effect failure.

max_replans: int: Maximum replans within one action attempt.

class embodichain.lab.sim.atomic_actions.ResolvedActionRequest[source]

Engine-owned immutable planning snapshot for one invocation revision.

Recovery replans reuse this object verbatim and vary only the PlanningContext. Deep-copying goal value payloads, policies, and skill options severs caller-owned mutable data before planning starts while retaining simulator-backed entity handles and private runtime caches.

Methods:

`__init__`(skill_id, goal, binding, ...[, ...])
`snapshot`()	Return an independently owned resolved-request snapshot.

__init__(skill_id, goal, binding, motion_policy, tracking_policy, recovery_policy, skill_options, phase_effect_gates=(), invocation_id=None, revision=0)

snapshot()[source]

Return an independently owned resolved-request snapshot.

Return type:: ResolvedActionRequest[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]

class embodichain.lab.sim.atomic_actions.RigidObjectSceneProvider[source]

Observe simulation rigid objects and maintain scene revisions.

The provider increments the general scene version when any tracked entity moves materially. For IDs declared as collision entities it additionally increments a per-environment collision-world revision, allowing one batch row to invalidate its trajectory without failing unrelated rows.

Parameters:

entities (Mapping[str, RigidObject]) – Stable entity IDs mapped to live simulation rigid objects.
collision_entity_ids (Sequence[str]) – Tracked IDs consumed as dynamic planner obstacles.
cfg (RigidObjectSceneProviderCfg | None) – Optional material-change thresholds.

Methods:

`__init__`(entities, *[, ...])
`snapshot`(*, timestamp, env_ids)	Capture object poses and advance material-change revisions.

__init__(entities, *, collision_entity_ids=(), cfg=None)[source]

snapshot(*, timestamp, env_ids)[source]

Capture object poses and advance material-change revisions.

Parameters:

timestamp (float) – Current simulation observation time.
env_ids (Tensor) – Stable correlation IDs whose order matches object rows.

Return type:

SceneSnapshot

Returns:

Versioned scene snapshot with per-environment collision revisions.

class embodichain.lab.sim.atomic_actions.RigidObjectSceneProviderCfg[source]

Material-pose thresholds used to advance scene revisions.

Methods:

`__init__`([translation_threshold, ...])
`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:

`rotation_threshold`	Minimum rotation in radians considered a scene change.
`translation_threshold`	Minimum translation in metres considered a scene change.

__init__(translation_threshold=<factory>, rotation_threshold=<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.

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.

rotation_threshold: float: Minimum rotation in radians considered a scene change.

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_threshold: float: Minimum translation in metres considered a scene change.

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.lab.sim.atomic_actions.RobotObservation[source]

Measured robot state used as the start of planning or replanning.

Methods:

`__init__`(timestamp, qpos, qvel[, qeffort, ...])
`with_qpos`(qpos)	Create a projected observation with a new position and zero velocity.

Attributes:

`batch_size`	Number of represented vectorized environments.
`robot_dof`	Number of robot joint-position columns.

__init__(timestamp, qpos, qvel, qeffort=None, root_pose=None, root_twist=None)

property batch_size: int: Number of represented vectorized environments.

property robot_dof: int: Number of robot joint-position columns.

with_qpos(qpos)[source]

Create a projected observation with a new position and zero velocity.

Parameters:: qpos (Tensor) – Projected joint positions with the same shape as this observation.
Return type:: RobotObservation
Returns:: New observation suitable for compiling the next action.

class embodichain.lab.sim.atomic_actions.RunnerStatus[source]

Lifecycle status owned by an ExecutionRunner.

Methods:

__new__(value)

__new__(value)

class embodichain.lab.sim.atomic_actions.RunnerStep[source]

Result of one non-blocking execution-runner update.

Methods:

__init__(status, timestamp, wait_duration, ...)

Attributes:

`is_waiting`	Whether no session tick was due during this update.
`message`	Terminal or failure diagnostic, when available.

__init__(status, timestamp, wait_duration, context, tick, dispatches, command_count, message=None)

property is_waiting: bool: Whether no session tick was due during this update.

message: str | None: Terminal or failure diagnostic, when available.

class embodichain.lab.sim.atomic_actions.RuntimeCommandFrame[source]

Synchronized endpoint commands for one batched runtime instant.

Parameters:

commands (tuple[EndpointCommand, ...]) – Commands dispatched together for this frame.
active_mask (Tensor) – Boolean environment rows allowed to execute commands. Transports must actively neutralize addressed targets for false rows rather than leaving a previously persistent command running.
env_ids (Tensor) – Stable environment identifiers for the batch rows.
hold_duration (Tensor) – Per-row delay before advancing to the next frame.

Methods:

`__init__`(commands, active_mask, env_ids, ...)
`snapshot`()	Return an independently owned command frame.
`with_active_mask`(active_mask)	Return a frame snapshot with a replacement active-row mask.

Attributes:

`batch_size`	Return the number of environment rows.
`device`	Return the shared frame device.
`targets`	Return owned targets in command order.

__init__(commands, active_mask, env_ids, hold_duration)

property batch_size: int: Return the number of environment rows.

property device: device: Return the shared frame device.

snapshot()[source]

Return an independently owned command frame.

Return type:: RuntimeCommandFrame

property targets: tuple[RuntimeEndpointTarget, ...]: Return owned targets in command order.

with_active_mask(active_mask)[source]

Return a frame snapshot with a replacement active-row mask.

Parameters:: active_mask (Tensor) – Boolean mask with one value per environment row.
Return type:: RuntimeCommandFrame
Returns:: Independently owned frame with unchanged commands and timing.

class embodichain.lab.sim.atomic_actions.RuntimeCommandPayload[source]

Immutable-by-ownership payload submitted to one runtime transport.

Attributes:

`batch_size`	Return the number of environment rows in this payload.
`device`	Return the device shared by this payload's batched values.
`transport_id`	Return the transport kind that accepts this payload.

Methods:

Return an independently owned payload snapshot.

abstract property batch_size: int: Return the number of environment rows in this payload.

abstract property device: device: Return the device shared by this payload’s batched values.

abstract snapshot()[source]

Return an independently owned payload snapshot.

Return type:: RuntimeCommandPayload

abstract property transport_id: str: Return the transport kind that accepts this payload.

class embodichain.lab.sim.atomic_actions.RuntimeEndpointTarget[source]

Stable controller destination produced by an endpoint adapter.

Targets contain immutable addressing data only. Live controllers, sockets, simulator entities, and other process-owned handles belong to an endpoint-command transport rather than this value.

Attributes:

`address_fingerprint`	Return the stable controller-address and safe-hold fingerprint.
`target_id`	Return the destination identifier within its transport.
`transport_id`	Return the registered transport kind used by this target.

Methods:

Return an independently owned target snapshot.

property address_fingerprint: Hashable

Return the stable controller-address and safe-hold fingerprint.

The default covers the exact target type and transport-scoped destination. Target types whose hold footprint depends on additional immutable addressing fields must override this property and include those fields. Replans and explicit revisions may replace payloads, but they may not change this fingerprint in place.

snapshot()[source]

Return an independently owned target snapshot.

Return type:: RuntimeEndpointTarget

abstract property target_id: str: Return the destination identifier within its transport.

abstract property transport_id: str: Return the registered transport kind used by this target.

class embodichain.lab.sim.atomic_actions.SceneEntityPose[source]

Late-bound pose derived from a versioned scene entity.

The semantic request remains stable while each call to AtomicAction.plan() resolves the latest scene pose. This is the bridge used by an execution session to replan moving goals.

Methods:

`__init__`(entity_id[, relative_pose, ...])
`snapshot`()	Return an independently owned late-bound pose value.

Attributes:

`entity_id`	Stable scene entity identifier.
`minimum_confidence`	Minimum accepted perception confidence.
`relative_pose`	Optional transform applied as `entity_pose @ relative_pose`.

__init__(entity_id, relative_pose=None, minimum_confidence=0.0)

entity_id: str: Stable scene entity identifier.

minimum_confidence: float: Minimum accepted perception confidence.

relative_pose: Tensor | None: Optional transform applied as entity_pose @ relative_pose.

snapshot()[source]

Return an independently owned late-bound pose value.

Return type:: SceneEntityPose
Returns:: Exact scene reference with an owned relative-pose tensor.

class embodichain.lab.sim.atomic_actions.SceneProvider[source]

Produce scene snapshots correlated with execution environments.

Implementations own scene-change detection and revision advancement. A snapshot’s entity rows must follow the supplied env_ids order. Scene and collision-world revisions must never regress for a stable environment.

Methods:

`__init__`(args, *kwargs)
`snapshot`(*, timestamp, env_ids)	Capture the latest versioned scene state.

__init__(*args, **kwargs)

snapshot(*, timestamp, env_ids)[source]

Capture the latest versioned scene state.

Parameters:

timestamp (float) – Observation timestamp supplied by the execution backend.
env_ids (Tensor) – Stable ordered environment correlation IDs.

Return type:

SceneSnapshot

Returns:

Scene snapshot whose batched entities follow env_ids order.

class embodichain.lab.sim.atomic_actions.SceneSnapshot[source]

Versioned scene state used to ground dynamic goals and obstacles.

Methods:

`__init__`(timestamp, version[, entities, ...])
`collision_obstacle_poses`(*, batch_size, ...)	Return collision obstacle poses in planning batch order.
`collision_world_revisions`(batch_size)	Expand the collision revision to one value per environment.
`empty`()	Create an empty initial scene snapshot.
`get_articulation_joint_state`(...)	Return an owned live joint observation for a canonical address.

Attributes:

`articulation_joints`	Live physical joint observations keyed by articulation and joint ID.
`collision_entity_ids`	Entity IDs whose poses update a planner's dynamic collision world.
`collision_world_revision`	Global or per-environment collision-world revision.

__init__(timestamp, version, entities=<factory>, collision_world_revision=0, collision_entity_ids=(), articulation_joints=<factory>)

articulation_joints: Mapping[tuple[str, str], ObservedArticulationJointState]: Live physical joint observations keyed by articulation and joint ID.

collision_entity_ids: tuple[str, ...]: Entity IDs whose poses update a planner’s dynamic collision world.

collision_obstacle_poses(*, batch_size, device, dtype)[source]

Return collision obstacle poses in planning batch order.

Parameters:

batch_size (int) – Number of planning environments.
device (device) – Planner tensor device.
dtype (dtype) – Planner tensor dtype.

Return type:

Mapping[str, Tensor]

Returns:

Mapping from configured collision entity ID to (B, 4, 4) pose.

collision_world_revision: int | tuple[int, ...]: Global or per-environment collision-world revision.

collision_world_revisions(batch_size)[source]

Expand the collision revision to one value per environment.

Parameters:: batch_size (int) – Number of environments represented by the planning context.
Return type:: tuple[int, ...]
Returns:: Per-environment monotonic revision tuple.
Raises:: ValueError – If an explicit revision tuple does not match the batch.

classmethod empty()[source]

Create an empty initial scene snapshot.

Return type:: SceneSnapshot

get_articulation_joint_state(articulation_id, joint_id)[source]

Return an owned live joint observation for a canonical address.

Return type:: ObservedArticulationJointState | None

class embodichain.lab.sim.atomic_actions.SimulationExecutionAdapter[source]

Adapt a simulation robot to observation, command, and clock protocols.

The adapter writes joint targets synchronously. Time advances only through sleep(), which converts the requested runner interval to an integral number of physics updates. This makes ExecutionRunner.run_until_blocked() deterministic and avoids wall-clock sleeps in headless simulation.

Parameters:

simulation (SimulationManager) – Simulation manager advanced by the execution clock.
robot (Robot) – Robot observed and commanded by the adapter.
physics_dt (float | None) – Optional physics period. Defaults to the simulation config.
control_dt (float | None) – Optional command period exposed to action interpolation. Defaults to physics_dt because that is the adapter’s minimum executable command cadence.
env_ids (Tensor | None) – Optional stable correlation IDs matching every robot row. They are not used as simulator indices; row order maps to robot instances.
scene_provider (SceneProvider | None) – Optional provider for versioned scene observations.
scene_supplier (Callable[[float], SceneSnapshot] | None) – Optional callback for versioned scene observations. It is mutually exclusive with scene_provider.
initial_time (float) – Initial elapsed simulation time in seconds.

Methods:

`__init__`(simulation, robot, *[, physics_dt, ...])
`cancel`(targets, *, timeout)	Acknowledge cancellation of synchronous simulation target writes.
`hold`(targets, context, *, timeout)	Set every represented joint endpoint to an observed-position hold.
`now`()	Return elapsed simulation time in seconds.
`observe`(task_state)	Capture full-robot state and the latest supplied scene snapshot.
`send`(command, *, timeout)	Write joint endpoint targets and neutralize inactive rows.
`sleep`(duration)	Advance physics by at least the requested duration.

Classes:

payload_type

alias of JointPositionPayload

__init__(simulation, robot, *, physics_dt=None, control_dt=None, env_ids=None, scene_provider=None, scene_supplier=None, initial_time=0.0)[source]

cancel(targets, *, timeout)[source]

Acknowledge cancellation of synchronous simulation target writes.

Parameters:

targets (tuple[RuntimeEndpointTarget, ...]) – Joint-position destinations whose queued work is cancelled.
timeout (float) – Positive acknowledgement deadline.

Return type:

Returns:

Accepted acknowledgement. The following hold call installs the actual safe target.

hold(targets, context, *, timeout)[source]

Set every represented joint endpoint to an observed-position hold.

Parameters:

targets (tuple[RuntimeEndpointTarget, ...]) – Joint-position destinations to place in a safe hold.
context (PlanningContext) – Latest observed positions and stable environment IDs.
timeout (float) – Positive acknowledgement deadline.

Return type:

Returns:

Accepted acknowledgement or a rejected diagnostic.

now()[source]

Return elapsed simulation time in seconds.

Return type:: float
Returns:: Elapsed simulation time in seconds.

observe(task_state)[source]

Capture full-robot state and the latest supplied scene snapshot.

Parameters:: task_state (TaskState) – Verified symbolic state owned by the execution session.
Return type:: PlanningContext
Returns:: Planning context timestamped with elapsed simulation time.

payload_type: alias of JointPositionPayload

send(command, *, timeout)[source]

Write joint endpoint targets and neutralize inactive rows.

Parameters:

command (RuntimeCommandFrame) – Joint-position endpoint frame. Inactive rows are replaced with observed positions by this transport.
timeout (float) – Positive acknowledgement deadline. Simulation writes are synchronous, so this is validated but otherwise unused.

Return type:

Returns:

Accepted acknowledgement or a rejected diagnostic.

sleep(duration)[source]

Advance physics by at least the requested duration.

Parameters:: duration (float) – Requested simulated duration in seconds.
Return type:: None

class embodichain.lab.sim.atomic_actions.SkillBindingContract[source]

Complete robot-independent binding contract for one atomic skill.

slots=() explicitly declares that a skill consumes no robot resource. None on SkillDescriptor instead means that no semantic binding contract was declared.

Methods:

__init__([slots, constraints])

Attributes:

slot_ids

Return required slot identifiers in declaration order.

__init__(slots=(), constraints=())

property slot_ids: tuple[str, ...]: Return required slot identifiers in declaration order.

class embodichain.lab.sim.atomic_actions.SkillDescriptor[source]

Machine-readable metadata for one registered atomic skill.

Methods:

__init__(skill_id, goal_type, options_type)

Attributes:

`binding_contract`	Explicit generic resource contract used by Task Program lowering.
`open_loop`	Whether completion reports motion execution without physical-effect proof.

__init__(skill_id, goal_type, options_type, agent_visible=True, open_loop=False, binding_contract=None)

binding_contract: SkillBindingContract | None: Explicit generic resource contract used by Task Program lowering.

open_loop: bool: Whether completion reports motion execution without physical-effect proof.

class embodichain.lab.sim.atomic_actions.SkillEndpointRequirement[source]

Capabilities and commands required from one slot-local endpoint.

Methods:

__init__(endpoint_id[, capabilities, ...])

Attributes:

`capabilities`	Open, namespaced all-of capability identifiers.
`endpoint_id`	Endpoint selector local to the containing participant slot.
`required_commands`	Semantic command names and their required typed command contracts.

__init__(endpoint_id, capabilities=frozenset({}), required_commands=<factory>)

capabilities: frozenset[str]: Open, namespaced all-of capability identifiers.

endpoint_id: str: Endpoint selector local to the containing participant slot.

required_commands: Mapping[str, type[ControlCommand]]: Semantic command names and their required typed command contracts.

class embodichain.lab.sim.atomic_actions.SkillResourceSlot[source]

One skill-local participant selected as an indivisible resource unit.

Methods:

__init__(slot_id, endpoints[, constraints])

Attributes:

`constraints`	Physical constraints among endpoint views in this participant.
`endpoints`	Endpoint requirements that the selected robot resource must satisfy.
`slot_id`	Skill-local participant name, such as `primary` or `source`.

__init__(slot_id, endpoints, constraints=())

constraints: tuple[DisjointSlotEndpoints, ...]: Physical constraints among endpoint views in this participant.

endpoints: tuple[SkillEndpointRequirement, ...]: Endpoint requirements that the selected robot resource must satisfy.

slot_id: str: Skill-local participant name, such as primary or source.

class embodichain.lab.sim.atomic_actions.Slide[source]

Open-loop approach, grasp, and axis-constrained sliding motion.

Classes:

`GoalType`	alias of `SlideGoal`
`OptionsType`	alias of `SlideOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`open_loop`	Whether the skill intentionally declares no verified physical effect.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of SlideGoal

OptionsType: alias of SlideOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='primary', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'open': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>, 'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)),), constraints=())

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

open_loop: ClassVar[bool] = True: Whether the skill intentionally declares no verified physical effect.

skill_id: ClassVar[str] = 'slide': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.SlideAffordance[source]

Target-local antipodal grasp and parent-joint translation geometry.

The positive translation-axis direction denotes approaching and pushing the articulated part closed. Pulling moves in the opposite direction. The mesh describes the actual graspable contact surface. The target pose is supplied separately by SceneEntityPose or a pose snapshot.

Methods:

`__init__`([object_label, custom_config, ...])
`resolve_from_object_geometry`(geometry)	Resolve the target-local prismatic axis and neighborhood sign.

Attributes:

`joint_limits`	Optional lower and upper translation limits in metres.
`joint_name`	Optional stable prismatic-joint name associated with the link.
`translation_axis`	Parent prismatic-joint axis, signed toward articulation geometry.

__init__(object_label='', custom_config=<factory>, translation_axis=<factory>, joint_name=None, joint_limits=None, *, mesh_vertices, mesh_triangles)

custom_config: dict[str, Any]: User-defined configuration payload.

joint_limits: tuple[float, float] | None = None: Optional lower and upper translation limits in metres.

joint_name: str | None = None: Optional stable prismatic-joint name associated with the link.

resolve_from_object_geometry(geometry)[source]

Resolve the target-local prismatic axis and neighborhood sign.

Return type:: None

translation_axis: Tensor: Parent prismatic-joint axis, signed toward articulation geometry.

class embodichain.lab.sim.atomic_actions.SlideGoal[source]

Translating articulation link described by a slide affordance.

Methods:

__init__(semantics, target_pose)

Attributes:

target_pose

Link pose snapshot or late-bound stable scene-entity reference.

__init__(semantics, target_pose)

target_pose: Tensor | SceneEntityPose: Link pose snapshot or late-bound stable scene-entity reference.

class embodichain.lab.sim.atomic_actions.SlideOptions[source]

Per-invocation sliding behavior for a translating articulation link.

Methods:

__init__([direction, hand_interp_steps, ...])

Attributes:

`approach_distance`	Pre-grasp distance opposite the approach/push axis.
`direction`	Whether to pull the part open or push it closed.
`hand_interp_steps`	Number of waypoints used for each close/open hand segment.
`translation_distance`	Distance traveled along the pull or push direction.

__init__(direction='pull', hand_interp_steps=5, approach_distance=0.1, translation_distance=0.15)

approach_distance: float: Pre-grasp distance opposite the approach/push axis.

direction: Literal['pull', 'push']: Whether to pull the part open or push it closed.

hand_interp_steps: int: Number of waypoints used for each close/open hand segment.

translation_distance: float: Distance traveled along the pull or push direction.

class embodichain.lab.sim.atomic_actions.StateDelta[source]

Expected task-state changes that require post-execution verification.

A mapping value of None removes the corresponding relation. Planning only declares this delta; an execution runtime applies it after verifying the semantic effect for the successful environment rows.

Methods:

`__init__`([held_object_updates, ...])
`apply`(state, update_mask)	Apply expected effects to selected environment rows.
`snapshot`()	Return an independently owned symbolic-effect snapshot.

Attributes:

`articulation_joint_updates`	Per-articulation/joint verified state replacements or removals.
`coordinated_held_object_updates`	Per-resource-pair coordinated attachment replacements or removals.
`held_object_updates`	Per-resource attachment replacements or removals.
`is_empty`	Whether this delta declares no symbolic state changes.

__init__(held_object_updates=<factory>, coordinated_held_object_updates=<factory>, articulation_joint_updates=<factory>)

apply(state, update_mask)[source]

Apply expected effects to selected environment rows.

This operation is used for hypothetical state propagation while compiling a sequence. A runtime must apply the same delta only after effect verification.

Parameters:

state (TaskState) – Input task state.
update_mask (Tensor) – Successful and verified rows, shape (num_envs,).

Return type:

Returns:

New task state with masked updates.

articulation_joint_updates: Mapping[tuple[str, str], ArticulationJointState | None]: Per-articulation/joint verified state replacements or removals.

coordinated_held_object_updates: Mapping[tuple[str, str], CoordinatedHeldObjectState | None]: Per-resource-pair coordinated attachment replacements or removals.

held_object_updates: Mapping[str, HeldObjectState | None]: Per-resource attachment replacements or removals.

property is_empty: bool: Whether this delta declares no symbolic state changes.

snapshot()[source]

Return an independently owned symbolic-effect snapshot.

Live simulation entities retain identity, while semantic metadata, affordance data, and every attachment tensor are copied.

Return type:: StateDelta
Returns:: Independently owned state delta.

class embodichain.lab.sim.atomic_actions.TaskState[source]

Symbolic task state, separate from measured robot state.

Methods:

`__init__`(batch_size, device[, held_objects, ...])
`empty`(batch_size, device)	Create an empty symbolic state.
`exclusive_held_object_mask`(resource)	Return environments where only `resource` holds its object.
`get_articulation_joint_state`(...)	Return verified state for one canonical articulation joint.
`get_coordinated_held_object`(first_resource, ...)	Return the relation for an ordered resource pair, if any.
`get_held_object`(resource)	Return the object held by `resource`, if any.
`held_object_mask`(resource)	Return environments where `resource` holds an object.

Attributes:

`articulation_joints`	Verified articulation states keyed by canonical articulation and joint IDs.
`batch_size`	Number of vectorized environments represented by the state.
`coordinated_held_objects`	Coordinated relations keyed by ordered logical task-state resource pairs.
`device`	Device used by per-environment masks and relation tensors.
`held_objects`	Single-manipulator held-object relations keyed by control resource.

__init__(batch_size, device, held_objects=<factory>, coordinated_held_objects=<factory>, articulation_joints=<factory>)

articulation_joints: Mapping[tuple[str, str], ArticulationJointState]: Verified articulation states keyed by canonical articulation and joint IDs.

batch_size: int: Number of vectorized environments represented by the state.

coordinated_held_objects: Mapping[tuple[str, str], CoordinatedHeldObjectState]: Coordinated relations keyed by ordered logical task-state resource pairs.

device: device | str: Device used by per-environment masks and relation tensors.

classmethod empty(batch_size, device)[source]

Create an empty symbolic state.

Parameters:

batch_size (int) – Number of represented environments.
device (device | str) – Tensor device used by the state.

Return type:

Returns:

Empty task state with explicit batch metadata.

exclusive_held_object_mask(resource)[source]

Return environments where only resource holds its object.

Object identity is established by the exact semantic record or by a shared non-null simulation entity. Labels and structural equality are deliberately ignored because distinct physical objects may look alike.

Parameters:: resource (str) – Manipulator control-resource name.
Return type:: Tensor
Returns:: Owned boolean mask with shape (batch_size,).

get_articulation_joint_state(articulation_id, joint_id)[source]

Return verified state for one canonical articulation joint.

Return type:: ArticulationJointState | None

get_coordinated_held_object(first_resource, second_resource)[source]

Return the relation for an ordered resource pair, if any.

Return type:: CoordinatedHeldObjectState | None

get_held_object(resource)[source]

Return the object held by resource, if any.

Return type:: HeldObjectState | None

held_object_mask(resource)[source]

Return environments where resource holds an object.

Parameters:: resource (str) – Manipulator control-resource name.
Return type:: Tensor
Returns:: Owned boolean mask with shape (batch_size,). Missing resources produce an all-false mask.

held_objects: Mapping[str, HeldObjectState]: Single-manipulator held-object relations keyed by control resource.

class embodichain.lab.sim.atomic_actions.TimedCommandSequence[source]

Ordered runtime command frames for one stable environment batch.

env_ids is authoritative even when frames is empty, preserving the batch size and device needed by compilation and execution boundaries.

Parameters:

frames (tuple[RuntimeCommandFrame, ...]) – Ordered command frames in execution order.
env_ids (Tensor) – Stable environment identifiers retained for empty sequences.

Methods:

`__init__`(frames, env_ids)
`snapshot`()	Return an independently owned timed sequence.

Attributes:

`batch_size`	Return the preserved environment batch size.
`device`	Return the preserved batch device.
`frame_count`	Return the number of command frames.
`targets`	Return unique owned destinations in first-use order.

__init__(frames, env_ids)

property batch_size: int: Return the preserved environment batch size.

property device: device: Return the preserved batch device.

property frame_count: int: Return the number of command frames.

snapshot()[source]

Return an independently owned timed sequence.

Return type:: TimedCommandSequence

property targets: tuple[RuntimeEndpointTarget, ...]: Return unique owned destinations in first-use order.

class embodichain.lab.sim.atomic_actions.TimedTerminalAcceptance[source]

Explicit terminal acceptance without endpoint feedback.

Methods:

__init__([settle_duration])

__init__(settle_duration=0.0)

class embodichain.lab.sim.atomic_actions.TimedTrackingSequence[source]

Tracking frames aligned by index with an authoritative command sequence.

Methods:

__init__(env_ids, frames)

Attributes:

`batch_size`	Return the represented environment count.
`device`	Return the sequence tensor device.
`frame_count`	Return the number of command-aligned tracking frames.

__init__(env_ids, frames)

property batch_size: int: Return the represented environment count.

property device: device: Return the sequence tensor device.

property frame_count: int: Return the number of command-aligned tracking frames.

class embodichain.lab.sim.atomic_actions.TimedTrajectory[source]

Full-robot joint trajectory with per-environment timing metadata.

Methods:

`__init__`(positions, velocities, ...)
`concatenate`(trajectories, *[, empty_like])	Concatenate trajectories along their waypoint dimension.
`empty`(*, batch_size, robot_dof, device, env_ids)	Create an empty trajectory with explicit batch and DoF dimensions.
`from_positions`(positions, *, env_ids, dt[, ...])	Build a trajectory from positions and explicit per-sample timing.
`from_uniform_step`(positions, *, env_ids, step_dt)	Build an explicitly uniform-time trajectory.
`hold_rows`(active_mask, hold_qpos)	Replace inactive rows with a fixed hold command.
`snapshot`()	Return an independently owned copy of this trajectory.

Attributes:

`batch_size`	Number of environment rows.
`dt`	Per-waypoint arrival intervals; the first sample normally has zero dt.
`duration`	Per-environment trajectory duration derived from waypoint intervals.
`robot_dof`	Number of full-robot command columns.
`waypoint_count`	Number of trajectory samples.

__init__(positions, velocities, accelerations, dt, env_ids)

property batch_size: int: Number of environment rows.

classmethod concatenate(trajectories, *, empty_like=None)[source]

Concatenate trajectories along their waypoint dimension.

Parameters:

trajectories (Sequence[TimedTrajectory]) – Compatible trajectories in execution order.
empty_like (PlanningContext | None) – Context used only when trajectories is empty.

Return type:

Returns:

Concatenated full-robot trajectory.

dt: Tensor: Per-waypoint arrival intervals; the first sample normally has zero dt.

property duration: Tensor: Per-environment trajectory duration derived from waypoint intervals.

classmethod empty(*, batch_size, robot_dof, device, env_ids)[source]

Create an empty trajectory with explicit batch and DoF dimensions.

Return type:: TimedTrajectory

classmethod from_positions(positions, *, env_ids, dt, velocities=None, accelerations=None)[source]

Build a trajectory from positions and explicit per-sample timing.

Parameters:

positions (Tensor) – Full-robot positions, shape (B, N, D).
env_ids (Tensor) – Environment identifiers, shape (B,).
dt (Tensor) – Per-sample arrival intervals, shape (B, N).
velocities (Tensor | None) – Optional joint velocities.
accelerations (Tensor | None) – Optional joint accelerations.

Return type:

Returns:

Validated timed trajectory.

classmethod from_uniform_step(positions, *, env_ids, step_dt, velocities=None, accelerations=None)[source]

Build an explicitly uniform-time trajectory.

The first waypoint has zero arrival time; every following waypoint uses step_dt. This factory is intended for interpolation algorithms whose cadence is selected by the caller, not for repairing untimed plans.

Parameters:

positions (Tensor) – Full-robot positions, shape (B, N, D).
env_ids (Tensor) – Environment identifiers, shape (B,).
step_dt (float) – Explicit interval between consecutive waypoints.
velocities (Tensor | None) – Optional joint velocities.
accelerations (Tensor | None) – Optional joint accelerations.

Return type:

Returns:

Validated uniformly timed trajectory.

hold_rows(active_mask, hold_qpos)[source]

Replace inactive rows with a fixed hold command.

Parameters:

active_mask (Tensor) – Rows allowed to execute this trajectory.
hold_qpos (Tensor) – Hold positions, shape (B, D).

Return type:

ArticulationAffordanceGeometry

Returns:

New trajectory with inactive rows frozen and derivatives zeroed.

property robot_dof: int: Number of full-robot command columns.

snapshot()[source]

Return an independently owned copy of this trajectory.

Return type:: TimedTrajectory
Returns:: A trajectory whose tensor storage can be mutated without changing the source trajectory.

property waypoint_count: int: Number of trajectory samples.

class embodichain.lab.sim.atomic_actions.TrackingCommandProjector[source]

Versioned pure projector from an endpoint command to desired state.

Methods:

`__init__`(args, *kwargs)
`project`(command, binding)	Project one command into the binding's desired tracking channel.

__init__(*args, **kwargs)

project(command, binding)[source]

Project one command into the binding’s desired tracking channel.

Return type:: TrackingState

class embodichain.lab.sim.atomic_actions.TrackingEvaluation[source]

Per-row metric result with unit-preserving component errors.

Methods:

__init__(channel_id, accepted_mask, ...[, ...])

__init__(channel_id, accepted_mask, valid_mask, normalized_error, component_errors=<factory>)

class embodichain.lab.sim.atomic_actions.TrackingEvaluatorRegistry[source]

Immutable exact-version metric-evaluator registry.

Methods:

__init__([evaluators])

__init__(evaluators=())[source]

class embodichain.lab.sim.atomic_actions.TrackingFeedbackAddress[source]

Immutable address understood by one tracking-feedback provider.

Attributes:

address_fingerprint

Return a stable, hashable address identity.

Methods:

snapshot()

Return an independently owned address snapshot.

abstract property address_fingerprint: Hashable: Return a stable, hashable address identity.

snapshot()[source]

Return an independently owned address snapshot.

Return type:: TrackingFeedbackAddress

class embodichain.lab.sim.atomic_actions.TrackingFeedbackBatch[source]

One synchronized typed observation from an exact feedback source.

Methods:

__init__(source, state, valid_mask, timestamp)

__init__(source, state, valid_mask, timestamp)

class embodichain.lab.sim.atomic_actions.TrackingFeedbackProvider[source]

Versioned live port that reads one exact tracking source.

Methods:

`__init__`(args, *kwargs)
`observe`(source, context)	Read one synchronized typed feedback batch.

__init__(*args, **kwargs)

observe(source, context)[source]

Read one synchronized typed feedback batch.

Return type:: TrackingFeedbackBatch

class embodichain.lab.sim.atomic_actions.TrackingFeedbackProviderRegistry[source]

Immutable exact-version feedback-provider registry.

Methods:

__init__([providers])

__init__(providers=())[source]

class embodichain.lab.sim.atomic_actions.TrackingFeedbackSourceRef[source]

Versioned provider route plus one immutable feedback address.

Methods:

`__init__`(provider_id, revision, address)
`snapshot`()	Return an independently owned source reference.

Attributes:

source_fingerprint

Return the exact versioned source identity.

__init__(provider_id, revision, address)

snapshot()[source]

Return an independently owned source reference.

Return type:: TrackingFeedbackSourceRef

property source_fingerprint: Hashable: Return the exact versioned source identity.

class embodichain.lab.sim.atomic_actions.TrackingFrame[source]

Desired endpoint states associated with one command frame.

Methods:

__init__([setpoints])

__init__(setpoints=())

class embodichain.lab.sim.atomic_actions.TrackingMetricCfg[source]

Immutable tolerance configuration dispatched by exact metric ID/revision.

Methods:

snapshot()

Return an independently owned metric configuration.

snapshot()[source]

Return an independently owned metric configuration.

Return type:: TrackingMetricCfg

class embodichain.lab.sim.atomic_actions.TrackingMetricEvaluator[source]

Versioned evaluator for one exact metric configuration type.

Methods:

`__init__`(args, *kwargs)
`evaluate`(desired, observed, valid_mask, metric)	Evaluate a desired and observed batch row by row.

__init__(*args, **kwargs)

evaluate(desired, observed, valid_mask, metric)[source]

Evaluate a desired and observed batch row by row.

Return type:: TrackingEvaluation

class embodichain.lab.sim.atomic_actions.TrackingPolicy[source]

Independent in-flight recovery signal and terminal acceptance contract.

Methods:

`__init__`(in_flight, terminal)
`joint_position`(*[, in_flight_max_abs_error, ...])	Create the built-in joint-position tracking and acceptance contract.
`timed`(*[, settle_duration])	Create an explicit time-only terminal contract with no tracking.

__init__(in_flight, terminal)

classmethod joint_position(*, in_flight_max_abs_error=0.05, terminal_max_abs_error=0.05, terminal_settle_timeout=0.5, consecutive_violations=1, consecutive_acceptances=1, grace_period=0.0)[source]

Create the built-in joint-position tracking and acceptance contract.

Return type:: TrackingPolicy

classmethod timed(*, settle_duration=0.0)[source]

Create an explicit time-only terminal contract with no tracking.

Return type:: TrackingPolicy

class embodichain.lab.sim.atomic_actions.TrackingProjectorRef[source]

Exact version of a command-to-tracking-state projector.

Methods:

`__init__`(projector_id, revision)
`snapshot`()	Return an independently owned projector route.

__init__(projector_id, revision)

snapshot()[source]

Return an independently owned projector route.

Return type:: TrackingProjectorRef

class embodichain.lab.sim.atomic_actions.TrackingProjectorRegistry[source]

Immutable exact-version command-projector registry.

Methods:

__init__([projectors])

__init__(projectors=())[source]

class embodichain.lab.sim.atomic_actions.TrackingRuntime[source]

Runtime facade for projecting commands and evaluating typed feedback.

Methods:

`__init__`(providers, projectors, evaluators)
`evaluate`(setpoint, feedback, metric)	Evaluate one observed setpoint with an exact metric implementation.
`evaluate_frame`(frame, metrics, context)	Observe and evaluate every setpoint required by one frame.
`observe`(setpoint, context)	Read the exact feedback source for one setpoint.
`project`(command, binding)	Project one command through the exact binding-owned projector.
`with_builtins`()	Create a runtime with context feedback and built-in typed metrics.

Attributes:

`evaluators`	Return the immutable exact-version evaluator registry.
`projectors`	Return the immutable exact-version projector registry.
`providers`	Return the immutable exact-version provider registry.

__init__(providers, projectors, evaluators)[source]

evaluate(setpoint, feedback, metric)[source]

Evaluate one observed setpoint with an exact metric implementation.

Return type:: TrackingEvaluation

evaluate_frame(frame, metrics, context)[source]

Observe and evaluate every setpoint required by one frame.

Return type:: Mapping[tuple[str, str, str], TrackingEvaluation]

property evaluators: TrackingEvaluatorRegistry: Return the immutable exact-version evaluator registry.

observe(setpoint, context)[source]

Read the exact feedback source for one setpoint.

Return type:: TrackingFeedbackBatch

project(command, binding)[source]

Project one command through the exact binding-owned projector.

Return type:: TrackingState

property projectors: TrackingProjectorRegistry: Return the immutable exact-version projector registry.

property providers: TrackingFeedbackProviderRegistry: Return the immutable exact-version provider registry.

classmethod with_builtins()[source]

Create a runtime with context feedback and built-in typed metrics.

Return type:: TrackingRuntime

class embodichain.lab.sim.atomic_actions.TrackingSetpoint[source]

One endpoint-local desired state and its typed feedback route.

Methods:

__init__(endpoint_key, binding, desired)

__init__(endpoint_key, binding, desired)

class embodichain.lab.sim.atomic_actions.TrackingState[source]

Immutable-by-ownership typed desired or observed tracking state.

Attributes:

`batch_size`	Return the represented environment count.
`device`	Return the tensor device.

Methods:

snapshot()

Return an independently owned state snapshot.

abstract property batch_size: int: Return the represented environment count.

abstract property device: device: Return the tensor device.

abstract snapshot()[source]

Return an independently owned state snapshot.

Return type:: TrackingState

class embodichain.lab.sim.atomic_actions.TrajectorySegment[source]

Named half-open waypoint range inside an action trajectory.

Segments describe semantic structure for inspection, visualization, and execution tracing. They do not create independent planning or recovery boundaries; recovery continues to operate on the enclosing action plan.

Methods:

`__init__`(name, start, stop)
`contains`(waypoint_index)	Return whether `waypoint_index` belongs to this segment.

Attributes:

waypoint_count

Number of waypoints in this segment.

__init__(name, start, stop)

contains(waypoint_index)[source]

Return whether waypoint_index belongs to this segment.

Return type:: bool

property waypoint_count: int: Number of waypoints in this segment.

class embodichain.lab.sim.atomic_actions.Twist[source]

Open-loop approach, grasp, twist, release, and retract motion.

Classes:

`GoalType`	alias of `TwistGoal`
`OptionsType`	alias of `TwistOptions`

Attributes:

`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`open_loop`	Whether the skill intentionally declares no verified physical effect.
`skill_id`	Stable registry identifier for this skill.

GoalType: alias of TwistGoal

OptionsType: alias of TwistOptions

binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(slots=(SkillResourceSlot(slot_id='primary', endpoints=(SkillEndpointRequirement(endpoint_id='motion', capabilities=frozenset({'kinematics.forward', 'motion.cartesian_pose'}), required_commands=mappingproxy({})), SkillEndpointRequirement(endpoint_id='grasp', capabilities=frozenset({'interaction.grasp'}), required_commands=mappingproxy({'open': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>, 'grasp': <class 'embodichain.lab.sim.atomic_actions.control.JointPositionCommand'>}))), constraints=(DisjointSlotEndpoints(endpoint_ids=('motion', 'grasp')),)),), constraints=())

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

open_loop: ClassVar[bool] = True: Whether the skill intentionally declares no verified physical effect.

skill_id: ClassVar[str] = 'twist': Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.TwistAffordance[source]

Target-local grasp point and parent-joint rotation geometry.

Methods:

`__init__`([object_label, custom_config, ...])
`get_grasp_pose`(target_pose)	Construct a deterministic world grasp pose from local geometry.
`require_axis_origin`()	Return the explicit or geometry-derived local rotation-axis point.
`resolve_from_object_geometry`(geometry)	Resolve the target-local revolute axis, sign, and joint origin.

Attributes:

`axis_origin`	Fallback axis point, overridden by revolute-joint origin metadata.
`grasp_position`	Explicit target-local center of the gripper contact region.
`joint_limits`	Optional lower and upper angular limits in radians.
`joint_name`	Optional stable articulation-joint name associated with the axis.
`twist_axis`	Parent revolute-joint axis, signed toward articulation geometry.

__init__(object_label='', custom_config=<factory>, twist_axis=<factory>, joint_name=None, joint_limits=None, *, grasp_position, axis_origin=None)

axis_origin: tuple[float, float, float] | None = None: Fallback axis point, overridden by revolute-joint origin metadata.

custom_config: dict[str, Any]: User-defined configuration payload.

get_grasp_pose(target_pose)[source]

Construct a deterministic world grasp pose from local geometry.

The pose z-axis follows twist_axis. The remaining axes are formed with an adaptive reference so the result is always in SO(3).

Return type:: Tensor
Returns:: Batched world-frame grasp poses with shape (B, 4, 4).
Raises:: ValueError – If target_pose is not a batched pose tensor.

grasp_position: tuple[float, float, float]: Explicit target-local center of the gripper contact region.

joint_limits: tuple[float, float] | None = None: Optional lower and upper angular limits in radians.

joint_name: str | None = None: Optional stable articulation-joint name associated with the axis.

require_axis_origin()[source]

Return the explicit or geometry-derived local rotation-axis point.

Return type:: tuple[float, float, float]
Returns:: Resolved target-local rotation-axis origin.
Raises:: ValueError – If neither a fallback nor articulation geometry supplied the rotation-axis origin.

resolve_from_object_geometry(geometry)[source]

Resolve the target-local revolute axis, sign, and joint origin.

Return type:: None

twist_axis: Tensor: Parent revolute-joint axis, signed toward articulation geometry.

class embodichain.lab.sim.atomic_actions.TwistGoal[source]

Target object described by a twist affordance.

Methods:

__init__(semantics, target_pose)

Attributes:

target_pose

Target pose snapshot or late-bound stable scene-entity reference.

__init__(semantics, target_pose)

target_pose: Tensor | SceneEntityPose: Target pose snapshot or late-bound stable scene-entity reference.

class embodichain.lab.sim.atomic_actions.TwistOptions[source]

Per-invocation twisting behavior.

Methods:

__init__([hand_interp_steps, ...])

Attributes:

`hand_interp_steps`	Number of waypoints used for each close/open hand segment.
`pre_grasp_distance`	Distance from the grasp pose along its negative z-axis.
`twist_angle`	Requested twist rotation in radians.
`twist_waypoint_count`	Number of Cartesian keyframes along the target's circular twist arc.

__init__(hand_interp_steps=5, twist_waypoint_count=8, pre_grasp_distance=0.1, twist_angle=0.7853981633974483)

hand_interp_steps: int: Number of waypoints used for each close/open hand segment.

pre_grasp_distance: float: Distance from the grasp pose along its negative z-axis.

twist_angle: float: Requested twist rotation in radians.

twist_waypoint_count: int: Number of Cartesian keyframes along the target’s circular twist arc.

class embodichain.lab.sim.atomic_actions.WholeBodyPoseTrackingEvaluator[source]

Evaluator for WholeBodyPoseTrackingMetric.

Classes:

metric_type

alias of WholeBodyPoseTrackingMetric

metric_type: alias of WholeBodyPoseTrackingMetric

class embodichain.lab.sim.atomic_actions.WholeBodyPoseTrackingMetric[source]

Independent base-pose and joint-position tolerances.

Methods:

__init__([translation_tolerance, ...])

__init__(translation_tolerance=0.02, rotation_tolerance=0.05, joint_position_tolerance=0.05)

class embodichain.lab.sim.atomic_actions.WholeBodyPoseTrackingState[source]

Batched base poses and joint positions for whole-body tracking.

Methods:

`__init__`(root_poses, joint_positions)
`snapshot`()	Return an independently owned state snapshot.

Attributes:

`batch_size`	Return the represented environment count.
`device`	Return the tensor device.

__init__(root_poses, joint_positions)

property batch_size: int: Return the represented environment count.

property device: device: Return the tensor device.

snapshot()[source]

Return an independently owned state snapshot.

Return type:: WholeBodyPoseTrackingState

embodichain.lab.sim.atomic_actions.create_simulation_atomic_action_engine(motion_generator, scene_entities, control_profiles=None, grasp_pose_generators=None, *, load_builtins=True, tracking_runtime=None)[source]

Create an engine whose initial context observes selected rigid objects.

This is the direct-simulation convenience path for offline planning. Entity IDs are derived from each rigid object’s stable uid; only explicitly supplied objects are observed. Advanced integrations that need aliases, articulation/link state, collision roles, or an external perception source should construct AtomicActionEngine with their own SceneProvider instead.

Parameters:

motion_generator (MotionGenerator) – Motion-generation backend owned by the engine.
scene_entities (Sequence[RigidObject]) – Non-empty sequence of simulation rigid objects to expose in automatically captured initial scene snapshots.
control_profiles (Mapping[str, ControlPartCommandProfile] | None) – Semantic commands keyed by robot control-part name.
grasp_pose_generators (Mapping[str, GraspPoseGenerator] | None) – Grasp-pose services keyed by grasp endpoint target.
load_builtins (bool) – Whether to install all built-in atomic actions.
tracking_runtime (TrackingRuntime | None) – Optional typed tracking runtime shared by action plans.

Return type:

AtomicActionEngine

Returns:

Engine configured with a rigid-object scene provider.

Raises:

TypeError – If scene_entities is not a sequence.
ValueError – If an entity lacks a stable UID or UIDs are duplicated.

embodichain.lab.sim.atomic_actions.sample_initial_articulation_geometry(provider, target_link_name, *, initial_qpos, initial_qpos_joint_names, body_scale, articulation_point_count=100000, target_point_count=5000)[source]

Sample initial articulation geometry for Atomic Action affordances.

The adapter evaluates FK at an explicitly named initial joint state, transforms every raw link mesh into the target link’s initial local frame, merges those meshes, and samples the target and merged surfaces with Open3D. It also transforms the nearest prismatic and revolute ancestor joint geometry into that frame.

Parameters:

provider (ArticulationGeometryProvider) – Structural source of deterministic link meshes, FK, and parent-joint topology.
target_link_name (str) – Link whose initial local frame defines the result.
initial_qpos (Tensor | Sequence[float]) – Initial joint positions with shape (J,).
initial_qpos_joint_names (Sequence[str]) – Names corresponding to initial_qpos.
body_scale (Tensor | Sequence[float]) – Configured articulation scale. Only unit scale is currently supported because raw meshes and FK must share one metric frame.
articulation_point_count (int) – Merged-articulation surface sample count. The same count is used for the non-target merged surface when one exists.
target_point_count (int) – Target-link surface sample count.

Return type:

Returns:

Owned typed affordance geometry in the target link’s initial frame.

Raises:

TypeError – If a name, joint-name sequence, or point count has the wrong type.
ValueError – If the target, initial state, scale, FK output, topology, or mesh geometry is invalid.

Articulation geometry adaptation#

The adapter converts deterministic articulation meshes, FK, and parent-joint topology into sampled Atomic Action affordance geometry. Initial state and scale are explicit inputs; the simulation object does not own sampling or semantic geometry keys.

class embodichain.lab.sim.atomic_actions.articulation_geometry.ArticulationGeometryProvider[source]#

Deterministic articulation facts required for geometry adaptation.

Implementations provide raw link meshes, FK, and immediate-parent-first joint topology. Initial configuration and scale are explicit adapter inputs so this protocol does not depend on an ArticulationCfg or PK-chain API.

device#: Device on which geometry tensors are assembled.

link_names#: Stable articulation link names.

Methods:

`__init__`(args, *kwargs)
`compute_fk`(qpos, *, link_names, qpos_joint_names)	Return root-frame link poses for the supplied named joint state.
`get_link_vert_face`(link_name)	Return one link-local triangle mesh.
`get_parent_joint_chain`(link_name)	Return parent joints ordered from the link toward the root.

__init__(*args, **kwargs)#

compute_fk(qpos, *, link_names, qpos_joint_names)[source]#

Return root-frame link poses for the supplied named joint state.

Parameters:

qpos (Tensor) – Joint positions with shape (B, J).
link_names (Sequence[str]) – Links whose poses should be returned.
qpos_joint_names (Sequence[str]) – Names corresponding to the last qpos axis.

Return type:

Tensor

Returns:

Root-frame poses with shape (B, L, 4, 4).

get_link_vert_face(link_name)[source]#

Return one link-local triangle mesh.

Parameters:: link_name (str) – Stable link name.
Return type:: tuple[Tensor, Tensor]
Returns:: Link-local vertices and triangle indices.

Attention

A non-empty link mesh must contain at least one non-degenerate triangle surface.

get_parent_joint_chain(link_name)[source]#

Return parent joints ordered from the link toward the root.

Parameters:: link_name (str) – Link at which to begin traversal.
Return type:: tuple[ArticulationJointGeometry, ...]
Returns:: Immediate-parent-first structural joint geometry.

class embodichain.lab.sim.atomic_actions.articulation_geometry.ArticulationJointGeometry[source]#

Structural joint geometry consumed by the articulation adapter.

name#: Stable joint name.

joint_type#: Normalized joint type such as fixed, prismatic, or revolute.

parent_link_name#: Stable parent-link name.

child_link_name#: Stable child-link name.

origin_pose#: Joint-frame pose in the parent-link frame, shape (4, 4).

axis#: Joint axis in the joint frame, shape (3,).

Methods:

__init__(*args, **kwargs)

__init__(*args, **kwargs)#

class embodichain.lab.sim.atomic_actions.articulation_geometry.ArticulationAffordanceGeometry[source]#

Owned sampled geometry for articulation-link affordances.

The point clouds and optional joint data are expressed in the target link’s initial local frame. Joint axes are normalized. Use to_object_geometry() at the ObjectSemantics boundary so the Atomic Action-specific string-key protocol remains in this module.

Parameters:

target_link_point_cloud (Tensor) – Sampled target-link surface, shape (N, 3).
articulation_point_cloud (Tensor) – Sampled merged-articulation surface, shape (M, 3).
prismatic_joint_axis (Tensor | None) – Optional nearest prismatic ancestor axis.
revolute_joint_axis (Tensor | None) – Optional nearest revolute ancestor axis.
revolute_axis_origin (Tensor | None) – Optional matching revolute-joint origin.
non_target_articulation_point_cloud (Tensor | None) – Sampled merged surface of every link except the target, shape (K, 3). An empty tensor records that the articulation has no non-target link surface. None is accepted only for geometry created without source-link provenance.

Methods:

`__init__`(target_link_point_cloud, ...[, ...])
`to_object_geometry`()	Return an owned `ObjectSemantics.geometry` dictionary.

__init__(target_link_point_cloud, articulation_point_cloud, prismatic_joint_axis=None, revolute_joint_axis=None, revolute_axis_origin=None, non_target_articulation_point_cloud=None)#

to_object_geometry()[source]#

Return an owned ObjectSemantics.geometry dictionary.

Return type:: dict[str, Tensor]
Returns:: A new real dictionary containing cloned point clouds and whichever optional joint entries are available.

embodichain.lab.sim.atomic_actions.articulation_geometry.sample_initial_articulation_geometry(provider, target_link_name, *, initial_qpos, initial_qpos_joint_names, body_scale, articulation_point_count=100000, target_point_count=5000)[source]#

Sample initial articulation geometry for Atomic Action affordances.

The adapter evaluates FK at an explicitly named initial joint state, transforms every raw link mesh into the target link’s initial local frame, merges those meshes, and samples the target and merged surfaces with Open3D. It also transforms the nearest prismatic and revolute ancestor joint geometry into that frame.

Parameters:

provider (ArticulationGeometryProvider) – Structural source of deterministic link meshes, FK, and parent-joint topology.
target_link_name (str) – Link whose initial local frame defines the result.
initial_qpos (Tensor | Sequence[float]) – Initial joint positions with shape (J,).
initial_qpos_joint_names (Sequence[str]) – Names corresponding to initial_qpos.
body_scale (Tensor | Sequence[float]) – Configured articulation scale. Only unit scale is currently supported because raw meshes and FK must share one metric frame.
articulation_point_count (int) – Merged-articulation surface sample count. The same count is used for the non-target merged surface when one exists.
target_point_count (int) – Target-link surface sample count.

Return type:

ArticulationAffordanceGeometry

Returns:

Owned typed affordance geometry in the target link’s initial frame.

Raises:

TypeError – If a name, joint-name sequence, or point count has the wrong type.
ValueError – If the target, initial state, scale, FK output, topology, or mesh geometry is invalid.

Semantic resource contracts#

class embodichain.lab.sim.atomic_actions.SkillDescriptor[source]#

Machine-readable metadata for one registered atomic skill.

Methods:

__init__(skill_id, goal_type, options_type)

Attributes:

`binding_contract`	Explicit generic resource contract used by Task Program lowering.
`open_loop`	Whether completion reports motion execution without physical-effect proof.

__init__(skill_id, goal_type, options_type, agent_visible=True, open_loop=False, binding_contract=None)#

binding_contract: SkillBindingContract | None#: Explicit generic resource contract used by Task Program lowering.

open_loop: bool#: Whether completion reports motion execution without physical-effect proof.

class embodichain.lab.sim.atomic_actions.SkillBindingContract[source]#

Complete robot-independent binding contract for one atomic skill.

slots=() explicitly declares that a skill consumes no robot resource. None on SkillDescriptor instead means that no semantic binding contract was declared.

Methods:

__init__([slots, constraints])

Attributes:

slot_ids

Return required slot identifiers in declaration order.

__init__(slots=(), constraints=())#

property slot_ids: tuple[str, ...]#: Return required slot identifiers in declaration order.

class embodichain.lab.sim.atomic_actions.SkillResourceSlot[source]#

One skill-local participant selected as an indivisible resource unit.

Methods:

__init__(slot_id, endpoints[, constraints])

Attributes:

`constraints`	Physical constraints among endpoint views in this participant.
`endpoints`	Endpoint requirements that the selected robot resource must satisfy.
`slot_id`	Skill-local participant name, such as `primary` or `source`.

__init__(slot_id, endpoints, constraints=())#

constraints: tuple[DisjointSlotEndpoints, ...]#: Physical constraints among endpoint views in this participant.

endpoints: tuple[SkillEndpointRequirement, ...]#: Endpoint requirements that the selected robot resource must satisfy.

slot_id: str#: Skill-local participant name, such as primary or source.

class embodichain.lab.sim.atomic_actions.SkillEndpointRequirement[source]#

Capabilities and commands required from one slot-local endpoint.

Methods:

__init__(endpoint_id[, capabilities, ...])

Attributes:

`capabilities`	Open, namespaced all-of capability identifiers.
`endpoint_id`	Endpoint selector local to the containing participant slot.
`required_commands`	Semantic command names and their required typed command contracts.

__init__(endpoint_id, capabilities=frozenset({}), required_commands=<factory>)#

capabilities: frozenset[str]#: Open, namespaced all-of capability identifiers.

endpoint_id: str#: Endpoint selector local to the containing participant slot.

required_commands: Mapping[str, type[ControlCommand]]#: Semantic command names and their required typed command contracts.

class embodichain.lab.sim.atomic_actions.DisjointSlotEndpoints[source]#

Require selected endpoints within one participant to be disjoint.

Methods:

__init__(endpoint_ids)

__init__(endpoint_ids)#

class embodichain.lab.sim.atomic_actions.DisjointResourceSlots[source]#

Require selected slots to have pairwise-disjoint physical claims.

Methods:

__init__(slots)

__init__(slots)#

Standard capability identifiers#

embodichain.lab.sim.atomic_actions.JOINT_POSITION_CAPABILITY = 'motion.joint_position'#

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

embodichain.lab.sim.atomic_actions.CARTESIAN_POSE_CAPABILITY = 'motion.cartesian_pose'#

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

embodichain.lab.sim.atomic_actions.FORWARD_KINEMATICS_CAPABILITY = 'kinematics.forward'#

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

embodichain.lab.sim.atomic_actions.INVERSE_KINEMATICS_CAPABILITY = 'kinematics.inverse'#

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

embodichain.lab.sim.atomic_actions.BATCH_INVERSE_KINEMATICS_CAPABILITY = 'kinematics.batch_inverse'#

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

embodichain.lab.sim.atomic_actions.GRASP_CAPABILITY = 'interaction.grasp'#

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

Planning and state#

class embodichain.lab.sim.atomic_actions.ActionBinding[source]#

Engine-owned generic endpoint bindings for one atomic action call.

Methods:

`__init__`(owner_id[, endpoints])
`endpoint`(slot_id, endpoint_id)	Return one action-local resolved endpoint.
`with_command_overrides`(overrides)	Return a binding snapshot with endpoint-scoped command overrides.

Attributes:

`endpoint_keys`	Return action-local endpoint keys in binding order.
`targets`	Return unique owned runtime targets in binding order.

__init__(owner_id, endpoints=())#

endpoint(slot_id, endpoint_id)[source]#

Return one action-local resolved endpoint.

Return type:: EndpointBinding

property endpoint_keys: tuple[tuple[str, str], ...]#: Return action-local endpoint keys in binding order.

property targets: tuple[RuntimeEndpointTarget, ...]#: Return unique owned runtime targets in binding order.

with_command_overrides(overrides)[source]#

Return a binding snapshot with endpoint-scoped command overrides.

Return type:: ActionBinding

class embodichain.lab.sim.atomic_actions.EndpointBinding[source]#

One action-local endpoint resolved to a runtime controller target.

Methods:

`__init__`(slot_id, endpoint_id, resource_id, ...)
`command`(name)	Return one owned semantic-command snapshot.
`joint_positions`(name, *, num_envs, device[, ...])	Resolve a named joint-position command for a planning batch.
`require_target`(target_type)	Return the runtime target after an explicit type check.
`snapshot`()	Return an independently owned endpoint-binding snapshot.
`tracking_channel`(channel_id)	Return one independently owned typed tracking-channel binding.
`with_commands`(overrides)	Return an endpoint snapshot with semantic-command overrides.

Attributes:

`destination_key`	Return the transport-scoped physical destination key.
`key`	Return the action-local `(slot, endpoint)` key.
`task_state_key`	Symbolic task-state key; defaults to `target.target_id`.

__init__(slot_id, endpoint_id, resource_id, adapter_id, target, task_state_key=None, tracking_channels=<factory>, capabilities=frozenset({}), commands=<factory>, claim_tokens=frozenset({}), joint_ids=())#

command(name)[source]#

Return one owned semantic-command snapshot.

Return type:: ControlCommand

property destination_key: tuple[str, str]#: Return the transport-scoped physical destination key.

joint_positions(name, *, num_envs, device, dtype=None)[source]#

Resolve a named joint-position command for a planning batch.

Return type:: Tensor

property key: tuple[str, str]#: Return the action-local (slot, endpoint) key.

require_target(target_type)[source]#

Return the runtime target after an explicit type check.

Return type:: TypeVar(TargetT, bound= RuntimeEndpointTarget)

snapshot()[source]#

Return an independently owned endpoint-binding snapshot.

Return type:: EndpointBinding

task_state_key: str | None#: Symbolic task-state key; defaults to target.target_id.

tracking_channel(channel_id)[source]#

Return one independently owned typed tracking-channel binding.

Return type:: EndpointTrackingChannelBinding

with_commands(overrides)[source]#

Return an endpoint snapshot with semantic-command overrides.

Return type:: EndpointBinding

class embodichain.lab.sim.atomic_actions.RuntimeEndpointTarget[source]#

Stable controller destination produced by an endpoint adapter.

Targets contain immutable addressing data only. Live controllers, sockets, simulator entities, and other process-owned handles belong to an endpoint-command transport rather than this value.

Attributes:

`address_fingerprint`	Return the stable controller-address and safe-hold fingerprint.
`target_id`	Return the destination identifier within its transport.
`transport_id`	Return the registered transport kind used by this target.

Methods:

Return an independently owned target snapshot.

property address_fingerprint: Hashable#

Return the stable controller-address and safe-hold fingerprint.

The default covers the exact target type and transport-scoped destination. Target types whose hold footprint depends on additional immutable addressing fields must override this property and include those fields. Replans and explicit revisions may replace payloads, but they may not change this fingerprint in place.

snapshot()[source]#

Return an independently owned target snapshot.

Return type:: RuntimeEndpointTarget

abstract property target_id: str#: Return the destination identifier within its transport.

abstract property transport_id: str#: Return the registered transport kind used by this target.

class embodichain.lab.sim.atomic_actions.JointPositionTarget[source]#

Joint-position destination backed by one robot control part.

Methods:

__init__(control_part, joint_ids)

Attributes:

`address_fingerprint`	Return the destination plus the joints that must remain holdable.
`target_id`	Return the robot control-part destination.
`transport_id`	Return the built-in joint-position transport identifier.

__init__(control_part, joint_ids)#

property address_fingerprint: Hashable#: Return the destination plus the joints that must remain holdable.

property target_id: str#: Return the robot control-part destination.

property transport_id: str#: Return the built-in joint-position transport identifier.

class embodichain.lab.sim.atomic_actions.ControlCommand[source]#

Immutable-by-ownership command associated with one control part.

Command subclasses own their payload and must return another owned value from snapshot(). This keeps engine profiles and resolved invocation requests isolated from caller-owned mutable tensors.

Methods:

`equivalent_to`(other)	Return whether `other` has exactly the same command semantics.
`snapshot`()	Return an independently owned copy of this command.

abstract equivalent_to(other)[source]#

Return whether other has exactly the same command semantics.

Return type:: bool

abstract snapshot()[source]#

Return an independently owned copy of this command.

Return type:: ControlCommand

class embodichain.lab.sim.atomic_actions.JointPositionCommand[source]#

A semantic command represented by one or batched joint positions.

positions has shape (control_dof,) or (num_envs, control_dof). A one-dimensional command is broadcast to the planning batch when resolved.

Methods:

`__init__`(positions)
`equivalent_to`(other)	Return whether `other` owns identical joint positions.
`resolve`(*, num_envs, control_dof, device[, ...])	Validate, move, and broadcast this command for a planning batch.
`snapshot`()	Return an independently owned command snapshot.

Attributes:

positions

Return an owned copy of the command payload.

__init__(positions)[source]#

equivalent_to(other)[source]#

Return whether other owns identical joint positions.

Return type:: bool

property positions: Tensor#: Return an owned copy of the command payload.

resolve(*, num_envs, control_dof, device, dtype=None)[source]#

Validate, move, and broadcast this command for a planning batch.

Parameters:

num_envs (int) – Number of selected environments.
control_dof (int) – Joint count of the resolved control part.
device (device | str) – Target planning device.
dtype (dtype | None) – Optional target dtype.

Return type:

Tensor

Returns:

Independently owned tensor with shape (num_envs, control_dof).

Raises:

ValueError – If the command shape does not match the control part or selected environment batch.

snapshot()[source]#

Return an independently owned command snapshot.

Return type:: JointPositionCommand

class embodichain.lab.sim.atomic_actions.ControlPartCommandProfile[source]#

Reusable semantic commands for one named robot control part.

Profiles are registered once on AtomicActionEngine, keyed by names from Robot.control_parts. They describe embodiment-specific meanings such as open, grasp or ready without coupling those values to an action implementation.

Methods:

`__init__`([commands])
`joint_positions`(**commands)	Build a profile whose entries are joint-position commands.
`snapshot`()	Return an independently owned profile snapshot.

__init__(commands=<factory>)#

classmethod joint_positions(**commands)[source]#

Build a profile whose entries are joint-position commands.

Return type:: ControlPartCommandProfile

snapshot()[source]#

Return an independently owned profile snapshot.

Return type:: ControlPartCommandProfile

class embodichain.lab.sim.atomic_actions.ActionControlOverrides[source]#

Per-invocation semantic commands keyed by slot and endpoint.

The first two keys match a skill’s (slot_id, endpoint_id) contract. The innermost mapping contains semantic command names. Overrides are captured in the invocation revision’s immutable planning snapshot.

Methods:

`__init__`([endpoints])
`as_flat_mapping`()	Return immutable overrides keyed by `(slot_id, endpoint_id)`.

Attributes:

is_empty

Whether this invocation defines no command overrides.

__init__(endpoints=<factory>)#

as_flat_mapping()[source]#

Return immutable overrides keyed by (slot_id, endpoint_id).

Return type:: Mapping[tuple[str, str], Mapping[str, ControlCommand]]

property is_empty: bool#: Whether this invocation defines no command overrides.

class embodichain.lab.sim.atomic_actions.ActionInvocation[source]#

One fully typed and endpoint-bound atomic skill request.

This is a runtime-domain object, not the JSON protocol emitted by an MLLM. An action compiler is responsible for converting a semantic SkillCallSpec into this grounded representation.

Methods:

__init__(skill_id, goal, binding[, ...])

Attributes:

`binding`	Generic skill endpoint bindings owned by the selected engine.
`control_overrides`	Optional semantic control commands for this invocation revision.
`goal`	Action-specific goal value object.
`invocation_id`	Optional correlation identifier propagated into execution traces.
`motion_policy`	Reusable motion-generation settings.
`phase_effect_gates`	Physical-effect gates enforced at named trajectory-segment entries.
`recovery_policy`	Bounded local execution recovery settings.
`revision`	Monotonic revision used when replacing a runtime invocation.
`skill_id`	Stable registered skill identifier.
`skill_options`	Optional per-invocation behavior override for the selected skill.
`tracking_policy`	Typed in-flight tracking and terminal-acceptance settings.

__init__(skill_id, goal, binding, motion_policy=<factory>, tracking_policy=<factory>, recovery_policy=<factory>, phase_effect_gates=(), skill_options=None, control_overrides=<factory>, invocation_id=None, revision=0)#

binding: ActionBinding#: Generic skill endpoint bindings owned by the selected engine.

control_overrides: ActionControlOverrides#: Optional semantic control commands for this invocation revision.

goal: TypeVar(GoalT)#: Action-specific goal value object.

invocation_id: str | None#: Optional correlation identifier propagated into execution traces.

motion_policy: MotionPolicy#: Reusable motion-generation settings.

phase_effect_gates: tuple[PhaseEffectGateRequirement, ...]#: Physical-effect gates enforced at named trajectory-segment entries.

recovery_policy: RecoveryPolicy#: Bounded local execution recovery settings.

revision: int#: Monotonic revision used when replacing a runtime invocation.

skill_id: str#: Stable registered skill identifier.

skill_options: Optional[TypeVar(OptionsT, bound= ActionOptions)]#: Optional per-invocation behavior override for the selected skill.

tracking_policy: TrackingPolicy#: Typed in-flight tracking and terminal-acceptance settings.

class embodichain.lab.sim.atomic_actions.ResolvedActionRequest[source]#

Engine-owned immutable planning snapshot for one invocation revision.

Recovery replans reuse this object verbatim and vary only the PlanningContext. Deep-copying goal value payloads, policies, and skill options severs caller-owned mutable data before planning starts while retaining simulator-backed entity handles and private runtime caches.

Methods:

`__init__`(skill_id, goal, binding, ...[, ...])
`snapshot`()	Return an independently owned resolved-request snapshot.

__init__(skill_id, goal, binding, motion_policy, tracking_policy, recovery_policy, skill_options, phase_effect_gates=(), invocation_id=None, revision=0)#

snapshot()[source]#

Return an independently owned resolved-request snapshot.

Return type:: ResolvedActionRequest[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]

class embodichain.lab.sim.atomic_actions.ActionOptions[source]#

Marker base for immutable, skill-specific runtime options.

Subclasses belong to action modules and contain only behavior that may vary between invocations. Robot resources and semantic targets do not belong in this object.

Methods:

__init__()

__init__()#

class embodichain.lab.sim.atomic_actions.MotionPolicy[source]#

Immutable motion-generation policy for one action invocation.

The policy is a runtime value object rather than application configuration. plan_opts is copied on construction so a caller-owned planner config cannot change an invocation after it has been created.

Attributes:

`dynamic_collision_mode`	How this invocation consumes live scene-snapshot collision entities.
`plan_opts`	Optional typed planner-specific options.
`sample_count`	Requested trajectory sample count when the backend does not preserve samples.
`strategy`	`motion_gen` or `ik_interp`.

Methods:

to_motion_gen_options(*, start_qpos, ...[, ...])

Translate this atomic policy into motion-generator options.

dynamic_collision_mode: DynamicCollisionMode#: How this invocation consumes live scene-snapshot collision entities.

plan_opts: PlanOptions | None#: Optional typed planner-specific options.

sample_count: int#: Requested trajectory sample count when the backend does not preserve samples.

strategy: Literal['motion_gen', 'ik_interp']#

motion_gen or ik_interp.

Type:: Motion strategy

to_motion_gen_options(*, start_qpos, control_part, sample_count=None, interpolation_dt=None, cartesian_linear=False)[source]#

Translate this atomic policy into motion-generator options.

Parameters:

start_qpos (Tensor) – Observed controlled-joint start positions.
control_part (str) – Bound robot control-part name.
sample_count (int | None) – Optional segment-local sample-count override.
interpolation_dt (float | None) – Explicit waypoint interval used only by deterministic interpolation.
cartesian_linear (bool) – Whether every supplied Cartesian keyframe is a required linear-path sample rather than a sparse endpoint.

Return type:

MotionGenOptions

Returns:

Independently owned options for MotionGenerator.

class embodichain.lab.sim.atomic_actions.RecoveryPolicy[source]#

Bounded local recovery policy used by the execution runtime.

Attributes:

`action_timeout`	Maximum time for one action attempt, including terminal effect verification.
`goal_rotation_threshold`	Dynamic-goal rotation threshold in radians (five degrees by default).
`goal_translation_threshold`	Dynamic-goal translation threshold in metres.
`max_action_retries`	Maximum whole-action retries after planning, execution, or effect failure.
`max_replans`	Maximum replans within one action attempt.

action_timeout: float#: Maximum time for one action attempt, including terminal effect verification.

goal_rotation_threshold: float#: Dynamic-goal rotation threshold in radians (five degrees by default).

goal_translation_threshold: float#: Dynamic-goal translation threshold in metres.

max_action_retries: int#: Maximum whole-action retries after planning, execution, or effect failure.

max_replans: int#: Maximum replans within one action attempt.

class embodichain.lab.sim.atomic_actions.PlanningContext[source]#

Complete side-effect-free input to AtomicAction.plan().

Methods:

`__init__`(robot, task, scene, env_ids[, ...])
`get_articulation_joint_state`(...)	Return verified state for one canonical articulation joint.
`get_coordinated_held_object`(first_resource, ...)	Return a coordinated held-object relation, if any.
`get_held_object`(resource)	Return the object held by `resource`, if any.
`project`(*, qpos, task)	Create the hypothetical context used to compile a following action.
`require_control_dt`()	Return the explicit command period required for interpolation.

Attributes:

`articulation_joints`	Verified articulation-joint states.
`batch_size`	Number of environments in this planning request.
`control_dt`	Explicit command period used by action-owned interpolation.
`coordinated_held_objects`	Coordinated held-object relations.
`held_objects`	Single-resource held-object relations.
`last_qpos`	Measured joint positions used as the planning start state.

__init__(robot, task, scene, env_ids, control_dt=None)#

property articulation_joints: Mapping[tuple[str, str], ArticulationJointState]#: Verified articulation-joint states.

property batch_size: int#: Number of environments in this planning request.

control_dt: float | None#: Explicit command period used by action-owned interpolation.

property coordinated_held_objects: Mapping[tuple[str, str], CoordinatedHeldObjectState]#: Coordinated held-object relations.

get_articulation_joint_state(articulation_id, joint_id)[source]#

Return verified state for one canonical articulation joint.

Return type:: ArticulationJointState | None

get_coordinated_held_object(first_resource, second_resource)[source]#

Return a coordinated held-object relation, if any.

Return type:: CoordinatedHeldObjectState | None

get_held_object(resource)[source]#

Return the object held by resource, if any.

Return type:: HeldObjectState | None

property held_objects: Mapping[str, HeldObjectState]#: Single-resource held-object relations.

property last_qpos: Tensor#: Measured joint positions used as the planning start state.

project(*, qpos, task)[source]#

Create the hypothetical context used to compile a following action.

Parameters:

qpos (Tensor) – Projected terminal joint positions.
task (TaskState) – Task state after applying expected effects.

Return type:

Returns:

New context. No measured state or simulator state is mutated.

require_control_dt()[source]#

Return the explicit command period required for interpolation.

Raises:: ValueError – If the caller did not provide control_dt.
Return type:: float

class embodichain.lab.sim.atomic_actions.RobotObservation[source]#

Measured robot state used as the start of planning or replanning.

Methods:

`__init__`(timestamp, qpos, qvel[, qeffort, ...])
`with_qpos`(qpos)	Create a projected observation with a new position and zero velocity.

Attributes:

`batch_size`	Number of represented vectorized environments.
`robot_dof`	Number of robot joint-position columns.

__init__(timestamp, qpos, qvel, qeffort=None, root_pose=None, root_twist=None)#

property batch_size: int#: Number of represented vectorized environments.

property robot_dof: int#: Number of robot joint-position columns.

with_qpos(qpos)[source]#

Create a projected observation with a new position and zero velocity.

Parameters:: qpos (Tensor) – Projected joint positions with the same shape as this observation.
Return type:: RobotObservation
Returns:: New observation suitable for compiling the next action.

class embodichain.lab.sim.atomic_actions.TaskState[source]#

Symbolic task state, separate from measured robot state.

Methods:

`__init__`(batch_size, device[, held_objects, ...])
`empty`(batch_size, device)	Create an empty symbolic state.
`exclusive_held_object_mask`(resource)	Return environments where only `resource` holds its object.
`get_articulation_joint_state`(...)	Return verified state for one canonical articulation joint.
`get_coordinated_held_object`(first_resource, ...)	Return the relation for an ordered resource pair, if any.
`get_held_object`(resource)	Return the object held by `resource`, if any.
`held_object_mask`(resource)	Return environments where `resource` holds an object.

Attributes:

`articulation_joints`	Verified articulation states keyed by canonical articulation and joint IDs.
`batch_size`	Number of vectorized environments represented by the state.
`coordinated_held_objects`	Coordinated relations keyed by ordered logical task-state resource pairs.
`device`	Device used by per-environment masks and relation tensors.
`held_objects`	Single-manipulator held-object relations keyed by control resource.

__init__(batch_size, device, held_objects=<factory>, coordinated_held_objects=<factory>, articulation_joints=<factory>)#

articulation_joints: Mapping[tuple[str, str], ArticulationJointState]#: Verified articulation states keyed by canonical articulation and joint IDs.

batch_size: int#: Number of vectorized environments represented by the state.

coordinated_held_objects: Mapping[tuple[str, str], CoordinatedHeldObjectState]#: Coordinated relations keyed by ordered logical task-state resource pairs.

device: device | str#: Device used by per-environment masks and relation tensors.

classmethod empty(batch_size, device)[source]#

Create an empty symbolic state.

Parameters:

batch_size (int) – Number of represented environments.
device (device | str) – Tensor device used by the state.

Return type:

Returns:

Empty task state with explicit batch metadata.

exclusive_held_object_mask(resource)[source]#

Return environments where only resource holds its object.

Object identity is established by the exact semantic record or by a shared non-null simulation entity. Labels and structural equality are deliberately ignored because distinct physical objects may look alike.

Parameters:: resource (str) – Manipulator control-resource name.
Return type:: Tensor
Returns:: Owned boolean mask with shape (batch_size,).

get_articulation_joint_state(articulation_id, joint_id)[source]#

Return verified state for one canonical articulation joint.

Return type:: ArticulationJointState | None

get_coordinated_held_object(first_resource, second_resource)[source]#

Return the relation for an ordered resource pair, if any.

Return type:: CoordinatedHeldObjectState | None

get_held_object(resource)[source]#

Return the object held by resource, if any.

Return type:: HeldObjectState | None

held_object_mask(resource)[source]#

Return environments where resource holds an object.

Parameters:: resource (str) – Manipulator control-resource name.
Return type:: Tensor
Returns:: Owned boolean mask with shape (batch_size,). Missing resources produce an all-false mask.

held_objects: Mapping[str, HeldObjectState]#: Single-manipulator held-object relations keyed by control resource.

class embodichain.lab.sim.atomic_actions.SceneSnapshot[source]#

Versioned scene state used to ground dynamic goals and obstacles.

Methods:

`__init__`(timestamp, version[, entities, ...])
`collision_obstacle_poses`(*, batch_size, ...)	Return collision obstacle poses in planning batch order.
`collision_world_revisions`(batch_size)	Expand the collision revision to one value per environment.
`empty`()	Create an empty initial scene snapshot.
`get_articulation_joint_state`(...)	Return an owned live joint observation for a canonical address.

Attributes:

`articulation_joints`	Live physical joint observations keyed by articulation and joint ID.
`collision_entity_ids`	Entity IDs whose poses update a planner's dynamic collision world.
`collision_world_revision`	Global or per-environment collision-world revision.

__init__(timestamp, version, entities=<factory>, collision_world_revision=0, collision_entity_ids=(), articulation_joints=<factory>)#

articulation_joints: Mapping[tuple[str, str], ObservedArticulationJointState]#: Live physical joint observations keyed by articulation and joint ID.

collision_entity_ids: tuple[str, ...]#: Entity IDs whose poses update a planner’s dynamic collision world.

collision_obstacle_poses(*, batch_size, device, dtype)[source]#

Return collision obstacle poses in planning batch order.

Parameters:

batch_size (int) – Number of planning environments.
device (device) – Planner tensor device.
dtype (dtype) – Planner tensor dtype.

Return type:

Mapping[str, Tensor]

Returns:

Mapping from configured collision entity ID to (B, 4, 4) pose.

collision_world_revision: int | tuple[int, ...]#: Global or per-environment collision-world revision.

collision_world_revisions(batch_size)[source]#

Expand the collision revision to one value per environment.

Parameters:: batch_size (int) – Number of environments represented by the planning context.
Return type:: tuple[int, ...]
Returns:: Per-environment monotonic revision tuple.
Raises:: ValueError – If an explicit revision tuple does not match the batch.

classmethod empty()[source]#

Create an empty initial scene snapshot.

Return type:: SceneSnapshot

get_articulation_joint_state(articulation_id, joint_id)[source]#

Return an owned live joint observation for a canonical address.

Return type:: ObservedArticulationJointState | None

class embodichain.lab.sim.atomic_actions.SceneEntityPose[source]#

Late-bound pose derived from a versioned scene entity.

The semantic request remains stable while each call to AtomicAction.plan() resolves the latest scene pose. This is the bridge used by an execution session to replan moving goals.

Methods:

`__init__`(entity_id[, relative_pose, ...])
`snapshot`()	Return an independently owned late-bound pose value.

Attributes:

`entity_id`	Stable scene entity identifier.
`minimum_confidence`	Minimum accepted perception confidence.
`relative_pose`	Optional transform applied as `entity_pose @ relative_pose`.

__init__(entity_id, relative_pose=None, minimum_confidence=0.0)#

entity_id: str#: Stable scene entity identifier.

minimum_confidence: float#: Minimum accepted perception confidence.

relative_pose: Tensor | None#: Optional transform applied as entity_pose @ relative_pose.

snapshot()[source]#

Return an independently owned late-bound pose value.

Return type:: SceneEntityPose
Returns:: Exact scene reference with an owned relative-pose tensor.

class embodichain.lab.sim.atomic_actions.StateDelta[source]#

Expected task-state changes that require post-execution verification.

A mapping value of None removes the corresponding relation. Planning only declares this delta; an execution runtime applies it after verifying the semantic effect for the successful environment rows.

Methods:

`__init__`([held_object_updates, ...])
`apply`(state, update_mask)	Apply expected effects to selected environment rows.
`snapshot`()	Return an independently owned symbolic-effect snapshot.

Attributes:

`articulation_joint_updates`	Per-articulation/joint verified state replacements or removals.
`coordinated_held_object_updates`	Per-resource-pair coordinated attachment replacements or removals.
`held_object_updates`	Per-resource attachment replacements or removals.
`is_empty`	Whether this delta declares no symbolic state changes.

__init__(held_object_updates=<factory>, coordinated_held_object_updates=<factory>, articulation_joint_updates=<factory>)#

apply(state, update_mask)[source]#

Apply expected effects to selected environment rows.

This operation is used for hypothetical state propagation while compiling a sequence. A runtime must apply the same delta only after effect verification.

Parameters:

state (TaskState) – Input task state.
update_mask (Tensor) – Successful and verified rows, shape (num_envs,).

Return type:

Returns:

New task state with masked updates.

articulation_joint_updates: Mapping[tuple[str, str], ArticulationJointState | None]#: Per-articulation/joint verified state replacements or removals.

coordinated_held_object_updates: Mapping[tuple[str, str], CoordinatedHeldObjectState | None]#: Per-resource-pair coordinated attachment replacements or removals.

held_object_updates: Mapping[str, HeldObjectState | None]#: Per-resource attachment replacements or removals.

property is_empty: bool#: Whether this delta declares no symbolic state changes.

snapshot()[source]#

Return an independently owned symbolic-effect snapshot.

Live simulation entities retain identity, while semantic metadata, affordance data, and every attachment tensor are copied.

Return type:: StateDelta
Returns:: Independently owned state delta.

class embodichain.lab.sim.atomic_actions.TimedTrajectory[source]#

Full-robot joint trajectory with per-environment timing metadata.

Methods:

`__init__`(positions, velocities, ...)
`concatenate`(trajectories, *[, empty_like])	Concatenate trajectories along their waypoint dimension.
`empty`(*, batch_size, robot_dof, device, env_ids)	Create an empty trajectory with explicit batch and DoF dimensions.
`from_positions`(positions, *, env_ids, dt[, ...])	Build a trajectory from positions and explicit per-sample timing.
`from_uniform_step`(positions, *, env_ids, step_dt)	Build an explicitly uniform-time trajectory.
`hold_rows`(active_mask, hold_qpos)	Replace inactive rows with a fixed hold command.
`snapshot`()	Return an independently owned copy of this trajectory.

Attributes:

`batch_size`	Number of environment rows.
`dt`	Per-waypoint arrival intervals; the first sample normally has zero dt.
`duration`	Per-environment trajectory duration derived from waypoint intervals.
`robot_dof`	Number of full-robot command columns.
`waypoint_count`	Number of trajectory samples.

__init__(positions, velocities, accelerations, dt, env_ids)#

property batch_size: int#: Number of environment rows.

classmethod concatenate(trajectories, *, empty_like=None)[source]#

Concatenate trajectories along their waypoint dimension.

Parameters:

trajectories (Sequence[TimedTrajectory]) – Compatible trajectories in execution order.
empty_like (PlanningContext | None) – Context used only when trajectories is empty.

Return type:

Returns:

Concatenated full-robot trajectory.

dt: Tensor#: Per-waypoint arrival intervals; the first sample normally has zero dt.

property duration: Tensor#: Per-environment trajectory duration derived from waypoint intervals.

classmethod empty(*, batch_size, robot_dof, device, env_ids)[source]#

Create an empty trajectory with explicit batch and DoF dimensions.

Return type:: TimedTrajectory

classmethod from_positions(positions, *, env_ids, dt, velocities=None, accelerations=None)[source]#

Build a trajectory from positions and explicit per-sample timing.

Parameters:

positions (Tensor) – Full-robot positions, shape (B, N, D).
env_ids (Tensor) – Environment identifiers, shape (B,).
dt (Tensor) – Per-sample arrival intervals, shape (B, N).
velocities (Tensor | None) – Optional joint velocities.
accelerations (Tensor | None) – Optional joint accelerations.

Return type:

Returns:

Validated timed trajectory.

classmethod from_uniform_step(positions, *, env_ids, step_dt, velocities=None, accelerations=None)[source]#

Build an explicitly uniform-time trajectory.

The first waypoint has zero arrival time; every following waypoint uses step_dt. This factory is intended for interpolation algorithms whose cadence is selected by the caller, not for repairing untimed plans.

Parameters:

positions (Tensor) – Full-robot positions, shape (B, N, D).
env_ids (Tensor) – Environment identifiers, shape (B,).
step_dt (float) – Explicit interval between consecutive waypoints.
velocities (Tensor | None) – Optional joint velocities.
accelerations (Tensor | None) – Optional joint accelerations.

Return type:

Returns:

Validated uniformly timed trajectory.

hold_rows(active_mask, hold_qpos)[source]#

Replace inactive rows with a fixed hold command.

Parameters:

active_mask (Tensor) – Rows allowed to execute this trajectory.
hold_qpos (Tensor) – Hold positions, shape (B, D).

Return type:

Returns:

New trajectory with inactive rows frozen and derivatives zeroed.

property robot_dof: int#: Number of full-robot command columns.

snapshot()[source]#

Return an independently owned copy of this trajectory.

Return type:: TimedTrajectory
Returns:: A trajectory whose tensor storage can be mutated without changing the source trajectory.

property waypoint_count: int#: Number of trajectory samples.

class embodichain.lab.sim.atomic_actions.PlanningFailure[source]#

Stable planning-failure classification used by recovery policy.

Parameters:

code (str) – Exact machine-readable failure code.
retryable (bool) – Whether action-level recovery may replan failed rows.

Methods:

__init__(code[, retryable])

__init__(code, retryable=True)#

class embodichain.lab.sim.atomic_actions.RuntimeCommandPayload[source]#

Immutable-by-ownership payload submitted to one runtime transport.

Attributes:

`batch_size`	Return the number of environment rows in this payload.
`device`	Return the device shared by this payload's batched values.
`transport_id`	Return the transport kind that accepts this payload.

Methods:

Return an independently owned payload snapshot.

abstract property batch_size: int#: Return the number of environment rows in this payload.

abstract property device: device#: Return the device shared by this payload’s batched values.

abstract snapshot()[source]#

Return an independently owned payload snapshot.

Return type:: RuntimeCommandPayload

abstract property transport_id: str#: Return the transport kind that accepts this payload.

class embodichain.lab.sim.atomic_actions.JointPositionPayload[source]#

Batched joint-position targets for the built-in robot transport.

Parameters:

positions (Tensor) – Joint positions with shape (batch_size, control_dof).
velocities (Tensor | None) – Optional joint velocities with the same shape and device.

Methods:

`__init__`(positions[, velocities])
`snapshot`()	Return an independently owned joint payload.

Attributes:

`batch_size`	Return the number of environment rows.
`device`	Return the tensor device.
`dof`	Return the number of controlled joints.
`transport_id`	Return the built-in joint-position transport identifier.

__init__(positions, velocities=None)#

property batch_size: int#: Return the number of environment rows.

property device: device#: Return the tensor device.

property dof: int#: Return the number of controlled joints.

snapshot()[source]#

Return an independently owned joint payload.

Return type:: JointPositionPayload

property transport_id: str#: Return the built-in joint-position transport identifier.

class embodichain.lab.sim.atomic_actions.EndpointCommand[source]#

One transport-compatible payload addressed to one runtime target.

Parameters:

target (RuntimeEndpointTarget) – Immutable destination resolved from an action endpoint.
payload (RuntimeCommandPayload) – Batched command value accepted by the target transport.

Methods:

`__init__`(target, payload)
`snapshot`()	Return an independently owned endpoint command.

Attributes:

`batch_size`	Return the payload batch size.
`destination_key`	Return the transport-scoped destination identifier.
`device`	Return the payload device.
`transport_id`	Return the common target and payload transport identifier.

__init__(target, payload)#

property batch_size: int#: Return the payload batch size.

property destination_key: tuple[str, str]#: Return the transport-scoped destination identifier.

property device: device#: Return the payload device.

snapshot()[source]#

Return an independently owned endpoint command.

Return type:: EndpointCommand

property transport_id: str#: Return the common target and payload transport identifier.

class embodichain.lab.sim.atomic_actions.RuntimeCommandFrame[source]#

Synchronized endpoint commands for one batched runtime instant.

Parameters:

commands (tuple[EndpointCommand, ...]) – Commands dispatched together for this frame.
active_mask (Tensor) – Boolean environment rows allowed to execute commands. Transports must actively neutralize addressed targets for false rows rather than leaving a previously persistent command running.
env_ids (Tensor) – Stable environment identifiers for the batch rows.
hold_duration (Tensor) – Per-row delay before advancing to the next frame.

Methods:

`__init__`(commands, active_mask, env_ids, ...)
`snapshot`()	Return an independently owned command frame.
`with_active_mask`(active_mask)	Return a frame snapshot with a replacement active-row mask.

Attributes:

`batch_size`	Return the number of environment rows.
`device`	Return the shared frame device.
`targets`	Return owned targets in command order.

__init__(commands, active_mask, env_ids, hold_duration)#

property batch_size: int#: Return the number of environment rows.

property device: device#: Return the shared frame device.

snapshot()[source]#

Return an independently owned command frame.

Return type:: RuntimeCommandFrame

property targets: tuple[RuntimeEndpointTarget, ...]#: Return owned targets in command order.

with_active_mask(active_mask)[source]#

Return a frame snapshot with a replacement active-row mask.

Parameters:: active_mask (Tensor) – Boolean mask with one value per environment row.
Return type:: RuntimeCommandFrame
Returns:: Independently owned frame with unchanged commands and timing.

class embodichain.lab.sim.atomic_actions.TimedCommandSequence[source]#

Ordered runtime command frames for one stable environment batch.

env_ids is authoritative even when frames is empty, preserving the batch size and device needed by compilation and execution boundaries.

Parameters:

frames (tuple[RuntimeCommandFrame, ...]) – Ordered command frames in execution order.
env_ids (Tensor) – Stable environment identifiers retained for empty sequences.

Methods:

`__init__`(frames, env_ids)
`snapshot`()	Return an independently owned timed sequence.

Attributes:

`batch_size`	Return the preserved environment batch size.
`device`	Return the preserved batch device.
`frame_count`	Return the number of command frames.
`targets`	Return unique owned destinations in first-use order.

__init__(frames, env_ids)#

property batch_size: int#: Return the preserved environment batch size.

property device: device#: Return the preserved batch device.

property frame_count: int#: Return the number of command frames.

snapshot()[source]#

Return an independently owned timed sequence.

Return type:: TimedCommandSequence

property targets: tuple[RuntimeEndpointTarget, ...]#: Return unique owned destinations in first-use order.

class embodichain.lab.sim.atomic_actions.ActionPlan[source]#

Scene-bound planning result for one grounded atomic action invocation.

An action owns one timed command sequence and one recovery boundary. Named TrajectorySegment values describe semantic structure within that sequence without implying independent planning or recovery boundaries.

expected_effects#: Terminal symbolic state changes committed only after physical-effect verification succeeds.

effect_candidates#: Nonterminal attachment baselines available to phase gates and in-flight guards without being committed to task state.

scene_dependency_monitor_until#: Optional exclusive waypoint-index upper bounds for individual scene_dependencies. An entity is monitored while the current waypoint index is smaller than its bound; 0 disables monitoring immediately, while an omitted entity remains monitored for the action’s full execution. Once the bound is reached, all pose changes for that entity are ignored, regardless of whether they were caused by the action or by an external disturbance.

scene_dependency_end_segment#: Optional last segment during which scene motion may invalidate and replan the action for every dependency.

Methods:

`__init__`(skill_id, plan_success, commands, ...)
`segment`(name)	Return a named trajectory segment.
`segment_at`(waypoint_index)	Return the segment containing a global action waypoint index.
`snapshot`()	Return an independently owned inspection snapshot of this plan.

Attributes:

`requires_effect_verification`	Whether execution must verify a terminal physical effect.
`success_all`	Whether every environment row planned successfully.

__init__(skill_id, plan_success, commands, recovery_policy, tracking_policy, planned_scene_version, planned_collision_world_revision, diagnostics, tracking=None, joint_trajectory=None, segments=(), scene_dependencies=(), scene_dependency_monitor_until=<factory>, scene_dependency_end_segment=None, collision_world_sensitive=False, replannable=True, expected_effects=<factory>, effect_candidates=<factory>, effect_verification=None, invocation_id=None, invocation_revision=0)#

property requires_effect_verification: bool#: Whether execution must verify a terminal physical effect.

segment(name)[source]#

Return a named trajectory segment.

Parameters:: name (str) – Exact segment name.
Return type:: TrajectorySegment
Returns:: Matching segment metadata.
Raises:: KeyError – If the plan has no segment with that name.

segment_at(waypoint_index)[source]#

Return the segment containing a global action waypoint index.

Return type:: TrajectorySegment

snapshot()[source]#

Return an independently owned inspection snapshot of this plan.

Runtime tracing and visualization need access to the exact plan that reached an execution boundary without being able to mutate the live session. Reconstructing the value through the public constructor also re-applies every plan invariant and snapshots all tensor-owning nested contracts.

Return type:: ActionPlan
Returns:: A validated plan with independently owned tensor storage.

property success_all: bool#: Whether every environment row planned successfully.

Engine and execution#

class embodichain.lab.sim.atomic_actions.AtomicAction[source]#

Side-effect-free planner for one semantically meaningful robot skill.

Actions own only typed default runtime options. An AtomicActionEngine binds its shared planning services before an action is invoked.

Attributes:

`GoalType`	Concrete goal dataclass or dataclasses accepted by this skill.
`agent_visible`	Whether an Action Agent should expose this skill by default.
`binding_contract`	Explicit robot-independent requirements for semantic discovery.
`default_options`	Return an owned copy of the action's default runtime options.
`device`	Return the concrete runtime device associated with the engine.
`is_bound`	Whether an engine has supplied this action's planning resources.
`motion_generator`	Return the engine-owned motion generator borrowed by this action.
`num_envs`	Number of environments owned by the bound robot.
`open_loop`	Whether the skill intentionally declares no verified physical effect.
`planning_services`	Return the engine-owned services borrowed by this action.
`robot`	Return the robot associated with the owning engine.
`robot_dof`	Number of full-robot degrees of freedom.
`skill_id`	Stable registry identifier for this skill.

Classes:

OptionsType

alias of ActionOptions

Methods:

`__init__`([default_options])
`build_command_plan`(request, context, *, ...)	Build a plan from transport-neutral runtime command frames.
`build_plan`(request, context, *, success, ...)	Build a validated action plan for a primitive implementation.
`descriptor`()	Return stable metadata used by registries and Action Agent adapters.
`failed_plan`(request, context, *[, message, ...])	Build a failed empty plan without changing task state.
`plan`(request, context)	Bind the current collision scene and invoke the skill planner.
`require_goal`(request)	Validate a resolved request and return its concrete goal.
`resolve_request`(invocation)	Validate and snapshot an invocation through engine-owned resources.

GoalType: ClassVar[type[Any] | tuple[type[Any], ...]]#: Concrete goal dataclass or dataclasses accepted by this skill.

OptionsType#: alias of ActionOptions

__init__(default_options=None)[source]#

agent_visible: ClassVar[bool] = True#: Whether an Action Agent should expose this skill by default.

binding_contract: ClassVar[SkillBindingContract | None] = None#

Explicit robot-independent requirements for semantic discovery.

Concrete action classes must declare this attribute in their own class body to opt into the semantic catalog. Inheriting another action’s contract does not silently expose a new skill identifier.

build_command_plan(request, context, *, success, commands, expected_effects=None, effect_candidates=None, effect_verification=None, replannable=True, diagnostics=None, segment_lengths=None, scene_dependency_monitor_until=None, scene_dependency_end_segment=None, joint_trajectory=None)[source]#

Build a plan from transport-neutral runtime command frames.

Tracking targets are projected from the command payloads through the typed channels declared by each bound endpoint. Semantic effects remain externally verified through the execution session.

Parameters:

request (ResolvedActionRequest[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]) – Resolved invocation snapshot being planned.
context (PlanningContext) – Planning input used for the plan.
success (bool | Tensor) – Per-environment planning success or scalar planner result.
commands (TimedCommandSequence) – Transport-neutral command sequence for the action.
expected_effects (StateDelta | None) – Symbolic effects to verify after execution.
effect_candidates (StateDelta | None) – Planned attachment baselines used by phase gates and in-flight held-object guards without committing task state.
effect_verification (EffectVerificationRequirement | None) – Optional explicit physical-effect boundary.
replannable (bool) – Whether the execution runtime may replan this action.
diagnostics (PlannerDiagnostics | None) – Optional retained planner diagnostics.
segment_lengths (Mapping[str, int] | None) – Optional ordered mapping from semantic segment names to command-frame counts. Zero-length entries are omitted.
scene_dependency_monitor_until (Mapping[str, int] | None) – Optional per-entity exclusive command-frame-index upper bound for scene-motion invalidation. An entity is monitored while the current frame index is smaller than its bound. 0 disables monitoring immediately; omitted dependencies remain monitored for the full action. Once the bound is reached, all pose changes for that entity are ignored.
scene_dependency_end_segment (str | None) – Optional last segment during which scene motion may invalidate and replan the action for every dependency.
joint_trajectory (TimedTrajectory | None) – Optional joint trajectory retained for offline compilation and inspection.

Return type:

Returns:

Side-effect-free action plan.

build_plan(request, context, *, success, trajectory, expected_effects=None, effect_candidates=None, effect_verification=None, replannable=True, diagnostics=None, segment_lengths=None, scene_dependency_monitor_until=None, scene_dependency_end_segment=None)[source]#

Build a validated action plan for a primitive implementation.

Parameters:

request (ResolvedActionRequest[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]) – Resolved invocation snapshot being planned.
context (PlanningContext) – Planning input used for the plan.
success (bool | Tensor) – Per-environment planning success or scalar planner result.
trajectory (TimedTrajectory) – Full-robot trajectory with explicit timing.
expected_effects (StateDelta | None) – Symbolic effects to verify after execution.
effect_candidates (StateDelta | None) – Planned attachment baselines used by phase gates and in-flight held-object guards without committing task state.
effect_verification (EffectVerificationRequirement | None) – Optional explicit physical-effect boundary. Use this when verification is required without a symbolic task- state delta.
replannable (bool) – Whether the execution runtime may replan this action.
diagnostics (PlannerDiagnostics | None) – Optional retained planner diagnostics.
segment_lengths (Mapping[str, int] | None) – Optional ordered mapping from semantic segment names to waypoint counts. Zero-length entries are omitted.
scene_dependency_monitor_until (Mapping[str, int] | None) – Optional per-entity exclusive waypoint-index upper bound for scene-motion invalidation. An entity is monitored while the current waypoint index is smaller than its bound. 0 disables monitoring immediately; omitted dependencies remain monitored for the full action. Once the bound is reached, all pose changes for that entity are ignored.
scene_dependency_end_segment (str | None) – Optional last segment during which scene motion may invalidate and replan the action for every dependency.

Return type:

Returns:

Side-effect-free action plan.

property default_options: OptionsT#: Return an owned copy of the action’s default runtime options.

classmethod descriptor()[source]#

Return stable metadata used by registries and Action Agent adapters.

Return type:: SkillDescriptor

property device: device#: Return the concrete runtime device associated with the engine.

failed_plan(request, context, *, message=None, failure_code='planning_failed', retryable=True)[source]#

Build a failed empty plan without changing task state.

Parameters:

request (ResolvedActionRequest[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]) – Resolved invocation that failed to plan.
context (PlanningContext) – Planning input used for the attempt.
message (str | None) – Optional diagnostic message.
failure_code (str) – Stable machine-readable planning failure code.
retryable (bool) – Whether execution may spend action-retry budget on the failed rows.

Return type:

Returns:

Failed action plan with an empty trajectory.

property is_bound: bool#: Whether an engine has supplied this action’s planning resources.

property motion_generator: MotionGenerator#: Return the engine-owned motion generator borrowed by this action.

property num_envs: int#: Number of environments owned by the bound robot.

open_loop: ClassVar[bool] = False#: Whether the skill intentionally declares no verified physical effect.

plan(request, context)[source]#

Bind the current collision scene and invoke the skill planner.

Parameters:

request (ResolvedActionRequest[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]) – Immutable, typed, and embodiment-resolved action request.
context (PlanningContext) – Latest observed robot, task, and scene state.

Return type:

Returns:

Scene-bound action plan with expected, uncommitted effects.

property planning_services: ActionPlanningServices#

Return the engine-owned services borrowed by this action.

Raises:: RuntimeError – If the action has not been registered or planned by an AtomicActionEngine.

require_goal(request)[source]#

Validate a resolved request and return its concrete goal.

Return type:: TypeVar(GoalT)

resolve_request(invocation)[source]#

Validate and snapshot an invocation through engine-owned resources.

Parameters:

invocation (ActionInvocation[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]) – Caller-owned invocation to resolve.

Return type:

ResolvedActionRequest[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]

Returns:

Immutable request reused by planning and recovery replans.

Raises:

ValueError – If the stable skill identifier does not match.
TypeError – If the goal or options type is incompatible.
KeyError – If a required binding role is missing.

property robot: Robot#: Return the robot associated with the owning engine.

property robot_dof: int#: Number of full-robot degrees of freedom.

skill_id: ClassVar[str]#: Stable registry identifier for this skill.

class embodichain.lab.sim.atomic_actions.AtomicActionEngine[source]#

Own planning resources and coordinate side-effect-free atomic actions.

Methods:

`__init__`(motion_generator[, ...])	Initialize one engine and bind its built-in action implementations.
`bind_control_parts`(skill_id, endpoints, *[, ...])	Build an advanced direct-core binding from control-part names.
`compile`(invocations[, context])	Compile a static sequence of grounded invocations.
`initial_context`(*[, task, scene, timestamp, ...])	Capture the robot state needed to start offline compilation.
`make_invocation`(skill_id, goal, *[, ...])	Construct a grounded invocation while naming the skill only once.
`plan`(invocation[, context])	Plan one registered invocation through the engine-owned backend.
`plan_request`(request[, context])	Plan an already-resolved request without rebuilding its snapshot.
`register`(action, *[, replace])	Register one action instance using its descriptor.
`resolve`(invocation)	Resolve a registered invocation into an engine-owned snapshot.
`start`(invocations[, context, eligible_mask])	Start incremental execution for a grounded invocation sequence.

Attributes:

`actions`	Registered action instances keyed by stable skill identifier.
`binding_owner_id`	Return the opaque owner identity required by action bindings.
`control_profiles`	Semantic command profiles registered for robot control parts.
`device`	Return the concrete planning device used by this engine.
`grasp_pose_generators`	Standalone grasp-pose services installed for endpoint targets.
`motion_generator`	Return the single motion generator owned by this engine.
`planning_services`	Engine-owned resources shared by every bound atomic action.
`robot`	Return the robot controlled by this engine.
`skill_catalog_revision`	Return the monotonic installed Atomic Skill catalog revision.
`skills`	Return explicitly declared, agent-visible installed skill metadata.
`tracking_runtime`	Typed endpoint-feedback runtime used by plans and sessions.

__init__(motion_generator, control_profiles=None, grasp_pose_generators=None, *, load_builtins=True, tracking_runtime=None, scene_provider=None)[source]#

Initialize one engine and bind its built-in action implementations.

Parameters:

motion_generator (MotionGenerator) – Engine-owned motion-generation backend.
control_profiles (Optional[Mapping[str, ControlPartCommandProfile]]) – Semantic commands keyed by robot control-part name.
grasp_pose_generators (Optional[Mapping[str, GraspPoseGenerator]]) – Standalone grasp-pose services keyed by the runtime target ID of each grasp endpoint, normally its robot control-part name.
load_builtins (bool) – Whether to instantiate and register every built-in action. Disable this for isolated tests or fully custom engines.
tracking_runtime (TrackingRuntime | None) – Optional exact-version feedback, projector, and metric registries. Built-in joint tracking is installed when omitted.
scene_provider (SceneProvider | None) – Optional default scene-observation source used by initial_context() when the caller does not supply an explicit scene snapshot. The provider is borrowed by reference.

property actions: dict[str, AtomicAction]#: Registered action instances keyed by stable skill identifier.

bind_control_parts(skill_id, endpoints, *, task_state_keys=None)[source]#

Build an advanced direct-core binding from control-part names.

Parameters:

skill_id (str) – Installed skill ID.
endpoints (Mapping[str, Mapping[str, str]]) – Nested slot_id -> endpoint_id -> control_part mapping.
task_state_keys (Optional[Mapping[str, str]]) – Optional explicit stable task-state key for each resource slot. See ActionPlanningServices.bind_control_parts() for inference rules when omitted.

Return type:

ActionBinding

Returns:

Engine-owned generic endpoint binding.

property binding_owner_id: str#: Return the opaque owner identity required by action bindings.

compile(invocations, context=None)[source]#

Compile a static sequence of grounded invocations.

Planning is side-effect free. Expected effects are applied only to the returned hypothetical projected_context so following actions can be checked against the expected state. No simulator or observed task state is mutated.

Parameters:

invocations (Iterable[ActionInvocation]) – Grounded action requests in execution order.
context (PlanningContext | None) – Optional initial planning context captured by the caller.

Return type:

CompiledTrajectory

Returns:

Concatenated timed trajectory, individual plans, and projected state.

Raises:

KeyError – If an invocation references an unregistered skill.
ValueError – If context, plan, or trajectory dimensions are incompatible.

property control_profiles: Mapping[str, ControlPartCommandProfile]#: Semantic command profiles registered for robot control parts.

property device: device#: Return the concrete planning device used by this engine.

property grasp_pose_generators: Mapping[str, GraspPoseGenerator]#: Standalone grasp-pose services installed for endpoint targets.

initial_context(*, task=None, scene=None, timestamp=0.0, control_dt=None)[source]#

Capture the robot state needed to start offline compilation.

Parameters:

task (TaskState | None) – Optional symbolic task state; an empty state is used otherwise.
scene (SceneSnapshot | None) – Optional explicit scene snapshot. It overrides the engine’s configured scene provider; an empty snapshot is used when both are absent.
timestamp (float) – Timestamp assigned to the captured robot observation.
control_dt (float | None) – Explicit command period for action-owned interpolation.

Return type:

Returns:

Planning context containing owned robot tensors.

make_invocation(skill_id, goal, *, control_parts=None, motion_policy=None, tracking_policy=None, recovery_policy=None, skill_options=None, control_overrides=None, invocation_id=None, revision=0)[source]#

Construct a grounded invocation while naming the skill only once.

control_parts uses the advanced direct-core binding path. Profile- based integrations resolve an ActionBinding in the semantic layer and construct ActionInvocation directly.

Parameters:

skill_id (str) – Stable identifier of an installed atomic skill.
goal (TypeVar(GoalT)) – Action-specific typed goal.
control_parts (Optional[Mapping[str, Mapping[str, str]]]) – Direct slot -> endpoint -> control_part mapping.
motion_policy (MotionPolicy | None) – Optional invocation motion policy.
tracking_policy (TrackingPolicy | None) – Optional typed tracking and terminal-acceptance policy.
recovery_policy (RecoveryPolicy | None) – Optional invocation recovery policy.
skill_options (Optional[TypeVar(OptionsT, bound= ActionOptions)]) – Optional action-specific invocation options.
control_overrides (ActionControlOverrides | None) – Optional endpoint-scoped command overrides.
invocation_id (str | None) – Optional correlation identifier.
revision (int) – Monotonic invocation revision.

Return type:

ActionInvocation[TypeVar(GoalT), TypeVar(OptionsT, bound= ActionOptions)]

Returns:

A standard ActionInvocation accepted by plan, compile, and start.

Raises:

ValueError – If control_parts is omitted.
KeyError – If the skill or control part is unknown.
TypeError – If an invocation field or binding input has an invalid type.

property motion_generator: MotionGenerator#: Return the single motion generator owned by this engine.

plan(invocation, context=None)[source]#

Plan one registered invocation through the engine-owned backend.

Parameters:

invocation (ActionInvocation) – Grounded request for a registered skill.
context (PlanningContext | None) – Optional latest planning state; captured when omitted.

Return type:

Returns:

Validated action plan.

Raises:

KeyError – If the invocation references an unregistered skill.

plan_request(request, context=None)[source]#

Plan an already-resolved request without rebuilding its snapshot.

Return type:: ActionPlan

property planning_services: ActionPlanningServices#: Engine-owned resources shared by every bound atomic action.

register(action, *, replace=False)[source]#

Register one action instance using its descriptor.

Parameters:

action (AtomicAction) – Configured action instance.
replace (bool) – Whether to replace an implementation already registered under the same stable skill identifier. Replacement is always explicit so extensions cannot silently shadow built-ins.

Raises:

TypeError – If action is not an AtomicAction.
ValueError – If it belongs to another engine or its skill identifier conflicts with an existing action.

Return type:

None

resolve(invocation)[source]#

Resolve a registered invocation into an engine-owned snapshot.

Return type:: ResolvedActionRequest

property robot: Robot#: Return the robot controlled by this engine.

property skill_catalog_revision: int#

Return the monotonic installed Atomic Skill catalog revision.

Installing or replacing an agent-visible implementation advances the revision even when its public descriptor is equal. External binding and compilation layers can therefore reject stale implementation snapshots.

property skills: Mapping[str, SkillDescriptor]#

Return explicitly declared, agent-visible installed skill metadata.

Process-wide type discovery, engine installation, and semantic exposure are separate boundaries. Only an action installed in this engine whose concrete class explicitly declares a generic binding contract appears here. Direct-core callers may continue to use every entry in actions.

start(invocations, context=None, *, eligible_mask=None)[source]#

Start incremental execution for a grounded invocation sequence.

Parameters:

invocations (Iterable[ActionInvocation]) – Grounded action requests in execution order.
context (PlanningContext | None) – Initial measured state and scene snapshot. The engine captures one when omitted.
eligible_mask (Tensor | None) – Optional per-environment cohort allowed to execute. Ineligible rows remain excluded for the whole session. All rows are eligible when omitted.

Return type:

ExecutionSession

Returns:

Stateful execution session advanced by session.tick(...).

property tracking_runtime: TrackingRuntime#: Typed endpoint-feedback runtime used by plans and sessions.

class embodichain.lab.sim.atomic_actions.ExecutionSession[source]#

Execute grounded invocations incrementally with bounded local recovery.

The session never steps a simulator itself. Each tick() consumes the latest observation and scene snapshot and emits at most one synchronized endpoint-command frame. A declared physical-effect boundary resolves only after the caller supplies a correlated EffectVerificationResult, and non-empty expected symbolic effects are committed for accepted rows only. Higher-level runtimes decide how to produce that result from their configured monitor selection.

Environment eligibility and recovery budgets are tracked per row. The waypoint cursor is batch-synchronized: a recoverable row replans the active cohort from the latest observation and restarts the action trajectory. Calls that mutate the session must be serialized by its owner; the session does not provide thread synchronization.

Methods:

`__init__`(engine, invocations, context, *[, ...])
`deactivate_rows`(env_mask, *, reason)	Permanently remove selected rows from this invocation sequence.
`revise_current`(invocation, *[, context])	Replace and replan the current invocation with a newer revision.
`tick`(context, *[, effect_result, ...])	Advance execution by one observation/command cycle.
`trajectory_segment`(name)	Return named segment metadata for the active action plan.

Attributes:

`active_commands`	Return an owned snapshot of the active action command sequence.
`active_plan`	Return an independently owned snapshot of the active action plan.
`effect_verification_pending`	Whether the current physical effect still requires verification.
`eligible_mask`	Rows still eligible to complete the full invocation sequence.
`held_object_guard_request`	Describe the phase that must be checked before the next command.
`latest_context`	Latest validated context with the session's verified task state.
`pending_effect`	Owned snapshot of the current effect boundary, when present.
`phase_effect_gate_request`	Return the blocking gate at the next trajectory-segment entry.
`plan_attempts`	Return every installed plan in deterministic recovery order.
`status`	Current session status.
`task_state`	Verified symbolic task state accumulated by this session.

__init__(engine, invocations, context, *, eligible_mask=None)[source]#

property active_commands: TimedCommandSequence#

Return an owned snapshot of the active action command sequence.

This inspection surface is intended for diagnostics and visualization. Mutating the returned tensors cannot affect execution state.

property active_plan: ActionPlan#

Return an independently owned snapshot of the active action plan.

This is a read-only diagnostics boundary for runtime metadata, visualization, and tests. Planning and recovery remain session-owned; mutating any tensor in the returned value cannot affect execution.

deactivate_rows(env_mask, *, reason)[source]#

Permanently remove selected rows from this invocation sequence.

Deactivation is sticky across action barriers and recovery replans. The next emitted command frame marks those rows inactive so the command sink can apply target-specific safe hold behavior.

Parameters:

env_mask (Tensor) – Rows requested for deactivation.
reason (str) – Human-readable event message.

Return type:

Tensor

Returns:

Owned mask of rows that changed from eligible to inactive.

Raises:

RuntimeError – If the session is already terminal.
ValueError – If reason is empty or the mask shape is invalid.

property effect_verification_pending: bool#: Whether the current physical effect still requires verification.

property eligible_mask: Tensor#

Rows still eligible to complete the full invocation sequence.

This is deliberately not named success_mask: while the session is running, eligibility does not imply that execution or semantic effects have succeeded.

property held_object_guard_request: HeldObjectGuardRequest | None#

Describe the phase that must be checked before the next command.

The request remains available while terminal acceptance is settling, using the final waypoint and segment identity. Once terminal physical effect verification begins, that verifier owns the boundary and this property returns None.

Returns:: Owned phase-aware guard request, or None when no command-phase guard is active.

property latest_context: PlanningContext#: Latest validated context with the session’s verified task state.

property pending_effect: EffectVerificationRequest | None#: Owned snapshot of the current effect boundary, when present.

property phase_effect_gate_request: PhaseEffectGateRequest | None#

Return the blocking gate at the next trajectory-segment entry.

Returns:: Owned request snapshot, or None when the next command is not blocked by a physical-effect gate.

property plan_attempts: tuple[ExecutionPlanAttempt, ...]#

Return every installed plan in deterministic recovery order.

The initial plan has generation zero. Each invocation revision, recovery replan, or whole-action retry appends a new generation instead of replacing earlier scene/collision evidence.

revise_current(invocation, *, context=None)[source]#

Replace and replan the current invocation with a newer revision.

The replacement is resolved into a new immutable request snapshot from context or the session’s latest observation. Retry and replan budgets restart for the new revision, while verified task state, the current batch barrier, and per-environment eligibility are preserved. Ordinary recovery replans continue to reuse this snapshot until another explicit revision. Once the action owns runtime destinations, the replacement must preserve their exact address fingerprints; changing controllers or safe-hold footprints requires a new invocation.

Parameters:

invocation (ActionInvocation) – Grounded replacement for the currently active skill. Its revision must be strictly greater than the active one, and its skill_id and invocation_id must identify the same logical call.
context (PlanningContext | None) – Optional fresh observation used to ground the replacement. A manually ticked caller may omit it to reuse latest_context. Runner-driven code stages revisions on ExecutionRunner, which supplies a due-time observation.

Raises:

TypeError – If invocation is not an ActionInvocation.
RuntimeError – If the session is no longer running or a physical effect is awaiting verification.
ValueError – If the replacement identifies another invocation or does not advance the revision, or if its plan changes the active runtime target addresses.

Return type:

None

property status: ExecutionStatus#: Current session status.

property task_state: TaskState#: Verified symbolic task state accumulated by this session.

tick(context, *, effect_result=None, phase_effect_gate_result=None, held_object_guard_result=None)[source]#

Advance execution by one observation/command cycle.

Parameters:

context (PlanningContext) – Latest measured robot and versioned scene state. Its task state is replaced by the session’s verified task state.
effect_result (EffectVerificationResult | None) – Optional correlated semantic-effect result for an action waiting at its terminal waypoint.
phase_effect_gate_result (PhaseEffectGateResult | None) – Optional correlated physical-effect decision for a blocked trajectory-segment entry.
held_object_guard_result (HeldObjectGuardResult | None) – Optional correlated in-flight held-object loss result for the current waypoint phase. None means the verifier found no applicable guard for this phase or no result was supplied.

Return type:

ExecutionTick

Returns:

Status, optional command, events, and current verified task state.

trajectory_segment(name)[source]#

Return named segment metadata for the active action plan.

Segment ranges are action-local and may change after a replan when a backend preserves its own sample count.

Return type:: TrajectorySegment

class embodichain.lab.sim.atomic_actions.ExecutionRunner[source]#

Connect an execution session to observation, controller, and time ports.

step() is non-blocking. It observes and advances the session only when the next command is due according to RuntimeCommandFrame.hold_duration. run_until_blocked() supplies the blocking loop for tutorials and simple applications. Controller rejection, timeout, observation failure, and session exceptions all trigger a best-effort cancel-then-hold sequence. Runner methods are designed for serialized event-loop use and are not thread-safe.

Parameters:

session (ExecutionSession) – Stateful atomic-action execution session.
observation_provider (ObservationProvider) – Source of fresh robot and scene observations.
command_sink (CommandSink) – Controller or simulation command boundary.
clock (ExecutionClock | None) – Optional scheduler clock. Defaults to monotonic wall time.
cfg (ExecutionRunnerCfg | None) – Optional acknowledgement, scheduling, and completion policy.

Methods:

`__init__`(session, observation_provider, ...)
`cancel`([reason])	Cancel controller work and hold the latest observed position.
`deactivate_rows`(env_mask, *, reason)	Permanently deactivate environment rows owned by this runner.
`revise_current`(invocation)	Stage a newer revision for the next scheduled observation boundary.
`run_until_blocked`(*[, effect_verifier, ...])	Run with clock-driven waiting until terminal or effect verification blocks.
`step`(*[, effect_result, effect_verifier, ...])	Perform one due observation/session/controller update without sleeping.

Attributes:

`command_count`	Number of active commands accepted by the sink.
`effect_verification_pending`	Whether execution is waiting for an external semantic-effect result.
`session`	Execution session advanced by this runner.
`status`	Current runner lifecycle status.

__init__(session, observation_provider, command_sink, *, clock=None, cfg=None)[source]#

cancel(reason='Execution cancelled by caller.')[source]#

Cancel controller work and hold the latest observed position.

Parameters:: reason (str) – Human-readable cancellation reason.
Return type:: RunnerStep
Returns:: Terminal runner step. The status is cancelled only when both cancel and hold are acknowledged; otherwise it is failed.

property command_count: int#: Number of active commands accepted by the sink.

deactivate_rows(env_mask, *, reason)[source]#

Permanently deactivate environment rows owned by this runner.

The runner refreshes its cached effect boundary so a verifier cannot submit a result correlated with a request that deactivation replaced. In-flight controller work is neutralized for those rows by the next due command frame according to the CommandSink contract.

Parameters:

env_mask (Tensor) – Rows requested for deactivation.
reason (str) – Human-readable event message.

Return type:

Tensor

Returns:

Owned mask of rows that changed from eligible to inactive.

Raises:

RuntimeError – If the runner is already terminal.
TypeError – If env_mask is not a tensor.
ValueError – If the mask or reason is invalid.

property effect_verification_pending: bool#: Whether execution is waiting for an external semantic-effect result.

revise_current(invocation)[source]#

Stage a newer revision for the next scheduled observation boundary.

Staging preserves the active frame deadline. When that deadline is due, step() observes fresh state, atomically plans and installs the replacement, and dispatches its first command. The submitted invocation is resolved into an owned snapshot immediately, so later caller mutation cannot alter the staged revision.

Parameters:

invocation (ActionInvocation) – Strictly newer revision of the active logical call.

Raises:

TypeError – If invocation is not an ActionInvocation.
RuntimeError – If this runner or its session is no longer running, or if a physical effect is awaiting verification.
ValueError – If session-level revision invariants are violated.

Return type:

None

run_until_blocked(*, effect_verifier=None, phase_effect_gate_verifier=None, held_object_guard_verifier=None, on_step=None, max_steps=100000)[source]#

Run with clock-driven waiting until terminal or effect verification blocks.

Parameters:

effect_verifier (Callable[[PlanningContext, EffectVerificationRequest], EffectVerificationResult] | None) – Optional synchronous callback used on fresh due-cycle observations while effect verification is pending. Without one, the method returns the running boundary so the caller can verify externally.
phase_effect_gate_verifier (Callable[[PlanningContext, PhaseEffectGateRequest], PhaseEffectGateResult] | None) – Optional synchronous callback used on fresh observations while a trajectory-segment entry is gated.
held_object_guard_verifier (Callable[[PlanningContext, HeldObjectGuardRequest], HeldObjectGuardResult | None] | None) – Optional synchronous phase-aware held-object verifier used before every due command cycle.
on_step (Callable[[RunnerStep], None] | None) – Optional callback for tracing or tutorial visualization.
max_steps (int) – Hard bound on loop iterations.

Return type:

Returns:

Terminal step, or a running step blocked on external verification.

property session: ExecutionSession#

Execution session advanced by this runner.

Call revise_current() or deactivate_rows() on the runner, rather than mutating the session directly, while this runner owns scheduling.

property status: RunnerStatus#: Current runner lifecycle status.

step(*, effect_result=None, effect_verifier=None, phase_effect_gate_result=None, phase_effect_gate_verifier=None, held_object_guard_verifier=None)[source]#

Perform one due observation/session/controller update without sleeping.

Parameters:

effect_result (EffectVerificationResult | None) – Optional correlated effect result. If this call occurs before the next cycle is due, it is not consumed and must be supplied again on a later call.
effect_verifier (Callable[[PlanningContext, EffectVerificationRequest], EffectVerificationResult] | None) – Optional synchronous verifier for the current pending request. It runs after a fresh due-cycle observation and before the session consumes the result. It is not called after the request deadline. Mutually exclusive with effect_result.
phase_effect_gate_result (PhaseEffectGateResult | None) – Optional externally produced result for the current blocking trajectory-segment entry gate.
phase_effect_gate_verifier (Callable[[PlanningContext, PhaseEffectGateRequest], PhaseEffectGateResult] | None) – Optional synchronous verifier for the current gate. It runs on a fresh due-cycle observation and is mutually exclusive with phase_effect_gate_result.
held_object_guard_verifier (Callable[[PlanningContext, HeldObjectGuardRequest], HeldObjectGuardResult | None] | None) – Optional synchronous phase-aware verifier. It receives a fresh observation and the current command-phase request before ExecutionSession.tick() and command dispatch. Returning None means the current phase has no applicable held-object guard.

Return type:

Returns:

Runner status, optional session tick, controller acknowledgements, and time remaining before another update is due.

class embodichain.lab.sim.atomic_actions.ExecutionRunnerCfg[source]#

Transport and scheduling policy for an ExecutionRunner.

Attributes:

`command_timeout`	Maximum time allowed for a command acknowledgement.
`hold_during_effect_verification`	Whether to hold observed state while terminal effects are pending.
`hold_on_completion`	Whether to issue a final hold after the session completes.
`minimum_cycle_time`	Minimum delay between feedback cycles, including passive hold cycles.
`safe_stop_timeout`	Maximum time allowed for each cancel or hold acknowledgement.

Methods:

validate([prefix])

Check the validity of configclass object.

command_timeout: float#: Maximum time allowed for a command acknowledgement.

hold_during_effect_verification: bool#

Whether to hold observed state while terminal effects are pending.

Disable this only for persistent transports whose last accepted command remains active without refresh, such as a position-controlled gripper that must retain contact preload. Failure and cancellation still perform the normal cancel-then-observed-hold safe stop.

hold_on_completion: bool#: Whether to issue a final hold after the session completes.

minimum_cycle_time: float#: Minimum delay between feedback cycles, including passive hold cycles.

safe_stop_timeout: float#: Maximum time allowed for each cancel or hold acknowledgement.

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.lab.sim.atomic_actions.ObservationProvider[source]#

Source of fresh planning contexts for feedback-driven execution.

Methods:

`__init__`(args, *kwargs)
`observe`(task_state)	Capture the latest robot and scene state.

__init__(*args, **kwargs)#

observe(task_state)[source]#

Capture the latest robot and scene state.

Parameters:: task_state (TaskState) – Runner-owned, externally verified symbolic task state.
Return type:: PlanningContext
Returns:: Fresh context with stable, ordered environment IDs.

class embodichain.lab.sim.atomic_actions.CommandSink[source]#

Controller boundary used by ExecutionRunner.

Methods:

`__init__`(args, *kwargs)
`cancel`(targets, *, timeout)	Cancel any controller-side command that has not completed.
`hold`(targets, context, *, timeout)	Apply transport-specific safe state to the supplied targets.
`send`(command, *, timeout)	Submit one synchronized endpoint-command frame.

__init__(*args, **kwargs)#

cancel(targets, *, timeout)[source]#

Cancel any controller-side command that has not completed.

Parameters:

targets (tuple[RuntimeEndpointTarget, ...]) – Runtime targets whose queued work must be cancelled.
timeout (float) – Maximum acknowledgement latency in seconds.

Return type:

Returns:

Transport or controller acknowledgement.

hold(targets, context, *, timeout)[source]#

Apply transport-specific safe state to the supplied targets.

Parameters:

targets (tuple[RuntimeEndpointTarget, ...]) – Runtime targets that may retain controller state.
context (PlanningContext) – Latest observation used by position-hold transports.
timeout (float) – Maximum acknowledgement latency in seconds.

Return type:

Returns:

Transport or controller acknowledgement.

send(command, *, timeout)[source]#

Submit one synchronized endpoint-command frame.

Parameters:

command (RuntimeCommandFrame) – Transport-neutral command frame with an active-row mask. The sink must actively neutralize inactive rows for every addressed target; omission is not a safe state for persistent controllers.
timeout (float) – Maximum acknowledgement latency in seconds.

Return type:

Returns:

Transport or controller acknowledgement.

class embodichain.lab.sim.atomic_actions.EndpointCommandTransport[source]#

Backend that owns one kind of runtime endpoint command.

Implementations own live simulator entities, device clients, or controller handles. Runtime command values retain only immutable addressing and payload data, so they remain independent of those process-owned resources.

Methods:

`__init__`(args, *kwargs)
`cancel`(targets, *, timeout)	Cancel outstanding commands for transport-local targets.
`hold`(targets, context, *, timeout)	Hold transport-local targets at their observed state.
`send`(frame, *, timeout)	Submit one transport-local command frame.

Attributes:

`payload_type`	Return the runtime payload type accepted by `send()`.
`transport_id`	Return the exact identifier used to register this transport.

__init__(*args, **kwargs)#

cancel(targets, *, timeout)[source]#

Cancel outstanding commands for transport-local targets.

Return type:: CommandAcknowledgement

hold(targets, context, *, timeout)[source]#

Hold transport-local targets at their observed state.

Return type:: CommandAcknowledgement

property payload_type: type[RuntimeCommandPayload]#: Return the runtime payload type accepted by send().

send(frame, *, timeout)[source]#

Submit one transport-local command frame.

Implementations must actively neutralize every inactive environment row for every addressed target. Silently skipping an inactive row is unsafe for persistent controllers such as base-velocity transports.

Return type:: CommandAcknowledgement

property transport_id: str#: Return the exact identifier used to register this transport.

class embodichain.lab.sim.atomic_actions.EndpointCommandRouter[source]#

Route generic endpoint operations to exact registered transports.

The router implements CommandSink structurally while avoiding a module-load dependency on runner. Acknowledgement types are imported only when an operation is executed, which keeps the transport boundary safe to import while the runner imports this module.

Parameters:

transports (Mapping[str, EndpointCommandTransport] | Iterable[EndpointCommandTransport]) – Either an exact transport_id -> transport mapping or an iterable of transports from which that mapping is built. Mapping keys must exactly equal each value’s declared transport_id.

Raises:

TypeError – If a registration does not implement the transport contract.
ValueError – If an identifier is invalid, a mapping key is not exact, or the same transport identifier is registered more than once.

Methods:

`__init__`(transports)
`cancel`(targets, *, timeout)	Route cancellation by target transport.
`hold`(targets, context, *, timeout)	Route an observed-state hold request by target transport.
`send`(frame, *, timeout)	Route one synchronized command frame by transport identifier.

Attributes:

transports

Return the immutable exact transport registry.

__init__(transports)[source]#

cancel(targets, *, timeout)[source]#

Route cancellation by target transport.

Parameters:

targets (tuple[RuntimeEndpointTarget, ...]) – Runtime destinations whose outstanding commands are cancelled.
timeout (float) – Maximum acknowledgement latency for each transport.

Return type:

Returns:

Aggregated acknowledgement. It is accepted only when every addressed transport accepts cancellation.

hold(targets, context, *, timeout)[source]#

Route an observed-state hold request by target transport.

Parameters:

targets (tuple[RuntimeEndpointTarget, ...]) – Runtime destinations to hold.
context (PlanningContext) – Fresh observation used by each transport to form its hold.
timeout (float) – Maximum acknowledgement latency for each transport.

Return type:

Returns:

Aggregated acknowledgement. It is accepted only when every addressed transport accepts its hold.

send(frame, *, timeout)[source]#

Route one synchronized command frame by transport identifier.

Dispatch is preflighted before any transport is called. An unknown transport or incompatible payload therefore rejects the whole frame without creating a partially dispatched operation.

Parameters:

frame (RuntimeCommandFrame) – Generic runtime command frame to split by transport.
timeout (float) – Maximum acknowledgement latency for each transport.

Return type:

Returns:

Aggregated acknowledgement. It is accepted only when every addressed transport accepts its local frame.

property transports: Mapping[str, EndpointCommandTransport]#: Return the immutable exact transport registry.

class embodichain.lab.sim.atomic_actions.ExecutionClock[source]#

Clock abstraction used for deterministic and simulation scheduling.

Methods:

`__init__`(args, *kwargs)
`now`()	Return a monotonic timestamp in seconds.
`sleep`(duration)	Wait or advance the execution backend by `duration` seconds.

__init__(*args, **kwargs)#

now()[source]#

Return a monotonic timestamp in seconds.

Return type:: float
Returns:: Monotonic timestamp in seconds.

sleep(duration)[source]#

Wait or advance the execution backend by duration seconds.

Parameters:: duration (float) – Non-negative duration in seconds.
Return type:: None

class embodichain.lab.sim.atomic_actions.MonotonicExecutionClock[source]#

Wall-clock implementation backed by time.

Methods:

`now`()	Return the current monotonic wall-clock time.
`sleep`(duration)	Sleep for a non-negative wall-clock duration.

now()[source]#

Return the current monotonic wall-clock time.

Return type:: float
Returns:: Monotonic wall-clock timestamp in seconds.

sleep(duration)[source]#

Sleep for a non-negative wall-clock duration.

Parameters:: duration (float) – Requested duration in seconds.
Return type:: None

embodichain.lab.sim.atomic_actions.create_simulation_atomic_action_engine(motion_generator, scene_entities, control_profiles=None, grasp_pose_generators=None, *, load_builtins=True, tracking_runtime=None)[source]#

Create an engine whose initial context observes selected rigid objects.

This is the direct-simulation convenience path for offline planning. Entity IDs are derived from each rigid object’s stable uid; only explicitly supplied objects are observed. Advanced integrations that need aliases, articulation/link state, collision roles, or an external perception source should construct AtomicActionEngine with their own SceneProvider instead.

Parameters:

motion_generator (MotionGenerator) – Motion-generation backend owned by the engine.
scene_entities (Sequence[RigidObject]) – Non-empty sequence of simulation rigid objects to expose in automatically captured initial scene snapshots.
control_profiles (Mapping[str, ControlPartCommandProfile] | None) – Semantic commands keyed by robot control-part name.
grasp_pose_generators (Mapping[str, GraspPoseGenerator] | None) – Grasp-pose services keyed by grasp endpoint target.
load_builtins (bool) – Whether to install all built-in atomic actions.
tracking_runtime (TrackingRuntime | None) – Optional typed tracking runtime shared by action plans.

Return type:

AtomicActionEngine

Returns:

Engine configured with a rigid-object scene provider.

Raises:

TypeError – If scene_entities is not a sequence.
ValueError – If an entity lacks a stable UID or UIDs are duplicated.

class embodichain.lab.sim.atomic_actions.SimulationExecutionAdapter[source]#

Adapt a simulation robot to observation, command, and clock protocols.

The adapter writes joint targets synchronously. Time advances only through sleep(), which converts the requested runner interval to an integral number of physics updates. This makes ExecutionRunner.run_until_blocked() deterministic and avoids wall-clock sleeps in headless simulation.

Parameters:

simulation (SimulationManager) – Simulation manager advanced by the execution clock.
robot (Robot) – Robot observed and commanded by the adapter.
physics_dt (float | None) – Optional physics period. Defaults to the simulation config.
control_dt (float | None) – Optional command period exposed to action interpolation. Defaults to physics_dt because that is the adapter’s minimum executable command cadence.
env_ids (Tensor | None) – Optional stable correlation IDs matching every robot row. They are not used as simulator indices; row order maps to robot instances.
scene_provider (SceneProvider | None) – Optional provider for versioned scene observations.
scene_supplier (Callable[[float], SceneSnapshot] | None) – Optional callback for versioned scene observations. It is mutually exclusive with scene_provider.
initial_time (float) – Initial elapsed simulation time in seconds.

Methods:

`__init__`(simulation, robot, *[, physics_dt, ...])
`cancel`(targets, *, timeout)	Acknowledge cancellation of synchronous simulation target writes.
`hold`(targets, context, *, timeout)	Set every represented joint endpoint to an observed-position hold.
`now`()	Return elapsed simulation time in seconds.
`observe`(task_state)	Capture full-robot state and the latest supplied scene snapshot.
`send`(command, *, timeout)	Write joint endpoint targets and neutralize inactive rows.
`sleep`(duration)	Advance physics by at least the requested duration.

Classes:

payload_type

alias of JointPositionPayload

__init__(simulation, robot, *, physics_dt=None, control_dt=None, env_ids=None, scene_provider=None, scene_supplier=None, initial_time=0.0)[source]#

cancel(targets, *, timeout)[source]#

Acknowledge cancellation of synchronous simulation target writes.

Parameters:

targets (tuple[RuntimeEndpointTarget, ...]) – Joint-position destinations whose queued work is cancelled.
timeout (float) – Positive acknowledgement deadline.

Return type:

Returns:

Accepted acknowledgement. The following hold call installs the actual safe target.

hold(targets, context, *, timeout)[source]#

Set every represented joint endpoint to an observed-position hold.

Parameters:

targets (tuple[RuntimeEndpointTarget, ...]) – Joint-position destinations to place in a safe hold.
context (PlanningContext) – Latest observed positions and stable environment IDs.
timeout (float) – Positive acknowledgement deadline.

Return type:

Returns:

Accepted acknowledgement or a rejected diagnostic.

now()[source]#

Return elapsed simulation time in seconds.

Return type:: float
Returns:: Elapsed simulation time in seconds.

observe(task_state)[source]#

Capture full-robot state and the latest supplied scene snapshot.

Parameters:: task_state (TaskState) – Verified symbolic state owned by the execution session.
Return type:: PlanningContext
Returns:: Planning context timestamped with elapsed simulation time.

payload_type#: alias of JointPositionPayload

send(command, *, timeout)[source]#

Write joint endpoint targets and neutralize inactive rows.

Parameters:

command (RuntimeCommandFrame) – Joint-position endpoint frame. Inactive rows are replaced with observed positions by this transport.
timeout (float) – Positive acknowledgement deadline. Simulation writes are synchronous, so this is validated but otherwise unused.

Return type: