Release FaceID 5.1 guest and site intelligence

CHANGELOG.md

@@ -3,6 +3,25 @@
 All notable changes to FaceID. The Home Assistant app shows this file in the
 update dialog; standalone users can watch GitHub releases.
 
+## 5.1.0 — 2026-08-11
+
+- **Temporary Guest Access:** create a guest from one quality-checked photo, limit
+  recognition to a date/time window, selected cameras and an entry count, then revoke
+  or delete it immediately from a friendly UI. Guest matching uses a stricter score and
+  margin; liveness plus a second factor are mandatory and FaceID never unlocks alone.
+- **Site map and estimated routes:** drag cameras onto a responsive site canvas, connect
+  plausible transitions and see each person's last observed camera. Saved links also
+  constrain live scenario and appearance-ReID paths; the UI clearly labels locations as
+  estimates rather than GPS positions.
+- **Anonymous traffic analytics:** camera/zone traffic share, peak hour and common
+  transitions are calculated from Frigate person events without using identities or
+  face images in the aggregate calculations.
+- **Route clip playlist:** every visit can play its Frigate recognition clips in
+  chronological camera order. The player shows route progress, advances automatically
+  and skips expired/missing clips without stopping the remaining journey.
+- **Operational backup:** guest templates, guest audit decisions and the site map are
+  included in safe backups and restores.
+
 ## 5.0.4 — 2026-08-11
 
 - **Camera participation from the UI:** every Frigate camera can now be enabled or

README.md

@@ -1,6 +1,11 @@
 # FaceID — self-hosted face recognition for Frigate + Home Assistant
 
-## FaceID 5.0: a product workflow for homes and offices
+## FaceID 5.1: a product workflow for homes and offices
+
+5.1 adds temporary guest passes with strict liveness/second-factor boundaries, a
+drag-and-drop camera site map, anonymous Frigate person-traffic analytics and a visit
+player that plays recognition clips in camera order. Map locations are explicitly
+estimated, and face recognition alone never authorizes a door unlock.
 
 Version 5 adds a dedicated **Users** area for normal day-to-day operation. Create a
 person, upload 5–10 photos, see a plain-language quality result for each photo, rename

app/__init__.py

@@ -1 +1 @@
-VERSION = "5.0.4"
+VERSION = "5.1.0"

app/access_control.py

@@ -11,15 +11,15 @@
 ROLE_TABS = {
     "admin": ["*"],
     "operator": [
-        "dashboard", "users", "unknowns", "visits", "intercom", "liveness",
+        "dashboard", "users", "guests", "site-map", "unknowns", "visits", "intercom", "liveness",
         "automations", "activity", "calibration", "privacy", "health-center",
     ],
-    "viewer": ["dashboard", "visits", "liveness", "activity"],
+    "viewer": ["dashboard", "site-map", "visits", "liveness", "activity"],
 }
 MUTATING_PREFIXES = {
     "operator": (
         "/api/persons", "/api/unknowns", "/api/audit",
-        "/api/intercom", "/api/liveness",
+        "/api/intercom", "/api/liveness", "/api/guests", "/api/site-map",
     ),
     "viewer": (),
 }

app/audit.py

@@ -759,6 +759,17 @@ def dashboard_summary(self):
             "processing": counts.get("processing", 0),
         }
 
+    def traffic_events(self, *, after_ts: float, limit: int = 10000) -> list[dict]:
+        """Anonymous person-event facts for zone traffic analytics."""
+        with self._lock, self._connection() as con:
+            con.row_factory = sqlite3.Row
+            rows = con.execute(
+                "SELECT event_id, camera, start_ts, end_ts, status FROM events "
+                "WHERE start_ts>=? AND status!='processing' ORDER BY start_ts LIMIT ?",
+                (float(after_ts), max(1, min(int(limit), 100000))),
+            ).fetchall()
+        return [dict(row) for row in rows]
+
     def labeled_events(self):
         with self._lock, self._connection() as con:
             con.row_factory = sqlite3.Row

app/backup_util.py

@@ -13,10 +13,11 @@
 log = logging.getLogger("faceid.backup")
 
 # Nur die unersetzliche Handarbeit sichern — nicht die Unknown-Queue oder Frigate-Vollbilder.
