Utilities

Environment creation

create_eval_envs

create_eval_envs(
    num_threads: int,
    num_envs_per_thread: int,
    env_seed: int,
    gym_id: str,
    wrappers: list[Wrapper] | None = None,
    vecenv_cls: type = AsyncVectorEnv,
    **gym_kwargs: Any,
) -> list[Env]

Create vectorized evaluation environments in the format expected by Evaluator.

Builds num_threads vectorized environments, each bundling num_envs_per_thread copies of gym_id. Each individual environment is seeded with a unique, incrementing seed derived from env_seed, and wrapped with the given wrappers.

Parameters:

Name	Type	Description	Default
num_threads	int	Number of vectorized environments to create (one per evaluation thread).	required
num_envs_per_thread	int	Number of parallel environments within each vectorized environment.	required
env_seed	int	Base seed. Environment i is seeded with env_seed + i.	required
gym_id	str	The Gymnasium environment ID passed to gym.make.	required
wrappers	list[Wrapper] | None	Gymnasium wrappers applied to each environment; first = innermost.	None
vecenv_cls	type	The vectorized-environment class to use (e.g. gym.vector.AsyncVectorEnv, gym.vector.SyncVectorEnv, or a Stable-Baselines3 VecEnv).	AsyncVectorEnv
**gym_kwargs	Any	Extra keyword arguments forwarded to gym.make (and thus to the environment), e.g. max_episode_steps.	{}

Returns:

Type	Description
list[Env]	A list of num_threads vectorized environments, ready to pass to Evaluator(env=...).

Source code in pydsmc/utils.py

def create_eval_envs(
    num_threads: int,
    num_envs_per_thread: int,
    env_seed: int,
    gym_id: str,
    wrappers: list[gym.Wrapper] | None = None,
    vecenv_cls: type = gym.vector.AsyncVectorEnv,  # Tested: gym.VectorEnv, sb3.VecEnv. gym VecEnvs recommended
    **gym_kwargs: Any,
) -> list[gym.Env]:
    """Create vectorized evaluation environments in the format expected by [`Evaluator`][pydsmc.Evaluator].

    Builds `num_threads` vectorized environments, each bundling `num_envs_per_thread` copies of
    `gym_id`. Each individual environment is seeded with a unique, incrementing seed derived from
    `env_seed`, and wrapped with the given `wrappers`.

    Args:
        num_threads: Number of vectorized environments to create (one per evaluation thread).
        num_envs_per_thread: Number of parallel environments within each vectorized environment.
        env_seed: Base seed. Environment `i` is seeded with `env_seed + i`.
        gym_id: The Gymnasium environment ID passed to `gym.make`.
        wrappers: Gymnasium wrappers applied to each environment; first = innermost.
        vecenv_cls: The vectorized-environment class to use (e.g. `gym.vector.AsyncVectorEnv`,
            `gym.vector.SyncVectorEnv`, or a Stable-Baselines3 `VecEnv`).
        **gym_kwargs: Extra keyword arguments forwarded to `gym.make` (and thus to the environment),
            e.g. `max_episode_steps`.

    Returns:
        A list of `num_threads` vectorized environments, ready to pass to `Evaluator(env=...)`.
    """
    if wrappers is None:
        wrappers = []

    def make_env(env_id, wrap_list=None):
        if wrap_list is None:
            wrap_list = []

        def _thunk():
            env = gym.make(gym_id, **gym_kwargs)
            for w in wrap_list:
                env = w(env)
            env.reset(
                seed=env_seed + env_id,
            )  # Seeding only necessary for SB3 I think, doesn't hurt either way
            return env

        return _thunk

    # create the environment
    envs = []
    for i in range(num_threads):
        env_fns = [
            make_env(i * num_envs_per_thread + j, wrappers) for j in range(num_envs_per_thread)
        ]
        env = vecenv_cls(
            env_fns,
        )  # Eval seems to be faster with gym vector envs in preliminary tests
        envs.append(env)

    return envs

Reading results

jsons_to_df

jsons_to_df(
    log_dir: Path | str | PathLike,
    save_path: Path | str | PathLike | None = None,
    *,
    include_timeseries: bool = False,
    first_converged_row: bool = False,
    save: bool = True,
    only_latest: bool = True,
) -> DataFrame

Aggregate the JSON-Lines logs of an evaluation run into a pandas DataFrame.

Reads the per-property results.jsonl files written under log_dir, merges them with the run and property settings, and returns the combined results. Optionally also writes the aggregated results to disk.

Parameters:

Name	Type	Description	Default
log_dir	Path | str | PathLike	Directory containing an evaluation run's logs (the run's settings.json and the per-property subdirectories).	required
save_path	Path | str | PathLike | None	Where to write the aggregated results. Defaults to log_dir. If it is a directory, a results.jsonl is written inside it; a .csv suffix writes CSV, otherwise JSON.	None
include_timeseries	bool	Include every intermediate result (the full time series) rather than just the final row per property.	False
first_converged_row	bool	Keep the first row where the interval converged (falling back to the last available row if none converged). Ignored if include_timeseries is set.	False
save	bool	Whether to write the aggregated results to save_path.	True
only_latest	bool	Only read the properties referenced in the run's settings.json (the latest run), rather than every property subdirectory present.	True

Returns:

Type	Description
DataFrame	A DataFrame with one row per property (or per intermediate result when include_timeseries is set).

Source code in pydsmc/json_parser.py

def jsons_to_df(
    log_dir: Path | str | os.PathLike,
    save_path: Path | str | os.PathLike | None = None,
    *,
    include_timeseries: bool = False,
    first_converged_row: bool = False,
    save: bool = True,
    only_latest: bool = True,
) -> pd.DataFrame:
    """Aggregate the JSON-Lines logs of an evaluation run into a pandas `DataFrame`.

    Reads the per-property `results.jsonl` files written under `log_dir`, merges them with the run
    and property settings, and returns the combined results. Optionally also writes the aggregated
    results to disk.

    Args:
        log_dir: Directory containing an evaluation run's logs (the run's `settings.json` and the
            per-property subdirectories).
        save_path: Where to write the aggregated results. Defaults to `log_dir`. If it is a
            directory, a `results.jsonl` is written inside it; a `.csv` suffix writes CSV, otherwise
            JSON.
        include_timeseries: Include every intermediate result (the full time series) rather than
            just the final row per property.
        first_converged_row: Keep the first row where the interval converged (falling back to the
            last available row if none converged). Ignored if `include_timeseries` is set.
        save: Whether to write the aggregated results to `save_path`.
        only_latest: Only read the properties referenced in the run's `settings.json` (the latest
            run), rather than every property subdirectory present.

    Returns:
        A `DataFrame` with one row per property (or per intermediate result when `include_timeseries` is set).
    """
    latest_results = __read_run_results(
        log_dir,
        include_timeseries=include_timeseries,
        first_converged_row=first_converged_row,
        only_latest=only_latest,
    )
    df = pd.DataFrame(latest_results)
    if save:
        save_path = save_path or log_dir
        save_path = Path(save_path)
        if save_path.is_dir():
            save_path = save_path / "results.jsonl"
        if save_path.name.endswith(".csv"):
            df.to_csv(save_path, index=False)
        else:
            with save_path.open("w") as f:
                json.dump(latest_results, f, indent=4)

    return df