Flowcean

https://flowcean.me/
Admin
Flowcean is a research-oriented Python toolkit for defining, simulating, identifying, evaluating,...
Tokens:345,853
Snippets:2,145
Trust Score:4.8
Update:1 month ago
Show doc for...
Tokens:
Raw
### Complete Flowcean Learning Pipeline

Source: https://flowcean.me/getting_started/new_project

The full code for the getting started example, combining CLI initialization, environment, learner, and active learning model. It prints the model and the learner's rewards.

```python
import flowcean.cli
from flowcean.core import learn_active


def main() -> None:
    flowcean.cli.initialize()

    environment = MyEnvironment(
        initial_state=0.0,
        max_value=10.0,
        max_num_iterations=1_000,
    )

    learner = MyLearner()

    model = learn_active(
        environment,
        learner,
    )
    print(model)  # Show which model was build
    print(learner.rewards)  # Show the reward over time


if __name__ == "__main__":
    main()

```

--------------------------------

### Run the trace prediction example

Source: https://flowcean.me/examples/trace_prediction

Run the example from the repository root. The output starts with the data flow and then prints the model results.

```bash
uv run --directory ./examples/trace_prediction python run.py

```

--------------------------------

### Adapter.start()

Source: https://flowcean.me/reference/flowcean/core

Abstract method to start the adapter. Called when the tool loop is started, it should initialize the adapter, start background processes, and establish connections to the CPS.

```APIDOC
## Adapter.start()

### Description
Start the adapter. This method is called when the tool loop is started. It should be used to initialize the adapter, start any background processes and establish connections to the CPS.

### Method
start()

### Parameters
None

### Returns
None

### Source Code
```python
@abstractmethod
def start(self) -> None:
    """Start the adapter.

    This method is called when the tool loop is started.
    It should be used to initialize the adapter, start any
    background processes and establish connections to the CPS.
    """
```
```

--------------------------------

### Install just to a custom directory

Source: https://flowcean.me/getting_started/prerequisites

Download and install just to a specified directory using the official install script.

```bash
curl --proto '=https' --tlsv1.2 -sSf https://just.systems/install.sh | bash -s -- --to <DEST>
```

--------------------------------

### Run the localization example

Source: https://flowcean.me/examples/robot_localization_failure

Commands to navigate to the example directory and execute the main run script using uv.

```bash
cd examples/robot_localization_failure
uv run run.py
```

--------------------------------

### Run the minimal hybrid system example

Source: https://flowcean.me/examples/hs_simple

Run the example from the repository root using uv. This command executes the main.py script in the examples/hs-simple directory.

```bash
uv run --directory ./examples/hs-simple python main.py

```

--------------------------------

### Install uv on Linux and macOS

Source: https://flowcean.me/getting_started/prerequisites

Install uv using the official install script for Linux and macOS.

```bash
curl -LsSf https://astral.sh/uv/install.sh | sh
```

--------------------------------

### Adapter start() method

Source: https://flowcean.me/reference/flowcean/core/adapter

Initializes the adapter, starts background processes, and establishes connections to the CPS when the tool loop begins.

```python
@abstractmethod
def start(self) -> None:
    """Start the adapter.

    This method is called when the tool loop is started.
    It should be used to initialize the adapter, start any
    background processes and establish connections to the CPS.
    """
```

--------------------------------

### Install uv on Windows

Source: https://flowcean.me/getting_started/prerequisites

Install uv using PowerShell on Windows.

```powershell
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
```

--------------------------------

### Preview documentation with just

Source: https://flowcean.me/getting_started/documentation

Starts a live preview server for the documentation.

```bash
just docs-serve
```

--------------------------------

### Adapter.start()

Source: https://flowcean.me/reference/flowcean/core/adapter

Starts the adapter, initializing it and establishing connections to the cyber-physical system.

```APIDOC
## Adapter.start()

### Description
Starts the adapter. This method is called when the tool loop is started. It should be used to initialize the adapter, start any background processes, and establish connections to the CPS.

### Method
start()

### Parameters
None

### Returns
None

### Example
```python
adapter.start()
```
```

--------------------------------

### CallbackManager Usage Example

Source: https://flowcean.me/reference/flowcean/core/callbacks/base

Example showing how to create a CallbackManager with RichCallback and LoggingCallback and invoke the on_learning_start method.

```APIDOC
## CallbackManager Usage Example

### Description
Demonstrates how to instantiate a CallbackManager with multiple callbacks and trigger the on_learning_start callback.

### Example
```python
from flowcean.core.callbacks import (
    CallbackManager,
    LoggingCallback,
    RichCallback,
)

manager = CallbackManager([RichCallback(), LoggingCallback()])
manager.on_learning_start(learner)
```
```

--------------------------------

### Install WSL on Windows

Source: https://flowcean.me/getting_started/prerequisites

Install Windows Subsystem for Linux using the wsl --install command.

```powershell
wsl --install
```

--------------------------------

### OPCAdapter YAML configuration example

Source: https://flowcean.me/reference/flowcean/adapter/opc/adapter

Example configuration file for the OPCAdapter. Specifies the server URL, pre-capture window length, inputs, and outputs.

