feat(audio): add independent track mixing and transcript timing

SKILL.md

@@ -71,6 +71,10 @@ Helpers (`helpers/transcribe.py`, `helpers/render.py`, etc.) live alongside this
 
 ## Helpers
 
+- **`mix_audio.py`, `map_transcript.py`** — mix independent audio tracks and map intact words onto the sample clock. See [audio mixing](references/audio-mixing.md).
+
+- **`source_scan.py`, `prepare_source.py`, `find_shot.py`, `project_state.py`** — inspect selected sources and retain provenance. See [source inspection](references/sources.md).
+
 - **`transcribe.py <video>`** — single-file Scribe call. `--num-speakers N` optional. Cached.
 - **`transcribe_batch.py <videos_dir>`** — 4-worker parallel transcription. Use for multi-take.
 - **`pack_transcripts.py --edit-dir <dir>`** — `transcripts/*.json` → `takes_packed.md` (phrase-level, break on silence ≥ 0.5s).

helpers/edit_clock.py

@@ -0,0 +1,104 @@
+"""One rational frame clock and one 48 kHz sample clock. Intervals are half-open."""
+
+from fractions import Fraction
+import math
+
+SAMPLE_RATE = 48000
+
+
+# require a finite positive clock rate before conversion
+def clock_rate(value):
+    if isinstance(value, bool):
+        raise ValueError("clock rate must be finite and positive")
+    try:
+        result = fraction(value)
+    except (ValueError, ZeroDivisionError, OverflowError) as exc:
+        raise ValueError("clock rate must be finite and positive") from exc
+    if result <= 0:
+        raise ValueError("clock rate must be finite and positive")
+    return result
+
+
+# convert decimal or rational input without introducing binary float rounding
+def fraction(value):
+    """Convert decimal or rational input without introducing binary float rounding."""
+    return value if isinstance(value, Fraction) else Fraction(str(value))
+
+
+# round a nonnegative rational value halfway upward
+def nearest(value):
+    """Round nonnegative rational values half up; avoid Python's banker rounding."""
+    value = fraction(value)
+    return (2 * value.numerator + value.denominator) // (2 * value.denominator)
+
+
+# convert seconds to a frame boundary using the requested rounding policy
+def seconds_to_frame(seconds, fps=30, mode="nearest"):
+    """Convert seconds to a frame boundary using the requested rounding policy."""
+    value = fraction(seconds) * clock_rate(fps)
+    if mode == "nearest":
+        return nearest(value)
+    if mode == "ceil":
+        return math.ceil(value)
+    if mode == "floor":
+        return math.floor(value)
+    raise ValueError("mode must be nearest, ceil, or floor")
+
+
+# map a frame boundary onto the audio sample clock
+def frame_to_sample(frame, fps=30, rate=SAMPLE_RATE):
+    """Map a frame boundary onto the audio sample clock."""
+    return nearest(fraction(frame) * clock_rate(rate) / clock_rate(fps))
+
+
+# map seconds onto the audio sample clock
+def seconds_to_sample(seconds, rate=SAMPLE_RATE):
+    """Map seconds onto the audio sample clock."""
+    return nearest(fraction(seconds) * clock_rate(rate))
+
+
+# floor a frame boundary to the centisecond clock used by ASS subtitles
+def ass_stamp(frame, fps=30):
+    """Floor to the ASS clock so a 30 fps end boundary cannot leak one frame."""
+    cs = math.floor(fraction(frame) * 100 / clock_rate(fps))
+    return f"{cs//360000}:{cs//6000%60:02}:{cs//100%60:02}.{cs%100:02}"
+
+
+# distribute a fixed frame budget across positive shot weights
+def allocate_frames(total, weights):
+    """Largest-remainder allocation preserves total; it does not choose edit points."""
+    if type(total) is not int or not weights or total < len(weights) or any(fraction(w) <= 0 for w in weights):
+        raise ValueError("positive weights and at least one frame per shot required")
+    exact = [fraction(w) * total / sum(map(fraction, weights)) for w in weights]
+    counts = [math.floor(v) for v in exact]
+    for i in sorted(
+        range(len(weights)), key=lambda i: (exact[i] - counts[i], -i), reverse=True
+    )[: total - sum(counts)]:
+        counts[i] += 1
+    if min(counts) < 1:
+        extra = total - len(weights)
+        exact = [fraction(w) * extra / sum(map(fraction, weights)) for w in weights]
+        counts = [1 + math.floor(v) for v in exact]
+        for i in sorted(range(len(weights)), key=lambda i: (exact[i] % 1, -i), reverse=True)[:total - sum(counts)]:
+            counts[i] += 1
+    return counts
+
+
+# reject shot intervals with gaps overlaps or an incorrect total
+def check_partition(shots, total):
+    """Reject shot intervals with gaps overlaps or an incorrect total."""
+    cursor = 0
+    for shot in shots:
+        start, end = shot["start_frame"], shot["end_frame"]
+        if (
+            type(start) is not int
+            or type(end) is not int
+            or start != cursor
+            or end <= start
+        ):
+            raise ValueError(
+                f'non-contiguous or invalid shot: {shot.get("id", "unknown")}'
+            )
+        cursor = end
+    if cursor != total:
+        raise ValueError(f"shots cover {cursor} frames, expected {total}")

