dsh/dual_camera_system/tests/test_spatial_scanner.py

import sys
import os
import tempfile
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import cv2
import numpy as np
import pytest
from core.spatial_scanner import SpatialScanner
from core.coord_utils import compute_sample_grid


class FakePTZ:
    def __init__(self):
        self.positions = []
    def goto_exact_position(self, pan, tilt, zoom):
        self.positions.append((pan, tilt, zoom))


def test_spatial_scanner_grid(tmp_path):
    ptz = FakePTZ()
    counter = {"n": 0}

    def frame_source():
        counter["n"] += 1
        return np.zeros((120, 160, 3), dtype=np.uint8)

    scanner = SpatialScanner("g1", ptz, frame_source, str(tmp_path), stabilize_time=0.0)
    result = scanner.run(pan_range=(0, 90), tilt_layers=(-10, 0), pan_step=30, zoom=1)
    assert len(result["samples"]) == 6
    assert result["panorama_path"] is not None
    assert len(ptz.positions) == 6


def test_invalid_pan_step_raises_value_error(tmp_path):
    ptz = FakePTZ()
    scanner = SpatialScanner("g1", ptz, lambda: None, str(tmp_path), stabilize_time=0.0)
    with pytest.raises(ValueError, match="pan_step must be positive"):
        scanner.run(pan_range=(0, 90), tilt_layers=(-10, 0), pan_step=0)


def test_compute_sample_grid_validation():
    with pytest.raises(ValueError, match="pan_step must be positive"):
        compute_sample_grid(pan_step=0)
    with pytest.raises(ValueError, match="tilt_layers must not be empty"):
        compute_sample_grid(tilt_layers=())
    with pytest.raises(ValueError, match="pan_range start must be less than end"):
        compute_sample_grid(pan_range=(180.0, 180.0))


def test_cancellation_stops_early(tmp_path):
    ptz = FakePTZ()
    counter = {"n": 0}

    def frame_source():
        counter["n"] += 1
        if counter["n"] == 2:
            scanner.cancel()
        return np.zeros((120, 160, 3), dtype=np.uint8)

    scanner = SpatialScanner("g1", ptz, frame_source, str(tmp_path), stabilize_time=0.0)
    result = scanner.run(pan_range=(0, 60), tilt_layers=(-10, 0), pan_step=30, zoom=1)
    assert len(result["samples"]) < 4
    assert scanner.progress["state"] == "cancelled"


def test_empty_sample_list_returns_no_panorama(tmp_path):
    ptz = FakePTZ()
    scanner = SpatialScanner("g1", ptz, lambda: None, str(tmp_path), stabilize_time=0.0)
    scanner._wait_frame = lambda timeout: None
    result = scanner.run(pan_range=(0, 60), tilt_layers=(-10, 0), pan_step=30, zoom=1)
    assert result["samples"] == []
    assert result["panorama_path"] is None
    assert scanner.progress["current"] == 0


def test_progress_callback_invoked(tmp_path):
    ptz = FakePTZ()
    progress_snapshots = []

    def frame_source():
        return np.zeros((120, 160, 3), dtype=np.uint8)

    def progress_callback(progress):
        progress_snapshots.append(dict(progress))

    scanner = SpatialScanner("g1", ptz, frame_source, str(tmp_path), stabilize_time=0.0)
    result = scanner.run(
        pan_range=(0, 60),
        tilt_layers=(-10, 0),
        pan_step=30,
        zoom=1,
        progress_callback=progress_callback,
    )
    expected_samples = len(result["samples"])
    assert expected_samples > 0
    # 扫描开始前会报告一次初始进度，之后每成功采集一个点报告一次
    assert len(progress_snapshots) == expected_samples + 1
    assert progress_snapshots[0]["current"] == 0
    assert progress_snapshots[0]["state"] == "scanning"
    assert progress_snapshots[-1]["current"] == expected_samples
    assert progress_snapshots[-1]["state"] == "scanning"


def test_prerun_cancellation_returns_empty_result(tmp_path):
    ptz = FakePTZ()
    scanner = SpatialScanner("g1", ptz, lambda: None, str(tmp_path), stabilize_time=0.0)
    scanner.cancel()
    result = scanner.run(pan_range=(0, 60), tilt_layers=(-10, 0), pan_step=30, zoom=1)
    assert result["samples"] == []
    assert result["panorama_path"] is None
    assert scanner.progress["state"] == "cancelled"


def test_panorama_does_not_blend_overlapping_samples(tmp_path):
    """重叠区域应直接覆盖，而不是加权融合导致虚化。"""
    ptz = FakePTZ()
    calls = {"n": 0}

    def frame_source():
        calls["n"] += 1
        # 第一张红色，第二张蓝色，两者水平视场约 55°，在 0°/30° 处明显重叠
        color = (0, 0, 255) if calls["n"] == 1 else (255, 0, 0)
        return np.full((100, 100, 3), color, dtype=np.uint8)

    scanner = SpatialScanner("g1", ptz, frame_source, str(tmp_path), stabilize_time=0.0)
    result = scanner.run(pan_range=(0, 60), tilt_layers=(0,), pan_step=30, zoom=1)
    assert result["panorama_path"] is not None
    panorama = cv2.imread(result["panorama_path"])
    assert panorama is not None

    # 在第二张图（pan=30°）的中心区域采样；直接覆盖应接近蓝色
    height, width = panorama.shape[:2]
    u = int((30 / 60) * width)
    v = int(((90 - 0) / 180) * height)
    roi = panorama[
        max(0, v - 50):min(height, v + 50),
        max(0, u - 50):min(width, u + 50),
    ]
    mask = roi.sum(axis=2) > 0
    assert mask.sum() > 0
    mean_color = roi[mask].mean(axis=0)

    blue = np.array([255, 0, 0], dtype=np.float32)
    blend = np.array([127, 0, 127], dtype=np.float32)
    dist_to_blue = np.linalg.norm(mean_color - blue)
    dist_to_blend = np.linalg.norm(mean_color - blend)

    # 均值应更接近蓝色，而不是红蓝融合色
    assert dist_to_blue < 60
    assert dist_to_blend > 90