```yaml
# Path to the OPC Server
server-url: "opc.tcp://127.0.0.1:4840"

# Specify the time window to record before the recording flag is set.
# This enables the capture and processing of data that triggers the
# recording flag.
pre_capture_window_length: 2

# Define Flowcean inputs which their respective feature name, the
# opc-id and the datatype of the feature.
# All inputs will be passed on to the Flowcean model when triggering an
# inference.
# Available data types are: float32, float64, int32, int64 and bool.
inputs:
    - feature: "feature_a"
      opc-id: "<OPC-ID String>"
      type: "int32"
    - feature: "feature_b"
      opc-id: "<OPC-ID String>"
      type: "float64"
    - ...

# Define the outputs from the Flowcean Model to the OPC Server.
# All outputs need to be mapped to on OPC field for the adapter to work
# correctly. If certain outputs are not needed / used, those should be

```

--------------------------------

### on_learning_start method

Source: https://flowcean.me/reference/flowcean/core/callbacks/logging

Logs a learning start event with the learner's name and optional context. The message format is '[learner.name] Learning started' with context appended as 'key=value' pairs.

```python
def on_learning_start(
    self,
    learner: Named,
    context: dict[str, Any] | None = None,
) -> None:
    """Log learning start event."""
    message = f"[{learner.name}] Learning started"
    if context:
        context_str = ", ".join(f"{k}={v}" for k, v in context.items())
        message += f" ({context_str})"
    self.logger.log(self.level_start, message)
```

--------------------------------

### Install just to ~/.local/bin and add to PATH

Source: https://flowcean.me/getting_started/prerequisites

Create the target directory, download and extract just, add it to the shell PATH, and verify with just --help.

```bash
# create ~/.local/bin
mkdir -p ~/.local/bin

# download and extract just to ~/.local/bin/just
curl --proto '=https' --tlsv1.2 -sSf https://just.systems/install.sh | bash -s -- --to ~/.local/bin

# add `~/.local/bin` to the paths that your shell searches for executables
# this line should be added to your shells initialization file,
# e.g. `~/.bashrc` or `~/.zshrc`
export PATH="$PATH:$HOME/.local/bin"

# just should now be executable
just --help
```

--------------------------------

### Install Python on Fedora

Source: https://flowcean.me/getting_started/prerequisites

Install Python3 using dnf on Fedora.

```bash
sudo dnf install python3
```

--------------------------------

### Verify Git installation

Source: https://flowcean.me/getting_started/prerequisites

Check the installed Git version.

```bash
git --version
```

--------------------------------

### start_mosaik - start mosaik simulation with SyncSimulator

Source: https://flowcean.me/reference/flowcean/mosaik/energy_system

Starts a mosaik simulation process. Configures the SyncSimulator, builds the scenario, connects sensor and actuator entities, and runs the simulation until the scenario end. Handles SimulationError by shutting down the world and sets sim_finished event.

```python
def start_mosaik(
    scenario: Scenario,
    *,
    sensors: list[str],
    actuators: list[str],
    sensor_queue: queue.Queue,
    actuator_queue: queue.Queue,
    sync_finished: threading.Event,
    sync_terminate: threading.Event,
    sim_finished: threading.Event,
) -> None:
    """Start the mosaik simulation process."""
    scenario.makelists.sim_config["SyncSimulator"] = {
        "python": "flowcean.mosaik.simulator:SyncSimulator",
    }
    scenario.build()
    world = scenario.world
    if not isinstance(world, mosaik.World):
        sim_finished.set()
        logger.error(
            "Malformed world object: %s (%s)",
            str(world),
            type(world),
        )
        return

    logger.debug("Starting SyncSimulator ...")
    sync_sim = world.start(
        "SyncSimulator",
        step_size=scenario.base.step_size,
        start_date=scenario.base.start_date,
        sensor_queue=sensor_queue,
        actuator_queue=actuator_queue,
        sync_finished=sync_finished,
        sync_terminate=sync_terminate,
        end=scenario.base.end,
    )

    logger.debug("Connecting sensor entities ...")
    for uid in sensors:
        sid, eid, attr = uid.split(".")
        full_id = f"{sid}.{eid}"
        sensor_model = sync_sim.Sensor(uid=uid)
        logger.debug("Connecting %s ...", full_id)
        world.connect(
            scenario.entities[full_id],
            sensor_model,
            (attr, "reading"),
        )

    logger.debug("Connecting actuator entities ...")
    for uid in actuators:
        sid, eid, attr = uid.split(".")
        full_id = f"{sid}.{eid}"
        actuator_model = sync_sim.Actuator(uid=uid)
        world.connect(
            actuator_model,
            scenario.entities[full_id],
            ("setpoint", attr),
            time_shifted=True,
            initial_data={"setpoint": None},
        )

    logger.info("Starting mosaik run ...")
    try:
        world.run(until=scenario.base.end)
    except SimulationError:
        logger.info("Simulation finished non-regular.")
        world.shutdown()
    else:
        logger.info("Simulation finished.")
    sim_finished.set()

```

--------------------------------

### Create a virtual environment and install Flowcean from PyPI using pip

Source: https://flowcean.me/getting_started/installation

