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