bacnet.py

BacNet Simulation Scripts（英文的）。用于数据采集。

class Null(Atomic):

    _appTag = Tag.nullAppTag

    def __init__(self, arg=None):
        self.value = ()

        if arg is None:
            pass
        elif isinstance(arg,Tag):
            self.Decode(arg)
        elif isinstance(arg,types.TupleType):
            if len(arg) != 0:
                raise ValueError, "empty tuple required"
        else:
            raise TypeError, "invalid constructor datatype"

    def Encode(self, tag):
        tag.SetAppData(Tag.nullAppTag, '')

    def Decode(self, tag):
        if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.nullAppTag):
            raise ValueError, "null application tag required"

        self.value = ()

    def __str__(self):
        return "Null"

#
#   Boolean
#

class Boolean(Atomic):

    _appTag = Tag.booleanAppTag

    def __init__(self, arg=None):
        self.value = False

        if arg is None:
            pass
        elif isinstance(arg,Tag):
            self.Decode(arg)
        elif isinstance(arg,types.BooleanType):
            self.value = arg
        else:
            raise TypeError, "invalid constructor datatype"

    def Encode(self, tag):
        tag.Set(Tag.applicationTagClass, Tag.booleanAppTag, int(self.value), '')

    def Decode(self, tag):
        if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.booleanAppTag):
            raise ValueError, "boolean application tag required"

        # get the data
        self.value = bool(tag.tagLVT)

    def __str__(self):
        return "Boolean(%s)" % (str(self.value), )

#
#   Unsigned
#

class Unsigned(Atomic):

    _appTag = Tag.unsignedAppTag

    def __init__(self,arg = None):
        self.value = 0L

        if arg is None:
            pass
        elif isinstance(arg,Tag):
            self.Decode(arg)
        elif isinstance(arg,types.IntType):
            if (arg < 0):
                raise ValueError, "unsigned integer required"
            self.value = long(arg)
        elif isinstance(arg,types.LongType):
            if (arg < 0):
                raise ValueError, "unsigned integer required"
            self.value = arg
        else:
            raise TypeError, "invalid constructor datatype"

    def Encode(self, tag):
        # rip apart the number
        data = [ord(c) for c in struct.pack('>L',self.value)]

        # reduce the value to the smallest number of octets
        while (len(data) > 1) and (data[0] == 0):
            del data[0]

        # encode the tag
        tag.SetAppData(Tag.unsignedAppTag, ''.join(chr(c) for c in data))

    def Decode(self, tag):
        if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.unsignedAppTag):
            raise ValueError, "unsigned application tag required"

        # get the data
        rslt = 0L
        for c in tag.tagData:
            rslt = (rslt << 8) + ord(c)

        # save the result
        self.value = rslt

    def __str__(self):
        return "Unsigned(%s)" % (self.value, )

#
#   Integer
#

class Integer(Atomic):

    _appTag = Tag.integerAppTag

    def __init__(self,arg = None):
        self.value = 0

        if arg is None:
            pass
        elif isinstance(arg,Tag):
            self.Decode(arg)
        elif isinstance(arg,types.IntType):
            self.value = arg
        elif isinstance(arg,types.LongType):
            self.value = arg
        else:
            raise TypeError, "invalid constructor datatype"

    def Encode(self, tag):
        # rip apart the number
        data = [ord(c) for c in struct.pack('>I',self.value)]

        # reduce the value to the smallest number of bytes, be
        # careful about sign extension
        if self.value < 0:
            while (len(data) > 1):
                if (data[0] != 255):
                    break
                if (data[1] < 128):
                    break
                del data[0]
        else:
            while (len(data) > 1):
                if (data[0] != 0):
                    break
                if (data[1] >= 128):
                    break
                del data[0]

        # encode the tag
        tag.SetAppData(Tag.integerAppTag, ''.join(chr(c) for c in data))

    def Decode(self, tag):
        if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.integerAppTag):
            raise ValueError, "integer application tag required"

        # get the data
        rslt = ord(tag.tagData[0])
        if (rslt & 0x80) != 0:
            rslt = (-1 << 8) | rslt

        for c in tag.tagData[1:]:
            rslt = (rslt << 8) | ord(c)

        # save the result
        self.value = rslt

    def __str__(self):
        return "Integer(%s)" % (self.value, )

#
#   Real
#

class Real(Atomic):

    _appTag = Tag.realAppTag

    def __init__(self, arg=None):
        self.value = 0.0

        if arg is None:
            pass
        elif isinstance(arg,Tag):
            self.Decode(arg)
        elif isinstance(arg,types.FloatType):
            self.value = arg
        else:
            raise TypeError, "invalid constructor datatype"

    def Encode(self, tag):
        # encode the tag
        tag.SetAppData(Tag.realAppTag, struct.pack('>f',self.value))

    def Decode(self, tag):
        if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.realAppTag):
            raise ValueError, "real application tag required"

        # extract the data
        self.value = struct.unpack('>f',tag.tagData)[0]

    def __str__(self):
        return "Real(%g)" % (self.value,)

#
#   Double
#

