music-assistant-server

1"""Client."""
2
3import logging
4from contextlib import suppress
5from typing import TYPE_CHECKING, Any, cast
6from xml.etree.ElementTree import Element
7
8from aiohttp import ClientResponseError
9from defusedxml import ElementTree
10
11from music_assistant.providers.bose_soundtouch.client.exceptions import (
12    ApiError,
13    NotFoundError,
14    SoundtouchError,
15)
16from music_assistant.providers.bose_soundtouch.client.schema.enums import Key, KeyState
17from music_assistant.providers.bose_soundtouch.client.schema.models import (
18    Bass,
19    BassCapabilities,
20    Info,
21    NowPlaying,
22    Presets,
23    Sources,
24    Volume,
25    Zone,
26)
27
28from .session_configuration import SessionConfiguration
29
30if TYPE_CHECKING:
31    from aiohttp.client import ClientResponse
32
33
34def xml_escape(value: str) -> str:
35    """Escape xml."""
36    return value.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
37
38
39def parse_string(string: str) -> str | bool | int:
40    """Parse a string value."""
41    if string.lower() == "true":
42        return True
43    if string.lower() == "false":
44        return False
45
46    with suppress(ValueError):
47        return int(string)
48
49    return string
50
51
52def create_zone_xml(zone: Zone, sender_ip: str | None = None) -> str:
53    """Create zone xml."""
54    assert zone.leader
55    assert zone.leader.mac is not None
56
57    if sender_ip:
58        xml = f'<zone master="{zone.leader.mac}" senderIPAddress="{sender_ip}">'
59    else:
60        xml = f'<zone master="{zone.leader.mac}">'
61    for member in zone.members:
62        xml += f'<member ipaddress="{member.ip}">{member.mac}</member>'
63    xml += "</zone>"
64    return xml
65
66
67def create_notification_xml(app_key: str, url: str, volume: int | None = None) -> str:
68    """Create xml used for notifications."""
69    volume_xml = f"<volume>{volume}</volume>" if volume is not None else ""
70    return (
71        "<play_info>"
72        f"<app_key>{xml_escape(app_key)}</app_key>"
73        f"<url>{xml_escape(url)}</url>"
74        "<service>Music Assistant</service>"
75        "<reason>Music Assistant</reason>"
76        f"{volume_xml}"
77        "</play_info>"
78    )
79
80
81class SoundtouchDevice:
82    """SoundtouchDevice."""
83
84    def __init__(self, session_configuration: SessionConfiguration) -> None:
85        """Initialize."""
86        self.session_config = session_configuration
87
88        if self.session_config.logger is None:
89            self.logger = logging.getLogger(__name__)
90            logging.basicConfig()
91            self.logger.setLevel(logging.DEBUG)
92        else:
93            self.logger = self.session_config.logger
94
95    async def press_key(self, key: Key) -> None:
96        """Press a key."""
97        body_press = f'<key state="{KeyState.PRESS}" sender="Gabbo">{key}</key>'
98        body_release = f'<key state="{KeyState.RELEASE}" sender="Gabbo">{key}</key>'
99        await self._post("key", body_press)
100        await self._post("key", body_release)
101
102    async def get_sources(self) -> Sources:
103        """Get sources."""
104        element = await self._get("sources")
105        return Sources.from_dict(
106            {"sources": [{**x.attrib, "source_name": x.text} for x in element]}
107        )
108
109    async def select_source(self, source: str, source_account: str | None = None) -> None:
110        """Select a source on the speaker."""
111        account = f' sourceAccount="{source_account}"' if source_account else ""
112        await self._post("select", f'<ContentItem source="{source}"{account}></ContentItem>')
113
114    async def get_bass_capabilities(self) -> BassCapabilities:
115        """Get bass capabilities."""
116        element = await self._get("bassCapabilities")
117        return BassCapabilities.from_dict(
118            {x.tag: x.text if x.text is None else parse_string(x.text) for x in element}
119        )
120
121    async def get_bass(self) -> Bass:
122        """Get bass if supported (BassCapabilities to verify!)."""
123        element = await self._get("bass")
124        return Bass.from_dict(
125            {x.tag: x.text if x.text is None else parse_string(x.text) for x in element}
126        )
127
128    async def set_bass(self, value: int) -> None:
129        """Set bass if supported (BassCapabilities to verify!)."""
130        await self._post("bass", f"<bass>{value}</bass>")
131
132    async def get_zone(self) -> Zone:
133        """Get zone."""
134        element = await self._get("getZone")
135        return Zone.from_dict(
136            {
137                "leader": {"mac": element.attrib.get("master")},
138                "members": [{**x.attrib, "mac": x.text} for x in element],
139            }
140        )
141
142    async def set_zone(self, zone: Zone, sender_ip: str | None = None) -> None:
143        """Set zone."""
144        if (
145            zone.leader is None
146            or zone.leader.mac is None
147            or zone.leader.ip is None
148            or not zone.members
149            or any(member.mac is None for member in zone.members)
150            or any(member.ip is None for member in zone.members)
151        ):
152            raise SoundtouchError("Zone information is incomplete.")
153
154        # verify, that leader is part of member, and move to first position if necessary
155        leader = next((member for member in zone.members if member.mac == zone.leader.mac), None)
156        if leader is None:
157            zone.members.insert(0, zone.leader)
158        leader_index = zone.members.index(zone.leader)
159        if leader_index != 0:
160            zone.members.pop(leader_index)
161            zone.members.insert(0, zone.leader)
162        await self._post("setZone", create_zone_xml(zone, sender_ip))
163
164    async def add_zone_members(self, zone: Zone) -> None:
165        """Add zone.members to a zone."""
166        await self._add_or_remove_zone_members(zone, add_members=True)
167
168    async def remove_zone_members(self, zone: Zone) -> None:
169        """Remove zone.members from a zone."""
170        await self._add_or_remove_zone_members(zone, add_members=False)
171
172    async def get_now_playing(self, endpoint: str = "nowPlaying") -> NowPlaying:
173        """Get now playing."""
174        try:
175            element = await self._get(endpoint)
176        except NotFoundError:
177            # both cases seem to be supported by the API
178            element = await self._get(endpoint="now_playing")
179        d: dict[str, Any] = element.attrib
180        for el in element:
181            if el.tag == "ContentItem":
182                item_name: str | None = None
183                for sub_el in el.iter():
184                    if sub_el.tag == "itemName":
185                        item_name = sub_el.text
186                d["content_item"] = {
187                    **el.attrib,
188                    "sourceAccount": el.attrib.get("sourceAccount")
189                    or element.attrib.get("sourceAccount"),
190                    "item_name": item_name,
191                }
192            elif el.tag == "time":
193                d["time_information"] = {
194                    "total": parse_string(el.get("total", "-1")),
195                    "position": None if el.text is None else parse_string(el.text),
196                }
197            elif el.tag == "art":
198                d["art"] = {"status": el.get("artImageStatus"), "url": el.text}
199            else:
200                d[el.tag] = el.text
201
202        return NowPlaying.from_dict(d)
203
204    async def get_track_info(self) -> NowPlaying:
205        """Get track info."""
206        return await self.get_now_playing("trackInfo")
207
208    async def get_volume(self) -> Volume:
209        """Get volume."""
210        element = await self._get("volume")
211        return Volume.from_dict(
212            {x.tag: x.text if x.text is None else parse_string(x.text) for x in element}
213        )
214
215    async def set_volume(self, volume: int, *, mute: bool) -> None:
216        """Set volume."""
217        mute_str = "true" if mute else "false"
218        xml = f"<volume>{volume}<muteenabled>{mute_str}</muteenabled></volume>"
219        await self._post("volume", xml)
220
221    async def get_presets(self) -> Presets:
222        """Get presets."""
223        element = await self._get("presets")
224        presets_list: list[dict[str, Any]] = []
225        for el in element:
226            preset_dict: dict[str, Any] = el.attrib
227            for sub_el in el:
228                if sub_el.tag == "ContentItem":
229                    item_name: str | None = None
230                    for sub_sub_el in sub_el.iter():
231                        if sub_sub_el.tag == "itemName":
232                            item_name = sub_sub_el.text
233                    preset_dict["content_item"] = {**sub_el.attrib, "item_name": item_name}
234            presets_list.append(preset_dict)
235        return Presets.from_dict({"presets": presets_list})
236
237    async def store_preset(self, preset_id: int, preset_url: str) -> None:
238        """Store a preset."""
239        xml = (
240            f'<preset id="{preset_id}">'
241            '<ContentItem source="LOCAL_INTERNET_RADIO" '
242            f'type="stationurl" location="{preset_url}" sourceAccount="" isPresetable="true">'
243            f"<itemName>Music Assistant Preset {preset_id}</itemName>"
244            "</ContentItem>"
245            "</preset>"
246        )
247        await self._post("storePreset", xml)
248
249    async def get_info(self) -> Info:
250        """Get info necessary for us."""
251        response = await self._get("info")
252        mac_addresses: set[str] = set()
253        ip_addresses: set[str] = set()
254        for network_info in response.findall("networkInfo"):
255            if mac := network_info.findtext("macAddress"):
256                mac_addresses.add(mac)
257            if ip := network_info.findtext("ipAddress"):
258                ip_addresses.add(ip)
259        software_version: str | None = None
260        for component in response.iter("component"):
261            if version := component.findtext("softwareVersion"):
262                software_version = version.split(" ", 1)[0]
263                break
264
265        # our connection ip should already be present, but just in case
266        ip_addresses.add(self.session_config.ip)
267
268        return Info(
269            device_id=response.attrib.get("deviceID", ""),
270            name=response.findtext("name") or "Bose SoundTouch",
271            model=response.findtext("type"),
272            mac_addresses=mac_addresses,
273            ip_addresses=ip_addresses,
274            software_version=software_version,
275        )
276
277    async def set_name(self, name: str) -> None:
278        """Set Name."""
279        xml = f"<name>{name}</name>"
280        await self._post("name", xml)
281
282    async def play_notification(self, app_key: str, url: str, volume: int | None = None) -> None:
283        """Plays notification as in previous client."""
284        xml = create_notification_xml(app_key, url, volume)
285        await self._post("speaker", xml)
286
287    async def _add_or_remove_zone_members(self, zone: Zone, *, add_members: bool = True) -> None:
288        """Add or remove members to a zone."""
289        if (
290            zone.leader is None
291            or zone.leader.mac is None
292            or any(member.mac is None for member in zone.members)
293            or any(member.ip is None for member in zone.members)
294        ):
295            raise SoundtouchError("Zone information is incomplete.")
296        if add_members:
297            await self._post("addZoneSlave", create_zone_xml(zone))
298            return
299        await self._post("removeZoneSlave", create_zone_xml(zone))
300
301    async def _get(self, endpoint: str, params: dict[str, str | int] | None = None) -> Element[str]:
302        """GET request to abs api."""
303
304        async def _request() -> ClientResponse:
305            return await self.session_config.session.get(
306                f"http://{self.session_config.ip}:{self.session_config.http_port}/{endpoint}",
307                params=params,
308                timeout=self.session_config.timeout,
309            )
310
311        response = await _request()
312
313        status = response.status
314        if status == 404:
315            raise NotFoundError
316        if response.content_type == "text/xml" and status == 200:
317            _response = await response.read()
318            return cast("Element[str]", ElementTree.fromstring(_response))
319        raise ApiError(f"API GET call to {endpoint} failed.")
320
321    async def _post(
322        self,
323        endpoint: str,
324        data: str | None = None,
325    ) -> bytes:
326        """POST request to api."""
327
328        async def _request() -> ClientResponse:
329            return await self.session_config.session.post(
330                f"http://{self.session_config.ip}:{self.session_config.http_port}/{endpoint}",
331                data=data,
332                raise_for_status=True,
333                timeout=self.session_config.timeout,
334            )
335
336        try:
337            response = await _request()
338        except ClientResponseError as exc:
339            if exc.code == 404:
340                raise NotFoundError from exc
341            raise ApiError(f"API POST call to {endpoint} failed.") from exc
342
343        return await response.read()
344