Merge branch 'synesthesiam-20250306-wyoming-info-websocket' into test-voice

This commit is contained in:
Bram Kragten 2025-03-26 09:07:51 +01:00
commit c4e9aeefea
4 changed files with 126 additions and 4 deletions

View File

@ -8,15 +8,19 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform from homeassistant.const import Platform
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.typing import ConfigType
from .const import ATTR_SPEAKER, DOMAIN from .const import ATTR_SPEAKER, DOMAIN
from .data import WyomingService from .data import WyomingService
from .devices import SatelliteDevice from .devices import SatelliteDevice
from .models import DomainDataItem from .models import DomainDataItem
from .websocket_api import async_register_websocket_api
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
SATELLITE_PLATFORMS = [ SATELLITE_PLATFORMS = [
Platform.ASSIST_SATELLITE, Platform.ASSIST_SATELLITE,
Platform.BINARY_SENSOR, Platform.BINARY_SENSOR,
@ -28,11 +32,19 @@ SATELLITE_PLATFORMS = [
__all__ = [ __all__ = [
"ATTR_SPEAKER", "ATTR_SPEAKER",
"DOMAIN", "DOMAIN",
"async_setup",
"async_setup_entry", "async_setup_entry",
"async_unload_entry", "async_unload_entry",
] ]
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Wyoming integration."""
async_register_websocket_api(hass)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Load Wyoming.""" """Load Wyoming."""
service = await WyomingService.create(entry.data["host"], entry.data["port"]) service = await WyomingService.create(entry.data["host"], entry.data["port"])

View File

@ -0,0 +1,42 @@
"""Wyoming Websocket API."""
import logging
from typing import Any
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.core import HomeAssistant, callback
from .const import DOMAIN
from .models import DomainDataItem
_LOGGER = logging.getLogger(__name__)
@callback
def async_register_websocket_api(hass: HomeAssistant) -> None:
"""Register the websocket API."""
websocket_api.async_register_command(hass, websocket_info)
@callback
@websocket_api.require_admin
@websocket_api.websocket_command({vol.Required("type"): "wyoming/info"})
def websocket_info(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""List service information for Wyoming all config entries."""
entry_items: dict[str, DomainDataItem] = hass.data.get(DOMAIN, {})
connection.send_result(
msg["id"],
{
"info": {
entry_id: item.service.info.to_dict()
for entry_id, item in entry_items.items()
}
},
)

View File

@ -121,7 +121,9 @@ def handle_config_entry(hass: HomeAssistant) -> ConfigEntry:
@pytest.fixture @pytest.fixture
async def init_wyoming_stt(hass: HomeAssistant, stt_config_entry: ConfigEntry): async def init_wyoming_stt(
hass: HomeAssistant, stt_config_entry: ConfigEntry
) -> ConfigEntry:
"""Initialize Wyoming STT.""" """Initialize Wyoming STT."""
with patch( with patch(
"homeassistant.components.wyoming.data.load_wyoming_info", "homeassistant.components.wyoming.data.load_wyoming_info",
@ -129,9 +131,13 @@ async def init_wyoming_stt(hass: HomeAssistant, stt_config_entry: ConfigEntry):
): ):
await hass.config_entries.async_setup(stt_config_entry.entry_id) await hass.config_entries.async_setup(stt_config_entry.entry_id)
return stt_config_entry
@pytest.fixture @pytest.fixture
async def init_wyoming_tts(hass: HomeAssistant, tts_config_entry: ConfigEntry): async def init_wyoming_tts(
hass: HomeAssistant, tts_config_entry: ConfigEntry
) -> ConfigEntry:
"""Initialize Wyoming TTS.""" """Initialize Wyoming TTS."""
with patch( with patch(
"homeassistant.components.wyoming.data.load_wyoming_info", "homeassistant.components.wyoming.data.load_wyoming_info",
@ -139,11 +145,13 @@ async def init_wyoming_tts(hass: HomeAssistant, tts_config_entry: ConfigEntry):
): ):
await hass.config_entries.async_setup(tts_config_entry.entry_id) await hass.config_entries.async_setup(tts_config_entry.entry_id)
return tts_config_entry
@pytest.fixture @pytest.fixture
async def init_wyoming_wake_word( async def init_wyoming_wake_word(
hass: HomeAssistant, wake_word_config_entry: ConfigEntry hass: HomeAssistant, wake_word_config_entry: ConfigEntry
): ) -> ConfigEntry:
"""Initialize Wyoming Wake Word.""" """Initialize Wyoming Wake Word."""
with patch( with patch(
"homeassistant.components.wyoming.data.load_wyoming_info", "homeassistant.components.wyoming.data.load_wyoming_info",
@ -151,6 +159,8 @@ async def init_wyoming_wake_word(
): ):
await hass.config_entries.async_setup(wake_word_config_entry.entry_id) await hass.config_entries.async_setup(wake_word_config_entry.entry_id)
return wake_word_config_entry
@pytest.fixture @pytest.fixture
async def init_wyoming_intent( async def init_wyoming_intent(

View File

@ -0,0 +1,58 @@
"""Websocket tests for Wyoming integration."""
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from tests.typing import WebSocketGenerator
async def test_info(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
init_components,
init_wyoming_stt: ConfigEntry,
init_wyoming_tts: ConfigEntry,
init_wyoming_wake_word: ConfigEntry,
init_wyoming_intent: ConfigEntry,
init_wyoming_handle: ConfigEntry,
) -> None:
"""Test info websocket command."""
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "wyoming/info"})
# result
msg = await client.receive_json()
assert msg["success"]
info = msg.get("result", {}).get("info", {})
# stt (speech-to-text) = asr (automated speech recognition)
assert init_wyoming_stt.entry_id in info
asr_info = info[init_wyoming_stt.entry_id].get("asr", [])
assert len(asr_info) == 1
assert asr_info[0].get("name") == "Test ASR"
# tts (text-to-speech)
assert init_wyoming_tts.entry_id in info
tts_info = info[init_wyoming_tts.entry_id].get("tts", [])
assert len(tts_info) == 1
assert tts_info[0].get("name") == "Test TTS"
# wake word detection
assert init_wyoming_wake_word.entry_id in info
wake_info = info[init_wyoming_wake_word.entry_id].get("wake", [])
assert len(wake_info) == 1
assert wake_info[0].get("name") == "Test Wake Word"
# intent recognition
assert init_wyoming_intent.entry_id in info
intent_info = info[init_wyoming_intent.entry_id].get("intent", [])
assert len(intent_info) == 1
assert intent_info[0].get("name") == "Test Intent"
# intent handling
assert init_wyoming_handle.entry_id in info
handle_info = info[init_wyoming_handle.entry_id].get("handle", [])
assert len(handle_info) == 1
assert handle_info[0].get("name") == "Test Handle"