Skip to content
lorawan-connection

Quickstart

Install the shared library from PyPI:

Terminal window
pip install lorawan-connection

Save the example library as sensecap_lorawan.py. Save the replay below as replay.py in the same directory, then run it:

Terminal window
python replay.py

The output is:

Device: Greenhouse
temperature=21.4, humidity=31.4

The mock connection supplies a device descriptor followed by a captured uplink. The collection creates the S2101 model, which decodes the reading and notifies its listener. No server is needed.

replay.py
"""Replay an S2101 capture without a network server."""


import asyncio
from datetime import UTC, datetime


from lorawan_connection import (
    AddedEvent,
    DeviceCollection,
    DeviceDescriptor,
    UplinkEvent,
)
from lorawan_connection.mock import MockConnection
from sensecap_lorawan import S2101




async def main() -> None:
    connection = MockConnection()
    devices = DeviceCollection(connection, [S2101])


    def device_added(device: S2101) -> None:
        print(f"Device: {device.descriptor.name}")
        device.add_update_listener(
            lambda: print(
                f"temperature={device.temperature}, humidity={device.humidity}"
            )
        )


    devices.subscribe_device_added(device_added)
    descriptor = DeviceDescriptor(
        stack="chirpstack",
        network_id="home",
        dev_eui="0201010101010101",
        name="Greenhouse",
        application_id="sensors",
        profile_id="s2101-profile",
        model_id=S2101.identifiers["chirpstack"][1],
        brand_id=S2101.identifiers["chirpstack"][0],
        manufacturer="Seeed Studio",
        model="SenseCAP S2101",
    )
    now = datetime.now(UTC)
    try:
        await devices.async_setup()
        connection.emit(AddedEvent(received_at=now, descriptor=descriptor))
        connection.emit(
            UplinkEvent(
                network_id="home",
                dev_eui=descriptor.dev_eui,
                received_at=now,
                data=bytes.fromhex("01011098530000010210A87A0000AF51"),
            )
        )
        device = devices.devices[descriptor.dev_eui]
        assert device.temperature == 21.4
        assert device.humidity == 31.4
    finally:
        devices.close()




if __name__ == "__main__":
    asyncio.run(main())

Example library

Section titled “Example library”

Save this as sensecap_lorawan.py alongside replay.py:

sensecap_lorawan.py
"""SenseCAP models consuming transport-independent LoRaWAN events."""


import logging
from datetime import datetime
from typing import override


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


_LOGGER = logging.getLogger(__name__)


VENDOR_ID = 0x02E8




def decode_s2101(data: bytes) -> dict[str, float | None]:
    """Decode channel-one measurements using Seeed's seven-byte records.


    The two-byte trailer is retained in the wire format. Seeed's reference
    decoder does not validate it. This implementation makes no CRC guarantee.
    """
    if len(data) < 9 or (len(data) - 2) % 7:
        raise ValueError("Invalid SenseCAP frame length")
    result: dict[str, float | None] = {}
    for offset in range(0, len(data) - 2, 7):
        channel = data[offset]
        measurement = int.from_bytes(data[offset + 1 : offset + 3], "little")
        key = {4097: "temperature", 4098: "humidity"}.get(measurement)
        if channel != 1 or key is None:
            continue
        value = (
            int.from_bytes(data[offset + 3 : offset + 7], "little", signed=True) / 1000
        )
        if value >= 2_000_000:
            _LOGGER.warning("SenseCAP %s sensor fault: %s", key, value)
            result[key] = None
            continue
        if key == "humidity" and not 0 <= value <= 100:
            raise ValueError("Humidity outside valid range")
        if key == "temperature" and not -40 <= value <= 85:
            raise ValueError("Temperature outside S2101 measurement range")
        result[key] = value
    return result




class S2101(Device):
    """S2101 measurements and update notifications."""


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


    def __init__(self, descriptor: DeviceDescriptor) -> None:
        """Initialize an unobserved model."""
        super().__init__(descriptor)
        self.temperature: float | None = None
        self.humidity: float | None = None
        self._updated: dict[str, datetime] = {}


    @override
    def handle_event(self, event: DeviceEvent) -> None:
        """Merge a valid partial measurement without clearing other values."""
        if self.closed or event.type != EventType.UPLINK:
            return
        if event.f_port not in (1, 2):
            return
        try:
            values = decode_s2101(event.data)
        except ValueError:
            return
        values = {
            key: value
            for key, value in values.items()
            if key not in self._updated or event.received_at >= self._updated[key]
        }
        if not values:
            return
        self._updated.update(dict.fromkeys(values, event.received_at))
        for key, value in values.items():
            setattr(self, key, value)
        self.notify()




class S2102(Device):
    """SenseCAP light intensity in lux."""


    identifiers = {
        "chirpstack": (VENDOR_ID, "92e11305-8190-41cc-84d0-ddc452cf1889"),
        "tts": ("sensecap", "sensecaps2102-light"),
    }


    def __init__(self, descriptor: DeviceDescriptor) -> None:
        super().__init__(descriptor)
        self.illuminance: float | None = None
        self._updated_at: datetime | None = None


    @override
    def handle_event(self, event: DeviceEvent) -> None:
        if self.closed or event.type != EventType.UPLINK or event.f_port not in (1, 2):
            return
        if self._updated_at is not None and event.received_at < self._updated_at:
            return
        data = event.data
        if len(data) < 9 or (len(data) - 2) % 7:
            return
        observed = False
        value: float | None = None
        for offset in range(0, len(data) - 2, 7):
            if (
                data[offset] != 1
                or int.from_bytes(data[offset + 1 : offset + 3], "little") != 4099
            ):
                continue
            reading = (
                int.from_bytes(data[offset + 3 : offset + 7], "little", signed=True)
                / 1000
            )
            if reading >= 2_000_000:
                value = None
            elif 0 <= reading <= 160_000:
                value = reading
            else:
                return
            observed = True
        if observed:
            self.illuminance = value
            self._updated_at = event.received_at
            self.notify()




type SenseCapDevice = S2101 | S2102




class SenseCapDeviceCollection(DeviceCollection[SenseCapDevice]):
    """Supported SenseCAP catalog models."""


    DEVICES = (S2101, S2102)

This example follows the Seeed reference decoders. It decodes channel-one temperature and humidity records, but does not validate the two-byte trailer. It has been tested with captured payloads and a simulated device. It has not been tested with physical SenseCAP hardware.