mirror of
https://github.com/wiltodelta/remove-ai-watermarks.git
synced 2026-08-31 17:50:35 +02:00
Reframe periodic pixel route as pipeline lattice; confirm and harden detection
The frozen periodic experts read an origin-anchored generation-pipeline lattice destroyed by a crop off the tile grid, not the crop-robust SynthID mark. Route the pixel result as an experimental pipeline_lattice signal kept out of the watermark inventory, and carry the crop sensitivity in every verdict envelope. Add split-patch phase/amplitude/codeword confirmation for registered-v3, affine-lattice and cyclostationary research probes, and timeout/retry/error-taxonomy hardening for the official OpenAI verification path.
This commit is contained in:
Victor Kuznetsov
@@ -0,0 +1,228 @@
|
||||
"""Probe complex cross-spectral coupling at preregistered carrier shifts."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import math
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import click
|
||||
import cv2
|
||||
import numpy as np
|
||||
from synthid_affine_lattice_probe import (
|
||||
_canonical_pixels,
|
||||
_opponent_channels,
|
||||
_patch_origins,
|
||||
template_harmonics,
|
||||
)
|
||||
from synthid_pixel_attack import load_rgb
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from numpy.typing import NDArray
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_OFF_CARRIER_OFFSETS = ((1, 0), (-1, 0), (0, 1), (0, -1), (1, 1), (-1, -1))
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CyclostationaryScore:
|
||||
"""Complex carrier-versus-neighbor contrast on disjoint patch groups."""
|
||||
|
||||
selection_carrier: float
|
||||
selection_off_carrier_median: float
|
||||
selection_contrast: float
|
||||
confirmation_carrier: float
|
||||
confirmation_off_carrier_median: float
|
||||
confirmation_contrast: float
|
||||
joint_contrast: float
|
||||
harmonic_count: int
|
||||
selection_patches: int
|
||||
confirmation_patches: int
|
||||
|
||||
|
||||
def _patch_cyclic_matrices(
|
||||
pixels: NDArray[Any],
|
||||
harmonics: NDArray[Any],
|
||||
*,
|
||||
tile_size: int,
|
||||
denoise_sigma: float,
|
||||
band_min: float,
|
||||
band_max: float,
|
||||
) -> NDArray[Any]:
|
||||
"""Return normalized complex cross-channel matrices for one patch."""
|
||||
patch_size = pixels.shape[0]
|
||||
if pixels.shape[:2] != (patch_size, patch_size) or patch_size % tile_size:
|
||||
raise ValueError("cyclostationary patches must be square multiples of the tile size")
|
||||
channels = _opponent_channels(np.asarray(pixels, dtype=np.float64))
|
||||
for channel in range(channels.shape[2]):
|
||||
residual = channels[:, :, channel]
|
||||
channels[:, :, channel] = residual - cv2.GaussianBlur(
|
||||
residual,
|
||||
(0, 0),
|
||||
sigmaX=denoise_sigma,
|
||||
sigmaY=denoise_sigma,
|
||||
borderType=cv2.BORDER_REFLECT_101,
|
||||
)
|
||||
window_1d = np.hanning(patch_size)
|
||||
window = window_1d[:, None] * window_1d[None, :]
|
||||
spectrum = np.fft.fft2(channels * window[:, :, None], axes=(0, 1))
|
||||
frequency_y = np.fft.fftfreq(patch_size)[:, None]
|
||||
frequency_x = np.fft.fftfreq(patch_size)[None, :]
|
||||
radius = np.sqrt(frequency_y * frequency_y + frequency_x * frequency_x)
|
||||
base_mask = (radius >= band_min) & (radius <= band_max)
|
||||
|
||||
offset_values = ((0, 0), *_OFF_CARRIER_OFFSETS)
|
||||
matrices = np.empty((len(harmonics), len(offset_values), 3, 3), dtype=np.complex128)
|
||||
for harmonic_index, (signed_row_value, signed_column_value) in enumerate(harmonics):
|
||||
alpha_y = round(float(signed_row_value) * patch_size / tile_size)
|
||||
alpha_x = round(float(signed_column_value) * patch_size / tile_size)
|
||||
for offset_index, (offset_y, offset_x) in enumerate(offset_values):
|
||||
shift_y = alpha_y + offset_y
|
||||
shift_x = alpha_x + offset_x
|
||||
shifted = np.roll(spectrum, shift=(-shift_y, -shift_x), axis=(0, 1))
|
||||
mask = base_mask & np.roll(base_mask, shift=(-shift_y, -shift_x), axis=(0, 1))
|
||||
base_values = spectrum[mask]
|
||||
shifted_values = shifted[mask]
|
||||
normalizer = math.sqrt(float(np.sum(np.abs(base_values) ** 2)) * float(np.sum(np.abs(shifted_values) ** 2)))
|
||||
if normalizer <= 1e-12:
|
||||
matrices[harmonic_index, offset_index] = 0.0
|
||||
else:
|
||||
matrices[harmonic_index, offset_index] = shifted_values.T @ np.conj(base_values) / normalizer
|
||||
return matrices
|
||||
|
||||
|
||||
def _group_score(values: list[NDArray[Any]], harmonic_weights: NDArray[Any]) -> tuple[float, float, float]:
|
||||
if not values:
|
||||
raise ValueError("cyclostationary score needs at least one patch")
|
||||
mean_matrices = np.mean(np.stack(values), axis=0)
|
||||
coherence = np.linalg.norm(mean_matrices, axis=(2, 3))
|
||||
carrier = float(np.sum(coherence[:, 0] * harmonic_weights))
|
||||
off_scores = [
|
||||
float(np.sum(coherence[:, offset_index] * harmonic_weights)) for offset_index in range(1, coherence.shape[1])
|
||||
]
|
||||
off_median = float(np.median(off_scores))
|
||||
return carrier, off_median, carrier - off_median
|
||||
|
||||
|
||||
def score_cyclostationary(
|
||||
pixels: NDArray[Any],
|
||||
template: NDArray[Any],
|
||||
*,
|
||||
period: float,
|
||||
patch_size: int = 256,
|
||||
grid_size: int = 4,
|
||||
harmonic_count: int = 8,
|
||||
denoise_sigma: float = 1.0,
|
||||
band_min: float = 0.05,
|
||||
band_max: float = 0.35,
|
||||
) -> CyclostationaryScore:
|
||||
"""Measure split-confirmed complex spectral coupling at one period."""
|
||||
if pixels.ndim != 3 or pixels.shape[2] != 3:
|
||||
raise ValueError("pixels must have shape (height, width, 3)")
|
||||
if not math.isfinite(period) or period <= 0.0:
|
||||
raise ValueError("period must be finite and positive")
|
||||
if not (0.0 < band_min < band_max < 0.5):
|
||||
raise ValueError("frequency band must satisfy 0 < min < max < 0.5")
|
||||
if not math.isfinite(denoise_sigma) or denoise_sigma <= 0.0:
|
||||
raise ValueError("denoise sigma must be finite and positive")
|
||||
|
||||
canonical = _canonical_pixels(pixels, template, period)
|
||||
tile_size = template.shape[0]
|
||||
harmonics, channel_weights, _coefficient_units = template_harmonics(template, harmonic_count)
|
||||
harmonic_weights = np.linalg.norm(channel_weights, axis=1)
|
||||
harmonic_weights /= np.sum(harmonic_weights)
|
||||
grouped_values: dict[int, list[NDArray[Any]]] = {0: [], 1: []}
|
||||
for origin_y, origin_x, group in _patch_origins(*canonical.shape[:2], patch_size, grid_size):
|
||||
aligned_y = (origin_y // tile_size) * tile_size
|
||||
aligned_x = (origin_x // tile_size) * tile_size
|
||||
patch = canonical[aligned_y : aligned_y + patch_size, aligned_x : aligned_x + patch_size]
|
||||
grouped_values[group].append(
|
||||
_patch_cyclic_matrices(
|
||||
patch,
|
||||
harmonics,
|
||||
tile_size=tile_size,
|
||||
denoise_sigma=denoise_sigma,
|
||||
band_min=band_min,
|
||||
band_max=band_max,
|
||||
)
|
||||
)
|
||||
selection_carrier, selection_null, selection_contrast = _group_score(grouped_values[0], harmonic_weights)
|
||||
confirmation_carrier, confirmation_null, confirmation_contrast = _group_score(grouped_values[1], harmonic_weights)
|
||||
return CyclostationaryScore(
|
||||
selection_carrier=selection_carrier,
|
||||
selection_off_carrier_median=selection_null,
|
||||
selection_contrast=selection_contrast,
|
||||
confirmation_carrier=confirmation_carrier,
|
||||
confirmation_off_carrier_median=confirmation_null,
|
||||
confirmation_contrast=confirmation_contrast,
|
||||
joint_contrast=min(selection_contrast, confirmation_contrast),
|
||||
harmonic_count=len(harmonics),
|
||||
selection_patches=len(grouped_values[0]),
|
||||
confirmation_patches=len(grouped_values[1]),
|
||||
)
|
||||
|
||||
|
||||
def _load_template(path: Path) -> NDArray[Any]:
|
||||
with np.load(path, allow_pickle=False) as artifact:
|
||||
template = np.asarray(artifact["template"], dtype=np.float64)
|
||||
if template.shape != (16, 16, 3) or not np.all(np.isfinite(template)):
|
||||
raise ValueError("template artifact does not contain a finite 16x16 RGB template")
|
||||
return template
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.argument("template_path", type=click.Path(exists=True, dir_okay=False, path_type=Path))
|
||||
@click.argument("images", nargs=-1, required=True, type=click.Path(exists=True, dir_okay=False, path_type=Path))
|
||||
@click.option("--period", type=click.FloatRange(min=1.0), required=True)
|
||||
@click.option("--input-scale", type=click.FloatRange(min=0.01), default=1.0, show_default=True)
|
||||
@click.option("--report-out", type=click.Path(dir_okay=False, path_type=Path), required=True)
|
||||
def main(
|
||||
template_path: Path,
|
||||
images: tuple[Path, ...],
|
||||
period: float,
|
||||
input_scale: float,
|
||||
report_out: Path,
|
||||
) -> None:
|
||||
"""Score IMAGES for complex cross-spectral carrier coupling."""
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
template = _load_template(template_path)
|
||||
rows = []
|
||||
for path in images:
|
||||
try:
|
||||
pixels = load_rgb(path)
|
||||
if input_scale != 1.0:
|
||||
width = max(1, round(pixels.shape[1] * input_scale))
|
||||
height = max(1, round(pixels.shape[0] * input_scale))
|
||||
interpolation = cv2.INTER_AREA if input_scale < 1.0 else cv2.INTER_CUBIC
|
||||
pixels = cv2.resize(pixels, (width, height), interpolation=interpolation)
|
||||
score = score_cyclostationary(pixels, template, period=period)
|
||||
except ValueError as error:
|
||||
rows.append({"path": str(path), "status": "unsupported", "reason": str(error)})
|
||||
log.warning("%s: unsupported: %s", path, error)
|
||||
continue
|
||||
rows.append({"path": str(path), "status": "scored", "score": asdict(score)})
|
||||
log.info("%s: joint contrast=%.6f", path, score.joint_contrast)
|
||||
report_out.parent.mkdir(parents=True, exist_ok=True)
|
||||
report_out.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"schema_version": 1,
|
||||
"template": str(template_path),
|
||||
"period": period,
|
||||
"input_scale": input_scale,
|
||||
"records": rows,
|
||||
},
|
||||
indent=2,
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
log.info("Wrote %d cyclostationary records: %s", len(rows), report_out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user