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

merge in the current samples branch before splitting off a new issue branch

Browse Source
This commit is contained in:
Joel Bender 2016-02-04 13:57:05 -05:00
commit f496a8b5ef
17 changed files with 1851 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
doc/source/index.rst
View File

@ -19,10 +19,11 @@ And then use the setup utility to install it::
$ python setup.py install
If you would like to participate in its development, please join the
`developers mailing list <https://lists.sourceforge.net/lists/listinfo/bacpypes-developers>`_. There is also a
`Google+ <https://plus.google.com/100756765082570761221/posts>`_ page that you
can add to your circles have have release notifications show up in your
stream.
`developers mailing list
<https://lists.sourceforge.net/lists/listinfo/bacpypes-developers>`_, join the
`chat room on Gitter <https://gitter.im/JoelBender/bacpypes>`_, and add the
`Google+ <https://plus.google.com/100756765082570761221/posts>`_ to your
circles have have release notifications show up in your stream.
Welcome aboard!

7
doc/source/releasenotes.rst
View File

@ -5,6 +5,13 @@ Release Notes
This page contains release notes.
Version 0.13.6
--------------
There have been lots of changes in the span between the previous published
version and this one and I haven't quite figured out how to extract the
relevent content from the git log. More to come.
Version 0.13.0
--------------

3
py25/bacpypes/app.py
View File

@ -151,6 +151,9 @@ class Application(ApplicationServiceElement, Logging):
# keep track of the local device
self.localDevice = localDevice
# bind the device object to this application
localDevice._app = self
# allow the address to be cast to the correct type
if isinstance(localAddress, Address):
self.localAddress = localAddress

48
py25/bacpypes/consolecmd.py
View File

@ -39,14 +39,19 @@ def console_interrupt(*args):
class ConsoleCmd(cmd.Cmd, Thread, Logging):
def __init__(self, prompt="> ", allow_exec=False, stdin=None, stdout=None):
def __init__(self, prompt="> ", stdin=None, stdout=None):
if _debug: ConsoleCmd._debug("__init__")
cmd.Cmd.__init__(self, stdin=stdin, stdout=stdout)
Thread.__init__(self, name="ConsoleCmd")
# save the prompt and exec option
self.prompt = prompt
self.allow_exec = allow_exec
# check to see if this is running interactive
self.interactive = sys.__stdin__.isatty()
# save the prompt for interactive sessions, otherwise be quiet
if self.interactive:
self.prompt = prompt
else:
self.prompt = ''
# gc counters
self.type2count = {}
@ -55,10 +60,6 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
# logging handlers
self.handlers = {}
# execution space for the user
self._locals = {}
self._globals = {}
# set a INT signal handler, ^C will only get sent to the
# main thread and there's no way to break the readline
# call initiated by this thread - sigh
@ -76,7 +77,7 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
if _debug: ConsoleCmd._debug(" - done cmdloop")
# tell the main thread to stop, this thread will exit
core.stop()
core.deferred(core.stop)
def onecmd(self, cmdString):
if _debug: ConsoleCmd._debug('onecmd %r', cmdString)
@ -231,6 +232,7 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
def do_exit(self, args):
"""Exits from the console."""
if _debug: ConsoleCmd._debug("do_exit %r", args)
return -1
def do_EOF(self, args):
@ -240,6 +242,8 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
def do_shell(self, args):
"""Pass command to a system shell when line begins with '!'"""
if _debug: ConsoleCmd._debug("do_shell %r", args)
os.system(args)
def do_help(self, args):
@ -247,7 +251,9 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
'help' or '?' with no arguments prints a list of commands for which help is available
'help <command>' or '? <command>' gives help on <command>
"""
## The only reason to define this method is for the help text in the doc string
if _debug: ConsoleCmd._debug("do_help %r", args)
# the only reason to define this method is for the help text in the doc string
cmd.Cmd.do_help(self, args)
def preloop(self):
@ -272,10 +278,14 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
if not isinstance(err, IOError):
self.stdout.write("history error: %s\n" % err)
cmd.Cmd.postloop(self) ## Clean up command completion
# clean up command completion
cmd.Cmd.postloop(self)
self.stdout.write("Exiting...\n")
core.stop()
if self.interactive:
self.stdout.write("Exiting...\n")
# tell the core we have stopped
core.deferred(core.stop)
def precmd(self, line):
""" This method is called after the line has been input but before
@ -294,16 +304,4 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
"""Do nothing on empty input line"""
pass
def default(self, line):
"""Called on an input line when the command prefix is not recognized.
If allow_exec is enabled, execute the line as Python code.
"""
if not self.allow_exec:
return cmd.Cmd.default(self, line)
try:
exec(line) in self._locals, self._globals
except Exception, err:
self.stdout.write("%s : %s\n" % (err.__class__, err))
bacpypes_debugging(ConsoleCmd)

