embodichain.lab.task_program.integrations

embodichain.lab.task_program.integrations#

Explicit provider, environment, and simulation Task Program integrations.

`AntipodalGraspAffordanceBinding`	Build one antipodal grasp affordance from a selected rigid-object mesh.
`ConfiguredHandOverPoseProvider`	Resolve a hand-over delivery target from immutable configuration.
`ContainerAffordanceBinding`	Declare one exact object target frame inside a container parent.
`ControlPartCommandPreset`	Named one-dimensional joint commands for one exact control part.
`ControlPartEvidenceProviderDeclaration`	Provider-free identity of one control-part evidence factory.
`ControlPartEvidenceProviderFactory`	Registration-owned factory for one live control-part evidence provider.
`ControlPartEndpointBinding`	Profile endpoint backed by one explicit robot control part.
`ControlPartResourceBinding`	Joint-backed robot resource containing control-part endpoints.
`CuroboParallelCommandSafetyValidator`	Validate the exact synchronized joint segment before transport dispatch.
`CuroboParallelSafetyValidatorFactory`	Create exact-sample collision gates for one aggregate control part.
`EndpointAdapterDeclaration`	Provider-free declaration of one exact endpoint adapter.
`TaskProgramAdapterFactory`	Bind one initialized environment to its Task Program adapter.
`TaskProgramEnvironmentAdapter`	Compile and run Task Programs through explicit environment factories.
`TaskProgramEnvironmentFactory`	Environment-owned factories for one explicit semantic integration.
`TaskProgramIntegrationCatalog`	Provider-free integration directory owned by one task registration.
`TaskProgramRuntimeAssembly`	Auditable result of one fresh environment runtime assembly.
`IntegrationFingerprintMismatch`	Raised when a live integration no longer matches its registration.
`ParallelCommandSafetyValidatorFactory`	Registration-owned factory for one authoritative live safety gate.
`ParallelSafetyDeclaration`	Provider-free identity and transport coverage of one safety factory.
`PlanningObservationPort`	Combined observation and full-qpos port required by the Gym runtime.
`RegisteredSemanticLowererDeclaration`	Provider-free identity of one registered semantic lowerer factory.
`RegisteredSemanticLowererFactory`	Registration-owned factory for one fresh live semantic lowerer.
`RigidizedArticulationAntipodalGraspBinding`	Build a root-frame grasp mesh from one rigidized articulation link.
`RuntimeTransportDeclaration`	Provider-free declaration of one ordered runtime transport encoder.
`SimulationArticulationBinding`	Explicit binding for one simulation articulation.
`SimulationArticulationLinkBinding`	Explicit canonical link backed by one native articulation link.
`SimulationRigidizedArticulationObjectBinding`	Bind one locked, floating articulation as a semantic object.
`SimulationTaskProgramAdapterFactory`	Create one standard simulation adapter after an environment is initialized.
`SimulationTaskProgramFactory`	Build every live Task Program component from one task registration.
`SimulationTaskProgramRegistration`	Exact immutable task-owned simulation integration registration.
`SimulationRigidObjectBinding`	Explicit binding for one simulation rigid object.
`SimulationRobotSkillProfileBinding`	Build a profile from typed resources with strict native validation.
`SimulationSceneBinding`	Build one authoritative registry from explicit simulation bindings.
`SimulationSegmentPolicyPort`	Execute built-in segment policies against explicitly bound simulation data.
`StandardExtensionDeclarations`	Cross-checked provider-free declarations for the standard factory.
`SupportSurfaceAffordanceBinding`	Declare one exact object target frame on a support parent.
`VersionedKey`	Built-in immutable sequence.
`create_simulation_task_program_adapter`(...)	Create a complete production adapter from one standard Gym environment.
`default_simulation_settle_presets`()	Return independently owned built-in post-policy presets.

Classes:

`CombinedTaskProgramExpansionFactory`	Apply one deterministic combined recipe to a Task Program call.
`TaskProgramExpansionRecord`	Owned logical candidates generated for one Task Program planning call.
`TaskProgramSourceAdapter`	Expose a Task Program's grounded Atomic plan through SourceAdapter.

class embodichain.lab.task_program.integrations.CombinedTaskProgramExpansionFactory[source]#

Apply one deterministic combined recipe to a Task Program call.

Selected calls receive their Affordance branch before Atomic planning. All skills share one row-local transform; explicit phase declarations may narrow the sample permissions for specialized deployments.

Methods:

`__init__`(profile, *, candidate_index[, ...])
`create_plan_transform`(request, *, engine)	Return the transform for calls selected by the episode schedule.
`prepare_planning_context`(request, context, ...)	Attach the selected call's Affordance branch before Atomic planning.

Attributes:

records

Return expansion records in planning-call order.

__init__(profile, *, candidate_index, recipe_indices=None, program_id, integration_id, robot_profile_id)[source]#

create_plan_transform(request, *, engine)[source]#

Return the transform for calls selected by the episode schedule.

Return type:: Optional[Callable[[ResolvedActionRequest, PlanningContext, ActionPlan], ActionPlan]]

prepare_planning_context(request, context, *, engine)[source]#

Attach the selected call’s Affordance branch before Atomic planning.

Return type:: PlanningContext

property records: tuple[TaskProgramExpansionRecord, ...]#: Return expansion records in planning-call order.

class embodichain.lab.task_program.integrations.TaskProgramExpansionRecord[source]#

Owned logical candidates generated for one Task Program planning call.

Methods:

`__init__`(workflow_id, workflow_call_index, ...)
`snapshot`()	Return a record with independently owned trajectory tensors.
`to_metadata`()	Return JSON-compatible candidate identity and factor provenance.

Attributes:

`affordance_requested`
`candidate_index`
`candidates`
`cycle_index`
`env_id`
`fallback_count`
`recipe_index`
`selected_candidate_id`
`templates`
`trajectory_requested`
`trajectory_selected`
`workflow_call_index`
`workflow_id`

__init__(workflow_id, workflow_call_index, candidate_index, selected_candidate_id, candidates, templates, recipe_index=None, cycle_index=None, affordance_requested=None, trajectory_requested=None, trajectory_selected=None, fallback_count=0, env_id=None)#

affordance_requested: int | None#

candidate_index: int#

candidates: tuple[CandidateSpec, ...]#

cycle_index: int | None#

env_id: int | None#

fallback_count: int#

recipe_index: int | None#

selected_candidate_id: str#

snapshot()[source]#

Return a record with independently owned trajectory tensors.

Return type:: TaskProgramExpansionRecord

templates: tuple[TrajectoryTemplate, ...]#

to_metadata()[source]#

Return JSON-compatible candidate identity and factor provenance.

