mirror/bacpypes
1
0
Fork 0
mirror of https://github.com/JoelBender/bacpypes synced 2025-09-28 22:15:23 +08:00
Code Issues Releases Wiki Activity

ready to release

Browse Source
This commit is contained in:
Joel Bender 2022-02-02 11:45:02 -05:00
parent 41104c2b56
commit 27ab4f18aa
11 changed files with 113 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
py25/bacpypes/basetypes.py
View File

@ -1295,7 +1295,7 @@ class PropertyIdentifier(Enumerated):
, 'integralConstant':49
, 'integralConstantUnits':50
, 'intervalOffset':195
, 'isUtc':344
, 'isUTC':344
, 'keySets':330
, 'lastAccessEvent':275
, 'lastAccessPoint':276

4
py25/bacpypes/object.py
View File

@ -1383,7 +1383,7 @@ class DateTimePatternValueObject(Object):
, OptionalProperty('outOfService', Boolean)
, OptionalProperty('priorityArray', PriorityArray)
, OptionalProperty('relinquishDefault', DateTime)
, OptionalProperty('isUtc', Boolean)
, OptionalProperty('isUTC', Boolean)
]
register_object_type(DateTimePatternValueObject)
@ -1400,7 +1400,7 @@ class DateTimeValueObject(Object):
, OptionalProperty('outOfService', Boolean)
, OptionalProperty('priorityArray', PriorityArray)
, OptionalProperty('relinquishDefault', DateTime)
, OptionalProperty('isUtc', Boolean)
, OptionalProperty('isUTC', Boolean)
]
register_object_type(DateTimeValueObject)

2
py27/bacpypes/basetypes.py
View File

@ -1518,7 +1518,7 @@ class PropertyIdentifier(Enumerated):
, 'ipv6PrefixLength':437
, 'ipv6ZoneIndex':446
, 'issueConfirmedNotifications':51
, 'isUtc':344
, 'isUTC':344
, 'keySets':330
, 'landingCallControl':471
, 'landingCalls': 470

4
py27/bacpypes/object.py
View File

@ -1573,7 +1573,7 @@ class DateTimePatternValueObject(Object):
, OptionalProperty('eventState', EventState)
, OptionalProperty('reliability', Reliability)
, OptionalProperty('outOfService', Boolean)
, OptionalProperty('isUtc', Boolean)
, OptionalProperty('isUTC', Boolean)
, OptionalProperty('priorityArray', PriorityArray)
, OptionalProperty('relinquishDefault', DateTime)
@ -1608,7 +1608,7 @@ class DateTimeValueObject(Object):
, OptionalProperty('outOfService', Boolean)
, OptionalProperty('priorityArray', PriorityArray)
, OptionalProperty('relinquishDefault', DateTime)
, OptionalProperty('isUtc', Boolean)
, OptionalProperty('isUTC', Boolean)
, OptionalProperty('reliabilityEvaluationInhibit', Boolean)
, OptionalProperty('eventDetectionEnable', Boolean)
, OptionalProperty('notificationClass', Unsigned)

5
py27/bacpypes/pdu.py
View File

@ -188,7 +188,10 @@ class Address:
Address._warning("route provided but not route aware: %r", addr)
if route_addr:
self.addrRoute = Address(int(route_addr))
if route_addr.startswith("0x"):
self.addrRoute = Address(xtob(route_addr[2:]))
else:
self.addrRoute = Address(int(route_addr))
if _debug: Address._debug(" - addrRoute: %r", self.addrRoute)
elif route_ip_addr:
if not route_ip_port:

2
py34/bacpypes/__init__.py
View File

@ -18,7 +18,7 @@ if _sys.platform not in _supported_platforms:
# Project Metadata
#
__version__ = '0.18.5'
__version__ = '0.18.6'
__author__ = 'Joel Bender'
__email__ = 'joel@carrickbender.com'

2
py34/bacpypes/basetypes.py
View File

@ -1518,7 +1518,7 @@ class PropertyIdentifier(Enumerated):
, 'ipv6PrefixLength':437
, 'ipv6ZoneIndex':446
, 'issueConfirmedNotifications':51
, 'isUtc':344
, 'isUTC':344
, 'keySets':330
, 'landingCallControl':471
, 'landingCalls': 470

