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