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