Suriyakumarvijayanayagam
dad04e8cda
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
169 lines
6.6 KiB
Python
169 lines
6.6 KiB
Python
import numpy as np
|
|
import pytest
|
|
|
|
from behavision.config import RecognitionSection
|
|
from behavision.gallery import Gallery, IdentityStore, VectorIndex
|
|
|
|
DIM = 16
|
|
|
|
|
|
def _unit(seed):
|
|
rng = np.random.default_rng(seed)
|
|
v = rng.normal(size=DIM).astype(np.float32)
|
|
return v / np.linalg.norm(v)
|
|
|
|
|
|
@pytest.fixture
|
|
def gallery(tmp_path):
|
|
store = IdentityStore(tmp_path / "test.db")
|
|
cfg = RecognitionSection(sighting_cooldown_seconds=0.0)
|
|
gal = Gallery(store, VectorIndex(DIM), cfg)
|
|
yield gal
|
|
store.close()
|
|
|
|
|
|
def test_auto_enroll_then_recognize(gallery):
|
|
emb = _unit(1)
|
|
first = gallery.resolve(emb, quality=0.9, camera_id="cam1")
|
|
assert first.kind == "new"
|
|
assert first.label.startswith("Visitor")
|
|
|
|
again = gallery.resolve(emb, quality=0.9, camera_id="cam1")
|
|
assert again.kind == "known"
|
|
assert again.identity_id == first.identity_id
|
|
assert again.similarity == pytest.approx(1.0, abs=1e-5)
|
|
|
|
|
|
def test_low_quality_face_is_not_enrolled(gallery):
|
|
res = gallery.resolve(_unit(2), quality=0.1, camera_id="cam1")
|
|
assert res.kind == "skipped"
|
|
assert gallery.store.stats()["identities"] == 0
|
|
|
|
|
|
def test_ambiguous_zone_creates_nothing(gallery):
|
|
base = _unit(3)
|
|
gallery.resolve(base, quality=0.9, camera_id="cam1")
|
|
# Build a vector with similarity between enroll(0.32) and match(0.42).
|
|
other = _unit(4)
|
|
other -= (other @ base) * base
|
|
other /= np.linalg.norm(other)
|
|
mid = 0.37 * base + np.sqrt(1 - 0.37 ** 2) * other
|
|
res = gallery.resolve(mid.astype(np.float32), quality=0.9, camera_id="cam1")
|
|
assert res.kind == "ambiguous"
|
|
assert gallery.store.stats()["identities"] == 1
|
|
|
|
|
|
def test_explicit_enroll_and_delete(gallery):
|
|
identity_id = gallery.enroll("Alice", [_unit(5), _unit(6)])
|
|
res = gallery.resolve(_unit(5), quality=0.9, camera_id="cam1")
|
|
assert res.kind == "known"
|
|
assert res.label == "Alice"
|
|
assert gallery.delete_identity(identity_id)
|
|
res2 = gallery.resolve(_unit(5), quality=0.9, camera_id="cam1")
|
|
assert res2.kind == "new" # gone from index too
|
|
|
|
|
|
def test_sighting_cooldown(tmp_path):
|
|
store = IdentityStore(tmp_path / "cd.db")
|
|
cfg = RecognitionSection(sighting_cooldown_seconds=9999)
|
|
gal = Gallery(store, VectorIndex(DIM), cfg)
|
|
emb = _unit(7)
|
|
gal.resolve(emb, quality=0.9, camera_id="cam1")
|
|
res = gal.resolve(emb, quality=0.9, camera_id="cam1")
|
|
assert res.kind == "known"
|
|
assert res.new_sighting is False
|
|
store.close()
|
|
|
|
|
|
def test_attributes_are_persisted_on_both_paths(gallery):
|
|
emb = _unit(8)
|
|
first = gallery.resolve(emb, quality=0.9, camera_id="cam1",
|
|
attributes={"gender": "Male", "age": 41})
|
|
assert first.kind == "new"
|
|
row = gallery.store.recent_sightings(1)[0]
|
|
assert row["attributes"] == {"gender": "Male", "age": 41}
|
|
|
|
again = gallery.resolve(emb, quality=0.9, camera_id="cam1",
|
|
attributes={"gender": "Male", "emotion": "neutral"})
|
|
assert again.kind == "known"
|
|
row = gallery.store.recent_sightings(1)[0]
|
|
assert row["attributes"] == {"gender": "Male", "emotion": "neutral"}
|
|
|
|
|
|
def test_sighting_without_attributes_stays_null(gallery):
|
|
gallery.resolve(_unit(9), quality=0.9, camera_id="cam1")
|
|
assert gallery.store.recent_sightings(1)[0]["attributes"] is None
|
|
|
|
|
|
def test_reinforce_fills_out_an_identity_born_with_one_embedding(gallery):
|
|
"""The cold-gallery fix: an identity created from a single view must be
|
|
able to accumulate other views during the same visit."""
|
|
base = _unit(20)
|
|
res = gallery.resolve(base, quality=0.9, camera_id="cam1")
|
|
assert gallery.store.embedding_count(res.identity_id) == 1
|
|
|
|
# a genuinely different view of the same person (sim below reinforce_threshold)
|
|
other = _unit(21)
|
|
other -= (other @ base) * base
|
|
other /= np.linalg.norm(other)
|
|
view2 = (0.5 * base + np.sqrt(1 - 0.25) * other).astype(np.float32)
|
|
assert gallery.reinforce_identity(res.identity_id, view2, quality=0.9)
|
|
assert gallery.store.embedding_count(res.identity_id) == 2
|
|
|
|
|
|
def test_reinforce_refuses_a_near_duplicate(gallery):
|
|
res = gallery.resolve(_unit(22), quality=0.9, camera_id="cam1")
|
|
# identical view adds nothing (sim 1.0 >= reinforce_threshold)
|
|
assert not gallery.reinforce_identity(res.identity_id, _unit(22), quality=0.9)
|
|
assert gallery.store.embedding_count(res.identity_id) == 1
|
|
|
|
|
|
def test_reinforce_refuses_low_quality_and_respects_the_cap(gallery):
|
|
res = gallery.resolve(_unit(23), quality=0.9, camera_id="cam1")
|
|
assert not gallery.reinforce_identity(res.identity_id, _unit(24), quality=0.1)
|
|
for i in range(10):
|
|
gallery.reinforce_identity(res.identity_id, _unit(30 + i), quality=0.9)
|
|
assert (gallery.store.embedding_count(res.identity_id)
|
|
<= gallery.cfg.max_embeddings_per_identity)
|
|
|
|
|
|
def test_reinforce_will_not_attach_another_persons_face(gallery):
|
|
"""A track that drifts onto a different face must not poison the gallery."""
|
|
a = gallery.resolve(_unit(40), quality=0.9, camera_id="cam1")
|
|
b = gallery.resolve(_unit(41), quality=0.9, camera_id="cam1")
|
|
assert a.identity_id != b.identity_id
|
|
# b's own vector offered as if it were a: top match is b, so refuse
|
|
assert not gallery.reinforce_identity(a.identity_id, _unit(41), quality=0.9)
|
|
assert gallery.store.embedding_count(a.identity_id) == 1
|
|
|
|
|
|
def _view(base, sim, seed):
|
|
"""A unit vector at a chosen cosine similarity to `base`."""
|
|
other = _unit(seed)
|
|
other -= (other @ base) * base
|
|
other /= np.linalg.norm(other)
|
|
return (sim * base + np.sqrt(1 - sim ** 2) * other).astype(np.float32)
|
|
|
|
|
|
def test_reinforce_refuses_a_view_it_would_call_a_different_person(gallery):
|
|
"""Below enroll_threshold, resolve() would mint a NEW identity - so
|
|
attaching the same vector to an existing one contradicts it. Measured on
|
|
the overhead camera: without this floor one identity held two vectors
|
|
0.195 apart."""
|
|
base = _unit(50)
|
|
res = gallery.resolve(base, quality=0.9, camera_id="cam1")
|
|
weak = _view(base, 0.20, 51) # below enroll_threshold 0.32
|
|
assert not gallery.reinforce_identity(res.identity_id, weak, quality=0.9)
|
|
assert gallery.store.embedding_count(res.identity_id) == 1
|
|
|
|
|
|
def test_reinforce_accepts_the_useful_band(gallery):
|
|
"""Between enroll_threshold and reinforce_threshold is exactly the view
|
|
worth learning: plausibly this person, usefully different."""
|
|
base = _unit(52)
|
|
res = gallery.resolve(base, quality=0.9, camera_id="cam1")
|
|
for sim, seed in ((0.35, 53), (0.50, 54)):
|
|
assert gallery.reinforce_identity(
|
|
res.identity_id, _view(base, sim, seed), quality=0.9), sim
|
|
assert gallery.store.embedding_count(res.identity_id) == 3
|