API Reference#

OPC Unified Architecture (OPC-UA)

OPC-UA is a platform-independent, service-oriented architecture for industrial automation and process control, commonly used for supervisory/SCADA-level communication with PLCs and other field devices.

Added in version 0.3.0.

High-level OPC-UA connection wrapper built on top of asyncua.

class icspacket.proto.opcua.connection.BrowseNode(node_id: NodeId, browse_name: str, display_name: str, node_class: NodeClass, type_definition: NodeId | None = None, value: Any = None, children: list[BrowseNode] = <factory>)[source]#

A node discovered while browsing an OPC-UA server’s address space, as returned by OPCUA_Connection.browse_details().

Variables:

  • node_id (ua.NodeId) – The node’s NodeId.

  • browse_name (str) – The node’s (non-localized) browse name.

  • display_name (str) – The node’s human-readable display name.

  • node_class (ua.NodeClass) – The node’s class, e.g. Object, Variable or Method.

  • type_definition (ua.NodeId | None) – The node’s type definition NodeId (e.g. the VariableType/ObjectType it was instantiated from), or None if it has none.

  • value (Any) – The node’s current value, if requested via values=True and the node is a Variable. None otherwise, or if the value could not be read.

  • children (list[BrowseNode]) – Child nodes, populated only when browsed with recursive=True.

exception icspacket.proto.opcua.connection.OPCUAProtocolError[source]#

Raised when an OPC-UA service request fails at the application level.

This wraps errors reported by asyncua (e.g. BadNodeIdUnknown, BadUserAccessDenied) so callers can catch a single icspacket exception type regardless of the underlying library.

class icspacket.proto.opcua.connection.OPCUA_Connection(timeout: float = 4.0, username: str | None = None, password: str | None = None, security_policy: str | None = None, security_mode: str = 'SignAndEncrypt', certificate: str | None = None, private_key: str | None = None, private_key_password: str | None = None, server_certificate: str | None = None, user_certificate: str | None = None, user_private_key: str | None = None, user_private_key_password: str | None = None)[source]#

Synchronous OPC-UA client connection built on asyncua.

This class provides a thin wrapper around asyncua.sync.Client, exposing node browsing/read/write as simple Python methods.

Example:

conn = OPCUA_Connection()
conn.connect(("192.168.1.50", 4840))
value = conn.read_value("ns=2;i=2")
conn.write_value("ns=2;i=3", "FAULT")
conn.close()

Note

NodeId syntax is entirely server-defined: some servers assign numeric identifiers (ns=2;i=2), others string identifiers (ns=2;s=Device1.Temperature). Use browse() (or discover_endpoints() for a server’s security settings) to find the actual identifiers exposed by a given server instead of assuming either form.

