embodichain.lab.task_program

embodichain.lab.task_program#

Declarative orchestration and execution language for embodied tasks.

Task Program is the provider-independent language between human or model authored task intent and executable Atomic Skills. Semantic Calls and their scene, profile, effect, and evidence contracts are internal language semantics; provider-specific assembly lives in integrations, while Gym owns only the environment execution bridge.

`TaskProgramCfg`	Strict, executable-free Task Program configuration.
`TaskProgramIntegrationCfg`	Static integration references selected by one Task Program.
`PoseCfg`	One declarative Cartesian pose using an XYZW quaternion.
`TargetRefCfg`	Reference to one top-level typed target provider.
`CyclicPoseTargetCfg`	Finite pose values selected cyclically by the enclosing repeat index.
`PickCfg`	Declarative request to acquire one registered object.
`PlaceCfg`	Declarative request to place one held object at one destination.
`HandOverCfg`	Declarative request to transfer one held object between resources.
`RegisteredSemanticCallCfg`	Safe declarative payload for one catalog-registered semantic call.
`InvokeCfg`	Invoke exactly one semantic call at the current program boundary.
`SequenceCfg`	Execute one non-empty ordered tuple of program nodes.
`RepeatCfg`	Repeat one child node a finite validated number of times.
`SegmentCfg`	Logical transaction with optional segment-specific language instruction.
`ParallelCfg`	Execute two or more branches concurrently and join at one barrier.
`BarrierCfg`	Explicit synchronization boundary owned by one parallel node.
`WaitStablePostCfg`	Wait for one registered entity to satisfy a named stability preset.
`ObjectNearTargetValidatorCfg`	Validate an object's position against one resolved target.
`ArticulationJointPositionValidatorCfg`	Validate one articulation joint against an inclusive position interval.
`TaskProgramCompiler`	Compile validated Task Program ASTs through one static scene manifest.
`CompiledTaskProgram`	Immutable overall instruction and bounded segments for task execution.
`load_task_program`(path, *[, base_dir, ...])	Safely load and strictly decode one JSON or YAML Task Program file.
`loads_task_program_json`(text, *[, ...])	Strictly parse and decode one untrusted Task Program JSON document.
`parse_task_program_json`(text, *[, max_bytes])	Parse one bounded Task Program JSON object without decoding its schema.
`decode_task_program`(data, *[, ...])	Decode untrusted JSON/YAML-shaped values into strict program config.
`validate_task_program`(config, context)	Resolve external references without observing or executing an environment.
`render_config_path`(path)	Render one configuration path using JSONPath-like notation.
`ConfigPath`	Built-in immutable sequence.
`ConfigPathPart`	Represent a PEP 604 union type
`SceneReferenceRole`	alias of `Literal`['entity', 'object', 'articulation', 'affordance', 'object_or_affordance']
`TaskProgramValidationContext`	Provider-free static validation boundary for external references.
`TaskProgramConfigError`	Base pathful diagnostic for Task Program configuration failures.
`TaskProgramDecodeError`	Raised when untrusted data does not match a supported strict schema.
`TaskProgramValidationError`	Raised when an explicit static integration context rejects a reference.
`TaskProgramCompileError`	Raised when a validated AST cannot lower to canonical semantic calls.

Schema#

class embodichain.lab.task_program.TaskProgramCfg[source]#

Strict, executable-free Task Program configuration.

instruction describes the overall task independently of the selected embodiment and recorder. Segment instructions describe individual phases.

Methods:

