mirror/bacpypes
1
0
Fork 0
mirror of https://github.com/JoelBender/bacpypes synced 2025-10-27 00:57:47 +08:00
Code Issues Releases Wiki Activity

merge in 78 to incorporate testing pieces

Browse Source
This commit is contained in:
Joel Bender
2017-09-19 21:39:01 -04:00
parent 1592589fc3 d637b802c6
commit ae96b548c6
52 changed files with 2635 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
py25/bacpypes/apdu.py
View File

@@ -1495,7 +1495,7 @@ class DeviceCommunicationControlRequestEnableDisable(Enumerated):
enumerations = \
{ 'enable':0
, 'disable':1
, 'defaultInitiation':2
, 'disableInitiation':2
}
class DeviceCommunicationControlRequest(ConfirmedRequestSequence):

45
py25/bacpypes/appservice.py
View File

@@ -11,7 +11,7 @@ from .debugging import ModuleLogger, DebugContents, bacpypes_debugging
from .comm import Client, ServiceAccessPoint, ApplicationServiceElement
from .task import OneShotTask
from .pdu import Address, LocalStation, RemoteStation
from .pdu import Address
from .apdu import AbortPDU, AbortReason, ComplexAckPDU, \
ConfirmedRequestPDU, Error, ErrorPDU, RejectPDU, SegmentAckPDU, \
SimpleAckPDU, UnconfirmedRequestPDU, apdu_types, \
@@ -85,7 +85,7 @@ class SSM(OneShotTask, DebugContents):
self.suspend_task()
# now install this
self.install_task(_time() + (msecs / 1000.0))
self.install_task(delta=msecs / 1000.0)
def stop_timer(self):
if _debug: SSM._debug("stop_timer")
@@ -104,7 +104,7 @@ class SSM(OneShotTask, DebugContents):
self.suspend_task()
# now install this
self.install_task(_time() + (msecs / 1000.0))
self.install_task(delta=msecs / 1000.0)
def set_state(self, newState, timer=0):
"""This function is called when the derived class wants to change state."""
@@ -1083,6 +1083,9 @@ class StateMachineAccessPoint(Client, ServiceAccessPoint):
self.segmentTimeout = 1500
self.maxSegmentsAccepted = 8
# device communication control
self.dccEnableDisable = 'enable'
# local device object provides these
if localDevice:
self.retryCount = localDevice.numberOfApduRetries
@@ -1121,6 +1124,22 @@ class StateMachineAccessPoint(Client, ServiceAccessPoint):
"""Packets coming up the stack are APDU's."""
if _debug: StateMachineAccessPoint._debug("confirmation %r", pdu)
# check device communication control
if self.dccEnableDisable == 'enable':
if _debug: StateMachineAccessPoint._debug(" - communications enabled")
elif self.dccEnableDisable == 'disable':
if (pdu.apduType == 0) and (pdu.apduService == 17):
if _debug: StateMachineAccessPoint._debug(" - continue with DCC request")
elif (pdu.apduType == 0) and (pdu.apduService == 20):
if _debug: StateMachineAccessPoint._debug(" - continue with reinitialize device")
elif (pdu.apduType == 1) and (pdu.apduService == 8):
if _debug: StateMachineAccessPoint._debug(" - continue with Who-Is")
else:
if _debug: StateMachineAccessPoint._debug(" - not a Who-Is, dropped")
return
elif self.dccEnableDisable == 'disableInitiation':
if _debug: StateMachineAccessPoint._debug(" - initiation disabled")
# make a more focused interpretation
atype = apdu_types.get(pdu.apduType)
if not atype:
@@ -1219,6 +1238,23 @@ class StateMachineAccessPoint(Client, ServiceAccessPoint):
a new transaction as a client."""
if _debug: StateMachineAccessPoint._debug("sap_indication %r", apdu)
# check device communication control
if self.dccEnableDisable == 'enable':
if _debug: StateMachineAccessPoint._debug(" - communications enabled")
elif self.dccEnableDisable == 'disable':
if _debug: StateMachineAccessPoint._debug(" - communications disabled")
return
elif self.dccEnableDisable == 'disableInitiation':
if _debug: StateMachineAccessPoint._debug(" - initiation disabled")
if (apdu.apduType == 1) and (apdu.apduService == 0):
if _debug: StateMachineAccessPoint._debug(" - continue with I-Am")
else:
if _debug: StateMachineAccessPoint._debug(" - not an I-Am")
return
if isinstance(apdu, UnconfirmedRequestPDU):
# deliver to the device
self.request(apdu)
@@ -1408,8 +1444,8 @@ class ApplicationServiceAccessPoint(ApplicationServiceElement, ServiceAccessPoin
# 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)
@@ -1487,3 +1523,4 @@ class ApplicationServiceAccessPoint(ApplicationServiceElement, ServiceAccessPoin
self.response(xpdu)
bacpypes_debugging(ApplicationServiceAccessPoint)

4
py25/bacpypes/bvllservice.py
View File

@@ -493,7 +493,7 @@ class BIPForeign(BIPSAP, Client, Server, OneShotTask, DebugContents):
# check for success
if pdu.bvlciResultCode == 0:
# schedule for a refresh
self.install_task(_time() + self.bbmdTimeToLive)
self.install_task(delta=self.bbmdTimeToLive)
return
@@ -541,7 +541,7 @@ class BIPForeign(BIPSAP, Client, Server, OneShotTask, DebugContents):
self.bbmdTimeToLive = ttl
# install this task to run when it gets a chance
self.install_task(0)
self.install_task(delta=0)
def unregister(self):
"""Drop the registration with a BBMD."""

4
py25/bacpypes/core.py
View File

@@ -161,7 +161,7 @@ def run(spin=SPIN, sigterm=stop, sigusr1=print_stack):
# call the functions
for fn, args, kwargs in fnlist:
# if _debug: run._debug(" - call: %r %r %r", fn, args, kwargs)
fn( *args, **kwargs)
fn(*args, **kwargs)
# done with this list
del fnlist
@@ -212,7 +212,7 @@ def run_once():
# call the functions
for fn, args, kwargs in fnlist:
if _debug: run_once._debug(" - call: %r %r %r", fn, args, kwargs)
fn( *args, **kwargs)
fn(*args, **kwargs)
# done with this list
del fnlist

6
py25/bacpypes/debugging.py
View File

@@ -5,6 +5,7 @@ Debugging
"""
import sys
import re
import logging
import binascii
from cStringIO import StringIO
@@ -40,9 +41,8 @@ def btox(data, sep=''):
def xtob(data, sep=''):
"""Interpret the hex encoding of a blob (string)."""
# if there is a separator, remove it
if sep:
data = ''.join(data.split(sep))
# remove the non-hex characters
data = re.sub("[^0-9a-fA-F]", '', data)
# interpret the hex
return binascii.unhexlify(data)

