openpectus.engine.opcua_hardware

Attributes

`logger`
`OPCUA_Types`

Classes

OPCUA_Hardware

Represents OPCUA hardware layer.

Functions

`_opcua_parent_path`(path)	Returns the parent path for an OPC-UA path.
`batched`(iterable, n)	Batch data into tuples of length n. The last batch may be shorter.

Module Contents

openpectus.engine.opcua_hardware.logger

openpectus.engine.opcua_hardware.OPCUA_Types

openpectus.engine.opcua_hardware._opcua_parent_path(path)

Returns the parent path for an OPC-UA path.

> path = “Objects/2:FT01/2:Totalizer/2:Total 1” > self._parent_path(path) “Objects/2:FT01/2:Totalizer”

openpectus.engine.opcua_hardware.batched(iterable, n)

Batch data into tuples of length n. The last batch may be shorter. Sourced from https://docs.python.org/3.11/library/itertools.html#itertools.batched. In Python 3.12 this will become a part of the itertools package and this function can be replaced by ‘from itertools import batched’.

Parameters:

iterable (Sequence[Any])
n (int)

class openpectus.engine.opcua_hardware.OPCUA_Hardware(host)

Bases: openpectus.engine.hardware.HardwareLayerBase

Represents OPCUA hardware layer.

Parameters:: host (str)

host: str

_client: asyncua.sync.Client

_max_nodes_per_read: int = 1000

_max_nodes_per_write: int = 1000

__str__()

Return type:: str

_browse_opcua_name_space_depth_first_until_path_is_valid(path)

Given a path to a node that does not exist, find the closest parent that does.

> path_to_nonexistant_node = “Objects/2:FT01/2:Totalizer/2:Totaal 1” > # User made a spelling mistake at the final node above. Oh dear… > # Suppose that “Objects/2:FT01/2:Totalizer” does exist though > self._browse_opcua_name_space_depth_first_until_path_is_valid( > path_to_nonexistant_node) [QualifiedName(NamespaceIndex=2, Name=’Total 1’), QualifiedName(NamespaceIndex=2, Name=’Total 2’), QualifiedName(NamespaceIndex=2, Name=’Total 3’)]

Parameters:: path (str)
Return type:: tuple[str, list[asyncua.ua.uatypes.QualifiedName]]

_query_server_operation_limits(): Ask the server about it’s operational capabilities.

validate_offline(): Perform checks that verify that the registers definition is valid. Raise on validation error. Virtual method.

validate_online()

Perform checks that verify online hardware access. Raise on validation error. Virtual method.

The default implementation checks that all Read registers can be read. Override to add additional checks.

_register_to_node(r)

Resolving a NodeId from a path requires an OPC-UA call. The NodeId’s are fixed so it makes sense to resolve them once and cache the result for future use.

Parameters:: r (openpectus.engine.hardware.Register)
Return type:: asyncua.sync.SyncNode

_registers_to_nodes(registers)

Parameters:: registers (Sequence[openpectus.engine.hardware.Register])
Return type:: list[asyncua.sync.SyncNode]

read(r)

Read single register value. Abstract method.

Parameters:: r (openpectus.engine.hardware.Register)
Return type:: Any

read_batch(registers)

Read batch of register values with a single OPC-UA call.

Parameters:: registers (Sequence[openpectus.engine.hardware.Register])
Return type:: list[Any]

_coerce_value_to_proper_type(value, r)

Parameters:

value (Any)
r (openpectus.engine.hardware.Register)

Return type:

Any

write(value, r)

Write single register value. Abstract method.

Parameters:

value (Any)
r (openpectus.engine.hardware.Register)

write_batch(values, registers)

Write batch of register values with a single OPC-UA call.

Parameters:

values (Sequence[Any])
registers (Sequence[openpectus.engine.hardware.Register])

connect(): Connect to OPC-UA server.

disconnect(): Disconnect hardware.