geointel/scripts/run_operator_hard_negative_detection_matrix.sh

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

usage() {
  cat >&2 <<'EOF'
Usage:
  OPERATOR_SAMPLE_MANIFEST_PATH=storage/operator-data/operator_samples_manifest.json \
  OPERATOR_BACKGROUND_SAMPLE_SLUGS="postel_bos lommel_heide kasterlee_bos" \
  QUALITY_MODEL_ASSET_IDS="geointel-building-yolov8n-expanded160e50-pt geointel-building-yolov8n-tile30-pt" \
  QUALITY_THRESHOLDS="0.50 0.25 0.15" \
  bash scripts/run_operator_hard_negative_detection_matrix.sh [base_url]

Optional environment:
  OPERATOR_SAMPLE_MANIFEST_PATH      Manifest from prepare_operator_real_data_samples.py.
  OPERATOR_BACKGROUND_SAMPLE_SLUGS   Optional comma/space separated filter. Defaults to samples marked background_candidate or allow_empty_reference.
  HARD_NEGATIVE_OUTPUT_DIR           Output directory, default: artifacts/detection-hard-negatives/<timestamp>.
  QUALITY_MODEL_ASSET_IDS            Space/comma separated local model asset IDs. Default: active configured model asset.
  QUALITY_TILE_SIZES                 Space/comma separated raster tile sizes, default: 640.
  QUALITY_TILE_OVERLAPS              Space/comma separated raster tile overlaps, default: 64.
  QUALITY_THRESHOLDS                 Space/comma separated confidence thresholds, default: 0.50 0.25 0.15.

This script scores false-positive pressure on documented operator background
samples only. It runs the existing raster upload, tile, configured-YOLO
preflight and detection API path, then counts detections. It does not upload
reference vectors, run QA/QC, use fixture detections, fetch providers or
download model weights.
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/operator_samples_manifest.json}"
OPERATOR_BACKGROUND_SAMPLE_SLUGS="${OPERATOR_BACKGROUND_SAMPLE_SLUGS:-}"
HARD_NEGATIVE_OUTPUT_DIR="${HARD_NEGATIVE_OUTPUT_DIR:-artifacts/detection-hard-negatives/$(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:-640}"
QUALITY_TILE_OVERLAPS="${QUALITY_TILE_OVERLAPS:-64}"
QUALITY_THRESHOLDS="${QUALITY_THRESHOLDS:-0.50 0.25 0.15}"

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

if [ ! -f "${OPERATOR_SAMPLE_MANIFEST_PATH}" ]; then
  echo "OPERATOR_SAMPLE_MANIFEST_PATH does not point to a readable manifest: ${OPERATOR_SAMPLE_MANIFEST_PATH}" >&2
  exit 2
fi

if ! command -v curl >/dev/null 2>&1; then
  echo "curl is required for hard-negative detection matrix verification" >&2
  exit 1
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 JSON parsing" >&2
  exit 1
fi

mkdir -p "${HARD_NEGATIVE_OUTPUT_DIR}"
sample_manifest_tsv="${HARD_NEGATIVE_OUTPUT_DIR}/hard_negative_samples.tsv"
matrix_manifest="${HARD_NEGATIVE_OUTPUT_DIR}/hard_negative_requests.tsv"

"${PYTHON_BIN}" - \
  "${ROOT}" \
  "${OPERATOR_SAMPLE_MANIFEST_PATH}" \
  "${OPERATOR_BACKGROUND_SAMPLE_SLUGS}" \
  "${sample_manifest_tsv}" <<'PY'
import json
import sys
from pathlib import Path

root = Path(sys.argv[1]).resolve()
manifest_path = Path(sys.argv[2])
slug_filter_raw = sys.argv[3]
output_path = Path(sys.argv[4])

payload = json.loads(manifest_path.read_text(encoding="utf-8-sig"))
samples = payload.get("samples") or []
if not samples:
    raise SystemExit("Operator sample manifest contains no samples")

requested_slugs = {
    value.strip().lower()
    for value in slug_filter_raw.replace(",", " ").split()
    if value.strip()
}


