Hereby we publish a python script for automatic analysis of generated MCQS and it can measure and compare the following metrics between sets of MCQs:
– Distractor Similarity, Readability
– Option Length Bias
– Flesch Reading-Ease
– Flesch–Kincaid grade level
#!/usr/bin/env python3
"""
MCQ Quality Analysis GUI (with Inferential Tests for 3x2x2 Design)
– Full statistical toolkit with assumption checks, effect sizes, CI, export,
logging, diagnostic plots, and Monte Carlo simulation.
– Enhanced: Raw data shows key length & mean distractor length; right-click copy;
explanatory legends for all test outputs.
– Enhanced error handling: cell counts, rank deficiency warnings, and diagnostic
messages printed in the GUI and log.
– Added "Flush Data" button to clear all loaded questions, folders, and UI state.
– Added descriptive statistics (mean, SD, N) for each factor and cell in ANOVA report.
– Legend now explains sum_sq, F, PR(>F).
– Statistics tab shows mean ± std for key and distractor lengths.
– Note added about zero FRE values in Statistics tab and Raw Data tab.
– When testing length_bias, ANOVA report also shows descriptive stats for key_length and distractor_length.
– NEW "Error" tab with integrity checks: empty questions, SHA hashes, duplicate IDs, missing keys, and zero FRE_full detection.
"""
import json
import os
import re
import sys
import traceback
import logging
import time
import warnings
import hashlib
from collections import defaultdict
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple, Any
import numpy as np
import pandas as pd
import scipy.stats as stats
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
# Statsmodels for ANOVA and post-hoc
import statsmodels.api as sm
from statsmodels.formula.api import ols
from statsmodels.stats.anova import anova_lm
from statsmodels.stats.multicomp import pairwise_tukeyhsd
from statsmodels.stats.diagnostic import het_breuschpagan
# Optional: textstat for validation
try:
import textstat
HAS_TEXTSTAT = True
except ImportError:
HAS_TEXTSTAT = False
# Force matplotlib to use TkAgg backend before importing pyplot
import matplotlib
matplotlib.use('TkAgg')
import matplotlib.pyplot as plt
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
import tkinter as tk
from tkinter import ttk, filedialog, messagebox, scrolledtext
# -----------------------------------------------------------------------------
# Logging setup
# -----------------------------------------------------------------------------
LOG_FILE = "metrics.log"
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(LOG_FILE, encoding='utf-8'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger("MCQ_Analysis")
# =============================================================================
# 1. Data Models and Parsing
# =============================================================================
@dataclass
class MCQQuestion:
"""Represents a single MCQ question with its components."""
question_id: str
stem: str
options: List[str]
key: str
condition: str
pdf_id: Optional[str] = None
domain: Optional[str] = None
pipeline: Optional[str] = None
prompt_type: Optional[str] = None
fre_stem: Optional[float] = None
fre_full: Optional[float] = None
fkgl_stem: Optional[float] = None
fkgl_full: Optional[float] = None
distractor_similarities: Optional[List[float]] = None
avg_distractor_similarity: Optional[float] = None
option_lengths: Optional[List[int]] = None
key_index: Optional[int] = None
def __post_init__(self):
if self.key and self.key.isalpha():
self.key_index = ord(self.key.upper()) - ord('A')
else:
self.key_index = -1
def parse_metadata_from_condition(condition: str) -> Dict[str, str]:
"""
Parse condition strings like 'P1sph' into pipeline, prompt_type, domain.
Expected patterns: P1sph, P2dph, P3sIM, etc.
"""
result = {'pipeline': 'Unknown', 'prompt_type': 'Unknown', 'domain': 'Unknown'}
if not condition:
return result
# Pipeline: P1, P2, P3
pipe_match = re.match(r'(P[123])', condition)
if pipe_match:
result['pipeline'] = pipe_match.group(1)
# Prompt: s or d
prompt_match = re.search(r'([sd])(?=ph|IM|im)', condition)
if prompt_match:
p = prompt_match.group(1)
result['prompt_type'] = 'Simple' if p.lower() == 's' else 'Detailed'
# Domain: ph or IM
domain_match = re.search(r'(ph|IM|im)$', condition)
if domain_match:
d = domain_match.group(1)
result['domain'] = 'Pharmacology' if d.lower() == 'ph' else 'InternalMedicine'
return result
def parse_generated_text(text: str, condition: str, pdf_id: Optional[str] = None) -> List[MCQQuestion]:
"""Parse raw generated text into MCQQuestion objects."""
questions = []
pattern = r'(?i)Question\s*(\d+)\s*:\s*(.*?)(?=\n\s*Question\s*\d+\s*:|\Z)'
blocks = re.findall(pattern, text, re.DOTALL)
if not blocks:
return questions
for num, block in blocks:
num = int(num.strip())
# Find all options
options = []
opt_re = re.compile(r'(?i)(?:\(([A-E])\)|([A-E])\.)\s*(.*?)(?=\s*(?:\([A-E]\)|[A-E]\.|Answer:|$))', re.DOTALL)
opt_matches = list(opt_re.finditer(block))
if len(opt_matches) != 5:
continue
opt_dict = {}
for m in opt_matches:
letter = m.group(1) or m.group(2)
text_opt = m.group(3).strip()
opt_dict[letter.upper()] = text_opt
options = [opt_dict.get(ch, "") for ch in "ABCDE"]
if any(not opt for opt in options):
continue
answer_re = re.search(r'(?i)Answer\s*:\s*([A-E])', block)
if not answer_re:
continue
key = answer_re.group(1).upper()
first_opt_match = re.search(r'(?i)\s*(?:\(A\)|A\.)\s*', block)
if first_opt_match:
stem_text = block[:first_opt_match.start()].strip()
else:
stem_text = block.split("Answer:")[0].strip()
stem_text = re.sub(r'(?i)Question\s*\d+\s*:\s*', '', stem_text).strip()
# Build unique ID including pdf_id if available
pdf_part = pdf_id if pdf_id else "unknown"
qid = f"{condition}_{pdf_part}_{num:03d}"
q = MCQQuestion(
question_id=qid,
stem=stem_text,
options=options,
key=key,
condition=condition,
pdf_id=pdf_id,
)
# Parse metadata
meta = parse_metadata_from_condition(condition)
q.pipeline = meta['pipeline']
q.prompt_type = meta['prompt_type']
q.domain = meta['domain']
questions.append(q)
return questions
def load_questions_from_json(json_file: str) -> List[MCQQuestion]:
"""Load questions from a JSON file."""
with open(json_file, 'r', encoding='utf-8') as f:
data = json.load(f)
questions = []
if isinstance(data, dict) and 'questions' in data:
data = data['questions']
if not isinstance(data, list):
raise ValueError("JSON must contain a list of question objects.")
for item in data:
qid = item.get('id') or item.get('question_id') or f"Q_{len(questions):03d}"
stem = item.get('stem', '')
options = item.get('options', [])
if len(options) != 5:
continue
key = item.get('key', '').upper()
if key not in 'ABCDE':
key = ''
condition = item.get('condition', 'unknown')
pdf_id = item.get('pdf_id', None)
q = MCQQuestion(
question_id=qid,
stem=stem,
options=options,
key=key,
condition=condition,
pdf_id=pdf_id,
)
# Parse metadata
meta = parse_metadata_from_condition(condition)
q.pipeline = meta['pipeline']
q.prompt_type = meta['prompt_type']
q.domain = meta['domain']
# Allow manual override if provided in JSON
q.domain = item.get('domain', q.domain)
q.pipeline = item.get('pipeline', q.pipeline)
q.prompt_type = item.get('prompt_type', q.prompt_type)
questions.append(q)
return questions
# =============================================================================
# 2. Metric Computation (with logging and optional textstat validation)
# =============================================================================
class MetricsCalculator:
"""Compute various quality metrics for MCQ questions."""
def __init__(self, embedding_model_name='all-mpnet-base-v2'):
logger.info("Loading Sentence-BERT model...")
self.embedding_model = SentenceTransformer(embedding_model_name)
logger.info("Model loaded.")
self._syllable_pattern = re.compile(r'[aeiouy]+', re.IGNORECASE)
def _count_syllables(self, word: str) -> int:
word = word.lower()
if word.endswith('e'):
word = word[:-1]
matches = self._syllable_pattern.findall(word)
count = len(matches)
if word.endswith('ed') and count > 1:
count -= 1
if word.endswith('es') and count > 1:
count -= 1
return max(1, count)
def _text_stats(self, text: str) -> Tuple[int, int, int]:
if not text:
return 0, 0, 0
sentences = re.split(r'[.!?]+', text)
sentences = [s for s in sentences if s.strip()]
num_sentences = len(sentences)
words = re.findall(r'\b\w+\b', text)
num_words = len(words)
syllables = sum(self._count_syllables(w) for w in words)
return num_words, num_sentences, syllables
def flesch_reading_ease(self, text: str) -> float:
words, sentences, syllables = self._text_stats(text)
if words == 0:
return 0.0
if sentences == 0:
sentences = 1
fre = 206.835 - 1.015 * (words / sentences) - 84.6 * (syllables / words)
return max(0.0, min(100.0, fre))
def flesch_kincaid_grade(self, text: str) -> float:
words, sentences, syllables = self._text_stats(text)
if words == 0:
return 0.0
if sentences == 0:
sentences = 1
fkgl = 0.39 * (words / sentences) + 11.8 * (syllables / words) - 15.59
return max(0.0, fkgl)
def compute_question_metrics(self, q: MCQQuestion) -> None:
logger.info(f"Computing metrics for question {q.question_id}")
q.fre_stem = self.flesch_reading_ease(q.stem)
q.fkgl_stem = self.flesch_kincaid_grade(q.stem)
full_text = q.stem + " " + " ".join(q.options)
q.fre_full = self.flesch_reading_ease(full_text)
q.fkgl_full = self.flesch_kincaid_grade(full_text)
# Optional validation with textstat
if HAS_TEXTSTAT:
ts_fre_stem = textstat.flesch_reading_ease(q.stem)
ts_fkgl_stem = textstat.flesch_kincaid_grade(q.stem)
if abs(q.fre_stem - ts_fre_stem) > 1.0:
logger.warning(f"FRE stem mismatch for {q.question_id}: {q.fre_stem:.2f} vs textstat {ts_fre_stem:.2f}")
if abs(q.fkgl_stem - ts_fkgl_stem) > 0.5:
logger.warning(f"FKGL stem mismatch for {q.question_id}: {q.fkgl_stem:.2f} vs textstat {ts_fkgl_stem:.2f}")
if not q.key or q.key_index < 0:
q.distractor_similarities = []
q.avg_distractor_similarity = None
else:
option_embeddings = self.embedding_model.encode(q.options, convert_to_numpy=True)
key_embedding = option_embeddings[q.key_index]
similarities = []
for i, emb in enumerate(option_embeddings):
if i == q.key_index:
continue
sim = cosine_similarity([key_embedding], [emb])[0][0]
similarities.append(sim)
q.distractor_similarities = similarities
q.avg_distractor_similarity = np.mean(similarities) if similarities else None
q.option_lengths = [len(opt) for opt in q.options]
logger.info(f"Finished metrics for {q.question_id}")
# =============================================================================
# 3. Analysis Engine (ENHANCED with export, diagnostic plots, Monte Carlo, and integrity checks)
# =============================================================================
class AnalysisEngine:
def __init__(self):
self.questions: List[MCQQuestion] = []
self.conditions: List[str] = []
self.df: Optional[pd.DataFrame] = None
self.metrics_calc = MetricsCalculator()
self._last_model = None # store model for residual checks
# Integrity checks
self.loaded_files: List[str] = [] # list of file paths loaded
self.pre_parsed_combined_hash: Optional[str] = None # SHA of all stems+options from raw JSONs combined
self.parsed_combined_hash: Optional[str] = None # SHA of all stems+options from parsed questions combined
def _compute_combined_pre_parsed_hash(self) -> str:
"""
Read all loaded JSON files, extract stems and options, sort globally by (filename, question_id),
concatenate, and return SHA-256.
"""
if not self.loaded_files:
return ""
text_parts = []
for filepath in sorted(self.loaded_files):
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
if isinstance(data, dict) and 'questions' in data:
data = data['questions']
if not isinstance(data, list):
continue
# Extract (question_id, stem, options) for each question
items = []
for item in data:
qid = item.get('id') or item.get('question_id') or "unknown"
stem = item.get('stem', '')
opts = item.get('options', [])
while len(opts) < 5:
opts.append('')
items.append((qid, stem, opts))
# Sort by qid within this file to be deterministic
items.sort(key=lambda x: x[0])
for qid, stem, opts in items:
text_parts.append(stem)
text_parts.extend(opts)
combined = "||".join(text_parts)
return hashlib.sha256(combined.encode('utf-8')).hexdigest() if combined else ""
def _compute_combined_parsed_hash(self) -> str:
"""Compute SHA from all questions' stems and options (after parsing)."""
if not self.questions:
return ""
# Sort by (condition, question_id) to ensure deterministic order across files
sorted_q = sorted(self.questions, key=lambda q: (q.condition, q.question_id))
text_parts = []
for q in sorted_q:
text_parts.append(q.stem)
text_parts.extend(q.options)
combined = "||".join(text_parts)
return hashlib.sha256(combined.encode('utf-8')).hexdigest() if combined else ""
def get_empty_questions(self) -> List[Dict[str, Any]]:
"""Return list of questions with empty stem or any empty option."""
issues = []
for q in self.questions:
empty_fields = []
if not q.stem.strip():
empty_fields.append("stem")
for i, opt in enumerate(q.options):
if not opt.strip():
empty_fields.append(f"option_{chr(ord('A')+i)}")
if empty_fields:
issues.append({
'question_id': q.question_id,
'empty_fields': empty_fields,
'condition': q.condition
})
return issues
def get_duplicate_ids(self) -> List[str]:
"""Return list of duplicate question IDs."""
ids = [q.question_id for q in self.questions]
seen = set()
dupes = set()
for qid in ids:
if qid in seen:
dupes.add(qid)
seen.add(qid)
return sorted(dupes)
def get_missing_keys(self) -> List[str]:
"""Return list of question IDs with no valid key (A-E)."""
return [q.question_id for q in self.questions if q.key_index < 0]
def get_zero_fre_full_questions(self) -> List[MCQQuestion]:
"""Return questions where fre_full == 0.0 (with full content)."""
return [q for q in self.questions if q.fre_full is not None and q.fre_full == 0.0]
def load_questions_from_json(self, json_path: str) -> int:
self.loaded_files.append(json_path)
questions = load_questions_from_json(json_path)
if not questions:
return 0
for q in questions:
self.metrics_calc.compute_question_metrics(q)
self.questions.extend(questions)
conds = set(q.condition for q in self.questions)
self.conditions = sorted(conds)
self._build_dataframe()
# Update combined hashes after loading
self.pre_parsed_combined_hash = self._compute_combined_pre_parsed_hash()
self.parsed_combined_hash = self._compute_combined_parsed_hash()
return len(questions)
def load_questions_from_directory(self, dir_path: str) -> int:
total = 0
for filename in os.listdir(dir_path):
if filename.endswith('.json'):
filepath = os.path.join(dir_path, filename)
self.loaded_files.append(filepath)
condition = os.path.splitext(filename)[0]
questions = load_questions_from_json(filepath)
for q in questions:
q.condition = condition
meta = parse_metadata_from_condition(condition)
q.pipeline = meta['pipeline']
q.prompt_type = meta['prompt_type']
q.domain = meta['domain']
if q.pdf_id is None:
q.pdf_id = filename
for q in questions:
self.metrics_calc.compute_question_metrics(q)
self.questions.extend(questions)
total += len(questions)
conds = set(q.condition for q in self.questions)
self.conditions = sorted(conds)
self._build_dataframe()
# Update combined hashes after loading all files
self.pre_parsed_combined_hash = self._compute_combined_pre_parsed_hash()
self.parsed_combined_hash = self._compute_combined_parsed_hash()
return total
def _build_dataframe(self):
data = []
for q in self.questions:
length_bias = None
key_len = None
mean_distractor_len = None
distractor_lens = []
if q.option_lengths and q.key_index >= 0:
key_len = q.option_lengths[q.key_index]
distractor_lens = [q.option_lengths[i] for i in range(5) if i != q.key_index]
if distractor_lens:
mean_distractor_len = np.mean(distractor_lens)
length_bias = np.mean(distractor_lens) - key_len
row = {
'question_id': q.question_id,
'pdf_id': q.pdf_id,
'condition': q.condition,
'pipeline': q.pipeline,
'prompt_type': q.prompt_type,
'domain': q.domain,
'stem': q.stem,
'key': q.key,
'key_index': q.key_index,
'fre_stem': q.fre_stem,
'fre_full': q.fre_full,
'fkgl_stem': q.fkgl_stem,
'fkgl_full': q.fkgl_full,
'avg_distractor_sim': q.avg_distractor_similarity,
'option_lengths': q.option_lengths,
'key_length': key_len,
'mean_distractor_length': mean_distractor_len,
'distractor_lengths': distractor_lens,
'length_bias': length_bias,
}
data.append(row)
self.df = pd.DataFrame(data)
if 'condition' in self.df.columns:
self.df['group_12'] = self.df['condition']
# ========== EXPORT ==========
def export_dataframe(self, filepath: str, format: str = 'csv') -> None:
"""Export the full dataframe (all questions + metrics) to a file."""
if self.df is None or self.df.empty:
raise ValueError("No data to export.")
# For Excel, we need to handle list columns by converting to strings
df_export = self.df.copy()
for col in ['option_lengths', 'distractor_lengths']:
if col in df_export.columns:
df_export[col] = df_export[col].apply(lambda x: str(x) if isinstance(x, list) else x)
if format.lower() == 'csv':
df_export.to_csv(filepath, index=False, encoding='utf-8-sig')
elif format.lower() in ('xlsx', 'excel'):
df_export.to_excel(filepath, index=False, engine='openpyxl')
else:
raise ValueError("Unsupported format. Use 'csv' or 'xlsx'.")
logger.info(f"Exported data to {filepath}")
# ========== DESCRIPTIVE / COMPARISON METHODS ==========
def get_condition_stats(self, condition: str) -> Dict[str, Any]:
if self.df is None:
return {}
cond_df = self.df[self.df['condition'] == condition]
if cond_df.empty:
return {}
stats_dict = {}
for metric in ['fre_stem', 'fre_full', 'fkgl_stem', 'fkgl_full']:
vals = cond_df[metric].dropna()
if not vals.empty:
mean = vals.mean()
sem = vals.sem()
ci_low, ci_high = stats.t.interval(0.95, len(vals)-1, loc=mean, scale=sem)
stats_dict[metric] = {
'mean': mean,
'std': vals.std(),
'count': len(vals),
'min': vals.min(),
'max': vals.max(),
'ci_low': ci_low,
'ci_high': ci_high,
}
sim_vals = cond_df['avg_distractor_sim'].dropna()
if not sim_vals.empty:
mean = sim_vals.mean()
sem = sim_vals.sem()
ci_low, ci_high = stats.t.interval(0.95, len(sim_vals)-1, loc=mean, scale=sem)
stats_dict['avg_distractor_sim'] = {
'mean': mean,
'std': sim_vals.std(),
'count': len(sim_vals),
'min': sim_vals.min(),
'max': sim_vals.max(),
'ci_low': ci_low,
'ci_high': ci_high,
}
key_counts = cond_df['key_index'].value_counts().sort_index()
key_dist = {chr(ord('A') + i): int(key_counts.get(i, 0)) for i in range(5)}
stats_dict['key_distribution'] = key_dist
stats_dict['key_total'] = len(cond_df)
key_lengths = cond_df['key_length'].dropna()
distractor_lengths = cond_df['distractor_lengths'].dropna()
all_distractor_lens = [l for sublist in distractor_lengths for l in sublist]
if not key_lengths.empty and all_distractor_lens:
stats_dict['key_length_mean'] = key_lengths.mean()
stats_dict['key_length_std'] = key_lengths.std()
stats_dict['distractor_length_mean'] = np.mean(all_distractor_lens)
stats_dict['distractor_length_std'] = np.std(all_distractor_lens)
stats_dict['length_bias'] = stats_dict['distractor_length_mean'] - stats_dict['key_length_mean']
else:
stats_dict['length_bias'] = None
return stats_dict
def compare_conditions(self, condition1: str, condition2: str, metric: str) -> Dict[str, Any]:
if self.df is None:
return {}
if condition1 not in self.conditions or condition2 not in self.conditions:
return {}
df1 = self.df[self.df['condition'] == condition1][metric].dropna()
df2 = self.df[self.df['condition'] == condition2][metric].dropna()
if df1.empty or df2.empty:
return {}
t_stat, p_val = stats.ttest_ind(df1, df2, equal_var=False)
return {
'condition1': condition1,
'condition2': condition2,
'metric': metric,
'mean1': df1.mean(),
'mean2': df2.mean(),
't_stat': t_stat,
'p_value': p_val,
'n1': len(df1),
'n2': len(df2)
}
def compare_groups(self, group1_conditions: List[str], group2_conditions: List[str]) -> Dict[str, Any]:
if self.df is None:
return {}
group1_df = self.df[self.df['condition'].isin(group1_conditions)]
group2_df = self.df[self.df['condition'].isin(group2_conditions)]
if group1_df.empty or group2_df.empty:
return {}
results = {}
numeric_metrics = ['fre_stem', 'fre_full', 'fkgl_stem', 'fkgl_full', 'avg_distractor_sim']
for metric in numeric_metrics:
vals1 = group1_df[metric].dropna()
vals2 = group2_df[metric].dropna()
if vals1.empty or vals2.empty:
continue
t_stat, p_val = stats.ttest_ind(vals1, vals2, equal_var=False)
n1, n2 = len(vals1), len(vals2)
var1, var2 = vals1.var(), vals2.var()
pooled_std = np.sqrt(((n1 - 1) * var1 + (n2 - 1) * var2) / (n1 + n2 - 2)) if (n1 + n2) > 2 else np.nan
cohen_d = (vals1.mean() - vals2.mean()) / pooled_std if pooled_std and pooled_std != 0 else np.nan
results[metric] = {
'mean1': vals1.mean(),
'mean2': vals2.mean(),
't_stat': t_stat,
'p_value': p_val,
'n1': n1,
'n2': n2,
'cohen_d': cohen_d
}
return results
# ========== INFERENTIAL TEST METHODS ==========
def get_available_dvs(self) -> List[str]:
if self.df is None:
return []
candidates = ['fre_stem', 'fre_full', 'fkgl_stem', 'fkgl_full', 'avg_distractor_sim', 'length_bias']
return [c for c in candidates if c in self.df.columns and self.df[c].notna().any()]
def check_assumptions(self, dv: str) -> Dict[str, Any]:
if self.df is None:
return {'error': 'No data loaded'}
df_clean = self.df[[dv, 'pipeline', 'prompt_type', 'domain']].dropna()
if df_clean.empty:
return {'error': f'No valid data for DV: {dv}'}
df_clean['pipeline'] = df_clean['pipeline'].astype('category')
df_clean['prompt_type'] = df_clean['prompt_type'].astype('category')
df_clean['domain'] = df_clean['domain'].astype('category')
model = ols(f'{dv} ~ C(pipeline) * C(prompt_type) * C(domain)', data=df_clean).fit()
residuals = model.resid
if len(residuals) >= 3:
shapiro_stat, shapiro_p = stats.shapiro(residuals)
else:
shapiro_stat, shapiro_p = np.nan, np.nan
df_clean['group'] = df_clean['pipeline'].astype(str) + '_' + df_clean['prompt_type'] + '_' + df_clean['domain']
groups = [group[dv].values for name, group in df_clean.groupby('group') if len(group) >= 2]
if len(groups) >= 2:
levene_stat, levene_p = stats.levene(*groups)
else:
levene_stat, levene_p = np.nan, np.nan
self._last_model = model
return {
'residual_normality': {'statistic': shapiro_stat, 'p_value': shapiro_p},
'homogeneity': {'statistic': levene_stat, 'p_value': levene_p},
'n_total': len(df_clean)
}
def run_3way_anova(self, dv: str) -> Dict[str, Any]:
"""
Run 3-way ANOVA with enhanced error handling, cell counts, and rank warnings.
"""
if self.df is None:
return {'error': 'No data loaded'}
df_clean = self.df[[dv, 'pipeline', 'prompt_type', 'domain']].dropna()
if df_clean.empty:
return {'error': f'No valid data for DV: {dv}'}
# ---- Cell count report ----
cell_counts = df_clean.groupby(['pipeline', 'prompt_type', 'domain']).size().reset_index(name='count')
empty_cells = cell_counts[cell_counts['count'] == 0]
if not empty_cells.empty:
logger.warning(f"Empty cells detected for DV {dv}: {empty_cells.to_dict('records')}")
# ---- Fit model with warning capture ----
df_clean['pipeline'] = df_clean['pipeline'].astype('category')
df_clean['prompt_type'] = df_clean['prompt_type'].astype('category')
df_clean['domain'] = df_clean['domain'].astype('category')
rank_warnings = []
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
try:
model = ols(f'{dv} ~ C(pipeline) * C(prompt_type) * C(domain)', data=df_clean).fit()
except Exception as e:
logger.error(f"Model fitting failed: {e}")
return {'error': f'Model fitting failed: {e}'}
# Collect warnings of interest
for warning in w:
if issubclass(warning.category, (UserWarning, RuntimeWarning)):
msg = str(warning.message)
# Filter to rank-related or division warnings
if any(kw in msg.lower() for kw in ['covariance', 'rank', 'divide', 'invalid']):
rank_warnings.append(msg)
# ---- ANOVA table ----
anova_table = anova_lm(model, typ=2)
ss_residual = anova_table.loc['Residual', 'sum_sq'] if 'Residual' in anova_table.index else np.nan
eta_sq_partial = {}
for idx in anova_table.index:
if idx == 'Residual':
continue
ss_effect = anova_table.loc[idx, 'sum_sq']
if ss_residual and (ss_effect + ss_residual) != 0:
eta_sq_partial[idx] = ss_effect / (ss_effect + ss_residual)
else:
eta_sq_partial[idx] = np.nan
self._last_model = model
return {
'anova_table': anova_table,
'model_summary': model.summary(),
'n_total': len(df_clean),
'partial_eta_squared': eta_sq_partial,
'cell_counts': cell_counts,
'empty_cells': empty_cells,
'rank_warnings': rank_warnings,
'data_used': df_clean # include the cleaned data for descriptive stats
}
def run_tukey_12groups(self, dv: str) -> Optional[pd.DataFrame]:
if self.df is None:
return None
df_clean = self.df[[dv, 'group_12']].dropna()
if df_clean.empty or len(df_clean['group_12'].unique()) < 2:
return None
tukey_result = pairwise_tukeyhsd(endog=df_clean[dv], groups=df_clean['group_12'], alpha=0.05)
result_df = pd.DataFrame(data=tukey_result.summary().data[1:], columns=tukey_result.summary().data[0])
return result_df
def run_key_position_chi2(self, factor: str = 'pipeline') -> Dict[str, Any]:
if self.df is None:
return {'error': 'No data'}
df_clean = self.df[[factor, 'key_index']].dropna()
if df_clean.empty:
return {'error': 'No valid key data'}
contingency = pd.crosstab(df_clean[factor], df_clean['key_index'])
chi2, p, dof, expected = stats.chi2_contingency(contingency)
return {
'chi2': chi2,
'p_value': p,
'dof': dof,
'contingency_table': contingency,
'expected': expected,
'significant': p < 0.05
}
def run_key_position_gof(self, condition: str) -> Dict[str, Any]:
if self.df is None:
return {'error': 'No data'}
df_cond = self.df[self.df['condition'] == condition]
if df_cond.empty:
return {'error': 'Condition not found'}
counts = df_cond['key_index'].value_counts().reindex(range(5), fill_value=0).values
if sum(counts) < 5:
return {'error': 'Too few observations'}
chi2, p = stats.chisquare(counts)
return {
'chi2': chi2,
'p_value': p,
'dof': 4,
'counts': counts,
'significant': p < 0.05,
'n': sum(counts)
}
def run_rank_anova(self, dv: str) -> Dict[str, Any]:
if self.df is None:
return {'error': 'No data'}
df_clean = self.df[[dv, 'pipeline', 'prompt_type', 'domain']].dropna()
if df_clean.empty:
return {'error': f'No valid data for DV: {dv}'}
df_clean['rank_dv'] = df_clean[dv].rank()
model = ols('rank_dv ~ C(pipeline) * C(prompt_type) * C(domain)', data=df_clean).fit()
anova_table = anova_lm(model, typ=2)
return {
'method': 'Rank-Based ANOVA (Non-parametric)',
'anova_table': anova_table,
'n_total': len(df_clean)
}
# ----- Diagnostic plots -----
def get_diagnostic_plots(self, dv: str) -> Tuple[plt.Figure, plt.Figure]:
"""Return a figure with QQ plot and residuals vs fitted for the fitted ANOVA model."""
if self.df is None or self._last_model is None:
raise ValueError("No model fitted. Run ANOVA first.")
# Use the stored model (should match the DV)
model = self._last_model
residuals = model.resid
fitted = model.fittedvalues
fig1, ax1 = plt.subplots(figsize=(5, 4))
sm.qqplot(residuals, line='s', ax=ax1)
ax1.set_title('QQ Plot of Residuals')
fig2, ax2 = plt.subplots(figsize=(5, 4))
ax2.scatter(fitted, residuals, alpha=0.6)
ax2.axhline(0, color='red', linestyle='--')
ax2.set_xlabel('Fitted values')
ax2.set_ylabel('Residuals')
ax2.set_title('Residuals vs Fitted')
return fig1, fig2
# ----- Monte Carlo Chi-Square simulation for key positions -----
def monte_carlo_key_chi2(self, factor: str = 'pipeline', n_sim: int = 1000) -> Dict[str, Any]:
"""
Permutation test for key-position independence.
Shuffle key positions within each factor level and compute Chi2.
"""
if self.df is None:
return {'error': 'No data'}
df_clean = self.df[[factor, 'key_index']].dropna()
if df_clean.empty:
return {'error': 'No valid key data'}
observed_chi2, _, _, _ = stats.chi2_contingency(pd.crosstab(df_clean[factor], df_clean['key_index']))
simulated_chi2 = []
# For speed, we can use numpy to shuffle
groups = df_clean.groupby(factor)
for _ in range(n_sim):
shuffled = []
for name, group in groups:
shuffled_keys = np.random.permutation(group['key_index'].values)
shuffled.append(pd.DataFrame({factor: group[factor].values, 'key_index': shuffled_keys}))
shuffled_df = pd.concat(shuffled, ignore_index=True)
cont = pd.crosstab(shuffled_df[factor], shuffled_df['key_index'])
chi2_sim, _, _, _ = stats.chi2_contingency(cont)
simulated_chi2.append(chi2_sim)
p_sim = (np.array(simulated_chi2) >= observed_chi2).mean()
return {
'observed_chi2': observed_chi2,
'simulated_chi2': simulated_chi2,
'p_sim': p_sim,
'n_sim': n_sim,
'significant_sim': p_sim < 0.05
}
# =============================================================================
# 4. GUI Application (updated with export, diagnostic plots, Monte Carlo, Flush, and Error tab)
# =============================================================================
class MCQAnalysisApp:
def __init__(self, root):
self.root = root
self.root.title("MCQ Quality Analysis - Thesis Tool (v5 with Legends & Copy)") # version bump
self.root.geometry("1300x900+50+50")
self.root.lift()
self.root.focus_force()
self.engine = AnalysisEngine()
self.current_condition = None
# Group storage
self.group_a_conditions = []
self.group_b_conditions = []
# Create main frames
self.top_frame = ttk.Frame(root)
self.top_frame.pack(side=tk.TOP, fill=tk.X, padx=10, pady=5)
self.mid_frame = ttk.Frame(root)
self.mid_frame.pack(side=tk.TOP, fill=tk.BOTH, expand=True, padx=10, pady=5)
self.bottom_frame = ttk.Frame(root)
self.bottom_frame.pack(side=tk.BOTTOM, fill=tk.X, padx=10, pady=5)
# Top: Load data, select condition, export, flush
self.load_btn = ttk.Button(self.top_frame, text="Load JSON File", command=self.load_json_file)
self.load_btn.pack(side=tk.LEFT, padx=5)
self.load_dir_btn = ttk.Button(self.top_frame, text="Load Directory (JSONs)", command=self.load_directory)
self.load_dir_btn.pack(side=tk.LEFT, padx=5)
self.export_btn = ttk.Button(self.top_frame, text="Export Raw Data", command=self.export_data)
self.export_btn.pack(side=tk.LEFT, padx=5)
self.flush_btn = ttk.Button(self.top_frame, text="Flush Data", command=self.flush_data)
self.flush_btn.pack(side=tk.LEFT, padx=5)
self.status_label = ttk.Label(self.top_frame, text="No data loaded")
self.status_label.pack(side=tk.LEFT, padx=20)
self.cond_label = ttk.Label(self.top_frame, text="Condition:")
self.cond_label.pack(side=tk.LEFT, padx=(20, 5))
self.cond_var = tk.StringVar()
self.cond_combo = ttk.Combobox(self.top_frame, textvariable=self.cond_var, state="readonly")
self.cond_combo.pack(side=tk.LEFT, padx=5)
self.cond_combo.bind('<<ComboboxSelected>>', self.on_condition_selected)
self.refresh_btn = ttk.Button(self.top_frame, text="Refresh Stats", command=self.update_display)
self.refresh_btn.pack(side=tk.LEFT, padx=10)
# Comparison section in bottom frame (individual conditions)
self.comp_label = ttk.Label(self.bottom_frame, text="Compare selected condition with:")
self.comp_label.pack(side=tk.LEFT, padx=5)
self.compare_cond_var = tk.StringVar()
self.compare_cond_combo = ttk.Combobox(self.bottom_frame, textvariable=self.compare_cond_var, state="readonly")
self.compare_cond_combo.pack(side=tk.LEFT, padx=5)
self.metric_var = tk.StringVar()
self.metric_combo = ttk.Combobox(self.bottom_frame, textvariable=self.metric_var, state="readonly")
self.metric_combo.pack(side=tk.LEFT, padx=5)
self.metric_combo['values'] = self.engine.get_available_dvs()
if self.metric_combo['values']:
self.metric_combo.set(self.metric_combo['values'][0])
self.compare_btn = ttk.Button(self.bottom_frame, text="Run t-test", command=self.run_comparison)
self.compare_btn.pack(side=tk.LEFT, padx=10)
self.comp_result_label = ttk.Label(self.bottom_frame, text="")
self.comp_result_label.pack(side=tk.LEFT, padx=10)
# Mid frame: Notebook for tabs
self.notebook = ttk.Notebook(self.mid_frame)
self.notebook.pack(fill=tk.BOTH, expand=True)
# Tab 1: Summary Statistics
self.stats_frame = ttk.Frame(self.notebook)
self.notebook.add(self.stats_frame, text="Statistics")
self.stats_text = tk.Text(self.stats_frame, wrap=tk.WORD, font=("Courier", 10))
self.stats_text.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
scrollbar = ttk.Scrollbar(self.stats_frame, orient=tk.VERTICAL, command=self.stats_text.yview)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
self.stats_text.config(yscrollcommand=scrollbar.set)
self.add_copy_menu(self.stats_text)
# Tab 2: Plots
self.plot_frame = ttk.Frame(self.notebook)
self.notebook.add(self.plot_frame, text="Plots")
self.figure = plt.Figure(figsize=(5, 4), dpi=100)
self.canvas = FigureCanvasTkAgg(self.figure, master=self.plot_frame)
self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True)
# Tab 3: Raw Data
self.data_frame = ttk.Frame(self.notebook)
self.notebook.add(self.data_frame, text="Raw Data")
self.data_tree = ttk.Treeview(self.data_frame)
self.data_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
scrollbar2 = ttk.Scrollbar(self.data_frame, orient=tk.VERTICAL, command=self.data_tree.yview)
scrollbar2.pack(side=tk.RIGHT, fill=tk.Y)
self.data_tree.config(yscrollcommand=scrollbar2.set)
# Add explanatory note for zero FRE_full values
self.data_note_label = ttk.Label(
self.data_frame,
text="Note: FRE_full = 0 indicates that the combined text (stem + options) contains no alphabetic words. "
"This may happen if options are empty or contain only numbers/symbols. "
"Check the 'option_lengths' column – if they are zero, the options are missing.",
foreground="gray",
wraplength=800
)
self.data_note_label.pack(side=tk.BOTTOM, fill=tk.X, padx=5, pady=5)
# Tab 4: Group Comparison (manual grouping)
self.group_frame = ttk.Frame(self.notebook)
self.notebook.add(self.group_frame, text="Group Comparison")
self.setup_group_comparison_tab()
# Tab 5: Inferential Tests
self.infer_frame = ttk.Frame(self.notebook)
self.notebook.add(self.infer_frame, text="Inferential Tests")
self.setup_inferential_tab()
# Tab 6: Error Checks (NEW)
self.error_frame = ttk.Frame(self.notebook)
self.notebook.add(self.error_frame, text="Error")
self.error_text = scrolledtext.ScrolledText(self.error_frame, wrap=tk.WORD, font=("Courier", 9))
self.error_text.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
self.add_copy_menu(self.error_text)
# Initial status
self.stats_text.insert(tk.END, "Welcome!\nLoad a JSON file or directory to begin.")
# Fill error tab with placeholder
self.error_text.insert(tk.END, "Integrity checks will appear here after data is loaded.\n")
# ========== Helper: Right-click copy menu ==========
def add_copy_menu(self, widget):
"""Add right-click context menu with Copy and Select All to a Text widget."""
menu = tk.Menu(widget, tearoff=0)
menu.add_command(label="Copy", command=lambda: self.copy_text(widget))
menu.add_command(label="Select All", command=lambda: widget.tag_add(tk.SEL, "1.0", tk.END))
menu.add_separator()
menu.add_command(label="Clear Selection", command=lambda: widget.tag_remove(tk.SEL, "1.0", tk.END))
def show_menu(event):
menu.tk_popup(event.x_root, event.y_root)
widget.bind("<Button-3>", show_menu)
def copy_text(self, widget):
try:
selected = widget.get(tk.SEL_FIRST, tk.SEL_LAST)
self.root.clipboard_clear()
self.root.clipboard_append(selected)
except tk.TclError:
# Nothing selected -> copy all
content = widget.get("1.0", tk.END)
self.root.clipboard_clear()
self.root.clipboard_append(content)
# ========== Helper: Statistical legend ==========
def statistical_legend(self) -> str:
return """
┌─────────────────────────────────────────────────────────────────────────────┐
│ LEGEND FOR STATISTICAL OUTPUTS │
├─────────────────────────────────────────────────────────────────────────────┤
│ p-value : Probability of observing the test statistic under the │
│ null hypothesis. p < 0.05 is typically considered │
│ statistically significant. │
│ η²p (eta-squared): Partial effect size – proportion of variance explained │
│ by that factor. 0.01=small, 0.06=medium, 0.14=large. │
│ Cohen's d : Standardised difference between two means. │
│ 0.2=small, 0.5=medium, 0.8=large. │
│ Chi² : Chi-square statistic – tests independence or goodness- │
│ of-fit. Larger values indicate stronger deviation from │
│ expected distribution. │
│ df : Degrees of freedom – number of independent values in │
│ the calculation. │
│ NaN : 'Not a Number' – indicates missing or undefined value, │
│ often because the test could not be computed (e.g., │
│ insufficient data, division by zero). │
│ Significant : Usually flagged when p < 0.05 (or simulated p < 0.05). │
│ sum_sq : Sum of squares – the total variability attributable to │
│ that factor (or residual). │
│ F : F-statistic – ratio of mean square of the factor to the │
│ mean square error. Larger values indicate stronger │
│ effect relative to error. │
│ PR(>F) : The p-value associated with the F-statistic. │
│ Small values (<0.05) suggest a significant effect. │
│ Interpretation : Always consider the context and effect sizes alongside │
│ p-values to avoid over-reliance on significance. │
└─────────────────────────────────────────────────────────────────────────────┘
"""
def append_legend(self, text_widget):
"""Append the statistical legend to the given text widget."""
text_widget.insert(tk.END, "\n\n" + self.statistical_legend())
# ========== Group Comparison Tab Setup ==========
def setup_group_comparison_tab(self):
parent = self.group_frame
left_panel = ttk.Frame(parent, width=300)
left_panel.pack(side=tk.LEFT, fill=tk.BOTH, expand=False, padx=5, pady=5)
right_panel = ttk.Frame(parent)
right_panel.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=5, pady=5)
ttk.Label(left_panel, text="Available Conditions", font=('Arial', 10, 'bold')).pack(anchor=tk.W)
self.avail_listbox = tk.Listbox(left_panel, selectmode=tk.EXTENDED, height=10)
self.avail_listbox.pack(fill=tk.BOTH, expand=True, pady=5)
btn_frame = ttk.Frame(left_panel)
btn_frame.pack(fill=tk.X, pady=5)
ttk.Button(btn_frame, text="Add to Group A", command=self.add_to_group_a).pack(side=tk.LEFT, padx=2)
ttk.Button(btn_frame, text="Add to Group B", command=self.add_to_group_b).pack(side=tk.LEFT, padx=2)
group_display_frame = ttk.Frame(left_panel)
group_display_frame.pack(fill=tk.BOTH, expand=True, pady=5)
ga_frame = ttk.Frame(group_display_frame)
ga_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=2)
ttk.Label(ga_frame, text="Group A", font=('Arial', 9, 'bold')).pack(anchor=tk.W)
self.group_a_listbox = tk.Listbox(ga_frame, selectmode=tk.SINGLE, height=5)
self.group_a_listbox.pack(fill=tk.BOTH, expand=True)
ga_btn_frame = ttk.Frame(ga_frame)
ga_btn_frame.pack(fill=tk.X, pady=2)
ttk.Button(ga_btn_frame, text="Remove", command=self.remove_from_group_a).pack(side=tk.LEFT, padx=2)
ttk.Button(ga_btn_frame, text="Clear", command=self.clear_group_a).pack(side=tk.LEFT, padx=2)
gb_frame = ttk.Frame(group_display_frame)
gb_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=2)
ttk.Label(gb_frame, text="Group B", font=('Arial', 9, 'bold')).pack(anchor=tk.W)
self.group_b_listbox = tk.Listbox(gb_frame, selectmode=tk.SINGLE, height=5)
self.group_b_listbox.pack(fill=tk.BOTH, expand=True)
gb_btn_frame = ttk.Frame(gb_frame)
gb_btn_frame.pack(fill=tk.X, pady=2)
ttk.Button(gb_btn_frame, text="Remove", command=self.remove_from_group_b).pack(side=tk.LEFT, padx=2)
ttk.Button(gb_btn_frame, text="Clear", command=self.clear_group_b).pack(side=tk.LEFT, padx=2)
self.compare_groups_btn = ttk.Button(left_panel, text="Compare Groups (t-test)", command=self.compare_groups)
self.compare_groups_btn.pack(pady=10)
ttk.Label(right_panel, text="Comparison Results", font=('Arial', 10, 'bold')).pack(anchor=tk.W)
columns = ('Metric', 'Mean_A', 'Mean_B', 't', 'p', 'Sig', "Cohen's d", 'n_A', 'n_B')
self.group_result_tree = ttk.Treeview(right_panel, columns=columns, show='headings', height=12)
for col in columns:
self.group_result_tree.heading(col, text=col)
self.group_result_tree.column(col, width=80, anchor=tk.CENTER)
self.group_result_tree.column('Metric', width=120, anchor=tk.W)
self.group_result_tree.pack(fill=tk.BOTH, expand=True, pady=5)
self.group_result_info = tk.Text(right_panel, height=4, wrap=tk.WORD, font=("Courier", 9))
self.group_result_info.pack(fill=tk.X, pady=5)
self.add_copy_menu(self.group_result_info)
# ========== Inferential Tests Tab Setup ==========
def setup_inferential_tab(self):
parent = self.infer_frame
ctrl_frame = ttk.Frame(parent)
ctrl_frame.pack(fill=tk.X, padx=10, pady=5)
ttk.Label(ctrl_frame, text="Select DV:").pack(side=tk.LEFT, padx=5)
self.infer_dv_var = tk.StringVar()
self.infer_dv_combo = ttk.Combobox(ctrl_frame, textvariable=self.infer_dv_var, state="readonly", width=20)
self.infer_dv_combo.pack(side=tk.LEFT, padx=5)
ttk.Button(ctrl_frame, text="Refresh DVs", command=self.refresh_infer_dvs).pack(side=tk.LEFT, padx=5)
btn_panel = ttk.Frame(parent)
btn_panel.pack(fill=tk.X, padx=10, pady=5)
ttk.Button(btn_panel, text="1. Check Assumptions (residuals)", command=self.run_assumptions).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_panel, text="2. Run 3-Way ANOVA", command=self.run_anova).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_panel, text="3. Run Tukey HSD (12 groups)", command=self.run_tukey).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_panel, text="4. Test Key-Position Bias (Chi2)", command=self.run_key_chi2).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_panel, text="5. Rank-Based ANOVA (Non-param)", command=self.run_rank_anova).pack(side=tk.LEFT, padx=5)
# NEW buttons for diagnostic plots and Monte Carlo
ttk.Button(btn_panel, text="6. Diagnostic Plots (QQ, Resid)", command=self.run_diagnostic_plots).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_panel, text="7. Monte Carlo Chi-Square", command=self.run_monte_carlo_chi2).pack(side=tk.LEFT, padx=5)
self.infer_result_text = scrolledtext.ScrolledText(parent, wrap=tk.WORD, font=("Courier", 9))
self.infer_result_text.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
self.add_copy_menu(self.infer_result_text)
self.infer_result_text.insert(tk.END, "Inferential Tests will appear here.\nLoad data and select a DV first.\n")
def refresh_infer_dvs(self):
dvs = self.engine.get_available_dvs()
self.infer_dv_combo['values'] = dvs
if dvs:
self.infer_dv_combo.set(dvs[0])
def get_selected_dv(self) -> Optional[str]:
dv = self.infer_dv_var.get()
if not dv or dv not in self.engine.get_available_dvs():
messagebox.showwarning("DV Missing", "Please select a valid Dependent Variable (DV).")
return None
return dv
# --- Assumptions check ---
def run_assumptions(self):
dv = self.get_selected_dv()
if not dv:
return
result = self.engine.check_assumptions(dv)
self.infer_result_text.delete(1.0, tk.END)
if 'error' in result:
self.infer_result_text.insert(tk.END, f"Error: {result['error']}")
self.append_legend(self.infer_result_text)
return
self.infer_result_text.insert(tk.END, f"Assumption Checks for DV: {dv}\n")
self.infer_result_text.insert(tk.END, "=" * 60 + "\n")
norm = result.get('residual_normality', {})
self.infer_result_text.insert(tk.END, f"Shapiro-Wilk test on residuals:\n")
self.infer_result_text.insert(tk.END, f" W = {norm.get('statistic', np.nan):.4f}, p = {norm.get('p_value', np.nan):.4f}\n")
if norm.get('p_value', 1) < 0.05:
self.infer_result_text.insert(tk.END, " → Residuals deviate from normality (p < 0.05). Consider rank-based ANOVA.\n")
else:
self.infer_result_text.insert(tk.END, " → Residuals are approximately normal (p >= 0.05).\n")
hom = result.get('homogeneity', {})
self.infer_result_text.insert(tk.END, f"\nLevene's test for homogeneity of variance:\n")
self.infer_result_text.insert(tk.END, f" W = {hom.get('statistic', np.nan):.4f}, p = {hom.get('p_value', np.nan):.4f}\n")
if hom.get('p_value', 1) < 0.05:
self.infer_result_text.insert(tk.END, " → Variance across groups is significantly different (p < 0.05).\n")
else:
self.infer_result_text.insert(tk.END, " → Variance is homogeneous across groups (p >= 0.05).\n")
self.infer_result_text.insert(tk.END, f"\nN_total = {result.get('n_total', 0)}\n")
self.append_legend(self.infer_result_text)
# --- ANOVA with effect sizes and diagnostics ---
def run_anova(self):
dv = self.get_selected_dv()
if not dv:
return
result = self.engine.run_3way_anova(dv)
self.infer_result_text.delete(1.0, tk.END)
if 'error' in result:
self.infer_result_text.insert(tk.END, f"Error: {result['error']}")
self.append_legend(self.infer_result_text)
return
# ---- Added: Descriptive statistics for each factor and cell ----
# Use the cleaned data from the result
df_clean = result.get('data_used')
if df_clean is not None and not df_clean.empty:
self.infer_result_text.insert(tk.END, "DESCRIPTIVE STATISTICS\n")
self.infer_result_text.insert(tk.END, "=" * 80 + "\n")
# Overall mean and SD for the DV
overall_mean = df_clean[dv].mean()
overall_std = df_clean[dv].std()
overall_n = len(df_clean)
self.infer_result_text.insert(tk.END, f"Overall (DV = {dv}): Mean = {overall_mean:.3f}, SD = {overall_std:.3f}, N = {overall_n}\n\n")
# Marginal means for each factor for the DV
for factor in ['pipeline', 'prompt_type', 'domain']:
desc = df_clean.groupby(factor)[dv].agg(['mean', 'std', 'count']).round(3)
self.infer_result_text.insert(tk.END, f"Marginal means for {factor} (DV = {dv}):\n")
self.infer_result_text.insert(tk.END, desc.to_string() + "\n\n")
# Cell means for the DV
cell_desc = df_clean.groupby(['pipeline', 'prompt_type', 'domain'])[dv].agg(['mean', 'std', 'count']).round(3)
self.infer_result_text.insert(tk.END, f"Cell means (pipeline × prompt_type × domain) for DV = {dv}:\n")
self.infer_result_text.insert(tk.END, cell_desc.to_string() + "\n")
# ---- If DV is length_bias, also show key_length and mean_distractor_length stats ----
if dv == 'length_bias':
# Get the indices of df_clean to pull the corresponding rows from self.engine.df
# df_clean contains the rows with non-null length_bias
# We'll add key_length and mean_distractor_length to the same descriptive tables
# Re-merge the original dataframe for those indices
orig_df = self.engine.df
# Get the rows used in df_clean by index
indices = df_clean.index
# Create a subset of orig_df with those indices
sub_df = orig_df.loc[indices]
# Ensure we have key_length and mean_distractor_length
if 'key_length' in sub_df.columns and 'mean_distractor_length' in sub_df.columns:
# Marginal means for each factor for key_length
self.infer_result_text.insert(tk.END, "\n--- Additional stats for length components ---\n")
for factor in ['pipeline', 'prompt_type', 'domain']:
# key_length
key_desc = sub_df.groupby(factor)['key_length'].agg(['mean', 'std', 'count']).round(3)
self.infer_result_text.insert(tk.END, f"Marginal means for {factor} (KEY LENGTH):\n")
self.infer_result_text.insert(tk.END, key_desc.to_string() + "\n")
# mean_distractor_length
dist_desc = sub_df.groupby(factor)['mean_distractor_length'].agg(['mean', 'std', 'count']).round(3)
self.infer_result_text.insert(tk.END, f"Marginal means for {factor} (DISTRACTOR LENGTH):\n")
self.infer_result_text.insert(tk.END, dist_desc.to_string() + "\n")
# Cell means for key_length and mean_distractor_length
cell_key = sub_df.groupby(['pipeline', 'prompt_type', 'domain'])['key_length'].agg(['mean', 'std', 'count']).round(3)
self.infer_result_text.insert(tk.END, "\nCell means for KEY LENGTH:\n")
self.infer_result_text.insert(tk.END, cell_key.to_string() + "\n")
cell_dist = sub_df.groupby(['pipeline', 'prompt_type', 'domain'])['mean_distractor_length'].agg(['mean', 'std', 'count']).round(3)
self.infer_result_text.insert(tk.END, "\nCell means for DISTRACTOR LENGTH:\n")
self.infer_result_text.insert(tk.END, cell_dist.to_string() + "\n")
self.infer_result_text.insert(tk.END, "\n" + "=" * 80 + "\n\n")
# ---- Display cell counts ----
cell_counts = result.get('cell_counts')
if cell_counts is not None and not cell_counts.empty:
self.infer_result_text.insert(tk.END, "Cell counts (pipeline × prompt_type × domain):\n")
self.infer_result_text.insert(tk.END, cell_counts.to_string(index=False) + "\n\n")
empty_cells = result.get('empty_cells')
if empty_cells is not None and not empty_cells.empty:
self.infer_result_text.insert(tk.END, "⚠ Empty cells detected! The ANOVA may not be reliable.\n")
self.infer_result_text.insert(tk.END, "Consider collecting more data or simplifying the model.\n\n")
# Show warnings
if result.get('rank_warnings'):
self.infer_result_text.insert(tk.END, "⚠ Warnings during model fitting (rank deficiency / division issues):\n")
for warn in result['rank_warnings']:
self.infer_result_text.insert(tk.END, f" {warn}\n")
self.infer_result_text.insert(tk.END, "\nThese indicate that some combinations of factors are missing or the model is over-specified.\n")
self.infer_result_text.insert(tk.END, "Interpret results with caution. Consider using the rank-based ANOVA as an alternative.\n\n")
# ---- ANOVA table ----
anova_df = result['anova_table']
eta = result.get('partial_eta_squared', {})
self.infer_result_text.insert(tk.END, f"3-Way Factorial ANOVA (Type II) for DV: {dv}\n")
self.infer_result_text.insert(tk.END, f"N_total = {result['n_total']}\n")
self.infer_result_text.insert(tk.END, "=" * 80 + "\n")
anova_df_with_eta = anova_df.copy()
anova_df_with_eta['η²p'] = [eta.get(idx, np.nan) for idx in anova_df.index]
self.infer_result_text.insert(tk.END, anova_df_with_eta.to_string() + "\n")
self.infer_result_text.insert(tk.END, "\n" + "=" * 80 + "\n")
self.infer_result_text.insert(tk.END, "Significant effects (p < 0.05):\n")
sig_effects = anova_df[anova_df['PR(>F)'] < 0.05]
if sig_effects.empty:
self.infer_result_text.insert(tk.END, "None.\n")
else:
self.infer_result_text.insert(tk.END, sig_effects.index.to_list())
self.append_legend(self.infer_result_text)
# --- Tukey ---
def run_tukey(self):
dv = self.get_selected_dv()
if not dv:
return
result_df = self.engine.run_tukey_12groups(dv)
self.infer_result_text.delete(1.0, tk.END)
if result_df is None or result_df.empty:
self.infer_result_text.insert(tk.END, "Tukey HSD could not be computed. Check data and groups.")
self.append_legend(self.infer_result_text)
return
self.infer_result_text.insert(tk.END, f"Tukey HSD Post-hoc (12 groups) for DV: {dv}\n")
self.infer_result_text.insert(tk.END, "Only significant pairwise comparisons shown (p < 0.05):\n")
self.infer_result_text.insert(tk.END, "=" * 80 + "\n")
sig_df = result_df[result_df['p-adj'] < 0.05]
if sig_df.empty:
self.infer_result_text.insert(tk.END, "No significant pairwise differences found.\n")
else:
self.infer_result_text.insert(tk.END, sig_df.to_string() + "\n")
self.append_legend(self.infer_result_text)
# --- Key-position Chi2 ---
def run_key_chi2(self):
self.infer_result_text.delete(1.0, tk.END)
factor = 'pipeline'
result = self.engine.run_key_position_chi2(factor)
if 'error' in result:
self.infer_result_text.insert(tk.END, f"Error: {result['error']}")
self.append_legend(self.infer_result_text)
return
self.infer_result_text.insert(tk.END, f"Chi-Square Test of Independence: Key-Position vs {factor}\n")
self.infer_result_text.insert(tk.END, "=" * 80 + "\n")
self.infer_result_text.insert(tk.END, f"Chi2 = {result['chi2']:.4f}, df = {result['dof']}, p = {result['p_value']:.4f}\n")
self.infer_result_text.insert(tk.END, f"Significant: {result['significant']}\n\n")
self.infer_result_text.insert(tk.END, "Contingency Table (rows = factor, cols = key position A-E):\n")
self.infer_result_text.insert(tk.END, result['contingency_table'].to_string() + "\n")
cond = self.cond_var.get()
if cond and cond in self.engine.conditions:
gof = self.engine.run_key_position_gof(cond)
if 'error' not in gof:
self.infer_result_text.insert(tk.END, f"\nGoodness-of-Fit (Uniform) for {cond}:\n")
self.infer_result_text.insert(tk.END, f"Chi2 = {gof['chi2']:.4f}, p = {gof['p_value']:.4f}, n={gof['n']}\n")
self.infer_result_text.insert(tk.END, f"Counts: A={gof['counts'][0]}, B={gof['counts'][1]}, C={gof['counts'][2]}, D={gof['counts'][3]}, E={gof['counts'][4]}\n")
self.append_legend(self.infer_result_text)
# --- Rank-based ANOVA ---
def run_rank_anova(self):
dv = self.get_selected_dv()
if not dv:
return
result = self.engine.run_rank_anova(dv)
self.infer_result_text.delete(1.0, tk.END)
if 'error' in result:
self.infer_result_text.insert(tk.END, f"Error: {result['error']}")
self.append_legend(self.infer_result_text)
return
self.infer_result_text.insert(tk.END, f"{result['method']} for DV: {dv}\n")
self.infer_result_text.insert(tk.END, f"N_total = {result['n_total']}\n")
self.infer_result_text.insert(tk.END, "=" * 80 + "\n")
self.infer_result_text.insert(tk.END, result['anova_table'].to_string() + "\n")
self.append_legend(self.infer_result_text)
# --- NEW: Diagnostic plots ---
def run_diagnostic_plots(self):
dv = self.get_selected_dv()
if not dv:
return
# Ensure a model is fitted (run ANOVA first)
if self.engine._last_model is None:
messagebox.showinfo("Info", "No ANOVA model fitted. Please run 'Run 3-Way ANOVA' first for this DV.")
return
try:
fig1, fig2 = self.engine.get_diagnostic_plots(dv)
# Show in a new window
plot_window = tk.Toplevel(self.root)
plot_window.title(f"Diagnostic Plots for {dv}")
plot_window.geometry("900x500")
notebook_plots = ttk.Notebook(plot_window)
notebook_plots.pack(fill=tk.BOTH, expand=True)
# QQ plot tab
frame1 = ttk.Frame(notebook_plots)
notebook_plots.add(frame1, text="QQ Plot")
canvas1 = FigureCanvasTkAgg(fig1, master=frame1)
canvas1.draw()
canvas1.get_tk_widget().pack(fill=tk.BOTH, expand=True)
# Residuals vs fitted tab
frame2 = ttk.Frame(notebook_plots)
notebook_plots.add(frame2, text="Residuals vs Fitted")
canvas2 = FigureCanvasTkAgg(fig2, master=frame2)
canvas2.draw()
canvas2.get_tk_widget().pack(fill=tk.BOTH, expand=True)
# Close button
ttk.Button(plot_window, text="Close", command=plot_window.destroy).pack(pady=5)
except Exception as e:
messagebox.showerror("Plot Error", f"Could not generate plots:\n{e}")
# --- NEW: Monte Carlo Chi-Square ---
def run_monte_carlo_chi2(self):
self.infer_result_text.delete(1.0, tk.END)
# Ask for number of simulations (optional)
n_sim = 1000
factor = 'pipeline'
result = self.engine.monte_carlo_key_chi2(factor, n_sim)
if 'error' in result:
self.infer_result_text.insert(tk.END, f"Error: {result['error']}")
self.append_legend(self.infer_result_text)
return
self.infer_result_text.insert(tk.END, f"Monte Carlo Chi-Square Test (Permutation) for Key-Position vs {factor}\n")
self.infer_result_text.insert(tk.END, "=" * 80 + "\n")
self.infer_result_text.insert(tk.END, f"Observed Chi2 = {result['observed_chi2']:.4f}\n")
self.infer_result_text.insert(tk.END, f"Simulated p-value (n={n_sim}): {result['p_sim']:.4f}\n")
self.infer_result_text.insert(tk.END, f"Significant (simulated): {result['significant_sim']}\n")
self.append_legend(self.infer_result_text)
# ========== Load / Refresh Methods ==========
def load_json_file(self):
filepath = filedialog.askopenfilename(
title="Select JSON file with MCQ data",
filetypes=[("JSON files", "*.json"), ("All files", "*.*")]
)
if not filepath:
return
try:
count = self.engine.load_questions_from_json(filepath)
self.status_label.config(text=f"Loaded {count} questions from {os.path.basename(filepath)}")
self.refresh_ui()
self.update_error_tab() # update error checks
except Exception as e:
messagebox.showerror("Error", f"Failed to load JSON: {e}\n{traceback.format_exc()}")
def load_directory(self):
dirpath = filedialog.askdirectory(title="Select directory with JSON files (one per condition)")
if not dirpath:
return
try:
count = self.engine.load_questions_from_directory(dirpath)
self.status_label.config(text=f"Loaded {count} questions from directory")
self.refresh_ui()
self.update_error_tab()
except Exception as e:
messagebox.showerror("Error", f"Failed to load directory: {e}\n{traceback.format_exc()}")
# --- Export ---
def export_data(self):
if self.engine.df is None or self.engine.df.empty:
messagebox.showwarning("No Data", "There is no data to export. Please load a dataset first.")
return
filepath = filedialog.asksaveasfilename(
defaultextension=".csv",
filetypes=[("CSV files", "*.csv"), ("Excel files", "*.xlsx"), ("All files", "*.*")],
title="Export Raw Data"
)
if not filepath:
return
try:
if filepath.lower().endswith('.xlsx'):
self.engine.export_dataframe(filepath, format='xlsx')
else:
self.engine.export_dataframe(filepath, format='csv')
messagebox.showinfo("Export Successful", f"Data exported to {filepath}")
except Exception as e:
messagebox.showerror("Export Error", f"Failed to export: {e}")
# --- Flush Data ---
def flush_data(self):
"""Clear all loaded data and reset the UI to a clean state."""
if messagebox.askyesno("Confirm Flush", "This will delete all loaded data and reset the interface. Proceed?"):
# Reset engine
self.engine.questions = []
self.engine.conditions = []
self.engine.df = None
self.engine._last_model = None
self.engine.loaded_files = []
self.engine.pre_parsed_combined_hash = None
self.engine.parsed_combined_hash = None
# Clear UI controls
self.cond_combo.set('')
self.cond_combo['values'] = []
self.compare_cond_combo.set('')
self.compare_cond_combo['values'] = []
self.metric_combo.set('')
self.metric_combo['values'] = []
self.infer_dv_combo.set('')
self.infer_dv_combo['values'] = []
# Clear text areas
self.stats_text.delete(1.0, tk.END)
self.stats_text.insert(tk.END, "Data flushed. Load a JSON file or directory to begin.")
self.infer_result_text.delete(1.0, tk.END)
self.infer_result_text.insert(tk.END, "Data flushed. Load data and select a DV first.")
self.group_result_info.delete(1.0, tk.END)
self.error_text.delete(1.0, tk.END)
self.error_text.insert(tk.END, "Integrity checks will appear here after data is loaded.\n")
# Clear treeviews
for item in self.data_tree.get_children():
self.data_tree.delete(item)
for item in self.group_result_tree.get_children():
self.group_result_tree.delete(item)
# Clear plots
self.figure.clear()
self.canvas.draw()
# Reset group lists
self.group_a_conditions = []
self.group_b_conditions = []
self.update_group_display()
self.update_available_conditions()
# Clear status
self.current_condition = None
self.status_label.config(text="Data flushed")
self.comp_result_label.config(text="")
def refresh_ui(self):
conditions = self.engine.conditions
self.cond_combo['values'] = conditions
if conditions:
self.cond_combo.set(conditions[0])
self.current_condition = conditions[0]
self.compare_cond_combo['values'] = [c for c in conditions if c != self.current_condition]
if self.compare_cond_combo['values']:
self.compare_cond_combo.set(self.compare_cond_combo['values'][0])
dvs = self.engine.get_available_dvs()
self.metric_combo['values'] = dvs
if dvs and not self.metric_combo.get():
self.metric_combo.set(dvs[0])
self.refresh_infer_dvs()
self.update_display()
else:
self.stats_text.delete(1.0, tk.END)
self.stats_text.insert(tk.END, "No data loaded.")
self.figure.clear()
self.canvas.draw()
self.clear_tree()
self.update_available_conditions()
self.group_a_conditions = []
self.group_b_conditions = []
self.update_group_display()
for item in self.group_result_tree.get_children():
self.group_result_tree.delete(item)
self.group_result_info.delete(1.0, tk.END)
def on_condition_selected(self, event=None):
self.current_condition = self.cond_var.get()
conditions = self.engine.conditions
others = [c for c in conditions if c != self.current_condition]
self.compare_cond_combo['values'] = others
if others:
self.compare_cond_combo.set(others[0])
self.update_display()
def update_display(self):
condition = self.cond_var.get()
if not condition or condition not in self.engine.conditions:
self.status_label.config(text="No valid condition selected.")
self.stats_text.delete(1.0, tk.END)
self.stats_text.insert(tk.END, "No valid condition selected.\nPlease load data and select a condition.")
return
self.current_condition = condition
try:
stats = self.engine.get_condition_stats(condition)
if not stats:
self.stats_text.delete(1.0, tk.END)
self.stats_text.insert(tk.END, f"No statistics available for condition '{condition}'.\n")
self.figure.clear()
self.canvas.draw()
self.clear_tree()
self.status_label.config(text=f"No stats for '{condition}'")
return
self.stats_text.delete(1.0, tk.END)
lines = []
lines.append(f"Condition: {condition}")
lines.append("=" * 60)
lines.append("Readability Metrics (mean ± std) [95% CI] [count]:")
for metric in ['fre_stem', 'fre_full', 'fkgl_stem', 'fkgl_full']:
s = stats.get(metric)
if s:
lines.append(f" {metric.upper()}: {s['mean']:.2f} ± {s['std']:.2f} "
f"CI [{s['ci_low']:.2f}–{s['ci_high']:.2f}] (n={s['count']})")
# If metric is fre_full and mean is 0, add a note
if metric == 'fre_full' and s['mean'] == 0.0 and s['count'] > 0:
lines.append(" → Zero FRE_full may indicate empty text or no words (check raw data).")
ds = stats.get('avg_distractor_sim')
if ds:
lines.append(f" AVG_DISTRACTOR_SIM: {ds['mean']:.4f} ± {ds['std']:.4f} "
f"CI [{ds['ci_low']:.4f}–{ds['ci_high']:.4f}] (n={ds['count']})")
key_dist = stats.get('key_distribution', {})
total = stats.get('key_total', 0)
if total > 0:
lines.append("Key Position Distribution:")
for letter in 'ABCDE':
cnt = key_dist.get(letter, 0)
pct = cnt / total * 100
lines.append(f" {letter}: {cnt} ({pct:.1f}%)")
lb = stats.get('length_bias')
if lb is not None:
km = stats.get('key_length_mean')
kstd = stats.get('key_length_std')
dm = stats.get('distractor_length_mean')
dstd = stats.get('distractor_length_std')
lines.append(f"Option Length Bias (Distractor mean - Key mean): {lb:.2f}")
if km is not None and kstd is not None:
lines.append(f" Key mean length: {km:.2f} ± {kstd:.2f}")
if dm is not None and dstd is not None:
lines.append(f" Distractor mean length: {dm:.2f} ± {dstd:.2f}")
else:
lines.append("Option Length Bias: Not enough data")
self.stats_text.insert(tk.END, "\n".join(lines))
self.update_plots(stats)
self.update_data_tree()
self.status_label.config(text=f"Displaying stats for '{condition}'")
except Exception as e:
self.status_label.config(text=f"Error updating display: {str(e)}")
messagebox.showerror("Display Error", f"An error occurred:\n{e}\n{traceback.format_exc()}")
def update_plots(self, stats):
self.figure.clear()
ax1 = self.figure.add_subplot(121)
ax2 = self.figure.add_subplot(122)
key_dist = stats.get('key_distribution', {})
total = stats.get('key_total', 0)
if total > 0:
letters = list('ABCDE')
counts = [key_dist.get(l, 0) for l in letters]
ax1.bar(letters, counts, color='skyblue')
ax1.set_title('Key Position Distribution')
ax1.set_xlabel('Option Position')
ax1.set_ylabel('Frequency')
else:
ax1.text(0.5, 0.5, 'No key data', ha='center', va='center')
ax1.set_title('Key Distribution')
km = stats.get('key_length_mean')
dm = stats.get('distractor_length_mean')
if km is not None and dm is not None:
ax2.bar(['Key', 'Distractors'], [km, dm], color=['green', 'orange'])
ax2.set_title('Mean Option Lengths')
ax2.set_ylabel('Mean length (characters)')
else:
ax2.text(0.5, 0.5, 'No length data', ha='center', va='center')
ax2.set_title('Length Bias')
self.figure.tight_layout()
self.canvas.draw()
def update_data_tree(self):
for item in self.data_tree.get_children():
self.data_tree.delete(item)
if self.engine.df is None:
return
condition = self.cond_var.get()
if condition not in self.engine.conditions:
return
cond_df = self.engine.df[self.engine.df['condition'] == condition]
if cond_df.empty:
return
# Show key length and mean distractor length in addition
cols = ['question_id', 'pdf_id', 'stem', 'key', 'fre_stem', 'fre_full', 'fkgl_stem', 'fkgl_full',
'avg_distractor_sim', 'key_length', 'mean_distractor_length']
self.data_tree['columns'] = cols
self.data_tree['show'] = 'headings'
for col in cols:
self.data_tree.heading(col, text=col)
# Adjust widths
if col == 'stem':
width = 200
elif col in ['key_length', 'mean_distractor_length']:
width = 120
else:
width = 100
self.data_tree.column(col, width=width)
for idx, row in cond_df.iterrows():
values = [row.get(c, '') for c in cols]
self.data_tree.insert('', tk.END, values=values)
def clear_tree(self):
for item in self.data_tree.get_children():
self.data_tree.delete(item)
def run_comparison(self):
if self.current_condition is None:
messagebox.showinfo("Info", "No condition selected.")
return
cond2 = self.compare_cond_var.get()
if not cond2:
messagebox.showinfo("Info", "Select a comparison condition.")
return
metric = self.metric_var.get()
if not metric:
messagebox.showinfo("Info", "Select a metric.")
return
result = self.engine.compare_conditions(self.current_condition, cond2, metric)
if not result:
messagebox.showinfo("Info", "Could not compute comparison.")
return
p = result['p_value']
sig = "p < 0.05 (significant)" if p < 0.05 else "p >= 0.05 (not significant)"
msg = (f"Comparison: {self.current_condition} vs {cond2}\n"
f"Metric: {metric}\n"
f"Mean1: {result['mean1']:.4f}, Mean2: {result['mean2']:.4f}\n"
f"t = {result['t_stat']:.4f}, p = {p:.4f}\n"
f"n1={result['n1']}, n2={result['n2']}\n"
f"Result: {sig}")
self.comp_result_label.config(text=msg)
# ========== Group Management Methods ==========
def update_available_conditions(self):
self.avail_listbox.delete(0, tk.END)
for cond in self.engine.conditions:
self.avail_listbox.insert(tk.END, cond)
def update_group_display(self):
self.group_a_listbox.delete(0, tk.END)
for cond in self.group_a_conditions:
self.group_a_listbox.insert(tk.END, cond)
self.group_b_listbox.delete(0, tk.END)
for cond in self.group_b_conditions:
self.group_b_listbox.insert(tk.END, cond)
def add_to_group_a(self):
selected = self.avail_listbox.curselection()
for idx in selected:
cond = self.avail_listbox.get(idx)
if cond not in self.group_a_conditions and cond not in self.group_b_conditions:
self.group_a_conditions.append(cond)
self.update_group_display()
def add_to_group_b(self):
selected = self.avail_listbox.curselection()
for idx in selected:
cond = self.avail_listbox.get(idx)
if cond not in self.group_b_conditions and cond not in self.group_a_conditions:
self.group_b_conditions.append(cond)
self.update_group_display()
def remove_from_group_a(self):
sel = self.group_a_listbox.curselection()
if sel:
cond = self.group_a_listbox.get(sel[0])
if cond in self.group_a_conditions:
self.group_a_conditions.remove(cond)
self.update_group_display()
def remove_from_group_b(self):
sel = self.group_b_listbox.curselection()
if sel:
cond = self.group_b_listbox.get(sel[0])
if cond in self.group_b_conditions:
self.group_b_conditions.remove(cond)
self.update_group_display()
def clear_group_a(self):
self.group_a_conditions = []
self.update_group_display()
def clear_group_b(self):
self.group_b_conditions = []
self.update_group_display()
def compare_groups(self):
if not self.group_a_conditions or not self.group_b_conditions:
messagebox.showinfo("Info", "Both groups must have at least one condition.")
return
overlap = set(self.group_a_conditions) & set(self.group_b_conditions)
if overlap:
if not messagebox.askyesno("Warning", f"Groups share conditions: {overlap}. Continue?"):
return
results = self.engine.compare_groups(self.group_a_conditions, self.group_b_conditions)
if not results:
self.group_result_info.delete(1.0, tk.END)
self.group_result_info.insert(tk.END, "No results. Possibly insufficient data for metrics.")
return
for item in self.group_result_tree.get_children():
self.group_result_tree.delete(item)
for metric, res in results.items():
p = res['p_value']
sig = "Yes" if p < 0.05 else "No"
values = (
metric,
f"{res['mean1']:.3f}",
f"{res['mean2']:.3f}",
f"{res['t_stat']:.3f}",
f"{p:.4f}",
sig,
f"{res['cohen_d']:.3f}" if not np.isnan(res['cohen_d']) else "NaN",
str(res['n1']),
str(res['n2'])
)
self.group_result_tree.insert('', tk.END, values=values)
info_lines = []
info_lines.append(f"Group A: {', '.join(self.group_a_conditions)} (n_questions total = {sum(1 for q in self.engine.questions if q.condition in self.group_a_conditions)})")
info_lines.append(f"Group B: {', '.join(self.group_b_conditions)} (n_questions total = {sum(1 for q in self.engine.questions if q.condition in self.group_b_conditions)})")
info_lines.append("t-test: Welch's t-test (unequal variances). Significance threshold: p < 0.05")
info_lines.append("Interpretation: If p < 0.05, the difference between group means is statistically significant.")
self.group_result_info.delete(1.0, tk.END)
self.group_result_info.insert(tk.END, "\n".join(info_lines))
# Add legend to group result info
self.group_result_info.insert(tk.END, "\n\n" + self.statistical_legend())
# ========== Error Tab Update ==========
def update_error_tab(self):
"""Populate the Error tab with integrity checks."""
self.error_text.delete(1.0, tk.END)
if not self.engine.questions:
self.error_text.insert(tk.END, "No data loaded.\n")
return
# 1. Empty questions
empty_issues = self.engine.get_empty_questions()
if empty_issues:
self.error_text.insert(tk.END, "⚠ EMPTY OR MISSING FIELDS\n")
self.error_text.insert(tk.END, "=" * 60 + "\n")
for issue in empty_issues:
self.error_text.insert(tk.END, f"Question: {issue['question_id']} (condition: {issue['condition']})\n")
self.error_text.insert(tk.END, f" Empty fields: {', '.join(issue['empty_fields'])}\n")
self.error_text.insert(tk.END, "\n")
else:
self.error_text.insert(tk.END, "✓ No empty fields found in any question.\n\n")
# 2. Duplicate IDs
dupes = self.engine.get_duplicate_ids()
if dupes:
self.error_text.insert(tk.END, "⚠ DUPLICATE QUESTION IDs\n")
self.error_text.insert(tk.END, "=" * 60 + "\n")
for qid in dupes:
self.error_text.insert(tk.END, f"Duplicate ID: {qid}\n")
self.error_text.insert(tk.END, "\n")
else:
self.error_text.insert(tk.END, "✓ No duplicate question IDs.\n\n")
# 3. Missing keys
missing_keys = self.engine.get_missing_keys()
if missing_keys:
self.error_text.insert(tk.END, "⚠ MISSING KEY POSITIONS\n")
self.error_text.insert(tk.END, "=" * 60 + "\n")
for qid in missing_keys:
self.error_text.insert(tk.END, f"Missing key: {qid}\n")
self.error_text.insert(tk.END, "\n")
else:
self.error_text.insert(tk.END, "✓ All questions have a valid key (A-E).\n\n")
# 4. SHA hashes (combined)
self.error_text.insert(tk.END, "🔐 FILE INTEGRITY (SHA-256)\n")
self.error_text.insert(tk.END, "=" * 60 + "\n")
pre_hash = self.engine.pre_parsed_combined_hash or "N/A"
post_hash = self.engine.parsed_combined_hash or "N/A"
self.error_text.insert(tk.END, f"Pre‑parsed combined SHA (from raw JSONs): {pre_hash}\n")
self.error_text.insert(tk.END, f"Parsed combined SHA (after import) : {post_hash}\n")
if pre_hash != "N/A" and post_hash != "N/A":
if pre_hash == post_hash:
self.error_text.insert(tk.END, "\n✓ Hashes match. Data integrity verified.\n")
else:
self.error_text.insert(tk.END, "\n❌ ERROR: Hashes DO NOT match! Data may have been altered during import.\n")
self.error_text.tag_add("error", "end-2l", "end-1l")
self.error_text.tag_config("error", foreground="red", font=("Courier", 9, "bold"))
else:
self.error_text.insert(tk.END, "\nUnable to verify integrity (missing hash).\n")
# 5. Zero FRE_full values (with full details)
zero_fre_questions = self.engine.get_zero_fre_full_questions()
if zero_fre_questions:
self.error_text.insert(tk.END, "\n⚠ ZERO FRE_FULL DETECTED\n")
self.error_text.insert(tk.END, "=" * 60 + "\n")
for q in zero_fre_questions[:20]: # limit to 20 for display
self.error_text.insert(tk.END, f"ID: {q.question_id} | Condition: {q.condition}\n")
self.error_text.insert(tk.END, f"Stem: {q.stem[:100]}{'...' if len(q.stem)>100 else ''}\n")
for i, opt in enumerate(q.options):
self.error_text.insert(tk.END, f" {chr(ord('A')+i)}: {opt[:80]}{'...' if len(opt)>80 else ''}\n")
self.error_text.insert(tk.END, "\n")
if len(zero_fre_questions) > 20:
self.error_text.insert(tk.END, f"... and {len(zero_fre_questions)-20} more questions with FRE_full = 0.\n")
self.error_text.insert(tk.END, "These questions have no alphabetic words in the combined stem+options.\n")
self.error_text.insert(tk.END, "Check the raw data for empty or non‑textual options.\n")
else:
self.error_text.insert(tk.END, "\n✓ No zero FRE_full values found.\n")
# 6. Total vs unique
total_q = len(self.engine.questions)
unique_q = len(set(q.question_id for q in self.engine.questions))
if total_q != unique_q:
self.error_text.insert(tk.END, f"\n⚠ Total questions: {total_q}, Unique IDs: {unique_q} (duplicates exist)\n")
else:
self.error_text.insert(tk.END, f"\n✓ Total questions: {total_q}, all IDs unique.\n")
# 7. NaN columns
if self.engine.df is not None:
nan_cols = self.engine.df.columns[self.engine.df.isna().any()].tolist()
if nan_cols:
self.error_text.insert(tk.END, f"\n⚠ Columns with NaN values: {', '.join(nan_cols)}\n")
self.error_text.insert(tk.END, "This may indicate missing data for some questions.\n")
else:
self.error_text.insert(tk.END, "\n✓ No NaN values in the dataframe.\n")
self.error_text.insert(tk.END, "\n" + "-" * 60 + "\n")
self.error_text.insert(tk.END, "All checks completed.\n")
# =============================================================================
# 5. Main Entry
# =============================================================================
def main():
root = tk.Tk()
app = MCQAnalysisApp(root)
root.mainloop()
if __name__ == "__main__":
try:
main()
except Exception as e:
print("=" * 60)
print("UNHANDLED EXCEPTION:")
traceback.print_exc()
print("=" * 60)
input("Press Enter to exit...")