SIENTIAPDE-1030

Add unit tests for connectors configuration, logger, workflows, and predictions batch

- Implement tests for MLflow, OPC, and Postgres configuration builders to validate environment variable handling and default values.
- Create tests for the logger to ensure default settings and handler configurations are correct.
- Add comprehensive tests for the FormatAndExportPrediction and PredictionProcess workflows, covering various scenarios including path flags and activity execution.
- Introduce tests for the PredictionsBatch workflow to verify the execution of local activities and child workflows.
- Include a values.yaml file for Kubernetes deployment configuration, specifying image details, service account settings, environment variables, and resource limits.

laborious/activities/activities.py

@@ -0,0 +1,53 @@
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    from laborious.activities.postgres import Postgres
+    from laborious.activities.mlflow import MLFlow
+    from laborious.activities.gates import Gates
+    from laborious.activities.opc import OPC
+    from typing import Any
+    from logging import Logger
+    from sientia_do.notifications.handlers import NotificationHandler
+
+
+class Activities(Postgres, MLFlow, Gates, OPC):
+
+    def __init__(self,
+                 postgres_config: dict[str, Any],
+                 mlflow_config: dict[str, Any],
+                 opc_config: dict[str, Any],
+                 logger: Logger, notification_handler: NotificationHandler):
+
+        # Initialize parent classes
+        Postgres.__init__(self, host=postgres_config['host'],
+                          port=postgres_config['port'],
+                          user=postgres_config['user'],
+                          password=postgres_config['password'],
+                          dbname=postgres_config['dbname'],
+                          min_connections=postgres_config['min_connections'],
+                          max_connections=postgres_config['max_connections'],
+                          logger=logger,
+                          notification_handler=notification_handler)
+
+        MLFlow.__init__(self, mlflow_host=mlflow_config['host'],
+                        mlflow_port=mlflow_config['port'],
+                        mlflow_username=mlflow_config['username'],
+                        mlflow_password=mlflow_config['password'],
+                        logger=logger,
+                        notification_handler=notification_handler)
+
+        Gates.__init__(self, logger=logger,
+                       notification_handler=notification_handler)
+
+        OPC.__init__(self,
+                     opc_servers=opc_config,
+                     logger=logger,
+                     notification_handler=notification_handler)
+
+    @activity.defn(name="prepare_activity")
+    async def prepare_activity(self, input_data: dict[str, Any]):
+        await super().prepare_activity(input_data)
+
+    def shutdown(self):
+        Postgres.close(self)
+        OPC.shutdown(self)

laborious/activities/base.py

