music-assistant-server

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