# ----------------------------------------------------------------------------
# Copyright (c) 2021-2026 DexForce Technology Co., Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ----------------------------------------------------------------------------
"""Provider-free compilation and lazy expansion of Task Program ASTs."""
from __future__ import annotations
from collections.abc import Iterator, Mapping
from dataclasses import dataclass, field
from types import MappingProxyType
from typing import TypeAlias
from embodichain.lab.task_program.semantics.calls import (
DeclarativeValue,
HandOver,
Pick,
Place,
RegisteredSemanticCall,
SemanticCallSpec,
SemanticPose,
)
from embodichain.lab.task_program.semantics.integration import SceneManifest
from embodichain.lab.task_program.semantics.scene import (
SceneAffordanceRef,
SceneArticulationRef,
SceneEntityRef,
SceneLinkRef,
SceneObjectRef,
SceneRegistry,
)
from ..language.schema import (
MAX_EXPANDED_CALLS,
MAX_REPEAT_COUNT,
ArticulationJointPositionValidatorCfg,
BarrierCfg,
CyclicPoseTargetCfg,
TaskProgramCfg,
TaskProgramIntegrationCfg,
HandOverCfg,
InvokeCfg,
ObjectNearTargetValidatorCfg,
ParallelCfg,
PickCfg,
PlaceCfg,
PoseCfg,
ProgramNodeCfg,
RegisteredSemanticCallCfg,
RepeatCfg,
SegmentCfg,
SequenceCfg,
TargetRefCfg,
WaitStablePostCfg,
)
from ..language.decoder import ConfigPath, TaskProgramConfigError, render_config_path
_SEMANTIC_CALL_TYPES = (
Pick,
Place,
HandOver,
RegisteredSemanticCall,
)
_SCENE_REF_TYPES = (
SceneEntityRef,
SceneObjectRef,
SceneArticulationRef,
SceneLinkRef,
SceneAffordanceRef,
)
[docs]
class TaskProgramCompileError(TaskProgramConfigError):
"""Raised when a validated AST cannot lower to canonical semantic calls."""
def _copy_scene_ref(reference: SceneEntityRef) -> SceneEntityRef:
"""Return one independent exact typed scene reference."""
if type(reference) not in _SCENE_REF_TYPES:
raise TypeError(f"Unsupported scene reference {type(reference).__name__}.")
return type(reference)(reference.entity_id)
[docs]
@dataclass(frozen=True, slots=True)
class CompiledRepeatFrame:
"""One lexical repeat occurrence in a compiled call or segment path."""
path: ConfigPath
iteration_index: int
count: int
def __post_init__(self) -> None:
if type(self.path) is not tuple:
raise TypeError("path must be a ConfigPath tuple.")
if type(self.iteration_index) is not int or not 0 <= self.iteration_index:
raise ValueError("iteration_index must be a non-negative integer.")
if type(self.count) is not int or self.count <= 0:
raise ValueError("count must be a positive integer.")
if self.iteration_index >= self.count:
raise ValueError("iteration_index must be smaller than count.")
[docs]
@dataclass(frozen=True, slots=True)
class CompiledTargetSelection:
"""Deterministic cyclic-target selection metadata for one occurrence."""
target_id: str
value_index: int
repeat_path: ConfigPath | None
repeat_iteration_index: int | None
def __post_init__(self) -> None:
if type(self.target_id) is not str or not self.target_id:
raise ValueError("target_id must be a non-empty string.")
if type(self.value_index) is not int or self.value_index < 0:
raise ValueError("value_index must be a non-negative integer.")
if (self.repeat_path is None) != (self.repeat_iteration_index is None):
raise ValueError(
"repeat_path and repeat_iteration_index must both be set or unset."
)
if self.repeat_path is not None and type(self.repeat_path) is not tuple:
raise TypeError("repeat_path must be a ConfigPath tuple or None.")
if self.repeat_iteration_index is not None and (
type(self.repeat_iteration_index) is not int
or self.repeat_iteration_index < 0
):
raise ValueError("repeat_iteration_index must be non-negative or None.")
def _snapshot_semantic_call(call: SemanticCallSpec) -> SemanticCallSpec:
"""Return one independently owned exact semantic-call value."""
if type(call) is Pick:
return Pick(
object=_copy_scene_ref(call.object),
grasp=(None if call.grasp is None else _copy_scene_ref(call.grasp)),
resources=dict(call.resources),
)
if type(call) is Place:
return Place(
object=_copy_scene_ref(call.object),
at=None if call.at is None else call.at.snapshot(),
on=None if call.on is None else _copy_scene_ref(call.on),
inside=None if call.inside is None else _copy_scene_ref(call.inside),
resources=dict(call.resources),
)
if type(call) is HandOver:
return HandOver(
object=_copy_scene_ref(call.object),
final_target=(
None if call.final_target is None else call.final_target.snapshot()
),
resources=dict(call.resources),
)
if type(call) is RegisteredSemanticCall:
return RegisteredSemanticCall(
call_id=call.call_id,
arguments=call.arguments,
resources=dict(call.resources),
)
raise TypeError("call must be an exact supported SemanticCallSpec value.")
[docs]
@dataclass(frozen=True, slots=True)
class CompiledTaskProgramCall:
"""One owned semantic call occurrence emitted by lazy program expansion."""
call_index: int
segment_call_index: int
call: SemanticCallSpec
source_path: ConfigPath
repeat_frames: tuple[CompiledRepeatFrame, ...] = ()
target_selections: tuple[CompiledTargetSelection, ...] = ()
def __post_init__(self) -> None:
if type(self.call_index) is not int or self.call_index < 0:
raise ValueError("call_index must be a non-negative integer.")
if type(self.segment_call_index) is not int or self.segment_call_index < 0:
raise ValueError("segment_call_index must be a non-negative integer.")
if type(self.call) not in _SEMANTIC_CALL_TYPES:
raise TypeError("call must be an exact supported SemanticCallSpec value.")
if type(self.source_path) is not tuple:
raise TypeError("source_path must be a ConfigPath tuple.")
frames = tuple(self.repeat_frames)
selections = tuple(self.target_selections)
if not all(type(frame) is CompiledRepeatFrame for frame in frames):
raise TypeError("repeat_frames must contain CompiledRepeatFrame values.")
if not all(
type(selection) is CompiledTargetSelection for selection in selections
):
raise TypeError(
"target_selections must contain CompiledTargetSelection values."
)
object.__setattr__(self, "call", _snapshot_semantic_call(self.call))
object.__setattr__(self, "repeat_frames", frames)
object.__setattr__(self, "target_selections", selections)
[docs]
@dataclass(frozen=True, slots=True)
class CompiledPostPolicy:
"""Owned post-policy config plus its canonical scene entity and source path."""
cfg: WaitStablePostCfg
entity: SceneEntityRef
source_path: ConfigPath
def __post_init__(self) -> None:
if type(self.cfg) is not WaitStablePostCfg:
raise TypeError("cfg must be exactly WaitStablePostCfg.")
if type(self.entity) not in _SCENE_REF_TYPES:
raise TypeError("entity must be an exact SceneEntityRef value.")
if type(self.source_path) is not tuple:
raise TypeError("source_path must be a ConfigPath tuple.")
object.__setattr__(
self,
"cfg",
WaitStablePostCfg(
entity=self.cfg.entity,
preset=self.cfg.preset,
kind=self.cfg.kind,
),
)
object.__setattr__(self, "entity", _copy_scene_ref(self.entity))
[docs]
@dataclass(frozen=True, slots=True)
class CompiledObjectNearTargetValidator:
"""Owned validator config with canonical object and resolved target pose."""
cfg: ObjectNearTargetValidatorCfg
object: SceneObjectRef
target_pose: SemanticPose
target_selection: CompiledTargetSelection
source_path: ConfigPath
def __post_init__(self) -> None:
if type(self.cfg) is not ObjectNearTargetValidatorCfg:
raise TypeError("cfg must be exactly ObjectNearTargetValidatorCfg.")
if type(self.object) is not SceneObjectRef:
raise TypeError("object must be exactly SceneObjectRef.")
if type(self.target_pose) is not SemanticPose:
raise TypeError("target_pose must be exactly SemanticPose.")
if type(self.target_selection) is not CompiledTargetSelection:
raise TypeError("target_selection must be exactly CompiledTargetSelection.")
if type(self.source_path) is not tuple:
raise TypeError("source_path must be a ConfigPath tuple.")
object.__setattr__(
self,
"cfg",
ObjectNearTargetValidatorCfg(
object=self.cfg.object,
target=self.cfg.target,
position_tolerance=self.cfg.position_tolerance,
kind=self.cfg.kind,
),
)
object.__setattr__(self, "object", _copy_scene_ref(self.object))
object.__setattr__(self, "target_pose", self.target_pose.snapshot())
[docs]
@dataclass(frozen=True, slots=True)
class CompiledArticulationJointPositionValidator:
"""Owned joint-position validator with its canonical articulation."""
cfg: ArticulationJointPositionValidatorCfg
articulation: SceneArticulationRef
source_path: ConfigPath
def __post_init__(self) -> None:
if type(self.cfg) is not ArticulationJointPositionValidatorCfg:
raise TypeError(
"cfg must be exactly ArticulationJointPositionValidatorCfg."
)
if type(self.articulation) is not SceneArticulationRef:
raise TypeError("articulation must be exactly SceneArticulationRef.")
if type(self.source_path) is not tuple:
raise TypeError("source_path must be a ConfigPath tuple.")
object.__setattr__(
self,
"cfg",
ArticulationJointPositionValidatorCfg(
articulation=self.cfg.articulation,
joint=self.cfg.joint,
minimum_position=self.cfg.minimum_position,
maximum_position=self.cfg.maximum_position,
kind=self.cfg.kind,
),
)
object.__setattr__(
self,
"articulation",
_copy_scene_ref(self.articulation),
)
CompiledTaskProgramValidator: TypeAlias = (
CompiledObjectNearTargetValidator | CompiledArticulationJointPositionValidator
)
_COMPILED_VALIDATOR_TYPES = (
CompiledObjectNearTargetValidator,
CompiledArticulationJointPositionValidator,
)
[docs]
@dataclass(frozen=True, slots=True)
class CompiledBarrier:
"""Explicit join semantics for one compiled parallel block."""
name: str
timeout_steps: int
failure_policy: str
source_path: ConfigPath
def __post_init__(self) -> None:
if type(self.name) is not str or not self.name:
raise ValueError("barrier name must be non-empty.")
if type(self.timeout_steps) is not int or self.timeout_steps <= 0:
raise ValueError("barrier timeout_steps must be positive.")
if self.failure_policy != "fail_fast":
raise ValueError("barrier failure_policy must be 'fail_fast'.")
if type(self.source_path) is not tuple:
raise TypeError("barrier source_path must be a ConfigPath tuple.")
[docs]
@dataclass(frozen=True, slots=True)
class CompiledParallelBranch:
"""One ordered semantic-call lane inside a parallel block."""
branch_index: int
calls: tuple[CompiledTaskProgramCall, ...]
source_path: ConfigPath
def __post_init__(self) -> None:
if type(self.branch_index) is not int or self.branch_index < 0:
raise ValueError("branch_index must be non-negative.")
calls = tuple(self.calls)
if not calls or not all(
type(call) is CompiledTaskProgramCall for call in calls
):
raise TypeError("parallel branch calls must be non-empty compiled calls.")
if type(self.source_path) is not tuple:
raise TypeError("parallel branch source_path must be a ConfigPath tuple.")
object.__setattr__(self, "calls", calls)
[docs]
@dataclass(frozen=True, slots=True)
class CompiledParallelBlock:
"""Two or more call lanes joined by an explicit deterministic barrier."""
branches: tuple[CompiledParallelBranch, ...]
barrier: CompiledBarrier
source_path: ConfigPath
def __post_init__(self) -> None:
branches = tuple(self.branches)
if len(branches) < 2 or not all(
type(branch) is CompiledParallelBranch for branch in branches
):
raise TypeError("parallel blocks require at least two compiled branches.")
if tuple(branch.branch_index for branch in branches) != tuple(
range(len(branches))
):
raise ValueError("parallel branch indices must be contiguous from zero.")
if type(self.barrier) is not CompiledBarrier:
raise TypeError("barrier must be exactly CompiledBarrier.")
if type(self.source_path) is not tuple:
raise TypeError("parallel source_path must be a ConfigPath tuple.")
object.__setattr__(self, "branches", branches)
[docs]
@dataclass(frozen=True, slots=True)
class CompiledTaskProgramSegment:
"""One independent explicit or implicit logical program segment."""
segment_index: int
segment_id: str
name: str
calls: tuple[CompiledTaskProgramCall, ...]
source_path: ConfigPath
repeat_frames: tuple[CompiledRepeatFrame, ...] = ()
post_policies: tuple[CompiledPostPolicy, ...] = ()
validators: tuple[CompiledTaskProgramValidator, ...] = ()
parallel_block: CompiledParallelBlock | None = None
implicit: bool = False
def __post_init__(self) -> None:
if type(self.segment_index) is not int or self.segment_index < 0:
raise ValueError("segment_index must be a non-negative integer.")
for field_name in ("segment_id", "name"):
value = getattr(self, field_name)
if type(value) is not str or not value:
raise ValueError(f"{field_name} must be a non-empty string.")
calls = tuple(self.calls)
if not calls or not all(
type(call) is CompiledTaskProgramCall for call in calls
):
raise TypeError(
"calls must contain at least one exact CompiledTaskProgramCall."
)
if tuple(call.segment_call_index for call in calls) != tuple(range(len(calls))):
raise ValueError("segment call indices must be contiguous from zero.")
if type(self.source_path) is not tuple:
raise TypeError("source_path must be a ConfigPath tuple.")
frames = tuple(self.repeat_frames)
post = tuple(self.post_policies)
validators = tuple(self.validators)
if not all(type(frame) is CompiledRepeatFrame for frame in frames):
raise TypeError("repeat_frames must contain CompiledRepeatFrame values.")
if not all(type(value) is CompiledPostPolicy for value in post):
raise TypeError("post_policies must contain CompiledPostPolicy values.")
if not all(type(value) in _COMPILED_VALIDATOR_TYPES for value in validators):
raise TypeError("validators must contain compiled validator values.")
if type(self.implicit) is not bool:
raise TypeError("implicit must be a bool.")
if self.implicit and (post or validators):
raise ValueError(
"Implicit segments cannot own post-policies or validators."
)
if self.parallel_block is not None:
if type(self.parallel_block) is not CompiledParallelBlock:
raise TypeError("parallel_block must be CompiledParallelBlock or None.")
flattened = tuple(
call for branch in self.parallel_block.branches for call in branch.calls
)
if flattened != calls:
raise ValueError(
"segment calls must equal parallel branch calls in branch order."
)
object.__setattr__(self, "calls", calls)
object.__setattr__(self, "repeat_frames", frames)
object.__setattr__(self, "post_policies", post)
object.__setattr__(self, "validators", validators)
[docs]
@dataclass(frozen=True, slots=True)
class CompiledTaskProgramAnalysis:
"""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.
"""
analysis_id: str
kind: str
calls: tuple[SemanticCallSpec, ...]
source_path: ConfigPath
segment_indices: tuple[int, ...]
execution_prefix_length: int
def __post_init__(self) -> None:
if type(self.analysis_id) is not str or not self.analysis_id:
raise ValueError("analysis_id must be a non-empty string.")
if self.kind not in {
"sequential_stretch",
"parallel_branch",
"sequential_suffix",
}:
raise ValueError("kind must identify a supported program analysis.")
calls = tuple(self.calls)
if not calls or not all(type(call) in _SEMANTIC_CALL_TYPES for call in calls):
raise TypeError("calls must contain supported semantic call values.")
if type(self.source_path) is not tuple:
raise TypeError("source_path must be a ConfigPath tuple.")
indices = tuple(self.segment_indices)
if not indices or any(type(index) is not int or index < 0 for index in indices):
raise ValueError("segment_indices must contain non-negative integers.")
if len(set(indices)) != len(indices) or tuple(sorted(indices)) != indices:
raise ValueError("segment_indices must be unique and ordered.")
if type(
self.execution_prefix_length
) is not int or not 1 <= self.execution_prefix_length <= len(calls):
raise ValueError(
"execution_prefix_length must select a non-empty prefix of calls."
)
object.__setattr__(
self,
"calls",
tuple(_snapshot_semantic_call(call) for call in calls),
)
object.__setattr__(self, "segment_indices", indices)
@dataclass(frozen=True, slots=True)
class _CallTemplate:
kind: str
source_path: ConfigPath
object: SceneObjectRef | None = None
grasp: SceneAffordanceRef | None = None
at_target_id: str | None = None
on: SceneObjectRef | SceneAffordanceRef | None = None
inside: SceneObjectRef | SceneAffordanceRef | None = None
final_target_id: str | None = None
call_id: str | None = None
arguments: Mapping[str, DeclarativeValue] | None = None
resources: tuple[tuple[str, str], ...] = ()
@dataclass(frozen=True, slots=True)
class _InvokeTemplate:
call: _CallTemplate
source_path: ConfigPath
@dataclass(frozen=True, slots=True)
class _SequenceTemplate:
items: tuple[_NodeTemplate, ...]
source_path: ConfigPath
@dataclass(frozen=True, slots=True)
class _RepeatTemplate:
count: int
body: _NodeTemplate
source_path: ConfigPath
@dataclass(frozen=True, slots=True)
class _BarrierTemplate:
name: str
timeout_steps: int
failure_policy: str
source_path: ConfigPath
@dataclass(frozen=True, slots=True)
class _ParallelTemplate:
branches: tuple[_NodeTemplate, ...]
barrier: _BarrierTemplate
source_path: ConfigPath
@dataclass(frozen=True, slots=True)
class _PostTemplate:
cfg: WaitStablePostCfg
entity: SceneEntityRef
source_path: ConfigPath
@dataclass(frozen=True, slots=True)
class _ObjectNearTargetValidatorTemplate:
cfg: ObjectNearTargetValidatorCfg
object: SceneObjectRef
target_id: str
source_path: ConfigPath
@dataclass(frozen=True, slots=True)
class _ArticulationJointPositionValidatorTemplate:
cfg: ArticulationJointPositionValidatorCfg
articulation: SceneArticulationRef
source_path: ConfigPath
_ValidatorTemplate: TypeAlias = (
_ObjectNearTargetValidatorTemplate | _ArticulationJointPositionValidatorTemplate
)
@dataclass(frozen=True, slots=True)
class _SegmentTemplate:
name: str
steps: _NodeTemplate
post: tuple[_PostTemplate, ...]
validators: tuple[_ValidatorTemplate, ...]
source_path: ConfigPath
_NodeTemplate = (
_InvokeTemplate
| _SequenceTemplate
| _RepeatTemplate
| _SegmentTemplate
| _ParallelTemplate
)
def _contains_parallel(template: _NodeTemplate) -> bool:
"""Return whether a compiled subtree owns a parallel block."""
if type(template) is _ParallelTemplate:
return True
if type(template) is _SequenceTemplate:
return any(_contains_parallel(child) for child in template.items)
if type(template) is _RepeatTemplate:
return _contains_parallel(template.body)
if type(template) is _SegmentTemplate:
return _contains_parallel(template.steps)
return False
@dataclass(slots=True)
class _ExpansionState:
segment_index: int = 0
call_index: int = 0
def _resolve_target(
target_id: str,
*,
targets: Mapping[str, tuple[SemanticPose, ...]],
repeat_frames: tuple[CompiledRepeatFrame, ...],
) -> tuple[SemanticPose, CompiledTargetSelection]:
"""Select one cyclic target from the nearest lexical repeat frame."""
values = targets[target_id]
repeat = repeat_frames[-1] if repeat_frames else None
value_index = 0 if repeat is None else repeat.iteration_index % len(values)
selection = CompiledTargetSelection(
target_id=target_id,
value_index=value_index,
repeat_path=None if repeat is None else repeat.path,
repeat_iteration_index=None if repeat is None else repeat.iteration_index,
)
return values[value_index].snapshot(), selection
def _instantiate_call(
template: _CallTemplate,
*,
call_index: int,
segment_call_index: int,
targets: Mapping[str, tuple[SemanticPose, ...]],
repeat_frames: tuple[CompiledRepeatFrame, ...],
) -> CompiledTaskProgramCall:
"""Instantiate one semantic call occurrence from static templates."""
resources = dict(template.resources)
selections: list[CompiledTargetSelection] = []
if template.kind == "pick":
assert template.object is not None
call: SemanticCallSpec = Pick(
object=_copy_scene_ref(template.object),
grasp=(None if template.grasp is None else _copy_scene_ref(template.grasp)),
resources=resources,
)
elif template.kind == "place":
assert template.object is not None
at: SemanticPose | None = None
if template.at_target_id is not None:
at, selection = _resolve_target(
template.at_target_id,
targets=targets,
repeat_frames=repeat_frames,
)
selections.append(selection)
call = Place(
object=_copy_scene_ref(template.object),
at=at,
on=None if template.on is None else _copy_scene_ref(template.on),
inside=(
None if template.inside is None else _copy_scene_ref(template.inside)
),
resources=resources,
)
elif template.kind == "hand_over":
assert template.object is not None
final_target: SemanticPose | None = None
if template.final_target_id is not None:
final_target, selection = _resolve_target(
template.final_target_id,
targets=targets,
repeat_frames=repeat_frames,
)
selections.append(selection)
call = HandOver(
object=_copy_scene_ref(template.object),
final_target=final_target,
resources=resources,
)
elif template.kind == "registered":
assert template.call_id is not None and template.arguments is not None
call = RegisteredSemanticCall(
call_id=template.call_id,
arguments=template.arguments,
resources=resources,
)
else: # pragma: no cover - compiler-owned templates prevent this
raise AssertionError(f"Unknown call template {template.kind!r}.")
return CompiledTaskProgramCall(
call_index=call_index,
segment_call_index=segment_call_index,
call=call,
source_path=template.source_path,
repeat_frames=repeat_frames,
target_selections=tuple(selections),
)
def _iter_call_templates(
template: _NodeTemplate,
*,
repeat_frames: tuple[CompiledRepeatFrame, ...],
) -> Iterator[tuple[_CallTemplate, tuple[CompiledRepeatFrame, ...]]]:
"""Expand call templates inside one explicit segment without segment splits."""
if type(template) is _InvokeTemplate:
yield template.call, repeat_frames
elif type(template) is _SequenceTemplate:
for child in template.items:
yield from _iter_call_templates(child, repeat_frames=repeat_frames)
elif type(template) is _RepeatTemplate:
for iteration_index in range(template.count):
frame = CompiledRepeatFrame(
path=template.source_path,
iteration_index=iteration_index,
count=template.count,
)
yield from _iter_call_templates(
template.body,
repeat_frames=(*repeat_frames, frame),
)
else: # pragma: no cover - nested segments are rejected during compilation
raise AssertionError("A nested segment reached call-only expansion.")
def _instantiate_parallel_block(
template: _ParallelTemplate,
*,
targets: Mapping[str, tuple[SemanticPose, ...]],
repeat_frames: tuple[CompiledRepeatFrame, ...],
state: _ExpansionState,
) -> tuple[CompiledParallelBlock, tuple[CompiledTaskProgramCall, ...]]:
"""Instantiate branch-local call order without serializing branch semantics."""
branches: list[CompiledParallelBranch] = []
flattened: list[CompiledTaskProgramCall] = []
segment_call_index = 0
for branch_index, branch_template in enumerate(template.branches):
calls: list[CompiledTaskProgramCall] = []
for call_template, call_repeat_frames in _iter_call_templates(
branch_template,
repeat_frames=repeat_frames,
):
call = _instantiate_call(
call_template,
call_index=state.call_index,
segment_call_index=segment_call_index,
targets=targets,
repeat_frames=call_repeat_frames,
)
calls.append(call)
flattened.append(call)
state.call_index += 1
segment_call_index += 1
branches.append(
CompiledParallelBranch(
branch_index=branch_index,
calls=tuple(calls),
source_path=template.branches[branch_index].source_path,
)
)
barrier = CompiledBarrier(
name=template.barrier.name,
timeout_steps=template.barrier.timeout_steps,
failure_policy=template.barrier.failure_policy,
source_path=template.barrier.source_path,
)
return (
CompiledParallelBlock(
branches=tuple(branches),
barrier=barrier,
source_path=template.source_path,
),
tuple(flattened),
)
def _segment_identity(
program_id: str,
*,
source_path: ConfigPath,
repeat_frames: tuple[CompiledRepeatFrame, ...],
implicit: bool,
) -> str:
"""Build one deterministic segment identity from lexical occurrence data."""
repeat_suffix = "".join(
f"@{render_config_path(frame.path)}[{frame.iteration_index}]"
for frame in repeat_frames
)
boundary = "implicit" if implicit else "segment"
return f"{program_id}:{boundary}:{render_config_path(source_path)}{repeat_suffix}"
def _iter_segments(
template: _NodeTemplate,
*,
program_id: str,
targets: Mapping[str, tuple[SemanticPose, ...]],
repeat_frames: tuple[CompiledRepeatFrame, ...],
state: _ExpansionState,
) -> Iterator[CompiledTaskProgramSegment]:
"""Lazily expand outer program structure into independent segments."""
if type(template) is _SequenceTemplate:
for child in template.items:
yield from _iter_segments(
child,
program_id=program_id,
targets=targets,
repeat_frames=repeat_frames,
state=state,
)
return
if type(template) is _RepeatTemplate:
for iteration_index in range(template.count):
frame = CompiledRepeatFrame(
path=template.source_path,
iteration_index=iteration_index,
count=template.count,
)
yield from _iter_segments(
template.body,
program_id=program_id,
targets=targets,
repeat_frames=(*repeat_frames, frame),
state=state,
)
return
if type(template) is _InvokeTemplate:
call = _instantiate_call(
template.call,
call_index=state.call_index,
segment_call_index=0,
targets=targets,
repeat_frames=repeat_frames,
)
state.call_index += 1
segment = CompiledTaskProgramSegment(
segment_index=state.segment_index,
segment_id=_segment_identity(
program_id,
source_path=template.source_path,
repeat_frames=repeat_frames,
implicit=True,
),
name=f"invoke:{call.call.semantic_id}",
calls=(call,),
source_path=template.source_path,
repeat_frames=repeat_frames,
implicit=True,
)
state.segment_index += 1
yield segment
return
if type(template) is _ParallelTemplate:
parallel_block, calls = _instantiate_parallel_block(
template,
targets=targets,
repeat_frames=repeat_frames,
state=state,
)
segment = CompiledTaskProgramSegment(
segment_index=state.segment_index,
segment_id=_segment_identity(
program_id,
source_path=template.source_path,
repeat_frames=repeat_frames,
implicit=True,
),
name=f"parallel:{parallel_block.barrier.name}",
calls=calls,
source_path=template.source_path,
repeat_frames=repeat_frames,
parallel_block=parallel_block,
implicit=True,
)
state.segment_index += 1
yield segment
return
assert type(template) is _SegmentTemplate
parallel_block: CompiledParallelBlock | None = None
if type(template.steps) is _ParallelTemplate:
parallel_block, instantiated_calls = _instantiate_parallel_block(
template.steps,
targets=targets,
repeat_frames=repeat_frames,
state=state,
)
calls = list(instantiated_calls)
else:
calls = []
for segment_call_index, (call_template, call_repeat_frames) in enumerate(
_iter_call_templates(template.steps, repeat_frames=repeat_frames)
):
calls.append(
_instantiate_call(
call_template,
call_index=state.call_index,
segment_call_index=segment_call_index,
targets=targets,
repeat_frames=call_repeat_frames,
)
)
state.call_index += 1
post_policies = tuple(
CompiledPostPolicy(
cfg=post.cfg,
entity=post.entity,
source_path=post.source_path,
)
for post in template.post
)
validators: list[CompiledTaskProgramValidator] = []
for validator in template.validators:
if type(validator) is _ObjectNearTargetValidatorTemplate:
target_pose, selection = _resolve_target(
validator.target_id,
targets=targets,
repeat_frames=repeat_frames,
)
validators.append(
CompiledObjectNearTargetValidator(
cfg=validator.cfg,
object=validator.object,
target_pose=target_pose,
target_selection=selection,
source_path=validator.source_path,
)
)
elif type(validator) is _ArticulationJointPositionValidatorTemplate:
validators.append(
CompiledArticulationJointPositionValidator(
cfg=validator.cfg,
articulation=validator.articulation,
source_path=validator.source_path,
)
)
else:
raise TypeError(
f"Unsupported internal validator template {type(validator).__name__}."
)
segment = CompiledTaskProgramSegment(
segment_index=state.segment_index,
segment_id=_segment_identity(
program_id,
source_path=template.source_path,
repeat_frames=repeat_frames,
implicit=False,
),
name=template.name,
calls=tuple(calls),
source_path=template.source_path,
repeat_frames=repeat_frames,
post_policies=post_policies,
validators=tuple(validators),
parallel_block=parallel_block,
implicit=False,
)
state.segment_index += 1
yield segment
[docs]
@dataclass(frozen=True, slots=True, init=False)
class CompiledTaskProgram:
"""Bounded provider-free segment snapshot used by preflight and execution."""
program_id: str
_integration: TaskProgramIntegrationCfg = field(repr=False, compare=False)
_segments: tuple[CompiledTaskProgramSegment, ...] = field(
repr=False,
compare=False,
)
[docs]
def __init__(self, *args: object, **kwargs: object) -> None:
"""Reject construction outside :class:`TaskProgramCompiler`."""
del args, kwargs
raise TypeError(
"CompiledTaskProgram values are created by TaskProgramCompiler."
)
@classmethod
def _create(
cls,
*,
program_id: str,
integration: TaskProgramIntegrationCfg,
segments: tuple[CompiledTaskProgramSegment, ...],
) -> CompiledTaskProgram:
"""Create one compiler-owned materialized program."""
if type(program_id) is not str or not program_id:
raise ValueError("program_id must be a non-empty string.")
if type(integration) is not TaskProgramIntegrationCfg:
raise TypeError("integration must be TaskProgramIntegrationCfg.")
values = tuple(segments)
if not values or not all(
type(segment) is CompiledTaskProgramSegment for segment in values
):
raise TypeError(
"segments must contain at least one CompiledTaskProgramSegment."
)
if tuple(segment.segment_index for segment in values) != tuple(
range(len(values))
):
raise ValueError("Materialized segment indices must be contiguous.")
flattened_calls = tuple(call for segment in values for call in segment.calls)
if len(flattened_calls) > MAX_EXPANDED_CALLS:
raise ValueError(
f"Materialized program exceeds {MAX_EXPANDED_CALLS} calls."
)
if tuple(call.call_index for call in flattened_calls) != tuple(
range(len(flattened_calls))
):
raise ValueError("Materialized call indices must be contiguous.")
instance = object.__new__(cls)
object.__setattr__(instance, "program_id", program_id)
object.__setattr__(
instance,
"_integration",
TaskProgramIntegrationCfg(
robot_profile=integration.robot_profile,
scene_registry=integration.scene_registry,
runtime_preset=integration.runtime_preset,
),
)
object.__setattr__(instance, "_segments", values)
return instance
@property
def integration(self) -> TaskProgramIntegrationCfg:
"""Return an independent integration-selection snapshot."""
return TaskProgramIntegrationCfg(
robot_profile=self._integration.robot_profile,
scene_registry=self._integration.scene_registry,
runtime_preset=self._integration.runtime_preset,
)
@property
def segment_count(self) -> int:
"""Return the number of materialized logical segments."""
return len(self._segments)
[docs]
def iter_segments(self) -> Iterator[CompiledTaskProgramSegment]:
"""Iterate the already materialized provider-free segments."""
return iter(self._segments)
[docs]
def preflight_analyses(self) -> tuple[CompiledTaskProgramAnalysis, ...]:
"""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.
"""
analyses: list[CompiledTaskProgramAnalysis] = []
stretch: list[CompiledTaskProgramSegment] = []
def flush_stretch() -> None:
if not stretch:
return
indices = tuple(segment.segment_index for segment in stretch)
calls = tuple(call.call for segment in stretch for call in segment.calls)
analyses.append(
CompiledTaskProgramAnalysis(
analysis_id=(
f"{self.program_id}:preflight:sequential:"
f"{indices[0]}-{indices[-1]}"
),
kind="sequential_stretch",
calls=calls,
source_path=stretch[0].source_path,
segment_indices=indices,
execution_prefix_length=len(calls),
)
)
stretch.clear()
for segment in self._segments:
block = segment.parallel_block
if block is None:
stretch.append(segment)
continue
flush_stretch()
for branch in block.branches:
calls = tuple(call.call for call in branch.calls)
analyses.append(
CompiledTaskProgramAnalysis(
analysis_id=(
f"{self.program_id}:preflight:parallel:"
f"{segment.segment_index}:{branch.branch_index}"
),
kind="parallel_branch",
calls=calls,
source_path=branch.source_path,
segment_indices=(segment.segment_index,),
execution_prefix_length=len(calls),
)
)
flush_stretch()
return tuple(analyses)
[docs]
def sequential_execution_analysis(
self,
segment_index: int,
) -> CompiledTaskProgramAnalysis:
"""Return current-segment prefix plus downstream sequential look-ahead.
Args:
segment_index: Index of the sequential segment about to execute.
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.
"""
if type(segment_index) is not int:
raise TypeError("segment_index must be an integer.")
if not 0 <= segment_index < len(self._segments):
raise IndexError(f"segment_index {segment_index!r} is outside the program.")
current = self._segments[segment_index]
if current.parallel_block is not None:
raise ValueError("Parallel segments do not have sequential look-ahead.")
window: list[CompiledTaskProgramSegment] = []
for segment in self._segments[segment_index:]:
if segment.parallel_block is not None:
break
window.append(segment)
calls = tuple(call.call for segment in window for call in segment.calls)
indices = tuple(segment.segment_index for segment in window)
return CompiledTaskProgramAnalysis(
analysis_id=(
f"{self.program_id}:execution:sequential:" f"{indices[0]}-{indices[-1]}"
),
kind="sequential_suffix",
calls=calls,
source_path=current.source_path,
segment_indices=indices,
execution_prefix_length=len(current.calls),
)
def __iter__(self) -> Iterator[CompiledTaskProgramSegment]:
return self.iter_segments()
def _materialize_program(
*,
program_id: str,
integration: TaskProgramIntegrationCfg,
targets: Mapping[str, tuple[SemanticPose, ...]],
root: _NodeTemplate,
) -> CompiledTaskProgram:
"""Expand one internal template directly into the public bounded snapshot."""
segments: list[CompiledTaskProgramSegment] = []
expanded_calls = 0
for segment in _iter_segments(
root,
program_id=program_id,
targets=targets,
repeat_frames=(),
state=_ExpansionState(),
):
expanded_calls += len(segment.calls)
if expanded_calls > MAX_EXPANDED_CALLS:
raise TaskProgramCompileError(
"expanded_call_limit",
segment.source_path,
"Program expansion exceeds the static limit of "
f"{MAX_EXPANDED_CALLS} semantic calls.",
)
segments.append(segment)
return CompiledTaskProgram._create(
program_id=program_id,
integration=integration,
segments=tuple(segments),
)
[docs]
class TaskProgramCompiler:
"""Compile validated Task Program ASTs through one static scene manifest."""
[docs]
def __init__(self, scene_manifest: SceneManifest) -> None:
"""Create one provider-free compiler.
Args:
scene_manifest: Canonical provider-free scene identity catalog.
"""
if type(scene_manifest) is not SceneManifest:
raise TypeError("scene_manifest must be exactly SceneManifest.")
self._scene_manifest = scene_manifest
[docs]
@classmethod
def from_scene_registry(cls, registry: SceneRegistry) -> TaskProgramCompiler:
"""Create a compiler from a provider-free SceneRegistry identity snapshot."""
return cls(SceneManifest.from_registry(registry))
def _resolve_scene(
self,
reference: str,
*,
expected_types: tuple[type[SceneEntityRef], ...],
path: ConfigPath,
) -> SceneEntityRef:
"""Resolve and validate one exact typed canonical scene reference."""
try:
resolved = self._scene_manifest.resolve(
reference,
path=path,
)
except TaskProgramConfigError:
raise
except (KeyError, TypeError, ValueError) as exc:
raise TaskProgramCompileError(
"scene_resolution_failed",
path,
str(exc),
) from exc
if type(resolved) not in expected_types:
raise TaskProgramCompileError(
"scene_reference_type_mismatch",
path,
f"Scene reference {reference!r} resolves to "
f"{type(resolved).__name__}, expected one of "
f"{tuple(value.__name__ for value in expected_types)}.",
)
return _copy_scene_ref(resolved)
@staticmethod
def _target_id(
reference: TargetRefCfg,
*,
targets: Mapping[str, tuple[SemanticPose, ...]],
path: ConfigPath,
) -> str:
"""Resolve one statically registered target ID."""
if type(reference) is not TargetRefCfg or reference.kind != "target_ref":
raise TaskProgramCompileError(
"invalid_target_reference",
path,
"Expected an exact target_ref configuration.",
)
if reference.target not in targets:
raise TaskProgramCompileError(
"unknown_target",
(*path, "target"),
f"Unknown target {reference.target!r}.",
)
return reference.target
def _compile_call(
self,
cfg: object,
*,
targets: Mapping[str, tuple[SemanticPose, ...]],
path: ConfigPath,
) -> _CallTemplate:
"""Lower one config call into a provider-free canonical template."""
if type(cfg) is PickCfg:
if cfg.kind != "pick":
raise TaskProgramCompileError(
"invalid_discriminator", (*path, "kind"), "Expected 'pick'."
)
object_ref = self._resolve_scene(
cfg.object,
expected_types=(SceneObjectRef,),
path=(*path, "object"),
)
grasp_ref = (
None
if cfg.grasp is None
else self._resolve_scene(
cfg.grasp,
expected_types=(SceneAffordanceRef,),
path=(*path, "grasp"),
)
)
return _CallTemplate(
kind="pick",
source_path=path,
object=object_ref,
grasp=grasp_ref,
resources=tuple(sorted(cfg.resources.items())),
)
if type(cfg) is PlaceCfg:
if cfg.kind != "place":
raise TaskProgramCompileError(
"invalid_discriminator", (*path, "kind"), "Expected 'place'."
)
object_ref = self._resolve_scene(
cfg.object,
expected_types=(SceneObjectRef,),
path=(*path, "object"),
)
at_target_id = (
None
if cfg.at is None
else self._target_id(
cfg.at,
targets=targets,
path=(*path, "at"),
)
)
on = (
None
if cfg.on is None
else self._resolve_scene(
cfg.on,
expected_types=(SceneObjectRef, SceneAffordanceRef),
path=(*path, "on"),
)
)
inside = (
None
if cfg.inside is None
else self._resolve_scene(
cfg.inside,
expected_types=(SceneObjectRef, SceneAffordanceRef),
path=(*path, "inside"),
)
)
return _CallTemplate(
kind="place",
source_path=path,
object=object_ref,
at_target_id=at_target_id,
on=on,
inside=inside,
resources=tuple(sorted(cfg.resources.items())),
)
if type(cfg) is HandOverCfg:
if cfg.kind != "hand_over":
raise TaskProgramCompileError(
"invalid_discriminator",
(*path, "kind"),
"Expected 'hand_over'.",
)
object_ref = self._resolve_scene(
cfg.object,
expected_types=(SceneObjectRef,),
path=(*path, "object"),
)
final_target_id = (
None
if cfg.final_target is None
else self._target_id(
cfg.final_target,
targets=targets,
path=(*path, "final_target"),
)
)
return _CallTemplate(
kind="hand_over",
source_path=path,
object=object_ref,
final_target_id=final_target_id,
resources=tuple(sorted(cfg.resources.items())),
)
if type(cfg) is RegisteredSemanticCallCfg:
if cfg.kind != "registered":
raise TaskProgramCompileError(
"invalid_discriminator",
(*path, "kind"),
"Expected 'registered'.",
)
snapshot = RegisteredSemanticCall(
call_id=cfg.call_id,
arguments=cfg.arguments,
resources=cfg.resources,
)
return _CallTemplate(
kind="registered",
source_path=path,
call_id=snapshot.call_id,
arguments=snapshot.arguments,
resources=tuple(sorted(snapshot.resources.items())),
)
raise TaskProgramCompileError(
"unsupported_call",
path,
f"Unsupported semantic call config {type(cfg).__name__}.",
)
def _compile_node(
self,
node: ProgramNodeCfg,
*,
targets: Mapping[str, tuple[SemanticPose, ...]],
path: ConfigPath,
inside_segment: bool,
inside_parallel: bool,
) -> _NodeTemplate:
"""Compile static AST structure without expanding repeats."""
if type(node) is InvokeCfg:
if node.kind != "invoke":
raise TaskProgramCompileError(
"invalid_discriminator", (*path, "kind"), "Expected 'invoke'."
)
return _InvokeTemplate(
call=self._compile_call(
node.call,
targets=targets,
path=(*path, "call"),
),
source_path=path,
)
if type(node) is SequenceCfg:
if node.kind != "sequence":
raise TaskProgramCompileError(
"invalid_discriminator",
(*path, "kind"),
"Expected 'sequence'.",
)
if not node.items:
raise TaskProgramCompileError(
"empty_sequence",
(*path, "items"),
"Sequence items must contain at least one program node.",
)
return _SequenceTemplate(
items=tuple(
self._compile_node(
child,
targets=targets,
path=(*path, "items", index),
inside_segment=inside_segment,
inside_parallel=inside_parallel,
)
for index, child in enumerate(node.items)
),
source_path=path,
)
if type(node) is RepeatCfg:
if node.kind != "repeat":
raise TaskProgramCompileError(
"invalid_discriminator", (*path, "kind"), "Expected 'repeat'."
)
if type(node.count) is not int or not 1 <= node.count <= MAX_REPEAT_COUNT:
raise TaskProgramCompileError(
"invalid_repeat_count",
(*path, "count"),
f"Repeat count must be an integer in [1, {MAX_REPEAT_COUNT}].",
)
return _RepeatTemplate(
count=node.count,
body=self._compile_node(
node.body,
targets=targets,
path=(*path, "body"),
inside_segment=inside_segment,
inside_parallel=inside_parallel,
),
source_path=path,
)
if type(node) is SegmentCfg:
if inside_parallel:
raise TaskProgramCompileError(
"segment_inside_parallel",
path,
"Parallel branches may contain only Invoke, Sequence, and "
"Repeat nodes; wrap the Parallel node in one Segment instead.",
)
if inside_segment:
raise TaskProgramCompileError(
"nested_segment",
path,
"Nested Segment nodes are ambiguous and forbidden.",
)
if node.kind != "segment":
raise TaskProgramCompileError(
"invalid_discriminator", (*path, "kind"), "Expected 'segment'."
)
post: list[_PostTemplate] = []
for index, cfg in enumerate(node.post):
post_path = (*path, "post", index)
if type(cfg) is not WaitStablePostCfg or cfg.kind != "wait_stable":
raise TaskProgramCompileError(
"unsupported_post_policy",
post_path,
"Supported schemas accept only exact wait_stable post policies.",
)
entity = self._resolve_scene(
cfg.entity,
expected_types=_SCENE_REF_TYPES,
path=(*post_path, "entity"),
)
post.append(
_PostTemplate(
cfg=WaitStablePostCfg(
entity=cfg.entity,
preset=cfg.preset,
kind=cfg.kind,
),
entity=entity,
source_path=post_path,
)
)
validators: list[_ValidatorTemplate] = []
for index, cfg in enumerate(node.validators):
validator_path = (*path, "validators", index)
if type(cfg) is ObjectNearTargetValidatorCfg:
if cfg.kind != "object_near_target":
raise TaskProgramCompileError(
"unsupported_validator",
validator_path,
"ObjectNearTargetValidatorCfg must use kind "
"'object_near_target'.",
)
if cfg.target not in targets:
raise TaskProgramCompileError(
"unknown_target",
(*validator_path, "target"),
f"Unknown target {cfg.target!r}.",
)
object_ref = self._resolve_scene(
cfg.object,
expected_types=(SceneObjectRef,),
path=(*validator_path, "object"),
)
validators.append(
_ObjectNearTargetValidatorTemplate(
cfg=ObjectNearTargetValidatorCfg(
object=cfg.object,
target=cfg.target,
position_tolerance=cfg.position_tolerance,
kind=cfg.kind,
),
object=object_ref,
target_id=cfg.target,
source_path=validator_path,
)
)
continue
if type(cfg) is ArticulationJointPositionValidatorCfg:
if cfg.kind != "articulation_joint_position":
raise TaskProgramCompileError(
"unsupported_validator",
validator_path,
"ArticulationJointPositionValidatorCfg must use kind "
"'articulation_joint_position'.",
)
articulation_ref = self._resolve_scene(
cfg.articulation,
expected_types=(SceneArticulationRef,),
path=(*validator_path, "articulation"),
)
validators.append(
_ArticulationJointPositionValidatorTemplate(
cfg=ArticulationJointPositionValidatorCfg(
articulation=cfg.articulation,
joint=cfg.joint,
minimum_position=cfg.minimum_position,
maximum_position=cfg.maximum_position,
kind=cfg.kind,
),
articulation=articulation_ref,
source_path=validator_path,
)
)
continue
raise TaskProgramCompileError(
"unsupported_validator",
validator_path,
f"Unsupported validator {type(cfg).__name__}.",
)
steps = self._compile_node(
node.steps,
targets=targets,
path=(*path, "steps"),
inside_segment=True,
inside_parallel=False,
)
if type(steps) is not _ParallelTemplate and _contains_parallel(steps):
raise TaskProgramCompileError(
"mixed_parallel_segment",
(*path, "steps"),
"A Segment may contain either a call-only program or one direct "
"Parallel node, not a mixed sequential/parallel tree.",
)
return _SegmentTemplate(
name=node.name,
steps=steps,
post=tuple(post),
validators=tuple(validators),
source_path=path,
)
if type(node) is ParallelCfg:
if inside_parallel:
raise TaskProgramCompileError(
"nested_parallel",
path,
"Nested Parallel nodes are forbidden.",
)
if node.kind != "parallel":
raise TaskProgramCompileError(
"invalid_discriminator",
(*path, "kind"),
"Expected 'parallel'.",
)
if len(node.branches) < 2:
raise TaskProgramCompileError(
"parallel_branch_count",
(*path, "branches"),
"Parallel requires at least two branches.",
)
if type(node.barrier) is not BarrierCfg:
raise TaskProgramCompileError(
"parallel_barrier_required",
(*path, "barrier"),
"Parallel.barrier must be an exact BarrierCfg.",
)
branches = tuple(
self._compile_node(
branch,
targets=targets,
path=(*path, "branches", index),
inside_segment=inside_segment,
inside_parallel=True,
)
for index, branch in enumerate(node.branches)
)
if any(_contains_parallel(branch) for branch in branches):
raise TaskProgramCompileError(
"nested_parallel",
(*path, "branches"),
"Nested Parallel nodes are forbidden.",
)
barrier = node.barrier
if barrier.kind != "barrier":
raise TaskProgramCompileError(
"invalid_discriminator",
(*path, "barrier", "kind"),
"Expected 'barrier'.",
)
if barrier.failure_policy != "fail_fast":
raise TaskProgramCompileError(
"unsupported_failure_policy",
(*path, "barrier", "failure_policy"),
"Barrier failure_policy must be exactly 'fail_fast'.",
)
return _ParallelTemplate(
branches=branches,
barrier=_BarrierTemplate(
name=barrier.name,
timeout_steps=barrier.timeout_steps,
failure_policy=barrier.failure_policy,
source_path=(*path, "barrier"),
),
source_path=path,
)
raise TaskProgramCompileError(
"unsupported_program_node",
path,
f"Unsupported program node {type(node).__name__}.",
)
@staticmethod
def _compile_targets(
targets: Mapping[str, CyclicPoseTargetCfg],
) -> Mapping[str, tuple[SemanticPose, ...]]:
"""Compile static pose providers without selecting repeat values."""
compiled: dict[str, tuple[SemanticPose, ...]] = {}
for target_id, target in targets.items():
path = ("targets", target_id)
if type(target) is not CyclicPoseTargetCfg or target.kind != "cyclic_pose":
raise TaskProgramCompileError(
"unsupported_target",
path,
"Supported schemas accept only exact cyclic_pose targets.",
)
poses: list[SemanticPose] = []
if not target.values:
raise TaskProgramCompileError(
"empty_target_values",
(*path, "values"),
"Cyclic target values must contain at least one pose.",
)
for index, pose in enumerate(target.values):
if type(pose) is not PoseCfg:
raise TaskProgramCompileError(
"invalid_pose",
(*path, "values", index),
"Target values must be exact PoseCfg values.",
)
poses.append(SemanticPose(pose.position, pose.quaternion_wxyz))
compiled[target_id] = tuple(poses)
return MappingProxyType(compiled)
[docs]
def compile(self, config: TaskProgramCfg) -> CompiledTaskProgram:
"""Compile one validated AST into a bounded provider-free program.
Args:
config: Strict, supported-version Task Program configuration.
Returns:
Immutable segments with repeat-local targets already resolved.
Raises:
TaskProgramCompileError: If typed scene resolution or AST lowering
fails.
"""
if type(config) is not TaskProgramCfg:
raise TypeError("config must be exactly TaskProgramCfg.")
targets = self._compile_targets(config.targets)
root = self._compile_node(
config.program,
targets=targets,
path=("program",),
inside_segment=False,
inside_parallel=False,
)
integration = TaskProgramIntegrationCfg(
robot_profile=config.integration.robot_profile,
scene_registry=config.integration.scene_registry,
runtime_preset=config.integration.runtime_preset,
)
return _materialize_program(
program_id=config.program_id,
integration=integration,
targets=targets,
root=root,
)
__all__: list[str] = []