/
/
/
1"""Helpers to make an UPnP request."""
2
3import aiohttp
4
5from music_assistant.helpers.upnp import (
6 get_xml_soap_media_info,
7 get_xml_soap_next,
8 get_xml_soap_pause,
9 get_xml_soap_play,
10 get_xml_soap_previous,
11 get_xml_soap_set_next_url,
12 get_xml_soap_set_url,
13 get_xml_soap_stop,
14 get_xml_soap_transport_info,
15)
16from music_assistant.models.player import PlayerMedia
17from music_assistant.providers.musiccast.constants import (
18 MC_DEVICE_UPNP_CTRL_ENDPOINT,
19 MC_DEVICE_UPNP_PORT,
20)
21from music_assistant.providers.musiccast.musiccast import MusicCastPhysicalDevice
22
23
24def get_headers(xml: str, soap_action: str) -> dict[str, str]:
25 """Get headers for MusicCast."""
26 return {
27 "Content-Type": 'text/xml; charset="utf-8"',
28 "SOAPACTION": f'"{soap_action}"',
29 "Accept": "*/*",
30 "User-Agent": "MusicCast/6.00 (Android)",
31 "Content-Length": str(len(xml)),
32 }
33
34
35def get_upnp_ctrl_url(physical_device: MusicCastPhysicalDevice) -> str:
36 """Get UPNP control URL."""
37 return f"http://{physical_device.device.device.ip}:{MC_DEVICE_UPNP_PORT}/{MC_DEVICE_UPNP_CTRL_ENDPOINT}"
38
39
40async def avt_play(
41 client: aiohttp.ClientSession,
42 physical_device: MusicCastPhysicalDevice,
43) -> None:
44 """Play."""
45 ctrl_url = get_upnp_ctrl_url(physical_device)
46 xml, soap_action = get_xml_soap_play()
47 headers = get_headers(xml, soap_action)
48 await client.post(ctrl_url, headers=headers, data=xml)
49
50
51async def avt_stop(
52 client: aiohttp.ClientSession,
53 physical_device: MusicCastPhysicalDevice,
54) -> None:
55 """Play."""
56 ctrl_url = get_upnp_ctrl_url(physical_device)
57 xml, soap_action = get_xml_soap_stop()
58 headers = get_headers(xml, soap_action)
59 await client.post(ctrl_url, headers=headers, data=xml)
60
61
62async def avt_pause(
63 client: aiohttp.ClientSession,
64 physical_device: MusicCastPhysicalDevice,
65) -> None:
66 """Play."""
67 ctrl_url = get_upnp_ctrl_url(physical_device)
68 xml, soap_action = get_xml_soap_pause()
69 headers = get_headers(xml, soap_action)
70 await client.post(ctrl_url, headers=headers, data=xml)
71
72
73async def avt_next(
74 client: aiohttp.ClientSession,
75 physical_device: MusicCastPhysicalDevice,
76) -> None:
77 """Play."""
78 ctrl_url = get_upnp_ctrl_url(physical_device)
79 xml, soap_action = get_xml_soap_next()
80 headers = get_headers(xml, soap_action)
81 await client.post(ctrl_url, headers=headers, data=xml)
82
83
84async def avt_previous(
85 client: aiohttp.ClientSession,
86 physical_device: MusicCastPhysicalDevice,
87) -> None:
88 """Play."""
89 ctrl_url = get_upnp_ctrl_url(physical_device)
90 xml, soap_action = get_xml_soap_previous()
91 headers = get_headers(xml, soap_action)
92 await client.post(ctrl_url, headers=headers, data=xml)
93
94
95async def avt_get_media_info(
96 client: aiohttp.ClientSession,
97 physical_device: MusicCastPhysicalDevice,
98) -> str:
99 """Get Media Info."""
100 ctrl_url = get_upnp_ctrl_url(physical_device)
101 xml, soap_action = get_xml_soap_media_info()
102 headers = get_headers(xml, soap_action)
103 response = await client.post(ctrl_url, headers=headers, data=xml)
104 response_text = await response.read()
105 return response_text.decode()
106
107
108async def avt_get_transport_info(
109 client: aiohttp.ClientSession,
110 physical_device: MusicCastPhysicalDevice,
111) -> str:
112 """Get Media Info."""
113 ctrl_url = get_upnp_ctrl_url(physical_device)
114 xml, soap_action = get_xml_soap_transport_info()
115 headers = get_headers(xml, soap_action)
116 response = await client.post(ctrl_url, headers=headers, data=xml)
117 response_text = await response.read()
118 return response_text.decode()
119
120
121async def avt_set_url(
122 client: aiohttp.ClientSession,
123 physical_device: MusicCastPhysicalDevice,
124 player_media: PlayerMedia,
125 enqueue: bool = False,
126) -> None:
127 """Set Url.
128
129 If device is playing, this will just continue with new media.
130 """
131 ctrl_url = get_upnp_ctrl_url(physical_device)
132 if enqueue:
133 xml, soap_action = get_xml_soap_set_next_url(player_media)
134 else:
135 xml, soap_action = get_xml_soap_set_url(player_media)
136 headers = get_headers(xml, soap_action)
137 await client.post(ctrl_url, headers=headers, data=xml)
138
139
140def search_xml(xml: str, tag: str) -> str | None:
141 """Search single line xml for these tags."""
142 start_str = f"<{tag}>"
143 end_str = f"</{tag}>"
144 start_int = xml.find(start_str)
145 end_int = xml.find(end_str)
146 if start_int == -1 or end_int == -1:
147 return None
148 return xml[start_int + len(start_str) : end_int]
149