asyncua.common package#

Submodules#

asyncua.common.callback module#

server side implementation of callback event

class asyncua.common.callback.Callback#

Bases: object

getName()#

setName(name)#

class asyncua.common.callback.CallbackService#

Bases: object

addListener(eventName, listener, priority=0)#

addSubscriber(subscriber)#

async call_listener(event, listener)#

async dispatch(eventName, event=None)#

removeListener(eventName, listener=None)#

class asyncua.common.callback.CallbackSubscriberInterface#

Bases: object

getSubscribedEvents()#

class asyncua.common.callback.CallbackType(value)#

Bases: Enum

The possible types of a Callback type.

Variables:

Null –
MonitoredItem –

ItemSubscriptionCreated = 1#

ItemSubscriptionDeleted = 3#

ItemSubscriptionModified = 2#

Null = 0#

PostRead = 7#

PostWrite = 5#

PreRead = 6#

PreWrite = 4#

class asyncua.common.callback.ServerItemCallback(request_params, response_params, user=None, is_external=False)#: Bases: Callback

asyncua.common.connection module#

class asyncua.common.connection.MessageChunk(security_policy, body=b'', msg_type=b'MSG', chunk_type=b'F')#

Bases: object

Message Chunk, as described in OPC UA specs Part 6, 6.7.2.

encrypted_size(plain_size)#

static from_binary(security_policy, data)#

static from_header_and_body(security_policy, header, buf, use_prev_key=False)#

static max_body_size(crypto, max_chunk_size)#

static message_to_chunks(security_policy, body, max_chunk_size, message_type=b'MSG', channel_id=1, request_id=1, token_id=1)#: Pack message body (as binary string) into one or more chunks. Size of each chunk will not exceed max_chunk_size. Returns a list of MessageChunks. SequenceNumber is not initialized here, it must be set by Secure Channel driver.

to_binary()#

class asyncua.common.connection.SecureConnection(security_policy, limits: TransportLimits)#

Bases: object

Common logic for client and server

close()#

is_open()#

message_to_binary(message, message_type=b'MSG', request_id=0)#: Convert OPC UA secure message to binary. The only supported types are SecureOpen, SecureMessage, SecureClose. If message_type is SecureMessage, the AlgorithmHeader should be passed as arg.

open(params, server)#: Called on server side to open secure channel.

receive_from_header_and_body(header, body)#: Convert MessageHeader and binary body to OPC UA TCP message (see OPC UA specs Part 6, 7.1: Hello, Acknowledge or ErrorMessage), or a Message object, or None (if intermediate chunk is received)

revolve_tokens()#: Revolve security tokens of the security channel. Start using the next security token negotiated during the renewal of the channel and remember the previous token until the other communication party

select_policy(uri, peer_certificate, mode=None)#

set_channel(params, request_type, client_nonce)#: Called on client side when getting secure channel data from server.

set_policy_factories(policies)#: Set a list of available security policies. Use this in servers with multiple endpoints with different security.

class asyncua.common.connection.TransportLimits(max_recv_buffer: int = 65535, max_send_buffer: int = 65535, max_chunk_count: int = 1601, max_message_size: int = 104857600)#

Bases: object

Limits of the tcp transport layer to prevent excessive resource usage

create_acknowledge_and_set_limits(msg: Hello) → Acknowledge#

create_hello_limits(msg: Hello) → Hello#

is_chunk_count_within_limit(sz: int) → bool#

is_msg_size_within_limit(sz: int) → bool#

max_chunk_count: int = 1601#

max_message_size: int = 104857600#

max_recv_buffer: int = 65535#

max_send_buffer: int = 65535#

update_client_limits(msg: Acknowledge) → None#

asyncua.common.copy_node_util module#

async asyncua.common.copy_node_util.copy_node(parent: Node, node: Node, nodeid: NodeId | None = None, recursive: bool = True) → List[Node]#: Copy a node or node tree as child of parent node

asyncua.common.event_objects module#

Autogenerated code from xml”

Model Uri:http://opcfoundation.org/UA/” Version:1.05.01” Publication date:2022-02-24T00:00:00Z”

File creation Date:2023-07-25 09:08:45.084676”

class asyncua.common.event_objects.AcknowledgeableCondition(sourcenode=None, message=None, severity=1)#

Bases: Condition

AcknowledgeableCondition:

class asyncua.common.event_objects.AlarmCondition(sourcenode=None, message=None, severity=1)#

Bases: AcknowledgeableCondition

AlarmCondition:

class asyncua.common.event_objects.AuditActivateSessionEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditSessionEvent

AuditActivateSessionEvent:

class asyncua.common.event_objects.AuditAddNodesEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditNodeManagementEvent

AuditAddNodesEvent:

class asyncua.common.event_objects.AuditAddReferencesEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditNodeManagementEvent

AuditAddReferencesEvent:

class asyncua.common.event_objects.AuditCancelEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditSessionEvent

AuditCancelEvent:

class asyncua.common.event_objects.AuditCertificateDataMismatchEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditCertificateEvent

AuditCertificateDataMismatchEvent:

class asyncua.common.event_objects.AuditCertificateEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditSecurityEvent

AuditCertificateEvent:

class asyncua.common.event_objects.AuditCertificateExpiredEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditCertificateEvent

AuditCertificateExpiredEvent:

class asyncua.common.event_objects.AuditCertificateInvalidEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditCertificateEvent

AuditCertificateInvalidEvent:

class asyncua.common.event_objects.AuditCertificateMismatchEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditCertificateEvent

AuditCertificateMismatchEvent:

class asyncua.common.event_objects.AuditCertificateRevokedEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditCertificateEvent

AuditCertificateRevokedEvent:

class asyncua.common.event_objects.AuditCertificateUntrustedEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditCertificateEvent

AuditCertificateUntrustedEvent:

class asyncua.common.event_objects.AuditChannelEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditSecurityEvent

AuditChannelEvent:

class asyncua.common.event_objects.AuditClientEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditEvent

AuditClientEvent:

class asyncua.common.event_objects.AuditClientUpdateMethodResultEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditClientEvent

AuditClientUpdateMethodResultEvent:

class asyncua.common.event_objects.AuditConditionAcknowledgeEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditConditionEvent

AuditConditionAcknowledgeEvent:

class asyncua.common.event_objects.AuditConditionCommentEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditConditionEvent

AuditConditionCommentEvent:

class asyncua.common.event_objects.AuditConditionConfirmEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditConditionEvent

AuditConditionConfirmEvent:

class asyncua.common.event_objects.AuditConditionEnableEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditConditionEvent

AuditConditionEnableEvent:

class asyncua.common.event_objects.AuditConditionEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditUpdateMethodEvent

AuditConditionEvent:

class asyncua.common.event_objects.AuditConditionOutOfServiceEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditConditionEvent

AuditConditionOutOfServiceEvent:

class asyncua.common.event_objects.AuditConditionResetEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditConditionEvent

AuditConditionResetEvent:

class asyncua.common.event_objects.AuditConditionRespondEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditConditionEvent

AuditConditionRespondEvent:

class asyncua.common.event_objects.AuditConditionShelvingEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditConditionEvent

AuditConditionShelvingEvent:

class asyncua.common.event_objects.AuditConditionSilenceEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditConditionEvent

AuditConditionSilenceEvent:

class asyncua.common.event_objects.AuditConditionSuppressionEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditConditionEvent

AuditConditionSuppressionEvent:

class asyncua.common.event_objects.AuditCreateSessionEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditSessionEvent

AuditCreateSessionEvent:

class asyncua.common.event_objects.AuditDeleteNodesEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditNodeManagementEvent

AuditDeleteNodesEvent:

class asyncua.common.event_objects.AuditDeleteReferencesEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditNodeManagementEvent

AuditDeleteReferencesEvent:

class asyncua.common.event_objects.AuditEvent(sourcenode=None, message=None, severity=1)#

Bases: BaseEvent

AuditEvent:

class asyncua.common.event_objects.AuditHistoryAnnotationUpdateEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditHistoryUpdateEvent

AuditHistoryAnnotationUpdateEvent:

class asyncua.common.event_objects.AuditHistoryAtTimeDeleteEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditHistoryDeleteEvent

AuditHistoryAtTimeDeleteEvent:

class asyncua.common.event_objects.AuditHistoryDeleteEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditHistoryUpdateEvent

AuditHistoryDeleteEvent:

class asyncua.common.event_objects.AuditHistoryEventDeleteEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditHistoryDeleteEvent

AuditHistoryEventDeleteEvent:

class asyncua.common.event_objects.AuditHistoryEventUpdateEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditHistoryUpdateEvent

AuditHistoryEventUpdateEvent:

class asyncua.common.event_objects.AuditHistoryRawModifyDeleteEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditHistoryDeleteEvent

AuditHistoryRawModifyDeleteEvent:

class asyncua.common.event_objects.AuditHistoryUpdateEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditUpdateEvent

AuditHistoryUpdateEvent:

class asyncua.common.event_objects.AuditHistoryValueUpdateEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditHistoryUpdateEvent

AuditHistoryValueUpdateEvent:

class asyncua.common.event_objects.AuditNodeManagementEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditEvent

AuditNodeManagementEvent:

class asyncua.common.event_objects.AuditOpenSecureChannelEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditChannelEvent

AuditOpenSecureChannelEvent:

class asyncua.common.event_objects.AuditProgramTransitionEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditUpdateStateEvent

AuditProgramTransitionEvent:

class asyncua.common.event_objects.AuditSecurityEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditEvent

AuditSecurityEvent:

class asyncua.common.event_objects.AuditSessionEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditSecurityEvent

AuditSessionEvent:

class asyncua.common.event_objects.AuditUpdateEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditEvent

AuditUpdateEvent:

class asyncua.common.event_objects.AuditUpdateMethodEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditEvent

AuditUpdateMethodEvent:

class asyncua.common.event_objects.AuditUpdateStateEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditUpdateMethodEvent

AuditUpdateStateEvent:

class asyncua.common.event_objects.AuditUrlMismatchEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditCreateSessionEvent

AuditUrlMismatchEvent:

class asyncua.common.event_objects.AuditWriteUpdateEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditUpdateEvent

AuditWriteUpdateEvent:

class asyncua.common.event_objects.BaseEvent(sourcenode=None, message=None, severity=1)#

Bases: Event

BaseEvent:

class asyncua.common.event_objects.BaseModelChangeEvent(sourcenode=None, message=None, severity=1)#

Bases: BaseEvent

BaseModelChangeEvent:

class asyncua.common.event_objects.CertificateExpirationAlarm(sourcenode=None, message=None, severity=1)#

Bases: SystemOffNormalAlarm

CertificateExpirationAlarm:

class asyncua.common.event_objects.CertificateUpdatedAuditEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditUpdateMethodEvent

CertificateUpdatedAuditEvent:

class asyncua.common.event_objects.Condition(sourcenode=None, message=None, severity=1)#

Bases: BaseEvent

Condition:

class asyncua.common.event_objects.DeviceFailureEvent(sourcenode=None, message=None, severity=1)#

Bases: SystemEvent

DeviceFailureEvent:

class asyncua.common.event_objects.DialogCondition(sourcenode=None, message=None, severity=1)#

Bases: Condition

DialogCondition:

class asyncua.common.event_objects.DiscrepancyAlarm(sourcenode=None, message=None, severity=1)#

Bases: AlarmCondition

DiscrepancyAlarm:

class asyncua.common.event_objects.DiscreteAlarm(sourcenode=None, message=None, severity=1)#

Bases: AlarmCondition

DiscreteAlarm:

class asyncua.common.event_objects.EventQueueOverflowEvent(sourcenode=None, message=None, severity=1)#

Bases: BaseEvent

EventQueueOverflowEvent:

class asyncua.common.event_objects.ExclusiveDeviationAlarm(sourcenode=None, message=None, severity=1)#

Bases: ExclusiveLimitAlarm

ExclusiveDeviationAlarm:

class asyncua.common.event_objects.ExclusiveLevelAlarm(sourcenode=None, message=None, severity=1)#

Bases: ExclusiveLimitAlarm

ExclusiveLevelAlarm:

class asyncua.common.event_objects.ExclusiveLimitAlarm(sourcenode=None, message=None, severity=1)#

Bases: LimitAlarm

ExclusiveLimitAlarm:

class asyncua.common.event_objects.ExclusiveRateOfChangeAlarm(sourcenode=None, message=None, severity=1)#

Bases: ExclusiveLimitAlarm

ExclusiveRateOfChangeAlarm:

class asyncua.common.event_objects.GeneralModelChangeEvent(sourcenode=None, message=None, severity=1)#

Bases: BaseModelChangeEvent

GeneralModelChangeEvent:

class asyncua.common.event_objects.InstrumentDiagnosticAlarm(sourcenode=None, message=None, severity=1)#

Bases: OffNormalAlarm

InstrumentDiagnosticAlarm:

class asyncua.common.event_objects.KeyCredentialAuditEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditUpdateMethodEvent

KeyCredentialAuditEvent:

class asyncua.common.event_objects.KeyCredentialDeletedAuditEvent(sourcenode=None, message=None, severity=1)#

Bases: KeyCredentialAuditEvent

KeyCredentialDeletedAuditEvent:

class asyncua.common.event_objects.KeyCredentialUpdatedAuditEvent(sourcenode=None, message=None, severity=1)#

Bases: KeyCredentialAuditEvent

KeyCredentialUpdatedAuditEvent:

class asyncua.common.event_objects.LimitAlarm(sourcenode=None, message=None, severity=1)#

Bases: AlarmCondition

LimitAlarm:

class asyncua.common.event_objects.NonExclusiveDeviationAlarm(sourcenode=None, message=None, severity=1)#

Bases: NonExclusiveLimitAlarm

NonExclusiveDeviationAlarm:

class asyncua.common.event_objects.NonExclusiveLevelAlarm(sourcenode=None, message=None, severity=1)#

Bases: NonExclusiveLimitAlarm

NonExclusiveLevelAlarm:

class asyncua.common.event_objects.NonExclusiveLimitAlarm(sourcenode=None, message=None, severity=1)#

Bases: LimitAlarm

NonExclusiveLimitAlarm:

class asyncua.common.event_objects.NonExclusiveRateOfChangeAlarm(sourcenode=None, message=None, severity=1)#

Bases: NonExclusiveLimitAlarm

NonExclusiveRateOfChangeAlarm:

class asyncua.common.event_objects.OffNormalAlarm(sourcenode=None, message=None, severity=1)#

Bases: DiscreteAlarm

OffNormalAlarm:

class asyncua.common.event_objects.ProgramTransitionAuditEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditUpdateStateEvent

ProgramTransitionAuditEvent:

class asyncua.common.event_objects.ProgramTransitionEvent(sourcenode=None, message=None, severity=1)#

Bases: TransitionEvent

ProgramTransitionEvent:

class asyncua.common.event_objects.ProgressEvent(sourcenode=None, message=None, severity=1)#

Bases: BaseEvent

ProgressEvent:

class asyncua.common.event_objects.PubSubCommunicationFailureEvent(sourcenode=None, message=None, severity=1)#

Bases: PubSubStatusEvent

PubSubCommunicationFailureEvent:

class asyncua.common.event_objects.PubSubStatusEvent(sourcenode=None, message=None, severity=1)#

Bases: SystemEvent

