/
/
/
1"""Streaming operations for KION Music."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncGenerator
7from typing import TYPE_CHECKING, Any
8
9import aiohttp
10from aiohttp import ClientPayloadError, ServerDisconnectedError
11from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
12from music_assistant_models.enums import ContentType, StreamType
13from music_assistant_models.errors import MediaNotFoundError
14from music_assistant_models.media_items import AudioFormat
15from music_assistant_models.streamdetails import StreamDetails
16
17from music_assistant.helpers.throttle_retry import BYPASS_THROTTLER
18
19from .constants import (
20 CONF_CODECS,
21 CONF_QUALITY,
22 CONF_TRANSPORT,
23 QUALITY_BALANCED,
24 QUALITY_EFFICIENT,
25 QUALITY_FILE_INFO_PARAMS,
26 QUALITY_HIGH,
27 QUALITY_LOSSLESS,
28 RADIO_TRACK_ID_SEP,
29 TRANSPORT_RAW,
30)
31
32if TYPE_CHECKING:
33 from yandex_music import DownloadInfo
34
35 from .provider import KionMusicProvider
36
37
38# Windowed-stream tuning constants
39_CHUNK_SIZE = 16384 # smaller than default 65536 for faster first-byte after retry
40_STREAM_TIMEOUT = aiohttp.ClientTimeout(total=None, sock_read=30)
41# Kion CDN drops TCP connections for slow consumers (observed at ~45s for raw transport
42# at real-time playback rate ~200 KB/s). By capping each Range request to 4 MB we download
43# each window quickly, preventing CDN drops for both raw and encrypted transports.
44_RANGE_WINDOW = 4 * 1024 * 1024 # 4 MB per Range request
45# AES-CTR block size in bytes (used for block-aligned Range requests in encrypted transport)
46_AES_BLOCK_SIZE = 16
47# Flat short delays for TCP drops (network glitches within a 4 MB window)
48_TCP_DROP_DELAYS = (0.5, 1.0, 2.0)
49# Exponential delays for true network stalls (read timeout)
50_STALL_DELAYS = (2.0, 4.0, 8.0)
51
52
53class KionMusicStreamingManager:
54 """Manages KION Music streaming operations."""
55
56 def __init__(self, provider: KionMusicProvider) -> None:
57 """
58 Initialize streaming manager.
59
60 :param provider: The KION Music provider instance.
61 """
62 self.provider = provider
63 self.client = provider.client
64 self.mass = provider.mass
65 self.logger = provider.logger
66
67 def _track_id_from_item_id(self, item_id: str) -> str:
68 """Extract API track ID from item_id (may be track_id@station_id for My Mix)."""
69 if RADIO_TRACK_ID_SEP in item_id:
70 return item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
71 return item_id
72
73 async def get_stream_details(self, item_id: str) -> StreamDetails:
74 """
75 Get stream details for a track.
76
77 Uses the unified /get-file-info endpoint for all quality tiers.
78 Falls back to /tracks/{id}/download-info if get-file-info fails.
79
80 :param item_id: Track ID or composite track_id@station_id for My Mix.
81 :return: StreamDetails for the track (item_id preserved for on_streamed).
82 :raises MediaNotFoundError: If stream URL cannot be obtained.
83 """
84 track_id = self._track_id_from_item_id(item_id)
85 track = await self.provider.get_track(item_id)
86 if not track:
87 raise MediaNotFoundError(f"Track {item_id} not found")
88
89 quality = (
90 str(self.provider.config.get_value(CONF_QUALITY) or QUALITY_BALANCED).strip().lower()
91 )
92 transport = (
93 str(self.provider.config.get_value(CONF_TRANSPORT) or TRANSPORT_RAW).strip().lower()
94 )
95
96 # Backward compatibility: old "lossless" config value
97 if quality == "lossless":
98 quality = QUALITY_LOSSLESS
99
100 fi_params = QUALITY_FILE_INFO_PARAMS.get(
101 quality, QUALITY_FILE_INFO_PARAMS[QUALITY_BALANCED]
102 )
103
104 # Allow advanced users to override codecs
105 codecs_override = str(self.provider.config.get_value(CONF_CODECS) or "").strip()
106 codecs = codecs_override or fi_params["codecs"]
107
108 self.logger.debug(
109 "Requesting stream for track %s: quality=%s, transport=%s, codecs=%s",
110 track_id,
111 quality,
112 transport,
113 codecs,
114 )
115
116 file_info = await self.client.get_track_file_info(
117 track_id,
118 quality=fi_params["quality"],
119 codecs=codecs,
120 transport=transport,
121 )
122
123 if file_info and file_info.get("url"):
124 url = file_info["url"]
125 codec = file_info.get("codec") or ""
126 needs_decryption = file_info.get("needs_decryption", False)
127
128 # Gather audio params: API response first, then probe container
129 bit_rate = file_info.get("bitrate") or file_info.get("bitrate_in_kbps") or 0
130 sample_rate = file_info.get("sample_rate") or 0
131 bit_depth = file_info.get("bit_depth") or 0
132
133 if (not sample_rate or not bit_depth) and not needs_decryption:
134 # Probe raw stream headers for real sample_rate/bit_depth
135 probed_sr, probed_bd = await self._probe_stream_params(url, codec)
136 sample_rate = sample_rate or probed_sr
137 bit_depth = bit_depth or probed_bd
138
139 self.logger.debug(
140 "Audio params for track %s: codec=%s, bit_rate=%s, sample_rate=%s, bit_depth=%s",
141 track_id,
142 codec,
143 bit_rate,
144 sample_rate,
145 bit_depth,
146 )
147
148 audio_format = self._build_audio_format(
149 codec,
150 bit_rate=bit_rate,
151 sample_rate=sample_rate,
152 bit_depth=bit_depth,
153 )
154
155 # Always use StreamType.CUSTOM with windowed Range requests to prevent CDN drops.
156 # can_seek=True only for codecs where bitrate * time yields a decodable byte
157 # offset â i.e. raw MP3. MP4-container codecs (aac-mp4, flac-mp4) need the
158 # ftyp/moov init atoms at the file start, so byte-offset seeks land in mdat
159 # with no codec config and produce undecodable data. Raw FLAC frames aren't
160 # fixed-size either, so byte-rate math doesn't land on a frame boundary.
161 # allow_seek=True lets ffmpeg handle time-based seeking via -ss in those cases.
162 byte_seekable = codec.lower() in ("mp3", "mpeg")
163 can_seek = not needs_decryption and bit_rate > 0 and byte_seekable
164 data: dict[str, Any] = {
165 "url": url,
166 "codec": codec,
167 "transport": transport,
168 "bit_rate": bit_rate,
169 # Stored for URL refresh on 4xx:
170 "fi_quality": fi_params["quality"],
171 "fi_codecs": codecs,
172 }
173 if needs_decryption and "key" in file_info:
174 data["decryption_key"] = file_info["key"]
175
176 return StreamDetails(
177 item_id=item_id,
178 provider=self.provider.instance_id,
179 audio_format=audio_format,
180 stream_type=StreamType.CUSTOM,
181 duration=track.duration,
182 data=data,
183 can_seek=can_seek,
184 allow_seek=not needs_decryption,
185 )
186
187 # Fallback: /tracks/{id}/download-info (defensive, should rarely trigger)
188 self.logger.warning(
189 "get-file-info failed for track %s, falling back to download-info", track_id
190 )
191 download_infos = await self.client.get_track_download_info(track_id, get_direct_links=True)
192 if not download_infos:
193 raise MediaNotFoundError(f"No stream info available for track {item_id}")
194
195 selected_info = self._select_best_quality(download_infos, quality)
196 if not selected_info or not selected_info.direct_link:
197 raise MediaNotFoundError(f"No stream URL available for track {item_id}")
198
199 self.logger.debug(
200 "Fallback stream for track %s: codec=%s, bitrate=%s",
201 track_id,
202 getattr(selected_info, "codec", None),
203 getattr(selected_info, "bitrate_in_kbps", None),
204 )
205
206 return StreamDetails(
207 item_id=item_id,
208 provider=self.provider.instance_id,
209 audio_format=self._build_audio_format(
210 selected_info.codec, bit_rate=selected_info.bitrate_in_kbps or 0
211 ),
212 stream_type=StreamType.HTTP,
213 duration=track.duration,
214 path=selected_info.direct_link,
215 can_seek=True,
216 allow_seek=True,
217 expiration=50, # download-info direct links expire after ~60s
218 )
219
220 def _select_best_quality(
221 self, download_infos: list[Any], preferred_quality: str | None
222 ) -> DownloadInfo | None:
223 """
224 Select the best quality download info based on user preference.
225
226 Used as fallback when get-file-info is unavailable.
227
228 :param download_infos: List of DownloadInfo objects.
229 :param preferred_quality: User's quality preference (efficient/high/balanced/superb).
230 :return: Best matching DownloadInfo or None.
231 """
232 if not download_infos:
233 return None
234
235 preferred_normalized = (preferred_quality or "").strip().lower()
236
237 # Sort by bitrate descending
238 sorted_infos = sorted(
239 download_infos,
240 key=lambda x: x.bitrate_in_kbps or 0,
241 reverse=True,
242 )
243
244 # Superb: Prefer FLAC (accept legacy "lossless" label for backward compatibility)
245 if preferred_normalized in {QUALITY_LOSSLESS, "lossless"}:
246 for codec in ("flac-mp4", "flac"):
247 for info in sorted_infos:
248 if info.codec and info.codec.lower() == codec:
249 return info
250 self.logger.warning(
251 "Superb quality (FLAC) requested but not available; using best available"
252 )
253 return sorted_infos[0]
254
255 # Efficient: Prefer lowest bitrate AAC/MP3
256 if preferred_normalized == QUALITY_EFFICIENT:
257 sorted_infos_asc = sorted(
258 download_infos,
259 key=lambda x: x.bitrate_in_kbps or 999,
260 )
261 for codec in ("aac-mp4", "aac", "he-aac-mp4", "he-aac", "mp3"):
262 for info in sorted_infos_asc:
263 if info.codec and info.codec.lower() == codec:
264 return info
265 return sorted_infos_asc[0]
266
267 # High: Prefer high bitrate MP3 (~320kbps)
268 if preferred_normalized == QUALITY_HIGH:
269 high_quality_mp3 = [
270 info
271 for info in sorted_infos
272 if info.codec
273 and info.codec.lower() == "mp3"
274 and info.bitrate_in_kbps
275 and info.bitrate_in_kbps >= 256
276 ]
277 if high_quality_mp3:
278 return high_quality_mp3[0]
279
280 for info in sorted_infos:
281 if info.codec and info.codec.lower() == "mp3":
282 return info
283
284 for info in sorted_infos:
285 if info.codec and info.codec.lower() not in ("flac", "flac-mp4"):
286 return info
287
288 return sorted_infos[0]
289
290 # Balanced (default): Prefer ~192kbps AAC
291 balanced_infos = [
292 info
293 for info in sorted_infos
294 if info.bitrate_in_kbps and 128 <= info.bitrate_in_kbps <= 256
295 ]
296 if balanced_infos:
297 for codec in ("aac-mp4", "aac", "he-aac-mp4", "he-aac", "mp3"):
298 for info in balanced_infos:
299 if info.codec and info.codec.lower() == codec:
300 return info
301 return balanced_infos[0]
302
303 return sorted_infos[0] if sorted_infos else None
304
305 def _get_content_type(self, codec: str | None) -> tuple[ContentType, ContentType]:
306 """
307 Determine container and codec type from Kion API codec string.
308
309 Kion API returns codec strings like "flac-mp4" (FLAC in MP4 container),
310 "aac-mp4" (AAC in MP4 container), or plain "flac", "mp3", "aac".
311 For MP4-container variants we return ContentType.MP4 as the container
312 and the actual audio codec via codec_type, matching the convention used
313 by the Yandex Music provider. This keeps container-aware behaviors in
314 MA core (mime type, seek handling, passthrough) correct.
315
316 :param codec: Codec string from Kion API.
317 :return: Tuple of (content_type/container, codec_type).
318 """
319 if not codec:
320 return ContentType.UNKNOWN, ContentType.UNKNOWN
321
322 codec_lower = codec.lower()
323
324 # MP4 container variants: codec is inside an MP4 container
325 if codec_lower == "flac-mp4":
326 return ContentType.MP4, ContentType.FLAC
327 if codec_lower in ("aac-mp4", "he-aac-mp4"):
328 return ContentType.MP4, ContentType.AAC
329
330 # Plain single-codec formats: codec is implied by content_type
331 if codec_lower == "flac":
332 return ContentType.FLAC, ContentType.UNKNOWN
333 if codec_lower in ("mp3", "mpeg"):
334 return ContentType.MP3, ContentType.UNKNOWN
335 if codec_lower in ("aac", "he-aac"):
336 return ContentType.AAC, ContentType.UNKNOWN
337
338 self.logger.debug("Unknown codec from Kion API: %s", codec)
339 return ContentType.UNKNOWN, ContentType.UNKNOWN
340
341 def _build_audio_format(
342 self,
343 codec: str | None,
344 *,
345 bit_rate: int = 0,
346 sample_rate: int = 0,
347 bit_depth: int = 0,
348 ) -> AudioFormat:
349 """
350 Build AudioFormat from codec string and optional stream metadata.
351
352 Values of 0 mean "unknown â let MA/ffmpeg detect from the actual stream".
353 Pass real values from the API response when available.
354
355 :param codec: Codec string from Kion API.
356 :param bit_rate: Bitrate in kbps (0 = unknown).
357 :param sample_rate: Sample rate in Hz (0 = unknown, detect from stream).
358 :param bit_depth: Bit depth (0 = unknown, detect from stream).
359 :return: Configured AudioFormat instance.
360 """
361 content_type, codec_type = self._get_content_type(codec)
362 kwargs: dict[str, Any] = {
363 "content_type": content_type,
364 "codec_type": codec_type,
365 }
366 # Only pass non-zero values; AudioFormat defaults to 44100/16 which
367 # MA/ffmpeg rely on. Passing 0 would override those defaults.
368 if bit_rate:
369 kwargs["bit_rate"] = bit_rate
370 if sample_rate:
371 kwargs["sample_rate"] = sample_rate
372 if bit_depth:
373 kwargs["bit_depth"] = bit_depth
374 return AudioFormat(**kwargs)
375
376 @staticmethod
377 def _parse_flac_streaminfo(header: bytes) -> tuple[int, int]:
378 """
379 Extract sample_rate and bit_depth from FLAC STREAMINFO block.
380
381 FLAC format: 4-byte magic "fLaC", then metadata blocks.
382 STREAMINFO is always the first block (type 0), 34 bytes payload.
383 Bytes 10-17 of STREAMINFO contain sample_rate (20 bits),
384 channels (3 bits), bit_depth (5 bits), total samples (36 bits).
385
386 :param header: First 42+ bytes of the FLAC stream.
387 :return: (sample_rate, bit_depth) or (0, 0) on parse failure.
388 """
389 if len(header) < 42 or header[:4] != b"fLaC":
390 return 0, 0
391 # STREAMINFO payload: 4 magic + 4 block header = 8 byte offset, 34 bytes long
392 # Bytes 10-13 of payload: sample_rate(20) | channels(3) | bps(5) | total(4 high)
393 payload = header[8:] # skip "fLaC" + block header
394 if len(payload) < 34:
395 return 0, 0
396 val = int.from_bytes(payload[10:14], "big")
397 sample_rate = (val >> 12) & 0xFFFFF
398 bit_depth = ((val >> 4) & 0x1F) + 1
399 return sample_rate, bit_depth
400
401 @staticmethod
402 def _parse_mp4_audio_params(header: bytes) -> tuple[int, int]:
403 """
404 Extract sample_rate and bit_depth from MP4/fMP4 container.
405
406 Scans for the 'dfLa' (FLAC-in-MP4) box, or falls back to parsing
407 the AudioSampleEntry in an 'mp4a' box to read sample size and sample rate.
408
409 :param header: First 8-32 KB of the MP4 stream.
410 :return: (sample_rate, bit_depth) or (0, 0) if not found.
411 """
412 # Quick scan for dfLa box (FLAC-in-MP4: contains FLAC STREAMINFO)
413 dfl_pos = header.find(b"dfLa")
414 if dfl_pos >= 4:
415 # dfLa box layout after "dfLa" type:
416 # 4 bytes version/flags
417 # 4 bytes STREAMINFO block header (type byte + 3-byte length)
418 # 34 bytes STREAMINFO payload
419 payload_start = dfl_pos + 4 + 4 + 4 # after type + version/flags + block header
420 payload = header[payload_start:]
421 if len(payload) >= 34:
422 val = int.from_bytes(payload[10:14], "big")
423 sample_rate = (val >> 12) & 0xFFFFF
424 bit_depth = ((val >> 4) & 0x1F) + 1
425 if 8000 <= sample_rate <= 384000 and 1 <= bit_depth <= 32:
426 return sample_rate, bit_depth
427
428 # Scan for mp4a AudioSampleEntry (AAC/generic audio in MP4)
429 mp4a_pos = header.find(b"mp4a")
430 if mp4a_pos >= 4:
431 # AudioSampleEntry: 4-byte size, "mp4a", 6 reserved, 2 data_ref,
432 # 2 version, 2 revision, 4 vendor, 2 channels, 2 sample_size,
433 # 2 compression_id, 2 packet_size, 4 sample_rate (16.16 fixed-point)
434 entry_start = mp4a_pos + 4 # after "mp4a"
435 entry = header[entry_start:]
436 if len(entry) >= 28:
437 sample_size = int.from_bytes(entry[18:20], "big")
438 sr_fixed = int.from_bytes(entry[24:28], "big")
439 sample_rate = sr_fixed >> 16
440 bit_depth = max(0, sample_size)
441 if 8000 <= sample_rate <= 384000:
442 return sample_rate, bit_depth
443
444 return 0, 0
445
446 async def _probe_stream_params(self, url: str, codec: str) -> tuple[int, int]:
447 """
448 Probe audio params by reading the first bytes of the stream.
449
450 Makes a small Range request to read container/stream headers,
451 then parses FLAC STREAMINFO or MP4 box structure.
452
453 :param url: Stream URL.
454 :param codec: Codec string from API (e.g. "flac-mp4", "flac").
455 :return: (sample_rate, bit_depth) or (0, 0) if probing fails.
456 """
457 codec_lower = (codec or "").lower()
458 # Determine how many bytes to read and which parser to use
459 if codec_lower == "flac":
460 probe_size = 64
461 parser = self._parse_flac_streaminfo
462 elif "-mp4" in codec_lower:
463 probe_size = 32768 # MP4 moov/stsd can be further in
464 parser = self._parse_mp4_audio_params
465 else:
466 return 0, 0 # lossy without container â let MA detect
467
468 try:
469 headers = {"Range": f"bytes=0-{probe_size - 1}"}
470 async with self.mass.http_session.get(
471 url,
472 headers=headers,
473 timeout=aiohttp.ClientTimeout(total=10),
474 ) as resp:
475 if resp.status not in (200, 206):
476 return 0, 0
477 header_bytes = await resp.content.read(probe_size)
478 self.logger.debug("Probe read %d bytes for codec=%s", len(header_bytes), codec)
479 result = parser(header_bytes)
480 self.logger.debug("Probe result: sample_rate=%d, bit_depth=%d", *result)
481 return result
482 except asyncio.CancelledError:
483 raise
484 except Exception:
485 self.logger.debug("Stream probe failed for codec=%s", codec)
486 return 0, 0
487
488 async def _refresh_stream_url(
489 self,
490 streamdetails: StreamDetails,
491 http_status: int,
492 bytes_yielded: int,
493 attempt: int,
494 max_retries: int,
495 ) -> bool:
496 """
497 Re-fetch an expired stream URL (works for both raw and encraw).
498
499 Updates streamdetails.data in-place with new URL (and key for encraw).
500
501 :return: True on success, False if retries exhausted.
502 """
503 if attempt >= max_retries:
504 return False
505 data = streamdetails.data
506 track_id = self._track_id_from_item_id(streamdetails.item_id)
507 self.logger.warning(
508 "Stream URL expired (HTTP %d) at %d bytes (attempt %d/%d) â re-fetching",
509 http_status,
510 bytes_yielded,
511 attempt + 1,
512 max_retries,
513 )
514 token = BYPASS_THROTTLER.set(True)
515 try:
516 file_info = await self.client.get_track_file_info(
517 track_id,
518 quality=data["fi_quality"],
519 codecs=data["fi_codecs"],
520 transport=data.get("transport", TRANSPORT_RAW),
521 )
522 finally:
523 BYPASS_THROTTLER.reset(token)
524 if file_info and file_info.get("url"):
525 data["url"] = file_info["url"]
526 if "decryption_key" in data and file_info.get("key"):
527 data["decryption_key"] = file_info["key"]
528 return True
529 return False
530
531 async def _decrypt_response_stream(
532 self,
533 response: Any,
534 key_bytes: bytes,
535 block_size: int,
536 bytes_delivered: int,
537 ) -> AsyncGenerator[bytes]:
538 """
539 Decrypt one HTTP response and yield plaintext chunks.
540
541 Aligns the AES-CTR counter to the correct block for resumption.
542 If the server ignores a Range header (200 instead of 206), resets the
543 counter to 0 and skips the already-delivered prefix transparently.
544
545 :param response: aiohttp ClientResponse (open context manager).
546 :param key_bytes: Raw AES key bytes.
547 :param block_size: AES block size (16 for CTR mode).
548 :param bytes_delivered: Total plaintext bytes already sent to the caller.
549 :return: Async generator yielding decrypted audio bytes.
550 """
551 block_start = (bytes_delivered // block_size) * block_size
552 block_skip = bytes_delivered - block_start
553
554 if block_start > 0 and response.status == 200:
555 self.logger.warning(
556 "Server ignored Range header at %d bytes (200 instead of 206)"
557 " â restarting decrypt from position 0, skipping %d already-sent bytes",
558 block_start,
559 bytes_delivered,
560 )
561 block_skip = bytes_delivered
562 block_num = (0).to_bytes(block_size, "big")
563 else:
564 block_num = (block_start // block_size).to_bytes(block_size, "big")
565
566 decryptor = Cipher(algorithms.AES(key_bytes), modes.CTR(block_num)).decryptor()
567 carry_skip = block_skip
568 async for chunk in response.content.iter_chunked(_CHUNK_SIZE):
569 decrypted = decryptor.update(chunk)
570 if carry_skip > 0:
571 skip = min(carry_skip, len(decrypted))
572 decrypted = decrypted[skip:]
573 carry_skip -= skip
574 if decrypted:
575 yield decrypted
576 final = decryptor.finalize()
577 if final:
578 yield final
579
580 def _handle_stream_error(
581 self,
582 err: Exception,
583 attempt: int,
584 max_retries: int,
585 bytes_yielded: int,
586 delays: tuple[float, ...],
587 label: str,
588 ) -> tuple[int, float]:
589 """
590 Increment retry counter, log a warning, or raise if retries are exhausted.
591
592 :param err: The exception that caused the retry.
593 :param attempt: Current retry attempt count (0-based).
594 :param max_retries: Maximum number of retries allowed.
595 :param bytes_yielded: Bytes delivered so far (for log context).
596 :param delays: Backoff delay sequence to pick from.
597 :param label: Short verb describing the failure (e.g. "dropped", "stalled").
598 :return: (new_attempt, retry_delay) tuple when retrying.
599 :raises MediaNotFoundError: When attempt count exceeds max_retries.
600 """
601 delay = delays[min(attempt, len(delays) - 1)]
602 attempt += 1
603 if attempt <= max_retries:
604 self.logger.warning(
605 "Stream %s at %d bytes (attempt %d/%d) â retrying",
606 label,
607 bytes_yielded,
608 attempt,
609 max_retries,
610 )
611 return attempt, delay
612 raise MediaNotFoundError(f"Stream {label} after retries were exhausted") from err
613
614 @staticmethod
615 def _is_content_range_eof(headers: Any, window_end: int) -> bool:
616 """
617 Return True when Content-Range indicates *window_end* reached the last file byte.
618
619 Parses ``Content-Range: bytes start-end/total`` and checks whether
620 ``window_end >= total - 1``. Returns False on any malformed header so
621 the caller falls back to the next window safely.
622 """
623 content_range = headers.get("Content-Range", "")
624 if not content_range.startswith("bytes "):
625 return False
626 try:
627 _, range_spec = content_range.split(" ", 1)
628 _, total_str = range_spec.split("/", 1)
629 total_str = total_str.strip()
630 return total_str.isdigit() and window_end >= int(total_str) - 1
631 except ValueError:
632 return False
633
634 async def _iter_raw_response(
635 self,
636 response: Any,
637 bytes_delivered: int,
638 block_start: int,
639 ) -> AsyncGenerator[bytes]:
640 """
641 Yield raw (unencrypted) chunks from one HTTP response.
642
643 If the server ignored the Range header (200 instead of 206), skips the
644 already-delivered prefix transparently.
645
646 :param response: aiohttp ClientResponse (open context manager).
647 :param bytes_delivered: Total bytes already sent to the caller.
648 :param block_start: Requested Range start offset.
649 :return: Async generator yielding raw audio bytes.
650 """
651 range_ignored = response.status == 200 and block_start > 0
652 skip_bytes = bytes_delivered if range_ignored else 0
653 async for raw_chunk in response.content.iter_chunked(_CHUNK_SIZE):
654 if skip_bytes > 0:
655 if len(raw_chunk) <= skip_bytes:
656 skip_bytes -= len(raw_chunk)
657 continue
658 raw_chunk = raw_chunk[skip_bytes:] # noqa: PLW2901
659 skip_bytes = 0
660 if raw_chunk:
661 yield raw_chunk
662
663 async def _handle_expired_url(
664 self,
665 streamdetails: StreamDetails,
666 response_status: int,
667 bytes_yielded: int,
668 attempt: int,
669 max_retries: int,
670 ) -> bytes:
671 """
672 Handle URL expiry (401/403/410) by refreshing and returning updated key.
673
674 On success returns the refreshed AES key bytes for encrypted streams,
675 or empty bytes for raw transport (caller ignores the return in that
676 case). Retry exhaustion raises ``MediaNotFoundError`` instead of
677 returning ``None``, so the function is guaranteed to produce a
678 ``bytes`` value when it returns.
679
680 :return: Updated AES key bytes (or empty bytes for raw transport).
681 :raises MediaNotFoundError: When refresh fails after retries exhausted.
682 """
683 if not await self._refresh_stream_url(
684 streamdetails,
685 response_status,
686 bytes_yielded,
687 attempt,
688 max_retries,
689 ):
690 raise MediaNotFoundError(
691 f"Stream URL expired (HTTP {response_status}) after retries exhausted"
692 )
693 data = streamdetails.data
694 if "decryption_key" in data:
695 try:
696 return bytes.fromhex(data["decryption_key"])
697 except ValueError as err:
698 raise MediaNotFoundError(f"Invalid decryption key: {err}") from err
699 return b""
700
701 @staticmethod
702 def _validate_encryption_key(data: dict[str, Any]) -> tuple[bool, bytes | None]:
703 """
704 Validate and extract encryption parameters from stream data.
705
706 :return: (is_encrypted, key_bytes) tuple.
707 :raises MediaNotFoundError: If AES key length is invalid.
708 """
709 if "decryption_key" not in data:
710 return False, None
711 try:
712 key_bytes = bytes.fromhex(data["decryption_key"])
713 except ValueError as err:
714 raise MediaNotFoundError(f"Invalid decryption key: {err}") from err
715 if len(key_bytes) not in (16, 24, 32):
716 raise MediaNotFoundError(f"Unsupported AES key length: {len(key_bytes)} bytes")
717 return True, key_bytes
718
719 def _calculate_seek_offset(
720 self, data: dict[str, Any], seek_position: int, is_encrypted: bool
721 ) -> int:
722 """
723 Calculate initial byte offset for raw transport seeking.
724
725 Byte-offset seeking is only safe for codecs where byte position and
726 time position are linearly related â i.e. raw MP3. MP4-container
727 formats (aac-mp4, flac-mp4) need the ftyp/moov init atoms at the
728 file start and raw FLAC frames are variable-size, so byte-offset
729 seeks there land in the middle of undecodable data. For those, we
730 return 0 and let ffmpeg handle time-based seeking via ``-ss``
731 (``allow_seek=True`` is set on the StreamDetails for that purpose).
732
733 :param data: Stream data dict (must contain 'bit_rate' in kbps).
734 :param seek_position: Seek offset in seconds.
735 :param is_encrypted: Whether the stream uses AES encryption.
736 :return: Byte offset to start streaming from (0 if not applicable).
737 """
738 if seek_position <= 0 or is_encrypted:
739 return 0
740 codec = str(data.get("codec") or "").lower()
741 if codec not in ("mp3", "mpeg"):
742 return 0
743 bit_rate = data.get("bit_rate") or 0
744 if not bit_rate:
745 return 0
746 byte_offset = seek_position * bit_rate * 1000 // 8
747 self.logger.debug(
748 "Seeking to %ds: byte offset %d (bitrate %d kbps)",
749 seek_position,
750 byte_offset,
751 bit_rate,
752 )
753 return byte_offset
754
755 async def get_audio_stream( # noqa: PLR0915
756 self, streamdetails: StreamDetails, seek_position: int = 0
757 ) -> AsyncGenerator[bytes]:
758 """
759 Return the audio stream via windowed Range requests.
760
761 Handles both raw (direct) and encraw (AES-CTR encrypted) transports.
762 Downloads in windowed Range requests of _RANGE_WINDOW bytes each to prevent
763 Kion CDN from dropping slow-consumer TCP connections.
764
765 On connection drop: flat short backoff (0.5s/1.0s/2.0s).
766 On read stall: exponential backoff (2s/4s/8s).
767 On URL expiry (HTTP 4xx): re-fetches URL and resumes from bytes_yielded.
768 Retry counter resets after each successful window.
769
770 :param streamdetails: Stream details with URL (and optional decryption key).
771 :param seek_position: Seek offset in seconds for raw transport (0 = from start).
772 :return: Async generator yielding audio bytes.
773 """
774 data = streamdetails.data
775 is_encrypted, key_bytes = self._validate_encryption_key(data)
776 initial_byte_offset = self._calculate_seek_offset(data, seek_position, is_encrypted)
777
778 max_retries = 6
779 bytes_yielded = initial_byte_offset
780 attempt = 0
781 retry_delay: float = 0.0
782
783 while True:
784 if attempt > 0:
785 await asyncio.sleep(retry_delay)
786
787 block_start = (
788 (bytes_yielded // _AES_BLOCK_SIZE) * _AES_BLOCK_SIZE
789 if is_encrypted
790 else bytes_yielded
791 )
792 window_end = block_start + _RANGE_WINDOW - 1
793
794 try:
795 async with self.mass.http_session.get(
796 data["url"],
797 headers={"Range": f"bytes={block_start}-{window_end}"},
798 timeout=_STREAM_TIMEOUT,
799 ) as response:
800 if response.status in (401, 403, 410):
801 new_key = await self._handle_expired_url(
802 streamdetails,
803 response.status,
804 bytes_yielded,
805 attempt,
806 max_retries,
807 )
808 if is_encrypted:
809 key_bytes = new_key
810 attempt += 1
811 retry_delay = 0.0
812 continue
813 if response.status == 416:
814 # Range Not Satisfiable â last complete window aligned with EOF,
815 # so our next request asked past the end. Treat as EOF.
816 return
817 try:
818 response.raise_for_status()
819 except Exception as err:
820 raise MediaNotFoundError(f"Failed to fetch stream: {err}") from err
821
822 bytes_before = bytes_yielded
823 if is_encrypted:
824 if key_bytes is None:
825 raise MediaNotFoundError("Missing decryption key")
826 block_skip = bytes_before - block_start
827 async for chunk in self._decrypt_response_stream(
828 response,
829 key_bytes,
830 _AES_BLOCK_SIZE,
831 bytes_yielded,
832 ):
833 bytes_yielded += len(chunk)
834 yield chunk
835 else:
836 range_ignored = response.status == 200 and block_start > 0
837 block_skip = bytes_before if range_ignored else 0
838 async for chunk in self._iter_raw_response(
839 response,
840 bytes_before,
841 block_start,
842 ):
843 bytes_yielded += len(chunk)
844 yield chunk
845
846 received = (bytes_yielded - bytes_before) + block_skip
847 if response.status == 200 or received < _RANGE_WINDOW:
848 return
849 if self._is_content_range_eof(response.headers, window_end):
850 return
851 attempt = 0
852 retry_delay = 0.0
853
854 except asyncio.CancelledError:
855 raise
856 except (ClientPayloadError, ServerDisconnectedError) as err:
857 attempt, retry_delay = self._handle_stream_error(
858 err,
859 attempt,
860 max_retries,
861 bytes_yielded,
862 _TCP_DROP_DELAYS,
863 "dropped",
864 )
865 except TimeoutError as err:
866 attempt, retry_delay = self._handle_stream_error(
867 err,
868 attempt,
869 max_retries,
870 bytes_yielded,
871 _STALL_DELAYS,
872 "stalled",
873 )
874