Return type:: dict[str, object]

trajectory_requested: int | None#

trajectory_selected: int | None#

workflow_call_index: int#

workflow_id: str#

class embodichain.lab.task_program.integrations.TaskProgramSourceAdapter[source]#

Expose a Task Program’s grounded Atomic plan through SourceAdapter.

Methods:

`__init__`(atomic_adapter)
`export_template`(source, *, context)	Delegate one grounded Atomic plan to the common template adapter.

Attributes:

`atomic_adapter`
`kind`

__init__(atomic_adapter)#

atomic_adapter: ActionPlanTemplateAdapter#

export_template(source, *, context)[source]#

Delegate one grounded Atomic plan to the common template adapter.

Parameters:

source (ActionPlan) – Grounded Atomic plan produced for one Task Program call.
context (SourceContext) – Coordinator-owned source and scene identity.

Return type:

TrajectoryTemplate

Returns:

Complete phase-annotated trajectory template.

kind: ClassVar[str] = 'task_program'#

Environment assembly#

The provider-independent language and compiler live in embodichain.lab.task_program. This package owns explicit environment, registration, and simulation assembly. Gym stepping remains in embodichain.lab.gym.envs.task_program.

class embodichain.lab.task_program.integrations.TaskProgramEnvironmentAdapter[source]#

Compile and run Task Programs through explicit environment factories.

Parameters:

factory (TaskProgramEnvironmentFactory) – Environment-owned live-provider and engine factory.
step_dt (float) – Authoritative Gym control cadence in seconds.
joint_command_mode (Literal['position', 'position_velocity']) – Expert joint targets emitted at the Gym boundary.
integration_catalog (TaskProgramIntegrationCatalog | None) – Optional immutable task-registration catalog used for provider-free compilation.
registration (SimulationTaskProgramRegistration | None) – Optional exact standard task registration. When present, every compiler/runtime extension comes exclusively from it.
call_catalog (SemanticCallCatalog | None) – Optional immutable semantic call catalog. The built-in catalog is used when omitted.
endpoint_adapters (Mapping[type[ResourceEndpoint], ResourceEndpointAdapter] | None) – Optional custom robot endpoint adapters.
registered_lowerers (Iterable[RegisteredSemanticLowerer]) – Explicit lowerers for registered semantic calls.
relation_grounders (Iterable[RelationTargetGrounder]) – Explicit relation-target grounding providers.
handover_pose_providers (Iterable[HandOverPoseProvider]) – Explicit embodiment hand-over providers.
effect_monitor_registry (EffectMonitorRegistry | None) – Optional exact-version monitor registry.
runtime_transports (Iterable[RuntimeTransportActionEncoder]) – Additional runtime-command-to-Gym encoders.
runner_cfg (ExecutionRunnerCfg | None) – Optional execution-runner policy.
post_policy_port (SegmentPostPolicyPort | None) – Optional environment post-policy executor.
validator_port (SegmentValidatorPort | None) – Optional environment segment validator.
parallel_safety_validator (ParallelCommandSafetyValidator | None) – Optional authoritative parallel safety gate.

A call to compile() snapshots only scene identities. A call to assemble_runtime() creates a fresh live runtime, which makes reset and episode ownership explicit and avoids retaining providers in compiled data.

Methods:

`__init__`(factory, *, step_dt[, ...])
`assemble_runtime`(integration)	Create a fresh fully connected semantic runtime.
`compile`(program)	Compile one program after exact integration-selection validation.
`create_bridge`(program, *[, ...])	Create a fresh Gym bridge for one provider-free compiled program.

Attributes:

`joint_command_mode`	Return the environment-owned expert joint command mode.
`robot_profile_id`	Return the exact robot profile ID accepted by this adapter.
`scene_registry_id`	Return the exact scene integration ID accepted by this adapter.
`step_dt`	Return the authoritative environment-step cadence.

__init__(factory, *, step_dt, joint_command_mode='position', integration_catalog=None, registration=None, call_catalog=None, endpoint_adapters=None, registered_lowerers=(), relation_grounders=(), handover_pose_providers=(), effect_monitor_registry=None, runtime_transports=(), runner_cfg=None, post_policy_port=None, validator_port=None, parallel_safety_validator=None)[source]#

assemble_runtime(integration)[source]#

Create a fresh fully connected semantic runtime.

Parameters:: integration (TaskProgramIntegrationCfg) – Exact scene, profile, and runtime-preset selection.
Return type:: TaskProgramRuntimeAssembly
Returns:: Owned assembly containing every validated runtime boundary.

compile(program)[source]#

Compile one program after exact integration-selection validation.

Parameters:: program (TaskProgramCfg) – Strict declarative program configuration.
Return type:: CompiledTaskProgram
Returns:: Provider-free lazily expanded compiled program.

create_bridge(program, *, expansion_profile=None, candidate_index=0, recipe_indices=None)[source]#

Create a fresh Gym bridge for one provider-free compiled program.

Parameters:: program (CompiledTaskProgram) – Program compiled for this adapter’s exact integration IDs.
Return type:: TaskProgramDemoBridge
Returns:: Lazy bridge sharing one newly assembled runtime, clock, and sink.

property joint_command_mode: Literal['position', 'position_velocity']#: Return the environment-owned expert joint command mode.

property robot_profile_id: str#

Return the exact robot profile ID accepted by this adapter.

Returns:: Stable robot-profile identifier.

property scene_registry_id: str#

Return the exact scene integration ID accepted by this adapter.

Returns:: Stable scene-registry identifier.

property step_dt: float#

Return the authoritative environment-step cadence.

Returns:: Positive control step duration in seconds.

class embodichain.lab.task_program.integrations.TaskProgramAdapterFactory[source]#

Bind one initialized environment to its Task Program adapter.

Methods:

`__init__`(args, *kwargs)
`create_adapter`(environment)	Create the adapter after the environment has built its live scene.

__init__(*args, **kwargs)#

create_adapter(environment)[source]#

Create the adapter after the environment has built its live scene.

Return type:: TaskProgramEnvironmentAdapter

class embodichain.lab.task_program.integrations.TaskProgramEnvironmentFactory[source]#

Environment-owned factories for one explicit semantic integration.

Implementations normally live in reusable robot/task integration modules, not in individual task motion planners. Every method is passed the exact objects selected earlier in the assembly so a factory cannot silently bind a different scene, robot profile, or engine.

Methods:

`__init__`(args, *kwargs)
`__new__`(**kwargs)

__init__(*args, **kwargs)#

__new__(**kwargs)#

class embodichain.lab.task_program.integrations.TaskProgramRuntimeAssembly[source]#

Auditable result of one fresh environment runtime assembly.

