#!/usr/bin/env bash
# The bus as the car will use it: raw frames to several subscribers, a decoder turning
# them into envelopes, several envelope subscribers. Runs on the car Orin against a
# vcan0, beside bus-bench.sh (which makes vcan0). Prints one JSON line per run.
#
#   usage: bus-mix.sh BUS_BENCH_BINARY CAPTURE.log [secs]
#   env:   CASES  "batch_ms signals max_hz" per run, ';'-separated (default: the matrix below)
#
# Loads keep the car's own CAN IDs and their relative rates: the capture as recorded
# (~890 frames/s), and the same capture played 5× faster (~4,450 frames/s, a full bus).
set -euo pipefail

BIN="$1"; CAPTURE="$2"; SECS="${3:-20}"
RAW_SINKS=3        # besides the decoder: svc-gw, telemetry-gw, a debug tool
ENV_SINKS=5        # svc-gw, recorder, svc-nav, svc-vision, telemetry-gw

ip link show vcan0 >/dev/null 2>&1 || { echo "vcan0 missing: run bus-bench.sh once"; exit 1; }
OUT=$(mktemp -d)
FAST="$OUT/fast.log"
awk 'NR == 1 { t0 = substr($1, 2, length($1) - 2) }
     { t = substr($1, 2, length($1) - 2); printf "(%.6f) %s %s\n", t0 + (t - t0) / 5, $2, $3 }' "$CAPTURE" > "$FAST"

ticks() { awk '{print $14 + $15}' "/proc/$1/stat"; }
HZ=$(getconf CLK_TCK)

# run NAME LOG BATCH_MS SIGNALS MAX_HZ
run() {
  local name="$1" log="$2" batch="$3" signals="$4" maxhz="$5" pids=() i
  rm -f "$OUT"/*.json
  for i in $(seq "$RAW_SINKS"); do
    "$BIN" sink --secs "$SECS" > "$OUT/raw$i.json" 2>/dev/null & pids+=($!)
  done
  for i in $(seq "$ENV_SINKS"); do
    "$BIN" sink --topic 'yukart/bench/vehicle/+' --secs "$SECS" > "$OUT/env$i.json" 2>/dev/null & pids+=($!)
  done
  "$BIN" decoder --signals "$signals" --max-hz "$maxhz" --secs $((SECS + 2)) > "$OUT/decoder.json" 2>/dev/null & pids+=($!)
  sleep 0.5
  "$BIN" bridge --batch-ms "$batch" --secs $((SECS + 3)) > "$OUT/bridge.json" 2>/dev/null & pids+=($!)
  sleep 0.5
  local broker m0 m1 t0 t1 loops
  loops=$(( SECS / 2 + 2 ))     # the fast log lasts 2 s; enough loops either way
  broker=$(pgrep -x mosquitto); m0=$(ticks "$broker"); t0=$(date +%s.%N)
  timeout "$((SECS + 1))" canplayer -l "$loops" -I "$log" vcan0=can_pcan >/dev/null 2>&1 || true
  m1=$(ticks "$broker"); t1=$(date +%s.%N)
  wait "${pids[@]}"
  python3 - "$name" "$OUT" "$((m1 - m0))" "$HZ" "$t0" "$t1" <<'PY'
import glob, json, sys
name, out, ticks, hz, t0, t1 = sys.argv[1:]
load = lambda p: json.load(open(p))
raw = [load(p) for p in sorted(glob.glob(f"{out}/raw*.json"))]
env = [load(p) for p in sorted(glob.glob(f"{out}/env*.json"))]
dec, br = load(f"{out}/decoder.json"), load(f"{out}/bridge.json")
worst = lambda xs, k: max(x["latency_ms"][k] for x in xs)
print(json.dumps({
    "run": name,
    "bridge": {k: br[k] for k in ("batch_ms", "frames", "messages", "refused_frames", "kernel_drops", "cpu_pct")},
    "raw_sinks": {"n": len(raw), "frames_per_s": min(x["frames_per_s"] for x in raw),
                  "lost": sum(x["lost_frames"] for x in raw),
                  "p99_ms": worst(raw, "p99"), "max_ms": worst(raw, "max")},
    "decoder": {k: dec[k] for k in ("signals_per_frame", "max_hz", "topics", "published_per_s", "refused", "cpu_pct")},
    "envelope_sinks": {"n": len(env), "per_s": min(x["messages_per_s"] for x in env),
                       "p50_ms": worst(env, "p50"), "p99_ms": worst(env, "p99"), "max_ms": worst(env, "max"),
                       "cpu_pct_each": max(x["cpu_pct"] for x in env)},
    "mosquitto_cpu_pct": round(100 * int(ticks) / int(hz) / (float(t1) - float(t0)), 1)}))
PY
}

CASES="${CASES:-10 3 0;10 8 0;10 8 20;0 8 0}"
for log in "$CAPTURE" "$FAST"; do
  rate=$([ "$log" = "$FAST" ] && echo "full bus" || echo "car")
  IFS=';' read -ra cases <<< "$CASES"
  for c in "${cases[@]}"; do
    read -r batch signals maxhz <<< "$c"
    run "$rate, raw batch ${batch} ms, ${signals} signals/frame, cap ${maxhz} Hz" "$log" "$batch" "$signals" "$maxhz"
  done
done