hvl_ccb.comm.opc¶
Communication protocol implementing an OPC UA connection. This protocol is used to interface with the “Supercube” PLC from Siemens.
- class OpcUaCommunication(config)[source]¶
Bases:
hvl_ccb.comm.base.CommunicationProtocol
Communication protocol implementing an OPC UA connection. Makes use of the package python-opcua.
- static config_cls()[source]¶
Return the default configdataclass class.
- Returns
a reference to the default configdataclass class
- init_monitored_nodes(node_id: Union[object, Iterable], ns_index: int) → None[source]¶
Initialize monitored nodes.
- Parameters
node_id – one or more strings of node IDs; node IDs are always casted via str() method here, hence do not have to be strictly string objects.
ns_index – the namespace index the nodes belong to.
- Raises
OpcUaCommunicationIOError – when protocol was not opened or can’t communicate with a OPC UA server
- property is_open: bool¶
Flag indicating if the communication port is open.
- Returns
True if the port is open, otherwise False
- open() → None[source]¶
Open the communication to the OPC UA server.
- Raises
OpcUaCommunicationIOError – when communication port cannot be opened.
- read(node_id, ns_index)[source]¶
Read a value from a node with id and namespace index.
- Parameters
node_id – the ID of the node to read the value from
ns_index – the namespace index of the node
- Returns
the value of the node object.
- Raises
OpcUaCommunicationIOError – when protocol was not opened or can’t communicate with a OPC UA server
- write(node_id, ns_index, value) → None[source]¶
Write a value to a node with name
name
.- Parameters
node_id – the id of the node to write the value to.
ns_index – the namespace index of the node.
value – the value to write.
- Raises
OpcUaCommunicationIOError – when protocol was not opened or can’t communicate with a OPC UA server
- class OpcUaCommunicationConfig(host: str, endpoint_name: str, port: int = 4840, sub_handler: hvl_ccb.comm.opc.OpcUaSubHandler = <hvl_ccb.comm.opc.OpcUaSubHandler object>, update_period: int = 500, wait_timeout_retry_sec: Union[int, float] = 1, max_timeout_retry_nr: int = 5)[source]¶
Bases:
object
Configuration dataclass for OPC UA Communciation.
- endpoint_name: str¶
Endpoint of the OPC server, this is a path like ‘OPCUA/SimulationServer’
- force_value(fieldname, value)¶
Forces a value to a dataclass field despite the class being frozen.
NOTE: you can define post_force_value method with same signature as this method to do extra processing after value has been forced on fieldname.
- Parameters
fieldname – name of the field
value – value to assign
- host: str¶
Hostname or IP-Address of the OPC UA server.
- is_configdataclass = True¶
- classmethod keys() → Sequence[str]¶
Returns a list of all configdataclass fields key-names.
- Returns
a list of strings containing all keys.
- max_timeout_retry_nr: int = 5¶
Maximal number of call re-tries on underlying OPC UA client timeout error
- classmethod optional_defaults() → Dict[str, object]¶
Returns a list of all configdataclass fields, that have a default value assigned and may be optionally specified on instantiation.
- Returns
a list of strings containing all optional keys.
- port: int = 4840¶
Port of the OPC UA server to connect to.
- classmethod required_keys() → Sequence[str]¶
Returns a list of all configdataclass fields, that have no default value assigned and need to be specified on instantiation.
- Returns
a list of strings containing all required keys.
- sub_handler: hvl_ccb.comm.opc.OpcUaSubHandler = <hvl_ccb.comm.opc.OpcUaSubHandler object>¶
object to use for handling subscriptions.
- update_period: int = 500¶
Update period for generating datachange events in OPC UA [milli seconds]
- wait_timeout_retry_sec: Union[int, float] = 1¶
Wait time between re-trying calls on underlying OPC UA client timeout error
- exception OpcUaCommunicationTimeoutError[source]¶
Bases:
hvl_ccb.comm.opc.OpcUaCommunicationIOError
OPC-UA communication timeout error.
- class OpcUaSubHandler[source]¶
Bases:
object
Base class for subscription handling of OPC events and data change events. Override methods from this class to add own handling capabilities.
To receive events from server for a subscription data_change and event methods are called directly from receiving thread. Do not do expensive, slow or network operation there. Create another thread if you need to do such a thing.