Merge remote-tracking branch 'origin/dev' into apply-bytes

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: J. Nick Koston <nick@home-assistant.io>

AGENTS.md

@@ -205,8 +205,9 @@ file does, and it is the authority when they disagree. The most useful starting
         for a table of structs copy the whole entry out and return the copy, never a pointer into flash.
         On other platforms they are a plain `memcpy` and dereference, so no `#ifdef` is needed. Declare log tags with
         `ESPHOME_LOG_TAG(TAG, "name")`, pass status text as `LOG_STR`, e.g.
-        `this->mark_failed(LOG_STR(ESP_LOG_MSG_COMM_FAIL))`, and give an action whose payload is a byte array
-        a `TEMPLATABLE_BYTES` field so a static code stays in flash.
+        `this->mark_failed(LOG_STR(ESP_LOG_MSG_COMM_FAIL))`. An action that only forwards a byte payload is
+        registered with `automation.ApplyBytes` (see Actions below), which keeps a constant payload in flash;
+        `TEMPLATABLE_BYTES` is for a `play()` that does more with the bytes.
     *   **Fixed sets of names** (status words, model names): a `PROGMEM_STRING_TABLE` (`esphome/core/progmem.h`),
         logged with `get_log_str` and `LOG_STR_ARG`, copied out with `ESPHOME_strncpy_P` before `publish_state`.
         A `dump_config` helper that picks a name returns `const LogString *` from `LOG_STR(...)`.
