code/scripts/continuous_phase_recall_snr_conf_sweep.py

#!/usr/bin/env python3
"""Sweep continuous-pick SNR thresholds and compare phase-wise count-matched confidence recall.

The script builds a reusable SQLite index from hourly continuous-picker JSONL files.
Recall follows the existing deployment diagnostic convention:

* manual P/S labels are restricted to stations/times covered by the waveform index;
* automatic picks are mapped to P or S by phase name;
* each label is matched to the nearest automatic pick at the same station and phase
  inside a search window, and counted as true positive if the residual is within
  the true-positive tolerance.
"""

from __future__ import annotations

import argparse
import csv
import importlib.util
import json
import math
import sqlite3
import sys
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable


ROOT = Path("odata/snr60_pnsn_v3_5120_20190706_20211113_reals_4_2_3_1_1p0_0p1_1p0")
DEFAULT_THRESHOLDS = [float(item) for item in range(0, 11)]
VALIDATION_THRESHOLDS = [4.25]


def load_module(path: Path, name: str):
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"Cannot import {path}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def iter_jsonl(path: Path) -> Iterable[dict[str, Any]]:
    with path.open("r", encoding="utf-8", errors="replace") as handle:
        for line in handle:
            if line.strip():
                yield json.loads(line)


def compact_day_from_epoch(value: float) -> str:
    return datetime.fromtimestamp(float(value), tz=timezone.utc).strftime("%Y%m%d")


def db_exec(conn: sqlite3.Connection, sql: str, params: tuple[Any, ...] = ()) -> None:
    conn.execute(sql, params)


def table_count(conn: sqlite3.Connection, table: str) -> int:
    return int(conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0])


def pick_source_signature(paths: list[Path]) -> list[dict[str, Any]]:
    return [
        {
            "path": str(path),
            "size": path.stat().st_size,
            "mtime_ns": path.stat().st_mtime_ns,
        }
        for path in paths
    ]


def ensure_schema(conn: sqlite3.Connection) -> None:
    conn.executescript(
        """
        PRAGMA journal_mode=WAL;
        PRAGMA synchronous=NORMAL;
        PRAGMA temp_store=MEMORY;

        CREATE TABLE IF NOT EXISTS meta (
            key TEXT PRIMARY KEY,
            value TEXT NOT NULL
        );

        CREATE TABLE IF NOT EXISTS picks (
            id INTEGER PRIMARY KEY,
            day TEXT NOT NULL,
            station_id TEXT NOT NULL,
            phase TEXT NOT NULL,
            phase_name TEXT,
            time REAL NOT NULL,
            phase_prob REAL,
            snr_ratio REAL,
            snr_db REAL,
            conf_rank_global INTEGER,
            conf_rank_phase INTEGER
        );

        CREATE TABLE IF NOT EXISTS labels (
            id INTEGER PRIMARY KEY,
            day TEXT NOT NULL,
            station_id TEXT NOT NULL,
            phase TEXT NOT NULL,
            time REAL NOT NULL,
            event_id TEXT,
            status TEXT
        );
        """
    )


def invalidate_index_if_needed(
    conn: sqlite3.Connection,
    *,
    pick_files: list[Path],
    label_json: Path,
    waveform_db: Path,
    days: list[str],
    force: bool,
) -> bool:
    signature = {
        "pick_files": pick_source_signature(pick_files),
        "label_json": str(label_json),
        "label_size": label_json.stat().st_size,
        "label_mtime_ns": label_json.stat().st_mtime_ns,
        "waveform_db": str(waveform_db),
        "waveform_db_size": waveform_db.stat().st_size,
        "waveform_db_mtime_ns": waveform_db.stat().st_mtime_ns,
        "days": days,
    }
    existing = conn.execute("SELECT value FROM meta WHERE key='source_signature'").fetchone()
    existing_value = None if existing is None else json.loads(existing[0])
    has_tables = table_count(conn, "picks") > 0 and table_count(conn, "labels") > 0
    rebuild = force or (existing_value != signature) or not has_tables
    if rebuild:
        conn.executescript(
            """
            DROP TABLE IF EXISTS picks;
            DROP TABLE IF EXISTS labels;
            DROP TABLE IF EXISTS meta;
            """
        )
        ensure_schema(conn)
        conn.execute(
            "INSERT OR REPLACE INTO meta(key, value) VALUES('source_signature', ?)",
            (json.dumps(signature, ensure_ascii=False, sort_keys=True),),
        )
    return rebuild