integration#: Owned integration-selection snapshot.

scene_registry#: Authoritative live scene registry.

robot_profile#: Declarative robot resource profile.

manifest#: Static scene/profile/call integration manifest.

engine#: Bound atomic action engine.

compiler#: Bound Semantic Call compiler.

observation_provider#: Shared planning and full-qpos provider.

evidence_collector#: Exact-version semantic evidence collector.

clock#: Shared environment-step clock.

command_encoder#: Runtime-frame to Gym-action encoder.

command_sink#: Buffered Gym command sink.

runner_cfg#: Runner policy selected by the integration runtime preset.

parallel_safety_validator#: Optional fresh registration-owned safety gate.

runtime#: Nonblocking Semantic Call runtime.

Methods:

__init__(integration, scene_registry, ...[, ...])

__init__(integration, scene_registry, robot_profile, manifest, engine, compiler, observation_provider, evidence_collector, clock, command_encoder, command_sink, runner_cfg, parallel_safety_validator, runtime, plan_transform_factory=None, expansion_trace_provider=None)#

class embodichain.lab.task_program.integrations.PlanningObservationPort[source]#

Combined observation and full-qpos port required by the Gym runtime.

Methods:

`__init__`(args, *kwargs)
`__new__`(**kwargs)

__init__(*args, **kwargs)#

__new__(**kwargs)#

Registration and extensions#

class embodichain.lab.task_program.integrations.TaskProgramIntegrationCatalog[source]#

Provider-free integration directory owned by one task registration.

Methods:

`__init__`(scene_registry_id, ...)
`preflight`(program)	Compile and statically link every expanded semantic call.
`validate_bound_endpoint_extensions`(bound_profile)	Match every live resolved endpoint to its fingerprinted declaration.
`validate_engine`(engine)	Require the live engine to expose every statically selected skill.
`validate_integration`(integration, *, path)	Validate exact scene, profile, and runtime-preset selection.
`validate_post_policy`(policy, *, path)	Validate one registered post-policy kind and named preset.
`validate_scene_reference`(reference, *, role, ...)	Validate one typed scene reference against its declared role.
`validate_semantic_call`(call, *, path)	Validate that the semantic call is present in the integration catalog.
`validate_validator`(validator, *, path)	Validate one registered segment-validator kind.

__init__(scene_registry_id, robot_profile_id, scene, robot_profile, call_catalog, relation_grounder_keys, settle_preset_ids, endpoint_adapter_declarations, runtime_transport_declarations, parallel_safety_declaration, control_part_evidence_declaration, registered_semantic_lowerer_declarations, fingerprint, _required_skills)#

preflight(program)[source]#

Compile and statically link every expanded semantic call.

Return type:: CompiledTaskProgram

validate_bound_endpoint_extensions(bound_profile)[source]#

Match every live resolved endpoint to its fingerprinted declaration.

Return type:: None

validate_engine(engine)[source]#

Require the live engine to expose every statically selected skill.

Return type:: None

validate_integration(integration, *, path)[source]#

Validate exact scene, profile, and runtime-preset selection.

Return type:: None

validate_post_policy(policy, *, path)[source]#

Validate one registered post-policy kind and named preset.

Return type:: None

validate_scene_reference(reference, *, role, path)[source]#

Validate one typed scene reference against its declared role.

Return type:: None

validate_semantic_call(call, *, path)[source]#

Validate that the semantic call is present in the integration catalog.

Return type:: None

validate_validator(validator, *, path)[source]#

Validate one registered segment-validator kind.

Return type:: None

class embodichain.lab.task_program.integrations.SimulationTaskProgramRegistration[source]#

Exact immutable task-owned simulation integration registration.

Methods:

`__init__`(scene_binding, robot_profile_binding)
`assert_unchanged`()	Reject nested declaration drift before live component creation.
`create_control_part_evidence_provider`(*, ...)	Create the registration-owned live control-part evidence provider.
`create_parallel_safety_validator`(*, ...)	Create and strictly validate the registration-owned live safety gate.
`create_registered_semantic_lowerers`(*, ...)	Create fresh live lowerers declared by this registration.
`validate_bound_profile`(bound_profile)	Validate resolved semantic endpoints against this registration.
`validate_engine`(engine)	Validate live atomic skills against this registration.
`validate_robot_profile`(profile)	Validate a live profile against its registered declaration.
`validate_scene_registry`(registry)	Validate a live registry against the registered scene declaration.

Attributes:

`endpoint_adapter_map`	Return custom live adapters keyed by their exact endpoint type.
`fingerprint`	Return the canonical registration fingerprint.

__init__(scene_binding, robot_profile_binding, call_catalog=<factory>, settle_presets=<factory>, relation_grounders=(), handover_pose_providers=(), endpoint_adapters=(), runtime_transports=(), parallel_safety_factory=None, control_part_evidence_factory=None, registered_semantic_lowerer_factories=())#

assert_unchanged()[source]#

Reject nested declaration drift before live component creation.

Return type:: None

create_control_part_evidence_provider(*, simulation, robot, scene_registry, engine, scene_provider)[source]#

Create the registration-owned live control-part evidence provider.

Parameters:

simulation (object) – Simulation that owns the selected robot and sensors.
robot (object) – Exact robot selected by the environment factory.
scene_registry (SceneRegistry) – Exact live semantic scene registry.
engine (AtomicActionEngine) – Atomic-action engine assembled for the robot.
scene_provider (SceneProvider) – Shared synchronized live scene provider.

Return type:

EffectEvidenceProvider | None

Returns:

Fresh registered provider, or None when no factory is declared.

create_parallel_safety_validator(*, simulation, robot, scene_registry, engine)[source]#

Create and strictly validate the registration-owned live safety gate.

Return type:: ParallelCommandSafetyValidator | None

create_registered_semantic_lowerers(*, simulation, robot, scene_registry, engine)[source]#

Create fresh live lowerers declared by this registration.

Return type:: tuple[RegisteredSemanticLowerer, ...]

property endpoint_adapter_map: Mapping[type[ResourceEndpoint], ResourceEndpointAdapter]#: Return custom live adapters keyed by their exact endpoint type.

property fingerprint: str#: Return the canonical registration fingerprint.

validate_bound_profile(bound_profile)[source]#

Validate resolved semantic endpoints against this registration.

Return type:: None

validate_engine(engine)[source]#

Validate live atomic skills against this registration.

Return type:: None

validate_robot_profile(profile)[source]#

Validate a live profile against its registered declaration.

Return type:: None

validate_scene_registry(registry)[source]#

Validate a live registry against the registered scene declaration.

Return type:: None

class embodichain.lab.task_program.integrations.IntegrationFingerprintMismatch[source]#

