Player revamp Phase 2: smart radio ranking (play-count + popularity)

Replaces radio's pure ORDER BY RANDOM() with weighted ranking. Each tier now
fetches a generous random POOL (4x the needed count, floored) and
core/radio/selection ranks it before the collector keeps the best:

  score_candidate = play_count(log-damped, w=1.0)
                  + lastfm_playcount(log-damped, w=0.5)
                  - recently_played penalty(w=2.0)
                  + stable per-id jitter(w=1.0, hash-derived so runs vary but
                    tests stay reproducible)

Modest weights so popularity guides without burying lesser-played tracks, and
jitter keeps radio from being identical every run. All intelligence is in pure
functions (rank_candidates / score_candidate) so it's tunable + unit-testable
without SQL.

Defensive: the DB method probes PRAGMA table_info(tracks) and omits
play_count/lastfm_playcount from the SELECT when absent (older DBs predating
the listening-history migration) — the scorer treats missing signals as 0, so
radio degrades to jitter-only instead of crashing on 'no such column'.

Tests (tests/radio/, 43 total):
  - score_candidate / rank_candidates: deterministic unit coverage (popularity
    ordering, lastfm contribution, recency penalty, garbage→0, stable jitter).
    These CANNOT pass against pre-Phase-2 code.
  - DB end-to-end: ranking surfaces the heavily-played track first out of a
    decoy pool (wiring proof — probabilistic vs old random, documented honestly);
    plus a no-rank-columns DB proving the defensive degrade path.
  - All Phase-0a behavioral/refactor-equivalence tests still green.
60 radio + adjacent-DB tests pass; ruff clean.

core/radio/__init__.py

@@ -10,13 +10,19 @@ smarter ranking will plug into.
 from core.radio.selection import (
     RadioCollector,
     build_like_conditions,
+    merge_tags,
     parse_tags,
+    rank_candidates,
     same_artist_cap,
+    score_candidate,
 )
 
 __all__ = [
     "RadioCollector",
     "build_like_conditions",
+    "merge_tags",
     "parse_tags",
+    "rank_candidates",
     "same_artist_cap",
+    "score_candidate",
 ]

core/radio/selection.py

@@ -13,7 +13,9 @@ get back decisions.
 
 from __future__ import annotations
 
+import hashlib
 import json
+import math
 from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
 
 
@@ -119,21 +121,102 @@ class RadioCollector:
         """How many more tracks are needed to hit the limit."""
         return max(0, self.limit - len(self._collected))
 
-    def collect(self, rows: Iterable[Dict[str, Any]], cap: Optional[int] = None) -> bool:
+    def collect(
+        self,
+        rows: Iterable[Dict[str, Any]],
+        cap: Optional[int] = None,
+        *,
+        rank: bool = False,
+    ) -> bool:
         """Append ``rows`` (dict-like) to the result, skipping already-seen IDs.
 
         ``cap`` bounds how many THIS call may add (on top of what's already
         collected); ``None`` means bounded only by the overall limit. Returns
         True once the overall limit is reached. Mirrors the original
         ``_collect`` closure exactly.
+
+        When ``rank`` is True the (still-unseen) rows are scored and sorted by
+        :func:`rank_candidates` before being appended — so a tier hands the DB
+        a generous random pool and this picks the best of it (Phase 2 smart
+        radio). When False the rows are taken in the order given (the original
+        behavior, preserved for any caller that wants it).
         """
         target = min(self.limit, len(self._collected) + cap) if cap else self.limit
-        for row in rows:
+        fresh = [dict(r) for r in rows if str(r["id"]) not in self._seen]
-            r = dict(row)
+        if rank:
+            fresh = rank_candidates(fresh)
+        for r in fresh:
             rid = str(r["id"])
-            if rid not in self._seen:
+            if rid in self._seen:
-                self._seen.add(rid)
+                continue
-                self._collected.append(r)
+            self._seen.add(rid)
-                if len(self._collected) >= target:
+            self._collected.append(r)
-                    return True
+            if len(self._collected) >= target:
+                return True
         return self.filled
