mirror of
https://github.com/JoelBender/bacpypes
synced 2025-10-27 00:57:47 +08:00
merge #207 into stage for next release
This commit is contained in:
Joel Bender
@@ -1,9 +1,9 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
"""
|
||||
This application presents a 'console' prompt to the user asking for readrange commands
|
||||
which create ReadRangeRequest PDUs, then lines up the coorresponding ReadRangeACK
|
||||
and prints the value.
|
||||
This application presents a 'console' prompt to the user asking for readrange
|
||||
commands which create ReadRangeRequest PDUs, then lines up the coorresponding
|
||||
ReadRangeACK and prints the value.
|
||||
"""
|
||||
|
||||
import sys
|
||||
@@ -17,10 +17,19 @@ from bacpypes.iocb import IOCB
|
||||
|
||||
from bacpypes.pdu import Address
|
||||
from bacpypes.object import get_datatype
|
||||
from bacpypes.apdu import ReadRangeRequest, ReadRangeACK
|
||||
from bacpypes.apdu import (
|
||||
ReadRangeRequest,
|
||||
Range,
|
||||
RangeByPosition,
|
||||
RangeBySequenceNumber,
|
||||
RangeByTime,
|
||||
ReadRangeACK,
|
||||
)
|
||||
|
||||
from bacpypes.app import BIPSimpleApplication
|
||||
from bacpypes.primitivedata import ObjectIdentifier
|
||||
from bacpypes.primitivedata import Date, Time, ObjectIdentifier
|
||||
from bacpypes.constructeddata import Array, List
|
||||
from bacpypes.basetypes import DateTime
|
||||
from bacpypes.local.device import LocalDeviceObject
|
||||
|
||||
# some debugging
|
||||
@@ -34,17 +43,23 @@ this_application = None
|
||||
# ReadRangeConsoleCmd
|
||||
#
|
||||
|
||||
|
||||
@bacpypes_debugging
|
||||
class ReadRangeConsoleCmd(ConsoleCmd):
|
||||
|
||||
def do_readrange(self, args):
|
||||
"""readrange <addr> <objid> <prop> [ <indx> ]"""
|
||||
"""readrange <addr> <objid> <prop> [ <indx> ]
|
||||
[ p <indx> <count> ]
|
||||
[ s <seq> <count> ]
|
||||
[ t <date> <time> <count> ]
|
||||
"""
|
||||
args = args.split()
|
||||
if _debug: ReadRangeConsoleCmd._debug("do_readrange %r", args)
|
||||
if _debug:
|
||||
ReadRangeConsoleCmd._debug("do_readrange %r", args)
|
||||
|
||||
try:
|
||||
addr, obj_id, prop_id = args[:3]
|
||||
obj_id = ObjectIdentifier(obj_id).value
|
||||
addr = Address(args.pop(0))
|
||||
obj_id = ObjectIdentifier(args.pop(0)).value
|
||||
prop_id = args.pop(0)
|
||||
|
||||
datatype = get_datatype(obj_id[0], prop_id)
|
||||
if not datatype:
|
||||
@@ -52,18 +67,49 @@ class ReadRangeConsoleCmd(ConsoleCmd):
|
||||
|
||||
# build a request
|
||||
request = ReadRangeRequest(
|
||||
objectIdentifier=obj_id,
|
||||
propertyIdentifier=prop_id,
|
||||
)
|
||||
request.pduDestination = Address(addr)
|
||||
destination=addr,
|
||||
objectIdentifier=obj_id, propertyIdentifier=prop_id
|
||||
)
|
||||
|
||||
if len(args) == 4:
|
||||
request.propertyArrayIndex = int(args[3])
|
||||
if _debug: ReadRangeConsoleCmd._debug(" - request: %r", request)
|
||||
# index is optional
|
||||
if args:
|
||||
if args[0].isdigit():
|
||||
if not issubclass(datatype, Array):
|
||||
raise ValueError("property is not an array")
|
||||
request.propertyArrayIndex = int(args.pop(0))
|
||||
datatype = datatype.subtype
|
||||
if not issubclass(datatype, List):
|
||||
raise ValueError("property is not a list")
|
||||
|
||||
# range is optional
|
||||
if args:
|
||||
range_type = args.pop(0)
|
||||
if range_type == "p":
|
||||
rbp = RangeByPosition(
|
||||
referenceIndex=int(args[0]), count=int(args[1])
|
||||
)
|
||||
request.range = Range(byPosition=rbp)
|
||||
elif range_type == "s":
|
||||
rbs = RangeBySequenceNumber(
|
||||
referenceSequenceNumber=int(args[0]), count=int(args[1])
|
||||
)
|
||||
request.range = Range(bySequenceNumber=rbs)
|
||||
elif range_type == "t":
|
||||
rbt = RangeByTime(
|
||||
referenceTime=DateTime(date=Date(args[0]), time=Time(args[1])),
|
||||
count=int(args[2]),
|
||||
)
|
||||
request.range = Range(byTime=rbt)
|
||||
else:
|
||||
raise ValueError("unknown range type: %r" % (range_type,))
|
||||
|
||||
if _debug:
|
||||
ReadRangeConsoleCmd._debug(" - request: %r", request)
|
||||
|
||||
# make an IOCB
|
||||
iocb = IOCB(request)
|
||||
if _debug: ReadRangeConsoleCmd._debug(" - iocb: %r", iocb)
|
||||
if _debug:
|
||||
ReadRangeConsoleCmd._debug(" - iocb: %r", iocb)
|
||||
|
||||
# give it to the application
|
||||
deferred(this_application.request_io, iocb)
|
||||
@@ -74,20 +120,29 @@ class ReadRangeConsoleCmd(ConsoleCmd):
|
||||
# do something for success
|
||||
if iocb.ioResponse:
|
||||
apdu = iocb.ioResponse
|
||||
if _debug:
|
||||
ReadRangeConsoleCmd._debug(" - apdu: %r", apdu)
|
||||
|
||||
# should be an ack
|
||||
if not isinstance(apdu, ReadRangeACK):
|
||||
if _debug: ReadRangeConsoleCmd._debug(" - not an ack")
|
||||
if _debug:
|
||||
ReadRangeConsoleCmd._debug(" - not an ack")
|
||||
return
|
||||
|
||||
# find the datatype
|
||||
datatype = get_datatype(apdu.objectIdentifier[0], apdu.propertyIdentifier)
|
||||
if _debug: ReadRangeConsoleCmd._debug(" - datatype: %r", datatype)
|
||||
datatype = get_datatype(
|
||||
apdu.objectIdentifier[0], apdu.propertyIdentifier
|
||||
)
|
||||
if _debug:
|
||||
ReadRangeConsoleCmd._debug(" - datatype: %r", datatype)
|
||||
if not datatype:
|
||||
raise TypeError("unknown datatype")
|
||||
|
||||
sys.stdout.write("firstSequenceNumber: %s\n" % (apdu.firstSequenceNumber,))
|
||||
sys.stdout.write("resultFlags: %s\n" % (apdu.resultFlags,))
|
||||
|
||||
# cast out the data into a list
|
||||
value = apdu.itemData[0].cast_out(datatype)
|
||||
value = apdu.itemData.cast_out(datatype)
|
||||
|
||||
# dump it out
|
||||
for i, item in enumerate(value):
|
||||
@@ -97,34 +152,40 @@ class ReadRangeConsoleCmd(ConsoleCmd):
|
||||
|
||||
# do something for error/reject/abort
|
||||
if iocb.ioError:
|
||||
sys.stdout.write(str(iocb.ioError) + '\n')
|
||||
sys.stdout.write(str(iocb.ioError) + "\n")
|
||||
|
||||
except Exception as error:
|
||||
ReadRangeConsoleCmd._exception("exception: %r", error)
|
||||
|
||||
|
||||
#
|
||||
# __main__
|
||||
#
|
||||
|
||||
|
||||
def main():
|
||||
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)
|
||||
if _debug:
|
||||
_log.debug("initialization")
|
||||
if _debug:
|
||||
_log.debug(" - args: %r", args)
|
||||
|
||||
# make a device object
|
||||
this_device = LocalDeviceObject(ini=args.ini)
|
||||
if _debug: _log.debug(" - this_device: %r", this_device)
|
||||
if _debug:
|
||||
_log.debug(" - this_device: %r", this_device)
|
||||
|
||||
# make a simple application
|
||||
this_application = BIPSimpleApplication(this_device, args.ini.address)
|
||||
|
||||
# make a console
|
||||
this_console = ReadRangeConsoleCmd()
|
||||
if _debug: _log.debug(" - this_console: %r", this_console)
|
||||
if _debug:
|
||||
_log.debug(" - this_console: %r", this_console)
|
||||
|
||||
# enable sleeping will help with threads
|
||||
enable_sleeping()
|
||||
@@ -135,5 +196,6 @@ def main():
|
||||
|
||||
_log.debug("fini")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
148
samples/ReadRangeServer.py
Executable file
148
samples/ReadRangeServer.py
Executable file
@@ -0,0 +1,148 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
"""
|
||||
Server with a Trend Log Object
|
||||
"""
|
||||
|
||||
from bacpypes.debugging import bacpypes_debugging, ModuleLogger
|
||||
from bacpypes.consolelogging import ConfigArgumentParser
|
||||
|
||||
from bacpypes.core import run
|
||||
from bacpypes.errors import ExecutionError
|
||||
from bacpypes.primitivedata import Date, Time
|
||||
from bacpypes.constructeddata import Array, List, SequenceOfAny
|
||||
from bacpypes.basetypes import DateTime, LogRecord, LogRecordLogDatum, StatusFlags
|
||||
from bacpypes.apdu import ReadRangeACK
|
||||
|
||||
from bacpypes.app import BIPSimpleApplication
|
||||
from bacpypes.local.device import LocalDeviceObject
|
||||
from bacpypes.object import PropertyError, TrendLogObject
|
||||
|
||||
|
||||
# some debugging
|
||||
_debug = 0
|
||||
_log = ModuleLogger(globals())
|
||||
|
||||
# globals
|
||||
this_application = None
|
||||
|
||||
|
||||
#
|
||||
# ReadRangeApplication
|
||||
#
|
||||
|
||||
|
||||
@bacpypes_debugging
|
||||
class ReadRangeApplication(BIPSimpleApplication):
|
||||
def __init__(self, *args):
|
||||
if _debug:
|
||||
ReadRangeApplication._debug("__init__ %r", args)
|
||||
BIPSimpleApplication.__init__(self, *args)
|
||||
|
||||
def do_ReadRangeRequest(self, apdu):
|
||||
if _debug:
|
||||
ReadRangeApplication._debug("do_ReadRangeRequest %r", apdu)
|
||||
|
||||
# extract the object identifier
|
||||
objId = apdu.objectIdentifier
|
||||
|
||||
# get the object
|
||||
obj = self.get_object_id(objId)
|
||||
if _debug:
|
||||
ReadRangeApplication._debug(" - object: %r", obj)
|
||||
|
||||
if not obj:
|
||||
raise ExecutionError(errorClass="object", errorCode="unknownObject")
|
||||
|
||||
# get the datatype
|
||||
datatype = obj.get_datatype(apdu.propertyIdentifier)
|
||||
if _debug:
|
||||
ReadRangeApplication._debug(" - datatype: %r", datatype)
|
||||
|
||||
# must be a list, or an array of lists
|
||||
if issubclass(datatype, List):
|
||||
pass
|
||||
elif (
|
||||
(apdu.propertyArrayIndex is not None)
|
||||
and issubclass(datatype, Array)
|
||||
and issubclass(datatype.subtype, List)
|
||||
):
|
||||
pass
|
||||
else:
|
||||
raise ExecutionError(errorClass="property", errorCode="propertyIsNotAList")
|
||||
|
||||
# get the value
|
||||
value = obj.ReadProperty(apdu.propertyIdentifier, apdu.propertyArrayIndex)
|
||||
if _debug:
|
||||
ReadRangeApplication._debug(" - value: %r", value)
|
||||
if value is None:
|
||||
raise PropertyError(apdu.propertyIdentifier)
|
||||
|
||||
# this is an ack
|
||||
resp = ReadRangeACK(context=apdu)
|
||||
resp.objectIdentifier = objId
|
||||
resp.propertyIdentifier = apdu.propertyIdentifier
|
||||
resp.propertyArrayIndex = apdu.propertyArrayIndex
|
||||
|
||||
resp.resultFlags = [1, 1, 0]
|
||||
resp.itemCount = len(value)
|
||||
|
||||
# save the result in the item data
|
||||
resp.itemData = SequenceOfAny()
|
||||
resp.itemData.cast_in(datatype(value))
|
||||
if _debug:
|
||||
ReadRangeApplication._debug(" - resp: %r", resp)
|
||||
|
||||
# return the result
|
||||
self.response(resp)
|
||||
|
||||
|
||||
#
|
||||
# __main__
|
||||
#
|
||||
|
||||
|
||||
def main():
|
||||
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(ini=args.ini)
|
||||
if _debug:
|
||||
_log.debug(" - this_device: %r", this_device)
|
||||
|
||||
# make a simple application
|
||||
this_application = ReadRangeApplication(this_device, args.ini.address)
|
||||
|
||||
timestamp = DateTime(date=Date().now().value, time=Time().now().value)
|
||||
# log_status = LogStatus([0,0,0])
|
||||
log_record_datum = LogRecordLogDatum(booleanValue=False)
|
||||
status_flags = StatusFlags([0, 0, 0, 0])
|
||||
log_record = LogRecord(
|
||||
timestamp=timestamp, logDatum=log_record_datum, statusFlags=status_flags
|
||||
)
|
||||
|
||||
trend_log_object = TrendLogObject(
|
||||
objectIdentifier=("trendLog", 1),
|
||||
objectName="Trend-Log-1",
|
||||
logBuffer=[log_record],
|
||||
)
|
||||
_log.debug(" - trend_log_object: %r", trend_log_object)
|
||||
this_application.add_object(trend_log_object)
|
||||
|
||||
_log.debug("running")
|
||||
|
||||
run()
|
||||
|
||||
_log.debug("fini")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user