mirror of
https://github.com/wiltodelta/remove-ai-watermarks.git
synced 2026-08-09 23:50:40 +02:00
Keep the pixel forensics in the library, drop the ai-score tooling
`scripts/ai_score.py` and the dataset scanner that fed it are gone: the detector they trained is not something this project runs, and the corpus lived outside the repository anyway. Nothing else referenced them. The scanner's pixel layer was worth keeping, so it moves into the package as `pixel_evidence.py` -- six families of scale-robust statistics (block-DCT histograms and Benford deviation, FFT band energies and CFA peaks, high-pass residual, error level, gradient, colour) measured in a single shared decode. The arithmetic was verified against the scanner over 60 corpus images, families and artifacts alike, before the scanner was removed; that comparison is no longer possible, which is why the tests now pin behavior instead: determinism, empty-not-wrong on images too small for a family, and one failing family not taking the others with it. It has no consumer. Nothing in the package reads it, and the module says so. `artifacts=True` returns the spatial layer -- perceptual hash, 128px thumbnail, coarse ELA/residual/phase maps. Those identify the source image rather than describe it, so they are opt-in and separate: everything else is a scalar or a fixed-length histogram nothing can be reconstructed from. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
b5da0510c9
commit
9a29dcac8a
@@ -1,126 +0,0 @@
|
||||
"""Tests for the standalone structural AI-generation scorer."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
||||
|
||||
import ai_score
|
||||
|
||||
|
||||
def _complete_record() -> dict[str, Any]:
|
||||
return {
|
||||
"noise": {"noise_std": 1.0, "noise_kurtosis": 2.0},
|
||||
"fft": {
|
||||
"cfa_peak": 3.0,
|
||||
"cfa_peaks": [2.5, 3.0],
|
||||
"fft_band_energy": list(range(8)),
|
||||
},
|
||||
"ela": {"ela_mean": 4.0, "ela_p95": 5.0},
|
||||
"gradient": {"laplacian_var": 6.0, "gradient_hist": list(range(10))},
|
||||
"color": {
|
||||
"saturation_mean": 0.5,
|
||||
"value_mean": 0.75,
|
||||
"color_hist_4x4x4": list(range(64)),
|
||||
},
|
||||
"dct": {
|
||||
"benford_mad": 0.1,
|
||||
"dct_ac_hist": [[1] * 21 for _ in range(8)],
|
||||
},
|
||||
"jpeg_forensics": {
|
||||
"subsampling": "4:4:4",
|
||||
"progressive": True,
|
||||
"quant_tables": {
|
||||
"0": list(range(1, 65)),
|
||||
"1": list(range(65, 129)),
|
||||
},
|
||||
"huffman_tables_hex": ["00ff", "abcd12"],
|
||||
"scan_count": 10,
|
||||
"restart_interval": 4,
|
||||
"precision_bits": 8,
|
||||
"adobe_transform": 1,
|
||||
"jfif": {"version": "1.1"},
|
||||
},
|
||||
"pil": {"width": 2000, "height": 1000},
|
||||
"content_format": "jpeg",
|
||||
}
|
||||
|
||||
|
||||
def test_v1_feature_schema_is_fixed_for_sparse_records() -> None:
|
||||
assert len(ai_score.feature_names("v1")) == 97
|
||||
assert len(ai_score.features_of({}, schema="v1")) == 97
|
||||
|
||||
|
||||
def test_v2_feature_schema_includes_existing_forensic_data() -> None:
|
||||
record = _complete_record()
|
||||
|
||||
names = ai_score.feature_names("v2")
|
||||
values = ai_score.features_of(record, schema="v2")
|
||||
by_name = dict(zip(names, values, strict=True))
|
||||
|
||||
assert len(names) == len(values) == 406
|
||||
assert by_name["cfa_peak_0"] == 2.5
|
||||
assert by_name["cfa_peak_1"] == 3.0
|
||||
assert by_name["dct_ac_0_0"] == 1 / 21
|
||||
assert by_name["dct_ac_7_20"] == 1 / 21
|
||||
assert by_name["jpeg_quant_0_0"] == 1.0
|
||||
assert by_name["jpeg_quant_1_63"] == 128.0
|
||||
assert by_name["jpeg_quant_table_count"] == 2.0
|
||||
assert by_name["jpeg_huffman_table_count"] == 2.0
|
||||
assert by_name["jpeg_huffman_total_bytes"] == 5.0
|
||||
assert by_name["jpeg_scan_count"] == 10.0
|
||||
assert by_name["jpeg_jfif_present"] == 1.0
|
||||
assert by_name["format_webp"] == 0.0
|
||||
assert by_name["format_isobmff"] == 0.0
|
||||
assert by_name["format_other"] == 0.0
|
||||
|
||||
|
||||
def test_v2_feature_schema_is_fixed_when_forensics_are_missing() -> None:
|
||||
names = ai_score.feature_names("v2")
|
||||
values = ai_score.features_of({}, schema="v2")
|
||||
|
||||
assert len(names) == len(values) == 406
|
||||
assert np.isnan(values[names.index("dct_ac_0_0")])
|
||||
assert np.isnan(values[names.index("jpeg_quant_0_0")])
|
||||
assert np.isnan(values[names.index("jpeg_scan_count")])
|
||||
|
||||
|
||||
def test_grouped_stratified_split_keeps_hashes_on_one_side() -> None:
|
||||
labels = np.asarray([1, 1, 1, 0, 0, 0, 1, 0])
|
||||
hashes = np.asarray(["a", "a", "b", "c", "c", "d", "e", "f"])
|
||||
|
||||
train, test = ai_score.grouped_stratified_split(labels, hashes, test_size=0.5, random_state=7)
|
||||
|
||||
assert set(hashes[train]).isdisjoint(set(hashes[test]))
|
||||
assert set(labels[train]) == {0, 1}
|
||||
assert set(labels[test]) == {0, 1}
|
||||
assert sorted(np.concatenate([train, test]).tolist()) == list(range(len(labels)))
|
||||
|
||||
|
||||
def test_grouped_stratified_split_rejects_conflicting_labels() -> None:
|
||||
labels = np.asarray([0, 1, 0, 1])
|
||||
hashes = np.asarray(["same", "same", "negative", "positive"])
|
||||
|
||||
with np.testing.assert_raises_regex(ValueError, "conflicting labels"):
|
||||
ai_score.grouped_stratified_split(labels, hashes)
|
||||
|
||||
|
||||
def test_temporal_holdout_excludes_hashes_seen_during_training() -> None:
|
||||
dates = np.asarray(["2026-01-01", "2026-01-01", "2026-01-02", "2026-01-03", "2026-01-04", "2026-01-04"])
|
||||
hashes = np.asarray(["repeated", "old", "middle", "new-a", "repeated", "new-b"])
|
||||
|
||||
train, test, cutoff = ai_score.temporal_holdout_split(dates, hashes, train_fraction=0.5)
|
||||
|
||||
assert cutoff == "2026-01-03"
|
||||
assert set(hashes[train]).isdisjoint(set(hashes[test]))
|
||||
assert set(hashes[test]) == {"new-a", "new-b"}
|
||||
|
||||
|
||||
def test_legacy_model_bundle_defaults_to_v1_schema() -> None:
|
||||
assert ai_score.model_schema({}) == "v1"
|
||||
assert ai_score.model_schema({"feature_schema": "v2"}) == "v2"
|
||||
@@ -0,0 +1,156 @@
|
||||
"""Tests for the experimental pixel-forensics collector.
|
||||
|
||||
It has no consumer, so there is no downstream behavior to pin. What these guard is
|
||||
the part a future consumer would rely on and could not discover from the code: that
|
||||
a family is empty rather than wrong when the image is too small for it, that one
|
||||
failing family does not lose the other five, and that the artifacts which identify
|
||||
the source image stay behind their opt-in.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
from remove_ai_watermarks.pixel_evidence import PixelEvidence, extract_pixel_evidence, is_available
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
FAMILIES = ("dct", "fft", "noise", "ela", "gradient", "color")
|
||||
|
||||
|
||||
def _textured(path: Path, size: tuple[int, int] = (192, 160), *, seed: int = 0) -> Path:
|
||||
"""A textured image. Flat colour would make several families degenerate (zero
|
||||
residual, empty gradient histogram) and hide a real break."""
|
||||
rng = np.random.default_rng(seed)
|
||||
base = rng.integers(0, 255, (size[1], size[0], 3), dtype=np.uint8)
|
||||
ramp = np.linspace(0, 255, size[0], dtype=np.uint8)[None, :, None]
|
||||
Image.fromarray(np.clip(base // 2 + ramp // 2, 0, 255).astype(np.uint8)).save(path)
|
||||
return path
|
||||
|
||||
|
||||
class TestFamilies:
|
||||
def test_every_family_is_measured_on_a_textured_image(self, tmp_path: Path):
|
||||
evidence = extract_pixel_evidence(_textured(tmp_path / "textured.png"))
|
||||
|
||||
assert evidence.decoded
|
||||
for family in FAMILIES:
|
||||
assert getattr(evidence, family), family
|
||||
|
||||
def test_the_same_image_measures_the_same_twice(self, tmp_path: Path):
|
||||
"""Determinism is what makes these comparable across runs and machines; the
|
||||
residual is computed in row chunks, which is exactly the kind of optimization
|
||||
that can perturb the last bits."""
|
||||
path = _textured(tmp_path / "textured.png")
|
||||
|
||||
first, second = extract_pixel_evidence(path), extract_pixel_evidence(path)
|
||||
|
||||
assert dataclasses.asdict(first) == dataclasses.asdict(second)
|
||||
|
||||
def test_source_dimensions_survive_the_downscale(self, tmp_path: Path):
|
||||
"""The recorded size is the SOURCE size, read before the 2048px cap. Recording
|
||||
the analysed size instead would still pass every statistic check, because
|
||||
those run on the downscaled array either way."""
|
||||
path = tmp_path / "oversize.png"
|
||||
Image.fromarray(np.zeros((80, 3000, 3), dtype=np.uint8)).save(path)
|
||||
|
||||
assert extract_pixel_evidence(path).decode == {"width": 3000, "height": 80}
|
||||
|
||||
|
||||
class TestDegenerateInputs:
|
||||
def test_undecodable_file_reports_the_error_and_stays_empty(self, tmp_path: Path):
|
||||
path = tmp_path / "broken.png"
|
||||
path.write_bytes(b"not an image")
|
||||
|
||||
evidence = extract_pixel_evidence(path, artifacts=True)
|
||||
|
||||
assert evidence.decoded is False
|
||||
assert "error" in evidence.decode
|
||||
assert all(getattr(evidence, family) == {} for family in FAMILIES)
|
||||
assert evidence.artifacts == {}
|
||||
|
||||
def test_image_too_small_for_a_family_leaves_it_empty(self, tmp_path: Path):
|
||||
"""8x8 is below the FFT's 32px floor but at the block DCT's. A consumer must
|
||||
not assume a fixed feature width, so the narrow case is pinned."""
|
||||
path = tmp_path / "tiny.png"
|
||||
Image.fromarray(np.arange(8 * 8 * 3, dtype=np.uint8).reshape(8, 8, 3)).save(path)
|
||||
|
||||
evidence = extract_pixel_evidence(path)
|
||||
|
||||
assert evidence.decoded is True
|
||||
assert evidence.fft == {}
|
||||
assert evidence.color != {}
|
||||
|
||||
def test_a_failing_family_does_not_lose_the_others(self, tmp_path: Path, monkeypatch):
|
||||
path = _textured(tmp_path / "textured.png")
|
||||
|
||||
def boom(*args, **kwargs):
|
||||
raise ValueError("family exploded")
|
||||
|
||||
monkeypatch.setattr("remove_ai_watermarks.pixel_evidence.color_features", boom)
|
||||
evidence = extract_pixel_evidence(path)
|
||||
|
||||
assert "error" in evidence.color
|
||||
assert evidence.dct != {}
|
||||
assert evidence.gradient != {}
|
||||
|
||||
|
||||
class TestArtifactsAreOptIn:
|
||||
"""The artifacts identify the source image -- a thumbnail is a picture, a
|
||||
perceptual hash matches one. Everything else is a scalar or a fixed-length
|
||||
histogram. That difference in kind is the reason for the flag, so the flag is
|
||||
what these guard."""
|
||||
|
||||
def test_off_by_default(self, tmp_path: Path):
|
||||
assert extract_pixel_evidence(_textured(tmp_path / "textured.png")).artifacts == {}
|
||||
|
||||
def test_on_request_it_returns_the_spatial_layer(self, tmp_path: Path):
|
||||
evidence = extract_pixel_evidence(_textured(tmp_path / "textured.png"), artifacts=True)
|
||||
|
||||
assert set(evidence.artifacts) == {"phash", "thumbnail_jpeg_b64", "ela_map", "noise_residual", "fft_phase"}
|
||||
assert len(evidence.artifacts["phash"]) == 16
|
||||
|
||||
def test_the_thumbnail_is_a_readable_image_of_the_source(self, tmp_path: Path):
|
||||
"""Stated plainly because it is the privacy claim: this field reconstructs
|
||||
the picture, at 128px."""
|
||||
import base64
|
||||
import io
|
||||
|
||||
evidence = extract_pixel_evidence(_textured(tmp_path / "textured.png"), artifacts=True)
|
||||
|
||||
with Image.open(io.BytesIO(base64.b64decode(evidence.artifacts["thumbnail_jpeg_b64"]))) as thumb:
|
||||
assert max(thumb.size) <= 128
|
||||
|
||||
def test_a_different_image_hashes_differently(self, tmp_path: Path):
|
||||
one = extract_pixel_evidence(_textured(tmp_path / "a.png", seed=1), artifacts=True)
|
||||
two = extract_pixel_evidence(_textured(tmp_path / "b.png", seed=2), artifacts=True)
|
||||
|
||||
assert one.artifacts["phash"] != two.artifacts["phash"]
|
||||
|
||||
def test_the_statistics_carry_no_array_payloads(self, tmp_path: Path):
|
||||
"""Without the flag nothing array-shaped may appear: that is what makes the
|
||||
default set aggregates rather than content."""
|
||||
evidence = extract_pixel_evidence(_textured(tmp_path / "textured.png"))
|
||||
|
||||
for family in FAMILIES:
|
||||
for key, value in getattr(evidence, family).items():
|
||||
assert isinstance(value, (int, float, str, list)), (family, key)
|
||||
if isinstance(value, list):
|
||||
for item in value:
|
||||
assert isinstance(item, (int, float, list)), (family, key)
|
||||
|
||||
|
||||
def test_is_available_reports_the_optional_dependency():
|
||||
assert is_available() is True # the test environment installs the pixels extra
|
||||
|
||||
|
||||
def test_evidence_is_frozen(tmp_path: Path):
|
||||
evidence = extract_pixel_evidence(_textured(tmp_path / "textured.png"))
|
||||
assert isinstance(evidence, PixelEvidence)
|
||||
with pytest.raises(dataclasses.FrozenInstanceError):
|
||||
evidence.decode = {} # type: ignore[misc]
|
||||
Reference in New Issue
Block a user