Add streaming video SynthID regeneration

tests/test_video.py

@@ -81,11 +81,33 @@ def _video_with_tc260_ebml(path: Path, *, value: bytes = _TC260_AIGC) -> Path:
     return path
 
 
+def _regeneration_metrics(
+    *,
+    frames: int = 24,
+    fps: float = 12.0,
+    width: int = 512,
+    height: int = 288,
+    psnr_db: float = 22.0,
+    temporal_residual_ratio: float = 1.2,
+):
+    from remove_ai_watermarks.video_invisible import RegenerationMetrics
+
+    return RegenerationMetrics(
+        frames=frames,
+        fps=fps,
+        width=width,
+        height=height,
+        psnr_db=psnr_db,
+        temporal_residual_ratio=temporal_residual_ratio,
+    )
+
+
 class TestVideoMetadataApi:
     def test_top_level_api_is_lazy_exported(self):
         import remove_ai_watermarks as raiw
 
         assert raiw.inspect_video_metadata is not None
+        assert raiw.remove_video_invisible is not None
         assert raiw.remove_video_metadata is not None
         assert raiw.remove_video_visible is not None
 
@@ -264,6 +286,107 @@ class TestVideoMetadataCli:
         assert "Unsupported video format" in result.output
 
 
+class TestVideoInvisibleApi:
+    def test_generates_unverified_candidate_and_strips_metadata(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        from remove_ai_watermarks import video_invisible
+        from remove_ai_watermarks.video import remove_video_invisible
+
+        source = _video_with_c2pa(tmp_path / "source.mp4")
+        output = tmp_path / "candidate.mp4"
+
+        def fake_regenerate(_source: Path, target: Path, **_kwargs: object):
+            target.write_bytes(_MP4_FTYP + _box(b"mdat", _VIDEO_PAYLOAD))
+            return _regeneration_metrics()
+
+        monkeypatch.setattr(video_invisible, "regenerate_video_candidate", fake_regenerate)
+
+        result = remove_video_invisible(source, output)
+
+        assert result.output == output
+        assert result.requires_external_verification is True
+        assert result.total_frames == 24
+        assert result.remaining_metadata == {}
+
+    def test_default_output_is_named_as_candidate(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        from remove_ai_watermarks import video_invisible
+        from remove_ai_watermarks.video import remove_video_invisible
+
+        source = _video_with_c2pa(tmp_path / "source.mp4")
+
+        def fake_regenerate(_source: Path, target: Path, **_kwargs: object):
+            target.write_bytes(_MP4_FTYP + _box(b"mdat", _VIDEO_PAYLOAD))
+            return _regeneration_metrics(
+                frames=2,
+                fps=2.0,
+                width=16,
+                height=16,
+                psnr_db=20.0,
+                temporal_residual_ratio=1.0,
+            )
+
+        monkeypatch.setattr(video_invisible, "regenerate_video_candidate", fake_regenerate)
+
+        result = remove_video_invisible(source)
+
+        assert result.output == tmp_path / "source_synthid_candidate.mp4"
+
+    def test_rejects_webm_regeneration(self, tmp_path: Path):
+        from remove_ai_watermarks.video import remove_video_invisible
+
+        source = _video_with_tc260_ebml(tmp_path / "source.webm")
+
+        with pytest.raises(ValueError, match="requires one of"):
+            remove_video_invisible(source)
+
+
+class TestVideoInvisibleCli:
+    def test_help_describes_external_verification(self):
+        runner = CliRunner()
+
+        result = runner.invoke(main, ["video", "invisible", "--help"])
+
+        assert result.exit_code == 0, result.output
+        assert "externally verifiable" in result.output
+
+    def test_reports_unverified_candidate(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        from remove_ai_watermarks import video
+
+        runner = CliRunner()
+        source = _video_with_c2pa(tmp_path / "source.mp4")
+        output = tmp_path / "candidate.mp4"
+
+        def fake_remove(_source: Path, target: Path, **_kwargs: object):
+            target.write_bytes(_MP4_FTYP + _box(b"mdat", _VIDEO_PAYLOAD))
+            return video.VideoInvisibleResult(
+                source=_source,
+                output=target,
+                noise_std=0.1,
+                metrics=_regeneration_metrics(),
+                remaining_metadata={},
+            )
+
+        monkeypatch.setattr(video, "remove_video_invisible", fake_remove)
+
+        result = runner.invoke(main, ["video", "invisible", str(source), "-o", str(output)])
+
+        assert result.exit_code == 0, result.output
+        assert "Candidate generated" in result.output
+        assert "UNVERIFIED" in result.output
+        assert "Gemini Flash" in result.output
+
+
 class TestSoraFrameLocalization:
     @staticmethod
     def _sora_like_frame() -> tuple[np.ndarray, tuple[int, int, int, int]]:

tests/test_video_invisible.py

@@ -0,0 +1,71 @@
+"""Regression tests for the video SynthID candidate engine."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from remove_ai_watermarks import video_encoding, video_invisible
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def test_availability_requires_both_optional_packages(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        video_invisible,
+        "find_spec",
+        lambda name: object() if name == "torch" else None,
+    )
+
+    assert video_invisible.is_available() is False
+
+
+def test_regeneration_rejects_noise_outside_unit_interval(tmp_path: Path) -> None:
+    with pytest.raises(ValueError, match="between 0 and 1"):
+        video_invisible.regenerate_video_candidate(
+            tmp_path / "source.mp4",
+            tmp_path / "candidate.mp4",
+            noise_std=1.01,
+        )
+
+
+def test_encoder_command_discards_metadata_and_copies_audio(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    source = tmp_path / "source.mp4"
+    output = tmp_path / "candidate.mp4"
+
+    monkeypatch.setattr(video_encoding.shutil, "which", lambda _name: "/usr/bin/ffmpeg")
+
+    command = video_encoding.raw_video_command(
+        source,
+        output,
+        width=8,
+        height=8,
+        fps=2.0,
+        strip_metadata=True,
+        crf=18,
+    )
+
+    metadata_index = command.index("-map_metadata")
+    assert command[metadata_index + 1] == "-1"
+    audio_codec_index = command.index("-c:a")
+    assert command[audio_codec_index + 1] == "copy"
+    assert "pipe:0" in command
+
+
+def test_stream_batches_consumes_only_one_batch_ahead() -> None:
+    consumed: list[int] = []
+
+    def values():
+        for value in range(5):
+            consumed.append(value)
+            yield value
+
+    batches = video_invisible._stream_batches(values(), 2)
+
+    assert next(iter(batches)) == [0, 1]
+    assert consumed == [0, 1]

tests/test_video_synthid_sweep.py

@@ -55,7 +55,7 @@ def test_shared_latent_noise_is_seeded(sweep: ModuleType) -> None:
 
 def test_psnr_is_infinite_for_identical_frames(sweep: ModuleType) -> None:
     frame = np.full((2, 8, 8, 3), 120, dtype=np.uint8)
-    assert sweep._psnr(frame, frame.copy()) == pytest.approx(float("inf"))
+    assert sweep.paired_psnr(frame, frame.copy()) == pytest.approx(float("inf"))
 
 
 def test_temporal_residual_ratio_is_one_for_identical_sequences(sweep: ModuleType) -> None:
@@ -63,5 +63,5 @@ def test_temporal_residual_ratio_is_one_for_identical_sequences(sweep: ModuleTyp
     second = first.copy()
     second[:, 8:16] = 80
     sequence = [first, second]
-    maps, baseline = sweep._temporal_reference(sequence)
-    assert sweep._temporal_residual_ratio([frame.copy() for frame in sequence], maps, baseline) == pytest.approx(1.0)
+    maps, baseline = sweep.build_temporal_reference(sequence)
+    assert sweep.temporal_residual_ratio([frame.copy() for frame in sequence], maps, baseline) == pytest.approx(1.0)