mirror/bacpypes
1
0
Fork 0
mirror of https://github.com/JoelBender/bacpypes synced 2025-10-27 00:57:47 +08:00
Code Issues Releases Wiki Activity

bring up-to-date with the 0.16.7 release

Browse Source
This commit is contained in:
Joel Bender
2018-02-16 23:15:56 -05:00
parent 45874f08bf
commit 55268cb508
6 changed files with 343 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
py25/bacpypes/bvllservice.py
View File

@@ -407,6 +407,54 @@ class BIPSimple(BIPSAP, Client, Server):
# send it upstream
self.response(xpdu)
elif isinstance(pdu, WriteBroadcastDistributionTable):
# build a response
xpdu = Result(code=0x0010, user_data=pdu.pduUserData)
xpdu.pduDestination = pdu.pduSource
# send it downstream
self.request(xpdu)
elif isinstance(pdu, ReadBroadcastDistributionTable):
# build a response
xpdu = Result(code=0x0020, user_data=pdu.pduUserData)
xpdu.pduDestination = pdu.pduSource
# send it downstream
self.request(xpdu)
elif isinstance(pdu, RegisterForeignDevice):
# build a response
xpdu = Result(code=0x0030, user_data=pdu.pduUserData)
xpdu.pduDestination = pdu.pduSource
# send it downstream
self.request(xpdu)
elif isinstance(pdu, ReadForeignDeviceTable):
# build a response
xpdu = Result(code=0x0040, user_data=pdu.pduUserData)
xpdu.pduDestination = pdu.pduSource
# send it downstream
self.request(xpdu)
elif isinstance(pdu, DeleteForeignDeviceTableEntry):
# build a response
xpdu = Result(code=0x0050, user_data=pdu.pduUserData)
xpdu.pduDestination = pdu.pduSource
# send it downstream
self.request(xpdu)
elif isinstance(pdu, DistributeBroadcastToNetwork):
# build a response
xpdu = Result(code=0x0060, user_data=pdu.pduUserData)
xpdu.pduDestination = pdu.pduSource
# send it downstream
self.request(xpdu)
else:
BIPSimple._warning("invalid pdu type: %s", type(pdu))
@@ -446,11 +494,6 @@ class BIPForeign(BIPSAP, Client, Server, OneShotTask, DebugContents):
def indication(self, pdu):
if _debug: BIPForeign._debug("indication %r", pdu)
# check the BBMD registration status, we may not be registered
if self.registrationStatus != 0:
if _debug: BIPForeign._debug(" - packet dropped, unregistered")
return
# check for local stations
if pdu.pduDestination.addrType == Address.localStationAddr:
# make an original unicast PDU
@@ -462,6 +505,11 @@ class BIPForeign(BIPSAP, Client, Server, OneShotTask, DebugContents):
# check for broadcasts
elif pdu.pduDestination.addrType == Address.localBroadcastAddr:
# check the BBMD registration status, we may not be registered
if self.registrationStatus != 0:
if _debug: BIPForeign._debug(" - packet dropped, unregistered")
return
# make an original broadcast PDU
xpdu = DistributeBroadcastToNetwork(pdu, user_data=pdu.pduUserData)
xpdu.pduDestination = self.bbmdAddress
@@ -498,6 +546,13 @@ class BIPForeign(BIPSAP, Client, Server, OneShotTask, DebugContents):
return
elif isinstance(pdu, OriginalUnicastNPDU):
# build a vanilla PDU
xpdu = PDU(pdu.pduData, source=pdu.pduSource, destination=pdu.pduDestination, user_data=pdu.pduUserData)
# send it upstream
self.response(xpdu)
# check the BBMD registration status, we may not be registered
if self.registrationStatus != 0:
if _debug: BIPForeign._debug(" - packet dropped, unregistered")
@@ -511,13 +566,6 @@ class BIPForeign(BIPSAP, Client, Server, OneShotTask, DebugContents):
# send this to the service access point
self.sap_response(pdu)
elif isinstance(pdu, OriginalUnicastNPDU):
# build a vanilla PDU
xpdu = PDU(pdu.pduData, source=pdu.pduSource, destination=pdu.pduDestination, user_data=pdu.pduUserData)
# send it upstream
self.response(xpdu)
elif isinstance(pdu, ForwardedNPDU):
# build a PDU with the source from the real source
xpdu = PDU(pdu.pduData, source=pdu.bvlciAddress, destination=LocalBroadcast(), user_data=pdu.pduUserData)
@@ -525,6 +573,54 @@ class BIPForeign(BIPSAP, Client, Server, OneShotTask, DebugContents):
# send it upstream
self.response(xpdu)
elif isinstance(pdu, WriteBroadcastDistributionTable):
# build a response
xpdu = Result(code=0x0010, user_data=pdu.pduUserData)
xpdu.pduDestination = pdu.pduSource
# send it downstream
self.request(xpdu)
elif isinstance(pdu, ReadBroadcastDistributionTable):
# build a response
xpdu = Result(code=0x0020, user_data=pdu.pduUserData)
xpdu.pduDestination = pdu.pduSource
# send it downstream
self.request(xpdu)
elif isinstance(pdu, RegisterForeignDevice):
# build a response
xpdu = Result(code=0x0030, user_data=pdu.pduUserData)
xpdu.pduDestination = pdu.pduSource
# send it downstream
self.request(xpdu)
elif isinstance(pdu, ReadForeignDeviceTable):
# build a response
xpdu = Result(code=0x0040, user_data=pdu.pduUserData)
xpdu.pduDestination = pdu.pduSource
# send it downstream
self.request(xpdu)
elif isinstance(pdu, DeleteForeignDeviceTableEntry):
# build a response
xpdu = Result(code=0x0050, user_data=pdu.pduUserData)
xpdu.pduDestination = pdu.pduSource
# send it downstream
self.request(xpdu)
elif isinstance(pdu, DistributeBroadcastToNetwork):
# build a response
xpdu = Result(code=0x0060, user_data=pdu.pduUserData)
xpdu.pduDestination = pdu.pduSource
# send it downstream
self.request(xpdu)
else:
BIPForeign._warning("invalid pdu type: %s", type(pdu))

10
py25/bacpypes/service/device.py
View File

@@ -113,13 +113,15 @@ class LocalDeviceObject(CurrentPropertyListMixIn, DeviceObject):
if 'objectIdentifier' not in kwargs:
raise RuntimeError("objectIdentifier is required")
# coerce the object identifier
object_identifier = kwargs['objectIdentifier']
if isinstance(object_identifier, (int, long)):
object_identifier = ('device', object_identifier)
# the object list is provided
if 'objectList' in kwargs:
raise RuntimeError("objectList is provided by LocalDeviceObject and cannot be overridden")
else:
kwargs['objectList'] = ArrayOf(ObjectIdentifier)([
kwargs['objectIdentifier'],
])
kwargs['objectList'] = ArrayOf(ObjectIdentifier)([object_identifier])
# check for a minimum value
if kwargs['maxApduLengthAccepted'] < 50: