fix: move radio TX to a dedicated task so HA WebSocket stays connected

Web command handlers ran on the async_tcp task and blocked up to 3s
waiting for loop() to bit-bang radio frames. async_tcp also services the
WebSocket on port 8080, so pongs stalled and Home Assistant
(ping_timeout=15) disconnected when several covers were commanded.

- Add a radio_tx FreeRTOS task with a job queue; sendFrame/repeatFrame
  and repeater frames enqueue instead of transmitting inline.
- Serialize CC1101 access with a recursive mutex (TX task, apply, end,
  frequency scan, RSSI read).
- Make Web::queueCommand fire-and-forget; drop the done semaphore.
- Route /setMyPosition and /unpairShade through the command queue.
- Clamp repeats to MAX_TX_REPEATS (40).
- Pin AsyncTCP to core 0 on dual-core targets.

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

platformio.ini

@@ -39,14 +39,23 @@ monitor_filters =
 	esp32_exception_decoder
 	log2file
 
+; Dual-core targets: keep the async web/socket task on core 0 so the radio
+; task on core 1 (with loop()) never competes with network handling.
+[dualcore]
+build_flags =
+	${env.build_flags}
+	-DCONFIG_ASYNC_TCP_RUNNING_CORE=0
+
 [env:esp32dev]
 board = esp32dev
 board_build.partitions = esp32_3MB.csv
+build_flags = ${dualcore.build_flags}
 
 [env:esp32devdbg]
 board = esp32dev
 build_type = debug
 board_build.partitions = esp32_3MB.csv
+build_flags = ${dualcore.build_flags}
 
 [env:esp32c3]
 board = esp32-c3-devkitm-1
@@ -54,6 +63,7 @@ board_build.partitions = esp32_3MB.csv
 
 [env:esp32s3]
 board = esp32-s3-devkitc-1
+build_flags = ${dualcore.build_flags}
 
 [env:esp32c6]
 platform = https://github.com/pioarduino/platform-espressif32/releases/download/stable/platform-espressif32.zip

src/Somfy.cpp

@@ -2,6 +2,8 @@
 #include <ELECHOUSE_CC1101_SRC_DRV.h>
 #include <SPI.h>
 #include <esp_task_wdt.h>
+#include <freertos/queue.h>
+#include <freertos/semphr.h>
 #include <esp_chip_info.h>
 #include "driver/gpio.h"
 #include "esp_log.h"
@@ -36,6 +38,22 @@ uint8_t rxmode = 0;  // Indicates whether the radio is in receive mode.  Just to
 #define SETMY_REPEATS 35
 #define TILT_REPEATS 15
 #define TX_QUEUE_DELAY 100
+#define TX_TASK_PRIORITY 5
+#define TX_TASK_STACK 4096
+
+// Serializes all access to the CC1101 (SPI) and the TX/RX pins between the
+// radio task, the main loop and the async web server task.
+static SemaphoreHandle_t radioMutex = nullptr;
+static QueueHandle_t txJobQueue = nullptr;
+static TaskHandle_t txTaskHandle = nullptr;
+struct RadioLock {
+  bool locked = true;  // Before the mutex exists there is nothing to serialize against.
+  bool owned = false;
+  RadioLock(TickType_t wait = portMAX_DELAY) {
+    if(radioMutex) locked = owned = xSemaphoreTakeRecursive(radioMutex, wait) == pdTRUE;
+  }
+  ~RadioLock() { if(owned) xSemaphoreGiveRecursive(radioMutex); }
+};
 
 int sort_asc(const void *cmp1, const void *cmp2) {
   int a = *((uint8_t *)cmp1);
@@ -246,7 +264,11 @@ void somfy_frame_t::decodeFrame(somfy_rx_t *rx) {
   this->hwsync = rx->cpt_synchro_hw;
   this->pulseCount = rx->pulseCount;
   this->bitLength = rx->bit_length;
-  this->rssi = ELECHOUSE_cc1101.getRssi();
+  {
+    // Don't stall the main loop behind a transmission just to read the RSSI.
+    RadioLock lock(0);
+    if(lock.locked) this->rssi = ELECHOUSE_cc1101.getRssi();
+  }
   this->decodeFrame(rx->payload);
 }
 byte somfy_frame_t::encode80Byte7(byte start, uint8_t repeat) {
@@ -1147,7 +1169,6 @@ void SomfyShade::checkMovement() {
         if(!isAtTarget()) {
           ESP_LOGD(TAG, "We are not at our tilt target: %.2f", this->tiltTarget);
           if(this->target != 100.0) SomfyRemote::sendCommand(somfy_commands::My, this->repeats);
-          delay(100);
           // We now need to move the tilt to the position we requested.
           this->moveToTiltTarget(this->tiltTarget);
         }
@@ -1199,7 +1220,6 @@ void SomfyShade::checkMovement() {
         if(!isAtTarget()) {
           ESP_LOGD(TAG, "We are not at our tilt target: %.2f", this->tiltTarget);
           if(this->target != 0.0) SomfyRemote::sendCommand(somfy_commands::My, this->repeats);
-          delay(100);
           // We now need to move the tilt to the position we requested.
           this->moveToTiltTarget(this->tiltTarget);
         }
@@ -1313,7 +1333,6 @@ void SomfyShade::checkMovement() {
     }
   }
   if(this->settingMyPos && this->isAtTarget()) {
-    delay(200);
     // Set this position before sending the command.  If you don't the processFrame function
     // will send the shade back to its original My position.
     if(this->tiltType != tilt_types::none) {
@@ -3835,33 +3854,30 @@ void SomfyRemote::repeatFrame(uint8_t repeat) {
     this->triggerGPIOs(this->lastFrame);
     return;
   }
-  somfy.transceiver.beginTransmit();
+  if(repeat > MAX_TX_REPEATS) repeat = MAX_TX_REPEATS;
+  // The radio task encodes the first frame with the current repeat counter and
+  // then counts up from there for each repeat frame.
   byte frm[10];
   this->lastFrame.encodeFrame(frm);
-  this->lastFrame.repeats++;
+  somfy_tx_job_t job;
-  somfy.transceiver.sendFrame(frm, this->bitLength == 56 ? 2 : 12, this->bitLength);
+  job.frame = this->lastFrame;
-  for(uint8_t i = 0; i < repeat; i++) {
+  job.repeats = repeat;
-    this->lastFrame.repeats++;
+  job.firstRepeat = this->lastFrame.repeats + 2;
-    if(this->lastFrame.bitLength == 80) this->lastFrame.encode80BitFrame(&frm[0], this->lastFrame.repeats);
+  this->lastFrame.repeats += repeat + 1;
-    somfy.transceiver.sendFrame(frm, this->bitLength == 56 ? 7 : 6, this->bitLength);
+  somfy.transceiver.queueTx(job);
-    esp_task_wdt_reset();
-  }
-  somfy.transceiver.endTransmit();
   //somfy.processFrame(this->lastFrame, true);
 }
 void SomfyShadeController::sendFrame(somfy_frame_t &frame, uint8_t repeat) {
-  somfy.transceiver.beginTransmit();
+  if(repeat > MAX_TX_REPEATS) repeat = MAX_TX_REPEATS;
+  // Encode once here so any side effects of the encoding on the frame
+  // (valid flag, default step size) are visible to the caller immediately.
   byte frm[10];
   frame.encodeFrame(frm);
-  this->transceiver.sendFrame(frm, frame.bitLength == 56 ? 2 : 12, frame.bitLength);
+  somfy_tx_job_t job;
-  for(uint8_t i = 0; i < repeat; i++) {
+  job.frame = frame;
-    // For each 80-bit frame we need to adjust the byte encoding for the
+  job.repeats = repeat;
-    // silence.
+  job.firstRepeat = 1;
-    if(frame.bitLength == 80) frame.encode80BitFrame(&frm[0], i + 1);
+  this->transceiver.queueTx(job);
-    this->transceiver.sendFrame(frm, frame.bitLength == 56 ? 7 : 6, frame.bitLength);
-    esp_task_wdt_reset();
-  }
-  this->transceiver.endTransmit();
 }
 bool SomfyShadeController::deleteShade(uint8_t shadeId) {
   for(uint8_t i = 0; i < SOMFY_MAX_SHADES; i++) {
@@ -4344,6 +4360,7 @@ float markFreq = 433.0f;
 int markRSSI = -100;
 uint32_t lastScan = 0;
 void Transceiver::beginFrequencyScan() {
+  RadioLock lock;
   if(this->config.enabled) {
     this->disableReceive();
     rxmode = 3;
@@ -4360,6 +4377,7 @@ void Transceiver::beginFrequencyScan() {
   }
 }
 void Transceiver::processFrequencyScan(bool received) {
+  RadioLock lock;
   if(this->config.enabled && rxmode == 3) {
     if(received) {
       currRSSI = ELECHOUSE_cc1101.getRssi();
@@ -4387,6 +4405,7 @@ void Transceiver::processFrequencyScan(bool received) {
   }
 }
 void Transceiver::endFrequencyScan() {
+  RadioLock lock;
   if(rxmode == 3) {
     rxmode = 0;
     if(interruptPin > 0) detachInterrupt(interruptPin); 
@@ -4545,6 +4564,8 @@ bool Transceiver::save() {
     return true;
 }
 bool Transceiver::end() {
+    this->flushTx();
+    RadioLock lock;
     this->disableReceive();
     return true;
 }
@@ -4779,6 +4800,10 @@ void transceiver_config_t::load() {
     //this->printBuffer = somfy.transceiver.printBuffer;
 }
 void transceiver_config_t::apply() {
+    // Frames queued under the old settings are discarded and any frame that is
+    // in flight completes before the radio is reconfigured.
+    somfy.transceiver.flushTx();
+    RadioLock lock;
     somfy.transceiver.disableReceive();
     bit_length = this->type;    
     if(this->enabled) {
@@ -4879,9 +4904,11 @@ void transceiver_config_t::apply() {
     //somfy.transceiver.printBuffer = this->printBuffer;
 }
 bool Transceiver::begin() {
+    if(!radioMutex) radioMutex = xSemaphoreCreateRecursiveMutex();
     this->config.load();
     this->config.apply();
     rx_queue.init();
+    this->startTxTask();
     return true;
 }
 void Transceiver::loop() {
@@ -4914,11 +4941,32 @@ void Transceiver::loop() {
     somfy.processWaitingFrame();
     // Check to see if there is anything in the buffer
     if(tx_queue.length > 0 && millis() > tx_queue.delay_time && somfy_rx.cpt_synchro_hw == 0) {
-      this->beginTransmit();
+      // Repeater frames are handed to the radio task like any other transmission.
-      somfy_tx_t tx;
+      somfy_tx_job_t job;
-      
+      job.raw = true;
-      tx_queue.pop(&tx);
+      tx_queue.pop(&job.rawTx);
-      {
+      this->queueTx(job);
+      tx_queue.delay_time = millis() + TX_QUEUE_DELAY;
+    }
+  }
+}
+bool Transceiver::queueTx(const somfy_tx_job_t &job) {
+  if(!txJobQueue) {
+    ESP_LOGE(TAG, "Radio task not started, dropping transmission");
+    return false;
+  }
+  if(xQueueSend(txJobQueue, &job, 0) != pdTRUE) {
+    ESP_LOGE(TAG, "Radio TX queue full, dropping transmission");
+    return false;
+  }
+  return true;
+}
+void Transceiver::flushTx() {
+  if(txJobQueue) xQueueReset(txJobQueue);
+}
+void Transceiver::transmitJob(somfy_tx_job_t &job) {
+  if(job.raw) {
+    somfy_tx_t &tx = job.rawTx;
     char frame_buf[128];
     int fpos = snprintf(frame_buf, sizeof(frame_buf), "Sending frame %d - %d-BIT [", tx.hwsync, tx.bit_length);
     for(uint8_t j = 0; j < 10; j++) {
@@ -4926,24 +4974,51 @@ void Transceiver::loop() {
     }
     snprintf(frame_buf + fpos, sizeof(frame_buf) - fpos, "]");
     ESP_LOGD(TAG, "%s", frame_buf);
-      }
+    this->beginTransmit();
     this->sendFrame(tx.payload, tx.hwsync, tx.bit_length);
-      tx_queue.delay_time = millis() + TX_QUEUE_DELAY;
+    this->endTransmit();
-      
+    return;
-      /*
-      while(tx_queue.length > 0 && tx_queue.pop(&tx)) {
-        Serial.printf("Sending frame %d - %d-BIT [", tx.hwsync, tx.bit_length);
-        for(uint8_t j = 0; j < 10; j++) {
-          Serial.print(tx.payload[j]);
-          if(j < 9) Serial.print(", ");
   }
-        Serial.println("]");
+  somfy_frame_t &frame = job.frame;
-        this->sendFrame(tx.payload, tx.hwsync, tx.bit_length);
+  byte frm[10];
+  frame.encodeFrame(frm);
+  this->beginTransmit();
+  this->sendFrame(frm, frame.bitLength == 56 ? 2 : 12, frame.bitLength);
+  for(uint8_t i = 0; i < job.repeats; i++) {
+    // For each 80-bit frame we need to adjust the byte encoding for the
+    // silence.
+    if(frame.bitLength == 80) frame.encode80BitFrame(&frm[0], (uint8_t)(job.firstRepeat + i));
+    this->sendFrame(frm, frame.bitLength == 56 ? 7 : 6, frame.bitLength);
   }
-      */
   this->endTransmit();
 }
+void Transceiver::txTask(void *arg) {
+  Transceiver *trans = static_cast<Transceiver *>(arg);
+  somfy_tx_job_t job;
+  for(;;) {
+    if(xQueueReceive(txJobQueue, &job, portMAX_DELAY) != pdTRUE) continue;
+    {
+      RadioLock lock;
+      trans->transmitJob(job);
     }
+    // Give the motors a full frame beat between separate transmissions.
+    vTaskDelay(pdMS_TO_TICKS(TX_QUEUE_DELAY));
+  }
+}
+void Transceiver::startTxTask() {
+  if(txTaskHandle) return;
+  if(!txJobQueue) txJobQueue = xQueueCreate(TX_JOB_QUEUE_SIZE, sizeof(somfy_tx_job_t));
+  if(!txJobQueue) {
+    ESP_LOGE(TAG, "Unable to create radio TX queue");
+    return;
+  }
+#if CONFIG_FREERTOS_UNICORE
+  xTaskCreate(Transceiver::txTask, "radio_tx", TX_TASK_STACK, this, TX_TASK_PRIORITY, &txTaskHandle);
+#else
+  // Run beside loop() so the network stack on the other core is never held up.
+  xTaskCreatePinnedToCore(Transceiver::txTask, "radio_tx", TX_TASK_STACK, this, TX_TASK_PRIORITY, &txTaskHandle, ARDUINO_RUNNING_CORE);
+#endif
+  if(!txTaskHandle) ESP_LOGE(TAG, "Unable to start radio TX task");
 }
 somfy_frame_t& Transceiver::lastFrame() { return this->frame; }
 void Transceiver::beginTransmit() {

src/Somfy.h

@@ -201,6 +201,19 @@ struct somfy_frame_t {
     bool isSynonym(somfy_frame_t &f);
     void copy(somfy_frame_t &f);
 };
+// Upper bound on repeats for a single transmission.  Each frame is ~140ms of
+// radio time so this keeps a single job to roughly 6 seconds.
+#define MAX_TX_REPEATS 40
+#define TX_JOB_QUEUE_SIZE 32
+// A transmission handed off to the radio task.  The frame is a copy so the
+// radio task never touches the remote's lastFrame.
+struct somfy_tx_job_t {
+  somfy_frame_t frame;
+  uint8_t repeats = 0;      // Number of repeat frames after the first frame.
+  uint8_t firstRepeat = 1;  // 80-bit repeat counter used for the first repeat frame.
+  bool raw = false;         // When true rawTx holds a pre-encoded frame (repeater).
+  somfy_tx_t rawTx;
+};
 
 class SomfyRoom {
   public:
@@ -488,8 +501,11 @@ struct transceiver_config_t {
 class Transceiver {
   private:
     static void handleReceive();
+    static void txTask(void *arg);
     bool _received = false;
     somfy_frame_t frame;
+    void startTxTask();
+    void transmitJob(somfy_tx_job_t &job);
   public:
     transceiver_config_t config;
     bool printBuffer = false;
@@ -505,6 +521,8 @@ class Transceiver {
     void disableReceive();
     somfy_frame_t& lastFrame();
     void sendFrame(byte *frame, uint8_t sync, uint8_t bitLength = 56);
+    bool queueTx(const somfy_tx_job_t &job);
+    void flushTx();
     void beginTransmit();
     void endTransmit();
     void emitFrame(somfy_frame_t *frame, somfy_rx_t *rx = nullptr);

src/Web.cpp

@@ -45,14 +45,29 @@ static const char _encoding_json[] = "application/json";
 static const char *TAG = "Web";
 
 static QueueHandle_t webCmdQueue = nullptr;
-static SemaphoreHandle_t webCmdDone = nullptr;
+
+// Radio work for these handlers runs on the main loop via the command queue.
+static void queueSetMyPosition(SomfyShade *shade, int8_t pos, int8_t tilt) {
+  web_command_t cmd = {};
+  cmd.type = web_cmd_t::set_my_position;
+  cmd.shadeId = shade->getShadeId();
+  cmd.position = pos;
+  cmd.tiltPosition = tilt;
+  webServer.queueCommand(cmd);
+}
+static void queueUnpairShade(SomfyShade *shade) {
+  web_command_t cmd = {};
+  cmd.type = web_cmd_t::unpair_shade;
+  cmd.shadeId = shade->getShadeId();
+  // Reflect the change in the response; the main loop sets it again and saves.
+  if(webServer.queueCommand(cmd)) shade->paired = false;
+}
 
 AsyncWebServer asyncServer(80);
 AsyncWebServer asyncApiServer(8081);
 void Web::startup() {
   ESP_LOGI(TAG, "Launching web server...");
   if(!webCmdQueue) webCmdQueue = xQueueCreate(WEB_CMD_QUEUE_SIZE, sizeof(web_command_t));
-  if(!webCmdDone) webCmdDone = xSemaphoreCreateBinary();
 
   asyncServer.on("/loginContext", HTTP_GET, [](AsyncWebServerRequest *request) {
     AsyncJsonResponse *response = new AsyncJsonResponse();
@@ -74,22 +89,19 @@ void Web::loop() {
   delay(1);
 }
 bool Web::queueCommand(const web_command_t &cmd) {
-  if(!webCmdQueue || !webCmdDone) return false;
+  if(!webCmdQueue) return false;
-  // Clear any stale signal
+  // This runs on the async_tcp task which also services the sockets so it must
-  xSemaphoreTake(webCmdDone, 0);
+  // never wait on the main loop or the radio.
-  if(xQueueSend(webCmdQueue, &cmd, pdMS_TO_TICKS(100)) != pdTRUE) {
+  web_command_t c = cmd;
+  if(c.repeat > MAX_TX_REPEATS) c.repeat = MAX_TX_REPEATS;
+  if(xQueueSend(webCmdQueue, &c, 0) != pdTRUE) {
     ESP_LOGE(TAG, "Command queue full, dropping command");
     return false;
   }
-  // Wait for main loop to process it
-  if(xSemaphoreTake(webCmdDone, pdMS_TO_TICKS(WEB_CMD_TIMEOUT_MS)) != pdTRUE) {
-    ESP_LOGW(TAG, "Command queue timeout waiting for processing");
-    return false;
-  }
   return true;
 }
 void Web::processQueue() {
-  if(!webCmdQueue || !webCmdDone) return;
+  if(!webCmdQueue) return;
   web_command_t cmd;
   while(xQueueReceive(webCmdQueue, &cmd, 0) == pdTRUE) {
     switch(cmd.type) {
@@ -156,8 +168,21 @@ void Web::processQueue() {
         }
         break;
       }
+      case web_cmd_t::set_my_position: {
+        SomfyShade *shade = somfy.getShadeById(cmd.shadeId);
+        if(shade) shade->setMyPosition(cmd.position, cmd.tiltPosition);
+        break;
+      }
+      case web_cmd_t::unpair_shade: {
+        SomfyShade *shade = somfy.getShadeById(cmd.shadeId);
+        if(shade) {
+          shade->sendCommand(somfy_commands::Prog, shade->bitLength == 56 ? 7 : 1);
+          shade->paired = false;
+          shade->save();
+        }
+        break;
+      }
     }
-    xSemaphoreGive(webCmdDone);
   }
 }
 bool Web::isAuthenticated(AsyncWebServerRequest *request, bool cfg) {
@@ -1601,7 +1626,7 @@ void Web::begin() {
         if(tilt < 0) tilt = shade->myPos;
         if(shade->tiltType == tilt_types::none) tilt = -1;
         if(pos >= 0 && pos <= 100)
-          shade->setMyPosition(shade->transformPosition(pos), shade->transformPosition(tilt));
+          queueSetMyPosition(shade, shade->transformPosition(pos), shade->transformPosition(tilt));
         AsyncJsonResp resp;
         resp.beginResponse(request, g_async_content, sizeof(g_async_content));
         resp.beginObject();
@@ -1628,7 +1653,7 @@ void Web::begin() {
       if(tilt < 0) tilt = shade->myPos;
       if(shade->tiltType == tilt_types::none) tilt = -1;
       if(pos >= 0 && pos <= 100)
-        shade->setMyPosition(shade->transformPosition(pos), shade->transformPosition(tilt));
+        queueSetMyPosition(shade, shade->transformPosition(pos), shade->transformPosition(tilt));
       AsyncJsonResp resp;
       resp.beginResponse(request, g_async_content, sizeof(g_async_content));
       resp.beginObject();
@@ -1753,12 +1778,7 @@ void Web::begin() {
         request->send(500, _encoding_json, F("{\"status\":\"ERROR\",\"desc\":\"Shade not found to unpair\"}"));
       }
       else {
-        if(shade->bitLength == 56)
+        queueUnpairShade(shade);
-          shade->sendCommand(somfy_commands::Prog, 7);
-        else
-          shade->sendCommand(somfy_commands::Prog, 1);
-        shade->paired = false;
-        shade->save();
         AsyncJsonResp resp;
         resp.beginResponse(request, g_async_content, sizeof(g_async_content));
         resp.beginObject();
@@ -1777,12 +1797,7 @@ void Web::begin() {
       request->send(500, _encoding_json, F("{\"status\":\"ERROR\",\"desc\":\"Shade not found to unpair\"}"));
     }
     else {
-      if(shade->bitLength == 56)
+      queueUnpairShade(shade);
-        shade->sendCommand(somfy_commands::Prog, 7);
-      else
-        shade->sendCommand(somfy_commands::Prog, 1);
-      shade->paired = false;
-      shade->save();
       AsyncJsonResp resp;
       resp.beginResponse(request, g_async_content, sizeof(g_async_content));
       resp.beginObject();

src/Web.h

@@ -6,8 +6,9 @@
 #ifndef webserver_h
 #define webserver_h
 
-#define WEB_CMD_QUEUE_SIZE 8
+// Commands are queued by the async web server and drained by the main loop.
-#define WEB_CMD_TIMEOUT_MS 3000
+// Big enough for Home Assistant firing a command at every shade at once.
+#define WEB_CMD_QUEUE_SIZE 32
 
 enum class web_cmd_t : uint8_t {
     shade_command,      // moveToTarget or sendCommand
@@ -18,6 +19,8 @@ enum class web_cmd_t : uint8_t {
     set_positions,      // set shade position directly
     shade_sensor,       // shade sensor command
     group_sensor,       // group sensor command
+    set_my_position,    // shade setMyPosition (position/tiltPosition)
+    unpair_shade,       // shade unpair (Prog) and clear paired flag
 };
 
 struct web_command_t {
@@ -67,8 +70,8 @@ class Web {
     void handleBackup(AsyncWebServerRequest *request);
     void handleReboot(AsyncWebServerRequest *request);
     void handleNotFound(AsyncWebServerRequest *request);
+    bool queueCommand(const web_command_t &cmd);
   private:
     void processQueue();
-    bool queueCommand(const web_command_t &cmd);
 };
 #endif