music-assistant-server

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