analytics/fixations.py - master - volkovss/eye_tracker

analytics/fixations.py
130 строк4 KB
volkovss
Initial commit: Eye Tracker v1.0
03 июн 2026, 13:18
03 июн 2026, 13:18f9d57a2
import numpy as np
from typing import List, Tuple
from dataclasses import dataclass


@dataclass
class Fixation:
    start_index: int
    end_index:   int
    duration_ms: float
    x: float
    y: float
    dispersion: float


@dataclass
class Saccade:
    start_index:  int
    end_index:    int
    amplitude_px: float
    start_x: float
    start_y: float
    end_x:   float
    end_y:   float


class IVTDetector:
    """
    Алгоритм I-VT (Velocity Threshold Identification).
    Классифицирует точки взгляда как фиксации или саккады
    на основании скорости перемещения взгляда.
    """

    def __init__(
        self,
        velocity_threshold_px_s: float = 30.0,
        min_fixation_ms:         float = 100.0,
        sample_rate_hz:          float = 30.0,
        confidence_threshold:    float = 0.3,
    ):
        self.vel_thresh    = velocity_threshold_px_s
        self.min_fix_ms    = min_fixation_ms
        self.sample_rate   = sample_rate_hz
        self.conf_thresh   = confidence_threshold

    def classify(
        self, gaze_points: List[dict]
    ) -> Tuple[List[Fixation], List[Saccade]]:
        """
        gaze_points — список словарей {x_px, y_px, confidence}.
        Возвращает (fixations, saccades).
        """
        valid = [p for p in gaze_points
                 if p.get("confidence", 0) >= self.conf_thresh]
        if len(valid) < 3:
            return [], []

        xs = np.array([p["x_px"] for p in valid], dtype=float)
        ys = np.array([p["y_px"] for p in valid], dtype=float)
        dt_s = 1.0 / self.sample_rate

        # скорость между соседними точками (px/s)
        vx = np.diff(xs) / dt_s
        vy = np.diff(ys) / dt_s
        vel = np.sqrt(vx**2 + vy**2)

        # True = фиксация, False = саккада
        is_fix = np.concatenate([[True], vel < self.vel_thresh])

        fixations: List[Fixation]  = []
        saccades:  List[Saccade]   = []
        dt_ms = dt_s * 1000

        i = 0
        while i < len(is_fix):
            j = i + 1
            while j < len(is_fix) and is_fix[j] == is_fix[i]:
                j += 1

            duration = (j - i) * dt_ms

            if is_fix[i]:
                if duration >= self.min_fix_ms:
                    fx, fy = float(np.mean(xs[i:j])), float(np.mean(ys[i:j]))
                    disp   = float(np.max(
                        np.sqrt((xs[i:j] - fx)**2 + (ys[i:j] - fy)**2)
                    ))
                    fixations.append(Fixation(
                        start_index=i, end_index=j,
                        duration_ms=duration,
                        x=fx, y=fy, dispersion=disp,
                    ))
            else:
                ei = min(j - 1, len(xs) - 1)
                amp = float(np.sqrt(
                    (xs[ei] - xs[i])**2 + (ys[ei] - ys[i])**2
                ))
                saccades.append(Saccade(
                    start_index=i, end_index=j,
                    amplitude_px=amp,
                    start_x=float(xs[i]),    start_y=float(ys[i]),
                    end_x=float(xs[ei]),     end_y=float(ys[ei]),
                ))

            i = j

        return fixations, saccades

    def compute_stats(
        self, fixations: List[Fixation], saccades: List[Saccade]
    ) -> dict:
        if not fixations:
            return {
                "total_fixations":          0,
                "avg_fixation_duration_ms": 0.0,
                "total_fixation_time_ms":   0.0,
                "saccade_count":            0,
                "avg_saccade_amplitude_px": 0.0,
            }

        durs = [f.duration_ms  for f in fixations]
        amps = [s.amplitude_px for s in saccades]

        return {
            "total_fixations":          len(fixations),
            "avg_fixation_duration_ms": float(np.mean(durs)),
            "total_fixation_time_ms":   float(np.sum(durs)),
            "saccade_count":            len(saccades),
            "avg_saccade_amplitude_px": float(np.mean(amps)) if amps else 0.0,
        }