-BACKUP_SUBDIRS = ("persons", "ignored", "body")
+BACKUP_SUBDIRS = ("persons", "ignored", "body", "guests")
 BACKUP_FILES = (
     "settings.json", "learning_runs.json", "frigate_sync.json",
-    "camera_profiles.json", "access_control.json", "schema.json",
+    "camera_profiles.json", "access_control.json", "guest_access.json",
+    "site_map.json", "schema.json",
 )
 
 
@@ -60,8 +61,8 @@ def build_backup_gz(data_dir: Path) -> bytes:
         if audit.is_file():
             _add_audit_snapshot(tar, audit)
         manifest = json.dumps({
-            "format": 5, "created": time.time(),
-            "includes": [*BACKUP_SUBDIRS, "settings", "learning-runs", "sync-ledger", "camera-profiles", "access-policy", "schema", "audit-history"],
+            "format": 6, "created": time.time(),
+            "includes": [*BACKUP_SUBDIRS, "settings", "learning-runs", "sync-ledger", "camera-profiles", "access-policy", "guest-access", "site-map", "schema", "audit-history"],
             "excludes": ["frigate-credentials", "mqtt-credentials", "clips", "media-cache"],
             "restore_note": "body classifier is rebuilt from reviewed material after restore",
         }, indent=2).encode("utf-8")

app/guest_access.py