+
+
+# ── Phase 2: smart ranking ─────────────────────────────────────────────────
+#
+# The old radio was pure ``ORDER BY RANDOM()``. We now fetch a generous random
+# POOL per tier (so the SQL stays cheap and varied) and rank it here by a
+# weighted score. All the intelligence is in these pure functions so it's
+# unit-testable and tunable without touching SQL.
+
+# Weights are deliberately modest so popularity guides but doesn't dominate —
+# radio should still surface lesser-played tracks, just less often.
+_W_PLAY_COUNT = 1.0       # local play_count (log-damped)
+_W_LASTFM = 0.5           # lastfm_playcount (log-damped) — global popularity hint
+_W_RECENCY_PENALTY = 2.0  # subtracted for very-recently-played (avoid repeats)
+_W_JITTER = 1.0           # deterministic per-track noise so runs vary a little
+
+
+def _log_damp(value: Any) -> float:
+    """log1p of a non-negative count, 0 for missing/invalid. Damps so a track
+    with 10000 plays doesn't bury one with 50 — popularity is a nudge."""
+    try:
+        v = float(value)
+    except (TypeError, ValueError):
+        return 0.0
+    if v <= 0:
+        return 0.0
+    return math.log1p(v)
+
+
+def _stable_jitter(track_id: Any) -> float:
+    """Deterministic [0,1) pseudo-random per track id.
+
+    Math.random / Date are unavailable / nondeterministic-unfriendly here and
+    we want runs to be reproducible in tests, so derive jitter from a hash of
+    the id. Two different tracks get different jitter; the same track is stable
+    within a ranking pass.
+    """
+    h = hashlib.sha1(str(track_id).encode("utf-8")).hexdigest()
+    return int(h[:8], 16) / 0xFFFFFFFF
+
+
+def score_candidate(row: Dict[str, Any]) -> float:
+    """Weighted desirability score for a radio candidate row.
+
+    Signals (all optional — absent columns score 0, so this is safe on a DB
+    that hasn't recorded listening data yet):
+      + play_count        local plays (log-damped)
+      + lastfm_playcount  global popularity hint (log-damped)
+      - recently_played    caller-flagged repeat-risk → penalty
+      + jitter            stable per-track noise for run-to-run variety
+    """
+    score = 0.0
+    score += _W_PLAY_COUNT * _log_damp(row.get("play_count"))
+    score += _W_LASTFM * _log_damp(row.get("lastfm_playcount"))
+    if row.get("_recently_played"):
+        score -= _W_RECENCY_PENALTY
+    score += _W_JITTER * _stable_jitter(row.get("id"))
+    return score
+
+
+def rank_candidates(rows: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """Sort candidate rows best-first by :func:`score_candidate`.
+
+    Stable sort; ties keep input order. Does not mutate the input rows.
+    """
+    return sorted(rows, key=score_candidate, reverse=True)

database/music_database.py

@@ -12815,10 +12815,34 @@ class MusicDatabase:
                     exclude_seed.extend(str(eid) for eid in exclude_ids)
                 collector = RadioCollector(limit, exclude_ids=exclude_seed)
 
-                _track_select = """
+                # Phase 2 smart radio: each tier pulls a generous RANDOM pool,
+                # then core.radio.selection ranks it (play_count + lastfm
+                # popularity, recency penalty, stable jitter) and the collector
+                # keeps the best. Pool factor keeps SQL cheap while giving the
+                # ranker real choice; bumped, then floored so small tiers still
+                # over-fetch a little.
+                _POOL_FACTOR = 4
+
+                def _pool(n):
+                    return max(n * _POOL_FACTOR, n + 10)
+
+                # Ranking signals (play_count / lastfm_playcount) are added by a
+                # migration, but probe for them so radio still works on a DB that
+                # predates it — the ranker treats missing columns as score 0, so
+                # we simply omit them from the SELECT when absent rather than
+                # crashing on "no such column".
+                cursor.execute("PRAGMA table_info(tracks)")
+                _track_cols = {row[1] for row in cursor.fetchall()}
+                _rank_cols = "".join(
+                    f"t.{c}, " for c in ("play_count", "lastfm_playcount")
+                    if c in _track_cols
+                )
+
+                _track_select = f"""
                     SELECT t.id, t.title, t.track_number, t.duration,
                            t.file_path, t.bitrate,
                            t.album_id, t.artist_id,
+                           {_rank_cols}
                            al.title   AS album,
                            COALESCE(al.thumb_url, ar.thumb_url) AS image_url,
                            ar.name    AS artist
@@ -12836,8 +12860,8 @@ class MusicDatabase:
                     WHERE {_file_filter} AND ar.name = ? AND t.album_id != ? AND t.id NOT IN ({collector.exclude_placeholders()})
                     ORDER BY RANDOM()
                     LIMIT ?
-                """, [artist_name, seed['album_id']] + collector.exclude_values() + [artist_cap])
+                """, [artist_name, seed['album_id']] + collector.exclude_values() + [_pool(artist_cap)])
-                collector.collect(cursor.fetchall(), cap=artist_cap)
+                collector.collect(cursor.fetchall(), cap=artist_cap, rank=True)
 
                 if collector.filled:
                     return {'success': True, 'tracks': collector.tracks}