4
py25/bacpypes/object.py
View File

@ -183,7 +183,9 @@ class Property(Logging):
self.identifier, obj, value, arrayIndex, priority, direct
)
if (not direct):
if direct:
if _debug: Property._debug(" - direct write")
else:
# see if it must be provided
if not self.optional and value is None:
raise ValueError("%s value required" % (self.identifier,))

3
py27/bacpypes/app.py
View File

@ -151,6 +151,9 @@ class Application(ApplicationServiceElement, Logging):
# keep track of the local device
self.localDevice = localDevice
# bind the device object to this application
localDevice._app = self
# allow the address to be cast to the correct type
if isinstance(localAddress, Address):
self.localAddress = localAddress

6
py27/bacpypes/appservice.py
View File

@ -1314,6 +1314,12 @@ class ApplicationServiceAccessPoint(ApplicationServiceElement, ServiceAccessPoin
# forward the encoded packet
self.request(xpdu)
# if the upper layers of the application did not assign an invoke ID,
# copy the one that was assigned on its way down the stack
if isinstance(apdu, ConfirmedRequestPDU) and apdu.apduInvokeID is None:
if _debug: ApplicationServiceAccessPoint._debug(" - pass invoke ID upstream %r", xpdu.apduInvokeID)
apdu.apduInvokeID = xpdu.apduInvokeID
def confirmation(self, apdu):
if _debug: ApplicationServiceAccessPoint._debug("confirmation %r", apdu)

49
py27/bacpypes/consolecmd.py
View File

@ -40,14 +40,19 @@ def console_interrupt(*args):
@bacpypes_debugging
class ConsoleCmd(cmd.Cmd, Thread, Logging):
def __init__(self, prompt="> ", allow_exec=False, stdin=None, stdout=None):
def __init__(self, prompt="> ", stdin=None, stdout=None):
if _debug: ConsoleCmd._debug("__init__")
cmd.Cmd.__init__(self, stdin=stdin, stdout=stdout)
Thread.__init__(self, name="ConsoleCmd")
# save the prompt and exec option
self.prompt = prompt
self.allow_exec = allow_exec
# check to see if this is running interactive
self.interactive = sys.__stdin__.isatty()
# save the prompt for interactive sessions, otherwise be quiet
if self.interactive:
self.prompt = prompt
else:
self.prompt = ''
# gc counters
self.type2count = {}
@ -56,10 +61,6 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
# logging handlers
self.handlers = {}
# execution space for the user
self._locals = {}
self._globals = {}
# set a INT signal handler, ^C will only get sent to the
# main thread and there's no way to break the readline
# call initiated by this thread - sigh
@ -77,7 +78,7 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
if _debug: ConsoleCmd._debug(" - done cmdloop")
# tell the main thread to stop, this thread will exit
core.stop()
core.deferred(core.stop)
def onecmd(self, cmdString):
if _debug: ConsoleCmd._debug('onecmd %r', cmdString)
@ -232,15 +233,19 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
def do_exit(self, args):
"""Exits from the console."""
if _debug: ConsoleCmd._debug("do_exit %r", args)
return -1
def do_EOF(self, args):
"""Exit on system end of file character"""
if _debug: ConsoleCmd._debug("do_EOF %r", args)
return self.do_exit(args)
def do_shell(self, args):
"""Pass command to a system shell when line begins with '!'"""
if _debug: ConsoleCmd._debug("do_shell %r", args)
os.system(args)
def do_help(self, args):
@ -248,7 +253,9 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
'help' or '?' with no arguments prints a list of commands for which help is available
'help <command>' or '? <command>' gives help on <command>
"""
## The only reason to define this method is for the help text in the doc string
if _debug: ConsoleCmd._debug("do_exit %r", args)
# the only reason to define this method is for the help text in the doc string
cmd.Cmd.do_help(self, args)
def preloop(self):
@ -273,10 +280,14 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
if not isinstance(err, IOError):
self.stdout.write("history error: %s\n" % err)
cmd.Cmd.postloop(self) ## Clean up command completion
# clean up command completion
cmd.Cmd.postloop(self)
self.stdout.write("Exiting...\n")
core.stop()
if self.interactive:
self.stdout.write("Exiting...\n")
# tell the core we have stopped
core.deferred(core.stop)
def precmd(self, line):
""" This method is called after the line has been input but before
@ -294,15 +305,3 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
def emptyline(self):
"""Do nothing on empty input line"""
pass
def default(self, line):
"""Called on an input line when the command prefix is not recognized.
If allow_exec is enabled, execute the line as Python code.
"""
if not self.allow_exec:
return cmd.Cmd.default(self, line)
try:
exec(line) in self._locals, self._globals
except Exception as err:
self.stdout.write("%s : %s\n" % (err.__class__, err))

25
py27/bacpypes/object.py
View File

@ -183,7 +183,9 @@ class Property(Logging):
self.identifier, obj, value, arrayIndex, priority, direct
)
if (not direct):
if direct:
if _debug: Property._debug(" - direct write")
else:
# see if it must be provided
if not self.optional and value is None:
raise ValueError("%s value required" % (self.identifier,))
@ -513,23 +515,26 @@ class Object(Logging):
previous_attrs = attrs
# build a list of properties "bottom up"
properties = []
property_names = []
for c in klasses:
properties.extend(getattr(c, 'properties', []))
properties = getattr(c, 'properties', [])
for property in properties:
if property.identifier not in property_names:
property_names.append(property.identifier)
# print out the values
for prop in properties:
value = prop.ReadProperty(self)
for property_name in property_names:
property_value = self._values.get(property_name, None)
# printing out property values that are None is tedious
if value is None:
if property_value is None:
continue
if hasattr(value, "debug_contents"):
file.write("%s%s\n" % (" " * indent, prop.identifier))
value.debug_contents(indent+1, file, _ids)
if hasattr(property_value, "debug_contents"):
file.write("%s%s\n" % (" " * indent, property_name))
property_value.debug_contents(indent+1, file, _ids)
else:
file.write("%s%s = %r\n" % (" " * indent, prop.identifier, value))
file.write("%s%s = %r\n" % (" " * indent, property_name, property_value))
#
# Standard Object Types

