From 46f417bf905cc4c2b3dbf11fbc3537586f99d8a5 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 14 Apr 2026 03:06:24 +0000 Subject: [PATCH 1/3] fix: pipeline correctness for quality, stale-vhs, timeout, and binary paths Co-authored-by: John Menke --- README.md | 19 +++++++- docs/demos/docgen.yaml | 8 ++++ src/docgen/binaries.py | 57 +++++++++++++++++++++++ src/docgen/cli.py | 10 +++- src/docgen/compose.py | 95 +++++++++++++++++++++++++++++++++----- src/docgen/config.py | 32 +++++++++++++ src/docgen/init.py | 8 ++++ src/docgen/manim_runner.py | 81 ++++++++++++++++++++++++++++---- src/docgen/vhs.py | 38 ++++++++++++++- tests/test_compose.py | 82 ++++++++++++++++++++++++++++++++ tests/test_config.py | 19 ++++++++ tests/test_manim_runner.py | 65 ++++++++++++++++++++++++++ 12 files changed, 489 insertions(+), 25 deletions(-) create mode 100644 src/docgen/binaries.py create mode 100644 tests/test_compose.py create mode 100644 tests/test_manim_runner.py diff --git a/README.md b/README.md index 861878b..593dbd6 100644 --- a/README.md +++ b/README.md @@ -55,7 +55,7 @@ docgen validate --pre-push # validate all outputs before committing | `docgen tts [--segment 01] [--dry-run]` | Generate TTS audio | | `docgen manim [--scene StackDAGScene]` | Render Manim animations | | `docgen vhs [--tape 02-quickstart.tape] [--strict]` | Render VHS terminal recordings | -| `docgen compose [01 02 03]` | Compose segments (audio + video) | +| `docgen compose [01 02 03] [--ffmpeg-timeout 900]` | Compose segments (audio + video) | | `docgen validate [--max-drift 2.75] [--pre-push]` | Run all validation checks | | `docgen concat [--config full-demo]` | Concatenate full demo files | | `docgen pages [--force]` | Generate index.html, pages.yml, .gitattributes, .gitignore | @@ -66,6 +66,23 @@ docgen validate --pre-push # validate all outputs before committing Create a `docgen.yaml` in your demos directory. See [examples/minimal-bundle/docgen.yaml](examples/minimal-bundle/docgen.yaml) for a starting point. +Useful pipeline options: + +```yaml +manim: + quality: 1080p30 # supports 480p15, 720p30, 1080p30, 1080p60, 1440p30, 1440p60, 2160p60 + manim_path: "" # optional explicit binary path (relative to docgen.yaml or absolute) + +vhs: + vhs_path: "" # optional explicit binary path (relative to docgen.yaml or absolute) + +compose: + ffmpeg_timeout_sec: 300 # can also be overridden with: docgen compose --ffmpeg-timeout N + warn_stale_vhs: true # warns if terminal/*.tape is newer than terminal/rendered/*.mp4 +``` + +If you edit a `.tape` file, run `docgen vhs` before `docgen compose` so compose does not use stale rendered terminal video. + ## System dependencies - **ffmpeg** — composition and probing diff --git a/docs/demos/docgen.yaml b/docs/demos/docgen.yaml index ee2b3e4..333999c 100644 --- a/docs/demos/docgen.yaml +++ b/docs/demos/docgen.yaml @@ -65,10 +65,18 @@ visual_map: manim: quality: 720p30 + manim_path: "" scenes: - DocgenOverviewScene - WizardGUIScene +vhs: + vhs_path: "" + +compose: + ffmpeg_timeout_sec: 300 + warn_stale_vhs: true + tts: model: gpt-4o-mini-tts voice: coral diff --git a/src/docgen/binaries.py b/src/docgen/binaries.py new file mode 100644 index 0000000..9a02afd --- /dev/null +++ b/src/docgen/binaries.py @@ -0,0 +1,57 @@ +"""Helpers for resolving external binary paths.""" + +from __future__ import annotations + +import os +import shutil +import sys +from dataclasses import dataclass, field +from pathlib import Path + + +@dataclass +class BinaryResolution: + name: str + path: str | None + tried: list[str] = field(default_factory=list) + + +def resolve_binary( + name: str, + *, + configured_path: str | None = None, + extra_candidates: list[str | Path] | None = None, +) -> BinaryResolution: + """Resolve an executable path using config overrides + sensible defaults.""" + candidates: list[str] = [] + + def _add(candidate: str | Path | None) -> None: + if not candidate: + return + p = str(candidate).strip() + if not p: + return + p = os.path.expanduser(os.path.expandvars(p)) + if p not in candidates: + candidates.append(p) + + _add(configured_path) + + # If docgen is running from a virtualenv, prefer binaries next to this python. + _add(Path(sys.executable).resolve().parent / name) + + for candidate in extra_candidates or []: + _add(candidate) + + which_hit = shutil.which(name) + if which_hit: + _add(which_hit) + + tried: list[str] = [] + for candidate in candidates: + tried.append(candidate) + cpath = Path(candidate) + if cpath.exists() and os.access(cpath, os.X_OK): + return BinaryResolution(name=name, path=str(cpath), tried=tried) + + return BinaryResolution(name=name, path=None, tried=tried) diff --git a/src/docgen/cli.py b/src/docgen/cli.py index 8d2ce48..72bb8df 100644 --- a/src/docgen/cli.py +++ b/src/docgen/cli.py @@ -124,8 +124,14 @@ def vhs(ctx: click.Context, tape: str | None, strict: bool) -> None: @main.command() @click.argument("segments", nargs=-1) +@click.option( + "--ffmpeg-timeout", + default=None, + type=int, + help="Override ffmpeg timeout in seconds (default from docgen.yaml compose.ffmpeg_timeout_sec).", +) @click.pass_context -def compose(ctx: click.Context, segments: tuple[str, ...]) -> None: +def compose(ctx: click.Context, segments: tuple[str, ...], ffmpeg_timeout: int | None) -> None: """Compose segments (audio + video via ffmpeg). Pass segment IDs to compose specific ones, or omit for the default set. @@ -133,7 +139,7 @@ def compose(ctx: click.Context, segments: tuple[str, ...]) -> None: from docgen.compose import Composer cfg = ctx.obj["config"] - comp = Composer(cfg) + comp = Composer(cfg, ffmpeg_timeout_sec=ffmpeg_timeout) target = list(segments) if segments else cfg.segments_default click.echo(f"=== Composing {len(target)} segments ===") comp.compose_segments(target) diff --git a/src/docgen/compose.py b/src/docgen/compose.py index d0c49d5..80e41fb 100644 --- a/src/docgen/compose.py +++ b/src/docgen/compose.py @@ -15,8 +15,13 @@ class ComposeError(RuntimeError): class Composer: - def __init__(self, config: Config) -> None: + def __init__(self, config: Config, ffmpeg_timeout_sec: int | None = None) -> None: self.config = config + self.ffmpeg_timeout_sec = ( + int(ffmpeg_timeout_sec) + if ffmpeg_timeout_sec is not None + else int(self.config.ffmpeg_timeout_sec) + ) def compose_segments(self, segment_ids: list[str], *, strict: bool = True) -> int: composed = 0 @@ -30,7 +35,9 @@ def compose_segments(self, segment_ids: list[str], *, strict: bool = True) -> in if vtype == "manim": ok = self._compose_simple(seg_id, self._manim_path(vmap), strict=strict) elif vtype == "vhs": - ok = self._compose_simple(seg_id, self._vhs_path(vmap), strict=strict) + video_path = self._vhs_path(vmap) + self._warn_if_stale_vhs(vmap, video_path) + ok = self._compose_simple(seg_id, video_path, strict=strict) elif vtype == "mixed": sources = [self._resolve_source(s) for s in vmap.get("sources", [])] ok = self._compose_mixed(seg_id, sources) @@ -223,16 +230,24 @@ def _find_audio(self, seg_id: str) -> Path | None: def _manim_path(self, vmap: dict[str, Any]) -> Path: src = vmap.get("source", "") - return self.config.animations_dir / "media" / "videos" / "scenes" / "720p30" / src + if not src: + return self.config.animations_dir / "media" / "videos" / "scenes" / "720p30" + + for base in self._manim_video_dirs(): + candidate = base / src + if candidate.exists(): + return candidate + return self._manim_video_dirs()[0] / src def _vhs_path(self, vmap: dict[str, Any]) -> Path: src = vmap.get("source", "") return self.config.terminal_dir / "rendered" / src def _resolve_source(self, source: str) -> Path: - manim_path = self.config.animations_dir / "media" / "videos" / "scenes" / "720p30" / source - if manim_path.exists(): - return manim_path + for base in self._manim_video_dirs(): + manim_path = base / source + if manim_path.exists(): + return manim_path vhs_path = self.config.terminal_dir / "rendered" / source if vhs_path.exists(): return vhs_path @@ -254,13 +269,69 @@ def _probe_duration(path: Path) -> float | None: except (ValueError, subprocess.TimeoutExpired, FileNotFoundError): return None - @staticmethod - def _run_ffmpeg(cmd: list[str]) -> None: + def _run_ffmpeg(self, cmd: list[str]) -> None: + timeout_sec = max(1, int(self.ffmpeg_timeout_sec)) + out_path = Path(cmd[-1]) if cmd else None try: - subprocess.run(cmd, check=True, capture_output=True, text=True, timeout=300) + subprocess.run(cmd, check=True, capture_output=True, text=True, timeout=timeout_sec) except FileNotFoundError: - print(" ERROR: ffmpeg not found in PATH") + raise ComposeError("ffmpeg not found in PATH") except subprocess.CalledProcessError as exc: - print(f" ERROR: ffmpeg failed: {exc.stderr[:300]}") + detail = (exc.stderr or exc.stdout or "")[:400] + raise ComposeError(f"ffmpeg failed: {detail}") except subprocess.TimeoutExpired: - print(" ERROR: ffmpeg timed out") + if out_path and out_path.exists() and out_path.stat().st_size > 0: + print( + f" WARNING: ffmpeg timed out after {timeout_sec}s, " + f"but output exists at {out_path}." + ) + return + raise ComposeError(f"ffmpeg timed out after {timeout_sec}s") + + def _manim_video_dirs(self) -> list[Path]: + root = self.config.animations_dir / "media" / "videos" + quality = str(self.config.manim_quality).strip().lower() + fallback_qualities = [ + quality, + "1080p30", + "1080p60", + "1440p30", + "1440p60", + "720p30", + "480p15", + "2160p60", + ] + + ordered_qualities: list[str] = [] + for q in fallback_qualities: + if q and q not in ordered_qualities: + ordered_qualities.append(q) + + dirs: list[Path] = [] + for q in ordered_qualities: + dirs.append(root / "scenes" / q) + dirs.append(root / q) + return dirs + + def _warn_if_stale_vhs(self, vmap: dict[str, Any], video_path: Path) -> None: + if not self.config.warn_stale_vhs: + return + + tape_name = str(vmap.get("tape", "")).strip() + if not tape_name: + source_name = str(vmap.get("source", "")).strip() + if source_name: + tape_name = f"{Path(source_name).stem}.tape" + if not tape_name: + return + + tape_path = self.config.terminal_dir / tape_name + if not tape_path.exists() or not video_path.exists(): + return + + if tape_path.stat().st_mtime > (video_path.stat().st_mtime + 1): + print( + " WARNING: tape is newer than rendered video " + f"({tape_path.name} > {video_path.name}). " + "Run `docgen vhs` before `docgen compose` to avoid stale output." + ) diff --git a/src/docgen/config.py b/src/docgen/config.py index 5ad77c3..8c8f9d3 100644 --- a/src/docgen/config.py +++ b/src/docgen/config.py @@ -88,6 +88,38 @@ def manim_scenes(self) -> list[str]: def manim_quality(self) -> str: return self.raw.get("manim", {}).get("quality", "720p30") + @property + def manim_path(self) -> str | None: + """Optional absolute/relative path to the Manim executable.""" + value = self.raw.get("manim", {}).get("manim_path") + return str(value) if value else None + + @property + def vhs_path(self) -> str | None: + """Optional absolute/relative path to the VHS executable.""" + value = self.raw.get("vhs", {}).get("vhs_path") + return str(value) if value else None + + # -- Compose ---------------------------------------------------------------- + + @property + def compose_config(self) -> dict[str, Any]: + defaults: dict[str, Any] = { + "ffmpeg_timeout_sec": 300, + "warn_stale_vhs": True, + } + defaults.update(self.raw.get("compose", {})) + return defaults + + @property + def ffmpeg_timeout_sec(self) -> int: + value = self.compose_config.get("ffmpeg_timeout_sec", 300) + return int(value) + + @property + def warn_stale_vhs(self) -> bool: + return bool(self.compose_config.get("warn_stale_vhs", True)) + # -- Validation ------------------------------------------------------------ @property diff --git a/src/docgen/init.py b/src/docgen/init.py index 563a639..4267cff 100644 --- a/src/docgen/init.py +++ b/src/docgen/init.py @@ -253,6 +253,14 @@ def _write_config(plan: InitPlan) -> str: "manim": { "quality": "720p30", "scenes": [f"Scene{s['id']}" for s in plan.segments], + "manim_path": "", + }, + "vhs": { + "vhs_path": "", + }, + "compose": { + "ffmpeg_timeout_sec": 300, + "warn_stale_vhs": True, }, "tts": { "model": plan.tts_model, diff --git a/src/docgen/manim_runner.py b/src/docgen/manim_runner.py index 682e50d..64e2f71 100644 --- a/src/docgen/manim_runner.py +++ b/src/docgen/manim_runner.py @@ -2,9 +2,13 @@ from __future__ import annotations +import re import subprocess +from pathlib import Path from typing import TYPE_CHECKING +from docgen.binaries import resolve_binary + if TYPE_CHECKING: from docgen.config import Config @@ -24,13 +28,24 @@ def render(self, scene: str | None = None) -> None: print(f"[manim] scenes.py not found at {scenes_file}") return - quality_flag = self._quality_flag() + quality_args, quality_label = self._quality_args() + manim_bin = self._resolve_manim_binary() + if not manim_bin: + return + + print(f"[manim] Rendering at {quality_label}") for s in scenes: - self._render_one(scenes_file, s, quality_flag) + self._render_one(manim_bin, scenes_file, s, quality_args) - def _render_one(self, scenes_file, scene_name: str, quality_flag: str) -> None: + def _render_one( + self, + manim_bin: str, + scenes_file: Path, + scene_name: str, + quality_args: list[str], + ) -> None: print(f"[manim] Rendering {scene_name}") - cmd = ["manim", quality_flag, str(scenes_file), scene_name] + cmd = [manim_bin, *quality_args, str(scenes_file), scene_name] try: subprocess.run( cmd, @@ -39,13 +54,61 @@ def _render_one(self, scenes_file, scene_name: str, quality_flag: str) -> None: timeout=300, ) except FileNotFoundError: - print("[manim] manim not found in PATH — install with: pip install manim") + print( + "[manim] manim executable not found. " + "Install with `pip install manim` in this environment or set " + "`manim.manim_path` in docgen.yaml." + ) except subprocess.CalledProcessError as exc: print(f"[manim] FAILED {scene_name}: exit code {exc.returncode}") except subprocess.TimeoutExpired: print(f"[manim] TIMEOUT {scene_name}") - def _quality_flag(self) -> str: - q = self.config.manim_quality - mapping = {"480p15": "-pql", "720p30": "-pqm", "1080p60": "-pqh"} - return mapping.get(q, "-pqm") + def _resolve_manim_binary(self) -> str | None: + configured = self.config.manim_path + if configured and not Path(configured).is_absolute(): + configured = str((self.config.base_dir / configured).resolve()) + + resolution = resolve_binary("manim", configured_path=configured) + if resolution.path: + return resolution.path + + print("[manim] manim executable not found.") + if resolution.tried: + print("[manim] Tried:") + for candidate in resolution.tried: + print(f" - {candidate}") + print( + "[manim] Fix: install with `pip install manim` in this env, " + "or set `manim.manim_path` in docgen.yaml." + ) + return None + + def _quality_args(self) -> tuple[list[str], str]: + q = str(self.config.manim_quality).strip().lower() + preset_map = { + "480p15": (["-pql"], "480p15 (-pql)"), + "720p30": (["-pqm"], "720p30 (-pqm)"), + "1080p60": (["-pqh"], "1080p60 (-pqh)"), + "2160p60": (["-pqp"], "2160p60 (-pqp)"), + } + if q in preset_map: + return preset_map[q] + + match = re.match(r"^(\d{3,4})p(\d{2})$", q) + if match: + height = int(match.group(1)) + fps = int(match.group(2)) + width = (height * 16) // 9 + if width % 2: + width += 1 + return ( + ["--resolution", f"{width},{height}", "--frame_rate", str(fps)], + f"{height}p{fps} (--resolution {width}x{height}, --frame_rate {fps})", + ) + + print( + f"[manim] WARNING: quality '{self.config.manim_quality}' not recognized; " + "falling back to 720p30 (-pqm)." + ) + return (["-pqm"], "720p30 (-pqm fallback)") diff --git a/src/docgen/vhs.py b/src/docgen/vhs.py index e1bed35..40618e9 100644 --- a/src/docgen/vhs.py +++ b/src/docgen/vhs.py @@ -10,6 +10,8 @@ from pathlib import Path from typing import TYPE_CHECKING +from docgen.binaries import resolve_binary + if TYPE_CHECKING: from docgen.config import Config @@ -105,9 +107,18 @@ def _clean_env() -> dict[str, str]: def _render_one(self, tape_path: Path, strict: bool) -> VHSResult: print(f"[vhs] Rendering {tape_path.name}") env = self._clean_env() + vhs_bin = self._resolve_vhs_binary() + if not vhs_bin: + return VHSResult( + tape=tape_path.name, + success=False, + errors=[ + "vhs not found. Install VHS or set vhs.vhs_path in docgen.yaml.", + ], + ) try: proc = subprocess.run( - ["vhs", str(tape_path)], + [vhs_bin, str(tape_path)], capture_output=True, text=True, timeout=300, @@ -151,3 +162,28 @@ def _scan_output(text: str) -> list[str]: found.append(line.strip()[:120]) break return found + + def _resolve_vhs_binary(self) -> str | None: + configured = self.config.vhs_path + if configured and not Path(configured).is_absolute(): + configured = str((self.config.base_dir / configured).resolve()) + + candidates = [ + Path.home() / "go" / "bin" / "vhs", + "/usr/local/bin/vhs", + "/snap/bin/vhs", + ] + resolution = resolve_binary("vhs", configured_path=configured, extra_candidates=candidates) + if resolution.path: + return resolution.path + + print("[vhs] VHS executable not found.") + if resolution.tried: + print("[vhs] Tried:") + for candidate in resolution.tried: + print(f" - {candidate}") + print( + "[vhs] Fix: install VHS and ensure it is executable, or set " + "`vhs.vhs_path` in docgen.yaml." + ) + return None diff --git a/tests/test_compose.py b/tests/test_compose.py new file mode 100644 index 0000000..20230b7 --- /dev/null +++ b/tests/test_compose.py @@ -0,0 +1,82 @@ +"""Tests for compose configuration behavior and source discovery.""" + +from __future__ import annotations + +import os +import time +from pathlib import Path + +import yaml + +from docgen.compose import Composer +from docgen.config import Config + + +def _write_cfg(tmp_path: Path, cfg: dict) -> Config: + path = tmp_path / "docgen.yaml" + path.write_text(yaml.dump(cfg), encoding="utf-8") + return Config.from_yaml(path) + + +def test_manim_source_uses_configured_quality_dir(tmp_path: Path) -> None: + cfg = { + "dirs": {"animations": "animations", "terminal": "terminal", "audio": "audio", "recordings": "recordings"}, + "segments": {"default": ["01"], "all": ["01"]}, + "visual_map": {"01": {"type": "manim", "source": "Scene01.mp4"}}, + "manim": {"quality": "1080p30"}, + } + c = _write_cfg(tmp_path, cfg) + target = tmp_path / "animations" / "media" / "videos" / "scenes" / "1080p30" + target.mkdir(parents=True, exist_ok=True) + (target / "Scene01.mp4").write_text("x", encoding="utf-8") + + composer = Composer(c) + resolved = composer._manim_path(c.visual_map["01"]) + assert resolved == target / "Scene01.mp4" + + +def test_stale_vhs_warning_printed(tmp_path: Path, capsys) -> None: + cfg = { + "dirs": {"terminal": "terminal", "audio": "audio", "recordings": "recordings", "animations": "animations"}, + "segments": {"default": ["01"], "all": ["01"]}, + "visual_map": {"01": {"type": "vhs", "source": "01-demo.mp4", "tape": "01-demo.tape"}}, + "compose": {"warn_stale_vhs": True}, + } + c = _write_cfg(tmp_path, cfg) + tape = tmp_path / "terminal" / "01-demo.tape" + video = tmp_path / "terminal" / "rendered" / "01-demo.mp4" + video.parent.mkdir(parents=True, exist_ok=True) + tape.parent.mkdir(parents=True, exist_ok=True) + tape.write_text("Type \"echo hi\"\n", encoding="utf-8") + video.write_text("video", encoding="utf-8") + # Ensure tape is newer than rendered video. + now = time.time() + os.utime(video, (now - 10, now - 10)) + os.utime(tape, (now, now)) + + composer = Composer(c) + composer._warn_if_stale_vhs(c.visual_map["01"], video) + out = capsys.readouterr().out + assert "tape is newer" in out + + +def test_stale_vhs_warning_can_be_disabled(tmp_path: Path, capsys) -> None: + cfg = { + "dirs": {"terminal": "terminal", "audio": "audio", "recordings": "recordings", "animations": "animations"}, + "segments": {"default": ["01"], "all": ["01"]}, + "visual_map": {"01": {"type": "vhs", "source": "01-demo.mp4", "tape": "01-demo.tape"}}, + "compose": {"warn_stale_vhs": False}, + } + c = _write_cfg(tmp_path, cfg) + tape = tmp_path / "terminal" / "01-demo.tape" + video = tmp_path / "terminal" / "rendered" / "01-demo.mp4" + video.parent.mkdir(parents=True, exist_ok=True) + tape.parent.mkdir(parents=True, exist_ok=True) + tape.write_text("Type \"echo hi\"\n", encoding="utf-8") + video.write_text("video", encoding="utf-8") + tape.touch() + + composer = Composer(c) + composer._warn_if_stale_vhs(c.visual_map["01"], video) + out = capsys.readouterr().out + assert out == "" diff --git a/tests/test_config.py b/tests/test_config.py index 4d04f8a..25d5054 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -56,6 +56,10 @@ def test_defaults(): assert c.manim_quality == "720p30" assert c.max_drift_sec == 2.75 assert c.ocr_config["sample_interval_sec"] == 2 + assert c.ffmpeg_timeout_sec == 300 + assert c.warn_stale_vhs is True + assert c.manim_path is None + assert c.vhs_path is None finally: cfg_path.unlink() @@ -69,3 +73,18 @@ def test_resolved_dirs(tmp_config): c = Config.from_yaml(tmp_config) assert c.narration_dir == tmp_config.parent / "narration" assert c.audio_dir == tmp_config.parent / "audio" + + +def test_binary_paths_and_compose_config(tmp_path): + cfg = { + "manim": {"manim_path": "/opt/bin/manim"}, + "vhs": {"vhs_path": "/opt/bin/vhs"}, + "compose": {"ffmpeg_timeout_sec": 900, "warn_stale_vhs": False}, + } + p = tmp_path / "docgen.yaml" + p.write_text(yaml.dump(cfg), encoding="utf-8") + c = Config.from_yaml(p) + assert c.manim_path == "/opt/bin/manim" + assert c.vhs_path == "/opt/bin/vhs" + assert c.ffmpeg_timeout_sec == 900 + assert c.warn_stale_vhs is False diff --git a/tests/test_manim_runner.py b/tests/test_manim_runner.py new file mode 100644 index 0000000..e97b9ca --- /dev/null +++ b/tests/test_manim_runner.py @@ -0,0 +1,65 @@ +"""Tests for Manim runner quality parsing and binary resolution.""" + +from __future__ import annotations + +from pathlib import Path + +import yaml + +from docgen.config import Config +from docgen.manim_runner import ManimRunner + + +def _config_with_quality(tmp_path: Path, quality: str) -> Config: + cfg = { + "dirs": {"animations": "animations"}, + "manim": {"quality": quality, "scenes": ["Scene01"]}, + "segments": {"default": ["01"], "all": ["01"]}, + } + p = tmp_path / "docgen.yaml" + p.write_text(yaml.dump(cfg), encoding="utf-8") + (tmp_path / "animations").mkdir(parents=True, exist_ok=True) + return Config.from_yaml(p) + + +def test_quality_1080p30_maps_to_resolution(tmp_path: Path) -> None: + cfg = _config_with_quality(tmp_path, "1080p30") + runner = ManimRunner(cfg) + args, label = runner._quality_args() + assert args == ["--resolution", "1920,1080", "--frame_rate", "30"] + assert "1080p30" in label + + +def test_quality_720p30_uses_preset_flag(tmp_path: Path) -> None: + cfg = _config_with_quality(tmp_path, "720p30") + runner = ManimRunner(cfg) + args, label = runner._quality_args() + assert args == ["-pqm"] + assert "720p30" in label + + +def test_quality_unknown_falls_back(tmp_path: Path) -> None: + cfg = _config_with_quality(tmp_path, "banana") + runner = ManimRunner(cfg) + args, _label = runner._quality_args() + assert args == ["-pqm"] + + +def test_resolve_manim_binary_from_config_path(tmp_path: Path) -> None: + manim_bin = tmp_path / "tools" / "manim" + manim_bin.parent.mkdir(parents=True, exist_ok=True) + manim_bin.write_text("#!/usr/bin/env bash\nexit 0\n", encoding="utf-8") + manim_bin.chmod(0o755) + + cfg = { + "dirs": {"animations": "animations"}, + "segments": {"default": ["01"], "all": ["01"]}, + "manim": {"quality": "720p30", "scenes": ["Scene01"], "manim_path": "tools/manim"}, + } + p = tmp_path / "docgen.yaml" + p.write_text(yaml.dump(cfg), encoding="utf-8") + (tmp_path / "animations").mkdir(parents=True, exist_ok=True) + + runner = ManimRunner(Config.from_yaml(p)) + resolved = runner._resolve_manim_binary() + assert resolved == str(manim_bin.resolve()) From 20dc59e87f2cc24b1c308b488255142bfb3e4446 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 14 Apr 2026 03:13:52 +0000 Subject: [PATCH 2/3] feat: add sync-vhs command and optional pipeline integration Co-authored-by: John Menke --- README.md | 18 +++ docs/demos/docgen.yaml | 4 + src/docgen/cli.py | 33 +++- src/docgen/config.py | 37 ++++- src/docgen/init.py | 4 + src/docgen/pipeline.py | 6 + src/docgen/tape_sync.py | 342 ++++++++++++++++++++++++++++++++++++++++ tests/test_config.py | 6 +- tests/test_tape_sync.py | 138 ++++++++++++++++ 9 files changed, 582 insertions(+), 6 deletions(-) create mode 100644 src/docgen/tape_sync.py create mode 100644 tests/test_tape_sync.py diff --git a/README.md b/README.md index 593dbd6..170f958 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,7 @@ docgen validate --pre-push # validate all outputs before committing | `docgen tts [--segment 01] [--dry-run]` | Generate TTS audio | | `docgen manim [--scene StackDAGScene]` | Render Manim animations | | `docgen vhs [--tape 02-quickstart.tape] [--strict]` | Render VHS terminal recordings | +| `docgen sync-vhs [--segment 01] [--dry-run]` | Rewrite VHS `Sleep` values from `animations/timing.json` | | `docgen compose [01 02 03] [--ffmpeg-timeout 900]` | Compose segments (audio + video) | | `docgen validate [--max-drift 2.75] [--pre-push]` | Run all validation checks | | `docgen concat [--config full-demo]` | Concatenate full demo files | @@ -75,6 +76,13 @@ manim: vhs: vhs_path: "" # optional explicit binary path (relative to docgen.yaml or absolute) + sync_from_timing: false # opt-in: allow tape Sleep rewrites from timing.json + typing_ms_per_char: 55 # typing estimate used by sync-vhs + max_typing_sec: 3.0 # per block cap for typing estimate + min_sleep_sec: 0.05 # floor for rewritten Sleep values + +pipeline: + sync_vhs_after_timestamps: false # opt-in: run sync-vhs automatically in generate-all/rebuild-after-audio compose: ffmpeg_timeout_sec: 300 # can also be overridden with: docgen compose --ffmpeg-timeout N @@ -83,6 +91,16 @@ compose: If you edit a `.tape` file, run `docgen vhs` before `docgen compose` so compose does not use stale rendered terminal video. +To auto-align tape pacing with generated narration: + +```bash +docgen timestamps +docgen sync-vhs --dry-run +docgen sync-vhs +docgen vhs +docgen compose +``` + ## System dependencies - **ffmpeg** — composition and probing diff --git a/docs/demos/docgen.yaml b/docs/demos/docgen.yaml index 333999c..921a136 100644 --- a/docs/demos/docgen.yaml +++ b/docs/demos/docgen.yaml @@ -72,6 +72,10 @@ manim: vhs: vhs_path: "" + sync_from_timing: true + typing_ms_per_char: 45 + max_typing_sec: 2.0 + min_sleep_sec: 0.2 compose: ffmpeg_timeout_sec: 300 diff --git a/src/docgen/cli.py b/src/docgen/cli.py index 72bb8df..c889e7c 100644 --- a/src/docgen/cli.py +++ b/src/docgen/cli.py @@ -122,6 +122,18 @@ def vhs(ctx: click.Context, tape: str | None, strict: bool) -> None: click.echo(f" {e}") +@main.command("sync-vhs") +@click.option("--segment", default=None, help="Sync tape(s) for one segment ID/name.") +@click.option("--dry-run", is_flag=True, help="Preview updates without writing files.") +@click.pass_context +def sync_vhs(ctx: click.Context, segment: str | None, dry_run: bool) -> None: + """Sync VHS Sleep durations from animations/timing.json.""" + from docgen.tape_sync import TapeSynchronizer + + cfg = ctx.obj["config"] + TapeSynchronizer(cfg).sync(segment=segment, dry_run=dry_run) + + @main.command() @click.argument("segments", nargs=-1) @click.option( @@ -225,22 +237,35 @@ def pages(ctx: click.Context, force: bool) -> None: @click.option("--skip-tts", is_flag=True) @click.option("--skip-manim", is_flag=True) @click.option("--skip-vhs", is_flag=True) +@click.option("--skip-tape-sync", is_flag=True, help="Skip optional sync-vhs stage after timestamps.") @click.pass_context -def generate_all(ctx: click.Context, skip_tts: bool, skip_manim: bool, skip_vhs: bool) -> None: +def generate_all( + ctx: click.Context, + skip_tts: bool, + skip_manim: bool, + skip_vhs: bool, + skip_tape_sync: bool, +) -> None: """Run full pipeline: TTS -> Manim -> VHS -> compose -> validate -> concat -> pages.""" from docgen.pipeline import Pipeline cfg = ctx.obj["config"] pipeline = Pipeline(cfg) - pipeline.run(skip_tts=skip_tts, skip_manim=skip_manim, skip_vhs=skip_vhs) + pipeline.run( + skip_tts=skip_tts, + skip_manim=skip_manim, + skip_vhs=skip_vhs, + skip_tape_sync=skip_tape_sync, + ) @main.command("rebuild-after-audio") +@click.option("--skip-tape-sync", is_flag=True, help="Skip optional sync-vhs stage after timestamps.") @click.pass_context -def rebuild_after_audio(ctx: click.Context) -> None: +def rebuild_after_audio(ctx: click.Context, skip_tape_sync: bool) -> None: """Rebuild everything after new audio: Manim -> VHS -> compose -> validate -> concat.""" from docgen.pipeline import Pipeline cfg = ctx.obj["config"] pipeline = Pipeline(cfg) - pipeline.run(skip_tts=True) + pipeline.run(skip_tts=True, skip_tape_sync=skip_tape_sync) diff --git a/src/docgen/config.py b/src/docgen/config.py index 8c8f9d3..cae089a 100644 --- a/src/docgen/config.py +++ b/src/docgen/config.py @@ -94,12 +94,47 @@ def manim_path(self) -> str | None: value = self.raw.get("manim", {}).get("manim_path") return str(value) if value else None + @property + def vhs_config(self) -> dict[str, Any]: + defaults: dict[str, Any] = { + "vhs_path": "", + "sync_from_timing": False, + "typing_ms_per_char": 35, + "max_typing_sec": 3.0, + "min_sleep_sec": 0.2, + } + defaults.update(self.raw.get("vhs", {})) + return defaults + @property def vhs_path(self) -> str | None: """Optional absolute/relative path to the VHS executable.""" - value = self.raw.get("vhs", {}).get("vhs_path") + value = self.vhs_config.get("vhs_path") return str(value) if value else None + @property + def sync_from_timing(self) -> bool: + return bool(self.vhs_config.get("sync_from_timing", False)) + + @property + def typing_ms_per_char(self) -> int: + return int(self.vhs_config.get("typing_ms_per_char", 35)) + + @property + def max_typing_sec(self) -> float: + return float(self.vhs_config.get("max_typing_sec", 3.0)) + + @property + def min_sleep_sec(self) -> float: + return float(self.vhs_config.get("min_sleep_sec", 0.2)) + + @property + def sync_vhs_after_timestamps(self) -> bool: + pipeline_cfg = self.raw.get("pipeline", {}) + if "sync_vhs_after_timestamps" in pipeline_cfg: + return bool(pipeline_cfg.get("sync_vhs_after_timestamps")) + return self.sync_from_timing + # -- Compose ---------------------------------------------------------------- @property diff --git a/src/docgen/init.py b/src/docgen/init.py index 4267cff..7a518e1 100644 --- a/src/docgen/init.py +++ b/src/docgen/init.py @@ -257,6 +257,10 @@ def _write_config(plan: InitPlan) -> str: }, "vhs": { "vhs_path": "", + "sync_from_timing": False, + "typing_ms_per_char": 55, + "max_typing_sec": 3.0, + "min_sleep_sec": 0.2, }, "compose": { "ffmpeg_timeout_sec": 300, diff --git a/src/docgen/pipeline.py b/src/docgen/pipeline.py index e1bf6ce..ecfa8c8 100644 --- a/src/docgen/pipeline.py +++ b/src/docgen/pipeline.py @@ -17,6 +17,7 @@ def run( skip_tts: bool = False, skip_manim: bool = False, skip_vhs: bool = False, + skip_tape_sync: bool = False, ) -> None: if not skip_tts: print("\n=== Stage: TTS ===") @@ -27,6 +28,11 @@ def run( from docgen.timestamps import TimestampExtractor TimestampExtractor(self.config).extract_all() + if self.config.sync_vhs_after_timestamps and not skip_tape_sync: + print("\n=== Stage: Sync VHS tape sleep timings ===") + from docgen.tape_sync import TapeSynchronizer + TapeSynchronizer(self.config).sync() + if not skip_manim: print("\n=== Stage: Manim ===") from docgen.manim_runner import ManimRunner diff --git a/src/docgen/tape_sync.py b/src/docgen/tape_sync.py new file mode 100644 index 0000000..2d0bdd9 --- /dev/null +++ b/src/docgen/tape_sync.py @@ -0,0 +1,342 @@ +"""Sync VHS Sleep values from animations/timing.json.""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from docgen.config import Config + + +@dataclass +class TapeSyncChange: + line_no: int + old_sleep_sec: float + new_sleep_sec: float + old_line: str + new_line: str + + +@dataclass +class TapeSyncResult: + tape: str + timing_key: str | None = None + duration_sec: float = 0.0 + blocks_found: int = 0 + changes: list[TapeSyncChange] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + wrote_file: bool = False + + +@dataclass +class _TapeBlock: + type_idx: int + enter_idx: int + sleep_idx: int + typed_text: str + + +class TapeSynchronizer: + def __init__(self, config: Config) -> None: + self.config = config + + def sync(self, segment: str | None = None, dry_run: bool = False) -> list[TapeSyncResult]: + timing = self._load_timing_json() + if not timing: + print("[sync-vhs] No timing.json data found. Run `docgen timestamps` first.") + return [] + + targets = self._collect_targets(segment=segment) + if not targets: + if segment: + print(f"[sync-vhs] No VHS targets matched segment filter '{segment}'.") + else: + print("[sync-vhs] No VHS tapes found to sync.") + return [] + + results: list[TapeSyncResult] = [] + for tape_path, timing_keys in targets: + result = self._sync_one( + tape_path=tape_path, + timing=timing, + timing_keys=timing_keys, + dry_run=dry_run, + ) + self._print_result(result, dry_run=dry_run) + results.append(result) + + changed = sum(1 for r in results if r.changes) + wrote = sum(1 for r in results if r.wrote_file) + print( + f"[sync-vhs] Done: {len(results)} tape(s), {changed} with changes, " + f"{wrote} file(s) written." + ) + return results + + def _collect_targets(self, segment: str | None) -> list[tuple[Path, list[str]]]: + query = segment.lower().strip() if segment else None + targets: list[tuple[Path, list[str]]] = [] + + for seg_id in sorted(self.config.visual_map): + vmap = self.config.visual_map.get(seg_id, {}) + if str(vmap.get("type", "")).lower() != "vhs": + continue + seg_name = self.config.resolve_segment_name(seg_id) + tape_name = str(vmap.get("tape", "")).strip() + if not tape_name: + source_name = str(vmap.get("source", "")).strip() + if source_name: + tape_name = f"{Path(source_name).stem}.tape" + if not tape_name: + continue + + tape_path = self.config.terminal_dir / tape_name + tape_stem = tape_path.stem + if query and query not in {seg_id.lower(), seg_name.lower(), tape_stem.lower()}: + continue + + timing_keys = [seg_name, seg_id, tape_stem] + targets.append((tape_path, self._unique_strings(timing_keys))) + + if targets: + return targets + + # Fallback for legacy projects without visual_map tape metadata. + for tape_path in sorted(self.config.terminal_dir.glob("*.tape")): + tape_stem = tape_path.stem + if query and query != tape_stem.lower(): + continue + targets.append((tape_path, [tape_stem])) + return targets + + @staticmethod + def _unique_strings(values: list[str]) -> list[str]: + seen: set[str] = set() + out: list[str] = [] + for value in values: + if value and value not in seen: + seen.add(value) + out.append(value) + return out + + def _load_timing_json(self) -> dict[str, Any]: + path = self.config.animations_dir / "timing.json" + if not path.exists(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + return data if isinstance(data, dict) else {} + except json.JSONDecodeError: + print(f"[sync-vhs] Invalid JSON: {path}") + return {} + + def _sync_one( + self, + tape_path: Path, + timing: dict[str, Any], + timing_keys: list[str], + dry_run: bool, + ) -> TapeSyncResult: + result = TapeSyncResult(tape=tape_path.name) + if not tape_path.exists(): + result.warnings.append(f"missing tape: {tape_path}") + return result + + timing_key = next((k for k in timing_keys if k in timing), None) + if not timing_key: + result.warnings.append(f"no timing key found (tried: {', '.join(timing_keys)})") + return result + result.timing_key = timing_key + + duration_sec = self._timing_duration_sec(timing[timing_key]) + result.duration_sec = duration_sec + if duration_sec <= 0: + result.warnings.append(f"timing data for '{timing_key}' has zero duration") + return result + + lines = tape_path.read_text(encoding="utf-8").splitlines() + blocks = self._find_blocks(lines) + result.blocks_found = len(blocks) + if not blocks: + result.warnings.append("no Type/Enter/Sleep blocks found after first Show") + return result + + window_sec = duration_sec / len(blocks) + ms_per_char = max(1, self.config.typing_ms_per_char) + max_typing_sec = max(0.0, self.config.max_typing_sec) + min_sleep_sec = max(0.0, self.config.min_sleep_sec) + + for block in blocks: + old_line = lines[block.sleep_idx] + old_sleep = self._parse_sleep_sec(old_line) + if old_sleep is None: + continue + + typing_est = min(max_typing_sec, (len(block.typed_text) * ms_per_char) / 1000.0) + new_sleep = max(min_sleep_sec, window_sec - typing_est) + new_line = self._format_sleep_line(new_sleep) + if new_line == old_line.strip(): + continue + + lines[block.sleep_idx] = new_line + result.changes.append( + TapeSyncChange( + line_no=block.sleep_idx + 1, + old_sleep_sec=old_sleep, + new_sleep_sec=new_sleep, + old_line=old_line, + new_line=new_line, + ) + ) + + if result.changes and not dry_run: + tape_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + result.wrote_file = True + return result + + @staticmethod + def _timing_duration_sec(entry: Any) -> float: + if not isinstance(entry, dict): + return 0.0 + + max_end = 0.0 + for key in ("words", "segments"): + values = entry.get(key, []) + if not isinstance(values, list): + continue + for item in values: + if not isinstance(item, dict): + continue + try: + max_end = max(max_end, float(item.get("end", 0))) + except (TypeError, ValueError): + continue + return max_end + + @staticmethod + def _find_blocks(lines: list[str]) -> list[_TapeBlock]: + show_idx = 0 + for i, line in enumerate(lines): + if line.strip().startswith("Show"): + show_idx = i + break + + blocks: list[_TapeBlock] = [] + i = show_idx + 1 + while i < len(lines): + current = lines[i].strip() + if not current.startswith("Type "): + i += 1 + continue + + type_idx = i + typed_text = TapeSynchronizer._extract_typed_text(current) + enter_idx: int | None = None + sleep_idx: int | None = None + + j = i + 1 + while j < len(lines): + nxt = lines[j].strip() + if nxt.startswith("Type "): + break + if enter_idx is None and nxt.startswith("Enter"): + enter_idx = j + elif enter_idx is not None and nxt.startswith("Sleep "): + sleep_idx = j + break + j += 1 + + if enter_idx is not None and sleep_idx is not None: + blocks.append( + _TapeBlock( + type_idx=type_idx, + enter_idx=enter_idx, + sleep_idx=sleep_idx, + typed_text=typed_text, + ) + ) + i = max(j, i + 1) + + return blocks + + @staticmethod + def _extract_typed_text(type_line: str) -> str: + payload = type_line[len("Type "):].strip() + return TapeSynchronizer._unquote(payload) + + @staticmethod + def _unquote(value: str) -> str: + if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}: + return value[1:-1] + return value + + @staticmethod + def _parse_sleep_sec(line: str) -> float | None: + match = re.match(r"^\s*Sleep\s+([0-9]*\.?[0-9]+)\s*(ms|s)?\s*$", line) + if not match: + return None + value = float(match.group(1)) + unit = (match.group(2) or "s").lower() + if unit == "ms": + return value / 1000.0 + return value + + @staticmethod + def _format_sleep_line(seconds: float) -> str: + if seconds < 1.0: + ms = max(1, int(round(seconds * 1000))) + return f"Sleep {ms}ms" + + rounded = round(seconds, 2) + if abs(rounded - round(rounded)) < 1e-9: + return f"Sleep {int(round(rounded))}s" + return f"Sleep {rounded:.2f}s" + + @staticmethod + def _print_result(result: TapeSyncResult, dry_run: bool) -> None: + prefix = "[sync-vhs] DRY-RUN" if dry_run else "[sync-vhs]" + key_msg = result.timing_key or "no timing key" + print( + f"{prefix} {result.tape}: key={key_msg}, duration={result.duration_sec:.2f}s, " + f"blocks={result.blocks_found}, changes={len(result.changes)}" + ) + for warning in result.warnings: + print(f"{prefix} WARN: {warning}") + for change in result.changes[:10]: + print( + f"{prefix} L{change.line_no}: {change.old_line.strip()} -> " + f"{change.new_line}" + ) + + +def sync_single_tape_from_timing( + tape_path: str | Path, + timing_entry: dict[str, Any], + *, + typing_ms_per_char: int = 45, + max_typing_sec: float = 3.0, + min_sleep_sec: float = 0.15, + dry_run: bool = False, +) -> TapeSyncResult: + """Pure helper used by tests and external callers.""" + path = Path(tape_path) + fake_cfg = type( + "_Cfg", + (), + { + "typing_ms_per_char": typing_ms_per_char, + "max_typing_sec": max_typing_sec, + "min_sleep_sec": min_sleep_sec, + }, + )() + syncer = TapeSynchronizer(fake_cfg) # type: ignore[arg-type] + return syncer._sync_one( # noqa: SLF001 - intentional internal reuse + tape_path=path, + timing={path.stem: timing_entry}, + timing_keys=[path.stem], + dry_run=dry_run, + ) diff --git a/tests/test_config.py b/tests/test_config.py index 25d5054..5fad1f2 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -78,8 +78,9 @@ def test_resolved_dirs(tmp_config): def test_binary_paths_and_compose_config(tmp_path): cfg = { "manim": {"manim_path": "/opt/bin/manim"}, - "vhs": {"vhs_path": "/opt/bin/vhs"}, + "vhs": {"vhs_path": "/opt/bin/vhs", "sync_from_timing": True, "typing_ms_per_char": 40}, "compose": {"ffmpeg_timeout_sec": 900, "warn_stale_vhs": False}, + "pipeline": {"sync_vhs_after_timestamps": True}, } p = tmp_path / "docgen.yaml" p.write_text(yaml.dump(cfg), encoding="utf-8") @@ -88,3 +89,6 @@ def test_binary_paths_and_compose_config(tmp_path): assert c.vhs_path == "/opt/bin/vhs" assert c.ffmpeg_timeout_sec == 900 assert c.warn_stale_vhs is False + assert c.sync_vhs_from_timing is True + assert c.sync_vhs_after_timestamps is True + assert c.typing_ms_per_char == 40 diff --git a/tests/test_tape_sync.py b/tests/test_tape_sync.py new file mode 100644 index 0000000..c443732 --- /dev/null +++ b/tests/test_tape_sync.py @@ -0,0 +1,138 @@ +"""Tests for syncing VHS Sleep values from timing.json.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import yaml + +from docgen.config import Config +from docgen.tape_sync import TapeSynchronizer + + +def _write_cfg(tmp_path: Path, cfg: dict) -> Config: + path = tmp_path / "docgen.yaml" + path.write_text(yaml.dump(cfg), encoding="utf-8") + return Config.from_yaml(path) + + +def test_sync_rewrites_sleep_values(tmp_path: Path) -> None: + cfg = { + "dirs": { + "audio": "audio", + "animations": "animations", + "terminal": "terminal", + "recordings": "recordings", + "narration": "narration", + }, + "segments": {"default": ["01"], "all": ["01"]}, + "segment_names": {"01": "01-demo"}, + "visual_map": {"01": {"type": "vhs", "tape": "01-demo.tape", "source": "01-demo.mp4"}}, + "vhs": { + "sync_from_timing": True, + "typing_ms_per_char": 100, + "max_typing_sec": 0.5, + "min_sleep_sec": 0.1, + }, + } + c = _write_cfg(tmp_path, cfg) + (tmp_path / "animations").mkdir(parents=True, exist_ok=True) + (tmp_path / "terminal").mkdir(parents=True, exist_ok=True) + + timing = { + "01-demo": { + "segments": [ + {"start": 0.0, "end": 2.0}, + {"start": 2.0, "end": 4.0}, + ] + } + } + (tmp_path / "animations" / "timing.json").write_text(json.dumps(timing), encoding="utf-8") + + tape = tmp_path / "terminal" / "01-demo.tape" + tape.write_text( + "\n".join( + [ + 'Set Shell "bash"', + "Show", + 'Type "echo one"', + "Enter", + "Sleep 5s", + 'Type "echo two"', + "Enter", + "Sleep 4s", + "", + ] + ), + encoding="utf-8", + ) + + results = TapeSynchronizer(c).sync() + assert len(results) == 1 + assert results[0].changes + new_text = tape.read_text(encoding="utf-8") + assert "Sleep 5s" not in new_text + assert "Sleep 4s" not in new_text + assert "Sleep " in new_text + + +def test_sync_dry_run_does_not_write(tmp_path: Path) -> None: + cfg = { + "dirs": {"animations": "animations", "terminal": "terminal"}, + "segments": {"default": ["01"], "all": ["01"]}, + "segment_names": {"01": "01-demo"}, + "visual_map": {"01": {"type": "vhs", "tape": "01-demo.tape", "source": "01-demo.mp4"}}, + } + c = _write_cfg(tmp_path, cfg) + (tmp_path / "animations").mkdir(parents=True, exist_ok=True) + (tmp_path / "terminal").mkdir(parents=True, exist_ok=True) + (tmp_path / "animations" / "timing.json").write_text( + json.dumps({"01-demo": {"segments": [{"start": 0.0, "end": 2.0}]}}), + encoding="utf-8", + ) + tape = tmp_path / "terminal" / "01-demo.tape" + original = "\n".join(['Show', 'Type "echo one"', "Enter", "Sleep 5s", ""]) + "\n" + tape.write_text(original, encoding="utf-8") + + results = TapeSynchronizer(c).sync(dry_run=True) + assert len(results) == 1 + assert results[0].changes + assert tape.read_text(encoding="utf-8") == original + + +def test_sync_segment_filter(tmp_path: Path) -> None: + cfg = { + "dirs": {"animations": "animations", "terminal": "terminal"}, + "segments": {"default": ["01", "02"], "all": ["01", "02"]}, + "segment_names": {"01": "01-demo", "02": "02-demo"}, + "visual_map": { + "01": {"type": "vhs", "tape": "01-demo.tape", "source": "01-demo.mp4"}, + "02": {"type": "vhs", "tape": "02-demo.tape", "source": "02-demo.mp4"}, + }, + } + c = _write_cfg(tmp_path, cfg) + (tmp_path / "animations").mkdir(parents=True, exist_ok=True) + (tmp_path / "terminal").mkdir(parents=True, exist_ok=True) + (tmp_path / "animations" / "timing.json").write_text( + json.dumps( + { + "01-demo": {"segments": [{"start": 0.0, "end": 2.0}]}, + "02-demo": {"segments": [{"start": 0.0, "end": 2.0}]}, + } + ), + encoding="utf-8", + ) + (tmp_path / "terminal" / "01-demo.tape").write_text( + "\n".join(["Show", 'Type "a"', "Enter", "Sleep 4s", ""]) + "\n", + encoding="utf-8", + ) + (tmp_path / "terminal" / "02-demo.tape").write_text( + "\n".join(["Show", 'Type "b"', "Enter", "Sleep 4s", ""]) + "\n", + encoding="utf-8", + ) + + results = TapeSynchronizer(c).sync(segment="01") + assert len(results) == 1 + assert results[0].tape == "01-demo.tape" + From 7b04c0e1a8f6e1049acafd4734e0c01b24c4659d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 14 Apr 2026 03:14:42 +0000 Subject: [PATCH 3/3] test: fix config assertion for sync-from-timing Co-authored-by: John Menke --- tests/test_config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_config.py b/tests/test_config.py index 5fad1f2..b616d94 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -89,6 +89,6 @@ def test_binary_paths_and_compose_config(tmp_path): assert c.vhs_path == "/opt/bin/vhs" assert c.ffmpeg_timeout_sec == 900 assert c.warn_stale_vhs is False - assert c.sync_vhs_from_timing is True + assert c.sync_from_timing is True assert c.sync_vhs_after_timestamps is True assert c.typing_ms_per_char == 40