lintunes/lintunes/mpris.py

"""MPRIS2 D-Bus interface so desktop media keys / playerctl control lintunes.

Registers org.mpris.MediaPlayer2.lintunes on the session bus with the
standard root and Player interfaces, backed by our Player object. Album art
is exported to a small cache dir so GNOME's media popup can show it
(mpris:artUrl must be a file the notification daemon can read).
"""
import os
from pathlib import Path

from PyQt6.QtCore import QMetaType, QObject, pyqtSlot, pyqtProperty, pyqtClassInfo
from PyQt6.QtDBus import (
    QDBusConnection, QDBusAbstractAdaptor, QDBusArgument, QDBusObjectPath,
    QDBusMessage,
)

from lintunes import tagging


SERVICE_NAME = "org.mpris.MediaPlayer2.lintunes"
OBJECT_PATH = "/org/mpris/MediaPlayer2"

ART_CACHE_LIMIT = 50


def _art_cache_dir() -> Path:
    base = os.environ.get("XDG_CACHE_HOME", str(Path.home() / ".cache"))
    return Path(base) / "lintunes" / "artwork"


def export_artwork(track) -> str | None:
    """Write the track's embedded art to the cache dir; return a file:// URL."""
    if track is None or not track.location or not track.persistent_id:
        return None
    cache_dir = _art_cache_dir()
    target = cache_dir / f"{track.persistent_id}.jpg"
    if not target.exists():
        data = tagging.read_embedded_artwork(track.location)
        if not data:
            return None
        cache_dir.mkdir(parents=True, exist_ok=True)
        _prune_cache(cache_dir)
        tmp = target.with_suffix(".tmp")
        tmp.write_bytes(data)
        tmp.replace(target)
    return "file://" + str(target)


def invalidate_artwork(track):
    """Drop a track's cached art so the next export re-reads the file (call
    after embedding new artwork)."""
    if track is None or not track.persistent_id:
        return
    try:
        (_art_cache_dir() / f"{track.persistent_id}.jpg").unlink()
    except OSError:
        pass


def _prune_cache(cache_dir: Path):
    files = sorted(cache_dir.glob("*.jpg"), key=lambda p: p.stat().st_mtime)
    for old in files[:max(0, len(files) - ART_CACHE_LIMIT)]:
        try:
            old.unlink()
        except OSError:
            pass


def setup_mpris(player, main_window) -> "MprisService | None":
    bus = QDBusConnection.sessionBus()
    if not bus.isConnected():
        return None
    service = MprisService(player, main_window)
    if not bus.registerService(SERVICE_NAME):
        return None
    bus.registerObject(OBJECT_PATH, service,
                       QDBusConnection.RegisterOption.ExportAdaptors)
    return service


class MprisService(QObject):
    def __init__(self, player, main_window):
        super().__init__()
        self.root_adaptor = MprisRootAdaptor(self, main_window)
        self.player_adaptor = MprisPlayerAdaptor(self, player)


@pyqtClassInfo("D-Bus Interface", "org.mpris.MediaPlayer2")
class MprisRootAdaptor(QDBusAbstractAdaptor):
    def __init__(self, parent, main_window):
        super().__init__(parent)
        self._window = main_window

    @pyqtProperty(bool)
    def CanQuit(self):
        return True

    @pyqtProperty(bool)
    def CanRaise(self):
        return True

    @pyqtProperty(bool)
    def HasTrackList(self):
        return False

    @pyqtProperty(str)
    def Identity(self):
        return "LinTunes"

    @pyqtProperty(str)
    def DesktopEntry(self):
        return "lintunes"

    @pyqtProperty("QStringList")
    def SupportedUriSchemes(self):
        return ["file"]

    @pyqtProperty("QStringList")
    def SupportedMimeTypes(self):
        return ["audio/mpeg", "audio/mp4", "audio/flac"]

    @pyqtSlot()
    def Raise(self):
        self._window.show()
        self._window.raise_()
        self._window.activateWindow()

    @pyqtSlot()
    def Quit(self):
        self._window.close()


