From 49fa60c66f0137202c4ee0c48fdfb846e56bce60 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 26 May 2025 16:45:50 -0300 Subject: [PATCH] SIENTIAPDE-1081 Enhance documentation across multiple modules with detailed parameter descriptions and usage examples --- README.md | 95 +++++++++++++++++++ laborious/activities/gates.py | 58 +++++------ laborious/activities/mlflow.py | 16 ++-- laborious/activities/opc.py | 22 +++-- laborious/activities/postgres.py | 21 ++-- laborious/utils/connectors_config.py | 4 + .../utils/filters/conditional_filters.py | 14 +++ laborious/utils/filters/mlflow_filters.py | 20 ++++ laborious/utils/logger.py | 8 ++ laborious/utils/policies.py | 17 ++++ .../utils/repository/model_repository.py | 80 +++++++++++----- laborious/utils/repository/opc_repository.py | 53 +++++++++-- laborious/workflows/predictions_batch.py | 25 ++--- .../format_and_export_prediction.py | 22 ++--- .../sub_workflows/prediction_process.py | 26 ++--- 15 files changed, 358 insertions(+), 123 deletions(-) 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 c4f9dd5..7eec6a7 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 ed96192..1783476 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. """ @@ -62,10 +62,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 e6a1dcc..800e16a 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/activities/postgres.py b/laborious/activities/postgres.py index 9fc0ff7..acc5ce2 100644 --- a/laborious/activities/postgres.py +++ b/laborious/activities/postgres.py @@ -6,7 +6,6 @@ with workflow.unsafe.imports_passed_through(): from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import QueuePool - from psycopg2.pool import ThreadedConnectionPool from pandas import read_sql_query, DataFrame from logging import Logger from sientia_do.notifications.handlers import NotificationHandler @@ -49,10 +48,10 @@ class Postgres(BaseActivity): Loads data from a custom query. Args: - query (str): The query to load data from. + - query (str): The query to load data from. Returns: - dict[str, dict]: The data from the query. + dict[str, Any]: The data from the query. """ self.logger.info(f"Fetching data from query: {query}") @@ -95,10 +94,10 @@ class Postgres(BaseActivity): Repeats the last prediction for a given model. Args: - query_items (dict[str, str]): The query items. Contains: - schema (str): The schema of the table. - table_name (str): The name of the table. - model (int): The model to repeat the prediction for. + - query_items (dict[str, str]): The query items. Contains: + - schema (str): The schema of the table. + - table_name (str): The name of the table. + - model (int): The model to repeat the prediction for. Returns: None @@ -144,10 +143,10 @@ class Postgres(BaseActivity): Exports data to a postgres table. Args: - input_data (dict[str, Any]): The data to export. Contains: - schema (str): The schema of the table. - table_name (str): The name of the table. - data (DataFrame): The data to export. + - input_data (dict[str, Any]): The data to export. Contains: + - schema (str): The schema of the table. + - table_name (str): The name of the table. + - data (DataFrame): The data to export. """ self.logger.debug( 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/logger.py b/laborious/utils/logger.py index 42a9cfd..64053fe 100644 --- a/laborious/utils/logger.py +++ b/laborious/utils/logger.py @@ -4,6 +4,14 @@ import sys def get_logger(name: str): + """ + Builds the logger. Gets the log level from the environment variable LOG_LEVEL. + If the log level is not set, it defaults to INFO. + Creates a stream handler and sets the log level. + Sets the formatter for the stream handler. + Adds the stream handler to the logger. + Returns the logger. + """ log_level = getenv('LOG_LEVEL', 'INFO').upper() logger = logging.getLogger(name) diff --git a/laborious/utils/policies.py b/laborious/utils/policies.py index 8c7449a..0ff573d 100644 --- a/laborious/utils/policies.py +++ b/laborious/utils/policies.py @@ -1,3 +1,20 @@ +""" + +Laborious Temporal Workflows Retry Policies + +This module defines retry policies for Temporal workflows in the Laborious system. +These policies ensure reliable execution of ML prediction workflows by automatically +retrying failed operations with exponential backoff. + +The retry_policy variable configures: +- Initial retry interval of 1 second +- Exponential backoff coefficient of 2.0 +- Maximum retry interval capped at 1 minute +- Maximum of 1 retry attempt + +This configuration helps prevent cascading failures while maintaining system responsiveness. + +""" from datetime import timedelta from temporalio.common import RetryPolicy diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index 3896411..ede7488 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -1,9 +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. @@ -13,7 +15,6 @@ import traceback import mlflow import pandas as pd from sientia.ModelServing import ModelServing -from pathlib import Path class MLFlowRepository(): @@ -49,7 +50,8 @@ class MLFlowRepository(): def get_artifact(self, destination: str, search_by: str, run_id: str = None, model_name: str = None, artifact_name: str = None) -> None: """ - Get an artifact in MLflow by experiment or model and save it to a destination path using API. + Get an artifact in MLflow by experiment or model and save it to a + destination path using API. If the artifact is searched by model, the latest production version will be used. Args: @@ -77,7 +79,9 @@ class MLFlowRepository(): Returns: dict: the metrics of the model including MSE and R2 """ - return self.model_serving.get_model_metrics(reference_data=None, real_data=real_data, predictions=predictions, type_flag=flag) + return self.model_serving.get_model_metrics( + reference_data=None, real_data=real_data, + predictions=predictions, type_flag=flag) def get_experiment_by_run_id(self, run_id: str) -> dict: # Get the run information using the run_id @@ -112,25 +116,26 @@ class MLFlowRepository(): Retrain a model with new data. Parameters: - data (pandas.DataFrame): The new data to use for retraining. - model_name (str): The name of the model to retrain. - metrics_list (list): The metrics to be used to compare the models. - compare_metrics (bool): If True, the retrain will only be considered if the new model is better than the current one. - If False, the retrain will always be considered. - split_dataset (bool): If True, the data will be split into X and Y and into training and testing sets. + - data (pandas.DataFrame): The new data to use for retraining. + - model_name (str): The name of the model to retrain. + - metrics_list (list): The metrics to be used to compare the models. + - compare_metrics (bool): If True, the retrain will only be considered if + the new model is better than the current one. + If False, the retrain will always be considered. + - split_dataset (bool): If True, the data will be split into X and Y and into training and testing sets. If False, the data will be used as a unique block for retraining. - update_report (bool): If True, a report will be created with the data of the retrained model. - update_transformation (bool): If True, the model will be updated in the MLflow tracking server. - update_prediction (bool): If True, the prediction model will be updated in the MLflow tracking server. - shuffle_data (bool): If True, the data will be shuffled before splitting. - model_type (str): The type of model to get metrics for. Ex: 'regression', 'classification'. + - update_report (bool): If True, a report will be created with the data of the retrained model. + - update_transformation (bool): If True, the model will be updated in the MLflow tracking server. + - update_prediction (bool): If True, the prediction model will be updated in the MLflow tracking server. + - shuffle_data (bool): If True, the data will be shuffled before splitting. + - model_type (str): The type of model to get metrics for. Ex: 'regression', 'classification'. Returns: - mlflow.sklearn.Model: The retrained prediction model. - mlflow.sklearn.Model: The retrained data model. - mse (float): The mean squared error of the retrained model. - r2 (float): The R-squared score of the retrained model. + - mlflow.sklearn.Model: The retrained prediction model. + - mlflow.sklearn.Model: The retrained data model. + - mse (float): The mean squared error of the retrained model. + - r2 (float): The R-squared score of the retrained model. """ # load predictor model @@ -218,13 +223,16 @@ class MLFlowRepository(): def update_production_model_by_run_id(self, run_id: str, model_name: str) -> dict: # Registrar o modelo - # Aqui estamos assumindo que você já tem um modelo salvo, caso contrário você precisará treiná-lo e salvá-lo primeiro. - # Se o modelo já está registrado, você pode usar o método register_model() ou pyfunc.load_model() para isso. + # Aqui estamos assumindo que você já tem um modelo salvo, + # caso contrário você precisará treiná-lo e salvá-lo primeiro. + # Se o modelo já está registrado, você pode usar o método + # register_model() ou pyfunc.load_model() para isso. mlflow.register_model( f"runs:/{run_id}/prediction_model", model_name) # Colocar a versão do modelo em produção - # Depois de registrar o modelo, precisamos pegar a versão mais recente do modelo e movê-lo para o estágio 'Production' + # Depois de registrar o modelo, precisamos pegar a versão mais + # recente do modelo e movê-lo para o estágio 'Production' client = mlflow.tracking.MlflowClient() # Obter a versão mais recente registrada do modelo @@ -257,10 +265,23 @@ class MLFlowRepository(): return metadata 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, - 'content': self.model_serving.get_cached_transform(model_name, data, model_retention).to_dict() + 'content': self.model_serving.get_cached_transform( + model_name, data, model_retention).to_dict() } except Exception as e: @@ -273,6 +294,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 425f59f..bc5a3e0 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 67cad49..bc343f2 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 de84328..55bd70b 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