Creates and activates a virtual environment, upgrades pip, and installs Flowcean from PyPI.

```bash
python -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install flowcean

```

--------------------------------

### Install PyQt6 with pip for interactive plotting

Source: https://flowcean.me/getting_started/installation

Installs the PyQt6 GUI backend using pip, for interactive Matplotlib windows.

```bash
python -m pip install PyQt6

```

--------------------------------

### Set up editable install with pip for Flowcean development

Source: https://flowcean.me/getting_started/installation

Creates a virtual environment and installs Flowcean in editable mode using pip, as a fallback when uv is not available.

```bash
python -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install -e .

```

--------------------------------

### RichSpinnerCallback.on_learning_start(learner, context=None)

Source: https://flowcean.me/reference/flowcean/core/callbacks

Displays a learning start message with a spinner. Starts a live display showing a spinner and the learner name.

```APIDOC
## on_learning_start(learner, context=None)

### Description
Displays a learning start message with a spinner. Starts a live display showing a spinner and the learner name.

### Parameters
- **learner** (Named) - Required - The learner that is starting.
- **context** (dict[str, Any] | None) - Optional - Additional context information. Defaults to None.

### Behavior
- Creates a Text object with a spinner and learner name.
- Starts a Live display with the text, refreshing at 10 times per second.

### Return Value
None
```

--------------------------------

### Report initialization example

Source: https://flowcean.me/reference/flowcean/core/report

An example of initializing a Report object with nested model metrics.

```python
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}, ... }, ... } ... )
```

--------------------------------

### Install Git on macOS with Homebrew

Source: https://flowcean.me/getting_started/prerequisites

Install Git using Homebrew on macOS.

```bash
brew install git
```

--------------------------------

### LoggingCallback.on_learning_start

Source: https://flowcean.me/reference/flowcean/core/callbacks/logging

Logs the start of a learning process with optional context information.

```APIDOC
## LoggingCallback.on_learning_start

### Description
Logs a message indicating that learning has started, including the learner's name and optional context key-value pairs.

### Method
on_learning_start(learner, context=None)

### Parameters
- **learner** (Named) - Required - The learner object with a name attribute
- **context** (dict[str, Any] | None) - Optional - Additional context information to include in the log message

### Returns
None
```

--------------------------------

### Usage example for tune_threshold with FBetaScore

Source: https://flowcean.me/reference/flowcean/core/strategies/offline

Shows how to call tune_threshold with an FBetaScore metric, retrieve the best threshold, and apply it to the model. The example uses a model, eval_env, inputs, and outputs that are assumed to be defined elsewhere.

```python
>>> from flowcean.sklearn.metrics.classification import FBetaScore
>>> metric = FBetaScore(beta=1.0)
>>> best_threshold, results = tune_threshold(
...     model, eval_env, inputs, outputs, metric
... )
>>> print(f"Best threshold: {best_threshold:.3f}")
>>> model.threshold = best_threshold  # Apply the best threshold
```

--------------------------------

### on_learning_start(learner, context=None)

Source: https://flowcean.me/reference/flowcean/core/callbacks/rich

Displays a learning start message with a spinner using the Rich Live component.

```python
def on_learning_start(
    self,
    learner: Named,
    context: dict[str, Any] | None = None,
) -> None:
    """Display learning start message with spinner."""
    text = Text()
    text.append("⠋", style="bold blue")
    text.append(f" [{learner.name}] Learning", style="bold")

    self._live = Live(text, console=self.console, refresh_per_second=10)
    self._live.start()
```

--------------------------------

### CallbackManager.on_learning_start

Source: https://flowcean.me/reference/flowcean/core/callbacks/base

Notifies all registered callbacks that the learning process has started.

```python
def on_learning_start(
    self,
    learner: Named,
    context: dict[str, Any] | None = None,
) -> None:
    """Notify all callbacks that learning has started."""
    for callback in self.callbacks:
        callback.on_learning_start(learner, context)
```

--------------------------------

### Explore dataset with Dash

Source: https://flowcean.me/examples/robot_localization_failure

Commands to navigate to the example directory and launch the interactive Dash visualization tool.

```bash
cd examples/robot_localization_failure
uv run explore.py
```

--------------------------------

### Install Git on Linux (Debian, Fedora, Arch)

Source: https://flowcean.me/getting_started/prerequisites

Install Git using the package manager for Debian-based distributions, Fedora, or Arch Linux.

```bash
# Debian-based distributions like Ubuntu
sudo apt-get install git
# Fedora
sudo dnf install git
# Arch Linux
sudo pacman -S git
```

--------------------------------

### RichCallback.on_learning_start

Source: https://flowcean.me/reference/flowcean/core/callbacks/rich

Displays a learning start message with the learner name and optional context metrics.

```APIDOC
## RichCallback.on_learning_start

### Description
Displays a learning start message with the learner name and optional context metrics.

### Method
on_learning_start(learner, context=None)

### Parameters
- **learner** (Named) - Required - The learner object that has a name attribute.
- **context** (dict[str, Any] | None) - Optional - A dictionary of context metrics to display if show_metrics is True.

### Behavior
Prints a message to the console indicating that learning has started, including the learner name and context metrics if provided and show_metrics is True.

### Example
```python
callback.on_learning_start(learner)
```
```

