offline

OfflineEnvironment

Bases: Environment, Protocol

Base class for offline environments.

Offline environments represent static, pre-recorded datasets collected upfront. They support the offline learning strategy where a fixed batch of data is processed at once to train a model. Offline environments can be transformed and chained together to create new datasets.

chain(*other)

Chain this offline environment with other offline environments.

Chaining offline environments will create a new incremental environment that will first observe the data from this environment and then the data from the other environments.

Parameters:

Name	Type	Description	Default
other	Environment	The other offline environments to chain.	()

Returns:

Type	Description
ChainedOfflineEnvironments	The chained offline environments.

Source code in src/flowcean/core/environment/offline.py

def chain(self, *other: Environment) -> ChainedOfflineEnvironments:
    """Chain this offline environment with other offline environments.

    Chaining offline environments will create a new incremental environment
    that will first observe the data from this environment and then the
    data from the other environments.

    Args:
        other: The other offline environments to chain.

    Returns:
        The chained offline environments.
    """
    return ChainedOfflineEnvironments([self, *other])

__add__(other)

Shorthand for chain.

Source code in src/flowcean/core/environment/offline.py

@final
def __add__(self, other: Environment) -> ChainedOfflineEnvironments:
    """Shorthand for `chain`."""
    return self.chain(other)

ChainedOfflineEnvironments(environments)

Bases: IncrementalEnvironment

Chained offline environments.

This environment chains multiple offline environments together. The environment will first observe the data from the first environment and then the data from the other environments.

Initialize the chained offline environments.

Parameters:

Name	Type	Description	Default
environments	Iterable[Environment]	The offline environments to chain.	required

Source code in src/flowcean/core/environment/offline.py

def __init__(self, environments: Iterable[Environment]) -> None:
    """Initialize the chained offline environments.

    Args:
        environments: The offline environments to chain.
    """
    self._environments = iter(environments)
    self._element = next(self._environments)