/
/
/
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
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 EMPTY_SECTION_ID, MERGE_SECTION_PROMPT, VALID_WEB_SEARCH_MODES
20from .helpers import slugify
21
22_slugify = slugify
23
24if TYPE_CHECKING:
25 from music_assistant_models.config_entries import ProviderConfig
26
27 from music_assistant.mass import MusicAssistant
28
29
30class AIRadioStorageMixin:
31 """Mixin with station/section persistence and normalization helpers."""
32
33 if TYPE_CHECKING:
34 mass: MusicAssistant
35 config: ProviderConfig
36 logger: logging.Logger
37 _sections_file: Path
38 _stations_file: Path
39 _sections: dict[str, dict[str, Any]]
40 _stations: dict[str, dict[str, Any]]
41 _hosts: dict[str, dict[str, Any]]
42
43 async def _load_sections(self) -> None:
44 """Load shared section definitions from disk."""
45 sections_file_exists = await asyncio.to_thread(self._sections_file.exists)
46 if not sections_file_exists:
47 defaults = self._default_sections_template()
48 self._sections = {item["id"]: item for item in defaults}
49 await self._write_sections()
50 return
51 async with aiofiles.open(self._sections_file) as file_handle:
52 content = await file_handle.read()
53 try:
54 payload = await async_json_loads(content)
55 except ValueError as err:
56 self.logger.error("Sections file is corrupt, restoring defaults: %s", err)
57 payload = {}
58 items = payload.get("sections", []) if isinstance(payload, dict) else []
59 parsed: dict[str, dict[str, Any]] = {}
60 if isinstance(items, list):
61 for item in items:
62 if not isinstance(item, dict):
63 continue
64 try:
65 normalized = self._normalize_section(item)
66 except Exception as err:
67 self.logger.warning("Skipping invalid section profile: %s", err)
68 continue
69 parsed[normalized["id"]] = normalized
70 if not parsed:
71 defaults = self._default_sections_template()
72 parsed = {item["id"]: item for item in defaults}
73 self._sections = parsed
74 await self._write_sections()
75 else:
76 self._sections = parsed
77
78 async def _write_sections(self) -> None:
79 """Persist shared section definitions to disk."""
80 payload = {
81 "version": 1,
82 "sections": sorted(self._sections.values(), key=lambda item: item["id"].lower()),
83 }
84 await self._write_json_file(self._sections_file, payload)
85
86 async def _load_stations(self) -> None:
87 """Load station profiles from disk."""
88 stations_file_exists = await asyncio.to_thread(self._stations_file.exists)
89 if not stations_file_exists:
90 self._stations = {}
91 return
92 async with aiofiles.open(self._stations_file) as file_handle:
93 content = await file_handle.read()
94 try:
95 payload = await async_json_loads(content)
96 except ValueError as err:
97 # keep the corrupt file on disk for inspection; it is only
98 # overwritten again once the user saves a station
99 self.logger.error("Stations file is corrupt, starting without stations: %s", err)
100 payload = {}
101 version = payload.get("version", 1) if isinstance(payload, dict) else 1
102 stations = payload.get("stations", []) if isinstance(payload, dict) else []
103 migrated = False
104 sections_changed = False
105 if isinstance(stations, list) and version <= 2 and stations:
106 sections_before = deepcopy(self._sections)
107 self._migrate_stations_v2_to_v3([s for s in stations if isinstance(s, dict)])
108 migrated = True
109 sections_changed = self._sections != sections_before
110 parsed: dict[str, dict[str, Any]] = {}
111 if isinstance(stations, list):
112 for station in stations:
113 if not isinstance(station, dict):
114 continue
115 try:
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 migrated and sections_changed:
123 await self._write_sections()
124 if migrated:
125 await self._write_hosts()
126 await self._write_stations()
127
128 async def _write_stations(self) -> None:
129 """Persist station profiles to disk."""
130 payload = {
131 "version": 3,
132 "stations": sorted(self._stations.values(), key=lambda item: item["name"]),
133 }
134 await self._write_json_file(self._stations_file, payload)
135
136 async def _write_json_file(self, target: Path, payload: dict[str, Any]) -> None:
137 """Write a JSON payload to disk without corrupting the target on failure."""
138 content = await async_json_dumps(payload, indent=True)
139 await asyncio.to_thread(self._sync_write_json_file, target, content)
140
141 def _materialize_sections(
142 self,
143 section_ids: list[str],
144 sections_map: dict[str, dict[str, Any]] | None = None,
145 ) -> tuple[list[dict[str, Any]], list[str]]:
146 """Resolve section ids to shared section objects."""
147 source = self._sections if sections_map is None else sections_map
148 sections: list[dict[str, Any]] = []
149 missing: list[str] = []
150 for section_id in section_ids:
151 section = source.get(section_id)
152 if section is None:
153 missing.append(section_id)
154 continue
155 sections.append(deepcopy(section))
156 return sections, missing
157
158 def _normalize_section(self, section: dict[str, Any]) -> dict[str, Any]:
159 """Validate and normalize a shared section definition."""
160 section_id = str(section.get("id", "")).strip()
161 if not section_id:
162 raise InvalidDataError("Section id is required")
163
164 section_type = str(section.get("type", "ai_text")).strip().lower()
165 if section_type not in {"ai_text", "ai_meta"}:
166 raise InvalidDataError(f"Section '{section_id}' has invalid type '{section_type}'")
167
168 name = str(section.get("name", section_id)).strip() or section_id
169 prompt = str(section.get("prompt", "")).strip()
170 if not prompt:
171 raise InvalidDataError(f"Section '{section_id}' prompt is required")
172
173 normalized: dict[str, Any] = {
174 "id": section_id,
175 "name": name,
176 "type": section_type,
177 "prompt": prompt,
178 }
179
180 if section_type == "ai_text":
181 web_search = str(section.get("web_search", "disabled")).strip().lower()
182 if web_search not in VALID_WEB_SEARCH_MODES:
183 web_search = "disabled"
184 normalized["web_search"] = web_search
185 raw_constraints = section.get("constraints")
186 max_chars = 0
187 if isinstance(raw_constraints, dict):
188 raw_max_chars = raw_constraints.get("max_chars", 0)
189 try:
190 max_chars = max(0, int(raw_max_chars or 0))
191 except (TypeError, ValueError) as err:
192 raise InvalidDataError(
193 f"Section '{section_id}' has non-numeric constraints.max_chars: "
194 f"{raw_max_chars!r}"
195 ) from err
196 if max_chars > 0:
197 normalized["constraints"] = {"max_chars": max_chars}
198 for passthrough_key in ("cover_image",):
199 if passthrough_key in section:
200 normalized[passthrough_key] = section[passthrough_key]
201 return normalized
202
203 def _normalize_station(self, station: dict[str, Any]) -> dict[str, Any]:
204 """Validate and normalize a station profile."""
205 station_id = str(station.get("id", "")).strip() or uuid4().hex[:8]
206 station_id = _slugify(station_id)
207 name = str(station.get("name", "")).strip()
208 if not name:
209 raise InvalidDataError("Station name is required")
210
211 source_playlist_id = str(station.get("source_playlist_id", "")).strip()
212 if not source_playlist_id:
213 raise InvalidDataError("Station source_playlist_id is required")
214 source_playlist_provider = str(station.get("source_playlist_provider", "library")).strip()
215 source_playlist_provider = source_playlist_provider or "library"
216
217 host_id = str(station.get("host_id", "")).strip()
218 if not host_id:
219 raise InvalidDataError("Station host_id is required")
220 if host_id not in self._hosts:
221 raise InvalidDataError(f"Station references unknown host: {host_id}")
222
223 def _require_number(field: str, raw: Any, default: float, cast: type) -> Any:
224 if raw is None or raw == "":
225 return default
226 try:
227 return cast(raw)
228 except (TypeError, ValueError) as err:
229 raise InvalidDataError(
230 f"Station '{name}' field {field!r} must be numeric (got {raw!r})"
231 ) from err
232
233 return {
234 "id": station_id,
235 "name": name,
236 "source_playlist_id": source_playlist_id,
237 "source_playlist_provider": source_playlist_provider,
238 "default_player_id": str(station.get("default_player_id") or ""),
239 "max_duration_minutes": max(
240 0.0,
241 _require_number(
242 "max_duration_minutes", station.get("max_duration_minutes"), 0.0, float
243 ),
244 ),
245 "shuffle_source_tracks": bool(station.get("shuffle_source_tracks", True)),
246 "host_id": host_id,
247 }
248
249 def _validate_section_order(
250 self,
251 section_order: list[Any],
252 valid_section_ids: set[str],
253 ) -> None:
254 """Validate section order references known shared section ids."""
255
256 def _ensure_known(section_id: str) -> None:
257 if section_id == EMPTY_SECTION_ID:
258 return
259 if section_id not in valid_section_ids:
260 raise InvalidDataError(
261 f"section_order references unknown section id '{section_id}'"
262 )
263
264 for rule in section_order:
265 if not isinstance(rule, dict):
266 raise InvalidDataError("Each section_order rule must be an object")
267 flow = rule.get("flow")
268 if not isinstance(flow, list) or not flow:
269 raise InvalidDataError("Each section_order rule requires non-empty flow")
270 for item in flow:
271 if not isinstance(item, dict):
272 raise InvalidDataError("Flow item must be an object")
273 if "MUST" in item:
274 _ensure_known(str(item.get("MUST", "")).strip())
275 continue
276 if "OPTIONAL" in item:
277 optional = item.get("OPTIONAL")
278 if not isinstance(optional, dict):
279 raise InvalidDataError("OPTIONAL flow must be an object")
280 _ensure_known(str(optional.get("section", "")).strip())
281 chance_raw = optional.get("chance", 0)
282 try:
283 float(chance_raw)
284 except (TypeError, ValueError) as err:
285 raise InvalidDataError("OPTIONAL chance must be numeric") from err
286 continue
287 if "ALTERNATIVE" in item:
288 alternative = item.get("ALTERNATIVE")
289 if not isinstance(alternative, dict):
290 raise InvalidDataError("ALTERNATIVE flow must be an object")
291 choices = alternative.get("choices")
292 if not isinstance(choices, list) or not choices:
293 raise InvalidDataError("ALTERNATIVE flow requires non-empty choices")
294 for choice in choices:
295 if not isinstance(choice, dict):
296 raise InvalidDataError("ALTERNATIVE choice must be an object")
297 _ensure_known(str(choice.get("section", "")).strip())
298 weight_raw = choice.get("weight", 1)
299 try:
300 weight = float(weight_raw)
301 except (TypeError, ValueError) as err:
302 raise InvalidDataError(
303 f"ALTERNATIVE choice weight must be numeric (got {weight_raw!r})"
304 ) from err
305 if weight <= 0:
306 raise InvalidDataError("ALTERNATIVE choice weight must be > 0")
307
308 def _default_sections_template(self) -> list[dict[str, Any]]:
309 """Return built-in shared section templates."""
310 return [
311 {
312 "id": "Song_Introduction_Start",
313 "name": "Song Introduction Start",
314 "type": "ai_text",
315 "web_search": "disabled",
316 "prompt": (
317 "The next track is <next_songinfo>. Open the program like a "
318 "polished radio host: brief welcome, confident energy, one "
319 "concrete hook about the song or artist, and a clean handoff "
320 "into the music."
321 ),
322 "constraints": {"max_chars": 650},
323 },
324 {
325 "id": "Song_Transition",
326 "name": "Song Transition",
327 "type": "ai_text",
328 "web_search": "allow",
329 "prompt": (
330 "The previous track was <prev_songinfo> and the next track is "
331 "<next_songinfo>. Create a natural radio transition that "
332 "connects both songs, sounds informed but concise, and avoids "
333 "filler or repetition."
334 ),
335 "constraints": {"max_chars": 650},
336 },
337 {
338 "id": "Global_News",
339 "name": "Global News",
340 "type": "ai_text",
341 "web_search": "force",
342 "prompt": (
343 "Create a short global news bulletin anchored to <timestamp>. "
344 "Use web search. Include two or three current items that are "
345 "broadly relevant, clearly separated, fact-focused, and "
346 "written for spoken delivery."
347 ),
348 "constraints": {"max_chars": 700},
349 },
350 {
351 "id": "Weather_Short",
352 "name": "Weather Short",
353 "type": "ai_text",
354 "web_search": "disabled",
355 "prompt": (
356 "Using <weather_hourly> and <timestamp>, deliver a short "
357 "spoken weather update with the current outlook, a useful "
358 "next-hours summary, and smooth radio phrasing."
359 ),
360 "constraints": {"max_chars": 500},
361 },
362 {
363 "id": "Song_Introduction_End",
364 "name": "Song Introduction End",
365 "type": "ai_text",
366 "web_search": "disabled",
367 "prompt": (
368 "The last track played was <prev_songinfo>. Close the program "
369 "with a memorable sign-off: brief reflection, warm farewell, "
370 "and language that sounds like the end of a real radio segment."
371 ),
372 "constraints": {"max_chars": 650},
373 },
374 {
375 "id": "Between_Songs_Smoother",
376 "name": "Between Songs Mix",
377 "type": "ai_meta",
378 "prompt": MERGE_SECTION_PROMPT,
379 },
380 ]
381
382 def _default_station_template(self) -> dict[str, Any]:
383 """Return the built-in station template."""
384 # must reference a real host so the template validates if saved verbatim; picks the
385 # first sorted host, or default_host on a bare install (the host template creates it)
386 hosts = sorted(self._hosts.values(), key=lambda item: item["name"])
387 return {
388 "id": "example_station",
389 "name": "Example AI Radio Station",
390 "source_playlist_id": "",
391 "source_playlist_provider": "library",
392 "default_player_id": "",
393 "max_duration_minutes": 0,
394 "shuffle_source_tracks": True,
395 "host_id": str(hosts[0]["id"]) if hosts else "default_host",
396 }
397
398 def _sync_write_json_file(self, target: Path, content: str) -> None:
399 """Atomically write JSON content to disk, fsyncing before the rename."""
400 tmp_file = target.with_name(f"{target.name}.tmp")
401 with tmp_file.open("w", encoding="utf-8") as file_handle:
402 file_handle.write(content)
403 file_handle.flush()
404 os.fsync(file_handle.fileno())
405 tmp_file.replace(target)
406