music-assistant-server

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