/
/
/
1"""Station and section storage/normalization mixin for AI Radio."""
2# mypy: disable-error-code=attr-defined
3
4from __future__ import annotations
5
6import asyncio
7import logging
8import os
9from copy import deepcopy
10from pathlib import Path
11from typing import TYPE_CHECKING, Any, cast
12from uuid import uuid4
13
14import aiofiles
15from music_assistant_models.errors import InvalidDataError
16
17from music_assistant.helpers.json import async_json_dumps, async_json_loads
18
19from .constants import (
20 DEFAULT_LLM_INSTRUCTIONS,
21 DEFAULT_WEATHER_PROVIDER,
22 DEFAULT_WEATHER_TIMEOUT_SECONDS,
23 EMPTY_SECTION_ID,
24 VALID_WEB_SEARCH_MODES,
25)
26from .helpers import slugify
27
28_slugify = slugify
29
30if TYPE_CHECKING:
31 from music_assistant_models.config_entries import ProviderConfig
32
33 from music_assistant.mass import MusicAssistant
34
35
36class AIRadioStorageMixin:
37 """Mixin with station/section persistence and normalization helpers."""
38
39 if TYPE_CHECKING:
40 mass: MusicAssistant
41 config: ProviderConfig
42 logger: logging.Logger
43 _sections_file: Path
44 _stations_file: Path
45 _sections: dict[str, dict[str, Any]]
46 _stations: dict[str, dict[str, Any]]
47
48 async def _load_sections(self) -> None:
49 """Load shared section definitions from disk."""
50 sections_file_exists = await asyncio.to_thread(self._sections_file.exists)
51 if not sections_file_exists:
52 defaults = self._default_sections_template()
53 self._sections = {item["id"]: item for item in defaults}
54 await self._write_sections()
55 return
56 async with aiofiles.open(self._sections_file) as file_handle:
57 content = await file_handle.read()
58 try:
59 payload = await async_json_loads(content)
60 except ValueError as err:
61 self.logger.error("Sections file is corrupt, restoring defaults: %s", err)
62 payload = {}
63 items = payload.get("sections", []) if isinstance(payload, dict) else []
64 parsed: dict[str, dict[str, Any]] = {}
65 if isinstance(items, list):
66 for item in items:
67 if not isinstance(item, dict):
68 continue
69 try:
70 normalized = self._normalize_section(item)
71 except Exception as err:
72 self.logger.warning("Skipping invalid section profile: %s", err)
73 continue
74 parsed[normalized["id"]] = normalized
75 if not parsed:
76 defaults = self._default_sections_template()
77 parsed = {item["id"]: item for item in defaults}
78 self._sections = parsed
79 await self._write_sections()
80 else:
81 self._sections = parsed
82
83 async def _write_sections(self) -> None:
84 """Persist shared section definitions to disk."""
85 payload = {
86 "version": 1,
87 "sections": sorted(self._sections.values(), key=lambda item: item["id"].lower()),
88 }
89 await self._write_json_file(self._sections_file, payload)
90
91 async def _load_stations(self) -> None:
92 """Load station profiles from disk."""
93 stations_file_exists = await asyncio.to_thread(self._stations_file.exists)
94 if not stations_file_exists:
95 self._stations = {}
96 return
97 async with aiofiles.open(self._stations_file) as file_handle:
98 content = await file_handle.read()
99 try:
100 payload = await async_json_loads(content)
101 except ValueError as err:
102 # keep the corrupt file on disk for inspection; it is only
103 # overwritten again once the user saves a station
104 self.logger.error("Stations file is corrupt, starting without stations: %s", err)
105 payload = {}
106 stations = payload.get("stations", []) if isinstance(payload, dict) else []
107 parsed: dict[str, dict[str, Any]] = {}
108 sections_changed = False
109 if isinstance(stations, list):
110 for station in stations:
111 if not isinstance(station, dict):
112 continue
113 try:
114 if self._upsert_embedded_sections_from_station(station):
115 sections_changed = True
116 normalized = self._normalize_station(station)
117 except Exception as err:
118 self.logger.warning("Skipping invalid station profile: %s", err)
119 continue
120 parsed[normalized["id"]] = normalized
121 self._stations = parsed
122 if sections_changed:
123 await self._write_sections()
124
125 async def _write_stations(self) -> None:
126 """Persist station profiles to disk."""
127 payload = {
128 "version": 2,
129 "stations": sorted(self._stations.values(), key=lambda item: item["name"]),
130 }
131 await self._write_json_file(self._stations_file, payload)
132
133 async def _write_json_file(self, target: Path, payload: dict[str, Any]) -> None:
134 """Write a JSON payload to disk without corrupting the target on failure."""
135 content = await async_json_dumps(payload, indent=True)
136 await asyncio.to_thread(self._sync_write_json_file, target, content)
137
138 def _materialize_sections(
139 self,
140 section_ids: list[str],
141 sections_map: dict[str, dict[str, Any]] | None = None,
142 ) -> tuple[list[dict[str, Any]], list[str]]:
143 """Resolve section ids to shared section objects."""
144 source = self._sections if sections_map is None else sections_map
145 sections: list[dict[str, Any]] = []
146 missing: list[str] = []
147 for section_id in section_ids:
148 section = source.get(section_id)
149 if section is None:
150 missing.append(section_id)
151 continue
152 sections.append(deepcopy(section))
153 return sections, missing
154
155 def _refresh_station_sections(self) -> None:
156 """Rebuild embedded station sections from selected shared ids."""
157 for station in self._stations.values():
158 section_ids = list(station.get("section_ids", []))
159 sections, _missing = self._materialize_sections(section_ids)
160 station["sections"] = sections
161
162 def _normalize_section(self, section: dict[str, Any]) -> dict[str, Any]:
163 """Validate and normalize a shared section definition."""
164 section_id = str(section.get("id", "")).strip()
165 if not section_id:
166 raise InvalidDataError("Section id is required")
167
168 section_type = str(section.get("type", "ai_text")).strip().lower()
169 if section_type not in {"ai_text", "ai_meta"}:
170 raise InvalidDataError(f"Section '{section_id}' has invalid type '{section_type}'")
171
172 name = str(section.get("name", section_id)).strip() or section_id
173 prompt = str(section.get("prompt", "")).strip()
174 if not prompt:
175 raise InvalidDataError(f"Section '{section_id}' prompt is required")
176
177 normalized: dict[str, Any] = {
178 "id": section_id,
179 "name": name,
180 "type": section_type,
181 "prompt": prompt,
182 }
183
184 if section_type == "ai_text":
185 web_search = str(section.get("web_search", "disabled")).strip().lower()
186 if web_search not in VALID_WEB_SEARCH_MODES:
187 web_search = "disabled"
188 normalized["web_search"] = web_search
189 raw_constraints = section.get("constraints")
190 max_chars = 0
191 if isinstance(raw_constraints, dict):
192 raw_max_chars = raw_constraints.get("max_chars", 0)
193 try:
194 max_chars = max(0, int(raw_max_chars or 0))
195 except (TypeError, ValueError) as err:
196 raise InvalidDataError(
197 f"Section '{section_id}' has non-numeric constraints.max_chars: "
198 f"{raw_max_chars!r}"
199 ) from err
200 if max_chars > 0:
201 normalized["constraints"] = {"max_chars": max_chars}
202 for passthrough_key in ("cover_image",):
203 if passthrough_key in section:
204 normalized[passthrough_key] = section[passthrough_key]
205 return normalized
206
207 def _normalize_general(self, source_general: Any) -> dict[str, Any]:
208 """Normalize station general settings to a known schema."""
209 defaults = cast("dict[str, Any]", deepcopy(self._default_station_template()["general"]))
210 if not isinstance(source_general, dict):
211 return defaults
212
213 def _text(key: str) -> str:
214 value = source_general.get(key, defaults[key])
215 return str(value or defaults[key]).strip() or str(defaults[key])
216
217 def _int(key: str) -> int:
218 value = source_general.get(key, defaults[key])
219 try:
220 return int(value)
221 except TypeError, ValueError:
222 return int(defaults[key])
223
224 return {
225 "instructions": str(source_general.get("instructions") or defaults["instructions"]),
226 "weather_provider": _text("weather_provider"),
227 "weather_timeout_seconds": _int("weather_timeout_seconds"),
228 }
229
230 def _upsert_embedded_sections_from_station(
231 self,
232 station: dict[str, Any],
233 sections_map: dict[str, dict[str, Any]] | None = None,
234 ) -> bool:
235 """Upsert embedded station sections and return whether shared sections changed."""
236 target = self._sections if sections_map is None else sections_map
237 changed = False
238 raw_sections = station.get("sections")
239 if not isinstance(raw_sections, list):
240 return changed
241
242 ids_from_embedded: list[str] = []
243 for item in raw_sections:
244 if not isinstance(item, dict):
245 continue
246 normalized = self._normalize_section(item)
247 ids_from_embedded.append(normalized["id"])
248 existing = target.get(normalized["id"])
249 if existing != normalized:
250 target[normalized["id"]] = normalized
251 changed = True
252
253 raw_ids = station.get("section_ids")
254 if not isinstance(raw_ids, list) or not raw_ids:
255 station["section_ids"] = ids_from_embedded
256 return changed
257
258 def _normalize_station(
259 self,
260 station: dict[str, Any],
261 sections_map: dict[str, dict[str, Any]] | None = None,
262 ) -> dict[str, Any]:
263 """Validate and normalize a station profile."""
264 known_sections = self._sections if sections_map is None else sections_map
265 station_id = str(station.get("id", "")).strip() or uuid4().hex[:8]
266 station_id = _slugify(station_id)
267 name = str(station.get("name", "")).strip()
268 if not name:
269 raise InvalidDataError("Station name is required")
270
271 source_playlist_id = str(station.get("source_playlist_id", "")).strip()
272 if not source_playlist_id:
273 raise InvalidDataError("Station source_playlist_id is required")
274 source_playlist_provider = str(station.get("source_playlist_provider", "library")).strip()
275 source_playlist_provider = source_playlist_provider or "library"
276
277 raw_section_order = station.get("section_order")
278 if not isinstance(raw_section_order, list) or not raw_section_order:
279 raise InvalidDataError("Station requires a non-empty 'section_order' list")
280
281 section_ids = self._collect_station_section_ids(station)
282 if not section_ids:
283 raise InvalidDataError("Station requires at least one section id")
284
285 sections, missing = self._materialize_sections(section_ids, known_sections)
286 if missing:
287 raise InvalidDataError(
288 f"Station references unknown sections: {', '.join(sorted(set(missing)))}"
289 )
290
291 self._validate_section_order(raw_section_order, set(section_ids))
292
293 general = self._normalize_general(station.get("general"))
294 merge_section_id = str(station.get("merge_section_id", "")).strip()
295 if merge_section_id:
296 if merge_section_id not in section_ids:
297 raise InvalidDataError("merge_section_id must be selected in station section_ids")
298 merge_section = known_sections.get(merge_section_id)
299 if not merge_section or str(merge_section.get("type", "")).strip().lower() != "ai_meta":
300 raise InvalidDataError("merge_section_id must reference an ai_meta section")
301
302 def _require_number(field: str, raw: Any, default: float, cast: type) -> Any:
303 if raw is None or raw == "":
304 return default
305 try:
306 return cast(raw)
307 except (TypeError, ValueError) as err:
308 raise InvalidDataError(
309 f"Station '{name}' field {field!r} must be numeric (got {raw!r})"
310 ) from err
311
312 return {
313 "id": station_id,
314 "name": name,
315 "source_playlist_id": source_playlist_id,
316 "source_playlist_provider": source_playlist_provider,
317 "default_player_id": str(station.get("default_player_id") or ""),
318 "max_duration_minutes": max(
319 0.0,
320 _require_number(
321 "max_duration_minutes", station.get("max_duration_minutes"), 0.0, float
322 ),
323 ),
324 "shuffle_source_tracks": bool(station.get("shuffle_source_tracks", True)),
325 "merge_section_id": merge_section_id,
326 "general": general,
327 "section_ids": section_ids,
328 "sections": sections,
329 "section_order": deepcopy(raw_section_order),
330 }
331
332 def _collect_station_section_ids(self, station: dict[str, Any]) -> list[str]:
333 """Collect station section ids from section_ids or legacy embedded sections."""
334 ids: list[str] = []
335 raw_ids = station.get("section_ids")
336 if isinstance(raw_ids, list):
337 ids.extend(str(item).strip() for item in raw_ids if str(item).strip())
338
339 if not ids:
340 raw_sections = station.get("sections")
341 if isinstance(raw_sections, list):
342 for section in raw_sections:
343 if not isinstance(section, dict):
344 continue
345 section_id = str(section.get("id", "")).strip()
346 if section_id:
347 ids.append(section_id)
348
349 deduped: list[str] = []
350 seen: set[str] = set()
351 for section_id in ids:
352 if section_id in seen:
353 continue
354 seen.add(section_id)
355 deduped.append(section_id)
356 return deduped
357
358 def _validate_section_order(
359 self,
360 section_order: list[Any],
361 valid_section_ids: set[str],
362 ) -> None:
363 """Validate section order references known shared section ids."""
364
365 def _ensure_known(section_id: str) -> None:
366 if section_id == EMPTY_SECTION_ID:
367 return
368 if section_id not in valid_section_ids:
369 raise InvalidDataError(
370 f"section_order references unknown section id '{section_id}'"
371 )
372
373 for rule in section_order:
374 if not isinstance(rule, dict):
375 raise InvalidDataError("Each section_order rule must be an object")
376 flow = rule.get("flow")
377 if not isinstance(flow, list) or not flow:
378 raise InvalidDataError("Each section_order rule requires non-empty flow")
379 for item in flow:
380 if not isinstance(item, dict):
381 raise InvalidDataError("Flow item must be an object")
382 if "MUST" in item:
383 _ensure_known(str(item.get("MUST", "")).strip())
384 continue
385 if "OPTIONAL" in item:
386 optional = item.get("OPTIONAL")
387 if not isinstance(optional, dict):
388 raise InvalidDataError("OPTIONAL flow must be an object")
389 _ensure_known(str(optional.get("section", "")).strip())
390 chance_raw = optional.get("chance", 0)
391 try:
392 float(chance_raw)
393 except (TypeError, ValueError) as err:
394 raise InvalidDataError("OPTIONAL chance must be numeric") from err
395 continue
396 if "ALTERNATIVE" in item:
397 alternative = item.get("ALTERNATIVE")
398 if not isinstance(alternative, dict):
399 raise InvalidDataError("ALTERNATIVE flow must be an object")
400 choices = alternative.get("choices")
401 if not isinstance(choices, list) or not choices:
402 raise InvalidDataError("ALTERNATIVE flow requires non-empty choices")
403 for choice in choices:
404 if not isinstance(choice, dict):
405 raise InvalidDataError("ALTERNATIVE choice must be an object")
406 _ensure_known(str(choice.get("section", "")).strip())
407 weight_raw = choice.get("weight", 1)
408 try:
409 weight = float(weight_raw)
410 except (TypeError, ValueError) as err:
411 raise InvalidDataError(
412 f"ALTERNATIVE choice weight must be numeric (got {weight_raw!r})"
413 ) from err
414 if weight <= 0:
415 raise InvalidDataError("ALTERNATIVE choice weight must be > 0")
416
417 def _default_sections_template(self) -> list[dict[str, Any]]:
418 """Return built-in shared section templates."""
419 return [
420 {
421 "id": "Song_Introduction_Start",
422 "name": "Song Introduction Start",
423 "type": "ai_text",
424 "web_search": "disabled",
425 "prompt": (
426 "The next track is <next_songinfo>. Open the program like a "
427 "polished radio host: brief welcome, confident energy, one "
428 "concrete hook about the song or artist, and a clean handoff "
429 "into the music."
430 ),
431 "constraints": {"max_chars": 650},
432 },
433 {
434 "id": "Song_Transition",
435 "name": "Song Transition",
436 "type": "ai_text",
437 "web_search": "allow",
438 "prompt": (
439 "The previous track was <prev_songinfo> and the next track is "
440 "<next_songinfo>. Create a natural radio transition that "
441 "connects both songs, sounds informed but concise, and avoids "
442 "filler or repetition."
443 ),
444 "constraints": {"max_chars": 650},
445 },
446 {
447 "id": "Global_News",
448 "name": "Global News",
449 "type": "ai_text",
450 "web_search": "force",
451 "prompt": (
452 "Create a short global news bulletin anchored to <timestamp>. "
453 "Use web search. Include two or three current items that are "
454 "broadly relevant, clearly separated, fact-focused, and "
455 "written for spoken delivery."
456 ),
457 "constraints": {"max_chars": 700},
458 },
459 {
460 "id": "Weather_Short",
461 "name": "Weather Short",
462 "type": "ai_text",
463 "web_search": "disabled",
464 "prompt": (
465 "Using <weather_hourly> and <timestamp>, deliver a short "
466 "spoken weather update with the current outlook, a useful "
467 "next-hours summary, and smooth radio phrasing."
468 ),
469 "constraints": {"max_chars": 500},
470 },
471 {
472 "id": "Song_Introduction_End",
473 "name": "Song Introduction End",
474 "type": "ai_text",
475 "web_search": "disabled",
476 "prompt": (
477 "The last track played was <prev_songinfo>. Close the program "
478 "with a memorable sign-off: brief reflection, warm farewell, "
479 "and language that sounds like the end of a real radio segment."
480 ),
481 "constraints": {"max_chars": 650},
482 },
483 {
484 "id": "Between_Songs_Smoother",
485 "name": "Between Songs Mix",
486 "type": "ai_meta",
487 "prompt": (
488 "Merge the drafts below into one coherent radio break. "
489 "Preserve factual content, remove duplication, and make the "
490 "final segment sound like one host speaking naturally.\n"
491 "<section_drafts>"
492 ),
493 },
494 ]
495
496 def _default_station_template(self) -> dict[str, Any]:
497 """Return the built-in station template."""
498 default_sections = self._default_sections_template()
499 default_section_ids = [item["id"] for item in default_sections]
500 return {
501 "id": "example_station",
502 "name": "Example AI Radio Station",
503 "source_playlist_id": "",
504 "source_playlist_provider": "library",
505 "default_player_id": "",
506 "max_duration_minutes": 0,
507 "shuffle_source_tracks": True,
508 "merge_section_id": "Between_Songs_Smoother",
509 "general": {
510 "instructions": DEFAULT_LLM_INSTRUCTIONS,
511 "weather_provider": DEFAULT_WEATHER_PROVIDER,
512 "weather_timeout_seconds": DEFAULT_WEATHER_TIMEOUT_SECONDS,
513 },
514 "section_ids": default_section_ids,
515 "sections": deepcopy(default_sections),
516 "section_order": [
517 {"when": "start_of_playlist", "flow": [{"MUST": "Song_Introduction_Start"}]},
518 {
519 "when": "between_songs",
520 "flow": [
521 {
522 "ALTERNATIVE": {
523 "choices": [
524 {"section": "Song_Transition", "weight": 100},
525 ]
526 }
527 },
528 {
529 "OPTIONAL": {
530 "section": "Weather_Short",
531 "chance": 0.2,
532 "guards": {
533 "min_gap_songs": 3,
534 "max_per_60min": 1,
535 "require_placeholders_present": ["<weather_hourly>"],
536 },
537 }
538 },
539 {
540 "OPTIONAL": {
541 "section": "Global_News",
542 "chance": 0.12,
543 "guards": {
544 "min_gap_songs": 4,
545 "max_per_60min": 1,
546 "require_placeholders_present": ["<timestamp>"],
547 },
548 }
549 },
550 ],
551 },
552 {"when": "end_of_playlist", "flow": [{"MUST": "Song_Introduction_End"}]},
553 ],
554 }
555
556 def _sync_write_json_file(self, target: Path, content: str) -> None:
557 """Atomically write JSON content to disk, fsyncing before the rename."""
558 tmp_file = target.with_name(f"{target.name}.tmp")
559 with tmp_file.open("w", encoding="utf-8") as file_handle:
560 file_handle.write(content)
561 file_handle.flush()
562 os.fsync(file_handle.fileno())
563 tmp_file.replace(target)
564