Authoring a Task Program in Python

Authoring a Task Program in Python#

This tutorial builds, validates, and compiles an Embodied Task Program entirely in Python. It focuses on the provider-independent language: no simulator, Gym environment, robot asset, or YAML file is required.

The result is a compiled semantic workflow, not robot commands. Continue with Creating a Modular Environment and Configure and Run an Embodied Task Program when you are ready to bind the same kind of program to a simulation environment. Use Atomic actions instead when application code should directly plan or execute individual Atomic Skills.

What you will build#

The example declares one scene object and a typed Pick-and-Place workflow:

Repeat (3 times)
└── Segment: move_cube
    ├── Pick cube
    ├── Place cube at the next cyclic target
    ├── Wait for cube to become stable
    └── Validate cube is near the selected target

Two target poses are selected cyclically, so the three repetitions use target indices 0, 1, and 0. Compilation expands those occurrences into stable segments and call indices without observing live scene state.

The code#

The complete runnable example is scripts/tutorials/task_program/build_and_compile.py:

Code for build_and_compile.py
  1# ----------------------------------------------------------------------------
  2# Copyright (c) 2021-2026 DexForce Technology Co., Ltd.
  3#
  4# Licensed under the Apache License, Version 2.0 (the "License");
  5# you may not use this file except in compliance with the License.
  6# You may obtain a copy of the License at
  7#
  8#     http://www.apache.org/licenses/LICENSE-2.0
  9#
 10# Unless required by applicable law or agreed to in writing, software
 11# distributed under the License is distributed on an "AS IS" BASIS,
 12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 13# See the License for the specific language governing permissions and
 14# limitations under the License.
 15# ----------------------------------------------------------------------------
 16
 17"""Build and compile a provider-independent Task Program in Python."""
 18
 19from __future__ import annotations
 20
 21from embodichain.lab.task_program import (
 22    CompiledTaskProgram,
 23    CyclicPoseTargetCfg,
 24    InvokeCfg,
 25    ObjectNearTargetValidatorCfg,
 26    PickCfg,
 27    PlaceCfg,
 28    PoseCfg,
 29    RepeatCfg,
 30    SegmentCfg,
 31    SequenceCfg,
 32    TargetRefCfg,
 33    TaskProgramCfg,
 34    TaskProgramCompiler,
 35    TaskProgramIntegrationCfg,
 36    WaitStablePostCfg,
 37)
 38from embodichain.lab.task_program.semantics import (
 39    SceneEntityManifest,
 40    SceneManifest,
 41    SceneObjectRef,
 42)
 43
 44__all__ = [
 45    "build_program",
 46    "build_scene_manifest",
 47    "main",
 48    "print_compiled_program",
 49]
 50
 51
 52def build_scene_manifest() -> SceneManifest:
 53    """Declare the provider-independent scene identities used by the program."""
 54    return SceneManifest(
 55        entries=(
 56            SceneEntityManifest(
 57                ref=SceneObjectRef("cube"),
 58                semantic_type="cube",
 59            ),
 60        )
 61    )
 62
 63
 64def build_program() -> TaskProgramCfg:
 65    """Build a typed Pick-and-Place Task Program without YAML or JSON."""
 66    drop_poses = CyclicPoseTargetCfg(
 67        values=(
 68            PoseCfg(
 69                position=(0.40, -0.20, 0.10),
 70                quaternion_wxyz=(1.0, 0.0, 0.0, 0.0),
 71            ),
 72            PoseCfg(
 73                position=(0.40, 0.20, 0.10),
 74                quaternion_wxyz=(1.0, 0.0, 0.0, 0.0),
 75            ),
 76        )
 77    )
 78    move_cube = SegmentCfg(
 79        name="move_cube",
 80        steps=SequenceCfg(
 81            items=(
 82                InvokeCfg(
 83                    call=PickCfg(
 84                        object="cube",
 85                        resources={"primary": "manipulator"},
 86                    )
 87                ),
 88                InvokeCfg(
 89                    call=PlaceCfg(
 90                        object="cube",
 91                        at=TargetRefCfg(target="drop_pose"),
 92                        resources={"primary": "manipulator"},
 93                    )
 94                ),
 95            )
 96        ),
 97        post=(WaitStablePostCfg(entity="cube"),),
 98        validators=(
 99            ObjectNearTargetValidatorCfg(
100                object="cube",
101                target="drop_pose",
102                position_tolerance=0.03,
103            ),
104        ),
105    )
106    return TaskProgramCfg(
107        program_id="python_pick_and_place",
108        integration=TaskProgramIntegrationCfg(
109            robot_profile="tutorial_robot",
110            scene_registry="tutorial_scene",
111            runtime_preset="trajectory",
112        ),
113        targets={"drop_pose": drop_poses},
114        program=RepeatCfg(count=3, body=move_cube),
115    )
116
117
118def print_compiled_program(compiled: CompiledTaskProgram) -> None:
119    """Print the deterministic expansion produced by the compiler."""
120    print(f"Program: {compiled.program_id}")
121    print(f"Segments: {compiled.segment_count}")
122    for segment in compiled:
123        repeat = segment.repeat_frames[-1]
124        call_names = " -> ".join(type(item.call).__name__ for item in segment.calls)
125        target = segment.calls[-1].target_selections[0]
126        print(
127            f"[{segment.segment_index}] {segment.name} "
128            f"(repeat {repeat.iteration_index + 1}/{repeat.count}): "
129            f"{call_names}; {target.target_id}[{target.value_index}]"
130        )
131
132
133def main() -> None:
134    """Compile and inspect the tutorial Task Program."""
135    scene_manifest = build_scene_manifest()
136    program = build_program()
137    compiled = TaskProgramCompiler(scene_manifest).compile(program)
138    print_compiled_program(compiled)
139
140
141if __name__ == "__main__":
142    main()

