mirror of
https://github.com/JoelBender/bacpypes
synced 2025-09-28 22:15:23 +08:00
bring the branch for #110 up to date with stage
This commit is contained in:
Joel Bender
commit
856cad7a7b
|
|
@ -534,6 +534,10 @@ class BIPForeignApplication(ApplicationIOController, WhoIsIAmServices, ReadWrite
|
|||
# can know if it should support segmentation
|
||||
self.smap = StateMachineAccessPoint(localDevice)
|
||||
|
||||
# the segmentation state machines need access to the same device
|
||||
# information cache as the application
|
||||
self.smap.deviceInfoCache = self.deviceInfoCache
|
||||
|
||||
# a network service access point will be needed
|
||||
self.nsap = NetworkServiceAccessPoint()
|
||||
|
||||
|
|
|
|||
|
|
@ -266,7 +266,7 @@ class Sequence(object):
|
|||
|
||||
# save the result
|
||||
setattr(self, element.name, value)
|
||||
except DecodingError:
|
||||
except (DecodingError, InvalidTag), err:
|
||||
# if the context tag was matched, the substructure has to be decoded
|
||||
# correctly.
|
||||
if element.context is None and element.optional:
|
||||
|
|
|
|||
|
|
@ -85,7 +85,7 @@ class ReadWritePropertyServices(Capability):
|
|||
if _debug: ReadWritePropertyServices._debug(" - resp: %r", resp)
|
||||
|
||||
except PropertyError:
|
||||
raise ExecutionError(errorClass='object', errorCode='unknownProperty')
|
||||
raise ExecutionError(errorClass='property', errorCode='unknownProperty')
|
||||
|
||||
# return the result
|
||||
self.response(resp)
|
||||
|
|
@ -130,7 +130,7 @@ class ReadWritePropertyServices(Capability):
|
|||
if _debug: ReadWritePropertyServices._debug(" - resp: %r", resp)
|
||||
|
||||
except PropertyError:
|
||||
raise ExecutionError(errorClass='object', errorCode='unknownProperty')
|
||||
raise ExecutionError(errorClass='property', errorCode='unknownProperty')
|
||||
|
||||
# return the result
|
||||
self.response(resp)
|
||||
|
|
|
|||
|
|
@ -530,6 +530,10 @@ class BIPForeignApplication(ApplicationIOController, WhoIsIAmServices, ReadWrite
|
|||
# can know if it should support segmentation
|
||||
self.smap = StateMachineAccessPoint(localDevice)
|
||||
|
||||
# the segmentation state machines need access to the same device
|
||||
# information cache as the application
|
||||
self.smap.deviceInfoCache = self.deviceInfoCache
|
||||
|
||||
# a network service access point will be needed
|
||||
self.nsap = NetworkServiceAccessPoint()
|
||||
|
||||
|
|
|
|||
|
|
@ -267,7 +267,7 @@ class Sequence(object):
|
|||
|
||||
# save the result
|
||||
setattr(self, element.name, value)
|
||||
except DecodingError:
|
||||
except (DecodingError, InvalidTag) as err:
|
||||
# if the context tag was matched, the substructure has to be decoded
|
||||
# correctly.
|
||||
if element.context is None and element.optional:
|
||||
|
|
|
|||
|
|
@ -287,6 +287,28 @@ class WhoHasIHaveServices(Capability):
|
|||
if _debug: WhoIsIAmServices._debug(" - no local device")
|
||||
return
|
||||
|
||||
# if this has limits, check them like Who-Is
|
||||
if apdu.limits is not None:
|
||||
# extract the parameters
|
||||
low_limit = apdu.limits.deviceInstanceRangeLowLimit
|
||||
high_limit = apdu.limits.deviceInstanceRangeHighLimit
|
||||
|
||||
# check for consistent parameters
|
||||
if (low_limit is None):
|
||||
raise MissingRequiredParameter("deviceInstanceRangeLowLimit required")
|
||||
if (low_limit < 0) or (low_limit > 4194303):
|
||||
raise ParameterOutOfRange("deviceInstanceRangeLowLimit out of range")
|
||||
if (high_limit is None):
|
||||
raise MissingRequiredParameter("deviceInstanceRangeHighLimit required")
|
||||
if (high_limit < 0) or (high_limit > 4194303):
|
||||
raise ParameterOutOfRange("deviceInstanceRangeHighLimit out of range")
|
||||
|
||||
# see we should respond
|
||||
if (self.localDevice.objectIdentifier[1] < low_limit):
|
||||
return
|
||||
if (self.localDevice.objectIdentifier[1] > high_limit):
|
||||
return
|
||||
|
||||
# find the object
|
||||
if apdu.object.objectIdentifier is not None:
|
||||
obj = self.objectIdentifier.get(apdu.object.objectIdentifier, None)
|
||||
|
|
@ -294,8 +316,10 @@ class WhoHasIHaveServices(Capability):
|
|||
obj = self.objectName.get(apdu.object.objectName, None)
|
||||
else:
|
||||
raise InconsistentParameters("object identifier or object name required")
|
||||
|
||||
# maybe we don't have it
|
||||
if not obj:
|
||||
raise ExecutionError(errorClass='object', errorCode='unknownObject')
|
||||
return
|
||||
|
||||
# send out the response
|
||||
self.i_have(obj, address=apdu.pduSource)
|
||||
|
|
|
|||
|
|
@ -86,7 +86,7 @@ class ReadWritePropertyServices(Capability):
|
|||
if _debug: ReadWritePropertyServices._debug(" - resp: %r", resp)
|
||||
|
||||
except PropertyError:
|
||||
raise ExecutionError(errorClass='object', errorCode='unknownProperty')
|
||||
raise ExecutionError(errorClass='property', errorCode='unknownProperty')
|
||||
|
||||
# return the result
|
||||
self.response(resp)
|
||||
|
|
@ -131,7 +131,7 @@ class ReadWritePropertyServices(Capability):
|
|||
if _debug: ReadWritePropertyServices._debug(" - resp: %r", resp)
|
||||
|
||||
except PropertyError:
|
||||
raise ExecutionError(errorClass='object', errorCode='unknownProperty')
|
||||
raise ExecutionError(errorClass='property', errorCode='unknownProperty')
|
||||
|
||||
# return the result
|
||||
self.response(resp)
|
||||
|
|
|
|||
|
|
@ -530,6 +530,10 @@ class BIPForeignApplication(ApplicationIOController, WhoIsIAmServices, ReadWrite
|
|||
# can know if it should support segmentation
|
||||
self.smap = StateMachineAccessPoint(localDevice)
|
||||
|
||||
# the segmentation state machines need access to the same device
|
||||
# information cache as the application
|
||||
self.smap.deviceInfoCache = self.deviceInfoCache
|
||||
|
||||
# a network service access point will be needed
|
||||
self.nsap = NetworkServiceAccessPoint()
|
||||
|
||||
|
|
|
|||
|
|
@ -129,7 +129,7 @@ class UDPMultiplexer:
|
|||
return
|
||||
|
||||
elif pdu.pduDestination.addrType == Address.localStationAddr:
|
||||
dest = pdu.pduDestination.addrTuple
|
||||
dest = unpack_ip_addr(pdu.pduDestination.addrAddr)
|
||||
if _debug: UDPMultiplexer._debug(" - requesting local station: %r", dest)
|
||||
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -267,7 +267,7 @@ class Sequence(object):
|
|||
|
||||
# save the result
|
||||
setattr(self, element.name, value)
|
||||
except DecodingError:
|
||||
except (DecodingError, InvalidTag) as err:
|
||||
# if the context tag was matched, the substructure has to be decoded
|
||||
# correctly.
|
||||
if element.context is None and element.optional:
|
||||
|
|
|
|||
|
|
@ -86,7 +86,7 @@ class ReadWritePropertyServices(Capability):
|
|||
if _debug: ReadWritePropertyServices._debug(" - resp: %r", resp)
|
||||
|
||||
except PropertyError:
|
||||
raise ExecutionError(errorClass='object', errorCode='unknownProperty')
|
||||
raise ExecutionError(errorClass='property', errorCode='unknownProperty')
|
||||
|
||||
# return the result
|
||||
self.response(resp)
|
||||
|
|
@ -131,7 +131,7 @@ class ReadWritePropertyServices(Capability):
|
|||
if _debug: ReadWritePropertyServices._debug(" - resp: %r", resp)
|
||||
|
||||
except PropertyError:
|
||||
raise ExecutionError(errorClass='object', errorCode='unknownProperty')
|
||||
raise ExecutionError(errorClass='property', errorCode='unknownProperty')
|
||||
|
||||
# return the result
|
||||
self.response(resp)
|
||||
|
|
|
|||
235
samples/DeviceDiscovery.py
Executable file
235
samples/DeviceDiscovery.py
Executable file
|
|
@ -0,0 +1,235 @@
|
|||
#!/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 BIPSimpleApplication
|
||||
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(BIPSimpleApplication):
|
||||
|
||||
def __init__(self, *args):
|
||||
if _debug: DiscoveryApplication._debug("__init__ %r", args)
|
||||
BIPSimpleApplication.__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
|
||||
BIPSimpleApplication.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)
|
||||
|
||||
# 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()
|
||||
239
samples/DeviceDiscoveryForeign.py
Executable file
239
samples/DeviceDiscoveryForeign.py
Executable file
|
|
@ -0,0 +1,239 @@
|
|||
#!/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()
|
||||
Loading…
Reference in New Issue
Block a user