core

Actable

Bases: Generic[Action], ABC

Base class for active environments.

Active environments require actions to be taken to advance.

act(action) abstractmethod

Act on the environment.

Parameters:

Name	Type	Description	Default
action	Action	The action to perform.	required

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

@abstractmethod
def act(self, action: Action) -> None:
    """Act on the environment.

    Args:
        action: The action to perform.
    """

ActiveEnvironment()

Bases: TransformedObservable, Stepable, Actable[Data]

Base class for active environments.

An active environment loads data in an interactive way, e.g., from a simulation or real system. The environment requires actions to be taken to advance. Data can be retrieved by observing the environment.

Initialize the active environment.

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

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

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[OfflineEnvironment]	The offline environments to chain.	required

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

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

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

IncrementalEnvironment()

Bases: TransformedObservable, Stepable, Iterable[Data]

Base class for incremental environments.

Incremental environments are environments that can be advanced by a step and provide a stream of data. The data can be observed at each step.

Initialize the incremental environment.

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

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

num_steps()

Return the number of steps in the environment.

Returns:

Type	Description
int | None	The number of steps in the environment, or None if the number of
int | None	steps is unknown.

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

def num_steps(self) -> int | None:
    """Return the number of steps in the environment.

    Returns:
        The number of steps in the environment, or None if the number of
        steps is unknown.
    """
    return None

Observable

Bases: Protocol

Protocol for observations.

observe() abstractmethod

Observe and return the observation.

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

@abstractmethod
def observe(self) -> Data:
    """Observe and return the observation."""
    raise NotImplementedError

TransformedObservable()

Bases: Observable

Base class for observations that carry a transform.

Attributes:

Name	Type	Description
transform	Transform	Transform

Initialize the observable.

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

def __init__(self) -> None:
    """Initialize the observable."""
    super().__init__()
    self.transform = Identity()

with_transform(transform)

Append a transform to the observation.

Parameters:

Name	Type	Description	Default
transform	Transform	Transform to append.	required

Returns:

Type	Description
Self	This observable with the appended transform.

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

def with_transform(
    self,
    transform: Transform,
) -> Self:
    """Append a transform to the observation.

    Args:
        transform: Transform to append.

    Returns:
        This observable with the appended transform.
    """
    self.transform = self.transform.chain(transform)
    return self

_observe() abstractmethod

Observe and return the observation without applying the transform.

This method must be implemented by subclasses.

Returns:

Type	Description
Data	The raw observation.

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

@abstractmethod
def _observe(self) -> Data:
    """Observe and return the observation without applying the transform.

    This method must be implemented by subclasses.

    Returns:
        The raw observation.
    """

__or__(transform)

Shortcut for with_transform.

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

def __or__(
    self,
    transform: Transform,
) -> Self:
    """Shortcut for `with_transform`."""
    return self.with_transform(transform)

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__()

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)

Finished

Bases: Exception

Exception raised when the environment is finished.

This exception is raised when the environment is finished, and no more data can be retrieved.

Stepable

Bases: ABC

Base class for stepable environments.

Stepable environments are environments that can be advanced by a step. Usually, this is combined with an observable to provide a stream of data.

step() abstractmethod

Advance the environment by one step.

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

@abstractmethod
def step(self) -> None:
    """Advance the environment by one step."""

ActiveLearner

Bases: ABC

Base class for active learners.

Active learners require actions to be taken to learn.

learn_active(action, observation) abstractmethod

Learn from actions and observations.

Parameters:

Name	Type	Description	Default
action	Data	The action performed.	required
observation	Data	The observation of the environment.	required

Returns:

Type	Description
Model	The model learned from the data.

Source code in src/flowcean/core/learner.py

@abstractmethod
def learn_active(self, action: Data, observation: Data) -> Model:
    """Learn from actions and observations.

    Args:
        action: The action performed.
        observation: The observation of the environment.

    Returns:
        The model learned from the data.
    """

propose_action(observation) abstractmethod

Propose an action based on an observation.

Parameters:

Name	Type	Description	Default
observation	Data	The observation of an environment.	required

