embodichain.lab.task_program.semantics

Return the registered extension identifier.

__init__(call_id, arguments=<factory>, *, resources=<factory>)

property semantic_id: str: Return the registered extension identifier.

class embodichain.lab.task_program.semantics.RegistrySceneProvider[source]

Stateful scene provider derived from an immutable registry.

Instances are created by SceneRegistry.make_scene_provider(); each instance owns its revision counters and material-pose baselines.

Parameters:

registry (SceneRegistry) – Immutable catalog that owns entity registrations.
translation_threshold (float) – Accumulated translation needed to publish a material scene change.
rotation_threshold (float) – Accumulated rotation needed to publish a material scene change.
batch_size (int | None) – Optional fixed execution batch size. Factory-created planning providers bind this value before their first observation.

Methods:

`__init__`(registry, *, translation_threshold, ...)
`snapshot`(*, timestamp, env_ids)	Observe all canonical entities and advance material revisions.

Attributes:

collision_world_mode

Return the configured or first-snapshot-resolved collision mode.

__init__(registry, *, translation_threshold, rotation_threshold, batch_size=None)[source]

property collision_world_mode: SceneCollisionWorldMode | None: Return the configured or first-snapshot-resolved collision mode.

snapshot(*, timestamp, env_ids)[source]

Observe all canonical entities and advance material revisions.

Parameters:

timestamp (float) – Non-negative monotonic observation timestamp.
env_ids (Tensor) – Stable ordered correlation IDs for every environment row.

Return type:

SceneSnapshot

Returns:

An immutable snapshot keyed only by canonical registry IDs.

class embodichain.lab.task_program.semantics.ResolvedResourceEndpoint[source]

Endpoint declaration resolved by one registered adapter.

Methods:

`__init__`(endpoint, adapter_id, runtime_target)
`conflicts_with`(other)	Return whether two endpoints address overlapping physical channels.

Attributes:

Return capabilities declared by the source endpoint.

__init__(endpoint, adapter_id, runtime_target, task_state_key=None, effect_sources=<factory>, tracking_channels=<factory>, command_profile_key=None, requires_command_profile=False, commands=<factory>, claim_tokens=frozenset({}), joint_ids=(), exclusive=True)

property capabilities: frozenset[str]: Return capabilities declared by the source endpoint.

conflicts_with(other)[source]

Return whether two endpoints address overlapping physical channels.

Return type:: bool

class embodichain.lab.task_program.semantics.ResolvedRobotResource[source]

Robot-validated resource with concrete endpoint joint IDs and claim.

Methods:

__init__(resource_id, endpoints, members, claim)

Attributes:

endpoint_joint_ids

Return ordered joint IDs for each resolved endpoint.

__init__(resource_id, endpoints, members, claim)

property endpoint_joint_ids: Mapping[str, tuple[int, ...]]: Return ordered joint IDs for each resolved endpoint.

class embodichain.lab.task_program.semantics.ResolvedSkillBinding[source]

One generic resource assignment lowered for the current action core.

Methods:

__init__(skill_id, resources, ...)

Attributes:

resource_ids

Return the selected logical resource ID for each skill-local slot.

__init__(skill_id, resources, action_binding, claim)

property resource_ids: Mapping[str, str]: Return the selected logical resource ID for each skill-local slot.

class embodichain.lab.task_program.semantics.ResourceBinding[source]

Generic mapping from skill-local slots to robot resource IDs.

Methods:

__init__(resources)

__init__(resources)

class embodichain.lab.task_program.semantics.ResourceClaim[source]

Physical leaf and joint claim used for deterministic conflict checks.

Methods:

`__init__`(leaf_resource_ids, joint_ids[, ...])
`combine`(claims)	Return the union of zero or more resource claims.
`conflicts_with`(other)	Return whether two claims overlap in a leaf or concrete joint.

__init__(leaf_resource_ids, joint_ids, claim_tokens=frozenset({}))

classmethod combine(claims)[source]

Return the union of zero or more resource claims.

Return type:: ResourceClaim

conflicts_with(other)[source]

Return whether two claims overlap in a leaf or concrete joint.

Return type:: bool

class embodichain.lab.task_program.semantics.ResourceEndpoint[source]

Extensible execution endpoint in a robot resource graph.

Endpoint subclasses add controller-specific addressing data. Capabilities stay on this common base so skill matching does not depend on any one controller kind.

Methods:

`__init__`(*[, capabilities])
`snapshot`()	Return an independently owned endpoint declaration.

Attributes:

SceneEntityRegistration

Open, namespaced capabilities provided by this exact endpoint.

__init__(*, capabilities=frozenset({}))

capabilities: frozenset[str]: Open, namespaced capabilities provided by this exact endpoint.

snapshot()[source]

Return an independently owned endpoint declaration.

Endpoint subclasses with payloads that cannot be deep-copied must override this method and return a new value of their exact type.

Return type:: ResourceEndpoint

class embodichain.lab.task_program.semantics.ResourceEndpointAdapter[source]

Resolve one endpoint kind without coupling profiles to its controller.

Attributes:

`adapter_id`	Stable adapter identifier used in diagnostics and resolved metadata.
`effect_evidence_source_keys`	Exact `(provider_id, revision)` effect-evidence routes emitted.
`endpoint_type`	Exact endpoint declaration type accepted by this adapter.
`runtime_target_types`	Exact immutable runtime-target value types this adapter may resolve.
`runtime_transport_ids`	Exact endpoint-command transport IDs this adapter may resolve.
`tracking_feedback_source_keys`	Exact `(provider_id, revision)` tracking-feedback routes emitted.
`tracking_projector_keys`	Exact `(projector_id, revision)` desired-state routes emitted.

Methods:

resolve(endpoint, *, engine)

Validate and resolve one endpoint against an action engine.

adapter_id: ClassVar[str]: Stable adapter identifier used in diagnostics and resolved metadata.

effect_evidence_source_keys: ClassVar[frozenset[tuple[str, str]]]: Exact (provider_id, revision) effect-evidence routes emitted.

endpoint_type: ClassVar[type[ResourceEndpoint]]: Exact endpoint declaration type accepted by this adapter.

abstract resolve(endpoint, *, engine)[source]

Validate and resolve one endpoint against an action engine.

Parameters:

endpoint (ResourceEndpoint) – Endpoint declaration of endpoint_type.
engine (AtomicActionEngine) – Engine whose robot, planner, and command profiles are bound.

Return type:

EndpointResolution

Returns:

Physical claims and supported lowering metadata.

runtime_target_types: ClassVar[tuple[type[RuntimeEndpointTarget], ...]]: Exact immutable runtime-target value types this adapter may resolve.

runtime_transport_ids: ClassVar[frozenset[str]]: Exact endpoint-command transport IDs this adapter may resolve.

tracking_feedback_source_keys: ClassVar[frozenset[tuple[str, str]]]: Exact (provider_id, revision) tracking-feedback routes emitted.

tracking_projector_keys: ClassVar[frozenset[tuple[str, str]]]: Exact (projector_id, revision) desired-state routes emitted.

class embodichain.lab.task_program.semantics.RobotResource[source]

Generic leaf or composite resource in one robot’s resource DAG.

A resource may expose any number of named endpoints. For example, one manipulation participant may expose motion and grasp endpoints, while a mobile base or whole-body controller may expose only motion. members describes physical claim composition and does not inherit endpoint capabilities.

Methods:

`__init__`(resource_id[, endpoints, members])
`snapshot`()	Return an independently owned resource declaration.

__init__(resource_id, endpoints=<factory>, members=())

snapshot()[source]

Return an independently owned resource declaration.

Return type:: RobotResource

class embodichain.lab.task_program.semantics.RobotSkillProfile[source]

Reusable declarative skill integration for one robot embodiment.

Methods:

`__init__`(profile_id, resources[, ...])
`action_control_profiles`()	Lower endpoint command profiles for the current action core.
`bind`(engine, *[, endpoint_adapters])	Validate this profile against one fully configured action engine.

Attributes:

grounding_providers

Semantic call ID to embodiment-owned named grounding provider ID.

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

action_control_profiles()[source]

Lower endpoint command profiles for the current action core.

Return type:: Mapping[str, ControlPartCommandProfile]
Returns:: Owned command profiles keyed by concrete robot control-part name.
Raises:: ValueError – If two endpoint declarations assign non-equivalent commands with the same semantic name to one control part.

bind(engine, *, endpoint_adapters=None)[source]

Validate this profile against one fully configured action engine.

Parameters:

engine (AtomicActionEngine) – Installed atomic-action engine for the target robot.
endpoint_adapters (Optional[Mapping[type[ResourceEndpoint], ResourceEndpointAdapter]]) – Optional exact endpoint-type adapters. Explicit entries extend the non-overridable built-in control-part adapter.

Return type:

BoundRobotSkillProfile

Returns:

Robot-, engine-, and adapter-validated profile view.

grounding_providers: Mapping[str, str]: Semantic call ID to embodiment-owned named grounding provider ID.

class embodichain.lab.task_program.semantics.ScalarEffectClause[source]

Raw force/wrench magnitude condition with monitor-owned thresholds.

Methods:

`__init__`(clause_id, expectation_id, source, ...)
`snapshot`()	Return an independently owned scalar clause.

__init__(clause_id, expectation_id, source, evidence_kind, expectation)

snapshot()[source]

Return an independently owned scalar clause.

Return type:: ScalarEffectClause

class embodichain.lab.task_program.semantics.ScalarEffectEvidenceBatch[source]

Raw per-row force or wrench-magnitude observations.

Methods:

`__init__`(evidence_id, evidence_kind, values, ...)
`snapshot`()	Return an independently owned evidence batch.
`to_metadata`()	Return raw scalar evidence as JSON-safe trace metadata.

__init__(evidence_id, evidence_kind, values, valid, acquisition_errors, timestamp, env_ids, observation_revision)

snapshot()[source]

Return an independently owned evidence batch.

Return type:: ScalarEffectEvidenceBatch

to_metadata()[source]

Return raw scalar evidence as JSON-safe trace metadata.

Return type:: dict[str, object]

class embodichain.lab.task_program.semantics.ScalarEffectEvidenceQuery[source]

Query for one raw force or wrench magnitude.

Methods:

`__init__`(clause, _expectation)
`snapshot`()	Return an independently owned scalar query.

Attributes:

`evidence_id`	Return the source clause ID.
`expectation`	Return an owned correlated expectation.
`source`	Return an owned source route.

__init__(clause, _expectation)

property evidence_id: str: Return the source clause ID.

property expectation: HeldObjectStateExpectation | CoordinatedHeldObjectCleanupExpectation | ArticulationJointStateExpectation: Return an owned correlated expectation.

