Skip to content

The environment layer — API reference

Every public symbol in skyfall_crl.env, generated from the source.

Generated page — built from the source docstrings, so this file is blank when read on GitHub. Run mkdocs serve to read it locally, or read the docstrings in the module itself.

Driving several environments at once

vector

Several independent worlds, stepped at once.

A step is a network round-trip, so throughput comes from running worlds concurrently rather than from making any one of them faster.

These environments are never reset automatically, and that is the whole point. Gymnasium's stock vector environments reset a sub-environment as soon as it reports done. Here a reset destroys the world and builds a new one, so auto-resetting at every truncation would throw away the agent's accumulated work and the financial position its reward is computed from. A world is meant to outlive the rollout windows taken from it, so truncation is reported and resetting is left to whoever is driving.

Nothing in the fan-out is MORPHEUS-specific -- it drives any gymnasium.Env whose episodes should survive truncation, which is the property that matters for a persistent environment. It lives here rather than in the core because that is where the environments are; importing it brings the adapter with it.

PersistentVectorEnv

Bases: VectorEnv

Runs N independent persistent worlds, one environment each.

Parameters

env_fns: One factory per world. Factories rather than instances because each world is configured independently -- and because building them here keeps the environments this object owns unambiguous. max_workers: Size of the thread pool. Stepping is I/O-bound, so threads are the right tool; defaults to one per environment.

envs instance-attribute 
envs: list[Env] = [make() for make in env_fns]
num_envs instance-attribute 
num_envs = len(self.envs)
single_observation_space instance-attribute 
single_observation_space = self.envs[0].observation_space
single_action_space instance-attribute 
single_action_space = self.envs[0].action_space
observation_space instance-attribute 
observation_space = batch_space(
    self.single_observation_space, self.num_envs
)
action_space instance-attribute 
action_space = batch_space(
    self.single_action_space, self.num_envs
)
metadata class-attribute instance-attribute 
metadata = {
    **self.envs[0].metadata,
    "autoreset_mode": AutoresetMode.DISABLED,
}
closed instance-attribute 
closed = False
reset 
reset(
    *,
    seed: SupportsIndex
    | Sequence[SupportsIndex]
    | None = None,
    options: dict[str, Any] | None = None,
) -> tuple[Any, dict[str, Any]]

Start an episode in every world.

step 
step(
    actions: Any,
) -> tuple[
    Any, np.ndarray, np.ndarray, np.ndarray, dict[str, Any]
]

Advance every world by one step.

No world is reset when it reports truncation: that decision belongs to the caller.

close 
close(**kwargs: Any) -> None

Release every world, each on its own terms about disposal.

The configuration service, as the environment reaches for it

regime

Which configuration a world is running under.

The service itself lives in skyfall_crl.regime, shared with the harness and the metrics rather than owned by the environment. Re-exported here because this is where the environment reaches for it.

DEFAULT_REGIME_ID module-attribute 

DEFAULT_REGIME_ID = 'baseline'

ConstantRegime

A world whose configuration never changes.

A run under this provider is stationary -- a useful control, and not a continual-learning setting.

regime_id 
regime_id(step: int) -> str
regime_at 
regime_at(step: int) -> Regime

Regime dataclass

One configuration: what it is called, how long it lasts, and how it behaves.

regime_id instance-attribute 
regime_id: str
duration_steps class-attribute instance-attribute 
duration_steps: int | None = None
axes class-attribute instance-attribute 
axes: Mapping[str, Any] = field(default_factory=dict)
origin class-attribute instance-attribute 
origin: str | None = None
identity property 
identity: str

What configuration this is, whatever it is called here.

Two entries share an identity when one repeats the other. Forgetting is measured over identities rather than labels, because meeting the same conditions twice is the whole of what it asks about.

get 
get(name: str, default: Any = None) -> Any
select 
select(names: Collection[str] | str) -> dict[str, Any]

The axes named here that this configuration actually sets.

exclude 
exclude(names: Collection[str] | str) -> dict[str, Any]

Every axis except the ones named.

Which axes an environment can be told about is the environment's business, not the schedule's, so the schedule offers the split and lets the adapter decide where it falls.

to_dict 
to_dict() -> dict[str, Any]

RegimeProvider

Bases: Protocol

Reports which configuration is active at a given step.

regime_id 
regime_id(step: int) -> str

The active configuration's name.

RegimeSchedule

The configurations a run passes through, and when.

Parameters

regimes: In order. Each needs a regime_id; duration_steps may be omitted on the last one to mean "until the run ends". An entry with alias_of repeats another configuration under a new name. shift_mode: abrupt switches at the boundary, cyclic loops back to the start, and linear_drift eases into the next configuration over drift_window steps.

shift_mode instance-attribute 
shift_mode = shift_mode
drift_window instance-attribute 
drift_window = max(int(drift_window), 1)
seed instance-attribute 
seed = int(seed)
regimes property 
regimes: tuple[Regime, ...]
total_steps property 
total_steps: int

Steps before the schedule repeats, or 0 if the last configuration runs forever.

from_dict classmethod 
from_dict(config: Mapping[str, Any]) -> RegimeSchedule
from_yaml_file classmethod 
from_yaml_file(path: str | Path) -> RegimeSchedule
regime_at 
regime_at(step: int) -> Regime

The configuration active at step.

regime_id 
regime_id(step: int) -> str

The active configuration's name -- the reporting contract the environment uses.

is_boundary 
is_boundary(step: int) -> bool

Whether step is where a configuration begins.

ScheduledRegime

Follows a schedule written in advance -- the default way to shift a world.

schedule may be a RegimeSchedule, the mapping one is built from, or the path to a YAML file holding it. The last two exist because a run is selected by configuration: {"provider": "scheduled", "params": {"schedule": ...}} builds this class directly, and a document cannot name an object. Accepting only the built form made that path silently produce a provider that raised on its first step.

schedule instance-attribute 
schedule = _as_schedule(schedule)
from_yaml_file classmethod 
from_yaml_file(path: str) -> ScheduledRegime
from_dict classmethod 
from_dict(config: Mapping[str, Any]) -> ScheduledRegime
regime_id 
regime_id(step: int) -> str
regime_at 
regime_at(step: int) -> Regime
is_boundary 
is_boundary(step: int) -> bool