SIENTIAPDE-994

Refactor activity methods and update requirements.txt to enhance functionality and remove deprecated filters. Added detailed docstrings for clarity and improved error handling in data processing workflows.

laborious/activities/base.py

@@ -11,6 +11,14 @@ class BaseActivity:
     def prepare_activity(self, schedule_name: str,
                          model_name: str,
                          model_id: str):
+        """
+        Prepare the activity for the notification handler.
+
+        Args:
+            schedule_name (str): The name of the schedule.
+            model_name (str): The name of the model.
+            model_id (str): The id of the model.
+        """
         self.notification_handler.base_notification.schedule_name = schedule_name
         self.notification_handler.base_notification.model_name = model_name
         self.notification_handler.base_notification.model_id = model_id

laborious/activities/gates.py

@@ -14,15 +14,29 @@ with workflow.unsafe.imports_passed_through():
 
 input_filter_functions = {
     'SPECIFIC_VARIABLES_NULL_VALUES': filter_specific_variables_null_values,
-    'EMPTY_DATA': filter_empty_data
+    'EMPTY_DATA': filter_empty_data,
+    'path_confidence': {
+        'stop': -1,
+        'continue': 2,
+        'repeat': -1
+    }
 }
 
-transform_filter_functions = {
-    'response_filter': {
-        'API_ERROR': api_error_filter,
+mlflow_response_filter_functions = {
+    'API_ERROR': api_error_filter,
+    'path_confidence': {
+        'stop': -1,
+        'continue': 10,
+        'repeat': -1
     },
-    'content_filter': {
-        'NAN_VALUES': nan_values_filter,
+}
+
+mlflow_content_filter_functions = {
+    'NAN_VALUES': nan_values_filter,
+    'path_confidence': {
+        'stop': -1,
+        'continue': 18,
+        'repeat': -1
     }
 }
 
@@ -40,13 +54,13 @@ class Gates(BaseActivity):
             input_data (dict): The input data. Contains:
                 filters (dict): The filters to apply.
                 data (dict[str, Any]): The data to filter.
+                path_priority (list[str]): The path priority.
         Returns:
-            tuple[str, int]: ('stop', -1) if some filter policy is 'stop', ('continue', 2)
-            if no filter policy is 'stop' and some filter policy is 'continue',
-            None if no filter is applied.
+            tuple[str, int]: (policy, confidence) based in priority list and filter configuration and functions.
         """
         filters = input_data['filters']
         data = DataFrame(input_data['data'])
+        path_priority = input_data['path_priority']
 
         filter_output = []
         for fil, config in filters.items():
@@ -63,23 +77,34 @@ class Gates(BaseActivity):
                     attachment_content=trace
                 )
 
-        if 'stop' in filter_output:
-            return 'stop', -1
-        elif 'continue' in filter_output:
-            return 'continue', 2
+        for path_flag in path_priority:
+            if path_flag in filter_output:
+                return path_flag, input_filter_functions['path_confidence'][path_flag]
 
         return None, 0
 
-    @activity.defn(name="mlflow_gate")
-    async def mlflow_gate(self, input_data: dict[str, Any]) -> tuple[str, int]:
-
+    @activity.defn(name="mlflow_response_gate")
+    async def mlflow_response_gate(self, input_data: dict[str, Any]) -> tuple[str, int]:
+        """
+        Filters the data based on the mlflow response 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 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, int]: (policy, confidence) based in priority list and filter configuration and functions.
+        """
         filters = input_data['filters']
-        data = DataFrame(input_data['data'])
+        data = input_data['data']
         gate_type = input_data['type']
+        path_priority = input_data['path_priority']
 
         filter_output = []
         for fil, config in filters.items():
-            if transform_filter_functions['response_filter'][fil](data, config):
+            if mlflow_response_filter_functions[fil](data, config):
                 filter_output.append(config['POLICY'])
                 self.notification_handler.build_and_send_notification(
                     notification_id=f"{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}",
@@ -89,18 +114,36 @@ class Gates(BaseActivity):
                     attachment_content=data['content']['traceback']
                 )
 
-        if 'stop' in filter_output:
-            return 'stop', -1
-        elif 'continue' in filter_output:
-            return 'continue', 10
+        for path_flag in path_priority:
+            if path_flag in filter_output:
+                return path_flag, mlflow_response_filter_functions['path_confidence'][path_flag]
 
-        if gate_type == 'predict':
-            return None, 0
+        return None, 0
 
-        data = DataFrame(data['content'])
+    @activity.defn(name="mlflow_content_gate")
+    async def mlflow_content_gate(self, input_data: dict[str, Any]) -> tuple[str, int]:
+        """
+        Filters the data based on the mlflow content 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 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, int]: (policy, confidence) based in priority list and filter configuration and functions.
+        """
+
+        filters = input_data['filters']
+        data = DataFrame(input_data['data'])
+        gate_type = input_data['type']
+        path_priority = input_data['path_priority']
+
+        filter_output = []
 
         for fil, config in filters.items():
-            if transform_filter_functions['content_filter'][fil](data, config):
+            if mlflow_content_filter_functions[fil](data, config):
                 filter_output.append(config['POLICY'])
                 self.notification_handler.build_and_send_notification(
                     notification_id=f"{gate_type.upper()}_GATE_CONTENT_FILTER__{fil}",
@@ -110,17 +153,26 @@ class Gates(BaseActivity):
                     attachment_content=data.to_string()
                 )
 
-        if 'stop' in filter_output:
-            return 'stop', -1
-        elif 'continue' in filter_output:
-            return 'continue', 18
+        for path_flag in path_priority:
+            if path_flag in filter_output:
+                return path_flag, mlflow_content_filter_functions['path_confidence'][path_flag]
 
         return None, 0
 
     @activity.defn(name="format_prediction")
-    async def format_prediction(self, input_data: dict[str, Any]) -> str:
+    async def format_prediction(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        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.
+        Returns:
+            dict: The formatted data.
+        """
         data = DataFrame(input_data['data'])
-
         data['timestamp'] = input_data['timestamp']
         data['model_id'] = input_data['model_id']
         data['prediction_confidence'] = input_data['prediction_confidence']
@@ -131,7 +183,21 @@ class Gates(BaseActivity):
         return data.to_dict()
 
     @activity.defn(name="format_default_prediction")
-    async def format_default_prediction(self, input_data: dict[str, Any]) -> str:
+    async def format_default_prediction(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Creates and formats the default prediction data, with zero value in prediction,
+        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.
+        Returns:
+            dict: The formatted data.
+        """
+
         return DataFrame({
             'prediction': [0],
             'response_time': [0],

laborious/activities/mlflow.py

@@ -8,7 +8,7 @@ with workflow.unsafe.imports_passed_through():
     from typing import Any
     from logging import Logger
     from sientia_do.notifications.handlers import NotificationHandler
-    from laborious.utils.model_repository import ModelMonitoringRepository
+    from laborious.utils.repository.model_repository import MLFlowRepository
     from sientia_do.notifications.models import NotificationLevel
 
 
@@ -21,12 +21,22 @@ class MLFlow(BaseActivity):
         self.mlflow_username = mlflow_username
         self.mlflow_password = mlflow_password
 
-        self.model_monitoring_repository = ModelMonitoringRepository(
+        self.model_monitoring_repository = MLFlowRepository(
             f"{mlflow_host}:{mlflow_port}", mlflow_username, mlflow_password
         )
 
     @activity.defn(name="request_transform")
     async def request_transform(self, input_data: dict[str, Any]) -> tuple[dict[str, Any], str]:
+        """
+        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.
+        Returns:
+            tuple[dict[str, Any], str]: The transformed data and the latest timestamp of the data.
+        """
         self.logger.info('Transforming data...')
         data = DataFrame(input_data['data'])
         model_name = input_data['model_name']
@@ -50,6 +60,16 @@ class MLFlow(BaseActivity):
 
     @activity.defn(name="request_predict")
     async def request_predict(self, input_data: dict[str, Any]) -> tuple[dict[str, Any], str]:
+        """
+        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.
+        Returns:
+            tuple[dict[str, Any], str]: The predicted data and the latest timestamp of the data.
+        """
         self.logger.info('Predicting data...')
         data = DataFrame(input_data['data'])
         model_name = input_data['model_name']

laborious/activities/opc.py

@@ -1,5 +1,3 @@
-import traceback
-from pandas import DataFrame
 from temporalio import activity, workflow
 
 
@@ -10,6 +8,8 @@ with workflow.unsafe.imports_passed_through():
     from laborious.utils.repository.opc_repository import OpcRepository
     from typing import Any
     from sientia_do.notifications.models import NotificationLevel
+    import traceback
+    from pandas import DataFrame
 
 
 class OPC(BaseActivity):
@@ -41,36 +41,52 @@ class OPC(BaseActivity):
 
     @activity.defn(name='write_opc_data')
     async def write_opc_data(self, input_data: dict[str, Any]):
+        """
+        Write prediction and confidence data to OPC servers. The two writing
+        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 to the OPC servers.
+                opc_servers (list[str]): The OPC servers to write to.
+                opc_output_config (dict[str, Any]): The OPC writing configuration. Contains:
+                    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:
+        """
         data = DataFrame(input_data['data'])
         _opc_servers = input_data['opc_servers']
         opc_output_config = input_data['opc_output_config']
 
-        for tag, config in opc_output_config['prediction_tags'].items():
-            try:
-                self.opc_repository.write_data(
-                    tag, data.head(1)['prediction'].values[0], config['data_type'])
-            except Exception as e:
-                trace = traceback.format_exc()
-                self.notification_handler.build_and_send_notification(
-                    notification_id="WRITE_OPC_PREDICTION_ERROR",
-                    message=f"Error writing data to OPC server: {e}",
-                    block="write_opc_data",
-                    level=NotificationLevel.ERROR,
-                    attachment_content=trace
-                )
-                self.logger.error(trace)
+        if 'prediction_tags' in opc_output_config:
+            for tag, config in opc_output_config['prediction_tags'].items():
+                try:
+                    self.opc_repository.write_data(
+                        tag, data.head(1)['prediction'].values[0], config['data_type'])
+                except Exception as e:
+                    trace = traceback.format_exc()
+                    self.notification_handler.build_and_send_notification(
+                        notification_id="WRITE_OPC_PREDICTION_ERROR",
+                        message=f"Error writing data to OPC server: {e}",
+                        block="write_opc_data",
+                        level=NotificationLevel.ERROR,
+                        attachment_content=trace
+                    )
+                    self.logger.error(trace)
 
-        for tag, config in opc_output_config['confidence_tags'].items():
-            try:
-                self.opc_repository.write_data(
-                    tag, data.head(1)['prediction_confidence'].values[0], config['data_type'])
-            except Exception as e:
-                trace = traceback.format_exc()
-                self.notification_handler.build_and_send_notification(
-                    notification_id="WRITE_OPC_CONFIDENCE_ERROR",
-                    message=f"Error writing data to OPC server: {e}",
-                    block="write_opc_data",
-                    level=NotificationLevel.ERROR,
-                    attachment_content=trace
-                )
-                self.logger.error(trace)
+        if 'confidence_tags' in opc_output_config:
+            for tag, config in opc_output_config['confidence_tags'].items():
+                try:
+                    self.opc_repository.write_data(
+                        tag, data.head(1)['prediction_confidence'].values[0], config['data_type'])
+                except Exception as e:
+                    trace = traceback.format_exc()
+                    self.notification_handler.build_and_send_notification(
+                        notification_id="WRITE_OPC_CONFIDENCE_ERROR",
+                        message=f"Error writing data to OPC server: {e}",
+                        block="write_opc_data",
+                        level=NotificationLevel.ERROR,
+                        attachment_content=trace
+                    )
+                    self.logger.error(trace)

laborious/activities/postgres.py

@@ -135,7 +135,10 @@ class Postgres(BaseActivity):
         Exports data to a postgres table.
 
         Args:
-            input_data (dict[str, Any]): 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.
         """
 
         schema = input_data["schema"]

laborious/utils/filters/base_filter.py

@@ -1,13 +0,0 @@
-from pandas import DataFrame
-
-class Filter:
-    def __init__(self, id: str):
-        self.id = id
-        self.warnings = []
-
-    @staticmethod
-    def method(df: DataFrame) -> DataFrame:
-        raise NotImplementedError
-    
-    def warning(self, message: str):
-        self.warnings.append(f'[{self.id}] - {message}')

laborious/utils/filters/conditional_filters.py

@@ -1,6 +1,5 @@
 from typing import List
 
-from laborious.utils.filters.base_filter import Filter
 from pandas import DataFrame
 
 

laborious/utils/filters/mlflow_filters.py

@@ -1,6 +1,5 @@
 import numpy as np
 from pandas import DataFrame
-from laborious.utils.filters.base_filter import Filter
 
 
 def api_error_filter(response: dict, _config: dict):

laborious/utils/repository/model_repository.py

@@ -16,7 +16,7 @@ from sientia.ModelServing import ModelServing
 from pathlib import Path
 
 
-class ModelMonitoringRepository():
+class MLFlowRepository():
     def __init__(self, host, username, password):
 
         self.model_serving = ModelServing(tracking_uri=host,

laborious/utils/repository/opc_repository.py

@@ -2,13 +2,7 @@ from pathlib import Path
 from asyncua.sync import Client
 from asyncua.crypto.security_policies import SecurityPolicyBasic256
 from asyncua.ua import DataValue, Variant, VariantType
-from datetime import datetime
-from time import sleep
 from logging import Logger
-from statistics import mean, median
-from typing import Callable
-
-from sientia_do.notifications.models import NotificationLevel
 
 data_type_map = {
     'float': VariantType.Float,

laborious/workflows/sub_workflows/format_and_export_prediction.py

@@ -11,7 +11,7 @@ class FormatAndExportPrediction():
     async def run(self, input_data: dict[str, Any]):
         path_flag = input_data['path_flag']
         data = input_data['data']
-        confidence = input_data['confidence']
+        prediction_confidence = input_data['prediction_confidence']
 
         if path_flag is None:
             # proceed with formatting and exporting
@@ -21,7 +21,7 @@ class FormatAndExportPrediction():
                     'data': data,
                     'timestamp': input_data['timestamp'],
                     'model_id': input_data['model_id'],
-                    'prediction_confidence': confidence,
+                    'prediction_confidence': prediction_confidence,
                 }
             )
 
@@ -32,7 +32,7 @@ class FormatAndExportPrediction():
                 {
                     'timestamp': input_data['timestamp'],
                     'model_id': input_data['model_id'],
-                    'prediction_confidence': confidence,
+                    'prediction_confidence': prediction_confidence,
                     'comment': input_data['comment']
                 }
             )

laborious/workflows/sub_workflows/prediction_process.py

@@ -13,8 +13,14 @@ class PredictionProcess():
     @workflow.run
     async def run(self, input_data: dict[str, Any]):
         data = input_data['data']
+        schema = input_data['schema']
+        table_name = input_data['table_name']
+        model = input_data['model']
+        filters = input_data['filters']
+        model_name = input_data['model_name']
+        model_retention = input_data['model_retention']
 
-        path_flag, _confidence = await workflow.execute_activity_method(
+        path_flag, confidence = await workflow.execute_activity_method(
             Gates.input_gate,
             {
                 'filters': input_data['filters'],
@@ -22,38 +28,102 @@ class PredictionProcess():
             }
         )
 
-        if path_flag == 'stop':
-            return
-
-        if path_flag == 'continue':
-            # repeat last prediction
-            await workflow.execute_activity_method(
-                Postgres.repeat_last_prediction,
-                {
-                    'schema': input_data['schema'],
-                    'table_name': input_data['table_name'],
-                    'model': input_data['model']
-                }
-            )
+        if await self.path_flag_handler(
+            data, path_flag, confidence, schema, table_name, model
+        ):
             return
 
         response_data, last_timestamp = await workflow.execute_activity_method(
             MLFlow.transform_data,
             {
                 'data': data,
-                'model_name': input_data['model_name'],
-                'model_retention': input_data['model_retention']
+                'model_name': model_name,
+                'model_retention': model_retention
             }
         )
 
         path_flag, confidence = await workflow.execute_activity_method(
             Gates.mlflow_gate,
             {
-                'filters': input_data['filters'],
+                'filters': filters,
                 'data': response_data,
                 'type': 'transform'
             }
         )
 
-        if path_flag == 'stop':
+        if await self.path_flag_handler(
+            data, path_flag, confidence, schema, table_name, model
+        ):
             return
+
+        response_data = await workflow.execute_activity_method(
+            MLFlow.request_predict,
+            {
+                'data': response_data,
+                'model_name': model_name,
+                'model_retention': model_retention
+            }
+        )
+
+        path_flag, confidence = await workflow.execute_activity_method(
+            Gates.mlflow_gate,
+            {
+                'filters': filters,
+                'data': response_data,
+                'type': 'predict'
+            }
+        )
+
+        if await self.path_flag_handler(
+            data, path_flag, confidence, schema, table_name, model
+        ):
+            return
+
+        await workflow.execute_child_workflow(
+            'format_and_export_prediction',
+            {
+                'path_flag': path_flag,
+                'data': response_data['content'],
+                'prediction_confidence': confidence,
+                'timestamp': response_data['timestamp'],
+                'model_id': model,
+                'model_name': model_name,
+                'model_retention': model_retention
+            }
+        )
+
+    async def path_flag_handler(self, data: dict[str, Any], path_flag: str,
+                                confidence: int, schema: str, table_name: str,
+                                model: str):
+        if path_flag == 'stop':
+            return True
+
+        elif path_flag == 'repeat':
+            # repeat last prediction
+            await workflow.execute_activity_method(
+                Postgres.repeat_last_prediction,
+                {
+                    'schema': schema,
+                    'table_name': table_name,
+                    'model': model
+                }
+            )
+            return True
+
+        elif path_flag == 'continue':
+            # call write workflow
+            workflow.execute_child_workflow(
+                'format_and_export_prediction',
+                {
+                    'path_flag': path_flag,
+                    'data': data,
+                    'prediction_confidence': confidence,
+                    'timestamp': last_timestamp,
+                    'model_id': model,
+                    'model_name': model_name,
+                    'model_retention': model_retention
+                }
+            )
+            return True
+
+        return False