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.

laborious/activities/opc.py

@@ -23,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'],
@@ -35,17 +35,17 @@ 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) -> 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.
@@ -56,7 +56,7 @@ class OPC(BaseActivity):
         """
 
         try:
-            return 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:
@@ -97,15 +97,15 @@ class OPC(BaseActivity):
 
         success = True
 
-        for server, config in opc_output_config.items():
+        for server_id, config in opc_output_config.items():
-            if self.opc_repository.get(server) is None:
+            if self.opc_repository.get(server_id) is None:
-                self.error(f"OPC server {server} not found", metadata)
+                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():
                     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'],
@@ -115,7 +115,7 @@ class OPC(BaseActivity):
             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'],

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

@@ -35,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
@@ -121,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,
@@ -165,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:
@@ -173,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 \
@@ -181,11 +181,11 @@ 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
@@ -210,7 +210,7 @@ class OpcRepository():
         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,
@@ -222,7 +222,7 @@ class OpcRepository():
 
         if data_type not in data_type_map:
             self.notification_handler.build_and_send_notification(
-                notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{self.name}",
+                notification_id=f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}",
                 message=f"Unsupported data type: {data_type}",
                 block="opc_repository",
                 level=NotificationLevel.ERROR
@@ -239,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,

values.yaml

@@ -153,7 +153,7 @@ env:
     value: "aignosi"
 
   - name: OPC_ID
-    value: "server-1"
+    value: "1"
   - name: OPC_URL
     value: "opc.tcp://sientia-opc-simulator-opc.sientia.svc.cluster.local:4840"