Register a stub transport for ESPHome serial proxies via USB

homeassistant/components/esphome/__init__.py

@@ -17,6 +17,7 @@
     CONF_HOST,
     CONF_PASSWORD,
     CONF_PORT,
+    EVENT_HOMEASSISTANT_STOP,
     __version__ as ha_version,
 )
 from homeassistant.core import HomeAssistant, callback
@@ -80,7 +81,10 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
 
     if "usb" in hass.config.components:
         async_register_serial_port_scanner(hass, _async_scan_serial_ports)
-        serial_proxy.set_hass_loop(hass.loop)
+        hass.bus.async_listen_once(
+            EVENT_HOMEASSISTANT_STOP,
+            serial_proxy.register_serialx_transport(hass.loop),
+        )
 
     return True
 

homeassistant/components/esphome/serial_proxy.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Callable
 from typing import cast
 
 from aioesphomeapi import APIClient
@@ -15,25 +16,17 @@
 from yarl import URL
 
 from homeassistant.config_entries import ConfigEntryState
-from homeassistant.core import HomeAssistant, async_get_hass
+from homeassistant.core import Event, HomeAssistant, async_get_hass, callback
 
 from .const import DOMAIN
 from .entry_data import ESPHomeConfigEntry
 
-SCHEME = "esphome-hass://"
-
 # This is required so that serialx can safely query Core for an instance of an
 # aioesphomeapi client. We cannot make any assumptions here, some packages run separate
 # asyncio event loops in dedicated threads.
 _HASS_LOOP: asyncio.AbstractEventLoop | None = None
 
 
-def set_hass_loop(loop: asyncio.AbstractEventLoop) -> None:
-    """Store a reference to the Core event loop."""
-    global _HASS_LOOP  # noqa: PLW0603  # pylint: disable=global-statement
-    _HASS_LOOP = loop
-
-
 def build_url(entry_id: str, port_name: str) -> URL:
     """Build a canonical `esphome-hass://` URL."""
     return URL.build(
@@ -105,9 +98,24 @@ class HassESPHomeSerialTransport(ESPHomeSerialTransport):
     _serial_cls = HassESPHomeSerial
 
 
-register_uri_handler(
-    scheme=SCHEME,
-    unique_scheme=SCHEME,
-    sync_cls=HassESPHomeSerial,
-    async_transport_cls=HassESPHomeSerialTransport,
-)
+def register_serialx_transport(
+    loop: asyncio.AbstractEventLoop,
+) -> Callable[[Event], None]:
+    """Register the ESPHome URI handler."""
+    global _HASS_LOOP  # noqa: PLW0603  # pylint: disable=global-statement
+    _HASS_LOOP = loop
+
+    unregister = register_uri_handler(
+        scheme="esphome-hass://",
+        unique_scheme="esphome-hass-internal://",  # The unique scheme must differ
+        sync_cls=HassESPHomeSerial,
+        async_transport_cls=HassESPHomeSerialTransport,
+    )
+
+    @callback
+    def _unregister(event: Event) -> None:
+        global _HASS_LOOP  # noqa: PLW0603  # pylint: disable=global-statement
+        unregister()
+        _HASS_LOOP = None
+
+    return _unregister

homeassistant/components/usb/__init__.py

@@ -35,6 +35,7 @@
 
 from .const import DOMAIN
 from .models import SerialDevice, USBDevice
+from .serial_proxy_stub import register_serialx_transport
 from .utils import (
     scan_serial_ports,
     usb_device_from_path,
@@ -187,6 +188,8 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     websocket_api.async_register_command(hass, websocket_usb_scan)
     websocket_api.async_register_command(hass, websocket_usb_list_serial_ports)
 
+    hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, register_serialx_transport())
+
     return True
 
 

homeassistant/components/usb/serial_proxy_stub.py

@@ -0,0 +1,43 @@
+"""ESPHome serial proxy URI handler stub for serialx."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+
+from serialx import register_uri_handler
+from serialx.platforms.serial_esphome import ESPHomeSerial, ESPHomeSerialTransport
+
+from homeassistant.core import Event, callback
+from homeassistant.exceptions import ConfigEntryNotReady
+
+
+class HassESPHomeSerialStub(ESPHomeSerial):
+    """ESPHomeSerial that throws `ConfigEntryNotReady` until ESPHome itself loads."""
+
+    async def _async_open(self) -> None:
+        """Open a connection."""
+        raise ConfigEntryNotReady("ESPHome has not loaded yet")
+
+
+class HassESPHomeSerialStubTransport(ESPHomeSerialTransport):
+    """Transport variant that constructs `HassESPHomeSerialStub`."""
+
+    transport_name = "esphome-hass"
+    _serial_cls = HassESPHomeSerialStub
+
+
+def register_serialx_transport() -> Callable[[Event], None]:
+    """Register the stub URI handler."""
+    unregister = register_uri_handler(
+        scheme="esphome-hass://",
+        unique_scheme="esphome-hass-usb://",
+        sync_cls=HassESPHomeSerialStub,
+        async_transport_cls=HassESPHomeSerialStubTransport,
+        weight=-1,  # We want the ESPHome integration transport to take precedence
+    )
+
+    @callback
+    def _unregister(event: Event) -> None:
+        unregister()
+
+    return _unregister

tests/components/usb/test_init.py

@@ -7,15 +7,17 @@
 from unittest.mock import MagicMock, Mock, call, patch, sentinel
 
 import pytest
-from serialx import SerialPortInfo
+from serialx import SerialPortInfo, create_serial_connection, serial_for_url
 
 from homeassistant import config_entries
 from homeassistant.components import usb
 from homeassistant.components.usb import DOMAIN, async_scan_serial_ports
 from homeassistant.components.usb.models import SerialDevice, USBDevice
+from homeassistant.components.usb.serial_proxy_stub import HassESPHomeSerialStub
 from homeassistant.components.usb.utils import usb_device_from_path
 from homeassistant.const import EVENT_HOMEASSISTANT_STARTED, EVENT_HOMEASSISTANT_STOP
 from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.setup import async_setup_component
 from homeassistant.util import dt as dt_util
 
@@ -1752,3 +1754,20 @@ async def test_list_serial_ports_os_error(
     assert not response["success"]
     assert response["error"]["code"] == "unknown_error"
     assert "Permission denied" in response["error"]["message"]
+
+
+async def test_serial_proxy_stub_sync(hass: HomeAssistant) -> None:
+    """Test ESPHome serial proxy stub."""
+    assert await async_setup_component(hass, DOMAIN, {})
+
+    serial_cls = serial_for_url("esphome-hass-usb://192.0.2.1")
+    assert isinstance(serial_cls, HassESPHomeSerialStub)
+
+    # Nothing actually opens, it just throws an error
+    with pytest.raises(ConfigEntryNotReady):
+        await create_serial_connection(
+            loop=asyncio.get_running_loop(),
+            protocol_factory=asyncio.Protocol,
+            url="esphome-hass-usb://192.0.2.1",
+            baudrate=115200,
+        )