diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 7a0c28f..f3730a4 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -99,6 +99,55 @@ class OPC(BaseActivity): ) raise e + def validate_server(self, server_id: str, metadata: dict[str, Any]) -> bool: + if self.opc_repository.get(server_id) is None: + message = f"OPC server {server_id} not found to perform write operation." + self.send_notification( + metadata=metadata, + notification_id="OPC_SERVER_NOT_FOUND", + message=message, + block="write_opc_data", + level=NotificationLevel.ERROR, + attachment_content=f"OPC servers: {list(self.opc_repository.keys())}" + ) + return False + return True + + def manage_output_tags( + self, server_id: str, config: dict[str, Any], data: DataFrame, + metadata: dict[str, Any], success: bool) -> bool: + if 'prediction_tags' in config: + for tag, tag_config in config['prediction_tags'].items(): + local_success = self.write_data( + server_id=server_id, + tag=tag, + data=data.head(1)['prediction'].values[0], + data_type=tag_config['data_type'], + tag_type='prediction', + metadata=metadata + ) + if local_success: + self.info( + f"Prediction data written to OPC server {server_id} for tag {tag}.", metadata) + success = success and local_success + + if 'confidence_tags' in config: + for tag, tag_config in config['confidence_tags'].items(): + local_success = self.write_data( + server_id=server_id, + tag=tag, + data=data.head(1)['prediction_confidence'].values[0], + data_type=tag_config['data_type'], + tag_type='confidence', + metadata=metadata + ) + if local_success: + self.info( + f"Confidence data written to OPC server {server_id} for tag {tag}.", metadata) + success = success and local_success + + return success + @activity.defn(name='write_opc_data') async def write_opc_data(self, input_data: dict[str, Any]) -> dict[Any, Any]: """ @@ -127,47 +176,13 @@ class OPC(BaseActivity): success = True for server_id, config in opc_output_config.items(): - if self.opc_repository.get(server_id) is None: - message = f"OPC server {server_id} not found to perform write operation." - self.send_notification( - metadata=metadata, - notification_id="OPC_SERVER_NOT_FOUND", - message=message, - block="write_opc_data", - level=NotificationLevel.ERROR, - attachment_content=f"OPC servers: {list(self.opc_repository.keys())}" - ) + + if not self.validate_server(server_id, metadata): success = False + continue - if 'prediction_tags' in config: - for tag, tag_config in config['prediction_tags'].items(): - local_success = self.write_data( - server_id=server_id, - tag=tag, - data=data.head(1)['prediction'].values[0], - data_type=tag_config['data_type'], - tag_type='prediction', - metadata=metadata - ) - if local_success: - self.info( - f"Prediction data written to OPC server {server_id} for tag {tag}.", metadata) - success = success and local_success - - if 'confidence_tags' in config: - for tag, tag_config in config['confidence_tags'].items(): - local_success = self.write_data( - server_id=server_id, - tag=tag, - data=data.head(1)['prediction_confidence'].values[0], - data_type=tag_config['data_type'], - tag_type='confidence', - metadata=metadata - ) - if local_success: - self.info( - f"Confidence data written to OPC server {server_id} for tag {tag}.", metadata) - success = success and local_success + success = success and self.manage_output_tags( + server_id, config, data, metadata, success) return self.process_confidence(data, success, metadata)