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