def build_pick_index(
    conn: sqlite3.Connection,
    *,
    eval_mod: Any,
    pick_files: list[Path],
    days: set[str],
    batch_size: int,
) -> dict[str, Any]:
    auto_counts = Counter()
    inserted = 0
    batch: list[tuple[Any, ...]] = []
    insert_sql = """
        INSERT INTO picks(day, station_id, phase, phase_name, time, phase_prob, snr_ratio, snr_db)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    """
    for path in pick_files:
        before_file = inserted + len(batch)
        for record in iter_jsonl(path):
            if record.get("record_type") != "phase_pick":
                continue
            phase_name = str(record.get("phase_name") or "")
            phase = eval_mod.AUTO_TO_LABEL_PHASE.get(phase_name)
            if phase is None:
                continue
            try:
                time_epoch = eval_mod.epoch(record["phase_time"])
            except Exception:
                continue
            day = compact_day_from_epoch(time_epoch)
            if day not in days:
                continue
            station_info = record.get("station_info") or {}
            station_id = eval_mod.norm_station_id(
                record.get("station_id") or station_info.get("station_id"),
                station_info.get("network"),
                station_info.get("station"),
                station_info.get("location"),
            )
            try:
                snr_ratio = float(record.get("snr"))
            except (TypeError, ValueError):
                snr_ratio = None
            snr_db = 10.0 * math.log10(snr_ratio) if snr_ratio and snr_ratio > 0.0 else None
            try:
                phase_prob = float(record.get("phase_prob"))
            except (TypeError, ValueError):
                phase_prob = None
            batch.append(
                (
                    day,
                    station_id,
                    phase,
                    phase_name,
                    time_epoch,
                    phase_prob,
                    snr_ratio,
                    snr_db,
                )
            )
            auto_counts[phase_name] += 1
            if len(batch) >= batch_size:
                conn.executemany(insert_sql, batch)
                inserted += len(batch)
                batch.clear()
        after_file = inserted + len(batch)
        print(
            f"[index] {path.name}: +{after_file - before_file:,} picks "
            f"({after_file:,} total pending commit)",
            file=sys.stderr,
            flush=True,
        )
    if batch:
        conn.executemany(insert_sql, batch)
        inserted += len(batch)
    conn.commit()
    return {"inserted_pick_count": inserted, "auto_counts_by_phase_name": dict(auto_counts)}


def build_label_index(
    conn: sqlite3.Connection,
    *,
    eval_mod: Any,
    label_json: Path,
    waveform_db: Path,
    days: set[str],
    batch_size: int,
) -> dict[str, Any]:
    coverage = eval_mod.Coverage(waveform_db)
    labels = []
    counts = Counter()
    insert_sql = """
        INSERT INTO labels(day, station_id, phase, time, event_id, status)
        VALUES (?, ?, ?, ?, ?, ?)
    """
    for label in eval_mod.iter_label_picks(label_json, None, None, coverage):
        day = compact_day_from_epoch(float(label["time"]))
        if day not in days:
            continue
        labels.append(
            (
                day,
                label["station_id"],
                label["phase"],
                float(label["time"]),
                label.get("event_id"),
                label.get("status"),
            )
        )
        counts[str(label["phase"])] += 1
        if len(labels) >= batch_size:
            conn.executemany(insert_sql, labels)
            labels.clear()
    if labels:
        conn.executemany(insert_sql, labels)
    conn.commit()
    print(f"[index] labels with waveform coverage: {sum(counts.values()):,}", file=sys.stderr, flush=True)
    return {
        "waveform_coverage_filter": coverage.enabled(),
        "label_counts_by_phase": dict(counts),
    }


