Add UPnP album art fallback

setup.cfg

@@ -16,6 +16,7 @@ install_requires =
     aiohttp>=3.8.5
     appdirs>=1.4.4
     async_upnp_client>=0.36.2
+    defusedxml>=0.7.1
     deprecated>=1.2.14
     aiofiles>=23.1.0
 python_requires = >=3.11

src/linkplay/bridge.py

@@ -24,9 +24,10 @@
     PlayingStatus,
     SpeakerType,
 )
-from linkplay.endpoint import LinkPlayEndpoint
+from linkplay.endpoint import LinkPlayApiEndpoint, LinkPlayEndpoint
 from linkplay.exceptions import LinkPlayInvalidDataException
 from linkplay.manufacturers import MANUFACTURER_WIIM, get_info_from_project
+from linkplay.upnp import LinkPlayUPnP
 from linkplay.utils import (
     equalizer_mode_from_number_mapping,
     equalizer_mode_to_number_mapping,
@@ -135,6 +136,7 @@ class LinkPlayPlayer:
     metainfo: dict[MetaInfo, dict[MetaInfoMetaData, str]]
 
     previous_playing_mode: PlayingMode | None = None
+    _upnp: LinkPlayUPnP | None = None
 
     def __init__(self, bridge: LinkPlayBridge):
         self.bridge = bridge
@@ -166,6 +168,12 @@ async def update_status(self) -> None:
         else:
             self.metainfo = {}
 
+        # Fall back to UPnP for album art when the HTTP API doesn't provide it
+        # (most non-WiiM devices, and some WiiM devices that occasionally omit
+        # it). The album art is injected into the existing metainfo structure.
+        if not self.album_art:
+            await self._update_album_art_from_upnp()
+
         # handle multiroom changes
         if self.bridge.device.controller is not None and (
             (
@@ -180,6 +188,23 @@ async def update_status(self) -> None:
             self.bridge.device.controller()
         self.previous_playing_mode = self.play_mode
 
+    async def _update_album_art_from_upnp(self) -> None:
+        """Fetch album art from the UPnP AVTransport service as a fallback."""
+        endpoint = self.bridge.endpoint
+        if not isinstance(endpoint, LinkPlayApiEndpoint):
+            return
+
+        if self._upnp is None:
+            self._upnp = LinkPlayUPnP(endpoint.host, endpoint.session)
+
+        metadata = await self._upnp.async_get_metadata()
+        if metadata is None or not metadata.album_art_uri:
+            return
+
+        self.metainfo.setdefault(MetaInfo.METADATA, {})[MetaInfoMetaData.ALBUM_ART] = (
+            metadata.album_art_uri
+        )
+
     async def next(self) -> None:
         """Play the next song in the playlist."""
         await self.bridge.request(LinkPlayCommand.NEXT)

src/linkplay/endpoint.py

@@ -41,10 +41,21 @@ def __init__(
             protocol == "https" and port != 443
         )
         port_suffix = f":{port}" if include_port else ""
+        self._host: str = endpoint
         self._endpoint: str = f"{protocol}://{endpoint}{port_suffix}"
 
         self._session: ClientSession = session
 
+    @property
+    def host(self) -> str:
+        """Return the host of the endpoint."""
+        return self._host
+
+    @property
+    def session(self) -> ClientSession:
+        """Return the aiohttp session of the endpoint."""
+        return self._session
+
     def to_dict(self):
         """Return the state of the LinkPlayEndpoint"""
         return {"endpoint": self._endpoint}

src/linkplay/upnp.py

@@ -0,0 +1,125 @@
+"""UPnP AVTransport helper for LinkPlay devices.
+
+Some LinkPlay devices (e.g. several Audio Pro and Arylic models) do not expose
+album art through the HTTP API, and a few do not expose ``getPlayerStatusEx``
+at all. The information is however available through the device's UPnP
+AVTransport service, which this module retrieves as a fallback.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+from urllib.parse import urlparse
+from xml.etree.ElementTree import Element, ParseError
+
+from aiohttp import ClientError, ClientSession
+from async_upnp_client.aiohttp import AiohttpSessionRequester
+from async_upnp_client.client import UpnpDevice, UpnpService
+from async_upnp_client.client_factory import UpnpFactory
+from async_upnp_client.exceptions import UpnpError
+from defusedxml.common import DefusedXmlException
+from defusedxml.ElementTree import fromstring
+
+from linkplay.consts import LOGGER
+
+UPNP_PORT: int = 49152
+UPNP_TIMEOUT: int = 2
+AVTRANSPORT_SERVICE: str = "urn:schemas-upnp-org:service:AVTransport:1"
+DESCRIPTION_URL: str = "http://{host}:{port}/description.xml"
+
+_DIDL_NAMESPACES: dict[str, str] = {
+    "didl": "urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/",
+    "dc": "http://purl.org/dc/elements/1.1/",
+    "upnp": "urn:schemas-upnp-org:metadata-1-0/upnp/",
+}
+
+
+@dataclass
+class LinkPlayUPnPMetadata:
+    """Media metadata retrieved from the UPnP AVTransport service."""
+
+    title: str | None = None
+    artist: str | None = None
+    album: str | None = None
+    album_art_uri: str | None = None
+
+
+class LinkPlayUPnP:
+    """Retrieves media information from a LinkPlay device over UPnP."""
+
+    def __init__(self, host: str, session: ClientSession):
+        self._host = host
+        self._factory = UpnpFactory(
+            AiohttpSessionRequester(session, timeout=UPNP_TIMEOUT)
+        )
+        self._device: UpnpDevice | None = None
+
+    async def _async_get_service(self) -> UpnpService | None:
+        """Return the AVTransport service, creating the device once."""
+        if self._device is None:
+            url = DESCRIPTION_URL.format(host=self._host, port=UPNP_PORT)
+            try:
+                self._device = await self._factory.async_create_device(url)
+            except (UpnpError, ClientError, asyncio.TimeoutError) as error:
+                LOGGER.debug("Failed to reach %s over UPnP: %s", self._host, error)
+                return None
+
+        if not self._device.has_service(AVTRANSPORT_SERVICE):
+            return None
+        return self._device.service(AVTRANSPORT_SERVICE)
+
+    async def async_get_metadata(self) -> LinkPlayUPnPMetadata | None:
+        """Return the current media metadata, or None if unavailable."""
+        service = await self._async_get_service()
+        if service is None:
+            return None
+
+        try:
+            result = await service.action("GetMediaInfo").async_call(InstanceID=0)
+        except (UpnpError, ClientError, asyncio.TimeoutError) as error:
+            LOGGER.debug("UPnP GetMediaInfo failed for %s: %s", self._host, error)
+            return None
+
+        metadata: str | None = result.get("CurrentURIMetaData")
+        if not metadata:
+            return None
+        return _parse_metadata(metadata)
+
+
+def _parse_metadata(didl: str) -> LinkPlayUPnPMetadata:
+    """Parse a DIDL-Lite metadata document into LinkPlayUPnPMetadata."""
+    try:
+        root = fromstring(didl)
+    except (ParseError, DefusedXmlException):
+        return LinkPlayUPnPMetadata()
+
+    item = root.find("didl:item", _DIDL_NAMESPACES)
+    if item is None:
+        return LinkPlayUPnPMetadata()
+
+    album_art_uri = _find_text(item, "upnp:albumArtURI")
+    if album_art_uri is not None and not _is_url(album_art_uri):
+        # Devices report placeholders such as "un_known" when no art is known.
+        album_art_uri = None
+
+    return LinkPlayUPnPMetadata(
+        title=_find_text(item, "dc:title"),
+        artist=_find_text(item, "upnp:artist"),
+        album=_find_text(item, "upnp:album"),
+        album_art_uri=album_art_uri,
+    )
+
+
+def _find_text(item: Element, path: str) -> str | None:
+    """Return the stripped text of a child element, or None when empty."""
+    element = item.find(path, _DIDL_NAMESPACES)
+    if element is None or element.text is None:
+        return None
+    return element.text.strip() or None
+
+
+def _is_url(value: str) -> bool:
+    """Return whether value is an absolute URL."""
+    parsed = urlparse(value)
+    return bool(parsed.scheme and parsed.netloc)

tests/linkplay/test_bridge.py

@@ -17,13 +17,16 @@
     EqualizerMode,
     LinkPlayCommand,
     LoopMode,
+    MetaInfo,
+    MetaInfoMetaData,
     MuteMode,
     PlayerAttribute,
     PlayingMode,
     PlayingStatus,
 )
 from linkplay.endpoint import LinkPlayApiEndpoint
 from linkplay.manufacturers import MANUFACTURER_WIIM
+from linkplay.upnp import LinkPlayUPnPMetadata
 
 
 def test_device_name():
@@ -50,6 +53,44 @@ async def test_device_update_status():
     assert device.uuid == "1234"
 
 
+async def test_player_update_status_album_art_upnp_fallback():
+    """A device without album art over HTTP falls back to UPnP."""
+    bridge = AsyncMock()
+    bridge.json_request.return_value = {PlayerAttribute.PLAYING_STATUS: "play"}
+    bridge.device.manufacturer = "Arylic"
+    bridge.device.controller = None
+    bridge.endpoint = LinkPlayApiEndpoint(
+        protocol="http", port=80, endpoint="1.2.3.4", session=AsyncMock()
+    )
+    player = LinkPlayPlayer(bridge)
+
+    metadata = LinkPlayUPnPMetadata(album_art_uri="https://art.example/cover.jpg")
+    with patch("linkplay.bridge.LinkPlayUPnP") as upnp_cls:
+        upnp_cls.return_value.async_get_metadata = AsyncMock(return_value=metadata)
+        await player.update_status()
+
+    upnp_cls.assert_called_once()
+    assert player.album_art == "https://art.example/cover.jpg"
+
+
+async def test_player_update_status_no_upnp_when_art_present():
+    """UPnP is not queried when the HTTP API already provides album art."""
+    bridge = AsyncMock()
+    bridge.json_request.side_effect = [
+        {PlayerAttribute.PLAYING_STATUS: "play"},
+        {MetaInfo.METADATA: {MetaInfoMetaData.ALBUM_ART: "http://art/from-wiim.jpg"}},
+    ]
+    bridge.device.manufacturer = MANUFACTURER_WIIM
+    bridge.device.controller = None
+    player = LinkPlayPlayer(bridge)
+
+    with patch("linkplay.bridge.LinkPlayUPnP") as upnp_cls:
+        await player.update_status()
+
+    upnp_cls.assert_not_called()
+    assert player.album_art == "http://art/from-wiim.jpg"
+
+
 async def test_device_reboot():
     """Tests if the device update is correctly called."""
     bridge = AsyncMock()
@@ -486,6 +527,8 @@ async def test_multiroom_set_volume_raises_value_error(volume: int):
 def mock_bridge():
     bridge = Mock(spec=LinkPlayBridge)
     bridge.device = Mock(spec=LinkPlayDevice)
+    # A non-LinkPlayApiEndpoint keeps update_status from attempting a UPnP call.
+    bridge.endpoint = Mock()
     return bridge
 
 
@@ -592,8 +635,11 @@ async def mock_session_call_api_json_side_effect(endpoint, session, command):
         # Create a LinkPlayPlayer instance with the mocked bridge
         player = LinkPlayPlayer(mock_bridge)
 
-        # Simulate the META_INFO request and exception handling
-        await player.update_status()
+        # Simulate the META_INFO request and exception handling. Patch UPnP as
+        # the empty album art would otherwise trigger the UPnP fallback.
+        with patch("linkplay.bridge.LinkPlayUPnP") as mock_upnp:
+            mock_upnp.return_value.async_get_metadata = AsyncMock(return_value=None)
+            await player.update_status()
 
         # Verify that metainfo is set to an empty dictionary after the exception
         assert player.metainfo == {}