Run it from the repository root:

python scripts/tutorials/task_program/build_and_compile.py

The output shows the deterministic expansion:

Program: python_pick_and_place
Segments: 3
[0] move_cube (repeat 1/3): Pick -> Place; drop_pose[0]
[1] move_cube (repeat 2/3): Pick -> Place; drop_pose[1]
[2] move_cube (repeat 3/3): Pick -> Place; drop_pose[0]

Declare provider-independent scene identities#

The compiler needs canonical identities and their semantic types, but it does not need simulator objects or state providers. A SceneManifest is the static catalog for that boundary:

def build_scene_manifest() -> SceneManifest:
    """Declare the provider-independent scene identities used by the program."""
    return SceneManifest(
        entries=(
            SceneEntityManifest(
                ref=SceneObjectRef("cube"),
                semantic_type="cube",
            ),
        )
    )

SceneObjectRef("cube") establishes that cube is an object rather than an articulation, link, or affordance. The compiler uses this typed identity to reject invalid or unknown references before a live environment is created.

Build the typed program tree#

Python authoring uses the same declarative schema as JSON or YAML. Constructors such as PickCfg, PlaceCfg, SequenceCfg, and RepeatCfg validate their own fields, so discriminator strings such as kind: pick do not need to be written manually:

def build_program() -> TaskProgramCfg:
    """Build a typed Pick-and-Place Task Program without YAML or JSON."""
    drop_poses = CyclicPoseTargetCfg(
        values=(
            PoseCfg(
                position=(0.40, -0.20, 0.10),
                quaternion_wxyz=(1.0, 0.0, 0.0, 0.0),
            ),
            PoseCfg(
                position=(0.40, 0.20, 0.10),
                quaternion_wxyz=(1.0, 0.0, 0.0, 0.0),
            ),
        )
    )
    move_cube = SegmentCfg(
        name="move_cube",
        steps=SequenceCfg(
            items=(
                InvokeCfg(
                    call=PickCfg(
                        object="cube",
                        resources={"primary": "manipulator"},
                    )
                ),
                InvokeCfg(
                    call=PlaceCfg(
                        object="cube",
                        at=TargetRefCfg(target="drop_pose"),
                        resources={"primary": "manipulator"},
                    )
                ),
            )
        ),
        post=(WaitStablePostCfg(entity="cube"),),
        validators=(
            ObjectNearTargetValidatorCfg(
                object="cube",
                target="drop_pose",
                position_tolerance=0.03,
            ),
        ),
    )
    return TaskProgramCfg(
        program_id="python_pick_and_place",
        integration=TaskProgramIntegrationCfg(
            robot_profile="tutorial_robot",
            scene_registry="tutorial_scene",
            runtime_preset="trajectory",
        ),
        targets={"drop_pose": drop_poses},
        program=RepeatCfg(count=3, body=move_cube),
    )

The three values in TaskProgramIntegrationCfg are exact identifiers for the robot profile, scene registry, and runtime preset that a trusted deployment will later provide. Provider-independent compilation preserves those selections but does not construct or contact their live providers.

The program remains executable-free even though it is authored in Python. Do not put callbacks, simulator objects, planners, or controller commands in a Task Program. Registered extensions also accept only declarative arguments; their executable lowerers belong to the trusted integration.

Compile and inspect the program#

TaskProgramCompiler resolves the scene references, expands the bounded repeat, selects each cyclic target, and assigns stable segment and call indices:

def print_compiled_program(compiled: CompiledTaskProgram) -> None:
    """Print the deterministic expansion produced by the compiler."""
    print(f"Program: {compiled.program_id}")
    print(f"Segments: {compiled.segment_count}")
    for segment in compiled:
        repeat = segment.repeat_frames[-1]
        call_names = " -> ".join(type(item.call).__name__ for item in segment.calls)
        target = segment.calls[-1].target_selections[0]
        print(
            f"[{segment.segment_index}] {segment.name} "
            f"(repeat {repeat.iteration_index + 1}/{repeat.count}): "
            f"{call_names}; {target.target_id}[{target.value_index}]"
        )


def main() -> None:
    """Compile and inspect the tutorial Task Program."""
    scene_manifest = build_scene_manifest()
    program = build_program()
    compiled = TaskProgramCompiler(scene_manifest).compile(program)
    print_compiled_program(compiled)


if __name__ == "__main__":
    main()

The returned CompiledTaskProgram is immutable and can be iterated more than once with the same result. Segment post-policies and validators are also resolved during compilation, but they run only after an environment adapter binds the compiled program to live services.

Understand the boundary#

This tutorial stops at the correct provider-independent boundary. Compilation does not:

  • observe object poses or robot state;

  • select planners or generate controller commands;

  • verify that the named robot profile and runtime preset are installed; or

  • call env.step().

Those responsibilities belong to the trusted integration, Atomic Skills, and the Gym execution bridge. Keeping them out of this example makes the Task Program language independently testable and keeps task intent reusable across compatible embodiments.

Diagnose an invalid reference#

Compiler failures include a stable error code and an exact source path. For example, compiling the same program against an empty scene manifest fails before any provider can be touched:

from embodichain.lab.task_program import (
    TaskProgramCompileError,
    TaskProgramCompiler,
    render_config_path,
)
from embodichain.lab.task_program.semantics import SceneManifest

try:
    TaskProgramCompiler(SceneManifest()).compile(build_program())
except TaskProgramCompileError as error:
    print(error.code, render_config_path(error.path))

Use the reported path to fix the source declaration. Do not catch a compile error and continue into execution with a partially resolved program.

Next steps#

Contents