/
/
/
1"""Helper(s) to create DIDL Lite metadata for Sonos/DLNA players."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6from xml.sax.saxutils import escape as xmlescape
7
8from music_assistant_models.enums import MediaType
9
10from music_assistant.constants import MASS_LOGO_ONLINE
11
12if TYPE_CHECKING:
13 from music_assistant.models.player import PlayerMedia
14
15
16# ruff: noqa: E501
17
18
19# XML
20def _get_soap_action(command: str) -> str:
21 return f"urn:schemas-upnp-org:service:AVTransport:1#{command}"
22
23
24def _get_body(command: str, arguments: str = "", service: str = "AVTransport") -> str:
25 return (
26 f'<u:{command} xmlns:u="urn:schemas-upnp-org:service:{service}:1">'
27 r"<InstanceID>0</InstanceID>"
28 f"{arguments}"
29 f"</u:{command}>"
30 )
31
32
33def _get_xml(body: str) -> str:
34 return (
35 r'<?xml version="1.0"?>'
36 r'<s:Envelope s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" xmlns:s="http://schemas.xmlsoap.org/soap/envelope/">'
37 r"<s:Body>"
38 f"{body}"
39 r"</s:Body>"
40 r"</s:Envelope>"
41 )
42
43
44def get_xml_soap_play() -> tuple[str, str]:
45 """Get UPnP xml and soap for Play."""
46 command = "Play"
47 arguments = r"<Speed>1</Speed>"
48 return _get_xml(_get_body(command, arguments)), _get_soap_action(command)
49
50
51def get_xml_soap_stop() -> tuple[str, str]:
52 """Get UPnP xml and soap for Stop."""
53 command = "Stop"
54 return _get_xml(_get_body(command)), _get_soap_action(command)
55
56
57def get_xml_soap_pause() -> tuple[str, str]:
58 """Get UPnP xml and soap for Pause."""
59 command = "Pause"
60 return _get_xml(_get_body(command)), _get_soap_action(command)
61
62
63def get_xml_soap_next() -> tuple[str, str]:
64 """Get UPnP xml and soap for Next."""
65 command = "Next"
66 return _get_xml(_get_body(command)), _get_soap_action(command)
67
68
69def get_xml_soap_previous() -> tuple[str, str]:
70 """Get UPnP xml and soap for Previous."""
71 command = "Previous"
72 return _get_xml(_get_body(command)), _get_soap_action(command)
73
74
75def get_xml_soap_transport_info() -> tuple[str, str]:
76 """Get UPnP xml and soap for GetTransportInfo."""
77 command = "GetTransportInfo"
78 return _get_xml(_get_body(command)), _get_soap_action(command)
79
80
81def get_xml_soap_media_info() -> tuple[str, str]:
82 """Get UPnP xml and soap for GetMediaInfo."""
83 command = "GetMediaInfo"
84 return _get_xml(_get_body(command)), _get_soap_action(command)
85
86
87def get_xml_soap_set_url(player_media: PlayerMedia) -> tuple[str, str]:
88 """Get UPnP xml and soap for SetAVTransportURI."""
89 metadata = create_didl_metadata_str(player_media)
90 command = "SetAVTransportURI"
91 arguments = (
92 f"<CurrentURI>{player_media.uri}</CurrentURI>"
93 "<CurrentURIMetaData>"
94 f"{metadata}"
95 "</CurrentURIMetaData>"
96 )
97 return _get_xml(_get_body(command, arguments)), _get_soap_action(command)
98
99
100def get_xml_soap_remove_all_tracks() -> tuple[str, str]:
101 """Get UPnP xml and soap for RemoveAllTracksFromQueue."""
102 command = "RemoveAllTracksFromQueue"
103 return _get_xml(_get_body(command)), _get_soap_action(command)
104
105
106def get_xml_soap_set_next_url(player_media: PlayerMedia) -> tuple[str, str]:
107 """Get UPnP xml and soap for SetNextAVTransportURI."""
108 metadata = create_didl_metadata_str(player_media)
109 command = "SetNextAVTransportURI"
110 arguments = (
111 f"<NextURI>{player_media.uri}</NextURI><NextURIMetaData>{metadata}</NextURIMetaData>"
112 )
113 return _get_xml(_get_body(command, arguments)), _get_soap_action(command)
114
115
116# RemoveTrackFromQueue
117def get_xml_soap_remove_track(object_id: str) -> tuple[str, str]:
118 """Get UPnP xml and soap for RemoveTrackFromQueue."""
119 command = "RemoveTrackFromQueue"
120 arguments = f"<ObjectID>{object_id}</ObjectID>"
121 return _get_xml(_get_body(command, arguments)), _get_soap_action(command)
122
123
124# AddURIToQueue
125def get_xml_soap_add_uri_to_queue(player_media: PlayerMedia) -> tuple[str, str]:
126 """Get UPnP xml and soap for AddURIToQueue."""
127 metadata = create_didl_metadata_str(player_media)
128 command = "AddURIToQueue"
129 arguments = (
130 f"<EnqueuedURI>{player_media.uri}</EnqueuedURI>"
131 f"<EnqueuedURIMetaData>{metadata}</EnqueuedURIMetaData>"
132 "<DesiredFirstTrackNumberEnqueued>1</DesiredFirstTrackNumberEnqueued>"
133 "<EnqueueAsNext>0</EnqueueAsNext>"
134 )
135 return _get_xml(_get_body(command, arguments)), _get_soap_action(command)
136
137
138# CreateSavedQueue
139def get_xml_soap_create_saved_queue(queue_name: str, player_media: PlayerMedia) -> tuple[str, str]:
140 """Get UPnP xml and soap for CreateSavedQueue."""
141 command = "CreateSavedQueue"
142 metadata = create_didl_metadata_str(player_media)
143 arguments = (
144 f"<Title>{xmlescape(queue_name)}</Title>"
145 f"<EnqueuedURI>{player_media.uri}</EnqueuedURI>"
146 f"<EnqueuedURIMetaData>{metadata}</EnqueuedURIMetaData>"
147 )
148 return _get_xml(_get_body(command, arguments)), _get_soap_action(command)
149
150
151# CreateQueue
152def get_xml_soap_create_queue() -> tuple[str, str]:
153 """Get UPnP xml and soap for CreateQueue."""
154 command = "CreateQueue"
155 arguments = (
156 "<QueueOwnerID>mass</QueueOwnerID>"
157 "<QueueOwnerContext>mass</QueueOwnerContext>"
158 "<QueuePolicy>0</QueuePolicy>"
159 )
160 return _get_xml(_get_body(command, arguments, "Queue")), _get_soap_action(command)
161
162
163# DIDL-LITE
164def create_didl_metadata(media: PlayerMedia) -> str:
165 """Create DIDL metadata string from url and PlayerMedia."""
166
167 def escape_metadata(data: str) -> str:
168 """Escape didl metadata."""
169 data = xmlescape(data)
170 # Escape non-ascii to decimal code.
171 result = ""
172 for char in data:
173 unicode_code = ord(char)
174 if unicode_code < 128:
175 # ascii
176 result += char
177 else:
178 result += f"&#{unicode_code};"
179 return result
180
181 ext = media.uri.split(".")[-1].split("?")[0]
182 image_url = media.image_url or MASS_LOGO_ONLINE
183 if media.media_type in (MediaType.FLOW_STREAM, MediaType.RADIO) or not media.duration:
184 # flow stream, radio or other duration-less stream
185 # Use streaming-optimized DLNA flags to prevent buffering
186 title = media.title or media.uri
187 return (
188 '<DIDL-Lite xmlns:dc="http://purl.org/dc/elements/1.1/" xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/" xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/" xmlns:dlna="urn:schemas-dlna-org:metadata-1-0/">'
189 f'<item id="flowmode" parentID="0" restricted="1">'
190 f"<dc:title>{escape_metadata(title)}</dc:title>"
191 f"<upnp:albumArtURI>{escape_metadata(image_url)}</upnp:albumArtURI>"
192 f"<dc:queueItemId>{escape_metadata(media.uri)}</dc:queueItemId>"
193 f"<dc:description>Music Assistant</dc:description>"
194 "<upnp:class>object.item.audioItem.audioBroadcast</upnp:class>"
195 f'<res protocolInfo="http-get:*:audio/{ext}:DLNA.ORG_OP=01;DLNA.ORG_CI=0;DLNA.ORG_FLAGS=01700000000000000000000000000000">{escape_metadata(media.uri)}</res>'
196 "</item>"
197 "</DIDL-Lite>"
198 )
199
200 assert media.queue_item_id is not None # for type checking
201
202 # For regular tracks with duration, use flags optimized for on-demand content
203 # DLNA.ORG_FLAGS=01500000000000000000000000000000 indicates:
204 # - Streaming transfer mode (bit 24)
205 # - Background transfer mode supported (bit 22)
206 # - DLNA v1.5 (bit 20)
207 duration_str = str(int(media.duration or 0) // 3600).zfill(2) + ":"
208 duration_str += str((int(media.duration or 0) % 3600) // 60).zfill(2) + ":"
209 duration_str += str(int(media.duration or 0) % 60).zfill(2)
210
211 return (
212 '<DIDL-Lite xmlns:dc="http://purl.org/dc/elements/1.1/" xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/" xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/" xmlns:r="urn:schemas-rinconnetworks-com:metadata-1-0/">'
213 f'<item id="{media.queue_item_id or xmlescape(media.uri)}" restricted="true" parentID="{media.source_id or ""}">'
214 f"<dc:title>{escape_metadata(media.title or media.uri)}</dc:title>"
215 f"<dc:creator>{escape_metadata(media.artist or '')}</dc:creator>"
216 f"<upnp:album>{escape_metadata(media.album or '')}</upnp:album>"
217 f"<upnp:artist>{escape_metadata(media.artist or '')}</upnp:artist>"
218 f"<dc:queueItemId>{escape_metadata(media.queue_item_id)}</dc:queueItemId>"
219 f"<dc:description>Music Assistant</dc:description>"
220 f"<upnp:albumArtURI>{escape_metadata(image_url)}</upnp:albumArtURI>"
221 "<upnp:class>object.item.audioItem.musicTrack</upnp:class>"
222 f'<res duration="{duration_str}" protocolInfo="http-get:*:audio/{ext}:DLNA.ORG_OP=01;DLNA.ORG_CI=0;DLNA.ORG_FLAGS=01500000000000000000000000000000">{escape_metadata(media.uri)}</res>'
223 '<desc id="cdudn" nameSpace="urn:schemas-rinconnetworks-com:metadata-1-0/">RINCON_AssociatedZPUDN</desc>'
224 "</item>"
225 "</DIDL-Lite>"
226 )
227
228
229def create_didl_metadata_str(media: PlayerMedia) -> str:
230 """Create (xml-escaped) DIDL metadata string from url and PlayerMedia."""
231 return xmlescape(create_didl_metadata(media))
232