helpers/edit_io.py

@@ -0,0 +1,111 @@
+"""Media IO and explicit source validation shared by the editing helpers."""
+
+import hashlib
+import json
+import subprocess
+from pathlib import Path
+
+
+# run a media command and surface its error output when it fails
+def run(args, log=None):
+    """Run a media command and surface its error output when it fails."""
+    result = subprocess.run([str(x) for x in args], capture_output=True)
+    if log is not None:
+        Path(log).write_bytes(result.stderr)
+    if result.returncode:
+        raise RuntimeError(
+            f"{args[0]} failed ({result.returncode}):\n"
+            + result.stderr.decode(errors="replace")[-6000:]
+        )
+    return result
+
+
+# read stream metadata and optionally count decoded frames with ffprobe
+def probe(path, frames=False):
+    """Read stream metadata and optionally count decoded frames with ffprobe."""
+    return json.loads(
+        run(
+            [
+                "ffprobe",
+                "-v",
+                "error",
+                *(["-count_frames"] if frames else []),
+                "-show_streams",
+                "-show_format",
+                "-of",
+                "json",
+                path,
+            ]
+        ).stdout
+    )
+
+
+# fingerprint file contents in bounded memory
+def sha256(path):
+    """Fingerprint file contents in bounded memory."""
+    digest = hashlib.sha256()
+    with Path(path).open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+# read a project or evidence document from disk
+def load_json(path):
+    """Read a project or evidence document from disk."""
+    return json.loads(Path(path).read_text())
+
+
+# write readable JSON while rejecting nonfinite measurement values
+def save_json(path, data, *, exclusive=False):
+    """Write readable JSON while rejecting nonfinite measurement values."""
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    payload = json.dumps(data, indent=2, allow_nan=False) + "\n"
+    with path.open("x" if exclusive else "w") as stream:
+        stream.write(payload)
+
+
+# resolve an artifact path relative to its project directory
+def resolve(root, name):
+    """Resolve an artifact path relative to its project directory."""
+    path = Path(name)
+    return path.resolve() if path.is_absolute() else (Path(root) / path).resolve()
+
+
+# require a declared render source and reject reference-only file aliases
+def source_path(manifest, root, source_id):
+    """Require a declared render source and reject reference-only file aliases."""
+    source = manifest["sources"][source_id]
+    if source.get("study_only") or not source.get("provenance"):
+        raise ValueError(
+            f"{source_id}: render sources need provenance and cannot be study-only"
+        )
+    path = resolve(root, source["file"])
+    if not path.is_file():
+        raise FileNotFoundError(path)
+    blocked_paths = manifest.get("study_media", [])
+    if not isinstance(blocked_paths, list):
+        raise ValueError("study_media must be a list of paths")
+    for blocked in blocked_paths:
+        other = resolve(root, blocked)
+        if path == other or (other.exists() and path.samefile(other)):
+            raise ValueError("study media entered the render graph")
+    return path
+
+
+# recover the last decodable JSON object from command output
+def last_json(text):
+    """Recover the last decodable JSON object from command output."""
+    candidates = []
+    for start, char in enumerate(text):
+        if char != "{":
+            continue
+        try:
+            value, length = json.JSONDecoder().raw_decode(text[start:])
+            candidates.append((start + length, -start, value))
+        except json.JSONDecodeError:
+            pass
+    if not candidates:
+        raise ValueError("no JSON measurement in command output")
+    return max(candidates, key=lambda row: row[:2])[2]