@@ -0,0 +1,26 @@
+from typing import Any
+from logging import Logger
+from temporalio import activity
+from sientia_do.notifications.handlers import NotificationHandler
+
+
+class BaseActivity:
+    def __init__(self, logger: Logger, notification_handler: NotificationHandler):
+        self.logger = logger
+        self.notification_handler = notification_handler
+
+    @activity.defn(name="prepare_activity")
+    async def prepare_activity(self, input_data: dict[str, Any]):
+        """
+        Prepare the activity for the notification handler.
+
+        Args:
+            workflow_name (str): The name of the workflow.
+            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.pipeline_name = input_data['workflow_name']
+        self.notification_handler.base_notification.schedule_name = input_data['schedule_name']
+        self.notification_handler.base_notification.model_name = input_data['model_name']
+        self.notification_handler.base_notification.model_id = input_data['model_id']

laborious/activities/gates.py

@@ -0,0 +1,297 @@
+from temporalio import activity, workflow
+
+
+with workflow.unsafe.imports_passed_through():
+    import traceback
+    from logging import Logger
+    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.notifications.models import NotificationLevel
+    from laborious.activities.base import BaseActivity
+    from laborious.utils.filters.mlflow_filters import nan_values_filter, api_error_filter
+    from typing import Any
+    from laborious.utils.filters.conditional_filters import (
+        filter_empty_data,
+        filter_specific_variables_null_values
+    )
+    from pandas import DataFrame
+    from datetime import datetime
+
+input_filter_functions = {
+    'SPECIFIC_VARIABLES_NULL_VALUES': filter_specific_variables_null_values,
+    'EMPTY_DATA': filter_empty_data,
+    'path_confidence': {
+        'STOP': -1,
+        'CONTINUE': 2,
+        'REPEAT': -1
+    }
+}
+
+mlflow_response_filter_functions = {
+    'API_ERROR': api_error_filter,
+    'path_confidence': {
+        'STOP': -1,
+        'CONTINUE': 10,
+        'REPEAT': -1
+    },
+}
+
+mlflow_content_filter_functions = {
+    'NAN_VALUES': nan_values_filter,
+    'path_confidence': {
+        'STOP': -1,
+        'CONTINUE': 18,
+        'REPEAT': -1
+    }
+}
+
+
+class Gates(BaseActivity):
+    def __init__(self, logger: Logger, notification_handler: NotificationHandler):
+        BaseActivity.__init__(self, logger, notification_handler)
+
+    @activity.defn(name="input_gate")
+    async def input_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
+        """
+        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.
+                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.
+        Returns:
+            tuple[str | None, int, str]: (policy, confidence) based in priority
+                list and filter configuration and functions.
+        """
+
+        self.logger.debug("Performing input gate...")
+
+        filters = input_data['filters']
+        data = DataFrame(input_data['data'])
+        path_priority = input_data['path_priority']
+
+        filter_output = []
+
+        self.logger.debug(f"Input data:\n {data}")
+        self.logger.debug(f"Filters: {filters}")
+
+        for fil, config in filters.items():
+            if fil not in input_filter_functions:
+                self.logger.error(f"Filter {fil} not found")
+                continue
+            try:
+                if input_filter_functions[fil](data, config):
+                    self.logger.debug(
+                        f"Data not passed the input filter {fil}:{config}")
+                    filter_output.append(config['POLICY'])
+            except Exception as e:
+                trace = traceback.format_exc()
+                self.notification_handler.build_and_send_notification(
+                    notification_id=f"INTPUT_GATE_ERROR__{fil}",
+                    message=f"Error in filter {fil}:{config}: \n {e}",
+                    block="input_gate",
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace
+                )
+
+        for path_flag in path_priority:
+            if path_flag in filter_output:
+                self.logger.debug(f"Input gate result: {path_flag}")
+                return path_flag, input_filter_functions['path_confidence'][path_flag], \
+                    "Input data with bad quality"
+
+        self.logger.debug("Nothing was filtered by the input gate")
+        return None, 0, ""
+
+    @activity.defn(name="mlflow_response_gate")
+    async def mlflow_response_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
+        """
+        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 | None, int, str]: (policy, confidence) based in priority list
+            and filter configuration and functions.
+        """
+
+        self.logger.debug("Performing mlflow response gate...")
+
+        filters = input_data['filters']
+        data = input_data['data']
+        gate_type = input_data['type']
+        path_priority = input_data['path_priority']
+
+        filter_output = []
+
+        self.logger.debug(f"Input data:\n {data}")
+        self.logger.debug(f"Filters: {filters}")
+
+        comments = []
+        for fil, config in filters.items():
+            if fil not in mlflow_response_filter_functions:
+                continue
+            try:
+                if mlflow_response_filter_functions[fil](data, config):
+                    filter_output.append(config['POLICY'])
+                    comments.append(data['content']['message'])
+                    self.notification_handler.build_and_send_notification(
+                        notification_id=f"{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}",
+                        message=data['content']['message'],
+                        block="mlflow_gate",
+                        level=NotificationLevel.WARNING,
+                        attachment_content=data['content']['traceback']
+                    )
+            except Exception as e:
+                trace = traceback.format_exc()
+                self.notification_handler.build_and_send_notification(
+                    notification_id=f"MLFLOW_GATE_RESPONSE_FILTER__{fil}",
+                    message=f"Error in filter {fil}:{config}: \n {e}",
+                    block="mlflow_gate",
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace
+                )
+
+        for path_flag in path_priority:
+            if path_flag in filter_output:
+                self.logger.debug(f"Mlflow response gate result: {path_flag}")
+                return path_flag, mlflow_response_filter_functions['path_confidence'][path_flag], \
+                    ", ".join(comments)
+
+        self.logger.debug("Nothing was filtered by the mlflow response gate")
+        return None, 0, ""
+
+    @activity.defn(name="mlflow_content_gate")
+    async def mlflow_content_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
+        """
+        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 | None, int, str]: (policy, confidence) based in priority
+                list and filter configuration and functions.
+        """
+
+        self.logger.debug("Performing mlflow content gate...")
+
+        filters = input_data['filters']
+        data = DataFrame(input_data['data'])
+        gate_type = input_data['type']
+        path_priority = input_data['path_priority']
+
+        filter_output = []
+
+        self.logger.debug(f"Input data:\n {data}")
+        self.logger.debug(f"Filters: {filters}")
+
+        for fil, config in filters.items():
+            if fil not in mlflow_content_filter_functions:
+                continue
+            try:
+                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}",
+                        message=f"Data not passed the content filter {fil}:{config}",
+                        block="mlflow_gate",
+                        level=NotificationLevel.WARNING,
+                        attachment_content=data.to_string()
+                    )
+            except Exception as e:
+                trace = traceback.format_exc()
+                self.notification_handler.build_and_send_notification(
+                    notification_id=f"MLFLOW_GATE_CONTENT_FILTER__{fil}",
+                    message=f"Error in filter {fil}:{config}: \n {e}",
+                    block="mlflow_gate",
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace
+                )
+
+        for path_flag in path_priority:
+            if path_flag in filter_output:
+                self.logger.debug(f"Mlflow content gate result: {path_flag}")
+                return path_flag, mlflow_content_filter_functions['path_confidence'][path_flag], \
+                    "Transformed data not passed the content filter"
+
+        self.logger.debug("Nothing was filtered by the mlflow content gate")
+        return None, 0, ""
+
+    @activity.defn(name="format_prediction")
+    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.
+        """
+        self.logger.debug("Formatting prediction...")
+
+        data = DataFrame(input_data['data'])
+        data['timestamp'] = input_data['timestamp']
+        data['model_id'] = input_data['model_id']
+        data['prediction_confidence'] = input_data['prediction_confidence']
+        data['prediction_status'] = 'Good'
+        data['comments'] = ""
+        data.sort_values(by='timestamp', inplace=True)
+
+        return data.to_dict()
+
+    @activity.defn(name="format_default_prediction")
+    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.
+        """
+
+        self.logger.debug("Formatting default prediction...")
+
+        return DataFrame({
+            'prediction': [0],
+            'response_time': [0],
+            'timestamp': [input_data['timestamp']],
+            'model_id': [input_data['model_id']],
+            'prediction_confidence': [input_data['prediction_confidence']],
+            'prediction_status': ['Bad'],
+            'comments': [input_data['comment']]
+        }).to_dict()
+
+    @activity.defn(name="get_last_timestamp")
+    async def get_last_timestamp(self, input_data: dict[str, Any]) -> str:
+        """
+        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.
+        Returns:
+            str: The last timestamp of the data.
+        """
+        data = DataFrame(input_data['data'])
+        if data.empty:
+            return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        return max(data['timestamp'].values.tolist())

laborious/activities/mlflow.py

@@ -0,0 +1,91 @@
+import numpy as np
+from pandas import DataFrame
+from temporalio import activity, workflow
+
+
+with workflow.unsafe.imports_passed_through():
+    from laborious.activities.base import BaseActivity
+    from laborious.utils.repository.model_repository import MLFlowRepository
+    from typing import Any
+    from logging import Logger
+    from sientia_do.notifications.handlers import NotificationHandler
+
+
+class MLFlow(BaseActivity):
+    def __init__(self, mlflow_host: str, mlflow_port: int, mlflow_username: str,
+                 mlflow_password: str, logger: Logger, notification_handler: NotificationHandler):
+        BaseActivity.__init__(self, logger, notification_handler)
+        self.mlflow_host = mlflow_host
+        self.mlflow_port = mlflow_port
+        self.mlflow_username = mlflow_username
+        self.mlflow_password = mlflow_password
+
+        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]) -> dict[str, Any]:
+        """
+        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.
+        Returns:
+            dict[str, Any]: The transformed data.
+        """
+        self.logger.info('Transforming data...')
+        data = DataFrame(input_data['data'])
+        model_name = input_data['model_name']
+        model_retention = input_data['model_retention']
+
+        self.logger.debug("Raw input data:")
+        self.logger.debug(data)
+
+        data = data.pivot(
+            index='timestamp', columns='variable',
+            values='value')
+        data.fillna(np.nan, inplace=True)
+        data.reset_index(inplace=True)
+        data.columns.name = None
+
+        self.logger.debug("Processed input data:")
+        self.logger.debug(data)
+
+        response_data = self.model_monitoring_repository.transform(
+            model_name, data, model_retention)
+
+        self.logger.debug("Response data:")
+        self.logger.debug(response_data)
+
+        return response_data
+
+    @activity.defn(name="request_predict")
+    async def request_predict(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        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:
+            dict[str, Any]: The predicted data.
+        """
+        self.logger.info('Predicting data...')
+        data = DataFrame(input_data['data'])
+        model_name = input_data['model_name']
+        model_retention = input_data['model_retention']
+
+        self.logger.debug(data)
+
+        data.replace(np.nan, None, inplace=True)
+
+        response_data = self.model_monitoring_repository.predict(
+            model_name, data, model_retention)
+
+        self.logger.debug(response_data)
+
+        return response_data