snapshot()[source]

Return an independently owned scalar query.

Return type:: ScalarEffectEvidenceQuery

property source: EffectEvidenceSourceRef: Return an owned source route.

class embodichain.lab.task_program.semantics.ScalarEffectObservation[source]

Callback-owned raw scalar values with explicit row validity.

Methods:

__init__(values[, valid, acquisition_errors])

__init__(values, valid=None, acquisition_errors=())

class embodichain.lab.task_program.semantics.ScalarEvidenceKind[source]

Raw scalar physical evidence channel.

Methods:

__new__(value)

__new__(value)

class embodichain.lab.task_program.semantics.ScalarExpectation[source]

Expected high/low magnitude band for scalar evidence.

Methods:

__new__(value)

__new__(value)

class embodichain.lab.task_program.semantics.SceneAffordanceRef[source]

Reference to one registered interaction affordance.

Methods:

__init__(entity_id)

__init__(entity_id)

class embodichain.lab.task_program.semantics.SceneArticulationEvidenceProvider[source]

Typed adapter for scene-articulation joint-state observations.

Integrations inject either a direct observer or a SceneProvider whose snapshot contains ObservedArticulationJointState values. The adapter never discovers live simulator objects from an environment. Repeated clauses share one synchronized snapshot and one sample per exact physical address.

Methods:

`__init__`([observer, scene_provider])
`collect`(queries, context)	Collect synchronized joint state for exact scene addresses.

__init__(observer=None, *, scene_provider=None)[source]

collect(queries, context)[source]

Collect synchronized joint state for exact scene addresses.

Return type:: Mapping[str, PoseRelationEvidenceBatch | BinaryEffectEvidenceBatch | ScalarEffectEvidenceBatch | JointStateEvidenceBatch]

class embodichain.lab.task_program.semantics.SceneArticulationJointStateProvider[source]

Observe canonical joints for one registered scene articulation.

Methods:

`__init__`(args, *kwargs)
`observe_joints`(*, timestamp, env_ids)	Return live joint observations whose rows follow `env_ids`.

__init__(*args, **kwargs)

observe_joints(*, timestamp, env_ids)[source]

Return live joint observations whose rows follow env_ids.

Return type:: Mapping[str, ObservedArticulationJointState]

class embodichain.lab.task_program.semantics.SceneArticulationRef[source]

Reference to one articulation registered in the semantic scene.

Methods:

__init__(entity_id)

__init__(entity_id)

class embodichain.lab.task_program.semantics.SceneCollisionRole[source]

How an entity participates in the planner collision world.

Methods:

__new__(value)

__new__(value)

class embodichain.lab.task_program.semantics.SceneCollisionWorldMode[source]

Batch-sharing policy for a dynamic planner collision world.

Methods:

__new__(value)

__new__(value)

class embodichain.lab.task_program.semantics.SceneDynamics[source]

Physical mobility classification owned by a scene registration.

Methods:

__new__(value)

__new__(value)

class embodichain.lab.task_program.semantics.SceneEntityManifest[source]

Provider-free static scene declaration using canonical registry metadata.

Methods:

`__init__`(ref[, aliases, parent, ...])
`from_metadata`(metadata)	Copy one canonical provider-free registry metadata value.

__init__(ref, aliases=(), parent=None, native_name=None, dynamics=SceneDynamics.UNKNOWN, collision_role=SceneCollisionRole.NONE, semantic_type=None, affordance_capabilities=frozenset({}), default_affordances=<factory>, affordance_payload_type=None, affordance_revision=None, relative_pose=None)

classmethod from_metadata(metadata)[source]

Copy one canonical provider-free registry metadata value.

Return type:: SceneEntityManifest

class embodichain.lab.task_program.semantics.SceneEntityMetadata[source]

Provider-free semantic metadata projected from one registration.

Parameters:

ref (SceneEntityRef) – Canonical typed entity reference.
aliases (tuple[str, ...]) – Boundary aliases, compared as an order-independent set.
parent (SceneEntityRef | None) – Canonical parent for links and affordances.
native_name (str | None) – Backend-local child name.
dynamics (SceneDynamics) – Physical mobility classification.
collision_role (SceneCollisionRole) – Planner collision classification.
semantic_type (str | None) – Optional application semantic type.
affordance_capabilities (frozenset[str]) – Open capabilities of an affordance.
default_affordances (Mapping[str, SceneAffordanceRef]) – Capability-scoped direct-child defaults.
affordance_payload_type (type[Affordance] | None) – Exact registered affordance value type.
affordance_revision (str | None) – Integrator-owned payload revision or fingerprint.
relative_pose (tuple[float, ...] | None) – Flattened parent-relative 4x4 pose, when declared.

Methods:

`__init__`(ref[, aliases, parent, ...])
`from_registration`(registration)	Project semantic metadata without copying a live payload/provider.

__init__(ref, aliases=(), parent=None, native_name=None, dynamics=SceneDynamics.UNKNOWN, collision_role=SceneCollisionRole.NONE, semantic_type=None, affordance_capabilities=frozenset({}), default_affordances=<factory>, affordance_payload_type=None, affordance_revision=None, relative_pose=None)

classmethod from_registration(registration)[source]

Project semantic metadata without copying a live payload/provider.

Return type:: SceneEntityMetadata

class embodichain.lab.task_program.semantics.SceneEntityRef[source]

Typed reference to one authoritative scene-registry entity.

Parameters:: entity_id (str) – Globally stable canonical registry identifier.

Methods:

__init__(entity_id)

Attributes:

entity_id

Globally stable authoritative registry identifier.

__init__(entity_id)

entity_id: str: Globally stable authoritative registry identifier.

class embodichain.lab.task_program.semantics.SceneEntityRegistration[source]

Immutable integration metadata for one authoritative scene entity.

Parent relationships, simulator-native names, pose sources, geometry, and affordances belong to the registry registration rather than the lightweight reference copied into semantic calls.

Parameters:

ref (SceneEntityRef) – Canonical typed reference.
state_provider (SceneEntityStateProvider | None) – Optional dynamic pose/confidence source.
joint_state_provider (SceneArticulationJointStateProvider | None) – Optional live articulation-joint source.
aliases (tuple[str, ...]) – External names normalized at the registry boundary.
parent (SceneEntityRef | None) – Canonical parent for a link or affordance.
native_name (str | None) – Backend-local member name under parent.
dynamics (SceneDynamics) – Physical mobility classification.
geometry_provider (SceneGeometryProvider | None) – Planner-facing collision geometry source.
collision_role (SceneCollisionRole) – Static, dynamic, or no planner collision role.
semantic_type (str | None) – Optional application semantic type.
affordance (Affordance | None) – Affordance value for an affordance registration.
affordance_capabilities (frozenset[str]) – Open semantic operations supported by an affordance registration.
default_affordances (Mapping[str, SceneAffordanceRef]) – Capability-to-child mapping owned by a parent object, articulation, or link registration.
affordance_revision (str | None) – Stable integrator-owned revision or fingerprint for capability-bearing affordance payload data.
relative_pose (Tensor | None) – Optional parent-relative affordance transform.

Methods:

__init__(ref[, state_provider, aliases, ...])

Attributes:

`affordance`	Affordance value owned by a `SceneAffordanceRef` registration.
`affordance_capabilities`	Open semantic capabilities declared by an affordance registration.
`affordance_revision`	Stable payload revision required by capability-bearing affordances.
`aliases`	External names normalized once at the registry boundary.
`collision_role`	Static/dynamic planner-obstacle role, or `none`.
`default_affordances`	Capability-scoped child affordances selected when multiple are valid.
`dynamics`	Static, kinematic, dynamic, or unknown mobility classification.
`geometry_provider`	Collision geometry source required for planner collision roles.
`joint_state_provider`	Explicit live joint source for an articulation registration.
`native_name`	Backend-local link or affordance name under `parent`.
`parent`	Canonical parent reference for a link or affordance.
`ref`	Canonical typed reference owned by the registry.
`relative_pose`	Optional parent-relative pose when no explicit state provider exists.
`semantic_type`	Optional application semantic type such as `container` or `tool`.
`state_provider`	Explicit dynamic pose/confidence source.

__init__(ref, state_provider=None, aliases=(), parent=None, native_name=None, dynamics=SceneDynamics.UNKNOWN, geometry_provider=None, collision_role=SceneCollisionRole.NONE, semantic_type=None, affordance=None, affordance_capabilities=frozenset({}), default_affordances=<factory>, affordance_revision=None, relative_pose=None, joint_state_provider=None)

affordance: Affordance | None: Affordance value owned by a SceneAffordanceRef registration.

affordance_capabilities: frozenset[str]: Open semantic capabilities declared by an affordance registration.

affordance_revision: str | None: Stable payload revision required by capability-bearing affordances.

aliases: tuple[str, ...]: External names normalized once at the registry boundary.

collision_role: SceneCollisionRole: Static/dynamic planner-obstacle role, or none.

default_affordances: Mapping[str, SceneAffordanceRef]: Capability-scoped child affordances selected when multiple are valid.

dynamics: SceneDynamics: Static, kinematic, dynamic, or unknown mobility classification.

geometry_provider: SceneGeometryProvider | None: Collision geometry source required for planner collision roles.

joint_state_provider: SceneArticulationJointStateProvider | None: Explicit live joint source for an articulation registration.

native_name: str | None: Backend-local link or affordance name under parent.

parent: SceneEntityRef | None: Canonical parent reference for a link or affordance.

ref: SceneEntityRef: Canonical typed reference owned by the registry.

relative_pose: Tensor | None: Optional parent-relative pose when no explicit state provider exists.

semantic_type: str | None: Optional application semantic type such as container or tool.

state_provider: SceneEntityStateProvider | None: Explicit dynamic pose/confidence source.

class embodichain.lab.task_program.semantics.SceneEntityStateProvider[source]

Observe one registered entity for an ordered environment batch.

Methods:

`__init__`(args, *kwargs)
`observe`(*, timestamp, env_ids)	Return the entity state whose rows follow `env_ids`.

__init__(*args, **kwargs)

observe(*, timestamp, env_ids)[source]

Return the entity state whose rows follow env_ids.

Parameters:

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

Return type:

EntityState

Returns:

Current pose and confidence for the registered entity.

class embodichain.lab.task_program.semantics.SceneGeometryProvider[source]

Provide one entity’s planner-facing collision geometry descriptor.

Methods:

`__init__`(args, *kwargs)
`get_geometry`()	Return the planner-facing geometry descriptor.

__init__(*args, **kwargs)

get_geometry()[source]

Return the planner-facing geometry descriptor.

