bacnet.py

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

            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"

        # funny cast to a bit
        self.value[bit] = value and 1 or 0

#
#   Enumerated
#

class Enumerated(Atomic):

    _appTag = Tag.enumeratedAppTag

    enumerations = {}
    _xlateTable = {}

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

        # see if the class has a translate table
        if not self.__class__.__dict__.has_key('_xlateTable'):
            ExpandEnumerations(self.__class__)

        # initialize the object
        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"

            # convert it to a string if you can
            try: self.value = self._xlateTable[arg]
            except KeyError: self.value = long(arg)
        elif isinstance(arg, types.LongType):
            if (arg < 0):
                raise ValueError, "unsigned integer required"

            # convert it to a string if you can
            try: self.value = self._xlateTable[arg]
            except KeyError: self.value = long(arg)
        elif isinstance(arg,types.StringType):
            if self._xlateTable.has_key(arg):
                self.value = arg
            else:
                raise ValueError, "undefined enumeration '%s'" % (arg,)
        else:
            raise TypeError, "invalid constructor datatype"

    def __getitem__(self, item):
        return self._xlateTable[item]

    def GetLong(self):
        if isinstance(self.value, types.LongType):
            return self.value
        elif isinstance(self.value, types.StringType):
            return long(self._xlateTable[self.value])
        else:
            raise TypeError, "%s is an invalid enumeration value datatype" % (type(self.value),)

    def keylist(self):
        """Return a list of names in order by value."""
        items = self.enumerations.items()
        items.sort(lambda a, b: cmp(a[1], b[1]))

        # last item has highest value
        rslt = [None] * (items[-1][1] + 1)

        # map the values
        for key, value in items:
            rslt[value] = key

        # return the result
        return rslt

    def __cmp__(self, other):
        """Special function to make sure comparisons are done in enumeration
        order, not alphabetic order."""
        # hoop jump it
        if not isinstance(other, self.__class__):
            other = self.__class__(other)

        # get the numeric version
        a = self.GetLong()
        b = other.GetLong()

        # now compare the values
        if (a < b):
            return -1
        elif (a > b):
            return 1
        else:
            return 0

    def Encode(self, tag):
        if isinstance(self.value, types.IntType):
            value = long(self.value)
        if isinstance(self.value, types.LongType):
            value = self.value
        elif isinstance(self.value, types.StringType):
            value = self._xlateTable[self.value]
        else:
            raise TypeError, "%s is an invalid enumeration value datatype" % (type(self.value),)

        # rip apart the number
        data = [ord(c) for c in struct.pack('>L',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.enumeratedAppTag, ''.join(chr(c) for c in data))

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

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

        # convert it to a string if you can
        try: rslt = self._xlateTable[rslt]
        except KeyError: pass

        # save the result
        self.value = rslt

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

#
#   ExpandEnumerations
#

# translate lowers to uppers, keep digits, toss everything else
_ExpandTranslateTable = ''.join([c.isalnum() and c.upper() or '-' for c in [chr(cc) for cc in range(256)]])
_ExpandDeleteChars = ''.join([chr(cc) for cc in range(256) if not chr(cc).isalnum()])

def ExpandEnumerations(klass):
    # build a value dictionary
    xlateTable = {}
    for name, value in klass.enumerations.iteritems():
        # save the results
        xlateTable[name] = value
        xlateTable[value] = name

        # translate the name for a class const
        name = name.translate(_ExpandTranslateTable, _ExpandDeleteChars)

        # save the name in the class
        setattr(klass, name, value)

    # save the dictionary in the class
    setattr(klass, '_xlateTable', xlateTable)

#
#   Date
#

class Date(Atomic):

    _appTag = Tag.dateAppTag

    DONT_CARE = 255

    def __init__(self, arg=None, year=255, month=255, day=255, dayOfWeek=255):
        self.value = (year, month, day, dayOfWeek)

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

    def Now(self):
        tup = time.gmtime()

        self.value = (tup[0]-1900, tup[1], tup[2], tup[6] + 1)

        return self

    def CalcDayOfWeek(self):
        """Calculate the correct day of the week."""
        # rip apart the value
        year, month, day, dayOfWeek = self.value

        # make sure all the components are defined
        if (year != 255) and (month != 255) and (day != 255):
            today = time.mktime( (year + 1900, month, day, 0, 0, 0, 0, 0, -1) )
            dayOfWeek = time.gmtime(today)[6] + 1

        # put it back together
        self.value = (year, month, day, dayOfWeek)

    def Encode(self, tag):
        # encode the tag
        tag.SetAppData(Tag.dateAppTag, ''.join(chr(c) for c in self.value))

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

        # rip apart the data
        self.value = tuple(ord(c) for c in tag.tagData)

    def __str__(self):
        # rip it apart
        year, month, day, dayOfWeek = self.value

        rslt = "Date("
        if month == 255:
            rslt += "*/"
        else:
            rslt += "%d/" % (month,)
        if day == 255:
            rslt += "*/"
        else:
            rslt += "%d/" % (day,)
        if year == 255:
            rslt += "* "
        else:
            rslt += "%d " % (year + 1900,)
        if dayOfWeek == 255:
            rslt += "*)"
        else:
            rslt += ['','Mon','Tue','Wed','Thu','Fri','Sat','Sun'][dayOfWeek] + ")"

        return rslt

#
#   Time
#

class Time(Atomic):

    _appTag = Tag.timeAppTag

    DONT_CARE = 255

    def __init__(self, arg=None, hour=255, minute=255, second=255, hundredth=255):
        # put it together
        self.value = (hour, minute, second, hundredth)

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

    def Now(self):
        now = time.time()
        tup = time.gmtime(now)

        self.value = (tup[3], tup[4], tup[5], int((now - int(now)) * 100))

        return self

    def Encode(self, tag):
        # encode the tag
        tag.SetAppData(Tag.timeAppTag, ''.join(chr(c) for c in self.value))

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

        # rip apart the data
        self.value = tuple(ord(c) for c in tag.tagData)

    def __str__(self):
        # rip it apart
        hour, minute, second, hundredth = self.value

        rslt = "Time("
        if hour == 255:
            rslt += "*:"
        else:
            rslt += "%02d:" % (hour,)
        if minute == 255:
            rslt += "*:"
        else:
            rslt += "%02d:" % (minute,)
        if second == 255:
            rslt += "*."
        else:
            rslt += "%02d." % (second,)
        if hundredth == 255:
            rslt += "*)"
        else:
            rslt += "%02d)" % (hundredth,)

        return rslt

#
#   ObjectType
#

class ObjectType(Enumerated):
    enumerations = \
        { 'analog-input':0 \
        , 'analog-output':1 \
        , 'analog-value':2 \
        , 'binary-input':3 \
        , 'binary-output':4 \
        , 'binary-value':5 \
        , 'calendar':6 \
        , 'command':7 \
        , 'device':8 \
        , 'event-enrollment':9 \
        , 'file':10 \
        , 'group':11 \
        , 'loop':12 \
        , 'multi-state-input':13 \
        , 'multi-state-output':14 \
        , 'notification-class':15 \
        , 'program':16 \
        , 'schedule':17 \
        , 'averaging':18 \
        , 'multi-state-value':19 \
        , 'trend-log':20 \
        , 'life-safety-point':21 \
        , 'life-safety-zone':22 \
        , 'accumulator':23 \
        , 'pulse-converter':24 \
        }

ExpandEnumerations(ObjectType)

#
#   ObjectIdentifier
#

class ObjectIdentifier(Atomic):

    _appTag = Tag.objectIdentifierAppTag
    objectTypeClass = ObjectType

    def __init__(self, *args):
        self.value = ('analog-input', 0)

        if len(args) == 0:
            pass
        elif len(args) == 1:
            arg = args[0]
            if isinstance(arg, Tag):
                self.Decode(arg)
            elif isinstance(arg, types.IntType):
                self.SetLong(long(arg))
            elif isinstance(arg, types.LongType):
                self.SetLong(arg)
            elif isinstance(arg, types.TupleType):
                self.SetTuple(*arg)
            else:
                raise TypeError, "invalid constructor datatype"
        elif len(args) == 2:
            self.SetTuple(*args)
        else:
            raise ValueError, "invalid constructor parameters"

    def SetTuple(self, objType, objInstance):
        # allow a type name as well as an integer
        if isinstance(objType, types.IntType):
            try:
                # try and make it pretty
                objType = self.objectTypeClass()[objType]
            except KeyError:
                pass
        elif isinstance(objType, types.StringType):
            try:
                # make sure the type is known
                self.objectTypeClass()[objType]
            except KeyError:
                raise ValueError, "unrecognized object type '%s'" % (objType,)
        else:
            raise TypeError, "invalid datatype for objType"

        # pack the components together
        self.value = (objType, objInstance)

    def GetTuple(self):
        """Return the unsigned integer tuple of the identifier."""
        objType, objInstance = self.value

        if isinstance(objType, types.IntType):
            pass
        elif isinstance(objType, types.LongType):
            objType = int(objType)
        elif isinstance(objType, types.StringType):
            # turn it back into an integer
            objType = self.objectTypeClass()[objType]
        else:
            raise TypeError, "invalid datatype for objType"

        # pack the components together
        return (objType, objInstance)

    def SetLong(self, value):
        # suck out the type
        objType = (value >> 22) & 0x03FF
        try:
            # try and make it pretty
            objType = self.objectTypeClass()[objType]
        except IndexError:
            pass

        # suck out the instance
        objInstance = value & 0x003FFFFF

        # save the result
        self.value = (objType, objInstance)

    def GetLong(self):
        """Return the unsigned integer representation of the identifier."""