| import torch |
| import numpy as np |
| import h5py |
| import os |
| import sys |
| from pathlib import Path |
| from tqdm import tqdm |
|
|
| |
| sys.path.append(str(Path(__file__).parent.parent)) |
|
|
| from src.models.student import LIPEV2Student, LIPEV2StudentGold, LIPEV2StudentBaseline |
|
|
| def evaluate_on_gaze360(model_path, h5_path, invert_yaw=False, invert_pitch=False): |
| device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') |
| print(f"Evaluating model: {model_path}") |
| print(f"On dataset: {h5_path}") |
| print(f"Correction: Invert Yaw={invert_yaw}, Invert Pitch={invert_pitch}") |
| print(f"Device: {device}") |
|
|
| |
| is_gold = 'gold' in model_path.lower() or 'id_5' in model_path.lower() or 'id_2' in model_path.lower() or 'id_6' in model_path.lower() or 'id_8' in model_path.lower() |
| is_baseline = 'baseline' in model_path.lower() |
| |
| if is_gold: |
| print("Architecture: V5-GOLD (DualPool / ID 5,6,8)") |
| model = LIPEV2StudentGold().to(device) |
| elif is_baseline: |
| print("Architecture: Baseline (Addition)") |
| model = LIPEV2StudentBaseline().to(device) |
| else: |
| print("Architecture: Standard (Concatenation / ID 1,3,4,7)") |
| model = LIPEV2Student().to(device) |
|
|
| state_dict = torch.load(model_path, map_location=device) |
| |
| |
| if 'n_averaged' in state_dict: |
| |
| new_state_dict = {} |
| for k, v in state_dict.items(): |
| if k.startswith('module.'): |
| new_state_dict[k[7:]] = v |
| else: |
| new_state_dict[k] = v |
| state_dict = new_state_dict |
|
|
| model.load_state_dict(state_dict, strict=False) |
| model.eval() |
|
|
| results = { |
| 'all': {'error': 0.0, 'count': 0}, |
| 'frontal_45': {'error': 0.0, 'count': 0} |
| } |
|
|
| with h5py.File(h5_path, 'r') as f: |
| left_patches = f['left_patches'][:] |
| right_patches = f['right_patches'][:] |
| landmarks = f['landmarks'][:] |
| gaze_gt = f['gaze'][:] |
|
|
| num_samples = left_patches.shape[0] |
| |
| with torch.no_grad(): |
| for i in tqdm(range(num_samples), desc="Testing"): |
| |
| lp = torch.from_numpy(left_patches[i]).float().unsqueeze(0).to(device) / 255.0 |
| rp = torch.from_numpy(right_patches[i]).float().unsqueeze(0).to(device) / 255.0 |
| lm = torch.from_numpy(landmarks[i]).float().view(1, -1).to(device) |
| gt = torch.from_numpy(gaze_gt[i]).float().to(device) |
|
|
| |
| if is_gold: |
| out = model(lp, lm) |
| else: |
| out = model(lp, lm, state='A') |
| |
| p_l, y_l = out[0], out[1] |
| |
| if is_gold: |
| out_r = model(rp, lm) |
| else: |
| out_r = model(rp, lm, state='A') |
| p_r, y_r = out_r[0], out_r[1] |
| |
| |
| def logits_to_deg(p_logits, y_logits): |
| idx = torch.arange(90).float().to(device) |
| p_prob = torch.softmax(p_logits, dim=1) |
| y_prob = torch.softmax(y_logits, dim=1) |
| p_deg = (torch.sum(p_prob * idx, dim=1) * 2 - 90) |
| y_deg = (torch.sum(y_prob * idx, dim=1) * 2 - 90) |
| return p_deg, y_deg |
|
|
| p_deg_l, y_deg_l = logits_to_deg(p_l, y_l) |
| p_deg_r, y_deg_r = logits_to_deg(p_r, y_r) |
| |
| p_final = (p_deg_l + p_deg_r) / 2 |
| y_final = (y_deg_l + y_deg_r) / 2 |
| |
| |
| if invert_pitch: p_final = -p_final |
| if invert_yaw: y_final = -y_final |
| |
| if args.swap_axes: |
| p_final, y_final = y_final, p_final |
| |
| |
| gt_deg = gt * (180.0 / np.pi) |
| pitch_gt = gt_deg[0] |
| yaw_gt = gt_deg[1] |
| |
| |
| def angles_to_unit_vector(pitch_deg, yaw_deg): |
| p = np.radians(pitch_deg) |
| y = np.radians(yaw_deg) |
| |
| |
| |
| vx = np.cos(p) * np.sin(y) |
| vy = np.sin(p) |
| vz = np.cos(p) * np.cos(y) |
| return np.array([vx, vy, vz]) |
|
|
| v_pred = angles_to_unit_vector(p_final.item(), y_final.item()) |
| v_gt = angles_to_unit_vector(pitch_gt.item(), yaw_gt.item()) |
| |
| |
| cos_sim = np.clip(np.dot(v_pred, v_gt), -1.0, 1.0) |
| angular_error = np.degrees(np.arccos(cos_sim)) |
| |
| |
| results['all']['error'] += angular_error |
| results['all']['count'] += 1 |
| |
| |
| if abs(yaw_gt.item()) <= 45.0: |
| results['frontal_45']['error'] += angular_error |
| results['frontal_45']['count'] += 1 |
|
|
| print(f"\n" + "="*45) |
| print(f"{'SUBSET':<20} | {'SAMPLES':<10} | {'Ang Error (deg)':<10}") |
| print(f"-"*45) |
| |
| for key, data in results.items(): |
| if data['count'] > 0: |
| mae = data['error'] / data['count'] |
| name = "All Cases" if key == 'all' else "Frontal +/- 45" |
| print(f"{name:<20} | {data['count']:<10} | {mae:.4f}") |
| |
| print(f"="*45) |
|
|
| if __name__ == "__main__": |
| import argparse |
| parser = argparse.ArgumentParser() |
| parser.add_argument('--model', type=str, default='checkpoints/baseline_v16/best_student_p11.pt') |
| parser.add_argument('--h5', type=str, default='data/processed/gaze360_robust_v16_test_B.h5') |
| parser.add_argument('--invert_yaw', action='store_true', default=False) |
| parser.add_argument('--invert_pitch', action='store_true', default=False) |
| parser.add_argument('--swap_axes', action='store_true', default=False) |
| args = parser.parse_args() |
| |
| evaluate_on_gaze360( |
| model_path=args.model, |
| h5_path=args.h5, |
| invert_yaw=args.invert_yaw, |
| invert_pitch=args.invert_pitch |
| ) |
|
|