mini_device revisited

gistfile1.txt

import asyncio

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

from bacpypes3.app import Application
from bacpypes3.local.analog import AnalogValueObject
from bacpypes3.local.binary import BinaryValueObject

_debug = 0
_log = ModuleLogger(globals())

INTERVAL = 1.0

# globals
test_av = None
test_bv = None


class SampleApplication(Application):
    async def update_values(self):
        test_values = [
            ("active", 1.0),
            ("inactive", 2.0),
            ("active", 3.0),
            ("inactive", 4.0),
        ]

        while True:
            await asyncio.sleep(INTERVAL)
            next_value = test_values.pop(0)
            test_values.append(next_value)
            if _debug:
                _log.debug("    - next_value: %r", next_value)

            test_av.presentValue = next_value[1]
            test_bv.presentValue = next_value[0]


async def main():
    global test_av, test_bv

    args = SimpleArgumentParser().parse_args()
    if _debug:
        _log.debug("args: %r", args)

    app = SampleApplication.from_args(args)
    if _debug:
        _log.debug("app: %r", app)

    test_av = AnalogValueObject(
        objectIdentifier=("analogValue", 1),
        objectName="av",
        presentValue=0.0,
        statusFlags=[0, 0, 0, 0],
        covIncrement=1.0,
    )
    _log.debug("    - test_av: %r", test_av)
    app.add_object(test_av)

    test_bv = BinaryValueObject(
        objectIdentifier=("binaryValue", 1),
        objectName="bv",
        presentValue="inactive",
        statusFlags=[0, 0, 0, 0],
    )
    _log.debug("    - test_bv: %r", test_bv)
    app.add_object(test_bv)

    asyncio.create_task(app.update_values())

    await asyncio.Future()


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