anomaly_score_db

def store_results(self, results: list[tuple[str, float]]) -> None:
    """Batch insert or update prediction results.

    Scores are rounded to fp32 precision before storage.  Existing
    rows with the same filename are replaced (``INSERT OR REPLACE``);
    replacement assigns a new rowid so the replaced row moves to the
    top of the ``updated_desc`` ordering.

    The commit also bumps the ``last_heartbeat`` metadata key. WAL
    growth is bounded passively by ``PRAGMA wal_autocheckpoint``; every
    ``checkpoint_every_n_commits`` commits an opportunistic
    ``PRAGMA wal_checkpoint(TRUNCATE)`` additionally tries to reset the
    .wal file to zero, which only succeeds when no reader holds it open.
    The checkpoint runs *outside* the transaction — it cannot execute
    inside one.

    Args:
        results: Sequence of ``(filename, score)`` tuples.

    Raises:
        sqlite3.DatabaseError: If the score write fails for a non-transient
            reason. A transient WAL-checkpoint lock from a concurrent reader
            is caught (the WAL stays bounded by autocheckpoint), not raised.
    """
    if not results:
        return

    rows = [(fn, _fp32_encode(score)) for fn, score in results]
    now = int(time.time())
    cur = self._conn.cursor()
    cur.executemany(
        "INSERT OR REPLACE INTO results(filename, score) VALUES (?, ?)",
        rows,
    )
    # Heartbeat in the same transaction as the scores — any reader
    # seeing a fresh heartbeat also sees the scores it advertises.
    cur.execute(
        "INSERT OR REPLACE INTO run_metadata(key, value) VALUES (?, ?)",
        (_HEARTBEAT_KEY, json.dumps(now)),
    )
    self._conn.commit()

    self._commits_since_checkpoint += 1
    if self._commits_since_checkpoint >= self._checkpoint_every_n_commits:
        self._commits_since_checkpoint = 0
        # Opportunistic shrink only. wal_autocheckpoint already bounds WAL
        # growth passively, so this TRUNCATE just resets the file to zero
        # when no reader holds it open. Under the live UI poller it usually
        # returns busy=1 and shrinks nothing — harmless, the WAL stays at its
        # bounded high-water mark until a quieter moment. Must never crash
        # the prediction write loop.
        #
        # Critically, drop busy_timeout to 0 for the duration: TRUNCATE
        # respects busy_timeout, so at the default 30 s it blocks the write
        # loop for the *full* 30 s every cadence interval while the gallery
        # poller holds a reader snapshot (measured: ~30 s spikes once per 25
        # commits, ~16% of prediction wall and the gap below peak img/s). At
        # 0 it gives up instantly when a reader is active and still shrinks
        # in the genuine quiet gaps. wal_autocheckpoint bounds the WAL either
        # way, so giving up costs nothing.
        try:
            cur.execute("PRAGMA busy_timeout=0")
            # Row is (busy, log_frames, checkpointed_frames); busy=1 means a
            # reader kept the WAL open so it could not be reset to zero.
            row = cur.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone()
            if row is not None and row[0]:
                logger.debug(
                    "WAL truncate blocked by reader (busy); WAL bounded by autocheckpoint"
                )
        except sqlite3.DatabaseError as exc:
            # A reader can block the WAL restart as either "database is
            # locked" (OperationalError) or "database table is locked"
            # (base DatabaseError). Both are transient — autocheckpoint keeps
            # the WAL bounded regardless. Anything else is a real error.
            if "locked" not in str(exc).lower():
                raise
            logger.debug("WAL checkpoint deferred (reader active): {}", exc)
        finally:
            cur.execute(f"PRAGMA busy_timeout={self._busy_timeout_ms}")

