mirror of
https://github.com/JoelBender/bacpypes
synced 2025-09-28 22:15:23 +08:00
240 lines
7.6 KiB
Python
Executable File
240 lines
7.6 KiB
Python
Executable File
#!/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 for each I-Am that is returned, reads
|
|
the object name of the device object (often called simply the device name).
|
|
"""
|
|
|
|
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.iocb import IOCB
|
|
|
|
from bacpypes.pdu import Address, GlobalBroadcast
|
|
from bacpypes.apdu import WhoIsRequest, IAmRequest, ReadPropertyRequest, ReadPropertyACK
|
|
from bacpypes.primitivedata import CharacterString
|
|
from bacpypes.basetypes import ServicesSupported
|
|
from bacpypes.errors import DecodingError
|
|
|
|
from bacpypes.app import BIPForeignApplication
|
|
from bacpypes.service.device import LocalDeviceObject
|
|
|
|
# some debugging
|
|
_debug = 1
|
|
_log = ModuleLogger(globals())
|
|
|
|
# globals
|
|
this_device = None
|
|
this_application = None
|
|
|
|
#
|
|
# DiscoveryApplication
|
|
#
|
|
|
|
@bacpypes_debugging
|
|
class DiscoveryApplication(BIPForeignApplication):
|
|
|
|
def __init__(self, *args):
|
|
if _debug: DiscoveryApplication._debug("__init__ %r", args)
|
|
BIPForeignApplication.__init__(self, *args)
|
|
|
|
# keep track of requests to line up responses
|
|
self.who_is_request = None
|
|
|
|
def request(self, apdu):
|
|
"""Sniff for Who-Is requests going downstream."""
|
|
if _debug: DiscoveryApplication._debug("request %r", apdu)
|
|
|
|
# save a copy of just the Who-Is request
|
|
if isinstance(apdu, WhoIsRequest):
|
|
self.who_is_request = apdu
|
|
|
|
# forward it along
|
|
BIPForeignApplication.request(self, apdu)
|
|
|
|
def do_IAmRequest(self, apdu):
|
|
"""Do something with incoming I-Am requests."""
|
|
if _debug: DiscoveryApplication._debug("do_IAmRequest %r", apdu)
|
|
|
|
# check for required parameters
|
|
if apdu.iAmDeviceIdentifier is None:
|
|
raise MissingRequiredParameter("iAmDeviceIdentifier required")
|
|
if apdu.maxAPDULengthAccepted is None:
|
|
raise MissingRequiredParameter("maxAPDULengthAccepted required")
|
|
if apdu.segmentationSupported is None:
|
|
raise MissingRequiredParameter("segmentationSupported required")
|
|
if apdu.vendorID is None:
|
|
raise MissingRequiredParameter("vendorID required")
|
|
|
|
# extract the device instance number
|
|
device_instance = apdu.iAmDeviceIdentifier[1]
|
|
if _debug: DiscoveryApplication._debug(" - device_instance: %r", device_instance)
|
|
|
|
# extract the source address
|
|
device_address = apdu.pduSource
|
|
if _debug: DiscoveryApplication._debug(" - device_address: %r", device_address)
|
|
|
|
# we didn't request anything yet
|
|
if not self.who_is_request:
|
|
return
|
|
|
|
if (self.who_is_request.deviceInstanceRangeLowLimit is not None) and \
|
|
(device_instance < self.who_is_request.deviceInstanceRangeLowLimit):
|
|
pass
|
|
elif (self.who_is_request.deviceInstanceRangeHighLimit is not None) and \
|
|
(device_instance > self.who_is_request.deviceInstanceRangeHighLimit):
|
|
pass
|
|
else:
|
|
# build a request for the object name
|
|
request = ReadPropertyRequest(
|
|
destination=apdu.pduSource,
|
|
objectIdentifier=apdu.iAmDeviceIdentifier,
|
|
propertyIdentifier='objectName',
|
|
)
|
|
|
|
# make an IOCB
|
|
iocb = IOCB(request)
|
|
if _debug: DiscoveryApplication._debug(" - iocb: %r", iocb)
|
|
|
|
# let us know when its complete
|
|
iocb.add_callback(self.device_discovered)
|
|
|
|
# give it to the application
|
|
self.request_io(iocb)
|
|
|
|
def device_discovered(self, iocb):
|
|
if _debug: DiscoveryApplication._debug("device_discovered %r", iocb)
|
|
|
|
# do something for error/reject/abort
|
|
if iocb.ioError:
|
|
sys.stdout.write(str(iocb.ioError) + '\n')
|
|
|
|
# do something for success
|
|
elif iocb.ioResponse:
|
|
apdu = iocb.ioResponse
|
|
|
|
# should be an ack
|
|
if not isinstance(apdu, ReadPropertyACK):
|
|
if _debug: DiscoveryApplication._debug(" - not an ack")
|
|
return
|
|
|
|
# pull out the name
|
|
device_name = apdu.propertyValue.cast_out(CharacterString)
|
|
if _debug: DiscoveryApplication._debug(" - device_name: %r", device_name)
|
|
|
|
# print out the response
|
|
sys.stdout.write("%s is at %s named %r\n" % (apdu.objectIdentifier[1], apdu.pduSource, device_name))
|
|
|
|
# do something with nothing?
|
|
else:
|
|
if _debug: DiscoveryApplication._debug(" - ioError or ioResponse expected")
|
|
|
|
|
|
#
|
|
# DiscoveryConsoleCmd
|
|
#
|
|
|
|
@bacpypes_debugging
|
|
class DiscoveryConsoleCmd(ConsoleCmd):
|
|
|
|
def do_whois(self, args):
|
|
"""whois [ <addr> ] [ <lolimit> <hilimit> ]"""
|
|
args = args.split()
|
|
if _debug: DiscoveryConsoleCmd._debug("do_whois %r", args)
|
|
|
|
try:
|
|
# gather the parameters
|
|
request = WhoIsRequest()
|
|
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:
|
|
DiscoveryConsoleCmd._exception("exception: %r", error)
|
|
|
|
def do_rtn(self, args):
|
|
"""rtn <addr> <net> ... """
|
|
args = args.split()
|
|
if _debug: DiscoveryConsoleCmd._debug("do_rtn %r", args)
|
|
|
|
# safe to assume only one adapter
|
|
adapter = this_application.nsap.adapters[0]
|
|
if _debug: DiscoveryConsoleCmd._debug(" - adapter: %r", adapter)
|
|
|
|
# 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.add_router_references(adapter, 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 = DiscoveryApplication(
|
|
this_device, args.ini.address,
|
|
Address(args.ini.foreignbbmd),
|
|
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)
|
|
|
|
# enable sleeping will help with threads
|
|
enable_sleeping()
|
|
|
|
_log.debug("running")
|
|
|
|
run()
|
|
|
|
_log.debug("fini")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|