Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions src/docgen/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,29 @@ def lint(ctx: click.Context, segment: str | None) -> None:
raise SystemExit(1)


@main.command("scene-lint")
@click.pass_context
def scene_lint(ctx: click.Context) -> None:
"""Lint Manim scene files for known pitfalls (weight=BOLD, positional color args)."""
from docgen.scene_lint import lint_scene_dir

cfg = ctx.obj["config"]
results = lint_scene_dir(cfg)

if not results:
click.echo("[scene-lint] No issues found")
return

for r in results:
click.echo(f" {r.path}")
for issue in r.issues:
click.echo(f" {issue}")

total = sum(len(r.issues) for r in results)
click.echo(f"\n[scene-lint] {total} issue(s) in {len(results)} file(s)")
raise SystemExit(1)


@main.command()
@click.option("--config-name", "concat_name", default=None, help="Concat config name.")
@click.pass_context
Expand Down
66 changes: 59 additions & 7 deletions src/docgen/compose.py
Original file line number Diff line number Diff line change
Expand Up @@ -223,14 +223,63 @@ 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
base = self.config.animations_dir / "media" / "videos" / "scenes"
return self._resolve_quality_path(base, src)

def _vhs_path(self, vmap: dict[str, Any]) -> Path:
src = vmap.get("source", "")
return self.config.terminal_dir / "rendered" / src
rendered = self.config.terminal_dir / "rendered" / src
if self.config.compose_warn_stale:
self._check_stale_vhs(src, rendered)
return rendered

def _check_stale_vhs(self, source_name: str, rendered: Path) -> None:
"""Warn if a .tape file is newer than its rendered mp4."""
if not rendered.exists():
return
stem = Path(source_name).stem
terminal_dir = self.config.terminal_dir
for tape in terminal_dir.glob(f"*{stem}*.tape"):
if tape.stat().st_mtime > rendered.stat().st_mtime:
print(
f" WARNING: {tape.name} is newer than {rendered.name} — "
f"run 'docgen vhs' to re-render before composing."
)
break

def _resolve_quality_path(self, base: Path, source: str) -> Path:
"""Find the rendered Manim file, trying the configured quality first,
then falling back to any available quality directory."""
from docgen.manim_runner import ManimRunner
preferred = ManimRunner(self.config).quality_subdir()

candidate = base / preferred / source
if candidate.exists():
return candidate

for subdir in ("1080p60", "1080p30", "1440p60", "720p30", "480p15", "2160p60"):
candidate = base / subdir / source
if candidate.exists():
if subdir != preferred:
print(
f" NOTE: using {subdir}/{source} "
f"(configured quality {preferred} not found)"
)
return candidate

no_scenes = base.parent / preferred / source
if no_scenes.exists():
return no_scenes
for subdir in ("1080p60", "1080p30", "1440p60", "720p30", "480p15", "2160p60"):
no_scenes = base.parent / subdir / source
if no_scenes.exists():
return no_scenes

return base / preferred / source

def _resolve_source(self, source: str) -> Path:
manim_path = self.config.animations_dir / "media" / "videos" / "scenes" / "720p30" / source
base = self.config.animations_dir / "media" / "videos" / "scenes"
manim_path = self._resolve_quality_path(base, source)
if manim_path.exists():
return manim_path
vhs_path = self.config.terminal_dir / "rendered" / source
Expand All @@ -254,13 +303,16 @@ 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 = self.config.ffmpeg_timeout
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)
except FileNotFoundError:
print(" ERROR: ffmpeg not found in PATH")
except subprocess.CalledProcessError as exc:
print(f" ERROR: ffmpeg failed: {exc.stderr[:300]}")
except subprocess.TimeoutExpired:
print(" ERROR: ffmpeg timed out")
print(
f" ERROR: ffmpeg timed out after {timeout}s. "
f"Increase compose.ffmpeg_timeout in docgen.yaml for long scenes."
)
16 changes: 16 additions & 0 deletions src/docgen/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,22 @@ def manim_scenes(self) -> list[str]:
def manim_quality(self) -> str:
return self.raw.get("manim", {}).get("quality", "720p30")

@property
def manim_font(self) -> str:
return self.raw.get("manim", {}).get("font", "Liberation Sans")

