#!/usr/bin/env python3 # ═════════════════════════════════════════════════════════════════════════════ # PLAYER COMPUTER — Ten Thousand Ducks (06/32) # by Gene Kogan · 2026 · https://genekogan.com/player_computer/ten_thousand_ducks # # A sea shanty for the 1992 container spill, and the one duck that made it home. # # This single file IS the piece: it draws every frame, synthesizes every sound, # and muxes them into the final video with ffmpeg. No other project files are # needed. You (or your agent) are invited to make a VARIATION of it: # # Generate a variation of this music video using only code. # Start from https://genekogan.com/player_computer/code/ten_thousand_ducks.py.txt # # The original render (for reference, yours should differ): # video: https://genekogan.com/player_computer/media/ten_thousand_ducks.mp4 # cover: https://genekogan.com/player_computer/media/ten_thousand_ducks.jpg # # Requirements: python3, numpy, pillow, and ffmpeg on PATH. # pip install numpy "pillow<13" # Speech/vocals (in pieces that have them) use the macOS `say` command; on # other platforms swap in espeak-ng / any TTS at the say_wav()/speak() calls, # or mute those lines. git provenance stamps degrade gracefully outside a repo. # Run: python3 ten_thousand_ducks.py (writes frames/, audio/, and the final mp4 # next to the script; writes ~8 GB of frames, # takes 5-15 min on a modern machine) # ═════════════════════════════════════════════════════════════════════════════ """ player_computer_final — "TEN THOUSAND DUCKS" (recut of second_wind/ten_thousand_ducks) Sea shanty, 96bpm, A minor, call-and-response. 26 bars (~67s). Sung + subtitled. Intro(1) V1(4) Chorus(6) V2(4) V3(4) Chorus(6) Out(1) Recut vs the second_wind original (46 bars / 1:58): each verse cut from four lines to its strongest couplet — V1 keeps the storm wave and the spill, V2 keeps the currents couplet (gyre / floes / seas where only the duck goes), V3 keeps the doorstep payoff. One chorus dropped (the hook still lands twice: after the spill, and as the finale once the duck is home). Shot order is now sequential per section so the storm always precedes the spill and the piece closes map -> doorstep -> medallion. Same music, same look, same lines that survive — nothing sped up, nothing truncated mid-thought. The true-ish story of the 1992 container spill: a storm takes a crate of 28,800 rubber ducks off a cargo ship in the North Pacific, and the ducks spend the next twenty years riding the currents — the gyre, the Arctic ice, beaches on three continents — until one of them, salt-faded and chipped, washes up on the retired sailor's own doorstep. A shanty because cargo lost at sea deserves one; a happy one because the sea returns it all, eventually. Look: scrimshaw. Fine sepia engraving on whalebone ivory — hatched sails and waves, a dotted route drawing itself across an engraved world map, banner ribbons carrying the count, a rope-framed duck medallion. The waves roll on the bar line; the crew stomps are the kick. Composition: engine : audio-first x shot-parallel content: audio-groove (stomp shanty, concertina, fiddle) x tts-voices (call-and-response vocoder) x dialogue-scenes (subtitles) FINAL CUT (player_computer_final): * Native 1920x1080. The scrimshaw is authored on a 1280x720 plate; S = 1.5 turns plate units into pixels at rasterisation time (ScaledDraw proxy), so every engraved line, rope curl and hatch spacing keeps its weight relative to the frame instead of thinning out. Grain is drawn on the plate grid and NEAREST-blown-up so the ivory tooth stays the same size. * No debug strip existed on this piece; the banners, the compass rose and the map legends are the engraver's own lettering and all stay. * Title flash: the TEN THOUSAND DUCKS card now carries "PLAYER COMPUTER" beneath it as a second engraved line with a hairline rule. """ import argparse, datetime, hashlib, math, os, subprocess, wave from pathlib import Path import numpy as np from PIL import Image, ImageDraw, ImageFont, ImageFilter NAME = "ten_thousand_ducks" TITLE = "TEN THOUSAND DUCKS" SETDIR = "player_computer_final" # ── delivery scale ─────────────────────────────────────────────────────────── # W/H stay the *plate*: the 1280x720 surface the scrimshaw is engraved on, and # every coordinate below is in plate units. OW/OH are real delivery pixels and # S = OH/H is the one number the look scales by. The ScaledDraw proxy multiplies # geometry and stroke widths at rasterisation time; font() scales sizes once. W, H, FPS = 1280, 720, 30 OW, OH = 1920, 1080 S = OH / H def P(v): return int(round(v*S)) def B(r): return r*S def _sxy(v, s): if isinstance(v, (list, tuple)): return [_sxy(u, s) for u in v] return v*s class ScaledDraw: """ImageDraw proxy: plate units in, pixels out. Only the first positional arg (xy) and `width` are touched — arc/chord/pieslice take angles positionally and those must pass through untouched.""" __slots__ = ("_d", "_s") _GEOM = frozenset(("line", "rectangle", "rounded_rectangle", "ellipse", "polygon", "arc", "chord", "pieslice", "point", "text")) def __init__(self, d, s): self._d, self._s = d, s def __getattr__(self, name): f = getattr(self._d, name) if name not in self._GEOM: return f s = self._s def wrapped(xy, *a, **kw): w = kw.get("width") if w is not None: kw["width"] = max(2, int(round(w*s))) return f(_sxy(xy, s), *a, **kw) return wrapped def mkdraw(im): d = ImageDraw.Draw(im) return d if S == 1.0 else ScaledDraw(d, S) def TL(d, txt, f): """textlength, converted back to plate units (fonts are pre-scaled).""" return d.textlength(txt, font=f)/S BPM = 96.0 BEAT = 60.0 / BPM BAR = 4 * BEAT SR = 44100 OUT = Path(__file__).parent FRAMES = OUT / "frames"; FRAMES.mkdir(exist_ok=True) AUD = OUT / "audio"; AUD.mkdir(exist_ok=True) ROOT = Path(__file__).resolve().parent # standalone: was repo root (used for git provenance) FONTS = ROOT / "fonts" SECTIONS = [ ("intro", 0, 1), ("verse1", 1, 5), ("chorus1", 5, 11), ("verse2", 11, 15), ("verse3", 15, 19), ("chorus2",19, 25), ("out", 25, 26), ] N_BARS = SECTIONS[-1][2] DUR = N_BARS * BAR + 2.5 N_FRAMES = int(DUR * FPS) MUSIC_DESC = f"sea shanty, {BPM:.0f}bpm, A minor, {N_BARS} bars, call-and-response" ENGINE_DESC = "storm / spill / map / floes / doorstep / medallion (scrimshaw)" def mtof(m): return 440.0 * 2.0 ** ((m - 69) / 12.0) _PC = {"C":0,"C#":1,"Db":1,"D":2,"D#":3,"Eb":3,"E":4,"F":5,"F#":6,"Gb":6, "G":7,"G#":8,"Ab":8,"A":9,"A#":10,"Bb":10,"B":11} def nf(name): i = 2 if (len(name) > 2 and name[1] in "#b") else 1 return mtof(12 * (int(name[i:]) + 1) + _PC[name[:i]]) def adsr(n, a, d, s, r): e = np.zeros(n) ai, di, ri = max(1, int(a*SR)), max(1, int(d*SR)), max(1, int(r*SR)) ai = min(ai, n); e[:ai] = np.linspace(0, 1, ai) if ai < n: dd = min(di, n - ai) e[ai:ai+dd] = np.linspace(1, s, dd); e[ai+dd:] = s if ri < n: e[-ri:] *= np.linspace(1, 0, ri) return e def voice(freq, dur, kind="saw", nh=26, c0=5200, c1=700, ck=8.0, res=0.0, detune=(0.0,), a=.005, d=.09, s=.7, r=.10, vib=(0.0, 0.0), seed=0): """Additive voice through a *moving* emulated filter (cutoff array). The cutoff glides c0->c1 at rate ck; `res` bumps harmonics near the cutoff. This is what gives plucks, reeses and stabs their motion. """ n = int(dur * SR) if n <= 0: return np.zeros(0) t = np.arange(n) / SR co = c1 + (c0 - c1) * np.exp(-t * ck) rng = np.random.RandomState(seed) out = np.zeros(n) vd, vr = vib for det in detune: f0 = freq * (1 + det * 0.006) for k in range(1, nh + 1): if kind == "saw": base = 1.0 / k elif kind == "square": base = (1.0 / k) if k % 2 else 0.0 elif kind == "tri": base = (1.0 / (k*k)) if k % 2 else 0.0 elif kind == "sine": base = 1.0 if k == 1 else 0.0 else: base = 1.0 / k if base == 0.0: continue fk = f0 * k if fk > SR * 0.45: break g = base / np.sqrt(1.0 + (fk / co) ** 4) if res: g = g + res * base * np.exp(-((fk - co) / (0.3 * co + 1)) ** 2) ph = rng.uniform(0, 2*np.pi) phase = 2*np.pi*fk*t + ph if vd: phase = phase + vd * np.sin(2*np.pi*vr*t) out += g * np.sin(phase) out /= len(detune) return out * adsr(n, a, d, s, r) def fm(freq, dur, ratio=2.0, index=4.0, idec=6.0, a=.002, d=.4, s=.0, r=.2, seed=0): """2-op FM — rhodes / bells / glassy leads.""" n = int(dur*SR); t = np.arange(n)/SR mod = np.sin(2*np.pi*freq*ratio*t) * index * np.exp(-t*idec) return np.sin(2*np.pi*freq*t + mod) * adsr(n, a, d, s, r) def ks(freq, dur, damp=0.996, seed=0): """Karplus-Strong pluck — guitar / harp.""" n = int(dur*SR); L = max(2, int(SR/freq)) rng = np.random.RandomState(seed) buf = rng.uniform(-1, 1, L) out = np.zeros(n); j = 0 for i in range(n): out[i] = buf[j] buf[j] = damp * 0.5 * (buf[j] + buf[(j+1) % L]) j = (j+1) % L return out * adsr(n, .001, .05, .85, .25) def bandshape(x, lo=0.0, hi=0.0, order=4): """Exact FFT band shaping. Noise sources go through this so nothing in the kit is a raw full-band blast (AESTHETIC 13a).""" n = len(x) if n < 8: return x X = np.fft.rfft(x); fq = np.maximum(np.fft.rfftfreq(n, 1/SR), 1e-6) g = np.ones_like(fq) if lo: g *= 1.0/np.sqrt(1.0 + (lo/fq)**order) if hi: g *= 1.0/np.sqrt(1.0 + (fq/hi)**order) return np.fft.irfft(X*g, n) def kick(dur=.30, f0=155, f1=48, punch=30, click=.5, seed=1): n = int(dur*SR); t = np.arange(n)/SR f = f1 + (f0-f1)*np.exp(-t*punch) body = np.sin(2*np.pi*np.cumsum(f)/SR) * np.exp(-t*10.5) ck = np.random.RandomState(seed).randn(n) * np.exp(-t*300) * click return np.tanh((body + ck) * 1.7) * .95 def snare(dur=.22, tone=196, bright=1.0, seed=2): n = int(dur*SR); t = np.arange(n)/SR rng = np.random.RandomState(seed) nz = bandshape(rng.randn(n), lo=280, hi=6200) body = np.sin(2*np.pi*tone*t) + .6*np.sin(2*np.pi*tone*1.58*t) return nz*np.exp(-t*19)*.85*bright + body*np.exp(-t*26)*.50 def hat(dur=.055, openh=False, seed=7): n = int(dur*SR); t = np.arange(n)/SR nz = bandshape(np.random.RandomState(seed).randn(n), lo=5200, hi=9800) return nz * np.exp(-t*(14 if openh else 85)) * .40 def ride(dur=.6, seed=9): n = int(dur*SR); t = np.arange(n)/SR bell = sum(np.sin(2*np.pi*f*t) for f in (522, 831, 1180, 1567, 2103)) nz = bandshape(np.random.RandomState(seed).randn(n), lo=3800, hi=9000) return bell*np.exp(-t*10)*.10 + nz*np.exp(-t*5)*.16 def rim(dur=.09, seed=3): n = int(dur*SR); t = np.arange(n)/SR return (np.sin(2*np.pi*1750*t) + .5*np.sin(2*np.pi*2600*t)) * np.exp(-t*90) * .5 def shaker(dur=.09, seed=5): n = int(dur*SR); t = np.arange(n)/SR nz = bandshape(np.random.RandomState(seed).randn(n), lo=3600, hi=8600) return nz * (np.exp(-t*40) * np.clip(t*260, 0, 1)) * .40 def crash(dur=1.6, seed=13): n = int(dur*SR); t = np.arange(n)/SR nz = bandshape(np.random.RandomState(seed).randn(n), lo=1400, hi=8200) return nz * (np.exp(-t*2.6) + .3*np.exp(-t*.6)) * .55 def riser(dur=2.0, seed=17): n = int(dur*SR); t = np.arange(n)/SR env = (t/dur) ** 1.7 sweep = np.sin(2*np.pi*np.cumsum(140 + 3000*(t/dur)**2)/SR) # noise through a band that *rises with the sweep* — pitched motion, # not a static full-band hiss (AESTHETIC 13a) rng = np.random.RandomState(seed) nz = np.zeros(n); blk = 2048 for i in range(0, n, blk): u = (i/max(1, n)) ** 1.4 fc = 300 + 5200*u seg = rng.randn(min(blk, n-i) + 256) nz[i:i+min(blk, n-i)] = bandshape(seg, lo=fc*.72, hi=fc*1.5)[:min(blk, n-i)] return (nz*env*.55 + sweep*env*.22) * .8 def vinyl(n, seed=23): """Surface noise: filtered hiss + sparse crackle.""" rng = np.random.RandomState(seed) hiss = bandshape(rng.randn(n), lo=140, hi=5200) * .022 cr = np.zeros(n) idx = rng.choice(n, size=max(1, n//2400), replace=False) cr[idx] = rng.uniform(-1, 1, len(idx)) * .10 cr = np.convolve(cr, np.exp(-np.arange(60)/9), "same") return hiss + cr def reverb(x, rt=1.6, mix=.3, seed=29, pre=0.02): """FFT convolution with a synthetic exponentially-decaying noise IR.""" n = int(rt*SR); t = np.arange(n)/SR rng = np.random.RandomState(seed) ir = rng.randn(n) * np.exp(-t*(5.0/rt)) ir[:int(pre*SR)] = 0 ir /= np.abs(ir).sum() / 40.0 + 1e-9 from numpy.fft import rfft, irfft L = 1 << int(np.ceil(np.log2(len(x) + n))) wet = irfft(rfft(x, L) * rfft(ir, L))[:len(x)] wet /= np.max(np.abs(wet)) + 1e-9 return x * (1-mix) + wet * mix * (np.max(np.abs(x)) + 1e-9) def delay(x, time=.25, fb=.38, mix=.25, taps=7): d = int(time*SR); out = x.copy() for i in range(1, taps+1): g = mix * (fb ** i); s = d*i if s >= len(x): break out[s:] += x[:len(x)-s] * g return out def lowpass(x, fc): a = np.exp(-2*np.pi*fc/SR); z = 0.0; y = np.empty_like(x) for i in range(len(x)): z = (1-a)*x[i] + a*z; y[i] = z return y class Song: """A multitrack canvas placed on an absolute bar/beat grid.""" def __init__(self, dur): self.n = int(dur*SR) self.tr = {} self.kick_t = [] def t(self, bar, step=0, swing=0.0): """absolute seconds of 16th-step `step` inside `bar`.""" sw = swing * (BEAT/4) if (step % 2) else 0.0 return bar*BAR + step*(BEAT/4) + sw def put(self, track, sig, at, g=1.0, pan=0.0): b = self.tr.setdefault(track, np.zeros((self.n, 2))) i = int(at*SR); j = min(self.n, i+len(sig)) if i >= self.n or j <= i: return th = (pan*.5+.5) * (np.pi/2) st = np.stack([sig[:j-i]*np.cos(th), sig[:j-i]*np.sin(th)], 1) * g b[i:j] += st def bus(self, track, fn): if track in self.tr: b = self.tr[track] self.tr[track] = np.stack([fn(b[:,0]), fn(b[:,1])], 1) def sec_env(self, levels, glide=0.35): """Section dynamics: a smooth per-sample gain built from {section_name: level}. Arrangement alone tends to come out flat — this is the macro arc the ear actually follows.""" env = np.ones(self.n) for nm, b0, b1 in SECTIONS: i0, i1 = int(b0*BAR*SR), min(self.n, int(b1*BAR*SR)) if i1 > i0: env[i0:i1] = levels.get(nm, 1.0) env[int(SECTIONS[-1][2]*BAR*SR):] = levels.get(SECTIONS[-1][0], 1.0) k = max(1, int(glide*SR)) return np.convolve(env, np.ones(k)/k, "same") def mixdown(self, gains, pump_depth=.30, pump_rel=.16, levels=None): mix = np.zeros((self.n, 2)) for k, b in self.tr.items(): mix += b * gains.get(k, 1.0) if levels: mix *= self.sec_env(levels)[:, None] if self.kick_t: env = np.ones(self.n); rl = int(pump_rel*SR) shape = 1 - pump_depth*np.exp(-np.arange(rl)/(pump_rel*SR/4)) for at in self.kick_t: i = int(at*SR); j = min(self.n, i+rl) if i < self.n: env[i:j] = np.minimum(env[i:j], shape[:j-i]) env = np.convolve(env, np.ones(320)/320, "same") mix *= env[:, None] # DC / sub-30Hz rumble trim (one-pole HP per channel, vectorised # via cumulative difference of a one-pole LP) a = math.exp(-2*math.pi*30.0/SR) for c in range(2): lp = np.empty(self.n); z = 0.0 col = mix[:, c] for i in range(0, self.n, 4096): blk = col[i:i+4096] for j in range(len(blk)): z = (1-a)*blk[j] + a*z; lp[i+j] = z mix[:, c] = col - lp mix = np.tanh(mix*1.25)/np.tanh(1.25) return mix / (np.max(np.abs(mix))+1e-9) * .94 def write(self, path, mix): with wave.open(str(path), "w") as w: w.setnchannels(2); w.setsampwidth(2); w.setframerate(SR) w.writeframes((np.clip(mix, -1, 1)*32767).astype(" macOS `say` (the canonical voices) -> espeak-ng / # espeak (Linux; language mapped from the say voice name, rate is wpm in both) # -> Windows SAPI (default voice, rate mapped from wpm) -> timed silence as # the last resort (duration from a chars/wpm heuristic, loud warning, never # cached so a later run with an engine present re-voices). A re-voiced film is # a different performance of the same score; that is by design. # Force a tier with POOP_TTS=say|espeak|sapi|none. def _tts_lang(voice): v = str(voice) if "Spanish" in v: return "es-mx" if "Mexico" in v else "es" if "Portuguese" in v: return "pt-br" if "Brazil" in v else "pt" if "English (UK)" in v: return "en-gb" return "en-us" def _tts_engine(): import shutil, platform want = os.environ.get("POOP_TTS", "").strip().lower() if want: return want if shutil.which("say"): return "say" if shutil.which("espeak-ng") or shutil.which("espeak"): return "espeak" if platform.system() == "Windows": return "sapi" return "none" def _tts_render(text, voice, rate, path): """Synthesize text -> mono 44.1k wav at `path` with the best available engine. Returns False if no engine (caller falls back to timed silence).""" import shutil, sys, base64 eng = _tts_engine() tmp = path.with_suffix(".tts.wav") try: if eng == "say": aiff = path.with_suffix(".aiff") subprocess.run(["say", "-v", voice, "-r", str(rate), "-o", str(aiff), text], check=True) subprocess.run(["ffmpeg", "-y", "-i", str(aiff), "-ar", str(SR), "-ac", "1", str(path)], check=True, capture_output=True) aiff.unlink(missing_ok=True) return True if eng == "espeak": exe = shutil.which("espeak-ng") or shutil.which("espeak") or "espeak-ng" subprocess.run([exe, "-v", _tts_lang(voice), "-s", str(int(rate)), "-w", str(tmp), str(text)], check=True) elif eng == "sapi": r = max(-10, min(10, round((int(rate) - 175) / 25))) esc = str(text).replace("'", "''") ps = ("Add-Type -AssemblyName System.Speech;" "$s=New-Object System.Speech.Synthesis.SpeechSynthesizer;" f"$s.Rate={r};$s.SetOutputToWaveFile('{tmp}');" f"$s.Speak('{esc}');$s.Dispose()") enc = base64.b64encode(ps.encode("utf-16-le")).decode() subprocess.run(["powershell", "-NoProfile", "-EncodedCommand", enc], check=True) else: return False subprocess.run(["ffmpeg", "-y", "-i", str(tmp), "-ar", str(SR), "-ac", "1", str(path)], check=True, capture_output=True) return True except Exception as e: print(f"[tts] {eng} failed ({e}) — falling back to timed silence", file=sys.stderr) return False finally: tmp.unlink(missing_ok=True) def _tts_silence(text, rate): import sys dur = max(0.6, len(str(text)) / (max(60, int(rate)) * 5.0 / 60.0)) print(f'[tts] no speech engine — timed silence ({dur:.2f}s): ' f'"{str(text)[:48]}"', file=sys.stderr) return np.zeros(int(dur * SR)) def say_wav(text, voice, rate, path): """text -> mono 44.1k voice wav (cached on disk; deterministic per engine).""" path = Path(path) if not path.exists(): if not _tts_render(text, voice, rate, path): return _tts_silence(text, rate) return read_wav(path) def fit(x, n): """Resample to exactly n samples. Shifts formants a little; pitch is the carrier's job, so this is free.""" if len(x) < 2: return np.zeros(n) return np.interp(np.linspace(0, len(x)-1, n), np.arange(len(x)), x) def carrier(f_per_sample, nh=30, detune=(0.0, -0.55, 0.62), vib=(0.0, 0.0)): """Band-limited additive carrier with continuous phase across note changes.""" n = len(f_per_sample) t = np.arange(n)/SR out = np.zeros(n) for d in detune: f = f_per_sample*(1 + d*0.005) if vib[0]: f = f*(1 + vib[0]*np.sin(2*np.pi*vib[1]*t)) ph = 2*np.pi*np.cumsum(f)/SR for k in range(1, nh+1): live = (f*k) < SR*0.45 if not live.any(): break out += np.sin(ph*k)/k * live return out/len(detune) def vocode(mod, car, nfft=1024, hop=256, bands=26, lo=110, hi=6500, gmax=12.0, rel=0.55, sib=0.06, tilt=4200.0): """Transfer mod's band envelope onto car. Gains are clamped and the band set is bounded — an unclamped vocoder turns carrier aliasing into hiss.""" n = max(len(mod), len(car)) mod = np.pad(mod, (0, n-len(mod))); car = np.pad(car, (0, n-len(car))) win = np.hanning(nfft); nfr = 1 + max(0, (n-nfft))//hop fr = np.fft.rfftfreq(nfft, 1/SR) edges = np.geomspace(lo, hi, bands+1) idx = [np.where((fr >= edges[b]) & (fr < edges[b+1]))[0] for b in range(bands)] keep = np.zeros(len(fr), bool) for ii in idx: keep[ii] = True out = np.zeros(n); wsum = np.zeros(n)+1e-9 prev = np.zeros(bands) for f in range(nfr): s = f*hop M = np.fft.rfft(mod[s:s+nfft]*win); C = np.fft.rfft(car[s:s+nfft]*win) am = np.abs(M); ac = np.abs(C) g = np.zeros(len(fr)) for b, ii in enumerate(idx): if not len(ii): continue em = np.sqrt((am[ii]**2).mean()); ec = np.sqrt((ac[ii]**2).mean()) gb = np.clip(em/(ec+1e-4), 0, gmax) gb = prev[b]*rel + gb*(1-rel) prev[b] = gb; g[ii] = gb out[s:s+nfft] += np.fft.irfft(C*g*keep)*win wsum[s:s+nfft] += win**2 # floor the window sum: at the ramp-in/out edges it -> 0 and the divide # detonates into a single enormous spike ws = np.maximum(wsum, 0.35*np.median(wsum[nfft:max(nfft+1, n-nfft)])) y = out/ws y[:hop] = 0.0; y[-hop:] = 0.0 y /= (np.max(np.abs(y))+1e-9) hp = np.zeros_like(mod); hp[1:] = mod[1:]-mod[:-1] for _ in range(2): hp = np.convolve(hp, [1, -0.93], "same") hp = np.clip(hp/(np.percentile(np.abs(hp), 99.5)+1e-9), -1, 1) a = math.exp(-2*math.pi*tilt/SR); z = 0.0; lp = np.empty_like(y) for i in range(len(y)): z = (1-a)*y[i] + a*z; lp[i] = z y = 0.55*y + 0.85*lp + sib*hp return y/(np.max(np.abs(y))+1e-9) def sing(text, notes, dur, voice="Moira", rate=170, cache=None, nh=30, detune=(0.0, -0.55, 0.62), vib=(0.012, 5.2), gliss=0.012, **vk): """A sung line. `notes` = [(freq, weight), …] carved across `dur` seconds.""" n = int(dur*SR) key = cache/("say_"+_h(text, voice, rate)+".wav") mod = fit(say_wav(text, voice, rate, key), n) tot = sum(w for _, w in notes) or 1.0 f = np.zeros(n); at = 0 for i, (fq, w) in enumerate(notes): ln = int(n*w/tot) if i < len(notes)-1 else n-at f[at:at+ln] = fq; at += ln if gliss: # portamento: smooth the note edges k = max(3, int(gliss*SR)); f = np.convolve(f, np.ones(k)/k, "same") f[:k] = f[k]; f[-k:] = f[-k-1] car = carrier(f, nh=nh, detune=detune, vib=vib) return vocode(mod, car, **vk) def speak(text, dur=None, voice="Alex", rate=170, cache=None, pitch=1.0): """Plain spoken line (no vocoder) — for verses that shouldn't sing.""" key = cache/("say_"+_h(text, voice, rate)+".wav") x = say_wav(text, voice, rate, key) if pitch != 1.0: x = fit(x, int(len(x)/pitch)) if dur: x = fit(x, int(dur*SR)) if len(x) > int(dur*SR) else \ np.pad(x, (0, int(dur*SR)-len(x))) return x/(np.max(np.abs(x))+1e-9) # ════════════════════════════════════════════════════════════════════════════ # THE SONG # ════════════════════════════════════════════════════════════════════════════ SW = 0.16 # Recut: each verse keeps its strongest couplet (original melodies kept with # their lines). Cut: the container/lashings line, the captain write-off line, # the beachcomber/schoolboy couplet, the retired-sailor and seagull lines. VERSES = { 1: [("In the winter storms of ninety-two a wave came for the deck,", [("E3", 1), ("A3", 1), ("B3", 1), ("C4", 1.4), ("B3", 1), ("A3", 1.6)]), ("Ten thousand yellow duckies spilled like sunrise on the foam,", [("E3", 1), ("G3", 1), ("A3", 1), ("B3", 1.4), ("C4", 1), ("B3", 1.4)])], 11: [("They rode the North Pacific gyre, they froze in Arctic floes,", [("E3", 1), ("A3", 1), ("B3", 1), ("C4", 1.4), ("B3", 1), ("A3", 1.6)]), ("They crossed the seas where nothing but the duck and current goes,", [("C4", 1), ("B3", 1), ("A3", 1), ("G3", 1.4), ("E3", 1.8)])], 15: [("One morning on his doorstep, salt-faded, chipped and small,", [("E3", 1), ("G3", 1), ("A3", 1), ("B3", 1.4), ("C4", 1), ("B3", 1.4)]), ("A duck sat looking up at him. The sea returns it all.", [("C4", 1), ("B3", 1), ("A3", 1.2), ("G3", 1), ("A3", 2.4)])], } CHORUS = [ ("Roll, little ducks, roll,", [("A3", 1.2), ("C4", 1), ("B3", 1), ("A3", 1.6)]), ("The current knows the way,", [("G3", 1), ("A3", 1), ("B3", 1.2), ("E3", 1.6)]), ("Ten thousand yellow sailors", [("A3", 1), ("B3", 1), ("C4", 1.2), ("D4", 1), ("C4", 1.2)]), ("Bound for home someday.", [("B3", 1), ("A3", 1), ("G3", 1.2), ("A3", 2.2)]), ] def stompf(seed=341): n = int(.25*SR); t = np.arange(n)/SR f = 45 + 90*np.exp(-t*44) body = np.sin(2*np.pi*np.cumsum(f)/SR)*np.exp(-t*14) wood = bandshape(np.random.RandomState(seed).randn(n), lo=120, hi=900)*np.exp(-t*36) return (body + wood*.5)*.9 def concertina(rootf, ivs, dur, seed=0): out = np.zeros(int(dur*SR)) for k2, iv in enumerate(ivs): sig = voice(rootf*2*2**(iv/12.0), dur*0.95, kind="saw", nh=14, c0=1700, c1=800, ck=1.0, detune=(-1.0, 1.1), vib=(.012, 5.0), a=.06, d=.3, s=.8, r=.3, seed=seed+k2) out[:len(sig)] += sig*.5 return out def gullcry(seed=347): n = int(.45*SR); t = np.arange(n)/SR f = 1200*(1 + .2*np.exp(-t*20)) - 380*(t/.45) x = np.sin(2*np.pi*np.cumsum(f)/SR) + .5*np.sin(2*2.01*np.pi*np.cumsum(f)/SR) return np.tanh(x*1.3)*np.sin(np.pi*np.clip(t/.45, 0, 1))**.6*.4 def squeak_duck(seed=349): n = int(.3*SR); t = np.arange(n)/SR f = 900 + 500*np.sin(np.pi*np.clip(t/.3, 0, 1)) x = np.sin(2*np.pi*np.cumsum(f)/SR)*(1+.6*np.sin(2*np.pi*30*t)) return bandshape(x, lo=500, hi=3200)*np.sin(np.pi*np.clip(t/.3, 0, 1))*.4 def build_song(): s = Song(DUR) R = np.random.RandomState(9600) def sec_of(bar): for nm, a, b in SECTIONS: if a <= bar < b: return nm return "out" # verses Am-G-Am-Em, choruses Am-F-G-Am — all in A natural minor PROGV = [("A1", (0, 3, 7)), ("G1", (0, 4, 7)), ("A1", (0, 3, 7)), ("E2", (0, 3, 7))] PROGC = [("A1", (0, 3, 7)), ("F1", (0, 4, 7)), ("G1", (0, 4, 7)), ("A1", (0, 3, 7))] for bar in range(N_BARS): sec = sec_of(bar) cho = sec.startswith("chorus") quiet = sec in ("intro", "out") prog = PROGC if cho else PROGV rootn, ivs = prog[(bar//2) % 4] rootf = nf(rootn) # ---- stomps + claps ------------------------------------------------ for st in (0, 8): at = s.t(bar, st, SW) s.put("drums", stompf(seed=bar*3+st), at, g=.85 if not quiet else .5) if st == 0: s.kick_t.append(at) for st in (4, 12): s.put("drums", snare(dur=.12, tone=280, bright=.65), s.t(bar, st, SW), g=.42 if cho else .3, pan=-.06) if cho: for st in (2, 10): s.put("drums", stompf(seed=bar*7+st), s.t(bar, st, SW), g=.4) s.put("drums", shaker(), s.t(bar, 6, SW), g=.1, pan=.3) # ---- concertina ---------------------------------------------------- s.put("conc", concertina(rootf, ivs, BAR*1.05, seed=bar*11), s.t(bar, 0), g=.22 if not quiet else .15) # ---- fiddle counterline ------------------------------------------- if cho or sec == "verse3": FID = [12, 10, 8, 7, 8, 10] for j in (0, 5, 8, 13): iv = FID[(bar+j) % 6] s.put("fid", voice(nf("A3")*2**(iv/12.0), BEAT*.7, kind="saw", nh=16, c0=2400, c1=1100, ck=2.6, vib=(.022, 6.0), a=.04, d=.2, s=.65, r=.15, seed=bar*17+j), s.t(bar, j, SW), g=.11, pan=.25) # ---- bass ---------------------------------------------------------- for st, iv in ((0, 0), (8, 7)): s.put("bass", voice(rootf, BEAT*1.5, kind="sine", nh=3, c0=190, c1=85, ck=2, a=.01, d=.3, s=.8, r=.3, seed=bar+st), s.t(bar, st, SW), g=.36) # ---- vocals ------------------------------------------------------------- SUBS = [] def crew_line(text, mel, at, dur, lead_only=False): melf = [(nf(nm), wg) for nm, wg in mel] lead = sing(text, melf, dur, voice="Alex", rate=140, cache=AUD, detune=(0.0, -0.5, 0.6), vib=(.012, 4.6)) s.put("vox", lead, at, g=.52, pan=0.0) if not lead_only: for vc, dt2, pn in (("Fred", (0.0, -1.1), -.2), ("Moira", (0.0, 0.9), .2)): hh = sing(text, melf, dur, voice=vc, rate=140, cache=AUD, detune=dt2, vib=(.010, 4.2)) s.put("vox", hh, at, g=.22, pan=pn) low = sing(text, [(f/2, wg) for f, wg in melf], dur, voice="Fred", rate=140, cache=AUD, detune=(0.0, -1.2), vib=(.006, 4.0)) s.put("vox", low, at, g=.16, pan=.05) SUBS.append((at, at + dur + 0.2, text if lead_only else text.upper())) for b0, lines in VERSES.items(): for i, (text, mel) in enumerate(lines): crew_line(text, mel, (b0 + i*2)*BAR + BEAT*0.3, BAR*2*0.86, lead_only=True) for b0 in (5, 19): for i, (text, mel) in enumerate(CHORUS): crew_line(text, mel, (b0 + i*1.5)*BAR + BEAT*0.25, BAR*1.5*0.9) np.savez(AUD/"subs.npz", t0=np.array([a for a, b, c in SUBS]), t1=np.array([b for a, b, c in SUBS]), tx=np.array([c for a, b, c in SUBS], dtype=object)) # ---- world -------------------------------------------------------------- n2 = int(DUR*SR); t2 = np.arange(n2)/SR waves = bandshape(np.random.RandomState(351).randn(n2), lo=150, hi=1600) am = .3 + .25*np.sin(2*np.pi*0.08*t2) + .2*np.sin(2*np.pi*0.035*t2+1) s.put("fx", waves*np.clip(am, 0, 1)*.2, 0.0, g=.8) thun = bandshape(np.random.RandomState(353).randn(int(2.5*SR)), lo=40, hi=300) tt3 = np.arange(len(thun))/SR s.put("fx", thun*np.exp(-tt3*1.6)*np.clip(tt3*8, 0, 1)*.7, 1*BAR + BEAT, g=.6) s.put("fx", gullcry(), 15*BAR + BEAT, g=.4, pan=.25) s.put("fx", gullcry(seed=348), 16*BAR + BEAT*2, g=.35, pan=-.2) s.put("fx", squeak_duck(), 17*BAR + BEAT*2, g=.5, pan=.1) s.put("fx", squeak_duck(seed=350), 25*BAR + BEAT*2, g=.55, pan=.05) s.bus("conc", lambda x: reverb(x, rt=2.0, mix=.28, seed=2001)) s.bus("fid", lambda x: reverb(x, rt=2.2, mix=.32, seed=2003)) s.bus("vox", lambda x: reverb(x, rt=2.2, mix=.3, seed=2007)) s.bus("fx", lambda x: reverb(x, rt=2.4, mix=.24, seed=2009)) mix = s.mixdown(dict(drums=1.0, conc=1.0, fid=1.0, bass=1.0, vox=1.0, fx=1.0), pump_depth=.10, pump_rel=.16, levels=dict(intro=.6, verse1=.82, chorus1=1.0, verse2=.85, verse3=.85, chorus2=1.0, out=.7)) wav = AUD / "final.wav" s.write(wav, mix) return wav, mix def analyze(mix): x = mix.mean(1) hop = SR / FPS; win = int(hop*1.7) E = {k: np.zeros(N_FRAMES) for k in ("rms", "low", "mid", "high")} for f in range(N_FRAMES): i = int(f*hop); seg = x[i:i+win] if len(seg) < 16: continue E["rms"][f] = np.sqrt((seg**2).mean()) sp = np.abs(np.fft.rfft(seg*np.hanning(len(seg)))) fr = np.fft.rfftfreq(len(seg), 1/SR) E["low"][f] = sp[fr < 170].sum() E["mid"][f] = sp[(fr >= 170) & (fr < 2400)].sum() E["high"][f] = sp[fr >= 2400].sum() for k in E: p = np.percentile(E[k], 96) + 1e-9 E[k] = np.clip(E[k]/p, 0, 1.25) lo = E["low"] flux = np.maximum(0, lo - np.concatenate([[0], lo[:-1]])) E["kick"] = np.clip(np.convolve(flux, [.25,.5,.25], "same") / (np.percentile(flux, 97)+1e-9), 0, 1) np.savez(AUD/"env.npz", **E) return E _ENV = {} def env(): if not _ENV: z = np.load(AUD/"env.npz") for k in z.files: _ENV[k] = z[k] return _ENV # ════════════════════════════════════════════════════════════════════════════ # ANIMATION HELPERS — motion is the point, so it gets its own primitives # ════════════════════════════════════════════════════════════════════════════ def ease_io(u): return u*u*(3-2*u) def ease_out(u): return 1-(1-u)**3 def ease_in(u): return u**3 def bounce(u, n=3): return abs(math.sin(u*math.pi*n))*(1-u) def lerp(a, b, u): return a + (b-a)*u def walk(t, speed=1.0): """Returns (leg_swing, body_bob, arm_swing) for a walk cycle at time t.""" p = t*speed*math.tau return math.sin(p), abs(math.sin(p))*-1.0, math.sin(p+math.pi) def spring(u, freq=3.0, damp=5.0): """Overshoot-and-settle, for things that arrive.""" if u <= 0: return 0.0 return 1 - math.exp(-damp*u)*math.cos(freq*math.tau*u) def arc(p0, p1, u, h=0.3): """Ballistic arc between two points.""" x = lerp(p0[0], p1[0], u) y = lerp(p0[1], p1[1], u) - math.sin(u*math.pi)*h*abs(p1[0]-p0[0]) return x, y def ramp(stops, n=256): stops = np.array(stops, np.float32) xs = np.linspace(0, 1, len(stops)); g = np.linspace(0, 1, n) return np.stack([np.interp(g, xs, stops[:, c]) for c in range(3)], 1) def apply_ramp(v01, lut): i = np.clip(v01*(len(lut)-1), 0, len(lut)-1).astype(np.int32) return lut[i] def value_noise(h, w, scale, seed): rng = np.random.RandomState(seed) gh, gw = int(h/scale)+2, int(w/scale)+2 g = rng.rand(gh, gw) ys = np.linspace(0, gh-1-1e-3, h); xs = np.linspace(0, gw-1-1e-3, w) y0 = ys.astype(int); x0 = xs.astype(int) fy = (ys-y0)[:, None]; fx = (fx0 := (xs-x0))[None, :] sy = fy*fy*(3-2*fy); sx = fx*fx*(3-2*fx) g00 = g[np.ix_(y0, x0)]; g01 = g[np.ix_(y0, x0+1)] g10 = g[np.ix_(y0+1, x0)]; g11 = g[np.ix_(y0+1, x0+1)] return (g00*(1-sx)+g01*sx)*(1-sy) + (g10*(1-sx)+g11*sx)*sy def fbm(h, w, scale, seed, oct=4): out = np.zeros((h, w)); amp = 1.0; nrm = 0.0 for o in range(oct): out += amp*value_noise(h, w, max(2, scale/(2**o)), seed+o) nrm += amp; amp *= .5 return out/nrm # ════════════════════════════════════════════════════════════════════════════ # SCRIMSHAW # ════════════════════════════════════════════════════════════════════════════ NEUTRAL = {"rms": .5, "low": .4, "mid": .4, "high": .3, "kick": .2} BONE = (236, 226, 204) BONE2 = (222, 210, 184) SEPIA = (66, 50, 40) SEP2 = (110, 88, 66) _BN = {} def bone(): if "b" not in _BN: im = Image.new("RGB", (OW, OH), BONE) d = mkdraw(im) # growth arcs for q in range(10): r = 300 + q*140 d.arc([W*0.5-r, H*1.1-r, W*0.5+r, H*1.1+r], 200, 340, fill=BONE2, width=2) R = np.random.RandomState(23) for _ in range(80): x, y = R.uniform(0, W), R.uniform(0, H) d.ellipse([x-1, y-1, x+1, y+1], fill=(214, 202, 176)) _BN["b"] = im return _BN["b"].copy() def eng_hatch(im, pts, ang=0.6, spacing=7, col=SEP2, w=2): """Hatch-fill a polygon by masking a line layer.""" mask = Image.new("L", (OW, OH), 0) mkdraw(mask).polygon(pts, fill=255) lay = Image.new("RGB", (OW, OH), BONE) dl = mkdraw(lay) diag = int(math.hypot(W, H)) ca, sa = math.cos(ang), math.sin(ang) for q in range(-diag//spacing, diag//spacing): off = q*spacing dl.line([(-sa*1000 + ca*off + W/2, ca*1000 + sa*off + H/2), (sa*1000 + ca*off + W/2, -ca*1000 + sa*off + H/2)], fill=col, width=w) im.paste(lay, (0, 0), mask) def rope_border(d, t): for q in range(0, W, 26): d.arc([q, 8, q+30, 34], 180, 360, fill=SEPIA, width=3) d.arc([q, H-36, q+30, H-10], 0, 180, fill=SEPIA, width=3) for q in range(0, H, 26): d.arc([8, q, 34, q+30], 90, 270, fill=SEPIA, width=3) d.arc([W-36, q, W-10, q+30], 270, 90, fill=SEPIA, width=3) def banner(d, cx, y, text, sc=1.0): f = font(int(26*sc), "Georgia Bold.ttf") tw = TL(d, text, f) x0, x1 = cx-tw/2-30*sc, cx+tw/2+30*sc d.polygon([(x0, y), (x1, y), (x1-14*sc, y+18*sc), (x1, y+36*sc), (x0, y+36*sc), (x0+14*sc, y+18*sc)], outline=SEPIA, width=3) d.line([x0+10*sc, y+8*sc, x1-10*sc, y+8*sc], fill=SEP2, width=1) d.line([x0+10*sc, y+28*sc, x1-10*sc, y+28*sc], fill=SEP2, width=1) d.text((cx-tw/2, y+5*sc), text, font=f, fill=SEPIA) def eng_waves(d, im, y0, t, rows=4, amp=1.0): """Rolling engraved wave curls; phase locked to the bar.""" ph = (t/BAR) % 1.0 for r in range(rows): y = y0 + r*44 for q in range(-1, W//72 + 2): x = q*72 + (ph*72 if r % 2 == 0 else -ph*72) d.arc([x, y, x+64, y+40], 180, 20, fill=SEPIA, width=3) d.arc([x+34, y+12, x+74, y+38], 200, 340, fill=SEP2, width=2) def duck(d, x, y, sc, t, ph=0.0, floating=True): bob = math.sin(t/BEAT*math.pi + ph)*3*sc if floating else 0 y = y + bob d.ellipse([x-13*sc, y-9*sc, x+13*sc, y+5*sc], outline=SEPIA, width=max(2, int(2*sc))) d.ellipse([x+4*sc, y-18*sc, x+16*sc, y-6*sc], outline=SEPIA, width=max(2, int(2*sc))) d.polygon([(x+15*sc, y-13*sc), (x+22*sc, y-11*sc), (x+15*sc, y-9*sc)], outline=SEPIA, width=1) d.ellipse([x+9*sc-1, y-14*sc-1, x+9*sc+2, y-14*sc+2], fill=SEPIA) # hatched breast shade d.arc([x-13*sc, y-9*sc, x+13*sc, y+5*sc], 40, 140, fill=SEP2, width=2) if sc > 3: d.arc([x-8*sc, y-7*sc, x+6*sc, y+3*sc], 300, 120, fill=SEPIA, width=int(1.2*sc)) d.ellipse([x+8*sc, y-13*sc, x+11*sc, y-10*sc], fill=BONE) if floating: d.line([x-20*sc, y+5*sc, x-13*sc, y+4*sc], fill=SEP2, width=2) d.line([x+13*sc, y+4*sc, x+20*sc, y+5*sc], fill=SEP2, width=2) def ship(d, im, x, y, sc, t, e, tilt=0.0): roll = math.sin(t/BAR*math.tau)*0.06 + tilt def rp(px, py): c, s2 = math.cos(roll), math.sin(roll) return (x + (px*c - py*s2)*sc, y + (px*s2 + py*c)*sc) hull = [rp(-150, 0), rp(150, 0), rp(110, 50), rp(-120, 50)] d.polygon(hull, outline=SEPIA, width=4) eng_hatch(im, hull, ang=0.2, spacing=6) d.line([rp(-150, 0), rp(150, 0)], fill=SEPIA, width=4) for mx, mh in ((-60, 190), (50, 230)): d.line([rp(mx, 0), rp(mx, -mh)], fill=SEPIA, width=4) for sy, sw2 in ((-mh*0.85, 60), (-mh*0.55, 76)): sail = [rp(mx-sw2, sy), rp(mx+sw2, sy), rp(mx+sw2*0.8, sy+mh*0.24), rp(mx-sw2*0.8, sy+mh*0.24)] d.polygon(sail, outline=SEPIA, width=3) eng_hatch(im, sail, ang=1.35, spacing=6, col=SEP2, w=1) d.line([rp(mx, -mh), rp(mx+18, -mh+8)], fill=SEPIA, width=2) d.line([rp(-150, 0), rp(-60, -190)], fill=SEP2, width=2) d.line([rp(150, 0), rp(50, -230)], fill=SEP2, width=2) def compass(d, cx, cy, r, t): d.ellipse([cx-r, cy-r, cx+r, cy+r], outline=SEPIA, width=3) d.ellipse([cx-r*0.7, cy-r*0.7, cx+r*0.7, cy+r*0.7], outline=SEP2, width=2) a0 = math.sin(t*0.4)*0.1 for q in range(8): a = a0 + q*math.tau/8 rr = r*0.92 if q % 2 == 0 else r*0.6 d.polygon([(cx+math.cos(a)*rr, cy+math.sin(a)*rr), (cx+math.cos(a+0.18)*r*0.24, cy+math.sin(a+0.18)*r*0.24), (cx+math.cos(a-0.18)*r*0.24, cy+math.sin(a-0.18)*r*0.24)], outline=SEPIA, width=2) f = font(20, "Georgia Bold.ttf") d.text((cx-7, cy-r-26), "N", font=f, fill=SEPIA) class Storm: def __init__(self, shot, rng): self.rng = rng; self.i0 = shot.i0 def frame(self, k, u, e): t = (self.i0+k)/FPS im = bone(); d = mkdraw(im) rope_border(d, t) # rain hatch R = np.random.RandomState(int(t*8) % 97) for _ in range(120): x, y = R.uniform(40, W-40), R.uniform(60, H*0.5) d.line([x, y, x-7, y+16], fill=SEP2, width=1) # lightning on rare kicks if e["kick"] > 0.8 and int(t*4) % 3 == 0: zx = W*0.72 d.line([zx, 60, zx-30, H*0.24, zx+10, H*0.32, zx-24, H*0.5], fill=SEPIA, width=4) heave = math.sin(t/BAR*math.tau)*26 ship(d, im, W*0.46, H*0.52 + heave, 1.0, t, e, tilt=0.10) # the container mid-tumble off the stern cu = (t/BAR/2) % 1.0 cx2, cy2 = arc((W*0.60, H*0.50 + heave), (W*0.88, H*0.72), ease_io(min(1, cu*1.4)), h=0.3) crate = [(cx2-34, cy2-22), (cx2+34, cy2-22), (cx2+34, cy2+22), (cx2-34, cy2+22)] d.polygon(crate, outline=SEPIA, width=3) d.line([cx2-34, cy2, cx2+34, cy2], fill=SEP2, width=2) d.line([cx2, cy2-22, cx2, cy2+22], fill=SEP2, width=2) eng_waves(d, im, H*0.68, t, rows=4) banner(d, W*0.5, H*0.08, "THE GALE OF '92") return np.asarray(im, np.float32) class Spill: def __init__(self, shot, rng): self.rng = rng; self.i0 = shot.i0 R = np.random.RandomState(int(rng.integers(1e6))) self.ducks = R.random((26, 3)) def frame(self, k, u, e): t = (self.i0+k)/FPS im = bone(); d = mkdraw(im) rope_border(d, t) # crate half-sunk, lid open, ducks pouring cx2, cy2 = W*0.26, H*0.46 crate = [(cx2-70, cy2-50), (cx2+70, cy2-50), (cx2+70, cy2+40), (cx2-70, cy2+40)] d.polygon(crate, outline=SEPIA, width=4) eng_hatch(im, crate, ang=0.3, spacing=8) d.polygon([(cx2-70, cy2-50), (cx2+10, cy2-96), (cx2+150, cy2-84), (cx2+70, cy2-50)], outline=SEPIA, width=3) for i, (dx, dy, dp) in enumerate(self.ducks): du = (u*1.4 + dp*0.5) % 1.2 px, py = arc((cx2+20, cy2-60), (W*(0.42+dx*0.5), H*(0.6+dy*0.26)), min(1, du), h=0.4) duck(d, px, py, 0.9+dp*0.5, t, ph=i, floating=du >= 1) eng_waves(d, im, H*0.64, t, rows=4) banner(d, W*0.5, H*0.08, "28,800 SOULS TO THE SEA") return np.asarray(im, np.float32) class MapEng: def __init__(self, shot, rng): self.rng = rng; self.i0 = shot.i0 def frame(self, k, u, e): t = (self.i0+k)/FPS im = bone(); d = mkdraw(im) rope_border(d, t) # crude engraved continents (Pacific-centred) asia = [(90, 150), (260, 110), (330, 180), (300, 300), (200, 360), (110, 300)] d.polygon(asia, outline=SEPIA, width=3) eng_hatch(im, asia, ang=0.5, spacing=9) amer = [(W-260, 90), (W-100, 120), (W-70, 260), (W-150, 420), (W-230, 520), (W-290, 340), (W-240, 200)] d.polygon(amer, outline=SEPIA, width=3) eng_hatch(im, amer, ang=0.5, spacing=9) arctic = [(200, 70), (W-300, 56), (W-260, 96), (240, 116)] d.polygon(arctic, outline=SEPIA, width=3) eng_hatch(im, arctic, ang=1.2, spacing=7) f = font(20, "Georgia Italic.ttf") d.text((150, 210), "ASIA", font=f, fill=SEPIA) d.text((W-220, 250), "AMERICA", font=f, fill=SEPIA) d.text((W*0.44, 70), "ARCTIC SEA", font=f, fill=SEPIA) d.text((W*0.40, H*0.52), "THE GREAT GYRE", font=f, fill=SEP2) # spill point + routes drawing on with global progress sp = (W*0.42, H*0.40) d.line([sp[0]-8, sp[1]-8, sp[0]+8, sp[1]+8], fill=SEPIA, width=3) d.line([sp[0]-8, sp[1]+8, sp[0]+8, sp[1]-8], fill=SEPIA, width=3) prog = np.clip((t/BAR - 5)/18.0, 0, 1) ROUTES = [ [sp, (W*0.55, H*0.55), (W*0.48, H*0.72), (W*0.36, H*0.62), (W*0.42, H*0.46)], # gyre loop [sp, (W*0.5, H*0.24), (W*0.55, H*0.13)], # north [sp, (W*0.62, H*0.44), (W*0.76, H*0.40), (W-150, H*0.36)], [sp, (W*0.34, H*0.50), (W*0.20, H*0.44)], ] for ri, route in enumerate(ROUTES): rl = prog*len(route)*1.1 - ri*0.15 for q in range(len(route)-1): seg = np.clip(rl - q, 0, 1) if seg <= 0: continue x0, y0 = route[q]; x1, y1 = route[q+1] n2 = 8 for j in range(int(n2*seg)): uu = (j+0.5)/n2 px, py = x0+(x1-x0)*uu, y0+(y1-y0)*uu d.ellipse([px-3, py-3, px+3, py+3], fill=SEPIA) # a duck at each route head hd = min(len(route)-1.001, max(0.0, rl-0.2)) qi = int(hd); uu = hd-qi x0, y0 = route[qi]; x1, y1 = route[qi+1] duck(d, x0+(x1-x0)*uu, y0+(y1-y0)*uu, 1.0, t, ph=ri) compass(d, W*0.13, H*0.78, 74, t) banner(d, W*0.5, H*0.885, "TWENTY YEARS ADRIFT", sc=1.0) return np.asarray(im, np.float32) class Floes: def __init__(self, shot, rng): self.rng = rng; self.i0 = shot.i0 def frame(self, k, u, e): t = (self.i0+k)/FPS im = bone(); d = mkdraw(im) rope_border(d, t) # aurora hatch curves for q in range(3): pts = [] for xx in range(60, W-40, 60): pts.append((xx, 90 + q*36 + math.sin(xx*0.01 + t*0.8 + q)*22)) d.line(pts, fill=SEP2, width=2) eng_waves(d, im, H*0.60, t, rows=3) # ice floes bobbing with ducks aboard for i, (fx, fw) in enumerate(((0.18, 180), (0.48, 240), (0.76, 150))): bob = math.sin(t/BAR*math.tau + i*1.4)*8 y = H*0.58 + bob floe = [(W*fx, y), (W*fx+fw, y), (W*fx+fw-24, y+34), (W*fx+20, y+30)] d.polygon(floe, outline=SEPIA, width=3) eng_hatch(im, floe, ang=0.1, spacing=6, col=SEP2, w=1) for q in range(2+i): duck(d, W*fx + 40 + q*46, y - 8, 1.1, t, ph=i+q, floating=False) # whale spout behind wx = W*0.62 d.arc([wx-90, H*0.47, wx+90, H*0.60], 180, 360, fill=SEPIA, width=4) sp = abs(math.sin(t*1.2)) for q in range(3): d.line([wx, H*0.47 - q*12*sp, wx - 8 + q*8, H*0.47 - q*16*sp - 10], fill=SEP2, width=2) banner(d, W*0.5, H*0.08, "THE ARCTIC DETOUR") return np.asarray(im, np.float32) class Doorstep: def __init__(self, shot, rng): self.rng = rng; self.i0 = shot.i0 def frame(self, k, u, e): t = (self.i0+k)/FPS im = bone(); d = mkdraw(im) rope_border(d, t) # sun rays hatched for q in range(9): a = math.pi + q*math.pi/8 d.line([W*0.82 + math.cos(a)*70, H*0.20 + math.sin(a)*70, W*0.82 + math.cos(a)*120, H*0.20 + math.sin(a)*120], fill=SEP2, width=2) d.ellipse([W*0.82-44, H*0.20-44, W*0.82+44, H*0.20+44], outline=SEPIA, width=3) # lighthouse lh = [(W*0.10, H*0.62), (W*0.16, H*0.62), (W*0.145, H*0.30), (W*0.115, H*0.30)] d.polygon(lh, outline=SEPIA, width=3) eng_hatch(im, lh, ang=0.0, spacing=10) d.rectangle([W*0.112, H*0.26, W*0.148, H*0.30], outline=SEPIA, width=3) # cottage door + step door = [(W*0.42, H*0.30), (W*0.60, H*0.30), (W*0.60, H*0.78), (W*0.42, H*0.78)] d.polygon(door, outline=SEPIA, width=4) eng_hatch(im, door, ang=1.57, spacing=9) d.ellipse([W*0.57, H*0.53, W*0.585, H*0.55], fill=SEPIA) d.rectangle([W*0.40, H*0.78, W*0.62, H*0.83], outline=SEPIA, width=3) # the old sailor bending down bend = 0.35 + 0.1*math.sin(t*0.8) sx, sy = W*0.70, H*0.80 d.line([sx, sy, sx-6, sy-60], fill=SEPIA, width=8) # legs d.line([sx-6, sy-60, sx-40*bend*2, sy-100+20*bend], fill=SEPIA, width=8) hx, hy = sx-46*bend*2, sy-112+26*bend d.ellipse([hx-12, hy-12, hx+12, hy+12], outline=SEPIA, width=3) d.arc([hx-16, hy-18, hx+16, hy-2], 180, 360, fill=SEPIA, width=4) d.line([sx-14, sy-84, hx+6, sy-66], fill=SEP2, width=5) # arm down # beard hatch d.arc([hx-10, hy-2, hx+10, hy+14], 0, 180, fill=SEP2, width=3) # the duck on the step, looking up duck(d, W*0.51, H*0.765, 1.5, t, ph=0.3, floating=False) banner(d, W*0.5, H*0.08, "THE SEA RETURNS IT ALL") return np.asarray(im, np.float32) class Medallion: def __init__(self, shot, rng): self.rng = rng; self.i0 = shot.i0 def frame(self, k, u, e): t = (self.i0+k)/FPS im = bone(); d = mkdraw(im) rope_border(d, t) cx, cy, r = W*0.5, H*0.48, H*0.30 # rope oval for q in range(0, 360, 14): a = math.radians(q) d.arc([cx+math.cos(a)*r*1.28-12, cy+math.sin(a)*r-12, cx+math.cos(a)*r*1.28+12, cy+math.sin(a)*r+12], q, q+180, fill=SEPIA, width=3) duck(d, cx-8, cy+30, 6.0, t, ph=0.0, floating=True) # chips + barnacle d.arc([cx+60, cy-70, cx+90, cy-40], 20, 160, fill=SEP2, width=2) d.ellipse([cx-80, cy+44, cx-64, cy+58], outline=SEP2, width=2) banner(d, W*0.5, H*0.86, "ONE OF THE TEN THOUSAND", sc=1.0) return np.asarray(im, np.float32) ENGINES = {"storm": Storm, "spill": Spill, "map": MapEng, "floes": Floes, "doorstep": Doorstep, "medallion": Medallion} PLAN = { "intro": (["medallion"], [16]), "verse1": (["storm", "spill"], [8, 8]), "chorus1": (["map", "spill"], [12, 12]), "verse2": (["map", "floes"], [8, 8]), "verse3": (["doorstep"], [16]), "chorus2": (["map", "doorstep"], [12, 12]), "out": (["medallion"], [16]), } CARDS = {"intro": "TEN THOUSAND DUCKS", "verse1": None, "chorus1": None, "verse2": None, "verse3": None, "chorus2": None, "out": None} SYSTEM_NAMES = ["LAT 44N", "LON 178E", "THE GYRE", "THE FLOES", "DUNDEE", "HOME"] class Shot: __slots__ = ("idx", "i0", "i1", "n", "engine", "section", "seed", "text", "card") def __init__(self, idx, i0, i1, engine, section, text=None, card=None): self.idx, self.i0, self.i1 = idx, i0, i1 self.n = i1 - i0 self.engine, self.section = engine, section self.seed = 90210 + idx*7919 self.text, self.card = text, card def build_shots(): """Deterministic. Recut change vs the source: engines run *sequentially* through each section's pool (not a seeded random pick) — at 26 bars the narrative order is the whole point (storm before spill, the finale closing map -> doorstep). Shot lengths still come from a menu.""" R = np.random.RandomState(5150) shots = []; idx = 0 for nm, b0, b1 in SECTIONS: engs, menu = PLAN[nm] t = b0*BAR; j = 0 while t < b1*BAR - 1e-6: step = menu[R.randint(len(menu))]*BEAT t2 = min(t+step, b1*BAR) if (b1*BAR - t2) < BEAT*1.5: t2 = b1*BAR # no orphan sliver i0, i1 = int(t*FPS), int(t2*FPS) if i1 > i0: eng = engs[j % len(engs)] txt = [SYSTEM_NAMES[(idx+q) % len(SYSTEM_NAMES)] for q in range(2)] shots.append(Shot(idx, i0, i1, eng, nm, txt, CARDS[nm] if j == 0 else None)) idx += 1; j += 1 t = t2 if shots: shots[-1].i1 = N_FRAMES; shots[-1].n = N_FRAMES - shots[-1].i0 return shots # ── portable font resolution (cross-platform; replaces the repo-only lookup) ── import warnings as _warnings _FONT_ALIASES = { "Menlo.ttc": ["Menlo.ttc", "DejaVuSansMono.ttf", "consola.ttf", "LiberationMono-Regular.ttf"], "Georgia.ttf": ["Georgia.ttf", "georgia.ttf", "DejaVuSerif.ttf", "LiberationSerif-Regular.ttf"], "Georgia Bold.ttf": ["Georgia Bold.ttf", "georgiab.ttf", "DejaVuSerif-Bold.ttf", "LiberationSerif-Bold.ttf"], "Georgia Italic.ttf": ["Georgia Italic.ttf", "georgiai.ttf", "DejaVuSerif-Italic.ttf", "LiberationSerif-Italic.ttf"], "Impact.ttf": ["Impact.ttf", "impact.ttf", "Anton-Regular.ttf", "DejaVuSans-Bold.ttf"], "Helvetica.ttc": ["Helvetica.ttc", "arial.ttf", "Arial.ttf", "DejaVuSans.ttf", "LiberationSans-Regular.ttf"], } def _font_dirs(): here = Path(__file__).resolve() dirs = [here.parent / "fonts"] + [p / "fonts" for p in list(here.parents)[1:4]] try: home = Path.home() except Exception: home = None dirs += [Path("/System/Library/Fonts"), Path("/System/Library/Fonts/Supplemental"), Path("/Library/Fonts"), Path("C:/Windows/Fonts"), Path("/usr/share/fonts"), Path("/usr/local/share/fonts")] if home: dirs += [home / "Library/Fonts", home / ".fonts", home / ".local/share/fonts"] return dirs _FONT_DIRS = _font_dirs() _FF = {} def _find_font(name): """Path of a usable font file for `name`, or None. Cached per name.""" if name in _FF: return _FF[name] found = None for cand in _FONT_ALIASES.get(name, [name]): for d in _FONT_DIRS: if not d.is_dir(): continue p = d / cand if p.is_file(): found = p; break try: found = next(iter(d.rglob(cand)), None) except OSError: found = None if found: break if found: break if found is None: _warnings.warn(f"font {name} not found in fonts/ or system font dirs; " f"using Pillow default (layout will differ)") _FF[name] = found return found def _load_font(p, size): """ImageFont for path `p` (from _find_font) at `size`; Pillow default if p is None.""" if p is None: try: return ImageFont.load_default(size=int(size)) except TypeError: return ImageFont.load_default() return ImageFont.truetype(str(p), size) _FC = {} def font(size, name="Menlo.ttc"): key = (size, name) if key not in _FC: p = _find_font(name) _FC[key] = _load_font(p, max(1, P(size))) return _FC[key] _VIG = {} def vignette(): if "v" not in _VIG: yy, xx = np.mgrid[0:OH, 0:OW] nx = (xx-OW/2)/(OW/2); ny = (yy-OH/2)/(OH/2) r = np.sqrt(nx**2+ny**2)/1.42 _VIG["v"] = np.clip(1.0-0.40*r**2.2, 0, 1)[..., None] return _VIG["v"] _SUBS = {} def subs(): if not _SUBS: p = AUD/"subs.npz" if p.exists(): z = np.load(p, allow_pickle=True) _SUBS["t0"], _SUBS["t1"], _SUBS["tx"] = z["t0"], z["t1"], z["tx"] else: _SUBS["t0"] = _SUBS["t1"] = np.zeros(0) _SUBS["tx"] = np.zeros(0, dtype=object) return _SUBS def post(arr, i, e, shot): a = arr.astype(np.float32) if isinstance(arr, np.ndarray) else \ np.asarray(arr, np.float32) lum = a.mean(2, keepdims=True)/255.0 a = a + (1-lum)*np.array([6, 2, -4], np.float32) a *= vignette()*0.2 + 0.8 rng = np.random.RandomState(23800 + i) if S == 1.0: a += rng.normal(0, 1.7, a.shape) else: # the ivory's tooth is a look, not a resolution: drawn on the plate # grid and blown up nearest-neighbour so a speck keeps its size gn = rng.normal(0, 1.7, (H, W, 3))*8.0 + 128.0 gi = Image.fromarray(np.clip(gn, 0, 255).astype(np.uint8)) a += (np.asarray(gi.resize((OW, OH), Image.NEAREST), np.float32) - 128.0)/8.0 out = Image.fromarray(np.clip(a, 0, 255).astype(np.uint8)) d = mkdraw(out) if shot.card: age = i - shot.i0 # recut: the intro is 1 bar (2.5s) now — fade the title card out # *before* the cut instead of letting it pop off (orig window 3.2s) if age < FPS*2.2: al = min(1.0, age/8.0)*min(1.0, (FPS*2.2-age)/12.0) f = font(46, "Georgia Bold.ttf") lw = TL(d, shot.card, f) col = tuple(int(c*al) for c in SEPIA) d.text((W/2-lw/2, H*0.10), shot.card, font=f, fill=col) # the show, engraved small under the title between two hairlines sub = "PLAYER COMPUTER" f2 = font(19, "Georgia Bold.ttf") sw = TL(d, sub, f2) sy = H*0.10 + 58 c2 = tuple(int(c*al) for c in SEP2) d.line([W/2-sw/2-52, sy+11, W/2-sw/2-16, sy+11], fill=c2, width=2) d.line([W/2+sw/2+16, sy+11, W/2+sw/2+52, sy+11], fill=c2, width=2) d.text((W/2-sw/2, sy), sub, font=f2, fill=c2) bh = int(H*0.045) d.rectangle([0, 0, W, bh], fill=(38, 30, 24)); d.rectangle([0, H-bh, W, H], fill=(38, 30, 24)) SB = subs(); t = i/FPS # not `S` — that is the delivery scale idx = np.where((SB["t0"] <= t) & (t < SB["t1"]))[0] if len(idx): line = str(SB["tx"][idx[-1]]) f = font(24, "Georgia Italic.ttf") lw = TL(d, line, f) y0 = H - bh - 44 d.rectangle([W/2-lw/2-12, y0-4, W/2+lw/2+12, y0+30], fill=(236, 226, 204)) d.rectangle([W/2-lw/2-12, y0-4, W/2+lw/2+12, y0+30], outline=SEPIA, width=2) d.text((W/2-lw/2, y0), line, font=f, fill=SEPIA) return out def render_shot(job): shot, force = job E = env() rng = np.random.default_rng(shot.seed) eng = ENGINES[shot.engine](shot, rng) made = 0 for k in range(shot.n): i = shot.i0 + k e = {kk: float(E[kk][min(i, N_FRAMES-1)]) for kk in E} p = FRAMES / f"f{i:05d}.png" u = k/max(1, shot.n-1) arr = eng.frame(k, u, e) # ALWAYS step the engine if p.exists() and not force: continue post(arr, i, e, shot).save(p, compress_level=1) made += 1 return f"shot {shot.idx:02d} {shot.engine:8s} {shot.section:9s} {made}/{shot.n}" def contact_sheet(shots): cols = 6; rows = (len(shots)+cols-1)//cols tw, th = 300, 193 sheet = Image.new("RGB", (cols*tw, rows*(th+24)), (10, 10, 14)) sd = ImageDraw.Draw(sheet) E = env() for n, sh in enumerate(shots): rng = np.random.default_rng(sh.seed) eng = ENGINES[sh.engine](sh, rng) mid = sh.n//2 arr = None for k in range(mid+1): i = sh.i0+k e = {kk: float(E[kk][min(i, N_FRAMES-1)]) for kk in E} arr = eng.frame(k, k/max(1, sh.n-1), e) im = post(arr, sh.i0+mid, e, sh).resize((tw, th), Image.LANCZOS) cx, cy = (n % cols)*tw, (n//cols)*(th+24) sheet.paste(im, (cx, cy)) sd.text((cx+5, cy+th+4), f"{sh.idx:02d} {sh.engine} · {sh.section} · {sh.i0/FPS:.1f}s", font=font(13), fill=(190, 195, 205)) p = OUT/"contact_sheet.png"; sheet.save(p) print(f"contact sheet -> {p} ({len(shots)} shots)") def main(): ap = argparse.ArgumentParser() ap.add_argument("--sheet", action="store_true") ap.add_argument("--shots", default="") ap.add_argument("--force", action="store_true") ap.add_argument("--mux-only", action="store_true") ap.add_argument("--audio-only", action="store_true") ap.add_argument("--jobs", type=int, default=min(14, os.cpu_count())) a = ap.parse_args() wav = AUD/"final.wav" if not wav.exists() or not (AUD/"env.npz").exists() or a.force: print(f"[1/3] song… {N_BARS} bars @ {BPM:.0f}bpm = {DUR:.1f}s") wav, mix = build_song(); analyze(mix) if a.audio_only: print(f"audio -> {wav}"); return shots = build_shots() if a.sheet: contact_sheet(shots); return if not a.mux_only: sel = set(int(x) for x in a.shots.split(",") if x.strip() != "") jobs = [(s, a.force) for s in shots if not sel or s.idx in sel] print(f"[2/3] frames… {len(jobs)} shots / {N_FRAMES} frames on {a.jobs} workers") import multiprocessing as mp with mp.get_context("fork").Pool(a.jobs) as pool: for r in pool.imap_unordered(render_shot, jobs): print(" ", r) print("[3/3] mux…") sd = SETDIR out = OUT/f"{NAME}.mp4" subprocess.run(["ffmpeg", "-y", "-framerate", str(FPS), "-i", str(FRAMES/"f%05d.png"), "-i", str(wav), "-c:v", "libx264", "-preset", "medium", "-crf", "20", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "256k", "-shortest", "-movflags", "+faststart", "-metadata", f"generator=renders/{sd}/{NAME}/render.py", "-metadata", f"title={sd} — {TITLE}", str(out)], check=True, capture_output=True) try: sha = subprocess.check_output(["git", "rev-parse", "--short", "HEAD"], cwd=ROOT).decode().strip() br = subprocess.check_output(["git", "rev-parse", "--abbrev-ref", "HEAD"], cwd=ROOT).decode().strip() except Exception: sha = br = "unknown" (OUT/"PROVENANCE.txt").write_text( f"generator: renders/{sd}/{NAME}/render.py\n" f"git: {sha} branch: {br}\n" f"timestamp: {datetime.datetime.now().astimezone().isoformat()}\n" f"duration: {DUR:.2f}s fps: {FPS} size: {OW}x{OH} (16:9, native)\n" f"music: {MUSIC_DESC}\n" f"sections: {' '.join(n for n,_,_ in SECTIONS)}\n" f"engines: {ENGINE_DESC} (shot-parallel, stateful per shot)\n") print(f"DONE {out} ({DUR:.1f}s)") if __name__ == "__main__": main()