--------------------------------

### LearnerCallback.on_learning_start

Source: https://flowcean.me/reference/flowcean/core/callbacks/base

Called when learning starts. This method is part of the LearnerCallback protocol and should be implemented by callback classes.

```APIDOC
## LearnerCallback.on_learning_start

### Description
Called when learning starts. This is an abstract method that must be implemented by any class implementing the LearnerCallback protocol.

### Method
on_learning_start

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

### Parameters
- **learner** (Named) - Required - The learner object that is starting learning.
- **context** (dict[str, Any] | None) - Optional - Additional context information for the learning start.

### Returns
None

### Source Code
```python
@abstractmethod
def on_learning_start(
    self,
    learner: Named,
    context: dict[str, Any] | None = None,
) -> None:
    """Called when learning starts."""
```
```

--------------------------------

### Expected output of the trace prediction example

Source: https://flowcean.me/examples/trace_prediction

The output starts with the data flow and then prints the model results.

```text
Trace prediction example
source trace rows: ...
transform: SlidingWindow(window_size=4)
training windows: ...
test windows: ...
evaluation:
prediction preview:

```

--------------------------------

### Usage example: RichCallback with RandomForestRegressorLearner

Source: https://flowcean.me/reference/flowcean/core/callbacks/rich

Shows how to instantiate a learner with a RichCallback to get console progress feedback.

```python
>>> from flowcean.core.callbacks import RichCallback
>>> from flowcean.sklearn import RandomForestRegressorLearner
>>> learner = RandomForestRegressorLearner(callbacks=[RichCallback()])
```

--------------------------------

### Unnest Example Data Setup

Source: https://flowcean.me/reference/flowcean/polars/transforms/unnest

Initializes a Polars DataFrame with a struct column 'c' containing fields 'a' and 't'.

```python
data_frame = pl.Series(
    "c",
    [
        {"a": 1, "t": 1},
        {"a": 4, "t": 2},
        {"a": 7, "t": 3},
        {"a": 10, "t": 4},
        {"a": 15, "t": 5},
    ],
).to_frame()
```

--------------------------------

### Build documentation with just

Source: https://flowcean.me/getting_started/documentation

Executes the docs recipe to build the documentation into the site directory. This process uses uv to manage the virtual environment and dependencies.

```bash
just docs
```

--------------------------------

### on_learning_start(learner, context=None)

Source: https://flowcean.me/reference/flowcean/core/callbacks/rich

Displays a learning start message with a spinner in the console. This method initializes a live display that shows a spinner with the learner's name.

```APIDOC
## on_learning_start(learner, context=None)

### Description
Displays a learning start message with a spinner in the console. This method initializes a live display that shows a spinner with the learner's name.

### Method
on_learning_start

### Parameters
- **learner** (Named) - Required - The learner object whose name is displayed.
- **context** (dict[str, Any] | None) - Optional - Additional context information.

### Behavior
- Creates a Text object with a blue spinner and the learner's name.
- Starts a Live display with a refresh rate of 10 frames per second.
```

--------------------------------

### Create a project and add Flowcean from PyPI using uv

Source: https://flowcean.me/getting_started/installation

Initializes a new Python project with uv and adds the Flowcean package from PyPI.

```bash
uv init flowcean-project
cd flowcean-project
uv add flowcean

```

--------------------------------

### Install Python on macOS with Homebrew

Source: https://flowcean.me/getting_started/prerequisites

Install Python using Homebrew on macOS.

```bash
brew install python
```

--------------------------------

### main(String[] args)

Source: https://flowcean.me/examples/java-automata/flowcean/AutomataLearner/Main.html

The main method is the entry point of the application. It instantiates a GRPC server and throws an IOException if the termination await fails.

```APIDOC
## main(String[] args)

### Description
Initial starting point. Instantiation of GRPC-Server.

### Method
public static void main(String[] args) throws IOException

### Parameters
- `args` (String[]) - Command Line Arguments

### Throws
- `IOException` - termination await exception
```

--------------------------------

### Install Flowcean via pip

Source: https://flowcean.me/

Use this command to install the Flowcean package from PyPI.

```bash
pip install flowcean
```

--------------------------------

### Verify Python installation

Source: https://flowcean.me/getting_started/prerequisites

Check the installed Python version using python3 or python.

```bash
python3 --version
```

```bash
python --version
```

--------------------------------

### Example: Using CallbackManager with RichCallback and LoggingCallback

Source: https://flowcean.me/reference/flowcean/core/callbacks/base

Shows how to instantiate a CallbackManager with RichCallback and LoggingCallback and trigger the on_learning_start callback with a learner object.

```python
>>> from flowcean.core.callbacks import (
...     CallbackManager,
...     LoggingCallback,
...     RichCallback,
... )
manager = CallbackManager([RichCallback(), LoggingCallback()])
manager.on_learning_start(learner)
```

--------------------------------

### Install Python on Arch Linux

Source: https://flowcean.me/getting_started/prerequisites

Install Python using pacman on Arch Linux.

```bash
sudo pacman -S python
```

--------------------------------

### Install PySR extra with pip