def add_indexes_and_ranks(conn: sqlite3.Connection) -> None:
    print("[index] creating lookup indexes", file=sys.stderr, flush=True)
    conn.executescript(
        """
        CREATE INDEX IF NOT EXISTS idx_picks_station_phase_time
            ON picks(station_id, phase, time);
        CREATE INDEX IF NOT EXISTS idx_picks_snr_phase
            ON picks(phase, snr_db);
        CREATE INDEX IF NOT EXISTS idx_picks_prob_global
            ON picks(phase_prob DESC, id ASC);
        CREATE INDEX IF NOT EXISTS idx_picks_prob_phase
            ON picks(phase, phase_prob DESC, id ASC);
        CREATE INDEX IF NOT EXISTS idx_labels_station_phase_time
            ON labels(station_id, phase, time);
        CREATE INDEX IF NOT EXISTS idx_labels_phase
            ON labels(phase);
        """
    )
    conn.commit()

    ranks_present = conn.execute(
        "SELECT COUNT(*) FROM picks WHERE conf_rank_global IS NOT NULL AND conf_rank_phase IS NOT NULL"
    ).fetchone()[0]
    if int(ranks_present) != table_count(conn, "picks"):
        print("[index] computing confidence ranks", file=sys.stderr, flush=True)
        conn.execute("DROP TABLE IF EXISTS pick_ranks")
        conn.execute(
            """
            CREATE TEMP TABLE pick_ranks AS
            SELECT
                id,
                ROW_NUMBER() OVER (ORDER BY phase_prob DESC, id ASC) AS conf_rank_global,
                ROW_NUMBER() OVER (PARTITION BY phase ORDER BY phase_prob DESC, id ASC) AS conf_rank_phase
            FROM picks
            """
        )
        conn.execute("CREATE INDEX pick_ranks_id ON pick_ranks(id)")
        conn.execute(
            """
            UPDATE picks
            SET
                conf_rank_global = (SELECT conf_rank_global FROM pick_ranks WHERE pick_ranks.id = picks.id),
                conf_rank_phase = (SELECT conf_rank_phase FROM pick_ranks WHERE pick_ranks.id = picks.id)
            """
        )
        conn.execute("DROP TABLE pick_ranks")
        conn.commit()
    print("[index] creating rank indexes", file=sys.stderr, flush=True)
    conn.executescript(
        """
        CREATE INDEX IF NOT EXISTS idx_picks_rank_global
            ON picks(conf_rank_global);
        CREATE INDEX IF NOT EXISTS idx_picks_rank_phase
            ON picks(phase, conf_rank_phase);
        """
    )
    conn.commit()


def ensure_index(
    db_path: Path,
    *,
    eval_script: Path,
    pick_dir: Path,
    label_json: Path,
    waveform_db: Path,
    days: list[str],
    batch_size: int,
    force: bool,
) -> tuple[sqlite3.Connection, dict[str, Any]]:
    db_path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(db_path))
    conn.row_factory = sqlite3.Row
    ensure_schema(conn)
    pick_files = sorted(pick_dir.glob("*.phase.jsonl"))
    if not pick_files:
        raise FileNotFoundError(f"No phase JSONL files found in {pick_dir}")
    eval_mod = load_module(eval_script, "pick_eval_no_nms")
    rebuild = invalidate_index_if_needed(
        conn,
        pick_files=pick_files,
        label_json=label_json,
        waveform_db=waveform_db,
        days=days,
        force=force,
    )
    index_summary: dict[str, Any] = {
        "db_path": str(db_path),
        "pick_dir": str(pick_dir),
        "pick_file_count": len(pick_files),
        "label_json": str(label_json),
        "waveform_db": str(waveform_db),
        "days": days,
        "rebuilt": rebuild,
    }
    if rebuild:
        with conn:
            index_summary.update(
                build_pick_index(
                    conn,
                    eval_mod=eval_mod,
                    pick_files=pick_files,
                    days=set(days),
                    batch_size=batch_size,
                )
            )
            index_summary.update(
                build_label_index(
                    conn,
                    eval_mod=eval_mod,
                    label_json=label_json,
                    waveform_db=waveform_db,
                    days=set(days),
                    batch_size=batch_size,
                )
            )
    add_indexes_and_ranks(conn)
    index_summary["pick_count"] = table_count(conn, "picks")
    index_summary["label_count"] = table_count(conn, "labels")
    index_summary["pick_counts_by_phase"] = {
        row["phase"]: row["n"]
        for row in conn.execute("SELECT phase, COUNT(*) AS n FROM picks GROUP BY phase")
    }
    index_summary["label_counts_by_phase"] = {
        row["phase"]: row["n"]
        for row in conn.execute("SELECT phase, COUNT(*) AS n FROM labels GROUP BY phase")
    }
    return conn, index_summary


