asyntest_py3_serial.py

# asyntest_py3_serial.py Test/demo of asychronous use of MessagePack under full python3 with pyserial

# Configured for pyserial. Link RX and TX pins.

# Copyright (c) 2024-2025 Peter Hinch Released under the MIT License see LICENSE

import asyncio
import umsgpack
import serial_asyncio


async def sender(swriter):
    obj = [
        1,
        True,
        False,
        0xFFFFFFFF,
        {"foo": b"\x80\x01\x02", "bar": [1, 2, 3, {"a": [1, 2, 3, {}]}]},
        -1,
        2.12345,
    ]

    while True:
        s = umsgpack.dumps(obj)
        swriter.write(s)
        await swriter.drain()
        await asyncio.sleep(5)
        obj[0] += 1


class stream_observer:
    def update(self, data: bytes) -> None:
        print(f"{data}")


async def receiver(sreader):
    recv_observer = stream_observer()
    while True:
        res = await umsgpack.aload(sreader, observer=recv_observer)
        print("Received:", res)


async def receiver_using_aloader(sreader):
    uart_aloader = umsgpack.aloader(sreader, observer=stream_observer())
    while True:
        res = await uart_aloader.load()
        print("Received (aloader):", res)


async def main():
    reader, writer = await serial_asyncio.open_serial_connection(url="/dev/ttyUSB0", baudrate=9600)
    asyncio.create_task(sender(writer))
    # asyncio.create_task(receiver(reader))
    asyncio.create_task(receiver_using_aloader(reader))
    while True:
        print("running...")
        await asyncio.sleep(20)


def test():
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Interrupted")


test()