3
py34/bacpypes/app.py
View File

@ -153,6 +153,9 @@ class Application(ApplicationServiceElement):
# keep track of the local device
self.localDevice = localDevice
# bind the device object to this application
localDevice._app = self
# allow the address to be cast to the correct type
if isinstance(localAddress, Address):
self.localAddress = localAddress

49
py34/bacpypes/consolecmd.py
View File

@ -40,14 +40,19 @@ def console_interrupt(*args):
@bacpypes_debugging
class ConsoleCmd(cmd.Cmd, Thread, Logging):
def __init__(self, prompt="> ", allow_exec=False, stdin=None, stdout=None):
def __init__(self, prompt="> ", stdin=None, stdout=None):
if _debug: ConsoleCmd._debug("__init__")
cmd.Cmd.__init__(self, stdin=stdin, stdout=stdout)
Thread.__init__(self, name="ConsoleCmd")
# save the prompt and exec option
self.prompt = prompt
self.allow_exec = allow_exec
# check to see if this is running interactive
self.interactive = sys.__stdin__.isatty()
# save the prompt for interactive sessions, otherwise be quiet
if self.interactive:
self.prompt = prompt
else:
self.prompt = ''
# gc counters
self.type2count = {}
@ -56,10 +61,6 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
# logging handlers
self.handlers = {}
# execution space for the user
self._locals = {}
self._globals = {}
# set a INT signal handler, ^C will only get sent to the
# main thread and there's no way to break the readline
# call initiated by this thread - sigh
@ -77,7 +78,7 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
if _debug: ConsoleCmd._debug(" - done cmdloop")
# tell the main thread to stop, this thread will exit
core.stop()
core.deferred(core.stop)
def onecmd(self, cmdString):
if _debug: ConsoleCmd._debug('onecmd %r', cmdString)
@ -233,15 +234,19 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
def do_exit(self, args):
"""Exits from the console."""
if _debug: ConsoleCmd._debug("do_exit %r", args)
return -1
def do_EOF(self, args):
"""Exit on system end of file character"""
if _debug: ConsoleCmd._debug("do_EOF %r", args)
return self.do_exit(args)
def do_shell(self, args):
"""Pass command to a system shell when line begins with '!'"""
if _debug: ConsoleCmd._debug("do_shell %r", args)
os.system(args)
def do_help(self, args):
@ -249,7 +254,9 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
'help' or '?' with no arguments prints a list of commands for which help is available
'help <command>' or '? <command>' gives help on <command>
"""
## The only reason to define this method is for the help text in the doc string
if _debug: ConsoleCmd._debug("do_help %r", args)
# the only reason to define this method is for the help text in the doc string
cmd.Cmd.do_help(self, args)
def preloop(self):
@ -274,10 +281,14 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
if not isinstance(err, IOError):
self.stdout.write("history error: %s\n" % err)
cmd.Cmd.postloop(self) ## Clean up command completion
# clean up command completion
cmd.Cmd.postloop(self)
self.stdout.write("Exiting...\n")
core.stop()
if self.interactive:
self.stdout.write("Exiting...\n")
# tell the core we have stopped
core.deferred(core.stop)
def precmd(self, line):
""" This method is called after the line has been input but before
@ -295,15 +306,3 @@ class ConsoleCmd(cmd.Cmd, Thread, Logging):
def emptyline(self):
"""Do nothing on empty input line"""
pass
def default(self, line):
"""Called on an input line when the command prefix is not recognized.
If allow_exec is enabled, execute the line as Python code.
"""
if not self.allow_exec:
return cmd.Cmd.default(self, line)
try:
exec(line) in self._locals, self._globals
except Exception as err:
self.stdout.write("%s : %s\n" % (err.__class__, err))

