Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
85cb234442
|
||
|
|
155c641012
|
No files matched your search
@@ -1,5 +1,27 @@
|
||||
# Changelog
|
||||
|
||||
## 1.1.0 — 2026-04-23
|
||||
|
||||
### Added
|
||||
|
||||
- `connect()` now validates and auto-adjusts the requested
|
||||
`LoRaConfig` against the device's advertised caps (`GET_INFO`):
|
||||
- `tx_power_dbm` is silently clamped into
|
||||
`[tx_power_min_dbm, tx_power_max_dbm]`. A clamp is logged at
|
||||
INFO. This keeps `dl.connect(config=my_config)` "just work" on
|
||||
boards with lower PA ceilings (e.g. SX1276 @ 20 dBm vs SX1262
|
||||
@ 22 dBm) without any per-board bookkeeping by the caller.
|
||||
- `freq_hz` outside the device's range, `sf` not in
|
||||
`supported_sf_bitmap`, or `bw` not in `supported_bw_bitmap` now
|
||||
raise :class:`ConfigNotSupported` *before* `SET_CONFIG` hits
|
||||
the wire. Silent shifts here would cross regulatory boundaries
|
||||
or change airtime/sensitivity without the caller noticing.
|
||||
- `Dongle.config` is now populated from `SetConfigResult.current`
|
||||
— the modulation the device actually stored, post-clamp — so
|
||||
callers can inspect what landed.
|
||||
- New `ConfigNotSupported` exception (subclass of `DongloraError`),
|
||||
exported at the package root.
|
||||
|
||||
## 1.0.1 — 2026-04-22
|
||||
|
||||
### Fixed
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
# DongLoRa Python Client
|
||||
|
||||
Python client library for talking to a DongLoRa device — either directly
|
||||
over USB or through the [mux daemon](https://github.com/donglora/mux-py).
|
||||
over USB or through the [mux daemon](https://git.swaits.com/donglora/mux-py).
|
||||
|
||||
## Install
|
||||
|
||||
|
||||
@@ -66,6 +66,7 @@ from donglora.errors import (
|
||||
BusyError,
|
||||
Cancelled,
|
||||
ChannelBusy,
|
||||
ConfigNotSupported,
|
||||
DeviceError,
|
||||
DongloraError,
|
||||
ErrorCode,
|
||||
@@ -131,6 +132,7 @@ __all__ = [
|
||||
"Cancelled",
|
||||
"Capability",
|
||||
"ChannelBusy",
|
||||
"ConfigNotSupported",
|
||||
"DeviceError",
|
||||
"Dongle",
|
||||
"DongloraError",
|
||||
|
||||
+90
-9
@@ -6,15 +6,21 @@ form it:
|
||||
1. Finds a DongLoRa USB device (or mux socket if one is available).
|
||||
2. Opens the transport at 115200 baud.
|
||||
3. Sends ``GET_INFO`` to confirm we're talking DongLoRa Protocol v2.
|
||||
4. Applies a sensible default :class:`LoRaConfig` via ``SET_CONFIG``.
|
||||
4. Applies a sensible default :class:`LoRaConfig` via ``SET_CONFIG``,
|
||||
auto-clamping ``tx_power_dbm`` against the device's advertised cap
|
||||
and raising :class:`ConfigNotSupported` for out-of-range frequency,
|
||||
spreading factor, or bandwidth.
|
||||
5. Spawns a keepalive thread to keep the session alive.
|
||||
6. Returns a :class:`Dongle` ready for ``.tx()`` / ``.rx()``.
|
||||
6. Returns a :class:`Dongle` ready for ``.tx()`` / ``.rx()``. The
|
||||
actually-applied config (post-clamp) is exposed on
|
||||
:attr:`Dongle.config`.
|
||||
|
||||
Every step is overrideable via keyword arguments. See ``connect``'s docstring.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import logging
|
||||
import os
|
||||
import socket
|
||||
@@ -25,7 +31,7 @@ import serial
|
||||
|
||||
from donglora.discovery import find_port, wait_for_device
|
||||
from donglora.dongle import Dongle
|
||||
from donglora.errors import DongloraError
|
||||
from donglora.errors import ConfigNotSupported, DongloraError
|
||||
from donglora.info import Info
|
||||
from donglora.modulation import LoRaConfig, Modulation
|
||||
from donglora.session import Session
|
||||
@@ -131,6 +137,67 @@ def _open_transport(port: str | None, timeout: float) -> Any:
|
||||
return ser
|
||||
|
||||
|
||||
def _prepare_config(info: Info, config: Modulation) -> Modulation:
|
||||
"""Validate and auto-adjust *config* against the device's advertised caps.
|
||||
|
||||
Per-field policy:
|
||||
|
||||
* ``tx_power_dbm``: clamped into ``[tx_power_min_dbm, tx_power_max_dbm]``.
|
||||
A clamp is logged at INFO — "give me max power" quietly returning less
|
||||
is the universally-expected behavior and not worth a hard error.
|
||||
* ``freq_hz``: rejected with :class:`ConfigNotSupported` when outside
|
||||
``[freq_min_hz, freq_max_hz]``. Silently shifting a 915 MHz request
|
||||
to 868 MHz (or vice versa) would cross regulatory boundaries.
|
||||
* ``sf``, ``bw``: rejected with :class:`ConfigNotSupported` when the
|
||||
corresponding capability bit isn't set. These change airtime and
|
||||
sensitivity dramatically; silent substitution is more confusing
|
||||
than helpful.
|
||||
|
||||
Non-LoRa modulations pass through untouched — the firmware rejects
|
||||
unsupported modulation IDs with ``EMODULATION`` on its own.
|
||||
"""
|
||||
if not isinstance(config, LoRaConfig):
|
||||
return config
|
||||
|
||||
cfg: LoRaConfig = config
|
||||
|
||||
if not (info.freq_min_hz <= cfg.freq_hz <= info.freq_max_hz):
|
||||
raise ConfigNotSupported(
|
||||
f"frequency {cfg.freq_hz} Hz outside device range "
|
||||
f"[{info.freq_min_hz}, {info.freq_max_hz}] Hz"
|
||||
)
|
||||
|
||||
if not (info.supported_sf_bitmap & (1 << cfg.sf)):
|
||||
supported = [i for i in range(16) if info.supported_sf_bitmap & (1 << i)]
|
||||
raise ConfigNotSupported(
|
||||
f"SF{cfg.sf} not supported by this device (supports SF{supported})"
|
||||
)
|
||||
|
||||
bw_bit = int(cfg.bw)
|
||||
if not (info.supported_bw_bitmap & (1 << bw_bit)):
|
||||
raise ConfigNotSupported(
|
||||
f"bandwidth {cfg.bw.name} (bit {bw_bit}) not in "
|
||||
f"supported_bw_bitmap 0x{info.supported_bw_bitmap:04X}"
|
||||
)
|
||||
|
||||
if cfg.tx_power_dbm > info.tx_power_max_dbm:
|
||||
log.info(
|
||||
"clamping tx_power_dbm: requested %d dBm, device max %d dBm",
|
||||
cfg.tx_power_dbm,
|
||||
info.tx_power_max_dbm,
|
||||
)
|
||||
cfg = dataclasses.replace(cfg, tx_power_dbm=info.tx_power_max_dbm)
|
||||
elif cfg.tx_power_dbm < info.tx_power_min_dbm:
|
||||
log.info(
|
||||
"clamping tx_power_dbm: requested %d dBm, device min %d dBm",
|
||||
cfg.tx_power_dbm,
|
||||
info.tx_power_min_dbm,
|
||||
)
|
||||
cfg = dataclasses.replace(cfg, tx_power_dbm=info.tx_power_min_dbm)
|
||||
|
||||
return cfg
|
||||
|
||||
|
||||
def connect(
|
||||
port: str | None = None,
|
||||
*,
|
||||
@@ -158,9 +225,16 @@ def connect(
|
||||
Modulation to apply at connect-time. Default:
|
||||
:meth:`LoRaConfig.default`. Pass ``None`` with
|
||||
``auto_configure=False`` to skip configuration entirely.
|
||||
With ``auto_configure=True`` (default), ``tx_power_dbm`` is
|
||||
silently clamped to the device's ``tx_power_max_dbm`` /
|
||||
``tx_power_min_dbm``; out-of-range ``freq_hz``, ``sf``, or
|
||||
``bw`` raise :class:`ConfigNotSupported`. Inspect
|
||||
:attr:`Dongle.config` after connect to see what the device
|
||||
actually stored.
|
||||
auto_configure
|
||||
If False, skip the ``SET_CONFIG`` step. The caller is then
|
||||
responsible for calling :meth:`Dongle.set_config` before TX/RX.
|
||||
If False, skip the ``SET_CONFIG`` step entirely — no
|
||||
validation, no clamping. The caller is then responsible for
|
||||
calling :meth:`Dongle.set_config` before TX/RX.
|
||||
keepalive
|
||||
If False, skip the keepalive daemon thread. The caller is
|
||||
responsible for periodic :meth:`Dongle.ping` calls to stay
|
||||
@@ -195,12 +269,18 @@ def connect(
|
||||
|
||||
applied: Modulation | None = None
|
||||
if auto_configure:
|
||||
applied = config if config is not None else LoRaConfig.default()
|
||||
requested: Modulation = config if config is not None else LoRaConfig.default()
|
||||
try:
|
||||
session.set_config(applied, timeout=max(timeout, 2.0))
|
||||
prepared = _prepare_config(info, requested)
|
||||
result = session.set_config(prepared, timeout=max(timeout, 2.0))
|
||||
except DongloraError:
|
||||
session.close()
|
||||
raise
|
||||
# Firmware echoes back the modulation it actually stored. Trust
|
||||
# that rather than our own `prepared` — it's the same value
|
||||
# today, but stays correct if firmware ever adds its own
|
||||
# normalization.
|
||||
applied = result.current if result is not None else prepared
|
||||
return session, info, applied
|
||||
|
||||
session, info, applied = _open_and_init()
|
||||
@@ -315,10 +395,11 @@ def _session_on(transport: Any) -> Dongle:
|
||||
raise DongloraError(
|
||||
f"device speaks DongLoRa Protocol v{info.proto_major}.{info.proto_minor}; this client requires v1.x",
|
||||
)
|
||||
applied = LoRaConfig.default()
|
||||
try:
|
||||
session.set_config(applied)
|
||||
prepared = _prepare_config(info, LoRaConfig.default())
|
||||
result = session.set_config(prepared)
|
||||
except DongloraError:
|
||||
session.close()
|
||||
raise
|
||||
applied = result.current if result is not None else prepared
|
||||
return Dongle(session, info, applied_config=applied, keepalive=True)
|
||||
+5
-1
@@ -99,7 +99,11 @@ class Dongle:
|
||||
|
||||
@property
|
||||
def config(self) -> Modulation | None:
|
||||
"""The most-recently-applied radio config, or ``None`` if none."""
|
||||
"""The radio config the device actually stored on the last
|
||||
``SET_CONFIG`` (including any post-clamp tx_power adjustment
|
||||
that :func:`donglora.connect` performed). ``None`` if no
|
||||
``SET_CONFIG`` has landed yet.
|
||||
"""
|
||||
return self._applied_config
|
||||
|
||||
def tx(self, data: bytes, *, skip_cad: bool = False, timeout: float = 10.0) -> TxDone:
|
||||
|
||||
@@ -42,6 +42,25 @@ class DongloraError(Exception):
|
||||
"""Base class for all DongLoRa client errors."""
|
||||
|
||||
|
||||
class ConfigNotSupported(DongloraError):
|
||||
"""The requested config contains a field the device cannot satisfy.
|
||||
|
||||
Raised client-side, before ``SET_CONFIG`` hits the wire, for fields
|
||||
that are *not* safe to silently auto-adjust:
|
||||
|
||||
* ``freq_hz`` outside the device's reported
|
||||
``freq_min_hz..freq_max_hz`` — silently shifting 915 MHz to
|
||||
868 MHz (or vice versa) crosses regulatory boundaries.
|
||||
* ``sf`` not in ``supported_sf_bitmap`` — changes airtime and
|
||||
sensitivity in ways the caller needs to know about.
|
||||
* ``bw`` not in ``supported_bw_bitmap`` — same as SF.
|
||||
|
||||
``tx_power_dbm`` is *not* in this list: it's clamped transparently
|
||||
because "give me max power" returning less than requested is the
|
||||
universally expected behavior.
|
||||
"""
|
||||
|
||||
|
||||
class FrameError(DongloraError):
|
||||
"""COBS decode, CRC check, or length check failed on an inbound frame."""
|
||||
|
||||
|
||||
+3
-3
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "donglora"
|
||||
version = "1.0.1"
|
||||
version = "1.1.0"
|
||||
description = "DongLoRa host library — talk LoRa from Python over USB"
|
||||
license = "MIT"
|
||||
requires-python = ">=3.10"
|
||||
@@ -28,8 +28,8 @@ meshcore = ["pycryptodome", "pynacl"]
|
||||
orac = ["pycryptodome", "pynacl", "anthropic"]
|
||||
|
||||
[project.urls]
|
||||
Repository = "https://github.com/donglora/client-py"
|
||||
Homepage = "https://github.com/donglora/client-py"
|
||||
Repository = "https://git.swaits.com/donglora/client-py"
|
||||
Homepage = "https://git.swaits.com/donglora/client-py"
|
||||
|
||||
[dependency-groups]
|
||||
dev = ["ruff", "pytest", "pytest-cov"]
|
||||
|
||||
@@ -0,0 +1,196 @@
|
||||
"""Unit tests for the client-side config preparation done by ``connect()``.
|
||||
|
||||
Validates the auto_configure contract:
|
||||
|
||||
* ``tx_power_dbm`` clamps silently (both high and low sides).
|
||||
* ``freq_hz`` / ``sf`` / ``bw`` raise :class:`ConfigNotSupported` when
|
||||
outside the device's advertised caps — silent substitution would hide
|
||||
regulatory or airtime-changing surprises from the caller.
|
||||
* Non-LoRa modulations pass through untouched.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
import donglora as dl
|
||||
from donglora.connect import _prepare_config
|
||||
from donglora.info import Info
|
||||
|
||||
|
||||
def _info(
|
||||
*,
|
||||
freq_min_hz: int = 863_000_000,
|
||||
freq_max_hz: int = 928_000_000,
|
||||
supported_sf_bitmap: int = 0x1FE0, # SF5..SF12
|
||||
supported_bw_bitmap: int = 0x03FF, # sub-GHz BWs 0..9
|
||||
tx_power_min_dbm: int = -9,
|
||||
tx_power_max_dbm: int = 22,
|
||||
) -> Info:
|
||||
"""Build an Info with only the caps-related fields that matter here."""
|
||||
return Info(
|
||||
proto_major=1,
|
||||
proto_minor=0,
|
||||
fw_major=0,
|
||||
fw_minor=0,
|
||||
fw_patch=0,
|
||||
radio_chip_id=int(dl.RadioChipId.SX1262),
|
||||
capability_bitmap=dl.Capability.LORA,
|
||||
supported_sf_bitmap=supported_sf_bitmap,
|
||||
supported_bw_bitmap=supported_bw_bitmap,
|
||||
max_payload_bytes=255,
|
||||
rx_queue_capacity=32,
|
||||
tx_queue_capacity=1,
|
||||
freq_min_hz=freq_min_hz,
|
||||
freq_max_hz=freq_max_hz,
|
||||
tx_power_min_dbm=tx_power_min_dbm,
|
||||
tx_power_max_dbm=tx_power_max_dbm,
|
||||
mcu_uid=b"\x00" * 6,
|
||||
radio_uid=b"",
|
||||
)
|
||||
|
||||
|
||||
# ── tx_power: clamp silently ──────────────────────────────────────
|
||||
|
||||
|
||||
def test_tx_power_above_max_is_clamped_down():
|
||||
cfg = dl.LoRaConfig.default()
|
||||
cfg = dl.LoRaConfig(
|
||||
freq_hz=cfg.freq_hz,
|
||||
bw=cfg.bw,
|
||||
sf=cfg.sf,
|
||||
cr=cfg.cr,
|
||||
sync_word=cfg.sync_word,
|
||||
tx_power_dbm=30,
|
||||
preamble_len=cfg.preamble_len,
|
||||
header_mode=cfg.header_mode,
|
||||
payload_crc=cfg.payload_crc,
|
||||
iq_invert=cfg.iq_invert,
|
||||
)
|
||||
out = _prepare_config(_info(tx_power_max_dbm=20), cfg)
|
||||
assert isinstance(out, dl.LoRaConfig)
|
||||
assert out.tx_power_dbm == 20
|
||||
# Other fields unchanged.
|
||||
assert out.freq_hz == cfg.freq_hz
|
||||
assert out.sf == cfg.sf
|
||||
|
||||
|
||||
def test_tx_power_below_min_is_clamped_up():
|
||||
cfg = dl.LoRaConfig.default()
|
||||
cfg = dl.LoRaConfig(
|
||||
freq_hz=cfg.freq_hz,
|
||||
bw=cfg.bw,
|
||||
sf=cfg.sf,
|
||||
cr=cfg.cr,
|
||||
sync_word=cfg.sync_word,
|
||||
tx_power_dbm=-30,
|
||||
preamble_len=cfg.preamble_len,
|
||||
header_mode=cfg.header_mode,
|
||||
payload_crc=cfg.payload_crc,
|
||||
iq_invert=cfg.iq_invert,
|
||||
)
|
||||
out = _prepare_config(_info(tx_power_min_dbm=2), cfg)
|
||||
assert isinstance(out, dl.LoRaConfig)
|
||||
assert out.tx_power_dbm == 2
|
||||
|
||||
|
||||
def test_tx_power_inside_range_is_untouched():
|
||||
cfg = dl.LoRaConfig.default()
|
||||
out = _prepare_config(_info(), cfg)
|
||||
assert out is cfg # same object, no replace() called
|
||||
|
||||
|
||||
# ── freq_hz: reject (regulatory boundaries) ───────────────────────
|
||||
|
||||
|
||||
def test_freq_below_min_raises():
|
||||
cfg = dl.LoRaConfig(
|
||||
freq_hz=300_000_000,
|
||||
bw=dl.LoRaBandwidth.KHZ_125,
|
||||
sf=7,
|
||||
cr=dl.LoRaCodingRate.CR_4_5,
|
||||
sync_word=0x3444,
|
||||
tx_power_dbm=14,
|
||||
preamble_len=8,
|
||||
header_mode=dl.LoRaHeaderMode.EXPLICIT,
|
||||
payload_crc=True,
|
||||
iq_invert=False,
|
||||
)
|
||||
with pytest.raises(dl.ConfigNotSupported, match="frequency"):
|
||||
_prepare_config(_info(freq_min_hz=863_000_000, freq_max_hz=928_000_000), cfg)
|
||||
|
||||
|
||||
def test_freq_above_max_raises():
|
||||
cfg = dl.LoRaConfig(
|
||||
freq_hz=2_400_000_000,
|
||||
bw=dl.LoRaBandwidth.KHZ_125,
|
||||
sf=7,
|
||||
cr=dl.LoRaCodingRate.CR_4_5,
|
||||
sync_word=0x3444,
|
||||
tx_power_dbm=14,
|
||||
preamble_len=8,
|
||||
header_mode=dl.LoRaHeaderMode.EXPLICIT,
|
||||
payload_crc=True,
|
||||
iq_invert=False,
|
||||
)
|
||||
with pytest.raises(dl.ConfigNotSupported, match="frequency"):
|
||||
_prepare_config(_info(freq_min_hz=137_000_000, freq_max_hz=1_020_000_000), cfg)
|
||||
|
||||
|
||||
# ── sf: reject (airtime / sensitivity change) ─────────────────────
|
||||
|
||||
|
||||
def test_sf5_rejected_on_sx1276_bitmap():
|
||||
"""SX1276 advertises SF6-SF12 (0x1FC0). SF5 must not silently upgrade."""
|
||||
cfg = dl.LoRaConfig(
|
||||
freq_hz=915_000_000,
|
||||
bw=dl.LoRaBandwidth.KHZ_125,
|
||||
sf=5,
|
||||
cr=dl.LoRaCodingRate.CR_4_5,
|
||||
sync_word=0x3444,
|
||||
tx_power_dbm=14,
|
||||
preamble_len=8,
|
||||
header_mode=dl.LoRaHeaderMode.EXPLICIT,
|
||||
payload_crc=True,
|
||||
iq_invert=False,
|
||||
)
|
||||
with pytest.raises(dl.ConfigNotSupported, match="SF5"):
|
||||
_prepare_config(_info(supported_sf_bitmap=0x1FC0), cfg)
|
||||
|
||||
|
||||
# ── bw: reject (airtime change) ────────────────────────────────────
|
||||
|
||||
|
||||
def test_bw_not_in_bitmap_raises():
|
||||
"""SX128x-only 200 kHz BW on a sub-GHz-only device must be rejected."""
|
||||
cfg = dl.LoRaConfig(
|
||||
freq_hz=915_000_000,
|
||||
bw=dl.LoRaBandwidth.KHZ_200, # bit 10 — SX128x only
|
||||
sf=7,
|
||||
cr=dl.LoRaCodingRate.CR_4_5,
|
||||
sync_word=0x3444,
|
||||
tx_power_dbm=14,
|
||||
preamble_len=8,
|
||||
header_mode=dl.LoRaHeaderMode.EXPLICIT,
|
||||
payload_crc=True,
|
||||
iq_invert=False,
|
||||
)
|
||||
with pytest.raises(dl.ConfigNotSupported, match="bandwidth"):
|
||||
_prepare_config(_info(supported_bw_bitmap=0x03FF), cfg)
|
||||
|
||||
|
||||
# ── Non-LoRa: pass through ─────────────────────────────────────────
|
||||
|
||||
|
||||
def test_non_lora_modulation_passes_through():
|
||||
# Frequency deliberately outside the info's range; _prepare_config
|
||||
# must not inspect it for non-LoRa modulations (firmware rejects
|
||||
# unsupported modulation IDs with EMODULATION on its own).
|
||||
fsk = dl.FskConfig(
|
||||
freq_hz=50_000_000,
|
||||
bitrate_bps=50_000,
|
||||
freq_dev_hz=25_000,
|
||||
rx_bw=0,
|
||||
)
|
||||
out = _prepare_config(_info(), fsk)
|
||||
assert out is fsk
|
||||
Reference in new issue
Block a user