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