asyncua.ua package#

Subpackages#

asyncua.ua.uaerrors package

Submodules#

asyncua.ua.attribute_ids module#

class asyncua.ua.attribute_ids.AttributeIds(value)#

Bases: IntEnum

An enumeration.

AccessLevel = 17#

AccessLevelEx = 27#

AccessRestrictions = 26#

ArrayDimensions = 16#

BrowseName = 3#

ContainsNoLoops = 11#

DataType = 14#

DataTypeDefinition = 23#

Description = 5#

DisplayName = 4#

EventNotifier = 12#

Executable = 21#

Historizing = 20#

InverseName = 10#

IsAbstract = 8#

MinimumSamplingInterval = 19#

NodeClass = 2#

NodeId = 1#

RolePermissions = 24#

Symmetric = 9#

UserAccessLevel = 18#

UserExecutable = 22#

UserRolePermissions = 25#

UserWriteMask = 7#

Value = 13#

ValueRank = 15#

WriteMask = 6#

asyncua.ua.object_ids module#

Note

This module is very large and currently excluded from the API documentation

asyncua.ua.status_codes module#

Note

This module is very large and currently excluded from the API documentation

asyncua.ua.ua_binary module#

Binary protocol specific functions and constants

class asyncua.ua.ua_binary.Primitives#

Bases: Primitives1

ByteString = <asyncua.ua.ua_binary._Bytes object>#

Bytes = <asyncua.ua.ua_binary._Bytes object>#

CharArray = <asyncua.ua.ua_binary._String object>#

DateTime = <asyncua.ua.ua_binary._DateTime object>#

Guid = <asyncua.ua.ua_binary._Guid object>#

Null = <asyncua.ua.ua_binary._Null object>#

String = <asyncua.ua.ua_binary._String object>#

class asyncua.ua.ua_binary.Primitives1#

Bases: object

Boolean = <asyncua.ua.ua_binary._Primitive1 object>#

Byte = <asyncua.ua.ua_binary._Primitive1 object>#

Char = <asyncua.ua.ua_binary._Primitive1 object>#

Double = <asyncua.ua.ua_binary._Primitive1 object>#

Float = <asyncua.ua.ua_binary._Primitive1 object>#

Int16 = <asyncua.ua.ua_binary._Primitive1 object>#

Int32 = <asyncua.ua.ua_binary._Primitive1 object>#

Int64 = <asyncua.ua.ua_binary._Primitive1 object>#

SByte = <asyncua.ua.ua_binary._Primitive1 object>#

UInt16 = <asyncua.ua.ua_binary._Primitive1 object>#

UInt32 = <asyncua.ua.ua_binary._Primitive1 object>#

UInt64 = <asyncua.ua.ua_binary._Primitive1 object>#

asyncua.ua.ua_binary.create_dataclass_serializer(dataclazz)#: Given a dataclass, return a function that serializes instances of this dataclass

asyncua.ua.ua_binary.create_enum_deserializer(uatype)#

asyncua.ua.ua_binary.create_enum_serializer(uatype)#

asyncua.ua.ua_binary.create_list_serializer(uatype, recursive: bool = False) → Callable[[Sequence[Any] | None], bytes]#: Given a type, return a function that takes a list of instances of that type and serializes it.

asyncua.ua.ua_binary.create_type_serializer(uatype)#: Create a binary serialization function for the given UA type

asyncua.ua.ua_binary.create_uatype_array_serializer(vtype)#

asyncua.ua.ua_binary.create_uatype_serializer(vtype)#

asyncua.ua.ua_binary.extensionobject_from_binary(data)#: Convert binary-coded ExtensionObject to a Python object. Returns an object, or None if TypeId is zero

asyncua.ua.ua_binary.extensionobject_to_binary(obj)#: Convert Python object to binary-coded ExtensionObject. If obj is None, convert to empty ExtensionObject (TypeId=0, no Body). Returns a binary string

asyncua.ua.ua_binary.field_serializer(ftype, dataclazz) → Callable[[Any], bytes]#

asyncua.ua.ua_binary.from_binary(uatype, data)#: unpack data given an uatype as a string or a python dataclass using ua types

asyncua.ua.ua_binary.header_from_binary(data)#

asyncua.ua.ua_binary.header_to_binary(hdr)#

asyncua.ua.ua_binary.list_to_binary(uatype, val)#

asyncua.ua.ua_binary.nodeid_from_binary(data)#

asyncua.ua.ua_binary.nodeid_to_binary(nodeid)#

asyncua.ua.ua_binary.pack_uatype(vtype, value)#

asyncua.ua.ua_binary.pack_uatype_array(vtype, array)#

