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