music-assistant-server

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