Files
Behavision/tests/test_commission.py
Suriyakumarvijayanayagam dad04e8cda Behavision: face recognition for retail, edge to head office
Five components that ship as one product:

- behavision/  the recognition engine. RTSP ingest, YuNet detection, IoU
               tracking, ArcFace embeddings, a FAISS/SQLite gallery, and a
               FastAPI dashboard. Identity is decided once per TRACK from an
               average of at least three embeddings, never per frame.
- agent/       the Go edge agent: supervises the engine, holds a durable
               spool, and drains it to MQTT. Nothing is acked before the
               broker confirms.
- desktop/     the shop PC application (Wails + React + tray).
- server/      the cloud API, MQTT consumer, reports and assistant.
- web/         platform.loyaly.ai, the head-office app, embedded in the
               server binary.

The gallery stores 512-float embeddings and timestamps - no images unless
`app.store_faces` is switched on. Those embeddings are biometric personal
data under GDPR and India's DPDP: template inversion reconstructs a
recognisable face from an ArcFace vector, so data/behavision.db is treated
as a biometric database and DELETE /api/visitors/{id} is a real erasure.

CLAUDE.md carries the reasoning behind every non-obvious decision here,
including the ones that were measured and the ones that were wrong first.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HViLj9gYNRtSr7YVZmW5sn
2026-09-04 11:14:18 +05:30

187 lines
7.2 KiB
Python

