music-assistant-server

1"""Player Provider for Sendspin."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import re
8from collections.abc import Callable
9from contextlib import suppress
10from copy import deepcopy
11from dataclasses import dataclass, field
12from ipaddress import ip_address
13from pathlib import Path
14from typing import TYPE_CHECKING, cast
15from urllib.parse import urlsplit
16from uuid import uuid4
17
18from aiosendspin.models.core import ClientHelloPayload
19from aiosendspin.models.core import DeviceInfo as SendspinDeviceInfo
20from aiosendspin.models.player import ClientHelloPlayerSupport, SupportedAudioFormat
21from aiosendspin.models.types import (
22    AudioCodec,
23    ManagementResult,
24    PairAbortReason,
25    PairMethod,
26    PlayerCommand,
27    role_family,
28)
29from aiosendspin.noise.driver import HandshakeAbortedError
30from aiosendspin.noise.pairing import (
31    LocalPairingAbortError,
32    PairingAbortError,
33    PairingAttempt,
34    PairingError,
35)
36from aiosendspin.noise.pairing_token import decode_token
37from aiosendspin.noise.trust_store import FileServerPairingStore
38from aiosendspin.server import (
39    ClientAddedEvent,
40    ClientRemovedEvent,
41    ClientUpdatedEvent,
42    SendspinEvent,
43    SendspinServer,
44)
45from music_assistant_models.config_entries import ConfigEntry
46from music_assistant_models.enums import (
47    ConfigEntryType,
48    EventType,
49    IdentifierType,
50    PlayerFeature,
51    PlayerType,
52    ProviderFeature,
53)
54from music_assistant_models.errors import AlreadyRegisteredError, SetupFailedError
55
56from music_assistant.constants import (
57    CONF_ENABLED,
58    CONF_ENTRY_MANUAL_DISCOVERY_IPS,
59    CONF_LOG_LEVEL,
60    CONF_PLAYERS,
61    CONF_PROVIDERS,
62    SENDSPIN_SERVER_PORT,
63    VERBOSE_LOG_LEVEL,
64)
65from music_assistant.helpers.util import format_ip_for_url
66from music_assistant.mass import MusicAssistant
67from music_assistant.models.player import Player
68from music_assistant.models.player_provider import PlayerProvider
69from music_assistant.providers.sendspin.bridge_role import (
70    BRIDGE_BIT_DEPTH,
71    BRIDGE_CHANNELS,
72    BRIDGE_ROLE_ID,
73    BRIDGE_SAMPLE_RATE,
74    BridgePlayerRole,
75)
76from music_assistant.providers.sendspin.constants import (
77    CONF_ALLOW_UNENCRYPTED,
78    CONF_MIN_PIN_LENGTH,
79    CONF_SENDSPIN_STATIC_DELAY,
80    CONF_VIRTUAL_PLAYER_OWNER,
81    DEFAULT_MIN_PIN_LENGTH,
82    VIRTUAL_PLAYER_ID_PREFIX,
83)
84from music_assistant.providers.sendspin.helpers import (
85    SecurityActionError,
86    effective_pair_methods,
87)
88from music_assistant.providers.sendspin.player import (
89    SendspinBasePlayer,
90    SendspinPlayer,
91    SendspinVisualizerPlayer,
92)
93from music_assistant.providers.sendspin.security import (
94    IDENTITY_FILENAME,
95    get_or_create_server_identity,
96)
97
98if TYPE_CHECKING:
99    from collections.abc import Awaitable, Sequence
100
101    from aiosendspin.models.core import PairMethodDescriptor
102    from aiosendspin.models.management import (
103        ManagementResultData,
104        ManagementSetPairingConfigPayload,
105    )
106    from aiosendspin.server.client import SendspinClient
107    from aiosendspin.server.connection import SendspinConnection
108    from aiosendspin.server.server import ExternalStreamStartRequest
109    from music_assistant_models.config_entries import ProviderConfig
110    from music_assistant_models.event import MassEvent
111    from music_assistant_models.provider import ProviderManifest
112
113    from music_assistant.providers.hass import HomeAssistantProvider
114
115
116DEFAULT_SENDSPIN_CLIENT_PORT = 8928
117DEFAULT_SENDSPIN_CLIENT_PATH = "/sendspin"
118VIRTUAL_PLAYER_REGISTER_TIMEOUT = 10.0
119VIRTUAL_PLAYER_CLEANUP_DELAYS = (0.0, 0.5, 2.0)
120VIRTUAL_PLAYER_CLEANUP_TIMEOUT = 2.0
121
122PIN_REQUEST_FEEDBACK_TIMEOUT = 2
123PIN_RETRY_IDLE_TIMEOUT = 300
124MANAGEMENT_REQUEST_TIMEOUT = 10
125MANAGEMENT_IDLE_TIMEOUT = 300
126
127
128@dataclass
129class PinPairingSession:
130    """State of an operator PIN pairing session for one client, across retry-in-place attempts."""
131
132    client_id: str
133    method: PairMethod
134    pin_future: asyncio.Future[str]
135    verify: bool = False
136    static: bool = False
137    task: asyncio.Task[None] | None = None
138    pin_request_event: asyncio.Event = field(default_factory=asyncio.Event)
139    error: Exception | None = None
140    retryable: bool = False
141
142    @property
143    def attempt_running(self) -> bool:
144        """Whether an attempt is currently in flight."""
145        return self.task is not None and not self.task.done()
146
147    @property
148    def awaiting_gesture(self) -> bool:
149        """Whether the attempt is still waiting for the client to enter pairing."""
150        return self.attempt_running and not self.pin_request_event.is_set()
151
152    @property
153    def awaiting_pin(self) -> bool:
154        """Whether the attempt is waiting for the operator to submit a PIN."""
155        return self.attempt_running and not self.pin_future.done()
156
157    @property
158    def can_retry(self) -> bool:
159        """Whether a failed attempt can be retried in place."""
160        return self.task is not None and self.task.done() and self.retryable
161
162    @property
163    def finished(self) -> bool:
164        """Whether the session reached a terminal outcome (no retry possible)."""
165        return self.task is not None and self.task.done() and not self.retryable
166
167
168@dataclass
169class ManagementSession:
170    """State of an operator device-management session for one client."""
171
172    client_id: str
173    connection: SendspinConnection
174    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
175
176
177def _pin_idle_task_id(client_id: str) -> str:
178    """Timer/task id for a client's pairing-retry idle timeout."""
179    return f"sendspin_pin_idle_{client_id}"
180
181
182def _management_idle_task_id(client_id: str) -> str:
183    """Timer/task id for a client's management-session idle timeout."""
184    return f"sendspin_management_idle_{client_id}"
185
186
187_MANAGEMENT_RESULT_ALERTS = {
188    ManagementResult.PERMISSION_DENIED: "management_error_permission_denied",
189    ManagementResult.ALREADY_EXISTS: "management_error_already_exists",
190    ManagementResult.INVALID: "management_error_invalid",
191    ManagementResult.NOT_FOUND: "management_error_not_found",
192    ManagementResult.STORAGE_EXHAUSTED: "management_error_storage_exhausted",
193}
194
195
196def _check_management_result(result: ManagementResult) -> None:
197    """Raise a structured error for a non-ok management result."""
198    if result is ManagementResult.OK:
199        return
200    alert_key = _MANAGEMENT_RESULT_ALERTS.get(result)
201    if alert_key is None:
202        raise SecurityActionError("management_error_generic", detail=result.value)
203    raise SecurityActionError(alert_key)
204
205
206def _manual_client_url(address: str) -> str:
207    """Convert a manually configured Sendspin host/IP to a client WebSocket URL."""
208    stripped_address = address.strip()
209    if not stripped_address:
210        raise ValueError("Address is empty")
211
212    if "://" in stripped_address:
213        return stripped_address
214
215    try:
216        parsed_ip = ip_address(stripped_address)
217    except ValueError:
218        pass
219    else:
220        return (
221            f"ws://{format_ip_for_url(str(parsed_ip))}:"
222            f"{DEFAULT_SENDSPIN_CLIENT_PORT}{DEFAULT_SENDSPIN_CLIENT_PATH}"
223        )
224
225    parsed_address = urlsplit(f"//{stripped_address}")
226    if parsed_address.hostname is None:
227        raise ValueError("Address does not contain a host")
228
229    return (
230        f"ws://{format_ip_for_url(parsed_address.hostname)}:"
231        f"{parsed_address.port or DEFAULT_SENDSPIN_CLIENT_PORT}"
232        f"{parsed_address.path or DEFAULT_SENDSPIN_CLIENT_PATH}"
233    )
234
235
236class SendspinProvider(PlayerProvider):
237    """Player Provider for Sendspin."""
238
239    reload_on_streams_network_change = True
240    server_api: SendspinServer
241    unregister_cbs: list[Callable[[], None]]
242    _pending_unregisters: dict[str, asyncio.Event]
243    _bridge_identifiers: dict[str, dict[IdentifierType, str]]
244    _bridge_underlying_players: dict[str, str]
245    _bridge_static_delay_defaults: dict[str, int]
246    _client_event_versions: dict[str, int]
247    _client_event_task_counts: dict[str, int]
248    _manual_ip_config: tuple[str, ...]
249    _virtual_players: dict[str, str]
250    _unloading: bool
251    _hass_available: bool
252
253    def __init__(
254        self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
255    ) -> None:
256        """Initialize a new Sendspin player provider."""
257        super().__init__(mass, manifest, config)
258        # Handle config option for manual IP's. Read a default here: at construction the
259        # config only carries the server defaults + stored raw values (the provider's typed
260        # option entries are resolved and applied by the config controller right after this).
261        manual_ip_config = cast(
262            "list[str]", config.get_value(CONF_ENTRY_MANUAL_DISCOVERY_IPS.key) or []
263        )
264        self._manual_ip_config = tuple(address for address in manual_ip_config if address.strip())
265        self._pending_unregisters = {}
266        self._bridge_identifiers = {}
267        self._headless_client_ids: set[str] = set()
268        self._bridge_underlying_players = {}
269        self._bridge_static_delay_defaults = {}
270        self._bridge_player_types: dict[str, PlayerType] = {}
271        self._client_event_versions = {}
272        self._client_event_task_counts = {}
273        self._virtual_players = {}
274        self._pin_sessions: dict[str, PinPairingSession] = {}
275        self._management_sessions: dict[str, ManagementSession] = {}
276        self._pairing_config_snapshots: dict[
277            str, tuple[SendspinConnection, ManagementResultData]
278        ] = {}
279        self._unloading = False
280        self._hass_available = False
281        self.unregister_cbs = []
282
283    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
284        """Return Config entries to configure this provider."""
285        return (
286            CONF_ENTRY_MANUAL_DISCOVERY_IPS,
287            ConfigEntry(
288                key=CONF_ALLOW_UNENCRYPTED,
289                type=ConfigEntryType.BOOLEAN,
290                default_value=True,
291            ),
292            ConfigEntry(
293                key=CONF_MIN_PIN_LENGTH,
294                type=ConfigEntryType.INTEGER,
295                range=(4, 12),
296                default_value=DEFAULT_MIN_PIN_LENGTH,
297            ),
298        )
299
300    async def handle_async_init(self) -> None:
301        """Load the persistent server identity and pairing store, then create the server."""
302        self._set_aiosendspin_log_level()
303        storage_dir = Path(self.mass.storage_path) / "sendspin"
304        identity_path = storage_dir / IDENTITY_FILENAME
305        try:
306            identity = await asyncio.to_thread(get_or_create_server_identity, storage_dir)
307        except ValueError as err:
308            raise SetupFailedError(
309                f"The Sendspin server identity at {identity_path} is corrupt: {err}. Restore it "
310                "from a backup, or remove the file to start fresh - every paired device will "
311                "then need to be re-paired."
312            ) from err
313        except OSError as err:
314            raise SetupFailedError(
315                f"Could not read the Sendspin server identity at {identity_path}: {err}. Fix the "
316                "file-access problem and reload; do not delete the file or every paired device "
317                "will need to be re-paired."
318            ) from err
319        pairing_store_path = storage_dir / "pairing_store.json"
320        try:
321            pairing_store = await FileServerPairingStore.open(pairing_store_path)
322        except (ValueError, TypeError, KeyError) as err:
323            raise SetupFailedError(
324                f"The Sendspin pairing store at {pairing_store_path} is corrupt: {err}. Restore it "
325                "from a backup, or remove the file to start fresh - this discards all pairings and "
326                "unpaired-access approvals."
327            ) from err
328        except OSError as err:
329            raise SetupFailedError(
330                f"Could not read the Sendspin pairing store at {pairing_store_path}: {err}. Fix the "
331                "file-access problem and reload; do not delete the file or all pairings will be "
332                "lost."
333            ) from err
334        self.server_api = SendspinServer(
335            self.mass.loop,
336            identity,
337            "Music Assistant",
338            self.mass.http_session,
339            pairing_store=pairing_store,
340            allow_unencrypted=cast("bool", self.config.get_value(CONF_ALLOW_UNENCRYPTED, True)),
341            min_pin_length=cast(
342                "int", self.config.get_value(CONF_MIN_PIN_LENGTH, DEFAULT_MIN_PIN_LENGTH)
343            ),
344        )
345        # Pitch (YINFFT) is the heaviest visualizer DSP and result quality is
346        # still very mixed, needs more testing. Disable it globally for now to
347        # spare low-power hosts.
348        self.server_api.set_visualizer_pitch_enabled(enabled=False)
349        self.unregister_cbs = [
350            self.server_api.add_event_listener(self.event_cb),
351            self.mass.subscribe(self._on_providers_updated, EventType.PROVIDERS_UPDATED),
352        ]
353        # seed the hass availability snapshot so the first (un)load is seen as a change
354        hass = self.mass.get_provider("hass")
355        self._hass_available = hass is not None and hass.available
356
357    async def update_config(self, config: ProviderConfig, changed_keys: set[str]) -> None:
358        """Handle logic when the config is updated."""
359        await super().update_config(config, changed_keys)
360        # a log level(-only) change does not reload the provider,
361        # so realign aiosendspin's logger here
362        if f"values/{CONF_LOG_LEVEL}" in changed_keys:
363            self._set_aiosendspin_log_level()
364
365    def event_cb(self, server: SendspinServer, event: SendspinEvent) -> None:
366        """Event callback registered to the sendspin server."""
367        match event:
368            case ClientAddedEvent(client_id):
369                event_version = self._begin_client_event(client_id)
370                self.mass.create_task(self._handle_client_added(client_id, event_version))
371            case ClientRemovedEvent(client_id):
372                event_version = self._begin_client_event(client_id)
373                self.mass.create_task(self._handle_client_removed(client_id, event_version))
374            case ClientUpdatedEvent(client_id):
375                event_version = self._begin_client_event(client_id)
376                self.mass.create_task(self._handle_client_updated(client_id, event_version))
377            case _:
378                self.logger.error("Unknown sendspin event: %s", event)
379
380    def on_player_enabled(self, player_id: str) -> None:
381        """Call (by config manager) when a player gets enabled."""
382        # A client that connected while disabled has no player object;
383        # replay the add event so re-enabling takes effect immediately.
384        if (
385            self.server_api.get_client(player_id) is not None
386            and self.mass.players.get_player(player_id) is None
387        ):
388            event_version = self._begin_client_event(player_id)
389            self.mass.create_task(self._handle_client_added(player_id, event_version))
390            return
391        super().on_player_enabled(player_id)
392
393    def register_bridge_identifiers(
394        self, client_id: str, identifiers: dict[IdentifierType, str]
395    ) -> None:
396        """
397        Pre-register extra identifiers for a bridge client.
398
399        Called by bridge managers (Chromecast, AirPlay) before registering an
400        external player, so that the resulting SendspinPlayer carries the parent
401        player's protocol-specific identifiers for cross-protocol matching.
402
403        :param client_id: The bridge client_id that will be used for registration.
404        :param identifiers: Extra identifiers to attach to the SendspinPlayer.
405        """
406        self._bridge_identifiers[client_id] = identifiers
407
408    def register_bridge_underlying_player(self, client_id: str, underlying_player_id: str) -> None:
409        """
410        Pre-register the underlying player a bridge client runs on top of.
411
412        Called by bridge managers before registering an external player, so that
413        the resulting SendspinPlayer carries the derived-transport edge and the
414        protocol linking layer can resolve its parent deterministically.
415
416        :param client_id: The bridge client_id that will be used for registration.
417        :param underlying_player_id: The player_id of the player the bridge rides on.
418        """
419        self._bridge_underlying_players[client_id] = underlying_player_id
420
421    def register_bridge_static_delay_default(self, client_id: str, default_ms: int) -> None:
422        """
423        Register a protocol-specific default static delay for a bridge client.
424
425        If the SendspinPlayer already exists, the default is applied immediately;
426        otherwise it is stashed and picked up when the player is created.
427
428        :param client_id: The bridge client_id for which the default applies.
429        :param default_ms: Model-specific default static delay in milliseconds.
430        """
431        existing = self.mass.players.get_player(client_id)
432        if isinstance(existing, SendspinPlayer):
433            existing.static_delay_default_ms = default_ms
434            # If no user-set value exists, push the new default to the device now
435            # so already-connected clients pick it up without a config edit.
436            if (
437                self.mass.config.get_raw_player_config_value(client_id, CONF_SENDSPIN_STATIC_DELAY)
438                is None
439            ):
440                self.mass.create_task(existing._apply_static_delay())
441            return
442        self._bridge_static_delay_defaults[client_id] = default_ms
443
444    def register_headless_client(self, client_id: str) -> None:
445        """
446        Mark a client id as headless: no MA player is registered for it.
447
448        Called by in-process consumers (e.g. the MilkDrop visualizer tap) whose
449        Sendspin clients exist purely to receive group audio and must never
450        surface as players anywhere in the UI or API.
451        """
452        self._headless_client_ids.add(client_id)
453
454    def register_bridge_player_type(self, client_id: str, player_type: PlayerType) -> None:
455        """
456        Pre-register a PlayerType override for a bridge client.
457
458        Called by bridge managers to set the player type for the resulting
459        player (e.g. PlayerType.LIGHT for Hue Entertainment bridges).
460        """
461        self._bridge_player_types[client_id] = player_type
462
463    async def apply_bridge_claim(
464        self,
465        client_id: str,
466        identifiers: dict[IdentifierType, str],
467        bridge_hello: ClientHelloPayload,
468        underlying_player_id: str | None = None,
469    ) -> bool:
470        """
471        Post-claim an already-registered SendspinPlayer as a bridge client.
472
473        Used when a bridge manager reaches setup_bridge AFTER the external client
474        has already connected on its own (e.g. a JS Cast receiver reconnecting to
475        the server before the Chromecast bridge could register). Attaches the
476        bridge's protocol-specific identifiers so cross-protocol matching can
477        link the SendspinPlayer to its native peer, and replays the bridge
478        hello's supported_commands restriction on the player features.
479
480        :param client_id: The Sendspin client_id whose player should be claimed.
481        :param identifiers: Protocol-specific identifiers (e.g. CAST_UUID) to
482            attach to the player for cross-protocol matching.
483        :param bridge_hello: The bridge's intended ClientHelloPayload. Its
484            player_support.supported_commands gates which volume/mute features
485            the player is allowed to expose.
486        :param underlying_player_id: The player_id of the player the bridge rides
487            on, establishing the derived-transport edge for protocol linking.
488        :return: True if a matching SendspinPlayer was found and updated.
489        """
490        player = self.mass.players.get_player(client_id)
491        if not isinstance(player, SendspinPlayer):
492            return False
493        for id_type, id_value in identifiers.items():
494            player.device_info.add_identifier(id_type, id_value)
495        if underlying_player_id is not None:
496            player._attr_underlying_player_id = underlying_player_id
497        bridge_supported_commands: list[PlayerCommand] = []
498        if bridge_hello.player_support:
499            bridge_supported_commands = list(bridge_hello.player_support.supported_commands)
500        if PlayerCommand.VOLUME in bridge_supported_commands:
501            player._attr_supported_features.add(PlayerFeature.VOLUME_SET)
502        else:
503            player._attr_supported_features.discard(PlayerFeature.VOLUME_SET)
504        if PlayerCommand.MUTE in bridge_supported_commands:
505            player._attr_supported_features.add(PlayerFeature.VOLUME_MUTE)
506        else:
507            player._attr_supported_features.discard(PlayerFeature.VOLUME_MUTE)
508        # Expose the claimed player as a protocol bridge, not a standalone web
509        # player. A JS Cast receiver advertises product_name="Web Browser" and
510        # would otherwise be classified as is_web_player → PlayerType.PLAYER
511        # (hidden). Restore protocol semantics so UI links it under its native peer.
512        player.is_web_player = False
513        player._attr_hidden_by_default = False
514        player._attr_expose_to_ha_by_default = True
515        player._attr_type = PlayerType.PROTOCOL
516        self.logger.info(
517            "Bridge claim applied to existing SendspinPlayer %s (client_id=%s)",
518            player.display_name,
519            client_id,
520        )
521        await self.mass.players.register_or_update(player)
522        return True
523
524    async def create_virtual_player(
525        self,
526        owner_instance_id: str,
527        display_name: str,
528        player_id: str | None = None,
529    ) -> str:
530        """
531        Create a hidden, server-side virtual Sendspin player.
532
533        A virtual player owns its own PlayerQueue and leads a native Sendspin
534        group, but never renders audio itself: the audio stream is delivered to
535        the guest players that are attached to it through standard grouping.
536        It is hidden from the UI and not exposed to Home Assistant by default,
537        and is automatically removed when the owning provider unloads.
538
539        :param owner_instance_id: Instance id of the (loaded) provider that owns
540            the virtual player and controls its lifecycle.
541        :param display_name: Human readable name for the virtual player.
542        :param player_id: Optional stable id for the virtual player; a random id
543            is generated when omitted. The id is always prefixed with
544            ``VIRTUAL_PLAYER_ID_PREFIX``.
545        :return: The player_id of the registered virtual player.
546        :raises SetupFailedError: If the virtual player can not be created.
547        """
548        if (owner := self.mass.get_provider(owner_instance_id)) is None:
549            raise SetupFailedError(f"Owner provider {owner_instance_id} is not loaded")
550        if owner.instance_id != owner_instance_id and (
551            len(self.mass.get_provider_instances(owner.domain)) > 1
552        ):
553            raise SetupFailedError(
554                f"Multiple instances exist for {owner_instance_id}: pass an exact instance id"
555            )
556        # normalize a provider domain to the actual instance id
557        owner_instance_id = owner.instance_id
558        if player_id is None:
559            player_id = uuid4().hex
560        elif not re.fullmatch(r"[a-zA-Z0-9_-]+", player_id):
561            raise SetupFailedError(
562                f"Invalid player_id {player_id}: only alphanumerics, '_' and '-' are allowed"
563            )
564        if not player_id.startswith(VIRTUAL_PLAYER_ID_PREFIX):
565            player_id = f"{VIRTUAL_PLAYER_ID_PREFIX}{player_id}"
566        if player_id in self._virtual_players or self.server_api.get_client(player_id) is not None:
567            raise SetupFailedError(f"Virtual player {player_id} already exists")
568        # a persisted config may only be reclaimed by the same owner
569        stored_owner = self._get_virtual_player_config_owner(player_id)
570        if stored_owner is not None and stored_owner != owner_instance_id:
571            raise SetupFailedError(f"Virtual player {player_id} is owned by {stored_owner}")
572        self._virtual_players[player_id] = owner_instance_id
573        try:
574            self._register_virtual_player_client(player_id, display_name)
575            await self._wait_for_virtual_player(player_id)
576        except asyncio.CancelledError:
577            await self._cleanup_failed_virtual_player_creation(player_id)
578            raise
579        except Exception as err:
580            await self._cleanup_failed_virtual_player_creation(player_id)
581            if isinstance(err, SetupFailedError):
582                raise
583            raise SetupFailedError(f"Failed to create virtual player {player_id}") from err
584        # persist the owner so orphaned configs can be swept after a restart
585        self.mass.config.set_raw_player_config_value(
586            player_id, CONF_VIRTUAL_PLAYER_OWNER, owner_instance_id
587        )
588        self.logger.info("Virtual player %s created for %s", player_id, owner_instance_id)
589        return player_id
590
591    async def remove_virtual_player(self, player_id: str) -> None:
592        """
593        Remove a virtual Sendspin player and permanently delete its configuration.
594
595        :param player_id: The player_id returned by create_virtual_player.
596        :raises ValueError: If the given player_id is not a (known) virtual player.
597        """
598        if not player_id.startswith(VIRTUAL_PLAYER_ID_PREFIX) or (
599            player_id not in self._virtual_players
600            and self._get_virtual_player_config_owner(player_id) is None
601        ):
602            raise ValueError(f"{player_id} is not a virtual player")
603        # unregister the player first so the client removed event handler
604        # can not race us with a non-permanent unregister
605        await self.mass.players.unregister(player_id, permanent=True)
606        if self.server_api.get_client(player_id) is not None:
607            await self.server_api.remove_client(player_id)
608        # the config may linger when the player was never registered
609        self.mass.players.delete_player_config(player_id)
610        self._virtual_players.pop(player_id, None)
611        self.logger.info("Virtual player %s removed", player_id)
612
613    def is_virtual_player(self, player_id: str) -> bool:
614        """Return whether the given player_id belongs to a registered virtual player."""
615        return player_id in self._virtual_players
616
617    def get_pin_session(self, client_id: str) -> PinPairingSession | None:
618        """Return the in-flight or just-finished PIN pairing session for a client."""
619        return self._pin_sessions.get(client_id)
620
621    def clear_pin_session(self, client_id: str) -> None:
622        """Drop a finished PIN pairing session (after its outcome has been shown)."""
623        session = self._pin_sessions.get(client_id)
624        if session is not None and session.finished:
625            self._cancel_pin_idle_timeout(client_id)
626            self._pin_sessions.pop(client_id, None)
627
628    async def start_pin_pairing(
629        self, client_id: str, *, verify: bool = False, static: bool = False
630    ) -> PinPairingSession:
631        """
632        Begin (or retry in place) an operator PIN pairing attempt with a connected client.
633
634        Returns once the client has asked for the PIN, the attempt has failed, or a short
635        feedback window has elapsed, so the caller's first render reflects whether the
636        device-side pairing gesture is still pending. The attempt keeps running until the PIN
637        is supplied via submit_pin (or it times out / is cancelled). A session left retryable
638        by a failed attempt is resumed in place, preserving the chosen method and verify mode.
639
640        :param verify: Re-verify an already-paired device's presence (dynamic PIN only).
641        :param static: Pair with the static PIN even when a dynamic PIN is offered.
642        """
643        session = self._pin_sessions.get(client_id)
644        if session is not None and (session.verify != verify or session.static != static):
645            # a stale session from an earlier run never resumes; the caller's
646            # static/verify choice must win
647            if session.attempt_running:
648                raise SecurityActionError("pairing_error_concurrent")
649            await self.cancel_pin_pairing(client_id)
650            session = None
651        if session is not None and session.can_retry:
652            self._begin_pin_attempt(session)
653            await self._pin_request_feedback(session)
654            return session
655        if session is not None and session.attempt_running:
656            return session
657        client = self.server_api.get_client(client_id)
658        info = client.info_or_none if client is not None else None
659        if info is None:
660            raise SecurityActionError("pairing_error_not_connected")
661        method = self._pick_pin_method(
662            effective_pair_methods(info, self.pairing_config_snapshot(client_id)),
663            verify=verify,
664            static=static,
665        )
666        session = PinPairingSession(
667            client_id=client_id,
668            method=method,
669            pin_future=self.mass.loop.create_future(),
670            verify=verify,
671            static=static,
672        )
673        self._pin_sessions[client_id] = session
674        self._begin_pin_attempt(session)
675        await self._pin_request_feedback(session)
676        return session
677
678    def submit_pin(self, client_id: str, pin: str) -> None:
679        """Deliver the operator-entered PIN to the in-flight pairing attempt."""
680        session = self._pin_sessions.get(client_id)
681        if session is None or session.task is None:
682            raise SecurityActionError("pairing_error_no_pin_session")
683        if not session.pin_future.done():
684            session.pin_future.set_result(pin.strip())
685
686    async def cancel_pin_pairing(self, client_id: str) -> None:
687        """Abort an in-flight or parked PIN pairing session, restoring normal service."""
688        session = self._pin_sessions.pop(client_id, None)
689        if session is None:
690            return
691        self._cancel_pin_idle_timeout(client_id)
692        await self._end_pairing_quietly(client_id)
693        if session.task is not None:
694            with suppress(Exception):
695                await session.task
696        await self._refresh_player(client_id)
697
698    async def pair_with_token(self, client_id: str, token_value: str) -> None:
699        """Pair a connected client using its pasted pairing token."""
700        session = self._pin_sessions.get(client_id)
701        if session is not None and session.attempt_running:
702            raise SecurityActionError("pairing_error_concurrent")
703        try:
704            token = decode_token(token_value)
705        except ValueError as err:
706            raise SecurityActionError("pairing_error_token_invalid") from err
707        if token.client_id != client_id:
708            raise SecurityActionError("pairing_error_token_mismatch")
709        try:
710            await self.server_api.initiate_pairing(
711                client_id,
712                PairingAttempt(PairMethod.PAIRING_PSK, pairing_psk=token.pairing_psk),
713            )
714        except PairingAbortError:
715            # Token pairing is single-shot; unpark the connection before surfacing the failure.
716            await self._end_pairing_quietly(client_id)
717            raise
718        except HandshakeAbortedError as err:
719            # A client that does not recognize the token's PSK closes the connection
720            # without an application-level error (spec); the server has disconnected it.
721            raise PairingError(
722                "the token was rejected by the device; make sure it is correct"
723            ) from err
724        await self._refresh_player(client_id)
725
726    async def unpair_client(self, client_id: str) -> None:
727        """Drop the pairing with a connected client (both sides forget the credential)."""
728        await self.server_api.unpair(client_id)
729        await self._refresh_player(client_id)
730
731    async def set_trusted_unpaired(self, client_id: str, enabled: bool) -> None:
732        """Approve or revoke unpaired (unauthenticated) playback for a client."""
733        if enabled:
734            await self.server_api.trust_unpaired(client_id)
735        else:
736            await self.server_api.untrust_unpaired(client_id)
737        await self._refresh_player(client_id)
738
739    def get_management_session(self, client_id: str) -> ManagementSession | None:
740        """Return the client's management session, dropping one whose connection is gone."""
741        session = self._management_sessions.get(client_id)
742        if session is None:
743            return None
744        client = self.server_api.get_client(client_id)
745        if client is None or client.connection is not session.connection:
746            self._drop_management_session(session)
747            return None
748        return session
749
750    def enter_management(self, client_id: str) -> ManagementSession:
751        """Open (or refresh) the operator management session for a paired connected client."""
752        if (session := self.get_management_session(client_id)) is not None:
753            self._arm_management_idle_timeout(session)
754            return session
755        try:
756            connection = self.server_api.enable_management(client_id)
757        except RuntimeError as err:
758            raise SecurityActionError("management_error_generic", detail=str(err)) from err
759        session = ManagementSession(client_id=client_id, connection=connection)
760        self._management_sessions[client_id] = session
761        self._arm_management_idle_timeout(session)
762        return session
763
764    def exit_management(self, client_id: str) -> None:
765        """Close the client's management session, restoring normal server admission."""
766        if (session := self._management_sessions.get(client_id)) is not None:
767            self._drop_management_session(session)
768
769    async def management_get_pairing_config(self, client_id: str) -> ManagementResultData:
770        """Fetch the device's pairing configuration over its management session."""
771        session = self._management_session_or_raise(client_id)
772        async with session.lock:
773            self._arm_management_idle_timeout(session)
774            result, data, _ = await self._management_call(
775                session.connection, session.connection.get_pairing_config()
776            )
777        _check_management_result(result)
778        self._pairing_config_snapshots[client_id] = (session.connection, data)
779        return data
780
781    def pairing_config_snapshot(self, client_id: str) -> ManagementResultData | None:
782        """
783        Return the last management-fetched pairing config for the client's current connection.
784
785        While the connection it was fetched on is still the active one, the snapshot is
786        fresher than the hello advertisement (which cannot change until reconnect); after a
787        reconnect the new hello is authoritative and the snapshot is dropped.
788        """
789        snapshot = self._pairing_config_snapshots.get(client_id)
790        if snapshot is None:
791            return None
792        connection, data = snapshot
793        client = self.server_api.get_client(client_id)
794        if client is None or client.connection is not connection:
795            self._pairing_config_snapshots.pop(client_id, None)
796            return None
797        return data
798
799    async def management_set_pairing_config(
800        self, client_id: str, patch: ManagementSetPairingConfigPayload
801    ) -> None:
802        """Apply a pairing-config patch on the device and refresh the cached snapshot."""
803        session = self._management_session_or_raise(client_id)
804        async with session.lock:
805            self._arm_management_idle_timeout(session)
806            result = await self._management_call(
807                session.connection, session.connection.set_pairing_config(patch)
808            )
809        _check_management_result(result)
810        await self.management_get_pairing_config(client_id)
811
812    @property
813    def supported_features(self) -> set[ProviderFeature]:
814        """Return the features supported by this Provider."""
815        return {
816            ProviderFeature.SYNC_PLAYERS,
817        }
818
819    async def loaded_in_mass(self) -> None:
820        """Call after the provider has been loaded."""
821        await super().loaded_in_mass()
822        self._remove_orphan_virtual_player_configs()
823        # Start server for handling incoming Sendspin connections from clients
824        # and mDNS discovery of new clients
825        await self.server_api.start_server(
826            port=SENDSPIN_SERVER_PORT,
827            host=self.mass.streams.bind_ip,
828            advertise_addresses=[self.mass.streams.publish_ip],
829        )
830        for address in self._manual_ip_config:
831            try:
832                url = _manual_client_url(address)
833            except ValueError as err:
834                self.logger.warning(
835                    "Ignoring invalid manual Sendspin client address %s: %s", address, err
836                )
837                continue
838            self.logger.debug("Connecting to manually configured Sendspin client at %s", url)
839            self.server_api.connect_to_client(
840                url,
841                retry_initial_connection=True,
842                retry_indefinitely=True,
843            )
844
845    async def unload(self, is_removed: bool = False) -> None:
846        """
847        Handle unload/close of the provider.
848
849        Called when provider is deregistered (e.g. MA exiting or config reloading).
850
851        :param is_removed: True when the provider is removed from the configuration.
852        """
853        self._unloading = True
854        # call_later timers are not swept by mass.stop(), so cancel them explicitly here.
855        for session in self._pin_sessions.values():
856            if session.task is not None:
857                session.task.cancel()
858            self._cancel_pin_idle_timeout(session.client_id)
859        self._pin_sessions.clear()
860        for management_session in self._management_sessions.values():
861            self.mass.cancel_timer(_management_idle_task_id(management_session.client_id))
862        self._management_sessions.clear()
863        self._pairing_config_snapshots.clear()
864        player_ids = [player.player_id for player in self.players]
865        # Stop the Sendspin server
866        await self.server_api.close()
867
868        for cb in self.unregister_cbs:
869            cb()
870        self.unregister_cbs = []
871        self._client_event_task_counts.clear()
872        self._client_event_versions.clear()
873        self._virtual_players.clear()
874        await asyncio.gather(
875            *(
876                self.mass.players.unregister(player_id, permanent=is_removed)
877                for player_id in player_ids
878            ),
879            return_exceptions=True,
880        )
881
882    def _set_aiosendspin_log_level(self) -> None:
883        """Keep aiosendspin's (very chatty) logging quiet unless verbose logging is enabled."""
884        # aiosendspin logs every protocol message of every client session at debug
885        # level, so only pass that through when verbose logging is enabled
886        if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
887            logging.getLogger("aiosendspin").setLevel(logging.DEBUG)
888        else:
889            logging.getLogger("aiosendspin").setLevel(self.logger.level + 10)
890
891    def _begin_client_event(self, client_id: str) -> int:
892        """Increment version and in-flight task count for a client event."""
893        version = self._client_event_versions.get(client_id, 0) + 1
894        self._client_event_versions[client_id] = version
895        self._client_event_task_counts[client_id] = (
896            self._client_event_task_counts.get(client_id, 0) + 1
897        )
898        return version
899
900    def _finish_client_event(self, client_id: str) -> None:
901        """Drop in-flight bookkeeping and prune version state when idle."""
902        task_count = self._client_event_task_counts.get(client_id, 0)
903        if task_count <= 1:
904            self._client_event_task_counts.pop(client_id, None)
905            self._client_event_versions.pop(client_id, None)
906            return
907        self._client_event_task_counts[client_id] = task_count - 1
908
909    def _is_current_client_event(self, client_id: str, event_version: int) -> bool:
910        """Return True if the event version is still the latest for the client."""
911        return self._client_event_versions.get(client_id) == event_version
912
913    async def _apply_hass_esphome_enrichment(self, players: Sequence[SendspinBasePlayer]) -> None:
914        """
915        Apply Home Assistant-sourced enrichment to ESPHome-backed Sendspin players.
916
917        Applies the HA display name and resolves the HA media_player entity that
918        announcements are relayed to: ESPHome devices support announcements
919        natively, but that capability is only reachable through the HA API.
920        Players are correlated by MAC address (the Sendspin client id).
921        """
922        esphome_players = [
923            player for player in players if player.device_info.manufacturer == "ESPHome"
924        ]
925        if not esphome_players:
926            return
927        hass = cast("HomeAssistantProvider | None", self.mass.get_provider("hass"))
928        if hass is None or not hass.available:
929            for player in esphome_players:
930                if isinstance(player, SendspinPlayer):
931                    player.set_hass_announce_entity(None)
932            return
933        try:
934            device_infos = await hass.get_media_player_device_infos(
935                [player.player_id for player in esphome_players], platform="esphome"
936            )
937        except Exception as err:
938            self.logger.warning("Failed to apply Home Assistant enrichment: %s", err)
939            return
940        for player in esphome_players:
941            device_info = device_infos.get(player.player_id.lower())
942            if device_info is not None and device_info["name"]:
943                player._attr_name = device_info["name"]
944            if isinstance(player, SendspinPlayer):
945                player.set_hass_announce_entity(
946                    device_info["announce_entity_id"] if device_info is not None else None
947                )
948
949    async def _refresh_hass_esphome_enrichment(self) -> None:
950        """Re-apply the HA enrichment to all registered ESPHome players (in place)."""
951        players = [
952            player
953            for player in self.players
954            if isinstance(player, SendspinBasePlayer)
955            and player.device_info.manufacturer == "ESPHome"
956        ]
957        if not players:
958            return
959        await self._apply_hass_esphome_enrichment(players)
960        for player in players:
961            if player.initialized.is_set():
962                player.update_state()
963
964    def _create_player(
965        self,
966        client_id: str,
967        sendspin_client: SendspinClient,
968        existing_player: Player | None,
969        initial_hello: ClientHelloPayload | None = None,
970    ) -> SendspinBasePlayer:
971        """
972        Create the appropriate player class based on client roles.
973
974        Priority: player role -> SendspinPlayer, metadata role -> DISPLAY,
975        visualizer role -> VISUALIZER. Bridge-registered type overrides the default.
976        """
977        extra_ids = self._bridge_identifiers.pop(client_id, None)
978        bridge_player_type = self._bridge_player_types.pop(client_id, None)
979        underlying_player_id = self._bridge_underlying_players.pop(client_id, None)
980        if underlying_player_id is None and existing_player is not None:
981            underlying_player_id = existing_player.underlying_player_id
982        static_delay_default_ms = self._bridge_static_delay_defaults.pop(client_id, None)
983        if static_delay_default_ms is None and isinstance(existing_player, SendspinPlayer):
984            static_delay_default_ms = existing_player.static_delay_default_ms
985
986        # Select on negotiated (not active) roles: activation follows pairing/trust state,
987        # which must not change the player class.
988        negotiated_families = {
989            role_family(role_id) for role_id in sendspin_client.negotiated_role_ids
990        }
991        has_player_role = "player" in negotiated_families
992        has_metadata_role = "metadata" in negotiated_families
993        has_visualizer_role = "visualizer" in negotiated_families
994
995        if has_player_role:
996            audio_player = SendspinPlayer(self, client_id, initial_hello=initial_hello)
997            if isinstance(existing_player, SendspinPlayer):
998                audio_player.preserve_control_features_from(existing_player)
999            player: SendspinBasePlayer = audio_player
1000        elif has_metadata_role or has_visualizer_role:
1001            default_type = PlayerType.DISPLAY if has_metadata_role else PlayerType.VISUALIZER
1002            viz_player = SendspinVisualizerPlayer(self, client_id, initial_hello=initial_hello)
1003            viz_player._attr_type = bridge_player_type or default_type
1004            player = viz_player
1005        else:
1006            audio_player = SendspinPlayer(self, client_id, initial_hello=initial_hello)
1007            if isinstance(existing_player, SendspinPlayer):
1008                audio_player.preserve_control_features_from(existing_player)
1009            player = audio_player
1010
1011        if extra_ids:
1012            for id_type, id_value in extra_ids.items():
1013                player.device_info.add_identifier(id_type, id_value)
1014        if underlying_player_id is not None:
1015            player._attr_underlying_player_id = underlying_player_id
1016        if static_delay_default_ms is not None and isinstance(player, SendspinPlayer):
1017            player.static_delay_default_ms = static_delay_default_ms
1018        return player
1019
1020    @staticmethod
1021    def _pick_pin_method(
1022        offered: list[PairMethodDescriptor], *, verify: bool = False, static: bool = False
1023    ) -> PairMethod:
1024        """
1025        Select the preferred usable PIN method from the client's offer.
1026
1027        :param verify: Restrict to dynamic PIN, the only method that proves device presence.
1028        :param static: Restrict to static PIN, overriding the dynamic-first default.
1029        """
1030        wanted: tuple[PairMethod, ...]
1031        if verify:
1032            wanted = (PairMethod.DYNAMIC_PIN,)
1033        elif static:
1034            wanted = (PairMethod.STATIC_PIN,)
1035        else:
1036            wanted = (PairMethod.DYNAMIC_PIN, PairMethod.STATIC_PIN)
1037        pin_methods = [descriptor for descriptor in offered if descriptor.method in wanted]
1038        for method in wanted:
1039            for descriptor in pin_methods:
1040                if descriptor.method is method and not descriptor.locked_out:
1041                    return descriptor.method
1042        if pin_methods:
1043            raise SecurityActionError("pairing_error_locked_out")
1044        raise SecurityActionError("pairing_error_no_pin_method")
1045
1046    def _begin_pin_attempt(self, session: PinPairingSession) -> None:
1047        """Start or restart a pairing attempt for the session, resetting per-attempt state."""
1048        self._cancel_pin_idle_timeout(session.client_id)
1049        session.error = None
1050        session.retryable = False
1051        session.pin_request_event.clear()
1052        if session.pin_future.done():
1053            session.pin_future = self.mass.loop.create_future()
1054        session.task = self.mass.create_task(self._run_pin_pairing(session))
1055
1056    async def _pin_request_feedback(self, session: PinPairingSession) -> None:
1057        """Wait briefly for the attempt to reach the PIN wait (or end), for an accurate render."""
1058        if session.task is None:
1059            return
1060
1061        async def pin_requested() -> None:
1062            await session.pin_request_event.wait()
1063
1064        waiter = self.mass.create_task(pin_requested())
1065        try:
1066            await asyncio.wait(
1067                {waiter, session.task},
1068                timeout=PIN_REQUEST_FEEDBACK_TIMEOUT,
1069                return_when=asyncio.FIRST_COMPLETED,
1070            )
1071        finally:
1072            waiter.cancel()
1073
1074    async def _run_pin_pairing(self, session: PinPairingSession) -> None:
1075        """Run one PIN pairing attempt, classifying the outcome for the UI."""
1076
1077        def pin_provider() -> asyncio.Future[str]:
1078            # Invoked only once the client's pair-init has arrived (post-gesture).
1079            session.pin_request_event.set()
1080            return session.pin_future
1081
1082        try:
1083            await self.server_api.initiate_pairing(
1084                session.client_id,
1085                PairingAttempt(session.method, pin_provider=pin_provider, verify=session.verify),
1086            )
1087        except PairingAbortError as err:
1088            if (
1089                isinstance(err, LocalPairingAbortError)
1090                and err.reason is PairAbortReason.USER_CANCELLED
1091            ):
1092                # Our own end_pairing cancelled this attempt; the connection is already restored.
1093                return
1094            session.error = err
1095            self.logger.debug("PIN pairing with %s aborted: %s", session.client_id, err)
1096            if err.reason is PairAbortReason.LOCKED_OUT:
1097                # Retrying is pointless once locked out; unpark the connection.
1098                await self._end_pairing_quietly(session.client_id)
1099            else:
1100                session.retryable = True
1101                self._arm_pin_idle_timeout(session)
1102        except Exception as err:
1103            # A non-abort failure: the server has already disconnected the client.
1104            session.error = err
1105            self.logger.debug("PIN pairing with %s failed: %s", session.client_id, err)
1106        else:
1107            await self._refresh_player(session.client_id)
1108        finally:
1109            if not session.pin_future.done():
1110                session.pin_future.cancel()
1111
1112    def _arm_pin_idle_timeout(self, session: PinPairingSession) -> None:
1113        """Schedule restoration of the connection if a failed attempt is left unretried."""
1114        self.mass.call_later(
1115            PIN_RETRY_IDLE_TIMEOUT,
1116            self._pin_idle_timeout,
1117            session,
1118            task_id=_pin_idle_task_id(session.client_id),
1119        )
1120
1121    def _cancel_pin_idle_timeout(self, client_id: str) -> None:
1122        """Cancel a pairing idle timeout, whether still pending or already firing."""
1123        task_id = _pin_idle_task_id(client_id)
1124        self.mass.cancel_timer(task_id)
1125        self.mass.cancel_task(task_id)
1126
1127    async def _pin_idle_timeout(self, session: PinPairingSession) -> None:
1128        """Terminate an abandoned retryable session, restoring the connection to service."""
1129        session.retryable = False
1130        session.error = TimeoutError("timed out waiting for a pairing retry")
1131        await self._end_pairing_quietly(session.client_id)
1132        await self._refresh_player(session.client_id)
1133
1134    async def _end_pairing_quietly(self, client_id: str) -> None:
1135        """End pairing on a client, tolerating an already-gone connection."""
1136        try:
1137            await self.server_api.end_pairing(client_id)
1138        except Exception as err:
1139            self.logger.debug("Ending pairing for %s failed: %s", client_id, err)
1140
1141    def _management_session_or_raise(self, client_id: str) -> ManagementSession:
1142        """Return the client's management session or raise if none is open."""
1143        session = self.get_management_session(client_id)
1144        if session is None:
1145            raise SecurityActionError("management_error_no_session")
1146        return session
1147
1148    async def _management_call[T](self, connection: SendspinConnection, request: Awaitable[T]) -> T:
1149        """Run a management request with a timeout, mapping transport failures to SecurityActionError."""
1150        try:
1151            async with asyncio.timeout(MANAGEMENT_REQUEST_TIMEOUT):
1152                return await request
1153        except TimeoutError as err:
1154            # Replies match requests by order with no id, so a timed-out request left in
1155            # flight would desync the next one; drop the connection to reset the channel.
1156            await connection.disconnect()
1157            raise SecurityActionError("management_error_timeout") from err
1158        except RuntimeError as err:
1159            raise SecurityActionError("management_error_generic", detail=str(err)) from err
1160
1161    def _arm_management_idle_timeout(self, session: ManagementSession) -> None:
1162        """(Re)start the idle countdown that closes an abandoned management session."""
1163        self.mass.call_later(
1164            MANAGEMENT_IDLE_TIMEOUT,
1165            self._drop_management_session,
1166            session,
1167            task_id=_management_idle_task_id(session.client_id),
1168        )
1169
1170    def _drop_management_session(self, session: ManagementSession) -> None:
1171        """Remove a management session, releasing its hold on the connection."""
1172        self.mass.cancel_timer(_management_idle_task_id(session.client_id))
1173        self._management_sessions.pop(session.client_id, None)
1174        with suppress(Exception):
1175            session.connection.disable_management()
1176
1177    async def _refresh_player(self, client_id: str) -> None:
1178        """Re-evaluate a registered player after a pairing/trust change (in place)."""
1179        player = self.mass.players.get_player(client_id)
1180        if not isinstance(player, SendspinBasePlayer) or not player.initialized.is_set():
1181            return
1182        # A trust change (de)activates roles; role instances are recreated on activation,
1183        # so pushed config (preferred format, static delay) must be re-applied.
1184        await player.on_config_updated()
1185        player.update_state()
1186
1187    async def _handle_client_added(self, client_id: str, event_version: int) -> None:
1188        """Handle a new client connection asynchronously."""
1189        try:
1190            if self._unloading or client_id in self._headless_client_ids:
1191                return
1192            sendspin_client = self.server_api.get_client(client_id)
1193            if sendspin_client is None:
1194                self.logger.debug("Client %s disconnected before add handling started", client_id)
1195                return
1196            bridge_hello_snapshot = None
1197            if (
1198                client_id in self._bridge_identifiers
1199                and (bridge_hello := sendspin_client.info_or_none) is not None
1200            ):
1201                # Snapshot the bridges hello before a reconnect can overwrite it
1202                bridge_hello_snapshot = deepcopy(bridge_hello)
1203            if pending_event := self._pending_unregisters.get(client_id):
1204                self.logger.debug(
1205                    "Waiting for pending unregister of %s before registering", client_id
1206                )
1207                await pending_event.wait()
1208                if not self._is_current_client_event(client_id, event_version):
1209                    self.logger.debug("Skipping stale add event for %s after waiting", client_id)
1210                    return
1211            # Check if client still exists (may have disconnected while waiting)
1212            sendspin_client = self.server_api.get_client(client_id)
1213            if sendspin_client is None:
1214                self.logger.debug("Client %s disconnected before hello completed", client_id)
1215                return
1216            # Wait for client hello to be processed (info becomes available)
1217            # ClientAddedEvent fires before the hello handshake completes
1218            for _ in range(50):  # Wait up to 5 seconds
1219                if sendspin_client.info_or_none is not None:
1220                    break
1221                await asyncio.sleep(0.1)
1222            else:
1223                self.logger.warning("Client %s hello not received within timeout", client_id)
1224                return
1225            if not self._is_current_client_event(client_id, event_version):
1226                self.logger.debug("Skipping stale add event for %s", client_id)
1227                return
1228            if not self.mass.config.get_raw_player_config_value(client_id, CONF_ENABLED, True):
1229                self.logger.debug("Ignoring disabled sendspin client: %s", client_id)
1230                return
1231            existing_player = self.mass.players.get_player(client_id)
1232            preserved_identifiers = (
1233                dict(existing_player.device_info.identifiers) if existing_player is not None else {}
1234            )
1235            if existing_player is not None:
1236                self.logger.debug("Refreshing existing player object for %s", client_id)
1237                await self.mass.players.unregister(client_id)
1238                if not self._is_current_client_event(client_id, event_version):
1239                    self.logger.debug("Skipping stale add event for %s after unregister", client_id)
1240                    return
1241                sendspin_client = self.server_api.get_client(client_id)
1242                if sendspin_client is None:
1243                    self.logger.debug("Client %s disconnected after unregister", client_id)
1244                    return
1245
1246            player = self._create_player(
1247                client_id, sendspin_client, existing_player, bridge_hello_snapshot
1248            )
1249            for id_type, id_value in preserved_identifiers.items():
1250                player.device_info.add_identifier(id_type, id_value)
1251            self.logger.debug("Client %s connected", client_id)
1252            await self._apply_hass_esphome_enrichment([player])
1253            if not self._is_current_client_event(client_id, event_version):
1254                self.logger.debug("Skipping stale add event for %s after HA enrichment", client_id)
1255                player._unsubscribe_client_callbacks()
1256                return
1257            try:
1258                await self.mass.players.register(player)
1259            except AlreadyRegisteredError:
1260                self.logger.debug(
1261                    "Client %s already registered while handling add event", client_id
1262                )
1263                player._unsubscribe_client_callbacks()
1264        finally:
1265            self._finish_client_event(client_id)
1266
1267    async def _handle_client_removed(self, client_id: str, event_version: int) -> None:
1268        """Handle a client disconnection asynchronously."""
1269        try:
1270            if self._unloading:
1271                return
1272            if client_id in self._headless_client_ids:
1273                self._headless_client_ids.discard(client_id)
1274                return
1275            self.logger.debug("Client %s disconnected", client_id)
1276            if not self._is_current_client_event(client_id, event_version):
1277                self.logger.debug("Skipping stale remove event for %s", client_id)
1278                return
1279            unregister_event = asyncio.Event()
1280            self._pending_unregisters[client_id] = unregister_event
1281            try:
1282                await self.mass.players.unregister(client_id)
1283            finally:
1284                self._pending_unregisters.pop(client_id, None)
1285                unregister_event.set()
1286        finally:
1287            self._finish_client_event(client_id)
1288
1289    async def _handle_client_updated(self, client_id: str, event_version: int) -> None:
1290        """Handle a client whose hello payload changed on reconnect."""
1291        try:
1292            if self._unloading:
1293                return
1294            if pending_event := self._pending_unregisters.get(client_id):
1295                self.logger.debug("Waiting for pending unregister of %s before updating", client_id)
1296                await pending_event.wait()
1297                if not self._is_current_client_event(client_id, event_version):
1298                    self.logger.debug("Skipping stale update event for %s after waiting", client_id)
1299                    return
1300            sendspin_client = self.server_api.get_client(client_id)
1301            if sendspin_client is None:
1302                return
1303            if not self._is_current_client_event(client_id, event_version):
1304                self.logger.debug("Skipping stale update event for %s", client_id)
1305                return
1306            existing_player = self.mass.players.get_player(client_id)
1307            if not isinstance(existing_player, SendspinBasePlayer):
1308                return
1309            previous_device_info = existing_player.device_info
1310            previous_type = existing_player.type
1311            existing_player._refresh_client_info(sendspin_client)
1312            if isinstance(existing_player, SendspinPlayer):
1313                existing_player.restore_bridge_identity(previous_device_info, previous_type)
1314            await self._apply_hass_esphome_enrichment([existing_player])
1315            if not self._is_current_client_event(client_id, event_version):
1316                self.logger.debug("Skipping stale update event for %s after refresh", client_id)
1317                return
1318            if previous_type == PlayerType.PROTOCOL and existing_player.type != PlayerType.PROTOCOL:
1319                existing_player.set_protocol_parent_id(None)
1320                existing_player._attr_underlying_player_id = None
1321            await self.mass.players.register_or_update(existing_player)
1322        finally:
1323            self._finish_client_event(client_id)
1324
1325    def _get_virtual_player_config_owner(self, player_id: str) -> str | None:
1326        """Return the owner instance id from a stored virtual player config, if any."""
1327        raw_conf = self.mass.config.get(f"{CONF_PLAYERS}/{player_id}")
1328        if not isinstance(raw_conf, dict) or raw_conf.get("provider") != self.instance_id:
1329            return None
1330        values = raw_conf.get("values")
1331        if not isinstance(values, dict):
1332            return None
1333        return cast("str | None", values.get(CONF_VIRTUAL_PLAYER_OWNER))
1334
1335    def _register_virtual_player_client(self, player_id: str, display_name: str) -> None:
1336        """Register the silent external Sendspin client backing a virtual player."""
1337        hello = ClientHelloPayload(
1338            client_id=player_id,
1339            name=display_name,
1340            version=1,
1341            supported_roles=[BRIDGE_ROLE_ID],
1342            device_info=SendspinDeviceInfo(
1343                product_name="Virtual Player",
1344                manufacturer="Music Assistant",
1345            ),
1346            player_support=ClientHelloPlayerSupport(
1347                supported_formats=[
1348                    SupportedAudioFormat(
1349                        codec=AudioCodec.PCM,
1350                        channels=BRIDGE_CHANNELS,
1351                        sample_rate=BRIDGE_SAMPLE_RATE,
1352                        bit_depth=BRIDGE_BIT_DEPTH,
1353                    )
1354                ],
1355                buffer_capacity=1_000,
1356                supported_commands=[],
1357            ),
1358        )
1359        sendspin_client = self.server_api.register_external_player(
1360            hello, on_stream_start=self._on_virtual_player_stream_start
1361        )
1362        for role in sendspin_client.roles_by_family("player"):
1363            if not isinstance(role, BridgePlayerRole):
1364                continue
1365            # audio delivered to the role is simply discarded: the virtual player
1366            # only anchors the group, the members receive the actual stream
1367            role.set_callbacks(
1368                on_audio_chunk=_virtual_player_noop,
1369                on_volume_change=_virtual_player_noop,
1370                on_mute_change=_virtual_player_noop,
1371                on_stream_start=_virtual_player_noop,
1372                on_stream_end=_virtual_player_noop,
1373            )
1374            role.setup_audio_requirements()
1375            role.set_timing(required_lead_time_ms=0, min_buffer_ms=0)
1376            break
1377
1378    async def _wait_for_virtual_player(self, player_id: str) -> None:
1379        """Wait until the virtual player is registered in MA with its queue."""
1380        deadline = self.mass.loop.time() + VIRTUAL_PLAYER_REGISTER_TIMEOUT
1381        while self.mass.loop.time() < deadline:
1382            if (
1383                self.mass.players.get_player(player_id) is not None
1384                and self.mass.player_queues.get(player_id) is not None
1385            ):
1386                return
1387            await asyncio.sleep(0.1)
1388        raise SetupFailedError(f"Virtual player {player_id} was not registered in time")
1389
1390    async def _cleanup_failed_virtual_player_creation(self, player_id: str) -> None:
1391        """
1392        Remove a virtual player after its creation does not complete.
1393
1394        :param player_id: Virtual player to remove.
1395        """
1396        last_error: Exception | None = None
1397        for delay in VIRTUAL_PLAYER_CLEANUP_DELAYS:
1398            if delay:
1399                await asyncio.sleep(delay)
1400            try:
1401                async with asyncio.timeout(VIRTUAL_PLAYER_CLEANUP_TIMEOUT):
1402                    await self.remove_virtual_player(player_id)
1403                return
1404            except Exception as err:
1405                last_error = err
1406        self.logger.warning(
1407            "Could not clean up failed virtual player creation %s: %s",
1408            player_id,
1409            last_error,
1410        )
1411
1412    def _on_virtual_player_stream_start(self, _request: ExternalStreamStartRequest) -> None:
1413        """Accept stream start requests for virtual players (nothing to connect)."""
1414
1415    async def _on_providers_updated(self, event: MassEvent) -> None:
1416        """Handle a change in the loaded providers."""
1417        # during (server) shutdown all providers unload; the startup sweep
1418        # takes care of configs whose owner is really gone
1419        if self._unloading or self.mass.closing:
1420            return
1421        # remove virtual players whose owning provider is no longer loaded
1422        for player_id, owner_instance_id in list(self._virtual_players.items()):
1423            if self.mass.get_provider(owner_instance_id) is not None:
1424                continue
1425            self.logger.debug(
1426                "Removing virtual player %s: owner %s unloaded", player_id, owner_instance_id
1427            )
1428            await self.remove_virtual_player(player_id)
1429        # (re)apply the HA-backed enrichment when the hass plugin (un)loads
1430        hass = self.mass.get_provider("hass")
1431        hass_available = hass is not None and hass.available
1432        if hass_available != self._hass_available:
1433            self._hass_available = hass_available
1434            await self._refresh_hass_esphome_enrichment()
1435
1436    def _remove_orphan_virtual_player_configs(self) -> None:
1437        """Delete stored configs of virtual players whose owner provider is gone."""
1438        all_player_configs = self.mass.config.get(CONF_PLAYERS, {})
1439        for player_id, raw_conf in list(all_player_configs.items()):
1440            if not isinstance(raw_conf, dict) or raw_conf.get("provider") != self.instance_id:
1441                continue
1442            values = raw_conf.get("values")
1443            if not isinstance(values, dict):
1444                values = {}
1445            owner_instance_id = values.get(CONF_VIRTUAL_PLAYER_OWNER)
1446            if owner_instance_id is None:
1447                continue
1448            if self.mass.config.get(f"{CONF_PROVIDERS}/{owner_instance_id}") is not None:
1449                # owner still configured; it will recreate its virtual players
1450                continue
1451            self.logger.debug("Removing orphan virtual player config %s", player_id)
1452            self.mass.players.delete_player_config(player_id)
1453
1454
1455def _virtual_player_noop(*_args: object) -> None:
1456    """No-op callback for virtual players, which never render audio locally."""
1457