/
/
/
1"""Utility helpers for AI Radio."""
2
3from __future__ import annotations
4
5import datetime
6import random
7import re
8from typing import Any
9
10from music_assistant_models.errors import MusicAssistantError
11
12from music_assistant.helpers.datetime import utc
13
14from .constants import EMPTY_SECTION_ID
15from .models import Slot
16
17
18def utc_now_iso() -> str:
19 """Return a UTC ISO timestamp."""
20 return utc().isoformat()
21
22
23def format_ai_radio_timestamp(moment: datetime.datetime) -> str:
24 """Format a moment for the <timestamp> placeholder, spelling out the weekday and month."""
25 # spelled out so the LLM never has to derive the weekday from a numeric date
26 return moment.strftime("%A %d %B %Y, %H:%M %Z")
27
28
29def slugify(value: str) -> str:
30 """Create a slug from arbitrary text."""
31 text = value.strip().lower()
32 text = re.sub(r"[^a-z0-9]+", "_", text)
33 text = text.strip("_")
34 return text or "station"
35
36
37def is_empty_section(section_id: str) -> bool:
38 """Return True when this section acts as a no-op marker."""
39 return section_id.strip().upper() == EMPTY_SECTION_ID
40
41
42def track_songinfo(track: dict[str, Any] | None) -> str:
43 """Return a display string for a track dictionary."""
44 if not track:
45 return ""
46 value = str(track.get("songinfo") or "").strip()
47 if value:
48 return value
49 artist = str(track.get("artist") or "").strip()
50 name = str(track.get("name") or "").strip()
51 return f"{artist} - {name}".strip(" -")
52
53
54def pick_weighted_choice(choices: list[dict[str, Any]], rng: random.Random) -> str:
55 """Pick one ALTERNATIVE section using weighted randomness."""
56 valid: list[tuple[str, float]] = []
57 for choice in choices:
58 section_id = str(choice.get("section", "")).strip()
59 weight = float(choice.get("weight", 1))
60 if section_id and weight > 0:
61 valid.append((section_id, weight))
62 if not valid:
63 raise MusicAssistantError("ALTERNATIVE has no valid section choices")
64 total = sum(weight for _, weight in valid)
65 target = rng.random() * total
66 cursor = 0.0
67 for section_id, weight in valid:
68 cursor += weight
69 if target <= cursor:
70 return section_id
71 return valid[-1][0]
72
73
74def build_slots(tracks: list[dict[str, Any]]) -> list[Slot]:
75 """Build insertion slots from a source track list."""
76 if not tracks:
77 return []
78
79 cumulative_minutes = [0.0]
80 total = 0.0
81 for track in tracks:
82 duration = track.get("duration")
83 seconds = float(duration) if isinstance(duration, (int, float)) and duration > 0 else 210.0
84 total += seconds / 60.0
85 cumulative_minutes.append(total)
86
87 slots: list[Slot] = []
88 slots.append(
89 Slot(
90 when="start_of_playlist",
91 at_index=0,
92 prev_index=None,
93 next_index=0,
94 very_next_index=1 if len(tracks) > 1 else None,
95 minute_mark=0.0,
96 )
97 )
98 for index in range(len(tracks) - 1):
99 slots.append(
100 Slot(
101 when="between_songs",
102 at_index=index + 1,
103 prev_index=index,
104 next_index=index + 1,
105 very_next_index=index + 2 if index + 2 < len(tracks) else None,
106 minute_mark=cumulative_minutes[index + 1],
107 )
108 )
109 slots.append(
110 Slot(
111 when="end_of_playlist",
112 at_index=len(tracks),
113 prev_index=len(tracks) - 1,
114 next_index=None,
115 very_next_index=None,
116 minute_mark=cumulative_minutes[-1],
117 )
118 )
119 return slots
120
121
122def soft_limit_text(text: str, max_chars: int, tolerance_ratio: float = 0.15) -> str:
123 """Trim generated text softly near sentence boundaries."""
124 if max_chars <= 0:
125 return text.strip()
126 slack = max(30, int(max_chars * tolerance_ratio))
127 hard_limit = max_chars + slack
128 cleaned = text.strip()
129 if len(cleaned) <= hard_limit:
130 return cleaned
131
132 candidate = cleaned[:hard_limit].rstrip()
133 sentence_ends = [match.end() for match in re.finditer(r"[.!?](?:\s|$)", candidate)]
134 if sentence_ends:
135 after_target = [index for index in sentence_ends if index >= max_chars]
136 if after_target:
137 return candidate[: after_target[0]].strip()
138 return candidate[: sentence_ends[-1]].strip()
139
140 last_space = candidate.rfind(" ")
141 if last_space > 0:
142 return candidate[:last_space].rstrip()
143 return candidate
144
145
146def coerce_float(value: Any, default: float) -> float:
147 """Convert arbitrary value to float with a safe fallback."""
148 try:
149 return float(value)
150 except TypeError, ValueError:
151 return default
152
153
154def coerce_int(value: Any, default: int) -> int:
155 """Convert arbitrary value to int with a safe fallback."""
156 try:
157 return int(value)
158 except TypeError, ValueError:
159 return default
160