"""Camera commissioning verdicts.
These messages are what an installer acts on at a customer site, so the
boundaries are tested against the real measured numbers: frontal faces at head
height score 0.70-0.82, the Office1 overhead corridor scored 0.32-0.45 against
a 0.65 gate, and frosted glass produced a flat 0.37 on every frame.
"""
import pytest
from behavision.commission import CommissionRun
GATE = 0.65
def _finished(qualities, gate=GATE, seconds=10.0):
"""A run that has already ended, holding these per-track qualities."""
run = CommissionRun("cam1", gate, seconds, now=0.0)
for q in qualities:
run.record(q, now=1.0)
return run.report(now=seconds + 1)
def test_a_well_placed_camera_passes():
r = _finished([0.72, 0.78, 0.70, 0.81, 0.75, 0.69])
assert r["verdict"] == "good"
assert r["quality"]["fraction_below_gate"] == 0.0
def test_the_office1_geometry_fails_with_a_placement_instruction():
"""The exact case this feature exists to catch: real faces, all under the
gate, nothing appearing broken."""
r = _finished([0.32, 0.38, 0.45, 0.41, 0.35, 0.44])
assert r["verdict"] == "poor"
assert r["quality"]["fraction_below_gate"] == 1.0
advice = " ".join(r["advice"]).lower()
assert "head height" in advice
# It must not suggest the shortcut that hides the problem.
assert "do not lower the quality gate" in advice
def test_a_mixed_camera_is_marginal_not_a_pass():
"""Half the visitors recognised is not a working camera, and calling it
one is how a site gets signed off broken."""
r = _finished([0.70, 0.72, 0.75, 0.40, 0.35, 0.38])
assert r["verdict"] == "marginal"
assert 0.2 < r["quality"]["fraction_below_gate"] <= 0.5
def test_no_faces_is_distinct_from_bad_placement():
"""The fixes are completely different — pointing the camera versus moving
it — so the verdicts must be too."""
r = _finished([])
assert r["verdict"] == "no_faces"
assert r["faces"] == 0
assert "walkway" in " ".join(r["advice"]).lower()
def test_a_constant_score_is_reported_as_an_artifact_not_a_face():
"""Frosted glass measured a flat 0.37 on every frame. A constant score
across many detections is the signature of a static object, and telling
the installer to move the camera would be wrong advice."""
r = _finished([0.37, 0.37, 0.371, 0.369, 0.37, 0.37, 0.37])
assert r["verdict"] == "artifact"
advice = " ".join(r["advice"]).lower()
assert "static" in advice
assert "head height" not in advice
def test_a_flat_score_above_the_gate_is_still_an_artifact():
"""The check is about the flatness, not the level - a bright poster can
score well and is still not a customer."""
r = _finished([0.74, 0.74, 0.741, 0.739, 0.74, 0.74])
assert r["verdict"] == "artifact"
def test_too_few_faces_is_inconclusive_not_a_verdict():
"""Three good samples is anecdote. Reporting it as a pass would sign off a
site on noise."""
r = _finished([0.75, 0.30, 0.72])
assert r["verdict"] == "inconclusive"
assert "again" in " ".join(r["advice"]).lower()
def test_verdict_uses_the_cameras_own_gate():
"""Gates describe a view. The same faces pass under a loosened per-camera
gate and fail under the strict default."""
faces = [0.50, 0.52, 0.48, 0.55, 0.51, 0.49]
assert _finished(faces, gate=0.65)["verdict"] == "poor"
assert _finished(faces, gate=0.45)["verdict"] == "good"
# -- run lifecycle ----------------------------------------------------------
def test_while_running_it_reports_progress_not_a_verdict():
run = CommissionRun("cam1", GATE, seconds=10.0, now=0.0)
run.record(0.7, now=1.0)
r = run.report(now=2.0)
assert r["verdict"] == "running"
assert r["running"] is True
assert r["faces"] == 1
assert r["elapsed"] == 2.0
def test_samples_after_the_window_are_ignored():
"""Otherwise a busy camera keeps changing its own verdict after the
installer has walked away and read the result."""
run = CommissionRun("cam1", GATE, seconds=10.0, now=0.0)
run.record(0.70, now=1.0)
run.record(0.10, now=99.0)
assert run.report(now=11.0)["faces"] == 1
def test_tracks_that_never_held_a_face_are_not_evidence():
"""best_quality 0 means no face was ever embedded on that track. Counting
it as a bad view would blame placement for a detection problem."""
run = CommissionRun("cam1", GATE, seconds=10.0, now=0.0)
run.record(0.0, now=1.0)
run.record(0.72, now=2.0)
assert run.report(now=11.0)["faces"] == 1
def test_cancel_stops_the_run_immediately():
run = CommissionRun("cam1", GATE, seconds=60.0, now=0.0)
assert run.running(now=1.0)
run.cancel()
assert not run.running(now=1.0)
run.record(0.9, now=2.0)
r = run.report(now=2.0)
assert r["cancelled"] is True
assert r["faces"] == 0
def test_a_very_short_window_is_clamped():
"""A 0-second check would report "no faces" before anyone could move."""
assert CommissionRun("cam1", GATE, seconds=0.0).seconds >= 5.0
def test_a_face_standing_still_is_not_reported_as_no_faces():
"""Found by running the dashboard: someone stands in front of the camera
to check it, their track never ends inside the window, and the check said
"no faces detected — check it is pointing at the walkway". That advice
moves a camera that is aimed correctly at a face."""
run = CommissionRun("cam1", gate=0.65, seconds=10, now=0.0)
for i in range(200): # a face in view for the whole window...
run.observe(1, now=0.1 * i)
r = run.report(now=20.0) # ...and not one completed pass
assert r["faces"] == 0
assert r["verdict"] == "no_completed_passes"
assert "walked past" in r["headline"]
joined = " ".join(r["advice"]).lower()
assert "pointed correctly" in joined
assert "walkway" not in joined, "must not advise re-aiming a working camera"
def test_a_truly_blind_camera_still_says_no_faces():
"""The distinction only earns its keep if the other branch survives."""
run = CommissionRun("cam1", gate=0.65, seconds=10, now=0.0)
r = run.report(now=20.0)
assert r["verdict"] == "no_faces"
assert "walkway" in " ".join(r["advice"])
def test_observe_is_ignored_once_the_window_closes():
"""Same rule record() already follows: the window is the measurement."""
run = CommissionRun("cam1", gate=0.65, seconds=10, now=0.0)
run.observe(1, now=50.0)
assert run.report(now=60.0)["verdict"] == "no_faces"
def test_progress_says_face_in_view_rather_than_zero_faces():
"""'watching... 0 faces so far' with a face plainly on screen reads as a
broken check, and is what made the underlying bug look normal."""
run = CommissionRun("cam1", gate=0.65, seconds=25, now=0.0)
run.observe(1, now=1.0)
r = run.report(now=2.0)
assert r["verdict"] == "running"
assert "face in view" in r["headline"]
assert r["frames_with_a_face"] == 1
def test_inconclusive_advice_reads_as_english():
"""Installer-facing copy: "1 samples is anecdote" is what a customer sees."""
run = CommissionRun("cam1", gate=0.65, seconds=10, now=0.0)
run.record(0.9, now=1.0)
advice = " ".join(run.report(now=20.0)["advice"])
assert "1 sample is" in advice and "1 samples" not in advice