active

ActiveInterface(uid, value, value_min, value_max, shape, dtype) dataclass

Interface to a feature in an active environment.

Represents a single feature of the environment, which can be either an input, an output, or the reward of the environment.

Parameters:

Name	Type	Description	Default
uid	str	Identifier of the feature inside the environment	required
value	int | float | NDArray[Any] | None	The value of the feature	required
value_min	SupportsFloat | NDArray[Any] | list[Any]	Simple representation of the minimum value	required
value_max	SupportsFloat | NDArray[Any] | list[Any]	Simple representation of the maximum value	required
shape	Sequence[int]	Tuple representing the shape of the value	required
dtype	type[floating[Any]] | type[integer[Any]]	Data type of this interface, e.g., numpy.float32	required

Observation(sensors, rewards) dataclass

An observation of an active environment.

The observation contains 'sensors', which are the raw observations of featured values, and rewards, which are a rated quantification of the environment state.

Parameters:

Name	Type	Description	Default
sensors	list[ActiveInterface]	List of interface objects, i.e., raw observations	required
rewards	list[ActiveInterface]	List of interface objects, i.e., rated state	required

Action(actuators) dataclass

An action in an active environment.

The action contains 'actuators', which represent setpoints in the environment. Each actuator targets exactly one input feature.

Parameters:

Name	Type	Description	Default
actuators	list[ActiveInterface]	List of interface objects, which are setpoints	required

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()
            action = learner.propose_action(observations)
            environment.act(action)
            environment.step()
            observations = environment.observe()
            model = learner.learn_active(action, observations)
    except StopLearning:
        pass
    if model is None:
        message = "No model was learned."
        raise RuntimeError(message)
    return model

evaluate_active(environment, model, metrics)

Evaluate on an active environment.

Evaluate a model that was trained for an active environment. The optimal output of the model is not known (unlike in supervised settings), therefore, the evaluation function(s) have to be provided manually. The evaluation function(s) are specific to an environment.

Action and observations going into the metric function with the relation that the nth entries of both lists contain the Action and the Observation that results from this action. Therefore, the first entry of Actions will not contain any values and the first entry of Observations contains the initial state of the environment.

Parameters:

Name	Type	Description	Default
environment	ActiveEnvironment	The active environment	required
model	Model	The model to evaluate	required
metrics	list[ActiveMetric]	list of metrics to evaluate against	required

Returns:

Type	Description
Report	The evaluation report.

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

def evaluate_active(
    environment: ActiveEnvironment,
    model: Model,
    metrics: list[ActiveMetric],
) -> Report:
    """Evaluate on an active environment.

    Evaluate a model that was trained for an active environment. The
    optimal output of the model is not known (unlike in supervised
    settings), therefore, the evaluation function(s) have to be
    provided manually. The evaluation function(s) are specific to an
    environment.

    Action and observations going into the metric function with the
    relation that the nth entries of both lists contain the Action and
    the Observation that results from this action. Therefore, the first
    entry of Actions will not contain any values and the first entry of
    Observations contains the initial state of the environment.

    Args:
        environment: The active environment
        model: The model to evaluate
        metrics: list of metrics to evaluate against

    Returns:
        The evaluation report.

    """
    observations: list[Observation] = []
    actions: list[Action] = [Action(actuators=[])]
    try:
        while True:
            observations.append(environment.observe())
            actions.append(model.predict(observations[-1]))
            environment.act(actions[-1])
            environment.step()
    except StopLearning:
        pass
    observations.append(environment.observe())
    entries: dict[str, ReportEntry] = {}
    entries[model.name] = ReportEntry(
        {metric.name: metric(observations, actions) for metric in metrics},
    )
    return Report(entries)