asyncua.ua.ua_binary.set_bit(data: int, offset: int) → int#

asyncua.ua.ua_binary.struct_from_binary(objtype: Type[T] | str, data: IO) → T#: unpack an ua struct. Arguments are an objtype as Python dataclass or string

asyncua.ua.ua_binary.struct_to_binary(obj)#

asyncua.ua.ua_binary.test_bit(data: int, offset: int) → int#

asyncua.ua.ua_binary.to_binary(uatype, val)#

asyncua.ua.ua_binary.uatcp_to_binary(message_type, message)#: Convert OPC UA TCP message (see OPC UA specs Part 6, 7.1) to binary. The only supported types are Hello, Acknowledge and ErrorMessage

asyncua.ua.ua_binary.unpack_uatype(vtype, data)#

asyncua.ua.ua_binary.unpack_uatype_array(vtype, data)#

asyncua.ua.ua_binary.unset_bit(data: int, offset: int) → int#

asyncua.ua.ua_binary.variant_from_binary(data)#

asyncua.ua.ua_binary.variant_to_binary(var)#

asyncua.ua.uaprotocol_auto module#

Note

This module is very large and currently excluded from the API documentation

asyncua.ua.uaprotocol_hand module#

class asyncua.ua.uaprotocol_hand.Acknowledge(ProtocolVersion: asyncua.ua.uatypes.UInt32 = 0, ReceiveBufferSize: asyncua.ua.uatypes.UInt32 = 65536, SendBufferSize: asyncua.ua.uatypes.UInt32 = 65536, MaxMessageSize: asyncua.ua.uatypes.UInt32 = 0, MaxChunkCount: asyncua.ua.uatypes.UInt32 = 0)#

Bases: object

MaxChunkCount: UInt32 = 0#

MaxMessageSize: UInt32 = 0#

ProtocolVersion: UInt32 = 0#

ReceiveBufferSize: UInt32 = 65536#

SendBufferSize: UInt32 = 65536#

class asyncua.ua.uaprotocol_hand.Argument(Name: asyncua.ua.uatypes.String = None, DataType: asyncua.ua.uatypes.NodeId = <factory>, ValueRank: asyncua.ua.uatypes.Int32 = -1, ArrayDimensions: List[asyncua.ua.uatypes.UInt32] = <factory>, Description: asyncua.ua.uatypes.LocalizedText = <factory>)#

Bases: Argument

ValueRank: Int32 = -1#

class asyncua.ua.uaprotocol_hand.AsymmetricAlgorithmHeader(SecurityPolicyURI: asyncua.ua.uatypes.String = 'http://opcfoundation.org/UA/SecurityPolicy#None', SenderCertificate: asyncua.ua.uatypes.ByteString = None, ReceiverCertificateThumbPrint: asyncua.ua.uatypes.ByteString = None)#

Bases: object

ReceiverCertificateThumbPrint: ByteString = None#

SecurityPolicyURI: String = 'http://opcfoundation.org/UA/SecurityPolicy#None'#

SenderCertificate: ByteString = None#

class asyncua.ua.uaprotocol_hand.ChunkType(Invalid: bytes = b'0', Single: bytes = b'F', Intermediate: bytes = b'C', Abort: bytes = b'A')#

Bases: object

Abort: bytes = b'A'#

Intermediate: bytes = b'C'#

Invalid: bytes = b'0'#

Single: bytes = b'F'#

class asyncua.ua.uaprotocol_hand.CryptographyNone#

Bases: object

Base class for symmetric/asymmetric cryptography

decrypt(data)#

encrypt(data)#

encrypted_block_size()#: Size of encrypted text block for block cipher.

min_padding_size()#

padding(size)#: Create padding for a block of given size. plain_size = size + len(padding) + signature_size() plain_size = N * plain_block_size()

plain_block_size()#: Size of plain text block for block cipher.

remove_padding(data)#

signature(data)#

signature_size()#

verify(data, signature)#: Verify signature and raise exception if signature is invalid

vsignature_size()#

class asyncua.ua.uaprotocol_hand.DataTypeAttributes(SpecifiedAttributes: asyncua.ua.uatypes.UInt32 = 0, DisplayName: asyncua.ua.uatypes.LocalizedText = <factory>, Description: asyncua.ua.uatypes.LocalizedText = <factory>, WriteMask: asyncua.ua.uatypes.UInt32 = 0, UserWriteMask: asyncua.ua.uatypes.UInt32 = 0, IsAbstract: asyncua.ua.uatypes.Boolean = True, DataTypeDefinition: asyncua.ua.uatypes.ExtensionObject = <factory>)#

Bases: DataTypeAttributes

