gitea_admin/sientia-dataops-laborious_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1081

Browse Source 
Enhance documentation across multiple modules with detailed parameter descriptions and usage examples
This commit is contained in:
vitor-aignosi
2025-05-26 16:45:50 -03:00
parent 5326051714
commit 10081b71d2
12 changed files with 294 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
README.md
View File

@@ -0,0 +1,95 @@
# Sientia DataOps Laborious
The Sientia DataOps Laborious is a Temporal-based workflow application that handles batch predictions and data processing for industrial data. It integrates with MLFlow for model management, PostgreSQL for data storage, and OPC for real-time data output. The module is designed to process data in a reliable and scalable manner using Temporal.io's workflow orchestration capabilities. It's get data from Scouter sinks, process it, make predictions using MLFlow models and generates metrics for the predictions.
## Key Features
- Batch predictions using MLFlow models
- Data transformation and preprocessing
- Workflow orchestration using Temporal.io
- Integration with PostgreSQL for data storage
- OPC integration for real-time data output
- Comprehensive error handling and notifications
- Configurable data filters and quality gates
- Scalable deployment architecture
## Workflows
### Predictions Batch
The main workflow that orchestrates batch predictions. Steps:
- prepare_activity: Prepares the activity with schedule and model information
- load_custom_query: Loads data using a custom query
- prediction_process: Executes the prediction process using the Prediction Process sub-workflow
#### Workflow inputs:
- `schedule_name`: The schedule name of the activity
- `model_name`: The model name of the activity
- `model_id`: The model id of the activity
- `query`: The custom query to load data
- `schema`: The schema of the data
- `table_name`: The name of the table to process
- `input_filters`: The filters to be applied during prediction
- `mlflow_transform_filters`: The filters to be applied during prediction
- `mlflow_predict_filters`: The filters to be applied during prediction
- `model_retention`: The model retention period in minutes
- `path_priority`: The path priority
### Prediction Process
Sub-workflow that handles individual prediction processing:
- get_last_timestamp: Gets the last timestamp of the data
- input_gate: Filters input data based on configured rules
- repeat_last_prediction: Repeats the last prediction if the data is empty
- request_transform: Makes predictions using MLFlow models
- mlflow_response_gate: Handles prediction or transform responses and filters
- mlflow_content_gate: Filters transform responses based on configured rules
- request_predict: Makes predictions using MLFlow models
- format_and_export_prediction: Formats and exports predictions using the
Format and Export Prediction sub-workflow
### Format and Export Prediction
Sub-workflow that handles prediction formatting and export:
- format_prediction: Formats prediction data if path flag is None
- format_default_prediction: Formats default prediction data if path flag is not None
- export_to_postgres: Exports formatted predictions to PostgreSQL
- write_to_opc: Writes predictions to OPC server
## Environment variables
- `POSTGRES_HOST`
- `POSTGRES_PORT`
- `POSTGRES_USER`
- `POSTGRES_PASSWORD`
- `POSTGRES_DBNAME`
- `POSTGRES_MIN_CONNECTIONS`
- `POSTGRES_MAX_CONNECTIONS`
- `MLFLOW_HOST`
- `MLFLOW_PORT`
- `MLFLOW_USERNAME`
- `MLFLOW_PASSWORD`
- `OPC_CONFIG` - json string containing the opc configuration for multiple opc servers
For single opc server use:
- `OPC_URL`
- `OPC_NAME`
- `OPC_SERVER_URI`
- `OPC_CERT_PATH`
- `OPC_PRIVATE_KEY_PATH`
- `OPC_SERVER_CERT_PATH`
- `OPC_RECONNECTION_INTERVAL`
- `TEMPORAL_HOST`
- `TEMPORAL_NAMESPACE`
## Application deployment
The application can be deployed using the following command:
```bash
helm upgrade --install sientia-dataops-laborious sientia/sientia-module -n sientia --create-namespace -f ./values.yaml
```

58
laborious/activities/gates.py
View File

