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