# -- Compose / ffmpeg ------------------------------------------------------

@property
def ffmpeg_timeout(self) -> int:
"""Timeout in seconds for ffmpeg subprocess calls."""
return int(self.raw.get("compose", {}).get("ffmpeg_timeout", 600))

@property
def compose_warn_stale(self) -> bool:
"""Warn during compose if a VHS .tape file is newer than its rendered mp4."""
return bool(self.raw.get("compose", {}).get("warn_stale_vhs", True))

# -- Validation ------------------------------------------------------------

@property
Expand Down
106 changes: 95 additions & 11 deletions src/docgen/manim_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,25 @@

from __future__ import annotations

import re
import shutil
import subprocess
import sys
from pathlib import Path
from typing import TYPE_CHECKING

if TYPE_CHECKING:
from docgen.config import Config

_PRESET_FLAGS: dict[str, list[str]] = {
"480p15": ["-ql"],
"720p30": ["-qm"],
"1080p60": ["-qh"],
"2160p60": ["-qp"],
}

_CUSTOM_RE = re.compile(r"^(\d+)p(\d+)$")


class ManimRunner:
def __init__(self, config: Config) -> None:
Expand All @@ -24,28 +37,99 @@ def render(self, scene: str | None = None) -> None:
print(f"[manim] scenes.py not found at {scenes_file}")
return

quality_flag = self._quality_flag()
manim_bin = self._find_manim()
if not manim_bin:
return

quality_flags = self._quality_flags()
for s in scenes:
self._render_one(scenes_file, s, quality_flag)
self._render_one(manim_bin, scenes_file, s, quality_flags)

def _render_one(self, scenes_file, scene_name: str, quality_flag: str) -> None:
print(f"[manim] Rendering {scene_name}")
cmd = ["manim", quality_flag, str(scenes_file), scene_name]
def _find_manim(self) -> str | None:
"""Locate the manim binary, checking the active venv first."""
venv = Path(sys.prefix) / "bin" / "manim"
if venv.is_file():
return str(venv)

found = shutil.which("manim")
if found:
return found

print(
"[manim] manim not found in PATH. "
"Install with: pip install manim (in this venv) "
"or set PATH to include the directory containing manim."
)
return None

def _render_one(
self, manim_bin: str, scenes_file: Path, scene_name: str, quality_flags: list[str]
) -> None:
quality_label = self.config.manim_quality
print(f"[manim] Rendering {scene_name} at {quality_label}")
cmd = [manim_bin, *quality_flags, str(scenes_file), scene_name]
try:
subprocess.run(
cmd,
check=True,
cwd=str(self.config.animations_dir),
timeout=300,
timeout=self.config.ffmpeg_timeout,
)
except FileNotFoundError:
print("[manim] manim not found in PATH — install with: pip install manim")
print(
"[manim] manim not found in PATH. "
"Install with: pip install manim"
)
except subprocess.CalledProcessError as exc:
print(f"[manim] FAILED {scene_name}: exit code {exc.returncode}")
except subprocess.TimeoutExpired:
print(f"[manim] TIMEOUT {scene_name}")
print(f"[manim] TIMEOUT {scene_name} (limit {self.config.ffmpeg_timeout}s)")

def _quality_flag(self) -> str:
def _quality_flags(self) -> list[str]:
"""Return CLI flags for Manim based on the configured quality string.

Recognised presets: 480p15, 720p30, 1080p60, 2160p60.
Arbitrary ``<height>p<fps>`` strings (e.g. ``1080p30``) are parsed
into explicit ``--resolution`` and ``--frame_rate`` flags.
"""
q = self.config.manim_quality
mapping = {"480p15": "-pql", "720p30": "-pqm", "1080p60": "-pqh"}
return mapping.get(q, "-pqm")
if q in _PRESET_FLAGS:
return list(_PRESET_FLAGS[q])

m = _CUSTOM_RE.match(q)
if m:
height = int(m.group(1))
fps = int(m.group(2))
width = _width_for_height(height)
print(f"[manim] Using custom quality {width}x{height} @ {fps}fps")
return ["--resolution", f"{width},{height}", "--frame_rate", str(fps)]