PubSubStatusEvent:

class asyncua.common.event_objects.PubSubTransportLimitsExceedEvent(sourcenode=None, message=None, severity=1)#

Bases: PubSubStatusEvent

PubSubTransportLimitsExceedEvent:

class asyncua.common.event_objects.RefreshEndEvent(sourcenode=None, message=None, severity=1)#

Bases: SystemEvent

RefreshEndEvent:

class asyncua.common.event_objects.RefreshRequiredEvent(sourcenode=None, message=None, severity=1)#

Bases: SystemEvent

RefreshRequiredEvent:

class asyncua.common.event_objects.RefreshStartEvent(sourcenode=None, message=None, severity=1)#

Bases: SystemEvent

RefreshStartEvent:

class asyncua.common.event_objects.RoleMappingRuleChangedAuditEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditUpdateMethodEvent

RoleMappingRuleChangedAuditEvent:

class asyncua.common.event_objects.SemanticChangeEvent(sourcenode=None, message=None, severity=1)#

Bases: BaseEvent

SemanticChangeEvent:

class asyncua.common.event_objects.SystemDiagnosticAlarm(sourcenode=None, message=None, severity=1)#

Bases: OffNormalAlarm

SystemDiagnosticAlarm:

class asyncua.common.event_objects.SystemEvent(sourcenode=None, message=None, severity=1)#

Bases: BaseEvent

SystemEvent:

class asyncua.common.event_objects.SystemOffNormalAlarm(sourcenode=None, message=None, severity=1)#

Bases: OffNormalAlarm

SystemOffNormalAlarm:

class asyncua.common.event_objects.SystemStatusChangeEvent(sourcenode=None, message=None, severity=1)#

Bases: SystemEvent

SystemStatusChangeEvent:

class asyncua.common.event_objects.TransitionEvent(sourcenode=None, message=None, severity=1)#

Bases: BaseEvent

TransitionEvent:

class asyncua.common.event_objects.TripAlarm(sourcenode=None, message=None, severity=1)#

Bases: OffNormalAlarm

TripAlarm:

class asyncua.common.event_objects.TrustListOutOfDateAlarm(sourcenode=None, message=None, severity=1)#

Bases: SystemOffNormalAlarm

TrustListOutOfDateAlarm:

class asyncua.common.event_objects.TrustListUpdatedAuditEvent(sourcenode=None, message=None, severity=1)#

Bases: AuditUpdateMethodEvent

TrustListUpdatedAuditEvent:

asyncua.common.events module#

class asyncua.common.events.Event(emitting_node=2253)#

Bases: object

OPC UA Event object. This is class in inherited by the common event objects such as BaseEvent, other auto standard events and custom events Events are used to trigger events on server side and are sent to clients for every events from server

Developer Warning: On server side the data type of attributes should be known, thus add properties using the add_property method!!!

add_property(name, val, datatype)#: Add a property to event and store its data type

add_variable(name, val, datatype)#: Add a variable to event and store its data type variables are able to have properties as children

static browse_path_to_attribute_name(browsePath)#

static from_event_fields(select_clauses, fields)#: Instantiate an Event object from a select_clauses and fields

static from_field_dict(fields)#: Create an Event object from a dict of name and variants

get_event_props_as_fields_dict()#: convert all properties and variables of the Event class to a dict of variants

to_event_fields(select_clauses)#: return a field list using a select clause and the object properties

to_event_fields_using_subscription_fields(select_clauses)#: Using a new select_clauses and the original select_clauses used during subscription, return a field list

async asyncua.common.events.get_event_obj_from_type_node(node)#: return an Event object from an event type node

async asyncua.common.events.get_event_objects_from_type_node(node: Node) → List[Node]#

async asyncua.common.events.get_event_properties_from_type_node(node: Node) → List[Node]#

async asyncua.common.events.get_event_variables_from_type_node(node: Node) → List[Node]#

async asyncua.common.events.get_filter_from_event_type(eventtypes: List[Node], where_clause_generation: bool = True)#

async asyncua.common.events.select_clauses_from_evtype(evtypes: List[Node])#

async asyncua.common.events.select_event_attributes_from_type_node(node: Node, attributeSelector)#

async asyncua.common.events.where_clause_from_evtype(evtypes: List[Node])#

asyncua.common.instantiate_util module#

Instantiate a new node and its child nodes from a node type.

async asyncua.common.instantiate_util.instantiate(parent: Node, node_type: Node, nodeid: NodeId | None = None, bname: QualifiedName | str | None = None, dname: LocalizedText | None = None, idx: int = 0, instantiate_optional: bool = True) → List[Node]#: instantiate a node type under a parent node. nodeid and browse name of new node can be specified, or just namespace index If they exists children of the node type, such as components, variables and properties are also instantiated

async asyncua.common.instantiate_util.is_abstract(node_type) → bool#

asyncua.common.manage_nodes module#

High level functions to create nodes

async asyncua.common.manage_nodes.create_data_type(parent: Node, nodeid: NodeId | str | int, bname: QualifiedName | str, description: str | None = None) → Node#: Create a new data type to be used in new variables, etc .. arguments are nodeid, browsename or namespace index, name

async asyncua.common.manage_nodes.create_encoding(parent, nodeid: NodeId | str | int, bname: QualifiedName | str) → Node#: Create a new encoding object to be instantiated in address space. arguments are nodeid, browsename or namespace index, name

async asyncua.common.manage_nodes.create_folder(parent: Node, nodeid: NodeId | str | int, bname: QualifiedName | str) → Node#: create a child node folder arguments are nodeid, browsename or namespace index, name

async asyncua.common.manage_nodes.create_method(parent: Node, *args) → Node#: create a child method object This is only possible on server side!! args are nodeid, browsename, method_to_be_called, [input argument types], [output argument types] or idx, name, method_to_be_called, [input argument types], [output argument types] if argument types is specified, child nodes advertising what arguments the method uses and returns will be created a callback is a method accepting the nodeid of the parent as first argument and variants after. returns a list of variants

async asyncua.common.manage_nodes.create_object(parent: Node, nodeid: NodeId | str | int, bname: QualifiedName | str, objecttype: NodeId | int | None = None, instantiate_optional: bool = True) → Node#: create a child node object arguments are nodeid, browsename, [objecttype] or namespace index, name, [objecttype] if objectype is given (a NodeId) then the type node is instantiated inclusive its child nodes

async asyncua.common.manage_nodes.create_object_type(parent: Node, nodeid: NodeId | str | int, bname: QualifiedName | str)#: Create a new object type to be instantiated in address space. arguments are nodeid, browsename or namespace index, name

async asyncua.common.manage_nodes.create_property(parent: Node, nodeid: NodeId | str | int, bname: QualifiedName | str, val: Any, varianttype: VariantType | None = None, datatype: NodeId | int | None = None) → Node#: create a child node property args are nodeid, browsename, value, [variant type] or idx, name, value, [variant type]

async asyncua.common.manage_nodes.create_reference_type(parent: Node, nodeid: NodeId | str | int, bname: QualifiedName | str | int, symmetric: bool = True, inversename: str | None = None) → Node#: Create a new reference type args are nodeid and browsename or idx and name

async asyncua.common.manage_nodes.create_variable(parent: Node, nodeid: NodeId | str | int, bname: QualifiedName | str, val: Any, varianttype: VariantType | None = None, datatype: NodeId | int | None = None) → Node#: create a child node variable args are nodeid, browsename, value, [variant type], [data type] or idx, name, value, [variant type], [data type]

async asyncua.common.manage_nodes.create_variable_type(parent: Node, nodeid: NodeId | str | int, bname: QualifiedName | str, datatype: NodeId | int) → Node#: Create a new variable type args are nodeid, browsename and datatype or idx, name and data type