Raised when a live integration no longer matches its registration.

Methods:

`__init__`(args, *kwargs)
`__new__`(**kwargs)

__init__(*args, **kwargs)#

__new__(**kwargs)#

class embodichain.lab.task_program.integrations.ControlPartEvidenceProviderDeclaration[source]#

Provider-free identity of one control-part evidence factory.

Parameters:

factory_type (type[object]) – Exact immutable factory implementation type.
provider_id (str) – Stable effect-evidence provider identifier.
revision (str) – Exact provider contract revision.

Methods:

__init__(factory_type, provider_id, revision)

__init__(factory_type, provider_id, revision)#

class embodichain.lab.task_program.integrations.ControlPartEvidenceProviderFactory[source]#

Registration-owned factory for one live control-part evidence provider.

Methods:

`__init__`(args, *kwargs)
`__new__`(**kwargs)

__init__(*args, **kwargs)#

__new__(**kwargs)#

class embodichain.lab.task_program.integrations.RegisteredSemanticLowererDeclaration[source]#

Provider-free identity of one registered semantic lowerer factory.

Parameters:

factory_type (type[object]) – Exact immutable factory implementation type.
call_id (str) – Registered semantic-call ID owned by the factory.
revision (str) – Exact factory contract revision.

Methods:

__init__(factory_type, call_id, revision)

__init__(factory_type, call_id, revision)#

class embodichain.lab.task_program.integrations.RegisteredSemanticLowererFactory[source]#

Registration-owned factory for one fresh live semantic lowerer.

Methods:

`__init__`(args, *kwargs)
`__new__`(**kwargs)

__init__(*args, **kwargs)#

__new__(**kwargs)#

class embodichain.lab.task_program.integrations.EndpointAdapterDeclaration[source]#

Provider-free declaration of one exact endpoint adapter.

Methods:

__init__(endpoint_type, adapter_type, ...)

__init__(endpoint_type, adapter_type, adapter_id, runtime_transport_ids, runtime_target_types, tracking_feedback_source_keys, tracking_projector_keys, effect_evidence_source_keys)#

class embodichain.lab.task_program.integrations.RuntimeTransportDeclaration[source]#

Provider-free declaration of one ordered runtime transport encoder.

Methods:

__init__(transport_type, transport_id, ...)

__init__(transport_type, transport_id, target_types, payload_types)#

class embodichain.lab.task_program.integrations.ParallelSafetyDeclaration[source]#

Provider-free identity and transport coverage of one safety factory.

Methods:

__init__(factory_type, validator_id, ...)

__init__(factory_type, validator_id, revision, supported_transport_ids)#

class embodichain.lab.task_program.integrations.ParallelCommandSafetyValidatorFactory[source]#

Registration-owned factory for one authoritative live safety gate.

Methods:

`__init__`(args, *kwargs)
`__new__`(**kwargs)

__init__(*args, **kwargs)#

__new__(**kwargs)#

class embodichain.lab.task_program.integrations.StandardExtensionDeclarations[source]#

Cross-checked provider-free declarations for the standard factory.

Methods:

__init__(endpoint_adapters, ...[, ...])

__init__(endpoint_adapters, runtime_transports, parallel_safety, control_part_evidence, registered_semantic_lowerers=<factory>)#

embodichain.lab.task_program.integrations.VersionedKey(*args, **kwargs)#

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable’s items.

If the argument is a tuple, the return value is the same object.

alias of tuple[str, str]

embodichain.lab.task_program.integrations.default_simulation_settle_presets()[source]#

Return independently owned built-in post-policy presets.

Return type:: Mapping[str, DynamicSettleMonitorCfg]

Simulation bindings#

class embodichain.lab.task_program.integrations.SimulationSceneBinding[source]#

Build one authoritative registry from explicit simulation bindings.

Methods:

`__init__`(registry_id[, rigid_objects, ...])
`build`(simulation)	Build the existing authoritative scene registry.
`declare`()	Project the complete provider-free scene declaration.

__init__(registry_id, rigid_objects=(), articulations=(), links=(), antipodal_grasps=(), support_surfaces=(), containers=(), collision_world_mode=None, *, rigidized_articulations=(), rigidized_articulation_grasps=())#

build(simulation)[source]#

Build the existing authoritative scene registry.

Parameters:: simulation (SimulationManager) – Live simulation used only for explicitly named lookups.
Return type:: SceneRegistry
Returns:: Immutable registry with typed roots, links, and affordances.

declare()[source]#

Project the complete provider-free scene declaration.

Return type:: SceneManifest

class embodichain.lab.task_program.integrations.SimulationRigidObjectBinding[source]#

Explicit binding for one simulation rigid object.

Methods:

__init__(entity_id, simulation_uid[, ...])

__init__(entity_id, simulation_uid, aliases=(), dynamics=SceneDynamics.UNKNOWN, collision_role=SceneCollisionRole.NONE, semantic_type=None, default_grasp_affordance=None, geometry_provider=None)#

class embodichain.lab.task_program.integrations.SimulationRigidizedArticulationObjectBinding[source]#

Bind one locked, floating articulation as a semantic object.

Parameters:

entity_id (str) – Canonical scene object ID.
simulation_uid (str) – Explicit native articulation UID.
locked_qpos (Mapping[str, float]) – Exact locked position for every native joint.
aliases (tuple[str, ...]) – Optional non-authoritative lookup aliases.
dynamics (SceneDynamics) – Semantic dynamics classification.
collision_role (SceneCollisionRole) – Collision-world participation role.
semantic_type (str | None) – Optional semantic object classification.
default_grasp_affordance (str | None) – Optional default grasp affordance ID.
geometry_provider (SceneGeometryProvider | None) – Optional planner-facing collision geometry provider.
joint_position_tolerance (float) – Positive joint-state tolerance no greater than 1e-3. Configured limit coincidence uses a separate fixed epsilon.
link_transform_tolerance (float) – Positive cross-arena transform tolerance no greater than 1e-5.

Attention

Live configured assembly accepts only floating roots whose native joint limits are finite and coincident at every declared lock position. The coincidence proof uses a fixed safety epsilon and cannot be weakened by an authoring tolerance.

Methods:

__init__(entity_id, simulation_uid, locked_qpos)

__init__(entity_id, simulation_uid, locked_qpos, aliases=(), dynamics=SceneDynamics.UNKNOWN, collision_role=SceneCollisionRole.NONE, semantic_type=None, default_grasp_affordance=None, geometry_provider=None, joint_position_tolerance=0.001, link_transform_tolerance=1e-05)#

class embodichain.lab.task_program.integrations.SimulationArticulationBinding[source]#