DataTypeDefinition: ExtensionObject#

class asyncua.ua.uaprotocol_hand.ErrorMessage(Error: asyncua.ua.uatypes.StatusCode = StatusCode(value=0), Reason: asyncua.ua.uatypes.String = '')#

Bases: object

Error: StatusCode = StatusCode(value=0)#

Reason: String = ''#

class asyncua.ua.uaprotocol_hand.Header(MessageType: None = None, ChunkType: None = None, ChannelId: int = 0)#

Bases: object

ChannelId: int = 0#

ChunkType: None = None#

MessageType: None = None#

add_size(size)#

body_size = 0#

header_size = 8#

static max_size()#

packet_size = 0#

class asyncua.ua.uaprotocol_hand.Hello(ProtocolVersion: asyncua.ua.uatypes.UInt32 = 0, ReceiveBufferSize: asyncua.ua.uatypes.UInt32 = 2147483647, SendBufferSize: asyncua.ua.uatypes.UInt32 = 2147483647, MaxMessageSize: asyncua.ua.uatypes.UInt32 = 2147483647, MaxChunkCount: asyncua.ua.uatypes.UInt32 = 2147483647, EndpointUrl: asyncua.ua.uatypes.String = '')#

Bases: object

EndpointUrl: String = ''#

MaxChunkCount: UInt32 = 2147483647#

MaxMessageSize: UInt32 = 2147483647#

ProtocolVersion: UInt32 = 0#

ReceiveBufferSize: UInt32 = 2147483647#

SendBufferSize: UInt32 = 2147483647#

class asyncua.ua.uaprotocol_hand.Message(chunks)#

Bases: object

SecurityHeader()#

SequenceHeader()#

body()#

request_id()#

class asyncua.ua.uaprotocol_hand.MessageType(Invalid: bytes = b'INV', Hello: bytes = b'HEL', Acknowledge: bytes = b'ACK', Error: bytes = b'ERR', SecureOpen: bytes = b'OPN', SecureClose: bytes = b'CLO', SecureMessage: bytes = b'MSG')#

Bases: object

Acknowledge: bytes = b'ACK'#

Error: bytes = b'ERR'#

Hello: bytes = b'HEL'#

Invalid: bytes = b'INV'#

SecureClose: bytes = b'CLO'#

SecureMessage: bytes = b'MSG'#

SecureOpen: bytes = b'OPN'#

class asyncua.ua.uaprotocol_hand.MethodAttributes(SpecifiedAttributes: asyncua.ua.uatypes.UInt32 = 0, DisplayName: asyncua.ua.uatypes.LocalizedText = <factory>, Description: asyncua.ua.uatypes.LocalizedText = <factory>, WriteMask: asyncua.ua.uatypes.UInt32 = 0, UserWriteMask: asyncua.ua.uatypes.UInt32 = 0, Executable: asyncua.ua.uatypes.Boolean = True, UserExecutable: asyncua.ua.uatypes.Boolean = True)#: Bases: MethodAttributes

class asyncua.ua.uaprotocol_hand.ObjectAttributes(SpecifiedAttributes: asyncua.ua.uatypes.UInt32 = 0, DisplayName: asyncua.ua.uatypes.LocalizedText = <factory>, Description: asyncua.ua.uatypes.LocalizedText = <factory>, WriteMask: asyncua.ua.uatypes.UInt32 = 0, UserWriteMask: asyncua.ua.uatypes.UInt32 = 0, EventNotifier: asyncua.ua.uatypes.Byte = 0)#: Bases: ObjectAttributes

class asyncua.ua.uaprotocol_hand.ObjectTypeAttributes(SpecifiedAttributes: asyncua.ua.uatypes.UInt32 = 0, DisplayName: asyncua.ua.uatypes.LocalizedText = <factory>, Description: asyncua.ua.uatypes.LocalizedText = <factory>, WriteMask: asyncua.ua.uatypes.UInt32 = 0, UserWriteMask: asyncua.ua.uatypes.UInt32 = 0, IsAbstract: asyncua.ua.uatypes.Boolean = True)#: Bases: ObjectTypeAttributes

class asyncua.ua.uaprotocol_hand.ReferenceTypeAttributes(SpecifiedAttributes: asyncua.ua.uatypes.UInt32 = 0, DisplayName: asyncua.ua.uatypes.LocalizedText = <factory>, Description: asyncua.ua.uatypes.LocalizedText = <factory>, WriteMask: asyncua.ua.uatypes.UInt32 = 0, UserWriteMask: asyncua.ua.uatypes.UInt32 = 0, IsAbstract: asyncua.ua.uatypes.Boolean = True, Symmetric: asyncua.ua.uatypes.Boolean = True, InverseName: asyncua.ua.uatypes.LocalizedText = <factory>)#: Bases: ReferenceTypeAttributes