Parameters:

  • timeout (float) – Timeout in seconds for OPC-UA service calls.

  • username (str | None) – Optional username for username/password authentication.

  • password (str | None) – Optional password for username/password authentication.

  • security_policy (str | None) – Secure channel security policy name, e.g. "Basic256Sha256", "Aes128Sha256RsaOaep" or "Aes256Sha256RsaPss". None (the default) disables secure channel security entirely (SecurityPolicy#None).

  • security_mode (str) – Message security mode to request, "Sign" or "SignAndEncrypt". Only relevant when security_policy is set.

  • certificate (str | None) – Path to the client’s application instance certificate (PEM/DER), required when security_policy is set.

  • private_key (str | None) – Path to the private key matching certificate, required when security_policy is set.

  • private_key_password (str | None) – Password protecting private_key, if it is encrypted.

  • server_certificate (str | None) – Expected server certificate (PEM/DER). If omitted, it is auto-discovered via an unauthenticated GetEndpoints call, matching the requested policy/mode.

  • user_certificate (str | None) – Certificate used for X509 user identity token authentication (distinct from channel security). Defaults to certificate when unset and user_private_key is given.

  • user_private_key (str | None) – Private key matching user_certificate. Defaults to private_key when unset.

  • user_private_key_password (str | None) – Password protecting user_private_key, if it is encrypted.

property client: Client#

The underlying asyncua.sync.Client instance.

connect(address: tuple[str, int] | str) → None[source]#

Connect and create a session with an OPC-UA server.

Parameters:

address (tuple[str, int] | str) – Either a full opc.tcp://host:port[/path] endpoint URL, or a plain (host, port) tuple.

Raises:

ConnectionError – If the connection or session creation fails, or if the security configuration is incomplete/invalid.

close() → None[source]#

Disconnect the OPC-UA session and secure channel.

Idempotent: calling this more than once, or before a connection was ever established, is a no-op.

Raises:

ConnectionClosedError – If the underlying client reports the connection as already/unexpectedly closed.

get_node(node_id: str | NodeId) → SyncNode[source]#

Resolve a node by its NodeId string (e.g. ns=2;s=MyVariable).

Parameters:

node_id (str | ua.NodeId) – NodeId string or NodeId instance.

Returns:

The resolved node handle.

Return type:

SyncNode

read_value(node_id: str | NodeId) → Any[source]#

Read the current value of a node’s Value attribute.

Parameters:

node_id (str | ua.NodeId) – Target node’s NodeId.

Returns:

The decoded Python value.

Raises:

OPCUAProtocolError – If the read service call fails.

write_value(node_id: str | NodeId, value: Any) → None[source]#

Write a value to a node’s Value attribute.

Parameters:

  • node_id (str | ua.NodeId) – Target node’s NodeId.

  • value – New value to write (a plain Python value or a Variant).

Raises:

OPCUAProtocolError – If the write service call fails.

browse(node_id: str | NodeId | None = None) → list[SyncNode][source]#

List the children of a node (or the Objects folder by default).

Parameters:

node_id (str | ua.NodeId | None) – NodeId to browse, or None for the root Objects folder.

Returns:

Child node handles.

Return type:

list[SyncNode]

browse_details(node_id: str | NodeId | None = None, *, recursive: bool = False, max_depth: int = 10, values: bool = False) → list[BrowseNode][source]#

List a node’s children with rich per-node metadata attached.

Unlike browse(), which only returns bare node handles (one extra round-trip per child is then needed just to resolve a name), this issues a single Browse service call per level to retrieve every child’s NodeId, BrowseName, DisplayName, NodeClass and TypeDefinition together.

Example:

conn = OPCUA_Connection()
conn.connect(("192.168.1.50", 4840))
for child in conn.browse_details(recursive=True, values=True):
    print(child)
Parameters:

  • node_id (str | ua.NodeId | None) – NodeId to browse, or None for the root Objects folder.

  • recursive (bool) – If True, also recursively browse every discovered child, building a full tree instead of a single level. A node reachable via more than one reference path (or an outright reference cycle) is only ever expanded once; repeat occurrences are still listed, just without children.

  • max_depth (int) – Maximum number of levels to descend when recursive is set (ignored otherwise). A server’s address space can be very large (e.g. the standard Server diagnostics subtree), so this bounds a scan’s depth by default.

  • values (bool) – If True, also read and attach the current value of every Variable node encountered (one extra round-trip per variable). A failed read (e.g. no read access) is ignored, leaving BrowseNode.value as None rather than aborting the browse.

Returns:

The requested node’s children, in server-defined order, with .children populated recursively when recursive is set.

Return type:

list[BrowseNode]

Raises:

OPCUAProtocolError – If browsing fails.

create_subscription(node_ids: Sequence[str | NodeId], interval: float = 500.0) → OPCUA_Subscription[source]#

Create a data-change subscription for one or more nodes.

The returned OPCUA_Subscription operates in “iterator mode”: notifications are buffered internally by asyncua and retrieved on demand via OPCUA_Subscription.next_event() or by iterating over it directly, rather than via a registered callback handler.

Parameters:

  • node_ids (Sequence[str | ua.NodeId]) – One or more NodeIds to monitor for value changes.

  • interval (float) – Requested publishing interval in milliseconds.

Returns:

A handle yielding data-change events as they arrive.

Return type:

OPCUA_Subscription

Raises:

OPCUAProtocolError – If the subscription cannot be created.

class icspacket.proto.opcua.connection.OPCUA_Subscription(subscription: Subscription)[source]#

Handle for an active OPC-UA data-change subscription.

Returned by OPCUA_Connection.create_subscription(). Wraps asyncua.sync.Subscription in iterator mode: no callback handler is registered, notifications are buffered internally and retrieved on demand.

Example:

sub = conn.create_subscription(["ns=2;i=2", "ns=2;i=3"])
for event in sub:
    print(event.node, event.value)
sub.close()

Also usable as a context manager, which calls close() on exit:

with conn.create_subscription(["ns=2;i=2"]) as sub:
    event = sub.next_event(timeout=5.0)
next_event(timeout: float | None = None) → DataChangeEvent | OpcEvent | StatusChangeEvent | None[source]#

Wait for and return the next data-change/event notification.

Parameters:

timeout (float | None) – Maximum time in seconds to wait, or None to block indefinitely.

Returns:

The next event, or None if timeout elapsed.

Return type:

SubEvent | None

close() → None[source]#

Delete the subscription on the server and release resources.

Idempotent: safe to call more than once.

icspacket.proto.opcua.connection.discover_endpoints(address: tuple[str, int] | str, timeout: float = 4.0) → list[EndpointDescription][source]#

Discover the endpoints advertised by an OPC-UA server.

Performs a lightweight, sessionless GetEndpoints service call: no authentication, secure channel security, or application session is established. Useful for probing which security policies, modes, and user token types a server supports before configuring a full OPCUA_Connection.

Parameters:

  • address (tuple[str, int] | str) – Either a full opc.tcp://host:port[/path] endpoint URL, or a plain (host, port) tuple.

  • timeout (float) – Timeout in seconds for the discovery call.

Returns:

The list of endpoints advertised by the server.

Return type:

list[ua.EndpointDescription]

Raises:

ConnectionError – If the discovery call fails.