[sx126x][sx127x] Fixes from code review

- Hold a send back for at most the airtime of the largest packet, worked
  out from the radio settings and at least 2 s, instead of a fixed 2 s
  that a LoRa packet at a high spreading factor outlasts
- End a busy period when a send finds the radio free, so a busy flag that
  never clears is released however rarely sends are tried
- Detect an FSK reception on the SX127x from its preamble and sync word
  flags
- Test the radio links' final validation

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

esphome/components/packet_interface/listen_before_talk.h

@@ -7,31 +7,37 @@ namespace esphome::packet_interface {
 /**
  * Decides whether a half-duplex radio link may send while its radio reports that it is receiving a packet.
  *
- * Sending is refused while the radio is busy, so the packet being received is not lost, but for at most MAX_BUSY_MS,
- * so a busy flag that never clears cannot stop sending for good. A refusal more than BUSY_GAP_MS after the previous
- * one starts a new busy period.
+ * Sending is refused while the radio is busy, so the packet being received is not lost. A busy period starts when a
+ * send first finds the radio busy, and ends when one finds it free. Once a busy period has lasted max_busy_ms, sends
+ * go ahead, so a busy flag that never clears cannot stop sending for good, however often sends are tried. Code
+ * generation sets max_busy_ms from the time the radio takes to receive the largest packet.
  */
 class ListenBeforeTalk {
  public:
-  /// Longer than any packet takes to receive.
-  static constexpr uint32_t MAX_BUSY_MS = 2000;
-  static constexpr uint32_t BUSY_GAP_MS = 500;
+  static constexpr uint32_t DEFAULT_MAX_BUSY_MS = 2000;
+
+  void set_max_busy(uint32_t max_busy_ms) { this->max_busy_ms_ = max_busy_ms; }
+  uint32_t get_max_busy() const { return this->max_busy_ms_; }
 
   /// True if a packet may be sent at \p now; \p busy says whether the radio is receiving.
   bool may_send(bool busy, uint32_t now) {
-    if (!busy)
+    if (!busy) {
+      this->busy_ = false;
       return true;
-    if (now - this->last_refused_ms_ > BUSY_GAP_MS)
+    }
+    if (!this->busy_) {
+      this->busy_ = true;
       this->busy_since_ms_ = now;
-    this->last_refused_ms_ = now;
-    return now - this->busy_since_ms_ >= MAX_BUSY_MS;
+    }
+    return now - this->busy_since_ms_ >= this->max_busy_ms_;
   }
   /// How long the current busy period has lasted at \p now.
   uint32_t busy_for(uint32_t now) const { return now - this->busy_since_ms_; }
 
  protected:
   uint32_t busy_since_ms_{0};
-  uint32_t last_refused_ms_{0};
+  uint32_t max_busy_ms_{DEFAULT_MAX_BUSY_MS};
+  bool busy_{false};
 };
 
 }  // namespace esphome::packet_interface

esphome/components/packet_interface/radio_airtime.py