@@ -0,0 +1,170 @@
+"""Temporary guest identities and fail-closed access eligibility decisions."""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import threading
+import time
+import uuid
+from pathlib import Path
+
+import cv2
+import numpy as np
+
+
+class GuestAccess:
+    def __init__(self, data_dir: Path, *, threshold: float = .62, margin: float = .12):
+        self.root = data_dir / "guests"
+        self.root.mkdir(parents=True, exist_ok=True)
+        self.events_path = data_dir / "guest_access.json"
+        self.threshold = max(.5, float(threshold))
+        self.margin = max(.08, float(margin))
+        self._lock = threading.RLock()
+
+    @staticmethod
+    def _atomic_json(path: Path, value) -> None:
+        temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
+        try:
+            with temporary.open("w", encoding="utf-8") as handle:
+                json.dump(value, handle, ensure_ascii=False, indent=2)
+                handle.flush(); os.fsync(handle.fileno())
+            os.replace(temporary, path)
+        finally:
+            temporary.unlink(missing_ok=True)
+
+    def _meta(self, guest_id: str) -> dict | None:
+        path = self.root / guest_id / "meta.json"
+        try:
+            value = json.loads(path.read_text(encoding="utf-8"))
+            return value if isinstance(value, dict) else None
+        except (OSError, json.JSONDecodeError):
+            return None
+
+    @staticmethod
+    def status(meta: dict, now: float | None = None) -> str:
+        now = float(now or time.time())
+        if meta.get("revoked"):
+            return "revoked"
+        if now < float(meta.get("valid_from") or 0):
+            return "future"
+        if now > float(meta.get("valid_until") or 0):
+            return "expired"
+        maximum = int(meta.get("max_entries") or 0)
+        if maximum and int(meta.get("entries_used") or 0) >= maximum:
+            return "used"
+        return "active"
+
+    def list(self) -> list[dict]:
+        with self._lock:
+            result = []
+            for folder in self.root.iterdir():
+                if not folder.is_dir():
+                    continue
+                meta = self._meta(folder.name)
+                if not meta:
+                    continue
+                result.append({**meta, "status": self.status(meta), "photo": f"media/guests/{folder.name}/face.jpg"})
+            return sorted(result, key=lambda item: (item["status"] != "active", item.get("valid_from", 0)))
+
+    def create(self, *, name: str, valid_from: float, valid_until: float,
+               max_entries: int, allowed_cameras: list[str], crop, embedding) -> dict:
+        name = str(name).strip()
+        if not name or len(name) > 100:
+            raise ValueError("guest name must contain 1-100 characters")
+        if float(valid_until) <= float(valid_from):
+            raise ValueError("valid_until must be after valid_from")
+        if float(valid_until) - float(valid_from) > 90 * 86400:
+            raise ValueError("guest access may not exceed 90 days")
+        guest_id = uuid.uuid4().hex[:12]
+        folder = self.root / guest_id
+        folder.mkdir()
+        meta = {
+            "id": guest_id, "name": name, "valid_from": float(valid_from),
+            "valid_until": float(valid_until), "max_entries": max(1, min(int(max_entries), 1000)),
+            "entries_used": 0, "allowed_cameras": sorted({str(x) for x in allowed_cameras if str(x)}),
+            "revoked": False, "created_ts": time.time(),
+        }
+        try:
+            if not cv2.imwrite(str(folder / "face.jpg"), crop, [cv2.IMWRITE_JPEG_QUALITY, 92]):
+                raise ValueError("could not store guest face")
+            with (folder / "embedding.npy").open("wb") as handle:
+                np.save(handle, np.asarray(embedding, dtype=np.float32))
+                handle.flush(); os.fsync(handle.fileno())
+            self._atomic_json(folder / "meta.json", meta)
+        except Exception:
+            shutil.rmtree(folder, ignore_errors=True)
+            raise
+        return {**meta, "status": self.status(meta), "photo": f"media/guests/{guest_id}/face.jpg"}
+
+    def revoke(self, guest_id: str) -> dict:
+        with self._lock:
+            meta = self._meta(guest_id)
+            if not meta:
+                raise KeyError(guest_id)
+            meta["revoked"] = True
+            self._atomic_json(self.root / guest_id / "meta.json", meta)
+            return {**meta, "status": self.status(meta)}
+
+    def delete(self, guest_id: str) -> None:
+        with self._lock:
+            folder = (self.root / guest_id).resolve()
+            if folder.parent != self.root.resolve() or not folder.is_dir():
+                raise KeyError(guest_id)
+            shutil.rmtree(folder)
+
+    def candidates(self, embedding, *, camera: str, now: float | None = None, limit: int = 2) -> list[tuple[str, str, float]]:
+        now = float(now or time.time())
+        rows = []
+        with self._lock:
+            for meta in self.list():
+                if self.status(meta, now) != "active":
+                    continue
+                allowed = meta.get("allowed_cameras") or []
+                if allowed and camera not in allowed:
+                    continue
+                try:
+                    reference = np.load(self.root / meta["id"] / "embedding.npy")
+                    score = float(np.asarray(reference, dtype=np.float32) @ np.asarray(embedding, dtype=np.float32))
+                except (OSError, ValueError):
+                    continue
+                rows.append((f"guest:{meta['id']}", meta["name"], score))
+        return sorted(rows, key=lambda item: item[2], reverse=True)[:max(1, limit)]
+
+    def evaluate(self, guest_id: str, *, camera: str, score: float, runner_up_score: float,
+                 liveness_confirmed: bool, second_factor: bool, event_id: str | None = None) -> dict:
+        """Return and audit eligibility. Only a complete decision consumes an entry."""
+        with self._lock:
+            meta = self._meta(guest_id)
+            if not meta:
+                raise KeyError(guest_id)
+            reasons = []
+            if self.status(meta) != "active": reasons.append("guest_not_active")
+            if meta.get("allowed_cameras") and camera not in meta["allowed_cameras"]: reasons.append("camera_not_allowed")
+            if float(score) < self.threshold: reasons.append("score_too_low")
+            if float(score) - float(runner_up_score) < self.margin: reasons.append("margin_too_low")
+            if not liveness_confirmed: reasons.append("liveness_required")
+            if not second_factor: reasons.append("second_factor_required")
+            authorized = not reasons
+            if authorized:
+                meta["entries_used"] = int(meta.get("entries_used") or 0) + 1
+                self._atomic_json(self.root / guest_id / "meta.json", meta)
+            event = {
+                "id": uuid.uuid4().hex, "guest_id": guest_id, "guest_name": meta["name"],
+                "event_id": event_id, "camera": camera, "score": round(float(score), 4),
+                "authorized": authorized, "reasons": reasons, "ts": time.time(),
+            }
+            try:
+                history = json.loads(self.events_path.read_text(encoding="utf-8"))
+                if not isinstance(history, list): history = []
+            except (OSError, json.JSONDecodeError):
+                history = []
+            self._atomic_json(self.events_path, [event, *history[:999]])
+            return {**event, "entries_used": meta["entries_used"], "entries_left": max(0, int(meta["max_entries"]) - int(meta["entries_used"]))}
+
+    def history(self, limit: int = 100) -> list[dict]:
+        try:
+            rows = json.loads(self.events_path.read_text(encoding="utf-8"))
+            return rows[:max(1, min(int(limit), 1000))] if isinstance(rows, list) else []
+        except (OSError, json.JSONDecodeError):
+            return []

