fix: report best-effort duration targets and shortfalls

This commit is contained in:
Affaan Mustafa
2026-09-07 14:17:02 +03:00
parent 1047b156c6
commit 262feffe1f
5 changed files with 110 additions and 4 deletions
+9
View File
@@ -29,6 +29,15 @@ python scripts/pipeline.py --genre example --root stylepacks \
--out out/review-v1.mp4
```
`--duration` is a **best-effort cadence target**, not an exact runtime. Complete
shots may produce a shorter or longer edit; the assembler does not duplicate
clips or add padding to meet the target. The output manifest retains actual
`duration` and adds `duration_contract` with requested and actual seconds,
shortfall, overrun, and the `cadence_target` policy. Differences of at least
one output frame are warned explicitly. No exact-duration mode is provided;
when an exact runtime is required, inspect the receipt and revise or reject
the cut before delivery.
The pipeline keeps each run's graded shot files because the editable FCPXML
and EDL reference them. It refuses output collisions and reports timeline
export failures. A completed render is not a saved editor project or creative
+1 -1
View File
@@ -566,7 +566,7 @@ def main() -> None:
help="WHAT HAPPENS: subject and action, e.g. 'a courier weaves "
"through night traffic'")
ap.add_argument("--duration", type=float, default=20.0,
help="target total seconds; shot lengths are drawn from the pack's cadence")
help="best-effort target seconds, not exact; shot lengths follow the pack's cadence and actual assembly duration is reported")
ap.add_argument("--base-video",
help="optional existing footage; each shot is conditioned on the frame "
"at its own timecode so generated shots supplement the edit")
+17 -2
View File
@@ -240,12 +240,25 @@ def forge(
break
kept.append(c); acc += d
if kept:
print(f" trimmed {len(order)} -> {len(kept)} shots to hit {duration:.1f}s")
print(f" trimmed {len(order)} -> {len(kept)} shots toward target {duration:.1f}s")
order = kept
out_path = Path(out)
asm.concat(order, out_path, fps=FPS)
final = frame_mod.probe(out_path)
duration_delta = final.duration - duration if duration is not None else 0.0
duration_contract = {
"policy": "cadence_target",
"requested_seconds": duration,
"actual_seconds": round(final.duration, 6),
"shortfall_seconds": round(max(0.0, -duration_delta), 6),
"overrun_seconds": round(max(0.0, duration_delta), 6),
}
if duration is not None and abs(duration_delta) + 1e-9 >= 1.0 / FPS:
print(f" WARNING: cadence target {duration:.3f}s produced {final.duration:.3f}s "
f"(shortfall {duration_contract['shortfall_seconds']:.3f}s, "
f"overrun {duration_contract['overrun_seconds']:.3f}s); "
"whole cadence shots are preserved without padding or duplication")
# Ship an EDITABLE timeline beside the flattened mp4.
#
@@ -288,6 +301,7 @@ def forge(
"generated_shots": len(gen_shots),
"base_shots": len(base_shots),
"duration": round(final.duration, 3),
"duration_contract": duration_contract,
"grade_strength": strength,
"timelines": timelines,
"shot_files": [str(Path(c).resolve()) for c in order],
@@ -311,7 +325,8 @@ def main() -> None:
ap.add_argument("--overlays", nargs="*", default=None, help="overlay image paths")
ap.add_argument("--overlay-every", type=int, default=4)
ap.add_argument("--overlay-opacity", type=float, default=0.3)
ap.add_argument("--duration", type=float, default=None)
ap.add_argument("--duration", type=float, default=None,
help="best-effort cadence target in seconds, not an exact output duration")
ap.add_argument("--strength", type=float, default=1.0, help="0-1 grade intensity")
ap.add_argument("--width", type=int, default=None)
ap.add_argument("--take-len", type=float, default=5.0)
+2 -1
View File
@@ -58,7 +58,8 @@ def main() -> None:
ap.add_argument("--fps", type=float, default=None, help="output frame rate")
ap.add_argument("--brief", default="", help="WHAT HAPPENS in the new piece")
ap.add_argument("--style-steer", default="", help="HOW IT LOOKS, per-run nudge")
ap.add_argument("--duration", type=float, default=12.0)
ap.add_argument("--duration", type=float, default=12.0,
help="best-effort cadence target in seconds, not an exact duration; actual result is reported")
ap.add_argument("--base-video", default=None, help="existing footage to supplement")
ap.add_argument("--base-ratio", type=float, default=0.35)
ap.add_argument("--out", default=None)
+81
View File
@@ -1,11 +1,13 @@
"""Requested image overlays must fail closed if compositing fails."""
import importlib.util
import io
import shutil
import subprocess
import sys
import tempfile
import unittest
from contextlib import redirect_stdout
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
@@ -174,5 +176,84 @@ class OverlayFailureTests(unittest.TestCase):
self.assertEqual(plate.read_bytes(), b"original image")
class DurationContractTests(unittest.TestCase):
def test_cadence_target_records_actual_duration_and_warns_on_frame_difference(self):
for requested, shortfall, overrun, warning in (
(2.0, 0.7, 0.0, True),
(1.3, 0.0, 0.0, False),
(1.3 + 1 / 30, 0.033333, 0.0, True),
(1.31, 0.01, 0.0, False),
(1.0, 0.0, 0.3, True),
(None, 0.0, 0.0, False),
):
with (
self.subTest(requested=requested),
tempfile.TemporaryDirectory() as directory,
):
root = Path(directory)
take, out = root / "take.mp4", root / "out.mp4"
take.write_bytes(b"original")
info = SimpleNamespace(width=320, height=180, fps=30, duration=1.3)
stats = SimpleNamespace(contrast=1, black_point=0, white_point=1)
stdout = io.StringIO()
def timeline(*args, **kwargs):
path = kwargs["out_path"]
path.touch()
return path
with (
patch.object(
forge.pack_mod,
"load",
return_value=SimpleNamespace(
grade_path="grade", cadence_path="cadence"
),
),
patch.object(forge.grade_mod, "load_stats", return_value=stats),
patch.object(
forge.cad_mod,
"load",
return_value=SimpleNamespace(
mean_shot=1, cuts_per_min=60, rhythm_variance=0
),
),
patch.object(forge.frame_mod, "probe", return_value=info),
patch.object(forge.asm, "normalize", return_value=take),
patch.object(forge.grade_mod, "grade_clip_direct"),
patch.object(forge.asm, "cut_take", return_value=[take]),
patch.object(forge.asm, "concat"),
patch.object(forge.tl_mod, "write_timeline", side_effect=timeline),
patch.object(forge.asm, "write_manifest") as manifest,
redirect_stdout(stdout),
):
forge.forge(
"look",
[str(take)],
str(out),
duration=requested,
work=str(root / "work"),
fps=30,
plan=[{"shots": [{"start": 0, "duration": 1.3}]}],
)
receipt = manifest.call_args.args[1]
self.assertEqual(receipt["duration"], 1.3)
self.assertEqual(
receipt["duration_contract"],
{
"policy": "cadence_target",
"requested_seconds": requested,
"actual_seconds": 1.3,
"shortfall_seconds": shortfall,
"overrun_seconds": overrun,
},
)
self.assertEqual(
"WARNING: cadence target" in stdout.getvalue(), warning
)
self.assertNotIn("to hit", stdout.getvalue())
self.assertEqual(take.read_bytes(), b"original")
if __name__ == "__main__":
unittest.main()