OakEmu-RL v0.1.0

Appearance

TransitionStorage ​

python
TransitionStorage(capacity: int, observation_shape: tuple[int, ...], observation_dtype: np.dtype) -> None

Pre-allocated circular-buffer storage for replay transitions and their reward-shaping metadata.

Owns the numpy columns (states, actions, rewards, next_states, dones, n_steps, and optional reward-shaping metadata), the wraparound insert mechanics, and npz payload (de)serialization. Sampling policy (uniform vs prioritized) is intentionally not this class's concern.

Initialize the storage.

Parameters:

  • capacity (int) – Maximum number of transitions to store.
  • observation_shape (tuple[int, ...]) – Shape of the observations.
  • observation_dtype (dtype) – The numpy dtype for storing observations.

Source

Properties ​

observation_shape ​

python
observation_shape: tuple[int, ...]

Get the shape of observations stored in the buffer.

Returns:

  • tuple[int, ...] – The observation shape tuple.

Source

reward_shaping_metadata_spec ​

python
reward_shaping_metadata_spec: dict[str, np.dtype]

Get the reward-shaping metadata columns stored in the buffer.

Returns:

  • dict[str, dtype] – A mapping of metadata key to numpy dtype.

Source

Methods ​

copy_reward_shaping_metadata ​

python
copy_reward_shaping_metadata(indices: np.ndarray, reward_shaping_metadata_out: dict[str, np.ndarray] | None) -> None

Copy sampled reward-shaping metadata into caller-provided output arrays.

Parameters:

  • indices (ndarray) – Sampled replay indices.
  • reward_shaping_metadata_out (dict[str, ndarray] | None) – Output arrays keyed like the stored metadata.

Source

insert ​

python
insert(state: list[np.ndarray], action: list[int], reward: list[float], next_state: list[np.ndarray], done: list[bool], n_steps: list[int], reward_shaping_metadata: Mapping[str, np.ndarray | int | float | list[int | float]] | None) -> np.ndarray

Write a batch of transitions into the circular buffer, advancing idx/size.

Parameters:

  • state (list[ndarray]) – List of states.
  • action (list[int]) – List of actions.
  • reward (list[float]) – List of rewards.
  • next_state (list[ndarray]) – List of next states.
  • done (list[bool]) – List of done flags.
  • n_steps (list[int]) – List of step counts.
  • reward_shaping_metadata (Mapping[str, ndarray | int | float | list[int | float]] | None) – Optional compact reward-shaping metadata columns.

Returns:

  • ndarray – The absolute buffer indices written, in insertion order.

Source

load_payload ​

python
load_payload(data: Mapping[str, Any]) -> None

Restore this storage's columns from a loaded npz payload, reallocating buffers.

Parameters:

  • data (Mapping[str, Any]) – The loaded npz mapping (as returned by np.load).

Source

to_payload ​

python
to_payload() -> dict[str, object]

Build the npz-serializable payload for this storage's current contents.

Returns:

  • dict[str, object] – A mapping of npz key to array/scalar value.

Source

Values ​

actions ​

python
actions = np.zeros((capacity,), dtype=np.int64)

Source

capacity ​

python
capacity = capacity

Source

dones ​

python
dones = np.zeros((capacity,), dtype=np.bool_)

Source

idx ​

python
idx = 0

Source

n_steps ​

python
n_steps = np.ones((capacity,), dtype=np.int32)

Source

next_states ​

python
next_states = np.zeros((capacity, *observation_shape), dtype=self.observation_dtype)

Source

observation_dtype ​

python
observation_dtype = np.dtype(observation_dtype)

Source

rewards ​

python
rewards = np.zeros((capacity,), dtype=np.float32)

Source

size ​

python
size = 0

Source

states ​

python
states = np.zeros((capacity, *observation_shape), dtype=self.observation_dtype)

Source