/
/
/
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
38
39@dataclass(slots=True)
40class SessionState:
41 """State container for an AI Radio run."""
42
43 session_id: str
44 station_id: str
45 status: str = "running"
46 created_at: str = field(default_factory=lambda: utc().isoformat())
47 started_at: str | None = None
48 ended_at: str | None = None
49 progress: dict[str, Any] = field(default_factory=dict)
50 result: dict[str, Any] = field(default_factory=dict)
51 error: str | None = None
52 skipped_sections: int = 0
53 last_render_error: str | None = None
54 task: asyncio.Task[Any] | None = field(default=None, repr=False, compare=False)
55 queue_id: str | None = field(default=None, repr=False, compare=False)
56
57 def as_dict(self) -> dict[str, Any]:
58 """Return session as a serializable dictionary."""
59 return {
60 "session_id": self.session_id,
61 "station_id": self.station_id,
62 "status": self.status,
63 "created_at": self.created_at,
64 "started_at": self.started_at,
65 "ended_at": self.ended_at,
66 "progress": self.progress,
67 "result": self.result,
68 "error": self.error,
69 "skipped_sections": self.skipped_sections,
70 "last_render_error": self.last_render_error,
71 }
72