@@ -0,0 +1,48 @@
+"""How long a radio packet takes to send, for the listen before talk limit of the radio links."""
+
+import math
+
+from esphome.types import ConfigType
+
+# Listen before talk holds a send back for at most the airtime of the largest packet, and never less than this
+MIN_MAX_BUSY_MS = 2000
+# Room for the preamble detection and the radio's own delays
+MARGIN = 1.25
+
+
+def packet_airtime_ms(radio: ConfigType, size: int) -> float:
+    """The time a radio configured as in radio takes to send a packet of size bytes.
+
+    LoRa follows the Semtech formula for an explicit header; the low data rate optimization is assumed when a symbol
+    takes more than 16 ms, as the drivers enable it. Other modulations count the preamble, sync word, length byte,
+    payload and CRC at the bitrate, treating the preamble size as bytes, which gives the longer time.
+    """
+    if str(radio["modulation"]) == "LORA":
+        spreading_factor = radio["spreading_factor"]
+        bandwidth_hz = (
+            float(str(radio["bandwidth"]).removesuffix("kHz").replace("_", ".")) * 1000
+        )
+        coding_rate = int(str(radio["coding_rate"])[-1]) - 4
+        symbol_s = (1 << spreading_factor) / bandwidth_hz
+        low_data_rate = 1 if symbol_s > 0.016 else 0
+        crc = 16 if radio.get("crc_enable", False) else 0
+        payload_symbols = 8 + max(
+            math.ceil(
+                (8 * size - 4 * spreading_factor + 28 + crc)
+                / (4 * (spreading_factor - 2 * low_data_rate))
+            )
+            * (coding_rate + 4),
+            0,
+        )
+        return (radio["preamble_size"] + 4.25 + payload_symbols) * symbol_s * 1000
+    bytes_sent = (
+        radio["preamble_size"] + len(radio.get("sync_value", [])) + 1 + size + 2
+    )
+    return bytes_sent * 8 / radio["bitrate"] * 1000
+
+
+def max_busy_ms(radio: ConfigType, max_packet_size: int) -> int:
+    """How long listen before talk may hold a send back."""
+    return max(
+        MIN_MAX_BUSY_MS, math.ceil(packet_airtime_ms(radio, max_packet_size) * MARGIN)
+    )

esphome/components/sx126x/packet_interface/__init__.py

@@ -1,14 +1,25 @@
 import esphome.codegen as cg
 from esphome.components import packet_interface
 from esphome.components.packet_interface import (
+    CONF_MAX_PACKET_SIZE,
     new_packet_interface,
     packet_interface_schema,
 )
+from esphome.components.packet_interface.radio_airtime import max_busy_ms
 import esphome.config_validation as cv
+from esphome.const import CONF_ID
+from esphome.core import CORE
 import esphome.final_validate as fv
 from esphome.types import ConfigType
 
-from .. import CONF_PAYLOAD_LENGTH, CONF_SX126X_ID, SX126x, SX126xListener, sx126x_ns
+from .. import (
+    CONF_PAYLOAD_LENGTH,
+    CONF_SX126X_ID,
+    DOMAIN,
+    SX126x,
+    SX126xListener,
+    sx126x_ns,
+)
 
 CODEOWNERS = ["@clydebarrow"]
 DEPENDENCIES = ["sx126x"]
@@ -41,4 +52,8 @@ FINAL_VALIDATE_SCHEMA = _final_validate
 
 async def to_code(config: ConfigType) -> None:
     radio = await cg.get_variable(config[CONF_SX126X_ID])
-    await new_packet_interface(config, radio)
+    var = await new_packet_interface(config, radio)
+    radio_config = next(
+        conf for conf in CORE.config[DOMAIN] if conf[CONF_ID] == config[CONF_SX126X_ID]
+    )
+    cg.add(var.set_max_busy(max_busy_ms(radio_config, config[CONF_MAX_PACKET_SIZE])))

esphome/components/sx126x/packet_interface/sx126x_packet_interface.h

@@ -14,6 +14,7 @@ class SX126xPacketInterface : public packet_interface::PacketInterface, public S
   explicit SX126xPacketInterface(SX126x *radio) : radio_(radio) {}
 
   void setup() override { this->radio_->register_listener(this); }
+  void set_max_busy(uint32_t max_busy_ms) { this->listen_before_talk_.set_max_busy(max_busy_ms); }
   bool send_to_interface(const packet_interface::PacketBuffer &data,
                          const packet_interface::PacketMetaData &meta_data) override;
   void on_packet(const std::vector<uint8_t> &packet, float rssi, float snr) override;

esphome/components/sx127x/packet_interface/__init__.py

@@ -5,7 +5,10 @@ from esphome.components.packet_interface import (
     new_packet_interface,
     packet_interface_schema,
 )
