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