dsh/dual_camera_system/tests/test_integration_polling.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
from polling_tracker import PollingTrackingCoordinator
from tracker import TrackedPerson, UltralyticsTracker


class FakePanorama:
    def __init__(self):
        self.connected = False
        self.streaming = False

    def connect(self):
        self.connected = True
        return True

    def start_stream_rtsp(self):
        self.streaming = True
        return True

    def get_frame(self):
        return np.zeros((480, 640, 3), dtype=np.uint8)

    def disconnect(self):
        self.connected = False
        self.streaming = False


class FakePTZ:
    def __init__(self):
        self.connected = False
        self.commands = []
        self.ptz_config = {"default_zoom": 8}

    def connect(self):
        self.connected = True
        return True

    def goto_exact_position(self, pan, tilt, zoom):
        self.commands.append((pan, tilt, zoom))
        return True

    def get_frame(self):
        return np.zeros((480, 640, 3), dtype=np.uint8)

    def get_current_position(self):
        return type("P", (), {"pan": 0, "tilt": 0, "zoom": 1})()

    def calculate_ptz_position(self, x, y, zoom=None):
        return x * 180, y * 90, zoom or 8

    def disconnect(self):
        self.connected = False


class FakeTracker:
    def __init__(self):
        self.frame_count = 0

    def update(self, frame):
        self.frame_count += 1
        return [
            TrackedPerson(track_id=1, bbox=(100, 100, 150, 200), center=(125, 150), confidence=0.9),
        ]


def test_integration_polling_start_stop():
    pan = FakePanorama()
    ptz = FakePTZ()
    tracker = FakeTracker()
    capture_dir = tempfile.mkdtemp(prefix="tracking_captures_")
    coord = PollingTrackingCoordinator(pan, ptz, tracker, config={
        "max_tracking_targets": 4,
        "ptz_stabilize_time": 0.05,
        "tracking_timeout": 1.0,
        "enable_upload": False,
        "capture_dir": capture_dir,
    })

    assert coord.start() is True
    import time
    time.sleep(0.3)
    coord.stop()

    assert pan.connected is False
    assert ptz.connected is False
    assert len(ptz.commands) > 0