|
|
|
|
|
import pandas as pd |
|
|
import numpy as np |
|
|
import matplotlib.pyplot as plt |
|
|
import matplotlib.patches as patches |
|
|
import seaborn as sns |
|
|
from scipy import stats |
|
|
from sklearn.preprocessing import StandardScaler |
|
|
from sklearn.ensemble import RandomForestClassifier |
|
|
from sklearn.model_selection import train_test_split |
|
|
from sklearn.metrics import classification_report, roc_auc_score |
|
|
import warnings |
|
|
import gradio as gr |
|
|
import os |
|
|
import git |
|
|
|
|
|
|
|
|
warnings.filterwarnings('ignore') |
|
|
plt.style.use('default') |
|
|
sns.set_palette("husl") |
|
|
|
|
|
class EnhancedAIvsRealGazeAnalyzer: |
|
|
def __init__(self): |
|
|
self.questions = ['Q1', 'Q2', 'Q3', 'Q4', 'Q5', 'Q6'] |
|
|
self.correct_answers = {'Pair1': 'B', 'Pair2': 'B', 'Pair3': 'B', 'Pair4': 'B', 'Pair5': 'B', 'Pair6': 'B'} |
|
|
self.combined_data = None |
|
|
self.fixation_data = {} |
|
|
self.valid_playback_participants = [] |
|
|
self.valid_playback_trials = {} |
|
|
self.model = None |
|
|
self.scaler = None |
|
|
self.feature_names = [] |
|
|
self.time_metrics = [] |
|
|
|
|
|
def _find_and_standardize_participant_col(self, df, filename): |
|
|
participant_col = next((c for c in df.columns if 'participant' in str(c).lower()), None) |
|
|
if not participant_col: |
|
|
raise ValueError(f"Could not find a 'participant' column in the file: {filename}") |
|
|
df = df.rename(columns={participant_col: 'participant_id'}) |
|
|
df['participant_id'] = df['participant_id'].astype(str) |
|
|
return df |
|
|
|
|
|
def load_and_process_data(self, base_path, response_file_path): |
|
|
print("--- Starting Robust Data Loading ---") |
|
|
response_df = pd.read_excel(response_file_path) |
|
|
response_df = self._find_and_standardize_participant_col(response_df, "GenAI Response.xlsx") |
|
|
for pair, ans in self.correct_answers.items(): |
|
|
if pair in response_df.columns: |
|
|
response_df[f'{pair}_Correct'] = (response_df[pair].astype(str).str.strip().str.upper() == ans) |
|
|
response_long = response_df.melt(id_vars=['participant_id'], value_vars=self.correct_answers.keys(), var_name='Pair') |
|
|
correctness_long = response_df.melt(id_vars=['participant_id'], value_vars=[f'{p}_Correct' for p in self.correct_answers.keys()], var_name='Pair_Correct_Col', value_name='Correct') |
|
|
correctness_long['Pair'] = correctness_long['Pair_Correct_Col'].str.replace('_Correct', '') |
|
|
response_long = response_long.merge(correctness_long[['participant_id', 'Pair', 'Correct']], on=['participant_id', 'Pair']) |
|
|
|
|
|
all_metrics_dfs = [] |
|
|
for q in self.questions: |
|
|
file_path = f"{base_path}/Filtered_GenAI_Metrics_cleaned_{q}.xlsx" |
|
|
if os.path.exists(file_path): |
|
|
print(f"Processing {file_path}...") |
|
|
xls = pd.ExcelFile(file_path) |
|
|
metrics_df = pd.read_excel(xls, sheet_name=0) |
|
|
metrics_df = self._find_and_standardize_participant_col(metrics_df, f"{q} Metrics") |
|
|
metrics_df['Question'] = q |
|
|
all_metrics_dfs.append(metrics_df) |
|
|
|
|
|
if len(xls.sheet_names) > 1: |
|
|
try: |
|
|
fix_df = pd.read_excel(xls, sheet_name=1) |
|
|
fix_df = self._find_and_standardize_participant_col(fix_df, f"{q} Fixations") |
|
|
fix_df.dropna(subset=['Fixation point X', 'Fixation point Y', 'Gaze event duration (ms)'], inplace=True) |
|
|
for participant_id, group in fix_df.groupby('participant_id'): |
|
|
self.fixation_data[(participant_id, q)] = group.reset_index(drop=True) |
|
|
if participant_id not in self.valid_playback_trials: |
|
|
self.valid_playback_trials[participant_id] = [] |
|
|
self.valid_playback_trials[participant_id].append(q) |
|
|
print(f" -> Successfully loaded {len(fix_df)} fixations for {q}.") |
|
|
except Exception as e: |
|
|
print(f" -> WARNING: Could not load fixation sheet for {q}. Error: {e}") |
|
|
|
|
|
if not all_metrics_dfs: raise ValueError("No aggregated metrics files were found.") |
|
|
self.combined_data = pd.concat(all_metrics_dfs, ignore_index=True) |
|
|
q_to_pair = {f'Q{i+1}': f'Pair{i+1}' for i in range(6)} |
|
|
self.combined_data['Pair'] = self.combined_data['Question'].map(q_to_pair) |
|
|
self.combined_data = self.combined_data.merge(response_long, on=['participant_id', 'Pair'], how='left') |
|
|
self.combined_data['Answer_Correctness'] = self.combined_data['Correct'].map({True: 'Correct', False: 'Incorrect'}) |
|
|
|
|
|
self.numeric_cols = self.combined_data.select_dtypes(include=np.number).columns.tolist() |
|
|
|
|
|
|
|
|
self.time_metrics = [c for c in self.numeric_cols if any(k in c.lower() for k in ['time', 'duration', 'fixation'])] |
|
|
|
|
|
self.valid_playback_participants = sorted(list(self.valid_playback_trials.keys())) |
|
|
print(f"--- Data Loading Successful. Found {len(self.valid_playback_participants)} participants with fixation data. ---") |
|
|
return self |
|
|
|
|
|
def run_prediction_model(self, test_size, n_estimators): |
|
|
leaky_features = ['participant_id', 'Correct', 'Total_Correct', 'Overall_Accuracy'] |
|
|
self.feature_names = [col for col in self.numeric_cols if col not in leaky_features] |
|
|
features = self.combined_data[self.feature_names].copy() |
|
|
target = self.combined_data['Answer_Correctness'].map({'Correct': 1, 'Incorrect': 0}) |
|
|
valid_indices = target.notna() |
|
|
features, target = features[valid_indices], target[valid_indices] |
|
|
features = features.fillna(features.median()).fillna(0) |
|
|
if len(target.unique()) < 2: return "Not enough data to train.", None, None, gr.Markdown("Model not trained yet.") |
|
|
X_train, X_test, y_train, y_test = train_test_split(features, target, test_size=test_size, random_state=42, stratify=target) |
|
|
self.scaler = StandardScaler().fit(X_train) |
|
|
self.model = RandomForestClassifier(n_estimators=int(n_estimators), random_state=42, class_weight='balanced').fit(self.scaler.transform(X_train), y_train) |
|
|
report = classification_report(y_test, self.model.predict(self.scaler.transform(X_test)), target_names=['Incorrect', 'Correct'], output_dict=True) |
|
|
auc_score = roc_auc_score(y_test, self.model.predict_proba(self.scaler.transform(X_test))[:, 1]) |
|
|
summary_md = f"### Model Performance\n- **AUC Score:** **{auc_score:.4f}**\n- **Overall Accuracy:** {report['accuracy']:.3f}" |
|
|
report_df = pd.DataFrame(report).transpose().round(3) |
|
|
feature_importance = pd.DataFrame({'Feature': self.feature_names, 'Importance': self.model.feature_importances_}).sort_values('Importance', ascending=False).head(15) |
|
|
fig, ax = plt.subplots(figsize=(10, 8)); sns.barplot(data=feature_importance, x='Importance', y='Feature', ax=ax, palette='viridis'); ax.set_title(f'Top 15 Predictive Features (n_estimators={int(n_estimators)})', fontsize=14); plt.tight_layout() |
|
|
return summary_md, report_df, fig, gr.Markdown("✅ **Model is ready!** You can now use the Gaze Playback tab.") |
|
|
|
|
|
def _recalculate_features_from_fixations(self, fixations_df): |
|
|
feature_vector = pd.Series(0.0, index=self.feature_names) |
|
|
if fixations_df.empty: return feature_vector.fillna(0).values.reshape(1, -1) |
|
|
if 'AOI name' in fixations_df.columns: |
|
|
for aoi_name, group in fixations_df.groupby('AOI name'): |
|
|
col_name = f'Total fixation duration on {aoi_name}' |
|
|
if col_name in feature_vector.index: |
|
|
feature_vector[col_name] = group['Gaze event duration (ms)'].sum() |
|
|
feature_vector['Total Recording Duration'] = fixations_df['Gaze event duration (ms)'].sum() |
|
|
return feature_vector.fillna(0).values.reshape(1, -1) |
|
|
|
|
|
def generate_gaze_playback(self, participant, question, fixation_num): |
|
|
if self.model is None: return "Please train a model in Tab 2 first.", None, gr.Slider(interactive=False) |
|
|
trial_key = (str(participant), question) |
|
|
if not participant or not question or trial_key not in self.fixation_data: |
|
|
return "Please select a valid trial.", None, gr.Slider(interactive=False, value=0) |
|
|
|
|
|
all_fixations = self.fixation_data[trial_key] |
|
|
fixation_num = int(fixation_num) |
|
|
slider_max = len(all_fixations) |
|
|
if fixation_num > slider_max: fixation_num = slider_max |
|
|
current_fixations = all_fixations.iloc[:fixation_num] |
|
|
|
|
|
partial_features = self._recalculate_features_from_fixations(current_fixations) |
|
|
prediction_prob = self.model.predict_proba(self.scaler.transform(partial_features))[0] |
|
|
prob_correct = prediction_prob[1] |
|
|
|
|
|
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8), gridspec_kw={'height_ratios': [4, 1]}) |
|
|
fig.suptitle(f"Gaze Playback for {participant} - {question}", fontsize=16, weight='bold') |
|
|
ax1.set_title(f"Displaying Fixations 1 through {fixation_num}/{slider_max}") |
|
|
ax1.set_xlim(0, 1920); ax1.set_ylim(1080, 0) |
|
|
ax1.set_aspect('equal'); ax1.tick_params(left=False, right=False, bottom=False, top=False, labelleft=False, labelbottom=False) |
|
|
ax1.add_patch(patches.Rectangle((0, 0), 1920/2, 1080, facecolor='#e0e0e0')) |
|
|
ax1.add_patch(patches.Rectangle((1920/2, 0), 1920/2, 1080, facecolor='#f0f0f0')) |
|
|
ax1.text(1920*0.25, 50, "Image A", ha='center', fontsize=14, alpha=0.7) |
|
|
ax1.text(1920*0.75, 50, "Image B", ha='center', fontsize=14, alpha=0.7) |
|
|
if not current_fixations.empty: |
|
|
points = current_fixations[['Fixation point X', 'Fixation point Y']] |
|
|
ax1.plot(points['Fixation point X'], points['Fixation point Y'], marker='o', color='grey', alpha=0.5, linestyle='-') |
|
|
ax1.scatter(points.iloc[-1]['Fixation point X'], points.iloc[-1]['Fixation point Y'], s=200, c='red', zorder=10, edgecolors='black', lw=2) |
|
|
|
|
|
ax2.set_xlim(0, 1); ax2.set_yticks([]) |
|
|
ax2.set_title("Live Prediction Confidence (Answer is 'Correct')") |
|
|
bar_color = 'green' if prob_correct > 0.5 else 'red' |
|
|
ax2.barh([0], [prob_correct], color=bar_color, height=0.5, edgecolor='black') |
|
|
ax2.axvline(0.5, color='black', linestyle='--', linewidth=1) |
|
|
ax2.text(prob_correct, 0, f" {prob_correct:.1%} ", va='center', ha='left' if prob_correct < 0.9 else 'right', color='white', weight='bold', fontsize=12) |
|
|
plt.tight_layout(rect=[0, 0, 1, 0.95]) |
|
|
|
|
|
trial_info = self.combined_data[(self.combined_data['participant_id'] == str(participant)) & (self.combined_data['Question'] == question)].iloc[0] |
|
|
summary_text = f"**Actual Answer:** `{trial_info['Answer_Correctness']}`" |
|
|
|
|
|
return summary_text, fig, gr.Slider(maximum=slider_max, value=fixation_num, interactive=True, step=1, minimum=0) |
|
|
|
|
|
def analyze_rq1_metric(self, metric): |
|
|
if not metric or metric not in self.combined_data.columns: return None, "Metric not found." |
|
|
correct = self.combined_data.loc[self.combined_data['Answer_Correctness'] == 'Correct', metric].dropna() |
|
|
incorrect = self.combined_data.loc[self.combined_data['Answer_Correctness'] == 'Incorrect', metric].dropna() |
|
|
if len(correct) < 2 or len(incorrect) < 2: return None, "Not enough data for both groups to compare." |
|
|
t_stat, p_val = stats.ttest_ind(incorrect, correct, equal_var=False, nan_policy='omit') |
|
|
fig, ax = plt.subplots(figsize=(8, 6)); sns.boxplot(data=self.combined_data, x='Answer_Correctness', y=metric, ax=ax, palette=['#66b3ff','#ff9999']); ax.set_title(f'Comparison of "{metric}" by Answer Correctness', fontsize=14); ax.set_xlabel("Answer Correctness"); ax.set_ylabel(metric); plt.tight_layout() |
|
|
summary = f"""### Analysis for: **{metric}**\n- **Mean (Correct Answers):** {correct.mean():.4f}\n- **Mean (Incorrect Answers):** {incorrect.mean():.4f}\n- **T-test p-value:** {p_val:.4f}\n\n**Conclusion:**\n- {'There is a **statistically significant** difference (p < 0.05).' if p_val < 0.05 else 'There is **no statistically significant** difference (p >= 0.05).'}""" |
|
|
return fig, summary |
|
|
|
|
|
def update_question_dropdown(self, participant): |
|
|
"""Dynamically updates the question dropdown based on the selected participant.""" |
|
|
valid_questions = self.valid_playback_trials.get(participant, []) |
|
|
return gr.Dropdown(choices=sorted(valid_questions), interactive=True, value=None, label="2. Select a Question") |
|
|
|
|
|
def handle_new_trial_selection(self, participant, question): |
|
|
"""Called when a new trial is selected. Resets the view to the first fixation.""" |
|
|
if not participant or not question: |
|
|
return "Select a trial to begin.", None, gr.Slider(value=0, interactive=False) |
|
|
initial_fixation_num = 1 |
|
|
return self.generate_gaze_playback(participant, question, initial_fixation_num) |
|
|
|
|
|
|
|
|
def setup_and_load_data(): |
|
|
repo_url = "https://github.com/RextonRZ/GenAIEyeTrackingCleanedDataset" |
|
|
repo_dir = "GenAIEyeTrackingCleanedDataset" |
|
|
if not os.path.exists(repo_dir): |
|
|
print(f"Cloning repository {repo_url}...") |
|
|
git.Repo.clone_from(repo_url, repo_dir) |
|
|
else: |
|
|
print("Data repository already exists.") |
|
|
base_path = repo_dir |
|
|
response_file_path = os.path.join(repo_dir, "GenAI Response.xlsx") |
|
|
analyzer = EnhancedAIvsRealGazeAnalyzer().load_and_process_data(base_path, response_file_path) |
|
|
return analyzer |
|
|
|
|
|
analyzer = setup_and_load_data() |
|
|
|
|
|
with gr.Blocks(theme=gr.themes.Soft()) as demo: |
|
|
gr.Markdown("# Interactive Dashboard: AI vs. Real Gaze Analysis") |
|
|
with gr.Tabs(): |
|
|
with gr.TabItem("📊 RQ1: Viewing Time vs. Correctness"): |
|
|
with gr.Row(): |
|
|
with gr.Column(scale=1): |
|
|
rq1_metric_dropdown=gr.Dropdown(choices=analyzer.time_metrics, label="Select a Time-Based Metric", value=analyzer.time_metrics[0] if analyzer.time_metrics else None) |
|
|
rq1_summary_output=gr.Markdown(label="Statistical Summary") |
|
|
with gr.Column(scale=2): |
|
|
rq1_plot_output=gr.Plot(label="Metric Comparison") |
|
|
|
|
|
with gr.TabItem("🤖 RQ2: Predicting Correctness from Gaze"): |
|
|
with gr.Row(): |
|
|
with gr.Column(scale=1): |
|
|
gr.Markdown("#### Tune Model Hyperparameters") |
|
|
rq2_test_size_slider=gr.Slider(minimum=0.1, maximum=0.5, step=0.05, value=0.3, label="Test Set Size") |
|
|
rq2_estimators_slider=gr.Slider(minimum=10, maximum=200, step=10, value=100, label="Number of Trees") |
|
|
rq2_status = gr.Markdown("Train a model to enable the Gaze Playback tab.") |
|
|
with gr.Column(scale=2): |
|
|
rq2_summary_output=gr.Markdown(label="Model Performance Summary") |
|
|
rq2_table_output=gr.Dataframe(label="Classification Report", interactive=False) |
|
|
rq2_plot_output=gr.Plot(label="Feature Importance") |
|
|
|
|
|
with gr.TabItem("👁️ Gaze Playback & Real-Time Prediction"): |
|
|
gr.Markdown("### See the Prediction Evolve with Every Glance!") |
|
|
with gr.Row(): |
|
|
with gr.Column(scale=1): |
|
|
playback_participant=gr.Dropdown(choices=analyzer.valid_playback_participants, label="1. Select a Participant") |
|
|
playback_question=gr.Dropdown(choices=[], label="2. Select a Question", interactive=False) |
|
|
gr.Markdown("3. Use the slider to play back fixations one by one.") |
|
|
playback_slider=gr.Slider(minimum=0, maximum=1, step=1, value=0, label="Fixation Number", interactive=False) |
|
|
playback_summary=gr.Markdown(label="Trial Info") |
|
|
with gr.Column(scale=2): |
|
|
playback_plot=gr.Plot(label="Gaze Playback & Live Prediction") |
|
|
|
|
|
|
|
|
outputs_rq2 = [rq2_summary_output, rq2_table_output, rq2_plot_output, rq2_status] |
|
|
outputs_playback = [playback_summary, playback_plot, playback_slider] |
|
|
|
|
|
rq1_metric_dropdown.change(fn=analyzer.analyze_rq1_metric, inputs=rq1_metric_dropdown, outputs=[rq1_plot_output, rq1_summary_output]) |
|
|
|
|
|
train_event = rq2_test_size_slider.release(fn=analyzer.run_prediction_model, inputs=[rq2_test_size_slider, rq2_estimators_slider], outputs=outputs_rq2) |
|
|
rq2_estimators_slider.release(fn=analyzer.run_prediction_model, inputs=[rq2_test_size_slider, rq2_estimators_slider], outputs=outputs_rq2) |
|
|
|
|
|
playback_participant.change( |
|
|
fn=analyzer.update_question_dropdown, |
|
|
inputs=playback_participant, |
|
|
outputs=playback_question |
|
|
) |
|
|
|
|
|
playback_question.change( |
|
|
fn=analyzer.handle_new_trial_selection, |
|
|
inputs=[playback_participant, playback_question], |
|
|
outputs=outputs_playback |
|
|
) |
|
|
|
|
|
playback_slider.release( |
|
|
fn=analyzer.generate_gaze_playback, |
|
|
inputs=[playback_participant, playback_question, playback_slider], |
|
|
outputs=outputs_playback |
|
|
) |
|
|
|
|
|
|
|
|
def initial_load(): |
|
|
|
|
|
rq1_fig, rq1_summary = analyzer.analyze_rq1_metric(analyzer.time_metrics[0] if analyzer.time_metrics else None) |
|
|
|
|
|
|
|
|
model_summary, report_df, feature_fig, status_md = analyzer.run_prediction_model(0.3, 100) |
|
|
|
|
|
|
|
|
return { |
|
|
rq1_plot_output: rq1_fig, |
|
|
rq1_summary_output: rq1_summary, |
|
|
rq2_summary_output: model_summary, |
|
|
rq2_table_output: report_df, |
|
|
rq2_plot_output: feature_fig, |
|
|
rq2_status: status_md |
|
|
} |
|
|
|
|
|
demo.load( |
|
|
fn=initial_load, |
|
|
outputs=[ |
|
|
rq1_plot_output, rq1_summary_output, |
|
|
rq2_summary_output, rq2_table_output, rq2_plot_output, rq2_status |
|
|
] |
|
|
) |
|
|
|
|
|
if __name__ == "__main__": |
|
|
demo.launch() |