`__init__`([program_id, integration, program, ...])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(program_id=<factory>, integration=<factory>, program=<factory>, targets=<factory>, instruction=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.TaskProgramIntegrationCfg[source]#

Static integration references selected by one Task Program.

Methods:

`__init__`([robot_profile, scene_registry, ...])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(robot_profile=<factory>, scene_registry=<factory>, runtime_preset=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.PoseCfg[source]#

One declarative Cartesian pose using an XYZW quaternion.

Methods:

`__init__`([position, quaternion_xyzw])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(position=<factory>, quaternion_xyzw=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.TargetRefCfg[source]#

Reference to one top-level typed target provider.

Methods:

`__init__`([target, kind])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(target=<factory>, kind=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.CyclicPoseTargetCfg[source]#

Finite pose values selected cyclically by the enclosing repeat index.

Methods:

`__init__`([values, kind])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(values=<factory>, kind=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.PickCfg[source]#

Declarative request to acquire one registered object.

Methods:

`__init__`([object, grasp, resources, kind])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

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

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.PlaceCfg[source]#

Declarative request to place one held object at one destination.

Methods:

`__init__`([object, at, on, inside, ...])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

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

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.HandOverCfg[source]#

Declarative request to transfer one held object between resources.

Methods:

`__init__`([object, final_target, resources, kind])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

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

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.RegisteredSemanticCallCfg[source]#

Safe declarative payload for one catalog-registered semantic call.

Methods:

`__init__`([call_id, arguments, resources, kind])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

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

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.InvokeCfg[source]#

Invoke exactly one semantic call at the current program boundary.

Methods:

`__init__`([call, kind])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(call=<factory>, kind=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.SequenceCfg[source]#

Execute one non-empty ordered tuple of program nodes.

Methods:

`__init__`([items, kind])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(items=<factory>, kind=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.RepeatCfg[source]#

Repeat one child node a finite validated number of times.

Methods:

`__init__`([count, body, kind])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(count=<factory>, body=<factory>, kind=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.SegmentCfg[source]#

Logical transaction with optional segment-specific language instruction.

instruction describes this segment for dataset annotation. When omitted, Gym demonstrations fall back to the episode’s overall task instruction.

Methods:

`__init__`([name, steps, post, validators, ...])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(name=<factory>, steps=<factory>, post=<factory>, validators=<factory>, kind=<factory>, instruction=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.ParallelCfg[source]#

Execute two or more branches concurrently and join at one barrier.

Methods:

`__init__`([branches, barrier, kind])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(branches=<factory>, barrier=<factory>, kind=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.BarrierCfg[source]#

Explicit synchronization boundary owned by one parallel node.

Methods:

`__init__`([name, timeout_steps, ...])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(name=<factory>, timeout_steps=<factory>, failure_policy=<factory>, kind=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.WaitStablePostCfg[source]#

Wait for one registered entity to satisfy a named stability preset.

Methods:

`__init__`([entity, preset, kind])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(entity=<factory>, preset=<factory>, kind=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.ObjectNearTargetValidatorCfg[source]#

Validate an object’s position against one resolved target.

Methods:

`__init__`([object, target, ...])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(object=<factory>, target=<factory>, position_tolerance=<factory>, kind=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.ObjectLiftedValidatorCfg[source]#

Require a world-Z increase since the owning segment started.

Parameters:

object (str) – Logical scene object whose observed height is measured.
minimum_lift (float) – Inclusive minimum increase in meters, relative to the observed pose captured before the segment starts executing.
kind (str) – Fixed object_lifted discriminator.

Methods:

`__init__`([object, minimum_lift, kind])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(object=<factory>, minimum_lift=<factory>, kind=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.ObjectRelativePositionValidatorCfg[source]#

Require an object’s position near a live reference plus a declared offset.

Each tolerance is exclusive; None disables checking that axis. At least one axis must be checked. world_axes preserves world-oriented offsets and tolerances; parent measures them in the live reference frame.

Parameters:

object (str) – Logical scene object to validate.
reference (str) – Logical scene object whose live position defines the origin.
position_offset (tuple[float, float, float]) – Desired XYZ displacement in the selected frame, in meters.
position_tolerance (tuple[float | None, float | None, float | None]) – Exclusive XYZ error bounds in meters; None disables the corresponding axis.
reference_frame (str) – world_axes (default) or the live reference’s parent frame.
kind (str) – Fixed object_relative_position discriminator.

Methods:

`__init__`([object, reference, ...])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(object=<factory>, reference=<factory>, position_offset=<factory>, position_tolerance=<factory>, kind=<factory>, reference_frame=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.ObjectUprightValidatorCfg[source]#

Require the object’s local Z axis to tilt less than an angle from world Z.

Parameters:

object (str) – Logical scene object whose local Z axis is measured.
maximum_tilt_degrees (float) – Exclusive maximum angle from world Z in degrees.
kind (str) – Fixed object_upright discriminator.

Methods:

`__init__`([object, maximum_tilt_degrees, kind])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(object=<factory>, maximum_tilt_degrees=<factory>, kind=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

class embodichain.lab.task_program.ArticulationJointPositionValidatorCfg[source]#

Validate one articulation joint against an inclusive position interval.

Methods:

`__init__`([articulation, joint, ...])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

__init__(articulation=<factory>, joint=<factory>, minimum_position=<factory>, maximum_position=<factory>, kind=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

Loading and validation#

embodichain.lab.task_program.load_task_program(path, *, base_dir=None, integration=None, validation_context=None)[source]#

Safely load and strictly decode one JSON or YAML Task Program file.

Relative paths are resolved from base_dir when provided. Otherwise, they retain normal pathlib.Path semantics and therefore resolve from the process working directory when opened.

Parameters:

path (str | PathLike[str]) – JSON, YAML, or YML file to load.
base_dir (str | PathLike[str] | None) – Optional directory used to resolve a relative path.
integration (TaskProgramIntegrationCfg | None) – Optional trusted integration selection injected into an unbound program before strict decoding.
validation_context (TaskProgramValidationContext | None) – Optional provider-free static reference validator applied after decoding either serialized format.

Return type:

TaskProgramCfg

Returns:

An owned, validated Task Program configuration.

Raises:

FileNotFoundError – If the resolved path is not a regular file.
ValueError – If the file is too large, has an unsupported extension, or contains ambiguous or invalid serialized data.
TaskProgramValidationError – If validation_context rejects an external reference.
UnicodeDecodeError – If the file is not valid UTF-8.

embodichain.lab.task_program.loads_task_program_json(text, *, integration=None, validation_context=None, max_bytes=4194304)[source]#

Strictly parse and decode one untrusted Task Program JSON document.

The input must be one plain JSON document. Markdown fences, trailing text, multiple documents, duplicate keys, non-finite numbers, and oversized input are rejected before the existing Task Program decoder is called.

Parameters:

text (str) – Untrusted JSON response text.
integration (TaskProgramIntegrationCfg | None) – Optional trusted integration selection injected before strict decoding. The source document must omit integration when this is provided.
validation_context (TaskProgramValidationContext | None) – Optional provider-free static reference validator.
max_bytes (int) – Maximum UTF-8 encoded response size.

Return type:

TaskProgramCfg

Returns:

Fully owned and internally validated Task Program configuration.

Raises:

TypeError – If text or max_bytes has the wrong exact type.
ValueError – If max_bytes is not positive.
TaskProgramDecodeError – If parsing or strict decoding fails.

embodichain.lab.task_program.parse_task_program_json(text, *, max_bytes=4194304)[source]#

Parse one bounded Task Program JSON object without decoding its schema.

This parse-only boundary lets a host-controlled frontend inspect or inject fields before calling decode_task_program(). It rejects duplicate keys, non-finite numbers, trailing content, invalid Unicode, excessive nesting, oversized UTF-8 input, and non-object top-level values. It does not validate the Task Program schema.

Parameters:

text (str) – Untrusted JSON document text.
max_bytes (int) – Maximum accepted UTF-8 encoded input size.

Return type:

dict[str, object]

Returns:

Exact JSON object mapping ready for explicit schema decoding.

Raises:

TypeError – If text or max_bytes has the wrong exact type.
ValueError – If max_bytes is not positive.
TaskProgramDecodeError – If strict JSON parsing fails.

embodichain.lab.task_program.decode_task_program(data, *, validation_context=None)[source]#

Decode untrusted JSON/YAML-shaped values into strict program config.

The schema contains sequential nodes plus deterministic parallel blocks with explicit barriers.

Parameters:

data (object) – Exact JSON-compatible mapping produced by a trusted parser.
validation_context (TaskProgramValidationContext | None) – Optional provider-free static reference validator.

Return type:

TaskProgramCfg

Returns:

Fully owned and internally validated Task Program configuration.

Raises:

TaskProgramDecodeError – If data is unsafe or violates the schema.
TaskProgramValidationError – If an explicit context rejects a reference.

embodichain.lab.task_program.validate_task_program(config, context)[source]#

Resolve external references without observing or executing an environment.

Parameters:

config (TaskProgramCfg) – Fully decoded and internally validated Task Program.
context (TaskProgramValidationContext) – Provider-free static integration/catalog/scene validator.

Raises:

TypeError – If either argument has the wrong contract.
TaskProgramValidationError – If an external reference is unavailable.

Return type:

None

embodichain.lab.task_program.render_config_path(path)[source]#

Render one configuration path using JSONPath-like notation.

Parameters:: path (tuple[str | int, ...]) – Tuple of mapping keys and sequence indices.
Return type:: str
Returns:: Stable human-readable path beginning at $.

embodichain.lab.task_program.ConfigPath(*args, **kwargs)#

Built-in immutable sequence.

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

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

alias of tuple[str | int, …]

embodichain.lab.task_program.ConfigPathPart = str | int#

Represent a PEP 604 union type

E.g. for int | str

embodichain.lab.task_program.SceneReferenceRole(*args, **kwargs)#: alias of Literal[‘entity’, ‘object’, ‘articulation’, ‘affordance’, ‘object_or_affordance’]

class embodichain.lab.task_program.TaskProgramValidationContext[source]#

Provider-free static validation boundary for external references.

Implementations may resolve profile, scene, preset, catalog, affordance, and resource IDs, but must not observe simulation state, construct planners, or execute calls.

Methods:

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

__init__(*args, **kwargs)#

__new__(**kwargs)#

class embodichain.lab.task_program.TaskProgramConfigError[source]#

Base pathful diagnostic for Task Program configuration failures.

Methods:

`__init__`(code, path, message)	Create one stable pathful diagnostic.
`__new__`(**kwargs)

__init__(code, path, message)[source]#

Create one stable pathful diagnostic.

Parameters:

code (str) – Machine-readable failure code.
path (tuple[str | int, ...]) – Exact configuration location.
message (str) – Human-readable explanation.

__new__(**kwargs)#

class embodichain.lab.task_program.TaskProgramDecodeError[source]#

Raised when untrusted data does not match a supported strict schema.

Methods:

`__init__`(code, path, message)	Create one stable pathful diagnostic.
`__new__`(**kwargs)

__init__(code, path, message)#

Create one stable pathful diagnostic.

Parameters:

code (str) – Machine-readable failure code.
path (tuple[str | int, ...]) – Exact configuration location.
message (str) – Human-readable explanation.

__new__(**kwargs)#

class embodichain.lab.task_program.TaskProgramValidationError[source]#

Raised when an explicit static integration context rejects a reference.

Methods:

`__init__`(code, path, message)	Create one stable pathful diagnostic.
`__new__`(**kwargs)

__init__(code, path, message)#

Create one stable pathful diagnostic.

Parameters:

code (str) – Machine-readable failure code.
path (tuple[str | int, ...]) – Exact configuration location.
message (str) – Human-readable explanation.

__new__(**kwargs)#

Compilation#

Compilation is provider-independent. Environment and simulator binding is documented under embodichain.lab.task_program.integrations; the final Gym lifecycle bridge is embodichain.lab.gym.envs.task_program.

class embodichain.lab.task_program.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

class embodichain.lab.task_program.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:

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

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

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, ...]

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.TaskProgramCompileError[source]#

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

Methods:

`__init__`(code, path, message)	Create one stable pathful diagnostic.
`__new__`(**kwargs)

__init__(code, path, message)#

Create one stable pathful diagnostic.

Parameters:

code (str) – Machine-readable failure code.
path (tuple[str | int, ...]) – Exact configuration location.
message (str) – Human-readable explanation.

__new__(**kwargs)#

Expansion integration#

Call-scoped expansion remains outside the Task Program language. Runtime requests identify one grounded Atomic planning call, while the integration adapter exposes the resulting Atomic plan through the source-neutral motion template contract.

`TaskProgramPlanRequest`	Stable Task Program identity for one grounded Atomic planning call.
`TaskProgramPlanTransformFactory`	Create an optional Atomic plan transform for one grounded call.

class embodichain.lab.task_program.runtime.expansion.TaskProgramPlanRequest[source]#

Stable Task Program identity for one grounded Atomic planning call.

Methods:

__init__(workflow_id, workflow_call_index, ...)

__init__(workflow_id, workflow_call_index, analysis_call_index, call, invocation)#

class embodichain.lab.task_program.runtime.expansion.TaskProgramPlanTransformFactory[source]#

Create an optional Atomic plan transform for one grounded call.

Implementations may additionally expose a prepare_planning_context method with the same request and engine keyword arguments. The executor calls that optional hook immediately before Atomic planning, which is the ownership boundary for recipe-selected Affordance sampling.

Methods:

`__init__`(args, *kwargs)
`create_plan_transform`(request, *, engine)	Return one session-scoped transform or `None` for no change.

__init__(*args, **kwargs)#

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

Return one session-scoped transform or None for no change.

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

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

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

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

Methods:

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

__init__(atomic_adapter)#

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

Delegate one grounded Atomic plan to the common template adapter.

Parameters:

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

Return type:

TrajectoryTemplate

Returns:

Complete phase-annotated trajectory template.

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

Owned logical candidates generated for one Task Program planning call.

Methods:

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

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

snapshot()[source]#

Return a record with independently owned trajectory tensors.

Return type:: TaskProgramExpansionRecord

to_metadata()[source]#

Return JSON-compatible candidate identity and factor provenance.

Return type:: dict[str, object]

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

Apply one deterministic combined recipe to a Task Program call.

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

Methods:

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

Attributes:

records

Return expansion records in planning-call order.

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

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

Return the transform for calls selected by the episode schedule.

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

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

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

Return type:: PlanningContext

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

MLLM frontend#

embodichain.agents.mllm.decode_mllm_task_program(response, *, integration, validation_context=None, max_bytes=4194304)[source]#

Decode one untrusted model response into the canonical program config.

The model response is a single plain JSON object containing program_id, targets, and program. The trusted host supplies integration; a response attempting to select its own integration is rejected rather than silently overwritten. Only the sequential subset and curated built-in calls are admitted, and robot resource overrides are forbidden.

Parameters:

response (str) – Untrusted model response containing one plain JSON document.
integration (TaskProgramIntegrationCfg) – Host-owned scene, robot-profile, and runtime-preset choice.
validation_context (TaskProgramValidationContext | None) – Optional provider-free static reference validator.
max_bytes (int) – Maximum UTF-8 encoded response size.

Return type:

TaskProgramCfg

Returns:

An owned canonical TaskProgramCfg.

Raises:

TypeError – If integration is not an exact integration config.
embodichain.lab.task_program.TaskProgramDecodeError – If JSON, schema, or MLLM policy validation fails.

embodichain.agents.mllm.compile_mllm_task_program(response, *, adapter, integration, validation_context=None, max_bytes=4194304)[source]#

Decode and compile a model response through the existing environment path.

This function introduces no MLLM-specific compiler. It delegates the owned config to TaskProgramEnvironmentAdapter.compile(), which performs the canonical scene resolution and Task Program lowering used by every other frontend.

Parameters:

response (str) – Untrusted model response containing one plain JSON document.
adapter (TaskProgramEnvironmentAdapter) – Existing trusted Task Program environment adapter.
integration (TaskProgramIntegrationCfg) – Host-owned scene, robot-profile, and runtime-preset choice.
validation_context (TaskProgramValidationContext | None) – Optional provider-free static reference validator.
max_bytes (int) – Maximum UTF-8 encoded response size.

Return type:

CompiledTaskProgram

Returns:

Provider-free program produced by the existing Task Program compiler.

Raises:

TypeError – If adapter or integration has the wrong exact type.
embodichain.lab.task_program.TaskProgramDecodeError – If JSON, schema, or MLLM policy validation fails.

embodichain.lab.task_program

Contents

embodichain.lab.task_program#

Schema#

Loading and validation#

Compilation#

Expansion integration#

MLLM frontend#