/
/
/
1"""DLNA Player."""
2
3import asyncio
4import functools
5import time
6from collections.abc import Awaitable, Callable, Coroutine, Sequence
7from contextlib import suppress
8from typing import TYPE_CHECKING, Any, Concatenate
9from urllib.parse import urlparse
10from xml.etree.ElementTree import ParseError
11
12import defusedxml.ElementTree as DefusedET
13from async_upnp_client.exceptions import UpnpError, UpnpResponseError
14from async_upnp_client.profiles.dlna import DmrDevice, TransportState
15from music_assistant_models.enums import IdentifierType, PlaybackState, PlayerFeature, PlayerType
16from music_assistant_models.errors import PlayerUnavailableError
17
18from music_assistant.constants import VERBOSE_LOG_LEVEL
19from music_assistant.helpers.upnp import create_didl_metadata
20from music_assistant.models.player import DeviceInfo, Player
21
22from .constants import PLAYER_CONFIG_ENTRIES
23
24if TYPE_CHECKING:
25 from async_upnp_client.client import UpnpDevice, UpnpService, UpnpStateVariable
26 from music_assistant_models.config_entries import ConfigEntry
27 from music_assistant_models.player import PlayerMedia
28
29 from .provider import DLNAPlayerProvider
30
31
32def catch_request_errors[DLNAPlayerT: "DLNAPlayer", **P, R](
33 func: Callable[Concatenate[DLNAPlayerT, P], Awaitable[R]],
34) -> Callable[Concatenate[DLNAPlayerT, P], Coroutine[Any, Any, R | None]]:
35 """Catch UpnpError errors."""
36
37 @functools.wraps(func)
38 async def wrapper(self: DLNAPlayerT, *args: P.args, **kwargs: P.kwargs) -> R | None:
39 """Catch UpnpError errors and check availability before and after request."""
40 self.last_command = time.time()
41 if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
42 self.logger.debug(
43 "Handling command %s for player %s",
44 func.__name__,
45 self.display_name,
46 )
47 if not self.available and func.__name__ not in ("pause", "stop"):
48 self.logger.warning("Device disappeared when trying to call %s", func.__name__)
49 return None
50 try:
51 return await func(self, *args, **kwargs)
52 except UpnpError as err:
53 self.force_poll = True
54 if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
55 self.logger.exception("Error during call %s", func.__name__)
56 else:
57 self.logger.error("Error during call %s: %r", func.__name__, str(err))
58 return None
59
60 return wrapper
61
62
63class DLNAPlayer(Player):
64 """
65 DLNA Player.
66
67 All DLNA players are considered generic protocol endpoints (PlayerType.PROTOCOL)
68 and will be wrapped in a UniversalPlayer. Devices with native provider support
69 (e.g., Sonos) are handled by their respective providers and will link to
70 the DLNA player as a protocol output.
71 """
72
73 # All DLNA devices are generic protocol endpoints - no vendor has native DLNA support in MA
74 _attr_type = PlayerType.PROTOCOL
75
76 def __init__(
77 self,
78 provider: DLNAPlayerProvider,
79 player_id: str,
80 description_url: str,
81 device: DmrDevice | None = None,
82 ) -> None:
83 """
84 Init Player.
85
86 The player_id is the udn.
87 """
88 super().__init__(provider, player_id)
89
90 self.device = device
91 self.description_url = description_url # last known location (description.xml) url
92
93 self.lock = asyncio.Lock() # Held when connecting or disconnecting the device
94
95 self.force_poll = False # used, if connection is lost
96
97 self._observed_playback_state: PlaybackState | None = None
98 self._playing_since: float | None = None
99
100 # ssdp_connect_failed: bool = False
101 #
102 # Track BOOTID in SSDP advertisements for device changes
103 self.bootid: int | None = None
104 self.last_seen = time.time()
105 self.last_command = time.time()
106
107 def set_available(self, available: bool) -> None:
108 """Set the availability of the player."""
109 self._attr_available = available
110
111 async def setup(self) -> bool:
112 """
113 Set up player in MA.
114
115 :return: True if setup was successful, False if device should be ignored.
116 """
117 await self._device_connect()
118
119 if self.device and not self.device.has_play_media:
120 self.logger.debug("Ignoring %s - no play capability", self.device.name)
121 return False
122
123 if self.device and await self._is_sonos_passive_speaker():
124 self.logger.debug("Ignoring %s - passive stereo pair speaker", self.device.name)
125 return False
126
127 self.set_static_attributes()
128 await self.mass.players.register_or_update(self)
129 return True
130
131 def set_static_attributes(self) -> None:
132 """Set static attributes."""
133 self._attr_needs_poll = True
134 self._attr_poll_interval = 30
135 self._set_player_features()
136
137 async def set_dynamic_attributes(self) -> None:
138 """Set dynamic attributes."""
139 available = self.device is not None and self.device.profile_device.available
140 self._attr_available = available
141 if not available:
142 return
143 assert self.device is not None # for type checking
144 self._attr_name = self.device.name
145 # a device reports an unknown volume as None (RenderingControl Volume/Mute unset
146 # or not reported yet), which must stay unknown instead of collapsing to 0/unmuted
147 volume_level = self.device.volume_level
148 self._attr_volume_level = int(volume_level * 100) if volume_level is not None else None
149 self._attr_volume_muted = self.device.is_volume_muted
150 _playback_state = self._get_playback_state()
151 assert _playback_state is not None # for type checking
152 prev_playback_state = self._observed_playback_state
153 self._observed_playback_state = _playback_state
154 if _playback_state != PlaybackState.PLAYING:
155 self._playing_since = None
156 elif prev_playback_state not in (None, PlaybackState.PLAYING):
157 self._playing_since = time.time()
158 self._attr_playback_state = _playback_state
159
160 _device_uri = self.device.current_track_uri or ""
161 try:
162 media_title = self.device.media_title
163 media_artist = self.device.media_artist
164 media_album = self.device.media_album_name
165 media_image_url = self.device.media_image_url
166 media_duration = self.device.media_duration
167 except ParseError as err:
168 # some devices (e.g. Bose SoundTouch) return malformed DIDL-Lite
169 # metadata XML - drop the metadata but keep the player updated
170 self.logger.debug(
171 "Ignoring malformed media metadata from device %s: %s", self.display_name, err
172 )
173 media_title = media_artist = media_album = media_image_url = None
174 media_duration = None
175 self.set_current_media(
176 uri=_device_uri,
177 clear_all=True,
178 title=media_title,
179 artist=media_artist,
180 album=media_album,
181 image_url=media_image_url,
182 duration=int(media_duration) if media_duration is not None else None,
183 )
184
185 # Let player controller determine active source, only override for known external sources
186 if _device_uri and _device_uri.startswith(self.mass.streams.base_url):
187 # MA stream - let controller determine source
188 self._attr_active_source = None
189 elif "spotify" in _device_uri:
190 # Spotify or Spotify Connect
191 self._attr_active_source = "spotify"
192 elif _device_uri:
193 # External HTTP source
194 self._attr_active_source = "http"
195 else:
196 # No URI - idle or unknown
197 self._attr_active_source = None
198 # TODO: extend this list with other possible sources
199 # a device reports 'no position' as None (RelativeTimePosition unset, sent as
200 # NOT_IMPLEMENTED or unparsable), so a reported 0 is a position like any other
201 # and is adopted instead of being discarded as 'not reported'.
202 if (media_position := self.device.media_position) is not None:
203 self._attr_elapsed_time = float(media_position)
204 if (position_updated_at := self.device.media_position_updated_at) is not None:
205 anchor = position_updated_at.timestamp()
206 if self._playing_since is not None:
207 # a device only re-stamps a position that actually changed, so shortly
208 # after a resume the timestamp still dates from before the pause. The
209 # position may not be extrapolated across the time it was not playing.
210 anchor = max(anchor, self._playing_since)
211 self._attr_elapsed_time_last_updated = anchor
212
213 async def get_config_entries(self) -> list[ConfigEntry]:
214 """Return all (provider/player specific) Config Entries for the given player (if any)."""
215 return [*PLAYER_CONFIG_ENTRIES]
216
217 # COMMANDS
218 @catch_request_errors
219 async def stop(self) -> None:
220 """Send STOP command to given player."""
221 assert self.device is not None # for type checking
222 if self.device.can_stop:
223 await self.device.async_stop()
224 return
225 # Some devices report stale/empty CurrentTransportActions while still
226 # accepting AVTransport Stop. Force-call Stop when action exists.
227 action = self.device._action("AVT", "Stop")
228 if action is not None:
229 await action.async_call(InstanceID=0)
230
231 @catch_request_errors
232 async def play(self) -> None:
233 """Send PLAY command to given player."""
234 assert self.device is not None # for type checking
235 await self.device.async_play()
236
237 @catch_request_errors
238 async def play_media(self, media: PlayerMedia) -> None:
239 """Handle PLAY MEDIA on given player."""
240 assert self.device is not None # for type checking
241 # always clear queue (by sending stop) first
242 if self.device.can_stop:
243 await self.stop()
244 url = await self.provider.mass.streams.resolve_stream_url(self.player_id, media)
245 didl_metadata = create_didl_metadata(media, url)
246 title = media.title or media.uri
247 # optimistically set the state here to help in case of a player
248 # that is slow or failing to report state changes.
249 prev_state = self._attr_playback_state
250 self.set_current_media(uri=url, clear_all=True)
251 self._attr_playback_state = PlaybackState.PLAYING
252 self._attr_elapsed_time = 0
253 self._attr_elapsed_time_last_updated = time.time()
254 try:
255 await self.device.async_set_transport_uri(url, title, didl_metadata)
256 await self.device.async_wait_for_can_play(10)
257 await self.device.async_play()
258 except Exception:
259 self._attr_playback_state = prev_state
260 raise
261 self.update_state()
262
263 @catch_request_errors
264 async def enqueue_next_media(self, media: PlayerMedia) -> None:
265 """Handle enqueuing of the next queue item on the player."""
266 assert self.device is not None # for type checking
267 url = await self.provider.mass.streams.resolve_stream_url(self.player_id, media)
268 didl_metadata = create_didl_metadata(media, url)
269 title = media.title or media.uri
270 try:
271 await self.device.async_set_next_transport_uri(url, title, didl_metadata)
272 except UpnpError:
273 self.logger.error(
274 "Enqueuing the next track failed for player %s - "
275 "the player probably doesn't support this. "
276 "Enable 'flow mode' for this player.",
277 self.display_name,
278 )
279
280 @catch_request_errors
281 async def pause(self) -> None:
282 """Send PAUSE command to given player."""
283 assert self.device is not None # for type checking
284
285 replace_pause_with_stop = self.get_config_value("replace_pause_with_stop", return_type=bool)
286
287 if replace_pause_with_stop and self.device.can_stop:
288 await self.stop()
289 return
290
291 if self.device.can_pause:
292 await self.device.async_pause()
293 return
294
295 # Some devices expose Pause but report stale CurrentTransportActions.
296 # Force-call Pause when action exists; otherwise fallback to Stop.
297 pause_action = self.device._action("AVT", "Pause")
298 if pause_action is not None:
299 await pause_action.async_call(InstanceID=0)
300 return
301 stop_action = self.device._action("AVT", "Stop")
302 if stop_action is not None:
303 await stop_action.async_call(InstanceID=0)
304
305 @catch_request_errors
306 async def volume_set(self, volume_level: int) -> None:
307 """Send VOLUME_SET command to given player."""
308 assert self.device is not None # for type checking
309 await self.device.async_set_volume_level(volume_level / 100)
310 self.mass.call_later(
311 0.25,
312 self._poll_volume_state,
313 task_id=f"dlna_poll_volume_{self.player_id}",
314 )
315
316 @catch_request_errors
317 async def volume_mute(self, muted: bool) -> None:
318 """Send VOLUME MUTE command to given player."""
319 assert self.device is not None # for type checking
320 await self.device.async_mute_volume(muted)
321 await self._poll_volume_state()
322
323 async def poll(self) -> None:
324 """Poll player for state updates."""
325 # try to reconnect the device if the connection was lost
326 if not self.device:
327 if not self.force_poll:
328 return
329 try:
330 await self._device_connect()
331 except UpnpError as err:
332 raise PlayerUnavailableError from err
333
334 assert self.device is not None
335
336 try:
337 now = time.time()
338 do_ping = self.force_poll or (now - self.last_seen) > 60
339 with suppress(ValueError, ParseError):
340 await self.device.async_update(do_ping=do_ping)
341 self.last_seen = now if do_ping else self.last_seen
342 except UpnpError as err:
343 # Some devices (e.g. Denon HEOS) return SOAP responses containing
344 # non-UTF-8 bytes in track metadata, which the underlying library
345 # surfaces as a UpnpCommunicationError wrapping UnicodeDecodeError.
346 # Treat this as a transient metadata issue and keep the player
347 # connected; the next poll will likely succeed.
348 if isinstance(err.__cause__, UnicodeDecodeError):
349 self.logger.debug("Ignoring non-UTF-8 SOAP response from device: %r", err)
350 return
351 self.logger.debug("Device unavailable: %r", err)
352 await self._device_disconnect()
353 raise PlayerUnavailableError from err
354 finally:
355 self.force_poll = False
356
357 async def on_unload(self) -> None:
358 """Handle logic when the player is unloaded from the Player controller."""
359 await super().on_unload()
360 await self._device_disconnect()
361
362 async def _device_connect(self) -> None:
363 """Connect DLNA/DMR Device."""
364 self.logger.debug("Connecting to device at %s", self.description_url)
365
366 async with self.lock:
367 if self.device:
368 self.logger.debug("Trying to connect when device already connected")
369 return
370
371 # Connect to the base UPNP device
372 if TYPE_CHECKING:
373 assert isinstance(self.provider, DLNAPlayerProvider) # for type checking
374 upnp_device = await self.provider.upnp_factory.async_create_device(self.description_url)
375
376 # Create profile wrapper
377 self.device = DmrDevice(upnp_device, self.provider.notify_server.event_handler)
378
379 # Subscribe to event notifications
380 try:
381 self.device.on_event = self._handle_event
382 await self.device.async_subscribe_services(auto_resubscribe=True)
383 except UpnpResponseError as err:
384 # Device rejected subscription request. This is OK, variables
385 # will be polled instead.
386 self.logger.debug("Device rejected subscription: %r", err)
387 except UpnpError as err:
388 # Don't leave the device half-constructed
389 self.device.on_event = None
390 self.device = None
391 self.logger.debug("Error while subscribing during device connect: %r", err)
392 raise
393 else:
394 # connect was successful, update device info
395 self._attr_device_info = DeviceInfo(
396 model=self.device.model_name,
397 manufacturer=self.device.manufacturer,
398 )
399 # Add UDN (player_id) as UUID identifier for matching with other protocols
400 # Strip the "uuid:" prefix if present for proper matching
401 uuid_value = self.player_id
402 if uuid_value.lower().startswith("uuid:"):
403 uuid_value = uuid_value[5:]
404 self._attr_device_info.add_identifier(IdentifierType.UUID, uuid_value)
405 # MAC address is NOT extracted from UUID because the last 12 chars
406 # of UPnP UUIDs are unreliable — many devices put random/model
407 # values there, not the real hardware MAC. Instead, the player
408 # controller resolves the real MAC via ARP during registration
409 # using the IP address extracted below.
410 # Try to extract just the IP from the URL for matching
411 ip_address = self.device.device.presentation_url or self.description_url
412 with suppress(ValueError):
413 parsed = urlparse(ip_address)
414 if parsed.hostname:
415 self._attr_device_info.add_identifier(
416 IdentifierType.IP_ADDRESS, parsed.hostname
417 )
418
419 def _handle_event(
420 self,
421 service: UpnpService,
422 state_variables: Sequence[UpnpStateVariable[Any]],
423 ) -> None:
424 """Handle state variable(s) changed event from DLNA device."""
425 if not state_variables:
426 # Indicates a failure to resubscribe, check if device is still available
427 self.force_poll = True
428 return
429 poll_first = False
430 if service.service_id == "urn:upnp-org:serviceId:AVTransport":
431 for state_variable in state_variables:
432 # Force a state refresh when player begins or pauses playback
433 # to update the position info.
434 if state_variable.name == "TransportState" and state_variable.value in (
435 TransportState.PLAYING,
436 TransportState.PAUSED_PLAYBACK,
437 ):
438 self.force_poll = True
439 poll_first = True
440 self.logger.log(
441 VERBOSE_LOG_LEVEL,
442 "Received new state from event for Player %s: %s",
443 self.display_name,
444 state_variable.value,
445 )
446 self.last_seen = time.time()
447 self.mass.create_task(self._update_player(poll_first=poll_first))
448
449 async def _update_player(self, poll_first: bool = False) -> None:
450 """
451 Update DLNA Player.
452
453 :param poll_first: Refresh the device state before reading it, so that the
454 position info belongs to the state that is about to be reported.
455 """
456 if poll_first:
457 # an unavailable device is reported as such by the state update below
458 with suppress(PlayerUnavailableError):
459 await self.poll()
460 prev_url = self._attr_current_media.uri if self._attr_current_media is not None else ""
461 prev_state = self.state
462 await self.set_dynamic_attributes()
463 current_url = self._attr_current_media.uri if self._attr_current_media is not None else ""
464 current_state = self.state
465
466 if (prev_url != current_url) or (prev_state != current_state):
467 # fetch track details on state or url change
468 self.force_poll = True
469
470 try:
471 self.update_state()
472 except KeyError, TypeError:
473 # at start the update might come faster than the config is initialized
474 await asyncio.sleep(2)
475 self.update_state()
476
477 def _set_player_features(self) -> None:
478 """Set Player Features based on config values and capabilities."""
479 assert self.device is not None # for type checking
480 supported_features: set[PlayerFeature] = set()
481
482 # Only add PLAY_MEDIA if the device actually supports playback
483 # Passive speakers (like stereo pair satellites) don't have play capability
484 if self.device.has_play_media:
485 supported_features.add(PlayerFeature.PLAY_MEDIA)
486 # there is no way to check if a dlna player support enqueuing
487 # so we simply assume it does and if it doesn't
488 # you'll find out at playback time and we log a warning
489 supported_features.add(PlayerFeature.ENQUEUE)
490 supported_features.add(PlayerFeature.GAPLESS_PLAYBACK)
491
492 if self.device.has_volume_level:
493 supported_features.add(PlayerFeature.VOLUME_SET)
494 if self.device.has_volume_mute:
495 supported_features.add(PlayerFeature.VOLUME_MUTE)
496 if self.device.has_pause:
497 supported_features.add(PlayerFeature.PAUSE)
498 self._attr_supported_features = supported_features
499
500 async def _is_sonos_passive_speaker(self) -> bool:
501 """
502 Check if this is a Sonos passive stereo pair speaker.
503
504 Queries the device's own topology. If that returns 403, the device is
505 considered passive (passive satellites and speakers with UPnP disabled
506 block topology queries). If successful, checks for Invisible="1" attribute.
507 """
508 if not self.device:
509 return False
510
511 manufacturer = (self.device.manufacturer or "").lower()
512 if "sonos" not in manufacturer:
513 return False
514
515 # Extract base UUID (strip "uuid:" prefix and "_MR" suffix)
516 our_uuid = self.player_id.removeprefix("uuid:").removesuffix("_MR")
517
518 # Query this device's topology
519 upnp_device = self.device.profile_device.root_device
520 result = await self._check_invisible_in_topology(upnp_device, our_uuid)
521
522 # Return the result: True if passive/403, False if active or check failed
523 return result if result is not None else False
524
525 async def _check_invisible_in_topology(
526 self, upnp_device: UpnpDevice, our_uuid: str
527 ) -> bool | None:
528 """
529 Check if our UUID is marked as Invisible in the topology.
530
531 :param upnp_device: UPnP device to query
532 :param our_uuid: Our device UUID to search for
533 :return: True if invisible/403 error, False if visible, None if check failed
534 """
535 zone_topology_service = None
536 for service in upnp_device.all_services:
537 if "ZoneGroupTopology" in service.service_type:
538 zone_topology_service = service
539 break
540
541 if not zone_topology_service:
542 return None
543
544 try:
545 action = zone_topology_service.action("GetZoneGroupState")
546 if not action:
547 return None
548
549 result = await action.async_call()
550 zone_group_state_xml = result.get("ZoneGroupState", "")
551 if not zone_group_state_xml:
552 return None
553
554 root = DefusedET.fromstring(zone_group_state_xml)
555 for member in root.iter("ZoneGroupMember"):
556 if member.get("UUID", "").upper() == our_uuid.upper():
557 return str(member.get("Invisible", "0")) == "1"
558
559 except UpnpResponseError as err:
560 # 403 Forbidden indicates passive satellite (blocks topology queries)
561 if "403" in str(err):
562 self.logger.debug(
563 "Sonos device %s returned 403 - treating as passive satellite",
564 our_uuid,
565 )
566 return True
567 self.logger.log(
568 VERBOSE_LOG_LEVEL,
569 "Error checking Sonos zone topology: %s",
570 err,
571 )
572 except (UpnpError, DefusedET.ParseError) as err:
573 self.logger.log(
574 VERBOSE_LOG_LEVEL,
575 "Error checking Sonos zone topology: %s",
576 err,
577 )
578
579 return None
580
581 def _get_playback_state(self) -> PlaybackState | None:
582 """Return current PlaybackState of the player."""
583 if self.device is None:
584 return None
585 if self.device.transport_state is None:
586 return PlaybackState.IDLE
587 if self.device.transport_state in (
588 TransportState.PLAYING,
589 TransportState.TRANSITIONING,
590 ):
591 return PlaybackState.PLAYING
592 if self.device.transport_state in (
593 TransportState.PAUSED_PLAYBACK,
594 TransportState.PAUSED_RECORDING,
595 ):
596 return PlaybackState.PAUSED
597 if self.device.transport_state == TransportState.VENDOR_DEFINED:
598 # Unable to map this state to anything reasonable, fallback to idle
599 return PlaybackState.IDLE
600
601 return PlaybackState.IDLE
602
603 async def _poll_volume_state(self) -> None:
604 """
605 Poll the device for current volume/mute state and update player.
606
607 Some DLNA devices don't send RenderingControl events for
608 volume/mute changes initiated via UPnP actions, and the library
609 skips polling RC state variables when subscribed to events.
610 This forces a targeted poll to keep state in sync.
611 """
612 if not self.device:
613 return
614 actions: list[str] = []
615 if self.device.has_volume_level:
616 actions.append("GetVolume")
617 if self.device.has_volume_mute:
618 actions.append("GetMute")
619 if not actions:
620 return
621 await self.device._async_poll_state_variables("RC", actions, InstanceID=0, Channel="Master")
622 await self._update_player()
623
624 async def _device_disconnect(self) -> None:
625 """Destroy connections to the device."""
626 async with self.lock:
627 if not self.device:
628 self.logger.debug("Disconnecting from device that's not connected")
629 return
630
631 self.logger.debug("Disconnecting from %s", self.device.name)
632
633 self.device.on_event = None
634 old_device = self.device
635 self.device = None
636 self.set_available(False)
637 await old_device.async_unsubscribe_services()
638 self.update_state()
639