Return type:: object
Returns:: Backend-consumable geometry or a live simulation entity.

class embodichain.lab.task_program.semantics.SceneLinkRef[source]

Reference to one registered articulation link.

Methods:

__init__(entity_id)

__init__(entity_id)

class embodichain.lab.task_program.semantics.SceneManifest[source]

Immutable provider-free scene catalog used before simulation starts.

Methods:

`__init__`([entries, collision_world_mode])
`from_registry`(registry)	Project a live registry without observing any dynamic provider.
`lookup`(identifier, *[, expected_type, path])	Return one static entry after canonical typed resolution.
`resolve`(identifier, *[, expected_type, path])	Resolve one canonical or alias reference with pathful diagnostics.
`resolve_affordance`(parent, *, capability[, ...])	Resolve one affordance using the same strict rule as SceneRegistry.
`validate_registry`(registry, *[, path])	Require a live registry to match this provider-free declaration.

Attributes:

entries

Return immutable provider-free entries in declaration order.

__init__(entries=(), *, collision_world_mode=None)[source]

property entries: tuple[SceneEntityManifest, ...]: Return immutable provider-free entries in declaration order.

classmethod from_registry(registry)[source]

Project a live registry without observing any dynamic provider.

Return type:: SceneManifest

lookup(identifier, *, expected_type=<class 'embodichain.lab.task_program.semantics.scene.SceneEntityRef'>, path=())[source]

Return one static entry after canonical typed resolution.

Return type:: SceneEntityManifest

resolve(identifier, *, expected_type=<class 'embodichain.lab.task_program.semantics.scene.SceneEntityRef'>, path=())[source]

Resolve one canonical or alias reference with pathful diagnostics.

Return type:: TypeVar(RefT, bound= SceneEntityRef)

resolve_affordance(parent, *, capability, explicit=None, path=())[source]

Resolve one affordance using the same strict rule as SceneRegistry.

Return type:: SceneAffordanceRef

validate_registry(registry, *, path=('integration', 'scene_registry'))[source]

Require a live registry to match this provider-free declaration.

Return type:: None

class embodichain.lab.task_program.semantics.SceneObjectRef[source]

Reference to one object registered in the semantic scene.

Methods:

__init__(entity_id)

__init__(entity_id)

class embodichain.lab.task_program.semantics.SceneRegistry[source]

Immutable authoritative catalog of semantic scene entities.

Canonical identifiers occupy one flat, globally unique namespace. Aliases are accepted only at lookup and integration boundaries and always resolve to a canonical typed reference before they leave the registry.

Parameters:

registrations (Iterable[SceneEntityRegistration]) – Complete scene registrations. The iterable is copied and cannot be extended after construction.
collision_world_mode (SceneCollisionWorldMode | None) – Explicit dynamic-collision batch policy. It may be omitted for a single environment, which resolves to shared. A multi-environment dynamic world must select a mode explicitly.

Methods:

`__init__`([registrations, collision_world_mode])
`affordances`(parent, *, capability)	Return compatible direct-child affordances without selecting one.
`collision_geometry_by_id`([role])	Materialize planner geometry under canonical registry IDs.
`from_simulation`(simulation, *[, ...])	Opt explicitly selected simulation entities into a registry.
`lookup`(identifier, *[, expected_type])	Return an owned registration after canonical typed resolution.
`make_planning_scene_provider`(...[, ...])	Create a provider after complete planner/registry validation.
`make_scene_provider`(*[, ...])	Create an independent provider without planner cross-validation.
`object_semantics`(object_ref, *, affordance)	Build one owned atomic-action semantic snapshot.
`resolve`(identifier, *[, expected_type])	Resolve a canonical ID or alias to a typed canonical reference.
`resolve_affordance`(parent, *, capability[, ...])	Select one compatible affordance with strict scoped-default rules.
`resolve_collision_world_mode`(*, batch_size)	Resolve the configured collision mode for an execution batch.
`validate_collision_integration`(...[, ...])	Validate registry/planner agreement before dynamic planning.

Attributes:

`aliases`	Return the immutable alias-to-canonical-ID index.
`collision_world_entity_ids`	Return every canonical ID represented in the planner world.
`dynamic_collision_entity_ids`	Return canonical IDs whose planner poses update dynamically.
`entity_metadata`	Return provider-free metadata without copying affordance payloads.
`entity_refs`	Return canonical typed references in registration order.
`registrations`	Return structurally independent registration values.
`static_collision_entity_ids`	Return canonical IDs baked into the static planner world.

__init__(registrations=(), *, collision_world_mode=None)[source]

affordances(parent, *, capability)[source]

Return compatible direct-child affordances without selecting one.

Parameters:

parent (str | SceneEntityRef) – Canonical ID, alias, or typed parent reference.
capability (str) – Required open affordance capability.

Return type:

tuple[SceneAffordanceRef, ...]

Returns:

Compatible canonical references sorted by canonical ID.

property aliases: Mapping[str, str]: Return the immutable alias-to-canonical-ID index.

collision_geometry_by_id(role=None)[source]

Materialize planner geometry under canonical registry IDs.

Parameters:: role (SceneCollisionRole | None) – Optional exact collision-role filter. Without a filter, all static and dynamic collision registrations are included. Registrations whose role is SceneCollisionRole.NONE never enter the planner collision world.
Return type:: Mapping[str, object]
Returns:: Fresh immutable canonical-ID-to-geometry mapping.

property collision_world_entity_ids: tuple[str, ...]: Return every canonical ID represented in the planner world.

property dynamic_collision_entity_ids: tuple[str, ...]: Return canonical IDs whose planner poses update dynamically.

property entity_metadata: tuple[SceneEntityMetadata, ...]: Return provider-free metadata without copying affordance payloads.

property entity_refs: tuple[SceneEntityRef, ...]: Return canonical typed references in registration order.

classmethod from_simulation(simulation, *, rigid_objects=None, articulations=None, collision_roles=None, geometry_providers=None, collision_world_mode=None)[source]

Opt explicitly selected simulation entities into a registry.

rigid_objects and articulations map authoritative registry IDs to simulation UIDs. UIDs become aliases automatically; unlisted simulation entities are never imported. Collision participation defaults to SceneCollisionRole.NONE.

Parameters:

simulation (SimulationManager) – Simulation manager used only for explicit UID lookup.
rigid_objects (Mapping[str, str] | None) – Canonical object IDs mapped to simulation UIDs.
articulations (Mapping[str, str] | None) – Canonical articulation IDs mapped to simulation UIDs.
collision_roles (Mapping[str, SceneCollisionRole] | None) – Optional collision roles keyed by canonical ID.
geometry_providers (Mapping[str, SceneGeometryProvider] | None) – Optional geometry overrides keyed by canonical ID. Selected rigid objects otherwise expose their live handles.
collision_world_mode (SceneCollisionWorldMode | None) – Optional dynamic collision batch-sharing mode.

Return type:

SceneRegistry

Returns:

Immutable registry containing only the explicitly selected entities.

lookup(identifier, *, expected_type=<class 'embodichain.lab.task_program.semantics.scene.SceneEntityRef'>)[source]

Return an owned registration after canonical typed resolution.

Parameters:

identifier (str | SceneEntityRef) – Canonical ID, alias, or typed canonical reference.
expected_type (type[TypeVar(RefT, bound= SceneEntityRef)]) – Required reference class.

Return type:

Returns:

A structurally independent copy of the matching registration.

make_planning_scene_provider(motion_generator, *, batch_size, translation_threshold=0.0001, rotation_threshold=0.001)[source]

Create a provider after complete planner/registry validation.

Parameters:

motion_generator (MotionGenerator) – Motion generator that will consume dynamic poses.
batch_size (int) – Number of execution environments.
translation_threshold (float) – Accumulated translation needed to publish a material scene change.
rotation_threshold (float) – Accumulated rotation needed to publish a material scene change.

Return type:

Returns:

A new independently stateful, planner-validated scene provider.

make_scene_provider(*, translation_threshold=0.0001, rotation_threshold=0.001, batch_size=None)[source]

Create an independent provider without planner cross-validation.

This factory is intended for perception and direct-core consumers. The canonical planning path must use make_planning_scene_provider() so planner IDs, capabilities, and collision-world mode cannot drift.

Parameters:

translation_threshold (float) – Accumulated translation needed to publish a material scene change.
rotation_threshold (float) – Accumulated rotation needed to publish a material scene change.
batch_size (int | None) – Optional fixed integration batch size. Supplying it validates the collision-world mode immediately and binds the provider to that row count.

Return type:

Returns:

A new provider with independent revisions and published baselines.

object_semantics(object_ref, *, affordance)[source]

Build one owned atomic-action semantic snapshot.

Parameters:

object_ref (str | SceneObjectRef) – Canonical object ID, alias, or typed reference.
affordance (str | SceneAffordanceRef) – Registered direct-child affordance for the object.

Return type:

ObjectSemantics

Returns:

Object semantics with an owned affordance payload and canonical ID.

Raises:

ValueError – If the affordance does not belong to the object.

property registrations: tuple[SceneEntityRegistration, ...]: Return structurally independent registration values.

resolve(identifier, *, expected_type=<class 'embodichain.lab.task_program.semantics.scene.SceneEntityRef'>)[source]

Resolve a canonical ID or alias to a typed canonical reference.

Parameters:

identifier (str | SceneEntityRef) – Canonical ID, alias, or already typed canonical ref.
expected_type (type[TypeVar(RefT, bound= SceneEntityRef)]) – Required reference class for typed lookup.

Return type:

TypeVar(RefT, bound= SceneEntityRef)

Returns:

Registry-owned canonical reference.

Raises:

KeyError – If the canonical ID or alias is unknown.
TypeError – If the supplied or resolved reference has the wrong type.

resolve_affordance(parent, *, capability, explicit=None)[source]

Select one compatible affordance with strict scoped-default rules.

Parameters:

parent (str | SceneEntityRef) – Entity that directly owns the affordance.
capability (str) – Required semantic affordance capability.
explicit (str | SceneAffordanceRef | None) – Optional explicit affordance ID or typed reference.

Return type:

SceneAffordanceRef

Returns:

One canonical compatible affordance reference.

Raises:

UnsupportedSceneAffordanceError – If no compatible affordance exists or an explicit affordance has the wrong parent/capability.
AmbiguousSceneAffordanceError – If multiple candidates exist without a scoped default.

resolve_collision_world_mode(*, batch_size)[source]

Resolve the configured collision mode for an execution batch.

Parameters:: batch_size (int) – Number of execution environments.
Return type:: SceneCollisionWorldMode | None
Returns:: The effective mode, or None when no dynamic collision entity is registered.

