Behavision/tests/test_live_picture.py

"""The live picture is the camera's latest frame, not the pipeline's.

Until this, the MJPEG stream served the frame the recognition pipeline had
most recently *finished* - so on a shop PC where detection plus identification
ran a few times a second, the live view ran a few times a second too, and every
picture it showed was already as old as that processing. It read as lag
because it was. These tests pin the decoupling: the picture comes from the
capture thread at its own rate, the boxes come from the pipeline at theirs,
and nothing is encoded for a camera nobody is watching.
"""
import time

import numpy as np

from behavision.config import CameraConfig, Config
from behavision.engine import CameraWorker
from behavision.tracking import Track


class StillSource:
    """A capture thread stand-in that hands out whatever frame it is given."""

    def __init__(self):
        self.frame = None
        self.ts = 0.0

    def set(self, frame):
        self.frame, self.ts = frame, time.time()

    def latest(self):
        return self.frame, self.ts

    def latest_since(self, known_ts):
        if self.frame is None or self.ts <= known_ts:
            return None, known_ts
        return self.frame, self.ts

    def stop(self):
        pass


def make_worker(tmp_path):
    cfg = Config()
    cfg.app.data_dir = tmp_path
    cam = CameraConfig(id="cam1", host="127.0.0.1", port=1, path="/none")
    w = CameraWorker(cam, cfg, detector=None, encoder=None, gallery=None,
                     bus=None, attrs=None)
    w.source = StillSource()
    return w


def grey(v=90):
    return np.full((120, 160, 3), v, dtype=np.uint8)


def resolved_track(box=(30, 30, 90, 100), label="Priya"):
    t = Track(id=1, box=box, kps=np.zeros((5, 2), dtype=np.float32), score=0.9)
    t.state, t.label, t.similarity = "resolved", label, 0.71
    return t


def test_the_picture_advances_with_the_camera_not_the_pipeline(tmp_path):
    w = make_worker(tmp_path)

    # Nothing captured yet: nothing to show, and no encode happened.
    assert w.latest_jpeg_since(0.0) == (None, 0.0)

    # A frame arrives from the camera. The pipeline has not touched it - and
    # the live view must not wait for it to.
    w.source.set(grey(80))
    jpeg1, ts1 = w.latest_jpeg_since(0.0)
    assert jpeg1 is not None and ts1 > 0

    # Same frame again: the stream asks "anything newer than ts1?" and the
    # answer is no. This is what stops duplicates going down the wire.
    assert w.latest_jpeg_since(ts1) == (None, ts1)

    # The camera produces a new frame; the pipeline still has not run.
    time.sleep(0.002)
    w.source.set(grey(160))
    jpeg2, ts2 = w.latest_jpeg_since(ts1)
    assert jpeg2 is not None and ts2 > ts1 and jpeg2 != jpeg1


def test_boxes_from_the_last_processed_frame_are_drawn_on_the_fresh_one(tmp_path):
    w = make_worker(tmp_path)
    w.source.set(grey())
    plain, _ = w.latest_jpeg_since(0.0)

    # The pipeline finishes a frame with one recognised person in it.
    w._remember_tracks([resolved_track()])

    # The NEXT camera frame - which the pipeline has not seen - still carries
    # the box, because a person does not vanish between two frames.
    time.sleep(0.002)
    w.source.set(grey())
    boxed, _ = w.latest_jpeg_since(0.0)
    assert boxed != plain, "a resolved track should be drawn on the live picture"


def test_a_stale_overlay_is_not_drawn(tmp_path):
    """A stalled pipeline must not leave a box floating over an empty spot."""
    w = make_worker(tmp_path)
    w.source.set(grey())
    plain, _ = w.latest_jpeg_since(0.0)

    w._remember_tracks([resolved_track()])
    # Pretend the pipeline last ran a while ago.
    with w._lock:
        w._overlay_ts = time.time() - 2.0

    time.sleep(0.002)
    w.source.set(grey())
    fresh, _ = w.latest_jpeg_since(0.0)
    assert fresh == plain, "boxes older than a second should not be drawn"


def test_only_tracks_matched_in_the_frame_are_drawn(tmp_path):
    """A track being coasted on misses has no face under it right now."""
    w = make_worker(tmp_path)
    missed = resolved_track()
    missed.misses = 3
    w._remember_tracks([missed])
    assert w._overlay == []

    seen = resolved_track()
    w._remember_tracks([seen])
    assert len(w._overlay) == 1
    (box, _color, text) = w._overlay[0]
    assert box == (30, 30, 90, 100) and text.startswith("Priya (")


def test_remembering_tracks_does_not_encode_or_copy(tmp_path):
    """The whole CPU argument: recording what to draw is a few tuples, and the
    frame is never touched. A shop PC with no viewer pays nothing."""
    w = make_worker(tmp_path)
    tracks = [resolved_track() for _ in range(5)]
    t0 = time.perf_counter()
    for _ in range(1000):
        w._remember_tracks(tracks)
    per_call_us = (time.perf_counter() - t0) / 1000 * 1e6
    # A JPEG encode of even a small frame is hundreds of microseconds; this
    # should be an order of magnitude under that.
    assert per_call_us < 100, f"remembering tracks took {per_call_us:.0f}us"