Adjust remote ESPHome log subscription level on logging change (#139308)

Author: bdraco

homeassistant/components/esphome/manager.py

@@ -35,6 +35,7 @@
     Platform,
 )
 from homeassistant.core import (
+    CALLBACK_TYPE,
     Event,
     EventStateChangedData,
     HomeAssistant,
@@ -95,6 +96,14 @@
     LogLevel.LOG_LEVEL_VERBOSE: logging.DEBUG,
     LogLevel.LOG_LEVEL_VERY_VERBOSE: logging.DEBUG,
 }
+LOGGER_TO_LOG_LEVEL = {
+    logging.NOTSET: LogLevel.LOG_LEVEL_VERY_VERBOSE,
+    logging.DEBUG: LogLevel.LOG_LEVEL_VERY_VERBOSE,
+    logging.INFO: LogLevel.LOG_LEVEL_CONFIG,
+    logging.WARNING: LogLevel.LOG_LEVEL_WARN,
+    logging.ERROR: LogLevel.LOG_LEVEL_ERROR,
+    logging.CRITICAL: LogLevel.LOG_LEVEL_ERROR,
+}
 # 7-bit and 8-bit C1 ANSI sequences
 # https://stackoverflow.com/questions/14693701/how-can-i-remove-the-ansi-escape-sequences-from-a-string-in-python
 ANSI_ESCAPE_78BIT = re.compile(
@@ -161,6 +170,8 @@ class ESPHomeManager:
     """Class to manage an ESPHome connection."""
 
     __slots__ = (
+        "_cancel_subscribe_logs",
+        "_log_level",
         "cli",
         "device_id",
         "domain_data",
@@ -194,6 +205,8 @@ def __init__(
         self.reconnect_logic: ReconnectLogic | None = None
         self.zeroconf_instance = zeroconf_instance
         self.entry_data = entry.runtime_data
+        self._cancel_subscribe_logs: CALLBACK_TYPE | None = None
+        self._log_level = LogLevel.LOG_LEVEL_NONE
 
     async def on_stop(self, event: Event) -> None:
         """Cleanup the socket client on HA close."""
@@ -368,15 +381,31 @@ async def on_connect(self) -> None:
 
     def _async_on_log(self, msg: SubscribeLogsResponse) -> None:
         """Handle a log message from the API."""
-        logger_level = LOG_LEVEL_TO_LOGGER.get(msg.level, logging.DEBUG)
-        if _LOGGER.isEnabledFor(logger_level):
-            log: bytes = msg.message
-            _LOGGER.log(
-                logger_level,
-                "%s: %s",
-                self.entry.title,
-                ANSI_ESCAPE_78BIT.sub(b"", log).decode("utf-8", "backslashreplace"),
-            )
+        log: bytes = msg.message
+        _LOGGER.log(
+            LOG_LEVEL_TO_LOGGER.get(msg.level, logging.DEBUG),
+            "%s: %s",
+            self.entry.title,
+            ANSI_ESCAPE_78BIT.sub(b"", log).decode("utf-8", "backslashreplace"),
+        )
+
+    @callback
+    def _async_get_equivalent_log_level(self) -> LogLevel:
+        """Get the equivalent ESPHome log level for the current logger."""
+        return LOGGER_TO_LOG_LEVEL.get(
+            _LOGGER.getEffectiveLevel(), LogLevel.LOG_LEVEL_VERY_VERBOSE
+        )
+
+    @callback
+    def _async_subscribe_logs(self, log_level: LogLevel) -> None:
+        """Subscribe to logs."""
+        if self._cancel_subscribe_logs is not None:
+            self._cancel_subscribe_logs()
+            self._cancel_subscribe_logs = None
+        self._log_level = log_level
+        self._cancel_subscribe_logs = self.cli.subscribe_logs(
+            self._async_on_log, self._log_level
+        )
 
     async def _on_connnect(self) -> None:
         """Subscribe to states and list entities on successful API login."""
@@ -390,7 +419,7 @@ async def _on_connnect(self) -> None:
         stored_device_name = entry.data.get(CONF_DEVICE_NAME)
         unique_id_is_mac_address = unique_id and ":" in unique_id
         if entry.options.get(CONF_SUBSCRIBE_LOGS):
-            cli.subscribe_logs(self._async_on_log, LogLevel.LOG_LEVEL_VERY_VERBOSE)
+            self._async_subscribe_logs(self._async_get_equivalent_log_level())
         results = await asyncio.gather(
             create_eager_task(cli.device_info()),
             create_eager_task(cli.list_entities_services()),
@@ -542,6 +571,10 @@ async def on_connect_error(self, err: Exception) -> None:
     def _async_handle_logging_changed(self, _event: Event) -> None:
         """Handle when the logging level changes."""
         self.cli.set_debug(_LOGGER.isEnabledFor(logging.DEBUG))
+        if self.entry.options.get(CONF_SUBSCRIBE_LOGS) and self._log_level != (
+            new_log_level := self._async_get_equivalent_log_level()
+        ):
+            self._async_subscribe_logs(new_log_level)
 
     async def async_start(self) -> None:
         """Start the esphome connection manager."""

tests/components/esphome/conftest.py

@@ -230,6 +230,7 @@ def __init__(
         )
         self.on_log_message: Callable[[SubscribeLogsResponse], None]
         self.device_info = device_info
+        self.current_log_level = LogLevel.LOG_LEVEL_NONE
 
     def set_state_callback(self, state_callback: Callable[[EntityState], None]) -> None:
         """Set the state callback."""
@@ -432,9 +433,11 @@ def _subscribe_home_assistant_states(
 
     def _subscribe_logs(
         on_log_message: Callable[[SubscribeLogsResponse], None], log_level: LogLevel
-    ) -> None:
+    ) -> Callable[[], None]:
         """Subscribe to log messages."""
         mock_device.set_on_log_message(on_log_message)
+        mock_device.current_log_level = log_level
+        return lambda: None
 
     def _subscribe_voice_assistant(
         *,

tests/components/esphome/test_manager.py

@@ -57,6 +57,7 @@ async def test_esphome_device_subscribe_logs(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Test configuring a device to subscribe to logs."""
+    assert await async_setup_component(hass, "logger", {"logger": {}})
     entry = MockConfigEntry(
         domain=DOMAIN,
         data={
@@ -76,6 +77,15 @@ async def test_esphome_device_subscribe_logs(
         states=[],
     )
     await hass.async_block_till_done()
+
+    await hass.services.async_call(
+        "logger",
+        "set_level",
+        {"homeassistant.components.esphome": "DEBUG"},
+        blocking=True,
+    )
+    assert device.current_log_level == LogLevel.LOG_LEVEL_VERY_VERBOSE
+
     caplog.set_level(logging.DEBUG)
     device.mock_on_log_message(
         Mock(level=LogLevel.LOG_LEVEL_INFO, message=b"test_log_message")
@@ -103,6 +113,28 @@ async def test_esphome_device_subscribe_logs(
     await hass.async_block_till_done()
     assert "test_debug_log_message" in caplog.text
 
+    await hass.services.async_call(
+        "logger",
+        "set_level",
+        {"homeassistant.components.esphome": "WARNING"},
+        blocking=True,
+    )
+    assert device.current_log_level == LogLevel.LOG_LEVEL_WARN
+    await hass.services.async_call(
+        "logger",
+        "set_level",
+        {"homeassistant.components.esphome": "ERROR"},
+        blocking=True,
+    )
+    assert device.current_log_level == LogLevel.LOG_LEVEL_ERROR
+    await hass.services.async_call(
+        "logger",
+        "set_level",
+        {"homeassistant.components.esphome": "INFO"},
+        blocking=True,
+    )
+    assert device.current_log_level == LogLevel.LOG_LEVEL_CONFIG
+
 
 async def test_esphome_device_service_calls_not_allowed(
     hass: HomeAssistant,