Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
258 lines
12 KiB
Python
258 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Create a leak-free rotated release portfolio without copying immutable imagery."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
import sys
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from pyproj import Transformer
|
|
from shapely.geometry import box
|
|
from shapely.ops import transform as shapely_transform
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
if str(SCRIPT_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPT_DIR))
|
|
|
|
from training_dataset_eligibility import ( # noqa: E402
|
|
TrainingEligibilityError,
|
|
assert_frozen_manifest_training_eligible,
|
|
)
|
|
from training_release_manifest import ( # noqa: E402
|
|
TrainingReleaseError,
|
|
assert_yolo_summary_bound_to_embedded_training_release,
|
|
)
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def parse_slugs(raw: str) -> set[str]:
|
|
return {value.strip() for value in raw.split(",") if value.strip()}
|
|
|
|
|
|
def validate_dataset_version(value: str) -> str:
|
|
version = value.strip()
|
|
if not version or any(character not in "abcdefghijklmnopqrstuvwxyz0123456789-" for character in version):
|
|
raise ValueError("dataset version must be a lowercase canonical slug")
|
|
return version
|
|
|
|
|
|
def write_json(path: Path, payload: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
|
|
|
|
|
def audit_spatial_leakage(samples: list[dict[str, Any]], buffer_m: float = 64.0) -> dict[str, Any]:
|
|
transformer = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True)
|
|
metric = []
|
|
for sample in samples:
|
|
bounds = sample.get("bbox_epsg4326")
|
|
if not isinstance(bounds, list) or len(bounds) != 4:
|
|
raise SystemExit(f"missing bbox for {sample['sample_slug']}")
|
|
metric.append((sample, shapely_transform(transformer.transform, box(*map(float, bounds)))))
|
|
findings = []
|
|
for index, (left, left_geometry) in enumerate(metric):
|
|
for right, right_geometry in metric[index + 1:]:
|
|
if left["split"] == right["split"]:
|
|
continue
|
|
distance = left_geometry.distance(right_geometry)
|
|
if distance < buffer_m:
|
|
findings.append({
|
|
"left": left["sample_slug"], "left_split": left["split"],
|
|
"right": right["sample_slug"], "right_split": right["split"],
|
|
"distance_m": distance,
|
|
})
|
|
return {"status": "ok" if not findings else "failed", "buffer_m": buffer_m, "findings": findings}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--corpus-manifest", type=Path, required=True)
|
|
parser.add_argument("--source-summary", type=Path, action="append", required=True)
|
|
parser.add_argument("--calibration-samples", required=True)
|
|
parser.add_argument("--test-samples", required=True)
|
|
parser.add_argument("--background-samples", required=True)
|
|
parser.add_argument("--internal-val-samples", required=True)
|
|
parser.add_argument("--version", required=True)
|
|
parser.add_argument("--output-dir", type=Path, required=True)
|
|
parser.add_argument(
|
|
"--fixture-mode",
|
|
action="store_true",
|
|
help="Accept only an explicitly fixture-only source corpus; never use for operational holdout rotation.",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
role_slugs = {
|
|
"calibration": parse_slugs(args.calibration_samples),
|
|
"test": parse_slugs(args.test_samples),
|
|
"background-test": parse_slugs(args.background_samples),
|
|
}
|
|
internal_val_slugs = parse_slugs(args.internal_val_samples)
|
|
all_holdouts = set().union(*role_slugs.values())
|
|
if sum(map(len, role_slugs.values())) != len(all_holdouts) or internal_val_slugs & all_holdouts:
|
|
raise SystemExit("rotated holdout lists overlap")
|
|
|
|
manifest = json.loads(args.corpus_manifest.read_text(encoding="utf-8"))
|
|
try:
|
|
assert_frozen_manifest_training_eligible(
|
|
args.corpus_manifest,
|
|
fixture_mode=args.fixture_mode,
|
|
verify_live=True,
|
|
)
|
|
for path in args.source_summary:
|
|
assert_yolo_summary_bound_to_embedded_training_release(
|
|
summary_path=path,
|
|
corpus_manifest=args.corpus_manifest,
|
|
fixture_mode=args.fixture_mode,
|
|
)
|
|
except (TrainingEligibilityError, TrainingReleaseError) as exc:
|
|
raise SystemExit(str(exc)) from exc
|
|
source_samples = {item["sample_slug"]: item for item in manifest["samples"]}
|
|
missing = sorted(all_holdouts - source_samples.keys())
|
|
if missing:
|
|
raise SystemExit(f"unknown holdout samples: {missing}")
|
|
nonfresh = sorted(slug for slug in all_holdouts if source_samples[slug]["split"] != "train")
|
|
if nonfresh:
|
|
raise SystemExit(f"rotated holdouts must come from the former train split: {nonfresh}")
|
|
invalid_internal = sorted(
|
|
slug for slug in internal_val_slugs
|
|
if slug not in source_samples or source_samples[slug]["split"] != "train"
|
|
)
|
|
if invalid_internal:
|
|
raise SystemExit(f"internal validation samples must come from former train: {invalid_internal}")
|
|
|
|
tiles_by_sample: dict[str, list[dict[str, Any]]] = {}
|
|
summary_hashes = {}
|
|
for path in args.source_summary:
|
|
summary = json.loads(path.read_text(encoding="utf-8"))
|
|
summary_hashes[str(path)] = sha256(path)
|
|
for tile in summary["tiles"]:
|
|
slug = tile["sample_slug"]
|
|
tiles_by_sample.setdefault(slug, []).append({**tile, "source_split": tile["split"]})
|
|
missing_tiles = sorted(source_samples.keys() - tiles_by_sample.keys())
|
|
if missing_tiles:
|
|
raise SystemExit(f"manifest samples have no source tiles: {missing_tiles}")
|
|
|
|
rotated_manifest = json.loads(json.dumps(manifest))
|
|
rotated_manifest["dataset_version"] = validate_dataset_version(args.version)
|
|
rotated_manifest["immutable"] = True
|
|
assignment = {slug: role for role, slugs in role_slugs.items() for slug in slugs}
|
|
for sample in rotated_manifest["samples"]:
|
|
sample["previous_split"] = sample["split"]
|
|
sample["split"] = (
|
|
"val" if sample["sample_slug"] in internal_val_slugs
|
|
else assignment.get(sample["sample_slug"], "train")
|
|
)
|
|
|
|
args.output_dir.mkdir(parents=True, exist_ok=True)
|
|
manifest_path = args.output_dir / "operator_samples_manifest.json"
|
|
write_json(manifest_path, rotated_manifest)
|
|
write_json(
|
|
args.output_dir / "corpus-freeze.json",
|
|
{
|
|
"schema_version": 2,
|
|
"dataset_version": rotated_manifest["dataset_version"],
|
|
"manifest_sha256": sha256(manifest_path),
|
|
"sample_count": len(rotated_manifest["samples"]),
|
|
"immutable": True,
|
|
"training_eligibility_policy": rotated_manifest["training_eligibility"]["policy_version"],
|
|
"fixture_mode": bool(args.fixture_mode),
|
|
},
|
|
)
|
|
leakage = audit_spatial_leakage(rotated_manifest["samples"])
|
|
write_json(args.output_dir / "spatial-leakage-audit.json", leakage)
|
|
if leakage["status"] != "ok":
|
|
raise SystemExit(f"rotated spatial leakage audit failed: {leakage['findings']}")
|
|
source_pairs = args.corpus_manifest.parent / "pairs"
|
|
for sample in rotated_manifest["samples"]:
|
|
source_audit = source_pairs / sample["sample_slug"] / "label-audit.json"
|
|
target_audit = args.output_dir / "pairs" / sample["sample_slug"] / "label-audit.json"
|
|
target_audit.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(source_audit, target_audit)
|
|
|
|
train_tiles: list[dict[str, Any]] = []
|
|
internal_val_tiles: list[dict[str, Any]] = []
|
|
role_counts: dict[str, Counter[str]] = {}
|
|
for role in ("calibration", "test", "background-test"):
|
|
tiles = []
|
|
role_counts[role] = Counter()
|
|
for slug in sorted(role_slugs[role]):
|
|
role_counts[role][source_samples[slug]["region"]] += 1
|
|
tiles.extend({**tile, "split": "val"} for tile in tiles_by_sample[slug])
|
|
role_dir = args.output_dir / role
|
|
summary_path = role_dir / "yolo_tile_dataset_summary.json"
|
|
write_json(summary_path, {"schema_version": 1, "output_dir": str(role_dir), "tiles": tiles})
|
|
|
|
for slug, tiles in tiles_by_sample.items():
|
|
if slug in all_holdouts:
|
|
continue
|
|
if slug in internal_val_slugs:
|
|
internal_val_tiles.extend({**tile, "split": "val"} for tile in tiles)
|
|
continue
|
|
for tile in tiles:
|
|
# Preserve the original train corpus's internal checkpoint split only.
|
|
if source_samples[slug]["split"] == "train" and tile["source_split"] == "val":
|
|
internal_val_tiles.append({**tile, "split": "val"})
|
|
else:
|
|
train_tiles.append({**tile, "split": "train"})
|
|
train_dir = args.output_dir / "train"
|
|
train_dir.mkdir(parents=True, exist_ok=True)
|
|
train_list = train_dir / "train.txt"
|
|
val_list = train_dir / "val.txt"
|
|
train_list.write_text("\n".join(tile["image_path"] for tile in train_tiles) + "\n", encoding="utf-8")
|
|
val_list.write_text("\n".join(tile["image_path"] for tile in internal_val_tiles) + "\n", encoding="utf-8")
|
|
(train_dir / "dataset.yaml").write_text(
|
|
f"path: {train_dir}\ntrain: {train_list}\nval: {val_list}\nnames:\n 0: building\n",
|
|
encoding="utf-8",
|
|
)
|
|
write_json(
|
|
train_dir / "yolo_tile_dataset_summary.json",
|
|
{"schema_version": 1, "output_dir": str(train_dir), "tiles": train_tiles + internal_val_tiles},
|
|
)
|
|
evidence = {
|
|
"schema_version": 1, "status": "ok", "strategy": "fresh-former-train-holdout-rotation",
|
|
"source_manifest": str(args.corpus_manifest), "source_manifest_sha256": sha256(args.corpus_manifest),
|
|
"dataset_version": rotated_manifest["dataset_version"],
|
|
"source_summary_sha256": summary_hashes, "rotated_manifest": str(manifest_path),
|
|
"rotated_manifest_sha256": sha256(manifest_path),
|
|
"spatial_leakage_status": leakage["status"],
|
|
"assignments": {role: sorted(slugs) for role, slugs in role_slugs.items()},
|
|
"internal_val_samples": sorted(internal_val_slugs),
|
|
"holdout_sample_counts_by_region": {
|
|
role: dict(sorted(counts.items())) for role, counts in role_counts.items()
|
|
},
|
|
"train_tile_count": len(train_tiles), "internal_val_tile_count": len(internal_val_tiles),
|
|
"protected_samples_in_training": sorted(all_holdouts.intersection({tile["sample_slug"] for tile in train_tiles + internal_val_tiles})),
|
|
"fit_samples_in_internal_validation": sorted(
|
|
{tile["sample_slug"] for tile in train_tiles} & {tile["sample_slug"] for tile in internal_val_tiles}
|
|
),
|
|
"fixture_mode": bool(args.fixture_mode),
|
|
"training_eligible": False,
|
|
"training_eligibility_reason": (
|
|
"A rotated split needs a new human review, label contract validation and immutable training release."
|
|
),
|
|
}
|
|
if evidence["protected_samples_in_training"]:
|
|
raise SystemExit("protected samples leaked into rotated training lists")
|
|
if evidence["fit_samples_in_internal_validation"] or not internal_val_tiles:
|
|
raise SystemExit("internal validation must be non-empty and AOI-disjoint from fit training")
|
|
write_json(args.output_dir / "holdout-rotation-evidence.json", evidence)
|
|
print(json.dumps(evidence, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|