/
/
/
1"""Streaming operations for Yandex 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 QUALITY_LOSSLESS
13
14if TYPE_CHECKING:
15 from yandex_music import DownloadInfo
16
17 from .provider import YandexMusicProvider
18
19
20class YandexMusicStreamingManager:
21 """Manages Yandex Music streaming operations."""
22
23 def __init__(self, provider: YandexMusicProvider) -> None:
24 """Initialize streaming manager.
25
26 :param provider: The Yandex 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 # Get download info
46 download_infos = await self.client.get_track_download_info(item_id, get_direct_links=True)
47 if not download_infos:
48 raise MediaNotFoundError(f"No stream info available for track {item_id}")
49
50 # Select best quality based on config
51 quality = self.provider.config.get_value("quality")
52 quality_str = str(quality) if quality is not None else None
53 selected_info = self._select_best_quality(download_infos, quality_str)
54
55 if not selected_info or not selected_info.direct_link:
56 raise MediaNotFoundError(f"No stream URL available for track {item_id}")
57
58 # Determine content type
59 content_type = self._get_content_type(selected_info.codec)
60 bitrate = selected_info.bitrate_in_kbps or 0
61
62 return StreamDetails(
63 item_id=item_id,
64 provider=self.provider.instance_id,
65 audio_format=AudioFormat(
66 content_type=content_type,
67 bit_rate=bitrate,
68 ),
69 stream_type=StreamType.HTTP,
70 duration=track.duration,
71 path=selected_info.direct_link,
72 can_seek=True,
73 allow_seek=True,
74 )
75
76 def _select_best_quality(
77 self, download_infos: list[Any], preferred_quality: str | None
78 ) -> DownloadInfo | None:
79 """Select the best quality download info.
80
81 :param download_infos: List of DownloadInfo objects.
82 :param preferred_quality: User's preferred quality setting.
83 :return: Best matching DownloadInfo or None.
84 """
85 if not download_infos:
86 return None
87
88 # Sort by bitrate descending
89 sorted_infos = sorted(
90 download_infos,
91 key=lambda x: x.bitrate_in_kbps or 0,
92 reverse=True,
93 )
94
95 # If user wants lossless, try to find FLAC first
96 if preferred_quality == QUALITY_LOSSLESS:
97 for info in sorted_infos:
98 if info.codec and info.codec.lower() == "flac":
99 return info
100
101 # Return highest bitrate
102 return sorted_infos[0] if sorted_infos else None
103
104 def _get_content_type(self, codec: str | None) -> ContentType:
105 """Determine content type from codec string.
106
107 :param codec: Codec string from Yandex API.
108 :return: ContentType enum value.
109 """
110 if not codec:
111 return ContentType.UNKNOWN
112
113 codec_lower = codec.lower()
114 if codec_lower == "flac":
115 return ContentType.FLAC
116 if codec_lower in ("mp3", "mpeg"):
117 return ContentType.MP3
118 if codec_lower == "aac":
119 return ContentType.AAC
120
121 return ContentType.UNKNOWN
122