[core] Add status_momentary_warning/error overloads without a name (#20362)

esphome/core/component.cpp

@@ -292,14 +292,17 @@ void Component::status_set_warning(const LogString *message) {
 }
 void Component::status_set_error() { this->status_set_error((const LogString *) nullptr); }
 void Component::status_set_error(const LogString *message) {
-  if (!this->set_status_flag_(STATUS_LED_ERROR))
-    return;
-  ESP_LOGE(TAG, "%s set Error flag: %s", LOG_STR_ARG(this->get_component_log_str()),
-           message ? LOG_STR_ARG(message) : LOG_STR_LITERAL("unspecified"));
-  if (message != nullptr) {
+  if (this->set_error_flag_(message) && message != nullptr) {
     store_component_error_message(this, message);
   }
 }
+bool Component::set_error_flag_(const LogString *message) {
+  if (!this->set_status_flag_(STATUS_LED_ERROR))
+    return false;
+  ESP_LOGE(TAG, "%s set Error flag: %s", LOG_STR_ARG(this->get_component_log_str()),
+           message ? LOG_STR_ARG(message) : LOG_STR_LITERAL("unspecified"));
+  return true;
+}
 void Component::status_clear_warning_slow_path_() {
   this->component_state_ &= ~STATUS_LED_WARNING;
   // Clear the app-wide STATUS_LED_WARNING bit only if setup has finished
@@ -329,6 +332,21 @@ void Component::status_momentary_error(const char *name, uint32_t length) {
   this->status_set_error();
   this->set_timeout(name, length, [this]() { this->status_clear_error(); });
 }
+void Component::status_momentary_warning(uint32_t length) {
+  this->status_momentary_warning(static_cast<const LogString *>(nullptr), length);
+}
+void Component::status_momentary_warning(const LogString *message, uint32_t length) {
+  this->status_set_warning(message);
+  this->set_timeout(InternalSchedulerID::STATUS_WARNING, length, [this]() { this->status_clear_warning(); });
+}
+void Component::status_momentary_error(uint32_t length) {
+  this->status_momentary_error(static_cast<const LogString *>(nullptr), length);
+}
+void Component::status_momentary_error(const LogString *message, uint32_t length) {
+  // Not stored: a cleared momentary error must not show up later as a FAILED reason
+  this->set_error_flag_(message);
+  this->set_timeout(InternalSchedulerID::STATUS_ERROR, length, [this]() { this->status_clear_error(); });
+}
 void Component::dump_config() {}
 
 // Function implementation of LOG_UPDATE_INTERVAL macro to reduce code size

esphome/core/component.h

@@ -67,6 +67,8 @@ inline constexpr uint32_t SCHEDULER_DONT_RUN = 4294967295UL;
 /// with component-level NUMERIC_ID values, even if the uint32_t values overlap.
 enum class InternalSchedulerID : uint32_t {
   POLLING_UPDATE = 0,  // PollingComponent interval
+  STATUS_WARNING = 1,  // status_momentary_warning() without a name
+  STATUS_ERROR = 2,    // status_momentary_error() without a name
 };
 
 // Forward declaration
@@ -307,6 +309,11 @@ class Component {
    */
   void status_momentary_warning(const char *name, uint32_t length = 5000);
 
+  /// Set warning status flag and clear it after `length` ms. A new call restarts the timeout.
+  /// `message` is logged in place of "unspecified" when the flag gets set.
+  void status_momentary_warning(uint32_t length = 5000);
+  void status_momentary_warning(const LogString *message, uint32_t length = 5000);
+
   /** Set error status flag and automatically clear it after a timeout.
    *
    * @param name Identifier for the timeout (used to cancel/replace existing timeouts with the same name).
@@ -316,6 +323,11 @@ class Component {
    */
   void status_momentary_error(const char *name, uint32_t length = 5000);
 
+  /// Set error status flag and clear it after `length` ms. A new call restarts the timeout.
+  /// `message` is logged in place of "unspecified"; unlike status_set_error() it is not stored.
+  void status_momentary_error(uint32_t length = 5000);
+  void status_momentary_error(const LogString *message, uint32_t length = 5000);
+
   bool has_overridden_loop() const { return (this->component_state_ & COMPONENT_HAS_LOOP) != 0; }
 
   /** Get the integration where this component was declared as a LogString for logging.
@@ -356,6 +368,8 @@ class Component {
   /// Note: Callers often use the return value to decide whether to log a warning/error,
   /// so once a flag is set, subsequent (potentially different) messages may be suppressed.
   bool set_status_flag_(uint8_t flag);
+  /// Set the error flag and log it; true when it was not set before.
+  bool set_error_flag_(const LogString *message);
 
   /** Set an interval function with a const char* name. Empty name means no cancelling possible.
    *

tests/integration/fixtures/status_momentary.yaml

@@ -0,0 +1,40 @@
+esphome:
+  name: status-momentary-test
+
+host:
+api:
+  actions:
+    - action: momentary_warning
+      variables:
+        length: int
+      then:
+        - lambda: "id(probe)->status_momentary_warning(length);"
+    - action: momentary_error
+      variables:
+        length: int
+      then:
+        - lambda: "id(probe)->status_momentary_error(length);"
+    - action: momentary_warning_message
+      then:
+        - lambda: 'id(probe)->status_momentary_warning(LOG_STR("probe warning reason"), 300);'
+    - action: momentary_error_message
+      then:
+        - lambda: 'id(probe)->status_momentary_error(LOG_STR("probe error reason"), 300);'
+
+logger:
+
+sensor:
+  - platform: template
+    id: probe
+    update_interval: 24h
+    lambda: return 1.0;
+
+binary_sensor:
+  - platform: template
+    name: Probe Warning
+    id: probe_warning
+    lambda: return id(probe)->status_has_warning();
+  - platform: template
+    name: Probe Error
+    id: probe_error
+    lambda: return id(probe)->status_has_error();

tests/integration/test_status_momentary.py

@@ -0,0 +1,87 @@
+"""Integration test for status_momentary_warning/error without a name."""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import BinarySensorState, EntityState
+import pytest
+
+from .state_utils import InitialStateHelper
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_status_momentary(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    loop = asyncio.get_running_loop()
+    warning_logged = loop.create_future()
+    error_logged = loop.create_future()
+
+    def on_log_line(line: str) -> None:
+        if (
+            "set Warning flag: probe warning reason" in line
+            and not warning_logged.done()
+        ):
+            warning_logged.set_result(True)
+        if "set Error flag: probe error reason" in line and not error_logged.done():
+            error_logged.set_result(True)
+
+    async with (
+        run_compiled(yaml_config, line_callback=on_log_line),
+        api_client_connected() as client,
+    ):
+        entities, services = await client.list_entities_services()
+        svc = {s.name: s for s in services}
+        keys = {e.object_id: e.key for e in entities}
+        warning_key = keys["probe_warning"]
+        error_key = keys["probe_error"]
+
+        changes: asyncio.Queue[tuple[int, bool, float]] = asyncio.Queue()
+
+        def on_state(state: EntityState) -> None:
+            if isinstance(state, BinarySensorState):
+                changes.put_nowait((state.key, state.state, loop.time()))
+
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+        await initial_state_helper.wait_for_initial_states()
+        assert initial_state_helper.initial_states[warning_key].state is False
+        assert initial_state_helper.initial_states[error_key].state is False
+
+        async def next_change(key: int) -> tuple[bool, float]:
+            while True:
+                k, value, when = await asyncio.wait_for(changes.get(), timeout=5)
+                if k == key:
+                    return value, when
+
+        # The flag sets at once and clears after the length
+        for service, key in (
+            ("momentary_warning", warning_key),
+            ("momentary_error", error_key),
+        ):
+            start = loop.time()
+            await client.execute_service(svc[service], {"length": 300})
+            assert (await next_change(key))[0] is True
+            value, cleared = await next_change(key)
+            assert value is False
+            assert cleared - start >= 0.25
+
+        # A second call restarts the timeout instead of adding a second one
+        start = loop.time()
+        await client.execute_service(svc["momentary_warning"], {"length": 1000})
+        assert (await next_change(warning_key))[0] is True
+        await asyncio.sleep(0.6)
+        await client.execute_service(svc["momentary_warning"], {"length": 1000})
+        value, cleared = await next_change(warning_key)
+        assert value is False
+        assert cleared - start >= 1.4
+
+        # A message replaces "unspecified" in the log line
+        await client.execute_service(svc["momentary_warning_message"], {})
+        await asyncio.wait_for(warning_logged, timeout=5)
+        await client.execute_service(svc["momentary_error_message"], {})
+        await asyncio.wait_for(error_logged, timeout=5)