Fix log runner using zeroconf sync close running in the event loop

- Use the log runner code from aioesphomeapi since it handles zeroconf
  correctly

- Avoid calling safe_print since it has a slow late import and does
  mulitple bytes/string conversions

esphome/components/api/client.py

@@ -1,19 +1,22 @@
 import asyncio
 import logging
 from datetime import datetime
-from typing import Optional
+from typing import Any, Optional
 
-from aioesphomeapi import APIClient, ReconnectLogic, APIConnectionError, LogLevel
+from aioesphomeapi import APIClient
-import zeroconf
+from aioesphomeapi.api_pb2 import SubscribeLogsResponse
+from aioesphomeapi.log_runner import async_run
+
+from esphome.const import CONF_KEY, CONF_PASSWORD, CONF_PORT, __version__
+from esphome.core import CORE
 
-from esphome.const import CONF_KEY, CONF_PORT, CONF_PASSWORD, __version__
-from esphome.util import safe_print
 from . import CONF_ENCRYPTION
 
 _LOGGER = logging.getLogger(__name__)
 
 
 async def async_run_logs(config, address):
+    """Run the logs command in the event loop."""
     conf = config["api"]
     port: int = int(conf[CONF_PORT])
     password: str = conf[CONF_PASSWORD]
@@ -28,44 +31,28 @@ async def async_run_logs(config, address):
         client_info=f"ESPHome Logs {__version__}",
         noise_psk=noise_psk,
     )
-    first_connect = True
+    dashboard = CORE.dashboard
 
-    def on_log(msg):
+    def on_log(msg: SubscribeLogsResponse) -> None:
-        time_ = datetime.now().time().strftime("[%H:%M:%S]")
+        """Handle a new log message."""
-        text = msg.message.decode("utf8", "backslashreplace")
+        time_ = datetime.now()
-        safe_print(time_ + text)
+        message: bytes = msg.message
-
+        text = message.decode("utf8", "backslashreplace")
-    async def on_connect():
+        if dashboard:
-        nonlocal first_connect
+            text = text.replace("\033", "\\033")
-        try:
+        print(f"[{time_.hour:02}:{time_.minute:02}:{time_.second:02}]{text}")
-            await cli.subscribe_logs(
-                on_log,
-                log_level=LogLevel.LOG_LEVEL_VERY_VERBOSE,
-                dump_config=first_connect,
-            )
-            first_connect = False
-        except APIConnectionError:
-            cli.disconnect()
-
-    async def on_disconnect(expected_disconnect: bool) -> None:
-        _LOGGER.warning("Disconnected from API")
-
-    zc = zeroconf.Zeroconf()
-    reconnect = ReconnectLogic(
-        client=cli,
-        on_connect=on_connect,
-        on_disconnect=on_disconnect,
-        zeroconf_instance=zc,
-    )
-    await reconnect.start()
 
+    stop = await async_run(cli, on_log)
     try:
         while True:
             await asyncio.sleep(60)
+    finally:
+        await stop()
+
+
+def run_logs(config: dict[str, Any], address: str) -> None:
+    """Run the logs command."""
+    try:
+        asyncio.run(async_run_logs(config, address))
     except KeyboardInterrupt:
-        await reconnect.stop()
+        pass
-        zc.close()
-
-
-def run_logs(config, address):
-    asyncio.run(async_run_logs(config, address))

requirements.txt

@@ -10,7 +10,7 @@ platformio==6.1.11  # When updating platformio, also update Dockerfile
 esptool==4.6.2
 click==8.1.7
 esphome-dashboard==20231107.0
-aioesphomeapi==18.2.7
+aioesphomeapi==18.4.0
 zeroconf==0.122.3
 
 # esp-idf requires this, but doesn't bundle it by default