laborious/activities/opc.py

@@ -0,0 +1,105 @@
+from temporalio import activity, workflow
+
+
+with workflow.unsafe.imports_passed_through():
+    from logging import Logger
+    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.notifications.models import NotificationLevel
+    from laborious.activities.base import BaseActivity
+    from laborious.utils.repository.opc_repository import OpcRepository
+    from typing import Any
+    import traceback
+    from pandas import DataFrame
+
+
+class OPC(BaseActivity):
+    def __init__(self, opc_servers: dict[str, dict[str, Any]],
+                 logger: Logger, notification_handler: NotificationHandler):
+
+        self.logger = logger
+        self.notification_handler = notification_handler
+        self.opc_servers = opc_servers
+
+        self.opc_repository = {}
+        for name, server in opc_servers.items():
+            self.opc_repository[name] = OpcRepository(
+                name=name,
+                url=server['url'],
+                logger=self.logger,
+                server_uri=server['server_uri'],
+                cert_path=server['cert_path'],
+                private_key_path=server['private_key_path'],
+                server_cert_path=server['server_cert_path'],
+                notification_handler=self.notification_handler,
+                reconnection_interval=server['reconnection_interval'],
+            )
+            self.opc_repository[name].connect()
+
+        BaseActivity.__init__(self, logger, notification_handler)
+
+    def write_data(self, server: str, tag: str, data: Any,
+                   data_type: str, tag_type: str):
+        try:
+            self.opc_repository[server].write_data(
+                tag, data, data_type)
+            self.logger.debug(f"Wrote {tag_type} to {tag}")
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.notification_handler.build_and_send_notification(
+                notification_id=f"WRITE_OPC_{tag_type.upper()}_ERROR",
+                message=f"Error writing data to OPC server: {e}",
+                block="write_opc_data",
+                level=NotificationLevel.ERROR,
+                attachment_content=trace
+            )
+            self.logger.error(trace)
+
+    @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_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.
+
+        Returns:
+        """
+        self.logger.debug("Writing data to OPC servers...")
+        data = DataFrame(input_data['data'])
+        opc_output_config = input_data['opc_output_config']
+        self.logger.debug(data)
+
+        for server, config in opc_output_config.items():
+            if self.opc_repository.get(server) is None:
+                self.logger.error(f"OPC server {server} not found")
+                continue
+
+            if 'prediction_tags' in config:
+                for tag, tag_config in config['prediction_tags'].items():
+                    self.write_data(
+                        server=server,
+                        tag=tag,
+                        data=data.head(1)['prediction'].values[0],
+                        data_type=tag_config['data_type'],
+                        tag_type='prediction'
+                    )
+            if 'confidence_tags' in config:
+                for tag, tag_config in config['confidence_tags'].items():
+                    self.write_data(
+                        server=server,
+                        tag=tag,
+                        data=data.head(1)['prediction_confidence'].values[0],
+                        data_type=tag_config['data_type'],
+                        tag_type='confidence'
+                    )
+
+    def shutdown(self):
+        for opc in self.opc_repository.values():
+            opc.disconnect()

laborious/activities/postgres.py

@@ -0,0 +1,181 @@
+import traceback
+from temporalio import workflow, activity
+
+from laborious.activities.base import BaseActivity
+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
+    from sientia_do.notifications.models import NotificationLevel
+    from typing import Any
+
+
+class Postgres(BaseActivity):
+    def __init__(self, host: str, port: int,
+                 user: str, password: str, dbname: str,
+                 min_connections: int, max_connections: int,
+                 logger: Logger, notification_handler: NotificationHandler):
+        self.host = host
+        self.port = port
+        self.user = user
+        self.password = password
+        self.dbname = dbname
+
+        # Create SQLAlchemy engine with connection pooling
+        self.engine = create_engine(
+            f'postgresql://{user}:{password}@{host}:{port}/{dbname}',
+            poolclass=QueuePool,
+            pool_size=min_connections,
+            max_overflow=max_connections - min_connections,
+            pool_pre_ping=True
+        )
+        self.session_factory = sessionmaker(bind=self.engine)
+
+        BaseActivity.__init__(self, logger, notification_handler)
+
+    def close(self):
+        self.engine.dispose()
+
+    def __del__(self):
+        self.close()
+
+    @activity.defn(name="load_custom_query")
+    async def load_custom_query(self, query: str) -> dict[str, Any]:
+        """
+        Loads data from a custom query.
+
+        Args:
+            query (str): The query to load data from.
+
+        Returns:
+            dict[str, dict]: The data from the query.
+        """
+        self.logger.info(f"Fetching data from query: {query}")
+
+        data = None
+        with self.session_factory() as session:
+            try:
+                data = read_sql_query(query, self.engine)
+
+            except Exception as e:
+                trace = traceback.format_exc()
+                self.notification_handler.build_and_send_notification(
+                    notification_id="ERROR_LOADING_CUSTOM_QUERY",
+                    message=f"Error fetching data from query: {e}",
+                    block="load_custom_query",
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace
+                )
+
+                self.logger.error(trace)
+
+                return {}
+            finally:
+                session.close()
+
+        if data is None:
+            return {}
+
+        # Converts any datetime datatype columns to string
+        for col in data.select_dtypes(include=['datetime64']).columns:
+            data[col] = data[col].dt.strftime('%Y-%m-%d %H:%M:%S')
+
+        self.logger.info(f"Fetched {len(data)} rows")
+        self.logger.debug(f"Data: \n{data.to_string()}")
+
+        return data.to_dict()
+
+    @activity.defn(name="repeat_last_prediction")
+    async def repeat_last_prediction(self, query_items: dict[str, str]):
+        """
+        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.
+
+        Returns:
+            None
+        """
+        schema = query_items["schema"]
+        table_name = query_items["table_name"]
+        model = query_items["model"]
+
+        repeat_query = f"""
+        INSERT INTO \"{schema}\".{table_name} (model_id, prediction, timestamp, response_time, prediction_status, prediction_confidence, created_at)
+        SELECT model_id, prediction, timestamp, response_time, prediction_status, prediction_confidence, NOW()
+        FROM \"{schema}\".{table_name}
+        WHERE model_id = {model}
+        ORDER BY timestamp DESC
+        LIMIT 1;
+        """
+        self.logger.info(f"Repeating last prediction for model {model}")
+        self.logger.debug(f"Query: {repeat_query}")
+
+        with self.session_factory() as session:
+            try:
+                session.execute(repeat_query)
+                session.commit()
+
+            except Exception as e:
+                trace = traceback.format_exc()
+                self.notification_handler.build_and_send_notification(
+                    notification_id="ERROR_REPEATING_LAST_PREDICTION",
+                    message=f"Error repeating last prediction: {e}",
+                    block="repeat_last_prediction",
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace
+                )
+
+                self.logger.error(trace)
+
+            finally:
+                session.close()
+
+    @activity.defn(name="export_data_to_postgres")
+    async def export_data_to_postgres(self, input_data: dict[str, Any]):
+        """
+        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.
+        """
+
+        self.logger.debug(
+            f"Exporting data to postgres: {input_data['data']}")
+
+        schema = input_data["schema"]
+        table_name = input_data["table_name"]
+        data = DataFrame(input_data["data"])
+
+        with self.session_factory() as session:
+            try:
+                data.to_sql(table_name, self.engine, schema=schema,
+                            if_exists="append", index=False)
+                session.commit()
+
+            except Exception as e:
+                trace = traceback.format_exc()
+                self.notification_handler.build_and_send_notification(
+                    notification_id="ERROR_EXPORTING_DATA_TO_POSTGRES",
+                    message=f"Error exporting data to postgres: {e}",
+                    block="export_data_to_postgres",
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace
+                )
+
+                self.logger.error(trace)
+
+            else:
+                self.logger.debug("Data exported to postgres")
+            finally:
+                session.close()

laborious/utils/connectors_config.py

@@ -0,0 +1,42 @@
+from os import getenv
+import json
+
+
+def build_postgres_config():
+    return {
+        'host': getenv('POSTGRES_HOST', 'localhost'),
+        'port': int(getenv('POSTGRES_PORT', '5432')),
+        'user': getenv('POSTGRES_USER', 'sientia'),
+        'password': getenv('POSTGRES_PASSWORD', 'sientia'),
+        'dbname': getenv('POSTGRES_DBNAME', 'sientia'),
+        'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')),
+        'max_connections': int(getenv('POSTGRES_MAX_CONNECTIONS', '20'))
+    }
+
+
+def build_mlflow_config():
+    return {
+        'host': getenv('MLFLOW_HOST', 'http://localhost'),
+        'port': int(getenv('MLFLOW_PORT', '5080')),
+        'username': getenv('MLFLOW_USERNAME', 'aignosi'),
+        'password': getenv('MLFLOW_PASSWORD', 'aignosi')
+    }
+
+
+def build_opc_config():
+    opc_raw = getenv('OPC_CONFIG', None)
+
+    if opc_raw:
+        return json.loads(opc_raw)
+
+    return {
+        'opc': {
+            'name': getenv('OPC_NAME', 'opc'),
+            'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'),
+            'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'),
+            'cert_path': getenv('OPC_CERT_PATH', None),
+            'private_key_path': getenv('OPC_PRIVATE_KEY_PATH', None),
+            'server_cert_path': getenv('OPC_SERVER_CERT_PATH', None),
+            'reconnection_interval': int(getenv('OPC_RECONNECTION_INTERVAL', '120'))
+        }
+    }

laborious/utils/filters/conditional_filters.py

@@ -0,0 +1,16 @@
+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.
+    """
+    return not data[
+        data['variable'].isin(config['VARIABLES']) & data['value'].isna()].empty
+
+
+def filter_empty_data(data: DataFrame, _config: dict) -> bool:
+    """
+    Returns True if the data is empty, False otherwise.
+    """
+    return data.empty

