/
/
/
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
11from music_assistant.helpers.audio import get_mime_type
12
13if TYPE_CHECKING:
14 from music_assistant.models.player import PlayerMedia
15
16
17# XML
18def _get_soap_action(command: str) -> str:
19 return f"urn:schemas-upnp-org:service:AVTransport:1#{command}"
20
21
22def _get_body(command: str, arguments: str = "", service: str = "AVTransport") -> str:
23 return (
24 f'<u:{command} xmlns:u="urn:schemas-upnp-org:service:{service}:1">'
25 r"<InstanceID>0</InstanceID>"
26 f"{arguments}"
27 f"</u:{command}>"
28 )
29
30
31def _get_xml(body: str) -> str:
32 return (
33 r'<?xml version="1.0"?>'
34 r'<s:Envelope s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" xmlns:s="http://schemas.xmlsoap.org/soap/envelope/">'
35 r"<s:Body>"
36 f"{body}"
37 r"</s:Body>"
38 r"</s:Envelope>"
39 )
40
41
42def get_xml_soap_play() -> tuple[str, str]:
43 """Get UPnP xml and soap for Play."""
44 command = "Play"
45 arguments = r"<Speed>1</Speed>"
46 return _get_xml(_get_body(command, arguments)), _get_soap_action(command)
47
48
49def get_xml_soap_stop() -> tuple[str, str]:
50 """Get UPnP xml and soap for Stop."""
51 command = "Stop"
52 return _get_xml(_get_body(command)), _get_soap_action(command)
53
54
55def get_xml_soap_pause() -> tuple[str, str]:
56 """Get UPnP xml and soap for Pause."""
57 command = "Pause"
58 return _get_xml(_get_body(command)), _get_soap_action(command)
59
60
61def get_xml_soap_next() -> tuple[str, str]:
62 """Get UPnP xml and soap for Next."""
63 command = "Next"
64 return _get_xml(_get_body(command)), _get_soap_action(command)
65
66
67def get_xml_soap_previous() -> tuple[str, str]:
68 """Get UPnP xml and soap for Previous."""
69 command = "Previous"
70 return _get_xml(_get_body(command)), _get_soap_action(command)
71
72
73def get_xml_soap_transport_info() -> tuple[str, str]:
74 """Get UPnP xml and soap for GetTransportInfo."""
75 command = "GetTransportInfo"
76 return _get_xml(_get_body(command)), _get_soap_action(command)
77
78
79def get_xml_soap_media_info() -> tuple[str, str]:
80 """Get UPnP xml and soap for GetMediaInfo."""
81 command = "GetMediaInfo"
82 return _get_xml(_get_body(command)), _get_soap_action(command)
83
84
85def get_xml_soap_set_url(player_media: PlayerMedia) -> tuple[str, str]:
86 """Get UPnP xml and soap for SetAVTransportURI."""
87 metadata = create_didl_metadata_str(player_media)
88 command = "SetAVTransportURI"
89 arguments = (
90 f"<CurrentURI>{player_media.uri}</CurrentURI>"
91 "<CurrentURIMetaData>"
92 f"{metadata}"
93 "</CurrentURIMetaData>"
94 )
95 return _get_xml(_get_body(command, arguments)), _get_soap_action(command)
96
97
98def get_xml_soap_remove_all_tracks() -> tuple[str, str]:
99 """Get UPnP xml and soap for RemoveAllTracksFromQueue."""
100 command = "RemoveAllTracksFromQueue"
101 return _get_xml(_get_body(command)), _get_soap_action(command)
102
103
104def get_xml_soap_set_next_url(player_media: PlayerMedia) -> tuple[str, str]:
105 """Get UPnP xml and soap for SetNextAVTransportURI."""
106 metadata = create_didl_metadata_str(player_media)
107 command = "SetNextAVTransportURI"
108 arguments = (
109 f"<NextURI>{player_media.uri}</NextURI><NextURIMetaData>{metadata}</NextURIMetaData>"
110 )
111 return _get_xml(_get_body(command, arguments)), _get_soap_action(command)
112
113
114# RemoveTrackFromQueue
115def get_xml_soap_remove_track(object_id: str) -> tuple[str, str]:
116 """Get UPnP xml and soap for RemoveTrackFromQueue."""
117 command = "RemoveTrackFromQueue"
118 arguments = f"<ObjectID>{object_id}</ObjectID>"
119 return _get_xml(_get_body(command, arguments)), _get_soap_action(command)
120
121
122# AddURIToQueue
123def get_xml_soap_add_uri_to_queue(player_media: PlayerMedia) -> tuple[str, str]:
124 """Get UPnP xml and soap for AddURIToQueue."""
125 metadata = create_didl_metadata_str(player_media)
126 command = "AddURIToQueue"
127 arguments = (
128 f"<EnqueuedURI>{player_media.uri}</EnqueuedURI>"
129 f"<EnqueuedURIMetaData>{metadata}</EnqueuedURIMetaData>"
130 "<DesiredFirstTrackNumberEnqueued>1</DesiredFirstTrackNumberEnqueued>"
131 "<EnqueueAsNext>0</EnqueueAsNext>"
132 )
133 return _get_xml(_get_body(command, arguments)), _get_soap_action(command)
134
135
136# CreateSavedQueue
137def get_xml_soap_create_saved_queue(queue_name: str, player_media: PlayerMedia) -> tuple[str, str]:
138 """Get UPnP xml and soap for CreateSavedQueue."""
139 command = "CreateSavedQueue"
140 metadata = create_didl_metadata_str(player_media)
141 arguments = (
142 f"<Title>{xmlescape(queue_name)}</Title>"
143 f"<EnqueuedURI>{player_media.uri}</EnqueuedURI>"
144 f"<EnqueuedURIMetaData>{metadata}</EnqueuedURIMetaData>"
145 )
146 return _get_xml(_get_body(command, arguments)), _get_soap_action(command)
147
148
149# CreateQueue
150def get_xml_soap_create_queue() -> tuple[str, str]:
151 """Get UPnP xml and soap for CreateQueue."""
152 command = "CreateQueue"
153 arguments = (
154 "<QueueOwnerID>mass</QueueOwnerID>"
155 "<QueueOwnerContext>mass</QueueOwnerContext>"
156 "<QueuePolicy>0</QueuePolicy>"
157 )
158 return _get_xml(_get_body(command, arguments, "Queue")), _get_soap_action(command)
159
160
161# DIDL-LITE
162def create_didl_metadata(media: PlayerMedia, url: str | None = None) -> str:
163 """Create DIDL metadata string from url and PlayerMedia."""
164 uri = url or media.uri
165
166 def escape_metadata(data: str) -> str:
167 """Escape didl metadata."""
168 data = xmlescape(data)
169 # Escape non-ascii to decimal code.
170 result = ""
171 for char in data:
172 unicode_code = ord(char)
173 if unicode_code < 128:
174 # ascii
175 result += char
176 else:
177 result += f"&#{unicode_code};"
178 return result
179
180 ext = uri.split(".")[-1].split("?")[0]
181 mime_type = get_mime_type(ext)
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 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(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:*:{mime_type}:DLNA.ORG_OP=01;DLNA.ORG_CI=0;DLNA.ORG_FLAGS=01700000000000000000000000000000">{escape_metadata(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 # the res duration describes the audio we hand over, which is shorter than
208 # the media item when playback starts at a seek position
209 stream_duration = int(media.stream_duration or media.duration or 0)
210 duration_str = str(stream_duration // 3600).zfill(2) + ":"
211 duration_str += str((stream_duration % 3600) // 60).zfill(2) + ":"
212 duration_str += str(stream_duration % 60).zfill(2)
213
214 return (
215 '<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/">'
216 f'<item id="{media.queue_item_id or xmlescape(uri)}" restricted="true" parentID="{media.source_id or ""}">'
217 f"<dc:title>{escape_metadata(media.title or uri)}</dc:title>"
218 f"<dc:creator>{escape_metadata(media.artist or '')}</dc:creator>"
219 f"<upnp:album>{escape_metadata(media.album or '')}</upnp:album>"
220 f"<upnp:artist>{escape_metadata(media.artist or '')}</upnp:artist>"
221 f"<dc:queueItemId>{escape_metadata(media.queue_item_id)}</dc:queueItemId>"
222 f"<dc:description>Music Assistant</dc:description>"
223 f"<upnp:albumArtURI>{escape_metadata(image_url)}</upnp:albumArtURI>"
224 "<upnp:class>object.item.audioItem.musicTrack</upnp:class>"
225 f'<res duration="{duration_str}" protocolInfo="http-get:*:{mime_type}:DLNA.ORG_OP=01;DLNA.ORG_CI=0;DLNA.ORG_FLAGS=01500000000000000000000000000000">{escape_metadata(uri)}</res>'
226 '<desc id="cdudn" nameSpace="urn:schemas-rinconnetworks-com:metadata-1-0/">RINCON_AssociatedZPUDN</desc>'
227 "</item>"
228 "</DIDL-Lite>"
229 )
230
231
232def create_didl_metadata_str(media: PlayerMedia) -> str:
233 """Create (xml-escaped) DIDL metadata string from url and PlayerMedia."""
234 return xmlescape(create_didl_metadata(media))
235