feat(gateway): implement serial communication handling in DaliDomainService and GatewayController

Co-authored-by: Copilot <copilot@github.com>

components/dali_domain/include/dali_domain.hpp

@@ -121,6 +121,9 @@ class DaliDomainService {
   DaliChannel* findChannelByIndex(uint8_t channel_index);
   const DaliChannel* findChannelByHardwareBus(uint8_t bus_id) const;
   bool hasSerialPort(int uart_port) const;
+  esp_err_t startSerialRxTask(DaliChannel& channel);
+  static void SerialRxTaskEntry(void* arg);
+  void serialRxTaskLoop(DaliChannel* channel);
   esp_err_t startRawFrameTask();
   static void RawFrameTaskEntry(void* arg);
   void rawFrameTaskLoop();

components/dali_domain/src/dali_domain.cpp

@@ -5,6 +5,7 @@
 #include "dali.hpp"
 #include "driver/uart.h"
 #include "esp_log.h"
+#include "freertos/queue.h"
 
 #include <algorithm>
 #include <utility>
@@ -14,6 +15,13 @@ namespace gateway {
 namespace {
 
 constexpr const char* kTag = "dali_domain";
+constexpr size_t kSerialRxPacketMaxBytes = 8;
+constexpr UBaseType_t kSerialRxQueueDepth = 8;
+
+struct SerialRxPacket {
+  size_t len{0};
+  uint8_t data[kSerialRxPacketMaxBytes]{};
+};
 
 bool SendHardwareFrame(uint8_t bus_id, const uint8_t* data, size_t len) {
   if (data == nullptr || len != 3) {
@@ -74,35 +82,66 @@ bool WriteSerialFrame(int uart_port, const uint8_t* data, size_t len) {
   return uart_write_bytes(static_cast<uart_port_t>(uart_port), data, len) == static_cast<int>(len);
 }
 
-std::vector<uint8_t> ReadSerialFrame(int uart_port, size_t len, uint32_t timeout_ms) {
+std::vector<uint8_t> PacketToVector(const SerialRxPacket& packet, size_t len = 0) {
-  std::vector<uint8_t> data(len, 0);
+  const size_t out_len = len == 0 ? packet.len : std::min(packet.len, len);
-  const int read_len = uart_read_bytes(static_cast<uart_port_t>(uart_port), data.data(), len,
+  return std::vector<uint8_t>(packet.data, packet.data + out_len);
-                                       pdMS_TO_TICKS(timeout_ms));
-  if (read_len <= 0) {
-    return {};
-  }
-  data.resize(static_cast<size_t>(read_len));
-  return data;
 }
 
-std::vector<uint8_t> TransactSerialFrame(int uart_port, uint32_t query_timeout_ms,
+void DrainSerialQueue(QueueHandle_t queue) {
-                                         const uint8_t* data, size_t len) {
+  if (queue == nullptr) {
+    return;
+  }
+  SerialRxPacket packet;
+  while (xQueueReceive(queue, &packet, 0) == pdTRUE) {
+  }
+}
+
+std::vector<uint8_t> ReadSerialFrame(QueueHandle_t queue, size_t len, uint32_t timeout_ms) {
+  if (queue == nullptr) {
+    return {};
+  }
+  SerialRxPacket packet;
+  if (xQueueReceive(queue, &packet, pdMS_TO_TICKS(timeout_ms)) != pdTRUE) {
+    return {};
+  }
+  return PacketToVector(packet, len);
+}
+
+std::vector<uint8_t> TransactSerialFrame(int uart_port, QueueHandle_t queue,
+                                         uint32_t query_timeout_ms, const uint8_t* data,
+                                         size_t len) {
+  if (data != nullptr && len > 0 && data[0] == 0x12) {
+    DrainSerialQueue(queue);
+  }
   if (!WriteSerialFrame(uart_port, data, len)) {
     return {0xFD};
   }
   if (data == nullptr || len == 0 || data[0] != 0x12) {
     return {0xFF};
   }
-  const auto response = ReadSerialFrame(uart_port, 2, query_timeout_ms);
+
-  if (response.empty()) {
+  const TickType_t timeout_ticks = pdMS_TO_TICKS(query_timeout_ms);
-    return {0xFE};
+  const TickType_t started = xTaskGetTickCount();
+  while ((xTaskGetTickCount() - started) < timeout_ticks) {
+    const TickType_t elapsed = xTaskGetTickCount() - started;
+    const TickType_t remaining = timeout_ticks > elapsed ? timeout_ticks - elapsed : 0;
+    SerialRxPacket packet;
+    if (xQueueReceive(queue, &packet, remaining) != pdTRUE) {
+      break;
     }
+    auto response = PacketToVector(packet, 2);
+    if (!response.empty() &&
+        (response[0] == 0xFF || response[0] == 0xFE || response[0] == 0xFD)) {
       return response;
     }
+  }
+  return {0xFE};
+}
 
 }  // namespace
 
 struct DaliDomainService::DaliChannel {
+  DaliDomainService* owner{nullptr};
   DaliChannelConfig config;
   DaliPhyKind phy_kind{DaliPhyKind::kCustom};
   DaliTransportHooks hooks;
@@ -110,6 +149,8 @@ struct DaliDomainService::DaliChannel {
   std::unique_ptr<Dali> dali;
   std::optional<DaliHardwareBusConfig> hardware_bus;
   std::optional<DaliSerialBusConfig> serial_bus;
+  QueueHandle_t serial_rx_queue{nullptr};
+  TaskHandle_t serial_rx_task_handle{nullptr};
 };
 
 DaliDomainService::DaliDomainService()
@@ -123,6 +164,7 @@ bool DaliDomainService::bindTransport(const DaliChannelConfig& config, DaliTrans
   }
 
   auto channel = std::make_unique<DaliChannel>();
+  channel->owner = this;
   channel->config = config;
   channel->hooks = std::move(hooks);
   channel->comm = std::make_unique<DaliComm>(channel->hooks.send, channel->hooks.read,
@@ -221,21 +263,28 @@ esp_err_t DaliDomainService::bindSerialBus(const DaliSerialBusConfig& config) {
   }
   uart_flush_input(uart);
 
+  QueueHandle_t serial_rx_queue = xQueueCreate(kSerialRxQueueDepth, sizeof(SerialRxPacket));
+  if (serial_rx_queue == nullptr) {
+    ESP_LOGE(kTag, "failed to create uart%d RX queue", config.uart_port);
+    return ESP_ERR_NO_MEM;
+  }
+
   DaliTransportHooks hooks;
   hooks.send = [uart_port = config.uart_port](const uint8_t* data, size_t len) {
     return WriteSerialFrame(uart_port, data, len);
   };
-  hooks.read = [uart_port = config.uart_port](size_t len, uint32_t timeout_ms) {
+  hooks.read = [serial_rx_queue](size_t len, uint32_t timeout_ms) {
-    return ReadSerialFrame(uart_port, len, timeout_ms);
+    return ReadSerialFrame(serial_rx_queue, len, timeout_ms);
   };
-  hooks.transact = [uart_port = config.uart_port, timeout_ms = config.query_timeout_ms](
+  hooks.transact = [uart_port = config.uart_port, serial_rx_queue,
-                       const uint8_t* data, size_t len) {
+                    timeout_ms = config.query_timeout_ms](const uint8_t* data, size_t len) {
-    return TransactSerialFrame(uart_port, timeout_ms, data, len);
+    return TransactSerialFrame(uart_port, serial_rx_queue, timeout_ms, data, len);
   };
   hooks.delay = [](uint32_t ms) { vTaskDelay(pdMS_TO_TICKS(ms)); };
 
   const DaliChannelConfig channel_config{config.channel_index, config.gateway_id, config.name};
   if (!bindTransport(channel_config, std::move(hooks))) {
+    vQueueDelete(serial_rx_queue);
     return ESP_ERR_INVALID_STATE;
   }
 
@@ -243,6 +292,13 @@ esp_err_t DaliDomainService::bindSerialBus(const DaliSerialBusConfig& config) {
   if (channel != nullptr) {
     channel->phy_kind = DaliPhyKind::kSerialUart;
     channel->serial_bus = config;
+    channel->serial_rx_queue = serial_rx_queue;
+    err = startSerialRxTask(*channel);
+    if (err != ESP_OK) {
+      ESP_LOGE(kTag, "failed to start uart%d RX task: %s", config.uart_port,
+               esp_err_to_name(err));
+      return err;
+    }
   }
   ESP_LOGI(kTag, "bound channel=%u gateway=%u serial uart%d tx=%d rx=%d baudrate=%lu",
            config.channel_index, config.gateway_id, config.uart_port, config.tx_pin, config.rx_pin,
@@ -472,6 +528,64 @@ const DaliDomainService::DaliChannel* DaliDomainService::findChannelByHardwareBu
   return it == channels_.end() ? nullptr : it->get();
 }
 
+esp_err_t DaliDomainService::startSerialRxTask(DaliChannel& channel) {
+  if (channel.serial_rx_task_handle != nullptr) {
+    return ESP_OK;
+  }
+  if (!channel.serial_bus.has_value() || channel.serial_rx_queue == nullptr) {
+    return ESP_ERR_INVALID_STATE;
+  }
+  const BaseType_t created = xTaskCreate(&DaliDomainService::SerialRxTaskEntry,
+                                         "dali_uart_rx", 4096, &channel, 4,
+                                         &channel.serial_rx_task_handle);
+  if (created != pdPASS) {
+    channel.serial_rx_task_handle = nullptr;
+    return ESP_ERR_NO_MEM;
+  }
+  return ESP_OK;
+}
+
+void DaliDomainService::SerialRxTaskEntry(void* arg) {
+  auto* channel = static_cast<DaliChannel*>(arg);
+  if (channel != nullptr && channel->owner != nullptr) {
+    channel->owner->serialRxTaskLoop(channel);
+  }
+  vTaskDelete(nullptr);
+}
+
+void DaliDomainService::serialRxTaskLoop(DaliChannel* channel) {
+  if (channel == nullptr || !channel->serial_bus.has_value() ||
+      channel->serial_rx_queue == nullptr) {
+    return;
+  }
+
+  const auto uart = static_cast<uart_port_t>(channel->serial_bus->uart_port);
+  while (true) {
+    SerialRxPacket packet;
+    const int read_len = uart_read_bytes(uart, packet.data, sizeof(packet.data),
+                                         pdMS_TO_TICKS(20));
+    if (read_len <= 0) {
+      continue;
+    }
+    packet.len = std::min<size_t>(static_cast<size_t>(read_len), sizeof(packet.data));
+    if (xQueueSend(channel->serial_rx_queue, &packet, 0) != pdTRUE) {
+      SerialRxPacket dropped;
+      xQueueReceive(channel->serial_rx_queue, &dropped, 0);
+      xQueueSend(channel->serial_rx_queue, &packet, 0);
+    }
+
+    if (packet.len != 2 && packet.len != 3) {
+      continue;
+    }
+    DaliRawFrame frame;
+    frame.channel_index = channel->config.channel_index;
+    frame.gateway_id = channel->config.gateway_id;
+    frame.phy_kind = channel->phy_kind;
+    frame.data.assign(packet.data, packet.data + packet.len);
+    notifyRawFrameSinks(frame);
+  }
+}
+
 esp_err_t DaliDomainService::startRawFrameTask() {
   if (raw_frame_task_handle_ != nullptr) {
     return ESP_OK;

components/gateway_controller/include/gateway_controller.hpp

@@ -16,6 +16,7 @@
 namespace gateway {
 
 class DaliDomainService;
+struct DaliRawFrame;
 class GatewayRuntime;
 
 struct GatewayControllerConfig {
@@ -112,6 +113,7 @@ class GatewayController {
   void refreshRuntimeGatewayNames();
   void publishPayload(uint8_t gateway_id, const std::vector<uint8_t>& payload);
   void publishFrame(const std::vector<uint8_t>& frame);
+  void handleDaliRawFrame(const DaliRawFrame& frame);
 
   uint8_t resolveInternalGroupRawAddress(uint8_t gateway_id, uint8_t raw_addr);
   static uint8_t normalizeGroupTargetType(uint8_t target_type);

components/gateway_controller/src/gateway_controller.cpp

@@ -114,6 +114,7 @@ esp_err_t GatewayController::start() {
   runtime_.setCommandAddressResolver([this](uint8_t gateway_id, uint8_t raw_addr) {
     return resolveInternalGroupRawAddress(gateway_id, raw_addr);
   });
+  dali_domain_.addRawFrameSink([this](const DaliRawFrame& frame) { handleDaliRawFrame(frame); });
 
   for (const auto& channel : dali_domain_.channelInfo()) {
     sceneStore(channel.gateway_id);
@@ -508,6 +509,31 @@ void GatewayController::publishFrame(const std::vector<uint8_t>& frame) {
   }
 }
 
+void GatewayController::handleDaliRawFrame(const DaliRawFrame& frame) {
+  if (frame.data.size() != 2 && frame.data.size() != 3) {
+    return;
+  }
+  if (setup_mode_ || dali_domain_.isAllocAddr(frame.gateway_id) ||
+      runtime_.hasActiveQueryCommand(frame.gateway_id)) {
+    return;
+  }
+
+  uint8_t addr = 0;
+  uint8_t data = 0;
+  if (frame.data.size() == 2) {
+    addr = frame.data[0];
+    data = frame.data[1];
+    if (addr == 0xBE) {
+      return;
+    }
+  } else {
+    addr = frame.data[1];
+    data = frame.data[2];
+  }
+
+  publishPayload(frame.gateway_id, {0x01, frame.gateway_id, addr, data});
+}
+
 uint8_t GatewayController::resolveInternalGroupRawAddress(uint8_t gateway_id, uint8_t raw_addr) {
   if (raw_addr < 0x80 || raw_addr > 0x9f) {
     return raw_addr;

components/gateway_runtime/include/gateway_runtime.hpp

@@ -98,6 +98,7 @@ class GatewayRuntime {
   std::optional<std::vector<uint8_t>> popNextCommand();
   void completeCurrentCommand();
   bool hasPendingQueryCommand(const std::vector<uint8_t>& command) const;
+  bool hasActiveQueryCommand(uint8_t gateway_id) const;
   CommandDropReason lastEnqueueDropReason() const;
 
   void setGatewayCount(size_t gateway_count);

components/gateway_runtime/src/gateway_runtime.cpp

@@ -334,6 +334,12 @@ bool GatewayRuntime::hasPendingQueryCommand(const std::vector<uint8_t>& command)
                      });
 }
 
+bool GatewayRuntime::hasActiveQueryCommand(uint8_t gateway_id) const {
+  LockGuard guard(command_lock_);
+  return current_command_.has_value() && isQueryCommand(*current_command_) &&
+         current_command_->size() > 2 && (*current_command_)[2] == gateway_id;
+}
+
 GatewayRuntime::CommandDropReason GatewayRuntime::lastEnqueueDropReason() const {
   LockGuard guard(command_lock_);
   return last_enqueue_drop_reason_;