app/main.py

@@ -28,6 +28,8 @@
 from .visits import VisitService
 from .liveness import LivenessDetector
 from .access_control import AccessControl
+from .guest_access import GuestAccess
+from .site_intelligence import SiteIntelligence
 from .migrations import run_migrations
 
 BASE = Path(__file__).resolve().parent.parent
@@ -151,6 +153,14 @@ def main():
         data_dir / "access_control.json",
         enabled=bool(f.get("access_control_enabled", False)),
     )
+    processor.guest_access = GuestAccess(
+        data_dir,
+        threshold=float(f.get("guest_match_threshold", max(.62, processor.match_thr + .08))),
+        margin=float(f.get("guest_match_margin", max(.12, processor.match_margin))),
+    )
+    processor.site_intelligence = SiteIntelligence(
+        data_dir / "site_map.json", audit, processor.camera_profiles, processor.visits,
+    )
     processor.liveness = LivenessDetector(
         model_path=str(f.get("liveness_model_path") or "/opt/faceid/models/liveness.onnx"),
         enabled=bool(f.get("liveness_enabled", True)),

app/mqtt_listener.py

@@ -481,7 +481,19 @@ def _process_face(self, eid: str, st: dict, img, face, quality=None, source="sna
             if st["liveness"].get("confirmed"):
                 st["liveness_blocked"] = False
         emb = face.normed_embedding
-        candidates = self.gallery.match_candidates(emb, limit=2)
+        candidates = self.gallery.match_candidates(emb, limit=3)
+        guest_access = getattr(self, "guest_access", None)
+        if guest_access is not None:
+            guest_candidates = guest_access.candidates(
+                emb, camera=st["camera"], now=st.get("start_time"), limit=2,
+            )
+            combined = sorted([*candidates, *guest_candidates], key=lambda item: item[2], reverse=True)
+            if combined and str(combined[0][0]).startswith("guest:"):
+                lead = float(combined[0][2]) - float(combined[1][2] if len(combined) > 1 else 0)
+                if float(combined[0][2]) >= guest_access.threshold and lead >= guest_access.margin:
+                    candidates = combined[:2]
+                else:
+                    candidates = [item for item in combined if not str(item[0]).startswith("guest:")][:2]
         slug, name, score = candidates[0] if candidates else (None, None, 0.0)
         _, runner_up, runner_up_score = (
             candidates[1] if len(candidates) > 1 else (None, None, 0.0)
@@ -545,6 +557,21 @@ def _process_face(self, eid: str, st: dict, img, face, quality=None, source="sna
             st["best_score"], st["best_person"] = decision.score, decision.person
             st["done"] = True
             st["final_decision"] = decision
+            if decision.slug and str(decision.slug).startswith("guest:"):
+                guest_id = str(decision.slug).split(":", 1)[1]
+                st["guest"] = {"id": guest_id, "name": decision.person}
+                access_result = guest_access.evaluate(
+                    guest_id, camera=st["camera"], score=decision.score,
+                    runner_up_score=decision.runner_up_score,
+                    liveness_confirmed=bool((st.get("liveness") or {}).get("confirmed")),
+                    second_factor=False, event_id=eid,
+                )
+                st["guest_access"] = access_result
+                if self.client:
+                    self.client.publish(
+                        f"{self.prefix}/v1/guest_access",
+                        json.dumps(access_result, ensure_ascii=False), retain=False,
+                    )
             self._publish_recognition(
                 eid, st, decision.person, decision.score, decision=decision
             )
@@ -818,6 +845,8 @@ def _post_event(
             "scenario": scenario,
             "body": st.get("body"),
             "liveness": st.get("liveness"),
+            "guest": st.get("guest"),
+            "guest_access": st.get("guest_access"),
         }
         if self.client and st.get("body"):
             self.client.publish(