Merge branch 'apply-bytes' into apply-bytes-migrate

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/cc1101/__init__.py

@@ -295,11 +295,6 @@ async def to_code(config):
         )
 
 
-# Actions
-SendPacketAction = ns.class_(
-    "SendPacketAction", automation.Action, cg.Parented.template(CC1101Component)
-)
-
 CC1101_ACTION_SCHEMA = cv.Schema(
     maybe_simple_id({cv.GenerateID(CONF_ID): cv.use_id(CC1101Component)})
 )
@@ -338,23 +333,11 @@ SEND_PACKET_ACTION_SCHEMA = cv.maybe_simple_value(
 )
 
 
-@automation.register_action(
+automation.register_apply_action(
     "cc1101.send_packet",
-    SendPacketAction,
     SEND_PACKET_ACTION_SCHEMA,
-    synchronous=True,
+    automation.ApplyBytes(CONF_DATA, "transmit_packet({})"),
 )
-async def send_packet_action_to_code(config, action_id, template_arg, args):
-    var = cg.new_Pvariable(action_id, template_arg)
-    await cg.register_parented(var, config[CONF_ID])
-    await automation.templatable_bytes(
-        config[CONF_DATA],
-        args,
-        var.set_data_template,
-        var.set_data_static,
-        "cc1101_data",
-    )
-    return var
 
 
 _CONFIG_VALIDATORS = {opt.schema: validator for opt, validator in CONFIG_MAP.items()}

esphome/components/cc1101/cc1101.h

@@ -132,13 +132,4 @@ class CC1101Component final : public Component,
   bool enter_tx_();
 };
 
-// Action Wrappers
-template<typename... Ts> class SendPacketAction final : public Action<Ts...>, public Parented<CC1101Component> {
-  TEMPLATABLE_BYTES(data)
-
-  void play(const Ts &...x) override {
-    this->data_.visit([this](const uint8_t *data, size_t len) { this->parent_->transmit_packet(data, len); }, x...);
-  }
-};
-
 }  // namespace esphome::cc1101

esphome/components/sx126x/__init__.py

@@ -11,10 +11,9 @@ from esphome.components.const import (
     CONF_RST_PIN,
 )
 import esphome.config_validation as cv
-from esphome.const import CONF_BUSY_PIN, CONF_DATA, CONF_FREQUENCY, CONF_ID
-from esphome.core import ID, TimePeriod
-from esphome.cpp_generator import MockObj
-from esphome.types import ConfigType, TemplateArgsType
+from esphome.const import CONF_BUSY_PIN, CONF_DATA, CONF_FREQUENCY
+from esphome.core import TimePeriod
+from esphome.types import ConfigType
 
 DOMAIN = "sx126x"
 MULTI_CONF = True
@@ -133,10 +132,6 @@ SHAPING = {
     "NONE": SX126xPulseShape.NO_FILTER,
 }
 
-SendPacketAction = sx126x_ns.class_(
-    "SendPacketAction", automation.Action, cg.Parented.template(SX126x)
-)
-
 
 def validate_raw_data(value: Any) -> bytes | list[int]:
     if isinstance(value, str):
@@ -324,25 +319,8 @@ SEND_PACKET_ACTION_SCHEMA = cv.maybe_simple_value(
 )
 
 
