travandClaude Opus 5.5
ad1d152948
The play journals keep running totals per track, which is enough for "most played ever" and nothing a Wrapped asks: top songs of this year, minutes listened, which month, what time of day, what got skipped. Those all come from one timestamped line per listen, so LinTunes now keeps that, while Preferences -> General -> "Keep a play history for my yearly Wrapped" is on. history/<machine>/<YYYY-MM>.jsonl, one writer per folder like the journals: local start time with its offset, ms actually heard (Player adds position steps under 1.5 s, so seeks and pauses don't count), how it ended, where it played and from what, and a name/artist/album snapshot that outlives a delete. Under 5 s is noise and never written. andTunes 0.3.0 writes the same lines on the Rabbit when sync says the desktop wants them, and sync copies each month home when the device's copy is longer. "Share this year's play history with friends" puts a trimmed copy (no ids, playlists or devices) in each friend's cassette folder, daily in November and December and monthly otherwise. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
381 lines
15 KiB
Python
381 lines
15 KiB
Python
"""Round 74: a listen history for the yearly Wrapped.
|
|
|
|
The play journals keep totals; a Wrapped needs one timestamped line per
|
|
listen. Pins: what Player counts as heard (seeks and pauses don't count), how
|
|
a listen ends, the 5 s noise floor, the per-machine month files, the Rabbit's
|
|
history coming home, the merge, and what friends receive and how often.
|
|
"""
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from lintunes import player as player_module
|
|
from lintunes.andtunes import plays as plays_mod
|
|
from lintunes.andtunes.manifest import build_manifest
|
|
from lintunes.cassette import wrapped
|
|
from lintunes.models import Library, Track
|
|
from lintunes.player import PlaybackSink, Player
|
|
from lintunes.storage import conflict_resolver, listen_history
|
|
from lintunes.storage.listen_history import ListenHistory, entry_for
|
|
|
|
|
|
class _Sink(PlaybackSink):
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self._playing = False
|
|
self._position = 0
|
|
|
|
def load(self, track, autoplay, start_ms):
|
|
self._playing = autoplay
|
|
self._position = start_ms
|
|
|
|
def play(self):
|
|
self._playing = True
|
|
|
|
def pause(self):
|
|
self._playing = False
|
|
|
|
def stop(self):
|
|
self._playing = False
|
|
|
|
def seek(self, position_ms):
|
|
self._position = position_ms
|
|
|
|
def position_ms(self):
|
|
return self._position
|
|
|
|
def duration_ms(self):
|
|
return 0
|
|
|
|
def is_playing(self):
|
|
return self._playing
|
|
|
|
def set_volume(self, level):
|
|
pass
|
|
|
|
def can_play(self, track):
|
|
return True
|
|
|
|
def unplayable_reason(self, track):
|
|
return ""
|
|
|
|
def shutdown(self):
|
|
pass
|
|
|
|
def tick(self, *positions):
|
|
for p in positions:
|
|
self._position = p
|
|
self.position_changed.emit(p)
|
|
|
|
|
|
def _player(qapp, tmp_path, count=2):
|
|
library = Library()
|
|
for tid in range(1, count + 1):
|
|
path = tmp_path / f"{tid}.mp3"
|
|
path.write_bytes(b"x")
|
|
library.tracks[tid] = Track(track_id=tid, name=f"Song {tid}",
|
|
artist="Artist", album="Album",
|
|
genre="Jazz", year=1961,
|
|
total_time=200_000, location=str(path))
|
|
manager = MagicMock(library=library)
|
|
with patch.multiple(player_module, QMediaPlayer=MagicMock(),
|
|
QAudioOutput=MagicMock(),
|
|
QAudioBufferOutput=MagicMock(),
|
|
QMediaDevices=MagicMock()):
|
|
player = Player(manager)
|
|
sink = _Sink()
|
|
player.set_sink(sink)
|
|
ended = []
|
|
player.listen_ended.connect(lambda *args: ended.append(args))
|
|
return player, sink, ended, manager
|
|
|
|
|
|
def _steps(start, stop, step=500):
|
|
return range(start, stop + 1, step)
|
|
|
|
|
|
class TestPlayerAccounting:
|
|
def test_skip_reports_time_heard(self, qapp, tmp_path):
|
|
player, sink, ended, _ = _player(qapp, tmp_path)
|
|
player.play_queue([1, 2], 0)
|
|
sink.tick(*_steps(500, 10_000))
|
|
player.next()
|
|
(key, track, started, heard, how), = ended
|
|
assert (key, track.track_id, heard, how) == (1, 1, 10_000, "skip")
|
|
assert started.tzinfo is not None # local time, with its offset
|
|
|
|
def test_a_seek_is_not_listening(self, qapp, tmp_path):
|
|
player, sink, ended, _ = _player(qapp, tmp_path)
|
|
player.play_queue([1, 2], 0)
|
|
sink.tick(500, 1000, 120_000, 120_500, 3000, 3500) # forward, then back
|
|
player.next()
|
|
assert ended[0][3] == 2000
|
|
|
|
def test_a_natural_finish(self, qapp, tmp_path):
|
|
player, sink, ended, manager = _player(qapp, tmp_path)
|
|
player.play_queue([1, 2], 0)
|
|
sink.tick(*_steps(500, 6000))
|
|
sink.ended.emit()
|
|
assert [(e[1].track_id, e[4]) for e in ended] == [(1, "finish")]
|
|
manager.record_play.assert_called_once_with(1)
|
|
|
|
def test_stop_and_quit(self, qapp, tmp_path):
|
|
player, sink, ended, _ = _player(qapp, tmp_path)
|
|
player.play_queue([1, 2], 0)
|
|
sink.tick(*_steps(500, 6000))
|
|
player.stop()
|
|
player.play_queue([2], 0)
|
|
sink.tick(*_steps(500, 6000))
|
|
player.shutdown()
|
|
assert [e[4] for e in ended] == ["stop", "quit"]
|
|
|
|
def test_nothing_heard_reports_nothing(self, qapp, tmp_path):
|
|
player, sink, ended, _ = _player(qapp, tmp_path)
|
|
player.play_queue([1, 2], 0)
|
|
player.next()
|
|
assert ended == []
|
|
|
|
def test_replaying_the_last_track_opens_a_new_listen(self, qapp, tmp_path):
|
|
player, sink, ended, _ = _player(qapp, tmp_path, count=1)
|
|
player.play_queue([1], 0)
|
|
sink.tick(*_steps(500, 6000))
|
|
sink.ended.emit() # queue ran out; track stays current
|
|
sink._position = 0
|
|
player.toggle_play()
|
|
sink.tick(*_steps(500, 7000))
|
|
player.stop()
|
|
assert [(e[3], e[4]) for e in ended] == [(6000, "finish"), (7000, "stop")]
|
|
|
|
|
|
def _track(**fields):
|
|
base = dict(track_id=7, name="Alhambra", artist="Ahmad Jamal",
|
|
album_artist="Ahmad Jamal", album="Alhambra", genre="Jazz",
|
|
year=1961, total_time=300_000)
|
|
base.update(fields)
|
|
return Track(**base)
|
|
|
|
|
|
WHEN = datetime(2026, 10, 2, 21, 14, 3, tzinfo=timezone(timedelta(hours=-7)))
|
|
|
|
|
|
class TestEntry:
|
|
def test_shape(self):
|
|
entry = entry_for(_track(), started=WHEN, heard_ms=200_000,
|
|
how="finish", src="desktop", ctx="playlist:ab12",
|
|
shuffle=True, track_id=7)
|
|
assert entry == {"t": "2026-10-02T21:14:03-07:00", "ms": 200_000,
|
|
"len": 300_000, "end": "finish", "src": "desktop",
|
|
"id": 7, "ctx": "playlist:ab12", "shuf": True,
|
|
"n": "Alhambra", "a": "Ahmad Jamal",
|
|
"aa": "Ahmad Jamal", "al": "Alhambra", "g": "Jazz",
|
|
"y": 1961}
|
|
|
|
def test_under_five_seconds_is_noise(self):
|
|
assert entry_for(_track(), started=WHEN, heard_ms=4999, how="skip",
|
|
src="desktop") is None
|
|
|
|
def test_friend_track_has_no_id(self):
|
|
entry = entry_for(_track(), started=WHEN, heard_ms=9000, how="skip",
|
|
src="desktop", track_id="cassette:tok:7")
|
|
assert "id" not in entry
|
|
|
|
|
|
class TestHistoryFiles:
|
|
def _entry(self, when, name="x"):
|
|
return entry_for(_track(name=name), started=when, heard_ms=60_000,
|
|
how="finish", src="desktop", track_id=7)
|
|
|
|
def test_month_files_per_machine(self, tmp_path):
|
|
mine, theirs = ListenHistory("aaa"), ListenHistory("bbb")
|
|
mine.add(self._entry(WHEN))
|
|
mine.add(self._entry(WHEN + timedelta(days=40), "november"))
|
|
theirs.add(self._entry(WHEN - timedelta(hours=1), "earlier"))
|
|
mine.flush(tmp_path)
|
|
theirs.flush(tmp_path)
|
|
names = sorted(str(p.relative_to(tmp_path))
|
|
for p in (tmp_path / "history").rglob("*.jsonl"))
|
|
assert names == ["history/aaa/2026-10.jsonl", "history/aaa/2026-11.jsonl",
|
|
"history/bbb/2026-10.jsonl"]
|
|
assert [e["n"] for e in listen_history.read_year(tmp_path, 2026)] == \
|
|
["earlier", "x", "november"]
|
|
assert not mine.dirty
|
|
|
|
def test_flush_appends(self, tmp_path):
|
|
history = ListenHistory("aaa")
|
|
for name in ("one", "two"):
|
|
history.add(self._entry(WHEN, name))
|
|
history.flush(tmp_path)
|
|
assert [e["n"] for e in listen_history.read_year(tmp_path, 2026)] == \
|
|
["one", "two"]
|
|
|
|
def test_torn_line_is_skipped(self, tmp_path):
|
|
history = ListenHistory("aaa")
|
|
history.add(self._entry(WHEN))
|
|
(path,) = history.flush(tmp_path)
|
|
with open(path, "a") as f:
|
|
f.write('{"t": "2026-10-0')
|
|
assert len(listen_history.read_year(tmp_path, 2026)) == 1
|
|
|
|
def test_other_years_are_left_out(self, tmp_path):
|
|
history = ListenHistory("aaa")
|
|
history.add(self._entry(WHEN.replace(year=2025)))
|
|
history.add(self._entry(WHEN))
|
|
history.flush(tmp_path)
|
|
assert len(listen_history.read_year(tmp_path, 2026)) == 1
|
|
|
|
def test_trim_for_sharing(self):
|
|
entry = self._entry(WHEN)
|
|
entry.update(ctx="playlist:x", shuf=True)
|
|
(trimmed,) = listen_history.trim_for_sharing([entry])
|
|
assert not {"id", "ctx", "src", "shuf"} & set(trimmed)
|
|
assert trimmed["a"] == "Ahmad Jamal" and trimmed["ms"] == 60_000
|
|
|
|
def test_conflict_merge_is_a_union(self, tmp_path):
|
|
folder = tmp_path / "history" / "andtunes-cafe"
|
|
folder.mkdir(parents=True)
|
|
a = json.dumps(self._entry(WHEN, "a"))
|
|
b = json.dumps(self._entry(WHEN + timedelta(minutes=5), "b"))
|
|
c = json.dumps(self._entry(WHEN + timedelta(minutes=9), "c"))
|
|
original = folder / "2026-10.jsonl"
|
|
original.write_text(f"{a}\n{b}\n")
|
|
conflict = folder / "2026-10.sync-conflict-20261002-211500-ABCDEFG.jsonl"
|
|
conflict.write_text(f"{a}\n{c}\n")
|
|
conflict_resolver.resolve_conflicts(tmp_path)
|
|
assert not conflict.exists()
|
|
assert [e["n"] for e in listen_history.read_lines(original)] == \
|
|
["a", "b", "c"]
|
|
|
|
|
|
class TestManagerAndGate:
|
|
def test_manager_writes_with_the_next_save(self, qapp, tmp_path):
|
|
from lintunes.library_manager import LibraryManager
|
|
manager = LibraryManager(Library(), tmp_path)
|
|
manager.record_listen(entry_for(_track(), started=WHEN, heard_ms=9000,
|
|
how="skip", src="desktop", track_id=7))
|
|
manager.record_listen(None)
|
|
manager.flush()
|
|
assert len(listen_history.read_year(tmp_path, 2026)) == 1
|
|
|
|
@pytest.mark.parametrize("on", [False, True])
|
|
def test_the_preference_gates_it(self, qapp, tmp_path, on):
|
|
from lintunes.gui.main_window import MainWindow
|
|
from lintunes.library_manager import LibraryManager
|
|
from lintunes.preferences import Preferences
|
|
manager = LibraryManager(Library(), tmp_path / "data")
|
|
prefs = Preferences(tmp_path / "data")
|
|
prefs.update_wrapped(track_history=on)
|
|
window = MainWindow(manager, prefs)
|
|
try:
|
|
window._listen_context = "followed:cassette:tok:pid"
|
|
window._on_listen_ended("cassette:tok:7", _track(), WHEN, 9000,
|
|
"skip")
|
|
pending = manager.listen_history.pending
|
|
if not on:
|
|
assert pending == []
|
|
else:
|
|
(entry,) = pending
|
|
assert entry["ctx"] == "friend:tok" and "id" not in entry
|
|
assert entry["src"] == "desktop"
|
|
finally:
|
|
window.close()
|
|
|
|
|
|
class TestRabbit:
|
|
def _line(self, name):
|
|
return json.dumps({"t": "2026-10-02T09:00:00-07:00", "ms": 9000,
|
|
"n": name, "src": "r1"}) + "\n"
|
|
|
|
def test_history_comes_home_when_longer(self, tmp_path):
|
|
root, data = tmp_path / "andTunes", tmp_path / "data"
|
|
month = root / "history" / "andtunes-cafe" / "2026-10.jsonl"
|
|
month.parent.mkdir(parents=True)
|
|
month.write_text(self._line("one"))
|
|
assert plays_mod.bring_back_history(root, data) == 1
|
|
dest = data / "history" / "andtunes-cafe" / "2026-10.jsonl"
|
|
assert dest.read_text() == self._line("one")
|
|
assert plays_mod.bring_back_history(root, data) == 0 # same length
|
|
month.write_text(self._line("one") + self._line("two") + '{"t": "20')
|
|
assert plays_mod.bring_back_history(root, data) == 1
|
|
assert dest.read_text() == self._line("one") + self._line("two")
|
|
|
|
def test_no_history_folder(self, tmp_path):
|
|
assert plays_mod.bring_back_history(tmp_path, tmp_path / "data") == 0
|
|
|
|
def test_manifest_flag(self):
|
|
assert "history" not in build_manifest([], [], generator="t")
|
|
assert build_manifest([], [], generator="t", history=True)["history"] is True
|
|
|
|
def test_sync_never_prunes_the_history(self):
|
|
from lintunes.andtunes.layout import PRESERVED_DIRS
|
|
assert "history" in PRESERVED_DIRS
|
|
|
|
|
|
def _utc(*args):
|
|
return datetime(*args, tzinfo=timezone.utc)
|
|
|
|
|
|
class TestSharingCadence:
|
|
def test_monthly_off_season_daily_in_season(self):
|
|
oct2 = _utc(2026, 10, 2)
|
|
assert wrapped.is_due("", oct2)
|
|
assert not wrapped.is_due("2026-09-20T00:00:00+00:00", oct2)
|
|
assert wrapped.is_due("2026-09-01T00:00:00+00:00", oct2)
|
|
nov = _utc(2026, 11, 20, 12)
|
|
assert not wrapped.is_due("2026-11-20T00:00:00+00:00", nov)
|
|
assert wrapped.is_due("2026-11-19T11:00:00+00:00", nov)
|
|
|
|
def test_january_finishes_last_year(self):
|
|
assert wrapped.years_to_write(_utc(2027, 1, 3)) == [2026, 2027]
|
|
assert wrapped.years_to_write(_utc(2026, 10, 3)) == [2026]
|
|
|
|
def _service(self, tmp_path, tokens=("t1", "t2")):
|
|
from lintunes.cassette.service import CassetteService
|
|
from lintunes.cassette.state import Friend
|
|
service = CassetteService(root=tmp_path / "cassette",
|
|
config_loader=lambda: {})
|
|
for token in tokens:
|
|
service._state.friends[token] = Friend(token=token, device_id="D",
|
|
name=token)
|
|
return service
|
|
|
|
def _data(self, tmp_path):
|
|
history = ListenHistory("aaa")
|
|
history.add(entry_for(_track(), started=WHEN, heard_ms=60_000,
|
|
how="finish", src="desktop", ctx="playlist:x",
|
|
track_id=7))
|
|
history.flush(tmp_path / "data")
|
|
return tmp_path / "data"
|
|
|
|
def test_publish_then_quiet_then_off(self, qapp, tmp_path):
|
|
from lintunes.cassette.state import out_dir
|
|
service = self._service(tmp_path)
|
|
data = self._data(tmp_path)
|
|
now = _utc(2026, 10, 2, 12)
|
|
assert service.publish_wrapped(data, True, now) == 2
|
|
out = out_dir(service.root, "t1") / "wrapped-2026.json"
|
|
doc = json.loads(out.read_text())
|
|
assert doc["year"] == 2026 and len(doc["plays"]) == 1
|
|
assert "ctx" not in doc["plays"][0] and "id" not in doc["plays"][0]
|
|
mtime = out.stat().st_mtime_ns
|
|
# Not due again for a month, and even when due, unchanged plays
|
|
# aren't rewritten.
|
|
assert service.publish_wrapped(data, True, now + timedelta(days=3)) == 0
|
|
assert service.publish_wrapped(data, True, now + timedelta(days=31)) == 0
|
|
assert out.stat().st_mtime_ns == mtime
|
|
# Switching it off takes it back.
|
|
assert service.publish_wrapped(data, False, now) == 2
|
|
assert not out.exists()
|
|
assert service.state().wrapped_published_at == ""
|
|
|
|
def test_new_friend_gets_theirs_without_waiting(self, qapp, tmp_path):
|
|
from lintunes.cassette.state import Friend, out_dir
|
|
service = self._service(tmp_path, tokens=("t1",))
|
|
data = self._data(tmp_path)
|
|
now = _utc(2026, 10, 2, 12)
|
|
service.publish_wrapped(data, True, now)
|
|
service._state.friends["t2"] = Friend(token="t2", device_id="E", name="t2")
|
|
assert service.publish_wrapped(data, True, now + timedelta(hours=1)) == 1
|
|
assert (out_dir(service.root, "t2") / "wrapped-2026.json").exists()
|