Returns:

Type	Description
Data	The action to perform.

Source code in src/flowcean/core/learner.py

@abstractmethod
def propose_action(self, observation: Data) -> Data:
    """Propose an action based on an observation.

    Args:
        observation: The observation of an environment.

    Returns:
        The action to perform.
    """

SupervisedIncrementalLearner

Bases: ABC

Base class for incremental supervised learners.

An incremental supervised learner learns from input-output pairs incrementally.

learn_incremental(inputs, outputs) abstractmethod

Learn from the data incrementally.

Parameters:

Name	Type	Description	Default
inputs	Data	The input data.	required
outputs	Data	The output data.	required

Returns:

Type	Description
Model	The model learned from the data.

Source code in src/flowcean/core/learner.py

@abstractmethod
def learn_incremental(self, inputs: Data, outputs: Data) -> Model:
    """Learn from the data incrementally.

    Args:
        inputs: The input data.
        outputs: The output data.

    Returns:
        The model learned from the data.
    """

SupervisedLearner

Bases: ABC

Base class for supervised learners.

A supervised learner learns from input-output pairs.

learn(inputs, outputs) abstractmethod

Learn from the data.

Parameters:

Name	Type	Description	Default
inputs	Data	The input data.	required
outputs	Data	The output data.	required

Returns:

Type	Description
Model	The model learned from the data.

Source code in src/flowcean/core/learner.py

@abstractmethod
def learn(self, inputs: Data, outputs: Data) -> Model:
    """Learn from the data.

    Args:
        inputs: The input data.
        outputs: The output data.

    Returns:
        The model learned from the data.
    """

OfflineMetric

Bases: ABC

Base class for metrics.

name property

Return the name of the metric.

Returns:

Type	Description
str	The name of the metric.

__call__(true, predicted) abstractmethod

Calculate the metric value for given true and predicted labels.

Parameters:

Name	Type	Description	Default
true	Data	True labels	required
predicted	Data	Predicted labels	required

Returns:

Type	Description
Reportable	Metric value

Source code in src/flowcean/core/metric.py

@abstractmethod
def __call__(self, true: Data, predicted: Data) -> Reportable:
    """Calculate the metric value for given true and predicted labels.

    Args:
        true: True labels
        predicted: Predicted labels

    Returns:
        Metric value
    """

Model

Bases: ABC

Base class for models.

A model is used to predict outputs for given inputs.

predict(input_features) abstractmethod

Predict outputs for the given inputs.

Parameters:

Name	Type	Description	Default
input_features	Data	The inputs for which to predict the outputs.	required

Returns:

Type	Description
Data	The predicted outputs.

Source code in src/flowcean/core/model.py

@abstractmethod
def predict(self, input_features: Data) -> Data:
    """Predict outputs for the given inputs.

    Args:
        input_features: The inputs for which to predict the outputs.

    Returns:
        The predicted outputs.
    """

save(file)

Save the model to the file.

This method can be used to save a flowcean model to a file or a file-like object. To save a model to a file use

with open("model.fml", "wb") as f:
    model.save(f)

The resulting file will contain the model any any attached transforms. It can be loaded again using the load method from the Model class.

Parameters:

Name	Type	Description	Default
file	Path | BinaryIO	The file like object to save the model to.	required

Source code in src/flowcean/core/model.py

@final
def save(self, file: Path | BinaryIO) -> None:
    """Save the model to the file.

    This method can be used to save a flowcean model to a file or a
    file-like object. To save a model to a file use

    ```python
    with open("model.fml", "wb") as f:
        model.save(f)
    ```

    The resulting file will contain the model any any attached transforms.
    It can be loaded again using the `load` method from the `Model` class.

    Args:
        file: The file like object to save the model to.
    """
    data = {
        "model": self.save_state(),
        "model_type": fullname(self),
    }
    if isinstance(file, Path):
        with file.open("wb") as f:
            pickle.dump(data, f)
    else:
        pickle.dump(data, file)

load(file) staticmethod

Load a model from file.