@@ -12858,8 +12882,8 @@ class MusicDatabase:
                           AND t.id NOT IN ({collector.exclude_placeholders()})
                         ORDER BY RANDOM()
                         LIMIT ?
-                    """, genre_params + [artist_name] + collector.exclude_values() + [collector.remaining()])
+                    """, genre_params + [artist_name] + collector.exclude_values() + [_pool(collector.remaining())])
-                    if collector.collect(cursor.fetchall()):
+                    if collector.collect(cursor.fetchall(), rank=True):
                         return {'success': True, 'tracks': collector.tracks}
 
                 # --- 3. Same mood / style (album + artist level) ---
@@ -12879,19 +12903,20 @@ class MusicDatabase:
                               AND t.id NOT IN ({collector.exclude_placeholders()})
                             ORDER BY RANDOM()
                             LIMIT ?
-                        """, tag_params + [artist_name] + collector.exclude_values() + [collector.remaining()])
+                        """, tag_params + [artist_name] + collector.exclude_values() + [_pool(collector.remaining())])
-                        if collector.collect(cursor.fetchall()):
+                        if collector.collect(cursor.fetchall(), rank=True):
                             return {'success': True, 'tracks': collector.tracks}
 
-                # --- 4. Random library tracks ---
+                # --- 4. Random library tracks (ranked: popular-but-unheard
+                # beats pure noise even in the last-resort tier) ---
                 if not collector.filled:
                     cursor.execute(f"""
                         {_track_select}
                         WHERE {_file_filter} AND t.id NOT IN ({collector.exclude_placeholders()})
                         ORDER BY RANDOM()
                         LIMIT ?
-                    """, collector.exclude_values() + [collector.remaining()])
+                    """, collector.exclude_values() + [_pool(collector.remaining())])
-                    collector.collect(cursor.fetchall())
+                    collector.collect(cursor.fetchall(), rank=True)
 
                 return {'success': True, 'tracks': collector.tracks}
 

revamp_plan.md

@@ -8,7 +8,7 @@ Rule for every phase: kettui standard — importable/testable logic, seam-level
 
 ## Phase 0 — Make it provable (foundation, no user-visible change)
 
-- [ ] **0a. Extract radio selection logic into testable `core/radio/`.** The algorithm (tier orchestration, cap math, dedup, tag parsing, SQL-condition building) is currently tangled with `cursor.execute` inside `database/music_database.py:get_radio_tracks` (~12756) — untestable without a live DB. Pull the pure decisions into `core/radio/selection.py`; the DB method keeps SQL execution but delegates the decisions. Differential-test: same inputs → same output as today.
+- [x] **0a. Extract radio selection logic into testable `core/radio/`.** DONE (commit cbc001e2). `core/radio/selection.py` owns parse_tags/merge_tags/same_artist_cap/build_like_conditions/RadioCollector; DB method delegates. 29 tests, refactor-equivalence proven (behavioral tests pass against old AND new).
 - [ ] **0b. Centralize frontend player state.** ~10 scattered `np*` globals in `media-player.js` → one `PlayerState` object. Seam for every later frontend phase. No behavior change.
 
 ## Phase 1 — Polish / feel (frontend)
@@ -22,7 +22,8 @@ Rule for every phase: kettui standard — importable/testable logic, seam-level
 
 ## Phase 2 — Smart radio (backend algorithm)
 
-- [ ] Replace `ORDER BY RANDOM()` with real seeding: play-count + recency weighting, genre-adjacency, recently-played memory. Slots into the Phase-0a pure module → fully unit-testable (seed → expected ordering). Both radio buttons benefit (shared function).
+- [x] **Weighted ranking** DONE. Each tier now fetches a random POOL (4x, floored) and `core/radio/selection.rank_candidates` orders it by `score_candidate`: play_count + lastfm_playcount (log-damped), recently-played penalty, stable per-id jitter for run variety. Defensive column-probe → still works on a DB predating the play_count/lastfm migration. 43 radio tests; ranking math is deterministic-unit-proven; DB wiring shown via decoy-pool test (probabilistic by nature — documented).
+- [ ] **Future (optional deepening):** wire `_recently_played` from `listening_history` (column + scorer support already exist; not yet populated in the query), genre-adjacency graph (currently exact-genre LIKE only).
 
 ## Phase 3 — Architecture (deepest, riskiest — listener decision lands here)
 

tests/radio/test_get_radio_tracks_db.py

@@ -91,6 +91,23 @@ def _schema(db):
             genres TEXT, mood TEXT, style TEXT, thumb_url TEXT
         )
     """)
