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