/
/
/
1"""Helpers to make an UPnP request."""
2
3import logging
4from contextlib import suppress
5from typing import TYPE_CHECKING
6
7import aiohttp
8
9from music_assistant.helpers.upnp import (
10 get_xml_soap_play,
11 get_xml_soap_set_next_url,
12 get_xml_soap_set_url,
13 get_xml_soap_stop,
14)
15from music_assistant.helpers.util import format_ip_for_url
16from music_assistant.models.player import PlayerMedia
17from music_assistant.providers.bose_soundtouch.const import UPNP_CONTROL_ENDPOINT, UPNP_PORT
18
19if TYPE_CHECKING:
20 from music_assistant.providers.bose_soundtouch.player import BoseSoundTouchPlayer
21
22LOGGER = logging.getLogger(__name__)
23
24
25def get_headers(soap_action: str) -> dict[str, str]:
26 """Get headers for Bose Soundtouch."""
27 return {
28 "Content-Type": 'text/xml; charset="utf-8"',
29 "SOAPACTION": f'"{soap_action}"',
30 "Accept": "*/*",
31 "User-Agent": "MusicAssistant",
32 }
33
34
35def get_upnp_ctrl_url(device: BoseSoundTouchPlayer) -> str:
36 """Get UPNP control URL."""
37 return f"http://{format_ip_for_url(device._client.session_config.ip)}:{UPNP_PORT}/{UPNP_CONTROL_ENDPOINT}"
38
39
40async def avt_play(
41 client: aiohttp.ClientSession,
42 physical_device: BoseSoundTouchPlayer,
43) -> None:
44 """Play."""
45 ctrl_url = get_upnp_ctrl_url(physical_device)
46 xml, soap_action = get_xml_soap_play()
47 await _post_soap(client, ctrl_url, xml, soap_action, "Play")
48
49
50async def avt_stop(
51 client: aiohttp.ClientSession,
52 physical_device: BoseSoundTouchPlayer,
53) -> None:
54 """Stop."""
55 ctrl_url = get_upnp_ctrl_url(physical_device)
56 xml, soap_action = get_xml_soap_stop()
57 await _post_soap(client, ctrl_url, xml, soap_action, "Stop")
58
59
60async def avt_set_url(
61 client: aiohttp.ClientSession,
62 physical_device: BoseSoundTouchPlayer,
63 player_media: PlayerMedia,
64 enqueue: bool = False,
65) -> None:
66 """
67 Set Url.
68
69 If device is playing, this will just continue with new media.
70 """
71 ctrl_url = get_upnp_ctrl_url(physical_device)
72 if enqueue:
73 xml, soap_action = get_xml_soap_set_next_url(player_media)
74 op_name = "SetNextAVTransportURI"
75 else:
76 xml, soap_action = get_xml_soap_set_url(player_media)
77 op_name = "SetAVTransportURI"
78 LOGGER.debug("AVT %s uri=%s", op_name, player_media.uri)
79 await _post_soap(client, ctrl_url, xml, soap_action, op_name)
80
81
82async def _post_soap(
83 client: aiohttp.ClientSession,
84 ctrl_url: str,
85 xml: str,
86 soap_action: str,
87 op_name: str,
88) -> aiohttp.ClientResponse:
89 """POST a SOAP request and log a warning on 4xx/5xx error responses."""
90 headers = get_headers(soap_action)
91 response = await client.post(ctrl_url, headers=headers, data=xml)
92 if response.status >= 400:
93 body_excerpt = ""
94 with suppress(aiohttp.ClientError, UnicodeError):
95 body_excerpt = (await response.read())[:300].decode(errors="replace")
96 LOGGER.warning(
97 "AVT %s failed: status=%s url=%s body=%s",
98 op_name,
99 response.status,
100 ctrl_url,
101 body_excerpt,
102 )
103 return response
104