Async SEN0395 Sensor

serio.py

import asyncio

import serialio

from sensation.sen0395 import SensorAsync, PresenceHandlerAsync


class Printer:

    async def __call__(self, p):
        print(f"printer async: {p}")


async def main():
    sensor = None
    try:
        serial_con = serialio.serial_for_url("serial:///dev/ttyAMA1", 115200)
        sensor = SensorAsync("my_sensor", serial_con)
        await serial_con.open()

        presence_handler = PresenceHandlerAsync()
        presence_handler.observers.append(Printer().__call__)
        sensor.handlers.append(presence_handler)

        await sensor.clear_buffer()
        sensor.start_reading()

        print(await sensor.configure_latency(0, 0))

        await asyncio.sleep(5)

        await sensor.stop_reading()

    finally:
        if sensor:
            await sensor.close()


asyncio.run(main())