168 lines
6.4 KiB
Python
168 lines
6.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Build a leak-free YOLO sampling manifest from failed release gates."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def file_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 dataset_validation_source(source_yaml: Path) -> str:
|
|
"""Preserve the source dataset's validation contract verbatim."""
|
|
for raw_line in source_yaml.read_text(encoding="utf-8").splitlines():
|
|
key, separator, value = raw_line.partition(":")
|
|
if separator and key.strip() == "val" and value.strip():
|
|
return value.strip()
|
|
raise ValueError(f"Source dataset YAML has no validation source: {source_yaml}")
|
|
|
|
|
|
def build_sampling(
|
|
*,
|
|
summary: dict[str, Any],
|
|
manifest: dict[str, Any],
|
|
assessment: dict[str, Any],
|
|
positive_repeat: int = 3,
|
|
negative_repeat: int = 4,
|
|
) -> tuple[list[str], dict[str, Any]]:
|
|
if assessment.get("status") != "continue_training_loop":
|
|
raise ValueError("Failure-driven sampling requires a failed assessment")
|
|
if positive_repeat < 1 or negative_repeat < 1:
|
|
raise ValueError("Repeat factors must be positive")
|
|
|
|
samples = {item["sample_slug"]: item for item in manifest["samples"]}
|
|
gates = assessment["gates"]
|
|
evaluation = assessment.get("test") or assessment.get("calibration")
|
|
if not evaluation or "regions" not in evaluation:
|
|
raise ValueError("Assessment has no regional calibration or test evidence")
|
|
regions = evaluation["regions"]
|
|
weak_recall_regions = {
|
|
region
|
|
for region, metrics in regions.items()
|
|
if metrics["f1"] < gates["min_region_f1"]
|
|
or metrics["recall"] < gates["min_region_recall"]
|
|
}
|
|
weak_precision_regions = {
|
|
region
|
|
for region, metrics in regions.items()
|
|
if metrics["precision"] < gates["min_region_precision"]
|
|
}
|
|
background = assessment.get("background")
|
|
background_failed = bool(
|
|
background
|
|
and background["pure_empty_false_positives"]
|
|
> gates["max_pure_empty_false_positives"]
|
|
)
|
|
|
|
image_paths: list[str] = []
|
|
repeat_counts: Counter[str] = Counter()
|
|
selected_samples: set[str] = set()
|
|
protected_samples: set[str] = set()
|
|
for tile in summary["tiles"]:
|
|
sample = samples[tile["sample_slug"]]
|
|
if sample["split"] != "train" or tile["split"] != "train":
|
|
protected_samples.add(tile["sample_slug"])
|
|
continue
|
|
region = sample["region"]
|
|
repeat = 1
|
|
if tile["label_count"] > 0 and region in weak_recall_regions:
|
|
repeat = positive_repeat
|
|
if tile["label_count"] == 0 and (background_failed or region in weak_precision_regions):
|
|
repeat = negative_repeat
|
|
path = str(Path(tile["image_path"]).resolve())
|
|
image_paths.extend([path] * repeat)
|
|
repeat_counts[region] += repeat
|
|
selected_samples.add(tile["sample_slug"])
|
|
|
|
if not image_paths:
|
|
raise ValueError("No train-only tiles selected")
|
|
metadata = {
|
|
"schema_version": 1,
|
|
"status": "ok",
|
|
"strategy": "failed-region-positive-and-hard-negative-repeat",
|
|
"failure_evidence_source": "test" if assessment.get("test") else "calibration",
|
|
"weak_recall_regions": sorted(weak_recall_regions),
|
|
"weak_precision_regions": sorted(weak_precision_regions),
|
|
"background_gate_failed": background_failed,
|
|
"positive_repeat": positive_repeat,
|
|
"negative_repeat": negative_repeat,
|
|
"source_train_tile_count": sum(
|
|
1
|
|
for tile in summary["tiles"]
|
|
if samples[tile["sample_slug"]]["split"] == "train" and tile["split"] == "train"
|
|
),
|
|
"sampled_train_entry_count": len(image_paths),
|
|
"sampled_entries_by_region": dict(sorted(repeat_counts.items())),
|
|
"selected_train_sample_count": len(selected_samples),
|
|
"protected_sample_count": len(protected_samples),
|
|
"protected_samples_in_training": [],
|
|
}
|
|
return image_paths, metadata
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--summary", type=Path, required=True)
|
|
parser.add_argument("--corpus-manifest", type=Path, required=True)
|
|
parser.add_argument("--assessment", type=Path, required=True)
|
|
parser.add_argument("--output-dir", type=Path, required=True)
|
|
parser.add_argument("--positive-repeat", type=int, default=3)
|
|
parser.add_argument("--negative-repeat", type=int, default=4)
|
|
args = parser.parse_args()
|
|
|
|
summary = json.loads(args.summary.read_text(encoding="utf-8"))
|
|
manifest = json.loads(args.corpus_manifest.read_text(encoding="utf-8"))
|
|
assessment = json.loads(args.assessment.read_text(encoding="utf-8"))
|
|
paths, metadata = build_sampling(
|
|
summary=summary,
|
|
manifest=manifest,
|
|
assessment=assessment,
|
|
positive_repeat=args.positive_repeat,
|
|
negative_repeat=args.negative_repeat,
|
|
)
|
|
args.output_dir.mkdir(parents=True, exist_ok=True)
|
|
train_list = args.output_dir / "train-failure-driven.txt"
|
|
train_list.write_text("\n".join(paths) + "\n", encoding="utf-8")
|
|
source_yaml = args.summary.parent / "dataset.yaml"
|
|
val_source = dataset_validation_source(source_yaml)
|
|
dataset_yaml = args.output_dir / "dataset.yaml"
|
|
dataset_yaml.write_text(
|
|
f"path: {args.output_dir}\n"
|
|
f"train: {train_list}\n"
|
|
f"val: {val_source}\n"
|
|
"names:\n 0: building\n",
|
|
encoding="utf-8",
|
|
)
|
|
metadata.update(
|
|
{
|
|
"summary": str(args.summary),
|
|
"summary_sha256": file_sha256(args.summary),
|
|
"corpus_manifest": str(args.corpus_manifest),
|
|
"corpus_manifest_sha256": file_sha256(args.corpus_manifest),
|
|
"assessment": str(args.assessment),
|
|
"assessment_sha256": file_sha256(args.assessment),
|
|
"source_dataset_yaml": str(source_yaml),
|
|
"train_list": str(train_list),
|
|
"dataset_yaml": str(dataset_yaml),
|
|
}
|
|
)
|
|
output = args.output_dir / "failure-driven-sampling.json"
|
|
output.write_text(json.dumps(metadata, indent=2), encoding="utf-8")
|
|
print(json.dumps(metadata, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|