embodichain.lab.gym.envs.task_program#
Gym lifecycle bridge for compiled Task Programs.
This package is the narrow Gym boundary for compiled Task Programs. It
adapts runtime output to lazy DemoSegment actions and synchronizes the
execution clock with ordinary environment steps; language, compilation,
catalogs, and simulation assembly live under
embodichain.lab.task_program.
Runner command sink that buffers actions for the Gym demo generator. |
|
Monotonic execution clock advanced only by explicit Gym steps. |
|
Raised when runtime timing cannot be represented on the Gym step grid. |
|
Callback-backed observation port that also exposes the latest qpos. |
|
Encode transport-neutral command frames to controller-ready Gym actions. |
|
Extensible lowering boundary for one runtime transport kind. |
|
Environment-aware program post-policy boundary. |
|
Environment-aware boundary for compiled program validators. |
|
Base error raised by the Task Program Gym bridge. |
|
Adapt sequential compiled program segments to lazy Gym demonstrations. |
|
Raised when a command frame names an unregistered transport. |
Classes:
Runner command sink that buffers actions for the Gym demo generator. |
|
Monotonic execution clock advanced only by explicit Gym steps. |
|
Callback-backed observation port that also exposes the latest qpos. |
|
Encode transport-neutral command frames to controller-ready Gym actions. |
|
Extensible lowering boundary for one runtime transport kind. |
|
Environment-aware program post-policy boundary. |
|
Environment-aware boundary for compiled program validators. |
|
Adapt sequential compiled program segments to lazy Gym demonstrations. |
Exceptions:
Raised when runtime timing cannot be represented on the Gym step grid. |
|
Base error raised by the Task Program Gym bridge. |
|
Raised when a command frame names an unregistered transport. |
- class embodichain.lab.gym.envs.task_program.BufferedGymCommandSink[source]
Runner command sink that buffers actions for the Gym demo generator.
Acceptance means the command was validated and copied into the local buffer; it does not claim that an environment transition already occurred.
Methods:
__init__(encoder, clock)cancel(targets, *, timeout)Discard accepted-but-not-yielded frames before a safe-stop hold.
Discard actions that were accepted locally but never yielded.
drain_safe_stop_action(*[, fallback])Select one buffered safe hold and discard every other local action.
hold(targets, context, *, timeout)Buffer one observed-position safe hold action.
pop()Pop the next accepted action and remember it as the active hold.
send(command, *, timeout)Validate, encode, and buffer one runtime command frame.
wait_hold(env_ids)Return an owned hold action for one runtime waiting step.
Attributes:
Return the monotonic count of actions accepted by this sink.
Return the exact environment-step clock used for timing checks.
Return the number of accepted actions not yet yielded to Gym.
- __init__(encoder, clock)[source]
- property accepted_action_count: int
Return the monotonic count of actions accepted by this sink.
- cancel(targets, *, timeout)[source]
Discard accepted-but-not-yielded frames before a safe-stop hold.
- Return type:
- property clock: EnvironmentStepClock
Return the exact environment-step clock used for timing checks.
- discard_pending()[source]
Discard actions that were accepted locally but never yielded.
- Return type:
None
- drain_safe_stop_action(*, fallback=None)[source]
Select one buffered safe hold and discard every other local action.
This method is used only by the demo abort handshake. A runtime acknowledgement proves local buffering, not
env.stepconsumption; therefore an interrupted generator must explicitly surface the final safe hold to the executor while dropping stale motion commands.- Return type:
ControllerAction|None
- hold(targets, context, *, timeout)[source]
Buffer one observed-position safe hold action.
- Return type:
- property pending_count: int
Return the number of accepted actions not yet yielded to Gym.
- pop()[source]
Pop the next accepted action and remember it as the active hold.
- Return type:
- send(command, *, timeout)[source]
Validate, encode, and buffer one runtime command frame.
- Return type:
- wait_hold(env_ids)[source]
Return an owned hold action for one runtime waiting step.
- Return type:
- class embodichain.lab.gym.envs.task_program.EnvironmentStepClock[source]
Monotonic execution clock advanced only by explicit Gym steps.
sleepintentionally raises. Calling synchronousSemanticCallExecutor.runwith this clock would otherwise advance execution without an environment transition. Demo integrations must use the nonblockingstart/steppath and calladvance_after_env_step()only after a yielded action was passed toenv.step.Methods:
__init__(step_dt, *[, initial_step])advance_after_env_step([steps])Advance time after
stepscompleted Gym environment transitions.now()Return deterministic environment time in seconds.
sleep(duration)Reject implicit waiting that is not backed by
env.step.steps_for_duration(duration, *[, field_name])Return an exact integer-grid representation of
duration.validate_frame(frame)Validate every row's command hold duration against the step grid.
Attributes:
Return the authoritative Gym control cadence.
Return the number of explicitly acknowledged environment steps.
- __init__(step_dt, *, initial_step=0)[source]
- advance_after_env_step(steps=1)[source]
Advance time after
stepscompleted Gym environment transitions.- Return type:
None
- now()[source]
Return deterministic environment time in seconds.
- Return type:
float
- sleep(duration)[source]
Reject implicit waiting that is not backed by
env.step.- Return type:
None
- property step_dt: float
Return the authoritative Gym control cadence.
- property step_index: int
Return the number of explicitly acknowledged environment steps.
- steps_for_duration(duration, *, field_name='duration')[source]
Return an exact integer-grid representation of
duration.Float32 command tensors receive a small ratio-space tolerance, but an incompatible cadence is never rounded or resampled.
- Return type:
int
- validate_frame(frame)[source]
Validate every row’s command hold duration against the step grid.
- Return type:
None
- exception embodichain.lab.gym.envs.task_program.EnvironmentStepTimingError[source]
Raised when runtime timing cannot be represented on the Gym step grid.
- class embodichain.lab.gym.envs.task_program.GymPlanningObservationProvider[source]
Callback-backed observation port that also exposes the latest qpos.
- Parameters:
capture (
Callable[[TaskState],PlanningContext]) – Callback accepting verifiedTaskStateand returning one freshPlanningContextfrom the Gym environment.
The callback is intentionally explicit: environment-specific scene, simulator, and registry access remains in environment integration code.
Methods:
__init__(capture)current_qpos(env_ids)Return latest full qpos rows in the requested stable-ID order.
observe(task_state)Capture and retain one fresh planning context.
- __init__(capture)[source]
- current_qpos(env_ids)[source]
Return latest full qpos rows in the requested stable-ID order.
- Return type:
Tensor
- observe(task_state)[source]
Capture and retain one fresh planning context.
- Return type:
- class embodichain.lab.gym.envs.task_program.RuntimeCommandFrameEncoder[source]
Encode transport-neutral command frames to controller-ready Gym actions.
- Parameters:
qpos_provider (
CurrentQposProvider) – Full-qpos source aligned to a frame’s explicitenv_ids.transports (
Iterable[RuntimeTransportActionEncoder]) – Optional additional transport encoders. The built-in joint-position encoder precedes them when enabled.include_joint_position (
bool) – Whether to install the built-in joint-position encoder. Standard assemblies disable it when their exact profile uses only custom endpoint transports.
Methods:
__init__(qpos_provider, *[, transports, ...])encode(frame)Encode one frame on top of a fresh full-qpos hold action.
encode_hold(targets, context)Encode an observed-position safe hold for addressed transports.
encode_idle_hold(env_ids)Return a fresh full-qpos hold when no transport was armed yet.
freeze()Permanently close transport registration for a standard assembly.
register_transport(transport, *[, replace])Register one shared transport-to-Gym action encoder.
Attributes:
Return whether runtime transport registration is permanently closed.
Return registered transport IDs in deterministic encoding order.
- __init__(qpos_provider, *, transports=(), include_joint_position=True)[source]
- encode(frame)[source]
Encode one frame on top of a fresh full-qpos hold action.
- Return type:
Union[Tensor,TensorDict[str,Tensor]]
- encode_hold(targets, context)[source]
Encode an observed-position safe hold for addressed transports.
- Return type:
Union[Tensor,TensorDict[str,Tensor]]
- encode_idle_hold(env_ids)[source]
Return a fresh full-qpos hold when no transport was armed yet.
- Return type:
Union[Tensor,TensorDict[str,Tensor]]
- freeze()[source]
Permanently close transport registration for a standard assembly.
- Return type:
None
- property is_frozen: bool
Return whether runtime transport registration is permanently closed.
- register_transport(transport, *, replace=False)[source]
Register one shared transport-to-Gym action encoder.
- Return type:
None
- property transport_ids: tuple[str, ...]
Return registered transport IDs in deterministic encoding order.
- class embodichain.lab.gym.envs.task_program.RuntimeTransportActionEncoder[source]
Extensible lowering boundary for one runtime transport kind.
An encoder receives the action produced by earlier registered transports and returns the next owned action value. This permits a future transport to promote the built-in tensor action to a
TensorDictwhen the Gym action manager exposes a structured controller boundary.Methods:
__init__(*args, **kwargs)encode(command, *, base_action, active_mask)Merge one addressed command into
base_action.hold(targets, *, base_action, context)Merge this transport's self-proven safe hold into
base_action.Attributes:
payload_typesExact runtime-payload types accepted by this encoder.
target_typesExact runtime-target types accepted by this encoder.
transport_idExact runtime transport ID handled by this encoder.
- __init__(*args, **kwargs)
- encode(command, *, base_action, active_mask)[source]
Merge one addressed command into
base_action.- Return type:
Union[Tensor,TensorDict[str,Tensor]]
- hold(targets, *, base_action, context)[source]
Merge this transport’s self-proven safe hold into
base_action.The transport remains authoritative for neutralizing its own controller; parallel command validation does not replace this transport-specific hold contract.
- Return type:
Union[Tensor,TensorDict[str,Tensor]]
-
payload_types:
ClassVar[tuple[type[RuntimeCommandPayload],...]] Exact runtime-payload types accepted by this encoder.
-
target_types:
ClassVar[tuple[type[RuntimeEndpointTarget],...]] Exact runtime-target types accepted by this encoder.
-
transport_id:
ClassVar[str] Exact runtime transport ID handled by this encoder.
- class embodichain.lab.gym.envs.task_program.SegmentPostPolicyPort[source]
Environment-aware program post-policy boundary.
Implementations may observe the environment after each resumed yield, but must return every controller action to this iterable. The bridge then routes those values through the ordinary demo executor and
env.step.Methods:
__init__(*args, **kwargs)actions(policy, *, segment, active_mask)Yield holds until
policycompletes for the active rows only.post_policy_metadata(policy, *, segment)Return JSON-safe metadata after one policy has run.
post_policy_result(policy, *, segment)Return one boolean or one boolean per environment row.
validate_policy(policy, *, segment)Validate one compiled policy without live observation or action.
- __init__(*args, **kwargs)
- actions(policy, *, segment, active_mask)[source]
Yield holds until
policycompletes for the active rows only.- Return type:
Iterable[Any]
- post_policy_metadata(policy, *, segment)[source]
Return JSON-safe metadata after one policy has run.
- Return type:
Mapping[str,Any]
- post_policy_result(policy, *, segment)[source]
Return one boolean or one boolean per environment row.
- Return type:
Any
- validate_policy(policy, *, segment)[source]
Validate one compiled policy without live observation or action.
- Return type:
None
- class embodichain.lab.gym.envs.task_program.SegmentValidatorPort[source]
Environment-aware boundary for compiled program validators.
Methods:
__init__(*args, **kwargs)validate(validator, *, segment)Return one boolean or one boolean per environment row.
validate_validator(validator, *, segment)Validate one compiled validator without observing the environment.
validator_metadata(validator, *, segment)Return JSON-safe metadata after one validator has run.
- __init__(*args, **kwargs)
- validate(validator, *, segment)[source]
Return one boolean or one boolean per environment row.
- Return type:
Any
- validate_validator(validator, *, segment)[source]
Validate one compiled validator without observing the environment.
- Return type:
None
- validator_metadata(validator, *, segment)[source]
Return JSON-safe metadata after one validator has run.
- Return type:
Mapping[str,Any]
- exception embodichain.lab.gym.envs.task_program.TaskProgramBridgeError[source]
Base error raised by the Task Program Gym bridge.
- class embodichain.lab.gym.envs.task_program.TaskProgramDemoBridge[source]
Adapt sequential compiled program segments to lazy Gym demonstrations.
- Parameters:
program (
CompiledTaskProgramPort) – Provider-free compiled Task Program.runtime (
SequentialSemanticCallExecutorPort) – Nonblocking semanticSemanticCallExecutorsurface.command_sink (
BufferedGymCommandSink) – The same buffered sink installed inruntime.clock (
EnvironmentStepClock) – The same environment-step clock installed inruntime.post_policy_port (
SegmentPostPolicyPort|None) – Optional environment-aware post-policy executor.validator_port (
SegmentValidatorPort|None) – Optional environment-aware validator executor.runner_cfg (
ExecutionRunnerCfg|None) – Runner transport policy selected by the runtime preset.parallel_safety_validator (
ParallelCommandSafetyValidator|None) – Optional authoritative physical-safety gate required before any parallel branch can start.
Parallel blocks retain their branch lanes and explicit barrier. They are lowered through
ParallelSemanticExecutor; they are never flattened into a sequential semantic-call list.Methods:
__init__(program, runtime, command_sink, ...)Lazily adapt compiled program segments to
DemoSegmentvalues.Attributes:
Return the environment-step clock used by this bridge.
Return the final row-local program acceptance mask.
Return whether every compiled segment completed its full lifecycle.
- __init__(program, runtime, command_sink, clock, *, post_policy_port=None, validator_port=None, runner_cfg=None, parallel_safety_validator=None)[source]
- property clock: EnvironmentStepClock
Return the environment-step clock used by this bridge.
- property completion_mask: Tensor
Return the final row-local program acceptance mask.
- Raises:
RuntimeError – If the segment iterator has not completed normally.
- iter_segments()[source]
Lazily adapt compiled program segments to
DemoSegmentvalues.Consumers must exhaust each segment’s actions and invoke its validator before requesting the next segment. Skipping either lifecycle boundary raises
TaskProgramBridgeErrorinstead of silently carrying stale row eligibility into downstream execution.- Return type:
Iterator[DemoSegment]
- property program_completed: bool
Return whether every compiled segment completed its full lifecycle.
- exception embodichain.lab.gym.envs.task_program.UnsupportedRuntimeTransportError[source]
Raised when a command frame names an unregistered transport.
- class embodichain.lab.gym.envs.task_program.TaskProgramDemoBridge[source]#
Adapt sequential compiled program segments to lazy Gym demonstrations.
- Parameters:
program (
CompiledTaskProgramPort) – Provider-free compiled Task Program.runtime (
SequentialSemanticCallExecutorPort) – Nonblocking semanticSemanticCallExecutorsurface.command_sink (
BufferedGymCommandSink) – The same buffered sink installed inruntime.clock (
EnvironmentStepClock) – The same environment-step clock installed inruntime.post_policy_port (
SegmentPostPolicyPort|None) – Optional environment-aware post-policy executor.validator_port (
SegmentValidatorPort|None) – Optional environment-aware validator executor.runner_cfg (
ExecutionRunnerCfg|None) – Runner transport policy selected by the runtime preset.parallel_safety_validator (
ParallelCommandSafetyValidator|None) – Optional authoritative physical-safety gate required before any parallel branch can start.
Parallel blocks retain their branch lanes and explicit barrier. They are lowered through
ParallelSemanticExecutor; they are never flattened into a sequential semantic-call list.Methods:
__init__(program, runtime, command_sink, ...)Lazily adapt compiled program segments to
DemoSegmentvalues.Attributes:
Return the environment-step clock used by this bridge.
Return the final row-local program acceptance mask.
Return whether every compiled segment completed its full lifecycle.
- __init__(program, runtime, command_sink, clock, *, post_policy_port=None, validator_port=None, runner_cfg=None, parallel_safety_validator=None)[source]#
- property clock: EnvironmentStepClock#
Return the environment-step clock used by this bridge.
- property completion_mask: Tensor#
Return the final row-local program acceptance mask.
- Raises:
RuntimeError – If the segment iterator has not completed normally.
- iter_segments()[source]#
Lazily adapt compiled program segments to
DemoSegmentvalues.Consumers must exhaust each segment’s actions and invoke its validator before requesting the next segment. Skipping either lifecycle boundary raises
TaskProgramBridgeErrorinstead of silently carrying stale row eligibility into downstream execution.- Return type:
Iterator[DemoSegment]
- property program_completed: bool#
Return whether every compiled segment completed its full lifecycle.
- class embodichain.lab.gym.envs.task_program.BufferedGymCommandSink[source]#
Runner command sink that buffers actions for the Gym demo generator.
Acceptance means the command was validated and copied into the local buffer; it does not claim that an environment transition already occurred.
Methods:
__init__(encoder, clock)cancel(targets, *, timeout)Discard accepted-but-not-yielded frames before a safe-stop hold.
Discard actions that were accepted locally but never yielded.
drain_safe_stop_action(*[, fallback])Select one buffered safe hold and discard every other local action.
hold(targets, context, *, timeout)Buffer one observed-position safe hold action.
pop()Pop the next accepted action and remember it as the active hold.
send(command, *, timeout)Validate, encode, and buffer one runtime command frame.
wait_hold(env_ids)Return an owned hold action for one runtime waiting step.
Attributes:
Return the monotonic count of actions accepted by this sink.
Return the exact environment-step clock used for timing checks.
Return the number of accepted actions not yet yielded to Gym.
- property accepted_action_count: int#
Return the monotonic count of actions accepted by this sink.
- cancel(targets, *, timeout)[source]#
Discard accepted-but-not-yielded frames before a safe-stop hold.
- Return type:
- property clock: EnvironmentStepClock#
Return the exact environment-step clock used for timing checks.
- discard_pending()[source]#
Discard actions that were accepted locally but never yielded.
- Return type:
None
- drain_safe_stop_action(*, fallback=None)[source]#
Select one buffered safe hold and discard every other local action.
This method is used only by the demo abort handshake. A runtime acknowledgement proves local buffering, not
env.stepconsumption; therefore an interrupted generator must explicitly surface the final safe hold to the executor while dropping stale motion commands.- Return type:
ControllerAction|None
- hold(targets, context, *, timeout)[source]#
Buffer one observed-position safe hold action.
- Return type:
- property pending_count: int#
Return the number of accepted actions not yet yielded to Gym.
- send(command, *, timeout)[source]#
Validate, encode, and buffer one runtime command frame.
- Return type:
- class embodichain.lab.gym.envs.task_program.EnvironmentStepClock[source]#
Monotonic execution clock advanced only by explicit Gym steps.
sleepintentionally raises. Calling synchronousSemanticCallExecutor.runwith this clock would otherwise advance execution without an environment transition. Demo integrations must use the nonblockingstart/steppath and calladvance_after_env_step()only after a yielded action was passed toenv.step.Methods:
__init__(step_dt, *[, initial_step])advance_after_env_step([steps])Advance time after
stepscompleted Gym environment transitions.now()Return deterministic environment time in seconds.
sleep(duration)Reject implicit waiting that is not backed by
env.step.steps_for_duration(duration, *[, field_name])Return an exact integer-grid representation of
duration.validate_frame(frame)Validate every row's command hold duration against the step grid.
Attributes:
Return the authoritative Gym control cadence.
Return the number of explicitly acknowledged environment steps.
- advance_after_env_step(steps=1)[source]#
Advance time after
stepscompleted Gym environment transitions.- Return type:
None
- property step_dt: float#
Return the authoritative Gym control cadence.
- property step_index: int#
Return the number of explicitly acknowledged environment steps.
- class embodichain.lab.gym.envs.task_program.GymPlanningObservationProvider[source]#
Callback-backed observation port that also exposes the latest qpos.
- Parameters:
capture (
Callable[[TaskState],PlanningContext]) – Callback accepting verifiedTaskStateand returning one freshPlanningContextfrom the Gym environment.
The callback is intentionally explicit: environment-specific scene, simulator, and registry access remains in environment integration code.
Methods:
__init__(capture)current_qpos(env_ids)Return latest full qpos rows in the requested stable-ID order.
observe(task_state)Capture and retain one fresh planning context.
- current_qpos(env_ids)[source]#
Return latest full qpos rows in the requested stable-ID order.
- Return type:
Tensor
- class embodichain.lab.gym.envs.task_program.RuntimeCommandFrameEncoder[source]#
Encode transport-neutral command frames to controller-ready Gym actions.
- Parameters:
qpos_provider (
CurrentQposProvider) – Full-qpos source aligned to a frame’s explicitenv_ids.transports (
Iterable[RuntimeTransportActionEncoder]) – Optional additional transport encoders. The built-in joint-position encoder precedes them when enabled.include_joint_position (
bool) – Whether to install the built-in joint-position encoder. Standard assemblies disable it when their exact profile uses only custom endpoint transports.
Methods:
__init__(qpos_provider, *[, transports, ...])encode(frame)Encode one frame on top of a fresh full-qpos hold action.
encode_hold(targets, context)Encode an observed-position safe hold for addressed transports.
encode_idle_hold(env_ids)Return a fresh full-qpos hold when no transport was armed yet.
freeze()Permanently close transport registration for a standard assembly.
register_transport(transport, *[, replace])Register one shared transport-to-Gym action encoder.
Attributes:
Return whether runtime transport registration is permanently closed.
Return registered transport IDs in deterministic encoding order.
- encode(frame)[source]#
Encode one frame on top of a fresh full-qpos hold action.
- Return type:
Union[Tensor,TensorDict[str,Tensor]]
- encode_hold(targets, context)[source]#
Encode an observed-position safe hold for addressed transports.
- Return type:
Union[Tensor,TensorDict[str,Tensor]]
- encode_idle_hold(env_ids)[source]#
Return a fresh full-qpos hold when no transport was armed yet.
- Return type:
Union[Tensor,TensorDict[str,Tensor]]
- freeze()[source]#
Permanently close transport registration for a standard assembly.
- Return type:
None
- property is_frozen: bool#
Return whether runtime transport registration is permanently closed.
- register_transport(transport, *, replace=False)[source]#
Register one shared transport-to-Gym action encoder.
- Return type:
None
- property transport_ids: tuple[str, ...]#
Return registered transport IDs in deterministic encoding order.
- class embodichain.lab.gym.envs.task_program.RuntimeTransportActionEncoder[source]#
Extensible lowering boundary for one runtime transport kind.
An encoder receives the action produced by earlier registered transports and returns the next owned action value. This permits a future transport to promote the built-in tensor action to a
TensorDictwhen the Gym action manager exposes a structured controller boundary.Methods:
- __init__(*args, **kwargs)#
- __new__(**kwargs)#
- class embodichain.lab.gym.envs.task_program.SegmentPostPolicyPort[source]#
Environment-aware program post-policy boundary.
Implementations may observe the environment after each resumed yield, but must return every controller action to this iterable. The bridge then routes those values through the ordinary demo executor and
env.step.Methods:
- __init__(*args, **kwargs)#
- __new__(**kwargs)#
- class embodichain.lab.gym.envs.task_program.SegmentValidatorPort[source]#
Environment-aware boundary for compiled program validators.
Methods:
- __init__(*args, **kwargs)#
- __new__(**kwargs)#
- class embodichain.lab.gym.envs.task_program.TaskProgramBridgeError[source]#
Base error raised by the Task Program Gym bridge.
Methods:
- __init__(*args, **kwargs)#
- __new__(**kwargs)#