Faster blended stitch, clip sound matched to the narration, and the score's choice during clips

backend/app/assistant/prompt.py

@@ -42,13 +42,15 @@
   - `card`: a title or chapter screen: `{"card":{"variant":"title"|"chapter","title":...,\
 "subtitle":...}}`. May have narration too.
   - `clip`: a video played with its own sound and no narration: `visual` with `trim` and \
-`"audio":"play"`.
+`"audio":"play"`. Its sound is matched to the narration's loudness automatically; \
+`visual.volume_db` (-30 to 20) moves it from there, only when asked.
   - Optional: `chapter` (a chapter mark at its start), `transition_in`, `duration` \
 (`min` or `fixed` seconds), `overlays`, `role` (title / intro / outro).
 - `audio_tracks`: music, sound effects or extra voice, each from a `start` anchor to an \
 `end` anchor (omit `end` to play once to its natural end), with `volume` (0-1; music \
 usually 0.12-0.2 under narration), `fade_in` / `fade_out` seconds, `duck_under_narration`, \
-`loop`.
+`loop`, and `during_clips`: omit it to treat clips like narration, `"mute"` to fade the \
+music out while a clip plays, `"play"` to keep it playing at its level.
 
 ## Time is anchored to structure, not seconds
 An anchor is `"video_start"`, `"video_end"`, or `{"segment":id,"beat":id?,"edge":"start"|\

backend/app/render/compile.py

@@ -68,6 +68,7 @@ class Unit:
     # Seconds into the file and how long, for a video; None plays it whole.
     trim: Optional[Tuple[float, float]] = None
     loop: bool = True
+    volume_db: float = 0.0            # a clip's own sound, on top of matching the narration
     panel: Optional[Panel] = None
     motion: Optional[Motion] = None   # resolved, `alternate` already cycled
     fit: str = "blur"
@@ -173,7 +174,8 @@ def plan_units(spec: VideoSpec, media: Dict[str, Media], warnings: List[str]) ->
                 warnings.append(f"Clip {segment.label or segment.id} has no video to play; left out")
                 continue
             units.append(Unit(**common, first_in_segment=True, kind="video_sound", media=found,
-                              trim=trim, loop=False, chapter=segment.chapter))
+                              trim=trim, loop=False, chapter=segment.chapter,
+                              volume_db=(visual.volume_db or 0.0) if visual else 0.0))
             continue
 
         if found and found.kind == "video":
@@ -399,6 +401,7 @@ class TrackPlacement:
     loop: bool
     duck: bool
     trim_in: float
+    during_clips: Optional[str] = None
 
 
 def place_tracks(spec: VideoSpec, media: Dict[str, Media], where: Placement,
@@ -441,7 +444,7 @@ def place_tracks(spec: VideoSpec, media: Dict[str, Media], where: Placement,
             track=track, media=found, start=start, length=length, loop=loop,
             duck=(track.duck_under_narration if track.duck_under_narration is not None
                   else track.kind == "music"),
-            trim_in=trim_in,
+            trim_in=trim_in, during_clips=track.during_clips,
         ))
     return placed
 

backend/app/render/renderer.py

@@ -225,6 +225,86 @@ def _stitch_blended(
         F.run(F.build_concat_command("stitch.txt", output), cwd=work_dir, timeout=1800)
 
 
+def _stitch_partial(
+    shot_files: List[str], durations: List[float], joins: list, output: str,
+    *, options: RenderOptions, preview: bool, work_dir: str,
+) -> None:
+    """Encode the picture again only around each blend, and copy the rest of it.
+
+    See `plan_partial_stitch`. The sound is built for the whole video in one pass,
+    exactly as the full stitch builds it, and muxed back in, so it is sample-exact
+    and never cut at a packet boundary.
+    """
+    fps = options.fps
+    probed = [F.probe_keyframes(os.path.join(work_dir, name), fps) for name in shot_files]
+    keyframes, counts = [k for k, _ in probed], [c for _, c in probed]
+    pieces = F.plan_partial_stitch(durations, joins, keyframes, fps)
+    # Where each shot starts on the timeline, which is where the sound puts it.
+    starts = [0.0]
+    for index in range(1, len(durations)):
+        join = joins[index] if index < len(joins) else None
+        starts.append(starts[-1] + durations[index - 1] - (join[1] if join else 0.0))
+    emitted = 0   # frames of picture laid down so far
+
+    cuts: Dict[int, List[int]] = {}
+    for piece in pieces:
+        if not piece.blend:
+            at = cuts.setdefault(piece.first, [])
+            at += [f for f in (piece.start_frame or None, piece.end_frame) if f is not None]
+    for shot, frames in cuts.items():
+        if frames:
+            cuts[shot] = sorted(set(frames))
+            F.run(F.build_cut_command(shot_files[shot], f"cut_{shot:04d}_%03d.mp4",
+                                      cuts[shot], fps), cwd=work_dir, timeout=600)
+
+    names: List[str] = []
+    for number, piece in enumerate(pieces):
+        if not piece.blend:
+            frames = cuts.get(piece.first) or []
+            if not frames:
+                name = f"cut_{piece.first:04d}_000.mp4"
+                F.run(F.build_video_copy_command(shot_files[piece.first], name),
+                      cwd=work_dir, timeout=600)
+                names.append(name)
+            else:
+                names.append(f"cut_{piece.first:04d}_"
+                             f"{sum(1 for f in frames if f <= piece.start_frame):03d}.mp4")
+            end = counts[piece.first] if piece.end_frame is None else piece.end_frame
+            emitted += end - piece.start_frame
+            continue
+        span = slice(piece.first, piece.last + 1)
+        sub_joins = [None] + list(joins[piece.first + 1:piece.last + 1])
+        # A run of short shots blended into each other is one blend piece, and as
+        # long as the video in the worst case: it is chunked like the full stitch.
+        chunks = F.plan_stitch_chunks(durations[span], sub_joins, fps)
+        for part, chunk in enumerate(chunks):
+            start = piece.start_frame if part == 0 else chunk.start_frame
+            end = piece.end_frame if part == len(chunks) - 1 else chunk.end_frame
+            first, last = piece.first + chunk.first, piece.first + chunk.last
+            # The copied stretches are whole frames, while the timeline is not: a
+            # shot planned at 13.242 s has 397 frames. Left alone, each blend would
+            # start up to a frame late and the error would add up along the video,
+            # where the full stitch measures every blend from the timeline. So the
+            # first shot's length here is what puts its blend back on the timeline.
+            lengths = list(durations[first:last + 1])
+            lengths[0] += (starts[first] + start / fps) - emitted / fps
+            name = f"blend_{number:03d}_{part:02d}.mp4"
+            F.run(F.build_stitch_command(
+                shot_files[first:last + 1], lengths, name,
+                joins=[None] + list(joins[first + 1:last + 1]), options=options,
+                preview=preview, start_frame=start, end_frame=end, streams="video"),
+                cwd=work_dir, timeout=F.shot_timeout(sum(durations[first:last + 1])))
+            names.append(name)
+            emitted += F.probe_keyframes(os.path.join(work_dir, name), fps)[1]
+
+    F.run(F.build_stitch_command(shot_files, durations, "stitch_audio.m4a", joins=joins,
+                                 options=options, preview=preview, streams="audio"),
+          cwd=work_dir, timeout=1800)
+    F.write_concat_list(os.path.join(work_dir, "stitch_video.txt"), names)
+    F.run(F.build_join_av_command("stitch_video.txt", "stitch_audio.m4a", output),
+          cwd=work_dir, timeout=1800)
+
+
 def render(
     *, spec: VideoSpec, preview: bool, work_dir: str, out_dir: str, data_dir: str, user_id: str,
     gam: GamClient, speech: NotesSpeech, progress: Progress,
@@ -280,6 +360,19 @@ def render(
                  for u, n in zip(units, narrated)]
     timeline = layout(units, durations, warnings)
 
+    # ── how loud the voice is, for clips to be matched to (S5) ─────────────────
+    # Measured, not assumed: voices and providers differ by several dB. R128 gates
+    # out the pauses, so the holds between lines don't pull it down.
+    narration_lufs = F.DEFAULT_LOUDNESS_LUFS
+    if any(u.kind == "video_sound" for u in units):
+        levels = []
+        for narration in narrated:
+            if narration.path and narration.duration > 0:
+                lufs = F.measure_loudness(os.path.join(work_dir, narration.path))
+                if lufs is not None:
+                    levels.append((lufs, narration.duration))
+        narration_lufs = F.combined_loudness(levels) or F.DEFAULT_LOUDNESS_LUFS
+
     # ── encode each unit ─────────────────────────────────────────────────────
     cache = shot_cache.ShotCache(data_dir, user_id)
     try:
@@ -341,6 +434,15 @@ def render(
                       cwd=work_dir, timeout=F.shot_timeout(length))
                 background = clip
 
+        # A clip's sound at the voice's loudness, then moved by its own setting:
+        # footage arrives anywhere from a whisper to a broadcast master. Measured
+        # over what plays, so a loud intro the trim cuts off doesn't count.
+        gain = 0.0
+        if unit.kind == "video_sound" and has_audio:
+            gain = round(F.clip_gain(F.measure_loudness(os.path.join(work_dir, background),
+                                                        duration),
+                                     narration_lufs, unit.volume_db), 1)
+
         # Watermark, overlay text and a quote or code panel, as one layer.
         if unit.chapter:
             current_chapter = unit.chapter
@@ -390,7 +492,7 @@ def argv(with_options: RenderOptions, with_path=path) -> List[str]:
                 duration=duration, output=output, options=with_options, preview=preview,
                 overlay_png=overlay_png, subtitle_file=shot_srt, background_has_audio=has_audio,
                 index=index, in_transition=in_transition, out_transition=out_transition,
-                loop=unit.loop, path=with_path,
+                loop=unit.loop, path=with_path, audio_gain_db=gain,
             )
 
         if narration.path is None and unit.kind != "video_sound":
@@ -424,8 +526,15 @@ def argv(with_options: RenderOptions, with_path=path) -> List[str]:
     stitched = "stitched.mp4"
     if any(blends):
         try:
-            _stitch_blended(shot_files, timeline.durations, blends, stitched,
-                            options=options, preview=preview, work_dir=work_dir)
+            try:
+                _stitch_partial(shot_files, timeline.durations, blends, stitched,
+                                options=options, preview=preview, work_dir=work_dir)
+            except F.FFmpegError as exc:
+                # The same blends with every frame encoded again: several times
+                # slower at full quality, but it relies on no keyframes or remuxing.
+                logger.warning("Partial stitch failed (%s); encoding it whole", exc)
+                _stitch_blended(shot_files, timeline.durations, blends, stitched,
+                                options=options, preview=preview, work_dir=work_dir)
         except F.FFmpegError as exc:
             # The blends cost a re-encode; if that fails, the shots still join by cuts,
             # and the timeline is re-laid without their overlaps to match. A clip's
@@ -460,9 +569,14 @@ def argv(with_options: RenderOptions, with_path=path) -> List[str]:
         progress("Mixing", 90, f"{len(tracks)} audio track{'s' if len(tracks) != 1 else ''}")
         mix = [F.MixTrack(path=t.media.path, start=t.start, length=t.length, trim_in=t.trim_in,
                           loop=t.loop, volume=t.track.volume, fade_in=t.track.fade_in,
-                          fade_out=t.track.fade_out, duck=t.duck) for t in tracks]
+                          fade_out=t.track.fade_out, duck=t.duck,
+                          during_clips=t.during_clips) for t in tracks]
+        # Where a clip plays its own sound, for a track that leaves for clips (S5).
+        clip_windows = [(timeline.starts[i], timeline.end(i)) for i, u in enumerate(units)
+                        if u.kind == "video_sound" and u.media and u.media.has_audio]
         F.run(F.build_mix_command(final, mix, "scored.mp4", duration=timeline.total,
-                                  duck_available="sidechaincompress" in F.available_filters()),
+                                  duck_available="sidechaincompress" in F.available_filters(),
+                                  clip_windows=clip_windows),
               cwd=work_dir, timeout=1800)
         final = "scored.mp4"
 

backend/app/spec/model.py

@@ -397,6 +397,9 @@ class Visual(_Model):
     trim: Optional[Trim] = None               # videos only
     audio: Literal["mute", "play"] = "mute"   # "play" only on a clip
     loop: bool = True                         # shorter than its segment: loop, or hold
+    # A clip's sound is matched to the narration's loudness by the renderer; this
+    # moves it from there, in dB (S5). Ignored where the footage plays muted.
+    volume_db: Optional[float] = Field(default=None, ge=-30, le=20)
 
 
 class Beat(_Model):
@@ -519,6 +522,10 @@ class AudioTrack(_Model):
     loop: Optional[bool] = None         # default: on for music
     volume: Unit = 0.18
     duck_under_narration: Optional[bool] = None   # default: on for music
+    # While a clip plays its own sound (S5). None: as under narration (a ducked
+    # track dips, one that isn't plays on). "mute" fades it out for the clip and
+    # back in after; "play" keeps it at its level, ducked under narration only.
+    during_clips: Optional[Literal["mute", "play"]] = None
     fade_in: float = Field(default=0.0, ge=0, le=15)
     fade_out: float = Field(default=0.0, ge=0, le=15)
 

backend/tests/test_render_compile.py

@@ -357,3 +357,32 @@ def test_the_overlay_pass_draws_each_layer_only_in_its_window_and_copies_the_sou
     assert "[l2]" in graph and "fade" not in graph.split("[1:v]")[0]
     assert argv[argv.index("-c:a") + 1] == "copy"
     assert argv.count("-c:a") == 1 and argv[argv.index("-map") + 1] == "[v2]"
+
+
+# ── clip sound levels and the score during clips (S5) ───────────────────────
+
+def test_a_clips_own_level_and_a_tracks_choice_during_clips_reach_the_render():
+    spec = spec_of({
+        "assets": {"v": {"from": "gam", "id": "g1", "type": "video"},
+                   "m": {"from": "gam", "id": "g2", "type": "audio"}},
+        "segments": [
+            {"id": "a", "kind": "clip", "visual": {"asset": "v", "audio": "play",
+                                                    "volume_db": -4.5}},
+            {"id": "b", "kind": "clip", "visual": {"asset": "v", "audio": "play"}},
+        ],
+        "audio_tracks": [{"id": "t", "asset": "m", "start": "video_start",
+                          "end": "video_end", "during_clips": "mute"}]})
+    units = plan_units(spec, {"v": VIDEO, "m": SONG}, [])
+    assert [u.volume_db for u in units] == [-4.5, 0.0]
+    timeline = layout(units, [6.0, 6.0], [])
+    where = placement(units, timeline, [{} for _ in units])
+    [track] = place_tracks(spec, {"v": VIDEO, "m": SONG}, where, [])
+    assert track.during_clips == "mute"
+
+
+def test_a_clip_level_outside_its_range_is_refused():
+    _spec, issues = parse_spec({
+        "assets": {"v": {"from": "gam", "id": "g1", "type": "video"}},
+        "segments": [{"id": "a", "kind": "clip",
+                      "visual": {"asset": "v", "audio": "play", "volume_db": 40}}]})
+    assert _spec is None and issues