def count_snr_picks(conn: sqlite3.Connection, threshold: float) -> dict[str, int]:
    rows = conn.execute(
        "SELECT phase, COUNT(*) AS n FROM picks WHERE snr_db >= ? GROUP BY phase",
        (threshold,),
    ).fetchall()
    counts = {"P": 0, "S": 0}
    counts.update({row["phase"]: int(row["n"]) for row in rows})
    return counts


def count_all_picks(conn: sqlite3.Connection) -> dict[str, int]:
    rows = conn.execute("SELECT phase, COUNT(*) AS n FROM picks GROUP BY phase").fetchall()
    counts = {"P": 0, "S": 0}
    counts.update({row["phase"]: int(row["n"]) for row in rows})
    return counts


def recall_query_sql(filter_sql: str, restrict_phase: bool) -> str:
    phase_clause = "WHERE l.phase = :phase" if restrict_phase else ""
    return f"""
        WITH candidate AS (
            SELECT
                l.id AS label_id,
                l.phase AS phase,
                p.time - l.time AS residual_s,
                ABS(p.time - l.time) AS abs_residual_s,
                ROW_NUMBER() OVER (
                    PARTITION BY l.id
                    ORDER BY ABS(p.time - l.time), p.id
                ) AS rn
            FROM labels AS l
            JOIN picks AS p
              ON p.station_id = l.station_id
             AND p.phase = l.phase
             AND p.time BETWEEN l.time - :search_window_s AND l.time + :search_window_s
            WHERE {filter_sql}
        ),
        nearest AS (
            SELECT label_id, phase, residual_s
            FROM candidate
            WHERE rn = 1
        )
        SELECT
            l.phase AS phase,
            COUNT(*) AS n_label,
            SUM(CASE WHEN nearest.residual_s IS NOT NULL
                      AND ABS(nearest.residual_s) <= :tp_tol_s
                     THEN 1 ELSE 0 END) AS n_tp,
            COUNT(nearest.label_id) AS n_with_candidate
        FROM labels AS l
        LEFT JOIN nearest ON nearest.label_id = l.id
        {phase_clause}
        GROUP BY l.phase
        ORDER BY l.phase
    """


def summarize_recall_rows(rows: list[sqlite3.Row]) -> dict[str, dict[str, Any]]:
    out = {"P": {"n_label": 0, "n_tp": 0, "n_with_candidate": 0, "recall": None},
           "S": {"n_label": 0, "n_tp": 0, "n_with_candidate": 0, "recall": None}}
    for row in rows:
        n_label = int(row["n_label"])
        n_tp = int(row["n_tp"] or 0)
        out[row["phase"]] = {
            "n_label": n_label,
            "n_tp": n_tp,
            "n_with_candidate": int(row["n_with_candidate"] or 0),
            "recall": n_tp / n_label if n_label else None,
        }
    return out


def evaluate_selection(
    conn: sqlite3.Connection,
    *,
    filter_sql: str,
    params: dict[str, Any],
    restrict_phase: str | None,
    tp_tol_s: float,
    search_window_s: float,
) -> dict[str, dict[str, Any]]:
    all_params = {
        **params,
        "tp_tol_s": tp_tol_s,
        "search_window_s": search_window_s,
    }
    if restrict_phase is not None:
        all_params["phase"] = restrict_phase
    rows = conn.execute(
        recall_query_sql(filter_sql, restrict_phase is not None),
        all_params,
    ).fetchall()
    return summarize_recall_rows(rows)


