music-assistant-server

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 logging
8from dataclasses import dataclass
9from typing import TYPE_CHECKING, Any, cast
10
11from music_assistant_models.enums import ContentType, StreamType
12from music_assistant_models.errors import (
13    InvalidDataError,
14    MediaNotFoundError,
15    MusicAssistantError,
16)
17from music_assistant_models.media_items import AudioFormat
18from music_assistant_models.streamdetails import StreamDetails
19
20from music_assistant.helpers.tags import async_parse_tags
21from music_assistant.helpers.tts import query_tts_engine, resolve_tts_stream_path
22
23from .constants import (
24    ATTR_HOST_ID,
25    ATTR_MAX_CHARS,
26    ATTR_PROMPT,
27    ATTR_RENDERED_TEXT,
28    ATTR_SESSION_ID,
29    ATTR_WEB_SEARCH_MODE,
30    CLIP_STREAMDETAILS_EXPIRATION,
31    DEFERRED_PLACEHOLDERS,
32    MIN_CLIP_MEDIA_LIFETIME,
33    TTS_SERVER_ERROR_MARKERS,
34)
35from .helpers import soft_limit_text
36
37if TYPE_CHECKING:
38    from music_assistant_models.enums import MediaType
39    from music_assistant_models.queue_item import QueueItem
40
41    from music_assistant.mass import MusicAssistant
42
43    from .models import SessionState
44
45
46@dataclass(slots=True)
47class _CachedClipMedia:
48    """Media previously minted for a clip, kept until it expires."""
49
50    path: str
51    stream_type: StreamType
52    audio_format: AudioFormat
53    duration: int | None
54    minted_at: float
55
56
57class AIRadioRenderMixin:
58    """Renders an AI Radio clip at the moment MA needs its audio."""
59
60    if TYPE_CHECKING:
61        mass: MusicAssistant
62        logger: logging.Logger
63        _hosts: dict[str, dict[str, Any]]
64        _sessions: dict[str, SessionState]
65
66    _render_locks: dict[str, asyncio.Lock]
67    _media_cache: dict[str, _CachedClipMedia]
68
69    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
70        """
71        Render the AI Radio clip with the given id and return its StreamDetails.
72
73        :param item_id: The clip id of the queue item MA wants to play.
74        :param media_type: The media type of the requested item.
75        """
76        queue_item = self._find_clip_item(item_id)
77        if queue_item is None:
78            raise MediaNotFoundError(f"AI Radio clip {item_id} is not in any queue")
79        prompt = str(queue_item.extra_attributes.get(ATTR_PROMPT) or "")
80        if not prompt:
81            self._record_skip(queue_item, "clip has no prompt to render")
82            raise MediaNotFoundError(f"AI Radio clip {item_id} has no prompt to render")
83
84        async with self._lock_for(item_id):
85            text = str(queue_item.extra_attributes.get(ATTR_RENDERED_TEXT) or "")
86            if not text:
87                text = await self._generate_script(queue_item, prompt, item_id)
88                queue_item.extra_attributes[ATTR_RENDERED_TEXT] = text
89                # the signal is what marks the items cache dirty and schedules the persist
90                self.mass.player_queues.signal_update(queue_item.queue_id, items_changed=True)
91            media = await self._cached_clip_media(queue_item, text, item_id)
92
93        return StreamDetails(
94            provider=self.instance_id,
95            item_id=item_id,
96            audio_format=media.audio_format,
97            media_type=media_type,
98            stream_type=media.stream_type,
99            path=media.path,
100            duration=media.duration,
101            # a talk clip has nothing worth seeking to, and a seek is the one path that
102            # would re-fetch a possibly-expired HA url mid-playback
103            can_seek=False,
104            allow_seek=False,
105            # a cache hit serves a url that was minted earlier, so it may only claim the life
106            # that url has left or the stream outlives the token behind it
107            expiration=self._remaining_media_lifetime(media),
108        )
109
110    def _lock_for(self, clip_id: str) -> asyncio.Lock:
111        """Return the per-clip render lock, creating it on first use."""
112        if not hasattr(self, "_render_locks"):
113            self._render_locks = {}
114        if clip_id not in self._render_locks:
115            self._render_locks[clip_id] = asyncio.Lock()
116        return self._render_locks[clip_id]
117
118    async def _cached_clip_media(
119        self, queue_item: QueueItem, text: str, clip_id: str
120    ) -> _CachedClipMedia:
121        """Return the clip's minted media, re-minting only once the cache entry has expired."""
122        if not hasattr(self, "_media_cache"):
123            self._media_cache = {}
124        now = asyncio.get_running_loop().time()
125        cached = self._media_cache.get(clip_id)
126        if cached is not None and self._remaining_media_lifetime(cached) > MIN_CLIP_MEDIA_LIFETIME:
127            return cached
128        # the caller holds the per-clip render lock, so of the several uncoordinated paths
129        # that resolve the same clip only the first one mints; the rest hit the cache above
130        path, stream_type, audio_format, duration = await self._mint_clip_media(
131            queue_item, text, clip_id
132        )
133        media = _CachedClipMedia(path, stream_type, audio_format, duration, now)
134        # clips are minted per queue item, so without pruning the cache grows for as long as
135        # the server runs. an entry past its window can never be served again anyway
136        for expired_id in [
137            key
138            for key, entry in self._media_cache.items()
139            if now - entry.minted_at >= CLIP_STREAMDETAILS_EXPIRATION
140        ]:
141            del self._media_cache[expired_id]
142        self._media_cache[clip_id] = media
143        return media
144
145    def _remaining_media_lifetime(self, media: _CachedClipMedia) -> int:
146        """Return the seconds the given minted media is still usable for."""
147        elapsed = asyncio.get_running_loop().time() - media.minted_at
148        return max(MIN_CLIP_MEDIA_LIFETIME, round(CLIP_STREAMDETAILS_EXPIRATION - elapsed))
149
150    def _tts_language(self, host_language: str | None = None) -> str | None:
151        """
152        Return the host's language, or the server locale, as a hyphenated language code.
153
154        :param host_language: The host's configured language override, if any.
155        """
156        locale = (host_language or "").strip() or self.mass.metadata.locale
157        return locale.replace("_", "-") if locale else None
158
159    def _find_clip_item(self, clip_id: str) -> QueueItem | None:
160        """Return the queue item holding the given clip, or None when no queue holds it."""
161        for queue_id in self._candidate_queue_ids(clip_id):
162            if (item := self._find_clip_in_queue(clip_id, queue_id)) is not None:
163                return item
164        return None
165
166    def _candidate_queue_ids(self, clip_id: str) -> list[str]:
167        """
168        Return the queue ids to search for a clip, the most likely one first.
169
170        The owning session knows its queue, but the session registry is empty after a
171        restart while the clip lives on in the persisted queue, so every queue stays a
172        candidate. Clip ids carry a uuid4-based session id, so a hit is unambiguous.
173        """
174        queue_ids = [queue.queue_id for queue in self.mass.player_queues.all()]
175        session = self._sessions.get(clip_id.rpartition("_")[0])
176        if session is not None and session.queue_id in queue_ids:
177            queue_ids.remove(session.queue_id)
178            queue_ids.insert(0, session.queue_id)
179        return queue_ids
180
181    def _find_clip_in_queue(self, clip_id: str, queue_id: str) -> QueueItem | None:
182        """Return the queue item holding the given clip, paging through the queue."""
183        page_size = 500
184        offset = 0
185        while True:
186            page = self.mass.player_queues.items(queue_id, limit=page_size, offset=offset)
187            if not page:
188                return None
189            for item in page:
190                if item.media_item is not None and item.media_item.item_id == clip_id:
191                    return item
192            if len(page) < page_size:
193                return None
194            offset += page_size
195
196    async def _generate_script(self, queue_item: QueueItem, prompt: str, clip_id: str) -> str:
197        """Resolve the deferred placeholders and generate the spoken script."""
198        attributes = queue_item.extra_attributes
199        deferred = await self._resolve_deferred_placeholders(prompt)
200        resolved = prompt
201        for key, value in deferred.items():
202            resolved = resolved.replace(key, value)
203        host = self._hosts.get(str(attributes.get(ATTR_HOST_ID) or "")) or {}
204        instructions = str(host.get("instructions") or "")
205        language = str(host.get("language") or "")
206        max_chars = int(attributes.get(ATTR_MAX_CHARS) or 0)
207        web_mode = str(attributes.get(ATTR_WEB_SEARCH_MODE) or "disabled")
208        try:
209            text = cast(
210                "str",
211                await self._generate_text(
212                    instructions=instructions,
213                    prompt=resolved,
214                    web_mode=web_mode,
215                    language=language,
216                ),
217            )
218        except Exception as err:
219            self.logger.warning(
220                "AI Radio clip %s (%s) failed to generate: %s", clip_id, queue_item.name, err
221            )
222            self._record_skip(queue_item, f"generation failed: {err}")
223            raise MediaNotFoundError(f"AI Radio clip {clip_id} failed to generate") from err
224        if max_chars > 0:
225            text = soft_limit_text(text, max_chars=max_chars)
226        self.logger.debug(
227            "AI Radio clip %s (%s) rendered: %d chars", clip_id, queue_item.name, len(text)
228        )
229        return text
230
231    async def _resolve_deferred_placeholders(self, prompt: str) -> dict[str, str]:
232        """Return freshly resolved values for the placeholders deferred until airtime."""
233        values = dict.fromkeys(DEFERRED_PLACEHOLDERS, "")
234        values["<timestamp>"] = self._configured_now().strftime("%Y-%m-%d %H:%M %Z")
235        # weather is the only deferred placeholder that costs a network round-trip, so it is
236        # only fetched when the prompt actually references it
237        weather_tokens = ("<weather_hourly>", "<weather_daily>")
238        if any(token in prompt for token in weather_tokens):
239            values.update(await self._prepare_weather_tokens())
240        return values
241
242    async def _mint_clip_media(
243        self, queue_item: QueueItem, text: str, clip_id: str
244    ) -> tuple[str, StreamType, AudioFormat, int | None]:
245        """Convert the script to playable audio via the configured TTS engine."""
246        host = self._hosts.get(str(queue_item.extra_attributes.get(ATTR_HOST_ID) or "")) or {}
247        engine_uid = str(host.get("tts_engine") or "") or None
248        language = self._tts_language(str(host.get("language") or ""))
249        try:
250            path, stream_type, audio_format = await self._render_tts_media(
251                text, engine_uid, language
252            )
253            # the probe is the first fetch, so a failed render surfaces here and not in playback
254            duration = await self._probe_duration(path)
255            return path, stream_type, audio_format, duration
256        except Exception as err:
257            self.logger.warning("AI Radio clip %s failed TTS: %s", clip_id, err)
258            self._record_skip(queue_item, f"TTS failed: {err}")
259            raise MediaNotFoundError(f"AI Radio clip {clip_id} failed TTS") from err
260
261    async def _render_tts_media(
262        self, text: str, engine_uid: str | None = None, language: str | None = None
263    ) -> tuple[str, StreamType, AudioFormat]:
264        """Ask the TTS engine for audio and return the path, stream type and format to play it."""
265        engine = await self._get_tts_engine(engine_uid)
266        try:
267            stream_details = await query_tts_engine(engine, text, language)
268        except TimeoutError, MusicAssistantError:
269            # a timeout or our own structured failure is not a language rejection, so a
270            # language-less retry would not help and would only double the wait
271            raise
272        except Exception as err:
273            if language is None:
274                raise
275            # some engines reject a language they don't support; fall back to the engine's
276            # own default voice rather than losing the clip entirely
277            self.logger.warning(
278                "AI Radio TTS engine '%s' rejected language '%s' (%s), retrying with its "
279                "default voice",
280                engine.uid,
281                language,
282                err,
283            )
284            stream_details = await query_tts_engine(engine, text, None)
285        path, stream_type = await resolve_tts_stream_path(engine, stream_details)
286        audio_format = stream_details.audio_format
287        if audio_format.content_type == ContentType.UNKNOWN:
288            audio_format = AudioFormat(content_type=ContentType.MP3)
289        return path, stream_type, audio_format
290
291    async def _probe_duration(self, path: str) -> int | None:
292        """Return the clip duration in seconds, or None when it cannot be determined."""
293        try:
294            tags = await async_parse_tags(path, require_duration=True)
295        except (InvalidDataError, OSError) as err:
296            if any(marker in str(err) for marker in TTS_SERVER_ERROR_MARKERS):
297                raise MusicAssistantError(
298                    "Error during TTS generation. Does your TTS provider have enough credit? "
299                    "Check the logs of your TTS provider for the reason."
300                ) from err
301            self.logger.warning("Could not determine AI Radio clip duration: %s", err)
302            return None
303        return int(tags.duration) if tags.duration else None
304
305    def _record_skip(self, queue_item: QueueItem, error: str) -> None:
306        """Record a skipped clip on its owning session."""
307        session_id = str(queue_item.extra_attributes.get(ATTR_SESSION_ID) or "")
308        if (session := self._sessions.get(session_id)) is None:
309            return
310        session.skipped_sections += 1
311        session.last_render_error = error
312