Godstrument/timewarp.py

"""
timewarp.py — give slow data a playhead you can scratch.

Live data is stuck in the present: you can freeze or bend it, but you can't
speed it up, because it's happening *now*. Recorded data is different — it has a
timeline. This module takes a window of history (a month of earthquakes, four
days of weather) and plays it back **compressed** — 30 days into 5 minutes — on
a loop, so glacial data finally has agency inside a track.

And because it has a playhead, you can *nudge* it: push the platter forward,
drag it back, scrub. A month of the Earth's seismicity becomes a loopable,
scratchable rhythm.

Two buffer modes:
  * continuous — interpolates a value at the playhead (temperature, wind...)
  * event      — fires discrete hits as the playhead crosses them (quakes...)

Playhead is tracked as a phase in [0,1) over the loaded window. Pure stdlib.
"""

from __future__ import annotations
import bisect

NUDGE_GAIN = 2.0   # how hard a full nudge shoves the platter


class Jog:
    """Live playhead control, fed by the hub over OSC (/jog/rate, /jog/nudge)."""

    def __init__(self, rate_min: float = 0.25, rate_max: float = 4.0):
        self.rate_min, self.rate_max = rate_min, rate_max
        self.rate_mult = 1.0
        self.nudge = 0.0

    def on_rate(self, addr, *a):
        if a:
            v = max(0.0, min(1.0, float(a[0])))
            self.rate_mult = self.rate_min + (self.rate_max - self.rate_min) * v

    def on_nudge(self, addr, *a):
        if a:
            self.nudge = (max(0.0, min(1.0, float(a[0]))) - 0.5) * 2.0


class ReplayBuffer:
    def __init__(self, mode: str = "continuous", into_seconds: float = 300.0):
        self.mode = mode
        self.into = max(1.0, into_seconds)   # real seconds for one full loop
        self.phase = 0.0
        # continuous
        self._phases: list[float] = []
        self._vals: list[float] = []
        # event: parallel sorted lists
        self._ev_phase: list[float] = []
        self._ev_payload: list[dict] = []
        self.loaded = False

    # ---- loading -------------------------------------------------------
    def load_continuous(self, times: list[float], values: list[float]):
        pairs = sorted((t, v) for t, v in zip(times, values) if v is not None)
        if len(pairs) < 2:
            return
        t0, t1 = pairs[0][0], pairs[-1][0]
        span = (t1 - t0) or 1.0
        self._phases = [(t - t0) / span for t, _ in pairs]
        self._vals = [float(v) for _, v in pairs]
        self.loaded = True

    def load_events(self, events: list[dict], t_key: str = "t"):
        """events: dicts each with a time under t_key; extra fields = payload."""
        evs = sorted(events, key=lambda e: e[t_key])
        if len(evs) < 1:
            return
        t0, t1 = evs[0][t_key], evs[-1][t_key]
        span = (t1 - t0) or 1.0
        self._ev_phase = [(e[t_key] - t0) / span for e in evs]
        self._ev_payload = evs
        self.loaded = True

    # ---- playback ------------------------------------------------------
    def tick(self, dt: float, rate_mult: float = 1.0, nudge: float = 0.0):
        """Advance the playhead. rate_mult scales base speed; nudge is bipolar
        (-1..1), a momentary platter shove. Returns crossed event payloads
        (event mode) or None (continuous)."""
        if not self.loaded:
            return [] if self.mode == "event" else None
        eff = rate_mult + nudge * NUDGE_GAIN          # effective multiplier
        dphase = (eff / self.into) * dt               # one loop = self.into sec
        # clamp a runaway single-step to at most one full loop
        if dphase > 1.0:
            dphase = 1.0
        elif dphase < -1.0:
            dphase = -1.0
        p0 = self.phase
        crossed = self._crossed(p0, dphase) if self.mode == "event" else None
        self.phase = (p0 + dphase) % 1.0
        return crossed

    def current(self) -> float:
        """Interpolated value at the playhead (continuous mode)."""
        if not self.loaded or not self._phases:
            return 0.0
        ph = self.phase
        ps = self._phases
        i = bisect.bisect_right(ps, ph)
        if i == 0 or i >= len(ps):
            # wrap segment between last and first sample
            a, b = ps[-1], ps[0] + 1.0
            va, vb = self._vals[-1], self._vals[0]
            x = ph if ph >= ps[-1] else ph + 1.0
        else:
            a, b = ps[i - 1], ps[i]
            va, vb = self._vals[i - 1], self._vals[i]
            x = ph
        f = (x - a) / (b - a) if b != a else 0.0
        return va + (vb - va) * f

    def _crossed(self, p0: float, dphase: float) -> list[dict]:
        if dphase == 0.0:
            return []
        out = []
        end = p0 + dphase
        for ep, payload in zip(self._ev_phase, self._ev_payload):
            for cand in (ep - 1.0, ep, ep + 1.0):
                if dphase > 0 and p0 < cand <= end:
                    out.append(payload)
                elif dphase < 0 and end <= cand < p0:
                    out.append(payload)
        return out