ESPHome binary sensor representing assist pipeline running (#91406)

* ESPHome binary sensor representing assist pipeline running

* Apply suggestions from code review

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>

* Rename to call active
Simplify with attrs a little

* Load binary sensor if voice assistant on device

* Add some tests

---------

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>

homeassistant/components/esphome/__init__.py

@@ -308,6 +308,7 @@ async def async_setup_entry(  # noqa: C901
             voice_assistant_udp_server.run_pipeline(handle_pipeline_event),
             "esphome.voice_assistant_udp_server.run_pipeline",
         )
+        entry_data.async_set_assist_pipeline_state(True)
 
         return port
 
@@ -315,6 +316,8 @@ async def async_setup_entry(  # noqa: C901
         """Stop a voice assistant pipeline."""
         nonlocal voice_assistant_udp_server
 
+        entry_data.async_set_assist_pipeline_state(False)
+
         if voice_assistant_udp_server is not None:
             voice_assistant_udp_server.stop()
             voice_assistant_udp_server = None
@@ -894,3 +897,40 @@ class EsphomeEntity(Entity, Generic[_InfoT, _StateT]):
         if not self._static_info.entity_category:
             return None
         return ENTITY_CATEGORIES.from_esphome(self._static_info.entity_category)
+
+
+class EsphomeAssistEntity(Entity):
+    """Define a base entity for Assist Pipeline entities."""
+
+    _attr_has_entity_name = True
+    _attr_should_poll = False
+
+    def __init__(self, entry_data: RuntimeEntryData) -> None:
+        """Initialize the binary sensor."""
+        self._entry_data: RuntimeEntryData = entry_data
+        self._attr_unique_id = (
+            f"{self._device_info.mac_address}-{self.entity_description.key}"
+        )
+
+    @property
+    def _device_info(self) -> EsphomeDeviceInfo:
+        assert self._entry_data.device_info is not None
+        return self._entry_data.device_info
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Return device registry information for this entity."""
+        return DeviceInfo(
+            connections={(dr.CONNECTION_NETWORK_MAC, self._device_info.mac_address)}
+        )
+
+    @callback
+    def _update(self) -> None:
+        self.async_write_ha_state()
+
+    async def async_added_to_hass(self) -> None:
+        """Register update callback."""
+        await super().async_added_to_hass()
+        self.async_on_remove(
+            self._entry_data.async_subscribe_assist_pipeline_update(self._update)
+        )

homeassistant/components/esphome/binary_sensor.py

@@ -6,13 +6,15 @@ from aioesphomeapi import BinarySensorInfo, BinarySensorState
 from homeassistant.components.binary_sensor import (
     BinarySensorDeviceClass,
     BinarySensorEntity,
+    BinarySensorEntityDescription,
 )
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.util.enum import try_parse_enum
 
-from . import EsphomeEntity, platform_async_setup_entry
+from . import EsphomeAssistEntity, EsphomeEntity, platform_async_setup_entry
+from .domain_data import DomainData
 
 
 async def async_setup_entry(
@@ -29,6 +31,11 @@ async def async_setup_entry(
         state_type=BinarySensorState,
     )
 
+    entry_data = DomainData.get(hass).get_entry_data(entry)
+    assert entry_data.device_info is not None
+    if entry_data.device_info.voice_assistant_version:
+        async_add_entities([EsphomeCallActiveBinarySensor(entry_data)])
+
 
 class EsphomeBinarySensor(
     EsphomeEntity[BinarySensorInfo, BinarySensorState], BinarySensorEntity
@@ -59,3 +66,17 @@ class EsphomeBinarySensor(
         if self._static_info.is_status_binary_sensor:
             return True
         return super().available
+
+
+class EsphomeCallActiveBinarySensor(EsphomeAssistEntity, BinarySensorEntity):
+    """A binary sensor implementation for ESPHome for use with assist_pipeline."""
+
+    entity_description = BinarySensorEntityDescription(
+        key="call_active",
+        translation_key="call_active",
+    )
+
+    @property
+    def is_on(self) -> bool | None:
+        """Return true if the binary sensor is on."""
+        return self._entry_data.assist_pipeline_state

homeassistant/components/esphome/entry_data.py

@@ -99,6 +99,10 @@ class RuntimeEntryData:
     _ble_connection_free_futures: list[asyncio.Future[int]] = field(
         default_factory=list
     )
+    assist_pipeline_update_callbacks: list[Callable[[], None]] = field(
+        default_factory=list
+    )
+    assist_pipeline_state: bool = False
 
     @property
     def name(self) -> str:
@@ -153,6 +157,24 @@ class RuntimeEntryData:
         self._ble_connection_free_futures.append(fut)
         return await fut
 
+    @callback
+    def async_set_assist_pipeline_state(self, state: bool) -> None:
+        """Set the assist pipeline state."""
+        self.assist_pipeline_state = state
+        for update_callback in self.assist_pipeline_update_callbacks:
+            update_callback()
+
+    def async_subscribe_assist_pipeline_update(
+        self, update_callback: Callable[[], None]
+    ) -> Callable[[], None]:
+        """Subscribe to assist pipeline updates."""
+
+        def _unsubscribe() -> None:
+            self.assist_pipeline_update_callbacks.remove(update_callback)
+
+        self.assist_pipeline_update_callbacks.append(update_callback)
+        return _unsubscribe
+
     @callback
     def async_remove_entity(
         self, hass: HomeAssistant, component_key: str, key: int
@@ -180,6 +202,9 @@ class RuntimeEntryData:
         if async_get_dashboard(hass):
             needed_platforms.add(Platform.UPDATE)
 
+        if self.device_info is not None and self.device_info.voice_assistant_version:
+            needed_platforms.add(Platform.BINARY_SENSOR)
+
         for info in infos:
             for info_type, platform in INFO_TYPE_TO_PLATFORM.items():
                 if isinstance(info, info_type):

homeassistant/components/esphome/strings.json

@@ -46,6 +46,13 @@
     },
     "flow_title": "{name}"
   },
+  "entity": {
+    "binary_sensor": {
+      "call_active": {
+        "name": "Call Active"
+      }
+    }
+  },
   "issues": {
     "ble_firmware_outdated": {
       "title": "Update {name} with ESPHome {version} or later",

tests/components/esphome/conftest.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 
 from unittest.mock import AsyncMock, Mock, patch
 
-from aioesphomeapi import APIClient, DeviceInfo
+from aioesphomeapi import APIClient, APIVersion, DeviceInfo
 import pytest
 from zeroconf import Zeroconf
 
@@ -101,6 +101,8 @@ def mock_client(mock_device_info):
     mock_client.device_info = AsyncMock(return_value=mock_device_info)
     mock_client.connect = AsyncMock()
     mock_client.disconnect = AsyncMock()
+    mock_client.list_entities_services = AsyncMock(return_value=([], []))
+    mock_client.api_version = APIVersion(99, 99)
 
     with patch("homeassistant.components.esphome.APIClient", mock_client), patch(
         "homeassistant.components.esphome.config_flow.APIClient", mock_client

tests/components/esphome/test_binary_sensor.py

@@ -0,0 +1,54 @@
+"""Test ESPHome binary sensors."""
+from unittest.mock import AsyncMock, Mock
+
+from aioesphomeapi import DeviceInfo
+
+from homeassistant.components.esphome import DOMAIN, DomainData
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+
+async def test_call_active(
+    hass: HomeAssistant,
+    mock_client,
+) -> None:
+    """Test call active binary sensor."""
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: "test.local", CONF_PORT: 6053, CONF_PASSWORD: ""},
+    )
+    entry.add_to_hass(hass)
+
+    device_info = DeviceInfo(
+        name="test",
+        friendly_name="Test",
+        voice_assistant_version=1,
+        mac_address="11:22:33:44:55:aa",
+        esphome_version="1.0.0",
+    )
+
+    mock_client.device_info = AsyncMock(return_value=device_info)
+    mock_client.subscribe_voice_assistant = AsyncMock(return_value=Mock())
+
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+    await hass.async_block_till_done()
+    await hass.async_block_till_done()
+
+    entry_data = DomainData.get(hass).get_entry_data(entry)
+
+    state = hass.states.get("binary_sensor.test_call_active")
+    assert state is not None
+    assert state.state == "off"
+
+    entry_data.async_set_assist_pipeline_state(True)
+
+    state = hass.states.get("binary_sensor.test_call_active")
+    assert state.state == "on"
+
+    entry_data.async_set_assist_pipeline_state(False)
+
+    state = hass.states.get("binary_sensor.test_call_active")
+    assert state.state == "off"