#!/usr/bin/env python3 """Composite the HyperFrames effect pack onto a finished cut. The pack (`assets/hyperframes/street_overlays_pack.mov`) is one ProRes 4444 file with a real alpha channel, holding every effect back to back in fixed 1.5-second slots. That is deliberate: rendering eight separate compositions means eight Chrome launches and eight encodes, where one strip is a single render and the compositor cuts the slot it needs by time. Placement is measured, not guessed. `find_overlay_zone.py` reports which rows of the frame the speaker occupies; an effect is only allowed into a band the speaker does not. That is the whole difference between graphics that look designed and graphics that look pasted: the former never fight the face. Effects are 0.5–1.2 s on screen. Longer and the viewer starts reading the graphic instead of the speaker, which is the opposite of why it is there. """ from __future__ import annotations import argparse import json import subprocess import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from media_probe import probe # noqa: E402 from skill_config import ASSETS, canvas as resolve_canvas, emit, require_binary, utf8_stdout # noqa: E402 # По умолчанию — пак данных, а не рисованный декор. # # Рисованный пак (street_overlays_pack.mov) остаётся в ассетах, но дефолтом быть # перестал: стрелки и звёздочки, нарисованные скриптом, выглядят самодельно и # уже один раз были сняты с готового монтажа. Правило записано в SKILL.md — # «декор берётся готовым ассетом, данные делает композиция», — но пока дефолт # указывал на декор, правило существовало только на бумаге, и звёздочка вернулась # в ролик на следующем же прогоне. PACK = ASSETS / "hyperframes" / "dataviz_pack.mov" LEGACY_PACK = ASSETS / "hyperframes" / "street_overlays_pack.mov" # name -> where it lives in the pack, and where it was drawn on the 1080x1920 # canvas. `cy` is what makes automatic placement possible: an effect drawn at a # height the speaker occupies has to move, and the only honest way to know that # is to compare its authored centre against a measured subject band. # `anchored` marks the effects whose position *is* the meaning — corner brackets # in the middle of the frame are not brackets, and smoke has to come off the # bottom edge — so those are never moved. SLOTS: dict[str, dict] = { "speed_lines": {"at": 0.05, "length": 1.25, "zone": "edges", "cy": 960, "h": 1500, "anchored": True, "directional": False, "use": "Разгон, перечисление, нарастание. Линии идут по краям и " "не заходят в среднюю треть."}, "burst": {"at": 1.55, "length": 1.25, "zone": "float", "cy": 752, "h": 264, "anchored": False, "directional": False, "use": "Короткий акцент на слове. Самый частый эффект пака."}, "arrow": {"at": 3.05, "length": 1.25, "zone": "float", "cy": 1472, "h": 106, "anchored": False, "directional": True, "use": "Указание на объект или на титр снизу слева."}, "underline": {"at": 4.55, "length": 1.25, "zone": "caption", "cy": 1614, "h": 40, "anchored": True, "directional": True, "use": "Подчёркивание под плашкой субтитра. Только на выводе."}, "brackets": {"at": 6.05, "length": 1.25, "zone": "frame", "cy": 960, "h": 1660, "anchored": True, "directional": False, "use": "Рамка-уголки: «смотри сюда», начало блока."}, "impact": {"at": 7.55, "length": 1.25, "zone": "float", "cy": 420, "h": 420, "anchored": False, "directional": False, "use": "Удар на цифре или выводе. Максимум два раза за ролик."}, "circle": {"at": 9.05, "length": 1.25, "zone": "float", "cy": 1497, "h": 255, "anchored": False, "directional": True, "use": "Обводка от руки вокруг того, что названо."}, "smoke": {"at": 10.55, "length": 1.4, "zone": "bottom", "cy": 1700, "h": 300, "anchored": True, "directional": False, "use": "Дым из нижних углов. Переход между блоками."}, } def subject_band_at(video: Path, at: float, width: int, height: int) -> tuple[float, float] | None: """Where the speaker is *at this moment*, not on average over the file. `find_overlay_zone.py` averages the whole clip, and that is the right answer for a fixed frame. It is the wrong answer once `render_punch_ins.py` has done its work: the crop changes several times a minute, so a panel placed against the file-wide average lands on the chest in the wide shots and on the chin in the tight ones. Measuring a one-second window around the event costs one ffmpeg call and removes the whole class of collision. Rough skin-tone mask, same method as the zone script — the task is not to find a face but to find where the person is, and for that a colour test is enough and has no failure mode on an unusual angle. """ import numpy as np done = subprocess.run( ["ffmpeg", "-v", "error", "-nostdin", "-ss", f"{max(0.0, at - 0.4):.3f}", "-i", str(video), "-t", "0.9", "-vf", "fps=4,scale=72:128,format=rgb24", "-f", "rawvideo", "-"], capture_output=True) if done.returncode or not done.stdout: return None data = np.frombuffer(done.stdout, dtype=np.uint8) frames = data.size // (72 * 128 * 3) if frames < 1: return None stack = data[:frames * 72 * 128 * 3].reshape(frames, 128, 72, 3).astype(np.int16) r, g, b = stack[..., 0], stack[..., 1], stack[..., 2] skin = ((r > 70) & (g > 35) & (b > 20) & (r > g + 12) & (r > b + 12)).mean(axis=0) rows = skin.mean(axis=1) hot = np.where(rows > max(0.06, rows.max() * 0.35))[0] if hot.size < 3: return None top = float(hot[0]) / 127.0 * height bottom = float(hot[-1]) / 127.0 * height return top, bottom def place(authored_cy: float, target_y: float, scale: float, height: int) -> int: """The `dy` that lands an effect's authored centre on `target_y`. Scale has to be in this arithmetic. `overlay` centres the scaled pack frame, so shrinking it by s also drags every drawn element toward the frame centre: the effect ends up at `(H - H*s)/2 + cy*s`, not at `cy`. Computing `dy` as `target - cy` is correct only at s = 1, and silently off by tens of pixels otherwise — which on a 334px band is the difference between the free strip and the speaker's forehead. """ scaled_centre = (height - height * scale) / 2.0 + authored_cy * scale return int(round(target_y - scaled_centre)) def best_free_band(zone: dict, height: int) -> dict | None: """The tallest band the speaker does not occupy, preferring the safe one. `find_overlay_zone.py` marks bands that collide with the platform's own UI — the bottom strip where TikTok and Reels put the caption, the handle and the buttons. A graphic there is not covered by the speaker but is covered by the app, which is the same problem with an extra step. """ bands = [band for band in (zone.get("free_bands") or []) if band.get("height", 0) > height * 0.09] if not bands: return None preferred = [band for band in bands if band.get("recommended")] return max(preferred or bands, key=lambda band: band.get("height", 0)) def ff(value: float) -> str: return f"{value:.6f}".rstrip("0").rstrip(".") or "0" def main() -> None: utf8_stdout() parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("input") parser.add_argument("output") parser.add_argument("--plan", help="JSON {overlays:[{at, effect, scale?, dx?, dy?, opacity?}]}") parser.add_argument("--retention-plan", help="retention_plan.json — derive overlays from its events") parser.add_argument("--pack", default=str(PACK)) parser.add_argument("--slots", help="JSON описание слотов пака. По умолчанию ищется " ".slots.json рядом с паком, иначе встроенная раскладка.") parser.add_argument("--zone", help="find_overlay_zone.py report — запасной вариант, " "если покадровый замер не удался") parser.add_argument("--face-fraction", type=float, default=0.50, help="Какая доля полосы спикера считается лицом и защищается") parser.add_argument("--caption-top", type=float, default=0.74, help="Доля высоты кадра, ниже которой живут субтитры. " "Графика туда не заходит.") parser.add_argument("--canvas", default="reels") parser.add_argument("--max-overlays", type=int, default=14) parser.add_argument("--crf", type=int, default=17) parser.add_argument("--preset", default="slow") parser.add_argument("--dry-run", action="store_true") parser.add_argument("--list", action="store_true") args = parser.parse_args() require_binary("ffmpeg") pack = Path(args.pack).resolve() # A pack is an asset. Adding a card to it must not be a code change, so the # slot table is loaded from a JSON file sitting next to the .mov and only # falls back to the built-in layout when there is none. slots = dict(SLOTS) slots_path = Path(args.slots) if args.slots else pack.with_suffix(".slots.json") slots_source = "встроенная раскладка" if slots_path.is_file(): loaded = json.loads(slots_path.read_text(encoding="utf-8")).get("slots") or {} if loaded: slots = loaded slots_source = str(slots_path) if args.list: emit({"pack": str(pack), "slots_from": slots_source, "slots": slots}) return source = Path(args.input).resolve() for path in (source, pack): if not path.is_file(): raise SystemExit(f"Not found: {path}") board = resolve_canvas(args.canvas) info = probe(source) video = info.get("video") or {} width = int(video.get("display_width") or board.width) height = int(video.get("display_height") or board.height) fps = int(round(float(video.get("avg_frame_rate") or board.fps))) duration = float(video.get("duration") or info.get("duration") or 0.0) overlays: list[dict] = [] if args.plan: payload = json.loads(Path(args.plan).read_text(encoding="utf-8")) overlays = payload.get("overlays") or payload elif args.retention_plan: plan = json.loads(Path(args.retention_plan).read_text(encoding="utf-8")) # Map the retention planner's own decisions onto graphics. `callout` # already means "a number was said and must be seen"; `reframe` events # are where the edit changes crop and can carry an accent. cycle = ["burst", "speed_lines", "brackets", "circle", "arrow", "impact"] index = 0 for event in plan.get("events", []): if event.get("kind") == "callout": overlays.append({"at": event["at"], "effect": "impact", "why": f"число «{event.get('trigger_word')}» должно быть видно"}) elif event.get("kind") == "reframe" and index % 2 == 0: overlays.append({"at": event["at"], "effect": cycle[index % len(cycle)], "why": f"акцент на рефрейме «{event.get('trigger_word')}»"}) index += 1 else: raise SystemExit("Нужен --plan или --retention-plan") # Keep them apart: two graphics inside a second read as one messy graphic. cleaned: list[dict] = [] for item in sorted(overlays, key=lambda o: float(o["at"])): at = float(item["at"]) name = item.get("effect") if name not in slots: raise SystemExit(f"Неизвестный эффект {name!r}. Есть: {sorted(slots)}") if at < 0.3 or at > duration - 0.6: continue if cleaned and at - cleaned[-1]["at"] < 1.6: continue cleaned.append({**item, "at": at, "effect": name}) cleaned = cleaned[:args.max_overlays] if not cleaned: emit({"status": "nothing_to_do", "note": "После фильтрации не осталось ни одной вставки."}) return # Vertical safety, measured **per event** rather than once for the file. # # A single file-wide band is the right answer for a locked-off frame and the # wrong one after `render_punch_ins.py`: the crop changes several times a # minute, so a panel placed against the average sits on the chest in the wide # shots and on the chin in the tight ones. That is exactly how the comparison # card ended up over the speaker's mouth. One extra ffmpeg call per event # removes the whole class of collision. fallback_band: tuple[float, float] | None = None if args.zone and Path(args.zone).is_file(): zone = json.loads(Path(args.zone).read_text(encoding="utf-8")) rows = zone.get("subject_rows") or zone.get("speaker_rows") or {} subject_top, subject_bottom = rows.get("top"), rows.get("bottom") if isinstance(subject_top, (int, float)) and isinstance(subject_bottom, (int, float)): fallback_band = (float(subject_top), float(subject_bottom)) # Captions own the bottom of the frame. A panel that clears the face but # lands on its own subtitles has solved nothing. caption_top = height * args.caption_top substitutions: list[str] = [] notes: list[str] = [] placed: list[dict] = [] for item in cleaned: slot = slots[item["effect"]] measured = subject_band_at(source, item["at"], width, height) or fallback_band if not measured: item["_dy"], item["_scale"] = int(item.get("dy", 0)), 1.0 placed.append(item) continue subject_top, subject_bottom = measured # Protect the *face*, not the silhouette: the band spans head to waist, # and treating all of it as off-limits leaves a ~350px strip that every # graphic then piles into. # # The fraction is 0.50, not the 0.42 this started at. 0.42 is roughly # right on a mid shot, but the punch-ins push the chin further down the # band, and the first render put a data panel 50px under a chin that was # actually 90px lower than the model said. A face guard that is only # correct at the average framing is not a guard. face_top = subject_top face_bottom = subject_top + args.face_fraction * (subject_bottom - subject_top) # Clearance so the graphic does not merely *touch* the chin. margin = height * 0.02 half = slot["h"] / 2.0 top_edge = slot["cy"] - half - margin bottom_edge = slot["cy"] + half + margin collides = not (bottom_edge <= face_top or top_edge >= face_bottom) if not collides and bottom_edge <= caption_top: item["_dy"], item["_scale"] = int(item.get("dy", 0)), 1.0 placed.append(item) continue if slot["directional"]: # A pointer moved off what it pointed at is not a pointer. substitutions.append( f"{item['at']:.2f}s: {item['effect']} снят — указывающий, перенос лишает смысла") continue # Two candidate homes: above the face, and between the face and the # captions. Pick whichever has more room, then scale to fit if needed. regions = [(0.0, max(0.0, face_top - 12)), (face_bottom + 12, caption_top)] region = max(regions, key=lambda r: r[1] - r[0]) room = region[1] - region[0] if room < slot["h"] * 0.45: substitutions.append( f"{item['at']:.2f}s: {item['effect']} снят — свободно всего {int(room)}px " f"при высоте {slot['h']}px") continue scale = min(1.0, round(room * 0.9 / slot["h"], 3)) target = (region[0] + region[1]) / 2.0 item["_dy"] = int(item.get("dy", 0)) + place(slot["cy"], target, scale, height) item["_scale"] = scale notes.append(f"{item['at']:.2f}s {item['effect']}: лицо {int(face_top)}..{int(face_bottom)}, " f"свободно {int(region[0])}..{int(region[1])} → {int(target)}" + (f" ×{scale}" if scale < 1.0 else "")) placed.append(item) cleaned = placed if not cleaned: emit({"status": "nothing_to_do", "substitutions": substitutions, "note": "Все вставки сняты по измеренной зоне."}) return zone_note = ("замер по каждому событию. " + " | ".join(notes)) if notes \ else "замер по каждому событию: ни одна вставка не пересекалась с лицом" command = ["ffmpeg", "-y", "-v", "warning", "-nostdin", "-i", str(source)] for item in cleaned: slot = slots[item["effect"]] command += ["-ss", ff(slot["at"]), "-t", ff(slot["length"]), "-i", str(pack)] graph: list[str] = [] label = "0:v" for index, item in enumerate(cleaned, start=1): slot = slots[item["effect"]] scale = float(item.get("scale", 1.0)) * float(item.get("_scale", 1.0)) dx = int(item.get("dx", 0)) dy = int(item.get("_dy", 0)) opacity = float(item.get("opacity", 1.0)) target_w = max(2, int(width * scale)) & ~1 chain = [f"scale={target_w}:-2:flags=bicubic", "format=rgba"] if opacity < 0.999: # colorchannelmixer on the alpha row is the only way to scale an # existing alpha channel; `format=rgba,geq` would be far slower. chain.append(f"colorchannelmixer=aa={opacity:.3f}") # The pack segment starts at its own zero; shift it to the event time so # `overlay` lines it up without needing `enable` arithmetic per frame. chain.append(f"setpts=PTS-STARTPTS+{ff(item['at'])}/TB") graph.append(f"[{index}:v]" + ",".join(chain) + f"[g{index}]") x = f"(W-w)/2+{dx}" y = f"(H-h)/2+{dy}" out = f"v{index}" graph.append( f"[{label}][g{index}]overlay=x={x}:y={y}:eof_action=pass:" f"enable='between(t,{ff(item['at'])},{ff(item['at'] + slot['length'])})'[{out}]" ) label = out graph.append(f"[{label}]format=yuv420p[vout]") report = { "status": "planned", "input": str(source), "pack": str(pack), "overlays": [{"at": round(o["at"], 3), "effect": o["effect"], "zone": slots[o["effect"]]["zone"], "use": slots[o["effect"]]["use"], "why": o.get("why")} for o in cleaned], "count": len(cleaned), "safe_zone": zone_note, "substitutions": substitutions, "note": ("Пак — один ProRes 4444 с настоящей альфой; слоты режутся по времени. " "Восемь отдельных композиций стоили бы восьми запусков Chrome."), } if args.dry_run: emit(report) return output = Path(args.output).resolve() output.parent.mkdir(parents=True, exist_ok=True) command += [ "-filter_complex", ";".join(graph), "-map", "[vout]", "-map", "0:a?", "-c:v", "libx264", "-preset", args.preset, "-crf", str(args.crf), "-profile:v", "high", "-pix_fmt", "yuv420p", "-r", str(fps), "-c:a", "copy", "-movflags", "+faststart", str(output), ] done = subprocess.run(command, capture_output=True, text=True, errors="replace") if done.returncode: report.update({"status": "failed", "graph": ";".join(graph), "tail": (done.stderr or "").strip()[-1800:]}) emit(report) raise SystemExit(done.returncode) actual = probe(output) report["status"] = "ok" report["output"] = str(output) report["actual_duration"] = actual.get("duration") drift = (round(actual["duration"] - duration, 4) if actual.get("duration") else None) report["duration_drift"] = drift report["drift_ok"] = drift is not None and abs(drift) <= max(0.08, 2.0 / fps) emit(report) if not report["drift_ok"]: raise SystemExit(2) if __name__ == "__main__": main()