def record_catalogues(self, source_to_catalogue: dict[str, str]) -> None:
    """Record which catalogue file each scored source came from (#506).

    Interns each distinct catalogue path into the ``catalogues`` table and
    points the matching ``results`` rows at it via ``catalogue_idx``, so a
    later gallery load reads only the catalogue that holds a source instead
    of scanning them all.  Sources without a ``results`` row (not scored)
    are silently ignored — only scored sources are ever browsed.

    Args:
        source_to_catalogue: Map from ``source_id`` (the ``results.filename``
            key) to the absolute catalogue path it was found in.  An empty
            map is a no-op; callers only record sources they actually found.
    """
    cur = self._conn.cursor()
    # Intern the distinct paths first; INSERT OR IGNORE keeps existing
    # indices stable so previously-recorded rows stay valid.  One
    # executemany rather than a Python-level loop of single executes.
    distinct_paths = {str(p) for p in source_to_catalogue.values()}
    cur.executemany(
        "INSERT OR IGNORE INTO catalogues(path) VALUES (?)",
        [(path,) for path in distinct_paths],
    )
    path_to_idx = {
        row["path"]: row["idx"] for row in cur.execute("SELECT idx, path FROM catalogues")
    }
    cur.executemany(
        "UPDATE results SET catalogue_idx = ? WHERE filename = ?",
        [(path_to_idx[str(cat)], str(sid)) for sid, cat in source_to_catalogue.items()],
    )
    self._conn.commit()

def get_catalogue_map(self, source_ids: list[str]) -> dict[str, str]:
    """Return the recorded catalogue path for each given scored source (#506).

    Args:
        source_ids: Source ids to look up.

    Returns:
        Map from ``source_id`` to its catalogue path, containing only the
        ids whose ``catalogue_idx`` has been recorded (others fall back to a
        full catalogue scan in the loader).  An empty input yields an empty
        map.
    """
    wanted = [str(s) for s in source_ids]
    rows = self._select_in_chunks(
        "SELECT r.filename AS filename, c.path AS path "
        "FROM results r JOIN catalogues c ON r.catalogue_idx = c.idx "
        "WHERE r.filename IN ({placeholders})",
        wanted,
    )
    return {row["filename"]: row["path"] for row in rows}

def set_metadata(self, key: str, value: Any) -> None:
    """Store a metadata key-value pair (JSON-serialised).

    Args:
        key: Metadata key (e.g. ``"model_path"``).
        value: Arbitrary JSON-serialisable value.
    """
    self._conn.execute(
        "INSERT OR REPLACE INTO run_metadata(key, value) VALUES (?, ?)",
        (key, json.dumps(value)),
    )
    self._conn.commit()

def set_metadata_batch(self, items: dict[str, Any]) -> None:
    """Store multiple metadata key-value pairs in one transaction.

    Args:
        items: Mapping of metadata keys to JSON-serialisable values.
    """
    self._conn.executemany(
        "INSERT OR REPLACE INTO run_metadata(key, value) VALUES (?, ?)",
        [(k, json.dumps(v)) for k, v in items.items()],
    )
    self._conn.commit()

def get_results(
    self,
    sort_by: str = "score_desc",
    limit: int = 100,
    offset: int = 0,
) -> list[dict]:
    """Query results with sorting and pagination.

    Args:
        sort_by: One of ``"score_desc"``, ``"score_asc"``,
            ``"score_mean_dist"``, ``"score_median_dist"``,
            ``"updated_desc"``, ``"updated_asc"``, ``"random"``.
        limit: Maximum number of rows to return.
        offset: Number of rows to skip.

    Returns:
        List of dicts with keys ``filename`` and ``score``.

    Raises:
        ValueError: If *sort_by* is not a recognised sort mode.
    """
    order = _SORT_CLAUSES.get(sort_by)
    if order is None:
        raise ValueError(f"Unknown sort_by={sort_by!r}. Valid options: {list(_SORT_CLAUSES)}")

    # Safe: sort_by is validated against _SORT_CLAUSES allowlist
    rows = self._safe_execute(
        f"""
        SELECT r.filename, r.score
        FROM results r
        ORDER BY {order}
        LIMIT ? OFFSET ?
        """,
        (limit, offset),
    )

    return [
        {
            "filename": row["filename"],
            "score": _fp32_decode(row["score"]),
        }
        for row in rows
    ]

def get_top_results(self, n: int) -> list[dict]:
    """Convenience wrapper: return the *n* highest-scoring results.

    Args:
        n: Number of top results to return.

    Returns:
        List of result dicts sorted by descending score.
    """
    return self.get_results(sort_by="score_desc", limit=n, offset=0)

def get_count(self) -> int:
    """Return total number of stored results.

    Returns:
        Row count of the results table.
    """
    rows = self._safe_execute("SELECT COUNT(*) AS cnt FROM results")
    return rows[0]["cnt"] if rows else 0

