Jens/geointel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(accuracy): refresh governed scan and training evidence

Browse Source
This commit is contained in:
Jens
2026-08-30 06:00:57 +02:00
parent c272220277
commit 0e3c1b20e9
27 changed files with 88787 additions and 318 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/activate_promoted_yolo_candidate.py
+77
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
@@ -13,6 +14,14 @@ SUPPORTED_MODEL_SUFFIXES = {".pt", ".onnx", ".engine"}
ENV_KEYS = ("GEOINTEL_INSTALL_AI", "YOLO_ENABLED", "YOLO_MODELS_DIR", "YOLO_MODEL_PATH")
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 _asset_id(path: Path) -> str:
raw = f"{path.stem}-{path.suffix.lower().lstrip('.')}"
normalized = re.sub(r"[^a-z0-9]+", "-", raw.lower()).strip("-")
@@ -116,6 +125,36 @@ def _gate_failures(report: dict[str, Any], candidate: dict[str, Any]) -> list[st
return failures
def _governed_release_failures(
release_gate: dict[str, Any],
*,
candidate_key: str,
model_sha256: str,
) -> list[str]:
"""Require the Phase-4/5 product decision that the legacy gate never had."""
failures: list[str] = []
if release_gate.get("status") != "pass":
failures.append("phase4_release_gate_not_passed")
if release_gate.get("product_benchmark_status") != "pass":
failures.append("product_benchmark_not_passed")
if release_gate.get("promotion_allowed") is not True:
failures.append("governed_promotion_not_allowed")
if release_gate.get("phase_decision") not in {"done", "ready", "complete"}:
failures.append("phase5_not_ready")
if release_gate.get("candidate_key") != candidate_key:
failures.append("governed_candidate_key_mismatch")
recorded_sha256 = str(release_gate.get("candidate_model_sha256") or "").lower()
if not re.fullmatch(r"[0-9a-f]{64}", recorded_sha256):
failures.append("governed_candidate_model_sha256_missing")
elif recorded_sha256 != model_sha256:
failures.append("governed_candidate_model_sha256_mismatch")
benchmark_sha256 = str(release_gate.get("benchmark_manifest_sha256") or "").lower()
if not re.fullmatch(r"[0-9a-f]{64}", benchmark_sha256):
failures.append("governed_benchmark_manifest_sha256_missing")
return failures
def _resolve_model_asset(models_dir: Path, model_asset_id: str) -> Path | None:
for path in _candidate_paths(models_dir):
if _asset_id(path) == model_asset_id:
@@ -126,6 +165,7 @@ def _resolve_model_asset(models_dir: Path, model_asset_id: str) -> Path | None:
def _base_payload(args: argparse.Namespace) -> dict[str, Any]:
return {
"promotion_report": str(Path(args.promotion_report).resolve()),
"phase4_release_gate_report": str(Path(args.phase4_release_gate_report).resolve()),
"candidate_key": args.candidate_key,
"models_dir": str(Path(args.models_dir).resolve()),
"env_file": str(Path(args.env_file).resolve()),
@@ -188,6 +228,33 @@ def activate(args: argparse.Namespace) -> tuple[int, dict[str, Any]]:
)
return 2, payload
model_sha256 = _sha256(selected_model)
try:
release_gate_path = Path(args.phase4_release_gate_report)
release_gate = _load_report(release_gate_path)
except (OSError, ValueError, json.JSONDecodeError) as exc:
payload.update({"status": "invalid_phase4_release_gate", "message": str(exc)})
return 2, payload
governance_failures = _governed_release_failures(
release_gate,
candidate_key=args.candidate_key,
model_sha256=model_sha256,
)
if governance_failures:
payload.update(
{
"status": "governed_release_not_approved",
"message": (
"Candidate passed the legacy diagnostic gates but lacks a matching "
"successful governed product benchmark and Phase-4/5 release decision."
),
"rejection_reasons": governance_failures,
"selected_host_model_path": str(selected_model),
"selected_model_sha256": model_sha256,
}
)
return 3, payload
selected_container_path = _container_path(selected_model, models_dir, args.container_model_dir)
updates = {
"GEOINTEL_INSTALL_AI": "true",
@@ -200,7 +267,9 @@ def activate(args: argparse.Namespace) -> tuple[int, dict[str, Any]]:
"status": "ready_to_apply",
"message": "Promoted YOLO candidate validated. Re-run with --apply to update the environment file.",
"selected_host_model_path": str(selected_model),
"selected_model_sha256": model_sha256,
"selected_container_model_path": selected_container_path,
"phase4_release_gate_sha256": _sha256(release_gate_path),
"env_updates": updates,
"docker_restart_required": True,
}
@@ -241,6 +310,14 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
description="Activate an existing local YOLO model only after a promotion report recommends the exact candidate key."
)
parser.add_argument("--promotion-report", required=True, help="Path to detection_model_promotion_report.json.")
parser.add_argument(
"--phase4-release-gate-report",
required=True,
help=(
"Path to the governed Phase-4/5 release-gate report. It must bind the exact "
"candidate key and model SHA-256 and explicitly allow promotion."
),
)
parser.add_argument("--candidate-key", required=True, help="Exact promoted candidate key from the report.")
parser.add_argument(
"--models-dir",
scripts/archive_technical_projects.py
+2 -2
View File
@@ -17,8 +17,8 @@ BACKEND_ROOT = ROOT / "backend" if (ROOT / "backend" / "app").is_dir() else ROOT
if str(BACKEND_ROOT) not in sys.path:
sys.path.insert(0, str(BACKEND_ROOT))
from app.db.session import SessionLocal
from app.models import Project
from app.db.session import SessionLocal # noqa: E402 - imported after backend path bootstrap
from app.models import Project # noqa: E402 - imported after backend path bootstrap
CANONICAL_PROJECT_NAMES = frozenset(
scripts/build_detection_model_promotion_report.py
+10
View File
@@ -324,7 +324,14 @@ def build_decisions(args: argparse.Namespace) -> dict[str, Any]:
)
return {
"schema_version": 2,
"generated_at": datetime.now(timezone.utc).isoformat(),
"promotion_authority": "legacy_diagnostic_only",
"runtime_activation_allowed": False,
"activation_requirement": (
"A separate successful Phase-4/5 release-gate report must bind the exact "
"candidate key and model SHA-256 before activation."
),
"positive_portfolio_path": str(portfolio_path),
"hard_negative_summary_paths": [str(path) for path in background_paths],
"background_split_summary_paths": [str(path) for path in split_summary_paths],
@@ -350,6 +357,9 @@ def write_markdown(report: dict[str, Any], path: Path) -> None:
f"- Hard-negative summaries: {len(report['hard_negative_summary_paths'])}",
f"- Background split summaries: {len(report['background_split_summary_paths'])}",
f"- Candidates: {report['candidate_count']}",
f"- Authority: `{report['promotion_authority']}`",
"- Runtime activation: **not authorized by this report**",
f"- Activation requirement: {report['activation_requirement']}",
"",
"## Gates",
"",
scripts/provision_waterinfo_station_history.py
+1 -1
View File
@@ -23,7 +23,7 @@ from typing import Any
import requests
from requests.adapters import HTTPAdapter
from shapely.geometry import Point, mapping, shape
from shapely.geometry import shape
from urllib3.util.retry import Retry
scripts/run_accuracy_phase3_full_data_scan.py
+52 -13
View File
@@ -21,7 +21,7 @@ from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
SCANNER_VERSION = "3.0.3"
SCANNER_VERSION = "3.1.0"
SCHEMA_VERSION = 1
DEFAULT_ROOTS = ("models", "datasets", "data", "storage", "artifacts", "output")
EXCLUDED_DIRS = {".git", "node_modules", ".next", "__pycache__", ".pytest_cache"}
@@ -101,13 +101,16 @@ def discover_files(repo_root: Path, roots: Iterable[str], evidence_dir: Path) ->
return sorted(set(files), key=lambda item: relative_path(item, repo_root))
def known_unreachable_items() -> list[dict[str, Any]]:
"""Boundaries identified by the Phase 1 inventory but not mounted locally."""
entries = (
("external://tower-corpora", "Tower corpora are not mounted in this project environment"),
("external://mounted-model-volumes", "Mounted model volumes are not available from this project environment"),
("external://production-postgis-or-api", "No production PostGIS/API endpoint is configured in the scan environment"),
)
def unreachable_items(entries: Iterable[tuple[str, str]]) -> list[dict[str, Any]]:
"""Create explicit scope records without inventing environment availability.
Older scanner versions always emitted three hard-coded unreachable items,
even when executed inside the production container where the corresponding
model and training mounts were available. Callers must now declare only
boundaries they have actually established as unavailable; requested roots
that do not exist are added automatically by :func:`missing_root_entries`.
"""
result = []
for path, reason in entries:
result.append(
@@ -140,6 +143,29 @@ def known_unreachable_items() -> list[dict[str, Any]]:
return result
def parse_unreachable_scope(value: str) -> tuple[str, str]:
path, separator, reason = value.partition("=")
if not separator or not path.strip() or not reason.strip():
raise argparse.ArgumentTypeError(
"--unreachable-scope must use PATH=CONCRETE_REASON"
)
return path.strip(), reason.strip()
def missing_root_entries(repo_root: Path, roots: Iterable[str]) -> list[tuple[str, str]]:
entries: list[tuple[str, str]] = []
for root in roots:
base = (repo_root / root).resolve(strict=False)
if not base.exists():
entries.append(
(
f"root://{root}",
f"Configured scan root is unavailable: {base}",
)
)
return entries
def infer_source(rel: str, payload: Any = None) -> str:
text = rel.lower()
if isinstance(payload, dict):
@@ -507,6 +533,17 @@ def parse_args() -> argparse.Namespace:
parser.add_argument("--batch-size", type=int, default=50)
parser.add_argument("--resume", action="store_true")
parser.add_argument("--roots", nargs="+", default=list(DEFAULT_ROOTS))
parser.add_argument(
"--unreachable-scope",
action="append",
default=[],
type=parse_unreachable_scope,
metavar="PATH=CONCRETE_REASON",
help=(
"Explicit non-filesystem or externally mounted boundary that was "
"verified as unavailable. May be repeated."
),
)
return parser.parse_args()
@@ -518,16 +555,18 @@ def main() -> int:
raise SystemExit("--batch-size must be positive")
paths = discover_files(repo_root, args.roots, output_dir)
inventory_entries = [{"path": relative_path(path, repo_root), "size_bytes": path.stat().st_size, "modified_ns": path.stat().st_mtime_ns} for path in paths]
unreachable_items = known_unreachable_items()
unavailable = unreachable_items(
[*missing_root_entries(repo_root, args.roots), *args.unreachable_scope]
)
inventory = {
"roots": list(args.roots),
"excluded_directories": sorted(EXCLUDED_DIRS),
"excluded_output_dir": relative_path(output_dir, repo_root),
"items": inventory_entries,
"unreachable_items": [{"path": item["path"], "reason": item["anomalies"][0]["message"]} for item in unreachable_items],
"unreachable_items": [{"path": item["path"], "reason": item["anomalies"][0]["message"]} for item in unavailable],
"accessible_item_count": len(inventory_entries),
"item_count": len(inventory_entries) + len(unreachable_items),
"inventory_hash": canonical_hash({"items": inventory_entries, "unreachable_items": [{"path": item["path"], "reason": item["anomalies"][0]["message"]} for item in unreachable_items]}),
"item_count": len(inventory_entries) + len(unavailable),
"inventory_hash": canonical_hash({"items": inventory_entries, "unreachable_items": [{"path": item["path"], "reason": item["anomalies"][0]["message"]} for item in unavailable]}),
}
checkpoint_path = output_dir / "scan-checkpoint.json"
checkpoint: dict[str, Any] | None = None
@@ -550,7 +589,7 @@ def main() -> int:
records[rel] = make_item(path, repo_root)
atomic_json(checkpoint_path, {"schema_version": 1, "scanner_version": SCANNER_VERSION, "started_at": started_at, "inventory": inventory, "cursor": min(offset + args.batch_size, len(paths)), "records": sorted(records.values(), key=lambda item: item["path"])})
items = [records[relative_path(path, repo_root)] for path in paths if relative_path(path, repo_root) in records]
items.extend(unreachable_items)
items.extend(unavailable)
scan_id = "p3-" + inventory["inventory_hash"][:16]
report = report_payload(items, inventory, scan_id, started_at=started_at, completed_at=utc_now(), repo_root=repo_root)
output_dir.mkdir(parents=True, exist_ok=True)
scripts/run_belgium_building_training_loop.py
+61 -8
View File
@@ -277,6 +277,54 @@ def failure_sampling_command(
return command
def protected_assessment_command(
*,
scripts_dir: Path,
calibration: Path,
test: Path,
background: Path,
output: Path,
selected_threshold: float,
min_aggregate_f1: float,
min_region_f1: float,
min_region_precision: float,
min_region_recall: float,
max_pure_empty_fp: int,
) -> list[str]:
"""Build the one-shot protected assessment command from frozen calibration gates.
The protected assessor must not silently reselect an operating point or
fall back to its own default gates. Keeping command construction in one
helper also makes the release boundary directly testable without opening
any protected inputs.
"""
return [
sys.executable,
str(scripts_dir / "assess_belgium_building_training_iteration.py"),
"--calibration",
str(calibration),
"--test",
str(test),
"--background",
str(background),
"--output",
str(output),
"--selected-threshold",
str(selected_threshold),
"--min-aggregate-f1",
str(min_aggregate_f1),
"--min-region-f1",
str(min_region_f1),
"--min-region-precision",
str(min_region_precision),
"--min-region-recall",
str(min_region_recall),
"--max-pure-empty-fp",
str(max_pure_empty_fp),
]
def resumable_training_command(yolo: str, checkpoint: Path) -> list[str]:
return [yolo, "train", f"resume={checkpoint}", "device=0"]
@@ -572,14 +620,19 @@ def main() -> int:
iteration_dir / f"{role}.log",
)
run(
[
sys.executable,
str(scripts_dir / "assess_belgium_building_training_iteration.py"),
"--calibration", str(reports["calibration"]),
"--test", str(reports["test"]),
"--background", str(reports["background"]),
"--output", str(assessment),
],
protected_assessment_command(
scripts_dir=scripts_dir,
calibration=reports["calibration"],
test=reports["test"],
background=reports["background"],
output=assessment,
selected_threshold=float(chosen["threshold"]),
min_aggregate_f1=args.min_aggregate_f1,
min_region_f1=args.min_region_f1,
min_region_precision=args.min_region_precision,
min_region_recall=args.min_region_recall,
max_pure_empty_fp=args.max_pure_empty_fp,
),
iteration_dir / "assessment.log",
allowed={0, 2},
)
scripts/seed_demo_workflow.py
+4 -2
View File
@@ -8,8 +8,10 @@ from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "backend"))
from app.db.session import SessionLocal
from app.services.demo_workflow_service import DemoWorkflowService
from app.db.session import SessionLocal # noqa: E402 - imported after backend path bootstrap
from app.services.demo_workflow_service import ( # noqa: E402 - imported after backend path bootstrap
DemoWorkflowService,
)
def main() -> int:
scripts/smoke_docs.py
+13 -3
View File
@@ -1,6 +1,16 @@
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
required_docs = ["docs/M5_OPERATIONAL_READINESS.md","docs/BUILD_GOVERNANCE.md","docs/CI_CD_SPECIFICATION.md","docs/HEALTHCHECK_CONTRACTS.md","docs/TROUBLESHOOTING_RUNBOOK.md","docs/RELEASE_PROCESS.md","docs/ROLLBACK_AND_RECOVERY.md"]
missing=[p for p in required_docs if not (ROOT/p).exists()]
if missing: raise SystemExit("Missing docs:\n"+"\n".join(missing))
required_docs = [
"docs/M5_OPERATIONAL_READINESS.md",
"docs/BUILD_GOVERNANCE.md",
"docs/CI_CD_SPECIFICATION.md",
"docs/HEALTHCHECK_CONTRACTS.md",
"docs/TROUBLESHOOTING_RUNBOOK.md",
"docs/RELEASE_PROCESS.md",
"docs/ROLLBACK_AND_RECOVERY.md",
]
missing = [p for p in required_docs if not (ROOT / p).exists()]
if missing:
raise SystemExit("Missing docs:\n" + "\n".join(missing))
print("Documentation smoke OK")