laborious/utils/filters/mlflow_filters.py

@@ -0,0 +1,22 @@
+import numpy as np
+from pandas import DataFrame
+
+
+def api_error_filter(response: dict, _config: dict):
+    if not response:
+        return True
+
+    if not response['success']:
+        return True
+
+    return False
+
+
+def nan_values_filter(predictions: DataFrame, _config: dict):
+    data = predictions.replace({None: np.nan}).drop(
+        columns=['timestamp'], errors='ignore').infer_objects(copy=False)
+
+    if data.isna().all().all():
+        return True
+
+    return False

laborious/utils/logger.py

@@ -0,0 +1,22 @@
+from os import getenv
+import logging
+import sys
+
+
+def get_logger(name: str):
+    log_level = getenv('LOG_LEVEL', 'INFO').upper()
+
+    logger = logging.getLogger(name)
+    logger.setLevel(log_level)
+    stream_handler = logging.StreamHandler(sys.stdout)
+    stream_handler.setLevel(log_level)
+
+    stream_handler.setFormatter(
+        logging.Formatter(
+            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+        )
+    )
+
+    logger.addHandler(stream_handler)
+
+    return logger

laborious/utils/policies.py

@@ -0,0 +1,9 @@
+from datetime import timedelta
+from temporalio.common import RetryPolicy
+
+retry_policy = RetryPolicy(
+    initial_interval=timedelta(seconds=1),
+    backoff_coefficient=2.0,
+    maximum_interval=timedelta(minutes=1),
+    maximum_attempts=1
+)

laborious/utils/repository/model_repository.py

