121 lines
4.7 KiB
Python
121 lines
4.7 KiB
Python
"""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)
|
|
)
|