/
/
/
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
338 section_ids: list[str] = []
339 seen: set[str] = set()
340 for item in host.get("section_ids", []):
341 if not isinstance(item, (str, int)):
342 continue
343 section_id = str(item).strip()
344 if not section_id or section_id in seen:
345 continue
346 seen.add(section_id)
347 section_ids.append(section_id)
348 if not section_ids:
349 raise InvalidDataError("Host requires at least one section id")
350 _sections, missing = self._materialize_sections(section_ids, known_sections)
351 if missing:
352 raise InvalidDataError(
353 f"Host references unknown sections: {', '.join(sorted(set(missing)))}"
354 )
355
356 raw_section_order = host.get("section_order")
357 if not isinstance(raw_section_order, list) or not raw_section_order:
358 raise InvalidDataError("Host requires a non-empty 'section_order' list")
359 self._validate_section_order(raw_section_order, set(section_ids))
360
361 merge_section_id = str(host.get("merge_section_id", "")).strip()
362 if merge_section_id:
363 if merge_section_id not in section_ids:
364 raise InvalidDataError("merge_section_id must be selected in host section_ids")
365 merge_section = known_sections.get(merge_section_id)
366 if not merge_section or str(merge_section.get("type", "")).strip().lower() != "ai_meta":
367 raise InvalidDataError("merge_section_id must reference an ai_meta section")
368
369 return {
370 "id": host_id,
371 "name": name,
372 "instructions": instructions,
373 "tts_engine": tts_engine,
374 "language": language,
375 "section_ids": section_ids,
376 "section_order": deepcopy(raw_section_order),
377 "merge_section_id": merge_section_id,
378 }
379
380 def _default_host_template(self) -> dict[str, Any]:
381 """Return the built-in host template."""
382 default_sections = self._default_sections_template()
383 default_section_ids = [item["id"] for item in default_sections]
384 return {
385 "id": "default_host",
386 "name": "Default Host",
387 "instructions": DEFAULT_LLM_INSTRUCTIONS,
388 "tts_engine": "",
389 "language": "",
390 "section_ids": default_section_ids,
391 "section_order": [
392 {"when": "start_of_playlist", "flow": [{"MUST": "Song_Introduction_Start"}]},
393 {
394 "when": "between_songs",
395 "flow": [
396 {
397 "ALTERNATIVE": {
398 "choices": [
399 {"section": "Song_Transition", "weight": 100},
400 ]
401 }
402 },
403 {
404 "OPTIONAL": {
405 "section": "Weather_Short",
406 "chance": 0.2,
407 "guards": {
408 "min_gap_songs": 3,
409 "max_per_60min": 1,
410 "require_placeholders_present": ["<weather_hourly>"],
411 },
412 }
413 },
414 {
415 "OPTIONAL": {
416 "section": "Global_News",
417 "chance": 0.12,
418 "guards": {
419 "min_gap_songs": 4,
420 "max_per_60min": 1,
421 "require_placeholders_present": ["<timestamp>"],
422 },
423 }
424 },
425 ],
426 },
427 {"when": "end_of_playlist", "flow": [{"MUST": "Song_Introduction_End"}]},
428 ],
429 "merge_section_id": "Between_Songs_Smoother",
430 }
431
432 def _default_preset_hosts(self) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]:
433 """Return the bundled preset hosts, each paired with the sections it references."""
434 return [_compile_preset_host(preset) for preset in PRESET_HOSTS]
435
436 def _migrate_stations_v2_to_v3(self, stations: list[dict[str, Any]]) -> None:
437 """Extract host profiles out of v2 stations and slim the stations in place."""
438 legacy_keys = ("general", "sections", "section_ids", "section_order", "merge_section_id")
439 seen: dict[str, str] = {}
440 for station in stations:
441 try:
442 for item in station.get("sections", []):
443 if not isinstance(item, dict):
444 continue
445 normalized_section = self._normalize_section(item)
446 if normalized_section["id"] not in self._sections:
447 self._sections[normalized_section["id"]] = normalized_section
448
449 general_raw = station.get("general")
450 general = general_raw if isinstance(general_raw, dict) else {}
451 instructions = (
452 str(general.get("instructions") or "").strip() or DEFAULT_LLM_INSTRUCTIONS
453 )
454 section_ids = [
455 str(item).strip()
456 for item in station.get("section_ids", [])
457 if str(item).strip()
458 ]
459 if not section_ids:
460 section_ids = [
461 str(item.get("id", "")).strip()
462 for item in station.get("sections", [])
463 if isinstance(item, dict) and str(item.get("id", "")).strip()
464 ]
465 section_order = station.get("section_order") or []
466 merge_section_id = str(station.get("merge_section_id", "")).strip()
467 fingerprint = json_dumps(
468 {
469 "instructions": instructions,
470 "section_ids": section_ids,
471 "section_order": section_order,
472 "merge_section_id": merge_section_id,
473 }
474 )
475 if fingerprint in seen:
476 host_id = seen[fingerprint]
477 else:
478 host = self._normalize_host(
479 {
480 "id": f"{station.get('name', 'host')}_host",
481 "name": f"{str(station.get('name', 'Host')).strip()} Host",
482 "instructions": instructions,
483 "section_ids": section_ids,
484 "section_order": section_order,
485 "merge_section_id": merge_section_id,
486 }
487 )
488 # a second distinct persona landing on the same slug must not overwrite
489 # the first
490 while host["id"] in self._hosts:
491 host["id"] = f"{host['id']}_{len(self._hosts)}"
492 self._hosts[host["id"]] = host
493 host_id = host["id"]
494 seen[fingerprint] = host_id
495 station["host_id"] = host_id
496 for key in legacy_keys:
497 station.pop(key, None)
498 except Exception as err:
499 self.logger.warning(
500 "Skipping station %s during host migration: %s",
501 station.get("id") or station.get("name") or "<unknown>",
502 err,
503 )
504
505
506def _compile_preset_host(preset: _PresetHost) -> tuple[dict[str, Any], list[dict[str, Any]]]:
507 """Build a host profile plus its sections from a preset, exactly as the frontend does."""
508 sections: list[dict[str, Any]] = []
509 start_flow: list[dict[str, Any]] = []
510 between_flow: list[dict[str, Any]] = []
511 end_flow: list[dict[str, Any]] = []
512 for segment in preset.segments:
513 # host-scoped id: sections live in one flat library, so two hosts must not
514 # collide on e.g. both having an "intro"
515 section_id = slugify(f"{preset.host_id}_{segment.segment_id}")
516 sections.append(
517 {
518 "id": section_id,
519 "name": segment.name,
520 "type": "ai_text",
521 "prompt": segment.prompt,
522 "web_search": segment.web_search,
523 "constraints": {"max_chars": segment.max_chars},
524 }
525 )
526 match segment.plays.kind:
527 case "start":
528 start_flow.append({"MUST": section_id})
529 case "end":
530 end_flow.append({"MUST": section_id})
531 case "every_song":
532 between_flow.append({"MUST": section_id})
533 case _:
534 between_flow.append(_optional_flow_item(section_id, segment))
535
536 section_order: list[dict[str, Any]] = []
537 if start_flow:
538 section_order.append({"when": "start_of_playlist", "flow": start_flow})
539 if between_flow:
540 section_order.append({"when": "between_songs", "flow": between_flow})
541 if end_flow:
542 section_order.append({"when": "end_of_playlist", "flow": end_flow})
543
544 merge_section_id = f"{preset.host_id}_smoother"
545 sections.append(
546 {
547 "id": merge_section_id,
548 "name": "Between Songs Mix",
549 "type": "ai_meta",
550 "prompt": MERGE_SECTION_PROMPT,
551 }
552 )
553 host = {
554 "id": preset.host_id,
555 "name": preset.name,
556 "instructions": preset.instructions,
557 "tts_engine": "",
558 "language": "",
559 "section_ids": [section["id"] for section in sections],
560 "section_order": section_order,
561 "merge_section_id": merge_section_id,
562 }
563 return host, sections
564
565
566def _optional_flow_item(section_id: str, segment: _PresetSegment) -> dict[str, Any]:
567 """Translate a recurring segment's cadence into an OPTIONAL flow item with guards."""
568 plays = segment.plays
569 max_per_60min = 0
570 if plays.kind == "every_n_songs":
571 songs = max(1, plays.value)
572 chance = min(1.0, 2 / songs)
573 # the guard allows a section once song_global - last_event >= min_gap_songs, and
574 # consecutive gaps are one song apart, so the cadence is the gap itself
575 min_gap_songs = songs
576 elif plays.kind == "every_n_min":
577 minutes = max(1, plays.value)
578 chance = 1.0
579 min_gap_songs = 0
580 max_per_60min = round(60 / minutes)
581 else:
582 chance = min(100, max(0, plays.value)) / 100
583 min_gap_songs = 1
584 return {
585 "OPTIONAL": {
586 "section": section_id,
587 "chance": chance,
588 "guards": {
589 "min_gap_songs": min_gap_songs,
590 "max_per_60min": max_per_60min,
591 "require_placeholders_present": [
592 token for token in GUARD_PLACEHOLDER_TOKENS if token in segment.prompt
593 ],
594 },
595 }
596 }
597