valid = ", ".join(sorted(_PRESET_FLAGS.keys()))
print(
f"[manim] WARNING: quality '{q}' not recognised, "
f"falling back to 720p30. Valid presets: {valid} "
f"(or use <height>p<fps> e.g. 1080p30)"
)
return ["-qm"]

def quality_subdir(self) -> str:
"""Return the subdirectory name Manim uses for the configured quality."""
q = self.config.manim_quality
preset_dirs = {
"480p15": "480p15",
"720p30": "720p30",
"1080p60": "1080p60",
"2160p60": "2160p60",
}
if q in preset_dirs:
return preset_dirs[q]
m = _CUSTOM_RE.match(q)
if m:
return q
return "720p30"


def _width_for_height(height: int) -> int:
"""Derive 16:9 width from height, rounding to even."""
w = int(height * 16 / 9)
return w + (w % 2)
24 changes: 22 additions & 2 deletions src/docgen/pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from __future__ import annotations

import shutil
from typing import TYPE_CHECKING

if TYPE_CHECKING:
Expand Down Expand Up @@ -41,8 +42,27 @@ def run(
print(f" WARNING: {r.tape} had errors: {r.errors}")

print("\n=== Stage: Compose ===")
from docgen.compose import Composer
Composer(self.config).compose_segments(self.config.segments_all)
from docgen.compose import Composer, ComposeError
composer = Composer(self.config)
try:
composer.compose_segments(self.config.segments_all)
except ComposeError:
if skip_manim:
raise
print(
"\n=== Stage: Manim (retry) ===\n"
"[pipeline] Compose hit FREEZE GUARD — this often happens on the "
"first run because Manim scenes need timing data from timestamps.\n"
"[pipeline] Clearing Manim cache and re-rendering…"
)
media_dir = self.config.animations_dir / "media"
if media_dir.exists():
shutil.rmtree(media_dir)
from docgen.manim_runner import ManimRunner as _MR
_MR(self.config).render()

print("\n=== Stage: Compose (retry) ===")
composer.compose_segments(self.config.segments_all)

print("\n=== Stage: Validate ===")
from docgen.validate import Validator
Expand Down
75 changes: 75 additions & 0 deletions src/docgen/scene_lint.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
"""Lint Manim scene files for known pitfalls: bold weight, positional color args."""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING

if TYPE_CHECKING:
from docgen.config import Config


@dataclass
class SceneLintResult:
path: str
passed: bool
issues: list[str] = field(default_factory=list)


_BOLD_PATTERN = re.compile(r"\bweight\s*=\s*BOLD\b")

_TEXT_POSITIONAL_COLOR = re.compile(
r"""Text\(\s*(?:f?["'][^"']*["']|[A-Za-z_]\w*)\s*,\s*["']#[0-9a-fA-F]""",
)

_TEXT_POSITIONAL_COLOR_VAR = re.compile(
r"""Text\(\s*(?:f?["'][^"']*["']|[A-Za-z_]\w*)\s*,\s*C_[A-Z_]+""",
)


def lint_scene_file(path: Path) -> SceneLintResult:
"""Scan a single Manim scene file for known issues."""
result = SceneLintResult(path=str(path), passed=True)

if not path.exists():
return result

text = path.read_text(encoding="utf-8")
for lineno, line in enumerate(text.splitlines(), 1):
stripped = line.strip()
if stripped.startswith("#"):
continue

if _BOLD_PATTERN.search(line):
result.issues.append(
f"Line {lineno}: weight=BOLD causes Pango font substitution — "
f"use font_size and color for emphasis instead"
)
result.passed = False

if _TEXT_POSITIONAL_COLOR.search(line) or _TEXT_POSITIONAL_COLOR_VAR.search(line):
result.issues.append(
f"Line {lineno}: color passed as positional arg to Text() — "
f"use keyword: Text(..., color=C_COLOR)"
)
result.passed = False

return result


def lint_scene_dir(config: Config) -> list[SceneLintResult]:
"""Lint all .py files under the animations directory."""
anim_dir = config.animations_dir
if not anim_dir.exists():
return []

results: list[SceneLintResult] = []
for py_file in sorted(anim_dir.glob("**/*.py")):
if py_file.name.startswith("_"):
continue
result = lint_scene_file(py_file)
if result.issues:
results.append(result)
return results
Loading