/
/
/
1"""Streaming operations for KION Music."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
6
7from music_assistant_models.enums import ContentType, StreamType
8from music_assistant_models.errors import MediaNotFoundError
9from music_assistant_models.media_items import AudioFormat
10from music_assistant_models.streamdetails import StreamDetails
11
12from .constants import CONF_QUALITY, QUALITY_LOSSLESS
13
14if TYPE_CHECKING:
15 from yandex_music import DownloadInfo
16
17 from .provider import KionMusicProvider
18
19
20class KionMusicStreamingManager:
21 """Manages KION Music streaming operations."""
22
23 def __init__(self, provider: KionMusicProvider) -> None:
24 """Initialize streaming manager.
25
26 :param provider: The KION Music provider instance.
27 """
28 self.provider = provider
29 self.client = provider.client
30 self.mass = provider.mass
31 self.logger = provider.logger
32
33 async def get_stream_details(self, item_id: str) -> StreamDetails:
34 """Get stream details for a track.
35
36 :param item_id: Track ID.
37 :return: StreamDetails for the track.
38 :raises MediaNotFoundError: If stream URL cannot be obtained.
39 """
40 # Get track info first
41 track = await self.provider.get_track(item_id)
42 if not track:
43 raise MediaNotFoundError(f"Track {item_id} not found")
44
45 quality = self.provider.config.get_value(CONF_QUALITY)
46 quality_str = str(quality) if quality is not None else None
47 preferred_normalized = (quality_str or "").strip().lower()
48 want_lossless = (
49 QUALITY_LOSSLESS in preferred_normalized or preferred_normalized == QUALITY_LOSSLESS
50 )
51
52 # When user wants lossless, try get-file-info first (FLAC; download-info often MP3 only)
53 if want_lossless:
54 self.logger.debug("Requesting lossless via get-file-info for track %s", item_id)
55 file_info = await self.client.get_track_file_info_lossless(item_id)
56 if file_info:
57 url = file_info.get("url")
58 codec = file_info.get("codec") or ""
59 if url and codec.lower() in ("flac", "flac-mp4"):
60 content_type = self._get_content_type(codec)
61 self.logger.debug(
62 "Stream selected for track %s via get-file-info: codec=%s",
63 item_id,
64 codec,
65 )
66 return StreamDetails(
67 item_id=item_id,
68 provider=self.provider.instance_id,
69 audio_format=AudioFormat(
70 content_type=content_type,
71 bit_rate=0,
72 ),
73 stream_type=StreamType.HTTP,
74 duration=track.duration,
75 path=url,
76 can_seek=True,
77 allow_seek=True,
78 )
79
80 # Default: use /tracks/.../download-info and select best quality
81 download_infos = await self.client.get_track_download_info(item_id, get_direct_links=True)
82 if not download_infos:
83 raise MediaNotFoundError(f"No stream info available for track {item_id}")
84
85 codecs_available = [
86 (getattr(i, "codec", None), getattr(i, "bitrate_in_kbps", None)) for i in download_infos
87 ]
88 self.logger.debug(
89 "Stream quality for track %s: config quality=%s, available codecs=%s",
90 item_id,
91 quality_str,
92 codecs_available,
93 )
94 selected_info = self._select_best_quality(download_infos, quality_str)
95
96 if not selected_info or not selected_info.direct_link:
97 raise MediaNotFoundError(f"No stream URL available for track {item_id}")
98
99 self.logger.debug(
100 "Stream selected for track %s: codec=%s, bitrate=%s",
101 item_id,
102 getattr(selected_info, "codec", None),
103 getattr(selected_info, "bitrate_in_kbps", None),
104 )
105
106 content_type = self._get_content_type(selected_info.codec)
107 bitrate = selected_info.bitrate_in_kbps or 0
108
109 return StreamDetails(
110 item_id=item_id,
111 provider=self.provider.instance_id,
112 audio_format=AudioFormat(
113 content_type=content_type,
114 bit_rate=bitrate,
115 ),
116 stream_type=StreamType.HTTP,
117 duration=track.duration,
118 path=selected_info.direct_link,
119 can_seek=True,
120 allow_seek=True,
121 )
122
123 def _select_best_quality(
124 self, download_infos: list[Any], preferred_quality: str | None
125 ) -> DownloadInfo | None:
126 """Select the best quality download info.
127
128 :param download_infos: List of DownloadInfo objects.
129 :param preferred_quality: User's preferred quality (e.g. "lossless" or "Lossless (FLAC)").
130 :return: Best matching DownloadInfo or None.
131 """
132 if not download_infos:
133 return None
134
135 # Normalize so we accept "lossless", "Lossless (FLAC)", etc.
136 preferred_normalized = (preferred_quality or "").strip().lower()
137 want_lossless = (
138 QUALITY_LOSSLESS in preferred_normalized or preferred_normalized == QUALITY_LOSSLESS
139 )
140
141 # Sort by bitrate descending
142 sorted_infos = sorted(
143 download_infos,
144 key=lambda x: x.bitrate_in_kbps or 0,
145 reverse=True,
146 )
147
148 # If user wants lossless, prefer flac-mp4 then flac (API formats ~2025)
149 if want_lossless:
150 for codec in ("flac-mp4", "flac"):
151 for info in sorted_infos:
152 if info.codec and info.codec.lower() == codec:
153 return info
154 self.logger.warning(
155 "Lossless (FLAC) requested but no FLAC in API response for this "
156 "track; using best available"
157 )
158
159 # Return highest bitrate
160 return sorted_infos[0] if sorted_infos else None
161
162 def _get_content_type(self, codec: str | None) -> ContentType:
163 """Determine content type from codec string.
164
165 :param codec: Codec string from API.
166 :return: ContentType enum value.
167 """
168 if not codec:
169 return ContentType.UNKNOWN
170
171 codec_lower = codec.lower()
172 if codec_lower in ("flac", "flac-mp4"):
173 return ContentType.FLAC
174 if codec_lower in ("mp3", "mpeg"):
175 return ContentType.MP3
176 if codec_lower == "aac":
177 return ContentType.AAC
178
179 return ContentType.UNKNOWN
180