Source: https://flowcean.me/getting_started/installation

Installs the optional PySR support extra for Flowcean when using pip.

```bash
python -m pip install "flowcean[pysr]"

```

--------------------------------

### Install Python on Debian-based Linux

Source: https://flowcean.me/getting_started/prerequisites

Update package lists and install Python3 on Debian-based distributions like Ubuntu.

```bash
sudo apt-get update
sudo apt-get install python3
```

--------------------------------

### Filter transform usage examples

Source: https://flowcean.me/reference/flowcean/polars

Examples of using the Filter transform with simple string expressions, logical AND, and nested OR conditions.

```python
Filter("x == 1")
```

```python
Filter(And(["x == 1", "t > 3"]))
```

```python
Filter(And(["x == 1", Or(["t > 3", "N < 15"])]))
```

--------------------------------

### Create Environment for Active Learning

Source: https://flowcean.me/getting_started/new_project

Sets up an environment representing your dataset/datastream/simulation. The MyEnvironment class is not yet defined and will be created later.

```python
environment = MyEnvironment(
    initial_state=0.0,
    max_value=10.0,
    max_num_iterations=1_000,
)

```

--------------------------------

### initialize(**kwargs)

Source: https://flowcean.me/reference/flowcean/cli

Initializes the experiment environment by loading the configuration and setting up logging according to its settings. Accepts additional configuration overrides via keyword arguments and returns the initialized configuration object.

```APIDOC
## initialize(**kwargs)

### Description
Initializes the experiment environment. Loads the configuration and sets up logging according to its settings.

### Method
Function

### Signature
initialize(**kwargs: Any) -> DictConfig | ListConfig

### Parameters
#### Keyword Arguments
- **kwargs** (Any) - Optional - Additional configuration overrides.

### Returns
- **DictConfig | ListConfig** - The initialized configuration object.

### Source Code
```python
def initialize(**kwargs: Any) -> DictConfig | ListConfig:
    """Initialize the experiment environment.

    Loads the configuration and sets up logging according to its settings.

    Args:
        **kwargs: Additional configuration overrides.

    Returns:
        The initialized configuration object.
    """
    conf = load_experiment_config(**kwargs)
    logging.basicConfig(**conf.logging.root)
    for logger_name, logger_conf in conf.logging.items():
        if logger_name == "root":
            continue
        logger = logging.getLogger(logger_name)
        logger.setLevel(logger_conf.level)
    return conf
```
```

--------------------------------

### LightningCallbackBridge.on_train_start

Source: https://flowcean.me/reference/flowcean/torch/lightning_learner

Called when training starts. Forwards the training start event to the callback manager with context including max epochs and batch size.

```APIDOC
## LightningCallbackBridge.on_train_start

### Description
Called when training starts. Forwards the training start event to the callback manager with context including max epochs and batch size.

### Method
Callback method

### Signature
```python
def on_train_start(
    self,
    trainer: lightning.Trainer,
    pl_module: lightning.LightningModule,
) -> None
```

### Parameters
- **trainer** (lightning.Trainer) - Required - The PyTorch Lightning trainer.
- **pl_module** (lightning.LightningModule) - Required - The Lightning module being trained.

### Behavior
- Constructs a context dictionary with keys 'max_epochs' and 'batch_size'.
- Calls `self.callback_manager.on_learning_start(self.learner, context)`.

### Returns
None
```

--------------------------------

### Domain.__init__ constructor

Source: https://flowcean.me/reference/flowcean/testing/domain/domain

Initializes the domain with a feature name. The feature_name parameter is required.

```python
def __init__(
    self,
    feature_name: str,
) -> None:
    """Initialize the domain.

    Args:
        feature_name: The name of the feature the domain belongs to.
    """
    self.feature_name = feature_name

```

--------------------------------

### Initialize OPC Adapter

Source: https://flowcean.me/reference/flowcean/adapter/opc/adapter

Initializes the adapter by loading configuration from a YAML file, connecting to the OPC server, and setting up input/output features and recording schemas.

