mis-custom-nodes / ComfyUI-SpectralSynthID /reverse-SynthID /src /analysis /synthid_codebook_finder.py
Download ComfyUI-SpectralSynthID/reverse-SynthID/src/analysis/synthid_codebook_finder.py from MarkLilly/mis-custom-nodes: direct link, hf CLI and curl.
- Browser
- Download file 28.1 kB
-
https://huggingface.co/MarkLilly/mis-custom-nodes/resolve/main/ComfyUI-SpectralSynthID/reverse-SynthID/src/analysis/synthid_codebook_finder.py
- Command line
-
hf download hf://MarkLilly/mis-custom-nodes/ComfyUI-SpectralSynthID/reverse-SynthID/src/analysis/synthid_codebook_finder.py
-
curl -L -o synthid_codebook_finder.py https://huggingface.co/MarkLilly/mis-custom-nodes/resolve/main/ComfyUI-SpectralSynthID/reverse-SynthID/src/analysis/synthid_codebook_finder.py
28.1 kB
| """ | |
| SynthID Watermark Codebook Discovery | |
| This script performs deep analysis on AI-generated images (with SynthID watermarks) | |
| to reverse-engineer the watermark embedding scheme. | |
| Based on SynthID paper, the watermark is: | |
| 1. Embedded via a neural network encoder during image generation | |
| 2. Designed to be imperceptible but robust to transformations | |
| 3. Uses a learned representation that encodes a binary message | |
| Analysis approaches: | |
| 1. Cross-image bit plane correlation - find consistent bit patterns | |
| 2. Frequency domain analysis - find carrier frequencies | |
| 3. Noise pattern extraction - find embedded signal in noise | |
| 4. DCT/DWT coefficient analysis - find quantization patterns | |
| 5. Statistical anomaly detection - find systematic deviations | |
| """ | |
| import os | |
| import sys | |
| import numpy as np | |
| import cv2 | |
| from PIL import Image | |
| from scipy.fft import fft2, fftshift, ifft2, dct, idct | |
| from scipy.stats import pearsonr, spearmanr, skew, kurtosis | |
| from scipy.ndimage import gaussian_filter, sobel | |
| # from skimage.restoration import denoise_wavelet # Avoid skimage dependency | |
| import pywt | |
| from collections import defaultdict | |
| import json | |
| from concurrent.futures import ProcessPoolExecutor, as_completed | |
| from tqdm import tqdm | |
| class SynthIDCodebookFinder: | |
| """ | |
| Discovers the SynthID watermark codebook by analyzing patterns | |
| across multiple AI-generated images. | |
| """ | |
| def __init__(self, target_size=(512, 512)): | |
| self.target_size = target_size | |
| self.n_images = 0 | |
| # Pattern accumulators | |
| self.bit_planes = {i: [] for i in range(8)} | |
| self.noise_patterns = [] | |
| self.fourier_magnitudes = [] | |
| self.fourier_phases = [] | |
| self.dct_patterns = [] | |
| self.wavelet_patterns = [] | |
| self.lsb_patterns = [] | |
| # Cross-image accumulators for finding consistent patterns | |
| self.lsb_avg = None | |
| self.noise_avg = None | |
| self.fourier_avg = None | |
| self.phase_coherence = None | |
| # Per-image features for clustering | |
| self.image_features = [] | |
| self.image_paths = [] | |
| def load_image(self, path: str) -> np.ndarray: | |
| """Load and preprocess image.""" | |
| img = cv2.imread(path) | |
| if img is None: | |
| return None | |
| img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) | |
| img = cv2.resize(img, self.target_size) | |
| return img | |
| def extract_lsb_pattern(self, img: np.ndarray) -> np.ndarray: | |
| """Extract LSB (Least Significant Bit) pattern from all channels.""" | |
| lsb = np.zeros((self.target_size[1], self.target_size[0], 3), dtype=np.uint8) | |
| for c in range(3): | |
| lsb[:, :, c] = img[:, :, c] & 1 | |
| return lsb | |
| def extract_bit_planes(self, img: np.ndarray) -> dict: | |
| """Extract all 8 bit planes from grayscale image.""" | |
| gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY) | |
| planes = {} | |
| for bit in range(8): | |
| planes[bit] = ((gray >> bit) & 1).astype(np.float32) | |
| return planes | |
| def extract_noise_pattern(self, img: np.ndarray) -> np.ndarray: | |
| """Extract noise residual using wavelet denoising.""" | |
| img_float = img.astype(np.float32) / 255.0 | |
| noise = np.zeros_like(img_float) | |
| for c in range(3): | |
| channel = img_float[:, :, c] | |
| # Use wavelet-based denoising manually | |
| denoised = self.wavelet_denoise(channel) | |
| noise[:, :, c] = channel - denoised | |
| return noise | |
| def wavelet_denoise(self, channel: np.ndarray, wavelet='db4', level=3) -> np.ndarray: | |
| """Manual wavelet denoising using pywt.""" | |
| # Decompose | |
| coeffs = pywt.wavedec2(channel, wavelet, level=level) | |
| # Estimate noise level from finest detail coefficients | |
| detail = coeffs[-1][0] # Horizontal detail at finest level | |
| sigma = np.median(np.abs(detail)) / 0.6745 | |
| # Threshold (soft thresholding with universal threshold) | |
| threshold = sigma * np.sqrt(2 * np.log(channel.size)) | |
| # Apply threshold to detail coefficients | |
| new_coeffs = [coeffs[0]] # Keep approximation | |
| for details in coeffs[1:]: | |
| new_details = tuple( | |
| pywt.threshold(d, threshold, mode='soft') for d in details | |
| ) | |
| new_coeffs.append(new_details) | |
| # Reconstruct | |
| denoised = pywt.waverec2(new_coeffs, wavelet) | |
| # Ensure same size | |
| denoised = denoised[:channel.shape[0], :channel.shape[1]] | |
| return denoised | |
| def extract_fourier_features(self, img: np.ndarray) -> tuple: | |
| """Extract Fourier magnitude and phase.""" | |
| gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY).astype(np.float32) | |
| f = fft2(gray) | |
| fshift = fftshift(f) | |
| magnitude = np.abs(fshift) | |
| phase = np.angle(fshift) | |
| return magnitude, phase | |
| def extract_dct_features(self, img: np.ndarray, block_size=8) -> np.ndarray: | |
| """Extract DCT coefficients pattern.""" | |
| gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY).astype(np.float32) | |
| H, W = gray.shape | |
| # Compute DCT on 8x8 blocks | |
| dct_coeffs = np.zeros((block_size, block_size), dtype=np.float64) | |
| count = 0 | |
| for i in range(0, H - block_size + 1, block_size): | |
| for j in range(0, W - block_size + 1, block_size): | |
| block = gray[i:i+block_size, j:j+block_size] | |
| dct_block = cv2.dct(block) | |
| dct_coeffs += np.abs(dct_block) | |
| count += 1 | |
| if count > 0: | |
| dct_coeffs /= count | |
| return dct_coeffs | |
| def extract_wavelet_features(self, img: np.ndarray) -> dict: | |
| """Extract wavelet coefficient patterns.""" | |
| gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY).astype(np.float32) / 255.0 | |
| # Multi-level wavelet decomposition | |
| coeffs = pywt.wavedec2(gray, 'haar', level=4) | |
| features = { | |
| 'cA': coeffs[0], # Approximation | |
| 'details': [] | |
| } | |
| for level, (cH, cV, cD) in enumerate(coeffs[1:], 1): | |
| features['details'].append({ | |
| 'level': level, | |
| 'horizontal': cH, | |
| 'vertical': cV, | |
| 'diagonal': cD | |
| }) | |
| return features | |
| def analyze_image(self, path: str) -> dict: | |
| """Perform full analysis on a single image.""" | |
| img = self.load_image(path) | |
| if img is None: | |
| return None | |
| features = {} | |
| # LSB analysis | |
| lsb = self.extract_lsb_pattern(img) | |
| features['lsb'] = lsb | |
| features['lsb_density'] = np.mean(lsb) | |
| # Bit planes | |
| bit_planes = self.extract_bit_planes(img) | |
| features['bit_planes'] = bit_planes | |
| features['bit_plane_densities'] = {b: np.mean(p) for b, p in bit_planes.items()} | |
| # Noise | |
| noise = self.extract_noise_pattern(img) | |
| features['noise'] = noise | |
| features['noise_std'] = np.std(noise) | |
| features['noise_structure'] = np.std(noise) / (np.mean(np.abs(noise)) + 1e-10) | |
| # Fourier | |
| magnitude, phase = self.extract_fourier_features(img) | |
| features['fourier_mag'] = magnitude | |
| features['fourier_phase'] = phase | |
| # DCT | |
| dct_coeffs = self.extract_dct_features(img) | |
| features['dct'] = dct_coeffs | |
| # Wavelet | |
| wavelet = self.extract_wavelet_features(img) | |
| features['wavelet'] = wavelet | |
| return features | |
| def add_image(self, path: str) -> bool: | |
| """Add an image to the analysis.""" | |
| features = self.analyze_image(path) | |
| if features is None: | |
| return False | |
| # Accumulate patterns | |
| self.lsb_patterns.append(features['lsb']) | |
| self.noise_patterns.append(features['noise']) | |
| self.fourier_magnitudes.append(features['fourier_mag']) | |
| self.fourier_phases.append(features['fourier_phase']) | |
| self.dct_patterns.append(features['dct']) | |
| for bit, plane in features['bit_planes'].items(): | |
| self.bit_planes[bit].append(plane) | |
| # Update running averages | |
| if self.lsb_avg is None: | |
| self.lsb_avg = features['lsb'].astype(np.float64) | |
| self.noise_avg = features['noise'].astype(np.float64) | |
| self.fourier_avg = features['fourier_mag'].astype(np.float64) | |
| self.phase_coherence = np.exp(1j * features['fourier_phase']) | |
| else: | |
| self.lsb_avg += features['lsb'].astype(np.float64) | |
| self.noise_avg += features['noise'].astype(np.float64) | |
| self.fourier_avg += features['fourier_mag'].astype(np.float64) | |
| self.phase_coherence += np.exp(1j * features['fourier_phase']) | |
| self.image_features.append({ | |
| 'lsb_density': features['lsb_density'], | |
| 'noise_std': features['noise_std'], | |
| 'noise_structure': features['noise_structure'], | |
| 'bit_plane_densities': features['bit_plane_densities'] | |
| }) | |
| self.image_paths.append(path) | |
| self.n_images += 1 | |
| return True | |
| def find_consistent_lsb_pattern(self) -> dict: | |
| """Find LSB bits that are consistent across all images.""" | |
| if self.n_images < 2: | |
| return {} | |
| avg = self.lsb_avg / self.n_images | |
| # Consistency map: pixels where LSB is always 0 or always 1 | |
| # High consistency = close to 0 or 1 | |
| consistency = np.abs(avg - 0.5) * 2 | |
| # Find highly consistent pixels (>90% consistency) | |
| consistent_mask = consistency > 0.9 | |
| # Per-channel analysis | |
| results = { | |
| 'overall_consistency': float(np.mean(consistency)), | |
| 'highly_consistent_ratio': float(np.mean(consistent_mask)), | |
| 'per_channel': {} | |
| } | |
| for c, name in enumerate(['R', 'G', 'B']): | |
| c_consistency = consistency[:, :, c] | |
| c_mask = consistent_mask[:, :, c] | |
| results['per_channel'][name] = { | |
| 'consistency': float(np.mean(c_consistency)), | |
| 'consistent_ratio': float(np.mean(c_mask)), | |
| 'consistent_value_mean': float(np.mean(avg[:, :, c][c_mask])) if np.any(c_mask) else 0.5 | |
| } | |
| # Find spatial pattern in consistent LSBs | |
| results['consistent_lsb_map'] = (avg * consistent_mask.astype(float)) | |
| return results | |
| def find_fourier_carriers(self) -> dict: | |
| """Find consistent frequency components (carrier frequencies).""" | |
| if self.n_images < 2: | |
| return {} | |
| avg_magnitude = self.fourier_avg / self.n_images | |
| phase_coherence = np.abs(self.phase_coherence) / self.n_images | |
| # Normalize magnitude | |
| log_mag = np.log1p(avg_magnitude) | |
| # Find peaks in magnitude spectrum | |
| H, W = avg_magnitude.shape | |
| center_y, center_x = H // 2, W // 2 | |
| # Create frequency coordinate grid | |
| y_coords, x_coords = np.ogrid[:H, :W] | |
| freq_r = np.sqrt((x_coords - center_x)**2 + (y_coords - center_y)**2) | |
| # Radial average | |
| max_r = int(np.min([center_x, center_y])) | |
| radial_profile = np.zeros(max_r) | |
| radial_phase = np.zeros(max_r) | |
| for r in range(max_r): | |
| mask = (freq_r >= r) & (freq_r < r + 1) | |
| if np.any(mask): | |
| radial_profile[r] = np.mean(log_mag[mask]) | |
| radial_phase[r] = np.mean(phase_coherence[mask]) | |
| # Find anomalous frequencies (high magnitude AND high phase coherence) | |
| combined_score = radial_profile * radial_phase | |
| mean_score = np.mean(combined_score) | |
| std_score = np.std(combined_score) | |
| anomalous_freqs = np.where(combined_score > mean_score + 2 * std_score)[0] | |
| # Analyze vertical center frequency (known pattern from previous analysis) | |
| vertical_strip = log_mag[:, center_x-2:center_x+3] | |
| vertical_energy = np.sum(vertical_strip, axis=1) | |
| # Find peaks in vertical profile | |
| from scipy.signal import find_peaks | |
| peaks, properties = find_peaks(vertical_energy, height=np.mean(vertical_energy) + np.std(vertical_energy)) | |
| results = { | |
| 'radial_profile': radial_profile.tolist(), | |
| 'radial_phase_coherence': radial_phase.tolist(), | |
| 'anomalous_frequencies': anomalous_freqs.tolist(), | |
| 'vertical_peaks_y': peaks.tolist(), | |
| 'phase_coherence_overall': float(np.mean(phase_coherence)), | |
| 'avg_magnitude_map': avg_magnitude, | |
| 'phase_coherence_map': phase_coherence | |
| } | |
| return results | |
| def find_noise_watermark(self) -> dict: | |
| """Find consistent noise pattern (embedded signal).""" | |
| if self.n_images < 2: | |
| return {} | |
| avg_noise = self.noise_avg / self.n_images | |
| # Analyze per-channel | |
| results = {'per_channel': {}} | |
| for c, name in enumerate(['R', 'G', 'B']): | |
| noise_c = avg_noise[:, :, c] | |
| # Structure ratio | |
| structure_ratio = np.std(noise_c) / (np.mean(np.abs(noise_c)) + 1e-10) | |
| # Spatial autocorrelation (watermarks often have structure) | |
| from scipy.ndimage import correlate | |
| autocorr = correlate(noise_c, noise_c[:50, :50]) | |
| results['per_channel'][name] = { | |
| 'mean': float(np.mean(noise_c)), | |
| 'std': float(np.std(noise_c)), | |
| 'structure_ratio': float(structure_ratio), | |
| 'max_val': float(np.max(noise_c)), | |
| 'min_val': float(np.min(noise_c)) | |
| } | |
| results['noise_pattern'] = avg_noise | |
| results['overall_structure_ratio'] = float(np.std(avg_noise) / (np.mean(np.abs(avg_noise)) + 1e-10)) | |
| return results | |
| def find_bit_plane_watermark(self) -> dict: | |
| """Analyze bit planes for consistent patterns.""" | |
| if self.n_images < 2: | |
| return {} | |
| results = {'per_bit': {}} | |
| for bit in range(8): | |
| planes = np.array(self.bit_planes[bit]) | |
| avg_plane = np.mean(planes, axis=0) | |
| # Consistency: how often is this bit the same value across images | |
| consistency = np.abs(avg_plane - 0.5) * 2 | |
| # Entropy of bit plane (low entropy = potential watermark) | |
| from scipy.stats import entropy as sp_entropy | |
| hist, _ = np.histogram(avg_plane.ravel(), bins=50) | |
| bit_entropy = sp_entropy(hist + 1e-10) | |
| results['per_bit'][bit] = { | |
| 'mean': float(np.mean(avg_plane)), | |
| 'consistency': float(np.mean(consistency)), | |
| 'entropy': float(bit_entropy), | |
| 'highly_consistent_ratio': float(np.mean(consistency > 0.8)) | |
| } | |
| if bit == 0: # LSB is most interesting | |
| results['lsb_avg_pattern'] = avg_plane | |
| results['lsb_consistency_map'] = consistency | |
| return results | |
| def find_dct_watermark(self) -> dict: | |
| """Analyze DCT coefficients for embedding patterns.""" | |
| if self.n_images < 2: | |
| return {} | |
| dct_array = np.array(self.dct_patterns) | |
| avg_dct = np.mean(dct_array, axis=0) | |
| std_dct = np.std(dct_array, axis=0) | |
| # Coefficient of variation (CV) - low CV means consistent | |
| cv = std_dct / (avg_dct + 1e-10) | |
| # Find most consistent coefficients | |
| consistent_positions = np.where(cv < 0.1) | |
| results = { | |
| 'avg_dct': avg_dct.tolist(), | |
| 'cv_dct': cv.tolist(), | |
| 'consistent_positions': list(zip(consistent_positions[0].tolist(), | |
| consistent_positions[1].tolist())), | |
| 'num_consistent': len(consistent_positions[0]) | |
| } | |
| return results | |
| def analyze_cross_image_correlation(self, sample_size=50) -> dict: | |
| """Analyze correlation between images to find shared patterns.""" | |
| if self.n_images < 2: | |
| return {} | |
| sample_size = min(sample_size, len(self.noise_patterns)) | |
| indices = np.random.choice(len(self.noise_patterns), sample_size, replace=False) | |
| # Noise correlation matrix | |
| noise_corrs = [] | |
| for i in range(len(indices)): | |
| for j in range(i + 1, len(indices)): | |
| n1 = self.noise_patterns[indices[i]].ravel() | |
| n2 = self.noise_patterns[indices[j]].ravel() | |
| corr, _ = pearsonr(n1, n2) | |
| noise_corrs.append(corr) | |
| # LSB correlation | |
| lsb_corrs = [] | |
| for i in range(len(indices)): | |
| for j in range(i + 1, len(indices)): | |
| l1 = self.lsb_patterns[indices[i]].ravel().astype(float) | |
| l2 = self.lsb_patterns[indices[j]].ravel().astype(float) | |
| corr, _ = pearsonr(l1, l2) | |
| lsb_corrs.append(corr) | |
| results = { | |
| 'noise_correlation': { | |
| 'mean': float(np.mean(noise_corrs)), | |
| 'std': float(np.std(noise_corrs)), | |
| 'max': float(np.max(noise_corrs)), | |
| 'min': float(np.min(noise_corrs)) | |
| }, | |
| 'lsb_correlation': { | |
| 'mean': float(np.mean(lsb_corrs)), | |
| 'std': float(np.std(lsb_corrs)), | |
| 'max': float(np.max(lsb_corrs)), | |
| 'min': float(np.min(lsb_corrs)) | |
| } | |
| } | |
| return results | |
| def extract_codebook(self) -> dict: | |
| """ | |
| Main method: Extract the SynthID codebook from analyzed images. | |
| """ | |
| print(f"Analyzing {self.n_images} images...") | |
| codebook = { | |
| 'n_images_analyzed': self.n_images, | |
| 'image_size': self.target_size, | |
| 'patterns': {} | |
| } | |
| print(" Finding consistent LSB patterns...") | |
| codebook['patterns']['lsb'] = self.find_consistent_lsb_pattern() | |
| print(" Finding Fourier carrier frequencies...") | |
| codebook['patterns']['fourier'] = self.find_fourier_carriers() | |
| print(" Finding noise watermark...") | |
| codebook['patterns']['noise'] = self.find_noise_watermark() | |
| print(" Finding bit plane patterns...") | |
| codebook['patterns']['bit_planes'] = self.find_bit_plane_watermark() | |
| print(" Finding DCT patterns...") | |
| codebook['patterns']['dct'] = self.find_dct_watermark() | |
| print(" Analyzing cross-image correlation...") | |
| codebook['patterns']['cross_correlation'] = self.analyze_cross_image_correlation() | |
| return codebook | |
| def save_visualization(codebook: dict, output_dir: str): | |
| """Save visualizations of discovered patterns.""" | |
| import matplotlib.pyplot as plt | |
| os.makedirs(output_dir, exist_ok=True) | |
| # 1. LSB consistency map | |
| if 'lsb' in codebook['patterns'] and 'consistent_lsb_map' in codebook['patterns']['lsb']: | |
| lsb_map = codebook['patterns']['lsb']['consistent_lsb_map'] | |
| fig, axes = plt.subplots(1, 3, figsize=(15, 5)) | |
| for c, name in enumerate(['R', 'G', 'B']): | |
| axes[c].imshow(lsb_map[:, :, c], cmap='hot') | |
| axes[c].set_title(f'Consistent LSB - {name}') | |
| axes[c].axis('off') | |
| plt.savefig(os.path.join(output_dir, 'lsb_consistency_map.png'), dpi=150, bbox_inches='tight') | |
| plt.close() | |
| # 2. Fourier magnitude | |
| if 'fourier' in codebook['patterns'] and 'avg_magnitude_map' in codebook['patterns']['fourier']: | |
| mag = codebook['patterns']['fourier']['avg_magnitude_map'] | |
| plt.figure(figsize=(10, 10)) | |
| plt.imshow(np.log1p(mag), cmap='viridis') | |
| plt.colorbar(label='Log Magnitude') | |
| plt.title('Average Fourier Magnitude Spectrum') | |
| plt.savefig(os.path.join(output_dir, 'fourier_magnitude.png'), dpi=150, bbox_inches='tight') | |
| plt.close() | |
| # Phase coherence | |
| if 'phase_coherence_map' in codebook['patterns']['fourier']: | |
| phase = codebook['patterns']['fourier']['phase_coherence_map'] | |
| plt.figure(figsize=(10, 10)) | |
| plt.imshow(phase, cmap='hot') | |
| plt.colorbar(label='Phase Coherence') | |
| plt.title('Phase Coherence (High = Consistent Phase)') | |
| plt.savefig(os.path.join(output_dir, 'phase_coherence.png'), dpi=150, bbox_inches='tight') | |
| plt.close() | |
| # 3. Noise pattern | |
| if 'noise' in codebook['patterns'] and 'noise_pattern' in codebook['patterns']['noise']: | |
| noise = codebook['patterns']['noise']['noise_pattern'] | |
| fig, axes = plt.subplots(1, 3, figsize=(15, 5)) | |
| for c, name in enumerate(['R', 'G', 'B']): | |
| im = axes[c].imshow(noise[:, :, c], cmap='RdBu', vmin=-0.01, vmax=0.01) | |
| axes[c].set_title(f'Avg Noise Pattern - {name}') | |
| axes[c].axis('off') | |
| plt.colorbar(im, ax=axes, label='Noise Value') | |
| plt.savefig(os.path.join(output_dir, 'noise_pattern.png'), dpi=150, bbox_inches='tight') | |
| plt.close() | |
| # 4. Bit plane consistency | |
| if 'bit_planes' in codebook['patterns'] and 'lsb_consistency_map' in codebook['patterns']['bit_planes']: | |
| lsb_cons = codebook['patterns']['bit_planes']['lsb_consistency_map'] | |
| plt.figure(figsize=(10, 10)) | |
| plt.imshow(lsb_cons, cmap='hot') | |
| plt.colorbar(label='Consistency') | |
| plt.title('LSB Consistency Map') | |
| plt.savefig(os.path.join(output_dir, 'lsb_bit_plane_consistency.png'), dpi=150, bbox_inches='tight') | |
| plt.close() | |
| # 5. Radial profile plot | |
| if 'fourier' in codebook['patterns'] and 'radial_profile' in codebook['patterns']['fourier']: | |
| radial = codebook['patterns']['fourier']['radial_profile'] | |
| phase_radial = codebook['patterns']['fourier']['radial_phase_coherence'] | |
| fig, ax1 = plt.subplots(figsize=(12, 6)) | |
| ax2 = ax1.twinx() | |
| ax1.plot(radial, 'b-', label='Magnitude') | |
| ax2.plot(phase_radial, 'r-', label='Phase Coherence') | |
| ax1.set_xlabel('Frequency (radius)') | |
| ax1.set_ylabel('Log Magnitude', color='b') | |
| ax2.set_ylabel('Phase Coherence', color='r') | |
| # Mark anomalous frequencies | |
| anomalous = codebook['patterns']['fourier']['anomalous_frequencies'] | |
| for f in anomalous: | |
| ax1.axvline(x=f, color='g', linestyle='--', alpha=0.5) | |
| plt.title('Radial Frequency Profile') | |
| plt.savefig(os.path.join(output_dir, 'radial_profile.png'), dpi=150, bbox_inches='tight') | |
| plt.close() | |
| print(f"Visualizations saved to {output_dir}") | |
| def main(): | |
| import argparse | |
| parser = argparse.ArgumentParser(description='Find SynthID watermark codebook') | |
| parser.add_argument('image_dir', type=str, help='Directory containing AI images') | |
| parser.add_argument('--output', type=str, default='./codebook_results', help='Output directory') | |
| parser.add_argument('--max-images', type=int, default=250, help='Maximum images to analyze') | |
| parser.add_argument('--size', type=int, default=512, help='Target image size') | |
| args = parser.parse_args() | |
| finder = SynthIDCodebookFinder(target_size=(args.size, args.size)) | |
| # Get image paths | |
| image_extensions = {'.png', '.jpg', '.jpeg', '.webp', '.bmp'} | |
| image_paths = [] | |
| for fname in os.listdir(args.image_dir): | |
| ext = os.path.splitext(fname)[1].lower() | |
| if ext in image_extensions: | |
| image_paths.append(os.path.join(args.image_dir, fname)) | |
| image_paths = image_paths[:args.max_images] | |
| print(f"Found {len(image_paths)} images to analyze") | |
| # Process images | |
| for path in tqdm(image_paths, desc="Processing images"): | |
| finder.add_image(path) | |
| # Extract codebook | |
| codebook = finder.extract_codebook() | |
| # Save results | |
| os.makedirs(args.output, exist_ok=True) | |
| # Save codebook (without numpy arrays for JSON) | |
| codebook_json = { | |
| 'n_images_analyzed': codebook['n_images_analyzed'], | |
| 'image_size': codebook['image_size'], | |
| 'patterns': {} | |
| } | |
| for pattern_type, pattern_data in codebook['patterns'].items(): | |
| codebook_json['patterns'][pattern_type] = {} | |
| for key, value in pattern_data.items(): | |
| if isinstance(value, np.ndarray): | |
| continue # Skip numpy arrays | |
| elif isinstance(value, dict): | |
| codebook_json['patterns'][pattern_type][key] = value | |
| else: | |
| codebook_json['patterns'][pattern_type][key] = value | |
| with open(os.path.join(args.output, 'codebook.json'), 'w') as f: | |
| json.dump(codebook_json, f, indent=2) | |
| # Save visualizations | |
| save_visualization(codebook, args.output) | |
| # Print summary | |
| print("\n" + "="*60) | |
| print("SYNTHID CODEBOOK ANALYSIS RESULTS") | |
| print("="*60) | |
| print(f"\nImages analyzed: {codebook['n_images_analyzed']}") | |
| if 'lsb' in codebook['patterns']: | |
| lsb = codebook['patterns']['lsb'] | |
| print(f"\nLSB Pattern:") | |
| print(f" Overall consistency: {lsb.get('overall_consistency', 0):.4f}") | |
| print(f" Highly consistent ratio: {lsb.get('highly_consistent_ratio', 0):.4f}") | |
| if 'fourier' in codebook['patterns']: | |
| fourier = codebook['patterns']['fourier'] | |
| print(f"\nFourier Analysis:") | |
| print(f" Phase coherence: {fourier.get('phase_coherence_overall', 0):.4f}") | |
| print(f" Anomalous frequencies: {fourier.get('anomalous_frequencies', [])}") | |
| print(f" Vertical peaks at y: {fourier.get('vertical_peaks_y', [])}") | |
| if 'noise' in codebook['patterns']: | |
| noise = codebook['patterns']['noise'] | |
| print(f"\nNoise Pattern:") | |
| print(f" Overall structure ratio: {noise.get('overall_structure_ratio', 0):.4f}") | |
| if 'bit_planes' in codebook['patterns']: | |
| bp = codebook['patterns']['bit_planes'] | |
| print(f"\nBit Plane Analysis:") | |
| for bit in range(8): | |
| if bit in bp.get('per_bit', {}): | |
| info = bp['per_bit'][bit] | |
| print(f" Bit {bit}: consistency={info['consistency']:.4f}, entropy={info['entropy']:.4f}") | |
| if 'cross_correlation' in codebook['patterns']: | |
| cc = codebook['patterns']['cross_correlation'] | |
| print(f"\nCross-Image Correlation:") | |
| print(f" Noise correlation mean: {cc.get('noise_correlation', {}).get('mean', 0):.4f}") | |
| print(f" LSB correlation mean: {cc.get('lsb_correlation', {}).get('mean', 0):.4f}") | |
| print(f"\nResults saved to: {args.output}") | |
| if __name__ == '__main__': | |
| main() | |