class asyncua.ua.uaprotocol_hand.SecurityPolicy(permissions=None)#

Bases: object

Base class for security policy

AsymmetricSignatureURI: str = ''#

URI = 'http://opcfoundation.org/UA/SecurityPolicy#None'#

make_local_symmetric_key(secret, seed)#

make_remote_symmetric_key(secret, seed, lifetime)#

secure_channel_nonce_length: int = 0#

signature_key_size: int = 0#

symmetric_key_size: int = 0#

class asyncua.ua.uaprotocol_hand.SecurityPolicyFactory(cls=<class 'asyncua.ua.uaprotocol_hand.SecurityPolicy'>, mode=MessageSecurityMode.None_, certificate=None, private_key=None, permission_ruleset=None)#

Bases: object

Helper class for creating server-side SecurityPolicy. Server has one certificate and private key, but needs a separate SecurityPolicy for every client and client’s certificate

create(peer_certificate)#

matches(uri, mode=None)#

class asyncua.ua.uaprotocol_hand.SequenceHeader(SequenceNumber: asyncua.ua.uatypes.UInt32 = None, RequestId: asyncua.ua.uatypes.UInt32 = None)#

Bases: object

RequestId: UInt32 = None#

SequenceNumber: UInt32 = None#

static max_size()#

class asyncua.ua.uaprotocol_hand.SymmetricAlgorithmHeader(TokenId: asyncua.ua.uatypes.UInt32 = 0)#

Bases: object

TokenId: UInt32 = 0#

static max_size()#

class asyncua.ua.uaprotocol_hand.VariableAttributes(SpecifiedAttributes: asyncua.ua.uatypes.UInt32 = 4002419, DisplayName: asyncua.ua.uatypes.LocalizedText = <factory>, Description: asyncua.ua.uatypes.LocalizedText = <factory>, WriteMask: asyncua.ua.uatypes.UInt32 = 0, UserWriteMask: asyncua.ua.uatypes.UInt32 = 0, Value: asyncua.ua.uatypes.Variant = <factory>, DataType: asyncua.ua.uatypes.NodeId = <factory>, ValueRank: asyncua.ua.uatypes.Int32 = 0, ArrayDimensions: List[asyncua.ua.uatypes.UInt32] = None, AccessLevel: asyncua.ua.uatypes.Byte = 1, UserAccessLevel: asyncua.ua.uatypes.Byte = 1, MinimumSamplingInterval: asyncua.ua.uatypes.Double = 0, Historizing: asyncua.ua.uatypes.Boolean = True)#

Bases: VariableAttributes

AccessLevel: Byte = 1#

ArrayDimensions: List[UInt32] = None#

Historizing: Boolean = True#

SpecifiedAttributes: UInt32 = 4002419#

UserAccessLevel: Byte = 1#

class asyncua.ua.uaprotocol_hand.VariableTypeAttributes(SpecifiedAttributes: asyncua.ua.uatypes.UInt32 = 0, DisplayName: asyncua.ua.uatypes.LocalizedText = <factory>, Description: asyncua.ua.uatypes.LocalizedText = <factory>, WriteMask: asyncua.ua.uatypes.UInt32 = 0, UserWriteMask: asyncua.ua.uatypes.UInt32 = 0, Value: asyncua.ua.uatypes.Variant = <factory>, DataType: asyncua.ua.uatypes.NodeId = <factory>, ValueRank: asyncua.ua.uatypes.Int32 = 0, ArrayDimensions: List[asyncua.ua.uatypes.UInt32] = <factory>, IsAbstract: asyncua.ua.uatypes.Boolean = True)#: Bases: VariableTypeAttributes

class asyncua.ua.uaprotocol_hand.ViewAttributes(SpecifiedAttributes: asyncua.ua.uatypes.UInt32 = 0, DisplayName: asyncua.ua.uatypes.LocalizedText = <factory>, Description: asyncua.ua.uatypes.LocalizedText = <factory>, WriteMask: asyncua.ua.uatypes.UInt32 = 0, UserWriteMask: asyncua.ua.uatypes.UInt32 = 0, ContainsNoLoops: asyncua.ua.uatypes.Boolean = True, EventNotifier: asyncua.ua.uatypes.Byte = 0)#: Bases: ViewAttributes

class asyncua.ua.uaprotocol_hand.XmlElement(Value: String = '')#

Bases: object

An XML element encoded as a UTF-8 string. :ivar Value: :vartype Value: String

Value: String = ''#

asyncua.ua.uatypes module#

implement ua datatypes

