sim_open.py

"""Can it work out which mood you are in when you open the app?

The listener here has two genuine moods and picks one each evening, the way
people actually do: something quiet late at night, something driving in the
afternoon. Averaging those two into one "usual taste" describes neither, so this
is the case that separates remembering a habit from reading a room.

Three arms, identical in every other respect:

  cold      no memory at all, every session starts from nothing
  habit     remembers where this listener usually sits, and opens there
  reading   remembers the modes, and spends the first few tracks working out
            which one is live tonight before committing
"""
from __future__ import annotations

import argparse
import tempfile
import warnings
from typing import List

import numpy as np

warnings.filterwarnings("ignore")

from ureco import config
from ureco.engine import Engine
from ureco.modes import Detection, ModeSet
from ureco.store import Store

GENRE = "129"


class TwoMoodPerson:
    def __init__(self, pool, rng):
        M = np.stack([t["features"].vec() for t in pool])
        # two real regions of this catalogue, far apart on the axes that matter
        self.moods = [np.percentile(M, 18, axis=0), np.percentile(M, 82, axis=0)]
        self.sensitivity = np.full(8, 0.3)
        self.sensitivity[[0, 1, 4]] = 2.4          # energy, tempo, brightness
        self.sensitivity /= self.sensitivity.mean()
        self.rng = rng
        self.tol = [self._cal(M, m) for m in self.moods]
        self.active = 0

    def _cal(self, M, mood):
        d = np.sqrt(((M - mood) ** 2 * self.sensitivity).mean(axis=1))
        return float(2.0 * np.percentile(d, 25))

    def new_session(self, hour_is_night: bool):
        # a habit, not a rule: quiet at night about 80% of the time
        p_quiet = 0.8 if hour_is_night else 0.2
        self.active = 0 if self.rng.random() < p_quiet else 1

    def react(self, track) -> float:
        f = track["features"].vec()
        m, tol = self.moods[self.active], self.tol[self.active]
        d = float(np.sqrt(((f - m) ** 2 * self.sensitivity).mean()))
        fit = max(0.0, 1.0 - d / tol)
        if fit + self.rng.normal(0, 0.13) >= 0.5:
            return float(np.clip(0.6 + 0.4 * fit + self.rng.normal(0, 0.08), 0.5, 1.0))
        return float(np.clip(0.05 + abs(self.rng.normal(0, 0.03)), 0.01, 0.3))


def run_session(engine, person, n, rng, night):
    person.new_session(night)
    hits, settled_at = [], None
    for i in range(n):
        q = engine.queue(limit=1)
        if not q:
            break
        t = q[0].track
        engine.start(t["id"])
        r = person.react(t)
        engine.record(t["id"], r * 30.0, 30.0)
        hits.append(r >= 0.5)
        if settled_at is None and engine.detection.settled:
            settled_at = i + 1
    return hits, settled_at


def run(seed, sessions, per_session, mode, verbose=False):
    rng = np.random.default_rng(seed)
    live = Store()
    pool = [t for t in live.analysed_tracks(4000) if str(t.get("genre_id")) == GENRE]
    live.close()
    if len(pool) < 60:
        return None

    store = Store(tempfile.mktemp(suffix=".db"))
    for t in pool:
        store.upsert_track(t, t["features"])
    person = TwoMoodPerson(pool, rng)

    rows = []
    for s in range(sessions):
        if mode == "cold":
            store.wipe_everything()
            for t in pool:
                store.upsert_track(t, t["features"])
        e = Engine(store, seed=seed * 100 + s, offline=True)
        e.lock_genre(GENRE, "Jazz")
        if mode != "reading":
            # habit arm: keep the profile, switch the mode-reading off
            e.modes = ModeSet()
            e.detection = Detection()
        night = (s % 2 == 0)
        hits, settled = run_session(e, person, per_session, rng, night)
        e.retrain()
        rows.append({
            "session": s + 1,
            "all": float(np.mean(hits)) if hits else 0.0,
            "first8": float(np.mean(hits[:8])) if len(hits) >= 8 else float("nan"),
            "after8": float(np.mean(hits[8:])) if len(hits) > 8 else float("nan"),
            "settled_at": settled,
            "n_modes": len(e.modes),
            "true_mood": person.active,
        })
        store.reset()
    return rows


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--seeds", type=int, default=10)
    ap.add_argument("--sessions", type=int, default=8)
    ap.add_argument("-n", type=int, default=36)
    a = ap.parse_args()

    arms = {}
    for mode in ("cold", "habit", "reading"):
        runs = [run(s + 1, a.sessions, a.n, mode) for s in range(a.seeds)]
        arms[mode] = [r for r in runs if r]

    def m(key, mode, lo=0):
        vals = [[r[i][key] for i in range(lo, a.sessions)] for r in arms[mode]]
        return float(np.nanmean(vals))

    print(f"a listener with two moods, {a.seeds} of them, "
          f"{a.sessions} sessions each, {a.n} tracks a session\n")
    print("  %-10s %-16s %-16s %-16s" % ("session", "cold", "habit memory", "reads the mood"))
    for i in range(a.sessions):
        row = []
        for mode in ("cold", "habit", "reading"):
            f8 = np.nanmean([r[i]["first8"] for r in arms[mode]])
            al = np.mean([r[i]["all"] for r in arms[mode]])
            row.append(f"{al:.0%}  (first8 {f8:.0%})")
        print("  %-10d %-16s %-16s %-16s" % (i + 1, *row))

    print()
    print("  %-26s %8s %8s %8s" % ("", "cold", "habit", "reading"))
    for label, key, lo in [("whole session", "all", 0), ("whole session, 3+", "all", 2),
                           ("first 8 tracks, 3+", "first8", 2),
                           ("after track 8, 3+", "after8", 2)]:
        print("  %-26s %7.0f%% %7.0f%% %7.0f%%" % (
            label, m(key, "cold", lo) * 100, m(key, "habit", lo) * 100, m(key, "reading", lo) * 100))

    settle = [r[i]["settled_at"] for r in arms["reading"]
              for i in range(2, a.sessions) if r[i]["settled_at"]]
    nmodes = [r[i]["n_modes"] for r in arms["reading"] for i in range(2, a.sessions)]
    print()
    print("  modes it found        : %.1f on average" % (np.mean(nmodes) if nmodes else 0))
    print("  tracks to settle      : %.1f (of %d in a session)"
          % (np.mean(settle) if settle else float("nan"), a.n))
    print("  reading vs habit, first 8 : %+.1f points"
          % ((m("first8", "reading", 2) - m("first8", "habit", 2)) * 100))
    print("  reading vs habit, overall : %+.1f points"
          % ((m("all", "reading", 2) - m("all", "habit", 2)) * 100))