yolo/backend/events.py

from __future__ import annotations

import json
import uuid
from datetime import UTC, datetime
from typing import Any

from .storage import Database, utc_now_iso

EVENT_STATUSES = {"pending", "acknowledged", "resolved"}


def _local_today_utc_boundary() -> str:
    """UTC Z-format timestamp of local midnight, for 'today' comparisons."""
    local_midnight = datetime.now().astimezone().replace(
        hour=0,
        minute=0,
        second=0,
        microsecond=0,
    )
    return (
        local_midnight.astimezone(UTC)
        .isoformat(timespec="seconds")
        .replace("+00:00", "Z")
    )


class EventStore:
    """Persistent alert event store backed by a single SQLite table.

    Keeps at most ``max_events`` rows (newest first, ordered by rowid so
    events created within the same second keep a stable insertion order).
    """

    def __init__(
        self,
        db: Database | None = None,
        max_events: int = 500,
    ) -> None:
        self._db = db or Database(":memory:")
        self._max_events = max(1, max_events)

    def create(
        self,
        session_id: str,
        classes: list[str],
        detections: list[dict[str, Any]],
        notification_channels: dict[str, bool],
    ) -> dict[str, Any]:
        confidences = [
            float(detection.get("confidence", 0.0))
            for detection in detections
            if detection.get("class") in classes
        ]
        event = {
            "id": uuid.uuid4().hex[:12],
            "created_at": utc_now_iso(),
            "session_id": session_id,
            "classes": classes,
            "max_confidence": round(max(confidences, default=0.0), 6),
            "detection_count": len(detections),
            "status": "pending",
            "notification_channels": notification_channels,
            "handled_at": None,
        }
        self._db.execute(
            """
            INSERT INTO events (
                id, created_at, session_id, classes, max_confidence,
                detection_count, status, notification_channels, handled_at
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                event["id"],
                event["created_at"],
                session_id,
                json.dumps(classes, ensure_ascii=False),
                event["max_confidence"],
                event["detection_count"],
                event["status"],
                json.dumps(notification_channels, ensure_ascii=False),
                None,
            ),
        )
        self._db.execute(
            """
            DELETE FROM events WHERE id IN (
                SELECT id FROM events ORDER BY rowid DESC LIMIT -1 OFFSET ?
            )
            """,
            (self._max_events,),
        )
        return event

    def list(
        self,
        limit: int = 100,
        status: str | None = None,
    ) -> list[dict[str, Any]]:
        rows = self._db.execute(
            "SELECT * FROM events ORDER BY rowid DESC",
        ).fetchall()
        events = [self._to_dict(row) for row in rows]
        if status:
            events = [event for event in events if event["status"] == status]
        return events[: max(1, min(limit, 500))]

    def update(self, event_id: str, status: str) -> dict[str, Any] | None:
        if status not in EVENT_STATUSES:
            raise ValueError(f"Unsupported event status: {status}")
        cursor = self._db.execute(
            """
            UPDATE events SET status = ?, handled_at = ?
            WHERE id = ?
            """,
            (status, None if status == "pending" else utc_now_iso(), event_id),
        )
        if cursor.rowcount == 0:
            return None
        row = self._db.execute(
            "SELECT * FROM events WHERE id = ?",
            (event_id,),
        ).fetchone()
        return self._to_dict(row)

    def summary(self) -> dict[str, int]:
        rows = self._db.execute(
            "SELECT created_at, classes, status FROM events",
        ).fetchall()
        today_boundary = _local_today_utc_boundary()
        today_events = 0
        counts = {status: 0 for status in EVENT_STATUSES}
        fire = 0
        smoke = 0
        for row in rows:
            counts[row["status"]] += 1
            classes = json.loads(row["classes"])
            if row["created_at"] >= today_boundary:
                today_events += 1
            fire += "fire" in classes
            smoke += "smoke" in classes
        return {
            "total": len(rows),
            "today": today_events,
            "pending": counts["pending"],
            "acknowledged": counts["acknowledged"],
            "resolved": counts["resolved"],
            "fire": fire,
            "smoke": smoke,
        }

    @staticmethod
    def _to_dict(row: Any) -> dict[str, Any]:
        return {
            "id": row["id"],
            "created_at": row["created_at"],
            "session_id": row["session_id"],
            "classes": json.loads(row["classes"]),
            "max_confidence": row["max_confidence"],
            "detection_count": row["detection_count"],
            "status": row["status"],
            "notification_channels": json.loads(row["notification_channels"]),
            "handled_at": row["handled_at"],
        }