class asyncua.ua.uatypes.AccessLevel(value)#

Bases: _MaskEnum

Bit index to indicate what the access level is.

Spec Part 3, appears multiple times, e.g. paragraph 5.6.2 Variable NodeClass

CurrentRead = 0#

CurrentWrite = 1#

HistoryRead = 2#

HistoryWrite = 3#

SemanticChange = 4#

StatusWrite = 5#

TimestampWrite = 6#

class asyncua.ua.uatypes.Boolean#: Bases: object

class asyncua.ua.uatypes.Byte#: Bases: int

class asyncua.ua.uatypes.ByteString#: Bases: bytes

class asyncua.ua.uatypes.ByteStringNodeId(Identifier: asyncua.ua.uatypes.Int32 | asyncua.ua.uatypes.String | asyncua.ua.uatypes.Guid | asyncua.ua.uatypes.ByteString = 0, NamespaceIndex: asyncua.ua.uatypes.Int16 = 0, NodeIdType: asyncua.ua.uatypes.NodeIdType = None)#: Bases: NodeId

class asyncua.ua.uatypes.Bytes#: Bases: bytes

class asyncua.ua.uatypes.CharArray#: Bases: str

class asyncua.ua.uatypes.DataValue(Value: ~asyncua.ua.uatypes.Variant | None = None, StatusCode_: ~asyncua.ua.uatypes.StatusCode | None = <factory>, SourceTimestamp: ~asyncua.ua.uatypes.DateTime | None = None, ServerTimestamp: ~asyncua.ua.uatypes.DateTime | None = None, SourcePicoseconds: ~asyncua.ua.uatypes.UInt16 | None = None, ServerPicoseconds: ~asyncua.ua.uatypes.UInt16 | None = None)#

Bases: object

A value with an associated timestamp, and quality. Automatically generated from xml , copied and modified here to fix errors in xml spec

Variables:

Value (Variant) –
StatusCode (StatusCode) –
SourceTimestamp (datetime) –
SourcePicoSeconds (int) –
ServerTimestamp (datetime) –
ServerPicoseconds (int) –

Encoding: Byte = 0#

ServerPicoseconds: UInt16 | None = None#

ServerTimestamp: DateTime | None = None#

SourcePicoseconds: UInt16 | None = None#

SourceTimestamp: DateTime | None = None#

property StatusCode#

StatusCode_: StatusCode | None#

Value: Variant | None = None#

data_type = NodeId(Identifier=25, NamespaceIndex=0, NodeIdType=<NodeIdType.TwoByte: 0>)#

class asyncua.ua.uatypes.DateTime#: Bases: datetime

Bases: object

A recursive structure containing diagnostic information associated with a status code.

Variables:

Encoding (Byte) –
SymbolicId (Int32) –
NamespaceURI (Int32) –
Locale (Int32) –
LocalizedText (Int32) –
AdditionalInfo (String) –
InnerStatusCode (StatusCode) –
InnerDiagnosticInfo (DiagnosticInfo) –

AdditionalInfo: String | None = None#

Encoding: Byte = 0#

InnerDiagnosticInfo: ExtensionObject | None = None#

InnerStatusCode: StatusCode | None = None#

Locale: Int32 | None = None#

LocalizedText: Int32 | None = None#

NamespaceURI: Int32 | None = None#

SymbolicId: Int32 | None = None#

data_type = NodeId(Identifier=25, NamespaceIndex=0, NodeIdType=<NodeIdType.TwoByte: 0>)#

class asyncua.ua.uatypes.Double(x=0, /)#: Bases: float

class asyncua.ua.uatypes.EventNotifier(value)#

Bases: _MaskEnum

Bit index to indicate how a node can be used for events.

Spec Part 3, appears multiple times, e.g. Paragraph 5.4 View NodeClass

HistoryRead = 2#

HistoryWrite = 3#

SubscribeToEvents = 0#

class asyncua.ua.uatypes.ExpandedNodeId(Identifier: asyncua.ua.uatypes.Int32 | asyncua.ua.uatypes.String | asyncua.ua.uatypes.Guid | asyncua.ua.uatypes.ByteString = 0, NamespaceIndex: asyncua.ua.uatypes.Int16 = 0, NodeIdType: asyncua.ua.uatypes.NodeIdType = None, NamespaceUri: asyncua.ua.uatypes.String | None = None, ServerIndex: asyncua.ua.uatypes.Int32 = 0)#

Bases: NodeId

NamespaceUri: String | None = None#

ServerIndex: Int32 = 0#

to_string()#