property static_collision_entity_ids: tuple[str, ...]: Return canonical IDs baked into the static planner world.

validate_collision_integration(motion_generator, *, batch_size, scene_provider=None)[source]

Validate registry/planner agreement before dynamic planning.

Parameters:

motion_generator (MotionGenerator) – Motion generator whose planner consumes obstacles.
batch_size (int) – Number of execution environments.
scene_provider (SceneProvider | None) – Optional external perception or hardware provider. Its concrete collision_entity_ids must agree exactly with the registry and planner declarations.

Return type:

SceneCollisionWorldMode | None

Returns:

Effective dynamic collision mode, or None without dynamic IDs.

class embodichain.lab.task_program.semantics.SemanticCallCatalog[source]

Immutable discovery catalog separated from engine installation.

Methods:

`__init__`(descriptors)
`discover`(call)	Discover metadata without installing or executing an implementation.
`with_descriptor`(descriptor)	Return a new catalog containing one additional descriptor.

Attributes:

descriptors

Return immutable descriptors keyed by exact semantic ID.

__init__(descriptors)[source]

property descriptors: Mapping[str, SemanticCallDescriptor]: Return immutable descriptors keyed by exact semantic ID.

discover(call)[source]

Discover metadata without installing or executing an implementation.

Parameters:

call (str | SemanticCallSpec) – Exact semantic ID or a call value.

Return type:

SemanticCallDescriptor

Returns:

Matching immutable descriptor.

Raises:

KeyError – If the exact call ID is unknown.
TypeError – If the call type disagrees with its descriptor.

with_descriptor(descriptor)[source]

Return a new catalog containing one additional descriptor.

Return type:: SemanticCallCatalog

class embodichain.lab.task_program.semantics.SemanticCallDescriptor[source]

Static catalog metadata for one semantic call kind.

Parameters:

call_id (str) – Stable semantic call identifier.
spec_type (type[SemanticCallSpec]) – Exact public call value type.
target_descriptor (SkillDescriptor | None) – Exact atomic goal/options/resource contract. It is inferred and non-overridable for curated calls and required for registered extensions.

Methods:

__init__(call_id, spec_type[, target_descriptor])

Attributes:

`binding_contract`	Return the resource contract from the canonical target descriptor.
`skill_id`	Return the atomic skill ID from the canonical target descriptor.

__init__(call_id, spec_type, target_descriptor=None)

property binding_contract: SkillBindingContract: Return the resource contract from the canonical target descriptor.

property skill_id: str: Return the atomic skill ID from the canonical target descriptor.

class embodichain.lab.task_program.semantics.SemanticCallSpec[source]

Base value contract shared by every declarative semantic call.

Parameters:: resources (Mapping[str, str]) – Optional skill-local slot to robot-resource overrides.

Methods:

`__init__`(*[, resources])
`to_metadata`()	Return this semantic call as deterministic JSON-safe data.

Attributes:

Return the stable catalog identifier for this call.

__init__(*, resources=<factory>)

property semantic_id: str: Return the stable catalog identifier for this call.

to_metadata()[source]

Return this semantic call as deterministic JSON-safe data.

Return type:: dict[str, object]

class embodichain.lab.task_program.semantics.SemanticDiagnostic[source]

Structured deterministic semantic-integration diagnostic.

Parameters:

code (str) – Stable machine-readable failure code.
path (tuple[str | int, ...]) – Complete configuration or program path.
message (str) – Human-readable explanation.
candidates (tuple[str, ...]) – Canonical candidate IDs, sorted when applicable.

Methods:

__init__(code, path, message[, candidates])

Attributes:

rendered_path

Return the path in dotted/indexed notation.

__init__(code, path, message, candidates=())

property rendered_path: str: Return the path in dotted/indexed notation.

class embodichain.lab.task_program.semantics.SemanticEffectKind[source]

Trace-level semantic effect category; clause types define behavior.

Methods:

__new__(value)

__new__(value)

class embodichain.lab.task_program.semantics.SemanticEffectSpec[source]

Grounded typed physical clauses and symbolic postconditions for one call.

Methods:

`__init__`(semantic_id, effect_kind, skill_id, ...)
`snapshot`()	Return an independently owned grounded effect contract.
`state_expectation`(expectation_id)	Return an owned state expectation by effect-local ID.
`to_metadata`()	Return this grounded effect contract as deterministic JSON-safe data.
`validate_request`(request)	Validate execution identity and typed symbolic postconditions.

__init__(semantic_id, effect_kind, skill_id, invocation_id, invocation_revision, env_ids, state_expectations, clauses)

snapshot()[source]

Return an independently owned grounded effect contract.

Return type:: SemanticEffectSpec

state_expectation(expectation_id)[source]

Return an owned state expectation by effect-local ID.

Return type:: HeldObjectStateExpectation | CoordinatedHeldObjectCleanupExpectation | ArticulationJointStateExpectation

to_metadata()[source]

Return this grounded effect contract as deterministic JSON-safe data.

Return type:: dict[str, object]

validate_request(request)[source]

Validate execution identity and typed symbolic postconditions.

Return type:: None

class embodichain.lab.task_program.semantics.SemanticIntegrationManifest[source]

Static scene/profile/catalog declaration validated before execution.

Parameters:

scene (SceneManifest) – Provider-free scene manifest.
robot_profile (RobotSkillProfile) – Declarative robot resource/profile snapshot.
call_catalog (SemanticCallCatalog) – Discoverable semantic call descriptors.
runtime_preset (str | None) – Optional integration-wide policy preset override.

Methods:

`__init__`(scene, robot_profile, call_catalog)
`bind`(scene_registry, engine, *[, ...])	Validate live scene and robot bindings without observing or planning.
`link_call`(call, *[, path])	Resolve static refs, affordances, and declared resource structure.

__init__(scene, robot_profile, call_catalog, runtime_preset=None)

bind(scene_registry, engine, *, endpoint_adapters=None)[source]

Validate live scene and robot bindings without observing or planning.

Return type:: BoundSemanticIntegration

link_call(call, *, path=('call',))[source]

Resolve static refs, affordances, and declared resource structure.

This method never observes scene providers, constructs an engine, samples a grasp, or runs a planner.

Return type:: LinkedSemanticCall

class embodichain.lab.task_program.semantics.SemanticPose[source]

Object-space pose expressed as position and a WXYZ quaternion.

The value owns normalized tensor snapshots and never exposes its internal tensors directly. A single pose or an environment batch is accepted.

Parameters:

position (Tensor | tuple[float, float, float] | list[float]) – Shape (3,) or (B, 3).
quaternion_wxyz (Tensor | tuple[float, float, float, float] | list[float]) – Shape (4,) or (B, 4). Finite, non-zero quaternions are normalized at construction.

Methods:

`__init__`(position, quaternion_wxyz)
`snapshot`()	Return an independently owned pose value.
`to_matrix`()	Convert the semantic pose to a homogeneous transform.
`to_metadata`()	Return the pose as deterministic JSON-safe semantic data.

Attributes:

`batch_size`	Return the explicit batch size, or `None` for one broadcast pose.
`position`	Return an independent position tensor.
`quaternion_wxyz`	Return an independent normalized quaternion tensor.

__init__(position, quaternion_wxyz)[source]

property batch_size: int | None: Return the explicit batch size, or None for one broadcast pose.

property position: Tensor: Return an independent position tensor.

property quaternion_wxyz: Tensor: Return an independent normalized quaternion tensor.

snapshot()[source]

Return an independently owned pose value.

Return type:: SemanticPose

to_matrix()[source]

Convert the semantic pose to a homogeneous transform.

Return type:: Tensor
Returns:: Shape (4, 4) for an unbatched pose or (B, 4, 4) for a batched pose.

to_metadata()[source]

Return the pose as deterministic JSON-safe semantic data.

Return type:: dict[str, object]

exception embodichain.lab.task_program.semantics.SemanticValidationError[source]

Raise one structured error at a static or live integration boundary.

Methods:

__init__(diagnostic)

__init__(diagnostic)[source]

class embodichain.lab.task_program.semantics.SkillPolicyPreset[source]

Policies and typed semantic-call option templates.

Methods:

`__init__`(preset_id, *, ...[, ...])	Own one policy bundle without exposing mutable nested configuration.
`action_option_template`(semantic_id)	Return one owned template for an exact semantic call ID.
`snapshot`()	Return an independently owned preset value.

Attributes:

`action_option_templates`	Return owned option templates keyed by exact semantic call ID.
`effect_monitors`	Return effect-monitor selections keyed by exact semantic call ID.
`motion_policy`	Return an independently owned motion policy.
`recovery_policy`	Return an independently owned recovery policy.
`required_planner`	Optional planner backend required by this preset.
`runner_cfg`	Return an independently owned runner configuration.
`tracking_policy`	Return independently owned endpoint-tracking settings.
`workflow_recovery_policy`	Return the bounded semantic-workflow recovery policy.

__init__(preset_id, *, action_option_templates, effect_assurance, effect_monitors=mappingproxy({}), motion_policy=None, tracking_policy=None, recovery_policy=None, workflow_recovery_policy=None, runner_cfg=None, required_planner=None)[source]: Own one policy bundle without exposing mutable nested configuration.

action_option_template(semantic_id)[source]

Return one owned template for an exact semantic call ID.

Raises:: KeyError – If this preset does not declare the semantic call.
Return type:: ActionOptions

property action_option_templates: Mapping[str, ActionOptions]: Return owned option templates keyed by exact semantic call ID.

property effect_monitors: Mapping[str, EffectMonitorRef]

Return effect-monitor selections keyed by exact semantic call ID.

Verified presets require a monitor for every curated effectful call they compile. Projected presets require this mapping to be empty.

property motion_policy: MotionPolicy: Return an independently owned motion policy.

property recovery_policy: RecoveryPolicy: Return an independently owned recovery policy.

required_planner: str | None: Optional planner backend required by this preset.

property runner_cfg: ExecutionRunnerCfg: Return an independently owned runner configuration.

snapshot()[source]

Return an independently owned preset value.

Return type:: SkillPolicyPreset

property tracking_policy: TrackingPolicy: Return independently owned endpoint-tracking settings.

property workflow_recovery_policy: WorkflowRecoveryPolicy: Return the bounded semantic-workflow recovery policy.

class embodichain.lab.task_program.semantics.SupportSurfaceAffordance[source]

Typed target frame for placing an object’s origin on a support surface.

The registered affordance pose is the desired object pose, expressed relative to its parent scene entity. The optional confidence threshold is enforced whenever that late-bound target pose is resolved.