def resolve_path(raw: str) -> str:
    path = Path(raw)
    if path.exists():
        return str(path)
    if raw.startswith("/app/"):
        candidate = root / raw.removeprefix("/app/")
        if candidate.exists():
            return str(candidate)
    candidate = root / raw
    if candidate.exists():
        return str(candidate)
    raise SystemExit(f"Sample file is not readable from this host: {raw}")


with output_path.open("w", encoding="utf-8") as handle:
    selected = 0
    for sample in samples:
        sample_slug = str(sample.get("sample_slug") or "").lower()
        sample_role = str(sample.get("sample_role") or "")
        allow_empty_reference = bool(sample.get("allow_empty_reference"))
        if not sample_slug:
            raise SystemExit("Operator sample is missing sample_slug")
        if requested_slugs and sample_slug not in requested_slugs:
            continue
        if not requested_slugs and sample_role != "background_candidate" and not allow_empty_reference:
            continue
        raster_path = resolve_path(str(sample.get("raster_path") or ""))
        reference_count = int(sample.get("reference_feature_count") or 0)
        handle.write(f"{sample_slug}\t{raster_path}\t{sample_role}\t{allow_empty_reference}\t{reference_count}\n")
        selected += 1

if selected == 0:
    raise SystemExit("No background_candidate samples matched OPERATOR_BACKGROUND_SAMPLE_SLUGS")
PY

model_requests_normalized="$(printf '%s' "${QUALITY_MODEL_ASSET_IDS}" | tr ',' ' ')"
tile_sizes_normalized="$(printf '%s' "${QUALITY_TILE_SIZES}" | tr ',' ' ')"
tile_overlaps_normalized="$(printf '%s' "${QUALITY_TILE_OVERLAPS}" | tr ',' ' ')"
thresholds_normalized="$(printf '%s' "${QUALITY_THRESHOLDS}" | tr ',' ' ')"

"${PYTHON_BIN}" - \
  "${model_requests_normalized}" \
  "${tile_sizes_normalized}" \
  "${tile_overlaps_normalized}" \
  "${thresholds_normalized}" \
  "${matrix_manifest}" <<'PY'
import re
import sys

models, tile_sizes, tile_overlaps, thresholds, output_path = sys.argv[1:6]


def split_values(raw: str, name: str) -> list[str]:
    values = [value.strip() for value in raw.split() if value.strip()]
    if not values:
        raise SystemExit(f"{name} must contain at least one value")
    return values


def safe_label(value: str) -> str:
    label = re.sub(r"[^A-Za-z0-9_.-]+", "_", value.strip())
    return label.replace(".", "p").replace("-", "_") or "active"


model_values = split_values(models, "QUALITY_MODEL_ASSET_IDS")
tile_size_values = split_values(tile_sizes, "QUALITY_TILE_SIZES")
tile_overlap_values = split_values(tile_overlaps, "QUALITY_TILE_OVERLAPS")
threshold_values = split_values(thresholds, "QUALITY_THRESHOLDS")

for value in tile_size_values:
    tile_size = int(value)
    if tile_size < 64:
        raise SystemExit(f"Tile size must be at least 64: {value}")

for value in tile_overlap_values:
    tile_overlap = int(value)
    if tile_overlap < 0:
        raise SystemExit(f"Tile overlap cannot be negative: {value}")

for value in threshold_values:
    threshold = float(value)
    if threshold < 0.0 or threshold > 1.0:
        raise SystemExit(f"Confidence threshold must be between 0 and 1: {value}")

with open(output_path, "w", encoding="utf-8") as handle:
    for model in model_values:
        for tile_size in tile_size_values:
            for overlap in tile_overlap_values:
                for threshold in threshold_values:
                    label = "_".join(
                        [
                            f"model_{safe_label(model)}",
                            f"tile_{safe_label(tile_size)}",
                            f"overlap_{safe_label(overlap)}",
                            f"threshold_{safe_label(threshold)}",
                        ]
                    )
                    handle.write(f"{model}\t{tile_size}\t{overlap}\t{threshold}\t{label}\n")
PY

json_field() {
  local file_path="$1"
  local expression="$2"
  "${PYTHON_BIN}" - "$file_path" "$expression" <<'PY'
import json
import sys

path, expression = sys.argv[1], sys.argv[2]
with open(path, "r", encoding="utf-8") as handle:
    payload = json.load(handle)
value = payload
for part in expression.split("."):
    if part:
        value = value[part]
print(value)
PY
}