@pyqtClassInfo("D-Bus Interface", "org.mpris.MediaPlayer2.Player")
class MprisPlayerAdaptor(QDBusAbstractAdaptor):
    def __init__(self, parent, player):
        super().__init__(parent)
        self._player = player
        player.playing_changed.connect(lambda _: self._notify(
            {"PlaybackStatus": self.PlaybackStatus}))
        player.track_changed.connect(lambda _: self._notify(
            {"Metadata": self._metadata(), "PlaybackStatus": self.PlaybackStatus}))

    # -- properties --

    @pyqtProperty(str)
    def PlaybackStatus(self):
        if self._player.is_playing():
            return "Playing"
        if self._player.current_track is not None:
            return "Paused"
        return "Stopped"

    @pyqtProperty(float)
    def Rate(self):
        return 1.0

    @pyqtProperty(float)
    def MinimumRate(self):
        return 1.0

    @pyqtProperty(float)
    def MaximumRate(self):
        return 1.0

    @pyqtProperty(float)
    def Volume(self):
        return 1.0

    @pyqtProperty("qlonglong")
    def Position(self):
        return self._player.position_ms() * 1000  # microseconds

    @pyqtProperty("QVariantMap")
    def Metadata(self):
        return self._metadata()

    @pyqtProperty(bool)
    def CanGoNext(self):
        return True

    @pyqtProperty(bool)
    def CanGoPrevious(self):
        return True

    @pyqtProperty(bool)
    def CanPlay(self):
        return True

    @pyqtProperty(bool)
    def CanPause(self):
        return True

    @pyqtProperty(bool)
    def CanSeek(self):
        return True

    @pyqtProperty(bool)
    def CanControl(self):
        return True

    # -- methods --

    @pyqtSlot()
    def Next(self):
        self._player.next()

    @pyqtSlot()
    def Previous(self):
        self._player.previous()

    @pyqtSlot()
    def Pause(self):
        self._player.pause()

    @pyqtSlot()
    def PlayPause(self):
        self._player.toggle_play()

    @pyqtSlot()
    def Play(self):
        self._player.play()

    @pyqtSlot()
    def Stop(self):
        self._player.stop()

    @pyqtSlot("qlonglong")
    def Seek(self, offset_us):
        self._player.seek(self._player.position_ms() + offset_us // 1000)

    @pyqtSlot(QDBusObjectPath, "qlonglong")
    def SetPosition(self, track_id, position_us):
        self._player.seek(position_us // 1000)

    @pyqtSlot(str)
    def OpenUri(self, uri):
        pass

    # -- helpers --

    def _metadata(self) -> dict:
        track = self._player.current_track
        if track is None:
            return {"mpris:trackid": QDBusObjectPath("/org/lintunes/track/none")}
        meta = {
            "mpris:trackid": QDBusObjectPath(f"/org/lintunes/track/{track.track_id}"),
            "xesam:title": track.name or "",
            "xesam:artist": [track.artist or ""],
            "xesam:album": track.album or "",
        }
        if track.total_time:
            meta["mpris:length"] = track.total_time * 1000  # microseconds
        if track.location:
            meta["xesam:url"] = "file://" + track.location
        art_url = export_artwork(track)
        if art_url:
            meta["mpris:artUrl"] = art_url
        return meta

    def _notify(self, changed: dict):
        QDBusConnection.sessionBus().send(build_properties_changed(changed))


def build_properties_changed(changed: dict) -> QDBusMessage:
    """A spec-correct Properties.PropertiesChanged signal for the Player
    interface. invalidated_properties must be marshalled as a D-Bus string
    array ("as"); a plain Python [] goes over as "av", and GDBus consumers —
    including gsd-media-keys, which keys its media-key MRU on seeing
    PlaybackStatus change — validate the signature and drop the signal.
    (Same pitfall as reveal_paths in track_table.)"""
    msg = QDBusMessage.createSignal(
        OBJECT_PATH, "org.freedesktop.DBus.Properties", "PropertiesChanged")
    invalidated = QDBusArgument()
    invalidated.beginArray(QMetaType(QMetaType.Type.QString.value).id())
    invalidated.endArray()
    msg.setArguments(["org.mpris.MediaPlayer2.Player", changed, invalidated])
    return msg