8158ca7c69
* Add connection test feature to assist_satellite * Add http to assist_satellite dependencies * Remove extra logging * Incorporate feedback * Fix tests * ruff * Apply suggestions from code review Co-authored-by: Bram Kragten <mail@bramkragten.nl> * Use asyncio.Event instead of dispatcher * Respond asap * Update homeassistant/components/assist_satellite/websocket_api.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> --------- Co-authored-by: Michael Hansen <mike@rhasspy.org> Co-authored-by: Paulus Schoutsen <balloob@gmail.com> Co-authored-by: Bram Kragten <mail@bramkragten.nl> Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
"""Assist satellite connection test."""
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
from aiohttp import web
|
|
|
|
from homeassistant.components.http import KEY_HASS, HomeAssistantView
|
|
|
|
from .const import CONNECTION_TEST_DATA
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
CONNECTION_TEST_CONTENT_TYPE = "audio/mpeg"
|
|
CONNECTION_TEST_FILENAME = "connection_test.mp3"
|
|
CONNECTION_TEST_URL_BASE = "/api/assist_satellite/connection_test"
|
|
|
|
|
|
class ConnectionTestView(HomeAssistantView):
|
|
"""View to serve an audio sample for connection test."""
|
|
|
|
requires_auth = False
|
|
url = f"{CONNECTION_TEST_URL_BASE}/{{connection_id}}"
|
|
name = "api:assist_satellite_connection_test"
|
|
|
|
async def get(self, request: web.Request, connection_id: str) -> web.Response:
|
|
"""Start a get request."""
|
|
_LOGGER.debug("Request for connection test with id %s", connection_id)
|
|
|
|
hass = request.app[KEY_HASS]
|
|
connection_test_data = hass.data[CONNECTION_TEST_DATA]
|
|
|
|
connection_test_event = connection_test_data.pop(connection_id, None)
|
|
|
|
if connection_test_event is None:
|
|
return web.Response(status=404)
|
|
|
|
connection_test_event.set()
|
|
|
|
audio_path = Path(__file__).parent / CONNECTION_TEST_FILENAME
|
|
audio_data = await hass.async_add_executor_job(audio_path.read_bytes)
|
|
|
|
return web.Response(body=audio_data, content_type=CONNECTION_TEST_CONTENT_TYPE)
|