opcua._FrozenClass.__setattr__() - Code Metrics - Inspection of "Merge pull request #78 from huazh/master" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed
Push — master ( abb445...558fdd )

by Olivier
created 2015-12-20 09:15 UTC
opcua._FrozenClass.setattr() A

↳ Parent: opcua._FrozenClass
Complexity

Conditions
Size

Total Lines
Duplication

Lines	0
Ratio	0 %
Code Coverage

Tests	3
CRAP Score	3.1406
Metric	Value
dl	0
loc	4
ccs	3
cts	4
cp	0.75
rs	10
cc	3
crap	3.1406
"""
implement ua datatypes
"""
import logging
from enum import Enum
from datetime import datetime, timedelta, tzinfo
from calendar import timegm
import sys
import os
if sys.version_info.major > 2:
    unicode = str

import uuid
import struct

import opcua.status_code as status_code
from opcua.object_ids import ObjectIds

logger = logging.getLogger('opcua.uaprotocol')

# types that will packed and unpacked directly using struct (string, bytes and datetime are handles as special cases
UaTypes = ("Boolean", "SByte", "Byte", "Int8", "UInt8", "Int16", "UInt16", "Int32", "UInt32", "Int64", "UInt64", "Float", "Double")


EPOCH_AS_FILETIME = 116444736000000000  # January 1, 1970 as MS file time
HUNDREDS_OF_NANOSECONDS = 10000000
FILETIME_EPOCH_AS_DATETIME = datetime(1601, 1, 1)


class UTC(tzinfo):

    """UTC"""

    def utcoffset(self, dt):
        return timedelta(0)

    def tzname(self, dt):
        return "UTC"

    def dst(self, dt):
        return timedelta(0)


# method copied from David Buxton <[email protected]> sample code
def datetime_to_win_epoch(dt):
    if (dt.tzinfo is None) or (dt.tzinfo.utcoffset(dt) is None):
        dt = dt.replace(tzinfo=UTC())
    ft = EPOCH_AS_FILETIME + (timegm(dt.timetuple()) * HUNDREDS_OF_NANOSECONDS)
    return ft + (dt.microsecond * 10)


