gitprov/gnommo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Better build

Browse Source
This commit is contained in:
gitprov
2026-07-25 13:24:00 +02:00
parent 7312fa2366
commit a91acca695
2 changed files with 163 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gnommo/cli.py
+40 -15
View File
@@ -3679,6 +3679,12 @@ def _print_render_plan_details(plan, marker_timings, slides: dict) -> None:
camera_events_by_time[t] = []
camera_events_by_time[t].append(event)
# Output-time position for every slide — i.e. when it actually appears in the
# finished video, WITH any narration pauses already added. Both aligned and
# interpolated slides print from this so the plan reads in one consistent
# (output) time base and matches the rendered result.
_slide_out = {e.slide_id: e.start_time for e in plan.slide_events}
# Detect slide markers that share a timestamp with the adjacent slide marker.
# Two slides at the same time means alignment is ambiguous — treat as an error.
slide_timings = [
@@ -3709,16 +3715,18 @@ def _print_render_plan_details(plan, marker_timings, slides: dict) -> None:
if timing.confidence < 1.0:
conf_str = f" ({timing.confidence:.0%})"
# Determine marker type for display
# Determine marker type for display. Slides print their FINAL (output)
# time — when the viewer sees them, pauses included — from the plan.
if marker_id in slides:
_st = _format_time(_slide_out.get(marker_id, timing.timestamp))
if marker_id in collision_ids:
collision_count += 1
print(
f' {marker_id:6} {time_str}{conf_str} COLLISION - same time as adjacent slide - "{context}"'
f' {marker_id:6} {_st}{conf_str} COLLISION - same time as adjacent slide - "{context}"'
)
else:
aligned_count += 1
print(f' {marker_id:6} {time_str}{conf_str} "{context}"')
print(f' {marker_id:6} {_st}{conf_str} "{context}"')
elif any(
marker_id.startswith(p)
for p in (
@@ -3786,7 +3794,8 @@ def _print_render_plan_details(plan, marker_timings, slides: dict) -> None:
print(f" {time_str} [{marker_id}]")
else:
aligned_count += 1
print(f' {marker_id:6} {time_str} "{context}"')
_st = _format_time(_slide_out.get(marker_id, timing.timestamp))
print(f' {marker_id:6} {_st} "{context}"')
else:
unaligned_count += 1
# Check if this is a slide that was interpolated into the plan
@@ -3795,7 +3804,9 @@ def _print_render_plan_details(plan, marker_timings, slides: dict) -> None:
(e for e in plan.slide_events if e.slide_id == marker_id), None
)
if interp_event:
interp_str = _format_time(interp_event.start_time)
interp_str = _format_time(
_slide_out.get(marker_id, interp_event.start_time)
)
print(f' {marker_id:6} ~{interp_str} INTERPOLATED - "{context}"')
else:
print(f' {marker_id:6} ??:??.?? NOT ALIGNED - "{context}"')
@@ -4778,8 +4789,8 @@ def _cmd_render_impl(
if plan.narration_segments:
print(f" Narration concat: {len(plan.narration_segments)} segment(s) at render time")
# Print detailed render plan with alignment info
_print_render_plan_details(plan, marker_timings, slides)
# (The detailed render plan is printed by the build/timing-layer block below,
# so the computation and its printout live together — not in the render stage.)
if plan.audio_events:
print(f"\n Audio effects:")
for event in plan.audio_events:
@@ -4830,10 +4841,23 @@ def _cmd_render_impl(
# 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)
# merge restores each id's human `adjustment` (the relative tweak).
_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.
if _existing_events:
# Override path: the marker_timings the plan used were EFFECTIVE
# narration (narration_time + adjustment). narration_time itself is
# owned by events.json — restore it so the effective value doesn't leak
# into the stored narration_time.
_by_id = {e["id"]: e for e in _existing_events}
for e in _events:
_old = _by_id.get(e["id"])
if _old is not None and _old.get("narration_time") is not None:
e["narration_time"] = _old["narration_time"]
else:
# Fresh align: recompute interpolated narration times between anchors.
_scaffold.reinterpolate_events(_events)
# final_time = (narration + adjustment) shifted by preceding pauses.
_scaffold.compute_final_times(_events)
_scaffold.write_events(project_path, _events)
_scaffold.write_scaffold(
project_path, _events, transcription, narration_schedule, plan.total_duration
@@ -4843,16 +4867,17 @@ def _cmd_render_impl(
_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}"
f"\n Timing layer: {_summ['exact']} exact, {_summ['interpolated']} interpolated "
f"→ {_scaffold.EVENTS_FILE} + {_scaffold.SCAFFOLD_FILE} + {_scaffold.TRANSCRIBED_FILE}"
)
# The full render plan is printed here at build time — render just executes it.
_print_render_plan_details(plan, marker_timings, slides)
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."
f" {_summ['interpolated']} event(s) interpolated. Nudge them in "
f"{_scaffold.EVENTS_FILE} via the \"adjustment\" field (seconds); it's a "
f"relative tweak that survives re-alignment."
)
print(
f" Compare {_scaffold.TRANSCRIBED_FILE} against manuscript.txt to see the drift, "
gnommo/scaffold.py
+117 -63
View File
@@ -89,6 +89,27 @@ def marker_type(marker_id: str, slides: dict, videos: dict, audio: dict) -> str:
return "other"
# Pause-variant video marker prefixes (mirrors transformer). These freeze the
# narration, so they're what makes narration_time and final_time diverge.
_PAUSE_MARKER_PREFIXES = (
"vftp:", "vfbp:", "vfmp:", "vf2tp:", "vf2bp:", "vf2mp:",
"vstp:", "vsbp:", "vsmp:",
)
def _pause_duration(marker_id: str, videos: dict) -> float:
"""Seconds a pause-variant video marker freezes the narration for, else 0."""
if not marker_id.startswith(_PAUSE_MARKER_PREFIXES):
return 0.0
handle = marker_id.split(":", 1)[1].lower() if ":" in marker_id else marker_id.lower()
vs = None
if videos:
vs = videos.get(handle) or next(
(v for k, v in videos.items() if k.lower() == handle), None
)
return float(getattr(vs, "pause_narration", 0.0) or 0.0) if vs else 0.0
# ── derive events from aligner output ─────────────────────────────────────────
def derive_events(
@@ -97,129 +118,157 @@ def derive_events(
videos: dict,
audio: dict,
) -> list[dict]:
"""Turn ordered MarkerTiming objects into event dicts with a mapping tag.
"""Turn ordered MarkerTiming objects into three-clock event dicts.
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.
Each event carries two clocks plus a tweak:
narration_time — aligner-owned position on the narration/transcript timeline
(interpolated for markers Whisper couldn't place). Read-only.
adjustment — a human tweak in seconds layered on narration_time. 0 here;
preserved across rebuilds by merge_events. The ONLY editable knob.
final_time — filled by compute_final_times: (narration_time + adjustment)
shifted forward by every pause before it. What the viewer sees.
Pause-narration videos also carry pause_duration so the offset is self-describing.
"""
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(
{
mapping = (
MAPPING_EXACT
if (placed and t.confidence >= _EXACT_THRESHOLD and not after_prev)
else MAPPING_INTERPOLATED
)
e = {
"type": marker_type(t.marker_id, slides, videos, audio),
"id": t.marker_id,
"time": round(t.timestamp, 3) if placed else None,
"narration_time": round(t.timestamp, 3) if placed else None,
"adjustment": 0.0,
"final_time": None,
"mapping": mapping,
"confidence": round(t.confidence, 3),
"context": (t.context or "")[:80],
}
)
pd = _pause_duration(t.marker_id, videos)
if pd:
e["pause_duration"] = pd
events.append(e)
_interpolate_missing_times(events)
_interpolate_narration(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.
"""
def _interpolate_narration(events: list[dict]) -> None:
"""Fill `narration_time: None` entries by linear interpolation between placed
neighbours. Head/tail runs 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:
if events[i]["narration_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:
while j < n and events[j]["narration_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
prev_time = events[i - 1]["narration_time"] if i > 0 else None
next_time = events[j]["narration_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)
events[i + k]["narration_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)
events[i + k]["narration_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)
events[i + k]["narration_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)
events[i + k]["narration_time"] = round(1.0 * (i + k), 3)
i = j
# ── final_time = narration + adjustment + pause offsets ───────────────────────
def compute_final_times(events: list[dict]) -> None:
"""Fill each event's final_time from its (narration_time + adjustment), shifted
forward by every pause whose own (narration + adjustment) is at or before it.
This is the single, clean definition of the narration→final mapping; the render
is fed the same effective narration times so its output matches these values.
Mutates events in place.
"""
def eff(e):
n = e.get("narration_time")
return None if n is None else n + e.get("adjustment", 0.0)
pauses = [
(eff(e), e.get("pause_duration", 0.0), id(e))
for e in events
if e.get("pause_duration") and eff(e) is not None
]
for e in events:
ee = eff(e)
if ee is None:
e["final_time"] = None
continue
shift = sum(dur for (pn, dur, oid) in pauses if pn <= ee and oid != id(e))
e["final_time"] = round(ee + shift, 3)
# ── re-interpolation on rebuild ───────────────────────────────────────────────
def reinterpolate_events(events: list[dict]) -> None:
"""Recompute every `interpolated` event's time between its anchor neighbours.
"""Recompute every `interpolated` event's narration_time between its anchors.
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.
Anchors are the events the aligner placed (mapping `exact`). Interpolated events
carry no real spoken cue, so their narration_time is only ever a guess spread
between anchors — recomputed each build. Adjustments are untouched (they layer
on top). Mutates events in place; call compute_final_times afterwards.
"""
for e in events:
if e.get("mapping") == MAPPING_INTERPOLATED:
e["time"] = None
_interpolate_missing_times(events)
e["narration_time"] = None
_interpolate_narration(events)
# ── merge: hand-edits win ─────────────────────────────────────────────────────
# ── merge: human tweaks survive re-align ──────────────────────────────────────
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`).
"""Carry each id's `adjustment` (the human tweak) from the existing events.json
onto the freshly-derived events. narration_time and mapping come from `fresh`
(aligner-owned); the tweak is relative so it stays meaningful even when the
alignment shifts. New ids are added; ids gone from the manuscript are dropped.
"""
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
if old is not None and old.get("adjustment"):
e["adjustment"] = old["adjustment"]
return fresh
# ── events <-> marker timings ─────────────────────────────────────────────────
def events_to_marker_timings(events: list[dict]) -> list[MarkerTiming]:
"""Reconstruct MarkerTiming objects from events for build_render_plan override."""
"""Reconstruct MarkerTiming objects for build_render_plan override.
Emits the EFFECTIVE narration time (narration_time + adjustment). The render's
pause pass then shifts these into final time, reproducing each event's stored
final_time exactly — so events.json is the single source of the render's timing.
"""
timings: list[MarkerTiming] = []
for e in events:
time = e.get("time")
n = e.get("narration_time")
eff = (n + e.get("adjustment", 0.0)) if n is not None else -1.0
timings.append(
MarkerTiming(
marker_id=e["id"],
timestamp=float(time) if time is not None else -1.0,
timestamp=eff,
context=e.get("context", ""),
confidence=float(e.get("confidence", 1.0)),
)
@@ -322,9 +371,15 @@ def build_transcribed_manuscript(events: list[dict], transcription: list) -> str
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.
"""
# The transcript is the NARRATION timeline, so place markers by their effective
# narration time (narration_time + adjustment), not final_time.
def _eff(e):
nt = e.get("narration_time")
return None if nt is None else nt + e.get("adjustment", 0.0)
evs = sorted(
[e for e in events if e.get("time") is not None],
key=lambda e: e["time"],
[(_eff(e), e) for e in events if _eff(e) is not None],
key=lambda pair: pair[0],
)
words = transcription or []
n = len(words)
@@ -342,8 +397,7 @@ def build_transcribed_manuscript(events: list[dict], transcription: list) -> str
buf.clear()
wi = 0
for e in evs:
t = e["time"]
for t, e in evs:
while wi < n and _w_start(words[wi]) < t:
buf.append(_w_word(words[wi]))
wi += 1