tracker/capture.py - master - volkovss/eye_tracker

tracker/capture.py
176 строк6 KB
volkovss
Initial commit: Eye Tracker v1.0
03 июн 2026, 13:18
03 июн 2026, 13:18f9d57a2
import cv2
import threading
import time
import numpy as np
from typing import Optional, Tuple
import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config import FRAME_WIDTH, FRAME_HEIGHT, FRAME_FPS


class Camera:
    """
    Потокобезопасная обёртка для одной камеры.
    Считывание кадров происходит в фоновом потоке,
    чтобы основной цикл не блокировался на read().
    """

    def __init__(self, index: int, name: str = "cam"):
        self.index = index
        self.name  = name
        self.cap: Optional[cv2.VideoCapture] = None

        self._frame: Optional[np.ndarray] = None
        self._lock    = threading.Lock()
        self._running = False
        self._thread: Optional[threading.Thread] = None

        self.fps_actual  = 0.0
        self._frame_cnt  = 0
        self._fps_ts     = time.time()

    # открыть камеру
    def open(self) -> bool:
        # Сначала пробуем DirectShow (Windows), потом любой бэкенд
        for backend in (cv2.CAP_DSHOW, cv2.CAP_ANY):
            cap = cv2.VideoCapture(self.index, backend)
            if cap.isOpened():
                cap.set(cv2.CAP_PROP_FRAME_WIDTH,  FRAME_WIDTH)
                cap.set(cv2.CAP_PROP_FRAME_HEIGHT, FRAME_HEIGHT)
                cap.set(cv2.CAP_PROP_FPS,          FRAME_FPS)
                cap.set(cv2.CAP_PROP_BUFFERSIZE,   1)
                self.cap = cap
                return True
        return False

    # запустить захват 
    def start(self):
        if not self.open():
            raise RuntimeError(f"[Camera] Не удалось открыть камеру {self.index} ({self.name})")
        self._running = True
        self._thread  = threading.Thread(target=self._loop, daemon=True, name=f"cam-{self.name}")
        self._thread.start()

    # остановить захват 
    def stop(self):
        self._running = False
        if self._thread:
            self._thread.join(timeout=2.0)
        if self.cap:
            self.cap.release()
            self.cap = None

    # фоновый цикл чтения
    def _loop(self):
        while self._running:
            if self.cap is None:
                break
            ret, frame = self.cap.read()
            if not ret:
                time.sleep(0.005)
                continue

            frame = cv2.resize(frame, (FRAME_WIDTH, FRAME_HEIGHT))

            with self._lock:
                self._frame = frame

            self._frame_cnt += 1
            now     = time.time()
            elapsed = now - self._fps_ts
            if elapsed >= 1.0:
                self.fps_actual = self._frame_cnt / elapsed
                self._frame_cnt = 0
                self._fps_ts    = now

    # прочитать последний кадр
    def read(self) -> Tuple[bool, Optional[np.ndarray]]:
        with self._lock:
            if self._frame is None:
                return False, None
            return True, self._frame.copy()

    @property
    def is_opened(self) -> bool:
        return self.cap is not None and self.cap.isOpened()


class DualCameraCapture:
    """
    Управляет двумя камерами одновременно.
    Обе камеры работают независимо в своих потоках,
    read() возвращает последние доступные кадры с обеих.

    mode: "binocular" | "left_only" | "right_only"
    """

    MODE_BINOCULAR  = "binocular"
    MODE_LEFT_ONLY  = "left_only"
    MODE_RIGHT_ONLY = "right_only"

    def __init__(self, left_index: int = 0, right_index: int = 1,
                 mode: str = "binocular"):
        self.cam_left  = Camera(left_index,  "left")
        self.cam_right = Camera(right_index, "right")
        self._frame_index = 0
        self.mode = mode

    def start(self) -> dict:
        """
        Запускает камеры в соответствии с режимом (mode).
        Возвращает словарь {"left": bool, "right": bool}.
        """
        status = {"left": False, "right": False}
        if self.mode in (self.MODE_BINOCULAR, self.MODE_LEFT_ONLY):
            try:
                self.cam_left.start()
                status["left"] = True
                print(f"[DualCam] Левая камера ({self.cam_left.index}) — OK")
            except RuntimeError as e:
                print(f"[DualCam] ОШИБКА левой камеры: {e}")

        if self.mode in (self.MODE_BINOCULAR, self.MODE_RIGHT_ONLY):
            try:
                self.cam_right.start()
                status["right"] = True
                print(f"[DualCam] Правая камера ({self.cam_right.index}) — OK")
            except RuntimeError as e:
                print(f"[DualCam] ОШИБКА правой камеры: {e}")

        print(f"[DualCam] Режим: {self.mode}")
        return status

    def stop(self):
        self.cam_left.stop()
        self.cam_right.stop()

    def read(self) -> Tuple[Optional[np.ndarray], Optional[np.ndarray], int]:
        """
        Возвращает (left_frame, right_frame, frame_index).
        В монокулярном режиме отключённая камера возвращает None.
        """
        fl, fr = None, None
        if self.mode in (self.MODE_BINOCULAR, self.MODE_LEFT_ONLY):
            ok_l, fl = self.cam_left.read()
            fl = fl if ok_l else None
        if self.mode in (self.MODE_BINOCULAR, self.MODE_RIGHT_ONLY):
            ok_r, fr = self.cam_right.read()
            fr = fr if ok_r else None
        self._frame_index += 1
        return fl, fr, self._frame_index

    def set_mode(self, mode: str):
        """Переключает режим камеры (без перезапуска)."""
        assert mode in (self.MODE_BINOCULAR, self.MODE_LEFT_ONLY, self.MODE_RIGHT_ONLY)
        self.mode = mode
        print(f"[DualCam] Режим изменён: {mode}")

    @property
    def status(self) -> dict:
        return {
            "left_fps":  round(self.cam_left.fps_actual,  1),
            "right_fps": round(self.cam_right.fps_actual, 1),
            "left_ok":   self.cam_left.is_opened,
            "right_ok":  self.cam_right.is_opened,
            "mode":      self.mode,
        }