/
/
/
1"""Helpers to make an UPnP request."""
2
3import logging
4from contextlib import suppress
5
6import aiohttp
7
8from music_assistant.helpers.upnp import (
9 get_xml_soap_media_info,
10 get_xml_soap_next,
11 get_xml_soap_pause,
12 get_xml_soap_play,
13 get_xml_soap_previous,
14 get_xml_soap_set_next_url,
15 get_xml_soap_set_url,
16 get_xml_soap_stop,
17 get_xml_soap_transport_info,
18)
19from music_assistant.helpers.util import format_ip_for_url
20from music_assistant.models.player import PlayerMedia
21from music_assistant.providers.musiccast.constants import (
22 MC_DEVICE_UPNP_CTRL_ENDPOINT,
23 MC_DEVICE_UPNP_PORT,
24)
25from music_assistant.providers.musiccast.musiccast import MusicCastPhysicalDevice
26
27LOGGER = logging.getLogger(__name__)
28
29
30def get_headers(xml: str, soap_action: str) -> dict[str, str]:
31 """Get headers for MusicCast."""
32 return {
33 "Content-Type": 'text/xml; charset="utf-8"',
34 "SOAPACTION": f'"{soap_action}"',
35 "Accept": "*/*",
36 "User-Agent": "MusicCast/6.00 (Android)",
37 "Content-Length": str(len(xml)),
38 }
39
40
41def get_upnp_ctrl_url(physical_device: MusicCastPhysicalDevice) -> str:
42 """Get UPNP control URL."""
43 return f"http://{format_ip_for_url(physical_device.device.device.ip)}:{MC_DEVICE_UPNP_PORT}/{MC_DEVICE_UPNP_CTRL_ENDPOINT}"
44
45
46async def _post_soap(
47 client: aiohttp.ClientSession,
48 ctrl_url: str,
49 xml: str,
50 soap_action: str,
51 op_name: str,
52) -> aiohttp.ClientResponse:
53 """POST a SOAP request and log a warning on 4xx/5xx error responses."""
54 headers = get_headers(xml, soap_action)
55 response = await client.post(ctrl_url, headers=headers, data=xml)
56 if response.status >= 400:
57 body_excerpt = ""
58 with suppress(aiohttp.ClientError, UnicodeError):
59 body_excerpt = (await response.read())[:300].decode(errors="replace")
60 LOGGER.warning(
61 "AVT %s failed: status=%s url=%s body=%s",
62 op_name,
63 response.status,
64 ctrl_url,
65 body_excerpt,
66 )
67 return response
68
69
70async def avt_play(
71 client: aiohttp.ClientSession,
72 physical_device: MusicCastPhysicalDevice,
73) -> None:
74 """Play."""
75 ctrl_url = get_upnp_ctrl_url(physical_device)
76 xml, soap_action = get_xml_soap_play()
77 await _post_soap(client, ctrl_url, xml, soap_action, "Play")
78
79
80async def avt_stop(
81 client: aiohttp.ClientSession,
82 physical_device: MusicCastPhysicalDevice,
83) -> None:
84 """Stop."""
85 ctrl_url = get_upnp_ctrl_url(physical_device)
86 xml, soap_action = get_xml_soap_stop()
87 await _post_soap(client, ctrl_url, xml, soap_action, "Stop")
88
89
90async def avt_pause(
91 client: aiohttp.ClientSession,
92 physical_device: MusicCastPhysicalDevice,
93) -> None:
94 """Pause."""
95 ctrl_url = get_upnp_ctrl_url(physical_device)
96 xml, soap_action = get_xml_soap_pause()
97 await _post_soap(client, ctrl_url, xml, soap_action, "Pause")
98
99
100async def avt_next(
101 client: aiohttp.ClientSession,
102 physical_device: MusicCastPhysicalDevice,
103) -> None:
104 """Next."""
105 ctrl_url = get_upnp_ctrl_url(physical_device)
106 xml, soap_action = get_xml_soap_next()
107 await _post_soap(client, ctrl_url, xml, soap_action, "Next")
108
109
110async def avt_previous(
111 client: aiohttp.ClientSession,
112 physical_device: MusicCastPhysicalDevice,
113) -> None:
114 """Previous."""
115 ctrl_url = get_upnp_ctrl_url(physical_device)
116 xml, soap_action = get_xml_soap_previous()
117 await _post_soap(client, ctrl_url, xml, soap_action, "Previous")
118
119
120async def avt_get_media_info(
121 client: aiohttp.ClientSession,
122 physical_device: MusicCastPhysicalDevice,
123) -> str:
124 """Get Media Info."""
125 ctrl_url = get_upnp_ctrl_url(physical_device)
126 xml, soap_action = get_xml_soap_media_info()
127 response = await _post_soap(client, ctrl_url, xml, soap_action, "GetMediaInfo")
128 if response.status >= 400:
129 raise aiohttp.ClientError(f"GetMediaInfo returned {response.status}")
130 response_text = await response.read()
131 return response_text.decode()
132
133
134async def avt_get_transport_info(
135 client: aiohttp.ClientSession,
136 physical_device: MusicCastPhysicalDevice,
137) -> str:
138 """Get Transport Info."""
139 ctrl_url = get_upnp_ctrl_url(physical_device)
140 xml, soap_action = get_xml_soap_transport_info()
141 response = await _post_soap(client, ctrl_url, xml, soap_action, "GetTransportInfo")
142 if response.status >= 400:
143 raise aiohttp.ClientError(f"GetTransportInfo returned {response.status}")
144 response_text = await response.read()
145 return response_text.decode()
146
147
148async def avt_set_url(
149 client: aiohttp.ClientSession,
150 physical_device: MusicCastPhysicalDevice,
151 player_media: PlayerMedia,
152 enqueue: bool = False,
153) -> None:
154 """
155 Set Url.
156
157 If device is playing, this will just continue with new media.
158 """
159 ctrl_url = get_upnp_ctrl_url(physical_device)
160 if enqueue:
161 xml, soap_action = get_xml_soap_set_next_url(player_media)
162 op_name = "SetNextAVTransportURI"
163 else:
164 xml, soap_action = get_xml_soap_set_url(player_media)
165 op_name = "SetAVTransportURI"
166 LOGGER.debug("AVT %s uri=%s", op_name, player_media.uri)
167 await _post_soap(client, ctrl_url, xml, soap_action, op_name)
168
169
170def search_xml(xml: str, tag: str) -> str | None:
171 """Search single line xml for these tags."""
172 start_str = f"<{tag}>"
173 end_str = f"</{tag}>"
174 start_int = xml.find(start_str)
175 end_int = xml.find(end_str)
176 if start_int == -1 or end_int == -1:
177 return None
178 return xml[start_int + len(start_str) : end_int]
179