offline

OfflineEnvironment()

Bases: TransformedObservable

Base class for offline environments.

Offline environments are used to represent datasets. They can be used to represent static datasets. Offline environments can be transformed and joined together to create new datasets.

Initialize the offline environment.

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

def __init__(self) -> None:
    """Initialize the offline environment."""
    super().__init__()

join(other)

Join this offline environment with another one.

Parameters:

Name	Type	Description	Default
other	OfflineEnvironment	The other offline environment to join.	required

Returns:

Type	Description
JoinedOfflineEnvironment	The joined offline environment.

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

def join(self, other: OfflineEnvironment) -> JoinedOfflineEnvironment:
    """Join this offline environment with another one.

    Args:
        other: The other offline environment to join.

    Returns:
        The joined offline environment.
    """
    return JoinedOfflineEnvironment([self, other])

__and__(other)

Shorthand for join.

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

def __and__(self, other: OfflineEnvironment) -> JoinedOfflineEnvironment:
    """Shorthand for `join`."""
    return self.join(other)

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	OfflineEnvironment	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: OfflineEnvironment) -> 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.
    """
    from flowcean.core.environment.chained import (
        ChainedOfflineEnvironments,
    )

    return ChainedOfflineEnvironments([self, *other])

__add__(other)

Shorthand for chain.

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

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

write_parquet(path)

Write the environment to a parquet file at the specified path.

Write the environment to a parquet file. Use a ParquetDataLoader to load the data back into flowcean as an environment.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to the parquet file where the data is written.	required

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

def write_parquet(self, path: Path | str) -> None:
    """Write the environment to a parquet file at the specified path.

    Write the environment to a parquet file. Use a `ParquetDataLoader` to
    load the data back into flowcean as an environment.

    Args:
        path: Path to the parquet file where the data is written.
    """
    self.observe().write_parquet(Path(path).with_suffix(".parquet"))

as_stream(batch_size)

Convert the offline environment to a streaming environment.

Parameters:

Name	Type	Description	Default
batch_size	int	The batch size of the streaming environment.	required

Returns:

Type	Description
StreamingOfflineEnvironment	The streaming environment.

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

def as_stream(self, batch_size: int) -> StreamingOfflineEnvironment:
    """Convert the offline environment to a streaming environment.

    Args:
        batch_size: The batch size of the streaming environment.

    Returns:
        The streaming environment.
    """
    from flowcean.environments.streaming import (
        StreamingOfflineEnvironment,
    )

    return StreamingOfflineEnvironment(self, batch_size)

JoinedOfflineEnvironment(environments)

Bases: OfflineEnvironment

Environment that joins multiple offline environments.

Attributes:

Name	Type	Description
environments	Iterable[OfflineEnvironment]	The offline environments to join.

Initialize the joined offline environment.

Parameters:

Name	Type	Description	Default
environments	Iterable[OfflineEnvironment]	The offline environments to join.	required

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

def __init__(self, environments: Iterable[OfflineEnvironment]) -> None:
    """Initialize the joined offline environment.

    Args:
        environments: The offline environments to join.
    """
    self.environments = environments
    super().__init__()