From 6a507f4d43b9cf3cf2c0786af4a4b78b0c71ae7f Mon Sep 17 00:00:00 2001 From: harsshks Date: Fri, 30 Jan 2026 00:23:35 +0530 Subject: [PATCH] feat(eyeTracking): improve batch_predict visualization payload Adds unified response structure with normalized gaze points and optional heatmap generation while preserving backward compatibility. Closes #47. --- app/routes/session.py | 35 ++++++++++++++--- app/services/gaze_tracker.py | 57 ++++++++++++++++++++------- app/services/heatmap.py | 76 ++++++++++++++++++++++++++++++++++++ 3 files changed, 147 insertions(+), 21 deletions(-) diff --git a/app/routes/session.py b/app/routes/session.py index 1db1859..fa24b92 100644 --- a/app/routes/session.py +++ b/app/routes/session.py @@ -1,6 +1,7 @@ # Necesary imports import os import re +import uuid import time import json import csv @@ -21,6 +22,7 @@ # from app.services import database as db from app.services import gaze_tracker +from app.services import heatmap as heatmap_service # Constants @@ -148,11 +150,13 @@ def calib_results(): return Response(json.dumps(data), status=200, mimetype='application/json') def batch_predict(): + predict_csv_path = None try: data = request.get_json() iris_data = data["iris_tracking_data"] screen_width = data.get("screen_width") screen_height = data.get("screen_height") + # model_X and model_Y are retrieved but not currently used in the simple predictor model_X = data.get("model_X", "Linear Regression") model_Y = data.get("model_Y", "Linear Regression") calib_id = data.get("calib_id") @@ -162,9 +166,12 @@ def batch_predict(): 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 / "temp_batch_predict.csv" + + # Use UUID to ensure unique filename for each request (prevent concurrency issues) + unique_id = str(uuid.uuid4()) + predict_csv_path = base_path / f"temp_batch_predict_{unique_id}.csv" - # CSV temporário + # Write temporary CSV 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" @@ -178,19 +185,35 @@ def batch_predict(): "right_iris_y": item["right_iris_y"], }) + # require screen dims for correct normalization and heatmap + if screen_width is None or screen_height is None: + return Response("Missing screen_width or screen_height", status=400) + result = gaze_tracker.predict_new_data_simple( calib_csv_path=calib_csv_path, predict_csv_path=predict_csv_path, iris_data=iris_data, - # model_X="Random Forest Regressor", - # model_Y="Random Forest Regressor", screen_width=screen_width, screen_height=screen_height, ) + + # attach a lightweight heatmap array if possible + try: + result_with_heat = heatmap_service.attach_heatmap_to_payload(result) + except Exception: + result_with_heat = result - return jsonify(convert_nan_to_none(result)) + return jsonify(convert_nan_to_none(result_with_heat)) except Exception as e: print("Erro batch_predict:", e) traceback.print_exc() - return Response("Erro interno", status=500) \ No newline at end of file + return Response("Erro interno", status=500) + + finally: + # Clean up temporary file + if predict_csv_path and predict_csv_path.exists(): + try: + predict_csv_path.unlink() + except Exception: + pass \ No newline at end of file diff --git a/app/services/gaze_tracker.py b/app/services/gaze_tracker.py index 7d1f7ce..4f35497 100644 --- a/app/services/gaze_tracker.py +++ b/app/services/gaze_tracker.py @@ -398,38 +398,65 @@ def predict_new_data_simple( predictions = [] for i in range(len(y_pred_x)): - # baseline dinâmico + # dynamic baseline 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 + # non-linear squash sx = squash(y_pred_x[i], SQUASH_LIMIT_X) sy = squash(y_pred_y[i], SQUASH_LIMIT_Y) px = x_center + float(sx) * x_scale py = y_center + float(sy) * y_scale + # normalized coordinates (0..1) when screen dims provided + x_norm = None + y_norm = None + if screen_width and screen_height: + try: + x_norm = float(px) / float(screen_width) + y_norm = float(py) / float(screen_height) + except Exception: + x_norm = None + y_norm = None + + # simple confidence heuristic based on normalized vertical iris difference + conf = None + try: + conf = float(np.exp(-np.abs(diff_py_norm[i]))) + conf = max(0.0, min(1.0, conf)) + except Exception: + conf = 1.0 + predictions.append({ "timestamp": iris_data[i].get("timestamp"), - "predicted_x": px, - "predicted_y": py, - "screen_width": screen_width, - "screen_height": screen_height, + "x": float(px), + "y": float(py), + "x_norm": x_norm, + "y_norm": y_norm, + "confidence": conf, }) # ============================ # LOGS # ============================ - 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("=========================") - - print("====== PIXEL SAMPLE ======") - for p in predictions[:15]: - print(f"x: {p['predicted_x']:.1f}, y: {p['predicted_y']:.1f}") + try: + 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("=========================") + except Exception: + pass + + # Top-level unified payload for visualizers + payload = { + "screen": {"width": screen_width, "height": screen_height}, + "predictions": predictions, + "normalized": bool(screen_width and screen_height), + "heatmap": None, + } - return predictions + return payload def normalizeData(data): diff --git a/app/services/heatmap.py b/app/services/heatmap.py index a7c18b2..884d2aa 100644 --- a/app/services/heatmap.py +++ b/app/services/heatmap.py @@ -1 +1,77 @@ +# To be added later on the project +import numpy as np + +def generate_heatmap(predictions, width, height, bins=(64, 64), sigma=1.5): + """ + Generate a simple 2D density heatmap from predictions. + + Args: + predictions (list): list of dicts with keys `x` and `y` (pixels). + width (int): screen width in pixels. + height (int): screen height in pixels. + bins (tuple): bins for (x_bins, y_bins). + sigma (float): gaussian smoothing sigma in bins. + + Returns: + list(list(float)): 2D array (y major) normalized to 0..1 suitable for JSON. + """ + if not predictions or not width or not height: + return None + + xs = [] + ys = [] + for p in predictions: + x = p.get("x") or p.get("predicted_x") + y = p.get("y") or p.get("predicted_y") + if x is None or y is None: + continue + xs.append(x) + ys.append(y) + + if len(xs) == 0: + return None + + x_bins, y_bins = bins + # Use numpy histogram2d (note: histogram2d expects x, y) + heat, xedges, yedges = np.histogram2d(xs, ys, bins=[x_bins, y_bins], range=[[0, width], [0, height]]) + + # transpose so rows correspond to y (top->bottom) + heat = heat.T + + # gaussian smoothing in frequency domain (approx) + try: + from scipy.ndimage import gaussian_filter + + heat = gaussian_filter(heat, sigma=sigma) + except Exception: + # fallback: simple local normalization if scipy not available + pass + + # normalize + mn = float(np.min(heat)) + mx = float(np.max(heat)) + if mx - mn > 0: + heat = (heat - mn) / (mx - mn) + else: + heat = heat * 0.0 + + return heat.tolist() + + +def attach_heatmap_to_payload(payload, bins=(64, 64), sigma=1.5): + """Attach generated heatmap to the payload produced by `predict_new_data_simple`. + + Returns a modified payload (copy) with `heatmap` populated if possible. + """ + if not payload or "predictions" not in payload: + return payload + + screen = payload.get("screen") or {} + width = screen.get("width") + height = screen.get("height") + + heat = generate_heatmap(payload["predictions"], width, height, bins=bins, sigma=sigma) + out = dict(payload) + out["heatmap"] = heat + return out # To be added later on the project \ No newline at end of file