196 lines
6.3 KiB
Python
196 lines
6.3 KiB
Python
"""Tests for CV region analysis stage — regression checks only."""
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from detect.models import BoundingBox, Frame
|
|
from detect.profiles.base import RegionAnalysisConfig
|
|
from detect.profiles.soccer import SoccerBroadcastProfile
|
|
|
|
|
|
# Load edges module directly — gpu/models/__init__.py has GPU-only imports
|
|
_spec = importlib.util.spec_from_file_location(
|
|
"cv_edges", Path("gpu/models/cv/edges.py"),
|
|
)
|
|
_edges_mod = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(_edges_mod)
|
|
detect_edges = _edges_mod.detect_edges
|
|
detect_edges_debug = _edges_mod.detect_edges_debug
|
|
|
|
|
|
def _make_frame(seq: int = 0, h: int = 1080, w: int = 1920) -> Frame:
|
|
"""Create a blank frame for testing."""
|
|
image = np.zeros((h, w, 3), dtype=np.uint8)
|
|
return Frame(sequence=seq, chunk_id=0, timestamp=seq * 0.5, image=image)
|
|
|
|
|
|
def _make_frame_with_lines(seq: int = 0) -> Frame:
|
|
"""Create a frame with two strong horizontal lines (simulates hoarding edges)."""
|
|
import cv2
|
|
|
|
image = np.zeros((1080, 1920, 3), dtype=np.uint8)
|
|
cv2.line(image, (100, 800), (1800, 800), (255, 255, 255), 3)
|
|
cv2.line(image, (100, 860), (1800, 860), (255, 255, 255), 3)
|
|
return Frame(sequence=seq, chunk_id=0, timestamp=seq * 0.5, image=image)
|
|
|
|
|
|
# --- Config ---
|
|
|
|
def test_soccer_profile_has_region_analysis_config():
|
|
profile = SoccerBroadcastProfile()
|
|
config = profile.region_analysis_config()
|
|
assert isinstance(config, RegionAnalysisConfig)
|
|
assert config.enabled is True
|
|
|
|
|
|
def test_region_analysis_config_defaults():
|
|
config = RegionAnalysisConfig()
|
|
assert config.edge_canny_low == 50
|
|
assert config.edge_canny_high == 150
|
|
assert config.edge_hough_threshold == 80
|
|
|
|
|
|
# --- Edge detection (GPU side, loaded standalone) ---
|
|
|
|
def test_detect_edges_blank_frame():
|
|
"""Blank frame should produce no regions."""
|
|
image = np.zeros((1080, 1920, 3), dtype=np.uint8)
|
|
results = detect_edges(image)
|
|
assert results == []
|
|
|
|
|
|
def test_detect_edges_with_lines():
|
|
"""Frame with parallel horizontal lines should produce at least one region."""
|
|
import cv2
|
|
|
|
image = np.zeros((1080, 1920, 3), dtype=np.uint8)
|
|
cv2.line(image, (100, 800), (1800, 800), (255, 255, 255), 3)
|
|
cv2.line(image, (100, 860), (1800, 860), (255, 255, 255), 3)
|
|
|
|
results = detect_edges(image)
|
|
assert len(results) >= 1
|
|
|
|
for r in results:
|
|
assert "x" in r and "y" in r and "w" in r and "h" in r
|
|
assert r["label"] == "edge_region"
|
|
assert 0 <= r["confidence"] <= 1
|
|
|
|
|
|
def test_detect_edges_returns_dict_format():
|
|
"""Each result must have the expected keys."""
|
|
import cv2
|
|
|
|
image = np.zeros((720, 1280, 3), dtype=np.uint8)
|
|
cv2.line(image, (50, 400), (1200, 400), (255, 255, 255), 2)
|
|
cv2.line(image, (50, 450), (1200, 450), (255, 255, 255), 2)
|
|
|
|
results = detect_edges(image)
|
|
if results:
|
|
r = results[0]
|
|
expected_keys = {"x", "y", "w", "h", "confidence", "label"}
|
|
assert set(r.keys()) == expected_keys
|
|
|
|
|
|
# --- Debug function ---
|
|
|
|
def test_detect_edges_debug_returns_overlays():
|
|
"""Debug function must return overlay images and counts."""
|
|
import cv2
|
|
|
|
image = np.zeros((1080, 1920, 3), dtype=np.uint8)
|
|
cv2.line(image, (100, 800), (1800, 800), (255, 255, 255), 3)
|
|
cv2.line(image, (100, 860), (1800, 860), (255, 255, 255), 3)
|
|
|
|
result = detect_edges_debug(image)
|
|
assert "regions" in result
|
|
assert "edge_overlay_b64" in result
|
|
assert "lines_overlay_b64" in result
|
|
assert "horizontal_count" in result
|
|
assert "pair_count" in result
|
|
assert isinstance(result["edge_overlay_b64"], str)
|
|
assert len(result["edge_overlay_b64"]) > 0 # non-empty base64
|
|
assert isinstance(result["lines_overlay_b64"], str)
|
|
assert len(result["lines_overlay_b64"]) > 0
|
|
assert result["horizontal_count"] >= 2
|
|
assert result["pair_count"] >= 1
|
|
assert len(result["regions"]) >= 1
|
|
|
|
|
|
def test_detect_edges_debug_blank_frame():
|
|
"""Debug on blank frame should still return structure."""
|
|
image = np.zeros((720, 1280, 3), dtype=np.uint8)
|
|
result = detect_edges_debug(image)
|
|
assert result["regions"] == []
|
|
assert result["horizontal_count"] == 0
|
|
assert result["pair_count"] == 0
|
|
assert isinstance(result["edge_overlay_b64"], str)
|
|
|
|
|
|
# --- Stage function ---
|
|
|
|
def test_stage_disabled(monkeypatch):
|
|
"""When disabled, returns empty dict."""
|
|
monkeypatch.setattr("detect.emit.push_detect_event", lambda *a, **kw: None)
|
|
|
|
from detect.stages.edge_detector import detect_edge_regions
|
|
|
|
config = RegionAnalysisConfig(enabled=False)
|
|
result = detect_edge_regions([_make_frame()], config, job_id="test")
|
|
assert result == {}
|
|
|
|
|
|
def test_stage_local_blank(monkeypatch):
|
|
"""Local mode on blank frames returns empty boxes."""
|
|
monkeypatch.setattr("detect.emit.push_detect_event", lambda *a, **kw: None)
|
|
|
|
from detect.stages.edge_detector import detect_edge_regions
|
|
|
|
config = RegionAnalysisConfig()
|
|
result = detect_edge_regions([_make_frame()], config, job_id="test")
|
|
assert isinstance(result, dict)
|
|
assert all(isinstance(v, list) for v in result.values())
|
|
|
|
|
|
def test_stage_local_with_lines(monkeypatch):
|
|
"""Local mode on frame with lines should find regions."""
|
|
monkeypatch.setattr("detect.emit.push_detect_event", lambda *a, **kw: None)
|
|
|
|
from detect.stages.edge_detector import detect_edge_regions
|
|
|
|
config = RegionAnalysisConfig()
|
|
frame = _make_frame_with_lines()
|
|
result = detect_edge_regions([frame], config, job_id="test")
|
|
|
|
boxes = result.get(frame.sequence, [])
|
|
assert len(boxes) >= 1
|
|
assert all(isinstance(b, BoundingBox) for b in boxes)
|
|
assert all(b.label == "edge_region" for b in boxes)
|
|
|
|
|
|
# --- Graph wiring ---
|
|
|
|
def test_detect_edges_in_nodes():
|
|
"""detect_edges must be in the pipeline node list."""
|
|
from detect.graph import NODES, NODE_FUNCTIONS
|
|
|
|
assert "detect_edges" in NODES
|
|
node_names = [name for name, _ in NODE_FUNCTIONS]
|
|
assert "detect_edges" in node_names
|
|
|
|
# Must be after filter_scenes, before detect_objects
|
|
idx = NODES.index("detect_edges")
|
|
assert NODES[idx - 1] == "filter_scenes"
|
|
assert NODES[idx + 1] == "detect_objects"
|
|
|
|
|
|
# --- State ---
|
|
|
|
def test_state_has_edge_regions_field():
|
|
from detect.state import DetectState
|
|
|
|
hints = DetectState.__annotations__
|
|
assert "edge_regions_by_frame" in hints
|