async asyncua.common.manage_nodes.delete_nodes(session: AbstractSession, nodes: Iterable[Node], recursive: bool = False, delete_target_references: bool = True) → Tuple[List[Node], List[StatusCode]]#: Delete specified nodes. Optionally delete recursively all nodes with a downward hierachic references to the node return the list of deleted node and the result

asyncua.common.methods module#

High level method related functions

async asyncua.common.methods.call_method(parent: Node, methodid: NodeId | QualifiedName | str, *args) → Any#: Call an OPC-UA method. methodid is browse name of child method or the nodeid of method as a NodeId object arguments are variants or python object convertible to variants. which may be of different types returns a list of values or a single value depending on the output of the method : param: parent Node

async asyncua.common.methods.call_method_full(parent: Node, methodid: NodeId | QualifiedName | str, *args) → CallMethodResult#: Call an OPC-UA method. methodid is browse name of child method or the nodeid of method as a NodeId object arguments are variants or python object convertible to variants. which may be of different types returns a CallMethodResult object with converted OutputArguments : param: parent Node

asyncua.common.methods.to_variant(*args: Iterable) → List[Variant]#: Create a list of ua.Variants from a given iterable of arguments.

asyncua.common.methods.uamethod(func)#: Method decorator to automatically convert arguments and output to and from variants

asyncua.common.node module#

High level node object, to access node attribute and browse address space

class asyncua.common.node.Node(session: AbstractSession, nodeid: Node | NodeId | str | int)#

Bases: object

High level node object, to access node attribute, browse and populate address space. Node objects are useful as-is but they do not expose the entire OPC-UA protocol. Feel free to look at the code of this class and call directly UA services methods to optimize your code

async add_data_type(nodeid: NodeId | str | int, bname: QualifiedName | str, description: str | None = None) → Node#

async add_folder(nodeid: NodeId | str | int, bname: QualifiedName | str) → Node#

async add_method(*args) → Node#

async add_object(nodeid: NodeId | str | int, bname: QualifiedName | str, objecttype: NodeId | int | None = None, instantiate_optional: bool = True) → Node#

async add_object_type(nodeid: NodeId | str | int, bname: QualifiedName | str) → Node#

async add_property(nodeid: NodeId | str | int, bname: QualifiedName | str, val: Any, varianttype: VariantType | None = None, datatype: NodeId | int | None = None) → Node#

async add_reference(target: Node | NodeId | str | int, reftype: int, forward: bool = True, bidirectional: bool = True) → None#: Add reference to node

async add_reference_type(nodeid: NodeId | str | int, bname: QualifiedName | str, symmetric: bool = True, inversename: str | None = None) → Node#

async add_variable(nodeid: NodeId | str | int, bname: QualifiedName | str, val: Any, varianttype: VariantType | None = None, datatype: NodeId | int | None = None) → Node#

async add_variable_type(nodeid: NodeId | str | int, bname: QualifiedName | str, datatype: NodeId | int) → Node#

async call_method(methodid: NodeId | QualifiedName | str, *args) → Any#

async delete(delete_references: bool = True, recursive: bool = False) → List[Node]#: Delete node from address space

async delete_reference(target: Node | NodeId | str | int, reftype: int, forward: bool = True, bidirectional: bool = True) → None#: Delete given node’s references from address space

async get_access_level() → Set[AccessLevel]#: Get the access level attribute of the node as a set of AccessLevel enum values.

async get_child(path: QualifiedName | str | Iterable[QualifiedName | str], return_all: Literal[False] = False) → Node#
async get_child(path: QualifiedName | str | Iterable[QualifiedName | str], return_all: Literal[True] = True) → List[Node]: get a child specified by its path from this node. A path might be: * a string representing a relative path as per UA spec * a qualified name * a list of string representing browsenames (legacy) * a list of qualified names

async get_children(refs: int = 33, nodeclassmask: NodeClass = NodeClass.Unspecified) → List[Node]#: Get all children of a node. By default hierarchical references and all node classes are returned. Other reference types may be given: References = 31 NonHierarchicalReferences = 32 HierarchicalReferences = 33 HasChild = 34 Organizes = 35 HasEventSource = 36 HasModellingRule = 37 HasEncoding = 38 HasDescription = 39 HasTypeDefinition = 40 GeneratesEvent = 41 Aggregates = 44 HasSubtype = 45 HasProperty = 46 HasComponent = 47 HasNotifier = 48 HasOrderedComponent = 49

async get_children_by_path(paths: Iterable[QualifiedName | str | Iterable[QualifiedName | str]], raise_on_partial_error: bool = True) → List[List[Node | None]]#: get children specified by their paths from this node. A path might be: * a string representing a relative path as per UA spec * a qualified name * a list of string representing browsenames (legacy) * a list of qualified names

async get_children_descriptions(refs: int = 33, nodeclassmask: NodeClass = NodeClass.Unspecified, includesubtypes: bool = True, result_mask: BrowseResultMask = BrowseResultMask.All) → List[ReferenceDescription]#

async get_description_refs() → List[Node]#

async get_encoding_refs() → List[Node]#

async get_methods() → List[Node]#: return methods of node. methods are child nodes with a reference of type HasComponent and a NodeClass of Method

async get_parent() → Node | None#: returns parent of the node. A Node may have several parents, the first found is returned. This method uses reverse references, a node might be missing such a link, thus we will not find its parent.

async get_path(max_length: int = 20, as_string: Literal[False] = False) → List[Node]#
async get_path(max_length: int = 20, as_string: Literal[True] = True) → List[str]: Attempt to find path of node from root node and return it as a list of Nodes. There might several possible paths to a node, this function will return one Some nodes may be missing references, so this method may return an empty list Since address space may have circular references, a max length is specified

async get_properties() → List[Node]#: return properties of node. properties are child nodes with a reference of type HasProperty and a NodeClass of Variable COROUTINE

async get_referenced_nodes(refs: int = 31, direction: BrowseDirection = BrowseDirection.Both, nodeclassmask: NodeClass = NodeClass.Unspecified, includesubtypes: bool = True) → List[Node]#: returns referenced nodes based on specific filter Parameters are the same as for get_references

async get_references(refs: int = 31, direction: BrowseDirection = BrowseDirection.Both, nodeclassmask: NodeClass = NodeClass.Unspecified, includesubtypes: bool = True, result_mask: BrowseResultMask = BrowseResultMask.All) → List[ReferenceDescription]#

returns references of the node based on specific filter defined with:

refs = ObjectId of the Reference direction = Browse direction for references nodeclassmask = filter nodes based on specific class includesubtypes = If true subtypes of the reference (ref) are also included result_mask = define what results information are requested

async get_user_access_level() → Set[AccessLevel]#: Get the user access level attribute of the node as a set of AccessLevel enum values.

async get_value() → Any#: Get value of a node as a python type. Only variables ( and properties) have values. An exception will be generated for other node types. WARNING: on server side, this function returns a ref to object in ua database. Do not modify it if it is a mutable object unless you know what you are doing

async get_variables() → List[Node]#: return variables of node. variables are child nodes with a reference of type HasComponent and a NodeClass of Variable

async history_read(details: ReadRawModifiedDetails, continuation_point: bytes | None = None) → HistoryReadResult#: Read raw history of a node, low-level function result code from server is checked and an exception is raised in case of error

async history_read_events(details: Iterable[ReadEventDetails]) → HistoryReadResult#: Read event history of a node, low-level function result code from server is checked and an exception is raised in case of error

static new_node(session, nodeid: NodeId) → Node#: Helper function to init nodes with out importing Node

async read_array_dimensions() → int#: Read and return ArrayDimensions attribute of node

async read_attribute(attr: AttributeIds, indexrange: str | None = None, raise_on_bad_status: bool = True) → DataValue#: Read one attribute of a node attributeid is a member of ua.AttributeIds indexrange is a NumericRange (a string; e.g. “1” or “1:3”. result code from server is checked and an exception is raised in case of error

