music-assistant-server

1"""Bose SoundTouch player implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import time
8from typing import TYPE_CHECKING, cast
9
10import aiohttp
11from music_assistant_models.enums import (
12    IdentifierType,
13    MediaType,
14    PlaybackState,
15    PlayerFeature,
16    PlayerType,
17)
18from music_assistant_models.errors import MusicAssistantError
19from music_assistant_models.player import DeviceInfo, PlayerSource
20
21from music_assistant.models.player import Player, PlayerMedia
22
23from .client import (
24    SoundTouchClient,
25    SoundTouchInfo,
26    SoundTouchNowPlaying,
27    extract_preset_id,
28    play_status_is_paused,
29    play_status_is_playing,
30)
31from .const import (
32    CONF_APP_KEY,
33    KEY_MUTE,
34    KEY_NEXT_TRACK,
35    KEY_PAUSE,
36    KEY_PLAY,
37    KEY_POWER,
38    KEY_PREV_TRACK,
39    NOTIFICATION_PORT,
40    PLAYER_ID_PREFIX,
41    PRESET_IDS,
42    RECONNECT_DELAY,
43    SOURCE_STANDBY,
44    WS_HEARTBEAT,
45    WS_SUBPROTOCOLS,
46)
47
48if TYPE_CHECKING:
49    from .provider import BoseSoundTouchProvider
50
51IDLE_POLL_INTERVAL = 30
52PLAYBACK_POLL_INTERVAL = 10
53
54
55class BoseSoundTouchPlayer(Player):
56    """Bose SoundTouch player in Music Assistant."""
57
58    def __init__(
59        self,
60        provider: BoseSoundTouchProvider,
61        player_id: str,
62        client: SoundTouchClient,
63        info: SoundTouchInfo,
64    ) -> None:
65        """Initialize the Player."""
66        super().__init__(provider, player_id)
67        self._client = client
68        self._device_id = info.device_id
69        self._attr_name = info.name
70        self._attr_type = PlayerType.PLAYER
71        # Playback (PLAY_MEDIA) is deliberately omitted: SoundTouch has no usable API to
72        # play an arbitrary stream, so audio is routed through a linked playback protocol
73        # (such as the DLNA renderer on the same device) via protocol linking.
74        self._attr_supported_features = {
75            PlayerFeature.POWER,
76            PlayerFeature.VOLUME_SET,
77            PlayerFeature.VOLUME_MUTE,
78            PlayerFeature.PAUSE,
79            PlayerFeature.NEXT_PREVIOUS,
80            PlayerFeature.SELECT_SOURCE,
81            PlayerFeature.SET_MEMBERS,
82        }
83        # Native announcements require a Bose developer app key; when configured the
84        # speaker plays them as an overlay (ducking and resuming the current playback).
85        app_key = provider.get_setup_value(CONF_APP_KEY)
86        self._app_key = str(app_key) if app_key else None
87        if self._app_key:
88            self._attr_supported_features.add(PlayerFeature.PLAY_ANNOUNCEMENT)
89        self._attr_can_group_with = {provider.instance_id}
90        self._attr_device_info = DeviceInfo(
91            model=info.model or "Bose SoundTouch",
92            manufacturer="Bose",
93            software_version=info.software_version,
94        )
95        self._attr_device_info.add_identifier(IdentifierType.UUID, info.device_id)
96        if info.mac_address:
97            self._attr_device_info.add_identifier(IdentifierType.MAC_ADDRESS, info.mac_address)
98        if info.ip_address:
99            self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, info.ip_address)
100        self._stop_event = asyncio.Event()
101        self._listener_task: asyncio.Task[None] | None = None
102
103    # --- Lifecycle ---
104
105    async def setup(self) -> None:
106        """Fetch initial state and start listening for device updates."""
107        await self._refresh_volume()
108        await self._refresh_now_playing()
109        await self._refresh_sources()
110        self._attr_needs_poll = True
111        self._attr_poll_interval = IDLE_POLL_INTERVAL
112        self._attr_available = True
113        self._listener_task = self.mass.create_task(self._listen())
114
115    async def poll(self) -> None:
116        """Poll the speaker as a safety net for missed websocket events."""
117        try:
118            await self._refresh_now_playing()
119            await self._refresh_volume()
120        except (aiohttp.ClientError, TimeoutError, OSError) as err:
121            self.logger.debug("Poll failed for %s: %s", self.name, err)
122            if self._attr_available:
123                self._attr_available = False
124                self.update_state()
125            return
126        if not self._attr_available:
127            self._attr_available = True
128        self._attr_poll_interval = (
129            PLAYBACK_POLL_INTERVAL
130            if self._attr_playback_state == PlaybackState.PLAYING
131            else IDLE_POLL_INTERVAL
132        )
133        self.update_state()
134
135    async def on_unload(self) -> None:
136        """Handle logic when the player is unloaded from the Player controller."""
137        self._stop_event.set()
138        if self._listener_task:
139            self._listener_task.cancel()
140            with contextlib.suppress(asyncio.CancelledError):
141                await self._listener_task
142            self._listener_task = None
143        await super().on_unload()
144
145    # --- Player commands ---
146
147    async def power(self, powered: bool) -> None:
148        """Handle POWER command on the player."""
149        now_playing = await self._client.get_now_playing()
150        currently_on = now_playing.source != SOURCE_STANDBY
151        if powered != currently_on:
152            # the POWER key toggles standby, so only send it when a change is needed
153            await self._client.press_key(KEY_POWER)
154        self._attr_powered = powered
155        self.update_state()
156
157    async def volume_set(self, volume_level: int) -> None:
158        """Handle VOLUME_SET command on the player."""
159        await self._client.set_volume(volume_level)
160        self._attr_volume_level = volume_level
161        self.update_state()
162
163    async def volume_mute(self, muted: bool) -> None:
164        """Handle VOLUME_MUTE command on the player."""
165        volume = await self._client.get_volume()
166        if volume.muted != muted:
167            # the MUTE key toggles mute, so only send it when a change is needed
168            await self._client.press_key(KEY_MUTE)
169        self._attr_volume_muted = muted
170        self.update_state()
171
172    async def play(self) -> None:
173        """Handle PLAY (resume) command on a native source."""
174        await self._client.press_key(KEY_PLAY)
175        self._attr_playback_state = PlaybackState.PLAYING
176        self.update_state()
177
178    async def pause(self) -> None:
179        """Handle PAUSE command on a native source."""
180        await self._client.press_key(KEY_PAUSE)
181        self._attr_playback_state = PlaybackState.PAUSED
182        self.update_state()
183
184    async def next_track(self) -> None:
185        """Handle NEXT_TRACK command on a native source."""
186        await self._client.press_key(KEY_NEXT_TRACK)
187
188    async def previous_track(self) -> None:
189        """Handle PREVIOUS_TRACK command on a native source."""
190        await self._client.press_key(KEY_PREV_TRACK)
191
192    async def select_source(self, source: str) -> None:
193        """Handle SELECT_SOURCE command on the player."""
194        source_name, _, source_account = source.partition(":")
195        await self._client.select_source(source_name, source_account or None)
196
197    async def play_announcement(
198        self, announcement: PlayerMedia, volume_level: int | None = None
199    ) -> None:
200        """Handle (native) playback of an announcement on the player."""
201        if not self._app_key:
202            return
203        self.logger.debug("Playing announcement %s on %s", announcement.uri, self.name)
204        await self._client.play_notification(self._app_key, announcement.uri, volume_level)
205
206    async def set_members(
207        self,
208        player_ids_to_add: list[str] | None = None,
209        player_ids_to_remove: list[str] | None = None,
210    ) -> None:
211        """Handle SET_MEMBERS command using the SoundTouch multiroom zone API."""
212        if player_ids_to_add:
213            members = self._member_tuples(player_ids_to_add)
214            if members:
215                zone = await self._client.get_zone()
216                if zone.master_id == self._device_id:
217                    await self._client.add_zone_slaves(self._device_id, members)
218                else:
219                    own = (self._device_id, self.device_info.ip_address or "")
220                    await self._client.set_zone(self._device_id, [own, *members])
221        if player_ids_to_remove:
222            members = self._member_tuples(player_ids_to_remove)
223            if members:
224                await self._client.remove_zone_slaves(self._device_id, members)
225        await self._refresh_zone()
226        for member_id in (player_ids_to_add or []) + (player_ids_to_remove or []):
227            self.mass.players.trigger_player_update(member_id)
228
229    # --- Public helpers ---
230
231    @property
232    def device_id(self) -> str:
233        """Return the Bose SoundTouch device id of this player."""
234        return self._device_id
235
236    def update_ip_address(self, ip_address: str) -> None:
237        """Update the speaker's IP address after a (re)discovery."""
238        if ip_address == self._client.ip_address:
239            return
240        self.logger.debug("Address updated to %s for player %s", ip_address, self.name)
241        self._client.ip_address = ip_address
242        self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, ip_address)
243        self.mass.players.trigger_player_update(self.player_id)
244
245    # --- Private helpers ---
246
247    async def _listen(self) -> None:
248        """Connect to the speaker's notification websocket and handle push updates."""
249        while not self._stop_event.is_set():
250            uri = f"ws://{self._client.ip_address}:{NOTIFICATION_PORT}"
251            try:
252                async with self.mass.http_session.ws_connect(
253                    uri, protocols=WS_SUBPROTOCOLS, heartbeat=WS_HEARTBEAT
254                ) as ws:
255                    self.logger.debug("Connected to SoundTouch websocket: %s", uri)
256                    if not self._attr_available:
257                        self._attr_available = True
258                        self.update_state()
259                    async for msg in ws:
260                        if self._stop_event.is_set():
261                            break
262                        if msg.type == aiohttp.WSMsgType.TEXT:
263                            await self._handle_message(msg.data)
264                        elif msg.type == aiohttp.WSMsgType.BINARY:
265                            await self._handle_message(msg.data.decode())
266                        elif msg.type in (
267                            aiohttp.WSMsgType.ERROR,
268                            aiohttp.WSMsgType.CLOSE,
269                            aiohttp.WSMsgType.CLOSED,
270                        ):
271                            break
272            except asyncio.CancelledError:
273                raise
274            except (aiohttp.ClientError, OSError, TimeoutError, UnicodeDecodeError) as err:
275                self.logger.debug(
276                    "SoundTouch websocket error for %s: %s. Reconnecting in %ss",
277                    self.name,
278                    err,
279                    RECONNECT_DELAY,
280                )
281            if not self._stop_event.is_set():
282                with contextlib.suppress(TimeoutError):
283                    await asyncio.wait_for(self._stop_event.wait(), timeout=RECONNECT_DELAY)
284
285    async def _handle_message(self, message: str) -> None:
286        """Handle a single websocket notification message."""
287        # a physical preset button press maps to the configured Music Assistant media;
288        # the bulk "presetsUpdated" notification (which lists all presets) is not a press
289        if (
290            "presetsUpdated" not in message
291            and (preset_id := extract_preset_id(message)) is not None
292        ):
293            await self._handle_preset(preset_id)
294            return
295        try:
296            if "volumeUpdated" in message:
297                await self._refresh_volume()
298            if "nowPlayingUpdated" in message:
299                await self._refresh_now_playing()
300            if "zoneUpdated" in message:
301                await self._refresh_zone()
302        except (aiohttp.ClientError, TimeoutError, OSError) as err:
303            self.logger.debug("Failed to refresh state for %s: %s", self.name, err)
304
305    async def _handle_preset(self, preset_id: int) -> None:
306        """Play the Music Assistant media configured for the given preset button."""
307        if preset_id not in PRESET_IDS:
308            return
309        # preset buttons are mapped once on the provider, shared by all its speakers
310        media_id = cast("BoseSoundTouchProvider", self.provider).get_preset_media(preset_id)
311        if not media_id:
312            self.logger.warning(
313                "Preset %s pressed on %s but no media is configured", preset_id, self.name
314            )
315            return
316        self.logger.info("Preset %s pressed on %s, playing %s", preset_id, self.name, media_id)
317        try:
318            await self.mass.player_queues.play_media(queue_id=self.player_id, media=media_id)
319        except MusicAssistantError:
320            self.logger.exception("Unable to play media for preset %s", preset_id)
321
322    async def _refresh_volume(self) -> None:
323        """Refresh volume state from the speaker."""
324        volume = await self._client.get_volume()
325        self._attr_volume_level = volume.level
326        self._attr_volume_muted = volume.muted
327        self.update_state()
328
329    async def _refresh_now_playing(self) -> None:
330        """Refresh playback state from the speaker."""
331        self._update_state_from_now_playing(await self._client.get_now_playing())
332
333    async def _refresh_sources(self) -> None:
334        """Refresh the list of selectable native sources from the speaker."""
335        try:
336            sources = await self._client.get_sources()
337        except (aiohttp.ClientError, TimeoutError, OSError) as err:
338            self.logger.debug("Failed to fetch sources for %s: %s", self.name, err)
339            return
340        self._attr_source_list = [
341            PlayerSource(
342                id=_source_id(source.source, source.source_account),
343                name=source.name,
344                passive=False,
345                can_play_pause=True,
346                can_seek=False,
347                can_next_previous=True,
348            )
349            for source in sources
350            if source.ready
351        ]
352
353    async def _refresh_zone(self) -> None:
354        """Refresh multiroom zone (group) membership from the speaker."""
355        zone = await self._client.get_zone()
356        if zone.master_id == self._device_id and zone.member_ids:
357            members = [self.player_id]
358            members.extend(
359                f"{PLAYER_ID_PREFIX}{member_id}"
360                for member_id in zone.member_ids
361                if member_id != self._device_id
362            )
363            self._attr_group_members = members
364        else:
365            self._attr_group_members = []
366        self.update_state()
367
368    def _update_state_from_now_playing(self, now_playing: SoundTouchNowPlaying) -> None:
369        """Update player state from a now_playing snapshot."""
370        self._attr_powered = now_playing.source != SOURCE_STANDBY
371        if not self._attr_powered:
372            self._attr_playback_state = PlaybackState.IDLE
373            self._attr_active_source = None
374            self._attr_current_media = None
375            self.update_state()
376            return
377
378        if play_status_is_playing(now_playing.play_status):
379            self._attr_playback_state = PlaybackState.PLAYING
380        elif play_status_is_paused(now_playing.play_status):
381            self._attr_playback_state = PlaybackState.PAUSED
382        else:
383            self._attr_playback_state = PlaybackState.IDLE
384
385        if now_playing.position is not None:
386            self._attr_elapsed_time = float(now_playing.position)
387            self._attr_elapsed_time_last_updated = time.time()
388
389        active_queue = self.mass.player_queues.get(self.player_id)
390        if active_queue and active_queue.current_item:
391            # Music Assistant is the active source; audio is rendered via the linked
392            # protocol and Music Assistant owns the metadata, so don't override it here.
393            self._attr_active_source = self.player_id
394        else:
395            # a native source (Bluetooth, AUX, Spotify, ...) is playing on the speaker
396            self._attr_active_source = _source_id(now_playing.source, now_playing.source_account)
397            if now_playing.title or now_playing.artist or now_playing.album:
398                self._attr_current_media = PlayerMedia(
399                    uri=f"soundtouch://{now_playing.source}",
400                    media_type=MediaType.UNKNOWN,
401                    title=now_playing.title,
402                    artist=now_playing.artist,
403                    album=now_playing.album,
404                    image_url=now_playing.image_url,
405                    duration=now_playing.duration,
406                    source_id=self._attr_active_source,
407                )
408            else:
409                self._attr_current_media = None
410        self.update_state()
411
412    def _member_tuples(self, player_ids: list[str]) -> list[tuple[str, str]]:
413        """Return (device_id, ip_address) tuples for the given SoundTouch player ids."""
414        members: list[tuple[str, str]] = []
415        for player_id in player_ids:
416            player = self.mass.players.get_player(player_id)
417            if isinstance(player, BoseSoundTouchPlayer) and (
418                ip_address := player.device_info.ip_address
419            ):
420                members.append((player.device_id, ip_address))
421        return members
422
423
424def _source_id(source: str, source_account: str | None) -> str:
425    """Build a stable source id from a SoundTouch source and account."""
426    return f"{source}:{source_account}" if source_account else source
427