db.py

"""PageInvaders -- SQLite persistence layer.

Replaces the old browser CSV export. Saved reports are written to a single
SQLite file (reports.db) using Python's standard-library sqlite3. There is
NO server and NO network: the database is just a file on disk, created next
to the executable (or this source file when run from source).

Public API:
    init()                 -> ensure reports.db and its tables exist.
    save_run(report)       -> insert one report (+ its per-step rows),
                              returns the new run id.
    db_path()              -> absolute path of the database file.

The `report` dict passed to save_run mirrors the end-of-run report the view
builds. Expected shape:
    {
      "algorithm": str, "frame_count": int, "vpages": int,
      "pattern": str, "tlb_enabled": bool, "write_prob": float,
      "total_accesses": int, "hits": int, "faults": int,
      "hit_ratio": float, "tlb_hit_ratio": float|None, "avg_emat": float,
      "dirty_evictions": int, "clean_evictions": int,
      "steps": [ {"step": int, "vpn": int, "write": int,
                  "result": str, "fault_rate_window": float}, ... ]
    }
"""
from __future__ import annotations

import os
import sqlite3
import sys
from contextlib import contextmanager
from datetime import datetime, timezone

DB_FILENAME = "reports.db"


def _base_dir() -> str:
    """Directory to place reports.db in.

    When frozen by PyInstaller (--onefile), sys.frozen is set and the running
    program lives in a temp dir, so we use the directory of the actual .exe
    (sys.executable) to keep the database next to it and persistent. When run
    from source we use this file's directory.
    """
    if getattr(sys, "frozen", False):
        return os.path.dirname(os.path.abspath(sys.executable))
    return os.path.dirname(os.path.abspath(__file__))


def db_path() -> str:
    return os.path.join(_base_dir(), DB_FILENAME)


def _connect() -> sqlite3.Connection:
    conn = sqlite3.connect(db_path())
    conn.execute("PRAGMA foreign_keys = ON;")
    return conn


@contextmanager
def _connection() -> sqlite3.Connection:
    conn = _connect()
    try:
        with conn:
            yield conn
    finally:
        conn.close()


def init() -> None:
    """Create the schema if it does not already exist. Safe to call on every
    app start."""
    with _connection() as conn:
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS runs (
                id              INTEGER PRIMARY KEY AUTOINCREMENT,
                created_at      TEXT    NOT NULL,
                algorithm       TEXT    NOT NULL,
                frame_count     INTEGER NOT NULL,
                vpages          INTEGER NOT NULL,
                pattern         TEXT    NOT NULL,
                tlb_enabled     INTEGER NOT NULL,
                write_prob      REAL    NOT NULL,
                total_accesses  INTEGER NOT NULL,
                hits            INTEGER NOT NULL,
                faults          INTEGER NOT NULL,
                hit_ratio       REAL    NOT NULL,
                tlb_hit_ratio   REAL,
                avg_emat        REAL    NOT NULL,
                dirty_evictions INTEGER NOT NULL,
                clean_evictions INTEGER NOT NULL
            );
            """
        )
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS steps (
                id                INTEGER PRIMARY KEY AUTOINCREMENT,
                run_id            INTEGER NOT NULL,
                step              INTEGER NOT NULL,
                vpn               INTEGER NOT NULL,
                write             INTEGER NOT NULL,
                result            TEXT    NOT NULL,
                fault_rate_window REAL    NOT NULL,
                FOREIGN KEY (run_id) REFERENCES runs (id) ON DELETE CASCADE
            );
            """
        )
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_steps_run_id ON steps (run_id);"
        )


def save_run(report: dict) -> int:
    """Insert one report and its per-step rows. Returns the new run id."""
    created_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
    with _connection() as conn:
        cur = conn.execute(
            """
            INSERT INTO runs (
                created_at, algorithm, frame_count, vpages, pattern,
                tlb_enabled, write_prob, total_accesses, hits, faults,
                hit_ratio, tlb_hit_ratio, avg_emat, dirty_evictions,
                clean_evictions
            ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
            """,
            (
                created_at,
                report["algorithm"],
                report["frame_count"],
                report["vpages"],
                report["pattern"],
                1 if report["tlb_enabled"] else 0,
                report["write_prob"],
                report["total_accesses"],
                report["hits"],
                report["faults"],
                report["hit_ratio"],
                report.get("tlb_hit_ratio"),
                report["avg_emat"],
                report["dirty_evictions"],
                report["clean_evictions"],
            ),
        )
        run_id = cur.lastrowid
        steps = report.get("steps") or []
        if steps:
            conn.executemany(
                """
                INSERT INTO steps (
                    run_id, step, vpn, write, result, fault_rate_window
                ) VALUES (?,?,?,?,?,?)
                """,
                [
                    (
                        run_id,
                        s["step"],
                        s["vpn"],
                        s["write"],
                        s["result"],
                        s["fault_rate_window"],
                    )
                    for s in steps
                ],
            )
    return run_id


def count_runs() -> int:
    """Convenience: how many reports are stored (used by tests / UI hints)."""
    with _connection() as conn:
        (n,) = conn.execute("SELECT COUNT(*) FROM runs").fetchone()
    return n