@@ -55,13 +55,13 @@ class Gates(BaseActivity):
Filters the data based on the filters. The return value is a tuple with the first element
being the policy and the second element being the confidence status.
Args:
input_data (dict): The input data. Contains:
filters (dict): The filters to apply.
- input_data (dict): The input data. Contains:
- filters (dict): The filters to apply.
The key is the filter name and the value is the filter configuration.
data (dict[str, Any]): The data to filter.
path_priority (list[str]): The path priority.
- data (dict[str, Any]): The data to filter.
- path_priority (list[str]): The path priority.
Returns:
tuple[str | None, int, str]: (policy, confidence) based in priority
tuple[str | None, int, str]: (policy, confidence, comments) based in priority
list and filter configuration and functions.
"""
@@ -111,13 +111,13 @@ class Gates(BaseActivity):
The return value is a tuple with the first element
being the policy and the second element being the confidence status.
Args:
input_data (dict): The input data. Contains:
filters (dict): The filter configuration to apply.
data (dict[str, Any]): The data to filter.
path_priority (list[str]): The path priority list.
type (str): The type of the gate.
- input_data (dict): The input data. Contains:
- filters (dict): The filter configuration to apply.
- data (dict[str, Any]): The data to filter.
- path_priority (list[str]): The path priority list.
- type (str): The type of the gate.
Returns:
tuple[str | None, int, str]: (policy, confidence) based in priority list
tuple[str | None, int, str]: (policy, confidence, comments) based in priority list
and filter configuration and functions.
"""
@@ -174,13 +174,13 @@ class Gates(BaseActivity):
The return value is a tuple with the first element
being the policy and the second element being the confidence status.
Args:
input_data (dict): The input data. Contains:
filters (dict): The filter configuration to apply.
data (dict[str, Any]): The data to filter.
path_priority (list[str]): The path priority list.
type (str): The type of the gate.
- input_data (dict): The input data. Contains:
- filters (dict): The filter configuration to apply.
- data (dict[str, Any]): The data to filter.
- path_priority (list[str]): The path priority list.
- type (str): The type of the gate.
Returns:
tuple[str | None, int, str]: (policy, confidence) based in priority
tuple[str | None, int, str]: (policy, confidence, comments) based in priority
list and filter configuration and functions.
"""
@@ -233,11 +233,11 @@ class Gates(BaseActivity):
"""
Formats the prediction data.
Args:
input_data (dict): The input data. Contains:
data (dict[str, Any]): The data to format.
timestamp (str): The timestamp of the data.
model_id (str): The id of the model.
prediction_confidence (float): The confidence of the prediction.
- input_data (dict): The input data. Contains:
- data (dict[str, Any]): The data to format.
- timestamp (str): The timestamp of the data.
- model_id (str): The id of the model.
- prediction_confidence (float): The confidence of the prediction.
Returns:
dict: The formatted data.
"""
@@ -260,11 +260,11 @@ class Gates(BaseActivity):
and usefull information in the other fields.
Args:
input_data (dict): The input data. Contains:
timestamp (str): The timestamp of the data.
model_id (str): The id of the model.
prediction_confidence (float): The confidence of the prediction.
comment (str): The comment of the prediction.
- input_data (dict): The input data. Contains:
- timestamp (str): The timestamp of the data.
- model_id (str): The id of the model.
- prediction_confidence (float): The confidence of the prediction.
- comment (str): The comment of the prediction.
Returns:
dict: The formatted data.
"""
@@ -286,8 +286,8 @@ class Gates(BaseActivity):
"""
Gets the last timestamp of the data.
Args:
input_data (dict): The input data. Contains:
data (dict[str, Any]): The data to get the last timestamp from.
- input_data (dict): The input data. Contains:
- data (dict[str, Any]): The data to get the last timestamp from.
Returns:
str: The last timestamp of the data.
"""

16
laborious/activities/mlflow.py
View File

@@ -29,10 +29,10 @@ class MLFlow(BaseActivity):
"""
Access MLFlow model to get the transformed data.
Args:
input_data (dict): The input data. Contains:
data (dict[str, Any]): The data to transform.
model_name (str): The name of the model.
model_retention (int): The retention of the model in minutes.
- input_data (dict): The input data. Contains:
- data (dict[str, Any]): The data to transform.
- model_name (str): The name of the model.
- model_retention (int): The retention time of the model, in minutes.
Returns:
dict[str, Any]: The transformed data.
"""
@@ -67,10 +67,10 @@ class MLFlow(BaseActivity):
"""
Access MLFlow model to get the predicted data.
Args:
input_data (dict): The input data. Contains:
data (dict[str, Any]): The data to predict.
model_name (str): The name of the model.
model_retention (int): The retention of the model.
- input_data (dict): The input data. Contains:
- data (dict[str, Any]): The data to predict.
- model_name (str): The name of the model.
- model_retention (int): The retention time of the model, in minutes.
Returns:
dict[str, Any]: The predicted data.
"""

22
laborious/activities/opc.py
View File

@@ -39,6 +39,17 @@ class OPC(BaseActivity):
def write_data(self, server: str, tag: str, data: Any,
data_type: str, tag_type: str):
"""
Write data to OPC server.
Args:
- server (str): The name of the OPC server.
- tag (str): The tag to write to.
- data (Any): The data to write.
- data_type (str): The data type.
- tag_type (str): The tag type.
"""
try:
self.opc_repository[server].write_data(
tag, data, data_type)
@@ -61,15 +72,14 @@ class OPC(BaseActivity):
operations are optional and independent of each other.
Args:
input_data (dict[str, Any]): The input data. Contains the following keys:
- data (dict[str, Any]): The dataframe that contains the data to write
- input_data(dict[str, Any]): The input data. Contains the following keys:
- data(dict[str, Any]): The dataframe that contains the data to write
to the OPC servers.
- opc_output_config (dict[str, Any]): The OPC writing configuration.
- opc_output_config(dict[str, Any]): The OPC writing configuration.
The keys are the OPC server names and the values contain:
prediction_tags (dict[str, Any]): The tags to write to the OPC servers.
confidence_tags (dict[str, Any]): The tags to write to the OPC servers.
- prediction_tags(dict[str, Any]): The tags to write to the OPC servers.
- confidence_tags(dict[str, Any]): The tags to write to the OPC servers.
Returns:
"""
self.logger.debug("Writing data to OPC servers...")
data = DataFrame(input_data['data'])

4
laborious/utils/connectors_config.py
View File

@@ -1,3 +1,7 @@
"""
Builds the configuration for the connectors.
"""
from os import getenv
import json

14
laborious/utils/filters/conditional_filters.py
View File

@@ -4,6 +4,13 @@ from pandas import DataFrame
def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool:
"""
Returns True if the specific columns have null values, False otherwise.
Args:
- data (DataFrame): The data to filter.
- config (dict): The configuration.
Returns:
bool: True if the specific columns have null values, False otherwise.
"""
return not data[
data['variable'].isin(config['VARIABLES']) & data['value'].isna()].empty
@@ -12,5 +19,12 @@ def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool
def filter_empty_data(data: DataFrame, _config: dict) -> bool:
"""
Returns True if the data is empty, False otherwise.
Args:
- data (DataFrame): The data to filter.
- _config (dict): The configuration.
Returns:
bool: True if the data is empty, False otherwise.
"""
return data.empty

20
laborious/utils/filters/mlflow_filters.py
View File

@@ -3,6 +3,16 @@ from pandas import DataFrame
def api_error_filter(response: dict, _config: dict):
"""
Returns True if the API response is empty or the 'success' key is False, False otherwise.
Args:
- response (dict): The API response.
- _config (dict): The configuration.
Returns:
bool: True if the API response is empty or the 'success' key is False, False otherwise.
"""
if not response:
return True
@@ -13,6 +23,16 @@ def api_error_filter(response: dict, _config: dict):
def nan_values_filter(predictions: DataFrame, _config: dict):
"""
Returns True if the predictions DataFrame contains only NaN values, False otherwise.
Args:
- predictions (DataFrame): The predictions DataFrame.
- _config (dict): The configuration.
Returns:
bool: True if the predictions DataFrame contains only NaN values, False otherwise.
"""
data = predictions.replace({None: np.nan}).drop(
columns=['timestamp'], errors='ignore').infer_objects(copy=False)

31
laborious/utils/repository/model_repository.py
View File

@@ -1,11 +1,11 @@
"""
Model Monitoring Repository
This module contains the ModelMonitoringRepository class, which is responsible
for handling the communication with the Model Monitoring API.
This module contains the ModelMonitoringRepository class,
which is responsible for handling the communication with the Model Monitoring API.
It includes the methods that are used to answer ModelMonitoringService requests using
the Model Monitoring API functions.
It includes the methods that are used to answer ModelMonitoringService
requests using the Model Monitoring API functions.
By Monitoring we mean the evaluation of the performance of models, the generation of reports.
@@ -23,6 +23,18 @@ class MLFlowRepository():
username=username, password=password)
def transform(self, model_name: str, data: pd.DataFrame, model_retention: int):
"""
Transform data using a model.
Parameters:
- model_name (str): The name of the model to use for transformation.
- data (pandas.DataFrame): The data to transform.
- model_retention (int): The number of minutes to keep the model.
Returns:
- dict: A dictionary containing the transformed data.
"""
try:
return {
'success': True,
@@ -40,6 +52,17 @@ class MLFlowRepository():
}
def predict(self, model_name: str, data: pd.DataFrame, model_retention: int):
"""
Predict data using a model.
Parameters:
- model_name (str): The name of the model to use for prediction.
- data (pandas.DataFrame): The data to predict.
- model_retention (int): The number of minutes to keep the model.
Returns:
- dict: A dictionary containing the predicted data.
"""
try:
start_time = datetime.now()
data = self.model_serving.get_cached_predict(

53
laborious/utils/repository/opc_repository.py
View File

@@ -1,12 +1,12 @@
import traceback
from logging import Logger
from datetime import datetime
from pathlib import Path
from asyncua.sync import Client
from asyncua.crypto.security_policies import SecurityPolicyBasic256
from asyncua.ua import DataValue, Variant, VariantType
from logging import Logger
from datetime import datetime
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.notifications.models import NotificationLevel
import traceback
data_type_map = {
'float': {
@@ -33,7 +33,8 @@ data_type_map = {
class OpcRepository():
def __init__(self, name: str, url: str, logger: Logger, notification_handler: NotificationHandler,
def __init__(self, name: str, url: str, logger: Logger,
notification_handler: NotificationHandler,
reconnection_interval: int = 60, server_uri: str = None, cert_path: str = None,
private_key_path: str = None, server_cert_path: str = None):
self.url = url
@@ -57,12 +58,12 @@ class OpcRepository():
Raises:
ValueError: If either the certificate path or private key path is not provided.
Attributes:
cert_path (str): Path to the client's certificate file.
private_key_path (str): Path to the client's private key file.
server_cert_path (str, optional): Path to the server's certificate file.
server_uri (str): The URI of the server to be used as the application URI.
client (opcua.Client): The OPC UA client instance.
logger (logging.Logger): Logger instance for logging information.
- cert_path (str): Path to the client's certificate file.
- private_key_path (str): Path to the client's private key file.
- server_cert_path (str, optional): Path to the server's certificate file.
- server_uri (str): The URI of the server to be used as the application URI.
- client (opcua.Client): The OPC UA client instance.
- logger (logging.Logger): Logger instance for logging information.
Security Settings:
- Security Policy: Basic256
- Secure Channel Timeout: 10,000,000 ms
@@ -105,6 +106,12 @@ class OpcRepository():
return self.try_connect()
def try_connect(self):
"""
Tries to connect to the OPC server.
Returns:
bool: True if the connection was successful, False otherwise.
"""
try:
self.last_reconnection_time = datetime.now()
self.client.connect()
@@ -122,6 +129,9 @@ class OpcRepository():
return False
def disconnect(self):
"""
Disconnects from the OPC server.
"""
if self.client is None:
return
self.client.disconnect()
@@ -129,12 +139,25 @@ class OpcRepository():
self.logger.info('Disconnected from OPC server')
def __del__(self):
"""
Disconnects from the OPC server when the object is destroyed.
"""
try:
self.disconnect()
except Exception as e:
self.logger.error(f"Error in destructor: {e}")
def validate_connection(self):
"""
Validates the connection to the OPC server.
If the connection is not established, it attempts to reconnect.
If the connection is established but the client is not connected,
it attempts to reconnect.
If the connection is established but the client is connected,
it checks if the client is connected to the OPC server.
If the client is not connected, it attempts to reconnect.
If the client is connected, it returns True.
"""
if self.client is None:
return self.connect()
@@ -168,6 +191,16 @@ class OpcRepository():
return True
def write_data(self, node, value, data_type):
"""
Writes data to the OPC server.
If the connection is not established, it attempts to reconnect.
If the connection is established but the client is not connected,
it attempts to reconnect.
If the connection is established but the client is connected,
it checks if the client is connected to the OPC server.
If the client is not connected, it attempts to reconnect.
If the client is connected, it returns True.
"""
if not self.validate_connection():
return
try:

25
laborious/workflows/predictions_batch.py
View File

@@ -19,19 +19,20 @@ class PredictionsBatch():
2. Loads data using a custom query and executes the prediction process
Args:
input_data (dict[str, Any]): The input data for the workflow.
- input_data (dict[str, Any]): The input data for the workflow.
Contains the following keys:
schedule_name (str): The name of the schedule.
model_name (str): The name of the model.
model_id (int): The id of the model.
query (str): The SQL query to be executed to load data.
schema (dict, optional): The schema definition for the data.
table_name (str, optional): The name of the table to process.
input_filters (dict, optional): Filters to be applied during prediction.
mlflow_transform_filters (dict, optional): Filters to be applied during prediction.
mlflow_predict_filters (dict, optional): Filters to be applied during prediction.
model_retention (int, optional): The model retention period in minutes.
path_priority (list[str]): The path priority.
- schedule_name (str): The name of the schedule.
- model_name (str): The name of the model.
- model_id (int): The id of the model.
- query (str): The SQL query to be executed to load data.
- schema (dict, optional): The schema definition for the data.
- table_name (str, optional): The name of the table to process.
- input_filters (dict, optional): Filters to be applied during prediction.
- mlflow_transform_filters (dict, optional): Filters to be applied
during prediction.
- mlflow_predict_filters (dict, optional): Filters to be applied during prediction.
- model_retention (int, optional): The model retention period in minutes.
- path_priority (list[str]): The path priority.
Returns:
None

22
laborious/workflows/sub_workflows/format_and_export_prediction.py
View File

@@ -21,17 +21,17 @@ class FormatAndExportPrediction():
Args:
input_data(dict[str, Any]): The input data for the workflow.
Contains the following keys:
path_flag(str): The path flag to determine the type of prediction to format
data(dict[str, Any]): The data to format
prediction_confidence(float): The prediction confidence to be registered
timestamp(str): The timestamp of the prediction, synchronized with the data
model_id(int): The model id of the prediction
model_name(str): The model name of the prediction
model_retention(str): The model retention of the prediction
comment(str): The comment to be registered
schema(str): The schema of the prediction
table_name(str): The table name of the prediction
opc_output_config(dict[str, Any]): The opc output config of the prediction
- path_flag(str): The path flag to determine the type of prediction to format
- data(dict[str, Any]): The data to format
- prediction_confidence(float): The prediction confidence to be registered
- timestamp(str): The timestamp of the prediction, synchronized with the data
- model_id(int): The model id of the prediction
- model_name(str): The model name of the prediction
- model_retention(str): The model retention of the prediction
- comment(str): The comment to be registered
- schema(str): The schema of the prediction
- table_name(str): The table name of the prediction
- opc_output_config(dict[str, Any]): The opc output config of the prediction
Returns:
bool: True if the workflow was successful, False otherwise.

26
laborious/workflows/sub_workflows/prediction_process.py
View File

@@ -19,19 +19,21 @@ class PredictionProcess():
2. Loads data using a custom query and executes the prediction process
Args:
input_data (dict[str, Any]): The input data for the workflow.
- input_data (dict[str, Any]): The input data for the workflow.
Contains the following keys:
data (dict[str, Any]): The data to be used for the prediction.
schema (str): The schema of the table.
table_name (str): The name of the table.
model_id (int): The id of the model.
input_filters (dict, optional): Filters to be applied during prediction.
mlflow_transform_filters (dict, optional): Filters to be applied during prediction.
mlflow_predict_filters (dict, optional): Filters to be applied during prediction.
model_name (str): The name of the model.
model_retention (int, optional): The model retention period in minutes.
path_priority (list[str]): The path priority.
opc_output_config (dict[str, Any]): The opc output config of the prediction.
- data (dict[str, Any]): The data to be used for the prediction.
- schema (str): The schema of the table.
- table_name (str): The name of the table.
- model_id (int): The id of the model.
- input_filters (dict, optional): Filters to be applied during prediction.
- mlflow_transform_filters (dict, optional): Filters to be
applied during prediction.
- mlflow_predict_filters (dict, optional): Filters to be
applied during prediction.
- model_name (str): The name of the model.
- model_retention (int, optional): The model retention period in minutes.
- path_priority (list[str]): The path priority.
- opc_output_config (dict[str, Any]): The opc output config of the prediction.
Returns:
None