require_json_data() {
  local file_path="$1"
  "${PYTHON_BIN}" - "$file_path" <<'PY'
import json
import sys

with open(sys.argv[1], "r", encoding="utf-8") as handle:
    payload = json.load(handle)
if "data" not in payload:
    raise SystemExit("Response is not a canonical GeoIntel data envelope")
PY
}

echo "== GeoIntel operator hard-negative detection matrix =="
echo "Base URL: ${BASE_URL}"
echo "Manifest: ${OPERATOR_SAMPLE_MANIFEST_PATH}"
echo "Background filter: ${OPERATOR_BACKGROUND_SAMPLE_SLUGS:-background_candidate samples}"
echo "Models: ${model_requests_normalized}"
echo "Tile sizes: ${tile_sizes_normalized}"
echo "Tile overlaps: ${tile_overlaps_normalized}"
echo "Thresholds: ${thresholds_normalized}"
echo "Output: ${HARD_NEGATIVE_OUTPUT_DIR}"

while IFS=$'\t' read -r sample_slug raster_path sample_role allow_empty_reference reference_feature_count; do
  echo "-- Background sample ${sample_slug}: role=${sample_role} allow_empty_reference=${allow_empty_reference} reference_features=${reference_feature_count} --"
  while IFS=$'\t' read -r model_request tile_size tile_overlap threshold run_label; do
    sample_output_dir="${HARD_NEGATIVE_OUTPUT_DIR}/${sample_slug}"
    mkdir -p "${sample_output_dir}"
    tmp_dir="$(mktemp -d)"
    run_log="${sample_output_dir}/${run_label}.log"
    run_summary="${sample_output_dir}/${run_label}_summary.json"
    model_env="${model_request}"
    if [ "${model_request}" = "__active__" ]; then
      model_env=""
    fi

    {
      echo "sample=${sample_slug} model=${model_request} tile=${tile_size} overlap=${tile_overlap} threshold=${threshold}"
      "${PYTHON_BIN}" - "${tmp_dir}/project_request.json" "${sample_slug}" "${model_request}" "${threshold}" <<'PY'
import json
import sys
from datetime import datetime, timezone

path, sample_slug, model_request, threshold = sys.argv[1:5]
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
payload = {
    "name": f"GeoIntel hard-negative {sample_slug} {model_request} {threshold} {stamp}",
    "description": "Operator hard-negative validation: raster upload, tiling, configured YOLO detection count, no QA reference.",
    "region": "Kempen",
}
with open(path, "w", encoding="utf-8") as handle:
    json.dump(payload, handle)
PY

      curl -fsS -X POST "${BASE_URL%/}/api/v1/projects" \
        -H "Content-Type: application/json" \
        --data-binary "@${tmp_dir}/project_request.json" > "${tmp_dir}/project.json"
      require_json_data "${tmp_dir}/project.json"
      project_id="$(json_field "${tmp_dir}/project.json" "data.id")"

      curl -fsS -X POST "${BASE_URL%/}/api/v1/projects/${project_id}/datasets/upload" \
        -F "file=@${raster_path}" \
        -F "dataset_type=raster" \
        -F "source=user_upload" \
        -F "dataset_role=source" \
        -F "source_name=manual" \
        -F 'source_metadata_json={"validation_workflow":"operator_hard_negative_detection_matrix","input_kind":"orthophoto_background_candidate"}' \
        -F 'provenance_metadata_json={"operator_supplied":true,"no_external_fetch":true}' \
        > "${tmp_dir}/raster_upload.json"
      require_json_data "${tmp_dir}/raster_upload.json"
      raster_dataset_id="$(json_field "${tmp_dir}/raster_upload.json" "data.id")"

      curl -fsS -X POST "${BASE_URL%/}/api/v1/projects/${project_id}/datasets/${raster_dataset_id}/raster/tile" \
        -H "Content-Type: application/json" \
        -d "{\"tile_size\":${tile_size},\"overlap\":${tile_overlap},\"output_name\":\"hard_negative_detection_tiles\"}" > "${tmp_dir}/tile.json"
      require_json_data "${tmp_dir}/tile.json"
      manifest_path="$(json_field "${tmp_dir}/tile.json" "data.result_json.manifest_path")"

      curl -fsS "${BASE_URL%/}/api/v1/detection/model-assets" > "${tmp_dir}/model_assets.json"
      require_json_data "${tmp_dir}/model_assets.json"
      model_asset_id="$("${PYTHON_BIN}" - "${tmp_dir}/model_assets.json" "${model_env}" <<'PY'
import json
import sys

path, requested = sys.argv[1:3]
with open(path, "r", encoding="utf-8") as handle:
    items = json.load(handle)["data"].get("items") or []
if requested:
    match = next((item for item in items if item.get("model_asset_id") == requested), None)
    if not match:
        raise SystemExit(f"Requested model asset was not found: {requested}")
    print(match["model_asset_id"])
else:
    active = next((item for item in items if item.get("active")), None)
    if not active:
        raise SystemExit("No active local model asset found; set QUALITY_MODEL_ASSET_IDS")
    print(active["model_asset_id"])
PY
)"

      curl -fsS -G "${BASE_URL%/}/api/v1/detection/yolo/preflight" \
        --data-urlencode "tile_manifest_path=${manifest_path}" \
        --data-urlencode "model_asset_id=${model_asset_id}" \
        --data-urlencode "check_model_load=false" > "${tmp_dir}/preflight.json"
      require_json_data "${tmp_dir}/preflight.json"
      "${PYTHON_BIN}" - "${tmp_dir}/preflight.json" <<'PY'
