From 4d2488e646906fc419966bdc75ebdaf415da8560 Mon Sep 17 00:00:00 2001 From: MagicMicky Date: Sat, 4 Jul 2026 12:01:39 +0200 Subject: [PATCH] Add UPnP album art fallback Many LinkPlay devices (e.g. Arylic and Audio Pro models) do not expose album art through the HTTP API; the art is only available through the device's UPnP AVTransport service. When getPlayerStatusEx does not provide album art, fall back to UPnP GetMediaInfo and read the albumArtURI, keeping the value in the existing metainfo structure so album_art works unchanged. The fallback is manufacturer-agnostic: it triggers whenever album art is missing, which also helps WiiM devices that occasionally omit it. Non-URL placeholders (e.g. "un_known") are discarded. Relates to #68. UPnP as a full replacement for devices without getPlayerStatusEx (including transport state and volume) is left as a follow-up. Builds on the exploration in the draft #102. --- setup.cfg | 1 + src/linkplay/bridge.py | 27 +++++++- src/linkplay/endpoint.py | 11 +++ src/linkplay/upnp.py | 125 ++++++++++++++++++++++++++++++++++ tests/linkplay/test_bridge.py | 50 +++++++++++++- tests/linkplay/test_upnp.py | 102 +++++++++++++++++++++++++++ 6 files changed, 313 insertions(+), 3 deletions(-) create mode 100644 src/linkplay/upnp.py create mode 100644 tests/linkplay/test_upnp.py diff --git a/setup.cfg b/setup.cfg index 25aaecd..6252418 100644 --- a/setup.cfg +++ b/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 diff --git a/src/linkplay/bridge.py b/src/linkplay/bridge.py index 0f6e5f3..68752a9 100644 --- a/src/linkplay/bridge.py +++ b/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) diff --git a/src/linkplay/endpoint.py b/src/linkplay/endpoint.py index 8deb2c2..1a1d8e5 100644 --- a/src/linkplay/endpoint.py +++ b/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} diff --git a/src/linkplay/upnp.py b/src/linkplay/upnp.py new file mode 100644 index 0000000..79d0ccf --- /dev/null +++ b/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) diff --git a/tests/linkplay/test_bridge.py b/tests/linkplay/test_bridge.py index 100a8f1..95422c4 100644 --- a/tests/linkplay/test_bridge.py +++ b/tests/linkplay/test_bridge.py @@ -17,6 +17,8 @@ EqualizerMode, LinkPlayCommand, LoopMode, + MetaInfo, + MetaInfoMetaData, MuteMode, PlayerAttribute, PlayingMode, @@ -24,6 +26,7 @@ ) 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 == {} diff --git a/tests/linkplay/test_upnp.py b/tests/linkplay/test_upnp.py new file mode 100644 index 0000000..77def4b --- /dev/null +++ b/tests/linkplay/test_upnp.py @@ -0,0 +1,102 @@ +"""Test UPnP album art retrieval.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +from async_upnp_client.exceptions import UpnpError +from linkplay.upnp import LinkPlayUPnP, _is_url, _parse_metadata + +# A real DIDL-Lite document as returned by an Arylic/Audio Pro device playing +# Spotify (album art is a fully-qualified Spotify CDN url). +DIDL_WITH_ART = ( + '' + '' + "La thune de ma femme" + "Hippocampe Fou" + "Consumed In Key" + "https://i.scdn.co/image/ab67616d0000b273a7" + "" +) + +# What a follower / idle device reports: the placeholder "un_known". +DIDL_UNKNOWN_ART = ( + '' + '' + "un_known" +) + + +def test_parse_metadata_with_art(): + """A valid DIDL document yields all fields and keeps the art url.""" + metadata = _parse_metadata(DIDL_WITH_ART) + assert metadata.title == "La thune de ma femme" + assert metadata.artist == "Hippocampe Fou" + assert metadata.album == "Consumed In Key" + assert metadata.album_art_uri == "https://i.scdn.co/image/ab67616d0000b273a7" + + +def test_parse_metadata_discards_placeholder_art(): + """The 'un_known' placeholder is not a url and is discarded.""" + metadata = _parse_metadata(DIDL_UNKNOWN_ART) + assert metadata.album_art_uri is None + assert metadata.title is None + + +def test_parse_metadata_malformed(): + """Malformed XML yields empty metadata instead of raising.""" + metadata = _parse_metadata(" yields empty metadata.""" + metadata = _parse_metadata( + '' + ) + assert metadata.title is None + assert metadata.album_art_uri is None + + +def test_is_url(): + """Only absolute urls are accepted.""" + assert _is_url("https://i.scdn.co/image/abc") + assert not _is_url("un_known") + assert not _is_url("") + + +async def test_async_get_metadata_returns_parsed(): + """GetMediaInfo metadata is parsed into a LinkPlayUPnPMetadata.""" + upnp = LinkPlayUPnP("1.2.3.4", session=AsyncMock()) + service = MagicMock() + service.action.return_value.async_call = AsyncMock( + return_value={"CurrentURIMetaData": DIDL_WITH_ART} + ) + with patch.object( + LinkPlayUPnP, "_async_get_service", AsyncMock(return_value=service) + ): + metadata = await upnp.async_get_metadata() + + assert metadata is not None + assert metadata.album_art_uri == "https://i.scdn.co/image/ab67616d0000b273a7" + + +async def test_async_get_metadata_no_service(): + """No AVTransport service yields None.""" + upnp = LinkPlayUPnP("1.2.3.4", session=AsyncMock()) + with patch.object(LinkPlayUPnP, "_async_get_service", AsyncMock(return_value=None)): + assert await upnp.async_get_metadata() is None + + +async def test_async_get_metadata_action_error(): + """A UPnP error while calling GetMediaInfo yields None.""" + upnp = LinkPlayUPnP("1.2.3.4", session=AsyncMock()) + service = MagicMock() + service.action.return_value.async_call = AsyncMock(side_effect=UpnpError()) + with patch.object( + LinkPlayUPnP, "_async_get_service", AsyncMock(return_value=service) + ): + assert await upnp.async_get_metadata() is None