copilot edge cases

esphome/analyze_memory/__init__.py

@@ -397,47 +397,38 @@ class MemoryAnalyzer:
             return pioenvs_dir
         return None
 
-    def _scan_cswtch_in_objects(
-        self, obj_dir: Path
-    ) -> dict[str, list[tuple[str, int]]]:
-        """Scan object files for CSWTCH symbols using a single nm invocation.
+    @staticmethod
+    def _parse_nm_cswtch_output(
+        output: str,
+        base_dir: Path | None,
+        cswtch_map: dict[str, list[tuple[str, int]]],
+    ) -> None:
+        """Parse nm output for CSWTCH symbols and add to cswtch_map.
 
-        Uses ``nm --print-file-name -S`` on all ``.o`` files at once.
-        Output format: ``/path/to/file.o:address size type name``
+        Handles both ``.o`` files and ``.a`` archives.
+
+        nm output formats::
+
+            .o files:  /path/file.o:hex_addr hex_size type name
+            .a files:  /path/lib.a:member.o:hex_addr hex_size type name
+
+        For ``.o`` files, paths are made relative to *base_dir* when possible.
+        For ``.a`` archives (detected by ``:`` in the file portion), paths are
+        formatted as ``archive_stem/member.o`` (e.g. ``liblwip2-536-feat/lwip-esp.o``).
 
         Args:
-            obj_dir: Directory containing object files (.pioenvs/<env>/)
-
-        Returns:
-            Dict mapping "CSWTCH$NNN:size" to list of (source_file, size) tuples.
+            output: Raw stdout from ``nm --print-file-name -S``.
+            base_dir: Base directory for computing relative paths of ``.o`` files.
+                      Pass ``None`` when scanning archives outside the build tree.
+            cswtch_map: Dict to populate, mapping ``"CSWTCH$N:size"`` to source list.
         """
-        cswtch_map: dict[str, list[tuple[str, int]]] = defaultdict(list)
-
-        if not self.nm_path:
-            return cswtch_map
-
-        # Find all .o files recursively, sorted for deterministic output
-        obj_files = sorted(obj_dir.rglob("*.o"))
-        if not obj_files:
-            return cswtch_map
-
-        _LOGGER.debug("Scanning %d object files for CSWTCH symbols", len(obj_files))
-
-        # Single nm call with --print-file-name for all object files
-        result = run_tool(
-            [self.nm_path, "--print-file-name", "-S"] + [str(f) for f in obj_files],
-            timeout=30,
-        )
-        if result is None or result.returncode != 0:
-            return cswtch_map
-
-        for line in result.stdout.splitlines():
+        for line in output.splitlines():
             if "CSWTCH$" not in line:
                 continue
 
             # Split on last ":" that precedes a hex address.
-            # nm --print-file-name format: filepath:hex_addr hex_size type name
-            # We split from the right: find the last colon followed by hex digits.
+            # For .o:  "filepath.o" : "hex_addr hex_size type name"
+            # For .a:  "filepath.a:member.o" : "hex_addr hex_size type name"
             parts_after_colon = line.rsplit(":", 1)
             if len(parts_after_colon) != 2:
                 continue
@@ -457,16 +448,89 @@ class MemoryAnalyzer:
             except ValueError:
                 continue
 
-            # Get relative path from obj_dir for readability
-            try:
-                rel_path = str(Path(file_path).relative_to(obj_dir))
-            except ValueError:
+            # Determine readable source path
+            # Use ".a:" to detect archive format (not bare ":" which matches
+            # Windows drive letters like "C:\...\file.o").
+            if ".a:" in file_path:
+                # Archive format: "archive.a:member.o" → "archive_stem/member.o"
+                archive_part, member = file_path.rsplit(":", 1)
+                archive_name = Path(archive_part).stem
+                rel_path = f"{archive_name}/{member}"
+            elif base_dir is not None:
+                try:
+                    rel_path = str(Path(file_path).relative_to(base_dir))
+                except ValueError:
+                    rel_path = file_path
+            else:
                 rel_path = file_path
 
             key = f"{sym_name}:{size}"
             cswtch_map[key].append((rel_path, size))
 
-        return cswtch_map
+    def _run_nm_cswtch_scan(
+        self,
+        files: list[Path],
+        base_dir: Path | None,
+        cswtch_map: dict[str, list[tuple[str, int]]],
+    ) -> None:
+        """Run nm on *files* and add any CSWTCH symbols to *cswtch_map*.
+
+        Args:
+            files: Object (``.o``) or archive (``.a``) files to scan.
+            base_dir: Base directory for relative path computation (see
+                      :meth:`_parse_nm_cswtch_output`).
+            cswtch_map: Dict to populate with results.
+        """
+        if not self.nm_path or not files:
+            return
+
+        _LOGGER.debug("Scanning %d files for CSWTCH symbols", len(files))
+
+        result = run_tool(
+            [self.nm_path, "--print-file-name", "-S"] + [str(f) for f in files],
+            timeout=30,
+        )
+        if result is None or result.returncode != 0:
+            _LOGGER.debug(
+                "nm failed or timed out scanning %d files for CSWTCH symbols",
+                len(files),
+            )
+            return
+
+        self._parse_nm_cswtch_output(result.stdout, base_dir, cswtch_map)
+
+    def _scan_cswtch_in_sdk_archives(
+        self, cswtch_map: dict[str, list[tuple[str, int]]]
+    ) -> None:
+        """Scan SDK library archives (.a) for CSWTCH symbols.
+
+        Prebuilt SDK libraries (e.g. lwip, bearssl) are not compiled from source,
+        so their CSWTCH symbols only exist inside ``.a`` archives.  Results are
+        merged into *cswtch_map* for keys not already found in ``.o`` files.
+
+        The same source file (e.g. ``lwip-esp.o``) often appears in multiple
+        library variants (``liblwip2-536.a``, ``liblwip2-1460-feat.a``, etc.),
+        so results are deduplicated by member name.
+        """
+        sdk_dirs = self._find_sdk_library_dirs()
+        if not sdk_dirs:
+            return
+
+        sdk_archives = sorted(a for sdk_dir in sdk_dirs for a in sdk_dir.glob("*.a"))
+
+        sdk_map: dict[str, list[tuple[str, int]]] = defaultdict(list)
+        self._run_nm_cswtch_scan(sdk_archives, None, sdk_map)
+
+        # Merge SDK results, deduplicating by member name.
+        for key, sources in sdk_map.items():
+            if key in cswtch_map:
+                continue
+            seen: dict[str, tuple[str, int]] = {}
+            for path, sz in sources:
+                member = Path(path).name
+                if member not in seen:
+                    seen[member] = (path, sz)
+            cswtch_map[key] = list(seen.values())
 
     def _source_file_to_component(self, source_file: str) -> str:
         """Map a source object file path to its component name.
