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