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
187 lines
7.2 KiB
Python
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
|