python/elidedb/motion.py

"""The motion vector channel: motion = temporal DIFFERENCE of appearance.

The Marengo lesson is that motion must be its own vector, never averaged
into appearance. The obvious implementations were measured and lost:
  X-CLIP (video-native, cross-frame attention)    close/open AUC 0.67
  VLM swap-contrast (query-time, 2 model calls)   0.86 (2B) / 0.91 (7B)
  ctx adapter (caption-trained)                   ~chance
  THIS: delta-appearance + query-swap difference  0.98 / 0.98

An event's motion vector is normalize(app(end) - app(start)) in SigLIP
space, from the per-frame vectors ALREADY on disk — ingest cost is a
numpy mean. A directional query's motion vector is
normalize(embed(q) - embed(swap(q))) — the same swap-contrast idea, moved
from VLM logits into embedding space, so it runs index-only in
microseconds. Both sides subtract their shared appearance component, so
what remains on each side is the DIRECTION of state change, and they
were trained (by SigLIP, incidentally) to point the same way.

Limits, measured: works when the state change is visually large
(drawer open<->close: 0.98); chance when a small object moves
(put-in/take-out: 0.46-0.50) — those queries are carried by the
appearance and caption-lexical channels instead. Non-directional queries
have no swap and the channel ABSTAINS (RRF median rank, no veto).
"""
from __future__ import annotations

import numpy as np
import pyarrow as pa


def index_motion(store, events_table="context_events", edge_s=1.5,
                 frames_table="frame_vectors"):
    """One motion vector per event span -> `motion_vectors` table.

    Event spans come from the gate-segmented event index so every retrieval
    unit carries both an appearance-family vector and a motion vector —
    the multi-vector layout, aligned. Rebuild = atomic replace commit.
    """
    import time
    import uuid

    from .embeddings import _vec_table
    from .log import FileEntry
    from .store import write_parquet
    t_start = time.time()

    ev = store.table(events_table).scan()
    fv, fvecs = _vec_table(store, frames_table)
    fs = np.asarray(fv.column("stream").to_pylist())
    ft = np.asarray([int(v) for v in fv.column("ts").to_pylist()])
    edge = int(edge_s * 1e9)

    # per-stream sorted timelines + searchsorted per event. The full-array
    # boolean-mask version was O(events x frames): at 100 h that is 50k
    # events over 1.8M frames = 16.5 MINUTES for what is 25 s of means.
    by_stream = {}
    for s in np.unique(fs):
        m = np.where(fs == s)[0]
        o = np.argsort(ft[m], kind="stable")
        by_stream[s] = (ft[m][o], fvecs[m[o]])

    rows_s, rows_a, rows_b, rows_v = [], [], [], []
    for s, a, b in zip(ev.column("stream").to_pylist(),
                       (int(v) for v in ev.column("ts").to_pylist()),
                       (int(v) for v in ev.column("t1").to_pylist())):
        if s not in by_stream:
            continue
        tt, vv = by_stream[s]
        lo, hi = np.searchsorted(tt, [a, b + 1])
        if hi - lo < 6:
            continue
        t0, t1 = int(tt[lo]), int(tt[hi - 1])
        e0 = np.searchsorted(tt, t0 + edge, side="right")
        e1 = np.searchsorted(tt, t1 - edge, side="left")
        v0 = vv[lo:max(e0, lo + 1)].mean(0)
        v1 = vv[min(e1, hi - 1):hi].mean(0)
        d = v1 - v0
        n = float(np.linalg.norm(d))
        if n < 1e-6:
            continue
        rows_s.append(s)
        rows_a.append(a)
        rows_b.append(b)
        rows_v.append((d / n).astype(np.float32))
    if not rows_v:
        return {"events": 0}

    dim = len(rows_v[0])
    tbl = pa.table({
        "ts": pa.array(rows_a, pa.int64()),
        "t1": pa.array(rows_b, pa.int64()),
        "stream": pa.array(rows_s),
        "vector": pa.array([v.tolist() for v in rows_v],
                           pa.list_(pa.float32(), dim)),
    })
    tab = store.table("motion_vectors")
    meta = {"model": "delta-appearance", "dim": dim, "edge_s": edge_s,
            "events_table": events_table}
    try:
        existing = tab.state().files
    except Exception:
        existing = []
    if existing:
        fname = f"part-{uuid.uuid4().hex[:12]}.parquet"
        path = tab.dir / fname
        write_parquet(tbl, path)
        tsv = tbl.column("ts")
        version = tab.log.commit(
            op="replace", kind="embeddings", schema=str(tbl.schema),
            add=[FileEntry(fname, len(tbl), path.stat().st_size,
                           tsv[0].as_py(), tsv[-1].as_py())],
            remove=[f.path for f in existing], meta=meta)
    else:
        version = tab.append(tbl, kind="embeddings", meta=meta)
    return {"events": len(tbl), "dim": dim, "version": version,
            "seconds": round(time.time() - t_start, 2)}