@@ -0,0 +1,297 @@
+"""
+Model Monitoring Repository
+
+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.
+
+By Monitoring we mean the evaluation of the performance of models, the generation of reports.
+
+"""
+from datetime import datetime
+import traceback
+import mlflow
+import pandas as pd
+from sientia.ModelServing import ModelServing
+
+
+class MLFlowRepository():
+    def __init__(self, host, username, password):
+
+        self.model_serving = ModelServing(tracking_uri=host,
+                                          username=username, password=password)
+
+    def get_current_data_df(self, current_data: pd.DataFrame, model_name: str, target: str):
+        """
+        Get the current data as a DataFrame and update the prediction and target columns
+
+        Parameters:
+        current_data (pd.DataFrame): the current data
+        model_name (str): the name of the model
+        target (str): the target column
+
+        Returns:
+        DataFrame: the current data as a DataFrame
+
+
+        """
+        predictions = current_data['prediction']
+
+        target = current_data[target]
+        current_data = self.model_serving.get_transformed_data(
+            model_name, current_data, by='model')
+        current_data['prediction'] = predictions
+        current_data['target'] = target
+
+        return pd.DataFrame(current_data).dropna()
+
+    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.
+        If the artifact is searched by model, the latest production version will be used.
+
+        Args:
+            destination: The destination path to save the artifact.
+            search_by: The way to search for the artifact ('experiment' or 'model').
+            run_id: The run ID of the experiment (if search_by is "experiment").
+            model_name: The name of the model (if search_by is "model").
+            artifact_name: The path of the artifact to download.
+
+        Returns:
+            artifact: The artifact(.csv) downloaded from MLflow.
+        """
+
+        self.model_serving.get_artifact(destination=destination, search_by=search_by,
+                                        run_id=run_id, model_name=model_name, artifact_name=artifact_name)
+
+    def calculate_model_metrics(self, real_data, predictions, flag):
+        """
+        Function to calculate the metrics of a model using API
+
+        Parameters:
+        real_data (array): the real data
+        predictions (array): the predictions
+
+        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)
+
+    def get_experiment_by_run_id(self, run_id: str) -> dict:
+        # Get the run information using the run_id
+        run = mlflow.get_run(run_id)
+
+        # Extract the experiment ID from the run
+        experiment_id = run.info.experiment_id
+
+        # Get the experiment details using the experiment ID
+        experiment = mlflow.get_experiment(experiment_id)
+        experiment_name = experiment.name
+        return experiment_name
+
+    def get_next_run_name(self, model_name: str) -> str:
+        """ 
+        Function to get the next run number of a specific model
+
+        Parameters:
+        model_name (str): the name of the model
+
+        Returns:
+        str: the next run number
+        """
+
+        runs = mlflow.search_runs(
+            experiment_names=[model_name], order_by=["start_time desc"])
+        next_run_number = len(runs) + 1
+        return f"{model_name}-{next_run_number}"
+
+    def retrain_model(self, data: pd.DataFrame, model_name: str) -> tuple:
+        """
+        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.
+                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'.
+
+
+        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.
+        """
+
+        # load predictor model
+        predictor_uri = f"models:/{model_name}/production"
+        # load transform model
+        latest_production_id = self.model_serving.get_model_run_id(
+            model_name, stage="Production"
+        )
+        transform_uri = self.model_serving.get_model_uri(
+            latest_production_id, prediction=False
+        )
+        # load
+        data_model = mlflow.sklearn.load_model(transform_uri)
+        prediction_model = mlflow.sklearn.load_model(predictor_uri)
+        data_model = data_model.fit(data)
+        treated_data = data_model.predict(data)
+        # align target column with treated_data
+        target_name = data_model.target_variable
+        y = data[target_name]
+        treated_data = pd.merge(
+            treated_data, y, left_index=True, right_index=True)
+        prediction_model = prediction_model.fit(treated_data)
+        # Example usage
+        experiment = self.get_experiment_by_run_id(latest_production_id)
+        pred_model_atributes = vars(prediction_model)  # load class attributes
+        data_model_atributes = vars(data_model)  # load class attributes
+        mlflow.set_experiment(experiment)
+        experiment_description = "Retrain model {model_name} with new data"
+        current_run_name = self.get_next_run_name(experiment)
+        with mlflow.start_run(
+            run_name=current_run_name, description=experiment_description
+        ) as _run:
+            # update transfomation model
+            # fixed parameters
+            for name_atribute, val_atribute in pred_model_atributes.items():
+                if name_atribute != "model":
+                    mlflow.log_param(name_atribute, val_atribute)
+            # update prediction model
+            for name_atribute, val_atribute in data_model_atributes.items():
+                if name_atribute != "model":
+                    mlflow.log_param(name_atribute, val_atribute)
+            # dynamic parameters, including model itself
+            mlflow.sklearn.log_model(data_model, "data_model")
+            file_path = f"laborious/data/raw_data_{model_name}.csv"
+            data.to_csv(
+                f"laborious/data/raw_data_{model_name}.csv", index=True)
+            # log the data raw
+            mlflow.log_artifact(file_path)
+
+            # dynamic parameters, including model itself
+            mlflow.sklearn.log_model(prediction_model, "prediction_model")
+            mlflow.log_param("retrain", True)
+
+            return "Model retrained successfully", experiment
+
+    def get_experiment(self, experiment_name: str) -> int:
+        experiment = mlflow.get_experiment_by_name(experiment_name)
+
+        if experiment is None:
+            raise ValueError(f'Experiment {experiment_name} not found')
+
+        return int(experiment.experiment_id)
+
+    def get_experiment_last_run(self, experiment_id: int) -> str:
+        runs = mlflow.search_runs(
+            experiment_ids=[experiment_id],
+            filter_string="",  # Sem filtro no MLflow ainda
+            output_format="pandas"
+        )
+
+        # Filtrar apenas as runs onde params.retrain == True
+        filtered_runs = runs[runs["params.retrain"] == 'True']
+
+        # Converter a coluna 'end_time' para datetime
+        filtered_runs['end_time'] = pd.to_datetime(filtered_runs['end_time'])
+
+        # Ordenar o DataFrame de forma descendente pela coluna 'end_time'
+        filtered_runs = filtered_runs.sort_values(
+            by='end_time', ascending=False)
+
+        # Pegar a última run_id do DataFrame filtrado e ordenado
+        latest_run_id = filtered_runs.iloc[0]['run_id']
+
+        return latest_run_id
+
+    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.
+        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'
+        client = mlflow.tracking.MlflowClient()
+
+        # Obter a versão mais recente registrada do modelo
+        model_versions = client.get_registered_model(
+            model_name).latest_versions
+        max_version = max(model_versions, key=lambda x: int(x.version)).version
+
+        # Mover a versão mais recente do modelo para o estágio de 'Production'
+        client.transition_model_version_stage(
+            name=model_name,
+            version=max_version,
+            stage="Production",
+            archive_existing_versions=True
+        )
+
+        return {
+            'model_name': model_name,
+            'version': max_version,
+            'mlflow_run_id': run_id
+        }
+
+    def update_production_model(self, experiment: str, model_name: str) -> dict:
+
+        experiment_id = self.get_experiment(experiment)
+        run_id = self.get_experiment_last_run(experiment_id)
+        metadata = self.update_production_model_by_run_id(run_id, model_name)
+
+        metadata['mlflow_experiment_id'] = experiment_id
+
+        return metadata
+
+    def transform(self, model_name: str, data: pd.DataFrame, model_retention: int):
+        try:
+            return {
+                'success': True,
+                'content': self.model_serving.get_cached_transform(
+                    model_name, data, model_retention).to_dict()
+            }
+
+        except Exception as e:
+            return {
+                'success': False,
+                'content': {
+                    'message': str(e),
+                    'traceback': traceback.format_exc()
+                }
+            }
+
+    def predict(self, model_name: str, data: pd.DataFrame, model_retention: int):
+        try:
+            start_time = datetime.now()
+            data = self.model_serving.get_cached_predict(
+                model_name, data, model_retention)[-1:]
+
+            end_time = datetime.now()
+            data = pd.DataFrame(data, columns=['prediction'])
+            data['response_time'] = (end_time - start_time).total_seconds()
+
+            return {
+                'success': True,
+                'content': data.to_dict()
+            }
+
+        except Exception as e:
+            return {
+                'success': False,
+                'content': {
+                    'message': str(e),
+                    'traceback': traceback.format_exc()
+                }
+            }

