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        # the notification API reports no completion, so wait out the announcement here:
206        # its audio is only served for as long as this call is running
207        duration = await self.mass.streams.get_announcement_duration(announcement)
208        await asyncio.sleep(duration or 10)
209
210    async def set_members(
211        self,
212        player_ids_to_add: list[str] | None = None,
213        player_ids_to_remove: list[str] | None = None,
214    ) -> None:
215        """Handle SET_MEMBERS command using the SoundTouch multiroom zone API."""
216        if player_ids_to_add:
217            members = self._member_tuples(player_ids_to_add)
218            if members:
219                zone = await self._client.get_zone()
220                if zone.master_id == self._device_id:
221                    await self._client.add_zone_slaves(self._device_id, members)
222                else:
223                    own = (self._device_id, self.device_info.ip_address or "")
224                    await self._client.set_zone(self._device_id, [own, *members])
225        if player_ids_to_remove:
226            members = self._member_tuples(player_ids_to_remove)
227            if members:
228                await self._client.remove_zone_slaves(self._device_id, members)
229        await self._refresh_zone()
230        for member_id in (player_ids_to_add or []) + (player_ids_to_remove or []):
231            self.mass.players.trigger_player_update(member_id)
232
233    # --- Public helpers ---
234
235    @property
236    def device_id(self) -> str:
237        """Return the Bose SoundTouch device id of this player."""
238        return self._device_id
239
240    def update_ip_address(self, ip_address: str) -> None:
241        """Update the speaker's IP address after a (re)discovery."""
242        if ip_address == self._client.ip_address:
243            return
244        self.logger.debug("Address updated to %s for player %s", ip_address, self.name)
245        self._client.ip_address = ip_address
246        self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, ip_address)
247        self.mass.players.trigger_player_update(self.player_id)
248
249    # --- Private helpers ---
250
251    async def _listen(self) -> None:
252        """Connect to the speaker's notification websocket and handle push updates."""
253        while not self._stop_event.is_set():
254            uri = f"ws://{self._client.ip_address}:{NOTIFICATION_PORT}"
255            try:
256                async with self.mass.http_session.ws_connect(
257                    uri, protocols=WS_SUBPROTOCOLS, heartbeat=WS_HEARTBEAT
258                ) as ws:
259                    self.logger.debug("Connected to SoundTouch websocket: %s", uri)
260                    if not self._attr_available:
261                        self._attr_available = True
262                        self.update_state()
263                    async for msg in ws:
264                        if self._stop_event.is_set():
265                            break
266                        if msg.type == aiohttp.WSMsgType.TEXT:
267                            await self._handle_message(msg.data)
268                        elif msg.type == aiohttp.WSMsgType.BINARY:
269                            await self._handle_message(msg.data.decode())
270                        elif msg.type in (
271                            aiohttp.WSMsgType.ERROR,
272                            aiohttp.WSMsgType.CLOSE,
273                            aiohttp.WSMsgType.CLOSED,
274                        ):
275                            break
276            except asyncio.CancelledError:
277                raise
278            except (aiohttp.ClientError, OSError, TimeoutError, UnicodeDecodeError) as err:
279                self.logger.debug(
280                    "SoundTouch websocket error for %s: %s. Reconnecting in %ss",
281                    self.name,
282                    err,
283                    RECONNECT_DELAY,
284                )
285            if not self._stop_event.is_set():
286                with contextlib.suppress(TimeoutError):
287                    await asyncio.wait_for(self._stop_event.wait(), timeout=RECONNECT_DELAY)
288
289    async def _handle_message(self, message: str) -> None:
290        """Handle a single websocket notification message."""
291        # a physical preset button press maps to the configured Music Assistant media;
292        # the bulk "presetsUpdated" notification (which lists all presets) is not a press
293        if (
294            "presetsUpdated" not in message
295            and (preset_id := extract_preset_id(message)) is not None
296        ):
297            await self._handle_preset(preset_id)
298            return
299        try:
300            if "volumeUpdated" in message:
301                await self._refresh_volume()
302            if "nowPlayingUpdated" in message:
303                await self._refresh_now_playing()
304            if "zoneUpdated" in message:
305                await self._refresh_zone()
306        except (aiohttp.ClientError, TimeoutError, OSError) as err:
307            self.logger.debug("Failed to refresh state for %s: %s", self.name, err)
308
309    async def _handle_preset(self, preset_id: int) -> None:
310        """Play the Music Assistant media configured for the given preset button."""
311        if preset_id not in PRESET_IDS:
312            return
313        # preset buttons are mapped once on the provider, shared by all its speakers
314        media_id = cast("BoseSoundTouchProvider", self.provider).get_preset_media(preset_id)
315        if not media_id:
316            self.logger.warning(
317                "Preset %s pressed on %s but no media is configured", preset_id, self.name
318            )
319            return
320        self.logger.info("Preset %s pressed on %s, playing %s", preset_id, self.name, media_id)
321        try:
322            await self.mass.player_queues.play_media(queue_id=self.player_id, media=media_id)
323        except MusicAssistantError:
324            self.logger.exception("Unable to play media for preset %s", preset_id)
325
326    async def _refresh_volume(self) -> None:
327        """Refresh volume state from the speaker."""
328        volume = await self._client.get_volume()
329        self._attr_volume_level = volume.level
330        self._attr_volume_muted = volume.muted
331        self.update_state()
332
333    async def _refresh_now_playing(self) -> None:
334        """Refresh playback state from the speaker."""
335        self._update_state_from_now_playing(await self._client.get_now_playing())
336
337    async def _refresh_sources(self) -> None:
338        """Refresh the list of selectable native sources from the speaker."""
339        try:
340            sources = await self._client.get_sources()
341        except (aiohttp.ClientError, TimeoutError, OSError) as err:
342            self.logger.debug("Failed to fetch sources for %s: %s", self.name, err)
343            return
344        self._attr_source_list = [
345            PlayerSource(
346                id=_source_id(source.source, source.source_account),
347                name=source.name,
348                passive=False,
349                can_play_pause=True,
350                can_seek=False,
351                can_next_previous=True,
352            )
353            for source in sources
354            if source.ready
355        ]
356
357    async def _refresh_zone(self) -> None:
358        """Refresh multiroom zone (group) membership from the speaker."""
359        zone = await self._client.get_zone()
360        if zone.master_id == self._device_id and zone.member_ids:
361            members = [self.player_id]
362            members.extend(
363                f"{PLAYER_ID_PREFIX}{member_id}"
364                for member_id in zone.member_ids
365                if member_id != self._device_id
366            )
367            self._attr_group_members = members
368        else:
369            self._attr_group_members = []
370        self.update_state()
371
372    def _update_state_from_now_playing(self, now_playing: SoundTouchNowPlaying) -> None:
373        """Update player state from a now_playing snapshot."""
374        self._attr_powered = now_playing.source != SOURCE_STANDBY
375        if not self._attr_powered:
376            self._attr_playback_state = PlaybackState.IDLE
377            self._attr_active_source = None
378            self._attr_current_media = None
379            self.update_state()
380            return
381
382        if play_status_is_playing(now_playing.play_status):
383            self._attr_playback_state = PlaybackState.PLAYING
384        elif play_status_is_paused(now_playing.play_status):
385            self._attr_playback_state = PlaybackState.PAUSED
386        else:
387            self._attr_playback_state = PlaybackState.IDLE
388
389        if now_playing.position is not None:
390            self._attr_elapsed_time = float(now_playing.position)
391            self._attr_elapsed_time_last_updated = time.time()
392
393        active_queue = self.mass.player_queues.get(self.player_id)
394        if active_queue and active_queue.current_item:
395            # Music Assistant is the active source; audio is rendered via the linked
396            # protocol and Music Assistant owns the metadata, so don't override it here.
397            self._attr_active_source = self.player_id
398        else:
399            # a native source (Bluetooth, AUX, Spotify, ...) is playing on the speaker
400            self._attr_active_source = _source_id(now_playing.source, now_playing.source_account)
401            if now_playing.title or now_playing.artist or now_playing.album:
402                self._attr_current_media = PlayerMedia(
403                    uri=f"soundtouch://{now_playing.source}",
404                    media_type=MediaType.UNKNOWN,
405                    title=now_playing.title,
406                    artist=now_playing.artist,
407                    album=now_playing.album,
408                    image_url=now_playing.image_url,
409                    duration=now_playing.duration,
410                    source_id=self._attr_active_source,
411                )
412            else:
413                self._attr_current_media = None
414        self.update_state()
415
416    def _member_tuples(self, player_ids: list[str]) -> list[tuple[str, str]]:
417        """Return (device_id, ip_address) tuples for the given SoundTouch player ids."""
418        members: list[tuple[str, str]] = []
419        for player_id in player_ids:
420            player = self.mass.players.get_player(player_id)
421            if isinstance(player, BoseSoundTouchPlayer) and (
422                ip_address := player.device_info.ip_address
423            ):
424                members.append((player.device_id, ip_address))
425        return members
426
427
428def _source_id(source: str, source_account: str | None) -> str:
429    """Build a stable source id from a SoundTouch source and account."""
430    return f"{source}:{source_account}" if source_account else source
431