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