laborious/utils/repository/opc_repository.py

@@ -0,0 +1,207 @@
+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': {
+        'converter': float,
+        'opc_type': VariantType.Float,
+    },
+    'double': {
+        'converter': float,
+        'opc_type': VariantType.Double,
+    },
+    'int': {
+        'converter': int,
+        'opc_type': VariantType.Int32,
+    },
+    'bool': {
+        'converter': bool,
+        'opc_type': VariantType.Boolean,
+    },
+    'str': {
+        'converter': str,
+        'opc_type': VariantType.String,
+    }
+}
+
+
+class OpcRepository():
+    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
+        self.name = name
+        self.server_uri = server_uri
+        self.cert_path = cert_path
+        self.private_key_path = private_key_path
+        self.server_cert_path = server_cert_path
+        self.logger = logger
+        self.error_count = 0
+        self.reconnection_interval = reconnection_interval
+        self.last_reconnection_time = None
+        self.notification_handler = notification_handler
+        self.client = None
+
+    def set_security(self):
+        """
+        Configures the security settings for the OPC UA client.
+        This method sets up the security policy, certificates, and timeouts
+        required for establishing a secure connection with the OPC UA server.
+        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.
+        Security Settings:
+            - Security Policy: Basic256
+            - Secure Channel Timeout: 10,000,000 ms
+            - Session Timeout: 10,000,000 ms
+        """
+
+        if not all([self.cert_path, self.private_key_path]):
+            raise ValueError(
+                "Certificate and private key paths must be provided for secure connection.")
+        cert = Path(self.cert_path)
+        private_key = Path(self.private_key_path)
+        server_cert = Path(
+            self.server_cert_path) if self.server_cert_path else None
+
+        self.client.application_uri = self.server_uri
+        self.logger.info('Setting security...')
+        self.client.set_security(
+            SecurityPolicyBasic256,
+            certificate=str(cert),
+            private_key=str(private_key),
+            server_certificate=str(server_cert)
+        )
+        self.client.secure_channel_timeout = 10000000
+        self.client.session_timeout = 10000000
+
+    def connect(self):
+        """
+        Establishes a connection to the OPC server.
+        This method initializes the OPC client using the provided URL and
+        sets up security if a certificate path is specified. It then
+        attempts to connect to the server and logs the connection status.
+        Raises:
+            Exception: If the connection to the OPC server fails.
+        """
+
+        self.client = Client(self.url)
+        if self.cert_path:
+            self.set_security()
+        self.logger.info('Starting connection...')
+        return self.try_connect()
+
+    def try_connect(self):
+        try:
+            self.last_reconnection_time = datetime.now()
+            self.client.connect()
+            return True
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.notification_handler.build_and_send_notification(
+                notification_id=f"OPC_CONNECTION_ERROR_{self.name}",
+                message=f"Failed to connect to OPC server: {e}",
+                block="opc_repository",
+                level=NotificationLevel.ERROR,
+                attachment_content=trace
+            )
+            self.logger.error(trace)
+            return False
+
+    def disconnect(self):
+        if self.client is None:
+            return
+        self.client.disconnect()
+        self.client = None
+        self.logger.info('Disconnected from OPC server')
+
+    def __del__(self):
+        try:
+            self.disconnect()
+        except Exception as e:
+            self.logger.error(f"Error in destructor: {e}")
+
+    def validate_connection(self):
+        if self.client is None:
+            return self.connect()
+
+        if self.error_count > 5:
+            self.logger.warning(
+                f"OPC server {self.name} will be disconnected due to multiple errors")
+            try:
+                self.disconnect()
+            except Exception as e:
+                trace = traceback.format_exc()
+                self.logger.error(f"Failed to disconnect from OPC server: {e}")
+                self.logger.error(trace)
+            self.logger.info(
+                f"Attempting to reconnect to OPC server {self.name}...")
+            return self.connect()
+
+        if hasattr(self.client, 'aio_obj') and self.client.aio_obj.uaclient.protocol is None or \
+                (hasattr(self.client.aio_obj.uaclient, 'protocol') and
+                 self.client.aio_obj.uaclient.protocol.state == "closed"):
+
+            self.logger.error(
+                f"OPC server {self.name} is not connected")
+            if (datetime.now() - self.last_reconnection_time).total_seconds(
+            ) > self.reconnection_interval:
+                self.logger.error(
+                    f"Trying to reconnect to OPC server {self.name}...")
+                return self.try_connect()
+
+            return False
+
+        return True
+
+    def write_data(self, node, value, data_type):
+        if not self.validate_connection():
+            return
+        try:
+            node = self.client.get_node(node)
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.notification_handler.build_and_send_notification(
+                notification_id=f"OPC_WRITE_GET_NODE_ERROR_{self.name}",
+                message=f"Failed to get node from OPC server: {e}",
+                block="opc_repository",
+                level=NotificationLevel.ERROR,
+                attachment_content=trace
+            )
+            self.logger.error(trace)
+            self.error_count += 1
+            return
+
+        data = data_type_map[data_type]['converter'](value)
+        self.logger.info(f'Writing {data} - {type(data)} to {node}')
+        ua_data = DataValue(
+            Variant(data, data_type_map[data_type]['opc_type']))
+
+        try:
+            node.write_value(ua_data)
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.notification_handler.build_and_send_notification(
+                notification_id=f"OPC_WRITE_DATA_ERROR_{self.name}",
+                message=f"Failed to write data to OPC server: {e}",
+                block="opc_repository",
+                level=NotificationLevel.ERROR,
+                attachment_content=trace
+            )
+            self.logger.error(trace)
+            self.error_count += 1
+            return
+        self.error_count = 0

laborious/worker/worker.py

