/
/
/
1"""Ynison WebSocket client for Yandex Music device synchronization."""
2
3from __future__ import annotations
4
5import asyncio
6import json
7import logging
8import random
9import secrets
10import time
11import uuid
12from collections.abc import Awaitable, Callable
13from contextlib import suppress
14from dataclasses import asdict, dataclass, field
15from typing import TYPE_CHECKING, Any
16
17import aiohttp
18from music_assistant_models.errors import LoginFailed
19
20if TYPE_CHECKING:
21 from ya_passport_auth import SecretStr
22
23from .constants import (
24 DEFAULT_APP_NAME,
25 DEFAULT_APP_VERSION,
26 DEVICE_TYPE_WEB,
27 RECONNECT_DELAYS,
28 WS_CONNECT_TIMEOUT,
29 WS_HEARTBEAT,
30 YNISON_ORIGIN,
31 YNISON_RECONNECT_ERROR_CODES,
32 YNISON_REDIRECT_URL,
33 YNISON_STATE_PATH,
34)
35
36
37class YnisonSendError(ConnectionError):
38 """
39 Raised by `YnisonClient._send(strict=True)` when a send cannot reach Ynison.
40
41 Indicates a transport-level failure (WebSocket not connected, write raised
42 ``ConnectionError`` / ``aiohttp.ClientError`` / ``RuntimeError`` / ``OSError``).
43 A reconnect is always scheduled before this is raised; callers should
44 translate it to the appropriate user-facing error (e.g.
45 ``PlayerCommandFailed``) or log-and-return for fire-and-forget paths.
46
47 Inherits from ``ConnectionError`` so existing broad transport-error
48 handlers continue to catch it.
49 """
50
51
52def make_version_block(device_id: str) -> dict[str, Any]:
53 """
54 Build a version sub-object authored by the given device.
55
56 Ynison expects string types for version and timestamp fields;
57 passing integers triggers 500 responses that terminate the WebSocket.
58 """
59 return {
60 "device_id": device_id,
61 "version": str(time.time_ns()),
62 "timestamp_ms": "0",
63 }
64
65
66def _stringify_version(version: Any) -> None:
67 """Coerce int `version.version`/`version.timestamp_ms` fields to str in-place."""
68 if not isinstance(version, dict):
69 return
70 for key in ("version", "timestamp_ms"):
71 val = version.get(key)
72 if isinstance(val, int) and not isinstance(val, bool):
73 version[key] = str(val)
74
75
76def normalize_player_state_timestamps(player_state: dict[str, Any]) -> None:
77 """
78 Coerce Ynison timestamp fields to strings in-place.
79
80 Ynison rejects integer `status.progress_ms`/`duration_ms`/`version.*`
81 (HTTP 500 + WS teardown), so we normalize inbound state at the ingestion
82 boundary. This guarantees that every outbound echo â whether via
83 `update_player_state` (from shallow-copied state) or `send_full_state`
84 on reconnect â carries string-typed timestamps by construction.
85 """
86 status = player_state.get("status")
87 if isinstance(status, dict):
88 for key in ("progress_ms", "duration_ms", "player_action_timestamp_ms"):
89 val = status.get(key)
90 if isinstance(val, int) and not isinstance(val, bool):
91 status[key] = str(val)
92 _stringify_version(status.get("version"))
93 queue = player_state.get("player_queue")
94 if isinstance(queue, dict):
95 _stringify_version(queue.get("version"))
96
97
98@dataclass
99class YnisonDeviceInfo:
100 """Device identification for Ynison registration."""
101
102 device_id: str
103 title: str
104 type: str = DEVICE_TYPE_WEB
105 app_name: str = DEFAULT_APP_NAME
106 app_version: str = DEFAULT_APP_VERSION
107
108
109@dataclass
110class YnisonState:
111 """Parsed Ynison state from the server."""
112
113 player_state: dict[str, Any] = field(default_factory=dict)
114 active_device_id: str | None = None
115 devices: list[dict[str, Any]] = field(default_factory=list)
116 # True iff the most recent state update carried a version block
117 # (on player_queue or status) authored by our own device_id â i.e.
118 # it is Ynison echoing back an update we originated. Consumers can
119 # inspect this to suppress feedback loops. False when no authored
120 # version block is present (e.g. status-only update from a peer
121 # that did not round-trip via our device).
122 last_update_is_echo: bool = False
123
124 @property
125 def current_track_id(self) -> str | None:
126 """Extract current track_id from player queue."""
127 queue = self.player_state.get("player_queue", {})
128 playable_list = queue.get("playable_list", [])
129 index = queue.get("current_playable_index", 0)
130 if playable_list and 0 <= index < len(playable_list):
131 playable_id = playable_list[index].get("playable_id")
132 if playable_id:
133 return str(playable_id)
134 return None
135
136 @property
137 def is_paused(self) -> bool:
138 """Return True if playback is paused."""
139 return bool(self.player_state.get("status", {}).get("paused", True))
140
141 @property
142 def progress_ms(self) -> int:
143 """Return current playback progress in milliseconds."""
144 return int(self.player_state.get("status", {}).get("progress_ms", 0))
145
146 @property
147 def duration_ms(self) -> int:
148 """Return current track duration in milliseconds."""
149 return int(self.player_state.get("status", {}).get("duration_ms", 0))
150
151
152# Type alias for the state update callback
153StateUpdateCallback = Callable[[YnisonState], Awaitable[None]]
154# Callback invoked on auth failure; should return a fresh token (or raise).
155AuthRefreshCallback = Callable[[], Awaitable["SecretStr"]]
156
157
158class YnisonClient:
159 """
160 WebSocket client for the Yandex Ynison protocol.
161
162 Manages the two-step connection (redirector â state service) and
163 provides methods to send state updates back to Ynison.
164 """
165
166 def __init__(
167 self,
168 token: SecretStr,
169 device_info: YnisonDeviceInfo,
170 on_state_update: StateUpdateCallback,
171 logger: logging.Logger,
172 http_session: aiohttp.ClientSession | None = None,
173 on_auth_failure: AuthRefreshCallback | None = None,
174 ) -> None:
175 """
176 Initialize Ynison client.
177
178 :param token: Yandex Music OAuth token (wrapped in SecretStr).
179 :param device_info: Device identification for Ynison.
180 :param on_state_update: Callback for state updates from Ynison.
181 :param logger: Logger instance.
182 :param http_session: Optional shared aiohttp session.
183 :param on_auth_failure: Optional callback invoked on auth failure during
184 reconnect. Should return a fresh SecretStr token. If not provided or
185 if the callback raises, reconnect proceeds with the current token.
186 """
187 self._token = token
188 self._device_info = device_info
189 self._on_state_update = on_state_update
190 self._logger = logger
191 self._external_session = http_session
192 self._on_auth_failure = on_auth_failure
193
194 self._ws: aiohttp.ClientWebSocketResponse | None = None
195 self._session: aiohttp.ClientSession | None = None
196 self._send_lock = asyncio.Lock()
197 self._message_task: asyncio.Task[None] | None = None
198 self._reconnect_task: asyncio.Task[None] | None = None
199 self._stop_event = asyncio.Event()
200 self._connected = False
201 self._has_connected_once = False
202
203 # Latest state from server
204 self.state = YnisonState()
205
206 # Reconnect settle window â first inbound state after reconnect can
207 # be our own stale broadcast (server retains state across reconnects
208 # and re-sends it). Provider-level handlers consult this watermark
209 # to discard the first â¤2s of post-reconnect state changes.
210 self._post_reconnect_settle_until: float = 0.0
211
212 @property
213 def connected(self) -> bool:
214 """Return True if connected to Ynison state service."""
215 return self._connected
216
217 @property
218 def in_post_reconnect_settle(self) -> bool:
219 """
220 True iff we're inside the 2 s post-reconnect settle window.
221
222 Provider handlers consult this to skip the first inbound state right
223 after a reconnect â that state can be a stale broadcast of our own
224 last-known view (server retained it across the WS hop) and acting on
225 it would re-fire pause/play commands the user never issued.
226 """
227 return time.monotonic() < self._post_reconnect_settle_until
228
229 @property
230 def device_id(self) -> str:
231 """Return our Ynison device_id (used when authoring outgoing state)."""
232 return self._device_info.device_id
233
234 async def connect(self) -> None:
235 """
236 Connect to Ynison (redirector â state service).
237
238 Raises on auth failure; auto-reconnects on transient errors.
239 """
240 self._stop_event.clear()
241 if self._external_session and self._external_session.closed:
242 raise RuntimeError("Provided http_session is closed")
243 self._session = self._external_session or aiohttp.ClientSession()
244
245 try:
246 # Step 1: Get redirect ticket
247 host, ticket, session_id = await self._get_redirect_ticket()
248
249 # Step 2: Connect to state service
250 await self._connect_state(host, ticket, session_id)
251 except LoginFailed:
252 await self.disconnect()
253 raise
254 except asyncio.CancelledError:
255 await self.disconnect()
256 raise
257 except Exception:
258 # Transient error â schedule reconnect instead of dying
259 self._logger.warning("Initial connection failed, scheduling reconnect", exc_info=True)
260 self._connected = False
261 if self._ws and not self._ws.closed:
262 await self._ws.close()
263 self._ws = None
264 if self._session and not self._external_session:
265 await self._session.close()
266 self._session = None
267 self._schedule_reconnect()
268
269 async def disconnect(self) -> None:
270 """Gracefully disconnect from Ynison."""
271 self._stop_event.set()
272 self._connected = False
273
274 if self._message_task and not self._message_task.done():
275 self._message_task.cancel()
276 with suppress(asyncio.CancelledError):
277 await self._message_task
278
279 if self._reconnect_task and not self._reconnect_task.done():
280 self._reconnect_task.cancel()
281 with suppress(asyncio.CancelledError):
282 await self._reconnect_task
283
284 if self._ws and not self._ws.closed:
285 await self._ws.close()
286 self._ws = None
287
288 if self._session and not self._external_session:
289 await self._session.close()
290 self._session = None
291
292 def update_token(self, token: SecretStr) -> None:
293 """Replace the stored OAuth token (e.g. after a refresh)."""
294 self._token = token
295
296 # ------------------------------------------------------------------
297 # Send methods
298 # ------------------------------------------------------------------
299
300 async def update_playing_status(
301 self,
302 progress_ms: int,
303 duration_ms: int,
304 paused: bool,
305 *,
306 strict: bool = False,
307 ) -> None:
308 """
309 Send playback status update to Ynison.
310
311 :param progress_ms: Current playback position in milliseconds.
312 :param duration_ms: Current track duration in milliseconds.
313 :param paused: Whether playback is paused.
314 :param strict: When ``True``, raise :class:`YnisonSendError` on
315 transport failure instead of silently scheduling a reconnect.
316 Delivery-critical callers (user commands, end-of-track signal)
317 opt in; heartbeat callers leave the default.
318 """
319 self._logger.debug(
320 "â update_playing_status: progress=%dms duration=%dms paused=%s",
321 progress_ms,
322 duration_ms,
323 paused,
324 )
325 msg = {
326 "update_playing_status": {
327 "playing_status": {
328 "progress_ms": str(progress_ms),
329 "duration_ms": str(duration_ms),
330 "paused": paused,
331 "playback_speed": 1.0,
332 },
333 },
334 }
335 await self._send(msg, strict=strict)
336
337 async def update_active_device(self, device_id: str) -> None:
338 """Request playback transfer to this device."""
339 msg = {
340 "update_active_device": {
341 "device_id_optional": device_id,
342 },
343 }
344 await self._send(msg)
345
346 async def update_session_params(self, mute_events_if_passive: bool = True) -> None:
347 """
348 Configure session params on the Ynison server.
349
350 `mute_events_if_passive=True` tells Ynison not to forward peer
351 state updates while we're not the active device. Reduces inbound
352 WS noise (and CPU) when running in `borrow` mode alongside other
353 active subscribers, and removes a class of false positives in
354 echo detection â fewer messages means fewer chances to misclassify.
355 """
356 msg = {
357 "update_session_params": {
358 "mute_events_if_passive": mute_events_if_passive,
359 },
360 }
361 self._logger.info(
362 "â update_session_params: mute_events_if_passive=%s", mute_events_if_passive
363 )
364 await self._send(msg)
365
366 async def sync_state_from_eov(self, actual_queue_id: str = "") -> None:
367 """
368 Request queue sync from the EOV (Unified Playback Queue) backend.
369
370 Asks the Ynison server to refresh the queue from the central EOV service.
371 Only works when this device is the active player. If the EOV queue
372 differs from actual_queue_id, the server broadcasts the updated state.
373
374 :param actual_queue_id: Current queue ID (empty string forces refresh).
375 """
376 msg = {
377 "sync_state_from_eov": {
378 "actual_queue_id": actual_queue_id,
379 },
380 **self._message_meta(),
381 }
382 self._logger.info("â sync_state_from_eov: queue_id=%r", actual_queue_id)
383 await self._send(msg)
384
385 async def update_player_state(
386 self,
387 player_state: dict[str, Any],
388 *,
389 strict: bool = False,
390 ) -> None:
391 """
392 Send player state update (queue changes, track skip).
393
394 Unlike send_full_state, this does NOT reset active device status.
395 Use this for track advances, queue modifications, repeat/shuffle changes.
396
397 :param player_state: Complete `player_state` dict to broadcast.
398 :param strict: When ``True``, raise :class:`YnisonSendError` on
399 transport failure instead of silently scheduling a reconnect.
400 Delivery-critical callers (queue advance after track end)
401 opt in; queue-list-replenish heartbeats leave the default.
402 """
403 queue = player_state.get("player_queue", {})
404 self._logger.info(
405 "â update_player_state: index=%s queue_len=%d entity_type=%s",
406 queue.get("current_playable_index"),
407 len(queue.get("playable_list", [])),
408 queue.get("entity_type", ""),
409 )
410 msg = {
411 "update_player_state": {
412 "player_state": player_state,
413 },
414 **self._message_meta(),
415 }
416 self._logger.debug("Sending player state: %s", json.dumps(msg)[:500])
417 await self._send(msg, strict=strict)
418
419 async def send_full_state(
420 self,
421 player_state: dict[str, Any] | None = None,
422 ) -> None:
423 """Send full state update (cold start, reconnect after offline)."""
424 state = player_state or self._build_initial_state()
425 msg = {
426 "update_full_state": {
427 "player_state": state,
428 "device": self._build_device_dict(),
429 "is_currently_active": False,
430 },
431 **self._message_meta(),
432 }
433 self._logger.debug("Sending full state: %s", json.dumps(msg)[:500])
434 await self._send(msg)
435
436 @staticmethod
437 def _message_meta() -> dict[str, Any]:
438 """
439 Return common envelope fields for state-mutating messages.
440
441 Ynison expects string-typed timestamps; integers cause 500 responses.
442 """
443 return {
444 "rid": str(uuid.uuid4()),
445 "player_action_timestamp_ms": str(int(time.time() * 1000)),
446 "activity_interception_type": "DO_NOT_INTERCEPT_BY_DEFAULT",
447 }
448
449 def _classify_state_as_echo(self, incoming_ps: dict[str, Any]) -> bool:
450 """
451 Return True iff `incoming_ps` is our own broadcast round-tripping.
452
453 Uses author check on BOTH queue.version.device_id and
454 status.version.device_id â only an update where every block was
455 authored by us is treated as echo. AND-logic is critical: a peer
456 queue change combined with our own status echo would otherwise
457 be silently swallowed (RC-1 in v1.9.1 live testing).
458
459 Why only `device_id` and not `version` value: Ynison's protobuf
460 comment marks `version.version` as `random(int64)`. The server
461 re-stamps it after every `update_playing_status` (we send blank,
462 server fills in). Comparing inbound `version` against an outbound
463 watermark is therefore meaningless â our own restamped echo can
464 carry any value. `device_id` is unique and preserved end-to-end,
465 so authorship is the only reliable echo signal.
466 """
467 own_id = self._device_info.device_id
468 queue_block = (incoming_ps.get("player_queue") or {}).get("version") or {}
469 status_block = (incoming_ps.get("status") or {}).get("version") or {}
470 queue_is_ours = queue_block.get("device_id") == own_id
471 status_is_ours = status_block.get("device_id") == own_id
472 return queue_is_ours and status_is_ours
473
474 # ------------------------------------------------------------------
475 # Connection internals
476 # ------------------------------------------------------------------
477
478 def _build_ws_protocol_header(
479 self,
480 redirect_ticket: str | None = None,
481 session_id: int | None = None,
482 ) -> str:
483 """Build Sec-WebSocket-Protocol header value."""
484 proto: dict[str, Any] = {
485 "Ynison-Device-Id": self._device_info.device_id,
486 "Ynison-Device-Info": json.dumps({"app_name": self._device_info.app_name, "type": 1}),
487 }
488 if redirect_ticket is not None:
489 proto["Ynison-Redirect-Ticket"] = redirect_ticket
490 if session_id is not None:
491 proto["Ynison-Session-Id"] = str(session_id)
492 return f"Bearer, v2, {json.dumps(proto)}"
493
494 def _build_headers(
495 self,
496 redirect_ticket: str | None = None,
497 session_id: int | None = None,
498 ) -> dict[str, str]:
499 """Build common WebSocket headers."""
500 return {
501 "Authorization": f"OAuth {self._token.get_secret()}",
502 "Origin": YNISON_ORIGIN,
503 "Sec-WebSocket-Protocol": self._build_ws_protocol_header(redirect_ticket, session_id),
504 }
505
506 def _build_device_dict(self) -> dict[str, Any]:
507 """Build device info dict for Ynison messages."""
508 info = asdict(self._device_info)
509 return {
510 "info": info,
511 "capabilities": {
512 "can_be_player": True,
513 "can_be_remote_controller": False,
514 },
515 "is_shadow": False,
516 }
517
518 def _build_initial_state(self) -> dict[str, Any]:
519 """Build initial player state (paused, empty queue)."""
520 device_id = self._device_info.device_id
521 return {
522 "status": {
523 "paused": True,
524 "duration_ms": "0",
525 "progress_ms": "0",
526 "playback_speed": 1,
527 "version": make_version_block(device_id),
528 },
529 "player_queue": {
530 "current_playable_index": -1,
531 "entity_id": "",
532 "entity_type": "VARIOUS",
533 "playable_list": [],
534 "options": {"repeat_mode": "NONE"},
535 "entity_context": "BASED_ON_ENTITY_BY_DEFAULT",
536 "version": make_version_block(device_id),
537 "from_optional": "",
538 },
539 }
540
541 async def _get_redirect_ticket(self) -> tuple[str, str, int]:
542 """
543 Connect to redirector and obtain redirect ticket.
544
545 :return: (host, redirect_ticket, session_id)
546 :raises LoginFailed: If authentication fails.
547 """
548 if self._session is None:
549 raise RuntimeError("HTTP session not initialized â call connect() first")
550 headers = self._build_headers()
551
552 ws_timeout = aiohttp.ClientWSTimeout(ws_close=WS_CONNECT_TIMEOUT)
553 try:
554 ws = await self._session.ws_connect(
555 YNISON_REDIRECT_URL,
556 headers=headers,
557 timeout=ws_timeout,
558 )
559 except aiohttp.WSServerHandshakeError as err:
560 if err.status in (401, 403):
561 raise LoginFailed("Ynison authentication failed â invalid token") from err
562 raise
563
564 try:
565 msg = await ws.receive(timeout=WS_CONNECT_TIMEOUT)
566 if msg.type in (aiohttp.WSMsgType.TEXT, aiohttp.WSMsgType.BINARY):
567 data = json.loads(msg.data)
568 else:
569 raise ConnectionError(f"Unexpected message type from redirector: {msg.type}")
570 finally:
571 await ws.close()
572
573 host = data.get("host", "")
574 ticket = data.get("redirect_ticket", "")
575 session_id = int(data.get("session_id", 0))
576
577 if not host or not ticket:
578 raise ConnectionError("Redirector response missing host or ticket")
579
580 self._logger.debug("Ynison redirect: host=%s, session_id=%d", host, session_id)
581 return host, ticket, session_id
582
583 async def _connect_state(self, host: str, ticket: str, session_id: int) -> None:
584 """Connect to Ynison state service and start message loop."""
585 if self._session is None:
586 raise RuntimeError("HTTP session not initialized â call connect() first")
587 url = f"wss://{host}{YNISON_STATE_PATH}"
588 headers = self._build_headers(redirect_ticket=ticket, session_id=session_id)
589
590 ws_timeout = aiohttp.ClientWSTimeout(ws_close=WS_CONNECT_TIMEOUT)
591 try:
592 self._ws = await self._session.ws_connect(
593 url, headers=headers, timeout=ws_timeout, heartbeat=WS_HEARTBEAT
594 )
595 except aiohttp.WSServerHandshakeError as err:
596 if err.status in (401, 403):
597 raise LoginFailed("Ynison authentication failed â invalid token") from err
598 raise
599 self._connected = True
600 self._logger.info("Connected to Ynison state service at %s", host)
601
602 # Always send a fresh initial state (empty/paused) â both on cold
603 # start and reconnect (v2.0). The previous behaviour replayed
604 # `self.state.player_state`, which after a heartbeat could carry
605 # `paused=True` and trigger an unintended pause on the still-running
606 # player when Ynison broadcast it back to us.
607 # If a player is already active (handoff in progress), the provider
608 # will reclaim ownership via `update_active_device` after the
609 # post-reconnect settle window expires.
610 if self._has_connected_once:
611 self._logger.info("Reconnect: sending fresh initial state (no stale replay)")
612 self._post_reconnect_settle_until = time.monotonic() + 2.0
613 await self.send_full_state()
614 # Best-effort: ask the server not to forward peer events while we
615 # are passive. Failure is non-fatal â we just receive more events.
616 try:
617 await self.update_session_params(mute_events_if_passive=True)
618 except Exception:
619 self._logger.debug("update_session_params failed", exc_info=True)
620
621 self._has_connected_once = True
622
623 # Start message loop
624 self._message_task = asyncio.create_task(self._message_loop())
625
626 async def _message_loop(self) -> None: # noqa: PLR0915
627 """Read messages from state service and dispatch callbacks."""
628 if self._ws is None:
629 raise RuntimeError("WebSocket not connected â call connect() first")
630 try:
631 async for msg in self._ws:
632 if self._stop_event.is_set():
633 break
634
635 if msg.type == aiohttp.WSMsgType.ERROR:
636 msg_data_preview = str(self._ws.exception())
637 elif not msg.data:
638 msg_data_preview = "<empty>"
639 elif isinstance(msg.data, str):
640 msg_data_preview = msg.data[:500]
641 elif isinstance(msg.data, bytes):
642 msg_data_preview = msg.data[:500].decode(errors="replace")
643 else:
644 msg_data_preview = str(msg.data)
645
646 self._logger.debug(
647 "Ynison msg type=%s, data=%s",
648 msg.type,
649 msg_data_preview,
650 )
651
652 if msg.type == aiohttp.WSMsgType.TEXT:
653 try:
654 data = json.loads(msg.data)
655 except json.JSONDecodeError:
656 self._logger.warning(
657 "Failed to parse Ynison message: %s",
658 msg.data[:200] if msg.data else "<empty>",
659 )
660 continue
661
662 if "error" in data:
663 error_info = data["error"]
664 error_code = error_info.get("details", {}).get("ynison-error-code", "")
665 self._logger.warning(
666 "Ynison error response: %s",
667 json.dumps(error_info)[:300],
668 )
669 if error_code in YNISON_RECONNECT_ERROR_CODES:
670 self._logger.info(
671 "Ynison re-balance error %s â breaking for immediate reconnect",
672 error_code,
673 )
674 break
675 continue
676
677 self._parse_state(data)
678 try:
679 await self._on_state_update(self.state)
680 except Exception:
681 self._logger.exception("Error in Ynison state update callback")
682 elif msg.type == aiohttp.WSMsgType.BINARY:
683 self._logger.debug(
684 "Ynison binary message (%d bytes)", len(msg.data) if msg.data else 0
685 )
686 elif msg.type == aiohttp.WSMsgType.ERROR:
687 self._logger.warning("Ynison WebSocket error: %s", self._ws.exception())
688 break
689 elif msg.type in (
690 aiohttp.WSMsgType.CLOSE,
691 aiohttp.WSMsgType.CLOSING,
692 aiohttp.WSMsgType.CLOSED,
693 ):
694 self._logger.debug(
695 "Ynison WS close: type=%s, close_code=%s, extra=%s",
696 msg.type,
697 self._ws.close_code,
698 msg.extra,
699 )
700 break
701 except asyncio.CancelledError:
702 return
703 except Exception:
704 self._logger.exception("Unexpected error in Ynison message loop")
705 self._logger.debug("Ynison message loop exited")
706
707 self._connected = False
708
709 if not self._stop_event.is_set() and (
710 self._reconnect_task is None or self._reconnect_task.done()
711 ):
712 self._logger.warning("Ynison connection lost, scheduling reconnect")
713 self._schedule_reconnect()
714
715 def _parse_state(self, data: dict[str, Any]) -> None:
716 """Parse PutYnisonStateResponse into YnisonState."""
717 old_track = self.state.current_track_id
718 old_index = self.state.player_state.get("player_queue", {}).get(
719 "current_playable_index", -1
720 )
721
722 # Replace each incoming player_state sub-object at the top level:
723 # Ynison sends entries like "player_queue" and "status" as complete
724 # objects, so merging nested dicts would retain stale keys that are
725 # absent from the update.
726 incoming_ps = data.get("player_state")
727 if incoming_ps is not None:
728 # Normalize timestamp fields before storing: Ynison rejects int
729 # `status.progress_ms`/`duration_ms`/`version.*` on outbound
730 # messages, and stored state is round-tripped via send_full_state
731 # (on reconnect) and update_player_state (on queue edits).
732 normalize_player_state_timestamps(incoming_ps)
733 existing_ps = self.state.player_state
734 for key, value in incoming_ps.items():
735 existing_ps[key] = value
736 # Echo detection: a state is our echo iff BOTH the queue and
737 # status version-blocks are authored by our `device_id`. See
738 # `_classify_state_as_echo` for why version values are not
739 # part of the check (Ynison documents `version.version` as
740 # `random(int64)` and the server re-stamps it).
741 self.state.last_update_is_echo = self._classify_state_as_echo(incoming_ps)
742 else:
743 self.state.last_update_is_echo = False
744 self.state.active_device_id = data.get(
745 "active_device_id_optional", self.state.active_device_id
746 )
747 self.state.devices = data.get("devices", self.state.devices)
748
749 new_track = self.state.current_track_id
750 queue = self.state.player_state.get("player_queue", {})
751 new_index = queue.get("current_playable_index", -1)
752 queue_len = len(queue.get("playable_list", []))
753 entity_type = queue.get("entity_type", "")
754
755 if old_track != new_track or old_index != new_index:
756 self._logger.info(
757 "Ynison queue change: track %sâ%s index %dâ%d queue_len=%d entity_type=%s",
758 old_track,
759 new_track,
760 old_index,
761 new_index,
762 queue_len,
763 entity_type,
764 )
765 else:
766 self._logger.debug(
767 "Ynison state update (no queue change): track=%s index=%d progress=%dms paused=%s",
768 new_track,
769 new_index,
770 self.state.progress_ms,
771 self.state.is_paused,
772 )
773
774 async def _reconnect(self) -> None:
775 """
776 Reconnect with exponential backoff, retrying indefinitely.
777
778 On authentication failure (LoginFailed), attempts to refresh the token
779 via the on_auth_failure callback before the next retry. The loop only
780 exits when `_stop_event` is set (via disconnect()) or on successful
781 reconnection; a reliable long-running plugin never permanently gives up.
782 """
783 attempt = 0
784 while not self._stop_event.is_set():
785 delay = RECONNECT_DELAYS[min(attempt, len(RECONNECT_DELAYS) - 1)]
786 # Add ±20% jitter to prevent thundering-herd reconnects
787 jitter = delay * 0.2 * (2 * random.random() - 1)
788 delay = max(0.5, delay + jitter)
789 self._logger.info("Ynison reconnect attempt %d in %.1fs", attempt + 1, delay)
790 await asyncio.sleep(delay)
791
792 if self._stop_event.is_set():
793 return
794
795 attempt += 1
796 try:
797 # Close stale WebSocket
798 if self._ws and not self._ws.closed:
799 await self._ws.close()
800 self._ws = None
801
802 # Re-create session if needed
803 if self._session is None or self._session.closed:
804 if self._external_session is not None:
805 if self._external_session.closed:
806 msg = "External HTTP session is closed"
807 raise RuntimeError(msg)
808 self._session = self._external_session
809 else:
810 self._session = aiohttp.ClientSession()
811
812 host, ticket, session_id = await self._get_redirect_ticket()
813 await self._connect_state(host, ticket, session_id)
814 self._logger.info("Ynison reconnected successfully")
815 return
816 except LoginFailed:
817 self._logger.warning("Ynison reconnect attempt %d failed: auth error", attempt)
818 if self._on_auth_failure:
819 try:
820 new_token = await self._on_auth_failure()
821 self._token = new_token
822 self._logger.info("Token refreshed, will retry with new token")
823 except Exception:
824 self._logger.warning("Token refresh failed", exc_info=True)
825 except asyncio.CancelledError:
826 return
827 except Exception:
828 self._logger.warning("Ynison reconnect attempt %d failed", attempt, exc_info=True)
829
830 async def _send(self, msg: dict[str, Any], *, strict: bool = False) -> None:
831 """
832 Send a JSON message to the state service (thread-safe).
833
834 :param msg: JSON-serialisable Ynison envelope.
835 :param strict: When ``True``, transport failures (disconnected socket
836 or write error) raise :class:`YnisonSendError` after scheduling a
837 reconnect. Default is the legacy fire-and-forget behaviour:
838 log + schedule reconnect + return.
839 """
840 async with self._send_lock:
841 if self._ws is None or self._ws.closed:
842 self._logger.debug("Cannot send to Ynison â not connected")
843 if strict:
844 raise YnisonSendError("Ynison WebSocket not connected")
845 return
846 try:
847 await self._ws.send_str(json.dumps(msg))
848 except (ConnectionError, aiohttp.ClientError, RuntimeError, OSError) as exc:
849 self._logger.warning("Failed to send message to Ynison, scheduling reconnect")
850 self._connected = False
851 self._schedule_reconnect()
852 if strict:
853 raise YnisonSendError("Ynison send failed") from exc
854
855 def _schedule_reconnect(self) -> None:
856 """
857 Schedule a background reconnect attempt if none is already in flight.
858
859 Idempotent: a single reconnect task is in flight at any time. Becomes
860 a no-op once :meth:`disconnect` has set ``_stop_event``.
861 """
862 if self._stop_event.is_set():
863 return
864 if self._reconnect_task is not None and not self._reconnect_task.done():
865 return
866 self._reconnect_task = asyncio.create_task(self._reconnect())
867
868
869def generate_device_id() -> str:
870 """Generate a 16-character hex device ID for Ynison registration."""
871 return secrets.token_hex(8)
872