diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ab1ce79 --- /dev/null +++ b/.gitignore @@ -0,0 +1,14 @@ +.venv/ +__pycache__/ +*.pyc +inference_logs/ +proof/frames/ +proof/narration/ +proof/segments/ +proof/work/ +proof/chrome-profile/ +proof/*.mp4 +proof/*.srt +proof/*-check/ +*.egg-info/ +.python-version diff --git a/proof/README.md b/proof/README.md new file mode 100644 index 0000000..c21f7b6 --- /dev/null +++ b/proof/README.md @@ -0,0 +1,42 @@ +# Proof movie — Hermes Function Calling + +A rebuildable movie that proves the function-calling **toolkit** works on +this machine. It does **not** load Hermes-2-Pro (that needs a GPU and +several gigabytes of weights). Yahoo Finance also 429'd from this network, +so the live tool shown is `code_interpreter` rather than a faked stock quote. + +Pipeline is the proving-it-works skill (`prime-radiant-inc/proving-it-works`): +terminal capture via ttyd + tmux + Chrome, local narration, burned-in +subtitles, then `check-movie`. + +## What the movie shows + +| Scene | Real code path | +|---|---| +| tools | `functions.get_openai_tools()` | +| parse | `utils.validate_and_extract_tool_calls()` | +| validate | `validator.validate_function_call_schema()` | +| execute | `functions.code_interpreter.invoke(...)` | + +## Rebuild + +```bash +# once: lightweight venv (no torch / flash-attn) +uv venv .venv +uv pip install --python .venv/bin/python \ + langchain==0.1.9 pydantic==2.6.2 jsonschema==4.21.1 \ + yfinance==0.2.36 pandas==2.2.0 beautifulsoup4 requests art pyyaml \ + pillow websockets + +# ttyd + uv on PATH, Google Chrome installed +proof/make-movie.sh +``` + +`proof/scripts/` are the proving-it-works helpers (MIT), vendored so the +movie can be re-cut without a plugin install. Title/end cards are PNG +stills from `render_cards.py` (PIL) so assemble does not depend on a +browser. + +Frames, narration wavs, and the encoded movie stay out of git — they are +scratch. `scenes.yaml`, `show.py`, `film.py`, `render_cards.py`, +`cards/*.png`, and `make-movie.sh` are the source of truth. diff --git a/proof/cards/end.png b/proof/cards/end.png new file mode 100644 index 0000000..8cde32b Binary files /dev/null and b/proof/cards/end.png differ diff --git a/proof/cards/title.png b/proof/cards/title.png new file mode 100644 index 0000000..f06c088 Binary files /dev/null and b/proof/cards/title.png differ diff --git a/proof/film.py b/proof/film.py new file mode 100755 index 0000000..b2c8ee7 --- /dev/null +++ b/proof/film.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python3 +"""Film a tmux+ttyd terminal of the Hermes toolkit via headless Chrome. + +Adapted from proving-it-works examples/film-terminal.py: real characters in +a real shell, screenshotted from the canvas ttyd draws. Software GL is +required or headless Chrome paints that canvas black. +""" + +from __future__ import annotations + +import asyncio +import base64 +import json +import subprocess +import sys +import time +import urllib.request +from pathlib import Path + +HERE = Path(__file__).resolve().parent +FPS = 2.0 +DEBUG_PORT = 7222 +SESSION = "hermes-demo" +CHROME = "/usr/bin/google-chrome" + + +def pane_command() -> str: + r = subprocess.run( + ["tmux", "display-message", "-p", "-t", SESSION, "#{pane_current_command}"], + capture_output=True, + text=True, + ) + return r.stdout.strip() + + +def wait_for_shell(timeout: float = 120) -> None: + deadline = time.time() + timeout + while time.time() < deadline: + if pane_command() in ("bash", "sh", "zsh"): + return + time.sleep(0.4) + raise SystemExit("shell never came back; refusing to type into a running program") + + +def send(keys: str, enter: bool = True) -> None: + cmd = ["tmux", "send-keys", "-t", SESSION, keys] + if enter: + cmd.append("Enter") + subprocess.run(cmd, check=True, capture_output=True) + + +def send_and_wait(keys: str, timeout: float = 120) -> None: + """Type a command, wait for it to start, then wait for the shell to return.""" + send(keys) + deadline = time.time() + 8 + while time.time() < deadline and pane_command() in ("bash", "sh", "zsh"): + time.sleep(0.1) + wait_for_shell(timeout=timeout) + + +async def cdp(ws, mid, method, params=None): + await ws.send(json.dumps({"id": mid, "method": method, "params": params or {}})) + while True: + msg = json.loads(await ws.recv()) + if msg.get("id") == mid: + return msg.get("result", {}) + + +async def shoot(ws, outdir: Path, stop: asyncio.Event, counter: list[int]) -> None: + import websockets # noqa: F401 — imported by caller + + mid = 1000 + while not stop.is_set(): + mid += 1 + try: + res = await asyncio.wait_for( + cdp(ws, mid, "Page.captureScreenshot", {"format": "png"}), timeout=3 + ) + data = res.get("data") + if data: + (outdir / f"f{counter[0]:05d}.png").write_bytes(base64.b64decode(data)) + counter[0] += 1 + except asyncio.TimeoutError: + pass + await asyncio.sleep(1 / FPS) + + +async def preflight(ws) -> None: + wait_for_shell() + send("echo PREFLIGHT_OK") + await asyncio.sleep(1.5) + res = await cdp(ws, 900, "Page.captureScreenshot", {"format": "png"}) + png = base64.b64decode(res["data"]) + from io import BytesIO + + from PIL import Image + + im = Image.open(BytesIO(png)).convert("L") + px = list(im.getdata()) + lit = sum(1 for v in px if v > 90) / len(px) + if lit < 0.002: + raise SystemExit( + f"preflight failed: terminal renders blank ({lit:.4%} lit pixels). " + "Check software GL flags before filming." + ) + print(f"preflight ok: {lit:.2%} of pixels lit") + send("clear") + await asyncio.sleep(0.8) + + +async def beats_tools() -> None: + await asyncio.sleep(1.2) + send_and_wait("python proof/show.py tools") + await asyncio.sleep(5) + + +async def beats_parse() -> None: + await asyncio.sleep(1.0) + send("clear") + await asyncio.sleep(0.6) + send_and_wait("python proof/show.py parse") + await asyncio.sleep(5) + + +async def beats_validate() -> None: + await asyncio.sleep(1.0) + send("clear") + await asyncio.sleep(0.6) + send_and_wait("python proof/show.py validate") + await asyncio.sleep(6) + + +async def beats_execute() -> None: + await asyncio.sleep(1.0) + send("clear") + await asyncio.sleep(0.6) + send_and_wait("python proof/show.py execute") + await asyncio.sleep(6) + + +BEATS = { + "tools": beats_tools, + "parse": beats_parse, + "validate": beats_validate, + "execute": beats_execute, +} + + +async def main(segment: str, url: str) -> None: + import websockets + + if segment not in BEATS: + raise SystemExit(f"unknown segment {segment!r}; expected {sorted(BEATS)}") + + outdir = HERE / "frames" / segment + outdir.mkdir(parents=True, exist_ok=True) + for old in outdir.glob("*.png"): + old.unlink() + + # Drop a leftover debug Chrome so this take owns DEBUG_PORT. + subprocess.run( + ["pkill", "-f", f"--remote-debugging-port={DEBUG_PORT}"], + capture_output=True, + ) + time.sleep(0.4) + + profile = HERE / "chrome-profile" + profile.mkdir(exist_ok=True) + chrome = subprocess.Popen( + [ + CHROME, + f"--remote-debugging-port={DEBUG_PORT}", + "--headless=new", + "--user-data-dir=" + str(profile), + "--window-size=1280,800", + "--hide-scrollbars", + "--no-sandbox", + "--force-device-scale-factor=2", + "--use-gl=angle", + "--use-angle=swiftshader", + "--enable-unsafe-swiftshader", + "about:blank", + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + time.sleep(3) + tabs = json.loads(urllib.request.urlopen(f"http://127.0.0.1:{DEBUG_PORT}/json").read()) + ws_url = [t for t in tabs if t["type"] == "page"][0]["webSocketDebuggerUrl"] + async with websockets.connect(ws_url, max_size=40 * 1024 * 1024) as ws: + await cdp(ws, 1, "Page.enable") + await cdp(ws, 3, "Page.navigate", {"url": url}) + await asyncio.sleep(4) + subprocess.run( + ["tmux", "refresh-client", "-t", SESSION], capture_output=True + ) + await asyncio.sleep(1) + await preflight(ws) + + stop = asyncio.Event() + counter = [0] + task = asyncio.create_task(shoot(ws, outdir, stop, counter)) + await BEATS[segment]() + stop.set() + await task + print(f"{segment}: {counter[0]} frames -> {outdir}") + finally: + chrome.terminate() + try: + chrome.wait(timeout=5) + except Exception: + chrome.kill() + + +if __name__ == "__main__": + if len(sys.argv) != 3: + raise SystemExit("usage: film.py ") + asyncio.run(main(sys.argv[1], sys.argv[2])) diff --git a/proof/make-movie.sh b/proof/make-movie.sh new file mode 100755 index 0000000..a060fc7 --- /dev/null +++ b/proof/make-movie.sh @@ -0,0 +1,55 @@ +#!/usr/bin/env bash +# Rebuild the Hermes Function Calling proof movie. +# Requires: ffmpeg, ffprobe, uv, ttyd, Google Chrome, the repo .venv +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +PROOF="$ROOT/proof" +SKILL="$PROOF/scripts" +SESSION=hermes-demo +TTYD_PORT=7681 +export PATH="$HOME/.local/bin:$ROOT/.venv/bin:$PATH" + +cd "$ROOT" + +if [[ ! -x "$ROOT/.venv/bin/python" ]]; then + echo "missing $ROOT/.venv — create it and pip-install langchain, pydantic, jsonschema, yfinance, pandas, art, pyyaml, pillow, websockets" >&2 + exit 1 +fi + +# --- terminal session served over HTTP --------------------------------- +tmux has-session -t "$SESSION" 2>/dev/null || \ + tmux new-session -d -s "$SESSION" -x 125 -y 34 -- bash -l + +# Point the pane at the repo with a short prompt. +tmux send-keys -t "$SESSION" "cd $ROOT" Enter +tmux send-keys -t "$SESSION" "export PATH=$ROOT/.venv/bin:\$HOME/.local/bin:\$PATH" Enter +tmux send-keys -t "$SESSION" "export PYTHONUNBUFFERED=1 PYTHONWARNINGS=ignore" Enter +tmux send-keys -t "$SESSION" "export PS1='$ '" Enter +sleep 0.4 + +if ! curl -fsS "http://127.0.0.1:${TTYD_PORT}/" >/dev/null 2>&1; then + ttyd -p "$TTYD_PORT" \ + -t fontSize=17 \ + -t 'fontFamily=DejaVu Sans Mono,monospace' \ + -t 'theme={"background":"#101014","foreground":"#e8e6e1"}' \ + tmux attach -t "$SESSION" >/tmp/ttyd-hermes.log 2>&1 & + # wait until it answers + for _ in $(seq 1 30); do + curl -fsS "http://127.0.0.1:${TTYD_PORT}/" >/dev/null 2>&1 && break + sleep 0.3 + done +fi + +python "$PROOF/render_cards.py" +python "$PROOF/film.py" tools "http://127.0.0.1:${TTYD_PORT}/" +python "$PROOF/film.py" parse "http://127.0.0.1:${TTYD_PORT}/" +python "$PROOF/film.py" validate "http://127.0.0.1:${TTYD_PORT}/" +python "$PROOF/film.py" execute "http://127.0.0.1:${TTYD_PORT}/" + +"$SKILL/narrate" "$PROOF/scenes.yaml" "$PROOF/narration/" +"$SKILL/assemble" "$PROOF/scenes.yaml" "$PROOF/silent-cut.mp4" +"$SKILL/make-subtitles" "$PROOF/narration/manifest.json" "$PROOF/movie.srt" \ + --offsets-json "$PROOF/segments/offsets.json" +"$SKILL/burn-subtitles" "$PROOF/silent-cut.mp4" "$PROOF/movie.srt" "$PROOF/movie.mp4" +"$SKILL/check-movie" "$PROOF/movie.mp4" diff --git a/proof/render_cards.py b/proof/render_cards.py new file mode 100755 index 0000000..c66bd7d --- /dev/null +++ b/proof/render_cards.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Render title/end cards as PNG so assemble does not need a browser.""" + +from pathlib import Path + +from PIL import Image, ImageDraw, ImageFont + +HERE = Path(__file__).resolve().parent +OUT = HERE / "cards" +W, H = 1280, 800 +BG = (16, 16, 20) +FG = (242, 242, 245) +SUB = (154, 154, 166) + +MONO = "/usr/share/fonts/truetype/dejavu/DejaVuSansMono-Bold.ttf" +SANS = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf" + + +def wrap(draw, text, font, max_width): + words = text.split() + lines, cur = [], "" + for w in words: + trial = (cur + " " + w).strip() + if draw.textlength(trial, font=font) <= max_width: + cur = trial + else: + if cur: + lines.append(cur) + cur = w + if cur: + lines.append(cur) + return lines + + +def card(title: str, subtitle: str, dest: Path) -> None: + img = Image.new("RGB", (W, H), BG) + draw = ImageDraw.Draw(img) + title_font = ImageFont.truetype(MONO, 42) + sub_font = ImageFont.truetype(SANS, 22) + max_w = int(W * 0.84) + t_lines = wrap(draw, title, title_font, max_w) + s_lines = wrap(draw, subtitle, sub_font, max_w) + gap = 22 + t_h = 50 + s_h = 32 + block = len(t_lines) * t_h + gap + len(s_lines) * s_h + y = (H - block) // 2 + for line in t_lines: + tw = draw.textlength(line, font=title_font) + draw.text(((W - tw) / 2, y), line, font=title_font, fill=FG) + y += t_h + y += gap + for line in s_lines: + tw = draw.textlength(line, font=sub_font) + draw.text(((W - tw) / 2, y), line, font=sub_font, fill=SUB) + y += s_h + dest.parent.mkdir(parents=True, exist_ok=True) + img.save(dest) + print(dest, dest.stat().st_size) + + +def main() -> None: + card( + "Hermes Function Calling", + "proving the toolkit — schemas, parser, validator, live tool execution", + OUT / "title.png", + ) + card( + "the contract holds without the GPU", + "schemas · parse · validate · execute — github.com/NousResearch/Hermes-Function-Calling", + OUT / "end.png", + ) + + +if __name__ == "__main__": + main() diff --git a/proof/scenes.yaml b/proof/scenes.yaml new file mode 100644 index 0000000..8a2f117 --- /dev/null +++ b/proof/scenes.yaml @@ -0,0 +1,57 @@ +fps: 30 +resolution: + width: 1280 + height: 800 +scenes: + - id: title + kind: image + src: cards/title.png + duration: 4 + narration: >- + This is Hermes Function Calling. The eight billion parameter model is + not on this machine. The toolkit that wraps it is, and that is what we + are about to run for real. + + - id: tools + kind: frames + src: frames/tools + rate: 2.0 + narration: >- + functions.py turns Python callables into OpenAI tool schemas. + Eleven tools come back. Here is get stock fundamentals: a symbol in, + a typed object out. + + - id: parse + kind: frames + src: frames/parse + rate: 2.0 + narration: >- + Hermes writes function calls inside tool call XML tags. The parser + pulls the JSON out of that markup so the rest of the stack can run it. + + - id: validate + kind: frames + src: frames/validate + rate: 2.0 + narration: >- + The validator checks every call against those live schemas. A Tesla + fundamentals request with a symbol is accepted. The same call with no + symbol is rejected. A function that does not exist is rejected too. + + - id: execute + kind: frames + src: frames/execute + rate: 2.0 + narration: >- + Yahoo finance rate limited this machine, so we did not fake a stock + quote. Instead we run the other first party tool, code interpreter, + on real Python. Squares of one through seven, total one hundred forty. + + - id: end + kind: image + src: cards/end.png + duration: 4.5 + narration: >- + Schema conversion, XML parse, argument checks, and a live tool. That + is the function calling contract. The language model is a caller, not + the implementation. diff --git a/proof/scripts/assemble b/proof/scripts/assemble new file mode 100755 index 0000000..c7834d0 --- /dev/null +++ b/proof/scripts/assemble @@ -0,0 +1,209 @@ +#!/usr/bin/env -S uv run --quiet --script +# /// script +# requires-python = ">=3.10" +# dependencies = ["pyyaml"] +# /// +"""Assemble scenes into one movie, each segment held to max(narration, visuals). + +Reads the same scenes file narrate does, so the narration you rendered and +the picture you recorded stay in step by construction: a segment lasts as +long as whichever of its two halves is longer, and the short one is padded +(video freezes its last frame, audio pads with silence). + +It also writes segments/offsets.json — where each scene starts in the final +cut — which make-subtitles consumes. Hand-computing those offsets is the +step that silently breaks every time you insert or reorder a scene. + +Scene kinds: + card title/caption rendered as HTML and screenshotted (needs a browser) + image a still you already have (a contact sheet, a diagram) + frames a directory of PNGs, played at `rate` fps + movie an existing movie, played as itself with its own audio + +Usage: + assemble SCENES.yaml OUT.mp4 [--narration DIR] [--work DIR] [--browser PATH] +""" + +import argparse +import json +import os +import shutil +import subprocess +import sys +from pathlib import Path + +import yaml + +BROWSERS = [ + "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome", + "/Applications/Chromium.app/Contents/MacOS/Chromium", + "chromium", "chromium-browser", "google-chrome", "google-chrome-stable", +] + +CARD_HTML = """ +

{TITLE}

{SUB}

+""" + + +def die(msg): + print(f"assemble: {msg}", file=sys.stderr) + sys.exit(1) + + +def run(cmd, timeout=None): + try: + r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + except subprocess.TimeoutExpired: + die(f"timed out ({timeout}s): {' '.join(map(str, cmd))[:240]}") + if r.returncode != 0: + die(f"{' '.join(map(str, cmd))}\n{r.stderr.strip()[:500]}") + return r + + +def dur(path): + r = run(["ffprobe", "-v", "error", "-show_entries", "format=duration", + "-of", "csv=p=0", str(path)]) + return float(r.stdout.strip()) + + +def find_browser(explicit): + for cand in ([explicit] if explicit else []) + BROWSERS: + if not cand: + continue + if os.path.sep in cand and Path(cand).exists(): + return cand + found = shutil.which(cand) + if found: + return found + return None + + +def make_card(scene, png, w, h, browser): + if not browser: + die("a `card` scene needs a browser (Chrome/Chromium) to render text; " + "pass --browser, or use an `image` scene you rendered yourself") + html = CARD_HTML.format( + w=w, h=h, bg=scene.get("background", "#101014"), + gap=max(16, h // 44), title=scene.get("title_size", max(28, h // 14)), + sub=scene.get("subtitle_size", max(16, h // 32)), + TITLE=scene.get("title", ""), SUB=scene.get("subtitle", "")) + tmp = png.with_suffix(".html") + tmp.write_text(html) + # as_uri() yields file:///abs/path — quote(str(path)) becomes file://rel + # which Chrome treats as a host and hangs on ERR_INVALID_URL. + url = Path(tmp).resolve().as_uri() + run([browser, "--headless=new", "--no-sandbox", "--disable-gpu", + "--hide-scrollbars", f"--screenshot={png}", f"--window-size={w},{h}", + "--force-device-scale-factor=1", url], timeout=45) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("scenes", type=Path) + ap.add_argument("out", type=Path) + ap.add_argument("--narration", type=Path, default=None) + ap.add_argument("--work", type=Path, default=None) + ap.add_argument("--browser", default=None) + args = ap.parse_args() + + for tool in ("ffmpeg", "ffprobe"): + if not shutil.which(tool): + die(f"{tool} not on PATH") + + doc = yaml.safe_load(args.scenes.read_text()) + base = args.scenes.parent + res = doc.get("resolution", {}) or {} + W, H = int(res.get("width", 1920)), int(res.get("height", 1080)) + FPS = int(doc.get("fps", 30)) + narration = args.narration or (base / "narration") + work = args.work or (base / "segments") + work.mkdir(parents=True, exist_ok=True) + browser = find_browser(args.browser) + + fit = (f"scale={W}:{H}:force_original_aspect_ratio=decrease," + f"pad={W}:{H}:(ow-iw)/2:(oh-ih)/2:color=#101014,setsar=1") + + offsets, clock, concat_lines = {}, 0.0, [] + for sc in doc["scenes"]: + sid = sc["id"] + kind = sc.get("kind", "frames") + seg = work / f"{sid}.mp4" + nar = narration / f"{sid}.wav" + nard = dur(nar) if nar.exists() else 0.0 + + if kind == "movie": + src = base / sc["src"] + target = dur(src) + inner_h = int(sc.get("height", int(H * 0.82))) + run(["ffmpeg", "-nostdin", "-y", "-v", "error", "-i", str(src), + "-vf", f"scale=-2:{inner_h},pad={W}:{H}:(ow-iw)/2:(oh-ih)/2:" + f"color=#101014,setsar=1", + "-af", f"volume={sc.get('gain_db', 0)}dB,apad", + "-r", str(FPS), "-t", f"{target:.3f}", + "-c:v", "libx264", "-preset", "medium", "-pix_fmt", "yuv420p", + "-c:a", "aac", "-ar", "44100", "-ac", "2", str(seg)]) + else: + if kind == "frames": + src = base / sc["src"] + n = len(list(Path(src).glob("*.png"))) + if not n: + die(f"scene {sid}: no PNGs in {src}") + rate = float(sc.get("rate", FPS)) + vis = n / rate + target = max(nard, vis) + vin = ["-framerate", str(rate), "-pattern_type", "glob", + "-i", str(Path(src) / "*.png")] + # freeze the last frame when narration outlasts the action + vf = fit + f",tpad=stop_mode=clone:stop_duration={max(0.0, target - vis):.3f}" + else: + if kind == "card": + img = work / f"card-{sid}.png" + make_card(sc, img, W, H, browser) + elif kind == "image": + img = base / sc["src"] + if not img.exists(): + die(f"scene {sid}: no such image {img}") + else: + die(f"scene {sid}: unknown kind {kind!r}") + target = max(nard, float(sc.get("duration", 3))) + vin = ["-loop", "1", "-i", str(img)] + vf = fit + + ain = (["-i", str(nar)] if nar.exists() + else ["-f", "lavfi", "-i", "anullsrc=r=44100:cl=stereo"]) + run(["ffmpeg", "-nostdin", "-y", "-v", "error", *vin, *ain, + "-vf", vf, "-af", "apad", "-r", str(FPS), "-t", f"{target:.3f}", + "-map", "0:v:0", "-map", "1:a:0", + "-c:v", "libx264", "-preset", "medium", "-pix_fmt", "yuv420p", + "-c:a", "aac", "-ar", "44100", "-ac", "2", str(seg)]) + + actual = dur(seg) + # only scenes that speak get a subtitle offset; a movie played as + # itself carries its own subtitles already + if nar.exists() and kind != "movie": + offsets[sid] = round(clock, 3) + clock += actual + concat_lines.append(f"file '{seg.resolve()}'") + print(f"{sid}: {actual:.1f}s{' (own audio)' if kind == 'movie' else ''}") + + listing = work / "concat.txt" + listing.write_text("\n".join(concat_lines) + "\n") + run(["ffmpeg", "-nostdin", "-y", "-v", "error", "-f", "concat", "-safe", "0", + "-i", str(listing), "-c", "copy", str(args.out)]) + (work / "offsets.json").write_text(json.dumps(offsets, indent=2)) + print(f"\nassembled {args.out} ({dur(args.out):.1f}s)") + print(f"scene offsets -> {work / 'offsets.json'} " + f"(feed to make-subtitles --offsets-json)") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/proof/scripts/burn-subtitles b/proof/scripts/burn-subtitles new file mode 100755 index 0000000..06b5f20 --- /dev/null +++ b/proof/scripts/burn-subtitles @@ -0,0 +1,100 @@ +#!/usr/bin/env -S uv run --quiet --script +# /// script +# requires-python = ">=3.10" +# /// +"""Put subtitles on a movie, by whichever route this ffmpeg supports. + +Burning them into the picture is what you want: subtitles survive Slack, +PR previews, phones, and anything that plays video without a subtitle UI. +That needs an ffmpeg built with libass, which many are not — Homebrew's +default macOS build has no `subtitles` filter at all, while Debian's does. +Rather than emit a command that works on half of machines, this checks and +falls back to an embedded soft-subtitle track, telling you which you got. + +Usage: + burn-subtitles IN.mp4 SUBS.srt OUT.mp4 [--font NAME] [--size N] + [--soft] [--margin PX] +""" + +import argparse +import shutil +import subprocess +import sys +from pathlib import Path + + +def has_libass(): + out = subprocess.run(["ffmpeg", "-hide_banner", "-filters"], + capture_output=True, text=True) + return any(line.split()[1:2] == ["subtitles"] + for line in out.stdout.splitlines() if line.strip()) + + +def run(cmd, cwd=None): + r = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd) + if r.returncode != 0: + print(" ".join(map(str, cmd)), file=sys.stderr) + print(r.stderr.strip()[:600], file=sys.stderr) + return r.returncode == 0 + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("movie", type=Path) + ap.add_argument("subs", type=Path) + ap.add_argument("out", type=Path) + ap.add_argument("--font", default="DejaVu Sans") + ap.add_argument("--size", type=int, default=16) + ap.add_argument("--margin", type=int, default=30) + ap.add_argument("--soft", action="store_true", + help="embed a soft track even if burning is available") + args = ap.parse_args() + + if not shutil.which("ffmpeg"): + sys.exit("ffmpeg not on PATH") + for f in (args.movie, args.subs): + if not f.exists(): + sys.exit(f"no such file: {f}") + + if not args.soft and has_libass(): + # ffmpeg 8 dropped positional filter options, so name it explicitly: + # `subtitles=movie.srt` parses on 5.x and fails on 8.x, but + # `subtitles=filename=movie.srt` works on both + # BorderStyle=3 draws a filled box behind the text. Outline-only + # subtitles are legible over a dark terminal and marginal over a + # white app screenshot; a demo movie cuts between both. + style = (f"FontName={args.font},Fontsize={args.size}," + f"BorderStyle=3,Outline=1,Shadow=0,MarginV={args.margin}," + f"PrimaryColour=&H00FFFFFF&,OutlineColour=&HB0101014&," + f"BackColour=&HB0101014&") + # run from the subtitle's directory: the filter treats ':' and '\' in + # paths as its own syntax, and quoting around that is a losing game + ok = run(["ffmpeg", "-nostdin", "-y", "-v", "error", + "-i", str(args.movie.resolve()), + "-vf", f"subtitles=filename={args.subs.name}:" + f"force_style='{style}'", + "-c:a", "copy", "-c:v", "libx264", "-preset", "medium", + "-pix_fmt", "yuv420p", str(args.out.resolve())], + cwd=str(args.subs.resolve().parent)) + if ok: + print(f"burned into the picture -> {args.out}") + return 0 + print("burn failed; falling back to a soft track", file=sys.stderr) + + ok = run(["ffmpeg", "-nostdin", "-y", "-v", "error", + "-i", str(args.movie), "-i", str(args.subs), + "-c", "copy", "-c:s", "mov_text", + "-metadata:s:s:0", "language=eng", str(args.out)]) + if not ok: + return 1 + print(f"embedded a soft subtitle track -> {args.out}") + if not args.soft: + print("NOTE: this ffmpeg has no libass, so the subtitles are a track a " + "player must choose to show, not pixels. Anything that autoplays " + "without subtitle UI (Slack, PR previews) will show none. Install " + "an ffmpeg with libass to burn them in.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/proof/scripts/check-movie b/proof/scripts/check-movie new file mode 100755 index 0000000..ad4cf05 --- /dev/null +++ b/proof/scripts/check-movie @@ -0,0 +1,261 @@ +#!/usr/bin/env -S uv run --quiet --script +# /// script +# requires-python = ">=3.10" +# dependencies = ["pillow"] +# /// +"""Mechanical gate for a proof/demo movie: catches the silent defects that +per-frame inspection structurally cannot see. + +A movie can pass every frame check and still be unwatchable, because the +defects live *between* frames: action crammed into the first seconds, a +narrator talking over a picture that died, a silent audio track. This +samples the picture and the sound on the same timeline and compares them. + +Thresholds are heuristics tuned against real good and bad movies. They +catch the egregious cases; they cannot tell you a movie is *right*. That is +what the contact sheet is for, and you have to actually look at it. + +Known blind spot: the picture is sampled at 1 Hz, so a visual beat shorter +than a second (a flash, a blank frame during a reload) falls between samples +and reads as "no change". Hold anything that matters for >1s. + +Usage: + check-movie MOVIE [--out DIR] [--no-expect-audio] + [--no-expect-subtitles] [--subs FILE] [--json] +""" + +import argparse +import array +import json +import math +import shutil +import subprocess +import sys +from pathlib import Path + +from PIL import Image + +THUMB_W = 320 # sampling width; the metric is a pixel fraction, so scale-free +PIXEL_DELTA = 8 # per-pixel grey delta that counts as "this pixel moved" +CHANGE_FRAC = 0.002 # >0.2% of pixels moved => the picture reached a new state +SPEECH_DB = -45.0 # windowed RMS above this counts as "someone is talking" +EARLY_ACTION = 0.40 # last change before this fraction of runtime => front-loaded +TAIL_TALK_S = 5.0 # ...and this many seconds of narration after it => broken +WARN_TAIL_S = 15.0 # frozen tail worth mentioning even when it passes +WARN_GAP_S = 30.0 # hold this long mid-movie and a viewer wonders if it froze + + +def die(msg): + print(f"FAIL {msg}") + sys.exit(2) + + +def grey(path): + with Image.open(path) as im: + return list(im.convert("L").tobytes()) + + +def sample_picture(movie, workdir): + """Per-second: fraction of pixels that moved since the previous second.""" + frames = workdir / "samples" + frames.mkdir(parents=True, exist_ok=True) + for old in frames.glob("*.png"): + old.unlink() + out = subprocess.run( + ["ffmpeg", "-nostdin", "-v", "error", "-i", str(movie), + "-vf", f"fps=1,scale={THUMB_W}:-1", "-f", "image2", str(frames / "s%05d.png")], + capture_output=True, text=True) + if out.returncode != 0: + die(f"frame sampling failed: {out.stderr.strip()[:200]}") + paths = sorted(frames.glob("s*.png")) + if not paths: + die("no video frames could be sampled") + fracs, prev = [], None + for p in paths: + px = grey(p) + if prev is not None: + n = min(len(px), len(prev)) + moved = sum(1 for i in range(n) if abs(px[i] - prev[i]) > PIXEL_DELTA) + fracs.append(moved / n) + prev = px + return paths, fracs + + +def sample_sound(movie, has_audio): + """Per-second RMS in dBFS.""" + if not has_audio: + return [] + out = subprocess.run( + ["ffmpeg", "-nostdin", "-v", "error", "-i", str(movie), + "-map", "0:a:0", "-ac", "1", "-ar", "8000", "-f", "s16le", "-"], + capture_output=True) + if out.returncode != 0 or not out.stdout: + die(f"audio decode failed: {out.stderr.decode()[:200]}") + pcm = array.array("h") + pcm.frombytes(out.stdout[: len(out.stdout) // 2 * 2]) + levels = [] + for start in range(0, len(pcm), 8000): + chunk = pcm[start:start + 8000] + if not chunk: + break + rms = math.sqrt(sum(float(s) * s for s in chunk) / len(chunk)) + levels.append(20 * math.log10(rms / 32768.0) if rms > 0 else -120.0) + return levels + + +def contact_sheet(paths, out_path, count=12): + picks = paths if len(paths) <= count else [ + paths[round(i * (len(paths) - 1) / (count - 1))] for i in range(count)] + thumbs = [Image.open(p).convert("RGB") for p in picks] + w, h = thumbs[0].size + # pick a column count that fills the grid exactly where possible: an + # empty cell reads as a black *frame*, which is a defect signal, and a + # sheet that lies about the movie defeats the point of the sheet + n = len(thumbs) + cols = next((c for c in (4, 3, 5, 2) if n % c == 0), min(4, n)) + rows = math.ceil(n / cols) + sheet = Image.new("RGB", (cols * w, rows * h), (48, 48, 52)) + for i, t in enumerate(thumbs): + sheet.paste(t, ((i % cols) * w, (i // cols) * h)) + sheet.save(out_path) + return [paths.index(p) for p in picks] + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("movie", type=Path) + ap.add_argument("--out", type=Path, default=None) + ap.add_argument("--no-expect-audio", dest="expect_audio", + action="store_false", default=True) + ap.add_argument("--no-expect-subtitles", dest="expect_subs", + action="store_false", default=True) + ap.add_argument("--subs", type=Path, default=None, + help="sidecar .srt (default: MOVIE.srt beside the movie)") + ap.add_argument("--json", action="store_true") + args = ap.parse_args() + + if not args.movie.exists(): + die(f"no such movie: {args.movie}") + for tool in ("ffmpeg", "ffprobe"): + if not shutil.which(tool): + die(f"{tool} not on PATH") + + workdir = args.out or args.movie.parent / f"{args.movie.stem}-check" + workdir.mkdir(parents=True, exist_ok=True) + + meta = subprocess.run( + ["ffprobe", "-v", "error", "-print_format", "json", + "-show_format", "-show_streams", str(args.movie)], + capture_output=True, text=True) + if meta.returncode != 0: + die(f"ffprobe failed: {meta.stderr.strip()[:200]}") + info = json.loads(meta.stdout) + vs = [s for s in info["streams"] if s["codec_type"] == "video"] + as_ = [s for s in info["streams"] if s["codec_type"] == "audio"] + if not vs: + die("no video stream") + duration = float(info["format"].get("duration", 0)) + + paths, fracs = sample_picture(args.movie, workdir) + levels = sample_sound(args.movie, bool(as_)) + changes = [i for i, f in enumerate(fracs) if f > CHANGE_FRAC] + talking = [i for i, lv in enumerate(levels) if lv >= SPEECH_DB] + span = len(fracs) or 1 + last_change = changes[-1] if changes else None + last_talk = talking[-1] if talking else None + + print(f"container {vs[0]['codec_name']} {vs[0]['width']}x{vs[0]['height']}, " + f"{duration:.1f}s, audio={'yes' if as_ else 'no'}") + print(f"picture reaches a new state in {len(changes)} of {span} seconds" + + (f"; last at {last_change}s" if last_change is not None else "")) + if levels: + print(f"sound audible in {len(talking)} of {len(levels)} seconds" + + (f"; last at {last_talk}s" if last_talk is not None else "")) + + failures, warnings = [], [] + if duration < 1: + failures.append(f"duration is {duration:.2f}s - that is not a movie") + if args.expect_audio and not as_: + failures.append("expected narration but there is no audio stream") + if levels and not talking: + failures.append("the audio track is silent end to end") + + # a narrated movie with no subtitles fails for everyone watching it muted + if as_ and args.expect_subs: + srt = args.subs or args.movie.with_suffix(".srt") + embedded = any(s["codec_type"] == "subtitle" for s in info["streams"]) + if srt.exists(): + last = 0.0 + for line in srt.read_text(errors="replace").splitlines(): + if "-->" in line: + end = line.split("-->")[1].strip().split()[0] + hh, mm, rest = end.split(":") + ss, _, ms = rest.partition(",") + last = max(last, int(hh) * 3600 + int(mm) * 60 + int(ss) + + int(ms or 0) / 1000) + # compare against where the narration ends, not the runtime: a + # silent end card is normal and must not read as missing subtitles + speech_end = float(last_talk + 1) if last_talk is not None else duration + print(f"subtitles {srt.name}, last cue ends at {last:.1f}s " + f"(narration ends {speech_end:.0f}s)") + if last < speech_end - 3.0: + failures.append( + f"subtitles stop at {last:.0f}s but the narration runs to " + f"{speech_end:.0f}s - {speech_end - last:.0f}s of speech " + f"has no subtitles") + elif embedded: + print("subtitles embedded subtitle stream present") + else: + failures.append( + f"narrated, but no subtitles: expected {srt.name} beside the " + f"movie (or an embedded track). Run make-subtitles and burn " + f"them in; pass --no-expect-subtitles only for a movie nobody " + f"will ever watch muted.") + if not changes: + failures.append("the picture never reaches a new state - this is a still, " + "not a movie") + else: + tail_talk = (last_talk - last_change) if last_talk is not None else 0 + frozen_frac = (span - last_change) / span + if last_change < EARLY_ACTION * span and tail_talk > TAIL_TALK_S: + failures.append( + f"every visible change happens in the first {last_change}s " + f"({100*last_change/span:.0f}% of runtime), then the picture is " + f"frozen for {span - last_change}s while narration keeps talking " + f"for {tail_talk:.0f}s of it. The demo is over before the " + f"explanation starts: pace the action to the narration.") + elif tail_talk > WARN_TAIL_S: + warnings.append(f"{tail_talk:.0f}s of narration after the last visible " + f"change ({100*frozen_frac:.0f}% of runtime frozen)") + gaps = [changes[i + 1] - changes[i] for i in range(len(changes) - 1)] + if gaps and max(gaps) > WARN_GAP_S: + warnings.append(f"{max(gaps)}s with no visible change mid-movie - " + f"intentional hold, or did something hang?") + + sheet = workdir / "contact-sheet.png" + idxs = contact_sheet(paths, sheet) + print(f"sheet {sheet}") + print(f" sampled at {', '.join(str(i) + 's' for i in idxs)}") + + for w in warnings: + print(f"WARN {w}") + for f in failures: + print(f"FAIL {f}") + + if args.json: + (workdir / "check.json").write_text(json.dumps( + {"duration": duration, "change_seconds": changes, + "talk_seconds": talking, "failures": failures, + "warnings": warnings}, indent=2)) + + if failures: + print("\nNOT SHIPPABLE. Fix, regenerate, re-run.") + return 1 + print("\nMechanical checks pass. NOW OPEN THE CONTACT SHEET AND LOOK AT IT: " + "this script cannot see wrong content, unreadable text, a missing " + "cursor, or narration that says something the picture contradicts.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/proof/scripts/make-subtitles b/proof/scripts/make-subtitles new file mode 100755 index 0000000..7c73473 --- /dev/null +++ b/proof/scripts/make-subtitles @@ -0,0 +1,125 @@ +#!/usr/bin/env -S uv run --quiet --script +# /// script +# requires-python = ">=3.10" +# /// +"""Build an SRT from narrate's manifest, timed to the measured clips. + +Subtitles are not decoration. A movie gets watched muted - in a PR, on a +phone, in an open-plan office, by someone who is deaf - and an unsubtitled +narrated movie simply doesn't communicate to those viewers. They also make +the movie searchable and let a reviewer check what was said without +listening. + +Cue timing is proportional to character count within each scene's measured +audio, which tracks speech closely enough for reading. If you need +word-exact timing, transcribe the rendered audio with a word-timestamp API +and use those offsets instead. + +Usage: + make-subtitles MANIFEST.json OUT.srt [--offsets SCENE=SECONDS ...] + [--max-chars N] [--max-secs S] +""" + +import argparse +import json +import sys +from pathlib import Path + +MAX_CHARS = 84 # two comfortable lines +MAX_SECS = 5.5 +MIN_SECS = 1.0 + + +def cue_chunks(text, max_chars): + """Split into cue-sized pieces on sentence, then clause, then word.""" + words, chunks, cur = text.split(), [], "" + for w in words: + candidate = f"{cur} {w}".strip() + if len(candidate) > max_chars and cur: + chunks.append(cur) + cur = w + else: + cur = candidate + if cur.endswith((".", "!", "?")) and len(cur) > max_chars * 0.45: + chunks.append(cur) + cur = "" + if cur: + chunks.append(cur) + return chunks or [text] + + +def wrap(line, width=42): + words, out, cur = line.split(), [], "" + for w in words: + if len(f"{cur} {w}".strip()) > width and cur: + out.append(cur) + cur = w + else: + cur = f"{cur} {w}".strip() + if cur: + out.append(cur) + return "\n".join(out[:2]) if len(out) <= 2 else "\n".join( + [" ".join(out[:len(out) // 2]), " ".join(out[len(out) // 2:])]) + + +def ts(seconds): + ms = int(round(seconds * 1000)) + h, ms = divmod(ms, 3600000) + m, ms = divmod(ms, 60000) + s, ms = divmod(ms, 1000) + return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("manifest", type=Path) + ap.add_argument("out", type=Path) + ap.add_argument("--offsets", nargs="*", default=[], + help="SCENE=SECONDS start overrides; without these, scenes " + "are assumed to run back to back in manifest order") + ap.add_argument("--offsets-json", type=Path, default=None, + help="segments/offsets.json from assemble - the reliable " + "way to time cues against the finished cut") + ap.add_argument("--max-chars", type=int, default=MAX_CHARS) + ap.add_argument("--max-secs", type=float, default=MAX_SECS) + args = ap.parse_args() + + manifest = json.loads(args.manifest.read_text()) + overrides = {} + if args.offsets_json: + overrides.update({k: float(v) for k, v in + json.loads(args.offsets_json.read_text()).items()}) + for spec in args.offsets: + k, _, v = spec.partition("=") + overrides[k] = float(v) + + # a scene with no offset and no place in the cut would silently land at + # the wrong time; skip it rather than mistime it + if overrides: + manifest = [e for e in manifest if e["id"] in overrides] + cues, clock = [], 0.0 + for entry in manifest: + start = overrides.get(entry["id"], clock) + dur = float(entry["duration"]) + chunks = cue_chunks(entry["text"], args.max_chars) + total_chars = sum(len(c) for c in chunks) or 1 + t = start + for chunk in chunks: + share = dur * (len(chunk) / total_chars) + share = max(MIN_SECS, min(share, args.max_secs)) + cues.append((t, min(t + share, start + dur), wrap(chunk))) + t += share + clock = start + dur + + lines = [] + for i, (a, b, text) in enumerate(cues, 1): + if b <= a: + b = a + MIN_SECS + lines += [str(i), f"{ts(a)} --> {ts(b)}", text, ""] + args.out.write_text("\n".join(lines)) + print(f"{len(cues)} cues, ends at {ts(cues[-1][1])} -> {args.out}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/proof/scripts/narrate b/proof/scripts/narrate new file mode 100755 index 0000000..5f99e21 --- /dev/null +++ b/proof/scripts/narrate @@ -0,0 +1,284 @@ +#!/usr/bin/env -S uv run --quiet --script +# /// script +# requires-python = ">=3.10" +# dependencies = ["pyyaml", "piper-tts"] +# /// +"""Render one narration clip per scene, and prove it says what you wrote. + +Engine selection is automatic: a cloud voice when a key is available, a +local neural voice (Piper) when there isn't one. The local path needs no +key, no network after the first voice download, and runs on macOS and +Linux alike - so a container with no secrets in it can still narrate. + +Input is a scenes file: a YAML list of scenes, each with `id` and +`narration`. Output is OUTDIR/.wav plus OUTDIR/manifest.json carrying +the exact text and measured duration of each clip, which is what +make-subtitles and the assembly step both read. + +Usage: + narrate SCENES.yaml OUTDIR [--engine auto|openai|openai-chat|piper] + [--voice NAME] [--force] +""" + +import argparse +import base64 +import difflib +import json +import os +import re +import subprocess +import sys +import urllib.request +import wave +from pathlib import Path + +import yaml + +OPENAI_TTS_MODEL = "gpt-4o-mini-tts" # deterministic: reads what you send +OPENAI_CHAT_MODEL = "gpt-audio-1.5" # better prosody, will ad-lib; gated +PIPER_VOICE = "en_US-lessac-medium" + + +def die(msg): + print(f"narrate: {msg}", file=sys.stderr) + sys.exit(1) + + +def openai_key(): + key = os.environ.get("OPENAI_API_KEY") + if key: + return key.strip() + try: + out = subprocess.run(["llm", "keys", "get", "openai"], + capture_output=True, text=True, timeout=15) + if out.returncode == 0 and out.stdout.strip(): + return out.stdout.strip() + except Exception: # noqa: BLE001 - llm not installed is a normal outcome + pass + return None + + +def norm(s): + return re.sub(r"[^a-z0-9 ]+", "", s.lower()).split() + + +ASR_SNIPPET = """ +import sys +from faster_whisper import WhisperModel +m = WhisperModel(sys.argv[2], device="cpu", compute_type="int8") +segs, _ = m.transcribe(sys.argv[1]) +print(" ".join(s.text.strip() for s in segs)) +""" + + +def transcribe_local(wav, model="base.en"): + """Transcribe with a local ASR, in its own uv env so narrate stays light. + Returns None when faster-whisper isn't available.""" + try: + out = subprocess.run( + ["uv", "run", "--quiet", "--with", "faster-whisper", "python3", + "-c", ASR_SNIPPET, str(wav), model], + capture_output=True, text=True, timeout=900) + except Exception: # noqa: BLE001 - no uv, no network: gate simply unavailable + return None + return out.stdout.strip() if out.returncode == 0 and out.stdout.strip() else None + + +def structural_drift(text, heard): + """How far a transcript diverges from the script, ignoring the noise an + ASR always makes. + + Exact word-matching is the wrong tool here: a small model mangles + unusual names ("smevals" -> "Mevil"), and - worse - a *dropped* word + scores as more similar than two mispronounced ones. What is detectable, + and what actually matters, is missing or invented CONTENT: a sentence + the voice skipped, or a preamble it invented. Returns + (length_delta_fraction, longest_run_of_missing_or_changed_words). + """ + want, got = norm(text), norm(heard) + delta = abs(len(got) - len(want)) / max(1, len(want)) + ops = difflib.SequenceMatcher(a=want, b=got).get_opcodes() + worst = max((i2 - i1 for tag, i1, i2, _, _ in ops if tag in ("delete", "replace")), + default=0) + return delta, worst + + +def post(url, key, body, want_json=True): + req = urllib.request.Request( + url, data=json.dumps(body).encode(), + headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=180) as r: + return json.load(r) if want_json else r.read() + + +def say_openai(key, text, out_wav, voice): + data = post("https://api.openai.com/v1/audio/speech", key, + {"model": OPENAI_TTS_MODEL, "voice": voice or "nova", + "input": text, "response_format": "wav"}, want_json=False) + out_wav.write_bytes(data) + return None # deterministic engine: nothing to gate + + +def say_openai_chat(key, text, out_wav, voice): + doc = post("https://api.openai.com/v1/chat/completions", key, { + "model": OPENAI_CHAT_MODEL, + "modalities": ["text", "audio"], + "audio": {"voice": voice or "nova", "format": "wav"}, + "messages": [{"role": "user", "content": + "Read this narration aloud, warm and clear, verbatim, " + "and say nothing else:\n\n" + text}], + }) + audio = doc["choices"][0]["message"]["audio"] + out_wav.write_bytes(base64.b64decode(audio["data"])) + return audio.get("transcript", "") + + +def say_piper(text, out_wav, voice): + from piper import PiperVoice + from piper.download_voices import download_voice + home = Path(os.environ.get("PIPER_VOICE_DIR", + Path.home() / ".cache" / "piper-voices")) + home.mkdir(parents=True, exist_ok=True) + name = voice or PIPER_VOICE + onnx = home / f"{name}.onnx" + if not onnx.exists(): + print(f" downloading local voice {name} (one time)…") + download_voice(name, home) + v = PiperVoice.load(str(onnx)) + with wave.open(str(out_wav), "wb") as w: + v.synthesize_wav(text, w) + return None + + +def duration(path): + out = subprocess.run( + ["ffprobe", "-v", "error", "-show_entries", "format=duration", + "-of", "csv=p=0", str(path)], capture_output=True, text=True) + return round(float(out.stdout.strip()), 3) + + +def main(): + if len(sys.argv) == 4 and sys.argv[1] == "--drift-check": + script = Path(sys.argv[2]).read_text() + heard = Path(sys.argv[3]).read_text() + delta, worst = structural_drift(script, heard) + bad = delta > 0.15 or worst >= 4 + print(f"length change {delta:.0%}, worst run {worst} -> " + f"{'MISMATCH' if bad else 'ok'}") + return 1 if bad else 0 + + ap = argparse.ArgumentParser() + ap.add_argument("scenes", type=Path) + ap.add_argument("outdir", type=Path) + ap.add_argument("--engine", default="auto", + choices=["auto", "openai", "openai-chat", "piper"]) + ap.add_argument("--voice", default=None) + ap.add_argument("--force", action="store_true") + ap.add_argument("--verify", default="auto", choices=["auto", "on", "off"], + help="listen back to each clip with a local ASR and flag " + "missing or invented content (default: on when the " + "engine can't tell you what it said)") + ap.add_argument("--asr-model", default="base.en") + args = ap.parse_args() + + doc = yaml.safe_load(args.scenes.read_text()) + scenes = [s for s in doc.get("scenes", []) if (s.get("narration") or "").strip()] + if not scenes: + die("no scenes with narration") + + key = openai_key() + engine = args.engine + if engine == "auto": + engine = "openai" if key else "piper" + if engine.startswith("openai") and not key: + die("no OPENAI_API_KEY (and `llm keys get openai` found nothing). " + "Use --engine piper for a local voice.") + print(f"engine: {engine}" + ("" if key or engine == "piper" else "")) + + # a deterministic cloud endpoint reads exactly what you send it, so the + # ear-check is optional there; anything else gets listened to by default + verify = args.verify == "on" or (args.verify == "auto" and engine != "openai") + + args.outdir.mkdir(parents=True, exist_ok=True) + # what the cached clips were rendered FROM: editing a line and keeping + # its old audio is a silent lie, and the movie will contradict itself + prior = {} + prior_path = args.outdir / "manifest.json" + if prior_path.exists(): + try: + prior = {e["id"]: e.get("text", "") for e in + json.loads(prior_path.read_text())} + except Exception: # noqa: BLE001 - a corrupt manifest just means no cache + prior = {} + manifest, failures = [], [] + + for sc in scenes: + sid = sc["id"] + text = " ".join((sc["narration"] or "").split()) + wav = args.outdir / f"{sid}.wav" + if wav.exists() and not args.force and prior.get(sid) == text: + print(f"{sid}: cached") + elif wav.exists() and not args.force and sid in prior: + print(f"{sid}: text changed since this clip was rendered - redoing") + args.force = True + else: + for attempt in (1, 2): + if engine == "openai": + claimed = say_openai(key, text, wav, args.voice) + elif engine == "openai-chat": + claimed = say_openai_chat(key, text, wav, args.voice) + else: + claimed = say_piper(text, wav, args.voice) + + # a chat model reports what it said: hold it to that exactly, + # because "Sure, here it is:" is the failure it introduces + if claimed is not None: + want, got = norm(text), norm(claimed) + drift = abs(len(want) - len(got)) + sum( + 1 for a, b in zip(want, got) if a != b) + if drift > max(2, len(want) // 25): + print(f"{sid}: engine ad-libbed (attempt {attempt}, " + f"drift {drift})") + continue + + # every engine: listen back. An ASR mangles unusual names, so + # only missing or invented CONTENT counts as a failure here. + if verify: + heard = transcribe_local(wav, args.asr_model) + if heard is None: + print(f"{sid}: ok (no local ASR available - gate skipped)") + break + delta, worst = structural_drift(text, heard) + if delta > 0.15 or worst >= 4: + print(f"{sid}: what came out does not match the script " + f"(attempt {attempt}: {delta:.0%} length change, " + f"{worst} words in a row wrong)") + print(f" heard: {heard[:120]}") + continue + print(f"{sid}: ok (verified by ear: {delta:.0%} length " + f"change, worst run {worst})") + break + print(f"{sid}: ok") + break + else: + failures.append(sid) + manifest.append({"id": sid, "text": text, "wav": wav.name, + "duration": duration(wav)}) + + (args.outdir / "manifest.json").write_text(json.dumps(manifest, indent=2)) + total = sum(m["duration"] for m in manifest) + print(f"\n{len(manifest)} clips, {total:.1f}s total -> {args.outdir}/manifest.json") + if engine == "piper": + print("local voice: it mispronounces unusual names rather than dropping " + "them - listen to one clip before you commit to a voice.") + if verify: + print("the ear-check catches missing or invented sentences, not " + "pronunciation: an ASR mangles jargon too.") + if failures: + print(f"FAILED verbatim delivery: {failures}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/proof/show.py b/proof/show.py new file mode 100755 index 0000000..ccf9bfe --- /dev/null +++ b/proof/show.py @@ -0,0 +1,145 @@ +#!/usr/bin/env python3 +"""Filmed CLI for the Hermes function-calling toolkit. + +Each subcommand exercises real repository code — tool schema conversion, + tool-call parsing, pydantic/jsonschema validation, and live tool +execution. Nothing here is staged or replayed from a fixture. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +logging.getLogger("function-calling-inference").setLevel(logging.CRITICAL) +logging.getLogger("yfinance").setLevel(logging.CRITICAL) + + +def banner(title: str) -> None: + line = "=" * 72 + print(line) + print(f" Hermes Function Calling · {title}") + print(line) + print() + + +def cmd_tools(_args: argparse.Namespace) -> int: + from functions import get_openai_tools + + tools = get_openai_tools() + banner("OpenAI tool schemas from functions.py") + print(f"{len(tools)} tools registered:\n") + for i, tool in enumerate(tools, 1): + fn = tool["function"] + print(f" {i:2}. {fn['name']}") + fund = next(t for t in tools if t["function"]["name"] == "get_stock_fundamentals") + print("\nConverted schema for get_stock_fundamentals:\n") + print(json.dumps(fund["function"]["parameters"], indent=2)) + return 0 + + +def cmd_parse(_args: argparse.Namespace) -> int: + from utils import validate_and_extract_tool_calls + + banner("Parse a Hermes from model output") + raw = ( + "\n" + '{"name": "code_interpreter",' + ' "arguments": {"code_markdown": "```python\\nprint(6 * 7)\\n```"}}\n' + "" + ) + print("Raw assistant message:\n") + print(raw) + print() + ok, calls, err = validate_and_extract_tool_calls(raw) + print(f"extracted_ok = {ok}") + if err: + print(f"error = {err}") + print("tool_calls =") + print(json.dumps(calls, indent=2)) + return 0 if ok else 1 + + +def cmd_validate(_args: argparse.Namespace) -> int: + from functions import get_openai_tools + from validator import validate_function_call_schema + + banner("Validate calls against the live tool schemas") + tools = get_openai_tools() + cases = [ + ( + "valid call", + { + "name": "get_stock_fundamentals", + "arguments": {"symbol": "TSLA"}, + }, + ), + ( + "missing required argument", + {"name": "get_stock_fundamentals", "arguments": {}}, + ), + ( + "unknown function", + {"name": "launch_the_missiles", "arguments": {"symbol": "TSLA"}}, + ), + ] + for label, call in cases: + ok, err = validate_function_call_schema(call, tools) + verdict = "ACCEPT" if ok else "REJECT" + print(f"{verdict:6} {label}") + print(f" call = {json.dumps(call)}") + if err: + print(f" why = {err}") + print() + return 0 + + +def cmd_execute(_args: argparse.Namespace) -> int: + from functions import code_interpreter + + banner("Execute a real tool: code_interpreter") + code_markdown = ( + "```python\n" + "squares = [n * n for n in range(1, 8)]\n" + "total = sum(squares)\n" + "print('squares', squares)\n" + "print('total', total)\n" + "```" + ) + print("Invoking functions.code_interpreter with this Python:\n") + print(code_markdown) + print("Live result (exec namespace, not a fixture):\n") + result = code_interpreter.invoke({"code_markdown": code_markdown}) + print(json.dumps(result, indent=2)) + expected = {"squares": [1, 4, 9, 16, 25, 36, 49], "total": 140} + if result != expected: + print(f"\nUNEXPECTED: {result!r} != {expected!r}", file=sys.stderr) + return 1 + print("\nOK squares 1..7 sum to 140") + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest="cmd", required=True) + sub.add_parser("tools", help="list converted OpenAI tool schemas") + sub.add_parser("parse", help="extract JSON from a block") + sub.add_parser("validate", help="accept a good call, reject bad ones") + sub.add_parser("execute", help="run code_interpreter for real") + args = parser.parse_args() + return { + "tools": cmd_tools, + "parse": cmd_parse, + "validate": cmd_validate, + "execute": cmd_execute, + }[args.cmd](args) + + +if __name__ == "__main__": + raise SystemExit(main())