def get_result_by_filename(self, filename: str) -> dict | None:
    """Look up a single result by its filename.

    Args:
        filename: The source filename to look up.

    Returns:
        Dict with keys ``filename`` and ``score``, or ``None`` if
        not found.
    """
    rows = self._safe_execute(
        """
        SELECT r.filename, r.score
        FROM results r
        WHERE r.filename = ?
        """,
        (filename,),
    )
    if not rows:
        logger.debug("No result found for filename: %s", filename)
        return None
    row = rows[0]
    return {
        "filename": row["filename"],
        "score": _fp32_decode(row["score"]),
    }

def get_all_scores(self) -> np.ndarray:
    """Return every stored score as a float32 numpy array.

    Loads the entire ``score`` column into memory — not suitable for
    databases with more than a few million rows.

    Returns:
        1-D array of all scores.
    """
    rows = self._safe_execute("SELECT score FROM results")
    if not rows:
        return np.empty(0, dtype=np.float32)
    # Both dtypes are pinned little-endian so the pair can never drift
    # apart.  Native/native would be equally correct — the value arrives
    # from SQLite as a Python int, so there is no on-the-wire byte order to
    # match; mixing the two views is the only thing that corrupts the
    # reinterpretation.
    raw = np.fromiter((r["score"] for r in rows), dtype="<u4", count=len(rows))
    return raw.view("<f4")

def get_score_range(self) -> tuple[float, float]:
    """Return ``(min_score, max_score)`` across all results.

    Returns:
        Tuple of minimum and maximum score values.

    Raises:
        ValueError: If the database contains no results.
    """
    rows = self._safe_execute("SELECT MIN(score) AS lo, MAX(score) AS hi FROM results")
    if not rows or rows[0]["lo"] is None:
        raise ValueError("No results in database")
    return (_fp32_decode(rows[0]["lo"]), _fp32_decode(rows[0]["hi"]))

def get_score_histogram(self, bins: int = 50) -> tuple[np.ndarray, np.ndarray]:
    """Compute a score histogram entirely in SQL.

    Args:
        bins: Number of histogram bins (default 50).

    Returns:
        ``(counts, bin_edges)`` — *counts* has length *bins*,
        *bin_edges* has length ``bins + 1``.  Empty bins have
        count 0.
    """
    # Use SQL to bin scores into [0, 1] range.
    # Scores are stored as fp32 bit patterns; decode via the fp32 UDF
    # before multiplying by the bin count.  Scores exactly equal to
    # 1.0 are clamped into the last bin.
    rows = self._safe_execute(
        """
        SELECT
            CASE
                WHEN CAST(fp32(score) * ? AS INTEGER) >= ? THEN ? - 1
                ELSE CAST(fp32(score) * ? AS INTEGER)
            END AS bin,
            COUNT(*) AS cnt
        FROM results
        GROUP BY bin
        ORDER BY bin
        """,
        (bins, bins, bins, bins),
    )

    counts = np.zeros(bins, dtype=np.int64)
    for row in rows:
        idx = int(row["bin"])
        if 0 <= idx < bins:
            counts[idx] = row["cnt"]

    bin_edges = np.linspace(0.0, 1.0, bins + 1)
    return counts, bin_edges

def is_processed(self, filename: str) -> bool:
    """Check whether *filename* already has a stored result.

    Args:
        filename: Source filename to check.

    Returns:
        ``True`` if a result row exists for this filename.
    """
    row = self._conn.execute("SELECT 1 FROM results WHERE filename = ?", (filename,)).fetchone()
    return row is not None

def get_unprocessed(self, filenames: list[str]) -> list[str]:
    """Filter *filenames* to only those without stored results.

    SELECT-only (no temp tables); input order and any duplicate entries are
    preserved.

    Args:
        filenames: Candidate filenames to check.

    Returns:
        Subset of *filenames* that do not yet appear in the
        results table.
    """
    if not filenames:
        return []

    # SELECT-only so this resume check takes no write lock. The previous
    # temp-table approach (CREATE/INSERT/DELETE) opened an IMMEDIATE write
    # transaction (the connection's isolation_level) and contended with the
    # live scoring writer for the single WAL writer slot — a lock-upgrade
    # collision then returns SQLITE_BUSY immediately, which busy_timeout
    # can't wait out, and killed a prediction chunk.
    rows = self._select_in_chunks(
        "SELECT filename FROM results WHERE filename IN ({placeholders})", filenames
    )
    processed = {row["filename"] for row in rows}
    return [f for f in filenames if f not in processed]

