98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import threading
|
|
import uuid
|
|
from collections import deque
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
EVENT_STATUSES = {"pending", "acknowledged", "resolved"}
|
|
|
|
|
|
class EventStore:
|
|
def __init__(self, max_events: int = 500) -> None:
|
|
self._events: deque[dict[str, Any]] = deque(maxlen=max_events)
|
|
self._lock = threading.Lock()
|
|
|
|
def create(
|
|
self,
|
|
session_id: str,
|
|
classes: list[str],
|
|
detections: list[dict[str, Any]],
|
|
notification_channels: dict[str, bool],
|
|
) -> dict[str, Any]:
|
|
now = datetime.now().astimezone()
|
|
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": now.isoformat(timespec="seconds"),
|
|
"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,
|
|
}
|
|
with self._lock:
|
|
self._events.appendleft(event)
|
|
return dict(event)
|
|
|
|
def list(
|
|
self,
|
|
limit: int = 100,
|
|
status: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
with self._lock:
|
|
events = [dict(event) for event in self._events]
|
|
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}")
|
|
with self._lock:
|
|
for event in self._events:
|
|
if event["id"] != event_id:
|
|
continue
|
|
event["status"] = status
|
|
event["handled_at"] = (
|
|
None
|
|
if status == "pending"
|
|
else datetime.now().astimezone().isoformat(
|
|
timespec="seconds"
|
|
)
|
|
)
|
|
return dict(event)
|
|
return None
|
|
|
|
def summary(self) -> dict[str, int]:
|
|
today = datetime.now().astimezone().date()
|
|
with self._lock:
|
|
events = [dict(event) for event in self._events]
|
|
today_events = [
|
|
event
|
|
for event in events
|
|
if datetime.fromisoformat(event["created_at"]).date() == today
|
|
]
|
|
return {
|
|
"total": len(events),
|
|
"today": len(today_events),
|
|
"pending": sum(
|
|
event["status"] == "pending" for event in events
|
|
),
|
|
"acknowledged": sum(
|
|
event["status"] == "acknowledged" for event in events
|
|
),
|
|
"resolved": sum(
|
|
event["status"] == "resolved" for event in events
|
|
),
|
|
"fire": sum("fire" in event["classes"] for event in events),
|
|
"smoke": sum("smoke" in event["classes"] for event in events),
|
|
}
|