mirror/bacpypes
1
0
Fork 0
mirror of https://github.com/JoelBender/bacpypes synced 2025-10-05 22:18:16 +08:00
Code Issues Releases Wiki Activity

refactor: Remove tutorials/WhoIsIAm.py

Browse Source 
This removes the Tutorial code and updates the old samples/WhoIsIAm.py to match it.
This commit is contained in:
Richard Littauer 2020-04-28 18:13:47 -04:00
parent d832f84319
commit 9ec89dbf11
4 changed files with 168 additions and 370 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
doc/source/gettingstarted/gettingstarted001.rst
View File

@ -244,7 +244,7 @@ them to be.
Now start the simplest tutorial application::
$ python samples/Tutorial/WhoIsIAm.py
$ python samples/WhoIsIAm.py
.. note::

18
doc/source/gettingstarted/gettingstarted002.rst
View File

@ -14,7 +14,7 @@ Getting Help
Whatever the command line parameters and additional options might be for
an application, you can start with help::
$ python Tutorial/WhoIsIAm.py --help
$ python samples/WhoIsIAm.py --help
usage: WhoIsIAm.py [-h] [--buggers] [--debug [DEBUG [DEBUG ...]]] [--color] [--ini INI]
This application presents a 'console' prompt to the user asking for Who-Is and
@ -43,7 +43,7 @@ Because BACpypes modules are deeply interconnected, dumping a complete list
of all of the logger names is a long list. Start out focusing on the
components of the WhoIsIAm.py application::
$ python Tutorial/WhoIsIAm.py --buggers | grep __main__
$ python samples/WhoIsIAm.py --buggers | grep __main__
__main__
__main__.WhoIsIAmApplication
__main__.WhoIsIAmConsoleCmd
@ -76,14 +76,14 @@ Debugging a Class
Debugging all of the classes and functions can generate a lot of output,
so it is useful to focus on a specific function or class::
$ python Tutorial/WhoIsIAm.py --debug __main__.WhoIsIAmApplication
$ python samples/WhoIsIAm.py --debug __main__.WhoIsIAmApplication
DEBUG:__main__.WhoIsIAmApplication:__init__ (<bacpypes.app.LocalDeviceObject object at 0x9bca8ac>, '128.253.109.40/24:47808')
>
The same method is used to debug the activity of a BACpypes module, for
example, there is a class called UDPActor in the UDP module::
$ python Tutorial/WhoIsIAm.py --ini BAC0.ini --debug bacpypes.udp.UDPActor
$ python samples/WhoIsIAm.py --ini BAC0.ini --debug bacpypes.udp.UDPActor
> DEBUG:bacpypes.udp.UDPActor:__init__ <bacpypes.udp.UDPDirector 128.253.109.255:47808 at 0xb6d40d6c> ('128.253.109.254', 47808)
DEBUG:bacpypes.udp.UDPActor:response <bacpypes.comm.PDU object at 0xb6d433cc>
<bacpypes.comm.PDU object at 0xb6d433cc>
@ -111,17 +111,17 @@ backupCount. For example, this invocation sends the main application debugging
to standard error and the debugging output of the bacpypes.udp module to the
traffic.txt file::
$ python Tutorial/WhoIsIAm.py --debug __main__ bacpypes.udp:traffic.txt
$ python samples/WhoIsIAm.py --debug __main__ bacpypes.udp:traffic.txt
By default the `maxBytes` is zero so there is no rotating file, but it can be
provided, for example this limits the file size to 1MB::
$ python Tutorial/WhoIsIAm.py --debug __main__ bacpypes.udp:traffic.txt:1048576
$ python samples/WhoIsIAm.py --debug __main__ bacpypes.udp:traffic.txt:1048576
If `maxBytes` is provided, then by default the `backupCount` is 10, but it can also
be specified, so this limits the output to one hundred files::
$ python Tutorial/WhoIsIAm.py --debug __main__ bacpypes.udp:traffic.txt:1048576:100
$ python samples/WhoIsIAm.py --debug __main__ bacpypes.udp:traffic.txt:1048576:100
.. caution::
@ -143,11 +143,11 @@ Rather than swapping INI files, you can simply provide the INI file on the
command line, overriding the default BACpypes.ini file. For example, I
have an INI file for port 47808::
$ python Tutorial/WhoIsIAm.py --ini BAC0.ini
$ python samples/WhoIsIAm.py --ini BAC0.ini
And another one for port 47809::
$ python Tutorial/WhoIsIAm.py --ini BAC1.ini
$ python samples/WhoIsIAm.py --ini BAC1.ini
And I switch back and forth between them.

181
samples/Tutorial/WhoIsIAm.py
View File

