embodichain.lab.task_program.compiler

embodichain.lab.task_program.compiler#

Task Program compilation and Semantic Call lowering.

The compiler expands a validated AST into immutable segments and calls. Semantic lowering is provider-independent; live planning starts only after an environment integration is assembled.

`CompiledArticulationJointPositionValidator`	Owned joint-position validator with its canonical articulation.
`CompiledBarrier`	Explicit join semantics for one compiled parallel block.
`CompiledObjectNearTargetValidator`	Owned validator config with canonical object and resolved target pose.
`CompiledParallelBlock`	Two or more call lanes joined by an explicit deterministic barrier.
`CompiledParallelBranch`	One ordered semantic-call lane inside a parallel block.
`CompiledPostPolicy`	Owned post-policy config plus its canonical scene entity and source path.
`CompiledRepeatFrame`	One lexical repeat occurrence in a compiled call or segment path.
`CompiledTargetSelection`	Deterministic cyclic-target selection metadata for one occurrence.
`CompiledTaskProgram`	Immutable overall instruction and bounded segments for task execution.
`CompiledTaskProgramAnalysis`	One owned canonical semantic-analysis window for a compiled program.
`CompiledTaskProgramCall`	One owned semantic call occurrence emitted by lazy program expansion.
`CompiledTaskProgramSegment`	One independent explicit or implicit logical program segment.
`CompiledTaskProgramValidator`	Represent a PEP 604 union type
`TaskProgramCompileError`	Raised when a validated AST cannot lower to canonical semantic calls.
`TaskProgramCompiler`	Compile validated Task Program ASTs through one static scene manifest.

Classes:

`CompiledArticulationJointPositionValidator`	Owned joint-position validator with its canonical articulation.
`CompiledBarrier`	Explicit join semantics for one compiled parallel block.
`CompiledObjectNearTargetValidator`	Owned validator config with canonical object and resolved target pose.
`CompiledObjectStateValidator`	Owned measured lift, relative position, or upright object contract.
`CompiledParallelBlock`	Two or more call lanes joined by an explicit deterministic barrier.
`CompiledParallelBranch`	One ordered semantic-call lane inside a parallel block.
`CompiledPostPolicy`	Owned post-policy config plus its canonical scene entity and source path.
`CompiledRepeatFrame`	One lexical repeat occurrence in a compiled call or segment path.
`CompiledTargetSelection`	Deterministic cyclic-target selection metadata for one occurrence.
`CompiledTaskProgram`	Immutable overall instruction and bounded segments for task execution.
`CompiledTaskProgramAnalysis`	One owned canonical semantic-analysis window for a compiled program.
`CompiledTaskProgramCall`	One owned semantic call occurrence emitted by lazy program expansion.
`CompiledTaskProgramSegment`	One independent explicit or implicit logical program segment.
`TaskProgramCompiler`	Compile validated Task Program ASTs through one static scene manifest.

Exceptions:

TaskProgramCompileError

Raised when a validated AST cannot lower to canonical semantic calls.

class embodichain.lab.task_program.compiler.CompiledArticulationJointPositionValidator[source]#

Owned joint-position validator with its canonical articulation.

Methods:

__init__(cfg, articulation, source_path)

Attributes:

`articulation`
`cfg`
`source_path`

__init__(cfg, articulation, source_path)#

articulation: SceneArticulationRef#

cfg: ArticulationJointPositionValidatorCfg | _ArticulationJointPositionValues#

source_path: tuple[str | int, ...]#

class embodichain.lab.task_program.compiler.CompiledBarrier[source]#

Explicit join semantics for one compiled parallel block.

Methods:

__init__(name, timeout_steps, ...)

Attributes:

`failure_policy`
`name`
`source_path`
`timeout_steps`

__init__(name, timeout_steps, failure_policy, source_path)#

failure_policy: str#

name: str#

source_path: tuple[str | int, ...]#

timeout_steps: int#

class embodichain.lab.task_program.compiler.CompiledObjectNearTargetValidator[source]#

Owned validator config with canonical object and resolved target pose.

Methods:

__init__(cfg, object, target_pose, ...)

Attributes:

`cfg`
`object`
`source_path`
`target_pose`
`target_selection`

__init__(cfg, object, target_pose, target_selection, source_path)#

cfg: ObjectNearTargetValidatorCfg | _ObjectNearTargetValues#

object: SceneObjectRef#

source_path: tuple[str | int, ...]#

target_pose: SemanticPose#

target_selection: CompiledTargetSelection#

class embodichain.lab.task_program.compiler.CompiledObjectStateValidator[source]#