@@ -604,7 +605,9 @@ file does, and it is the authority when they disagree. The most useful starting
         )
         ```
         The `ApplyField`, `ApplyCall` and `register_apply_action` docstrings in `esphome/automation.py` cover
-        the rest; `cover.control` and `cover.template.publish` are in-tree examples. `TEMPLATABLE_VALUE` with
+        the rest; `cover.control` and `cover.template.publish` are in-tree examples. An action whose input is a
+        byte payload passes `automation.ApplyBytes(CONF_DATA, "send({})")` in the same call; a constant payload
+        stays in flash and a lambda's vector is forwarded as data and size. `TEMPLATABLE_VALUE` with
         `cg.templatable` stays for actions whose `play()` has real logic beyond forwarding values.
 
     *   **Conditions:**

esphome/automation.py

@@ -59,10 +59,15 @@ async def templatable_bytes(
     if cg.is_template(value):
         fn = await cg.templatable(value, args, cg.std_vector.template(cg.uint8))
         cg.add(set_template(fn))
-    elif len(value) > 0xFFFF:
-        raise EsphomeError(f"Byte payload is {len(value)} bytes; the maximum is 65535")
     else:
-        cg.add(set_static(progmem_bytes(table_name, value), len(value)))
+        cg.add(set_static(*_static_bytes(table_name, value)))
+
+
+def _static_bytes(table_name: str, value: bytes | list[int]) -> tuple[MockObj, int]:
+    """The shared PROGMEM table for a constant payload and its length."""
+    if (size := len(value)) > 0xFFFF:
+        raise EsphomeError(f"Byte payload is {size} bytes; the maximum is 65535")
+    return progmem_bytes(table_name, value), size
 
 
 def maybe_simple_id(*validators):
@@ -356,6 +361,33 @@ class ApplyField:
         return ApplyCall(target, ((self.conf_key, self.type_, self.const_fn),))
 
 
+@dataclass(frozen=True)
+class ApplyBytes:
+    """A byte payload key forwarded as ``target``, whose ``{}`` stands for ``data, len``.
+
+    A constant payload is a shared PROGMEM table passed as is; on ESP8266 it is first copied
+    out, to the stack up to ``stack`` bytes and to the heap beyond, unless ``progmem_target``
+    names a call that reads flash itself. A lambda's ``std::vector<uint8_t>`` is passed as
+    its data and size. The template may name the parent as ``{parent}``. An absent key
+    emits nothing.
+    """
+
+    conf_key: str | tuple[str, ...]
+    target: str
+    stack: int = 64
+    progmem_target: str | None = None
+
+    def __post_init__(self) -> None:
+        for target in (self.target, self.progmem_target):
+            if target is None:
+                continue
+            fields = [
+                f for _, f, _, _ in string.Formatter().parse(target) if f is not None
+            ]
+            if fields.count("") != 1 or any(f not in ("", "parent") for f in fields):
+                raise ValueError(f"apply bytes target {target!r} needs one {{}}")
+
+
 def _config_lookup(config: ConfigType, key: str | tuple[str, ...]) -> Any:
     if isinstance(key, str):
         return config.get(key)
@@ -488,7 +520,7 @@ def _apply_values(config: ConfigType, members: list[tuple[Any, Any, Any]]) -> li
 def register_apply_action(
     name: str,
     schema: cv.Schema,
-    *fields: ApplyField | ApplyCall,
+    *fields: ApplyField | ApplyCall | ApplyBytes,
     call: str | None = None,
     id_key: str = CONF_ID,
 ) -> None:
@@ -504,7 +536,7 @@ def register_apply_action(
 def apply_action_builder(
     name: str,
     schema: cv.Schema,
-    *fields: ApplyField | ApplyCall,
+    *fields: ApplyField | ApplyCall | ApplyBytes,
     call: str | None = None,
     id_key: str = CONF_ID,
 ) -> Callable[..., Awaitable[MockObj]]:
@@ -517,17 +549,20 @@ def apply_action_builder(
     and ``apply_call.perform()`` is appended.
     """
     # An action stores the value, so a std::string constant stays in flash on ESP8266.
-    statements_spec = [
-        (
-            c.target,
-            _default_std_string(c.members, flash_string),
-        )
-        for c in (f if isinstance(f, ApplyCall) else f.call() for f in fields)
-    ]
     _check_key_in_schema(name, schema, id_key)
-    for _, members in statements_spec:
+    statements_spec: list[ApplyBytes | tuple[str, list[tuple[Any, Any, Any]]]] = []
+    for f in fields:
+        if isinstance(f, ApplyBytes):
+            _check_key_in_schema(name, schema, f.conf_key)
+            statements_spec.append(f)
+            continue
+        c = f if isinstance(f, ApplyCall) else f.call()
+        members = _default_std_string(c.members, flash_string)
         for conf_key, _, _ in members:
             _check_key_in_schema(name, schema, conf_key)
+        statements_spec.append((c.target, members))
+    # Named like the generated function, so it cannot clash with an SDK symbol
+    table_name = f"esphome__{name.replace('.', '_')}__bytes"
 
     async def builder(
         config: ConfigType,
@@ -539,7 +574,15 @@ def apply_action_builder(
         lambda_args = _apply_lambda_args(args)
         receiver = "apply_call." if call else f"{parent}->"
         statements: list[str] = []
-        for target, members in statements_spec:
+        for spec in statements_spec:
+            if isinstance(spec, ApplyBytes):
+                statements.extend(
+                    await _render_bytes(
+                        spec, config, receiver, parent, lambda_args, table_name
+                    )
+                )
+                continue
+            target, members = spec
             values = _apply_values(config, members)
             if not _apply_call_active(members, values):
                 continue
@@ -560,6 +603,56 @@ def apply_action_builder(
     return builder
 
 
+async def _render_bytes(
+    spec: ApplyBytes,
+    config: ConfigType,
+    receiver: str,
+    parent: str,
+    lambda_args: TemplateArgsType,
+    table_name: str,
+) -> list[str]:
+    """The statements for one ``ApplyBytes``: what ``TemplatableBytes::visit`` does, inlined."""
+    value = _config_lookup(config, spec.conf_key)
+    if value is None:
+        return []
+
+    # The temporary carries the generated-name prefix so a trigger argument cannot shadow it
+    def send(target: str, data: str) -> str:
+        return f"{receiver}{target.format(data, parent=parent)};"
+
+    if isinstance(value, Lambda):
+        inner = await cg.process_lambda(
+            value, lambda_args, return_type=cg.std_vector.template(cg.uint8)
+        )
+        return [
+            "{",
+            f"  const std::vector<uint8_t> esphome__bytes = ({call_lambda(inner)});",
+            f"  {send(spec.target, 'esphome__bytes.data(), esphome__bytes.size()')}",
+            "}",
+        ]
+    table, size = _static_bytes(table_name, value)
+    # Qualified so a trigger argument cannot shadow a reused table; an empty payload is nullptr
+    args = f"::{table}, {size}" if size else f"{table}, {size}"
+    if CORE.is_esp8266 and spec.progmem_target is not None:
+        return [send(spec.progmem_target, args)]
+    if not CORE.is_esp8266:
+        return [send(spec.target, args)]
+    # A PROGMEM table must be read with aligned loads on ESP8266, so copy it out first;
+    # the length is known here, so the copy is an exact array, on the heap past ``stack``.
+    # An empty payload still gets a one byte array: the raw lwIP socket rejects a null buffer.
+    if size <= spec.stack:
+        buffer, data = f"uint8_t esphome__bytes[{max(size, 1)}];", "esphome__bytes"
+    else:
+        buffer, data = (
+            f"auto esphome__bytes = std::make_unique<uint8_t[]>({size});",
+            "esphome__bytes.get()",
+        )
+    statements = ["{", f"  {buffer}"]
+    if size:
+        statements.append(f"  esphome::progmem_memcpy({data}, {args});")
+    return [*statements, f"  {send(spec.target, f'{data}, {size}')}", "}"]
+
+
 def _apply_call_active(members: list[tuple[Any, Any, Any]], values: list[Any]) -> bool:
     """An ``ApplyCall`` is emitted unless it has keys and none of them is set."""
     return not members or any(value is not None for value in values)

esphome/components/api/__init__.py

@@ -7,7 +7,7 @@ from esphome import automation
 import esphome.codegen as cg
 from esphome.components.const import CONF_DESCRIPTION, CONF_HOST
 from esphome.components.logger import request_log_listener
-from esphome.components.network import DOMAIN as NETWORK_DOMAIN
+from esphome.components.network import require_ipv4, require_ipv6
 
 # ENCRYPTION_SCHEMA and validate_encryption_key are re-exported for external
 # components and downstream consumers that import them from api
@@ -28,7 +28,6 @@ from esphome.const import (
     CONF_DATA,
     CONF_DATA_TEMPLATE,
     CONF_DELAY,
-    CONF_ENABLE_IPV6,
     CONF_ENCRYPTION,
     CONF_EVENT,
     CONF_ID,
@@ -52,7 +51,6 @@ from esphome.const import (
 )
 from esphome.core import CORE, ID, CoroPriority, EsphomeError, coroutine_with_priority
 from esphome.cpp_generator import Expression, MockObj, TemplateArgsType
-import esphome.final_validate as fv
 from esphome.helpers import fnv1_hash
 from esphome.schema_extractors import SCHEMA_EXTRACT, schema_extractor
 from esphome.types import ConfigFragmentType, ConfigType
@@ -333,7 +331,10 @@ def _validate_outgoing_host(value: str) -> IPv4Address | IPv6Address:
                 "give the address without the '%' part"
             )
         if (mapped := address.ipv4_mapped) is not None:
-            return mapped
+            address = mapped
+    # The build must carry the host's address family for the dial to work
+    require = require_ipv6 if isinstance(address, IPv6Address) else require_ipv4
+    require("api.outgoing_connection")(value)
     return address
 
 
@@ -479,31 +480,12 @@ def _action_strings(conf: ConfigType, has_metadata: bool) -> list[str | None]:
     return strings
 
 
-def _validate_outgoing_host_ipv6(config: ConfigType) -> ConfigType:
-    """An IPv6 host can never be parsed, so never dialed, without IPv6."""
-    if (
-        (outgoing := config.get(CONF_OUTGOING_CONNECTION)) is None
-        or (host := outgoing.get(CONF_HOST)) is None
-        or host.version != 6
-    ):
-        return config
-    network_conf = fv.full_config.get().get(NETWORK_DOMAIN) or {}
-    if not network_conf.get(CONF_ENABLE_IPV6):
-        raise cv.Invalid(
-            "outgoing_connection host is an IPv6 address but IPv6 is not "
-            "enabled; set 'network: enable_ipv6: true'",
-            path=[CONF_OUTGOING_CONNECTION, CONF_HOST],
-        )
-    return config
-
-
 def _validate_wizard(config: ConfigType) -> ConfigType:
     wizard.final_validate(config)
     return config
 
 
 FINAL_VALIDATE_SCHEMA = cv.All(
-    _validate_outgoing_host_ipv6,
     _validate_wizard,
 )
 

esphome/components/api/api_outgoing_connection.cpp

@@ -38,12 +38,14 @@ static bool peer_to_target(APIConnection *conn, SavedOutgoingTarget &target) {
     const auto *bytes = reinterpret_cast<const uint8_t *>(&addr6->sin6_addr);
     uint32_t prefix[3];
     memcpy(prefix, bytes, sizeof(prefix));
+#if USE_NETWORK_IPV4
     // A dual-stack listener reports an IPv4 peer as ::ffff:a.b.c.d
     if (prefix[0] == 0 && prefix[1] == 0 && prefix[2] == htonl(0xFFFFUL)) {
       target.family = AF_INET;
       memcpy(target.addr, bytes + sizeof(prefix), sizeof(struct in_addr));
       return true;
     }
+#endif
     // A link-local target is only reachable through the interface it came in
     // on. Device platforms number interfaces from one; a host build can hand
     // out an index too large to store, and a truncated one dials the wrong
@@ -57,6 +59,7 @@ static bool peer_to_target(APIConnection *conn, SavedOutgoingTarget &target) {
     return true;
   }
 #endif
+#if USE_NETWORK_IPV4
   if (family != AF_INET) {
     return false;
   }
@@ -64,6 +67,9 @@ static bool peer_to_target(APIConnection *conn, SavedOutgoingTarget &target) {
   target.family = AF_INET;
   memcpy(target.addr, &addr4->sin_addr, sizeof(addr4->sin_addr));
   return true;
+#else
+  return false;
+#endif
 }
 #endif
 
@@ -85,6 +91,7 @@ socklen_t OutgoingConnectionManager::target_sockaddr_(struct sockaddr_storage *a
     return sizeof(*addr6);
   }
 #endif
+#if USE_NETWORK_IPV4
   if (this->saved_.family != AF_INET) {
     return 0;
   }
@@ -94,6 +101,9 @@ socklen_t OutgoingConnectionManager::target_sockaddr_(struct sockaddr_storage *a
   addr4->sin_port = htons(API_OUTGOING_CONNECTION_PORT);
   memcpy(&addr4->sin_addr, this->saved_.addr, sizeof(addr4->sin_addr));
   return sizeof(*addr4);
+#else
+  return 0;
+#endif
 #endif
 }
 

esphome/components/e131/__init__.py

@@ -1,5 +1,5 @@
 import esphome.codegen as cg
-from esphome.components import socket
+from esphome.components import network, socket
 from esphome.components.light.effects import register_addressable_effect
 from esphome.components.light.types import AddressableLightEffect
 import esphome.config_validation as cv
@@ -38,6 +38,7 @@ CONFIG_SCHEMA = cv.All(
             ),
         }
     ).extend(cv.COMPONENT_SCHEMA),
+    network.require_ipv4("e131"),
     socket.consume_sockets(1, "e131", socket.SocketType.UDP),
 )
 

esphome/components/esp32/__init__.py

@@ -711,6 +711,15 @@ def set_core_data(config):
     return config
 
 
+def _require_ipv4_on_arduino(config: ConfigType) -> ConfigType:
+    """The Arduino core's IPAddress.cpp does not compile with IPv4 disabled."""
+    if CORE.using_arduino:
+        from esphome.components import network
+
+        network.require_ipv4(PLATFORM_ESP32)(config)
+    return config
+
+
 def get_esp32_variant(core_obj=None):
     return (core_obj or CORE).data[KEY_ESP32][KEY_VARIANT]
 
@@ -2494,6 +2503,7 @@ CONFIG_SCHEMA = cv.All(
     _set_default_framework,
     _check_versions,
     set_core_data,
+    _require_ipv4_on_arduino,
     cv.has_at_least_one_key(CONF_BOARD, CONF_VARIANT),
 )
 

esphome/components/esp8266/__init__.py

@@ -7,6 +7,7 @@ from typing import Any
 
 from esphome.build_helpers.native import native_backend
 import esphome.codegen as cg
+from esphome.components import network
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_BOARD,
@@ -369,6 +370,7 @@ CONFIG_SCHEMA = cv.All(
     _warn_platformio_toolchain,
     _validate_native_toolchain,
     set_core_data,
+    network.require_ipv4(PLATFORM_ESP8266),
 )
 
 

esphome/components/ethernet/__init__.py

@@ -8,6 +8,7 @@ from esphome.components.network import (
     get_network_priority,
     get_priority_interfaces_from_full_config,
     ip_address_literal,
+    require_ipv4,
 )
 from esphome.config_helpers import (
     filter_source_files_from_defines,
@@ -22,6 +23,7 @@ from esphome.const import (
     CONF_DNS2,
     CONF_DOMAIN,
     CONF_ENABLE_ON_BOOT,
+    CONF_ETHERNET,
     CONF_GATEWAY,
     CONF_ID,
     CONF_INTERRUPT_PIN,
@@ -556,6 +558,7 @@ CONFIG_SCHEMA = cv.All(
         upper=True,
     ),
     _validate,
+    require_ipv4(CONF_ETHERNET),
 )
 
 

esphome/components/host/__init__.py

@@ -1,6 +1,7 @@
 import sys
 
 import esphome.codegen as cg
+from esphome.components import network
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_MAC_ADDRESS,
@@ -44,6 +45,7 @@ CONFIG_SCHEMA = cv.All(
     # is no PlatformIO backend, so a CLI --toolchain must name this one
     cv.resolve_toolchain("host", (Toolchain.HOST,), Toolchain.HOST),
     set_core_data,
+    network.require_ipv4(PLATFORM_HOST),
 )
 
 

esphome/components/i2s_audio/speaker/__init__.py

@@ -107,13 +107,10 @@ def _set_stream_limits(config: ConfigType) -> ConfigType:
         )(config)
         return config
 
-    # The original ESP32 cannot lay out sub-16-bit slots that match ESPHome's packed audio, so the smallest
-    # stream it accepts is 16-bit (see start_i2s_driver); the other variants handle 8-bit.
-    min_bits_per_sample = 16 if esp32.get_esp32_variant() == esp32.VARIANT_ESP32 else 8
-
-    # The configured bits per sample sets the I2S slot width, but the speaker narrows wider streams down to it
-    # in place before clocking them out (see start_i2s_driver). Advertise up to 32-bit so those wider streams
-    # are accepted rather than forcing an upstream conversion.
+    # The configured bits per sample sets the I2S slot width, but the speaker converts streams of other widths
+    # before clocking them out (see start_i2s_driver), widening them on the original ESP32. Advertise 8- to 32-bit
+    # so those streams are accepted rather than forcing an upstream conversion.
+    min_bits_per_sample = 8
     max_bits_per_sample = 32
 
     if config[CONF_I2S_MODE] == CONF_PRIMARY:
@@ -153,10 +150,12 @@ def _validate_esp32_variant(config: ConfigType) -> ConfigType:
         if variant not in INTERNAL_DAC_VARIANTS:
             raise cv.Invalid(f"{variant} does not have an internal DAC")
     elif variant == esp32.VARIANT_ESP32 and config[CONF_BITS_PER_SAMPLE] == 8:
-        # The original ESP32 I2S peripheral packs each sample into a whole number of 16-bit words, so an
-        # 8-bit slot does not line up with ESPHome's tightly packed audio (see start_i2s_driver). Reject it
-        # at config time rather than emitting corrupted output at runtime.
-        raise cv.Invalid("8-bit audio is not supported on the original ESP32")
+        # An 8-bit slot is rejected on the original ESP32: no known DAC needs one, and it is untested there.
+        # 8-bit streams still work; they are widened to the 16- or 32-bit slot (see start_i2s_driver).
+        raise cv.Invalid(
+            f"{CONF_BITS_PER_SAMPLE} cannot be 8bit on the original ESP32",
+            path=[CONF_BITS_PER_SAMPLE],
+        )
     return config
 
 

esphome/components/i2s_audio/speaker/i2s_audio_speaker.h

@@ -208,8 +208,8 @@ class I2SAudioSpeakerBase : public I2SAudioOut, public speaker::Speaker, public
 
   audio::AudioStreamInfo current_stream_info_;  // Format of the audio in the ring buffer (the I2S input)
   // Format actually clocked out of the I2S peripheral. Same channel count and sample rate as
-  // current_stream_info_, but the bits per sample may be narrower when the incoming stream is wider than
-  // the speaker's configured slot bit width. Set by start_i2s_driver before the speaker task starts.
+  // current_stream_info_, but the bits per sample may be converted to the speaker's configured slot bit width.
+  // Set by start_i2s_driver before the speaker task starts.
   audio::AudioStreamInfo output_stream_info_;
 
   gpio_num_t dout_pin_;

esphome/components/i2s_audio/speaker/i2s_audio_speaker_standard.cpp

@@ -38,6 +38,11 @@ static constexpr size_t I2S_EVENT_QUEUE_COUNT = DMA_BUFFERS_COUNT * 2;
 // DMA_BUFFER_DURATION_MS, so a multiple of that gives plenty of slack against scheduling jitter
 // without masking real failures.
 static constexpr TickType_t WRITE_TIMEOUT_TICKS = pdMS_TO_TICKS(DMA_BUFFER_DURATION_MS * (DMA_BUFFERS_COUNT + 1));
+#ifdef USE_ESP32_VARIANT_ESP32
+// Stack buffer that widened audio passes through on the original ESP32. Small enough for the speaker task's stack,
+// large enough that the extra i2s_channel_write calls cost little.
+static constexpr size_t WIDEN_BUFFER_BYTES = 512;
+#endif
 
 // Requested frames per DMA buffer for the given stream, clamped so the byte size stays within the ESP-IDF
 // maximum DMA descriptor size. This is only the value handed to the channel config: ESP-IDF may still adjust
@@ -66,9 +71,8 @@ void I2SAudioSpeaker::dump_config() {
   }
   ESP_LOGCONFIG(TAG, "  Communication format: %s", fmt_str);
   if (this->slot_bit_width_ != I2S_SLOT_BIT_WIDTH_AUTO) {
-    // The width of each I2S slot. It is also the narrowing ceiling: streams wider than this are narrowed to
-    // it. A stream narrower than the slot is left at its own width and clocked into the wider slot, so this
-    // is not necessarily the sample data width (which depends on the incoming stream).
+    // The width of each I2S slot. Wider streams are narrowed to it. Narrower streams are padded into the slot by
+    // the peripheral, except on the original ESP32, where they are widened to it.
     ESP_LOGCONFIG(TAG, "  Slot bit width: %u", (unsigned) static_cast<uint32_t>(this->slot_bit_width_));
   }
 }
@@ -92,11 +96,14 @@ void I2SAudioSpeaker::run_speaker_task() {
   const size_t ring_buffer_size =
       (this->current_stream_info_.ms_to_bytes(ring_buffer_duration) / source_alignment) * source_alignment;
 
-  // Per-frame byte widths and whether the task must narrow the bit depth before writing to the I2S peripheral.
+  // Per-frame byte widths and whether the task must convert the bit depth before writing to the I2S peripheral.
   const uint8_t channels = this->current_stream_info_.get_channels();
   const uint8_t input_bytes_per_sample = this->current_stream_info_.get_bits_per_sample() / 8;
   const uint8_t output_bytes_per_sample = this->output_stream_info_.get_bits_per_sample() / 8;
-  const bool narrowing = input_bytes_per_sample != output_bytes_per_sample;
+  const bool narrowing = input_bytes_per_sample > output_bytes_per_sample;
+#ifdef USE_ESP32_VARIANT_ESP32
+  const bool widening = input_bytes_per_sample < output_bytes_per_sample;
+#endif
 
   // ESP-IDF may allocate smaller (or cache-line-rounded) DMA buffers than dma_buffer_frames() requested: it
   // clamps each descriptor to the max DMA descriptor size and, on targets that route internal memory through
@@ -323,20 +330,26 @@ void I2SAudioSpeaker::run_speaker_task() {
           const size_t output_bytes = this->output_stream_info_.frames_to_bytes(frames_to_write);
 
           uint8_t *chunk = audio_source->mutable_data();
-          if (narrowing) {
-            // Narrow the bit depth in place: output exactly aliases input with the same channel count and a
-            // smaller width, which copy_frames handles as a single forward pass. Only the frames about to be
-            // consumed are overwritten, so any unprocessed tail stays intact for the next iteration.
-            esp_audio_libs::pcm_convert::copy_frames(chunk, chunk, input_bytes_per_sample, channels,
-                                                     output_bytes_per_sample, channels, frames_to_write);
-          }
-          this->swap_esp32_mono_samples_(chunk, output_bytes);
-
           size_t bw = 0;
-          i2s_channel_write(this->tx_handle_, chunk, output_bytes, &bw, WRITE_TIMEOUT_TICKS);
+#ifdef USE_ESP32_VARIANT_ESP32
+          if (widening) {
+            bw = this->write_widened_(chunk, frames_to_write);
+          } else
+#endif
+          {
+            if (narrowing) {
+              // Narrow the bit depth in place: output exactly aliases input with the same channel count and a
+              // smaller width, which copy_frames handles as a single forward pass. Only the frames about to be
+              // consumed are overwritten, so any unprocessed tail stays intact for the next iteration.
+              esp_audio_libs::pcm_convert::copy_frames(chunk, chunk, input_bytes_per_sample, channels,
+                                                       output_bytes_per_sample, channels, frames_to_write);
+            }
+            this->swap_esp32_mono_samples_(chunk, output_bytes);
+            i2s_channel_write(this->tx_handle_, chunk, output_bytes, &bw, WRITE_TIMEOUT_TICKS);
+          }
           if (bw != output_bytes) {
             // A short write breaks DMA descriptor alignment for every subsequent event. Drop the chunk rather
-            // than retry it: it was already narrowed in place.
+            // than retry it: it may already have been converted in place.
             ESP_LOGE(TAG, "Partial DMA write (%u of %u bytes), resyncing DMA lockstep", (unsigned) bw,
                      (unsigned) output_bytes);
             audio_source->consume(input_bytes);
@@ -416,26 +429,26 @@ esp_err_t I2SAudioSpeaker::start_i2s_driver(audio::AudioStreamInfo &audio_stream
     return ESP_ERR_NOT_SUPPORTED;
   }
 
-  // When the stream is wider than the configured slot bit width, the speaker task narrows each frame in place
-  // before handing it to the I2S peripheral. Compute the output format here so the driver, DMA buffers, and
-  // the task's conversion all agree on the clocked-out width. A stream no wider than the slot width is passed
-  // through unchanged (the slot may still be wider than the data, the existing behavior).
+  // Compute the output width so the driver, DMA buffers, and the task's conversion all agree. A stream wider than
+  // the configured slot is narrowed to it, and a narrower stream keeps its width for the peripheral to pad into the
+  // slot. The original ESP32 cannot clock a slot wider than its data, so there every stream is converted to the
+  // slot's memory width instead.
   uint8_t output_bits_per_sample = audio_stream_info.get_bits_per_sample();
   if (this->slot_bit_width_ != I2S_SLOT_BIT_WIDTH_AUTO) {
+#ifdef USE_ESP32_VARIANT_ESP32
+    output_bits_per_sample = this->slot_memory_bits_();
+#else
     const uint8_t configured_bits = static_cast<uint8_t>(this->slot_bit_width_);
     if (output_bits_per_sample > configured_bits) {
       output_bits_per_sample = configured_bits;
     }
+#endif
   }
   this->output_stream_info_ = audio::AudioStreamInfo(output_bits_per_sample, audio_stream_info.get_channels(),
                                                      audio_stream_info.get_sample_rate());
 
 #ifdef USE_ESP32_VARIANT_ESP32
-  // The original ESP32 I2S peripheral stores each sample in a whole number of 16-bit words (a 24-bit sample
-  // occupies 4 bytes in the DMA buffer, an 8-bit sample 2 bytes), but ESPHome's audio pipeline packs samples
-  // tightly (3 bytes for 24-bit, 1 for 8-bit). The two layouts only line up when the bit depth is a multiple
-  // of 16. The check is on the output width since that is what reaches the peripheral; a wider input is fine
-  // as long as it narrows to a 16- or 32-bit slot.
+  // Only a slot left at AUTO can reach this; every configured slot is converted to 16- or 32-bit samples above
   if (output_bits_per_sample % 16 != 0) {
     ESP_LOGE(TAG, "ESP32 supports only 16- or 32-bit output, got %u-bit", (unsigned) output_bits_per_sample);
     return ESP_ERR_NOT_SUPPORTED;
@@ -487,8 +500,8 @@ esp_err_t I2SAudioSpeaker::start_i2s_driver(audio::AudioStreamInfo &audio_stream
 
 #ifdef USE_I2S_AUDIO_FULL_DUPLEX
 audio::AudioStreamInfo I2SAudioSpeaker::full_duplex_stream_info_() const {
-  return audio::AudioStreamInfo(static_cast<uint8_t>(this->slot_bit_width_),
-                                this->slot_mode_ == I2S_SLOT_MODE_STEREO ? 2 : 1, this->sample_rate_);
+  return audio::AudioStreamInfo(this->slot_memory_bits_(), this->slot_mode_ == I2S_SLOT_MODE_STEREO ? 2 : 1,
+                                this->sample_rate_);
 }
 
 void I2SAudioSpeaker::build_full_duplex_config(i2s_chan_config_t &chan_cfg, i2s_std_config_t &std_cfg) {
@@ -530,8 +543,14 @@ void I2SAudioSpeaker::build_i2s_config_(const audio::AudioStreamInfo &output_str
     slot_mask = I2S_STD_SLOT_BOTH;
   }
 
-  // Configure the data bit width from the output (post-narrowing) format, which is what is clocked out.
-  const i2s_data_bit_width_t data_bit_width = (i2s_data_bit_width_t) output_stream_info.get_bits_per_sample();
+  // Configure the data bit width from the output (post-conversion) format, which is what is clocked out.
+  i2s_data_bit_width_t data_bit_width = (i2s_data_bit_width_t) output_stream_info.get_bits_per_sample();
+#ifdef USE_ESP32_VARIANT_ESP32
+  // The original ESP32 clocks data at the slot width; a 24-bit slot sends the top 24 bits of each 32-bit sample
+  if (this->slot_bit_width_ != I2S_SLOT_BIT_WIDTH_AUTO) {
+    data_bit_width = (i2s_data_bit_width_t) this->slot_bit_width_;
+  }
+#endif
   i2s_std_slot_config_t slot_cfg;
   switch (this->i2s_comm_fmt_) {
     case I2SCommFmt::PCM:
@@ -545,23 +564,10 @@ void I2SAudioSpeaker::build_i2s_config_(const audio::AudioStreamInfo &output_str
       break;
   }
 
-#ifdef USE_ESP32_VARIANT_ESP32
-  // There seems to be a bug on the ESP32 (non-variant) platform where setting the slot bit width higher than the
-  // bits per sample causes the audio to play too fast. Setting the ws_width to the configured slot bit width seems
-  // to make it play at the correct speed while sending more bits per slot.
-  if (this->slot_bit_width_ != I2S_SLOT_BIT_WIDTH_AUTO) {
-    uint32_t configured_bit_width = static_cast<uint32_t>(this->slot_bit_width_);
-    slot_cfg.ws_width = configured_bit_width;
-    if (configured_bit_width > 16) {
-      slot_cfg.msb_right = false;
-    }
-  }
-#else
   slot_cfg.slot_bit_width = this->slot_bit_width_;
   if (this->slot_bit_width_ != I2S_SLOT_BIT_WIDTH_AUTO) {
     slot_cfg.ws_width = static_cast<uint32_t>(this->slot_bit_width_);
   }
-#endif  // USE_ESP32_VARIANT_ESP32
   slot_cfg.slot_mask = slot_mask;
 
   i2s_std_gpio_config_t gpio_cfg = this->parent_->get_pin_config();
@@ -574,6 +580,45 @@ void I2SAudioSpeaker::build_i2s_config_(const audio::AudioStreamInfo &output_str
   };
 }
 
+uint8_t I2SAudioSpeaker::slot_memory_bits_() const {
+#ifdef USE_ESP32_VARIANT_ESP32
+  // The original ESP32 stores a 24-bit slot's samples in 32-bit words
+  if (this->slot_bit_width_ == I2S_SLOT_BIT_WIDTH_24BIT) {
+    return 32;
+  }
+#endif
+  return static_cast<uint8_t>(this->slot_bit_width_);
+}
+
+#ifdef USE_ESP32_VARIANT_ESP32
+size_t I2SAudioSpeaker::write_widened_(const uint8_t *input, uint32_t frames) {
+  alignas(4) uint8_t widened[WIDEN_BUFFER_BYTES];
+  const uint8_t channels = this->current_stream_info_.get_channels();
+  const uint8_t input_bytes_per_sample = this->current_stream_info_.get_bits_per_sample() / 8;
+  const uint8_t output_bytes_per_sample = this->output_stream_info_.get_bits_per_sample() / 8;
+  const uint32_t frames_per_pass = WIDEN_BUFFER_BYTES / this->output_stream_info_.frames_to_bytes(1);
+
+  size_t bytes_written = 0;
+  while (frames > 0) {
+    const uint32_t pass_frames = std::min(frames, frames_per_pass);
+    esp_audio_libs::pcm_convert::copy_frames(input, widened, input_bytes_per_sample, channels, output_bytes_per_sample,
+                                             channels, pass_frames);
+    const size_t pass_bytes = this->output_stream_info_.frames_to_bytes(pass_frames);
+    // 8-bit mono widened to 16-bit mono needs the pair swap; passes hold whole pairs since writes are pair-aligned
+    this->swap_esp32_mono_samples_(widened, pass_bytes);
+    size_t bw = 0;
+    i2s_channel_write(this->tx_handle_, widened, pass_bytes, &bw, WRITE_TIMEOUT_TICKS);
+    bytes_written += bw;
+    if (bw != pass_bytes) {
+      break;
+    }
+    input += this->current_stream_info_.frames_to_bytes(pass_frames);
+    frames -= pass_frames;
+  }
+  return bytes_written;
+}
+#endif  // USE_ESP32_VARIANT_ESP32
+
 }  // namespace esphome::i2s_audio
 
 #endif  // USE_ESP32

esphome/components/i2s_audio/speaker/i2s_audio_speaker_standard.h

@@ -38,6 +38,17 @@ class I2SAudioSpeaker final : public I2SAudioSpeakerBase {
   audio::AudioStreamInfo full_duplex_stream_info_() const;
 #endif
 
+  /// @brief Bits per sample that the configured slot takes in DMA memory.
+  uint8_t slot_memory_bits_() const;
+
+#ifdef USE_ESP32_VARIANT_ESP32
+  /// @brief Widens frames to the output bit depth through a small stack buffer and writes them to the peripheral.
+  /// @param input Frames in the incoming stream format
+  /// @param frames Number of frames to widen and write
+  /// @return Bytes written, which is short of the widened size if a write timed out
+  size_t write_widened_(const uint8_t *input, uint32_t frames);
+#endif
+
   /// @brief Frame multiple for DMA buffers and writes; 2 keeps ESP32 mono sample pairs word-aligned.
   static uint32_t dma_frame_multiple(const audio::AudioStreamInfo &stream_info) {
     return esp32_mono_swap_needed(stream_info) ? 2 : 1;

esphome/components/libretiny/__init__.py

@@ -3,6 +3,7 @@ import logging
 from pathlib import Path
 
 import esphome.codegen as cg
+from esphome.components.const import CONF_LIBRETINY
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_BOARD,
@@ -304,6 +305,15 @@ FRAMEWORK_SCHEMA = cv.All(
 
 CONFIG_SCHEMA = cv.All(_notify_old_style, cv.require_platformio_toolchain("LibreTiny"))
 
+
+def _require_ipv4(config: dict) -> dict:
+    """Declares that LibreTiny builds need IPv4."""
+    # Imported here: logger imports this module, so a module level import would load network for every config
+    from esphome.components import network
+
+    return network.require_ipv4(CONF_LIBRETINY)(config)
+
+
 BASE_SCHEMA = cv.Schema(
     {
         cv.GenerateID(): cv.declare_id(LTComponent),
@@ -318,6 +328,7 @@ BASE_SCHEMA = cv.Schema(
 BASE_SCHEMA.add_extra(_detect_variant)
 BASE_SCHEMA.add_extra(cv.require_platformio_toolchain("LibreTiny"))
 BASE_SCHEMA.add_extra(_update_core_data)
+BASE_SCHEMA.add_extra(_require_ipv4)
 
 
 def _configure_lwip(config: dict) -> None:

esphome/components/mqtt/__init__.py

@@ -1,6 +1,6 @@
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import logger, socket
+from esphome.components import logger, network, socket
 from esphome.components.esp32 import (
     add_idf_component,
     idf_version,
@@ -35,6 +35,7 @@ from esphome.const import (
     CONF_KEEPALIVE,
     CONF_LEVEL,
     CONF_LOG_TOPIC,
+    CONF_MQTT,
     CONF_ON_CONNECT,
     CONF_ON_DISCONNECT,
     CONF_ON_JSON_MESSAGE,
@@ -365,6 +366,7 @@ CONFIG_SCHEMA = cv.All(
         ]
     ),
     _consume_mqtt_sockets,
+    network.require_ipv4(CONF_MQTT),
 )
 
 

esphome/components/network/__init__.py

@@ -1,3 +1,5 @@
+from collections.abc import Callable
+from dataclasses import dataclass, field
 import ipaddress
 import logging
 from typing import Any
@@ -51,6 +53,16 @@ TCPIP_RECVMBOX_OPTIMIZED = 64
 TCP_SEND_BUFFER_MIN = 2880
 TCP_SEND_BUFFER_MAX = 65535
 
+CONF_ENABLE_IPV4 = "enable_ipv4"
+
+
+@dataclass
+class NetworkData:
+    require_ipv4: set[str] = field(default_factory=set)
+    require_ipv6: set[str] = field(default_factory=set)
+    request_ipv4_off: set[str] = field(default_factory=set)
+
+
 # Network priority tracking infrastructure
 # Components can query this to determine their relative setup priority.
 # CORE.data[KEY_NETWORK_PRIORITY] is a list of dicts of the form
@@ -264,11 +276,35 @@ def has_high_performance_networking() -> bool:
     return CORE.data.get(KEY_HIGH_PERFORMANCE_NETWORKING, False)
 
 
-def validate_ipv6(value: bool) -> bool:
-    if CORE.is_nrf52 and not value:
-        raise cv.Invalid("On nRF52, enable_ipv6 must be true")
+def _get_data() -> NetworkData:
+    if DOMAIN not in CORE.data:
+        CORE.data[DOMAIN] = NetworkData()
+    return CORE.data[DOMAIN]
 
-    return value
+
+def _ip_request(
+    update: Callable[[NetworkData], None],
+) -> Callable[[ConfigType], ConfigType]:
+    def validator(config: ConfigType) -> ConfigType:
+        update(_get_data())
+        return config
+
+    return validator
+
+
+def require_ipv4(name: str) -> Callable[[ConfigType], ConfigType]:
+    """Return a validator declaring that component `name` needs IPv4."""
+    return _ip_request(lambda data: data.require_ipv4.add(name))
+
+
+def require_ipv6(name: str) -> Callable[[ConfigType], ConfigType]:
+    """Return a validator declaring that component `name` needs IPv6."""
+    return _ip_request(lambda data: data.require_ipv6.add(name))
+
+
+def request_ipv4_off(name: str) -> Callable[[ConfigType], ConfigType]:
+    """Return a validator asking for IPv4 to default to off, on behalf of `name`."""
+    return _ip_request(lambda data: data.request_ipv4_off.add(name))
 
 
 def get_network_priority(iface: str) -> float | None:
@@ -348,33 +384,30 @@ def _validate_priority_list(value: Any) -> list[dict[str, str]]:
     return entries
 
 
+# Shared so _final_validate()'s auto-flip to True enforces the same version floor as
+# the schema does for a user-typed 'enable_ipv6: true'.
+_ENABLE_IPV6_FRAMEWORK_VERSION_VALIDATOR = cv.require_framework_version(
+    bk72xx_arduino=cv.Version(1, 7, 0),
+    esp_idf=cv.Version(0, 0, 0),
+    esp32_arduino=cv.Version(0, 0, 0),
+    esp8266_arduino=cv.Version(0, 0, 0),
+    host=cv.Version(0, 0, 0),
+    rp2_arduino=cv.Version(0, 0, 0),
+    nrf52_zephyr=cv.Version(0, 0, 0),
+)
+
+
 CONFIG_SCHEMA = cv.All(
     cv.Schema(
         {
             cv.GenerateID(): cv.declare_id(NetworkComponent),
-            cv.SplitDefault(
-                CONF_ENABLE_IPV6,
-                bk72xx=False,
-                esp32=False,
-                esp8266=False,
-                host=False,
-                rp2=False,
-                nrf52=True,
-            ): cv.All(
+            cv.Optional(CONF_ENABLE_IPV4): cv.boolean,
+            cv.Optional(CONF_ENABLE_IPV6): cv.All(
                 cv.boolean,
                 cv.Any(
-                    cv.require_framework_version(
-                        bk72xx_arduino=cv.Version(1, 7, 0),
-                        esp_idf=cv.Version(0, 0, 0),
-                        esp32_arduino=cv.Version(0, 0, 0),
-                        esp8266_arduino=cv.Version(0, 0, 0),
-                        host=cv.Version(0, 0, 0),
-                        rp2_arduino=cv.Version(0, 0, 0),
-                        nrf52_zephyr=cv.Version(0, 0, 0),
-                    ),
+                    _ENABLE_IPV6_FRAMEWORK_VERSION_VALIDATOR,
                     cv.boolean_false,
                 ),
-                validate_ipv6,
             ),
             cv.Optional(CONF_MIN_IPV6_ADDR_COUNT, default=0): cv.positive_int,
             cv.Optional(CONF_ENABLE_HIGH_PERFORMANCE): cv.All(
@@ -392,13 +425,50 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
+def _resolve_ip_family(
+    config: ConfigType, key: str, family: str, requirers: set[str], default: bool
+) -> bool:
+    """Resolve `key` to a bool; return True when a component requires the family."""
+    if not requirers:
+        config.setdefault(key, default)
+        return False
+    if config.get(key) is False:
+        raise cv.Invalid(
+            f"{family} is required by {', '.join(sorted(requirers))}, but "
+            f"'network: {key}: false' explicitly disables it"
+        )
+    config[key] = True
+    return True
+
+
 def _final_validate(config: ConfigType) -> None:
-    """Check that every interface named in 'priority' has a corresponding component block."""
-    full = fv.full_config.get()
+    full_config = fv.full_config.get()
+    data = _get_data()
+
+    if _resolve_ip_family(
+        config, CONF_ENABLE_IPV6, "IPv6", data.require_ipv6, default=False
+    ):
+        # Turning IPv6 on must not bypass the version gate cv.Any() enforces.
+        _ENABLE_IPV6_FRAMEWORK_VERSION_VALIDATOR(True)
+    _resolve_ip_family(
+        config,
+        CONF_ENABLE_IPV4,
+        "IPv4",
+        data.require_ipv4,
+        default=not data.request_ipv4_off,
+    )
+
+    if not config[CONF_ENABLE_IPV4] and not config[CONF_ENABLE_IPV6]:
+        raise cv.Invalid(
+            "'enable_ipv4: false' requires 'enable_ipv6: true' -- disabling both "
+            "leaves no IP stack"
+        )
+
+    # Check that every interface named in 'priority' has a corresponding component block.
     priority_list = config.get(CONF_PRIORITY, [])
     for entry in priority_list:
         iface = entry["interface"]
-        if iface not in full:
+        if iface not in full_config:
             raise cv.Invalid(
                 f"'{iface}' is listed in 'network: priority:' but no '{iface}:' "
                 f"component is configured",
@@ -430,7 +500,8 @@ FINAL_VALIDATE_SCHEMA = _final_validate
 @coroutine_with_priority(CoroPriority.NETWORK)
 async def to_code(config: ConfigType) -> None:
     cg.add_define("USE_NETWORK")
-    # ESP32 with Arduino uses ESP-IDF network APIs directly, no Arduino Network library needed
+    enable_ipv6 = config[CONF_ENABLE_IPV6]
+    enable_ipv4 = config[CONF_ENABLE_IPV4]
 
     # Store the user-declared network priority list in CORE.data so that ethernet,
     # wifi and other network components can query it via get_network_priority()
@@ -557,7 +628,8 @@ async def to_code(config: ConfigType) -> None:
 
     if CORE.is_nrf52:
         zephyr_add_prj_conf("NETWORKING", True)
-        zephyr_add_prj_conf("NET_IPV6", True)
+        zephyr_add_prj_conf("NET_IPV4", enable_ipv4)
+        zephyr_add_prj_conf("NET_IPV6", enable_ipv6)
         zephyr_add_prj_conf("NET_TCP", True)
         zephyr_add_prj_conf("NET_UDP", True)
         # The nRF Connect SDK replaces mbedTLS with PSA/Oberon crypto and does not provide the
@@ -582,28 +654,28 @@ async def to_code(config: ConfigType) -> None:
         zephyr_add_prj_conf("NET_TCP_MAX_RECV_WINDOW_SIZE", 2280)
         zephyr_add_prj_conf("NET_TCP_MAX_SEND_WINDOW_SIZE", 2280)
 
-    if (enable_ipv6 := config.get(CONF_ENABLE_IPV6, None)) is not None:
-        cg.add_define("USE_NETWORK_IPV6", enable_ipv6)
-        if enable_ipv6:
-            cg.add_define(
-                "USE_NETWORK_MIN_IPV6_ADDR_COUNT", config[CONF_MIN_IPV6_ADDR_COUNT]
-            )
-        if CORE.is_esp32:
-            if CORE.using_arduino:
-                add_idf_sdkconfig_option("CONFIG_LWIP_IPV6", True)
-                add_idf_sdkconfig_option("CONFIG_LWIP_IPV6_AUTOCONFIG", True)
-            else:
-                add_idf_sdkconfig_option("CONFIG_LWIP_IPV6", enable_ipv6)
-                add_idf_sdkconfig_option("CONFIG_LWIP_IPV6_AUTOCONFIG", enable_ipv6)
-        elif enable_ipv6:
-            cg.add_build_flag("-DCONFIG_LWIP_IPV6")
-            cg.add_build_flag("-DCONFIG_LWIP_IPV6_AUTOCONFIG")
-            if CORE.is_bk72xx:
-                cg.add_build_flag("-DCONFIG_IPV6")
-            if CORE.is_esp8266:
-                cg.add_build_flag("-DPIO_FRAMEWORK_ARDUINO_LWIP2_IPV6_LOW_MEMORY")
-            if CORE.is_rp2:
-                cg.add_build_flag("-DPIO_FRAMEWORK_ARDUINO_ENABLE_IPV6")
+    cg.add_define("USE_NETWORK_IPV6", enable_ipv6)
+    if enable_ipv6:
+        cg.add_define(
+            "USE_NETWORK_MIN_IPV6_ADDR_COUNT", config[CONF_MIN_IPV6_ADDR_COUNT]
+        )
+    if CORE.is_esp32:
+        add_idf_sdkconfig_option("CONFIG_LWIP_IPV6", enable_ipv6)
+        add_idf_sdkconfig_option("CONFIG_LWIP_IPV6_AUTOCONFIG", enable_ipv6)
+    elif enable_ipv6:
+        cg.add_build_flag("-DCONFIG_LWIP_IPV6")
+        cg.add_build_flag("-DCONFIG_LWIP_IPV6_AUTOCONFIG")
+        if CORE.is_bk72xx:
+            cg.add_build_flag("-DCONFIG_IPV6")
+        if CORE.is_esp8266:
+            cg.add_build_flag("-DPIO_FRAMEWORK_ARDUINO_LWIP2_IPV6_LOW_MEMORY")
+        if CORE.is_rp2:
+            cg.add_build_flag("-DPIO_FRAMEWORK_ARDUINO_ENABLE_IPV6")
+
+    cg.add_define("USE_NETWORK_IPV4", enable_ipv4)
+    if CORE.is_esp32:
+        add_idf_sdkconfig_option("CONFIG_LWIP_IPV4", enable_ipv4)
+
     # Pvariable creation lives in a separate coroutine at NETWORK_SERVICES so it
     # emits after wifi/ethernet at COMMUNICATION. This keeps compile-time config
     # (above) separate from C++ object lifecycle and allows wiring in interface

esphome/components/network/ip_address.h

@@ -15,14 +15,14 @@
 #if USE_ARDUINO
 #include <Arduino.h>
 #include <IPAddress.h>
-#endif /* USE_ADRDUINO */
+#endif /* USE_ARDUINO */
 
 #ifdef USE_HOST
 #include <arpa/inet.h>
 #if USE_NETWORK_IPV6
 using ip4_addr_t = struct in_addr;
 using ip6_addr_t = struct in6_addr;
-struct ip_addr_t {
+struct ip_addr_t {  // NOLINT(readability-identifier-naming) mirrors the lwIP name the other platforms use
   union {
     struct in6_addr ip6;
     struct in_addr ip4;
@@ -55,9 +55,23 @@ using ip4_addr_t = in_addr;
 #include <zephyr/net/net_ip.h>
 #include <zephyr/net/socket.h>
 #include <zephyr/posix/arpa/inet.h>
-using ip_addr_t = struct in6_addr;
-static inline int ipaddr_aton(const char *cp, ip_addr_t *addr) { return inet_pton(AF_INET6, cp, addr) == 1 ? 1 : 0; }
+#ifndef CONFIG_NET_IPV6
+#error "Zephyr builds need CONFIG_NET_IPV6"
 #endif
+using ip_addr_t = struct in6_addr;
+static inline int ipaddr_aton(const char *cp, ip_addr_t *addr) {
+#if defined(CONFIG_NET_IPV4)
+  // Dual-stack: a plain IPv4 literal must still parse -- store it as a v4-mapped
+  // IPv6 address (::ffff:a.b.c.d), matching the 4-octet constructor below.
+  struct in_addr addr4;
+  if (inet_pton(AF_INET, cp, &addr4) == 1) {
+    net_ipv6_addr_create_v4_mapped(&addr4, addr);
+    return 1;
+  }
+#endif /* CONFIG_NET_IPV4 */
+  return inet_pton(AF_INET6, cp, addr) == 1 ? 1 : 0;
+}
+#endif /* USE_ZEPHYR */
 
 #if USE_ESP32_FRAMEWORK_ARDUINO
 #define arduino_ns Arduino_h
@@ -65,16 +79,24 @@ static inline int ipaddr_aton(const char *cp, ip_addr_t *addr) { return inet_pto
 #define arduino_ns arduino
 #elif USE_ARDUINO
 #define arduino_ns
-#endif
+#endif /* USE_ESP32_FRAMEWORK_ARDUINO */
 
 #ifdef USE_ESP32
 #include <esp_netif.h>
-#endif
+#endif /* USE_ESP32 */
 
 namespace esphome::network {
 
 /// Buffer size for IP address string (IPv6 max: 39 chars + null)
-static constexpr size_t IP_ADDRESS_BUFFER_SIZE = 40;
+static constexpr size_t IP_ADDRESS_BUFFER_SIZE =
+#if defined(USE_ZEPHYR)
+    // Mainline Zephyr's inet_ntop() (net_addr_ntop() in subsys/net/ip/utils.c) never checks
+    // its size argument, so an undersized buffer overflows instead of failing cleanly. Size
+    // for the worst case: an IPv4-mapped IPv6 address needs up to INET6_ADDRSTRLEN (46).
+    46;
+#else
+    40;
+#endif
 
 /// Lowercase hex digits in IP address string (A-F -> a-f for IPv6 per RFC 5952)
 inline void lowercase_ip_str(char *buf) {
@@ -86,19 +108,58 @@ inline void lowercase_ip_str(char *buf) {
 
 struct IPAddress {
  public:
-#ifdef USE_ZEPHYR
+#if defined(USE_ZEPHYR)
+  // With CONFIG_NET_IPV4 (dual stack), IPv4 addresses are stored v4-mapped (::ffff:a.b.c.d).
   IPAddress() { memset(&ip_addr_, 0, sizeof(ip_addr_)); }
+#if defined(CONFIG_NET_IPV4)
+  IPAddress(uint8_t first, uint8_t second, uint8_t third, uint8_t fourth) {
+    struct in_addr addr4{
+        .s_addr = htonl(((uint32_t) first << 24) | ((uint32_t) second << 16) | ((uint32_t) third << 8) | fourth)};
+    net_ipv6_addr_create_v4_mapped(&addr4, &ip_addr_);
+  }
+#endif /* CONFIG_NET_IPV4 */
   IPAddress(const std::string &in_address) : ip_addr_{} { ipaddr_aton(in_address.c_str(), &ip_addr_); }
   IPAddress(const struct in6_addr *other_ip) { ip_addr_ = *other_ip; }
   IPAddress(const struct sockaddr_in6 *addr) { ip_addr_ = addr->sin6_addr; }
 
   operator struct in6_addr() const { return ip_addr_; }
 
-  bool is_set() const { return !net_ipv6_is_addr_unspecified(&ip_addr_); }
+  bool is_set() const {
+#if defined(CONFIG_NET_IPV4)
+    if (net_ipv6_addr_is_v4_mapped(&ip_addr_))
+      return ip_addr_.s6_addr32[3] != 0;
+#endif /* CONFIG_NET_IPV4 */
+    return !net_ipv6_is_addr_unspecified(&ip_addr_);
+  }
+#if defined(CONFIG_NET_IPV4)
+  // V4-mapped (::ffff:a.b.c.d) addresses represent IPv4 traffic carried over the AF_INET6 socket
+  // Zephyr always uses when CONFIG_NET_IPV6 is on; anything else is real IPv6.
+  bool is_ip4() const { return this->is_set() && net_ipv6_addr_is_v4_mapped(&ip_addr_); }
+  bool is_ip6() const { return this->is_set() && !net_ipv6_addr_is_v4_mapped(&ip_addr_); }
+#else
   bool is_ip4() const { return false; }
   bool is_ip6() const { return this->is_set(); }
-  bool is_multicast() const { return net_ipv6_is_addr_mcast(&ip_addr_); }
+#endif /* CONFIG_NET_IPV4 */
+  bool is_multicast() const {
+#if defined(CONFIG_NET_IPV4)
+    if (net_ipv6_addr_is_v4_mapped(&ip_addr_)) {
+      struct in_addr addr4{};
+      addr4.s_addr = ip_addr_.s6_addr32[3];
+      return net_ipv4_is_addr_mcast(&addr4);
+    }
+#endif /* CONFIG_NET_IPV4 */
+    return net_ipv6_is_addr_mcast(&ip_addr_);
+  }
   char *str_to(char *buf) const {
+#if defined(CONFIG_NET_IPV4)
+    if (this->is_ip4()) {
+      struct in_addr addr4{};
+      addr4.s_addr = ip_addr_.s6_addr32[3];
+      if (inet_ntop(AF_INET, &addr4, buf, IP_ADDRESS_BUFFER_SIZE) == nullptr)
+        buf[0] = '\0';
+      return buf;
+    }
+#endif /* CONFIG_NET_IPV4 */
     if (inet_ntop(AF_INET6, &ip_addr_, buf, IP_ADDRESS_BUFFER_SIZE) == nullptr)
       buf[0] = '\0';
     return buf;
@@ -134,8 +195,8 @@ struct IPAddress {
   operator ip_addr_t() const { return this->ip_addr_; }
   bool is_set() const {
     if (this->ip_addr_.type == IPADDR_TYPE_V6) {
-      static constexpr uint8_t zero[sizeof(struct in6_addr)] = {};
-      return memcmp(this->ip_addr_.u_addr.ip6.s6_addr, zero, sizeof(zero)) != 0;
+      static constexpr uint8_t ZERO[sizeof(struct in6_addr)] = {};
+      return memcmp(this->ip_addr_.u_addr.ip6.s6_addr, ZERO, sizeof(ZERO)) != 0;
     }
     return this->ip_addr_.u_addr.ip4.s_addr != 0;
   }
@@ -187,59 +248,72 @@ struct IPAddress {
     return buf;  // IPv4 only, no hex letters to lowercase
   }
 #endif  // USE_NETWORK_IPV6
-#else
+#else   // LWIP
   IPAddress() { ip_addr_set_zero(&ip_addr_); }
+#if LWIP_IPV4
   IPAddress(uint8_t first, uint8_t second, uint8_t third, uint8_t fourth) {
     IP_ADDR4(&ip_addr_, first, second, third, fourth);
   }
+#endif /* LWIP_IPV4 */
   IPAddress(const ip_addr_t *other_ip) { ip_addr_copy(ip_addr_, *other_ip); }
   IPAddress(const char *in_address) { ipaddr_aton(in_address, &ip_addr_); }
   IPAddress(const std::string &in_address) { ipaddr_aton(in_address.c_str(), &ip_addr_); }
+#if LWIP_IPV4
   IPAddress(ip4_addr_t *other_ip) {
     memcpy((void *) &ip_addr_, (void *) other_ip, sizeof(ip4_addr_t));
 #if LWIP_IPV6
     ip_addr_.type = IPADDR_TYPE_V4;
-#endif
+#endif /* LWIP_IPV6 */
   }
+#endif /* LWIP_IPV4 */
 #if USE_ARDUINO
   IPAddress(const arduino_ns::IPAddress &other_ip) { ip_addr_set_ip4_u32(&ip_addr_, other_ip); }
-#endif
+#endif /* USE_ARDUINO */
 #if LWIP_IPV6
   IPAddress(ip6_addr_t *other_ip) {
     memcpy((void *) &ip_addr_, (void *) other_ip, sizeof(ip6_addr_t));
-    ip_addr_.type = IPADDR_TYPE_V6;
+    IP_SET_TYPE_VAL(ip_addr_, IPADDR_TYPE_V6);
   }
 #endif /* LWIP_IPV6 */
 
 #ifdef USE_ESP32
 #if LWIP_IPV6
   IPAddress(esp_ip6_addr_t *other_ip) {
-    memcpy((void *) &ip_addr_.u_addr.ip6, (void *) other_ip, sizeof(esp_ip6_addr_t));
-    ip_addr_.type = IPADDR_TYPE_V6;
+    memcpy((void *) ip_2_ip6(&ip_addr_), (void *) other_ip, sizeof(esp_ip6_addr_t));
+    IP_SET_TYPE_VAL(ip_addr_, IPADDR_TYPE_V6);
   }
 #endif /* LWIP_IPV6 */
+#if LWIP_IPV4
   IPAddress(esp_ip4_addr_t *other_ip) {
     memcpy((void *) &ip_addr_, (void *) other_ip, sizeof(esp_ip4_addr_t));
 #if LWIP_IPV6
     ip_addr_.type = IPADDR_TYPE_V4;
-#endif
+#endif /* LWIP_IPV6 */
   }
+#endif /* LWIP_IPV4 */
   IPAddress(esp_ip_addr_t *other_ip) {
 #if LWIP_IPV6
     memcpy((void *) &ip_addr_, (void *) other_ip, sizeof(ip_addr_));
-#else
+#elif LWIP_IPV4
     memcpy((void *) &ip_addr_, (void *) &other_ip->u_addr.ip4, sizeof(ip_addr_));
-#endif
+#endif /* LWIP_IPV6 */
   }
   operator esp_ip_addr_t() const {
-    esp_ip_addr_t tmp;
+    esp_ip_addr_t tmp{};
 #if LWIP_IPV6
     memcpy((void *) &tmp, (void *) &ip_addr_, sizeof(ip_addr_));
-#else
+#if !LWIP_IPV4
+    tmp.type = IPADDR_TYPE_V6;
+#endif /* !LWIP_IPV4 */
+#elif LWIP_IPV4
     memcpy((void *) &tmp.u_addr.ip4, (void *) &ip_addr_, sizeof(ip_addr_));
+    tmp.type = IPADDR_TYPE_V4;
+#else
+#error "Neither LWIP_IPV4 nor LWIP_IPV6 is enabled"
 #endif /* LWIP_IPV6 */
     return tmp;
   }
+#if LWIP_IPV4
   operator esp_ip4_addr_t() const {
     esp_ip4_addr_t tmp;
 #if LWIP_IPV6
@@ -249,16 +323,17 @@ struct IPAddress {
 #endif /* LWIP_IPV6 */
     return tmp;
   }
+#endif /* LWIP_IPV4 */
 #endif /* USE_ESP32 */
 
   operator ip_addr_t() const { return ip_addr_; }
-#if LWIP_IPV6
+#if LWIP_IPV6 && LWIP_IPV4
   operator ip4_addr_t() const { return *ip_2_ip4(&ip_addr_); }
-#endif /* LWIP_IPV6 */
+#endif /* LWIP_IPV6 && LWIP_IPV4 */
 
 #if USE_ARDUINO
   operator arduino_ns::IPAddress() const { return ip_addr_get_ip4_u32(&ip_addr_); }
-#endif
+#endif /* USE_ARDUINO */
 
   bool is_set() const { return !ip_addr_isany(&ip_addr_); }  // NOLINT(readability-simplify-boolean-expr)
   bool is_ip4() const { return IP_IS_V4(&ip_addr_); }
@@ -273,17 +348,19 @@ struct IPAddress {
   }
   bool operator==(const IPAddress &other) const { return ip_addr_cmp(&ip_addr_, &other.ip_addr_); }
   bool operator!=(const IPAddress &other) const { return !ip_addr_cmp(&ip_addr_, &other.ip_addr_); }
+#if LWIP_IPV4
   IPAddress &operator+=(uint8_t increase) {
     if (IP_IS_V4(&ip_addr_)) {
 #if LWIP_IPV6
       (((u8_t *) (&ip_addr_.u_addr.ip4))[3]) += increase;
 #else
       (((u8_t *) (&ip_addr_.addr))[3]) += increase;
-#endif /* LWIP_IPV6 */
+#endif  /* LWIP_IPV6 */
     }
     return *this;
   }
-#endif
+#endif  /* LWIP_IPV4 */
+#endif  // LWIP
 
  protected:
   ip_addr_t ip_addr_;
@@ -292,4 +369,4 @@ struct IPAddress {
 using IPAddresses = std::array<IPAddress, 5>;
 
 }  // namespace esphome::network
-#endif
+#endif /* NETWORK */

esphome/components/nrf52/__init__.py

@@ -11,6 +11,7 @@ from esphome import pins
 from esphome.build_helpers import pch
 from esphome.build_helpers.ccache import resolve_ccache_path
 import esphome.codegen as cg
+from esphome.components import network
 from esphome.components.zephyr import (
     Section,
     add_extra_script,
@@ -322,6 +323,8 @@ CONFIG_SCHEMA = cv.All(
     _default_self_reset_pin,
     _resolve_toolchain,
     set_framework,
+    network.require_ipv6(PLATFORM_NRF52),
+    network.request_ipv4_off(PLATFORM_NRF52),
 )
 
 

esphome/components/openthread/__init__.py

@@ -31,7 +31,7 @@ from esphome.components.esp32 import (
     require_vfs_select,
 )
 from esphome.components.mdns import MDNSComponent, enable_mdns_storage
-from esphome.components.network import DOMAIN as NETWORK_DOMAIN, add_use_address
+from esphome.components.network import add_use_address, request_ipv4_off, require_ipv6
 from esphome.components.nrf52.framework import include_west_project
 from esphome.components.zephyr import zephyr_add_prj_conf
 from esphome.config_helpers import filter_source_files_from_platform
@@ -40,7 +40,6 @@ from esphome.const import (
     CONF_AP,
     CONF_BAUD_RATE,
     CONF_CHANNEL,
-    CONF_ENABLE_IPV6,
     CONF_ENABLE_ON_BOOT,
     CONF_FRAMEWORK,
     CONF_HARDWARE_UART,
@@ -50,6 +49,7 @@ from esphome.const import (
     CONF_LOGGER,
     CONF_NETWORKS,
     CONF_NUMBER,
+    CONF_OPENTHREAD,
     CONF_OUTPUT_POWER,
     CONF_RESET_PIN,
     CONF_RX_PIN,
@@ -448,17 +448,13 @@ CONFIG_SCHEMA = cv.All(
     _validate_platform,
     _validate,
     _require_vfs,
+    require_ipv6(CONF_OPENTHREAD),
+    request_ipv4_off(CONF_OPENTHREAD),
 )
 
 
 def _final_validate(config: ConfigType) -> ConfigType:
     full_config = fv.full_config.get()
-    network_config = full_config.get(NETWORK_DOMAIN, {})
-    if not network_config.get(CONF_ENABLE_IPV6, False):
-        raise cv.Invalid(
-            "OpenThread requires IPv6 to be enabled in the network component. "
-            "Please set `enable_ipv6: true` in the `network` configuration."
-        )
 
     wifi_config = full_config.get(CONF_WIFI)
     if CONF_BORDER_ROUTER not in config:

esphome/components/rp2/__init__.py

@@ -7,6 +7,7 @@ import subprocess
 from typing import Any
 
 import esphome.codegen as cg
+from esphome.components import network
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_BOARD,
@@ -318,6 +319,7 @@ CONFIG_SCHEMA = cv.All(
     _detect_variant,
     cv.require_platformio_toolchain("RP2"),
     set_core_data,
+    network.require_ipv4(PLATFORM_RP2),
 )
 
 

esphome/components/socket/__init__.py

@@ -211,9 +211,15 @@ def require_tcp_listener() -> None:
 
 
 def consume_role_sockets(component: str) -> Callable[[ConfigType], ConfigType]:
-    """Socket accounting for a role keyed client or server schema."""
+    """Socket accounting for a role keyed client or server schema.
+
+    Also requires IPv4, since the TCP link and listener helpers are IPv4 only.
+    """
 
     def validator(config: ConfigType) -> ConfigType:
+        from esphome.components import network
+
+        network.require_ipv4(component)(config)
         if config[CONF_ROLE] == "server":
             consume_sockets(1, component, SocketType.TCP_LISTEN)(config)
         return consume_sockets(1, component)(config)

esphome/components/socket/headers.h

@@ -204,6 +204,11 @@ using socklen_t = uint32_t;
 
 #if defined(USE_SOCKET_IMPL_LWIP_TCP) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS) || defined(USE_SOCKET_IMPL_BSD_SOCKETS)
 
+// Build tests compile socket without network, which normally defines this
+#ifndef USE_NETWORK_IPV4
+#define USE_NETWORK_IPV4 1
+#endif
+
 namespace esphome::socket {
 
 // Maximum length for formatted socket address string (IP address without port)

esphome/components/socket/socket.cpp

@@ -42,54 +42,55 @@ bool socket_ready_fd(int /*fd*/, bool /*loop_monitored*/) { return true; }
 #endif
 
 // Platform-specific inet_ntop wrappers
+#if USE_NETWORK_IPV4
 #if defined(USE_SOCKET_IMPL_LWIP_TCP)
 // LWIP raw TCP (ESP8266) uses inet_ntoa_r which takes struct by value
 static inline const char *esphome_inet_ntop4(const void *addr, char *buf, size_t size) {
   inet_ntoa_r(*reinterpret_cast<const struct in_addr *>(addr), buf, size);
   return buf;
 }
-#if USE_NETWORK_IPV6
-static inline const char *esphome_inet_ntop6(const void *addr, char *buf, size_t size) {
-  inet6_ntoa_r(*reinterpret_cast<const ip6_addr_t *>(addr), buf, size);
-  return buf;
-}
-#endif
 #elif defined(USE_SOCKET_IMPL_LWIP_SOCKETS)
 // LWIP sockets (LibreTiny, ESP32 Arduino)
 static inline const char *esphome_inet_ntop4(const void *addr, char *buf, size_t size) {
   return lwip_inet_ntop(AF_INET, addr, buf, size);
 }
-#if USE_NETWORK_IPV6
-static inline const char *esphome_inet_ntop6(const void *addr, char *buf, size_t size) {
-  return lwip_inet_ntop(AF_INET6, addr, buf, size);
-}
-#endif
 #elif defined(USE_ZEPHYR)
 // Zephyr BSD sockets — use Zephyr native address formatting via POSIX-subset wrappers.
 // <zephyr/net/socket.h> is already included transitively through <sys/socket.h>.
 static inline const char *esphome_inet_ntop4(const void *addr, char *buf, size_t size) {
   return zsock_inet_ntop(AF_INET, addr, buf, size);
 }
-// IPv6 is always enabled on nRF52 (config validation enforces enable_ipv6=True),
-// but the guard is retained for consistency with other platform blocks.
-#if USE_NETWORK_IPV6
-static inline const char *esphome_inet_ntop6(const void *addr, char *buf, size_t size) {
-  return zsock_inet_ntop(AF_INET6, addr, buf, size);
-}
-#endif
 #else
 // BSD sockets (host, ESP32-IDF)
 static inline const char *esphome_inet_ntop4(const void *addr, char *buf, size_t size) {
   return inet_ntop(AF_INET, addr, buf, size);
 }
+#endif
+#endif /* USE_NETWORK_IPV4 */
+
 #if USE_NETWORK_IPV6
+#if defined(USE_SOCKET_IMPL_LWIP_TCP)
+static inline const char *esphome_inet_ntop6(const void *addr, char *buf, size_t size) {
+  inet6_ntoa_r(*reinterpret_cast<const ip6_addr_t *>(addr), buf, size);
+  return buf;
+}
+#elif defined(USE_SOCKET_IMPL_LWIP_SOCKETS)
+static inline const char *esphome_inet_ntop6(const void *addr, char *buf, size_t size) {
+  return lwip_inet_ntop(AF_INET6, addr, buf, size);
+}
+#elif defined(USE_ZEPHYR)
+static inline const char *esphome_inet_ntop6(const void *addr, char *buf, size_t size) {
+  return zsock_inet_ntop(AF_INET6, addr, buf, size);
+}
+#else
 static inline const char *esphome_inet_ntop6(const void *addr, char *buf, size_t size) {
   return inet_ntop(AF_INET6, addr, buf, size);
 }
 #endif
-#endif
+#endif /* USE_NETWORK_IPV6 */
 
 bool sockaddr_to_ipv4(const struct sockaddr *addr, uint32_t *out) {
+#if USE_NETWORK_IPV4
   if (addr->sa_family == AF_INET) {
     *out = reinterpret_cast<const struct sockaddr_in *>(addr)->sin_addr.s_addr;
     return true;
@@ -105,23 +106,31 @@ bool sockaddr_to_ipv4(const struct sockaddr *addr, uint32_t *out) {
     }
   }
 #endif
+#else
+  (void) addr;
+  (void) out;
+#endif /* USE_NETWORK_IPV4 */
   return false;
 }
 
 // Format sockaddr into caller-provided buffer, returns length written (excluding null)
 size_t format_sockaddr_to(const struct sockaddr *addr_ptr, socklen_t len, std::span<char, SOCKADDR_STR_LEN> buf) {
+#if USE_NETWORK_IPV4
   if (addr_ptr->sa_family == AF_INET && len >= sizeof(const struct sockaddr_in)) {
     const auto *addr = reinterpret_cast<const struct sockaddr_in *>(addr_ptr);
     if (esphome_inet_ntop4(&addr->sin_addr, buf.data(), buf.size()) != nullptr)
       return strlen(buf.data());
   }
+#endif /* USE_NETWORK_IPV4 */
 #if USE_NETWORK_IPV6
-  else if (addr_ptr->sa_family == AF_INET6 && len >= sizeof(sockaddr_in6)) {
+  if (addr_ptr->sa_family == AF_INET6 && len >= sizeof(sockaddr_in6)) {
     const auto *addr = reinterpret_cast<const struct sockaddr_in6 *>(addr_ptr);
+#if USE_NETWORK_IPV4
     uint32_t v4;
     // Format a v4 mapped peer as plain IPv4.
     if (sockaddr_to_ipv4(addr_ptr, &v4) && esphome_inet_ntop4(&v4, buf.data(), buf.size()) != nullptr)
       return strlen(buf.data());
+#endif /* USE_NETWORK_IPV4 */
     if (esphome_inet_ntop6(&addr->sin6_addr, buf.data(), buf.size()) != nullptr)
       return strlen(buf.data());
   }
@@ -235,6 +244,10 @@ template<typename F> static bool foreach_eligible_ipv6_if(F &&callback) {
 
 bool join_multicast_group(UDPSocket *sock, const char *ip_address, uint32_t *if_index_out) {
   if (strchr(ip_address, ':') == nullptr) {
+#if !USE_NETWORK_IPV4
+    errno = EAFNOSUPPORT;
+    return false;
+#else
     struct in_addr ipv4mc{};
 #ifdef USE_ZEPHYR
     if (zsock_inet_pton(AF_INET, ip_address, &ipv4mc) != 1) {
@@ -262,6 +275,7 @@ bool join_multicast_group(UDPSocket *sock, const char *ip_address, uint32_t *if_
       *if_index_out = 0;
     }
     return true;
+#endif  // USE_NETWORK_IPV4
   }
 #ifdef USE_SOCKET_IPV6
   struct ipv6_mreq imreq6{};
@@ -373,7 +387,7 @@ socklen_t set_sockaddr(struct sockaddr *addr, socklen_t addrlen, const char *ip_
 #else
     // Use LWIP-specific functions
     ip6_addr_t ip6;
-    if (inet6_aton(ip_address, &ip6) == 0) {
+    if (inet6_aton(ip_address, &ip6) != 1) {
       errno = EINVAL;
       return 0;
     }
@@ -382,6 +396,7 @@ socklen_t set_sockaddr(struct sockaddr *addr, socklen_t addrlen, const char *ip_
     return sizeof(sockaddr_in6);
   }
 #endif /* USE_NETWORK_IPV6 */
+#if USE_NETWORK_IPV4
   if (addrlen < sizeof(sockaddr_in)) {
     errno = EINVAL;
     return 0;
@@ -404,6 +419,11 @@ socklen_t set_sockaddr(struct sockaddr *addr, socklen_t addrlen, const char *ip_
 #endif
   server->sin_port = htons(port);
   return sizeof(sockaddr_in);
+#else
+  memset(addr, 0, addrlen);
+  errno = EAFNOSUPPORT;
+  return 0;
+#endif /* USE_NETWORK_IPV4 */
 }
 
 #if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS)

esphome/components/statsd/__init__.py

@@ -1,5 +1,5 @@
 import esphome.codegen as cg
-from esphome.components import binary_sensor, sensor, socket
+from esphome.components import binary_sensor, network, sensor, socket
 from esphome.components.const import CONF_HOST
 import esphome.config_validation as cv
 from esphome.const import (
@@ -48,6 +48,7 @@ CONFIG_SCHEMA = cv.All(
             ),
         }
     ).extend(cv.polling_component_schema("10s")),
+    network.require_ipv4("statsd"),
     socket.consume_sockets(1, "statsd", socket.SocketType.UDP),
 )
 

esphome/components/udp/__init__.py

@@ -4,8 +4,8 @@ from typing import Any
 from esphome import automation
 from esphome.automation import Trigger
 import esphome.codegen as cg
+from esphome.components import network
 from esphome.components.const import CONF_ON_RECEIVE
-from esphome.components.network import DOMAIN as NETWORK_DOMAIN
 from esphome.components.packet_transport import (
     CONF_BINARY_SENSORS,
     CONF_ENCRYPTION,
@@ -15,13 +15,7 @@ from esphome.components.packet_transport import (
     CONF_SENSORS,
 )
 import esphome.config_validation as cv
-from esphome.const import (
-    CONF_DATA,
-    CONF_ENABLE_IPV6,
-    CONF_ID,
-    CONF_PORT,
-    CONF_TRIGGER_ID,
-)
+from esphome.const import CONF_DATA, CONF_ID, CONF_PORT, CONF_TRIGGER_ID
 from esphome.core import ID
 from esphome.cpp_generator import MockObj, TemplateArgsType
 import esphome.final_validate as fv
@@ -106,7 +100,6 @@ def _final_validate(config: ConfigType) -> None:
     from esphome.components import socket
 
     full_config = fv.full_config.get()
-    ipv6_enabled = full_config.get(NETWORK_DOMAIN, {}).get(CONF_ENABLE_IPV6, False)
     raw_lwip = (
         full_config.get(socket.DOMAIN, {}).get(socket.CONF_IMPLEMENTATION)
         == socket.IMPLEMENTATION_LWIP_TCP
@@ -122,11 +115,6 @@ def _final_validate(config: ConfigType) -> None:
                 raise cv.Invalid(
                     f"IPv6 address '{addr}' is not supported on this platform", [key]
                 )
-            if not ipv6_enabled:
-                raise cv.Invalid(
-                    f"IPv6 address '{addr}' requires 'enable_ipv6: true' in the 'network' component",
-                    [key],
-                )
 
 
 FINAL_VALIDATE_SCHEMA = _final_validate
@@ -136,6 +124,12 @@ def _ipv6_addresses(config: ConfigType) -> list[IPv6Address]:
     return [addr for addr in config[CONF_ADDRESSES] if addr.version == 6]
 
 
+def _require_ipv6_for_ipv6_addresses(config: ConfigType) -> ConfigType:
+    if _ipv6_addresses(config) or config[CONF_LISTEN_ADDRESS].version == 6:
+        network.require_ipv6("udp")(config)
+    return config
+
+
 def _has_ipv4_address(config: ConfigType) -> bool:
     return any(addr.version == 4 for addr in config[CONF_ADDRESSES])
 
@@ -178,6 +172,8 @@ CONFIG_SCHEMA = cv.All(
         }
     ).extend(RELOCATED),
     _consume_udp_sockets,
+    network.require_ipv4("udp"),
+    _require_ipv6_for_ipv6_addresses,
 )
 
 

esphome/components/voice_assistant/__init__.py

@@ -1,6 +1,12 @@
 from esphome import automation, core
 import esphome.codegen as cg
-from esphome.components import media_player, micro_wake_word, microphone, speaker
+from esphome.components import (
+    media_player,
+    micro_wake_word,
+    microphone,
+    network,
+    speaker,
+)
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_ID,
@@ -169,6 +175,7 @@ CONFIG_SCHEMA = cv.All(
         }
     ).extend(cv.COMPONENT_SCHEMA),
     cv.key_requires((CONF_ON_TTS_STREAM_START, CONF_ON_TTS_STREAM_END), CONF_SPEAKER),
+    network.require_ipv4("voice_assistant"),
 )
 
 FINAL_VALIDATE_SCHEMA = cv.All(

esphome/components/wake_on_lan/button.py

@@ -1,5 +1,5 @@
 import esphome.codegen as cg
-from esphome.components import button, socket
+from esphome.components import button, network, socket
 import esphome.config_validation as cv
 from esphome.const import CONF_ID
 from esphome.types import ConfigType
@@ -22,6 +22,7 @@ CONFIG_SCHEMA = cv.All(
             cv.Required(CONF_TARGET_MAC_ADDRESS): cv.mac_address,
         }
     ),
+    network.require_ipv4("wake_on_lan"),
     socket.consume_sockets(1, "wake_on_lan", socket.SocketType.UDP),
 )
 

esphome/components/wifi/__init__.py

@@ -20,6 +20,7 @@ from esphome.components.network import (
     has_high_performance_wifi,
     ip_address_literal,
     require_high_performance_wifi,
+    require_ipv4,
 )
 from esphome.components.psram import is_guaranteed as psram_is_guaranteed
 from esphome.config_helpers import filter_source_files_from_platform
@@ -702,6 +703,7 @@ CONFIG_SCHEMA = cv.All(
     _apply_min_auth_mode_default,
     _validate,
     _report_provisioning_credentials,
+    require_ipv4(DOMAIN),
 )
 
 

esphome/components/wireguard/__init__.py

@@ -3,7 +3,7 @@ import re
 
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import time
+from esphome.components import network, time
 from esphome.components.esp32 import CORE, add_idf_sdkconfig_option
 import esphome.config_validation as cv
 from esphome.const import (
@@ -92,6 +92,7 @@ CONFIG_SCHEMA = cv.All(
             PLATFORM_LN882X,
         ]
     ),
+    network.require_ipv4("wireguard"),
 )
 
 

esphome/core/defines.h

@@ -241,6 +241,13 @@
 #define NETWORK_ACTIVE_INTERFACE_LISTENER_COUNT 1
 #define USE_NETWORK
 #define USE_NETWORK_DEFAULT_ROUTE
+#define USE_NETWORK_IPV4 true
+// lwIP is built without IPv6 on ESP8266, RP2 and LibreTiny
+#if defined(USE_ESP32) || defined(USE_HOST)
+#define USE_NETWORK_IPV6 true
+#else
+#define USE_NETWORK_IPV6 false
+#endif
 #define USE_NETWORK_PRIMARY_INTERFACE_WIFI
 #define USE_NEXTION_COMMAND_SPACING
 #define USE_NEXTION_CONF_START_UP_PAGE

script/check_idf_py_equivalence.py

@@ -51,6 +51,11 @@ def _ninja_logs(skip_bootloader: bool) -> list[str]:
 
 
 BOOTLOADER_BYPRODUCT = re.compile(r"(^|/build/)bootloader/")
+# The build step of any CMake ExternalProject. One with BUILD_ALWAYS (ESP-IDF's
+# ulp_embed_binary) re-runs on every build, and ninja re-records all of that
+# edge's byproducts with the new end time even though the files are untouched.
+# Real work on them still changes the hashed firmware.
+EXTERNAL_PROJECT_BUILD_STAMP = re.compile(r"-stamp/[^/]+-build$")
 MACRO_CHANGED = (
     "IDF changed __build_process_project_includes; update "
     "IDF_BOOTLOADER_OVERRIDE in esphome/build_gen/espidf.py"
@@ -96,10 +101,24 @@ def _ninja_mtimes(build_path: Path, logs: list[str]) -> dict[tuple[str, str], st
     for name in logs:
         log = build_path / name
         lines = log.read_text(encoding="utf-8").splitlines() if log.is_file() else []
-        for fields in (line.split("\t") for line in lines if not line.startswith("#")):
-            if len(fields) >= 4 and (
-                name != TOP_NINJA_LOG or not BOOTLOADER_BYPRODUCT.search(fields[3])
-            ):
+        rows = [
+            fields
+            for fields in (
+                line.split("\t") for line in lines if not line.startswith("#")
+            )
+            if len(fields) >= 5
+        ]
+        # Every output of an edge shares its command hash, so the stamp names
+        # the whole external project group.
+        external = {
+            fields[4]
+            for fields in rows
+            if EXTERNAL_PROJECT_BUILD_STAMP.search(fields[3])
+        }
+        for fields in rows:
+            if fields[4] in external:
+                continue
+            if name != TOP_NINJA_LOG or not BOOTLOADER_BYPRODUCT.search(fields[3]):
                 mtimes[name, fields[3]] = fields[2]
     return mtimes
 

tests/component_tests/api/test_outgoing_connection.py

@@ -6,6 +6,7 @@ from pathlib import Path
 
 import pytest
 
+from esphome.components import network
 from esphome.components.api import CONFIG_SCHEMA, FINAL_VALIDATE_SCHEMA
 from esphome.components.esp32 import KEY_BOARD, KEY_VARIANT, VARIANT_ESP32
 import esphome.config_validation as cv
@@ -134,22 +135,20 @@ def test_outgoing_connection_folds_v4_mapped_host(
     assert FINAL_VALIDATE_SCHEMA(config) is not None
 
 
-def test_outgoing_connection_ipv6_host_requires_ipv6(
-    set_core_config: SetCoreConfigCallable,
+@pytest.mark.parametrize(
+    ("host", "required", "other"),
+    [
+        ("192.168.1.2", "require_ipv4", "require_ipv6"),
+        ("::ffff:192.168.1.2", "require_ipv4", "require_ipv6"),
+        ("fd00::1", "require_ipv6", "require_ipv4"),
+    ],
+)
+def test_outgoing_connection_host_requires_its_address_family(
+    set_core_config: SetCoreConfigCallable, host: str, required: str, other: str
 ) -> None:
+    """The host's family is registered with network, which resolves enable_ipv4/6 later."""
     set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
-    config = CONFIG_SCHEMA(_api_config({"host": "fd00::1"}))
-    with pytest.raises(cv.Invalid, match="IPv6 is not"):
-        FINAL_VALIDATE_SCHEMA(config)
-
-
-def test_outgoing_connection_ipv6_host_passes_with_ipv6_enabled(
-    set_core_config: SetCoreConfigCallable,
-) -> None:
-    set_core_config(
-        PlatformFramework.ESP32_IDF,
-        platform_data=ESP32_PLATFORM_DATA,
-        full_config={"network": {"enable_ipv6": True}},
-    )
-    config = CONFIG_SCHEMA(_api_config({"host": "fd00::1"}))
-    assert FINAL_VALIDATE_SCHEMA(config) is not None
+    CONFIG_SCHEMA(_api_config({"host": host}))
+    data = network._get_data()
+    assert "api.outgoing_connection" in getattr(data, required)
+    assert "api.outgoing_connection" not in getattr(data, other)

tests/component_tests/conftest.py

@@ -55,6 +55,7 @@ def reset_core() -> Generator[None]:
     """Reset CORE after each test."""
     yield
     CORE.reset()
+    CORE.testing_mode = False  # reset() leaves it, so a test enabling it would leak
 
 
 @pytest.fixture(autouse=True)

tests/component_tests/network/config/bk72xx_no_network.yaml

@@ -0,0 +1,5 @@
+esphome:
+  name: platformtest
+
+bk72xx:
+  board: generic-bk7252

tests/component_tests/network/config/esp32_arduino_ipv4_disabled_rejected.yaml

@@ -0,0 +1,11 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+  framework:
+    type: arduino
+
+network:
+  enable_ipv6: true
+  enable_ipv4: false

tests/component_tests/network/config/esp32_arduino_ipv6_no_ipv4_requirement.yaml

@@ -0,0 +1,12 @@
+esphome:
+  name: ardtest
+
+esp32:
+  board: esp32dev
+  framework:
+    type: arduino
+
+network:
+  enable_ipv6: true
+
+api:

tests/component_tests/network/config/esp32_arduino_no_network.yaml

@@ -0,0 +1,7 @@
+esphome:
+  name: platformtest
+
+esp32:
+  board: esp32dev
+  framework:
+    type: arduino

tests/component_tests/network/config/esp32_idf_no_network.yaml

@@ -0,0 +1,5 @@
+esphome:
+  name: platformtest
+
+esp32:
+  board: esp32dev

tests/component_tests/network/config/esp8266_ipv4_disabled_rejected.yaml

@@ -0,0 +1,9 @@
+esphome:
+  name: test
+
+esp8266:
+  board: esp01_1m
+
+network:
+  enable_ipv6: true
+  enable_ipv4: false

tests/component_tests/network/config/esp8266_no_network.yaml

@@ -0,0 +1,5 @@
+esphome:
+  name: platformtest
+
+esp8266:
+  board: esp01_1m

tests/component_tests/network/config/host_ipv6_no_ipv4_requirement.yaml

@@ -0,0 +1,9 @@
+esphome:
+  name: hosttest
+
+host:
+
+network:
+  enable_ipv6: true
+
+api:

tests/component_tests/network/config/ipv4_and_ipv6_both_required.yaml

@@ -0,0 +1,12 @@
+esphome:
+  name: test
+
+esp32:
+  variant: esp32c6
+  framework:
+    type: esp-idf
+
+udp:
+
+openthread:
+  tlv: 0e080000000000010000000300001035060004001fffe00208e227ac6a7f24052f0708fdb753eb517cb4d3051062b2442a928d9ea3b947a1618fc4085a030f4f70656e5468726561642d393837330102987304105330d857354330133c05e1fd7ae81a910c0402a0f7f8

tests/component_tests/network/config/ipv4_required_with_ipv6.yaml

@@ -0,0 +1,14 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+network:
+  enable_ipv6: true
+
+wifi:
+  ssid: "test_ssid"
+  password: "test_password"

tests/component_tests/network/config/ipv6_off_nothing_required.yaml

@@ -0,0 +1,10 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+network:
+  enable_ipv6: false

tests/component_tests/network/config/ipv6_only.yaml

@@ -0,0 +1,10 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+network:
+  enable_ipv6: true

tests/component_tests/network/config/ipv6_only_ipv4_explicit_disable.yaml

@@ -0,0 +1,11 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+network:
+  enable_ipv6: true
+  enable_ipv4: false

tests/component_tests/network/config/nrf52_network_no_openthread.yaml

@@ -0,0 +1,10 @@
+esphome:
+  name: nrf52test
+
+nrf52:
+  board: adafruit_itsybitsy_nrf52840
+  bootloader: adafruit_nrf52_sd140_v6
+
+network:
+
+api:

tests/component_tests/network/config/nrf52_no_network.yaml

@@ -0,0 +1,6 @@
+esphome:
+  name: platformtest
+
+nrf52:
+  board: adafruit_itsybitsy_nrf52840
+  bootloader: adafruit_nrf52_sd140_v6

tests/component_tests/network/config/openthread_ipv4_explicitly_on.yaml

@@ -0,0 +1,13 @@
+esphome:
+  name: test
+
+esp32:
+  variant: esp32c6
+  framework:
+    type: esp-idf
+
+network:
+  enable_ipv4: true
+
+openthread:
+  tlv: 0e080000000000010000000300001035060004001fffe00208e227ac6a7f24052f0708fdb753eb517cb4d3051062b2442a928d9ea3b947a1618fc4085a030f4f70656e5468726561642d393837330102987304105330d857354330133c05e1fd7ae81a910c0402a0f7f8

tests/component_tests/network/config/openthread_no_explicit_network.yaml

@@ -0,0 +1,10 @@
+esphome:
+  name: test
+
+esp32:
+  variant: esp32c6
+  framework:
+    type: esp-idf
+
+openthread:
+  tlv: 0e080000000000010000000300001035060004001fffe00208e227ac6a7f24052f0708fdb753eb517cb4d3051062b2442a928d9ea3b947a1618fc4085a030f4f70656e5468726561642d393837330102987304105330d857354330133c05e1fd7ae81a910c0402a0f7f8

tests/component_tests/network/config/openthread_with_wake_on_lan.yaml

@@ -0,0 +1,15 @@
+esphome:
+  name: test
+
+esp32:
+  variant: esp32c6
+  framework:
+    type: esp-idf
+
+openthread:
+  tlv: 0e080000000000010000000300001035060004001fffe00208e227ac6a7f24052f0708fdb753eb517cb4d3051062b2442a928d9ea3b947a1618fc4085a030f4f70656e5468726561642d393837330102987304105330d857354330133c05e1fd7ae81a910c0402a0f7f8
+
+button:
+  - platform: wake_on_lan
+    name: Wake PC
+    target_mac_address: AA:BB:CC:DD:EE:FF

tests/component_tests/network/config/rp2_no_network.yaml

@@ -0,0 +1,5 @@
+esphome:
+  name: platformtest
+
+rp2:
+  board: rpipicow

tests/component_tests/network/config/tcp_uart_ipv4_disabled_rejected.yaml

@@ -0,0 +1,18 @@
+esphome:
+  name: test
+
+esp32:
+  variant: esp32c6
+  framework:
+    type: esp-idf
+
+openthread:
+  tlv: 0e080000000000010000000300001035060004001fffe00208e227ac6a7f24052f0708fdb753eb517cb4d3051062b2442a928d9ea3b947a1618fc4085a030f4f70656e5468726561642d393837330102987304105330d857354330133c05e1fd7ae81a910c0402a0f7f8
+
+network:
+  enable_ipv4: false
+
+tcp_uart:
+  - id: tcp_uart_server
+    role: server
+    port: 5020

tests/component_tests/network/config/uart_tcp_ipv4_disabled_rejected.yaml

@@ -0,0 +1,25 @@
+esphome:
+  name: test
+
+esp32:
+  variant: esp32c6
+  framework:
+    type: esp-idf
+
+openthread:
+  tlv: 0e080000000000010000000300001035060004001fffe00208e227ac6a7f24052f0708fdb753eb517cb4d3051062b2442a928d9ea3b947a1618fc4085a030f4f70656e5468726561642d393837330102987304105330d857354330133c05e1fd7ae81a910c0402a0f7f8
+
+network:
+  enable_ipv4: false
+
+uart:
+  id: uart_bus
+  tx_pin: 4
+  rx_pin: 5
+  baud_rate: 115200
+
+uart_tcp:
+  - id: bridge
+    uart_id: uart_bus
+    role: server
+    port: 502

tests/component_tests/network/config/udp_ipv4_disabled_rejected.yaml

@@ -0,0 +1,16 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+network:
+  enable_ipv6: true
+  enable_ipv4: false
+
+udp:
+  id: my_udp
+  listen_address: 239.0.60.53
+  addresses: ["239.0.60.53"]

tests/component_tests/network/config/wake_on_lan_ipv4_disabled_rejected.yaml

@@ -0,0 +1,18 @@
+esphome:
+  name: test
+
+esp32:
+  variant: esp32c6
+  framework:
+    type: esp-idf
+
+openthread:
+  tlv: 0e080000000000010000000300001035060004001fffe00208e227ac6a7f24052f0708fdb753eb517cb4d3051062b2442a928d9ea3b947a1618fc4085a030f4f70656e5468726561642d393837330102987304105330d857354330133c05e1fd7ae81a910c0402a0f7f8
+
+network:
+  enable_ipv4: false
+
+button:
+  - platform: wake_on_lan
+    name: Wake PC
+    target_mac_address: AA:BB:CC:DD:EE:FF

tests/component_tests/network/config/wifi_ipv4_disabled_rejected.yaml

@@ -0,0 +1,15 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+network:
+  enable_ipv6: true
+  enable_ipv4: false
+
+wifi:
+  ssid: "test_ssid"
+  password: "test_password"

tests/component_tests/network/test_network.py

@@ -0,0 +1,191 @@
+"""Tests for network component."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+from voluptuous import Invalid
+
+from esphome.components.network import (
+    CONF_ENABLE_IPV4,
+    _final_validate,
+    _get_data,
+    request_ipv4_off,
+    require_ipv4,
+    require_ipv6,
+)
+from esphome.config import load_config
+from esphome.const import CONF_ENABLE_IPV6, KEY_FRAMEWORK_VERSION, PlatformFramework
+from esphome.core import CORE, Version
+from tests.component_tests.types import SetCoreConfigCallable
+
+
+def _define_value(name: str) -> str | None:
+    define = next((d for d in CORE.defines if d.name == name), None)
+    return None if define is None else str(define.value)
+
+
+def test_require_ipv4_records_name_and_passes_config_through() -> None:
+    """require_ipv4() returns the config unchanged and records the caller once."""
+    config = {"foo": "bar"}
+    validator = require_ipv4("foo")
+    assert validator(config) is config
+    validator({})
+    assert _get_data().require_ipv4 == {"foo"}
+
+
+def test_require_ipv6_and_request_ipv4_off_record_state() -> None:
+    """require_ipv6() and request_ipv4_off() record their callers."""
+    require_ipv6("bar")({})
+    request_ipv4_off("bar")({})
+    data = _get_data()
+    assert data.require_ipv6 == {"bar"}
+    assert data.request_ipv4_off == {"bar"}
+
+
+@pytest.mark.parametrize(
+    ("config_file", "ipv4", "ipv6"),
+    [
+        ("ipv6_only_ipv4_explicit_disable.yaml", "false", "true"),
+        ("ipv6_only.yaml", "true", "true"),
+        ("ipv4_required_with_ipv6.yaml", "true", "true"),
+        ("wifi_only.yaml", "true", "false"),
+        ("openthread_no_explicit_network.yaml", "false", "true"),
+        ("openthread_ipv4_explicitly_on.yaml", "true", "true"),
+        ("ipv6_off_nothing_required.yaml", "true", "false"),
+        ("ipv4_and_ipv6_both_required.yaml", "true", "true"),
+        ("openthread_with_wake_on_lan.yaml", "true", "true"),
+        ("esp32_arduino_ipv6_no_ipv4_requirement.yaml", "true", "true"),
+        ("host_ipv6_no_ipv4_requirement.yaml", "true", "true"),
+        ("nrf52_network_no_openthread.yaml", "false", "true"),
+    ],
+)
+def test_ip_family_defines(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+    config_file: str,
+    ipv4: str,
+    ipv6: str,
+) -> None:
+    """Each config resolves USE_NETWORK_IPV4 and USE_NETWORK_IPV6 as expected."""
+    generate_main(component_config_path(config_file))
+    assert _define_value("USE_NETWORK_IPV4") == ipv4
+    assert _define_value("USE_NETWORK_IPV6") == ipv6
+
+
+@pytest.mark.parametrize(
+    "config_file",
+    [
+        "wifi_ipv4_disabled_rejected.yaml",
+        "wake_on_lan_ipv4_disabled_rejected.yaml",
+        "udp_ipv4_disabled_rejected.yaml",
+        "tcp_uart_ipv4_disabled_rejected.yaml",
+        "uart_tcp_ipv4_disabled_rejected.yaml",
+        "esp8266_ipv4_disabled_rejected.yaml",
+        "esp32_arduino_ipv4_disabled_rejected.yaml",
+    ],
+)
+def test_require_ipv4_registration_rejects_explicit_disable(
+    component_config_path: Callable[[str], Path],
+    config_file: str,
+) -> None:
+    """A component that requires IPv4 rejects an explicit 'enable_ipv4: false'."""
+    CORE.config_path = component_config_path(config_file)
+    res = load_config({})
+    assert any("explicitly disables it" in str(err) for err in res.errors)
+
+
+@pytest.mark.parametrize(
+    ("config_file", "component"),
+    [
+        ("tcp_uart_ipv4_disabled_rejected.yaml", "tcp_uart"),
+        ("uart_tcp_ipv4_disabled_rejected.yaml", "uart_tcp"),
+    ],
+)
+def test_tcp_link_users_named_in_ipv4_error(
+    component_config_path: Callable[[str], Path],
+    config_file: str,
+    component: str,
+) -> None:
+    """The IPv4 requirement registered by the socket helper names the component."""
+    CORE.config_path = component_config_path(config_file)
+    res = load_config({})
+    assert any(f"IPv4 is required by {component}" in str(err) for err in res.errors)
+
+
+def test_final_validate_rejects_explicit_ipv6_disable_when_required() -> None:
+    """An explicit 'enable_ipv6: false' conflicting with require_ipv6() is rejected."""
+    require_ipv6("test")({})
+    with pytest.raises(Invalid, match="explicitly disables it"):
+        _final_validate({CONF_ENABLE_IPV6: False})
+
+
+def test_final_validate_accepts_explicit_ipv6_disable_when_not_required() -> None:
+    """An explicit 'enable_ipv6: false' is accepted when nothing requires IPv6."""
+    _final_validate({CONF_ENABLE_IPV6: False})
+
+
+def test_final_validate_resolves_ipv6_default_when_required(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    """require_ipv6() turns an absent 'enable_ipv6' on."""
+    set_core_config(
+        PlatformFramework.HOST_NATIVE,
+        core_data={KEY_FRAMEWORK_VERSION: Version(0, 0, 0)},
+    )
+    require_ipv6("test")({})
+    config: dict[str, bool] = {}
+    _final_validate(config)
+    assert config[CONF_ENABLE_IPV6] is True
+
+
+def test_final_validate_rejects_explicit_ipv4_disable_when_required() -> None:
+    """An explicit 'enable_ipv4: false' conflicting with require_ipv4() is rejected."""
+    require_ipv4("test")({})
+    with pytest.raises(Invalid, match="explicitly disables it"):
+        _final_validate({CONF_ENABLE_IPV4: False})
+
+
+def test_final_validate_accepts_explicit_ipv4_disable_when_not_required() -> None:
+    """An explicit 'enable_ipv4: false' is accepted when IPv6 is on."""
+    _final_validate({CONF_ENABLE_IPV4: False, CONF_ENABLE_IPV6: True})
+
+
+def test_final_validate_accepts_ipv4_default_when_required() -> None:
+    """An explicit 'enable_ipv4: true' never conflicts with require_ipv4()."""
+    require_ipv4("test")({})
+    _final_validate({CONF_ENABLE_IPV4: True})
+
+
+@pytest.mark.parametrize(
+    "config",
+    [{CONF_ENABLE_IPV4: False}, {CONF_ENABLE_IPV4: False, CONF_ENABLE_IPV6: False}],
+)
+def test_final_validate_rejects_both_address_families_disabled(
+    config: dict[str, bool],
+) -> None:
+    """Disabling IPv4 with IPv6 off leaves no IP stack and is rejected."""
+    with pytest.raises(Invalid, match="leaves no IP stack"):
+        _final_validate(config)
+
+
+@pytest.mark.parametrize(
+    "config_file",
+    [
+        "esp8266_no_network.yaml",
+        "rp2_no_network.yaml",
+        "bk72xx_no_network.yaml",
+        "esp32_idf_no_network.yaml",
+        "esp32_arduino_no_network.yaml",
+        "nrf52_no_network.yaml",
+    ],
+)
+def test_no_ip_defines_without_network_component(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+    config_file: str,
+) -> None:
+    """Platform IP requests emit no IP family defines when network is not loaded."""
+    generate_main(component_config_path(config_file))
+    assert _define_value("USE_NETWORK_IPV4") is None
+    assert _define_value("USE_NETWORK_IPV6") is None

tests/component_tests/openthread/test_openthread.py

@@ -182,10 +182,6 @@ def test_border_router_rcp_codegen(
             lambda config, _: config[CONF_WIFI].update({CONF_AP: {}}),
             "does not support Wi-Fi AP",
         ),
-        (
-            lambda config, _: config[CONF_NETWORK].update({CONF_ENABLE_IPV6: False}),
-            "requires IPv6 to be enabled",
-        ),
     ],
 )
 def test_border_router_rejects_incompatible_config(

tests/component_tests/udp/test_udp_listen_address.py

@@ -4,12 +4,14 @@ from ipaddress import ip_address
 
 import pytest
 
+from esphome.components import network
 from esphome.components.udp import (
     CONF_ADDRESSES,
     CONF_LISTEN_ADDRESS,
     _address,
     _final_validate,
     _listen_address,
+    _require_ipv6_for_ipv6_addresses,
 )
 import esphome.config_validation as cv
 from esphome.const import CONF_ENABLE_IPV6, PlatformFramework
@@ -61,29 +63,12 @@ _RAW_LWIP_IPV6 = {
     "network": {CONF_ENABLE_IPV6: True},
     "socket": {"implementation": "lwip_tcp"},
 }
-_NO_IPV6 = "requires 'enable_ipv6: true'"
 _RAW_LWIP = "not supported on this platform"
 
 
 @pytest.mark.parametrize(
     ("addresses", "listen_address", "platform", "full_config", "message"),
     [
-        pytest.param(
-            ["ff12::cafe"],
-            "255.255.255.255",
-            PlatformFramework.ESP32_IDF,
-            {},
-            _NO_IPV6,
-            id="ipv6_destination_without_ipv6",
-        ),
-        pytest.param(
-            ["255.255.255.255"],
-            "ff12::cafe",
-            PlatformFramework.ESP32_IDF,
-            {},
-            _NO_IPV6,
-            id="ipv6_listen_without_ipv6",
-        ),
         pytest.param(
             ["ff12::cafe"],
             "255.255.255.255",
@@ -120,6 +105,31 @@ def test_final_validate_rejects_unsupported_ipv6(
         )
 
 
+@pytest.mark.parametrize(
+    ("addresses", "listen_address", "required"),
+    [
+        (["ff12::cafe"], "255.255.255.255", True),
+        (["255.255.255.255"], "ff12::cafe", True),
+        (["255.255.255.255"], "255.255.255.255", False),
+    ],
+)
+def test_ipv6_address_requires_ipv6_from_network(
+    set_core_config: SetCoreConfigCallable,
+    addresses: list[str],
+    listen_address: str,
+    required: bool,
+) -> None:
+    """An IPv6 destination or listen address registers udp with network.require_ipv6()."""
+    set_core_config(PlatformFramework.ESP32_IDF)
+    _require_ipv6_for_ipv6_addresses(
+        {
+            CONF_ADDRESSES: [ip_address(a) for a in addresses],
+            CONF_LISTEN_ADDRESS: ip_address(listen_address),
+        }
+    )
+    assert ("udp" in network._get_data().require_ipv6) is required
+
+
 def test_final_validate_accepts_ipv6_when_enabled(
     set_core_config: SetCoreConfigCallable,
 ) -> None:

tests/components/api/test-outgoing-connection-ipv6.esp32-c6-idf.yaml

@@ -0,0 +1,12 @@
+packages:
+  common: !include common.yaml
+
+# OpenThread turns IPv4 off, so the dial path builds without any IPv4 code
+openthread:
+  tlv: 0e080000000000010000000300001035060004001fffe00208e227ac6a7f24052f0708fdb753eb517cb4d3051062b2442a928d9ea3b947a1618fc4085a030f4f70656e5468726561642d393837330102987304105330d857354330133c05e1fd7ae81a910c0402a0f7f8
+
+api:
+  outgoing_connection:
+    host: fd53:145f:ed22:ad81::1
+    port: 6054
+    delay: 60s

tests/components/network/test-ipv6.esp32-ard.yaml

@@ -0,0 +1,5 @@
+substitutions:
+  network_enable_ipv6: "true"
+
+packages:
+  network: !include common.yaml

tests/components/network/test.esp32-c6-idf.yaml

@@ -0,0 +1,12 @@
+esp32:
+  framework:
+    log_level: DEBUG
+
+network:
+  enable_ipv6: true
+  enable_ipv4: false
+
+api:
+
+openthread:
+  tlv: 0e080000000000010000000300001035060004001fffe00208e227ac6a7f24052f0708fdb753eb517cb4d3051062b2442a928d9ea3b947a1618fc4085a030f4f70656e5468726561642d393837330102987304105330d857354330133c05e1fd7ae81a910c0402a0f7f8

tests/components/network/test.nrf52-adafruit.yaml

@@ -1,5 +1,6 @@
 network:
   enable_ipv6: true
+  enable_ipv4: false
 
 openthread:
   tlv: 0E080000000000010000

tests/components/network/test.nrf52-mcumgr.yaml

@@ -1,5 +1,6 @@
 network:
   enable_ipv6: true
+  enable_ipv4: true
 
 openthread:
   tlv: 0E080000000000010000

tests/integration/README.md

@@ -45,7 +45,7 @@ The `InitialStateHelper` class solves a common problem in integration tests: whe
 - Tracks all entities (except stateless ones like buttons)
 - Swallows the first state broadcast for each entity
 - Forwards all subsequent state changes to your test callback
-- Provides `wait_for_initial_states()` to synchronize before test actions
+- Provides `wait_for_initial_states()` to synchronize before test actions; it fails the test on timeout, naming the entities still without a state
 
 **When to use it:**
 - Any test that triggers entity state changes and needs to verify them
@@ -232,11 +232,9 @@ initial_state_helper = InitialStateHelper(entities)
 # Subscribe with the wrapper that filters initial states
 client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
-# Wait for all initial states to be broadcast
-try:
-    await initial_state_helper.wait_for_initial_states()
-except TimeoutError:
-    pytest.fail("Timeout waiting for initial states")
+# Wait for all initial states to be broadcast; on timeout this fails the test
+# and lists the entities that never sent one
+await initial_state_helper.wait_for_initial_states()
 
 # Now perform your test actions - on_state will only receive new changes
 # ... trigger state changes ...

tests/integration/fixtures/uart_mock_modbus_broadcast_write.yaml

@@ -139,7 +139,7 @@ button:
     on_press:
       - lambda: |-
           // Start polling now that the test has subscribed.
-          id(modbus_controller_1).set_update_interval(1000);
+          id(modbus_controller_1).set_update_interval(300);
           id(modbus_controller_1).start_poller();
           // Broadcast (address 0) write single register: reg 0x10 = 777 on every server.
           // PDU is function code + data (no address/CRC); the hub prepends address 0 and appends CRC.

tests/integration/fixtures/uart_mock_modbus_client_inline.yaml

@@ -46,7 +46,7 @@ modbus:
     role: client
     turnaround_time: 10ms
     # Short wait so the no-reply cases (address 2 below) time out well within the test window.
-    send_wait_time: 500ms
+    send_wait_time: 200ms
 
 modbus_server:
   - address: 1

tests/integration/fixtures/uart_mock_modbus_loopback.yaml

@@ -75,7 +75,7 @@ modbus_controller:
   - address: 1
     modbus_id: virtual_modbus_controller
     id: modbus_controller_1
-    update_interval: 1s
+    update_interval: 500ms
 
 modbus_server:
   - address: 1

tests/integration/state_utils.py

@@ -19,6 +19,8 @@ from aioesphomeapi import (
     UserService,
 )
 
+from .wait_utils import wait_or_fail
+
 _LOGGER = logging.getLogger(__name__)
 
 S = TypeVar("S", bound=EntityState)
@@ -28,6 +30,7 @@ async def wait_for_state(
     client: APIClient,
     predicate: Callable[[EntityState], bool],
     timeout: float = 5.0,
+    msg: str = "Timed out waiting for a matching state",
 ) -> EntityState:
     """Subscribe to states and wait for one matching ``predicate``.
 
@@ -42,21 +45,24 @@ async def wait_for_state(
         predicate: Callable invoked for every received state; the first
             state for which it returns ``True`` is returned.
         timeout: Maximum time to wait in seconds.
+        msg: Failure message if no matching state arrives in time; the
+            states received are listed after it.
 
     Returns:
         The first state matching ``predicate``.
-
-    Raises:
-        asyncio.TimeoutError: If no matching state arrives within ``timeout``.
     """
     future: asyncio.Future[EntityState] = asyncio.get_running_loop().create_future()
+    received: list[EntityState] = []
 
     def on_state(state: EntityState) -> None:
-        if not future.done() and predicate(state):
+        if future.done():
+            return
+        received.append(state)
+        if predicate(state):
             future.set_result(state)
 
     client.subscribe_states(on_state)
-    return await asyncio.wait_for(future, timeout=timeout)
+    return await wait_or_fail(future, timeout, msg, received=received)
 
 
 class StateWaiter:
@@ -377,15 +383,23 @@ class InitialStateHelper:
         return wrapper
 
     async def wait_for_initial_states(self, timeout: float = 5.0) -> None:
-        """Wait for all initial states to be received.
+        """Wait for all initial states to be received, failing the test on timeout.
 
         Args:
             timeout: Maximum time to wait in seconds
-
-        Raises:
-            asyncio.TimeoutError: If initial states aren't received within timeout
         """
-        await asyncio.wait_for(self._initial_states_received, timeout=timeout)
+        await wait_or_fail(
+            self._initial_states_received,
+            timeout,
+            "Timed out waiting for initial states",
+            still_waiting=lambda: sorted(
+                f"{type(entity).__name__} {entity.object_id}"
+                for entity in (
+                    self._entities_by_id[entity_id]
+                    for entity_id in self._wait_initial_states
+                )
+            ),
+        )
 
 
 class SensorStateCollector:
@@ -471,12 +485,29 @@ class SensorStateCollector:
         )
 
     async def wait_for_all(self, timeout: float = 3.0) -> None:
-        """Wait until all sensors and binary sensors have at least one value."""
+        """Wait until every tracked sensor has a value, failing the test on timeout."""
         if self._all_have_values():
             return
         future: asyncio.Future[bool] = asyncio.get_running_loop().create_future()
         self._waiters.append((self._all_have_values, future))
-        await asyncio.wait_for(future, timeout=timeout)
+        tracked = (self.sensor_states, self.binary_states, self.text_sensor_states)
+        await wait_or_fail(
+            future,
+            timeout,
+            "Timed out waiting for every sensor to report",
+            still_waiting=lambda: [
+                name
+                for states in tracked
+                for name, values in states.items()
+                if not values
+            ],
+            received=lambda: {
+                name: values
+                for states in tracked
+                for name, values in states.items()
+                if values
+            },
+        )
 
     def add_waiter(self, condition: Callable[[], bool]) -> asyncio.Future[bool]:
         """Add a custom waiter that resolves when condition returns True.
@@ -613,12 +644,7 @@ class SensorTracker:
         )
         initial_state_helper = InitialStateHelper(entities)
         client.subscribe_states(initial_state_helper.on_state_wrapper(self.on_state))
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            import pytest
-
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
         if match_initial_states:
             for state in initial_state_helper.initial_states.values():
                 self.on_state(state, first_pending_only=True)

tests/integration/test_binary_sensor_autorepeat_filter.py

@@ -58,10 +58,7 @@ async def test_binary_sensor_autorepeat_filter(
             initial_state_helper.on_state_wrapper(collector.on_state)
         )
 
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         autorepeat_states = collector.binary_states["autorepeat_sensor"]
 

tests/integration/test_binary_sensor_invalidate_state.py

@@ -73,10 +73,7 @@ async def test_binary_sensor_invalidate_state(
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
         # Wait for initial states
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Check initial state - should be missing (unknown)
         initial_state = initial_state_helper.initial_states.get(binary_sensor.key)

tests/integration/test_climate_custom_modes.py

@@ -55,10 +55,7 @@ async def test_climate_custom_fan_modes_and_presets(
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
         # Wait for all initial states to be broadcast
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Verify enum presets are exposed (from preset: config map)
         assert ClimatePreset.AWAY in test_climate.supported_presets, (

tests/integration/test_entity_missing_state.py

@@ -79,10 +79,7 @@ async def test_entity_missing_state(
         initial_state_helper = InitialStateHelper(entities)
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Nothing has published yet, so all of them must report unknown
         for info in stateful:

tests/integration/test_host_mode_climate_basic_state.py

@@ -45,7 +45,8 @@ async def test_host_mode_climate_basic_state(
                 and state.target_humidity == 20.0
             )
 
-        try:
-            await wait_for_state(client, is_default_state)
-        except TimeoutError:
-            pytest.fail("Climate did not converge to expected default state")
+        await wait_for_state(
+            client,
+            is_default_state,
+            msg="Climate did not converge to expected default state",
+        )

tests/integration/test_host_mode_climate_control.py

@@ -46,10 +46,7 @@ async def test_host_mode_climate_control(
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
         # Wait for all initial states to be broadcast
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Adjust setpoints
         client.climate_command(

tests/integration/test_light_constant_brightness.py

@@ -79,10 +79,7 @@ async def test_light_constant_brightness(
 
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         async def send_and_wait(
             light_key: int, timeout: float = 5.0, **kwargs: Any

tests/integration/test_sensor_filters_ring_buffer.py

@@ -84,10 +84,7 @@ async def test_sensor_filters_ring_buffer(
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
         # Wait for initial states to be sent before pressing button
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Find the publish button
         publish_button = require_entity(entities, "publish_values_button")

tests/integration/test_sensor_filters_sliding_window.py

@@ -116,10 +116,7 @@ async def test_sensor_filters_sliding_window(
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
         # Wait for initial states to be sent before pressing button
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Find the publish button
         publish_button = require_entity(entities, "publish_values_button")
@@ -241,10 +238,7 @@ async def test_sensor_filters_nan_handling(
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
         # Wait for initial states
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Find the publish button
         publish_button = require_entity(entities, "publish_nan_values_button")
@@ -336,10 +330,7 @@ async def test_sensor_filters_ring_buffer_wraparound(
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
         # Wait for initial state
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial state")
+        await initial_state_helper.wait_for_initial_states()
 
         # Find the publish button
         publish_button = require_entity(entities, "publish_wraparound_button")

tests/integration/test_sensor_timeout_filter.py

@@ -90,10 +90,7 @@ async def test_sensor_timeout_filter(
         initial_state_helper = InitialStateHelper(entities)
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Helper to find buttons by object_id substring
         def find_button(object_id_substring: str) -> int:

tests/integration/test_status_flags.py

@@ -71,10 +71,7 @@ async def test_status_flags(
         # state changes triggered by our service calls.
         initial_state_helper = InitialStateHelper(entities)
         client.subscribe_states(initial_state_helper.on_state_wrapper(tracker.on_state))
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         async def call(name: str) -> None:
             await client.execute_service(svc[name], {})

tests/integration/test_strftime_to.py

@@ -61,10 +61,7 @@ async def test_strftime_to(
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
         # Wait for initial states to be broadcast
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Wait for all expected states
         await wait_or_fail(

tests/integration/test_template_climate_basic.py

@@ -88,10 +88,7 @@ async def test_template_climate_basic(
         client.subscribe_states(
             initial_state_helper.on_state_wrapper(lambda state: None)
         )
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         initial = initial_state_helper.initial_states.get(test_climate.key)
         assert initial is not None, "No initial climate state received"

tests/integration/test_template_climate_custom_modes.py

@@ -71,10 +71,7 @@ async def test_template_climate_custom_modes(
         client.subscribe_states(
             initial_state_helper.on_state_wrapper(lambda state: None)
         )
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         initial = initial_state_helper.initial_states.get(test_climate.key)
         assert initial is not None, "No initial climate state received"

tests/integration/test_template_climate_nonoptimistic.py

@@ -66,10 +66,7 @@ async def test_template_climate_nonoptimistic(
         )
 
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         initial = initial_state_helper.initial_states.get(test_climate.key)
         assert initial is not None, "No initial climate state received"

tests/integration/test_template_climate_on_control_ordering.py

@@ -54,10 +54,7 @@ async def test_template_climate_on_control_ordering(
         client.subscribe_states(
             initial_state_helper.on_state_wrapper(lambda state: None)
         )
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         client.climate_command(test_climate.key, mode=ClimateMode.HEAT)
         state = await wait_for_climate_state()

tests/integration/test_template_climate_publish_all_fields.py

@@ -68,18 +68,14 @@ async def test_template_climate_publish_all_fields(
         publish_button = require_entity(entities, "publish_all", ButtonInfo)
 
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         client.button_command(publish_button.key)
-        try:
-            state = await wait_for_state(
-                client, lambda s: isinstance(s, aioesphomeapi.ClimateState)
-            )
-        except TimeoutError:
-            pytest.fail("Timeout waiting for the published climate state")
+        state = await wait_for_state(
+            client,
+            lambda s: isinstance(s, aioesphomeapi.ClimateState),
+            msg="Timeout waiting for the published climate state",
+        )
 
         assert state.current_temperature == pytest.approx(20.0, abs=0.1)
         assert state.current_humidity == pytest.approx(60.0, abs=0.1)

tests/integration/test_template_climate_sensor_push.py

@@ -57,10 +57,7 @@ async def test_template_climate_sensor_push(
         publish_humidity = require_entity(entities, "publish_humidity", ButtonInfo)
 
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         initial = initial_state_helper.initial_states.get(test_climate.key)
         assert initial is not None, "No initial climate state received"

tests/integration/test_template_climate_set_actions.py

@@ -65,10 +65,7 @@ async def test_template_climate_set_actions(
         client.subscribe_states(
             initial_state_helper.on_state_wrapper(lambda state: None)
         )
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Both traits are derived from the low/high and humidity set actions, not declared.
         assert test_climate.supports_two_point_target_temperature

tests/integration/test_template_climate_two_point_temperature.py

@@ -66,10 +66,7 @@ async def test_template_climate_two_point_temperature(
             initial_state_helper.on_state_wrapper(lambda state: None)
         )
 
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         initial = initial_state_helper.initial_states.get(test_climate.key)
         assert initial is not None, "No initial climate state received"

tests/integration/test_text_command.py

@@ -55,10 +55,7 @@ async def test_text_command(
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
         # Wait for all initial states to be received
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Verify initial states were received
         assert test_text.key in initial_state_helper.initial_states

tests/integration/test_uart_mock_ld2410.py

@@ -90,24 +90,14 @@ async def test_uart_mock_ld2410(
             initial_state_helper.on_state_wrapper(collector.on_state)
         )
 
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)
 
         # Wait for Phase 1 - all sensors and binary sensors have at least one value
-        try:
-            await collector.wait_for_all(timeout=3.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Timeout waiting for Phase 1 frame. Received:\n"
-                f"  sensor_states: {collector.sensor_states}\n"
-                f"  binary_states: {collector.binary_states}"
-            )
+        await collector.wait_for_all(timeout=3.0)
 
         # Phase 1 values: moving=100, still=120, energy=50/25, detect=300
         assert collector.sensor_states["moving_distance"][0] == pytest.approx(100.0)
@@ -214,24 +204,14 @@ async def test_uart_mock_ld2410_engineering(
             initial_state_helper.on_state_wrapper(collector.on_state)
         )
 
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)
 
         # Wait for Phase 1 - all sensors and binary sensors have at least one value
-        try:
-            await collector.wait_for_all(timeout=3.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Timeout waiting for Phase 1 frame. Received:\n"
-                f"  sensor_states: {collector.sensor_states}\n"
-                f"  binary_states: {collector.binary_states}"
-            )
+        await collector.wait_for_all(timeout=3.0)
 
         # Phase 1 values (engineering mode frame):
         # moving=30, energy=100, still=30, energy=100, detect=0

tests/integration/test_uart_mock_ld2410_query_ack.py

@@ -119,10 +119,7 @@ async def test_uart_mock_ld2410_query_ack(
         initial_state_helper = InitialStateHelper(entities)
         client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
 
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)

tests/integration/test_uart_mock_ld2410s.py

@@ -117,10 +117,7 @@ async def _start_scenario(client: APIClient, collector: SensorStateCollector) ->
     collector.build_key_mapping(entities)
     initial_state_helper = InitialStateHelper(entities)
     client.subscribe_states(initial_state_helper.on_state_wrapper(collector.on_state))
-    try:
-        await initial_state_helper.wait_for_initial_states()
-    except TimeoutError:
-        pytest.fail("Timeout waiting for initial states")
+    await initial_state_helper.wait_for_initial_states()
     start_btn = require_entity(entities, "start_scenario", ButtonInfo)
     client.button_command(start_btn.key)
 

tests/integration/test_uart_mock_ld2412.py

@@ -89,10 +89,7 @@ async def _wait_initial(
     """Subscribe to states and wait until every entity has sent its initial state."""
     initial_state_helper = InitialStateHelper(entities)
     client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-    try:
-        await initial_state_helper.wait_for_initial_states()
-    except TimeoutError:
-        pytest.fail("Timeout waiting for initial states")
+    await initial_state_helper.wait_for_initial_states()
     return initial_state_helper
 
 
@@ -197,14 +194,7 @@ async def test_uart_mock_ld2412(
         client.button_command(start_btn.key)
 
         # Wait for Phase 1 - all sensors and binary sensors have at least one value
-        try:
-            await collector.wait_for_all(timeout=3.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Timeout waiting for Phase 1 frame. Received:\n"
-                f"  sensor_states: {collector.sensor_states}\n"
-                f"  binary_states: {collector.binary_states}"
-            )
+        await collector.wait_for_all(timeout=3.0)
 
         # Phase 1 values: moving=100, still=120, energy=50/25, detect=100
         assert collector.sensor_states["moving_distance"][0] == pytest.approx(100.0)
@@ -303,14 +293,7 @@ async def test_uart_mock_ld2412_engineering(
         client.button_command(start_btn.key)
 
         # Wait for Phase 1 - all sensors and binary sensors have at least one value
-        try:
-            await collector.wait_for_all(timeout=3.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Timeout waiting for Phase 1 frame. Received:\n"
-                f"  sensor_states: {collector.sensor_states}\n"
-                f"  binary_states: {collector.binary_states}"
-            )
+        await collector.wait_for_all(timeout=3.0)
 
         # Phase 1 values (engineering mode frame):
         # moving=30, energy=100, still=30, energy=100, detect=30
@@ -406,14 +389,7 @@ async def test_uart_mock_ld2412_engineering_truncated(
         client.button_command(start_btn.key)
 
         # Wait for Phase 1 — valid engineering frame establishes baseline
-        try:
-            await collector.wait_for_all(timeout=3.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Timeout waiting for Phase 1 frame. Received:\n"
-                f"  sensor_states: {collector.sensor_states}\n"
-                f"  binary_states: {collector.binary_states}"
-            )
+        await collector.wait_for_all(timeout=3.0)
 
         # Phase 1 baseline: gate_0_move=100, light=87
         assert collector.sensor_states["gate_0_move_energy"][0] == pytest.approx(100.0)

tests/integration/test_uart_mock_ld2420.py

@@ -81,24 +81,14 @@ async def test_uart_mock_ld2420(
             initial_state_helper.on_state_wrapper(collector.on_state)
         )
 
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)
 
         # Wait for Phase 1 - all sensors and binary sensors have at least one value
-        try:
-            await collector.wait_for_all(timeout=3.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Timeout waiting for Phase 1 frame. Received:\n"
-                f"  sensor_states: {collector.sensor_states}\n"
-                f"  binary_states: {collector.binary_states}"
-            )
+        await collector.wait_for_all(timeout=3.0)
 
         # Phase 1 values: moving=100, has_target=true
         assert collector.sensor_states["moving_distance"][0] == pytest.approx(100.0)
@@ -191,24 +181,14 @@ async def test_uart_mock_ld2420_simple(
             initial_state_helper.on_state_wrapper(collector.on_state)
         )
 
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)
 
         # Wait for Phase 1 - all sensors and binary sensors have at least one value
-        try:
-            await collector.wait_for_all(timeout=3.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Timeout waiting for Phase 1 frame. Received:\n"
-                f"  sensor_states: {collector.sensor_states}\n"
-                f"  binary_states: {collector.binary_states}"
-            )
+        await collector.wait_for_all(timeout=3.0)
 
         # Phase 1: simple mode "ON Range 0100\r\n" → distance=100, presence=true
         assert collector.sensor_states["moving_distance"][0] == pytest.approx(100.0)

tests/integration/test_uart_mock_ld2420_config_exit.py

@@ -51,10 +51,7 @@ async def test_uart_mock_ld2420_config_exit(
         client.subscribe_states(
             initial_state_helper.on_state_wrapper(collector.on_state)
         )
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)

tests/integration/test_uart_mock_ld2450.py

@@ -103,25 +103,14 @@ async def test_uart_mock_ld2450(
             initial_state_helper.on_state_wrapper(collector.on_state)
         )
 
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         # Start the UART mock scenario now that we're subscribed
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)
 
         # Wait for Phase 1 - all sensors and binary sensors have at least one value
-        try:
-            await collector.wait_for_all(timeout=5.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Timeout waiting for Phase 1 frame. Received:\n"
-                f"  sensor_states: {collector.sensor_states}\n"
-                f"  binary_states: {collector.binary_states}\n"
-                f"  text_states: {collector.text_sensor_states}"
-            )
+        await collector.wait_for_all(timeout=5.0)
 
         # Phase 1 values:
         # Target 1: X=-500, Y=1000, Speed=-50 (approaching), Res=320

tests/integration/test_uart_mock_ld2460.py

@@ -85,10 +85,7 @@ async def test_uart_mock_ld2460(
         client.subscribe_states(
             initial_state_helper.on_state_wrapper(collector.on_state)
         )
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+        await initial_state_helper.wait_for_initial_states()
 
         start_btn = require_entity(entities, "start_scenario", ButtonInfo)
         client.button_command(start_btn.key)

tests/integration/test_uart_mock_modbus.py

@@ -23,6 +23,7 @@ from collections.abc import Callable
 from aioesphomeapi import ButtonInfo, NumberInfo, SwitchInfo, TextSensorState
 import pytest
 
+from .log_utils import LineWaiter
 from .state_utils import SensorTracker, require_entity, wait_for_state
 from .types import APIClientConnectedFactory, RunCompiledFunction
 from .wait_utils import wait_or_fail
@@ -79,6 +80,44 @@ def _make_modbus_line_callback() -> tuple[Callable[[str], None], list[str], list
     return line_callback, error_log_lines, warning_log_lines
 
 
+async def _wait_until(
+    tracker: SensorTracker,
+    names: list[str],
+    predicate: Callable[[dict[str, list[float]]], bool],
+) -> None:
+    """Return once *predicate* holds, re-checking whenever one of *names* gets a state."""
+    while not predicate(tracker.sensor_states):
+        futures = [tracker.expect_any(name) for name in names]
+        await asyncio.wait(futures, return_when=asyncio.FIRST_COMPLETED)
+        for future in futures:
+            future.cancel()
+
+
+async def _wait_for_reads(
+    tracker: SensorTracker, name: str, count: int, timeout: float = 5.0
+) -> None:
+    """Wait until *name* has received at least *count* states in total."""
+    await wait_or_fail(
+        _wait_until(tracker, [name], lambda states: len(states[name]) >= count),
+        timeout,
+        f"{name} did not reach {count} states",
+        states=tracker.sensor_states[name],
+    )
+
+
+async def _wait_until_quiet(
+    tracker: SensorTracker, names: list[str], quiet: float
+) -> None:
+    """Return once none of *names* has received a state for *quiet* seconds."""
+    while True:
+        futures = [tracker.expect_any(name) for name in names]
+        done, _ = await asyncio.wait(futures, timeout=quiet)
+        for future in futures:
+            future.cancel()
+        if not done:
+            return
+
+
 def _assert_no_modbus_errors(
     error_log_lines: list[str], warning_log_lines: list[str]
 ) -> None:
@@ -105,6 +144,7 @@ async def test_uart_mock_modbus(
 ) -> None:
     """Test basic modbus data parsing."""
 
+    waiter = LineWaiter()
     tracker = SensorTracker(
         [
             "basic_register",
@@ -123,20 +163,30 @@ async def test_uart_mock_modbus(
     exception_response_changed = tracker.expect_any("exception_response")
 
     async with (
-        run_compiled(yaml_config),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         await tracker.setup_and_start_scenario(client)
 
         await tracker.await_change(delayed_response_changed, "delayed_response")
         await tracker.await_change(basic_register_changed, "basic_register")
-        # Run all "must not change" checks concurrently — each waits the full
-        # timeout, so sequential execution would multiply the wall time.
+        # Wait until each failing read has resolved: devices 2 and 3 gave up, device 1 got the exception.
+        await waiter.wait_for_each(
+            "Modbus device=2 set offline",
+            "Modbus device=3 set offline",
+            "Error function code: 0x83 exception: 2, address: 1",
+            timeout=5.0,
+        )
+        # The late reply lands ~100ms after device 2 gives up; the window also covers the API batch delay.
         await asyncio.gather(
-            tracker.await_must_not_change(late_response_changed, "late_response"),
-            tracker.await_must_not_change(no_response_changed, "no_response"),
             tracker.await_must_not_change(
-                exception_response_changed, "exception_response"
+                late_response_changed, "late_response", timeout=0.5
+            ),
+            tracker.await_must_not_change(
+                no_response_changed, "no_response", timeout=0.5
+            ),
+            tracker.await_must_not_change(
+                exception_response_changed, "exception_response", timeout=0.5
             ),
         )
 
@@ -812,23 +862,19 @@ async def test_uart_mock_modbus_continuous(
     """
 
     tracker = SensorTracker(["continuous_reg"])
+    waiter = LineWaiter()
 
     async with (
-        run_compiled(yaml_config),
+        run_compiled(yaml_config, line_callback=waiter.callback),
         api_client_connected() as client,
     ):
         # setup_and_start_scenario presses the Start Scenario button, whose on_press triggers the
         # controller's first update(). With continuous that one read re-queues and streams; without it
         # the next poll would not run until the 30s update_interval elapses.
         entities = await tracker.setup_and_start_scenario(client)
-        # Count reads over a window far shorter than the update_interval. Absent continuous polling we
-        # would see ~1 (the triggered poll); continuous re-queues, so the bus fills with reads.
-        await asyncio.sleep(3.0)
-        reads = len(tracker.sensor_states["continuous_reg"])
-        assert reads >= 5, (
-            "expected many continuous reads within the window (update_interval is 30s, so absent "
-            f"continuous polling we would see ~1), got {reads}"
-        )
+        # Absent continuous polling we would see ~1 read (the triggered poll) within the 30s
+        # update_interval; continuous re-queues, so the bus fills with reads.
+        await _wait_for_reads(tracker, "continuous_reg", 5)
 
         # Recovery path: a live continuous poll that starts failing goes offline, and the next update()
         # re-arms it once the device answers again. Silence the server so the poll's reads time out; with
@@ -837,22 +883,35 @@ async def test_uart_mock_modbus_continuous(
         start = require_entity(entities, "start_scenario", ButtonInfo)
 
         client.switch_command(silence.key, True)
-        await asyncio.sleep(1.0)  # let the poll fail and the device trip offline
+        await waiter.wait_for("Modbus device=1 set offline", timeout=5.0)
         plateau = len(tracker.sensor_states["continuous_reg"])
-        await asyncio.sleep(1.0)  # offline: no polls should land
-        assert len(tracker.sensor_states["continuous_reg"]) == plateau, (
-            "reads kept arriving after the server was silenced - the failed continuous poll did not stop"
-        )
+
+        def timeouts() -> int:
+            return sum(
+                "Stop waiting for response from 1 " in line for line in waiter.lines
+            )
+
+        # The silenced server answers nothing, so count request timeouts, not states. Going offline
+        # still sends one probe (offline_skip_updates is 0) and its retry, each timing out after a bus
+        # wait of up to a second, so the count may grow twice more; a continuous poll that failed to
+        # stop would keep it growing every few hundred milliseconds instead of settling.
+        sent = timeouts()
+        assert sent, "no request timed out before the device went offline"
+        for _ in range(4):
+            await asyncio.sleep(1.5)
+            if timeouts() == sent:
+                break
+            sent = timeouts()
+        else:
+            pytest.fail(
+                "the failed continuous poll kept sending requests after the device went offline"
+            )
 
         # Answer again and trigger update(): the offline probe recovers the device and the continuous
         # poll re-arms, so streaming resumes.
         client.switch_command(silence.key, False)
         client.button_command(start.key)
-        await asyncio.sleep(3.0)
-        resumed = len(tracker.sensor_states["continuous_reg"]) - plateau
-        assert resumed >= 5, (
-            f"continuous polling did not resume after the device recovered (got {resumed} new reads)"
-        )
+        await _wait_for_reads(tracker, "continuous_reg", plateau + 5)
 
 
 @pytest.mark.asyncio
@@ -921,15 +980,34 @@ async def test_uart_mock_modbus_fairness(
     ):
         entities = await tracker.setup_and_start_scenario(client)
 
-        # Let both controllers hammer the bus for a while.
-        await asyncio.sleep(2.0)
+        # Let both controllers hammer the bus until each has issued a fair sample of requests.
+        counters = ["requests_1", "requests_2"]
+        await wait_or_fail(
+            _wait_until(
+                tracker,
+                counters,
+                lambda states: (
+                    all(states[name] for name in counters)
+                    and min(states[name][-1] for name in counters) >= 10
+                ),
+            ),
+            5.0,
+            "controllers did not poll",
+            states=tracker.sensor_states,
+        )
 
         # Stop polling so the counters settle to a final, stable value (state
         # coalescing means intermediate values may be skipped, but the final
         # value is always delivered once changes stop).
         stop_btn = require_entity(entities, "stop_scenario", ButtonInfo)
         client.button_command(stop_btn.key)
-        await asyncio.sleep(0.5)
+        # Quiet window longer than the API batch delay (100ms)
+        await wait_or_fail(
+            _wait_until_quiet(tracker, counters, 0.3),
+            3.0,
+            "counters did not settle after stop",
+            states=tracker.sensor_states,
+        )
 
         assert tracker.sensor_states["requests_1"], "controller 1 issued no requests"
         assert tracker.sensor_states["requests_2"], "controller 2 issued no requests"