Merge pull request #1706 from roboflow/opc-ua-improvements

OPC-UA improvements for ignition

Author: hansent

inference/enterprise/workflows/enterprise_blocks/sinks/opc_writer/v1.py

@@ -43,6 +43,20 @@
 
 **Note:** The data type you send must match the expected type of the target OPC UA variable.
 
+### Node Lookup Mode
+The block supports two methods for locating OPC UA nodes via the `node_lookup_mode` parameter:
+
+- **`hierarchical` (default)**: Uses standard OPC UA hierarchical path navigation. The block navigates
+  through the address space using `get_child()`. Each component in the `object_name` path is
+  automatically prefixed with the namespace index.
+  - **Example**: `object_name="Roboflow/Crane_11"` → path `0:Objects/2:Roboflow/2:Crane_11/2:Variable`
+  - **Best for**: Traditional OPC UA servers with hierarchical address spaces
+
+- **`direct`**: Uses direct NodeId string access. The block constructs a NodeId as
+  `ns={namespace};s={object_name}/{variable_name}` and accesses it directly via `get_node()`.
+  - **Example**: `object_name="[Sample_Tags]/Ramp"` → NodeId `ns=2;s=[Sample_Tags]/Ramp/South_Person_Count`
+  - **Best for**: Ignition SCADA systems and other servers using string-based NodeId identifiers
+
 ### Cooldown
 To prevent excessive traffic to the OPC UA server, the block includes a `cooldown_seconds` parameter, 
 which defaults to **5 seconds**. During the cooldown period:
@@ -190,6 +204,15 @@ class BlockManifest(WorkflowBlockManifest):
         },
         examples=[10, "$inputs.cooldown_seconds"],
     )
+    node_lookup_mode: Union[
+        Selector(kind=[STRING_KIND]),
+        Literal["hierarchical", "direct"],
+    ] = Field(
+        default="hierarchical",
+        description="Method to locate the OPC UA node: 'hierarchical' uses path navigation, "
+        "'direct' uses NodeId strings (for Ignition-style string-based tags).",
+        examples=["hierarchical", "direct"],
+    )
 
     @classmethod
     def describe_outputs(cls) -> List[OutputDefinition]:
@@ -238,8 +261,10 @@ def run(
         fire_and_forget: bool = True,
         disable_sink: bool = False,
         cooldown_seconds: int = 5,
+        node_lookup_mode: Literal["hierarchical", "direct"] = "hierarchical",
     ) -> BlockResult:
         if disable_sink:
