Merge branch 'dev' into apply-switch-momentary

AGENTS.md

@@ -443,6 +443,8 @@ file does, and it is the authority when they disagree. The most useful starting
           uses `button.ApplyButton` (`switch.ApplySwitch`, `number.ApplyNumber`, `select.ApplySelect`) in its schema
           and `apply=` on the `sub_*` call. The body is a call on the hub with `entity` and `state`, `value` or
           `index`/`option` in scope; on a switch, number or select `publish="before"` or `"after"` publishes the state.
+          `publish="momentary"` is only for legacy action switches (restart, factory reset): it publishes off and runs
+          the body only when turned on. Never use it for a new entity; an action is a button.
           ```python
           switches = switch.sub_switches(config, parent=hub)
           await switches(CONF_BLUETOOTH, None, apply="set_bluetooth(state)", publish="after")

esphome/components/switch/__init__.py

@@ -303,6 +303,7 @@ APPLY_ENTITY = ApplyEntity(
     ApplySwitch,
     ((SwitchPtr, "entity"), (cg.bool_, "state")),
     "entity->publish_state(state);",
+    momentary=("entity->publish_state(false);", "state"),
 )
 
 

esphome/core/entity_helpers.py

@@ -494,7 +494,7 @@ async def _setup_entity_impl(var: MockObj, config: ConfigType, platform: str) ->
     config[_KEY_ICON_IDX] = icon_idx
 
 
-ApplyPublish = Literal["before", "after"] | None
+ApplyPublish = Literal["before", "after", "momentary"] | None
 
 
 @dataclass(frozen=True, slots=True)
@@ -502,12 +502,14 @@ 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.
+    ``momentary`` is (acknowledge statement, condition) for ``publish="momentary"``.
     """
 
     cls: MockObjClass
     params: tuple[tuple[SafeExpType, str], ...]
     publish: str | None = None
     locals: tuple[tuple[str, str], ...] = ()
+    momentary: tuple[str, str] | None = None
 
 
 _APPLY_DOMAIN = "entity_apply"
@@ -549,7 +551,9 @@ class SubEntities:
 
         With ``apply`` the entity is the domain's ``Apply*`` class and its control runs
         ``parent->apply``, or ``apply`` as a namespaced free statement when no parent is bound, with
-        ``publish`` ("before"/"after") placing the domain's publish.
+        ``publish`` ("before"/"after") placing the domain's publish. ``publish="momentary"`` instead
+        acknowledges first (a switch publishes off) and runs ``apply`` only when turned on. It is only for
+        legacy action switches; a new action entity is a button.
         """
         entity_id: ID = conf[CONF_ID]
         is_apply_class = self.apply_entity is not None and str(entity_id.type) == str(
@@ -570,8 +574,12 @@ class SubEntities:
             )
         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 publish not in (None, "before", "after", "momentary"):
+            raise ValueError(
+                f'{entity_id}: publish= must be "before", "after" or "momentary"'
+            )
+        if publish == "momentary" and self.apply_entity.momentary is None:
+            raise ValueError(f'{entity_id}: publish="momentary" is not supported here')
 
         if apply is None:
             var = await self.new_entity(conf, *args, **kwargs)
@@ -592,11 +600,16 @@ class SubEntities:
         if isinstance(parent, ID):
             parent = await get_variable(parent)
         apply_entity = self.apply_entity
-        body = [f"{apply};" if parent is None else f"{parent_ref(parent)}->{apply};"]
+        call = f"{apply};" if parent is None else f"{parent_ref(parent)}->{apply};"
         if publish == "before":
-            body.insert(0, apply_entity.publish)
+            body = [apply_entity.publish, call]
         elif publish == "after":
-            body.append(apply_entity.publish)
+            body = [call, apply_entity.publish]
+        elif publish == "momentary":
+            acknowledge, condition = apply_entity.momentary
+            body = [acknowledge, f"if ({condition}) {{ {call} }}"]
+        else:
+            body = [call]
         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))

tests/integration/state_utils.py

@@ -595,8 +595,8 @@ class SensorStateCollector:
 class SensorTracker:
     """Data-driven sensor state tracker with expected-value futures.
 
-    Tracks sensor, binary sensor and text sensor state updates and resolves
-    futures when they report specific expected values. Eliminates per-sensor future
+    Tracks sensor and binary sensor state updates and resolves futures when
+    they report specific expected values. Eliminates per-sensor future
     boilerplate.
 
     Usage::
@@ -608,20 +608,12 @@ class SensorTracker:
     """
 
     def __init__(self, sensor_names: list[str]) -> None:
-        self.sensor_states: dict[str, list[float | bool | str]] = {
-            name: [] for name in sensor_names
-        }
+        self.sensor_states: dict[str, list[float]] = {name: [] for name in sensor_names}
         self.key_to_sensor: dict[int, str] = {}
         self._expectations: dict[str, list[tuple[object, asyncio.Future]]] = {}
 
     _ANY = object()  # Sentinel: match any value
 
-    def map_entities(self, entities: list[EntityInfo]) -> None:
-        """Map the tracked names to the keys of ``entities``."""
-        self.key_to_sensor.update(
-            build_key_to_entity_mapping(entities, list(self.sensor_states))
-        )
-
     def expect(self, name: str, value: object) -> asyncio.Future:
         """Register an expected value for *name* and return a future for it."""
         future: asyncio.Future = asyncio.get_running_loop().create_future()
@@ -646,7 +638,7 @@ class SensorTracker:
                 connect-time states so they cannot satisfy a later phase.
         """
         if (
-            not isinstance(state, (SensorState, BinarySensorState, TextSensorState))
+            not isinstance(state, (SensorState, BinarySensorState))
             or state.missing_state
         ):
             return
@@ -716,7 +708,9 @@ class SensorTracker:
                 satisfy a later phase.
         """
         entities, _ = await client.list_entities_services()
-        self.map_entities(entities)
+        self.key_to_sensor.update(
+            build_key_to_entity_mapping(entities, list(self.sensor_states.keys()))
+        )
         initial_state_helper = InitialStateHelper(entities)
         client.subscribe_states(initial_state_helper.on_state_wrapper(self.on_state))
         await initial_state_helper.wait_for_initial_states()

tests/integration/test_api_get_time_response_timezone.py

@@ -6,7 +6,7 @@ from aioesphomeapi import connection as api_connection
 from aioesphomeapi.api_pb2 import GetTimeResponse
 import pytest
 
-from .state_utils import SensorTracker
+from .state_utils import SensorTracker, build_key_to_entity_mapping
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 # 2024-01-01 00:00:00 UTC
@@ -32,7 +32,7 @@ async def test_api_get_time_response_timezone(
     async with run_compiled(yaml_config), api_client_connected() as client:
         entities, _ = await client.list_entities_services()
         tracker = SensorTracker(["tz_offset"])
-        tracker.map_entities(entities)
+        tracker.key_to_sensor = build_key_to_entity_mapping(entities, ["tz_offset"])
         client.subscribe_states(tracker.on_state)
 
         await tracker.await_change(tracker.expect_any("tz_offset"), "tz_offset")

tests/integration/test_status_flags.py

@@ -19,7 +19,11 @@ import asyncio
 
 import pytest
 
-from .state_utils import SensorTracker, subscribe_after_initial_states
+from .state_utils import (
+    SensorTracker,
+    build_key_to_entity_mapping,
+    subscribe_after_initial_states,
+)
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 # Time to let the host-mode main loop run so status_led_light.loop() can
@@ -63,7 +67,9 @@ async def test_status_flags(
                 "status_led_last_state",
             ]
         )
-        tracker.map_entities(entities)
+        tracker.key_to_sensor.update(
+            build_key_to_entity_mapping(entities, list(tracker.sensor_states.keys()))
+        )
 
         # Swallow initial state broadcasts so the test only reacts to
         # state changes triggered by our service calls.

tests/integration/test_template_time.py

@@ -2,10 +2,14 @@
 
 from __future__ import annotations
 
+import asyncio
+
+from aioesphomeapi import EntityState, TextSensorState
 import pytest
 
-from .state_utils import SensorTracker, subscribe_after_initial_states
+from .state_utils import build_key_to_entity_mapping, subscribe_after_initial_states
 from .types import APIClientConnectedFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
 
 # Timestamps are exposed as text sensors (rather than 32-bit float sensor states,
 # which cannot represent a UNIX epoch exactly) so the exact value can be checked.
@@ -26,10 +30,18 @@ async def test_template_time(
     async with run_compiled(yaml_config), api_client_connected() as client:
         entities, _ = await client.list_entities_services()
 
-        tracker = SensorTracker(list(EXPECTED_STATES))
-        tracker.map_entities(entities)
-        futures = tracker.expect_all(EXPECTED_STATES)
-        await subscribe_after_initial_states(
-            client, entities, on_state=tracker.on_state
-        )
-        await tracker.await_all(futures, timeout=3.0)
+        key_to_name = build_key_to_entity_mapping(entities, list(EXPECTED_STATES))
+        events = {name: asyncio.Event() for name in EXPECTED_STATES}
+
+        def on_state(state: EntityState) -> None:
+            if isinstance(state, TextSensorState) and not state.missing_state:
+                name = key_to_name.get(state.key)
+                if name is not None and state.state == EXPECTED_STATES[name]:
+                    events[name].set()
+
+        await subscribe_after_initial_states(client, entities, on_state=on_state)
+
+        for name, event in events.items():
+            await wait_or_fail(
+                event, 3.0, f"Timeout waiting for {name} to report its expected value"
+            )

tests/unit_tests/core/test_entity_helpers.py

@@ -1479,6 +1479,7 @@ _SWITCH_APPLY = entity_helpers.ApplyEntity(
     _APPLY_SWITCH_CLS,
     ((RawExpression("switch_::Switch *"), "entity"), (RawExpression("bool"), "state")),
     "entity->publish_state(state);",
+    momentary=("entity->publish_state(false);", "state"),
 )
 _BUTTON_APPLY = entity_helpers.ApplyEntity(
     cg.esphome_ns.namespace("button").class_("ApplyButton"),
@@ -1670,6 +1671,12 @@ async def test_sub_entities_apply_parent_id(
             {"apply": "x()", "publish": "Before"},
             "must be",
         ),
+        (
+            _SELECT_APPLY,
+            _apply_conf("k_id", _SELECT_APPLY.cls),
+            {"apply": "x()", "publish": "momentary"},
+            "not supported",
+        ),
     ],
 )
 async def test_sub_entities_apply_errors(
@@ -1689,6 +1696,35 @@ async def test_sub_entities_apply_errors(
     assert apply_functions == []
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("parent", "apply", "body"),
+    [
+        (
+            _HUB,
+            "restart()",
+            "entity->publish_state(false);\nif (state) { ::hub->restart(); }",
+        ),
+        (
+            None,
+            "restart::restart_device()",
+            "entity->publish_state(false);\nif (state) { restart::restart_device(); }",
+        ),
+    ],
+)
+async def test_sub_entities_apply_momentary(
+    setup_test_environment: list[str],
+    apply_functions: list[tuple[str, str, str]],
+    parent: MockObj | None,
+    apply: str,
+    body: str,
+) -> None:
+    """A momentary switch publishes off first and acts only when turned on."""
+    sub = _apply_sub([], {"r": _apply_conf("r_sw")}, parent=parent)
+    await sub("r", None, apply=apply, publish="momentary")
+    assert apply_functions[0][2] == body
+
+
 @pytest.mark.asyncio
 async def test_sub_entities_apply_without_parent_needs_namespaced_call(
     setup_test_environment: list[str],