+from esphome.components.packet_interface.radio_airtime import max_busy_ms
 import esphome.config_validation as cv
+from esphome.const import CONF_ID
+from esphome.core import CORE
 import esphome.final_validate as fv
 from esphome.types import ConfigType
 
@@ -13,6 +16,7 @@ from .. import (
     CONF_MODULATION,
     CONF_PAYLOAD_LENGTH,
     CONF_SX127X_ID,
+    DOMAIN,
     SX127x,
     SX127xListener,
     sx127x_ns,
@@ -62,4 +66,8 @@ FINAL_VALIDATE_SCHEMA = _final_validate
 
 async def to_code(config: ConfigType) -> None:
     radio = await cg.get_variable(config[CONF_SX127X_ID])
-    await new_packet_interface(config, radio)
+    var = await new_packet_interface(config, radio)
+    radio_config = next(
+        conf for conf in CORE.config[DOMAIN] if conf[CONF_ID] == config[CONF_SX127X_ID]
+    )
+    cg.add(var.set_max_busy(max_busy_ms(radio_config, config[CONF_MAX_PACKET_SIZE])))

esphome/components/sx127x/packet_interface/sx127x_packet_interface.h

@@ -14,6 +14,7 @@ class SX127xPacketInterface : public packet_interface::PacketInterface, public S
   explicit SX127xPacketInterface(SX127x *radio) : radio_(radio) {}
 
   void setup() override { this->radio_->register_listener(this); }
+  void set_max_busy(uint32_t max_busy_ms) { this->listen_before_talk_.set_max_busy(max_busy_ms); }
   bool send_to_interface(const packet_interface::PacketBuffer &data,
                          const packet_interface::PacketMetaData &meta_data) override;
   void on_packet(const std::vector<uint8_t> &packet, float rssi, float snr) override;

esphome/components/sx127x/sx127x.cpp

@@ -358,8 +358,11 @@ bool SX127x::is_receiving() {
   // In receive mode DIO0 signals a received packet; the transmit FIFO would overwrite it
   if (this->dio0_pin_ != nullptr && this->dio0_pin_->digital_read())
     return true;
-  if (this->modulation_ != MOD_LORA)
-    return false;
+  if (this->modulation_ != MOD_LORA) {
+    // Set from a detected preamble or sync word until the packet has been read
+    return (this->read_register_(REG_IRQ_FLAGS_1) & (IRQ_FLAGS_1_PREAMBLE_DETECT | IRQ_FLAGS_1_SYNC_ADDRESS_MATCH)) !=
+           0;
+  }
   return (this->read_register_(REG_MODEM_STAT) &
           (MODEM_STAT_SIGNAL_DETECTED | MODEM_STAT_SIGNAL_SYNCHRONIZED | MODEM_STAT_HEADER_VALID)) != 0;
 }

esphome/components/sx127x/sx127x_reg.h

@@ -45,6 +45,7 @@ enum SX127xReg : uint8_t {
   REG_PAYLOAD_LENGTH_LSB = 0x32,
   REG_FIFO_THRESH = 0x35,
   REG_IMAGE_CAL = 0x3B,
+  REG_IRQ_FLAGS_1 = 0x3E,
   // LoRa registers
   REG_FIFO_ADDR_PTR = 0x0D,
   REG_FIFO_TX_BASE_ADDR = 0x0E,
@@ -259,6 +260,11 @@ enum SX127xIrqFlags : uint8_t {
   CAD_DETECTED = 0x01,
 };
 
+enum SX127xIrqFlags1 : uint8_t {
+  IRQ_FLAGS_1_PREAMBLE_DETECT = 0x02,
+  IRQ_FLAGS_1_SYNC_ADDRESS_MATCH = 0x01,
+};
+
 enum SX127xModemStat : uint8_t {
   MODEM_STAT_SIGNAL_DETECTED = 0x01,
   MODEM_STAT_SIGNAL_SYNCHRONIZED = 0x02,

tests/components/packet_interface/listen_before_talk.cpp

@@ -4,6 +4,7 @@
 namespace esphome::packet_interface::lbt_testing {
 
 static constexpr uint32_t START = 10000;
+static constexpr uint32_t MAX_BUSY = ListenBeforeTalk::DEFAULT_MAX_BUSY_MS;
 
 TEST(ListenBeforeTalkTest, SendsWhenNotBusy) {
   ListenBeforeTalk lbt;
@@ -15,23 +16,37 @@ TEST(ListenBeforeTalkTest, RefusesWhileBusyUpToLimit) {
   EXPECT_FALSE(lbt.may_send(true, START));
   // Refused every 50 ms, as the reliability layer retries
   uint32_t now = START;
-  while (now - START < ListenBeforeTalk::MAX_BUSY_MS - 50) {
+  while (now - START < MAX_BUSY - 50) {
     now += 50;
     EXPECT_FALSE(lbt.may_send(true, now));
   }
   now += 50;
   EXPECT_TRUE(lbt.may_send(true, now));
-  EXPECT_EQ(lbt.busy_for(now), ListenBeforeTalk::MAX_BUSY_MS);
+  EXPECT_EQ(lbt.busy_for(now), MAX_BUSY);
 }
 
-TEST(ListenBeforeTalkTest, GapStartsNewBusyPeriod) {
+TEST(ListenBeforeTalkTest, BusyFlagThatNeverClearsReleasedForSlowSends) {
+  ListenBeforeTalk lbt;
+  // A send every 10 s, as session keep-alives, always finding the radio busy
+  EXPECT_FALSE(lbt.may_send(true, START));
+  EXPECT_TRUE(lbt.may_send(true, START + 10000));
+}
+
+TEST(ListenBeforeTalkTest, FreeRadioEndsBusyPeriod) {
   ListenBeforeTalk lbt;
   EXPECT_FALSE(lbt.may_send(true, START));
-  EXPECT_FALSE(lbt.may_send(true, START + 400));
-  EXPECT_EQ(lbt.busy_for(START + 400), 400);
-  uint32_t later = START + 400 + ListenBeforeTalk::BUSY_GAP_MS + 1;
-  EXPECT_FALSE(lbt.may_send(true, later));
-  EXPECT_EQ(lbt.busy_for(later), 0);
+  EXPECT_TRUE(lbt.may_send(false, START + 1500));
+  // A new busy period starts, so the earlier one does not count
+  EXPECT_FALSE(lbt.may_send(true, START + MAX_BUSY + 100));
+  EXPECT_EQ(lbt.busy_for(START + MAX_BUSY + 100), 0);
+}
+
+TEST(ListenBeforeTalkTest, LimitCanBeLonger) {
+  ListenBeforeTalk lbt;
+  lbt.set_max_busy(8000);
+  EXPECT_FALSE(lbt.may_send(true, START));
+  EXPECT_FALSE(lbt.may_send(true, START + 7999));
+  EXPECT_TRUE(lbt.may_send(true, START + 8000));
 }
 
 }  // namespace esphome::packet_interface::lbt_testing

tests/unit_tests/test_lora_packet_interface.py

@@ -3,9 +3,10 @@
 import pytest
 
 from esphome import config_validation as cv
-from esphome.components.packet_interface import CONF_MAX_PACKET_SIZE
+from esphome.components.packet_interface import CONF_MAX_PACKET_SIZE, radio_airtime
 from esphome.components.sx126x import packet_interface as sx126x_packet_interface
 from esphome.components.sx127x import packet_interface as sx127x_packet_interface
+import esphome.final_validate as fv
 
 
 @pytest.mark.parametrize(
@@ -22,3 +23,108 @@ def test_max_packet_size_is_radio_limit(
     assert schema({radio_key: "radio"})[CONF_MAX_PACKET_SIZE] == 255
     with pytest.raises(cv.Invalid):
         schema({radio_key: "radio", CONF_MAX_PACKET_SIZE: 256})
+
+
+def _lora(**options: object) -> dict:
+    return {
+        "modulation": "LORA",
+        "spreading_factor": 7,
+        "bandwidth": "125_0kHz",
+        "coding_rate": "CR_4_5",
+        "preamble_size": 8,
+        "crc_enable": False,
+        **options,
+    }
+
+
+@pytest.mark.parametrize(
+    ("radio", "airtime_ms"),
+    [
+        # Worked by hand from the Semtech formula, for a 255 byte packet with an explicit header
+        (_lora(spreading_factor=7), 394.5),
+        (_lora(spreading_factor=7, crc_enable=True), 399.6),
+        (_lora(spreading_factor=10), 2254.8),
+        # A symbol takes over 16 ms, so the low data rate optimization lengthens the packet
+        (_lora(spreading_factor=12), 9019.4),
+    ],
+)
+def test_lora_airtime(radio: dict, airtime_ms: float) -> None:
+    """The airtime of the largest LoRa packet follows the Semtech formula."""
+    assert radio_airtime.packet_airtime_ms(radio, 255) == pytest.approx(
+        airtime_ms, abs=0.5
+    )
+
+
+def test_max_busy_follows_airtime() -> None:
+    """Listen before talk waits at least 2 s, and longer than the largest packet takes."""
+    assert radio_airtime.max_busy_ms(_lora(), 255) == 2000
+    assert radio_airtime.max_busy_ms(_lora(spreading_factor=12), 255) > 9019
+    fsk = {
+        "modulation": "FSK",
+        "preamble_size": 4,
+        "bitrate": 1200,
+        "sync_value": [1, 2],
+    }
+    assert radio_airtime.packet_airtime_ms(fsk, 64) == pytest.approx(
+        (4 + 2 + 1 + 64 + 2) * 8 / 1200 * 1000
+    )
+
+
+class _FullConfig(dict):
+    """Enough of the full configuration for the radio links' final validation."""
+
+    def __init__(self, domain: str, radio: dict) -> None:
+        super().__init__({domain: [radio]})
+        self.domain = domain
+
+    def get_path_for_id(self, id_: object) -> list:
+        return [self.domain, 0, "id"]
+
+    def get_config_for_path(self, path: list) -> dict:
+        return self[path[0]][path[1]]
+
+
+@pytest.mark.parametrize(
+    ("module", "domain", "radio", "link", "match"),
+    [
+        (
+            sx126x_packet_interface,
+            "sx126x",
+            {"payload_length": 10},
+            {},
+            "payload_length",
+        ),
+        (
+            sx127x_packet_interface,
+            "sx127x",
+            _lora(payload_length=10),
+            {},
+            "payload_length",
+        ),
+        (
+            sx127x_packet_interface,
+            "sx127x",
+            {"modulation": "FSK", "payload_length": 0},
+            {CONF_MAX_PACKET_SIZE: 65},
+            "at most 64",
+        ),
+    ],
+)
+def test_radio_settings_checked(
+    module: object, domain: str, radio: dict, link: dict, match: str
+) -> None:
+    """Packets vary in length, and FSK on an SX127x holds at most 64 bytes."""
+    token = fv.full_config.set(_FullConfig(domain, radio))
+    try:
+        with pytest.raises(cv.Invalid, match=match):
+            module._final_validate(
+                {f"{domain}_id": "radio", CONF_MAX_PACKET_SIZE: 64, **link}
+            )
+        fixed = {**radio, "payload_length": 0}
+        fv.full_config.set(_FullConfig(domain, fixed))
+        if CONF_MAX_PACKET_SIZE not in link:
+            assert module._final_validate(
+                {f"{domain}_id": "radio", CONF_MAX_PACKET_SIZE: 64}
+            )
+    finally:
+        fv.full_config.reset(token)