def append_result_rows(
    rows: list[dict[str, Any]],
    *,
    threshold: float,
    threshold_label: str,
    selection: str,
    confidence_scope: str | None,
    auto_counts: dict[str, int],
    recall: dict[str, dict[str, Any]],
) -> None:
    for phase in ("P", "S"):
        phase_recall = recall[phase]
        rows.append(
            {
                "threshold_db": threshold,
                "threshold_label": threshold_label,
                "selection": selection,
                "confidence_scope": confidence_scope or "",
                "phase": phase,
                "auto_count_phase": auto_counts.get(phase, 0),
                "auto_count_total": auto_counts.get("P", 0) + auto_counts.get("S", 0),
                "n_label": phase_recall["n_label"],
                "n_tp": phase_recall["n_tp"],
                "n_with_candidate": phase_recall["n_with_candidate"],
                "recall": phase_recall["recall"],
            }
        )


def run_sweep(
    conn: sqlite3.Connection,
    thresholds: list[float],
    *,
    tp_tol_s: float,
    search_window_s: float,
) -> list[dict[str, Any]]:
    results: list[dict[str, Any]] = []
    full_counts = count_all_picks(conn)
    print("[sweep] full, no SNR filter", file=sys.stderr, flush=True)
    full_recall = evaluate_selection(
        conn,
        filter_sql="1 = 1",
        params={},
        restrict_phase=None,
        tp_tol_s=tp_tol_s,
        search_window_s=search_window_s,
    )
    append_result_rows(
        results,
        threshold=-1.0,
        threshold_label="Full",
        selection="snr",
        confidence_scope=None,
        auto_counts=full_counts,
        recall=full_recall,
    )
    for scope, label_filter, params in [
        (
            "phase_count_matched",
            """
                (
                    (p.phase = 'P' AND p.conf_rank_phase <= :n_p)
                    OR
                    (p.phase = 'S' AND p.conf_rank_phase <= :n_s)
                )
            """,
            {"n_p": full_counts["P"], "n_s": full_counts["S"]},
        ),
        (
            "global_count_matched",
            "p.conf_rank_global <= :n_total",
            {"n_total": full_counts["P"] + full_counts["S"]},
        ),
    ]:
        recall = evaluate_selection(
            conn,
            filter_sql=label_filter,
            params=params,
            restrict_phase=None,
            tp_tol_s=tp_tol_s,
            search_window_s=search_window_s,
        )
        append_result_rows(
            results,
            threshold=-1.0,
            threshold_label="Full",
            selection="confidence",
            confidence_scope=scope,
            auto_counts=full_counts,
            recall=recall,
        )

    for threshold in thresholds:
        print(f"[sweep] threshold {threshold:g} dB", file=sys.stderr, flush=True)
        snr_counts = count_snr_picks(conn, threshold)
        snr_recall = evaluate_selection(
            conn,
            filter_sql="p.snr_db >= :threshold",
            params={"threshold": threshold},
            restrict_phase=None,
            tp_tol_s=tp_tol_s,
            search_window_s=search_window_s,
        )
        append_result_rows(
            results,
            threshold=threshold,
            threshold_label=f"{threshold:g}",
            selection="snr",
            confidence_scope=None,
            auto_counts=snr_counts,
            recall=snr_recall,
        )

        phase_conf_recall = evaluate_selection(
            conn,
            filter_sql="""
                (
                    (p.phase = 'P' AND p.conf_rank_phase <= :n_p)
                    OR
                    (p.phase = 'S' AND p.conf_rank_phase <= :n_s)
                )
            """,
            params={"n_p": snr_counts["P"], "n_s": snr_counts["S"]},
            restrict_phase=None,
            tp_tol_s=tp_tol_s,
            search_window_s=search_window_s,
        )
        append_result_rows(
            results,
            threshold=threshold,
            threshold_label=f"{threshold:g}",
            selection="confidence",
            confidence_scope="phase_count_matched",
            auto_counts=snr_counts,
            recall=phase_conf_recall,
        )

        total_count = snr_counts["P"] + snr_counts["S"]
        global_conf_recall = evaluate_selection(
            conn,
            filter_sql="p.conf_rank_global <= :n_total",
            params={"n_total": total_count},
            restrict_phase=None,
            tp_tol_s=tp_tol_s,
            search_window_s=search_window_s,
        )
        append_result_rows(
            results,
            threshold=threshold,
            threshold_label=f"{threshold:g}",
            selection="confidence",
            confidence_scope="global_count_matched",
            auto_counts=snr_counts,
            recall=global_conf_recall,
        )
    return results


