1
0
mirror of https://github.com/JoelBender/bacpypes synced 2025-09-28 22:15:23 +08:00

bring in #88, needs to be verified

This commit is contained in:
Joel Bender 2016-08-25 23:24:30 -04:00
commit b7308b3861
28 changed files with 137 additions and 192 deletions

View File

@ -21,10 +21,6 @@ from bacpypes.basetypes import PriorityValue, PriorityArray
_debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
#
# CommandableMixin
#

View File

@ -63,4 +63,3 @@ def main():
if __name__ == "__main__":
main()

View File

@ -17,10 +17,6 @@ from bacpypes.object import MultiStateValueObject
_debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
#
# __main__
#

View File

@ -26,13 +26,7 @@ from bacpypes.constructeddata import Array
_debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
this_console = None
# point list
# set according to your device
# point list, set according to your device
point_list = [
('1.2.3.4', 'analogValue', 1, 'presentValue'),
('1.2.3.4', 'analogValue', 2, 'presentValue'),

View File

@ -21,10 +21,6 @@ from bacpypes.errors import ExecutionError
_debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
#
# RandomUnsignedValueProperty
#

View File

@ -28,10 +28,6 @@ _log = ModuleLogger(globals())
# settings
RANDOM_OBJECT_COUNT = int(os.getenv('RANDOM_OBJECT_COUNT', 10))
# globals
this_device = None
this_application = None
#
# RandomValueProperty
#
@ -125,6 +121,5 @@ def main():
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -27,7 +27,6 @@ _debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
this_console = None
@ -151,7 +150,9 @@ class ReadPropertyConsoleCmd(ConsoleCmd):
# __main__
#
try:
def main():
global this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -189,7 +190,5 @@ try:
_log.debug("fini")
except Exception as error:
_log.exception("an error has occurred: %s", error)
finally:
_log.debug("finally")
if __name__ == "__main__":
main()

View File

@ -26,9 +26,7 @@ _debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
this_console = None
#
# ReadPropertyAnyApplication
@ -136,6 +134,8 @@ class ReadPropertyAnyConsoleCmd(ConsoleCmd):
#
def main():
global this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()

View File

@ -206,6 +206,8 @@ class ReadPropertyMultipleConsoleCmd(ConsoleCmd):
#
def main():
global this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()

View File

@ -25,10 +25,6 @@ from bacpypes.errors import ExecutionError
_debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
#
# RandomValueProperty
#

View File

@ -24,9 +24,7 @@ _debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
this_console = None
#
# ReadRangeApplication
@ -124,7 +122,9 @@ class ReadRangeConsoleCmd(ConsoleCmd):
# __main__
#
try:
def main():
global this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -160,7 +160,7 @@ try:
run()
except Exception as error:
_log.exception("an error has occurred: %s", error)
finally:
_log.debug("finally")
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -227,7 +227,9 @@ class TestConsoleCmd(ConsoleCmd):
# __main__
#
try:
def main():
global this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -263,7 +265,7 @@ try:
run()
except Exception as err:
_log.exception("an error has occurred: %s", err)
finally:
_log.debug("finally")
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -218,5 +218,7 @@ def main():
run()
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -30,9 +30,7 @@ _debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
this_console = None
#
# ReadPropertyApplication
@ -234,7 +232,9 @@ class ReadWritePropertyConsoleCmd(ConsoleCmd):
# __main__
#
try:
def main():
global this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -270,7 +270,7 @@ try:
run()
except Exception as error:
_log.exception("an error has occurred: %s", error)
finally:
_log.debug("finally")
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -27,11 +27,6 @@ from bacpypes.constructeddata import Array
_debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
this_console = None
# point list
point_list = [
('1.2.3.4', 'analogValue', 1, 'presentValue'),
@ -150,7 +145,7 @@ class PrairieDog(BIPSimpleApplication, RecurringTask):
# __main__
#
try:
def main():
# parse the command line arguments
parser = ConfigArgumentParser(description=__doc__)
@ -176,6 +171,7 @@ try:
# make a dog
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()
@ -188,7 +184,7 @@ try:
run()
except Exception as error:
_log.exception("an error has occurred: %s", error)
finally:
_log.debug("finally")
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -44,7 +44,7 @@ class PrairieDog(RecurringTask):
# __main__
#
try:
def main():
# parse the command line arguments
parser = ArgumentParser(description=__doc__)
@ -68,7 +68,7 @@ try:
run()
except Exception as error:
_log.exception("an error has occurred: %s", error)
finally:
_log.debug("finally")
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -25,10 +25,6 @@ from bacpypes.app import LocalDeviceObject, BIPSimpleApplication
_debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
#
# SampleApplication
#
@ -61,7 +57,7 @@ class SampleApplication(BIPSimpleApplication):
# __main__
#
try:
def main():
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -93,8 +89,7 @@ try:
run()
except Exception as err:
_log.exception("an error has occurred: %s", err)
finally:
_log.debug("finally")
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -20,11 +20,6 @@ from bacpypes.app import LocalDeviceObject, BIPSimpleApplication
_debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
this_console = None
#
# SampleApplication
#
@ -70,7 +65,7 @@ class SampleConsoleCmd(ConsoleCmd):
# __main__
#
try:
def main():
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -106,8 +101,7 @@ try:
run()
except Exception as error:
_log.exception("an error has occurred: %s", error)
finally:
_log.debug("finally")
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -24,9 +24,7 @@ _debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
this_console = None
# how the application should respond
rsvp = (True, None, None)
@ -189,7 +187,9 @@ class SubscribeCOVConsoleCmd(ConsoleCmd):
# __main__
#
try:
def main():
global this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -217,6 +217,7 @@ try:
# make a console
this_console = SubscribeCOVConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)
# enable sleeping will help with threads
enable_sleeping()
@ -225,7 +226,7 @@ try:
run()
except Exception as error:
_log.exception("an error has occurred: %s", error)
finally:
_log.debug("finally")
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -24,9 +24,6 @@ _log = ModuleLogger(globals())
SERVER_HOST = os.getenv('SERVER_HOST', 'any')
SERVER_PORT = int(os.getenv('SERVER_PORT', 9000))
# globals
server_address = None
#
# EchoMaster
#
@ -89,8 +86,7 @@ def main():
host = args.host
if host == "any":
host = ''
port = args.port
server_address = (host, port)
server_address = (host, args.port)
if _debug: _log.debug(" - server_address: %r", server_address)
# create a director listening to the address
@ -114,4 +110,3 @@ def main():
if __name__ == "__main__":
main()

View File

@ -175,7 +175,7 @@ class BroadcastReceiver(Client):
# __main__
#
try:
def main():
# parse the command line arguments
parser = ArgumentParser(description=__doc__)
parser.add_argument("address",
@ -235,7 +235,7 @@ try:
run()
except Exception as e:
_log.exception("an error has occurred: %s", e)
finally:
_log.debug("finally")
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -84,59 +84,54 @@ register_object_type(VendorAVObject, vendor_id=vendor_id)
@bacpypes_debugging
def main():
if _debug: main._debug("initialization")
global vendor_id
try:
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
if _debug: main._debug("initialization")
if _debug: main._debug(" - args: %r", args)
if _debug: _log.debug("initialization")
if _debug: _log.debug(" - args: %r", args)
# make a device object
this_device = LocalDeviceObject(
objectName=args.ini.objectname,
objectIdentifier=int(args.ini.objectidentifier),
maxApduLengthAccepted=int(args.ini.maxapdulengthaccepted),
segmentationSupported=args.ini.segmentationsupported,
vendorIdentifier=vendor_id,
)
# make a device object
this_device = LocalDeviceObject(
objectName=args.ini.objectname,
objectIdentifier=int(args.ini.objectidentifier),
maxApduLengthAccepted=int(args.ini.maxapdulengthaccepted),
segmentationSupported=args.ini.segmentationsupported,
vendorIdentifier=vendor_id,
)
# make a sample application
this_application = BIPSimpleApplication(this_device, args.ini.address)
# 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)
# 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
# let the device object know
this_device.protocolServicesSupported = services_supported.value
# make some objects
ravo1 = VendorAVObject(
objectIdentifier=(513, 1), objectName='Random1'
)
if _debug: main._debug(" - ravo1: %r", ravo1)
# make some objects
ravo1 = VendorAVObject(
objectIdentifier=(513, 1), objectName='Random1'
)
if _debug: _log.debug(" - ravo1: %r", ravo1)
ravo2 = VendorAVObject(
objectIdentifier=(513, 2), objectName='Random2'
)
if _debug: main._debug(" - ravo2: %r", ravo2)
ravo2 = VendorAVObject(
objectIdentifier=(513, 2), objectName='Random2'
)
if _debug: _log.debug(" - ravo2: %r", ravo2)
# add it to the device
this_application.add_object(ravo1)
this_application.add_object(ravo2)
if _debug: main._debug(" - object list: %r", this_device.objectList)
# add it to the device
this_application.add_object(ravo1)
this_application.add_object(ravo2)
if _debug: _log.debug(" - object list: %r", this_device.objectList)
if _debug: main._debug("running")
if _debug: _log.debug("running")
run()
run()
except Exception as error:
main._exception("an error has occurred: %s", error)
finally:
if _debug: main._debug("finally")
if _debug: _log.debug("fini")
if __name__ == '__main__':
main()

View File

@ -31,6 +31,9 @@ import VendorAVObject
_debug = 0
_log = ModuleLogger(globals())
# globals
this_application = None
#
# ReadPropertyApplication
#
@ -97,8 +100,6 @@ class ReadWritePropertyConsoleCmd(ConsoleCmd):
def do_read(self, args):
"""read <addr> <type> <inst> <prop> [ <indx> ]"""
global this_application
args = args.split()
if _debug: ReadWritePropertyConsoleCmd._debug("do_read %r", args)
@ -141,8 +142,6 @@ class ReadWritePropertyConsoleCmd(ConsoleCmd):
def do_write(self, args):
"""write <addr> <type> <inst> <prop> <value> [ <indx> ] [ <priority> ]"""
global this_application
args = args.split()
ReadWritePropertyConsoleCmd._debug("do_write %r", args)
@ -235,16 +234,14 @@ class ReadWritePropertyConsoleCmd(ConsoleCmd):
# main
#
@bacpypes_debugging
def main():
if _debug: main._debug("initialization")
global this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
if _debug: main._debug("initialization")
if _debug: main._debug(" - args: %r", args)
if _debug: _log.debug("initialization")
if _debug: _log.debug(" - args: %r", args)
# make a device object
this_device = LocalDeviceObject(
@ -272,10 +269,11 @@ def main():
# enable sleeping will help with threads
enable_sleeping()
main._debug("running")
_log.debug("running")
run()
_log.debug("fini")
if __name__ == '__main__':
main()

View File

@ -21,10 +21,6 @@ from bacpypes.app import LocalDeviceObject, BIPSimpleApplication
_debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
# counters
who_has_counter = defaultdict(int)
i_have_counter = defaultdict(int)
@ -76,7 +72,7 @@ class WhoHasIHaveApplication(BIPSimpleApplication):
# __main__
#
try:
def main():
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -107,6 +103,8 @@ try:
# run until stopped, ^C works
run()
_log.debug("fini")
print("----- Who Has -----")
for (src, objname), count in sorted(who_has_counter.items()):
print("%-20s %-30s %4d" % (src, objname, count))
@ -117,7 +115,5 @@ try:
print("%-20s %-20s %-20s %-20s %4d" % (src, devid, objid, objname, count))
print("")
except Exception as error:
_log.exception("an error has occurred: %s", error)
finally:
_log.debug("finally")
if __name__ == "__main__":
main()

View File

@ -26,9 +26,7 @@ _debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
this_console = None
#
# WhoIsIAmApplication
@ -162,7 +160,9 @@ class WhoIsIAmConsoleCmd(ConsoleCmd):
# __main__
#
try:
def main():
global this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -200,6 +200,7 @@ try:
# make a console
this_console = WhoIsIAmConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)
# enable sleeping will help with threads
enable_sleeping()
@ -208,8 +209,7 @@ try:
run()
except Exception as error:
_log.exception("an error has occurred: %s", error)
finally:
_log.debug("finally")
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -20,10 +20,6 @@ from bacpypes.app import LocalDeviceObject, BIPSimpleApplication
_debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
# counters
who_is_counter = defaultdict(int)
i_am_counter = defaultdict(int)
@ -73,7 +69,7 @@ class WhoIsIAmApplication(BIPSimpleApplication):
# __main__
#
try:
def main():
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -103,6 +99,8 @@ try:
run()
_log.debug("fini")
print("----- Who Is -----")
for (src, lowlim, hilim), count in sorted(who_is_counter.items()):
print("%-20s %8s %8s %4d" % (src, lowlim, hilim, count))
@ -113,7 +111,5 @@ try:
print("%-20s %8d %4d" % (src, devid, count))
print("")
except Exception as error:
_log.exception("an error has occurred: %s", error)
finally:
_log.debug("finally")
if __name__ == "__main__":
main()

View File

@ -28,7 +28,6 @@ _log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
this_console = None
#
# WhoIsIAmApplication
@ -163,7 +162,7 @@ class WhoIsIAmConsoleCmd(ConsoleCmd):
#
def main():
global this_device, this_application, this_console
global this_device, this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -217,7 +216,8 @@ def main():
run()
_log.debug("finally")
_log.debug("fini")
if __name__ == "__main__":
main()

View File

@ -103,7 +103,9 @@ class WhoIsRouterConsoleCmd(ConsoleCmd):
# __main__
#
try:
def main():
global this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -125,7 +127,7 @@ try:
run()
except Exception as error:
_log.exception("an error has occurred: %s", error)
finally:
_log.debug("finally")
_log.debug("fini")
if __name__ == "__main__":
main()