music-assistant-server

1"""Bose SoundTouch player provider implementation."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, cast
6
7import aiohttp
8from zeroconf import ServiceStateChange
9
10from music_assistant.constants import CONF_ENTRY_MANUAL_DISCOVERY_IPS
11from music_assistant.helpers.util import get_primary_ip_address_from_zeroconf
12from music_assistant.models.player_provider import PlayerProvider
13
14from .client import SoundTouchClient
15from .config import (
16    ACTION_ASSIGN,
17    ACTION_SEARCH,
18    CONF_SEARCH_MEDIA_TYPE,
19    CONF_SEARCH_QUERY,
20    CONF_SEARCH_RESULT,
21    CONF_SEARCH_TARGET,
22    PRESET_KEY_PREFIX,
23    build_preset_config_entries,
24    preset_media_key,
25)
26from .const import PLAYER_ID_PREFIX, PRESET_IDS
27from .player import BoseSoundTouchPlayer
28
29if TYPE_CHECKING:
30    from music_assistant_models.config_entries import (
31        ConfigActionResult,
32        ConfigEntry,
33        ProviderConfig,
34    )
35    from zeroconf.asyncio import AsyncServiceInfo
36
37
38def _search_values_to_reset(changed_keys: set[str]) -> tuple[str, ...]:
39    """Return search fields invalidated by an earlier step changing."""
40    if f"values/{CONF_SEARCH_MEDIA_TYPE}" in changed_keys:
41        return (CONF_SEARCH_QUERY, CONF_SEARCH_RESULT, CONF_SEARCH_TARGET)
42    if f"values/{CONF_SEARCH_QUERY}" in changed_keys:
43        return (CONF_SEARCH_RESULT, CONF_SEARCH_TARGET)
44    if f"values/{CONF_SEARCH_RESULT}" in changed_keys:
45        return (CONF_SEARCH_TARGET,)
46    return ()
47
48
49class BoseSoundTouchProvider(PlayerProvider):
50    """Player provider for Bose SoundTouch speakers."""
51
52    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
53        """Return Config entries to configure this provider."""
54        return (CONF_ENTRY_MANUAL_DISCOVERY_IPS, *await build_preset_config_entries(self))
55
56    async def handle_config_action(
57        self, action: str
58    ) -> tuple[ConfigEntry, ...] | ConfigActionResult | None:
59        """
60        Handle a preset search/assignment button press and re-render the entries.
61
62        The search button refreshes the shared result list. The assignment button copies
63        the selected result to the chosen physical preset.
64
65        :param action: The action id of the pressed button.
66        """
67        if action not in (ACTION_SEARCH, ACTION_ASSIGN):
68            return await super().handle_config_action(action)
69        if action == ACTION_SEARCH:
70            self._reset_search_values(CONF_SEARCH_RESULT, CONF_SEARCH_TARGET)
71        if action == ACTION_ASSIGN:
72            selected = str(self.get_config_value(CONF_SEARCH_RESULT, "") or "")
73            target = str(self.get_config_value(CONF_SEARCH_TARGET, "") or "")
74            if selected and target.isdigit() and (preset_id := int(target)) in PRESET_IDS:
75                self._update_config_value(preset_media_key(preset_id), selected, immediate=True)
76                self._reset_search_values(
77                    CONF_SEARCH_MEDIA_TYPE,
78                    CONF_SEARCH_QUERY,
79                    CONF_SEARCH_RESULT,
80                    CONF_SEARCH_TARGET,
81                )
82        return (
83            CONF_ENTRY_MANUAL_DISCOVERY_IPS,
84            *await build_preset_config_entries(self, refresh_results=action == ACTION_SEARCH),
85        )
86
87    async def update_config(self, config: ProviderConfig, changed_keys: set[str]) -> None:
88        """Handle logic when the config is updated."""
89        for key in _search_values_to_reset(changed_keys):
90            self._update_config_value(key, "", immediate=True)
91            if entry := config.values.get(key):
92                entry.value = ""
93        # the preset mappings are read on demand when a button is pressed, so hide those
94        # keys from the base implementation: reloading the provider for a preset edit
95        # would needlessly drop and rediscover every speaker
96        await super().update_config(
97            config,
98            {key for key in changed_keys if not key.startswith(f"values/{PRESET_KEY_PREFIX}")},
99        )
100
101    def get_preset_media(self, preset_id: int) -> str:
102        """
103        Return the media URI mapped to the given physical preset button (empty if unset).
104
105        :param preset_id: The physical preset button number (1-6).
106        """
107        return str(self.get_config_value(preset_media_key(preset_id), "") or "")
108
109    async def loaded_in_mass(self) -> None:
110        """Call after the provider has been loaded."""
111        manual_ips = cast("list[str]", self.config.get_value(CONF_ENTRY_MANUAL_DISCOVERY_IPS.key))
112        for ip_address in manual_ips:
113            if stripped := ip_address.strip():
114                await self.try_add_player(stripped)
115
116    async def on_mdns_service_state_change(
117        self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
118    ) -> None:
119        """Handle MDNS service state callback."""
120        if not info or state_change == ServiceStateChange.Removed:
121            # availability is tracked by the player itself (websocket + polling)
122            return
123        ip_address = get_primary_ip_address_from_zeroconf(info)
124        if not ip_address:
125            return
126        # if we already know a player on this address, just trigger an update
127        if existing := self._get_player_by_ip(ip_address):
128            self.mass.players.trigger_player_update(existing.player_id)
129            return
130        # debounce setup to avoid duplicate work on rapid mDNS updates
131        task_id = f"setup_soundtouch_{ip_address}"
132        self.mass.call_later(2, self.try_add_player, ip_address, task_id=task_id)
133
134    async def try_add_player(self, ip_address: str) -> None:
135        """Try to add a Bose SoundTouch speaker as a player."""
136        client = SoundTouchClient(self.mass.http_session, ip_address)
137        try:
138            info = await client.get_info()
139        except (aiohttp.ClientError, TimeoutError, OSError) as err:
140            self.logger.debug("Failed to query SoundTouch device at %s: %s", ip_address, err)
141            return
142        if not info.device_id:
143            self.logger.debug("SoundTouch device at %s returned no device id", ip_address)
144            return
145
146        player_id = f"{PLAYER_ID_PREFIX}{info.device_id}"
147        if existing := self.mass.players.get_player(player_id):
148            # already known: refresh its address and bail out
149            assert isinstance(existing, BoseSoundTouchPlayer)
150            existing.update_ip_address(ip_address)
151            return
152
153        player = BoseSoundTouchPlayer(self, player_id, client, info)
154        try:
155            await player.setup()
156            await self.mass.players.register_or_update(player)
157        except Exception:
158            self.logger.exception("Failed to register SoundTouch player %s", info.name)
159            await player.on_unload()
160            return
161        self.logger.info("Registered Bose SoundTouch player: %s (%s)", info.name, ip_address)
162
163    def _reset_search_values(self, *keys: str) -> None:
164        """Reset persisted fields that belong to later search steps."""
165        for key in keys:
166            self._update_config_value(key, "", immediate=True)
167
168    def _get_player_by_ip(self, ip_address: str) -> BoseSoundTouchPlayer | None:
169        """Return an existing SoundTouch player with the given IP address (if any)."""
170        for player in self.players:
171            if (
172                isinstance(player, BoseSoundTouchPlayer)
173                and player.device_info.ip_address == ip_address
174            ):
175                return player
176        return None
177