This method can be used to load a previously saved flowcean model from a file or a file-like object. To load a model from a file use

with open("model.fml", "rb") as f:
    model = Model.load(f)

The load method will automatically determine the model type and and any attached transforms and will load them into the correct model class.

As this method uses the pickle module to load the model, it is not safe to load models from untrusted sources as this could lead to arbitrary code execution!

Parameters:

Name	Type	Description	Default
file	Path | BinaryIO	The file like object to load the model from.	required

Source code in src/flowcean/core/model.py

@staticmethod
def load(file: Path | BinaryIO) -> Model:
    """Load a model from file.

    This method can be used to load a previously saved flowcean model from
    a file or a file-like object.
    To load a model from a file use

    ```python
    with open("model.fml", "rb") as f:
        model = Model.load(f)
    ```

    The `load` method will automatically determine the model type and and
    any attached transforms and will load them into the correct model
    class.

    As this method uses the `pickle` module to load the model, it is not
    safe to load models from untrusted sources as this could lead to
    arbitrary code execution!

    Args:
        file: The file like object to load the model from.
    """
    # Read the general model from the file
    if isinstance(file, Path):
        with file.open("rb") as f:
            data = pickle.load(f)  # noqa: S301
    else:
        data = pickle.load(file)  # noqa: S301
    if not isinstance(data, dict):
        msg = "Invalid model file"
        raise ValueError(msg)  # noqa: TRY004, it's really a value error and not a type error
    data = cast(dict[str, Any], data)

    # Create a model based on the type
    model_type = data["model_type"]
    module_name, class_name = model_type.rsplit(".", 1)
    module = importlib.import_module(module_name)
    model_class = getattr(module, class_name)

    return model_class.load_from_state(data["model"])

save_state() abstractmethod

Save the model state to a dictionary.

To save the model to a file, use the save method. To create a model from a state dictionary, use the load_from_state method.

Returns:

Type	Description
dict[str, Any]	A dictionary containing the model state.

Source code in src/flowcean/core/model.py

@abstractmethod
def save_state(self) -> dict[str, Any]:
    """Save the model state to a dictionary.

    To save the model to a file, use the `save` method.
    To create a model from a state dictionary, use the `load_from_state`
    method.

    Returns:
        A dictionary containing the model state.
    """

load_from_state(state) abstractmethod classmethod

Load the model from a state dictionary.

To load a model from a file use the load method. To save the model state to a dictionary, use the save_state method.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	A dictionary containing the model state.	required

Source code in src/flowcean/core/model.py

@classmethod
@abstractmethod
def load_from_state(cls, state: dict[str, Any]) -> Model:
    """Load the model from a state dictionary.

    To load a model from a file use the `load` method.
    To save the model state to a dictionary, use the `save_state` method.

    Args:
        state: A dictionary containing the model state.
    """

ModelWithTransform(model, input_transform, output_transform) dataclass

Bases: Model

Model that carries a transform.

Attributes:

Name	Type	Description
model	Model	Model
transform	Model	Transform

Report(entries)

A report containing reportables.

Initialize the report.

Parameters:

Name	Type	Description	Default
entries	dict[str, Reportable]	The report entries.	required

Source code in src/flowcean/core/report.py

def __init__(self, entries: dict[str, Reportable]) -> None:
    """Initialize the report.

    Args:
        entries: The report entries.
    """
    self.entries = entries

__str__()

Return a string representation of the report.

Source code in src/flowcean/core/report.py

def __str__(self) -> str:
    """Return a string representation of the report."""
    return "\n".join(
        f"{name}: {value}" for name, value in self.entries.items()
    )

StopLearning

Bases: Exception

Stop learning.

This exception is raised when the learning process should stop.

ChainedTransforms(*transforms)

Bases: Transform, FitOnce, FitIncremetally

A transform that is a chain of other transforms.

Initialize the chained transforms.

Parameters:

Name	Type	Description	Default
transforms	Transform	The transforms to chain.	()

Source code in src/flowcean/core/transform.py

