fix(mpris): read properties independent of the introspection table
This commit is contained in:
+1
-1
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "lrx-cli"
|
||||
version = "0.7.11"
|
||||
version = "0.7.12"
|
||||
description = "Fetch line-synced lyrics for your music player."
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.13"
|
||||
|
||||
@@ -61,7 +61,7 @@ MULTI_CANDIDATE_DELAY_S = 0.2 # delay between sequential lyric fetches
|
||||
LEGACY_CONFIDENCE = 50.0
|
||||
|
||||
# User-Agents
|
||||
UA_BROWSER = "Mozilla/5.0 (X11; Linux x86_64; rv:152.0) Gecko/20100101 Firefox/152.0"
|
||||
UA_BROWSER = "Mozilla/5.0 (Macintosh; Intel Mac OS X 15_8_0) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/27.0 Safari/605.1.15"
|
||||
UA_LRX = f"LRX-CLI {APP_VERSION} (https://github.com/Uyanide/lrx-cli)"
|
||||
|
||||
MUSIXMATCH_COOLDOWN_MS = 600_000 # 10 minutes
|
||||
|
||||
+41
-29
@@ -8,7 +8,7 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from dbus_next.aio.message_bus import MessageBus
|
||||
from dbus_next.constants import BusType
|
||||
from dbus_next.constants import BusType, MessageType
|
||||
from dbus_next.message import Message
|
||||
from loguru import logger
|
||||
from typing import Optional, List, Any
|
||||
@@ -38,17 +38,39 @@ async def _list_mpris_players(bus: MessageBus) -> List[str]:
|
||||
return []
|
||||
|
||||
|
||||
async def get_player_property(
|
||||
bus: MessageBus, player_name: str, prop: str
|
||||
) -> Optional[Any]:
|
||||
"""Read an org.mpris.MediaPlayer2.Player property with a raw Properties.Get.
|
||||
|
||||
Raw calls keep property access independent of the peer's introspection data.
|
||||
"""
|
||||
reply = await bus.call(
|
||||
Message(
|
||||
destination=player_name,
|
||||
path="/org/mpris/MediaPlayer2",
|
||||
interface="org.freedesktop.DBus.Properties",
|
||||
member="Get",
|
||||
signature="ss",
|
||||
body=["org.mpris.MediaPlayer2.Player", prop],
|
||||
)
|
||||
)
|
||||
if reply is None:
|
||||
return None
|
||||
if reply.message_type != MessageType.METHOD_RETURN:
|
||||
raise RuntimeError(
|
||||
f"Properties.Get({prop}) failed: "
|
||||
f"{reply.body[0] if reply.body else reply.error_name or 'no error body'}"
|
||||
)
|
||||
if not reply.body:
|
||||
return None
|
||||
return reply.body[0]
|
||||
|
||||
|
||||
async def _get_playback_status(bus: MessageBus, player_name: str) -> Optional[str]:
|
||||
"""Get PlaybackStatus ('Playing', 'Paused', 'Stopped') for a player."""
|
||||
try:
|
||||
introspection = await bus.introspect(player_name, "/org/mpris/MediaPlayer2")
|
||||
proxy = bus.get_proxy_object(
|
||||
player_name, "/org/mpris/MediaPlayer2", introspection
|
||||
)
|
||||
props = proxy.get_interface("org.freedesktop.DBus.Properties")
|
||||
status_var = await getattr(props, "call_get")(
|
||||
"org.mpris.MediaPlayer2.Player", "PlaybackStatus"
|
||||
)
|
||||
status_var = await get_player_property(bus, player_name, "PlaybackStatus")
|
||||
return status_var.value if status_var else None
|
||||
except Exception as e:
|
||||
logger.debug(f"Could not get playback status for {player_name}: {e}")
|
||||
@@ -128,13 +150,15 @@ async def _fetch_metadata_dbus(
|
||||
specific_player: Optional[str],
|
||||
preferred_player: str,
|
||||
player_blacklist: tuple[str, ...],
|
||||
bus: Optional[MessageBus] = None,
|
||||
) -> Optional[TrackMeta]:
|
||||
bus = None
|
||||
try:
|
||||
bus = await MessageBus(bus_type=BusType.SESSION).connect()
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to connect to DBus: {e}")
|
||||
return None
|
||||
owns_bus = bus is None
|
||||
if bus is None:
|
||||
try:
|
||||
bus = await MessageBus(bus_type=BusType.SESSION).connect()
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to connect to DBus: {e}")
|
||||
return None
|
||||
|
||||
try:
|
||||
player_name = await _select_player(
|
||||
@@ -148,20 +172,8 @@ async def _fetch_metadata_dbus(
|
||||
|
||||
logger.debug(f"Using player: {player_name}")
|
||||
|
||||
introspection = await bus.introspect(player_name, "/org/mpris/MediaPlayer2")
|
||||
proxy = bus.get_proxy_object(
|
||||
player_name, "/org/mpris/MediaPlayer2", introspection
|
||||
)
|
||||
|
||||
props_iface = proxy.get_interface("org.freedesktop.DBus.Properties")
|
||||
if not props_iface:
|
||||
logger.error(f"Player {player_name} doesn't support Properties interface.")
|
||||
return None
|
||||
|
||||
try:
|
||||
metadata_var: Any = await getattr(props_iface, "call_get")(
|
||||
"org.mpris.MediaPlayer2.Player", "Metadata"
|
||||
)
|
||||
metadata_var = await get_player_property(bus, player_name, "Metadata")
|
||||
if not metadata_var:
|
||||
logger.error("Empty metadata received.")
|
||||
return None
|
||||
@@ -215,7 +227,7 @@ async def _fetch_metadata_dbus(
|
||||
return None
|
||||
|
||||
finally:
|
||||
if bus:
|
||||
if owns_bus and bus:
|
||||
bus.disconnect()
|
||||
|
||||
|
||||
|
||||
@@ -17,7 +17,7 @@ from dbus_next.message import Message
|
||||
from loguru import logger
|
||||
|
||||
from ..models import TrackMeta
|
||||
from ..mpris import pick_active_player
|
||||
from ..mpris import get_player_property, pick_active_player
|
||||
|
||||
|
||||
def _variant_value(item: object) -> object | None:
|
||||
@@ -70,7 +70,6 @@ class PlayerMonitor:
|
||||
_target: PlayerTarget
|
||||
players: dict[str, PlayerState]
|
||||
_bus: MessageBus | None
|
||||
_props_cache: dict[str, object]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -88,7 +87,6 @@ class PlayerMonitor:
|
||||
self._target = target or PlayerTarget()
|
||||
self.players: dict[str, PlayerState] = {}
|
||||
self._bus: MessageBus | None = None
|
||||
self._props_cache: dict[str, object] = {}
|
||||
|
||||
async def start(self) -> None:
|
||||
"""Start DBus monitoring and populate initial player snapshot."""
|
||||
@@ -99,33 +97,10 @@ class PlayerMonitor:
|
||||
|
||||
async def close(self) -> None:
|
||||
"""Stop DBus monitoring and close bus connection."""
|
||||
self._props_cache.clear()
|
||||
if self._bus:
|
||||
self._bus.disconnect()
|
||||
self._bus = None
|
||||
|
||||
async def _get_player_props(self, bus_name: str) -> object | None:
|
||||
"""Return cached DBus Properties interface for player, creating it if missing."""
|
||||
if not self._bus:
|
||||
return None
|
||||
if bus_name in self._props_cache:
|
||||
return self._props_cache[bus_name]
|
||||
|
||||
try:
|
||||
introspection = await self._bus.introspect(
|
||||
bus_name, "/org/mpris/MediaPlayer2"
|
||||
)
|
||||
proxy = self._bus.get_proxy_object(
|
||||
bus_name, "/org/mpris/MediaPlayer2", introspection
|
||||
)
|
||||
props = proxy.get_interface("org.freedesktop.DBus.Properties")
|
||||
self._props_cache[bus_name] = props
|
||||
return props
|
||||
except Exception as e:
|
||||
logger.debug(f"Failed to prepare DBus props for {bus_name}: {e}")
|
||||
self._props_cache.pop(bus_name, None)
|
||||
return None
|
||||
|
||||
async def _add_match_rules(self) -> None:
|
||||
"""Register signal subscriptions needed by monitor."""
|
||||
if not self._bus:
|
||||
@@ -190,16 +165,13 @@ class PlayerMonitor:
|
||||
|
||||
async def _fetch_player_state(self, bus_name: str) -> Optional[PlayerState]:
|
||||
"""Read current playback status and metadata from one player service."""
|
||||
props = await self._get_player_props(bus_name)
|
||||
if props is None:
|
||||
if not self._bus:
|
||||
return None
|
||||
try:
|
||||
status_var = await getattr(props, "call_get")(
|
||||
"org.mpris.MediaPlayer2.Player", "PlaybackStatus"
|
||||
)
|
||||
metadata_var = await getattr(props, "call_get")(
|
||||
"org.mpris.MediaPlayer2.Player", "Metadata"
|
||||
status_var = await get_player_property(
|
||||
self._bus, bus_name, "PlaybackStatus"
|
||||
)
|
||||
metadata_var = await get_player_property(self._bus, bus_name, "Metadata")
|
||||
status = status_var.value if status_var else "Stopped"
|
||||
track = self._track_from_metadata(
|
||||
metadata_var.value if metadata_var else {}
|
||||
@@ -207,7 +179,6 @@ class PlayerMonitor:
|
||||
return PlayerState(bus_name=bus_name, status=status, track=track)
|
||||
except Exception as e:
|
||||
logger.debug(f"Failed to read state for {bus_name}: {e}")
|
||||
self._props_cache.pop(bus_name, None)
|
||||
return None
|
||||
|
||||
def _track_from_metadata(self, metadata: dict[str, object]) -> Optional[TrackMeta]:
|
||||
@@ -270,9 +241,6 @@ class PlayerMonitor:
|
||||
added = sorted(after - before)
|
||||
removed = sorted(before - after)
|
||||
|
||||
for bus_name in removed:
|
||||
self._props_cache.pop(bus_name, None)
|
||||
|
||||
self.players = updated
|
||||
|
||||
if added or removed:
|
||||
@@ -371,19 +339,15 @@ class PlayerMonitor:
|
||||
|
||||
async def get_position_ms(self, bus_name: str) -> Optional[int]:
|
||||
"""Read player-reported position in milliseconds."""
|
||||
props = await self._get_player_props(bus_name)
|
||||
if props is None:
|
||||
if not self._bus:
|
||||
return None
|
||||
try:
|
||||
position_var = await getattr(props, "call_get")(
|
||||
"org.mpris.MediaPlayer2.Player", "Position"
|
||||
)
|
||||
position_var = await get_player_property(self._bus, bus_name, "Position")
|
||||
if position_var is None:
|
||||
return None
|
||||
return max(0, int(position_var.value) // 1000)
|
||||
except Exception as e:
|
||||
logger.debug(f"Failed to read position from {bus_name}: {e}")
|
||||
self._props_cache.pop(bus_name, None)
|
||||
return None
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,158 @@
|
||||
"""Regression tests: MPRIS property reads must be issued as raw
|
||||
org.freedesktop.DBus.Properties.Get messages, independent of introspection.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
from typing import Callable, Optional
|
||||
|
||||
import pytest
|
||||
from dbus_next.constants import MessageType
|
||||
from dbus_next.message import Message
|
||||
from dbus_next.signature import Variant
|
||||
|
||||
from lrx_cli.models import TrackMeta
|
||||
from lrx_cli.mpris import _fetch_metadata_dbus, _get_playback_status
|
||||
from lrx_cli.watch.player import PlayerMonitor
|
||||
|
||||
PLAYER = "org.mpris.MediaPlayer2.generic1"
|
||||
|
||||
PLAYER_METADATA = {
|
||||
"mpris:trackid": Variant("o", "/org/example/Tracks/42"),
|
||||
"mpris:length": Variant("x", 239_761_000),
|
||||
"mpris:artUrl": Variant("s", "http://example.com/cover.jpg"),
|
||||
"xesam:album": Variant("s", "album"),
|
||||
"xesam:artist": Variant("as", ["artist"]),
|
||||
"xesam:title": Variant("s", "title"),
|
||||
"xesam:url": Variant("s", "http://example.com/stream"),
|
||||
}
|
||||
|
||||
|
||||
class FakeBus:
|
||||
"""Test double exposing only the low-level call/disconnect surface."""
|
||||
|
||||
def __init__(self, reply: Callable[[Message], Optional[Message]]) -> None:
|
||||
self._reply = reply
|
||||
self.calls: list[Message] = []
|
||||
self.disconnected = False
|
||||
|
||||
async def call(self, message: Message) -> Optional[Message]:
|
||||
self.calls.append(message)
|
||||
return self._reply(message)
|
||||
|
||||
def disconnect(self) -> None:
|
||||
self.disconnected = True
|
||||
|
||||
|
||||
def _return(signature: str, body: list) -> Message:
|
||||
return Message(
|
||||
message_type=MessageType.METHOD_RETURN,
|
||||
reply_serial=1,
|
||||
signature=signature,
|
||||
body=body,
|
||||
)
|
||||
|
||||
|
||||
def _names_reply(names: list[str]) -> Message:
|
||||
return _return("as", [names])
|
||||
|
||||
|
||||
def _metadata_reply() -> Message:
|
||||
return _return("v", [Variant("a{sv}", PLAYER_METADATA)])
|
||||
|
||||
|
||||
def _state_reply(msg: Message) -> Optional[Message]:
|
||||
"""Reply to ListNames and the PlaybackStatus/Metadata property reads."""
|
||||
if msg.member == "ListNames":
|
||||
return _names_reply([PLAYER])
|
||||
if msg.member == "Get":
|
||||
if msg.body[1] == "PlaybackStatus":
|
||||
return _return("v", [Variant("s", "Playing")])
|
||||
if msg.body[1] == "Metadata":
|
||||
return _metadata_reply()
|
||||
return None
|
||||
|
||||
|
||||
def _monitor(bus: FakeBus) -> PlayerMonitor:
|
||||
monitor = PlayerMonitor(
|
||||
on_players_changed=lambda: None,
|
||||
on_seeked=lambda bus_name, position_ms: None,
|
||||
on_playback_status=lambda bus_name, status: None,
|
||||
player_blacklist=(),
|
||||
)
|
||||
monitor._bus = bus # type: ignore[assignment]
|
||||
return monitor
|
||||
|
||||
|
||||
def test_get_playback_status_sends_raw_properties_get() -> None:
|
||||
def reply(msg: Message) -> Optional[Message]:
|
||||
if msg.member == "Get":
|
||||
return _return("v", [Variant("s", "Playing")])
|
||||
return None
|
||||
|
||||
bus = FakeBus(reply)
|
||||
status = asyncio.run(_get_playback_status(bus, PLAYER)) # type: ignore[arg-type]
|
||||
|
||||
assert status == "Playing"
|
||||
request = bus.calls[0]
|
||||
assert request.destination == PLAYER
|
||||
assert request.path == "/org/mpris/MediaPlayer2"
|
||||
assert request.interface == "org.freedesktop.DBus.Properties"
|
||||
assert request.member == "Get"
|
||||
assert request.body == ["org.mpris.MediaPlayer2.Player", "PlaybackStatus"]
|
||||
|
||||
|
||||
def test_fetch_metadata_dbus_reads_metadata_via_raw_properties_get() -> None:
|
||||
bus = FakeBus(_state_reply)
|
||||
track = asyncio.run(_fetch_metadata_dbus(None, "", (), bus=bus)) # type: ignore[arg-type]
|
||||
|
||||
# trackid without a Spotify prefix normalizes to None
|
||||
assert track == TrackMeta(
|
||||
trackid=None,
|
||||
length=239_761,
|
||||
album="album",
|
||||
artist="artist",
|
||||
title="title",
|
||||
url="http://example.com/stream",
|
||||
)
|
||||
assert not any(msg.member == "Introspect" for msg in bus.calls)
|
||||
# injected bus stays owned by the caller
|
||||
assert bus.disconnected is False
|
||||
|
||||
|
||||
def test_fetch_metadata_dbus_disconnects_self_created_bus(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
bus = FakeBus(_state_reply)
|
||||
|
||||
class FakeMessageBus:
|
||||
@staticmethod
|
||||
async def connect() -> FakeBus:
|
||||
return bus
|
||||
|
||||
monkeypatch.setattr("lrx_cli.mpris.MessageBus", lambda bus_type: FakeMessageBus)
|
||||
track = asyncio.run(_fetch_metadata_dbus(None, "", ()))
|
||||
|
||||
assert track is not None
|
||||
assert track.title == "title"
|
||||
assert bus.disconnected is True
|
||||
|
||||
|
||||
def test_player_monitor_refresh_reads_state_via_raw_properties_get() -> None:
|
||||
monitor = _monitor(FakeBus(_state_reply))
|
||||
asyncio.run(monitor.refresh())
|
||||
|
||||
state = monitor.players.get(PLAYER)
|
||||
assert state is not None
|
||||
assert state.status == "Playing"
|
||||
assert state.track is not None
|
||||
assert state.track.title == "title"
|
||||
|
||||
|
||||
def test_player_monitor_get_position_ms_reads_raw_position() -> None:
|
||||
def reply(msg: Message) -> Optional[Message]:
|
||||
if msg.member == "Get":
|
||||
return _return("v", [Variant("x", 231_008_000)])
|
||||
return None
|
||||
|
||||
monitor = _monitor(FakeBus(reply))
|
||||
assert asyncio.run(monitor.get_position_ms(PLAYER)) == 231_008
|
||||
Reference in New Issue
Block a user