/
volkovss
/
eye_tracker
Обзор
Документация
Войти
/
volkovss
/
eye_tracker
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
tracker/session_manager.py
260 строк
9 KB
volkovss
Initial commit: Eye Tracker v1.0
03 июн 2026, 13:18
03 июн 2026, 13:18
f9d57a2
Код
Авторство
О чём код?
import threading import time from datetime import datetime, timezone from typing import Optional, List, Callable from dataclasses import dataclass from database.models import ( Session as DBSession, GazeData as DBGazeData, HardwareConfig, SoftwareConfig, EnvironmentParams, CalibrationData, Log, get_db_session, ) from tracker.calibration import GazeMapper from tracker.pupil import PupilResult from tracker.metadata import collect_all @dataclass class GazePoint: timestamp: str frame_index: int x_px: float y_px: float x_norm: float y_norm: float confidence: float left_pupil_x: Optional[float] left_pupil_y: Optional[float] right_pupil_x: Optional[float] right_pupil_y: Optional[float] class SessionManager: FLUSH_INTERVAL = 2.0 # секунды между записями в БД BUFFER_LIMIT = 300 # принудительный сброс при переполнении def __init__(self, engine): self.engine = engine self._db = get_db_session(engine) self.current_session: Optional[DBSession] = None self.gaze_mapper: Optional[GazeMapper] = None self._buffer: List[GazePoint] = [] self._lock = threading.Lock() self._thread: Optional[threading.Thread] = None self._running = False self.screen_w = 1920 self.screen_h = 1080 self.on_gaze: Optional[Callable] = None # колбэк для live UI # старт сессии def start_session( self, user_id: int, device_id: int, task_name: str, gaze_mapper: GazeMapper, capture_frame = None, ) -> int: meta = collect_all(capture_frame) # создаём запись сессии session = DBSession( user_id=user_id, device_id=device_id, task_name=task_name, start_ts=datetime.utcnow(), status="active" ) self._db.add(session) self._db.flush() sid = session.session_id # аппаратная конфигурация hw = meta["hardware"] self._db.add(HardwareConfig( session_id=sid, cpu_model=hw.get("cpu_model"), cpu_cores=hw.get("cpu_cores"), ram_mb=hw.get("ram_mb"), gpu_info=hw.get("gpu_info"), )) # программная конфигурация sw = meta["software"] self._db.add(SoftwareConfig( session_id=sid, os_name=sw.get("os_name"), os_version=sw.get("os_version"), python_version=sw.get("python_version"), opencv_version=sw.get("opencv_version"), app_version=sw.get("app_version"), )) # параметры окружения env = meta["environment"] self.screen_w = env.get("screen_width_px", 1920) self.screen_h = env.get("screen_height_px", 1080) self._db.add(EnvironmentParams( session_id=sid, ambient_lux_estimate=env.get("ambient_lux_estimate"), screen_width_px=self.screen_w, screen_height_px=self.screen_h, screen_dpi=env.get("screen_dpi"), distance_estimate_cm=60.0, lighting_condition=env.get("lighting_condition"), )) self._db.commit() self.current_session = session self.gaze_mapper = gaze_mapper # запускаем поток сброса буфера self._running = True self._thread = threading.Thread( target=self._flush_loop, daemon=True, name="gaze-flush" ) self._thread.start() self._log(sid, "INFO", f"Сессия {sid} запущена", "session") return sid # добавление точки взгляда def add_gaze( self, left_pupil: Optional[PupilResult], right_pupil: Optional[PupilResult], frame_index: int, ): if not self.current_session or not self.gaze_mapper: return # Расчёт экранной позиции взгляда gaze = self.gaze_mapper.predict_binocular( (left_pupil.x, left_pupil.y) if left_pupil else None, (right_pupil.x, right_pupil.y) if right_pupil else None, ) gx = gaze[0] if gaze else 0.0 gy = gaze[1] if gaze else 0.0 lc = left_pupil.confidence if left_pupil else 0.0 rc = right_pupil.confidence if right_pupil else 0.0 conf = (lc + rc) / 2 if (left_pupil and right_pupil) else max(lc, rc) pt = GazePoint( timestamp=datetime.now(timezone.utc).isoformat(), frame_index=frame_index, x_px=gx, y_px=gy, x_norm=gx / max(self.screen_w, 1), y_norm=gy / max(self.screen_h, 1), confidence=conf, left_pupil_x=left_pupil.x if left_pupil else None, left_pupil_y=left_pupil.y if left_pupil else None, right_pupil_x=right_pupil.x if right_pupil else None, right_pupil_y=right_pupil.y if right_pupil else None, ) with self._lock: self._buffer.append(pt) # Принудительный сброс при переполнении буфера if len(self._buffer) >= self.BUFFER_LIMIT: batch = self._buffer.copy() self._buffer.clear() threading.Thread(target=self._write_batch, args=(batch,), daemon=True).start() if self.on_gaze: self.on_gaze(pt) # остановка сессии def stop_session(self) -> int: self._running = False if self._thread: self._thread.join(timeout=5.0) self._flush_buffer() # финальный сброс if self.current_session: self.current_session.end_ts = datetime.utcnow() self.current_session.status = "completed" self._db.commit() sid = self.current_session.session_id self._log(sid, "INFO", "Сессия завершена", "session") self._db.commit() self.current_session = None return sid return -1 # внутренние методы def _flush_loop(self): while self._running: time.sleep(self.FLUSH_INTERVAL) self._flush_buffer() def _flush_buffer(self): with self._lock: if not self._buffer: return batch = self._buffer.copy() self._buffer.clear() self._write_batch(batch) def _write_batch(self, batch: list): if not self.current_session: return sid = self.current_session.session_id rows = [ DBGazeData( session_id=sid, timestamp=p.timestamp, frame_index=p.frame_index, x_px=p.x_px, y_px=p.y_px, x_norm=p.x_norm, y_norm=p.y_norm, confidence=p.confidence, left_pupil_x=p.left_pupil_x, left_pupil_y=p.left_pupil_y, right_pupil_x=p.right_pupil_x, right_pupil_y=p.right_pupil_y, ) for p in batch ] try: self._db.bulk_save_objects(rows) self._db.commit() except Exception as e: self._db.rollback() print(f"[SessionManager] Ошибка записи в БД: {e}") def _log(self, session_id: int, level: str, msg: str, module: str): self._db.add(Log( session_id=session_id, level=level, message=msg, module=module )) try: self._db.commit() except Exception: self._db.rollback() def save_calibration( self, calibration_json: str, polynomial_json: str, rms_error: float, num_points: int, ) -> int: if not self.current_session: raise RuntimeError("Нет активной сессии") cal = CalibrationData( session_id=self.current_session.session_id, calibration_points_json=calibration_json, polynomial_coeffs_json=polynomial_json, calibration_error_metric=rms_error, num_points=num_points, ) self._db.add(cal) self._db.commit() self.current_session.calibration_id = cal.cal_id self._db.commit() return cal.cal_id