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