splatter-tennis/assets/_vidroll.py

# _vidroll.py — a run cycle out of a VIDEO model. One still frame in, one stride's worth of
# sprites out.
#
#   uv run --with pillow --with numpy --with requests python assets/_vidroll.py pf --film
#   uv run --with pillow --with numpy --with requests python assets/_vidroll.py pf --cut
#   uv run --with pillow --with numpy --with requests python assets/_vidroll.py pf --cut --install
#   keys: p (player m), pf (player f)   [the opponents run two-frame pairs and do not need this]
#
# WHY THIS EXISTS, and why the still pipeline does not do it. `_runroll.py` asks one image model for
# a ROW of run poses, and across six sessions it never once returned a real gait: it draws three
# phases and repeats them with the same leg leading, which Oliver called "ropeskipping, not running"
# — amplitude wiggle, no swap. Words did not fix it (a frame-by-frame LEFT/RIGHT skeleton spec, a
# closing checklist, a loop clause: all of it lost). A VIDEO model has no such problem, because it
# is not drawing poses, it is moving a body through time — the legs alternate because that is what
# legs do. One 5-second clip is one guaranteed-coherent stride, and it costs about $0.42.
#
# THE RULES THAT MADE IT WORK (each one paid for with a bad clip):
#   * PIN THE POSE AT BOTH ENDS. first_frame AND last_frame set to the same still. Her first clip,
#     pinned only at the front, rotated her into a side profile by the second half.
#   * The prompt says the camera never moves and the runner never turns. Say both.
#   * MEASURE THE STRIDE, do not eyeball it: `--cut` scores every frame against every other with
#     _famref.legs/stride_diff and takes the period with the lowest self-difference, then samples
#     N phases evenly across ONE period. Her clip strides in 6 frames at 16 fps, his in 23 — a
#     fixed "every fifth frame" gave a cycle that read as a stagger.
#   * The frames go through `_stride.cut`, the same palette-snap-to-settled-art the leg-swap twin
#     uses. Raw clip frames are soft and their racket beds fill in; the cut is what makes them
#     sprites. Whole frames beat splicing: an earlier pass kept only the clip's legs under his own
#     drawn torso, and matching hips frame by frame was a week of seams (her skirt made it worse).
#
# THE CLIP'S FIRST FRAME IS THE WHOLE JOB. Generate it as a single still first — for her, one
# `_pose_ref_run_*`-style identity reference plus HIS approved run frame as the pose picture ("she
# is the person; his pose is the instruction"), which landed on roll one. A crisp first frame with
# an open racket bed gives a clean clip; a soft one gives a clip whose racket wobbles.
import base64, json, os, sys, time

import requests
from PIL import Image

HERE = os.path.dirname(os.path.abspath(__file__))
SCRATCH = os.path.join(HERE, '_refcheck')
MODEL = os.environ.get('SPLAT_VIDEO_MODEL', 'kwaivgi/kling-v3.0-std')
FPS = 16          # extraction rate: fine enough to find the period, coarse enough to score quickly
N = 6             # phases installed — one stride, evenly sampled

CHARS = {
 'p':  {'still': 'p_run1', 'like': 'p_idle1', 'install': ('p_run', ''),  'who': 'The man',
        'they': ('he', 'his', 'him')},
 'pf': {'still': 'p_run1_f', 'like': 'p_idle1_f', 'install': ('p_run', '_f'), 'who': 'The woman',
        'they': ('she', 'her', 'her')},
}


def key():
    if os.environ.get('OPENROUTER_API_KEY'): return os.environ['OPENROUTER_API_KEY']
    for line in open(os.path.join(HERE, '..', '.env')):
        if line.startswith('OPENROUTER_API_KEY='):
            return line.split('=', 1)[1].strip().strip('"').strip("'")
    sys.exit('no OPENROUTER_API_KEY')


def prompt(k):
    c = CHARS[k]; s, poss, obj = c['they']
    return (
     f'Pixel-art video game sprite animation. {c["who"]} in the image keeps RUNNING AWAY from the '
     f'camera, seen from behind, travelling slightly to the right. Two full running strides: '
     f'{poss} legs alternate clearly, left knee driving up then right knee driving up, the trailing '
     f'foot kicking up behind {obj} each time. {s.capitalize()} grips the wooden tennis racket in '
     f'{poss} right hand the whole time and never lets go. {s.upper()} NEVER TURNS: the camera '
     f'stays behind {obj} for every single frame, {poss} face is never shown, {s} never rotates '
     f'into a side profile or toward the camera. THE CAMERA IS COMPLETELY STATIC — no pan, no zoom, '
     f'no dolly. {s.capitalize()} stays in the middle of the frame at the same size. The background '
     'stays ONE FLAT SOLID MAGENTA #FF00FF, empty — no court, no scenery, no shadow under '
     f'{poss} feet, no ground line, no contact shadow. The drawing style stays hard-edged pixel art '
     'the whole time.')


