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