async read_attributes(attrs: Iterable[AttributeIds]) → List[DataValue]#: Read several attributes of a node list of DataValue is returned

async read_browse_name() → QualifiedName#: Get browse name of a node. A browse name is a QualifiedName object composed of a string(name) and a namespace index.

async read_data_type() → NodeId#: get data type of node as NodeId

async read_data_type_as_variant_type() → VariantType#: get data type of node as VariantType This only works if node is a variable, otherwise type may not be convertible to VariantType

async read_data_type_definition() → DataTypeDefinition#: read data type definition attribute of node only DataType nodes following spec >= 1.04 have that attribute

async read_data_value(raise_on_bad_status: bool = True) → DataValue#: Get value of a node as a DataValue object. Only variables (and properties) have values. An exception will be generated for other node types. DataValue contain a variable value as a variant as well as server and source timestamps

async read_description() → LocalizedText#: get description attribute class of node

async read_display_name() → LocalizedText#: get DisplayName attribute of node

async read_event_history(starttime: datetime | None = None, endtime: datetime | None = None, numvalues: int = 0, evtypes: Node | NodeId | str | int | Iterable[Node | NodeId | str | int] = 2041) → List[Event]#: Read event history of a source node result code from server is checked and an exception is raised in case of error If numvalues is > 0 and number of events in period is > numvalues then result will be truncated

async read_event_notifier() → Set[EventNotifier]#: Get the event notifier attribute of the node as a set of EventNotifier enum values.

async read_node_class() → NodeClass#: get node class attribute of node

async read_params(params: ReadParameters) → List[DataValue]#

async read_raw_history(starttime: datetime | None = None, endtime: datetime | None = None, numvalues: int = 0, return_bounds: bool = True) → List[DataValue]#: Read raw history of a node result code from server is checked and an exception is raised in case of error If numvalues is > 0 and number of events in period is > numvalues then result will be truncated

async read_type_definition() → NodeId | None#: returns type definition of the node.

async read_value() → Any#: Get value of a node as a python type. Only variables ( and properties) have values. An exception will be generated for other node types. WARNING: on server side, this function returns a ref to object in ua database. Do not modify it if it is a mutable object unless you know what you are doing

async read_value_rank() → int#: Read and return ValueRank attribute of node

async register() → None#: Register node for faster read and write access (if supported by server) Rmw: This call modifies the nodeid of the node, the original nodeid is available as node.basenodeid

async set_attr_bit(attr: AttributeIds, bit: int) → None#

async set_data_value(value: Any, varianttype: VariantType | None = None) → None#: Write value of a node. Only variables(properties) have values. An exception will be generated for other node types. value argument is either: * a python built-in type, converted to opc-ua optionally using the variantype argument. * a ua.Variant, varianttype is then ignored * a ua.DataValue, you then have full control over data send to server WARNING: On server side, ref to object is directly saved in our UA db, if this is a mutable object and you modify it afterward, then the object in db will be modified without any data change event generated

async set_event_notifier(values) → None#

Set the event notifier attribute.

Parameters:: values – an iterable of EventNotifier enum values.

async set_modelling_rule(mandatory: bool) → None#: Add a modelling rule reference to Node. When creating a new object type, its variable and child nodes will not be instantiated if they do not have modelling rule if mandatory is None, the modelling rule is removed

async set_read_only() → None#: Set a node as read-only for clients. A node is always writable on server side.

async set_value(value: Any, varianttype: VariantType | None = None) → None#: Write value of a node. Only variables(properties) have values. An exception will be generated for other node types. value argument is either: * a python built-in type, converted to opc-ua optionally using the variantype argument. * a ua.Variant, varianttype is then ignored * a ua.DataValue, you then have full control over data send to server WARNING: On server side, ref to object is directly saved in our UA db, if this is a mutable object and you modify it afterward, then the object in db will be modified without any data change event generated

async set_writable(writable: bool = True) → None#: Set node as writable by clients. A node is always writable on server side.

async unregister() → None#

async unset_attr_bit(attr: AttributeIds, bit: int) → None#

async write_array_dimensions(value: int) → None#: Set attribute ArrayDimensions of node make sure it has the correct data type

async write_attribute(attributeid: AttributeIds, datavalue: DataValue, indexrange: str | None = None) → None#: Set an attribute of a node attributeid is a member of ua.AttributeIds datavalue is a ua.DataValue object indexrange is a NumericRange (a string; e.g. “1” or “1:3”.

See https://reference.opcfoundation.org/v104/Core/docs/Part4/7.22/)

async write_data_type_definition(sdef: DataTypeDefinition) → None#: write data type definition attribute of node only DataType nodes following spec >= 1.04 have that attribute

async write_params(params: WriteParameters) → List[StatusCode]#

async write_value(value: Any, varianttype: VariantType | None = None) → None#: Write value of a node. Only variables(properties) have values. An exception will be generated for other node types. value argument is either: * a python built-in type, converted to opc-ua optionally using the variantype argument. * a ua.Variant, varianttype is then ignored * a ua.DataValue, you then have full control over data send to server WARNING: On server side, ref to object is directly saved in our UA db, if this is a mutable object and you modify it afterward, then the object in db will be modified without any data change event generated

async write_value_rank(value: int) → None#: Set attribute ValueRank of node

asyncua.common.node_factory module#

asyncua.common.node_factory.make_node(session: AbstractSession, nodeid: NodeId) → Node#: Node factory Needed no break cyclical import of Node

asyncua.common.shortcuts module#

class asyncua.common.shortcuts.Shortcuts(server)#

Bases: object

This object contains Node objects to some commonly used nodes

HasComponent: Node#

HasEncoding: Node#

HasProperty: Node#

Organizes: Node#

base_data_type: Node#

base_event_type: Node#

base_object_type: Node#

base_structure_type: Node#

base_union_type: Node#

base_variable_type: Node#

data_types: Node#

enum_data_type: Node#

event_types: Node#

folder_type: Node#

namespace_array: Node#

namespaces: Node#

object_types: Node#

objects: Node#

opc_binary: Node#

option_set_type: Node#

reference_types: Node#

root: Node#

server: Node#

server_state: Node#

service_level: Node#

types: Node#

variable_types: Node#

asyncua.common.statemachine module#

https://reference.opcfoundation.org/v104/Core/docs/Part10/ https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.1/ Basic statemachines described in OPC UA Spec.: StateMachineType FiniteStateMachineType ExclusiveLimitStateMachineType - not implemented FileTransferStateMachineType - not implemented ProgramStateMachineType - not implemented ShelvedStateMachineType - not implemented Relevant information: Overview - https://reference.opcfoundation.org/v104/Core/docs/Part10/5.2.3/#5.2.3.1 States - https://reference.opcfoundation.org/v104/Core/docs/Part10/5.2.3/#5.2.3.2 Transitions - https://reference.opcfoundation.org/v104/Core/docs/Part10/5.2.3/#5.2.3.3 Events - https://reference.opcfoundation.org/v104/Core/docs/Part10/5.2.5/

class asyncua.common.statemachine.ExclusiveLimitStateMachine(server=None, parent=None, idx=None, name=None)#

Bases: FiniteStateMachine

NOT IMPLEMENTED “ExclusiveLimitStateMachineType”

class asyncua.common.statemachine.FileTransferStateMachine(server=None, parent=None, idx=None, name=None)#

Bases: FiniteStateMachine

NOT IMPLEMENTED “FileTransferStateMachineType” https://reference.opcfoundation.org/v104/Core/ObjectTypes/FileTransferStateMachineType/ https://reference.opcfoundation.org/v104/Core/docs/Part5/C.4.6/

class asyncua.common.statemachine.FiniteStateMachine(server: Server | None = None, parent: Node | None = None, idx: int | None = None, name: str | None = None)#