4
py25/bacpypes/iocb.py
View File

@@ -208,7 +208,7 @@ class IOCB(DebugContents):
self.ioTimeout = FunctionTask(self.abort, err)
# (re)schedule it
self.ioTimeout.install_task(_time() + delay)
self.ioTimeout.install_task(delay=delay)
def __repr__(self):
xid = id(self)
@@ -767,7 +767,7 @@ class IOQController(IOController):
# schedule a call in the future
task = FunctionTask(IOQController._wait_trigger, self)
task.install_task(_time() + self.wait_time)
task.install_task(delay=self.wait_time)
else:
# change our state

2
py25/bacpypes/service/cov.py
View File

@@ -304,7 +304,7 @@ class COVIncrementCriteria(COVDetection):
# continue
COVDetection.send_cov_notifications(self)
bacpypes_debugging(COVIncrementCriteria)
bacpypes_debugging(COVDetection)
class AccessDoorCriteria(COVDetection):

97
py25/bacpypes/service/device.py
View File

@@ -7,11 +7,12 @@ from ..pdu import GlobalBroadcast
from ..primitivedata import Date, Time, ObjectIdentifier
from ..constructeddata import ArrayOf
from ..apdu import WhoIsRequest, IAmRequest, IHaveRequest
from ..apdu import WhoIsRequest, IAmRequest, IHaveRequest, SimpleAckPDU, Error
from ..errors import ExecutionError, InconsistentParameters, \
MissingRequiredParameter, ParameterOutOfRange
from ..object import register_object_type, registered_object_types, \
Property, DeviceObject
from ..task import FunctionTask
# some debugging
_debug = 0
@@ -89,6 +90,11 @@ class LocalDeviceObject(DeviceObject):
if attr not in kwargs:
kwargs[attr] = value
for key, value in kwargs.items():
if key.startswith("_"):
setattr(self, key, value)
del kwargs[key]
# check for registration
if self.__class__ not in registered_object_types.values():
if 'vendorIdentifier' not in kwargs:
@@ -124,6 +130,8 @@ class LocalDeviceObject(DeviceObject):
if 'objectList' not in self.propertyList:
self.propertyList.append('objectList')
bacpypes_debugging(LocalDeviceObject)
#
# Who-Is I-Am Services
#
@@ -286,6 +294,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)
@@ -293,8 +323,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)
@@ -338,3 +370,64 @@ class WhoHasIHaveServices(Capability):
### check to see if the application is looking for this object
bacpypes_debugging(WhoHasIHaveServices)
#
# Device Communication Control
#
class DeviceCommunicationControlServices(Capability):
def __init__(self):
if _debug: DeviceCommunicationControlServices._debug("__init__")
Capability.__init__(self)
# task to run if there is a time duration
self._dcc_enable_task = None
def do_DeviceCommunicationControlRequest(self, apdu):
if _debug: DeviceCommunicationControlServices._debug("do_CommunicationControlRequest, %r", apdu)
if getattr(self.localDevice, "_dcc_password", None):
if not apdu.password or apdu.password != getattr(self.localDevice, "_dcc_password"):
raise ExecutionError(errorClass="security", errorCode="passwordFailure")
if apdu.enableDisable == "enable":
self.enable_communications()
else:
# disable or disableInitiation
self.disable_communications(apdu.enableDisable)
# if there is a time duration, it's in minutes
if apdu.timeDuration:
self._dcc_enable_task = FunctionTask(self.enable_communications)
self._dcc_enable_task.install_task(delta=apdu.timeDuration * 60)
if _debug: DeviceCommunicationControlServices._debug(" - enable scheduled")
# respond with a simple ack
self.response(SimpleAckPDU(context=apdu))
def enable_communications(self):
if _debug: DeviceCommunicationControlServices._debug("enable_communications")
# tell the State Machine Access Point
self.smap.dccEnableDisable = 'enable'
# if an enable task was scheduled, cancel it
if self._dcc_enable_task:
self._dcc_enable_task.suspend_task()
self._dcc_enable_task = None
def disable_communications(self, enable_disable):
if _debug: DeviceCommunicationControlServices._debug("disable_communications %r", enable_disable)
# tell the State Machine Access Point
self.smap.dccEnableDisable = enable_disable
# if an enable task was scheduled, cancel it
if self._dcc_enable_task:
self._dcc_enable_task.suspend_task()
self._dcc_enable_task = None
bacpypes_debugging(DeviceCommunicationControlServices)

