music-assistant-server
98.6 KB•PY
provider.py
98.6 KB • 2,231 lines • python
1"""Yandex Ynison plugin provider for Music Assistant."""
2
3from __future__ import annotations
4
5import asyncio
6import hashlib
7import random
8import time
9from collections.abc import AsyncGenerator, Callable
10from contextlib import aclosing, suppress
11from dataclasses import dataclass
12from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast
13
14from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
15from music_assistant_models.enums import (
16 ConfigEntryType,
17 ContentType,
18 EventType,
19 MediaType,
20 ProviderFeature,
21 ProviderType,
22 SourceControl,
23 StreamType,
24)
25from music_assistant_models.errors import (
26 InvalidDataError,
27 LoginFailed,
28 MediaNotFoundError,
29 PlayerCommandFailed,
30 SetupFailedError,
31 UnsupportedFeaturedException,
32)
33from music_assistant_models.media_items import AudioSource, ProviderMapping
34from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
35from ya_passport_auth import SecretStr
36from ya_passport_auth.ma import BorrowedCredentialSource
37
38from music_assistant.controllers.streams.constants import STREAM_SLOT_PLAYBACK_WAIT_TIMEOUT
39from music_assistant.helpers.ffmpeg import get_ffmpeg_stream
40from music_assistant.helpers.throttle_retry import BYPASS_THROTTLER, ThrottlerManager
41from music_assistant.models.plugin import PluginProvider, SourceControlValue
42
43from .auth import refresh_music_token
44from .constants import (
45 CONF_ALLOW_PLAYER_SWITCH,
46 CONF_DEVICE_ID,
47 CONF_MASS_PLAYER_ID,
48 CONF_OUTPUT_BIT_DEPTH,
49 CONF_OUTPUT_SAMPLE_RATE,
50 CONF_TOKEN,
51 CONF_X_TOKEN,
52 CONF_YM_INSTANCE,
53 DEFAULT_DISPLAY_NAME,
54 OUTPUT_AUTO,
55 YANDEX_MUSIC_CONF_QUALITY,
56 YANDEX_MUSIC_LOSSLESS_QUALITIES,
57 YM_INSTANCE_OWN,
58)
59from .protocols import YandexMusicProviderLike
60from .streaming import (
61 PCM_LOSSLESS_PARAMS,
62 PCM_LOSSY_PARAMS,
63 PROBE_ARGS,
64 make_pcm_format,
65)
66from .ynison_client import (
67 YnisonClient,
68 YnisonDeviceInfo,
69 YnisonSendError,
70 YnisonState,
71 generate_device_id,
72 make_version_block,
73)
74
75if TYPE_CHECKING:
76 from music_assistant_models.config_entries import ProviderConfig
77 from music_assistant_models.event import MassEvent
78 from music_assistant_models.media_items import AudioFormat
79 from music_assistant_models.provider import ProviderManifest
80
81 from music_assistant.mass import MusicAssistant
82
83# How often (seconds) to sync progress to MA UI and Ynison.
84_PROGRESS_SYNC_INTERVAL = 5.0
85
86# Grace window after our own REPLACE/seek during which incoming Ynison
87# progress updates are treated as our own echo (not a user seek).
88_ECHO_GRACE_PERIOD = 3.0
89
90# Bound on the synchronous pre-fetch in _prefetch_format_for_track. A slow
91# pre-fetch is treated like a failed one â fall back to the current format
92# and let the in-stream `_get_stream_details_with_retry` handle retries.
93_PREFETCH_FORMAT_TIMEOUT = 2.5
94
95# Idempotency cache TTL for outbound peer-commands.
96_COMMAND_IDEMPOTENCY_TTL = 1.0
97
98# stable id for the single AudioSource this provider exposes;
99# combined with the provider instance_id this forms the persistent uri
100AUDIO_SOURCE_ID = "main"
101
102# Retry settings for transient Yandex API failures
103_API_MAX_RETRIES = 3
104_API_INITIAL_BACKOFF = 2.0
105_API_MAX_BACKOFF = 30.0
106
107# Cache TTL for stream details (seconds)
108_STREAM_DETAILS_CACHE_TTL = 300 # 5 minutes
109
110# In-memory music-token cache TTL (seconds). Yandex music tokens live ~60 min;
111# 50 min leaves 10 min headroom before the server would reject them. Tied to
112# the borrow-mode-with-only-x_token + 401-storm path described in spec 0004.
113_MUSIC_TOKEN_TTL_S = 50 * 60
114
115# Maximum number of distinct x_token entries kept in the own-mode music-token
116# cache (borrow mode caches inside BorrowedCredentialSource). 4 keeps headroom
117# for an x_token rotation with one refresh in flight.
118_MUSIC_TOKEN_CACHE_MAX = 4
119
120# Accepted non-auto values for output format overrides; mirrors the options
121# offered in CONF_OUTPUT_SAMPLE_RATE / CONF_OUTPUT_BIT_DEPTH config entries.
122# Used defensively to reject stale/tampered values without raising.
123_VALID_SAMPLE_RATES: frozenset[str] = frozenset({"44100", "48000", "96000"})
124_VALID_BIT_DEPTHS: frozenset[str] = frozenset({"16", "24"})
125
126
127class _StreamOwnerMismatchError(InvalidDataError):
128 """Raised when linked-provider stream details belong to another instance."""
129
130
131@dataclass(frozen=True)
132class _CachedToken:
133 """
134 Music token entry in the in-memory cache.
135
136 `expires_monotonic` is compared against the provider's `_now()` seam.
137 """
138
139 token: SecretStr
140 expires_monotonic: float
141
142
143def _hash_x_token(x_token: str) -> str:
144 """
145 Return the SHA-256 hex digest of an x_token, used as cache key.
146
147 The raw x_token is never stored in dict keys (defence-in-depth against
148 accidental log / dump leakage of the cache structure).
149 """
150 return hashlib.sha256(x_token.encode("utf-8")).hexdigest()
151
152
153class YandexYnisonProvider(PluginProvider):
154 """Implementation of the Yandex Music Connect (Ynison) Plugin."""
155
156 # PluginProvider base does not declare `is_streaming_provider`; MA's
157 # audio-analysis path raises AttributeError for live sources without
158 # an explicit opt-out. Analysing transient external-source tracks
159 # buys nothing.
160 is_streaming_provider: bool = False
161
162 def __init__(
163 self,
164 mass: MusicAssistant,
165 manifest: ProviderManifest,
166 config: ProviderConfig,
167 supported_features: set[ProviderFeature],
168 ) -> None:
169 """Initialize the Ynison plugin provider."""
170 super().__init__(mass, manifest, config, supported_features)
171
172 # Setup identity and playback options
173 self._default_player_id: str = cast("str", self.get_setup_value(CONF_MASS_PLAYER_ID)) or ""
174 # the display name snapshot the Ynison device connected with (set at init)
175 self._advertised_name: str | None = None
176 allow_switch_value = self.config.get_value(CONF_ALLOW_PLAYER_SWITCH)
177 self._allow_player_switch: bool = (
178 cast("bool", allow_switch_value) if allow_switch_value is not None else True
179 )
180 self._cfg_sample_rate: str = (
181 cast("str", self.config.get_value(CONF_OUTPUT_SAMPLE_RATE)) or OUTPUT_AUTO
182 )
183 self._cfg_bit_depth: str = (
184 cast("str", self.config.get_value(CONF_OUTPUT_BIT_DEPTH)) or OUTPUT_AUTO
185 )
186
187 # Token source â None = own (manually entered CONF_TOKEN);
188 # otherwise the instance_id of a linked yandex_music provider to borrow from.
189 ym_instance_value = cast("str | None", self.get_setup_value(CONF_YM_INSTANCE))
190 self._ym_instance_id: str | None = (
191 ym_instance_value
192 if ym_instance_value and ym_instance_value != YM_INSTANCE_OWN
193 else None
194 )
195 # Borrow mode: read-only credential source over the linked
196 # yandex_music instance (shared auth layer). The owner stays the
197 # single writer/rotator of persisted credentials; minted music
198 # tokens are cached in-memory inside the source (TTL + LRU +
199 # coalesced refreshes per its spec).
200 self._borrow_source: BorrowedCredentialSource | None = (
201 BorrowedCredentialSource(self.mass, self._ym_instance_id)
202 if self._ym_instance_id is not None
203 else None
204 )
205
206 # Device ID â persist in config so re-registration uses the same ID
207 device_id = cast("str | None", self.config.get_value(CONF_DEVICE_ID))
208 if not device_id:
209 device_id = generate_device_id()
210 self._update_config_value(CONF_DEVICE_ID, device_id)
211 self._device_id: str = device_id
212
213 # Runtime state
214 self._active_player_id: str | None = None
215 self._ynison: YnisonClient | None = None
216 self._runner_task: asyncio.Task[None] | None = None
217 self._on_unload_callbacks: list[Callable[..., None]] = []
218 self._yandex_provider: YandexMusicProviderLike | None = None
219 self._current_streaming_track_id: str | None = None
220 self._track_changed_event = asyncio.Event()
221 self._stream_stop_event = asyncio.Event()
222 self._seek_position_ms: int = 0
223 self._seek_grace_until: float = 0.0
224 self._last_player_update_time: float = 0.0
225 self._actual_duration_ms: int = 0
226 self._prefetched_list: list[dict[str, Any]] | None = None
227 self._prefetch_task: asyncio.Task[Any] | None = None
228 self._normalized_params: dict[str, Any] = PCM_LOSSY_PARAMS
229 self._normalized_format: AudioFormat = make_pcm_format(PCM_LOSSY_PARAMS)
230
231 # Rate limiter for Yandex API calls (max 2 req/s)
232 self._api_throttler = ThrottlerManager(rate_limit=2, period=1.0)
233
234 # Progress tracking â byte counter is the single source of truth
235 # during active streaming; Ynison echoes are detected via
236 # YnisonState.last_update_is_echo and ignored.
237 self._streaming_progress_ms: int = 0
238
239 # AudioSource MediaItem + per-stream state
240 self._stream_metadata = StreamMetadata(
241 title=f"Yandex Music Connect | {self._display_name}",
242 )
243 self._audio_source = AudioSource(
244 item_id=AUDIO_SOURCE_ID,
245 provider=self.instance_id,
246 name=self.name,
247 provider_mappings={
248 ProviderMapping(
249 item_id=AUDIO_SOURCE_ID,
250 provider_domain=self.domain,
251 provider_instance=self.instance_id,
252 # Fresh AudioFormat copy: AudioFormat is mutable and MA's
253 # FFMpeg._log_reader_task sets `input_format.codec_type`
254 # in-place. Sharing `self._normalized_format` here would
255 # let that mutation leak into the ProviderMapping and into
256 # later StreamDetails snapshots.
257 audio_format=make_pcm_format(self._normalized_params),
258 )
259 },
260 can_play_pause=False,
261 can_seek=False,
262 can_next_previous=False,
263 exclusive=True,
264 allow_external_trigger=True,
265 )
266 # _in_use_by_player tracks the queue currently consuming our stream
267 self._in_use_by_player: str | None = None
268 # _active_session_id is the controller-provided token for the current
269 # stream request â used to reject stale on_source_unselected callbacks
270 # after a same-queue reconnect supersedes the previous request.
271 self._active_session_id: str | None = None
272
273 # Idempotency cache for outbound peer-commands. Suppresses duplicate
274 # (action, key) pairs inside `_COMMAND_IDEMPOTENCY_TTL` â protects
275 # against echo-storms where the same Ynison broadcast lands on our
276 # state-handler twice in quick succession.
277 self._command_idempotency: dict[tuple[str, str | None], float] = {}
278
279 # "Ynison paused us externally â expect a resume that needs
280 # `play_media` re-issuance." Set in `_pause_playback`, read in
281 # `_activate_playback`. Survives a stray `_stream_stop_event` clear
282 # independent of the stop signal (which covers non-pause stop reasons).
283 self._externally_paused: bool = False
284
285 # In-memory music-token cache keyed by SHA-256(x_token). 50-min TTL,
286 # 4-entry LRU. Coalesces concurrent refresh attempts via a single
287 # asyncio.Lock so a reconnect storm makes at most one Passport call.
288 # `_now` is a seam for tests to advance the clock.
289 self._token_cache: dict[str, _CachedToken] = {}
290 self._token_refresh_lock = asyncio.Lock()
291 self._now: Callable[[], float] = time.monotonic
292
293 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
294 """
295 Return Config entries to configure this provider.
296
297 Account, player and device identity are collected by the interactive setup flow;
298 only runtime playback options live here.
299 """
300 return (
301 ConfigEntry(
302 key=CONF_ALLOW_PLAYER_SWITCH,
303 type=ConfigEntryType.BOOLEAN,
304 default_value=True,
305 ),
306 ConfigEntry(
307 key=CONF_OUTPUT_SAMPLE_RATE,
308 type=ConfigEntryType.STRING,
309 default_value=OUTPUT_AUTO,
310 options=[
311 ConfigValueOption(OUTPUT_AUTO),
312 ConfigValueOption("44100"),
313 ConfigValueOption("48000"),
314 ConfigValueOption("96000"),
315 ],
316 advanced=True,
317 ),
318 ConfigEntry(
319 key=CONF_OUTPUT_BIT_DEPTH,
320 type=ConfigEntryType.STRING,
321 default_value=OUTPUT_AUTO,
322 options=[
323 ConfigValueOption(OUTPUT_AUTO),
324 ConfigValueOption("16"),
325 ConfigValueOption("24"),
326 ],
327 advanced=True,
328 ),
329 ConfigEntry(
330 key=CONF_DEVICE_ID,
331 type=ConfigEntryType.STRING,
332 hidden=True,
333 required=False,
334 ),
335 )
336
337 # ------------------------------------------------------------------
338 # Provider lifecycle
339 # ------------------------------------------------------------------
340
341 async def handle_async_init(self) -> None:
342 """Handle async initialization of the provider."""
343 if not self.get_setup_value(CONF_MASS_PLAYER_ID):
344 raise SetupFailedError(
345 "No connected Music Assistant player is configured",
346 translation_key="no_connected_player",
347 translation_owner=self.translation_owner,
348 )
349 if self._ym_instance_id is not None:
350 self.logger.info(
351 "Borrowing credentials from yandex_music instance '%s'",
352 self._ym_instance_id,
353 )
354 else:
355 self.logger.info("Using manually configured Yandex Music token (no auto-refresh)")
356 token = await self._resolve_token()
357
358 self._advertised_name = self._display_name
359 device_info = YnisonDeviceInfo(
360 device_id=self._device_id,
361 title=self._advertised_name,
362 )
363
364 self._ynison = YnisonClient(
365 token=token,
366 device_info=device_info,
367 on_state_update=self._handle_ynison_state,
368 logger=self.logger,
369 on_auth_failure=self._refresh_ynison_token,
370 )
371
372 self._runner_task = self.mass.create_task(self._ynison.connect())
373
374 # Subscribe to provider events to detect linked yandex_music provider
375 self._on_unload_callbacks.append(
376 self.mass.subscribe(
377 self._on_provider_event,
378 EventType.PROVIDERS_UPDATED,
379 )
380 )
381 # the advertised device name is snapshotted into the Ynison connection, so a
382 # rename of the connected player needs a reload to re-advertise correctly
383 self._on_unload_callbacks.append(
384 self.mass.subscribe(
385 self._on_connected_player_event,
386 # PLAYER_UPDATED covers provider-originated renames; the handler
387 # no-ops unless the display name actually changed
388 (EventType.PLAYER_ADDED, EventType.PLAYER_CONFIG_UPDATED, EventType.PLAYER_UPDATED),
389 id_filter=self._default_player_id,
390 )
391 )
392 # Initial check for matching provider
393 self.mass.create_task(self._check_yandex_provider_match())
394
395 async def unload(self, is_removed: bool = False) -> None:
396 """Handle close/cleanup of the provider."""
397 if self._prefetch_task and not self._prefetch_task.done():
398 self._prefetch_task.cancel()
399 with suppress(asyncio.CancelledError):
400 await self._prefetch_task
401
402 if self._ynison:
403 await self._ynison.disconnect()
404
405 if self._runner_task and not self._runner_task.done():
406 self._runner_task.cancel()
407 with suppress(asyncio.CancelledError):
408 await self._runner_task
409
410 for callback in self._on_unload_callbacks:
411 with suppress(KeyError):
412 callback()
413
414 async def get_audio_sources(self) -> list[AudioSource]:
415 """Return the AudioSources this plugin currently exposes."""
416 return [self._audio_source]
417
418 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
419 """
420 Return StreamDetails for streaming the Yandex Music Connect audio.
421
422 Side-effect-free: ownership is claimed in on_source_selected (which the
423 streams controller fires before this method on the actual stream
424 request). Keeping this idempotent means preload paths like
425 player_queues._load_item can fetch streamdetails without claiming the
426 source and blocking a subsequent cross-queue handoff.
427 """
428 if item_id != AUDIO_SOURCE_ID:
429 raise MediaNotFoundError(f"Unknown AudioSource: {item_id}")
430 return StreamDetails(
431 provider=self.instance_id,
432 item_id=item_id,
433 # Fresh AudioFormat copy per call: MA's ffmpeg mutates
434 # input_format.codec_type in place, so a shared instance would
435 # propagate that mutation into future stream-details snapshots.
436 audio_format=make_pcm_format(self._normalized_params),
437 media_type=MediaType.AUDIO_SOURCE,
438 stream_type=StreamType.CUSTOM,
439 stream_metadata=self._stream_metadata,
440 )
441
442 async def on_source_control(
443 self,
444 source_id: str,
445 action: SourceControl,
446 value: SourceControlValue = None,
447 ) -> None:
448 """Proxy playback control commands to Yandex via the linked Yandex Music provider."""
449 if source_id != AUDIO_SOURCE_ID:
450 return
451 if action == SourceControl.PLAY:
452 await self._on_play()
453 elif action == SourceControl.PAUSE:
454 await self._on_pause()
455 elif action == SourceControl.NEXT:
456 await self._on_next()
457 elif action == SourceControl.PREVIOUS:
458 await self._on_previous()
459 elif (
460 action == SourceControl.SEEK
461 # tolerate float positions from internal callers; bool is an int
462 # subclass, so a misrouted toggle must not become a 1-second seek
463 and isinstance(value, (int, float))
464 and not isinstance(value, bool)
465 ):
466 await self._on_seek(int(value))
467
468 async def get_audio_stream( # noqa: PLR0915
469 self, streamdetails: StreamDetails, seek_position: int = 0
470 ) -> AsyncGenerator[bytes]:
471 """
472 Return continuous audio stream following Ynison track changes.
473
474 Streams the current track, then waits for track changes and streams
475 the next track automatically. Runs until the source is deselected.
476
477 The PCM format is frozen at session start to match what the outer
478 ffmpeg captured from ``self._normalized_format``. If
479 ``_update_normalized_format()`` fires mid-session (e.g. a provider
480 reload), the new format takes effect only on the *next* session â
481 preventing bit-depth/sample-rate mismatches that cause noise.
482 """
483 self._stream_stop_event.clear()
484 # snapshot the consumer at session start; the rest of this generator
485 # treats the queue_id as the player_id (they are the same by convention).
486 # The lock may legitimately be empty here â MA's `_load_item` preload
487 # path drives the generator to fill an initial audio buffer BEFORE
488 # `on_source_selected` has been dispatched, so `_in_use_by_player` is
489 # still None on that call. `had_claim` records whether a lock was
490 # already in force at entry; only in that case do we enforce
491 # cross-session invariants on the loop and the `finally` cleanup.
492 player_id = self._in_use_by_player or ""
493 had_claim = self._in_use_by_player is not None
494 # Snapshot the active session id too so a same-queue reconnect (which
495 # updates _active_session_id but not _in_use_by_player) is treated as a
496 # superseding session: the loop exits early, and the finally clear
497 # below skips the release so it doesn't clobber the new claim.
498 captured_session_id = self._active_session_id
499
500 # MA's streams controller may pass a non-zero seek_position (e.g. a
501 # resume initiated through a path that does NOT go through Ynison and
502 # therefore did not set `_seek_position_ms`). Honor it as the seed for
503 # the upcoming track. The Ynison-driven seek path (`_activate_playback`
504 # / `_on_seek`) keeps writing `_seek_position_ms` directly, which
505 # subsequent track iterations consume â only the seed differs.
506 if seek_position > 0 and self._seek_position_ms == 0:
507 self._seek_position_ms = seek_position * 1000
508
509 # Freeze format for this streaming session so every inner ffmpeg
510 # produces data matching the outer ffmpeg's captured input_format.
511 session_params: dict[str, Any] = dict(self._normalized_params)
512 session_fmt: AudioFormat = make_pcm_format(session_params)
513
514 try:
515 while not self._stream_stop_event.is_set() and (
516 # Preload path: no claim was active at entry â drive the
517 # loop purely off Ynison state and the stop event.
518 not had_claim or not self._session_lost(player_id, captured_session_id)
519 ):
520 if not self._ynison or not self._ynison.state.current_track_id:
521 # Wait for a track to appear
522 self._track_changed_event.clear()
523 try:
524 await asyncio.wait_for(self._track_changed_event.wait(), timeout=30.0)
525 except TimeoutError:
526 continue
527 continue
528
529 # Clear event before reading state so any subsequent update
530 # re-sets the event instead of being silently cleared.
531 self._track_changed_event.clear()
532 track_id = self._ynison.state.current_track_id
533 self._current_streaming_track_id = track_id
534
535 # `_pause_playback` set the stop event; finalize.
536 if self._ynison.state.is_paused:
537 return
538
539 if not self._yandex_provider:
540 self.logger.warning(
541 "No linked Yandex Music provider â cannot stream track %s", track_id
542 )
543 self._stream_stop_event.set()
544 if self._in_use_by_player == player_id:
545 await self.mass.players.cmd_stop(player_id)
546 return
547
548 # Stream the current track
549 seek_ms = self._seek_position_ms
550 self._seek_position_ms = 0
551 bytes_yielded = 0
552 self._streaming_progress_ms = seek_ms
553 last_progress_sync = time.monotonic()
554
555 track_fmt = make_pcm_format(session_params)
556 track_stream = self._stream_track(
557 track_id, seek_ms=seek_ms, session_params=session_params
558 )
559 # aclosing: breaking out below must finalize the generator right away,
560 # otherwise the linked provider's stream slot stays charged until GC.
561 async with aclosing(track_stream):
562 async for chunk in track_stream:
563 yield chunk
564 bytes_yielded += len(chunk)
565 now_mono = time.monotonic()
566 if now_mono - last_progress_sync >= _PROGRESS_SYNC_INTERVAL:
567 last_progress_sync = now_mono
568 await self._sync_progress(
569 seek_ms, bytes_yielded, player_id, session_fmt
570 )
571 if (
572 self._track_changed_event.is_set()
573 or self._stream_stop_event.is_set()
574 or (had_claim and self._session_lost(player_id, captured_session_id))
575 ):
576 break
577
578 # Align to PCM frame boundary â prevents misalignment in MA's
579 # downstream ffmpeg when a track stream is interrupted mid-chunk.
580 # We pad with zeros (can't un-yield bytes already sent downstream).
581 frame_size = (track_fmt.bit_depth // 8) * track_fmt.channels
582 if frame_size > 0:
583 excess = bytes_yielded % frame_size
584 if excess:
585 yield b"\x00" * (frame_size - excess)
586
587 # Don't clear _current_streaming_track_id yet â keep it set
588 # during advance/wait so Ynison echo of the same track doesn't
589 # trigger a false track-change detection in _activate_playback.
590
591 if self._stream_stop_event.is_set():
592 break
593
594 # Differentiate "track finished naturally" from "inner loop
595 # broke out early". Signalling completion on an
596 # interrupted track makes Yandex auto-advance the queue â
597 # surfaces as an unwanted skip on pause / handoff.
598 broke_for_pause = self._ynison is not None and self._ynison.state.is_paused
599 broke_for_session_change = had_claim and self._session_lost(
600 player_id, captured_session_id
601 )
602 natural_end = (
603 not self._track_changed_event.is_set()
604 and not broke_for_pause
605 and not broke_for_session_change
606 and self._ynison is not None
607 )
608 if natural_end:
609 self.logger.info("Track %s finished, advancing to next", track_id)
610 await self._signal_track_completion()
611 if not await self._wait_for_track_change(track_id):
612 self._stream_stop_event.set()
613 break
614
615 # Clear before next iteration â the new track ID will be set at
616 # the top of the loop from the latest Ynison state.
617 self._current_streaming_track_id = None
618 finally:
619 # Release ownership only if THIS generator owned the claim at
620 # entry AND no one else has superseded it since. The double-guard
621 # protects against a same-queue reconnect refreshing the session
622 # id without changing the queue id; clearing the lock on the old
623 # generator's teardown would otherwise clobber the new session's
624 # claim. `had_claim` keeps the preload path from touching the lock
625 # at all (no claim ever existed to release).
626 if had_claim and not self._session_lost(player_id, captured_session_id):
627 self._in_use_by_player = None
628 self._current_streaming_track_id = None
629
630 async def on_source_selected(
631 self,
632 source_id: str,
633 player_id: str,
634 owner_player_id: str,
635 stream_session_id: str,
636 ) -> None:
637 """Handle callback when this AudioSource has been selected/started on a player."""
638 if source_id != AUDIO_SOURCE_ID or not player_id:
639 return
640
641 # Check if manual player switching is allowed
642 if not self._allow_player_switch:
643 current_target = self._get_target_player_id()
644 if player_id != current_target and current_target:
645 # Redirect to the configured target, but only once per
646 # idempotency window. The target may be a sendspin bridge /
647 # sync-group whose stream is consumed under a player id that
648 # never equals `current_target`, so each redirect re-triggers
649 # selection here. Re-issuing `play_media` on every rejection
650 # turns that into an unbounded AudioError storm; the raise
651 # below still aborts every wrong-player stream regardless.
652 if self._idempotent("source_redirect", current_target):
653 self.logger.debug(
654 "Player switching disabled, redirecting selection from %s to %s",
655 player_id,
656 current_target,
657 )
658 await self.mass.player_queues.play_media(
659 current_target, str(self._audio_source.uri)
660 )
661 msg = f"Player switching is disabled; source must remain on {current_target}"
662 raise RuntimeError(msg)
663
664 # Stop previous player if switching. The lock claim a few lines below
665 # replaces the previous queue's claim; the previous stream loop notices
666 # the queue change and exits cleanly.
667 if self._active_player_id and self._active_player_id != player_id:
668 prev_player_id = self._active_player_id
669 self.logger.info(
670 "Source selected on %s, stopping %s",
671 player_id,
672 prev_player_id,
673 )
674 try:
675 await self.mass.players.cmd_stop(prev_player_id)
676 except Exception as err:
677 self.logger.debug(
678 "Failed to stop previous player %s: %s",
679 prev_player_id,
680 err,
681 )
682
683 # Claim ownership for this queue. The lock lives here (not in
684 # get_stream_details) so preload paths can fetch streamdetails without
685 # accidentally blocking a subsequent cross-queue handoff at the actual
686 # stream request.
687 self._in_use_by_player = owner_player_id
688 # Record this request's session id so a later on_source_unselected can
689 # tell whether it is the live teardown or a stale callback from a
690 # superseded same-queue request.
691 self._active_session_id = stream_session_id
692 self._active_player_id = player_id
693 self.logger.debug("Active player set to: %s", player_id)
694
695 async def on_source_unselected(
696 self, source_id: str, owner_player_id: str, stream_session_id: str
697 ) -> None:
698 """Release the queue-scoped exclusive claim when MA tears down the stream."""
699 if source_id != AUDIO_SOURCE_ID:
700 return
701 # Reject stale callbacks: only release if this is still the active
702 # session. A owner_player_id check alone is not sufficient â same-queue
703 # reconnects (player drops + reopens the same stream URL before the
704 # original request's finally fires) would otherwise let the old
705 # request's late callback clear the live claim of the new stream.
706 if self._active_session_id != stream_session_id:
707 return
708 self._active_session_id = None
709 if self._in_use_by_player == owner_player_id:
710 self._in_use_by_player = None
711
712 async def _wait_for_track_change(self, old_track_id: str, timeout: float = 30.0) -> bool:
713 """
714 Wait for Ynison to report a different track, ignoring echoes.
715
716 After _signal_track_completion sends update_playing_status, Ynison
717 echoes back the same track with updated progress. Only return True
718 once current_track_id actually differs from old_track_id.
719 """
720 deadline = time.monotonic() + timeout
721 while not self._stream_stop_event.is_set():
722 # Check state BEFORE clearing the event. Ynison may have already
723 # advanced between _signal_track_completion() returning and this
724 # method running; clearing first would drop the set() that went
725 # with the state update, leaving us to wait until timeout.
726 # Check is race-free: no await between the read and clear() below.
727 # None means empty/unreadable queue â treat as "not advanced."
728 if self._ynison:
729 current = self._ynison.state.current_track_id
730 if current is not None and current != old_track_id:
731 return True
732 self._track_changed_event.clear()
733 remaining = deadline - time.monotonic()
734 if remaining <= 0:
735 break
736 try:
737 await asyncio.wait_for(self._track_changed_event.wait(), timeout=remaining)
738 except TimeoutError:
739 break
740 self.logger.info("No new track from Ynison after completion, stopping stream")
741 return False
742
743 async def _stream_track(
744 self,
745 track_id: str,
746 seek_ms: int = 0,
747 session_params: dict[str, Any] | None = None,
748 ) -> AsyncGenerator[bytes]:
749 """
750 Stream a single track, normalizing to fixed PCM via per-track ffmpeg.
751
752 Every track is decoded through its own ffmpeg process to produce a
753 fixed PCM output (s16le or s24le based on YM quality setting). This
754 ensures MA's single ffmpeg process never encounters mid-stream format
755 changes (codec, bit depth, sample rate).
756
757 *session_params* â frozen format dict from the enclosing
758 ``get_audio_stream()`` session. Falls back to the current
759 ``_normalized_params`` when called outside a session.
760 """
761 provider = self._yandex_provider
762 if provider is None:
763 self.logger.warning(
764 "Linked Yandex Music provider unavailable â stopping track %s",
765 track_id,
766 )
767 self._stream_stop_event.set()
768 return
769 # In-flight stream fetch outranks unrelated 429 cooldowns:
770 # dropping a stream the user is actively trying to play is
771 # worse than risking another captcha. Prefetch deliberately
772 # stays throttled (see `_prefetch_format_for_track`).
773 bypass_token = BYPASS_THROTTLER.set(True)
774 try:
775 stream_details = await self._get_stream_details_with_retry(track_id, provider=provider)
776 except Exception:
777 self.logger.exception("Failed to get stream details for track %s", track_id)
778 self._stream_stop_event.set()
779 return
780 finally:
781 BYPASS_THROTTLER.reset(bypass_token)
782
783 if not self._linked_provider_is_current(provider):
784 self.logger.warning(
785 "Linked Yandex Music provider changed mid-stream â stopping track %s",
786 track_id,
787 )
788 self._stream_stop_event.set()
789 return
790
791 await self._update_metadata_from_stream(stream_details, seek_ms)
792 if not self._linked_provider_is_current(provider):
793 self.logger.warning(
794 "Linked Yandex Music provider changed while preparing track %s",
795 track_id,
796 )
797 self._stream_stop_event.set()
798 return
799
800 # No -re here: MA's realtime pacer is the single pacing authority for
801 # AudioSources. Pacing the decode a second time would pin it to realtime
802 # and forbid the small read-ahead that absorbs CDN jitter; back-pressure
803 # through the generator chain still bounds memory.
804 extra_input_args = list(PROBE_ARGS)
805 if seek_ms > 0:
806 extra_input_args += ["-ss", f"{seek_ms / 1000.0:.3f}"]
807
808 # Use session format when available, otherwise current normalized params
809 params = session_params if session_params is not None else self._normalized_params
810 out_fmt = make_pcm_format(params)
811 # Log the output rate + bit depth alongside the source format: with the
812 # passthrough fast path this PCM IS the delivered audio, so the line must
813 # let an operator read rate passthrough vs a resample, not just codec.
814 self.logger.info(
815 "Streaming track %s â %s/%dHz/%dbit: input=%s seek=%dms",
816 track_id,
817 out_fmt.content_type.value,
818 out_fmt.sample_rate,
819 out_fmt.bit_depth,
820 stream_details.audio_format,
821 seek_ms,
822 )
823 async with provider.acquire_stream_slot(STREAM_SLOT_PLAYBACK_WAIT_TIMEOUT):
824 if not self._linked_provider_is_current(provider):
825 self.logger.warning(
826 "Linked Yandex Music provider changed before starting track %s",
827 track_id,
828 )
829 self._stream_stop_event.set()
830 return
831 raw_stream = provider.get_audio_stream(stream_details)
832 ffmpeg_stream = get_ffmpeg_stream(
833 audio_input=raw_stream,
834 input_format=stream_details.audio_format,
835 output_format=out_fmt,
836 extra_input_args=extra_input_args,
837 )
838 async with aclosing(raw_stream), aclosing(ffmpeg_stream):
839 async for chunk in ffmpeg_stream:
840 if not self._linked_provider_is_current(provider):
841 self.logger.warning(
842 "Linked Yandex Music provider changed while streaming track %s",
843 track_id,
844 )
845 self._stream_stop_event.set()
846 break
847 yield chunk
848
849 async def _get_stream_details_with_retry(
850 self,
851 track_id: str,
852 media_type: MediaType = MediaType.TRACK,
853 *,
854 provider: YandexMusicProviderLike | None = None,
855 ) -> StreamDetails:
856 """Fetch stream details with caching, throttling, and retry."""
857 # Capture the linked yandex_music provider into a local ref at entry.
858 # self._yandex_provider can flip to None mid-await when the linked
859 # MusicProvider is unloaded (see _check_yandex_provider_match, which
860 # runs as a background task on provider-loaded/unloaded events).
861 # Dereferencing the attribute after an await would raise
862 # AttributeError and hard-stop the audio generator.
863 provider = provider or self._yandex_provider
864 if provider is None:
865 raise LoginFailed(
866 "Linked Yandex Music provider is not loaded â cannot fetch stream details"
867 )
868
869 cache_key = self._stream_details_cache_key(provider.instance_id, track_id)
870 cached = await self.mass.cache.get(
871 cache_key,
872 provider=self.instance_id,
873 base_class=StreamDetails,
874 )
875 if cached is not None:
876 cached_streamdetails = cast("StreamDetails", cached)
877 if cached_streamdetails.provider == provider.instance_id:
878 self.logger.debug("Stream details cache hit for %s", track_id)
879 return cached_streamdetails
880 await self.mass.cache.delete(cache_key, provider=self.instance_id)
881 self.logger.warning(
882 "Discarded stream details for %s owned by %s instead of %s",
883 track_id,
884 cached_streamdetails.provider,
885 provider.instance_id,
886 )
887
888 backoff = _API_INITIAL_BACKOFF
889 last_err: Exception | None = None
890 for attempt in range(_API_MAX_RETRIES):
891 async with self._api_throttler.acquire() as delay:
892 if delay > 0:
893 self.logger.debug("get_stream_details throttled %.1fs", delay)
894 try:
895 sd = await provider.get_stream_details(track_id, media_type)
896 if sd.provider != provider.instance_id:
897 raise _StreamOwnerMismatchError(
898 f"Stream details for {track_id} belong to {sd.provider}, "
899 f"expected {provider.instance_id}"
900 )
901 # StreamDetails.data has serialize="omit", so to_dict()
902 # strips it. Manually include it so cached entries keep
903 # the URL / decryption key needed by get_audio_stream().
904 cache_value = sd.to_dict()
905 cache_value["data"] = sd.data
906 # Respect the provider's expiration (e.g. yandex_music sets
907 # 50 s because CDN URLs expire after ~60 s). Fall back to
908 # our default TTL when the provider does not override.
909 cache_ttl = min(_STREAM_DETAILS_CACHE_TTL, sd.expiration)
910 if cache_ttl > 0:
911 await self.mass.cache.set(
912 cache_key,
913 cache_value,
914 expiration=cache_ttl,
915 provider=self.instance_id,
916 )
917 return sd
918 except asyncio.CancelledError:
919 raise
920 except _StreamOwnerMismatchError:
921 raise
922 except Exception as err:
923 last_err = err
924 if attempt < _API_MAX_RETRIES - 1:
925 jitter = backoff * random.uniform(0.75, 1.25)
926 self.logger.warning(
927 "get_stream_details attempt %d/%d failed: %s, retrying in %.1fs",
928 attempt + 1,
929 _API_MAX_RETRIES,
930 err,
931 jitter,
932 )
933 await asyncio.sleep(jitter)
934 backoff = min(backoff * 2, _API_MAX_BACKOFF)
935 msg = f"get_stream_details failed after {_API_MAX_RETRIES} attempts for {track_id}"
936 raise RuntimeError(msg) from last_err
937
938 async def _invalidate_stream_cache(
939 self, track_id: str, provider_instance_id: str | None = None
940 ) -> None:
941 """
942 Evict cached stream details for a track so the next fetch is fresh.
943
944 :param track_id: Track whose cached stream details should be dropped.
945 :param provider_instance_id: Linked provider instance that owns the entry,
946 defaulting to the currently linked one.
947 """
948 if provider_instance_id is None:
949 if self._yandex_provider is None:
950 return
951 provider_instance_id = self._yandex_provider.instance_id
952 cache_key = self._stream_details_cache_key(provider_instance_id, track_id)
953 await self.mass.cache.delete(cache_key, provider=self.instance_id)
954 self.logger.debug("Invalidated stream cache for %s", track_id)
955
956 @staticmethod
957 def _stream_details_cache_key(provider_instance_id: str, track_id: str) -> str:
958 """Return the cache key for one linked provider instance and track."""
959 return f"ynison_sd_{provider_instance_id}_{track_id}"
960
961 def _linked_provider_is_current(self, provider: YandexMusicProviderLike) -> bool:
962 """Return whether the captured linked provider still owns streaming."""
963 return self._yandex_provider is provider and provider.available
964
965 # ------------------------------------------------------------------
966 # Token handling
967 # ------------------------------------------------------------------
968
969 async def _refresh_via_x_token(self, x_token: str) -> SecretStr:
970 """
971 Refresh the music token from an x_token, caching the result.
972
973 Within :data:`_MUSIC_TOKEN_TTL_S` of a successful refresh, subsequent
974 calls for the same x_token return the cached :class:`SecretStr`
975 without hitting Yandex Passport. Concurrent callers coalesce via
976 :attr:`_token_refresh_lock`.
977
978 :param x_token: Long-lived session token to exchange for a music
979 token. Hashed before use as a cache key; the raw value is
980 never stored in dict keys or logs.
981 :returns: Fresh or cached music-scoped :class:`SecretStr`.
982 :raises LoginFailed: When Yandex explicitly rejects the x_token
983 (propagated from :func:`provider.auth.refresh_music_token`).
984 :raises ResourceTemporarilyUnavailable: On transient Passport
985 failures (network, rate limit) â retry later, credentials
986 are still good.
987 """
988 cache_key = _hash_x_token(x_token)
989 cached = self._token_cache.get(cache_key)
990 now = self._now()
991 if cached is not None and cached.expires_monotonic > now:
992 return cached.token
993
994 async with self._token_refresh_lock:
995 # Double-check inside the lock â a peer caller may have refreshed
996 # while we were waiting for the lock, in which case we reuse
997 # their fresh entry instead of issuing a duplicate Passport call.
998 cached = self._token_cache.get(cache_key)
999 now = self._now()
1000 if cached is not None and cached.expires_monotonic > now:
1001 return cached.token
1002
1003 token = await refresh_music_token(SecretStr(x_token))
1004 self._store_cached_token(cache_key, token)
1005 return token
1006
1007 def _store_cached_token(self, cache_key: str, token: SecretStr) -> None:
1008 """
1009 Insert a cache entry, enforcing the LRU bound.
1010
1011 Refreshing an existing key bumps its position to most-recent. When
1012 a new key would push the cache over :data:`_MUSIC_TOKEN_CACHE_MAX`,
1013 the oldest entry is evicted first.
1014 """
1015 # Reordering: pop-then-set positions the (possibly-new) key as
1016 # most-recent in Python's insertion-ordered dict.
1017 self._token_cache.pop(cache_key, None)
1018 while len(self._token_cache) >= _MUSIC_TOKEN_CACHE_MAX:
1019 oldest = next(iter(self._token_cache))
1020 self._token_cache.pop(oldest)
1021 self._token_cache[cache_key] = _CachedToken(
1022 token=token,
1023 expires_monotonic=self._now() + _MUSIC_TOKEN_TTL_S,
1024 )
1025
1026 def _invalidate_cached_token(self, x_token: str) -> None:
1027 """Drop the cache entry for an x_token (e.g. after a 401)."""
1028 self._token_cache.pop(_hash_x_token(x_token), None)
1029
1030 async def _resolve_token(self) -> SecretStr:
1031 """
1032 Resolve the Yandex Music OAuth token for the Ynison connection.
1033
1034 In borrow mode: read from the linked yandex_music provider's config.
1035 If only x_token is present (YM hasn't refreshed yet), do a cached
1036 in-memory refresh without writing back â YM owns token persistence.
1037
1038 In own mode: return CONF_TOKEN if set; otherwise, when CONF_X_TOKEN
1039 is present (QR-with-Remember-session path), cached in-memory refresh.
1040 """
1041 if self._borrow_source is not None:
1042 return await self._borrow_source.resolve_music_token()
1043
1044 token = cast("str | None", self.get_setup_value(CONF_TOKEN))
1045 if token:
1046 return SecretStr(token)
1047 x_token = cast("str | None", self.get_setup_value(CONF_X_TOKEN))
1048 if x_token:
1049 self.logger.debug("Own-mode token not present â refreshing from stored x_token")
1050 return await self._refresh_via_x_token(x_token)
1051 raise LoginFailed("No Yandex Music token configured")
1052
1053 async def _refresh_ynison_token(self) -> SecretStr:
1054 """
1055 Refresh the OAuth token for Ynison reconnection.
1056
1057 Called by YnisonClient on auth failure (401/403) during reconnect.
1058
1059 In borrow mode: re-read the linked YM instance's x_token and refresh
1060 in-memory only (no config writes â YM owns token persistence).
1061
1062 In own mode: refresh from stored CONF_X_TOKEN when present (QR with
1063 "Remember session" enabled). When absent (manual token paste only),
1064 surface LoginFailed so the user knows to paste a new token.
1065
1066 The cached token entry for the current x_token is invalidated up
1067 front â this method is reached only on a server-rejected token, so
1068 the cached value is provably stale.
1069 """
1070 if self._borrow_source is not None:
1071 ym_music_token, ym_x_token = self._borrow_source.read_tokens()
1072 if ym_x_token is None:
1073 raise LoginFailed("Cannot refresh: linked Yandex Music instance has no x_token")
1074 # Both the minted entry AND the owner's persisted token may be the
1075 # value the server just rejected â invalidate both so the source
1076 # can't re-serve either; it will mint fresh from x_token.
1077 if ym_music_token is not None:
1078 self._borrow_source.invalidate(ym_music_token)
1079 self._borrow_source.invalidate(ym_x_token)
1080 self.logger.info("Refreshing Yandex Music token for Ynison reconnect (borrow mode)")
1081 return await self._borrow_source.resolve_music_token()
1082
1083 x_token = cast("str | None", self.get_setup_value(CONF_X_TOKEN))
1084 if x_token:
1085 self._invalidate_cached_token(x_token)
1086 self.logger.info("Refreshing Yandex Music token for Ynison reconnect (own mode)")
1087 return await self._refresh_via_x_token(x_token)
1088
1089 raise LoginFailed(
1090 "Token expired and no stored x_token to refresh from. Re-authenticate "
1091 "via QR or paste a fresh Yandex Music token."
1092 )
1093
1094 # ------------------------------------------------------------------
1095 # Ynison state handling
1096 # ------------------------------------------------------------------
1097
1098 async def _handle_ynison_state(self, state: YnisonState) -> None:
1099 """Handle state update from Ynison."""
1100 is_our_device = state.active_device_id == self._device_id
1101
1102 # Detailed queue logging for diagnostics
1103 queue = state.player_state.get("player_queue", {})
1104 playable_list = queue.get("playable_list", [])
1105 current_index = queue.get("current_playable_index", -1)
1106 entity_type = queue.get("entity_type", "")
1107 entity_id = queue.get("entity_id", "")
1108 track_id = state.current_track_id
1109 self.logger.debug(
1110 "Ynison state: active_device=%s (ours=%s) track=%s "
1111 "index=%d/%d entity=%s type=%s paused=%s progress=%dms",
1112 state.active_device_id,
1113 is_our_device,
1114 track_id,
1115 current_index,
1116 len(playable_list),
1117 entity_id[:40] if entity_id else "<none>",
1118 entity_type,
1119 state.is_paused,
1120 state.progress_ms,
1121 )
1122
1123 # Post-reconnect settle window: the first inbound state after a WS
1124 # reconnect may reflect pre-reconnect peer state (active device etc).
1125 # Acting on it would re-issue play_media, mirror a stale paused flag
1126 # to MA, or worst case clobber a fresh local claim. The 2 s window in
1127 # YnisonClient._connect_state gives the server time to emit a state
1128 # broadcast that reflects our re-registered presence; until then we
1129 # only log.
1130 if self._ynison and self._ynison.in_post_reconnect_settle:
1131 self.logger.debug(
1132 "Skipping state inside post-reconnect settle window (track=%s paused=%s)",
1133 track_id,
1134 state.is_paused,
1135 )
1136 return
1137
1138 if is_our_device and not state.is_paused:
1139 self.logger.info(
1140 "Ynison â playing (track=%s progress=%dms)", track_id, state.progress_ms
1141 )
1142 # Pre-fetch next batch when playing second-to-last track
1143 self._maybe_prefetch(current_index, playable_list, entity_id, entity_type)
1144 await self._activate_playback(state)
1145 elif is_our_device and state.is_paused:
1146 self.logger.info(
1147 "Ynison â paused (track=%s progress=%dms)", track_id, state.progress_ms
1148 )
1149 await self._pause_playback()
1150 elif self._in_use_by_player:
1151 self.logger.info(
1152 "Ynison â other device active (was=%s), clearing",
1153 state.active_device_id,
1154 )
1155 self._clear_active_player()
1156
1157 async def _activate_playback(self, state: YnisonState) -> None: # noqa: PLR0915
1158 """Activate playback on the target MA player."""
1159 target_player_id = self._get_target_player_id()
1160 if not target_player_id:
1161 self.logger.warning("Ynison active on our device but no MA player available")
1162 return
1163
1164 # Resume after pause / fresh start: either signal triggers
1165 # play_media below. `_externally_paused` survives a stray stop-event
1166 # clear; the stop event covers non-pause stop reasons
1167 # (`_stream_track` warning branch, `_clear_active_player`).
1168 needs_reselect = self._stream_stop_event.is_set() or self._externally_paused
1169 self._stream_stop_event.clear()
1170 self._externally_paused = False
1171
1172 # Start playback via the standard play_media flow if not already active.
1173 # Guard on _active_player_id (set immediately) rather than in_use_by_queue
1174 # (set by get_stream_details when the streams controller picks up the request)
1175 # to prevent queuing redundant play_media calls during the ~5s gap.
1176 if self._active_player_id != target_player_id or needs_reselect:
1177 # Pre-fetch the upcoming track's real format BEFORE submitting
1178 # play_media so the AudioSource's provider_mapping carries the
1179 # right audio_format when the streams controller calls
1180 # get_stream_details(). Skip on same-track same-player resume â
1181 # the cached format is still correct for that case.
1182 upcoming = state.current_track_id
1183 switching_player = self._active_player_id != target_player_id
1184 self._active_player_id = target_player_id
1185 if upcoming and (switching_player or upcoming != self._current_streaming_track_id):
1186 await self._prefetch_format_for_track(upcoming)
1187 self.mass.create_task(
1188 self.mass.player_queues.play_media(target_player_id, str(self._audio_source.uri))
1189 )
1190
1191 # Signal track change if track_id changed
1192 significant_change = False
1193 new_track = state.current_track_id
1194 if new_track and new_track != self._current_streaming_track_id:
1195 self.logger.info("Track changed: %s -> %s", self._current_streaming_track_id, new_track)
1196 self._current_streaming_track_id = new_track
1197 self._seek_position_ms = state.progress_ms
1198 self._track_changed_event.set()
1199 significant_change = True
1200 # Grace period: ignore seek detection for a few seconds after
1201 # track change â Ynison echoes can report stale progress that
1202 # looks like a large drift.
1203 self._seek_grace_until = time.monotonic() + _ECHO_GRACE_PERIOD
1204 elif new_track and new_track == self._current_streaming_track_id:
1205 # Same-track resume after pause: explicitly seek to the Ynison position
1206 # so the new stream starts at the right offset.
1207 if needs_reselect:
1208 self._seek_position_ms = state.progress_ms
1209 self._track_changed_event.set()
1210 self._seek_grace_until = time.monotonic() + _ECHO_GRACE_PERIOD
1211 significant_change = True
1212 else:
1213 # Detect seek: compare Ynison progress against our stream position.
1214 # Ignore Ynison echoes (updates authored by our own device_id) to
1215 # prevent feedback loops where our own progress triggers false seeks.
1216 now = time.monotonic()
1217 if now < self._seek_grace_until:
1218 pass # Skip during grace period after track change or seek
1219 elif state.last_update_is_echo:
1220 pass # Echo of our own update â ignore
1221 else:
1222 our_ms = self._streaming_progress_ms
1223 if our_ms >= 0:
1224 verdict = self._classify_drift(state.progress_ms, our_ms)
1225 if verdict == "seek":
1226 drift_ms = abs(state.progress_ms - our_ms)
1227 self.logger.info(
1228 "Seek detected on track %s: "
1229 "expected ~%dms, Ynison at %dms (drift %dms)",
1230 new_track,
1231 our_ms,
1232 state.progress_ms,
1233 int(drift_ms),
1234 )
1235 self._seek_position_ms = state.progress_ms
1236 self._track_changed_event.set()
1237 self._seek_grace_until = now + _ECHO_GRACE_PERIOD
1238 significant_change = True
1239 elif verdict == "queue_rebuild":
1240 self.logger.debug(
1241 "Drift on track %s classified as queue-rebuild "
1242 "echo (Ynison=%dms, ours=%dms) â not seeking",
1243 new_track,
1244 state.progress_ms,
1245 our_ms,
1246 )
1247
1248 # Update metadata from state
1249 self._update_metadata(state)
1250
1251 # Always trigger player update on significant changes;
1252 # throttle regular updates to avoid UI churn (every 5 seconds).
1253 # Use force_update on seek/track change so the server broadcasts a full
1254 # PLAYER_UPDATED event instead of a lightweight elapsed-time-only one
1255 # that the frontend may not handle for AudioSource players.
1256 now_mono = time.monotonic()
1257 if significant_change or needs_reselect or now_mono - self._last_player_update_time >= 5.0:
1258 self.mass.players.trigger_player_update(
1259 target_player_id, force_update=significant_change
1260 )
1261 self._last_player_update_time = now_mono
1262
1263 def _update_metadata(self, state: YnisonState) -> None:
1264 """Update AudioSource metadata from Ynison state."""
1265 meta = self._stream_metadata
1266
1267 # Update duration (prefer actual from stream_details) and elapsed time
1268 best_duration = self._best_duration_ms()
1269 if best_duration:
1270 meta.duration = best_duration // 1000
1271 # Only update elapsed from Ynison when NOT actively streaming â
1272 # during streaming, _sync_progress provides byte-accurate progress.
1273 if state.progress_ms is not None and not self._in_use_by_player:
1274 meta.elapsed_time = state.progress_ms // 1000
1275 meta.elapsed_time_last_updated = time.time()
1276
1277 # Extract track info from player state if available
1278 queue = state.player_state.get("player_queue", {})
1279 playable_list = queue.get("playable_list", [])
1280 index = queue.get("current_playable_index", 0)
1281 if playable_list and 0 <= index < len(playable_list):
1282 playable = playable_list[index]
1283 title = playable.get("title")
1284 if title:
1285 meta.title = title
1286 cover = playable.get("cover_url_optional")
1287 if cover and not cover.startswith("http"):
1288 cover = f"https://{cover}"
1289 if cover:
1290 # Replace %% placeholder with size
1291 cover = cover.replace("%%", "400x400")
1292 meta.image_url = cover
1293
1294 async def _update_metadata_from_stream(
1295 self, stream_details: StreamDetails, seek_ms: int = 0
1296 ) -> None:
1297 """Update AudioSource metadata from stream details (authoritative for duration)."""
1298 meta = self._stream_metadata
1299 if stream_details.duration:
1300 meta.duration = stream_details.duration
1301 self._actual_duration_ms = stream_details.duration * 1000
1302 # Push the real duration to Ynison so the YM app shows
1303 # the correct value (we send duration_ms=0 on advance to
1304 # prevent stale propagation, so this corrects it).
1305 if self._ynison:
1306 await self._send_progress_to_ynison(
1307 progress_ms=seek_ms,
1308 duration_ms=self._actual_duration_ms,
1309 paused=self._ynison.state.is_paused,
1310 )
1311 meta.elapsed_time = seek_ms // 1000 if seek_ms else 0
1312 meta.elapsed_time_last_updated = time.time()
1313 # `trigger_player_update` expects a player_id; `_in_use_by_player` is
1314 # a queue identifier which only happens to coincide with player_id
1315 # when there is no protocol bridge. Use `_active_player_id` â the
1316 # real player wrapping our stream (bridge if any).
1317 if self._active_player_id:
1318 self.mass.players.trigger_player_update(self._active_player_id, force_update=True)
1319
1320 async def _send_progress_to_ynison(
1321 self,
1322 progress_ms: int,
1323 duration_ms: int,
1324 paused: bool,
1325 *,
1326 strict: bool = False,
1327 ) -> None:
1328 """
1329 Send progress to Ynison.
1330
1331 Progress is clamped to duration because Ynison rejects updates where
1332 progress > duration (error 400030001) and disconnects the WebSocket.
1333 The byte counter can slightly overshoot duration at end-of-stream.
1334
1335 Echo detection is done upstream via YnisonState.last_update_is_echo,
1336 which is set when Ynison rebroadcasts an update we authored.
1337
1338 :param progress_ms: Current playback position in milliseconds.
1339 :param duration_ms: Current track duration in milliseconds.
1340 :param paused: Whether playback is paused.
1341 :param strict: When ``True``, propagate transport failures as
1342 :class:`provider.ynison_client.YnisonSendError`. Used by user-command
1343 and end-of-track callers. Heartbeat callers leave the default.
1344 """
1345 if duration_ms <= 0:
1346 # Ynison rejects progress > duration; skip until duration is known.
1347 return
1348 if not self._ynison or not self._ynison.connected:
1349 if strict:
1350 raise YnisonSendError("Ynison not connected")
1351 return
1352 progress_ms = min(progress_ms, duration_ms)
1353 await self._ynison.update_playing_status(
1354 progress_ms=progress_ms,
1355 duration_ms=duration_ms,
1356 paused=paused,
1357 strict=strict,
1358 )
1359
1360 def _bytes_to_ms(self, byte_count: int, fmt: AudioFormat | None = None) -> int:
1361 """Convert PCM byte count to milliseconds using the given format."""
1362 bps = (fmt or self._normalized_format).pcm_sample_size
1363 if bps == 0:
1364 return 0
1365 return (byte_count * 1000) // bps
1366
1367 async def _sync_progress(
1368 self,
1369 seek_ms: int,
1370 bytes_yielded: int,
1371 player_id: str | None,
1372 fmt: AudioFormat | None = None,
1373 ) -> None:
1374 """Push real playback progress to MA metadata and Ynison."""
1375 elapsed_ms = seek_ms + self._bytes_to_ms(bytes_yielded, fmt)
1376 self._streaming_progress_ms = elapsed_ms
1377 # Update MA metadata
1378 meta = self._stream_metadata
1379 if meta:
1380 meta.elapsed_time = elapsed_ms // 1000
1381 meta.elapsed_time_last_updated = time.time()
1382 if player_id:
1383 self.mass.players.trigger_player_update(player_id)
1384 # Update Ynison so the Yandex app shows correct position
1385 await self._send_progress_to_ynison(
1386 progress_ms=elapsed_ms,
1387 duration_ms=self._best_duration_ms(),
1388 paused=False,
1389 )
1390
1391 async def _pause_playback(self) -> None:
1392 """
1393 Release the active player on external pause.
1394
1395 ``cmd_stop`` is the only mechanism that flips ``PlaybackState``
1396 to IDLE for an AudioSource queue item; ``cmd_pause`` and
1397 ``queue.pause`` both short-circuit back to ``on_source_control``
1398 and leave MA's state untouched. Pattern matches upstream
1399 ``AriaCastReceiver._handle_playback_state_update``. Resume
1400 re-runs ``play_media`` (preload + ffmpeg startup) so it costs
1401 a few seconds â the alternative kept resume instant but left
1402 MA's UI stuck on PLAYING.
1403 """
1404 target = self._in_use_by_player
1405 if not target:
1406 self.logger.info("Pause requested but no active queue (_in_use_by_player is None)")
1407 return
1408 self.logger.info("Pause: cmd_stop(%s)", target)
1409 # stop event ends the audio generator; finally clears the lock.
1410 self._stream_stop_event.set()
1411 try:
1412 await self.mass.players.cmd_stop(target)
1413 except Exception:
1414 # cmd_stop is the only mechanism that flips MA's PlaybackState
1415 # to IDLE for an AudioSource. A silent failure here resurrects
1416 # the very UX bug this code path exists to fix.
1417 self.logger.warning(
1418 "cmd_stop(%s) failed during external pause â MA UI may stay PLAYING",
1419 target,
1420 exc_info=True,
1421 )
1422 return
1423 # Demote `_active_player_id` from the bridge MA streams to
1424 # (e.g. `spb_*`) back to the queue id; queues live on the bare
1425 # UUID. Without this, resume's `play_media(_active_player_id,
1426 # â¦)` would target the bridge and raise
1427 # `PlayerUnavailableError`. Post-success only so a failure
1428 # path keeps the bridge id intact for the next attempt.
1429 self._active_player_id = target
1430 self._externally_paused = True
1431
1432 # ------------------------------------------------------------------
1433 # Player selection
1434 # ------------------------------------------------------------------
1435
1436 async def _on_connected_player_event(self, event: MassEvent) -> None:
1437 """Reload the provider when the connected player's display name changed."""
1438 del event
1439 if self._advertised_name is None or self._display_name == self._advertised_name:
1440 return
1441 self.logger.info(
1442 "Connected player was renamed; reloading to re-advertise as '%s'",
1443 self._display_name,
1444 )
1445 task_id = f"load_provider_{self.instance_id}"
1446 self.mass.call_later(1, self.mass.load_provider_config, self.config, task_id=task_id)
1447
1448 @property
1449 def _display_name(self) -> str:
1450 """Return the advertised device name: the connected player's display name."""
1451 if player := self.mass.players.get_player(self._default_player_id):
1452 return player.display_name
1453 # on a cold boot the player registers after this provider connects, so fall
1454 # back to its stored config name (the name sticks for the whole connection)
1455 stored_name = self.mass.config.get_raw_player_config_value(
1456 self._default_player_id, "name"
1457 ) or self.mass.config.get_raw_player_config_value(self._default_player_id, "default_name")
1458 return str(stored_name) if stored_name else DEFAULT_DISPLAY_NAME
1459
1460 def _get_target_player_id(self) -> str | None:
1461 """Determine the target player ID for playback."""
1462 # If there's an active player, validate it still exists
1463 if self._active_player_id:
1464 if self.mass.players.get_player(self._active_player_id):
1465 return self._active_player_id
1466 self._active_player_id = None
1467
1468 # Configured player (mandatory; enforced at load)
1469 if self.mass.players.get_player(self._default_player_id):
1470 return self._default_player_id
1471
1472 self.logger.warning(
1473 "Configured default player '%s' no longer exists",
1474 self._default_player_id,
1475 )
1476 return None
1477
1478 def _session_lost(self, player_id: str, session_id: str | None) -> bool:
1479 """
1480 Return ``True`` when our claim no longer matches the live session.
1481
1482 :param player_id: Queue id captured at generator entry.
1483 :param session_id: ``_active_session_id`` captured at generator entry.
1484 """
1485 return self._in_use_by_player != player_id or self._active_session_id != session_id
1486
1487 def _idempotent(self, action: str, key: str | None) -> bool:
1488 """
1489 Return ``True`` if ``(action, key)`` was not seen within the TTL window.
1490
1491 :param action: A short string identifying the command kind.
1492 :param key: Sub-key inside the action namespace, or ``None``.
1493 """
1494 now = time.monotonic()
1495 for stale_key in [
1496 k for k, ts in self._command_idempotency.items() if now - ts > _COMMAND_IDEMPOTENCY_TTL
1497 ]:
1498 self._command_idempotency.pop(stale_key, None)
1499 composite = (action, key)
1500 last = self._command_idempotency.get(composite)
1501 if last is not None and now - last < _COMMAND_IDEMPOTENCY_TTL:
1502 return False
1503 self._command_idempotency[composite] = now
1504 return True
1505
1506 @staticmethod
1507 def _classify_drift(
1508 ynison_ms: int,
1509 our_ms: int,
1510 threshold_ms: int = 3000,
1511 ) -> Literal["ignore", "queue_rebuild", "seek"]:
1512 """
1513 Classify drift between Ynison-reported and our local position.
1514
1515 Returns one of:
1516
1517 - ``"ignore"`` â drift at or below ``threshold_ms``; no seek needed.
1518 - ``"queue_rebuild"`` â Ynison reports near-zero progress while we
1519 are past 5s into the track; treat as a RADIO queue-rebuild echo,
1520 not a user seek (otherwise we'd yank playback to the start every
1521 time the rotor station refills the queue).
1522 - ``"seek"`` â genuine drift; honor it.
1523
1524 :param ynison_ms: Position reported by Ynison in milliseconds.
1525 :param our_ms: Position tracked locally in milliseconds.
1526 :param threshold_ms: Minimum drift to consider non-ignorable.
1527 """
1528 drift = abs(ynison_ms - our_ms)
1529 if drift <= threshold_ms:
1530 return "ignore"
1531 if ynison_ms < 1000 and our_ms > 5000:
1532 return "queue_rebuild"
1533 return "seek"
1534
1535 async def _prefetch_format_for_track(self, track_id: str) -> None:
1536 """
1537 Pre-fetch stream details for *track_id* and adapt PCM format.
1538
1539 Best-effort: bounded by ``_PREFETCH_FORMAT_TIMEOUT`` so a slow Yandex
1540 API does not stall ``_activate_playback``. On timeout / error the
1541 current format stays in place and the in-stream
1542 ``_get_stream_details_with_retry`` handles retries.
1543
1544 :param track_id: Yandex Music track id to query.
1545 """
1546 if not self._yandex_provider:
1547 return
1548 try:
1549 stream_details = await asyncio.wait_for(
1550 self._get_stream_details_with_retry(track_id),
1551 timeout=_PREFETCH_FORMAT_TIMEOUT,
1552 )
1553 except TimeoutError:
1554 self.logger.info(
1555 "Pre-fetch of stream details for %s exceeded %.1fs â "
1556 "keeping current format; in-stream fetch will retry",
1557 track_id,
1558 _PREFETCH_FORMAT_TIMEOUT,
1559 )
1560 return
1561 except Exception:
1562 self.logger.warning(
1563 "Pre-fetch of stream details failed for %s â keeping current format",
1564 track_id,
1565 exc_info=True,
1566 )
1567 return
1568 old_sr = self._normalized_params.get("sample_rate")
1569 old_bd = self._normalized_params.get("bit_depth")
1570 self._update_normalized_format(hint=stream_details.audio_format)
1571 new_sr = self._normalized_params.get("sample_rate")
1572 new_bd = self._normalized_params.get("bit_depth")
1573 if (old_sr, old_bd) != (new_sr, new_bd):
1574 self.logger.info(
1575 "Pre-fetch adapted format for %s: %dHz/%dbit -> %dHz/%dbit (source=%s)",
1576 track_id,
1577 old_sr or 0,
1578 old_bd or 0,
1579 new_sr or 0,
1580 new_bd or 0,
1581 stream_details.audio_format,
1582 )
1583
1584 def _clear_active_player(self) -> None:
1585 """Clear the active player and reset plugin state."""
1586 prev_player_id = self._active_player_id
1587 # the owner is the user-facing MA player; _active_player_id can be the protocol
1588 # player that consumed the stream, which is not what holds the source session
1589 owner_player_id = self._in_use_by_player
1590 source_session = (
1591 self.mass.players.get_audio_source_session(owner_player_id) if owner_player_id else None
1592 )
1593 self._active_player_id = None
1594 self._in_use_by_player = None
1595 self._active_session_id = None
1596 self._stream_stop_event.set()
1597 self._streaming_progress_ms = 0
1598 self._prefetched_list = None
1599 self._command_idempotency.clear()
1600 self._externally_paused = False
1601 if self._prefetch_task and not self._prefetch_task.done():
1602 self._prefetch_task.cancel()
1603
1604 if prev_player_id:
1605 self.logger.debug(
1606 "Playback ended on player %s, clearing active player",
1607 prev_player_id,
1608 )
1609 if owner_player_id:
1610 # give the source back as well as stopping: a session left on the player
1611 # keeps it publishing this source, so its own queue stays unreachable
1612 self.mass.create_task(
1613 self.mass.players.deselect_source(
1614 owner_player_id,
1615 provider_instance_id=self.instance_id,
1616 source_id=AUDIO_SOURCE_ID,
1617 playback_session_id=(
1618 source_session.playback_session_id if source_session else None
1619 ),
1620 )
1621 )
1622 self.mass.players.trigger_player_update(prev_player_id)
1623
1624 # ------------------------------------------------------------------
1625 # Yandex Music provider matching
1626 # ------------------------------------------------------------------
1627
1628 def _on_provider_event(self, event: MassEvent) -> None:
1629 """Handle provider added/removed events."""
1630 self.mass.create_task(self._check_yandex_provider_match())
1631
1632 async def _check_yandex_provider_match(self) -> None:
1633 """
1634 Check if a Yandex Music provider is available for audio streaming.
1635
1636 In borrow mode (self._ym_instance_id set), match strictly by instance_id
1637 so that audio and credentials come from the same account. In own mode,
1638 accept any yandex_music music-provider (prior behavior).
1639 """
1640 for provider in self.mass.get_providers():
1641 if provider.domain != "yandex_music" or provider.type != ProviderType.MUSIC:
1642 continue
1643 if self._ym_instance_id is not None and provider.instance_id != self._ym_instance_id:
1644 continue
1645 self.logger.debug("Found Yandex Music provider â enabling playback control")
1646 self._yandex_provider = cast("YandexMusicProviderLike", provider)
1647 self._update_normalized_format()
1648 self._update_source_capabilities()
1649 return
1650
1651 if self._yandex_provider is not None:
1652 self.logger.debug(
1653 "Yandex Music provider no longer available â disabling playback control"
1654 )
1655 self._yandex_provider = None
1656 self._update_source_capabilities()
1657
1658 def _snap_rate_to_player(self, rate: int) -> int:
1659 """
1660 Snap *rate* down to the nearest sample rate the target player accepts.
1661
1662 Best-effort: returns *rate* unchanged when no target player or
1663 supported-rate set can be resolved, and never raises.
1664
1665 :param rate: The sample rate the hint / floor logic chose.
1666 :return: A rate the target player can play (``rate`` itself when it is
1667 already supported or no player is resolvable).
1668 """
1669 # Mirror MA's _select_audio_source_pcm_format so the declared format
1670 # equals what the AudioSource passthrough picks â keeping MA off its
1671 # second resampling ffmpeg.
1672 try:
1673 player_id = self._get_target_player_id()
1674 if not player_id:
1675 return rate
1676 player = self.mass.players.get_player(player_id)
1677 if player is None:
1678 return rate
1679 supported = [sr for sr, _ in player.get_supported_sample_rates()]
1680 if not supported or rate in supported:
1681 return rate
1682 return max((r for r in supported if r <= rate), default=min(supported))
1683 except Exception:
1684 self.logger.debug(
1685 "Could not snap sample rate to player capabilities; keeping %d Hz",
1686 rate,
1687 exc_info=True,
1688 )
1689 return rate
1690
1691 def _update_normalized_format(self, hint: AudioFormat | None = None) -> None:
1692 """
1693 Set PCM normalization profile based on config and YM quality.
1694
1695 Priority: explicit config values > hint from real stream_details >
1696 auto-detection from YM quality. The hint is fed by
1697 ``_prefetch_format_for_track`` when ``CONF_OUTPUT_SAMPLE_RATE`` is
1698 ``auto`` so the AudioSource ``provider_mapping.audio_format`` matches
1699 the actual source rate of the upcoming track. Without a hint, falls
1700 back to YM-quality-based detection (superb/lossless â 24bit/44.1kHz,
1701 else â 16bit/44.1kHz). The resulting auto/hint rate is then snapped
1702 down to the nearest rate the target player supports; a valid explicit
1703 override is delivered verbatim and never snapped.
1704
1705 Creates fresh AudioFormat instances each time to prevent mutation by
1706 MA's FFMpeg._log_reader_task (which sets input_format.codec_type
1707 in-place on the object passed as input_format to the outer ffmpeg).
1708
1709 :param hint: Optional real source AudioFormat (from a stream-details
1710 pre-fetch). Lifts auto mode from the quality-based default to the
1711 track's actual sample rate and bit depth.
1712 """
1713 # Start with auto-detected base from YM quality config
1714 # (yandex_music does not expose get_quality(); read from its ProviderConfig instead)
1715 quality = ""
1716 if self._yandex_provider is not None:
1717 provider_config = getattr(self._yandex_provider, "config", None)
1718 if provider_config is not None and hasattr(provider_config, "get_value"):
1719 config_quality = provider_config.get_value(YANDEX_MUSIC_CONF_QUALITY)
1720 if isinstance(config_quality, str):
1721 quality = config_quality
1722 is_lossless = quality in YANDEX_MUSIC_LOSSLESS_QUALITIES
1723 base = dict(PCM_LOSSLESS_PARAMS if is_lossless else PCM_LOSSY_PARAMS)
1724 # Promote auto-base from the real stream details when available.
1725 # Validate the hint against the same allow-lists we use for explicit
1726 # config overrides â a Yandex API hiccup that returns an unsupported
1727 # rate (or 0) must not poison the AudioSource provider_mapping or the
1728 # outer ffmpeg input_format.
1729 if hint is not None:
1730 if hint.sample_rate and str(hint.sample_rate) in _VALID_SAMPLE_RATES:
1731 base["sample_rate"] = hint.sample_rate
1732 if hint.bit_depth and str(hint.bit_depth) in _VALID_BIT_DEPTHS:
1733 base["bit_depth"] = hint.bit_depth
1734
1735 # Apply config overrides. MA's ConfigEntry options constrain the UI to
1736 # known-good strings, but a stale persisted value or hand-edited config
1737 # could still surface something unparsable or off-list â fall back to
1738 # the auto-detected base with a warning instead of crashing the load.
1739 sample_rate = base["sample_rate"]
1740 bit_depth = base["bit_depth"]
1741 explicit_rate = False
1742 if self._cfg_sample_rate != OUTPUT_AUTO:
1743 if self._cfg_sample_rate in _VALID_SAMPLE_RATES:
1744 sample_rate = int(self._cfg_sample_rate)
1745 explicit_rate = True
1746 else:
1747 self.logger.warning(
1748 "Invalid %s=%r; falling back to auto-detected %d Hz",
1749 CONF_OUTPUT_SAMPLE_RATE,
1750 self._cfg_sample_rate,
1751 sample_rate,
1752 )
1753 # Snap the auto / hint / floor rate to a value the target player accepts
1754 # so the declared format matches what MA's AudioSource passthrough picks
1755 # and no second resampling ffmpeg is spawned. A valid explicit override
1756 # is delivered verbatim and is never snapped.
1757 if not explicit_rate:
1758 sample_rate = self._snap_rate_to_player(sample_rate)
1759 if self._cfg_bit_depth != OUTPUT_AUTO:
1760 if self._cfg_bit_depth in _VALID_BIT_DEPTHS:
1761 bit_depth = int(self._cfg_bit_depth)
1762 else:
1763 self.logger.warning(
1764 "Invalid %s=%r; falling back to auto-detected %d-bit",
1765 CONF_OUTPUT_BIT_DEPTH,
1766 self._cfg_bit_depth,
1767 bit_depth,
1768 )
1769
1770 content_type = ContentType.PCM_S24LE if bit_depth == 24 else ContentType.PCM_S16LE
1771 new_params: dict[str, Any] = {
1772 "content_type": content_type,
1773 "sample_rate": sample_rate,
1774 "bit_depth": bit_depth,
1775 "channels": 2,
1776 }
1777
1778 # Warn if format changes while a player is actively streaming â the
1779 # active session keeps using its frozen snapshot; the new format takes
1780 # effect on the next session.
1781 old = self._normalized_params
1782 if self._in_use_by_player and (
1783 old.get("content_type") != content_type
1784 or old.get("sample_rate") != sample_rate
1785 or old.get("bit_depth") != bit_depth
1786 ):
1787 self.logger.warning(
1788 "Normalization format changed while streaming â new format "
1789 "(%s/%dHz/%dbit) will apply on next session",
1790 content_type.value,
1791 sample_rate,
1792 bit_depth,
1793 )
1794
1795 self._normalized_params = new_params
1796 # Fresh copy for each caller so no shared mutable state
1797 self._normalized_format = make_pcm_format(self._normalized_params)
1798 # rebuild the AudioSource so its ProviderMapping carries the new audio_format
1799 self._audio_source = self._build_audio_source()
1800 self.logger.debug(
1801 "Normalization format: %s/%dHz/%dbit",
1802 self._normalized_format.content_type.value,
1803 self._normalized_format.sample_rate,
1804 self._normalized_format.bit_depth,
1805 )
1806
1807 def _update_source_capabilities(self) -> None:
1808 """Rebuild AudioSource so capability flags reflect linked provider availability."""
1809 self._audio_source = self._build_audio_source()
1810 # The session publishes the controls from the object it holds, so hand it the
1811 # rebuilt one: the new capability flags reach the UI without waiting for the
1812 # source to be selected again.
1813 if not self._in_use_by_player:
1814 return
1815 self.mass.players.refresh_source(self._in_use_by_player, self._audio_source)
1816
1817 def _build_audio_source(self) -> AudioSource:
1818 """Construct the AudioSource MediaItem with current capability flags."""
1819 has_provider = self._yandex_provider is not None
1820 return AudioSource(
1821 item_id=AUDIO_SOURCE_ID,
1822 provider=self.instance_id,
1823 name=self.name,
1824 provider_mappings={
1825 ProviderMapping(
1826 item_id=AUDIO_SOURCE_ID,
1827 provider_domain=self.domain,
1828 provider_instance=self.instance_id,
1829 # Fresh AudioFormat copy â `self._normalized_format` is a
1830 # shared mutable that MA's ffmpeg sets `codec_type` on
1831 # in-place. Sharing it would let that mutation leak into
1832 # the rebuilt AudioSource and any future stream-details.
1833 audio_format=make_pcm_format(self._normalized_params),
1834 )
1835 },
1836 can_play_pause=has_provider,
1837 can_seek=has_provider,
1838 can_next_previous=has_provider,
1839 exclusive=True,
1840 allow_external_trigger=True,
1841 )
1842
1843 # ------------------------------------------------------------------
1844 # Playback control callbacks
1845 # ------------------------------------------------------------------
1846
1847 def _best_duration_ms(self) -> int:
1848 """Return the best known duration: actual from stream, or Ynison state as fallback."""
1849 if self._actual_duration_ms > 0:
1850 return self._actual_duration_ms
1851 if self._ynison:
1852 return self._ynison.state.duration_ms
1853 return 0
1854
1855 def _require_connected_ynison(self) -> YnisonClient:
1856 """
1857 Return the live Ynison client or raise an MA player-control error.
1858
1859 :raises UnsupportedFeaturedException: When the provider's Ynison
1860 client has not been initialised yet (pre-`handle_async_init`
1861 or post-`unload`).
1862 :raises PlayerCommandFailed: When the Ynison WebSocket is currently
1863 disconnected (e.g. mid-reconnect after a transient network
1864 error). Surface to MA so the UI shows a clear failure toast
1865 instead of accepting the command and stalling.
1866 """
1867 if not self._ynison:
1868 raise UnsupportedFeaturedException("Ynison client not initialized")
1869 if not self._ynison.connected:
1870 raise PlayerCommandFailed("Ynison WebSocket disconnected")
1871 return self._ynison
1872
1873 async def _on_play(self) -> None:
1874 """Handle play command â send resume to Ynison."""
1875 client = self._require_connected_ynison()
1876 if not self._idempotent("on_play", None):
1877 return
1878 state = client.state
1879 try:
1880 await self._send_progress_to_ynison(
1881 progress_ms=state.progress_ms,
1882 duration_ms=self._best_duration_ms(),
1883 paused=False,
1884 strict=True,
1885 )
1886 except YnisonSendError as exc:
1887 raise PlayerCommandFailed("Ynison send failed") from exc
1888
1889 async def _on_pause(self) -> None:
1890 """Handle pause command â send pause to Ynison."""
1891 client = self._require_connected_ynison()
1892 if not self._idempotent("on_pause", None):
1893 return
1894 state = client.state
1895 try:
1896 await self._send_progress_to_ynison(
1897 progress_ms=state.progress_ms,
1898 duration_ms=self._best_duration_ms(),
1899 paused=True,
1900 strict=True,
1901 )
1902 except YnisonSendError as exc:
1903 raise PlayerCommandFailed("Ynison send failed") from exc
1904
1905 # Entity types that use server-side "radio" queue replenishment.
1906 # Currently only RADIO (personal wave, genre stations).
1907 # Add "WAVE" here if/when Yandex supports it via the same
1908 # rotor_station_tracks API.
1909 _RADIO_ENTITY_TYPES: ClassVar[set[str]] = {"RADIO"}
1910
1911 def _maybe_prefetch(
1912 self,
1913 current_index: int,
1914 playable_list: list[dict[str, Any]],
1915 entity_id: str,
1916 entity_type: str,
1917 ) -> None:
1918 """Kick off background prefetch when nearing the end of the queue."""
1919 if entity_type not in self._RADIO_ENTITY_TYPES:
1920 return
1921 if not self._yandex_provider or not playable_list:
1922 return
1923 # second-to-last or last â trigger prefetch near end of queue
1924 if current_index < len(playable_list) - 2:
1925 return
1926 # Already prefetched or prefetch in progress
1927 if self._prefetched_list is not None:
1928 return
1929 if self._prefetch_task and not self._prefetch_task.done():
1930 return
1931
1932 self.logger.info(
1933 "Pre-fetching tracks (at index %d/%d, entity=%s)",
1934 current_index,
1935 len(playable_list),
1936 entity_id[:40] if entity_id else "<none>",
1937 )
1938
1939 async def _do_prefetch() -> None:
1940 result = await self._replenish_radio_queue(entity_id, entity_type, playable_list)
1941 if result:
1942 self._prefetched_list = result
1943 # Push expanded queue to Ynison immediately so the YM app
1944 # sees upcoming tracks and enables the "next" button.
1945 await self._update_queue_list(result)
1946
1947 self._prefetch_task = self.mass.create_task(_do_prefetch())
1948
1949 async def _signal_track_completion(self) -> None:
1950 """
1951 Signal that the current track finished playing.
1952
1953 Ynison is a state-sync protocol â the active device must advance
1954 current_playable_index itself.
1955
1956 If the next index is within the playable list, we advance immediately.
1957 If we're at the end (typical for RADIO/wave with short queues),
1958 we fetch more tracks via the Yandex Music API, append them to the
1959 playable_list, and then advance.
1960 """
1961 if not self._ynison:
1962 return
1963 state = self._ynison.state
1964 duration = self._best_duration_ms()
1965 queue = state.player_state.get("player_queue", {})
1966 current_index = queue.get("current_playable_index", 0)
1967 playable_list = queue.get("playable_list", [])
1968 entity_type = queue.get("entity_type", "")
1969 entity_id = queue.get("entity_id", "")
1970 next_index = current_index + 1
1971
1972 self.logger.info(
1973 "Track finished at index %d/%d (entity=%s type=%s), "
1974 "advancing to index %d (duration=%dms)",
1975 current_index,
1976 len(playable_list),
1977 entity_id[:40] if entity_id else "<none>",
1978 entity_type,
1979 next_index,
1980 duration,
1981 )
1982 self._actual_duration_ms = 0
1983
1984 # 1. Report that playback reached the end.
1985 # Echo tracking is handled by _send_progress_to_ynison.
1986 # `strict=True`: a dropped end-of-track signal stalls the YM app on
1987 # the just-finished track. We log and continue â the reconnect is
1988 # already scheduled and the queue-advance below sees the same WS state
1989 # â but we don't reraise (this is end-of-stream, there's no command to
1990 # fail back to the user).
1991 try:
1992 await self._send_progress_to_ynison(
1993 progress_ms=duration, duration_ms=duration, paused=False, strict=True
1994 )
1995 except YnisonSendError:
1996 self.logger.warning(
1997 "Track-completion signal dropped (Ynison transport failure); "
1998 "queue advance will retry once the WS reconnects",
1999 exc_info=True,
2000 )
2001
2002 if next_index < len(playable_list):
2003 # 2a. Queue has room â advance immediately.
2004 # Clear stale prefetch data so _maybe_prefetch can trigger for
2005 # the new queue tail on subsequent state updates.
2006 self._prefetched_list = None
2007 await self._advance_queue_index(next_index)
2008 elif entity_type in self._RADIO_ENTITY_TYPES:
2009 # 2b. At end of RADIO queue â use prefetched data or fetch now
2010 expanded: list[dict[str, Any]] | None = None
2011 if self._prefetched_list:
2012 self.logger.info("Using pre-fetched queue (%d items)", len(self._prefetched_list))
2013 expanded = self._prefetched_list
2014 self._prefetched_list = None
2015 elif self._prefetch_task and not self._prefetch_task.done():
2016 self.logger.info("Waiting for in-flight prefetch...")
2017 await self._prefetch_task
2018 expanded = self._prefetched_list
2019 self._prefetched_list = None
2020 else:
2021 expanded = await self._replenish_radio_queue(entity_id, entity_type, playable_list)
2022 if expanded and next_index < len(expanded):
2023 await self._advance_queue_index(next_index, expanded_list=expanded)
2024 elif expanded:
2025 self.logger.warning(
2026 "Expanded queue has %d items but next_index=%d â re-fetching",
2027 len(expanded),
2028 next_index,
2029 )
2030 fresh = await self._replenish_radio_queue(entity_id, entity_type, expanded)
2031 if fresh and next_index < len(fresh):
2032 await self._advance_queue_index(next_index, expanded_list=fresh)
2033 else:
2034 self.logger.warning("Still cannot advance after re-fetch")
2035 else:
2036 self.logger.warning(
2037 "Could not replenish queue (entity=%s type=%s), cannot advance",
2038 entity_id,
2039 entity_type,
2040 )
2041 else:
2042 self.logger.info(
2043 "End of non-radio queue (entity=%s type=%s), playback complete",
2044 entity_id[:40] if entity_id else "<none>",
2045 entity_type,
2046 )
2047
2048 async def _replenish_radio_queue(
2049 self,
2050 entity_id: str,
2051 entity_type: str,
2052 playable_list: list[dict[str, Any]],
2053 ) -> list[dict[str, Any]] | None:
2054 """
2055 Fetch more tracks from Yandex Music API and return expanded playable_list.
2056
2057 The active device is responsible for replenishing RADIO/wave queues.
2058 Ynison only syncs state â it does NOT generate new tracks.
2059 """
2060 if not self._yandex_provider:
2061 self.logger.warning("No yandex_music provider available for radio replenishment")
2062 return None
2063
2064 # Determine the last track ID for pagination
2065 last_track_id: str | None = None
2066 if playable_list:
2067 last_track_id = playable_list[-1].get("playable_id")
2068
2069 self.logger.info(
2070 "Fetching more tracks for %s station %s (queue=%s)",
2071 entity_type,
2072 entity_id,
2073 last_track_id,
2074 )
2075
2076 try:
2077 tracks, batch_id = await self._yandex_provider.get_rotor_station_tracks(
2078 entity_id, queue=last_track_id
2079 )
2080 except Exception:
2081 self.logger.exception("Failed to fetch radio tracks for %s", entity_id)
2082 return None
2083
2084 if not tracks:
2085 self.logger.warning("No tracks returned for station %s", entity_id)
2086 return None
2087
2088 # Determine the 'from' field from existing items
2089 from_field = ""
2090 if playable_list:
2091 from_field = playable_list[0].get("from", "")
2092
2093 # Convert tracks to Ynison playable_list format
2094 new_items: list[dict[str, Any]] = []
2095 for track in tracks:
2096 album_id = ""
2097 if hasattr(track, "albums") and track.albums:
2098 album_id = str(track.albums[0].id) if track.albums[0].id else ""
2099 cover = ""
2100 if hasattr(track, "cover_uri") and track.cover_uri:
2101 cover = track.cover_uri
2102 new_items.append(
2103 {
2104 "playable_id": str(track.id),
2105 "album_id_optional": album_id,
2106 "playable_type": "TRACK",
2107 "from": from_field,
2108 "title": track.title or "",
2109 "cover_url_optional": cover,
2110 }
2111 )
2112
2113 self.logger.info(
2114 "Fetched %d new tracks for station %s (batch=%s)",
2115 len(new_items),
2116 entity_id,
2117 batch_id,
2118 )
2119
2120 return list(playable_list) + new_items
2121
2122 async def _advance_queue_index(
2123 self,
2124 next_index: int,
2125 *,
2126 expanded_list: list[dict[str, Any]] | None = None,
2127 ) -> None:
2128 """
2129 Send update_player_state to advance the queue to next_index.
2130
2131 If expanded_list is provided, it replaces the playable_list
2132 (used after radio queue replenishment).
2133
2134 Waits up to 10 s for reconnection if Ynison is temporarily
2135 disconnected (e.g. after a transient error).
2136 """
2137 if not self._ynison:
2138 return
2139 if not self._ynison.connected:
2140 self.logger.info("Waiting for Ynison reconnection before advancing queueâ¦")
2141 for _ in range(10):
2142 await asyncio.sleep(1)
2143 if not self._ynison or self._ynison.connected:
2144 break
2145 if not self._ynison or not self._ynison.connected:
2146 self.logger.warning("Cannot advance queue â Ynison still disconnected")
2147 return
2148 state = self._ynison.state
2149 queue = state.player_state.get("player_queue", {})
2150 device_id = self._ynison.device_id
2151 new_state = dict(state.player_state)
2152 new_state["player_queue"] = dict(queue)
2153 new_state["player_queue"]["current_playable_index"] = next_index
2154 new_state["player_queue"]["version"] = make_version_block(device_id)
2155 if expanded_list is not None:
2156 new_state["player_queue"]["playable_list"] = expanded_list
2157 new_state["status"] = dict(new_state.get("status", {}))
2158 new_state["status"]["progress_ms"] = "0"
2159 new_state["status"]["duration_ms"] = "0"
2160 new_state["status"]["paused"] = False
2161 new_state["status"]["version"] = make_version_block(device_id)
2162 # `strict=True`: a dropped queue-advance leaves `_wait_for_track_change`
2163 # spinning for its full 30 s timeout. Log and return â the next
2164 # reconnect-broadcast picks up our authored version block and resyncs.
2165 try:
2166 await self._ynison.update_player_state(player_state=new_state, strict=True)
2167 except YnisonSendError:
2168 self.logger.warning(
2169 "Queue-advance dropped (Ynison transport failure); "
2170 "stream will stall until reconnect-broadcast resyncs",
2171 exc_info=True,
2172 )
2173
2174 async def _update_queue_list(self, expanded_list: list[dict[str, Any]]) -> None:
2175 """
2176 Push an expanded playable_list to Ynison without changing index or progress.
2177
2178 Called right after prefetch completes so the YM app sees upcoming
2179 tracks and enables the "next" button.
2180 """
2181 if not self._ynison or not self._ynison.connected:
2182 return
2183 state = self._ynison.state
2184 queue = state.player_state.get("player_queue", {})
2185 device_id = self._ynison.device_id
2186 new_state = dict(state.player_state)
2187 new_state["player_queue"] = dict(queue)
2188 new_state["player_queue"]["playable_list"] = expanded_list
2189 new_state["player_queue"]["version"] = make_version_block(device_id)
2190 await self._ynison.update_player_state(player_state=new_state)
2191
2192 async def _on_next(self) -> None:
2193 """Handle next track command â signal track end so Yandex advances."""
2194 self._require_connected_ynison()
2195 await self._signal_track_completion()
2196
2197 async def _on_previous(self) -> None:
2198 """Handle previous track command â update queue index in Ynison."""
2199 client = self._require_connected_ynison()
2200 queue = client.state.player_state.get("player_queue", {})
2201 current_index = queue.get("current_playable_index", 0)
2202 if current_index > 0:
2203 self._actual_duration_ms = 0
2204 await self._advance_queue_index(current_index - 1)
2205
2206 async def _on_seek(self, position: int) -> None:
2207 """
2208 Handle seek command â send position update to Ynison.
2209
2210 :param position: Position in seconds from Music Assistant.
2211 """
2212 client = self._require_connected_ynison()
2213 seek_ms = position * 1000
2214 state = client.state
2215 try:
2216 await self._send_progress_to_ynison(
2217 progress_ms=seek_ms,
2218 duration_ms=self._best_duration_ms(),
2219 paused=state.is_paused,
2220 strict=True,
2221 )
2222 except YnisonSendError as exc:
2223 # Do not mutate `_seek_position_ms` / `_seek_grace_until` on failure
2224 # â local stream state must not drift past a send that never landed.
2225 raise PlayerCommandFailed("Ynison send failed") from exc
2226 # Also trigger local stream restart so seek takes effect
2227 # immediately without waiting for the Ynison echo.
2228 self._seek_position_ms = seek_ms
2229 self._seek_grace_until = time.monotonic() + _ECHO_GRACE_PERIOD
2230 self._track_changed_event.set()
2231