365 lines
14 KiB
Python
365 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
from collections import Counter, defaultdict
|
|
from datetime import datetime, timezone
|
|
from itertools import combinations
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def artifact_record(path: Path, *, include_hash: bool = True) -> dict[str, Any]:
|
|
stat = path.stat()
|
|
return {
|
|
"path": str(path),
|
|
"size_bytes": stat.st_size,
|
|
"modified_at": datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat(),
|
|
"sha256": sha256_file(path) if include_hash else None,
|
|
}
|
|
|
|
|
|
def manifest_summary(path: Path) -> dict[str, Any]:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
samples = payload.get("samples") if isinstance(payload.get("samples"), list) else []
|
|
region_split = Counter(
|
|
f"{sample.get('region')}/{sample.get('split')}" for sample in samples
|
|
)
|
|
contexts = Counter(str(sample.get("context")) for sample in samples)
|
|
roles = Counter(str(sample.get("sample_role")) for sample in samples)
|
|
background_samples = [
|
|
{
|
|
"sample_slug": sample.get("sample_slug"),
|
|
"region": sample.get("region"),
|
|
"reference_feature_count": sample.get("reference_feature_count"),
|
|
"require_empty": sample.get("require_empty"),
|
|
}
|
|
for sample in samples
|
|
if sample.get("split") == "background-test"
|
|
]
|
|
pure_empty = [
|
|
sample
|
|
for sample in background_samples
|
|
if sample.get("require_empty") is True
|
|
and int(sample.get("reference_feature_count") or 0) == 0
|
|
]
|
|
|
|
cross_split_raster_hashes: dict[str, set[str]] = defaultdict(set)
|
|
cross_split_dataset_ids: dict[str, set[str]] = defaultdict(set)
|
|
for sample in samples:
|
|
split = str(sample.get("split"))
|
|
if sample.get("raster_sha256"):
|
|
cross_split_raster_hashes[str(sample["raster_sha256"])].add(split)
|
|
if sample.get("raster_dataset_id"):
|
|
cross_split_dataset_ids[str(sample["raster_dataset_id"])].add(split)
|
|
|
|
spatial = cross_split_spatial_summary(samples)
|
|
perceptual = cross_split_dhash_summary(samples)
|
|
return {
|
|
**artifact_record(path),
|
|
"schema_version": payload.get("schema_version"),
|
|
"dataset_version": payload.get("dataset_version"),
|
|
"immutable": payload.get("immutable"),
|
|
"sample_count": len(samples),
|
|
"region_split_counts": dict(sorted(region_split.items())),
|
|
"context_counts": dict(sorted(contexts.items())),
|
|
"sample_role_counts": dict(sorted(roles.items())),
|
|
"background_test_samples": background_samples,
|
|
"pure_empty_background_count": len(pure_empty),
|
|
"pure_empty_background_by_region": dict(
|
|
sorted(Counter(str(item["region"]) for item in pure_empty).items())
|
|
),
|
|
"exact_cross_split_raster_hash_duplicate_count": sum(
|
|
1 for splits in cross_split_raster_hashes.values() if len(splits) > 1
|
|
),
|
|
"cross_split_raster_dataset_id_duplicate_count": sum(
|
|
1 for splits in cross_split_dataset_ids.values() if len(splits) > 1
|
|
),
|
|
"spatial_independence": spatial,
|
|
"perceptual_duplicate_screen": perceptual,
|
|
}
|
|
|
|
|
|
def cross_split_spatial_summary(samples: list[dict[str, Any]]) -> dict[str, Any]:
|
|
try:
|
|
from pyproj import Transformer
|
|
from shapely.geometry import box
|
|
from shapely.ops import transform
|
|
except ImportError:
|
|
return {"available": False, "reason": "geospatial_dependencies_unavailable"}
|
|
|
|
transformer = Transformer.from_crs(4326, 31370, always_xy=True)
|
|
records: list[tuple[str, str, Any]] = []
|
|
for sample in samples:
|
|
bbox_values = sample.get("bbox_epsg4326")
|
|
if not isinstance(bbox_values, list) or len(bbox_values) != 4:
|
|
continue
|
|
geometry = transform(transformer.transform, box(*[float(value) for value in bbox_values]))
|
|
records.append((str(sample.get("sample_slug")), str(sample.get("split")), geometry))
|
|
|
|
distances: list[tuple[float, str, str, str, str]] = []
|
|
for left, right in combinations(records, 2):
|
|
if left[1] == right[1]:
|
|
continue
|
|
distance = float(left[2].distance(right[2]))
|
|
distances.append((distance, left[0], left[1], right[0], right[1]))
|
|
distances.sort(key=lambda item: item[0])
|
|
return {
|
|
"available": True,
|
|
"crs": "EPSG:31370",
|
|
"cross_split_pair_count": len(distances),
|
|
"minimum_distance_m": distances[0][0] if distances else None,
|
|
"pairs_below_64_m": sum(1 for item in distances if item[0] < 64.0),
|
|
"pairs_below_2000_m": sum(1 for item in distances if item[0] < 2000.0),
|
|
"closest_pairs": [
|
|
{
|
|
"distance_m": item[0],
|
|
"left_sample": item[1],
|
|
"left_split": item[2],
|
|
"right_sample": item[3],
|
|
"right_split": item[4],
|
|
}
|
|
for item in distances[:20]
|
|
],
|
|
"claim_boundary": (
|
|
"AOI bounding-box distance is a screening check only; it does not prove "
|
|
"municipality, flight-strip, building-instance or imagery independence."
|
|
),
|
|
}
|
|
|
|
|
|
def _dhash(path: Path) -> int:
|
|
from PIL import Image
|
|
|
|
with Image.open(path) as image:
|
|
resized = image.convert("L").resize((9, 8))
|
|
pixel_source = (
|
|
resized.get_flattened_data()
|
|
if hasattr(resized, "get_flattened_data")
|
|
else resized.getdata()
|
|
)
|
|
pixels = list(pixel_source)
|
|
value = 0
|
|
for row in range(8):
|
|
offset = row * 9
|
|
for column in range(8):
|
|
value = (value << 1) | int(pixels[offset + column] > pixels[offset + column + 1])
|
|
return value
|
|
|
|
|
|
def cross_split_dhash_summary(samples: list[dict[str, Any]]) -> dict[str, Any]:
|
|
hashes: list[tuple[str, str, int]] = []
|
|
failures: list[dict[str, str]] = []
|
|
for sample in samples:
|
|
raster_path = Path(str(sample.get("raster_path") or ""))
|
|
if not raster_path.is_file():
|
|
failures.append({"sample_slug": str(sample.get("sample_slug")), "reason": "missing_raster"})
|
|
continue
|
|
try:
|
|
hashes.append(
|
|
(
|
|
str(sample.get("sample_slug")),
|
|
str(sample.get("split")),
|
|
_dhash(raster_path),
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
failures.append({"sample_slug": str(sample.get("sample_slug")), "reason": str(exc)})
|
|
|
|
matches: list[dict[str, Any]] = []
|
|
minimum: int | None = None
|
|
compared = 0
|
|
for left, right in combinations(hashes, 2):
|
|
if left[1] == right[1]:
|
|
continue
|
|
compared += 1
|
|
distance = (left[2] ^ right[2]).bit_count()
|
|
minimum = distance if minimum is None else min(minimum, distance)
|
|
if distance <= 4:
|
|
matches.append(
|
|
{
|
|
"left_sample": left[0],
|
|
"left_split": left[1],
|
|
"right_sample": right[0],
|
|
"right_split": right[1],
|
|
"hamming_distance": distance,
|
|
}
|
|
)
|
|
return {
|
|
"algorithm": "64-bit difference hash over 9x8 grayscale resize",
|
|
"screened_raster_count": len(hashes),
|
|
"cross_split_pair_count": compared,
|
|
"minimum_hamming_distance": minimum,
|
|
"pairs_at_or_below_4": matches,
|
|
"failures": failures,
|
|
"claim_boundary": (
|
|
"This bounded perceptual screen is not semantic, instance-level or "
|
|
"flight-strip deduplication and cannot establish split independence."
|
|
),
|
|
}
|
|
|
|
|
|
def corpus_audit_summary(path: Path) -> dict[str, Any]:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
keys = (
|
|
"status",
|
|
"dataset_version",
|
|
"manifest_immutable",
|
|
"sample_count",
|
|
"split_counts",
|
|
"input_feature_count",
|
|
"accepted_feature_count",
|
|
"decision_counts",
|
|
"temporal_unknown_sample_count",
|
|
"spatial_leakage_status",
|
|
"reviewed_sample_count",
|
|
"review_complete",
|
|
"failures",
|
|
)
|
|
return {
|
|
**artifact_record(path),
|
|
**{key: payload.get(key) for key in keys},
|
|
"review_queue_count": len(payload.get("review_queue") or []),
|
|
"claim_boundary": (
|
|
"Automated status and an empty failure list do not substitute for "
|
|
"human label acceptance."
|
|
),
|
|
}
|
|
|
|
|
|
def calibration_summary(path: Path) -> dict[str, Any]:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
sweeps = payload.get("sweeps") if isinstance(payload.get("sweeps"), list) else []
|
|
|
|
def sweep_at(threshold: float) -> dict[str, Any] | None:
|
|
return next(
|
|
(
|
|
{
|
|
"threshold": item.get("threshold"),
|
|
"aggregate": item.get("aggregate"),
|
|
"regions": item.get("regions"),
|
|
"pure_empty_false_positives": item.get("pure_empty_false_positives"),
|
|
}
|
|
for item in sweeps
|
|
if abs(float(item.get("threshold")) - threshold) < 1e-12
|
|
),
|
|
None,
|
|
)
|
|
|
|
return {
|
|
**artifact_record(path),
|
|
"model": payload.get("model"),
|
|
"summary": payload.get("summary"),
|
|
"split": payload.get("split"),
|
|
"match_iou": payload.get("match_iou"),
|
|
"tile_count": payload.get("tile_count"),
|
|
"inference_imgsz": payload.get("inference_imgsz"),
|
|
"sweep_count": len(sweeps),
|
|
"threshold_0_15": sweep_at(0.15),
|
|
"threshold_0_02": sweep_at(0.02),
|
|
"claim_boundary": (
|
|
"This is calibration-split, tile-level bbox evidence; it is not a "
|
|
"protected-test, unique-building or national release result."
|
|
),
|
|
}
|
|
|
|
|
|
def inventory(root: Path) -> dict[str, Any]:
|
|
model_files = sorted(
|
|
path
|
|
for path in (root / "models").glob("*")
|
|
if path.is_file() and path.suffix.lower() in {".pt", ".pth", ".onnx", ".engine", ".ckpt", ".safetensors"}
|
|
)
|
|
training_root = root / "storage" / "training"
|
|
checkpoints = sorted(
|
|
(
|
|
path
|
|
for path in training_root.glob("building-be-*/**/*")
|
|
if path.is_file() and path.suffix.lower() in {".pt", ".pth", ".ckpt", ".onnx", ".engine", ".safetensors"}
|
|
),
|
|
key=lambda path: path.stat().st_mtime,
|
|
reverse=True,
|
|
)
|
|
reports = sorted(
|
|
(
|
|
path
|
|
for path in training_root.glob("building-be-*/**/*.json")
|
|
if path.is_file()
|
|
),
|
|
key=lambda path: path.stat().st_mtime,
|
|
reverse=True,
|
|
)
|
|
manifests = sorted(
|
|
(root / "storage" / "operator-data").glob("building-be-*/operator_samples_manifest.json"),
|
|
key=lambda path: path.stat().st_mtime,
|
|
reverse=True,
|
|
)
|
|
return {
|
|
"model_assets": [artifact_record(path) for path in model_files],
|
|
"training_checkpoint_count": len(checkpoints),
|
|
"training_checkpoint_bytes": sum(path.stat().st_size for path in checkpoints),
|
|
"newest_training_checkpoints": [
|
|
artifact_record(path, include_hash=False) for path in checkpoints[:30]
|
|
],
|
|
"training_json_report_count": len(reports),
|
|
"newest_training_json_reports": [
|
|
artifact_record(path, include_hash=False) for path in reports[:100]
|
|
],
|
|
"operator_manifest_count": len(manifests),
|
|
"operator_manifests": [artifact_record(path) for path in manifests],
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="Collect a bounded, read-only ML/data-lineage snapshot from the deployed GeoIntel volume."
|
|
)
|
|
parser.add_argument("--app-root", default="/app")
|
|
args = parser.parse_args()
|
|
root = Path(args.app_root).resolve()
|
|
|
|
paths = {
|
|
"v56_audit": root / "storage/training/building-be-v56-corpus-audit-r1/belgium-building-corpus-audit.json",
|
|
"v56_manifest": root / "storage/operator-data/building-be-v56-hard-negative-instance-roofs-r1/operator_samples_manifest.json",
|
|
"v58_calibration": root / "storage/training/building-be-v58-v56-clean-pretrained-r1/preview-epoch-015/calibration.json",
|
|
"v62_calibration": root / "storage/training/building-be-v62-v58-flanders-balanced-transfer-r1/preview-epoch-006/calibration-routed.json",
|
|
"v66_manifest": root / "storage/operator-data/building-be-v66-lowrise-temporal-r1/operator_samples_manifest.json",
|
|
}
|
|
missing = [str(path) for path in paths.values() if not path.is_file()]
|
|
if missing:
|
|
raise SystemExit("Required lineage artifacts are missing: " + ", ".join(missing))
|
|
|
|
payload = {
|
|
"schema_version": 1,
|
|
"captured_at": datetime.now(timezone.utc).isoformat(),
|
|
"read_only": True,
|
|
"inventory": inventory(root),
|
|
"v56": {
|
|
"corpus_audit": corpus_audit_summary(paths["v56_audit"]),
|
|
"manifest": manifest_summary(paths["v56_manifest"]),
|
|
},
|
|
"v58": {"calibration": calibration_summary(paths["v58_calibration"])},
|
|
"v62": {"calibration": calibration_summary(paths["v62_calibration"])},
|
|
"v66": {"manifest": manifest_summary(paths["v66_manifest"])},
|
|
"global_claim_boundary": (
|
|
"Inventory and historical calibration evidence do not prove human "
|
|
"label acceptance, strict split independence, calibration, protected-test "
|
|
"performance, national validity or release readiness."
|
|
),
|
|
}
|
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|