From bc424a48f48952f95b90862b88182d956ffc0b48 Mon Sep 17 00:00:00 2001 From: Victor Kuznetsov Date: Sun, 23 Aug 2026 11:14:53 -0700 Subject: [PATCH] Remove lattice detector sources from the public package --- .../_synthid_confirmation.py | 288 -------- .../_synthid_registered.py | 664 ------------------ .../assets/synthid_periodic_tile_2048_v1.npz | Bin 7501 -> 0 bytes src/remove_ai_watermarks/synthid_detector.py | 543 -------------- 4 files changed, 1495 deletions(-) delete mode 100644 src/remove_ai_watermarks/_synthid_confirmation.py delete mode 100644 src/remove_ai_watermarks/_synthid_registered.py delete mode 100644 src/remove_ai_watermarks/assets/synthid_periodic_tile_2048_v1.npz delete mode 100644 src/remove_ai_watermarks/synthid_detector.py diff --git a/src/remove_ai_watermarks/_synthid_confirmation.py b/src/remove_ai_watermarks/_synthid_confirmation.py deleted file mode 100644 index 0ed6cf9..0000000 --- a/src/remove_ai_watermarks/_synthid_confirmation.py +++ /dev/null @@ -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, - ) diff --git a/src/remove_ai_watermarks/_synthid_registered.py b/src/remove_ai_watermarks/_synthid_registered.py deleted file mode 100644 index 8b6e4e5..0000000 --- a/src/remove_ai_watermarks/_synthid_registered.py +++ /dev/null @@ -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 diff --git a/src/remove_ai_watermarks/assets/synthid_periodic_tile_2048_v1.npz b/src/remove_ai_watermarks/assets/synthid_periodic_tile_2048_v1.npz deleted file mode 100644 index 33bbf4cb6817e65edc72c919065b278abdaca6b7..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 7501 zcmchcWl&t(nudWO!JP#6;2r`&1C55@78+^XT^nf#!Gc3@4epv?jZ4tr5ZtYC4Q>N< zPiF2pXXe)Ym^*9N-fQow_3rnrZ+*Ykqa^nPkqr(G?&-r#4L9sK6Mg^3z=I=zGqrQH zF>nUCnm9UH*x9n%+PlNU;lTMntbH6(|Cr!XcAssN;`H_?bt#pme+H)NVViTWz0mXhs!*i4CXHq~4d1gjOl!I_dk}QB$jYa^^E`^MY45##`B$yH| z48N10JtQ$Vu`n}t{2DP#+SwSr|K;|9=Kx7WZ-gT>|Wf0L~WHCZKvxQQZ!!Lx6x26Lw7-ji`;fxe#MaKj$pqwNVP<3SdsjaF(U8*rx-m^N|F?~a z%z2@Y{g=*^4e}5Eodn~t9ws*S)&|Zdze9X@{ttJvXaAVf;?G41BhI@|^CW@H*)oL^ z^t3!=Cg@CRGQ>W_J_{;5@-U#ptY z_)aI;OI_(SRYh4}-7hWErY5+!_o46gw@v(tu84GBL(JZHOA9;OP@$sB$q8o{+kWWI z)s)DEs@;GyzS?v&8(&l~LRyLZ9Et<4nzx4DDSd_JJ!f}Bx14yDVAI1G0DS?dMK$H$ zHY*h;So_UPTV(C+8RKoD~jq}Tn-7hxbCW&Uqc^s83WY+Q)^ z3C}IW?>bt0-41jHE}k-uW(qSM4EfjCSv3nJceY=wM_)!$CJocYWVDD2qJmu4i7lF6 zh%NXyMZ{^7aZadi$W@KXGq&&u!mPE?eBOSihi$jFRxWVeuX)iq8soXF4J3hw^5ox` zW4BJ$4x!}vNQsF-;WCeW@%vIeRgO=bn7Pz9TY zV^!BYOWmj`tmivd!r9t4JMty>CtvB0lM=3r0$IbXDX?y{fQZ$WI!v@A#5fHy4#slv zZ?YP;E_{ zI~-h*2U;%J{ER>PO-G_{B4_o$qpUZ2)`8Qwqk4EKZ)xDxP?S)!;KI^&G~#O}k!*dp z)CL061vejCHw+HdU&FP?9TB`VU%fGe(rhf=YwaD6UP@p|)_;~8pbpqv>s=22$$B9& zq3KaXMUtl@tnC*`(4PHsTX02HnDem0<5bCuTrch$q4s-d_&u*(9J^p|SBY|Yw2eQ; zMUBUp^uP+E43Uri!fq#Gwtd0}MKfCE>4x*qMk!C;y}EYC48V;R-Wb=%OZd1}LHi`O zS8cPWQL$k=J>b=DPweT(pC8-+AG;zy8)z}8E?IkHTE}Su*&5gUppJrR;gZM0rb>rf z77nVLtQHvSEp2d^U#>N+s#vJbk}SKbSco+m(Df?nO)xdWL!q-R+;1*-m^sLo33BEt z(`IU$a2s^O_AX>TEwf1G7k5}5%6h1M(Xl1^HPDi9w?_E{6Q-AZNysh6&#vnzDO&5P|Gnp+M!21LQ3ENv`HOYtEgVJnJ-t4x?#P1r>Q-s_s+Y90Rc@ z>%?(<`=c&YWk24YYAbdDG-H8TtRajYS)cYNp{ZE5s)(jF4b!w^l+{+-a$HfFa z5mgsEcAJLYqn+q0D}5-QrXmrI;6wE&DgSPe4(sbM<_;m#;Q#zgEO4V$!^0cOSHF%{ zViEHR$uC*OCHuT{Q~gXO)QpV~NysxLqY05CG*T;lVz6zxvy9YpKhSTgHM!^xQ|+cf z>9Vk&o9rbmsMssCzw4%#hn79^^YSMTtKsW-XD>GPg_R`*X$bghDIoUK&X6`2M&eeY z3Szl1w-Gh(slrXl0AdNpg1w<7EulnY5p(dwPv6G-2vr+CEhQ1>kx?Zs)b>|H;ZbfQ zT#8=%n(n`l#>vWs!7?0#{gdqb6|6Gtn@(5tk%(vTFn0H4yN!s77Y(PaJ7;8dX$d+) zv&M*&V5~Vv&#G_t4+hrp!BJmn*{voBfZ#4s*NNkOE2+joklhGuE}dbN=bFPo7J(y)hV|CDi^c)4ZTW& zDIS=6uPZImgac^$;6#8(hy zzE0s(VZYN9nW;;sFXzE^cyrw(*UcFOk-dF;6rb-f7>lC}8p@1)*HR}FXfTPMY9{an zRV39hL>4v-QwC5b5PGC;oJ7Upn%6_?pJUqy)@#%B4(~k=B3xYvF9Dl z%NI6SK5q$~DcO)|91x?l9?KT*_M;`V26&p=Cwwz!lSeL;CWn;|=zf_0NO-A!6FuBE zF7GP56FgHxGS@=h+cS5PB|X;sHmw4*9(RvGX!X6(z)1u%@<;DH>oRzu`$L(XXnNBUbZi? zE?a50*85)SBQ4Ck!CQc-*9pYNi=LicrS<`@CcDGppeUz?MPlk+7BQ@$7|6ocU^&Z} zHrNcPccF}?=hJpfc<-_Fx3+b~K|j5U=G+@*6A1H0n(p;vZg7PE3B++nxWnOXJ^i(t zGuMP({Ro3kS~zblJ7Hwb%VqG-pP<>%AHy0ddv6>#$i4d&k8evtkvvlRS72W-sOnIA zYdHK@DJ?|O$YRVZu&9n>_xsx2lqH{`VdbnC)Vi3=o<%Jc-FDMXc)x7qp-|B?>JXf0-sqkoqh2WsWW}mGk(Q>-{KFW_!C*=6)yS zlVZY;(cG1F3PD3s3Y>VLSE9)nqjG)$n|DR>FDx#Syuw%BmIp2H9f2Zh2--VaRMCUL zcJOl3nSHz*2(;^w3iqeXt;tKT3;SK4S@Yl0`P#+8Cz;GNjrvR<27EialFHXvN}s!p zs23nVRdi|I@pQ7*`3S77uUu9^E zXpW(V1<5cl((LM(5&C`z*bUZxE)avprF8 z^TG5WU0dKHu@rB>AM2Vlep?~cqeR0XjH#ejzmb-9^kJluL#&v6r1hIlV)ko6q)YWW z6t+~n<+9m&V4)uA4V$1u2t}@#MtgIG@tw!hc@cRv686BPV?m-xQQX!neIb3zc;DMF zh9kjHz_8VQ>0&OK7H6N82D=9r9JUS%tNa`>vR+cgu|k1{B+TN`T9QjHdXGGRs_TGHE!Ewbk^m%OZ>dLZW&^Ml)s>i*SKEqmF5uRZ$UVK zcHnt;*>}L2pHGNtmf!!n4nzbXI8ezBPvY25XlB|u5B3l7_9<+`-mA10;%QucjhTIe zoDSK;Xj^pj%r6hiU!6_!5vfXN6pQ|17F+6GrmyIHRJ9I=`J_n3#w~QuSJkd*IDDT5 zW4?fl((4uO6NXtde)?nI#o;QgVJ&i<6G+VKF}ye4=OU%u5}pK$JF1SSlFZg>yziHm z+yJu~8M6USg{{-b<8?_Y!wgezD$Xk;{9=+R=}Gv^TEfGkNBezDN%* zVfnbvg5(92zqF74>8dPKL||IhrhT+PqoeETMPHEcsf=b#o#CxTU3RTrqKek1a<>yA z96B57wc?!38d>Iron?9c&!h-|1D2>eppL`T@K+J# zY3X;90HGuO@~Q}aI8>)f~$IfD^w9I_zZs`qNa&~XKzH| zN=L)lmNjH6PFK&UQ#>Z??s;{orhUmckPXU7`7%KQ0rW*KK*~ZSD1d@Q2iqOSx`yZH zr`Cyb_MLM|$7FqGIn_@a;O{PQF%~UTr{J+B;lJ7mIXK@QsZ1BF@StgWy3omWS7s&u zgspk)S7)B(<_b(?^vrMb~7j*P+qjy3xxiL(b+DZUu3VkB;x}oArgd_@oQV z^7Owz&Xz}3ol{FR3q%`s{iCHB2*}Dse+!&4oN%_k_nFrJJbp`+%Zo&m zBN8zm!-z0)AL4Z?yu#IbwG~`mOqNkK?Xo>3hqrL9$ss>vLKRWXb4zihf8JiWQ48z@ z@Aa*B7k02#eGva93lwf9gBjiwnPgpa&`R<~dc7wdriJ=#VefX7X{_xc^t?TrwWy9k zayxN3gJ2f;TU#(g>w^%G1FH)iEgTNFt8|A0)(w|1XVU8q2T0j^OBS^mia2lj@p-d% zr3f}>Zxm=z)?{(jO!$zpTzMocEYJzhj0tDTH6tJhYlx&@@KH8nl zu76p52bNBnymk~soj)6F9FH=WXlqN0)x-RL6g1yS+N_p+30Z+;x=b%=aqBN}6JWan zx6yD{0yq>tgvi4tPF_n8P8a+cgve5_r&FBg|Z!4hl%BtQysdVTNj@5(L-w zB6bDWTuKacZ;PYG{oad}vy^h>xI2o0=-9LS_zlzawa1DfLVOPswG0?YlHS8qn((wj*rG>|R_EIUGEEGO)R=Odd zly*2)Ba?MM9i_P|M?X6l_-TxIVP&ubF-2<+L-qtb)kkr_-VGrB)AxFy&dW&tSB^lHFZ&6bU*EyqT;=|C;d*s!!@<97e80*g=17bkuhZ=ltsgq#N6heJ6LMtXx6U3cTx(nosrUq zbp{yR$MF~JdkIOzze0-i>?KSGS|9q3d*Ib!)Nl%YiUv!?Y+4#_V){e?1<=u6 z8@dLC`JS3{_d83;-Pt@9|7@@md=k#L#9i5Wor#W~{L{NP`1xi==m#B{;3^i1Z5nXl zyU_}0G7p^B$04jTNz2CD|l%pCIM z(>$OsyY3$@SmuN))51dm2M*V*)OVz%(j3YTvmwKa%Hl-dIob2d2{SU_+d}B4i7|SC zG&8oAlJ?c^%eKBIG-SYo2tm+>SlN-}`g0r4x#0%aS-&$nevKw~eyScylp&o@7yHY; zr0!wRlq$xIYP%Et;kfr|k>2rTU+M8RbsaFFmrw$?ArNG11~0VNz*dhV46W_vYWTa( zI|^3W^f<02C4@a1WH~;Bz9U=kDy)y;wancSc&<1iK$h9-dyv=qdo#)XMTxTxnCgP> zt|3vEo|E1TZ7h5`B?$30Ib+Fe!<)843EQ5X3tTQFgHl#IAJ11F_+c(%=CtWE-E)q6 z#oNl)xIcV|`B^=2D)yH~3Pvuh8*v&(WG-0sa%V9k z3%qh70hF`fuToReOyaNw@1iuxcC!Hzb1xS)e27PDwYSbdAYW_7*7 z@&F`waiJqN?f!;8vKv#1LdH(ZWT=%%Y93w^PthhU|!h5(Ba zcwO`g8zhQpapqkby|u&j7zn=&pUBfOBSD z*;MD_D$+_R77Dul_Y>J7DLCA8y3J=@;{<@kM3NxbvJxvpvu!EI@kg8hyj9e0H8NUG zWMM;Y<&chLJYXDMK)$a0U4am^d$Ep=NGjEK zOOZCFmj6XMH=Uy4G_ub=s1Q7jl8$VQ_7lBnJU?r!!#=bA)LEv%O-kwf9dc^@S}M;^ zHMDz_KjWhjh1r7NhxCuQzyxe>V&rUM46=P7TE9=~f8zqV|0`Tj`J?gr)h{^#xCdka zkHGfi&tUBlmp!60yg%$7rw{Ot&%a}~zv>@x*l#`BqvVhA0Y2p6?Z2SVztW-p7Cp=# z;4vPY|C8eURrbhye%oRHNf!Pf`wNKrS7NL`i3t$?5dV+HkC^9?-aKv%?w_>D58A&$ zpTAl^vYAKj^2lH?{v@Y+@?iZ>q~@>6$A)?AgTEX1lED4FeU#*okRSj5^x;~1D2XQN H 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", - )