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 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
    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"