Opcua Connection

OpcuaConnection

class eta_nexus.connections.OpcuaConnection(url: str, usr: str | None = None, pwd: str | None = None, *, nodes: Nodes[OpcuaNode] | None = None, key_cert: KeyCertPair | None = None, **kwargs: Any)[source]

The OPC UA Connection class allows reading and writing from and to OPC UA servers. Additionally, it implements a subscription method, which reads continuously in a specified interval.

Parameters:

url – URL of the OPC UA Server.
usr – Username in OPC UA for login.
pwd – Password in OPC UA for login.
nodes – List of nodes to use for all operations.

logger: Logger = <Logger eta_nexus.connections.opcua_connection (WARNING)>

connection: Client

classmethod from_ids(ids: Sequence[str], url: str, usr: str | None = None, pwd: str | None = None) → OpcuaConnection[source]

Initialize the connection object from an OPC UA protocol through the node IDs.

Parameters:

ids – Identification of the Node.
url – URL for connection.
usr – Username in OPC UA for login.
pwd – Password in OPC UA for login.

Returns:

OpcuaConnection object.

read(nodes: OpcuaNode | Nodes[OpcuaNode] | None = None) → pd.DataFrame[source]

Read some manually selected values from OPC UA capable controller.

Parameters:: nodes – Single node or list/set of nodes to read from.
Returns:: pandas.DataFrame containing current values of the OPC UA-variables.
Raises:: ConnectionError – When an error occurs during reading.

write(values: Mapping[OpcuaNode, Primitive]) → None[source]

Writes some manually selected values on OPC UA capable controller.

Parameters:: values – Dictionary of nodes and data to write {node: value}.
Raises:: ConnectionError – When an error occurs during reading.

subscribe(handler: SubscriptionHandler, nodes: OpcuaNode | Nodes[OpcuaNode] | None = None, interval: TimeStep = 1) → None[source]

Subscribe to nodes and call handler when new data is available. Basic architecture of the subscription is the client- server communication via subscription notify. This function works asynchronously. Subscriptions must always be closed using the close_sub function (use try, finally!).

Parameters:

nodes – Single node or list/set of nodes to subscribe to.
handler – SubscriptionHandler object with a push method that accepts node, value pairs.
interval – Interval for receiving new data. It is interpreted as seconds when given as an integer.

close_sub() → None[source]: Close an open subscription.

url_parsed: ParseResult: URL of the server to connect to

usr: str | None: Username for login to server

pwd: str | None: Password for login to server

exc: BaseException | None

OpcuaNode

class eta_nexus.nodes.OpcuaNode(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)[source]

Node for the OPC UA protocol.

opc_id: str | None: Node ID of the OPC UA Node.

opc_path_str: str | None: Path to the OPC UA node.

opc_ns: int | None: Namespace of the OPC UA Node.

opc_id_type: str: Type of the OPC UA Node ID Specification.

opc_name: str: Name of the OPC UA Node.

evolve(**kwargs: Any) → Self[source]

Returns a new node instance by copying the current node and changing only specified keyword arguments.

This allows for seamless node instantiation with only a few changes.

Adjusted attributes handling according to OpcuaNode instantiation logic as in ‘__attrs_post_init__’.

Parameters:: kwargs – Keyword arguments to change.
Returns:: New instance of the node.

connection_identifier() → str: Unique identifier for the connection that is associated with the node (i.e. would be created by Connection.from_node())

name: str: Name for the node.

url: str: URL of the connection.

url_parsed: ParseResult: Parse result object of the URL (in case more post-processing is required).

usr: str | None: Username for login to the connection (default: None).

pwd: str | None: Password for login to the connection (default: None).

interval: str | None: Interval

dtype: Callable | None: Data type of the node (for value conversion). Note that strings will be interpreted as utf-8 encoded. If you do not want this behaviour, use ‘bytes’.