Parameters:: minimum_confidence (float) – Minimum confidence accepted while resolving the late-bound target pose.

Methods:

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

__init__(object_label='', custom_config=<factory>, minimum_confidence=0.0)

class embodichain.lab.task_program.semantics.SymbolicStateDomain[source]

Typed mapping domains owned by TaskState.

Methods:

__new__(value)

__new__(value)

class embodichain.lab.task_program.semantics.SymbolicStateKey[source]

Provider-free key for one exact symbolic TaskState write.

The domain makes otherwise similar string and pair addresses impossible to conflate during static parallel analysis. This contract intentionally describes only exact keys; dynamic or opaque effects must not manufacture a guessed key.

Methods:

`__init__`(domain, address)
`articulation_joint`(articulation_id, joint_id)	Build one articulation-joint mapping key.
`coordinated_held_object`(...)	Build one ordered coordinated-held-object mapping key.
`held_object`(task_state_key)	Build one held-object mapping key.

Attributes:

rendered

Return a deterministic domain-qualified diagnostic form.

__init__(domain, address)

classmethod articulation_joint(articulation_id, joint_id)[source]

Build one articulation-joint mapping key.

Return type:: SymbolicStateKey

classmethod coordinated_held_object(first_task_state_key, second_task_state_key)[source]

Build one ordered coordinated-held-object mapping key.

Return type:: SymbolicStateKey

classmethod held_object(task_state_key)[source]

Build one held-object mapping key.

Return type:: SymbolicStateKey

property rendered: str: Return a deterministic domain-qualified diagnostic form.

exception embodichain.lab.task_program.semantics.UnsupportedSceneAffordanceError[source]: Raised when a parent has no affordance for a required capability.

exception embodichain.lab.task_program.semantics.UnsupportedSkillError[source]: Raised when no robot-resource assignment can satisfy a skill.

class embodichain.lab.task_program.semantics.WorkflowRecoveryPolicy[source]

Bound workflow-level recovery for curated semantic effect failures.

The atomic action remains the owner of replans and whole-action retries. This policy applies only after that action emits RECOVERY_REQUIRED and returns control to the Task Program workflow executor.

Parameters:: max_recovery_attempts (int) – Maximum recovery cycles for each environment row at one semantic-call boundary. Zero disables workflow recovery.

Methods:

`__init__`([max_recovery_attempts])
`snapshot`()	Return an independently owned immutable policy.

__init__(max_recovery_attempts=0)

snapshot()[source]

Return an independently owned immutable policy.

Return type:: WorkflowRecoveryPolicy

embodichain.lab.task_program.semantics.build_effect_evidence_queries(spec)[source]

Build one independently owned typed query per effect clause.

Parameters:: spec (SemanticEffectSpec) – Grounded semantic effect contract.
Return type:: tuple[PoseRelationEvidenceQuery | BinaryEffectEvidenceQuery | ScalarEffectEvidenceQuery | JointStateEvidenceQuery, ...]
Returns:: Queries in the contract’s deterministic clause order.

embodichain.lab.task_program.semantics.builtin_semantic_call_catalog()[source]

Build the curated catalog for installed manipulation primitives.

Return type:: SemanticCallCatalog
Returns:: A fresh immutable catalog. Atomic implementations remain uninstalled; callers bind them to an engine through the separate runtime path.

Semantic calls and catalog#

class embodichain.lab.task_program.semantics.SemanticCallSpec[source]#

Base value contract shared by every declarative semantic call.

Parameters:: resources (Mapping[str, str]) – Optional skill-local slot to robot-resource overrides.

Methods:

`__init__`(*[, resources])
`to_metadata`()	Return this semantic call as deterministic JSON-safe data.

Attributes:

Return the stable catalog identifier for this call.

__init__(*, resources=<factory>)#

property semantic_id: str#: Return the stable catalog identifier for this call.

to_metadata()[source]#

Return this semantic call as deterministic JSON-safe data.

Return type:: dict[str, object]

class embodichain.lab.task_program.semantics.SemanticPose[source]#

Object-space pose expressed as position and a WXYZ quaternion.

The value owns normalized tensor snapshots and never exposes its internal tensors directly. A single pose or an environment batch is accepted.

Parameters:

position (Tensor | tuple[float, float, float] | list[float]) – Shape (3,) or (B, 3).
quaternion_wxyz (Tensor | tuple[float, float, float, float] | list[float]) – Shape (4,) or (B, 4). Finite, non-zero quaternions are normalized at construction.

Methods:

`__init__`(position, quaternion_wxyz)
`snapshot`()	Return an independently owned pose value.
`to_matrix`()	Convert the semantic pose to a homogeneous transform.
`to_metadata`()	Return the pose as deterministic JSON-safe semantic data.

Attributes:

`batch_size`	Return the explicit batch size, or `None` for one broadcast pose.
`position`	Return an independent position tensor.
`quaternion_wxyz`	Return an independent normalized quaternion tensor.

__init__(position, quaternion_wxyz)[source]#

property batch_size: int | None#: Return the explicit batch size, or None for one broadcast pose.

property position: Tensor#: Return an independent position tensor.

property quaternion_wxyz: Tensor#: Return an independent normalized quaternion tensor.

snapshot()[source]#

Return an independently owned pose value.

Return type:: SemanticPose

to_matrix()[source]#

Convert the semantic pose to a homogeneous transform.

Return type:: Tensor
Returns:: Shape (4, 4) for an unbatched pose or (B, 4, 4) for a batched pose.

to_metadata()[source]#

Return the pose as deterministic JSON-safe semantic data.

Return type:: dict[str, object]

class embodichain.lab.task_program.semantics.Pick[source]#

Pick one registered object using an optional explicit grasp affordance.

Parameters:

object (SceneObjectRef) – Authoritative semantic object reference.
grasp (SceneAffordanceRef | None) – Optional explicit grasp affordance. Omission requests deterministic registry selection.
resources (Mapping[str, str]) – Optional skill-local resource overrides.

Methods:

__init__(object[, grasp, resources])

__init__(object, grasp=None, *, resources=<factory>)#

class embodichain.lab.task_program.semantics.Place[source]#

Place a held object at exactly one semantic destination.

Parameters:

object (SceneObjectRef) – Authoritative held-object reference.
at (SemanticPose | None) – Absolute object-space pose.
on (SceneObjectRef | SceneAffordanceRef | None) – Object or affordance supporting an on relation.
inside (SceneObjectRef | SceneAffordanceRef | None) – Object or affordance supporting an inside relation.
resources (Mapping[str, str]) – Optional skill-local resource overrides.

Methods:

__init__(object[, at, on, inside, resources])

__init__(object, at=None, on=None, inside=None, *, resources=<factory>)#

class embodichain.lab.task_program.semantics.HandOver[source]#

Pick up, transfer, and place an object with two robot resources.

Parameters:

object (SceneObjectRef) – Authoritative scene-object reference. The object must not already be held when the unified action starts.
final_target (SemanticPose | None) – Optional final object-space delivery pose.
resources (Mapping[str, str]) – Optional skill-local resource overrides. Select the second candidate with the destination slot when needed.

Methods:

__init__(object[, final_target, resources])

__init__(object, final_target=None, *, resources=<factory>)#

class embodichain.lab.task_program.semantics.RegisteredSemanticCall[source]#

Safe value payload for a catalog-registered semantic extension.

Parameters:

call_id (str) – Stable extension identifier discovered in a semantic catalog.
arguments (Mapping[str, None | bool | int | float | str | SceneEntityRef | SemanticPose | tuple[None | bool | int | float | str | SceneEntityRef | SemanticPose | tuple[DeclarativeValue, ...] | Mapping[str, DeclarativeValue], ...] | Mapping[str, None | bool | int | float | str | SceneEntityRef | SemanticPose | tuple[DeclarativeValue, ...] | Mapping[str, DeclarativeValue]]]) – Nested declarative data. Executable or live values are rejected at construction.
resources (Mapping[str, str]) – Optional skill-local resource overrides.

Methods:

__init__(call_id[, arguments, resources])

Attributes:

Return the registered extension identifier.

__init__(call_id, arguments=<factory>, *, resources=<factory>)#

property semantic_id: str#: Return the registered extension identifier.

class embodichain.lab.task_program.semantics.SemanticCallDescriptor[source]#

Static catalog metadata for one semantic call kind.

Parameters:

call_id (str) – Stable semantic call identifier.
spec_type (type[SemanticCallSpec]) – Exact public call value type.
target_descriptor (SkillDescriptor | None) – Exact atomic goal/options/resource contract. It is inferred and non-overridable for curated calls and required for registered extensions.

Methods:

__init__(call_id, spec_type[, target_descriptor])

Attributes:

`binding_contract`	Return the resource contract from the canonical target descriptor.
`skill_id`	Return the atomic skill ID from the canonical target descriptor.

__init__(call_id, spec_type, target_descriptor=None)#

property binding_contract: SkillBindingContract#: Return the resource contract from the canonical target descriptor.

property skill_id: str#: Return the atomic skill ID from the canonical target descriptor.

class embodichain.lab.task_program.semantics.SemanticCallCatalog[source]#

Immutable discovery catalog separated from engine installation.

Methods:

`__init__`(descriptors)
`discover`(call)	Discover metadata without installing or executing an implementation.
`with_descriptor`(descriptor)	Return a new catalog containing one additional descriptor.

Attributes:

descriptors

Return immutable descriptors keyed by exact semantic ID.

__init__(descriptors)[source]#

property descriptors: Mapping[str, SemanticCallDescriptor]#: Return immutable descriptors keyed by exact semantic ID.

discover(call)[source]#

Discover metadata without installing or executing an implementation.

Parameters:

call (str | SemanticCallSpec) – Exact semantic ID or a call value.

Return type:

SemanticCallDescriptor

Returns:

Matching immutable descriptor.

Raises:

KeyError – If the exact call ID is unknown.
TypeError – If the call type disagrees with its descriptor.

with_descriptor(descriptor)[source]#

Return a new catalog containing one additional descriptor.

Return type:: SemanticCallCatalog

embodichain.lab.task_program.semantics.builtin_semantic_call_catalog()[source]#

Build the curated catalog for installed manipulation primitives.

Return type:: SemanticCallCatalog
Returns:: A fresh immutable catalog. Atomic implementations remain uninstalled; callers bind them to an engine through the separate runtime path.

Semantic integration#

class embodichain.lab.task_program.semantics.SceneEntityManifest[source]#

Provider-free static scene declaration using canonical registry metadata.

Methods:

`__init__`(ref[, aliases, parent, ...])
`from_metadata`(metadata)	Copy one canonical provider-free registry metadata value.