@ -1,181 +0,0 @@
#!/usr/bin/env python
"""
This application presents a 'console' prompt to the user asking for Who-Is and I-Am
commands which create the related APDUs, then lines up the corresponding I-Am
for incoming traffic and prints out the contents.
"""
import sys
from bacpypes.debugging import bacpypes_debugging, ModuleLogger
from bacpypes.consolelogging import ConfigArgumentParser
from bacpypes.consolecmd import ConsoleCmd
from bacpypes.core import run, enable_sleeping
from bacpypes.pdu import Address, GlobalBroadcast
from bacpypes.apdu import WhoIsRequest, IAmRequest
from bacpypes.errors import DecodingError
from bacpypes.app import BIPSimpleApplication
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 1
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
#
# WhoIsIAmApplication
#
@bacpypes_debugging
class WhoIsIAmApplication(BIPSimpleApplication):
def __init__(self, *args):
if _debug: WhoIsIAmApplication._debug("__init__ %r", args)
BIPSimpleApplication.__init__(self, *args)
# keep track of requests to line up responses
self._request = None
def request(self, apdu):
if _debug: WhoIsIAmApplication._debug("request %r", apdu)
# save a copy of the request
self._request = apdu
# forward it along
BIPSimpleApplication.request(self, apdu)
def confirmation(self, apdu):
if _debug: WhoIsIAmApplication._debug("confirmation %r", apdu)
# forward it along
BIPSimpleApplication.confirmation(self, apdu)
def indication(self, apdu):
if _debug: WhoIsIAmApplication._debug("indication %r", apdu)
if (isinstance(self._request, WhoIsRequest)) and (isinstance(apdu, IAmRequest)):
device_type, device_instance = apdu.iAmDeviceIdentifier
if device_type != 'device':
raise DecodingError("invalid object type")
if (self._request.deviceInstanceRangeLowLimit is not None) and \
(device_instance < self._request.deviceInstanceRangeLowLimit):
pass
elif (self._request.deviceInstanceRangeHighLimit is not None) and \
(device_instance > self._request.deviceInstanceRangeHighLimit):
pass
else:
# print out the contents
sys.stdout.write('pduSource = ' + repr(apdu.pduSource) + '\n')
sys.stdout.write('iAmDeviceIdentifier = ' + str(apdu.iAmDeviceIdentifier) + '\n')
sys.stdout.write('maxAPDULengthAccepted = ' + str(apdu.maxAPDULengthAccepted) + '\n')
sys.stdout.write('segmentationSupported = ' + str(apdu.segmentationSupported) + '\n')
sys.stdout.write('vendorID = ' + str(apdu.vendorID) + '\n')
sys.stdout.flush()
# forward it along
BIPSimpleApplication.indication(self, apdu)
#
# WhoIsIAmConsoleCmd
#
@bacpypes_debugging
class WhoIsIAmConsoleCmd(ConsoleCmd):
def do_whois(self, args):
"""whois [ <addr> ] [ <lolimit> <hilimit> ]"""
args = args.split()
if _debug: WhoIsIAmConsoleCmd._debug("do_whois %r", args)
try:
# gather the parameters
if (len(args) == 1) or (len(args) == 3):
addr = Address(args[0])
del args[0]
else:
addr = GlobalBroadcast()
if len(args) == 2:
lolimit = int(args[0])
hilimit = int(args[1])
else:
lolimit = hilimit = None
# code lives in the device service
this_application.who_is(lolimit, hilimit, addr)
except Exception as error:
WhoIsIAmConsoleCmd._exception("exception: %r", error)
def do_iam(self, args):
"""iam"""
args = args.split()
if _debug: WhoIsIAmConsoleCmd._debug("do_iam %r", args)
# code lives in the device service
this_application.i_am()
def do_rtn(self, args):
"""rtn <addr> <net> ... """
args = args.split()
if _debug: WhoIsIAmConsoleCmd._debug("do_rtn %r", args)
# provide the address and a list of network numbers
router_address = Address(args[0])
network_list = [int(arg) for arg in args[1:]]
# pass along to the service access point
this_application.nsap.update_router_references(None, router_address, network_list)
#
# __main__
#
def main():
global this_device
global this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_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=int(args.ini.vendoridentifier),
)
# make a simple application
this_application = WhoIsIAmApplication(this_device, args.ini.address)
# make a console
this_console = WhoIsIAmConsoleCmd()
if _debug: _log.debug(" - this_console: %r", this_console)
# enable sleeping will help with threads
enable_sleeping()
_log.debug("running")
run()
_log.debug("fini")
if __name__ == "__main__":
main()

83
samples/WhoIsIAm.py Executable file → Normal file
View File