4
py34/bacpypes/object.py
View File

@ -183,7 +183,9 @@ class Property(Logging):
self.identifier, obj, value, arrayIndex, priority, direct
)
if (not direct):
if direct:
if _debug: Property._debug(" - direct write")
else:
# see if it must be provided
if not self.optional and value is None:
raise ValueError("%s value required" % (self.identifier,))

1172
samples/COVMixin.py Executable file
View File

File diff suppressed because it is too large Load Diff

67
samples/ConsoleCmd.py Executable file
View File

@ -0,0 +1,67 @@
#!/usr/bin/python
"""
This application is a template for applications that use the ConsoleCmd class.
"""
import sys
from bacpypes.debugging import bacpypes_debugging, ModuleLogger
from bacpypes.consolelogging import ArgumentParser
from bacpypes.consolecmd import ConsoleCmd
from bacpypes.core import run
# some debugging
_debug = 0
_log = ModuleLogger(globals())
# globals
this_console = None
#
# ConsoleCmdTemplate
#
class ConsoleCmdTemplate(ConsoleCmd):
def do_echo(self, args):
"""echo ..."""
args = args.split()
if _debug: ConsoleCmdTemplate._debug("do_echo %r", args)
sys.stdout.write(' '.join(args) + '\n')
bacpypes_debugging(ConsoleCmdTemplate)
def main():
global this_console
# build a parser for the command line arguments
parser = ArgumentParser(description=__doc__)
# sample additional argument to change the prompt
parser.add_argument(
"--prompt", type=str,
default="> ",
help="change the prompt",
)
# parse the command line arguments
args = parser.parse_args()
if _debug: _log.debug("initialization")
if _debug: _log.debug(" - args: %r", args)
# make a console
this_console = ConsoleCmdTemplate(prompt=args.prompt)
_log.debug("running")
run()
if __name__ == "__main__":
main()

