From cd6090d2aec1b6b481531af6ba5c9926301d2bdc Mon Sep 17 00:00:00 2001 From: Alexander Refsum Jensenius Date: Mon, 24 Aug 2026 23:04:46 +0200 Subject: [PATCH 1/3] WIP: a lean one-pass extractor for long recordings. NOT TRUSTED YET. Deliberately not exported and deliberately not used for tonight's run. It is here so the work and the measurements behind it are not lost, and so the next session starts from the bug rather than from the idea. WHAT IT IS FOR. mg_motion is built for a clip and for a person looking at the result; on 120 s of 1080p the cost decomposes as 245 s for motion_analysis='all' with motiongrams, 215 s for qom with motiongrams, and 62 s for qom without. The motiongrams are 71 per cent and the area of motion another 12. This converts each motion frame to greyscale ONCE and takes qom and both videogram columns from it, writes into preallocated memmaps so nothing reallocates, and stores the videogram as a pyramid of extremes so a two-hour session can be zoomed like an audio waveform. Serially it ran 120 s of video in 60 s; over 8 processes, 27 s. WHY IT IS NOT TRUSTED. Its qom differs from mg_motion's, and worse, it differs BETWEEN IDENTICAL RUNS OF ITSELF. Non-determinism means frames are being misaligned somewhere in how this drives ffmpeg_cmd, not that a parameter is wrong. Two real bugs were found and fixed on the way and neither was the cause: - `-t` was placed between the video input and the `color=` inputs that filter_frame_ffmpeg appends, so it bound to one of THOSE inputs instead of the output, changing which frames the filter saw. - pipe reads were not looped to a full frame; a short read slides every later frame across the boundary. The same two commands run through plain subprocess.Popen produce identical frames, so the fault is on this side, not ffmpeg's. WHAT WAS MEASURED AND REJECTED, so nobody repeats it: ffmpeg emitting gray is slower than rgb24 because the chain is RGB-native; signalstats on MGT's own chain gives r=0.90 against QomRaw because the threshold filter's floor adds a constant; scale=1:H:flags=area inside ffmpeg takes 106 s against 60 s; packet size from the bitstream is free but correlates only r=0.33; and gray+tblend+signalstats takes 118 s for r=0.94. ffmpeg wins at decoding and filtering and loses at reductions. Next step is a test that fails on the current code: extract two overlapping ranges and assert the shared frames are equal, which the non-determinism will break immediately. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Xg1f739wddu5M4s3UwdNkn --- musicalgestures/_tracks.py | 406 +++++++++++++++++++++++++++++++++++++ 1 file changed, 406 insertions(+) create mode 100644 musicalgestures/_tracks.py diff --git a/musicalgestures/_tracks.py b/musicalgestures/_tracks.py new file mode 100644 index 0000000..7bcd4d4 --- /dev/null +++ b/musicalgestures/_tracks.py @@ -0,0 +1,406 @@ +"""One pass over a long recording, and everything a timeline needs afterwards. + +`mg_motion` is built for a clip and for a person looking at the result: it can write +a motion video, plots, motiongrams and a data file, and it computes centroid and area +whether or not you asked for them. That generosity is the right default for interactive +use and the wrong one for a two-hour session, where the cost decomposes like this on +120 s of 1080p video: + + motion_analysis='all', motiongrams on 245 s + motion_analysis='qom', motiongrams on 215 s + motion_analysis='qom', motiongrams off 62 s + +The motiongrams are 71 per cent of it and the area of motion another 12. This module +does the one pass those numbers argue for: **convert each motion frame to greyscale +once, and take everything from that** --- the quantity of motion, and both videogram +columns. `centroid()` converts to greyscale internally and then throws the conversion +away; doing it once and reusing it is most of the saving, and working on one channel +rather than three is the rest. + +**Nothing is appended to a growing array.** The frame count is known before the pass +starts, so the columns go into a preallocated memory-mapped file. That is not a +micro-optimisation: growing these by `np.append` is what made a session take an +extrapolated 215 hours before 2026-08-24. + +**The videogram is stored as a pyramid, the way an audio editor stores peaks.** A +column per frame is finer than any page can show --- 50 columns per second on an A4 +width is one column per 20 pixels even when zoomed to a single action --- but the +whole session at that rate is 475,680 columns and cannot be drawn at all. So each +level halves the one below it by taking the extremes rather than the mean, because a +brief movement must survive being zoomed out of; averaging is what makes a spike +disappear at low magnification. Levels are built once, after the pass, from the base +that is already on disk, and cost a geometric series: less than the base again. + +Reading is then a slice: pick the level whose width is nearest the pixels available +and take the columns for the time range wanted. +""" +from __future__ import annotations + +import json +import os +from concurrent.futures import ProcessPoolExecutor +from pathlib import Path + +import cv2 +import numpy as np + +import musicalgestures +from musicalgestures._filter import filter_frame_ffmpeg +from musicalgestures._utils import MgProgressbar, ffmpeg_cmd + +#: Coarser levels stop here: below this a level is too narrow to be worth a file. +MIN_LEVEL_COLUMNS = 64 + + +def _analysis_dir(video, out_dir=None) -> Path: + """`analysis//` beside the video unless told otherwise.""" + video = Path(video) + root = Path(out_dir) if out_dir else video.parent / "analysis" + d = root / video.stem + d.mkdir(parents=True, exist_ok=True) + return d + + +def _frame_count(video, fps, duration) -> int: + """Frames to expect, from the container's own duration. + + Deliberately an estimate with room in it: the memmaps are opened at this size and + trimmed to what actually arrived, because a container that misreports its duration + should cost a truncated file rather than a crash halfway through a two-hour pass. + """ + return int(round(float(duration) * float(fps))) + 8 + + + + +def _read_exact(stream, n: int) -> bytes | None: + """Read exactly n bytes, or None at end of stream. + + **A pipe read can return short.** `stdout.read(n)` on a pipe is not obliged to + give n bytes, and treating a short read as a whole frame slides every later frame + across the boundary --- which shows up as output that changes between identical + runs rather than as an error. Loop until the frame is complete. + """ + buf = bytearray() + while len(buf) < n: + part = stream.read(n - len(buf)) + if not part: + return None + buf.extend(part) + return bytes(buf) + + +def _chunk_worker(args) -> int: + """Extract one time range into its slice of the memmaps. Runs in its own process. + + **Each chunk starts one frame early and throws that frame away.** The motion frame + is a difference against the preceding frame, so the first frame after a seek has no + predecessor and is not a motion frame at all. Overlapping by one and discarding it + makes a chunked pass identical to a serial one instead of merely close. + """ + (video, d, i0, n_frames, t0, fps, W, H, n_total, + filtertype, threshold, blur, use_median, kernel_size, plate_every) = args + import cv2 as _cv2 + import numpy as _np + import musicalgestures as _mg + from musicalgestures._filter import filter_frame_ffmpeg as _ffilter + from musicalgestures._utils import ffmpeg_cmd as _ffcmd + + d = Path(d) + lead = 1.0 / fps + seek = max(0.0, t0 - lead) + drop = 1 if t0 > 0 else 0 + dur = (n_frames + drop) / fps + + #: -ss goes before the input it seeks; -t must come AFTER -filter_complex so it + #: is an OUTPUT option. filter_frame_ffmpeg appends further inputs (infinite + #: `color=` sources for the threshold filter), so a -t placed between the video + #: and them binds to one of THOSE inputs instead, changing which frames the + #: filter sees. That produced an envelope differing from the serial one on almost + #: every frame while looking like a chunking bug. + cmd = ["ffmpeg", "-y", "-ss", f"{seek:.6f}", "-i", str(video)] + cmd, chain = _ffilter(str(video), cmd, True, blur, filtertype, + threshold, kernel_size, use_median) + cmd += ["-filter_complex", chain[:-1], "-t", f"{dur:.6f}", + "-f", "rawvideo", "-pix_fmt", "rgb24", "-"] + + qom = _np.memmap(d / "qom.f4", dtype=_np.float32, mode="r+", shape=(n_total,)) + vg = _np.memmap(d / "videogram_v.u1", dtype=_np.uint8, mode="r+", shape=(n_total, H)) + hg = _np.memmap(d / "videogram_h.u1", dtype=_np.uint8, mode="r+", shape=(n_total, W)) + + #: total_time drives the progress bar's arithmetic, so it must be a number + #: even when no bar is wanted --- None makes it subtract from nothing. + proc = _ffcmd(cmd, total_time=dur, pipe="read", stream=False) + nbytes = W * H * 3 + seen = written = 0 + plates = [] + while written < n_frames: + buf = _read_exact(proc.stdout, nbytes) + if buf is None: + break + seen += 1 + if seen <= drop: + continue + frame = _np.frombuffer(buf, dtype=_np.uint8).reshape(H, W, 3) + grey = _cv2.cvtColor(frame, _cv2.COLOR_BGR2GRAY) + j = i0 + written + qom[j] = float(_cv2.sumElems(grey)[0]) + vg[j] = grey.mean(axis=1).round().astype(_np.uint8) + hg[j] = grey.mean(axis=0).round().astype(_np.uint8) + if plate_every and j % plate_every == 0: + plates.append(frame.copy()) + written += 1 + proc.terminate() + qom.flush(); vg.flush(); hg.flush() + if plates: + _np.save(d / f".plate_{i0}.npy", _np.stack(plates)) + (d / f".done_{i0}").write_text(str(written)) + return written + + +def extract_tracks(video, out_dir=None, filtertype="Regular", threshold=0.05, + blur="None", use_median=False, kernel_size=5, + plate_every=None, progress=True) -> dict: + """Quantity of motion and both videogram bases, in one pass over the video. + + Args: + video: path to the recording. + out_dir: where `analysis//` goes. Defaults to beside the video. + filtertype, threshold, blur, use_median, kernel_size: passed to the same + ffmpeg filter chain `mg_motion` uses, so the motion frames are identical. + plate_every: keep one raw frame in this many for the room plate, or None to + keep none. The frames are sampled across the whole recording, so a plate + built from them describes the whole room rather than one stretch. + progress: show a progress bar. + + Returns: + dict: paths written, and the parameters that made them. + """ + video = Path(video) + mgv = musicalgestures.MgVideo(str(video)) + W, H, fps = mgv.width, mgv.height, float(mgv.fps) + n_max = _frame_count(video, fps, mgv.length / fps if mgv.length else 0) \ + if mgv.length else 10 ** 7 + if mgv.length: + n_max = int(mgv.length) + 8 + + d = _analysis_dir(video, out_dir) + qom_path = d / "qom.f4" + vgram_path = d / "videogram_v.u1" # one column per frame, height H + hgram_path = d / "videogram_h.u1" # one row per frame, width W + + qom = np.memmap(qom_path, dtype=np.float32, mode="w+", shape=(n_max,)) + vg = np.memmap(vgram_path, dtype=np.uint8, mode="w+", shape=(n_max, H)) + hg = np.memmap(hgram_path, dtype=np.uint8, mode="w+", shape=(n_max, W)) + + cmd = ["ffmpeg", "-y", "-i", str(video)] + cmd, chain = filter_frame_ffmpeg(str(video), cmd, True, blur, filtertype, + threshold, kernel_size, use_median) + cmd += ["-filter_complex", chain[:-1], "-f", "rawvideo", "-pix_fmt", "rgb24", "-"] + + plates: list = [] + pb = MgProgressbar(total=n_max, prefix="Tracks:") if progress else None + process = ffmpeg_cmd(cmd, total_time=mgv.length, pipe="read") + + i = 0 + nbytes = W * H * 3 + while i < n_max: + buf = _read_exact(process.stdout, nbytes) + if buf is None: + break + frame = np.frombuffer(buf, dtype=np.uint8).reshape(H, W, 3) + #: ONE conversion, three uses. centroid() does this conversion internally and + #: discards it; the videogram columns then redo the work on three channels. + grey = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) + qom[i] = float(cv2.sumElems(grey)[0]) + vg[i] = grey.mean(axis=1).round().astype(np.uint8) + hg[i] = grey.mean(axis=0).round().astype(np.uint8) + if plate_every and i % plate_every == 0: + plates.append(frame.copy()) + if pb: + pb.progress(i) + i += 1 + process.terminate() + if pb: + pb.progress(n_max) + + n = i + qom.flush(); vg.flush(); hg.flush() + del qom, vg, hg + _truncate(qom_path, n * 4) + _truncate(vgram_path, n * H) + _truncate(hgram_path, n * W) + + meta = { + "video": str(video), "frames": n, "fps": fps, "width": W, "height": H, + "duration_s": n / fps, + "filtertype": filtertype, "threshold": threshold, "blur": blur, + "use_median": use_median, "kernel_size": kernel_size, + "qom": qom_path.name, "videogram_v": vgram_path.name, + "videogram_h": hgram_path.name, + "note": ("qom is the sum of the greyscale motion frame, the same quantity " + "mg_motion writes as QomRaw. The videogram bases hold one column " + "per frame; read them through pyramid levels rather than whole."), + } + if plates: + plate = np.median(np.stack(plates), axis=0).astype(np.uint8) + cv2.imwrite(str(d / "room_plate.png"), plate) + meta["room_plate"] = "room_plate.png" + meta["plate_frames"] = len(plates) + meta["plate_note"] = ("MEDIAN, not mean. A mean over frames with performers in " + "different places keeps a faint ghost of each of them " + "everywhere they stood; a median removes them, because at " + "any pixel they are a minority of the samples.") + (d / "tracks.json").write_text(json.dumps(meta, indent=1) + "\n") + return meta + + +def _truncate(path: Path, nbytes: int) -> None: + """Cut a memmap file back to the rows that were actually written.""" + with open(path, "r+b") as fh: + fh.truncate(nbytes) + + +def build_pyramid(analysis_dir, which="videogram_v") -> list[Path]: + """Halve a videogram base repeatedly, keeping extremes rather than means. + + Level 0 is the base, one column per frame. Level k is 2^k frames per column, and + each column holds the greatest value of the columns beneath it. **Extremes, not + means**: a movement lasting a few frames is exactly what a viewer zooms out to + find, and averaging is what makes it vanish at low magnification. + + Returns the paths written, coarsest last. + """ + d = Path(analysis_dir) + meta = json.loads((d / "tracks.json").read_text()) + n, H, W = meta["frames"], meta["height"], meta["width"] + span = H if which == "videogram_v" else W + base = np.memmap(d / meta[which], dtype=np.uint8, mode="r", shape=(n, span)) + + out, level, cur = [], 0, np.asarray(base) + while cur.shape[0] > MIN_LEVEL_COLUMNS: + level += 1 + m = cur.shape[0] // 2 + pair = cur[: m * 2].reshape(m, 2, span) + cur = pair.max(axis=1) + p = d / f"{which}.L{level}.u1" + np.asarray(cur, dtype=np.uint8).tofile(p) + out.append(p) + meta.setdefault("pyramid", {})[which] = [p.name for p in out] + (d / "tracks.json").write_text(json.dumps(meta, indent=1) + "\n") + return out + + +def read_columns(analysis_dir, start_s=0.0, end_s=None, max_columns=2000, + which="videogram_v") -> tuple[np.ndarray, float]: + """The videogram for a time range, at the coarsest level that still fills the width. + + This is how an audio editor draws a waveform: choose the level whose resolution + the display can use and read a slice of it, rather than reading everything and + throwing most of it away. + + Returns (columns, seconds_per_column). + """ + d = Path(analysis_dir) + meta = json.loads((d / "tracks.json").read_text()) + n, fps = meta["frames"], meta["fps"] + span = meta["height"] if which == "videogram_v" else meta["width"] + end_s = meta["duration_s"] if end_s is None else end_s + want = max(1, int((end_s - start_s) * fps)) + + #: Choose the level whose column count for this range is nearest below the + #: pixels available. Reading a finer level and decimating in the reader would + #: undo the point of having levels at all. + level, stride = 0, 1 + while want // (stride * 2) >= max_columns and stride * 2 <= n: + stride *= 2 + level += 1 + if level == 0: + arr = np.memmap(d / meta[which], dtype=np.uint8, mode="r", shape=(n, span)) + else: + name = f"{which}.L{level}.u1" + rows = n // stride + arr = np.memmap(d / name, dtype=np.uint8, mode="r", shape=(rows, span)) + + lo = int(start_s * fps) // stride + hi = int(end_s * fps) // stride + return np.asarray(arr[lo:hi]), stride / fps + + +def extract_tracks_parallel(video, out_dir=None, workers=None, chunk_s=120.0, + filtertype="Regular", threshold=0.05, blur="None", + use_median=False, kernel_size=5, plate_every=None, + resume=True) -> dict: + """The same pass, split over processes by time. Resumable. + + The work is embarrassingly parallel because each frame's motion depends only on + its predecessor, so a chunk needs one frame of lead-in and nothing else. Workers + write into disjoint slices of the same memory-mapped files, which is why no + merging step is needed and why a crashed worker costs one chunk rather than the run. + + `resume=True` skips chunks that already left a marker, so restarting after a + failure at hour five does not redo hours one to four --- the lesson the SINS + producers learned by truncating a completed table. + """ + video = Path(video) + mgv = musicalgestures.MgVideo(str(video)) + W, H, fps = mgv.width, mgv.height, float(mgv.fps) + n_total = int(mgv.length) + 8 + d = _analysis_dir(video, out_dir) + + for name, dt, shape in (("qom.f4", np.float32, (n_total,)), + ("videogram_v.u1", np.uint8, (n_total, H)), + ("videogram_h.u1", np.uint8, (n_total, W))): + if not (d / name).exists() or not resume: + m = np.memmap(d / name, dtype=dt, mode="w+", shape=shape) + m.flush(); del m + + per = max(1, int(round(chunk_s * fps))) + jobs = [] + for i0 in range(0, n_total, per): + n_frames = min(per, n_total - i0) + if resume and (d / f".done_{i0}").exists(): + continue + jobs.append((str(video), str(d), i0, n_frames, i0 / fps, fps, W, H, n_total, + filtertype, threshold, blur, use_median, kernel_size, plate_every)) + + workers = workers or max(1, min(os.cpu_count() or 2, 8)) + if jobs: + with ProcessPoolExecutor(max_workers=workers) as pool: + list(pool.map(_chunk_worker, jobs)) + + #: The true frame count is where the last chunk stopped, not the estimate. + written = 0 + for f in sorted(d.glob(".done_*"), key=lambda q: int(q.name.split("_")[1])): + i0 = int(f.name.split("_")[1]) + written = max(written, i0 + int(f.read_text() or 0)) + n = written or n_total + + _truncate(d / "qom.f4", n * 4) + _truncate(d / "videogram_v.u1", n * H) + _truncate(d / "videogram_h.u1", n * W) + + meta = {"video": str(video), "frames": n, "fps": fps, "width": W, "height": H, + "duration_s": n / fps, "filtertype": filtertype, "threshold": threshold, + "blur": blur, "use_median": use_median, "kernel_size": kernel_size, + "workers": workers, "chunk_s": chunk_s, + "qom": "qom.f4", "videogram_v": "videogram_v.u1", + "videogram_h": "videogram_h.u1", + "note": ("qom is the sum of the greyscale motion frame, the quantity " + "mg_motion writes as QomRaw. Chunks overlap by one frame and " + "discard it, because the first frame after a seek has no " + "predecessor to differ from.")} + + plate_files = sorted(d.glob(".plate_*.npy")) + if plate_files: + stack = np.concatenate([np.load(f) for f in plate_files]) + cv2.imwrite(str(d / "room_plate.png"), + np.median(stack, axis=0).astype(np.uint8)) + meta["room_plate"] = "room_plate.png" + meta["plate_frames"] = int(stack.shape[0]) + meta["plate_note"] = ("MEDIAN, not mean: a mean keeps a faint ghost of each " + "performer everywhere they stood.") + for f in plate_files: + f.unlink() + (d / "tracks.json").write_text(json.dumps(meta, indent=1) + "\n") + return meta From 6eb32b3dd82cc3bdab64c009511fd53a1eb42c5e Mon Sep 17 00:00:00 2001 From: Alexander Refsum Jensenius Date: Mon, 24 Aug 2026 23:26:36 +0200 Subject: [PATCH 2/3] The non-determinism was two output specs feeding one stdout ffmpeg_cmd(pipe="read") appends its own output arguments --- -f image2pipe -pix_fmt bgr24 -vcodec rawvideo - --- and this module had already appended its own. ffmpeg was given two outputs and wrote BOTH into the same stdout, interleaved, so the frames were wrong and, because interleaving depends on buffering, different between identical runs. mg_motion works because it stops at -filter_complex and lets ffmpeg_cmd finish the command. SERIAL IS NOW EXACT: byte-for-byte equal to mg_motion's QomRaw over 6005 frames of 1080p. Two earlier fixes on the way --- a -t binding to a color= input rather than the output, and pipe reads not looped to a whole frame --- were real bugs but not this one. THE PARALLEL PATH IS STILL NOT EXACT and is not to be used. On 1920x1080 at 50 fps with eight workers and 15 s chunks it repeats one frame at the LAST chunk seam: -ss before -i seeks to a keyframe, so the frames decoded before the target are not always the single frame the worker drops. All seven interior seams are exact; one frame in 6005 is not. tests/test_tracks.py asserts agreement with mg_motion and determinism, both of which fail on the old code. Its third test, parallel against serial, PASSES on a small synthetic clip and does NOT reproduce the seam artefact --- so it says so in its own docstring rather than reading like cover. It was first written as a strict xfail and XPASSed, which is how that was discovered. mypy clean across 71 files. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Xg1f739wddu5M4s3UwdNkn --- musicalgestures/_tracks.py | 30 ++++++++++--- tests/test_tracks.py | 89 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 113 insertions(+), 6 deletions(-) create mode 100644 tests/test_tracks.py diff --git a/musicalgestures/_tracks.py b/musicalgestures/_tracks.py index 7bcd4d4..f2b9ee9 100644 --- a/musicalgestures/_tracks.py +++ b/musicalgestures/_tracks.py @@ -56,7 +56,7 @@ def _analysis_dir(video, out_dir=None) -> Path: """`analysis//` beside the video unless told otherwise.""" video = Path(video) root = Path(out_dir) if out_dir else video.parent / "analysis" - d = root / video.stem + d: Path = root / video.stem d.mkdir(parents=True, exist_ok=True) return d @@ -99,7 +99,7 @@ def _chunk_worker(args) -> int: makes a chunked pass identical to a serial one instead of merely close. """ (video, d, i0, n_frames, t0, fps, W, H, n_total, - filtertype, threshold, blur, use_median, kernel_size, plate_every) = args + filtertype, threshold, blur, use_median, kernel_size, plate_every, is_last) = args import cv2 as _cv2 import numpy as _np import musicalgestures as _mg @@ -121,8 +121,19 @@ def _chunk_worker(args) -> int: cmd = ["ffmpeg", "-y", "-ss", f"{seek:.6f}", "-i", str(video)] cmd, chain = _ffilter(str(video), cmd, True, blur, filtertype, threshold, kernel_size, use_median) - cmd += ["-filter_complex", chain[:-1], "-t", f"{dur:.6f}", - "-f", "rawvideo", "-pix_fmt", "rgb24", "-"] + #: STOP AT -filter_complex. ffmpeg_cmd(pipe="read") appends its OWN output + #: arguments --- `-f image2pipe -pix_fmt bgr24 -vcodec rawvideo -` --- so adding + #: an output spec here gives ffmpeg two outputs and it writes BOTH into the same + #: stdout, interleaved. That produced frames that were wrong and, because the + #: interleaving depends on buffering, different between identical runs. The pixel + #: format is therefore bgr24, which is what COLOR_BGR2GRAY below expects. + cmd += ["-filter_complex", chain[:-1]] + #: The LAST chunk runs to the end of file rather than to a computed duration. + #: A -t derived from an estimated frame count can stop a frame early or late + #: against the container's real end; the loop already stops at n_frames, so the + #: bound is redundant there and was costing one frame at the tail. + if not is_last: + cmd += ["-t", f"{dur:.6f}"] qom = _np.memmap(d / "qom.f4", dtype=_np.float32, mode="r+", shape=(n_total,)) vg = _np.memmap(d / "videogram_v.u1", dtype=_np.uint8, mode="r+", shape=(n_total, H)) @@ -196,7 +207,13 @@ def extract_tracks(video, out_dir=None, filtertype="Regular", threshold=0.05, cmd = ["ffmpeg", "-y", "-i", str(video)] cmd, chain = filter_frame_ffmpeg(str(video), cmd, True, blur, filtertype, threshold, kernel_size, use_median) - cmd += ["-filter_complex", chain[:-1], "-f", "rawvideo", "-pix_fmt", "rgb24", "-"] + #: STOP AT -filter_complex. ffmpeg_cmd(pipe="read") appends its OWN output + #: arguments --- `-f image2pipe -pix_fmt bgr24 -vcodec rawvideo -` --- so adding + #: an output spec here gives ffmpeg two outputs and it writes BOTH into the same + #: stdout, interleaved. That produced frames that were wrong and, because the + #: interleaving depends on buffering, different between identical runs. The pixel + #: format is therefore bgr24, which is what COLOR_BGR2GRAY below expects. + cmd += ["-filter_complex", chain[:-1]] plates: list = [] pb = MgProgressbar(total=n_max, prefix="Tracks:") if progress else None @@ -362,7 +379,8 @@ def extract_tracks_parallel(video, out_dir=None, workers=None, chunk_s=120.0, if resume and (d / f".done_{i0}").exists(): continue jobs.append((str(video), str(d), i0, n_frames, i0 / fps, fps, W, H, n_total, - filtertype, threshold, blur, use_median, kernel_size, plate_every)) + filtertype, threshold, blur, use_median, kernel_size, plate_every, + i0 + n_frames >= n_total)) workers = workers or max(1, min(os.cpu_count() or 2, 8)) if jobs: diff --git a/tests/test_tracks.py b/tests/test_tracks.py new file mode 100644 index 0000000..af330e7 --- /dev/null +++ b/tests/test_tracks.py @@ -0,0 +1,89 @@ +"""The lean extractor must agree with mg_motion, and with itself. + +`_tracks.extract_tracks` exists to do in one pass what `mg_motion` does in several, +for recordings too long for the general path. It is only worth having if it produces +the same numbers, so that is what these assert --- against `mg_motion` itself, which +is the known answer. + +The bug these were written for: an earlier version appended its own output arguments +before calling `ffmpeg_cmd(pipe="read")`, which appends its OWN. ffmpeg was given two +outputs and wrote both into the same stdout, interleaved. The result was frames that +were wrong and, because interleaving depends on buffering, **different between +identical runs**. Nothing in the suite would have caught it, so these exist. +""" +import csv +import subprocess + +import numpy as np +import musicalgestures as mg +from musicalgestures._tracks import extract_tracks, extract_tracks_parallel + + +def _synth(path, seconds=4, fps=25, size="320x240"): + subprocess.run( + ["ffmpeg", "-v", "error", "-y", "-f", "lavfi", + "-i", f"testsrc=size={size}:rate={fps}:duration={seconds}", + "-pix_fmt", "yuv420p", str(path)], + check=True, capture_output=True) + return str(path) + + +def _qom_from_tracks(meta, d): + return np.memmap(d / "qom.f4", dtype=np.float32, mode="r").astype(float) + + +def _qom_from_mg_motion(path): + mg.MgVideo(path).motion(motion_analysis="qom", save_motiongrams=False, + save_video=False, save_plot=False, save_data=True, + normalize=False) + csv_path = str(path).rsplit(".", 1)[0] + "_motion.csv" + return np.array([float(r["QomRaw"]) for r in csv.DictReader(open(csv_path))]) + + +def test_qom_matches_mg_motion(tmp_path): + """The whole point: same numbers as the path it replaces.""" + v = _synth(tmp_path / "v.mp4") + meta = extract_tracks(v, out_dir=tmp_path / "out", progress=False) + mine = _qom_from_tracks(meta, tmp_path / "out" / "v") + truth = _qom_from_mg_motion(v) + n = min(len(mine), len(truth)) + assert n > 50, "clip too short to be a real comparison" + np.testing.assert_allclose(mine[:n], truth[:n], rtol=0, atol=1e-3) + + +def test_extraction_is_deterministic(tmp_path): + """Run it twice and get the same answer. + + This is the test that would have caught the interleaved-output bug on the day it + was written, and it is cheap. An extractor whose answer depends on buffering is + not an extractor. + """ + v = _synth(tmp_path / "v.mp4") + a = extract_tracks(v, out_dir=tmp_path / "a", progress=False) + b = extract_tracks(v, out_dir=tmp_path / "b", progress=False) + qa = _qom_from_tracks(a, tmp_path / "a" / "v") + qb = _qom_from_tracks(b, tmp_path / "b" / "v") + assert np.array_equal(qa, qb), "two identical runs disagreed" + + +def test_parallel_matches_serial(tmp_path): + """Chunked extraction must equal serial extraction. + + **This passes here and does NOT cover the fault that matters.** On the real + material --- 1920x1080 at 50 fps, 120 s, eight workers, 15 s chunks --- the + parallel path repeats one frame at the last chunk seam: `-ss` before `-i` seeks + to a keyframe, so the number of frames decoded before the target is not always + the single frame the worker drops. On this small synthetic clip the seek lands + exactly and the artefact does not appear. + + So this guards the easy case only, and the parallel path should not be used until + a fixture reproduces the hard one. Written down because a passing test that does + not exercise the known bug is worse than no test: it reads like cover. + """ + v = _synth(tmp_path / "v.mp4", seconds=8) + s = extract_tracks(v, out_dir=tmp_path / "s", progress=False) + p = extract_tracks_parallel(v, out_dir=tmp_path / "p", workers=4, chunk_s=2) + qs = _qom_from_tracks(s, tmp_path / "s" / "v") + qp = _qom_from_tracks(p, tmp_path / "p" / "v") + n = min(len(qs), len(qp)) + assert np.array_equal(qs[:n], qp[:n]) From a84f91aa265f29fbcd5055f53aa54fb3ae4af748 Mon Sep 17 00:00:00 2001 From: Alexander Refsum Jensenius Date: Mon, 24 Aug 2026 23:50:19 +0200 Subject: [PATCH 3/3] The parallel path is exact: trim by time, never count frames after a seek Chunks were seeked with -ss and then had exactly one frame dropped as the difference filter's lead-in. -ss before -i lands on a keyframe, so the frames arriving before the target are not always one, and a chunk could repeat its predecessor's value --- one wrong frame in 6,005 on 1080p/50 fps. Fixed structurally rather than by adjusting the count: seek a whole second early, then let ffmpeg's `trim` keep the wanted range BY TIMESTAMP. There is no count left to get wrong. Parallel output is now byte-identical to serial AND to mg_motion's QomRaw across all 6,005 frames of the 120 s 1080p clip. Cost of the lead-in is one second of decode per chunk: 6.7 per cent at the 15 s chunks used for testing, under 0.2 per cent at the 600 s chunks a real session would use. THE TEST FOR THIS DOES NOT COVER IT, AND SAYS SO. Written twice --- once with a default GOP, once with keyframes forced off the chunk grid --- and both versions pass against the broken code. The artefact needs the real material to appear, and a 1080p 120 s fixture does not belong in a unit suite. Rather than leave a test whose name implies coverage it does not have, the docstring states what it does guard, what it does not, and that the real protection is structural: if the worker ever goes back to counting frames after a seek, this test will not notice. 722 tests pass, mypy clean across 71 files. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Xg1f739wddu5M4s3UwdNkn --- musicalgestures/_tracks.py | 28 +++++++++++-------- tests/test_tracks.py | 57 ++++++++++++++++++++++++++++---------- 2 files changed, 58 insertions(+), 27 deletions(-) diff --git a/musicalgestures/_tracks.py b/musicalgestures/_tracks.py index f2b9ee9..b196f47 100644 --- a/musicalgestures/_tracks.py +++ b/musicalgestures/_tracks.py @@ -107,10 +107,14 @@ def _chunk_worker(args) -> int: from musicalgestures._utils import ffmpeg_cmd as _ffcmd d = Path(d) - lead = 1.0 / fps + #: SEEK EARLY AND TRIM BY TIME, rather than seeking close and dropping a frame. + #: `-ss` before `-i` lands on a keyframe, so how many frames arrive before the + #: target is not fixed --- dropping exactly one left a repeated frame at a seam. + #: A whole second of lead-in guarantees the difference filter has a predecessor, + #: and `trim` then keeps precisely the wanted range by timestamp. + lead = 1.0 if t0 > 0 else 0.0 seek = max(0.0, t0 - lead) - drop = 1 if t0 > 0 else 0 - dur = (n_frames + drop) / fps + dur = lead + n_frames / fps #: -ss goes before the input it seeks; -t must come AFTER -filter_complex so it #: is an OUTPUT option. filter_frame_ffmpeg appends further inputs (infinite @@ -127,11 +131,14 @@ def _chunk_worker(args) -> int: #: stdout, interleaved. That produced frames that were wrong and, because the #: interleaving depends on buffering, different between identical runs. The pixel #: format is therefore bgr24, which is what COLOR_BGR2GRAY below expects. - cmd += ["-filter_complex", chain[:-1]] - #: The LAST chunk runs to the end of file rather than to a computed duration. - #: A -t derived from an estimated frame count can stop a frame early or late - #: against the container's real end; the loop already stops at n_frames, so the - #: bound is redundant there and was costing one frame at the tail. + trim = "" + if lead: + #: Relative to the seek point, keep from `lead` onwards. setpts restarts the + #: clock so downstream sees a normal stream. + trim = f",trim=start={lead:.6f},setpts=PTS-STARTPTS" + cmd += ["-filter_complex", chain[:-1] + trim] + #: The last chunk runs to end of file; earlier ones are bounded so a worker does + #: not decode the rest of a two-hour recording it will discard. if not is_last: cmd += ["-t", f"{dur:.6f}"] @@ -143,15 +150,12 @@ def _chunk_worker(args) -> int: #: even when no bar is wanted --- None makes it subtract from nothing. proc = _ffcmd(cmd, total_time=dur, pipe="read", stream=False) nbytes = W * H * 3 - seen = written = 0 + written = 0 plates = [] while written < n_frames: buf = _read_exact(proc.stdout, nbytes) if buf is None: break - seen += 1 - if seen <= drop: - continue frame = _np.frombuffer(buf, dtype=_np.uint8).reshape(H, W, 3) grey = _cv2.cvtColor(frame, _cv2.COLOR_BGR2GRAY) j = i0 + written diff --git a/tests/test_tracks.py b/tests/test_tracks.py index af330e7..64aadf7 100644 --- a/tests/test_tracks.py +++ b/tests/test_tracks.py @@ -66,24 +66,51 @@ def test_extraction_is_deterministic(tmp_path): assert np.array_equal(qa, qb), "two identical runs disagreed" -def test_parallel_matches_serial(tmp_path): - """Chunked extraction must equal serial extraction. - - **This passes here and does NOT cover the fault that matters.** On the real - material --- 1920x1080 at 50 fps, 120 s, eight workers, 15 s chunks --- the - parallel path repeats one frame at the last chunk seam: `-ss` before `-i` seeks - to a keyframe, so the number of frames decoded before the target is not always - the single frame the worker drops. On this small synthetic clip the seek lands - exactly and the artefact does not appear. - - So this guards the easy case only, and the parallel path should not be used until - a fixture reproduces the hard one. Written down because a passing test that does - not exercise the known bug is worse than no test: it reads like cover. +def _synth_keyframes(path, seconds=10, fps=25, size="320x240", gop=37): + """A clip whose keyframes deliberately do NOT align with chunk boundaries. + + `gop=37` at 25 fps puts a keyframe every 1.48 s, so a chunk starting on a round + second lands mid-GOP and `-ss` seeks backwards to the keyframe. How many frames + then arrive before the target varies --- which is exactly what the old + drop-exactly-one-frame logic got wrong. + """ + subprocess.run( + ["ffmpeg", "-v", "error", "-y", "-f", "lavfi", + "-i", f"testsrc=size={size}:rate={fps}:duration={seconds}", + "-g", str(gop), "-pix_fmt", "yuv420p", str(path)], + check=True, capture_output=True) + return str(path) + + +def test_parallel_matches_serial_across_unaligned_keyframes(tmp_path): + """Chunked extraction must equal serial extraction, seams included. + + The fault this exists for: chunks were seeked with `-ss` and then had exactly one + frame dropped as the difference filter's lead-in. `-ss` before `-i` lands on a + keyframe, so the frames arriving before the target are not always one, and a chunk + could repeat its predecessor's value. On 1080p/50 fps that showed up as a single + wrong frame in 6,005 --- small enough to dismiss and wrong all the same. + + **THIS TEST DOES NOT REPRODUCE THAT FAULT, and saying so is the point.** It was + written twice --- once with a default GOP, once with keyframes forced off the chunk + grid as here --- and BOTH versions pass against the broken drop-exactly-one-frame + code. The artefact needs the real material to appear: 1920x1080 at 50 fps, 120 s, + eight workers, 15 s chunks, where it showed up as one wrong frame in 6,005. A + fixture that heavy does not belong in a unit suite. + + So this guards the easy case and no more. What actually prevents the fault coming + back is structural rather than tested: the worker no longer counts frames after a + seek at all --- it seeks a second early and lets `trim` keep the wanted range by + timestamp, so there is no count to get wrong. If that ever reverts to counting, + this test will not notice. The comment in `_chunk_worker` says the same thing at + the place where it would happen. """ - v = _synth(tmp_path / "v.mp4", seconds=8) + v = _synth_keyframes(tmp_path / "v.mp4") s = extract_tracks(v, out_dir=tmp_path / "s", progress=False) p = extract_tracks_parallel(v, out_dir=tmp_path / "p", workers=4, chunk_s=2) qs = _qom_from_tracks(s, tmp_path / "s" / "v") qp = _qom_from_tracks(p, tmp_path / "p" / "v") n = min(len(qs), len(qp)) - assert np.array_equal(qs[:n], qp[:n]) + assert n > 200 + bad = np.flatnonzero(qs[:n] != qp[:n]) + assert len(bad) == 0, f"{len(bad)} frames differ at/after chunk seams: {bad[:8]}"