def __init__(
    self,
    *transforms: Transform,
) -> None:
    """Initialize the chained transforms.

    Args:
        transforms: The transforms to chain.
    """
    self.transforms = transforms

FitIncremetally

Bases: ABC

A mixin for transforms that need to be fitted to data incrementally.

fit_incremental(data) abstractmethod

Fit to the data incrementally.

Parameters:

Name	Type	Description	Default
data	Data	The data to fit to.	required

Source code in src/flowcean/core/transform.py

@abstractmethod
def fit_incremental(self, data: Data) -> None:
    """Fit to the data incrementally.

    Args:
        data: The data to fit to.
    """

FitOnce

Bases: ABC

A mixin for transforms that need to be fitted to data once.

fit(data) abstractmethod

Fit to the data.

Parameters:

Name	Type	Description	Default
data	Data	The data to fit to.	required

Source code in src/flowcean/core/transform.py

@abstractmethod
def fit(self, data: Data) -> None:
    """Fit to the data.

    Args:
        data: The data to fit to.
    """

Identity()

Bases: Transform

A transform that does nothing.

Initialize the identity transform.

Source code in src/flowcean/core/transform.py

def __init__(self) -> None:
    """Initialize the identity transform."""
    super().__init__()

Transform

Bases: ABC

Base class for all transforms.

apply(data) abstractmethod

Apply the transform to data.

Parameters:

Name	Type	Description	Default
data	Data	The data to transform.	required

Returns:

Type	Description
Data	The transformed data.

Source code in src/flowcean/core/transform.py

@abstractmethod
def apply(self, data: Data) -> Data:
    """Apply the transform to data.

    Args:
        data: The data to transform.

    Returns:
        The transformed data.
    """

__call__(data)

Apply the transform to data.

Parameters:

Name	Type	Description	Default
data	Data	The data to transform.	required

Returns:

Type	Description
Data	The transformed data.

Source code in src/flowcean/core/transform.py

def __call__(self, data: Data) -> Data:
    """Apply the transform to data.

    Args:
        data: The data to transform.

    Returns:
        The transformed data.
    """
    return self.apply(data)

chain(other)

Chain this transform with other transforms.

This can be used to chain multiple transforms together. Chained transforms are applied left to right.

Example

chained_transform = TransformA().chain(TransformB())

Parameters:

Name	Type	Description	Default
other	Transform	The transforms to chain.	required

Returns:

Type	Description
Transform	A new Chain transform.

Source code in src/flowcean/core/transform.py

def chain(
    self,
    other: Transform,
) -> Transform:
    """Chain this transform with other transforms.

    This can be used to chain multiple transforms together.
    Chained transforms are applied left to right.

    Example:
        ```python
        chained_transform = TransformA().chain(TransformB())
        ```

    Args:
        other: The transforms to chain.

    Returns:
        A new Chain transform.
    """
    return ChainedTransforms(self, other)

__or__(other)

Shorthand for chaining transforms.

Example

chained_transform = TransformA() | TransformB()

Parameters:

Name	Type	Description	Default
other	Transform	The transform to chain.	required

Returns:

Type	Description
Transform	A new Chain transform.

Source code in src/flowcean/core/transform.py

def __or__(
    self,
    other: Transform,
) -> Transform:
    """Shorthand for chaining transforms.

    Example:
        ```python
        chained_transform = TransformA() | TransformB()
        ```

    Args:
        other: The transform to chain.

    Returns:
        A new Chain transform.
    """
    return self.chain(other)

inverse()

Get the inverse of the transform.

Returns:

Type	Description
Transform	The inverse of the transform.

Source code in src/flowcean/core/transform.py

def inverse(self) -> Transform:
    """Get the inverse of the transform.

    Returns:
        The inverse of the transform.
    """
    raise NotImplementedError

learn_active(environment, learner)

Learn from an active environment.

Learn from an active environment by interacting with it and learning from the observations. The learning process stops when the environment ends or when the learner requests to stop.

Parameters:

Name	Type	Description	Default
environment	ActiveEnvironment	The active environment.	required
learner	ActiveLearner	The active learner.	required

