SIENTIAPDE-1110

SIENTIAPDE-1110 Refactor OPC activity to use server IDs instead of names, update values.yaml for OPC_ID, and enhance OpcRepository initialization for improved clarity and consistency.

.github/workflows/quality-gate.yml

@@ -3,19 +3,17 @@ name: Quality gate
 on:
   push:
     branches:
-      - '**'
+      - main
   pull_request:
     branches:
-      - '**'
+      - main
     types: [ opened, synchronize, reopened ]
 
 jobs:
   sonar:
     name: SonarQube Analysis
     runs-on: ubuntu-latest
-    permissions:
+    permissions: write-all
-      contents: read
-
     steps:
       - name: ⬇️ Checkout Code
         uses: actions/checkout@v4
@@ -64,46 +62,13 @@ jobs:
           python -m pip install --upgrade pip
           pip install -r ${{ steps.prepare-requirements.outputs.PROCESSED_REQUIREMENTS_FILE }}
           pip install pytest pytest-cov pytest-asyncio
-      
-      - name: ⬇️ Setup Node.js 18
-        uses: actions/setup-node@v4
-        with:
-          node-version: 18
-
-      - name: 📥 Setup SonarScanner
-        uses: warchant/setup-sonar-scanner@v7
 
       - name: 🧪 Run Tests with Pytest
         run: |
-          set +e 
           pytest tests --junitxml=pytest.xml --cov=laborious --cov-report=xml --cov-report=term
-          PYTEST_EXIT_CODE=$?
-          set -e
 
-          if [ $PYTEST_EXIT_CODE -eq 0 ]; then
+      - name: Run SonarQube Analysis
-            echo "Pytest executado com sucesso."
+        uses: SonarSource/sonarqube-scan-action@v5
-          elif [ $PYTEST_EXIT_CODE -eq 5 ]; then
-            echo "Pytest finalizado com código 5 (Nenhum teste encontrado). Tratando como sucesso para este workflow."
-            exit 0 
-          else
-            echo "Pytest falhou com código de saída $PYTEST_EXIT_CODE."
-            exit $PYTEST_EXIT_CODE 
-          fi
-
-      - name: 📊 Run SonarQube Analysis
         env:
-          SONAR_PROJECT_KEY: ${{ secrets.SONAR_PROJECT_KEY }}
-          SONAR_HOST_URL: ${{ secrets.SONAR_HOST_URL }}
           SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
-        run: |
+          SONAR_HOST_URL: ${{ secrets.SONAR_HOST_URL }}
-          sonar-scanner \
-            -Dsonar.projectKey=$SONAR_PROJECT_KEY \
-            -Dsonar.sources=laborious \
-            -Dsonar.tests=tests \
-            -Dsonar.python.coverage.reportPaths=coverage.xml \
-            -Dsonar.python.xunit.reportPath=pytest.xml \
-            -Dsonar.host.url=$SONAR_HOST_URL \
-            -Dsonar.token=$SONAR_TOKEN \
-            -Dsonar.python.version=3.11 \
-            -Dsonar.projectVersion=1.0.0 \
-            -Dsonar.coverage.exclusions=laborious/worker/worker.py

laborious/activities/activities.py

@@ -3,11 +3,11 @@ from temporalio import activity, workflow
 with workflow.unsafe.imports_passed_through():
     from sientia_do.temporal.activities.postgres import Postgres
     from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.temporal.utils.logger import Logger
     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
 
 
 class Activities(Postgres, MLFlow, Gates, OPC):
@@ -44,10 +44,6 @@ class Activities(Postgres, MLFlow, Gates, OPC):
                      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/gates.py

