Skip to content
lorawan-connection

Devices and collections

A device model holds reported values and exposes command methods for one physical device. This example implements a Dragino LT-22222-L.

Complete device example

Section titled “Complete device example”

Subclass Device and implement handle_event() to decode incoming events. The base class provides identity, LoRaWAN status, update subscriptions, and command delivery.

Save this as dragino_lorawan.py. It includes the DraginoDevices collection used below.

dragino_lorawan.py
"""Dragino device models consuming shared LoRaWAN events."""


import logging
from datetime import datetime
from typing import override


from lorawan_connection import (
    Device,
    DeviceCollection,
    DeviceDescriptor,
    DeviceEvent,
    EventType,
)


from .lht65 import LHT65


_LOGGER = logging.getLogger(__name__)


VENDOR_ID = 676




class LT22222(Device):
    """Model for the Dragino LT-22222-L."""


    model_name = "LT-22222-L"


    identifiers = {
        "chirpstack": (VENDOR_ID, "cb0a7bef-eaa0-4c61-a0b6-ce33e6ecbc4f"),
        "tts": ("dragino", "lt22222-l"),
    }


    def __init__(self, descriptor: DeviceDescriptor) -> None:
        """Initialize values as unobserved until a device report arrives."""
        super().__init__(descriptor)
        self.mode: int | None = None
        self.relays: dict[int, bool | None] = {1: None, 2: None}
        self.digital_outputs: dict[int, bool | None] = {1: None, 2: None}
        self.digital_inputs: dict[int, bool | None] = {1: None, 2: None}
        self.voltages: dict[int, float | None] = {1: None, 2: None}
        self.currents: dict[int, float | None] = {1: None, 2: None}
        self.digital_counts: dict[int, int | None] = {1: None, 2: None}
        self.voltage_count: int | None = None


    @override
    def handle_event(self, event: DeviceEvent) -> None:
        if self.closed or event.type != EventType.UPLINK:
            return
        data = event.data
        if event.f_port != 2 or len(data) != 11:
            return
        # Trigger reports contain flags, not input measurements or output states.
        if not 1 <= (data[10] & 0x3F) <= 5:
            _LOGGER.debug("Ignoring LT-22222-L report mode %s", data[10] & 0x3F)
            return
        mode = data[10] & 0x3F
        relays: dict[int, bool | None] = {
            1: bool(data[8] & 0x80),
            2: bool(data[8] & 0x40),
        }
        digital_outputs: dict[int, bool | None] = {
            1: bool(data[8] & 1),
            2: bool(data[8] & 2),
        }
        digital_inputs: dict[int, bool | None] = {1: None, 2: None}
        voltages: dict[int, float | None] = {1: None, 2: None}
        currents: dict[int, float | None] = {1: None, 2: None}
        digital_counts: dict[int, int | None] = {1: None, 2: None}
        voltage_count = None


        if mode in (1, 5):
            voltages = {
                1: int.from_bytes(data[0:2], signed=True) / 1000,
                2: int.from_bytes(data[2:4], signed=True) / 1000,
            }
        if mode in (1, 3, 5):
            currents[1] = int.from_bytes(data[4:6], signed=True) / 1000
        if mode in (1, 3):
            currents[2] = int.from_bytes(data[6:8], signed=True) / 1000
        if mode == 1:
            digital_inputs = {1: bool(data[8] & 8), 2: bool(data[8] & 16)}
        if mode in (2, 3, 4):
            digital_counts[1] = int.from_bytes(data[0:4])
        if mode == 2:
            digital_counts[2] = int.from_bytes(data[4:8])
        if mode == 4:
            voltage_count = int.from_bytes(data[4:8])
        if mode == 5:
            digital_counts[1] = int.from_bytes(data[6:8])


        self.mode = mode
        self.relays = relays
        self.digital_outputs = digital_outputs
        self.digital_inputs = digital_inputs
        self.voltages = voltages
        self.currents = currents
        self.digital_counts = digital_counts
        self.voltage_count = voltage_count
        self.notify()


    async def async_set_relay(
        self, channel: int, on: bool, *, expires_at: datetime | None = None
    ) -> None:
        """Await command acknowledgement. Telemetry updates reported state."""
        if channel not in (1, 2):
            raise ValueError("Relay channel must be 1 or 2")
        states = [0x11, 0x11]
        states[channel - 1] = int(on)
        await self.async_send_downlink(
            data=bytes((0x03, *states)),
            f_port=2,
            expires_at=expires_at,
        )


    async def async_set_digital_output(
        self, channel: int, on: bool, *, expires_at: datetime | None = None
    ) -> None:
        """Enable an active-low output and await its command acknowledgement."""
        if channel not in (1, 2):
            raise ValueError("Digital output channel must be 1 or 2")
        # The protocol includes DO3 even though this model has no third output.
        states = [0x11, 0x11, 0x11]
        states[channel - 1] = int(on)
        await self.async_send_downlink(
            data=bytes((0x02, *states)),
            f_port=2,
            expires_at=expires_at,
        )




