v0.1.4: playback-control provenance log
LinTunes has spontaneously resumed playback for ~4s (then paused) four times while trav was away — suspected phantom media-key events from the Audioengine USB DAC's HID 'keyboard', but the pathway (MPRIS vs the focused-window key filter) is unproven. Every control path now drops a timestamped line in ~/.cache/lintunes/control-events.log: MPRIS Player methods, the media-key eventFilter (with source input device where the compositor exposes it), and Player.toggle_play/pause with position. New lintunes/eventlog.py, 1MB rotation, never raises; conftest autouse fixture keeps tests off the real log file. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
22
TASKS.md
22
TASKS.md
@ -3,6 +3,28 @@
|
||||
Legend: `[ ]` todo · `[~]` in progress · `[x]` done.
|
||||
When a round closes, move its finished items to `tasks-done.md`.
|
||||
|
||||
## Round 23 — playback-control provenance log (v0.1.4)
|
||||
|
||||
Tests in `tests/test_round23.py`. Diagnostic round → patch bump **0.1.4**.
|
||||
|
||||
Background: LinTunes has resumed playback by itself for ~4s (then paused)
|
||||
four times while trav was away/asleep (Jul 5, 7, 8?, 10). Fingerprint each
|
||||
time: current track resumed from its paused position, paused ~4s later.
|
||||
Suspected phantom media-key events (the Audioengine 2+ USB DAC registers a
|
||||
HID *keyboard*), but the pathway is unproven — hence instrumentation.
|
||||
|
||||
- [x] New `lintunes/eventlog.py` — `log_control(source, action, detail)`
|
||||
appends timestamped one-liners to
|
||||
`~/.cache/lintunes/control-events.log` (1 MB rotate, never raises)
|
||||
and mirrors to the `lintunes.control` logger.
|
||||
- [x] Provenance hooks: every MPRIS Player method (`mpris.py`; PyQt6 lacks
|
||||
QDBusContext so no sender — run dbus-monitor alongside when the
|
||||
sender matters), the media-key `eventFilter` in `main_window.py`
|
||||
(logs the source input device where the compositor exposes it), and
|
||||
`Player.toggle_play`/`pause` (state + position).
|
||||
- [x] `tests/conftest.py`: autouse fixture isolates the log path so tests
|
||||
never write the real ~/.cache file.
|
||||
|
||||
## Round 22 — start time honored on same-source replay (v0.1.3)
|
||||
|
||||
Plan reference: `~/.claude/plans/could-you-look-into-generic-balloon.md`.
|
||||
|
||||
@ -1,3 +1,3 @@
|
||||
"""LinTunes — iTunes-style music library manager and player for Linux."""
|
||||
|
||||
__version__ = "0.1.3"
|
||||
__version__ = "0.1.4"
|
||||
|
||||
37
lintunes/eventlog.py
Normal file
37
lintunes/eventlog.py
Normal file
@ -0,0 +1,37 @@
|
||||
"""Append-only provenance log for playback-control events.
|
||||
|
||||
Every path that can start/stop playback (MPRIS calls, the media-key
|
||||
eventFilter, the Player's transport methods) drops a one-line record here,
|
||||
so "the app played by itself" incidents can be attributed after the fact:
|
||||
which pathway fired, when, and — for D-Bus — who sent it. Written to
|
||||
~/.cache/lintunes/control-events.log (survives app restarts and doesn't
|
||||
depend on how the app was launched); mirrored to the standard logger.
|
||||
"""
|
||||
import logging
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
log = logging.getLogger("lintunes.control")
|
||||
|
||||
_MAX_BYTES = 1_000_000
|
||||
|
||||
|
||||
def _log_path() -> Path:
|
||||
return Path.home() / ".cache" / "lintunes" / "control-events.log"
|
||||
|
||||
|
||||
def log_control(source: str, action: str, detail: str = ""):
|
||||
"""Record a playback-control event; never raises."""
|
||||
line = f"{time.strftime('%F %T')} [{source}] {action}"
|
||||
if detail:
|
||||
line += f" {detail}"
|
||||
log.info("[%s] %s %s", source, action, detail)
|
||||
try:
|
||||
path = _log_path()
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if path.exists() and path.stat().st_size > _MAX_BYTES:
|
||||
path.replace(path.with_suffix(".log.old"))
|
||||
with open(path, "a", encoding="utf-8") as f:
|
||||
f.write(line + "\n")
|
||||
except Exception:
|
||||
pass # a provenance logger must never break playback
|
||||
@ -10,6 +10,7 @@ from PyQt6.QtGui import QAction, QKeySequence
|
||||
|
||||
from lintunes import mpris, tagging, theme
|
||||
from lintunes.art_search import AlbumArtFetcher
|
||||
from lintunes.eventlog import log_control
|
||||
from lintunes.inhibit import SleepInhibitor
|
||||
from lintunes.player import Player
|
||||
from lintunes.importers import file_importer
|
||||
@ -551,16 +552,32 @@ class MainWindow(QMainWindow):
|
||||
return False
|
||||
key = event.key()
|
||||
if key == Qt.Key.Key_Space:
|
||||
self._log_key(event, "space")
|
||||
self.play_pause()
|
||||
return True
|
||||
if key == Qt.Key.Key_Right:
|
||||
self._log_key(event, "right")
|
||||
self.player.next()
|
||||
return True
|
||||
if key == Qt.Key.Key_Left:
|
||||
self._log_key(event, "left")
|
||||
self.player.previous()
|
||||
return True
|
||||
return super().eventFilter(obj, event)
|
||||
|
||||
@staticmethod
|
||||
def _log_key(event, name: str):
|
||||
"""Provenance for keys the filter acts on: which input device sent
|
||||
the press (compositor permitting) — see eventlog module docstring."""
|
||||
detail = ""
|
||||
try:
|
||||
device = event.device()
|
||||
if device is not None:
|
||||
detail = f"device={device.name()!r}"
|
||||
except Exception:
|
||||
pass
|
||||
log_control("keyfilter", name, detail)
|
||||
|
||||
def _on_playing_changed(self, playing: bool):
|
||||
if playing:
|
||||
self._inhibitor.inhibit()
|
||||
|
||||
@ -15,6 +15,7 @@ from PyQt6.QtDBus import (
|
||||
)
|
||||
|
||||
from lintunes import tagging
|
||||
from lintunes.eventlog import log_control
|
||||
|
||||
|
||||
SERVICE_NAME = "org.mpris.MediaPlayer2.lintunes"
|
||||
@ -200,36 +201,49 @@ class MprisPlayerAdaptor(QDBusAbstractAdaptor):
|
||||
|
||||
# -- methods --
|
||||
|
||||
def _log(self, action: str):
|
||||
# PyQt6 doesn't wrap QDBusContext, so the caller can't be identified
|
||||
# in-process; run `dbus-monitor` alongside when the sender matters.
|
||||
log_control("mpris", action)
|
||||
|
||||
@pyqtSlot()
|
||||
def Next(self):
|
||||
self._log("Next")
|
||||
self._player.next()
|
||||
|
||||
@pyqtSlot()
|
||||
def Previous(self):
|
||||
self._log("Previous")
|
||||
self._player.previous()
|
||||
|
||||
@pyqtSlot()
|
||||
def Pause(self):
|
||||
self._log("Pause")
|
||||
self._player.pause()
|
||||
|
||||
@pyqtSlot()
|
||||
def PlayPause(self):
|
||||
self._log("PlayPause")
|
||||
self._player.toggle_play()
|
||||
|
||||
@pyqtSlot()
|
||||
def Play(self):
|
||||
self._log("Play")
|
||||
self._player.play()
|
||||
|
||||
@pyqtSlot()
|
||||
def Stop(self):
|
||||
self._log("Stop")
|
||||
self._player.stop()
|
||||
|
||||
@pyqtSlot("qlonglong")
|
||||
def Seek(self, offset_us):
|
||||
self._log("Seek")
|
||||
self._player.seek(self._player.position_ms() + offset_us // 1000)
|
||||
|
||||
@pyqtSlot(QDBusObjectPath, "qlonglong")
|
||||
def SetPosition(self, track_id, position_us):
|
||||
self._log("SetPosition")
|
||||
self._player.seek(position_us // 1000)
|
||||
|
||||
@pyqtSlot(str)
|
||||
|
||||
@ -6,6 +6,7 @@ from PyQt6.QtCore import QObject, QUrl, pyqtSignal
|
||||
from PyQt6.QtMultimedia import (
|
||||
QMediaPlayer, QAudioOutput, QAudioBufferOutput, QMediaDevices, QAudio)
|
||||
|
||||
from lintunes.eventlog import log_control
|
||||
from lintunes.models import Track
|
||||
|
||||
|
||||
@ -197,6 +198,9 @@ class Player(QObject):
|
||||
# ---- transport ----
|
||||
|
||||
def toggle_play(self):
|
||||
log_control("player", "toggle_play",
|
||||
f"playing={self.is_playing()} pos={self.position_ms()}ms "
|
||||
f"track={self._current_track.name if self._current_track else None!r}")
|
||||
if self.is_playing():
|
||||
self._paused_at = time.monotonic()
|
||||
self._media.pause()
|
||||
@ -221,6 +225,7 @@ class Player(QObject):
|
||||
|
||||
def pause(self):
|
||||
if self.is_playing():
|
||||
log_control("player", "pause", f"pos={self.position_ms()}ms")
|
||||
self._paused_at = time.monotonic()
|
||||
self._media.pause()
|
||||
|
||||
|
||||
@ -8,6 +8,15 @@ import pytest
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolate_eventlog(tmp_path, monkeypatch):
|
||||
"""Keep the control-provenance log out of the real ~/.cache during tests
|
||||
(Player/MPRIS methods log unconditionally)."""
|
||||
from lintunes import eventlog
|
||||
monkeypatch.setattr(eventlog, "_log_path",
|
||||
lambda: tmp_path / "control-events.log")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def qapp():
|
||||
"""A Qt application so QObject/QTimer and real widgets work in tests.
|
||||
|
||||
105
tests/test_round23.py
Normal file
105
tests/test_round23.py
Normal file
@ -0,0 +1,105 @@
|
||||
"""Round 23: playback-control provenance log.
|
||||
|
||||
Every pathway that can start/stop playback now drops a one-line record via
|
||||
lintunes.eventlog (MPRIS methods with D-Bus sender, the media-key
|
||||
eventFilter with source device, Player transport methods with position), so
|
||||
"the app played by itself at night" incidents are attributable after the
|
||||
fact from ~/.cache/lintunes/control-events.log.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from lintunes import eventlog
|
||||
from lintunes.models import Library, Track
|
||||
from lintunes import player as player_module
|
||||
from lintunes.player import Player
|
||||
|
||||
|
||||
def _patch_log(monkeypatch, tmp_path):
|
||||
path = tmp_path / "control-events.log"
|
||||
monkeypatch.setattr(eventlog, "_log_path", lambda: path)
|
||||
return path
|
||||
|
||||
|
||||
class TestEventLog:
|
||||
def test_writes_timestamped_line(self, monkeypatch, tmp_path):
|
||||
path = _patch_log(monkeypatch, tmp_path)
|
||||
eventlog.log_control("test", "action", "k=v")
|
||||
line = path.read_text().strip()
|
||||
assert line.endswith("[test] action k=v")
|
||||
assert line[:4].isdigit() # starts with a date
|
||||
|
||||
def test_detail_optional(self, monkeypatch, tmp_path):
|
||||
path = _patch_log(monkeypatch, tmp_path)
|
||||
eventlog.log_control("test", "solo")
|
||||
assert path.read_text().strip().endswith("[test] solo")
|
||||
|
||||
def test_rotates_when_large(self, monkeypatch, tmp_path):
|
||||
path = _patch_log(monkeypatch, tmp_path)
|
||||
path.write_bytes(b"x" * (eventlog._MAX_BYTES + 1))
|
||||
eventlog.log_control("test", "rotated")
|
||||
assert path.with_suffix(".log.old").exists()
|
||||
assert "rotated" in path.read_text()
|
||||
|
||||
def test_never_raises_on_io_error(self, monkeypatch):
|
||||
monkeypatch.setattr(eventlog, "_log_path",
|
||||
lambda: type("P", (), {})()) # broken path object
|
||||
eventlog.log_control("test", "no-crash") # must not raise
|
||||
|
||||
|
||||
class TestMprisLogging:
|
||||
def test_playpause_logs_and_forwards(self, qapp, monkeypatch, tmp_path):
|
||||
path = _patch_log(monkeypatch, tmp_path)
|
||||
from lintunes.mpris import MprisService
|
||||
player = MagicMock()
|
||||
service = MprisService(player, MagicMock())
|
||||
service.player_adaptor.PlayPause()
|
||||
player.toggle_play.assert_called_once()
|
||||
text = path.read_text()
|
||||
assert "[mpris] PlayPause" in text
|
||||
|
||||
def test_all_methods_log(self, qapp, monkeypatch, tmp_path):
|
||||
path = _patch_log(monkeypatch, tmp_path)
|
||||
from lintunes.mpris import MprisService
|
||||
service = MprisService(MagicMock(), MagicMock())
|
||||
adaptor = service.player_adaptor
|
||||
adaptor.Play()
|
||||
adaptor.Pause()
|
||||
adaptor.Next()
|
||||
adaptor.Previous()
|
||||
adaptor.Stop()
|
||||
text = path.read_text()
|
||||
for member in ("Play", "Pause", "Next", "Previous", "Stop"):
|
||||
assert f"[mpris] {member}" in text
|
||||
|
||||
|
||||
class TestPlayerLogging:
|
||||
def _player(self, qapp, tracks):
|
||||
library = Library()
|
||||
for t in tracks:
|
||||
library.tracks[t.track_id] = t
|
||||
manager = MagicMock()
|
||||
manager.library = library
|
||||
with patch.multiple(
|
||||
player_module,
|
||||
QMediaPlayer=MagicMock(),
|
||||
QAudioOutput=MagicMock(),
|
||||
QAudioBufferOutput=MagicMock(),
|
||||
QMediaDevices=MagicMock(),
|
||||
):
|
||||
return Player(manager)
|
||||
|
||||
def test_toggle_play_logs_state_and_position(self, qapp, monkeypatch,
|
||||
tmp_path):
|
||||
path = _patch_log(monkeypatch, tmp_path)
|
||||
(tmp_path / "a.mp3").write_bytes(b"x")
|
||||
track = Track(track_id=1, name="Song A",
|
||||
location=str(tmp_path / "a.mp3"), total_time=100_000)
|
||||
player = self._player(qapp, [track])
|
||||
player._media.position.return_value = 4321
|
||||
player._media.playbackState.return_value = None # != PlayingState
|
||||
player.play_queue([1], 0)
|
||||
player.toggle_play()
|
||||
text = path.read_text()
|
||||
assert "[player] toggle_play" in text
|
||||
assert "pos=4321ms" in text and "'Song A'" in text
|
||||
Reference in New Issue
Block a user