/
/
/
1"""Data models for AI Radio."""
2
3from __future__ import annotations
4
5import asyncio
6from dataclasses import dataclass, field
7from typing import Any
8
9from music_assistant.helpers.datetime import utc
10
11
12@dataclass(slots=True)
13class Slot:
14 """Insertion slot between source tracks."""
15
16 when: str
17 at_index: int
18 prev_index: int | None
19 next_index: int | None
20 very_next_index: int | None
21 minute_mark: float
22
23
24@dataclass(slots=True)
25class PlannedSection:
26 """A section that should be generated for a run."""
27
28 order: int
29 clip_id: str
30 section_id: str
31 section_name: str
32 when: str
33 insert_at_index: int
34 prompt: str
35 max_chars: int
36 web_search_mode: str
37 # when true, a failed weather fetch skips the clip instead of airing it without a forecast
38 weather_required: bool = False
39 # the guard history events this plan claimed, as (section_id, (song, minute)). a caller
40 # that drops the plan can drop these too, so a clip that never aired carries no weight
41 history_events: list[tuple[str, tuple[int, float]]] = field(default_factory=list)
42
43
44@dataclass(slots=True)
45class DJQueueState:
46 """State container for one sticky queue DJ."""
47
48 queue_id: str
49 host_id: str
50 dj_session_id: str
51 clip_counter: int = 0
52 songs_before_window: int = 0
53 minutes_before_window: float = 0.0
54 # queue_item_ids of the tracks whose preceding gap this session already settled, by
55 # injecting a clip, by leaving it empty on purpose or because it became unusable
56 decided_gap_ids: set[str] = field(default_factory=set)
57 history: dict[str, list[tuple[int, float]]] = field(default_factory=dict)
58 # a freshly armed state may only plan once the previous host's clips are cleared
59 ready: bool = False
60 replan_pending: bool = False
61 lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False, compare=False)
62 task: asyncio.Task[Any] | None = field(default=None, repr=False, compare=False)
63
64
65@dataclass(slots=True)
66class SessionState:
67 """State container for an AI Radio run."""
68
69 session_id: str
70 station_id: str
71 status: str = "running"
72 created_at: str = field(default_factory=lambda: utc().isoformat())
73 started_at: str | None = None
74 ended_at: str | None = None
75 progress: dict[str, Any] = field(default_factory=dict)
76 result: dict[str, Any] = field(default_factory=dict)
77 error: str | None = None
78 skipped_sections: int = 0
79 last_render_error: str | None = None
80 task: asyncio.Task[Any] | None = field(default=None, repr=False, compare=False)
81 queue_id: str | None = field(default=None, repr=False, compare=False)
82
83 def as_dict(self) -> dict[str, Any]:
84 """Return session as a serializable dictionary."""
85 return {
86 "session_id": self.session_id,
87 "station_id": self.station_id,
88 "queue_id": self.queue_id,
89 "status": self.status,
90 "created_at": self.created_at,
91 "started_at": self.started_at,
92 "ended_at": self.ended_at,
93 "progress": self.progress,
94 "result": self.result,
95 "error": self.error,
96 "skipped_sections": self.skipped_sections,
97 "last_render_error": self.last_render_error,
98 }
99