class asyncua.ua.uatypes.ExtensionObject(TypeId: ~asyncua.ua.uatypes.NodeId = NodeId(Identifier=0, NamespaceIndex=0, NodeIdType=<NodeIdType.TwoByte: 0>), Body: ~asyncua.ua.uatypes.ByteString | None = None)#

Bases: object

Any UA object packed as an ExtensionObject

Variables:

TypeId (NodeId) –
Body (bytes) –

Body: ByteString | None = None#

Encoding: Byte = 0#

TypeId: NodeId = NodeId(Identifier=0, NamespaceIndex=0, NodeIdType=<NodeIdType.TwoByte: 0>)#

class asyncua.ua.uatypes.Float(x=0, /)#: Bases: float

class asyncua.ua.uatypes.FourByteNodeId(Identifier: asyncua.ua.uatypes.Int32 | asyncua.ua.uatypes.String | asyncua.ua.uatypes.Guid | asyncua.ua.uatypes.ByteString = 0, NamespaceIndex: asyncua.ua.uatypes.Int16 = 0, NodeIdType: asyncua.ua.uatypes.NodeIdType = None)#: Bases: NodeId

class asyncua.ua.uatypes.Guid(hex=None, bytes=None, bytes_le=None, fields=None, int=None, version=None, *, is_safe=SafeUUID.unknown)#

Bases: UUID

int#

is_safe#

class asyncua.ua.uatypes.GuidNodeId(Identifier: asyncua.ua.uatypes.Int32 | asyncua.ua.uatypes.String | asyncua.ua.uatypes.Guid | asyncua.ua.uatypes.ByteString = 0, NamespaceIndex: asyncua.ua.uatypes.Int16 = 0, NodeIdType: asyncua.ua.uatypes.NodeIdType = None)#: Bases: NodeId

class asyncua.ua.uatypes.Int16#: Bases: int

class asyncua.ua.uatypes.Int32#: Bases: int

class asyncua.ua.uatypes.Int64#: Bases: int

asyncua.ua.uatypes.Integer#: alias of Variant

class asyncua.ua.uatypes.LocalizedText(Text=None, Locale=None)#

Bases: object

A string qualified with a namespace index.

Encoding: Byte = 0#

Locale: String | None = None#

Text: String | None = None#

static from_string(string)#

to_string()#

class asyncua.ua.uatypes.NodeId(Identifier: Int32 | String | Guid | ByteString = 0, NamespaceIndex: Int16 = 0, NodeIdType: NodeIdType | None = None)#

Bases: object

NodeId Object

Args:: identifier: The identifier might be an int, a string, bytes or a Guid namespaceidx(int): The index of the namespace nodeidtype(NodeIdType): The type of the nodeid if it cannot be guessed, or you want something special like twobyte nodeid or fourbytenodeid

Variables:

Identifier (NodeId) –
NamespaceIndex (Int) –
NamespaceUri (String) –
ServerIndex (Int) –

Identifier: Int32 | String | Guid | ByteString = 0#

NamespaceIndex: Int16 = 0#

NodeIdType: NodeIdType = None#

check_identifier_type_compatibility()#: Check whether the given identifier can be interpreted as the given node identifier type.

static from_string(string)#

has_null_identifier()#

is_null()#

to_binary()#

to_string()#

class asyncua.ua.uatypes.NodeIdType(value)#

Bases: IntEnum

An enumeration.

ByteString = 5#

FourByte = 1#

Guid = 4#

Numeric = 2#

String = 3#

TwoByte = 0#

class asyncua.ua.uatypes.Null#: Bases: object

class asyncua.ua.uatypes.NumericNodeId(Identifier: asyncua.ua.uatypes.Int32 | asyncua.ua.uatypes.String | asyncua.ua.uatypes.Guid | asyncua.ua.uatypes.ByteString = 0, NamespaceIndex: asyncua.ua.uatypes.Int16 = 0, NodeIdType: asyncua.ua.uatypes.NodeIdType = None)#: Bases: NodeId

class asyncua.ua.uatypes.QualifiedName(Name=None, NamespaceIndex=0)#

Bases: object

A string qualified with a namespace index.

Name: String = ''#

NamespaceIndex: UInt16 = 0#

static from_string(string, default_idx=0)#

to_string()#

class asyncua.ua.uatypes.RelativePath(Elements: ~typing.List[~asyncua.ua.uatypes.RelativePathElement] = <factory>)#

Bases: object

https://reference.opcfoundation.org/v105/Core/docs/Part4/7.31

Variables:: Elements (RelativePathElement) –

Elements: List[RelativePathElement]#

data_type = NodeId(Identifier=540, NamespaceIndex=0, NodeIdType=<NodeIdType.FourByte: 1>)#

static from_string(string: str)#

to_string() → str#

class asyncua.ua.uatypes.RelativePathElement(ReferenceTypeId: ~asyncua.ua.uatypes.NodeId = <factory>, IsInverse: ~asyncua.ua.uatypes.Boolean = True, IncludeSubtypes: ~asyncua.ua.uatypes.Boolean = True, TargetName: ~asyncua.ua.uatypes.QualifiedName = <factory>)#

Bases: object

https://reference.opcfoundation.org/v105/Core/docs/Part4/7.31

Variables:

ReferenceTypeId (NodeId) –
IsInverse (Boolean) –
IncludeSubtypes (Boolean) –
TargetName (QualifiedName) –

IncludeSubtypes: Boolean = True#

IsInverse: Boolean = True#

ReferenceTypeId: NodeId#

TargetName: QualifiedName#

data_type = NodeId(Identifier=537, NamespaceIndex=0, NodeIdType=<NodeIdType.FourByte: 1>)#

class asyncua.ua.uatypes.SByte#: Bases: int

class asyncua.ua.uatypes.SecurityPolicyType(value)#

Bases: Enum

The supported types of SecurityPolicy.

“None” “Basic128Rsa15_Sign” “Basic128Rsa15_SignAndEncrypt” “Basic256_Sign” “Basic256_SignAndEncrypt” “Basic256Sha256_Sign” “Basic256Sha256_SignAndEncrypt” “Aes128_Sha256_RsaOaep_Sign” “Aes128_Sha256_RsaOaep_SignAndEncrypt” “Aes256_Sha256_RsaPss_Sign” “Aes256_Sha256_RsaPss_SignAndEncrypt”

Aes128Sha256RsaOaep_Sign = 7#

Aes128Sha256RsaOaep_SignAndEncrypt = 8#

Aes256Sha256RsaPss_Sign = 9#

Aes256Sha256RsaPss_SignAndEncrypt = 10#

Basic128Rsa15_Sign = 1#

Basic128Rsa15_SignAndEncrypt = 2#

Basic256Sha256_Sign = 5#

Basic256Sha256_SignAndEncrypt = 6#

Basic256_Sign = 3#

Basic256_SignAndEncrypt = 4#

NoSecurity = 0#

class asyncua.ua.uatypes.StatusCode(value: UInt32 = 0)#

Bases: object

Variables:

value (int) –
name (string) –
doc (string) –

check()#: Raises an exception if the status code is anything else than 0 (good).

property doc#

is_bad()#

https://reference.opcfoundation.org/v104/Core/docs/Part4/7.34.1/ 11 Reserved for future use. All Clients should treat a StatusCode with this severity as “Bad”.

return True if status is Bad (10) or (11).

is_good()#: return True if status is Good (00).

is_uncertain()#: return True if status is Uncertain (01).

property name#

value: UInt32 = 0#

class asyncua.ua.uatypes.String#: Bases: str

class asyncua.ua.uatypes.StringNodeId(Identifier: asyncua.ua.uatypes.Int32 | asyncua.ua.uatypes.String | asyncua.ua.uatypes.Guid | asyncua.ua.uatypes.ByteString = 0, NamespaceIndex: asyncua.ua.uatypes.Int16 = 0, NodeIdType: asyncua.ua.uatypes.NodeIdType = None)#: Bases: NodeId

class asyncua.ua.uatypes.TwoByteNodeId(Identifier: asyncua.ua.uatypes.Int32 | asyncua.ua.uatypes.String | asyncua.ua.uatypes.Guid | asyncua.ua.uatypes.ByteString = 0, NamespaceIndex: asyncua.ua.uatypes.Int16 = 0, NodeIdType: asyncua.ua.uatypes.NodeIdType = None)#: Bases: NodeId

class asyncua.ua.uatypes.UInt16#: Bases: int

class asyncua.ua.uatypes.UInt32#: Bases: int

class asyncua.ua.uatypes.UInt64#: Bases: int

asyncua.ua.uatypes.UInteger#: alias of Variant

class asyncua.ua.uatypes.UaUnion#

Bases: object

class to identify unions

class asyncua.ua.uatypes.ValueRank(value)#

Bases: IntEnum

Defines dimensions of a variable. This enum does not support all cases since ValueRank support any n>0 but since it is an IntEnum it can be replaced by a normal int

Any = -2#

FourDimensions = 4#

OneDimension = 1#

OneOrMoreDimensions = 0#

Scalar = -1#

ScalarOrOneDimension = -3#

ThreeDimensions = 3#

TwoDimensions = 2#

class asyncua.ua.uatypes.Variant(Value: Any | None = None, VariantType: VariantType | None = None, Dimensions: List[Int32] | None = None, is_array: bool | None = None)#