Bases: StateMachine

Implementation of an FiniteStateMachineType a little more advanced than the basic one if you need to know the available states and transition from clientside

async set_available_states(states: List[NodeId])#

async set_available_transitions(transitions: List[NodeId])#

class asyncua.common.statemachine.ProgramStateMachine(server=None, parent=None, idx=None, name=None)#

Bases: FiniteStateMachine

https://reference.opcfoundation.org/v104/Core/docs/Part10/4.2.3/ Implementation of an ProgramStateMachine its quite a complex statemachine with the optional possibility to make the statchange from clientside via opcua-methods

class asyncua.common.statemachine.ShelvedStateMachine(server=None, parent=None, idx=None, name=None)#

Bases: FiniteStateMachine

NOT IMPLEMENTED “ShelvedStateMachineType”

class asyncua.common.statemachine.State(id, name: str | None = None, number: int | None = None, node: Node | None = None)#

Bases: object

Helperclass for States (StateVariableType) https://reference.opcfoundation.org/v104/Core/docs/Part5/B.4.3/ name: type string will be converted automatically to qualifiedname

-> Name is a QualifiedName which uniquely identifies the current state within the StateMachineType.

id: “BaseVariableType” Id is a name which uniquely identifies the current state within the StateMachineType. A subtype may restrict the DataType. number: Number is an integer which uniquely identifies the current state within the StateMachineType.

class asyncua.common.statemachine.StateMachine(server: Server | None = None, parent: Node | None = None, idx: int | None = None, name: str | None = None)#

Bases: object

Implementation of an StateMachineType (most basic type) CurrentState: Mandatory “StateVariableType” LastTransition: Optional “TransitionVariableType” Generates TransitionEvent’s

async add_state(state: ~asyncua.common.statemachine.State, state_type: ~asyncua.ua.uatypes.NodeId = NodeId(Identifier=2307, NamespaceIndex=0, NodeIdType=<NodeIdType.FourByte: 1>), optionals: bool = False)#: this method adds a state object to the statemachines address space state: State, InitialStateType: ua.NodeId(2309, 0), StateType: ua.NodeId(2307, 0), ChoiceStateType: ua.NodeId(15109,0),

async add_transition(transition: ~asyncua.common.statemachine.Transition, transition_type: ~asyncua.ua.uatypes.NodeId = NodeId(Identifier=2310, NamespaceIndex=0, NodeIdType=<NodeIdType.FourByte: 1>), optionals: bool = False)#: this method adds a transition object to the statemachines address space transition: Transition, transition_type: ua.NodeId(2310, 0),

async change_state(state: State, transition: Transition | None = None, event_msg: str | LocalizedText | None = None, severity: int = 500)#: method to change the state of the statemachine state: “State” mandatory transition: “Transition” optional event_msg: “LocalizedText” optional severity: “Int” optional

async init(statemachine: Node)#: initialize and get subnodes

async install(optionals: bool = False)#: setup adressspace

class asyncua.common.statemachine.Transition(id, name: str | None = None, number: int | None = None, node: Node | None = None)#

Bases: object

Helper class for Transitions (TransitionVariableType) https://reference.opcfoundation.org/v104/Core/docs/Part5/B.4.4/ name: type string will be converted automatically to qualifiedname

-> Name is a QualifiedName which uniquely identifies a transition within the StateMachineType.

id: “BaseVariableType” Id is a name which uniquely identifies a Transition within the StateMachineType. A subtype may restrict the DataType. number: Number is an integer which uniquely identifies the current state within the StateMachineType. transitiontime: TransitionTime specifies when the transition occurred. effectivetransitiontime: EffectiveTransitionTime specifies the time when the current state or one of its substates was entered. If, for example, a StateA is active and – while active – switches several times between its substates SubA and SubB, then the TransitionTime stays at the point in time when StateA became active whereas the EffectiveTransitionTime changes with each change of a substate.

asyncua.common.structures module#

Support for custom structures in client and server We only support a subset of features but should be enough for custom structures

class asyncua.common.structures.EnumType(name)#

Bases: object

get_code()#

class asyncua.common.structures.EnumeratedValue(name, value)#: Bases: object

class asyncua.common.structures.Field(name)#: Bases: object

class asyncua.common.structures.Struct(name)#

Bases: object

get_code()#

class asyncua.common.structures.StructGenerator#

Bases: object

get_python_classes(env=None)#

make_model_from_file(path)#

make_model_from_string(xml)#

save_to_file(path, register=False)#

set_typeid(name, typeid)#

async asyncua.common.structures.load_enums(server, env=None, force=False)#: Read enumeration data types on server and generate python Enums in ua scope for them

async asyncua.common.structures.load_type_definitions(server, nodes=None)#: Download xml from given variable node defining custom structures. If no node is given, attempts to import variables from all nodes under “0:OPC Binary” the code is generated and imported on the fly. If you know the structures are not going to be modified it might be interesting to copy the generated files and include them in you code

asyncua.common.structures104 module#

class asyncua.common.structures104.DataTypeSorter(data_type: NodeId, name: str, desc: ReferenceDescription, sdef: StructureDefinition)#

Bases: object

depends_on(other: DataTypeSorter)#

dtype_index: Dict[NodeId, DataTypeSorter] = {}#

referenced_dtypes: Set[NodeId] = {}#

asyncua.common.structures104.clean_name(name)#: Remove characters that might be present in OPC UA structures but cannot be part of Python class names

asyncua.common.structures104.get_default_value(uatype, enums=None, hack=False, optional=False)#

async asyncua.common.structures104.load_basetype_alias_xml_import(server, name, nodeid, parent_datatype_nid)#: Insert alias for a datatype used for xml import

async asyncua.common.structures104.load_custom_struct(node: Node) → Any#

async asyncua.common.structures104.load_custom_struct_xml_import(node_id: NodeId, attrs: DataTypeAttributes)#: This function is used to load custom structs from xmlimporter

async asyncua.common.structures104.load_data_type_definitions(server: Server | Client, base_node: Node = None, overwrite_existing=False) → Dict#: Read DataTypeDefinition attribute on all Structure and Enumeration defined on server and generate Python objects in ua namespace to be used to talk with server

async asyncua.common.structures104.load_enum_xml_import(node_id: NodeId, attrs: DataTypeAttributes, option_set: bool)#: This function is used to load enums from xmlimporter

async asyncua.common.structures104.load_enums(server: Server | Client, base_node: Node = None, option_set: bool = False) → Dict#

asyncua.common.structures104.make_basetype_code(name, parent_datatype)#: alias basetypes

asyncua.common.structures104.make_enum_code(name, edef, option_set)#: if node has a DataTypeDefinition attribute, generate enum code

asyncua.common.structures104.make_structure_code(data_type, struct_name, sdef, log_error=True)#: given a StructureDefinition object, generate Python code

async asyncua.common.structures104.new_enum(server: Server | Client, idx: int | NodeId, name: int | QualifiedName, values: List[str], option_set: bool = False) → Node#

async asyncua.common.structures104.new_struct(server: Server | Client, idx: int | NodeId, name: int | QualifiedName, fields: List[StructureField], is_union: bool = False) → Tuple[Node, List[Node]]#: simple way to create a new structure return the created data type node and the list of encoding nodes

asyncua.common.structures104.new_struct_field(name: str, dtype: NodeId | Node | VariantType, array: bool = False, optional: bool = False, description: str = '') → StructureField#: simple way to create a StructureField

asyncua.common.subscription module#

high level interface to subscriptions

class asyncua.common.subscription.DataChangeNotif(subscription_data: SubscriptionItemData, monitored_item: MonitoredItemNotification)#

Bases: object

To be send to clients for every datachange notification from server.

class asyncua.common.subscription.DataChangeNotificationHandler(*args, **kwargs)#

Bases: Protocol

