mirror of
https://github.com/wiltodelta/remove-ai-watermarks.git
synced 2026-08-31 09:40:38 +02:00
Add production SynthID routing and OpenAI verification
This commit is contained in:
Victor Kuznetsov
@@ -0,0 +1,195 @@
|
||||
"""Suppress the recovered periodic carrier without image regeneration.
|
||||
|
||||
This research tool controls the project's local fixed-template score. A local
|
||||
score reversal is not evidence that a provider SynthID verifier will stop
|
||||
detecting the image.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import math
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import click
|
||||
from PIL import Image
|
||||
from synthid_pixel_attack import load_rgb, measure # pyright: ignore[reportUnknownVariableType]
|
||||
from synthid_research_manifest import artifact_sha256
|
||||
from synthid_tile_attack import subtract_tiled_template
|
||||
|
||||
from remove_ai_watermarks.synthid_detector import (
|
||||
TILE_THRESHOLD,
|
||||
_geometry_supported, # pyright: ignore[reportPrivateUsage]
|
||||
_load_template, # pyright: ignore[reportPrivateUsage]
|
||||
folded_template_score,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from numpy.typing import NDArray
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def apply_template(pixels: NDArray[Any], template: NDArray[Any], *, amplitude: float) -> NDArray[Any]:
|
||||
"""Subtract AMPLITUDE times periodic TEMPLATE from arbitrary RGB PIXELS."""
|
||||
return subtract_tiled_template(pixels, template, strength=amplitude)
|
||||
|
||||
|
||||
def carrier_score(pixels: NDArray[Any], template: NDArray[Any], sigma: float) -> float:
|
||||
"""Return the local fixed-template carrier score for PIXELS."""
|
||||
score, _folded = folded_template_score(pixels, template, sigma)
|
||||
return score
|
||||
|
||||
|
||||
def find_minimum_amplitude(
|
||||
pixels: NDArray[Any],
|
||||
template: NDArray[Any],
|
||||
sigma: float,
|
||||
*,
|
||||
target_score: float,
|
||||
maximum_amplitude: float,
|
||||
iterations: int,
|
||||
) -> tuple[float, NDArray[Any], float]:
|
||||
"""Return the smallest searched amplitude whose score reaches TARGET_SCORE."""
|
||||
if not math.isfinite(target_score):
|
||||
raise ValueError("target score must be finite")
|
||||
if not math.isfinite(maximum_amplitude) or maximum_amplitude <= 0.0:
|
||||
raise ValueError("maximum amplitude must be finite and positive")
|
||||
if iterations < 1:
|
||||
raise ValueError("iterations must be positive")
|
||||
maximum_pixels = apply_template(pixels, template, amplitude=maximum_amplitude)
|
||||
maximum_score = carrier_score(maximum_pixels, template, sigma)
|
||||
if maximum_score > target_score:
|
||||
raise ValueError(
|
||||
f"maximum amplitude {maximum_amplitude:g} reached score {maximum_score:.6f}, "
|
||||
f"above target {target_score:.6f}"
|
||||
)
|
||||
|
||||
low = 0.0
|
||||
high = maximum_amplitude
|
||||
best_pixels = maximum_pixels
|
||||
best_score = maximum_score
|
||||
for _iteration in range(iterations):
|
||||
middle = (low + high) / 2.0
|
||||
candidate = apply_template(pixels, template, amplitude=middle)
|
||||
candidate_score = carrier_score(candidate, template, sigma)
|
||||
if candidate_score <= target_score:
|
||||
high = middle
|
||||
best_pixels = candidate
|
||||
best_score = candidate_score
|
||||
else:
|
||||
low = middle
|
||||
return high, best_pixels, best_score
|
||||
|
||||
|
||||
def suppress_carrier(
|
||||
pixels: NDArray[Any],
|
||||
*,
|
||||
target_score: float = -0.25,
|
||||
maximum_amplitude: float = 40.0,
|
||||
iterations: int = 8,
|
||||
) -> tuple[NDArray[Any], dict[str, float | int | str]]:
|
||||
"""Suppress a locally detected carrier and return pixels plus measurements."""
|
||||
height, width = pixels.shape[:2]
|
||||
if not _geometry_supported(width, height):
|
||||
raise ValueError(f"unsupported decoded geometry: {width}x{height}")
|
||||
if target_score >= TILE_THRESHOLD:
|
||||
raise ValueError(f"target score must be below the detector threshold {TILE_THRESHOLD:.6f}")
|
||||
template, sigma, _model_height, _model_width, tile_height, tile_width = _load_template()
|
||||
original_score = carrier_score(pixels, template, sigma)
|
||||
if original_score < TILE_THRESHOLD:
|
||||
raise ValueError(
|
||||
f"local carrier is not detected: score {original_score:.6f} is below threshold {TILE_THRESHOLD:.6f}"
|
||||
)
|
||||
|
||||
started = time.perf_counter()
|
||||
amplitude, candidate, candidate_score = find_minimum_amplitude(
|
||||
pixels,
|
||||
template,
|
||||
sigma,
|
||||
target_score=target_score,
|
||||
maximum_amplitude=maximum_amplitude,
|
||||
iterations=iterations,
|
||||
)
|
||||
quality = measure(pixels, candidate, name="adaptive-carrier", path=Path("<memory>"))
|
||||
return candidate, {
|
||||
"status": "local_carrier_suppressed",
|
||||
"detector_scope": "local fixed-template carrier, not provider-verified SynthID removal",
|
||||
"width": width,
|
||||
"height": height,
|
||||
"tile_height": tile_height,
|
||||
"tile_width": tile_width,
|
||||
"threshold": TILE_THRESHOLD,
|
||||
"target_score": target_score,
|
||||
"original_score": original_score,
|
||||
"candidate_score": candidate_score,
|
||||
"amplitude": amplitude,
|
||||
"maximum_amplitude": maximum_amplitude,
|
||||
"iterations": iterations,
|
||||
"residual_rms": quality.residual_rms,
|
||||
"psnr_db": quality.psnr_db,
|
||||
"ssim": quality.ssim,
|
||||
"changed_pixel_fraction": quality.changed_pixel_fraction,
|
||||
"elapsed_seconds": time.perf_counter() - started,
|
||||
}
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.argument("source", type=click.Path(exists=True, dir_okay=False, path_type=Path))
|
||||
@click.argument("output", type=click.Path(dir_okay=False, path_type=Path))
|
||||
@click.option("--target-score", type=float, default=-0.25, show_default=True)
|
||||
@click.option("--maximum-amplitude", type=click.FloatRange(min=0.0, min_open=True), default=40.0, show_default=True)
|
||||
@click.option("--iterations", type=click.IntRange(min=1), default=8, show_default=True)
|
||||
@click.option("--report-out", type=click.Path(dir_okay=False, path_type=Path))
|
||||
def main(
|
||||
source: Path,
|
||||
output: Path,
|
||||
target_score: float,
|
||||
maximum_amplitude: float,
|
||||
iterations: int,
|
||||
report_out: Path | None,
|
||||
) -> None:
|
||||
"""Write a lossless PNG with the recovered local carrier suppressed."""
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
if output.suffix.lower() != ".png":
|
||||
raise click.BadParameter("output must use the .png extension", param_hint="output")
|
||||
report_path = report_out or output.with_suffix(".json")
|
||||
for path in (output, report_path):
|
||||
if path.exists():
|
||||
raise click.ClickException(f"refusing to overwrite existing file: {path}")
|
||||
try:
|
||||
candidate, report = suppress_carrier(
|
||||
load_rgb(source), # pyright: ignore[reportUnknownArgumentType]
|
||||
target_score=target_score,
|
||||
maximum_amplitude=maximum_amplitude,
|
||||
iterations=iterations,
|
||||
)
|
||||
except ValueError as error:
|
||||
raise click.ClickException(str(error)) from error
|
||||
output.parent.mkdir(parents=True, exist_ok=True)
|
||||
report_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
Image.fromarray(candidate, mode="RGB").save(output, format="PNG", compress_level=9)
|
||||
report.update(
|
||||
{
|
||||
"source": str(source.resolve()),
|
||||
"source_sha256": artifact_sha256(source),
|
||||
"output": str(output.resolve()),
|
||||
"output_sha256": artifact_sha256(output),
|
||||
}
|
||||
)
|
||||
report_path.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n", encoding="utf-8")
|
||||
log.info(
|
||||
"Suppressed local carrier %.6f -> %.6f at %.2f dB PSNR; wrote %s",
|
||||
report["original_score"],
|
||||
report["candidate_score"],
|
||||
report["psnr_db"],
|
||||
output,
|
||||
)
|
||||
log.info("Research caveat: this is not provider-verified SynthID removal")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,352 @@
|
||||
"""Calibrate a versioned SynthID expert bank without forcing binary verdicts.
|
||||
|
||||
This research utility combines already-computed pixel-only expert scores. It
|
||||
does not inspect provenance, metadata, filenames, or provider labels at
|
||||
inference. Expert support must be determined from predeclared geometry or model
|
||||
scope, never from the observed score.
|
||||
|
||||
The clean null is a union test: any supported expert may provide positive
|
||||
evidence, so its smallest empirical upper-tail p-value receives a Bonferroni
|
||||
correction. The watermarked hypothesis is itself a union over possible encoder
|
||||
states and can be rejected only when every configured expert has complete
|
||||
coverage and gives a small empirical lower-tail p-value.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import bisect
|
||||
import json
|
||||
import logging
|
||||
import math
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
from typing import Literal, cast
|
||||
|
||||
import click
|
||||
from synthid_research_manifest import artifact_sha256
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
CascadeVerdict = Literal["detected", "not_detected", "abstain"]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ExpertCalibration:
|
||||
"""Frozen positive and negative score distributions for one expert."""
|
||||
|
||||
name: str
|
||||
positive_scores: tuple[float, ...]
|
||||
negative_scores: tuple[float, ...]
|
||||
higher_is_positive: bool = True
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.name:
|
||||
raise ValueError("expert name must not be empty")
|
||||
if not self.positive_scores or not self.negative_scores:
|
||||
raise ValueError(f"expert {self.name!r} needs positive and negative calibration scores")
|
||||
if not all(math.isfinite(score) for score in (*self.positive_scores, *self.negative_scores)):
|
||||
raise ValueError(f"expert {self.name!r} contains a non-finite calibration score")
|
||||
direction = 1.0 if self.higher_is_positive else -1.0
|
||||
object.__setattr__(self, "positive_scores", tuple(sorted(direction * score for score in self.positive_scores)))
|
||||
object.__setattr__(self, "negative_scores", tuple(sorted(direction * score for score in self.negative_scores)))
|
||||
|
||||
def orient(self, score: float) -> float:
|
||||
"""Return SCORE in the common higher-means-more-positive direction."""
|
||||
return score if self.higher_is_positive else -score
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CascadeConfig:
|
||||
"""Calibration distributions and two-sided decision levels."""
|
||||
|
||||
experts: tuple[ExpertCalibration, ...]
|
||||
positive_alpha: float
|
||||
negative_alpha: float
|
||||
coverage_complete: bool
|
||||
scope: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.experts:
|
||||
raise ValueError("at least one expert is required")
|
||||
names = [expert.name for expert in self.experts]
|
||||
if len(set(names)) != len(names):
|
||||
raise ValueError("expert names must be unique")
|
||||
for label, value in (("positive_alpha", self.positive_alpha), ("negative_alpha", self.negative_alpha)):
|
||||
if not 0.0 < value <= 1.0:
|
||||
raise ValueError(f"{label} must be in (0, 1]")
|
||||
if not self.scope:
|
||||
raise ValueError("detector scope must not be empty")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ExpertObservation:
|
||||
"""One expert score, or an explicit unsupported result."""
|
||||
|
||||
name: str
|
||||
supported: bool
|
||||
score: float | None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.name:
|
||||
raise ValueError("observation expert name must not be empty")
|
||||
if self.supported:
|
||||
if self.score is None or not math.isfinite(self.score):
|
||||
raise ValueError(f"supported expert {self.name!r} needs a finite score")
|
||||
elif self.score is not None:
|
||||
raise ValueError(f"unsupported expert {self.name!r} must not provide a score")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ExpertEvidence:
|
||||
"""Two empirical p-values for one supported expert."""
|
||||
|
||||
name: str
|
||||
score: float
|
||||
clean_null_p_value: float
|
||||
watermarked_p_value: float
|
||||
positive_calibration_count: int
|
||||
negative_calibration_count: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CascadeResult:
|
||||
"""Auditable tri-state verdict for one observation record."""
|
||||
|
||||
verdict: CascadeVerdict
|
||||
reason: str
|
||||
clean_null_p_value: float | None
|
||||
watermarked_p_value: float | None
|
||||
supported_expert_count: int
|
||||
configured_expert_count: int
|
||||
coverage_complete: bool
|
||||
evidence: tuple[ExpertEvidence, ...]
|
||||
|
||||
|
||||
def _upper_tail_p_value(sorted_scores: tuple[float, ...], score: float) -> float:
|
||||
"""Smoothed empirical probability of a calibration score at least SCORE."""
|
||||
tail_count = len(sorted_scores) - bisect.bisect_left(sorted_scores, score)
|
||||
return (tail_count + 1.0) / (len(sorted_scores) + 1.0)
|
||||
|
||||
|
||||
def _lower_tail_p_value(sorted_scores: tuple[float, ...], score: float) -> float:
|
||||
"""Smoothed empirical probability of a calibration score at most SCORE."""
|
||||
tail_count = bisect.bisect_right(sorted_scores, score)
|
||||
return (tail_count + 1.0) / (len(sorted_scores) + 1.0)
|
||||
|
||||
|
||||
def classify_observations(config: CascadeConfig, observations: tuple[ExpertObservation, ...]) -> CascadeResult:
|
||||
"""Combine one explicit observation from every configured expert."""
|
||||
calibration_by_name = {expert.name: expert for expert in config.experts}
|
||||
observation_by_name = {observation.name: observation for observation in observations}
|
||||
if len(observation_by_name) != len(observations):
|
||||
raise ValueError("observation expert names must be unique")
|
||||
if observation_by_name.keys() != calibration_by_name.keys():
|
||||
missing = sorted(calibration_by_name.keys() - observation_by_name.keys())
|
||||
unknown = sorted(observation_by_name.keys() - calibration_by_name.keys())
|
||||
raise ValueError(f"observations must cover the configured bank; missing={missing}, unknown={unknown}")
|
||||
|
||||
evidence: list[ExpertEvidence] = []
|
||||
for calibration in config.experts:
|
||||
observation = observation_by_name[calibration.name]
|
||||
if not observation.supported:
|
||||
continue
|
||||
if observation.score is None:
|
||||
raise RuntimeError("validated supported observation lost its score")
|
||||
oriented_score = calibration.orient(observation.score)
|
||||
evidence.append(
|
||||
ExpertEvidence(
|
||||
name=calibration.name,
|
||||
score=observation.score,
|
||||
clean_null_p_value=_upper_tail_p_value(calibration.negative_scores, oriented_score),
|
||||
watermarked_p_value=_lower_tail_p_value(calibration.positive_scores, oriented_score),
|
||||
positive_calibration_count=len(calibration.positive_scores),
|
||||
negative_calibration_count=len(calibration.negative_scores),
|
||||
)
|
||||
)
|
||||
|
||||
if not evidence:
|
||||
return CascadeResult(
|
||||
verdict="abstain",
|
||||
reason="unsupported",
|
||||
clean_null_p_value=None,
|
||||
watermarked_p_value=None,
|
||||
supported_expert_count=0,
|
||||
configured_expert_count=len(config.experts),
|
||||
coverage_complete=config.coverage_complete,
|
||||
evidence=(),
|
||||
)
|
||||
|
||||
supported_count = len(evidence)
|
||||
clean_null_p_value = min(1.0, supported_count * min(item.clean_null_p_value for item in evidence))
|
||||
watermarked_p_value = max(item.watermarked_p_value for item in evidence)
|
||||
rejects_clean_null = clean_null_p_value <= config.positive_alpha
|
||||
full_support = supported_count == len(config.experts)
|
||||
rejects_watermarked = config.coverage_complete and full_support and watermarked_p_value <= config.negative_alpha
|
||||
|
||||
if rejects_clean_null and rejects_watermarked:
|
||||
verdict: CascadeVerdict = "abstain"
|
||||
reason = "conflicting_evidence"
|
||||
elif rejects_clean_null:
|
||||
verdict = "detected"
|
||||
reason = "watermarked_hypothesis_supported"
|
||||
elif rejects_watermarked:
|
||||
verdict = "not_detected"
|
||||
reason = "unwatermarked_hypothesis_supported"
|
||||
elif config.coverage_complete and not full_support:
|
||||
verdict = "abstain"
|
||||
reason = "incomplete_support"
|
||||
elif not config.coverage_complete and watermarked_p_value <= config.negative_alpha:
|
||||
verdict = "abstain"
|
||||
reason = "incomplete_coverage"
|
||||
else:
|
||||
verdict = "abstain"
|
||||
reason = "insufficient_evidence"
|
||||
|
||||
return CascadeResult(
|
||||
verdict=verdict,
|
||||
reason=reason,
|
||||
clean_null_p_value=clean_null_p_value,
|
||||
watermarked_p_value=watermarked_p_value,
|
||||
supported_expert_count=supported_count,
|
||||
configured_expert_count=len(config.experts),
|
||||
coverage_complete=config.coverage_complete,
|
||||
evidence=tuple(evidence),
|
||||
)
|
||||
|
||||
|
||||
def _mapping(value: object, label: str) -> dict[str, object]:
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError(f"{label} must be an object")
|
||||
return cast("dict[str, object]", value)
|
||||
|
||||
|
||||
def _sequence(value: object, label: str) -> list[object]:
|
||||
if not isinstance(value, list):
|
||||
raise ValueError(f"{label} must be an array")
|
||||
return cast("list[object]", value)
|
||||
|
||||
|
||||
def _scores(value: object, label: str) -> tuple[float, ...]:
|
||||
scores: list[float] = []
|
||||
for index, score in enumerate(_sequence(value, label)):
|
||||
if isinstance(score, bool) or not isinstance(score, (int, float)):
|
||||
raise ValueError(f"{label}[{index}] must be a number")
|
||||
scores.append(float(score))
|
||||
return tuple(scores)
|
||||
|
||||
|
||||
def _number(value: object, label: str) -> float:
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||
raise ValueError(f"{label} must be a number")
|
||||
return float(value)
|
||||
|
||||
|
||||
def _boolean(value: object, label: str) -> bool:
|
||||
if not isinstance(value, bool):
|
||||
raise ValueError(f"{label} must be a boolean")
|
||||
return value
|
||||
|
||||
|
||||
def _string(value: object, label: str) -> str:
|
||||
if not isinstance(value, str) or not value:
|
||||
raise ValueError(f"{label} must be a non-empty string")
|
||||
return value
|
||||
|
||||
|
||||
def load_config(path: Path) -> CascadeConfig:
|
||||
"""Load a schema-versioned calibration manifest."""
|
||||
payload = _mapping(json.loads(path.read_text(encoding="utf-8")), "calibration manifest")
|
||||
if payload.get("schema_version") != 1:
|
||||
raise ValueError("unsupported calibration manifest schema")
|
||||
experts: list[ExpertCalibration] = []
|
||||
for index, raw_expert in enumerate(_sequence(payload.get("experts"), "experts")):
|
||||
expert = _mapping(raw_expert, f"experts[{index}]")
|
||||
experts.append(
|
||||
ExpertCalibration(
|
||||
name=_string(expert.get("name"), f"experts[{index}].name"),
|
||||
positive_scores=_scores(expert.get("positive_scores"), f"experts[{index}].positive_scores"),
|
||||
negative_scores=_scores(expert.get("negative_scores"), f"experts[{index}].negative_scores"),
|
||||
higher_is_positive=_boolean(
|
||||
expert.get("higher_is_positive", True),
|
||||
f"experts[{index}].higher_is_positive",
|
||||
),
|
||||
)
|
||||
)
|
||||
return CascadeConfig(
|
||||
experts=tuple(experts),
|
||||
positive_alpha=_number(payload.get("positive_alpha"), "positive_alpha"),
|
||||
negative_alpha=_number(payload.get("negative_alpha"), "negative_alpha"),
|
||||
coverage_complete=_boolean(payload.get("coverage_complete", False), "coverage_complete"),
|
||||
scope=_string(payload.get("scope"), "scope"),
|
||||
)
|
||||
|
||||
|
||||
def load_observation_records(path: Path) -> list[tuple[str, tuple[ExpertObservation, ...]]]:
|
||||
"""Load named score records with explicit support for every expert."""
|
||||
payload = _mapping(json.loads(path.read_text(encoding="utf-8")), "observation manifest")
|
||||
if payload.get("schema_version") != 1:
|
||||
raise ValueError("unsupported observation manifest schema")
|
||||
records: list[tuple[str, tuple[ExpertObservation, ...]]] = []
|
||||
for record_index, raw_record in enumerate(_sequence(payload.get("records"), "records")):
|
||||
record = _mapping(raw_record, f"records[{record_index}]")
|
||||
record_id = _string(record.get("id"), f"records[{record_index}].id")
|
||||
observations: list[ExpertObservation] = []
|
||||
for observation_index, raw_observation in enumerate(
|
||||
_sequence(record.get("observations"), f"records[{record_index}].observations")
|
||||
):
|
||||
observation = _mapping(raw_observation, f"records[{record_index}].observations[{observation_index}]")
|
||||
raw_score = observation.get("score")
|
||||
observations.append(
|
||||
ExpertObservation(
|
||||
name=_string(
|
||||
observation.get("name"),
|
||||
f"records[{record_index}].observations[{observation_index}].name",
|
||||
),
|
||||
supported=_boolean(
|
||||
observation.get("supported", False),
|
||||
f"records[{record_index}].observations[{observation_index}].supported",
|
||||
),
|
||||
score=None
|
||||
if raw_score is None
|
||||
else _number(raw_score, f"records[{record_index}].observations[{observation_index}].score"),
|
||||
)
|
||||
)
|
||||
records.append((record_id, tuple(observations)))
|
||||
return records
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.argument("calibration_path", type=click.Path(exists=True, dir_okay=False, path_type=Path))
|
||||
@click.argument("observation_path", 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)
|
||||
def main(calibration_path: Path, observation_path: Path, report_out: Path) -> None:
|
||||
"""Classify precomputed expert scores using CALIBRATION_PATH."""
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
config = load_config(calibration_path)
|
||||
rows: list[dict[str, object]] = []
|
||||
verdict_counts: dict[CascadeVerdict, int] = {"detected": 0, "not_detected": 0, "abstain": 0}
|
||||
for record_id, observations in load_observation_records(observation_path):
|
||||
result = classify_observations(config, observations)
|
||||
verdict_counts[result.verdict] += 1
|
||||
rows.append({"id": record_id, "result": asdict(result)})
|
||||
report_out.parent.mkdir(parents=True, exist_ok=True)
|
||||
report_out.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"schema_version": 1,
|
||||
"scope": config.scope,
|
||||
"calibration_sha256": artifact_sha256(calibration_path),
|
||||
"observation_sha256": artifact_sha256(observation_path),
|
||||
"counts": verdict_counts,
|
||||
"records": rows,
|
||||
},
|
||||
indent=2,
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
log.info("Wrote %d conformal cascade verdicts: %s", len(rows), report_out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,69 @@
|
||||
"""Score images and apply the conservative SynthID expert-bank router."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from dataclasses import asdict
|
||||
from pathlib import Path
|
||||
from typing import TypedDict
|
||||
|
||||
import click
|
||||
from synthid_conformal_cascade import ExpertObservation
|
||||
from synthid_routed_expert_bank import classify_routed
|
||||
from synthid_runtime_expert_scores import ExpertScore, score_path
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class RoutedImage(TypedDict):
|
||||
"""One scored image and its conservative routed result."""
|
||||
|
||||
id: str
|
||||
path: str
|
||||
width: int
|
||||
height: int
|
||||
observations: list[ExpertScore]
|
||||
result: dict[str, object]
|
||||
|
||||
|
||||
def detect_path(path: Path) -> RoutedImage:
|
||||
"""Score and conservatively route one image PATH."""
|
||||
scored = score_path(path)
|
||||
observations = tuple(
|
||||
ExpertObservation(
|
||||
name=observation["name"],
|
||||
supported=observation["supported"],
|
||||
score=observation["score"],
|
||||
)
|
||||
for observation in scored["observations"]
|
||||
)
|
||||
return {**scored, "result": asdict(classify_routed(observations))}
|
||||
|
||||
|
||||
@click.command()
|
||||
@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)
|
||||
def main(images: tuple[Path, ...], report_out: Path) -> None:
|
||||
"""Score and route IMAGES through the conservative pixel expert bank."""
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
records = [detect_path(path) for path in images]
|
||||
detected = sum(record["result"].get("verdict") == "detected" for record in records)
|
||||
report_out.parent.mkdir(parents=True, exist_ok=True)
|
||||
report_out.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"schema_version": 1,
|
||||
"counts": {"detected": detected, "abstain": len(records) - detected},
|
||||
"records": records,
|
||||
},
|
||||
indent=2,
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
log.info("Wrote %d routed image verdicts: %s", len(records), report_out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,151 @@
|
||||
"""Route SynthID pixel experts without an unsafe union of overlapping positives.
|
||||
|
||||
The registered expert owns its measured scale-search range and the large expert
|
||||
owns its separately challenged native large-image range. A fixed-only crossing
|
||||
remains auditable evidence but cannot produce a bank-level detection. The bank
|
||||
never claims absence because encoder-version coverage is incomplete.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
from typing import Literal
|
||||
|
||||
import click
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(PROJECT_ROOT / "src"))
|
||||
|
||||
from synthid_conformal_cascade import ( # noqa: E402
|
||||
ExpertObservation,
|
||||
load_observation_records,
|
||||
)
|
||||
from synthid_research_manifest import artifact_sha256 # noqa: E402
|
||||
|
||||
from remove_ai_watermarks import synthid_detector # noqa: E402
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
RoutedVerdict = Literal["detected", "abstain"]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RoutedBankResult:
|
||||
"""One conservative bank-level decision with every expert score retained."""
|
||||
|
||||
verdict: RoutedVerdict
|
||||
reason: str
|
||||
selected_expert: str | None
|
||||
fixed_supported: bool
|
||||
fixed_score: float | None
|
||||
registered_supported: bool
|
||||
registered_score: float | None
|
||||
large_supported: bool
|
||||
large_score: float | None
|
||||
|
||||
|
||||
def classify_routed(observations: tuple[ExpertObservation, ...]) -> RoutedBankResult:
|
||||
"""Route explicit fixed, registered, and large observations without an OR rule."""
|
||||
by_name = {observation.name: observation for observation in observations}
|
||||
if len(by_name) != len(observations):
|
||||
raise ValueError("observation expert names must be unique")
|
||||
expected = {
|
||||
synthid_detector.DETECTOR_ID,
|
||||
synthid_detector.REGISTERED_DETECTOR_ID,
|
||||
synthid_detector.LARGE_DETECTOR_ID,
|
||||
}
|
||||
if by_name.keys() != expected:
|
||||
missing = sorted(expected - by_name.keys())
|
||||
unknown = sorted(by_name.keys() - expected)
|
||||
raise ValueError(f"observations must cover the routed bank; missing={missing}, unknown={unknown}")
|
||||
|
||||
fixed = by_name[synthid_detector.DETECTOR_ID]
|
||||
registered = by_name[synthid_detector.REGISTERED_DETECTOR_ID]
|
||||
large = by_name[synthid_detector.LARGE_DETECTOR_ID]
|
||||
if large.supported:
|
||||
if large.score is None:
|
||||
raise RuntimeError("validated large observation lost its score")
|
||||
if large.score >= synthid_detector.LARGE_THRESHOLD:
|
||||
verdict: RoutedVerdict = "detected"
|
||||
reason = "large_threshold_crossed"
|
||||
selected_expert: str | None = large.name
|
||||
else:
|
||||
verdict = "abstain"
|
||||
reason = "large_below_threshold"
|
||||
selected_expert = None
|
||||
elif registered.supported:
|
||||
if registered.score is None:
|
||||
raise RuntimeError("validated registered observation lost its score")
|
||||
if registered.score >= synthid_detector.REGISTERED_THRESHOLD:
|
||||
verdict = "detected"
|
||||
reason = "registered_threshold_crossed"
|
||||
selected_expert = registered.name
|
||||
else:
|
||||
verdict = "abstain"
|
||||
reason = (
|
||||
"fixed_only_ambiguous"
|
||||
if fixed.supported and fixed.score is not None and fixed.score >= synthid_detector.TILE_THRESHOLD
|
||||
else "registered_below_threshold"
|
||||
)
|
||||
selected_expert = None
|
||||
elif fixed.supported:
|
||||
verdict = "abstain"
|
||||
reason = (
|
||||
"fixed_only_geometry_uncalibrated"
|
||||
if fixed.score is not None and fixed.score >= synthid_detector.TILE_THRESHOLD
|
||||
else "registered_unsupported"
|
||||
)
|
||||
selected_expert = None
|
||||
else:
|
||||
verdict = "abstain"
|
||||
reason = "unsupported"
|
||||
selected_expert = None
|
||||
|
||||
return RoutedBankResult(
|
||||
verdict=verdict,
|
||||
reason=reason,
|
||||
selected_expert=selected_expert,
|
||||
fixed_supported=fixed.supported,
|
||||
fixed_score=fixed.score,
|
||||
registered_supported=registered.supported,
|
||||
registered_score=registered.score,
|
||||
large_supported=large.supported,
|
||||
large_score=large.score,
|
||||
)
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.argument("observation_path", 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)
|
||||
def main(observation_path: Path, report_out: Path) -> None:
|
||||
"""Route a three-expert pixel score manifest from OBSERVATION_PATH."""
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
counts: dict[RoutedVerdict, int] = {"detected": 0, "abstain": 0}
|
||||
rows: list[dict[str, object]] = []
|
||||
for record_id, observations in load_observation_records(observation_path):
|
||||
result = classify_routed(observations)
|
||||
counts[result.verdict] += 1
|
||||
rows.append({"id": record_id, "result": asdict(result)})
|
||||
report_out.parent.mkdir(parents=True, exist_ok=True)
|
||||
report_out.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"schema_version": 1,
|
||||
"observation_sha256": artifact_sha256(observation_path),
|
||||
"counts": counts,
|
||||
"records": rows,
|
||||
},
|
||||
indent=2,
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
log.info("Wrote %d routed expert-bank verdicts: %s", len(rows), report_out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,119 @@
|
||||
"""Export fixed, large, and scale-registered SynthID observations for images.
|
||||
|
||||
The output is an input manifest for ``synthid_conformal_cascade.py``. All
|
||||
experts consume decoded RGB pixels only. Unsupported geometry is recorded
|
||||
explicitly and never represented by a synthetic score.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, TypedDict
|
||||
|
||||
import click
|
||||
import numpy as np
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from numpy.typing import NDArray
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(PROJECT_ROOT / "src"))
|
||||
|
||||
from synthid_pixel_attack import load_rgb # noqa: E402
|
||||
from synthid_research_manifest import artifact_sha256 # noqa: E402
|
||||
|
||||
from remove_ai_watermarks import synthid_detector # noqa: E402
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
FIXED_EXPERT_NAME = synthid_detector.DETECTOR_ID
|
||||
REGISTERED_EXPERT_NAME = synthid_detector.REGISTERED_DETECTOR_ID
|
||||
LARGE_EXPERT_NAME = synthid_detector.LARGE_DETECTOR_ID
|
||||
|
||||
|
||||
class ExpertScore(TypedDict):
|
||||
"""One JSON-safe runtime expert observation."""
|
||||
|
||||
name: str
|
||||
supported: bool
|
||||
score: float | None
|
||||
|
||||
|
||||
class ScoredImage(TypedDict):
|
||||
"""One hash-pinned image with every runtime expert observation."""
|
||||
|
||||
id: str
|
||||
path: str
|
||||
width: int
|
||||
height: int
|
||||
observations: list[ExpertScore]
|
||||
|
||||
|
||||
def _observation(name: str, supported: bool, score: float | None) -> ExpertScore:
|
||||
return {"name": name, "supported": supported, "score": score}
|
||||
|
||||
|
||||
def score_pixels(pixels: NDArray[np.uint8]) -> list[ExpertScore]:
|
||||
"""Return explicit fixed, registered, and large observations for RGB PIXELS."""
|
||||
if pixels.ndim != 3 or pixels.shape[2] != 3 or pixels.dtype != np.uint8:
|
||||
raise ValueError("pixels must be an RGB uint8 array")
|
||||
bgr_pixels = np.ascontiguousarray(pixels[:, :, ::-1])
|
||||
native = synthid_detector.detect_synthid("decoded-image", image=bgr_pixels)
|
||||
registered = synthid_detector.detect_synthid("decoded-image", image=bgr_pixels, register_scale=True)
|
||||
fixed = _observation(FIXED_EXPERT_NAME, False, None)
|
||||
large = _observation(LARGE_EXPERT_NAME, False, None)
|
||||
native_observation = _observation(native.detector, native.status != "unsupported", native.score)
|
||||
if native.detector == FIXED_EXPERT_NAME:
|
||||
fixed = native_observation
|
||||
elif native.detector == LARGE_EXPERT_NAME:
|
||||
large = native_observation
|
||||
else:
|
||||
raise RuntimeError(f"unexpected default SynthID expert: {native.detector}")
|
||||
return [
|
||||
fixed,
|
||||
_observation(REGISTERED_EXPERT_NAME, registered.status != "unsupported", registered.score),
|
||||
large,
|
||||
]
|
||||
|
||||
|
||||
def score_path(path: Path) -> ScoredImage:
|
||||
"""Decode PATH once and return one hash-pinned observation record."""
|
||||
pixels = load_rgb(path)
|
||||
height, width = pixels.shape[:2]
|
||||
return {
|
||||
"id": artifact_sha256(path),
|
||||
"path": str(path),
|
||||
"width": width,
|
||||
"height": height,
|
||||
"observations": score_pixels(pixels),
|
||||
}
|
||||
|
||||
|
||||
@click.command()
|
||||
@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)
|
||||
def main(images: tuple[Path, ...], report_out: Path) -> None:
|
||||
"""Score IMAGES with every shipped pixel expert."""
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
records = [score_path(path) for path in images]
|
||||
report_out.parent.mkdir(parents=True, exist_ok=True)
|
||||
report_out.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"schema_version": 1,
|
||||
"experts": [FIXED_EXPERT_NAME, REGISTERED_EXPERT_NAME, LARGE_EXPERT_NAME],
|
||||
"records": records,
|
||||
},
|
||||
indent=2,
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
log.info("Wrote %d three-expert score records: %s", len(records), report_out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -25,15 +25,26 @@ log = logging.getLogger(__name__)
|
||||
|
||||
def subtract_tiled_template(pixels: np.ndarray, template: np.ndarray, *, strength: float) -> np.ndarray:
|
||||
"""Subtract STRENGTH times TEMPLATE repeated over PIXELS."""
|
||||
if strength < 0.0:
|
||||
raise ValueError("strength must be nonnegative")
|
||||
if not np.isfinite(strength) or strength < 0.0:
|
||||
raise ValueError("strength must be finite and nonnegative")
|
||||
if pixels.ndim != 3 or pixels.shape[2] != 3:
|
||||
raise ValueError("pixels must have shape (height, width, 3)")
|
||||
if template.ndim != 3 or template.shape[2] != 3:
|
||||
raise ValueError("template must have shape (tile height, tile width, 3)")
|
||||
height, width = pixels.shape[:2]
|
||||
tile_height, tile_width = template.shape[:2]
|
||||
if template.shape[2:] != (3,) or height % tile_height != 0 or width % tile_width != 0:
|
||||
raise ValueError("template does not tile the pixel geometry")
|
||||
repeated = np.tile(template, (height // tile_height, width // tile_width, 1))
|
||||
result = pixels.astype(np.float64) - strength * repeated
|
||||
return np.clip(np.rint(result), 0, 255).astype(np.uint8)
|
||||
if tile_height == 0 or tile_width == 0:
|
||||
raise ValueError("template dimensions must be positive")
|
||||
|
||||
result = np.empty_like(pixels, dtype=np.uint8)
|
||||
repeats_x = (width + tile_width - 1) // tile_width
|
||||
for top in range(0, height, 256):
|
||||
bottom = min(top + 256, height)
|
||||
template_rows = template[np.arange(top, bottom) % tile_height]
|
||||
repeated = np.tile(template_rows, (1, repeats_x, 1))[:, :width]
|
||||
stripe = pixels[top:bottom].astype(np.float64) - strength * repeated
|
||||
result[top:bottom] = np.clip(np.rint(stripe), 0, 255).astype(np.uint8)
|
||||
return result
|
||||
|
||||
|
||||
def parse_positive_floats(value: str, *, option_name: str) -> tuple[float, ...]:
|
||||
|
||||
Reference in New Issue
Block a user