__init__(ref, aliases=(), parent=None, native_name=None, dynamics=SceneDynamics.UNKNOWN, collision_role=SceneCollisionRole.NONE, semantic_type=None, affordance_capabilities=frozenset({}), default_affordances=<factory>, affordance_payload_type=None, affordance_revision=None, relative_pose=None)#

classmethod from_metadata(metadata)[source]#

Copy one canonical provider-free registry metadata value.

Return type:: SceneEntityManifest

class embodichain.lab.task_program.semantics.SceneManifest[source]#

Immutable provider-free scene catalog used before simulation starts.

Methods:

`__init__`([entries, collision_world_mode])
`from_registry`(registry)	Project a live registry without observing any dynamic provider.
`lookup`(identifier, *[, expected_type, path])	Return one static entry after canonical typed resolution.
`resolve`(identifier, *[, expected_type, path])	Resolve one canonical or alias reference with pathful diagnostics.
`resolve_affordance`(parent, *, capability[, ...])	Resolve one affordance using the same strict rule as SceneRegistry.
`validate_registry`(registry, *[, path])	Require a live registry to match this provider-free declaration.

Attributes:

entries

Return immutable provider-free entries in declaration order.

__init__(entries=(), *, collision_world_mode=None)[source]#

property entries: tuple[SceneEntityManifest, ...]#: Return immutable provider-free entries in declaration order.

classmethod from_registry(registry)[source]#

Project a live registry without observing any dynamic provider.

Return type:: SceneManifest

lookup(identifier, *, expected_type=<class 'embodichain.lab.task_program.semantics.scene.SceneEntityRef'>, path=())[source]#

Return one static entry after canonical typed resolution.

Return type:: SceneEntityManifest

resolve(identifier, *, expected_type=<class 'embodichain.lab.task_program.semantics.scene.SceneEntityRef'>, path=())[source]#

Resolve one canonical or alias reference with pathful diagnostics.

Return type:: TypeVar(RefT, bound= SceneEntityRef)

resolve_affordance(parent, *, capability, explicit=None, path=())[source]#

Resolve one affordance using the same strict rule as SceneRegistry.

Return type:: SceneAffordanceRef

validate_registry(registry, *, path=('integration', 'scene_registry'))[source]#

Require a live registry to match this provider-free declaration.

Return type:: None

class embodichain.lab.task_program.semantics.SemanticIntegrationManifest[source]#

Static scene/profile/catalog declaration validated before execution.

Parameters:

scene (SceneManifest) – Provider-free scene manifest.
robot_profile (RobotSkillProfile) – Declarative robot resource/profile snapshot.
call_catalog (SemanticCallCatalog) – Discoverable semantic call descriptors.
runtime_preset (str | None) – Optional integration-wide policy preset override.

Methods:

`__init__`(scene, robot_profile, call_catalog)
`bind`(scene_registry, engine, *[, ...])	Validate live scene and robot bindings without observing or planning.
`link_call`(call, *[, path])	Resolve static refs, affordances, and declared resource structure.

__init__(scene, robot_profile, call_catalog, runtime_preset=None)#

bind(scene_registry, engine, *, endpoint_adapters=None)[source]#

Validate live scene and robot bindings without observing or planning.

Return type:: BoundSemanticIntegration

link_call(call, *, path=('call',))[source]#

Resolve static refs, affordances, and declared resource structure.

This method never observes scene providers, constructs an engine, samples a grasp, or runs a planner.

Return type:: LinkedSemanticCall

class embodichain.lab.task_program.semantics.SemanticDiagnostic[source]#

Structured deterministic semantic-integration diagnostic.

Parameters:

code (str) – Stable machine-readable failure code.
path (tuple[str | int, ...]) – Complete configuration or program path.
message (str) – Human-readable explanation.
candidates (tuple[str, ...]) – Canonical candidate IDs, sorted when applicable.

Methods:

__init__(code, path, message[, candidates])

Attributes:

rendered_path

Return the path in dotted/indexed notation.

__init__(code, path, message, candidates=())#

property rendered_path: str#: Return the path in dotted/indexed notation.

class embodichain.lab.task_program.semantics.SemanticValidationError[source]#

Raise one structured error at a static or live integration boundary.

Methods:

__init__(diagnostic)

__init__(diagnostic)[source]#

Robot resources and profiles#

class embodichain.lab.task_program.semantics.RobotSkillProfile[source]#

Reusable declarative skill integration for one robot embodiment.

Methods:

`__init__`(profile_id, resources[, ...])
`action_control_profiles`()	Lower endpoint command profiles for the current action core.
`bind`(engine, *[, endpoint_adapters])	Validate this profile against one fully configured action engine.

Attributes:

grounding_providers

Semantic call ID to embodiment-owned named grounding provider ID.

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

action_control_profiles()[source]#

Lower endpoint command profiles for the current action core.

Return type:: Mapping[str, ControlPartCommandProfile]
Returns:: Owned command profiles keyed by concrete robot control-part name.
Raises:: ValueError – If two endpoint declarations assign non-equivalent commands with the same semantic name to one control part.

bind(engine, *, endpoint_adapters=None)[source]#

Validate this profile against one fully configured action engine.

Parameters:

engine (AtomicActionEngine) – Installed atomic-action engine for the target robot.
endpoint_adapters (Optional[Mapping[type[ResourceEndpoint], ResourceEndpointAdapter]]) – Optional exact endpoint-type adapters. Explicit entries extend the non-overridable built-in control-part adapter.

Return type:

BoundRobotSkillProfile

Returns:

Robot-, engine-, and adapter-validated profile view.

grounding_providers: Mapping[str, str]#: Semantic call ID to embodiment-owned named grounding provider ID.

class embodichain.lab.task_program.semantics.BoundRobotSkillProfile[source]#

Robot- and engine-validated view of a RobotSkillProfile.

Methods:

`__init__`(profile, engine, *[, endpoint_adapters])
`assert_current`()	Reject this binding after the engine's skill catalog changes.
`candidates`(skill_id[, selections])	Return every valid complete resource assignment deterministically.
`preset`([preset_id, skill_id])	Resolve an explicit, per-skill, or profile-default policy preset.
`resolve`(skill_id[, selections])	Resolve one skill with strict capability matching and disambiguation.

Attributes:

`engine`	Return the exact action engine used to validate this profile.
`profile_id`	Return the stable profile identifier.
`resources`	Return resolved generic robot resources keyed by logical ID.
`skills`	Return installed Atomic Skills fully supported by this profile.
`source_profile`	Return the immutable profile object used to create this binding.

__init__(profile, engine, *, endpoint_adapters=None)[source]#

assert_current()[source]#

Reject this binding after the engine’s skill catalog changes.

Raises:: RuntimeError – If actions were registered or replaced after binding.
Return type:: None

candidates(skill_id, selections=None)[source]#

Return every valid complete resource assignment deterministically.

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

property engine: AtomicActionEngine#: Return the exact action engine used to validate this profile.

preset(preset_id=None, *, skill_id=None)[source]#

Resolve an explicit, per-skill, or profile-default policy preset.

Return type:: SkillPolicyPreset

property profile_id: str#: Return the stable profile identifier.

resolve(skill_id, selections=None)[source]#

Resolve one skill with strict capability matching and disambiguation.

Return type:: ResolvedSkillBinding

property resources: Mapping[str, ResolvedRobotResource]#: Return resolved generic robot resources keyed by logical ID.

property skills: Mapping[str, SkillDescriptor]#: Return installed Atomic Skills fully supported by this profile.

property source_profile: RobotSkillProfile#: Return the immutable profile object used to create this binding.

class embodichain.lab.task_program.semantics.RobotResource[source]#

Generic leaf or composite resource in one robot’s resource DAG.

A resource may expose any number of named endpoints. For example, one manipulation participant may expose motion and grasp endpoints, while a mobile base or whole-body controller may expose only motion. members describes physical claim composition and does not inherit endpoint capabilities.

Methods:

`__init__`(resource_id[, endpoints, members])
`snapshot`()	Return an independently owned resource declaration.

__init__(resource_id, endpoints=<factory>, members=())#

snapshot()[source]#

Return an independently owned resource declaration.

Return type:: RobotResource

class embodichain.lab.task_program.semantics.ResourceEndpoint[source]#

Extensible execution endpoint in a robot resource graph.

Endpoint subclasses add controller-specific addressing data. Capabilities stay on this common base so skill matching does not depend on any one controller kind.

Methods:

`__init__`(*[, capabilities])
`snapshot`()	Return an independently owned endpoint declaration.

Attributes:

Open, namespaced capabilities provided by this exact endpoint.

__init__(*, capabilities=frozenset({}))#

capabilities: frozenset[str]#: Open, namespaced capabilities provided by this exact endpoint.

snapshot()[source]#

Return an independently owned endpoint declaration.

Endpoint subclasses with payloads that cannot be deep-copied must override this method and return a new value of their exact type.

Return type:: ResourceEndpoint

class embodichain.lab.task_program.semantics.ResourceEndpointAdapter[source]#

Resolve one endpoint kind without coupling profiles to its controller.

Attributes:

`adapter_id`	Stable adapter identifier used in diagnostics and resolved metadata.
`effect_evidence_source_keys`	Exact `(provider_id, revision)` effect-evidence routes emitted.
`endpoint_type`	Exact endpoint declaration type accepted by this adapter.
`runtime_target_types`	Exact immutable runtime-target value types this adapter may resolve.
`runtime_transport_ids`	Exact endpoint-command transport IDs this adapter may resolve.
`tracking_feedback_source_keys`	Exact `(provider_id, revision)` tracking-feedback routes emitted.
`tracking_projector_keys`	Exact `(projector_id, revision)` desired-state routes emitted.

Methods:

resolve(endpoint, *, engine)

Validate and resolve one endpoint against an action engine.

adapter_id: ClassVar[str]#: Stable adapter identifier used in diagnostics and resolved metadata.

effect_evidence_source_keys: ClassVar[frozenset[tuple[str, str]]]#: Exact (provider_id, revision) effect-evidence routes emitted.

endpoint_type: ClassVar[type[ResourceEndpoint]]#: Exact endpoint declaration type accepted by this adapter.

abstract resolve(endpoint, *, engine)[source]#

Validate and resolve one endpoint against an action engine.

Parameters:

endpoint (ResourceEndpoint) – Endpoint declaration of endpoint_type.
engine (AtomicActionEngine) – Engine whose robot, planner, and command profiles are bound.

Return type:

EndpointResolution

Returns:

Physical claims and supported lowering metadata.

runtime_target_types: ClassVar[tuple[type[RuntimeEndpointTarget], ...]]#: Exact immutable runtime-target value types this adapter may resolve.

