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