Core¶

Bases: ABC

Abstract base class for adapters.

A component connecting Flowcean to cyber-physical systems. An adapter handles the interface between learned models and real cyber-physical systems during deployment. It manages communication with the system's sensors (for observations) and actuators (for applying actions), enabling the deployment of trained models in production.

start() -> None

stop() -> None

get_data() -> Data

send_data(data: Data) -> None

Manage multiple callbacks through a single interface.

Initialize the callback manager.

callbacks = callbacks or []

on_learning_start(learner: Named, context: dict[str, Any] | None = None) -> None

on_learning_progress(learner: Named, progress: float | None = None, metrics: dict[str, Any] | None = None) -> None

on_learning_end(learner: Named, model: Model, metrics: dict[str, Any] | None = None) -> None

on_learning_error(learner: Named, error: Exception) -> None

Mixin to add callback support to learners.

callback_manager: CallbackManager

Bases: Protocol

Protocol for learner callbacks.

on_learning_start(learner: Named, context: dict[str, Any] | None = None) -> None

on_learning_progress(learner: Named, progress: float | None = None, metrics: dict[str, Any] | None = None) -> None

on_learning_end(learner: Named, model: Model, metrics: dict[str, Any] | None = None) -> None

on_learning_error(learner: Named, error: Exception) -> None

Bases: LearnerCallback

Standard Python logging callback.

Initialize the logging callback.

logger = logger or logging.getLogger('flowcean.learner')

level_start = level_start

level_progress = level_progress

level_end = level_end

level_error = level_error

on_learning_start(learner: Named, context: dict[str, Any] | None = None) -> None

on_learning_progress(learner: Named, progress: float | None = None, metrics: dict[str, Any] | None = None) -> None

on_learning_end(learner: Named, model: Model, metrics: dict[str, Any] | None = None) -> None

on_learning_error(learner: Named, error: Exception) -> None

Bases: LearnerCallback

Rich console callback with adaptive progress display.

Initialize the Rich callback.

console = console or Console()

show_metrics = show_metrics

on_learning_start(learner: Named, context: dict[str, Any] | None = None) -> None

on_learning_progress(learner: Named, progress: float | None = None, metrics: dict[str, Any] | None = None) -> None

on_learning_end(learner: Named, model: Model, metrics: dict[str, Any] | None = None) -> None

on_learning_error(learner: Named, error: Exception) -> None

Bases: LearnerCallback

Simplified Rich callback with just a spinner.

Initialize the Rich spinner callback.

console = console or Console()

on_learning_start(learner: Named, context: dict[str, Any] | None = None) -> None

on_learning_progress(learner: Named, progress: float | None = None, metrics: dict[str, Any] | None = None) -> None

on_learning_end(learner: Named, model: Model, metrics: dict[str, Any] | None = None) -> None

on_learning_error(learner: Named, error: Exception) -> None

A callback that intentionally produces no output.

on_learning_start(learner: Named, context: dict[str, Any] | None = None) -> None

on_learning_progress(learner: Named, progress: float | None = None, metrics: dict[str, Any] | None = None) -> None

on_learning_end(learner: Named, model: Model, metrics: dict[str, Any] | None = None) -> None

on_learning_error(learner: Named, error: Exception) -> None

Bases: Protocol

Base class for active environments.

Active environments require actions to be taken to advance.

act(action: Data) -> None

Bases: Environment, Stepable, Actable, Protocol

An environment supporting active learning through interaction.

An active environment loads data interactively from simulations or real systems. The learner influences the environment by selecting actions, which the environment responds to with observations and rewards. This supports active learning strategies where the learner explores the environment to optimize its behavior.

Bases: Named, Protocol

Abstraction for data sources in learning and evaluation.

An environment describes the possible data sources for the learning and evaluation procedure. Environments can be offline (pre-recorded datasets), incremental (streaming data), or active (interactive systems where the learner can influence the environment). All environments support applying transforms to observations.

transform: Transform = Identity()

append_transform(transform: Transform) -> Self

observe() -> Data

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.

Bases: Environment, Stepable, Iterable[Data], Protocol

An environment providing incremental (streaming) learning data.

Incremental environments provide data as a continuous stream of samples or small batches. The environment is advanced by stepping through data, observing the current state at each step. This supports incremental learning (also known as passive online learning), where the model is continuously updated as new data arrives.

num_steps() -> int | None

Bases: Protocol

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

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:

step() -> None

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: Environment) -> ChainedOfflineEnvironments

Bases: Named, Protocol

Base class for active learners.

Active learners require actions to be taken to learn.

learn_active(action: Data, observation: Data) -> Model

propose_action(observation: Data) -> Data

Bases: Named, Protocol

Base class for incremental supervised learners.

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

learn_incremental(inputs: Data, outputs: Data) -> Model

Bases: Named, Protocol