def framed(src, out, size=(1024, 1024), scale=3):
    """The sprite on flat magenta, big enough for a video model to see the pixels."""
    im = Image.open(src).convert('RGBA')
    im = im.resize((im.width * scale, im.height * scale), Image.NEAREST)
    c = Image.new('RGB', size, (255, 0, 255))
    c.paste(im, ((size[0] - im.width) // 2, size[1] - im.height - 60), im)
    c.save(out)
    return out


def u(p):
    return 'data:image/png;base64,' + base64.b64encode(open(p, 'rb').read()).decode()


def film(k):
    os.makedirs(SCRATCH, exist_ok=True)
    first = framed(os.path.join(HERE, CHARS[k]['still'] + '.png'), os.path.join(SCRATCH, f'vid_{k}_first.png'))
    img = {'type': 'image_url', 'image_url': {'url': u(first)}}
    r = requests.post('https://openrouter.ai/api/v1/videos',
        headers={'Authorization': f'Bearer {key()}', 'Content-Type': 'application/json'},
        json={'model': MODEL, 'prompt': prompt(k), 'duration': 5, 'aspect_ratio': '1:1',
              'generate_audio': False,
              # BOTH ends pinned to the same still: that is what holds the camera behind the runner
              'frame_images': [dict(img, frame_type='first_frame'), dict(img, frame_type='last_frame')]},
        timeout=600)
    if r.status_code >= 400: sys.exit(f'{k}: HTTP {r.status_code} {r.text[:300]}')
    job = r.json()
    poll = job.get('polling_url') or f"https://openrouter.ai/api/v1/videos/{job['id']}"
    for _ in range(120):
        time.sleep(10)
        p = requests.get(poll, headers={'Authorization': f'Bearer {key()}'}, timeout=120).json()
        print(' ', p.get('status'), flush=True)
        if p.get('status') in ('completed', 'succeeded', 'success'):
            url = (p.get('unsigned_urls') or [None])[0]
            # the "unsigned" url still wants the key — without it you save a 67-byte 401 body
            v = requests.get(url, headers={'Authorization': f'Bearer {key()}'}, timeout=600).content
            out = os.path.join(SCRATCH, f'vid_{k}.mp4')
            open(out, 'wb').write(v)
            print(f'{k}: {out} {len(v)} bytes   cost {p.get("usage", {}).get("cost")}')
            return
        if p.get('status') in ('failed', 'error', 'cancelled'):
            sys.exit(json.dumps(p)[:400])
    sys.exit(f'{k}: still pending after 20 minutes')


def cut(k, install=False):
    import subprocess
    sys.path.insert(0, HERE)
    import _stride
    from _famref import legs, stride_diff
    mp4 = os.path.join(SCRATCH, f'vid_{k}.mp4')
    if not os.path.exists(mp4): sys.exit(f'{mp4}: no clip — run `_vidroll.py {k} --film` first')
    d = os.path.join(SCRATCH, f'vidf_{k}')
    subprocess.run(['rm', '-rf', d], check=False); os.makedirs(d)
    subprocess.run(['ffmpeg', '-v', 'error', '-i', mp4, '-vf', f'fps={FPS}', f'{d}/f%03d.png'], check=True)
    fs = sorted(os.listdir(d))[6:46]                 # skip the held first frames
    ims = []
    for f in fs:
        im, err = _stride.cut(os.path.join(d, f), CHARS[k]['like'])
        if im: ims.append(im)
    if len(ims) < 12: sys.exit(f'{k}: only {len(ims)} frames survived the cut')
    L = [legs(i) for i in ims]
    # THE PERIOD IS MEASURED, NOT GUESSED: the lag whose frames look most like the frames one lag
    # earlier IS the stride. (A run that lopes and one that snaps differ by 4x — see the header.)
    scores = [(sum(stride_diff(L[i], L[i + p]) for i in range(len(L) - p)) / (len(L) - p), p)
              for p in range(4, min(26, len(L) - 2))]
    per = min(scores)[1]
    print(f'{k}: {len(ims)} frames, stride = {per} frames at {FPS} fps ({per / FPS:.2f} s)')
    sel = [round(i * per / N) for i in range(N)]
    pre, suf = CHARS[k]['install']
    outdir = HERE if install else os.path.join(SCRATCH, f'vidcut_{k}')
    os.makedirs(outdir, exist_ok=True)
    import numpy as np
    for n, i in enumerate(sel, 1):
        im = ims[i]
        # feet-centred like every other run cut: drawChar centres a frame by its total WIDTH, and
        # the racket's reach varies across a cycle, so an uncentred cycle slides sideways per frame
        a = np.array(im); op = a[..., 3] > 40
        ys, xs = np.nonzero(op[int(im.height * 0.66):])
        off = round(2 * (xs.mean() - im.width / 2)) if len(xs) else 0
        if off:
            p = Image.new('RGBA', (im.width + abs(off), im.height), (0, 0, 0, 0))
            p.paste(im, (0 if off > 0 else abs(off), 0)); im = p
        im.save(os.path.join(outdir, f'{pre}{n}{suf}.png'))
    print(f'{k}: {N} frames -> {outdir}' + ('' if install else '  (review, then --install)'))
    print('   BUMP SPRV in render.js, or the year-long asset cache serves the old frames.')


if __name__ == '__main__':
    args = [a for a in sys.argv[1:] if not a.startswith('-')]
    if len(args) != 1 or args[0] not in CHARS:
        sys.exit(f'usage: _vidroll.py <key> [--film|--cut] [--install]   keys: {" ".join(CHARS)}')
    k = args[0]
    if '--film' in sys.argv: film(k)
    elif '--cut' in sys.argv: cut(k, install='--install' in sys.argv)
    else: sys.exit('pass --film or --cut')