mirror of
https://github.com/JoelBender/bacpypes
synced 2025-10-27 00:57:47 +08:00
manually merge in the appropriate differences between issue-38 and issue-48
This commit is contained in:
Joel Bender
@@ -69,10 +69,8 @@ class Tag(object):
|
||||
|
||||
def set(self, tclass, tnum, tlvt=0, tdata=''):
|
||||
"""set the values of the tag."""
|
||||
if isinstance(tdata, bytearray):
|
||||
tdata = bytes(tdata)
|
||||
elif not isinstance(tdata, bytes):
|
||||
raise TypeError("tag data must be bytes or bytearray")
|
||||
if not isinstance(tdata, str):
|
||||
raise TypeError("tag data must be str")
|
||||
|
||||
self.tagClass = tclass
|
||||
self.tagNumber = tnum
|
||||
@@ -81,10 +79,8 @@ class Tag(object):
|
||||
|
||||
def set_app_data(self, tnum, tdata):
|
||||
"""set the values of the tag."""
|
||||
if isinstance(tdata, bytearray):
|
||||
tdata = bytes(tdata)
|
||||
elif not isinstance(tdata, bytes):
|
||||
raise TypeError("tag data must be bytes or bytearray")
|
||||
if not isinstance(tdata, str):
|
||||
raise TypeError("tag data must be str")
|
||||
|
||||
self.tagClass = Tag.applicationTagClass
|
||||
self.tagNumber = tnum
|
||||
@@ -174,7 +170,7 @@ class Tag(object):
|
||||
|
||||
# application tagged boolean now has data
|
||||
if (self.tagNumber == Tag.booleanAppTag):
|
||||
return ContextTag(context, bytearray([self.tagLVT]))
|
||||
return ContextTag(context, chr(self.tagLVT))
|
||||
else:
|
||||
return ContextTag(context, self.tagData)
|
||||
|
||||
@@ -563,11 +559,11 @@ class Unsigned(Atomic):
|
||||
|
||||
def encode(self, tag):
|
||||
# rip apart the number
|
||||
data = bytearray(struct.pack('>L', self.value))
|
||||
data = 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]
|
||||
while (len(data) > 1) and (data[0] == '\x00'):
|
||||
data = data[1:]
|
||||
|
||||
# encode the tag
|
||||
tag.set_app_data(Tag.unsignedAppTag, data)
|
||||
@@ -613,24 +609,24 @@ class Integer(Atomic):
|
||||
|
||||
def encode(self, tag):
|
||||
# rip apart the number
|
||||
data = bytearray(struct.pack('>I', self.value & 0xFFFFFFFF))
|
||||
data = struct.pack('>I', self.value & 0xFFFFFFFF)
|
||||
|
||||
# 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):
|
||||
if (data[0] != '\xFF'):
|
||||
break
|
||||
if (data[1] < 128):
|
||||
if (data[1] < '\x80'):
|
||||
break
|
||||
del data[0]
|
||||
data = data[1:]
|
||||
else:
|
||||
while (len(data) > 1):
|
||||
if (data[0] != 0):
|
||||
if (data[0] != '\x00'):
|
||||
break
|
||||
if (data[1] >= 128):
|
||||
if (data[1] >= '\x80'):
|
||||
break
|
||||
del data[0]
|
||||
data = data[1:]
|
||||
|
||||
# encode the tag
|
||||
tag.set_app_data(Tag.integerAppTag, data)
|
||||
@@ -640,7 +636,7 @@ class Integer(Atomic):
|
||||
raise ValueError("integer application tag required")
|
||||
|
||||
# byte array easier to deal with
|
||||
tag_data = bytearray(tag.tagData)
|
||||
tag_data = [ord(c) for c in tag.tagData]
|
||||
|
||||
# get the data
|
||||
rslt = tag_data[0]
|
||||
@@ -746,8 +742,8 @@ class OctetString(Atomic):
|
||||
pass
|
||||
elif isinstance(arg, Tag):
|
||||
self.decode(arg)
|
||||
elif isinstance(arg, (bytes, bytearray)):
|
||||
self.value = bytes(arg)
|
||||
elif isinstance(arg, str):
|
||||
self.value = arg
|
||||
elif isinstance(arg, OctetString):
|
||||
self.value = arg.value
|
||||
else:
|
||||
@@ -800,11 +796,10 @@ class CharacterString(Atomic):
|
||||
if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.characterStringAppTag):
|
||||
raise ValueError("character string application tag required")
|
||||
|
||||
# byte array easier to deal with
|
||||
tag_data = bytearray(tag.tagData)
|
||||
tag_data = tag.tagData
|
||||
|
||||
# extract the data
|
||||
self.strEncoding = tag_data[0]
|
||||
self.strEncoding = ord(tag_data[0])
|
||||
self.strValue = tag_data[1:]
|
||||
|
||||
# normalize the value
|
||||
@@ -824,7 +819,7 @@ class CharacterString(Atomic):
|
||||
self.value = '### unknown encoding: %d ###' % (self.strEncoding,)
|
||||
|
||||
def __str__(self):
|
||||
return "CharacterString(%d,X'%s')" % (self.strEncoding, btox(self.strValue))
|
||||
return "CharacterString(%d," % (self.strEncoding,) + repr(self.strValue) + ")"
|
||||
|
||||
#
|
||||
# BitString
|
||||
@@ -870,7 +865,7 @@ class BitString(Atomic):
|
||||
unused = used and (8 - used) or 0
|
||||
|
||||
# start with the number of unused bits
|
||||
data = bytearray([unused])
|
||||
data = [unused]
|
||||
|
||||
# build and append each packed octet
|
||||
bits = self.value + [0] * unused
|
||||
@@ -881,13 +876,13 @@ class BitString(Atomic):
|
||||
data.append(x)
|
||||
|
||||
# encode the tag
|
||||
tag.set_app_data(Tag.bitStringAppTag, data)
|
||||
tag.set_app_data(Tag.bitStringAppTag, ''.join(chr(i) for i in data))
|
||||
|
||||
def decode(self, tag):
|
||||
if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.bitStringAppTag):
|
||||
raise ValueError("bit string application tag required")
|
||||
|
||||
tag_data = bytearray(tag.tagData)
|
||||
tag_data = [ord(c) for c in tag.tagData]
|
||||
|
||||
# extract the number of unused bits
|
||||
unused = tag_data[0]
|
||||
@@ -1055,11 +1050,11 @@ class Enumerated(Atomic):
|
||||
raise TypeError("%s is an invalid enumeration value datatype" % (type(self.value),))
|
||||
|
||||
# rip apart the number
|
||||
data = bytearray(struct.pack('>L', value))
|
||||
data = struct.pack('>L', value)
|
||||
|
||||
# reduce the value to the smallest number of octets
|
||||
while (len(data) > 1) and (data[0] == 0):
|
||||
del data[0]
|
||||
while (len(data) > 1) and (data[0] == '\x00'):
|
||||
data = data[1:]
|
||||
|
||||
# encode the tag
|
||||
tag.set_app_data(Tag.enumeratedAppTag, data)
|
||||
@@ -1108,87 +1103,167 @@ def expand_enumerations(klass):
|
||||
# Date
|
||||
#
|
||||
|
||||
_mm = r'(?P<month>0?\d|1[0-4]|odd|even|255|[*])'
|
||||
_dd = r'(?P<day>[0-3]?\d|last|odd|even|255|[*])'
|
||||
_yy = r'(?P<year>\d{2}|255|[*])'
|
||||
_yyyy = r'(?P<year>\d{4}|255|[*])'
|
||||
_dow = r'(?P<dow>[1-7]|mon|tue|wed|thu|fri|sat|sun|255|[*])'
|
||||
|
||||
_special_mon = {'*': 255, 'odd': 13, 'even': 14, None: 255}
|
||||
_special_mon_inv = {255: '*', 13: 'odd', 14: 'even'}
|
||||
|
||||
_special_day = {'*': 255, 'last': 32, 'odd': 33, 'even': 34, None: 255}
|
||||
_special_day_inv = {255: '*', 32: 'last', 33: 'odd', 34: 'even'}
|
||||
|
||||
_special_dow = {'*': 255, 'mon': 1, 'tue': 2, 'wed': 3, 'thu': 4, 'fri': 5, 'sat': 6, 'sun': 7}
|
||||
_special_dow_inv = {255: '*', 1: 'mon', 2: 'tue', 3: 'wed', 4: 'thu', 5: 'fri', 6: 'sat', 7: 'sun'}
|
||||
|
||||
|
||||
def _merge(*args):
|
||||
"""Create a composite pattern and compile it."""
|
||||
return re.compile(r'^' + r'[/-]'.join(args) + r'(?:\s+' + _dow + ')?$')
|
||||
|
||||
|
||||
# make a list of compiled patterns
|
||||
_date_patterns = [
|
||||
_merge(_yyyy, _mm, _dd),
|
||||
_merge(_mm, _dd, _yyyy),
|
||||
_merge(_dd, _mm, _yyyy),
|
||||
_merge(_yy, _mm, _dd),
|
||||
_merge(_mm, _dd, _yy),
|
||||
_merge(_dd, _mm, _yy),
|
||||
]
|
||||
|
||||
|
||||
class Date(Atomic):
|
||||
|
||||
_app_tag = Tag.dateAppTag
|
||||
_date_regex = re.compile(r"^([*]|\d+)[/]([*]|\d+)[/]([*]|\d+)(?:\s([*]|\w+))?$")
|
||||
_day_names = ['', 'Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']
|
||||
|
||||
DONT_CARE = 255
|
||||
|
||||
def __init__(self, arg=None, year=255, month=255, day=255, dayOfWeek=255):
|
||||
self.value = (year, month, day, dayOfWeek)
|
||||
|
||||
def __init__(self, arg=None, year=255, month=255, day=255, day_of_week=255):
|
||||
self.value = (year, month, day, day_of_week)
|
||||
|
||||
if arg is None:
|
||||
pass
|
||||
elif isinstance(arg,Tag):
|
||||
elif isinstance(arg, Tag):
|
||||
self.decode(arg)
|
||||
elif isinstance(arg, tuple):
|
||||
self.value = arg
|
||||
elif isinstance(arg, str):
|
||||
date_match = Date._date_regex.match(arg)
|
||||
if not date_match:
|
||||
raise ValueError("invalid date pattern")
|
||||
date_groups = date_match.groups()
|
||||
# lower case everything
|
||||
arg = arg.lower()
|
||||
|
||||
# day/month/year
|
||||
tup_list = []
|
||||
for s in date_groups[:3]:
|
||||
if s == '*':
|
||||
tup_list.append(255)
|
||||
elif s in None:
|
||||
tup_list.append(0)
|
||||
else:
|
||||
tup_list.append(int(s))
|
||||
# make a list of the contents from matching patterns
|
||||
matches = []
|
||||
for p in _date_patterns:
|
||||
m = p.match(arg)
|
||||
if m:
|
||||
matches.append(m.groupdict())
|
||||
|
||||
# clean up the year
|
||||
if (tup_list[2] < 100):
|
||||
tup_list[2] += 2000
|
||||
tup_list[2] -= 1900
|
||||
# try to find a good one
|
||||
match = None
|
||||
if not matches:
|
||||
raise ValueError("unmatched")
|
||||
|
||||
# day-of-week madness
|
||||
dow = date_groups[3]
|
||||
if dow is None:
|
||||
tup_list.append(0)
|
||||
elif (dow == '*'):
|
||||
tup_list.append(255)
|
||||
elif dow.isdigit():
|
||||
tup_list.append(int(dow))
|
||||
# if there is only one, success
|
||||
if len(matches) == 1:
|
||||
match = matches[0]
|
||||
else:
|
||||
dow = dow.title()
|
||||
if dow not in Date._day_names:
|
||||
raise ValueError("invalid day name")
|
||||
tup_list.append(Date._day_names.index(dow))
|
||||
# check to see if they really are the same
|
||||
for a, b in zip(matches[:-1],matches[1:]):
|
||||
if a != b:
|
||||
raise ValueError("ambiguous")
|
||||
break
|
||||
else:
|
||||
match = matches[0]
|
||||
|
||||
# extract the year and normalize
|
||||
year = match['year']
|
||||
if (year == '*') or (not year):
|
||||
year = 255
|
||||
else:
|
||||
year = int(year)
|
||||
if (year == 255):
|
||||
pass
|
||||
elif year < 35:
|
||||
year += 2000
|
||||
elif year < 100:
|
||||
year += 1900
|
||||
|
||||
# extract the month and normalize
|
||||
month = match['month']
|
||||
if month in _special_mon:
|
||||
month = _special_mon[month]
|
||||
else:
|
||||
month = int(month)
|
||||
if month > 14:
|
||||
print("invalid month")
|
||||
|
||||
# extract the day and normalize
|
||||
day = match['day']
|
||||
if day in _special_day:
|
||||
day = _special_day[day]
|
||||
else:
|
||||
day = int(day)
|
||||
if day > 34:
|
||||
print("invalid day")
|
||||
|
||||
# extract the day-of-week and normalize
|
||||
day_of_week = match['dow']
|
||||
if day_of_week in _special_dow:
|
||||
day_of_week = _special_dow[day_of_week]
|
||||
elif not day_of_week:
|
||||
pass
|
||||
else:
|
||||
day_of_week = int(day_of_week)
|
||||
|
||||
# year becomes the correct octet
|
||||
if year != 255:
|
||||
year -= 1900
|
||||
|
||||
# save the value
|
||||
self.value = (year, month, day, day_of_week)
|
||||
|
||||
# calculate the day of the week
|
||||
if not day_of_week:
|
||||
self.CalcDayOfWeek()
|
||||
|
||||
self.value = tuple(tup_list)
|
||||
elif isinstance(arg, Date):
|
||||
self.value = arg.value
|
||||
|
||||
else:
|
||||
raise TypeError("invalid constructor datatype")
|
||||
|
||||
def now(self):
|
||||
tup = time.localtime()
|
||||
|
||||
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
|
||||
year, month, day, day_of_week = 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
|
||||
# assume the worst
|
||||
day_of_week = 255
|
||||
|
||||
# check for special values
|
||||
if year == 255:
|
||||
pass
|
||||
elif month in _special_mon_inv:
|
||||
pass
|
||||
elif day in _special_day_inv:
|
||||
pass
|
||||
else:
|
||||
try:
|
||||
today = time.mktime( (year + 1900, month, day, 0, 0, 0, 0, 0, -1) )
|
||||
day_of_week = time.gmtime(today)[6] + 1
|
||||
except OverflowError:
|
||||
pass
|
||||
|
||||
# put it back together
|
||||
self.value = (year, month, day, dayOfWeek)
|
||||
self.value = (year, month, day, day_of_week)
|
||||
|
||||
def now(self):
|
||||
tup = time.localtime()
|
||||
self.value = (tup[0]-1900, tup[1], tup[2], tup[6] + 1)
|
||||
return self
|
||||
|
||||
def encode(self, tag):
|
||||
# encode the tag
|
||||
tag.set_app_data(Tag.dateAppTag, bytearray(self.value))
|
||||
tag.set_app_data(Tag.dateAppTag, ''.join(chr(i) for i in self.value))
|
||||
|
||||
def decode(self, tag):
|
||||
if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.dateAppTag):
|
||||
@@ -1198,28 +1273,21 @@ class Date(Atomic):
|
||||
self.value = tuple(ord(c) for c in tag.tagData)
|
||||
|
||||
def __str__(self):
|
||||
"""String representation of the date."""
|
||||
# rip it apart
|
||||
year, month, day, dayOfWeek = self.value
|
||||
year, month, day, day_of_week = self.value
|
||||
|
||||
rslt = "Date("
|
||||
if month == 255:
|
||||
rslt += "*/"
|
||||
else:
|
||||
rslt += "%d/" % (month,)
|
||||
if day == 255:
|
||||
rslt += "*/"
|
||||
else:
|
||||
rslt += "%d/" % (day,)
|
||||
if year == 255:
|
||||
rslt += "* "
|
||||
year = "*"
|
||||
else:
|
||||
rslt += "%d " % (year + 1900,)
|
||||
if dayOfWeek == 255:
|
||||
rslt += "*)"
|
||||
else:
|
||||
rslt += Date._day_names[dayOfWeek] + ")"
|
||||
year = str(year + 1900)
|
||||
|
||||
month = _special_mon_inv.get(month, str(month))
|
||||
day = _special_day_inv.get(day, str(day))
|
||||
day_of_week = _special_dow_inv.get(day_of_week, str(day_of_week))
|
||||
|
||||
return "%s(%s-%s-%s %s)" % (self.__class__.__name__, year, month, day, day_of_week)
|
||||
|
||||
return rslt
|
||||
|
||||
#
|
||||
# Time
|
||||
@@ -1280,7 +1348,7 @@ class Time(Atomic):
|
||||
|
||||
def encode(self, tag):
|
||||
# encode the tag
|
||||
tag.set_app_data(Tag.timeAppTag, bytearray(self.value))
|
||||
tag.set_app_data(Tag.timeAppTag, ''.join(chr(c) for c in self.value))
|
||||
|
||||
def decode(self, tag):
|
||||
if (tag.tagClass != Tag.applicationTagClass) or (tag.tagNumber != Tag.timeAppTag):
|
||||
|
||||
Reference in New Issue
Block a user