music-assistant-server

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