diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 85a9ebe..c99a54d 100644 --- a/laborious/activities/opc.py +++ b/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( diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 4a436e6..031e171 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/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( - notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}", - message=f"Unsupported data type: {data_type} | metadata: {metadata}", - block="opc_repository", - level=NotificationLevel.ERROR - ) - return False + return False, { + "notification_id": f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}", + "message": f"Unsupported data type: {data_type} | metadata: {metadata}", + "block": "opc_repository", + "level": NotificationLevel.ERROR + } 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, {} diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index 10f820c..e20f7ab 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/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, diff --git a/requirements.txt b/requirements.txt index 9b6205c..a62c1e1 100644 --- a/requirements.txt +++ b/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 diff --git a/values.yaml b/values.yaml index 06be537..8d80a21 100644 --- a/values.yaml +++ b/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"