geointel/scripts/run_mol_operational_validation.sh

#!/usr/bin/env bash
set -euo pipefail

usage() {
  cat >&2 <<'EOF'
Usage:
  OPERATOR_SAMPLE_MANIFEST_PATH=storage/operator-data/mol-operational-1024/operator_samples_manifest.json \
  bash scripts/run_mol_operational_validation.sh [base_url]

Optional environment:
  MOL_POSITIVE_SAMPLE_SLUGS    Positive Mol holdouts. Default: mol_achterbos mol_gompel mol_donk mol_postel.
  MOL_BACKGROUND_SAMPLE_SLUGS  Mol background controls. Default: postel_bos.
  MOL_VALIDATION_OUTPUT_DIR    Output directory, default: artifacts/mol-operational-validation/<timestamp>.
  QUALITY_MODEL_ASSET_IDS      Local model asset ID, default: active configured model.
  QUALITY_TILE_SIZES           Default: 512.
  QUALITY_TILE_OVERLAPS        Default: 64.
  QUALITY_THRESHOLDS           Default: 0.15.
  REAL_IOU_THRESHOLD           Default: 0.25.

The runner never downloads weights, fetches product providers or uses fixture
outputs. Prepare the documented real orthophoto/GRB files explicitly first.
EOF
}

ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "${ROOT}"

BASE_URL="${1:-${GE_INTEL_BASE_URL:-http://localhost:1202}}"
OPERATOR_SAMPLE_MANIFEST_PATH="${OPERATOR_SAMPLE_MANIFEST_PATH:-storage/operator-data/mol-operational-1024/operator_samples_manifest.json}"
MOL_POSITIVE_SAMPLE_SLUGS="${MOL_POSITIVE_SAMPLE_SLUGS:-mol_achterbos mol_gompel mol_donk mol_postel}"
MOL_BACKGROUND_SAMPLE_SLUGS="${MOL_BACKGROUND_SAMPLE_SLUGS:-postel_bos}"
MOL_VALIDATION_OUTPUT_DIR="${MOL_VALIDATION_OUTPUT_DIR:-artifacts/mol-operational-validation/$(date -u +%Y%m%dT%H%M%SZ)}"
QUALITY_MODEL_ASSET_IDS="${QUALITY_MODEL_ASSET_IDS:-${REAL_MODEL_ASSET_ID:-__active__}}"
QUALITY_TILE_SIZES="${QUALITY_TILE_SIZES:-512}"
QUALITY_TILE_OVERLAPS="${QUALITY_TILE_OVERLAPS:-64}"
QUALITY_THRESHOLDS="${QUALITY_THRESHOLDS:-0.15}"
REAL_IOU_THRESHOLD="${REAL_IOU_THRESHOLD:-0.25}"

if [ "${BASE_URL}" = "-h" ] || [ "${BASE_URL}" = "--help" ]; then
  usage
  exit 0
fi

if [ ! -f "${OPERATOR_SAMPLE_MANIFEST_PATH}" ]; then
  echo "Mol operator manifest is not readable: ${OPERATOR_SAMPLE_MANIFEST_PATH}" >&2
  exit 2
fi

if [ -n "${PYTHON_BIN:-}" ]; then
  PYTHON_BIN="${PYTHON_BIN}"
else
  PYTHON_BIN=""
  for candidate in python3 python.exe python; do
    if command -v "${candidate}" >/dev/null 2>&1 && "${candidate}" -c "import json, sys" >/dev/null 2>&1; then
      PYTHON_BIN="${candidate}"
      break
    fi
  done
fi
if [ -z "${PYTHON_BIN}" ]; then
  echo "A Python interpreter is required for Mol operational validation" >&2
  exit 1
fi

mkdir -p "${MOL_VALIDATION_OUTPUT_DIR}"
positive_output_dir="${MOL_VALIDATION_OUTPUT_DIR}/positive-qa"
background_output_dir="${MOL_VALIDATION_OUTPUT_DIR}/background-control"