```python
    Args:
        config_path: Path to the YAML configuration file containing
                the OPC server URL and feature definitions.
        pull_frequency: Frequency in Hz at which to poll the OPC server
            for new data.
    """
    super().__init__()

    self.pull_frequency = pull_frequency

    # Load the configuration from the provided path
    with Path(config_path).open() as yaml_file:
        yaml_data = YAML(typ="safe").load(yaml_file)

    # Initialize the adapter fields
    self.input_features = {}
    self.output_features = {}

    schema: dict[str, Any] = {}

    # Initialize the OPC client
    self.client = Client(yaml_data["server-url"])
    self.client.connect()
    self.client.load_type_definitions()

    # Set the log level of the opcua library to WARNING to reduce
    # verbosity
    logging.getLogger("opcua").setLevel(logging.WARNING)

    # Start a keep-alive mechanism to ensure the connection stays alive
    self.client.keepalive = KeepAlive(
        self.client,
        min(
            self.client.session_timeout,
            self.client.secure_channel_timeout,
        )
        * 0.7,
    )
    self.client.keepalive.start()

    # Get inputs from the YAML configuration
    # Each input has a feature name, an opc-id, and a type
    for feature in yaml_data["inputs"]:
        self.input_features[feature["feature"]] = self.client.get_node(
            feature["opc-id"],
        )
        # Build the input schema from opc to flowcean
        schema[feature["feature"]] = _type_mapping_str2pl[feature["type"]]

    for feature in yaml_data["outputs"]:
        self.output_features[feature["feature"]] = self.client.get_node(
            feature["opc-id"],
        )

    streaming_flag_opc_id = yaml_data["stream_flag"]
    connection_flag_opc_id = yaml_data["connection_flag"]
    prediction_flag_opc_id = yaml_data["prediction_flag"]
    self.pre_capture_window_length = timedelta(
        seconds=yaml_data["pre_capture_window_length"],
    )

    # Setup the remaining opc nodes
    self.streaming_flag_node = self.client.get_node(
        streaming_flag_opc_id,
    )
    self.connection_flag_node = self.client.get_node(
        connection_flag_opc_id,
    )
    self.prediction_flag_node = self.client.get_node(
        prediction_flag_opc_id,
    )

    # Initialize the recording DataFrame with the derived schema
    self.input_schema = schema.copy()
    # Add a special column for the recorded time - this column will
    # only be used internally and dropped before returning the data
    # to flowcean
    schema["_recorded_time"] = pl.Datetime
    self.recorded_data = pl.DataFrame(schema=schema)
```

--------------------------------

### OPCAdapter Initialization Method

Source: https://flowcean.me/reference/flowcean/adapter/opc

The __init__ method initializes the OPC adapter with a path to the YAML configuration file and an optional polling frequency.

```python
def __init__(
    self,
    config_path: str | Path,
    *,
    pull_frequency: float = 150.0,
) -> None:
    r"""Initialize the OPC adapter.

    Initializes a new OPC to Flowcean adapter instance.
    The adapter reads its configuration from a YAML file.
    The path to this file is provided in the config_path argument and the
    file has the following structure:
    ```
    # Path to the OPC Server
    server-url: "opc.tcp://127.0.0.1:4840"

    # Specify the time window to record before the recording flag is set.
    # This enables the capture and processing of data that triggers the
    # recording flag.
    pre_capture_window_length: 2

    # Define Flowcean inputs which their respective feature name, the
    # opc-id and the datatype of the feature.
    # All inputs will be passed on to the Flowcean model when triggering an
    # inference.
    # Available data types are: float32, float64, int32, int64 and bool.
    inputs:
        - feature: "feature_a"
          opc-id: "<OPC-ID String>"
          type: "int32"
        - feature: "feature_b"
          opc-id: "<OPC-ID String>"
          type: "float64"
        - ...
```

--------------------------------

### thermostat(ambient=20.0, heating_power=5.0, cooling_rate=0.3, hysteresis=2.0, initial_state=None)

Source: https://flowcean.me/reference/flowcean/hybrid/benchmarks/thermostat

Creates a thermostat benchmark system. This function returns a HybridSystem configured for thermostat switching based on the provided parameters.

```APIDOC
## thermostat(ambient=20.0, heating_power=5.0, cooling_rate=0.3, hysteresis=2.0, initial_state=None)

### Description
Creates a thermostat benchmark system. This function returns a HybridSystem configured for thermostat switching based on the provided parameters.

### Method
Function

### Signature
```python
def thermostat(
    ambient: float = 20.0,
    heating_power: float = 5.0,
    cooling_rate: float = 0.3,
    hysteresis: float = 2.0,
    initial_state: np.ndarray | None = None,
) -> HybridSystem:
```

### Parameters
- **ambient** (float) - Optional - Ambient temperature. Default: `20.0`
- **heating_power** (float) - Optional - Heating input strength. Default: `5.0`
- **cooling_rate** (float) - Optional - Cooling coefficient. Default: `0.3`
- **hysteresis** (float) - Optional - Temperature hysteresis band width. Default: `2.0`
- **initial_state** (ndarray | None) - Optional - Optional initial temperature. Default: `None`

### Returns
- **HybridSystem** - HybridSystem configured for thermostat switching.

### Source Code
Source code in `src/flowcean/hybrid/benchmarks/thermostat.py`