4
py34/bacpypes/object.py
View File

@ -1574,7 +1574,7 @@ class DateTimePatternValueObject(Object):
, OptionalProperty('eventState', EventState)
, OptionalProperty('reliability', Reliability)
, OptionalProperty('outOfService', Boolean)
, OptionalProperty('isUtc', Boolean)
, OptionalProperty('isUTC', Boolean)
, OptionalProperty('priorityArray', PriorityArray)
, OptionalProperty('relinquishDefault', DateTime)
@ -1609,7 +1609,7 @@ class DateTimeValueObject(Object):
, OptionalProperty('outOfService', Boolean)
, OptionalProperty('priorityArray', PriorityArray)
, OptionalProperty('relinquishDefault', DateTime)
, OptionalProperty('isUtc', Boolean)
, OptionalProperty('isUTC', Boolean)
, OptionalProperty('reliabilityEvaluationInhibit', Boolean)
, OptionalProperty('eventDetectionEnable', Boolean)
, OptionalProperty('notificationClass', Unsigned)

5
py34/bacpypes/pdu.py
View File

@ -204,7 +204,10 @@ class Address:
Address._warning("route provided but not route aware: %r", addr)
if route_addr:
self.addrRoute = Address(int(route_addr))
if route_addr.startswith("0x"):
self.addrRoute = Address(xtob(route_addr[2:]))
else:
self.addrRoute = Address(int(route_addr))
if _debug: Address._debug(" - addrRoute: %r", self.addrRoute)
elif route_ip_addr:
if not route_ip_port:

35
samples/WhoIsIAm.py
View File

@ -43,25 +43,23 @@ class WhoIsIAmApplication(BIPSimpleApplication):
# keep track of requests to line up responses
self._request = None
def process_io(self, iocb):
if _debug: WhoIsIAmApplication._debug("process_io %r", iocb)
def request(self, apdu):
if _debug: WhoIsIAmApplication._debug("request %r", apdu)
# save a copy of the request
self._request = iocb.args[0]
if isinstance(apdu, WhoIsRequest):
self._request = apdu
# forward it along
BIPSimpleApplication.process_io(self, iocb)
def confirmation(self, apdu):
if _debug: WhoIsIAmApplication._debug("confirmation %r", apdu)
# forward it along
BIPSimpleApplication.confirmation(self, apdu)
BIPSimpleApplication.request(self, apdu)
def indication(self, apdu):
if _debug: WhoIsIAmApplication._debug("indication %r", apdu)
if (isinstance(self._request, WhoIsRequest)) and (isinstance(apdu, IAmRequest)):
if not self._request:
if _debug: WhoIsIAmApplication._debug(" - no pending request")
elif isinstance(apdu, IAmRequest):
device_type, device_instance = apdu.iAmDeviceIdentifier
if device_type != 'device':
raise DecodingError("invalid object type")
@ -84,6 +82,11 @@ class WhoIsIAmApplication(BIPSimpleApplication):
# forward it along
BIPSimpleApplication.indication(self, apdu)
def confirmation(self, apdu):
if _debug: WhoIsIAmApplication._debug("confirmation %r", apdu)
# forward it along
BIPSimpleApplication.confirmation(self, apdu)
#
# WhoIsIAmConsoleCmd
@ -117,6 +120,16 @@ class WhoIsIAmConsoleCmd(ConsoleCmd):
except Exception as error:
WhoIsIAmConsoleCmd._exception("exception: %r", error)
def do_any(self, args):
"""any
Print all of the I-Am's received as if an unconstrained Who-Is was
sent out, without actually sending an unconstrained Who-Is.
"""
this_application._request = WhoIsRequest()
this_application._request.deviceInstanceRangeLowLimit = 0
this_application._request.deviceInstanceRangeHighLimit = 4194303
def do_iam(self, args):
"""iam"""
args = args.split()

95
tests/test_pdu/test_address.py
View File

