hvl_ccb.comm.opc
Communication protocol implementing an OPC UA connection. This protocol is used to interface with the “Cube” PLC from Siemens.
- class OpcUaCommunication(config)[source]
Bases:
CommunicationProtocolCommunication 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: 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. —DEPRECATED! DO NOT USE!!!—
- 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 | ~ipaddress.IPv4Address | ~ipaddress.IPv6Address, endpoint_name: str, port: int = 4840, sub_handler: ~hvl_ccb.comm.opc.OpcUaSubHandler = <hvl_ccb.comm.opc.OpcUaSubHandler object>, update_parameter: ~asyncua.ua.uaprotocol_auto.CreateSubscriptionParameters = CreateSubscriptionParameters(RequestedPublishingInterval=1000, RequestedLifetimeCount=300, RequestedMaxKeepAliveCount=22, MaxNotificationsPerPublish=10000, PublishingEnabled=True, Priority=0), wait_timeout_retry_sec: int | float = 1, max_timeout_retry_nr: int = 5)[source]
Bases:
objectConfiguration 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 | IPv4Address | IPv6Address
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: OpcUaSubHandler = <hvl_ccb.comm.opc.OpcUaSubHandler object>
object to use for handling subscriptions.
- update_parameter: CreateSubscriptionParameters = CreateSubscriptionParameters(RequestedPublishingInterval=1000, RequestedLifetimeCount=300, RequestedMaxKeepAliveCount=22, MaxNotificationsPerPublish=10000, PublishingEnabled=True, Priority=0)
Values are given as a ua.CreateSubscriptionParameters as these parameters are requested by the OPC server. Other values will lead to an automatic revision of the parameters and a warning in the opc-logger, cf. MR !173
- wait_timeout_retry_sec: int | float = 1
Wait time between re-trying calls on underlying OPC UA client timeout error
- exception OpcUaCommunicationIOError[source]
Bases:
OSError,CommunicationErrorOPC-UA communication I/O error.
- exception OpcUaCommunicationTimeoutError[source]
Bases:
OpcUaCommunicationIOErrorOPC-UA communication timeout error.
- class OpcUaSubHandler[source]
Bases:
objectBase 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.