analytics/session_analyzer.py - master - volkovss/eye_tracker

analytics/session_analyzer.py
136 строк5 KB
volkovss
Initial commit: Eye Tracker v1.0
03 июн 2026, 13:18
03 июн 2026, 13:18f9d57a2
import os
import json
import base64
from datetime import datetime
from typing import Optional

import cv2
import numpy as np

from database.models import (
    get_db_session, GazeData, AggregatedStats,
    EnvironmentParams, Session as DBSession,
)
from analytics.fixations import IVTDetector
from analytics.heatmap   import generate_heatmap, generate_scanpath


class SessionAnalyzer:

    HEATMAP_DIR = os.path.join(
        os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
        "web", "static", "heatmaps",
    )

    def __init__(self, engine):
        self.engine = engine
        os.makedirs(self.HEATMAP_DIR, exist_ok=True)

    # полный анализ сессии
    def analyze(self, session_id: int) -> dict:
        db   = get_db_session(self.engine)
        rows = (
            db.query(GazeData)
            .filter_by(session_id=session_id)
            .order_by(GazeData.frame_index)
            .all()
        )

        if not rows:
            return {"error": "Нет данных взгляда для этой сессии"}

        env      = db.query(EnvironmentParams).filter_by(session_id=session_id).first()
        screen_w = (env.screen_width_px  if env and env.screen_width_px  else 1920)
        screen_h = (env.screen_height_px if env and env.screen_height_px else 1080)

        points = [
            {"x_px": r.x_px, "y_px": r.y_px, "confidence": r.confidence}
            for r in rows
        ]

        # детекция фиксаций (I-VT алгоритм)
        detector             = IVTDetector(
            velocity_threshold_px_s=30.0,
            min_fixation_ms=100.0,
            sample_rate_hz=30.0,
        )
        fixations, saccades  = detector.classify(points)
        stats                = detector.compute_stats(fixations, saccades)

        # качество данных
        good    = sum(1 for p in points if p["confidence"] >= 0.3)
        quality = good / max(len(points), 1) * 100

        # тепловая карта
        heatmap  = generate_heatmap(points, screen_w, screen_h, sigma=40)
        hm_path  = os.path.join(self.HEATMAP_DIR, f"s{session_id}_heatmap.png")
        cv2.imwrite(hm_path, heatmap)

        # scanpath
        scanpath = generate_scanpath(points, screen_w, screen_h, fixations=fixations)
        sp_path  = os.path.join(self.HEATMAP_DIR, f"s{session_id}_scanpath.png")
        cv2.imwrite(sp_path, scanpath)

        # сохранение / обновление AggregatedStats
        agg = db.query(AggregatedStats).filter_by(session_id=session_id).first()
        if not agg:
            agg = AggregatedStats(session_id=session_id)
            db.add(agg)

        agg.total_fixations          = stats["total_fixations"]
        agg.avg_fixation_duration_ms = stats["avg_fixation_duration_ms"]
        agg.total_fixation_time_ms   = stats["total_fixation_time_ms"]
        agg.saccade_count            = stats["saccade_count"]
        agg.avg_saccade_amplitude_px = stats["avg_saccade_amplitude_px"]
        agg.total_data_points        = len(rows)
        agg.data_quality_pct         = quality
        agg.heatmap_path             = hm_path
        agg.computed_at              = datetime.utcnow()
        db.commit()
        db.close()

        return {
            "session_id":   session_id,
            "total_points": len(rows),
            "quality_pct":  round(quality, 1),
            "screen_w":     screen_w,
            "screen_h":     screen_h,
            "stats":        stats,
            "fixations": [
                {"x": f.x, "y": f.y, "duration_ms": round(f.duration_ms, 1)}
                for f in fixations
            ],
            "saccades": [
                {"amplitude_px": round(s.amplitude_px, 1),
                 "start_x": round(s.start_x, 1), "start_y": round(s.start_y, 1),
                 "end_x":   round(s.end_x,   1), "end_y":   round(s.end_y,   1)}
                for s in saccades
            ],
        }

    # чтение картинок
    def get_image_b64(self, session_id: int, kind: str) -> Optional[str]:
        """kind: 'heatmap' или 'scanpath'"""
        path = os.path.join(self.HEATMAP_DIR, f"s{session_id}_{kind}.png")
        if not os.path.exists(path):
            return None
        with open(path, "rb") as f:
            return base64.b64encode(f.read()).decode()

    # сводка без пересчёта
    def get_summary(self, session_id: int) -> Optional[dict]:
        db  = get_db_session(self.engine)
        agg = db.query(AggregatedStats).filter_by(session_id=session_id).first()
        if not agg:
            return None
        result = {
            "total_fixations":          agg.total_fixations,
            "avg_fixation_duration_ms": agg.avg_fixation_duration_ms,
            "total_fixation_time_ms":   agg.total_fixation_time_ms,
            "saccade_count":            agg.saccade_count,
            "avg_saccade_amplitude_px": agg.avg_saccade_amplitude_px,
            "total_data_points":        agg.total_data_points,
            "data_quality_pct":         agg.data_quality_pct,
        }
        db.close()
        return result