Explicit binding for one simulation articulation.

Methods:

__init__(entity_id, simulation_uid[, ...])

__init__(entity_id, simulation_uid, aliases=(), dynamics=SceneDynamics.UNKNOWN, collision_role=SceneCollisionRole.NONE, semantic_type=None, geometry_provider=None)#

class embodichain.lab.task_program.integrations.SimulationArticulationLinkBinding[source]#

Explicit canonical link backed by one native articulation link.

Methods:

__init__(entity_id, articulation_id, ...[, ...])

__init__(entity_id, articulation_id, native_link_name, aliases=(), dynamics=SceneDynamics.UNKNOWN, semantic_type=None)#

class embodichain.lab.task_program.integrations.AntipodalGraspAffordanceBinding[source]#

Build one antipodal grasp affordance from a selected rigid-object mesh.

internal_axis upgrades the payload to AxisAlignAffordance while preserving the same antipodal grasp capability. The axis is expressed in the rigid object’s local frame and can be reused by held-object skills such as Pour.

Methods:

__init__(entity_id, object_id, native_name, ...)

__init__(entity_id, object_id, native_name, revision, aliases=(), relative_pose=(1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0), mesh_env_id=0, internal_axis=None)#

class embodichain.lab.task_program.integrations.RigidizedArticulationAntipodalGraspBinding[source]#

Build a root-frame grasp mesh from one rigidized articulation link.

Parameters:

entity_id (str) – Canonical affordance ID.
object_id (str) – Parent rigidized-articulation object ID.
grasp_link (str) – Native link supplying the grasp mesh.
native_name (str) – Stable native name for the affordance.
revision (str) – Affordance payload revision.
aliases (tuple[str, ...]) – Optional non-authoritative lookup aliases.
relative_pose (tuple[float, ...]) – Flattened affordance-to-object SE(3) transform.

Methods:

__init__(entity_id, object_id, grasp_link, ...)

__init__(entity_id, object_id, grasp_link, native_name, revision, aliases=(), relative_pose=(1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0))#

class embodichain.lab.task_program.integrations.ContainerAffordanceBinding[source]#

Declare one exact object target frame inside a container parent.

Parameters:

entity_id (str) – Canonical ID of the placement affordance.
parent_id (str) – Canonical object, articulation, or link parent ID.
native_name (str) – Stable native name of this target frame.
aliases (tuple[str, ...]) – Optional non-authoritative lookup aliases.
object_target_pose (tuple[float, ...]) – Desired object pose relative to the parent.
object_target_frame (str) – parent composes the parent transform; world_axes uses its position with world-oriented target axes.
minimum_confidence (float) – Minimum parent/affordance observation confidence.
is_default (bool) – Whether this is the parent’s default Place(inside=...) frame.

Methods:

__init__(entity_id, parent_id, native_name)

__init__(entity_id, parent_id, native_name, aliases=(), object_target_pose=(1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0), minimum_confidence=0.0, is_default=False, object_target_frame='parent')#

class embodichain.lab.task_program.integrations.SupportSurfaceAffordanceBinding[source]#

Declare one exact object target frame on a support parent.

Parameters:

entity_id (str) – Canonical ID of the placement affordance.
parent_id (str) – Canonical object, articulation, or link parent ID.
native_name (str) – Stable native name of this target frame.
aliases (tuple[str, ...]) – Optional non-authoritative lookup aliases.
object_target_pose (tuple[float, ...]) – Desired object pose relative to the parent.
object_target_frame (str) – parent composes the complete parent transform; world_axes adds its translation to the parent position while keeping the declared orientation and offset in world axes.
minimum_confidence (float) – Minimum parent/affordance observation confidence.
is_default (bool) – Whether this is the parent’s default Place(on=...) frame.

Methods:

__init__(entity_id, parent_id, native_name)

__init__(entity_id, parent_id, native_name, aliases=(), object_target_pose=(1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0), minimum_confidence=0.0, is_default=False, object_target_frame='parent')#

class embodichain.lab.task_program.integrations.ControlPartCommandPreset[source]#

Named one-dimensional joint commands for one exact control part.

Methods:

`__init__`(preset_id, control_part, commands)
`build`(*, control_dof)	Build a command profile after validating the native control width.
`declare`()	Build a provider-free command profile from declared tuple widths.

__init__(preset_id, control_part, commands)#

build(*, control_dof)[source]#

Build a command profile after validating the native control width.

Return type:: ControlPartCommandProfile

declare()[source]#

Build a provider-free command profile from declared tuple widths.

Return type:: ControlPartCommandProfile

class embodichain.lab.task_program.integrations.ControlPartEndpointBinding[source]#

Profile endpoint backed by one explicit robot control part.

Methods:

`__init__`(endpoint_id, control_part, capabilities)
`build`(robot)	Build a joint-backed endpoint after native control-part validation.
`declare`()	Return the endpoint contract without reading a robot.

__init__(endpoint_id, control_part, capabilities, command_preset=None)#

build(robot)[source]#

Build a joint-backed endpoint after native control-part validation.

Return type:: ResourceEndpoint

declare()[source]#

Return the endpoint contract without reading a robot.

Return type:: ResourceEndpoint

class embodichain.lab.task_program.integrations.ControlPartResourceBinding[source]#

Joint-backed robot resource containing control-part endpoints.

Methods:

`__init__`(resource_id[, endpoints, members])
`build`(robot)	Build a resource containing strictly validated control-part endpoints.
`declare`()	Return the resource graph without reading native control parts.

__init__(resource_id, endpoints=(), members=())#

build(robot)[source]#

Build a resource containing strictly validated control-part endpoints.

Return type:: RobotResource

declare()[source]#

Return the resource graph without reading native control parts.

Return type:: RobotResource

class embodichain.lab.task_program.integrations.SimulationRobotSkillProfileBinding[source]#

Build a profile from typed resources with strict native validation.

Methods:

`__init__`(profile_id, resources[, ...])
`build`(robot)	Build the existing profile after validating every typed resource.
`declare`()	Project the complete provider-free robot skill profile.

__init__(profile_id, resources, command_presets=(), defaults=<factory>, presets=(), default_preset=None, skill_presets=<factory>, grounding_providers=<factory>)#

build(robot)[source]#

Build the existing profile after validating every typed resource.

Parameters:: robot (Robot) – Live robot selected by the simulation factory.
Return type:: RobotSkillProfile
Returns:: Reusable, engine-independent robot skill profile.

declare()[source]#

Project the complete provider-free robot skill profile.

Return type:: RobotSkillProfile

class embodichain.lab.task_program.integrations.SimulationTaskProgramFactory[source]#

Build every live Task Program component from one task registration.

Parameters:

simulation (SimulationManager) – Exact live simulation that owns robot and scene UIDs.
robot (Robot) – Exact robot selected for planning and evidence acquisition.
registration (SimulationTaskProgramRegistration) – Frozen scene, profile, and extension declarations.
step_dt (float) – Authoritative Gym control cadence.
joint_command_mode (Literal['position', 'position_velocity']) – Environment-owned expert joint command mode.
planner_cfg (BasePlannerCfg | None) – Explicit planner configuration. None selects TOPPRA for robot.uid.
planner_config – Optional executable-free planner declaration from an execution policy. It is converted to planner_cfg after the selected runtime robot is known.
motion_generator_factory (Callable[[], MotionGenerator] | None) – Optional fresh-generator factory. It is mutually exclusive with planner_cfg and intended for custom planners and isolated tests.
grasp_pose_generators (Mapping[str, GraspPoseGenerator] | None) – Standalone grasp-pose services keyed by grasp endpoint target ID, normally a robot control-part name.
translation_threshold (float) – Material scene translation threshold.
rotation_threshold (float) – Material scene rotation threshold.

The Gym cadence is attached to each live PlanningContext. Motion policy presets describe behavior; they do not own the environment’s command period.

Methods:

`__init__`(simulation, robot, registration, *, ...)
`create_adapter`()	Create the exact Gym adapter with shared simulation policy ports.
`create_atomic_action_engine`(profile)	Create a fresh engine around the selected planner and exact profile.
`create_effect_evidence_providers`(*, ...)	Create built-in control-part and articulation evidence providers.
`create_parallel_command_safety_validator`(*, ...)	Create the registration-owned validator for this runtime assembly.
`create_planning_observation_provider`(*, ...)	Create one planning port and planner-validated shared scene provider.
`create_registered_semantic_lowerers`(*, ...)	Create fresh registration-owned lowerers for this runtime assembly.
`create_robot_skill_profile`()	Build the robot profile used to lower calls to Atomic Skills.
`create_scene_registry`()	Build one fresh authoritative registry from the task registration.
`from_environment`(environment, *, registration)	Create a registered factory from the standard Gym environment surface.
`registration_owned_segment_policy_ports`()	Return registration-owned post-policy and validator ports.

Attributes:

`robot_profile_id`	Return the exact configured robot-profile ID.
`scene_registry_id`	Return the exact configured scene-registry ID.
`segment_policy_port`	Return the shared simulation post-policy and validator port.
`step_dt`	Return the authoritative Gym control cadence.
`task_program_registration`	Return the exact standard registration owned by this factory.

__init__(simulation, robot, registration, *, step_dt, joint_command_mode='position', planner_cfg=None, motion_generator_factory=None, grasp_pose_generators=None, translation_threshold=0.0001, rotation_threshold=0.001)[source]#

create_adapter()[source]#

Create the exact Gym adapter with shared simulation policy ports.

Return type:: TaskProgramEnvironmentAdapter

create_atomic_action_engine(profile)[source]#

Create a fresh engine around the selected planner and exact profile.

Return type:: AtomicActionEngine

create_effect_evidence_providers(*, scene_registry, engine, observation_provider)[source]#

Create built-in control-part and articulation evidence providers.

Return type:: Iterable[EffectEvidenceProvider]

create_parallel_command_safety_validator(*, scene_registry, engine, observation_provider)[source]#

Create the registration-owned validator for this runtime assembly.

Return type:: ParallelCommandSafetyValidator

create_planning_observation_provider(*, scene_registry, engine, clock)[source]#

Create one planning port and planner-validated shared scene provider.

Return type:: PlanningObservationPort

create_registered_semantic_lowerers(*, scene_registry, engine)[source]#

Create fresh registration-owned lowerers for this runtime assembly.

Return type:: tuple[RegisteredSemanticLowerer, ...]

create_robot_skill_profile()[source]#

Build the robot profile used to lower calls to Atomic Skills.

Return type:: RobotSkillProfile

create_scene_registry()[source]#

Build one fresh authoritative registry from the task registration.

Return type:: SceneRegistry

classmethod from_environment(environment, *, registration, planner_cfg=None, planner_config=None, motion_generator_factory=None, grasp_pose_generators=None, translation_threshold=0.0001, rotation_threshold=0.001)[source]#

Create a registered factory from the standard Gym environment surface.

Return type:: SimulationTaskProgramFactory

registration_owned_segment_policy_ports()[source]#

Return registration-owned post-policy and validator ports.

Return type:: tuple[SimulationSegmentPolicyPort, SimulationSegmentPolicyPort]

property robot_profile_id: str#: Return the exact configured robot-profile ID.

property scene_registry_id: str#: Return the exact configured scene-registry ID.

property segment_policy_port: SimulationSegmentPolicyPort#: Return the shared simulation post-policy and validator port.

property step_dt: float#: Return the authoritative Gym control cadence.

property task_program_registration: SimulationTaskProgramRegistration#: Return the exact standard registration owned by this factory.

class embodichain.lab.task_program.integrations.SimulationTaskProgramAdapterFactory[source]#

Create one standard simulation adapter after an environment is initialized.

Parameters:

registration (SimulationTaskProgramRegistration) – Immutable provider-free task integration.
grasp_pose_generator_factories (Mapping[str, Callable[[], GraspPoseGenerator]] | None) – Zero-argument factories keyed by runtime grasp endpoint target ID. Each environment receives fresh services.
planner_config (Mapping[str, object] | None) – Optional execution-policy planner declaration. The live factory supplies the selected robot and simulation instance values.

Methods:

`__init__`(registration, *[, ...])
`create_adapter`(environment)	Create fresh grasp services and bind the initialized environment.

Attributes:

registration

Return the exact static registration owned by this factory.

__init__(registration, *, grasp_pose_generator_factories=None, planner_config=None)[source]#

create_adapter(environment)[source]#

Create fresh grasp services and bind the initialized environment.

Return type:: TaskProgramEnvironmentAdapter

property registration: SimulationTaskProgramRegistration#: Return the exact static registration owned by this factory.

class embodichain.lab.task_program.integrations.ConfiguredHandOverPoseProvider[source]#

Resolve a hand-over delivery target from immutable configuration.

The provider carries object-space poses rather than arm trajectories. The shared semantic compiler and atomic HandOver implementation remain responsible for grasp selection, IK, motion generation, transfer, release, and delivery. Keeping the numeric declaration as tuple fields also makes the provider suitable for task-registration catalog fingerprinting.

Parameters:

final_position (tuple[float, float, float]) – World-frame object delivery position.
final_quaternion_xyzw (tuple[float, float, float, float]) – World-frame object delivery orientation.

