music-assistant-server

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