diff --git a/app/routes/session.py b/app/routes/session.py index 0840f8b..541ce1e 100644 --- a/app/routes/session.py +++ b/app/routes/session.py @@ -1,4 +1,4 @@ -# Necesary imports +# Necessary imports import os import re import time @@ -28,6 +28,63 @@ ALLOWED_EXTENSIONS = {"txt", "webm"} COLLECTION_NAME = "session" +def summarize_numeric_values(values, screen_width=None, screen_height=None): + numeric_values = [ + value for value in values + if isinstance(value, (int, float, np.integer, np.floating)) + and np.isfinite(value) + ] + array = np.asarray(numeric_values, dtype=float) + return { + "min": float(np.min(array)) if array.size else None, + "max": float(np.max(array)) if array.size else None, + "mean": float(np.mean(array)) if array.size else None, + "std": float(np.std(array)) if array.size else None, + "constant": bool(array.size and np.all(array == array[0])), + "invalid": len(values) - len(numeric_values), + "strings": sum(isinstance(value, str) for value in values), + "outside_screen_range": int( + np.sum( + (array < 0) | + (array > max(screen_width or 0, screen_height or 0)) + ) + ) if array.size else 0, + } + +def summarize_batch_input(iris_data, screen_width=None, screen_height=None): + fields = ["left_iris_x", "right_iris_x", "left_iris_y", "right_iris_y"] + timestamps = [ + sample.get("timestamp") for sample in iris_data + if isinstance(sample, dict) + and isinstance(sample.get("timestamp"), (int, float)) + and np.isfinite(sample.get("timestamp")) + ] + intervals = np.diff(timestamps).tolist() if len(timestamps) > 1 else [] + return { + "sample_count": len(iris_data), + "invalid_samples": sum( + not isinstance(sample, dict) or any( + not isinstance(sample.get(field), (int, float)) + or not np.isfinite(sample.get(field)) + for field in fields + ) + for sample in iris_data + ), + "timestamp_range": { + "min": min(timestamps) if timestamps else None, + "max": max(timestamps) if timestamps else None, + }, + "timestamp_intervals": summarize_numeric_values(intervals), + "fields": { + field: summarize_numeric_values( + [sample.get(field) if isinstance(sample, dict) else None for sample in iris_data], + screen_width, + screen_height, + ) + for field in fields + }, + } + def sanitize_filename(name): """Sanitize a user-provided filename to prevent path traversal attacks. @@ -77,9 +134,13 @@ def convert_nan_to_none(obj): def calib_results(): from_ruxailab = json.loads(request.form['from_ruxailab']) - user_id = json.loads(request.form['user_id']) - study_id = json.loads(request.form['study_id']) - file_name = sanitize_filename(json.loads(request.form['file_name'])) + user_id = json.loads( + request.form.get('user_id', 'null') + ) + study_id = json.loads( + request.form.get('study_id', 'null') + ) + fallback_file_name = sanitize_filename(json.loads(request.form['file_name'])) fixed_points = json.loads(request.form['fixed_circle_iris_points']) calib_points = json.loads(request.form['calib_circle_iris_points']) screen_height = json.loads(request.form['screen_height']) @@ -87,12 +148,28 @@ def calib_results(): model = json.loads(request.form.get('model', '"Linear Regression"')) k = json.loads(request.form['k']) - # Generate csv dataset of calibration points + timestamp = int(time.time() * 1000) + + if ( + user_id + and isinstance(user_id, str) + and user_id.strip() + and study_id + and isinstance(study_id, str) + and study_id.strip() + ): + file_name = sanitize_filename( + f"{user_id}_{study_id}_{timestamp}" + ) + else: + file_name = fallback_file_name + + # Generate a CSV dataset from the calibration points os.makedirs( f"{Path().absolute()}/app/services/calib_validation/csv/data/", exist_ok=True ) - # Generate csv of calibration points with following columns + # Generate a CSV file containing the calibration points with the following columns calib_csv_file = f"{Path().absolute()}/app/services/calib_validation/csv/data/{file_name}_fixed_train_data.csv" csv_columns = [ "left_iris_x", @@ -105,7 +182,7 @@ def calib_results(): "screen_width", ] - # Save calibration points to CSV file + # Save the calibration points to the CSV file try: # Open CSV file with open(calib_csv_file, "w") as csvfile: @@ -122,7 +199,7 @@ def calib_results(): except IOError: print("I/O error") - # Generate csv of iris points of session + # Generate a CSV file containing the calibration iris samples os.makedirs( f"{Path().absolute()}/app/services/calib_validation/csv/data/", exist_ok=True ) @@ -150,18 +227,30 @@ def calib_results(): "model": model, "screen_height": screen_height, "screen_width": screen_width, - "k": k + "k": k, + "fixed_points": fixed_points, + "calib_points": calib_points, + "timestamp": timestamp, } FUNCTIONS_ENDPOINT_URL = os.getenv('FUNCTIONS_ENDPOINT_URL') FUNCTIONS_ENDPOINT_URL+='/receiveCalibration' - print("file_name:", file_name) + print( + "[calibration] sending calibration", + { + "session_id": file_name, + "user_id": user_id, + "study_id": study_id, + "fixed_points": len(fixed_points), + "calib_points": len(calib_points), + } + ) resp = requests.post(FUNCTIONS_ENDPOINT_URL, json=payload) - print("Enviado para RuxaiLab:", resp.status_code, resp.text) + print("Sent to RuxaiLab:", resp.status_code, resp.text) except Exception as e: - print("Erro ao enviar para RuxaiLab:", e) + print("Error sending to RuxaiLab:", e) # Convert NaN values to None before returning JSON data = convert_nan_to_none(data) @@ -170,48 +259,404 @@ def calib_results(): def batch_predict(): try: data = request.get_json() + iris_data = data["iris_tracking_data"] + calib_points = data["calib_points"] + fixed_points = data["fixed_points"] + screen_width = data.get("screen_width") screen_height = data.get("screen_height") model_name = data.get("model_name", "Linear Regression") + k = data.get("k", 5) calib_id = data.get("calib_id") - if not calib_id: - return Response("Missing calib_id", status=400) - - calib_id = sanitize_filename(calib_id) - - base_path = Path().absolute() / "app/services/calib_validation/csv/data" - calib_csv_path = base_path / f"{calib_id}_fixed_train_data.csv" - predict_csv_path = base_path / f"temp_batch_predict_{uuid.uuid4().hex}.csv" - - # CSV temporário - with open(predict_csv_path, "w", newline="") as csvfile: - writer = csv.DictWriter(csvfile, fieldnames=[ - "left_iris_x", "left_iris_y", "right_iris_x", "right_iris_y" - ]) - writer.writeheader() - for item in iris_data: - writer.writerow({ - "left_iris_x": item["left_iris_x"], - "left_iris_y": item["left_iris_y"], - "right_iris_x": item["right_iris_x"], - "right_iris_y": item["right_iris_y"], - }) - + print("[eye-tracking-diagnostic] batch input", { + "screen_width": screen_width, + "screen_height": screen_height, + "calib_id": calib_id, + "model_name": model_name, + "k": k, + "calib_points_count": len(calib_points), + "fixed_points_count": len(fixed_points), + **summarize_batch_input( + iris_data, + screen_width, + screen_height, + ), + }) + + if not calib_points: + return Response("Missing calib_points", status=400) + + if not fixed_points: + return Response("Missing fixed_points", status=400) + + # --------------------------------------------------------- + # Predict live/session gaze data + # --------------------------------------------------------- result = gaze_tracker.predict_new_data_simple( - calib_csv_path=calib_csv_path, - predict_csv_path=predict_csv_path, + fixed_points=fixed_points, iris_data=iris_data, model_name=model_name, screen_width=screen_width, screen_height=screen_height, ) - os.remove(predict_csv_path) - - return jsonify(convert_nan_to_none(result)) + + predicted_x = [ + item.get("predicted_x") + for item in result + ] + + predicted_y = [ + item.get("predicted_y") + for item in result + ] + # --------------------------------------------------------- + # Predict calibration samples for model evaluation + # --------------------------------------------------------- + + calibration_result = ( + gaze_tracker.predict_new_data_simple( + fixed_points=fixed_points, + iris_data=calib_points, + model_name=model_name, + screen_width=screen_width, + screen_height=screen_height, + ) + ) + + calibration_metrics = calculate_calibration_metrics( + fixed_points=fixed_points, + calibration_result=calibration_result, + calib_points_count=len(calib_points), + ) + + print("[eye-tracking-diagnostic] calibration metrics", { + "samples": calibration_metrics["samples"], + "points": calibration_metrics["points"], + "mean_error_px": calibration_metrics["mean_error_px"], + "median_error_px": calibration_metrics["median_error_px"], + "rmse_px": calibration_metrics["rmse_px"], + "horizontal_mae_px": calibration_metrics[ + "horizontal_mae_px" + ], + "vertical_mae_px": calibration_metrics[ + "vertical_mae_px" + ], + "accuracy_50px": calibration_metrics[ + "accuracy_50px" + ], + "accuracy_100px": calibration_metrics[ + "accuracy_100px" + ], + "quality": calibration_metrics["quality"], + }) + + + print("[eye-tracking-diagnostic] batch response", { + "prediction_count": len(result), + "predicted_x": summarize_numeric_values(predicted_x), + "predicted_y": summarize_numeric_values(predicted_y), + "screen_width": screen_width, + "screen_height": screen_height, + }) + + return jsonify( + convert_nan_to_none({ + "predictions": result, + "metrics": { + "calibration": calibration_metrics, + "session": { + "prediction_count": len(result), + }, + }, + }) + ) except Exception as e: - print("Erro batch_predict:", e) + print("Error in batch_predict:", e) traceback.print_exc() - return Response("Erro interno", status=500) \ No newline at end of file + return Response("Internal Error", status=500) + +def calculate_calibration_metrics( + fixed_points, + calibration_result, + calib_points_count, +): + """ + Calculate calibration performance metrics by comparing predictions + from calib_points against their corresponding fixed calibration targets. + + calib_points are expected to be ordered by target point, with the + same number of samples collected for each calibration target. + """ + + if not fixed_points or not calibration_result: + return { + "samples": 0, + "points": 0, + "mean_error_px": None, + "median_error_px": None, + "rmse_px": None, + "horizontal_mae_px": None, + "vertical_mae_px": None, + "accuracy_50px": None, + "accuracy_100px": None, + "quality": "Unknown", + "points_detail": [], + } + + # --------------------------------------------------------- + # Get unique calibration targets preserving their order. + # --------------------------------------------------------- + + calibration_targets = [] + seen_targets = set() + + for point in fixed_points: + point_x = point.get("point_x") + point_y = point.get("point_y") + + if point_x is None or point_y is None: + continue + + target = ( + float(point_x), + float(point_y), + ) + + if target not in seen_targets: + seen_targets.add(target) + calibration_targets.append(target) + + if not calibration_targets: + return { + "samples": 0, + "points": 0, + "mean_error_px": None, + "median_error_px": None, + "rmse_px": None, + "horizontal_mae_px": None, + "vertical_mae_px": None, + "accuracy_50px": None, + "accuracy_100px": None, + "quality": "Unknown", + "points_detail": [], + } + + # --------------------------------------------------------- + # calib_points are ordered by target point. + # --------------------------------------------------------- + + total_predictions = min( + len(calibration_result), + calib_points_count, + ) + + point_count = len(calibration_targets) + samples_per_point = total_predictions // point_count + + if samples_per_point <= 0: + return { + "samples": 0, + "points": point_count, + "mean_error_px": None, + "median_error_px": None, + "rmse_px": None, + "horizontal_mae_px": None, + "vertical_mae_px": None, + "accuracy_50px": None, + "accuracy_100px": None, + "quality": "Unknown", + "points_detail": [], + } + + errors = [] + horizontal_errors = [] + vertical_errors = [] + + points_detail = [] + + # --------------------------------------------------------- + # Evaluate each calibration point independently. + # --------------------------------------------------------- + + for point_index, (target_x, target_y) in enumerate( + calibration_targets + ): + start_index = point_index * samples_per_point + end_index = start_index + samples_per_point + + point_predictions = calibration_result[ + start_index:end_index + ] + + point_errors = [] + point_horizontal_errors = [] + point_vertical_errors = [] + + for prediction in point_predictions: + predicted_x = prediction.get("predicted_x") + predicted_y = prediction.get("predicted_y") + + if ( + predicted_x is None + or predicted_y is None + ): + continue + + try: + predicted_x = float(predicted_x) + predicted_y = float(predicted_y) + + horizontal_error = abs( + predicted_x - target_x + ) + + vertical_error = abs( + predicted_y - target_y + ) + + euclidean_error = ( + (predicted_x - target_x) ** 2 + + (predicted_y - target_y) ** 2 + ) ** 0.5 + + horizontal_errors.append(horizontal_error) + vertical_errors.append(vertical_error) + errors.append(euclidean_error) + + point_horizontal_errors.append( + horizontal_error + ) + point_vertical_errors.append( + vertical_error + ) + point_errors.append(euclidean_error) + + except (TypeError, ValueError): + continue + + # Use the existing numeric summarizer instead of manually + # calculating min/max/mean/std for each error collection. + point_error_stats = summarize_numeric_values( + point_errors + ) + + point_accuracy_50 = ( + np.mean( + np.asarray(point_errors) <= 50 + ) * 100 + if point_errors + else None + ) + + point_accuracy_100 = ( + np.mean( + np.asarray(point_errors) <= 100 + ) * 100 + if point_errors + else None + ) + + points_detail.append({ + "target_x": target_x, + "target_y": target_y, + "samples": len(point_errors), + "mean_error_px": ( + round(point_error_stats["mean"], 2) + if point_error_stats["mean"] is not None + else None + ), + "std_error_px": ( + round(point_error_stats["std"], 2) + if point_error_stats["std"] is not None + else None + ), + "accuracy_50px": ( + round(float(point_accuracy_50), 2) + if point_accuracy_50 is not None + else None + ), + "accuracy_100px": ( + round(float(point_accuracy_100), 2) + if point_accuracy_100 is not None + else None + ), + }) + + # --------------------------------------------------------- + # Global metrics + # --------------------------------------------------------- + + error_stats = summarize_numeric_values(errors) + horizontal_stats = summarize_numeric_values( + horizontal_errors + ) + vertical_stats = summarize_numeric_values( + vertical_errors + ) + + if not errors: + return { + "samples": 0, + "points": point_count, + "mean_error_px": None, + "median_error_px": None, + "rmse_px": None, + "horizontal_mae_px": None, + "vertical_mae_px": None, + "accuracy_50px": None, + "accuracy_100px": None, + "quality": "Unknown", + "points_detail": points_detail, + } + + errors_array = np.asarray(errors, dtype=float) + + median_error = float(np.median(errors_array)) + + rmse = float( + np.sqrt(np.mean(np.square(errors_array))) + ) + + accuracy_50 = float( + np.mean(errors_array <= 50) * 100 + ) + + accuracy_100 = float( + np.mean(errors_array <= 100) * 100 + ) + + mean_error = error_stats["mean"] + + # --------------------------------------------------------- + # Calibration quality + # --------------------------------------------------------- + + if mean_error <= 40: + quality = "Excellent" + elif mean_error <= 75: + quality = "Good" + elif mean_error <= 120: + quality = "Fair" + else: + quality = "Poor" + + return { + "samples": len(errors), + "points": point_count, + "mean_error_px": round(mean_error, 2), + "median_error_px": round(median_error, 2), + "rmse_px": round(rmse, 2), + "horizontal_mae_px": round( + horizontal_stats["mean"], 2 + ), + "vertical_mae_px": round( + vertical_stats["mean"], 2 + ), + "accuracy_50px": round(accuracy_50, 2), + "accuracy_100px": round(accuracy_100, 2), + "quality": quality, + "points_detail": points_detail, + } \ No newline at end of file diff --git a/app/services/gaze_tracker.py b/app/services/gaze_tracker.py index 9f5b68e..182505f 100644 --- a/app/services/gaze_tracker.py +++ b/app/services/gaze_tracker.py @@ -120,7 +120,7 @@ } def squash(v, limit=1.0): - """Squash não-linear estilo WebGazer""" + """Non-linear squash inspired by WebGazer""" return np.tanh(v / limit) def train_and_predict(model_name, X_train, y_train, X_test, y_test, label): @@ -272,45 +272,82 @@ def predict(data, k, model_X, model_Y): return data def predict_new_data_simple( - calib_csv_path, - predict_csv_path, + fixed_points, iris_data, model_name="Linear Regression", screen_width=None, screen_height=None, -): +): # ============================ - # CONFIG (WebGazer-inspired) + # CONFIG # ============================ BASELINE_ALPHA = 0.01 - SQUASH_LIMIT_X = 1.0 SQUASH_LIMIT_Y = 1.0 - Y_GAIN = 1.2 # adjustment to compensate for vertical bias + Y_GAIN = 1.2 # ============================ - # LOAD TRAIN + # LOAD CALIBRATION # ============================ - df_train = pd.read_csv(calib_csv_path) + df_train = pd.DataFrame(fixed_points) + + def diagnostic_stats(values): + values = np.asarray(values, dtype=float) + + return { + "min": float(np.min(values)) if values.size else None, + "max": float(np.max(values)) if values.size else None, + "mean": float(np.mean(values)) if values.size else None, + "std": float(np.std(values)) if values.size else None, + } + + print("[eye-tracking-diagnostic] calibration artifact stats", { + "source": "request.fixed_points", + "rows": len(df_train), + "screen_width": screen_width, + "screen_height": screen_height, + "fields": { + field: diagnostic_stats(df_train[field].values) + for field in [ + "left_iris_x", + "right_iris_x", + "left_iris_y", + "right_iris_y", + "point_x", + "point_y", + ] + }, + "calibration_screen_dimensions": { + "width": diagnostic_stats(df_train["screen_width"].values) + if "screen_width" in df_train else None, + "height": diagnostic_stats(df_train["screen_height"].values) + if "screen_height" in df_train else None, + }, + "unique_point_x": sorted( + df_train["point_x"].dropna().unique().tolist() + ), + }) x_center = screen_width / 2 y_center = screen_height / 2 - # normalize targets to [-1, 1] space - y_train_x = (df_train["point_x"].values.astype(float) - x_center) / (screen_width / 2) - y_train_y = (df_train["point_y"].values.astype(float) - y_center) / (screen_height / 2) - - # ensure laterality + # ============================ + # ENSURE LATERALITY + # ============================ if df_train["left_iris_x"].mean() < df_train["right_iris_x"].mean(): df_train["left_iris_x"], df_train["right_iris_x"] = ( df_train["right_iris_x"].copy(), df_train["left_iris_x"].copy(), ) + if df_train["left_iris_y"].mean() < df_train["right_iris_y"].mean(): df_train["left_iris_y"], df_train["right_iris_y"] = ( df_train["right_iris_y"].copy(), df_train["left_iris_y"].copy(), ) + # ============================ + # CALIBRATION IRIS DATA + # ============================ left_x = df_train["left_iris_x"].values.astype(float) right_x = df_train["right_iris_x"].values.astype(float) left_y = df_train["left_iris_y"].values.astype(float) @@ -318,20 +355,102 @@ def predict_new_data_simple( mean_x = (left_x + right_x) / 2 diff_x = left_x - right_x + mean_y = (left_y + right_y) / 2 diff_y = left_y - right_y - # baseline inicial (WebGazer) - ref_mean_x = np.mean(mean_x) - ref_mean_y = np.mean(mean_y) + # ============================ + # REMOVE CALIBRATION OUTLIERS + # ============================ + def robust_mask(values, threshold=6.0): + values = np.asarray(values, dtype=float) + + median = np.median(values) + mad = np.median(np.abs(values - median)) + + if mad < 1e-9: + return np.ones(len(values), dtype=bool) + + robust_z = 0.6745 * (values - median) / mad + + return np.abs(robust_z) <= threshold + + valid_mask_x = ( + robust_mask(left_x) + & robust_mask(right_x) + & robust_mask(mean_x) + ) + + removed_count = int(np.sum(~valid_mask_x)) + + if removed_count > 0: + print( + "[eye-tracking-diagnostic] calibration outliers removed", + { + "removed": removed_count, + "remaining": int(np.sum(valid_mask_x)), + }, + ) + + left_x = left_x[valid_mask_x] + right_x = right_x[valid_mask_x] + mean_x = mean_x[valid_mask_x] + diff_x = diff_x[valid_mask_x] + + left_y = left_y[valid_mask_x] + right_y = right_y[valid_mask_x] + mean_y = mean_y[valid_mask_x] + diff_y = diff_y[valid_mask_x] + + point_x = df_train["point_x"].values.astype(float)[valid_mask_x] + point_y = df_train["point_y"].values.astype(float)[valid_mask_x] + else: + point_x = df_train["point_x"].values.astype(float) + point_y = df_train["point_y"].values.astype(float) + + print("[eye-tracking-diagnostic] calibration after filtering", { + "rows": len(mean_x), + "removed": removed_count, + "mean_x": diagnostic_stats(mean_x), + "diff_x": diagnostic_stats(diff_x), + "mean_y": diagnostic_stats(mean_y), + "diff_y": diagnostic_stats(diff_y), + }) + + # ============================ + # BASELINE + # ============================ + ref_mean_x = float(np.mean(mean_x)) + ref_mean_y = float(np.mean(mean_y)) rel_x = mean_x - ref_mean_x rel_y = mean_y - ref_mean_y + print("[eye-tracking-diagnostic] calibration baseline", { + "ref_mean_x": ref_mean_x, + "ref_mean_y": ref_mean_y, + "rel_x": diagnostic_stats(rel_x), + "rel_y": diagnostic_stats(rel_y), + }) + + # ============================ + # NORMALIZE TARGETS + # ============================ + y_train_x = ( + (point_x - x_center) + / (screen_width / 2) + ) + + y_train_y = ( + (point_y - y_center) + / (screen_height / 2) + ) + # ============================ # PHYSICAL NORMALIZATION Y # ============================ iris_y_scale = np.std(mean_y) + 1e-6 + diff_y_norm = diff_y / iris_y_scale rel_y_norm = rel_y / iris_y_scale @@ -339,100 +458,256 @@ def predict_new_data_simple( # FEATURES # ============================ X_train_x = np.column_stack([ - left_x, right_x, mean_x, diff_x, rel_x + diff_x, + rel_x, ]) X_train_y = np.column_stack([ - diff_y_norm, rel_y_norm + diff_y_norm, + rel_y_norm, ]) + print("[eye-tracking-diagnostic] training features", { + "X_columns": ["diff_x", "rel_x"], + "X_diff_x": diagnostic_stats(X_train_x[:, 0]), + "X_rel_x": diagnostic_stats(X_train_x[:, 1]), + "Y_columns": ["diff_y_norm", "rel_y_norm"], + }) + # ============================ # MODELS # ============================ + model_x = models_gaze_engineered.get( + model_name, + models_gaze_engineered["Linear Regression"], + ) + + model_y = models.get( + model_name, + models["Linear Regression"], + ) - model_x=models_gaze_engineered.get(model_name,models_gaze_engineered['Linear Regression']) - model_y=models.get(model_name,models['Linear Regression']) + model_x_key_exists = model_name in models_gaze_engineered + model_y_key_exists = model_name in models + + print("[eye-tracking-diagnostic] model selection", { + "requested_model_name": model_name, + "model_x_configured": model_x_key_exists, + "model_y_configured": model_y_key_exists, + "model_x_type": type(model_x).__name__, + "model_y_type": type(model_y).__name__, + "fallback": not ( + model_x_key_exists + and model_y_key_exists + ), + "calibration_source": "request.fixed_points", + }) + # ============================ + # FIT + # ============================ model_x.fit(X_train_x, y_train_x) model_y.fit(X_train_y, y_train_y) # ============================ - # Real scale (calibration) - normalize predicted values to screen coordinates + # TARGET SCALE # ============================ - x_range = np.percentile(y_train_x, 95) - np.percentile(y_train_x, 5) - y_range = np.percentile(y_train_y, 95) - np.percentile(y_train_y, 5) + x_range = ( + np.percentile(y_train_x, 95) + - np.percentile(y_train_x, 5) + ) + + y_range = ( + np.percentile(y_train_y, 95) + - np.percentile(y_train_y, 5) + ) - x_scale = max(x_range / 2, 1e-6) * (screen_width / 2) - y_scale = max(y_range / 2, 1e-6) * (screen_height / 2) + x_scale = screen_width / 2 + + y_scale = ( + max(y_range / 2, 1e-6) + * (screen_height / 2) + ) # ============================ - # LOAD PREDICT + # LOAD PREDICTION DATA # ============================ - df_pred = pd.read_csv(predict_csv_path) + df_pred = pd.DataFrame(iris_data) if df_pred["left_iris_x"].mean() < df_pred["right_iris_x"].mean(): df_pred["left_iris_x"], df_pred["right_iris_x"] = ( df_pred["right_iris_x"].copy(), df_pred["left_iris_x"].copy(), ) + if df_pred["left_iris_y"].mean() < df_pred["right_iris_y"].mean(): df_pred["left_iris_y"], df_pred["right_iris_y"] = ( df_pred["right_iris_y"].copy(), df_pred["left_iris_y"].copy(), ) + # ============================ + # PREDICTION IRIS DATA + # ============================ left_px = df_pred["left_iris_x"].values.astype(float) right_px = df_pred["right_iris_x"].values.astype(float) + left_py = df_pred["left_iris_y"].values.astype(float) right_py = df_pred["right_iris_y"].values.astype(float) mean_px = (left_px + right_px) / 2 diff_px = left_px - right_px + mean_py = (left_py + right_py) / 2 diff_py = left_py - right_py - # baseline relativo + # ============================ + # RELATIVE FEATURES + # ============================ rel_px = mean_px - ref_mean_x rel_py = mean_py - ref_mean_y + # ========================================================== + # ALIGN PREDICTION DISTRIBUTION WITH CALIBRATION + # ========================================================== + # The prediction data may have a different relative X distribution + # from the calibration data due to changes in head position or camera + # alignment between calibration and prediction. + # + # Center the prediction distribution around the calibration mean + # to keep the input features within the distribution observed + # during training and reduce unwanted extrapolation. + prediction_rel_x_mean = float(np.mean(rel_px)) + calibration_rel_x_mean = float(np.mean(rel_x)) + + rel_px = ( + rel_px + - prediction_rel_x_mean + + calibration_rel_x_mean + ) + + print("[eye-tracking-diagnostic] rel_x alignment", { + "calibration_mean": calibration_rel_x_mean, + "prediction_mean_before": prediction_rel_x_mean, + "prediction_mean_after": float(np.mean(rel_px)), + "prediction_before": diagnostic_stats( + mean_px - ref_mean_x + ), + "prediction_after": diagnostic_stats(rel_px), + }) + diff_py_norm = diff_py / iris_y_scale rel_py_norm = rel_py / iris_y_scale + # ============================ + # PREDICTION FEATURES + # ============================ X_pred_x = np.column_stack([ - left_px, right_px, mean_px, diff_px, rel_px + diff_px, + rel_px, ]) X_pred_y = np.column_stack([ - diff_py_norm, rel_py_norm + diff_py_norm, + rel_py_norm, ]) + print("[eye-tracking-diagnostic] X features", { + "column_order": [ + "diff_x", + "rel_x", + ], + "calibration": { + "diff_x": diagnostic_stats(X_train_x[:, 0]), + "rel_x": diagnostic_stats(X_train_x[:, 1]), + }, + "prediction": { + "diff_x": diagnostic_stats(X_pred_x[:, 0]), + "rel_x": diagnostic_stats(X_pred_x[:, 1]), + }, + "baseline": { + "ref_mean_x": ref_mean_x, + "prediction_mean_x": float(np.mean(mean_px)), + "prediction_rel_x_mean": float(np.mean(rel_px)), + }, + }) + + # ============================ + # RAW PREDICTIONS + # ============================ y_pred_x = model_x.predict(X_pred_x) + + print("[eye-tracking-diagnostic] raw X prediction", { + "min": float(np.min(y_pred_x)), + "max": float(np.max(y_pred_x)), + "mean": float(np.mean(y_pred_x)), + "std": float(np.std(y_pred_x)), + "p5": float(np.percentile(y_pred_x, 5)), + "p25": float(np.percentile(y_pred_x, 25)), + "p50": float(np.percentile(y_pred_x, 50)), + "p75": float(np.percentile(y_pred_x, 75)), + "p95": float(np.percentile(y_pred_x, 95)), + "first_5": y_pred_x[:5].tolist(), + "last_5": y_pred_x[-5:].tolist(), + }) + y_pred_y = model_y.predict(X_pred_y) - # remove bias vertical + # ============================ + # REMOVE VERTICAL BIAS + # ============================ y_pred_y = y_pred_y - np.mean(y_pred_y) - y_pred_y = y_pred_y * Y_GAIN - # ============================ - # PREDICTION LOOP (WebGazer) + # PREDICTION LOOP # ============================ predictions = [] - window_size = 30 + dynamic_ref_x = ref_mean_x + dynamic_ref_y = ref_mean_y for i in range(len(y_pred_x)): - # baseline dinâmico - ref_mean_x = BASELINE_ALPHA * mean_px[i] + (1 - BASELINE_ALPHA) * ref_mean_x - ref_mean_y = BASELINE_ALPHA * mean_py[i] + (1 - BASELINE_ALPHA) * ref_mean_y - # squash não-linear - sx = squash(y_pred_x[i], SQUASH_LIMIT_X) - sy = squash(y_pred_y[i], SQUASH_LIMIT_Y) + dynamic_ref_x = ( + BASELINE_ALPHA * mean_px[i] + + (1 - BASELINE_ALPHA) * dynamic_ref_x + ) + + dynamic_ref_y = ( + BASELINE_ALPHA * mean_py[i] + + (1 - BASELINE_ALPHA) * dynamic_ref_y + ) + + # ============================ + # X + # ============================ + sx = np.clip( + y_pred_x[i], + -1.0, + 1.0, + ) + + # ============================ + # Y + # ============================ + sy = squash( + y_pred_y[i], + SQUASH_LIMIT_Y, + ) + + # ============================ + # SCREEN PIXELS + # ============================ + px = ( + x_center + + float(sx) * x_scale + ) - px = x_center + float(sx) * x_scale - py = y_center + float(sy) * y_scale + py = ( + y_center + + float(sy) * y_scale + ) predictions.append({ "timestamp": iris_data[i].get("timestamp"), @@ -443,19 +718,56 @@ def predict_new_data_simple( }) # ============================ - # LOGS + # X TRANSFORMATION DIAGNOSTIC + # ============================ + squashed_x = np.clip( + y_pred_x, + -1.0, + 1.0, + ) + + predicted_x = np.asarray([ + prediction["predicted_x"] + for prediction in predictions + ]) + + print("[eye-tracking-diagnostic] X transformation", { + "x_range": float(x_range), + "x_scale": float(x_scale), + "raw_prediction": diagnostic_stats(y_pred_x), + "squashed_prediction": diagnostic_stats(squashed_x), + "predicted_x": { + **diagnostic_stats(predicted_x), + "p5": float(np.percentile(predicted_x, 5)), + "p50": float(np.percentile(predicted_x, 50)), + "p95": float(np.percentile(predicted_x, 95)), + }, + }) + + # ============================ + # FINAL DEBUG # ============================ print("====== MODEL DEBUG ======") - print(f"y_pred_x: {np.min(y_pred_x):.3f} → {np.max(y_pred_x):.3f}") - print(f"y_pred_y: {np.min(y_pred_y):.3f} → {np.max(y_pred_y):.3f}") + print( + f"y_pred_x: " + f"{np.min(y_pred_x):.3f} → " + f"{np.max(y_pred_x):.3f}" + ) + print( + f"y_pred_y: " + f"{np.min(y_pred_y):.3f} → " + f"{np.max(y_pred_y):.3f}" + ) print("=========================") print("====== PIXEL SAMPLE ======") + for p in predictions[:15]: - print(f"x: {p['predicted_x']:.1f}, y: {p['predicted_y']:.1f}") + print( + f"x: {p['predicted_x']:.1f}, " + f"y: {p['predicted_y']:.1f}" + ) return predictions - - def normalizeData(data): return (data - np.min(data)) / (np.max(data) - np.min(data))