/
/
/
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 PRESET_KEY_PREFIX,
17 build_preset_config_entries,
18 parse_preset_action,
19 preset_media_key,
20 preset_selected_media_key,
21)
22from .const import PLAYER_ID_PREFIX
23from .player import BoseSoundTouchPlayer
24
25if TYPE_CHECKING:
26 from music_assistant_models.config_entries import (
27 ConfigActionResult,
28 ConfigEntry,
29 ProviderConfig,
30 )
31 from zeroconf.asyncio import AsyncServiceInfo
32
33
34class BoseSoundTouchProvider(PlayerProvider):
35 """Player provider for Bose SoundTouch speakers."""
36
37 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
38 """Return Config entries to configure this provider."""
39 return (CONF_ENTRY_MANUAL_DISCOVERY_IPS, *await build_preset_config_entries(self))
40
41 async def handle_config_action(
42 self, action: str
43 ) -> tuple[ConfigEntry, ...] | ConfigActionResult | None:
44 """
45 Handle a preset search/select button press and re-render the entries.
46
47 A select button persists the currently selected search result as that preset's
48 media URI; a search button only re-runs that preset's media search. Values are
49 read from the (already persisted) stored config; no in-flight form is passed.
50
51 :param action: The action id of the pressed button.
52 """
53 preset_id, is_select = parse_preset_action(action)
54 if preset_id is None:
55 return await super().handle_config_action(action)
56 if is_select and (
57 selected := str(self.get_config_value(preset_selected_media_key(preset_id), "") or "")
58 ):
59 self.mass.config.set_raw_provider_config_value(
60 self.instance_id, preset_media_key(preset_id), selected
61 )
62 # only the search button runs a (slow) media search; selecting merely persists the
63 # already fetched choice, which stays selectable without searching again
64 return (
65 CONF_ENTRY_MANUAL_DISCOVERY_IPS,
66 *await build_preset_config_entries(
67 self, refresh_preset_id=None if is_select else preset_id
68 ),
69 )
70
71 async def update_config(self, config: ProviderConfig, changed_keys: set[str]) -> None:
72 """Handle logic when the config is updated."""
73 # the preset mappings are read on demand when a button is pressed, so hide those
74 # keys from the base implementation: reloading the provider for a preset edit
75 # would needlessly drop and rediscover every speaker
76 await super().update_config(
77 config,
78 {key for key in changed_keys if not key.startswith(f"values/{PRESET_KEY_PREFIX}")},
79 )
80
81 def get_preset_media(self, preset_id: int) -> str:
82 """
83 Return the media URI mapped to the given physical preset button (empty if unset).
84
85 :param preset_id: The physical preset button number (1-6).
86 """
87 return str(self.get_config_value(preset_media_key(preset_id), "") or "")
88
89 async def loaded_in_mass(self) -> None:
90 """Call after the provider has been loaded."""
91 manual_ips = cast("list[str]", self.config.get_value(CONF_ENTRY_MANUAL_DISCOVERY_IPS.key))
92 for ip_address in manual_ips:
93 if stripped := ip_address.strip():
94 await self.try_add_player(stripped)
95
96 async def on_mdns_service_state_change(
97 self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
98 ) -> None:
99 """Handle MDNS service state callback."""
100 if not info or state_change == ServiceStateChange.Removed:
101 # availability is tracked by the player itself (websocket + polling)
102 return
103 ip_address = get_primary_ip_address_from_zeroconf(info)
104 if not ip_address:
105 return
106 # if we already know a player on this address, just trigger an update
107 if existing := self._get_player_by_ip(ip_address):
108 self.mass.players.trigger_player_update(existing.player_id)
109 return
110 # debounce setup to avoid duplicate work on rapid mDNS updates
111 task_id = f"setup_soundtouch_{ip_address}"
112 self.mass.call_later(2, self.try_add_player, ip_address, task_id=task_id)
113
114 async def try_add_player(self, ip_address: str) -> None:
115 """Try to add a Bose SoundTouch speaker as a player."""
116 client = SoundTouchClient(self.mass.http_session, ip_address)
117 try:
118 info = await client.get_info()
119 except (aiohttp.ClientError, TimeoutError, OSError) as err:
120 self.logger.debug("Failed to query SoundTouch device at %s: %s", ip_address, err)
121 return
122 if not info.device_id:
123 self.logger.debug("SoundTouch device at %s returned no device id", ip_address)
124 return
125
126 player_id = f"{PLAYER_ID_PREFIX}{info.device_id}"
127 if existing := self.mass.players.get_player(player_id):
128 # already known: refresh its address and bail out
129 assert isinstance(existing, BoseSoundTouchPlayer)
130 existing.update_ip_address(ip_address)
131 return
132
133 player = BoseSoundTouchPlayer(self, player_id, client, info)
134 try:
135 await player.setup()
136 await self.mass.players.register_or_update(player)
137 except Exception:
138 self.logger.exception("Failed to register SoundTouch player %s", info.name)
139 await player.on_unload()
140 return
141 self.logger.info("Registered Bose SoundTouch player: %s (%s)", info.name, ip_address)
142
143 def _get_player_by_ip(self, ip_address: str) -> BoseSoundTouchPlayer | None:
144 """Return an existing SoundTouch player with the given IP address (if any)."""
145 for player in self.players:
146 if (
147 isinstance(player, BoseSoundTouchPlayer)
148 and player.device_info.ip_address == ip_address
149 ):
150 return player
151 return None
152