OakEmu-RL v0.1.0

Appearance

TransitionPipeline ​

python
TransitionPipeline(reward_shaping: RewardShapingPipeline, transition_accumulator: TransitionAccumulator, replay_writer: ReplayWriter, submission_lock: RLock, *, initial_steps: int = 0) -> None

Turns environment step results into shaped, accumulated transitions submitted to the replay buffer, and owns the environment-step counter that advances on every processed step.

Every method that touches shared staging state (the n-step buffer, the transition accumulator, the step counter) runs under a single submission lock, which is also handed to the trainer's TrajectoryEpisodeCoordinator -- holding it (via freeze) blocks both replay submission and trajectory-store writes, which is what makes a DQN checkpoint save consistent with the actor thread still running.

Initialize the pipeline.

Parameters:

  • reward_shaping (RewardShapingPipeline) – The actor-side reward-shaping pipeline.
  • transition_accumulator (TransitionAccumulator) – Accumulates n-step transitions into replay batches.
  • replay_writer (ReplayWriter) – Dedicated replay-writer runtime for replay inserts.
  • submission_lock (RLock) – Lock shared with the trainer's trajectory runtime.
  • initial_steps (int) – The step count to resume from (for example after a checkpoint restore).

Source

Properties ​

replay_queue_depth ​

python
replay_queue_depth: int

Get the number of replay batches waiting to be inserted.

Returns:

  • int – The replay queue depth.

Source

staged_transition_count ​

python
staged_transition_count: int

Get the number of transitions currently staged, not yet submitted to the replay buffer.

Returns:

  • int – The number of locally buffered transitions.

Source

steps ​

python
steps: int

Get the total number of environment steps processed.

Returns:

  • int – The total step count.

Source

Methods ​

flush ​

python
flush() -> None

Flush any buffered transitions to the replay buffer.

Source

freeze ​

python
freeze() -> Generator[None]

Freeze replay submission, flush actor-side staging, and drain the replay writer.

Returns:

  • Generator[None] – A context manager that keeps replay submission frozen for the duration of the block.

Source

process_step_result ​

python
process_step_result(result: MultiEnvResult, env_state: EnvironmentState) -> bool

Shape the step's reward, accumulate/submit the resulting transitions, and advance state.

Parameters:

  • result (MultiEnvResult) – The environment step result.
  • env_state (EnvironmentState) – The state for this environment.

Returns:

  • bool – True if the episode is now done (terminated or truncated).

Source

replace_replay_buffer ​

python
replace_replay_buffer(replay_buffer: ReplayBuffer) -> None

Rebind the replay writer to a new replay buffer.

Parameters:

  • replay_buffer (ReplayBuffer) – The new replay buffer.

Source

restore_steps ​

python
restore_steps(steps: int) -> None

Restore the step counter from a checkpoint.

Parameters:

  • steps (int) – The step count to resume from.

Source