Returns:

Type	Description
Model	The model learned from the environment.

Source code in src/flowcean/core/strategies/active.py

def learn_active(
    environment: ActiveEnvironment,
    learner: ActiveLearner,
) -> Model:
    """Learn from an active environment.

    Learn from an active environment by interacting with it and
    learning from the observations. The learning process stops when the
    environment ends or when the learner requests to stop.

    Args:
        environment: The active environment.
        learner: The active learner.

    Returns:
        The model learned from the environment.
    """
    model = None
    try:
        while True:
            observations = environment.observe().collect(streaming=True)
            action = learner.propose_action(observations)
            environment.act(action)
            environment.step()
            observations = environment.observe().collect(streaming=True)
            model = learner.learn_active(action, observations)
    except StopLearning:
        pass
    if model is None:
        message = "No model was learned."
        raise RuntimeError(message)
    return model

deploy(environment, model, input_transforms=None, output_transforms=None)

Deploy a trained model to a custom environment.

Parameters:

Name	Type	Description	Default
environment	ActiveEnvironment | IncrementalEnvironment	custom system environment	required
model	ModelWithTransform	the trained model	required
input_transforms	Transform | None	system specific transforms for model input	None
output_transforms	Transform | None	system specific transforms for system input	None

Source code in src/flowcean/core/strategies/deploy.py

def deploy(
    environment: ActiveEnvironment | IncrementalEnvironment,
    model: ModelWithTransform,
    input_transforms: Transform | None = None,
    output_transforms: Transform | None = None,
) -> None:
    """Deploy a trained model to a custom environment.

    Args:
        environment: custom system environment
        model: the trained model
        input_transforms: system specific transforms for model input
        output_transforms: system specific transforms for system input
    """
    if input_transforms is None:
        input_transforms = Identity()
    if output_transforms is None:
        output_transforms = Identity()

    observation = environment.observe()
    output = model.predict(input_transforms.apply(observation))

    if isinstance(environment, ActiveEnvironment):
        environment.act(output_transforms.apply(output).collect())

learn_incremental(environment, learner, inputs, outputs, input_transform=None, output_transform=None)

Learn from a incremental environment.

Learn from a incremental environment by incrementally learning from the input-output pairs. The learning process stops when the environment ends.

Parameters:

Name	Type	Description	Default
environment	IncrementalEnvironment	The incremental environment.	required
learner	SupervisedIncrementalLearner	The supervised incremental learner.	required
inputs	list[str]	The input feature names.	required
outputs	list[str]	The output feature names.	required
input_transform	Transform | None	The transform to apply to the input features.	None
output_transform	Transform | None	The transform to apply to the output features. Its inverse will be part of the final model.	None

Returns:

Type	Description
Model	The model learned from the environment.

Source code in src/flowcean/core/strategies/incremental.py

def learn_incremental(
    environment: IncrementalEnvironment,
    learner: SupervisedIncrementalLearner,
    inputs: list[str],
    outputs: list[str],
    input_transform: Transform | None = None,
    output_transform: Transform | None = None,
) -> Model:
    """Learn from a incremental environment.

    Learn from a incremental environment by incrementally learning from
    the input-output pairs. The learning process stops when the environment
    ends.

    Args:
        environment: The incremental environment.
        learner: The supervised incremental learner.
        inputs: The input feature names.
        outputs: The output feature names.
        input_transform: The transform to apply to the input features.
        output_transform: The transform to apply to the output features.
            Its inverse will be part of the final model.

    Returns:
        The model learned from the environment.
    """
    model = None
    for data in environment:
        input_features = data.select(inputs)
        output_features = data.select(outputs)

        if isinstance(input_transform, FitIncremetally):
            input_transform.fit_incremental(input_features)

        if input_transform is not None:
            input_features = input_transform.apply(input_features)

        if isinstance(output_transform, FitIncremetally):
            output_transform.fit_incremental(output_features)

        if output_transform is not None:
            output_features = output_transform.apply(output_features)

        model = learner.learn_incremental(
            input_features,
            output_features,
        )

    if model is None:
        message = "No data found in environment."
        raise ValueError(message)

    if input_transform is None and output_transform is None:
        return model

    if output_transform is not None:
        output_transform = output_transform.inverse()

    return ModelWithTransform(
        model=model,
        input_transform=input_transform,
        output_transform=output_transform,
    )

