src/database.py

# backend/src/database.py

import sqlite3
from pathlib import Path
from datetime import datetime, timezone

DB_PATH = Path(__file__).resolve().parent.parent / "parking.db"


def get_connection() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH, timeout=30, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    # WAL mode: allows concurrent reads alongside writes (avoids "database is locked")
    conn.execute("PRAGMA journal_mode=WAL")
    # Wait up to 10 s for any lock before raising an error
    conn.execute("PRAGMA busy_timeout=10000")
    return conn


def init_db() -> None:
    """Create tables if they do not already exist."""
    conn = get_connection()
    cur = conn.cursor()

    cur.executescript("""

        CREATE TABLE IF NOT EXISTS occupancy_log (

            id          INTEGER PRIMARY KEY AUTOINCREMENT,

            slot_id     TEXT    NOT NULL,

            status      TEXT    NOT NULL CHECK(status IN ('occupied', 'empty')),

            confidence  REAL    NOT NULL,

            logged_at   TEXT    NOT NULL

        );



        CREATE TABLE IF NOT EXISTS predictions (

            id              INTEGER PRIMARY KEY AUTOINCREMENT,

            slot_id         TEXT    NOT NULL,

            horizon_minutes INTEGER NOT NULL,

            vacancy_prob    REAL    NOT NULL,

            predicted_at    TEXT    NOT NULL

        );



        CREATE INDEX IF NOT EXISTS idx_occupancy_slot

            ON occupancy_log(slot_id, logged_at);



        CREATE INDEX IF NOT EXISTS idx_predictions_slot

            ON predictions(slot_id, predicted_at);

    """)

    conn.commit()
    conn.close()


def log_occupancy(records: list[dict]) -> None:
    """

    Insert a batch of occupancy readings.



    Each record must have:

        slot_id    : str

        status     : 'occupied' | 'empty'

        confidence : float

    """
    now = datetime.now(timezone.utc).isoformat()
    rows = [(r["slot_id"], r["status"], r["confidence"], now) for r in records]

    conn = get_connection()
    conn.executemany(
        "INSERT INTO occupancy_log (slot_id, status, confidence, logged_at) "
        "VALUES (?, ?, ?, ?)",
        rows,
    )
    conn.commit()
    conn.close()


def get_latest_occupancy() -> list[dict]:
    """

    Return the most recent status for every slot.

    Uses a subquery to pick the MAX logged_at per slot_id.

    """
    conn = get_connection()
    rows = conn.execute("""

        SELECT ol.slot_id, ol.status, ol.confidence, ol.logged_at

        FROM   occupancy_log ol

        INNER JOIN (

            SELECT slot_id, MAX(logged_at) AS max_ts

            FROM   occupancy_log

            GROUP  BY slot_id

        ) latest ON ol.slot_id = latest.slot_id

                 AND ol.logged_at = latest.max_ts

        ORDER BY ol.slot_id

    """).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_full_history(limit: int = 500, offset: int = 0) -> list[dict]:
    """Return a paginated slice of occupancy_log, ordered by time."""
    conn = get_connection()
    rows = conn.execute(
        "SELECT slot_id, status, confidence, logged_at "
        "FROM   occupancy_log "
        "ORDER  BY logged_at DESC "
        "LIMIT  ? OFFSET ?",
        (limit, offset),
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def save_predictions(records: list[dict]) -> None:
    """

    Insert Prophet forecast results.



    Each record must have:

        slot_id         : str

        horizon_minutes : int

        vacancy_prob    : float

    """
    now = datetime.now(timezone.utc).isoformat()
    rows = [
        (r["slot_id"], r["horizon_minutes"], r["vacancy_prob"], now)
        for r in records
    ]
    conn = get_connection()
    conn.executemany(
        "INSERT INTO predictions (slot_id, horizon_minutes, vacancy_prob, predicted_at) "
        "VALUES (?, ?, ?, ?)",
        rows,
    )
    conn.commit()
    conn.close()


def get_latest_predictions(horizon_minutes: int) -> list[dict]:
    """Return the most recent forecast for each slot at the requested horizon."""
    conn = get_connection()
    rows = conn.execute("""

        SELECT p.slot_id, p.vacancy_prob, p.predicted_at

        FROM   predictions p

        INNER JOIN (

            SELECT slot_id, MAX(predicted_at) AS max_ts

            FROM   predictions

            WHERE  horizon_minutes = ?

            GROUP  BY slot_id

        ) latest ON p.slot_id   = latest.slot_id

                 AND p.predicted_at = latest.max_ts

        WHERE p.horizon_minutes = ?

        ORDER BY p.slot_id

    """, (horizon_minutes, horizon_minutes)).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_analytics_summary() -> dict:
    """

    Return aggregated analytics without sending raw rows.

    Computes:

      - total_readings: total rows in occupancy_log

      - avg_occupancy_pct: overall average occupancy percentage

      - peak_hour: 0-23 hour with highest average occupancy

      - busiest_slot: slot_id with most occupied readings

      - hourly_trend: list of {hour, occupied, empty} for last 48 buckets

    """
    conn = get_connection()

    # Total readings
    total = conn.execute("SELECT COUNT(*) AS cnt FROM occupancy_log").fetchone()["cnt"]

    # Average occupancy %
    avg_row = conn.execute("""

        SELECT ROUND(100.0 * SUM(CASE WHEN status = 'occupied' THEN 1 ELSE 0 END)

               / NULLIF(COUNT(*), 0), 1) AS avg_pct

        FROM occupancy_log

    """).fetchone()
    avg_pct = float(avg_row["avg_pct"]) if avg_row["avg_pct"] is not None else 0.0

    # Peak hour (0-23) — SQLite strftime extracts UTC hour
    peak_row = conn.execute("""

        SELECT CAST(strftime('%H', logged_at) AS INTEGER) AS hr,

               SUM(CASE WHEN status = 'occupied' THEN 1 ELSE 0 END) AS occ_count

        FROM   occupancy_log

        GROUP  BY hr

        ORDER  BY occ_count DESC

        LIMIT  1

    """).fetchone()
    peak_hour = int(peak_row["hr"]) if peak_row else 0

    # Busiest slot (most occupied readings)
    busy_row = conn.execute("""

        SELECT slot_id, COUNT(*) AS cnt

        FROM   occupancy_log

        WHERE  status = 'occupied'

        GROUP  BY slot_id

        ORDER  BY cnt DESC

        LIMIT  1

    """).fetchone()
    busiest_slot = busy_row["slot_id"] if busy_row else None

    # Hourly trend: last 48 distinct hours bucketed by strftime('%Y-%m-%dT%H:00', logged_at)
    trend_rows = conn.execute("""

        SELECT strftime('%Y-%m-%dT%H:00', logged_at) AS hour,

               SUM(CASE WHEN status = 'occupied' THEN 1 ELSE 0 END) AS occupied,

               SUM(CASE WHEN status = 'empty'    THEN 1 ELSE 0 END) AS empty

        FROM   occupancy_log

        GROUP  BY hour

        ORDER  BY hour DESC

        LIMIT  48

    """).fetchall()
    # Reverse so chronological order (oldest first)
    hourly_trend = list(reversed([dict(r) for r in trend_rows]))

    conn.close()

    return {
        "total_readings":   total,
        "avg_occupancy_pct": avg_pct,
        "peak_hour":         peak_hour,
        "busiest_slot":      busiest_slot,
        "hourly_trend":      hourly_trend,
    }