runtime_transport_ids: ClassVar[frozenset[str]]#: Exact endpoint-command transport IDs this adapter may resolve.

tracking_feedback_source_keys: ClassVar[frozenset[tuple[str, str]]]#: Exact (provider_id, revision) tracking-feedback routes emitted.

tracking_projector_keys: ClassVar[frozenset[tuple[str, str]]]#: Exact (projector_id, revision) desired-state routes emitted.

class embodichain.lab.task_program.semantics.EndpointResolution[source]#

Adapter-produced runtime destination and claim metadata for one endpoint.

Methods:

__init__(runtime_target[, task_state_key, ...])

Attributes:

`claim_tokens`	Adapter-defined physical/controller claims beyond robot joint IDs.
`command_profile_key`	Profile key that owns semantic commands for this endpoint, when any.
`effect_sources`	Provider-routed raw observation sources keyed by open channel ID.
`exclusive`	Whether this execution endpoint must declare a physical claim.
`joint_ids`	Ordered robot joint IDs controlled by the endpoint, when applicable.
`requires_command_profile`	Whether a missing `command_profile_key` entry invalidates binding.
`runtime_target`	Typed immutable destination consumed by an endpoint command transport.
`task_state_key`	Optional symbolic state key; profile binding defaults to its resource ID.
`tracking_channels`	Typed feedback source and desired-state projector by channel ID.

__init__(runtime_target, task_state_key=None, effect_sources=<factory>, tracking_channels=<factory>, command_profile_key=None, requires_command_profile=False, claim_tokens=frozenset({}), joint_ids=(), exclusive=True)#

claim_tokens: frozenset[str]#: Adapter-defined physical/controller claims beyond robot joint IDs.

command_profile_key: str | None#: Profile key that owns semantic commands for this endpoint, when any.

effect_sources: Mapping[str, EffectEvidenceSourceRef]#: Provider-routed raw observation sources keyed by open channel ID.

exclusive: bool#: Whether this execution endpoint must declare a physical claim.

joint_ids: tuple[int, ...]#: Ordered robot joint IDs controlled by the endpoint, when applicable.

requires_command_profile: bool#: Whether a missing command_profile_key entry invalidates binding.

runtime_target: RuntimeEndpointTarget#: Typed immutable destination consumed by an endpoint command transport.

task_state_key: str | None#: Optional symbolic state key; profile binding defaults to its resource ID.

tracking_channels: Mapping[str, EndpointTrackingChannelBinding]#: Typed feedback source and desired-state projector by channel ID.

class embodichain.lab.task_program.semantics.ControlPartEndpoint[source]#

One named execution endpoint backed by a robot control part.

Capabilities are explicit and never inferred from the endpoint name, joint count, other endpoints, or composite resource members.

Methods:

__init__(control_part[, command_profile, ...])

Attributes:

`command_profile`	Optional generic command-profile ID; defaults to `control_part`.
`control_part`	Key from the bound robot's `control_parts` mapping.

__init__(control_part, command_profile=None, *, capabilities=frozenset({}))#

command_profile: str | None#: Optional generic command-profile ID; defaults to control_part.

control_part: str#: Key from the bound robot’s control_parts mapping.

class embodichain.lab.task_program.semantics.ControlPartEndpointAdapter[source]#

Resolve joint-backed ControlPartEndpoint declarations.

Attributes:

`adapter_id`	Stable adapter identifier used in diagnostics and resolved metadata.
`effect_evidence_source_keys`	Exact `(provider_id, revision)` effect-evidence routes emitted.
`runtime_target_types`	Exact immutable runtime-target value types this adapter may resolve.
`runtime_transport_ids`	Exact endpoint-command transport IDs this adapter may resolve.
`tracking_feedback_source_keys`	Exact `(provider_id, revision)` tracking-feedback routes emitted.
`tracking_projector_keys`	Exact `(projector_id, revision)` desired-state routes emitted.

Classes:

endpoint_type

alias of ControlPartEndpoint

Methods:

resolve(endpoint, *, engine)

Resolve a robot control part and verify its standard capabilities.

adapter_id: ClassVar[str] = 'control_part'#: Stable adapter identifier used in diagnostics and resolved metadata.

effect_evidence_source_keys: ClassVar[frozenset[tuple[str, str]]] = frozenset({('builtin.control_part', '1')})#: Exact (provider_id, revision) effect-evidence routes emitted.

endpoint_type#: alias of ControlPartEndpoint

resolve(endpoint, *, engine)[source]#

Resolve a robot control part and verify its standard capabilities.

Return type:: EndpointResolution

runtime_target_types: ClassVar[tuple[type[RuntimeEndpointTarget], ...]] = (<class 'embodichain.lab.sim.atomic_actions.bindings.JointPositionTarget'>,)#: Exact immutable runtime-target value types this adapter may resolve.

runtime_transport_ids: ClassVar[frozenset[str]] = frozenset({'robot.joint_position'})#: Exact endpoint-command transport IDs this adapter may resolve.

tracking_feedback_source_keys: ClassVar[frozenset[tuple[str, str]]] = frozenset({('planning_context.robot', '1')})#: Exact (provider_id, revision) tracking-feedback routes emitted.

tracking_projector_keys: ClassVar[frozenset[tuple[str, str]]] = frozenset({('joint_position_payload', '1')})#: Exact (projector_id, revision) desired-state routes emitted.

class embodichain.lab.task_program.semantics.ResourceBinding[source]#

Generic mapping from skill-local slots to robot resource IDs.

Methods:

__init__(resources)

__init__(resources)#

class embodichain.lab.task_program.semantics.ResolvedResourceEndpoint[source]#

Endpoint declaration resolved by one registered adapter.

Methods:

`__init__`(endpoint, adapter_id, runtime_target)
`conflicts_with`(other)	Return whether two endpoints address overlapping physical channels.

Attributes:

SceneEntityRegistration

Return capabilities declared by the source endpoint.

__init__(endpoint, adapter_id, runtime_target, task_state_key=None, effect_sources=<factory>, tracking_channels=<factory>, command_profile_key=None, requires_command_profile=False, commands=<factory>, claim_tokens=frozenset({}), joint_ids=(), exclusive=True)#

property capabilities: frozenset[str]#: Return capabilities declared by the source endpoint.

conflicts_with(other)[source]#

Return whether two endpoints address overlapping physical channels.

Return type:: bool

class embodichain.lab.task_program.semantics.ResolvedRobotResource[source]#

Robot-validated resource with concrete endpoint joint IDs and claim.

Methods:

__init__(resource_id, endpoints, members, claim)

Attributes:

endpoint_joint_ids

Return ordered joint IDs for each resolved endpoint.

__init__(resource_id, endpoints, members, claim)#

property endpoint_joint_ids: Mapping[str, tuple[int, ...]]#: Return ordered joint IDs for each resolved endpoint.

class embodichain.lab.task_program.semantics.ResolvedSkillBinding[source]#

One generic resource assignment lowered for the current action core.

Methods:

__init__(skill_id, resources, ...)

Attributes:

resource_ids

Return the selected logical resource ID for each skill-local slot.

__init__(skill_id, resources, action_binding, claim)#

property resource_ids: Mapping[str, str]#: Return the selected logical resource ID for each skill-local slot.

class embodichain.lab.task_program.semantics.ResourceClaim[source]#

Physical leaf and joint claim used for deterministic conflict checks.

Methods:

`__init__`(leaf_resource_ids, joint_ids[, ...])
`combine`(claims)	Return the union of zero or more resource claims.
`conflicts_with`(other)	Return whether two claims overlap in a leaf or concrete joint.

__init__(leaf_resource_ids, joint_ids, claim_tokens=frozenset({}))#

classmethod combine(claims)[source]#

Return the union of zero or more resource claims.

Return type:: ResourceClaim

conflicts_with(other)[source]#

Return whether two claims overlap in a leaf or concrete joint.

Return type:: bool

class embodichain.lab.task_program.semantics.SkillPolicyPreset[source]#

Policies and typed semantic-call option templates.

Methods:

`__init__`(preset_id, *, ...[, ...])	Own one policy bundle without exposing mutable nested configuration.
`action_option_template`(semantic_id)	Return one owned template for an exact semantic call ID.
`snapshot`()	Return an independently owned preset value.

Attributes:

`action_option_templates`	Return owned option templates keyed by exact semantic call ID.
`effect_monitors`	Return effect-monitor selections keyed by exact semantic call ID.
`motion_policy`	Return an independently owned motion policy.
`recovery_policy`	Return an independently owned recovery policy.
`required_planner`	Optional planner backend required by this preset.
`runner_cfg`	Return an independently owned runner configuration.
`tracking_policy`	Return independently owned endpoint-tracking settings.
`workflow_recovery_policy`	Return the bounded semantic-workflow recovery policy.

__init__(preset_id, *, action_option_templates, effect_assurance, effect_monitors=mappingproxy({}), motion_policy=None, tracking_policy=None, recovery_policy=None, workflow_recovery_policy=None, runner_cfg=None, required_planner=None)[source]#: Own one policy bundle without exposing mutable nested configuration.

action_option_template(semantic_id)[source]#

Return one owned template for an exact semantic call ID.

Raises:: KeyError – If this preset does not declare the semantic call.
Return type:: ActionOptions

property action_option_templates: Mapping[str, ActionOptions]#: Return owned option templates keyed by exact semantic call ID.

property effect_monitors: Mapping[str, EffectMonitorRef]#

Return effect-monitor selections keyed by exact semantic call ID.

Verified presets require a monitor for every curated effectful call they compile. Projected presets require this mapping to be empty.

property motion_policy: MotionPolicy#: Return an independently owned motion policy.

property recovery_policy: RecoveryPolicy#: Return an independently owned recovery policy.

required_planner: str | None#: Optional planner backend required by this preset.

property runner_cfg: ExecutionRunnerCfg#: Return an independently owned runner configuration.

snapshot()[source]#

Return an independently owned preset value.

Return type:: SkillPolicyPreset

property tracking_policy: TrackingPolicy#: Return independently owned endpoint-tracking settings.

property workflow_recovery_policy: WorkflowRecoveryPolicy#: Return the bounded semantic-workflow recovery policy.

class embodichain.lab.task_program.semantics.EffectAssurance[source]#

Authority used to advance semantic state after command completion.

Methods:

__new__(value)

__new__(value)#

Profile errors#

class embodichain.lab.task_program.semantics.ProfileValidationError[source]#

Raised when a robot skill profile disagrees with its engine or robot.

Methods:

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

__init__(*args, **kwargs)#

__new__(**kwargs)#

