/
/
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 # the guard history events this plan claimed, as (section_id, (song, minute)). a caller
38 # that drops the plan can drop these too, so a clip that never aired carries no weight
39 history_events: list[tuple[str, tuple[int, float]]] = field(default_factory=list)
40
41
42@dataclass(slots=True)
43class DJQueueState:
44 """State container for one sticky queue DJ."""
45
46 queue_id: str
47 host_id: str
48 dj_session_id: str
49 clip_counter: int = 0
50 songs_before_window: int = 0
51 minutes_before_window: float = 0.0
52 # queue_item_ids of the tracks whose preceding gap this session already settled, by
53 # injecting a clip, by leaving it empty on purpose or because it became unusable
54 decided_gap_ids: set[str] = field(default_factory=set)
55 history: dict[str, list[tuple[int, float]]] = field(default_factory=dict)
56 # a freshly armed state may only plan once the previous host's clips are cleared
57 ready: bool = False
58 replan_pending: bool = False
59 lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False, compare=False)
60 task: asyncio.Task[Any] | None = field(default=None, repr=False, compare=False)
61
62
63@dataclass(slots=True)
64class SessionState:
65 """State container for an AI Radio run."""
66
67 session_id: str
68 station_id: str
69 status: str = "running"
70 created_at: str = field(default_factory=lambda: utc().isoformat())
71 started_at: str | None = None
72 ended_at: str | None = None
73 progress: dict[str, Any] = field(default_factory=dict)
74 result: dict[str, Any] = field(default_factory=dict)
75 error: str | None = None
76 skipped_sections: int = 0
77 last_render_error: str | None = None
78 task: asyncio.Task[Any] | None = field(default=None, repr=False, compare=False)
79 queue_id: str | None = field(default=None, repr=False, compare=False)
80
81 def as_dict(self) -> dict[str, Any]:
82 """Return session as a serializable dictionary."""
83 return {
84 "session_id": self.session_id,
85 "station_id": self.station_id,
86 "queue_id": self.queue_id,
87 "status": self.status,
88 "created_at": self.created_at,
89 "started_at": self.started_at,
90 "ended_at": self.ended_at,
91 "progress": self.progress,
92 "result": self.result,
93 "error": self.error,
94 "skipped_sections": self.skipped_sections,
95 "last_render_error": self.last_render_error,
96 }
97