music-assistant-server

1"""Sendspin Player implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from collections.abc import Callable
8from contextlib import suppress
9from io import BytesIO
10from typing import TYPE_CHECKING, ClassVar, cast
11
12from aiosendspin.models import AudioCodec, MediaCommand
13from aiosendspin.models.management import (
14    ManagementSetPairingConfigPayload,
15    SetDynamicPinConfig,
16    SetStaticPinConfig,
17    SetUnpairedAccessConfig,
18)
19from aiosendspin.models.types import PairMethod, PlaybackStateType, PlayerCommand
20from aiosendspin.models.types import RepeatMode as SendspinRepeatMode
21from aiosendspin.models.visualizer import BeatAvailability, BeatTiming
22from aiosendspin.noise.driver import HandshakeAbortedError
23from aiosendspin.noise.pairing import PairingError
24from aiosendspin.noise.trust_store import PskCategory
25from aiosendspin.server import ClientEvent, GroupEvent, SendspinGroup, VolumeChangedEvent
26from aiosendspin.server.audio import AudioFormat as SendspinAudioFormat
27from aiosendspin.server.client import DisconnectBehaviour
28from aiosendspin.server.events import (
29    ClientGroupChangedEvent,
30    GroupDeletedEvent,
31    GroupMemberAddedEvent,
32    GroupMemberRemovedEvent,
33    GroupStateChangedEvent,
34)
35from aiosendspin.server.roles import (
36    ArtworkGroupRole,
37    ColorGroupRole,
38    ControllerEvent,
39    ControllerGroupRole,
40    ControllerNextEvent,
41    ControllerPauseEvent,
42    ControllerPlayEvent,
43    ControllerPreviousEvent,
44    ControllerRepeatEvent,
45    ControllerSeekEvent,
46    ControllerSeekRelativeEvent,
47    ControllerShuffleEvent,
48    ControllerStopEvent,
49    MetadataGroupRole,
50    VisualizerGroupRole,
51)
52from aiosendspin.server.roles.color.state import Color
53from aiosendspin.server.roles.metadata.state import Metadata
54from aiosendspin.server.roles.player.events import StaticDelayChangedEvent
55from aiosendspin.server.roles.player.types import PlayerRoleProtocol
56from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
57from music_assistant_models.constants import PLAYER_CONTROL_NONE
58from music_assistant_models.enums import (
59    ConfigEntryType,
60    IdentifierType,
61    MediaType,
62    PlaybackState,
63    PlayerFeature,
64    PlayerType,
65    ProviderType,
66    RepeatMode,
67)
68from music_assistant_models.errors import MediaNotFoundError, PlayerCommandFailed
69from music_assistant_models.media_items import Album, Artist, is_track
70from music_assistant_models.player import DeviceInfo
71from PIL import Image
72
73from music_assistant.constants import HIDDEN_ANNOUNCE_VOLUME_CONFIG_ENTRIES
74from music_assistant.controllers.streams.audio_analysis import SMART_FADES_ANALYSIS_DOMAIN
75from music_assistant.helpers.util import is_valid_mac_address, join_task
76from music_assistant.models.player import Player, PlayerMedia
77from music_assistant.models.setup_flow import AbortFlow, StepExpiredError
78
79from .constants import (
80    BRIDGE_PREFIX,
81    CONF_ACTION_MANAGEMENT_DYNAMIC_PIN_DISABLE,
82    CONF_ACTION_MANAGEMENT_DYNAMIC_PIN_ENABLE,
83    CONF_ACTION_MANAGEMENT_ENTER,
84    CONF_ACTION_MANAGEMENT_EXIT,
85    CONF_ACTION_MANAGEMENT_STATIC_PIN_DISABLE,
86    CONF_ACTION_MANAGEMENT_STATIC_PIN_ENABLE,
87    CONF_ACTION_MANAGEMENT_UNPAIRED_DISABLE,
88    CONF_ACTION_MANAGEMENT_UNPAIRED_ENABLE,
89    CONF_ACTION_REVOKE_UNPAIRED,
90    CONF_ACTION_UNPAIR,
91    CONF_CAST_AUDIO_UNSUPPORTED,
92    CONF_PAIRING_METHOD,
93    CONF_PAIRING_PIN,
94    CONF_PAIRING_TOKEN,
95    CONF_SENDSPIN_STATIC_DELAY,
96    DEFAULT_SENDSPIN_STATIC_DELAY,
97    PAIR_METHOD_PIN,
98    PAIR_METHOD_STATIC_PIN,
99    PAIR_METHOD_TOKEN,
100    PAIR_METHOD_UNPAIRED,
101)
102from .helpers import (
103    AlertText,
104    SecurityActionError,
105    action_entry,
106    alert_entry,
107    effective_pair_methods,
108    effective_unpaired_access,
109    error_alert,
110    mac_from_bridge_client_id,
111)
112from .playback import SendspinPlaybackSession
113
114# Supported group commands for Sendspin players
115SUPPORTED_GROUP_COMMANDS = [
116    MediaCommand.PLAY,
117    MediaCommand.PAUSE,
118    MediaCommand.STOP,
119    MediaCommand.NEXT,
120    MediaCommand.PREVIOUS,
121    MediaCommand.REPEAT_OFF,
122    MediaCommand.REPEAT_ONE,
123    MediaCommand.REPEAT_ALL,
124    MediaCommand.SHUFFLE,
125    MediaCommand.UNSHUFFLE,
126    MediaCommand.SEEK,
127    MediaCommand.SEEK_RELATIVE,
128]
129
130# Config constants for Sendspin audio format
131CONF_PREFERRED_SENDSPIN_FORMAT = "preferred_sendspin_format"
132SENDSPIN_FORMAT_AUTOMATIC = "automatic"
133
134
135def format_to_option_value(fmt: SupportedAudioFormat) -> str:
136    """Convert SupportedAudioFormat to "codec:sample_rate:bit_depth:channels"."""
137    return f"{fmt.codec.value}:{fmt.sample_rate}:{fmt.bit_depth}:{fmt.channels}"
138
139
140def option_value_to_format(value: str) -> tuple[AudioCodec, SendspinAudioFormat] | None:
141    """
142    Parse option value back to (AudioCodec, SendspinAudioFormat).
143
144    :param value: Option value in format "codec:sample_rate:bit_depth:channels".
145    :return: Tuple of (AudioCodec, SendspinAudioFormat) or None if parsing fails.
146    """
147    try:
148        codec_str, sample_rate_str, bit_depth_str, channels_str = value.split(":")
149        codec = AudioCodec(codec_str)
150        audio_format = SendspinAudioFormat(
151            sample_rate=int(sample_rate_str),
152            bit_depth=int(bit_depth_str),
153            channels=int(channels_str),
154        )
155        return (codec, audio_format)
156    except ValueError, KeyError:
157        return None
158
159
160def format_to_display_string(fmt: SupportedAudioFormat) -> str:
161    """Convert to display string like "FLAC 48kHz/24bit stereo"."""
162    codec_name = fmt.codec.name
163    sample_rate_khz = fmt.sample_rate / 1000
164    # Format sample rate: show as integer if whole number, otherwise one decimal
165    if sample_rate_khz == int(sample_rate_khz):
166        sample_rate_str = f"{int(sample_rate_khz)}kHz"
167    else:
168        sample_rate_str = f"{sample_rate_khz:.1f}kHz"
169    if fmt.channels == 2:
170        channels_str = "stereo"
171    elif fmt.channels == 1:
172        channels_str = "mono"
173    else:
174        channels_str = f"{fmt.channels}ch"
175    return f"{codec_name} {sample_rate_str}/{fmt.bit_depth}bit {channels_str}"
176
177
178_MANAGEMENT_ACTIONS = {
179    CONF_ACTION_MANAGEMENT_UNPAIRED_ENABLE: ManagementSetPairingConfigPayload(
180        unpaired_access=SetUnpairedAccessConfig(enabled=True)
181    ),
182    CONF_ACTION_MANAGEMENT_UNPAIRED_DISABLE: ManagementSetPairingConfigPayload(
183        unpaired_access=SetUnpairedAccessConfig(enabled=False)
184    ),
185    CONF_ACTION_MANAGEMENT_STATIC_PIN_ENABLE: ManagementSetPairingConfigPayload(
186        static_pin=SetStaticPinConfig(enabled=True)
187    ),
188    CONF_ACTION_MANAGEMENT_STATIC_PIN_DISABLE: ManagementSetPairingConfigPayload(
189        static_pin=SetStaticPinConfig(enabled=False)
190    ),
191    CONF_ACTION_MANAGEMENT_DYNAMIC_PIN_ENABLE: ManagementSetPairingConfigPayload(
192        dynamic_pin=SetDynamicPinConfig(enabled=True)
193    ),
194    CONF_ACTION_MANAGEMENT_DYNAMIC_PIN_DISABLE: ManagementSetPairingConfigPayload(
195        dynamic_pin=SetDynamicPinConfig(enabled=False)
196    ),
197}
198
199# Seconds the setup flow waits for the device to enter pairing mode (show its PIN).
200PAIR_GESTURE_TIMEOUT = 120.0
201# Seconds the setup flow waits for pairing to complete after the PIN is submitted.
202PAIR_CONFIRM_TIMEOUT = 30.0
203
204# Terminal pairing-error slugs that map to a dedicated setup_flow.abort reason;
205# anything else falls back to the generic "pairing_failed" abort.
206_PAIRING_ABORT_REASONS = {
207    "pairing_error_concurrent": "pairing_error_concurrent",
208    "pairing_error_locked_out": "pairing_error_locked_out",
209    "pairing_error_no_pin_method": "no_pair_methods",
210    "pairing_error_not_connected": "pairing_error_not_connected",
211}
212
213
214def _pin_error_slug(error: Exception | None) -> str:
215    """Return the strings.json errors slug for a retryable PIN failure (re-rendered form)."""
216    if error is None:
217        return "pairing_error_generic"
218    return error_alert(error).key
219
220
221def _pairing_abort_reason(error: Exception | None) -> str:
222    """Return the setup_flow.abort reason slug for a terminal pairing failure."""
223    if error is None:
224        return "pairing_failed"
225    key = error.alert_key if isinstance(error, SecurityActionError) else error_alert(error).key
226    return _PAIRING_ABORT_REASONS.get(key, "pairing_failed")
227
228
229if TYPE_CHECKING:
230    from aiosendspin.models.core import ClientHelloPayload
231    from aiosendspin.models.management import ManagementResultData, PairingMethodConfig
232    from aiosendspin.models.player import SupportedAudioFormat
233    from aiosendspin.noise.trust_store import ServerPairingRecord
234    from aiosendspin.server.client import SendspinClient
235    from music_assistant_models.media_items import MediaItemPalette
236    from music_assistant_models.player_queue import PlayerQueue
237    from music_assistant_models.queue_item import QueueItem
238
239    from music_assistant.controllers.player_queues.state import PlayerQueueData
240    from music_assistant.models.setup_flow import SetupSession
241    from music_assistant.providers.chromecast.sendspin_bridge import SendspinBridgeManager
242    from music_assistant.providers.hass import HomeAssistantProvider
243
244    from .provider import PinPairingSession, SendspinProvider
245
246
247class SendspinBasePlayer(Player):
248    """
249    Base class for Sendspin players in Music Assistant.
250
251    Provides shared device-info, group membership, and event handling logic
252    that is common to both audio and non-audio (visualizer) player types.
253    """
254
255    api: SendspinClient
256    unsub_event_cb: Callable[[], None] | None
257    unsub_group_event_cb: Callable[[], None] | None
258    is_web_player: bool = False
259    # transient alert produced by the most recent security action, surfaced on the next
260    # config-entries render (the settings page re-renders after invoking an action)
261    _pending_security_alert: AlertText | None = None
262
263    def __init__(
264        self,
265        provider: SendspinProvider,
266        player_id: str,
267        initial_hello: ClientHelloPayload | None = None,
268    ) -> None:
269        """
270        Initialize the base Sendspin player.
271
272        :param provider: The Sendspin provider instance.
273        :param player_id: The unique player identifier.
274        :param initial_hello: Optional hello payload from the client.
275        """
276        super().__init__(provider, player_id)
277        sendspin_client = provider.server_api.get_client(player_id)
278        assert sendspin_client is not None
279        self.api = sendspin_client
280        self.unsub_event_cb = None
281        self.unsub_group_event_cb = None
282        self.logger = self.provider.logger.getChild(player_id)
283        self._attr_can_group_with = {provider.instance_id}
284        self._attr_power_control = PLAYER_CONTROL_NONE
285        self._refresh_client_info(sendspin_client, hello_payload=initial_hello)
286        self._subscribe_client_callbacks()
287
288    def event_cb(self, client: SendspinClient, event: ClientEvent) -> None:
289        """Event callback registered to the sendspin client."""
290        match event:
291            case ClientGroupChangedEvent(new_group=new_group):
292                if self.unsub_group_event_cb is not None:
293                    self.unsub_group_event_cb()
294                self.unsub_group_event_cb = new_group.add_event_listener(self.group_event_cb)
295                self._on_group_changed(new_group)
296                self.update_state()
297
298    def group_event_cb(self, group: SendspinGroup, event: GroupEvent) -> None:
299        """Event callback registered to the sendspin group this player belongs to."""
300        if self.synced_to is not None:
301            # Only handle group events as the leader, except for:
302            # - GroupMemberRemovedEvent: to handle being removed from a group
303            # - GroupStateChangedEvent: to update playback state when leader stops/disconnects
304            if not isinstance(event, (GroupMemberRemovedEvent, GroupStateChangedEvent)):
305                return
306        match event:
307            case GroupStateChangedEvent(state=state):
308                match state:
309                    case PlaybackStateType.PLAYING:
310                        self._attr_playback_state = PlaybackState.PLAYING
311                    case PlaybackStateType.PAUSED:
312                        self._attr_playback_state = PlaybackState.PAUSED
313                    case PlaybackStateType.STOPPED:
314                        self._attr_playback_state = PlaybackState.IDLE
315                        self._attr_elapsed_time = 0
316                        self._attr_elapsed_time_last_updated = time.time()
317                        self._on_group_stopped()
318                self.update_state()
319            case GroupMemberAddedEvent(client_id=client_id):
320                is_group_leader = (
321                    bool(group.clients) and group.clients[0].client_id == self.player_id
322                )
323                if is_group_leader and (
324                    not self._attr_group_members or self._attr_group_members[0] != self.player_id
325                ):
326                    self._attr_group_members = [self.player_id, *self._attr_group_members]
327                if client_id not in self._attr_group_members:
328                    self._attr_group_members.append(client_id)
329                    self.update_state()
330                self._schedule_membership_sync(group)
331            case GroupMemberRemovedEvent(client_id=client_id):
332                self.mass.create_task(self._handle_group_member_removed(group, client_id))
333                self._schedule_membership_sync(group)
334            case GroupDeletedEvent():
335                pass
336
337    async def on_unload(self) -> None:
338        """Handle logic when the player is unloaded from the Player controller."""
339        await super().on_unload()
340        self._unsubscribe_client_callbacks()
341
342    @property
343    def needs_setup(self) -> bool:
344        """
345        Whether the device is connected and encrypted but not yet usable for playback.
346
347        An unpaired device that has not been paired or allowed unpaired playback still
348        connects, but the server activates no roles for it. Reporting needs_setup keeps it
349        out of the ready-to-play targets while its settings (and the pairing actions) stay
350        reachable. Legacy unencrypted devices and the built-in web player can play as-is.
351        """
352        if self._is_bridge_or_web_player:
353            return False
354
355        # Deliberately no is_connected gate: between a (re)connect's hello and its first
356        # client/state, security and active roles are already valid while is_connected is not.
357
358        if self.api.connection_security is None:
359            return False
360        return not self.api.active_roles
361
362    @property
363    def setup_reason(self) -> str | None:
364        """Return the reason this device needs setup (pairing), or None when it does not."""
365        return "pairing_required" if self.needs_setup else None
366
367    async def get_config_entries(self) -> list[ConfigEntry]:
368        """Return all (provider/player specific) Config Entries for the player."""
369        entries = await super().get_config_entries()
370        entries.extend(await self._get_security_config_entries())
371        return entries
372
373    async def handle_config_action(self, action: str) -> list[ConfigEntry]:
374        """Run an unpair/management action, then re-render the entries."""
375        if self.api.connection is not None:
376            provider = cast("SendspinProvider", self.provider)
377            alert = await self._handle_security_action(provider, action)
378            if alert is not None:
379                self._pending_security_alert = alert
380        return await self.get_config_entries()
381
382    async def run_setup_flow(self, session: SetupSession) -> None:
383        """
384        Drive pairing (or an unpaired-access grant) for this encrypted Sendspin device.
385
386        An unpaired device picks between the offered pair methods and - when the device
387        permits it - playback without pairing; re-running the flow on a paired device
388        verifies its presence via a dynamic PIN. Pairing succeeds as a side effect of
389        the provider pairing calls; the flow finishes with no persisted values. Bridge/
390        web players and unencrypted (legacy) connections have nothing to pair.
391
392        :param session: The setup flow session used to interact with the user.
393        """
394        security = self.api.connection_security
395        if self._is_bridge_or_web_player or security is None:
396            raise AbortFlow("nothing_to_configure")
397        provider = cast("SendspinProvider", self.provider)
398        record = await provider.server_api.pairing_store.record_by_client_id(self.player_id)
399        if security.psk_category is PskCategory.LONG_TERM and record is not None:
400            await self._run_verify_presence_flow(session, provider, record)
401            await session.finish({})
402            return
403        trusted_unpaired = (
404            await provider.server_api.pairing_store.trusted_unpaired(self.player_id) is not None
405        )
406        options = self._pairing_method_options(provider, offer_unpaired=not trusted_unpaired)
407        if not options:
408            raise AbortFlow("no_pair_methods")
409        if len(options) == 1 and options[0] != PAIR_METHOD_UNPAIRED:
410            method = options[0]
411        else:
412            # a lone unpaired-access option still renders the form: granting
413            # unauthenticated playback needs an explicit user choice
414            values = await session.form(
415                [
416                    ConfigEntry(
417                        key=CONF_PAIRING_METHOD,
418                        type=ConfigEntryType.STRING,
419                        required=True,
420                        default_value=options[0],
421                        options=[ConfigValueOption(value=option) for option in options],
422                    )
423                ],
424                step_id="select_method",
425            )
426            method = str(values[CONF_PAIRING_METHOD])
427        if method == PAIR_METHOD_UNPAIRED:
428            await provider.set_trusted_unpaired(self.player_id, enabled=True)
429        elif method == PAIR_METHOD_TOKEN:
430            await self._run_token_pairing_flow(session, provider)
431        else:
432            await self._run_pin_pairing_flow(
433                session, provider, static=method == PAIR_METHOD_STATIC_PIN
434            )
435        await session.finish({})
436
437    @property
438    def _is_bridge_or_web_player(self) -> bool:
439        """Whether this is a protocol bridge or built-in web/app player (skips pairing setup)."""
440        return self.player_id.startswith(BRIDGE_PREFIX) or self.is_web_player
441
442    def _subscribe_client_callbacks(self) -> None:
443        """Subscribe to client and group events for the currently bound client."""
444        self.api.disconnect_behaviour = DisconnectBehaviour.UNGROUP
445        self.unsub_event_cb = self.api.add_event_listener(self.event_cb)
446        self.unsub_group_event_cb = self.api.group.add_event_listener(self.group_event_cb)
447
448    def _unsubscribe_client_callbacks(self) -> None:
449        """Unsubscribe any active client and group listeners."""
450        if self.unsub_event_cb is not None:
451            with suppress(Exception):
452                self.unsub_event_cb()
453            self.unsub_event_cb = None
454        if self.unsub_group_event_cb is not None:
455            with suppress(Exception):
456                self.unsub_group_event_cb()
457            self.unsub_group_event_cb = None
458
459    def _refresh_client_info(
460        self,
461        sendspin_client: SendspinClient,
462        hello_payload: ClientHelloPayload | None = None,
463    ) -> None:
464        """
465        Refresh shared player attributes from a Sendspin client hello/info payload.
466
467        :param sendspin_client: The Sendspin client instance.
468        :param hello_payload: Optional hello payload to use instead of client info.
469        """
470        client_info = hello_payload or sendspin_client.info
471        preserved_identifiers = dict(self._attr_device_info.identifiers)
472        self._attr_name = client_info.name
473        if device_info := client_info.device_info:
474            self._attr_device_info = DeviceInfo(
475                model=device_info.product_name or "Unknown model",
476                manufacturer=device_info.manufacturer or "Unknown Manufacturer",
477                software_version=device_info.software_version,
478            )
479        else:
480            self._attr_device_info = DeviceInfo()
481        for id_type, id_value in preserved_identifiers.items():
482            self._attr_device_info.add_identifier(id_type, id_value)
483        # Add player_id as MAC identifier for protocol linking (if it's a valid MAC)
484        # This enables linking with bridged players (e.g., AirPlay via Sendspin bridge)
485        if IdentifierType.MAC_ADDRESS not in self._attr_device_info.identifiers:
486            if _mac := mac_from_bridge_client_id(self.player_id):
487                self._attr_device_info.add_identifier(IdentifierType.MAC_ADDRESS, _mac)
488            elif is_valid_mac_address(self.player_id):
489                self._attr_device_info.add_identifier(IdentifierType.MAC_ADDRESS, self.player_id)
490        self._attr_available = True
491
492    @property
493    def _artwork_role(self) -> ArtworkGroupRole | None:
494        """Get the ArtworkGroupRole for this player's group."""
495        role = self.api.group.group_role("artwork")
496        if isinstance(role, ArtworkGroupRole):
497            return role
498        return None
499
500    @property
501    def _metadata_role(self) -> MetadataGroupRole | None:
502        """Get the MetadataGroupRole for this player's group."""
503        role = self.api.group.group_role("metadata")
504        if isinstance(role, MetadataGroupRole):
505            return role
506        return None
507
508    @property
509    def _color_role(self) -> ColorGroupRole | None:
510        """Get the ColorGroupRole for this player's group."""
511        role = self.api.group.group_role("color")
512        if isinstance(role, ColorGroupRole):
513            return role
514        return None
515
516    @property
517    def _visualizer_role(self) -> VisualizerGroupRole | None:
518        """Get the VisualizerGroupRole for this player's group."""
519        role = self.api.group.group_role("visualizer")
520        if isinstance(role, VisualizerGroupRole):
521            return role
522        return None
523
524    @property
525    def _controller_role(self) -> ControllerGroupRole | None:
526        """Get the ControllerGroupRole for this player's group."""
527        role = self.api.group.group_role("controller")
528        if isinstance(role, ControllerGroupRole):
529            return role
530        return None
531
532    @property
533    def _player_role(self) -> PlayerRoleProtocol | None:
534        """Get the player role for this client (not group role)."""
535        for role in self.api.roles_by_family("player"):
536            if isinstance(role, PlayerRoleProtocol):
537                return role
538        return None
539
540    def _on_group_changed(self, new_group: SendspinGroup) -> None:
541        """
542        Handle group change logic.
543
544        Syncs playback state from the new group and schedules membership sync.
545        Override in subclasses for additional behaviour.
546
547        :param new_group: The new group this player has been assigned to.
548        """
549        # Sync playback state from the new group
550        match new_group.state:
551            case PlaybackStateType.PLAYING:
552                self._attr_playback_state = PlaybackState.PLAYING
553            case PlaybackStateType.PAUSED:
554                self._attr_playback_state = PlaybackState.PAUSED
555            case PlaybackStateType.STOPPED:
556                self._attr_playback_state = PlaybackState.IDLE
557                self._attr_elapsed_time = 0
558                self._attr_elapsed_time_last_updated = time.time()
559        # Update in case this is a newly created group
560        # GroupMemberAddedEvent or GroupMemberRemovedEvent will be fired before this
561        # so group members are already up to date at this point
562        self._schedule_membership_sync(new_group)
563
564    def _on_group_stopped(self) -> None:
565        """Handle the group transitioning to STOPPED state."""
566
567    async def _sync_membership_from_group(self, group: SendspinGroup) -> None:
568        """
569        Sync MA player group membership from authoritative group state.
570
571        :param group: The Sendspin group to sync from.
572        """
573        # Ignore stale events from a group we no longer belong to.
574        if group is not self.api.group:
575            return
576        group_client_ids = [client.client_id for client in group.clients]
577        is_leader = bool(group_client_ids) and group_client_ids[0] == self.player_id
578        desired_group_members = group_client_ids if is_leader else []
579        if self._attr_group_members != desired_group_members:
580            self._attr_group_members = desired_group_members
581            self.update_state()
582
583    def _schedule_membership_sync(self, group: SendspinGroup) -> None:
584        """Schedule a coalesced membership reconciliation task for this player."""
585        self.mass.create_task(
586            self._sync_membership_from_group(group),
587            task_id=f"sendspin_membership_sync_{self.player_id}",
588            abort_existing=True,
589        )
590
591    async def _handle_group_member_removed(self, group: SendspinGroup, client_id: str) -> None:
592        """Handle a group member being removed asynchronously."""
593        if client_id == self.player_id:
594            was_leader = (
595                bool(self._attr_group_members) and self._attr_group_members[0] == self.player_id
596            )
597            if was_leader and len(group.clients) > 0:
598                # We were removed as the group leader but other clients remain.
599                # Don't stop the group -- the PushStream keeps running for
600                # remaining members (playback session was transferred in set_members).
601                self.logger.debug(
602                    "Player %s removed as group leader; group continues for remaining members",
603                    self.player_id,
604                )
605            elif not was_leader:
606                self.logger.debug(
607                    "Player %s removed from group as non-leader; keeping old group playing",
608                    self.player_id,
609                )
610            # Clear members for our detached/solo state.
611            self._attr_group_members = []
612            self.update_state()
613        elif client_id in self._attr_group_members:
614            # Someone else left our group
615            self._attr_group_members.remove(client_id)
616            self.update_state()
617
618    async def _get_security_config_entries(self) -> list[ConfigEntry]:
619        """Build the pairing/security section entries."""
620        if self._is_bridge_or_web_player:
621            return []
622        provider = cast("SendspinProvider", self.provider)
623        # surface (and clear) any alert produced by the most recent action
624        alert: AlertText | None = self._pending_security_alert
625        self._pending_security_alert = None
626        status, actions = await self._security_state_entries(provider)
627        entries = [status] if status is not None else []
628        if alert is not None:
629            entries.append(alert_entry(alert))
630        return entries + actions
631
632    async def _security_state_entries(
633        self, provider: SendspinProvider
634    ) -> tuple[ConfigEntry | None, list[ConfigEntry]]:
635        """Return the current security-status entry and the available action entries."""
636        if not self.api.is_connected:
637            # The action dropped the connection (e.g. unpair); the client will reconnect
638            return ConfigEntry(key="security_status_disconnected", type=ConfigEntryType.LABEL), []
639
640        security = self.api.connection_security
641        if security is None:
642            return ConfigEntry(key="security_status_unencrypted", type=ConfigEntryType.ALERT), []
643
644        record = await provider.server_api.pairing_store.record_by_client_id(self.player_id)
645
646        # psk_category is fixed at handshake time: right after an unpair the live connection
647        # still reports LONG_TERM while the record is already gone. The settings view refetches
648        # in exactly that window and will not observe the later reconnect until reopened, so
649        # require the record too or else render the unpaired end state immediately.
650        if security.psk_category is PskCategory.LONG_TERM and record is not None:
651            return (
652                ConfigEntry(key="security_status_paired", type=ConfigEntryType.LABEL),
653                await self._paired_entries(
654                    provider, provider.pairing_config_snapshot(self.player_id)
655                ),
656            )
657
658        trusted_unpaired = (
659            await provider.server_api.pairing_store.trusted_unpaired(self.player_id) is not None
660        )
661        if not trusted_unpaired:
662            return None, []
663        return (
664            ConfigEntry(key="security_status_unpaired", type=ConfigEntryType.ALERT),
665            [action_entry(CONF_ACTION_REVOKE_UNPAIRED)],
666        )
667
668    async def _paired_entries(
669        self,
670        provider: SendspinProvider,
671        pairing_config: ManagementResultData | None,
672    ) -> list[ConfigEntry]:
673        """Return the action entries for the paired view (management, unpair)."""
674        entries: list[ConfigEntry] = []
675        if provider.get_management_session(self.player_id) is not None:
676            # Render from the snapshot the enter/patch actions keep fresh; only fetch if it is
677            # unexpectedly empty (e.g. a session opened without a fetch).
678            management_config = pairing_config
679            if management_config is None:
680                try:
681                    management_config = await provider.management_get_pairing_config(self.player_id)
682                except SecurityActionError as err:
683                    provider.exit_management(self.player_id)
684                    entries.append(alert_entry(error_alert(err)))
685            if management_config is not None:
686                entries.extend(self._management_section_entries(management_config))
687                return entries
688        entries.append(action_entry(CONF_ACTION_MANAGEMENT_ENTER, advanced=True))
689        entries.append(action_entry(CONF_ACTION_UNPAIR, advanced=True))
690        return entries
691
692    @staticmethod
693    def _management_section_entries(config: ManagementResultData) -> list[ConfigEntry]:
694        """Return the entries for the open device-management section."""
695        entries: list[ConfigEntry] = [
696            ConfigEntry(key="management_status", type=ConfigEntryType.LABEL)
697        ]
698        if config.unpaired_access is not None:
699            action = (
700                CONF_ACTION_MANAGEMENT_UNPAIRED_DISABLE
701                if config.unpaired_access.enabled
702                else CONF_ACTION_MANAGEMENT_UNPAIRED_ENABLE
703            )
704            entries.append(action_entry(action))
705        entries.extend(
706            SendspinBasePlayer._management_pin_method_entries(
707                config.static_pin,
708                CONF_ACTION_MANAGEMENT_STATIC_PIN_ENABLE,
709                CONF_ACTION_MANAGEMENT_STATIC_PIN_DISABLE,
710            )
711        )
712        entries.extend(
713            SendspinBasePlayer._management_pin_method_entries(
714                config.dynamic_pin,
715                CONF_ACTION_MANAGEMENT_DYNAMIC_PIN_ENABLE,
716                CONF_ACTION_MANAGEMENT_DYNAMIC_PIN_DISABLE,
717            )
718        )
719        entries.append(action_entry(CONF_ACTION_MANAGEMENT_EXIT))
720        return entries
721
722    @staticmethod
723    def _management_pin_method_entries(
724        method: PairingMethodConfig | None,
725        enable_action: str,
726        disable_action: str,
727    ) -> list[ConfigEntry]:
728        """Return the toggle for one PIN pairing method, empty if the device lacks it."""
729        if method is None:
730            return []
731        action = disable_action if method.enabled else enable_action
732        return [action_entry(action)]
733
734    async def _handle_security_action(
735        self, provider: SendspinProvider, action: str
736    ) -> AlertText | None:
737        """Execute an unpair/management action, returning a localized alert on failure."""
738        try:
739            if action == CONF_ACTION_UNPAIR:
740                await provider.unpair_client(self.player_id)
741            elif action == CONF_ACTION_REVOKE_UNPAIRED:
742                await provider.set_trusted_unpaired(self.player_id, enabled=False)
743            elif action == CONF_ACTION_MANAGEMENT_ENTER:
744                provider.enter_management(self.player_id)
745                try:
746                    await provider.management_get_pairing_config(self.player_id)
747                except SecurityActionError:
748                    provider.exit_management(self.player_id)
749                    raise
750            elif action == CONF_ACTION_MANAGEMENT_EXIT:
751                provider.exit_management(self.player_id)
752            elif action in _MANAGEMENT_ACTIONS:
753                await provider.management_set_pairing_config(
754                    self.player_id, _MANAGEMENT_ACTIONS[action]
755                )
756        except (
757            HandshakeAbortedError,
758            PairingError,
759            TimeoutError,
760            OSError,
761            SecurityActionError,
762            ValueError,
763        ) as err:
764            return error_alert(err)
765        return None
766
767    def _pairing_method_options(
768        self, provider: SendspinProvider, *, offer_unpaired: bool
769    ) -> list[str]:
770        """
771        Return the pairing-method option values the device currently offers for setup.
772
773        :param offer_unpaired: Include the unpaired-playback grant when the device permits it.
774        """
775        info = self.api.info_or_none
776        pairing_config = provider.pairing_config_snapshot(self.player_id)
777        pair_methods = effective_pair_methods(info, pairing_config)
778        usable_pin_methods = {
779            descriptor.method
780            for descriptor in pair_methods
781            if descriptor.method in (PairMethod.DYNAMIC_PIN, PairMethod.STATIC_PIN)
782            and not descriptor.locked_out
783        }
784        options: list[str] = []
785        if usable_pin_methods:
786            options.append(PAIR_METHOD_PIN)
787            # Static PIN is only a distinct, meaningful choice when both PIN methods are usable.
788            if usable_pin_methods >= {PairMethod.DYNAMIC_PIN, PairMethod.STATIC_PIN}:
789                options.append(PAIR_METHOD_STATIC_PIN)
790        if any(descriptor.method is PairMethod.PAIRING_PSK for descriptor in pair_methods):
791            options.append(PAIR_METHOD_TOKEN)
792        if offer_unpaired and effective_unpaired_access(info, pairing_config):
793            options.append(PAIR_METHOD_UNPAIRED)
794        return options
795
796    async def _pairing_succeeded(
797        self, provider: SendspinProvider, pin_session: PinPairingSession
798    ) -> bool:
799        """Whether the attempt finished cleanly (or a long-term pairing record now exists)."""
800        if pin_session.finished and pin_session.error is None:
801            return True
802        if pin_session.verify:
803            return False
804        # a confirm wait that outlived its deadline: the record is the proof of success
805        return (
806            await provider.server_api.pairing_store.record_by_client_id(self.player_id) is not None
807        )
808
809    async def _run_verify_presence_flow(
810        self, session: SetupSession, provider: SendspinProvider, record: ServerPairingRecord
811    ) -> None:
812        """Confirm a paired device's physical presence via its dynamic PIN."""
813        info = self.api.info_or_none
814        pairing_config = provider.pairing_config_snapshot(self.player_id)
815        offers_dynamic_pin = any(
816            descriptor.method is PairMethod.DYNAMIC_PIN and not descriptor.locked_out
817            for descriptor in effective_pair_methods(info, pairing_config)
818        )
819        # presence proven by a dynamic-PIN pairing itself needs no re-verification
820        if not offers_dynamic_pin or PairMethod.DYNAMIC_PIN in record.pair_methods:
821            raise AbortFlow("already_paired")
822        await self._run_pin_pairing_flow(session, provider, static=False, verify=True)
823
824    async def _run_pin_pairing_flow(
825        self,
826        session: SetupSession,
827        provider: SendspinProvider,
828        *,
829        static: bool,
830        verify: bool = False,
831    ) -> None:
832        """
833        Pair via PIN: gesture wait, PIN entry and the retry-in-place loop.
834
835        A retryable failure re-renders the PIN form (start_pin_pairing resumes the session in
836        place); a terminal failure aborts the flow and an expired gesture wait propagates as a
837        timed_out abort. On any non-success exit the finally tears down a device-side session
838        still in flight - including when the flow is cancelled.
839
840        :param static: Pair with the device's static PIN instead of a dynamic one.
841        :param verify: Verify an already-paired device's presence (dynamic PIN only).
842        """
843        succeeded = False
844        errors: dict[str, str] | None = None
845        try:
846            while True:
847                try:
848                    pin_session = await provider.start_pin_pairing(
849                        self.player_id, static=static, verify=verify
850                    )
851                except SecurityActionError as err:
852                    raise AbortFlow(_pairing_abort_reason(err)) from err
853                if pin_session.awaiting_gesture:
854                    await session.progress_until(
855                        pin_session.pin_request_event.wait(),
856                        step_id="awaiting_gesture",
857                        text="awaiting_gesture",
858                        expires_in=PAIR_GESTURE_TIMEOUT,
859                    )
860                if not pin_session.awaiting_pin:
861                    # The attempt ended before a PIN could be entered.
862                    if await self._pairing_succeeded(provider, pin_session):
863                        succeeded = True
864                        return
865                    if pin_session.can_retry:
866                        errors = {"base": _pin_error_slug(pin_session.error)}
867                        continue
868                    raise AbortFlow(_pairing_abort_reason(pin_session.error))
869                pin_values = await session.form(
870                    [ConfigEntry(key=CONF_PAIRING_PIN, type=ConfigEntryType.STRING, required=True)],
871                    step_id="verify_pin" if verify else "enter_pin",
872                    errors=errors,
873                )
874                errors = None
875                try:
876                    provider.submit_pin(self.player_id, str(pin_values[CONF_PAIRING_PIN]).strip())
877                except SecurityActionError as err:
878                    # the session ended underneath us (cancelled/timed out); start afresh
879                    errors = {"base": err.alert_key}
880                    continue
881                task = pin_session.task
882                if task is not None and not task.done():
883                    # Join the pairing task: the step deadline must not cancel it.
884                    with suppress(StepExpiredError):
885                        await session.progress_until(
886                            join_task(task),
887                            step_id="confirming",
888                            text="confirming",
889                            expires_in=PAIR_CONFIRM_TIMEOUT,
890                        )
891                if await self._pairing_succeeded(provider, pin_session):
892                    succeeded = True
893                    return
894                if pin_session.can_retry:
895                    errors = {"base": _pin_error_slug(pin_session.error)}
896                    continue
897                raise AbortFlow(_pairing_abort_reason(pin_session.error))
898        finally:
899            if succeeded:
900                provider.clear_pin_session(self.player_id)
901            elif provider.get_pin_session(self.player_id) is not None:
902                await provider.cancel_pin_pairing(self.player_id)
903
904    async def _run_token_pairing_flow(
905        self, session: SetupSession, provider: SendspinProvider
906    ) -> None:
907        """Pair via a pasted pairing token, re-rendering the form on a recoverable failure."""
908        errors: dict[str, str] | None = None
909        while True:
910            token_values = await session.form(
911                [ConfigEntry(key=CONF_PAIRING_TOKEN, type=ConfigEntryType.STRING, required=True)],
912                step_id="enter_token",
913                errors=errors,
914            )
915            try:
916                await provider.pair_with_token(
917                    self.player_id, str(token_values[CONF_PAIRING_TOKEN]).strip()
918                )
919            except (
920                SecurityActionError,
921                PairingError,
922                HandshakeAbortedError,
923                TimeoutError,
924                OSError,
925            ) as err:
926                errors = {"base": error_alert(err).key}
927                continue
928            return
929
930
931class SendspinPlayer(SendspinBasePlayer):
932    """A sendspin audio player in Music Assistant."""
933
934    _attr_type = PlayerType.PROTOCOL
935
936    last_sent_artwork_url: str | None = None
937    last_sent_artist_artwork_url: str | None = None
938    _last_beat_queue_item_id: str | None = None
939    _last_beat_anchor_us: int | None = None
940    # Background poller that retries _send_beat_schedule when analysis is
941    # not yet available. Cancelled on track change / stop / successful push.
942    _beat_retry_task: asyncio.Task[None] | None = None
943    # Queue item the current poller is targeting (so a track switch cancels it).
944    _beat_retry_queue_item_id: str | None = None
945    playback_session: SendspinPlaybackSession
946    static_delay_default_ms: int = DEFAULT_SENDSPIN_STATIC_DELAY
947    # HA media_player entity announcements are relayed to (ESPHome-backed devices)
948    _hass_announce_entity_id: str | None = None
949
950    @property
951    def requires_flow_mode(self) -> bool:
952        """Return if the player requires flow mode."""
953        return True
954
955    def __init__(
956        self,
957        provider: SendspinProvider,
958        player_id: str,
959        initial_hello: ClientHelloPayload | None = None,
960    ) -> None:
961        """Initialize the Player."""
962        super().__init__(provider, player_id, initial_hello)
963        hello_payload = initial_hello or self.api.info
964        self.playback_session = SendspinPlaybackSession(self)
965        self._attr_supported_features = {
966            PlayerFeature.PLAY_MEDIA,
967            PlayerFeature.SET_MEMBERS,
968            PlayerFeature.MULTI_DEVICE_DSP,
969        }
970        # Keep volume/mute features of the first registration as a workaround for Cast.
971        if hello_payload.player_support:
972            _supported_commands = hello_payload.player_support.supported_commands
973            if PlayerCommand.VOLUME in _supported_commands:
974                self._attr_supported_features.add(PlayerFeature.VOLUME_SET)
975            if PlayerCommand.MUTE in _supported_commands:
976                self._attr_supported_features.add(PlayerFeature.VOLUME_MUTE)
977
978    @property
979    def supported_sample_rates(self) -> list[tuple[int, int]] | None:
980        """Return supported (sample_rate, bit_depth) tuples derived from the player role."""
981        # not cached: the player role / reported formats can change after the
982        # client (re)registers, so we always re-resolve from the live role state
983        if (player_role := self._player_role) is not None:
984            formats = player_role.get_supported_formats() or []
985            rates = sorted({(fmt.sample_rate, fmt.bit_depth) for fmt in formats})
986            if rates:
987                return rates
988        return [(44100, 16)]
989
990    def preserve_control_features_from(self, other: SendspinPlayer) -> None:
991        """Keep the first registration's volume/mute features as a workaround for Cast."""
992        for feature in (PlayerFeature.VOLUME_SET, PlayerFeature.VOLUME_MUTE):
993            if feature in other.supported_features:
994                self._attr_supported_features.add(feature)
995            else:
996                self._attr_supported_features.discard(feature)
997
998    def set_hass_announce_entity(self, entity_id: str | None) -> None:
999        """
1000        Set or clear the Home Assistant entity used to relay announcements.
1001
1002        ESPHome devices support announcements natively (ducking any running
1003        playback), but that capability is only reachable through their Home
1004        Assistant media_player entity; the PLAY_ANNOUNCEMENT feature follows it.
1005
1006        :param entity_id: The HA media_player entity id, or None to clear.
1007        """
1008        self._hass_announce_entity_id = entity_id
1009        if entity_id is not None:
1010            self._attr_supported_features.add(PlayerFeature.PLAY_ANNOUNCEMENT)
1011        else:
1012            self._attr_supported_features.discard(PlayerFeature.PLAY_ANNOUNCEMENT)
1013
1014    async def play_announcement(
1015        self, announcement: PlayerMedia, volume_level: int | None = None
1016    ) -> None:
1017        """Handle (provider native) playback of an announcement on given player."""
1018        entity_id = self._hass_announce_entity_id
1019        hass = cast("HomeAssistantProvider | None", self.mass.get_provider("hass"))
1020        if entity_id is None or hass is None or not hass.available:
1021            raise PlayerCommandFailed(
1022                f"Announcement relay via Home Assistant is not available for {self.display_name}"
1023            )
1024        self.logger.info(
1025            "Playing announcement %s on %s (via Home Assistant)",
1026            announcement.uri,
1027            self.display_name,
1028        )
1029        if volume_level is not None:
1030            # the device's announcement pipeline plays at its own volume;
1031            # the announce volume config entries are hidden for this player
1032            self.logger.debug("Ignoring announcement volume level for player %s", self.display_name)
1033        await hass.play_announcement_on_entity(entity_id, announcement)
1034        self.logger.debug("Playing announcement on %s completed", self.display_name)
1035
1036    def restore_bridge_identity(
1037        self, previous_device_info: DeviceInfo, previous_type: PlayerType
1038    ) -> None:
1039        """Keep bridge players exposed as protocol bridges after client attach updates."""
1040        if previous_type != PlayerType.PROTOCOL:
1041            return
1042        if not (
1043            IdentifierType.CAST_UUID in previous_device_info.identifiers
1044            or IdentifierType.AIRPLAY_ID in previous_device_info.identifiers
1045        ):
1046            return
1047        refreshed_identifiers = dict(self._attr_device_info.identifiers)
1048        self._attr_device_info = DeviceInfo(
1049            model=previous_device_info.model,
1050            manufacturer=previous_device_info.manufacturer,
1051            software_version=self._attr_device_info.software_version,
1052        )
1053        for id_type, id_value in refreshed_identifiers.items():
1054            self._attr_device_info.add_identifier(id_type, id_value)
1055        self.is_web_player = False
1056        self._attr_hidden_by_default = False
1057        self._attr_expose_to_ha_by_default = True
1058        self._attr_type = PlayerType.PROTOCOL
1059
1060    def _subscribe_client_callbacks(self) -> None:
1061        """Subscribe to client and group events for the currently bound client."""
1062        super()._subscribe_client_callbacks()
1063        self.api.disconnect_behaviour = DisconnectBehaviour.STOP
1064        if controller_role := self._controller_role:
1065            controller_role.set_supported_commands(SUPPORTED_GROUP_COMMANDS)
1066
1067    def _refresh_client_info(
1068        self,
1069        sendspin_client: SendspinClient,
1070        hello_payload: ClientHelloPayload | None = None,
1071    ) -> None:
1072        """Refresh player attributes from a Sendspin client hello/info payload."""
1073        super()._refresh_client_info(sendspin_client, hello_payload=hello_payload)
1074        client_info = hello_payload or sendspin_client.info
1075        if device_info := client_info.device_info:
1076            # determine if this is a web/app player based on product name or manufacturer
1077            # TODO: make this part of the spec and let clients explicitly report if they
1078            # are a web/app player instead of relying on heuristics
1079            self.is_web_player = (
1080                device_info.product_name
1081                in (
1082                    "Web Browser",
1083                    "Web Player",
1084                    "Mobile Application",
1085                    "PWA",
1086                )
1087                or device_info.manufacturer == "Music Assistant"
1088            )
1089        else:
1090            self.is_web_player = False
1091        if client_info.player_support:
1092            for role in sendspin_client.roles_by_family("player"):
1093                volume = role.get_player_volume()
1094                muted = role.get_player_muted()
1095                if volume is not None:
1096                    self._attr_volume_level = volume
1097                if muted is not None:
1098                    self._attr_volume_muted = muted
1099                if volume is not None or muted is not None:
1100                    break
1101        # virtual players are server-owned anchors that follow the same
1102        # hidden/standalone semantics as web players, without relying on the
1103        # device-info heuristics above
1104        is_standalone = self.is_web_player or cast(
1105            "SendspinProvider", self.provider
1106        ).is_virtual_player(self.player_id)
1107        self._attr_expose_to_ha_by_default = not is_standalone
1108        self._attr_hidden_by_default = is_standalone
1109        # register web/app player as native player type because it doesn't need to be linked
1110        # every web/app player is just a standalone player.
1111        self._attr_type = PlayerType.PLAYER if is_standalone else PlayerType.PROTOCOL
1112
1113    def event_cb(self, client: SendspinClient, event: ClientEvent) -> None:
1114        """Event callback registered to the sendspin client."""
1115        match event:
1116            case VolumeChangedEvent(volume=volume, muted=muted):
1117                self._attr_volume_level = volume
1118                self._attr_volume_muted = muted
1119                self.update_state()
1120            case StaticDelayChangedEvent(static_delay_ms=delay_ms):
1121                self.logger.debug("Static delay changed to %d ms", delay_ms)
1122                current = self.config.get_value(
1123                    CONF_SENDSPIN_STATIC_DELAY, self.static_delay_default_ms
1124                )
1125                if current != delay_ms:
1126                    self.mass.config.set_raw_player_config_value(
1127                        self.player_id, CONF_SENDSPIN_STATIC_DELAY, delay_ms
1128                    )
1129            case _:
1130                super().event_cb(client, event)
1131
1132    def _on_group_changed(self, new_group: SendspinGroup) -> None:
1133        """Handle group change with controller commands and playback session cancellation."""
1134        if controller_role := self._controller_role:
1135            controller_role.set_supported_commands(SUPPORTED_GROUP_COMMANDS)
1136        # Cancel active playback - push stream belongs to the old group
1137        self.mass.create_task(self.playback_session.cancel("group changed"))
1138        super()._on_group_changed(new_group)
1139
1140    def _on_group_stopped(self) -> None:
1141        """Cancel playback session when group stops and we are the leader."""
1142        if self.synced_to is not None:
1143            return
1144        # Bind the cancel to the session task that is live right now: by the time
1145        # the deferred task below runs, play_media may already have started a fresh
1146        # session, which must not be torn down by this stale group-stopped event.
1147        stale_task = self.playback_session.playback_task
1148        if stale_task is None or stale_task.done():
1149            return
1150        self.mass.create_task(self._cancel_stale_playback_session(stale_task))
1151
1152    async def _cancel_stale_playback_session(self, task: asyncio.Task[None]) -> None:
1153        """Cancel the playback session only if the given task is still the active one."""
1154        if self.playback_session.playback_task is not task:
1155            return
1156        await self.playback_session.cancel("group stopped")
1157
1158    def group_event_cb(self, group: SendspinGroup, event: GroupEvent) -> None:
1159        """Event callback registered to the sendspin group this player belongs to."""
1160        # Leader only: a synced follower's self.state.current_media is a reference to
1161        # the leader's PlayerMedia object, so the refresh below would mutate the leader's
1162        # anchor from every follower. The metadata push (also leader-only) is what these
1163        # refreshes exist to serve, so followers have nothing to do here.
1164        is_resume = (
1165            isinstance(event, GroupStateChangedEvent)
1166            and event.state == PlaybackStateType.PLAYING
1167            and self._attr_playback_state == PlaybackState.PAUSED
1168            and self.synced_to is None
1169        )
1170        if is_resume:
1171            # _attr_elapsed_time_last_updated is only advanced by playback.py's commit
1172            # loop, which stops while paused - so it's still anchored to the moment
1173            # playback paused. Fast-forward it now, before update_state() below flips
1174            # playback_state to PLAYING, so corrected_elapsed_time doesn't extrapolate
1175            # across the paused span.
1176            self._attr_elapsed_time_last_updated = time.time()
1177        super().group_event_cb(group, event)
1178        if is_resume and self.state.current_media is not None:
1179            # send_current_media_metadata() (scheduled below) reads self.state.current_media,
1180            # which the queue controller rebuilds from its own cached elapsed-time anchor -
1181            # only refreshed via a 500ms-debounced callback, so it's still stale here even
1182            # after the fix above. Patch this update's snapshot directly so the imminent
1183            # metadata push doesn't race ahead of that debounce with a stale value.
1184            self.state.current_media.elapsed_time_last_updated = time.time()
1185        match event:
1186            case GroupStateChangedEvent(state=state) if self.synced_to is None and state in (
1187                PlaybackStateType.PLAYING,
1188                PlaybackStateType.PAUSED,
1189            ):
1190                # Push progress explicitly: current_media's identity is unchanged across
1191                # pause/resume so update_state() above won't debounce a metadata push
1192                # through the normal media-changed callback.
1193                self.mass.create_task(
1194                    self.send_current_media_metadata(),
1195                    task_id=f"sendspin_metadata_{self.player_id}",
1196                    abort_existing=True,
1197                )
1198            case ControllerEvent() as controller_event:
1199                if self.synced_to is None:
1200                    self.mass.create_task(self._handle_controller_event(controller_event))
1201
1202    async def _handle_controller_event(self, event: ControllerEvent) -> None:
1203        """Handle a controller event from the ControllerGroupRole."""
1204        queue = self.mass.player_queues.get_active_queue(self.player_id)
1205        match event:
1206            case ControllerPlayEvent():
1207                await self.mass.players.cmd_play(self.player_id)
1208            case ControllerPauseEvent():
1209                await self.mass.players.cmd_pause(self.player_id)
1210            case ControllerStopEvent():
1211                await self.mass.players.cmd_stop(self.player_id)
1212            case ControllerNextEvent():
1213                await self.mass.players.cmd_next_track(self.player_id)
1214            case ControllerPreviousEvent():
1215                await self.mass.players.cmd_previous_track(self.player_id)
1216            case ControllerRepeatEvent(mode=mode) if queue:
1217                match mode:
1218                    case SendspinRepeatMode.OFF:
1219                        self.mass.player_queues.set_repeat(queue.queue_id, RepeatMode.OFF)
1220                    case SendspinRepeatMode.ONE:
1221                        self.mass.player_queues.set_repeat(queue.queue_id, RepeatMode.ONE)
1222                    case SendspinRepeatMode.ALL:
1223                        self.mass.player_queues.set_repeat(queue.queue_id, RepeatMode.ALL)
1224            case ControllerShuffleEvent(shuffle=shuffle) if queue:
1225                await self.mass.player_queues.set_shuffle(queue.queue_id, shuffle_enabled=shuffle)
1226            case ControllerSeekEvent(position_ms=position_ms) if (
1227                queue and queue.current_item and queue.current_item.duration
1228            ):
1229                # Clamp in case track duration changed after we advertised the seek range.
1230                duration_ms = int(queue.current_item.duration * 1000)
1231                await self.mass.player_queues.seek(
1232                    queue.queue_id, max(0, min(position_ms, duration_ms)) // 1000
1233                )
1234            case ControllerSeekRelativeEvent(offset_ms=offset_ms) if (
1235                queue and queue.current_item and queue.current_item.duration
1236            ):
1237                # Clamp current position + offset to the 0..duration range.
1238                target_ms = int(queue.corrected_elapsed_time * 1000) + offset_ms
1239                duration_ms = int(queue.current_item.duration * 1000)
1240                await self.mass.player_queues.seek(
1241                    queue.queue_id, max(0, min(target_ms, duration_ms)) // 1000
1242                )
1243
1244    async def _sync_membership_from_group(self, group: SendspinGroup) -> None:
1245        """Sync MA/player + playback session membership from authoritative group state."""
1246        # Ignore stale events from a group we no longer belong to.
1247        if group is not self.api.group:
1248            return
1249        group_client_ids = [client.client_id for client in group.clients]
1250        is_leader = bool(group_client_ids) and group_client_ids[0] == self.player_id
1251        desired_group_members = group_client_ids if is_leader else []
1252        desired_session_members = group_client_ids[1:] if is_leader else []
1253        if self._attr_group_members != desired_group_members:
1254            self._attr_group_members = desired_group_members
1255            self.update_state()
1256        # Only use STOP when we actually lead other members.
1257        self.api.disconnect_behaviour = (
1258            DisconnectBehaviour.STOP
1259            if is_leader and len(desired_session_members) > 0
1260            else DisconnectBehaviour.UNGROUP
1261        )
1262        await self.playback_session.sync_members(set(desired_session_members))
1263
1264    async def volume_set(self, volume_level: int) -> None:
1265        """Handle VOLUME_SET command on the player."""
1266        roles = self.api.roles_by_family("player")
1267        for role in roles:
1268            role.set_player_volume(volume_level)
1269
1270    async def volume_mute(self, muted: bool) -> None:
1271        """Handle VOLUME MUTE command on the player."""
1272        roles = self.api.roles_by_family("player")
1273        for role in roles:
1274            role.set_player_mute(muted)
1275        # Native clients don't always emit a VolumeChangedEvent in response to a
1276        # mute command, so update our state directly to keep MA in sync.
1277        self._attr_volume_muted = muted
1278        self.update_state()
1279
1280    async def stop(self) -> None:
1281        """Stop command."""
1282        self.logger.debug("Received STOP command on player %s", self.display_name)
1283        self.mark_stop_called()
1284        self._attr_current_media = None
1285        self._attr_playback_state = PlaybackState.IDLE
1286        self._attr_elapsed_time = 0
1287        self._attr_elapsed_time_last_updated = time.time()
1288        self.update_state()
1289        # group.stop() snapshots the live position, which it can only do while the push
1290        # stream is up - cancelling first leaves it re-emitting a stale anchor. Teardown
1291        # goes in finally so a failing group stop can't strand the session, and nothing
1292        # may await between the two: the STOPPED event cancels this session inline, and a
1293        # suspension in between would let that cancel cut pipeline teardown short.
1294        try:
1295            await self.api.group.stop()
1296        finally:
1297            await self.playback_session.cancel("stop command")
1298
1299    def _get_cast_bridge_manager(self) -> SendspinBridgeManager | None:
1300        """Return the Chromecast provider's Sendspin bridge manager, if loaded."""
1301        chromecast_provider = self.mass.get_provider("chromecast")
1302        if chromecast_provider is None:
1303            return None
1304        manager = getattr(chromecast_provider, "bridge_manager", None)
1305        if manager is None:
1306            return None
1307        return cast("SendspinBridgeManager", manager)
1308
1309    async def play_media(self, media: PlayerMedia) -> None:
1310        """Play media command."""
1311        self.logger.debug(
1312            "Received PLAY_MEDIA command on player %s with uri %s", self.display_name, media.uri
1313        )
1314
1315        self._attr_current_media = media
1316        self._attr_elapsed_time = None
1317        self._attr_elapsed_time_last_updated = None
1318
1319        await self.playback_session.cancel("new media requested")
1320
1321        # Cast-only: reset future before start() to avoid racing _on_stream_start
1322        # and to cancel any stale pending future from a previous timed-out attempt.
1323        cast_app_ready: asyncio.Future[None] | None = None
1324        if (mgr := self._get_cast_bridge_manager()) and (
1325            bridge := mgr.get_bridge_by_client_id(self.player_id)
1326        ):
1327            cast_app_ready = bridge.reset_cast_app_ready()
1328
1329        await self.playback_session.start(media)
1330        self.update_state()
1331
1332        if cast_app_ready is None:
1333            return
1334        try:
1335            await asyncio.wait_for(asyncio.shield(cast_app_ready), timeout=30.0)
1336        except BaseException as exc:
1337            if not cast_app_ready.done():
1338                cast_app_ready.cancel()
1339            # Cancel the playback session so we don't keep streaming to a
1340            # device that never reported ready.
1341            with suppress(Exception):
1342                await self.playback_session.cancel("cast app readiness failed")
1343            if isinstance(exc, TimeoutError):
1344                raise PlayerCommandFailed(
1345                    f"Cast app on {self.display_name} did not report ready within 30s",
1346                    translation_key="cast_app_not_ready",
1347                    translation_owner=self.translation_owner,
1348                    translation_args=[self.display_name],
1349                ) from None
1350            raise
1351
1352    async def on_config_updated(self) -> None:
1353        """Handle logic when the PlayerConfig is first loaded or updated."""
1354        await self._apply_preferred_format()
1355        await self._apply_static_delay()
1356
1357    async def _apply_preferred_format(self) -> None:
1358        """Read config and set/clear the players preferred format."""
1359        player_role = self._player_role
1360        if player_role is None:
1361            return
1362
1363        config_value = cast(
1364            "str",
1365            self.config.get_value(CONF_PREFERRED_SENDSPIN_FORMAT, SENDSPIN_FORMAT_AUTOMATIC),
1366        )
1367        if config_value == SENDSPIN_FORMAT_AUTOMATIC:
1368            # Automatic mode: clear override and let client decide.
1369            player_role.set_preferred_format(None, None)
1370            return
1371
1372        parsed = option_value_to_format(config_value)
1373        if parsed is None:
1374            self.logger.warning(
1375                "Invalid audio format config value '%s' for player %s",
1376                config_value,
1377                self.display_name,
1378            )
1379            return
1380
1381        codec, audio_format = parsed
1382        if not player_role.set_preferred_format(audio_format, codec):
1383            self.logger.warning(
1384                "Failed to set preferred audio format %s %s for player %s",
1385                codec.name,
1386                audio_format,
1387                self.display_name,
1388            )
1389
1390    async def _apply_static_delay(self) -> None:
1391        """Read config and send set_static_delay command if supported."""
1392        player_role = self._player_role
1393        if player_role is None:
1394            return
1395
1396        config_value = cast(
1397            "int",
1398            self.config.get_value(CONF_SENDSPIN_STATIC_DELAY, self.static_delay_default_ms),
1399        )
1400        player_role.set_static_delay(config_value)
1401
1402    async def set_members(
1403        self,
1404        player_ids_to_add: list[str] | None = None,
1405        player_ids_to_remove: list[str] | None = None,
1406    ) -> None:
1407        """Handle SET_MEMBERS command on the player."""
1408        for player_id in player_ids_to_remove or []:
1409            member_player = self.mass.players.get_player(player_id, True)
1410            member_player = cast("SendspinPlayer", member_player)
1411
1412            # Dynamic leader switch: transfer the active playback session to the
1413            # next remaining group member before removing ourselves from the group.
1414            # This keeps the PushStream alive for the remaining members.
1415            if (
1416                player_id == self.player_id
1417                and self.playback_session.playback_task is not None
1418                and not self.playback_session.playback_task.done()
1419            ):
1420                remaining = [c for c in self.api.group.clients if c.client_id != self.player_id]
1421                if remaining:
1422                    new_owner_id = remaining[0].client_id
1423                    new_owner = self.mass.players.get_player(new_owner_id)
1424                    if isinstance(new_owner, SendspinPlayer):
1425                        self.logger.info(
1426                            "Transferring playback session to %s for dynamic leader switch",
1427                            new_owner.display_name,
1428                        )
1429                        await self.playback_session.transfer_to(new_owner)
1430                        new_owner.playback_session = self.playback_session
1431                        self.playback_session = SendspinPlaybackSession(self)
1432
1433            await self.api.group.remove_client(member_player.api)
1434        # Cast-only: reset futures before add so a fatal error on a Cast-bridged
1435        # member (e.g. AudioContext unsupported) raises PlayerCommandFailed.
1436        # Only track readiness while streaming, only then add_client launches the app.
1437        bridge_manager = self._get_cast_bridge_manager()
1438        pending_cast: list[tuple[SendspinPlayer, asyncio.Future[None]]] = []
1439        try:
1440            for player_id in player_ids_to_add or []:
1441                member_player = cast(
1442                    "SendspinPlayer", self.mass.players.get_player(player_id, True)
1443                )
1444                if (
1445                    self.api.group.has_active_stream
1446                    and bridge_manager
1447                    and (bridge := bridge_manager.get_bridge_by_client_id(player_id))
1448                ):
1449                    pending_cast.append((member_player, bridge.reset_cast_app_ready()))
1450                await self.api.group.add_client(member_player.api)
1451
1452            if pending_cast:
1453                try:
1454                    await asyncio.wait_for(
1455                        asyncio.gather(*(asyncio.shield(f) for _, f in pending_cast)),
1456                        timeout=30.0,
1457                    )
1458                except TimeoutError:
1459                    stuck = [m.display_name for m, f in pending_cast if not f.done()]
1460                    raise PlayerCommandFailed(
1461                        f"Cast app on {', '.join(stuck)} did not report ready within 30s",
1462                        translation_key="cast_app_members_not_ready",
1463                        translation_owner=self.translation_owner,
1464                        translation_args=[", ".join(stuck)],
1465                    ) from None
1466        except BaseException:
1467            # Roll back Cast members we just added so a failed group operation
1468            # doesn't leave dead members in the Sendspin group.
1469            for member, _ in pending_cast:
1470                with suppress(Exception):
1471                    await self.api.group.remove_client(member.api)
1472            raise
1473        finally:
1474            for _, f in pending_cast:
1475                if not f.done():
1476                    f.cancel()
1477        # self.group_members will be updated by the group event callback
1478
1479    async def _send_album_artwork(self, current_media: PlayerMedia) -> str | None:
1480        """
1481        Send album artwork to the sendspin group.
1482
1483        Args:
1484            current_media: The current player media.
1485        """
1486        # image_url is resolved per-source upstream (radio / Spotify Connect / queue items).
1487        artwork_url = current_media.image_url
1488        if artwork_url != self.last_sent_artwork_url:
1489            self.last_sent_artwork_url = artwork_url
1490            if artwork_url is not None:
1491                # Fetch from the resolved URL so the bytes match artwork_url, even when
1492                # radio now-playing art differs from the queue item's own image.
1493                try:
1494                    image_data = await self.mass.metadata.get_thumbnail(
1495                        artwork_url, provider="builtin"
1496                    )
1497                except MediaNotFoundError:
1498                    # artwork file was removed from disk; skip rather than crash the send
1499                    image_data = None
1500                if isinstance(image_data, bytes):
1501                    # decode through the guard so undecodable art (e.g. SVG) is skipped, not crashed
1502                    image = await self._decode_artwork(image_data)
1503                    if image is not None and (artwork_role := self._artwork_role) is not None:
1504                        await artwork_role.set_album_artwork(image)
1505            elif (artwork_role := self._artwork_role) is not None:
1506                await artwork_role.set_album_artwork(None)
1507
1508        return artwork_url
1509
1510    async def _send_artist_artwork(self, current_item: QueueItem) -> None:
1511        """Send artist artwork to the sendspin group."""
1512        artist_artwork_url: str | None = None
1513
1514        if current_item.media_item is not None and is_track(current_item.media_item):
1515            artists = current_item.media_item.artists
1516            if artists:
1517                primary_artist = artists[0]
1518                # Prefer a full library artist (has reliable up-to-date artwork) over
1519                # the ItemMapping in the queue item, which often has image=None.
1520                result = await self.mass.music.get_library_item_by_prov_id(
1521                    MediaType.ARTIST, primary_artist.item_id, primary_artist.provider
1522                )
1523                artist_item = result if isinstance(result, Artist) else None
1524                image = artist_item.image if artist_item is not None else primary_artist.image
1525                if image is not None:
1526                    artist_artwork_url = self.mass.metadata.get_image_url(image)
1527
1528        if artist_artwork_url != self.last_sent_artist_artwork_url:
1529            self.last_sent_artist_artwork_url = artist_artwork_url
1530            if artist_artwork_url is not None:
1531                # Fetch bytes from the already-resolved URL to avoid the secondary
1532                # provider lookup that get_image_data_for_item triggers for ItemMappings.
1533                try:
1534                    artist_image_data = await self.mass.metadata.get_thumbnail(
1535                        artist_artwork_url, provider="builtin"
1536                    )
1537                except MediaNotFoundError:
1538                    # artwork file was removed from disk; skip rather than crash the send
1539                    artist_image_data = None
1540                if isinstance(artist_image_data, bytes):
1541                    artist_image = await self._decode_artwork(artist_image_data)
1542                    if (
1543                        artist_image is not None
1544                        and (artwork_role := self._artwork_role) is not None
1545                    ):
1546                        await artwork_role.set_artist_artwork(artist_image)
1547            elif (artwork_role := self._artwork_role) is not None:
1548                await artwork_role.set_artist_artwork(None)
1549
1550    async def _decode_artwork(self, image_data: bytes) -> Image.Image | None:
1551        """
1552        Decode artwork bytes into a Pillow image, returning None if undecodable.
1553
1554        :param image_data: Raw image bytes to decode.
1555        """
1556
1557        def _open() -> Image.Image:
1558            img = Image.open(BytesIO(image_data))
1559            img.load()
1560            return img
1561
1562        try:
1563            return await asyncio.to_thread(_open)
1564        except OSError as err:
1565            self.logger.debug("Skipping undecodable artwork: %s", err)
1566            return None
1567
1568    def _on_player_media_updated(self) -> None:
1569        """Handle callback when the current media of the player is updated."""
1570        if self.synced_to is not None:
1571            # Only leader sends metadata
1572            return
1573        self.mass.create_task(
1574            self.send_current_media_metadata(),
1575            task_id=f"sendspin_metadata_{self.player_id}",
1576            abort_existing=True,
1577        )
1578
1579    async def _clear_current_media_metadata(self) -> None:
1580        """Clear all metadata and artwork from the sendspin group."""
1581        # Stop any in-flight beat-analysis polling task
1582        self._cancel_beat_retry()
1583        if (metadata_role := self._metadata_role) is not None:
1584            metadata_role.set_metadata(Metadata())
1585        if (visualizer_role := self._visualizer_role) is not None:
1586            visualizer_role.clear_beat_schedule()
1587            # Reset to PENDING so beats are re-deferred until the next track's analysis lands.
1588            visualizer_role.set_beat_availability(BeatAvailability.PENDING)
1589        if (artwork_role := self._artwork_role) is not None:
1590            await artwork_role.set_album_artwork(None)
1591            await artwork_role.set_artist_artwork(None)
1592        if (color_role := self._color_role) is not None:
1593            color_role.clear()
1594        if (controller_role := self._controller_role) is not None:
1595            controller_role.set_seek_max_ms(None)
1596        self.last_sent_artwork_url = None
1597        self.last_sent_artist_artwork_url = None
1598        self._last_beat_queue_item_id = None
1599        self._last_beat_anchor_us = None
1600
1601    def _publish_repeat_shuffle(self, repeat: SendspinRepeatMode, *, shuffle: bool) -> None:
1602        """
1603        Push repeat/shuffle to controller state for current-spec clients.
1604
1605        Clients implementing the older spec version still read the copy mirrored
1606        onto metadata state, for now.
1607        """
1608        if (controller_role := self._controller_role) is not None:
1609            controller_role.set_repeat(repeat)
1610            controller_role.set_shuffle(shuffle=shuffle)
1611
1612    def _compute_track_progress_ms(self, current_media: PlayerMedia, *, is_playing: bool) -> int:
1613        """
1614        Resolve current track position in ms from queue/media elapsed time.
1615
1616        Prefer queue/media elapsed as source of truth. Only interpolate while
1617        actively playing; for paused/idle states keep the last fixed position.
1618        """
1619        elapsed_time: float | None = (
1620            float(current_media.elapsed_time) if current_media.elapsed_time is not None else None
1621        )
1622        if is_playing and current_media.corrected_elapsed_time is not None:
1623            elapsed_time = current_media.corrected_elapsed_time
1624        if elapsed_time is None:
1625            elapsed_time = self.corrected_elapsed_time if is_playing else self.elapsed_time
1626        return max(0, int(elapsed_time * 1000)) if elapsed_time is not None else 0
1627
1628    async def _refresh_beat_schedule(self) -> None:
1629        """
1630        Re-attempt beat hydration only, without re-pushing metadata/artwork.
1631
1632        Lighter than `send_current_media_metadata`: the retry poller uses this so
1633        late-arriving analysis lands without re-running the full pipeline.
1634        """
1635        current_media = self.state.current_media
1636        if current_media is None:
1637            return
1638        queue_item: QueueItem | None = None
1639        queue: PlayerQueue | None = None
1640        if current_media.source_id and current_media.queue_item_id:
1641            queue = self.mass.player_queues.get(current_media.source_id)
1642            queue_item = self.mass.player_queues.get_item(
1643                current_media.source_id, current_media.queue_item_id
1644            )
1645        is_playing = self.state.playback_state == PlaybackState.PLAYING
1646        track_progress = self._compute_track_progress_ms(current_media, is_playing=is_playing)
1647        await self._send_beat_schedule(queue, queue_item, track_progress, is_playing)
1648
1649    async def send_current_media_metadata(self) -> None:
1650        """Send the current media metadata to the sendspin group."""
1651        if not self.available:
1652            return
1653        current_media = self.state.current_media
1654        if current_media is None:
1655            await self._clear_current_media_metadata()
1656            return
1657        # check if we are playing a MA queue item
1658        queue_item: QueueItem | None = None
1659        queue: PlayerQueue | None = None
1660        if current_media.source_id and current_media.queue_item_id:
1661            queue = self.mass.player_queues.get(current_media.source_id)
1662            queue_item = self.mass.player_queues.get_item(
1663                current_media.source_id, current_media.queue_item_id
1664            )
1665
1666        # Runs even without a queue item so radio / Spotify Connect streams still get art.
1667        await self._send_album_artwork(current_media)
1668        if queue_item:
1669            await self._send_artist_artwork(queue_item)
1670
1671        track_number: int | None = None
1672        year: int | None = None
1673        album_artist: str | None = None
1674        if queue_item and queue_item.media_item and is_track(queue_item.media_item):
1675            track = queue_item.media_item
1676            track_number = track.track_number or None
1677            album_mapping = track.album
1678            if album_mapping is not None:
1679                year = album_mapping.year
1680                if not isinstance(album_mapping, Album):
1681                    # Cheap DB-only lookup, no external API call; None if not in library
1682                    result = await self.mass.music.get_library_item_by_prov_id(
1683                        MediaType.ALBUM, album_mapping.item_id, album_mapping.provider
1684                    )
1685                    full_album: Album | None = result if isinstance(result, Album) else None
1686                else:
1687                    full_album = album_mapping
1688                if full_album and full_album.artists:
1689                    album_artist = full_album.artist_str
1690
1691        track_duration = current_media.duration or 0
1692        if controller_role := self._controller_role:
1693            controller_role.set_seek_max_ms(int(track_duration * 1000) if track_duration else None)
1694        repeat = SendspinRepeatMode.OFF
1695        if queue and queue.repeat_mode == RepeatMode.ALL:
1696            repeat = SendspinRepeatMode.ALL
1697        elif queue and queue.repeat_mode == RepeatMode.ONE:
1698            repeat = SendspinRepeatMode.ONE
1699
1700        shuffle = queue.shuffle_enabled if queue else False
1701        is_playing = self.state.playback_state == PlaybackState.PLAYING
1702        track_progress = self._compute_track_progress_ms(current_media, is_playing=is_playing)
1703
1704        metadata = Metadata(
1705            title=current_media.title,
1706            artist=current_media.artist,
1707            album_artist=album_artist,
1708            album=current_media.album,
1709            artwork_url=current_media.image_url,
1710            year=year,
1711            track=track_number,
1712            track_duration=track_duration * 1000 if track_duration is not None else None,
1713            track_progress=track_progress,
1714            playback_speed=1000 if is_playing else 0,
1715            repeat=repeat,
1716            shuffle=shuffle,
1717        )
1718
1719        # Send metadata to the group
1720        if (metadata_role := self._metadata_role) is not None:
1721            metadata_role.set_metadata(metadata)
1722
1723        self._publish_repeat_shuffle(repeat, shuffle=shuffle)
1724
1725        # Send color palette derived from the cover art (already computed by
1726        # the players controller with the Sendspin defined minimum contrast values).
1727        if (color_role := self._color_role) is not None:
1728            self._send_color_palette(color_role, current_media.palette)
1729
1730        await self._send_beat_schedule(queue, queue_item, track_progress, is_playing)
1731
1732    def _send_color_palette(
1733        self, color_role: ColorGroupRole, palette: MediaItemPalette | None
1734    ) -> None:
1735        """Push the palette already attached to current_media to the sendspin group."""
1736        if palette is None:
1737            color_role.clear()
1738            return
1739        color_role.set_color(
1740            Color(
1741                background_dark=palette.background_dark,
1742                background_light=palette.background_light,
1743                primary=palette.primary,
1744                accent=palette.accent,
1745                on_dark=palette.on_dark,
1746                on_light=palette.on_light,
1747            )
1748        )
1749
1750    @staticmethod
1751    def _flow_track_offset_us(pq_data: PlayerQueueData | None, queue_item: QueueItem) -> int | None:
1752        """
1753        Return the current track's flow-stream start offset (minus file seek), in µs.
1754
1755        Sums the streamed duration of every track committed before the current
1756        one in the queue-flow stream, so beats anchor to this track's start
1757        rather than the first track's. Returns None when the flow log hasn't
1758        recorded this track yet, so the caller falls back to the progress anchor.
1759
1760        A track whose intro was crossfade-trimmed loses its raw seek position
1761        (only the elapsed-inflated value survives), so its anchor can be off by
1762        up to the crossfade duration. Exact handling needs a raw-seek field on
1763        the flow log entry.
1764        """
1765        if pq_data is None or queue_item.streamdetails is None:
1766            return None
1767        log = pq_data.flow_mode_stream_log
1768        if not log or log[-1].queue_item_id != queue_item.queue_item_id:
1769            return None
1770        track_flow_start_s = sum(entry.seconds_streamed or 0.0 for entry in log[:-1])
1771        file_seek_s = float(queue_item.streamdetails.seek_position or 0)
1772        return int((track_flow_start_s - file_seek_s) * 1_000_000)
1773
1774    async def _send_beat_schedule(
1775        self,
1776        queue: PlayerQueue | None,
1777        queue_item: QueueItem | None,
1778        track_progress_ms: int,
1779        is_playing: bool,
1780    ) -> None:
1781        """Hydrate per-track beat timings from audio analysis and push to visualizer."""
1782        visualizer_role = self._visualizer_role
1783        if visualizer_role is None:
1784            return
1785        if not is_playing or queue_item is None or queue_item.streamdetails is None:
1786            visualizer_role.clear_beat_schedule()
1787            self._last_beat_queue_item_id = None
1788            self._last_beat_anchor_us = None
1789            self._cancel_beat_retry()
1790            return
1791        # smart_fades is the only AA provider that emits beats. Without it, no
1792        # beats will ever arrive for this source.
1793        if not any(
1794            p.available and p.domain == "smart_fades"
1795            for p in self.mass.get_providers(ProviderType.AUDIO_ANALYSIS)
1796        ):
1797            visualizer_role.clear_beat_schedule()
1798            visualizer_role.set_beat_availability(BeatAvailability.UNAVAILABLE)
1799            self._last_beat_queue_item_id = None
1800            self._last_beat_anchor_us = None
1801            self._cancel_beat_retry()
1802            return
1803        provider = cast("SendspinProvider", self.provider)
1804        now_us = provider.server_api.clock.now_us()
1805        # Anchor beats to the current track's spot in the flow stream's audio
1806        # timeline. Falls back to the progress-derived anchor until the first
1807        # chunk commits or while the flow log hasn't recorded this track yet.
1808        anchor_us: int | None = None
1809        pq_data = self.mass.player_queues.queue_data_or_none(queue.queue_id) if queue else None
1810        offset_us = self._flow_track_offset_us(pq_data, queue_item)
1811        if offset_us is not None:
1812            anchor_us = self.playback_session.flow_track_anchor_us(offset_us)
1813        if anchor_us is None:
1814            anchor_us = now_us - track_progress_ms * 1000
1815        # Re-push only on track change or seek (anchor jumps beyond natural drift).
1816        if (
1817            queue_item.queue_item_id == self._last_beat_queue_item_id
1818            and self._last_beat_anchor_us is not None
1819            and abs(anchor_us - self._last_beat_anchor_us) < 500_000
1820        ):
1821            return
1822        sd = queue_item.streamdetails
1823        analysis = await self.mass.streams.audio_analysis.get_audio_analysis(
1824            sd.item_id,
1825            sd.provider,
1826            media_type=sd.media_type,
1827            priority=(SMART_FADES_ANALYSIS_DOMAIN,),
1828        )
1829        if analysis is None or analysis.beats is None or len(analysis.beats) == 0:
1830            visualizer_role.clear_beat_schedule()
1831            # Analysis may still be running (offline NN takes ~5-10 s). Kick a
1832            # poller so beats land once available, without waiting for the
1833            # next player media update.
1834            # This could be solved more elegantly with an event instead of this
1835            # poller, but that means changes outside the sendspin provider, which
1836            # is riskier.
1837            self._schedule_beat_retry(queue_item.queue_item_id)
1838            return  # don't poison the cache; retry asynchronously
1839        # Analysis is in — no more retries needed.
1840        self._cancel_beat_retry()
1841        downbeats = (
1842            {float(d) for d in analysis.downbeats} if analysis.downbeats is not None else set()
1843        )
1844        beats: list[BeatTiming] = []
1845        for b in analysis.beats:
1846            beat_us = int(anchor_us + float(b) * 1_000_000)
1847            if beat_us < now_us:
1848                continue
1849            beats.append(BeatTiming(timestamp_us=beat_us, is_downbeat=float(b) in downbeats))
1850        visualizer_role.clear_beat_schedule()
1851        if beats:
1852            visualizer_role.append_beat_schedule(beats)
1853        self._last_beat_queue_item_id = queue_item.queue_item_id
1854        self._last_beat_anchor_us = anchor_us
1855
1856    # Initial backoff for the beat-analysis poller. The neural beat tracker
1857    # in smart_fades takes ~5-10 s; retry every 3 s until it lands. Capped so
1858    # tracks that won't ever have analysis don't poll forever.
1859    _BEAT_RETRY_INTERVAL_S: ClassVar[float] = 3.0
1860    _BEAT_RETRY_MAX_ATTEMPTS: ClassVar[int] = 30  # ~90 s of wait
1861
1862    def _schedule_beat_retry(self, queue_item_id: str) -> None:
1863        """
1864        Start the background poller for late-arriving beat analysis.
1865
1866        If a poller is already running for this queue item, no-op.
1867        """
1868        if (
1869            self._beat_retry_task is not None
1870            and not self._beat_retry_task.done()
1871            and self._beat_retry_queue_item_id == queue_item_id
1872        ):
1873            return
1874        self._cancel_beat_retry()
1875        self._beat_retry_queue_item_id = queue_item_id
1876        self._beat_retry_task = asyncio.create_task(self._beat_retry_loop(queue_item_id))
1877
1878    def _cancel_beat_retry(self) -> None:
1879        """Cancel the beat-analysis poller if running."""
1880        if self._beat_retry_task is not None and not self._beat_retry_task.done():
1881            self._beat_retry_task.cancel()
1882        self._beat_retry_task = None
1883        self._beat_retry_queue_item_id = None
1884
1885    async def _beat_retry_loop(self, queue_item_id: str) -> None:
1886        """Retry beat-schedule hydration until beats land or the track changes."""
1887        try:
1888            for _ in range(self._BEAT_RETRY_MAX_ATTEMPTS):
1889                await asyncio.sleep(self._BEAT_RETRY_INTERVAL_S)
1890                if self._beat_retry_queue_item_id != queue_item_id:
1891                    return  # superseded by another poller
1892                current = self.current_media
1893                if current is None or current.queue_item_id != queue_item_id:
1894                    return  # track changed
1895                if self._last_beat_queue_item_id == queue_item_id:
1896                    return  # beats were pushed via some other path
1897                # Re-attempt beat hydration only; _send_beat_schedule will push
1898                # beats now or schedule another retry.
1899                try:
1900                    await self._refresh_beat_schedule()
1901                except Exception:
1902                    self.logger.exception("Beat-schedule retry failed for %s", queue_item_id)
1903                    continue
1904                if self._last_beat_queue_item_id == queue_item_id:
1905                    return
1906            # Cap exhausted without beats landing. Flip the visualizer to
1907            # UNAVAILABLE so clients (Hue, web) stop waiting and the peak
1908            # walker / static palette path takes over for this track.
1909            if (
1910                self._beat_retry_queue_item_id == queue_item_id
1911                and (current := self.current_media) is not None
1912                and current.queue_item_id == queue_item_id
1913                and (visualizer_role := self._visualizer_role) is not None
1914            ):
1915                visualizer_role.set_beat_availability(BeatAvailability.UNAVAILABLE)
1916        except asyncio.CancelledError:
1917            return
1918        finally:
1919            if self._beat_retry_queue_item_id == queue_item_id:
1920                self._beat_retry_queue_item_id = None
1921
1922    async def get_config_entries(self) -> list[ConfigEntry]:
1923        """Return all (provider/player specific) Config Entries for the player."""
1924        entries: list[ConfigEntry] = []
1925        # Show alert if this Cast device is known to lack AudioContext support
1926        if self.mass.config.get_raw_player_config_value(
1927            self.player_id, CONF_CAST_AUDIO_UNSUPPORTED
1928        ):
1929            entries.append(
1930                ConfigEntry(
1931                    key="cast_audio_unsupported",
1932                    type=ConfigEntryType.ALERT,
1933                    required=False,
1934                )
1935            )
1936        entries.extend(await super().get_config_entries())
1937        # Build dynamic format options from player's supported formats
1938        player_role = self._player_role
1939        if player_role is not None:
1940            supported_formats = player_role.get_supported_formats()
1941            if supported_formats:
1942                format_options = [
1943                    ConfigValueOption(SENDSPIN_FORMAT_AUTOMATIC),
1944                ]
1945                for fmt in supported_formats:
1946                    format_options.append(
1947                        ConfigValueOption(
1948                            format_to_option_value(fmt), title=format_to_display_string(fmt)
1949                        )
1950                    )
1951                entries.append(
1952                    ConfigEntry(
1953                        key=CONF_PREFERRED_SENDSPIN_FORMAT,
1954                        type=ConfigEntryType.STRING,
1955                        category="protocol_generic",
1956                        default_value=SENDSPIN_FORMAT_AUTOMATIC,
1957                        options=format_options,
1958                        advanced=True,
1959                    )
1960                )
1961
1962        if (
1963            player_role is not None
1964            and PlayerCommand.SET_STATIC_DELAY in player_role.state_supported_commands
1965        ):
1966            entries.append(
1967                ConfigEntry(
1968                    key=CONF_SENDSPIN_STATIC_DELAY,
1969                    type=ConfigEntryType.INTEGER,
1970                    required=False,
1971                    default_value=self.static_delay_default_ms,
1972                    range=(0, 5000),
1973                    immediate_apply=True,
1974                    # Not a advanced option since this will only show up for players where it is likely
1975                    # necessary to adjust the delay.
1976                    advanced=False,
1977                )
1978            )
1979
1980        if self._hass_announce_entity_id is not None:
1981            # announcements are relayed to the device via Home Assistant,
1982            # which has no volume control for announcements
1983            entries.extend(HIDDEN_ANNOUNCE_VOLUME_CONFIG_ENTRIES)
1984
1985        return entries
1986
1987    async def on_unload(self) -> None:
1988        """Handle logic when the player is unloaded from the Player controller."""
1989        await self.playback_session.close()
1990        await super().on_unload()
1991
1992
1993class SendspinVisualizerPlayer(SendspinBasePlayer):
1994    """A non-audio Sendspin player for visualizer/lighting devices."""
1995
1996    _attr_type = PlayerType.VISUALIZER
1997    _attr_hidden_by_default = True
1998    _attr_expose_to_ha_by_default = False
1999
2000    def __init__(
2001        self,
2002        provider: SendspinProvider,
2003        player_id: str,
2004        initial_hello: ClientHelloPayload | None = None,
2005    ) -> None:
2006        """
2007        Initialize the visualizer player.
2008
2009        :param provider: The Sendspin provider instance.
2010        :param player_id: The unique player identifier.
2011        :param initial_hello: Optional hello payload from the client.
2012        """
2013        super().__init__(provider, player_id, initial_hello)
2014        self._attr_supported_features = {PlayerFeature.SET_MEMBERS}
2015
2016    async def set_members(
2017        self,
2018        player_ids_to_add: list[str] | None = None,
2019        player_ids_to_remove: list[str] | None = None,
2020    ) -> None:
2021        """Handle SET_MEMBERS command for the visualizer player."""
2022        for player_id in player_ids_to_remove or []:
2023            member = self.mass.players.get_player(player_id, True)
2024            if isinstance(member, SendspinBasePlayer):
2025                await self.api.group.remove_client(member.api)
2026        for player_id in player_ids_to_add or []:
2027            member = self.mass.players.get_player(player_id, True)
2028            if isinstance(member, SendspinBasePlayer):
2029                await self.api.group.add_client(member.api)
2030