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:14:21 -05:00
parent 1c1ebce4b8
commit f23dd4fef4
15 changed files with 1023 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
py25/bacpypes/app.py
View File

@@ -473,7 +473,7 @@ class BIPSimpleApplication(ApplicationIOController, WhoIsIAmServices, ReadWriteP
def __init__(self, localDevice, localAddress, deviceInfoCache=None, aseID=None):
if _debug: BIPSimpleApplication._debug("__init__ %r %r deviceInfoCache=%r aseID=%r", localDevice, localAddress, deviceInfoCache, aseID)
ApplicationIOController.__init__(self, localDevice, deviceInfoCache, aseID=aseID)
ApplicationIOController.__init__(self, localDevice, localAddress, deviceInfoCache, aseID=aseID)
# local address might be useful for subclasses
if isinstance(localAddress, Address):
@@ -528,9 +528,14 @@ bacpypes_debugging(BIPSimpleApplication)
class BIPForeignApplication(ApplicationIOController, WhoIsIAmServices, ReadWritePropertyServices):
def __init__(self, localDevice, localAddress, bbmdAddress, bbmdTTL, aseID=None):
if _debug: BIPForeignApplication._debug("__init__ %r %r %r %r aseID=%r", localDevice, localAddress, bbmdAddress, bbmdTTL, aseID)
ApplicationIOController.__init__(self, localDevice, aseID=aseID)
def __init__(self, localDevice, localAddress, bbmdAddress, bbmdTTL, deviceInfoCache=None, aseID=None):
if _debug:
BIPForeignApplication._debug(
"__init__ %r %r %r %r deviceInfoCache=%r aseID=%r",
localDevice, localAddress, bbmdAddress, bbmdTTL,
deviceInfoCache, aseID,
)
ApplicationIOController.__init__(self, localDevice, localAddress, deviceInfoCache, aseID=aseID)
# local address might be useful for subclasses
if isinstance(localAddress, Address):

41
py25/bacpypes/basetypes.py
View File

@@ -1604,6 +1604,16 @@ class ErrorType(Sequence):
, Element('errorCode', ErrorCode)
]
class LightingCommand(Sequence):
sequenceElements = \
[ Element('operation', LightingOperation, 0)
, Element('targetLevel', Real, 1, True)
, Element('rampRate', Real, 2, True)
, Element('stepIncrement', Real, 3, True)
, Element('fadeTime', Unsigned, 4, True)
, Element('priority', Unsigned, 5, True)
]
class ObjectPropertyReference(Sequence):
sequenceElements = \
[ Element('objectIdentifier', ObjectIdentifier, 0)
@@ -1792,8 +1802,21 @@ class CalendarEntry(Choice):
]
class ChannelValue(Choice):
choiceElements = [
### needs help
choiceElements = \
[ Element('null', Null)
, Element('real', Real)
, Element('enumerated', Enumerated)
, Element('unsigned', Unsigned)
, Element('boolean', Boolean)
, Element('integer', Integer)
, Element('double', Double)
, Element('time', Time)
, Element('characterString', CharacterString)
, Element('octetString', OctetString)
, Element('bitString', BitString)
, Element('date', Date)
, Element('objectidentifier', ObjectIdentifier)
, Element('lightingCommand', LightingCommand, 0)
]
class ClientCOV(Choice):
@@ -2068,16 +2091,6 @@ class KeyIdentifier(Sequence):
, Element('keyId', Unsigned, 1)
]
class LightingCommand(Sequence):
sequenceElements = \
[ Element('operation', LightingOperation, 0)
, Element('targetLevel', Real, 1) ### optional
, Element('rampRate', Real, 2) ### optional
, Element('stepIncrement', Real, 3) ### optional
, Element('fadeTime', Unsigned, 4) ### optional
, Element('priority', Unsigned, 5) ### optional
]
class LogDataLogData(Choice):
choiceElements = \
[ Element('booleanValue', Boolean, 0)
@@ -2336,14 +2349,14 @@ class PriorityValue(Choice):
, Element('enumerated', Enumerated)
, Element('unsigned', Unsigned)
, Element('boolean', Boolean)
, Element('signed', Integer)
, Element('integer', Integer)
, Element('double', Double)
, Element('time', Time)
, Element('characterString', CharacterString)
, Element('octetString', OctetString)
, Element('bitString', BitString)
, Element('date', Date)
, Element('objectid', ObjectIdentifier)
, Element('objectidentifier', ObjectIdentifier)
, Element('constructedValue', Any, 0)
, Element('datetime', DateTime, 1)
]

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:

6
py25/bacpypes/service/object.py
View File

@@ -24,7 +24,6 @@ ArrayOfPropertyIdentifier = ArrayOf(PropertyIdentifier)
# CurrentPropertyList
#
@bacpypes_debugging
class CurrentPropertyList(Property):
def __init__(self):
@@ -60,17 +59,20 @@ class CurrentPropertyList(Property):
def WriteProperty(self, obj, value, arrayIndex=None, priority=None, direct=False):
raise ExecutionError(errorClass='property', errorCode='writeAccessDenied')
bacpypes_debugging(CurrentPropertyList)
#
# CurrentPropertyListMixIn
#
@bacpypes_debugging
class CurrentPropertyListMixIn(Object):
properties = [
CurrentPropertyList(),
]
bacpypes_debugging(CurrentPropertyListMixIn)
#
# ReadProperty and WriteProperty Services
#