import argparse
|
import json
|
import sys
|
import tempfile
|
import unittest
|
from pathlib import Path
|
|
import cv2
|
import numpy as np
|
import rasterio
|
from PIL import Image
|
from rasterio.transform import from_origin
|
|
|
CAPABILITY = Path(__file__).resolve().parents[1]
|
sys.path.insert(0, str(CAPABILITY))
|
|
from generate_validation_inputs import generate # noqa: E402
|
from run_anomaly_detection import ( # noqa: E402
|
FEATURE_NAMES,
|
aligned_rule_scores,
|
axis_positions,
|
extract_features,
|
fit_aligned_rule_model,
|
fit_local_change_model,
|
fit_models,
|
clean_local_change_flag,
|
local_change_scores,
|
run,
|
)
|
|
|
def patterned_image(width: int = 192, height: int = 160) -> np.ndarray:
|
yy, xx = np.mgrid[:height, :width]
|
return np.stack(
|
[
|
(xx * 3 + yy) % 256,
|
(xx + yy * 2 + 40) % 256,
|
(xx * 2 + yy * 3 + 80) % 256,
|
],
|
axis=2,
|
).astype(np.uint8)
|
|
|
class AnomalyDetectionTests(unittest.TestCase):
|
def test_axis_positions_anchor_final_tile(self):
|
self.assertEqual(axis_positions(100, 32, 20), [0, 20, 40, 60, 68])
|
with self.assertRaises(ValueError):
|
axis_positions(20, 32, 16)
|
|
def test_features_explain_blur_and_colour(self):
|
source = patterned_image(64, 64)
|
blurred = cv2.GaussianBlur(source, (31, 31), 0)
|
magenta = np.full((64, 64, 3), (230, 20, 210), dtype=np.uint8)
|
source_features = extract_features(source)
|
blurred_features = extract_features(blurred)
|
magenta_features = extract_features(magenta)
|
laplacian_index = FEATURE_NAMES.index("laplacian_variance")
|
saturation_index = FEATURE_NAMES.index("hsv_mean_s")
|
self.assertLess(blurred_features[laplacian_index], source_features[laplacian_index])
|
self.assertGreater(magenta_features[saturation_index], source_features[saturation_index])
|
|
def test_isolation_model_is_repeatable(self):
|
rng = np.random.default_rng(7)
|
features = rng.normal(size=(80, len(FEATURE_NAMES)))
|
first = fit_models(features, 0.995, 42)
|
second = fit_models(features, 0.995, 42)
|
self.assertAlmostEqual(first["isolation_threshold"], second["isolation_threshold"], places=12)
|
np.testing.assert_allclose(
|
first["reference_isolation_scores"], second["reference_isolation_scores"]
|
)
|
|
def test_aligned_rules_detect_local_change_despite_global_weather_shift(self):
|
rng = np.random.default_rng(12)
|
base = rng.normal(size=(60, len(FEATURE_NAMES)))
|
references = [
|
base + rng.normal(scale=0.08, size=base.shape) + offset
|
for offset in (-0.3, -0.1, 0.0, 0.15, 0.35)
|
]
|
model = fit_aligned_rule_model(references, 0.995)
|
target = base + 0.22
|
target[25:29, FEATURE_NAMES.index("dark_ratio")] += 8.0
|
scores, reasons = aligned_rule_scores(
|
target, model["aligned_center"], model["aligned_scale"]
|
)
|
flagged = set(np.flatnonzero(scores > model["rule_threshold"]))
|
self.assertTrue(set(range(25, 29)).issubset(flagged))
|
self.assertEqual(FEATURE_NAMES[int(reasons[26])], "dark_ratio")
|
|
def test_local_change_channel_detects_small_unseen_structure_without_class_rules(self):
|
base = patterned_image(192, 160)
|
references = [
|
cv2.convertScaleAbs(base, alpha=alpha, beta=beta)
|
for alpha, beta in ((0.9, 5), (1.0, 0), (1.08, 8))
|
]
|
target = cv2.convertScaleAbs(base, alpha=1.02, beta=3)
|
checker = (np.indices((28, 28)).sum(axis=0) % 2 * 255).astype(np.uint8)
|
target[72:100, 104:132] = np.stack([checker, 255 - checker, checker], axis=2)
|
model = fit_local_change_model(references, 0.995)
|
scores, _ = local_change_scores(target, model)
|
detected = clean_local_change_flag(scores)
|
self.assertGreater(float(detected[72:100, 104:132].mean()), 0.75)
|
self.assertLess(float(detected.mean()), 0.1)
|
|
def test_cli_contract_produces_visual_and_structured_outputs(self):
|
with tempfile.TemporaryDirectory() as directory:
|
root = Path(directory)
|
reference = root / "reference"
|
inputs = root / "inputs"
|
reference.mkdir()
|
inputs.mkdir()
|
normal = patterned_image()
|
Image.fromarray(normal).save(reference / "normal.png")
|
Image.fromarray(cv2.convertScaleAbs(normal, alpha=0.94, beta=4)).save(
|
reference / "normal-dim.png"
|
)
|
Image.fromarray(cv2.convertScaleAbs(normal, alpha=1.05, beta=2)).save(
|
reference / "normal-bright.png"
|
)
|
Image.fromarray(normal).save(inputs / "normal.png")
|
anomalous = normal.copy()
|
anomalous[48:96, 64:112] = (5, 5, 5)
|
Image.fromarray(anomalous).save(inputs / "anomalous.png")
|
output = root / "output"
|
metadata = run(
|
argparse.Namespace(
|
reference=reference,
|
input=inputs,
|
output=output,
|
tile_size=32,
|
stride=16,
|
threshold_quantile=0.99,
|
random_state=42,
|
truth_manifest=None,
|
)
|
)
|
self.assertEqual(metadata["input_count"], 2)
|
self.assertGreaterEqual(metadata["reference_tile_count"], 50)
|
for record in metadata["images"]:
|
for key in (
|
"rule_heatmap_file",
|
"isolation_heatmap_file",
|
"overlay_file",
|
"mask_file",
|
"vector_file",
|
"tiles_file",
|
):
|
self.assertTrue((output / record[key]).is_file(), key)
|
stored = json.loads((output / "run_metadata.json").read_text(encoding="utf-8"))
|
self.assertEqual(stored["classification"], "B")
|
with self.assertRaises(ValueError):
|
run(
|
argparse.Namespace(
|
reference=reference,
|
input=inputs,
|
output=output,
|
tile_size=32,
|
stride=16,
|
threshold_quantile=0.99,
|
random_state=42,
|
truth_manifest=None,
|
)
|
)
|
|
def test_projected_geotiff_preserves_crs(self):
|
with tempfile.TemporaryDirectory() as directory:
|
root = Path(directory)
|
source = patterned_image(96, 96)
|
reference = root / "reference.tif"
|
target = root / "target.tif"
|
transform = from_origin(500000, 4000000, 2, 2)
|
for path, values in ((reference, source), (target, source.copy())):
|
if path == target:
|
values[32:64, 32:64] = (250, 10, 230)
|
with rasterio.open(
|
path,
|
"w",
|
driver="GTiff",
|
width=96,
|
height=96,
|
count=3,
|
dtype="uint8",
|
transform=transform,
|
crs="EPSG:3857",
|
) as dst:
|
dst.write(np.moveaxis(values, 2, 0))
|
output = root / "out"
|
metadata = run(
|
argparse.Namespace(
|
reference=reference,
|
input=target,
|
output=output,
|
tile_size=16,
|
stride=8,
|
threshold_quantile=0.99,
|
random_state=42,
|
truth_manifest=None,
|
)
|
)
|
record = metadata["images"][0]
|
self.assertTrue(record["georeferenced"])
|
self.assertEqual(record["crs"], "EPSG:3857")
|
with rasterio.open(output / record["mask_file"]) as src:
|
self.assertEqual(src.crs.to_string(), "EPSG:3857")
|
self.assertEqual(src.transform, transform)
|
|
def test_too_few_reference_tiles_fails(self):
|
rng = np.random.default_rng(1)
|
with self.assertRaisesRegex(ValueError, "At least 50"):
|
fit_models(rng.normal(size=(49, len(FEATURE_NAMES))), 0.995, 42)
|
|
def test_validation_generator_preserves_source_bytes(self):
|
with tempfile.TemporaryDirectory() as directory:
|
root = Path(directory)
|
normal = root / "normal.jpg"
|
difficult = root / "difficult.jpg"
|
Image.fromarray(patterned_image()).save(normal)
|
Image.fromarray(np.flipud(patterned_image())).save(difficult)
|
result = generate(normal, difficult, root / "raw", root / "processed")
|
self.assertEqual(result["normal_source_sha256"], result["reference_copy_sha256"])
|
self.assertEqual(result["difficult_source_sha256"], result["difficult_copy_sha256"])
|
manifest = json.loads(
|
(root / "processed" / "truth-manifest.json").read_text(encoding="utf-8")
|
)
|
self.assertEqual(len(manifest["images"]["07-16-injected.png"]["regions"]), 3)
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