datachange_notification(node: Node, val: Any, data: DataChangeNotif) → None#: called for every datachange notification from server

class asyncua.common.subscription.EventNotificationHandler(*args, **kwargs)#

Bases: Protocol

event_notification(event: EventNotificationList) → None#: called for every event notification from server

class asyncua.common.subscription.StatusChangeNotificationHandler(*args, **kwargs)#

Bases: Protocol

status_change_notification(status: StatusChangeNotification) → None#: called for every status change notification from server

class asyncua.common.subscription.SubHandler#

Bases: object

Subscription Handler. To receive events from server for a subscription This class is just a sample class. Whatever class having these methods can be used

datachange_notification(node: Node, val: Any, data: DataChangeNotif) → None#: called for every datachange notification from server

event_notification(event: EventNotificationList) → None#: called for every event notification from server

status_change_notification(status: StatusChangeNotification) → None#: called for every status change notification from server

class asyncua.common.subscription.Subscription(server: InternalSession | UaClient, params: ua.CreateSubscriptionParameters, handler: SubscriptionHandler)#

Bases: object

Subscription object returned by Server or Client objects. The object represent a subscription to an opc-ua server. This is a high level class, especially subscribe_data_change and subscribe_events methods. If more control is necessary look at code and/or use create_monitored_items method. :param server: InternalSession or UaClient

async create_monitored_items(monitored_items: Iterable[MonitoredItemCreateRequest]) → List[int | StatusCode]#: low level method to have full control over subscription parameters. Client handle must be unique since it will be used as key for internal registration of data.

async deadband_monitor(var: Node, deadband_val: Double, deadbandtype: UInt32 = 1, queuesize: int = 0, attr: AttributeIds = ua.AttributeIds.Value) → int#
async deadband_monitor(var: Iterable[Node], deadband_val: Double, deadbandtype: UInt32 = 1, queuesize: int = 0, attr: AttributeIds = ua.AttributeIds.Value) → List[int | StatusCode]: Method to create a subscription with a Deadband Value. Default deadband value type is absolute. Return a handle which can be used to unsubscribe :param var: Variable to which you want to subscribe :param deadband_val: Absolute float value :param deadbandtype: Default value is 1 (absolute), change to 2 for percentage deadband :param queuesize: Wanted queue size, default is 1 :param attr: Attribute ID

async delete() → None#: Delete subscription on server. This is automatically done by Client and Server classes on exit.

async init() → CreateSubscriptionResult#

async modify_monitored_item(handle: int, new_samp_time: Double, new_queuesize: int = 0, mod_filter_val: int = -1) → List[MonitoredItemModifyResult]#: Modify a monitored item. :param handle: Handle returned when originally subscribing :param new_samp_time: New wanted sample time :param new_queuesize: New wanted queuesize, default is 0 :param mod_filter_val: New deadband filter value :return: Return a Modify Monitored Item Result

async publish_callback(publish_result: PublishResult) → None#: Handle PublishResult callback.

async set_monitoring_mode(monitoring: MonitoringMode) → List[StatusCode]#

The monitoring mode parameter is used to enable/disable the sampling of MonitoredItems (Samples don’t queue on the server side)

Parameters:: monitoring – The monitoring mode to apply
Returns:: Return a Set Monitoring Mode Result

async set_publishing_mode(publishing: bool) → List[StatusCode]#

Disable publishing of NotificationMessages for the subscription, but doesn’t discontinue the sending of keep-alive Messages, nor change the monitoring mode.

Parameters:: publishing – The publishing mode to apply
Returns:: Return a Set Publishing Mode Result

Subscribe to alarm and condition events from a node. Default node is Server node. In many servers the server node is the only one you can subscribe to. If evtypes is not provided, evtype defaults to ConditionType. If evtypes is a list or tuple of custom event types, the events will be filtered to the supplied types. A handle (integer value) is returned which can be used to modify/cancel the subscription.

Parameters:

sourcenode – int, str, ua.NodeId or Node
evtypes – ua.ObjectIds, str, ua.NodeId or Node
evfilter – ua.EventFilter which provides the SelectClauses and WhereClause
queuesize – 0 for default queue size, 1 for minimum queue size, n for FIFO queue,

MaxUInt32 for max queue size :return: Handle for changing/cancelling of the subscription

async subscribe_data_change(nodes: Node, attr: AttributeIds = ua.AttributeIds.Value, queuesize: int = 0, monitoring=ua.MonitoringMode.Reporting, sampling_interval: Double = 0.0) → int#

async subscribe_data_change(nodes: Node | Iterable[Node], attr: AttributeIds = ua.AttributeIds.Value, queuesize: int = 0, monitoring=ua.MonitoringMode.Reporting, sampling_interval: Double = 0.0) → List[int | StatusCode]

Subscribe to data change events of one or multiple nodes. The default attribute used for the subscription is Value. Return value is a handle which can be used to modify/cancel the subscription. The handle is an integer value for single Nodes. If the creation of the subscription fails an UaStatusCodeError is raised. If multiple Nodes are supplied, a List of integers or ua.StatusCode objects is returned. A list of StatusCode objects are returned to indicate that the subscription has failed (no exception will be raised in this case). If more control is necessary the create_monitored_items method can be used directly.

Parameters:

nodes – One Node or an Iterable of Nodes
attr – The Node attribute you want to subscribe to
queuesize – 0 or 1 for default queue size (shall be 1 - no queuing), n for FIFO queue
sampling_interval – ua.Duration

Returns:

Handle for changing/cancelling of the subscription

Subscribe to events from a node. Default node is Server node. In most servers the server node is the only one you can subscribe to. If evtypes is not provided, evtype defaults to BaseEventType. If evtypes is a list or tuple of custom event types, the events will be filtered to the supplied types. A handle (integer value) is returned which can be used to modify/cancel the subscription.

Parameters:

sourcenode – int, str, ua.NodeId or Node
evtypes – ua.ObjectIds, str, ua.NodeId or Node
evfilter – ua.EventFilter which provides the SelectClauses and WhereClause
queuesize – 0 for default queue size, 1 for minimum queue size, n for FIFO queue,

MaxUInt32 for max queue size :param where_clause_generation: No where_clause generation when no eventfilter is provided. Need for TwinCAT, Codesys :return: Handle for changing/cancelling of the subscription

async unsubscribe(handle: int | Iterable[int]) → None#: Unsubscribe from datachange or events using the handle returned while subscribing. If you delete the subscription, you do not need to unsubscribe. :param handle: The handle that was returned when subscribing to the node/nodes

async update(params: ModifySubscriptionParameters) → ModifySubscriptionResult#

asyncua.common.subscription.SubscriptionHandler#

Protocol class representing subscription handlers to receive events from server.

alias of Union[DataChangeNotificationHandler, EventNotificationHandler, StatusChangeNotificationHandler]

class asyncua.common.subscription.SubscriptionItemData#

Bases: object

To store useful data from a monitored item.

asyncua.common.type_dictionary_builder module#

class asyncua.common.type_dictionary_builder.DataTypeDictionaryBuilder(server, idx, ns_urn, dict_name, dict_node_id=None)#

Bases: object

add_field(type_name, variable_name, struct_name, is_array=False)#

async create_data_type(type_name, nodeid=None, init=True)#

async init()#

async set_dict_byte_string()#

class asyncua.common.type_dictionary_builder.OPCTypeDictionaryBuilder(ns_urn)#

Bases: object

add_field(type_name, variable_name, struct_name, is_array=False)#

append_struct(name)#

get_dict_value()#

indent(elem, level=0)#

class asyncua.common.type_dictionary_builder.StructNode(type_dict, data_type, name, node_ids)#

Bases: object

add_field(type_name, field_name, is_array=False)#

asyncua.common.type_dictionary_builder.get_ua_class(ua_class_name)#

