Build a device library

Package your device models with a public collection that accepts a Connection. Keep decoding, units, and command encoding in the library. Applications supply the connection and display the data.

Organize the package

Add lorawan-connection as a dependency in your package’s pyproject.toml. Use a devices/ folder with one file per model, such as devices/s2101.py. Keep each model and its decoder together. For a SenseCAP library, the layout could be:

pyproject.toml
sensecap_lorawan/
  __init__.py       # collection and public exports
  __main__.py       # command-line entry point
  devices/
    __init__.py
    s2101.py        # S2101 model and decoder
tests/
  conftest.py       # collection fixture
  devices/
    test_s2101.py

The quickstart example contains the S2101 model and decoder. Put those in devices/s2101.py. Move decoders shared by several models into a separate module.

Expose a collection

Define the collection in __init__.py, alongside the public model exports:

from lorawan_connection import DeviceCollection

from .devices.s2101 import S2101


class SenseCapDeviceCollection(DeviceCollection[S2101]):
    DEVICES = (S2101,)


__all__ = ["S2101", "SenseCapDeviceCollection"]

For several device classes, use their union as the collection’s type parameter and list them in DEVICES. Keep the collection exported from __init__.py if you move its implementation to another module.

Keep the application boundary clear

Keep backend imports in application code such as the CLI. Device models use the shared event and downlink interfaces.

An application with an open connection can call monitor() below. Cancelling it closes the collection while leaving the connection open. See Connecting to ChirpStack for connection setup.

import asyncio

from lorawan_connection import Connection
from sensecap_lorawan import S2101, SenseCapDeviceCollection


async def monitor(connection: Connection) -> None:
    devices = SenseCapDeviceCollection(connection)

    def device_added(device: S2101) -> None:
        def show_readings() -> None:
            print(device.descriptor.name, device.temperature, device.humidity)

        show_readings()
        device.add_update_listener(show_readings)
        device.add_remove_listener(
            lambda: print("Removed:", device.descriptor.name)
        )

    devices.subscribe_device_added(device_added)
    try:
        await devices.async_setup()
        await asyncio.Event().wait()
    finally:
        devices.close()

Values can be None until the first report. The model’s removal listener runs on device removal or replacement. Collection shutdown clears it without calling it.

Add a CLI and tests

The CLI helper connects the library to a server and prints device updates. Use fixture events to test attributes, commands, and listeners with captured payloads.