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()