/
/
/
1"""Just-in-time clip rendering for AI Radio."""
2# mypy: disable-error-code="attr-defined"
3
4from __future__ import annotations
5
6import asyncio
7import json
8import logging
9from dataclasses import dataclass, replace
10from typing import TYPE_CHECKING, Any, cast
11
12from music_assistant_models.enums import ContentType, StreamType, VolumeNormalizationMode
13from music_assistant_models.errors import (
14 InvalidDataError,
15 MediaNotFoundError,
16 MusicAssistantError,
17)
18from music_assistant_models.media_items import AudioFormat
19from music_assistant_models.streamdetails import StreamDetails
20
21from music_assistant.constants import (
22 CONF_VALUE_DISABLED,
23 CONF_VALUE_ENABLED,
24 CONF_VOLUME_NORMALIZATION,
25 CONF_VOLUME_NORMALIZATION_TARGET,
26 CONF_VOLUME_NORMALIZATION_TRACKS,
27)
28from music_assistant.helpers.audio import parse_loudnorm
29from music_assistant.helpers.ffmpeg import get_ffmpeg_stream
30from music_assistant.helpers.process import check_output
31from music_assistant.helpers.tags import async_parse_tags
32from music_assistant.helpers.tts import (
33 query_tts_engine_with_language_fallback,
34 resolve_tts_language,
35 resolve_tts_stream_path,
36)
37
38from .constants import (
39 ATTR_HOST_ID,
40 ATTR_MAX_CHARS,
41 ATTR_PROMPT,
42 ATTR_RENDERED_TEXT,
43 ATTR_SESSION_ID,
44 ATTR_WEATHER_REQUIRED,
45 ATTR_WEB_SEARCH_MODE,
46 CLIP_STREAMDETAILS_EXPIRATION,
47 CONF_TTS_LOUDNESS_BOOST,
48 DEFAULT_TTS_LOUDNESS_BOOST,
49 DEFERRED_PLACEHOLDERS,
50 LOUDNESS_MEASURE_TIMEOUT,
51 MIN_CLIP_MEDIA_LIFETIME,
52 MIN_LOUDNESS_REFERENCE_SECONDS,
53 NO_WEATHER_DATA_INSTRUCTION,
54 TTS_CLIP_PCM_FORMAT,
55 TTS_PEAK_CEILING_DB,
56 TTS_SERVER_ERROR_MARKERS,
57 TTS_SPEECHNORM_FILTER,
58 WEATHER_PLACEHOLDER_TOKENS,
59)
60from .helpers import coerce_int, format_ai_radio_timestamp, soft_limit_text
61
62if TYPE_CHECKING:
63 from collections.abc import AsyncGenerator
64
65 from music_assistant_models.config_entries import ProviderConfig
66 from music_assistant_models.enums import MediaType
67 from music_assistant_models.queue_item import QueueItem
68
69 from music_assistant.mass import MusicAssistant
70
71 from .models import SessionState
72
73
74@dataclass(slots=True)
75class _CachedClipMedia:
76 """Media previously minted for a clip, kept until it expires."""
77
78 path: str
79 stream_type: StreamType
80 audio_format: AudioFormat
81 duration: int | None
82 minted_at: float
83 loudness: float | None
84
85
86@dataclass(slots=True)
87class _ClipAudio:
88 """What get_audio_stream needs to serve a levelled clip, carried on StreamDetails.data."""
89
90 path: str
91 input_format: AudioFormat
92 gain_db: float
93
94
95class AIRadioRenderMixin:
96 """Renders an AI Radio clip at the moment MA needs its audio."""
97
98 if TYPE_CHECKING:
99 mass: MusicAssistant
100 config: ProviderConfig
101 logger: logging.Logger
102 _hosts: dict[str, dict[str, Any]]
103 _sessions: dict[str, SessionState]
104
105 _render_locks: dict[str, asyncio.Lock]
106 _media_cache: dict[str, _CachedClipMedia]
107 _engine_loudness: dict[tuple[str, str, str], float]
108
109 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
110 """
111 Render the AI Radio clip with the given id and return its StreamDetails.
112
113 :param item_id: The clip id of the queue item MA wants to play.
114 :param media_type: The media type of the requested item.
115 """
116 queue_item = self._find_clip_item(item_id)
117 if queue_item is None:
118 raise MediaNotFoundError(f"AI Radio clip {item_id} is not in any queue")
119 prompt = str(queue_item.extra_attributes.get(ATTR_PROMPT) or "")
120 if not prompt:
121 self._record_skip(queue_item, "clip has no prompt to render")
122 raise MediaNotFoundError(f"AI Radio clip {item_id} has no prompt to render")
123
124 async with self._lock_for(item_id):
125 text = str(queue_item.extra_attributes.get(ATTR_RENDERED_TEXT) or "")
126 if not text:
127 text = await self._generate_script(queue_item, prompt, item_id)
128 queue_item.extra_attributes[ATTR_RENDERED_TEXT] = text
129 # the signal is what marks the items cache dirty and schedules the persist
130 self.mass.player_queues.signal_update(queue_item.queue_id, items_changed=True)
131 media = await self._cached_clip_media(queue_item, text, item_id)
132
133 streamdetails = StreamDetails(
134 provider=self.instance_id,
135 item_id=item_id,
136 audio_format=media.audio_format,
137 media_type=media_type,
138 stream_type=media.stream_type,
139 path=media.path,
140 duration=media.duration,
141 # a talk clip has nothing worth seeking to, and a seek is the one path that
142 # would re-fetch a possibly-expired HA url mid-playback
143 can_seek=False,
144 allow_seek=False,
145 # a cache hit serves a url that was minted earlier, so it may only claim the life
146 # that url has left or the stream outlives the token behind it
147 expiration=self._remaining_media_lifetime(media),
148 )
149 gain_db = self._loudness_gain(queue_item.queue_id, media.loudness)
150 if gain_db is not None:
151 # core never normalizes a sound effect, so the clip is levelled here or it
152 # airs noticeably quieter than the music around it
153 streamdetails.stream_type = StreamType.CUSTOM
154 # core mirrors what ffmpeg reports onto this object, so it gets a copy of the
155 # constant rather than a handle on the one every clip shares
156 streamdetails.decoded_audio_format = replace(TTS_CLIP_PCM_FORMAT)
157 streamdetails.data = _ClipAudio(media.path, media.audio_format, gain_db)
158 return streamdetails
159
160 async def get_audio_stream(
161 self, streamdetails: StreamDetails, seek_position: int = 0
162 ) -> AsyncGenerator[bytes]:
163 """
164 Return the levelled audio of a spoken clip as PCM.
165
166 :param streamdetails: The StreamDetails previously returned by get_stream_details.
167 :param seek_position: Ignored, a spoken clip cannot be seeked.
168 """
169 clip = cast("_ClipAudio", streamdetails.data)
170 async for chunk in get_ffmpeg_stream(
171 audio_input=clip.path,
172 input_format=clip.input_format,
173 output_format=TTS_CLIP_PCM_FORMAT,
174 filter_params=[
175 TTS_SPEECHNORM_FILTER,
176 f"volume={round(clip.gain_db, 2)}dB",
177 f"alimiter=limit={TTS_PEAK_CEILING_DB}dB:level=false:latency=true",
178 ],
179 ):
180 yield chunk
181
182 def _lock_for(self, clip_id: str) -> asyncio.Lock:
183 """Return the per-clip render lock, creating it on first use."""
184 if not hasattr(self, "_render_locks"):
185 self._render_locks = {}
186 if clip_id not in self._render_locks:
187 self._render_locks[clip_id] = asyncio.Lock()
188 return self._render_locks[clip_id]
189
190 async def _cached_clip_media(
191 self, queue_item: QueueItem, text: str, clip_id: str
192 ) -> _CachedClipMedia:
193 """Return the clip's minted media, re-minting only once the cache entry has expired."""
194 if not hasattr(self, "_media_cache"):
195 self._media_cache = {}
196 now = asyncio.get_running_loop().time()
197 cached = self._media_cache.get(clip_id)
198 if cached is not None and self._remaining_media_lifetime(cached) > MIN_CLIP_MEDIA_LIFETIME:
199 return cached
200 # the caller holds the per-clip render lock, so of the several uncoordinated paths
201 # that resolve the same clip only the first one mints; the rest hit the cache above
202 path, stream_type, audio_format, duration, loudness = await self._mint_clip_media(
203 queue_item, text, clip_id
204 )
205 media = _CachedClipMedia(path, stream_type, audio_format, duration, now, loudness)
206 # clips are minted per queue item, so without pruning the cache grows for as long as
207 # the server runs. an entry past its window can never be served again anyway
208 for expired_id in [
209 key
210 for key, entry in self._media_cache.items()
211 if now - entry.minted_at >= CLIP_STREAMDETAILS_EXPIRATION
212 ]:
213 del self._media_cache[expired_id]
214 self._media_cache[clip_id] = media
215 return media
216
217 def _remaining_media_lifetime(self, media: _CachedClipMedia) -> int:
218 """Return the seconds the given minted media is still usable for."""
219 elapsed = asyncio.get_running_loop().time() - media.minted_at
220 return max(MIN_CLIP_MEDIA_LIFETIME, round(CLIP_STREAMDETAILS_EXPIRATION - elapsed))
221
222 def _wanted_loudness(self, queue_id: str) -> float | None:
223 """Return the level in LUFS a clip should air at, or None when it should air as is."""
224 normalization = self.mass.config.get_effective_player_queue_config_value(
225 queue_id, CONF_VOLUME_NORMALIZATION, CONF_VALUE_ENABLED
226 )
227 if normalization == CONF_VALUE_DISABLED:
228 return None
229 # the queue switch only says normalization may run; the tracks around the clip are
230 # the ones it has to match, and their own preference can still turn it off
231 tracks_mode = self.mass.streams.get_config_value(CONF_VOLUME_NORMALIZATION_TRACKS)
232 if tracks_mode == VolumeNormalizationMode.DISABLED.value:
233 return None
234 target = self.mass.streams.get_config_value(
235 CONF_VOLUME_NORMALIZATION_TARGET, return_type=int
236 )
237 boost = coerce_int(
238 self.config.get_value(CONF_TTS_LOUDNESS_BOOST), DEFAULT_TTS_LOUDNESS_BOOST
239 )
240 return target + boost
241
242 def _loudness_gain(self, queue_id: str, loudness: float | None) -> float | None:
243 """Return the dB to lift the clip by, or None when it should air untouched."""
244 if loudness is None or (wanted := self._wanted_loudness(queue_id)) is None:
245 return None
246 # the reference is taken behind speechnorm, which lands close to the target on its
247 # own, so this trim is small and runs in either direction
248 return wanted - loudness
249
250 def _tts_language(self, host_language: str | None = None) -> str | None:
251 """
252 Return the host's language, or the server locale, as a hyphenated language code.
253
254 :param host_language: The host's configured language override, if any.
255 """
256 if override := (host_language or "").strip():
257 return override.replace("_", "-")
258 return resolve_tts_language(self.mass)
259
260 def _find_clip_item(self, clip_id: str) -> QueueItem | None:
261 """Return the queue item holding the given clip, or None when no queue holds it."""
262 for queue_id in self._candidate_queue_ids(clip_id):
263 if (item := self._find_clip_in_queue(clip_id, queue_id)) is not None:
264 return item
265 return None
266
267 def _candidate_queue_ids(self, clip_id: str) -> list[str]:
268 """
269 Return the queue ids to search for a clip, the most likely one first.
270
271 The owning session knows its queue, but the session registry is empty after a
272 restart while the clip lives on in the persisted queue, so every queue stays a
273 candidate. Clip ids carry a uuid4-based session id, so a hit is unambiguous.
274 """
275 queue_ids = [queue.queue_id for queue in self.mass.player_queues.all()]
276 session = self._sessions.get(clip_id.rpartition("_")[0])
277 if session is not None and session.queue_id in queue_ids:
278 queue_ids.remove(session.queue_id)
279 queue_ids.insert(0, session.queue_id)
280 return queue_ids
281
282 def _find_clip_in_queue(self, clip_id: str, queue_id: str) -> QueueItem | None:
283 """Return the queue item holding the given clip, paging through the queue."""
284 page_size = 500
285 offset = 0
286 while True:
287 page = self.mass.player_queues.items(queue_id, limit=page_size, offset=offset)
288 if not page:
289 return None
290 for item in page:
291 if item.media_item is not None and item.media_item.item_id == clip_id:
292 return item
293 if len(page) < page_size:
294 return None
295 offset += page_size
296
297 async def _generate_script(self, queue_item: QueueItem, prompt: str, clip_id: str) -> str:
298 """Resolve the deferred placeholders and generate the spoken script."""
299 attributes = queue_item.extra_attributes
300 deferred = await self._resolve_deferred_placeholders(prompt)
301 empty_weather_tokens = [
302 token
303 for token in WEATHER_PLACEHOLDER_TOKENS
304 if token in prompt and not deferred.get(token)
305 ]
306 if empty_weather_tokens:
307 if attributes.get(ATTR_WEATHER_REQUIRED):
308 error = "weather data unavailable for a weather-required clip"
309 self.logger.warning(
310 "AI Radio clip %s (%s) skipped: %s", clip_id, queue_item.name, error
311 )
312 self._record_skip(queue_item, error)
313 raise MediaNotFoundError(f"AI Radio clip {clip_id} has no weather data")
314 # weather is optional in this clip, so the LLM must skip it rather than invent it
315 for token in empty_weather_tokens:
316 deferred[token] = NO_WEATHER_DATA_INSTRUCTION
317 resolved = prompt
318 for key, value in deferred.items():
319 resolved = resolved.replace(key, value)
320 host = self._hosts.get(str(attributes.get(ATTR_HOST_ID) or "")) or {}
321 instructions = str(host.get("instructions") or "")
322 language = str(host.get("language") or "")
323 max_chars = int(attributes.get(ATTR_MAX_CHARS) or 0)
324 web_mode = str(attributes.get(ATTR_WEB_SEARCH_MODE) or "disabled")
325 try:
326 text = cast(
327 "str",
328 await self._generate_text(
329 instructions=instructions,
330 prompt=resolved,
331 web_mode=web_mode,
332 language=language,
333 ),
334 )
335 except Exception as err:
336 self.logger.warning(
337 "AI Radio clip %s (%s) failed to generate: %s", clip_id, queue_item.name, err
338 )
339 self._record_skip(queue_item, f"generation failed: {err}")
340 raise MediaNotFoundError(f"AI Radio clip {clip_id} failed to generate") from err
341 if max_chars > 0:
342 text = soft_limit_text(text, max_chars=max_chars)
343 self.logger.debug(
344 "AI Radio clip %s (%s) rendered: %d chars", clip_id, queue_item.name, len(text)
345 )
346 return text
347
348 async def _resolve_deferred_placeholders(self, prompt: str) -> dict[str, str]:
349 """Return freshly resolved values for the placeholders deferred until airtime."""
350 values = dict.fromkeys(DEFERRED_PLACEHOLDERS, "")
351 values["<timestamp>"] = format_ai_radio_timestamp(self._configured_now())
352 # weather is the only deferred placeholder that costs a network round-trip, so it is
353 # only fetched when the prompt actually references it
354 if any(token in prompt for token in WEATHER_PLACEHOLDER_TOKENS):
355 values.update(await self._prepare_weather_tokens())
356 return values
357
358 async def _mint_clip_media(
359 self, queue_item: QueueItem, text: str, clip_id: str
360 ) -> tuple[str, StreamType, AudioFormat, int | None, float | None]:
361 """Convert the script to playable audio via the configured TTS engine."""
362 host = self._hosts.get(str(queue_item.extra_attributes.get(ATTR_HOST_ID) or "")) or {}
363 engine_uid = str(host.get("tts_engine") or "") or None
364 language = self._tts_language(str(host.get("language") or ""))
365 options = host.get("options") or {}
366 try:
367 path, stream_type, audio_format = await self._render_tts_media(
368 text, engine_uid, language, options
369 )
370 # the probe is the first fetch, so a failed render surfaces here and not in playback
371 duration = await self._probe_duration(path)
372 except Exception as err:
373 self.logger.warning("AI Radio clip %s failed TTS: %s", clip_id, err)
374 self._record_skip(queue_item, f"TTS failed: {err}")
375 raise MediaNotFoundError(f"AI Radio clip {clip_id} failed TTS") from err
376 # measuring costs a fetch and a decode on the just-in-time render path, so it only
377 # runs where the reading has somewhere to go
378 loudness = (
379 await self._reference_loudness(engine_uid, language, options, path, duration)
380 if self._wanted_loudness(queue_item.queue_id) is not None
381 else None
382 )
383 return path, stream_type, audio_format, duration, loudness
384
385 async def _reference_loudness(
386 self,
387 engine_uid: str | None,
388 language: str | None,
389 options: dict[str, Any],
390 path: str,
391 duration: int | None,
392 ) -> float | None:
393 """Return the loudness in LUFS to level this clip against, or None when unknown."""
394 if not hasattr(self, "_engine_loudness"):
395 self._engine_loudness = {}
396 # engine, language and options together decide which voice speaks, and clips from one
397 # voice land within a dB of each other, so measuring one of them is enough
398 key = (engine_uid or "", language or "", json.dumps(options, sort_keys=True, default=str))
399 if (cached := self._engine_loudness.get(key)) is not None:
400 return cached
401 if (loudness := await self._measure_loudness(path)) is None:
402 return None
403 if (duration or 0) >= MIN_LOUDNESS_REFERENCE_SECONDS:
404 self._engine_loudness[key] = loudness
405 return loudness
406
407 async def _measure_loudness(self, path: str) -> float | None:
408 """Return the integrated loudness of the given audio in LUFS, or None when it fails."""
409 try:
410 returncode, output = await check_output(
411 "ffmpeg",
412 "-hide_banner",
413 "-nostats",
414 "-i",
415 path,
416 # measure behind speechnorm: it is what the gain is applied on top of, and it
417 # levels the clip itself, so the reading has to come from its output or the
418 # gain corrects for a level that no longer reaches it
419 "-af",
420 f"{TTS_SPEECHNORM_FILTER},loudnorm=print_format=json",
421 "-f",
422 "null",
423 "-",
424 timeout=LOUDNESS_MEASURE_TIMEOUT,
425 )
426 except (OSError, TimeoutError) as err:
427 self.logger.debug("Could not measure AI Radio clip loudness: %s", err)
428 return None
429 if returncode != 0:
430 self.logger.debug("Could not measure AI Radio clip loudness: ffmpeg failed")
431 return None
432 return parse_loudnorm(output)
433
434 async def _render_tts_media(
435 self,
436 text: str,
437 engine_uid: str | None = None,
438 language: str | None = None,
439 options: dict[str, Any] | None = None,
440 ) -> tuple[str, StreamType, AudioFormat]:
441 """Ask the TTS engine for audio and return the path, stream type and format to play it."""
442 engine = await self._get_tts_engine(engine_uid)
443 stream_details = await query_tts_engine_with_language_fallback(
444 engine, text, language, logger=self.logger, options=options
445 )
446 path, stream_type = await resolve_tts_stream_path(engine, stream_details)
447 audio_format = stream_details.audio_format
448 if audio_format.content_type == ContentType.UNKNOWN:
449 audio_format = AudioFormat(content_type=ContentType.MP3)
450 return path, stream_type, audio_format
451
452 async def _probe_duration(self, path: str) -> int | None:
453 """Return the clip duration in seconds, or None when it cannot be determined."""
454 try:
455 tags = await async_parse_tags(path, require_duration=True)
456 except (InvalidDataError, OSError) as err:
457 if any(marker in str(err) for marker in TTS_SERVER_ERROR_MARKERS):
458 # the engine reports no reason of its own (Home Assistant answers a failed
459 # render with an empty 500), so the probe's message is the only clue there is
460 raise MusicAssistantError(
461 f"{err}. The TTS engine failed to generate the audio it handed out. "
462 "Check the logs of the TTS engine for the reason (for a Home Assistant "
463 "engine that is the Home Assistant core log). A cloud engine may be "
464 "out of credit or having an outage."
465 ) from err
466 self.logger.warning("Could not determine AI Radio clip duration: %s", err)
467 return None
468 return int(tags.duration) if tags.duration else None
469
470 def _record_skip(self, queue_item: QueueItem, error: str) -> None:
471 """Record a skipped clip on its owning session."""
472 session_id = str(queue_item.extra_attributes.get(ATTR_SESSION_ID) or "")
473 if (session := self._sessions.get(session_id)) is None:
474 return
475 session.skipped_sections += 1
476 session.last_render_error = error
477