@ -46,6 +46,21 @@ class MatchAddressMixin:
assert addr.addrAddr == xtob(a)
@bacpypes_debugging
def setup_module():
"""This function is called once at the beginning of all of the tests
in this module."""
if _debug: setup_module._debug("setup_module")
settings.route_aware = True
@bacpypes_debugging
def teardown_module():
"""This function is called once at the end of the tests in this module."""
if _debug: teardown_module._debug("teardown_module")
settings.route_aware = False
@bacpypes_debugging
class TestAddress(unittest.TestCase, MatchAddressMixin):
@ -441,13 +456,13 @@ class TestRemoteStation(unittest.TestCase, MatchAddressMixin):
self.match_address(test_addr, 4, 1, 6, '01020304bac0')
assert str(test_addr) == "1:1.2.3.4"
test_addr = RemoteStation(1, xtob('01020304bac1'))
self.match_address(test_addr, 4, 1, 6, '01020304bac1')
assert str(test_addr) == "1:1.2.3.4:47809"
def test_remote_station_ints_routed(self):
if _debug: TestRemoteStation._debug("test_remote_station_ints_routed")
if not settings.route_aware:
if _debug: TestRemoteStation._debug(" - not route aware")
return
# test integer
test_addr = RemoteStation(1, 1, route=Address("1.2.3.4"))
self.match_address(test_addr, 4, 1, 1, '01')
@ -457,6 +472,18 @@ class TestRemoteStation(unittest.TestCase, MatchAddressMixin):
self.match_address(test_addr, 4, 1, 1, 'fe')
assert str(test_addr) == "1:254@1.2.3.4"
test_addr = RemoteStation(1, 254, route=Address("1.2.3.4:47809"))
self.match_address(test_addr, 4, 1, 1, 'fe')
assert str(test_addr) == "1:254@1.2.3.4:47809"
test_addr = RemoteStation(1, 254, route=Address("0x01020304BAC0"))
self.match_address(test_addr, 4, 1, 1, 'fe')
assert str(test_addr) == "1:254@1.2.3.4"
test_addr = RemoteStation(1, 254, route=Address("0x01020304BAC1"))
self.match_address(test_addr, 4, 1, 1, 'fe')
assert str(test_addr) == "1:254@1.2.3.4:47809"
# test station address
with self.assertRaises(ValueError):
RemoteStation(1, -1)
@ -466,10 +493,6 @@ class TestRemoteStation(unittest.TestCase, MatchAddressMixin):
def test_remote_station_bytes_routed(self):
if _debug: TestRemoteStation._debug("test_remote_station_bytes_routed")
if not settings.route_aware:
if _debug: TestRemoteStation._debug(" - not route aware")
return
# multi-byte strings are hex encoded
test_addr = RemoteStation(1, xtob('0102'), route=Address("1.2.3.4"))
self.match_address(test_addr, 4, 1, 2, '0102')
@ -479,11 +502,34 @@ class TestRemoteStation(unittest.TestCase, MatchAddressMixin):
self.match_address(test_addr, 4, 1, 3, '010203')
assert str(test_addr) == "1:0x010203@1.2.3.4"
test_addr = RemoteStation(1, xtob('010203'), route=Address("1.2.3.4:47809"))
self.match_address(test_addr, 4, 1, 3, '010203')
assert str(test_addr) == "1:0x010203@1.2.3.4:47809"
test_addr = RemoteStation(1, xtob('010203'), route=Address("0x01020304BAC0"))
self.match_address(test_addr, 4, 1, 3, '010203')
assert str(test_addr) == "1:0x010203@1.2.3.4"
test_addr = RemoteStation(1, xtob('010203'), route=Address("0x01020304BAC1"))
self.match_address(test_addr, 4, 1, 3, '010203')
assert str(test_addr) == "1:0x010203@1.2.3.4:47809"
# match with an IPv4 address
test_addr = RemoteStation(1, xtob('01020304bac0'), route=Address("1.2.3.4"))
self.match_address(test_addr, 4, 1, 6, '01020304bac0')
assert str(test_addr) == "1:1.2.3.4@1.2.3.4"
test_addr = RemoteStation(1, xtob('01020304bac0'), route=Address("1.2.3.4:47809"))
self.match_address(test_addr, 4, 1, 6, '01020304bac0')
assert str(test_addr) == "1:1.2.3.4@1.2.3.4:47809"
test_addr = RemoteStation(1, xtob('01020304bac0'), route=Address("0x01020304BAC0"))
self.match_address(test_addr, 4, 1, 6, '01020304bac0')
assert str(test_addr) == "1:1.2.3.4@1.2.3.4"
test_addr = RemoteStation(1, xtob('01020304bac0'), route=Address("0x01020304BAC1"))
self.match_address(test_addr, 4, 1, 6, '01020304bac0')
assert str(test_addr) == "1:1.2.3.4@1.2.3.4:47809"
@bacpypes_debugging
class TestLocalBroadcast(unittest.TestCase, MatchAddressMixin):
@ -498,10 +544,6 @@ class TestLocalBroadcast(unittest.TestCase, MatchAddressMixin):
def test_local_broadcast_routed(self):
if _debug: TestLocalBroadcast._debug("test_local_broadcast_routed")
if not settings.route_aware:
if _debug: TestLocalBroadcast._debug(" - not route aware")
return
test_addr = LocalBroadcast(route=Address("1.2.3.4"))
self.match_address(test_addr, 1, None, None, None)
assert str(test_addr) == "*@1.2.3.4"
@ -532,10 +574,6 @@ class TestRemoteBroadcast(unittest.TestCase, MatchAddressMixin):
def test_remote_broadcast_routed(self):
if _debug: TestRemoteBroadcast._debug("test_remote_broadcast_routed")
if not settings.route_aware:
if _debug: TestRemoteBroadcast._debug(" - not route aware")
return
# match
test_addr = RemoteBroadcast(1, route=Address("1.2.3.4"))
self.match_address(test_addr, 3, 1, None, None)
@ -555,14 +593,13 @@ class TestGlobalBroadcast(unittest.TestCase, MatchAddressMixin):
def test_global_broadcast_routed(self):
if _debug: TestGlobalBroadcast._debug("test_global_broadcast_routed")
if not settings.route_aware:
if _debug: TestGlobalBroadcast._debug(" - not route aware")
return
test_addr = GlobalBroadcast(route=Address("1.2.3.4"))
self.match_address(test_addr, 5, None, None, None)
assert str(test_addr) == "*:*@1.2.3.4"
test_addr = GlobalBroadcast(route=Address("1.2.3.4:47809"))
self.match_address(test_addr, 5, None, None, None)
assert str(test_addr) == "*:*@1.2.3.4:47809"
@bacpypes_debugging
class TestAddressEquality(unittest.TestCase, MatchAddressMixin):
@ -581,13 +618,23 @@ class TestAddressEquality(unittest.TestCase, MatchAddressMixin):
def test_address_equality_str_routed(self):
if _debug: TestAddressEquality._debug("test_address_equality_str_routed")
if not settings.route_aware:
if _debug: TestAddressEquality._debug(" - not route aware")
return
assert Address("3:4@6.7.8.9") == RemoteStation(3, 4, route=Address("6.7.8.9"))
assert Address("3:4@0x06070809BAC0") == RemoteStation(3, 4, route=Address("6.7.8.9"))
assert Address("3:4@6.7.8.9:47809") == RemoteStation(3, 4, route=Address("6.7.8.9:47809"))
assert Address("3:4@0x06070809BAC1") == RemoteStation(3, 4, route=Address("6.7.8.9:47809"))
assert Address("5:*@6.7.8.9") == RemoteBroadcast(5, route=Address("6.7.8.9"))
assert Address("5:*@0x06070809BAC0") == RemoteBroadcast(5, route=Address("6.7.8.9"))
assert Address("5:*@6.7.8.9:47809") == RemoteBroadcast(5, route=Address("6.7.8.9:47809"))
assert Address("5:*@0x06070809BAC1") == RemoteBroadcast(5, route=Address("6.7.8.9:47809"))
assert Address("*:*@6.7.8.9") == GlobalBroadcast(route=Address("6.7.8.9"))
assert Address("*:*@0x06070809BAC0") == GlobalBroadcast(route=Address("6.7.8.9"))
assert Address("*:*@6.7.8.9:47809") == GlobalBroadcast(route=Address("6.7.8.9:47809"))
assert Address("*:*@0x06070809BAC1") == GlobalBroadcast(route=Address("6.7.8.9:47809"))
def test_address_equality_unicode(self):
if _debug: TestAddressEquality._debug("test_address_equality_unicode")