/
/
/
1"""Host (personality) storage/normalization mixin for AI Radio."""
2# mypy: disable-error-code=attr-defined
3
4from __future__ import annotations
5
6import asyncio
7import logging
8from copy import deepcopy
9from dataclasses import dataclass
10from typing import TYPE_CHECKING, Any, Literal
11
12import aiofiles
13from music_assistant_models.errors import InvalidDataError
14
15from music_assistant.helpers.json import async_json_loads, json_dumps
16
17from .constants import DEFAULT_LLM_INSTRUCTIONS, MERGE_SECTION_PROMPT
18from .helpers import slugify
19
20if TYPE_CHECKING:
21 from pathlib import Path
22
23 from music_assistant.mass import MusicAssistant
24
25# placeholders a segment's prompt needs before it may be scheduled, in the same
26# order the frontend's host compiler emits them
27GUARD_PLACEHOLDER_TOKENS = ("<weather_hourly>", "<timestamp>")
28
29
30@dataclass(frozen=True, slots=True)
31class _Plays:
32 """When a preset segment plays; 'value' carries its songs/minutes/percent, if any."""
33
34 kind: Literal["start", "end", "every_song", "every_n_songs", "every_n_min", "occasionally"]
35 value: int = 0
36
37
38@dataclass(frozen=True, slots=True)
39class _PresetSegment:
40 """One spoken segment of a preset host."""
41
42 segment_id: str
43 name: str
44 prompt: str
45 web_search: str
46 max_chars: int
47 plays: _Plays
48
49
50@dataclass(frozen=True, slots=True)
51class _PresetHost:
52 """A bundled host persona, kept in sync with the frontend's host editor presets."""
53
54 host_id: str
55 name: str
56 instructions: str
57 segments: tuple[_PresetSegment, ...]
58
59
60SONG_TRANSITION_PROMPT = (
61 "The previous track was <prev_songinfo> and the next track is <next_songinfo>. "
62 "Create a natural radio transition that connects both songs, sounds informed "
63 "but concise, and avoids filler or repetition."
64)
65
66PRESET_HOSTS: tuple[_PresetHost, ...] = (
67 _PresetHost(
68 host_id="morning_show",
69 name="Morning show",
70 instructions=(
71 "Host personality: warm, energetic, upbeat morning-show host who sounds "
72 "fully awake and glad to be on air. Program instructions: write for spoken "
73 "delivery, keep segments concise, avoid bullet-point phrasing, avoid "
74 "cliches, mention concrete details when available, and maintain a "
75 "believable radio flow between sections."
76 ),
77 segments=(
78 _PresetSegment(
79 segment_id="intro",
80 name="Intro",
81 prompt=(
82 "The next track is <next_songinfo>. Open the morning show like a "
83 "warm, upbeat host: brief good-morning greeting, one concrete hook "
84 "about the song or artist, and a clean handoff into the music."
85 ),
86 web_search="disabled",
87 max_chars=650,
88 plays=_Plays("start"),
89 ),
90 _PresetSegment(
91 segment_id="transition",
92 name="Transition",
93 prompt=(
94 "The previous track was <prev_songinfo> and the next track is "
95 "<next_songinfo>. Create a natural, energetic morning-show "
96 "transition that connects both songs, sounds informed but concise, "
97 "and avoids filler or repetition."
98 ),
99 web_search="allow",
100 max_chars=650,
101 plays=_Plays("every_n_songs", 3),
102 ),
103 _PresetSegment(
104 segment_id="weather",
105 name="Weather",
106 prompt=(
107 "Using <weather_hourly> and <timestamp>, deliver a short spoken "
108 "weather update with the current outlook, a useful next-hours "
109 "summary, and smooth morning-show phrasing."
110 ),
111 web_search="disabled",
112 max_chars=500,
113 plays=_Plays("every_n_min", 60),
114 ),
115 _PresetSegment(
116 segment_id="news",
117 name="News",
118 prompt=(
119 "Create a short global news bulletin anchored to <timestamp>. Use "
120 "web search. Include two or three current items that are broadly "
121 "relevant, clearly separated, fact-focused, and written for spoken "
122 "delivery."
123 ),
124 web_search="force",
125 max_chars=700,
126 plays=_Plays("every_n_min", 60),
127 ),
128 _PresetSegment(
129 segment_id="sign_off",
130 name="Sign-off",
131 prompt=(
132 "The last track played was <prev_songinfo>. Close the morning show "
133 "with a memorable sign-off: brief reflection, warm farewell, and "
134 "language that sounds like the end of a real radio segment."
135 ),
136 web_search="disabled",
137 max_chars=650,
138 plays=_Plays("end"),
139 ),
140 ),
141 ),
142 _PresetHost(
143 host_id="minimal_dj",
144 name="Minimal DJ",
145 instructions=(
146 "Host personality: minimal, calm, understated DJ who lets the music lead. "
147 "Program instructions: keep every segment brief, avoid small talk, avoid "
148 "cliches, and never overshadow the songs with unnecessary commentary."
149 ),
150 segments=(
151 _PresetSegment(
152 segment_id="transition",
153 name="Transition",
154 prompt=(
155 "The previous track was <prev_songinfo> and the next track is "
156 "<next_songinfo>. Give a short, minimal DJ transition: one or two "
157 "sentences, calm tone, no filler, just enough to bridge the songs."
158 ),
159 web_search="disabled",
160 max_chars=300,
161 plays=_Plays("every_n_songs", 3),
162 ),
163 ),
164 ),
165 _PresetHost(
166 host_id="music_nerd",
167 name="Music nerd",
168 instructions=(
169 "Host personality: knowledgeable, enthusiastic music nerd who loves sharing "
170 "context without lecturing. Program instructions: write for spoken delivery, "
171 "keep segments concise, favor concrete facts over generic praise, avoid "
172 "cliches, and maintain a believable radio flow between sections."
173 ),
174 segments=(
175 _PresetSegment(
176 segment_id="intro",
177 name="Intro",
178 prompt=(
179 "The next track is <next_songinfo>. Open the program like a "
180 "knowledgeable music host: brief welcome, one genuinely interesting "
181 "detail about the artist or genre, and a clean handoff into the music."
182 ),
183 web_search="disabled",
184 max_chars=650,
185 plays=_Plays("start"),
186 ),
187 _PresetSegment(
188 segment_id="artist_fact",
189 name="Artist fact",
190 prompt=(
191 "The next track is <next_songinfo>. Share one specific, "
192 "well-researched fact about the artist, the recording, or its "
193 "influence. Keep it precise and avoid generic trivia."
194 ),
195 web_search="allow",
196 max_chars=500,
197 plays=_Plays("every_n_songs", 2),
198 ),
199 _PresetSegment(
200 segment_id="transition",
201 name="Transition",
202 prompt=SONG_TRANSITION_PROMPT,
203 web_search="allow",
204 max_chars=650,
205 plays=_Plays("occasionally", 20),
206 ),
207 ),
208 ),
209 _PresetHost(
210 host_id="party_host",
211 name="Party host",
212 instructions=(
213 "Host personality: high-energy, confident party host who keeps the crowd "
214 "hyped. Program instructions: write for spoken delivery, keep segments "
215 "concise, avoid bullet-point phrasing, avoid cliches, and maintain a "
216 "believable, energetic radio flow between sections."
217 ),
218 segments=(
219 _PresetSegment(
220 segment_id="hype_intro",
221 name="Hype intro",
222 prompt=(
223 "The next track is <next_songinfo>. Open the party like a hype "
224 "radio host: high energy, one confident line about the song or "
225 "artist, and a clean handoff that gets people moving."
226 ),
227 web_search="disabled",
228 max_chars=650,
229 plays=_Plays("start"),
230 ),
231 _PresetSegment(
232 segment_id="shout_out",
233 name="Shout-out",
234 prompt=(
235 "The previous track was <prev_songinfo> and the next track is "
236 "<next_songinfo>. Deliver a high-energy party transition with a "
237 "quick shout-out vibe: keep it fun, confident, and concise, and "
238 "avoid filler or repetition."
239 ),
240 web_search="allow",
241 max_chars=650,
242 plays=_Plays("every_n_songs", 3),
243 ),
244 _PresetSegment(
245 segment_id="sign_off",
246 name="Sign-off",
247 prompt=(
248 "The last track played was <prev_songinfo>. Close the party with a "
249 "memorable, high-energy sign-off: brief hype recap, warm farewell, "
250 "and language that sounds like the end of a real party set."
251 ),
252 web_search="disabled",
253 max_chars=650,
254 plays=_Plays("end"),
255 ),
256 ),
257 ),
258)
259
260
261class AIRadioHostsMixin:
262 """Mixin with host persistence and normalization helpers."""
263
264 if TYPE_CHECKING:
265 mass: MusicAssistant
266 logger: logging.Logger
267 _hosts_file: Path
268 _stations_file: Path
269 _hosts: dict[str, dict[str, Any]]
270 _sections: dict[str, dict[str, Any]]
271
272 async def _load_hosts(self) -> None:
273 """Load host profiles from disk."""
274 hosts_file_exists = await asyncio.to_thread(self._hosts_file.exists)
275 if not hosts_file_exists:
276 self._hosts = {}
277 return
278 async with aiofiles.open(self._hosts_file) as file_handle:
279 content = await file_handle.read()
280 try:
281 payload = await async_json_loads(content)
282 except ValueError as err:
283 # keep the corrupt file on disk for inspection; it is only
284 # overwritten again once a host is saved
285 self.logger.error("Hosts file is corrupt, starting without hosts: %s", err)
286 payload = {}
287 items = payload.get("hosts", []) if isinstance(payload, dict) else []
288 parsed: dict[str, dict[str, Any]] = {}
289 if isinstance(items, list):
290 for item in items:
291 if not isinstance(item, dict):
292 continue
293 try:
294 normalized = self._normalize_host(item)
295 except Exception as err:
296 self.logger.warning("Skipping invalid host profile: %s", err)
297 continue
298 parsed[normalized["id"]] = normalized
299 self._hosts = parsed
300
301 async def _write_hosts(self) -> None:
302 """Persist host profiles to disk."""
303 payload = {
304 "version": 1,
305 "hosts": sorted(self._hosts.values(), key=lambda item: item["name"]),
306 }
307 await self._write_json_file(self._hosts_file, payload)
308
309 async def _seed_preset_hosts(self) -> None:
310 """Seed the bundled preset hosts, but only on a fresh (never configured) install."""
311 for storage_file in (self._hosts_file, self._stations_file):
312 if await asyncio.to_thread(storage_file.exists):
313 return
314 for host, sections in self._default_preset_hosts():
315 for section in sections:
316 self._sections[section["id"]] = section
317 self._hosts[host["id"]] = host
318 await self._write_sections()
319 await self._write_hosts()
320 self.logger.info("Seeded %d preset hosts on this fresh install", len(PRESET_HOSTS))
321
322 def _normalize_host(
323 self,
324 host: dict[str, Any],
325 sections_map: dict[str, dict[str, Any]] | None = None,
326 ) -> dict[str, Any]:
327 """Validate and normalize a host profile."""
328 known_sections = self._sections if sections_map is None else sections_map
329 host_id = slugify(str(host.get("id", "")).strip() or str(host.get("name", "")).strip())
330 name = str(host.get("name", "")).strip()
331 if not name:
332 raise InvalidDataError("Host name is required")
333
334 instructions = str(host.get("instructions") or "").strip() or DEFAULT_LLM_INSTRUCTIONS
335 tts_engine = str(host.get("tts_engine") or "").strip()
336 language = str(host.get("language") or "").strip()
337 options_raw = host.get("options")
338 options = (
339 {str(key): value for key, value in options_raw.items()}
340 if isinstance(options_raw, dict)
341 else {}
342 )
343
344 section_ids: list[str] = []
345 seen: set[str] = set()
346 for item in host.get("section_ids", []):
347 if not isinstance(item, (str, int)):
348 continue
349 section_id = str(item).strip()
350 if not section_id or section_id in seen:
351 continue
352 seen.add(section_id)
353 section_ids.append(section_id)
354 if not section_ids:
355 raise InvalidDataError("Host requires at least one section id")
356 _sections, missing = self._materialize_sections(section_ids, known_sections)
357 if missing:
358 raise InvalidDataError(
359 f"Host references unknown sections: {', '.join(sorted(set(missing)))}"
360 )
361
362 raw_section_order = host.get("section_order")
363 if not isinstance(raw_section_order, list) or not raw_section_order:
364 raise InvalidDataError("Host requires a non-empty 'section_order' list")
365 self._validate_section_order(raw_section_order, set(section_ids))
366
367 merge_section_id = str(host.get("merge_section_id", "")).strip()
368 if merge_section_id:
369 if merge_section_id not in section_ids:
370 raise InvalidDataError("merge_section_id must be selected in host section_ids")
371 merge_section = known_sections.get(merge_section_id)
372 if not merge_section or str(merge_section.get("type", "")).strip().lower() != "ai_meta":
373 raise InvalidDataError("merge_section_id must reference an ai_meta section")
374
375 return {
376 "id": host_id,
377 "name": name,
378 "instructions": instructions,
379 "tts_engine": tts_engine,
380 "language": language,
381 "options": options,
382 "section_ids": section_ids,
383 "section_order": deepcopy(raw_section_order),
384 "merge_section_id": merge_section_id,
385 }
386
387 def _default_host_template(self) -> dict[str, Any]:
388 """Return the built-in host template."""
389 default_sections = self._default_sections_template()
390 default_section_ids = [item["id"] for item in default_sections]
391 return {
392 "id": "default_host",
393 "name": "Default Host",
394 "instructions": DEFAULT_LLM_INSTRUCTIONS,
395 "tts_engine": "",
396 "language": "",
397 "options": {},
398 "section_ids": default_section_ids,
399 "section_order": [
400 {"when": "start_of_playlist", "flow": [{"MUST": "Song_Introduction_Start"}]},
401 {
402 "when": "between_songs",
403 "flow": [
404 {
405 "ALTERNATIVE": {
406 "choices": [
407 {"section": "Song_Transition", "weight": 100},
408 ]
409 }
410 },
411 {
412 "OPTIONAL": {
413 "section": "Weather_Short",
414 "chance": 0.2,
415 "guards": {
416 "min_gap_songs": 3,
417 "max_per_60min": 1,
418 "require_placeholders_present": ["<weather_hourly>"],
419 },
420 }
421 },
422 {
423 "OPTIONAL": {
424 "section": "Global_News",
425 "chance": 0.12,
426 "guards": {
427 "min_gap_songs": 4,
428 "max_per_60min": 1,
429 "require_placeholders_present": ["<timestamp>"],
430 },
431 }
432 },
433 ],
434 },
435 {"when": "end_of_playlist", "flow": [{"MUST": "Song_Introduction_End"}]},
436 ],
437 "merge_section_id": "Between_Songs_Smoother",
438 }
439
440 def _default_preset_hosts(self) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]:
441 """Return the bundled preset hosts, each paired with the sections it references."""
442 return [_compile_preset_host(preset) for preset in PRESET_HOSTS]
443
444 def _migrate_stations_v2_to_v3(self, stations: list[dict[str, Any]]) -> None:
445 """Extract host profiles out of v2 stations and slim the stations in place."""
446 legacy_keys = ("general", "sections", "section_ids", "section_order", "merge_section_id")
447 seen: dict[str, str] = {}
448 for station in stations:
449 try:
450 for item in station.get("sections", []):
451 if not isinstance(item, dict):
452 continue
453 normalized_section = self._normalize_section(item)
454 if normalized_section["id"] not in self._sections:
455 self._sections[normalized_section["id"]] = normalized_section
456
457 general_raw = station.get("general")
458 general = general_raw if isinstance(general_raw, dict) else {}
459 instructions = (
460 str(general.get("instructions") or "").strip() or DEFAULT_LLM_INSTRUCTIONS
461 )
462 section_ids = [
463 str(item).strip()
464 for item in station.get("section_ids", [])
465 if str(item).strip()
466 ]
467 if not section_ids:
468 section_ids = [
469 str(item.get("id", "")).strip()
470 for item in station.get("sections", [])
471 if isinstance(item, dict) and str(item.get("id", "")).strip()
472 ]
473 section_order = station.get("section_order") or []
474 merge_section_id = str(station.get("merge_section_id", "")).strip()
475 fingerprint = json_dumps(
476 {
477 "instructions": instructions,
478 "section_ids": section_ids,
479 "section_order": section_order,
480 "merge_section_id": merge_section_id,
481 }
482 )
483 if fingerprint in seen:
484 host_id = seen[fingerprint]
485 else:
486 host = self._normalize_host(
487 {
488 "id": f"{station.get('name', 'host')}_host",
489 "name": f"{str(station.get('name', 'Host')).strip()} Host",
490 "instructions": instructions,
491 "section_ids": section_ids,
492 "section_order": section_order,
493 "merge_section_id": merge_section_id,
494 }
495 )
496 # a second distinct persona landing on the same slug must not overwrite
497 # the first
498 while host["id"] in self._hosts:
499 host["id"] = f"{host['id']}_{len(self._hosts)}"
500 self._hosts[host["id"]] = host
501 host_id = host["id"]
502 seen[fingerprint] = host_id
503 station["host_id"] = host_id
504 for key in legacy_keys:
505 station.pop(key, None)
506 except Exception as err:
507 self.logger.warning(
508 "Skipping station %s during host migration: %s",
509 station.get("id") or station.get("name") or "<unknown>",
510 err,
511 )
512
513
514def _compile_preset_host(preset: _PresetHost) -> tuple[dict[str, Any], list[dict[str, Any]]]:
515 """Build a host profile plus its sections from a preset, exactly as the frontend does."""
516 sections: list[dict[str, Any]] = []
517 start_flow: list[dict[str, Any]] = []
518 between_flow: list[dict[str, Any]] = []
519 end_flow: list[dict[str, Any]] = []
520 for segment in preset.segments:
521 # host-scoped id: sections live in one flat library, so two hosts must not
522 # collide on e.g. both having an "intro"
523 section_id = slugify(f"{preset.host_id}_{segment.segment_id}")
524 sections.append(
525 {
526 "id": section_id,
527 "name": segment.name,
528 "type": "ai_text",
529 "prompt": segment.prompt,
530 "web_search": segment.web_search,
531 "constraints": {"max_chars": segment.max_chars},
532 }
533 )
534 match segment.plays.kind:
535 case "start":
536 start_flow.append({"MUST": section_id})
537 case "end":
538 end_flow.append({"MUST": section_id})
539 case "every_song":
540 between_flow.append({"MUST": section_id})
541 case _:
542 between_flow.append(_optional_flow_item(section_id, segment))
543
544 section_order: list[dict[str, Any]] = []
545 if start_flow:
546 section_order.append({"when": "start_of_playlist", "flow": start_flow})
547 if between_flow:
548 section_order.append({"when": "between_songs", "flow": between_flow})
549 if end_flow:
550 section_order.append({"when": "end_of_playlist", "flow": end_flow})
551
552 merge_section_id = f"{preset.host_id}_smoother"
553 sections.append(
554 {
555 "id": merge_section_id,
556 "name": "Between Songs Mix",
557 "type": "ai_meta",
558 "prompt": MERGE_SECTION_PROMPT,
559 }
560 )
561 host = {
562 "id": preset.host_id,
563 "name": preset.name,
564 "instructions": preset.instructions,
565 "tts_engine": "",
566 "language": "",
567 "options": {},
568 "section_ids": [section["id"] for section in sections],
569 "section_order": section_order,
570 "merge_section_id": merge_section_id,
571 }
572 return host, sections
573
574
575def _optional_flow_item(section_id: str, segment: _PresetSegment) -> dict[str, Any]:
576 """Translate a recurring segment's cadence into an OPTIONAL flow item with guards."""
577 plays = segment.plays
578 max_per_60min = 0
579 if plays.kind == "every_n_songs":
580 songs = max(1, plays.value)
581 chance = min(1.0, 2 / songs)
582 # the guard allows a section once song_global - last_event >= min_gap_songs, and
583 # consecutive gaps are one song apart, so the cadence is the gap itself
584 min_gap_songs = songs
585 elif plays.kind == "every_n_min":
586 minutes = max(1, plays.value)
587 chance = 1.0
588 min_gap_songs = 0
589 max_per_60min = round(60 / minutes)
590 else:
591 chance = min(100, max(0, plays.value)) / 100
592 min_gap_songs = 1
593 return {
594 "OPTIONAL": {
595 "section": section_id,
596 "chance": chance,
597 "guards": {
598 "min_gap_songs": min_gap_songs,
599 "max_per_60min": max_per_60min,
600 "require_placeholders_present": [
601 token for token in GUARD_PLACEHOLDER_TOKENS if token in segment.prompt
602 ],
603 },
604 }
605 }
606