233
samples/SubscribeCOV.py Executable file
View File

@ -0,0 +1,233 @@
#!/usr/bin/python
"""
This application presents a 'console' prompt to the user asking for read commands
which create ReadPropertyRequest PDUs, then lines up the coorresponding ReadPropertyACK
and prints the value.
"""
import sys
from bacpypes.debugging import bacpypes_debugging, ModuleLogger
from bacpypes.consolelogging import ConfigArgumentParser
from bacpypes.consolecmd import ConsoleCmd
from bacpypes.core import run
from bacpypes.pdu import Address
from bacpypes.app import LocalDeviceObject, BIPSimpleApplication
from bacpypes.object import get_object_class, get_datatype
from bacpypes.apdu import SubscribeCOVRequest, SimpleAckPDU, \
Error, RejectPDU, AbortPDU
from bacpypes.primitivedata import Unsigned
from bacpypes.constructeddata import Array
from bacpypes.basetypes import ServicesSupported
# some debugging
_debug = 0
_log = ModuleLogger(globals())
# globals
this_device = None
this_application = None
this_console = None
# how the application should respond
rsvp = (True, None, None)
#
# SubscribeCOVApplication
#
class SubscribeCOVApplication(BIPSimpleApplication):
def __init__(self, *args):
if _debug: SubscribeCOVApplication._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: SubscribeCOVApplication._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: SubscribeCOVApplication._debug("confirmation %r", apdu)
# continue normally
super(SubscribeCOVApplication, self).confirmation(apdu)
def indication(self, apdu):
if _debug: SubscribeCOVApplication._debug("indication %r", apdu)
# continue normally
super(SubscribeCOVApplication, self).indication(apdu)
def do_ConfirmedCOVNotificationRequest(self, apdu):
if _debug: SubscribeCOVApplication._debug("do_ConfirmedCOVNotificationRequest %r", apdu)
global rsvp
if rsvp[0]:
# success
response = SimpleAckPDU(context=apdu)
if _debug: SubscribeCOVApplication._debug(" - simple_ack: %r", response)
elif rsvp[1]:
# reject
response = RejectPDU(reason=rsvp[1], context=apdu)
if _debug: SubscribeCOVApplication._debug(" - reject: %r", response)
elif rsvp[2]:
# abort
response = AbortPDU(reason=rsvp[2], context=apdu)
if _debug: SubscribeCOVApplication._debug(" - abort: %r", response)
# return the result
self.response(response)
def do_UnconfirmedCOVNotificationRequest(self, apdu):
if _debug: SubscribeCOVApplication._debug("do_UnconfirmedCOVNotificationRequest %r", apdu)
bacpypes_debugging(SubscribeCOVApplication)
#
# SubscribeCOVConsoleCmd
#
class SubscribeCOVConsoleCmd(ConsoleCmd):
def do_subscribe(self, args):
"""subscribe addr proc_id obj_type obj_inst [ confirmed ] [ lifetime ]
"""
args = args.split()
if _debug: SubscribeCOVConsoleCmd._debug("do_subscribe %r", args)
try:
addr, proc_id, obj_type, obj_inst = args[:4]
proc_id = int(proc_id)
if obj_type.isdigit():
obj_type = int(obj_type)
elif not get_object_class(obj_type):
raise ValueError, "unknown object type"
obj_inst = int(obj_inst)
if len(args) >= 5:
issue_confirmed = args[4]
if issue_confirmed == '-':
issue_confirmed = None
else:
issue_confirmed = issue_confirmed.lower() == 'true'
if _debug: SubscribeCOVConsoleCmd._debug(" - issue_confirmed: %r", issue_confirmed)
else:
issue_confirmed = None
if len(args) >= 6:
lifetime = args[5]
if lifetime == '-':
lifetime = None
else:
lifetime = int(lifetime)
if _debug: SubscribeCOVConsoleCmd._debug(" - lifetime: %r", lifetime)
else:
lifetime = None
# build a request
request = SubscribeCOVRequest(
subscriberProcessIdentifier=proc_id,
monitoredObjectIdentifier=(obj_type, obj_inst),
)
request.pduDestination = Address(addr)
# optional parameters
if issue_confirmed is not None:
request.issueConfirmedNotifications = issue_confirmed
if lifetime is not None:
request.lifetime = lifetime
if _debug: SubscribeCOVConsoleCmd._debug(" - request: %r", request)
# give it to the application
this_application.request(request)
except Exception, e:
SubscribeCOVConsoleCmd._exception("exception: %r", e)
def do_ack(self, args):
"""ack
"""
args = args.split()
if _debug: SubscribeCOVConsoleCmd._debug("do_ack %r", args)
global rsvp
rsvp = (True, None, None)
def do_reject(self, args):
"""reject reason
"""
args = args.split()
if _debug: SubscribeCOVConsoleCmd._debug("do_subscribe %r", args)
global rsvp
rsvp = (False, args[0], None)
def do_abort(self, args):
"""abort reason
"""
args = args.split()
if _debug: SubscribeCOVConsoleCmd._debug("do_subscribe %r", args)
global rsvp
rsvp = (False, None, args[0])
bacpypes_debugging(SubscribeCOVConsoleCmd)
#
# __main__
#
try:
# 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 = 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()
_log.debug("running")
run()
except Exception, e:
_log.exception("an error has occurred: %s", e)
finally:
_log.debug("finally")

