music-assistant-server

1"""
2Lightweight async client for the Bose SoundTouch local API.
3
4SoundTouch speakers expose a simple XML-over-HTTP API on port 8090 for control
5and status, plus a websocket notification channel on port 8080 for push updates.
6"""
7
8from __future__ import annotations
9
10from dataclasses import dataclass, field
11from typing import TYPE_CHECKING, cast
12
13from aiohttp import ClientTimeout
14from defusedxml import ElementTree as DefusedET
15
16from .const import API_PORT, BUFFERING_STATE, PAUSE_STATE, PLAY_STATE, REQUEST_TIMEOUT
17
18if TYPE_CHECKING:
19    from xml.etree.ElementTree import Element
20
21    from aiohttp import ClientSession
22
23
24@dataclass
25class SoundTouchInfo:
26    """Static device information from the /info endpoint."""
27
28    device_id: str
29    name: str
30    model: str | None = None
31    mac_address: str | None = None
32    ip_address: str | None = None
33    software_version: str | None = None
34
35
36@dataclass
37class SoundTouchNowPlaying:
38    """Current playback state from the /now_playing endpoint."""
39
40    source: str
41    source_account: str | None = None
42    title: str | None = None
43    artist: str | None = None
44    album: str | None = None
45    image_url: str | None = None
46    duration: int | None = None
47    position: int | None = None
48    play_status: str | None = None
49
50
51@dataclass
52class SoundTouchVolume:
53    """Volume state from the /volume endpoint."""
54
55    level: int = 0
56    muted: bool = False
57
58
59@dataclass
60class SoundTouchSource:
61    """A selectable source from the /sources endpoint."""
62
63    source: str
64    source_account: str | None
65    name: str
66    ready: bool
67
68
69@dataclass
70class SoundTouchZone:
71    """Zone (multiroom) state from the /getZone endpoint."""
72
73    master_id: str | None = None
74    member_ids: list[str] = field(default_factory=list)
75
76
77class SoundTouchClient:
78    """Async client for a single Bose SoundTouch speaker's local API."""
79
80    def __init__(self, session: ClientSession, ip_address: str) -> None:
81        """
82        Initialize the client.
83
84        :param session: The shared aiohttp client session to use for requests.
85        :param ip_address: The IP address of the SoundTouch speaker.
86        """
87        self._session = session
88        self.ip_address = ip_address
89
90    async def get_info(self) -> SoundTouchInfo:
91        """Return static device information."""
92        return parse_info(await self._get("info"), self.ip_address)
93
94    async def get_now_playing(self) -> SoundTouchNowPlaying:
95        """Return the current playback state."""
96        return parse_now_playing(await self._get("now_playing"))
97
98    async def get_volume(self) -> SoundTouchVolume:
99        """Return the current volume state."""
100        return parse_volume(await self._get("volume"))
101
102    async def set_volume(self, level: int) -> None:
103        """Set the absolute volume level (0..100)."""
104        level = max(0, min(100, level))
105        await self._post("volume", f"<volume>{level}</volume>")
106
107    async def press_key(self, key: str) -> None:
108        """Send a key press (press followed by release) to the speaker."""
109        body_press = f'<key state="press" sender="Gabbo">{key}</key>'
110        body_release = f'<key state="release" sender="Gabbo">{key}</key>'
111        await self._post("key", body_press)
112        await self._post("key", body_release)
113
114    async def get_sources(self) -> list[SoundTouchSource]:
115        """Return the list of sources available on the speaker."""
116        return parse_sources(await self._get("sources"))
117
118    async def select_source(self, source: str, source_account: str | None = None) -> None:
119        """Select a source on the speaker."""
120        account = f' sourceAccount="{source_account}"' if source_account else ""
121        await self._post("select", f'<ContentItem source="{source}"{account}></ContentItem>')
122
123    async def play_notification(self, app_key: str, url: str, volume: int | None = None) -> None:
124        """
125        Play an audio notification (announcement) on the speaker.
126
127        Uses the Bose audio notification API which overlays the given URL on top of
128        the current playback (ducking it) and resumes afterwards. Requires a Bose
129        developer app key.
130        """
131        await self._post("speaker", _build_notification_xml(app_key, url, volume))
132
133    async def get_zone(self) -> SoundTouchZone:
134        """Return the current multiroom zone state."""
135        return parse_zone(await self._get("getZone"))
136
137    async def set_zone(self, master_id: str, members: list[tuple[str, str]]) -> None:
138        """
139        Create a multiroom zone with this speaker as master.
140
141        :param master_id: The device id of the zone master.
142        :param members: List of (device_id, ip_address) tuples for all zone members.
143        """
144        await self._post("setZone", _build_zone_xml(master_id, members))
145
146    async def add_zone_slaves(self, master_id: str, members: list[tuple[str, str]]) -> None:
147        """Add the given members to this speaker's existing zone."""
148        await self._post("addZoneSlave", _build_zone_xml(master_id, members))
149
150    async def remove_zone_slaves(self, master_id: str, members: list[tuple[str, str]]) -> None:
151        """Remove the given members from this speaker's zone."""
152        await self._post("removeZoneSlave", _build_zone_xml(master_id, members))
153
154    async def _get(self, path: str) -> Element:
155        url = f"http://{self.ip_address}:{API_PORT}/{path}"
156        async with self._session.get(url, timeout=_TIMEOUT) as resp:
157            resp.raise_for_status()
158            return _parse_xml(await resp.text())
159
160    async def _post(self, path: str, body: str) -> None:
161        url = f"http://{self.ip_address}:{API_PORT}/{path}"
162        async with self._session.post(url, data=body, timeout=_TIMEOUT) as resp:
163            resp.raise_for_status()
164
165
166def parse_info(root: Element, fallback_ip: str | None = None) -> SoundTouchInfo:
167    """Parse a /info response into a SoundTouchInfo."""
168    mac_address: str | None = None
169    ip_address: str | None = None
170    for network_info in root.findall("networkInfo"):
171        if (mac := network_info.findtext("macAddress")) and not mac_address:
172            mac_address = mac
173        if (ip := network_info.findtext("ipAddress")) and not ip_address:
174            ip_address = ip
175    software_version: str | None = None
176    for component in root.iter("component"):
177        if version := component.findtext("softwareVersion"):
178            software_version = version.split(" ", 1)[0]
179            break
180    return SoundTouchInfo(
181        device_id=root.attrib.get("deviceID", ""),
182        name=root.findtext("name") or "Bose SoundTouch",
183        model=root.findtext("type"),
184        mac_address=mac_address,
185        ip_address=ip_address or fallback_ip,
186        software_version=software_version,
187    )
188
189
190def parse_now_playing(root: Element) -> SoundTouchNowPlaying:
191    """Parse a /now_playing response into a SoundTouchNowPlaying."""
192    art_url: str | None = None
193    if (art := root.find("art")) is not None and art.text:
194        if art.attrib.get("artImageStatus") == "IMAGE_PRESENT":
195            art_url = art.text
196    duration: int | None = None
197    position: int | None = None
198    if (time_elem := root.find("time")) is not None:
199        duration = _int_or_none(time_elem.attrib.get("total"))
200        position = _int_or_none(time_elem.text)
201    return SoundTouchNowPlaying(
202        source=root.attrib.get("source", ""),
203        source_account=root.attrib.get("sourceAccount") or None,
204        title=root.findtext("track") or root.findtext("stationName"),
205        artist=root.findtext("artist"),
206        album=root.findtext("album"),
207        image_url=art_url,
208        duration=duration,
209        position=position,
210        play_status=root.findtext("playStatus"),
211    )
212
213
214def parse_volume(root: Element) -> SoundTouchVolume:
215    """Parse a /volume response into a SoundTouchVolume."""
216    return SoundTouchVolume(
217        level=_int_or_none(root.findtext("actualvolume")) or 0,
218        muted=(root.findtext("muteenabled") or "").lower() == "true",
219    )
220
221
222def parse_sources(root: Element) -> list[SoundTouchSource]:
223    """Parse a /sources response into a list of SoundTouchSource."""
224    sources: list[SoundTouchSource] = []
225    for item in root.findall("sourceItem"):
226        source = item.attrib.get("source", "")
227        if not source:
228            continue
229        sources.append(
230            SoundTouchSource(
231                source=source,
232                source_account=item.attrib.get("sourceAccount") or None,
233                name=(item.text or source).strip(),
234                ready=item.attrib.get("status") == "READY",
235            )
236        )
237    return sources
238
239
240def parse_zone(root: Element) -> SoundTouchZone:
241    """Parse a /getZone response into a SoundTouchZone."""
242    master_id = root.attrib.get("master")
243    member_ids = [member.text for member in root.findall("member") if member.text]
244    return SoundTouchZone(master_id=master_id or None, member_ids=member_ids)
245
246
247def play_status_is_playing(play_status: str | None) -> bool:
248    """Return whether the given playStatus represents active playback."""
249    return play_status in (PLAY_STATE, BUFFERING_STATE)
250
251
252def play_status_is_paused(play_status: str | None) -> bool:
253    """Return whether the given playStatus represents paused playback."""
254    return play_status == PAUSE_STATE
255
256
257def extract_preset_id(message: str) -> int | None:
258    """
259    Extract a Bose SoundTouch favorite/preset id from a websocket XML message.
260
261    Detects a physical preset button press from the speaker's notification channel.
262    """
263    try:
264        root = DefusedET.fromstring(message)
265    except DefusedET.ParseError:
266        return None
267
268    preset_id = next(
269        (
270            element.attrib.get("id")
271            for element in root.iter()
272            if _local_name(element.tag) == "preset"
273        ),
274        None,
275    )
276
277    try:
278        return int(preset_id) if preset_id else None
279    except TypeError, ValueError:
280        return None
281
282
283def _parse_xml(payload: str) -> Element:
284    return cast("Element", DefusedET.fromstring(payload))
285
286
287def _local_name(tag: str) -> str:
288    """Return an XML tag name without its namespace prefix."""
289    return tag.rsplit("}", 1)[-1]
290
291
292def _int_or_none(value: str | None) -> int | None:
293    if value is None:
294        return None
295    try:
296        return int(value)
297    except ValueError, TypeError:
298        return None
299
300
301def _build_zone_xml(master_id: str, members: list[tuple[str, str]]) -> str:
302    member_xml = "".join(
303        f'<member ipaddress="{ip_address}">{device_id}</member>'
304        for device_id, ip_address in members
305    )
306    return f'<zone master="{master_id}">{member_xml}</zone>'
307
308
309def _build_notification_xml(app_key: str, url: str, volume: int | None = None) -> str:
310    volume_xml = f"<volume>{volume}</volume>" if volume is not None else ""
311    return (
312        "<play_info>"
313        f"<app_key>{_xml_escape(app_key)}</app_key>"
314        f"<url>{_xml_escape(url)}</url>"
315        "<service>Music Assistant</service>"
316        "<reason>Music Assistant</reason>"
317        f"{volume_xml}"
318        "</play_info>"
319    )
320
321
322def _xml_escape(value: str) -> str:
323    return value.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
324
325
326_TIMEOUT = ClientTimeout(total=REQUEST_TIMEOUT)
327