Base class for supervised learners.

A supervised learner learns from input-output pairs.

learn(inputs: Data, outputs: Data) -> Model

Bases: Named, Protocol

Base class for metrics for active environments.

Bases: Named, Protocol

Quantitative measure for evaluating model performance.

Metrics compare predictions with true outputs to assess model quality.

prepare(data: Data) -> Data

compute(true: Data, predicted: Data) -> Reportable

Bases: Model, Protocol

Protocol for classifier models with threshold-based predictions.

Extends Model with probability predictions and a configurable decision threshold. Implement this protocol for classifiers that support predict_proba.

threshold: float

predict_proba(input_features: Data) -> Data

Bases: Named, Protocol

Base class for models.

A model is used to predict outputs for given inputs.

pre_transform: Transform = Identity()

post_transform: Transform = Identity()

preprocess(input_features: Data) -> Data

predict(input_features: Data) -> Data

postprocess(output: Data) -> Data

save(file: Path | str | BinaryIO) -> None

model.save("model.fml")

load(file: Path | str | BinaryIO) -> Model

model = Model.load("model.fml")

Bases: dict[str, ReportEntry]

A structured container for evaluation results of multiple models.

The Report maps model names to their metric results. For each model:

• top-level keys are metric names,
• values are either:
  • a single Reportable (e.g., scalar metric result), or
  • a nested mapping from submetric names to Reportable objects (e.g., per-class F1 scores, per-feature regression errors, or multi-output results).

This hierarchical structure allows uniform representation of both simple metrics and complex hierarchical metrics across multiple models.

Example:

report = Report( ... { ... "model_a": { ... "accuracy": 0.95, ... "f1": {"class_0": 0.91, "class_1": 0.89}, ... }, ... "model_b": { ... "mae": {"feature_x": 0.2, "feature_y": 0.3}, ... }, ... } ... )

great_table() -> GT

pretty_print(header_style: StyleType = 'bold magenta', metric_style: StyleType = 'cyan', value_style: StyleType = 'green', title_style: StyleType = 'bold yellow') -> None

Bases: Protocol

An action in an active environment.

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

Parameters:

actuators: list[ActiveInterface]

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:

uid: str

value: int | float | NDArray[Any] | None

value_min: SupportsFloat | NDArray[Any] | list[Any]

value_max: SupportsFloat | NDArray[Any] | list[Any]

shape: Sequence[int]

dtype: type[floating[Any]] | type[integer[Any]]

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:

sensors: list[ActiveInterface]

rewards: list[ActiveInterface]

Bases: Exception

Stop learning.

This exception is raised when the learning process should stop.

Bases: Invertible, Transform

A composition of multiple transforms applied sequentially.

Chained transforms are applied left-to-right. Useful for building preprocessing pipelines.

Initialize the chained transforms.

Parameters:

transforms: Sequence[Transform] = transforms

apply(data: Data) -> Data

chain(other: Transform) -> Transform

fit(data: Data) -> Self

fit_incremental(data: Data) -> Self

inverse() -> Transform

Bases: Invertible, Transform

A no-op transform that returns data unchanged.

Often used as a placeholder or default transform.

Initialize the identity transform.

apply(data: Data) -> Data

inverse() -> Transform

chain(other: Transform) -> Transform

Bases: Protocol

Protocol for transforms that support inversion.

An invertible transform can undo its effect via inverse().

>>> scaler = Standardize().fit(data)
>>> restored = scaler.inverse()(scaler(data))

inverse() -> Transform

Bases: Transform, Invertible, Protocol

Bases: Transform, Invertible

A transform wrapping a function.

Useful for quick one-off transformations without creating a dedicated class.

>>> to_float = Lambda(lambda df: df.cast(pl.Float64))
>>> normalized = Lambda(
...     lambda df: (df - df.mean()) / df.std(),
...     inverse_func=lambda df: df * df.std() + df.mean(),
... )

Initialize the lambda transform.

Parameters:

func: Callable[[Data], Data] = func

inverse_func: Callable[[Data], Data] | None = inverse_func

apply(data: Data) -> Data

inverse() -> Transform

Bases: Named, Protocol

Base protocol for all transforms in Flowcean.

A transform is a reusable operation that modifies data. Examples include preprocessing (e.g., standardization), feature engineering (e.g., feature selection, PCA), or augmentation (noise injection, synthetic features).

Transforms are composable via the | operator, allowing complex transformation pipelines to be expressed in a clean and functional style:

>>> transform = Select(features=["x"]) | Standardize()
>>> transformed = transform(dataset)

apply(data: Data) -> Data

chain(other: Transform) -> Transform

chained = TransformA().chain(TransformB())
chained(data)  # Equivalent to TransformB(TransformA(data))

fit(data: Data) -> Self

fit_incremental(data: Data) -> Self