"""Camera commissioning: is this camera placed well enough to recognise faces? The Office1 camera was installed, ran for weeks, and recognised almost nobody. Nothing was broken — the overhead angle tilted every face down and the frosted glass backlit them, so ArcFace never received a view it could embed stably. It took reading vectors out of SQLite by hand to find that out. This turns that diagnosis into an install step. The person installing walks past a few times and gets one of two answers: "this camera is good" or "move it to head height facing the approach direction". A site cannot be signed off broken and then discovered three weeks later from a footfall report that was always zero. It measures the *live pipeline*, not a separate probe: every finished track reports the best face quality it managed. That is the right question — not "were the frames sharp" but "did a person walking past produce at least one view worth enrolling" — and it is the same number `fraction_below_gate` on the dashboard is built from, so the wizard and the running system cannot disagree. """ from __future__ import annotations import threading import time from typing import Optional # Verdict boundaries, from measured data on real cameras (see CLAUDE.md): # frontal faces at head height score 0.70-0.82, the overhead corridor scores # 0.32-0.45 against a 0.65 gate. The fractions below are of faces that fall # under whatever gate that camera is configured with. GOOD_BELOW_GATE = 0.20 POOR_BELOW_GATE = 0.50 # A real walk-past varies; a static artifact does not. Frosted-glass tracks # measured a flat 0.37 on every frame, and a constant score across many # detections is the signature of a thing, not a person. FLAT_SPREAD = 0.03 FLAT_MIN_SAMPLES = 6 # Below this many faces the numbers are anecdote, not measurement. MIN_SAMPLES = 5 DEFAULT_SECONDS = 25.0 def quantile(ordered: "list[float]", frac: float) -> float: if not ordered: return 0.0 idx = min(len(ordered) - 1, int(round(frac * (len(ordered) - 1)))) return ordered[idx] class CommissionRun: """One timed placement check on one camera. Written by the worker thread as tracks end, read by API threads polling for the result, hence the lock. """ def __init__(self, camera_id: str, gate: float, seconds: float = DEFAULT_SECONDS, now: "float | None" = None): self.camera_id = camera_id self.gate = gate self.seconds = max(5.0, float(seconds)) self.started_at = now if now is not None else time.time() self._lock = threading.Lock() self._qualities: "list[float]" = [] # Frames on which at least one face was being tracked. A face in view # and a face that completed a pass are different observations, and # only the second one produces a quality sample. self._live_frames = 0 self._cancelled = False # -- written by the worker thread ----------------------------------- def record(self, best_quality: float, now: "float | None" = None) -> None: """One finished track's best view. Tracks that never held a face at all are not evidence about placement — they are evidence about detection — so they are dropped.""" if best_quality <= 0: return if not self.running(now): return with self._lock: self._qualities.append(float(best_quality)) def observe(self, live_faces: int, now: "float | None" = None) -> None: """One frame's worth of live tracking, whether or not anything ended. Without this the check cannot tell "the camera sees nobody" from "somebody is standing in front of it right now", because both produce zero finished tracks — and those two states need opposite advice. """ if live_faces <= 0 or not self.running(now): return with self._lock: self._live_frames += 1 # -- read by API threads -------------------------------------------- def running(self, now: "float | None" = None) -> bool: if self._cancelled: return False now = now if now is not None else time.time() return now - self.started_at < self.seconds def cancel(self) -> None: self._cancelled = True def report(self, now: "float | None" = None) -> dict: now = now if now is not None else time.time() with self._lock: ordered = sorted(self._qualities) live = self._live_frames running = self.running(now) out = { "camera_id": self.camera_id, "gate": round(self.gate, 3), "seconds": self.seconds, "elapsed": round(min(now - self.started_at, self.seconds), 1), "running": running, "cancelled": self._cancelled, "faces": len(ordered), "frames_with_a_face": live, "quality": _spread(ordered, self.gate), } out.update(self._verdict(ordered, running, live)) return out # -- internals ------------------------------------------------------ def _verdict(self, ordered: "list[float]", running: bool, live: int = 0) -> dict: n = len(ordered) if running: done = f"{n} pass{'' if n == 1 else 'es'} completed" # Saying "0 faces" while a face is plainly on screen reads as a # broken check, so report what is actually happening. seen = " · face in view" if live else "" return {"verdict": "running", "headline": f"watching… {done}{seen}", "advice": ["Walk past the camera the way a customer " "would, and out of the frame."]} if n == 0 and live: # A face was tracked the whole time and never left. The camera is # aimed correctly and the old advice ("check it is pointing at the # walkway") would send an installer to move a camera looking # straight at them — which is how a good camera gets made bad. return {"verdict": "no_completed_passes", "headline": "a face was in view, but nobody walked past", "advice": [ "The camera is detecting a face, so it is pointed " "correctly — but no one completed a pass.", "This check scores the best view of each person as " "they leave the frame, which is what recognition " "actually uses, so standing still measures nothing.", "Walk through the frame and out of it, a few times, " "then run the check again."]} if n == 0: # Streaming but nothing detected. Distinguishing this from "placed # badly" matters: the fix is completely different. return {"verdict": "no_faces", "headline": "no faces detected", "advice": [ "The camera is streaming but saw no face at all.", "Check it is pointing at the walkway, not the ceiling " "or floor, and that someone walked through the frame.", "If people did walk past, the view is too far, too " "dark, or too steep for the detector."]} below = sum(1 for q in ordered if q < self.gate) / n p50 = quantile(ordered, 0.50) spread = quantile(ordered, 0.95) - quantile(ordered, 0.05) if n >= FLAT_MIN_SAMPLES and spread < FLAT_SPREAD: # Every detection scoring the same is not a camera problem to # solve by moving it - it is not seeing people at all. return {"verdict": "artifact", "headline": f"every detection scored {p50:.2f} — this is " "probably not a face", "advice": [ "A constant score across every detection is the " "signature of a static object, not a person.", "Glass, a poster, a reflection or a mannequin in view " "will do this.", "Point the camera away from it, or raise " "detection.score_threshold for this camera."]} if n < MIN_SAMPLES: return {"verdict": "inconclusive", "headline": f"only {n} face{'' if n == 1 else 's'} seen — " "not enough to judge", "advice": [ f"Median quality was {p50:.2f}, but {n} " f"sample{'' if n == 1 else 's'} is anecdote, not " "measurement.", "Run the check again and walk past several times, " "ideally with more than one person."]} if below <= GOOD_BELOW_GATE: return {"verdict": "good", "headline": f"good placement — median quality {p50:.2f}", "advice": [ f"{below:.0%} of faces fell below the {self.gate:.2f} " "enrollment gate. This camera can enrol and recognise " "people reliably."]} if below <= POOR_BELOW_GATE: return {"verdict": "marginal", "headline": f"usable but weak — {below:.0%} of faces are " "below the gate", "advice": [ f"Median quality {p50:.2f} against a " f"{self.gate:.2f} gate: roughly {below:.0%} of " "visitors will be seen and then discarded.", "Angling it to face the approach direction, or " "lowering it toward head height, usually fixes this.", "If the position cannot change, lower this camera's " "min_enroll_quality — but only this camera's."]} return {"verdict": "poor", "headline": f"poor placement — {below:.0%} of faces are below " "the gate", "advice": [ f"Median quality {p50:.2f} against a {self.gate:.2f} " "gate. Most people who walk past will not be enrolled or " "recognised, and nothing will look broken.", "Move the camera to roughly head height, facing the " "direction people approach from.", "An overhead camera tilts every face downward, which is " "the single most common cause of this result.", "Backlighting — a window or lit glass behind the " "subject — is the second most common.", "Re-run this check after moving it. Do not lower the " "quality gate to make this message go away: it converts a " "visible miss into an invisible wrong match."]} def _spread(ordered: "list[float]", gate: float) -> dict: out = {"n": len(ordered)} if not ordered: return out out["p05"] = round(quantile(ordered, 0.05), 3) out["p50"] = round(quantile(ordered, 0.50), 3) out["p95"] = round(quantile(ordered, 0.95), 3) out["fraction_below_gate"] = round( sum(1 for v in ordered if v < gate) / len(ordered), 3) return out