def get_processed_filenames(self) -> set[str]:
    """Return the set of all processed filenames.

    For very large databases this loads all filenames into memory.
    Prefer :meth:`get_unprocessed` for incremental resume checks.

    Returns:
        Set of filenames with stored results.
    """
    rows = self._conn.execute("SELECT filename FROM results").fetchall()
    return {row["filename"] for row in rows}

def validate_compatibility(self, cfg: DotMap) -> tuple[bool, str]:
    """Check whether *cfg* matches the stored run metadata.

    Only keys listed in :data:`_COMPAT_KEYS` are compared.  The
    schema-version entry is always present (written on open) but is
    not part of the compatibility check — a schema mismatch is
    already rejected in :meth:`__init__`.

    ``cfg`` is the prediction config; this method calls
    :func:`build_compat_metadata` internally so all DB-related
    metadata extraction lives next to the DB.  Callers that already
    hold a built dict can use :meth:`set_metadata_batch` directly.

    Args:
        cfg: Prediction config with ``model_path``, ``net``, and
            ``normalisation`` populated.

    Returns:
        ``(is_compatible, message)`` — *message* describes the first
        mismatch found, or is empty on success.  A fresh DB (no
        compat keys stored yet) is always compatible.
    """
    cfg_metadata = build_compat_metadata(cfg)
    stored = self.get_metadata()

    for key, label in _COMPAT_KEYS.items():
        if key not in stored:
            logger.debug("Compatibility key %r not in stored metadata; skipping", key)
            continue
        if stored[key] != cfg_metadata[key]:
            return (
                False,
                f"{label} mismatch: stored={stored[key]!r}, current={cfg_metadata[key]!r}",
            )

    return True, ""

def get_metadata(self) -> dict[str, Any]:
    """Return all stored metadata as a dict.

    Returns:
        Mapping of metadata keys to their JSON-deserialised values.
    """
    rows = self._conn.execute("SELECT key, value FROM run_metadata").fetchall()
    result = {row["key"]: json.loads(row["value"]) for row in rows}
    if not result:
        logger.debug("No metadata found in prediction DB")
    return result

def get_heartbeat(self) -> int | None:
    """Return the unix-epoch timestamp of the last successful commit.

    Refreshed by :meth:`store_results` in the same transaction as
    the scores.  A stale value (more than a few minutes behind the
    wall clock) means the writer subprocess is paused or crashed.

    Returns:
        Integer unix timestamp, or ``None`` if no batch has committed
        since the DB was created.
    """
    rows = self._safe_execute("SELECT value FROM run_metadata WHERE key = ?", (_HEARTBEAT_KEY,))
    if not rows:
        return None
    return int(json.loads(rows[0]["value"]))

def close(self) -> None:
    """Close the database connection.

    Acquires ``_close_lock`` to wait for any in-flight
    ``_safe_execute`` calls to finish, then nulls ``_conn`` so
    subsequent reads return ``[]``.  The actual ``conn.close()``
    runs after the lock is released — at that point no thread can
    hold a reference to the connection.

    On the way out:

    * ``PRAGMA wal_checkpoint(TRUNCATE)`` folds the WAL back into
      the main file so a reopened DB does not carry a ``*.wal``
      leftover from the last run.
    * ``PRAGMA optimize`` lets SQLite regenerate statistics on
      indexes that have changed since they were last ANALYZEd.

    A best-effort wrapper swallows per-pragma errors so a write
    lock held by something else (or a broken connection) can't
    block shutdown.
    """
    with self._close_lock:
        conn = self._conn
        self._conn = None
        if conn is not None:
            for stmt in ("PRAGMA wal_checkpoint(TRUNCATE)", "PRAGMA optimize"):
                try:
                    conn.execute(stmt)
                except sqlite3.Error as exc:
                    logger.debug("{} failed during close: {}", stmt, exc)
    if conn is not None:
        conn.close()
    logger.debug("AnomalyScoreDB closed: {}", self._db_path)