@@ -0,0 +1,109 @@
+from temporalio import workflow, client
+from temporalio.worker import Worker
+import sys
+
+with workflow.unsafe.imports_passed_through():
+    import os
+    import asyncio
+    from laborious.workflows.predictions_batch import PredictionsBatch
+    from laborious.workflows.sub_workflows.prediction_process import PredictionProcess
+    from laborious.workflows.sub_workflows.format_and_export_prediction import \
+        FormatAndExportPrediction
+    from laborious.activities.activities import Activities
+    from laborious.utils.logger import get_logger
+    from laborious.utils.connectors_config import (
+        build_postgres_config,
+        build_mlflow_config,
+        build_opc_config
+    )
+    from sientia_do.notifications.handlers import NotificationHandler
+
+
+async def main():
+    host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
+    logger = get_logger(__name__)
+
+    logger.info('Starting Worker...')
+
+    logger.info('Starting Notification Handler...')
+
+    notification_handler = NotificationHandler(
+        servers=os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'http://localhost:9092'),
+        logger=logger,
+        project_name=os.getenv('PROJECT_NAME', 'laborious'),
+        pipeline_name='-',
+        trigger_name='-',
+        model_name='-',
+        model='-'
+    )
+
+    logger.info('Starting Activities...')
+
+    activities = Activities(
+        postgres_config=build_postgres_config(),
+        mlflow_config=build_mlflow_config(),
+        opc_config=build_opc_config(),
+        logger=logger,
+        notification_handler=notification_handler
+    )
+
+    logger.info('Starting Temporal Client...')
+
+    temporal_client = await client.Client.connect(
+        target_host=host,
+        namespace=os.getenv('TEMPORAL_NAMESPACE', 'default')
+    )
+
+    logger.info('Starting Workers...')
+
+    workers = [
+        Worker(
+            temporal_client,
+            task_queue='predictions-queue',
+            workflows=[PredictionsBatch, PredictionProcess,
+                       FormatAndExportPrediction],
+            activities=[
+                # Base
+                activities.prepare_activity,
+                # MLFlow
+                activities.request_predict,
+                activities.request_transform,
+                # Gates
+                activities.input_gate,
+                activities.mlflow_response_gate,
+                activities.mlflow_content_gate,
+                activities.format_prediction,
+                activities.format_default_prediction,
+                activities.get_last_timestamp,
+                # OPC
+                activities.write_opc_data,
+                # Postgres
+                activities.load_custom_query,
+                activities.repeat_last_prediction,
+                activities.export_data_to_postgres
+            ]
+        )
+    ]
+
+    handlers = []
+    for w in workers:
+        handlers.append(w.run())
+
+    logger.info('Workers started successfully')
+
+    try:
+        # This will run the workers and wait for them to complete.
+        # If an exception occurs in any of the worker handlers, it will be propagated here.
+        await asyncio.gather(*handlers)
+    except BaseException as e:
+        logger.error("An unhandled exception occurred: %s", e, exc_info=True)
+    finally:
+        if notification_handler:
+            notification_handler.shutdown()
+        if activities:
+            activities.shutdown()
+        # Exit with a non-zero status code to indicate failure to Kubernetes
+        sys.exit(1)
+
+if __name__ == '__main__':
+    asyncio.run(main())

laborious/workflows/predictions_batch.py

