#!/usr/bin/env python3 """ ESP32-S3 LD2453 Radar — Web Visualizer for Orange Pi ====================================================== - Kết nối WebSocket tới ESP32 nhận dữ liệu radar realtime - Mở Web Server (FastAPI) cho phép trình duyệt xem đồ họa đẹp - Broadcast dữ liệu từ ESP32 tới tất cả các client trình duyệt đang kết nối - Hỗ trợ REST API lấy zones, cập nhật zones (proxy tới ESP32) Sử dụng: python visualizer.py # Mặc định port 8000 python visualizer.py 192.168.1.100 --port 8080 # Tùy chỉnh port python visualizer.py 192.168.1.100 --host 0.0.0.0 # Mở cho truy cập ngoài Mở trình duyệt vào: http://:8000 """ import argparse import asyncio import json import logging import math import sys import time from contextlib import asynccontextmanager from dataclasses import asdict, dataclass, field from pathlib import Path from typing import List, Set # ------------------------------------------------------------ # Third-party imports (install via requirements.txt) # ------------------------------------------------------------ try: import websockets 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}") sys.exit(1) # ============================================================ # Data structures — keep same semantics as original # ============================================================ @dataclass class Target: slot: int valid: bool = False x_mm: int = 0 y_mm: int = 0 speed_cm_s: int = 0 distance_mm: float = 0.0 angle_deg: float = 0.0 last_seen: float = 0.0 @dataclass class Zone: id: int x1: int = 0 y1: int = 0 x2: int = 0 y2: int = 0 occupied: bool = False @dataclass class RadarState: node_id: str = "RADAR_A" sequence: int = 0 timestamp_ms: int = 0 target_count: int = 0 targets: List[Target] = field( default_factory=lambda: [Target(i + 1) for i in range(3)] ) zones: List[Zone] = field(default_factory=lambda: [Zone(i) for i in range(1, 4)]) frames_received: int = 0 last_update: float = 0.0 esp32_connected: bool = False def to_json(self) -> dict: """Serialize to something the browser can render.""" now = time.time() return { "node_id": self.node_id, "sequence": self.sequence, "timestamp_ms": self.timestamp_ms, "target_count": self.target_count, "targets": [ { "slot": t.slot, "valid": t.valid, "x_mm": t.x_mm, "y_mm": t.y_mm, "speed_cm_s": t.speed_cm_s, "distance_mm": round(t.distance_mm, 1), "angle_deg": round(t.angle_deg, 2), "last_seen": ( round(now - t.last_seen, 2) if t.valid or (t.last_seen and t.last_seen > 0) else None ), } for t in self.targets ], "zones": [ { "id": z.id, "x1": z.x1, "y1": z.y1, "x2": z.x2, "y2": z.y2, "occupied": z.occupied, } for z in self.zones ], "frames_received": self.frames_received, "last_update": ( round(now - self.last_update, 2) if self.last_update and self.last_update > 0 else None ), "esp32_connected": self.esp32_connected, "server_time": round(now, 2), } async def update_from_frame(self, frame: dict): self.node_id = frame.get("node_id", self.node_id) self.sequence = frame.get("sequence", 0) self.timestamp_ms = frame.get("timestamp", 0) self.target_count = frame.get("target_count", 0) now = time.time() PHYS_Y_MIN = 1 PHYS_Y_MAX = 6500 PHYS_X_ABS = 6500 valid_targets_this_frame = 0 targets_data = frame.get("targets", []) for t_data in targets_data: slot = t_data.get("id", 0) if 1 <= slot <= 3: t = self.targets[slot - 1] t.slot = slot t.valid = bool(t_data.get("valid", False)) t.x_mm = int(t_data.get("x", 0)) t.y_mm = int(t_data.get("y", 0)) t.speed_cm_s = int(t_data.get("speed", 0)) t.distance_mm = float(t_data.get("distance", 0.0)) t.angle_deg = float(t_data.get("angle", 0.0)) y_ok = PHYS_Y_MIN <= t.y_mm <= PHYS_Y_MAX x_ok = -PHYS_X_ABS <= t.x_mm <= PHYS_X_ABS if not (y_ok and x_ok): t.valid = False if t.valid and (t.distance_mm == 0 or t.angle_deg == 0): t.distance_mm = math.hypot(t.x_mm, t.y_mm) if t.distance_mm > 0: t.angle_deg = math.degrees(math.atan2(t.x_mm, t.y_mm)) if t.valid: t.last_seen = now valid_targets_this_frame += 1 self.target_count = valid_targets_this_frame self.frames_received += 1 self.last_update = now store = globals().get("STORE") if store is not None: try: await store.maybe_record(self.targets, now) except Exception: pass # ============================================================ # 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, } # ============================================================ # Global app state # ============================================================ STATE = RadarState() # Default zones — matches the original ESP32 layout STATE.zones[0] = Zone(1, x1=-4000, y1=1, x2=-1, y2=4000) STATE.zones[1] = Zone(2, x1=1, y1=1, x2=4000, y2=4000) STATE.zones[2] = Zone(3, x1=-4001, y1=4001, x2=4001, y2=6000) CONNECTED_BROWSERS: Set[WebSocket] = set() BROWSER_LOCK = asyncio.Lock() ESP32_HOST: str = "" # populated from CLI logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s", ) log = logging.getLogger("radar-web") # ============================================================ # Helper: update zone occupancy (redundant but useful for validation) # ============================================================ def _update_zone_occupancy_from_targets(state: RadarState): for z in state.zones: z.occupied = False for t in state.targets: if not t.valid: continue if (min(z.x1, z.x2) <= t.x_mm <= max(z.x1, z.x2) and min(z.y1, z.y2) <= t.y_mm <= max(z.y1, z.y2)): z.occupied = True break # ============================================================ # Heartbeat: broadcast last known state to browsers every 1s # even if ESP32 hasn't sent a new frame yet (parser stub scenario). # ============================================================ async def heartbeat_broadcast(): while True: try: await broadcast_to_browsers() except Exception: pass await asyncio.sleep(1.0) # ============================================================ # ESP32 WebSocket client — runs as a background task # ============================================================ async def esp32_ws_forever(esp32_host: str): """Loop forever: connect to ESP32 /ws, read frames, update STATE, broadcast.""" uri = f"ws://{esp32_host}/ws" reconnect_delay = 2.0 while True: try: STATE.esp32_connected = False log.info(f"Đang kết nối ESP32 tại {uri} ...") # Push initial "not connected" state to any browser that joined early await broadcast_to_browsers() async with websockets.connect( uri, ping_interval=30, ping_timeout=10, close_timeout=5, open_timeout=8, max_size=1_000_000, ) as ws: STATE.esp32_connected = True log.info(f"Đã kết nối WebSocket tới ESP32 ({esp32_host})") # Fetch zones from ESP32 after successful connect await fetch_zones_from_esp32(esp32_host) # Broadcast immediately so browsers see zones right after connect await broadcast_to_browsers() async for message in ws: try: frame = json.loads(message) except json.JSONDecodeError: log.warning(f"Lỗi parse JSON từ ESP32: {message[:80]}...") continue await STATE.update_from_frame(frame) _update_zone_occupancy_from_targets(STATE) # Always push the latest state await broadcast_to_browsers() log.info("ESP32 WebSocket đóng, đang kết nối lại...") STATE.esp32_connected = False await broadcast_to_browsers() except websockets.exceptions.ConnectionClosed as e: STATE.esp32_connected = False log.warning(f"Mất kết nối ESP32 ({e}), thử lại sau {reconnect_delay}s") await broadcast_to_browsers() await asyncio.sleep(reconnect_delay) except OSError as e: STATE.esp32_connected = False log.warning(f"Không thể kết nối ESP32: {e}. Thử lại sau {reconnect_delay}s") await broadcast_to_browsers() await asyncio.sleep(reconnect_delay) except asyncio.CancelledError: log.info("Task ESP32 WS bị hủy (shutdown).") break except Exception as e: STATE.esp32_connected = False log.exception(f"Lỗi ESP32 WS loop: {type(e).__name__}: {e}") await asyncio.sleep(reconnect_delay) async def fetch_zones_from_esp32(esp32_host: str): """Fetch zone layout from ESP32 /zones REST endpoint.""" url = f"http://{esp32_host}/zones" try: timeout = aiohttp.ClientTimeout(total=3) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get(url) as resp: if resp.status == 200: data = await resp.json() for z_data in data: idx = z_data.get("id", 0) if 1 <= idx <= 3: z = STATE.zones[idx - 1] z.id = idx z.x1 = int(z_data.get("x1", z.x1)) z.y1 = int(z_data.get("y1", z.y1)) z.x2 = int(z_data.get("x2", z.x2)) z.y2 = int(z_data.get("y2", z.y2)) z.occupied = bool(z_data.get("occupied", False)) log.info(f"Đã lấy {len(data)} zones từ ESP32") else: log.warning(f"GET /zones trả về HTTP {resp.status}") except Exception as e: log.warning(f"Không lấy được zones từ ESP32 (sẽ dùng mặc định): {e}") async def post_zones_to_esp32(esp32_host: str, zones_payload): """Forward a zone update to ESP32 /updateZones.""" url = f"http://{esp32_host}/updateZones" try: timeout = aiohttp.ClientTimeout(total=5) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post(url, json=zones_payload) as resp: body = await resp.text() if resp.status in (200, 204): # Apply locally immediately for z_data in zones_payload: idx = z_data.get("id", 0) if 1 <= idx <= 3: z = STATE.zones[idx - 1] z.id = idx z.x1 = int(z_data.get("x1", z.x1)) z.y1 = int(z_data.get("y1", z.y1)) z.x2 = int(z_data.get("x2", z.x2)) z.y2 = int(z_data.get("y2", z.y2)) await broadcast_to_browsers() return {"status": "ok", "http_status": resp.status, "body": body} else: raise HTTPException(status_code=resp.status, detail=body) except HTTPException: raise except Exception as e: raise HTTPException(status_code=502, detail=f"Lỗi khi gọi ESP32: {e}") # ============================================================ # Broadcast to all connected browsers # ============================================================ async def broadcast_to_browsers(): payload = json.dumps(STATE.to_json()) if not CONNECTED_BROWSERS: return async with BROWSER_LOCK: dead: List[WebSocket] = [] for ws in list(CONNECTED_BROWSERS): try: await ws.send_text(payload) except Exception: dead.append(ws) for ws in dead: CONNECTED_BROWSERS.discard(ws) # ============================================================ # FastAPI app setup # ============================================================ 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): """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()), ] yield 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( title="ESP32 LD2453 Radar Web Visualizer", lifespan=lifespan, ) app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") # --- HTTP routes ------------------------------------------------ @app.get("/") async def get_index(): index_file = TEMPLATES_DIR / "index.html" if not index_file.exists(): return { "error": "index.html không tồn tại. Hãy đảm bảo thư mục templates/ có index.html." } return FileResponse(str(index_file)) @app.get("/api/state") async def get_state(): """REST endpoint lấy trạng thái hiện tại (polling dự phòng).""" return STATE.to_json() @app.get("/api/zones") async def get_zones(): return [ {"id": z.id, "x1": z.x1, "y1": z.y1, "x2": z.x2, "y2": z.y2, "occupied": z.occupied} for z in STATE.zones ] @app.post("/api/zones") async def update_zones(payload: list): """Nhận cập nhật zones từ trình duyệt và forward tới ESP32.""" if not ESP32_HOST: raise HTTPException(status_code=500, detail="ESP32 host chưa được cấu hình") 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): await ws.accept() # Gửi trạng thái ngay lập tức để client có data lần đầu try: await ws.send_text(json.dumps(STATE.to_json())) except Exception: return async with BROWSER_LOCK: CONNECTED_BROWSERS.add(ws) log.info(f"Browser kết nối: {ws.client} (tổng: {len(CONNECTED_BROWSERS)})") try: # Browser chủ yếu nhận, nhưng chúng ta vẫn đọc để phát hiện ngắt kết nối while True: data = await ws.receive_text() # Có thể mở rộng: nhận lệnh từ trình duyệt (ví dụ: yêu cầu cập nhật zones) try: msg = json.loads(data) if msg.get("type") == "updateZones": zones = msg.get("zones", []) await post_zones_to_esp32(ESP32_HOST, zones) except Exception: pass except WebSocketDisconnect: pass finally: async with BROWSER_LOCK: CONNECTED_BROWSERS.discard(ws) log.info(f"Browser ngắt kết nối (còn lại: {len(CONNECTED_BROWSERS)})") # ============================================================ # Entry point # ============================================================ def main(): global ESP32_HOST parser = argparse.ArgumentParser( description="ESP32 LD2453 Radar — Web Visualizer (Xem trên trình duyệt)" ) parser.add_argument( "esp32", help="Địa chỉ IP ESP32 (ví dụ: 192.168.1.100)", ) parser.add_argument( "--host", default="0.0.0.0", help="Host lắng nghe của Web Server (mặc định: 0.0.0.0 — cho phép truy cập từ ngoài)", ) parser.add_argument( "--port", type=int, default=8000, help="Port của Web Server (mặc định: 8000)", ) parser.add_argument( "--verbose", "-v", action="store_true", help="Hiển thị log chi tiết (DEBUG level)", ) args = parser.parse_args() if args.verbose: logging.getLogger().setLevel(logging.DEBUG) ESP32_HOST = args.esp32.strip() # Try importing uvicorn — keep it here so help text still shows without install try: import uvicorn except ImportError: print("Lỗi: Cần cài uvicorn: pip install uvicorn") sys.exit(1) print() print("=" * 62) print(" ESP32-S3 LD2453 Radar — Web Visualizer") print("=" * 62) print(f" ESP32 : ws://{ESP32_HOST}/ws") print(f" Web UI : http://{args.host}:{args.port}/") print(f" WS UI : ws://:{args.port}/ws") print(f" Browsers: 0 client đang kết nối") print("=" * 62) print(" Nhấn Ctrl+C để dừng") print("=" * 62) print() try: uvicorn.run(app, host=args.host, port=args.port, log_level="warning") except KeyboardInterrupt: print("\nThoát.") if __name__ == "__main__": main()