class embodichain.lab.task_program.semantics.UnsupportedSkillError[source]#

Raised when no robot-resource assignment can satisfy a skill.

Methods:

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

__init__(*args, **kwargs)#

__new__(**kwargs)#

class embodichain.lab.task_program.semantics.AmbiguousSkillBindingError[source]#

Raised when multiple assignments remain without a complete default.

Methods:

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

__init__(*args, **kwargs)#

__new__(**kwargs)#

Effects and evidence#

class embodichain.lab.task_program.semantics.SemanticEffectSpec[source]#

Grounded typed physical clauses and symbolic postconditions for one call.

Methods:

`__init__`(semantic_id, effect_kind, skill_id, ...)
`snapshot`()	Return an independently owned grounded effect contract.
`state_expectation`(expectation_id)	Return an owned state expectation by effect-local ID.
`to_metadata`()	Return this grounded effect contract as deterministic JSON-safe data.
`validate_request`(request)	Validate execution identity and typed symbolic postconditions.

__init__(semantic_id, effect_kind, skill_id, invocation_id, invocation_revision, env_ids, state_expectations, clauses)#

snapshot()[source]#

Return an independently owned grounded effect contract.

Return type:: SemanticEffectSpec

state_expectation(expectation_id)[source]#

Return an owned state expectation by effect-local ID.

Return type:: HeldObjectStateExpectation | CoordinatedHeldObjectCleanupExpectation | ArticulationJointStateExpectation

to_metadata()[source]#

Return this grounded effect contract as deterministic JSON-safe data.

Return type:: dict[str, object]

validate_request(request)[source]#

Validate execution identity and typed symbolic postconditions.

Return type:: None

class embodichain.lab.task_program.semantics.EffectMonitorRef[source]#

Versioned, declarative reference to an effect-monitor factory.

Methods:

`__init__`(monitor_id, revision[, params])
`snapshot`()	Return an independently owned declarative reference.
`to_metadata`()	Return a deterministic JSON-safe monitor selection.

__init__(monitor_id, revision, params=<factory>)#

snapshot()[source]#

Return an independently owned declarative reference.

Return type:: EffectMonitorRef

to_metadata()[source]#

Return a deterministic JSON-safe monitor selection.

Return type:: dict[str, object]

class embodichain.lab.task_program.semantics.EffectMonitor[source]#

Stateful verifier owned by one grounded semantic call.

Methods:

observe(request, evidence)

Consume one synchronized raw observation and decide requested rows.

Attributes:

`resolved_params`	Return all resolved monitor thresholds for trace metadata.
`spec`	Return an independently owned effect contract.

abstract observe(request, evidence)[source]#

Consume one synchronized raw observation and decide requested rows.

Return type:: EffectMonitorDecision

abstract property resolved_params: Mapping[str, None | bool | int | float | str | tuple[None | bool | int | float | str | tuple[EffectMonitorParam, ...] | Mapping[str, EffectMonitorParam], ...] | Mapping[str, None | bool | int | float | str | tuple[EffectMonitorParam, ...] | Mapping[str, EffectMonitorParam]]]#: Return all resolved monitor thresholds for trace metadata.

abstract property spec: SemanticEffectSpec#: Return an independently owned effect contract.

class embodichain.lab.task_program.semantics.EffectEvidenceCollector[source]#

Dispatch and normalize a synchronized observation for one effect spec.

Methods:

`__init__`(registry)
`collect`(spec, *, timestamp, observation_revision)	Acquire and strictly synchronize evidence for every effect clause.

Attributes:

registry

Return the immutable provider registry.

__init__(registry)[source]#

collect(spec, *, timestamp, observation_revision, env_ids=None)[source]#

Acquire and strictly synchronize evidence for every effect clause.

Parameters:

spec (SemanticEffectSpec) – Grounded semantic effect contract.
timestamp (float) – Backend observation time for this acquisition tick.
observation_revision (int) – Runtime-owned observation revision.
env_ids (Tensor | None) – Optional ordered subset of spec.env_ids. Acquisition failures must remain present as rows with valid=False.

Return type:

Mapping[str, PoseRelationEvidenceBatch | BinaryEffectEvidenceBatch | ScalarEffectEvidenceBatch | JointStateEvidenceBatch]

Returns:

Immutable mapping keyed exactly by effect clause ID.

property registry: EffectEvidenceProviderRegistry#: Return the immutable provider registry.

Registry and provider#

class embodichain.lab.task_program.semantics.SceneRegistry[source]#

Immutable authoritative catalog of semantic scene entities.

Canonical identifiers occupy one flat, globally unique namespace. Aliases are accepted only at lookup and integration boundaries and always resolve to a canonical typed reference before they leave the registry.

Parameters:

registrations (Iterable[SceneEntityRegistration]) – Complete scene registrations. The iterable is copied and cannot be extended after construction.
collision_world_mode (SceneCollisionWorldMode | None) – Explicit dynamic-collision batch policy. It may be omitted for a single environment, which resolves to shared. A multi-environment dynamic world must select a mode explicitly.

Methods:

`__init__`([registrations, collision_world_mode])
`affordances`(parent, *, capability)	Return compatible direct-child affordances without selecting one.
`collision_geometry_by_id`([role])	Materialize planner geometry under canonical registry IDs.
`from_simulation`(simulation, *[, ...])	Opt explicitly selected simulation entities into a registry.
`lookup`(identifier, *[, expected_type])	Return an owned registration after canonical typed resolution.
`make_planning_scene_provider`(...[, ...])	Create a provider after complete planner/registry validation.
`make_scene_provider`(*[, ...])	Create an independent provider without planner cross-validation.
`object_semantics`(object_ref, *, affordance)	Build one owned atomic-action semantic snapshot.
`resolve`(identifier, *[, expected_type])	Resolve a canonical ID or alias to a typed canonical reference.
`resolve_affordance`(parent, *, capability[, ...])	Select one compatible affordance with strict scoped-default rules.
`resolve_collision_world_mode`(*, batch_size)	Resolve the configured collision mode for an execution batch.
`validate_collision_integration`(...[, ...])	Validate registry/planner agreement before dynamic planning.

Attributes:

`aliases`	Return the immutable alias-to-canonical-ID index.
`collision_world_entity_ids`	Return every canonical ID represented in the planner world.
`dynamic_collision_entity_ids`	Return canonical IDs whose planner poses update dynamically.
`entity_metadata`	Return provider-free metadata without copying affordance payloads.
`entity_refs`	Return canonical typed references in registration order.
`registrations`	Return structurally independent registration values.
`static_collision_entity_ids`	Return canonical IDs baked into the static planner world.

__init__(registrations=(), *, collision_world_mode=None)[source]#

affordances(parent, *, capability)[source]#

Return compatible direct-child affordances without selecting one.

Parameters:

parent (str | SceneEntityRef) – Canonical ID, alias, or typed parent reference.
capability (str) – Required open affordance capability.

Return type:

tuple[SceneAffordanceRef, ...]

Returns:

Compatible canonical references sorted by canonical ID.

property aliases: Mapping[str, str]#: Return the immutable alias-to-canonical-ID index.

collision_geometry_by_id(role=None)[source]#

Materialize planner geometry under canonical registry IDs.

Parameters:: role (SceneCollisionRole | None) – Optional exact collision-role filter. Without a filter, all static and dynamic collision registrations are included. Registrations whose role is SceneCollisionRole.NONE never enter the planner collision world.
Return type:: Mapping[str, object]
Returns:: Fresh immutable canonical-ID-to-geometry mapping.

property collision_world_entity_ids: tuple[str, ...]#: Return every canonical ID represented in the planner world.

property dynamic_collision_entity_ids: tuple[str, ...]#: Return canonical IDs whose planner poses update dynamically.

property entity_metadata: tuple[SceneEntityMetadata, ...]#: Return provider-free metadata without copying affordance payloads.

property entity_refs: tuple[SceneEntityRef, ...]#: Return canonical typed references in registration order.

classmethod from_simulation(simulation, *, rigid_objects=None, articulations=None, collision_roles=None, geometry_providers=None, collision_world_mode=None)[source]#

Opt explicitly selected simulation entities into a registry.

rigid_objects and articulations map authoritative registry IDs to simulation UIDs. UIDs become aliases automatically; unlisted simulation entities are never imported. Collision participation defaults to SceneCollisionRole.NONE.

Parameters:

simulation (SimulationManager) – Simulation manager used only for explicit UID lookup.
rigid_objects (Mapping[str, str] | None) – Canonical object IDs mapped to simulation UIDs.
articulations (Mapping[str, str] | None) – Canonical articulation IDs mapped to simulation UIDs.
collision_roles (Mapping[str, SceneCollisionRole] | None) – Optional collision roles keyed by canonical ID.
geometry_providers (Mapping[str, SceneGeometryProvider] | None) – Optional geometry overrides keyed by canonical ID. Selected rigid objects otherwise expose their live handles.
collision_world_mode (SceneCollisionWorldMode | None) – Optional dynamic collision batch-sharing mode.

Return type:

SceneRegistry

Returns:

Immutable registry containing only the explicitly selected entities.

lookup(identifier, *, expected_type=<class 'embodichain.lab.task_program.semantics.scene.SceneEntityRef'>)[source]#

Return an owned registration after canonical typed resolution.

Parameters:

identifier (str | SceneEntityRef) – Canonical ID, alias, or typed canonical reference.
expected_type (type[TypeVar(RefT, bound= SceneEntityRef)]) – Required reference class.

Return type:

Returns:

A structurally independent copy of the matching registration.

make_planning_scene_provider(motion_generator, *, batch_size, translation_threshold=0.0001, rotation_threshold=0.001)[source]#

Create a provider after complete planner/registry validation.

Parameters:

motion_generator (MotionGenerator) – Motion generator that will consume dynamic poses.
batch_size (int) – Number of execution environments.
translation_threshold (float) – Accumulated translation needed to publish a material scene change.
rotation_threshold (float) – Accumulated rotation needed to publish a material scene change.

Return type:

Returns:

A new independently stateful, planner-validated scene provider.

make_scene_provider(*, translation_threshold=0.0001, rotation_threshold=0.001, batch_size=None)[source]#

Create an independent provider without planner cross-validation.

This factory is intended for perception and direct-core consumers. The canonical planning path must use make_planning_scene_provider() so planner IDs, capabilities, and collision-world mode cannot drift.

Parameters:

translation_threshold (float) – Accumulated translation needed to publish a material scene change.
rotation_threshold (float) – Accumulated rotation needed to publish a material scene change.
batch_size (int | None) – Optional fixed integration batch size. Supplying it validates the collision-world mode immediately and binds the provider to that row count.

Return type: