CalvinBackup/remove-ai-watermarks
1
0
Fork 0
mirror of https://github.com/wiltodelta/remove-ai-watermarks.git synced 2026-08-31 09:40:38 +02:00
Code Issues Packages Projects Releases Wiki Activity

Remove lattice detector sources from the public package

Browse Source
This commit is contained in:
Victor Kuznetsov
2026-08-23 11:14:53 -07:00
parent d22872f84d
commit bc424a48f4
4 changed files with 0 additions and 1495 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/remove_ai_watermarks/_synthid_confirmation.py
-288
View File
@@ -1,288 +0,0 @@
"""Independent split-patch confirmation for the registered SynthID carrier."""
# The optional numeric libraries do not provide complete types for this path.
# pyright: reportMissingTypeStubs=false, reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
from __future__ import annotations
import math
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
import cv2
import numpy as np
from remove_ai_watermarks.synthid_detector import fold_residual_template, unit_tile
if TYPE_CHECKING:
from numpy.typing import NDArray
MIN_PERIOD = 10.0
MIN_COHERENCE = 0.30
MIN_AMPLITUDE = 0.0
H5_PERIOD = (18.0, 18.6)
H5_MIN = 0.13
STRONG_COHERENCE_PERIOD = (18.6, 20.0)
STRONG_COHERENCE_MIN = 0.40
WEAK_H5_PERIOD = (20.0, 22.0)
WEAK_H5_MIN = 0.02
PATCH_SIZE = 256
GRID_SIZE = 4
HARMONIC_COUNT = 16
@dataclass(frozen=True)
class RegisteredConfirmationComponents:
"""Auditable split-patch confirmation components for one fixed period."""
period: float
joint_coherence: float
joint_amplitude: float
unknown_codeword_fixed_confirmation: float
selection_patches: int
confirmation_patches: int
@property
def passes(self) -> bool:
"""Whether every frozen period-aware confirmation gate passes."""
return registered_confirmation_passes(
self.period,
self.joint_coherence,
self.joint_amplitude,
self.unknown_codeword_fixed_confirmation,
)
def registered_confirmation_passes(
period: float,
joint_coherence: float,
joint_amplitude: float,
unknown_codeword_fixed_confirmation: float,
) -> bool:
"""Apply the single frozen registered-carrier confirmation rule."""
if period < MIN_PERIOD:
return False
if joint_coherence < MIN_COHERENCE or joint_amplitude < MIN_AMPLITUDE:
return False
if H5_PERIOD[0] <= period < H5_PERIOD[1]:
return unknown_codeword_fixed_confirmation >= H5_MIN
if STRONG_COHERENCE_PERIOD[0] <= period < STRONG_COHERENCE_PERIOD[1]:
return joint_coherence >= STRONG_COHERENCE_MIN
if WEAK_H5_PERIOD[0] <= period < WEAK_H5_PERIOD[1]:
return unknown_codeword_fixed_confirmation >= WEAK_H5_MIN
return True
def _opponent_channels(values: NDArray[Any]) -> NDArray[Any]:
red = values[:, :, 0]
green = values[:, :, 1]
blue = values[:, :, 2]
return np.stack((green, red - green, blue - 0.5 * (red + green)), axis=2)
def _template_harmonics(template: NDArray[Any]) -> tuple[NDArray[Any], NDArray[Any]]:
opponent = _opponent_channels(np.asarray(template, dtype=np.float64))
spectrum = np.fft.fft2(opponent, axes=(0, 1))
height, width = template.shape[:2]
candidates: list[tuple[float, int, int]] = []
for row in range(height):
signed_row = row if row <= height // 2 else row - height
for column in range(width):
signed_column = column if column <= width // 2 else column - width
if signed_row < 0 or (signed_row == 0 and signed_column <= 0):
continue
power = float(np.sum(np.abs(spectrum[row, column]) ** 2))
candidates.append((power, signed_row, signed_column))
candidates.sort(reverse=True)
selected = candidates[:HARMONIC_COUNT]
harmonics = np.asarray([(row, column) for _power, row, column in selected], dtype=np.float64)
coefficients = np.asarray([spectrum[int(row) % height, int(column) % width] for row, column in harmonics])
weights = np.abs(coefficients)
weight_sum = float(np.sum(weights))
if weight_sum <= 0.0:
raise ValueError("template has no nonzero periodic harmonics")
return harmonics, weights / weight_sum
def _patch_origins(height: int, width: int) -> list[tuple[int, int, int]]:
if height < PATCH_SIZE or width < PATCH_SIZE:
raise ValueError("registered confirmation needs both image sides to be at least 256 pixels")
y_values = np.linspace(0, height - PATCH_SIZE, min(GRID_SIZE, height // PATCH_SIZE), dtype=np.int64)
x_values = np.linspace(0, width - PATCH_SIZE, min(GRID_SIZE, width // PATCH_SIZE), dtype=np.int64)
origins = [
(int(y), int(x), (y_index + x_index) % 2)
for y_index, y in enumerate(np.unique(y_values))
for x_index, x in enumerate(np.unique(x_values))
]
if {group for _y, _x, group in origins} != {0, 1}:
raise ValueError("registered confirmation needs two independent patch groups")
return origins
def _bilinear_sample(spectrum: NDArray[Any], y: NDArray[Any], x: NDArray[Any]) -> NDArray[Any]:
height, width = spectrum.shape
y_floor = np.floor(y)
x_floor = np.floor(x)
y0 = y_floor.astype(np.int64) % height
x0 = x_floor.astype(np.int64) % width
y1 = (y0 + 1) % height
x1 = (x0 + 1) % width
dy = y - y_floor
dx = x - x_floor
return (
spectrum[y0, x0] * (1.0 - dy) * (1.0 - dx)
+ spectrum[y1, x0] * dy * (1.0 - dx)
+ spectrum[y0, x1] * (1.0 - dy) * dx
+ spectrum[y1, x1] * dy * dx
)
def _patch_unit_values(
pixels: NDArray[Any],
origin_y: int,
origin_x: int,
period: float,
harmonics: NDArray[Any],
denoise_sigma: float,
) -> NDArray[Any]:
patch = np.asarray(
pixels[origin_y : origin_y + PATCH_SIZE, origin_x : origin_x + PATCH_SIZE],
dtype=np.float32,
)
channels = _opponent_channels(patch)
window_1d = np.hanning(PATCH_SIZE).astype(np.float32)
window = window_1d[:, None] * window_1d[None, :]
frequencies_y = harmonics[:, 0] / period
frequencies_x = harmonics[:, 1] / period
sample_y = frequencies_y * PATCH_SIZE
sample_x = frequencies_x * PATCH_SIZE
sampled = np.empty((len(harmonics), 3), dtype=np.complex128)
for channel in range(3):
residual = channels[:, :, channel]
residual -= cv2.GaussianBlur(
residual,
(0, 0),
sigmaX=denoise_sigma,
sigmaY=denoise_sigma,
borderType=cv2.BORDER_REFLECT_101,
)
sampled[:, channel] = _bilinear_sample(np.fft.fft2(residual * window), sample_y, sample_x)
sampled *= np.exp(-2j * math.pi * (frequencies_y * origin_y + frequencies_x * origin_x))[:, None]
magnitudes = np.abs(sampled)
return np.divide(sampled, magnitudes, out=np.zeros_like(sampled), where=magnitudes > 1e-12)
def _coherence(values: list[NDArray[Any]], weights: NDArray[Any]) -> float:
coherence = np.abs(np.mean(np.stack(values), axis=0))
return float(np.sum(coherence * weights))
def _unknown_codeword_fixed_confirmation(
selection_values: list[NDArray[Any]],
confirmation_values: list[NDArray[Any]],
weights: NDArray[Any],
) -> float:
cross_codeword = np.mean(np.stack(confirmation_values), axis=0) * np.conj(
np.mean(np.stack(selection_values), axis=0)
)
confirmation_mask = np.arange(len(weights)) % 2 == 1
masked_weights = weights[confirmation_mask]
return float(np.abs(np.sum(cross_codeword[confirmation_mask] * masked_weights)) / np.sum(masked_weights))
def _canonical_pixels(pixels: NDArray[Any], template: NDArray[Any], period: float) -> NDArray[Any]:
width = max(template.shape[1], round(pixels.shape[1] * template.shape[1] / period))
height = max(template.shape[0], round(pixels.shape[0] * template.shape[0] / period))
if (height, width) == pixels.shape[:2]:
return pixels
interpolation = cv2.INTER_AREA if width < pixels.shape[1] else cv2.INTER_CUBIC
return np.asarray(cv2.resize(pixels, (width, height), interpolation=interpolation))
def _cyclic_correlations(template: NDArray[Any], tile: NDArray[Any]) -> NDArray[Any]:
template_spectrum = np.fft.fft2(template, axes=(0, 1))
tile_spectrum = np.fft.fft2(tile, axes=(0, 1))
return np.fft.ifft2(np.sum(template_spectrum * np.conj(tile_spectrum), axis=2)).real
def _joint_amplitude(
pixels: NDArray[Any],
template: NDArray[Any],
period: float,
denoise_sigma: float,
) -> tuple[float, int, int]:
canonical = _canonical_pixels(pixels, template, period)
tile_height, tile_width = template.shape[:2]
grouped_units: dict[int, list[NDArray[Any]]] = {0: [], 1: []}
origins = _patch_origins(*canonical.shape[:2])
for origin_y, origin_x, group in origins:
aligned_y = (origin_y // tile_height) * tile_height
aligned_x = (origin_x // tile_width) * tile_width
folded = fold_residual_template(
canonical[aligned_y : aligned_y + PATCH_SIZE, aligned_x : aligned_x + PATCH_SIZE],
tile_height=tile_height,
tile_width=tile_width,
denoise_sigma=denoise_sigma,
)
unit, _norm = unit_tile(folded)
grouped_units[group].append(unit)
selection_tile, _selection_norm = unit_tile(np.mean(grouped_units[0], axis=0))
confirmation_tile, _confirmation_norm = unit_tile(np.mean(grouped_units[1], axis=0))
selection_correlations = _cyclic_correlations(template, selection_tile)
confirmation_correlations = _cyclic_correlations(template, confirmation_tile)
shift_y, shift_x = np.unravel_index(int(np.argmax(selection_correlations)), selection_correlations.shape)
return (
min(
float(selection_correlations[shift_y, shift_x]),
float(confirmation_correlations[shift_y, shift_x]),
),
len(grouped_units[0]),
len(grouped_units[1]),
)
def registered_confirmation_components(
pixels: NDArray[Any],
template: NDArray[Any],
period: float,
denoise_sigma: float,
) -> RegisteredConfirmationComponents:
"""Measure the frozen split-patch gates at one registered carrier 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("registered period must be finite and positive")
harmonics, weights = _template_harmonics(template)
grouped_values: dict[int, list[NDArray[Any]]] = {0: [], 1: []}
for origin_y, origin_x, group in _patch_origins(*pixels.shape[:2]):
grouped_values[group].append(
_patch_unit_values(
pixels,
origin_y,
origin_x,
period,
harmonics,
denoise_sigma,
)
)
amplitude, selection_patches, confirmation_patches = _joint_amplitude(
pixels,
template,
period,
denoise_sigma,
)
return RegisteredConfirmationComponents(
period=period,
joint_coherence=min(
_coherence(grouped_values[0], weights),
_coherence(grouped_values[1], weights),
),
joint_amplitude=amplitude,
unknown_codeword_fixed_confirmation=_unknown_codeword_fixed_confirmation(
grouped_values[0],
grouped_values[1],
weights,
),
selection_patches=selection_patches,
confirmation_patches=confirmation_patches,
)
src/remove_ai_watermarks/_synthid_registered.py
-664
View File
@@ -1,664 +0,0 @@
"""Opt-in scale registration for the measured periodic SynthID carrier."""
# The optional numeric libraries do not provide complete types for this path.
# pyright: reportMissingTypeStubs=false, reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
from __future__ import annotations
import itertools
import math
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
import cv2
import numpy as np
from remove_ai_watermarks._synthid_confirmation import (
RegisteredConfirmationComponents,
registered_confirmation_components,
)
from remove_ai_watermarks.synthid_detector import folded_template_score
if TYPE_CHECKING:
from numpy.typing import NDArray
_PYRAMID_SCALES = (0.75, 1.0, 1.25)
_SEARCH_PERIODS = np.linspace(5.0, 32.0, 541, dtype=np.float64)
_CANONICAL_PERIODS = np.linspace(7.5, 24.5, 1701, dtype=np.float64)
_OPPONENT_SEARCH_PERIODS = np.linspace(7.5, 14.5, 141, dtype=np.float64)
_FINE_OPPONENT_COARSE_PERIODS = np.linspace(7.5, 9.0, 31, dtype=np.float64)
_FINE_OPPONENT_PROBE_SIZE = 384
_PERIOD_THRESHOLDS = (
(7.5, 8.5, 0.3770629524888979),
(8.5, 10.0, 0.25174716660523494),
(10.0, 12.0, 0.284692023502354),
(12.0, 14.0, 0.19794247706938645),
(14.0, 16.0, 0.33930082812296375),
(16.0, 18.0, 0.28915284982686323),
(18.0, 20.0, 0.22885510746595789),
(20.0, 22.0, 0.24570317032768269),
(22.0, 24.5, 0.3142958338390489),
)
REGISTERED_HIGH_BAND_THRESHOLD = 0.075
OPPONENT_REGISTERED_MIN_PERIOD = 7.9
OPPONENT_REGISTERED_MAX_PERIOD = 12.0
OPPONENT_REGISTERED_CODEC_VETO_MAX_PERIOD = 8.1
OPPONENT_REGISTERED_MAX_P8_EDGE_RATIO = 1.05
FINE_OPPONENT_REGISTERED_MIN_PERIOD = 7.5
FINE_OPPONENT_REGISTERED_MAX_PERIOD = 9.0
OPPONENT_REGISTERED_FIXED_MIN = 0.16
OPPONENT_REGISTERED_RED_GREEN_MIN = 0.60
OPPONENT_REGISTERED_BLUE_YELLOW_MIN = 0.55
@dataclass(frozen=True)
class RegisteredComponents:
"""Calibrated components of one scale-registered decision."""
raw_score: float
amplitude_threshold: float
selected_period: float
spectral_period: float
high_band_score: float
confirmation: RegisteredConfirmationComponents | None = None
@property
def base_decision_score(self) -> float:
"""Return the unchanged registered-v2 decision statistic."""
if self.selected_period != self.spectral_period:
return 0.0
return min(
self.raw_score / self.amplitude_threshold,
self.high_band_score / REGISTERED_HIGH_BAND_THRESHOLD,
)
@property
def decision_score(self) -> float:
"""Return the base score only after split confirmation passes."""
base_score = self.base_decision_score
if base_score < 1.0:
return base_score
if self.confirmation is None or not self.confirmation.passes:
return 0.0
return base_score
@dataclass(frozen=True)
class OpponentRegisteredComponents:
"""Auditable margins for the bounded opponent-color fallback."""
selected_period: float
spectral_period: float
spectral_score: float
fixed_score: float
red_green_spatial: float
blue_yellow_spatial: float
candidate_count: int
red_green_p8_edge_ratio: float | None
blue_yellow_p8_edge_ratio: float | None
@property
def base_decision_score(self) -> float:
"""Return the minimum normalized color-carrier margin."""
return min(
self.fixed_score / OPPONENT_REGISTERED_FIXED_MIN,
self.red_green_spatial / OPPONENT_REGISTERED_RED_GREEN_MIN,
self.blue_yellow_spatial / OPPONENT_REGISTERED_BLUE_YELLOW_MIN,
)
@property
def decision_score(self) -> float:
"""Return the margin only inside the independently challenged period band."""
if not OPPONENT_REGISTERED_MIN_PERIOD <= self.selected_period <= OPPONENT_REGISTERED_MAX_PERIOD:
return 0.0
if self.selected_period <= OPPONENT_REGISTERED_CODEC_VETO_MAX_PERIOD:
ratios = (self.red_green_p8_edge_ratio, self.blue_yellow_p8_edge_ratio)
if any(value is None or value > OPPONENT_REGISTERED_MAX_P8_EDGE_RATIO for value in ratios):
return 0.0
return self.base_decision_score
@property
def fine_decision_score(self) -> float:
"""Return the margin for the separately calibrated fine-period expert."""
if not FINE_OPPONENT_REGISTERED_MIN_PERIOD <= self.selected_period <= FINE_OPPONENT_REGISTERED_MAX_PERIOD:
return 0.0
if self.selected_period <= OPPONENT_REGISTERED_CODEC_VETO_MAX_PERIOD:
ratios = (self.red_green_p8_edge_ratio, self.blue_yellow_p8_edge_ratio)
if any(value is None or value > OPPONENT_REGISTERED_MAX_P8_EDGE_RATIO for value in ratios):
return 0.0
return self.base_decision_score
def _resize(pixels: NDArray[Any], width: int, height: int) -> NDArray[Any]:
interpolation = cv2.INTER_AREA if width < pixels.shape[1] else cv2.INTER_CUBIC
return np.asarray(cv2.resize(pixels, (width, height), interpolation=interpolation))
def _template_frequency_features(
template: NDArray[Any],
) -> tuple[NDArray[Any], NDArray[Any], NDArray[Any]]:
spectrum = np.fft.fft2(template, axes=(0, 1))
power = np.sum(np.abs(spectrum) ** 2, axis=2)
power[0, 0] = 0.0
indices = np.argsort(power.ravel())[::-1][:30]
rows, columns = np.unravel_index(indices, power.shape)
height, width = template.shape[:2]
signed_rows = np.where(rows <= height // 2, rows, rows - height)
signed_columns = np.where(columns <= width // 2, columns, columns - width)
harmonics = np.column_stack((signed_rows, signed_columns)).astype(np.float64)
return harmonics, spectrum[rows, columns], spectrum
def _bilinear_sample(
spectrum: NDArray[Any],
y: NDArray[Any],
x: NDArray[Any],
) -> NDArray[Any]:
height, width = spectrum.shape
y_floor = np.floor(y)
x_floor = np.floor(x)
y0 = y_floor.astype(np.int64) % height
x0 = x_floor.astype(np.int64) % width
y1 = (y0 + 1) % height
x1 = (x0 + 1) % width
dy = y - y_floor
dx = x - x_floor
return (
spectrum[y0, x0] * (1.0 - dy) * (1.0 - dx)
+ spectrum[y1, x0] * dy * (1.0 - dx)
+ spectrum[y0, x1] * (1.0 - dy) * dx
+ spectrum[y1, x1] * dy * dx
)
def _spectral_curve(
pixels: NDArray[Any],
periods: NDArray[Any],
harmonics: NDArray[Any],
coefficients: NDArray[Any],
) -> NDArray[Any]:
height, width = pixels.shape[:2]
y = (periods[:, None] ** -1) * harmonics[None, :, 0] * height
x = (periods[:, None] ** -1) * harmonics[None, :, 1] * width
sampled = np.empty((len(periods), len(harmonics), 3), dtype=np.complex128)
for channel in range(3):
residual = pixels[:, :, channel].astype(np.float32)
residual -= cv2.GaussianBlur(
residual,
(0, 0),
sigmaX=1.0,
sigmaY=1.0,
borderType=cv2.BORDER_REFLECT_101,
)
spectrum = np.fft.fft2(residual)
sampled[:, :, channel] = _bilinear_sample(spectrum, y % height, x % width)
numerator = np.real(np.sum(np.conj(coefficients)[None, :, :] * sampled, axis=(1, 2)))
denominator = np.linalg.norm(coefficients) * np.linalg.norm(sampled, axis=(1, 2))
return np.divide(
numerator,
denominator,
out=np.zeros_like(numerator),
where=denominator > 0.0,
)
def _period_candidates(
periods: NDArray[Any],
scores: NDArray[Any],
count: int = 3,
) -> list[float]:
candidates: list[float] = []
for index in np.argsort(scores)[::-1]:
period = float(periods[index])
if any(abs(period - existing_period) < 0.25 for existing_period in candidates):
continue
candidates.append(period)
if len(candidates) == count:
break
return candidates
def _period_threshold(period: float) -> float:
for index, (lower, upper, threshold) in enumerate(_PERIOD_THRESHOLDS):
if lower <= period < upper or (index == len(_PERIOD_THRESHOLDS) - 1 and period == upper):
return threshold
raise ValueError(f"registered period {period} is outside the calibrated range")
def _high_band_score(
folded: NDArray[Any],
template_spectrum: NDArray[Any],
) -> float:
folded_spectrum = np.fft.fft2(folded, axes=(0, 1))
tile_height, tile_width = template_spectrum.shape[:2]
y_coordinates = np.minimum(np.arange(tile_height), tile_height - np.arange(tile_height))
x_coordinates = np.minimum(np.arange(tile_width), tile_width - np.arange(tile_width))
radius = np.sqrt(y_coordinates[:, None] ** 2 + x_coordinates[None, :] ** 2)
correlations = []
for lower, upper in ((4.5, 6.5), (6.5, 12.0)):
mask = (radius >= lower) & (radius < upper)
selected_folded = folded_spectrum[mask]
selected_template = template_spectrum[mask]
denominator = np.linalg.norm(selected_folded) * np.linalg.norm(selected_template)
correlations.append(
float(np.real(np.vdot(selected_template, selected_folded)) / denominator) if denominator > 0.0 else 0.0
)
return min(correlations)
def _best_canonical(
pixels: NDArray[Any],
periods: list[float],
template: NDArray[Any],
sigma: float,
) -> tuple[float, NDArray[Any], NDArray[Any], float]:
best_score = -math.inf
best_canonical: NDArray[Any] | None = None
best_folded: NDArray[Any] | None = None
best_period: float | None = None
for period in periods:
predicted_width = round(pixels.shape[1] * template.shape[1] / period)
for delta in range(-4, 5):
width = predicted_width + delta
height = round(pixels.shape[0] * width / pixels.shape[1])
canonical = _resize(pixels, width, height)
score, folded = folded_template_score(canonical, template, sigma)
if score > best_score:
best_score = score
best_canonical = canonical
best_folded = folded
best_period = period
if best_canonical is None or best_folded is None or best_period is None:
raise RuntimeError("scale registration produced no canonical view")
return float(best_score), best_canonical, best_folded, best_period
def _quadrant_median(
canonical: NDArray[Any],
template: NDArray[Any],
sigma: float,
) -> float:
tile_height, tile_width = template.shape[:2]
split_y = max(tile_height, (canonical.shape[0] // (2 * tile_height)) * tile_height)
split_x = max(tile_width, (canonical.shape[1] // (2 * tile_width)) * tile_width)
scores = []
for region in (
canonical[:split_y, :split_x],
canonical[:split_y, split_x:],
canonical[split_y:, :split_x],
canonical[split_y:, split_x:],
):
score, _folded = folded_template_score(region, template, sigma)
scores.append(score)
return float(np.median(scores))
def _pyramid_locked_mean(
pixels: NDArray[Any],
harmonics: NDArray[Any],
coefficients: NDArray[Any],
base_curve: NDArray[Any],
) -> float:
curves = []
candidates = []
for scale in _PYRAMID_SCALES:
if scale == 1.0:
curve = base_curve
else:
level = _resize(
pixels,
max(16, round(pixels.shape[1] * scale)),
max(16, round(pixels.shape[0] * scale)),
)
curve = _spectral_curve(level, _SEARCH_PERIODS, harmonics, coefficients)
curves.append(curve)
candidates.append(_period_candidates(_SEARCH_PERIODS, curve))
combinations = itertools.product(*candidates)
def spread(combination: tuple[float, ...]) -> float:
normalized_periods = [
candidate / scale
for candidate, scale in zip(
combination,
_PYRAMID_SCALES,
strict=True,
)
]
return float(np.std(np.log(normalized_periods)))
best = min(
combinations,
key=spread,
)
base_period = float(np.median([candidate / scale for candidate, scale in zip(best, _PYRAMID_SCALES, strict=True)]))
locked = [
float(np.interp(base_period * scale, _SEARCH_PERIODS, curve))
for curve, scale in zip(curves, _PYRAMID_SCALES, strict=True)
]
return float(np.mean(locked))
def _opponent_pair(values: NDArray[Any]) -> NDArray[Any]:
"""Return Red-minus-Green and Blue-minus-Yellow color planes."""
red = values[:, :, 0]
green = values[:, :, 1]
blue = values[:, :, 2]
return np.stack((red - green, blue - 0.5 * (red + green)), axis=2)
def _opponent_period_curve(
pixels: NDArray[Any],
template: NDArray[Any],
periods: NDArray[Any] = _OPPONENT_SEARCH_PERIODS,
) -> NDArray[Any]:
"""Return signed opponent-color coherence across the frozen search grid."""
template_opponent = _opponent_pair(np.asarray(template, dtype=np.float64))
template_spectrum = np.fft.fft2(template_opponent, axes=(0, 1))
power = np.sum(np.abs(template_spectrum) ** 2, axis=2)
power[0, 0] = 0.0
indices = np.argsort(power.ravel())[::-1][:30]
rows, columns = np.unravel_index(indices, power.shape)
height, width = template.shape[:2]
signed_rows = np.where(rows <= height // 2, rows, rows - height)
signed_columns = np.where(columns <= width // 2, columns, columns - width)
harmonics = np.column_stack((signed_rows, signed_columns)).astype(np.float64)
coefficients = template_spectrum[rows, columns]
image_height, image_width = pixels.shape[:2]
sample_y = periods[:, None] ** -1 * harmonics[None, :, 0] * image_height
sample_x = periods[:, None] ** -1 * harmonics[None, :, 1] * image_width
sampled = np.empty((len(periods), len(harmonics), 2), dtype=np.complex128)
image_opponent = _opponent_pair(np.asarray(pixels, dtype=np.float32))
for channel in range(2):
residual = image_opponent[:, :, channel]
residual -= cv2.GaussianBlur(
residual,
(0, 0),
sigmaX=1.0,
sigmaY=1.0,
borderType=cv2.BORDER_REFLECT_101,
)
sampled[:, :, channel] = _bilinear_sample(
np.fft.fft2(residual),
sample_y % image_height,
sample_x % image_width,
)
numerator = np.real(np.sum(np.conj(coefficients)[None, :, :] * sampled, axis=(1, 2)))
denominator = np.linalg.norm(coefficients) * np.linalg.norm(sampled, axis=(1, 2))
return np.divide(numerator, denominator, out=np.zeros_like(numerator), where=denominator > 0.0)
def _opponent_period_candidates(scores: NDArray[Any], count: int = 3) -> list[int]:
"""Return separated period indices in descending spectral-score order."""
candidates: list[int] = []
for index in np.argsort(scores)[::-1]:
period = float(_OPPONENT_SEARCH_PERIODS[index])
if any(abs(period - float(_OPPONENT_SEARCH_PERIODS[prior])) < 0.5 for prior in candidates):
continue
candidates.append(int(index))
if len(candidates) == count:
break
return candidates
def _canonical_at_period(
pixels: NDArray[Any],
template: NDArray[Any],
period: float,
) -> NDArray[Any]:
"""Resample PIXELS so PERIOD maps to the frozen template period."""
width = max(template.shape[1], round(pixels.shape[1] * template.shape[1] / period))
height = max(template.shape[0], round(pixels.shape[0] * template.shape[0] / period))
if (height, width) == pixels.shape[:2]:
return pixels
return _resize(pixels, width, height)
def _correlation(left: NDArray[Any], right: NDArray[Any]) -> float:
"""Return the signed real cosine between equal-shaped arrays."""
denominator = float(np.linalg.norm(left) * np.linalg.norm(right))
return float(np.real(np.vdot(right, left)) / denominator) if denominator > 0.0 else 0.0
def _period8_edge_ratio(values: NDArray[Any]) -> float:
"""Measure native 8-pixel block edges relative to non-block phases."""
phase_values = np.zeros(8, dtype=np.float64)
for axis in (0, 1):
differences = np.abs(np.diff(values, axis=axis))
indices = np.arange(differences.shape[axis])
for phase in range(8):
selected = indices[(indices + 1) % 8 == phase]
phase_values[phase] += 0.5 * float(np.take(differences, selected, axis=axis).mean())
baseline = float(np.median(phase_values[[1, 2, 3, 5, 6, 7]]))
return float(phase_values[0] / baseline) if baseline > 1e-9 else math.inf
def _period8_opponent_edge_ratios(pixels: NDArray[Any]) -> tuple[float, float]:
"""Return codec-grid ratios for the two opponent-color planes."""
opponent = _opponent_pair(np.asarray(pixels, dtype=np.float32))
return _period8_edge_ratio(opponent[:, :, 0]), _period8_edge_ratio(opponent[:, :, 1])
def _opponent_components_at_period(
pixels: NDArray[Any],
template: NDArray[Any],
sigma: float,
period: float,
*,
spectral_period: float,
spectral_score: float,
candidate_count: int,
period8_edge_ratios: tuple[float, float] | None = None,
) -> OpponentRegisteredComponents:
"""Measure one period without selecting it from the image being scored."""
canonical = _canonical_at_period(pixels, template, period)
fixed_score, folded = folded_template_score(canonical, template, sigma)
folded_opponent = _opponent_pair(folded)
template_opponent = _opponent_pair(template)
red_green_p8_edge_ratio, blue_yellow_p8_edge_ratio = period8_edge_ratios or (None, None)
return OpponentRegisteredComponents(
selected_period=period,
spectral_period=spectral_period,
spectral_score=spectral_score,
fixed_score=fixed_score,
red_green_spatial=_correlation(folded_opponent[:, :, 0], template_opponent[:, :, 0]),
blue_yellow_spatial=_correlation(folded_opponent[:, :, 1], template_opponent[:, :, 1]),
candidate_count=candidate_count,
red_green_p8_edge_ratio=red_green_p8_edge_ratio,
blue_yellow_p8_edge_ratio=blue_yellow_p8_edge_ratio,
)
def opponent_registered_components(
pixels: NDArray[Any],
template: NDArray[Any],
sigma: float,
) -> OpponentRegisteredComponents:
"""Measure the bounded lossless-resize carrier in opponent-color space."""
curve = _opponent_period_curve(pixels, template)
candidate_indices = _opponent_period_candidates(curve)
observations: list[OpponentRegisteredComponents] = []
period8_edge_ratios: tuple[float, float] | None = None
for index in candidate_indices:
period = float(_OPPONENT_SEARCH_PERIODS[index])
if period <= OPPONENT_REGISTERED_CODEC_VETO_MAX_PERIOD and period8_edge_ratios is None:
period8_edge_ratios = _period8_opponent_edge_ratios(pixels)
observations.append(
_opponent_components_at_period(
pixels,
template,
sigma,
period,
spectral_period=float(_OPPONENT_SEARCH_PERIODS[int(np.argmax(curve))]),
spectral_score=float(curve[index]),
candidate_count=len(candidate_indices),
period8_edge_ratios=period8_edge_ratios,
)
)
if not observations:
raise RuntimeError("opponent-color registration produced no candidates")
return max(observations, key=lambda observation: observation.base_decision_score)
def _fine_opponent_period_groups(curve: NDArray[Any]) -> list[list[float]]:
"""Return fine period grids around separated absolute spectral peaks."""
centers: list[float] = []
for index in np.argsort(np.abs(curve))[::-1]:
period = float(_FINE_OPPONENT_COARSE_PERIODS[index])
if any(abs(period - existing) < 0.2 for existing in centers):
continue
centers.append(period)
if len(centers) == 3:
break
return [
sorted(
{
round(float(period), 2)
for period in np.arange(center - 0.36, center + 0.361, 0.01)
if FINE_OPPONENT_REGISTERED_MIN_PERIOD <= period <= FINE_OPPONENT_REGISTERED_MAX_PERIOD
}
)
for center in centers
]
def fine_opponent_registered_components(
pixels: NDArray[Any],
template: NDArray[Any],
sigma: float,
) -> OpponentRegisteredComponents:
"""Select and score the calibrated fine-period lossless-resize expert."""
curve = _opponent_period_curve(pixels, template, _FINE_OPPONENT_COARSE_PERIODS)
spectral_index = int(np.argmax(np.abs(curve)))
spectral_period = float(_FINE_OPPONENT_COARSE_PERIODS[spectral_index])
period_groups = _fine_opponent_period_groups(curve)
probe = pixels[
: min(_FINE_OPPONENT_PROBE_SIZE, pixels.shape[0]),
: min(_FINE_OPPONENT_PROBE_SIZE, pixels.shape[1]),
]
candidate_count = sum(len(group) for group in period_groups)
unique_periods = sorted({period for group in period_groups for period in group})
probe_by_period = {
period: _opponent_components_at_period(
probe,
template,
sigma,
period,
spectral_period=spectral_period,
spectral_score=float(np.interp(period, _FINE_OPPONENT_COARSE_PERIODS, curve)),
candidate_count=candidate_count,
)
for period in unique_periods
}
probe_groups = [[probe_by_period[period] for period in group] for group in period_groups]
probe_observations = [observation for group in probe_groups for observation in group]
finalist_periods = {
observation.selected_period
for group in probe_groups
for observation in sorted(group, key=lambda value: value.base_decision_score, reverse=True)[:2]
}
finalist_periods.update(
observation.selected_period
for observation in sorted(
probe_observations,
key=lambda value: value.base_decision_score,
reverse=True,
)[:5]
)
period8_edge_ratios = (
_period8_opponent_edge_ratios(pixels)
if any(period <= OPPONENT_REGISTERED_CODEC_VETO_MAX_PERIOD for period in finalist_periods)
else None
)
observations = [
_opponent_components_at_period(
pixels,
template,
sigma,
period,
spectral_period=spectral_period,
spectral_score=float(np.interp(period, _FINE_OPPONENT_COARSE_PERIODS, curve)),
candidate_count=len(probe_observations),
period8_edge_ratios=period8_edge_ratios,
)
for period in sorted(finalist_periods)
]
if not observations:
raise RuntimeError("fine opponent-color registration produced no candidates")
return max(observations, key=lambda observation: observation.base_decision_score)
def registered_components(
pixels: NDArray[Any],
template: NDArray[Any],
sigma: float,
) -> RegisteredComponents:
"""Measure a carrier after bounded scale registration."""
harmonics, coefficients, template_spectrum = _template_frequency_features(template)
combined_periods = np.concatenate((_SEARCH_PERIODS, _CANONICAL_PERIODS))
combined_curve = _spectral_curve(pixels, combined_periods, harmonics, coefficients)
base_curve = combined_curve[: len(_SEARCH_PERIODS)]
canonical_curve = combined_curve[len(_SEARCH_PERIODS) :]
candidates = _period_candidates(_CANONICAL_PERIODS, canonical_curve)
baseline, canonical, folded, selected_period = _best_canonical(pixels, candidates, template, sigma)
quadrant = _quadrant_median(canonical, template, sigma)
pyramid = _pyramid_locked_mean(
pixels,
harmonics,
coefficients,
base_curve,
)
raw_score = float((baseline + quadrant + pyramid) / 3.0)
components = RegisteredComponents(
raw_score=raw_score,
amplitude_threshold=_period_threshold(selected_period),
selected_period=selected_period,
spectral_period=candidates[0],
high_band_score=_high_band_score(folded, template_spectrum),
)
if components.base_decision_score < 1.0:
return components
try:
confirmation = registered_confirmation_components(
pixels,
template,
selected_period,
sigma,
)
except ValueError:
return components
return RegisteredComponents(
raw_score=components.raw_score,
amplitude_threshold=components.amplitude_threshold,
selected_period=components.selected_period,
spectral_period=components.spectral_period,
high_band_score=components.high_band_score,
confirmation=confirmation,
)
def registered_score(
pixels: NDArray[Any],
template: NDArray[Any],
sigma: float,
) -> float:
"""Return the calibrated registered decision statistic."""
return registered_components(pixels, template, sigma).decision_score
def opponent_registered_score(
pixels: NDArray[Any],
template: NDArray[Any],
sigma: float,
) -> float:
"""Return the bounded opponent-color fallback decision statistic."""
return opponent_registered_components(pixels, template, sigma).decision_score
def fine_opponent_registered_score(
pixels: NDArray[Any],
template: NDArray[Any],
sigma: float,
) -> float:
"""Return the separately calibrated fine-period decision statistic."""
return fine_opponent_registered_components(pixels, template, sigma).fine_decision_score
src/remove_ai_watermarks/assets/synthid_periodic_tile_2048_v1.npz
BIN
View File
Binary file not shown.
src/remove_ai_watermarks/synthid_detector.py
-543
View File
@@ -1,543 +0,0 @@
"""Detect the confirmed periodic SynthID image carrier at calibrated image sizes.
This is a positive-only detector for one measured carrier epoch, not Google's
private payload decoder. A positive result is strong local evidence for the
carrier. An indeterminate result means only that the selected detector did not
find it; image sizes outside that mode's calibrated range are reported separately.
The numeric runtime requires the ``pixels`` extra. Imports remain lazy so the
package's metadata-only paths stay dependency-light.
"""
# The optional numeric libraries do not provide complete types for this path.
# pyright: reportMissingTypeStubs=false, reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
from __future__ import annotations
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal
if TYPE_CHECKING:
from numpy.typing import NDArray
SynthIDDetectionStatus = Literal["detected", "indeterminate", "unsupported"]
DETECTOR_ID = "synthid-periodic-tile-v2"
REGISTERED_DETECTOR_ID = "synthid-periodic-tile-registered-v3"
OPPONENT_REGISTERED_DETECTOR_ID = "synthid-periodic-tile-opponent-registered-v1"
FINE_OPPONENT_REGISTERED_DETECTOR_ID = "synthid-periodic-tile-opponent-fine-registered-v1"
LARGE_DETECTOR_ID = "synthid-periodic-tile-large-v1"
MODEL_FILENAME = "synthid_periodic_tile_2048_v1.npz"
# The template remains frozen at this model geometry. Runtime images are never
# resized. The supported pixel-count interval is the separately challenged domain:
# below it too few repetitions make the positive-only statistic unreliable, and
# above it resource use and specificity have not been calibrated.
MODEL_WIDTH = 2048
MODEL_HEIGHT = 2048
MIN_SUPPORTED_PIXELS = 1_000_000
MAX_SUPPORTED_PIXELS = 18_000_000
TILE_THRESHOLD = 0.17357069773071196
REGISTERED_MIN_SUPPORTED_PIXELS = 250_000
REGISTERED_MAX_SUPPORTED_PIXELS = 10_000_000
# Registered-v3 can confirm a positive only when both disjoint checkerboard
# groups contain a complete frozen 256-pixel patch. Narrower geometries need a
# separately calibrated adaptive-patch expert and must not masquerade as misses.
REGISTERED_MIN_SIDE = 256
# The registered score preserves the minimum normalized v2 margin only after
# independent split-patch phase, amplitude, and held-out codeword confirmation.
REGISTERED_THRESHOLD = 1.0
# The opponent-color fallback is a narrower precision-first route for lossless
# scale changes. Smaller rasters retained a natural period-10 false positive.
OPPONENT_REGISTERED_THRESHOLD = 1.0
OPPONENT_REGISTERED_MIN_PIXELS = 1_000_000
OPPONENT_REGISTERED_MIN_SIDE = 768
# Fine-period registration is separately frozen for the dense 0.47-0.55
# lossless-resize challenge. Its more expensive selector is bounded to the
# geometry range covered by the locked and reserve negative sets.
FINE_OPPONENT_REGISTERED_THRESHOLD = 1.05
FINE_OPPONENT_REGISTERED_MIN_PIXELS = 1_000_000
FINE_OPPONENT_REGISTERED_MAX_PIXELS = 5_000_000
FINE_OPPONENT_REGISTERED_MIN_SIDE = 768
# The large-image score combines all-window fixed and spatial opponent gates
# with an any-window signed opponent mid-band gate. The one vulnerable portrait
# geometry has an additional Green mid-band upper gate.
LARGE_THRESHOLD = 1.0
LARGE_MIN_PIXELS = 10_000_000
LARGE_MAX_PIXELS = 18_000_000
LARGE_WINDOW = 2_048
LARGE_PHASE = 16
LARGE_FIXED_SCORE_MIN = 0.14
LARGE_RED_GREEN_SPATIAL_MIN = 0.90
LARGE_BLUE_YELLOW_SPATIAL_MIN = 0.70
LARGE_BLUE_YELLOW_MID_BAND_MAX = -0.15
LARGE_PORTRAIT_GEOMETRY = (3_072, 5_504)
LARGE_PORTRAIT_GREEN_MID_BAND_MAX = 0.06
INSTALL_HINT = "install the pixel extra: uv add 'remove-ai-watermarks[pixels]'"
@dataclass(frozen=True)
class SynthIDDetection:
"""One local periodic-lattice verdict.
The family this reports is NOT the watermark, and the field names say so. The
statistic is destroyed by a crop of seven pixels, while SynthID's published
evaluation retains 99.97% TPR under aggressive crop and resize, so what
crosses the threshold is a generation-pipeline lattice anchored at the image
origin. It identifies the pipeline, not the mark, and `docs/synthid.md`
carries the measurement.
"""
status: SynthIDDetectionStatus
width: int
height: int
score: float | None
threshold: float
detector: str = DETECTOR_ID
reason: str | None = None
signal_family: str = "generation-pipeline-lattice"
provider_scope: str = "provider-neutral"
backend: str = "local-pixel"
metadata_used_for_verdict: bool = False
pixels_preserved: bool = True
# Consumers cannot be expected to read a caveat in prose, so the two measured
# failure modes travel with every verdict.
tile_aligned_crop_required: bool = True
identifies_watermark: bool = False
@property
def detected(self) -> bool:
"""Whether the supported carrier crossed its frozen threshold."""
return self.status == "detected"
def to_dict(self) -> dict[str, str | int | float | bool | None]:
"""Return a JSON-safe result without a local file path."""
return {
"status": self.status,
"width": self.width,
"height": self.height,
"score": self.score,
"threshold": self.threshold,
"detector": self.detector,
"reason": self.reason,
"signal_family": self.signal_family,
"provider_scope": self.provider_scope,
"backend": self.backend,
"metadata_used_for_verdict": self.metadata_used_for_verdict,
"pixels_preserved": self.pixels_preserved,
"tile_aligned_crop_required": self.tile_aligned_crop_required,
"identifies_watermark": self.identifies_watermark,
}
@dataclass(frozen=True)
class LargeImageComponents:
"""Auditable margins for the calibrated large-image carrier branch."""
width: int
height: int
minimum_fixed_score: float
minimum_red_green_spatial: float
minimum_blue_yellow_spatial: float
minimum_blue_yellow_mid_band: float
maximum_green_mid_band: float
@property
def decision_score(self) -> float:
"""Return the minimum normalized gate margin; one is the boundary."""
margins = [
self.minimum_fixed_score / LARGE_FIXED_SCORE_MIN,
self.minimum_red_green_spatial / LARGE_RED_GREEN_SPATIAL_MIN,
self.minimum_blue_yellow_spatial / LARGE_BLUE_YELLOW_SPATIAL_MIN,
self.minimum_blue_yellow_mid_band / LARGE_BLUE_YELLOW_MID_BAND_MAX,
]
if (self.width, self.height) == LARGE_PORTRAIT_GEOMETRY:
margins.append(1.0 + LARGE_PORTRAIT_GREEN_MID_BAND_MAX - self.maximum_green_mid_band)
return min(margins)
def is_available() -> bool:
"""True when the optional numeric runtime is installed."""
from remove_ai_watermarks.optional_deps import module_available
return module_available("cv2", "numpy")
@lru_cache(maxsize=1)
def _load_template() -> tuple[NDArray[Any], float, int, int, int, int]:
"""Load and validate the bundled pickle-free detector model."""
import numpy as np
model_path = Path(__file__).parent / "assets" / MODEL_FILENAME
with np.load(model_path, allow_pickle=False) as artifact:
if int(artifact["format_version"]) != 1:
raise RuntimeError("unsupported SynthID detector model format")
height = int(artifact["height"])
width = int(artifact["width"])
tile_height = int(artifact["tile_height"])
tile_width = int(artifact["tile_width"])
denoise_sigma = float(artifact["denoise_sigma"])
template = np.asarray(artifact["template"], dtype=np.float64)
if not _geometry_supported(width, height):
raise RuntimeError("bundled SynthID detector has unexpected geometry")
if template.shape != (tile_height, tile_width, 3):
raise RuntimeError("bundled SynthID detector has an invalid template shape")
if not np.all(np.isfinite(template)) or not np.isclose(np.linalg.norm(template), 1.0):
raise RuntimeError("bundled SynthID detector has an invalid template")
if not np.isfinite(denoise_sigma) or denoise_sigma <= 0.0:
raise RuntimeError("bundled SynthID detector has an invalid denoise sigma")
return template, denoise_sigma, height, width, tile_height, tile_width
def fold_residual_template(
pixels: NDArray[Any],
*,
tile_height: int,
tile_width: int,
denoise_sigma: float,
) -> NDArray[Any]:
"""Estimate a zero-mean periodic residual template by modulo folding."""
import cv2
import numpy as np
if pixels.ndim != 3 or pixels.shape[2] != 3:
raise ValueError("pixels must have shape (height, width, 3)")
if tile_height < 1 or tile_width < 1 or denoise_sigma <= 0.0:
raise ValueError("tile dimensions and denoise sigma must be positive")
height, width = pixels.shape[:2]
if height < tile_height or width < tile_width:
raise ValueError("image geometry must be at least as large as the tile geometry")
divisible = height % tile_height == 0 and width % tile_width == 0
full_height = height - height % tile_height
full_width = width - width % tile_width
repeats_y = full_height // tile_height
repeats_x = full_width // tile_width
remaining_height = height - full_height
remaining_width = width - full_width
counts = np.full((tile_height, tile_width), repeats_y * repeats_x, dtype=np.int64)
counts[:remaining_height] += repeats_x
counts[:, :remaining_width] += repeats_y
counts[:remaining_height, :remaining_width] += 1
# OpenCV filters channels independently. Processing one channel at a time
# keeps the 18 MP upper bound from requiring two full three-channel float32
# buffers in addition to the decoded image.
folded = np.empty((tile_height, tile_width, 3), dtype=np.float64)
for channel in range(3):
residual = pixels[:, :, channel].astype(np.float32)
residual -= cv2.GaussianBlur(
residual,
(0, 0),
sigmaX=denoise_sigma,
sigmaY=denoise_sigma,
borderType=cv2.BORDER_REFLECT_101,
)
if divisible:
folded[:, :, channel] = residual.reshape(
repeats_y,
tile_height,
repeats_x,
tile_width,
).mean(axis=(0, 2), dtype=np.float64)
continue
folded_sum = (
residual[:full_height, :full_width]
.reshape(
repeats_y,
tile_height,
repeats_x,
tile_width,
)
.sum(axis=(0, 2), dtype=np.float64)
)
if remaining_height:
bottom = residual[full_height:, :full_width].reshape(
remaining_height,
repeats_x,
tile_width,
)
folded_sum[:remaining_height] += bottom.sum(axis=1, dtype=np.float64)
if remaining_width:
right = residual[:full_height, full_width:].reshape(
repeats_y,
tile_height,
remaining_width,
)
folded_sum[:, :remaining_width] += right.sum(axis=0, dtype=np.float64)
if remaining_height and remaining_width:
folded_sum[:remaining_height, :remaining_width] += residual[
full_height:,
full_width:,
]
folded[:, :, channel] = folded_sum / counts
return folded - np.mean(folded, axis=(0, 1), keepdims=True)
def unit_tile(tile: NDArray[Any]) -> tuple[NDArray[Any], float]:
"""Return TILE normalized by its L2 norm and the original norm."""
import numpy as np
norm = float(np.linalg.norm(tile))
if norm == 0.0:
return np.zeros_like(tile, dtype=np.float64), 0.0
return np.asarray(tile, dtype=np.float64) / norm, norm
def _image_size(image_path: Path) -> tuple[int, int]:
from PIL import Image
with Image.open(image_path) as image:
return image.size
def _geometry_supported(width: int, height: int) -> bool:
"""Whether the image has a calibrated number of periodic-tile samples."""
pixels = width * height
return MIN_SUPPORTED_PIXELS <= pixels <= MAX_SUPPORTED_PIXELS
def _registered_geometry_supported(width: int, height: int) -> bool:
"""Whether scale registration was challenged at this decoded size."""
pixels = width * height
return (
min(width, height) >= REGISTERED_MIN_SIDE
and REGISTERED_MIN_SUPPORTED_PIXELS <= pixels <= REGISTERED_MAX_SUPPORTED_PIXELS
)
def _large_geometry_supported(width: int, height: int) -> bool:
"""Whether fixed phase-aligned windows cover the calibrated large range."""
pixels = width * height
return min(width, height) >= LARGE_WINDOW and LARGE_MIN_PIXELS < pixels <= LARGE_MAX_PIXELS
def _opponent_registered_geometry_supported(width: int, height: int) -> bool:
"""Whether the opponent-color fallback passed its frozen geometry challenge."""
pixels = width * height
return (
min(width, height) >= OPPONENT_REGISTERED_MIN_SIDE
and OPPONENT_REGISTERED_MIN_PIXELS <= pixels <= REGISTERED_MAX_SUPPORTED_PIXELS
)
def _fine_opponent_registered_geometry_supported(width: int, height: int) -> bool:
"""Whether the fine-period selector passed its frozen geometry challenge."""
pixels = width * height
return (
min(width, height) >= FINE_OPPONENT_REGISTERED_MIN_SIDE
and FINE_OPPONENT_REGISTERED_MIN_PIXELS <= pixels <= FINE_OPPONENT_REGISTERED_MAX_PIXELS
)
def folded_template_score(
pixels: NDArray[Any],
template: NDArray[Any],
denoise_sigma: float,
) -> tuple[float, NDArray[Any]]:
"""Fold PIXELS at the model geometry and score the normalized tile."""
tile_height, tile_width = template.shape[:2]
folded = fold_residual_template(
pixels,
tile_height=tile_height,
tile_width=tile_width,
denoise_sigma=denoise_sigma,
)
normalized, _norm = unit_tile(folded)
return float((template * normalized).sum()), folded
def _large_window_starts(length: int) -> tuple[int, ...]:
"""Return phase-aligned starts that cover both edges without resampling."""
if length < LARGE_WINDOW:
raise ValueError("large-image sides must be at least 2,048 pixels")
last = ((length - LARGE_WINDOW) // LARGE_PHASE) * LARGE_PHASE
starts = list(range(0, last + 1, LARGE_WINDOW))
if starts[-1] != last:
starts.append(last)
return tuple(starts)
def _correlation(left: NDArray[Any], right: NDArray[Any]) -> float:
import numpy as np
denominator = float(np.linalg.norm(left) * np.linalg.norm(right))
return float(np.real(np.vdot(right, left)) / denominator) if denominator > 0.0 else 0.0
def _large_window_components(
folded: NDArray[Any],
template: NDArray[Any],
) -> tuple[float, float, float, float]:
"""Measure the four color-phase features used by the large branch."""
import numpy as np
folded_red_green = folded[:, :, 0] - folded[:, :, 1]
template_red_green = template[:, :, 0] - template[:, :, 1]
folded_blue_yellow = folded[:, :, 2] - 0.5 * (folded[:, :, 0] + folded[:, :, 1])
template_blue_yellow = template[:, :, 2] - 0.5 * (template[:, :, 0] + template[:, :, 1])
height, width = folded.shape[:2]
y_coordinates = np.minimum(np.arange(height), height - np.arange(height))
x_coordinates = np.minimum(np.arange(width), width - np.arange(width))
radius = np.sqrt(y_coordinates[:, None] ** 2 + x_coordinates[None, :] ** 2)
mid_band = (radius >= 4.5) & (radius < 6.5)
blue_yellow_mid = _correlation(
np.fft.fft2(folded_blue_yellow)[mid_band],
np.fft.fft2(template_blue_yellow)[mid_band],
)
green_mid = _correlation(
np.fft.fft2(folded[:, :, 1])[mid_band],
np.fft.fft2(template[:, :, 1])[mid_band],
)
return (
_correlation(folded_red_green, template_red_green),
_correlation(folded_blue_yellow, template_blue_yellow),
blue_yellow_mid,
green_mid,
)
def large_image_components(
pixels: NDArray[Any],
template: NDArray[Any],
denoise_sigma: float,
) -> LargeImageComponents:
"""Score all phase-aligned 2,048-pixel windows of one large RGB image."""
if pixels.ndim != 3 or pixels.shape[2] != 3:
raise ValueError("pixels must have shape (height, width, 3)")
height, width = pixels.shape[:2]
if not _large_geometry_supported(width, height):
raise ValueError("image geometry is outside the calibrated large-image range")
minimum_fixed = float("inf")
minimum_red_green = float("inf")
minimum_blue_yellow = float("inf")
minimum_blue_yellow_mid = float("inf")
maximum_green_mid = -float("inf")
for y in _large_window_starts(height):
for x in _large_window_starts(width):
window = pixels[y : y + LARGE_WINDOW, x : x + LARGE_WINDOW]
fixed_score, folded = folded_template_score(window, template, denoise_sigma)
red_green, blue_yellow, blue_yellow_mid, green_mid = _large_window_components(
folded,
template,
)
minimum_fixed = min(minimum_fixed, fixed_score)
minimum_red_green = min(minimum_red_green, red_green)
minimum_blue_yellow = min(minimum_blue_yellow, blue_yellow)
minimum_blue_yellow_mid = min(minimum_blue_yellow_mid, blue_yellow_mid)
maximum_green_mid = max(maximum_green_mid, green_mid)
return LargeImageComponents(
width=width,
height=height,
minimum_fixed_score=minimum_fixed,
minimum_red_green_spatial=minimum_red_green,
minimum_blue_yellow_spatial=minimum_blue_yellow,
minimum_blue_yellow_mid_band=minimum_blue_yellow_mid,
maximum_green_mid_band=maximum_green_mid,
)
def detect_synthid(
image_path: str | Path,
*,
image: NDArray[Any] | None = None,
register_scale: bool | None = None,
) -> SynthIDDetection:
"""Detect the supported periodic carrier in IMAGE_PATH.
``indeterminate`` means that the frozen periodic carrier did not cross its
calibrated threshold; it is not a clean-image guarantee. The default
production router uses scale registration through 10 megapixels and the
native large-image expert above that boundary. Set ``register_scale`` to
``True`` to force registration or ``False`` to run the legacy fixed-period
diagnostic below the large-image boundary.
"""
path = Path(image_path)
if image is None:
width, height = _image_size(path)
else:
if image.ndim != 3 or image.shape[2] != 3:
raise ValueError("image must be a three-channel BGR array")
height, width = image.shape[:2]
large_mode = register_scale is not True and width * height > LARGE_MIN_PIXELS
registered_mode = register_scale is True or (register_scale is None and not large_mode)
if registered_mode:
geometry_supported = _registered_geometry_supported(width, height)
threshold = REGISTERED_THRESHOLD
detector_id = REGISTERED_DETECTOR_ID
unsupported_reason = (
"registered-v3 requires 250,000-10,000,000 decoded pixels and both dimensions to be at least 256 pixels"
)
elif large_mode:
geometry_supported = _large_geometry_supported(width, height)
threshold = LARGE_THRESHOLD
detector_id = LARGE_DETECTOR_ID
unsupported_reason = (
"large-v1 requires more than 10,000,000 through 18,000,000 decoded pixels "
"and at least two phase-aligned 2048-pixel windows"
)
else:
geometry_supported = _geometry_supported(width, height)
threshold = TILE_THRESHOLD
detector_id = DETECTOR_ID
unsupported_reason = "fixed-v2 requires 1,000,000-18,000,000 decoded pixels"
if not geometry_supported:
return SynthIDDetection(
status="unsupported",
width=width,
height=height,
score=None,
threshold=threshold,
detector=detector_id,
reason=unsupported_reason,
)
if not is_available():
raise RuntimeError(f"SynthID pixel detection needs numpy and OpenCV; {INSTALL_HINT}")
import numpy as np
from PIL import Image
template, sigma, *_model = _load_template()
if image is None:
with Image.open(path) as source:
pixels = np.asarray(source.convert("RGB"), dtype=np.uint8)
else:
pixels = np.asarray(image[:, :, ::-1], dtype=np.uint8)
if pixels.shape != (height, width, 3):
raise RuntimeError("decoded image geometry does not match its header")
if registered_mode:
from remove_ai_watermarks._synthid_registered import (
fine_opponent_registered_score,
opponent_registered_score,
registered_score,
)
score = registered_score(pixels, template, sigma)
if score < REGISTERED_THRESHOLD and _opponent_registered_geometry_supported(width, height):
opponent_score = opponent_registered_score(pixels, template, sigma)
if opponent_score >= OPPONENT_REGISTERED_THRESHOLD:
score = opponent_score
threshold = OPPONENT_REGISTERED_THRESHOLD
detector_id = OPPONENT_REGISTERED_DETECTOR_ID
if score < threshold and _fine_opponent_registered_geometry_supported(width, height):
fine_score = fine_opponent_registered_score(pixels, template, sigma)
if fine_score >= FINE_OPPONENT_REGISTERED_THRESHOLD:
score = fine_score
threshold = FINE_OPPONENT_REGISTERED_THRESHOLD
detector_id = FINE_OPPONENT_REGISTERED_DETECTOR_ID
elif large_mode:
score = large_image_components(pixels, template, sigma).decision_score
else:
score, _folded = folded_template_score(pixels, template, sigma)
detected = score >= threshold
return SynthIDDetection(
status="detected" if detected else "indeterminate",
width=width,
height=height,
score=score,
threshold=threshold,
detector=detector_id,
reason=None if detected else "the selected carrier expert did not cross every calibrated gate",
)