Bases: object

Create an OPC-UA Variant object. if no argument a Null Variant is created. if not variant type is given, attempts to guess type from python type if a variant is given as value, the new objects becomes a copy of the argument

Variables:

Value (Any supported type) –
VariantType (VariantType) –
Dimension –
is_array (If the variant is an array. Always True if Dimension is specified) –

Dimensions: List[Int32] | None = None#

Value: Any = None#

VariantType: VariantType = None#

is_array: bool | None = None#

class asyncua.ua.uatypes.VariantType(value)#

Bases: Enum

The possible types of a variant.

Variables:

Null –
Boolean –
SByte –
Byte –
Int16 –
UInt16 –
Int32 –
UInt32 –
Int64 –
UInt64 –
Float –
Double –
String –
DateTime –
Guid –
ByteString –
XmlElement –
NodeId –
ExpandedNodeId –
StatusCode –
QualifiedName –
LocalizedText –
ExtensionObject –
DataValue –
Variant –
DiagnosticInfo –

Boolean = 1#

Byte = 3#

ByteString = 15#

DataValue = 23#

DateTime = 13#

DiagnosticInfo = 25#

Double = 11#

ExpandedNodeId = 18#

ExtensionObject = 22#

Float = 10#

Guid = 14#

Int16 = 4#

Int32 = 6#

Int64 = 8#

LocalizedText = 21#

NodeId = 17#

Null = 0#

QualifiedName = 20#

SByte = 2#

StatusCode = 19#

String = 12#

UInt16 = 5#

UInt32 = 7#

UInt64 = 9#

Variant = 24#

XmlElement = 16#

class asyncua.ua.uatypes.VariantTypeCustom(val)#

Bases: object

Looks like sometime we get variant with other values than those defined in VariantType. FIXME: We should not need this class, as far as I understand the spec variants can only be of VariantType

class asyncua.ua.uatypes.WriteMask(value)#

Bases: _MaskEnum

Bit index to indicate which attribute of a node is writable

Spec Part 3, Paragraph 5.2.7 WriteMask

AccessLevel = 0#

ArrayDimensions = 1#

BrowseName = 2#

ContainsNoLoops = 3#

DataType = 4#

Description = 5#

DisplayName = 6#

EventNotifier = 7#

Executable = 8#

Historizing = 9#

InverseName = 10#

IsAbstract = 11#

MinimumSamplingInterval = 12#

NodeClass = 13#

NodeId = 14#

Symmetric = 15#

UserAccessLevel = 16#

UserExecutable = 17#

UserWriteMask = 18#

ValueForVariableType = 21#

ValueRank = 19#

WriteMask = 20#

asyncua.ua.uatypes.datatype_to_varianttype(int_type)#: Takes a NodeId or int and return a VariantType This is only supported if int_type < 63 due to VariantType encoding At low level we do not have access to address space thus decoding is limited a better version of this method can be find in ua_utils.py

asyncua.ua.uatypes.datetime_to_win_epoch(dt: datetime)#

asyncua.ua.uatypes.flatten(mylist)#

asyncua.ua.uatypes.flatten_and_get_shape(mylist)#

asyncua.ua.uatypes.get_default_value(vtype)#: Given a variant type return default value for this type

asyncua.ua.uatypes.get_extensionobject_class_type(typeid)#: Returns the registered class type for typid of an extension object

asyncua.ua.uatypes.get_shape(mylist)#

asyncua.ua.uatypes.get_win_epoch()#

asyncua.ua.uatypes.register_basetype(name, nodeid, class_type)#: Register a new alias of basetypes for automatic decoding and make them available in ua module

asyncua.ua.uatypes.register_enum(name, nodeid, class_type)#: Register a new enum for automatic decoding and make them available in ua module

asyncua.ua.uatypes.register_extension_object(name, encoding_nodeid, class_type, datatype_nodeid=None)#: Register a new extension object for automatic decoding and make them available in ua module

asyncua.ua.uatypes.type_allow_subclass(uatype)#

asyncua.ua.uatypes.type_from_allow_subtype(uatype)#

asyncua.ua.uatypes.type_from_list(uatype)#

asyncua.ua.uatypes.type_from_optional(uatype)#

asyncua.ua.uatypes.type_is_list(uatype)#

asyncua.ua.uatypes.type_is_union(uatype)#

asyncua.ua.uatypes.type_string_from_type(uatype)#

asyncua.ua.uatypes.types_from_union(uatype, origin=None)#

asyncua.ua.uatypes.types_or_list_from_union(uatype)#

asyncua.ua.uatypes.win_epoch_to_datetime(epch)#