+            logging.debug("OPC Writer disabled by disable_sink parameter")
             return {
                 "disabled": True,
                 "throttling_status": False,
@@ -261,6 +286,7 @@ def run(
             }
 
         value_str = str(value)
+        logging.debug(f"OPC Writer converting value '{value_str}' to type {value_type}")
         try:
             if value_type in [BOOLEAN_KIND, "Boolean"]:
                 decoded_value = value_str.strip().lower() in ("true", "1")
@@ -272,7 +298,9 @@ def run(
                 decoded_value = value_str
             else:
                 raise ValueError(f"Unsupported value type: {value_type}")
+            logging.debug(f"OPC Writer successfully converted value to {decoded_value}")
         except ValueError as exc:
+            logging.error(f"OPC Writer failed to convert value: {exc}")
             return {
                 "disabled": False,
                 "error_status": True,
@@ -290,9 +318,11 @@ def run(
             variable_name=variable_name,
             value=decoded_value,
             timeout=timeout,
+            node_lookup_mode=node_lookup_mode,
         )
         self._last_notification_fired = datetime.now()
         if fire_and_forget and self._background_tasks:
+            logging.debug("OPC Writer submitting write task to background tasks")
             self._background_tasks.add_task(opc_writer_handler)
             return {
                 "disabled": False,
@@ -301,14 +331,19 @@ def run(
                 "message": "Writing to the OPC UA server in the background task",
             }
         if fire_and_forget and self._thread_pool_executor:
+            logging.debug("OPC Writer submitting write task to thread pool executor")
             self._thread_pool_executor.submit(opc_writer_handler)
             return {
                 "disabled": False,
                 "error_status": False,
                 "throttling_status": False,
                 "message": "Writing to the OPC UA server in the background task",
             }
+        logging.debug("OPC Writer executing synchronous write")
         error_status, message = opc_writer_handler()
+        logging.debug(
+            f"OPC Writer write completed: error_status={error_status}, message={message}"
+        )
         return {
             "disabled": False,
             "error_status": error_status,
@@ -317,6 +352,30 @@ def run(
         }
 
 
+def get_available_namespaces(client: Client) -> List[str]:
+    """
+    Get list of available namespaces from OPC UA server.
+    Returns empty list if unable to fetch namespaces.
+    """
+    try:
+        get_namespace_array = sync_async_client_method(AsyncClient.get_namespace_array)(
+            client
+        )
+        return get_namespace_array()
+    except Exception as exc:
+        logging.info(f"Failed to get namespace array (non-fatal): {exc}")
+        return ["<unable to fetch namespaces>"]
+
+
+def safe_disconnect(client: Client) -> None:
+    """Safely disconnect from OPC UA server, swallowing any errors"""
+    try:
+        logging.debug("OPC Writer disconnecting from server")
+        client.disconnect()
+    except Exception as exc:
+        logging.debug(f"OPC Writer disconnect error (non-fatal): {exc}")
+
+
 def opc_connect_and_write_value(
     url: str,
     namespace: str,
@@ -326,7 +385,11 @@ def opc_connect_and_write_value(
     variable_name: str,
     value: Union[bool, float, int, str],
     timeout: int,
+    node_lookup_mode: Literal["hierarchical", "direct"] = "hierarchical",
 ) -> Tuple[bool, str]:
+    logging.debug(
+        f"OPC Writer attempting to connect and write value={value} to {url}/{object_name}/{variable_name}"
+    )
     try:
         _opc_connect_and_write_value(
             url=url,
@@ -337,9 +400,14 @@ def opc_connect_and_write_value(
             variable_name=variable_name,
             value=value,
             timeout=timeout,
+            node_lookup_mode=node_lookup_mode,
+        )
+        logging.debug(
+            f"OPC Writer successfully wrote value to {url}/{object_name}/{variable_name}"
         )
         return False, "Value set successfully"
     except Exception as exc:
+        logging.error(f"OPC Writer failed to write value: {exc}")
         return (
             True,
             f"Failed to write {value} to {object_name}:{variable_name} in {url}. Internal error details: {exc}.",
@@ -355,21 +423,28 @@ def _opc_connect_and_write_value(
     variable_name: str,
     value: Union[bool, float, int, str],
     timeout: int,
+    node_lookup_mode: Literal["hierarchical", "direct"] = "hierarchical",
 ):
+    logging.debug(f"OPC Writer creating client for {url} with timeout={timeout}")
     client = Client(url=url, sync_wrapper_timeout=timeout)
     if user_name and password:
         client.set_user(user_name)
         client.set_password(password)
     try:
+        logging.debug(f"OPC Writer connecting to {url}")
         client.connect()
+        logging.debug("OPC Writer successfully connected to server")
     except BadUserAccessDenied as exc:
-        client.disconnect()
+        logging.error(f"OPC Writer authentication failed: {exc}")
+        safe_disconnect(client)
         raise Exception(f"AUTH ERROR: {exc}")
     except OSError as exc:
-        client.disconnect()
+        logging.error(f"OPC Writer network error during connection: {exc}")
+        safe_disconnect(client)
         raise Exception(f"NETWORK ERROR: {exc}")
     except Exception as exc:
-        client.disconnect()
+        logging.error(f"OPC Writer unhandled connection error: {type(exc)} {exc}")
+        safe_disconnect(client)
         raise Exception(f"UNHANDLED ERROR: {type(exc)} {exc}")
     get_namespace_index = sync_async_client_method(AsyncClient.get_namespace_index)(
         client
@@ -378,33 +453,79 @@ def _opc_connect_and_write_value(
     try:
         if namespace.isdigit():
             nsidx = int(namespace)
+            logging.debug(f"OPC Writer using numeric namespace index: {nsidx}")
         else:
             nsidx = get_namespace_index(namespace)
     except ValueError as exc:
-        client.disconnect()
-        raise Exception(f"WRONG NAMESPACE ERROR: {exc}")
-    except Exception as exc:
-        client.disconnect()
-        raise Exception(f"UNHANDLED ERROR: {type(exc)} {exc}")
-
-    try:
-        var = client.nodes.root.get_child(
-            f"0:Objects/{nsidx}:{object_name}/{nsidx}:{variable_name}"
+        namespaces = get_available_namespaces(client)
+        logging.error(f"OPC Writer invalid namespace: {exc}")
+        logging.error(f"Available namespaces: {namespaces}")
+        safe_disconnect(client)
+        raise Exception(
+            f"WRONG NAMESPACE ERROR: {exc}. Available namespaces: {namespaces}"
         )
-    except BadNoMatch as exc:
-        client.disconnect()
-        raise Exception(f"WRONG OBJECT OR PROPERTY ERROR: {exc}")
     except Exception as exc:
-        client.disconnect()
-        raise Exception(f"UNHANDLED ERROR: {type(exc)} {exc}")
+        namespaces = get_available_namespaces(client)
+        logging.error(f"OPC Writer unhandled namespace error: {type(exc)} {exc}")
+        logging.error(f"Available namespaces: {namespaces}")
+        safe_disconnect(client)
+        raise Exception(
+            f"UNHANDLED ERROR: {type(exc)} {exc}. Available namespaces: {namespaces}"
+        )
+
+    if node_lookup_mode == "direct":
+        # Direct NodeId access for Ignition-style string identifiers
+        try:
+            node_id = f"ns={nsidx};s={object_name}/{variable_name}"
+            logging.debug(f"OPC Writer using direct NodeId access: {node_id}")
+            var = client.get_node(node_id)
+            # Verify the node exists by reading its attributes
+            var.read_browse_name()
+            logging.debug(
+                f"OPC Writer successfully found variable node using direct NodeId"
+            )
+        except Exception as exc:
+            logging.error(f"OPC Writer direct NodeId access failed: {exc}")
+            safe_disconnect(client)
+            raise Exception(
+                f"WRONG OBJECT OR PROPERTY ERROR: Could not find node with direct NodeId '{node_id}'. Error: {exc}"
+            )
+    else:
+        # Hierarchical path navigation (standard OPC UA)
+        try:
+            # Split object_name on "/" and prepend namespace index to each component
+            object_components = object_name.split("/")
+            object_path = "/".join([f"{nsidx}:{comp}" for comp in object_components])
+            node_path = f"0:Objects/{object_path}/{nsidx}:{variable_name}"
+            logging.debug(f"OPC Writer using hierarchical path: {node_path}")
+            var = client.nodes.root.get_child(node_path)
+            logging.debug(
+                f"OPC Writer successfully found variable node using hierarchical path"
+            )
+        except BadNoMatch as exc:
+            logging.error(f"OPC Writer hierarchical path not found: {exc}")
+            safe_disconnect(client)
+            raise Exception(
+                f"WRONG OBJECT OR PROPERTY ERROR: Could not find node at hierarchical path '{node_path}'. Error: {exc}"
+            )
+        except Exception as exc:
+            logging.error(f"OPC Writer unhandled node lookup error: {type(exc)} {exc}")
+            safe_disconnect(client)
+            raise Exception(f"UNHANDLED ERROR: {type(exc)} {exc}")
 
     try:
+        logging.debug(f"OPC Writer writing value '{value}' to variable")
         var.write_value(value)
+        logging.info(
+            f"OPC Writer successfully wrote  '{value}'  to variable at {object_name}/{variable_name}"
+        )
     except BadTypeMismatch as exc:
-        client.disconnect()
+        logging.error(f"OPC Writer type mismatch error: {exc}")
+        safe_disconnect(client)
         raise Exception(f"WRONG TYPE ERROR: {exc}")
     except Exception as exc:
-        client.disconnect()
+        logging.error(f"OPC Writer unhandled write error: {type(exc)} {exc}")
+        safe_disconnect(client)
         raise Exception(f"UNHANDLED ERROR: {type(exc)} {exc}")
 
-    client.disconnect()
+    safe_disconnect(client)