mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-08-08 18:03:18 +02:00
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.
This commit is contained in:
+15
-33
@@ -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)
|
||||
|
||||
+13
-1
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user