embodichain.lab.gym.envs.task_program

Contents

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.

BufferedGymCommandSink

Runner command sink that buffers actions for the Gym demo generator.

EnvironmentStepClock

Monotonic execution clock advanced only by explicit Gym steps.

EnvironmentStepTimingError

Raised when runtime timing cannot be represented on the Gym step grid.

GymPlanningObservationProvider

Callback-backed observation port that also exposes the latest qpos.

RuntimeCommandFrameEncoder

Encode transport-neutral command frames to controller-ready Gym actions.

RuntimeTransportActionEncoder

Extensible lowering boundary for one runtime transport kind.

SegmentPostPolicyPort

Environment-aware program post-policy boundary.

SegmentValidatorPort

Environment-aware boundary for compiled program validators.

TaskProgramBridgeError

Base error raised by the Task Program Gym bridge.

TaskProgramDemoBridge

Adapt sequential compiled program segments to lazy Gym demonstrations.

UnsupportedRuntimeTransportError

Raised when a command frame names an unregistered transport.

Classes:

BufferedGymCommandSink

Runner command sink that buffers actions for the Gym demo generator.

EnvironmentStepClock

Monotonic execution clock advanced only by explicit Gym steps.

GymPlanningObservationProvider

Callback-backed observation port that also exposes the latest qpos.

RuntimeCommandFrameEncoder

Encode transport-neutral command frames to controller-ready Gym actions.

RuntimeTransportActionEncoder

Extensible lowering boundary for one runtime transport kind.

SegmentPostPolicyPort

Environment-aware program post-policy boundary.

SegmentValidatorPort

Environment-aware boundary for compiled program validators.

TaskProgramDemoBridge

Adapt sequential compiled program segments to lazy Gym demonstrations.

Exceptions:

EnvironmentStepTimingError

Raised when runtime timing cannot be represented on the Gym step grid.

TaskProgramBridgeError

Base error raised by the Task Program Gym bridge.

UnsupportedRuntimeTransportError

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_pending()

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:

accepted_action_count

Return the monotonic count of actions accepted by this sink.

clock

Return the exact environment-step clock used for timing checks.

pending_count

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:

CommandAcknowledgement

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.step consumption; 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:

CommandAcknowledgement

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:

ControllerAction

send(command, *, timeout)[source]

Validate, encode, and buffer one runtime command frame.

Return type:

CommandAcknowledgement

wait_hold(env_ids)[source]

Return an owned hold action for one runtime waiting step.

Return type:

ControllerAction

class embodichain.lab.gym.envs.task_program.EnvironmentStepClock[source]

Monotonic execution clock advanced only by explicit Gym steps.

sleep intentionally raises. Calling synchronous SemanticCallExecutor.run with this clock would otherwise advance execution without an environment transition. Demo integrations must use the nonblocking start/step path and call advance_after_env_step() only after a yielded action was passed to env.step.

Methods:

__init__(step_dt, *[, initial_step])

advance_after_env_step([steps])

Advance time after steps completed 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:

step_dt

Return the authoritative Gym control cadence.

step_index

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 steps completed 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 verified TaskState and returning one fresh PlanningContext from 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:

PlanningContext

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 explicit env_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:

is_frozen

Return whether runtime transport registration is permanently closed.

transport_ids

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 TensorDict when 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_types

Exact runtime-payload types accepted by this encoder.

target_types

Exact runtime-target types accepted by this encoder.

transport_id

Exact 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 policy completes 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 policy completes 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 semantic SemanticCallExecutor surface.

  • command_sink (BufferedGymCommandSink) – The same buffered sink installed in runtime.

  • clock (EnvironmentStepClock) – The same environment-step clock installed in runtime.

  • 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, ...)

iter_segments()

Lazily adapt compiled program segments to DemoSegment values.

Attributes:

clock

Return the environment-step clock used by this bridge.

completion_mask

Return the final row-local program acceptance mask.

program_completed

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 DemoSegment values.

Consumers must exhaust each segment’s actions and invoke its validator before requesting the next segment. Skipping either lifecycle boundary raises TaskProgramBridgeError instead 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 semantic SemanticCallExecutor surface.

  • command_sink (BufferedGymCommandSink) – The same buffered sink installed in runtime.

  • clock (EnvironmentStepClock) – The same environment-step clock installed in runtime.

  • 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, ...)

iter_segments()

Lazily adapt compiled program segments to DemoSegment values.

Attributes:

clock

Return the environment-step clock used by this bridge.

completion_mask

Return the final row-local program acceptance mask.

program_completed

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 DemoSegment values.

Consumers must exhaust each segment’s actions and invoke its validator before requesting the next segment. Skipping either lifecycle boundary raises TaskProgramBridgeError instead 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_pending()

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:

accepted_action_count

Return the monotonic count of actions accepted by this sink.

clock

Return the exact environment-step clock used for timing checks.

pending_count

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:

CommandAcknowledgement

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.step consumption; 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:

CommandAcknowledgement

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:

ControllerAction

send(command, *, timeout)[source]#

Validate, encode, and buffer one runtime command frame.

Return type:

CommandAcknowledgement

wait_hold(env_ids)[source]#

Return an owned hold action for one runtime waiting step.

Return type:

ControllerAction

class embodichain.lab.gym.envs.task_program.EnvironmentStepClock[source]#

Monotonic execution clock advanced only by explicit Gym steps.

sleep intentionally raises. Calling synchronous SemanticCallExecutor.run with this clock would otherwise advance execution without an environment transition. Demo integrations must use the nonblocking start/step path and call advance_after_env_step() only after a yielded action was passed to env.step.

Methods:

__init__(step_dt, *[, initial_step])

advance_after_env_step([steps])

Advance time after steps completed 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:

step_dt

Return the authoritative Gym control cadence.

step_index

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 steps completed 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

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 verified TaskState and returning one fresh PlanningContext from 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:

PlanningContext

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 explicit env_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:

is_frozen

Return whether runtime transport registration is permanently closed.

transport_ids

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 TensorDict when the Gym action manager exposes a structured controller boundary.

Methods:

__init__(*args, **kwargs)

__new__(**kwargs)

__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)

__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)

__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)

__init__(*args, **kwargs)#
__new__(**kwargs)#
class embodichain.lab.gym.envs.task_program.EnvironmentStepTimingError[source]#

Raised when runtime timing cannot be represented on the Gym step grid.

Methods:

__init__(*args, **kwargs)

__new__(**kwargs)

__init__(*args, **kwargs)#
__new__(**kwargs)#
class embodichain.lab.gym.envs.task_program.UnsupportedRuntimeTransportError[source]#

Raised when a command frame names an unregistered transport.

Methods:

__init__(*args, **kwargs)

__new__(**kwargs)

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