type DraginoDevice = LT22222 | LHT65




class DraginoDevices(DeviceCollection[DraginoDevice]):
    """Supported Dragino catalog models."""


    DEVICES = (LT22222, LHT65)

Values start as None until a report arrives. The model uses channel dictionaries. Other models can use individual attributes, nested objects, or dataclasses.

Decode events and update state

Section titled “Decode events and update state”

Implement handle_event() to check the event type and supported port, then decode the payload. Keep byte layouts, units, checksums, and vendor error values in the model or its decoder. Python’s int.from_bytes and struct can read binary fields.

Validate the whole payload before changing attributes. Invalid data leaves the previous values intact. A partial reading updates only the values it contains, unless the device protocol says an omitted value is no longer available.

One message can update several values. Update them all before calling notify() so listeners see the complete update. Notify for valid reports even when values are unchanged. Models choose their own timestamp ordering and stale-data rules. The collection does not impose them.

Read LoRaWAN device status

Section titled “Read LoRaWAN device status”

The base device stores StatusEvent before calling your handle_event() method. It exposes battery percentage, external power, and the device’s downlink SNR. Your handler does not need to call super() for status or acknowledgement handling.

Read device.battery_level, device.external_power_source, and device.downlink_margin. These start as None. device.latest_status contains the complete report and its timestamp. Status reports notify update listeners after your handler finishes.

These values come from LoRaWAN device-status reports. Keep battery readings decoded from a vendor payload in the model’s own attributes.

Read values and observe changes

Section titled “Read values and observe changes”

Read the latest reported values from model attributes without contacting the device. Subscribe to changes in the collection’s device-added callback:

from dragino_lorawan import LT22222




def device_added(device: LT22222) -> None:
    def show_values() -> None:
        print(device.descriptor.name)
        print(device.voltages, device.digital_inputs, device.relays)


    show_values()
    device.add_update_listener(show_values)

Listeners run synchronously with no arguments. Registration does not replay state, so call show_values() once for the initial values. add_update_listener() returns an unsubscribe function. Closing the model also removes its listeners.

Send commands

Section titled “Send commands”

Applications call the model’s async methods. For example:

import asyncio


from dragino_lorawan import LT22222




async def turn_on(device: LT22222) -> None:
    async with asyncio.timeout(30):
        await device.async_set_relay(1, True)

The call completes after a device acknowledgement or raises TimeoutError after 30 seconds. Keep the event subscription running so acknowledgements can arrive. An ACK confirms delivery. Reported state changes with the next uplink.

Implement a command method

Section titled “Implement a command method”

Validate the caller’s arguments and encode them using the device’s protocol. Then await self.async_send_downlink(data=payload, f_port=port) from the model’s async method. The base class supplies the device’s DevEUI and waits for its ACK. Keep reported attributes unchanged until a device report arrives.

By default, wait_for_ack=True requests a confirmed downlink. A model can pass wait_for_ack=False to send an unconfirmed command and return after the server accepts it. Confirmation follows this setting automatically.

If a command should expire while queued, pass a timezone-aware expires_at datetime. Queue expiry limits how long the server retains a command. The caller’s timeout limits how long the application waits.

The Dragino example exposes optional expiry on both output methods:

