diff --git a/Brain/src/yovision_brain/vision/__init__.py b/Brain/src/yovision_brain/vision/__init__.py new file mode 100644 index 0000000..94943d4 --- /dev/null +++ b/Brain/src/yovision_brain/vision/__init__.py @@ -0,0 +1,16 @@ +"""Anonymous detection and single-stream tracking.""" + +from .detector import LumaBlobDetector, TorchLumaBlobDetector +from .models import BoundingBox, Detection, Detector, DetectorMetadata, TrackedObject +from .tracker import SingleStreamTracker + +__all__ = [ + "BoundingBox", + "Detection", + "Detector", + "DetectorMetadata", + "LumaBlobDetector", + "SingleStreamTracker", + "TorchLumaBlobDetector", + "TrackedObject", +] diff --git a/Brain/src/yovision_brain/vision/detector.py b/Brain/src/yovision_brain/vision/detector.py new file mode 100644 index 0000000..1e26806 --- /dev/null +++ b/Brain/src/yovision_brain/vision/detector.py @@ -0,0 +1,120 @@ +"""Deterministic anonymous blob detectors with no biometric semantics.""" + +from __future__ import annotations + +from collections.abc import Sequence + +from yovision_brain.decode import DecodedFrame, DecoderError + +from .models import BoundingBox, Detection, DetectorMetadata + +_METADATA = DetectorMetadata( + name="yovision-luma-blob", + version="1.0.0", + source="YoVision Brain first-party deterministic algorithm", + license="No external model license; no learned weights are distributed", + weights="none", +) + + +def _components(mask: Sequence[Sequence[bool]], minimum_area: int) -> tuple[BoundingBox, ...]: + height = len(mask) + width = len(mask[0]) if height else 0 + visited: set[tuple[int, int]] = set() + boxes: list[BoundingBox] = [] + for y in range(height): + for x in range(width): + if not mask[y][x] or (x, y) in visited: + continue + pending = [(x, y)] + visited.add((x, y)) + points: list[tuple[int, int]] = [] + while pending: + current_x, current_y = pending.pop() + points.append((current_x, current_y)) + for neighbor in ( + (current_x - 1, current_y), + (current_x + 1, current_y), + (current_x, current_y - 1), + (current_x, current_y + 1), + ): + nx, ny = neighbor + if 0 <= nx < width and 0 <= ny < height and mask[ny][nx] and neighbor not in visited: + visited.add(neighbor) + pending.append(neighbor) + if len(points) >= minimum_area: + xs, ys = zip(*points) + boxes.append(BoundingBox(min(xs), min(ys), max(xs) + 1, max(ys) + 1)) + return tuple(sorted(boxes, key=lambda box: (box.top, box.left, box.bottom, box.right))) + + +def _validate_frame(frame: DecodedFrame) -> None: + if frame.pixel_format not in {"rgb24", "yuv444p"}: + raise DecoderError(f"anonymous detector does not support pixel format {frame.pixel_format!r}") + expected = frame.width * frame.height * 3 + if len(frame.payload) != expected: + raise DecoderError(f"vision frame has {len(frame.payload)} bytes; expected {expected}") + + +class LumaBlobDetector: + """Small CPU reference detector used for deterministic integration tests.""" + + metadata = _METADATA + + def __init__(self, *, threshold: int = 200, minimum_area: int = 1) -> None: + if not 0 <= threshold <= 255 or minimum_area < 1: + raise ValueError("invalid luma detector threshold or minimum area") + self._threshold = threshold + self._minimum_area = minimum_area + + def detect(self, frame: DecodedFrame) -> tuple[Detection, ...]: + _validate_frame(frame) + if frame.pixel_format == "rgb24": + pixels = [ + max(frame.payload[index : index + 3]) + for index in range(0, len(frame.payload), 3) + ] + else: + pixels = list(frame.payload[: frame.width * frame.height]) + mask = [ + [pixels[y * frame.width + x] >= self._threshold for x in range(frame.width)] + for y in range(frame.height) + ] + return tuple( + Detection(box=box, category="anonymous_target", confidence=1.0) + for box in _components(mask, self._minimum_area) + ) + + +class TorchLumaBlobDetector: + """PyTorch CPU/GPU smoke backend; it contains no external model weights.""" + + metadata = DetectorMetadata( + name="yovision-torch-luma-blob", + version="1.0.0", + source="YoVision Brain first-party PyTorch tensor implementation", + license="PyTorch BSD-3-Clause; no external model weights", + weights="none", + ) + + def __init__(self, *, threshold: int = 200, minimum_area: int = 1, device: str = "cpu") -> None: + self._threshold = threshold + self._minimum_area = minimum_area + self._device = device + + def detect(self, frame: DecodedFrame) -> tuple[Detection, ...]: + _validate_frame(frame) + try: + import torch + except ImportError as exc: + raise RuntimeError("PyTorch runtime is required for TorchLumaBlobDetector") from exc + values = torch.tensor(list(frame.payload), dtype=torch.uint8, device=self._device) + if frame.pixel_format == "rgb24": + luma = values.reshape(frame.height, frame.width, 3).amax(dim=2) + else: + luma = values[: frame.width * frame.height].reshape(frame.height, frame.width) + mask = (luma >= self._threshold).cpu().tolist() + return tuple( + Detection(box=box, category="anonymous_target", confidence=1.0) + for box in _components(mask, self._minimum_area) + ) diff --git a/Brain/src/yovision_brain/vision/models.py b/Brain/src/yovision_brain/vision/models.py new file mode 100644 index 0000000..d1862b8 --- /dev/null +++ b/Brain/src/yovision_brain/vision/models.py @@ -0,0 +1,52 @@ +"""Privacy-preserving vision ports and observations.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Protocol + +from yovision_brain.decode import DecodedFrame + + +@dataclass(frozen=True, slots=True) +class DetectorMetadata: + name: str + version: str + source: str + license: str + weights: str + + +@dataclass(frozen=True, slots=True) +class BoundingBox: + left: int + top: int + right: int + bottom: int + + @property + def area(self) -> int: + return max(0, self.right - self.left) * max(0, self.bottom - self.top) + + +@dataclass(frozen=True, slots=True) +class Detection: + box: BoundingBox + category: str + confidence: float + + +@dataclass(frozen=True, slots=True) +class TrackedObject: + track_id: str + box: BoundingBox + category: str + confidence: float + frame_sequence: int + timestamp_ns: int + + +class Detector(Protocol): + metadata: DetectorMetadata + + def detect(self, frame: DecodedFrame) -> tuple[Detection, ...]: ... diff --git a/Brain/src/yovision_brain/vision/tracker.py b/Brain/src/yovision_brain/vision/tracker.py new file mode 100644 index 0000000..8eb6066 --- /dev/null +++ b/Brain/src/yovision_brain/vision/tracker.py @@ -0,0 +1,83 @@ +"""Session-local single-stream IoU tracker.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from .models import BoundingBox, Detection, TrackedObject + + +def _iou(first: BoundingBox, second: BoundingBox) -> float: + intersection = BoundingBox( + max(first.left, second.left), + max(first.top, second.top), + min(first.right, second.right), + min(first.bottom, second.bottom), + ).area + union = first.area + second.area - intersection + return intersection / union if union else 0.0 + + +@dataclass(slots=True) +class _Track: + track_id: str + detection: Detection + missed: int = 0 + + +class SingleStreamTracker: + """Tracks anonymous boxes only within one process session and one stream.""" + + def __init__(self, *, iou_threshold: float = 0.2, max_missed: int = 2) -> None: + if not 0.0 <= iou_threshold <= 1.0 or max_missed < 0: + raise ValueError("invalid tracker threshold or missed-frame limit") + self._iou_threshold = iou_threshold + self._max_missed = max_missed + self._tracks: dict[str, _Track] = {} + self._next_id = 1 + + def update( + self, + detections: tuple[Detection, ...], + *, + frame_sequence: int, + timestamp_ns: int, + ) -> tuple[TrackedObject, ...]: + unmatched_tracks = set(self._tracks) + results: list[TrackedObject] = [] + for detection in detections: + candidates = [ + (track_id, _iou(self._tracks[track_id].detection.box, detection.box)) + for track_id in unmatched_tracks + if self._tracks[track_id].detection.category == detection.category + ] + track_id, score = max(candidates, key=lambda item: item[1], default=("", -1.0)) + if score < self._iou_threshold: + track_id = f"track-{self._next_id:06d}" + self._next_id += 1 + self._tracks[track_id] = _Track(track_id, detection) + else: + unmatched_tracks.remove(track_id) + self._tracks[track_id].detection = detection + self._tracks[track_id].missed = 0 + results.append( + TrackedObject( + track_id=track_id, + box=detection.box, + category=detection.category, + confidence=detection.confidence, + frame_sequence=frame_sequence, + timestamp_ns=timestamp_ns, + ) + ) + for track_id in unmatched_tracks: + track = self._tracks[track_id] + track.missed += 1 + if track.missed > self._max_missed: + del self._tracks[track_id] + return tuple(results) + + def finish(self) -> tuple[str, ...]: + ended = tuple(sorted(self._tracks)) + self._tracks.clear() + return ended diff --git a/Brain/tests/fixtures/vision/README.md b/Brain/tests/fixtures/vision/README.md new file mode 100644 index 0000000..2817ea7 --- /dev/null +++ b/Brain/tests/fixtures/vision/README.md @@ -0,0 +1,5 @@ +# Brain vision fixtures + +Vision tests create anonymous geometric RGB frames in memory. Never add faces, +customer recordings, biometric templates, camera credentials, or unreviewed +model weights to this directory. diff --git a/Brain/tests/vision/test_detection_tracking.py b/Brain/tests/vision/test_detection_tracking.py new file mode 100644 index 0000000..e22f698 --- /dev/null +++ b/Brain/tests/vision/test_detection_tracking.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +import pytest + +from yovision_brain.decode import DecodedFrame +from yovision_brain.vision import ( + BoundingBox, + Detection, + LumaBlobDetector, + SingleStreamTracker, + TorchLumaBlobDetector, +) + + +def frame(payload: bytes, *, sequence: int = 0, width: int = 4, height: int = 3) -> DecodedFrame: + return DecodedFrame(sequence, sequence * 40_000_000, "camera", "main", width, height, "rgb24", payload) + + +def rgb(values: list[int]) -> bytes: + return b"".join(bytes((value, value, value)) for value in values) + + +def detection(left: int, top: int, right: int, bottom: int) -> Detection: + return Detection(BoundingBox(left, top, right, bottom), "anonymous_target", 0.9) + + +def test_detector_emits_only_anonymous_observations() -> None: + payload = rgb([0, 255, 255, 0, 0, 255, 255, 0, 0, 0, 0, 0]) + result = LumaBlobDetector(minimum_area=2).detect(frame(payload)) + assert result == (Detection(BoundingBox(1, 0, 3, 2), "anonymous_target", 1.0),) + assert LumaBlobDetector.metadata.weights == "none" + assert "external model license" in LumaBlobDetector.metadata.license + + +def test_empty_frame_has_no_detection() -> None: + assert LumaBlobDetector().detect(frame(rgb([0] * 12))) == () + + +def test_tracker_keeps_session_id_across_motion_and_short_occlusion() -> None: + tracker = SingleStreamTracker(iou_threshold=0.1, max_missed=2) + first = tracker.update((detection(0, 0, 3, 3),), frame_sequence=0, timestamp_ns=0) + assert first[0].track_id == "track-000001" + assert tracker.update((), frame_sequence=1, timestamp_ns=1) == () + resumed = tracker.update((detection(1, 0, 4, 3),), frame_sequence=2, timestamp_ns=2) + assert resumed[0].track_id == "track-000001" + assert tracker.finish() == ("track-000001",) + + +def test_disappeared_track_ends_and_new_target_gets_new_id() -> None: + tracker = SingleStreamTracker(max_missed=1) + first = tracker.update((detection(0, 0, 2, 2),), frame_sequence=0, timestamp_ns=0) + tracker.update((), frame_sequence=1, timestamp_ns=1) + tracker.update((), frame_sequence=2, timestamp_ns=2) + second = tracker.update((detection(0, 0, 2, 2),), frame_sequence=3, timestamp_ns=3) + assert first[0].track_id == "track-000001" + assert second[0].track_id == "track-000002" + + +def test_track_ids_are_session_local() -> None: + one = SingleStreamTracker().update((detection(0, 0, 1, 1),), frame_sequence=0, timestamp_ns=0) + two = SingleStreamTracker().update((detection(0, 0, 1, 1),), frame_sequence=0, timestamp_ns=0) + assert one[0].track_id == two[0].track_id == "track-000001" + + +def test_torch_backend_cpu_smoke_uses_no_external_weights() -> None: + pytest.importorskip("torch") + result = TorchLumaBlobDetector().detect(frame(rgb([0, 255] + [0] * 10))) + assert result[0].category == "anonymous_target" + assert TorchLumaBlobDetector.metadata.weights == "none"