2
py25/bacpypes/task.py
View File

@@ -135,7 +135,7 @@ def OneShotFunction(fn, *args, **kwargs):
if not _task_manager:
_unscheduled_tasks.append(task)
else:
task.install_task(_task_manager.get_time())
task.install_task(delta=0)
return task

36
py25/bacpypes/vlan.py
View File

@@ -12,9 +12,9 @@ from copy import deepcopy
from .errors import ConfigurationError
from .debugging import ModuleLogger, bacpypes_debugging
from .core import deferred
from .pdu import Address
from .comm import Client, Server, bind
from .task import OneShotFunction
# some debugging
_debug = 0
@@ -56,7 +56,7 @@ class Network:
""" Process a PDU by sending a copy to each node as dictated by the
addressing and if a node is promiscuous.
"""
if _debug: Network._debug("[%s]process_pdu %r", self.name, pdu)
if _debug: Network._debug("process_pdu(%s) %r", self.name, pdu)
# randomly drop a packet
if self.drop_percent != 0.0:
@@ -65,19 +65,20 @@ class Network:
return
if pdu.pduDestination == self.broadcast_address:
for n in self.nodes:
if (pdu.pduSource != n.address):
if _debug: Network._debug(" - match: %r", n)
n.response(deepcopy(pdu))
if _debug: Network._debug(" - broadcast")
for node in self.nodes:
if (pdu.pduSource != node.address):
if _debug: Network._debug(" - match: %r", node)
node.response(deepcopy(pdu))
else:
for n in self.nodes:
if n.promiscuous or (pdu.pduDestination == n.address):
if _debug: Network._debug(" - match: %r", n)
n.response(deepcopy(pdu))
if _debug: Network._debug(" - unicast")
for node in self.nodes:
if node.promiscuous or (pdu.pduDestination == node.address):
if _debug: Network._debug(" - match: %r", node)
node.response(deepcopy(pdu))
def __len__(self):
""" Simple way to determine the number of nodes in the network. """
if _debug: Network._debug("__len__")
return len(self.nodes)
bacpypes_debugging(Network)
@@ -115,7 +116,7 @@ class Node(Server):
def indication(self, pdu):
"""Send a message."""
if _debug: Node._debug("[%s]indication %r", self.name, pdu)
if _debug: Node._debug("indication(%s) %r", self.name, pdu)
# make sure we're connected
if not self.lan:
@@ -128,8 +129,15 @@ class Node(Server):
elif (not self.spoofing) and (pdu.pduSource != self.address):
raise RuntimeError("spoofing address conflict")
# actual network delivery is deferred
deferred(self.lan.process_pdu, pdu)
# actual network delivery is a zero-delay task
OneShotFunction(self.lan.process_pdu, pdu)
def __repr__(self):
return "<%s(%s) at %s>" % (
self.__class__.__name__,
self.name,
hex(id(self)),
)
bacpypes_debugging(Node)