-@automation.register_action(
+automation.register_apply_action(
     "sx126x.send_packet",
-    SendPacketAction,
     SEND_PACKET_ACTION_SCHEMA,
-    synchronous=True,
+    automation.ApplyBytes(CONF_DATA, "transmit_packet({})", stack=MAX_PACKET_SIZE),
 )
-async def send_packet_action_to_code(
-    config: ConfigType,
-    action_id: ID,
-    template_arg: cg.TemplateArguments,
-    args: TemplateArgsType,
-) -> MockObj:
-    var = cg.new_Pvariable(action_id, template_arg)
-    await cg.register_parented(var, config[CONF_ID])
-    await automation.templatable_bytes(
-        config[CONF_DATA],
-        args,
-        var.set_data_template,
-        var.set_data_static,
-        "sx126x_data",
-    )
-    return var

esphome/components/sx126x/automation.h

@@ -1,18 +0,0 @@
-#pragma once
-
-#include "esphome/core/component.h"
-#include "esphome/core/automation.h"
-#include "esphome/components/sx126x/sx126x.h"
-
-namespace esphome::sx126x {
-
-template<typename... Ts> class SendPacketAction final : public Action<Ts...>, public Parented<SX126x> {
-  TEMPLATABLE_BYTES(data)
-
-  void play(const Ts &...x) override {
-    this->data_.template visit<SX126X_MAX_PACKET_SIZE>(
-        [this](const uint8_t *data, size_t len) { this->parent_->transmit_packet(data, len); }, x...);
-  }
-};
-
-}  // namespace esphome::sx126x

esphome/components/sx127x/__init__.py

@@ -11,9 +11,7 @@ from esphome.components.const import (
     CONF_RST_PIN,
 )
 import esphome.config_validation as cv
-from esphome.const import CONF_DATA, CONF_FREQUENCY, CONF_ID
-from esphome.core import ID
-from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.const import CONF_DATA, CONF_FREQUENCY
 from esphome.types import ConfigType
 
 DOMAIN = "sx127x"
@@ -117,10 +115,6 @@ SHAPING = {
     "NONE": SX127xPaRamp.SHAPING_NONE,
 }
 
-SendPacketAction = sx127x_ns.class_(
-    "SendPacketAction", automation.Action, cg.Parented.template(SX127x)
-)
-
 
 def validate_raw_data(value: Any) -> bytes | list[int]:
     if isinstance(value, str):
@@ -291,25 +285,8 @@ SEND_PACKET_ACTION_SCHEMA = cv.maybe_simple_value(
 )
 
 
-@automation.register_action(
+automation.register_apply_action(
     "sx127x.send_packet",
-    SendPacketAction,
     SEND_PACKET_ACTION_SCHEMA,
-    synchronous=True,
+    automation.ApplyBytes(CONF_DATA, "transmit_packet({})", stack=MAX_PACKET_SIZE),
 )
-async def send_packet_action_to_code(
-    config: ConfigType,
-    action_id: ID,
-    template_arg: cg.TemplateArguments,
-    args: TemplateArgsType,
-) -> MockObj:
-    var = cg.new_Pvariable(action_id, template_arg)
-    await cg.register_parented(var, config[CONF_ID])
-    await automation.templatable_bytes(
-        config[CONF_DATA],
-        args,
-        var.set_data_template,
-        var.set_data_static,
-        "sx127x_data",
-    )
-    return var

esphome/components/sx127x/automation.h

@@ -1,19 +0,0 @@
-#pragma once
-
-#include "esphome/core/component.h"
-#include "esphome/core/automation.h"
-#include "esphome/components/sx127x/sx127x.h"
-
-namespace esphome::sx127x {
-
-template<typename... Ts> class SendPacketAction final : public Action<Ts...>, public Parented<SX127x> {
-  TEMPLATABLE_BYTES(data)
-
- public:
-  void play(const Ts &...x) override {
-    this->data_.template visit<SX127X_MAX_PACKET_SIZE>(
-        [this](const uint8_t *data, size_t len) { this->parent_->transmit_packet(data, len); }, x...);
-  }
-};
-
-}  // namespace esphome::sx127x

esphome/components/uart/__init__.py

@@ -121,7 +121,6 @@ HOST_BAUD_RATES = [
 ]
 
 UARTDevice = uart_ns.class_("UARTDevice")
-UARTWriteAction = uart_ns.class_("UARTWriteAction", automation.Action)
 UARTDebugger = uart_ns.class_("UARTDebugger", cg.Component, automation.Action)
 UARTDummyReceiver = uart_ns.class_("UARTDummyReceiver", cg.Component)
 MULTI_CONF = True
@@ -593,9 +592,8 @@ def new_uart_component(
     return cg.new_bus_component(register_uart_device, config, *args)
 
 
-@automation.register_action(
+automation.register_apply_action(
     "uart.write",
-    UARTWriteAction,
     cv.maybe_simple_value(
         {
             cv.GenerateID(): cv.use_id(UARTComponent),
@@ -603,15 +601,11 @@ def new_uart_component(
         },
         key=CONF_DATA,
     ),
-    synchronous=True,
+    # A constant stays in flash: write_array_progmem copies it out in chunks on ESP8266
+    automation.ApplyBytes(
+        CONF_DATA, "write_array({})", progmem_target="write_array_progmem({})"
+    ),
 )
-async def uart_write_to_code(config, action_id, template_arg, args):
-    var = cg.new_Pvariable(action_id, template_arg)
-    await cg.register_parented(var, config[CONF_ID])
-    await automation.templatable_bytes(
-        config[CONF_DATA], args, var.set_data_template, var.set_data_static, "uart_data"
-    )
-    return var
 
 
 @coroutine_with_priority(CoroPriority.FINAL)

esphome/components/uart/automation.h

@@ -1,22 +0,0 @@
-#pragma once
-
-#include "uart.h"
-#include "esphome/core/automation.h"
-
-#include <vector>
-
-namespace esphome::uart {
-
-template<typename... Ts> class UARTWriteAction final : public Action<Ts...>, public Parented<UARTComponent> {
-  TEMPLATABLE_BYTES(data)
-
-  void play(const Ts &...x) override {
-    if (this->data_.is_static()) {
-      this->parent_->write_array_progmem(this->data_.data(), this->data_.size());
-    } else {
-      this->parent_->write_array(this->data_.value(x...));
-    }
-  }
-};
-
-}  // namespace esphome::uart

esphome/components/udp/__init__.py

@@ -29,7 +29,6 @@ DOMAIN = "udp"
 MULTI_CONF = True
 udp_ns = cg.esphome_ns.namespace("udp")
 UDPComponent = udp_ns.class_("UDPComponent", cg.Component)
-UDPWriteAction = udp_ns.class_("UDPWriteAction", automation.Action)
 trigger_argname = "data"
 # Listener callback type (non-owning span from UDP component)
 listener_args = cg.std_span.template(cg.uint8.operator("const"))
@@ -240,17 +239,23 @@ def validate_raw_data(value: Any) -> bytes | list[int]:
     )
 
 
-@automation.register_action(
+_UDP_WRITE_SCHEMA = cv.maybe_simple_value(
+    {
+        cv.GenerateID(): cv.use_id(UDPComponent),
+        cv.Required(CONF_DATA): cv.templatable(validate_raw_data),
+    },
+    key=CONF_DATA,
+)
+_udp_write_builder = automation.apply_action_builder(
     "udp.write",
-    UDPWriteAction,
-    cv.maybe_simple_value(
-        {
-            cv.GenerateID(): cv.use_id(UDPComponent),
-            cv.Required(CONF_DATA): cv.templatable(validate_raw_data),
-        },
-        key=CONF_DATA,
-    ),
-    synchronous=True,
+    _UDP_WRITE_SCHEMA,
+    # Constant payloads up to this size are sent from the stack on ESP8266
+    automation.ApplyBytes(CONF_DATA, "send_packet({})", stack=128),
+)
+
+
+@automation.register_action(
+    "udp.write", automation.ApplyAction, _UDP_WRITE_SCHEMA, synchronous=True
 )
 async def udp_write_to_code(
     config: ConfigType,
@@ -258,11 +263,6 @@ async def udp_write_to_code(
     template_arg: cg.TemplateArguments,
     args: TemplateArgsType,
 ) -> MockObj:
-    var = cg.new_Pvariable(action_id, template_arg)
     udp_var = await cg.get_variable(config[CONF_ID])
-    await cg.register_parented(var, udp_var)
     cg.add(udp_var.set_should_broadcast())
-    await automation.templatable_bytes(
-        config[CONF_DATA], args, var.set_data_template, var.set_data_static, "udp_data"
-    )
-    return var
+    return await _udp_write_builder(config, action_id, template_arg, args)

esphome/components/udp/automation.h

@@ -1,24 +0,0 @@
-#pragma once
-
-#include "udp_component.h"
-#ifdef USE_NETWORK
-#include "esphome/core/automation.h"
-
-namespace esphome::udp {
-
-/// Constant payloads up to this size are sent from the stack on ESP8266.
-static constexpr size_t STACK_PAYLOAD_SIZE = 128;
-
-template<typename... Ts> class UDPWriteAction final : public Action<Ts...>, public Parented<UDPComponent> {
-  TEMPLATABLE_BYTES(data)
-
-  void play(const Ts &...x) override {
-    // One sendto per packet, so a failed allocation never sends a partial packet
-    this->data_.template visit<STACK_PAYLOAD_SIZE>(
-        [this](const uint8_t *data, size_t len) { this->parent_->send_packet(data, len); }, x...);
-  }
-};
-
-}  // namespace esphome::udp
-
-#endif

tests/component_tests/cc1101/test_cc1101_payload_tables.py

@@ -14,7 +14,7 @@ def test_constant_payloads_share_progmem_tables(
     generate_main: Callable[[str | Path], str],
     component_config_path: Callable[[str], Path],
 ) -> None:
-    """Equal send_packet payloads share one table; lambdas stay templates."""
+    """Equal send_packet payloads share one table; a lambda is sent from its vector."""
     main_cpp = generate_main(component_config_path("payload_tables.yaml"))
 
     tables = re.findall(
@@ -23,8 +23,10 @@ def test_constant_payloads_share_progmem_tables(
     assert [v for _, v in tables] == [
         "{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08}"
     ]
-    assert main_cpp.count(f"set_data_static({tables[0][0]}, 8);") == 2
-    assert "set_data_template(" in main_cpp
+    assert main_cpp.count(f"->transmit_packet(::{tables[0][0]}, 8);") == 2
+    assert (
+        "->transmit_packet(esphome__bytes.data(), esphome__bytes.size());" in main_cpp
+    )
 
 
 @pytest.mark.parametrize("size", [64, 65])

tests/component_tests/sx126x/test_packet_tables.py

@@ -14,7 +14,7 @@ def test_constant_packets_share_progmem_tables(
     generate_main: Callable[[str | Path], str],
     component_config_path: Callable[[str], Path],
 ) -> None:
-    """Equal constant packets share one table; lambdas stay templates."""
+    """Equal constant packets share one table; a lambda is sent from its vector."""
     main_cpp = generate_main(component_config_path("packet_tables.yaml"))
 
     tables = dict(
@@ -26,8 +26,10 @@ def test_constant_packets_share_progmem_tables(
         ["{0xC5, 0x51, 0x78, 0x82}", "{0x68, 0x69}"]
     )
     shared = next(k for k, v in tables.items() if v == "{0xC5, 0x51, 0x78, 0x82}")
-    assert main_cpp.count(f"set_data_static({shared}, 4);") == 2
-    assert "set_data_template(" in main_cpp
+    assert main_cpp.count(f"->transmit_packet(::{shared}, 4);") == 2
+    assert (
+        "->transmit_packet(esphome__bytes.data(), esphome__bytes.size());" in main_cpp
+    )
     assert "set_sx1261" not in main_cpp
 
 

tests/component_tests/sx127x/test_packet_tables.py

@@ -14,7 +14,7 @@ def test_constant_packets_share_progmem_tables(
     generate_main: Callable[[str | Path], str],
     component_config_path: Callable[[str], Path],
 ) -> None:
-    """Equal constant packets share one table; lambdas stay templates."""
+    """Equal constant packets share one table; a lambda is sent from its vector."""
     main_cpp = generate_main(component_config_path("packet_tables.yaml"))
 
     tables = dict(
@@ -26,8 +26,10 @@ def test_constant_packets_share_progmem_tables(
         ["{0xC5, 0x51, 0x78, 0x82}", "{0x68, 0x69}"]
     )
     shared = next(k for k, v in tables.items() if v == "{0xC5, 0x51, 0x78, 0x82}")
-    assert main_cpp.count(f"set_data_static({shared}, 4);") == 2
-    assert "set_data_template(" in main_cpp
+    assert main_cpp.count(f"->transmit_packet(::{shared}, 4);") == 2
+    assert (
+        "->transmit_packet(esphome__bytes.data(), esphome__bytes.size());" in main_cpp
+    )
 
 
 def test_packet_data_length_limit() -> None:

tests/component_tests/uart/config/payload_tables_esp8266.yaml

@@ -0,0 +1,34 @@
+---
+esphome:
+  name: test
+  on_boot:
+    then:
+      - uart.write: [0x01, 0x02, 0x03]
+      - uart.write: !lambda return {0x09};
+      - uart.write: ""
+
+esp8266:
+  board: d1_mini
+
+uart:
+  tx_pin: GPIO1
+  rx_pin: GPIO3
+  baud_rate: 9600
+
+switch:
+  - platform: uart
+    id: shared_switch
+    name: Shared Switch
+    data: [0x01, 0x02, 0x03]
+  - platform: uart
+    id: on_off_switch
+    name: On Off Switch
+    data:
+      turn_on: "ON"
+      turn_off: [0x04]
+
+button:
+  - platform: uart
+    id: shared_button
+    name: Shared Button
+    data: [0x01, 0x02, 0x03]

tests/component_tests/uart/test_payload_tables.py

@@ -23,6 +23,19 @@ def test_constant_payloads_share_progmem_tables(
     )
     assert f"shared_switch->set_data_on({shared}, 3);" in main_cpp
     assert f"shared_button->set_data({shared}, 3);" in main_cpp
-    assert f"set_data_static({shared}, 3);" in main_cpp
-    assert "set_data_static(nullptr, 0);" in main_cpp
-    assert "set_data_template(" in main_cpp
+    assert f"->write_array(::{shared}, 3);" in main_cpp
+    assert "->write_array(nullptr, 0);" in main_cpp
+    assert "->write_array(esphome__bytes.data(), esphome__bytes.size());" in main_cpp
+
+
+def test_esp8266_write_action_reads_flash_directly(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """On ESP8266 the write action hands the table to write_array_progmem instead of copying it."""
+    main_cpp = generate_main(component_config_path("payload_tables_esp8266.yaml"))
+
+    assert "->write_array_progmem(::uart_data, 3);" in main_cpp
+    assert "->write_array_progmem(nullptr, 0);" in main_cpp
+    assert "->write_array(esphome__bytes.data(), esphome__bytes.size());" in main_cpp
+    assert "progmem_memcpy" not in main_cpp

tests/component_tests/udp/test_udp_write.py

@@ -8,10 +8,17 @@ def test_write_payloads_share_a_progmem_table(
     generate_main: Callable[[str | Path], str],
     component_config_path: Callable[[str], Path],
 ) -> None:
-    """Identical constant payloads share one flash table; an empty one needs none."""
+    """Identical constant payloads share one flash table and are copied out of it on
+    ESP8266 before sending; an empty one needs none; a lambda is sent from its vector."""
     main_cpp = generate_main(component_config_path("udp_write.yaml"))
 
-    assert main_cpp.count("udp_data[] PROGMEM = {0xAA, 0x55, 0x01, 0x02};") == 1
-    assert main_cpp.count("->set_data_static(udp_data, 4);") == 2
-    assert "->set_data_static(nullptr, 0);" in main_cpp
-    assert "->set_data_template(" in main_cpp
+    table = "esphome__udp_write__bytes"
+    assert main_cpp.count(f"{table}[] PROGMEM = {{0xAA, 0x55, 0x01, 0x02}};") == 1
+    assert (
+        main_cpp.count(f"esphome::progmem_memcpy(esphome__bytes, ::{table}, 4);") == 2
+    )
+    assert main_cpp.count("->send_packet(esphome__bytes, 4);") == 2
+    assert "uint8_t esphome__bytes[1];" in main_cpp
+    assert "->send_packet(esphome__bytes, 0);" in main_cpp
+    assert "->send_packet(esphome__bytes.data(), esphome__bytes.size());" in main_cpp
+    assert main_cpp.count("->set_should_broadcast();") == 4

tests/unit_tests/test_automation.py

@@ -11,6 +11,7 @@ from esphome.automation import (
     CONF_CHOICE_ID,
     CONF_CHOOSE,
     ApplyAction,
+    ApplyBytes,
     ApplyCall,
     ApplyCondition,
     ApplyField,
@@ -1281,6 +1282,145 @@ async def test_apply_string_ref_literal_stays_in_flash_on_esp8266(
     assert f"::{PARENT_OBJ}->set_payload({rendered});" in _apply_definition(mock_cg)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("platform", ["esp32", "esp8266"])
+async def test_apply_bytes_lambda_is_a_vector_in_a_block(
+    registries: tuple[Registry, Registry], mock_cg: MockCodegen, platform: str
+) -> None:
+    """A lambda payload is called once into a scoped vector and passed as data and size."""
+    fields = (ApplyBytes("data", "send({})"),)
+    config = {"data": Lambda("return {0x01, x};")}
+    await _run_apply_action(
+        registries, fields, config, args=[(cg.uint8, "x")], platform=platform
+    )
+    text = _apply_definition(mock_cg)
+    # Parenthesized so a reduced `return a, b;` stays one initializer
+    assert "const std::vector<uint8_t> esphome__bytes = (" in text
+    assert "-> std::vector<uint8_t>" in text
+    assert (
+        f"::{PARENT_OBJ}->send(esphome__bytes.data(), esphome__bytes.size());" in text
+    )
+    assert "progmem_memcpy" not in text
+
+
+@pytest.mark.asyncio
+async def test_apply_bytes_constant_is_a_shared_table(
+    registries: tuple[Registry, Registry], mock_cg: MockCodegen
+) -> None:
+    """A constant payload is the shared PROGMEM table and its length, named after the action."""
+    CORE.config = {}
+    fields = (ApplyBytes("data", "send({})"),)
+    await _run_apply_action(registries, fields, {"data": [0xA1, 0x02]})
+    assert f"::{PARENT_OBJ}->send(::esphome__my_apply__bytes, 2);" in _apply_definition(
+        mock_cg
+    )
+
+    mock_cg.new_pvariable.reset_mock()
+    await _run_apply_action(registries, fields, {"data": b""})
+    assert f"::{PARENT_OBJ}->send(nullptr, 0);" in _apply_definition(mock_cg)
+
+    mock_cg.new_pvariable.reset_mock()
+    await _run_apply_action(registries, fields, {})
+    assert "send(" not in _apply_definition(mock_cg)
+
+    with pytest.raises(EsphomeError, match="maximum is 65535"):
+        await _run_apply_action(registries, fields, {"data": bytes(0x10000)})
+
+
+@pytest.mark.asyncio
+async def test_apply_bytes_constant_is_copied_out_of_flash_on_esp8266(
+    registries: tuple[Registry, Registry], mock_cg: MockCodegen
+) -> None:
+    """On ESP8266 a constant goes through a stack buffer, or straight to a flash-aware target."""
+    CORE.config = {}
+    fields = (ApplyBytes("data", "send({})", stack=128),)
+    await _run_apply_action(
+        registries, fields, {"data": [0xA1, 0x02]}, platform="esp8266"
+    )
+    text = _apply_definition(mock_cg)
+    lines = [
+        "uint8_t esphome__bytes[2];",
+        "esphome::progmem_memcpy(esphome__bytes, ::esphome__my_apply__bytes, 2);",
+        f"::{PARENT_OBJ}->send(esphome__bytes, 2);",
+    ]
+    positions = [text.index(line) for line in lines]
+    assert positions == sorted(positions)
+
+    # Past the stack budget the copy goes to the heap
+    mock_cg.new_pvariable.reset_mock()
+    fields = (ApplyBytes("data", "send({})", stack=1),)
+    await _run_apply_action(
+        registries, fields, {"data": [0xA1, 0x02]}, platform="esp8266"
+    )
+    text = _apply_definition(mock_cg)
+    assert "auto esphome__bytes = std::make_unique<uint8_t[]>(2);" in text
+    assert (
+        "esphome::progmem_memcpy(esphome__bytes.get(), ::esphome__my_apply__bytes, 2);"
+        in text
+    )
+    assert f"::{PARENT_OBJ}->send(esphome__bytes.get(), 2);" in text
+
+    mock_cg.new_pvariable.reset_mock()
+    fields = (ApplyBytes("data", "send({})", progmem_target="send_progmem({})"),)
+    await _run_apply_action(
+        registries, fields, {"data": [0xA1, 0x02]}, platform="esp8266"
+    )
+    text = _apply_definition(mock_cg)
+    assert f"::{PARENT_OBJ}->send_progmem(::esphome__my_apply__bytes, 2);" in text
+    assert "progmem_memcpy" not in text
+
+    # An empty payload never touches flash but still passes a non-null buffer
+    mock_cg.new_pvariable.reset_mock()
+    fields = (ApplyBytes("data", "send({})"),)
+    await _run_apply_action(registries, fields, {"data": []}, platform="esp8266")
+    text = _apply_definition(mock_cg)
+    assert "uint8_t esphome__bytes[1];" in text
+    assert f"::{PARENT_OBJ}->send(esphome__bytes, 0);" in text
+    assert "progmem_memcpy" not in text
+
+
+@pytest.mark.asyncio
+async def test_apply_bytes_beside_other_fields_and_call_object(
+    registries: tuple[Registry, Registry], mock_cg: MockCodegen
+) -> None:
+    CORE.config = {}
+    fields = (
+        ApplyField("channel", "set_channel", cg.uint8),
+        ApplyBytes("data", "write({})"),
+    )
+    await _run_apply_action(
+        registries, fields, {"channel": 3, "data": [1]}, call="make_call"
+    )
+    text = _apply_definition(mock_cg)
+    lines = [
+        f"auto apply_call = ::{PARENT_OBJ}->make_call();",
+        "apply_call.set_channel(3);",
+        "apply_call.write(::esphome__my_apply__bytes, 1);",
+        "apply_call.perform();",
+    ]
+    positions = [text.index(line) for line in lines]
+    assert positions == sorted(positions)
+
+
+def test_apply_bytes_checks_its_shape(registries: tuple[Registry, Registry]) -> None:
+    # Escaped braces are not placeholders
+    assert ApplyBytes("data", "send({{}}, {})").target == "send({{}}, {})"
+    with pytest.raises(ValueError, match="needs one {}"):
+        ApplyBytes("data", "send({{}})")
+    with pytest.raises(ValueError, match="needs one {}"):
+        ApplyBytes("data", "send()")
+    with pytest.raises(ValueError, match="needs one {}"):
+        ApplyBytes("data", "send({})", progmem_target="send_progmem({}, {})")
+    with pytest.raises(ValueError, match="'dat' is not in the schema"):
+        register_apply_action(
+            "my.bad",
+            cv.Schema(
+                {cv.Required(CONF_ID): cv.string, cv.Required("data"): cv.string}
+            ),
+            ApplyBytes("dat", "send({})"),
+        )
+
+
 @pytest.mark.asyncio
 async def test_register_apply_condition_predicate(
     registries: tuple[Registry, Registry], mock_cg: MockCodegen