@ -2,7 +2,7 @@
"""
This application presents a 'console' prompt to the user asking for Who-Is and I-Am
commands which create the related APDUs, then lines up the coorresponding I-Am
commands which create the related APDUs, then lines up the corresponding I-Am
for incoming traffic and prints out the contents.
"""
@ -12,8 +12,7 @@ from bacpypes.debugging import bacpypes_debugging, ModuleLogger
from bacpypes.consolelogging import ConfigArgumentParser
from bacpypes.consolecmd import ConsoleCmd
from bacpypes.core import run, deferred, enable_sleeping
from bacpypes.iocb import IOCB
from bacpypes.core import run, enable_sleeping
from bacpypes.pdu import Address, GlobalBroadcast
from bacpypes.apdu import WhoIsRequest, IAmRequest
@ -23,7 +22,7 @@ from bacpypes.app import BIPSimpleApplication
from bacpypes.local.device import LocalDeviceObject
# some debugging
_debug = 0
_debug = 1
_log = ModuleLogger(globals())
# globals
@ -68,10 +67,10 @@ class WhoIsIAmApplication(BIPSimpleApplication):
raise DecodingError("invalid object type")
if (self._request.deviceInstanceRangeLowLimit is not None) and \
(device_instance < self._request.deviceInstanceRangeLowLimit):
(device_instance < self._request.deviceInstanceRangeLowLimit):
pass
elif (self._request.deviceInstanceRangeHighLimit is not None) and \
(device_instance > self._request.deviceInstanceRangeHighLimit):
(device_instance > self._request.deviceInstanceRangeHighLimit):
pass
else:
# print out the contents
@ -85,6 +84,7 @@ class WhoIsIAmApplication(BIPSimpleApplication):
# forward it along
BIPSimpleApplication.indication(self, apdu)
#
# WhoIsIAmConsoleCmd
#
@ -93,60 +93,37 @@ class WhoIsIAmApplication(BIPSimpleApplication):
class WhoIsIAmConsoleCmd(ConsoleCmd):
def do_whois(self, args):
"""whois [ <addr>] [ <lolimit> <hilimit> ]"""
"""whois [ <addr> ] [ <lolimit> <hilimit> ]"""
args = args.split()
if _debug: WhoIsIAmConsoleCmd._debug("do_whois %r", args)
try:
# build a request
request = WhoIsRequest()
# gather the parameters
if (len(args) == 1) or (len(args) == 3):
request.pduDestination = Address(args[0])
addr = Address(args[0])
del args[0]
else:
request.pduDestination = GlobalBroadcast()
addr = GlobalBroadcast()
if len(args) == 2:
request.deviceInstanceRangeLowLimit = int(args[0])
request.deviceInstanceRangeHighLimit = int(args[1])
if _debug: WhoIsIAmConsoleCmd._debug(" - request: %r", request)
lolimit = int(args[0])
hilimit = int(args[1])
else:
lolimit = hilimit = None
# make an IOCB
iocb = IOCB(request)
if _debug: WhoIsIAmConsoleCmd._debug(" - iocb: %r", iocb)
# code lives in the device service
this_application.who_is(lolimit, hilimit, addr)
# give it to the application
this_application.request_io(iocb)
except Exception as err:
WhoIsIAmConsoleCmd._exception("exception: %r", err)
except Exception as error:
WhoIsIAmConsoleCmd._exception("exception: %r", error)
def do_iam(self, args):
"""iam"""
args = args.split()
if _debug: WhoIsIAmConsoleCmd._debug("do_iam %r", args)
try:
# build a request
request = IAmRequest()
request.pduDestination = GlobalBroadcast()
# set the parameters from the device object
request.iAmDeviceIdentifier = this_device.objectIdentifier
request.maxAPDULengthAccepted = this_device.maxApduLengthAccepted
request.segmentationSupported = this_device.segmentationSupported
request.vendorID = this_device.vendorIdentifier
if _debug: WhoIsIAmConsoleCmd._debug(" - request: %r", request)
# make an IOCB
iocb = IOCB(request)
if _debug: WhoIsIAmConsoleCmd._debug(" - iocb: %r", iocb)
# give it to the application
this_application.request_io(iocb)
except Exception as err:
WhoIsIAmConsoleCmd._exception("exception: %r", err)
# code lives in the device service
this_application.i_am()
def do_rtn(self, args):
"""rtn <addr> <net> ... """
@ -162,11 +139,12 @@ class WhoIsIAmConsoleCmd(ConsoleCmd):
#
# main
# __main__
#
def main():
global this_device, this_application
global this_device
global this_application
# parse the command line arguments
args = ConfigArgumentParser(description=__doc__).parse_args()
@ -175,14 +153,16 @@ def main():
if _debug: _log.debug(" - args: %r", args)
# make a device object
this_device = LocalDeviceObject(ini=args.ini)
if _debug: _log.debug(" - this_device: %r", this_device)
this_device = LocalDeviceObject(
objectName=args.ini.objectname,
objectIdentifier=int(args.ini.objectidentifier),
maxApduLengthAccepted=int(args.ini.maxapdulengthaccepted),
segmentationSupported=args.ini.segmentationsupported,
vendorIdentifier=int(args.ini.vendoridentifier),
)
# make a simple application
this_application = WhoIsIAmApplication(
this_device, args.ini.address,
)
if _debug: _log.debug(" - this_application: %r", this_application)
this_application = WhoIsIAmApplication(this_device, args.ini.address)
# make a console
this_console = WhoIsIAmConsoleCmd()
@ -197,6 +177,5 @@ def main():
_log.debug("fini")
if __name__ == "__main__":
main()