#!/usr/bin/env bash
# One training cycle on the Spark, start to finish, unattended.
#
#   nohup bash ~/AareLC_train/tools/spark_cycle.sh v5 > ~/cycle_v5.log 2>&1 &
#   bash ~/AareLC_train/tools/spark_cycle.sh v5 --quick            # also the fast settings-A run first
#   MODELS="n s m l" bash ~/AareLC_train/tools/spark_cycle.sh v5   # the promoted recipe at several model sizes
#
# Steps: wait for a running pipeline batch; grow the gold set by its rule (studio `gold`
# stage: gold batches, share, floors); export the reviewed pool as dataset <tag> into this
# repo; stop vLLM; re-measure every registry model on the grown gold set so the gate and
# the comparison stand on the same frames; train the promoted recipe from the base weights,
# one model size after the other; gate each (paired bootstrap against the active model);
# only the nano run may promote itself, a bigger model is a candidate until its speed at the
# beamline is known (its latency on this machine is measured and kept next to the run);
# compare every model on the unseen frames; push; start vLLM again if it was running.
#
# What earlier cycles taught, built in:
#   - a failed gate is a verdict, not a crash: train_run.py is called once per run;
#   - a full-resolution-mask training does not fit next to vLLM in 121 GB: vLLM is stopped
#     while the trainings run and started again on exit if it was up before;
#   - the trainings run one at a time and each waits for MEM_FLOOR_GB of free memory, so the
#     studio server, the inference workers and a reviewers' batch keep their room;
#   - an out-of-memory training is tried once more at half the batch size;
#   - torch's native JIT wants Python headers the Spark lacks: switched off, headers on CPATH;
#   - a grown gold set invalidates stored predictions: everything is re-measured first.
set -eo pipefail
TAG="${1:?usage: spark_cycle.sh <dataset tag, e.g. v5> [--quick]}"; QUICK="${2:-}"
MODELS="${MODELS:-n}"               # yolo26 sizes to train, e.g. "n s m l"
MEM_FLOOR_GB="${MEM_FLOOR_GB:-40}"  # a training starts only with this much memory available
export PATH=$HOME/bin:$PATH
export CPATH=$HOME/local/root/usr/include:$HOME/local/root/usr/include/python3.12
export TORCH_DISABLE_NATIVE_JIT=1
export PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True
PY=$HOME/AareLC_studio/.venv/bin/python
STUDIO=$HOME/AareLC_studio; TRAIN=$HOME/AareLC_train; CFG=pipeline.prod.json
say() { echo "== $(date +%H:%M) $*"; }
avail_gb() { free -g | awk 'NR==2{print $7}'; }

while pgrep -f "[p]ipeline.run auto" >/dev/null; do sleep 30; done   # the reviewers' batch goes first
cd "$STUDIO"; set -a; . ~/.aaredb_env; set +a
say "gold: growing by rule"
$PY -m src.pipeline.run gold --config $CFG 2>&1 | grep -E "^gold" | cut -c1-220 || true
say "export + import into AareLC_train (dataset $TAG)"
$PY -m src.pipeline.run export --config $CFG --dest "data/pipeline_prod/dataset_$(date +%F)_$TAG" --train-repo "$TRAIN" --train-tag "$TAG" 2>&1 | grep -E "^export|dataset |committed|tagged|gold" | cut -c1-220 || true
cd "$TRAIN"
DS=$(git tag -l "dataset/*_$TAG" | tail -1); [ -n "$DS" ] || { echo "no dataset tag for $TAG"; exit 1; }
DSDIR="data/datasets/${DS#dataset/}"
say "dataset $DS: $(ls $DSDIR/images/train | wc -l) train / $(ls $DSDIR/images/val | wc -l) val; gold $(ls data/gold/labels | wc -l) frames"

# Base weights for every size asked for (Ultralytics fetches a missing one from its releases).
for size in $MODELS; do
  base="models/base/yolo26${size}-seg.pt"
  [ -f "$base" ] && continue
  say "fetching $base"
  (cd models/base && $PY -c "from ultralytics import YOLO; YOLO('yolo26${size}-seg.pt')" >/dev/null 2>&1) || { echo "!! could not fetch $base"; exit 1; }
  git add "$base" && git commit -q -m "base weights yolo26${size}-seg" || true
done

VLLM_WAS_UP=0
docker ps --format "{{.Names}}" | grep -q "^vllm$" && VLLM_WAS_UP=1
vllm_back() {
  [ "$VLLM_WAS_UP" = 1 ] || { echo "$(date) vllm was not running before the cycle: left stopped" >> ~/vllm_restart.log; return 0; }
  for i in 1 2 3; do
    docker start vllm >> ~/vllm_restart.log 2>&1 || true; sleep 60
    if docker ps --format "{{.Names}}" | grep -q "^vllm$"; then echo "$(date) vllm up (try $i)" >> ~/vllm_restart.log; return 0; fi
  done
  echo "$(date) vllm did NOT come back" >> ~/vllm_restart.log
}
trap vllm_back EXIT
docker stop vllm >/dev/null 2>&1 || true
say "vLLM stopped (was up: $VLLM_WAS_UP), $(avail_gb) GB available"

