Send confirmed event notification samples

send-confirmed-event-notifications.json

[
    {
        "process-identifier": 1,
        "initiating-device-identifier": "device,999",
        "event-object-identifier": "analog-value,1",
        "time-stamp": {
            "time": "01:02:03"
        },
        "notification-class": 1,
        "priority": 0,
        "event-type": "floating-limit",
        "message-text": "optional message text",
        "notify-type": "alarm",
        "ack-required": false,
        "from-state": "normal",
        "to-state": "fault",
        "event-values": {
            "floating-limit": {
                "reference-value": 75.3,
                "status-flags": [0, 1, 0, 0],
                "setpoint-value": 75.0,
                "error-limit": 0.5
            }
        }
    }
]

send-confirmed-event-notifications.py

"""
Send Confirmed Event Notifications
"""

import asyncio
import json
import sys

from bacpypes3.debugging import ModuleLogger
from bacpypes3.argparse import SimpleArgumentParser
from bacpypes3.app import Application

from bacpypes3.pdu import Address
from bacpypes3.apdu import ConfirmedEventNotificationRequest, ErrorRejectAbortNack
from bacpypes3.json import json_to_sequence

# some debugging
_debug = 0
_log = ModuleLogger(globals())


async def main() -> None:
    app = None
    try:
        args = SimpleArgumentParser().parse_args()
        if _debug:
            _log.debug("args: %r", args)

        # build an application
        app = Application.from_args(args)
        if _debug:
            _log.debug("app: %r", app)

        with open("send-confirmed-event-notifications.json") as json_file:
            packet_samples = json.load(json_file)

        while sample_number := input("? "):
            # build the request
            apdu = json_to_sequence(
                packet_samples[int(sample_number)], ConfirmedEventNotificationRequest
            )

            # set the destination
            apdu.pduDestination = Address("10.0.1.70")
            if _debug:
                _log.debug("request: %r", apdu)

            try:
                response = await app.request(apdu)
                if _debug:
                    _log.debug("response: %r", response)
            except ErrorRejectAbortNack as err:
                sys.stderr.write(f"error/reject/abort/nack: {err}\n")

    finally:
        if app:
            app.close()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        if _debug:
            _log.debug("keyboard interrupt")