@@ -3,10 +3,10 @@ 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 sientia_do.temporal.activities.base import BaseActivity
+    from sientia_do.temporal.utils.logger import Logger
     from laborious.utils.filters.mlflow_filters import nan_values_filter, api_error_filter
     from typing import Any
     from laborious.utils.filters.conditional_filters import (
@@ -65,7 +65,13 @@ class Gates(BaseActivity):
                 list and filter configuration and functions.
         """
 
-        self.logger.debug("Performing input gate...")
+        self.debug(f"Input data: {input_data}")
+
+        metadata = input_data['metadata']
+
+        self.debug("Performing input gate...", metadata)
+
+        self.debug(f"Input data: {input_data}", metadata)
 
         filters = input_data['filters']
         data = DataFrame(input_data['data'])
@@ -73,17 +79,17 @@ class Gates(BaseActivity):
 
         filter_output = []
 
-        self.logger.debug(f"Input data:\n {data}")
+        self.debug(f"Input data:\n {data}", metadata)
-        self.logger.debug(f"Filters: {filters}")
+        self.debug(f"Filters: {filters}", metadata)
 
         for fil, config in filters.items():
             if fil not in input_filter_functions:
-                self.logger.error(f"Filter {fil} not found")
+                self.error(f"Filter {fil} not found", metadata)
                 continue
             try:
                 if input_filter_functions[fil](data, config['config']):
-                    self.logger.debug(
+                    self.debug(
-                        f"Data not passed the input filter {fil}:{config}")
+                        f"Data not passed the input filter {fil}:{config}", metadata)
                     filter_output.append(config['policy'])
             except Exception as e:
                 trace = traceback.format_exc()
@@ -97,11 +103,11 @@ class Gates(BaseActivity):
 
         for path_flag in path_priority:
             if path_flag in filter_output:
-                self.logger.debug(f"Input gate result: {path_flag}")
+                self.debug(f"Input gate result: {path_flag}", metadata)
                 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")
+        self.debug("Nothing was filtered by the input gate", metadata)
         return None, 0, ""
 
     @activity.defn(name="mlflow_response_gate")
@@ -121,7 +127,8 @@ class Gates(BaseActivity):
             and filter configuration and functions.
         """
 
-        self.logger.debug("Performing mlflow response gate...")
+        metadata = input_data['metadata']
+        self.debug("Performing mlflow response gate...", metadata)
 
         filters = input_data['filters']
         data = input_data['data']
@@ -130,8 +137,8 @@ class Gates(BaseActivity):
 
         filter_output = []
 
-        self.logger.debug(f"Input data:\n {data}")
+        self.debug(f"Input data:\n {data}", metadata)
-        self.logger.debug(f"Filters: {filters}")
+        self.debug(f"Filters: {filters}", metadata)
 
         comments = []
         for fil, config in filters.items():
@@ -160,11 +167,12 @@ class Gates(BaseActivity):
 
         for path_flag in path_priority:
             if path_flag in filter_output:
-                self.logger.debug(f"Mlflow response gate result: {path_flag}")
+                self.debug(
+                    f"Mlflow response gate result: {path_flag}", metadata)
                 return path_flag, mlflow_response_filter_functions['path_confidence'][path_flag], \
                     ", ".join(comments)
 
-        self.logger.debug("Nothing was filtered by the mlflow response gate")
+        self.debug("Nothing was filtered by the mlflow response gate", metadata)
         return None, 0, ""
 
     @activity.defn(name="mlflow_content_gate")
@@ -184,7 +192,8 @@ class Gates(BaseActivity):
                 list and filter configuration and functions.
         """
 
-        self.logger.debug("Performing mlflow content gate...")
+        metadata = input_data['metadata']
+        self.debug("Performing mlflow content gate...", metadata)
 
         filters = input_data['filters']
         data = DataFrame(input_data['data'])
@@ -193,8 +202,8 @@ class Gates(BaseActivity):
 
         filter_output = []
 
-        self.logger.debug(f"Input data:\n {data}")
+        self.debug(f"Input data:\n {data}", metadata)
-        self.logger.debug(f"Filters: {filters}")
+        self.debug(f"Filters: {filters}", metadata)
 
         for fil, config in filters.items():
             if fil not in mlflow_content_filter_functions:
@@ -221,15 +230,16 @@ class Gates(BaseActivity):
 
         for path_flag in path_priority:
             if path_flag in filter_output:
-                self.logger.debug(f"Mlflow content gate result: {path_flag}")
+                self.debug(
+                    f"Mlflow content gate result: {path_flag}", metadata)
                 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")
+        self.debug("Nothing was filtered by the mlflow content gate", metadata)
         return None, 0, ""
 
     @activity.defn(name="format_prediction")
-    async def format_prediction(self, input_data: dict[str, Any]) -> dict[str, Any]:
+    async def format_prediction(self, input_data: dict[str, Any]) -> dict[Any, Any]:
         """
         Formats the prediction data.
         Args:
@@ -241,7 +251,8 @@ class Gates(BaseActivity):
         Returns:
             dict: The formatted data.
         """
-        self.logger.debug("Formatting prediction...")
+        metadata = input_data['metadata']
+        self.debug("Formatting prediction...", metadata)
 
         data = DataFrame(input_data['data'])
         data['timestamp'] = input_data['timestamp']
@@ -254,7 +265,7 @@ class Gates(BaseActivity):
         return data.to_dict()
 
     @activity.defn(name="format_default_prediction")
-    async def format_default_prediction(self, input_data: dict[str, Any]) -> dict[str, Any]:
+    async def format_default_prediction(self, input_data: dict[str, Any]) -> dict[Any, Any]:
         """
         Creates and formats the default prediction data, with zero value in prediction,
         and usefull information in the other fields.
@@ -269,7 +280,8 @@ class Gates(BaseActivity):
             dict: The formatted data.
         """
 
-        self.logger.debug("Formatting default prediction...")
+        metadata = input_data['metadata']
+        self.debug("Formatting default prediction...", metadata)
 
         return DataFrame({
             'prediction': [0],

laborious/activities/mlflow.py

@@ -6,9 +6,9 @@ from temporalio import activity, workflow
 with workflow.unsafe.imports_passed_through():
     from sientia_do.temporal.activities.base import BaseActivity
     from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.temporal.utils.logger import Logger
     from laborious.utils.repository.model_repository import MLFlowRepository
     from typing import Any
-    from logging import Logger
 
 
 class MLFlow(BaseActivity):
@@ -36,13 +36,19 @@ class MLFlow(BaseActivity):
         Returns:
             dict[str, Any]: The transformed data.
         """
-        self.logger.info('Transforming data...')
+        metadata = input_data['metadata']
+        self.debug('Transforming data...', metadata)
         data = DataFrame(input_data['data'])
         model_name = input_data['model_name']
         model_retention = input_data['model_retention']
 
-        self.logger.debug("Raw input data:")
+        self.debug("Raw input data:", metadata)
-        self.logger.debug(data)
+        self.debug(data, metadata)
+
+        # Sort by created_at in descending order and keep first occurrence of each variable/timestamp pair
+        data = data.sort_values('created_at', ascending=False).drop_duplicates(
+            subset=['variable', 'timestamp'], keep='first'
+        )
 
         data = data.pivot(
             index='timestamp', columns='variable',
@@ -51,14 +57,14 @@ class MLFlow(BaseActivity):
         data.reset_index(inplace=True)
         data.columns.name = None
 
-        self.logger.debug("Processed input data:")
+        self.debug("Processed input data:", metadata)
-        self.logger.debug(data)
+        self.debug(data, metadata)
 
         response_data = self.model_monitoring_repository.transform(
             model_name, data, model_retention)
 
-        self.logger.debug("Response data:")
+        self.debug("Response data:", metadata)
-        self.logger.debug(response_data)
+        self.debug(response_data, metadata)
 
         return response_data
 
@@ -74,18 +80,19 @@ class MLFlow(BaseActivity):
         Returns:
             dict[str, Any]: The predicted data.
         """
-        self.logger.info('Predicting data...')
+        metadata = input_data['metadata']
+        self.debug('Predicting data...', metadata)
         data = DataFrame(input_data['data'])
         model_name = input_data['model_name']
         model_retention = input_data['model_retention']
 
-        self.logger.debug(data)
+        self.debug(data, metadata)
 
         data.replace(np.nan, None, inplace=True)
 
         response_data = self.model_monitoring_repository.predict(
             model_name, data, model_retention)
 
-        self.logger.debug(response_data)
+        self.debug(response_data, metadata)
 
         return response_data

laborious/activities/opc.py

@@ -2,15 +2,17 @@ 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 sientia_do.temporal.activities.base import BaseActivity
+    from sientia_do.temporal.utils.logger import Logger
     from laborious.utils.repository.opc_repository import OpcRepository
     from typing import Any
     import traceback
     from pandas import DataFrame
 
+OPC_WRITTING_ERROR_CONFIDENCE = 12
+
 
 class OPC(BaseActivity):
     def __init__(self, opc_servers: dict[str, dict[str, Any]],
@@ -21,9 +23,9 @@ class OPC(BaseActivity):
         self.opc_servers = opc_servers
 
         self.opc_repository = {}
-        for name, server in opc_servers.items():
+        for id, server in opc_servers.items():
-            self.opc_repository[name] = OpcRepository(
+            self.opc_repository[id] = OpcRepository(
-                name=name,
+                id=server['id'],
                 url=server['url'],
                 logger=self.logger,
                 server_uri=server['server_uri'],
@@ -33,25 +35,28 @@ class OPC(BaseActivity):
                 notification_handler=self.notification_handler,
                 reconnection_interval=server['reconnection_interval'],
             )
-            self.opc_repository[name].connect()
+            self.opc_repository[id].connect()
 
         BaseActivity.__init__(self, logger, notification_handler)
 
-    def write_data(self, server: str, tag: str, data: Any,
+    def write_data(self, server_id: str, tag: str, data: Any,
-                   data_type: str, tag_type: str):
+                   data_type: str, tag_type: str) -> bool:
         """
         Write data to OPC server.
 
         Args:
-            - server (str): The name of the OPC server.
+            - server_id (str): The id of the OPC server.
             - tag (str): The tag to write to.
             - data (Any): The data to write.
             - data_type (str): The data type.
             - tag_type (str): The tag type.
+
+        Returns:
+            - bool: True if the data was written successfully, False otherwise.
         """
 
         try:
-            self.opc_repository[server].write_data(
+            return self.opc_repository[server_id].write_data(
                 tag, data, data_type)
             self.logger.debug(f"Wrote {tag_type} to {tag}")
         except Exception as e:
@@ -63,10 +68,10 @@ class OPC(BaseActivity):
                 level=NotificationLevel.ERROR,
                 attachment_content=trace
             )
-            self.logger.error(trace)
+            raise e
 
     @activity.defn(name='write_opc_data')
-    async def write_opc_data(self, input_data: dict[str, Any]):
+    async def write_opc_data(self, input_data: dict[str, Any]) -> dict[Any, Any]:
         """
         Write prediction and confidence data to OPC servers. The two writing
         operations are optional and independent of each other.
@@ -80,36 +85,72 @@ class OPC(BaseActivity):
                     - 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:
+            - dict[Any, Any]: The data that was written to the OPC servers.
+
         """
-        self.logger.debug("Writing data to OPC servers...")
+        metadata = input_data['metadata']
+        self.debug("Writing data to OPC servers...", metadata)
         data = DataFrame(input_data['data'])
         opc_output_config = input_data['opc_output_config']
-        self.logger.debug(data)
+        self.debug(data, metadata)
 
-        for server, config in opc_output_config.items():
+        success = True
-            if self.opc_repository.get(server) is None:
+
-                self.logger.error(f"OPC server {server} not found")
+        for server_id, config in opc_output_config.items():
+            if self.opc_repository.get(server_id) is None:
+                self.error(f"OPC server {server_id} not found", metadata)
                 continue
 
             if 'prediction_tags' in config:
                 for tag, tag_config in config['prediction_tags'].items():
-                    self.write_data(
+                    success = success and self.write_data(
-                        server=server,
+                        server_id=server_id,
                         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,
+                        server_id=server_id,
                         tag=tag,
                         data=data.head(1)['prediction_confidence'].values[0],
                         data_type=tag_config['data_type'],
                         tag_type='confidence'
                     )
 
+        return self.process_confidence(data, success, metadata)
+
+    def process_confidence(self, data: DataFrame, success: bool, metadata: dict[str, Any]) -> dict[Any, Any]:
+        """
+        Processes the confidence of OPC server write operations and updates the DataFrame accordingly.
+
+        If the write operation was not successful, sets the 'prediction_confidence' column in the DataFrame
+        to a predefined error confidence value and logs a debug message. Otherwise, logs a success message.
+
+        Args:
+            data (DataFrame): The DataFrame containing the data to be processed.
+            success (bool): Indicates whether the data was successfully written to the OPC servers.
+
+        Returns:
+            dict[Any, Any]: The processed data as a dictionary.
+        """
+
+        if not success:
+            data['prediction_confidence'] = OPC_WRITTING_ERROR_CONFIDENCE
+            self.debug(
+                "Some data could not be written to OPC servers, setting confidence to "
+                f"{OPC_WRITTING_ERROR_CONFIDENCE}."
+            )
+
+        else:
+            self.debug("Data written to OPC servers successfully.", metadata)
+
+        return data.to_dict()
+
     def shutdown(self):
         for opc in self.opc_repository.values():
             opc.disconnect()

laborious/utils/connectors_config.py

@@ -35,7 +35,7 @@ def build_opc_config():
 
     return {
         'opc': {
-            'name': getenv('OPC_NAME', 'opc'),
+            'id': getenv('OPC_ID', '1'),
             '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),

laborious/utils/repository/opc_repository.py

@@ -2,9 +2,11 @@ import traceback
 from logging import Logger
 from datetime import datetime
 from pathlib import Path
+from typing import Any
 from asyncua.sync import Client
 from asyncua.crypto.security_policies import SecurityPolicyBasic256
 from asyncua.ua import DataValue, Variant, VariantType
+from regex import F
 from sientia_do.notifications.handlers import NotificationHandler
 from sientia_do.notifications.models import NotificationLevel
 
@@ -33,12 +35,12 @@ data_type_map = {
 
 
 class OpcRepository():
-    def __init__(self, name: str, url: str, logger: Logger,
+    def __init__(self, id: 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.id = id
         self.server_uri = server_uri
         self.cert_path = cert_path
         self.private_key_path = private_key_path
@@ -119,7 +121,7 @@ class OpcRepository():
         except Exception as e:
             trace = traceback.format_exc()
             self.notification_handler.build_and_send_notification(
-                notification_id=f"OPC_CONNECTION_ERROR_{self.name}",
+                notification_id=f"OPC_CONNECTION_ERROR_{self.id}",
                 message=f"Failed to connect to OPC server: {e}",
                 block="opc_repository",
                 level=NotificationLevel.ERROR,
@@ -163,7 +165,7 @@ class OpcRepository():
 
         if self.error_count > 5:
             self.logger.warning(
-                f"OPC server {self.name} will be disconnected due to multiple errors")
+                f"OPC server {self.id} will be disconnected due to multiple errors")
             try:
                 self.disconnect()
             except Exception as e:
@@ -171,7 +173,7 @@ class OpcRepository():
                 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}...")
+                f"Attempting to reconnect to OPC server {self.id}...")
             return self.connect()
 
         if hasattr(self.client, 'aio_obj') and self.client.aio_obj.uaclient.protocol is None or \
@@ -179,18 +181,18 @@ class OpcRepository():
                  self.client.aio_obj.uaclient.protocol.state == "closed"):
 
             self.logger.error(
-                f"OPC server {self.name} is not connected")
+                f"OPC server {self.id} 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}...")
+                    f"Trying to reconnect to OPC server {self.id}...")
                 return self.try_connect()
 
             return False
 
         return True
 
-    def write_data(self, node, value, data_type):
+    def write_data(self, node: str, value: Any, data_type: str) -> bool:
         """
         Writes data to the OPC server.
         If the connection is not established, it attempts to reconnect.
@@ -202,13 +204,13 @@ class OpcRepository():
         If the client is connected, it returns True.
         """
         if not self.validate_connection():
-            return
+            return False
         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}",
+                notification_id=f"OPC_WRITE_GET_NODE_ERROR_{self.id}",
                 message=f"Failed to get node from OPC server: {e}",
                 block="opc_repository",
                 level=NotificationLevel.ERROR,
@@ -216,7 +218,16 @@ class OpcRepository():
             )
             self.logger.error(trace)
             self.error_count += 1
-            return
+            return False
+
+        if data_type not in data_type_map:
+            self.notification_handler.build_and_send_notification(
+                notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}",
+                message=f"Unsupported data type: {data_type}",
+                block="opc_repository",
+                level=NotificationLevel.ERROR
+            )
+            return False
 
         data = data_type_map[data_type]['converter'](value)
         self.logger.info(f'Writing {data} - {type(data)} to {node}')
@@ -228,7 +239,7 @@ class OpcRepository():
         except Exception as e:
             trace = traceback.format_exc()
             self.notification_handler.build_and_send_notification(
-                notification_id=f"OPC_WRITE_DATA_ERROR_{self.name}",
+                notification_id=f"OPC_WRITE_DATA_ERROR_{self.id}",
                 message=f"Failed to write data to OPC server: {e}",
                 block="opc_repository",
                 level=NotificationLevel.ERROR,
@@ -236,5 +247,7 @@ class OpcRepository():
             )
             self.logger.error(trace)
             self.error_count += 1
-            return
+            return False
         self.error_count = 0
+
+        return True

laborious/worker/worker.py

@@ -60,8 +60,6 @@ async def main():
             workflows=[PredictionsBatch, PredictionProcess,
                        FormatAndExportPrediction],
             activities=[
-                # Base
-                activities.prepare_activity,
                 # MLFlow
                 activities.request_predict,
                 activities.request_transform,
@@ -93,7 +91,7 @@ async def main():
         # 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)
+        logger.error(f"An unhandled exception occurred: {e}")
     finally:
         if notification_handler:
             notification_handler.shutdown()

laborious/workflows/predictions_batch.py

@@ -40,27 +40,28 @@ class PredictionsBatch():
             Exception: If any of the required parameters are missing or if the workflow fails.
         """
 
-        await workflow.execute_local_activity_method(
+        metadata = {
-            Activities.prepare_activity,
+            'metadata': {
-            {
                 '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'],
+            {
+                **metadata,
+                'query': input_data['query'],
+            },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=60)
         )
 
         # Prepare input for prediction_process workflow
         prediction_input = {
+            'metadata': metadata,
             'data': data,
             'schema': input_data['schema'],
             'table_name': input_data['table_name'],

laborious/workflows/sub_workflows/format_and_export_prediction.py

@@ -36,6 +36,7 @@ class FormatAndExportPrediction():
         Returns:
             bool: True if the workflow was successful, False otherwise.
         """
+        metadata = input_data['metadata']
         path_flag = input_data['path_flag']
         data = input_data['data']
         prediction_confidence = input_data['prediction_confidence']
@@ -45,6 +46,7 @@ class FormatAndExportPrediction():
             prediction = await workflow.execute_local_activity_method(
                 Activities.format_prediction,
                 {
+                    **metadata,
                     'data': data,
                     'timestamp': input_data['timestamp'],
                     'model_id': input_data['model_id'],
@@ -59,6 +61,7 @@ class FormatAndExportPrediction():
             prediction = await workflow.execute_local_activity_method(
                 Activities.format_default_prediction,
                 {
+                    **metadata,
                     'timestamp': input_data['timestamp'],
                     'model_id': input_data['model_id'],
                     'prediction_confidence': prediction_confidence,
@@ -68,22 +71,11 @@ class FormatAndExportPrediction():
                 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(
+        prediction = await workflow.execute_activity_method(
             Activities.write_opc_data,
             {
+                **metadata,
                 'opc_output_config': input_data['opc_output_config'],
                 'data': prediction
             },
@@ -91,5 +83,15 @@ class FormatAndExportPrediction():
             start_to_close_timeout=timedelta(seconds=60)
         )
 
-        await postgres_holder
+        # write to postgres
-        await opc_holder
+        await workflow.execute_activity_method(
+            Activities.export_data_to_postgres,
+            {
+                **metadata,
+                'schema': input_data['schema'],
+                'table_name': input_data['table_name'],
+                'data': prediction,
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )

laborious/workflows/sub_workflows/prediction_process.py

@@ -41,6 +41,7 @@ class PredictionProcess():
             Exception: If any of the required parameters are missing or if the workflow fails.
         """
 
+        metadata = input_data['metadata']
         data = input_data['data']
         model_id = input_data['model_id']
         model_name = input_data['model_name']
@@ -49,19 +50,26 @@ class PredictionProcess():
         last_timestamp = await workflow.execute_local_activity_method(
             Activities.get_last_timestamp,
             {
+                **metadata,
                 'data': data
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(minutes=1),
         )
 
+        gate_input = {
+            **metadata,
+            'filters': input_data['input_filters'],
+            'data': data,
+            'path_priority': input_data['path_priority']
+        }
+
+        print(f"Metadata e input atualizadas {gate_input}")
+        print(f"Metadata: {metadata}")
+
         path_flag, confidence, comment = await workflow.execute_local_activity_method(
             Activities.input_gate,
-            {
+            gate_input,
-                'filters': input_data['input_filters'],
-                'data': data,
-                'path_priority': input_data['path_priority']
-            },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(minutes=1),
         )
@@ -74,6 +82,7 @@ class PredictionProcess():
         response_data = await workflow.execute_local_activity_method(
             Activities.request_transform,
             {
+                **metadata,
                 'data': data,
                 'model_name': model_name,
                 'model_retention': model_retention
@@ -85,6 +94,7 @@ class PredictionProcess():
         path_flag, confidence, comment = await workflow.execute_local_activity_method(
             Activities.mlflow_response_gate,
             {
+                **metadata,
                 'filters': input_data['mlflow_transform_filters'],
                 'data': response_data,
                 'type': 'transform',
@@ -104,6 +114,7 @@ class PredictionProcess():
         path_flag, confidence, comment = await workflow.execute_local_activity_method(
             Activities.mlflow_content_gate,
             {
+                **metadata,
                 'filters': input_data['mlflow_transform_filters'],
                 'data': transformed_data,
                 'type': 'transform',
@@ -121,6 +132,7 @@ class PredictionProcess():
         response_data = await workflow.execute_local_activity_method(
             Activities.request_predict,
             {
+                **metadata,
                 'data': transformed_data,
                 'model_name': model_name,
                 'model_retention': model_retention
@@ -132,6 +144,7 @@ class PredictionProcess():
         path_flag, confidence, comment = await workflow.execute_local_activity_method(
             Activities.mlflow_response_gate,
             {
+                **metadata,
                 'filters': input_data['mlflow_predict_filters'],
                 'data': response_data,
                 'type': 'predict',
@@ -149,6 +162,7 @@ class PredictionProcess():
         await workflow.execute_child_workflow(
             'format_and_export_prediction',
             {
+                'metadata': metadata,
                 'path_flag': path_flag,
                 'data': response_data['content'],
                 'prediction_confidence': confidence,
@@ -187,13 +201,15 @@ class PredictionProcess():
             bool: True if the prediction should be stopped, False otherwise.
         """
 
+        metadata = input_data['metadata']
+
         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
+        path_flag = path_flag.upper() if path_flag else ''
 
         if path_flag == 'STOP':
             return True
@@ -203,6 +219,7 @@ class PredictionProcess():
             await workflow.execute_activity_method(
                 Activities.repeat_last_prediction,
                 {
+                    **metadata,
                     'schema': schema,
                     'table_name': table_name,
                     'model': model_id,
@@ -218,6 +235,7 @@ class PredictionProcess():
             await workflow.execute_child_workflow(
                 'format_and_export_prediction',
                 {
+                    'metadata': metadata,
                     'path_flag': path_flag,
                     'data': data,
                     'prediction_confidence': confidence,

requirements.txt

@@ -3,5 +3,5 @@ psycopg2-binary
 sqlalchemy
 asyncua
 redis
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.17
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.2.0
 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.1

sonar-project.properties

@@ -0,0 +1,11 @@
+sonar.projectKey=Aignosi_sientia-dataops-laborious_temporal_beaec423-6c42-4f26-8134-b676287b499d
+sonar.projectName=sientia-dataops-laborious_temporal
+sonar.sources=laborious
+sonar.tests=tests
+sonar.projectVersion=1.0.0
+sonar.coverage.exclusions=laborious/worker/worker.py
+sonar.qualitygate.wait=true
+sonar.qualitygate.timeout=300
+sonar.python.coverage.reportPaths=coverage.xml
+sonar.python.xunit.reportPath=pytest.xml
+sonar.python.version=3.11

tests/laborious/activities/test_mlflow.py

@@ -47,10 +47,18 @@ async def test_request_transform(mock_max, mock_dataframe, mlflow):
     # Mock input data
     input_data = {
         'data': [
-            {'timestamp': '2024-01-01', 'variable': 'var1', 'value': 1.0},
+            {'timestamp': '2024-01-01', 'variable': 'var1',
-            {'timestamp': '2024-01-01', 'variable': 'var2', 'value': 2.0},
+                'value': 1.0, 'created_at': '2024-01-01 12:00:00'},
-            {'timestamp': '2024-01-02', 'variable': 'var1', 'value': 3.0},
+            {'timestamp': '2024-01-01', 'variable': 'var2',
-            {'timestamp': '2024-01-02', 'variable': 'var2', 'value': 4.0}
+                'value': 2.0, 'created_at': '2024-01-01 12:00:00'},
+            {'timestamp': '2024-01-02', 'variable': 'var1',
+                'value': 3.0, 'created_at': '2024-01-02 12:00:00'},
+            {'timestamp': '2024-01-02', 'variable': 'var2',
+                'value': 4.0, 'created_at': '2024-01-02 12:00:00'},
+            {'timestamp': '2024-01-02', 'variable': 'var1',
+                'value': 1.0, 'created_at': '2024-01-01 12:00:00'},
+            {'timestamp': '2024-01-02', 'variable': 'var2',
+                'value': 1.0, 'created_at': '2024-01-01 12:00:00'}
         ],
         'model_name': 'test_model',
         'model_retention': 30
@@ -60,6 +68,9 @@ async def test_request_transform(mock_max, mock_dataframe, mlflow):
     expected_response = {'prediction': [0.5, 0.6]}
     mlflow.model_monitoring_repository.transform.return_value = expected_response
 
+    mock_dataframe.return_value.sort_values.return_value = mock_dataframe.return_value
+    mock_dataframe.return_value.drop_duplicates.return_value = mock_dataframe.return_value
+
     # Call the method
     response_data = await mlflow.request_transform(input_data)
 

tests/laborious/activities/test_opc.py

@@ -1,4 +1,5 @@
 from unittest.mock import patch, MagicMock, ANY, call
+from pandas import DataFrame
 from pytest import fixture, mark
 from laborious.activities.opc import NotificationLevel
 
@@ -75,7 +76,7 @@ def test___init__(mock_opc_repository):
 
 @fixture
 @patch("laborious.activities.opc.OpcRepository")
-def opc(_mock_opc_repository):
+def opc(mock_opc_repository):
     servers = {
         'server1': {
             'url': 'http://localhost:8080',
@@ -86,6 +87,9 @@ def opc(_mock_opc_repository):
             'reconnection_interval': 60,
         }
     }
+    mock_opc_repository.write_data = MagicMock(
+        return_value=True
+    )
     return OPC(
         opc_servers=servers,
         logger=MagicMock(),
@@ -103,8 +107,8 @@ WRITE_DATA_CASES = [
 
 @mark.parametrize('tag,data_type,data', WRITE_DATA_CASES)
 def test_write_data_success(opc, tag, data_type, data):
-    opc.write_data(server='server1', tag=tag, data=data,
+    assert opc.write_data(server='server1', tag=tag, data=data,
-                   data_type=data_type, tag_type='prediction')
+                          data_type=data_type, tag_type='prediction')
     opc.opc_repository['server1'].write_data.assert_called_once_with(
         tag, data, data_type)
 
@@ -112,16 +116,22 @@ def test_write_data_success(opc, tag, data_type, data):
 def test_write_data_exception(opc):
     opc.opc_repository['server1'].write_data.side_effect = Exception(
         "Test error")
-    opc.write_data(server='server1', tag='tag1', data=50,
+
-                   data_type='int', tag_type='prediction')
+    try:
-    opc.notification_handler.build_and_send_notification.assert_called_once_with(
+        opc.write_data(server='server1', tag='tag1', data=50,
-        notification_id="WRITE_OPC_PREDICTION_ERROR",
+                       data_type='int', tag_type='prediction')
-        message="Error writing data to OPC server: Test error",
+
-        block="write_opc_data",
+    except Exception:
-        level=NotificationLevel.ERROR,
+        opc.notification_handler.build_and_send_notification.assert_called_once_with(
-        attachment_content=ANY
+            notification_id="WRITE_OPC_PREDICTION_ERROR",
-    )
+            message="Error writing data to OPC server: Test error",
-    opc.logger.error.assert_called_once()
+            block="write_opc_data",
+            level=NotificationLevel.ERROR,
+            attachment_content=ANY
+        )
+
+    else:
+        assert False, "Expected an exception to be raised"
 
 
 @mark.asyncio
@@ -146,9 +156,11 @@ async def test_write_opc_data_success(opc):
 
     # Act
     opc.write_data = MagicMock()
-    await opc.write_opc_data(input_data)
+    opc.process_confidence = MagicMock(return_value={'data': 'data'})
+    output = await opc.write_opc_data(input_data)
 
     # Assert
+    assert output == {'data': 'data'}
     opc.write_data.assert_has_calls([
         call(
             server='server1',
@@ -191,6 +203,18 @@ async def test_write_opc_data_empty_config(opc):
     opc.opc_repository['server1'].write_data.assert_not_called()
 
 
+@mark.parametrize('data,success,expected', [
+    (DataFrame({'prediction_confidence': [0]}), True, 0),
+    (DataFrame({'prediction_confidence': [0]}), False, 12),
+])
+def test_process_confidence(opc, data, success, expected):
+    # Act
+    result = opc.process_confidence(data, success)
+
+    # Assert
+    assert result['prediction_confidence'][0] == expected
+
+
 def test_shutdown(opc):
     opc.shutdown()
     opc.opc_repository['server1'].disconnect.assert_called_once()

tests/laborious/utils/repository/test_opc_repository.py

@@ -224,6 +224,24 @@ def test_write_data_get_node_failed(opc_repository):
     assert opc_repository.error_count == 1
 
 
+def test_write_data_invalid_data_type(opc_repository, mock_client):
+    opc_repository.validate_connection = MagicMock(return_value=True)
+    opc_repository.client = mock_client
+    mock_node = MagicMock()
+    mock_client.get_node.return_value = mock_node
+
+    opc_repository.write_data("ns=2;s=TestNode", 42.0, "invalid_type")
+    opc_repository.validate_connection.assert_called_once()
+    mock_client.get_node.assert_called_once_with("ns=2;s=TestNode")
+
+    opc_repository.notification_handler.build_and_send_notification.assert_called_once_with(
+        notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.name}",
+        message="Unsupported data type: invalid_type",
+        block="opc_repository",
+        level=NotificationLevel.ERROR
+    )
+
+
 def test_write_data(opc_repository, mock_client):
     opc_repository.validate_connection = MagicMock(return_value=True)
     opc_repository.client = mock_client

tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py

@@ -40,17 +40,7 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction):
             retry_policy=ANY,
             start_to_close_timeout=ANY
         )])
-    workflow_mock.execute_activity_method.assert_has_calls([
+
-        call(
-            Activities.export_data_to_postgres,
-            {
-                'schema': input_data['schema'],
-                'table_name': input_data['table_name'],
-                'data': workflow_mock.execute_local_activity_method.return_value
-            },
-            retry_policy=ANY,
-            start_to_close_timeout=ANY
-        )])
     workflow_mock.execute_activity_method.assert_has_calls([
         call(
             Activities.write_opc_data,
@@ -63,6 +53,18 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction):
         )
     ])
 
+    workflow_mock.execute_activity_method.assert_has_calls([
+        call(
+            Activities.export_data_to_postgres,
+            {
+                'schema': input_data['schema'],
+                'table_name': input_data['table_name'],
+                'data': workflow_mock.execute_activity_method.return_value
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )])
+
     assert workflow_mock.execute_activity_method.call_count == 2
     assert workflow_mock.execute_local_activity_method.call_count == 1
 
@@ -99,18 +101,7 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction
             start_to_close_timeout=ANY
         )
     ])
-    workflow_mock.execute_activity_method.assert_has_calls([
+
-        call(
-            Activities.export_data_to_postgres,
-            {
-                'schema': input_data['schema'],
-                'table_name': input_data['table_name'],
-                'data': workflow_mock.execute_local_activity_method.return_value
-            },
-            retry_policy=ANY,
-            start_to_close_timeout=ANY
-        )
-    ])
     workflow_mock.execute_activity_method.assert_has_calls([
         call(
             Activities.write_opc_data,
@@ -123,5 +114,18 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction
         )
     ])
 
+    workflow_mock.execute_activity_method.assert_has_calls([
+        call(
+            Activities.export_data_to_postgres,
+            {
+                'schema': input_data['schema'],
+                'table_name': input_data['table_name'],
+                'data': workflow_mock.execute_activity_method.return_value
+            },
+            retry_policy=ANY,
+            start_to_close_timeout=ANY
+        )
+    ])
+
     assert workflow_mock.execute_activity_method.call_count == 2
     assert workflow_mock.execute_local_activity_method.call_count == 1

values.yaml

@@ -11,7 +11,7 @@ image:
   # This sets the pull policy for images.
   pullPolicy: Always
   # Overrides the image tag whose default is the chart appVersion.
-  tag: "0.1.1"
+  tag: "0.2.2"
 
 # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
 imagePullSecrets:
@@ -123,7 +123,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1097-realizar-testes-basicos-no-cluster-suse-linux"
+    value: "SIENTIAPDE-1110-criar-testes-e-2-e"
   - name: PYTHON_APP
     value: "laborious.worker.worker"
 
@@ -152,10 +152,10 @@ env:
   - name: MLFLOW_PASSWORD
     value: "aignosi"
 
-  - name: OPC_NAME
+  - name: OPC_ID
-    value: "server-1"
+    value: "1"
   - name: OPC_URL
-    value: "opc.tcp://sientia-opc-simulator.sientia.svc.cluster.local:4840"
+    value: "opc.tcp://sientia-opc-simulator-opc.sientia.svc.cluster.local:4840"
 
   - name: KAFKA_BOOTSTRAP_SERVERS
     value: "kafka.kafka.svc.cluster.local:9092"
@@ -178,9 +178,9 @@ ssh:
 
 # kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp
 
-# helm upgrade --install sientia-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.1.0-uat
+# helm upgrade --install sientia-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.4.0-uat
 
 # kubectl create secret generic git-ssh-key-sientia-laborious-worker \
 #   --namespace sientia \
 #   --from-file=ssh-privatekey=git_key \
-#   --type=kubernetes.io/ssh-auth
+#   --type=kubernetes.io/ssh-auth