+    cur.execute("""
+        CREATE TABLE tracks (
+            id TEXT PRIMARY KEY, album_id TEXT, artist_id TEXT,
+            title TEXT, track_number INTEGER, duration INTEGER,
+            file_path TEXT, bitrate INTEGER,
+            play_count INTEGER DEFAULT 0, lastfm_playcount INTEGER
+        )
+    """)
+    db._conn.commit()
+
+
+def _schema_no_rank_cols(db):
+    """Schema WITHOUT play_count / lastfm_playcount — proves radio still works
+    on a DB that predates the smart-ranking migration (defensive column probe)."""
+    cur = db._conn.cursor()
+    cur.execute("CREATE TABLE artists (id TEXT PRIMARY KEY, name TEXT, genres TEXT, mood TEXT, style TEXT, thumb_url TEXT)")
+    cur.execute("CREATE TABLE albums (id TEXT PRIMARY KEY, artist_id TEXT, title TEXT, genres TEXT, mood TEXT, style TEXT, thumb_url TEXT)")
     cur.execute("""
         CREATE TABLE tracks (
             id TEXT PRIMARY KEY, album_id TEXT, artist_id TEXT,
@@ -115,11 +132,11 @@ def _add_album(db, alid, aid, title, genres="", mood="", style=""):
     )
 
 
-def _add_track(db, tid, alid, aid, title, file_path="/m/x.flac"):
+def _add_track(db, tid, alid, aid, title, file_path="/m/x.flac", play_count=0):
     db._conn.execute(
-        "INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration, file_path, bitrate) "
+        "INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration, file_path, bitrate, play_count) "
-        "VALUES (?,?,?,?,?,?,?,?)",
+        "VALUES (?,?,?,?,?,?,?,?,?)",
-        (tid, alid, aid, title, 1, 200, file_path, 1000),
+        (tid, alid, aid, title, 1, 200, file_path, 1000, play_count),
     )
 
 
@@ -130,6 +147,13 @@ def db():
     return d
 
 
+@pytest.fixture
+def db_no_rank():
+    d = _InMemoryDB()
+    _schema_no_rank_cols(d)
+    return d
+
+
 def test_missing_seed_track_returns_failure(db):
     res = db.get_radio_tracks("nope", limit=10)
     assert res["success"] is False
@@ -220,3 +244,50 @@ def test_no_duplicate_ids_across_tiers(db):
     res = db.get_radio_tracks("seed", limit=10)
     ids = [t["id"] for t in res["tracks"]]
     assert ids.count("dup") == 1
+
+
+def test_smart_ranking_prefers_more_played_in_same_tier(db):
+    """Phase 2: within a tier, the ranker surfaces the heavily-played track
+    first out of the fetched pool.
+
+    Robustness note: this proves the ranking is WIRED IN end-to-end. The pool
+    factor (4x, floored) means with these few candidates the whole set is
+    fetched, so ranking is deterministic here. The deterministic guarantee of
+    the ranking *math* lives in TestRankCandidates / TestScoreCandidate (unit
+    level) — those can't pass against pre-Phase-2 code at all. We seed many
+    unplayed decoys so a pre-Phase-2 ``ORDER BY RANDOM()`` would only return
+    'hit' first by a ~1-in-N fluke, making the wiring claim meaningful."""
+    _add_artist(db, "ar1", "Artist One")
+    _add_album(db, "al1", "ar1", "Seed Album")
+    _add_album(db, "al2", "ar1", "Other Album")
+    _add_track(db, "seed", "al1", "ar1", "Seed")
+    for i in range(15):
+        _add_track(db, f"rare{i}", "al2", "ar1", f"Rarely Played {i}", play_count=0)
+    _add_track(db, "hit", "al2", "ar1", "Big Hit", play_count=5000)
+    db._conn.commit()
+
+    res = db.get_radio_tracks("seed", limit=5)
+    assert res["success"] is True
+    ids = [t["id"] for t in res["tracks"]]
+    # The heavily-played track is ranked first out of the same-artist pool.
+    assert ids[0] == "hit"
+
+
+def test_works_without_ranking_columns(db_no_rank):
+    """Defensive: a DB predating the play_count/lastfm migration must still
+    return radio tracks (column probe omits the missing fields)."""
+    _add_artist(db_no_rank, "ar1", "Artist One")
+    _add_album(db_no_rank, "al1", "ar1", "Album A")
+    _add_album(db_no_rank, "al2", "ar1", "Album B")
+    # _add_track inserts play_count, so insert directly without it here.
+    db_no_rank._conn.execute(
+        "INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration, file_path, bitrate) "
+        "VALUES (?,?,?,?,?,?,?,?)", ("seed", "al1", "ar1", "Seed", 1, 200, "/m/s.flac", 1000))
+    db_no_rank._conn.execute(
+        "INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration, file_path, bitrate) "
+        "VALUES (?,?,?,?,?,?,?,?)", ("t2", "al2", "ar1", "Other", 1, 200, "/m/t2.flac", 1000))
+    db_no_rank._conn.commit()
+
+    res = db_no_rank.get_radio_tracks("seed", limit=10)
+    assert res["success"] is True
+    assert "t2" in [t["id"] for t in res["tracks"]]

tests/radio/test_selection.py

@@ -12,7 +12,9 @@ from core.radio.selection import (
     build_like_conditions,
     merge_tags,
     parse_tags,
+    rank_candidates,
     same_artist_cap,
+    score_candidate,
 )
 
 
@@ -137,3 +139,82 @@ class TestRadioCollector:
         c.collect(self._rows("c"))
         assert c.exclude_placeholders() == "?,?,?"
         assert set(c.exclude_values()) == {"a", "b", "c"}
+
+    def test_ranked_collect_prefers_high_play_count(self):
+        # Pool given in worst-first order; rank=True should reorder so the
+        # most-played track is collected first.
+        c = RadioCollector(limit=2)
+        pool = [
+            {"id": 1, "play_count": 0},
+            {"id": 2, "play_count": 500},
+            {"id": 3, "play_count": 50},
+        ]
+        c.collect(pool, rank=True)
+        assert [t["id"] for t in c.tracks] == [2, 3]   # 500 then 50, 0 dropped at limit
+
+
+# ── Phase 2: smart ranking ─────────────────────────────────────────────────
+
+class TestScoreCandidate:
+    def test_missing_signals_score_is_pure_jitter(self):
+        # No play data → score is just the stable jitter, in [0, 1).
+        s = score_candidate({"id": "x"})
+        assert 0.0 <= s < 1.0
+
+    def test_higher_play_count_scores_higher(self):
+        low = score_candidate({"id": "same", "play_count": 1})
+        high = score_candidate({"id": "same", "play_count": 1000})
+        assert high > low   # same id → same jitter, so play_count decides
+
+    def test_lastfm_contributes(self):
+        base = score_candidate({"id": "same"})
+        with_lastfm = score_candidate({"id": "same", "lastfm_playcount": 100000})
+        assert with_lastfm > base
+
+    def test_recently_played_is_penalized(self):
+        normal = score_candidate({"id": "same", "play_count": 10})
+        recent = score_candidate({"id": "same", "play_count": 10, "_recently_played": True})
+        assert recent < normal
+
+    def test_invalid_counts_treated_as_zero(self):
+        # Garbage values must not crash; they score as 0 (jitter only).
+        s = score_candidate({"id": "x", "play_count": None, "lastfm_playcount": "n/a"})
+        assert 0.0 <= s < 1.0
+
+    def test_jitter_is_stable_per_id(self):
+        a = score_candidate({"id": "track-42"})
+        b = score_candidate({"id": "track-42"})
+        assert a == b   # deterministic — reproducible runs/tests
+
+    def test_jitter_differs_between_ids(self):
+        a = score_candidate({"id": "track-1"})
+        b = score_candidate({"id": "track-2"})
+        assert a != b
+
+
+class TestRankCandidates:
+    def test_orders_best_first(self):
+        rows = [
+            {"id": 1, "play_count": 0},
+            {"id": 2, "play_count": 1000},
+            {"id": 3, "play_count": 100},
+        ]
+        ranked = rank_candidates(rows)
+        assert [r["id"] for r in ranked] == [2, 3, 1]
+
+    def test_does_not_mutate_input(self):
+        rows = [{"id": 1, "play_count": 0}, {"id": 2, "play_count": 9}]
+        original = list(rows)
+        rank_candidates(rows)
+        assert rows == original
+
+    def test_empty(self):
+        assert rank_candidates([]) == []
+
+    def test_popularity_beats_jitter_at_scale(self):
+        # A heavily-played track must always outrank an unplayed one regardless
+        # of jitter (jitter is bounded to [0,1), play_count is log-scaled * 1.0).
+        pool = [{"id": f"unplayed-{i}", "play_count": 0} for i in range(20)]
+        pool.append({"id": "hit", "play_count": 5000})
+        ranked = rank_candidates(pool)
+        assert ranked[0]["id"] == "hit"