def motion_scores(store, qv, qv_swap):
    """(stream, t0, t1, score) per event: motion vectors against the
    query-direction vector. Pure matmul over a version-cached matrix."""
    from .embeddings import _vec_table
    tbl, vecs = _vec_table(store, "motion_vectors")
    qd = np.asarray(qv, np.float32) - np.asarray(qv_swap, np.float32)
    n = float(np.linalg.norm(qd))
    if n < 1e-6:
        return []
    sc = vecs @ (qd / n)
    return list(zip(tbl.column("stream").to_pylist(),
                    (int(v) for v in tbl.column("ts").to_pylist()),
                    (int(v) for v in tbl.column("t1").to_pylist()),
                    sc.tolist()))


_MOT_IDX = {}


def motion_lookup(store, qv, qv_swap):
    """lookup(stream, t0, t1) -> score of the recording CONTAINING the span,
    or nan. The per-stream sorted span index is cached per table version;
    a query pays one matmul + searchsorted per segment. The Python-dict
    scan this replaces cost ~70 ms per directional query at 50k recordings
    (measured at the 100 h scale test)."""
    from .embeddings import _vec_table
    ver = store.table("motion_vectors").state().version
    key = (str(store.dir), ver)
    if key not in _MOT_IDX:
        tbl, _ = _vec_table(store, "motion_vectors")
        ss = np.asarray(tbl.column("stream").to_pylist())
        sa = np.asarray([int(v) for v in tbl.column("ts").to_pylist()])
        sb = np.asarray([int(v) for v in tbl.column("t1").to_pylist()])
        idx = {}
        for s in np.unique(ss):
            m = np.where(ss == s)[0]
            o = np.argsort(sa[m], kind="stable")
            idx[s] = (sa[m][o], sb[m][o], m[o])
        if len(_MOT_IDX) > 8:
            _MOT_IDX.clear()
        _MOT_IDX[key] = idx
    idx = _MOT_IDX[key]
    _, vecs = _vec_table(store, "motion_vectors")
    qd = np.asarray(qv, np.float32) - np.asarray(qv_swap, np.float32)
    n = float(np.linalg.norm(qd))
    if n < 1e-6:
        return lambda s, a, b: float("nan")
    sc = vecs @ (qd / n)

    def lookup(s, a, b):
        if s not in idx:
            return float("nan")
        t0s, t1s, rows = idx[s]
        j = int(np.searchsorted(t0s, a, side="right")) - 1
        if j >= 0 and b <= int(t1s[j]) + 1:
            return float(sc[rows[j]])
        return float("nan")
    return lookup


def motion_candidates(store, qv, qv_swap, streams, t0s, top=24):
    """DIRECTION-FIRST RECALL, content-gated. Given a broad appearance-
    plausible window set (their streams + start times), return the top
    recordings ranked by motion score as (stream, rec_t0, rec_t1, score).

    Why the gate: unrestricted motion recall was measured useless — the
    tiny direction cosines (~0.07) drown corpus-wide in random tabletop
    deltas. Why recall at all: at the 100 h scale, appearance ordering
    alone never surfaces true close/open recordings into the candidate
    pool (put-in clips carry a stronger 'drawer' signal), so ranking-only
    motion had nothing correct to rank. Appearance answers WHAT is
    plausible; motion picks WHICH of those changed the right way."""
    from .embeddings import _vec_table
    ver = store.table("motion_vectors").state().version
    key = (str(store.dir), ver)
    if key not in _MOT_IDX:
        motion_lookup(store, qv, qv_swap)      # builds the cache
    idx = _MOT_IDX[key]
    _, vecs = _vec_table(store, "motion_vectors")
    qd = np.asarray(qv, np.float32) - np.asarray(qv_swap, np.float32)
    n = float(np.linalg.norm(qd))
    if n < 1e-6:
        return []
    sc = vecs @ (qd / n)

    streams = np.asarray(streams)
    t0s = np.asarray(t0s)
    hit_rows = set()
    for s in np.unique(streams):
        if s not in idx:
            continue
        r0, r1, rows = idx[s]
        w = t0s[streams == s]
        j = np.searchsorted(r0, w, side="right") - 1
        ok = j >= 0
        hit_rows.update(int(r) for r in np.unique(rows[j[ok]]))
    if not hit_rows:
        return []
    hits = sorted(hit_rows, key=lambda r: -sc[r])[:top]
    tbl, _ = _vec_table(store, "motion_vectors")
    ss = tbl.column("stream")
    sa = tbl.column("ts")
    sb = tbl.column("t1")
    return [(ss[r].as_py(), int(sa[r].as_py()), int(sb[r].as_py()),
             float(sc[r])) for r in hits]