class Double(Atomic):

    _appTag = Tag.doubleAppTag

    def __init__(self,arg = None):
        self.value = 0.0

        if arg is None:
            pass
        elif isinstance(arg,Tag):
            self.Decode(arg)
        elif isinstance(arg,types.FloatType):
            self.value = arg
        else:
            raise TypeError, "invalid constructor datatype"

    def Encode(self, tag):
        # encode the tag
        tag.SetAppData(Tag.doubleAppTag, struct.pack('>d',self.value))

    def Decode(self, tag):
        if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.doubleAppTag):
            raise ValueError, "double application tag required"

        # extract the data
        self.value = struct.unpack('>d',tag.tagData)[0]

    def __str__(self):
        return "Double(%g)" % (self.value,)

#
#   OctetString
#

class OctetString(Atomic):

    _appTag = Tag.octetStringAppTag

    def __init__(self, arg=None):
        self.value = ''

        if arg is None:
            pass
        elif isinstance(arg,Tag):
            self.Decode(arg)
        elif isinstance(arg,types.StringType):
            self.value = arg
        else:
            raise TypeError, "invalid constructor datatype"

    def Encode(self, tag):
        # encode the tag
        tag.SetAppData(Tag.octetStringAppTag, self.value)

    def Decode(self, tag):
        if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.octetStringAppTag):
            raise ValueError, "octet string application tag required"

        self.value = tag.tagData

    def __str__(self):
        return "OctetString(X'" + StringToHex(self.value) + "')"

#
#   CharacterString
#

class CharacterString(Atomic):

    _appTag = Tag.characterStringAppTag

    def __init__(self, arg=None):
        self.value = ''
        self.strEncoding = 0

        if arg is None:
            pass
        elif isinstance(arg,Tag):
            self.Decode(arg)
        elif isinstance(arg,types.StringType):
            self.value = arg
        else:
            raise TypeError, "invalid constructor datatype"

    def Encode(self, tag):
        # encode the tag
        tag.SetAppData(Tag.characterStringAppTag, chr(self.strEncoding)+self.value)

    def Decode(self, tag):
        if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.characterStringAppTag):
            raise ValueError, "character string application tag required"

        # extract the data
        self.strEncoding = tag.tagData[0]
        self.value = tag.tagData[1:]

    def __str__(self):
        return "CharacterString(%d," % (self.strEncoding,) + repr(self.value) + ")"

#
#   BitString
#

class BitString(Atomic):

    _appTag = Tag.bitStringAppTag
    bitNames = {}
    bitLen = 0

    def __init__(self, arg = None):
        self.value = [0] * self.bitLen

        if arg is None:
            pass
        elif isinstance(arg,Tag):
            self.Decode(arg)
        elif isinstance(arg,types.ListType):
            allInts = allStrings = True
            for elem in arg:
                allInts = allInts and ((elem == 0) or (elem == 1))
                allStrings = allStrings and self.bitNames.has_key(elem)

            if allInts:
                self.value = arg
            elif allStrings:
                for bit in arg:
                    bit = self.bitNames[bit]
                    if (bit < 0) or (bit > len(self.value)):
                        raise IndexError, "constructor element out of range"
                    self.value[bit] = 1
            else:
                raise TypeError, "invalid constructor list element(s)"
        else:
            raise TypeError, "invalid constructor datatype"

    def Encode(self, tag):
        # compute the unused bits to fill out the string
        _, used = divmod(len(self.value), 8)
        unused = used and (8 - used) or 0

        # start with the number of unused bits
        data = chr(unused)

        # build and append each packed octet
        bits = self.value + [0] * unused
        for i in range(0,len(bits),8):
            x = 0
            for j in range(0,8):
                x |= bits[i + j] << (7 - j)
            data += chr(x)

        # encode the tag
        tag.SetAppData(Tag.bitStringAppTag, data)

    def Decode(self, tag):
        if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.bitStringAppTag):
            raise ValueError, "bit string application tag required"

        # extract the number of unused bits
        unused = ord(tag.tagData[0])

        # extract the data
        data = []
        for c in tag.tagData[1:]:
            x = ord(c)
            for i in range(8):
                if (x & (1 << (7 - i))) != 0:
                    data.append( 1 )
                else:
                    data.append( 0 )

        # trim off the unused bits
        self.value = data[:-unused]

    def __str__(self):
        # flip the bit names
        bitNames = {}
        for key, value in self.bitNames.iteritems():
            bitNames[value] = key

        # build a list of values and/or names
        valueList = []
        for value, index in zip(self.value,range(len(self.value))):
            if bitNames.has_key(index):
                if value:
                    valueList.append(bitNames[index])
                else:
                    valueList.append('!' + bitNames[index])
            else:
                valueList.append(str(value))

        # bundle it together
        return "BitString(" + ','.join(valueList) + ")"

    def __getitem__(self,bit):
        if isinstance(bit,types.IntType):
            pass
        elif isinstance(bit,types.StringType):
            if not self.bitNames.has_key(bit):
                raise IndexError, "unknown bit name '%s'" % (bit,)

            bit = self.bitNames[bit]
        else:
            raise TypeError, "bit index must be an integer or bit name"

        if (bit < 0) or (bit > len(self.value)):
            raise IndexError, "list index out of range"

        return self.value[bit]

    def __setitem__(self,bit,value):
        if isinstance(bit,types.IntType):
            pass
        elif isinstance(bit,types.StringType):