From 4913195b9d77882bac87e1dddea44684efa1fa13 Mon Sep 17 00:00:00 2001 From: jenstandstad Date: Thu, 23 Jul 2026 19:04:15 +0200 Subject: [PATCH] Adding fixes to video build --- gnommo/cli.py | 126 +++++++++++++++- gnommo/scaffold.py | 343 ++++++++++++++++++++++++++++++++++++++++++ gnommo/transformer.py | 24 ++- 3 files changed, 480 insertions(+), 13 deletions(-) create mode 100644 gnommo/scaffold.py diff --git a/gnommo/cli.py b/gnommo/cli.py index 25a9fa2..eb7067e 100644 --- a/gnommo/cli.py +++ b/gnommo/cli.py @@ -126,6 +126,7 @@ Examples: "preprocess", "pre", "trim", + "build", "render", "grade", "all", @@ -167,6 +168,12 @@ Examples: action="store_true", help="Show what would be done without executing", ) + parser.add_argument( + "--realign", + action="store_true", + help="For build/render: discard existing events.json times and re-align " + "manuscript markers to the transcript from scratch (e.g. after re-recording)", + ) parser.add_argument( "--slides", type=str, @@ -368,6 +375,16 @@ Examples: args.processed, args.alpha_quality, ) + elif action == "build": + return cmd_build( + project_path, + args.verbose, + args.dry_run, + args.slides, + args.res, + args.force, + realign=args.realign, + ) elif action == "render": return cmd_render( project_path, @@ -377,6 +394,7 @@ Examples: args.res, args.force, chunk_slides=args.chunk_slides, + realign=args.realign, ) elif action == "grade": return cmd_grade( @@ -4160,6 +4178,35 @@ def _build_merged_transcription(project_path: Path, config, verbose: bool = Fals return merged or None +def cmd_build( + project_path: Path, + verbose: bool, + dry_run: bool, + slides_arg: str = None, + res: str = "full", + force: bool = False, + realign: bool = False, +) -> int: + """Build the timing scaffold (events.json + scaffold.json) without rendering. + + Aligns manuscript markers to the transcript (or reuses/updates events.json + times), writes the editable events.json + compiled scaffold.json, and stops. + Edit events.json to nudge slide/video timings, then run `render`. Pass + --realign to discard existing events.json times and re-align from scratch + (e.g. after re-recording narration). + """ + return cmd_render( + project_path, + verbose, + dry_run, + slides_arg, + res, + force, + plan_only=True, + realign=realign, + ) + + def cmd_render( project_path: Path, verbose: bool, @@ -4169,8 +4216,17 @@ def cmd_render( force: bool = False, chunk_slides: int = 0, _output_path_override: Path = None, + plan_only: bool = False, + realign: bool = False, ) -> int: - """Render final video.""" + """Render final video. + + Two-stage timing model: this builds/refreshes events.json + scaffold.json (the + editable timing layer) and then renders. When events.json already exists its + times are used verbatim (hand-edits win, no re-alignment) unless realign=True. + With plan_only=True it stops after writing the scaffold — that's the `build` + command. + """ from .parser import ( parse_audio, parse_manuscript, @@ -4186,12 +4242,13 @@ def cmd_render( from .preprocessor import RES_CONFIGS, ensure_downscaled_files_exist # Parse slide range if provided + _verb = "Building scaffold" if plan_only else "Rendering" slide_range = None if slides_arg: slide_range = _parse_slide_range(slides_arg) - print(f"Rendering: {project_path.name} (slides {slides_arg})") + print(f"{_verb}: {project_path.name} (slides {slides_arg})") else: - print(f"Rendering: {project_path.name}") + print(f"{_verb}: {project_path.name}") # Show resolution mode if res != "full": @@ -4329,8 +4386,17 @@ def cmd_render( print(f" Warning: {w}") print(" Passed.") - # Stage 3: Transform (includes on-the-fly alignment) - print("\n[3/4] Building render plan...") + # Stage 3: Transform (alignment, unless a prior events.json pins the times) + from . import scaffold as _scaffold + + _existing_events = None if realign else _scaffold.read_events(project_path) + _timings_override = ( + _scaffold.events_to_marker_timings(_existing_events) if _existing_events else None + ) + if _timings_override is not None: + print("\n[3/4] Building render plan (using events.json timings)...") + else: + print("\n[3/4] Building render plan (aligning to transcript)...") plan, marker_timings = build_render_plan( project_path, config, @@ -4344,6 +4410,7 @@ def cmd_render( slide_range=slide_range, narration_schedule=narration_schedule, narration_source=narration_source, + marker_timings_override=_timings_override, ) if plan.time_offset > 0: print(f" Time offset: {plan.time_offset:.1f}s (partial render)") @@ -4397,6 +4464,46 @@ def cmd_render( ] _write_tasks_file(project_path, missing_videos, alignment_issues) + # --- Timing layer: derive events, keep hand-edits, persist events.json + scaffold.json --- + # Only for a full-timeline pass — partial (--slides) and internal chunk renders + # read the override but must never overwrite the complete events.json/scaffold.json. + if slide_range is None and _output_path_override is None: + _events = _scaffold.derive_events(marker_timings, slides, videos, audio) + _events = _scaffold.merge_events(_events, _existing_events) + # Recompute interpolated (non-anchor) times between exact/manual anchors, + # so fixing a few anchor times and rebuilding redistributes the rest. + _scaffold.reinterpolate_events(_events) + _scaffold.write_events(project_path, _events) + _scaffold.write_scaffold( + project_path, _events, transcription, narration_schedule, plan.total_duration + ) + # Spoken transcript with the aligned markers interleaved — diff against + # manuscript.txt to see where the recording drifts from the script. + _scaffold.write_transcribed_manuscript(project_path, _events, transcription) + _summ = _scaffold.mapping_summary(_events) + print( + f"\n Timing layer: {_summ['exact']} exact, {_summ['interpolated']} interpolated, " + f"{_summ['manual']} manual → {_scaffold.EVENTS_FILE} + {_scaffold.SCAFFOLD_FILE} " + f"+ {_scaffold.TRANSCRIBED_FILE}" + ) + if plan_only: + if _summ["interpolated"]: + print( + f" {_summ['interpolated']} event(s) interpolated. Fix a few anchor times in " + f"{_scaffold.EVENTS_FILE} and set their \"mapping\" to \"manual\" to pin them; " + f"rebuild redistributes the rest between anchors." + ) + print( + f" Compare {_scaffold.TRANSCRIBED_FILE} against manuscript.txt to see the drift, " + f"then run 'gnommo -p {project_path.name} render'." + ) + else: + print(f" Run 'gnommo -p {project_path.name} render' to produce the video.") + return 0 + elif plan_only: + # Partial build (--slides): nothing to persist for the full scaffold. + return 0 + # Check for unaligned markers unaligned = [t for t in marker_timings if t.timestamp < 0] if unaligned: @@ -5343,8 +5450,13 @@ def cmd_all( if _files_modified_since(project_path, t0, "narration.json"): cascade_force = True - print("\n>>> Step 5/8: Render\n") - result = cmd_render(project_path, verbose, dry_run, res=res, force=cascade_force) + print("\n>>> Step 5/8: Build scaffold + Render\n") + # render also writes events.json/scaffold.json. When upstream changed + # (cascade_force), re-align from the transcript; otherwise honour any + # hand-edits already in events.json. + result = cmd_render( + project_path, verbose, dry_run, res=res, force=cascade_force, realign=cascade_force + ) if result != 0: return result diff --git a/gnommo/scaffold.py b/gnommo/scaffold.py new file mode 100644 index 0000000..a12662b --- /dev/null +++ b/gnommo/scaffold.py @@ -0,0 +1,343 @@ +"""events.json + scaffold.json — the timing layer between alignment and render. + +The render stage is split in two: + + build aligns manuscript markers to the transcript, then writes + events.json — one editable entry per marker: {type, id, time, mapping} + scaffold.json — the compiled timeline (transcription + narration + schedule + merged events + total_duration) + render loads the scaffold's event times (honouring any hand-edits to + events.json) and renders WITHOUT re-aligning. + +events.json is the authoritative, GUI-editable seam: a future gnommoweb +review-stage timeline can read it, let the user drag slide-marker / video-start +events around, and write it straight back. Because it stores absolute +final-timeline times with stable ids, edits always win over the fuzzy aligner. + +mapping values: + exact a confident fuzzy match to the spoken words + interpolated placed by the aligner's repair/fallback path, or filled in + between neighbours because the phrase wasn't found at all + manual set by a human / the GUI; never recomputed by build +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Optional + +from .models import CAMERA_PRESETS +from .transformer import MarkerTiming + +EVENTS_FILE = "events.json" +SCAFFOLD_FILE = "scaffold.json" +TRANSCRIBED_FILE = "manuscript_transcribed.txt" + +MAPPING_EXACT = "exact" +MAPPING_INTERPOLATED = "interpolated" +MAPPING_MANUAL = "manual" + +# Confidence at/above this counts as an exact match (mirrors the aligner default). +_EXACT_THRESHOLD = 0.6 + + +# ── marker classification ───────────────────────────────────────────────────── + +def marker_type(marker_id: str, slides: dict, videos: dict, audio: dict) -> str: + """Classify a marker id as slide / video / audio / camera / other.""" + if slides and marker_id in slides: + return "slide" + if videos and marker_id in videos: + return "video" + if audio and marker_id in audio: + return "audio" + if marker_id in CAMERA_PRESETS: + return "camera" + return "other" + + +# ── derive events from aligner output ───────────────────────────────────────── + +def derive_events( + marker_timings: list[MarkerTiming], + slides: dict, + videos: dict, + audio: dict, +) -> list[dict]: + """Turn ordered MarkerTiming objects into event dicts with a mapping tag. + + Markers the aligner failed to place (timestamp < 0) are interpolated between + their nearest placed neighbours so nothing silently vanishes — they surface + as `interpolated` for a human/GUI to nudge. + """ + events: list[dict] = [] + for t in marker_timings: + placed = t.timestamp is not None and t.timestamp >= 0 + after_prev = (t.context or "").startswith("(after previous)") + if not placed: + mapping = MAPPING_INTERPOLATED + elif t.confidence >= _EXACT_THRESHOLD and not after_prev: + mapping = MAPPING_EXACT + else: + mapping = MAPPING_INTERPOLATED + events.append( + { + "type": marker_type(t.marker_id, slides, videos, audio), + "id": t.marker_id, + "time": round(t.timestamp, 3) if placed else None, + "mapping": mapping, + "confidence": round(t.confidence, 3), + "context": (t.context or "")[:80], + } + ) + + _interpolate_missing_times(events) + return events + + +def _interpolate_missing_times(events: list[dict]) -> None: + """Fill `time: None` entries by linear interpolation between placed neighbours. + + Runs at the head/tail (no anchor on one side) are spread at +1s steps from the + nearest known time (or 0.0). Mutates events in place. + """ + n = len(events) + i = 0 + while i < n: + if events[i]["time"] is not None: + i += 1 + continue + # [i, j) is a run of missing times. + j = i + while j < n and events[j]["time"] is None: + j += 1 + prev_time = events[i - 1]["time"] if i > 0 else None + next_time = events[j]["time"] if j < n else None + gap = j - i + if prev_time is not None and next_time is not None: + step = (next_time - prev_time) / (gap + 1) + for k in range(gap): + events[i + k]["time"] = round(prev_time + step * (k + 1), 3) + elif prev_time is not None: # trailing run + for k in range(gap): + events[i + k]["time"] = round(prev_time + 1.0 * (k + 1), 3) + elif next_time is not None: # leading run + base = max(0.0, next_time - gap) + for k in range(gap): + events[i + k]["time"] = round(base + 1.0 * k, 3) + else: # nothing placed at all + for k in range(gap): + events[i + k]["time"] = round(1.0 * (i + k), 3) + i = j + + +# ── re-interpolation on rebuild ─────────────────────────────────────────────── + +def reinterpolate_events(events: list[dict]) -> None: + """Recompute every `interpolated` event's time between its anchor neighbours. + + Anchors are the events a human/aligner trusts — mapping `exact` or `manual`. + Interpolated events carry no real spoken cue, so their time is only ever a + guess spread between the surrounding anchors. Recomputing them on each build + is what makes the workflow usable: fix a couple of real anchor times (mark + them `manual`), rebuild, and the visual/build slides between them redistribute + themselves instead of staying frozen where an earlier bad alignment put them. + + Mutates events in place. + """ + for e in events: + if e.get("mapping") == MAPPING_INTERPOLATED: + e["time"] = None + _interpolate_missing_times(events) + + +# ── merge: hand-edits win ───────────────────────────────────────────────────── + +def merge_events(fresh: list[dict], existing: Optional[list[dict]]) -> list[dict]: + """Overlay freshly-derived events with any existing per-id edits. + + For every id already in events.json we keep its stored `time` and `mapping` + (that's the user's / GUI's edit or a previously accepted alignment). Fresh + ids not present before are added; ids no longer in the manuscript are dropped + (they simply aren't in `fresh`). + """ + if not existing: + return fresh + by_id = {e.get("id"): e for e in existing} + out: list[dict] = [] + for e in fresh: + old = by_id.get(e["id"]) + if old is not None: + if old.get("time") is not None: + e["time"] = old["time"] + if old.get("mapping"): + e["mapping"] = old["mapping"] + out.append(e) + return out + + +# ── events <-> marker timings ───────────────────────────────────────────────── + +def events_to_marker_timings(events: list[dict]) -> list[MarkerTiming]: + """Reconstruct MarkerTiming objects from events for build_render_plan override.""" + timings: list[MarkerTiming] = [] + for e in events: + time = e.get("time") + timings.append( + MarkerTiming( + marker_id=e["id"], + timestamp=float(time) if time is not None else -1.0, + context=e.get("context", ""), + confidence=float(e.get("confidence", 1.0)), + ) + ) + return timings + + +# ── serialization ───────────────────────────────────────────────────────────── + +def read_events(project_path: Path) -> Optional[list[dict]]: + p = project_path / EVENTS_FILE + if not p.exists(): + return None + try: + return json.loads(p.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + return None + + +def write_events(project_path: Path, events: list[dict]) -> Path: + p = project_path / EVENTS_FILE + p.write_text(json.dumps(events, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + return p + + +def read_scaffold(project_path: Path) -> Optional[dict]: + p = project_path / SCAFFOLD_FILE + if not p.exists(): + return None + try: + return json.loads(p.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + return None + + +def write_scaffold( + project_path: Path, + events: list[dict], + transcription: list, + narration_schedule: list, + total_duration: float, +) -> Path: + """Write the compiled timeline document render consumes. + + Stores times + ids (not heavy slide/video objects — those are re-resolved + from disk at render). Includes the transcription so the future timeline GUI + can show the spoken words under the draggable markers. + """ + scaffold = { + "version": 1, + "total_duration": round(total_duration, 3), + "narration": [ + { + "seg_id": s.seg_id, + "source_path": str(s.source_path), + "skip": round(s.skip, 3), + "take": (round(s.take, 3) if s.take is not None else None), + "duration": round(s.duration, 3), + "offset": round(s.offset, 3), + } + for s in (narration_schedule or []) + ], + "events": events, + "transcription": [ + {"word": w.word, "start": round(w.start, 3), "end": round(w.end, 3)} + for w in (transcription or []) + ], + } + p = project_path / SCAFFOLD_FILE + p.write_text(json.dumps(scaffold, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + return p + + +# ── transcribed manuscript (for diffing against the script) ─────────────────── + +def _w_start(w) -> float: + return w["start"] if isinstance(w, dict) else w.start + + +def _w_word(w) -> str: + return w["word"] if isinstance(w, dict) else w.word + + +def _fmt_marker(e: dict) -> str: + """Render a marker line, mirroring manuscript.txt so the two files diff cleanly.""" + mid = e["id"] + t = e.get("type") + if t == "video": + return f"[video:{mid}]" + if t == "audio": + return f"[audio:{mid}]" + return f"[{mid}]" # slide / camera / other + + +def build_transcribed_manuscript(events: list[dict], transcription: list) -> str: + """Interleave the aligned markers into the ACTUAL spoken transcript. + + Walks the transcript along the timeline and drops each marker (at its aligned + time) on its own line followed by the words spoken until the next marker — + the same shape as manuscript.txt, but with what was really said. Diff the two + to see exactly where the recording drifted from the script. + """ + evs = sorted( + [e for e in events if e.get("time") is not None], + key=lambda e: e["time"], + ) + words = transcription or [] + n = len(words) + + lines: list[str] = [ + "# Auto-generated from the spoken transcript with aligned markers.", + "# Diff against manuscript.txt to see where the recording drifts from the script.", + "", + ] + buf: list[str] = [] + + def _flush() -> None: + if buf: + lines.append(" ".join(buf)) + buf.clear() + + wi = 0 + for e in evs: + t = e["time"] + while wi < n and _w_start(words[wi]) < t: + buf.append(_w_word(words[wi])) + wi += 1 + _flush() + if lines and lines[-1] != "": + lines.append("") + lines.append(_fmt_marker(e)) + while wi < n: + buf.append(_w_word(words[wi])) + wi += 1 + _flush() + + return "\n".join(lines).rstrip() + "\n" + + +def write_transcribed_manuscript( + project_path: Path, events: list[dict], transcription: list +) -> Path: + p = project_path / TRANSCRIBED_FILE + p.write_text(build_transcribed_manuscript(events, transcription), encoding="utf-8") + return p + + +def mapping_summary(events: list[dict]) -> dict: + """Count events by mapping for a concise build/render summary.""" + out = {MAPPING_EXACT: 0, MAPPING_INTERPOLATED: 0, MAPPING_MANUAL: 0} + for e in events: + out[e.get("mapping", MAPPING_INTERPOLATED)] = out.get(e.get("mapping", MAPPING_INTERPOLATED), 0) + 1 + return out diff --git a/gnommo/transformer.py b/gnommo/transformer.py index e714fb5..d108a83 100644 --- a/gnommo/transformer.py +++ b/gnommo/transformer.py @@ -639,6 +639,7 @@ def build_render_plan( slide_range: Optional[tuple[str, Optional[str]]] = None, narration_schedule: Optional[list] = None, narration_source: Optional[VideoSource] = None, + marker_timings_override: Optional[list["MarkerTiming"]] = None, ) -> tuple[RenderPlan, list[MarkerTiming]]: """ Build a complete render plan from manuscript and transcription. @@ -650,6 +651,12 @@ def build_render_plan( manuscript_text: The manuscript.txt content (source of truth for markers) transcription: Word-level timestamps from whisper transcription slide_range: Optional tuple of (start_slide, end_slide) for partial rendering. + marker_timings_override: When provided (e.g. loaded from events.json / + scaffold.json), these timings are used verbatim instead of aligning + against the transcript. Their timestamps are already final-timeline + values, so the narration-skip adjustment below is skipped for them. + This is the seam that lets `render` consume a hand-edited scaffold + without re-running (and re-breaking on) fuzzy alignment. Returns: Tuple of (RenderPlan, list of MarkerTiming for display) @@ -657,10 +664,14 @@ def build_render_plan( audio = audio or {} audio_dir = audio_dir or project_path - # Align markers to transcription timestamps - marker_timings = align_markers_to_transcription( - manuscript_text, transcription, slides=slides, videos=videos, audio=audio - ) + # Align markers to transcription timestamps — unless caller supplied timings + # (from the scaffold/events layer), in which case those win verbatim. + if marker_timings_override is not None: + marker_timings = marker_timings_override + else: + marker_timings = align_markers_to_transcription( + manuscript_text, transcription, slides=slides, videos=videos, audio=audio + ) # Find shared_assets directory shared_assets_dir = None @@ -702,8 +713,9 @@ def build_render_plan( full_duration = get_video_duration(video_path) # Apply skip offset: if narration starts at `skip` seconds, subtract it from - # all marker timestamps so they line up with the trimmed timeline. - if narration_skip > 0: + # all marker timestamps so they line up with the trimmed timeline. Skipped for + # override timings, which are already expressed in the final timeline. + if narration_skip > 0 and marker_timings_override is None: for timing in marker_timings: if timing.timestamp >= 0: timing.timestamp = max(0.0, timing.timestamp - narration_skip)