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