145
samples/TCPClient.py Normal file
View File

@ -0,0 +1,145 @@
#!/usr/bin/python
"""
This simple TCP client application connects to a server and sends the text
entered in the console. There is no conversion from incoming streams of
content into a line or any other higher-layer concept of a packet.
"""
from bacpypes.debugging import bacpypes_debugging, ModuleLogger
from bacpypes.core import run, stop
from bacpypes.task import TaskManager
from bacpypes.comm import PDU, Client, Server, bind, ApplicationServiceElement
from bacpypes.consolelogging import ArgumentParser
from bacpypes.console import ConsoleClient
from bacpypes.tcp import TCPClientDirector
# some debugging
_debug = 0
_log = ModuleLogger(globals())
# globals
server_address = None
# defaults
default_server_host = '127.0.0.1'
default_server_port = 9000
#
# MiddleMan
#
class MiddleMan(Client, Server):
"""
An instance of this class sits between the TCPClientDirector and the
console client. Downstream packets from a console have no concept of a
destination, so this is added to the PDUs before being sent to the
director. The source information in upstream packets is ignored by the
console client.
"""
def indication(self, pdu):
if _debug: MiddleMan._debug("indication %r", pdu)
global server_address
# no data means EOF, stop
if not pdu.pduData:
stop()
return
# pass it along
self.request(PDU(pdu.pduData, destination=server_address))
def confirmation(self, pdu):
if _debug: MiddleMan._debug("confirmation %r", pdu)
# pass it along
self.response(pdu)
bacpypes_debugging(MiddleMan)
#
# MiddleManASE
#
class MiddleManASE(ApplicationServiceElement):
def indication(self, addPeer=None, delPeer=None):
"""
This function is called by the TCPDirector when the client connects to
or disconnects from a server. It is called with addPeer or delPeer
keyword parameters, but not both.
"""
if _debug: MiddleManASE._debug('indication addPeer=%r delPeer=%r', addPeer, delPeer)
if addPeer:
if _debug: MiddleManASE._debug(" - add peer %s", addPeer)
if delPeer:
if _debug: MiddleManASE._debug(" - delete peer %s", delPeer)
# if there are no clients, quit
if not self.elementService.clients:
if _debug: MiddleManASE._debug(" - quitting")
stop()
bacpypes_debugging(MiddleManASE)
def main():
"""
Main function, called when run as an application.
"""
global server_address
# parse the command line arguments
parser = ArgumentParser(description=__doc__)
parser.add_argument(
"host", nargs='?',
help="address of host",
default=default_server_host,
)
parser.add_argument(
"port", nargs='?', type=int,
help="server port",
default=default_server_port,
)
args = parser.parse_args()
if _debug: _log.debug("initialization")
if _debug: _log.debug(" - args: %r", args)
# extract the server address and port
host = args.host
port = args.port
server_address = (host, port)
if _debug: _log.debug(" - server_address: %r", server_address)
# build the stack
this_console = ConsoleClient()
if _debug: _log.debug(" - this_console: %r", this_console)
this_middle_man = MiddleMan()
if _debug: _log.debug(" - this_middle_man: %r", this_middle_man)
this_director = TCPClientDirector()
if _debug: _log.debug(" - this_director: %r", this_director)
bind(this_console, this_middle_man, this_director)
bind(MiddleManASE(), this_director)
# create a task manager for scheduled functions
task_manager = TaskManager()
if _debug: _log.debug(" - task_manager: %r", task_manager)
# don't wait to connect
this_director.connect(server_address)
if _debug: _log.debug("running")
run()
if __name__ == "__main__":
main()

