lintunes/scripts/watch_mpris_senders.py

#!/usr/bin/env python3
"""Attribute every D-Bus call into LinTunes' MPRIS interface to the sender.

Run this in the background during normal use (and overnight) to catch the
"phantom playback" incidents: lintunes' control-events.log shows those resumes
arriving as [mpris] PlayPause / Play, but can't name *who* sent the call (PyQt6
has no QDBusContext — see lintunes/mpris.py). This script does the other half:
it tails `dbus-monitor --session`, filters for method calls on
org.mpris.MediaPlayer2(.Player), and for each one resolves the sender's unique
bus name (:1.NN) to a PID + command via the D-Bus daemon, writing a provenance
line to ~/.cache/lintunes/mpris-senders.log (and stdout).

Read-only: it never touches lintunes, the library, or any music file.

Usage:
    python3 scripts/watch_mpris_senders.py            # log + echo to stdout
    python3 scripts/watch_mpris_senders.py --quiet    # log only
    python3 scripts/watch_mpris_senders.py --follow  # also tail the log file

Then keep it running. When you next hear the ~4 s of phantom audio, look at the
last lines of ~/.cache/lintunes/mpris-senders.log — it will name the process
(almost certainly gsd-media-keys fed by bluetoothd/wireplumber, but let's prove
it rather than guess).
"""
import argparse
import os
import subprocess
import sys
import time
from pathlib import Path

LOG_PATH = Path.home() / ".cache" / "lintunes" / "mpris-senders.log"

# Members of the Player interface that can start/change playback. We log all
# MPRIS calls actually, but flag these in the human log as "playback-affecting".
PLAYBACK_MEMBERS = {
    "Play", "PlayPause", "Pause", "Stop", "Next", "Previous",
    "Seek", "SetPosition", "OpenUri",
}


def _service_pid(sender: str) -> tuple[int | None, str]:
    """Resolve a D-Bus unique name (:1.NN) to (pid, comm). Returns
    (None, reason) when the bus daemon can't (or won't) tell us."""
    try:
        out = subprocess.run(
            ["busctl", "--user", "call",
             "org.freedesktop.DBus", "/org/freedesktop/DBus",
             "org.freedesktop.DBus", "GetConnectionUnixProcessID", "s", sender],
            capture_output=True, text=True, timeout=5,
        )
    except (FileNotFoundError, subprocess.SubprocessError) as exc:
        return None, f"busctl error: {exc}"
    if out.returncode != 0:
        return None, f"busctl rc={out.returncode}: {out.stderr.strip()}"
    parts = out.stdout.split()
    if len(parts) < 2 or parts[0] != "u":
        return None, f"unparsed busctl output: {out.stdout.strip()!r}"
    try:
        pid = int(parts[1])
    except ValueError:
        return None, f"non-numeric pid: {parts[1]!r}"
    # Get the human name of the process for the log. /proc/<pid>/comm is short
    # (truncated to 15 chars by the kernel); cmdline gives the full invocation
    # but is NUL-separated and may include the executable path — join and trim.
    comm = f"pid={pid}"
    try:
        with open(f"/proc/{pid}/comm", encoding="utf-8", errors="replace") as f:
            comm = f.read().strip() or comm
        with open(f"/proc/{pid}/cmdline", "rb") as f:
            args = f.read().replace(b"\x00", b" ").decode(
                "utf-8", errors="replace").strip()
        if args:
            comm = f"{comm} :: {args[:160]}"
    except OSError as exc:
        comm = f"{comm} (comm unreadable: {exc})"
    return pid, comm


def _parse_method_call(line: str):
    """Pull (sender, destination, member, interface) out of a dbus-monitor
    "method call" line, or return None if it isn't one / isn't parseable."""
    if not line.startswith("method call"):
        return None
    if "org.mpris.MediaPlayer2" not in line:
        return None
    fields = {}
    # dbus-monitor writes "key=value" tokens separated by whitespace, and a
    # trailing "path=...; interface=...; member=..." group. Splitting on ';' or
    # whitespace and taking the last '=' segment of each token covers both.
    for tok in line.replace(";", " ").split():
        if "=" in tok:
            k, _, v = tok.partition("=")
            fields.setdefault(k, v)
    if "member" not in fields or "sender" not in fields:
        return None
    return fields


def _timestamp() -> str:
    return time.strftime("%F %T")


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--quiet", action="store_true",
                    help="don't echo to stdout (still writes the log file)")
    ap.add_argument("--follow", action="store_true",
                    help="also tail the log file to stdout as it's written")
    args = ap.parse_args()

    LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
    # dbus-monitor emits two preamble lines naming the bus type; keep going.
    proc = subprocess.Popen(
        ["dbus-monitor", "--session"],
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
        text=True, bufsize=1,
    )
    # Cache sender->(pid, comm) so repeat presses from the same process don't
    # retry the D-Bus daemon call every line. Refresh if the pid ever changes.
    pid_cache: dict[str, tuple[int | None, str]] = {}

    print(f"[{time.strftime('%F %T')}] watching session D-Bus for MPRIS calls "
          f"-> {LOG_PATH}", file=sys.stderr)
    try:
        for line in proc.stdout:            # dbus-monitor blocks here
            line = line.rstrip()
            parsed = _parse_method_call(line)
            if parsed is None:
                continue
            sender = parsed["sender"]
            member = parsed["member"]
            dest = parsed.get("destination", "?")
            iface = parsed.get("interface", "?")
            pid_comm = pid_cache.get(sender)
            if pid_comm is None:
                pid, comm = _service_pid(sender)
                pid_cache[sender] = (pid, comm)
            pid, comm = pid_cache[sender]
            marker = "<<" if member in PLAYBACK_MEMBERS else "  "
            rec = (f"{_timestamp()} {marker} {member:10s} "
                   f"sender={sender} dest={dest} iface={iface} {comm}")
            with open(LOG_PATH, "a", encoding="utf-8") as f:
                f.write(rec + "\n")
            if not args.quiet:
                print(rec)
    except KeyboardInterrupt:
        pass
    finally:
        proc.terminate()
        proc.wait()
        if args.follow:
            print(f"\n[{time.strftime('%F %T')}] stopped; log: {LOG_PATH}",
                  file=sys.stderr)


if __name__ == "__main__":
    main()