From a066a082569e98d4cab9661301b64a21284f8abb Mon Sep 17 00:00:00 2001 From: kaffeindecaf Date: Sat, 24 Oct 2026 13:05:00 +0200 Subject: [PATCH 1/3] feat: add timestamp-preserving causality inversion --- .github/workflows/tests.yml | 30 +++ README.md | 72 ++++++- ROADMAP.md | 20 +- benchmarks/inversion.json | 143 ++++++++++++++ effex.py | 96 +++++++++- frame_pipeline.py | 37 ++++ requirements-inversion.txt | 3 + stateful_effects.py | 210 +++++++++++++++++++++ tests/benchmark_inversion.py | 88 +++++++++ tests/inversion_regression.py | 342 ++++++++++++++++++++++++++++++++++ tests/menu_smoke.py | 43 ++++- tests/run_tests.py | 1 + timestamped_frames.py | 118 ++++++++++++ 13 files changed, 1192 insertions(+), 11 deletions(-) create mode 100644 .github/workflows/tests.yml create mode 100644 benchmarks/inversion.json create mode 100644 frame_pipeline.py create mode 100644 requirements-inversion.txt create mode 100644 tests/benchmark_inversion.py create mode 100644 tests/inversion_regression.py create mode 100644 timestamped_frames.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..630e389 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,30 @@ +name: tests + +on: + pull_request: + push: + branches: [main, 'feat/**'] + +permissions: + contents: read + +jobs: + regression: + runs-on: ubuntu-24.04 + timeout-minutes: 25 + env: + TMPDIR: ${{ runner.temp }} + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + - name: install ffmpeg and optional processors + run: | + sudo apt-get update + sudo apt-get install -y ffmpeg + python -m pip install -r requirements-collapse.txt -r requirements-inversion.txt ruff + - name: check changed processor code + run: ruff check --select F frame_pipeline.py timestamped_frames.py stateful_effects.py tests/inversion_regression.py + - name: run every suite with optional dependencies present + run: python tests/run_tests.py diff --git a/README.md b/README.md index 3a3f697..f9b89ed 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,11 @@ python3 -m pip install -r requirements-collapse.txt These packages are optional and loaded only when rendering stateful effects. Ordinary FFmpeg effects and `--list` still work without them. +Causality inversion uses NumPy and PyAV to carry RGB frames with their original +presentation timestamps. Install `requirements-inversion.txt`; its aligned mode +also needs `requirements-collapse.txt` for CPU optical flow. Timestamped export +uses FFmpeg's `-enc_time_base demux` option (tested with FFmpeg 7.1). + ## commands ```sh @@ -50,6 +55,7 @@ Ordinary FFmpeg effects and `--list` still work without them. ./effex.py sediment+deepfry input/clip.mp4 --set sediment.decay=30 --set sediment.color=warm --set deepfry.strength=60 ./effex.py collapse input/clip.mp4 --set intensity=80 --set seed=17 ./effex.py collapse input/clip.mp4 --set mode=seam --set seam_strength=80 +./effex.py inversion input/clip.mp4 --set lookahead=0.5 --set offset=0.3 --set region=bands ./effex.py datamosh input/a.mp4 --set second=input/b.mp4 --set cut=dice --set survival=50 --set seed=123 ./effex.py stretch input/clip.mp4 --dry-run ``` @@ -64,7 +70,7 @@ Ordinary FFmpeg effects and `--list` still work without them. - datamosh can join two videos in one codec stream. pick the second clip in the menu or pass `--set second=path`; set the trim, keyframe survival, seed, fill, grit, and audio. - single-pass previews, render progress, ctrl-c cancellation, and cleanup of partial renders. -datamosh and timescratch use their own multi-pass pipelines, so they cannot be chained with other effects. Temporal sediment uses a bounded 128 MiB history buffer. Stateful effects resample input to the probed frame rate, do not preserve original variable-rate timestamps, and have no preview yet. Multiple stateful processors must be adjacent; FFmpeg filters can go before and after them. +datamosh and timescratch use their own multi-pass pipelines, so they cannot be chained with other effects. Temporal sediment uses a bounded 128 MiB history buffer. Without inversion, stateful chains still normalize input to the probed frame rate. Chains containing inversion use timestamped transport and preserve the video presentation timeline, including variable-rate input and nonzero start times. Stateful previews are not available. Multiple stateful processors must be adjacent; FFmpeg filters can go before and after them. ## channel lobotomy @@ -168,6 +174,69 @@ averaged 6.68 ms at 360p and 12.87 ms at 720p. Persistent state was about 0.38 MiB for elastic and 0.53 MiB for seam mode in these cases. These are measurements of this fixture on this host, not a promise for every input. +## causality inversion + +`inversion` is under **TIME & MEMORY**. Selected regions borrow actual pixels +from a later decoded frame. It never predicts or fabricates future content. + +| option | range / values | default | +| --- | --- | --- | +| `lookahead` | 0-5 seconds of buffer horizon | 0.5 | +| `offset` | 0-lookahead seconds into the future | 0.5 | +| `strength` | 0-100%, opacity or replacement coverage | 65% | +| `mode` | blend, replace, tear, align | blend | +| `region` | full, bands, blocks, patch | full | +| `mask_speed` | 0-4 mask/tear cycles per second | 0.25 | +| `alignment` | 0-100%, flow gain in align mode | 100% | +| `feather` | 0-100%, region edge softness | 10% | +| `seed` | 0-2147483647; 0 derives masks from source | 0 | + +Blend mixes current and future RGB values. Replace uses strength as seeded +patch coverage, so half strength replaces some pixels rather than making all +pixels half-transparent. Tear shifts horizontal bands from the future image. +Align estimates reduced-resolution forward flow and samples the future image +at those coordinates to bring it toward the current scene. All modes can use +the same region masks and feathering; alignment only affects `mode=align`. + +The buffer adds up to `lookahead` seconds of video processing delay, plus one +decoded frame used to determine packet duration. Output timestamps do not move. +Live input cannot provide genuine future pixels without that buffering; the +current CLI accepts files, not live streams. A future request uses the first +available frame at or after the target time. Near EOF, or before a detected +scene cut, it clamps to the latest available frame in that segment. The final +frame therefore stays unchanged. Zero strength or zero offset passes through +without keeping a look-ahead queue. + +Each inversion queue is capped at 128 frames and 128 MiB of persistent state. +Transient RGB/float masks, alignment maps and encoder buffers are additional; +the cap is not a limit on total process RSS. +Impossible nominal-rate settings fail before decoding with a resize/lower-horizon +hint. An unusually dense VFR burst can fill the queue; its oldest frame then +uses the latest already-buffered future instead of exceeding the cap. Explicit +reset, rewind, frame discontinuity or a new clip clears pending state. Scene +cuts drain the old segment before starting the next, preventing future-scene +pixels from leaking backward across cuts. + +The shared renderer keeps original rational video PTS through NUT RGB pipes, +including chains with sediment or collapse in either adjacent order. Unfiltered +AAC/MP3/AC3/EAC3/ALAC/Opus audio is stream-copied to avoid encoder priming shifts. +Other audio codecs or chains with audio filters are encoded to AAC; encoder +delay can apply. Temporal FFmpeg filters such as slowmo still deliberately +change timestamps. Masks/flow use seconds for analysis; exported PTS stays exact. + +Tests cover early object appearance, mode differences, EOF/short/static clips, +scene boundaries, resets, bounded queues, caller ownership, VFR/nonzero-start +timestamps, copied audio packets, short pipe writes, cancellation and cleanup. +Run `python3 tests/run_tests.py inversion` for the focused suite. + +`python3 tests/benchmark_inversion.py --frames 60 --output benchmarks/inversion.json` +records each mode at 640x360 and 1280x720 in fresh workers. With a 0.5-second +horizon at 30 fps, measured queue state was about 9.89 MiB at 360p and 39.55 MiB +at 720p. At 720p, processor-only averages including EOF drain were 11.93 ms +blend, 15.10 ms replace, 24.32 ms tear and 23.10 ms align. These exclude +FFmpeg decode/encode and are not end-to-end real-time claims. The JSON records +peak process RSS separately from bounded processor state. + ## files and tests `input/` holds source clips. `output/` holds renders. Git ignores files in both folders except `.gitkeep`, so clips and renders stay local. Both paths can be changed with `--input-dir` / `--output-dir` or `EFFEX_INPUT_DIR` / `EFFEX_OUTPUT_DIR`. @@ -178,6 +247,7 @@ run the test suite with: python3 tests/run_tests.py python3 tests/run_tests.py psychedelic # pixel behavior, options and real renders python3 tests/run_tests.py collapse seams # deformation, carving and shared pipeline +python3 tests/run_tests.py inversion # future pixels, original PTS and EOF fallback ``` tests use generated clips and temporary folders; they do not overwrite files in `input/` or `output/`. diff --git a/ROADMAP.md b/ROADMAP.md index f3e971a..7e884c5 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -4,10 +4,10 @@ purpose: pick one item per session, finish it end-to-end (code + test + docs), then check it off with a done note saying what actually landed. format: `[ ]` open, `[x]` done, done notes indented under the item. state: input/ holds source videos, output/ holds renders and previews, effex.py -is the tool (v0.12.0, 23 effects, chains of two or more), tests/ has nine +is the tool (v0.12.0, 24 effects, chains of two or more), tests/ has ten suites. verify before closing an item with: - python3 tests/run_tests.py # cli + datamosh + menu + layout + install + stateful + psychedelic + collapse + seams + python3 tests/run_tests.py # cli + datamosh + menu + layout + install + stateful + psychedelic + collapse + seams + inversion python3 tests/make_clip.py # drop a lavfi test clip into input/ ./install.sh # effex onto ~/.local/bin @@ -350,7 +350,7 @@ decision. ### stateful glitch effects -Six effects from the requested design prompt. Temporal Sediment and Elastic Collapse are implemented; the remaining four are checklist items, not implementations. +Six effects from the requested design prompt. Temporal Sediment, Elastic Collapse and Causality Inversion are implemented; the remaining three are checklist items, not implementations. Shared rules for the stateful effects: - Trace how `effex.py` scans/probes clips, builds `Plan`s, invokes custom runners, and encodes output. The core stays FFmpeg-filter based; Temporal Sediment adds a narrow per-frame adapter for stateful effects. @@ -404,11 +404,23 @@ Shared rules for the stateful effects: - Reuse a shared flow field if the pipeline can safely provide one; otherwise estimate dense flow from previous/current grayscale frames at a configurable analysis resolution, starting with Farneback flow if OpenCV is accepted. Smooth and clamp vectors, apply a tunable gain, and remap the current frame. Verify the displacement sign with a synthetic moving shape before tuning the look. - Options: flow gain, nonlinear gain, motion threshold, analysis resolution, smoothing, max displacement, optional trail amount/decay, seed for randomized modulation. - Handle the first frame without history. Test direction/sign, stronger displacement at higher gain, relative stability on static footage, valid shape/dtype, and CPU cost. -- [ ] E2.49 - causality inversion (prompt effect 20): let selected regions borrow pixels from later video frames. +- [x] E2.49 - causality inversion (prompt effect 20): let selected regions borrow pixels from later video frames. - Use a bounded look-ahead buffer of frames and timestamps. Start with future-frame blend and replacement; add region tearing or motion alignment as separate modes. Preserve output timestamps and dimensions. - Document the look-ahead delay. Live input cannot provide genuine future frames without buffering or prediction. Near clip end, reduce the blend or use the latest available frame. - Options: look-ahead seconds, blend/replacement strength, offset, region/mask mode, mask speed, motion alignment, edge feather, seed. - Test that a moving synthetic object appears early by the configured offset, blend and replacement differ, unavailable future frames take the documented fallback, and memory/timestamps remain bounded and valid. + _Done: `inversion` adds bounded look-ahead with separate blend, replacement, + band-tearing and CPU-flow-aligned modes. Region masks, speed, feathering, + strength, horizon/offset, alignment and seed are configurable. Queues cap at + 128 frames/128 MiB, reject impossible nominal settings and clamp dense VFR + bursts/EOF/cuts to available future content. Streaming stages retain rational + PTS and dimensions through optional PyAV/NUT transport inside the existing + FFmpeg render lifecycle; compatible unfiltered audio is copied. Tests verify + early moving objects, distinct modes, ownership, resets/cuts, budget bounds, + EOF, exact VFR/nonzero-start PTS, audio packet timing, stage ordering, short + writes, failure/cancel cleanup and the menu. Separate 360p/720p benchmarks for + all four modes are saved in benchmarks/inversion.json. README documents + buffering latency, live-input limits, fallback and optional dependencies._ --- diff --git a/benchmarks/inversion.json b/benchmarks/inversion.json new file mode 100644 index 0000000..cb54109 --- /dev/null +++ b/benchmarks/inversion.json @@ -0,0 +1,143 @@ +{ + "recorded_utc": "2026-10-24T10:56:21.541926+00:00", + "platform": "Linux-7.1.13+parrot7-amd64-x86_64-with-glibc2.41", + "scope": "processor only, fresh worker per mode/size; includes EOF drain, excludes decode/encode", + "results": [ + { + "mode": "blend", + "width": 640, + "height": 360, + "frames": 60, + "mean_ms_including_eof": 3.0951608495039786, + "push_p95_ms": 5.521897197468206, + "eof_drain_ms": 30.323145001602825, + "peak_buffer_mib": 9.890716552734375, + "peak_process_rss_mib": 74.19921875, + "lookahead_seconds": 0.5, + "fps": 30, + "opencv_threads": 1, + "python": "3.14.0", + "numpy": "2.5.4", + "opencv": "5.0.0" + }, + { + "mode": "blend", + "width": 1280, + "height": 720, + "frames": 60, + "mean_ms_including_eof": 11.926763082859301, + "push_p95_ms": 12.431164700319641, + "eof_drain_ms": 163.31714000261854, + "peak_buffer_mib": 39.553802490234375, + "peak_process_rss_mib": 144.90234375, + "lookahead_seconds": 0.5, + "fps": 30, + "opencv_threads": 1, + "python": "3.14.0", + "numpy": "2.5.4", + "opencv": "5.0.0" + }, + { + "mode": "replace", + "width": 640, + "height": 360, + "frames": 60, + "mean_ms_including_eof": 3.4171822838591956, + "push_p95_ms": 4.974954102362972, + "eof_drain_ms": 40.226283999800216, + "peak_buffer_mib": 9.891693115234375, + "peak_process_rss_mib": 78.8984375, + "lookahead_seconds": 0.5, + "fps": 30, + "opencv_threads": 1, + "python": "3.14.0", + "numpy": "2.5.4", + "opencv": "5.0.0" + }, + { + "mode": "replace", + "width": 1280, + "height": 720, + "frames": 60, + "mean_ms_including_eof": 15.104414433032312, + "push_p95_ms": 18.992216050901334, + "eof_drain_ms": 198.32324899471132, + "peak_buffer_mib": 39.554779052734375, + "peak_process_rss_mib": 149.8671875, + "lookahead_seconds": 0.5, + "fps": 30, + "opencv_threads": 1, + "python": "3.14.0", + "numpy": "2.5.4", + "opencv": "5.0.0" + }, + { + "mode": "tear", + "width": 640, + "height": 360, + "frames": 60, + "mean_ms_including_eof": 6.345966033162161, + "push_p95_ms": 9.825213200383587, + "eof_drain_ms": 78.20343400089769, + "peak_buffer_mib": 9.890716552734375, + "peak_process_rss_mib": 74.515625, + "lookahead_seconds": 0.5, + "fps": 30, + "opencv_threads": 1, + "python": "3.14.0", + "numpy": "2.5.4", + "opencv": "5.0.0" + }, + { + "mode": "tear", + "width": 1280, + "height": 720, + "frames": 60, + "mean_ms_including_eof": 24.321544300498015, + "push_p95_ms": 29.347579599198067, + "eof_drain_ms": 317.7273420005804, + "peak_buffer_mib": 39.553802490234375, + "peak_process_rss_mib": 147.859375, + "lookahead_seconds": 0.5, + "fps": 30, + "opencv_threads": 1, + "python": "3.14.0", + "numpy": "2.5.4", + "opencv": "5.0.0" + }, + { + "mode": "align", + "width": 640, + "height": 360, + "frames": 60, + "mean_ms_including_eof": 8.36438881669892, + "push_p95_ms": 13.55484554587747, + "eof_drain_ms": 111.04231100034667, + "peak_buffer_mib": 9.890716552734375, + "peak_process_rss_mib": 80.078125, + "lookahead_seconds": 0.5, + "fps": 30, + "opencv_threads": 1, + "python": "3.14.0", + "numpy": "2.5.4", + "opencv": "5.0.0" + }, + { + "mode": "align", + "width": 1280, + "height": 720, + "frames": 60, + "mean_ms_including_eof": 23.098397633293644, + "push_p95_ms": 23.494914998445893, + "eof_drain_ms": 321.40718099981314, + "peak_buffer_mib": 39.553802490234375, + "peak_process_rss_mib": 154.203125, + "lookahead_seconds": 0.5, + "fps": 30, + "opencv_threads": 1, + "python": "3.14.0", + "numpy": "2.5.4", + "opencv": "5.0.0" + } + ] +} diff --git a/effex.py b/effex.py index 0b652bc..e880737 100755 --- a/effex.py +++ b/effex.py @@ -33,7 +33,7 @@ from pathlib import Path from typing import Callable, Optional, cast -from stateful_effects import ElasticCollapse, TemporalSediment +from stateful_effects import CausalityInversion, ElasticCollapse, TemporalSediment VERSION = "0.12.0" @@ -1958,6 +1958,50 @@ def build_collapse(ctx: EffectContext) -> Plan: ) +INVERSION_PARAMS = ( + Param("lookahead", "look-ahead secs", "buffer horizon in seconds; adds processing latency", + "float", 0.5, minimum=0, maximum=5, step=0.1), + Param("offset", "future offset", "seconds to borrow ahead; cannot exceed look-ahead", + "float", 0.5, minimum=0, maximum=5, step=0.1), + Param("strength", "strength", "blend opacity or replacement coverage", + "percent", 65, minimum=0, maximum=100, step=5), + Param("mode", "mode", "how future pixels enter the current image", "choice", "blend", + choices=(("blend", "blend", "mix current and future pixels"), + ("replace", "replace", "replace seeded regions instead of mixing"), + ("tear", "region tearing", "shift bands from the future frame"), + ("align", "motion aligned", "warp future toward current with CPU flow"))), + Param("region", "region mask", "which regions may borrow future pixels", "choice", "full", + choices=(("full", "full frame", "every pixel"), + ("bands", "bands", "moving horizontal stripes"), + ("blocks", "blocks", "moving checker regions"), + ("patch", "moving patch", "a drifting circular patch"))), + Param("mask_speed", "mask speed", "mask and tear cycles per second; 0 holds still", + "float", 0.25, minimum=0, maximum=4, step=0.1), + Param("alignment", "motion alignment", "flow gain in aligned mode; 0 disables alignment", + "percent", 100, minimum=0, maximum=100, step=5), + Param("feather", "edge feather", "soften region mask boundaries", + "percent", 10, minimum=0, maximum=100, step=5), + Param("seed", "random seed", "0 derives repeatable masks from the source", + "int", 0, minimum=0, maximum=2147483647, step=1), +) + + +def build_inversion(ctx: EffectContext) -> Plan: + lookahead = float(ctx.opts.get("lookahead", 0.5)) + offset = float(ctx.opts.get("offset", 0.5)) + if offset > lookahead: + raise ValueError("inversion offset must not exceed lookahead seconds") + width, height = _source_dims(ctx.clip, ctx.frame) + return Plan( + vf="stateful causality inversion timestamped frame pass", + out_suffix=ctx.suffix or "_INVERTED", width=width, height=height, + notes=[f"buffers up to {lookahead:g}s to borrow {offset:g}s into the future", + "preserves video PTS, dimensions and compatible unfiltered audio packets", + "EOF/cuts clamp to latest frame; 128 MiB/128-frame buffer budget", + "requires requirements-inversion.txt; aligned mode also needs headless OpenCV"], + ) + + EFFECTS = [ Effect(key="stretch", name="stretch", blurb="squash the frame into another shape", @@ -2074,6 +2118,14 @@ def build_collapse(ctx: EffectContext) -> Plan: "evolve content-aware elastic field or bounded low-resolution seam map", "smooth displacement, CPU remap, encode with later filters and audio", )), + Effect(key="inversion", name="causality inversion", + blurb="borrow real pixels from buffered future frames", + tag="bounded look-ahead with timestamp-preserving RGB transport", + build=build_inversion, params=INVERSION_PARAMS, + frame_processor=CausalityInversion, + stages=("decode timestamped RGB frames through earlier FFmpeg filters", + "buffer future frames, mix selected regions, clamp unavailable future at EOF", + "encode on original video PTS with later filters and source audio")), Effect(key="sediment", name="temporal sediment", blurb="deposit frames into bounded, decaying visual history", tag="stateful Python frame processor", @@ -2097,7 +2149,7 @@ def build_collapse(ctx: EffectContext) -> Plan: "lobotomy", "neon", )), ("TIME & MEMORY", ( - "trails", "timescratch", "sediment", "dropout", + "trails", "timescratch", "sediment", "dropout", "inversion", )), ("CODEC & BLOCKS", ( "datamosh", "blockswap", "crunch", @@ -2363,6 +2415,10 @@ def apply_cli_opts(chain: Chain, args) -> tuple: if effect is None: return opts, (f"{chain.name} has no shape option for --size") opts[effect.key][shape_key(effect)] = f"{args.width}x{args.height}" + if "inversion" in opts: + mine = opts["inversion"] + if mine["offset"] > mine["lookahead"]: + return opts, "inversion offset must not exceed lookahead seconds" return opts, "" @@ -2551,6 +2607,13 @@ def run_stateful_chain(chain: Chain, ctx: EffectContext, clip: Clip, tools: Tool except Exception as exc: return False, f"could not initialize {chain.name}: {exc}", 0.0 + timestamped = any(getattr(p, "requires_timestamps", False) for p in processors) + if timestamped: + try: + import av # noqa: F401 - dependency check before launching FFmpeg + except ImportError: + return False, "Causality Inversion needs requirements-inversion.txt (PyAV)", 0.0 + prefix = ",".join(p.vf for e, p in plans[:first] if p.vf and e.frame_processor is None) suffix = ",".join(p.vf for e, p in plans[last + 1:] @@ -2567,6 +2630,15 @@ def run_stateful_chain(chain: Chain, ctx: EffectContext, clip: Clip, tools: Tool "-f", "rawvideo", "-pixel_format", "rgb24", "-video_size", f"{width}x{height}", "-framerate", fps_text, "-i", "pipe:0"] + if timestamped: + decode_cmd = [tools.ffmpeg, "-hide_banner", "-nostdin", "-loglevel", "error", + "-nostats", "-copyts", "-i", str(clip.path), "-map", "0:v:0", + "-vf", ",".join(([prefix] if prefix else []) + ["format=rgb24"]), + "-an", "-c:v", "rawvideo", "-enc_time_base", "demux", + "-fps_mode", "passthrough", "-avoid_negative_ts", "disabled", + "-f", "nut", "pipe:1"] + encode_cmd = [tools.ffmpeg, "-hide_banner", "-nostdin", "-y", "-copyts", + "-f", "nut", "-i", "pipe:0"] if clip.audio: encode_cmd += ["-i", str(clip.path)] encode_cmd += merged.in_extra @@ -2576,14 +2648,19 @@ def run_stateful_chain(chain: Chain, ctx: EffectContext, clip: Clip, tools: Tool if clip.audio: if merged.af: encode_cmd += ["-af", merged.af] - encode_cmd += ["-map", "1:a:0?", "-c:a", "aac", "-b:a", "192k", - "-ac", "2"] + if timestamped and not merged.af and clip.acodec in ("aac", "mp3", "ac3", "eac3", "alac", "opus"): + encode_cmd += ["-map", "1:a:0?", "-c:a", "copy"] + else: + encode_cmd += ["-map", "1:a:0?", "-c:a", "aac", "-b:a", "192k", "-ac", "2"] else: encode_cmd += ["-an"] encode_cmd += ["-c:v", "libx264", "-preset", ctx.preset] if not merged.abr: encode_cmd += ["-crf", str(ctx.crf)] encode_cmd += ["-pix_fmt", "yuv420p", "-movflags", "+faststart"] + if timestamped: + encode_cmd += ["-fps_mode", "passthrough", "-enc_time_base", "demux", + "-avoid_negative_ts", "disabled"] encode_cmd += merged.out_extra encode_cmd += ["-progress", "pipe:1", "-nostats", "-loglevel", "error", str(dst)] @@ -2617,6 +2694,7 @@ def feed_frames(encoder_input, stop_event) -> None: def drain_decoder(pipe): for line in iter(pipe.readline, b""): decoder_errors.append(line.decode("utf-8", "replace").rstrip()) + del decoder_errors[:-40] try: decoder = subprocess.Popen(decode_cmd, stdout=subprocess.PIPE, @@ -2628,8 +2706,14 @@ def drain_decoder(pipe): target=drain_decoder, args=(decode_err,), daemon=True) error_thread.start() index = 0 + if timestamped: + from frame_pipeline import process_packets + from timestamped_frames import read_packets, write_packets + packets = read_packets(decode_out, {"clip": clip, "fps": fps}, stop_event) + index = write_packets(process_packets(packets, processors), encoder_input, + (height, width, 3), fps, stop_event) frame_bytes = width * height * 3 - while not stop_event.is_set(): + while not timestamped and not stop_event.is_set(): raw = read_frame(decode_out, frame_bytes) if stop_event.is_set(): break @@ -2679,6 +2763,8 @@ def drain_decoder(pipe): detail += "\n" + "\n".join(decoder_errors) raise RuntimeError(detail) from exc finally: + for processor in processors: + processor.reset() if decoder is not None: for pipe in (decoder.stdin, decoder.stdout, decoder.stderr): if pipe is not None and not pipe.closed: diff --git a/frame_pipeline.py b/frame_pipeline.py new file mode 100644 index 0000000..e0b9846 --- /dev/null +++ b/frame_pipeline.py @@ -0,0 +1,37 @@ +"""Timestamp-carrying packets and composable streaming frame stages. + +No numeric/video dependencies are imported until a processor needs them. +""" +from dataclasses import dataclass, field, replace +from fractions import Fraction +from typing import Any + + +@dataclass(frozen=True) +class FramePacket: + frame: Any + index: int + timestamp: Fraction + time_base: Fraction + duration: Fraction = Fraction(0) + context: dict = field(default_factory=dict) + + +def _stage(packets, processor): + if hasattr(processor, "push"): + for packet in packets: + yield from processor.push(packet) + yield from processor.flush() + else: + for packet in packets: + result = processor.process(packet.frame, packet.index, + float(packet.timestamp), packet.context) + yield replace(packet, frame=result) + + +def process_packets(packets, processors): + """Compose streaming and immediate processors, draining each stage at EOF.""" + stream = packets + for processor in processors: + stream = _stage(stream, processor) + yield from stream diff --git a/requirements-inversion.txt b/requirements-inversion.txt new file mode 100644 index 0000000..75e4f19 --- /dev/null +++ b/requirements-inversion.txt @@ -0,0 +1,3 @@ +-r requirements-stateful.txt +# Timestamped NUT transport; aligned mode additionally needs requirements-collapse.txt. +av>=14 diff --git a/stateful_effects.py b/stateful_effects.py index 3d404fa..b702812 100644 --- a/stateful_effects.py +++ b/stateful_effects.py @@ -1,5 +1,7 @@ """Bounded, stateful frame processors used by effex's raw-frame runner.""" import math +from collections import deque +from dataclasses import replace from typing import Any, Optional @@ -485,3 +487,211 @@ def process(self, frame, frame_index: int, timestamp: Optional[float], map_y += np.arange(height, dtype=np.float32)[:, None] return cv.remap(frame, map_x, map_y, interpolation=cv.INTER_LINEAR, borderMode=cv.BORDER_REFLECT_101) + + +class CausalityInversion: + """Bounded future-pixel borrowing; emit each packet on its original timeline.""" + + requires_timestamps = True + MAX_STATE_BYTES: int = 128 * 1024 * 1024 + MAX_FRAMES = 128 + DEFAULTS = { + "lookahead": 0.5, "offset": 0.5, "strength": 65, "mode": "blend", + "region": "full", "mask_speed": 0.25, "alignment": 100, + "feather": 10, "seed": 0, + } + + def __init__(self): + self._np: Any = None + self._config = {} + self._shape = (0, 0, 0) + self._queue = deque() + self._seed = 0 + self._capacity = 0 + self._next_index = None + self._last_timestamp = None + self._scene_sample: Any = None + self._coverage: Any = None + self.capacity_fallbacks = 0 + + def initialize(self, config: dict, frame_shape: tuple, fps: float, seed: int) -> None: + try: + import numpy as np + except ImportError as exc: + raise RuntimeError("Causality Inversion needs requirements-inversion.txt") from exc + if (len(frame_shape) != 3 or frame_shape[2] != 3 + or any(int(v) < 1 for v in frame_shape)): + raise ValueError("Causality Inversion expects an HxWx3 RGB frame") + self._np = np + self._shape = tuple(int(v) for v in frame_shape) + self._config = dict(self.DEFAULTS) + for key, default in self.DEFAULTS.items(): + if key in ("mode", "region"): + value = str(config.get(key, default)) + choices = (("blend", "replace", "tear", "align") if key == "mode" + else ("full", "bands", "blocks", "patch")) + if value not in choices: + raise ValueError(f"invalid inversion {key}: {value}") + self._config[key] = value + else: + high = (5 if key in ("lookahead", "offset") else + 4 if key == "mask_speed" else + 2147483647 if key == "seed" else 100) + value = ElasticCollapse._number(config.get(key, default), default, 0, high) + self._config[key] = (value if key in ("lookahead", "offset", "mask_speed") + else int(value)) + if self._config["offset"] > self._config["lookahead"]: + raise ValueError("inversion offset must not exceed lookahead seconds") + if self._config["mode"] == "align" and self._config["alignment"]: + try: + import cv2 # noqa: F401 + except ImportError as exc: + raise RuntimeError("Aligned inversion needs requirements-collapse.txt") from exc + self._seed = int(ElasticCollapse._number(seed, 0, 0, 2147483647)) + frame_bytes = math.prod(self._shape) + self._capacity = min(self.MAX_FRAMES, (self.MAX_STATE_BYTES - 65536) // frame_bytes) + delay = self._config["lookahead"] if self._config["strength"] and self._config["offset"] else 0 + rate = ElasticCollapse._number(fps, 30.0, 0.001, 1000) + needed = math.ceil(delay * rate) + 1 + if needed > self._capacity: + raise ValueError( + f"inversion lookahead needs {needed} frames but the 128 MiB/128-frame " + f"budget fits {self._capacity}; lower lookahead or resize before inversion" + ) + self.reset() + + @property + def state_nbytes(self) -> int: + total = sum(packet.frame.nbytes for packet in self._queue) + return int(total + sum(a.nbytes for a in (self._scene_sample, self._coverage) + if a is not None)) + + @property + def buffered_frames(self) -> int: + return len(self._queue) + + @property + def capacity(self) -> int: + return self._capacity + + def serialize_config(self) -> dict: + return dict(self._config) + + def reset(self) -> None: + self._queue.clear() + self._next_index = None + self._last_timestamp = None + self._scene_sample = None + self._coverage = None + self.capacity_fallbacks = 0 + + def _mask(self, timestamp): + np = self._np + height, width = self._shape[:2] + y = np.arange(height, dtype=np.float32)[:, None] / max(1, height - 1) + x = np.arange(width, dtype=np.float32)[None, :] / max(1, width - 1) + phase = (self._seed % 997) / 997 * 2 * math.pi + t = float(timestamp) * self._config["mask_speed"] * 2 * math.pi + region = self._config["region"] + if region == "full": + return np.ones((height, width), np.float32) + if region == "bands": + signed = np.broadcast_to(np.sin(y * 8 * math.pi + t + phase), (height, width)) + elif region == "blocks": + signed = np.sin(x * 8 * math.pi + t + phase) * np.sin(y * 6 * math.pi - t) + else: + cx, cy = 0.5 + 0.23 * math.sin(t + phase), 0.5 + 0.23 * math.cos(t + phase) + signed = 0.25 - np.sqrt((x - cx) ** 2 + (y - cy) ** 2) + feather = self._config["feather"] / 100.0 + return (np.clip(0.5 + signed / (2 * feather), 0, 1).astype(np.float32) + if feather else (signed >= 0).astype(np.float32)) + + def _borrow(self, current, future): + np = self._np + mode = self._config["mode"] + height, width = self._shape[:2] + if mode == "tear": + rows = np.arange(height)[:, None] + phase = float(current.timestamp) * self._config["mask_speed"] * 2 * math.pi + offsets = np.rint(np.sin(rows // max(1, height // 8) + phase + + self._seed) * max(1, width // 8)).astype(np.int32) + return future.frame[rows, (np.arange(width)[None, :] + offsets) % width] + if mode != "align" or not self._config["alignment"] or min(height, width) < 8: + return future.frame + import cv2 as cv + scale = min(1.0, 192 / max(height, width)) + size = (max(8, round(width * scale)), max(8, round(height * scale))) + a = cv.cvtColor(cv.resize(current.frame, size), cv.COLOR_RGB2GRAY) + b = cv.cvtColor(cv.resize(future.frame, size), cv.COLOR_RGB2GRAY) + flow = cv.calcOpticalFlowFarneback(a, b, None, 0.5, 3, 21, 5, 7, 1.5, 0) + flow = cv.resize(flow, (width, height)) + gain = self._config["alignment"] / 100.0 + mx = np.arange(width, dtype=np.float32)[None, :] + np.clip(flow[:, :, 0] * width / size[0], -width / 4, width / 4) * gain + my = np.arange(height, dtype=np.float32)[:, None] + np.clip(flow[:, :, 1] * height / size[1], -height / 4, height / 4) * gain + return cv.remap(future.frame, mx, my, cv.INTER_LINEAR, borderMode=cv.BORDER_REFLECT_101) + + def _mix(self, current, future): + np = self._np + strength = self._config["strength"] / 100.0 + if strength == 0 or current.timestamp == future.timestamp: + return replace(current, frame=current.frame.copy()) + borrowed = self._borrow(current, future) + mask = self._mask(current.timestamp) + if self._config["mode"] == "replace": + if self._coverage is None: + self._coverage = np.random.default_rng(self._seed).random((16, 16)).astype(np.float32) + height, width = self._shape[:2] + rows = np.arange(height) * 16 // height + cols = np.arange(width) * 16 // width + alpha = (self._coverage[rows[:, None], cols[None, :]] < strength).astype(np.float32) + alpha = (alpha * mask)[:, :, None] + else: + alpha = (strength * mask)[:, :, None] + result = current.frame.astype(np.float32) * (1 - alpha) + borrowed.astype(np.float32) * alpha + return replace(current, frame=np.ascontiguousarray(np.clip(result, 0, 255).round().astype(np.uint8))) + + def _emit_oldest(self): + current = self._queue.popleft() + target = float(current.timestamp) + self._config["offset"] + future = next((p for p in self._queue if float(p.timestamp) + 1e-9 >= target), + self._queue[-1] if self._queue else current) + return self._mix(current, future) + + def push(self, packet): + if self._np is None: + raise RuntimeError("initialize Causality Inversion before buffering") + np = self._np + if (not isinstance(packet.frame, np.ndarray) or packet.frame.shape != self._shape + or packet.frame.dtype != np.uint8): + raise ValueError(f"expected uint8 RGB frame {self._shape}") + stamp = float(packet.timestamp) + if not math.isfinite(stamp): + raise ValueError("inversion needs finite presentation timestamps") + if ((self._next_index is not None and packet.index != self._next_index) + or (self._last_timestamp is not None and stamp < self._last_timestamp)): + self.reset() + sample = packet.frame[::max(1, self._shape[0] // 32), + ::max(1, self._shape[1] // 32)].copy() + cut = (self._scene_sample is not None and + float(np.abs(sample.astype(np.float32) - self._scene_sample.astype(np.float32)).mean()) >= 140.25) + if cut or packet.context.get("scene_cut"): + yield from self.flush() + self._scene_sample = sample + self._next_index = packet.index + 1 + self._last_timestamp = stamp + if self._config["strength"] == 0 or self._config["offset"] == 0: + yield replace(packet, frame=packet.frame.copy()) + return + while len(self._queue) >= self._capacity: + self.capacity_fallbacks += 1 + yield self._emit_oldest() + self._queue.append(replace(packet, frame=packet.frame.copy())) + while self._queue and stamp - float(self._queue[0].timestamp) + 1e-9 >= self._config["lookahead"]: + yield self._emit_oldest() + + def flush(self): + """EOF/cut: clamp unavailable future requests to the last buffered frame.""" + while self._queue: + yield self._emit_oldest() + self._scene_sample = None + self._coverage = None diff --git a/tests/benchmark_inversion.py b/tests/benchmark_inversion.py new file mode 100644 index 0000000..2ea31aa --- /dev/null +++ b/tests/benchmark_inversion.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 +"""Benchmark bounded inversion at 360p/720p; no FFmpeg time is included.""" +import argparse +import json +import platform +import resource +import subprocess +import sys +import time +from datetime import datetime, timezone +from fractions import Fraction +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO)) + + +def worker(mode, width, height, frames): + import cv2 + import numpy as np + from frame_pipeline import FramePacket + from stateful_effects import CausalityInversion + cv2.setNumThreads(1) + y, x = np.indices((height, width)) + source = np.stack((x % 256, y % 256, ((x // 24 + y // 24) % 2) * 180), axis=-1).astype(np.uint8) + del x, y + processor = CausalityInversion() + processor.initialize({"mode": mode}, source.shape, 30, 17) + timings, buffered, outputs = [], 0, 0 + for i in range(frames): + packet = FramePacket(np.roll(source, i * 2, axis=1), i, Fraction(i, 30), + Fraction(1, 30), Fraction(1, 30)) + start = time.perf_counter() + for result in processor.push(packet): + assert result.frame.shape == source.shape + outputs += 1 + timings.append((time.perf_counter() - start) * 1000) + buffered = max(buffered, processor.state_nbytes) + start = time.perf_counter() + for _ in processor.flush(): + outputs += 1 + drain_ms = (time.perf_counter() - start) * 1000 + assert outputs == frames and processor.state_nbytes == 0 + return {"mode": mode, "width": width, "height": height, "frames": frames, + "mean_ms_including_eof": (sum(timings) + drain_ms) / frames, + "push_p95_ms": float(np.percentile(timings, 95)), "eof_drain_ms": drain_ms, + "peak_buffer_mib": buffered / 1048576, + "peak_process_rss_mib": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024, + "lookahead_seconds": 0.5, "fps": 30, "opencv_threads": 1, + "python": platform.python_version(), "numpy": np.__version__, "opencv": cv2.__version__} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--frames", type=int, default=60) + parser.add_argument("--output", type=Path) + parser.add_argument("--worker", choices=("blend", "replace", "tear", "align")) + parser.add_argument("--width", type=int, default=640) + parser.add_argument("--height", type=int, default=360) + args = parser.parse_args() + if args.frames < 1: + parser.error("--frames must be positive") + if args.worker: + print(json.dumps(worker(args.worker, args.width, args.height, args.frames))) + return + rows = [] + for mode in ("blend", "replace", "tear", "align"): + for width, height in ((640, 360), (1280, 720)): + result = subprocess.run([sys.executable, __file__, "--worker", mode, "--width", str(width), + "--height", str(height), "--frames", str(args.frames)], + text=True, capture_output=True, check=True) + row = json.loads(result.stdout) + rows.append(row) + print(f"{mode:7} {width}x{height}: {row['mean_ms_including_eof']:.2f} ms/frame, " + f"{row['peak_buffer_mib']:.2f} MiB state, {row['peak_process_rss_mib']:.1f} MiB RSS") + assert len(rows) == 8 + report = {"recorded_utc": datetime.now(timezone.utc).isoformat(), "platform": platform.platform(), + "scope": "processor only, fresh worker per mode/size; includes EOF drain, excludes decode/encode", + "results": rows} + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(report, indent=2) + "\n") + else: + print(json.dumps(report, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/tests/inversion_regression.py b/tests/inversion_regression.py new file mode 100644 index 0000000..18a7808 --- /dev/null +++ b/tests/inversion_regression.py @@ -0,0 +1,342 @@ +#!/usr/bin/env python3 +"""Causality inversion: real future pixels, bounded queues and timestamped export.""" +import json +import importlib.util +import sys +import tempfile +import unittest +from fractions import Fraction +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +try: + import numpy as np +except ImportError: + print(" SKIP inversion tests need requirements-inversion.txt") + sys.exit(0) + +import stateful_effects +from harness import EFFEX, audio_codec, env_for, make_clip, run, video_dims + + +class InversionTests(unittest.TestCase): + def processor(self, shape=(24, 32, 3), seed=17, **config): + self.assertTrue(hasattr(stateful_effects, "CausalityInversion"), + "CausalityInversion must be implemented") + p = stateful_effects.CausalityInversion() + p.initialize(config, shape, 10, seed) + return p + + @staticmethod + def packet(frame, index, timestamp=None): + from frame_pipeline import FramePacket + return FramePacket(frame, index, Fraction(index, 10) if timestamp is None else timestamp, + Fraction(1, 1000), Fraction(1, 10)) + + def test_moving_object_appears_early_without_shifting_output_times(self): + p = self.processor(lookahead=0.2, offset=0.2, strength=100, + mode="replace", region="full", feather=0) + frames = [] + output = [] + for i in range(6): + frame = np.zeros((24, 32, 3), np.uint8) + frame[8:14, 2 + i * 4:6 + i * 4] = 255 + frames.append(frame.copy()) + output.extend(p.push(self.packet(frame, i))) + output.extend(p.flush()) + self.assertEqual([p.index for p in output], list(range(6))) + self.assertEqual([p.timestamp for p in output], [Fraction(i, 10) for i in range(6)]) + np.testing.assert_array_equal(output[0].frame, frames[2]) + np.testing.assert_array_equal(output[1].frame, frames[3]) + np.testing.assert_array_equal(output[-2].frame, frames[-1]) + np.testing.assert_array_equal(output[-1].frame, frames[-1]) + self.assertEqual(p.state_nbytes, 0) + + def test_blend_and_replace_have_different_strength_semantics(self): + results = [] + for mode in ("blend", "replace"): + p = self.processor(lookahead=0.1, offset=0.1, strength=50, mode=mode, + region="full", feather=0) + first = np.zeros((24, 32, 3), np.uint8) + future = np.full_like(first, 100) + self.assertEqual(list(p.push(self.packet(first, 0))), []) + output = list(p.push(self.packet(future, 1))) + results.append(output[0].frame) + self.assertEqual(set(np.unique(results[0])), {50}) + self.assertEqual(set(np.unique(results[1])), {0, 100}) + + def test_tearing_is_a_separate_future_pixel_mode(self): + frame = np.zeros((24, 32, 3), np.uint8) + future = np.tile(np.arange(32, dtype=np.uint8)[None, :, None], (24, 1, 3)) + p = self.processor(lookahead=0.1, offset=0.1, strength=100, mode="tear", + region="full", feather=0) + list(p.push(self.packet(frame, 0))) + out = list(p.push(self.packet(future, 1)))[0].frame + self.assertFalse(np.array_equal(out, future)) + self.assertTrue(set(np.unique(out)).issubset(set(np.unique(future)))) + + def test_regions_feather_seed_and_mask_speed(self): + first = np.zeros((24, 32, 3), np.uint8) + future = np.full_like(first, 100) + outputs = [] + for region in ("full", "bands", "blocks", "patch"): + p = self.processor(lookahead=0.1, offset=0.1, strength=100, + mode="blend", region=region, feather=0) + list(p.push(self.packet(first, 0))) + out = list(p.push(self.packet(future, 1)))[0].frame + outputs.append(out) + if region != "full": + self.assertEqual(set(np.unique(out)), {0, 100}) + for i in range(len(outputs)): + for j in range(i): + self.assertFalse(np.array_equal(outputs[i], outputs[j])) + p = self.processor(lookahead=0.1, offset=0.1, strength=100, + region="bands", feather=50) + list(p.push(self.packet(first, 0))) + softened = list(p.push(self.packet(future, 1)))[0].frame + self.assertTrue(np.any((softened > 0) & (softened < 100))) + a = self.processor(lookahead=0.1, offset=0.1, region="patch", mask_speed=1) + b = self.processor(lookahead=0.1, offset=0.1, region="patch", mask_speed=1) + c = self.processor(lookahead=0.1, offset=0.1, region="patch", mask_speed=1, seed=99) + collected = [[], [], []] + for i in range(10): + source = first if i % 2 == 0 else future + for index, processor in enumerate((a, b, c)): + collected[index].extend(processor.push(self.packet(source, i))) + self.assertTrue(all(np.array_equal(x.frame, y.frame) + for x, y in zip(collected[0], collected[1]))) + self.assertTrue(any(not np.array_equal(x.frame, y.frame) + for x, y in zip(collected[0], collected[2]))) + + def test_alignment_warps_future_toward_current_with_correct_sign(self): + try: + import cv2 + except ImportError: + self.skipTest("aligned mode needs requirements-collapse.txt") + rng = np.random.default_rng(1) + current = cv2.GaussianBlur(rng.integers(0, 180, (64, 96, 3), dtype=np.uint8), (3, 3), 0) + future = np.roll(current, 4, axis=1) + errors = [] + for mode in ("blend", "align"): + p = self.processor(shape=current.shape, lookahead=0.1, offset=0.1, + strength=100, mode=mode, alignment=100, feather=0) + list(p.push(self.packet(current, 0))) + out = list(p.push(self.packet(future, 1)))[0].frame + errors.append(float(np.abs(out[8:-8, 8:-8].astype(float) + - current[8:-8, 8:-8]).mean())) + self.assertLess(errors[1], errors[0] * 0.6, errors) + + def test_ownership_passthrough_reset_and_buffer_budget(self): + shape = (24, 32, 3) + source = np.zeros(shape, np.uint8) + p = self.processor(lookahead=0.2, offset=0.2, strength=100, mode="replace") + list(p.push(self.packet(source, 0))) + source[:] = 100 + list(p.push(self.packet(source, 1))) + p.reset() + self.assertEqual(p.buffered_frames, 0) + self.assertEqual(p.state_nbytes, 0) + for config in (dict(strength=0), dict(lookahead=0, offset=0)): + p = self.processor(**config) + output = list(p.push(self.packet(source, 0))) + np.testing.assert_array_equal(output[0].frame, source) + self.assertFalse(np.shares_memory(output[0].frame, source)) + p = self.processor(lookahead=0.2, offset=0.2) + p.MAX_STATE_BYTES = 65536 + source.nbytes * 5 + p.initialize(dict(lookahead=0.2, offset=0.2), shape, 10, 17) + output = [] + for i in range(100): + output.extend(p.push(self.packet(source, i, Fraction(i, 1000)))) + self.assertLessEqual(p.buffered_frames, 5) + self.assertLessEqual(p.state_nbytes, p.MAX_STATE_BYTES) + self.assertGreater(p.capacity_fallbacks, 0) + output.extend(p.flush()) + self.assertEqual([q.index for q in output], list(range(100))) + self.assertEqual(p.state_nbytes, 0) + with self.assertRaisesRegex(ValueError, "budget"): + self.processor(shape=(2160, 3840, 3), lookahead=1, offset=1) + with self.assertRaisesRegex(ValueError, "offset"): + self.processor(lookahead=0.1, offset=0.5) + + def test_irregular_timestamps_eof_and_metadata(self): + p = self.processor(lookahead=0.2, offset=0.2, strength=100, mode="replace") + stamps = [Fraction(n, 100) for n in (0, 3, 11, 27, 31)] + output, frames = [], [] + for i, stamp in enumerate(stamps): + frame = np.full((24, 32, 3), i * 20, np.uint8) + frames.append(frame.copy()) + output.extend(p.push(self.packet(frame, i, stamp))) + frame[:] = 255 # buffered snapshots must not alias the caller + output.extend(p.flush()) + self.assertEqual([q.timestamp for q in output], stamps) + self.assertEqual([q.duration for q in output], [Fraction(1, 10)] * 5) + np.testing.assert_array_equal(output[0].frame, frames[3]) + np.testing.assert_array_equal(output[-2].frame, frames[-1]) + p = self.processor() + self.assertEqual(list(p.flush()), []) + lone = self.packet(frames[0], 0) + self.assertEqual(list(p.push(lone)), []) + np.testing.assert_array_equal(list(p.flush())[0].frame, frames[0]) + + def test_scene_cuts_do_not_borrow_across_scenes(self): + from dataclasses import replace + p = self.processor(lookahead=0.2, offset=0.2, strength=100, mode="replace") + black = np.zeros((24, 32, 3), np.uint8) + white = np.full_like(black, 255) + list(p.push(self.packet(black, 0))) + out = list(p.push(replace(self.packet(white, 1), context={"scene_cut": True}))) + np.testing.assert_array_equal(out[0].frame, black) + p.reset() + for index in range(3): + list(p.push(self.packet(black, index))) + # An explicit rewind discards the previous timeline, rather than leaking it. + self.assertEqual(list(p.push(self.packet(white, 0))), []) + self.assertEqual([q.index for q in p.flush()], [0]) + + def test_streaming_stage_composition_preserves_packet_metadata(self): + from frame_pipeline import process_packets + from stateful_effects import TemporalSediment + packets = [self.packet(np.full((24, 32, 3), i * 20, np.uint8), i) for i in range(5)] + for reverse in (False, True): + inversion = self.processor(lookahead=0.1, offset=0.1) + sediment = TemporalSediment() + sediment.initialize({"intensity": 0}, (24, 32, 3), 10, 17) + stages = [inversion, sediment] if not reverse else [sediment, inversion] + output = list(process_packets(iter(packets), stages)) + self.assertEqual([q.timestamp for q in output], [q.timestamp for q in packets]) + self.assertEqual(len(output), len(packets)) + + @unittest.skipUnless(importlib.util.find_spec("av"), "install requirements-inversion.txt") + def test_real_cli_preserves_vfr_pts_and_render_modes(self): + import effex + self.assertTrue("inversion" in effex.EFFECT_BY_KEY, "inversion must be registered") + with tempfile.TemporaryDirectory(prefix="effex-inversion-") as directory: + root = Path(directory) + out = root / "out" + out.mkdir() + env = env_for(root, out) + clip = root / "av.mp4" + self.assertTrue(make_clip(clip, seconds=1, size="160x90")) + for setting in ("lookahead=-1", "offset=6", "strength=101", "mode=bad", + "region=bad", "mask_speed=5", "alignment=-1", "feather=101", + "seed=-1", "lookahead=0.1"): + invalid = run([EFFEX, "inversion", clip, "--set", setting, "--dry-run"], env_extra=env) + self.assertEqual(invalid.returncode, 2, invalid.stderr.decode()) + self.assertNotIn(b"Traceback", invalid.stderr) + modes = ["blend", "replace", "tear"] + if importlib.util.find_spec("cv2"): + modes.append("align") + for mode in modes: + proc = run([EFFEX, "inversion", clip, "--set", f"mode={mode}", + "--set", "region=bands", "--json", "--preset", "ultrafast"], env_extra=env) + self.assertEqual(proc.returncode, 0, proc.stderr.decode()) + data = json.loads(proc.stdout) + self.assertEqual(video_dims(data["output"]), (160, 90)) + self.assertEqual(audio_codec(data["output"]), "aac") + for chain in ("neon+inversion+deepfry", "inversion+sediment", + "sediment+inversion", "collapse+inversion", "inversion+collapse", + "stretch+inversion"): + if "collapse" in chain and importlib.util.find_spec("cv2") is None: + continue + args = [EFFEX, chain, clip, "--json", "--preset", "ultrafast"] + if chain.startswith("stretch"): + args += ["--size", "128x72"] + result = run(args, env_extra=env) + self.assertEqual(result.returncode, 0, result.stderr.decode()) + data = json.loads(result.stdout) + self.assertEqual(video_dims(data["output"]), (128, 72) if chain.startswith("stretch") else (160, 90)) + def audio_pts(path): + result = run(["ffprobe", "-v", "error", "-select_streams", "a:0", + "-show_entries", "packet=pts_time", "-of", "json", path]) + return [p["pts_time"] for p in json.loads(result.stdout)["packets"]] + copied = run([EFFEX, "inversion", clip, "--json"], env_extra=env) + self.assertEqual(audio_pts(clip), audio_pts(json.loads(copied.stdout)["output"])) + for start in (0, 5): + vfr = root / f"vfr-{start}.mkv" + made = run(["ffmpeg", "-v", "error", "-y", "-copyts", "-f", "lavfi", "-i", + "testsrc2=s=160x90:r=10:d=1.5", "-vf", + f"select='not(eq(mod(n,4),3))',setpts=PTS+{start}/TB", + "-fps_mode", "vfr", "-c:v", "ffv1", vfr]) + self.assertEqual(made.returncode, 0, made.stderr.decode()) + proc = run([EFFEX, "inversion", vfr, "--json", "--preset", "ultrafast"], env_extra=env) + self.assertEqual(proc.returncode, 0, proc.stderr.decode()) + destination = json.loads(proc.stdout)["output"] + def pts(path): + result = run(["ffprobe", "-v", "error", "-select_streams", "v:0", + "-show_entries", "frame=pts_time", "-of", "json", path]) + return [Fraction(f["pts_time"]) for f in json.loads(result.stdout)["frames"]] + self.assertEqual(pts(destination), pts(vfr)) + + def test_nut_roundtrip_short_writes_and_timestamp_guards(self): + import io + import threading + try: + import av # noqa: F401 + except ImportError: + self.skipTest("timestamped transport needs requirements-inversion.txt") + from timestamped_frames import _PipeWriter, read_packets, write_packets + event = threading.Event() + class ShortWriter(io.BytesIO): + def write(self, data): + return super().write(bytes(data[:7])) + destination = ShortWriter() + writer = _PipeWriter(destination, event) + self.assertEqual(writer.write(b"abcdefghijklmnopqrstuvwxyz"), 26) + self.assertEqual(destination.getvalue(), b"abcdefghijklmnopqrstuvwxyz") + event.set() + with self.assertRaises(InterruptedError): + writer.write(b"cancelled") + event.clear() + packets = [self.packet(np.full((24, 32, 3), i * 20, np.uint8), i, Fraction(n, 1000)) + for i, n in enumerate((5000, 5033, 5117, 5200))] + destination = ShortWriter() + self.assertEqual(write_packets(iter(packets), destination, (24, 32, 3), 30, event), 4) + source = io.BytesIO(destination.getvalue()) + restored = list(read_packets(source, {}, event)) + self.assertEqual([q.timestamp for q in restored], [q.timestamp for q in packets]) + for a, b in zip(restored, packets): + np.testing.assert_array_equal(a.frame, b.frame) + with self.assertRaisesRegex(ValueError, "non-monotonic"): + write_packets(iter([packets[1], packets[0]]), io.BytesIO(), (24, 32, 3), 30, event) + proc = run([sys.executable, "-S", EFFEX, "--list", "--json"]) + self.assertEqual(proc.returncode, 0, proc.stderr.decode()) + + + @unittest.skipUnless(importlib.util.find_spec("av"), "install requirements-inversion.txt") + def test_failure_and_cancellation_remove_partial_output(self): + import os + import signal + import subprocess + import time + with tempfile.TemporaryDirectory(prefix="effex-inversion-cancel-") as directory: + root = Path(directory) + clip = root / "clip.mp4" + self.assertTrue(make_clip(clip, seconds=20, size="640x360")) + bad = root / "render.invalid" + failed = run([EFFEX, "inversion", clip, "--output", bad, "--json"], timeout=15) + self.assertNotEqual(failed.returncode, 0) + self.assertFalse(bad.exists()) + target = root / "cancelled.mp4" + env = dict(os.environ, **env_for(root, root)) + process = subprocess.Popen([sys.executable, EFFEX, "inversion", clip, + "--output", target, "--json"], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) + try: + deadline = time.monotonic() + 10 + while not target.exists() and process.poll() is None and time.monotonic() < deadline: + time.sleep(0.02) + self.assertIsNone(process.poll(), "render ended before cancellation could be exercised") + self.assertTrue(target.exists(), "encoder did not create its partial output") + process.send_signal(signal.SIGINT) + process.communicate(timeout=10) + self.assertNotEqual(process.returncode, 0) + self.assertFalse(target.exists(), "cancelled timestamped output was not removed") + finally: + if process.poll() is None: + process.kill() + process.communicate(timeout=5) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/menu_smoke.py b/tests/menu_smoke.py index d2b42cf..7e7ce3b 100644 --- a/tests/menu_smoke.py +++ b/tests/menu_smoke.py @@ -874,6 +874,47 @@ def scenario_collapse(c: Checker): d.close() +def scenario_inversion(c: Checker): + """Buffered future borrowing has editable controls and uses the same renderer.""" + import importlib.util + from harness import newest_mp4 + if importlib.util.find_spec("av") is None or importlib.util.find_spec("numpy") is None: + print(" SKIP inversion menu tests need requirements-inversion.txt") + return + in_dir, out_dir = temp_folders() + make_clip(in_dir / "clip.mp4", seconds=1, size="160x90") + d = start(env_for(in_dir, out_dir)) + try: + d.wait_for("main menu") + d.send(b"p") + d.wait_for("pick a video") + d.send(ENTER) + d.wait_for("pick an effect") + next_effect_group(d, 2) + d.send(DOWN * 4) + d.send(ENTER) + c.check("inversion options open", wait_last(d, "inversion options")) + c.check("inversion exposes buffer and mask controls", all( + text in last_frame(d) for text in ("buffer horizon", "future offset", + "blend opacity", "region mask", "edge feather"))) + press_and_wait(d, RIGHT) + c.check("lookahead is editable", "0.6" in last_frame(d)) + c.check("buffered effect offers no unsupported preview", "p preview" not in last_frame(d)) + d.send(b"r") + c.check("inversion confirm opens", wait_last(d, "render this video?")) + d.send(ENTER) + c.check("inversion menu render finishes", wait_last(d, "wrote", timeout=30)) + out = newest_mp4(out_dir) + c.check("inversion menu render preserves dimensions and audio", + out is not None and video_dims(out) == (160, 90) and audio_codec(out) == "aac") + d.send(ENTER) + wait_last(d, "main menu") + d.send(b"q") + c.check("inversion exits cleanly", d.expect_exit(6) == 0) + finally: + d.close() + + def main() -> int: c = Checker("menu smoke") if not require_tools(c): @@ -885,7 +926,7 @@ def main() -> int: scenario_vhs, scenario_channelswap, scenario_timescratch, scenario_temporal_sediment, scenario_chain, scenario_psychedelic, scenario_collapse, - scenario_backout): + scenario_inversion, scenario_backout): print(f" -- {scenario.__name__}") scenario(c) return c.summary() diff --git a/tests/run_tests.py b/tests/run_tests.py index cb80bab..f3f11c1 100644 --- a/tests/run_tests.py +++ b/tests/run_tests.py @@ -26,6 +26,7 @@ "psychedelic": HERE / "psychedelic_regression.py", "collapse": HERE / "collapse_regression.py", "seams": HERE / "seam_carving_regression.py", + "inversion": HERE / "inversion_regression.py", } diff --git a/timestamped_frames.py b/timestamped_frames.py new file mode 100644 index 0000000..9998718 --- /dev/null +++ b/timestamped_frames.py @@ -0,0 +1,118 @@ +"""Optional NUT transport preserving presentation times through RGB processors. + +FFmpeg still owns decoding/encoding and process cleanup. PyAV carries timestamped +RGB frames over those pipes; nothing here starts a second renderer. +""" +from dataclasses import replace +from fractions import Fraction + +from frame_pipeline import FramePacket + + +def read_packets(pipe, context, stop_event): + import av + + pending = None + previous_gap = Fraction(0) + with av.open(pipe, "r", format="nut") as source: + for index, frame in enumerate(source.decode(video=0)): + if stop_event.is_set(): + return + if frame.pts is None or frame.time_base is None: + raise ValueError("timestamped decoder returned a frame without PTS") + base = Fraction(frame.time_base) + packet = FramePacket(frame.to_ndarray(format="rgb24"), index, + frame.pts * base, base, + (frame.duration or 0) * base, dict(context)) + if pending is not None: + gap = packet.timestamp - pending.timestamp + if gap <= 0: + raise ValueError("inversion requires strictly increasing video PTS") + previous_gap = gap + yield replace(pending, duration=gap) + pending = packet + if pending is not None: + yield replace(pending, duration=pending.duration or previous_gap or Fraction(1, 30)) + + +class _PipeWriter: + """PyAV expects full writes; FileIO on a pipe can return short writes.""" + + def __init__(self, pipe, stop_event): + self.pipe = pipe + self.stop_event = stop_event + self.position = 0 + + def write(self, data): + remaining = memoryview(data) + size = len(remaining) + while remaining: + if self.stop_event.is_set(): + raise InterruptedError("timestamped frame export cancelled") + count = self.pipe.write(remaining) + if count is None or count <= 0: + raise BrokenPipeError("FFmpeg encoder stopped reading timestamped frames") + remaining = remaining[count:] + self.position += count + return size + + def tell(self): + return self.position + + def flush(self): + self.pipe.flush() + + +def write_packets(packets, pipe, shape, fps, stop_event): + import av + import numpy as np + + iterator = iter(packets) + first = next(iterator, None) + if first is None: + raise ValueError("decoder produced no video frames") + base = Fraction(first.time_base) + writer = _PipeWriter(pipe, stop_event) + count = 0 + previous = None + with av.open(writer, "w", format="nut") as output: + stream = output.add_stream("rawvideo") + stream.width, stream.height = shape[1], shape[0] + stream.pix_fmt = "rgb24" + stream.time_base = base + stream.codec_context.time_base = base + stream.codec_context.framerate = Fraction(str(fps)).limit_denominator(1000000) + + def send(packet): + if (not isinstance(packet.frame, np.ndarray) or packet.frame.shape != shape + or packet.frame.dtype != np.uint8): + raise ValueError("stateful processor returned an invalid RGB frame") + pts = packet.timestamp / base + duration = packet.duration / base + if pts.denominator != 1 or duration.denominator != 1: + raise ValueError("processor changed timestamps outside the decoder time base") + frame = av.VideoFrame.from_ndarray(np.ascontiguousarray(packet.frame), format="rgb24") + frame.pts, frame.time_base = int(pts), base + frame.duration = int(duration) + for encoded in stream.encode(frame): + # rawvideo encode drops duration unless it is restored on packets. + encoded.pts = encoded.dts = int(pts) + encoded.time_base, encoded.duration = base, int(duration) + output.mux(encoded) + + for packet in _prepend(first, iterator): + if stop_event.is_set(): + return count + if previous is not None and packet.timestamp <= previous: + raise ValueError("processor returned non-monotonic presentation timestamps") + send(packet) + previous = packet.timestamp + count += 1 + for encoded in stream.encode(): + output.mux(encoded) + return count + + +def _prepend(first, iterator): + yield first + yield from iterator From ebb2b1dab98bcdf953d7ac96a360fd264ce23a7b Mon Sep 17 00:00:00 2001 From: kaffeindecaf Date: Sat, 24 Oct 2026 13:07:06 +0200 Subject: [PATCH 2/3] fix: use valid workflow environment context --- .github/workflows/tests.yml | 2 -- 1 file changed, 2 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 630e389..b966fb5 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -12,8 +12,6 @@ jobs: regression: runs-on: ubuntu-24.04 timeout-minutes: 25 - env: - TMPDIR: ${{ runner.temp }} steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 From eadf387983734361c8dea49c2cc0f4674215f23b Mon Sep 17 00:00:00 2001 From: kaffeindecaf Date: Sat, 24 Oct 2026 13:16:31 +0200 Subject: [PATCH 3/3] fix: preserve dropout frames on older ffmpeg --- effex.py | 2 ++ tests/cli_regression.py | 2 ++ 2 files changed, 4 insertions(+) diff --git a/effex.py b/effex.py index e880737..20af5e1 100755 --- a/effex.py +++ b/effex.py @@ -1666,6 +1666,8 @@ def build_dropout(ctx: EffectContext) -> Plan: return Plan( vf=f"select='gt(random(0),{chance / 100:g})'", out_suffix="_DROPOUT", + # FFmpeg 6 defaults to CFR here and recreates frames that select removed. + out_extra=["-fps_mode", "vfr"], width=w, height=h, notes=[ diff --git a/tests/cli_regression.py b/tests/cli_regression.py index b966f05..44a2375 100644 --- a/tests/cli_regression.py +++ b/tests/cli_regression.py @@ -376,6 +376,8 @@ def frame_hash(path: Path, at: str) -> str: env_extra=env) c.check("dropout dry-run carries the random select", "select='gt(random(0),0.3)'" in proc.stdout.decode()) + c.check("dropout prevents CFR from recreating removed frames", + "-fps_mode vfr" in proc.stdout.decode()) proc = run([EFFEX, "dropout", clip, "--set", "chance=30", "--json"], env_extra=env) dropped = Path(json.loads(proc.stdout.decode())["output"])