Owned measured lift, relative position, or upright object contract.

Parameters:

cfg (ObjectLiftedValidatorCfg | ObjectRelativePositionValidatorCfg | ObjectUprightValidatorCfg | _ObjectStateValues) – Typed measured-object validator configuration to snapshot.
object (SceneObjectRef) – Canonical object reference resolved during static compilation.
source_path (tuple[str | int, ...]) – Validator location in the original program configuration.
reference (SceneObjectRef | None) – Canonical reference object, required only for relative position validation.

Methods:

__init__(cfg, object, source_path[, reference])

Attributes:

`cfg`
`object`
`reference`
`source_path`

__init__(cfg, object, source_path, reference=None)#

cfg: ObjectLiftedValidatorCfg | ObjectRelativePositionValidatorCfg | ObjectUprightValidatorCfg | _ObjectStateValues#

object: SceneObjectRef#

reference: SceneObjectRef | None#

source_path: tuple[str | int, ...]#

class embodichain.lab.task_program.compiler.CompiledParallelBlock[source]#

Two or more call lanes joined by an explicit deterministic barrier.

Methods:

__init__(branches, barrier, source_path)

Attributes:

`barrier`
`branches`
`source_path`

__init__(branches, barrier, source_path)#

barrier: CompiledBarrier#

branches: tuple[CompiledParallelBranch, ...]#

source_path: tuple[str | int, ...]#

class embodichain.lab.task_program.compiler.CompiledParallelBranch[source]#

One ordered semantic-call lane inside a parallel block.

Methods:

__init__(branch_index, calls, source_path)

Attributes:

`branch_index`
`calls`
`source_path`

__init__(branch_index, calls, source_path)#

branch_index: int#

calls: tuple[CompiledTaskProgramCall, ...]#

source_path: tuple[str | int, ...]#

class embodichain.lab.task_program.compiler.CompiledPostPolicy[source]#

Owned post-policy config plus its canonical scene entity and source path.

Methods:

__init__(cfg, entity, source_path)

Attributes:

`cfg`
`entity`
`source_path`

__init__(cfg, entity, source_path)#

cfg: WaitStablePostCfg | _WaitStableValues#

entity: SceneEntityRef#

source_path: tuple[str | int, ...]#

class embodichain.lab.task_program.compiler.CompiledRepeatFrame[source]#

One lexical repeat occurrence in a compiled call or segment path.

Methods:

__init__(path, iteration_index, count)

Attributes:

`count`
`iteration_index`
`path`

__init__(path, iteration_index, count)#

count: int#

iteration_index: int#

path: tuple[str | int, ...]#

class embodichain.lab.task_program.compiler.CompiledTargetSelection[source]#

Deterministic cyclic-target selection metadata for one occurrence.

Methods:

__init__(target_id, value_index, ...)

Attributes:

`repeat_iteration_index`
`repeat_path`
`target_id`
`value_index`

__init__(target_id, value_index, repeat_path, repeat_iteration_index)#

repeat_iteration_index: int | None#

repeat_path: tuple[str | int, ...] | None#

target_id: str#

value_index: int#

class embodichain.lab.task_program.compiler.CompiledTaskProgram[source]#

Immutable overall instruction and bounded segments for task execution.

Methods:

`__init__`(args, *kwargs)	Reject construction outside `TaskProgramCompiler`.
`iter_segments`()	Iterate the already materialized provider-free segments.
`preflight_analyses`()	Return full-program analyses split only at parallel barriers.
`sequential_execution_analysis`(segment_index)	Return current-segment prefix plus downstream sequential look-ahead.

Attributes:

`instruction`
`integration`	Return an independent integration-selection snapshot.
`program_id`
`segment_count`	Return the number of materialized logical segments.

__init__(*args, **kwargs)[source]#: Reject construction outside TaskProgramCompiler.

instruction: str | None#

property integration: TaskProgramIntegrationCfg#: Return an independent integration-selection snapshot.

iter_segments()[source]#

Iterate the already materialized provider-free segments.

Return type:: Iterator[CompiledTaskProgramSegment]

preflight_analyses()[source]#

Return full-program analyses split only at parallel barriers.

Consecutive sequential segments form one static workflow, preserving their object-state flow and cross-segment target look-ahead. Each parallel branch is analyzed independently; no state or target inference crosses the barrier in either direction.

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

program_id: str#

property segment_count: int#: Return the number of materialized logical segments.

sequential_execution_analysis(segment_index)[source]#

Return current-segment prefix plus downstream sequential look-ahead.

Parameters:

segment_index (int) – Index of the sequential segment about to execute.

