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

add a traffic log, application needs IOController for sending notifications

Browse Source
This commit is contained in:
Joel Bender 2018-06-22 01:16:43 -04:00
parent 5078d59a15
commit 1fdfea1f0b
1 changed files with 11 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
tests/test_service/helpers.py
View File

@ -10,12 +10,12 @@ from bacpypes.comm import Client, bind
from bacpypes.pdu import Address, LocalBroadcast
from bacpypes.vlan import Network, Node
from bacpypes.app import Application
from bacpypes.app import ApplicationIOController
from bacpypes.appservice import StateMachineAccessPoint, ApplicationServiceAccessPoint
from bacpypes.netservice import NetworkServiceAccessPoint, NetworkServiceElement
from bacpypes.local.device import LocalDeviceObject
from ..state_machine import StateMachine, StateMachineGroup
from ..state_machine import StateMachine, StateMachineGroup, TrafficLog
from ..time_machine import reset_time_machine, run_time_machine
@ -39,8 +39,12 @@ class ApplicationNetwork(StateMachineGroup):
reset_time_machine()
if _debug: ApplicationNetwork._debug(" - time machine reset")
# create a traffic log
self.traffic_log = TrafficLog()
# make a little LAN
self.vlan = Network(broadcast_address=LocalBroadcast())
self.vlan.traffic_log = self.traffic_log
# test device object
self.td_device_object = LocalDeviceObject(
@ -84,6 +88,9 @@ class ApplicationNetwork(StateMachineGroup):
for direction, pdu in state_machine.transaction_log:
ApplicationNetwork._debug(" %s %s", direction, str(pdu))
# traffic log has what was processed on each vlan
self.traffic_log.dump(ApplicationNetwork._debug)
# check for success
all_success, some_failed = super(ApplicationNetwork, self).check_for_success()
ApplicationNetwork._debug(" - all_success, some_failed: %r, %r", all_success, some_failed)
@ -131,7 +138,7 @@ class SnifferNode(Client, StateMachine):
#
@bacpypes_debugging
class ApplicationStateMachine(Application, StateMachine):
class ApplicationStateMachine(ApplicationIOController, StateMachine):
def __init__(self, localDevice, vlan):
if _debug: ApplicationStateMachine._debug("__init__ %r %r", localDevice, vlan)
@ -141,7 +148,7 @@ class ApplicationStateMachine(Application, StateMachine):
if _debug: ApplicationStateMachine._debug(" - address: %r", self.address)
# continue with initialization
Application.__init__(self, localDevice, self.address)
ApplicationIOController.__init__(self, localDevice, self.address)
StateMachine.__init__(self, name=localDevice.objectName)
# include a application decoder