def write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
    fieldnames = [
        "threshold_db",
        "threshold_label",
        "selection",
        "confidence_scope",
        "phase",
        "auto_count_phase",
        "auto_count_total",
        "n_label",
        "n_tp",
        "n_with_candidate",
        "recall",
    ]
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)


def plot_results(
    rows: list[dict[str, Any]],
    out_png: Path,
    out_pdf: Path,
    *,
    confidence_scope: str,
    confidence_label: str,
) -> None:
    import matplotlib.pyplot as plt

    plot_rows = [
        row
        for row in rows
        if (row.get("threshold_label") == "Full" or float(row["threshold_db"]).is_integer())
        and row["selection"] in {"snr", "confidence"}
        and row["confidence_scope"] in {"", confidence_scope}
    ]
    colors = {"P": "#2f6f9f", "S": "#b45f3c"}
    fig, axes = plt.subplots(1, 2, figsize=(8.0, 3.3), sharey=True)
    for ax, phase in zip(axes, ("P", "S"), strict=True):
        phase_rows = [row for row in plot_rows if row["phase"] == phase]
        for selection, scope, label, linestyle, marker in [
            ("snr", "", "SNR filter", "-", "o"),
            ("confidence", confidence_scope, confidence_label, "--", "s"),
        ]:
            series = sorted(
                [
                    row
                    for row in phase_rows
                    if row["selection"] == selection and row["confidence_scope"] == scope
                ],
                key=lambda item: float(item["threshold_db"]),
            )
            ax.plot(
                [float(row["threshold_db"]) for row in series],
                [100.0 * float(row["recall"]) for row in series],
                linestyle=linestyle,
                marker=marker,
                markersize=4,
                linewidth=1.8,
                color=colors[phase],
                alpha=1.0 if selection == "snr" else 0.72,
                label=label,
            )
        for marker_x in (5.0, 10.0):
            ax.axvline(marker_x, color="#8a8a8a", linewidth=0.8, linestyle=":", zorder=0)
        ax.set_title(f"{phase}-phase recall")
        ax.set_xlabel("SNR threshold (dB)")
        ax.set_xlim(-1.25, 10.2)
        ax.set_xticks([-1, 0, 2, 4, 6, 8, 10])
        ax.set_xticklabels(["Full", "0", "2", "4", "6", "8", "10"])
        ax.grid(True, axis="y", color="#dddddd", linewidth=0.7)
        ax.spines["top"].set_visible(False)
        ax.spines["right"].set_visible(False)
    axes[0].set_ylabel("Recall of covered manual picks (%)")
    axes[0].legend(frameon=False, loc="lower left", fontsize=8)
    fig.suptitle("Continuous phase-pick recall under SNR vs count-matched confidence filtering", y=1.02)
    fig.tight_layout()
    out_png.parent.mkdir(parents=True, exist_ok=True)
    fig.savefig(out_png, dpi=300, bbox_inches="tight")
    fig.savefig(out_pdf, bbox_inches="tight")
    plt.close(fig)


