1167 lines
54 KiB
Python
1167 lines
54 KiB
Python
#!/usr/bin/env python3
|
|
"""Seal and verify immutable YOLO training-release inputs.
|
|
|
|
Training data is deliberately treated as a release artifact rather than a
|
|
mutable ``dataset.yaml`` file. The sidecars written by this module bind the
|
|
exact YAML bytes, every train/validation image and matching label, the frozen
|
|
corpus manifest and the human-review decision evidence. Verification re-hashes
|
|
all of those inputs immediately before a training or resume command can run.
|
|
|
|
The module avoids a YAML dependency on purpose. GeoIntel-generated YOLO YAML
|
|
files use a small, auditable top-level scalar subset (``path``, ``train`` and
|
|
``val``); unsupported YAML shapes fail closed instead of being guessed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from collections.abc import Mapping
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
from training_dataset_eligibility import (
|
|
TrainingEligibilityError,
|
|
assert_frozen_manifest_training_eligible,
|
|
frozen_manifest_training_eligibility_failures,
|
|
)
|
|
|
|
|
|
TRAINING_RELEASE_CONTRACT_VERSION = "geointel-training-release/v1"
|
|
TRAINING_RELEASE_FREEZE_VERSION = "geointel-training-release-freeze/v1"
|
|
TRAINING_ASSET_MANIFEST_VERSION = "geointel-yolo-training-assets/v1"
|
|
TRAINING_LABEL_CONTRACT_MANIFEST_VERSION = "geointel-yolo-training-label-contracts/v1"
|
|
_SHA256_RE = re.compile(r"^[0-9a-f]{64}$", re.IGNORECASE)
|
|
_IMAGE_SUFFIXES = {".bmp", ".jpeg", ".jpg", ".png", ".tif", ".tiff", ".webp"}
|
|
|
|
|
|
class TrainingReleaseError(ValueError):
|
|
"""Raised when a training release is absent, mutable or incomplete."""
|
|
|
|
|
|
def file_sha256(path: Path) -> str:
|
|
"""Return the SHA-256 of one regular file."""
|
|
|
|
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 _canonical_path(path: Path) -> str:
|
|
return str(path.expanduser().resolve(strict=False))
|
|
|
|
|
|
def _canonical_json_bytes(value: Any) -> bytes:
|
|
return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode(
|
|
"utf-8"
|
|
)
|
|
|
|
|
|
def _write_immutable_json(path: Path, payload: Mapping[str, Any]) -> None:
|
|
"""Create an immutable sidecar, or accept an identical idempotent rerun."""
|
|
|
|
encoded = json.dumps(dict(payload), ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
|
if path.exists():
|
|
if path.read_text(encoding="utf-8") != encoded:
|
|
raise TrainingReleaseError(
|
|
f"Immutable training-release artifact already exists with different content: {path}"
|
|
)
|
|
return
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(encoded, encoding="utf-8")
|
|
temporary.replace(path)
|
|
|
|
|
|
def training_release_paths(train_yaml: Path) -> dict[str, Path]:
|
|
"""Return the only accepted sidecar locations for a dataset YAML."""
|
|
|
|
yaml_path = train_yaml.resolve(strict=False)
|
|
return {
|
|
"release_manifest": yaml_path.with_name(yaml_path.name + ".geointel-training-release.json"),
|
|
"release_freeze": yaml_path.with_name(yaml_path.name + ".geointel-training-release-freeze.json"),
|
|
"asset_manifest": yaml_path.with_name(yaml_path.name + ".geointel-training-assets.json"),
|
|
"label_contract_manifest": yaml_path.with_name(yaml_path.name + ".geointel-training-label-contracts.json"),
|
|
}
|
|
|
|
|
|
def _yaml_scalar(raw: str, *, field: str, yaml_path: Path) -> str:
|
|
value = raw.strip()
|
|
if not value:
|
|
raise TrainingReleaseError(f"YOLO YAML field {field!r} is empty: {yaml_path}")
|
|
if value.startswith(("[", "{", "|", ">", "&", "*", "!")):
|
|
raise TrainingReleaseError(
|
|
f"YOLO YAML field {field!r} uses an unsupported non-scalar form: {yaml_path}"
|
|
)
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
|
|
value = value[1:-1]
|
|
if " #" in value:
|
|
value = value.split(" #", 1)[0].rstrip()
|
|
if not value:
|
|
raise TrainingReleaseError(f"YOLO YAML field {field!r} is empty: {yaml_path}")
|
|
return value
|
|
|
|
|
|
def parse_yolo_dataset_yaml(train_yaml: Path) -> dict[str, str]:
|
|
"""Parse the intentionally small, generated YOLO YAML contract.
|
|
|
|
The parser accepts only direct top-level scalar ``path``, ``train`` and
|
|
``val`` keys. It never tries to infer a list, anchor or nested YAML shape.
|
|
Such a file must be converted to an explicit generated release first.
|
|
"""
|
|
|
|
try:
|
|
raw_text = train_yaml.read_text(encoding="utf-8")
|
|
except OSError as exc:
|
|
raise TrainingReleaseError(f"Training YAML is unreadable: {train_yaml}") from exc
|
|
fields: dict[str, str] = {}
|
|
for line_number, raw_line in enumerate(raw_text.splitlines(), start=1):
|
|
if not raw_line.strip() or raw_line.lstrip().startswith("#"):
|
|
continue
|
|
if raw_line[:1].isspace():
|
|
continue
|
|
key, separator, value = raw_line.partition(":")
|
|
if not separator:
|
|
continue
|
|
field = key.strip()
|
|
if field not in {"path", "train", "val"}:
|
|
continue
|
|
if field in fields:
|
|
raise TrainingReleaseError(
|
|
f"YOLO YAML field {field!r} occurs more than once at line {line_number}: {train_yaml}"
|
|
)
|
|
fields[field] = _yaml_scalar(value, field=field, yaml_path=train_yaml)
|
|
missing = sorted({"train", "val"} - fields.keys())
|
|
if missing:
|
|
raise TrainingReleaseError(
|
|
f"YOLO YAML is missing required field(s) {', '.join(missing)}: {train_yaml}"
|
|
)
|
|
return fields
|
|
|
|
|
|
def _resolve_dataset_reference(raw: str, *, yaml_path: Path, dataset_root: Path) -> Path:
|
|
path = Path(raw).expanduser()
|
|
if path.is_absolute():
|
|
return path.resolve(strict=False)
|
|
return (dataset_root / path).resolve(strict=False)
|
|
|
|
|
|
def _resolve_list_image(raw: str, *, list_path: Path, dataset_root: Path) -> Path:
|
|
candidate = Path(raw).expanduser()
|
|
if candidate.is_absolute():
|
|
return candidate.resolve(strict=False)
|
|
list_relative = (list_path.parent / candidate).resolve(strict=False)
|
|
root_relative = (dataset_root / candidate).resolve(strict=False)
|
|
if list_relative.is_file():
|
|
return list_relative
|
|
if root_relative.is_file():
|
|
return root_relative
|
|
return list_relative
|
|
|
|
|
|
def _label_path_for_image(image_path: Path) -> Path:
|
|
parts = list(image_path.parts)
|
|
image_index = next(
|
|
(index for index in range(len(parts) - 1, -1, -1) if parts[index].lower() == "images"),
|
|
None,
|
|
)
|
|
if image_index is None:
|
|
return image_path.with_suffix(".txt")
|
|
parts[image_index] = "labels"
|
|
return Path(*parts).with_suffix(".txt")
|
|
|
|
|
|
def _image_paths_for_split(
|
|
*,
|
|
split: str,
|
|
source_path: Path,
|
|
dataset_root: Path,
|
|
) -> tuple[list[Path], dict[str, Any]]:
|
|
if source_path.is_dir():
|
|
images = sorted(
|
|
(candidate.resolve(strict=False) for candidate in source_path.rglob("*") if candidate.is_file() and candidate.suffix.lower() in _IMAGE_SUFFIXES),
|
|
key=lambda candidate: _canonical_path(candidate),
|
|
)
|
|
source = {"kind": "directory", "path": _canonical_path(source_path)}
|
|
elif source_path.is_file() and source_path.suffix.lower() == ".txt":
|
|
image_lines = [line.strip() for line in source_path.read_text(encoding="utf-8").splitlines() if line.strip()]
|
|
images = [
|
|
_resolve_list_image(line, list_path=source_path, dataset_root=dataset_root)
|
|
for line in image_lines
|
|
]
|
|
source = {
|
|
"kind": "list",
|
|
"path": _canonical_path(source_path),
|
|
"sha256": file_sha256(source_path),
|
|
"entry_count": len(images),
|
|
}
|
|
elif source_path.is_file() and source_path.suffix.lower() in _IMAGE_SUFFIXES:
|
|
images = [source_path.resolve(strict=False)]
|
|
source = {"kind": "image", "path": _canonical_path(source_path)}
|
|
else:
|
|
raise TrainingReleaseError(
|
|
f"YOLO YAML {split!r} source is neither an image directory, a list nor an image: {source_path}"
|
|
)
|
|
if not images:
|
|
raise TrainingReleaseError(f"YOLO YAML {split!r} source contains no images: {source_path}")
|
|
return images, source
|
|
|
|
|
|
def build_yolo_asset_manifest(train_yaml: Path) -> dict[str, Any]:
|
|
"""Return deterministic hashes for every exact train/validation pair."""
|
|
|
|
yaml_path = train_yaml.resolve(strict=False)
|
|
if not yaml_path.is_file():
|
|
raise TrainingReleaseError(f"Training YAML does not exist: {yaml_path}")
|
|
yaml_fields = parse_yolo_dataset_yaml(yaml_path)
|
|
root_field = yaml_fields.get("path")
|
|
dataset_root = (
|
|
_resolve_dataset_reference(root_field, yaml_path=yaml_path, dataset_root=yaml_path.parent)
|
|
if root_field is not None
|
|
else yaml_path.parent.resolve(strict=False)
|
|
)
|
|
entries: list[dict[str, Any]] = []
|
|
split_sources: dict[str, dict[str, Any]] = {}
|
|
counts: dict[str, int] = {}
|
|
for split in ("train", "val"):
|
|
source_path = _resolve_dataset_reference(
|
|
yaml_fields[split],
|
|
yaml_path=yaml_path,
|
|
dataset_root=dataset_root,
|
|
)
|
|
images, source = _image_paths_for_split(
|
|
split=split,
|
|
source_path=source_path,
|
|
dataset_root=dataset_root,
|
|
)
|
|
split_sources[split] = source
|
|
counts[split] = len(images)
|
|
for entry_index, image_path in enumerate(images):
|
|
if not image_path.is_file() or image_path.suffix.lower() not in _IMAGE_SUFFIXES:
|
|
raise TrainingReleaseError(
|
|
f"YOLO {split!r} entry is not a supported readable image: {image_path}"
|
|
)
|
|
label_path = _label_path_for_image(image_path)
|
|
if not label_path.is_file():
|
|
raise TrainingReleaseError(
|
|
f"YOLO {split!r} image has no matching label file: {image_path} -> {label_path}"
|
|
)
|
|
entries.append(
|
|
{
|
|
"split": split,
|
|
"entry_index": entry_index,
|
|
"image_path": _canonical_path(image_path),
|
|
"image_sha256": file_sha256(image_path),
|
|
"label_path": _canonical_path(label_path),
|
|
"label_sha256": file_sha256(label_path),
|
|
}
|
|
)
|
|
payload: dict[str, Any] = {
|
|
"schema_version": 1,
|
|
"contract_version": TRAINING_ASSET_MANIFEST_VERSION,
|
|
"dataset_yaml_path": _canonical_path(yaml_path),
|
|
"dataset_yaml_sha256": file_sha256(yaml_path),
|
|
"dataset_root": _canonical_path(dataset_root),
|
|
"split_sources": split_sources,
|
|
"counts": counts,
|
|
"entries": entries,
|
|
}
|
|
payload["asset_manifest_sha256"] = hashlib.sha256(_canonical_json_bytes(payload)).hexdigest()
|
|
return payload
|
|
|
|
|
|
def _label_contract_dependencies() -> tuple[Any, Any, Any, Any]:
|
|
"""Import backend-only contract helpers only when a release is sealed.
|
|
|
|
The script is invoked from both the repository and the container where the
|
|
backend package can live at a different relative path. A lazy import
|
|
keeps the small YAML/hash inspection functions usable without opening a
|
|
backend dependency, while operational releases cannot bypass validation.
|
|
"""
|
|
|
|
import sys
|
|
|
|
repo_root = Path(__file__).resolve().parents[1]
|
|
app_root = repo_root if (repo_root / "app").is_dir() else repo_root / "backend"
|
|
if str(app_root) not in sys.path:
|
|
sys.path.insert(0, str(app_root))
|
|
try:
|
|
from app.services.data_contract_validation import (
|
|
LineageEvidence,
|
|
TransformationEvidence,
|
|
build_label_validation_input,
|
|
validate_registered_asset,
|
|
)
|
|
except Exception as exc: # pragma: no cover - deployment dependency failure
|
|
raise TrainingReleaseError("Versioned label-contract validator is unavailable") from exc
|
|
return LineageEvidence, TransformationEvidence, build_label_validation_input, validate_registered_asset
|
|
|
|
|
|
def _parse_yolo_label_records(label_path: Path) -> tuple[bytes, tuple[dict[str, Any], ...]]:
|
|
"""Parse one YOLO text label exactly; blank content is a candidate negative."""
|
|
|
|
try:
|
|
raw = label_path.read_bytes()
|
|
text = raw.decode("utf-8")
|
|
except (OSError, UnicodeDecodeError) as exc:
|
|
raise TrainingReleaseError(f"YOLO label is unreadable UTF-8: {label_path}") from exc
|
|
records: list[dict[str, Any]] = []
|
|
for line_number, line in enumerate(text.splitlines(), start=1):
|
|
stripped = line.strip()
|
|
if not stripped:
|
|
continue
|
|
fields = stripped.split()
|
|
if len(fields) != 5:
|
|
raise TrainingReleaseError(
|
|
f"YOLO label line must contain class and four coordinates: {label_path}:{line_number}"
|
|
)
|
|
try:
|
|
class_id = int(fields[0])
|
|
except ValueError as exc:
|
|
raise TrainingReleaseError(f"YOLO class id is invalid: {label_path}:{line_number}") from exc
|
|
try:
|
|
coordinates = [float(value) for value in fields[1:]]
|
|
except ValueError as exc:
|
|
raise TrainingReleaseError(f"YOLO coordinate is invalid: {label_path}:{line_number}") from exc
|
|
records.append(
|
|
{
|
|
"class_id": class_id,
|
|
"x_center": coordinates[0],
|
|
"y_center": coordinates[1],
|
|
"width": coordinates[2],
|
|
"height": coordinates[3],
|
|
}
|
|
)
|
|
return raw, tuple(records)
|
|
|
|
|
|
def _sample_for_label_asset(
|
|
*,
|
|
label_path: Path,
|
|
split: str,
|
|
samples: list[Mapping[str, Any]],
|
|
) -> Mapping[str, Any]:
|
|
"""Bind every exported label path to exactly one frozen corpus sample."""
|
|
|
|
stem = label_path.stem
|
|
candidates: list[Mapping[str, Any]] = []
|
|
for sample in samples:
|
|
slug = str(sample.get("sample_slug") or "").strip()
|
|
if slug and (stem == slug or stem.startswith(f"{slug}_")):
|
|
candidates.append(sample)
|
|
if len(candidates) != 1:
|
|
raise TrainingReleaseError(
|
|
f"YOLO label cannot be bound to exactly one frozen corpus sample: {label_path}"
|
|
)
|
|
sample = candidates[0]
|
|
sample_split = str(sample.get("split") or "").strip().lower()
|
|
if sample_split != split:
|
|
raise TrainingReleaseError(
|
|
f"YOLO label split does not match its frozen corpus sample: {label_path} ({split!r} != {sample_split!r})"
|
|
)
|
|
return sample
|
|
|
|
|
|
def _accepted_review_metadata(sample_slug: str, review: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""Return sample-specific accepted review evidence for a pure negative."""
|
|
|
|
if review.get("status") != "accepted" or review.get("review_complete") is not True:
|
|
raise TrainingReleaseError(
|
|
f"Pure-background label requires accepted human review: {sample_slug}"
|
|
)
|
|
evidence = review.get("evidence")
|
|
if not isinstance(evidence, Mapping):
|
|
raise TrainingReleaseError(f"Pure-background label review evidence is missing: {sample_slug}")
|
|
accepted = evidence.get("accepted_sample_slugs")
|
|
if not isinstance(accepted, list) or sample_slug not in {str(value) for value in accepted}:
|
|
raise TrainingReleaseError(f"Pure-background label sample was not accepted by review: {sample_slug}")
|
|
reviewers = evidence.get("reviewer_ids")
|
|
timestamps = evidence.get("reviewed_at_by_sample")
|
|
artifacts = evidence.get("reviewed_artifact_sha256_by_sample")
|
|
if (
|
|
not isinstance(reviewers, list)
|
|
or not reviewers
|
|
or not isinstance(timestamps, Mapping)
|
|
or not isinstance(artifacts, Mapping)
|
|
):
|
|
raise TrainingReleaseError(f"Pure-background label review evidence is incomplete: {sample_slug}")
|
|
reviewer = next((str(value).strip() for value in reviewers if isinstance(value, str) and value.strip()), "")
|
|
reviewed_at = timestamps.get(sample_slug)
|
|
artifact_sha256 = artifacts.get(sample_slug)
|
|
if not reviewer or not isinstance(reviewed_at, str) or not _is_sha256(artifact_sha256):
|
|
raise TrainingReleaseError(f"Pure-background label review evidence is invalid: {sample_slug}")
|
|
return {
|
|
"review_decision": "accepted",
|
|
"reviewer_id": reviewer,
|
|
"reviewed_at": reviewed_at,
|
|
"review_artifact_sha256": artifact_sha256,
|
|
}
|
|
|
|
|
|
def _source_evidence_for_label(sample: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
"""Extract immutable parent identities from the frozen pair decision."""
|
|
|
|
pair = sample.get("training_eligibility")
|
|
if not isinstance(pair, Mapping) or pair.get("eligible") is not True:
|
|
raise TrainingReleaseError(f"Label sample has no eligible frozen training pair: {sample.get('sample_slug')}")
|
|
parent_evidence: dict[str, dict[str, Any]] = {}
|
|
for role, field_name in (("raster", "raster_dataset_id"), ("reference", "reference_dataset_id")):
|
|
decision = pair.get(role)
|
|
if not isinstance(decision, Mapping) or decision.get("eligible") is not True:
|
|
raise TrainingReleaseError(f"Label sample has no eligible {role} parent: {sample.get('sample_slug')}")
|
|
evidence = decision.get("evidence")
|
|
if not isinstance(evidence, Mapping):
|
|
raise TrainingReleaseError(f"Label sample has no {role} provenance evidence: {sample.get('sample_slug')}")
|
|
expected_id = str(sample.get(field_name) or "")
|
|
if not expected_id or str(evidence.get("dataset_id") or "") != expected_id:
|
|
raise TrainingReleaseError(f"Label sample {role} parent identity is not bound: {sample.get('sample_slug')}")
|
|
checksum = evidence.get("checksum_sha256")
|
|
if not _is_sha256(checksum):
|
|
raise TrainingReleaseError(f"Label sample {role} parent checksum is invalid: {sample.get('sample_slug')}")
|
|
parent_evidence[role] = dict(evidence)
|
|
return parent_evidence["raster"], parent_evidence["reference"]
|
|
|
|
|
|
def build_training_label_contract_manifest(
|
|
*,
|
|
corpus_manifest_path: Path,
|
|
asset_manifest: Mapping[str, Any],
|
|
review: Mapping[str, Any],
|
|
fixture_mode: bool,
|
|
) -> dict[str, Any]:
|
|
"""Validate every release label under a versioned contract before training.
|
|
|
|
Empty label text is never inferred as a negative. It receives the
|
|
`pure_background` mode only after its exact frozen sample, source parents,
|
|
split and accepted review evidence have been bound into this sidecar.
|
|
"""
|
|
|
|
corpus = _read_json(corpus_manifest_path, label="Frozen corpus manifest")
|
|
samples_raw = corpus.get("samples")
|
|
if not isinstance(samples_raw, list) or not samples_raw:
|
|
raise TrainingReleaseError("Frozen corpus manifest contains no samples for label-contract validation")
|
|
samples = [sample for sample in samples_raw if isinstance(sample, Mapping)]
|
|
if len(samples) != len(samples_raw):
|
|
raise TrainingReleaseError("Frozen corpus manifest contains an invalid sample for label-contract validation")
|
|
entries_raw = asset_manifest.get("entries")
|
|
if not isinstance(entries_raw, list) or not entries_raw:
|
|
raise TrainingReleaseError("Training asset manifest contains no labels for label-contract validation")
|
|
asset_manifest_sha256 = asset_manifest.get("asset_manifest_sha256")
|
|
if not _is_sha256(asset_manifest_sha256):
|
|
raise TrainingReleaseError("Training asset manifest checksum is invalid for label-contract validation")
|
|
corpus_sha256 = file_sha256(corpus_manifest_path)
|
|
LineageEvidence, TransformationEvidence, build_label_validation_input, validate_registered_asset = _label_contract_dependencies()
|
|
entries: list[dict[str, Any]] = []
|
|
for asset_entry in entries_raw:
|
|
if not isinstance(asset_entry, Mapping):
|
|
raise TrainingReleaseError("Training asset manifest contains an invalid label entry")
|
|
split = str(asset_entry.get("split") or "").strip().lower()
|
|
label_path_raw = asset_entry.get("label_path")
|
|
image_sha256 = asset_entry.get("image_sha256")
|
|
if split not in {"train", "val"} or not isinstance(label_path_raw, str) or not _is_sha256(image_sha256):
|
|
raise TrainingReleaseError("Training asset manifest label entry is incomplete")
|
|
label_path = Path(label_path_raw).expanduser().resolve(strict=False)
|
|
raw, records = _parse_yolo_label_records(label_path)
|
|
sample = _sample_for_label_asset(label_path=label_path, split=split, samples=samples)
|
|
sample_slug = str(sample.get("sample_slug") or "").strip()
|
|
raster, reference = _source_evidence_for_label(sample)
|
|
source_registry_id = reference.get("source_registry_id")
|
|
source_snapshot_id = reference.get("source_snapshot_id")
|
|
if not isinstance(source_registry_id, str) or not source_registry_id or not isinstance(source_snapshot_id, str) or not source_snapshot_id:
|
|
raise TrainingReleaseError(f"Label source registry/snapshot is not bound: {sample_slug}")
|
|
label_mode = "objects" if records else "pure_background"
|
|
metadata: dict[str, Any] = {
|
|
"image_checksum_sha256": image_sha256,
|
|
"class_ontology_version": "geointel-building-yolo/v1",
|
|
"source_corpus_manifest_sha256": corpus_sha256,
|
|
"label_mode": label_mode,
|
|
}
|
|
if label_mode == "pure_background":
|
|
metadata.update(
|
|
{
|
|
"sample_slug": sample_slug,
|
|
"split": split,
|
|
"raster_dataset_id": sample.get("raster_dataset_id"),
|
|
"reference_dataset_id": sample.get("reference_dataset_id"),
|
|
**_accepted_review_metadata(sample_slug, review),
|
|
}
|
|
)
|
|
label_input = build_label_validation_input(
|
|
asset_id=f"yolo-label:{_canonical_path(label_path)}",
|
|
label_records=records,
|
|
label_mode=label_mode,
|
|
checksum_sha256=file_sha256(label_path),
|
|
computed_checksum_sha256=file_sha256(label_path),
|
|
content=raw,
|
|
source_registry_id=source_registry_id,
|
|
source_snapshot_id=source_snapshot_id,
|
|
imported_at=datetime.now(timezone.utc),
|
|
metadata=metadata,
|
|
temporal_unknown_reason="YOLO label inherits the governed raster/reference temporal assessment.",
|
|
source_version_unknown_reason="The immutable corpus manifest is the label release version.",
|
|
lineage=LineageEvidence(
|
|
upstream_asset_ids=(str(sample.get("raster_dataset_id")), str(sample.get("reference_dataset_id"))),
|
|
upstream_checksums_sha256=(str(raster["checksum_sha256"]), str(reference["checksum_sha256"])),
|
|
transformations=(
|
|
TransformationEvidence(
|
|
name="geointel-yolo-label-export",
|
|
version="1.0.0",
|
|
checksum_sha256=asset_manifest_sha256,
|
|
),
|
|
),
|
|
),
|
|
)
|
|
report = validate_registered_asset(label_input)
|
|
if report.validation_status.value != "passed":
|
|
issue_codes = ",".join(sorted(issue.code for issue in report.issues))
|
|
raise TrainingReleaseError(f"YOLO label contract failed for {label_path}: {issue_codes}")
|
|
entries.append(
|
|
{
|
|
"split": split,
|
|
"entry_index": asset_entry.get("entry_index"),
|
|
"image_path": asset_entry.get("image_path"),
|
|
"image_sha256": image_sha256,
|
|
"label_path": _canonical_path(label_path),
|
|
"label_sha256": file_sha256(label_path),
|
|
"sample_slug": sample_slug,
|
|
"label_mode": label_mode,
|
|
"data_contract": {
|
|
"key": report.data_contract_key,
|
|
"version": report.data_contract_version,
|
|
"fingerprint_sha256": report.contract_fingerprint_sha256,
|
|
"validation_status": report.validation_status.value,
|
|
},
|
|
}
|
|
)
|
|
payload: dict[str, Any] = {
|
|
"schema_version": 1,
|
|
"contract_version": TRAINING_LABEL_CONTRACT_MANIFEST_VERSION,
|
|
"corpus_manifest_path": _canonical_path(corpus_manifest_path),
|
|
"corpus_manifest_sha256": corpus_sha256,
|
|
"asset_manifest_content_sha256": asset_manifest_sha256,
|
|
"fixture_mode": fixture_mode,
|
|
"entries": entries,
|
|
"counts": {
|
|
"total": len(entries),
|
|
"objects": sum(entry["label_mode"] == "objects" for entry in entries),
|
|
"pure_background": sum(entry["label_mode"] == "pure_background" for entry in entries),
|
|
},
|
|
}
|
|
payload["label_contract_manifest_sha256"] = hashlib.sha256(_canonical_json_bytes(payload)).hexdigest()
|
|
return payload
|
|
|
|
|
|
def _is_sha256(value: Any) -> bool:
|
|
return isinstance(value, str) and _SHA256_RE.fullmatch(value) is not None
|
|
|
|
|
|
def _parse_utc_timestamp(value: Any) -> bool:
|
|
if not isinstance(value, str) or not value.strip():
|
|
return False
|
|
try:
|
|
parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return False
|
|
return parsed.tzinfo is not None
|
|
|
|
|
|
def human_review_audit_failures(
|
|
audit: Mapping[str, Any],
|
|
*,
|
|
corpus_manifest_path: Path | None = None,
|
|
corpus_manifest_sha256: str | None = None,
|
|
) -> list[str]:
|
|
"""Validate the persisted, accepted human-review evidence required for training."""
|
|
|
|
failures: list[str] = []
|
|
if audit.get("status") != "ok":
|
|
failures.append("review_audit_status_not_ok")
|
|
if audit.get("review_complete") is not True:
|
|
failures.append("review_complete_not_true")
|
|
if audit.get("manifest_immutable") is not True:
|
|
failures.append("review_audit_manifest_not_immutable")
|
|
if audit.get("spatial_leakage_status") != "ok":
|
|
failures.append("review_audit_spatial_leakage_not_ok")
|
|
if corpus_manifest_path is not None and audit.get("corpus_manifest_path") != _canonical_path(corpus_manifest_path):
|
|
failures.append("review_audit_corpus_manifest_path_mismatch")
|
|
if corpus_manifest_sha256 is not None and audit.get("corpus_manifest_sha256") != corpus_manifest_sha256:
|
|
failures.append("review_audit_corpus_manifest_checksum_mismatch")
|
|
evidence = audit.get("human_review_evidence")
|
|
if not isinstance(evidence, Mapping):
|
|
return sorted(set(failures + ["accepted_human_review_evidence_missing"]))
|
|
if not isinstance(evidence.get("review_decisions_path"), str) or not evidence["review_decisions_path"]:
|
|
failures.append("review_decisions_path_missing")
|
|
if not _is_sha256(evidence.get("review_decisions_sha256")):
|
|
failures.append("review_decisions_checksum_invalid")
|
|
elif isinstance(evidence.get("review_decisions_path"), str) and evidence["review_decisions_path"]:
|
|
decision_path = Path(evidence["review_decisions_path"]).expanduser().resolve(strict=False)
|
|
if not decision_path.is_file() or evidence["review_decisions_sha256"] != file_sha256(decision_path):
|
|
failures.append("review_decisions_artifact_checksum_mismatch")
|
|
required_count = evidence.get("required_sample_count")
|
|
accepted_count = evidence.get("accepted_sample_count")
|
|
if not isinstance(required_count, int) or required_count < 1:
|
|
failures.append("review_required_sample_count_invalid")
|
|
if not isinstance(accepted_count, int) or not isinstance(required_count, int) or accepted_count != required_count:
|
|
failures.append("review_accepted_sample_count_incomplete")
|
|
accepted_slugs = evidence.get("accepted_sample_slugs")
|
|
if not isinstance(accepted_slugs, list) or not isinstance(required_count, int) or len(accepted_slugs) != required_count:
|
|
failures.append("review_accepted_sample_evidence_incomplete")
|
|
reviewers = evidence.get("reviewer_ids")
|
|
if not isinstance(reviewers, list) or not reviewers or not all(isinstance(value, str) and value for value in reviewers):
|
|
failures.append("reviewer_identity_evidence_missing")
|
|
timestamps = evidence.get("reviewed_at_by_sample")
|
|
if not isinstance(timestamps, Mapping) or not isinstance(accepted_slugs, list):
|
|
failures.append("review_timestamp_evidence_missing")
|
|
elif any(not _parse_utc_timestamp(timestamps.get(str(slug))) for slug in accepted_slugs):
|
|
failures.append("review_timestamp_evidence_invalid")
|
|
artifacts = evidence.get("reviewed_artifact_sha256_by_sample")
|
|
artifact_paths = evidence.get("reviewed_artifact_path_by_sample")
|
|
if (
|
|
not isinstance(artifacts, Mapping)
|
|
or not isinstance(artifact_paths, Mapping)
|
|
or not isinstance(accepted_slugs, list)
|
|
):
|
|
failures.append("reviewed_artifact_evidence_missing")
|
|
else:
|
|
for slug in accepted_slugs:
|
|
artifact_hash = artifacts.get(str(slug))
|
|
artifact_path_raw = artifact_paths.get(str(slug))
|
|
if not _is_sha256(artifact_hash) or not isinstance(artifact_path_raw, str) or not artifact_path_raw:
|
|
failures.append("reviewed_artifact_evidence_invalid")
|
|
continue
|
|
artifact_path = Path(artifact_path_raw).expanduser().resolve(strict=False)
|
|
if not artifact_path.is_file() or artifact_hash != file_sha256(artifact_path):
|
|
failures.append("reviewed_artifact_checksum_mismatch")
|
|
return sorted(set(failures))
|
|
|
|
|
|
def _read_json(path: Path, *, label: str) -> dict[str, Any]:
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise TrainingReleaseError(f"{label} is unreadable: {path}") from exc
|
|
if not isinstance(payload, dict):
|
|
raise TrainingReleaseError(f"{label} must be a JSON object: {path}")
|
|
return payload
|
|
|
|
|
|
def _review_evidence_for_release(
|
|
*,
|
|
review_audit_path: Path | None,
|
|
corpus_manifest_path: Path,
|
|
corpus_manifest_sha256: str,
|
|
fixture_mode: bool,
|
|
) -> dict[str, Any]:
|
|
if fixture_mode:
|
|
return {
|
|
"status": "fixture_relaxed",
|
|
"fixture_only": True,
|
|
"review_complete": False,
|
|
"reason": "fixture-only frozen corpus explicitly permits test-only review relaxation",
|
|
}
|
|
if review_audit_path is None:
|
|
raise TrainingReleaseError(
|
|
"Operational training release requires --review-audit with accepted human-review evidence"
|
|
)
|
|
audit_path = review_audit_path.resolve(strict=False)
|
|
audit = _read_json(audit_path, label="Human-review audit")
|
|
failures = human_review_audit_failures(
|
|
audit,
|
|
corpus_manifest_path=corpus_manifest_path,
|
|
corpus_manifest_sha256=corpus_manifest_sha256,
|
|
)
|
|
if failures:
|
|
raise TrainingReleaseError(
|
|
"Human-review audit is not eligible for operational training: " + ", ".join(failures)
|
|
)
|
|
evidence = audit["human_review_evidence"]
|
|
assert isinstance(evidence, Mapping)
|
|
return {
|
|
"status": "accepted",
|
|
"fixture_only": False,
|
|
"review_complete": True,
|
|
"audit_path": _canonical_path(audit_path),
|
|
"audit_sha256": file_sha256(audit_path),
|
|
"evidence": dict(evidence),
|
|
}
|
|
|
|
|
|
def create_training_release_manifest(
|
|
*,
|
|
train_yaml: Path,
|
|
corpus_manifest: Path,
|
|
review_audit_path: Path | None = None,
|
|
fixture_mode: bool = False,
|
|
live_session_factory: Callable[[], Any] | None = None,
|
|
live_dataset_model: type[Any] | None = None,
|
|
) -> dict[str, Path]:
|
|
"""Seal one exact training YAML against a frozen corpus and review decision.
|
|
|
|
A normal release cannot be created without a passed human-review audit.
|
|
Fixture mode is accepted only after the corpus eligibility gate proves that
|
|
the source manifest *and* its freeze sidecar are explicitly fixture-only.
|
|
"""
|
|
|
|
yaml_path = train_yaml.resolve(strict=False)
|
|
corpus_path = corpus_manifest.resolve(strict=False)
|
|
try:
|
|
assert_frozen_manifest_training_eligible(
|
|
corpus_path,
|
|
fixture_mode=fixture_mode,
|
|
verify_live=True,
|
|
session_factory=live_session_factory,
|
|
dataset_model=live_dataset_model,
|
|
)
|
|
except TrainingEligibilityError as exc:
|
|
raise TrainingReleaseError(str(exc)) from exc
|
|
if not yaml_path.is_file():
|
|
raise TrainingReleaseError(f"Training YAML does not exist: {yaml_path}")
|
|
paths = training_release_paths(yaml_path)
|
|
asset_payload = build_yolo_asset_manifest(yaml_path)
|
|
corpus_freeze_path = corpus_path.parent / "corpus-freeze.json"
|
|
if not corpus_freeze_path.is_file():
|
|
raise TrainingReleaseError(f"Frozen corpus sidecar is missing: {corpus_freeze_path}")
|
|
corpus_hash = file_sha256(corpus_path)
|
|
review = _review_evidence_for_release(
|
|
review_audit_path=review_audit_path,
|
|
corpus_manifest_path=corpus_path,
|
|
corpus_manifest_sha256=corpus_hash,
|
|
fixture_mode=fixture_mode,
|
|
)
|
|
label_contract_payload = build_training_label_contract_manifest(
|
|
corpus_manifest_path=corpus_path,
|
|
asset_manifest=asset_payload,
|
|
review=review,
|
|
fixture_mode=fixture_mode,
|
|
)
|
|
_write_immutable_json(paths["asset_manifest"], asset_payload)
|
|
_write_immutable_json(paths["label_contract_manifest"], label_contract_payload)
|
|
release: dict[str, Any] = {
|
|
"schema_version": 1,
|
|
"contract_version": TRAINING_RELEASE_CONTRACT_VERSION,
|
|
"immutable": True,
|
|
"status": "eligible",
|
|
"fixture_mode": fixture_mode,
|
|
"dataset_yaml": {
|
|
"path": _canonical_path(yaml_path),
|
|
"sha256": file_sha256(yaml_path),
|
|
},
|
|
"corpus": {
|
|
"manifest_path": _canonical_path(corpus_path),
|
|
"manifest_sha256": corpus_hash,
|
|
"freeze_path": _canonical_path(corpus_freeze_path),
|
|
"freeze_sha256": file_sha256(corpus_freeze_path),
|
|
},
|
|
"asset_manifest": {
|
|
"path": _canonical_path(paths["asset_manifest"]),
|
|
"sha256": file_sha256(paths["asset_manifest"]),
|
|
"content_sha256": asset_payload["asset_manifest_sha256"],
|
|
"train_entry_count": asset_payload["counts"]["train"],
|
|
"val_entry_count": asset_payload["counts"]["val"],
|
|
},
|
|
"label_contract_manifest": {
|
|
"path": _canonical_path(paths["label_contract_manifest"]),
|
|
"sha256": file_sha256(paths["label_contract_manifest"]),
|
|
"content_sha256": label_contract_payload["label_contract_manifest_sha256"],
|
|
"entry_count": label_contract_payload["counts"]["total"],
|
|
"pure_background_entry_count": label_contract_payload["counts"]["pure_background"],
|
|
},
|
|
"human_review": review,
|
|
}
|
|
_write_immutable_json(paths["release_manifest"], release)
|
|
freeze = {
|
|
"schema_version": 1,
|
|
"contract_version": TRAINING_RELEASE_FREEZE_VERSION,
|
|
"immutable": True,
|
|
"release_manifest_sha256": file_sha256(paths["release_manifest"]),
|
|
"fixture_mode": fixture_mode,
|
|
"dataset_yaml_sha256": release["dataset_yaml"]["sha256"],
|
|
"corpus_manifest_sha256": corpus_hash,
|
|
"asset_manifest_sha256": release["asset_manifest"]["sha256"],
|
|
"label_contract_manifest_sha256": release["label_contract_manifest"]["sha256"],
|
|
}
|
|
_write_immutable_json(paths["release_freeze"], freeze)
|
|
return paths
|
|
|
|
|
|
def training_release_failures(
|
|
*,
|
|
train_yaml: Path,
|
|
corpus_manifest: Path | None = None,
|
|
fixture_mode: bool = False,
|
|
live_session_factory: Callable[[], Any] | None = None,
|
|
live_dataset_model: type[Any] | None = None,
|
|
) -> list[str]:
|
|
"""Return every deterministic reason a YAML cannot enter training now."""
|
|
|
|
yaml_path = train_yaml.resolve(strict=False)
|
|
paths = training_release_paths(yaml_path)
|
|
failures: list[str] = []
|
|
if not paths["release_manifest"].is_file():
|
|
return ["training_release_manifest_missing"]
|
|
if not paths["release_freeze"].is_file():
|
|
return ["training_release_freeze_missing"]
|
|
try:
|
|
release = _read_json(paths["release_manifest"], label="Training release manifest")
|
|
freeze = _read_json(paths["release_freeze"], label="Training release freeze")
|
|
except TrainingReleaseError as exc:
|
|
return [str(exc)]
|
|
if release.get("contract_version") != TRAINING_RELEASE_CONTRACT_VERSION:
|
|
failures.append("training_release_contract_invalid")
|
|
if release.get("immutable") is not True:
|
|
failures.append("training_release_not_immutable")
|
|
if release.get("status") != "eligible":
|
|
failures.append("training_release_not_eligible")
|
|
if release.get("fixture_mode") is not fixture_mode:
|
|
failures.append("training_release_fixture_mode_mismatch")
|
|
if freeze.get("contract_version") != TRAINING_RELEASE_FREEZE_VERSION:
|
|
failures.append("training_release_freeze_contract_invalid")
|
|
if freeze.get("immutable") is not True:
|
|
failures.append("training_release_freeze_not_immutable")
|
|
if freeze.get("fixture_mode") is not fixture_mode:
|
|
failures.append("training_release_freeze_fixture_mode_mismatch")
|
|
if freeze.get("release_manifest_sha256") != file_sha256(paths["release_manifest"]):
|
|
failures.append("training_release_manifest_checksum_mismatch")
|
|
|
|
yaml_evidence = release.get("dataset_yaml")
|
|
if not isinstance(yaml_evidence, Mapping):
|
|
failures.append("training_release_yaml_evidence_missing")
|
|
else:
|
|
if yaml_evidence.get("path") != _canonical_path(yaml_path):
|
|
failures.append("training_release_yaml_path_mismatch")
|
|
if not yaml_path.is_file() or yaml_evidence.get("sha256") != file_sha256(yaml_path):
|
|
failures.append("training_release_yaml_checksum_mismatch")
|
|
if freeze.get("dataset_yaml_sha256") != yaml_evidence.get("sha256"):
|
|
failures.append("training_release_freeze_yaml_binding_mismatch")
|
|
|
|
corpus = release.get("corpus")
|
|
corpus_path: Path | None = None
|
|
if not isinstance(corpus, Mapping):
|
|
failures.append("training_release_corpus_evidence_missing")
|
|
else:
|
|
declared = corpus.get("manifest_path")
|
|
if not isinstance(declared, str) or not declared:
|
|
failures.append("training_release_corpus_path_missing")
|
|
else:
|
|
corpus_path = Path(declared).expanduser().resolve(strict=False)
|
|
if corpus_manifest is not None and corpus_path != corpus_manifest.resolve(strict=False):
|
|
failures.append("training_release_corpus_path_mismatch")
|
|
if not corpus_path.is_file():
|
|
failures.append("training_release_corpus_missing")
|
|
else:
|
|
corpus_hash = file_sha256(corpus_path)
|
|
if corpus.get("manifest_sha256") != corpus_hash:
|
|
failures.append("training_release_corpus_checksum_mismatch")
|
|
expected_freeze = corpus_path.parent / "corpus-freeze.json"
|
|
if corpus.get("freeze_path") != _canonical_path(expected_freeze):
|
|
failures.append("training_release_corpus_freeze_path_mismatch")
|
|
if not expected_freeze.is_file() or corpus.get("freeze_sha256") != file_sha256(expected_freeze):
|
|
failures.append("training_release_corpus_freeze_checksum_mismatch")
|
|
if freeze.get("corpus_manifest_sha256") != corpus_hash:
|
|
failures.append("training_release_freeze_corpus_binding_mismatch")
|
|
failures.extend(
|
|
frozen_manifest_training_eligibility_failures(
|
|
corpus_path,
|
|
fixture_mode=fixture_mode,
|
|
verify_live=True,
|
|
session_factory=live_session_factory,
|
|
dataset_model=live_dataset_model,
|
|
)
|
|
)
|
|
|
|
asset_evidence = release.get("asset_manifest")
|
|
if not isinstance(asset_evidence, Mapping):
|
|
failures.append("training_release_asset_manifest_evidence_missing")
|
|
else:
|
|
if asset_evidence.get("path") != _canonical_path(paths["asset_manifest"]):
|
|
failures.append("training_release_asset_manifest_path_mismatch")
|
|
if not paths["asset_manifest"].is_file():
|
|
failures.append("training_release_asset_manifest_missing")
|
|
else:
|
|
stored_assets = _read_json(paths["asset_manifest"], label="Training asset manifest")
|
|
asset_file_hash = file_sha256(paths["asset_manifest"])
|
|
if asset_evidence.get("sha256") != asset_file_hash:
|
|
failures.append("training_release_asset_manifest_checksum_mismatch")
|
|
if freeze.get("asset_manifest_sha256") != asset_file_hash:
|
|
failures.append("training_release_freeze_asset_binding_mismatch")
|
|
try:
|
|
expected_assets = build_yolo_asset_manifest(yaml_path)
|
|
except TrainingReleaseError as exc:
|
|
failures.append(f"training_release_assets_unreadable:{exc}")
|
|
else:
|
|
if stored_assets != expected_assets:
|
|
failures.append("training_release_asset_manifest_content_mismatch")
|
|
if asset_evidence.get("content_sha256") != expected_assets["asset_manifest_sha256"]:
|
|
failures.append("training_release_asset_content_checksum_mismatch")
|
|
|
|
review = release.get("human_review")
|
|
if not isinstance(review, Mapping):
|
|
failures.append("training_release_human_review_missing")
|
|
elif fixture_mode:
|
|
if review.get("status") != "fixture_relaxed" or review.get("fixture_only") is not True:
|
|
failures.append("training_release_fixture_review_relaxation_invalid")
|
|
else:
|
|
if review.get("status") != "accepted" or review.get("review_complete") is not True:
|
|
failures.append("training_release_human_review_not_accepted")
|
|
audit_path_raw = review.get("audit_path")
|
|
if not isinstance(audit_path_raw, str) or not audit_path_raw:
|
|
failures.append("training_release_review_audit_path_missing")
|
|
else:
|
|
audit_path = Path(audit_path_raw).expanduser().resolve(strict=False)
|
|
if not audit_path.is_file() or review.get("audit_sha256") != file_sha256(audit_path):
|
|
failures.append("training_release_review_audit_checksum_mismatch")
|
|
else:
|
|
audit = _read_json(audit_path, label="Human-review audit")
|
|
corpus_hash = file_sha256(corpus_path) if corpus_path and corpus_path.is_file() else None
|
|
failures.extend(
|
|
human_review_audit_failures(
|
|
audit,
|
|
corpus_manifest_path=corpus_path,
|
|
corpus_manifest_sha256=corpus_hash,
|
|
)
|
|
)
|
|
if review.get("evidence") != audit.get("human_review_evidence"):
|
|
failures.append("training_release_human_review_evidence_mismatch")
|
|
|
|
label_contract_evidence = release.get("label_contract_manifest")
|
|
if not isinstance(label_contract_evidence, Mapping):
|
|
failures.append("training_release_label_contract_manifest_evidence_missing")
|
|
else:
|
|
if label_contract_evidence.get("path") != _canonical_path(paths["label_contract_manifest"]):
|
|
failures.append("training_release_label_contract_manifest_path_mismatch")
|
|
if not paths["label_contract_manifest"].is_file():
|
|
failures.append("training_release_label_contract_manifest_missing")
|
|
else:
|
|
label_file_hash = file_sha256(paths["label_contract_manifest"])
|
|
if label_contract_evidence.get("sha256") != label_file_hash:
|
|
failures.append("training_release_label_contract_manifest_checksum_mismatch")
|
|
if freeze.get("label_contract_manifest_sha256") != label_file_hash:
|
|
failures.append("training_release_freeze_label_contract_binding_mismatch")
|
|
try:
|
|
stored_label_contracts = _read_json(
|
|
paths["label_contract_manifest"],
|
|
label="Training label-contract manifest",
|
|
)
|
|
except TrainingReleaseError as exc:
|
|
failures.append(f"training_release_label_contract_manifest_unreadable:{exc}")
|
|
else:
|
|
if (
|
|
corpus_path is None
|
|
or not corpus_path.is_file()
|
|
or not isinstance(review, Mapping)
|
|
or not yaml_path.is_file()
|
|
):
|
|
failures.append("training_release_label_contract_inputs_unavailable")
|
|
else:
|
|
try:
|
|
expected_label_contracts = build_training_label_contract_manifest(
|
|
corpus_manifest_path=corpus_path,
|
|
asset_manifest=build_yolo_asset_manifest(yaml_path),
|
|
review=review,
|
|
fixture_mode=fixture_mode,
|
|
)
|
|
except TrainingReleaseError as exc:
|
|
failures.append(f"training_release_label_contract_invalid:{exc}")
|
|
else:
|
|
if stored_label_contracts != expected_label_contracts:
|
|
failures.append("training_release_label_contract_manifest_content_mismatch")
|
|
if (
|
|
label_contract_evidence.get("content_sha256")
|
|
!= expected_label_contracts.get("label_contract_manifest_sha256")
|
|
):
|
|
failures.append("training_release_label_contract_content_checksum_mismatch")
|
|
counts = expected_label_contracts.get("counts")
|
|
if not isinstance(counts, Mapping):
|
|
failures.append("training_release_label_contract_counts_invalid")
|
|
else:
|
|
if label_contract_evidence.get("entry_count") != counts.get("total"):
|
|
failures.append("training_release_label_contract_entry_count_mismatch")
|
|
if label_contract_evidence.get("pure_background_entry_count") != counts.get("pure_background"):
|
|
failures.append("training_release_label_contract_background_count_mismatch")
|
|
return sorted(set(failures))
|
|
|
|
|
|
def assert_training_release_eligible(
|
|
*,
|
|
train_yaml: Path,
|
|
corpus_manifest: Path | None = None,
|
|
fixture_mode: bool = False,
|
|
live_session_factory: Callable[[], Any] | None = None,
|
|
live_dataset_model: type[Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Fail closed unless the exact YAML/corpus/assets/review release is valid."""
|
|
|
|
failures = training_release_failures(
|
|
train_yaml=train_yaml,
|
|
corpus_manifest=corpus_manifest,
|
|
fixture_mode=fixture_mode,
|
|
live_session_factory=live_session_factory,
|
|
live_dataset_model=live_dataset_model,
|
|
)
|
|
if failures:
|
|
raise TrainingReleaseError(
|
|
"Training release is not eligible: " + ", ".join(failures)
|
|
)
|
|
return _read_json(training_release_paths(train_yaml)["release_manifest"], label="Training release manifest")
|
|
|
|
|
|
def assert_yolo_summary_bound_to_training_release(
|
|
*,
|
|
summary_path: Path,
|
|
train_yaml: Path,
|
|
corpus_manifest: Path | None = None,
|
|
fixture_mode: bool = False,
|
|
live_session_factory: Callable[[], Any] | None = None,
|
|
live_dataset_model: type[Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Verify that a tile summary is an exact view of an eligible release.
|
|
|
|
Derived training tools must not accept a merely similarly named JSON
|
|
summary. This gate binds every selected image/label pair to the already
|
|
re-hashed asset manifest and requires the same immutable corpus/release.
|
|
It is deliberately strict: a transform that changes image bytes needs a
|
|
new governed corpus and release instead of inheriting production authority.
|
|
"""
|
|
|
|
yaml_path = train_yaml.resolve(strict=False)
|
|
release = assert_training_release_eligible(
|
|
train_yaml=yaml_path,
|
|
corpus_manifest=corpus_manifest,
|
|
fixture_mode=fixture_mode,
|
|
live_session_factory=live_session_factory,
|
|
live_dataset_model=live_dataset_model,
|
|
)
|
|
summary = _read_json(summary_path.resolve(strict=False), label="YOLO tile summary")
|
|
release_paths = training_release_paths(yaml_path)
|
|
release_manifest_path = release_paths["release_manifest"]
|
|
corpus = release.get("corpus")
|
|
assets = release.get("asset_manifest")
|
|
yaml_evidence = release.get("dataset_yaml")
|
|
if not isinstance(corpus, Mapping) or not isinstance(assets, Mapping) or not isinstance(yaml_evidence, Mapping):
|
|
raise TrainingReleaseError("Training release lacks corpus, asset or YAML evidence for summary binding")
|
|
|
|
required_bindings = {
|
|
"dataset_yaml": _canonical_path(yaml_path),
|
|
"training_release_manifest": _canonical_path(release_manifest_path),
|
|
"training_release_manifest_sha256": file_sha256(release_manifest_path),
|
|
"source_manifest_sha256": corpus.get("manifest_sha256"),
|
|
}
|
|
for field_name, expected in required_bindings.items():
|
|
if summary.get(field_name) != expected:
|
|
raise TrainingReleaseError(
|
|
f"YOLO tile summary {field_name!r} is not bound to the eligible training release"
|
|
)
|
|
if yaml_evidence.get("path") != _canonical_path(yaml_path):
|
|
raise TrainingReleaseError("Training release YAML evidence does not match the supplied tile summary YAML")
|
|
if summary.get("training_asset_manifest") != _canonical_path(release_paths["asset_manifest"]):
|
|
raise TrainingReleaseError("YOLO tile summary points to a different training asset manifest")
|
|
|
|
stored_assets = _read_json(release_paths["asset_manifest"], label="Training asset manifest")
|
|
entries = stored_assets.get("entries")
|
|
tiles = summary.get("tiles")
|
|
if not isinstance(entries, list) or not entries or not isinstance(tiles, list) or not tiles:
|
|
raise TrainingReleaseError("YOLO tile summary or training asset manifest has no entries")
|
|
expected_pairs: set[tuple[str, str, str]] = set()
|
|
for asset in entries:
|
|
if not isinstance(asset, Mapping):
|
|
raise TrainingReleaseError("Training asset manifest has an invalid entry")
|
|
split = str(asset.get("split") or "").strip().lower()
|
|
image_path = asset.get("image_path")
|
|
label_path = asset.get("label_path")
|
|
if split not in {"train", "val"} or not isinstance(image_path, str) or not isinstance(label_path, str):
|
|
raise TrainingReleaseError("Training asset manifest entry is incomplete")
|
|
expected_pairs.add((split, _canonical_path(Path(image_path)), _canonical_path(Path(label_path))))
|
|
if len(expected_pairs) != len(entries):
|
|
raise TrainingReleaseError("Training asset manifest contains duplicate image/label pairs")
|
|
|
|
observed_pairs: set[tuple[str, str, str]] = set()
|
|
for index, tile in enumerate(tiles):
|
|
if not isinstance(tile, Mapping):
|
|
raise TrainingReleaseError(f"YOLO tile summary entry {index} is invalid")
|
|
split = str(tile.get("split") or "").strip().lower()
|
|
image_path = tile.get("image_path")
|
|
label_path = tile.get("label_path")
|
|
if split not in {"train", "val"} or not isinstance(image_path, str) or not isinstance(label_path, str):
|
|
raise TrainingReleaseError(f"YOLO tile summary entry {index} is incomplete")
|
|
pair = (split, _canonical_path(Path(image_path)), _canonical_path(Path(label_path)))
|
|
if pair not in expected_pairs:
|
|
raise TrainingReleaseError(f"YOLO tile summary entry {index} is not in the immutable training assets")
|
|
if pair in observed_pairs:
|
|
raise TrainingReleaseError(f"YOLO tile summary repeats immutable training asset {index}")
|
|
observed_pairs.add(pair)
|
|
if observed_pairs != expected_pairs:
|
|
raise TrainingReleaseError("YOLO tile summary is not a complete immutable view of the training assets")
|
|
return release
|
|
|
|
|
|
def assert_yolo_summary_bound_to_embedded_training_release(
|
|
*,
|
|
summary_path: Path,
|
|
corpus_manifest: Path | None = None,
|
|
fixture_mode: bool = False,
|
|
live_session_factory: Callable[[], Any] | None = None,
|
|
live_dataset_model: type[Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Resolve a generated summary's YAML only long enough to verify its release.
|
|
|
|
The embedded path has no authority on its own. The delegated exact-binding
|
|
check below requires it to match the immutable release manifest and asset
|
|
manifest before a derived sampler may read any tile from the summary.
|
|
"""
|
|
|
|
summary = _read_json(summary_path.resolve(strict=False), label="YOLO tile summary")
|
|
raw_yaml = summary.get("dataset_yaml")
|
|
if not isinstance(raw_yaml, str) or not raw_yaml.strip():
|
|
raise TrainingReleaseError("YOLO tile summary has no dataset YAML binding")
|
|
return assert_yolo_summary_bound_to_training_release(
|
|
summary_path=summary_path,
|
|
train_yaml=Path(raw_yaml),
|
|
corpus_manifest=corpus_manifest,
|
|
fixture_mode=fixture_mode,
|
|
live_session_factory=live_session_factory,
|
|
live_dataset_model=live_dataset_model,
|
|
)
|
|
|
|
|
|
def _main() -> int:
|
|
parser = argparse.ArgumentParser(description="Seal or verify an immutable GeoIntel YOLO training release.")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
create = subparsers.add_parser("create", help="Create immutable sidecars for one frozen training release.")
|
|
create.add_argument("--train-yaml", type=Path, required=True)
|
|
create.add_argument("--corpus-manifest", type=Path, required=True)
|
|
create.add_argument("--review-audit", type=Path)
|
|
create.add_argument("--fixture-mode", action="store_true")
|
|
verify = subparsers.add_parser("verify", help="Verify a release immediately before training or resume.")
|
|
verify.add_argument("--train-yaml", type=Path, required=True)
|
|
verify.add_argument("--corpus-manifest", type=Path)
|
|
verify.add_argument("--fixture-mode", action="store_true")
|
|
args = parser.parse_args()
|
|
try:
|
|
if args.command == "create":
|
|
paths = create_training_release_manifest(
|
|
train_yaml=args.train_yaml,
|
|
corpus_manifest=args.corpus_manifest,
|
|
review_audit_path=args.review_audit,
|
|
fixture_mode=args.fixture_mode,
|
|
)
|
|
print(json.dumps({key: _canonical_path(value) for key, value in paths.items()}, indent=2))
|
|
else:
|
|
release = assert_training_release_eligible(
|
|
train_yaml=args.train_yaml,
|
|
corpus_manifest=args.corpus_manifest,
|
|
fixture_mode=args.fixture_mode,
|
|
)
|
|
print(json.dumps({"status": "eligible", "release": release}, indent=2))
|
|
except (TrainingReleaseError, TrainingEligibilityError) as exc:
|
|
raise SystemExit(str(exc)) from exc
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(_main())
|