From 1d7a6270d4f3760cb8e9cafc80f0532067b0d6ad Mon Sep 17 00:00:00 2001 From: Victor Kuznetsov Date: Tue, 15 Sep 2026 16:20:31 -0700 Subject: [PATCH] Add phase-free spectral pipeline probe --- docs/synthid-classifiers.md | 39 +++ scripts/README.md | 1 + scripts/spectral_pipeline_probe.py | 346 ++++++++++++++++++++++++++ tests/test_spectral_pipeline_probe.py | 81 ++++++ 4 files changed, 467 insertions(+) create mode 100644 scripts/spectral_pipeline_probe.py create mode 100644 tests/test_spectral_pipeline_probe.py diff --git a/docs/synthid-classifiers.md b/docs/synthid-classifiers.md index 95c3ad5..fd0bca9 100644 --- a/docs/synthid-classifiers.md +++ b/docs/synthid-classifiers.md @@ -23,6 +23,45 @@ are explicit negatives, a strict rule can abstain, and any watermark claim uses an independent oracle. CLIP content embeddings and the 124-d origin-locked residual bank are different features for different jobs. +### Phase-free spectral pipeline probe, 2026-09-15 + +Mark's public +[`bigdatamark/synthid-research`](https://huggingface.co/datasets/bigdatamark/synthid-research) +corpus pairs 300 prompts across Gemini 2048-pixel JPEG, OpenAI 1254-pixel PNG, +and FLUX 1280-pixel PNG exports. It is useful for testing source-pipeline +features, but it is not a causal SynthID corpus: renderer, dimensions, encoder, +and watermark state all change together. + +The paper-inspired probe in [`scripts/spectral_pipeline_probe.py`](../scripts/spectral_pipeline_probe.py) +discards Fourier phase, subtracts each image's radial log-magnitude envelope, +pools luma plus two opponent-color spectra, and fits a dual ridge classifier. +Matching prompt ids stay together in a grouped five-fold evaluation. The +256-pixel / 32x32-grid / ridge-1000 configuration was selected after an +exploratory comparison on these same folds, so this is a development result, +not blind validation. Its complete-corpus confusion matrix was: + +| Actual pipeline | Gemini | OpenAI | FLUX | +| --- | ---: | ---: | ---: | +| Gemini | 220 | 25 | 55 | +| OpenAI | 33 | 219 | 48 | +| FLUX | 75 | 48 | 177 | + +Accuracy was 616/900 (68.4%), against 33.3% chance. Applying each transform +only to held-out rows gave 67.3% after a two-pixel top-left crop, 67.8% after a +95% resize, 67.9% after JPEG quality 75, and 68.4% after Gaussian blur radius +0.7. This is materially more transform-stable than the origin-locked lattice, +but it remains a source/export classifier. There are no photographs, foreign +generators beyond FLUX, independently generated 1K Gemini files, or same-image +SynthID on/off pairs in the corpus. No runtime detector, threshold, or product +verdict is justified by this result. + +The representation follows the frequency-domain premise of Zhang et al., +[*Detecting and Simulating Artifacts in GAN Fake Images*](https://arxiv.org/abs/1907.06515), +not their ResNet-34 implementation. Their own post-processing results also +support treating augmentation and architecture shift as first-class tests, +rather than reading an untransformed source-classification score as watermark +evidence. + Krawetz's Gemini-chat TPR critique is a verifier-quality claim, not a feature we can ship. [Lead Stories, 2026-07](https://leadstories.com/analysis/2026/07/google-gemini-synthid-detector-confuses-results-within-same-chat.html) documented Gemini repeating the first file's SynthID verdict inside a diff --git a/scripts/README.md b/scripts/README.md index 52d1eeb..a4775fc 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -50,6 +50,7 @@ explicitly names a tracked canonical result. | `visible_recall_sample.py` | Build an unbiased blinded sample for recall measurement. | | `visible_sheets.py` | Build blinded contact sheets for relaxation candidates. | | `watermark_benchmark_cohort.py` | Build the deterministic synthetic DWT-DCT and TrustMark cohort consumed by `watermark_benchmark.py`. | +| `spectral_pipeline_probe.py` | Measure a prompt-grouped, phase-free spectral source-pipeline hypothesis without presenting it as SynthID detection. | | `watermark_benchmark_real_cohort.py` | Build a publication-cleared, provider-paired and content-stratified real-image cohort. | | `retrain_photo_classify.py` | CPU-retrain the 2026-08-31 photo heads from a sha256-keyed cache pack, no images. | | `publish_photo_classify_hf.py` | Upload the photo-classify card and freeze weights to Hugging Face `wiltodelta/raiw-photo-classify`. Manual; the Action `publish-photo-classify-hf.yml` is the write-token path. | diff --git a/scripts/spectral_pipeline_probe.py b/scripts/spectral_pipeline_probe.py new file mode 100644 index 0000000..0d6cc76 --- /dev/null +++ b/scripts/spectral_pipeline_probe.py @@ -0,0 +1,346 @@ +"""Measure a phase-free spectral source-pipeline hypothesis. + +This is a research harness, not a SynthID detector. It classifies complete +generation/export pipelines whose renderer, output geometry, encoding, and +watermark state may all be confounded. A watermark claim requires a causal +marked/clean contrast and an independent oracle. + +The input directory contains one subdirectory per class. Matching filename +stems are treated as one prompt group and always stay in the same fold. + +Example: + uv run python scripts/spectral_pipeline_probe.py DATASET \ + --class-dir gemini --class-dir openai --class-dir flux \ + --attack original --attack crop2 --attack resize95 \ + --attack jpeg75 --attack blur \ + --report-out .local-eval/spectral-pipeline-report.json +""" + +from __future__ import annotations + +import io +import json +import logging +from dataclasses import dataclass +from pathlib import Path +from typing import cast + +import click +import numpy as np +from numpy.typing import NDArray +from PIL import Image, ImageFilter + +log = logging.getLogger(__name__) + +IMAGE_SUFFIXES = frozenset({".jpg", ".jpeg", ".png", ".webp"}) +ATTACKS = ("original", "crop2", "resize95", "jpeg75", "blur") + +FloatArray = NDArray[np.float64] +IntArray = NDArray[np.int64] +BoolArray = NDArray[np.bool_] + + +@dataclass(frozen=True) +class SpectralGeometry: + """Precomputed coordinates for one phase-free spectral representation.""" + + size: int + grid: int + radius: NDArray[np.int32] + radius_counts: IntArray + passband: BoolArray + + +@dataclass(frozen=True) +class FoldModel: + """One standardized dual-ridge model and its held-out rows.""" + + test_rows: BoolArray + mean: FloatArray + scale: FloatArray + active: BoolArray + weights: FloatArray + + +def make_geometry(size: int = 256, grid: int = 32) -> SpectralGeometry: + """Build the radial detrending and pooling geometry.""" + if size < 16 or grid < 2 or size % grid: + raise ValueError("size must be at least 16 and divisible by grid") + yy, xx = np.mgrid[:size, :size] + radius = np.rint(np.hypot(yy - size // 2, xx - size // 2)).astype(np.int32) + radius_counts = np.bincount(radius.ravel()) + passband = (radius >= 4) & (radius <= size * 0.48) + return SpectralGeometry(size, grid, radius, radius_counts, passband) + + +def phase_free_log_spectrum(channel: FloatArray) -> FloatArray: + """Return centered log FFT magnitude, deliberately discarding phase.""" + if channel.ndim != 2: + raise ValueError("channel must be a two-dimensional array") + centered = np.asarray(channel, dtype=np.float64) - float(np.mean(channel)) + return np.log1p(np.abs(np.fft.fftshift(np.fft.fft2(centered)))) + + +def spectral_feature(pixels: FloatArray, geometry: SpectralGeometry) -> FloatArray: + """Extract pooled radial-residual spectra from luma and opponent color.""" + if pixels.shape != (geometry.size, geometry.size, 3): + raise ValueError(f"pixels must have shape {(geometry.size, geometry.size, 3)}, got {pixels.shape}") + red, green, blue = np.moveaxis(np.asarray(pixels, dtype=np.float64), 2, 0) + channels = ( + 0.299 * red + 0.587 * green + 0.114 * blue, + red - green, + blue - (red + green) / 2.0, + ) + block = geometry.size // geometry.grid + features: list[FloatArray] = [] + for channel in channels: + spectrum = phase_free_log_spectrum(channel) + radial_sum = np.bincount(geometry.radius.ravel(), weights=spectrum.ravel()) + radial_mean = radial_sum / geometry.radius_counts + residual = np.where( + geometry.passband, + spectrum - radial_mean[geometry.radius], + 0.0, + ) + pooled = residual.reshape(geometry.grid, block, geometry.grid, block).mean(axis=(1, 3)) + features.append(pooled.ravel()) + return np.concatenate(features) + + +def _apply_attack(image: Image.Image, attack: str) -> Image.Image: + """Apply one declared robustness transform before canonical resizing.""" + if attack == "original": + return image + if attack == "crop2": + if image.width <= 4 or image.height <= 4: + raise ValueError("crop2 requires both image dimensions to exceed four pixels") + return image.crop((2, 2, image.width, image.height)) + if attack == "resize95": + return image.resize( + (max(1, round(image.width * 0.95)), max(1, round(image.height * 0.95))), + Image.Resampling.LANCZOS, + ) + if attack == "jpeg75": + encoded = io.BytesIO() + image.save(encoded, format="JPEG", quality=75) + encoded.seek(0) + with Image.open(encoded) as decoded: + return decoded.convert("RGB") + if attack == "blur": + return image.filter(ImageFilter.GaussianBlur(0.7)) + raise ValueError(f"unsupported attack: {attack}") + + +def feature_from_path(path: Path, geometry: SpectralGeometry, attack: str) -> FloatArray: + """Decode one path, apply ATTACK, and return its phase-free feature.""" + with Image.open(path) as source: + return _feature_from_image(source.convert("RGB"), geometry, attack) + + +def _feature_from_image(image: Image.Image, geometry: SpectralGeometry, attack: str) -> FloatArray: + """Extract one attacked feature from an already decoded RGB image.""" + attacked = _apply_attack(image, attack) + canonical = attacked.resize((geometry.size, geometry.size), Image.Resampling.LANCZOS) + pixels = np.asarray(canonical, dtype=np.float64) / 255.0 + return spectral_feature(pixels, geometry) + + +def discover_grouped_paths( + root: Path, + class_dirs: tuple[str, ...], +) -> tuple[list[Path], IntArray, IntArray, list[str]]: + """Return class-major paths over filename stems shared by every class.""" + if len(class_dirs) < 2 or len(set(class_dirs)) != len(class_dirs): + raise ValueError("at least two unique class directories are required") + per_class: list[dict[str, Path]] = [] + for class_dir in class_dirs: + directory = root / class_dir + if not directory.is_dir(): + raise ValueError(f"class directory does not exist: {directory}") + mapping: dict[str, Path] = {} + for path in sorted(directory.iterdir()): + if path.is_file() and path.suffix.lower() in IMAGE_SUFFIXES: + if path.stem in mapping: + raise ValueError(f"duplicate filename stem in {directory}: {path.stem}") + mapping[path.stem] = path + per_class.append(mapping) + common_names: set[str] = set(per_class[0]) + for mapping in per_class[1:]: + common_names.intersection_update(mapping) + group_names = sorted(common_names) + if not group_names: + raise ValueError("class directories have no shared filename stems") + paths: list[Path] = [] + labels: list[int] = [] + groups: list[int] = [] + for class_index, mapping in enumerate(per_class): + for group_index, group_name in enumerate(group_names): + paths.append(mapping[group_name]) + labels.append(class_index) + groups.append(group_index) + return paths, np.asarray(labels), np.asarray(groups), group_names + + +def fit_grouped_models( + features: FloatArray, + labels: IntArray, + groups: IntArray, + *, + class_count: int, + folds: int = 5, + ridge: float = 1000.0, +) -> list[FoldModel]: + """Fit prompt-grouped dual ridge models without test-fold leakage.""" + if features.ndim != 2 or len(features) != len(labels) or len(labels) != len(groups): + raise ValueError("features, labels, and groups must describe the same rows") + if folds < 2 or len(np.unique(groups)) < folds: + raise ValueError("fold count must be between two and the number of groups") + if class_count < 2 or ridge <= 0: + raise ValueError("class_count must be at least two and ridge must be positive") + targets = np.eye(class_count, dtype=np.float64)[labels] + models: list[FoldModel] = [] + for fold in range(folds): + test_rows = groups % folds == fold + train_rows = ~test_rows + train = features[train_rows] + mean = np.mean(train, axis=0) + scale = np.std(train, axis=0) + active = scale > 1e-8 + train_features = (train[:, active] - mean[active]) / scale[active] + kernel = train_features @ train_features.T + dual_weights = cast( + "FloatArray", + np.linalg.solve( + kernel + ridge * np.eye(len(kernel)), + targets[train_rows], + ), + ) + weights = train_features.T @ dual_weights + models.append(FoldModel(test_rows, mean, scale, active, weights)) + return models + + +def predict_grouped(models: list[FoldModel], features: FloatArray) -> IntArray: + """Predict every row only with the model that held out its group.""" + predictions = np.full(len(features), -1, dtype=np.int64) + for model in models: + test_features = (features[model.test_rows][:, model.active] - model.mean[model.active]) / model.scale[ + model.active + ] + scores = test_features @ model.weights + predictions[model.test_rows] = np.argmax(scores, axis=1) + if np.any(predictions < 0): + raise ValueError("fold models did not cover every row") + return predictions + + +def confusion_matrix(labels: IntArray, predictions: IntArray, class_count: int) -> IntArray: + """Count actual rows by predicted class.""" + matrix = np.zeros((class_count, class_count), dtype=np.int64) + np.add.at(matrix, (labels, predictions), 1) + return matrix + + +def _extract_all_attacks( + paths: list[Path], + geometry: SpectralGeometry, + attacks: tuple[str, ...], +) -> dict[str, FloatArray]: + """Decode each path once and extract every requested attack feature.""" + log.info("Extracting %s attack views from %s images", len(attacks), len(paths)) + rows: dict[str, list[FloatArray]] = {attack: [] for attack in attacks} + for path in paths: + with Image.open(path) as source: + image = source.convert("RGB") + for attack in attacks: + rows[attack].append(_feature_from_image(image, geometry, attack)) + return {attack: np.asarray(features) for attack, features in rows.items()} + + +@click.command() +@click.argument("dataset", type=click.Path(exists=True, file_okay=False, path_type=Path)) +@click.option( + "class_dirs", + "--class-dir", + multiple=True, + required=True, + help="Class subdirectory; repeat in label order.", +) +@click.option( + "attacks", + "--attack", + multiple=True, + type=click.Choice(ATTACKS), + default=("original",), + show_default=True, +) +@click.option("--size", type=click.IntRange(min=16), default=256, show_default=True) +@click.option("--grid", type=click.IntRange(min=2), default=32, show_default=True) +@click.option("--folds", type=click.IntRange(min=2), default=5, show_default=True) +@click.option("--ridge", type=click.FloatRange(min=0.0, min_open=True), default=1000.0, show_default=True) +@click.option("--report-out", type=click.Path(dir_okay=False, path_type=Path), required=True) +def main( + dataset: Path, + class_dirs: tuple[str, ...], + attacks: tuple[str, ...], + size: int, + grid: int, + folds: int, + ridge: float, + report_out: Path, +) -> None: + """Run grouped spectral pipeline classification over DATASET.""" + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") + if "original" not in attacks: + raise click.UsageError("--attack original is required because every model trains on original exports") + unique_attacks = tuple(dict.fromkeys(attacks)) + try: + geometry = make_geometry(size, grid) + paths, labels, groups, group_names = discover_grouped_paths(dataset, class_dirs) + features_by_attack = _extract_all_attacks(paths, geometry, unique_attacks) + original = features_by_attack["original"] + models = fit_grouped_models( + original, + labels, + groups, + class_count=len(class_dirs), + folds=folds, + ridge=ridge, + ) + except ValueError as error: + raise click.ClickException(str(error)) from error + + results: dict[str, object] = {} + for attack in unique_attacks: + features = features_by_attack[attack] + predictions = predict_grouped(models, features) + matrix = confusion_matrix(labels, predictions, len(class_dirs)) + results[attack] = { + "accuracy": float(np.trace(matrix) / np.sum(matrix)), + "confusion_matrix": matrix.tolist(), + } + + report = { + "claim": "phase-free generation/export pipeline classification; not SynthID detection", + "causal_limit": ( + "Renderer, output geometry, encoding, and watermark state vary together. " + "Only a marked/clean causal contrast plus an independent oracle can isolate the watermark." + ), + "evaluation_limit": ( + "Grouped cross-validation estimates this corpus only. Hyperparameters selected on these folds " + "need an independent locked corpus before they support a generalization claim." + ), + "dataset": str(dataset), + "classes": list(class_dirs), + "group_count": len(group_names), + "image_count": len(paths), + "configuration": {"size": size, "grid": grid, "folds": folds, "ridge": ridge}, + "results": results, + } + report_out.parent.mkdir(parents=True, exist_ok=True) + report_out.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") + log.info("Wrote spectral pipeline report: %s", report_out) + + +if __name__ == "__main__": + main() diff --git a/tests/test_spectral_pipeline_probe.py b/tests/test_spectral_pipeline_probe.py new file mode 100644 index 0000000..01e0bfb --- /dev/null +++ b/tests/test_spectral_pipeline_probe.py @@ -0,0 +1,81 @@ +"""Tests for the phase-free spectral pipeline research harness.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +from PIL import Image + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) + +import spectral_pipeline_probe as probe + + +def test_log_spectrum_discards_translation_phase() -> None: + rng = np.random.default_rng(17) + channel = rng.normal(size=(32, 32)) + + original = probe.phase_free_log_spectrum(channel) + translated = probe.phase_free_log_spectrum(np.roll(channel, shift=(5, -7), axis=(0, 1))) + + assert np.allclose(original, translated, rtol=1e-12, atol=1e-12) + + +def test_spectral_feature_is_finite_and_has_declared_shape() -> None: + geometry = probe.make_geometry(size=32, grid=4) + yy, xx = np.mgrid[:32, :32] + carrier = np.cos(2 * np.pi * (5 * xx + 3 * yy) / 32) + pixels = np.stack((carrier, np.roll(carrier, 1, axis=0), np.roll(carrier, 2, axis=1)), axis=2) + + feature = probe.spectral_feature(pixels, geometry) + + assert feature.shape == (3 * 4 * 4,) + assert np.all(np.isfinite(feature)) + assert np.linalg.norm(feature) > 0 + + +def test_discovery_keeps_only_stems_shared_by_every_class(tmp_path: Path) -> None: + for class_name in ("first", "second"): + directory = tmp_path / class_name + directory.mkdir() + for stem in ("001", "002"): + Image.new("RGB", (8, 8), "white").save(directory / f"{stem}.png") + Image.new("RGB", (8, 8), "black").save(tmp_path / "first" / "unpaired.png") + + paths, labels, groups, names = probe.discover_grouped_paths(tmp_path, ("first", "second")) + + assert names == ["001", "002"] + assert [path.stem for path in paths] == ["001", "002", "001", "002"] + assert labels.tolist() == [0, 0, 1, 1] + assert groups.tolist() == [0, 1, 0, 1] + + +def test_grouped_ridge_never_trains_on_the_test_group() -> None: + groups = np.tile(np.arange(14), 3) + labels = np.repeat(np.arange(3), 14) + class_centers = np.eye(3)[labels] * 8.0 + group_nuisance = np.column_stack((groups % 2, groups % 3, groups % 5)) * 0.01 + features = class_centers + group_nuisance + + models = probe.fit_grouped_models(features, labels, groups, class_count=3, folds=5, ridge=1.0) + predictions = probe.predict_grouped(models, features) + + assert np.array_equal(predictions, labels) + fold_assignment = np.full(len(groups), -1) + for fold, model in enumerate(models): + assert not np.any(fold_assignment[model.test_rows] >= 0) + fold_assignment[model.test_rows] = fold + assert np.all(fold_assignment >= 0) + for group in np.unique(groups): + assert len(np.unique(fold_assignment[groups == group])) == 1 + + +def test_confusion_matrix_uses_actual_rows_and_predicted_columns() -> None: + labels = np.asarray([0, 0, 1, 1, 2]) + predictions = np.asarray([0, 1, 1, 2, 0]) + + matrix = probe.confusion_matrix(labels, predictions, class_count=3) + + assert matrix.tolist() == [[1, 1, 0], [0, 1, 1], [1, 0, 0]]