Merge remote-tracking branch 'origin/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/copy/sensor/__init__.py

@@ -13,19 +13,10 @@ from esphome.const import (
 from esphome.core.entity_helpers import inherit_property_from
 from esphome.types import ConfigType
 
-from .. import copy_ns
-
-CopySensor = copy_ns.class_("CopySensor", sensor.Sensor, cg.Component)
-
-
-CONFIG_SCHEMA = (
-    sensor.sensor_schema(CopySensor)
-    .extend(
-        {
-            cv.Required(CONF_SOURCE_ID): cv.use_id(sensor.Sensor),
-        }
-    )
-    .extend(cv.COMPONENT_SCHEMA)
+CONFIG_SCHEMA = sensor.sensor_schema(sensor.Sensor).extend(
+    {
+        cv.Required(CONF_SOURCE_ID): cv.use_id(sensor.Sensor),
+    }
 )
 
 FINAL_VALIDATE_SCHEMA = cv.All(
@@ -40,6 +31,4 @@ FINAL_VALIDATE_SCHEMA = cv.All(
 
 async def to_code(config: ConfigType) -> None:
     var = await sensor.new_sensor(config)
-    await cg.register_component(var, config)
-
-    await cg.add_variable(var.set_source, config[CONF_SOURCE_ID])
+    sensor.follow_sensor(var, await cg.get_variable(config[CONF_SOURCE_ID]))

esphome/components/copy/sensor/copy_sensor.cpp

@@ -1,16 +0,0 @@
-#include "copy_sensor.h"
-#include "esphome/core/log.h"
-
-namespace esphome::copy {
-
-ESPHOME_LOG_TAG(TAG, "copy.sensor");
-
-void CopySensor::setup() {
-  source_->add_on_state_callback([this](float value) { this->publish_state(value); });
-  if (source_->has_state())
-    this->publish_state(source_->state);
-}
-
-void CopySensor::dump_config() { LOG_SENSOR("", "Copy Sensor", this); }
-
-}  // namespace esphome::copy

esphome/components/copy/sensor/copy_sensor.h

@@ -1,18 +0,0 @@
-#pragma once
-
-#include "esphome/core/component.h"
-#include "esphome/components/sensor/sensor.h"
-
-namespace esphome::copy {
-
-class CopySensor final : public sensor::Sensor, public Component {
- public:
-  void set_source(sensor::Sensor *source) { source_ = source; }
-  void setup() override;
-  void dump_config() override;
-
- protected:
-  sensor::Sensor *source_;
-};
-
-}  // namespace esphome::copy

esphome/components/copy/text_sensor/__init__.py

@@ -5,19 +5,10 @@ from esphome.const import CONF_ENTITY_CATEGORY, CONF_ICON, CONF_SOURCE_ID
 from esphome.core.entity_helpers import inherit_property_from
 from esphome.types import ConfigType
 
-from .. import copy_ns
-
-CopyTextSensor = copy_ns.class_("CopyTextSensor", text_sensor.TextSensor, cg.Component)
-
-
-CONFIG_SCHEMA = (
-    text_sensor.text_sensor_schema(CopyTextSensor)
-    .extend(
-        {
-            cv.Required(CONF_SOURCE_ID): cv.use_id(text_sensor.TextSensor),
-        }
-    )
-    .extend(cv.COMPONENT_SCHEMA)
+CONFIG_SCHEMA = text_sensor.text_sensor_schema(text_sensor.TextSensor).extend(
+    {
+        cv.Required(CONF_SOURCE_ID): cv.use_id(text_sensor.TextSensor),
+    }
 )
 
 FINAL_VALIDATE_SCHEMA = cv.All(
@@ -28,6 +19,4 @@ FINAL_VALIDATE_SCHEMA = cv.All(
 
 async def to_code(config: ConfigType) -> None:
     var = await text_sensor.new_text_sensor(config)
-    await cg.register_component(var, config)
-
-    await cg.add_variable(var.set_source, config[CONF_SOURCE_ID])
+    text_sensor.follow_text_sensor(var, await cg.get_variable(config[CONF_SOURCE_ID]))

esphome/components/copy/text_sensor/copy_text_sensor.cpp

@@ -1,16 +0,0 @@
-#include "copy_text_sensor.h"
-#include "esphome/core/log.h"
-
-namespace esphome::copy {
-
-ESPHOME_LOG_TAG(TAG, "copy.text_sensor");
-
-void CopyTextSensor::setup() {
-  source_->add_on_state_callback([this](const std::string &value) { this->publish_state(value); });
-  if (source_->has_state())
-    this->publish_state(source_->state);
-}
-
-void CopyTextSensor::dump_config() { LOG_TEXT_SENSOR("", "Copy Sensor", this); }
-
-}  // namespace esphome::copy

esphome/components/copy/text_sensor/copy_text_sensor.h

@@ -1,18 +0,0 @@
-#pragma once
-
-#include "esphome/core/component.h"
-#include "esphome/components/text_sensor/text_sensor.h"
-
-namespace esphome::copy {
-
-class CopyTextSensor final : public text_sensor::TextSensor, public Component {
- public:
-  void set_source(text_sensor::TextSensor *source) { source_ = source; }
-  void setup() override;
-  void dump_config() override;
-
- protected:
-  text_sensor::TextSensor *source_;
-};
-
-}  // namespace esphome::copy

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

@@ -501,7 +501,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)
@@ -509,12 +509,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"
@@ -573,7 +575,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(
@@ -594,8 +598,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)
@@ -616,11 +624,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/unit_tests/core/test_entity_helpers.py

@@ -1522,6 +1522,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"),
@@ -1713,6 +1714,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(
@@ -1732,6 +1739,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],