#!/usr/bin/env python3 """Schedule what changes on screen, so nothing stays still long enough to lose the viewer. This is the difference between a recording and an edit. Published measurements of short-form retention agree on one number more than any other: a talking head that never changes frame holds roughly 41% of its viewers, and the same content with an on-screen change every 3–5 seconds holds roughly 58%. Nothing else in the pipeline — not the grade, not the music, not the font — moves retention by seventeen points. So the schedule is built against a *budget of stillness*, not against taste: * no window longer than `--max-static` may pass without something changing; * a change is a cut, a reframe, a b-roll window, a graphic, or a caption treatment — anything the eye registers as new; * changes are placed on word boundaries, never mid-word, and never on top of each other; * the first three seconds get the densest cadence, because that is where the decision to stay is made. *What* changes at each point is chosen from the transcript, not at random. A number wants a callout. A concrete, picturable noun wants b-roll. A pronoun or an abstraction wants a reframe, because there is nothing to show. Choosing by content is what keeps the result from feeling like a zoom applied on a timer. The output is a plan. `render_punch_ins.py` executes the reframes; b-roll windows are handed to the sourcing scripts; graphics go to the HyperFrames layer. """ from __future__ import annotations import argparse import json import re import subprocess import sys from pathlib import Path import numpy as np sys.path.insert(0, str(Path(__file__).resolve().parent)) from media_probe import probe # noqa: E402 from skill_config import canvas as resolve_canvas, emit, require_binary, utf8_stdout # noqa: E402 NUMBER = re.compile(r"\d") # Words that name something a camera could point at. Kept deliberately small and # domain-agnostic: a long "visual noun" list becomes a topic model that is wrong # on every new subject. Anything not matched falls through to a reframe, which # is always safe. CONCRETE_HINT = re.compile( r"(карт|видеокарт|компьютер|экран|код|сет[ьи]|мозг|данн|график|таблиц|фото|" r"изображ|камер|телефон|лаборатор|сервер|чип|процессор|программ|приложен|" r"робот|машин|устройств|город|человек|лиц|рук|глаз)", re.IGNORECASE) ABSTRACT = re.compile(r"^(это|то|тот|та|те|он|она|они|мы|вы|я|его|её|их|как|что|" r"который|которые|потому|поэтому|и|а|но|же|ли|бы|уже|ещё|еще)$", re.IGNORECASE) def subject_box(path: Path, samples: int = 90) -> dict: """Where the speaker is, from temporal variance rather than face detection. A talking head is the only thing in the frame that changes: the wall, the chair and the shelf are constant across the whole clip. Averaging the frame-to-frame difference over the file therefore produces a map whose hot region *is* the subject, with no model, no weights and no failure mode on an unusual face angle. It also degrades gracefully — on footage where nothing moves it returns the centre, which is what a reframe should do anyway. """ done = subprocess.run( ["ffmpeg", "-v", "error", "-nostdin", "-i", str(path), "-vf", f"fps=2,scale=64:114,format=gray,tblend=all_mode=difference", "-frames:v", str(samples), "-f", "rawvideo", "-"], capture_output=True) if done.returncode or not done.stdout: return {"cx": 0.5, "cy": 0.42, "confidence": 0.0, "note": "кадры не прочитаны — центр по умолчанию"} data = np.frombuffer(done.stdout, dtype=np.uint8).astype(np.float64) frames = data.size // (64 * 114) if frames < 4: return {"cx": 0.5, "cy": 0.42, "confidence": 0.0, "note": "слишком мало кадров"} stack = data[:frames * 64 * 114].reshape(frames, 114, 64) energy = stack.mean(axis=0) energy = np.maximum(0.0, energy - np.percentile(energy, 60)) total = energy.sum() if total < 1e-6: return {"cx": 0.5, "cy": 0.42, "confidence": 0.0, "note": "движения в кадре нет — кадрируем по центру"} ys, xs = np.mgrid[0:114, 0:64] cx = float((energy * xs).sum() / total) / 63.0 cy = float((energy * ys).sum() / total) / 113.0 # How concentrated the motion is: a single speaker gives a tight blob, a busy # street gives a flat map and a centroid that means nothing. share = float(energy[energy > np.percentile(energy, 90)].sum() / total) return {"cx": round(cx, 3), "cy": round(cy, 3), "confidence": round(share, 3), "note": ("центр движения в кадре; уверенность — доля энергии в верхних 10% " "ячеек, ниже 0.25 означает, что в кадре двигается не один объект")} def load_timeline(path: Path) -> tuple[list[dict], list[float], float]: data = json.loads(path.read_text(encoding="utf-8")) words = data.get("words") or [] if not words: raise SystemExit(f"No words in {path}") cuts = [] for clip in data.get("clips") or []: start = float(clip.get("global_start", 0.0)) if start > 0.01: cuts.append(round(start, 3)) cold = data.get("cold_open") or {} if cold.get("length"): cuts.append(round(float(cold["length"]), 3)) duration = float(data.get("expected_duration") or (words[-1]["end"] if words else 0.0)) return words, sorted(set(cuts)), duration def classify(word: str) -> str: token = word.strip(" ,.!?;:«»\"()") if NUMBER.search(token): return "callout" if ABSTRACT.match(token): return "reframe" if CONCRETE_HINT.search(token) and len(token) > 4: return "broll" return "reframe" def main() -> None: utf8_stdout() parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("video") parser.add_argument("timeline") parser.add_argument("--out", default="retention_plan.json") parser.add_argument("--max-static", type=float, default=4.0, help="Longest run of screen time with no change (measured " "optimum is 3-5 s; 5 s is the documented ceiling)") parser.add_argument("--opening-static", type=float, default=2.0, help="Tighter ceiling for the first --opening-seconds") parser.add_argument("--opening-seconds", type=float, default=6.0) parser.add_argument("--min-gap", type=float, default=1.6, help="Two changes closer than this read as a stutter") parser.add_argument("--zoom", type=float, default=1.14, help="Punch-in scale. Above ~1.25 a 1080p source starts to soften") parser.add_argument("--zoom-variation", type=float, default=0.05, help="How much successive punch-ins differ, so they do not " "all look like the same move") parser.add_argument("--allow-broll", action="store_true", help="Emit b-roll windows as well as reframes") parser.add_argument("--canvas", default="reels") args = parser.parse_args() require_binary("ffmpeg") video = Path(args.video).resolve() words, cuts, duration = load_timeline(Path(args.timeline).resolve()) board = resolve_canvas(args.canvas) info = probe(video) actual = float((info.get("video") or {}).get("duration") or info.get("duration") or duration) subject = subject_box(video) # Existing change events: every cut the recut already made, plus the start. existing = sorted(set([0.0] + [c for c in cuts if 0 < c < actual])) # Walk the file and fill any gap that exceeds the budget. events: list[dict] = [] scheduled = list(existing) index = 0 cursor = 0.0 while cursor < actual - 0.5: ceiling = (args.opening_static if cursor < args.opening_seconds else args.max_static) following = next((t for t in scheduled if t > cursor + 0.01), actual) if following - cursor <= ceiling: cursor = following continue target = cursor + ceiling # Land on a word boundary: a reframe inside a word reads as a glitch. candidates = [w for w in words if abs(float(w["start"]) - target) < ceiling * 0.6] if not candidates: cursor = target continue word = min(candidates, key=lambda w: abs(float(w["start"]) - target)) at = round(float(word["start"]), 3) if any(abs(at - t) < args.min_gap for t in scheduled): cursor = target continue kind = classify(word["word"]) if kind == "broll" and not args.allow_broll: kind = "reframe" # Alternate the move so consecutive punch-ins are not identical. step = args.zoom + args.zoom_variation * (1 if index % 2 == 0 else -1) drift = 0.06 * (1 if index % 4 in (0, 3) else -1) end_word = next((w for w in words if float(w["start"]) > at + 2.2), None) until = round(float(end_word["start"]) if end_word else min(actual, at + 2.6), 3) events.append({ "at": at, "until": until, "kind": kind, "trigger_word": word["word"], "why": { "callout": "в кадре произносится число — оно должно быть видно, а не только слышно", "broll": "названо то, что можно показать; перебивка сильнее любого зума", "reframe": "показывать нечего, поэтому меняется крупность — это самый дешёвый " "и самый частый интеррапт", }[kind], "zoom": round(step, 3) if kind == "reframe" else None, "center": ({"cx": round(min(0.82, max(0.18, subject["cx"] + drift)), 3), "cy": round(min(0.72, max(0.24, subject["cy"])), 3)} if kind == "reframe" else None), "sfx": ({"id": "click_tick", "gain_db": -26} if kind == "reframe" else {"id": "pop_ui", "gain_db": -22}), }) scheduled.append(at) scheduled.sort() index += 1 cursor = at # ------------------------------------------------------------------ report all_changes = sorted(set(scheduled)) gaps = [round(b - a, 2) for a, b in zip(all_changes, all_changes[1:])] if all_changes and actual - all_changes[-1] > 0.1: gaps.append(round(actual - all_changes[-1], 2)) still = sum(gap for gap in gaps if gap > args.max_static) before_changes = sorted(set(existing)) before_gaps = [b - a for a, b in zip(before_changes, before_changes[1:])] or [actual] if before_changes and actual - before_changes[-1] > 0.1: before_gaps.append(actual - before_changes[-1]) plan = { "version": "1.0", "video": str(video), "duration": round(actual, 3), "canvas": args.canvas, "subject": subject, "budget": { "max_static_seconds": args.max_static, "opening_static_seconds": args.opening_static, "min_gap_seconds": args.min_gap, "basis": ("Опубликованные измерения удержания коротких видео: статичная " "говорящая голова держит ~41% зрителей, та же длина со сменой " "картинки каждые 3–5 секунд — ~58%. Потолок статики — 5 секунд."), }, "before": { "changes": len(before_changes), "mean_gap": round(float(np.mean(before_gaps)), 2), "max_gap": round(float(np.max(before_gaps)), 2), "seconds_over_budget": round(sum(g for g in before_gaps if g > args.max_static), 1), "share_over_budget": round(sum(g for g in before_gaps if g > args.max_static) / actual, 3), }, "after": { "changes": len(all_changes), "mean_gap": round(float(np.mean(gaps)), 2) if gaps else 0.0, "max_gap": round(float(np.max(gaps)), 2) if gaps else 0.0, "seconds_over_budget": round(still, 1), "share_over_budget": round(still / actual, 3), }, "events": events, "counts": { kind: sum(1 for event in events if event["kind"] == kind) for kind in ("reframe", "broll", "callout") }, "next": [ "python scripts/render_punch_ins.py retention_plan.json IN.mp4 OUT.mp4", "Окна kind=broll отдай в stock_media.py или youtube_source.py — " "перебивка на реальном материале держит лучше любого зума.", "Окна kind=callout — это слой графики HyperFrames, а не субтитр.", ], "checks": [ "Ни один reframe не должен попасть внутрь произносимого слова — план ставит " "их на границы, но после ручной правки это надо проверить заново.", "Если уверенность subject ниже 0.25, кадрирование считай неизмеренным и " "задай центр вручную.", ], } Path(args.out).write_text(json.dumps(plan, ensure_ascii=False, indent=2), encoding="utf-8") emit({**{k: v for k, v in plan.items() if k != "events"}, "events_count": len(events), "events_preview": events[:6], "out": args.out}) if __name__ == "__main__": main()