music-assistant-server

1"""Runtime execution mixin for AI Radio."""
2# mypy: disable-error-code="attr-defined"
3
4from __future__ import annotations
5
6import asyncio
7import datetime
8import logging
9import random
10import time
11from collections import defaultdict
12from copy import deepcopy
13from pathlib import Path
14from typing import TYPE_CHECKING, Any, cast
15from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
16
17from aiohttp import ClientTimeout
18from music_assistant_models.enums import (
19    EventType,
20    ImageType,
21    MediaType,
22    PlaybackState,
23)
24from music_assistant_models.errors import MusicAssistantError
25from music_assistant_models.media_items import (
26    MediaItemImage,
27    ProviderMapping,
28    SoundEffect,
29    UniqueList,
30)
31
32from music_assistant.controllers.player_queues.helpers import build_queue_item
33from music_assistant.helpers.datetime import now, utc
34from music_assistant.helpers.json import json_loads
35from music_assistant.helpers.plugin_engines import resolve_ai_engine, resolve_tts_engine
36from music_assistant.helpers.uri import create_uri
37
38from .constants import (
39    AI_QUERY_TIMEOUT_SECONDS,
40    ATTR_HOST_ID,
41    ATTR_MAX_CHARS,
42    ATTR_PROMPT,
43    ATTR_SESSION_ID,
44    ATTR_STATION_ID,
45    ATTR_WEB_SEARCH_MODE,
46    CONF_AI_ENGINE,
47    CONF_TIMEZONE,
48    CONF_TTS_ENGINE,
49    CONF_WEATHER_CITY,
50    CONF_WEATHER_COUNTRY,
51    CONF_WEATHER_PROVIDER,
52    CONF_WEATHER_TIMEOUT,
53    DEFAULT_LLM_INSTRUCTIONS,
54    DEFAULT_WEATHER_PROVIDER,
55    DEFAULT_WEATHER_TIMEOUT_SECONDS,
56    DEFERRED_PLACEHOLDERS,
57    SHOW_START_TIMEOUT_SECONDS,
58    TTS_PRONUNCIATION_INSTRUCTIONS,
59    VALID_WEB_SEARCH_MODES,
60    WEB_SEARCH_MODE_RANK,
61)
62from .helpers import (
63    build_slots,
64    coerce_float,
65    coerce_int,
66    is_empty_section,
67    pick_weighted_choice,
68    slugify,
69    track_songinfo,
70    utc_now_iso,
71)
72from .models import (
73    PlannedSection,
74    SessionState,
75    Slot,
76)
77
78if TYPE_CHECKING:
79    from music_assistant_models.config_entries import ConfigValueType, ProviderConfig
80    from music_assistant_models.event import MassEvent
81    from music_assistant_models.media_items import PlayableMediaItemType
82    from music_assistant_models.queue_item import QueueItem
83
84    from music_assistant.mass import MusicAssistant
85    from music_assistant.models.plugin import AIEngine, TTSEngine
86
87
88# the sticky queue DJ re-plans on every queue change, so an uncached forecast lookup would
89# add two HTTP round trips to each one. Weather does not move meaningfully within this window
90WEATHER_TOKENS_CACHE_SECONDS = 300
91
92
93class AIRadioRuntimeMixin:
94    """Mixin with all runtime logic for AI Radio runs."""
95
96    # (fetched_at, tokens) of the last weather lookup, shared by the show and DJ paths
97    _weather_tokens_cache: tuple[float, dict[str, str]] | None = None
98
99    if TYPE_CHECKING:
100        mass: MusicAssistant
101        config: ProviderConfig
102        logger: logging.Logger
103        _sessions: dict[str, SessionState]
104
105        def get_setup_value(self, key: str, default: ConfigValueType = None) -> ConfigValueType:
106            """Return a value collected by this provider's setup flow."""
107
108        def _schedule_replan(self, queue_id: str) -> None:
109            """Request a replan pass for the given queue."""
110
111        async def set_queue_dj(self, queue_id: str, host_id: str | None) -> dict[str, str]:
112            """Enable, switch or disable the sticky AI DJ on a queue."""
113
114    def _set_session_progress(
115        self,
116        session: SessionState,
117        phase: str,
118        **details: Any,
119    ) -> None:
120        """Set progress payload with a stable phase key."""
121        session.progress = {
122            "phase": phase,
123            # Keep legacy key for compatibility with older UI code.
124            "step": phase,
125            **details,
126        }
127
128    def _build_program(self, station: dict[str, Any], host: dict[str, Any]) -> dict[str, Any]:
129        """Merge a station and its host into the dict the planner consumes."""
130        sections, missing = self._materialize_sections(list(host.get("section_ids", [])))
131        if missing:
132            raise MusicAssistantError(
133                f"Host references unknown sections: {', '.join(sorted(set(missing)))}"
134            )
135        return {
136            **deepcopy(station),
137            "host_id": str(host.get("id", "")),
138            "instructions": str(host.get("instructions", "")),
139            "tts_engine": str(host.get("tts_engine", "")),
140            "language": str(host.get("language", "")),
141            "sections": sections,
142            "section_order": deepcopy(host.get("section_order", [])),
143            "merge_section_id": str(host.get("merge_section_id", "")),
144        }
145
146    async def _run_session(self, session_id: str, program: dict[str, Any]) -> None:
147        """Run one session in the background."""
148        session = self._sessions[session_id]
149        session.started_at = utc_now_iso()
150        self.logger.info(
151            "AI Radio run started: session=%s station=%s",
152            session.session_id,
153            session.station_id,
154        )
155        try:
156            result = await self._run_show(session, program)
157            session.result = result
158            queue_stopped = result.get("ended_reason") == "queue_stopped"
159            session.status = "stopped" if queue_stopped else "completed"
160            self.logger.info(
161                "AI Radio run %s: session=%s station=%s",
162                session.status,
163                session.session_id,
164                session.station_id,
165            )
166        except asyncio.CancelledError:
167            session.status = "stopped"
168            self.logger.info(
169                "AI Radio run cancelled: session=%s station=%s",
170                session.session_id,
171                session.station_id,
172            )
173            raise
174        except Exception as err:
175            session.status = "failed"
176            session.error = str(err).strip() or err.__class__.__name__
177            self.logger.exception("AI Radio session failed: %s", err)
178        finally:
179            session.ended_at = utc_now_iso()
180            # a show session blocks queue DJ replans while it runs, so ending it must
181            # re-arm the DJ itself instead of waiting on the next queue change
182            if session.queue_id:
183                self._schedule_replan(session.queue_id)
184
185    async def _run_show(
186        self,
187        session: SessionState,
188        program: dict[str, Any],
189    ) -> dict[str, Any]:
190        """Plan and queue the whole show in one pass, then start playback."""
191        program = deepcopy(program)
192        self.logger.debug(
193            "Show starting for station '%s' (%s)",
194            program.get("name", "AI Radio"),
195            program.get("id", ""),
196        )
197        self._set_session_progress(session, "fetch_source_tracks")
198        # runtime_tokens only feeds the require_placeholders_present guards below; its
199        # resolved text is discarded here and re-fetched fresh when each clip renders
200        runtime_tokens = await self._prepare_runtime_tokens(program)
201        player_id = str(program.get("default_player_id") or "").strip()
202        if not player_id:
203            raise MusicAssistantError("AI Radio requires a target player")
204        if not self.mass.players.get_player(player_id):
205            raise MusicAssistantError(f"Unknown target player: {player_id}")
206
207        tracks, playlist_name = await self._fetch_source_tracks(program)
208        tracks = self._apply_source_shuffle(tracks, program)
209        tracks = self._apply_track_duration_limit(tracks, program)
210        if not tracks:
211            raise MusicAssistantError("No source tracks available after applying station limits")
212
213        # a grouped player plays from the group leader's queue, so resolve the
214        # active queue up front and target that one for queueing and polling
215        queue_id = player_id
216        active_queue = self.mass.player_queues.get_active_queue(player_id)
217        if active_queue is not None:
218            queue_id = str(active_queue.queue_id)
219        # a queue runs one host at a time; the show is now that host, so any sticky
220        # DJ assignment on the queue is cleared before the show takes it over
221        await self.set_queue_dj(queue_id, None)
222        self.mass.player_queues.clear(queue_id)
223        session.queue_id = queue_id
224
225        # a shuffled queue reorders each batch, scattering sections away from their tracks
226        await self.mass.player_queues.set_shuffle(queue_id, False)
227
228        cumulative_minutes = [0.0]
229        for track in tracks:
230            duration = track.get("duration")
231            seconds = (
232                float(duration) if isinstance(duration, (int, float)) and duration > 0 else 210.0
233            )
234            cumulative_minutes.append(cumulative_minutes[-1] + (seconds / 60.0))
235
236        self._set_session_progress(session, "planning_sections", total_tracks=len(tracks))
237        planned_sections, _history = self._plan_sections(
238            session_id=session.session_id,
239            tracks=tracks,
240            program=program,
241            track_index_offset=0,
242            minute_offset=0.0,
243            history_state={},
244            allowed_slot_when=None,
245            runtime_tokens=runtime_tokens,
246        )
247        queue_items = self._compose_queue_items(
248            queue_id=queue_id,
249            session=session,
250            program=program,
251            tracks=tracks,
252            sections=planned_sections,
253        )
254        if not queue_items:
255            raise MusicAssistantError("No queue entries were generated")
256
257        self._set_session_progress(
258            session,
259            "initializing_queue",
260            total_tracks=len(tracks),
261            queue_entries=len(queue_items),
262            queue_id=queue_id,
263        )
264        # load() stages the items without starting playback, so every clip already carries its
265        # prompt by the time anything can ask for its audio
266        await self.mass.player_queues.load(
267            queue_id,
268            queue_items=queue_items,
269            keep_remaining=False,
270            keep_played=False,
271            shuffle=False,
272        )
273        await self.mass.player_queues.play_index(queue_id, 0)
274        self._set_session_progress(
275            session,
276            "running",
277            total_tracks=len(tracks),
278            queue_entries=len(queue_items),
279            queue_id=queue_id,
280        )
281        has_clips = any(ATTR_SESSION_ID in item.extra_attributes for item in queue_items)
282        ended_reason = await self._await_show_end(
283            session, queue_id, len(queue_items) - 1, has_clips=has_clips
284        )
285        return {
286            "ended_reason": ended_reason,
287            "source_playlist_name": playlist_name,
288            "source_tracks": len(tracks),
289            "queue_id": queue_id,
290            "queue_entries": len(queue_items),
291            "planned_sections": len(planned_sections),
292            "skipped_sections": session.skipped_sections,
293        }
294
295    async def _await_show_end(
296        self, session: SessionState, queue_id: str, last_index: int, *, has_clips: bool
297    ) -> str:
298        """
299        Block until this session's show is over and report why it ended.
300
301        :param session: The session whose clips are in the queue.
302        :param queue_id: The queue playing the show.
303        :param last_index: Queue index of the final entry this session enqueued.
304        :param has_clips: Whether this run enqueued any AI Radio clips at all. A clip-free
305            show (every section was skipped by its rules) must not be mistaken for one whose
306            clips were cleared out from under it, so that rule is skipped entirely here.
307        :return: ``"source_exhausted"`` when the show played out, ``"queue_stopped"`` when the
308            queue was stopped or taken over before reaching the end.
309        :raises MusicAssistantError: if playback never starts within
310            :data:`SHOW_START_TIMEOUT_SECONDS`.
311        """
312        finished = asyncio.Event()
313        playback_started = asyncio.Event()
314        # a queue that has not started yet must never be mistaken for a stopped one
315        playback_seen = False
316        ended_reason = "queue_stopped"
317
318        def _check_show_state() -> None:
319            nonlocal playback_seen, ended_reason
320            queue = self.mass.player_queues.get(queue_id)
321            if queue is None:
322                finished.set()
323                return
324            if queue.state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
325                playback_seen = True
326                playback_started.set()
327            if has_clips and not self._session_has_clips(queue_id, session.session_id):
328                finished.set()
329                return
330            if not playback_seen or queue.state != PlaybackState.IDLE:
331                return
332            # playing out and being stopped both end IDLE, so position is the discriminator
333            current_index = queue.current_index
334            if current_index is not None and current_index >= last_index:
335                ended_reason = "source_exhausted"
336            self.logger.info(
337                "Queue %s went idle at index %s of %s, ending show (%s)",
338                queue_id,
339                current_index,
340                last_index,
341                ended_reason,
342            )
343            finished.set()
344
345        def _on_queue_event(_event: MassEvent) -> None:
346            _check_show_state()
347
348        unsubscribe = self.mass.subscribe(
349            _on_queue_event,
350            (EventType.QUEUE_UPDATED, EventType.QUEUE_ITEMS_UPDATED, EventType.PLAYER_REMOVED),
351            id_filter=queue_id,
352        )
353        try:
354            # the queue may already have gone away, or (for a show with clips) already lost
355            # them, by the time this subscribes; IDLE-after-playout still needs a fresh event,
356            # since playback_seen is not latched yet
357            _check_show_state()
358            await self._await_playback_start(playback_started, finished)
359            await finished.wait()
360        finally:
361            unsubscribe()
362        return ended_reason
363
364    async def _await_playback_start(
365        self, playback_started: asyncio.Event, finished: asyncio.Event
366    ) -> None:
367        """
368        Wait for the show to either start playing or end before it ever did.
369
370        :param playback_started: Set once the queue is first observed playing or paused.
371        :param finished: Set once the show is over, however that came about.
372        :raises MusicAssistantError: if neither happens within
373            :data:`SHOW_START_TIMEOUT_SECONDS`.
374        """
375        if playback_started.is_set() or finished.is_set():
376            return
377        # a player that never comes online (or whose clips all fail) must not pin this
378        # session's "running" status, and its max-concurrent-runs slot, forever
379        wait_tasks = (
380            asyncio.ensure_future(playback_started.wait()),
381            asyncio.ensure_future(finished.wait()),
382        )
383        try:
384            done, _pending = await asyncio.wait(
385                wait_tasks,
386                timeout=SHOW_START_TIMEOUT_SECONDS,
387                return_when=asyncio.FIRST_COMPLETED,
388            )
389        finally:
390            for task in wait_tasks:
391                if not task.done():
392                    task.cancel()
393        if not done:
394            raise MusicAssistantError(
395                f"Playback did not start within {SHOW_START_TIMEOUT_SECONDS}s"
396            )
397
398    def _session_has_clips(self, queue_id: str, session_id: str) -> bool:
399        """Return whether any queue item still belongs to the given session."""
400        page_size = 500
401        offset = 0
402        while True:
403            page = self.mass.player_queues.items(queue_id, limit=page_size, offset=offset)
404            if not page:
405                return False
406            if any(item.extra_attributes.get(ATTR_SESSION_ID) == session_id for item in page):
407                return True
408            if len(page) < page_size:
409                return False
410            offset += page_size
411
412    async def _fetch_source_tracks(
413        self, station: dict[str, Any]
414    ) -> tuple[list[dict[str, Any]], str]:
415        """Load and normalize source playlist tracks."""
416        playlist_id = str(station.get("source_playlist_id", "")).strip()
417        provider = str(station.get("source_playlist_provider", "library")).strip() or "library"
418        if not playlist_id:
419            raise MusicAssistantError("Station is missing source_playlist_id")
420
421        playlist = await self.mass.music.playlists.get(playlist_id, provider)
422        playlist_name = playlist.name
423        tracks = [track async for track in self.mass.music.playlists.tracks(playlist_id, provider)]
424        normalized: list[dict[str, Any]] = []
425        for track in tracks:
426            artist = ""
427            track_artists = getattr(track, "artists", None)
428            if isinstance(track_artists, list) and track_artists:
429                artist = str(track_artists[0].name)
430            uri = await self._track_to_uri(track)
431            if not uri:
432                self.logger.warning(
433                    "Skipping source track with no resolvable uri: %s - %s (item_id=%s)",
434                    artist,
435                    track.name,
436                    track.item_id,
437                )
438                continue
439            normalized.append(
440                {
441                    "index": len(normalized),
442                    "item_id": track.item_id,
443                    "name": track.name,
444                    "artist": artist,
445                    "songinfo": f"{artist} - {track.name}".strip(" -"),
446                    "duration": track.duration,
447                    "uri": uri,
448                    "media_item": track,
449                }
450            )
451        return normalized, playlist_name
452
453    async def _track_to_uri(self, track: PlayableMediaItemType) -> str:
454        """Resolve a stable URI for a source track."""
455        if track.uri:
456            return track.uri
457        ordered_mappings = sorted(
458            track.provider_mappings,
459            key=lambda mapping: mapping.quality,
460            reverse=True,
461        )
462        for mapping in ordered_mappings:
463            if not mapping.available:
464                continue
465            return create_uri(MediaType.TRACK, mapping.provider_instance, mapping.item_id)
466        return ""
467
468    def _apply_source_shuffle(
469        self, tracks: list[dict[str, Any]], station: dict[str, Any]
470    ) -> list[dict[str, Any]]:
471        """Return the source tracks in random order when the station asks for it."""
472        if not station.get("shuffle_source_tracks", True) or not tracks:
473            return tracks
474        indices = list(range(len(tracks)))
475        random.Random().shuffle(indices)
476        result: list[dict[str, Any]] = []
477        for new_index, old_index in enumerate(indices):
478            updated = deepcopy(tracks[old_index])
479            updated["index"] = new_index
480            updated["source_index"] = old_index
481            result.append(updated)
482        self.logger.info("Shuffled %d source tracks", len(result))
483        return result
484
485    def _apply_track_duration_limit(
486        self, tracks: list[dict[str, Any]], station: dict[str, Any]
487    ) -> list[dict[str, Any]]:
488        """Truncate the given tracks to the configured playtime cap, preserving their order."""
489        max_duration = float(station.get("max_duration_minutes", 0) or 0)
490        if max_duration <= 0 or not tracks:
491            return tracks
492        chosen: list[int] = []
493        total_minutes = 0.0
494        for index, track in enumerate(tracks):
495            duration = track.get("duration")
496            seconds = (
497                float(duration) if isinstance(duration, (int, float)) and duration > 0 else 210.0
498            )
499            chosen.append(index)
500            total_minutes += seconds / 60.0
501            if total_minutes > max_duration:
502                break
503        result: list[dict[str, Any]] = []
504        for new_index, old_index in enumerate(chosen):
505            updated = deepcopy(tracks[old_index])
506            updated["index"] = new_index
507            updated["source_index"] = old_index
508            result.append(updated)
509        self.logger.info(
510            "Applied source playtime cap: %.1f min requested, %d -> %d tracks selected",
511            max_duration,
512            len(tracks),
513            len(result),
514        )
515        return result
516
517    def _plan_sections(  # noqa: PLR0915
518        self,
519        session_id: str,
520        tracks: list[dict[str, Any]],
521        program: dict[str, Any],
522        track_index_offset: int,
523        minute_offset: float,
524        history_state: dict[str, list[tuple[int, float]]],
525        allowed_slot_when: list[str] | None,
526        runtime_tokens: dict[str, str],
527        decided_next_item_ids: set[str] | None = None,
528    ) -> tuple[list[PlannedSection], dict[str, list[tuple[int, float]]]]:
529        """Evaluate section rules and produce planning entries."""
530        sections = program.get("sections", [])
531        section_order = program.get("section_order", [])
532        if not isinstance(sections, list) or not sections:
533            raise MusicAssistantError("Station has no sections configured")
534        if not isinstance(section_order, list) or not section_order:
535            raise MusicAssistantError("Station has no section_order configured")
536
537        section_by_id = {
538            str(section.get("id", "")).strip(): section
539            for section in sections
540            if str(section.get("id", "")).strip()
541        }
542        slots = build_slots(tracks)
543        history = {section_id: list(events) for section_id, events in history_state.items()}
544        selected: list[tuple[str, Slot, dict[str, str]]] = []
545        rng = random.Random()
546
547        def slot_event(slot: Slot) -> tuple[int, float]:
548            song_local = slot.next_index if slot.next_index is not None else len(tracks)
549            return track_index_offset + song_local, minute_offset + slot.minute_mark
550
551        def register_event(section_id: str, slot: Slot) -> None:
552            if is_empty_section(section_id):
553                return
554            history.setdefault(section_id, []).append(slot_event(slot))
555
556        for slot in slots:
557            if allowed_slot_when and slot.when not in allowed_slot_when:
558                continue
559            if (
560                decided_next_item_ids
561                and slot.when == "between_songs"
562                and slot.next_index is not None
563                and str(tracks[slot.next_index].get("item_id", "")) in decided_next_item_ids
564            ):
565                # the caller settled this slot in an earlier run: re-evaluating it would
566                # consume a chance roll and register its event a second time
567                continue
568            matching_rules = [
569                rule for rule in section_order if str(rule.get("when", "")).strip() == slot.when
570            ]
571            if not matching_rules:
572                continue
573            static, deferred = self._resolve_placeholders(
574                program=program,
575                tracks=tracks,
576                slot=slot,
577                runtime_tokens=runtime_tokens,
578            )
579            # guards may require a deferred token to be present, so they see the merged view;
580            # only the static half is substituted into the stored prompt
581            guard_values = {**deferred, **static}
582            for rule in matching_rules:
583                flow = rule.get("flow", [])
584                if not isinstance(flow, list):
585                    continue
586                for flow_item in flow:
587                    if not isinstance(flow_item, dict):
588                        continue
589                    if "MUST" in flow_item:
590                        section_id = str(flow_item["MUST"]).strip()
591                        if not section_id:
592                            continue
593                        if is_empty_section(section_id):
594                            continue
595                        selected.append((section_id, slot, static))
596                        register_event(section_id, slot)
597                        continue
598                    if "ALTERNATIVE" in flow_item:
599                        alternative = flow_item["ALTERNATIVE"]
600                        if not isinstance(alternative, dict):
601                            continue
602                        section_id = pick_weighted_choice(alternative.get("choices", []), rng)
603                        if is_empty_section(section_id):
604                            continue
605                        selected.append((section_id, slot, static))
606                        register_event(section_id, slot)
607                        continue
608                    if "OPTIONAL" in flow_item:
609                        optional = flow_item["OPTIONAL"]
610                        if not isinstance(optional, dict):
611                            continue
612                        section_id = str(optional.get("section", "")).strip()
613                        if not section_id:
614                            continue
615                        chance_raw = coerce_float(optional.get("chance"), 0.0)
616                        chance = chance_raw / 100.0 if chance_raw > 1 else chance_raw
617                        if rng.random() > chance:
618                            continue
619                        guards = optional.get("guards", {}) if isinstance(optional, dict) else {}
620                        if not self._passes_optional_guards(
621                            section_id=section_id,
622                            guards=guards if isinstance(guards, dict) else {},
623                            history=history,
624                            slot=slot,
625                            tracks=tracks,
626                            placeholders=guard_values,
627                            track_index_offset=track_index_offset,
628                            minute_offset=minute_offset,
629                        ):
630                            continue
631                        if is_empty_section(section_id):
632                            continue
633                        selected.append((section_id, slot, static))
634                        register_event(section_id, slot)
635
636        merge_section_id = str(program.get("merge_section_id", "")).strip()
637        meta_section = section_by_id.get(merge_section_id) if merge_section_id else None
638        grouped: dict[str, list[tuple[str, Slot, dict[str, str]]]] = defaultdict(list)
639        for item in selected:
640            section_id, slot, placeholders = item
641            key = f"{slot.when}:{slot.at_index}"
642            grouped[key].append((section_id, slot, placeholders))
643
644        planned: list[PlannedSection] = []
645        order_index = 0
646        processed_keys: set[str] = set()
647        for section_id, slot, placeholders in selected:
648            key = f"{slot.when}:{slot.at_index}"
649            grouped_items = grouped[key]
650            if (
651                len(grouped_items) > 1
652                and slot.when == "between_songs"
653                and meta_section
654                and key not in processed_keys
655            ):
656                processed_keys.add(key)
657                merged = self._build_meta_section_plan(
658                    grouped_items=grouped_items,
659                    meta_section=meta_section,
660                    placeholders=placeholders,
661                    order=order_index,
662                    section_by_id=section_by_id,
663                    session_id=session_id,
664                    history_events=[(item[0], slot_event(item[1])) for item in grouped_items],
665                )
666                planned.append(merged)
667                order_index += 1
668                continue
669            if key in processed_keys:
670                continue
671            section = section_by_id.get(section_id)
672            if not section:
673                continue
674            if str(section.get("type", "ai_text")).strip().lower() != "ai_text":
675                continue
676            prompt = self._apply_placeholders(str(section.get("prompt", "")), placeholders)
677            max_chars = int((section.get("constraints") or {}).get("max_chars", 0) or 0)
678            if max_chars > 0:
679                prompt += (
680                    f"\n\nTarget length: around {max_chars} characters. It may exceed by up to "
681                    "15% if needed to finish naturally. Never stop mid-sentence."
682                )
683            planned.append(
684                PlannedSection(
685                    order=order_index,
686                    clip_id=f"{session_id}_{order_index:03d}",
687                    section_id=section_id,
688                    section_name=self._resolve_section_name(section, section_id),
689                    when=slot.when,
690                    insert_at_index=slot.at_index,
691                    prompt=prompt,
692                    max_chars=max_chars,
693                    web_search_mode=self._resolve_web_search_mode(section, section_id),
694                    history_events=[(section_id, slot_event(slot))],
695                )
696            )
697            order_index += 1
698
699        return planned, history
700
701    def _passes_optional_guards(
702        self,
703        section_id: str,
704        guards: dict[str, Any],
705        history: dict[str, list[tuple[int, float]]],
706        slot: Slot,
707        tracks: list[dict[str, Any]],
708        placeholders: dict[str, str],
709        track_index_offset: int,
710        minute_offset: float,
711    ) -> bool:
712        """Evaluate OPTIONAL section guards."""
713        min_gap_songs = coerce_int(guards.get("min_gap_songs"), 0)
714        max_per_60min = coerce_int(guards.get("max_per_60min"), 0)
715        required_placeholders = guards.get("require_placeholders_present", [])
716        events = history.get(section_id, [])
717        song_local = slot.next_index if slot.next_index is not None else len(tracks)
718        song_global = track_index_offset + song_local
719        minute_global = minute_offset + slot.minute_mark
720
721        if min_gap_songs > 0 and events:
722            if song_global - events[-1][0] < min_gap_songs:
723                return False
724        if max_per_60min > 0:
725            in_window = [event for event in events if (minute_global - event[1]) <= 60.0]
726            if len(in_window) >= max_per_60min:
727                return False
728        if isinstance(required_placeholders, list):
729            for token in required_placeholders:
730                if not placeholders.get(str(token), "").strip():
731                    return False
732        return True
733
734    def _build_meta_section_plan(
735        self,
736        grouped_items: list[tuple[str, Slot, dict[str, str]]],
737        meta_section: dict[str, Any],
738        placeholders: dict[str, str],
739        order: int,
740        section_by_id: dict[str, dict[str, Any]],
741        session_id: str,
742        history_events: list[tuple[str, tuple[int, float]]],
743    ) -> PlannedSection:
744        """Build a merged ai_meta section for one slot."""
745        section_ids = [item[0] for item in grouped_items]
746        slot = grouped_items[0][1]
747        prompt_lines: list[str] = []
748        total_max_chars = 0
749        max_web_mode = "disabled"
750        merged_names: list[str] = []
751        for index, section_id in enumerate(section_ids, start=1):
752            section = section_by_id.get(section_id, {})
753            section_name = self._resolve_section_name(section, section_id)
754            merged_names.append(section_name)
755            prompt_base = self._apply_placeholders(str(section.get("prompt", "")), placeholders)
756            max_chars = int((section.get("constraints") or {}).get("max_chars", 0) or 0)
757            total_max_chars += max_chars
758            prompt_lines.append(f"{index}. [{section_id}] {prompt_base}")
759            mode = self._resolve_web_search_mode(section, section_id)
760            if WEB_SEARCH_MODE_RANK[mode] > WEB_SEARCH_MODE_RANK[max_web_mode]:
761                max_web_mode = mode
762
763        meta_prompt = self._apply_placeholders(str(meta_section.get("prompt", "")), placeholders)
764        prompt_block = "\n".join(prompt_lines)
765        if "<section_drafts>" in meta_prompt:
766            meta_prompt = meta_prompt.replace("<section_drafts>", prompt_block)
767        else:
768            meta_prompt = f"{meta_prompt}\n\nSection prompts:\n{prompt_block}\n"
769        meta_prompt += (
770            "\n\nCreate one single moderator script that naturally combines all requested parts. "
771            "Return plain text only."
772        )
773        if total_max_chars > 0:
774            meta_prompt += (
775                f"\n\nTarget length: around {total_max_chars} characters total. It may exceed "
776                "by up to 15% if needed to finish naturally. Never stop mid-sentence."
777            )
778        section_id = f"multi_{'_'.join(slugify(item) for item in section_ids)}"
779        section_name = " + ".join(dict.fromkeys(merged_names))
780        return PlannedSection(
781            order=order,
782            clip_id=f"{session_id}_{order:03d}",
783            section_id=section_id,
784            section_name=section_name,
785            when=slot.when,
786            insert_at_index=slot.at_index,
787            prompt=meta_prompt,
788            max_chars=total_max_chars,
789            web_search_mode=max_web_mode,
790            history_events=history_events,
791        )
792
793    def _compose_queue_items(
794        self,
795        queue_id: str,
796        session: SessionState,
797        program: dict[str, Any],
798        tracks: list[dict[str, Any]],
799        sections: list[PlannedSection],
800    ) -> list[QueueItem]:
801        """
802        Build the queue items for a whole show.
803
804        Clips carry their render state in ``extra_attributes`` from the moment they are built, so
805        a clip is renderable as soon as the queue holds it.
806
807        :param queue_id: The queue the items are built for.
808        :param session: The session that owns the show.
809        :param program: The station+host program being played.
810        :param tracks: The normalized source tracks, in play order.
811        :param sections: The planned sections to interleave between them.
812        """
813        sections_by_index: dict[int, list[PlannedSection]] = defaultdict(list)
814        for item in sections:
815            sections_by_index[item.insert_at_index].append(item)
816        items: list[QueueItem] = []
817        for index in range(len(tracks) + 1):
818            for section in sorted(sections_by_index.get(index, []), key=lambda item: item.order):
819                items.append(
820                    self._section_to_clip_item(queue_id, session.session_id, program, section)
821                )
822            if index < len(tracks) and (media_item := tracks[index].get("media_item")) is not None:
823                items.append(build_queue_item(queue_id, media_item))
824        return items
825
826    def _section_to_clip_item(
827        self,
828        queue_id: str,
829        session_id: str,
830        program: dict[str, Any],
831        section: PlannedSection,
832    ) -> QueueItem:
833        """Build the queue item for a not-yet-rendered clip."""
834        clip = SoundEffect(
835            item_id=section.clip_id,
836            provider=self.instance_id,
837            name=section.section_name,
838            provider_mappings={
839                ProviderMapping(
840                    item_id=section.clip_id,
841                    provider_domain=self.domain,
842                    provider_instance=self.instance_id,
843                )
844            },
845        )
846        clip.metadata.images = UniqueList(
847            [
848                MediaItemImage(
849                    type=ImageType.THUMB,
850                    path=self._ai_radio_cover_image_path(),
851                    provider="builtin",
852                    remotely_accessible=False,
853                )
854            ]
855        )
856        queue_item = build_queue_item(queue_id, clip)
857        # the section name already travels as the item's own name, so it is not duplicated here
858        queue_item.extra_attributes.update(
859            {
860                ATTR_SESSION_ID: session_id,
861                ATTR_STATION_ID: str(program.get("id") or ""),
862                ATTR_HOST_ID: str(program.get("host_id") or ""),
863                ATTR_PROMPT: section.prompt,
864                ATTR_MAX_CHARS: section.max_chars,
865                ATTR_WEB_SEARCH_MODE: section.web_search_mode,
866            }
867        )
868        return queue_item
869
870    @staticmethod
871    def _ai_radio_cover_image_path() -> str:
872        """Return the explicit AI Radio playlist cover image path."""
873        return str(Path(__file__).with_name("air.png"))
874
875    async def _prepare_runtime_tokens(self, program: dict[str, Any]) -> dict[str, str]:
876        """Prepare runtime tokens (including weather placeholders) for one run."""
877        if not self._program_uses_weather_placeholders(program):
878            return {}
879        return await self._prepare_weather_tokens()
880
881    async def _prepare_weather_tokens(self) -> dict[str, str]:
882        """Return the weather placeholder tokens, fetching them at most once per cache window."""
883        cached = self._weather_tokens_cache
884        if cached is not None and (time.monotonic() - cached[0]) < WEATHER_TOKENS_CACHE_SECONDS:
885            return dict(cached[1])
886        tokens = await self._fetch_weather_tokens()
887        # failed and disabled lookups are cached too, so a broken forecast source cannot
888        # put its timeout in front of every replan pass
889        self._weather_tokens_cache = (time.monotonic(), tokens)
890        return dict(tokens)
891
892    async def _fetch_weather_tokens(self) -> dict[str, str]:
893        """Fetch and format weather placeholder tokens from the configured provider."""
894        runtime_tokens: dict[str, str] = {}
895
896        weather_provider = (
897            str(self.config.get_value(CONF_WEATHER_PROVIDER) or DEFAULT_WEATHER_PROVIDER)
898            .strip()
899            .lower()
900        )
901        if weather_provider in {"", "none", "disabled", "off"}:
902            return runtime_tokens
903        if weather_provider != "open_meteo":
904            self.logger.warning(
905                "Unsupported weather provider '%s' for AI Radio station",
906                weather_provider,
907            )
908            return runtime_tokens
909
910        city, country = self._extract_location()
911        if not city or not country:
912            self.logger.warning(
913                "Weather placeholders used but no location configured "
914                "(set the weather_city/weather_country provider options)"
915            )
916            return runtime_tokens
917
918        configured_timeout = self.config.get_value(CONF_WEATHER_TIMEOUT)
919        timeout_seconds = max(5, coerce_int(configured_timeout, DEFAULT_WEATHER_TIMEOUT_SECONDS))
920        try:
921            weather_hourly, weather_daily = await self._fetch_open_meteo_weather(
922                city=city,
923                country=country,
924                timeout_seconds=timeout_seconds,
925            )
926        except Exception as err:
927            self.logger.warning(
928                "Weather lookup failed for '%s, %s': %s",
929                city,
930                country,
931                err,
932            )
933            return runtime_tokens
934
935        if weather_hourly:
936            runtime_tokens["<weather_hourly>"] = weather_hourly
937        if weather_daily:
938            runtime_tokens["<weather_daily>"] = weather_daily
939        return runtime_tokens
940
941    def _program_uses_weather_placeholders(self, program: dict[str, Any]) -> bool:
942        """Return whether the program references weather placeholders."""
943        weather_tokens = ("<weather_hourly>", "<weather_daily>")
944        for section in program.get("sections", []):
945            prompt = str(section.get("prompt", ""))
946            if any(token in prompt for token in weather_tokens):
947                return True
948
949        for rule in program.get("section_order", []):
950            flow = rule.get("flow", [])
951            for item in flow:
952                optional = item.get("OPTIONAL")
953                if not optional:
954                    continue
955                guards = optional.get("guards", {})
956                required = guards.get("require_placeholders_present", [])
957                if any(str(token) in weather_tokens for token in required):
958                    return True
959        return False
960
961    def _extract_location(self) -> tuple[str, str]:
962        """Extract weather location (city/country) from the provider config."""
963        city = str(self.config.get_value(CONF_WEATHER_CITY) or "").strip()
964        country = str(self.config.get_value(CONF_WEATHER_COUNTRY) or "").strip()
965        return city, country
966
967    def _configured_now(self) -> datetime.datetime:
968        """Return the current time in the configured timezone, falling back to host local time."""
969        tz_name = str(self.config.get_value(CONF_TIMEZONE) or "").strip()
970        if tz_name:
971            try:
972                return utc().astimezone(ZoneInfo(tz_name))
973            except ZoneInfoNotFoundError, ValueError:
974                # a typo must not take the run down, but it should not pass unnoticed either
975                self.logger.warning(
976                    "Ignoring invalid timezone %r, falling back to the host timezone", tz_name
977                )
978        return now()
979
980    async def _fetch_open_meteo_weather(
981        self,
982        city: str,
983        country: str,
984        timeout_seconds: int,
985    ) -> tuple[str, str]:
986        """Fetch weather strings from Open-Meteo for weather placeholders."""
987        geocode_params: dict[str, str | int] = {
988            "name": city,
989            "count": 10,
990            "language": "en",
991            "format": "json",
992        }
993        country_code = country.upper() if len(country) == 2 and country.isalpha() else ""
994        if country_code:
995            geocode_params["country"] = country_code
996        geocode = await self._open_meteo_get_json(
997            "https://geocoding-api.open-meteo.com/v1/search",
998            geocode_params,
999            timeout_seconds,
1000        )
1001        results = geocode.get("results", [])
1002        if not isinstance(results, list) or not results:
1003            raise MusicAssistantError(f"No geocoding result for {city}, {country}")
1004
1005        selected = results[0]
1006        country_lc = country.lower()
1007        for candidate in results:
1008            if not isinstance(candidate, dict):
1009                continue
1010            candidate_country = str(candidate.get("country", "")).strip().lower()
1011            candidate_country_code = str(candidate.get("country_code", "")).strip().upper()
1012            if candidate_country and candidate_country == country_lc:
1013                selected = candidate
1014                break
1015            if country_code and candidate_country_code == country_code:
1016                selected = candidate
1017                break
1018
1019        if not isinstance(selected, dict):
1020            raise MusicAssistantError(f"No valid geocoding result for {city}, {country}")
1021        latitude_value: object = selected.get("latitude")
1022        longitude_value: object = selected.get("longitude")
1023        if not isinstance(latitude_value, (int, float, str)) or not isinstance(
1024            longitude_value, (int, float, str)
1025        ):
1026            raise MusicAssistantError(
1027                f"Geocoding result for {city}, {country} has invalid coordinates"
1028            )
1029        try:
1030            lat = float(latitude_value)
1031            lon = float(longitude_value)
1032        except ValueError as err:
1033            raise MusicAssistantError(
1034                f"Geocoding result for {city}, {country} has invalid coordinates"
1035            ) from err
1036        timezone_name = str(selected.get("timezone") or "UTC")
1037        forecast = await self._open_meteo_get_json(
1038            "https://api.open-meteo.com/v1/forecast",
1039            {
1040                "latitude": lat,
1041                "longitude": lon,
1042                "current": "temperature_2m,apparent_temperature,weather_code",
1043                "hourly": "temperature_2m,precipitation_probability,weather_code",
1044                "daily": (
1045                    "temperature_2m_max,temperature_2m_min,"
1046                    "precipitation_probability_max,weather_code"
1047                ),
1048                "forecast_days": 3,
1049                "timezone": timezone_name,
1050            },
1051            timeout_seconds,
1052        )
1053        return self._format_weather_strings(forecast)
1054
1055    async def _open_meteo_get_json(
1056        self,
1057        base_url: str,
1058        params: dict[str, Any],
1059        timeout_seconds: int,
1060    ) -> dict[str, Any]:
1061        """Perform one Open-Meteo GET request."""
1062        async with self.mass.http_session.get(
1063            base_url,
1064            params=params,
1065            timeout=ClientTimeout(total=timeout_seconds),
1066        ) as response:
1067            payload = await response.read()
1068            if response.status >= 400:
1069                raise MusicAssistantError(
1070                    f"Open-Meteo request failed ({response.status}): "
1071                    f"{payload.decode(errors='ignore')}"
1072                )
1073        data = json_loads(payload)
1074        if not isinstance(data, dict):
1075            raise MusicAssistantError("Open-Meteo response is not a JSON object")
1076        return data
1077
1078    def _format_weather_strings(self, payload: dict[str, Any]) -> tuple[str, str]:
1079        """Format Open-Meteo payload into weather placeholder strings."""
1080        hourly = payload.get("hourly", {})
1081        daily = payload.get("daily", {})
1082        current = payload.get("current", {})
1083        if not isinstance(hourly, dict):
1084            hourly = {}
1085        if not isinstance(daily, dict):
1086            daily = {}
1087        if not isinstance(current, dict):
1088            current = {}
1089
1090        hourly_times = hourly.get("time", [])
1091        hourly_temp = hourly.get("temperature_2m", [])
1092        hourly_prec = hourly.get("precipitation_probability", [])
1093        if not isinstance(hourly_times, list):
1094            hourly_times = []
1095        if not isinstance(hourly_temp, list):
1096            hourly_temp = []
1097        if not isinstance(hourly_prec, list):
1098            hourly_prec = []
1099
1100        current_time = str(current.get("time") or "").strip()
1101        start_index = 0
1102        if current_time and current_time in hourly_times:
1103            start_index = int(hourly_times.index(current_time))
1104
1105        max_items = min(len(hourly_times), len(hourly_temp), len(hourly_prec))
1106        hourly_parts: list[str] = []
1107        for index in range(start_index, min(start_index + 6, max_items)):
1108            ts = str(hourly_times[index]).replace("T", " ")
1109            hourly_parts.append(
1110                f"{ts}: {self._format_number(hourly_temp[index])}C, "
1111                f"rain {self._format_number(hourly_prec[index])}%"
1112            )
1113        current_text = ""
1114        if current:
1115            current_text = (
1116                f"now {self._format_number(current.get('temperature_2m'))}C "
1117                f"(feels {self._format_number(current.get('apparent_temperature'))}C)"
1118            )
1119        weather_hourly = "; ".join(([current_text] if current_text else []) + hourly_parts)
1120
1121        daily_times = daily.get("time", [])
1122        max_t = daily.get("temperature_2m_max", [])
1123        min_t = daily.get("temperature_2m_min", [])
1124        max_prec = daily.get("precipitation_probability_max", [])
1125        if not isinstance(daily_times, list):
1126            daily_times = []
1127        if not isinstance(max_t, list):
1128            max_t = []
1129        if not isinstance(min_t, list):
1130            min_t = []
1131        if not isinstance(max_prec, list):
1132            max_prec = []
1133        daily_parts: list[str] = []
1134        for index in range(min(len(daily_times), len(max_t), len(min_t), len(max_prec))):
1135            daily_parts.append(
1136                f"{daily_times[index]}: "
1137                f"{self._format_number(min_t[index])}-{self._format_number(max_t[index])}C, "
1138                f"rain {self._format_number(max_prec[index])}%"
1139            )
1140        weather_daily = "; ".join(daily_parts)
1141        return weather_hourly, weather_daily
1142
1143    def _format_number(self, value: Any) -> str:
1144        """Format weather numeric values compactly for prompts."""
1145        try:
1146            # the host reads these out loud, where a decimal place only clutters the line
1147            numeric = round(float(value))
1148        except Exception:
1149            return str(value)
1150        return str(numeric)
1151
1152    def _resolve_placeholders(
1153        self,
1154        program: dict[str, Any],
1155        tracks: list[dict[str, Any]],
1156        slot: Slot,
1157        runtime_tokens: dict[str, str],
1158    ) -> tuple[dict[str, str], dict[str, str]]:
1159        """
1160        Resolve placeholders for one slot, split by when they are substituted.
1161
1162        :param program: The station+host program being planned.
1163        :param tracks: The track list the slot indexes into.
1164        :param slot: The insertion slot being filled.
1165        :param runtime_tokens: Weather tokens fetched for this run.
1166        :return: ``(static, deferred)`` — static values are fixed by the track order and are
1167            substituted at plan time; deferred values describe the moment of airing and are
1168            substituted at render time.
1169        """
1170        prev_track = tracks[slot.prev_index] if slot.prev_index is not None else None
1171        next_track = tracks[slot.next_index] if slot.next_index is not None else None
1172        very_next_track = tracks[slot.very_next_index] if slot.very_next_index is not None else None
1173        static = {
1174            "<prev_songinfo>": track_songinfo(prev_track),
1175            "<next_songinfo>": track_songinfo(next_track),
1176            "<very_next_songinfo>": track_songinfo(very_next_track),
1177        }
1178        deferred = dict.fromkeys(DEFERRED_PLACEHOLDERS, "")
1179        deferred["<timestamp>"] = self._configured_now().strftime("%Y-%m-%d %H:%M %Z")
1180        for key, value in runtime_tokens.items():
1181            if str(key) in DEFERRED_PLACEHOLDERS:
1182                deferred[str(key)] = str(value)
1183            else:
1184                static[str(key)] = str(value)
1185        return static, deferred
1186
1187    def _apply_placeholders(self, prompt: str, values: dict[str, str]) -> str:
1188        """Apply placeholder replacements in a prompt."""
1189        text = prompt
1190        for key, value in values.items():
1191            text = text.replace(key, value)
1192        return text
1193
1194    def _resolve_section_name(self, section: dict[str, Any], fallback_id: str) -> str:
1195        """Resolve section display name."""
1196        name = str(section.get("name", "")).strip()
1197        return name or fallback_id.replace("_", " ")
1198
1199    def _resolve_web_search_mode(self, section: dict[str, Any], section_id: str) -> str:
1200        """Resolve and validate section web search mode."""
1201        mode = str(section.get("web_search", "disabled")).strip().lower()
1202        if mode not in VALID_WEB_SEARCH_MODES:
1203            raise MusicAssistantError(
1204                f"Invalid web_search mode '{mode}' in section '{section_id}'. "
1205                f"Allowed: {sorted(VALID_WEB_SEARCH_MODES)}"
1206            )
1207        return mode
1208
1209    async def _generate_text(
1210        self, instructions: str, prompt: str, web_mode: str, language: str | None = None
1211    ) -> str:
1212        """Generate one section text using the configured AI engine."""
1213        instructions = instructions.strip() or DEFAULT_LLM_INSTRUCTIONS
1214        query_parts: list[str] = []
1215        if instructions:
1216            query_parts.append(f"Program instructions:\n{instructions}")
1217        query_parts.append(f"Pronunciation rules:\n{TTS_PRONUNCIATION_INSTRUCTIONS}")
1218        # stated as a default so a station can still ask for another language in its instructions
1219        query_parts.append(
1220            "Unless the program instructions ask for another language, write the output "
1221            f"in the language matching the locale '{language or self.mass.metadata.locale}'."
1222        )
1223        if web_mode == "force":
1224            query_parts.append(
1225                "Web mode: force. Use current up-to-date information where relevant."
1226            )
1227        elif web_mode == "allow":
1228            query_parts.append(
1229                "Web mode: allow. Use current information if it improves the answer."
1230            )
1231        query_parts.append(
1232            f"Task: Write one concise spoken radio section.\n\n{prompt}\n\nReturn plain text only."
1233        )
1234        query = "\n\n".join(query_parts)
1235        engine = await self._get_ai_engine()
1236        self.logger.debug(
1237            "AI query prepared: engine=%s web_mode=%s query_chars=%d",
1238            engine.uid,
1239            web_mode,
1240            len(query),
1241        )
1242        try:
1243            async with asyncio.timeout(AI_QUERY_TIMEOUT_SECONDS) as query_timeout:
1244                response = await engine.provider.ai_query(query, engine_id=engine.id)
1245        except Exception as err:
1246            # expired() tells our own cap apart from a timeout raised inside the engine
1247            if isinstance(err, TimeoutError) and query_timeout.expired():
1248                raise MusicAssistantError(
1249                    f"AI engine '{engine.uid}' did not respond within {AI_QUERY_TIMEOUT_SECONDS}s"
1250                ) from err
1251            error_name = err.__class__.__name__
1252            error_text = str(err).strip()
1253            if error_name == "NotConnected":
1254                raise MusicAssistantError(
1255                    "AI engine "
1256                    f"'{engine.uid}' is not connected. Reconnect the provider "
1257                    "(for example Home Assistant) and retry."
1258                ) from err
1259            details = error_text or error_name
1260            raise MusicAssistantError(f"AI engine '{engine.uid}' query failed: {details}") from err
1261        if not response or not str(response).strip():
1262            raise MusicAssistantError(
1263                f"AI engine '{engine.uid}' returned an empty response for section text"
1264            )
1265        text = str(response).strip()
1266        self.logger.debug(
1267            "AI query response received: engine=%s chars=%d",
1268            engine.uid,
1269            len(text),
1270        )
1271        return text
1272
1273    async def _get_ai_engine(self) -> AIEngine:
1274        """Return the engine used for AI_QUERY tasks, honouring the configured selection."""
1275        selected = cast("str | None", self.get_setup_value(CONF_AI_ENGINE))
1276        if engine := await resolve_ai_engine(self.mass, selected):
1277            return engine
1278        raise MusicAssistantError(
1279            "No AI engine available. Set up a plugin that provides AI (for example Home "
1280            "Assistant with an ai_task entity) and select it in the AI Radio settings."
1281        )
1282
1283    async def _stop_session_queue(self, session: SessionState) -> None:
1284        """Stop playback of the queue a run was playing on."""
1285        queue_id = session.queue_id
1286        if queue_id is None:
1287            return
1288        queue = self.mass.player_queues.get(queue_id)
1289        if queue is None or getattr(queue, "state", None) == PlaybackState.IDLE:
1290            return
1291        try:
1292            await self.mass.player_queues.stop(queue_id)
1293        except MusicAssistantError as err:
1294            self.logger.debug("Could not stop queue %s: %s", queue_id, err)
1295
1296    async def _get_tts_engine(self, engine_uid: str | None = None) -> TTSEngine:
1297        """Return the engine used for TTS tasks, preferring a host-specific engine_uid."""
1298        if engine_uid:
1299            if engine := await resolve_tts_engine(self.mass, engine_uid):
1300                return engine
1301            self.logger.warning(
1302                "Host TTS engine %s is unavailable, falling back to the provider default",
1303                engine_uid,
1304            )
1305        selected = cast("str | None", self.get_setup_value(CONF_TTS_ENGINE))
1306        if engine := await resolve_tts_engine(self.mass, selected):
1307            return engine
1308        raise MusicAssistantError(
1309            "No text-to-speech engine available. Set up a plugin that provides text-to-speech "
1310            "(for example Home Assistant with a TTS entity) and select it in the AI Radio "
1311            "settings."
1312        )
1313