@@ -0,0 +1,89 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from laborious.activities.activities import Activities
+    from typing import Any
+    from laborious.utils.policies import retry_policy
+    from datetime import timedelta
+
+
+@workflow.defn(name="predictions_batch")
+class PredictionsBatch():
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]):
+        """
+        This workflow runs a batch of predictions based on the input data.
+
+        The workflow executes in two main steps:
+        1. Prepares the activity with schedule and model information
+        2. Loads data using a custom query and executes the prediction process
+
+        Args:
+            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.
+        Returns:
+            None
+
+        Raises:
+            Exception: If any of the required parameters are missing or if the workflow fails.
+        """
+
+        await workflow.execute_local_activity_method(
+            Activities.prepare_activity,
+            {
+                'schedule_name': input_data['schedule_name'],
+                'model_name': input_data['model_name'],
+                'model_id': input_data['model_id'],
+                'workflow_name': 'predictions_batch'
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        data = await workflow.execute_local_activity_method(
+            Activities.load_custom_query,
+            input_data['query'],
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        # Prepare input for prediction_process workflow
+        prediction_input = {
+            'data': data,
+            'schema': input_data['schema'],
+            'table_name': input_data['table_name'],
+            'model_id': input_data['model_id'],
+            'model_name': input_data['model_name'],
+            'input_filters': input_data.get('input_filters', {
+                'EMPTY_DATA': {
+                    'POLICY': 'STOP'
+                }
+            }),
+            'mlflow_transform_filters': input_data.get('mlflow_transform_filters', {
+                'API_ERROR': {
+                    'POLICY': 'STOP'
+                }
+            }),
+            'mlflow_predict_filters': input_data.get('mlflow_predict_filters', {
+                'API_ERROR': {
+                    'POLICY': 'STOP'
+                }
+            }),
+            'model_retention': input_data.get('model_retention', 60),
+            'path_priority': input_data.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT']),
+            'opc_output_config': input_data.get('opc_output_config', {})
+        }
+
+        await workflow.execute_child_workflow(
+            'prediction_process', prediction_input)

laborious/workflows/sub_workflows/format_and_export_prediction.py

@@ -0,0 +1,95 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from laborious.activities.activities import Activities
+    from typing import Any
+    from datetime import timedelta
+    from laborious.utils.policies import retry_policy
+
+
+@workflow.defn(name="format_and_export_prediction")
+class FormatAndExportPrediction():
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]):
+        """
+        This workflow formats and exports predictions based on path_flag:
+        - If path_flag is None: formats prediction
+            using input data, timestamp, model_id and confidence
+        - If path_flag exists: creates default prediction
+            with timestamp, model_id, confidence and comment
+        Finally exports formatted prediction to postgres table
+        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
+
+        Returns:
+            bool: True if the workflow was successful, False otherwise.
+        """
+        path_flag = input_data['path_flag']
+        data = input_data['data']
+        prediction_confidence = input_data['prediction_confidence']
+
+        if path_flag is None:
+            # proceed with formatting and exporting
+            prediction = await workflow.execute_local_activity_method(
+                Activities.format_prediction,
+                {
+                    'data': data,
+                    'timestamp': input_data['timestamp'],
+                    'model_id': input_data['model_id'],
+                    'prediction_confidence': prediction_confidence,
+                },
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=60)
+            )
+
+        else:
+            # create default prediction
+            prediction = await workflow.execute_local_activity_method(
+                Activities.format_default_prediction,
+                {
+                    'timestamp': input_data['timestamp'],
+                    'model_id': input_data['model_id'],
+                    'prediction_confidence': prediction_confidence,
+                    'comment': input_data['comment']
+                },
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=60)
+            )
+
+        # write to postgres
+        postgres_holder = workflow.execute_activity_method(
+            Activities.export_data_to_postgres,
+            {
+                'schema': input_data['schema'],
+                'table_name': input_data['table_name'],
+                'data': prediction
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        # write to opc
+        opc_holder = workflow.execute_activity_method(
+            Activities.write_opc_data,
+            {
+                'opc_output_config': input_data['opc_output_config'],
+                'data': prediction
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        await postgres_holder
+        await opc_holder

laborious/workflows/sub_workflows/prediction_process.py

@@ -0,0 +1,233 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from laborious.activities.activities import Activities
+    from typing import Any
+    from laborious.utils.policies import retry_policy
+    from datetime import timedelta
+
+
+@workflow.defn(name="prediction_process")
+class PredictionProcess():
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]):
+        """
+        This workflow runs a prediction process based on the input data.
+
+        The workflow executes in two main steps:
+        1. Prepares the activity with schedule and model information
+        2. Loads data using a custom query and executes the prediction process
+
+        Args:
+            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.
+        Returns:
+            None
+
+        Raises:
+            Exception: If any of the required parameters are missing or if the workflow fails.
+        """
+
+        data = input_data['data']
+        model_id = input_data['model_id']
+        model_name = input_data['model_name']
+        model_retention = input_data['model_retention']
+
+        last_timestamp = await workflow.execute_local_activity_method(
+            Activities.get_last_timestamp,
+            {
+                'data': data
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(minutes=1),
+        )
+
+        path_flag, confidence, comment = await workflow.execute_local_activity_method(
+            Activities.input_gate,
+            {
+                'filters': input_data['input_filters'],
+                'data': data,
+                'path_priority': input_data['path_priority']
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(minutes=1),
+        )
+
+        if await self.path_flag_handler(
+            data, path_flag, input_data, confidence, last_timestamp, comment
+        ):
+            return
+
+        response_data = await workflow.execute_local_activity_method(
+            Activities.request_transform,
+            {
+                'data': data,
+                'model_name': model_name,
+                'model_retention': model_retention
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(minutes=1),
+        )
+
+        path_flag, confidence, comment = await workflow.execute_local_activity_method(
+            Activities.mlflow_response_gate,
+            {
+                'filters': input_data['mlflow_transform_filters'],
+                'data': response_data,
+                'type': 'transform',
+                'path_priority': input_data['path_priority']
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(minutes=1),
+        )
+
+        if await self.path_flag_handler(
+            data, path_flag, input_data, confidence, last_timestamp, comment
+        ):
+            return
+
+        transformed_data = response_data['content']
+
+        path_flag, confidence, comment = await workflow.execute_local_activity_method(
+            Activities.mlflow_content_gate,
+            {
+                'filters': input_data['mlflow_transform_filters'],
+                'data': transformed_data,
+                'type': 'transform',
+                'path_priority': input_data['path_priority']
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(minutes=1),
+        )
+
+        if await self.path_flag_handler(
+            data, path_flag, input_data, confidence, last_timestamp, comment
+        ):
+            return
+
+        response_data = await workflow.execute_local_activity_method(
+            Activities.request_predict,
+            {
+                'data': transformed_data,
+                'model_name': model_name,
+                'model_retention': model_retention
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(minutes=1),
+        )
+
+        path_flag, confidence, comment = await workflow.execute_local_activity_method(
+            Activities.mlflow_response_gate,
+            {
+                'filters': input_data['mlflow_predict_filters'],
+                'data': response_data,
+                'type': 'predict',
+                'path_priority': input_data['path_priority']
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(minutes=1),
+        )
+
+        if await self.path_flag_handler(
+            data, path_flag, input_data, confidence, last_timestamp, comment
+        ):
+            return
+
+        await workflow.execute_child_workflow(
+            'format_and_export_prediction',
+            {
+                'path_flag': path_flag,
+                'data': response_data['content'],
+                'prediction_confidence': confidence,
+                'timestamp': last_timestamp,
+                'model_id': model_id,
+                'model_name': model_name,
+                'model_retention': model_retention,
+                'opc_output_config': input_data['opc_output_config'],
+                'schema': input_data['schema'],
+                'table_name': input_data['table_name'],
+                'comment': comment
+            }
+        )
+
+    async def path_flag_handler(self, data: dict[str, Any], path_flag: str,
+                                input_data: dict[str, Any], confidence: int,
+                                last_timestamp: str, comment: str):
+        """
+        This function handles the path flag and the confidence of the prediction.
+        It returns True if the prediction should be stopped. If path_flag is 'repeat',
+        it repeats the last prediction.
+        If path_flag is 'continue', it calls the write workflow. If path_flag is 'stop',
+        it stops the prediction process.
+        Args:
+            data (dict[str, Any]): The data to be used for the prediction.
+            path_flag (str): The path flag to determine the type of prediction to format
+            confidence (int): The confidence of the prediction
+            schema (str): The schema of the prediction
+            table_name (str): The table name of the prediction
+            model_id (int): The model id of the prediction
+            last_timestamp (str): The timestamp of the last prediction
+            model_name (str): The model name of the prediction
+            model_retention (int): The model retention of the prediction
+            comment (str): The comment of the prediction
+        Returns:
+            bool: True if the prediction should be stopped, False otherwise.
+        """
+
+        schema = input_data['schema']
+        table_name = input_data['table_name']
+        model_id = input_data['model_id']
+        model_name = input_data['model_name']
+        model_retention = input_data['model_retention']
+
+        path_flag = path_flag.upper() if path_flag else None
+
+        if path_flag == 'STOP':
+            return True
+
+        elif path_flag == 'REPEAT':
+            # repeat last prediction
+            await workflow.execute_activity_method(
+                Activities.repeat_last_prediction,
+                {
+                    'schema': schema,
+                    'table_name': table_name,
+                    'model_id': model_id
+                },
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(minutes=1),
+            )
+            return True
+
+        elif path_flag == 'CONTINUE':
+            # call write workflow
+            await workflow.execute_child_workflow(
+                'format_and_export_prediction',
+                {
+                    'path_flag': path_flag,
+                    'data': data,
+                    'prediction_confidence': confidence,
+                    'timestamp': last_timestamp,
+                    'model_id': model_id,
+                    'model_name': model_name,
+                    'model_retention': model_retention,
+                    'schema': schema,
+                    'table_name': table_name,
+                    'comment': comment,
+                    'opc_output_config': input_data['opc_output_config']
+                }
+            )
+            return True
+
+        return False