[core] Add Apply entities so hub forwarding buttons, switches, numbers and selects need no class (#20935)

esphome/automation.py

@@ -430,12 +430,20 @@ def _apply_function(
 ) -> MockObj:
     """Emit the generated function and declare ``id_`` as the ``ApplyAction`` or
     ``ApplyCondition`` templated on it, so ``play()`` calls it directly."""
-    fn = cg.static_function(
-        f"esphome__{id_.id}__fn", return_type, lambda_args, statements
-    )
+    fn = apply_static_function(id_, return_type, lambda_args, statements)
     return cg.new_Pvariable(id_, cg.TemplateArguments(fn, *template_arg))
 
 
+def apply_static_function(
+    id_: ID,
+    return_type: SafeExpType,
+    params: TemplateArgsType,
+    statements: list[str],
+) -> MockObj:
+    """Emit the static function generated for ``id_`` and return an expression naming it."""
+    return cg.static_function(f"esphome__{id_.id}__fn", return_type, params, statements)
+
+
 async def _render_values(
     name: str,
     target: str,

esphome/components/button/__init__.py

@@ -23,6 +23,7 @@ from esphome.const import (
 )
 from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import (
+    ApplyEntity,
     SubEntities,
     entity_duplicate_validator,
     queue_entity_register,
@@ -46,6 +47,7 @@ DEVICE_CLASSES = [
 button_ns = cg.esphome_ns.namespace("button")
 Button = button_ns.class_("Button", cg.EntityBase)
 ButtonPtr = Button.operator("ptr")
+ApplyButton = button_ns.class_("ApplyButton", Button)
 
 validate_device_class = cv.one_of(*DEVICE_CLASSES, lower=True, space="_")
 
@@ -155,11 +157,14 @@ async def new_button(config: ConfigType, *args: SafeExpType) -> MockObj:
     return var
 
 
+APPLY_ENTITY = ApplyEntity(ApplyButton, ((ButtonPtr, "entity"),))
+
+
 def sub_buttons(
     config: ConfigType, *, parent: MockObj | ID | None = None
 ) -> SubEntities:
-    """Return a SubEntities bound to new_button."""
-    return SubEntities(new_button, config, parent)
+    """Return a SubEntities bound to new_button; ``apply`` statements can use ``entity``."""
+    return SubEntities(new_button, config, parent, APPLY_ENTITY)
 
 
 BUTTON_PRESS_SCHEMA = maybe_simple_id(

esphome/components/button/button.h

@@ -46,4 +46,14 @@ class Button : public EntityBase {
   LazyCallbackManager<void()> press_callback_{};
 };
 
+/// Calls the codegen-generated Fn(this) on a press; the hub and any publish are baked into Fn.
+template<auto Fn> class ApplyButton final : public Button {
+ public:
+  // Not "= default": placement new would zero-fill .bss that is already zero.
+  ApplyButton() {}
+
+ protected:
+  void press_action() override { Fn(this); }
+};
+
 }  // namespace esphome::button

esphome/components/number/__init__.py

@@ -82,6 +82,7 @@ from esphome.const import (
 from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
 from esphome.core.config import UNIT_OF_MEASUREMENT_MAX_LENGTH
 from esphome.core.entity_helpers import (
+    ApplyEntity,
     SubEntities,
     entity_duplicate_validator,
     queue_entity_register,
@@ -160,6 +161,7 @@ IS_PLATFORM_COMPONENT = True
 number_ns = cg.esphome_ns.namespace("number")
 Number = number_ns.class_("Number", cg.EntityBase)
 NumberPtr = Number.operator("ptr")
+ApplyNumber = number_ns.class_("ApplyNumber", Number)
 
 # Triggers
 ValueRangeTrigger = number_ns.class_(
@@ -321,11 +323,18 @@ async def new_number(config, *args, min_value: float, max_value: float, step: fl
     return var
 
 
+APPLY_ENTITY = ApplyEntity(
+    ApplyNumber,
+    ((NumberPtr, "entity"), (cg.float_, "value")),
+    "entity->publish_state(value);",
+)
+
+
 def sub_numbers(
     config: ConfigType, *, parent: MockObj | ID | None = None
 ) -> SubEntities:
-    """Return a SubEntities bound to new_number."""
-    return SubEntities(new_number, config, parent)
+    """Return a SubEntities bound to new_number; ``apply`` statements can use ``entity`` and ``value``."""
+    return SubEntities(new_number, config, parent, APPLY_ENTITY)
 
 
 NUMBER_IN_RANGE_CONDITION_SCHEMA = cv.All(

esphome/components/number/number.h

@@ -54,4 +54,14 @@ class Number : public EntityBase {
   LazyCallbackManager<void(float)> state_callback_;
 };
 
+/// Calls the codegen-generated Fn(this, value) on a new value; the hub and any publish are baked into Fn.
+template<auto Fn> class ApplyNumber final : public Number {
+ public:
+  // Not "= default": placement new would zero-fill .bss that is already zero.
+  ApplyNumber() {}
+
+ protected:
+  void control(float value) override { Fn(this, value); }
+};
+
 }  // namespace esphome::number

esphome/components/select/__init__.py

@@ -19,6 +19,7 @@ from esphome.const import (
 )
 from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import (
+    ApplyEntity,
     SubEntities,
     entity_duplicate_validator,
     queue_entity_register,
@@ -35,6 +36,7 @@ IS_PLATFORM_COMPONENT = True
 select_ns = cg.esphome_ns.namespace("select")
 Select = select_ns.class_("Select", cg.EntityBase)
 SelectPtr = Select.operator("ptr")
+ApplySelect = select_ns.class_("ApplySelect", Select)
 
 # Conditions
 SelectIsCondition = select_ns.class_("SelectIsCondition", automation.Condition)
@@ -121,11 +123,19 @@ async def new_select(config, *args, options: list[str]):
     return var
 
 
+APPLY_ENTITY = ApplyEntity(
+    ApplySelect,
+    ((SelectPtr, "entity"), (cg.size_t, "index")),
+    "entity->publish_state(index);",
+    (("option", "const char *option = entity->option_at(index);"),),
+)
+
+
 def sub_selects(
     config: ConfigType, *, parent: MockObj | ID | None = None
 ) -> SubEntities:
-    """Return a SubEntities bound to new_select."""
-    return SubEntities(new_select, config, parent)
+    """Return a SubEntities bound to new_select; ``apply`` statements can use ``entity``, ``index`` and ``option``."""
+    return SubEntities(new_select, config, parent, APPLY_ENTITY)
 
 
 @coroutine_with_priority(CoroPriority.CORE)

esphome/components/select/select.h

@@ -112,4 +112,14 @@ class Select : public EntityBase {
   LazyCallbackManager<void(size_t)> state_callback_;
 };
 
+/// Calls the codegen-generated Fn(this, index) on a new index; the hub and any publish are baked into Fn.
+template<auto Fn> class ApplySelect final : public Select {
+ public:
+  // Not "= default": placement new would zero-fill .bss that is already zero.
+  ApplySelect() {}
+
+ protected:
+  void control(size_t index) override { Fn(this, index); }
+};
+
 }  // namespace esphome::select

esphome/components/switch/__init__.py

@@ -22,6 +22,7 @@ from esphome.const import (
 )
 from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import (
+    ApplyEntity,
     SubEntities,
     entity_duplicate_validator,
     queue_entity_register,
@@ -43,6 +44,7 @@ DEVICE_CLASSES = [
 switch_ns = cg.esphome_ns.namespace("switch_")
 Switch = switch_ns.class_("Switch", cg.EntityBase)
 SwitchPtr = Switch.operator("ptr")
+ApplySwitch = switch_ns.class_("ApplySwitch", Switch)
 
 SwitchRestoreMode = switch_ns.enum("SwitchRestoreMode")
 
@@ -176,11 +178,18 @@ async def new_switch(config, *args):
     return var
 
 
+APPLY_ENTITY = ApplyEntity(
+    ApplySwitch,
+    ((SwitchPtr, "entity"), (cg.bool_, "state")),
+    "entity->publish_state(state);",
+)
+
+
 def sub_switches(
     config: ConfigType, *, parent: MockObj | ID | None = None
 ) -> SubEntities:
-    """Return a SubEntities bound to new_switch."""
-    return SubEntities(new_switch, config, parent)
+    """Return a SubEntities bound to new_switch; ``apply`` statements can use ``entity`` and ``state``."""
+    return SubEntities(new_switch, config, parent, APPLY_ENTITY)
 
 
 SWITCH_ACTION_SCHEMA = maybe_simple_id(

esphome/components/switch/switch.h

@@ -147,4 +147,14 @@ class Switch : public EntityBase {
 #define LOG_SWITCH(prefix, type, obj) log_switch((TAG), (prefix), LOG_STR_LITERAL(type), (obj))
 void log_switch(const char *tag, const char *prefix, const char *type, Switch *obj);
 
+/// Calls the codegen-generated Fn(this, state) on a state write; the hub and any publish are baked into Fn.
+template<auto Fn> class ApplySwitch final : public Switch {
+ public:
+  // Not "= default": placement new would zero-fill .bss that is already zero.
+  ApplySwitch() {}
+
+ protected:
+  void write_state(bool state) override { Fn(this, state); }
+};
+
 }  // namespace esphome::switch_

esphome/core/entity_helpers.py

@@ -2,8 +2,10 @@ from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 import functools
 import logging
-from typing import Any
+import re
+from typing import Any, Literal
 
+from esphome.automation import apply_static_function, parent_ref
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import (
@@ -23,7 +25,7 @@ from esphome.core.config import (
     ICON_MAX_LENGTH,
     UNIT_OF_MEASUREMENT_MAX_LENGTH,
 )
-from esphome.cpp_generator import MockObj, RawStatement, add, get_variable
+from esphome.cpp_generator import MockObj, MockObjClass, RawStatement, add, get_variable
 from esphome.cpp_types import App
 import esphome.final_validate as fv
 from esphome.helpers import (
@@ -468,33 +470,116 @@ async def _setup_entity_impl(var: MockObj, config: ConfigType, platform: str) ->
     config[_KEY_ICON_IDX] = icon_idx
 
 
+ApplyPublish = Literal["before", "after"] | None
+
+
+@dataclass(frozen=True, slots=True)
+class ApplyEntity:
+    """A domain's ``Apply*`` class, its generated function's parameters and its publish statement.
+
+    ``locals`` are (name, declaration) pairs, declared only when the body uses the name.
+    """
+
+    cls: MockObjClass
+    params: tuple[tuple[SafeExpType, str], ...]
+    publish: str | None = None
+    locals: tuple[tuple[str, str], ...] = ()
+
+
+_APPLY_DOMAIN = "entity_apply"
+
+
 @dataclass(frozen=True, slots=True)
 class SubEntities:
     """Create the child entity configured under key, set parent (if bound) and pass it to setter.
 
-    Extra arguments go to new_entity. Returns None only when key is absent, so the result can be
-    used directly as a condition.
+    Returns None only when key is absent. ``apply``/``publish`` are documented on ``create``.
     """
 
     new_entity: Callable[..., Awaitable[MockObj]]
     config: ConfigType
     parent: MockObj | ID | None = None
+    apply_entity: ApplyEntity | None = None
 
     async def __call__(
         self,
         key: str,
-        setter: Callable[[MockObj], Expression],
+        setter: Callable[[MockObj], Expression] | None,
         *args: SafeExpType,
         **kwargs: Any,
     ) -> MockObj | None:
         if (conf := self.config.get(key)) is None:
             return None
-        var = await self.new_entity(conf, *args, **kwargs)
-        if self.parent is not None:
-            await cg.register_parented(var, self.parent)
-        add(setter(var))
+        return await self.create(conf, setter, *args, **kwargs)
+
+    async def create(
+        self,
+        conf: ConfigType,
+        setter: Callable[[MockObj], Expression] | None,
+        *args: SafeExpType,
+        apply: str | None = None,
+        publish: ApplyPublish = None,
+        **kwargs: Any,
+    ) -> MockObj:
+        """Create the entity from ``conf``; ``setter`` may be None when the hub keeps no pointer.
+
+        With ``apply`` the entity is the domain's ``Apply*`` class and its control runs
+        ``parent->apply``, with ``publish`` ("before"/"after") placing the domain's publish.
+        """
+        entity_id: ID = conf[CONF_ID]
+        is_apply_class = self.apply_entity is not None and str(entity_id.type) == str(
+            self.apply_entity.cls
+        )
+        if apply is None and is_apply_class:
+            raise ValueError(f"{entity_id}: {entity_id.type} needs apply=")
+        if apply is not None and not is_apply_class:
+            raise ValueError(
+                f"{entity_id}: apply= needs the domain's Apply class, not {entity_id.type}"
+            )
+        if apply is not None and self.parent is None:
+            raise ValueError(f"{entity_id}: apply= needs a parent")
+        if publish is not None and (apply is None or self.apply_entity.publish is None):
+            raise ValueError(f"{entity_id}: publish= needs apply= on a stateful domain")
+        if publish not in (None, "before", "after"):
+            raise ValueError(f'{entity_id}: publish= must be "before" or "after"')
+
+        if apply is None:
+            var = await self.new_entity(conf, *args, **kwargs)
+            if self.parent is not None:
+                await cg.register_parented(var, self.parent)
+        else:
+            fn = await self._apply_function(entity_id, apply, publish)
+            var = await self.new_entity(conf, cg.TemplateArguments(fn), *args, **kwargs)
+        if setter is not None:
+            add(setter(var))
         return var
 
+    async def _apply_function(
+        self, entity_id: ID, apply: str, publish: ApplyPublish
+    ) -> SafeExpType:
+        """Emit the control function once per distinct body, so equal entities share one class."""
+        parent = self.parent
+        if isinstance(parent, ID):
+            parent = await get_variable(parent)
+        apply_entity = self.apply_entity
+        body = [f"{parent_ref(parent)}->{apply};"]
+        if publish == "before":
+            body.insert(0, apply_entity.publish)
+        elif publish == "after":
+            body.append(apply_entity.publish)
+        for name, decl in apply_entity.locals:
+            # Declared right before its first use, so calls keep their order.
+            uses = (i for i, line in enumerate(body) if re.search(rf"\b{name}\b", line))
+            if (first := next(uses, None)) is not None:
+                body.insert(first, decl)
+        functions: dict[str, SafeExpType] = CORE.data.setdefault(_APPLY_DOMAIN, {})
+        key = "\n".join([*(f"{t} {n}" for t, n in apply_entity.params), *body])
+        if (fn := functions.get(key)) is None:
+            fn = functions[key] = apply_static_function(
+                entity_id, cg.void, list(apply_entity.params), body
+            )
+        return fn
+
 
 def inherit_property_from(property_to_inherit, parent_id_property, transform=None):
     """Validator that inherits a configuration property from another entity, for use with FINAL_VALIDATE_SCHEMA.

tests/components/button/apply_button_test.cpp

@@ -0,0 +1,27 @@
+#include <gtest/gtest.h>
+
+#include "esphome/components/button/button.h"
+
+namespace esphome::button::testing {
+
+static Button *last_entity = nullptr;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+static int presses = 0;                // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+
+static void on_press(Button *entity) {
+  last_entity = entity;
+  presses++;
+}
+
+TEST(ApplyButton, PressRunsTheFunctionWithTheEntity) {
+  last_entity = nullptr;
+  presses = 0;
+  ApplyButton<on_press> button;
+  button.press();
+  button.press();
+  EXPECT_EQ(presses, 2);
+  EXPECT_EQ(last_entity, &button);
+}
+
+TEST(ApplyButton, HoldsNoMoreThanTheBase) { EXPECT_EQ(sizeof(ApplyButton<on_press>), sizeof(Button)); }
+
+}  // namespace esphome::button::testing

tests/components/number/apply_number_test.cpp

@@ -0,0 +1,31 @@
+#include <gtest/gtest.h>
+
+#include <vector>
+
+#include "esphome/components/number/number.h"
+
+namespace esphome::number::testing {
+
+static Number *last_entity = nullptr;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+static std::vector<float> values;      // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+
+static void on_control(Number *entity, float value) {
+  last_entity = entity;
+  values.push_back(value);
+}
+
+TEST(ApplyNumber, ControlRunsTheFunctionWithTheValue) {
+  last_entity = nullptr;
+  values.clear();
+  ApplyNumber<on_control> number;
+  number.traits.set_min_value(0);
+  number.traits.set_max_value(10);
+  number.traits.set_step(1);
+  number.make_call().set_value(4).perform();
+  EXPECT_EQ(values, (std::vector<float>{4}));
+  EXPECT_EQ(last_entity, &number);
+}
+
+TEST(ApplyNumber, HoldsNoMoreThanTheBase) { EXPECT_EQ(sizeof(ApplyNumber<on_control>), sizeof(Number)); }
+
+}  // namespace esphome::number::testing

tests/components/select/apply_select_test.cpp

@@ -0,0 +1,27 @@
+#include <gtest/gtest.h>
+
+#include <string>
+#include <vector>
+
+#include "esphome/components/select/select.h"
+
+namespace esphome::select::testing {
+
+static constexpr const char *const APPLY_OPTIONS[] = {"low", "medium", "high"};
+
+static std::vector<std::string> chosen;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+
+static void on_control(Select *entity, size_t index) { chosen.emplace_back(entity->option_at(index)); }
+
+TEST(ApplySelect, ControlRunsTheFunctionWithTheIndex) {
+  chosen.clear();
+  ApplySelect<on_control> select;
+  select.traits.set_options_static(APPLY_OPTIONS, 3);
+  select.make_call().set_index(2).perform();
+  select.make_call().set_option("low").perform();
+  EXPECT_EQ(chosen, (std::vector<std::string>{"high", "low"}));
+}
+
+TEST(ApplySelect, HoldsNoMoreThanTheBase) { EXPECT_EQ(sizeof(ApplySelect<on_control>), sizeof(Select)); }
+
+}  // namespace esphome::select::testing

tests/components/switch/apply_switch_test.cpp

@@ -0,0 +1,37 @@
+#include <gtest/gtest.h>
+
+#include <vector>
+
+#include "esphome/components/switch/switch.h"
+
+namespace esphome::switch_::testing {
+
+static Switch *last_entity = nullptr;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+static std::vector<bool> writes;       // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+
+static void on_write(Switch *entity, bool state) {
+  last_entity = entity;
+  writes.push_back(state);
+}
+
+TEST(ApplySwitch, WriteRunsTheFunctionWithTheState) {
+  last_entity = nullptr;
+  writes.clear();
+  ApplySwitch<on_write> sw;
+  sw.turn_on();
+  sw.turn_off();
+  EXPECT_EQ(writes, (std::vector<bool>{true, false}));
+  EXPECT_EQ(last_entity, &sw);
+}
+
+TEST(ApplySwitch, InversionIsAppliedBeforeTheFunction) {
+  writes.clear();
+  ApplySwitch<on_write> sw;
+  sw.set_inverted(true);
+  sw.turn_on();
+  EXPECT_EQ(writes, (std::vector<bool>{false}));
+}
+
+TEST(ApplySwitch, HoldsNoMoreThanTheBase) { EXPECT_EQ(sizeof(ApplySwitch<on_write>), sizeof(Switch)); }
+
+}  // namespace esphome::switch_::testing

tests/unit_tests/core/test_entity_helpers.py

@@ -7,6 +7,8 @@ from typing import Any
 
 import pytest
 
+import esphome.codegen as cg
+from esphome.components import button, number, select, switch
 from esphome.config_validation import Invalid
 from esphome.const import (
     CONF_DEVICE_CLASS,
@@ -33,7 +35,7 @@ from esphome.core.entity_helpers import (
     setup_entity,
     setup_unit_of_measurement,
 )
-from esphome.cpp_generator import MockObj
+from esphome.cpp_generator import MockObj, MockObjClass, RawExpression
 from esphome.helpers import fnv1_hash, sanitize, snake_case
 
 from .common import load_config_from_fixture
@@ -1255,10 +1257,11 @@ async def test_sub_entities_creates_and_sets(
 
     monkeypatch.setattr(entity_helpers.cg, "register_parented", fake_register_parented)
     hub = MockObj("hub", "->")
-    sub = entity_helpers.SubEntities(new_entity, {"child": {}}, parent=hub)
+    conf = {CONF_ID: ID("child")}
+    sub = entity_helpers.SubEntities(new_entity, {"child": conf}, parent=hub)
 
     assert await sub("child", hub.set_child, 1, step=2) is created
-    assert calls == [({}, (1,), {"step": 2})]
+    assert calls == [(conf, (1,), {"step": 2})]
     assert setup_test_environment == ["parent child hub", "hub->set_child(child)"]
 
 
@@ -1294,3 +1297,314 @@ async def test_sub_entities_without_parent(
 
     assert await sub("child", hub.set_child) is not None
     assert setup_test_environment == ["hub->set_child(child)"]
+
+
+@pytest.fixture
+def apply_functions(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str, str]]:
+    """Record every generated apply function as (name, parameters, body)."""
+    emitted: list[tuple[str, str, str]] = []
+
+    def fake_static_function(
+        name: str, return_type: Any, parameters: list[tuple[Any, str]], body: list[str]
+    ) -> RawExpression:
+        assert str(return_type) == "void"
+        params = ", ".join(f"{type_} {arg}" for type_, arg in parameters)
+        emitted.append((name, params, "\n".join(body)))
+        return RawExpression(name)
+
+    monkeypatch.setattr(entity_helpers.cg, "static_function", fake_static_function)
+    CORE.data.pop(entity_helpers._APPLY_DOMAIN, None)
+    return emitted
+
+
+_HUB = MockObj("hub", "->")
+_APPLY_SWITCH_CLS = cg.esphome_ns.namespace("switch_").class_("ApplySwitch")
+_SWITCH_APPLY = entity_helpers.ApplyEntity(
+    _APPLY_SWITCH_CLS,
+    ((RawExpression("switch_::Switch *"), "entity"), (RawExpression("bool"), "state")),
+    "entity->publish_state(state);",
+)
+_BUTTON_APPLY = entity_helpers.ApplyEntity(
+    cg.esphome_ns.namespace("button").class_("ApplyButton"),
+    ((RawExpression("button::Button *"), "entity"),),
+)
+
+
+def _apply_conf(name: str, type_: MockObjClass = _APPLY_SWITCH_CLS) -> dict[str, ID]:
+    return {CONF_ID: ID(name, type=type_)}
+
+
+def _apply_sub(
+    calls: list[tuple[Any, ...]],
+    config: dict[str, Any],
+    apply_entity: entity_helpers.ApplyEntity = _SWITCH_APPLY,
+    parent: MockObj | ID | None = _HUB,
+) -> entity_helpers.SubEntities:
+    async def new_entity(conf: dict[str, Any], *args: Any, **kwargs: Any) -> MockObj:
+        calls.append((conf[CONF_ID].id, *(str(a) for a in args)))
+        return MockObj(conf[CONF_ID].id)
+
+    return entity_helpers.SubEntities(new_entity, config, parent, apply_entity)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("publish", "body"),
+    [
+        (None, "::hub->set_bluetooth(state);"),
+        (
+            "before",
+            "entity->publish_state(state);\n::hub->set_bluetooth(state);",
+        ),
+        ("after", "::hub->set_bluetooth(state);\nentity->publish_state(state);"),
+    ],
+)
+async def test_sub_entities_apply_generates_function(
+    monkeypatch: pytest.MonkeyPatch,
+    setup_test_environment: list[str],
+    apply_functions: list[tuple[str, str, str]],
+    publish: str | None,
+    body: str,
+) -> None:
+    """With apply the entity is templated on a generated function and stores no parent."""
+
+    async def fail_register_parented(var: MockObj, parent: MockObj) -> None:
+        raise AssertionError("must not be called")
+
+    monkeypatch.setattr(entity_helpers.cg, "register_parented", fail_register_parented)
+    calls: list[tuple[Any, ...]] = []
+    sub = _apply_sub(calls, {"bt": _apply_conf("bt_sw")})
+
+    var = await sub(
+        "bt", _HUB.set_bt_switch, apply="set_bluetooth(state)", publish=publish
+    )
+    assert str(var) == "bt_sw"
+    assert calls == [("bt_sw", "<esphome__bt_sw__fn>")]
+    assert apply_functions == [
+        ("esphome__bt_sw__fn", "switch_::Switch * entity, bool state", body)
+    ]
+    assert setup_test_environment == ["hub->set_bt_switch(bt_sw)"]
+
+
+_SELECT_APPLY = entity_helpers.ApplyEntity(
+    cg.esphome_ns.namespace("select").class_("ApplySelect"),
+    ((RawExpression("select::Select *"), "entity"), (RawExpression("size_t"), "index")),
+    "entity->publish_state(index);",
+    (("option", "const char *option = entity->option_at(index);"),),
+)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("apply", "publish", "body"),
+    [
+        (
+            "set_mode(option)",
+            "after",
+            (
+                "const char *option = entity->option_at(index);\n"
+                "::hub->set_mode(option);\nentity->publish_state(index);"
+            ),
+        ),
+        (
+            "set_mode(option)",
+            "before",
+            (
+                "entity->publish_state(index);\n"
+                "const char *option = entity->option_at(index);\n"
+                "::hub->set_mode(option);"
+            ),
+        ),
+        (
+            "set_mode(index, options_)",
+            "after",
+            "::hub->set_mode(index, options_);\nentity->publish_state(index);",
+        ),
+    ],
+)
+async def test_sub_entities_apply_locals(
+    setup_test_environment: list[str],
+    apply_functions: list[tuple[str, str, str]],
+    apply: str,
+    publish: str,
+    body: str,
+) -> None:
+    """A local is declared before its first use, and only when named as a whole identifier."""
+    conf = _apply_conf("m_sel", _SELECT_APPLY.cls)
+    sub = _apply_sub([], {"m": conf}, _SELECT_APPLY)
+    await sub("m", None, apply=apply, publish=publish)
+    assert apply_functions == [
+        ("esphome__m_sel__fn", "select::Select * entity, size_t index", body)
+    ]
+
+
+@pytest.mark.asyncio
+async def test_sub_entities_apply_shares_equal_functions(
+    setup_test_environment: list[str], apply_functions: list[tuple[str, str, str]]
+) -> None:
+    """Entities with the same statement share one function, so one class instantiation."""
+    calls: list[tuple[Any, ...]] = []
+    config = {
+        "a": _apply_conf("a_sw"),
+        "b": _apply_conf("b_sw"),
+        "c": _apply_conf("c_sw"),
+    }
+    sub = _apply_sub(calls, config)
+    await sub("a", _HUB.set_a, apply="refresh()", publish="before")
+    await sub("b", _HUB.set_b, apply="refresh()", publish="before")
+    await sub("c", _HUB.set_c, apply="refresh()")
+    assert calls == [
+        ("a_sw", "<esphome__a_sw__fn>"),
+        ("b_sw", "<esphome__a_sw__fn>"),
+        ("c_sw", "<esphome__c_sw__fn>"),
+    ]
+    assert [name for name, _, _ in apply_functions] == [
+        "esphome__a_sw__fn",
+        "esphome__c_sw__fn",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_sub_entities_apply_parent_id(
+    monkeypatch: pytest.MonkeyPatch,
+    setup_test_environment: list[str],
+    apply_functions: list[tuple[str, str, str]],
+) -> None:
+    """A parent given as an ID is looked up once and named from global scope."""
+
+    async def fake_get_variable(id_: ID) -> MockObj:
+        assert id_ == ID("my_hub")
+        return MockObj("my_hub", "->")
+
+    monkeypatch.setattr(entity_helpers, "get_variable", fake_get_variable)
+    calls: list[tuple[Any, ...]] = []
+    sub = _apply_sub(calls, {"r": _apply_conf("r_sw")}, parent=ID("my_hub"))
+    await sub("r", MockObj("my_hub", "->").set_r, apply="restart()")
+    assert apply_functions[0][2] == "::my_hub->restart();"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("apply_entity", "conf", "kwargs", "match"),
+    [
+        (_SWITCH_APPLY, _apply_conf("k_id"), {}, "needs apply="),
+        (
+            _SWITCH_APPLY,
+            _apply_conf("k_id", cg.esphome_ns.namespace("x").class_("MySwitch")),
+            {"apply": "x()"},
+            "needs the domain's Apply class",
+        ),
+        (
+            _BUTTON_APPLY,
+            _apply_conf(
+                "k_id", cg.esphome_ns.namespace("button").class_("ApplyButton")
+            ),
+            {"apply": "restart()", "publish": "before"},
+            "publish= needs",
+        ),
+        (
+            _SWITCH_APPLY,
+            _apply_conf("k_id", cg.esphome_ns.namespace("x").class_("MySwitch")),
+            {"publish": "before"},
+            "publish= needs",
+        ),
+        (
+            _SWITCH_APPLY,
+            _apply_conf("k_id"),
+            {"apply": "x()", "publish": "Before"},
+            "must be",
+        ),
+    ],
+)
+async def test_sub_entities_apply_errors(
+    setup_test_environment: list[str],
+    apply_functions: list[tuple[str, str, str]],
+    apply_entity: entity_helpers.ApplyEntity,
+    conf: dict[str, ID],
+    kwargs: dict[str, str],
+    match: str,
+) -> None:
+    """A mismatch between the declared class and apply= fails in Python, not in C++."""
+    calls: list[tuple[Any, ...]] = []
+    sub = _apply_sub(calls, {"k": conf}, apply_entity)
+    with pytest.raises(ValueError, match=match):
+        await sub("k", MockObj("hub", "->").set_k, **kwargs)
+    assert calls == []
+    assert apply_functions == []
+
+
+@pytest.mark.asyncio
+async def test_sub_entities_apply_needs_parent(
+    setup_test_environment: list[str], apply_functions: list[tuple[str, str, str]]
+) -> None:
+    calls: list[tuple[Any, ...]] = []
+    sub = _apply_sub(calls, {"k": _apply_conf("k_id")}, parent=None)
+    with pytest.raises(ValueError, match="needs a parent"):
+        await sub("k", None, apply="x()")
+
+
+@pytest.mark.parametrize(
+    ("sub_fn", "cls", "params", "publish"),
+    [
+        (button.sub_buttons, button.ApplyButton, "button::Button * entity", None),
+        (
+            switch.sub_switches,
+            switch.ApplySwitch,
+            "switch_::Switch * entity, bool state",
+            "entity->publish_state(state);",
+        ),
+        (
+            number.sub_numbers,
+            number.ApplyNumber,
+            "number::Number * entity, float value",
+            "entity->publish_state(value);",
+        ),
+        (
+            select.sub_selects,
+            select.ApplySelect,
+            "select::Select * entity, size_t index",
+            "entity->publish_state(index);",
+        ),
+    ],
+)
+def test_domain_apply_entities(
+    sub_fn: Callable[..., entity_helpers.SubEntities],
+    cls: MockObjClass,
+    params: str,
+    publish: str | None,
+) -> None:
+    """Each domain's sub helper declares its Apply class and the function it calls."""
+    apply_entity = sub_fn({}).apply_entity
+    assert apply_entity.cls is cls
+    assert ", ".join(f"{type_} {arg}" for type_, arg in apply_entity.params) == params
+    assert apply_entity.publish == publish
+
+
+@pytest.mark.asyncio
+async def test_sub_entities_without_setter(
+    setup_test_environment: list[str], apply_functions: list[tuple[str, str, str]]
+) -> None:
+    """A hub that does not keep the entity passes no setter; nothing else is emitted."""
+    calls: list[tuple[Any, ...]] = []
+    sub = _apply_sub(calls, {"vent": _apply_conf("vent_sw")})
+    assert await sub("vent", None, apply="vent_door()") is not None
+    assert calls == [("vent_sw", "<esphome__vent_sw__fn>")]
+    assert setup_test_environment == []
+
+
+@pytest.mark.asyncio
+async def test_sub_entities_create_from_whole_config(
+    setup_test_environment: list[str], apply_functions: list[tuple[str, str, str]]
+) -> None:
+    """create() builds the entity from a config that is the entity itself."""
+    calls: list[tuple[Any, ...]] = []
+    sub = _apply_sub(calls, {})
+    var = await sub.create(
+        _apply_conf("rf_sw"),
+        _HUB.set_rf_switch,
+        apply="set_rf_mode(state)",
+        publish="before",
+    )
+    assert str(var) == "rf_sw"
+    assert calls == [("rf_sw", "<esphome__rf_sw__fn>")]
+    assert setup_test_environment == ["hub->set_rf_switch(rf_sw)"]