music-assistant-server

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