From b9579b6c15c4cc573fd6385c7831184b7fa3a4b0 Mon Sep 17 00:00:00 2001 From: agessaman Date: Wed, 15 Jul 2026 20:35:07 -0700 Subject: [PATCH] fix(airtime): delegate airtime math to the shared core estimator AirtimeManager.calculate_airtime now uses openhop_core's calculate_lora_airtime_ms, which matches RadioLib's getTimeOnAir. This adopts the radio driver's symbol-time auto-LDRO rule in place of the SF/BW shorthand, which diverged from hardware at 16.384 ms symbol settings such as SF12 @ 250 kHz and SF10 @ 62.5 kHz. Test oracles follow the same rule, with literal RadioLib-derived regression values for the divergent settings. --- repeater/airtime.py | 48 +++++++++++------------------------ tests/test_airtime.py | 14 +++++++++- tests/test_packet_duration.py | 3 ++- 3 files changed, 30 insertions(+), 35 deletions(-) diff --git a/repeater/airtime.py b/repeater/airtime.py index b9e8141..c21ddc2 100644 --- a/repeater/airtime.py +++ b/repeater/airtime.py @@ -1,9 +1,8 @@ import logging -import math import time from typing import Tuple -from openhop_core.protocol.packet_utils import coding_rate_denominator +from openhop_core.protocol.packet_utils import calculate_lora_airtime_ms logger = logging.getLogger("AirtimeManager") @@ -39,9 +38,12 @@ class AirtimeManager: explicit_header: bool = True, ) -> float: """ - Calculate LoRa packet airtime using the Semtech reference formula. + Calculate LoRa packet airtime via the shared core estimator. - Reference: https://www.semtech.com/design-support/lora-calculator + Delegates to ``calculate_lora_airtime_ms``, which matches RadioLib's + ``getTimeOnAir`` (the firmware reference), including its symbol-time + low-data-rate-optimization auto rule. Coding rate accepts either the + denominator form (5..8) or the legacy index form (1..4). Args: payload_len: Payload length in bytes @@ -55,35 +57,15 @@ class AirtimeManager: Returns: Airtime in milliseconds """ - sf = spreading_factor or self.spreading_factor - bw_hz = bandwidth_hz or self.bandwidth - # Configs and radio drivers use either the denominator form (5..8) or - # the legacy index form (1..4); normalize before the Semtech formula. - cr = coding_rate_denominator(coding_rate or self.coding_rate) - preamble_len = preamble_len or self.preamble_length - crc = 1 if crc_enabled else 0 - h = 0 if explicit_header else 1 # H=0 for explicit, H=1 for implicit - - # Low data rate optimization: required for SF11/SF12 at 125kHz - de = 1 if (sf >= 11 and bw_hz <= 125000) else 0 - - # Symbol time in milliseconds: T_sym = 2^SF / BW_kHz - t_sym = (2**sf) / (bw_hz / 1000) - - # Preamble time: T_preamble = (n_preamble + 4.25) * T_sym - t_preamble = (preamble_len + 4.25) * t_sym - - # Payload symbol calculation (Semtech formula): - # n_payload = 8 + ceil(max(8*PL - 4*SF + 28 + 16*CRC - 20*H, 0) / (4*(SF - 2*DE))) * CR - numerator = max(8 * payload_len - 4 * sf + 28 + 16 * crc - 20 * h, 0) - denominator = 4 * (sf - 2 * de) - n_payload = 8 + math.ceil(numerator / denominator) * cr - - # Payload time - t_payload = n_payload * t_sym - - # Total packet airtime - return t_preamble + t_payload + return calculate_lora_airtime_ms( + payload_len, + spreading_factor or self.spreading_factor, + bandwidth_hz or self.bandwidth, + coding_rate or self.coding_rate, + preamble_len or self.preamble_length, + crc_enabled=crc_enabled, + explicit_header=explicit_header, + ) def can_transmit(self, airtime_ms: float) -> Tuple[bool, float]: enforcement_enabled = self.config.get("duty_cycle", {}).get("enforcement_enabled", True) diff --git a/tests/test_airtime.py b/tests/test_airtime.py index 8f005bf..d6feb8e 100644 --- a/tests/test_airtime.py +++ b/tests/test_airtime.py @@ -17,7 +17,8 @@ def _semtech_airtime_ms(payload_len: int, sf: int, bw_hz: int, cr: int, preamble """Independent Semtech reference formula used as oracle in tests.""" crc = 1 h = 0 # explicit header - de = 1 if (sf >= 11 and bw_hz <= 125000) else 0 + # LDRO follows the radio driver's auto rule: on when symbol time >= 16 ms + de = 1 if (2**sf) / (bw_hz / 1000) >= 16.0 else 0 t_sym = (2**sf) / (bw_hz / 1000) t_preamble = (preamble + 4.25) * t_sym numerator = max(8 * payload_len - 4 * sf + 28 + 16 * crc - 20 * h, 0) @@ -97,6 +98,17 @@ def test_long_range_preset_has_higher_airtime_than_fast_preset_for_same_payload( assert long_mgr.calculate_airtime(payload_len) > fast_mgr.calculate_airtime(payload_len) +def test_ldro_follows_symbol_time_rule_like_radiolib(): + """Settings with 16.384 ms symbols use LDRO on real hardware (RadioLib's + auto rule) even though they fall outside the old "SF >= 11 and + BW <= 125 kHz" shorthand. Expected values generated with RadioLib 7.6.0's + SX126x::calculateTimeOnAir integer arithmetic (microseconds / 1000).""" + sf12_wide = _make_mgr(sf=12, bw_hz=250000, cr=5, preamble=8) + assert math.isclose(sf12_wide.calculate_airtime(50), 1150.976, rel_tol=1e-9) + sf10_narrow = _make_mgr(sf=10, bw_hz=62500, cr=6, preamble=8) + assert math.isclose(sf10_narrow.calculate_airtime(32), 1216.512, rel_tol=1e-9) + + def test_legacy_coding_rate_index_matches_denominator_form(): """A config using the legacy index form (1..4) must compute the same airtime as the equivalent denominator form (5..8), not undercount diff --git a/tests/test_packet_duration.py b/tests/test_packet_duration.py index 3c646c8..31f58f6 100644 --- a/tests/test_packet_duration.py +++ b/tests/test_packet_duration.py @@ -18,7 +18,8 @@ def _semtech_airtime_ms(payload_len: int, sf: int, bw_hz: int, cr: int, preamble """ crc = 1 h = 0 # explicit header - de = 1 if (sf >= 11 and bw_hz <= 125000) else 0 + # LDRO follows the radio driver's auto rule: on when symbol time >= 16 ms + de = 1 if (2**sf) / (bw_hz / 1000) >= 16.0 else 0 t_sym = (2**sf) / (bw_hz / 1000) t_preamble = (preamble + 4.25) * t_sym numerator = max(8 * payload_len - 4 * sf + 28 + 16 * crc - 20 * h, 0)