129 lines
5.2 KiB
Python
129 lines
5.2 KiB
Python
"""Causal feature computation shared by offline training and the real-time filter.
|
|
|
|
Both `train_filter_model.py` (batch, pandas) and `adaptive_filter.py` (online,
|
|
per-sample) must derive identical features from identical inputs, or a model
|
|
trained offline will not behave the same way when driven by live samples.
|
|
This module is the single source of truth for that feature set.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
from collections import deque
|
|
|
|
FEATURE_WINDOW = 5
|
|
MAX_CONTINUOUS_GAP_S = 0.5
|
|
|
|
FEATURE_COLUMNS = [
|
|
"velocity_deg_s",
|
|
"accel_deg_s2",
|
|
"dispersion_deg",
|
|
"confidence",
|
|
"velocity_deg_s_mean",
|
|
"velocity_deg_s_std",
|
|
"dispersion_deg_mean",
|
|
]
|
|
|
|
EVENT_LABELS = ("fixation", "saccade", "pursuit")
|
|
|
|
|
|
def px_to_deg(distance_px: float, cm_per_px: float, viewing_distance_cm: float) -> float:
|
|
"""Convert a pixel distance to visual angle, given screen and viewing-distance calibration.
|
|
|
|
Every recording can be a different physical screen at a different
|
|
distance, so any pixel-unit feature (dispersion, acceleration) is not
|
|
comparable across sessions unless converted through this. velocity_deg_s
|
|
already goes through the equivalent conversion at record time.
|
|
"""
|
|
distance_cm = distance_px * cm_per_px
|
|
return math.degrees(2.0 * math.atan2(distance_cm / 2.0, viewing_distance_cm))
|
|
|
|
|
|
def add_rolling_features(df, session_col: str = "session_id"):
|
|
"""Add the *_mean / *_std rolling columns to a DataFrame in place, returning it.
|
|
|
|
Rolling windows are computed per session so one recording's trailing
|
|
samples never leak into the next. min_periods=1 keeps the first few
|
|
samples of a session usable (with a noisier estimate) instead of NaN.
|
|
"""
|
|
grouped = df.groupby(session_col, sort=False)
|
|
velocity_roll = grouped["velocity_deg_s"].rolling(FEATURE_WINDOW, min_periods=1)
|
|
df["velocity_deg_s_mean"] = velocity_roll.mean().reset_index(level=0, drop=True)
|
|
df["velocity_deg_s_std"] = velocity_roll.std().reset_index(level=0, drop=True).fillna(0.0)
|
|
df["dispersion_deg_mean"] = (
|
|
grouped["dispersion_deg"].rolling(FEATURE_WINDOW, min_periods=1)
|
|
.mean().reset_index(level=0, drop=True)
|
|
)
|
|
return df
|
|
|
|
|
|
class OnlineFeatureTracker:
|
|
"""Incrementally computes the same features as add_rolling_features, one raw sample at a time.
|
|
|
|
Mirrors the instantaneous velocity/accel/dispersion logic in
|
|
gaze_exercises.py's record_sample so a live filter sees the exact
|
|
features the model was trained on.
|
|
"""
|
|
|
|
def __init__(self, cm_per_px: float, viewing_distance_cm: float) -> None:
|
|
self.cm_per_px = cm_per_px
|
|
self.viewing_distance_cm = viewing_distance_cm
|
|
self.history: deque[tuple[float, float, float]] = deque(maxlen=FEATURE_WINDOW)
|
|
self.velocity_history: deque[float] = deque(maxlen=FEATURE_WINDOW)
|
|
self.dispersion_history: deque[float] = deque(maxlen=FEATURE_WINDOW)
|
|
self.prev_sample: tuple[float, float, float] | None = None
|
|
self.prev_velocity_deg_s: float | None = None
|
|
|
|
def reset(self) -> None:
|
|
self.history.clear()
|
|
self.velocity_history.clear()
|
|
self.dispersion_history.clear()
|
|
self.prev_sample = None
|
|
self.prev_velocity_deg_s = None
|
|
|
|
def update(self, timestamp: float, x: float, y: float, confidence: float) -> dict | None:
|
|
velocity_deg_s = accel_deg_s2 = None
|
|
if self.prev_sample is not None:
|
|
prev_t, prev_x, prev_y = self.prev_sample
|
|
dt = timestamp - prev_t
|
|
if 0 < dt <= MAX_CONTINUOUS_GAP_S:
|
|
velocity_px_s = math.dist((prev_x, prev_y), (x, y)) / dt
|
|
velocity_deg_s = px_to_deg(velocity_px_s, self.cm_per_px, self.viewing_distance_cm)
|
|
if self.prev_velocity_deg_s is not None:
|
|
accel_deg_s2 = (velocity_deg_s - self.prev_velocity_deg_s) / dt
|
|
self.prev_sample = (timestamp, x, y)
|
|
self.prev_velocity_deg_s = velocity_deg_s
|
|
|
|
self.history.append((x, y, timestamp))
|
|
dispersion_deg = None
|
|
if len(self.history) >= 2:
|
|
xs = [p[0] for p in self.history]
|
|
ys = [p[1] for p in self.history]
|
|
dispersion_px = (max(xs) - min(xs)) + (max(ys) - min(ys))
|
|
dispersion_deg = px_to_deg(dispersion_px, self.cm_per_px, self.viewing_distance_cm)
|
|
|
|
if velocity_deg_s is None or dispersion_deg is None:
|
|
return None
|
|
|
|
self.velocity_history.append(velocity_deg_s)
|
|
self.dispersion_history.append(dispersion_deg)
|
|
|
|
n = len(self.velocity_history)
|
|
velocity_deg_s_mean = sum(self.velocity_history) / n
|
|
if n >= 2:
|
|
variance = sum((v - velocity_deg_s_mean) ** 2 for v in self.velocity_history) / n
|
|
velocity_deg_s_std = math.sqrt(variance)
|
|
else:
|
|
velocity_deg_s_std = 0.0
|
|
dispersion_deg_mean = sum(self.dispersion_history) / len(self.dispersion_history)
|
|
|
|
return {
|
|
"velocity_deg_s": velocity_deg_s,
|
|
"accel_deg_s2": accel_deg_s2 if accel_deg_s2 is not None else 0.0,
|
|
"dispersion_deg": dispersion_deg,
|
|
"confidence": confidence,
|
|
"velocity_deg_s_mean": velocity_deg_s_mean,
|
|
"velocity_deg_s_std": velocity_deg_s_std,
|
|
"dispersion_deg_mean": dispersion_deg_mean,
|
|
}
|