say "re-measuring every registry model on the $(ls data/gold/labels | wc -l) gold frames"
for d in models/registry/*/; do
  v=$(basename "$d"); [ -f "$d/weights.pt" ] || continue
  $PY tools/measure_weights.py "$d/weights.pt" --version "$v" 2>&1 | tail -n 1
done
BEAMLINE_PT=$STUDIO/models/best_yolo26n-seg-overlap-false_2026-04-12.pt   # the model before the loop lives in the studio
[ -f "$BEAMLINE_PT" ] && $PY tools/measure_weights.py "$BEAMLINE_PT" --version yolo26n-seg-overlap-false-2026-04-12 2>&1 | tail -n 1
git add models/registry data/gold && git commit -q -m "gold grown to $(ls data/gold/labels | wc -l) frames; every registry model re-measured on it" || true

wait_for_memory() {  # up to two hours for the floor, then go anyway and say so
  local waited=0
  while [ "$(avail_gb)" -lt "$MEM_FLOOR_GB" ] && [ $waited -lt 7200 ]; do sleep 60; waited=$((waited + 60)); done
  [ "$(avail_gb)" -ge "$MEM_FLOOR_GB" ] || echo "   only $(avail_gb) GB available after two hours; starting anyway"
}
run() {  # size, tag suffix, note, extra train_run.py args...
  local size="$1" suffix="$2" note="$3"; shift 3
  local base="models/base/yolo26${size}-seg.pt"
  local id; id=$(python3 tools/ledger.py new --dataset "$DS" --tag "$TAG-$suffix" --base "$base" --note "$note")
  git add runs && git commit -q -m "run $id: exported on $DS" || true
  wait_for_memory
  say "training $id from $base ($(avail_gb) GB available)"
  set +e
  $PY tools/train_run.py "$id" --device 0 --workers 8 "$@" > "runs/$id/train.log" 2>&1 < /dev/null
  local rc=$?            # 0 = gate passed (or no baseline), 2 = gate failed: both are verdicts
  if [ $rc -ne 0 ] && [ $rc -ne 2 ] && grep -aqi "out of memory" "runs/$id/train.log"; then
    say "$id ran out of memory at batch 16; once more at batch 8"
    sleep 60
    $PY tools/train_run.py "$id" --device 0 --workers 8 --batch 8 --force "$@" > "runs/$id/train.log" 2>&1 < /dev/null
    rc=$?
  fi
  set -e
  [ $rc -eq 0 ] || [ $rc -eq 2 ] || echo "!! $id ended with exit code $rc: $(grep -aiE "error|out of memory" runs/$id/train.log | tail -1 | cut -c1-160)"
  local v; v=$(python3 -c "import json;print(json.load(open('runs/$id/run.json')).get('version') or '')")
  if [ -n "$v" ] && [ -d "runs/$id/gold_predictions" ]; then mkdir -p "runs/_measure_$v" && cp -r "runs/$id/gold_predictions" "runs/_measure_$v/"; fi
  python3 -c "import json; m=json.load(open('runs/$id/metrics.json')); g=m.get('gate') or {}; b=g.get('bootstrap') or {}; print('   gate:', g.get('pass'), '| rule:', g.get('rule'), '| F1', g.get('f1_delta'), 'mask', round(g.get('mask_score',0)-g.get('mask_score_baseline',0),4), '| win F1', (b.get('f1') or {}).get('win'), 'win mask', (b.get('mask_score') or {}).get('win'))" 2>/dev/null || true
  # the answer time the live server would see, in PyTorch on this machine (TensorRT at the beamline is faster)
  local best="runs/$id/ultralytics/train/weights/best.pt"
  if [ -f "$best" ]; then
    $PY tools/bench_latency.py "$best" --out "runs/$id/latency.json" 2>/dev/null | sed 's/^/   latency: /' | cut -c1-200 || true
  fi
  python3 tools/ledger.py render >/dev/null 2>&1 || true
  git add runs models && git commit -q -m "run $id: $(python3 -c "import json;print(json.load(open('runs/$id/run.json'))['status'])")" || true
}
[ "$QUICK" = "--quick" ] && run n recipeA "settings A (fliplr 0.5, mask_ratio 4) on dataset $TAG: quick signal, gate only, Spark" --no-promote --set fliplr=0.5 mask_ratio=4
for size in $MODELS; do
  if [ "$size" = "n" ]; then
    run n noflip-mask1 "promoted recipe (fliplr 0, mask_ratio 1) on dataset $TAG: candidate for the beamlines, Spark" --set fliplr=0 mask_ratio=1
  else
    run "$size" "${size}-noflip-mask1" "promoted recipe (fliplr 0, mask_ratio 1) with yolo26${size} on dataset $TAG: gate only until its beamline speed is known, Spark" --no-promote --set fliplr=0 mask_ratio=1
  fi
done

say "every model on the unseen gold frames"
$PY tools/compare_unseen.py 2>&1 | cut -c1-200 || true
git add runs && git commit -q -m "unseen-frame comparison after cycle $TAG" || true
say "active model now: $(python3 -c 'import json;print(json.load(open("models/active/production.json"))["version"])')"
bash ~/sync_push.sh 2>&1 | tail -2
say "cycle $TAG done"