def win_epoch_to_datetime(epch):
    return FILETIME_EPOCH_AS_DATETIME + timedelta(microseconds=epch // 10)


def build_array_format_py2(prefix, length, fmtchar):
    return prefix + str(length) + fmtchar


def build_array_format_py3(prefix, length, fmtchar):
    return prefix + str(length) + chr(fmtchar)


if sys.version_info.major < 3:
    build_array_format = build_array_format_py2
else:
    build_array_format = build_array_format_py3


def pack_uatype_array_primitive(st, value, length):
    if length == 1:
        return b'\x01\x00\x00\x00' + st.pack(value[0])
    else:
        return struct.pack(build_array_format("<i", length, st.format[1]), length, *value)


def pack_uatype_array(uatype, value):
    if value is None:
        return b'\xff\xff\xff\xff'
    length = len(value)
    if length == 0:
        return b'\x00\x00\x00\x00'
    if uatype in uatype2struct:
        return pack_uatype_array_primitive(uatype2struct[uatype], value, length)
    b = []
    b.append(uatype_Int32.pack(length))
    for val in value:
        b.append(pack_uatype(uatype, val))
    return b"".join(b)


def pack_uatype(uatype, value):
    if uatype in uatype2struct:
        return uatype2struct[uatype].pack(value)
    elif uatype == "Null":
        return b''
    elif uatype == "String":
        return pack_string(value)
    elif uatype in ("CharArray", "ByteString"):
        return pack_bytes(value)
    elif uatype == "DateTime":
        return pack_datetime(value)
    elif uatype == "ExtensionObject":
        # dependency loop: classes in uaprotocol_auto use Variant defined in this file,
        # but Variant can contain any object from uaprotocol_auto as ExtensionObject.
        # Using local import to avoid import loop
        from opcua.uaprotocol_auto import extensionobject_to_binary
        return extensionobject_to_binary(value)
    else:
        return value.to_binary()

uatype_Int8 = struct.Struct("<b")
uatype_SByte = uatype_Int8
uatype_Int16 = struct.Struct("<h")
uatype_Int32 = struct.Struct("<i")
uatype_Int64 = struct.Struct("<q")
uatype_UInt8 = struct.Struct("<B")
uatype_Char = uatype_UInt8
uatype_Byte = uatype_UInt8
uatype_UInt16 = struct.Struct("<H")
uatype_UInt32 = struct.Struct("<I")
uatype_UInt64 = struct.Struct("<Q")
uatype_Boolean = struct.Struct("<?")
uatype_Double = struct.Struct("<d")
uatype_Float = struct.Struct("<f")

uatype2struct = {
    "Int8": uatype_Int8,
    "SByte": uatype_SByte,
    "Int16": uatype_Int16,
    "Int32": uatype_Int32,
    "Int64": uatype_Int64,
    "UInt8": uatype_UInt8,
    "Char": uatype_Char,
    "Byte": uatype_Byte,
    "UInt16": uatype_UInt16,
    "UInt32": uatype_UInt32,
    "UInt64": uatype_UInt64,
    "Boolean": uatype_Boolean,
    "Double": uatype_Double,
    "Float": uatype_Float,
}


def unpack_uatype(uatype, data):
    if uatype in uatype2struct:
        st = uatype2struct[uatype]
        return st.unpack(data.read(st.size))[0]
    elif uatype == "String":
        return unpack_string(data)
    elif uatype in ("CharArray", "ByteString"):
        return unpack_bytes(data)
    elif uatype == "DateTime":
        return unpack_datetime(data)
    elif uatype == "ExtensionObject":
        # dependency loop: classes in uaprotocol_auto use Variant defined in this file,
        # but Variant can contain any object from uaprotocol_auto as ExtensionObject.
        # Using local import to avoid import loop
        from opcua.uaprotocol_auto import extensionobject_from_binary
        return extensionobject_from_binary(data)
    else:
        glbs = globals()
        if uatype in glbs:
            klass = glbs[uatype]
            if hasattr(klass, 'from_binary'):
                return klass.from_binary(data)
        raise Exception("can not unpack unknown uatype %s" % uatype)


def unpack_uatype_array(uatype, data):
    length = uatype_Int32.unpack(data.read(4))[0]
    if length == -1:
        return None
    elif length == 0:
        return []
    elif uatype in uatype2struct:
        st = uatype2struct[uatype]
        if length == 1:
            return list(st.unpack(data.read(st.size)))
        else:
            arrst = struct.Struct(build_array_format("<", length, st.format[1]))
            return list(arrst.unpack(data.read(arrst.size)))
    else:
        result = []
        for _ in range(0, length):
            result.append(unpack_uatype(uatype, data))
        return result


def pack_datetime(value):
    epch = datetime_to_win_epoch(value)
    return uatype_Int64.pack(epch)


def unpack_datetime(data):
    epch = uatype_Int64.unpack(data.read(8))[0]
    return win_epoch_to_datetime(epch)


def pack_string(string):
    if type(string) is unicode:
        string = string.encode('utf-8')
    length = len(string)
    if length == 0:
        return b'\xff\xff\xff\xff'
    return uatype_Int32.pack(length) + string

pack_bytes = pack_string


def unpack_bytes(data):
    length = uatype_Int32.unpack(data.read(4))[0]
    if length == -1:
        return b''
    return data.read(length)


def py3_unpack_string(data):
    b = unpack_bytes(data)
    return b.decode("utf-8")


if sys.version_info.major < 3:
    unpack_string = unpack_bytes
else:
    unpack_string = py3_unpack_string


def test_bit(data, offset):
    mask = 1 << offset
    return data & mask


def set_bit(data, offset):
    mask = 1 << offset
    return data | mask


class _FrozenClass(object):

    """
    make it impossible to add members to a class.
    This is a hack since I found out that most bugs are due to misspelling a variable in protocol
    """
    _freeze = False

    def __setattr__(self, key, value):
        if self._freeze and not hasattr(self, key):
            raise TypeError("Error adding member '{}' to class '{}', class is frozen, members are {}".format(key, self.__class__.__name__, self.__dict__.keys()))
        object.__setattr__(self, key, value)

if "PYOPCUA_NO_TYPO_CHECK" in os.environ:
    # typo check is cpu consuming, but it will make debug easy.
    # if typo check is not need (in production), please set env PYOPCUA_NO_TYPO_CHECK.
    # this will make all uatype class inherit from object intead of _FrozenClass
    # and skip the typo check.
    FrozenClass = object
else:
    FrozenClass = _FrozenClass


class Guid(FrozenClass):

    def __init__(self):
        self.uuid = uuid.uuid4()
        self._freeze = True

    def to_binary(self):
        return self.uuid.bytes

    @staticmethod
    def from_binary(data):
        g = Guid()
        g.uuid = uuid.UUID(bytes=data.read(16))
        return g

    def __eq__(self, other):
        return isinstance(other, Guid) and self.uuid == other.uuid


class StatusCode(FrozenClass):

    """
    :ivar value:
    :vartype value: int
    :ivar name:
    :vartype name: string
    :ivar doc:
    :vartype doc: string
    """

    def __init__(self, value=0):
        self.value = value
        self.name, self.doc = status_code.get_name_and_doc(value)
        self._freeze = True

    def to_binary(self):
        return uatype_UInt32.pack(self.value)

    @staticmethod
    def from_binary(data):
        val = uatype_UInt32.unpack(data.read(4))[0]
        sc = StatusCode(val)
        return sc

    def check(self):
        """
        raise en exception if status code is anything else than 0
        use is is_good() method if not exception is desired
        """
        if self.value != 0:
            raise Exception("{}({})".format(self.doc, self.name))

    def is_good(self):
        """
        return True if status is Good.
        """
        if self.value == 0:
            return True
        return False

    def __str__(self):
        return 'StatusCode({})'.format(self.name)
    __repr__ = __str__


class NodeIdType(object):
    TwoByte = 0
    FourByte = 1
    Numeric = 2
    String = 3
    Guid = 4
    ByteString = 5


class NodeId(FrozenClass):

    """
    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 cannor be guess or you want something special like twobyte nodeid or fourbytenodeid


    :ivar Identifier:
    :vartype Identifier: NodeId
    :ivar NamespaceIndex:
    :vartype NamespaceIndex: Int
    :ivar NamespaceUri:
    :vartype NamespaceUri: String
    :ivar ServerIndex:
    :vartype ServerIndex: Int
    """

    def __init__(self, identifier=None, namespaceidx=0, nodeidtype=None):
        self.Identifier = identifier
        self.NamespaceIndex = namespaceidx
        self.NodeIdType = nodeidtype
        self.NamespaceUri = ""
        self.ServerIndex = 0
        self._freeze = True
        if self.Identifier is None:
            self.Identifier = 0
            self.NodeIdType = NodeIdType.TwoByte
            return
        if self.NodeIdType is None:
            if isinstance(self.Identifier, int):
                self.NodeIdType = NodeIdType.Numeric
            elif isinstance(self.Identifier, str):
                self.NodeIdType = NodeIdType.String
            elif isinstance(self.Identifier, bytes):
                self.NodeIdType = NodeIdType.ByteString
            else:
                raise Exception("NodeId: Could not guess type of NodeId, set NodeIdType")

    def __key(self):
        if self.NodeIdType in (NodeIdType.TwoByte, NodeIdType.FourByte, NodeIdType.Numeric):  # twobyte, fourbyte and numeric may represent the same node
            return self.NamespaceIndex, self.Identifier
        else:
            return self.NodeIdType, self.NamespaceIndex, self.Identifier

    def __eq__(self, node):
        return isinstance(node, NodeId) and self.__key() == node.__key()

    def __hash__(self):
        return hash(self.__key())

    @staticmethod
    def from_string(string):
        l = string.split(";")
        identifier = None
        namespace = 0
        ntype = None
        srv = None
        nsu = None
        for el in l:
            if not el:
                continue
            k, v = el.split("=")
            k = k.strip()
            v = v.strip()
            if k == "ns":
                namespace = int(v)
            elif k == "i":
                ntype = NodeIdType.Numeric
                identifier = int(v)
            elif k == "s":
                ntype = NodeIdType.String
                identifier = v
            elif k == "g":
                ntype = NodeIdType.Guid
                identifier = v
            elif k == "b":
                ntype = NodeIdType.ByteString
                identifier = v
            elif k == "srv":
                srv = v
            elif k == "nsu":
                nsu = v
        if identifier is None:
            raise Exception("Could not parse nodeid string: " + string)
        nodeid = NodeId(identifier, namespace, ntype)
        nodeid.NamespaceUri = nsu
        nodeid.ServerIndex = srv
        return nodeid

    def to_string(self):
        string = ""
        if self.NamespaceIndex != 0:
            string += "ns={};".format(self.NamespaceIndex)
        ntype = None
        if self.NodeIdType == NodeIdType.Numeric:
            ntype = "i"
        elif self.NodeIdType == NodeIdType.String:
            ntype = "s"
        elif self.NodeIdType == NodeIdType.TwoByte:
            ntype = "i"
        elif self.NodeIdType == NodeIdType.FourByte:
            ntype = "i"
        elif self.NodeIdType == NodeIdType.Guid:
            ntype = "g"
        elif self.NodeIdType == NodeIdType.ByteString:
            ntype = "b"
        string += "{}={}".format(ntype, self.Identifier)
        if self.ServerIndex:
            string = "srv=" + str(self.ServerIndex) + string
        if self.NamespaceUri:
            string += "nsu={}".format(self.NamespaceUri)
        return string

    def __str__(self):
        return "NodeId({})".format(self.to_string())
    __repr__ = __str__

    def to_binary(self):
        if self.NodeIdType == NodeIdType.TwoByte:
            return struct.pack("<BB", self.NodeIdType, self.Identifier)
        elif self.NodeIdType == NodeIdType.FourByte:
            return struct.pack("<BBH", self.NodeIdType, self.NamespaceIndex, self.Identifier)
        elif self.NodeIdType == NodeIdType.Numeric:
            return struct.pack("<BHI", self.NodeIdType, self.NamespaceIndex, self.Identifier)
        elif self.NodeIdType == NodeIdType.String:
            return struct.pack("<BH", self.NodeIdType, self.NamespaceIndex) + \
                pack_string(self.Identifier)
        elif self.NodeIdType == NodeIdType.ByteString:
            return struct.pack("<BH", self.NodeIdType, self.NamespaceIndex) + \
                pack_bytes(self.Identifier)
        else:
            return struct.pack("<BH", self.NodeIdType, self.NamespaceIndex) + \
                self.Identifier.to_binary()

    @staticmethod
    def from_binary(data):
        nid = NodeId()
        encoding = ord(data.read(1))
        nid.NodeIdType = encoding & 0b00111111

        if nid.NodeIdType == NodeIdType.TwoByte:
            nid.Identifier = ord(data.read(1))
        elif nid.NodeIdType == NodeIdType.FourByte:
            nid.NamespaceIndex, nid.Identifier = struct.unpack("<BH", data.read(3))
        elif nid.NodeIdType == NodeIdType.Numeric:
            nid.NamespaceIndex, nid.Identifier = struct.unpack("<HI", data.read(6))
        elif nid.NodeIdType == NodeIdType.String:
            nid.NamespaceIndex = uatype_UInt16.unpack(data.read(2))[0]
            nid.Identifier = unpack_string(data)
        elif nid.NodeIdType == NodeIdType.ByteString:
            nid.NamespaceIndex = uatype_UInt16.unpack(data.read(2))[0]
            nid.Identifier = unpack_bytes(data)
        elif nid.NodeIdType == NodeIdType.Guid:
            nid.NamespaceIndex = uatype_UInt16.unpack(data.read(2))[0]
            nid.Identifier = Guid.from_binary(data)
        else:
            raise Exception("Unknown NodeId encoding: " + str(nid.NodeIdType))

        if test_bit(encoding, 6):
            nid.NamespaceUri = unpack_string(data)
        if test_bit(encoding, 7):
            nid.ServerIndex = uatype_UInt32.unpack(data.read(4))[0]

        return nid


class TwoByteNodeId(NodeId):

    def __init__(self, identifier):
        NodeId.__init__(self, identifier, 0, NodeIdType.TwoByte)


class FourByteNodeId(NodeId):

    def __init__(self, identifier, namespace=0):
        NodeId.__init__(self, identifier, namespace, NodeIdType.FourByte)


class NumericNodeId(NodeId):

    def __init__(self, identifier, namespace=0):
        NodeId.__init__(self, identifier, namespace, NodeIdType.Numeric)


class ByteStringNodeId(NodeId):

    def __init__(self, identifier, namespace=0):
        NodeId.__init__(self, identifier, namespace, NodeIdType.ByteString)


class GuidNodeId(NodeId):

    def __init__(self, identifier, namespace=0):
        NodeId.__init__(self, identifier, namespace, NodeIdType.Guid)


class StringNodeId(NodeId):

    def __init__(self, identifier, namespace=0):
        NodeId.__init__(self, identifier, namespace, NodeIdType.String)


ExpandedNodeId = NodeId


class QualifiedName(FrozenClass):

    '''
    A string qualified with a namespace index.
    '''

    def __init__(self, name="", namespaceidx=0):
        self.NamespaceIndex = namespaceidx
        self.Name = name
        self._freeze = True

    def to_string(self):
        return "{}:{}".format(self.NamespaceIndex, self.Name)

    @staticmethod
    def from_string(string):
        if ":" in string:
            idx, name = string.split(":", 1)
        else:
            idx = 0
            name = string
        return QualifiedName(name, int(idx))

    def to_binary(self):
        packet = []
        packet.append(uatype_UInt16.pack(self.NamespaceIndex))
        packet.append(pack_string(self.Name))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = QualifiedName()
        obj.NamespaceIndex = uatype_UInt16.unpack(data.read(2))[0]
        obj.Name = unpack_string(data)
        return obj

    def __eq__(self, bname):
        return isinstance(bname, QualifiedName) and self.Name == bname.Name and self.NamespaceIndex == bname.NamespaceIndex

    def __str__(self):
        return 'QualifiedName({}:{})'.format(self.NamespaceIndex, self.Name)

    __repr__ = __str__


class LocalizedText(FrozenClass):

    '''
    A string qualified with a namespace index.
    '''

    def __init__(self, text=""):
        self.Encoding = 0
        self.Text = text
        if type(self.Text) is unicode:
            self.Text = self.Text.encode('utf-8')
        if self.Text:
            self.Encoding |= (1 << 1)
        self.Locale = b''
        self._freeze = True

    def to_binary(self):
        packet = []
        if self.Locale:
            self.Encoding |= (1 << 0)
        if self.Text:
            self.Encoding |= (1 << 1)
        packet.append(uatype_UInt8.pack(self.Encoding))
        if self.Locale:
            packet.append(pack_bytes(self.Locale))
        if self.Text:
            packet.append(pack_bytes(self.Text))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        obj = LocalizedText()
        obj.Encoding = ord(data.read(1))
        if obj.Encoding & (1 << 0):
            obj.Locale = unpack_bytes(data)
        if obj.Encoding & (1 << 1):
            obj.Text = unpack_bytes(data)
        return obj

    def to_string(self):
        # FIXME: use local
        return self.Text.decode()

    def __str__(self):
        return 'LocalizedText(' + 'Encoding:' + str(self.Encoding) + ', ' + \
            'Locale:' + str(self.Locale) + ', ' + \
            'Text:' + str(self.Text) + ')'
    __repr__ = __str__

    def __eq__(self, other):
        if isinstance(other, LocalizedText) and self.Locale == other.Locale and self.Text == other.Text:
            return True
        return False


class VariantType(Enum):

    '''
    The possible types of a variant.

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



    '''
    Null = 0
    Boolean = 1
    SByte = 2
    Byte = 3
    Int16 = 4
    UInt16 = 5
    Int32 = 6
    UInt32 = 7
    Int64 = 8
    UInt64 = 9
    Float = 10
    Double = 11
    String = 12
    DateTime = 13
    Guid = 14
    ByteString = 15
    XmlElement = 16
    NodeId = 17
    ExpandedNodeId = 18
    StatusCode = 19
    QualifiedName = 20
    LocalizedText = 21
    ExtensionObject = 22
    DataValue = 23
    Variant = 24
    DiagnosticInfo = 25


class Variant(FrozenClass):

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

    :ivar Value:
    :vartype Value: Any supported type
    :ivar VariantType:
    :vartype VariantType: VariantType
    """

    def __init__(self, value=None, varianttype=None, encoding=0):
        self.Encoding = encoding
        self.Value = value
        self.VariantType = varianttype
        if isinstance(value, Variant):
            self.Value = value.Value
            self.VariantType = value.VariantType
        if self.VariantType is None:
            if type(self.Value) in (list, tuple):
                if len(self.Value) == 0:
                    raise Exception("could not guess UA variable type")
                self.VariantType = self._guess_type(self.Value[0])
            else:
                self.VariantType = self._guess_type(self.Value)
        self._freeze = True

    def __eq__(self, other):
        if isinstance(other, Variant) and self.VariantType == other.VariantType and self.Value == other.Value:
            return True
        return False

    def _guess_type(self, val):
        if val is None:
            return VariantType.Null
        elif isinstance(val, bool):
            return VariantType.Boolean
        elif isinstance(val, float):
            return VariantType.Double
        elif isinstance(val, int):
            return VariantType.Int64
        elif type(val) in (str, unicode):
            return VariantType.String
        elif isinstance(val, bytes):
            return VariantType.ByteString
        elif isinstance(val, datetime):
            return VariantType.DateTime
        else:
            if isinstance(val, object):
                try:
                    return getattr(VariantType, val.__class__.__name__)
                except AttributeError:
                    return VariantType.ExtensionObject
            else:
                raise Exception("Could not guess UA type of {} with type {}, specify UA type".format(val, type(val)))

    def __str__(self):
        return "Variant(val:{},type:{})".format(self.Value, self.VariantType)
    __repr__ = __str__

    def to_binary(self):
        b = []
        mask = self.Encoding & 0b01111111
        self.Encoding = (self.VariantType.value | mask)
        if type(self.Value) in (list, tuple):
            self.Encoding |= (1 << 7)
            b.append(uatype_UInt8.pack(self.Encoding))
            b.append(pack_uatype_array(self.VariantType.name, self.Value))
        else:
            b.append(uatype_UInt8.pack(self.Encoding))
            b.append(pack_uatype(self.VariantType.name, self.Value))
        return b"".join(b)

    @staticmethod
    def from_binary(data):
        encoding = ord(data.read(1))
        vtype = VariantType(encoding & 0b01111111)
        if vtype == VariantType.Null:
            return Variant(None, vtype, encoding)
        if encoding & (1 << 7):
            value = unpack_uatype_array(vtype.name, data)
        else:
            value = unpack_uatype(vtype.name, data)
        return Variant(value, vtype, encoding)


class DataValue(FrozenClass):

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

    :ivar Value:
    :vartype Value: Variant
    :ivar StatusCode:
    :vartype StatusCode: StatusCode
    :ivar SourceTimestamp:
    :vartype SourceTimestamp: datetime
    :ivar SourcePicoSeconds:
    :vartype SourcePicoSeconds: int
    :ivar ServerTimestamp:
    :vartype ServerTimestamp: datetime
    :ivar ServerPicoseconds:
    :vartype ServerPicoseconds: int

    '''

    def __init__(self, variant=None, status=None):
        self.Encoding = 0
        if not isinstance(variant, Variant):
            variant = Variant(variant)
        self.Value = variant
        if status is None:
            self.StatusCode = StatusCode()
        else:
            self.StatusCode = status
        self.SourceTimestamp = None  # DateTime()
        self.SourcePicoseconds = None
        self.ServerTimestamp = None  # DateTime()
        self.ServerPicoseconds = None
        self._freeze = True

    def to_binary(self):
        packet = []
        if self.Value:
            self.Encoding |= (1 << 0)
        if self.StatusCode:
            self.Encoding |= (1 << 1)
        if self.SourceTimestamp:
            self.Encoding |= (1 << 2)
        if self.ServerTimestamp:
            self.Encoding |= (1 << 3)
        if self.SourcePicoseconds:
            self.Encoding |= (1 << 4)
        if self.ServerPicoseconds:
            self.Encoding |= (1 << 5)
        packet.append(uatype_UInt8.pack(self.Encoding))
        if self.Value:
            packet.append(self.Value.to_binary())
        if self.StatusCode:
            packet.append(self.StatusCode.to_binary())
        if self.SourceTimestamp:
            packet.append(pack_datetime(self.SourceTimestamp))  # self.SourceTimestamp.to_binary())
        if self.ServerTimestamp:
            packet.append(pack_datetime(self.ServerTimestamp))  # self.ServerTimestamp.to_binary())
        if self.SourcePicoseconds:
            packet.append(uatype_UInt16.pack(self.SourcePicoseconds))
        if self.ServerPicoseconds:
            packet.append(uatype_UInt16.pack(self.ServerPicoseconds))
        return b''.join(packet)

    @staticmethod
    def from_binary(data):
        encoding = ord(data.read(1))
        if encoding & (1 << 0):
            value = Variant.from_binary(data)
        else:
            value = None
        if encoding & (1 << 1):
            status = StatusCode.from_binary(data)
        else:
            status = None
        obj = DataValue(value, status)
        obj.Encoding = encoding
        if obj.Encoding & (1 << 2):
            obj.SourceTimestamp = unpack_datetime(data)  # DateTime.from_binary(data)
        if obj.Encoding & (1 << 3):
            obj.ServerTimestamp = unpack_datetime(data)  # DateTime.from_binary(data)
        if obj.Encoding & (1 << 4):
            obj.SourcePicoseconds = uatype_UInt16.unpack(data.read(2))[0]
        if obj.Encoding & (1 << 5):
            obj.ServerPicoseconds = uatype_UInt16.unpack(data.read(2))[0]
        return obj

    def __str__(self):
        s = 'DataValue(Value:{}'.format(self.Value)
        if self.StatusCode is not None:
            s += ', StatusCode:{}'.format(self.StatusCode)
        if self.SourceTimestamp is not None:
            s += ', SourceTimestamp:{}'.format(self.SourceTimestamp)
        if self.ServerTimestamp is not None:
            s += ', ServerTimestamp:{}'.format(self.ServerTimestamp)
        if self.SourcePicoseconds is not None:
            s += ', SourcePicoseconds:{}'.format(self.SourcePicoseconds)
        if self.ServerPicoseconds is not None:
            s += ', ServerPicoseconds:{}'.format(self.ServerPicoseconds)
        s += ')'
        return s

    __repr__ = __str__


__nodeid_counter = 2000


def generate_nodeid(idx):
    global __nodeid_counter
    __nodeid_counter += 1
    return NodeId(__nodeid_counter, idx)

FreeOpcUa / python-opcua

Push — master ( abb445...558fdd )

opcua._FrozenClass.__setattr__() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like

opcua._FrozenClass.setattr() A
