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