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

LocalDeviceObject shuffling, automatic protocolServicesSupported

Browse Source
This commit is contained in:
Joel Bender 2018-02-24 00:37:04 -05:00
parent b700b09856
commit c25b086211
50 changed files with 61 additions and 537 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
py27/bacpypes/__init__.py
View File

@ -69,6 +69,8 @@ from . import apdu
from . import app
from . import appservice
from . import local
from . import service
#

122
py27/bacpypes/service/device.py
View File

@ -4,136 +4,16 @@ from ..debugging import bacpypes_debugging, ModuleLogger
from ..capability import Capability
from ..pdu import GlobalBroadcast
from ..primitivedata import Date, Time, ObjectIdentifier
from ..constructeddata import ArrayOf
from ..apdu import WhoIsRequest, IAmRequest, IHaveRequest, SimpleAckPDU, Error
from ..apdu import WhoIsRequest, IAmRequest, IHaveRequest, SimpleAckPDU
from ..errors import ExecutionError, InconsistentParameters, \
MissingRequiredParameter, ParameterOutOfRange
from ..object import register_object_type, registered_object_types, \
Property, DeviceObject
from ..task import FunctionTask
from .object import CurrentPropertyListMixIn
# some debugging
_debug = 0
_log = ModuleLogger(globals())
#
# CurrentDateProperty
#
class CurrentDateProperty(Property):
def __init__(self, identifier):
Property.__init__(self, identifier, Date, default=(), optional=True, mutable=False)
def ReadProperty(self, obj, arrayIndex=None):
# access an array
if arrayIndex is not None:
raise TypeError("{0} is unsubscriptable".format(self.identifier))
# get the value
now = Date()
now.now()
return now.value
def WriteProperty(self, obj, value, arrayIndex=None, priority=None, direct=False):
raise ExecutionError(errorClass='property', errorCode='writeAccessDenied')
#
# CurrentTimeProperty
#
class CurrentTimeProperty(Property):
def __init__(self, identifier):
Property.__init__(self, identifier, Time, default=(), optional=True, mutable=False)
def ReadProperty(self, obj, arrayIndex=None):
# access an array
if arrayIndex is not None:
raise TypeError("{0} is unsubscriptable".format(self.identifier))
# get the value
now = Time()
now.now()
return now.value
def WriteProperty(self, obj, value, arrayIndex=None, priority=None, direct=False):
raise ExecutionError(errorClass='property', errorCode='writeAccessDenied')
#
# LocalDeviceObject
#
@bacpypes_debugging
class LocalDeviceObject(CurrentPropertyListMixIn, DeviceObject):
properties = \
[ CurrentTimeProperty('localTime')
, CurrentDateProperty('localDate')
]
defaultProperties = \
{ 'maxApduLengthAccepted': 1024
, 'segmentationSupported': 'segmentedBoth'
, 'maxSegmentsAccepted': 16
, 'apduSegmentTimeout': 5000
, 'apduTimeout': 3000
, 'numberOfApduRetries': 3
}
def __init__(self, **kwargs):
if _debug: LocalDeviceObject._debug("__init__ %r", kwargs)
# fill in default property values not in kwargs
for attr, value in LocalDeviceObject.defaultProperties.items():
if attr not in kwargs:
kwargs[attr] = value
for key, value in kwargs.items():
if key.startswith("_"):
setattr(self, key, value)
del kwargs[key]
# check for registration
if self.__class__ not in registered_object_types.values():
if 'vendorIdentifier' not in kwargs:
raise RuntimeError("vendorIdentifier required to auto-register the LocalDeviceObject class")
register_object_type(self.__class__, vendor_id=kwargs['vendorIdentifier'])
# check for properties this class implements
if 'localDate' in kwargs:
raise RuntimeError("localDate is provided by LocalDeviceObject and cannot be overridden")
if 'localTime' in kwargs:
raise RuntimeError("localTime is provided by LocalDeviceObject and cannot be overridden")
# the object identifier is required for the object list
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")
kwargs['objectList'] = ArrayOf(ObjectIdentifier)([object_identifier])
# check for a minimum value
if kwargs['maxApduLengthAccepted'] < 50:
raise ValueError("invalid max APDU length accepted")
# dump the updated attributes
if _debug: LocalDeviceObject._debug(" - updated kwargs: %r", kwargs)
# proceed as usual
super(LocalDeviceObject, self).__init__(**kwargs)
#
# Who-Is I-Am Services
#