import json
import sys

with open(sys.argv[1], "r", encoding="utf-8") as handle:
    data = json.load(handle)["data"]
if data.get("status") != "ready":
    raise SystemExit(f"YOLO preflight is not ready: {data.get('status')} {data.get('message')}")
if data.get("will_download_models") is not False:
    raise SystemExit("YOLO preflight reported possible model download")
PY

      "${PYTHON_BIN}" - "${tmp_dir}/run_request.json" "${project_id}" "${raster_dataset_id}" "${model_asset_id}" "${manifest_path}" "${threshold}" <<'PY'
import json
import sys

path, project_id, dataset_id, model_asset_id, tile_manifest_path, confidence = sys.argv[1:7]
payload = {
    "project_id": project_id,
    "dataset_id": dataset_id,
    "model_id": "yolo-configured",
    "model_asset_id": model_asset_id,
    "confidence_threshold": float(confidence),
    "tile_manifest_path": tile_manifest_path,
}
with open(path, "w", encoding="utf-8") as handle:
    json.dump(payload, handle)
PY

      curl -fsS -X POST "${BASE_URL%/}/api/v1/detection/run" \
        -H "Content-Type: application/json" \
        --data-binary "@${tmp_dir}/run_request.json" > "${tmp_dir}/detection_run.json"
      require_json_data "${tmp_dir}/detection_run.json"
      analysis_run_id="$(json_field "${tmp_dir}/detection_run.json" "data.analysis_run_id")"
      detection_count="$(json_field "${tmp_dir}/detection_run.json" "data.detection_count")"

      curl -fsS "${BASE_URL%/}/api/v1/detection/runs/${analysis_run_id}/detections" > "${tmp_dir}/detections.json"
      require_json_data "${tmp_dir}/detections.json"
      detections_list_count="$("${PYTHON_BIN}" - "${tmp_dir}/detections.json" <<'PY'
import json
import sys

with open(sys.argv[1], "r", encoding="utf-8") as handle:
    data = json.load(handle)["data"]
print(len(data.get("items") or []))
PY
)"

      tile_count="$(json_field "${tmp_dir}/preflight.json" "data.tile_count")"
      "${PYTHON_BIN}" - \
        "${run_summary}" \
        "${sample_slug}" \
        "${sample_role}" \
        "${allow_empty_reference}" \
        "${reference_feature_count}" \
        "${model_request}" \
        "${model_asset_id}" \
        "${tile_size}" \
        "${tile_overlap}" \
        "${threshold}" \
        "${project_id}" \
        "${raster_dataset_id}" \
        "${manifest_path}" \
        "${analysis_run_id}" \
        "${detection_count}" \
        "${detections_list_count}" \
        "${tile_count}" \
        "${run_log}" <<'PY'
