strategies

StopLearning

Bases: Exception

Stop learning.

This exception is raised when the learning process should stop.

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

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,
    )