115
samples/TCPServer.py Executable file
View File

@ -0,0 +1,115 @@
#!/usr/bin/python
"""
This simple TCP server application listens for one or more client connections
and echos the incoming lines back to the client. There is no conversion from
incoming streams of content into a line or any other higher-layer concept
of a packet.
"""
import sys
import logging
from bacpypes.debugging import bacpypes_debugging, ModuleLogger
from bacpypes.consolelogging import ArgumentParser
from bacpypes.core import run
from bacpypes.comm import PDU, Client, bind, ApplicationServiceElement
from bacpypes.tcp import TCPServerDirector
# some debugging
_debug = 0
_log = ModuleLogger(globals())
# globals
server_address = None
# defaults
default_server_host = '127.0.0.1'
default_server_port = 9000
#
# EchoMaster
#
class EchoMaster(Client):
def confirmation(self, pdu):
if _debug: EchoMaster._debug('confirmation %r', pdu)
self.request(PDU(pdu.pduData, destination=pdu.pduSource))
bacpypes_debugging(EchoMaster)
#
# MiddleManASE
#
class MiddleManASE(ApplicationServiceElement):
def indication(self, addPeer=None, delPeer=None):
"""
This function is called by the TCPDirector when the client connects to
or disconnects from a server. It is called with addPeer or delPeer
keyword parameters, but not both.
"""
if _debug: MiddleManASE._debug('indication addPeer=%r delPeer=%r', addPeer, delPeer)
if addPeer:
if _debug: MiddleManASE._debug(" - add peer %s", addPeer)
if delPeer:
if _debug: MiddleManASE._debug(" - delete peer %s", delPeer)
bacpypes_debugging(MiddleManASE)
#
# __main__
#
def main():
# parse the command line arguments
parser = ArgumentParser(description=__doc__)
parser.add_argument(
"--host", nargs='?',
help="listening address of server",
default=default_server_host,
)
parser.add_argument(
"--port", nargs='?', type=int,
help="server port",
default=default_server_port,
)
args = parser.parse_args()
if _debug: _log.debug("initialization")
if _debug: _log.debug(" - args: %r", args)
# extract the server address and port
host = args.host
if host == "any":
host = ''
port = args.port
server_address = (host, port)
if _debug: _log.debug(" - server_address: %r", server_address)
# create a director listening to the address
this_director = TCPServerDirector(server_address)
if _debug: _log.debug(" - this_director: %r", this_director)
# create an echo
echo_master = EchoMaster()
if _debug: _log.debug(" - echo_master: %r", echo_master)
# bind everything together
bind(echo_master, this_director)
bind(MiddleManASE(), this_director)
_log.debug("running")
run()
if __name__ == "__main__":
main()