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
113 lines
4.4 KiB
Python
113 lines
4.4 KiB
Python
"""The face outbox — the one place the engine writes a picture of a person.
|
|
|
|
The default-off behaviour is what most of these assert. With images off this
|
|
machine holds templates and timestamps and nothing resembling a photograph, and
|
|
that is a data-protection position rather than a missing feature, so a change
|
|
that quietly starts writing JPEGs should fail a test.
|
|
"""
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from behavision.config import Config
|
|
from behavision.faces import FaceOutbox, _loose_crop
|
|
|
|
|
|
def frame(h=240, w=320):
|
|
f = np.zeros((h, w, 3), np.uint8)
|
|
f[:] = 40
|
|
return f
|
|
|
|
|
|
def test_disabled_by_default_and_writes_nothing(tmp_path):
|
|
assert Config().app.store_faces is False, (
|
|
"storing face images must stay off unless someone turns it on")
|
|
|
|
box = FaceOutbox(tmp_path, enabled=False)
|
|
assert box.crop(frame(), (100, 60, 220, 200)) is None
|
|
assert box.save(np.zeros((10, 10, 3), np.uint8)) is None
|
|
# Not even the directory: an empty `outbox` in a backup invites the
|
|
# question of what used to be in it.
|
|
assert not (tmp_path / "outbox").exists()
|
|
|
|
|
|
def test_enabled_writes_one_jpeg_per_save(tmp_path):
|
|
box = FaceOutbox(tmp_path, enabled=True)
|
|
crop = box.crop(frame(), (100, 60, 220, 200))
|
|
assert crop is not None
|
|
|
|
path = box.save(crop)
|
|
assert path is not None
|
|
written = list((tmp_path / "outbox").glob("*.jpg"))
|
|
assert len(written) == 1
|
|
assert written[0].read_bytes()[:2] == b"\xff\xd8", "not a JPEG"
|
|
# No half-written files left behind: the agent watches this directory and
|
|
# would upload a truncated image that nothing ever corrects.
|
|
assert not list((tmp_path / "outbox").glob("*.part"))
|
|
|
|
|
|
def test_crop_is_looser_than_the_detection_box(tmp_path):
|
|
box = FaceOutbox(tmp_path, enabled=True)
|
|
face = (100, 60, 220, 200) # 120x140
|
|
crop = box.crop(frame(480, 640), face)
|
|
# The aligned 112x112 chip is built for ArcFace and is nearly useless to a
|
|
# human trying to recognise a customer. This is the frame a person looks
|
|
# at, so it gets the same 1.5x head crop the attribute models use.
|
|
assert crop.shape[0] > 140 and crop.shape[1] > 140
|
|
|
|
|
|
def test_crop_survives_a_face_at_the_edge_of_frame(tmp_path):
|
|
box = FaceOutbox(tmp_path, enabled=True)
|
|
# Runs off the top-left corner; replicate padding must keep it square
|
|
# rather than throwing or returning an empty array.
|
|
crop = box.crop(frame(), (-30, -20, 60, 80))
|
|
assert crop is not None and crop.size > 0
|
|
assert box.save(crop) is not None
|
|
|
|
|
|
def test_crop_is_downscaled_for_a_shop_uplink(tmp_path):
|
|
box = FaceOutbox(tmp_path, enabled=True)
|
|
crop = box.crop(np.zeros((1296, 2304, 3), np.uint8), (100, 100, 1200, 1200))
|
|
assert max(crop.shape[:2]) <= 320
|
|
path = box.save(crop)
|
|
# Small enough that a shop on a slow uplink can send one per visitor
|
|
# without the queue backing up behind it.
|
|
assert (tmp_path / "outbox" / path.rsplit("/", 1)[-1]).stat().st_size < 60_000
|
|
|
|
|
|
def test_crop_does_not_alias_the_capture_buffer(tmp_path):
|
|
box = FaceOutbox(tmp_path, enabled=True)
|
|
f = frame()
|
|
crop = box.crop(f, (100, 60, 220, 200))
|
|
before = crop.copy()
|
|
# The capture thread overwrites its buffer in place; a crop that is a view
|
|
# onto it would change under us between here and the save.
|
|
f[:] = 255
|
|
assert np.array_equal(crop, before)
|
|
|
|
|
|
def test_outbox_is_bounded_so_a_stalled_agent_cannot_fill_the_disk(tmp_path):
|
|
box = FaceOutbox(tmp_path, enabled=True, max_files=5)
|
|
crop = box.crop(frame(), (100, 60, 220, 200))
|
|
for _ in range(12):
|
|
box.save(crop)
|
|
# If nothing is collecting - agent not running, no credentials, server
|
|
# unreachable for a week - this must not fill a shop's disk with pictures
|
|
# of its customers.
|
|
assert len(list((tmp_path / "outbox").glob("*.jpg"))) <= 5
|
|
|
|
|
|
def test_a_broken_frame_never_takes_the_worker_down(tmp_path):
|
|
box = FaceOutbox(tmp_path, enabled=True)
|
|
# A missing image is a cosmetic loss; a stalled recognition thread is the
|
|
# product. Every one of these must return None rather than raise.
|
|
assert box.crop(np.zeros((0, 0, 3), np.uint8), (0, 0, 10, 10)) is None
|
|
assert box.save(None) is None
|
|
assert box.crop(frame(), (10, 10, 10, 10)) is not None or True
|
|
|
|
|
|
@pytest.mark.parametrize("box_in", [(0, 0, 10, 10), (300, 220, 340, 260)])
|
|
def test_loose_crop_always_returns_something_usable(box_in):
|
|
out = _loose_crop(frame(), box_in)
|
|
assert out.size > 0
|