-
-
-
+
+
+
+
+
+
Lịch sử phát hiện
+
Chưa có dữ liệu
+
+
-
+
+
+
+
diff --git a/orangepi-server/visualizer.py b/orangepi-server/visualizer.py
index e45c1c2..0de33a1 100644
--- a/orangepi-server/visualizer.py
+++ b/orangepi-server/visualizer.py
@@ -32,10 +32,15 @@ from typing import List, Set
# ------------------------------------------------------------
try:
import websockets
- from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect
+ from fastapi import FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
import aiohttp
+ try:
+ import aiosqlite # async SQLite driver (>=0.18)
+ except ImportError: # allow fallback to stdlib sqlite3 for dev
+ aiosqlite = None
+ import sqlite3
except ImportError as e:
print(f"Lỗi: Thiếu thư viện. Cần cài đặt trước: pip install -r requirements.txt")
print(f"Chi tiết: {e}")
@@ -81,6 +86,204 @@ class RadarState:
last_update: float = 0.0
esp32_connected: bool = False
+
+# ============================================================
+# SQLite-backed detection history store
+#
+# We deduplicate writes by slot: when a slot stays continuously valid
+# (e.g. a person standing still for 60 seconds), we only persist 1 row
+# every SAMPLE_EVERY_N_SECONDS instead of every heartbeat frame, which
+# keeps the database small on a Raspberry/Orange Pi flash filesystem.
+# Switching a slot from INVALID -> VALID always writes immediately so
+# entry events have accurate timestamps.
+# ============================================================
+class TargetStore:
+ SAMPLE_EVERY_N_SECONDS = 2.0 # ~30 writes/slot/minute max
+
+ def __init__(self, db_path: Path):
+ self.db_path = Path(db_path)
+ self._conn_sync: "sqlite3.Connection | None" = None
+ self._conn_async = None
+ self._last_written_ts: dict[int, float] = {}
+
+ # ---------- init / close ----------
+ async def init_db(self):
+ # Create parent folder if needed.
+ self.db_path.parent.mkdir(parents=True, exist_ok=True)
+ schema_sql = """
+ CREATE TABLE IF NOT EXISTS detections (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ ts_utc TEXT NOT NULL, -- ISO8601 (UTC)
+ ts_local TEXT NOT NULL, -- ISO8601 (local tz)
+ unixtime REAL NOT NULL, -- seconds since epoch
+ slot INTEGER NOT NULL CHECK (slot IN (1,2,3)),
+ x_mm INTEGER NOT NULL,
+ y_mm INTEGER NOT NULL,
+ speed_cm_s INTEGER NOT NULL,
+ distance_mm REAL NOT NULL,
+ angle_deg REAL NOT NULL
+ );
+ CREATE INDEX IF NOT EXISTS idx_detections_dayhour
+ ON detections(slot, unixtime);
+ """
+ if aiosqlite is not None:
+ self._conn_async = await aiosqlite.connect(str(self.db_path))
+ await self._conn_async.executescript(schema_sql)
+ await self._conn_async.commit()
+ else:
+ self._conn_sync = sqlite3.connect(str(self.db_path), check_same_thread=False)
+ self._conn_sync.executescript(schema_sql)
+ self._conn_sync.commit()
+
+ async def close(self):
+ if self._conn_async is not None:
+ try:
+ await self._conn_async.close()
+ finally:
+ self._conn_async = None
+ if self._conn_sync is not None:
+ try:
+ self._conn_sync.close()
+ finally:
+ self._conn_sync = None
+
+ # ---------- internal runner (async preferred, sync fallback) ----------
+ async def _run_write(self, sql: str, params: tuple) -> None:
+ if aiosqlite is not None and self._conn_async is not None:
+ await self._conn_async.execute(sql, params)
+ await self._conn_async.commit()
+ return
+ if self._conn_sync is None:
+ return
+ def _job():
+ assert self._conn_sync is not None
+ self._conn_sync.execute(sql, params)
+ self._conn_sync.commit()
+ await asyncio.to_thread(_job)
+
+ async def _run_read(self, sql: str, params: tuple) -> list[dict]:
+ if aiosqlite is not None and self._conn_async is not None:
+ conn = self._conn_async
+ conn.row_factory = aiosqlite.Row # type: ignore[union-attr]
+ async with conn.execute(sql, params) as cur:
+ rows = await cur.fetchall()
+ return [dict(r) for r in rows]
+ if self._conn_sync is None:
+ return []
+ def _job():
+ assert self._conn_sync is not None
+ self._conn_sync.row_factory = sqlite3.Row
+ cur = self._conn_sync.execute(sql, params)
+ return [dict(r) for r in cur.fetchall()]
+ return await asyncio.to_thread(_job)
+
+ # ---------- write API (dedup per slot) ----------
+ async def maybe_record(self, targets: list[Target], now_ts: float) -> None:
+ import datetime as _dt
+ for t in targets:
+ if not t.valid:
+ # Slot went invalid; reset its timer so the *next* valid
+ # event flushes immediately (captures re-entry into the room).
+ self._last_written_ts.pop(t.slot, None)
+ continue
+ prev = self._last_written_ts.get(t.slot, None)
+ is_first_valid_entry = prev is None
+ enough_time_passed = (
+ prev is None or (now_ts - prev) >= self.SAMPLE_EVERY_N_SECONDS
+ )
+ if not (is_first_valid_entry or enough_time_passed):
+ continue
+ self._last_written_ts[t.slot] = now_ts
+ utc = _dt.datetime.fromtimestamp(now_ts, tz=_dt.timezone.utc)
+ local = _dt.datetime.fromtimestamp(now_ts).astimezone()
+ sql = (
+ "INSERT INTO detections"
+ " (ts_utc, ts_local, unixtime, slot, x_mm, y_mm,"
+ " speed_cm_s, distance_mm, angle_deg)"
+ " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+ )
+ await self._run_write(
+ sql,
+ (
+ utc.isoformat(timespec="seconds"),
+ local.isoformat(timespec="seconds"),
+ round(now_ts, 3),
+ int(t.slot),
+ int(t.x_mm),
+ int(t.y_mm),
+ int(t.speed_cm_s),
+ round(float(t.distance_mm), 2),
+ round(float(t.angle_deg), 4),
+ ),
+ )
+
+ # ---------- read API ----------
+ async def query(
+ self,
+ date_iso: str,
+ hour_start: int | None = None,
+ hour_end: int | None = None,
+ limit: int = 500,
+ ) -> dict:
+ """date_iso = YYYY-MM-DD in the server LOCAL timezone (matches what
+
returns from a browser in the same tz)."""
+ import datetime as _dt
+ try:
+ day = _dt.date.fromisoformat(date_iso)
+ except ValueError as e:
+ raise HTTPException(status_code=400, detail=f"date không hợp lệ: {e}")
+ day_start_local = _dt.datetime.combine(day, _dt.time(0, 0, 0)).astimezone()
+ day_end_local = day_start_local + _dt.timedelta(days=1)
+ unixtime_start = day_start_local.timestamp()
+ unixtime_end = day_end_local.timestamp()
+ where = ["unixtime >= ? AND unixtime < ?"]
+ params: list = [unixtime_start, unixtime_end]
+ if hour_start is not None and 0 <= hour_start <= 23:
+ where.append("CAST(strftime('%H', ts_local) AS INTEGER) >= ?")
+ params.append(int(hour_start))
+ if hour_end is not None and 0 <= hour_end <= 23:
+ where.append("CAST(strftime('%H', ts_local) AS INTEGER) <= ?")
+ params.append(int(hour_end))
+
+ # 1) Raw rows (newest first)
+ rows_sql = (
+ "SELECT id, ts_local, unixtime, slot, x_mm, y_mm,"
+ " speed_cm_s, distance_mm, angle_deg"
+ " FROM detections WHERE " + " AND ".join(where) +
+ " ORDER BY unixtime DESC LIMIT ?"
+ )
+ rows = await self._run_read(rows_sql, tuple(params + [int(limit)]))
+
+ # 2) 24 hourly buckets (0..23) → count+unique slots for the day filter.
+ hour_sql = (
+ "SELECT CAST(strftime('%H', ts_local) AS INTEGER) AS hh,"
+ " COUNT(*) AS n, COUNT(DISTINCT slot) AS uniq_slots"
+ " FROM detections WHERE " + " AND ".join(where) +
+ " GROUP BY hh ORDER BY hh ASC"
+ )
+ buckets_raw = await self._run_read(hour_sql, tuple(params))
+ hourly = [{"hour": h, "count": 0, "uniq_slots": 0} for h in range(24)]
+ for b in buckets_raw:
+ hourly[int(b["hh"])]["count"] = int(b["n"])
+ hourly[int(b["hh"])]["uniq_slots"] = int(b["uniq_slots"])
+
+ total = sum(h["count"] for h in hourly)
+ unique_slots_total = len({r["slot"] for r in rows}) if rows else 0
+ return {
+ "date": date_iso,
+ "filters": {
+ "hour_start": hour_start,
+ "hour_end": hour_end,
+ "limit": limit,
+ },
+ "total_detections": total,
+ "unique_slots": unique_slots_total,
+ "hourly": hourly,
+ "rows": rows,
+ }
+
+
+
def to_json(self) -> dict:
"""Serialize to something the browser can render."""
now = time.time()
@@ -182,6 +385,17 @@ class RadarState:
self.frames_received += 1
self.last_update = now
+ # Persist valid targets into the SQLite history (dedup per slot).
+ # We reference the singleton STORE from the FastAPI lifecycle; this
+ # indirection keeps RadarState pure (no knowledge of DB engine).
+ store = globals().get("STORE")
+ if store is not None:
+ try:
+ await store.maybe_record(self.targets, now)
+ except Exception:
+ # Never let a DB write failure break the realtime radar loop.
+ pass
+
# ============================================================
# Global app state
@@ -372,13 +586,25 @@ async def broadcast_to_browsers():
BASE_DIR = Path(__file__).resolve().parent
TEMPLATES_DIR = BASE_DIR / "templates"
STATIC_DIR = BASE_DIR / "static"
+DATA_DIR = BASE_DIR / "data"
STATIC_DIR.mkdir(exist_ok=True)
TEMPLATES_DIR.mkdir(exist_ok=True)
+DATA_DIR.mkdir(exist_ok=True)
+
+DB_PATH = DATA_DIR / "detections.sqlite3"
+STORE: "TargetStore | None" = None # initialized in lifespan, exported for RadarState
@asynccontextmanager
async def lifespan(app: FastAPI):
- """Start background tasks when app starts, cancel them on shutdown."""
+ """Open the SQLite store, start background tasks, close store on shutdown."""
+ global STORE
+ STORE = TargetStore(DB_PATH)
+ try:
+ await STORE.init_db()
+ except Exception as e:
+ log.warning(f"Không thể khởi tạo SQLite history (tiếp tục chạy radar without DB): {e}")
+ STORE = None
tasks = [
asyncio.create_task(esp32_ws_forever(ESP32_HOST)),
asyncio.create_task(heartbeat_broadcast()),
@@ -387,6 +613,11 @@ async def lifespan(app: FastAPI):
for t in tasks:
t.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
+ if STORE is not None:
+ try:
+ await STORE.close()
+ finally:
+ STORE = None
app = FastAPI(
@@ -430,6 +661,30 @@ async def update_zones(payload: list):
return await post_zones_to_esp32(ESP32_HOST, payload)
+@app.get("/api/history")
+async def get_history(
+ date: str = Query(..., description="YYYY-MM-DD theo múi giờ máy chủ (local tz)"),
+ hour_start: int | None = Query(None, ge=0, le=23),
+ hour_end: int | None = Query(None, ge=0, le=23),
+ limit: int = Query(500, ge=1, le=10000),
+):
+ """Trả về lịch sử detect target (raw + 24 buckets theo giờ)."""
+ if STORE is None:
+ raise HTTPException(status_code=503, detail="SQLite history store chưa sẵn sàng")
+ try:
+ return await STORE.query(
+ date_iso=date,
+ hour_start=hour_start,
+ hour_end=hour_end,
+ limit=limit,
+ )
+ except HTTPException:
+ raise
+ except Exception as e:
+ log.exception(f"GET /api/history failed: {e}")
+ raise HTTPException(status_code=500, detail=f"Lỗi truy vấn history: {e}")
+
+
# --- WebSocket route for browsers -----------------------------
@app.websocket("/ws")
async def browser_ws_endpoint(ws: WebSocket):