Skip to content

Adapters

Adapters connect deployed models to data sources and sinks. The common Adapter interface is defined in flowcean.core. See the adapter guide for the deployment workflow.

adapter

Classes

DataFrameAdapter 

DataFrameAdapter(source: DataFrame, input_features: Iterable[str], result_path: str)

Bases: Adapter

Adapter wrapper for a DataFrame.

This class allows to use a DataFrame as an Adapter and an input source for a tool loop. The tool receives data from the source DataFrame row by row. The tool loops results are collected in a result DataFrame which is written to a CSV file at the end of the process.

Attributes
df instance-attribute 
df: LazyFrame = cast('pl.LazyFrame', source.observe()).select(input_features)
df_len instance-attribute 
df_len: int = self.df.select(pl.len()).collect().item()
result_df instance-attribute 
result_df: DataFrame
count class-attribute instance-attribute 
count: int = 0
result_path instance-attribute 
result_path: str = result_path
Methods:
start 
start() -> None
stop 
stop() -> None
get_data 
get_data() -> LazyFrame
send_data 
send_data(data: LazyFrame) -> None

opc

Classes

OPCAdapter 

OPCAdapter(config_path: str | Path, *, pull_frequency: float = 150.0)

Bases: Adapter

Flowcean adapter for OPC (Open Platform Communications) protocol.

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"
    - ...

# 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
# dropped using transforms. The available datatypes are the same as for
# the input features.
outputs:
    - feature: "result_feature"
    opc-id: "<OPC-ID String>"
    type: "int32"

# Flags are special OPC fields, which are used to communicate between
# the adapter and the OPC server.

# The streaming flag is set by the OPC server to `True` when Flowcean
# should start to record data. Once set back to `False`, the data is
# forwarded to the model, an inference step is executed and the results
# are send back to the OPC server.
stream_flag: "<OPC-ID String>"

# The prediction flag is set to `True` by the adapter when it
# successfully sends data to the OPC server.
# This can be used on the OPC side to process the newly received data.
# The flag is *not* reset by the adapter, so it will remain `True`
# until the OPC server resets it.
prediction_flag: "<OPC-ID String>"

# The connection flag is set to `True` by the adapter when it
# successfully connects to the OPC server and set to `False` when it
# disconnects from the server.
# This can be used to monitor the connection status of the adapter from
# the OPC side.
connection_flag: "<OPC-ID String>"

Parameters:

Name Type Description Default
config_path str | Path

Path to the YAML configuration file containing the OPC server URL and feature definitions.

 required
pull_frequency float

Frequency in Hz at which to poll the OPC server for new data.

 150.0
Attributes
opc_client instance-attribute 
opc_client: Client
pull_frequency instance-attribute 
pull_frequency: float = pull_frequency
recorded_data instance-attribute 
recorded_data: DataFrame = pl.DataFrame(schema=schema)
input_schema instance-attribute 
input_schema: dict[str, Any] = schema.copy()
pre_capture_window_length instance-attribute 
pre_capture_window_length: timedelta = timedelta(seconds=yaml_data['pre_capture_window_length'])
input_features instance-attribute 
input_features: dict[str, Node] = {}
output_features instance-attribute 
output_features: dict[str, Node] = {}
streaming_flag_node instance-attribute 
streaming_flag_node: Node = self.client.get_node(streaming_flag_opc_id)
connection_flag_node instance-attribute 
connection_flag_node: Node = self.client.get_node(connection_flag_opc_id)
prediction_flag_node instance-attribute 
prediction_flag_node: Node = self.client.get_node(prediction_flag_opc_id)
streaming_handler instance-attribute 
streaming_handler: StreamingHandler
streaming_sub instance-attribute 
streaming_sub: Subscription
client instance-attribute 
client = Client(yaml_data['server-url'])
Methods:
start 
start() -> None
stop 
stop() -> None
get_data 
get_data() -> Data

Get data from the OPC server and return it to flowcean.

send_data 
send_data(data: DataFrame | LazyFrame) -> None

Send data to the OPC server.

Send a polars DataFrame or LazyFrame to the OPC server. The data must contain all required output features, otherwise a ValueError is raised.

Parameters:

Name Type Description Default
data DataFrame | LazyFrame

Polars DataFrame or LazyFrame containing the data to send.

 required