def parse_thresholds(values: list[str] | None) -> list[float]:
    if not values:
        return DEFAULT_THRESHOLDS
    thresholds: list[float] = []
    for value in values:
        if ":" in value:
            start, stop, step = [float(item) for item in value.split(":", 2)]
            current = start
            while current <= stop + step * 1e-6:
                thresholds.append(round(current, 6))
                current += step
        else:
            thresholds.append(float(value))
    return sorted(dict.fromkeys(thresholds))


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, default=ROOT)
    parser.add_argument("--pick-dir", type=Path, default=None)
    parser.add_argument("--label-json", type=Path, default=Path("data/datax/data/label/annotations_for_continuous_hdf5.json"))
    parser.add_argument("--waveform-db", type=Path, default=Path("data/datax/data/index/waveform_index.sqlite"))
    parser.add_argument("--eval-script", type=Path, default=Path("odata/evaluate_pick_recall_no_nms.py"))
    parser.add_argument("--outdir", type=Path, default=Path("outputs/continuous_phase_recall_snr_conf_sweep"))
    parser.add_argument("--db", type=Path, default=None)
    parser.add_argument("--days", nargs="+", default=["20190706", "20211113"])
    parser.add_argument("--thresholds", nargs="*", default=None, help="Values or start:stop:step, default 0..10.")
    parser.add_argument("--include-validation-thresholds", action="store_true")
    parser.add_argument("--tp-tol-s", type=float, default=1.5)
    parser.add_argument("--search-window-s", type=float, default=5.0)
    parser.add_argument("--batch-size", type=int, default=20000)
    parser.add_argument("--force-rebuild", action="store_true")
    args = parser.parse_args()

    pick_dir = args.pick_dir or (args.root / "hourly" / "all")
    db_path = args.db or (args.outdir / "pick_recall_sweep.sqlite")
    thresholds = parse_thresholds(args.thresholds)
    if args.include_validation_thresholds:
        thresholds = sorted(dict.fromkeys([*thresholds, *VALIDATION_THRESHOLDS]))

    conn, index_summary = ensure_index(
        db_path,
        eval_script=args.eval_script,
        pick_dir=pick_dir,
        label_json=args.label_json,
        waveform_db=args.waveform_db,
        days=args.days,
        batch_size=args.batch_size,
        force=args.force_rebuild,
    )
    try:
        rows = run_sweep(
            conn,
            thresholds,
            tp_tol_s=args.tp_tol_s,
            search_window_s=args.search_window_s,
        )
    finally:
        conn.close()

    args.outdir.mkdir(parents=True, exist_ok=True)
    csv_path = args.outdir / "continuous_phase_recall_snr_conf_sweep.csv"
    summary_path = args.outdir / "continuous_phase_recall_snr_conf_sweep_summary.json"
    png_path = args.outdir / "continuous_phase_recall_snr_conf_sweep.png"
    pdf_path = args.outdir / "continuous_phase_recall_snr_conf_sweep.pdf"
    global_png_path = args.outdir / "continuous_phase_recall_snr_conf_sweep_global_count_matched.png"
    global_pdf_path = args.outdir / "continuous_phase_recall_snr_conf_sweep_global_count_matched.pdf"
    write_csv(csv_path, rows)
    plot_results(
        rows,
        png_path,
        pdf_path,
        confidence_scope="phase_count_matched",
        confidence_label="Confidence, same phase count",
    )
    plot_results(
        rows,
        global_png_path,
        global_pdf_path,
        confidence_scope="global_count_matched",
        confidence_label="Confidence, same total count",
    )
    summary = {
        "index": index_summary,
        "thresholds_db": thresholds,
        "matching": {
            "tp_tolerance_s": args.tp_tol_s,
            "search_window_s": args.search_window_s,
            "label_denominator": "manual P/S labels with waveform coverage",
            "snr_db_definition": "10*log10(JSONL snr ratio)",
            "confidence_phase_count_matched": "for each phase, keep the same number of P or S picks as that phase's SNR filter, sorted by phase_prob",
            "confidence_global_count_matched": "keep the same total P+S count as the SNR filter, sorted by phase_prob; included for comparability with earlier deployment diagnostic",
        },
        "outputs": {
            "csv": str(csv_path),
            "png": str(png_path),
            "pdf": str(pdf_path),
            "global_count_matched_png": str(global_png_path),
            "global_count_matched_pdf": str(global_pdf_path),
            "sqlite": str(db_path),
        },
        "rows": rows,
    }
    summary_path.write_text(json.dumps(summary, indent=2, ensure_ascii=False), encoding="utf-8")
    print(json.dumps({"summary": str(summary_path), "csv": str(csv_path), "png": str(png_path), "pdf": str(pdf_path)}, indent=2))


if __name__ == "__main__":
    main()