55
py27/bacpypes/service/object.py
View File

@ -7,11 +7,11 @@ from ..basetypes import ErrorType, PropertyIdentifier
from ..primitivedata import Atomic, Null, Unsigned
from ..constructeddata import Any, Array, ArrayOf
from ..apdu import Error, \
from ..apdu import \
SimpleAckPDU, ReadPropertyACK, ReadPropertyMultipleACK, \
ReadAccessResult, ReadAccessResultElement, ReadAccessResultElementChoice
from ..errors import ExecutionError
from ..object import Property, Object, PropertyError
from ..object import PropertyError
# some debugging
_debug = 0
@ -20,57 +20,6 @@ _log = ModuleLogger(globals())
# handy reference
ArrayOfPropertyIdentifier = ArrayOf(PropertyIdentifier)
#
# CurrentPropertyList
#
@bacpypes_debugging
class CurrentPropertyList(Property):
def __init__(self):
if _debug: CurrentPropertyList._debug("__init__")
Property.__init__(self, 'propertyList', ArrayOfPropertyIdentifier, default=None, optional=True, mutable=False)
def ReadProperty(self, obj, arrayIndex=None):
if _debug: CurrentPropertyList._debug("ReadProperty %r %r", obj, arrayIndex)
# make a list of the properties that have values
property_list = [k for k, v in obj._values.items()
if v is not None
and k not in ('objectName', 'objectType', 'objectIdentifier', 'propertyList')
]
if _debug: CurrentPropertyList._debug(" - property_list: %r", property_list)
# sort the list so it's stable
property_list.sort()
# asking for the whole thing
if arrayIndex is None:
return ArrayOfPropertyIdentifier(property_list)
# asking for the length
if arrayIndex == 0:
return len(property_list)
# asking for an index
if arrayIndex > len(property_list):
raise ExecutionError(errorClass='property', errorCode='invalidArrayIndex')
return property_list[arrayIndex - 1]
def WriteProperty(self, obj, value, arrayIndex=None, priority=None, direct=False):
raise ExecutionError(errorClass='property', errorCode='writeAccessDenied')
#
# CurrentPropertyListMixIn
#
@bacpypes_debugging
class CurrentPropertyListMixIn(Object):
properties = [
CurrentPropertyList(),
]
#
# ReadProperty and WriteProperty Services
#

9
samples/AccumulatorObject.py
View File

@ -15,7 +15,7 @@ from bacpypes.basetypes import DateTime, Scale
from bacpypes.object import AccumulatorObject
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
from bacpypes.service.object import ReadWritePropertyMultipleServices
# some debugging
@ -84,13 +84,6 @@ def main():
# add the additional service
this_application.add_capability(ReadWritePropertyMultipleServices)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a random input object
accumulator = AccumulatorObject(
objectIdentifier=('accumulator', 1),

3
samples/BBMD2VLANRouter.py
View File

@ -24,7 +24,8 @@ from bacpypes.bvllservice import BIPBBMD, AnnexJCodec, UDPMultiplexer
from bacpypes.app import Application
from bacpypes.appservice import StateMachineAccessPoint, ApplicationServiceAccessPoint
from bacpypes.service.device import LocalDeviceObject, WhoIsIAmServices
from bacpypes.local.device import LocalDeviceObject
from bacpypes.service.device import WhoIsIAmServices
from bacpypes.service.object import ReadWritePropertyServices
from bacpypes.primitivedata import Real

9
samples/COVClient.py
View File

@ -20,7 +20,7 @@ from bacpypes.apdu import SubscribeCOVRequest, \
SimpleAckPDU, RejectPDU, AbortPDU
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -228,13 +228,6 @@ def main():
# make a simple application
this_application = SubscribeCOVApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = SubscribeCOVConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

2
samples/COVServer.py
View File

@ -18,7 +18,7 @@ from bacpypes.task import RecurringTask
from bacpypes.app import BIPSimpleApplication
from bacpypes.object import AnalogValueObject, BinaryValueObject
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
from bacpypes.service.cov import ChangeOfValueServices
# some debugging

4
samples/CommandableMixin.py
View File

@ -26,8 +26,8 @@ from bacpypes.object import Property, ReadableProperty, WritableProperty, \
TimeValueObject, TimePatternValueObject, ChannelObject
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.object import CurrentPropertyListMixIn
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.object import CurrentPropertyListMixIn
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0

9
samples/DeviceCommunicationControl.py
View File

@ -18,7 +18,7 @@ from bacpypes.pdu import Address
from bacpypes.apdu import DeviceCommunicationControlRequest
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -134,13 +134,6 @@ def main():
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = DCCConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/DeviceDiscovery.py
View File

@ -22,7 +22,7 @@ from bacpypes.basetypes import ServicesSupported
from bacpypes.errors import DecodingError
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 1
@ -211,13 +211,6 @@ def main():
# make a simple application
this_application = DiscoveryApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = DiscoveryConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/DeviceDiscoveryForeign.py
View File

@ -22,7 +22,7 @@ from bacpypes.basetypes import ServicesSupported
from bacpypes.errors import DecodingError
from bacpypes.app import BIPForeignApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 1
@ -215,13 +215,6 @@ def main():
int(args.ini.foreignttl),
)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = DiscoveryConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/HTTPServer.py
View File