import json
import sys

(
    output_path,
    sample_slug,
    sample_role,
    allow_empty_reference,
    reference_feature_count,
    model_request,
    model_asset_id,
    tile_size,
    tile_overlap,
    threshold,
    project_id,
    raster_dataset_id,
    manifest_path,
    analysis_run_id,
    detection_count,
    detections_list_count,
    tile_count,
    run_log,
) = sys.argv[1:19]

detections = int(detection_count)
listed = int(detections_list_count)
tiles = max(int(tile_count), 1)
if detections != listed:
    raise SystemExit(f"Detection run count {detections} did not match list count {listed}")
summary = {
    "sample_slug": sample_slug,
    "sample_role": sample_role,
    "allow_empty_reference": allow_empty_reference == "True",
    "reference_feature_count": int(reference_feature_count),
    "model_request": model_request,
    "model_asset_id": model_asset_id,
    "tile_size": int(tile_size),
    "tile_overlap": int(tile_overlap),
    "threshold": float(threshold),
    "project_id": project_id,
    "raster_dataset_id": raster_dataset_id,
    "manifest_path": manifest_path,
    "analysis_run_id": analysis_run_id,
    "tile_count": tiles,
    "detection_count": detections,
    "false_positive_pressure": detections / tiles,
    "run_log": run_log,
}
with open(output_path, "w", encoding="utf-8") as handle:
    json.dump(summary, handle, indent=2, sort_keys=True)
print(
    "sample={sample_slug} model={model_asset_id} tile={tile_size} overlap={tile_overlap} "
    "threshold={threshold} detections={detection_count} false_positive_pressure={false_positive_pressure}".format(
        **summary
    )
)
PY
    } >"${run_log}" 2>&1

    cat "${run_log}" | tail -n 1
    rm -rf "${tmp_dir}"
  done < "${matrix_manifest}"
done < "${sample_manifest_tsv}"

"${PYTHON_BIN}" - "${HARD_NEGATIVE_OUTPUT_DIR}" "${BASE_URL}" "${OPERATOR_SAMPLE_MANIFEST_PATH}" <<'PY'
import glob
import json
import sys
from datetime import datetime, timezone
from pathlib import Path

output_dir = Path(sys.argv[1])
base_url = sys.argv[2]
manifest_path = sys.argv[3]

items = []
for path in sorted(glob.glob(str(output_dir / "*" / "*_summary.json"))):
    with open(path, "r", encoding="utf-8") as handle:
        items.append(json.load(handle))

if not items:
    raise SystemExit("No hard-negative detection summaries were produced")


def key_lowest_pressure(item: dict) -> tuple[float, int]:
    return (float(item.get("false_positive_pressure") or 0), int(item.get("detection_count") or 0))


best_by_lowest_pressure = min(items, key=key_lowest_pressure)
summary = {
    "generated_at": datetime.now(timezone.utc).isoformat(),
    "base_url": base_url,
    "operator_sample_manifest_path": manifest_path,
    "sample_count": len({item["sample_slug"] for item in items}),
    "run_count": len(items),
    "best_by_lowest_pressure": best_by_lowest_pressure,
    "items": items,
}
summary_path = output_dir / "hard_negative_matrix_summary.json"
summary_path.write_text(json.dumps(summary, indent=2, sort_keys=True), encoding="utf-8")

print("")
print("Operator hard-negative detection summary")
print("sample\tmodel\ttile\toverlap\tthreshold\tdetections\tfalse_positive_pressure")
for item in items:
    print(
        "{sample_slug}\t{model_asset_id}\t{tile_size}\t{tile_overlap}\t{threshold:.2f}\t{detection_count}\t{false_positive_pressure}".format(
            **item
        )
    )
print("")
print(f"Summary: {summary_path}")
item = best_by_lowest_pressure
print(
    f"best_by_lowest_pressure sample={item['sample_slug']} model={item['model_asset_id']} "
    f"tile={item['tile_size']} overlap={item['tile_overlap']} threshold={item['threshold']:.2f} "
    f"detections={item['detection_count']} false_positive_pressure={item['false_positive_pressure']}"
)
PY