helpers/find_shot.py

@@ -0,0 +1,100 @@
+"""Find visual correspondence candidates in independently acquired source footage."""
+
+import argparse
+import math
+from pathlib import Path
+import cv2
+import numpy as np
+from PIL import Image
+from source_scan import catalog, selected_frames
+from edit_io import save_json
+
+
+# score local image features that agree on one geometric transformation
+def correspondence(query, candidate):
+    """Score local image features that agree on one geometric transformation."""
+    detector = cv2.SIFT_create(nfeatures=1000)
+    qa, qd = detector.detectAndCompute(
+        cv2.cvtColor(np.asarray(query), cv2.COLOR_RGB2GRAY), None
+    )
+    ca, cd = detector.detectAndCompute(
+        cv2.cvtColor(np.asarray(candidate), cv2.COLOR_RGB2GRAY), None
+    )
+    if qd is None or cd is None or len(cd) < 2:
+        return {"inliers": 0, "matches": 0, "reprojection_px": None}
+    pairs = cv2.BFMatcher().knnMatch(qd, cd, k=2)
+    matches = [
+        p[0] for p in pairs if len(p) == 2 and p[0].distance < 0.72 * p[1].distance
+    ]
+    if len(matches) < 4:
+        return {"inliers": 0, "matches": len(matches), "reprojection_px": None}
+    a = np.float32([qa[m.queryIdx].pt for m in matches])
+    b = np.float32([ca[m.trainIdx].pt for m in matches])
+    matrix, mask = cv2.findHomography(a, b, cv2.RANSAC, 3)
+    if matrix is None or mask is None:
+        return {"inliers": 0, "matches": len(matches), "reprojection_px": None}
+    projected = cv2.perspectiveTransform(a[:, None, :], matrix)[:, 0, :]
+    valid = mask[:, 0].astype(bool)
+    if not valid.any():
+        return {"inliers": 0, "matches": len(matches), "reprojection_px": None}
+    return {
+        "inliers": int(valid.sum()),
+        "matches": len(matches),
+        "reprojection_px": float(
+            np.linalg.norm(projected[valid] - b[valid], axis=1).mean()
+        ),
+        "homography": matrix.tolist(),
+    }
+
+
+# rank sampled source frames against a query image for later visual review
+def search(query, source, every=1, limit=12):
+    """Rank sampled source frames against a query image for later visual review."""
+    if not math.isfinite(every) or every <= 0:
+        raise ValueError("sampling interval must be positive")
+    index = catalog(source)
+    selected = []
+    next_time = index["frames"][0]["pts"]
+    for row in index["frames"]:
+        if row["pts"] >= next_time:
+            selected.append(row["frame"])
+            next_time = row["pts"] + every
+    with Image.open(query) as im:
+        reference = im.convert("RGB")
+        reference.thumbnail((640, 640))
+    results = []
+    for frame, image in selected_frames(source, selected, 640):
+        results.append(
+            {
+                "frame": frame,
+                "pts": index["frames"][frame]["pts"],
+                **correspondence(reference, image),
+            }
+        )
+    return {
+        "source_sha256": index["sha256"],
+        "candidates": sorted(
+            results, key=lambda r: (-r["inliers"], r["reprojection_px"] if r["reprojection_px"] is not None else 1e9)
+        )[:limit],
+        "limit": "Visual candidates need native-frame and semantic review; coarse sampling cannot certify exact action timing",
+    }
+
+
+# search source footage while preventing the report from replacing an input
+def main():
+    """Search source footage while preventing the report from replacing an input."""
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("query")
+    p.add_argument("source")
+    p.add_argument("--every", type=float, default=1)
+    p.add_argument("--out", required=True)
+    a = p.parse_args()
+    if Path(a.out).resolve() in (Path(a.query).resolve(), Path(a.source).resolve()):
+        p.error("output would overwrite input")
+    if Path(a.out).exists() or Path(a.out).is_symlink():
+        p.error("output already exists choose a new report path")
+    save_json(a.out, search(a.query, a.source, a.every), exclusive=True)
+
+
+if __name__ == "__main__":
+    main()