/
/
/
1"""
2Live announcements: speech captured by a client and played on a player once it is spoken.
3
4A client holds an authenticated WebSocket on the MA webserver and pushes raw PCM frames
5for as long as the user speaks. Those frames land in a LiveAnnouncementSession, which
6serves them as a WAV url on the stream server. Once the clip is complete it is announced
7like any other: the announcement renderer pulls that url the same way it pulls a TTS
8clip, so every player plays it exactly as it plays the announcements it already knows.
9
10Wire format on the inbound WebSocket:
11- text {"type":"auth","token":...} - first message, omitted for Ingress connections
12- text {"type":"start","player_id":...,"sample_rate":...,"channels":...,
13 "pre_announce":...,"volume_level":...}
14- text {"type":"started"} back, after which audio is accepted
15- binary frames of raw little-endian signed 16-bit PCM in the announced format
16- text (any) to end the clip, or simply close the connection
17- text {"type":"finished"} or {"type":"error","message":...} back once it has played
18
19A rejected client is closed with code 4001 and the reason in the close frame.
20"""
21
22from __future__ import annotations
23
24import asyncio
25import secrets
26from contextlib import aclosing, suppress
27from typing import TYPE_CHECKING, NamedTuple
28
29from aiohttp import WSMsgType, web
30from music_assistant_models.auth import Scope
31from music_assistant_models.enums import ContentType
32from music_assistant_models.errors import MusicAssistantError, PlayerCommandFailed
33from music_assistant_models.media_items import AudioFormat
34from orjson import dumps
35
36from music_assistant.constants import HOMEASSISTANT_SYSTEM_USER
37from music_assistant.controllers.webserver.helpers.auth_middleware import (
38 get_authenticated_user,
39 has_scope,
40 is_request_from_ingress,
41 set_current_user,
42)
43from music_assistant.helpers.audio import create_streaming_wave_header
44from music_assistant.helpers.json import JSON_DECODE_EXCEPTIONS, json_loads
45
46from .announcements import MAX_ANNOUNCEMENT_SECONDS, MAX_CLIP_SECONDS
47
48if TYPE_CHECKING:
49 import logging
50 from collections.abc import AsyncGenerator, Callable
51
52 from music_assistant.mass import MusicAssistant
53
54# The client connects here to push its audio; the announcement itself is served from
55# the stream server, whose route only ever carries the session id.
56LIVE_ANNOUNCEMENT_ROUTE = "/live_announcement"
57LIVE_ANNOUNCEMENT_STREAM_PATH = "/live_announcement/{session_id}.wav"
58
59# Live audio is always uncompressed 16-bit PCM; only the rate and channel count vary,
60# and the client reports both in its start message. The clip is held in memory while it
61# is spoken, so the accepted range stops at what a capture device plausibly delivers -
62# anything beyond it is only a bigger buffer, since announcements render to 44.1kHz.
63LIVE_ANNOUNCEMENT_BIT_DEPTH = 16
64MIN_SAMPLE_RATE = 8000
65MAX_SAMPLE_RATE = 96000
66MAX_CHANNELS = 2
67
68# A client that stops sending without saying so holds the player's playback lock for as
69# long as it stays connected, so silence ends the clip on its own. The idle timeout only
70# measures the gap between frames, so a clip is bounded in wall clock time as well -
71# otherwise a trickle of frames could hold the player forever.
72IDLE_TIMEOUT = 10
73MAX_SESSION_SECONDS = MAX_ANNOUNCEMENT_SECONDS
74# Maximum time to wait for the handshake messages that precede the audio.
75HANDSHAKE_TIMEOUT = 10
76# Live announcements are spoken one at a time by a person, so a handful of sessions is
77# already generous; the cap bounds the audio that can be buffered at once.
78MAX_CONCURRENT_SESSIONS = 4
79# Room for the whole clip to play out and the player to be restored. A player provider
80# that never returns would otherwise hold one of the session slots until a restart.
81ANNOUNCEMENT_TIMEOUT = MAX_CLIP_SECONDS + 60
82# A close frame carries 125 bytes, two of which are taken by the status code.
83MAX_CLOSE_REASON_BYTES = 123
84
85
86class LiveAnnouncementStart(NamedTuple):
87 """The validated contents of a client's start message."""
88
89 player_id: str
90 audio_format: AudioFormat
91 pre_announce: bool | None
92 volume_level: int | None
93
94
95class LiveAnnouncementSession:
96 """
97 The audio of a single live announcement, buffered while it is being spoken.
98
99 The whole clip is kept until the session is dropped, so the renderer - which only
100 starts pulling once the clip is complete - can read it from the start, and can do so
101 again if it is served more than once.
102 """
103
104 def __init__(self, session_id: str, url: str, audio_format: AudioFormat) -> None:
105 """
106 Initialize the session.
107
108 :param session_id: The id that identifies this session in its stream url.
109 :param url: The url the announcement audio is served on.
110 :param audio_format: The format of the PCM frames the client sends.
111 """
112 self.session_id = session_id
113 self.url = url
114 self.audio_format = audio_format
115 self._chunks: list[bytes] = []
116 self._total_bytes = 0
117 self._finished = False
118 self._audio_added = asyncio.Condition()
119
120 @property
121 def duration(self) -> float:
122 """Return the duration (in seconds) of the audio received so far."""
123 return self._total_bytes / self.audio_format.pcm_sample_size
124
125 async def write(self, chunk: bytes) -> None:
126 """
127 Add a chunk of PCM audio to the clip.
128
129 :param chunk: Raw PCM audio in this session's format.
130 """
131 async with self._audio_added:
132 if self._finished:
133 return
134 self._chunks.append(chunk)
135 self._total_bytes += len(chunk)
136 self._audio_added.notify_all()
137
138 async def finish(self) -> None:
139 """End the clip, releasing any reader waiting for more audio."""
140 async with self._audio_added:
141 self._finished = True
142 self._audio_added.notify_all()
143
144 async def read(self) -> AsyncGenerator[bytes]:
145 """Yield the clip from the start, waiting for audio that is still being spoken."""
146 index = 0
147 while True:
148 async with self._audio_added:
149 while index >= len(self._chunks):
150 if self._finished:
151 return
152 await self._audio_added.wait()
153 chunk = self._chunks[index]
154 index += 1
155 # yielded outside the lock, so a slow reader never blocks the client
156 yield chunk
157
158
159class LiveAnnouncementManager:
160 """Owner of the live announcements that are currently being spoken."""
161
162 def __init__(self, mass: MusicAssistant, logger: logging.Logger) -> None:
163 """
164 Initialize the manager.
165
166 :param mass: The MusicAssistant instance.
167 :param logger: Logger of the streams controller this manager belongs to.
168 """
169 self.mass = mass
170 self.logger = logger.getChild("live_announcements")
171 self._sessions: dict[str, LiveAnnouncementSession] = {}
172 self._connections: set[web.WebSocketResponse] = set()
173 self._unregister: Callable[[], None] | None = None
174 self._closing = False
175
176 @property
177 def active_sessions(self) -> int:
178 """Return the number of live announcements currently in progress."""
179 return len(self._sessions)
180
181 def setup(self) -> None:
182 """Register the inbound audio route on the MA webserver."""
183 # a reload closes and sets this manager up again, so the shutdown flag is
184 # cleared here - it would otherwise silence every clip until a restart
185 self._closing = False
186 # this route rides on the webserver instead of the stream server because it is
187 # the only one of the two that is authenticated (and reachable over https).
188 self._unregister = self.mass.webserver.register_dynamic_route(
189 LIVE_ANNOUNCEMENT_ROUTE, self.handle_ws, "GET"
190 )
191
192 async def close(self) -> None:
193 """Unregister the route and disconnect any client that is still speaking."""
194 # closing a connection reads as a client that stopped speaking, so without this
195 # a recording in progress would announce itself on a server that is going down
196 self._closing = True
197 if self._unregister is not None:
198 self._unregister()
199 self._unregister = None
200 for ws in list(self._connections):
201 with suppress(Exception):
202 await ws.close()
203
204 async def serve_stream(self, request: web.Request) -> web.StreamResponse:
205 """Serve the audio of a live announcement to the announcement renderer."""
206 session_id = request.match_info["session_id"]
207 if (session := self._sessions.get(session_id)) is None:
208 raise web.HTTPNotFound(reason=f"Unknown live announcement: {session_id}")
209 resp = web.StreamResponse(status=200, reason="OK")
210 resp.content_type = "audio/wav"
211 resp.enable_chunked_encoding()
212 await resp.prepare(request)
213 # the header declares an open-ended length: how long the user will speak is
214 # not known when the first bytes go out
215 await resp.write(create_streaming_wave_header(session.audio_format))
216 # aclosing releases the reader immediately when the renderer goes away, instead
217 # of leaving it parked on the session until garbage collection finalizes it
218 audio = session.read()
219 async with aclosing(audio):
220 async for chunk in audio:
221 try:
222 await resp.write(chunk)
223 except ConnectionResetError, BrokenPipeError:
224 break
225 return resp
226
227 async def handle_ws(self, request: web.Request) -> web.WebSocketResponse:
228 """Serve one client connection: receive the spoken audio and announce it."""
229 ws = web.WebSocketResponse(heartbeat=25)
230 await ws.prepare(request)
231 self._connections.add(ws)
232 try:
233 if not await self._authenticate(request, ws):
234 return ws
235 if (start := await self._read_start_message(request, ws)) is None:
236 return ws
237 if len(self._sessions) >= MAX_CONCURRENT_SESSIONS:
238 await self._reject(request, ws, "Too many live announcements in progress")
239 return ws
240 await self._run_session(ws, start)
241 finally:
242 self._connections.discard(ws)
243 if not ws.closed:
244 await ws.close()
245 return ws
246
247 async def _run_session(self, ws: web.WebSocketResponse, start: LiveAnnouncementStart) -> None:
248 """Announce what the client speaks, from the start message up to the stop."""
249 session = self._create_session(start.audio_format)
250 try:
251 await self._send(ws, {"type": "started"})
252 await self._receive_audio(ws, session)
253 finally:
254 await session.finish()
255 if not session.duration or self._closing:
256 # nothing was spoken (a mis-tap, or a client that never sent audio): announcing
257 # it would interrupt whatever the player is doing to play silence. the same
258 # applies while shutting down, where the clip was cut short by us.
259 self._sessions.pop(session.session_id, None)
260 await self._send(ws, {"type": "finished"})
261 return
262 # only now the clip is complete: players that announce natively need it whole up
263 # front - AirPlay renders it to a file and schedules one synchronized instant for
264 # every group member from its exact duration, which a growing clip cannot give.
265 # Its own task, so a client that drops still gets what it spoke played out, and
266 # it owns the session, dropping it once the audio has been consumed.
267 announcement = self.mass.create_task(
268 self._play_live_announcement(
269 session,
270 start.player_id,
271 pre_announce=start.pre_announce,
272 volume_level=start.volume_level,
273 )
274 )
275 if ws.closed:
276 # there is no longer anyone to report the outcome to, so release the
277 # connection handler and let the announcement play itself out
278 return
279 try:
280 await announcement
281 except MusicAssistantError as err:
282 # a typed error carries a message meant for the person who spoke
283 self.logger.warning("Live announcement to player %s failed: %s", start.player_id, err)
284 await self._send(ws, {"type": "error", "message": str(err)})
285 return
286 except Exception:
287 # anything else is a defect rather than a failed announcement, so it is logged
288 # with its traceback and reported without leaking its internals
289 self.logger.exception("Live announcement to player %s failed", start.player_id)
290 await self._send(ws, {"type": "error", "message": "The announcement failed."})
291 return
292 await self._send(ws, {"type": "finished"})
293
294 async def _receive_audio(
295 self, ws: web.WebSocketResponse, session: LiveAnnouncementSession
296 ) -> None:
297 """Buffer the audio frames the client sends until it stops speaking."""
298 try:
299 async with asyncio.timeout(MAX_SESSION_SECONDS):
300 await self._read_frames(ws, session)
301 except TimeoutError:
302 self.logger.warning(
303 "Live announcement %s ended: it ran for the maximum of %s seconds",
304 session.session_id,
305 MAX_SESSION_SECONDS,
306 )
307
308 async def _read_frames(
309 self, ws: web.WebSocketResponse, session: LiveAnnouncementSession
310 ) -> None:
311 """Read audio frames until the client stops speaking or falls silent."""
312 while True:
313 try:
314 async with asyncio.timeout(IDLE_TIMEOUT):
315 msg = await ws.receive()
316 except TimeoutError:
317 self.logger.warning(
318 "Live announcement %s ended: no audio for %s seconds",
319 session.session_id,
320 IDLE_TIMEOUT,
321 )
322 return
323 if msg.type != WSMsgType.BINARY:
324 # any text frame is the client saying it is done; anything else is the
325 # connection going away, which means the same thing
326 return
327 if not msg.data:
328 # carries no audio, so it must not grow the clip that is held in memory
329 continue
330 await session.write(msg.data)
331 if session.duration >= MAX_ANNOUNCEMENT_SECONDS:
332 self.logger.warning(
333 "Live announcement %s reached the maximum length of %s seconds",
334 session.session_id,
335 MAX_ANNOUNCEMENT_SECONDS,
336 )
337 return
338
339 async def _play_live_announcement(
340 self,
341 session: LiveAnnouncementSession,
342 player_id: str,
343 pre_announce: bool | None,
344 volume_level: int | None,
345 ) -> None:
346 """Play the session audio as an announcement and drop the session afterwards."""
347 try:
348 # the player was available when the handshake accepted it, but a whole clip
349 # has been spoken since; an unavailable player drops the command downstream
350 # without a word, which would be reported back as a clip that played
351 player = self.mass.players.get_player(player_id)
352 if player is None or not player.available:
353 raise PlayerCommandFailed(f"Player {player_id} is no longer available.")
354 # a player that plays announcements natively hands off to its provider, which
355 # has no deadline of its own - one that never returns would otherwise keep
356 # this session (and its slot) alive for as long as the server runs
357 async with asyncio.timeout(ANNOUNCEMENT_TIMEOUT):
358 await self.mass.players.play_announcement(
359 player_id,
360 url=session.url,
361 pre_announce=pre_announce,
362 volume_level=volume_level,
363 )
364 except TimeoutError as err:
365 # reported to the client rather than swallowed: a timeout means the player
366 # never confirmed it played, so "finished" would be a guess
367 raise PlayerCommandFailed("The announcement did not finish playing.") from err
368 finally:
369 self._sessions.pop(session.session_id, None)
370
371 def _create_session(self, audio_format: AudioFormat) -> LiveAnnouncementSession:
372 """Register a new session and return it."""
373 session_id = secrets.token_urlsafe(16)
374 path = LIVE_ANNOUNCEMENT_STREAM_PATH.format(session_id=session_id)
375 session = LiveAnnouncementSession(
376 session_id, f"{self.mass.streams.base_url}{path}", audio_format
377 )
378 self._sessions[session_id] = session
379 return session
380
381 async def _authenticate(self, request: web.Request, ws: web.WebSocketResponse) -> bool:
382 """
383 Authenticate a client connection, mirroring the visualizer relay handshake.
384
385 Ingress requests are authenticated by Home Assistant via headers; everything
386 else must send `{"type": "auth", "token": ...}` first.
387
388 :param request: The incoming HTTP request.
389 :param ws: The prepared WebSocket response.
390 :return: True when the client may announce.
391 """
392 is_ingress = is_request_from_ingress(request)
393 if is_ingress:
394 user = await get_authenticated_user(request)
395 elif (message := await self._read_message(request, ws, "auth")) is None:
396 return False
397 elif not (token := message.get("token")):
398 return await self._reject(request, ws, "Auth message without a token")
399 else:
400 user = await self.mass.webserver.auth.authenticate_with_token(str(token))
401 if user is None:
402 return await self._reject(request, ws, "Authentication failed")
403 if not is_ingress and user.username == HOMEASSISTANT_SYSTEM_USER:
404 # the token of the Home Assistant system user is only meant to travel over
405 # the ingress connection, mirroring the policy of the websocket api
406 return await self._reject(
407 request, ws, "Home Assistant system user not allowed on regular webserver"
408 )
409 if not has_scope(user, Scope.PLAYERS_CONTROL):
410 return await self._reject(request, ws, "Not allowed to control players")
411 # the announcement is dispatched as a task, which inherits this context, so the
412 # player command it runs sees the user and applies their player restrictions
413 set_current_user(user)
414 return True
415
416 async def _read_start_message(
417 self, request: web.Request, ws: web.WebSocketResponse
418 ) -> LiveAnnouncementStart | None:
419 """Read and validate the start message, returning None when it is unusable."""
420 if (message := await self._read_message(request, ws, "start")) is None:
421 return None
422 player_id = message.get("player_id")
423 player = self.mass.players.get_player(player_id) if isinstance(player_id, str) else None
424 # an unavailable player silently drops the command downstream, which would
425 # otherwise be reported to the client as an announcement that played
426 if not isinstance(player_id, str) or player is None or not player.available:
427 await self._reject(request, ws, f"Unknown or unavailable player: {player_id}")
428 return None
429 sample_rate = message.get("sample_rate")
430 if (
431 not isinstance(sample_rate, int)
432 or not MIN_SAMPLE_RATE <= sample_rate <= MAX_SAMPLE_RATE
433 ):
434 await self._reject(request, ws, f"Unsupported sample rate: {sample_rate}")
435 return None
436 channels = message.get("channels", 1)
437 if not isinstance(channels, int) or not 1 <= channels <= MAX_CHANNELS:
438 await self._reject(request, ws, f"Unsupported channel count: {channels}")
439 return None
440 pre_announce = message.get("pre_announce")
441 volume_level = message.get("volume_level")
442 return LiveAnnouncementStart(
443 player_id=player_id,
444 audio_format=AudioFormat(
445 content_type=ContentType.PCM_S16LE,
446 sample_rate=sample_rate,
447 bit_depth=LIVE_ANNOUNCEMENT_BIT_DEPTH,
448 channels=channels,
449 ),
450 pre_announce=pre_announce if isinstance(pre_announce, bool) else None,
451 volume_level=volume_level if isinstance(volume_level, int) else None,
452 )
453
454 async def _read_message(
455 self, request: web.Request, ws: web.WebSocketResponse, expected_type: str
456 ) -> dict[str, object] | None:
457 """Read one json message of the expected type, returning None when it is not."""
458 try:
459 async with asyncio.timeout(HANDSHAKE_TIMEOUT):
460 msg = await ws.receive()
461 except TimeoutError:
462 await self._reject(request, ws, f"Timeout waiting for the {expected_type} message")
463 return None
464 if msg.type != WSMsgType.TEXT:
465 await self._reject(request, ws, f"Expected a {expected_type} message")
466 return None
467 try:
468 data = json_loads(msg.data)
469 except JSON_DECODE_EXCEPTIONS:
470 await self._reject(request, ws, f"Invalid JSON in the {expected_type} message")
471 return None
472 if not isinstance(data, dict) or data.get("type") != expected_type:
473 await self._reject(request, ws, f"Expected a {expected_type} message")
474 return None
475 return data
476
477 async def _reject(self, request: web.Request, ws: web.WebSocketResponse, reason: str) -> bool:
478 """
479 Close a connection that may not (or cannot) announce, and log why.
480
481 The reason travels in the close frame, where the client reads it to tell the
482 user what went wrong instead of showing a bare disconnect.
483 """
484 self.logger.warning("Rejected live announcement from %s: %s", request.remote, reason)
485 # a close frame carries at most 125 bytes, 2 of which are the status code, and a
486 # reason quoting client input can exceed that - which the browser drops entirely.
487 # the decode/encode round trip drops a character the cut landed in the middle of,
488 # since an invalid utf-8 reason would be rejected just the same.
489 message = reason.encode()[:MAX_CLOSE_REASON_BYTES].decode(errors="ignore").encode()
490 await ws.close(code=4001, message=message)
491 return False
492
493 async def _send(self, ws: web.WebSocketResponse, message: dict[str, object]) -> None:
494 """Send one json message, ignoring a client that already went away."""
495 with suppress(ConnectionError, RuntimeError):
496 await ws.send_str(dumps(message).decode())
497