/
/
/
1"""
2Rendering of announcement audio (optional pre-announce chime + announcement).
3
4A single announcement is regularly consumed more than once: by the player's own http
5fetch, by a HEAD probe preceding it, and by every member of a group announcement. An
6AnnouncementRender decodes the audio once and keeps the whole clip in memory, so every
7one of those readers is served from there, each in the format it needs. Because the
8clip is complete, its exact duration is known without probing the source again.
9"""
10
11from __future__ import annotations
12
13import asyncio
14import logging
15from collections.abc import AsyncGenerator
16from contextlib import aclosing, suppress
17from typing import TYPE_CHECKING
18
19from music_assistant_models.enums import ContentType
20from music_assistant_models.errors import AudioError
21from music_assistant_models.media_items import AudioFormat
22
23from music_assistant.constants import MASS_LOGGER_NAME, VERBOSE_LOG_LEVEL
24from music_assistant.helpers.audio import calculate_content_length
25from music_assistant.helpers.ffmpeg import get_ffmpeg_stream
26
27if TYPE_CHECKING:
28 from music_assistant.controllers.players.helpers import AnnounceData
29
30LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.announcements")
31
32# Announcements are always rendered to this format; readers that need something else
33# (an encoded http stream, a different sample rate) convert from it per reader.
34ANNOUNCEMENT_PCM_FORMAT = AudioFormat(
35 content_type=ContentType.PCM_S16LE,
36 sample_rate=44100,
37 bit_depth=16,
38 channels=2,
39)
40
41# Upper bounds on the audio taken from each source. Announcements are short by nature;
42# these only guard against a source that never ends (e.g. a radio stream handed to the
43# announcement api, or an endless custom chime), which would otherwise render forever.
44MAX_CHIME_SECONDS = 30
45MAX_ANNOUNCEMENT_SECONDS = 300
46
47# The longest clip a render can produce, for callers that need to bound a wait on an
48# announcement whose length is not known.
49MAX_CLIP_SECONDS = MAX_CHIME_SECONDS + MAX_ANNOUNCEMENT_SECONDS
50
51# Maximum time to wait for a (slow) announcement source.
52DEFAULT_RENDER_TIMEOUT = 30
53
54
55class AnnouncementRender:
56 """
57 A single announcement clip (pre-announce chime + announcement), held in memory.
58
59 Announcement clips are short and are read by several consumers at once, each from
60 the start, so the clip is kept whole until the render is closed.
61 """
62
63 def __init__(self, announcement_url: str, pre_announce: bool, pre_announce_url: str) -> None:
64 """
65 Initialize the render. Call start() to begin rendering.
66
67 :param announcement_url: URL of the announcement audio.
68 :param pre_announce: Whether to prepend the pre-announce chime.
69 :param pre_announce_url: URL of the pre-announce chime.
70 """
71 self.announcement_url = announcement_url
72 self.pre_announce = pre_announce
73 self.pre_announce_url = pre_announce_url
74 self.ref_count = 0
75 self._chunks: list[bytes] = []
76 self._total_bytes = 0
77 self._closed = False
78 self._task: asyncio.Task[None] | None = None
79 self._audio_added = asyncio.Condition()
80 self._ready = asyncio.Event()
81 self._finished = asyncio.Event()
82
83 @property
84 def duration(self) -> float:
85 """Return the duration (in seconds) of the audio rendered so far."""
86 return self._total_bytes / ANNOUNCEMENT_PCM_FORMAT.pcm_sample_size
87
88 @property
89 def key(self) -> str:
90 """Return the key that identifies the audio of this render."""
91 return _render_key(self.announcement_url, self.pre_announce, self.pre_announce_url)
92
93 def start(self) -> None:
94 """Start rendering the announcement audio."""
95 self._task = asyncio.get_running_loop().create_task(self._render())
96
97 async def get_stream(self, output_format: AudioFormat) -> AsyncGenerator[bytes]:
98 """
99 Stream the announcement audio in the given output format.
100
101 Starts at the beginning of the clip and waits for audio that is still being
102 rendered, so any number of consumers can read it at the same time.
103
104 :param output_format: The format to deliver the audio in.
105 """
106 if output_format == ANNOUNCEMENT_PCM_FORMAT:
107 async for chunk in self._read():
108 yield chunk
109 return
110 async for chunk in get_ffmpeg_stream(
111 audio_input=self._read(),
112 input_format=ANNOUNCEMENT_PCM_FORMAT,
113 output_format=output_format,
114 ):
115 yield chunk
116
117 async def wait_ready(self, timeout: float = DEFAULT_RENDER_TIMEOUT) -> bool:
118 """
119 Wait until there is audio available to start playback with.
120
121 Returns False if the source did not deliver any audio within the timeout.
122
123 :param timeout: Maximum time to wait for the first audio.
124 """
125 try:
126 await asyncio.wait_for(self._ready.wait(), timeout)
127 except TimeoutError:
128 LOGGER.warning("Timeout waiting for announcement audio from %s", self.announcement_url)
129 return False
130 return self._total_bytes > 0
131
132 async def wait_finished(self, timeout: float = DEFAULT_RENDER_TIMEOUT) -> float | None:
133 """
134 Wait for the render to finish and return the exact duration in seconds.
135
136 Returns None if the render did not finish within the timeout.
137
138 :param timeout: Maximum time to wait for the render to finish.
139 """
140 try:
141 await asyncio.wait_for(self._finished.wait(), timeout)
142 except TimeoutError:
143 LOGGER.warning(
144 "Timeout waiting for announcement %s to be rendered", self.announcement_url
145 )
146 return None
147 return self.duration
148
149 async def close(self) -> None:
150 """Discard the rendered audio and stop reading from the source."""
151 if self._task and not self._task.done():
152 self._task.cancel()
153 with suppress(asyncio.CancelledError):
154 await self._task
155 async with self._audio_added:
156 self._closed = True
157 self._chunks.clear()
158 self._ready.set()
159 self._finished.set()
160 self._audio_added.notify_all()
161
162 async def _render(self) -> None:
163 """Decode the chime and the announcement into the clip."""
164 try:
165 if self.pre_announce:
166 await self._decode(self.pre_announce_url, MAX_CHIME_SECONDS)
167 # only the announcement itself marks the clip ready to play: a player
168 # started on the chime alone would run dry waiting for a slow source
169 await self._decode(self.announcement_url, MAX_ANNOUNCEMENT_SECONDS, signal_ready=True)
170 except AudioError as err:
171 # a failed source ends the clip instead of propagating to every reader:
172 # they receive whatever was rendered so far (possibly nothing) and stop
173 LOGGER.warning(
174 "Failed to fetch announcement audio from %s: %s", self.announcement_url, err
175 )
176 except Exception:
177 # same reasoning, for anything the ffmpeg helper did not wrap in an AudioError
178 LOGGER.exception("Error rendering announcement audio from %s", self.announcement_url)
179 finally:
180 async with self._audio_added:
181 self._ready.set()
182 self._finished.set()
183 self._audio_added.notify_all()
184 LOGGER.log(
185 VERBOSE_LOG_LEVEL,
186 "Rendered %.2f seconds of announcement audio for %s",
187 self.duration,
188 self.announcement_url,
189 )
190
191 async def _decode(self, url: str, max_seconds: int, signal_ready: bool = False) -> None:
192 """Decode one source url and add it to the clip."""
193 fmt = url.rsplit(".", maxsplit=1)[-1]
194 stream = get_ffmpeg_stream(
195 audio_input=url,
196 input_format=AudioFormat(content_type=ContentType.try_parse(fmt)),
197 output_format=ANNOUNCEMENT_PCM_FORMAT,
198 chunk_size=calculate_content_length(ANNOUNCEMENT_PCM_FORMAT, 1),
199 extra_input_args=["-t", str(max_seconds)],
200 )
201 # aclosing guarantees the ffmpeg process is torn down immediately when the
202 # render is cancelled, instead of lingering until garbage collection
203 # finalizes the abandoned generator.
204 async with aclosing(stream):
205 async for chunk in stream:
206 async with self._audio_added:
207 self._chunks.append(chunk)
208 self._total_bytes += len(chunk)
209 if signal_ready:
210 self._ready.set()
211 self._audio_added.notify_all()
212
213 async def _read(self) -> AsyncGenerator[bytes]:
214 """Yield the clip from the start, waiting for audio that is still rendering."""
215 index = 0
216 while True:
217 async with self._audio_added:
218 while index >= len(self._chunks):
219 if self._closed or self._finished.is_set():
220 return
221 await self._audio_added.wait()
222 chunk = self._chunks[index]
223 index += 1
224 # yielded outside the lock, so a slow reader never holds up the render
225 yield chunk
226
227
228class AnnouncementRenderer:
229 """
230 Owner of the announcements that are currently in progress.
231
232 Tracks both which announcement each player is playing - the http route only knows
233 the player it serves - and the renders that produce the audio for them. Players
234 that announce the same audio share a single render.
235 """
236
237 def __init__(self) -> None:
238 """Initialize the renderer."""
239 self._renders: dict[str, AnnouncementRender] = {}
240 self._by_player: dict[str, AnnounceData] = {}
241
242 @property
243 def active_announcements(self) -> int:
244 """Return the number of announcements currently in progress."""
245 return len(self._by_player)
246
247 @property
248 def active_renders(self) -> int:
249 """Return the number of announcement renders currently in use."""
250 return len(self._renders)
251
252 def register(self, player_id: str, announce_data: AnnounceData) -> AnnouncementRender:
253 """
254 Register an announcement for a player and start rendering its audio.
255
256 Every register must be paired with an unregister.
257
258 :param player_id: The player the announcement is played on.
259 :param announce_data: The announcement to play.
260 """
261 self._by_player[player_id] = announce_data
262 return self.acquire(announce_data)
263
264 async def unregister(self, player_id: str, render: AnnouncementRender) -> None:
265 """
266 Release an announcement that was registered for a player.
267
268 :param player_id: The player the announcement was registered for.
269 :param render: The render previously returned by register().
270 """
271 self._by_player.pop(player_id, None)
272 await self.release(render)
273
274 def get_for_player(self, player_id: str) -> AnnounceData | None:
275 """
276 Return the announcement registered for the given player, if any.
277
278 :param player_id: The player to look up.
279 """
280 return self._by_player.get(player_id)
281
282 def acquire(self, announce_data: AnnounceData) -> AnnouncementRender:
283 """
284 Get the render for the given announcement, starting it if it is not running yet.
285
286 Every acquire must be paired with a release.
287
288 :param announce_data: The announcement to render.
289 """
290 key = _announce_data_key(announce_data)
291 if (render := self._renders.get(key)) is None:
292 render = AnnouncementRender(
293 announce_data["announcement_url"],
294 announce_data["pre_announce"],
295 announce_data["pre_announce_url"],
296 )
297 self._renders[key] = render
298 render.start()
299 render.ref_count += 1
300 return render
301
302 def get(self, announce_data: AnnounceData) -> AnnouncementRender | None:
303 """
304 Return the active render for the given announcement, if there is one.
305
306 :param announce_data: The announcement to look up.
307 """
308 return self._renders.get(_announce_data_key(announce_data))
309
310 async def release(self, render: AnnouncementRender) -> None:
311 """
312 Release a reference to a render, tearing it down when it was the last one.
313
314 :param render: The render previously obtained from acquire().
315 """
316 render.ref_count -= 1
317 if render.ref_count > 0:
318 return
319 if self._renders.get(render.key) is render:
320 del self._renders[render.key]
321 await render.close()
322
323
324def _render_key(announcement_url: str, pre_announce: bool, pre_announce_url: str) -> str:
325 """Return the key that identifies the audio of an announcement."""
326 # The key covers the audio only - not which player fetches it - so all players
327 # (and probes) that want this exact audio are served from a single render.
328 return f"{announcement_url}|{pre_announce_url if pre_announce else ''}"
329
330
331def _announce_data_key(announce_data: AnnounceData) -> str:
332 """Return the render key for the given announcement."""
333 return _render_key(
334 announce_data["announcement_url"],
335 announce_data["pre_announce"],
336 announce_data["pre_announce_url"],
337 )
338