Return type:

CompiledTaskProgramAnalysis

Returns:

Analysis beginning at the selected segment and ending immediately before the next parallel barrier or the end of the program.

Raises:

IndexError – If segment_index is outside this program.
ValueError – If the selected segment is itself parallel.

class embodichain.lab.task_program.compiler.CompiledTaskProgramAnalysis[source]#

One owned canonical semantic-analysis window for a compiled program.

execution_prefix_length separates calls that the current segment owns from downstream calls included only for static state-flow and target look-ahead. Preflight analyses set the prefix to the complete window.

Methods:

__init__(analysis_id, kind, calls, ...)

Attributes:

`analysis_id`
`calls`
`execution_prefix_length`
`kind`
`segment_indices`
`source_path`

__init__(analysis_id, kind, calls, source_path, segment_indices, execution_prefix_length)#

analysis_id: str#

calls: tuple[SemanticCallSpec, ...]#

execution_prefix_length: int#

kind: str#

segment_indices: tuple[int, ...]#

source_path: tuple[str | int, ...]#

class embodichain.lab.task_program.compiler.CompiledTaskProgramCall[source]#

One owned semantic call occurrence emitted by lazy program expansion.

Methods:

__init__(call_index, segment_call_index, ...)

Attributes:

`call`
`call_index`
`repeat_frames`
`segment_call_index`
`source_path`
`target_selections`

__init__(call_index, segment_call_index, call, source_path, repeat_frames=(), target_selections=())#

call: SemanticCallSpec#

call_index: int#

repeat_frames: tuple[CompiledRepeatFrame, ...]#

segment_call_index: int#

source_path: tuple[str | int, ...]#

target_selections: tuple[CompiledTargetSelection, ...]#

class embodichain.lab.task_program.compiler.CompiledTaskProgramSegment[source]#

One independent explicit or implicit logical program segment.

Methods:

__init__(segment_index, segment_id, name, ...)

Attributes:

`calls`
`implicit`
`instruction`
`name`
`parallel_block`
`post_policies`
`repeat_frames`
`segment_id`
`segment_index`
`source_path`
`validators`

__init__(segment_index, segment_id, name, calls, source_path, repeat_frames=(), post_policies=(), validators=(), parallel_block=None, implicit=False, instruction=None)#

calls: tuple[CompiledTaskProgramCall, ...]#

implicit: bool#

instruction: str | None#

name: str#

parallel_block: CompiledParallelBlock | None#

post_policies: tuple[CompiledPostPolicy, ...]#

repeat_frames: tuple[CompiledRepeatFrame, ...]#

segment_id: str#

segment_index: int#

source_path: tuple[str | int, ...]#

validators: tuple[CompiledObjectNearTargetValidator | CompiledArticulationJointPositionValidator | CompiledObjectStateValidator, ...]#

exception embodichain.lab.task_program.compiler.TaskProgramCompileError[source]#: Raised when a validated AST cannot lower to canonical semantic calls.

class embodichain.lab.task_program.compiler.TaskProgramCompiler[source]#

Compile validated Task Program ASTs through one static scene manifest.

Methods:

`__init__`(scene_manifest)	Create one provider-free compiler.
`compile`(config)	Compile one validated AST into a bounded provider-free program.
`from_scene_registry`(registry)	Create a compiler from a provider-free SceneRegistry identity snapshot.

__init__(scene_manifest)[source]#

Create one provider-free compiler.

Parameters:: scene_manifest (SceneManifest) – Canonical provider-free scene identity catalog.

compile(config)[source]#

Compile one validated AST into a bounded provider-free program.

Parameters:: config (TaskProgramCfg) – Strict, supported-version Task Program configuration.
Return type:: CompiledTaskProgram
Returns:: Immutable segments with repeat-local targets already resolved.
Raises:: TaskProgramCompileError – If typed scene resolution or AST lowering fails.

classmethod from_scene_registry(registry)[source]#

Create a compiler from a provider-free SceneRegistry identity snapshot.

Return type:: TaskProgramCompiler

Validator type aliases#

These aliases describe the compiled validator variants.

embodichain.lab.task_program.compiler.CompiledTaskProgramValidator = embodichain.lab.task_program.compiler.program.CompiledObjectNearTargetValidator | embodichain.lab.task_program.compiler.program.CompiledArticulationJointPositionValidator | embodichain.lab.task_program.compiler.program.CompiledObjectStateValidator#

Represent a PEP 604 union type

E.g. for int | str

embodichain.lab.task_program.compiler

Contents

embodichain.lab.task_program.compiler#

Validator type aliases#