| import cv2
|
| import numpy as np
|
| import time
|
| import os
|
| import random
|
| import pyautogui
|
|
|
| import warnings
|
| warnings.filterwarnings("ignore", category=DeprecationWarning)
|
|
|
| from preprocessing.preprocessing_pipeline import (
|
| create_face_mesh,
|
| estimate_camera_matrix,
|
| estimate_head_pose,
|
| compute_iris_radius,
|
| compute_ear,
|
| step1_normalize,
|
| step2_illumination,
|
| LEFT_EYE_INDICES,
|
| LEFT_EAR_INDICES,
|
| LEFT_IRIS_INDICES,
|
| RIGHT_EYE_INDICES,
|
| RIGHT_EAR_INDICES,
|
| RIGHT_IRIS_INDICES,
|
| )
|
|
|
| from inference_pipeline import InsightUXPipeline
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| POSE_NORM_SCALE = 30.0
|
|
|
| def normalize_pose(head_pose):
|
| return np.array([
|
| head_pose.pitch / POSE_NORM_SCALE,
|
| head_pose.yaw / POSE_NORM_SCALE,
|
| head_pose.roll / POSE_NORM_SCALE,
|
| ], dtype=np.float32)
|
|
|
|
|
|
|
|
|
|
|
| HEAD_PITCH_COMPENSATION = 0.0
|
|
|
| def compensate_pitch(raw_pitch, head_pitch_deg):
|
| return raw_pitch - np.radians(head_pitch_deg) * HEAD_PITCH_COMPENSATION
|
|
|
|
|
|
|
|
|
|
|
| HEAD_YAW_COMPENSATION = 0.0
|
|
|
| def compensate_yaw(raw_yaw, head_yaw_deg):
|
| return raw_yaw - np.radians(head_yaw_deg) * HEAD_YAW_COMPENSATION
|
|
|
|
|
|
|
|
|
|
|
|
|
| LIGHT_MIN_BRIGHTNESS = 60
|
| LIGHT_MAX_BRIGHTNESS = 200
|
| BLINK_EAR_THRESHOLD = 0.20
|
| MIN_SAMPLES_OK = 15
|
|
|
|
|
|
|
|
|
|
|
|
|
| def check_lighting(frame):
|
| gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
| brightness = np.mean(gray)
|
| if brightness < LIGHT_MIN_BRIGHTNESS:
|
| return False, "Too dark - increase lighting"
|
| elif brightness > LIGHT_MAX_BRIGHTNESS:
|
| return False, "Too bright - reduce lighting"
|
| return True, None
|
|
|
|
|
| def check_face_center(lms, frame_shape):
|
| H, W = frame_shape[:2]
|
| xs = [lm.x * W for lm in lms]
|
| ys = [lm.y * H for lm in lms]
|
| cx = np.mean(xs)
|
| cy = np.mean(ys)
|
| if cx < W * 0.35:
|
| return False, "Move face RIGHT in frame"
|
| elif cx > W * 0.65:
|
| return False, "Move face LEFT in frame"
|
| if cy < H * 0.35:
|
| return False, "Sit closer or lower your camera"
|
| elif cy > H * 0.65:
|
| return False, "Sit further or raise your camera"
|
| return True, None
|
|
|
|
|
| def get_full_feedback(frame, lms, frame_shape, yaw, roll):
|
| msgs = []
|
| ok = True
|
| H, W = frame_shape[:2]
|
|
|
| nose_y = lms[1].y * H
|
| forehead_y = lms[10].y * H
|
| chin_y = lms[152].y * H
|
| midpoint_y = (forehead_y + chin_y) / 2.0
|
| face_h = chin_y - forehead_y
|
|
|
| if face_h > 1:
|
| pitch_ratio = (nose_y - midpoint_y) / face_h
|
| if pitch_ratio < -0.10:
|
| msgs.append("Lift your head UP")
|
| ok = False
|
| elif pitch_ratio > 0.15:
|
| msgs.append("Tilt your head DOWN slightly")
|
| ok = False
|
|
|
| if yaw < -15:
|
| msgs.append("Turn face slightly RIGHT")
|
| ok = False
|
| elif yaw > 15:
|
| msgs.append("Turn face slightly LEFT")
|
| ok = False
|
|
|
| if roll < -10:
|
| msgs.append("Tilt head slightly RIGHT")
|
| ok = False
|
| elif roll > 10:
|
| msgs.append("Tilt head slightly LEFT")
|
| ok = False
|
|
|
| light_ok, light_msg = check_lighting(frame)
|
| if not light_ok:
|
| msgs.append(light_msg)
|
| ok = False
|
|
|
| center_ok, center_msg = check_face_center(lms, frame_shape)
|
| if not center_ok:
|
| msgs.append(center_msg)
|
| ok = False
|
|
|
| if ok:
|
| msgs.append("Perfect! Hold still...")
|
|
|
| return ok, msgs
|
|
|
|
|
|
|
|
|
|
|
|
|
| def face_orientation_gate(face_mesh, cap, cam_matrix_ref):
|
| print("\nChecking face setup before calibration...")
|
| print("Position your face straight, centred, at normal laptop distance.")
|
|
|
| HOLD_SECONDS = 2.0
|
| good_since = None
|
| cam_matrix_loc = cam_matrix_ref[0]
|
|
|
| cv2.namedWindow("Face Check", cv2.WINDOW_NORMAL)
|
| cv2.resizeWindow("Face Check", 640, 420)
|
|
|
| while True:
|
| ret, frame = cap.read()
|
| if not ret:
|
| continue
|
|
|
| if cam_matrix_loc is None:
|
| cam_matrix_loc = estimate_camera_matrix(frame.shape)
|
| cam_matrix_ref[0] = cam_matrix_loc
|
|
|
| rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
|
| results = face_mesh.process(rgb)
|
| display = frame.copy()
|
| H, W = display.shape[:2]
|
|
|
| if not results.multi_face_landmarks:
|
| good_since = None
|
| cv2.putText(display, "No face detected - look at camera",
|
| (20, 40), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2)
|
| else:
|
| lms = results.multi_face_landmarks[0].landmark
|
| head_pose = estimate_head_pose(lms, frame.shape, cam_matrix_loc)
|
|
|
| if head_pose is None:
|
| good_since = None
|
| cv2.putText(display, "Pose failed - move slightly",
|
| (20, 40), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2)
|
| else:
|
| is_good, msgs = get_full_feedback(
|
| frame, lms, frame.shape, head_pose.yaw, head_pose.roll
|
| )
|
|
|
| if is_good:
|
| if good_since is None:
|
| good_since = time.time()
|
| elapsed = time.time() - good_since
|
| remaining = max(0, HOLD_SECONDS - elapsed)
|
|
|
| bar_w = int((elapsed / HOLD_SECONDS) * (W - 40))
|
| bar_w = min(bar_w, W - 40)
|
| cv2.rectangle(display, (20, H-50), (W-20, H-25), (40, 40, 40), -1)
|
| cv2.rectangle(display, (20, H-50), (20+bar_w, H-25), (0, 220, 0), -1)
|
| cv2.putText(display, f"Hold still... {remaining:.1f}s",
|
| (20, 40), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2)
|
|
|
| if elapsed >= HOLD_SECONDS:
|
| cv2.putText(display, "Starting calibration!",
|
| (20, 80), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 255, 0), 2)
|
| cv2.imshow("Face Check", display)
|
| cv2.waitKey(800)
|
| cv2.destroyWindow("Face Check")
|
| return
|
| else:
|
| good_since = None
|
| for i, msg in enumerate(msgs):
|
| cv2.putText(display, msg, (20, 40 + i * 35),
|
| cv2.FONT_HERSHEY_SIMPLEX, 0.75, (0, 165, 255), 2)
|
|
|
| cv2.putText(display,
|
| f"Yaw:{head_pose.yaw:+.1f} Roll:{head_pose.roll:+.1f}",
|
| (20, H - 65), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (160, 160, 160), 1)
|
|
|
| cv2.imshow("Face Check", display)
|
| if cv2.waitKey(1) & 0xFF == 27:
|
| cv2.destroyWindow("Face Check")
|
| return
|
|
|
|
|
|
|
|
|
|
|
|
|
| def screen_to_gaze_angles(sx, sy, screen_w, screen_h):
|
| nx = (sx - screen_w / 2) / (screen_w / 2)
|
| ny = (sy - screen_h / 2) / (screen_h / 2)
|
| pitch = float(np.arctan(ny * 0.4))
|
| yaw = float(np.arctan(nx * 0.6))
|
| return pitch, yaw
|
|
|
|
|
| def finetune_on_calibration_v4(
|
| onnx_path,
|
| left_patches, right_patches, head_poses, screen_points,
|
| screen_w, screen_h,
|
| ckpt_path="checkpoints/best_model_v4.pt",
|
| out_onnx_path=None,
|
| steps=300, lr=5e-4,
|
| ):
|
| try:
|
| import torch
|
| import sys
|
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
| from models.model_v4 import GazeCNNv4
|
| except ImportError as e:
|
| print(f"[Fine-tune] Skipped: {e}")
|
| return False
|
|
|
| if not os.path.exists(ckpt_path):
|
| print(f"[Fine-tune] Skipped: checkpoint not found at {ckpt_path}")
|
| return False
|
|
|
| if out_onnx_path is None:
|
| out_onnx_path = onnx_path
|
|
|
| device = torch.device("cpu")
|
| model = GazeCNNv4().to(device)
|
| ckpt = torch.load(ckpt_path, map_location=device)
|
| model.load_state_dict(ckpt["model_state_dict"])
|
|
|
| for p in model.parameters():
|
| p.requires_grad = False
|
| for p in model.fc_a.parameters(): p.requires_grad = True
|
| for p in model.stream_b.parameters(): p.requires_grad = True
|
| for p in model.fusion.parameters(): p.requires_grad = True
|
|
|
| trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
|
| print(f"[Fine-tune] Trainable: {trainable:,} parameters")
|
|
|
| optimizer = torch.optim.Adam(
|
| filter(lambda p: p.requires_grad, model.parameters()),
|
| lr=lr, weight_decay=1e-4
|
| )
|
|
|
| def angular_loss(pred, target):
|
| def to_vec(a):
|
| p, y = a[:, 0], a[:, 1]
|
| v = torch.stack([torch.cos(p)*torch.sin(y),
|
| torch.sin(p),
|
| torch.cos(p)*torch.cos(y)], dim=1)
|
| return v / (v.norm(dim=1, keepdim=True) + 1e-8)
|
| cos_sim = (to_vec(pred) * to_vec(target)).sum(dim=1)
|
| return (1 - cos_sim).mean()
|
|
|
| patches_l, poses_t, gazes = [], [], []
|
| for l, r, pose, sp in zip(left_patches, right_patches, head_poses, screen_points):
|
| def norm(x): return (x.astype(np.float32) / 255.0 - 0.5) / 0.5
|
| patches_l.append(np.stack([norm(l), norm(r)], axis=0))
|
| poses_t.append(pose)
|
| gazes.append(screen_to_gaze_angles(sp[0], sp[1], screen_w, screen_h))
|
|
|
| patches_t = torch.tensor(np.array(patches_l), dtype=torch.float32)
|
| poses_t = torch.tensor(np.array(poses_t), dtype=torch.float32)
|
| gazes_t = torch.tensor(np.array(gazes), dtype=torch.float32)
|
|
|
| model.eval()
|
| with torch.no_grad():
|
| feat_a_raw = model.pool(model.backbone(patches_t)).flatten(1)
|
|
|
| model.fc_a.train()
|
| model.stream_b.train()
|
| model.fusion.train()
|
|
|
| print(f"[Fine-tune] {steps} steps on {len(patches_l)} samples (backbone cached)...")
|
| for step in range(steps):
|
| optimizer.zero_grad()
|
| feat_a = model.fc_a(feat_a_raw)
|
| feat_b = model.stream_b(poses_t)
|
| fused = torch.cat([feat_a, feat_b], dim=1)
|
| pred = model.fusion(fused)
|
| loss = angular_loss(pred, gazes_t)
|
| loss.backward()
|
| optimizer.step()
|
| if (step + 1) % 50 == 0:
|
| print(f" Step {step+1}/{steps} | Loss: {loss.item():.5f}")
|
|
|
| model.eval()
|
| dp = torch.zeros(1, 2, 36, 60)
|
| dpose = torch.zeros(1, 3)
|
| torch.onnx.export(
|
| model, (dp, dpose), out_onnx_path,
|
| input_names = ["eye_patch_binocular", "head_pose"],
|
| output_names = ["gaze"],
|
| dynamic_axes = {"eye_patch_binocular": {0: "batch"},
|
| "head_pose": {0: "batch"}, "gaze": {0: "batch"}},
|
| opset_version=14,
|
| dynamo=False,
|
| )
|
| print(f"[Fine-tune] Adapted ONNX saved: {out_onnx_path}")
|
| return True
|
|
|
|
|
|
|
|
|
|
|
|
|
| def filter_outlier_points(gaze_vectors, screen_points, lefts, rights, poses,
|
| head_pitches_deg, sigma=2.0):
|
| gv = np.array(gaze_vectors)
|
| med_p = np.median(gv[:, 0])
|
| med_y = np.median(gv[:, 1])
|
| std_p = max(np.std(gv[:, 0]), 1e-6)
|
| std_y = max(np.std(gv[:, 1]), 1e-6)
|
|
|
| keep = (np.abs(gv[:, 0] - med_p) / std_p <= sigma)
|
| keep &= (np.abs(gv[:, 1] - med_y) / std_y <= sigma)
|
|
|
| n_removed = int((~keep).sum())
|
| if n_removed:
|
| print(f"[Outlier filter] Removed {n_removed} point(s) (>{sigma}σ)")
|
|
|
| idx = np.where(keep)[0].tolist()
|
| return (
|
| [gaze_vectors[i] for i in idx],
|
| [screen_points[i] for i in idx],
|
| [lefts[i] for i in idx],
|
| [rights[i] for i in idx],
|
| [poses[i] for i in idx],
|
| [head_pitches_deg[i] for i in idx],
|
| )
|
|
|
|
|
|
|
|
|
|
|
|
|
| ONNX_PATH = "models/gaze_cnn_v4.onnx"
|
| CKPT_PATH = "checkpoints/best_model_v4.pt"
|
|
|
| SCREEN_W, SCREEN_H = pyautogui.size()
|
|
|
| points = [
|
| (0.02, 0.02), (0.35, 0.02), (0.65, 0.02), (0.98, 0.02),
|
| (0.02, 0.35), (0.35, 0.35), (0.65, 0.35), (0.98, 0.35),
|
| (0.02, 0.65), (0.35, 0.65), (0.65, 0.65), (0.98, 0.65),
|
| (0.02, 0.98), (0.35, 0.98), (0.65, 0.98), (0.98, 0.98),
|
| ]
|
|
|
| def get_duration(py):
|
|
|
|
|
|
|
|
|
|
|
| return 5 if (py <= 0.4 or py >= 0.6) else 3
|
|
|
| random.Random(7).shuffle(points)
|
|
|
| FINETUNE_SAMPLES_PER_POINT = 5
|
|
|
|
|
|
|
|
|
|
|
|
|
| pipeline = InsightUXPipeline(ONNX_PATH)
|
| face_mesh = create_face_mesh(static_image_mode=False)
|
| cap = cv2.VideoCapture(0)
|
|
|
| cam_matrix = None
|
|
|
| gaze_vectors = []
|
| screen_points = []
|
| head_pitches_deg = []
|
| rep_lefts = []
|
| rep_rights = []
|
| rep_poses = []
|
| rep_hp_deg = []
|
|
|
| ft_lefts = []
|
| ft_rights = []
|
| ft_poses = []
|
| ft_sp = []
|
|
|
|
|
|
|
| session_brightness = []
|
| session_blink_skips = 0
|
| session_light_skips = 0
|
| points_with_low_yield = []
|
|
|
| cam_matrix_ref = [None]
|
| face_orientation_gate(face_mesh, cap, cam_matrix_ref)
|
| if cam_matrix_ref[0] is not None:
|
| cam_matrix = cam_matrix_ref[0]
|
|
|
| cv2.namedWindow("Calibration", cv2.WINDOW_NORMAL)
|
| cv2.setWindowProperty("Calibration", cv2.WND_PROP_FULLSCREEN, cv2.WINDOW_FULLSCREEN)
|
|
|
| total = len(points)
|
| print(f"Calibration started - {total} points")
|
|
|
| for idx, (px, py) in enumerate(points):
|
| sx, sy = int(px * SCREEN_W), int(py * SCREEN_H)
|
| duration = get_duration(py)
|
|
|
| settle_s = 1.5 if idx == 0 else 0.4
|
| settle_start = time.time()
|
| while time.time() - settle_start < settle_s:
|
| ret, frame = cap.read()
|
| if not ret:
|
| continue
|
| screen = np.zeros((SCREEN_H, SCREEN_W, 3), dtype=np.uint8)
|
| cv2.circle(screen, (sx, sy), 18, (80, 80, 80), -1)
|
| msg = "Get ready..." if idx == 0 else "Settling..."
|
| cv2.putText(screen, msg, (50, 50),
|
| cv2.FONT_HERSHEY_SIMPLEX, 1, (180, 180, 180), 2)
|
| cv2.imshow("Calibration", screen)
|
| cv2.waitKey(1)
|
|
|
| samples = []
|
| l_list = []
|
| r_list = []
|
| p_list = []
|
| hp_list = []
|
|
|
| point_blink_skips = 0
|
| point_light_skips = 0
|
|
|
| start = time.time()
|
|
|
| while time.time() - start < duration:
|
| ret, frame = cap.read()
|
| if not ret:
|
| continue
|
|
|
| if cam_matrix is None:
|
| cam_matrix = estimate_camera_matrix(frame.shape)
|
|
|
|
|
|
|
| light_ok, light_msg = check_lighting(frame)
|
| gray_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
| session_brightness.append(float(np.mean(gray_frame)))
|
| if not light_ok:
|
| point_light_skips += 1
|
| screen = np.zeros((SCREEN_H, SCREEN_W, 3), dtype=np.uint8)
|
| cv2.circle(screen, (sx, sy), 18, (0, 140, 255), 2)
|
| cv2.putText(screen, f"Point {idx+1}/{total} - {light_msg}",
|
| (50, 50), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 140, 255), 2)
|
| cv2.imshow("Calibration", screen)
|
| cv2.waitKey(1)
|
| continue
|
|
|
| rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
|
| results = face_mesh.process(rgb)
|
|
|
| if not results.multi_face_landmarks:
|
| continue
|
|
|
| lms = results.multi_face_landmarks[0].landmark
|
| head_pose = estimate_head_pose(lms, frame.shape, cam_matrix)
|
| if head_pose is None:
|
| continue
|
|
|
|
|
|
|
|
|
| ear_l = compute_ear(lms, LEFT_EAR_INDICES, frame.shape)
|
| ear_r = compute_ear(lms, RIGHT_EAR_INDICES, frame.shape)
|
| avg_ear = (ear_l + ear_r) / 2.0
|
| if avg_ear < BLINK_EAR_THRESHOLD:
|
| point_blink_skips += 1
|
| screen = np.zeros((SCREEN_H, SCREEN_W, 3), dtype=np.uint8)
|
| cv2.circle(screen, (sx, sy), 18, (180, 180, 0), 2)
|
| cv2.putText(screen, f"Point {idx+1}/{total} - Blink detected, skipping frame",
|
| (50, 50), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (180, 180, 0), 2)
|
| cv2.imshow("Calibration", screen)
|
| cv2.waitKey(1)
|
| continue
|
|
|
|
|
| pose_vec = normalize_pose(head_pose)
|
|
|
| def process_eye(eye_idx, ear_idx, iris_idx):
|
| s1 = step1_normalize(frame, lms, head_pose, eye_idx, ear_idx, iris_idx)
|
| if not s1.is_open:
|
| return None
|
| ir = compute_iris_radius(lms, iris_idx, frame.shape)
|
| s2 = step2_illumination(s1, ir)
|
| return s2.blended if s2.is_usable else None
|
|
|
| l = process_eye(LEFT_EYE_INDICES, LEFT_EAR_INDICES, LEFT_IRIS_INDICES)
|
| r = process_eye(RIGHT_EYE_INDICES, RIGHT_EAR_INDICES, RIGHT_IRIS_INDICES)
|
|
|
| if l is None and r is None:
|
| continue
|
| if l is None: l = r
|
| if r is None: r = l
|
|
|
| _, _, raw_pitch, raw_yaw = pipeline.predict_gaze_vector(l, pose_vec, r)
|
|
|
| pitch = compensate_pitch(raw_pitch, head_pose.pitch)
|
| yaw = compensate_yaw(raw_yaw, head_pose.yaw)
|
|
|
| samples.append([pitch, yaw])
|
| l_list.append(l)
|
| r_list.append(r)
|
| p_list.append(pose_vec)
|
| hp_list.append(head_pose.pitch)
|
|
|
| elapsed = time.time() - start
|
| screen = np.zeros((SCREEN_H, SCREEN_W, 3), dtype=np.uint8)
|
| angle = int(360 * elapsed / duration)
|
| cv2.ellipse(screen, (sx, sy), (28, 28), -90, 0, angle, (0, 180, 0), 3)
|
| cv2.circle(screen, (sx, sy), 18, (0, 255, 0), -1)
|
| cv2.putText(screen, f"Point {idx+1}/{total} - Look at the dot",
|
| (50, 50), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
|
| if point_blink_skips or point_light_skips:
|
| cv2.putText(screen, f"skipped: {point_blink_skips} blink, {point_light_skips} lighting",
|
| (50, 85), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (140, 140, 140), 1)
|
| cv2.imshow("Calibration", screen)
|
|
|
| if cv2.waitKey(1) & 0xFF == 27:
|
| break
|
|
|
| session_blink_skips += point_blink_skips
|
| session_light_skips += point_light_skips
|
|
|
| if not samples:
|
| print(f"Point {idx+1}: no samples, skipping.")
|
| points_with_low_yield.append(idx + 1)
|
| continue
|
|
|
| if len(samples) < MIN_SAMPLES_OK:
|
| print(f" ⚠ Point {idx+1}: only {len(samples)} valid samples "
|
| f"({point_blink_skips} blink-skipped, {point_light_skips} light-skipped) — may be unreliable")
|
| points_with_low_yield.append(idx + 1)
|
|
|
| avg = np.median(samples, axis=0)
|
| avg_hp_deg = float(np.median(hp_list))
|
|
|
| gaze_vectors.append(list(avg))
|
| screen_points.append([sx, sy])
|
| head_pitches_deg.append(avg_hp_deg)
|
|
|
| mid = len(l_list) // 2
|
| rep_lefts.append(l_list[mid])
|
| rep_rights.append(r_list[mid])
|
| rep_poses.append(p_list[mid])
|
| rep_hp_deg.append(hp_list[mid])
|
|
|
| n_avail = len(l_list)
|
| n_take = min(FINETUNE_SAMPLES_PER_POINT, n_avail)
|
| indices = np.linspace(0, n_avail - 1, n_take, dtype=int)
|
| for i in indices:
|
| ft_lefts.append(l_list[i])
|
| ft_rights.append(r_list[i])
|
| ft_poses.append(p_list[i])
|
| ft_sp.append([sx, sy])
|
|
|
| print(f"Point {idx+1:2d}/{total} | pitch={avg[0]:.4f} yaw={avg[1]:.4f} "
|
| f"| head_pitch={avg_hp_deg:.1f}deg | {len(indices)} frames kept "
|
| f"| skipped: {point_blink_skips} blink, {point_light_skips} lighting")
|
|
|
| cap.release()
|
| cv2.destroyAllWindows()
|
|
|
| if len(gaze_vectors) < 4:
|
| print(f"ERROR: only {len(gaze_vectors)} points collected, need at least 4.")
|
| exit(1)
|
|
|
|
|
|
|
|
|
| (gaze_vectors, screen_points,
|
| rep_lefts, rep_rights, rep_poses,
|
| rep_hp_deg) = filter_outlier_points(
|
| gaze_vectors, screen_points,
|
| rep_lefts, rep_rights, rep_poses,
|
| head_pitches_deg, sigma=3.0
|
| )
|
|
|
| n_pts = len(screen_points)
|
| if n_pts < 4:
|
| print("ERROR: too many outliers removed, need at least 4 clean points.")
|
| exit(1)
|
|
|
| ft_lefts_clean = []
|
| ft_rights_clean = []
|
| ft_poses_clean = []
|
| ft_sp_clean = []
|
| for i in range(n_pts):
|
| for _ in range(FINETUNE_SAMPLES_PER_POINT):
|
| ft_lefts_clean.append(rep_lefts[i])
|
| ft_rights_clean.append(rep_rights[i])
|
| ft_poses_clean.append(rep_poses[i])
|
| ft_sp_clean.append(screen_points[i])
|
|
|
| print(f"\n--- Fine-tuning CNN on your eyes ({len(ft_lefts_clean)} frames) ---")
|
| finetuned = finetune_on_calibration_v4(
|
| onnx_path = ONNX_PATH,
|
| left_patches = ft_lefts_clean,
|
| right_patches = ft_rights_clean,
|
| head_poses = ft_poses_clean,
|
| screen_points = ft_sp_clean,
|
| screen_w = SCREEN_W,
|
| screen_h = SCREEN_H,
|
| ckpt_path = CKPT_PATH,
|
| out_onnx_path = ONNX_PATH,
|
| steps = 80,
|
| lr = 1e-4,
|
| )
|
|
|
| print("\n--- Fitting RBF calibration ---")
|
| pipeline = InsightUXPipeline(ONNX_PATH)
|
|
|
| adapted_gaze_vectors = []
|
| for i in range(n_pts):
|
| l = rep_lefts[i]
|
| r = rep_rights[i]
|
| pose = rep_poses[i]
|
| hp = rep_hp_deg[i]
|
| hy = float(pose[1]) * POSE_NORM_SCALE
|
|
|
| _, _, raw_pitch, raw_yaw = pipeline.predict_gaze_vector(l, pose, r)
|
|
|
| pitch = compensate_pitch(raw_pitch, hp)
|
| yaw = compensate_yaw(raw_yaw, hy)
|
| adapted_gaze_vectors.append([pitch, yaw])
|
|
|
| assert len(adapted_gaze_vectors) == len(screen_points), \
|
| f"Length mismatch: {len(adapted_gaze_vectors)} gaze vs {len(screen_points)} screen"
|
|
|
| pipeline.calibration.calibrate(
|
| np.array(adapted_gaze_vectors),
|
| np.array(screen_points)
|
| )
|
| pipeline.calibration.save("calibration.pkl")
|
|
|
| import pickle
|
| baseline_pitch = float(np.median(rep_hp_deg)) if len(rep_hp_deg) else 0.0
|
| baseline_yaw = float(np.median([float(p[1]) * POSE_NORM_SCALE for p in rep_poses])) if len(rep_poses) else 0.0
|
| baseline_roll = float(np.median([float(p[2]) * POSE_NORM_SCALE for p in rep_poses])) if len(rep_poses) else 0.0
|
| with open("baseline_pose.pkl", "wb") as f:
|
| pickle.dump({"pitch": baseline_pitch, "yaw": baseline_yaw, "roll": baseline_roll}, f)
|
| print(f"Saved baseline_pose.pkl: pitch={baseline_pitch:+.2f}, yaw={baseline_yaw:+.2f}, roll={baseline_roll:+.2f}")
|
|
|
| print(f"\nCalibration complete! {n_pts}/{total} points used.")
|
| if finetuned:
|
| print("CNN fine-tuned on your eyes + RBF calibration fitted.")
|
| else:
|
| print("RBF calibration fitted (CNN fine-tuning skipped).")
|
|
|
|
|
|
|
|
|
|
|
|
|
| print("\n================ CALIBRATION QUALITY SUMMARY ================")
|
| if session_brightness:
|
| avg_bright = float(np.mean(session_brightness))
|
| pct_dark = 100 * sum(1 for b in session_brightness if b < LIGHT_MIN_BRIGHTNESS) / len(session_brightness)
|
| pct_bright = 100 * sum(1 for b in session_brightness if b > LIGHT_MAX_BRIGHTNESS) / len(session_brightness)
|
| print(f"Average brightness: {avg_bright:.0f} (comfortable range: {LIGHT_MIN_BRIGHTNESS}-{LIGHT_MAX_BRIGHTNESS})")
|
| if pct_dark > 10:
|
| print(f"⚠ Lighting was too DARK for {pct_dark:.0f}% of frames. "
|
| f"Add a light source facing your face, or face a window, before recalibrating.")
|
| if pct_bright > 10:
|
| print(f"⚠ Lighting was too BRIGHT for {pct_bright:.0f}% of frames "
|
| f"(backlight or a light directly behind you?). Try facing away from strong light sources.")
|
| if pct_dark <= 10 and pct_bright <= 10:
|
| print("Lighting was consistently good throughout.")
|
|
|
| if session_blink_skips > 0:
|
| print(f"Blinking accounted for {session_blink_skips} skipped frames across the session "
|
| f"— normal, this is expected and was handled automatically.")
|
|
|
| if points_with_low_yield:
|
| print(f"⚠ These points had low sample counts and may be less accurate: "
|
| f"{', '.join(str(p) for p in points_with_low_yield)}. "
|
| f"If tracking feels off in that part of the screen, consider recalibrating.")
|
| else:
|
| print("All points collected a healthy number of samples.")
|
| print("===============================================================")
|
|
|
|
|
| print("\n--- Pitch by screen row (should increase top -> bottom) ---")
|
| gv_arr = np.array(adapted_gaze_vectors)
|
| sp_arr = np.array(screen_points)
|
| thresholds = [(0, SCREEN_H*0.25, "Top (y<25%) "),
|
| (SCREEN_H*0.25, SCREEN_H*0.5, "Mid-hi (25-50%) "),
|
| (SCREEN_H*0.5, SCREEN_H*0.75, "Mid-lo (50-75%) "),
|
| (SCREEN_H*0.75, SCREEN_H+1, "Bottom (y>75%) ")]
|
| for lo, hi, label in thresholds:
|
| mask = (sp_arr[:, 1] >= lo) & (sp_arr[:, 1] < hi)
|
| if mask.any():
|
| print(f" {label}: avg pitch = {gv_arr[mask, 0].mean():.4f}")
|
|
|
| print("\n--- Yaw by screen column (should increase left -> right) ---")
|
| col_thresholds = [(0, SCREEN_W*0.25, "Left (x<25%) "),
|
| (SCREEN_W*0.25, SCREEN_W*0.5, "Mid-lf (25-50%) "),
|
| (SCREEN_W*0.5, SCREEN_W*0.75, "Mid-rt (50-75%) "),
|
| (SCREEN_W*0.75, SCREEN_W+1, "Right (x>75%) ")]
|
| for lo, hi, label in col_thresholds:
|
| mask = (sp_arr[:, 0] >= lo) & (sp_arr[:, 0] < hi)
|
| if mask.any():
|
| print(f" {label}: avg yaw = {gv_arr[mask, 1].mean():.4f}") |