asyncua.common.ua_utils module#

Useful methods and classes not belonging anywhere and depending on asyncua library

asyncua.common.ua_utils.copy_dataclass_attr(dc_source, dc_dest) → None#: Copy the common attributes of dc_source to dc_dest

asyncua.common.ua_utils.data_type_to_string(dtype)#

async asyncua.common.ua_utils.data_type_to_variant_type(dtype_node)#: Given a Node datatype, find out the variant type to encode data. This is not exactly straightforward…

async asyncua.common.ua_utils.get_base_data_type(datatype)#

Looks up the base datatype of the provided datatype Node The base datatype is either: A primitive type (ns=0, i<=21) or a complex one (ns=0 i>21 and i<30) like Enum and Struct.

Args:: datatype: NodeId of a datype of a variable
Returns:: NodeId of datatype base or None in case base datype can not be determined

asyncua.common.ua_utils.get_default_value(uatype)#

async asyncua.common.ua_utils.get_node_children(node, nodes=None)#: Get recursively all children of a node

async asyncua.common.ua_utils.get_node_subtypes(node, nodes=None)#

async asyncua.common.ua_utils.get_node_supertype(node)#: return node supertype or None

async asyncua.common.ua_utils.get_node_supertypes(node, includeitself=False, skipbase=True)#: return get all subtype parents of node recursive :param node: can be an ua.Node or ua.NodeId :param includeitself: include also node to the list :param skipbase don’t include the toplevel one :returns list of ua.Node, top parent first

async asyncua.common.ua_utils.get_nodes_of_namespace(server, namespaces=None)#

Get the nodes of one or more namespaces . Args:

server: opc ua server to use namespaces: list of string uri or int indexes of the namespace to export

Returns:: List of nodes that are part of the provided namespaces

async asyncua.common.ua_utils.is_child_present(node, browsename)#: return if a browsename is present a child from the provide node :param node: node wherein to find the browsename :param browsename: browsename to search :returns returns True if the browsename is present else False

async asyncua.common.ua_utils.is_subtype(node, supertype)#: return if a node is a subtype of a specified nodeid

asyncua.common.ua_utils.string_to_val(string, vtype)#: Convert back a string to a python or python-asyncua object Note: no error checking is done here, supplying null strings could raise exceptions (datetime and guid)

asyncua.common.ua_utils.string_to_variant(string, vtype)#: convert back a string to an ua.Variant

asyncua.common.ua_utils.val_to_string(val, truncate=False)#: convert a python object or python-asyncua object to a string which should be easy to understand for human easy to modify, and not too hard to parse back ….not easy meant for UI or command lines if truncate is true then huge strings or bytes are truncated

asyncua.common.ua_utils.value_to_datavalue(val, varianttype=None)#: convert anything to a DataValue using varianttype

asyncua.common.ua_utils.variant_to_string(var)#: convert a variant to a string which should be easy to understand for human easy to modify, and not too hard to parse back ….not easy meant for UI or command lines

asyncua.common.utils module#

Helper function and classes that do not rely on asyncua library. Helper function and classes depending on ua object are in ua_utils.py

class asyncua.common.utils.Buffer(data, start_pos=0, size=-1)#

Bases: object

Alternative to io.BytesIO making debug easier and added a few convenience methods.

copy(size=-1)#: return a shadow copy, optionally only copy ‘size’ bytes

property cur_pos#

read(size)#: read and pop number of bytes for buffer

rewind(cur_pos=0)#: rewind the buffer

skip(size)#: skip size bytes in buffer

exception asyncua.common.utils.NotEnoughData#: Bases: UaError

exception asyncua.common.utils.ServiceError(code)#: Bases: UaError

exception asyncua.common.utils.SocketClosedException#: Bases: UaError

asyncua.common.utils.create_nonce(size=32)#

asyncua.common.utils.fields_with_resolved_types(class_or_instance: Any, globalns: Dict[str, Any] | None = None, localns: Dict[str, Any] | None = None, include_extras: bool = False) → Tuple[Field, ...]#

Return a tuple describing the fields of this dataclass.

Accepts a dataclass or an instance of one. Tuple elements are of type Field. ForwardRefs and string types will be resolved.

asyncua.common.xmlexporter module#

from a list of nodes in the address space, build an XML file format is the one from opc-ua specification

class asyncua.common.xmlexporter.XmlExporter(server: Server | Client, export_values: bool = False)#

Bases: object

If it is required that for _extobj_to_etree members to the value should be written in a certain order it can be added to the dictionary below.

async add_etree_datatype(obj)#: Add a UA data type element to the XML etree

async add_etree_method(node)#

async add_etree_object(node)#: Add a UA object element to the XML etree

async add_etree_object_type(node)#: Add a UA object type element to the XML etree

async add_etree_reference_type(obj)#

async add_etree_variable(node)#: Add a UA variable element to the XML etree

async add_etree_variable_type(node)#: Add a UA variable type element to the XML etree

async add_variable_common(node, el, export_value: bool)#

async build_etree(node_list)#

Create an XML etree object from a list of nodes; Namespaces used by nodes are always exported for consistency. Args:

node_list: list of Node objects for export

Returns:

dump_etree()#: Dump etree to console for debugging Returns:

extobj_ordered_elements = {NodeId(Identifier=296, NamespaceIndex=0, NodeIdType=<NodeIdType.FourByte: 1>): ['Name', 'DataType', 'ValueRank', 'ArrayDimensions', 'Description']}#

async member_to_etree(el, name, dtype, val)#

async node_to_etree(node)#

Add the necessary XML sub elements to the etree for exporting the node Args:

node: Node object which will be added to XML etree

Returns:

async value_to_etree(el, dtype_name, dtype, node)#

async write_xml(xmlpath, pretty=True)#

Write the XML etree in the exporter object to a file Args:

xmlpath: string representing the path/file name pretty: add spaces and newlines, to be more readable

Returns:

asyncua.common.xmlexporter.indent(elem, level=0)#: copy and paste from http://effbot.org/zone/element-lib.htm#prettyprint it basically walks your tree and adds spaces and newlines so the tree is printed in a nice way

asyncua.common.xmlimporter module#

add nodes defined in XML to address space format is the one from opc-ua specification

class asyncua.common.xmlimporter.XmlImporter(server: Server | Client, strict_mode: bool = True)#

Bases: object

async add_datatype(obj, no_namespace_migration=False)#

async add_method(obj, no_namespace_migration=False)#

async add_object(obj, no_namespace_migration=False)#

async add_object_type(obj, no_namespace_migration=False)#

async add_reference_type(obj, no_namespace_migration=False)#

async add_variable(obj, no_namespace_migration=False)#

async add_variable_type(obj, no_namespace_migration=False)#

async import_xml(xmlpath=None, xmlstring=None)#: import xml and return added nodes

make_objects(node_data)#

asyncua.common.xmlparser module#

parse xml file from asyncua-spec

class asyncua.common.xmlparser.ExtObj#: Bases: object

class asyncua.common.xmlparser.Field(data)#: Bases: object

class asyncua.common.xmlparser.NodeData#: Bases: object

class asyncua.common.xmlparser.RefStruct#: Bases: object

class asyncua.common.xmlparser.XMLParser#

Bases: object

get_aliases() → dict#: Return the used node aliases in this import file

get_node_datas()#

get_nodeset_namespaces() → List[Tuple[str, String, DateTime]]#: Get all namespaces that are registered with version and date_time

get_used_namespaces()#: Return the used namespace uris in this import file

static list_required_models(xmlpath, xmlstring)#: Try getting required XML Models, before parsing NodeSet

async parse(xmlpath=None, xmlstring=None)#

parse_sync(xmlpath=None, xmlstring=None)#

asyncua.common.xmlparser.ua_type_to_python(val, uatype_as_str)#: Converts a string value to a python value according to ua_utils.