Core Utilities API¶

Core APIs cover persistence, shared logging, hidden episode caps, rendering, and experiment logging.

Persistence¶

src.core.io.create_dir ¶

create_dir(path: str)

src.core.io.save_to_file ¶

save_to_file(obj, dir, filename)

src.core.io.load_from_file ¶

load_from_file(dir, filename)

src.core.io.load_from_file_if_exists ¶

load_from_file_if_exists(
    directory: str | Path,
    filename: str,
    default: Any = None,
) -> Any

Logging¶

src.core.logging.configure_logging ¶

configure_logging(
    *,
    min_level: str | int = INFO,
    logs_dir: str | Path | None = None,
    console_enabled: bool = True,
    file_enabled: bool = True,
    console_sink: TextIO | None = None,
    log_format: str = DEFAULT_LOG_FORMAT,
) -> None

Configure shared logging sinks for the current interpreter.

src.core.logging.log ¶

log(
    message: object,
    *,
    type: str | None = None,
    priority: str | int = INFO,
    console: bool | None = None,
    file: bool | None = None,
    exception: BaseException | bool | None = None,
) -> None

Write a message to the shared console/file logger.

Episode and Video Wrappers¶

src.core.wrappers.infer_episode_cap ¶

infer_episode_cap(env: Any) -> int | None

Find the hidden max_steps episode cap on an env/wrapper stack.

src.core.wrappers.apply_hidden_episode_cap ¶

apply_hidden_episode_cap(
    *,
    active_agents: Iterable[str],
    terminations: dict[str, bool],
    truncations: dict[str, bool],
    episode_steps: int,
    episode_cap: int | None,
) -> tuple[dict[str, bool], bool]

Return updated truncations plus whether the hidden cap fired this step.

src.core.wrappers.record_video_parallel ¶

RecordVideo ¶

RecordVideo(
    env: ParallelEnv,
    video_folder: str,
    episode_trigger: Callable[[int], bool] | None = None,
    step_trigger: Callable[[int], bool] | None = None,
    video_length: int = 0,
    name_prefix: str = "rl-video",
    fps: int | None = None,
    disable_logger: bool = True,
    gc_trigger: Callable[[int], bool]
    | None = lambda episode: True,
)

Bases: BaseParallelWrapper

Wraps a Parallel environment with to output interval-based recordings.

Parameters:

Name	Type	Description	Default
`env`	`ParallelEnv`	The parallel environment that will be trapped.	required
`video_folder`	`str`	The folder where the recordings will be stored.	required
`episode_trigger`	`Callable[[int], bool] \| None`	Function that accepts an integer and returns `True` to start recording an episode.	`None`
`step_trigger`	`Callable[[int], bool] \| None`	Function that accepts an integer that should return `True` on the n-th environment step that the recording should be started, where n sums over all previous episodes.	`None`
`video_length`	`int`	The length of recorded episodes. If 0, entire episodes are recorded. Otherwise, snippets of the specified length are captured.	`0`
`name_prefix`	`str`	Will be prepended to the filename of the recordings. Defaults to "rl-video".	`'rl-video'`
`fps`	`int \| None`	The frame per second in the video. Provides a custom video fps for environment, if `None` then the environment metadata `render_fps` key is used if it exists, otherwise a default value of 30 is used.	`None`
`disable_logger`	`bool`	Whether to disable moviepy logger or not, default it is disabled	`True`
`gc_trigger`	`_type_`	Function that accepts an integer and returns `True` iff garbage collection should be performed after this episode	`lambda episode: True`

Raises:

Type	Description
`ValueError`	If the render_mode is not suitable, an image must be returned per render step
`DependencyNotInstalled`	If `MoviePy` is not installed

reset ¶

reset(
    seed: int | None = None, options: dict | None = None
) -> tuple[dict[AgentID, ObsType], dict[AgentID, dict]]

Reset the environment and eventually starts a new recording.

step ¶

step(
    actions: dict[AgentID, ActionType],
) -> tuple[
    dict[AgentID, ObsType],
    dict[AgentID, float],
    dict[AgentID, bool],
    dict[AgentID, bool],
    dict[AgentID, dict],
]

Steps through the environment using actions, recording frames if self.recording.

render ¶

render() -> RenderFrame | list[RenderFrame]

Compute the render frames as specified by render_mode attribute during initialization of the environment.

close ¶

close()

Closes the wrapper then the video recorder.

start_recording ¶

start_recording(video_name: str)

Start a new recording. If it is already recording, stops the current recording before starting the new one.

stop_recording ¶

stop_recording()

Stop current recording and saves the video.

del ¶

__del__()

Warn the user in case last video wasn't saved.

Experiment Configuration¶

src.core.wandb.create_base_config ¶

create_base_config(
    env_name: str,
    num_envs: int,
    n_agents: int,
    timesteps: int,
    num_runs: int,
    rollouts: int,
    learning_epochs: int,
    mini_batches: int,
    learning_rate: float,
    max_risk: float,
    use_wandb: bool,
    shield_patience: int = 64,
    shield_opponent_mode: str = "monotone_floor",
    shield_reward_mode: str = "bayesian",
    shield_credible_delta: float = 0.05,
    shield_credible_schedule: str = "summable",
    experiment_tag: str = "",
    seed_offset: int = 0,
    wm_version: str = "v0",
    om_version: str = "v0",
    ippo_version: str = "v0",
    lag_version: str = "v0",
    shield_version: str = "v0",
    final_policy_eval_episodes: int = 0,
    final_policy_eval_horizons: tuple[int, ...] = (
        1,
        10,
        50,
        100,
        200,
    ),
    final_policy_eval_seed_offset: int = 1000000,
    final_policy_eval_confidence: float = 0.95,
    **overrides: Any,
) -> dict

Create a base config dict with required values, allowing arbitrary overrides/additions via **overrides.

src.core.wandb.start_run ¶

start_run(
    algo: str,
    run_idx: int,
    base_cfg: dict,
    extra_cfg: dict | None = None,
)

src.core.wandb.log_history ¶

log_history(run, history, prefix: str = '')

Supports

list[EpisodeLog]
list[dict]-like
dict of lists

For EpisodeLog

logs one summary point per episode at step = t_end.