evaluate_offline(model, environment, inputs, outputs, metrics)

Evaluate a model on an offline environment.

Evaluate a model on an offline environment by predicting the outputs from the inputs and comparing them to the true outputs.

Parameters:

Name	Type	Description	Default
model	Model	The model to evaluate.	required
environment	OfflineEnvironment	The offline environment.	required
inputs	list[str]	The input feature names.	required
outputs	list[str]	The output feature names.	required
metrics	list[OfflineMetric]	The metrics to evaluate the model with.	required

Returns:

Type	Description
Report	The evaluation report.

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

def evaluate_offline(
    model: Model,
    environment: OfflineEnvironment,
    inputs: list[str],
    outputs: list[str],
    metrics: list[OfflineMetric],
) -> Report:
    """Evaluate a model on an offline environment.

    Evaluate a model on an offline environment by predicting the outputs from
    the inputs and comparing them to the true outputs.

    Args:
        model: The model to evaluate.
        environment: The offline environment.
        inputs: The input feature names.
        outputs: The output feature names.
        metrics: The metrics to evaluate the model with.

    Returns:
        The evaluation report.
    """
    data = environment.observe()
    input_features = data.select(inputs)
    output_features = data.select(outputs)
    predictions = model.predict(input_features)
    return Report(
        {
            metric.name: metric(output_features, predictions.lazy())
            for metric in metrics
        },
    )

learn_offline(environment, learner, inputs, outputs, *, input_transform=None, output_transform=None)

Learn from an offline environment.

Learn from an offline environment by learning from the input-output pairs.

Parameters:

Name	Type	Description	Default
environment	OfflineEnvironment	The offline environment.	required
learner	SupervisedLearner	The supervised learner.	required
inputs	list[str]	The input feature names.	required
outputs	list[str]	The output feature names.	required
input_transform	Transform | None	The transform to apply to the input features. Will be part of the final model.	None
output_transform	Transform | None	The transform to apply to the output features. Its inverse will be part of the final model.	None

Returns:

Type	Description
Model	The model learned from the environment.

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

def learn_offline(
    environment: OfflineEnvironment,
    learner: SupervisedLearner,
    inputs: list[str],
    outputs: list[str],
    *,
    input_transform: Transform | None = None,
    output_transform: Transform | None = None,
) -> Model:
    """Learn from an offline environment.

    Learn from an offline environment by learning from the input-output pairs.

    Args:
        environment: The offline environment.
        learner: The supervised learner.
        inputs: The input feature names.
        outputs: The output feature names.
        input_transform: The transform to apply to the input features.
            Will be part of the final model.
        output_transform: The transform to apply to the output features.
            Its inverse will be part of the final model.

    Returns:
        The model learned from the environment.
    """
    logger.info("Learning with offline strategy")
    data = environment.observe()
    logger.info("Selecting input and output features")
    input_features = data.select(inputs)
    output_features = data.select(outputs)

    if isinstance(input_transform, FitOnce):
        input_transform.fit(input_features)
    elif isinstance(input_transform, FitIncremetally):
        input_transform.fit_incremental(input_features)

    if input_transform is not None:
        input_features = input_transform.apply(input_features)

    if isinstance(output_transform, FitOnce):
        output_transform.fit(output_features)
    elif isinstance(output_transform, FitIncremetally):
        output_transform.fit_incremental(output_features)

    if output_transform is not None:
        output_features = output_transform.apply(output_features)

    logger.info("Learning model")
    model = learner.learn(inputs=input_features, outputs=output_features)

    if input_transform is None and output_transform is None:
        return model

    if output_transform is not None:
        output_transform = output_transform.inverse()

    return ModelWithTransform(
        model,
        input_transform,
        output_transform,
    )