Methods:

`__init__`(final_position, final_quaternion_xyzw)
`resolve`(call, *, context, bound)	Return an independently owned object-space delivery target.

__init__(final_position, final_quaternion_xyzw)#

resolve(call, *, context, bound)[source]#

Return an independently owned object-space delivery target.

Parameters:

call (HandOver) – Canonical hand-over semantic call.
context (PlanningContext) – Latest immutable planning observation.
bound (BoundSemanticCall) – Engine/profile-bound hand-over call.

Return type:

HandOverPoseTargets

Returns:

Configured final object-space target.

class embodichain.lab.task_program.integrations.CuroboParallelCommandSafetyValidator[source]#

Validate the exact synchronized joint segment before transport dispatch.

Parameters:

robot (object) – Live robot supplying measured joint state and control-part IDs.
motion_generator (MotionGenerator) – Runtime motion generator backed by exact cuRobo.
scene_registry (SceneRegistry) – Authoritative live collision-scene registry.
validation_control_part (str) – Aggregate control part for merged commands.
max_joint_step (float) – Maximum displacement between collision samples.
max_interpolation_samples (int) – Fail-closed sample-count upper bound.

Methods:

`__init__`(*, robot, motion_generator, ...)
`validate`(*, branch_frames, merged_frame)	Reject a merged command whose exact interpolated segment collides.

__init__(*, robot, motion_generator, scene_registry, validation_control_part, max_joint_step, max_interpolation_samples)[source]#

validate(*, branch_frames, merged_frame)[source]#

Reject a merged command whose exact interpolated segment collides.

Return type:: None

class embodichain.lab.task_program.integrations.CuroboParallelSafetyValidatorFactory[source]#

Create exact-sample collision gates for one aggregate control part.

validation_control_part must contain every joint that any parallel branch can command. A common dual-arm example is "dual_arm". The cuRobo model for that part remains the authoritative bounds, self-collision, and world-collision model.

Parameters:

validation_control_part (str) – Aggregate robot control part containing every joint that a parallel lane may command.
max_joint_step (float) – Maximum absolute joint displacement between collision samples in radians or the joint’s native linear unit.
max_interpolation_samples (int) – Fail-closed upper bound on samples per frame.

Methods:

`__init__`(validation_control_part[, ...])
`create`(*, simulation, robot, scene_registry, ...)	Create one fresh validator bound to the assembled live runtime.

__init__(validation_control_part, max_joint_step=0.025, max_interpolation_samples=256)#

create(*, simulation, robot, scene_registry, engine)[source]#

Create one fresh validator bound to the assembled live runtime.

Return type:: CuroboParallelCommandSafetyValidator

class embodichain.lab.task_program.integrations.SimulationSegmentPolicyPort[source]#

Execute built-in segment policies against explicitly bound simulation data.

Parameters:

simulation (SimulationManager) – Live simulation used only for UIDs declared in scene_binding.
robot (Robot) – Live robot used to produce full target-qpos holds while the post-policy observes settling.
scene_binding (SimulationSceneBinding) – Exact canonical-to-native scene declaration.
settle_presets (Mapping[str, DynamicSettleMonitorCfg] | None) – Named settling policies. None installs the shared rigid_object and articulation presets.
env_ids (Tensor | None) – Optional stable logical row IDs. They describe correlation, not simulator row indices; simulator rows remain ordered exactly as returned by the robot and bound entities.

The same instance implements both SegmentPostPolicyPort and SegmentValidatorPort. Unknown policy types, presets, canonical IDs, or native entities fail before an action is emitted.

Methods:

`__init__`(simulation, robot, scene_binding, *)
`actions`(policy, *, segment, active_mask)	Yield full target-qpos hold actions until rows settle or time out.
`begin_segment`(*, segment, active_mask)	Capture lift baselines just before the segment plans and executes.
`post_policy_metadata`(policy, *, segment)	Return the latest JSON-safe settling trace for one policy.
`post_policy_result`(policy, *, segment)	Return the latest independently owned per-row settling result.
`validate`(validator, *, segment)	Observe one explicitly bound entity against its validator contract.
`validate_policy`(policy, *, segment)	Validate one post-policy against static bindings without observation.
`validate_validator`(validator, *, segment)	Validate one validator against static bindings without observation.
`validator_metadata`(validator, *, segment)	Return an owned JSON-safe trace for one completed validator.

Attributes:

settle_preset_ids

Return installed post-policy preset IDs in declaration order.

__init__(simulation, robot, scene_binding, *, settle_presets=None, env_ids=None)[source]#

actions(policy, *, segment, active_mask)[source]#

Yield full target-qpos hold actions until rows settle or time out.

Parameters:

policy (Any) – Exact compiled wait_stable policy.
segment (Any) – Exact segment that owns policy.
active_mask (Tensor) – Rows that remain eligible after runtime execution and preceding post-policies. Inactive rows are held safely but do not participate in settling, timeout, or success results.

Yields:

Fresh full target-qpos hold commands consumed by ordinary env.step(). Reading drive targets instead of measured joint positions preserves contact preload in position-controlled tools. Rows inactive when the policy starts use fresh measured qpos holds.

Return type:

Iterator[Tensor]

Timeout is a normal row-local result boundary. Timed-out rows are exposed through post_policy_result() and post_policy_metadata(); no batch-level exception is raised.

begin_segment(*, segment, active_mask)[source]#

Capture lift baselines just before the segment plans and executes.

This method only reads observed poses. Ineligible rows remain rejected, and non-finite observations reject their own rows during validation.

Return type:: None

post_policy_metadata(policy, *, segment)[source]#

Return the latest JSON-safe settling trace for one policy.

The trace is available after the policy generator has started. A terminal trace has status "settled", "timed_out", or "skipped" when no rows remain active; an early demo interruption intentionally retains the latest "running" snapshot for diagnosis.

Return type:: Mapping[str, object]

post_policy_result(policy, *, segment)[source]#

Return the latest independently owned per-row settling result.

Return type:: Tensor

property settle_preset_ids: tuple[str, ...]#: Return installed post-policy preset IDs in declaration order.

validate(validator, *, segment)[source]#

Observe one explicitly bound entity against its validator contract.

Parameters:

validator (Any) – Exact compiled segment validator.
segment (Any) – Exact segment that owns validator.

Return type:

Tensor

Returns:

Boolean tensor with one result per simulation row.

validate_policy(policy, *, segment)[source]#

Validate one post-policy against static bindings without observation.

This method reads only the compiled declaration, installed preset table, and entities resolved when the port was constructed. It never samples velocity or qpos and never emits a controller action.

Return type:: None

validate_validator(validator, *, segment)[source]#

