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 pathlib import Path
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
21
22from .constants import (
23    ATTR_MAX_CHARS,
24    ATTR_PROMPT,
25    ATTR_RENDERED_TEXT,
26    ATTR_SESSION_ID,
27    ATTR_STATION_ID,
28    ATTR_WEB_SEARCH_MODE,
29    CLIP_STREAMDETAILS_EXPIRATION,
30    DEFERRED_PLACEHOLDERS,
31    TTS_QUERY_TIMEOUT_SECONDS,
32    TTS_SERVER_ERROR_MARKERS,
33)
34from .helpers import soft_limit_text
35
36if TYPE_CHECKING:
37    from music_assistant_models.enums import MediaType
38    from music_assistant_models.queue_item import QueueItem
39
40    from music_assistant.mass import MusicAssistant
41
42    from .models import SessionState
43
44
45class AIRadioRenderMixin:
46    """Renders an AI Radio clip at the moment MA needs its audio."""
47
48    if TYPE_CHECKING:
49        mass: MusicAssistant
50        logger: logging.Logger
51        _stations: dict[str, dict[str, Any]]
52        _sessions: dict[str, SessionState]
53
54    _render_locks: dict[str, asyncio.Lock]
55
56    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
57        """
58        Render the AI Radio clip with the given id and return its StreamDetails.
59
60        :param item_id: The clip id of the queue item MA wants to play.
61        :param media_type: The media type of the requested item.
62        """
63        queue_item = self._find_clip_item(item_id)
64        if queue_item is None:
65            raise MediaNotFoundError(f"AI Radio clip {item_id} is not in any queue")
66        prompt = str(queue_item.extra_attributes.get(ATTR_PROMPT) or "")
67        if not prompt:
68            self._record_skip(queue_item, "clip has no prompt to render")
69            raise MediaNotFoundError(f"AI Radio clip {item_id} has no prompt to render")
70
71        async with self._lock_for(item_id):
72            text = str(queue_item.extra_attributes.get(ATTR_RENDERED_TEXT) or "")
73            if not text:
74                text = await self._generate_script(queue_item, prompt, item_id)
75                queue_item.extra_attributes[ATTR_RENDERED_TEXT] = text
76                # the signal is what marks the items cache dirty and schedules the persist
77                self.mass.player_queues.signal_update(queue_item.queue_id, items_changed=True)
78            path, stream_type, audio_format, duration = await self._mint_clip_media(
79                queue_item, text, item_id
80            )
81
82        return StreamDetails(
83            provider=self.instance_id,
84            item_id=item_id,
85            audio_format=audio_format,
86            media_type=media_type,
87            stream_type=stream_type,
88            path=path,
89            duration=duration,
90            # a talk clip has nothing worth seeking to, and a seek is the one path that
91            # would re-fetch a possibly-expired HA url mid-playback
92            can_seek=False,
93            allow_seek=False,
94            expiration=CLIP_STREAMDETAILS_EXPIRATION,
95        )
96
97    def _lock_for(self, clip_id: str) -> asyncio.Lock:
98        """Return the per-clip render lock, creating it on first use."""
99        if not hasattr(self, "_render_locks"):
100            self._render_locks = {}
101        if clip_id not in self._render_locks:
102            self._render_locks[clip_id] = asyncio.Lock()
103        return self._render_locks[clip_id]
104
105    def _find_clip_item(self, clip_id: str) -> QueueItem | None:
106        """Return the queue item holding the given clip, or None when no queue holds it."""
107        for queue_id in self._candidate_queue_ids(clip_id):
108            if (item := self._find_clip_in_queue(clip_id, queue_id)) is not None:
109                return item
110        return None
111
112    def _candidate_queue_ids(self, clip_id: str) -> list[str]:
113        """
114        Return the queue ids to search for a clip, the most likely one first.
115
116        The owning session knows its queue, but the session registry is empty after a
117        restart while the clip lives on in the persisted queue, so every queue stays a
118        candidate. Clip ids carry a uuid4-based session id, so a hit is unambiguous.
119        """
120        queue_ids = [queue.queue_id for queue in self.mass.player_queues.all()]
121        session = self._sessions.get(clip_id.rpartition("_")[0])
122        if session is not None and session.queue_id in queue_ids:
123            queue_ids.remove(session.queue_id)
124            queue_ids.insert(0, session.queue_id)
125        return queue_ids
126
127    def _find_clip_in_queue(self, clip_id: str, queue_id: str) -> QueueItem | None:
128        """Return the queue item holding the given clip, paging through the queue."""
129        page_size = 500
130        offset = 0
131        while True:
132            page = self.mass.player_queues.items(queue_id, limit=page_size, offset=offset)
133            if not page:
134                return None
135            for item in page:
136                if item.media_item is not None and item.media_item.item_id == clip_id:
137                    return item
138            if len(page) < page_size:
139                return None
140            offset += page_size
141
142    async def _generate_script(self, queue_item: QueueItem, prompt: str, clip_id: str) -> str:
143        """Resolve the deferred placeholders and generate the spoken script."""
144        attributes = queue_item.extra_attributes
145        station = self._stations.get(str(attributes.get(ATTR_STATION_ID) or "")) or {}
146        deferred = await self._resolve_deferred_placeholders(station)
147        resolved = prompt
148        for key, value in deferred.items():
149            resolved = resolved.replace(key, value)
150        max_chars = int(attributes.get(ATTR_MAX_CHARS) or 0)
151        web_mode = str(attributes.get(ATTR_WEB_SEARCH_MODE) or "disabled")
152        try:
153            text = cast(
154                "str",
155                await self._generate_text(station=station, prompt=resolved, web_mode=web_mode),
156            )
157        except Exception as err:
158            self.logger.warning(
159                "AI Radio clip %s (%s) failed to generate: %s", clip_id, queue_item.name, err
160            )
161            self._record_skip(queue_item, f"generation failed: {err}")
162            raise MediaNotFoundError(f"AI Radio clip {clip_id} failed to generate") from err
163        if max_chars > 0:
164            text = soft_limit_text(text, max_chars=max_chars)
165        self.logger.debug(
166            "AI Radio clip %s (%s) rendered: %d chars", clip_id, queue_item.name, len(text)
167        )
168        return text
169
170    async def _resolve_deferred_placeholders(self, station: dict[str, Any]) -> dict[str, str]:
171        """Return freshly resolved values for the placeholders deferred until airtime."""
172        values = dict.fromkeys(DEFERRED_PLACEHOLDERS, "")
173        values["<timestamp>"] = self._configured_now().strftime("%Y-%m-%d %H:%M %Z")
174        values.update(await self._prepare_runtime_tokens(station))
175        return values
176
177    async def _mint_clip_media(
178        self, queue_item: QueueItem, text: str, clip_id: str
179    ) -> tuple[str, StreamType, AudioFormat, int | None]:
180        """Convert the script to playable audio via the configured TTS engine."""
181        try:
182            path, stream_type, audio_format = await self._render_tts_media(text)
183            # the probe is the first fetch, so a failed render surfaces here and not in playback
184            duration = await self._probe_duration(path)
185            return path, stream_type, audio_format, duration
186        except Exception as err:
187            self.logger.warning("AI Radio clip %s failed TTS: %s", clip_id, err)
188            self._record_skip(queue_item, f"TTS failed: {err}")
189            raise MediaNotFoundError(f"AI Radio clip {clip_id} failed TTS") from err
190
191    async def _render_tts_media(self, text: str) -> tuple[str, StreamType, AudioFormat]:
192        """Ask the TTS engine for audio and return the path, stream type and format to play it."""
193        engine = await self._get_tts_engine()
194        try:
195            async with asyncio.timeout(TTS_QUERY_TIMEOUT_SECONDS) as query_timeout:
196                stream_details = await engine.provider.get_tts_message(text, engine_id=engine.id)
197        except TimeoutError as err:
198            # expired() tells our own cap apart from a timeout raised inside the engine
199            if not query_timeout.expired():
200                raise
201            raise MusicAssistantError(
202                f"TTS engine '{engine.uid}' did not respond within {TTS_QUERY_TIMEOUT_SECONDS}s"
203            ) from err
204        path = str(getattr(stream_details, "path", "") or "").strip()
205        if path.startswith(("http://", "https://", "rtsp://", "rtmp://")):
206            stream_type = StreamType.HTTP
207        elif path and Path(path).is_absolute() and await asyncio.to_thread(Path(path).is_file):
208            stream_type = StreamType.LOCAL_FILE
209        else:
210            raise InvalidDataError(
211                f"TTS engine '{engine.uid}' returned an unusable stream path: "
212                f"{path or '<empty>'}. StreamDetails.path must be a fetchable "
213                "http(s)/rtsp/rtmp URL or the absolute path of an existing local file."
214            )
215        audio_format = stream_details.audio_format
216        if audio_format.content_type == ContentType.UNKNOWN:
217            audio_format = AudioFormat(content_type=ContentType.MP3)
218        return path, stream_type, audio_format
219
220    async def _probe_duration(self, path: str) -> int | None:
221        """Return the clip duration in seconds, or None when it cannot be determined."""
222        try:
223            tags = await async_parse_tags(path, require_duration=True)
224        except (InvalidDataError, OSError) as err:
225            if any(marker in str(err) for marker in TTS_SERVER_ERROR_MARKERS):
226                raise MusicAssistantError(
227                    "Error during TTS generation. Does your TTS provider have enough credit? "
228                    "Check the logs of your TTS provider for the reason."
229                ) from err
230            self.logger.warning("Could not determine AI Radio clip duration: %s", err)
231            return None
232        return int(tags.duration) if tags.duration else None
233
234    def _record_skip(self, queue_item: QueueItem, error: str) -> None:
235        """Record a skipped clip on its owning session."""
236        session_id = str(queue_item.extra_attributes.get(ATTR_SESSION_ID) or "")
237        if (session := self._sessions.get(session_id)) is None:
238            return
239        session.skipped_sections += 1
240        session.last_render_error = error
241