From 5e5a3976ba844c5398da457eeafb3da4c051945e Mon Sep 17 00:00:00 2001 From: Victor Kuznetsov Date: Mon, 10 Aug 2026 11:07:32 -0700 Subject: [PATCH] Add registered SynthID phase probing --- docs/synthid-detector-removal-plan.md | 55 +++++ docs/synthid.md | 16 +- scripts/synthid_color_space_probe.py | 25 ++- scripts/synthid_phase_carrier.py | 81 ++++++- scripts/synthid_phase_registration.py | 92 ++++++++ scripts/synthid_v3_codebook_probe.py | 278 +++++++++++++++++++----- tests/test_synthid_phase_carrier.py | 27 +++ tests/test_synthid_v3_codebook_probe.py | 57 +++++ 8 files changed, 556 insertions(+), 75 deletions(-) create mode 100644 scripts/synthid_phase_registration.py diff --git a/docs/synthid-detector-removal-plan.md b/docs/synthid-detector-removal-plan.md index c81359e..abbda33 100644 --- a/docs/synthid-detector-removal-plan.md +++ b/docs/synthid-detector-removal-plan.md @@ -1024,6 +1024,61 @@ labels, matched transformations, and an image-level detection loss. External generator corpora, including difficult non-target providers, remain hard negative and FPR-challenge sets only. +### 2026-08-10: low-content controls and registered phase carrier + +A same-resolution low-content matrix compared independently generated solid +outputs from two target model families against three per-image controls: exact +mean fill, amplitude-matched Gaussian noise, and a phase-randomized residual +with preserved Fourier magnitude. Raw stationary-wavelet summaries transferred +between the two target families with AUCs of 0.982 and 1.000, and reached 0.973 +when blue and green were held out by color. This was not watermark evidence. +The frozen classifier accepted every one of 1,869 external negatives because +it had learned the distinction between real generator texture and artificial +controls. Removing absolute wavelet energy reduced external-negative +acceptance only to 61.6%, with similar 58.1-67.3% acceptance across all three +source classes. Both low-content wavelet branches are rejected as presence +detectors until real non-target solid outputs provide matched negatives. + +The numeric V3 audit loader was then extended to support both dense and sparse +format-v2 profiles without pickle. Exact-profile evaluation exposed a sharp +encoder-version boundary. The 1024x1024 profile accepted none of 231 target +provider images and none of 26 exact-geometry negatives. The 1536x2816 profile +accepted 30 of 55 target-provider images, including all four temporal-test +images, while rejecting the one exact-geometry negative available in the +closed corpus. The independently fitted phase model accepted 24 of those 55 +and also accepted all four temporal-test images. This is positive evidence for +a geometry- and epoch-specific carrier, not a universal SynthID decoder. + +On the four temporal-test positives, the fixed V3 score survived JPEG-95 and a +75% downscale on all four images, survived JPEG-85 on two, and failed after a +5% center crop or a one-pixel translation on all four. Bounded analytical +translation registration recovered all four shifted images and selected the +known `(-1, -1)` offset. Searching up to 16 pixels produced no positives among +50 exact-resolution and 144 canonicalized frozen negatives. The shared +registration implementation now serves both the numeric V3 probe and the +independently fitted phase model. + +A discovery-only scale-and-translation view search recovered all four 5% +cropped temporal images with the independently fitted model after lowering the +active-support gate from 0.50 to 0.40. It produced zero positives on the 194 +frozen negatives and on the same preregistered 3,000-image COCO challenge used +by the identity scorer. The latter result has a zero-error one-sided 95% bound +of 0.0998% only for that abstention challenge: every COCO image remained +outside carrier support, with a maximum active fraction of 0.201. The scale +rule is not frozen because its support threshold was selected after inspecting +the crop examples. It requires a new temporal positive holdout before it can +join the detector rule. + +The current actionable research candidate remains a positive-only, +provider-specific expert for the supported 1536x2816 carrier epoch. Identity +and bounded translation views use the frozen phase and support thresholds; +unsupported geometry, insufficient carrier magnitude, and ambiguous phase +return `abstain`. Vendor attribution may select the expert that supplied accepted +evidence, but it must not turn an abstention into a provider label. The next +calibration gate still requires at least 3,000 native-support negatives, +same-provider oracle negatives, matched non-target solid outputs, and a new +temporal positive that influenced neither profile nor threshold. + ## Decision record The program has four possible honest outcomes per provider: diff --git a/docs/synthid.md b/docs/synthid.md index 5ab7b83..30cdd0b 100644 --- a/docs/synthid.md +++ b/docs/synthid.md @@ -279,8 +279,20 @@ decision. Source labels therefore remain suitable for vendor triage and hard negative evaluation, not for establishing a SynthID detector without counterfactual or oracle watermark labels. -The protocol, exact limitations, and next experiments are recorded in the -[`detector and removal research plan`](synthid-detector-removal-plan.md). +A later low-content matrix also rejected wavelet energy and normalized +wavelet-shape classifiers: they separated real target outputs from artificial +flat, Gaussian, and phase-random controls, then accepted 61.6-100% of real +external negatives. The surviving branch is narrower. A provider-specific +1536x2816 phase carrier detected all four temporal-test positives, bounded +translation registration recovered all four one-pixel shifts, and the joint +phase/support rule produced zero positives on 194 frozen negatives. A +scale-and-translation discovery rule also produced zero positives on a +preregistered 3,000-image COCO challenge, but every COCO image was outside +carrier support and the scale threshold was selected post hoc. The result is +therefore a positive-only, geometry- and epoch-specific expert with abstention, +not a universal SynthID detector. Exact measurements and remaining calibration +gates are in the +[`detector and removal research plan`](synthid-detector-removal-plan.md#2026-08-10-low-content-controls-and-registered-phase-carrier). A controlled study (June 2026, clean v0.8.6 with text/face protection OFF, native resolution on this repo's default SDXL pipeline) measured the minimum diff --git a/scripts/synthid_color_space_probe.py b/scripts/synthid_color_space_probe.py index baed365..3d451d0 100644 --- a/scripts/synthid_color_space_probe.py +++ b/scripts/synthid_color_space_probe.py @@ -18,6 +18,7 @@ import cv2 import numpy as np from PIL import Image from synthid_phase_carrier import _leave_one_out_coherence +from synthid_phase_registration import extract_frequency_values from synthid_v3_codebook_probe import load_v3_model log = logging.getLogger(__name__) @@ -138,16 +139,6 @@ def _load_rgb(path: Path, *, height: int, width: int) -> np.ndarray: return np.asarray(rgb, dtype=np.float64) -def _extract_values(pixels: np.ndarray, bins: np.ndarray) -> np.ndarray: - """Extract complex rFFT values at sparse BINS from three-channel PIXELS.""" - values = np.empty(len(bins), dtype=np.complex128) - for channel in range(3): - positions = np.flatnonzero(bins[:, 2] == channel) - spectrum = np.fft.rfft2(pixels[:, :, channel]) - values[positions] = spectrum[bins[positions, 0], bins[positions, 1]] - return values - - def discover_model( paths: list[Path], *, @@ -180,7 +171,12 @@ def discover_model( image_values = np.empty((len(paths), len(bins)), dtype=np.complex128) for index, path in enumerate(paths): rgb = _load_rgb(path, height=height, width=width) - image_values[index] = _extract_values(transform_color_space(rgb, color_space), bins) + image_values[index] = extract_frequency_values( + transform_color_space(rgb, color_space), + bins[:, 0], + bins[:, 1], + bins[:, 2], + ) magnitudes = np.abs(image_values) units = np.divide(image_values, magnitudes, out=np.zeros_like(image_values), where=magnitudes != 0.0) unit_sum = np.sum(units, axis=0) @@ -213,7 +209,12 @@ def score_image(path: Path, model: ColorPhaseModel) -> ColorPhaseScore: """Score PATH against MODEL and expose additive channel evidence.""" rgb = _load_rgb(path, height=model.height, width=model.width) bins = np.column_stack((model.rows, model.columns, model.channels)) - values = _extract_values(transform_color_space(rgb, model.color_space), bins) + values = extract_frequency_values( + transform_color_space(rgb, model.color_space), + bins[:, 0], + bins[:, 1], + bins[:, 2], + ) magnitude_gate = np.minimum(np.abs(values) / (model.expected_magnitudes + 1e-12), 1.0) contributions = model.weights * magnitude_gate * np.cos(np.angle(values) - model.phases) active_weights = model.weights * magnitude_gate diff --git a/scripts/synthid_phase_carrier.py b/scripts/synthid_phase_carrier.py index 8c4fce4..92b8cc6 100644 --- a/scripts/synthid_phase_carrier.py +++ b/scripts/synthid_phase_carrier.py @@ -16,6 +16,7 @@ from pathlib import Path import click import numpy as np from PIL import Image +from synthid_phase_registration import MAX_TRANSLATION_SHIFT, extract_frequency_values, register_phase_translations log = logging.getLogger(__name__) @@ -44,6 +45,18 @@ class PhaseCarrierScore: peak_count: int +@dataclass(frozen=True) +class RegisteredPhaseCarrierScore: + """Best phase-carrier score across a bounded translation search.""" + + path: str + score: float + active_weight_fraction: float + peak_count: int + row_shift: int + column_shift: int + + def _load_rgb(path: Path, *, height: int, width: int, canonicalize_geometry: bool = False) -> np.ndarray: """Load PATH as float64 RGB, optionally resizing to model geometry.""" with Image.open(path) as image: @@ -161,6 +174,11 @@ def discover_model( ) +def _frequency_values(pixels: np.ndarray, model: PhaseCarrierModel) -> np.ndarray: + """Return MODEL's selected complex coefficients from PIXELS.""" + return extract_frequency_values(pixels, model.rows, model.columns, model.channels) + + def score_image( path: Path, model: PhaseCarrierModel, @@ -174,13 +192,7 @@ def score_image( width=model.width, canonicalize_geometry=canonicalize_geometry, ) - values = np.empty(len(model.rows), dtype=np.complex128) - for channel in range(3): - positions = np.flatnonzero(model.channels == channel) - if len(positions) == 0: - continue - spectrum = np.fft.rfft2(pixels[:, :, channel]) - values[positions] = spectrum[model.rows[positions], model.columns[positions]] + values = _frequency_values(pixels, model) magnitude_gate = np.minimum(np.abs(values) / (model.expected_magnitudes + 1e-12), 1.0) active_weights = model.weights * magnitude_gate active_weight = float(np.sum(active_weights)) @@ -197,6 +209,41 @@ def score_image( ) +def score_translations( + path: Path, + model: PhaseCarrierModel, + *, + max_shift: int = 4, + canonicalize_geometry: bool = False, +) -> RegisteredPhaseCarrierScore: + """Return the best score after compensating bounded integer translations.""" + pixels = _load_rgb( + path, + height=model.height, + width=model.width, + canonicalize_geometry=canonicalize_geometry, + ) + registration = register_phase_translations( + _frequency_values(pixels, model), + phases=model.phases, + weights=model.weights, + expected_magnitudes=model.expected_magnitudes, + rows=model.rows, + columns=model.columns, + height=model.height, + width=model.width, + max_shift=max_shift, + ) + return RegisteredPhaseCarrierScore( + path=str(path), + score=registration.score, + active_weight_fraction=registration.active_weight_fraction, + peak_count=len(model.rows), + row_shift=registration.row_shift, + column_shift=registration.column_shift, + ) + + def save_model(path: Path, model: PhaseCarrierModel) -> None: """Save MODEL as a validated numeric NPZ artifact.""" path.parent.mkdir(parents=True, exist_ok=True) @@ -287,21 +334,39 @@ def discover( @click.argument("images", nargs=-1, required=True, type=click.Path(exists=True, dir_okay=False, path_type=Path)) @click.option("--report-out", type=click.Path(dir_okay=False, path_type=Path), required=True) @click.option("--canonicalize-geometry", is_flag=True, help="Resize inputs to the model geometry before scoring.") +@click.option("--max-shift", type=click.IntRange(min=0, max=MAX_TRANSLATION_SHIFT), default=0, show_default=True) def score( model_path: Path, images: tuple[Path, ...], report_out: Path, canonicalize_geometry: bool, + max_shift: int, ) -> None: """Score IMAGES with MODEL_PATH.""" model = load_model(model_path) + scores = ( + [asdict(score_image(image, model, canonicalize_geometry=canonicalize_geometry)) for image in images] + if max_shift == 0 + else [ + asdict( + score_translations( + image, + model, + max_shift=max_shift, + canonicalize_geometry=canonicalize_geometry, + ) + ) + for image in images + ] + ) payload = { "model": str(model_path), "height": model.height, "width": model.width, "peak_count": len(model.rows), "canonicalize_geometry": canonicalize_geometry, - "scores": [asdict(score_image(image, model, canonicalize_geometry=canonicalize_geometry)) for image in images], + "max_shift": max_shift, + "scores": scores, } report_out.parent.mkdir(parents=True, exist_ok=True) report_out.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") diff --git a/scripts/synthid_phase_registration.py b/scripts/synthid_phase_registration.py new file mode 100644 index 0000000..6e4f1e3 --- /dev/null +++ b/scripts/synthid_phase_registration.py @@ -0,0 +1,92 @@ +"""Shared bounded translation registration for phase-carrier probes.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np + +MAX_TRANSLATION_SHIFT = 32 + + +@dataclass(frozen=True) +class TranslationRegistration: + """Best phase score and offset from one bounded translation search.""" + + score: float + active_weight_fraction: float + row_shift: int + column_shift: int + + +def phase_adjustment( + rows: np.ndarray, + columns: np.ndarray, + *, + height: int, + width: int, + row_shift: int | np.ndarray, + column_shift: int | np.ndarray, +) -> np.ndarray: + """Return the Fourier phase adjustment for one integer translation.""" + signed_rows = np.where(rows > height // 2, rows - height, rows) + return 2.0 * np.pi * (signed_rows * row_shift / height + columns * column_shift / width) + + +def extract_frequency_values( + pixels: np.ndarray, + rows: np.ndarray, + columns: np.ndarray, + channels: np.ndarray, +) -> np.ndarray: + """Extract sparse three-channel rFFT coefficients from PIXELS.""" + values = np.empty(len(rows), dtype=np.complex128) + for channel in range(3): + positions = np.flatnonzero(channels == channel) + if len(positions) == 0: + continue + spectrum = np.fft.rfft2(pixels[:, :, channel]) + values[positions] = spectrum[rows[positions], columns[positions]] + return values + + +def register_phase_translations( + values: np.ndarray, + *, + phases: np.ndarray, + weights: np.ndarray, + expected_magnitudes: np.ndarray, + rows: np.ndarray, + columns: np.ndarray, + height: int, + width: int, + max_shift: int, +) -> TranslationRegistration: + """Find the strongest phase alignment over bounded integer translations.""" + if not 0 <= max_shift <= MAX_TRANSLATION_SHIFT: + raise ValueError(f"max_shift must be between 0 and {MAX_TRANSLATION_SHIFT}") + magnitude_gate = np.minimum(np.abs(values) / (expected_magnitudes + 1e-12), 1.0) + active_weights = weights * magnitude_gate + active_weight = float(np.sum(active_weights)) + if active_weight == 0.0: + return TranslationRegistration(0.0, 0.0, 0, 0) + + shifts = np.arange(-max_shift, max_shift + 1) + adjustment = phase_adjustment( + rows[:, None, None], + columns[:, None, None], + height=height, + width=width, + row_shift=shifts[None, :, None], + column_shift=shifts[None, None, :], + ) + difference = np.angle(values) - phases + scores = np.sum(active_weights[:, None, None] * np.cos(difference[:, None, None] + adjustment), axis=0) + scores /= active_weight + row_index, column_index = np.unravel_index(int(np.argmax(scores)), scores.shape) + return TranslationRegistration( + score=float(scores[row_index, column_index]), + active_weight_fraction=active_weight, + row_shift=int(shifts[row_index]), + column_shift=int(shifts[column_index]), + ) diff --git a/scripts/synthid_v3_codebook_probe.py b/scripts/synthid_v3_codebook_probe.py index fcf8828..bb65d97 100644 --- a/scripts/synthid_v3_codebook_probe.py +++ b/scripts/synthid_v3_codebook_probe.py @@ -1,9 +1,9 @@ """Independently evaluate a numeric reverse-SynthID V3 NPZ codebook. -The loader accepts only the documented numeric format-v2 arrays and disables -pickle. It does not import or execute third-party code. Scores are exploratory: -the external reference provenance and labels still require independent oracle -validation before this can support a SynthID detector claim. +The loader accepts only the documented dense or sparse numeric format-v2 arrays +and disables pickle. It does not import or execute third-party code. Scores are +exploratory: the external reference provenance and labels still require +independent oracle validation before this can support a SynthID detector claim. """ from __future__ import annotations @@ -16,8 +16,15 @@ from pathlib import Path import click import numpy as np from PIL import Image +from synthid_phase_registration import ( + MAX_TRANSLATION_SHIFT, + extract_frequency_values, + phase_adjustment, + register_phase_translations, +) log = logging.getLogger(__name__) +LOG_2 = np.log(2.0) @dataclass(frozen=True) @@ -45,15 +52,139 @@ class V3Score: peak_count: int +@dataclass(frozen=True) +class V3RegisteredScore: + """Best phase-alignment score across a bounded translation search.""" + + path: str + phase_score: float + axial_phase_score: float + active_weight_fraction: float + peak_count: int + row_shift: int + column_shift: int + + +@dataclass(frozen=True) +class _CarrierCandidates: + """Selected numeric carrier arrays from one V3 profile.""" + + weights: np.ndarray + rows: np.ndarray + columns: np.ndarray + channels: np.ndarray + phases: np.ndarray + magnitudes: np.ndarray + + def _load_sparse_channel(artifact: np.lib.npyio.NpzFile, prefix: str, channel: int) -> tuple[np.ndarray, ...]: """Load one sparse channel without reconstructing full image-sized arrays.""" indices = np.asarray(artifact[f"{prefix}idx_{channel}"], dtype=np.uint32) - magnitudes = np.exp2(np.asarray(artifact[f"{prefix}mag_{channel}"], dtype=np.float64)) - 1.0 - phases = np.asarray(artifact[f"{prefix}phase_{channel}"], dtype=np.float64) + log_magnitudes = np.asarray(artifact[f"{prefix}mag_{channel}"]) + phases = np.asarray(artifact[f"{prefix}phase_{channel}"]) coherence = np.asarray(artifact[f"{prefix}cons_{channel}"], dtype=np.float64) / 255.0 - if not (indices.shape == magnitudes.shape == phases.shape == coherence.shape): + if not (indices.shape == log_magnitudes.shape == phases.shape == coherence.shape): raise ValueError("sparse profile arrays have inconsistent shapes") - return indices, magnitudes, phases, coherence + return indices, log_magnitudes, phases, coherence + + +def _top_positions(selection: np.ndarray, tie_breaker: np.ndarray, peak_count: int) -> np.ndarray: + """Return deterministic descending positions for the strongest candidates.""" + if len(selection) < peak_count: + raise ValueError(f"profile exposes only {len(selection)} eligible bins") + cutoff = np.partition(selection, -peak_count)[-peak_count] + stronger = np.flatnonzero(selection > cutoff) + tied = np.flatnonzero(selection == cutoff) + remaining = peak_count - len(stronger) + if remaining < len(tied): + tied = tied[np.argpartition(tie_breaker[tied], -remaining)[-remaining:]] + positions = np.concatenate((stronger, tied)) + order = np.lexsort((-tie_breaker[positions], -selection[positions])) + return positions[order] + + +def _dense_candidates( + artifact: np.lib.npyio.NpzFile, + prefix: str, + *, + height: int, + width: int, + min_radius: float, + peak_count: int, +) -> _CarrierCandidates: + """Select candidate arrays from one dense numeric profile.""" + shape = (height, width // 2 + 1, 3) + log_magnitudes = np.asarray(artifact[f"{prefix}mag"]) + phases = np.asarray(artifact[f"{prefix}phase"]) + coherence = np.asarray(artifact[f"{prefix}cons"], dtype=np.float64) / 255.0 + if not (log_magnitudes.shape == phases.shape == coherence.shape == shape): + raise ValueError("dense profile arrays have inconsistent shapes") + + rows = np.arange(height) + signed_rows = np.where(rows > height // 2, rows - height, rows) + columns = np.arange(shape[1]) + radius = np.sqrt(np.square(signed_rows[:, None]) + np.square(columns[None, :])) + valid_spatial = (radius >= min_radius) & (columns[None, :] > 0) + valid = np.broadcast_to(valid_spatial[:, :, None], shape) + flat_valid = np.flatnonzero(valid) + selection = (np.square(coherence) * log_magnitudes * LOG_2).ravel()[flat_valid] + positions = _top_positions(selection, flat_valid, peak_count) + selected = flat_valid[positions] + selected_rows, selected_columns, selected_channels = np.unravel_index(selected, shape) + return _CarrierCandidates( + weights=selection[positions], + rows=selected_rows, + columns=selected_columns, + channels=selected_channels, + phases=np.asarray(phases.ravel()[selected], dtype=np.float64), + magnitudes=np.exp2(np.asarray(log_magnitudes.ravel()[selected], dtype=np.float64)) - 1.0, + ) + + +def _sparse_candidates( + artifact: np.lib.npyio.NpzFile, + prefix: str, + *, + height: int, + width: int, + min_radius: float, + peak_count: int, +) -> _CarrierCandidates: + """Select candidate arrays from one sparse numeric profile.""" + half_width = width // 2 + 1 + selections: list[np.ndarray] = [] + candidate_rows: list[np.ndarray] = [] + candidate_columns: list[np.ndarray] = [] + candidate_channels: list[np.ndarray] = [] + candidate_phases: list[np.ndarray] = [] + candidate_log_magnitudes: list[np.ndarray] = [] + for channel in range(3): + indices, log_magnitudes, phases, coherence = _load_sparse_channel(artifact, prefix, channel) + rows, columns = np.unravel_index(indices, (height, half_width)) + signed_rows = np.where(rows > height // 2, rows - height, rows) + radius = np.sqrt(np.square(signed_rows) + np.square(columns)) + valid = (radius >= min_radius) & (columns > 0) + selections.append((np.square(coherence) * log_magnitudes * LOG_2)[valid]) + candidate_rows.append(rows[valid]) + candidate_columns.append(columns[valid]) + candidate_channels.append(np.full(np.count_nonzero(valid), channel, dtype=np.int8)) + candidate_phases.append(phases[valid]) + candidate_log_magnitudes.append(log_magnitudes[valid]) + + selection = np.concatenate(selections) + rows = np.concatenate(candidate_rows) + columns = np.concatenate(candidate_columns) + channels = np.concatenate(candidate_channels) + tie_breaker = np.ravel_multi_index((rows, columns, channels), (height, half_width, 3)) + selected = _top_positions(selection, tie_breaker, peak_count) + return _CarrierCandidates( + weights=selection[selected], + rows=rows[selected], + columns=columns[selected], + channels=channels[selected], + phases=np.asarray(np.concatenate(candidate_phases)[selected], dtype=np.float64), + magnitudes=np.exp2(np.asarray(np.concatenate(candidate_log_magnitudes)[selected], dtype=np.float64)) - 1.0, + ) def load_v3_model( @@ -66,44 +197,27 @@ def load_v3_model( ) -> V3CarrierModel: """Load top phase-consistent bins from a numeric V3 codebook profile.""" prefix = f"{height}x{width}/" - half_width = width // 2 + 1 - candidates: list[tuple[float, int, int, int, float, float]] = [] with np.load(path, allow_pickle=False) as artifact: if int(artifact["format_version"]) != 2: raise ValueError("only numeric V3 format version 2 is supported") - if not bool(int(artifact[f"{prefix}sparse"])): - raise ValueError("only sparse profiles are supported by this audit loader") - for channel in range(3): - indices, magnitudes, phases, coherence = _load_sparse_channel(artifact, prefix, channel) - rows, columns = np.unravel_index(indices, (height, half_width)) - signed_rows = np.where(rows > height // 2, rows - height, rows) - radius = np.sqrt(np.square(signed_rows) + np.square(columns)) - valid = (radius >= min_radius) & (columns > 0) - selection = np.square(coherence) * np.log1p(magnitudes) - for index in np.flatnonzero(valid): - candidates.append( - ( - float(selection[index]), - int(rows[index]), - int(columns[index]), - channel, - float(phases[index]), - float(magnitudes[index]), - ) - ) - if len(candidates) < peak_count: - raise ValueError(f"profile exposes only {len(candidates)} eligible bins") - selected = sorted(candidates, reverse=True)[:peak_count] - raw_weights = np.asarray([item[0] for item in selected], dtype=np.float64) + loader = _sparse_candidates if bool(int(artifact[f"{prefix}sparse"])) else _dense_candidates + candidates = loader( + artifact, + prefix, + height=height, + width=width, + min_radius=min_radius, + peak_count=peak_count, + ) return V3CarrierModel( height=height, width=width, - rows=np.asarray([item[1] for item in selected], dtype=np.int32), - columns=np.asarray([item[2] for item in selected], dtype=np.int32), - channels=np.asarray([item[3] for item in selected], dtype=np.int8), - phases=np.asarray([item[4] for item in selected], dtype=np.float64), - weights=raw_weights / np.sum(raw_weights), - expected_magnitudes=np.asarray([item[5] for item in selected], dtype=np.float64), + rows=np.asarray(candidates.rows, dtype=np.int32), + columns=np.asarray(candidates.columns, dtype=np.int32), + channels=np.asarray(candidates.channels, dtype=np.int8), + phases=candidates.phases, + weights=candidates.weights / np.sum(candidates.weights), + expected_magnitudes=candidates.magnitudes, ) @@ -116,16 +230,18 @@ def _load_profile_rgb(path: Path, model: V3CarrierModel) -> np.ndarray: return np.asarray(image, dtype=np.float64) -def score_image(path: Path, model: V3CarrierModel) -> V3Score: - """Score PATH against selected V3 phase bins.""" +def _frequency_values(path: Path, model: V3CarrierModel) -> np.ndarray: + """Return the selected complex coefficients from PATH.""" pixels = _load_profile_rgb(path, model) - values = np.empty(len(model.rows), dtype=np.complex128) - for channel in range(3): - positions = np.flatnonzero(model.channels == channel) - if len(positions) == 0: - continue - spectrum = np.fft.fft2(pixels[:, :, channel]) - values[positions] = spectrum[model.rows[positions], model.columns[positions]] + return extract_frequency_values(pixels, model.rows, model.columns, model.channels) + + +def _score_values( + values: np.ndarray, + model: V3CarrierModel, + phase_offsets: np.ndarray | float = 0.0, +) -> tuple[float, float, float]: + """Return phase, axial-phase, and active-weight scores for VALUES.""" phase_difference = np.angle(values) - model.phases magnitude_gate = np.minimum(np.abs(values) / (model.expected_magnitudes + 1e-12), 1.0) active_weights = model.weights * magnitude_gate @@ -134,8 +250,15 @@ def score_image(path: Path, model: V3CarrierModel) -> V3Score: phase_score = 0.0 axial_score = 0.0 else: - phase_score = float(np.sum(active_weights * np.cos(phase_difference)) / active_weight) - axial_score = float(np.sum(active_weights * np.cos(2.0 * phase_difference)) / active_weight) + adjusted = phase_difference + phase_offsets + phase_score = float(np.sum(active_weights * np.cos(adjusted)) / active_weight) + axial_score = float(np.sum(active_weights * np.cos(2.0 * adjusted)) / active_weight) + return phase_score, axial_score, active_weight + + +def score_image(path: Path, model: V3CarrierModel) -> V3Score: + """Score PATH against selected V3 phase bins.""" + phase_score, axial_score, active_weight = _score_values(_frequency_values(path, model), model) return V3Score( path=str(path), phase_score=phase_score, @@ -145,22 +268,71 @@ def score_image(path: Path, model: V3CarrierModel) -> V3Score: ) +def score_translations(path: Path, model: V3CarrierModel, *, max_shift: int = 4) -> V3RegisteredScore: + """Return the best score after compensating bounded integer translations.""" + values = _frequency_values(path, model) + registration = register_phase_translations( + values, + phases=model.phases, + weights=model.weights, + expected_magnitudes=model.expected_magnitudes, + rows=model.rows, + columns=model.columns, + height=model.height, + width=model.width, + max_shift=max_shift, + ) + selected_adjustment = phase_adjustment( + model.rows, + model.columns, + height=model.height, + width=model.width, + row_shift=registration.row_shift, + column_shift=registration.column_shift, + ) + phase_score, axial_score, active_weight = _score_values(values, model, selected_adjustment) + return V3RegisteredScore( + path=str(path), + phase_score=phase_score, + axial_phase_score=axial_score, + active_weight_fraction=active_weight, + peak_count=len(model.rows), + row_shift=registration.row_shift, + column_shift=registration.column_shift, + ) + + @click.command() @click.argument("codebook", type=click.Path(exists=True, dir_okay=False, path_type=Path)) @click.argument("images", nargs=-1, required=True, type=click.Path(exists=True, dir_okay=False, path_type=Path)) @click.option("--height", type=click.IntRange(min=64), required=True) @click.option("--width", type=click.IntRange(min=64), required=True) @click.option("--peak-count", type=click.IntRange(min=1), default=256, show_default=True) +@click.option("--max-shift", type=click.IntRange(min=0, max=MAX_TRANSLATION_SHIFT), default=0, show_default=True) @click.option("--report-out", type=click.Path(dir_okay=False, path_type=Path), required=True) -def main(codebook: Path, images: tuple[Path, ...], height: int, width: int, peak_count: int, report_out: Path) -> None: +def main( + codebook: Path, + images: tuple[Path, ...], + height: int, + width: int, + peak_count: int, + max_shift: int, + report_out: Path, +) -> None: """Score IMAGES against one exact-resolution profile from CODEBOOK.""" model = load_v3_model(codebook, height=height, width=width, peak_count=peak_count) + scores = ( + [asdict(score_image(image, model)) for image in images] + if max_shift == 0 + else [asdict(score_translations(image, model, max_shift=max_shift)) for image in images] + ) payload = { "codebook": str(codebook), "height": height, "width": width, "peak_count": peak_count, - "scores": [asdict(score_image(image, model)) for image in images], + "max_shift": max_shift, + "scores": scores, } report_out.parent.mkdir(parents=True, exist_ok=True) report_out.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") diff --git a/tests/test_synthid_phase_carrier.py b/tests/test_synthid_phase_carrier.py index 62ce0fe..464e470 100644 --- a/tests/test_synthid_phase_carrier.py +++ b/tests/test_synthid_phase_carrier.py @@ -118,3 +118,30 @@ def test_scoring_can_canonicalize_geometry(tmp_path: Path) -> None: assert score.path == str(mismatch) assert score.peak_count == 4 + + +def test_translation_search_recovers_shifted_carrier(tmp_path: Path) -> None: + positives: list[Path] = [] + for index in range(4): + path = tmp_path / f"positive-{index}.png" + _write_image(path, phase=0.4, seed=index) + positives.append(path) + heldout = tmp_path / "heldout.png" + shifted = tmp_path / "shifted.png" + _write_image(heldout, phase=0.4, seed=10) + with Image.open(heldout) as source: + pixels = np.asarray(source).copy() + Image.fromarray(np.roll(pixels, shift=(1, 1), axis=(0, 1)), mode="RGB").save(shifted) + model = carrier.discover_model(positives, peak_count=8, min_radius=1.0) + + fixed = carrier.score_image(shifted, model) + unregistered = carrier.score_translations(shifted, model, max_shift=0) + registered = carrier.score_translations(shifted, model, max_shift=2) + + assert unregistered.score == pytest.approx(fixed.score) + assert unregistered.active_weight_fraction == pytest.approx(fixed.active_weight_fraction) + assert registered.score > fixed.score + assert abs(registered.row_shift) <= 2 + assert abs(registered.column_shift) <= 2 + with pytest.raises(ValueError, match="between 0 and 32"): + carrier.score_translations(shifted, model, max_shift=33) diff --git a/tests/test_synthid_v3_codebook_probe.py b/tests/test_synthid_v3_codebook_probe.py index 6c5df84..08177a1 100644 --- a/tests/test_synthid_v3_codebook_probe.py +++ b/tests/test_synthid_v3_codebook_probe.py @@ -30,6 +30,29 @@ def _write_codebook(path: Path, *, height: int, width: int, phase: float) -> Non np.savez(path, **payload) +def _write_dense_codebook(path: Path, *, height: int, width: int, phase: float) -> None: + half_width = width // 2 + 1 + magnitudes = np.zeros((height, half_width, 3), dtype=np.float16) + phases = np.zeros_like(magnitudes) + coherence = np.zeros_like(magnitudes, dtype=np.uint8) + rows = np.asarray([7, 11, 13, 17]) + columns = np.asarray([5, 9, 12, 15]) + for channel in range(3): + magnitudes[rows, columns, channel] = np.log2(1.0 + np.asarray([1000.0, 10.0, 10.0, 10.0])) + phases[rows, columns, channel] = phase + coherence[rows, columns, channel] = 255 + np.savez( + path, + format_version=np.asarray(2), + **{ + f"{height}x{width}/sparse": np.asarray(0), + f"{height}x{width}/mag": magnitudes, + f"{height}x{width}/phase": phases, + f"{height}x{width}/cons": coherence, + }, + ) + + def _write_carrier(path: Path, *, height: int, width: int, phase: float) -> None: yy, xx = np.mgrid[:height, :width] carrier = 80.0 + 20.0 * np.cos(2.0 * np.pi * (7.0 * yy / height + 5.0 * xx / width) + phase) @@ -51,6 +74,40 @@ def test_numeric_codebook_scores_matching_phase(tmp_path: Path) -> None: assert score.phase_score > 0.0 +def test_dense_numeric_codebook_scores_matching_phase(tmp_path: Path) -> None: + height = width = 64 + codebook = tmp_path / "dense-codebook.npz" + image = tmp_path / "image.png" + _write_dense_codebook(codebook, height=height, width=width, phase=0.4) + _write_carrier(image, height=height, width=width, phase=0.4) + + model = probe.load_v3_model(codebook, height=height, width=width, peak_count=4, min_radius=1.0) + score = probe.score_image(image, model) + + assert score.peak_count == 4 + assert score.phase_score > 0.0 + + +def test_translation_search_recovers_shifted_carrier(tmp_path: Path) -> None: + height = width = 64 + codebook = tmp_path / "codebook.npz" + image = tmp_path / "image.png" + shifted = tmp_path / "shifted.png" + _write_codebook(codebook, height=height, width=width, phase=0.4) + _write_carrier(image, height=height, width=width, phase=0.4) + with Image.open(image) as source: + pixels = np.asarray(source).copy() + Image.fromarray(np.roll(pixels, shift=(1, 1), axis=(0, 1)), mode="RGB").save(shifted) + model = probe.load_v3_model(codebook, height=height, width=width, peak_count=4, min_radius=1.0) + + fixed = probe.score_image(shifted, model) + registered = probe.score_translations(shifted, model, max_shift=2) + + assert registered.phase_score > fixed.phase_score + assert abs(registered.row_shift) <= 2 + assert abs(registered.column_shift) <= 2 + + def test_rejects_wrong_format(tmp_path: Path) -> None: artifact = tmp_path / "bad.npz" np.savez(artifact, format_version=np.asarray(1))