"${PYTHON_BIN}" - \
  "${OPERATOR_SAMPLE_MANIFEST_PATH}" \
  "${MOL_POSITIVE_SAMPLE_SLUGS}" \
  "${MOL_BACKGROUND_SAMPLE_SLUGS}" <<'PY'
import json
import sys
from pathlib import Path

manifest_path = Path(sys.argv[1])
positive_slugs = {value.strip().lower() for value in sys.argv[2].replace(",", " ").split() if value.strip()}
background_slugs = {value.strip().lower() for value in sys.argv[3].replace(",", " ").split() if value.strip()}
payload = json.loads(manifest_path.read_text(encoding="utf-8-sig"))
samples = {str(sample.get("sample_slug") or "").lower(): sample for sample in payload.get("samples") or []}
required = positive_slugs | background_slugs
missing = sorted(required - samples.keys())
if missing:
    raise SystemExit(f"Mol manifest is missing required samples: {', '.join(missing)}")
for slug in sorted(positive_slugs):
    sample = samples[slug]
    if sample.get("municipality") != "Mol":
        raise SystemExit(f"Positive sample is not attributed to Mol: {slug}")
    if sample.get("recommended_split") != "val":
        raise SystemExit(f"Positive Mol operational sample is not a validation holdout: {slug}")
    if int(sample.get("reference_feature_count") or 0) < 1:
        raise SystemExit(f"Positive Mol operational sample has no GRB references: {slug}")
for slug in sorted(background_slugs):
    sample = samples[slug]
    if sample.get("municipality") != "Mol":
        raise SystemExit(f"Background sample is not attributed to Mol: {slug}")
    if sample.get("sample_role") != "background_candidate" and not sample.get("allow_empty_reference"):
        raise SystemExit(f"Mol background sample is not explicitly marked as background: {slug}")
PY

echo "== GeoIntel Mol operational validation =="
echo "Base URL: ${BASE_URL}"
echo "Manifest: ${OPERATOR_SAMPLE_MANIFEST_PATH}"
echo "Positive holdouts: ${MOL_POSITIVE_SAMPLE_SLUGS}"
echo "Background controls: ${MOL_BACKGROUND_SAMPLE_SLUGS}"
echo "Output: ${MOL_VALIDATION_OUTPUT_DIR}"

OPERATOR_SAMPLE_MANIFEST_PATH="${OPERATOR_SAMPLE_MANIFEST_PATH}" \
OPERATOR_SAMPLE_SLUGS="${MOL_POSITIVE_SAMPLE_SLUGS}" \
MULTI_SAMPLE_OUTPUT_DIR="${positive_output_dir}" \
QUALITY_MODEL_ASSET_IDS="${QUALITY_MODEL_ASSET_IDS}" \
QUALITY_TILE_SIZES="${QUALITY_TILE_SIZES}" \
QUALITY_TILE_OVERLAPS="${QUALITY_TILE_OVERLAPS}" \
QUALITY_THRESHOLDS="${QUALITY_THRESHOLDS}" \
REAL_IOU_THRESHOLD="${REAL_IOU_THRESHOLD}" \
bash scripts/run_multi_sample_detection_quality_matrix.sh "${BASE_URL}"

OPERATOR_SAMPLE_MANIFEST_PATH="${OPERATOR_SAMPLE_MANIFEST_PATH}" \
OPERATOR_BACKGROUND_SAMPLE_SLUGS="${MOL_BACKGROUND_SAMPLE_SLUGS}" \
HARD_NEGATIVE_OUTPUT_DIR="${background_output_dir}" \
QUALITY_MODEL_ASSET_IDS="${QUALITY_MODEL_ASSET_IDS}" \
QUALITY_TILE_SIZES="${QUALITY_TILE_SIZES}" \
QUALITY_TILE_OVERLAPS="${QUALITY_TILE_OVERLAPS}" \
QUALITY_THRESHOLDS="${QUALITY_THRESHOLDS}" \
bash scripts/run_operator_hard_negative_detection_matrix.sh "${BASE_URL}"

"${PYTHON_BIN}" - \
  "${positive_output_dir}/multi_sample_quality_summary.json" \
  "${background_output_dir}/hard_negative_matrix_summary.json" \
  "${MOL_VALIDATION_OUTPUT_DIR}" \
  "${BASE_URL}" \
  "${OPERATOR_SAMPLE_MANIFEST_PATH}" <<'PY'
import json
import statistics
import sys
from datetime import datetime, timezone
from pathlib import Path

positive_path = Path(sys.argv[1])
background_path = Path(sys.argv[2])
output_dir = Path(sys.argv[3])
base_url = sys.argv[4]
manifest_path = sys.argv[5]
positive = json.loads(positive_path.read_text(encoding="utf-8"))
background = json.loads(background_path.read_text(encoding="utf-8"))
positive_items = positive.get("items") or []
background_items = background.get("items") or []
if not positive_items or not background_items:
    raise SystemExit("Mol operational validation did not produce both positive and background evidence")

def metric_values(key: str) -> list[float]:
    return [float(item[key]) for item in positive_items if item.get(key) is not None]

f1_values = metric_values("f1_score")
precision_values = metric_values("precision")
recall_values = metric_values("recall")
summary = {
    "schema_version": 1,
    "status": "evidence_ready",
    "generated_at": datetime.now(timezone.utc).isoformat(),
    "base_url": base_url,
    "operator_sample_manifest_path": manifest_path,
    "positive_summary_path": str(positive_path),
    "background_summary_path": str(background_path),
    "positive_sample_count": int(positive.get("sample_count") or 0),
    "positive_run_count": len(positive_items),
    "background_sample_count": int(background.get("sample_count") or 0),
    "background_run_count": len(background_items),
    "mean_precision": statistics.fmean(precision_values) if precision_values else None,
    "mean_recall": statistics.fmean(recall_values) if recall_values else None,
    "mean_f1": statistics.fmean(f1_values) if f1_values else None,
    "minimum_f1": min(f1_values) if f1_values else None,
    "total_matches": sum(int(item.get("matches") or 0) for item in positive_items),
    "total_false_positives": sum(int(item.get("false_positives") or 0) for item in positive_items),
    "total_false_negatives": sum(int(item.get("false_negatives") or 0) for item in positive_items),
    "total_background_detections": sum(int(item.get("detection_count") or 0) for item in background_items),
    "zero_detection_background_runs": sum(1 for item in background_items if int(item.get("detection_count") or 0) == 0),
    "project_ids": [item.get("project_id") for item in positive_items + background_items],
    "area_ids": [item.get("area_id") for item in positive_items + background_items],
    "analysis_run_ids": [item.get("analysis_run_id") for item in positive_items + background_items],
    "quality_check_ids": [item.get("quality_check_id") for item in positive_items],
    "positive_items": positive_items,
    "background_items": background_items,
}
summary_path = output_dir / "mol_operational_validation_summary.json"
summary_path.write_text(json.dumps(summary, indent=2, sort_keys=True), encoding="utf-8")

def fmt(value):
    return "n/a" if value is None else f"{value:.4f}"

lines = [
    "# Mol operational validation",
    "",
    f"- Status: `{summary['status']}`",
    f"- Positive samples/runs: `{summary['positive_sample_count']}` / `{summary['positive_run_count']}`",
    f"- Background samples/runs: `{summary['background_sample_count']}` / `{summary['background_run_count']}`",
    f"- Mean precision: `{fmt(summary['mean_precision'])}`",
    f"- Mean recall: `{fmt(summary['mean_recall'])}`",
    f"- Mean F1: `{fmt(summary['mean_f1'])}`",
    f"- Minimum F1: `{fmt(summary['minimum_f1'])}`",
    f"- Total matches / FP / FN: `{summary['total_matches']}` / `{summary['total_false_positives']}` / `{summary['total_false_negatives']}`",
    f"- Background detections: `{summary['total_background_detections']}`",
    "",
    "`evidence_ready` records completed persisted workflows; it is not an automatic model-promotion decision.",
]
(output_dir / "mol_operational_validation_summary.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
print(json.dumps({"status": summary["status"], "summary_path": str(summary_path)}, indent=2))
PY