```python
def thermostat(
    ambient: float = 20.0,
    heating_power: float = 5.0,
    cooling_rate: float = 0.3,
    hysteresis: float = 2.0,
    initial_state: np.ndarray | None = None,
) -> HybridSystem:
    """Create a thermostat benchmark system.

    Args:
        ambient: Ambient temperature.
        heating_power: Heating input strength.
        cooling_rate: Cooling coefficient.
        hysteresis: Temperature hysteresis band width.
        initial_state: Optional initial temperature.

    Returns:
        HybridSystem configured for thermostat switching.
    """

    def heating(
        _t: float,
        state: np.ndarray,
        params: Parameters,
        _input_stream: InputStream,
    ) -> np.ndarray:
        temperature = state[0]
        return np.array(
            [
                -params["cooling_rate"] * (temperature - params["ambient"])
                + params["heating_power"],
            ],
        )

    def cooling(
        _t: float,
        state: np.ndarray,
        params: Parameters,
        _input_stream: InputStream,
    ) -> np.ndarray:
        temperature = state[0]
        return np.array(
            [
                -params["cooling_rate"] * (temperature - params["ambient"]),
            ],
        )

    def event_surface_high(
        t: float,
        state: np.ndarray,
        params: Parameters,
        input_stream: InputStream,
    ) -> float:
        target = float(input_stream(t)[0])
        return state[0] - (target + 0.5 * params["hysteresis"])

    def event_surface_low(
        t: float,
        state: np.ndarray,
        params: Parameters,
        input_stream: InputStream,
    ) -> float:
        target = float(input_stream(t)[0])
        return state[0] - (target - 0.5 * params["hysteresis"])

    heating_dynamics = ContinuousDynamics(heating, label="heating")
    cooling_dynamics = ContinuousDynamics(cooling, label="cooling")
    heating_mode = Location(heating_dynamics, label="heating")
    cooling_mode = Location(cooling_dynamics, label="cooling")

    to_cooling = Transition(
        source=heating_mode,
        target=cooling_mode,
        event=EventSurface(
            event_surface_high,
            direction=CrossingDirection.RISING,
            label="too_hot",
        ),
    )
    to_heating = Transition(
        source=cooling_mode,
        target=heating_mode,
        event=EventSurface(
            event_surface_low,
            direction=CrossingDirection.FALLING,
            label="too_cold",
        ),
    )

    if initial_state is None:
        initial_state = np.array([ambient], dtype=float)

    return HybridSystem(
        locations=[heating_mode, cooling_mode],
        transitions=[to_cooling, to_heating],
        initial_location=heating_mode,
        initial_state=initial_state,
        parameters={
            "ambient": ambient,
            "heating_power": heating_power,
            "cooling_rate": cooling_rate,
            "hysteresis": hysteresis,
        },
    )
```

```

--------------------------------

### Example of creating a Report with hierarchical metrics

Source: https://flowcean.me/reference/flowcean/core

Example showing how to construct a Report object with nested metric results for multiple models. The structure maps model names to metric names, where values can be scalars or nested mappings for submetrics.

```python
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},
        },
    }
)
```

--------------------------------

### get_default_callbacks()

Source: https://flowcean.me/reference/flowcean/core/callbacks/support

Function to get the default callbacks for learners.

```APIDOC
## `get_default_callbacks()`

### Description
Returns the default callbacks for learners. Currently returns an empty list so learners stay silent unless callbacks are provided.

### Returns
- **list[LearnerCallback]** - An empty list.

### Source Code
```python
def get_default_callbacks() -> list[LearnerCallback]:
    """Return the default callbacks for learners.

    Returns:
        An empty list so learners stay silent unless callbacks are provided.
    """
    return []
```
```

--------------------------------

### parser()

Source: https://flowcean.me/examples/java-automata/io/flowcean/learner/grpc/LearnerOuterClass.DataField.html

Static method to get the parser for DataField.

```APIDOC
## parser()

### Description
Returns the parser for DataField.

### Method
parser

### Signature
public static com.google.protobuf.Parser<LearnerOuterClass.DataField> parser()

### Returns
- **com.google.protobuf.Parser<LearnerOuterClass.DataField>** - The parser instance.
```

--------------------------------

### start_mosaik

Source: https://flowcean.me/reference/flowcean/mosaik/energy_system

Starts the mosaik simulation process. Configures the SyncSimulator, connects sensor and actuator entities, and runs the simulation until the specified end time.

```APIDOC
## start_mosaik

### Description
Starts the mosaik simulation process. Configures the SyncSimulator, connects sensor and actuator entities, and runs the simulation until the specified end time.

### Method
Function

### Signature
start_mosaik(scenario, *, sensors, actuators, sensor_queue, actuator_queue, sync_finished, sync_terminate, sim_finished)

### Parameters
- **scenario** (Scenario) - Required - The scenario object containing the simulation setup.
- **sensors** (list[str]) - Required - List of sensor entity identifiers in the format 'sid.eid.attr'.
- **actuators** (list[str]) - Required - List of actuator entity identifiers in the format 'sid.eid.attr'.
- **sensor_queue** (queue.Queue) - Required - Queue for sensor data.
- **actuator_queue** (queue.Queue) - Required - Queue for actuator data.
- **sync_finished** (threading.Event) - Required - Event to signal synchronization completion.
- **sync_terminate** (threading.Event) - Required - Event to signal termination.
- **sim_finished** (threading.Event) - Required - Event to signal simulation completion.

### Returns
None

### Behavior
- Configures the SyncSimulator in the scenario's sim_config.
- Builds the scenario and retrieves the world.
- Starts the SyncSimulator with the given step size, start date, queues, and end time.
- Connects sensor entities to the SyncSimulator's Sensor model.
- Connects actuator entities to the SyncSimulator's Actuator model with time-shifted connections.
- Runs the world simulation until the specified end time.
- Sets the sim_finished event after the simulation completes.
```

--------------------------------

### SilentCallback.on_learning_start

Source: https://flowcean.me/reference/flowcean/core/callbacks/base

A no-op implementation of on_learning_start that does nothing when learning starts.

```APIDOC
## SilentCallback.on_learning_start

### Description
A no-op implementation of on_learning_start that does nothing when learning starts.