@ -22,7 +22,7 @@ from bacpypes.apdu import ReadPropertyRequest, WhoIsRequest
from bacpypes.app import BIPSimpleApplication
from bacpypes.object import get_object_class, get_datatype
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -190,13 +190,6 @@ try:
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# local host, special port
HOST, PORT = "", int(args.port)
server = ThreadedTCPServer((HOST, PORT), ThreadedHTTPRequestHandler)

9
samples/HandsOnLab/Sample1_SimpleApplication.py
View File

@ -13,7 +13,7 @@ from bacpypes.consolelogging import ConfigArgumentParser
from bacpypes.core import run
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 1
@ -72,13 +72,6 @@ def main():
this_application = SampleApplication(this_device, args.ini.address)
if _debug: _log.debug(" - this_application: %r", this_application)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
_log.debug("running")
run()

9
samples/HandsOnLab/Sample2_WhoIsIAmApplication.py
View File

@ -15,7 +15,7 @@ from bacpypes.consolelogging import ConfigArgumentParser
from bacpypes.core import run
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -90,13 +90,6 @@ def main():
# make a sample application
this_application = WhoIsIAmApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
_log.debug("running")
run()

9
samples/HandsOnLab/Sample3_WhoHasIHaveApplication.py
View File

@ -16,7 +16,7 @@ from bacpypes.consolelogging import ConfigArgumentParser
from bacpypes.core import run
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -92,13 +92,6 @@ def main():
# make a sample application
this_application = WhoHasIHaveApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
_log.debug("running")
# run until stopped, ^C works

9
samples/HandsOnLab/Sample4_RandomAnalogValueObject.py
View File

