diff --git a/README.md b/README.md index e69de29..d69a75c 100644 --- a/README.md +++ b/README.md @@ -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 +``` diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 799fed5..f0796ff 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -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. """ diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index a209ed1..9b8a58a 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -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. """ diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 997b7c2..11cd77c 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -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']) diff --git a/laborious/utils/connectors_config.py b/laborious/utils/connectors_config.py index 80ed24d..0515cc2 100644 --- a/laborious/utils/connectors_config.py +++ b/laborious/utils/connectors_config.py @@ -1,3 +1,7 @@ +""" +Builds the configuration for the connectors. +""" + from os import getenv import json diff --git a/laborious/utils/filters/conditional_filters.py b/laborious/utils/filters/conditional_filters.py index 57cd3fd..d98cebe 100644 --- a/laborious/utils/filters/conditional_filters.py +++ b/laborious/utils/filters/conditional_filters.py @@ -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 diff --git a/laborious/utils/filters/mlflow_filters.py b/laborious/utils/filters/mlflow_filters.py index 9936018..f6f1efc 100644 --- a/laborious/utils/filters/mlflow_filters.py +++ b/laborious/utils/filters/mlflow_filters.py @@ -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) diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index ed5936b..b3473c6 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -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( diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index e674354..cafcc41 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -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: diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py index 9c4ab61..d80a9d4 100644 --- a/laborious/workflows/predictions_batch.py +++ b/laborious/workflows/predictions_batch.py @@ -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 diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index 68a5958..d904fee 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py @@ -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. diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index df740b3..e4fc8cc 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -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