### Method
Instance method

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

### Parameters
- **learner** (Named) - Required - The learner object that is starting learning.
- **context** (dict[str, Any] | None) - Optional - Additional context information. Defaults to None.

### Returns
None
```

--------------------------------

### simulate(system, t_span, x0=None, location0=None, *, input_stream=None, capture_inputs=None, capture_derivatives=False, max_jumps=256, rtol=1e-07, atol=1e-09, max_step=None, dense_output=False, sample_times=None, sample_dt=None)

Source: https://flowcean.me/reference/flowcean/hybrid/simulator

Simulates a hybrid system and returns a trace. This function takes a HybridSystem, time span, and optional initial conditions and solver settings.

```APIDOC
## simulate(system, t_span, x0=None, location0=None, *, input_stream=None, capture_inputs=None, capture_derivatives=False, max_jumps=256, rtol=1e-07, atol=1e-09, max_step=None, dense_output=False, sample_times=None, sample_dt=None)

### Description
Simulates a hybrid system and returns a trace. This function takes a HybridSystem, time span, and optional initial conditions and solver settings.

### Parameters
- **system** (HybridSystem) - Required - Hybrid system to simulate.
- **t_span** (tuple[float, float]) - Required - Start and end time for integration.
- **x0** (Iterable[float] | None) - Optional - Optional initial state override. Default: None.
- **location0** (Location | None) - Optional - Optional initial location override. Default: None.
- **input_stream** (InputStream | None) - Optional - Optional input stream accessor for callbacks. Default: None.
- **capture_inputs** (bool | None) - Optional - Input capture mode. If None, capture iff an input stream is provided. Default: None.
- **capture_derivatives** (bool) - Optional - Whether to re-evaluate Location.dynamics.flow on the returned trace grid and store the sampled derivatives in Trace.dx. This assumes pure flow callbacks under repeated evaluation. Scalar derivative returns are accepted only for single-state systems. Default: False.
- **max_jumps** (int) - Optional - Maximum number of transitions allowed. Default: 256.
- **rtol** (float) - Optional - Relative tolerance for the solver. Default: 1e-07.
- **atol** (float) - Optional - Absolute tolerance for the solver. Default: 1e-09.
- **max_step** (float | None) - Optional - Optional maximum step size. Default: None.
- **dense_output** (bool) - Optional - Whether to build a continuous solution per segment. Default: False.
- **sample_times** (Iterable[float] | None) - Optional - Monotone time grid to sample from the dense solution. Default: None.
- **sample_dt** (float | None) - Optional - Fixed sampling interval to generate a time grid. Default: None.

### Returns
- **Trace** - The simulation trace with location labels and events.
```

--------------------------------

### initialize function implementation

Source: https://flowcean.me/reference/flowcean/cli

Loads the experiment configuration and sets up logging according to the configuration's logging settings. It returns the initialized configuration object. The function accepts additional configuration overrides via keyword arguments.

```python
def initialize(**kwargs: Any) -> DictConfig | ListConfig:
    """Initialize the experiment environment.

    Loads the configuration and sets up logging according to its settings.

    Args:
        **kwargs: Additional configuration overrides.

    Returns:
        The initialized configuration object.
    """
    conf = load_experiment_config(**kwargs)
    logging.basicConfig(**conf.logging.root)
    for logger_name, logger_conf in conf.logging.items():
        if logger_name == "root":
            continue
        logger = logging.getLogger(logger_name)
        logger.setLevel(logger_conf.level)
    return conf
```

--------------------------------

### GrpcPassiveAutomataLearner.__init__(backend)

Source: https://flowcean.me/reference/flowcean/grpc/learner

Creates a GrpcPassiveAutomataLearner instance with a specified backend. The backend provides the server address and connection details for the gRPC channel.

```APIDOC
## GrpcPassiveAutomataLearner.__init__(backend)

### Description
Creates a GrpcPassiveAutomataLearner instance with a specified backend. The backend provides the server address and connection details for the gRPC channel.

### Method
Constructor

### Signature
python
def __init__(self, backend: _Backend) -> None


### Parameters
- **backend** (`_Backend`) - Required - The backend to use for the learner.

### Details
- Establishes an insecure gRPC channel to the backend's server address.
- Sets maximum send and receive message lengths.
- Creates a LearnerStub for communication.

### Example
python
# Example usage
backend = _AddressBackend("localhost:50051")
learner = GrpcPassiveAutomataLearner(backend)

```

--------------------------------

### FilterExpr.get()

Source: https://flowcean.me/reference/flowcean/polars/transforms/filter

Abstract method to get the polars expression for this filter.

```APIDOC
## FilterExpr.get()

### Description
Get the polars expression for this filter.

### Method
Abstract method

### Signature
```python
@abstractmethod
def get(self) -> pl.Expr:
    """Get the polars expression for this filter."""
```

### Returns
- **pl.Expr** - The polars expression for this filter.
```

--------------------------------

### Run incremental learning script

Source: https://flowcean.me/examples/one_tank_incremental

Executes the incremental learning script from the repository root using uv.

```bash
uv run --directory ./examples/one_tank python run_incremental.py
```
Tokens:
Raw