diff --git a/docs/synthid-detector-removal-plan.md b/docs/synthid-detector-removal-plan.md index 7ab3f1e..2a0ba44 100644 --- a/docs/synthid-detector-removal-plan.md +++ b/docs/synthid-detector-removal-plan.md @@ -262,8 +262,10 @@ Goal: learn watermark presence from counterfactual image-level evidence that a fixed template misses. The primary model uses a full canonical field with raw RGB or luminance/chroma, -absolute-amplitude fine and coarse residuals, and an optional frequency branch. -Locally normalized patch evidence remains a frozen ablation, not the primary +absolute-amplitude fine and coarse residuals, and optional stationary-wavelet +and complex-frequency branches. Every optional representation is encoded +separately and fused late; early channel concatenation is not a valid ablation. +Locally normalized patch evidence remains a frozen baseline, not the primary input. Multi-view registration is aggregated into one image-level presence logit. Payload-like, phase, localization, and content-watermarkability heads are auxiliary and must prove an incremental held-out benefit. @@ -937,6 +939,52 @@ Until the counterfactual-label and full-field gates pass, the current residual CNN remains a useful vendor-triage stage for a future cascade, but it is not a SynthID detector and should not be optimized as the final decision surface. +### 2026-08-09: paired wavelet and spectral ablation + +The paired spectral harness was extended with three-level undecimated `db2` +wavelets, complex Fourier phase coherence and power, cepstral peaks, and a +cyclic clean/marked permutation control. The implementation streams wavelet and +spectrum field accumulators rather than stacking transformed fields across +pairs. + +On 60 public TrustMark P training pairs at canonical size 256, the true +residuals had mean inter-pair RGB NCC of 0.097-0.121. Cyclically mismatching each +clean image with the next marked image reduced all three channels to about +-0.017. The strongest coherent Fourier cluster occurred around vertical offsets +9-11 and was strongest in the blue channel. Wavelet repeatability was greatest +in horizontal detail bands at levels 1 and 2. These are causal residual +measurements for the open proxy, not evidence about SynthID's carrier. + +The representations then faced a source-disjoint single-image test with 60 +training, 15 calibration, and 15 test pairs under identity, JPEG-90, 0.8 resize, +and 5% crop transformations. A fixed spectral-template score reached only 0.618 +identity AUC and collapsed to 0.498 under crop. A 318-feature wavelet/spectral +summary reached 0.653 aggregate AUC and zero TPR at the threshold above every +calibration negative. + +Full-field neural ablations at size 128 gave the following aggregate results: + +| Input | AUC | Paired wins | Test FPR | Test TPR | +| --- | ---: | ---: | ---: | ---: | +| Spatial RGB plus residuals | 0.723 | 57/60 | 0% | 11.7% | +| SWT maps only | 0.671 | 59/60 | 0% | 0% | +| Complex FFT maps only | 0.516 | 41/60 | 0% | 0% | +| Spatial plus SWT, late fusion | 0.665 | 59/60 | 5.0% | 8.3% | +| Spatial plus SWT plus FFT, early fusion | 0.639 | 57/60 | 1.7% | 1.7% | +| Spatial plus SWT plus FFT, late fusion | 0.698 | 54/60 | 0% | 0% | + +The table uses one threshold above every calibration negative; test FPR is +reported separately because a clean calibration result does not guarantee a +clean locked test. + +The SWT branch contains weak, unusually consistent paired evidence, but it did +not improve the spatial model's source-disjoint discrimination in this small +proxy. The FFT map branch was indistinguishable from chance. Therefore neither +representation advances into the primary detector by default. SWT remains a +late-fusion ablation for a larger paired corpus; complex spectral analysis +remains a residual-discovery diagnostic unless a future held-out test reverses +this result. + ## Decision record The program has four possible honest outcomes per provider: diff --git a/docs/synthid.md b/docs/synthid.md index a84c23f..0b4a34d 100644 --- a/docs/synthid.md +++ b/docs/synthid.md @@ -258,6 +258,16 @@ research fidelity gate. The OpenAI periodic residual also reduced all three local model scores on 11 of 11 additional images before and after JPEG-90, but neither provider candidate has a negative matching-provider oracle verdict, so neither is an established remover. + +A later paired open-method control tested stationary wavelet and complex +spectral representations without treating another watermark as SynthID. On a +source-disjoint TrustMark P proxy, the SWT branch moved 59 of 60 transformed +clean/watermarked pairs in the correct direction but did not improve the +spatial model's aggregate AUC. Complex FFT maps were indistinguishable from +chance, and fixed phase scoring collapsed under crop. Wavelets therefore remain +a gated late-fusion ablation, while complex spectral analysis remains a +residual-discovery tool rather than a presence score. + The protocol, exact limitations, and next experiments are recorded in the [`detector and removal research plan`](synthid-detector-removal-plan.md). diff --git a/scripts/synthid_spectral_probe.py b/scripts/synthid_spectral_probe.py index 5c082a1..3859804 100644 --- a/scripts/synthid_spectral_probe.py +++ b/scripts/synthid_spectral_probe.py @@ -1,9 +1,9 @@ -"""Discover and score a shared spectral carrier from exact image pairs. +"""Analyze wavelet and spectral structure from exact image pairs. This is a research harness, not a production SynthID detector. Pair provenance and oracle labels remain external evidence. The harness deliberately separates -template discovery from single-image scoring and stores arrays in NPZ without -pickle. +paired residual discovery from single-image fixed-template scoring, includes a +permuted-pair control, and stores arrays in NPZ without pickle. Usage: uv run python scripts/synthid_spectral_probe.py discover \ @@ -19,16 +19,22 @@ from __future__ import annotations import json import logging -from dataclasses import asdict, dataclass +from dataclasses import asdict, dataclass, field from itertools import combinations from pathlib import Path import click import numpy as np +import pywt from PIL import Image, ImageFilter log = logging.getLogger(__name__) +WAVELET = "db2" +WAVELET_LEVELS = 3 +SPECTRAL_PEAKS = 32 +CEPSTRAL_PEAKS = 16 + @dataclass(frozen=True) class PairMeasurement: @@ -58,6 +64,17 @@ class ImageScore: peak_count: int +@dataclass +class _WaveletAccumulator: + """Streaming summary state for one wavelet band.""" + + coefficient_sum: np.ndarray + energy_sum: np.ndarray + coefficient_norm_sum: np.ndarray + energy_norm_sum: np.ndarray + rms_values: list[np.ndarray] = field(default_factory=list) + + def load_rgb(path: Path) -> np.ndarray: """Load PATH as float64 RGB pixels.""" with Image.open(path) as image: @@ -111,6 +128,198 @@ def channel_ncc(first: np.ndarray, second: np.ndarray) -> tuple[float, float, fl return tuple(float(value) for value in values) +def pairwise_ncc_from_normalized_sum( + total: np.ndarray, + squared_norm_sum: np.ndarray, + count: int, +) -> tuple[float, float, float]: + """Recover mean pairwise NCC from the sum of normalized RGB arrays.""" + if count < 2: + return (0.0, 0.0, 0.0) + pair_count = count * (count - 1) / 2 + values = (np.sum(np.square(total), axis=(0, 1)) - squared_norm_sum) / (2.0 * pair_count) + return tuple(float(value) for value in values) + + +def stationary_wavelet_bands( + residual: np.ndarray, + *, + wavelet: str = WAVELET, + levels: int = WAVELET_LEVELS, +) -> dict[str, np.ndarray]: + """Return undecimated detail bands while preserving canonical coordinates.""" + if residual.ndim != 3 or residual.shape[2] != 3: + raise ValueError("residual must be an RGB array") + if levels < 1: + raise ValueError("levels must be positive") + divisor = 2**levels + if residual.shape[0] % divisor or residual.shape[1] % divisor: + raise ValueError(f"residual dimensions must be divisible by {divisor}") + + coefficients = pywt.swt2(residual, wavelet, level=levels, axes=(0, 1)) + bands: dict[str, np.ndarray] = {} + for level, (_approximation, details) in enumerate(reversed(coefficients), start=1): + for orientation, values in zip(("h", "v", "d"), details, strict=True): + bands[f"level_{level}_{orientation}"] = np.asarray(values, dtype=np.float64) + return bands + + +def wavelet_report(residuals: list[np.ndarray]) -> dict[str, object]: + """Summarize multi-scale wavelet energy and cross-pair repeatability.""" + if not residuals: + raise ValueError("at least one residual is required") + accumulators: dict[str, _WaveletAccumulator] = {} + for residual in residuals: + for name, values in stationary_wavelet_bands(residual).items(): + coefficient = normalized_channels(values) + energy = normalized_channels(np.square(values)) + accumulator = accumulators.get(name) + if accumulator is None: + accumulator = _WaveletAccumulator( + coefficient_sum=np.zeros_like(coefficient), + energy_sum=np.zeros_like(energy), + coefficient_norm_sum=np.zeros(3, dtype=np.float64), + energy_norm_sum=np.zeros(3, dtype=np.float64), + ) + accumulators[name] = accumulator + accumulator.coefficient_sum += coefficient + accumulator.energy_sum += energy + accumulator.coefficient_norm_sum += np.sum(np.square(coefficient), axis=(0, 1)) + accumulator.energy_norm_sum += np.sum(np.square(energy), axis=(0, 1)) + accumulator.rms_values.append(np.sqrt(np.mean(np.square(values), axis=(0, 1)))) + + bands: list[dict[str, object]] = [] + for name, accumulator in accumulators.items(): + rms = np.asarray(accumulator.rms_values) + bands.append( + { + "name": name, + "median_rms": [float(value) for value in np.median(rms, axis=0)], + "rms_cv": [ + float(value) + for value in np.divide( + np.std(rms, axis=0), + np.mean(rms, axis=0), + out=np.zeros(3, dtype=np.float64), + where=np.mean(rms, axis=0) > 1e-12, + ) + ], + "coefficient_ncc": pairwise_ncc_from_normalized_sum( + accumulator.coefficient_sum, + accumulator.coefficient_norm_sum, + len(residuals), + ), + "energy_map_ncc": pairwise_ncc_from_normalized_sum( + accumulator.energy_sum, + accumulator.energy_norm_sum, + len(residuals), + ), + } + ) + return {"wavelet": WAVELET, "levels": WAVELET_LEVELS, "bands": bands} + + +def _top_half_plane_offsets(values: np.ndarray, count: int, *, min_radius: float) -> list[tuple[int, int, int]]: + """Return top ROW/COLUMN/CHANNEL indices from one centered half-plane.""" + height, width, channels = values.shape + center_y, center_x = height // 2, width // 2 + yy, xx = np.ogrid[:height, :width] + radius = np.sqrt(np.square(yy - center_y) + np.square(xx - center_x)) + half_plane = (yy > center_y) | ((yy == center_y) & (xx >= center_x)) + valid = (radius >= min_radius) & half_plane + masked = np.where(valid[:, :, None], values, -np.inf) + limit = min(count, int(np.sum(np.isfinite(masked)))) + flat = np.argpartition(masked.ravel(), -limit)[-limit:] + flat = flat[np.argsort(masked.ravel()[flat])[::-1]] + return [tuple(int(value) for value in np.unravel_index(index, (height, width, channels))) for index in flat] + + +def spectral_report(residuals: list[np.ndarray]) -> dict[str, object]: + """Summarize complex phase coherence, power, and cepstral periodicity.""" + if not residuals: + raise ValueError("at least one residual is required") + unit_sum = np.zeros(residuals[0].shape, dtype=np.complex128) + power_sum = np.zeros(residuals[0].shape, dtype=np.float64) + for residual in residuals: + spectrum = np.fft.fftshift(np.fft.fft2(normalized_channels(residual), axes=(0, 1)), axes=(0, 1)) + magnitude = np.abs(spectrum) + unit_sum += np.divide(spectrum, magnitude, out=np.zeros_like(spectrum), where=magnitude > 1e-12) + power_sum += np.square(magnitude) + coherence = np.abs(unit_sum / len(residuals)) + power = power_sum / len(residuals) + weighted = coherence * np.sqrt(power) + height, width, _ = coherence.shape + center_y, center_x = height // 2, width // 2 + total_power = np.sum(power, axis=(0, 1)) + peaks = [] + for row, column, channel in _top_half_plane_offsets(weighted, SPECTRAL_PEAKS, min_radius=4.0): + peaks.append( + { + "dy": row - center_y, + "dx": column - center_x, + "channel": channel, + "phase_coherence": float(coherence[row, column, channel]), + "power_fraction": float(power[row, column, channel] / max(total_power[channel], 1e-12)), + } + ) + + mean_power = np.mean(power, axis=2) + cepstrum = np.abs(np.fft.fftshift(np.fft.ifft2(np.fft.ifftshift(np.log1p(mean_power))))) + cepstral_cube = cepstrum[:, :, None] + cepstral = [ + {"dy": row - center_y, "dx": column - center_x, "magnitude": float(cepstrum[row, column])} + for row, column, _channel in _top_half_plane_offsets(cepstral_cube, CEPSTRAL_PEAKS, min_radius=2.0) + ] + return { + "phase_coherence_median": float(np.median(coherence)), + "phase_coherence_p95": float(np.quantile(coherence, 0.95)), + "peaks": peaks, + "cepstral_peaks": cepstral, + } + + +def _canonical_rgb(path: Path, size: int) -> np.ndarray: + """Return floating-point RGB pixels at the canonical analysis size.""" + with Image.open(path) as source: + image = source.convert("RGB").resize((size, size), Image.Resampling.BILINEAR) + return np.asarray(image, dtype=np.float64) + + +def permutation_control( + residuals: list[np.ndarray], + measurements: list[PairMeasurement], +) -> dict[str, object] | None: + """Compare true residual repeatability with deliberately mismatched pairs.""" + if len(measurements) < 2: + return None + size = residuals[0].shape[0] + true_sum = np.zeros_like(residuals[0]) + mismatched_sum = np.zeros_like(residuals[0]) + true_norm_sum = np.zeros(3, dtype=np.float64) + mismatched_norm_sum = np.zeros(3, dtype=np.float64) + mismatched_rms: list[float] = [] + for residual in residuals: + normalized = normalized_channels(residual) + true_sum += normalized + true_norm_sum += np.sum(np.square(normalized), axis=(0, 1)) + for index, measurement in enumerate(measurements): + other = measurements[(index + 1) % len(measurements)] + clean = _canonical_rgb(Path(measurement.clean), size) + marked = _canonical_rgb(Path(other.marked), size) + mismatched = marked - clean + normalized = normalized_channels(mismatched) + mismatched_sum += normalized + mismatched_norm_sum += np.sum(np.square(normalized), axis=(0, 1)) + mismatched_rms.append(float(np.sqrt(np.mean(np.square(mismatched))))) + return { + "strategy": "cyclic marked-image permutation", + "true_pair_ncc": pairwise_ncc_from_normalized_sum(true_sum, true_norm_sum, len(residuals)), + "mismatched_pair_ncc": pairwise_ncc_from_normalized_sum(mismatched_sum, mismatched_norm_sum, len(residuals)), + "true_median_rms": float(np.median([np.sqrt(np.mean(np.square(residual))) for residual in residuals])), + "mismatched_median_rms": float(np.median(mismatched_rms)), + } + + def build_template(residuals: list[np.ndarray]) -> np.ndarray: """Average canonical residuals after per-channel normalization.""" if not residuals: @@ -227,7 +436,7 @@ def load_template(path: Path) -> tuple[np.ndarray, np.ndarray]: def discovery_report( residuals: list[np.ndarray], measurements: list[PairMeasurement], peaks: np.ndarray ) -> dict[str, object]: - """Build a JSON-safe report with pair statistics and cross-pair NCC.""" + """Build a JSON-safe paired wavelet and spectral discovery report.""" pairwise = [ { "first": measurements[first].marked, @@ -241,12 +450,15 @@ def discovery_report( "pairs": [asdict(measurement) for measurement in measurements], "pairwise": pairwise, "peaks": peaks.tolist(), + "wavelet": wavelet_report(residuals), + "spectral": spectral_report(residuals), + "permutation_control": permutation_control(residuals, measurements), } @click.group() def main() -> None: - """Discover and score an experimental shared spectral carrier.""" + """Analyze paired residuals and score an experimental spectral template.""" logging.basicConfig(level=logging.INFO, format="%(message)s") diff --git a/tests/test_synthid_spectral_probe.py b/tests/test_synthid_spectral_probe.py index c9ce250..d3286fa 100644 --- a/tests/test_synthid_spectral_probe.py +++ b/tests/test_synthid_spectral_probe.py @@ -3,6 +3,7 @@ from __future__ import annotations import sys +from itertools import combinations from pathlib import Path import numpy as np @@ -101,3 +102,81 @@ def test_discovery_report_contains_cross_pair_ncc(tmp_path: Path): assert report["pair_count"] == 2 assert len(report["pairwise"]) == 1 + assert report["wavelet"]["wavelet"] == "db2" + assert len(report["wavelet"]["bands"]) == 9 + assert report["spectral"]["peaks"] + assert report["spectral"]["cepstral_peaks"] + assert report["permutation_control"] is not None + + +def test_wavelet_report_finds_repeatable_multiscale_carrier(tmp_path: Path): + first = _write_pair(tmp_path, "first", (96, 96), 100) + second = _write_pair(tmp_path, "second", (128, 80), 140) + first_residual, _ = probe.pair_residual(*first, size=64) + second_residual, _ = probe.pair_residual(*second, size=64) + + report = probe.wavelet_report([first_residual, second_residual]) + + assert len(report["bands"]) == 9 + assert max(max(band["coefficient_ncc"]) for band in report["bands"]) > 0.7 + assert all(len(band["median_rms"]) == 3 for band in report["bands"]) + + +def test_streaming_pairwise_ncc_matches_explicit_pairs(): + rng = np.random.default_rng(7) + arrays = [rng.normal(size=(16, 16, 3)) for _ in range(5)] + normalized = [probe.normalized_channels(values) for values in arrays] + + explicit = np.mean( + [probe.channel_ncc(arrays[first], arrays[second]) for first, second in combinations(range(len(arrays)), 2)], + axis=0, + ) + squared_norm_sum = np.sum([np.sum(np.square(values), axis=(0, 1)) for values in normalized], axis=0) + streamed = probe.pairwise_ncc_from_normalized_sum( + np.sum(normalized, axis=0), + squared_norm_sum, + len(normalized), + ) + + assert np.allclose(streamed, explicit) + + +def test_streaming_pairwise_ncc_handles_zero_norm_channels(): + arrays = [np.zeros((8, 8, 3)), np.zeros((8, 8, 3))] + normalized = [probe.normalized_channels(values) for values in arrays] + + streamed = probe.pairwise_ncc_from_normalized_sum( + np.sum(normalized, axis=0), + np.zeros(3), + len(normalized), + ) + + assert streamed == (0.0, 0.0, 0.0) + + +def test_spectral_report_finds_injected_frequency(tmp_path: Path): + first = _write_pair(tmp_path, "first", (96, 96), 100) + second = _write_pair(tmp_path, "second", (128, 80), 140) + first_residual, _ = probe.pair_residual(*first, size=64) + second_residual, _ = probe.pair_residual(*second, size=64) + + report = probe.spectral_report([first_residual, second_residual]) + + assert any(abs(peak["dy"]) == 14 and abs(peak["dx"]) == 14 for peak in report["peaks"]) + assert report["phase_coherence_p95"] >= report["phase_coherence_median"] + + +def test_permutation_control_breaks_exact_pair_repeatability(tmp_path: Path): + first = _write_pair(tmp_path, "first", (96, 96), 100) + second = _write_pair(tmp_path, "second", (128, 80), 140) + first_residual, first_measurement = probe.pair_residual(*first, size=64) + second_residual, second_measurement = probe.pair_residual(*second, size=64) + + report = probe.permutation_control( + [first_residual, second_residual], + [first_measurement, second_measurement], + ) + + assert report is not None + assert np.mean(report["true_pair_ncc"]) > np.mean(report["mismatched_pair_ncc"]) + assert report["mismatched_median_rms"] > report["true_median_rms"]