diff --git a/app/main.py b/app/main.py index a9c3486..3917ad8 100644 --- a/app/main.py +++ b/app/main.py @@ -316,6 +316,82 @@ async def whitelist_remove(plate: str, back: str = Form("")): return RedirectResponse(back or "/whitelist", status_code=303) +ZONE_PATH = "/data/zone.json" + + +@app.get("/config/zone", response_class=HTMLResponse) +async def zone_page(request: Request): + import json + zone: dict = {} + if os.path.exists(ZONE_PATH): + try: + with open(ZONE_PATH) as f: + zone = json.load(f) + except Exception: + pass + + latest_frame = None + events = database.get_events(limit=1) + if events: + eid = events[0]["id"] + bf = os.path.join(EVENTS_DIR, eid, "best_frame.jpg") + if os.path.exists(bf): + latest_frame = f"events/{eid}/best_frame.jpg" + else: + first = sorted(glob.glob(os.path.join(EVENTS_DIR, eid, "frame_*.jpg"))) + if first: + latest_frame = f"events/{eid}/{os.path.basename(first[0])}" + + return templates.TemplateResponse("zone.html", { + "request": request, + "zone": zone, + "latest_frame": latest_frame, + }) + + +@app.post("/config/zone") +async def save_zone(request: Request): + import json + data = await request.json() + points = data.get("points", []) + if len(points) == 0: + if os.path.exists(ZONE_PATH): + os.unlink(ZONE_PATH) + return {"ok": True, "deleted": True} + if len(points) < 3: + raise HTTPException(400, "Minimum 3 points requis") + with open(ZONE_PATH, "w") as f: + json.dump({"points": points}, f) + return {"ok": True, "points": len(points)} + + +@app.get("/config/snapshot") +async def live_snapshot(): + import tempfile, cv2 + url = watcher._rtsp_url() + with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f: + tmp_path = f.name + + def _grab(): + subprocess.run([ + "ffmpeg", "-y", "-rtsp_transport", "tcp", + "-i", url, "-vframes", "1", "-q:v", "2", tmp_path, + ], capture_output=True, timeout=10) + return os.path.exists(tmp_path) and os.path.getsize(tmp_path) > 0 + + loop = asyncio.get_event_loop() + ok = await loop.run_in_executor(None, _grab) + if ok: + with open(tmp_path, "rb") as f: + data = f.read() + os.unlink(tmp_path) + from fastapi.responses import Response + return Response(content=data, media_type="image/jpeg") + if os.path.exists(tmp_path): + os.unlink(tmp_path) + raise HTTPException(500, "Snapshot RTSP impossible") + + @app.post("/event/{event_id}/test-lpr") async def test_lpr(event_id: str, frame_path: str = Form(""), bbox: str = Form("")): import json, base64, cv2, numpy as np diff --git a/app/templates/index.html b/app/templates/index.html index eeddaf6..121395f 100644 --- a/app/templates/index.html +++ b/app/templates/index.html @@ -35,6 +35,7 @@ {{ total }} passage{{ 's' if total != 1 else '' }} 📊 Stats 🛡 Whitelist + ⬡ Zone diff --git a/app/templates/zone.html b/app/templates/zone.html new file mode 100644 index 0000000..75218ca --- /dev/null +++ b/app/templates/zone.html @@ -0,0 +1,298 @@ + + + + + + CamWatch — Zone de détection + + + + + +
+ ← Retour + ⬡ + Zone de détection + +
+ +
+ + +
+
+ +
+
+ + {% if zone.points %}Zone active — {{ zone.points | length }} points{% else %}Aucune zone définie{% endif %} + +
+ + +
+
+
+ + +
+
+ {% if zone.points %} + ✓ Zone active — {{ zone.points | length }} points définis. + Seuls les mouvements et détections à l'intérieur de cette zone seront pris en compte. + {% else %} + Aucune zone définie — toute l'image est utilisée pour la détection. + {% endif %} +
+
+ + {% if zone.points %} + + {% endif %} + +
+
+ + +
+ + +
+ 💡 Dessine autour de la partie de l'image où les voitures passent (ex: ta voie d'accès, la rue devant le portail). Exclue ton jardin, la végétation, le ciel. +
+
+ +
+ + + + diff --git a/app/watcher.py b/app/watcher.py index 7a4f3ff..a48deaf 100644 --- a/app/watcher.py +++ b/app/watcher.py @@ -35,7 +35,28 @@ _last_event_time = 0.0 _analyzer: PlateAnalyzer | None = None import re as _re +import json as _json + _FR_PLATE_RE = _re.compile(r'^([A-Z]{2})(\d{3})([A-Z]{2})$') +_ZONE_PATH = "/data/zone.json" + + +def _read_zone_points() -> list | None: + try: + with open(_ZONE_PATH) as f: + pts = _json.load(f).get("points", []) + if len(pts) >= 3: + return pts + except Exception: + pass + return None + + +def _zone_mask(h: int, w: int, pts: list) -> np.ndarray: + poly = np.array([[int(x * w), int(y * h)] for x, y in pts], dtype=np.int32) + mask = np.zeros((h, w), dtype=np.uint8) + cv2.fillPoly(mask, [poly], 255) + return mask def _normalize_plate(raw: str) -> str: """Validate and format a French plate (6-8 alphanumeric chars). @@ -189,6 +210,9 @@ def _process_clip(event_id: str, event_dir: str, clip_path: str, camera_name: st loaded: list[tuple[np.ndarray, str, float, float]] = [] # frame, path, sharpness, motion prev_small: np.ndarray | None = None + zone_pts = _read_zone_points() + zone_mask_small: np.ndarray | None = None # computed lazily on first frame + for path in frame_files: frame = cv2.imread(path) if frame is None: @@ -197,7 +221,16 @@ def _process_clip(event_id: str, event_dir: str, clip_path: str, camera_name: st lap = cv2.Laplacian(gray, cv2.CV_64F).var() h, w = gray.shape small = cv2.resize(gray, (_MOTION_W, _MOTION_W * h // w)) - motion = float(np.mean(np.abs(small.astype(np.float32) - prev_small.astype(np.float32)))) if prev_small is not None else 0.0 + if zone_mask_small is None and zone_pts: + sh, sw = small.shape + zone_mask_small = cv2.resize(_zone_mask(h, w, zone_pts), (sw, sh)) + if prev_small is not None: + diff = np.abs(small.astype(np.float32) - prev_small.astype(np.float32)) + if zone_mask_small is not None: + diff = diff * (zone_mask_small.astype(np.float32) / 255.0) + motion = float(np.mean(diff)) + else: + motion = 0.0 prev_small = small loaded.append((frame, path, lap, motion)) @@ -221,6 +254,13 @@ def _process_clip(event_id: str, event_dir: str, clip_path: str, camera_name: st scored: list[tuple[float, np.ndarray, str]] = [] for _, frame, path in top15: plates = _analyzer.detect_plates(frame) if _analyzer else [] + if plates and zone_pts: + fh, fw = frame.shape[:2] + poly = np.array([[int(x * fw), int(y * fh)] for x, y in zone_pts], dtype=np.int32) + plates = [p for p in plates + if cv2.pointPolygonTest(poly.reshape(-1, 1, 2), + ((p[0] + p[2]) / 2, (p[1] + p[3]) / 2), + False) >= 0] score = _frame_score(frame, plates) scored.append((score, frame, path)) scored.sort(key=lambda x: -x[0])