SIENTIAPDE-1163

Refactor OPC connection handling to improve error notifications and update GITHUB_BRANCH in values.yaml for MongoDB integration. Change requirements.txt to point to local dataops library path.

laborious/activities/opc.py

@@ -22,7 +22,7 @@ class OPC(BaseActivity):
         self.notification_handler = notification_handler
         self.opc_servers = opc_servers
 
-        self.opc_repository = {}
+        self.opc_repository: dict[str, OpcRepository] = {}
         for id, server in opc_servers.items():
             self.opc_repository[id] = OpcRepository(
                 id=server['id'],
@@ -35,8 +35,22 @@ class OPC(BaseActivity):
                 notification_handler=self.notification_handler,
                 reconnection_interval=server['reconnection_interval'],
             )
-            self.opc_repository[id].connect()
+            is_connected, error_data = self.opc_repository[id].connect()
-
+            if not is_connected:
+                self.send_notification(
+                    metadata={
+                        'model_id': '-',
+                        'model_name': '-',
+                        'workflow_name': '-',
+                        'schedule_name': 'INITIALIZATION'
+                    },
+                    notification_id=error_data['notification_id'],
+                    message=error_data['message'],
+                    block=error_data['block'],
+                    level=error_data.get('level', NotificationLevel.ERROR),
+                    attachment_content=error_data.get(
+                        'attachment_content', None)
+                )
         BaseActivity.__init__(self, logger, notification_handler)
 
     def write_data(self, server_id: str, tag: str, data: Any,
@@ -56,8 +70,20 @@ class OPC(BaseActivity):
         """
 
         try:
-            return self.opc_repository[server_id].write_data(
+            is_success, error_data = self.opc_repository[server_id].write_data(
                 tag, data, data_type, self.logger, metadata)
+            if not is_success:
+                self.send_notification(
+                    metadata=metadata,
+                    notification_id=error_data['notification_id'],
+                    message=error_data['message'],
+                    block=error_data['block'],
+                    level=error_data.get('level', NotificationLevel.ERROR),
+                    attachment_content=error_data.get(
+                        'attachment_content', None)
+                )
+                return False
+            return True
         except Exception as e:
             trace = traceback.format_exc()
             self.send_notification(

laborious/utils/repository/opc_repository.py

@@ -91,7 +91,7 @@ class OpcRepository():
         self.client.secure_channel_timeout = 10000000
         self.client.session_timeout = 10000000
 
-    def connect(self):
+    def connect(self) -> tuple[bool, dict[str, Any]]:
         """
         Establishes a connection to the OPC server.
         This method initializes the OPC client using the provided URL and
@@ -107,7 +107,7 @@ class OpcRepository():
         self.logger.info(f'Starting connection to OPC server {self.id}...')
         return self.try_connect()
 
-    def try_connect(self):
+    def try_connect(self) -> tuple[bool, dict[str, Any]]:
         """
         Tries to connect to the OPC server.
 
@@ -118,18 +118,18 @@ class OpcRepository():
         try:
             self.last_reconnection_time = datetime.now()
             self.client.connect()
-            return True
+            return True, {}
         except Exception as e:
             trace = traceback.format_exc()
-            self.notification_handler.build_and_send_notification(
-                notification_id=f"OPC_CONNECTION_ERROR_{self.id}",
-                message=f"Failed to connect to OPC server: {e}",
-                block="opc_repository",
-                level=NotificationLevel.ERROR,
-                attachment_content=trace
-            )
             self.logger.error(trace)
-            return False
+
+            return False, {
+                "notification_id": f"OPC_CONNECTION_ERROR_{self.id}",
+                "message": f"Failed to connect to OPC server: {e}",
+                "block": "opc_repository",
+                "level": NotificationLevel.ERROR,
+                "attachment_content": trace
+            }
 
     def disconnect(self):
         """
@@ -153,7 +153,7 @@ class OpcRepository():
         except Exception as e:
             self.logger.error(f"Error in destructor: {e}")
 
-    def validate_connection(self):
+    def validate_connection(self) -> tuple[bool, dict[str, Any]]:
         """
         Validates the connection to the OPC server.
         If the connection is not established, it attempts to reconnect.
@@ -193,12 +193,12 @@ class OpcRepository():
                     f"Trying to reconnect to OPC server {self.id}...")
                 return self.connect()
 
-            return False
+            return False, {}
 
-        return True
+        return True, {}
 
     def write_data(self, node: str, value: Any, data_type: str,
-                   logger: Logger, metadata: dict[str, Any]) -> bool:
+                   logger: Logger, metadata: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
         """
         Writes data to the OPC server.
         If the connection is not established, it attempts to reconnect.
@@ -209,31 +209,33 @@ class OpcRepository():
         If the client is not connected, it attempts to reconnect.
         If the client is connected, it returns True.
         """
-        if not self.validate_connection():
+
-            return False
+        is_connected, error = self.validate_connection()
+
+        if not is_connected:
+            return False, error
+
         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.id}",
-                message=f"Failed to get node from OPC server: {e} | metadata: {metadata}",
-                block="opc_repository",
-                level=NotificationLevel.ERROR,
-                attachment_content=trace
-            )
             logger.custom_error(trace, metadata.get('schedule_name', 'N/A'))
             self.error_count += 1
-            return False
+            return False, {
+                "notification_id": f"OPC_WRITE_GET_NODE_ERROR_{self.id}",
+                "message": f"Failed to get node from OPC server: {e} | metadata: {metadata}",
+                "block": "opc_repository",
+                "level": NotificationLevel.ERROR,
+                "attachment_content": trace
+            }
 
         if data_type not in data_type_map:
-            self.notification_handler.build_and_send_notification(
+            return False, {
-                notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}",
+                "notification_id": f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}",
-                message=f"Unsupported data type: {data_type} | metadata: {metadata}",
+                "message": f"Unsupported data type: {data_type} | metadata: {metadata}",
-                block="opc_repository",
+                "block": "opc_repository",
-                level=NotificationLevel.ERROR
+                "level": NotificationLevel.ERROR
-            )
+            }
-            return False
 
         data = data_type_map[data_type]['converter'](value)
         logger.custom_info(
@@ -256,16 +258,15 @@ class OpcRepository():
             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.id}",
-                message=f"Failed to write data to OPC server: {e} | metadata: {metadata}",
-                block="opc_repository",
-                level=NotificationLevel.ERROR,
-                attachment_content=trace
-            )
             logger.custom_error(trace, metadata.get('schedule_name', 'N/A'))
             self.error_count += 1
-            return False
+            return False, {
+                "notification_id": f"OPC_WRITE_DATA_ERROR_{self.id}",
+                "message": f"Failed to write data to OPC server: {e} | metadata: {metadata}",
+                "block": "opc_repository",
+                "level": NotificationLevel.ERROR,
+                "attachment_content": trace
+            }
         self.error_count = 0
 
-        return True
+        return True, {}

laborious/workflows/sub_workflows/prediction_process.py

@@ -64,9 +64,6 @@ class PredictionProcess():
             '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,

requirements.txt

@@ -3,5 +3,5 @@ psycopg2-binary
 sqlalchemy
 asyncua
 redis
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.0
+/home/grezewave/Documents/projects/sientia/sientia-dataops-library
 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.1

values.yaml

@@ -123,7 +123,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1154-otimizar-conexao-com-banco-de-dados-e-paralelismo"
+    value: "SIENTIAPDE-1163-alterar-dinamica-de-notificacoes-para-usar-o-mongodb-ao-inves-do-kafka"
   - name: PYTHON_APP
     value: "laborious.worker.worker"