await device.async_set_relay(1, True)  # No queue expiry.
await device.async_set_digital_output(1, True, expires_at=deadline)

deadline must be a timezone-aware datetime. ChirpStack supports queue expiry. TTS rejects an explicit deadline. Without expiry, a queued command can execute later, including after the caller stops waiting. No automatic queue clearing occurs.

Handle failures

Section titled “Handle failures”

A rejected command or negative acknowledgement raises DownlinkError. Sending also raises DownlinkError when the model is closed or has no sender. Applications can catch this and TimeoutError where they report the result to the user.

Cancellation stops waiting. It does not remove a command already queued on the server, which may still reach the device. Close the collection on connection loss to end pending acknowledgement waits.

Manage devices with a collection

Section titled “Manage devices with a collection”

A collection manages device instances for one logical network. DraginoDevices declares its supported models in DEVICES. The collection selects a class by matching its identifiers against each incoming descriptor.

Each stack has its own catalog. Declare the native brand and model IDs together:

class S2101(Device):
    identifiers = {
        "chirpstack": (744, "fc455aa2-01cf-492b-9359-a5d8c9a0e1b3"),
        "tts": ("sensecap", "sensecaps2101-temp-humid"),
    }


    # Implement handle_event() and expose the decoded readings.

The collection matches (stack, brand_id, model_id). The same device model can therefore decode messages from either stack. A duplicate identity within one stack raises ValueError. Matching IDs in different stacks are independent.

This function creates a collection and prints each model update:

from dragino_lorawan import DraginoDevices, LT22222
from lorawan_connection import Connection, DeviceCollection




def device_added(device: LT22222) -> None:
    def show_values() -> None:
        print(device.descriptor.name, device.voltages, device.relays)


    device.add_update_listener(show_values)
    show_values()




def create_collection(connection: Connection) -> DeviceCollection[LT22222]:
    collection = DraginoDevices(connection)
    collection.subscribe_device_added(device_added)
    return collection

In the ChirpStack connection example, replace the SenseCAP collection and its device-added callback with these definitions. Create it with create_collection(connection), then call await collection.async_setup(). The collection subscribes to its registered vendors and attaches the command sender before announcing each device.

You can also construct a collection directly from model classes. Inside create_collection, the equivalent construction is:

from lorawan_connection import DeviceCollection


collection = DeviceCollection(connection, [LT22222])

Backends and test doubles implement the same Connection protocol. Several collections can share a connection. Closing one leaves the others and the transport open.

Include unrecognized devices

Section titled “Include unrecognized devices”

For a device listing, omit model classes. The collection creates a generic Device for every device exposed by the connection, even without a catalog identity:

from lorawan_connection import Connection, Device, DeviceCollection




async def create_inventory(connection: Connection) -> DeviceCollection[Device]:
    collection = DeviceCollection(connection)


    def device_added(device: Device) -> None:
        def show_status() -> None:
            print(device.descriptor.name, device.battery_level)


        device.add_update_listener(show_status)
        show_status()


    collection.subscribe_device_added(device_added)
    await collection.async_setup()
    return collection

Generic devices retain reported LoRaWAN status without decoding application bytes. A battery value remains unknown until a usable status report arrives.

Collections with registered models include only matching devices. A generic collection and a vendor collection can share the same connection. Close each collection when its consumer no longer needs live updates.

Device lifecycle

Section titled “Device lifecycle”

subscribe_device_added() reports existing models immediately and future additions as they arrive. The collection announces a model before passing its first event to it, so the callback can attach listeners before readings arrive. Current instances are available in collection.devices, keyed by DevEUI.

A metadata update replaces the descriptor on the existing instance and notifies its update listeners. A change to its vendor or catalog identity replaces the model. Removal closes the model and calls listeners registered with subscribe_device_removed().

To observe one model directly, register a callback on it:

unsubscribe = device.add_remove_listener(
    lambda: print(f"Removed: {device.descriptor.name}")
)

The callback runs after removal or replacement closes that model. Ordinary connection shutdown clears this listener without reporting removal.

When stopping, call collection.close(). It unsubscribes from device events, closes every model, removes listeners, and ends pending command waits. Close operations are safe to repeat.