Validate one validator against static bindings without observation.

Return type:: None

validator_metadata(validator, *, segment)[source]#

Return an owned JSON-safe trace for one completed validator.

Return type:: Mapping[str, object]

embodichain.lab.task_program.integrations.create_simulation_task_program_adapter(environment, *, registration, planner_cfg=None, planner_config=None, motion_generator_factory=None, grasp_pose_generators=None, translation_threshold=0.0001, rotation_threshold=0.001)[source]#

Create a complete production adapter from one standard Gym environment.

This is the intended task-side one-line integration. Standard tasks pass one immutable registration that owns their pre-simulation catalog and every runtime extension.

Parameters:

environment (SimulationTaskProgramEnvironment) – Standard Gym simulation environment exposing sim, robot, and step_dt.
registration (SimulationTaskProgramRegistration) – Required immutable task registration.
planner_cfg (BasePlannerCfg | None) – Optional planner configuration owned by the factory.
planner_config (Mapping[str, object] | None) – Optional executable-free planner declaration from an execution policy.
motion_generator_factory (Callable[[], MotionGenerator] | None) – Optional factory for one fresh motion generator.
grasp_pose_generators (Mapping[str, GraspPoseGenerator] | None) – Standalone grasp-pose services keyed by grasp endpoint target ID.
translation_threshold (float) – Scene translation revision threshold.
rotation_threshold (float) – Scene rotation revision threshold.

Return type:

TaskProgramEnvironmentAdapter

Returns:

Complete production Task Program environment adapter.

Implementation module exports#

`CuroboParallelCommandSafetyValidator`	Validate the exact synchronized joint segment before transport dispatch.
`CuroboParallelSafetyValidatorFactory`	Create exact-sample collision gates for one aggregate control part.

`TaskProgramIntegrationCatalog`	Provider-free integration directory owned by one task registration.
`IntegrationFingerprintMismatch`	Raised when a live integration no longer matches its registration.
`SimulationTaskProgramRegistration`	Exact immutable task-owned simulation integration registration.

`ControlPartEvidenceProviderDeclaration`	Provider-free identity of one control-part evidence factory.
`ControlPartEvidenceProviderFactory`	Registration-owned factory for one live control-part evidence provider.
`EndpointAdapterDeclaration`	Provider-free declaration of one exact endpoint adapter.
`ParallelCommandSafetyValidatorFactory`	Registration-owned factory for one authoritative live safety gate.
`ParallelSafetyDeclaration`	Provider-free identity and transport coverage of one safety factory.
`RegisteredSemanticLowererDeclaration`	Provider-free identity of one registered semantic lowerer factory.
`RegisteredSemanticLowererFactory`	Registration-owned factory for one fresh live semantic lowerer.
`RuntimeTransportDeclaration`	Provider-free declaration of one ordered runtime transport encoder.
`StandardExtensionDeclarations`	Cross-checked provider-free declarations for the standard factory.
`VersionedKey`	Exact `(provider_or_projector_id, revision)` registry key.
`build_standard_extension_declarations`(*, ...)	Cross-check standard-runtime extensions against one exact profile.
`declare_control_part_evidence_factory`(factory)	Read one control-part evidence factory's exact static identity.
`declare_endpoint_adapter`(adapter)	Read one endpoint adapter's exact static extension contract.
`declare_parallel_safety_factory`(factory)	Read one safety factory's exact static identity and coverage.
`declare_registered_semantic_lowerer_factory`(factory)	Read one semantic lowerer factory's exact static identity.
`declare_runtime_transport`(transport)	Read one runtime encoder's exact static target/payload contract.
`validate_immutable_extension_declaration`(...)	Accept only a deeply immutable frozen dataclass or stateless instance.

embodichain.lab.task_program.integrations.extensions.build_standard_extension_declarations(*, profile, endpoint_adapters, runtime_transports, parallel_safety_factory, control_part_evidence_factory=None, registered_semantic_lowerer_factories=())[source]#

Cross-check standard-runtime extensions against one exact profile.

The built-in control-part adapter and joint-position transport cannot be overridden. They are installed first only when the profile uses a ControlPartEndpoint; a pure-custom profile contains only its custom declarations. Custom adapters and transports must cover exactly the endpoint types and transport IDs used by the registered profile; unused declarations fail closed.

Return type:: StandardExtensionDeclarations

embodichain.lab.task_program.integrations.extensions.declare_control_part_evidence_factory(factory)[source]#

Read one control-part evidence factory’s exact static identity.

Parameters:: factory (ControlPartEvidenceProviderFactory) – Immutable registration-owned factory declaration.
Return type:: ControlPartEvidenceProviderDeclaration
Returns:: Provider-free exact factory identity.

embodichain.lab.task_program.integrations.extensions.declare_endpoint_adapter(adapter)[source]#

Read one endpoint adapter’s exact static extension contract.

Return type:: EndpointAdapterDeclaration

embodichain.lab.task_program.integrations.extensions.declare_parallel_safety_factory(factory)[source]#

Read one safety factory’s exact static identity and coverage.

Return type:: ParallelSafetyDeclaration

embodichain.lab.task_program.integrations.extensions.declare_registered_semantic_lowerer_factory(factory)[source]#

Read one semantic lowerer factory’s exact static identity.

Parameters:

factory (RegisteredSemanticLowererFactory) – Immutable registration-owned lowerer factory.

Return type:

RegisteredSemanticLowererDeclaration

Returns:

Provider-free call, revision, and factory-type declaration.

Raises:

TypeError – If the factory has no callable creator or is mutable.
ValueError – If its call ID or revision is not an exact identifier.

embodichain.lab.task_program.integrations.extensions.declare_runtime_transport(transport)[source]#

Read one runtime encoder’s exact static target/payload contract.

Return type:: RuntimeTransportDeclaration

embodichain.lab.task_program.integrations.extensions.validate_immutable_extension_declaration(value, *, field_name)[source]#

Accept only a deeply immutable frozen dataclass or stateless instance.

Frozen dataclass fields may contain only immutable scalar values, types, enums with immutable values, exact tuples, exact frozensets, and recursively frozen dataclasses. Mutable leaves such as mappings, lists, sets, bytearrays, and tensors are rejected because registration-owned live extensions are shared with an assembled runtime. A non-dataclass extension must not have instance or slot state at all.

Return type:: None

embodichain.lab.task_program.integrations.extensions.VersionedKey(*args, **kwargs)#

Exact (provider_or_projector_id, revision) registry key.

alias of tuple[str, str]

embodichain.lab.task_program.integrations

Contents

embodichain.lab.task_program.integrations#

Environment assembly#

Registration and extensions#

Simulation bindings#

Implementation module exports#