@ -21,7 +21,7 @@ from bacpypes.object import AnalogValueObject, Property, register_object_type
from bacpypes.errors import ExecutionError
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -99,13 +99,6 @@ def main():
# make a sample application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make some random input objects
for i in range(1, RANDOM_OBJECT_COUNT+1):
ravo = RandomAnalogValueObject(

11
samples/LocalScheduleObject.py
View File

@ -22,8 +22,8 @@ from bacpypes.basetypes import CalendarEntry, DailySchedule, DateRange, \
from bacpypes.object import register_object_type, get_datatype, WritableProperty, ScheduleObject
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.object import CurrentPropertyListMixIn
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.object import CurrentPropertyListMixIn
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -573,13 +573,6 @@ def main():
# make a sample application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
#
# Simple daily schedule (actually a weekly schedule with every day
# being identical.

9
samples/MultiStateValueObject.py
View File

@ -15,7 +15,7 @@ from bacpypes.constructeddata import ArrayOf
from bacpypes.object import MultiStateValueObject
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -44,13 +44,6 @@ def main():
# make a sample application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a multistate value object
msvo = MultiStateValueObject(
objectIdentifier=('multiStateValue', 1),

9
samples/MultipleReadProperty.py
View File

@ -23,7 +23,7 @@ from bacpypes.primitivedata import Unsigned
from bacpypes.constructeddata import Array
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -142,13 +142,6 @@ def main():
# make a simple application
this_application = ReadPointListApplication(point_list, this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# fire off a request when the core has a chance
deferred(this_application.next_request)

10
samples/MultipleReadPropertyHammer.py
View File

@ -31,7 +31,8 @@ from bacpypes.bvllservice import BIPSimple, AnnexJCodec, UDPMultiplexer
from bacpypes.apdu import ReadPropertyRequest
from bacpypes.service.device import LocalDeviceObject, WhoIsIAmServices
from bacpypes.local.device import LocalDeviceObject
from bacpypes.service.device import WhoIsIAmServices
from bacpypes.service.object import ReadWritePropertyServices
@ -211,13 +212,6 @@ def main():
# make a simple application
this_application = ReadPointListApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
_log.debug("running")
run()

9
samples/MultipleReadPropertyThreaded.py
View File

@ -24,7 +24,7 @@ from bacpypes.primitivedata import Unsigned
from bacpypes.constructeddata import Array
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -134,13 +134,6 @@ def main():
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# create a thread and
read_thread = ReadPointListThread(point_list)
if _debug: _log.debug(" - read_thread: %r", read_thread)

9
samples/RandomAnalogValueSleep.py
View File

@ -22,7 +22,7 @@ from bacpypes.object import AnalogValueObject, Property, register_object_type
from bacpypes.errors import ExecutionError
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -119,13 +119,6 @@ def main():
# make a sample application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make some random input objects
for i in range(1, RANDOM_OBJECT_COUNT+1):
ravo = RandomAnalogValueObject(

9
samples/ReadAllProperties.py
View File

@ -23,7 +23,7 @@ from bacpypes.constructeddata import Array
from bacpypes.app import BIPSimpleApplication
from bacpypes.object import get_object_class, get_datatype
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -168,13 +168,6 @@ def main():
# make a simple application
this_application = ReadPropertyApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# fire off a request when the core has a chance
deferred(this_application.next_request)

9
samples/ReadProperty.py
View File

@ -22,7 +22,7 @@ from bacpypes.constructeddata import Array
from bacpypes.app import BIPSimpleApplication
from bacpypes.object import get_object_class, get_datatype
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -162,13 +162,6 @@ def main():
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = ReadPropertyConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/ReadProperty25.py
View File

@ -22,7 +22,7 @@ from bacpypes.constructeddata import Array
from bacpypes.app import BIPSimpleApplication
from bacpypes.object import get_object_class, get_datatype
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -164,13 +164,6 @@ def main():
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = ReadPropertyConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/ReadPropertyAny.py
View File

@ -23,7 +23,7 @@ from bacpypes.apdu import ReadPropertyRequest, Error, AbortPDU, ReadPropertyACK
from bacpypes.primitivedata import Tag
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
@ -136,13 +136,6 @@ def main():
this_application = BIPSimpleApplication(this_device, args.ini.address)
if _debug: _log.debug(" - this_application: %r", this_application)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = ReadPropertyAnyConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/ReadPropertyMultiple.py
View File

@ -25,7 +25,7 @@ from bacpypes.constructeddata import Array
from bacpypes.basetypes import PropertyIdentifier
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -214,13 +214,6 @@ def main():
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = ReadPropertyMultipleConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/ReadPropertyMultiple25.py
View File

@ -25,7 +25,7 @@ from bacpypes.constructeddata import Array
from bacpypes.basetypes import PropertyIdentifier
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -220,13 +220,6 @@ def main():
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = ReadPropertyMultipleConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/ReadPropertyMultipleServer.py
View File

@ -18,7 +18,7 @@ from bacpypes.object import AnalogValueObject, Property, register_object_type
from bacpypes.errors import ExecutionError
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject, DeviceCommunicationControlServices
from bacpypes.service.device import DeviceCommunicationControlServices
from bacpypes.service.object import ReadWritePropertyMultipleServices
# some debugging
@ -124,13 +124,6 @@ def main():
this_application.add_object(ravo2)
_log.debug(" - object list: %r", this_device.objectList)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
_log.debug("running")
run()

11
samples/ReadPropertyMultipleServer25.py
View File

@ -18,8 +18,8 @@ from bacpypes.object import AnalogValueObject, Property, register_object_type
from bacpypes.errors import ExecutionError
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import ReadWritePropertyMultipleServices, \
LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
from bacpypes.service.device import ReadWritePropertyMultipleServices
# some debugging
_debug = 0
@ -122,13 +122,6 @@ def main():
this_application.add_object(ravo2)
_log.debug(" - object list: %r", this_device.objectList)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
_log.debug("running")
run()

9
samples/ReadRange.py
View File

@ -20,7 +20,7 @@ from bacpypes.object import get_object_class, get_datatype
from bacpypes.apdu import ReadRangeRequest, ReadRangeACK
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -132,13 +132,6 @@ def main():
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = ReadRangeConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/ReadWriteEventMessageTexts.py
View File

@ -23,7 +23,7 @@ from bacpypes.primitivedata import Unsigned, CharacterString
from bacpypes.constructeddata import Array, ArrayOf, Any
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -241,13 +241,6 @@ def main():
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = ReadWritePropertyConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/ReadWriteFile.py
View File

@ -31,7 +31,7 @@ from bacpypes.apdu import Error, AbortPDU, \
AtomicWriteFileACK
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -322,13 +322,6 @@ def main():
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = TestConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

13
samples/ReadWriteFileServer.py
View File

@ -15,9 +15,9 @@ from bacpypes.consolelogging import ConfigArgumentParser
from bacpypes.core import run
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.service.file import FileServices, \
LocalRecordAccessFileObject, LocalStreamAccessFileObject
from bacpypes.local.device import LocalDeviceObject
from bacpypes.local.file import FileServices, LocalRecordAccessFileObject, LocalStreamAccessFileObject
from bacpypes.service.file import FileServices
# some debugging
_debug = 0
@ -188,13 +188,6 @@ def main():
# add the capability to server file content
this_application.add_capability(FileServices)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a record access file, add to the device
f1 = TestRecordFile(
objectIdentifier=('file', 1),

9
samples/ReadWriteProperty.py
View File

@ -27,7 +27,7 @@ from bacpypes.primitivedata import Null, Atomic, Boolean, Unsigned, Integer, \
from bacpypes.constructeddata import Array, Any, AnyAtomic
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -282,13 +282,6 @@ def main():
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = ReadWritePropertyConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/RecurringMultipleReadProperty.py
View File

@ -24,7 +24,7 @@ from bacpypes.primitivedata import Unsigned
from bacpypes.constructeddata import Array
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -176,13 +176,6 @@ def main():
this_application = PrairieDog(args.interval, this_device, args.ini.address)
if _debug: _log.debug(" - this_application: %r", this_application)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
_log.debug("running")
run()

9
samples/ThreadedReadProperty.py
View File

@ -24,7 +24,7 @@ from bacpypes.primitivedata import Unsigned
from bacpypes.constructeddata import Array
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -172,13 +172,6 @@ def main():
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
thread_list = []
# loop through the address and point lists

9
samples/Tutorial/SampleConsoleCmd-Answer.py
View File

@ -15,7 +15,7 @@ from bacpypes.consolecmd import ConsoleCmd
from bacpypes.core import run, enable_sleeping
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -112,13 +112,6 @@ def main():
# make a sample application
this_application = SampleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = SampleConsoleCmd()

9
samples/Tutorial/SampleConsoleCmd.py
View File

@ -15,7 +15,7 @@ from bacpypes.consolecmd import ConsoleCmd
from bacpypes.core import run, enable_sleeping
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -84,13 +84,6 @@ def main():
# make a sample application
this_application = SampleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = SampleConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

19
samples/Tutorial/WhoIsIAm.py
View File

@ -20,7 +20,7 @@ from bacpypes.basetypes import ServicesSupported
from bacpypes.errors import DecodingError
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 1
@ -167,26 +167,9 @@ def main():
vendorIdentifier=int(args.ini.vendoridentifier),
)
# build a bit string that knows about the bit names
pss = ServicesSupported()
pss['whoIs'] = 1
pss['iAm'] = 1
pss['readProperty'] = 1
pss['writeProperty'] = 1
# set the property value to be just the bits
this_device.protocolServicesSupported = pss.value
# make a simple application
this_application = WhoIsIAmApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = WhoIsIAmConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/VendorAVObject.py
View File

@ -20,7 +20,7 @@ from bacpypes.object import AnalogValueObject, Property, register_object_type
from bacpypes.errors import ExecutionError
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -106,13 +106,6 @@ def main():
# make a sample application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make some objects
ravo1 = VendorAVObject(
objectIdentifier=(513, 1), objectName='Random1'

9
samples/VendorReadWriteProperty.py
View File

@ -26,7 +26,7 @@ from bacpypes.primitivedata import Tag, Null, Atomic, Integer, Unsigned, Real
from bacpypes.constructeddata import Array, Any
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
import VendorAVObject
@ -254,13 +254,6 @@ def main():
# make a simple application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = ReadWritePropertyConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

19
samples/WhoIsIAm.py
View File

@ -21,7 +21,7 @@ from bacpypes.basetypes import ServicesSupported
from bacpypes.errors import DecodingError
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -189,29 +189,12 @@ def main():
)
if _debug: _log.debug(" - this_device: %r", this_device)
# build a bit string that knows about the bit names
pss = ServicesSupported()
pss['whoIs'] = 1
pss['iAm'] = 1
pss['readProperty'] = 1
pss['writeProperty'] = 1
# set the property value to be just the bits
this_device.protocolServicesSupported = pss.value
# make a simple application
this_application = WhoIsIAmApplication(
this_device, args.ini.address,
)
if _debug: _log.debug(" - this_application: %r", this_application)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = WhoIsIAmConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

19
samples/WhoIsIAmForeign.py
View File

@ -31,7 +31,7 @@ from bacpypes.basetypes import ServicesSupported
from bacpypes.errors import DecodingError
from bacpypes.app import BIPForeignApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
@ -200,16 +200,6 @@ def main():
)
if _debug: _log.debug(" - this_device: %r", this_device)
# build a bit string that knows about the bit names
pss = ServicesSupported()
pss['whoIs'] = 1
pss['iAm'] = 1
pss['readProperty'] = 1
pss['writeProperty'] = 1
# set the property value to be just the bits
this_device.protocolServicesSupported = pss.value
# make a simple application
this_application = WhoIsIAmApplication(
this_device, args.ini.address,
@ -218,13 +208,6 @@ def main():
)
if _debug: _log.debug(" - this_application: %r", this_application)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = WhoIsIAmConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

9
samples/WriteSomething.py
View File

@ -19,7 +19,7 @@ from bacpypes.iocb import IOCB
from bacpypes.pdu import Address
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
from bacpypes.primitivedata import TagList, OpeningTag, ClosingTag, ContextTag
from bacpypes.constructeddata import Any
@ -130,13 +130,6 @@ def main():
this_application = BIPSimpleApplication(this_device, args.ini.address)
if _debug: _log.debug(" - this_application: %r", this_application)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make a console
this_console = WriteSomethingConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)

4
sandbox/local_schedule_object_t1.py
View File

@ -22,8 +22,8 @@ from bacpypes.basetypes import CalendarEntry, DailySchedule, DateRange, \
from bacpypes.object import register_object_type, get_datatype, WritableProperty, ScheduleObject
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.object import CurrentPropertyListMixIn
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.object import CurrentPropertyListMixIn
from bacpypes.local.device import LocalDeviceObject
### testing
import time

11
sandbox/local_schedule_object_t2.py
View File

@ -23,8 +23,8 @@ from bacpypes.object import register_object_type, get_datatype, \
WritableProperty, ScheduleObject, AnalogValueObject
from bacpypes.app import Application
from bacpypes.service.object import CurrentPropertyListMixIn
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.object import CurrentPropertyListMixIn
from bacpypes.local.device import LocalDeviceObject
### testing
import time
@ -661,13 +661,6 @@ def main():
# make a floating application, no network interface
this_application = Application(this_device)
# get the services supported
services_supported = this_application.get_services_supported()
if _debug: _log.debug(" - services_supported: %r", services_supported)
# let the device object know
this_device.protocolServicesSupported = services_supported.value
#
# Simple daily schedule (actually a weekly schedule with every day
# being identical.

1
setup.py
View File

@ -55,6 +55,7 @@ setup(
url="https://github.com/JoelBender/bacpypes",
packages=[
'bacpypes',
'bacpypes.local',
'bacpypes.service',
],
package_dir={

2
tests/test_service/helpers.py
View File

@ -13,7 +13,7 @@ from bacpypes.vlan import Network, Node
from bacpypes.app import Application
from bacpypes.appservice import StateMachineAccessPoint, ApplicationServiceAccessPoint
from bacpypes.netservice import NetworkServiceAccessPoint, NetworkServiceElement
from bacpypes.service.device import LocalDeviceObject
from bacpypes.local.device import LocalDeviceObject
from ..state_machine import StateMachine, StateMachineGroup
from ..time_machine import reset_time_machine, run_time_machine

2
tests/test_service/test_object.py
View File

@ -16,7 +16,7 @@ from bacpypes.constructeddata import ArrayOf
from bacpypes.object import register_object_type, ReadableProperty, \
WritableProperty, Object
from bacpypes.service.object import CurrentPropertyListMixIn
from bacpypes.local.object import CurrentPropertyListMixIn
# some debugging
_debug = 0