@@ -505,17 +569,25 @@ class MemoryAnalyzer:
 
         CSWTCH symbols are compiler-generated lookup tables for switch statements.
         They are local symbols, so the same name can appear in different object files.
-        This method scans .o files to attribute them to their source components.
+        This method scans .o files and SDK archives to attribute them to their
+        source components.
         """
         obj_dir = self._find_object_files_dir()
         if obj_dir is None:
             _LOGGER.debug("No object files directory found, skipping CSWTCH analysis")
             return
 
-        # Scan object files for CSWTCH symbols
-        cswtch_map = self._scan_cswtch_in_objects(obj_dir)
+        # Scan build-dir object files for CSWTCH symbols
+        cswtch_map: dict[str, list[tuple[str, int]]] = defaultdict(list)
+        self._run_nm_cswtch_scan(sorted(obj_dir.rglob("*.o")), obj_dir, cswtch_map)
+
+        # Also scan SDK library archives (.a) for CSWTCH symbols.
+        # Prebuilt SDK libraries (e.g. lwip, bearssl) are not compiled from source
+        # so their symbols only exist inside .a archives, not as loose .o files.
+        self._scan_cswtch_in_sdk_archives(cswtch_map)
+
         if not cswtch_map:
-            _LOGGER.debug("No CSWTCH symbols found in object files")
+            _LOGGER.debug("No CSWTCH symbols found in object files or SDK archives")
             return
 
         # Collect CSWTCH symbols from the ELF (already parsed in sections)

esphome/codegen.py

@@ -87,7 +87,6 @@ from esphome.cpp_types import (  # noqa: F401
     size_t,
     std_ns,
     std_shared_ptr,
-    std_span,
     std_string,
     std_string_ref,
     std_vector,

esphome/components/epaper_spi/epaper_spi_spectra_e6.h

@@ -4,7 +4,7 @@
 
 namespace esphome::epaper_spi {
 
-class EPaperSpectraE6 final : public EPaperBase {
+class EPaperSpectraE6 : public EPaperBase {
  public:
   EPaperSpectraE6(const char *name, uint16_t width, uint16_t height, const uint8_t *init_sequence,
                   size_t init_sequence_length)

esphome/components/epaper_spi/epaper_waveshare.h

@@ -6,7 +6,7 @@ namespace esphome::epaper_spi {
 /**
  * An epaper display that needs LUTs to be sent to it.
  */
-class EpaperWaveshare final : public EPaperMono {
+class EpaperWaveshare : public EPaperMono {
  public:
   EpaperWaveshare(const char *name, uint16_t width, uint16_t height, const uint8_t *init_sequence,
                   size_t init_sequence_length, const uint8_t *lut, size_t lut_length, const uint8_t *partial_lut,

esphome/components/logger/logger_zephyr.cpp

@@ -68,7 +68,7 @@ void HOT Logger::write_msg_(const char *msg, uint16_t len) {
 #ifdef CONFIG_PRINTK
   // Requires the debug component and an active SWD connection.
   // It is used for pyocd rtt -t nrf52840
-  printk("%.*s", static_cast<int>(len), msg);
+  k_str_out(const_cast<char *>(msg), len);
 #endif
   if (this->uart_dev_ == nullptr) {
     return;

esphome/components/packet_transport/packet_transport.cpp

@@ -396,9 +396,9 @@ static bool process_rolling_code(Provider &provider, PacketDecoder &decoder) {
 /**
  * Process a received packet
  */
-void PacketTransport::process_(std::span<const uint8_t> data) {
+void PacketTransport::process_(const std::vector<uint8_t> &data) {
   auto ping_key_seen = !this->ping_pong_enable_;
-  PacketDecoder decoder(data.data(), data.size());
+  PacketDecoder decoder((data.data()), data.size());
   char namebuf[256]{};
   uint8_t byte;
   FuData rdata{};

esphome/components/packet_transport/packet_transport.h

@@ -9,9 +9,8 @@
 #include "esphome/components/binary_sensor/binary_sensor.h"
 #endif
 
-#include <map>
-#include <span>
 #include <vector>
+#include <map>
 
 /**
  * Providing packet encoding functions for exchanging data with a remote host.
@@ -114,7 +113,7 @@ class PacketTransport : public PollingComponent {
   virtual bool should_send() { return true; }
 
   // to be called by child classes when a data packet is received.
-  void process_(std::span<const uint8_t> data);
+  void process_(const std::vector<uint8_t> &data);
   void send_data_(bool all);
   void flush_();
   void add_data_(uint8_t key, const char *id, float data);

esphome/components/udp/__init__.py

@@ -13,7 +13,7 @@ from esphome.components.packet_transport import (
 import esphome.config_validation as cv
 from esphome.const import CONF_DATA, CONF_ID, CONF_PORT, CONF_TRIGGER_ID
 from esphome.core import ID
-from esphome.cpp_generator import MockObj
+from esphome.cpp_generator import literal
 
 CODEOWNERS = ["@clydebarrow"]
 DEPENDENCIES = ["network"]
@@ -23,12 +23,8 @@ MULTI_CONF = True
 udp_ns = cg.esphome_ns.namespace("udp")
 UDPComponent = udp_ns.class_("UDPComponent", cg.Component)
 UDPWriteAction = udp_ns.class_("UDPWriteAction", automation.Action)
-trigger_argname = "data"
-# Listener callback type (non-owning span from UDP component)
-listener_args = cg.std_span.template(cg.uint8.operator("const"))
-listener_argtype = [(listener_args, trigger_argname)]
-# Automation/trigger type (owned vector, safe for deferred actions like delay)
 trigger_args = cg.std_vector.template(cg.uint8)
+trigger_argname = "data"
 trigger_argtype = [(trigger_args, trigger_argname)]
 
 CONF_ADDRESSES = "addresses"
@@ -122,13 +118,7 @@ async def to_code(config):
             trigger_id, trigger_argtype, on_receive
         )
         trigger_lambda = await cg.process_lambda(
-            trigger.trigger(
-                cg.std_vector.template(cg.uint8)(
-                    MockObj(trigger_argname).begin(),
-                    MockObj(trigger_argname).end(),
-                )
-            ),
-            listener_argtype,
+            trigger.trigger(literal(trigger_argname)), trigger_argtype
         )
         cg.add(var.add_listener(trigger_lambda))
         cg.add(var.set_should_listen())

esphome/components/udp/packet_transport/udp_transport.cpp

@@ -12,7 +12,7 @@ bool UDPTransport::should_send() { return network::is_connected(); }
 void UDPTransport::setup() {
   PacketTransport::setup();
   if (!this->providers_.empty() || this->is_encrypted_()) {
-    this->parent_->add_listener([this](std::span<const uint8_t> data) { this->process_(data); });
+    this->parent_->add_listener([this](std::vector<uint8_t> &buf) { this->process_(buf); });
   }
 }
 

esphome/components/udp/udp_component.cpp

@@ -103,8 +103,8 @@ void UDPComponent::setup() {
 }
 
 void UDPComponent::loop() {
+  auto buf = std::vector<uint8_t>(MAX_PACKET_SIZE);
   if (this->should_listen_) {
-    std::array<uint8_t, MAX_PACKET_SIZE> buf;
     for (;;) {
 #if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS)
       auto len = this->listen_socket_->read(buf.data(), buf.size());
@@ -116,9 +116,9 @@ void UDPComponent::loop() {
 #endif
       if (len <= 0)
         break;
-      size_t packet_len = static_cast<size_t>(len);
-      ESP_LOGV(TAG, "Received packet of length %zu", packet_len);
-      this->packet_listeners_.call(std::span<const uint8_t>(buf.data(), packet_len));
+      buf.resize(len);
+      ESP_LOGV(TAG, "Received packet of length %zu", len);
+      this->packet_listeners_.call(buf);
     }
   }
 }

esphome/components/udp/udp_component.h

@@ -10,9 +10,7 @@
 #ifdef USE_SOCKET_IMPL_LWIP_TCP
 #include <WiFiUdp.h>
 #endif
-#include <array>
 #include <initializer_list>
-#include <span>
 #include <vector>
 
 namespace esphome::udp {
@@ -28,7 +26,7 @@ class UDPComponent : public Component {
   void set_broadcast_port(uint16_t port) { this->broadcast_port_ = port; }
   void set_should_broadcast() { this->should_broadcast_ = true; }
   void set_should_listen() { this->should_listen_ = true; }
-  void add_listener(std::function<void(std::span<const uint8_t>)> &&listener) {
+  void add_listener(std::function<void(std::vector<uint8_t> &)> &&listener) {
     this->packet_listeners_.add(std::move(listener));
   }
   void setup() override;
@@ -43,7 +41,7 @@ class UDPComponent : public Component {
   uint16_t broadcast_port_{};
   bool should_broadcast_{};
   bool should_listen_{};
-  CallbackManager<void(std::span<const uint8_t>)> packet_listeners_{};
+  CallbackManager<void(std::vector<uint8_t> &)> packet_listeners_{};
 
 #if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS)
   std::unique_ptr<socket::Socket> broadcast_socket_ = nullptr;

esphome/cpp_types.py

@@ -12,7 +12,6 @@ std_shared_ptr = std_ns.class_("shared_ptr")
 std_string = std_ns.class_("string")
 std_string_ref = std_ns.namespace("string &")
 std_vector = std_ns.class_("vector")
-std_span = std_ns.class_("span")
 uint8 = global_ns.namespace("uint8_t")
 uint16 = global_ns.namespace("uint16_t")
 uint32 = global_ns.namespace("uint32_t")

tests/integration/test_udp.py

@@ -93,34 +93,23 @@ async def udp_listener(port: int = 0) -> AsyncGenerator[tuple[int, UDPReceiver]]
         sock.close()
 
 
-def _get_free_udp_port() -> int:
-    """Get a free UDP port by binding to port 0 and releasing."""
-    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    sock.bind(("127.0.0.1", 0))
-    port = sock.getsockname()[1]
-    sock.close()
-    return port
-
-
 @pytest.mark.asyncio
 async def test_udp_send_receive(
     yaml_config: str,
     run_compiled: RunCompiledFunction,
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
-    """Test UDP component can send and receive messages."""
+    """Test UDP component can send messages with multiple addresses configured."""
+    # Track log lines to verify dump_config output
     log_lines: list[str] = []
-    receive_event = asyncio.Event()
 
     def on_log_line(line: str) -> None:
         log_lines.append(line)
-        if "Received UDP:" in line:
-            receive_event.set()
 
-    async with udp_listener() as (broadcast_port, receiver):
-        listen_port = _get_free_udp_port()
-        config = yaml_config.replace("UDP_LISTEN_PORT_PLACEHOLDER", str(listen_port))
-        config = config.replace("UDP_BROADCAST_PORT_PLACEHOLDER", str(broadcast_port))
+    async with udp_listener() as (udp_port, receiver):
+        # Replace placeholders in the config
+        config = yaml_config.replace("UDP_LISTEN_PORT_PLACEHOLDER", str(udp_port + 1))
+        config = config.replace("UDP_BROADCAST_PORT_PLACEHOLDER", str(udp_port))
 
         async with (
             run_compiled(config, line_callback=on_log_line),
@@ -180,19 +169,3 @@ async def test_udp_send_receive(
             assert "Address: 127.0.0.2" in log_text, (
                 f"Address 127.0.0.2 not found in dump_config. Log: {log_text[-2000:]}"
             )
-
-            # Test receiving a UDP packet (exercises on_receive with std::span)
-            test_payload = b"TEST_RECEIVE_UDP"
-            send_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-            try:
-                send_sock.sendto(test_payload, ("127.0.0.1", listen_port))
-            finally:
-                send_sock.close()
-
-            try:
-                await asyncio.wait_for(receive_event.wait(), timeout=5.0)
-            except TimeoutError:
-                pytest.fail(
-                    f"on_receive did not fire. Expected 'Received UDP:' in logs. "
-                    f"Last log lines: {log_lines[-20:]}"
-                )