Ignora Símbolo
" elif is_timeout: if correct: # Timeout was the correct action (e.g., Attention non-target, Memory suppressor) feedback_html = "T/O (OK)
" else: # Timeout occurred when a response was needed feedback_html = "T/O (Error)
" elif correct: feedback_html = "✓
" else: # Incorrect response key pressed feedback_html = "❌
" # --- State Update --- state["test_feedback"] = feedback_html state["test_user_response"] = key if not is_timeout else "T/O" state["awaiting_input"] = False # Input processed or timed out state["last_processed_index"] = stimulus_index # Mark as processed # --- Score Update --- if correct: state["positive_score"] = state.get("positive_score", 0) + 1 else: # Only count errors where a response *should* have been made or an *incorrect* response was made. # Don't penalize "correct" timeouts twice. is_correct_timeout_scenario = is_timeout and correct if not is_correct_timeout_scenario: state["negative_score"] = state.get("negative_score", 0) + 1 # --- Trial Result Logging --- trial_info = { 'idx': stimulus_index, 'stim': str(stimulus), 'resp': state["test_user_response"], 'ok': correct, 'rt': round(rt, 3) if rt != -1 else -1, 'to': is_timeout, 'req_key': required_key, # Test specific details 'prev_stim': state.get("test_last_stimulus", '') if current_test == "Atencion" else None, 'is_ax': is_target_after_cue if current_test == "Atencion" else None, 'is_sa': is_similar_distractor_alone if current_test == "Atencion" else None, 'is_congruent': is_congruent if current_test == "Inhibicion" else None, 'is_letter': is_letter_trial if current_test == "Memoria" else None, 'supp_ignored': suppressor_ignored if current_test == "Memory" else None, 'rule_idx': current_rule_index if current_test == "Flexibilidad" else None, 'is_switch': state.get("is_switch_trial", False) if current_test == "Flexibilidad" else None, 'is_intf': is_interference_trial if current_test == "Flexibilidad" else None, 'intf_key': active_int_key if current_test == "Flexibilidad" and is_interference_trial else None, } if not isinstance(state.get("current_trial_results"), list): state["current_trial_results"] = [] state["current_trial_results"].append(trial_info) log_message(f"Processed Idx:{stimulus_index} -> Correct:{correct}, T/O:{is_timeout}, RT:{trial_info['rt']:.3f}. Scores: +{state['positive_score']}/-{state['negative_score']}") return state def calculate_detailed_scores(test_name, trial_results, params, expected_response_info, sequence): if not trial_results: return {'precision': 0, 'analysis': "Sin datos de prueba.
", 'avg_rt': 0, 'rt_sd': 0} n = len(trial_results) correct_trials = sum(1 for r in trial_results if r['ok']) # Precision calculation: Correct trials / Total trials. precision = (correct_trials / n * 100) if n > 0 else 0 # Timeouts: Any trial marked with 'to: True' timeouts = sum(1 for r in trial_results if r['to']) timeout_percent = (timeouts / n * 100) if n > 0 else 0 # Valid RTs: Correct, non-timeout trials with positive RT valid_rts = [r['rt'] for r in trial_results if r['ok'] and not r['to'] and r.get('rt', -1) > 0] avg_rt = sum(valid_rts) / len(valid_rts) if valid_rts else 0 rt_sd = 0 if PANDAS_AVAILABLE and len(valid_rts) > 1: try: rt_sd = pd.Series(valid_rts).std() except Exception: rt_sd = 0 # Fallback if pandas std fails elif len(valid_rts) > 1: mean = avg_rt; variance = sum([(rt - mean) ** 2 for rt in valid_rts]) / (len(valid_rts) - 1); rt_sd = variance ** 0.5 else: rt_sd = 0 rt_analysis = f"- TR Medio (Correcto): {avg_rt:.3f}s" if avg_rt > 0 else "" if rt_sd > 0: rt_analysis += f" (±{rt_sd:.3f}s)" if avg_rt > 1e-6 and rt_sd > 0 and len(valid_rts) > 3 and (rt_sd / avg_rt) > 0.5 : # Check Coefficient of Variation rt_analysis += " - Alta Variabilidad TR" analysis = [f"Precisión Total: {precision:.1f}% ({correct_trials}/{n})"] # Timeouts: Report if high percentage of *any* timeout (correct or incorrect) if timeout_percent > 25: analysis.append(f"- T/O Altos: {timeouts}/{n} ({timeout_percent:.0f}% > 25%)") if rt_analysis: analysis.append(rt_analysis) analysis.append("Error en análisis detallado.
") # Clean up analysis HTML analysis_html = "".join(f"{line}
" for line in analysis if isinstance(line, str) and line.strip() and line != "", "
" display_text = ""; style_color = "#00FF00"; extra_class = "" try: if test_name == "Atencion": display_text = str(stim_raw) elif test_name == "Inhibicion": if isinstance(stim_raw, tuple) and len(stim_raw) == 2: word_key, color_name_key = stim_raw display_text = INHIB_WORDS.get(word_key, str(word_key)) style_color = INHIB_COLORS.get(color_name_key, "#FFFFFF") else: raise ValueError("Invalid stimulus format for Inhibicion") elif test_name == "Memoria": if isinstance(stim_raw, str): if stim_raw in MEM_SUPPRESS_SYMBOLS: display_text = stim_raw; extra_class = " stimulus-suppressor"; style_color = "#888888" elif stim_raw in MEM_LETTERS: display_text = stim_raw else: display_text = stim_raw # Fallback for unexpected string else: raise ValueError("Invalid stimulus type for Memoria") elif test_name == "Flexibilidad": if isinstance(stim_raw, tuple) and len(stim_raw) == 3: number, color_code, _ = stim_raw # Don't display the interference key itself display_text = str(number) color_map = params.get('colors', FLEX_RULE_COLORS) style_color = color_map.get(color_code, "#00FF00") # Default to green if color code unknown interference_info = params.get('interference', {}) interference_color_code = interference_info.get('color_char', FLEX_INTERFERENCE_COLOR_CHAR) if color_code == interference_color_code: extra_class = " stimulus-interference" else: raise ValueError("Invalid stimulus format for Flexibilidad") else: display_text = str(stim_raw) # Default fallback # Basic HTML escaping display_text_safe = display_text.replace("&", "&").replace("<", "<").replace(">", ">") if not display_text_safe or not display_text_safe.strip(): display_text_safe = " " # Ensure non-empty return f"
{display_text_safe}
" except Exception as e: log_message(f"ERROR formatting stimulus: Stim={stim_raw}, Test={test_name}\n{traceback.format_exc()}") return "ERR
" # Error display def get_test_buttons_visibility(state): # Start with all hidden visibility = [gr.update(visible=False)] * 11 stage = state.get("stage", "") test_name = state.get("current_test", "") params = state.get("test_params", {}) stimulus_raw = state.get("test_stimulus") is_awaiting = state.get("awaiting_input", False) if stage.startswith("test_") and is_awaiting and test_name: try: if test_name == "Atencion": target_key = params.get('target', ATTN_TARGET) distractor_key = params.get('similar_distractor_key', ATTN_SIMILAR_DISTRACTOR_KEY) # Always show both buttons, user decides when to press based on rules visibility[0] = gr.update(visible=True, value=target_key.upper()) # Target button (A->X) visibility[1] = gr.update(visible=True, value=f"{distractor_key.upper()} (Similar)") # Distractor button (Similar Alone) elif test_name == "Inhibicion": # Always show both buttons visibility[2] = gr.update(visible=True) # Correct button visibility[3] = gr.update(visible=True) # Incorrect button elif test_name == "Memoria": # Show buttons ONLY if it's a letter trial, not a suppressor is_letter = isinstance(stimulus_raw, str) and stimulus_raw in MEM_LETTERS if is_letter: match_key = params.get('match_key', MEM_MATCH_KEY) nomatch_key = params.get('nomatch_key', MEM_NOMATCH_KEY) visibility[4] = gr.update(visible=True, value=f"{match_key.upper()} (Igual)") # Match button visibility[5] = gr.update(visible=True, value=f"{nomatch_key.upper()} (Dif.)") # No-match button elif test_name == "Flexibilidad": if not isinstance(stimulus_raw, tuple) or len(stimulus_raw) != 3: return visibility # Invalid stimulus, keep buttons hidden _number, color_code, active_int_key = stimulus_raw interference_info = params.get('interference', {}) interference_color = interference_info.get('color_char', FLEX_INTERFERENCE_COLOR_CHAR) is_interference_trial = (color_code == interference_color) if is_interference_trial: # Show only the interference button visibility[10] = gr.update(visible=True, value=f"{active_int_key.upper()} (ROJO!)") else: # Show buttons for the *active rule* (using snapshot) rule_keys = params.get('rule_keys', FLEX_RULE_KEYS) current_rule_index = state.get("test_current_rule_idx_snapshot", 0) if current_rule_index == 0: # Par/Impar rule active if len(rule_keys) > 0 and len(rule_keys[0]) == 2: key1, key2 = rule_keys[0] visibility[6] = gr.update(visible=True, value=f"{key1.upper()} (Par)") visibility[7] = gr.update(visible=True, value=f"{key2.upper()} (Impar)") elif current_rule_index == 1: # Alto/Bajo rule active if len(rule_keys) > 1 and len(rule_keys[1]) == 2: key1, key2 = rule_keys[1] visibility[8] = gr.update(visible=True, value=f"{key1.upper()} (Alto>5)") visibility[9] = gr.update(visible=True, value=f"{key2.upper()} (Bajo≤5)") except Exception as e: log_message(f"ERROR getting button visibility for {test_name}: {e}\n{traceback.format_exc()}") return [gr.update(visible=False)] * 11 # Fallback: hide all on error return visibility def get_stage_visibility(stage): visibility_map = { "welcome": stage == "welcome", "set_alias": stage == "set_alias", "menu": stage == "menu", "instructions": stage == "instructions", "test": stage.startswith("test_"), "results": stage == "results", "history": stage == "history" } block_names = ["welcome", "set_alias", "menu", "instructions", "test", "results", "history"] return [gr.update(visible=visibility_map.get(name, False)) for name in block_names] def determine_flex_rule(state, trial_index): # This determines the rule *for the current trial_index* based on switches *before* it if state.get("current_test") != "Flexibilidad": return state.get("test_current_rule_idx", 0) # Return current if not flex expected_info = state.get("test_expected_response", {}) switches = expected_info.get('switches', set()) initial_rule = state.get("initial_flex_rule", 0) # Rule for trial 0 # Count switches that occurred strictly *before* this trial index num_switches_passed = sum(1 for switch_trial_idx in switches if trial_index > switch_trial_idx) # If even number of switches passed, use initial rule. If odd, use the other rule. current_rule_index = initial_rule if num_switches_passed % 2 == 0 else 1 - initial_rule return current_rule_index def get_instructions_text(test_name, params): level = params.get('level', 1) icon = TEST_ICONS.get(test_name, '❓') timeout_base = params.get('response_timeout_base', 1.5) trials = params.get('trials', '?') lines = [f"### {icon} Instrucciones: {test_name} [Nivel {level}]", "{target.upper()} SOLO si ves la letra '{target}' inmediatamente después de la letra '{cue}'.")
lines.append(f"2. Presiona {distractor_key} si ves una letra similar distractora ({similar_distractors_str}) Y ésta NO sigue inmediatamente a la letra '{cue}'.")
lines.append(f"- NO PRESIONES NADA en todos los demás casos ( '{cue}' sola, '{target}' sola, '{cue}' seguida de otra letra NO target, o cualquier otra letra no similar).")
lines.append("- Debes discriminar entre el target A->X, los distractores similares aislados, y el resto.")
elif test_name == "Inhibicion":
lines.append("Objetivo: Supresión de Respuesta Prepotente (Stroop Modificado).")
lines.append("- Aparecerán nombres de colores escritos en diferentes colores de tinta.")
lines.append(f"- Si la PALABRA COINCIDE con el COLOR de la tinta (Ej: 'VERDE' en VERDE), presiona el botón Correcto.")
lines.append(f"- Si la PALABRA NO COINCIDE con el COLOR de la tinta (Ej: 'VERDE' en ROJO), presiona el botón Incorrecto.")
lines.append("- Debes responder en CADA ensayo, indicando si hay coincidencia o no.")
lines.append("- ¡CONCÉNTRATE y evita responder según la palabra solamente!")
elif test_name == "Memoria":
n_back = params.get('n_back', 1); match_key = params.get('match_key', MEM_MATCH_KEY).upper()
nomatch_key = params.get('nomatch_key', MEM_NOMATCH_KEY).upper()
suppress_prob = params.get('suppress_prob', 0); symbols_str = ", ".join(MEM_SUPPRESS_SYMBOLS)
lines.append(f"Objetivo: Memoria de Trabajo '{n_back}-Back' con Supresión de Distractores.")
if suppress_prob > 0: lines.append(f"- Aparecerán letras y, a veces, símbolos distractores ({symbols_str}).")
else: lines.append(f"- Aparecerán letras.")
if suppress_prob > 0: lines.append("- IGNORA COMPLETAMENTE los símbolos distractores. NO respondas a ellos.")
lines.append(f"- SOLO para las LETRAS: Presiona {match_key} si la letra actual es IGUAL a la letra que apareció hace {n_back} posiciones atrás (contando solo letras).")
lines.append(f"- Presiona {nomatch_key} si la letra actual es DIFERENTE a la letra de hace {n_back} posiciones (contando solo letras).")
lines.append(f"- Para las primeras {n_back} letras que aparezcan, presiona siempre {nomatch_key}.")
elif test_name == "Flexibilidad":
rules = params.get('rules', FLEX_RULES); rule_keys = params.get('rule_keys', FLEX_RULE_KEYS)
interference_info = params.get('interference', {}); base_int_key = interference_info.get('base_key', FLEX_INTERFERENCE_BASE_KEY).upper()
alt_int_key = interference_info.get('alt_key', FLEX_INTERFERENCE_ALT_KEY).upper(); key_switch_level = interference_info.get('key_switch_level', 99)
int_color_char = interference_info.get('color_char', FLEX_INTERFERENCE_COLOR_CHAR); int_color_hex = params.get('colors', {}).get(int_color_char, '#FF3333')
lines.append("Objetivo: Flexibilidad Cognitiva Avanzada (Cambio de Tarea + Interferencia).")
lines.append("Tarea Principal: Aplica la REGLA ACTIVA al NÚMERO que aparece.")
try:
k1r1, k2r1 = rule_keys[0][0].upper(), rule_keys[0][1].upper(); k1r2, k2r2 = rule_keys[1][0].upper(), rule_keys[1][1].upper()
lines.append(f"- Regla '{rules[0]}': {k1r1} si es Par, {k2r1} si es Impar.")
lines.append(f"- Regla '{rules[1]}': {k1r2} si es Alto (>5), {k2r2} si es Bajo (≤5).")
except (IndexError, TypeError): lines.append("- Error cargando reglas/teclas."); log_message("ERROR generating Flex instructions: Rule keys invalid.")
lines.append("- La regla activa (Par/Impar o Alto/Bajo) CAMBIARÁ aleatoriamente durante la prueba SIN AVISO. ¡Debes adaptarte!")
lines.append(f"- Tarea de Interferencia: Si el número aparece en color ROJO ({int_color_char}), IGNORA la regla activa y presiona la tecla de interferencia designada ({base_int_key} o {alt_int_key}).")
if level >= key_switch_level: lines.append(f"- ¡ALERTA NIVEL ALTO! La tecla de interferencia (rojos) empieza siendo {base_int_key}, pero puede CAMBIAR a {alt_int_key} durante la prueba.")
else: lines.append(f"- La tecla de interferencia (rojos) es siempre {base_int_key}.")
lines.append(f"⚠️ ADVERTENCIA ⚠️
Este NO es un test médico. Es un DESAFÍO COGNITIVO EXTREMO.
Diseñado para explorar límites, no para diagnóstico. Resultados TEMÁTICOS.
ALTA DIFICULTAD: Niveles avanzados pueden ser frustrantes.
Si tienes dudas sobre tu cognición, CONSULTA A UN PROFESIONAL.
Al continuar, ACEPTAS estas condiciones.
Rendimiento depende del dispositivo y concentración.
""", elem_classes="welcome-text") accept_button = gr.Button("Entendido, Acepto el Desafío", elem_classes="btn-matrix-accept", scale=1) with gr.Column(visible=False, elem_classes="main-content-box", elem_id="alias-block") as alias_block: gr.Markdown("### Designación de Agente", elem_classes="matrix-title") alias_input=gr.Textbox(label="Ingresa tu Alias:", placeholder="Ej: Neo, Trinity...", lines=1, max_lines=1, scale=3, elem_id="alias-input-box") alias_submit_button=gr.Button("Confirmar Alias", elem_classes="btn-matrix", scale=1) alias_feedback = gr.Markdown("", elem_id="alias_feedback") with gr.Column(visible=False, elem_classes="main-content-box", elem_id="menu-block") as menu_block: gr.Markdown("### Terminal Principal", elem_classes="matrix-title") agent_info=gr.Markdown("Agente: ??? | Nivel de Acceso: ???", elem_id="agent-info-menu") start_sim_button=gr.Button("▶️ 1. Iniciar Nueva Simulación", elem_classes="btn-matrix btn-menu") change_alias_button=gr.Button("👤 2. Cambiar Designación (Alias)", elem_classes="btn-matrix btn-menu") view_history_button=gr.Button("📜 3. Ver Registros Previos", elem_classes="btn-matrix btn-menu") reset_level_button=gr.Button("⏪ 4. Reset Nivel a 1", elem_classes="btn-matrix btn-menu btn-exit") with gr.Column(visible=False, elem_classes="main-content-box", elem_id="instructions-block") as instructions_block: instructions_title=gr.Markdown("#### Cargando Briefing...", elem_classes="matrix-subtitle") instructions_text=gr.HTML("...", elem_id="instr-text") start_test_button=gr.Button("¡INICIAR PRUEBA AHORA!", elem_classes="btn-matrix-accept") with gr.Column(visible=False, elem_classes="main-content-box", elem_id="test-block") as test_block: test_title=gr.Markdown("#### Ejecutando Simulación...", elem_classes="matrix-subtitle") progress_indicator=gr.Markdown("Progreso: 0/0", elem_classes="progress-indicator") score_display = gr.Markdown("Score: +0 -0", elem_id="score-display") timer_display = gr.Markdown("T-Max: ---s", elem_id="timer-display") stimulus_display=gr.HTML("", elem_id="stimulus-display") feedback_display=gr.HTML(" ", elem_id="feedback-display") with gr.Row(equal_height=True, variant="compact"): attn_target_btn = gr.Button("X", elem_classes="btn-matrix-response", scale=1, visible=False, elem_id="attn-target-btn") attn_distractor_btn = gr.Button("D (Similar)", elem_classes="btn-matrix-response btn-attn-distractor", scale=1, visible=False, elem_id="attn-distractor-btn") with gr.Row(equal_height=True, variant="compact"): inhib_correct_btn = gr.Button("Correcto", elem_classes="btn-matrix-response btn-inhib-correct", scale=1, visible=False, elem_id="inhib-correct-btn") inhib_incorrect_btn = gr.Button("Incorrecto", elem_classes="btn-matrix-response btn-inhib-incorrect", scale=1, visible=False, elem_id="inhib-incorrect-btn") with gr.Row(equal_height=True, variant="compact"): mem_s_btn = gr.Button("S (Igual)", elem_classes="btn-matrix-response", scale=1, visible=False, elem_id="mem-s-btn") mem_n_btn = gr.Button("N (Dif.)", elem_classes="btn-matrix-response", scale=1, visible=False, elem_id="mem-n-btn") with gr.Row(equal_height=True, variant="compact"): fx_p_btn = gr.Button("P (Par)", elem_classes="btn-matrix-response", visible=False, scale=1, elem_id="fx-p-btn") fx_i_btn = gr.Button("I (Impar)", elem_classes="btn-matrix-response", visible=False, scale=1, elem_id="fx-i-btn") with gr.Row(equal_height=True, variant="compact"): fx_a_btn = gr.Button("A (Alto>5)", elem_classes="btn-matrix-response", visible=False, scale=1, elem_id="fx-a-btn") fx_b_btn = gr.Button("B (Bajo≤5)", elem_classes="btn-matrix-response", visible=False, scale=1, elem_id="fx-b-btn") with gr.Row(equal_height=True, variant="compact"): fx_int_btn = gr.Button("X (ROJO!)", elem_classes="btn-matrix-response btn-red", visible=False, scale=1, elem_id="fx-int-btn") with gr.Column(visible=False, elem_classes="main-content-box", elem_id="results-block") as results_block: results_title=gr.Markdown("### Reporte de Simulación", elem_classes="matrix-title") results_summary=gr.HTML("...", elem_id="results-summary") results_level_msg=gr.HTML("...", elem_id="results-level") results_analysis_title=gr.Markdown("--- Análisis Táctico Detallado (V8.2 Clean Final) ---", elem_classes="matrix-subtitle", visible=True) results_analysis=gr.HTML("
Calculando...
", elem_id="results-analysis") results_info=gr.Markdown("Registro guardado (si alias definido). Desafío extremo completado.
", elem_id="results-info") results_back_button=gr.Button("Volver al Terminal Principal", elem_classes="btn-matrix") with gr.Column(visible=False, elem_classes="main-content-box", elem_id="history-block") as history_block: gr.Markdown("### Archivos de Simulaciones Previas", elem_classes="matrix-title") hist_df=gr.DataFrame(elem_id="history-table", wrap=True, visible=PANDAS_AVAILABLE, label="Top Registros (Max 25)") hist_html=gr.HTML("...
", elem_id="history-html", visible=not PANDAS_AVAILABLE) hist_back_btn=gr.Button("Volver al Terminal Principal", elem_classes="btn-matrix") # --- Helper Functions for Gradio --- all_blocks = [welcome_block, alias_block, menu_block, instructions_block, test_block, results_block, history_block] all_response_buttons = [ attn_target_btn, attn_distractor_btn, inhib_correct_btn, inhib_incorrect_btn, mem_s_btn, mem_n_btn, fx_p_btn, fx_i_btn, fx_a_btn, fx_b_btn, fx_int_btn ] test_trial_ui_updates = [stimulus_display, feedback_display, progress_indicator, score_display, timer_display] + all_response_buttons + [distraction_overlay] results_display_outputs = [results_title, results_summary, results_level_msg, results_analysis_title, results_analysis] history_display_outputs = [hist_df, hist_html] test_block_ui_components = [test_title, progress_indicator, score_display, timer_display, stimulus_display, feedback_display, *all_response_buttons, distraction_overlay] def update_menu_info(state_dict): alias = state_dict.get('alias', '???') level = state_dict.get('level', 1) return gr.update(value=f"Agente: {alias} | Nivel de Acceso: {level}") def confirm_alias_wrapper(current_state_dict, alias_str): next_state = current_state_dict # Modify directly alias = alias_str.strip()[:25] if isinstance(alias_str, str) else "" feedback_msg = ""; next_stage = next_state.get("stage", "welcome") if alias and len(alias) >= 2: next_state["alias"] = alias; next_stage = "menu"; gr.Info(f"Alias '{alias}' confirmado."); feedback_msg = "" log_message(f"Alias set to: {alias}") else: feedback_msg = "Alias inválido. Mínimo 2 caracteres.
"; gr.Warning("Alias inválido."); next_stage = "set_alias" next_state["stage"] = next_stage; visibility_updates = get_stage_visibility(next_stage) menu_info_update = update_menu_info(next_state); alias_feedback_update = gr.update(value=feedback_msg) return [next_state] + visibility_updates + [menu_info_update, alias_feedback_update] def start_simulation_wrapper(current_state_dict): next_state = current_state_dict # Modify directly if not next_state.get("alias"): log_message("Start simulation attempt without alias.") gr.Warning("Se requiere alias para iniciar."); next_state["stage"] = "set_alias" dummy_instr = [gr.update()] * 2 return [next_state] + get_stage_visibility("set_alias") + [update_menu_info(next_state)] + dummy_instr current_level = next_state.get("level", 1) log_message(f"Starting simulation round for Alias: {next_state.get('alias')}, Level: {current_level}") test_order = random.sample(AVAILABLE_TESTS, len(AVAILABLE_TESTS)) next_state["current_test_order"] = test_order first_test_index = 0; first_test_name = test_order[first_test_index] try: params = get_difficulty_params(first_test_name, current_level) instruction_text = get_instructions_text(first_test_name, params) flex_int_key = params.get('interference',{}).get('base_key', FLEX_INTERFERENCE_BASE_KEY) if first_test_name == "Flexibilidad" else "?" except Exception as e: log_message(f"ERROR preparing first test {first_test_name}: {e}\n{traceback.format_exc()}") gr.Error(f"Error Crítico preparando prueba {first_test_name}: {e}"); next_state["stage"] = "menu"; dummy_instr = [gr.update()] * 2 return [next_state] + get_stage_visibility("menu") + [update_menu_info(next_state)] + dummy_instr # Reset relevant fields for the new round using initial_state as template fields_to_reset = [ "current_test", "current_test_index", "test_params", "test_sequence", "test_expected_response", "test_trial_index", "current_stimulus_index", "last_processed_index", "test_stimulus", "test_user_response", "test_feedback", "test_last_stimulus", "test_start_time", "test_stimulus_show_time", "current_trial_timeout", "current_trial_iti", "awaiting_input", "initial_flex_rule", "test_current_rule_idx", "test_current_rule_idx_snapshot", "is_switch_trial", "flex_active_interference_key", "current_scores", "current_trial_results", "round_results", "_distraction_active", "positive_score", "negative_score" ] for field in fields_to_reset: next_state[field] = deepcopy(initial_state[field]) # Deepcopy only for reset values initial_flex_rule_for_round = random.randint(0, 1) if first_test_name == "Flexibilidad" else 0 next_state.update({ "stage": "instructions", "current_test": first_test_name, "current_test_index": first_test_index, "test_params": params, "current_scores": {test: 0.0 for test in test_order}, # Init scores "initial_flex_rule": initial_flex_rule_for_round, "test_current_rule_idx": initial_flex_rule_for_round, "test_current_rule_idx_snapshot": initial_flex_rule_for_round, "flex_active_interference_key": flex_int_key, "_level_before_results": current_level }) visibility_updates = get_stage_visibility("instructions"); menu_info_update = update_menu_info(next_state) instr_title_update = gr.update(value=f"#### {TEST_ICONS.get(first_test_name, '')} Briefing: {first_test_name} [Nivel {current_level}]") instr_text_update = gr.update(value=instruction_text) log_message(f"Showing instructions for first test: {first_test_name}") return [next_state] + visibility_updates + [menu_info_update, instr_title_update, instr_text_update] def start_test_wrapper(current_state_dict): next_state = current_state_dict # Modify directly current_test = next_state.get("current_test"); params = next_state.get("test_params", {}); level = next_state.get("level", 1) if not current_test or not params: log_message("ERROR: start_test_wrapper called with invalid state (no test/params).") gr.Error("Estado inválido: Falta info de prueba."); next_state["stage"] = "menu" visibility_updates = get_stage_visibility("menu"); menu_info_update = update_menu_info(next_state) # Need to return updates for all potentially visible components to hide them dummy_instr = [gr.update()] * 2; dummy_test_ui = [gr.update()] * len(test_block_ui_components); dummy_results = [gr.update()] * len(results_display_outputs); dummy_history = [gr.update()] * len(history_display_outputs) return [next_state] + visibility_updates + [menu_info_update] + dummy_instr + dummy_test_ui + dummy_results + dummy_history log_message(f"Starting test: {current_test}, Level: {level}") try: sequence, expected_info = generate_sequence_for_test(current_test, params) # Params might have been updated by generator (e.g., trials count) next_state['test_params'] = params except (RuntimeError, ValueError, Exception) as e: log_message(f"FATAL: Error generating sequence for {current_test}: {e}\n{traceback.format_exc()}") gr.Error(f"Error Crítico generando secuencia para {current_test}: {e}"); next_state["stage"] = "menu" visibility_updates = get_stage_visibility("menu"); menu_info_update = update_menu_info(next_state) dummy_instr = [gr.update()] * 2; dummy_test_ui = [gr.update()] * len(test_block_ui_components); dummy_results = [gr.update()] * len(results_display_outputs); dummy_history = [gr.update()] * len(history_display_outputs) return [next_state] + visibility_updates + [menu_info_update] + dummy_instr + dummy_test_ui + dummy_results + dummy_history next_stage = f"test_{current_test.lower()}" # Reset specific test-run variables initial_flex_rule_for_test = next_state.get("initial_flex_rule", 0) # Keep the round's initial rule if needed flex_int_key = params.get('interference',{}).get('base_key', FLEX_INTERFERENCE_BASE_KEY) if current_test == "Flexibilidad" else "?" next_state.update({ "stage": next_stage, "test_sequence": sequence, "test_expected_response": expected_info, "test_trial_index": 0, "current_stimulus_index": -1, "last_processed_index": -99, # Reset processed index "test_start_time": time.time(), "test_feedback": " ", "test_last_stimulus": '', "test_current_rule_idx": initial_flex_rule_for_test, # Start with the correct initial rule "test_current_rule_idx_snapshot": initial_flex_rule_for_test, "flex_active_interference_key": flex_int_key, "current_trial_results": [], "awaiting_input": False, "_distraction_active": False, "positive_score": 0, "negative_score": 0 }) # Reset scores for this specific test visibility_updates = get_stage_visibility(next_stage); menu_info_update = update_menu_info(next_state) instr_updates = [gr.update(), gr.update()] # Hide instruction text block test_title_update = gr.update(value=f"#### {TEST_ICONS.get(current_test, '')} Ejecutando: {current_test} [Nivel {level}]") progress_update = gr.update(value=f"Progreso: 0/{len(sequence)}"); stimulus_update = gr.HTML("Listo...
") feedback_update = gr.HTML(" "); timer_update = gr.update(value="T-Max: ---s"); score_update = gr.update(value="Score: +0 -0") # Reset score display button_updates = [gr.update(visible=False)] * len(all_response_buttons) # Hide all buttons initially distraction_update = gr.update(value="") results_dummies = [gr.update()] * len(results_display_outputs); history_dummies = [gr.update()] * len(history_display_outputs) log_message(f"Test setup complete. Sequence length: {len(sequence)}. Starting trial flow...") return ([next_state] + visibility_updates + [menu_info_update] + instr_updates + [test_title_update, progress_update, score_update, timer_update, stimulus_update, feedback_update] + button_updates + [distraction_update] + results_dummies + history_dummies) def run_trial_flow(state_at_start_of_flow): state = state_at_start_of_flow # Use the state passed in try: sequence = state.get("test_sequence", []); params = state.get("test_params", {}); test_duration = len(sequence); current_test = state.get("current_test", "Unknown") if test_duration == 0 or not current_test or current_test == "Unknown": log_message(f"ERROR: run_trial_flow started with invalid sequence/test. SeqLen={test_duration}, Test={current_test}") raise ValueError("Secuencia de prueba vacía o test inválido.") base_timeout = params.get('response_timeout_base', RESPONSE_WINDOW_TIMEOUT_BASE); timeout_var = params.get('response_timeout_variability', 0) base_iti = params.get('iti_base', INTER_TRIAL_INTERVAL_BASE); iti_var = params.get('iti_variability', 0) feedback_delay = params.get('feedback_delay', FEEDBACK_BASE_DELAY); distraction_prob = params.get('distraction_prob', 0) flex_int_info = params.get('interference', {}) if current_test == "Flexibilidad" else {} flex_key_switch_point = state.get("test_expected_response", {}).get('interference_key_switch_point', -1) if current_test == "Flexibilidad" else -1 flex_base_key = flex_int_info.get('base_key', FLEX_INTERFERENCE_BASE_KEY); flex_alt_key = flex_int_info.get('alt_key', FLEX_INTERFERENCE_ALT_KEY) log_message(f"run_trial_flow: Entering loop for {test_duration} trials.") while state["test_trial_index"] < test_duration: current_trial_idx = state["test_trial_index"] log_message(f"run_trial_flow: Starting Trial {current_trial_idx+1}/{test_duration}") # --- Inter-Trial Interval (ITI) --- iti_variability_amount = random.uniform(-iti_var, iti_var) * base_iti; current_iti = max(INTER_TRIAL_INTERVAL_MIN, base_iti + iti_variability_amount) state["current_trial_iti"] = current_iti; log_message(f"run_trial_flow: ITI={current_iti:.3f}s") time.sleep(current_iti) # --- Pre-Stimulus Setup --- state["current_stimulus_index"] = current_trial_idx # Set index *before* showing stim state["test_last_stimulus"] = state.get("test_stimulus", '') # Store previous stim for Attention test state["test_stimulus"] = sequence[current_trial_idx] # Determine Flex rule/key *before* showing stimulus if current_test == "Flexibilidad": active_rule_idx = determine_flex_rule(state, current_trial_idx); state["test_current_rule_idx"] = active_rule_idx state["test_current_rule_idx_snapshot"] = active_rule_idx # Snapshot rule for this trial state["is_switch_trial"] = (current_trial_idx in state.get("test_expected_response", {}).get('switches', set())) current_active_int_key = flex_alt_key if flex_key_switch_point != -1 and current_trial_idx >= flex_key_switch_point else flex_base_key state["flex_active_interference_key"] = current_active_int_key # Update stimulus tuple with the active interference key for this trial if isinstance(state["test_stimulus"], tuple) and len(state["test_stimulus"]) >= 2: stim_num, stim_color, _ = state["test_stimulus"]; state["test_stimulus"] = (stim_num, stim_color, current_active_int_key) log_message(f"run_trial_flow: Flex - RuleIdxSnap: {active_rule_idx}, IsSwitch: {state['is_switch_trial']}, ActiveIntKey: {current_active_int_key}") # Calculate timeout for this specific trial timeout_variability_amount = random.uniform(-timeout_var, timeout_var) * base_timeout; current_timeout = max(RESPONSE_WINDOW_TIMEOUT_MIN, base_timeout + timeout_variability_amount) state["current_trial_timeout"] = current_timeout # --- Show Stimulus & Start Response Window --- state["awaiting_input"] = True state["test_feedback"] = " " # Clear previous feedback state["test_stimulus_show_time"] = time.time() # Record exact show time # Prepare UI updates stimulus_html = format_stimulus_html(state) feedback_html = gr.HTML(state["test_feedback"]) # Show cleared feedback progress_html = f"Progreso: {current_trial_idx + 1}/{test_duration}" score_html = gr.update(value=f"Score: +{state.get('positive_score', 0)} -{state.get('negative_score', 0)}") timer_html = gr.update(value=f"T-Max: {current_timeout:.2f}s") button_visibility = get_test_buttons_visibility(state) # Distraction Flash (Optional) distraction_update_html = "" if random.random() < distraction_prob: state["_distraction_active"] = True; distraction_update_html = "" log_message("run_trial_flow: Distraction flash ON") # Yield state + UI updates with distraction active yield [state, gr.update(value=stimulus_html), feedback_html, gr.update(value=progress_html), score_html, timer_html] + button_visibility + [gr.update(value=distraction_update_html)] time.sleep(0.06); # Duration of flash state["_distraction_active"] = False; distraction_update_html = "" log_message("run_trial_flow: Distraction flash OFF") # Yield again to turn off flash (state is same, only overlay changes) # We need to yield *all* elements again, even if only one changed for Gradio Blocks yield [state, gr.update(value=stimulus_html), feedback_html, gr.update(value=progress_html), score_html, timer_html] + button_visibility + [gr.update(value=distraction_update_html)] else: # Yield state + UI updates without distraction flash log_message(f"run_trial_flow: Showing stimulus Idx:{current_trial_idx} '{state['test_stimulus']}', Timeout: {current_timeout:.3f}s") yield ([state, gr.update(value=stimulus_html), feedback_html, gr.update(value=progress_html), score_html, timer_html] + button_visibility + [gr.update(value=distraction_update_html)]) # --- Wait for Response or Timeout --- # Instead of sleep, use a loop checking state frequently. This allows click handler to update state. start_wait_time = time.time() response_received_in_loop = False while time.time() - start_wait_time < current_timeout: # Check if the state indicates a response was processed for *this* trial if not state.get("awaiting_input", True) and state.get("last_processed_index", -99) == current_trial_idx: log_message(f"run_trial_flow: Response detected for Idx:{current_trial_idx} within loop.") response_received_in_loop = True break # Exit wait loop, response was handled by click wrapper time.sleep(0.02) # Check state roughly every 20ms # --- Process Timeout (if necessary) --- # Re-check state *after* the wait period if not response_received_in_loop: # Double check if response came in *exactly* at the end, or if state is still awaiting input if state.get("awaiting_input", False) and state.get("last_processed_index", -99) != current_trial_idx: log_message(f"run_trial_flow: Timeout occurred for Idx:{current_trial_idx}. Processing T/O.") state = process_response(state, None, is_timeout=True) # Update state with T/O result # Yield Timeout Feedback state and UI updates stim_upd = gr.update(value="") # Clear stimulus fb_upd = gr.HTML(state["test_feedback"]) # Show T/O feedback score_upd = gr.update(value=f"Score: +{state.get('positive_score', 0)} -{state.get('negative_score', 0)}") btn_upd = [gr.update(visible=False)] * len(all_response_buttons) # Hide buttons distr_upd = gr.update(value="") # Ensure overlay off timer_upd = gr.update(value="") # Clear timer yield [state, stim_upd, fb_upd, gr.update(), score_upd, timer_upd] + btn_upd + [distr_upd] log_message(f"run_trial_flow: Waiting feedback delay after T/O: {feedback_delay:.3f}s") time.sleep(feedback_delay) # Yield Clear Timeout Feedback state and UI updates yield [state, gr.update(), gr.HTML(" "), gr.update(), gr.update(), gr.update()] + [gr.update()]*len(all_response_buttons) + [gr.update()] else: log_message(f"run_trial_flow: Wait ended for Idx:{current_trial_idx}, but response already processed (awaiting={state.get('awaiting_input')}, last_proc={state.get('last_processed_index')}). No T/O action needed.") else: log_message(f"run_trial_flow: Response for Idx:{current_trial_idx} handled by click wrapper. Proceeding.") # --- Move to Next Trial --- state["test_trial_index"] += 1 log_message(f"run_trial_flow: End of Trial {current_trial_idx+1}. Moving to next.") # --- End of Test --- log_message(f"run_trial_flow: Test '{current_test}' finished all {test_duration} trials.") state["awaiting_input"] = False # Ensure state reflects test end # Final yield to clear screen and ensure state is updated yield [state, gr.update(value="
"), gr.HTML(" "), gr.update(), gr.update(value=f"Score: +{state.get('positive_score', 0)} -{state.get('negative_score', 0)}"), gr.update(value="")] + [gr.update(visible=False)]*len(all_response_buttons) + [gr.update(value="")] except Exception as e: log_message(f"ERROR FATAL during run_trial_flow for {state.get('current_test', '??')}: {e}\n{traceback.format_exc()}") gr.Error(f"Error durante la prueba {state.get('current_test', '??')}. Volviendo al menú."); state["stage"] = "menu"; state["awaiting_input"] = False # Ensure input stops # Yield error state and clear test UI yield [state, gr.update(value="
ERROR
"), gr.HTML("ERROR
"), gr.update(), gr.update(value="Score: ERROR"), gr.update()] + [gr.update(visible=False)]*len(all_response_buttons) + [gr.update(value="")] def process_click_wrapper(current_state_dict, button_signal): state_at_click = current_state_dict # Use directly stimulus_index_at_click = state_at_click.get("current_stimulus_index", -1) is_awaiting = state_at_click.get("awaiting_input", False) last_processed = state_at_click.get("last_processed_index", -99) # Conditions to IGNORE the click: # 1. Not currently awaiting input. # 2. This specific stimulus index has already been processed. if not is_awaiting or last_processed == stimulus_index_at_click: log_message(f"Ignoring click for Idx:{stimulus_index_at_click}. Awaiting={is_awaiting}, LastProcessed={last_processed}") # Yield the *unchanged* state and no UI updates (or rather, dummy updates) yield [current_state_dict] + [gr.update()] * len(test_trial_ui_updates) return # Determine the actual key pressed based on the button signal actual_key_to_process = None if button_signal == 'placeholder_flex_int': # Special case for dynamic Flex Int key actual_key_to_process = state_at_click.get("flex_active_interference_key", FLEX_INTERFERENCE_BASE_KEY) # Use the key stored in state for this trial elif isinstance(button_signal, str): actual_key_to_process = button_signal # Use the static key from the button map else: log_message(f"ERROR: Click detected but the button signal is invalid: {button_signal}") yield [current_state_dict] + [gr.update()] * len(test_trial_ui_updates) # No state change on error return log_message(f"Click received for Idx:{stimulus_index_at_click}. Key: '{actual_key_to_process}'. Processing response.") # Process the response - this MODIFIES the state dictionary inplace next_state = process_response(state_at_click, actual_key_to_process, is_timeout=False) feedback_delay = next_state.get("test_params", {}).get("feedback_delay", FEEDBACK_BASE_DELAY) # Yield State with Feedback + Updated UI stim_upd = gr.update(value="") # Clear stimulus fb_upd = gr.HTML(next_state["test_feedback"]) # Show feedback (✓, ❌) score_upd = gr.update(value=f"Score: +{next_state.get('positive_score', 0)} -{next_state.get('negative_score', 0)}") button_visibility_update = [gr.update(visible=False)] * len(all_response_buttons) # Hide buttons after click distr_upd = gr.update(value="") # Ensure overlay off timer_upd = gr.update(value="") # Clear timer display yield ([next_state, stim_upd, fb_upd, gr.update(), score_upd, timer_upd] + button_visibility_update + [distr_upd]) # Wait for feedback duration log_message(f"Showing feedback for Idx:{stimulus_index_at_click}. Delay: {feedback_delay:.3f}s") time.sleep(feedback_delay) # Yield State + Clear Feedback UI # State remains the same as after processing, just clear the visual feedback log_message(f"Clearing feedback for Idx:{stimulus_index_at_click}.") yield [next_state, gr.update(), gr.HTML(" "), gr.update(), gr.update(), gr.update()] + [gr.update()]*len(all_response_buttons) + [gr.update()] def finish_current_test_wrapper(state_dict_from_flow): state = state_dict_from_flow # Modify directly current_test = state.get("current_test"); trial_results = state.get("current_trial_results", []); params = state.get("test_params", {}) expected_response_info = state.get("test_expected_response", {}); sequence = state.get("test_sequence", []) current_scores = state.get("current_scores", {}); test_order = state.get("current_test_order", []); current_test_index = state.get("current_test_index", -1) current_level = state.get("level", 1); alias = state.get("alias"); level_at_start_of_round = state.get("_level_before_results", current_level) log_message(f"Finishing test: {current_test} (Index: {current_test_index}). Num results: {len(trial_results)}") if not current_test or current_test_index < 0 or not test_order: log_message(f"WARN: Invalid state at finish_current_test. Test={current_test}, Index={current_test_index}. Resetting to menu.") state["stage"] = "menu"; # Reset fields related to the test run fields_to_reset = ["current_test", "current_test_index", "test_params", "test_sequence", "awaiting_input", "round_results", "current_trial_results", "positive_score", "negative_score", "test_expected_response", "current_stimulus_index", "last_processed_index"] for field in fields_to_reset: state[field] = deepcopy(initial_state[field]) visibility_updates = get_stage_visibility("menu"); menu_info_update = update_menu_info(state); dummy_instr = [gr.update()] * 2; dummy_test_ui = [gr.update()] * len(test_block_ui_components); dummy_results = [gr.update()] * len(results_display_outputs); dummy_history = [gr.update()] * len(history_display_outputs) return [state] + visibility_updates + [menu_info_update] + dummy_instr + dummy_test_ui + dummy_results + dummy_history # Calculate score for the completed test try: score_details = calculate_detailed_scores(current_test, trial_results, params, expected_response_info, sequence) precision = score_details['precision']; analysis_html = score_details['analysis'] log_message(f"Score calculated for {current_test}: Precision={precision:.1f}%") except Exception as e: log_message(f"ERROR calculating scores for {current_test}: {e}\n{traceback.format_exc()}") precision = 0.0; analysis_html = "
Error al generar análisis detallado.
" # Store score and analysis current_scores[current_test] = precision if "round_results" not in state or state["round_results"] is None: state["round_results"] = {"detailed_analysis": {}} if "detailed_analysis" not in state["round_results"]: state["round_results"]["detailed_analysis"] = {} state["round_results"]["detailed_analysis"][current_test] = analysis_html # --- Check if more tests in this round --- next_test_index = current_test_index + 1 if next_test_index < len(test_order): # --- Prepare for Next Test --- next_test_name = test_order[next_test_index]; log_message(f"Preparing next test: {next_test_name} (Index: {next_test_index})") next_stage = "instructions" try: next_params = get_difficulty_params(next_test_name, current_level) instruction_text = get_instructions_text(next_test_name, next_params) flex_int_key = next_params.get('interference',{}).get('base_key', FLEX_INTERFERENCE_BASE_KEY) if next_test_name == "Flexibilidad" else "?" except Exception as e: log_message(f"ERROR preparing next test {next_test_name}: {e}\n{traceback.format_exc()}") gr.Error(f"Error Crítico preparando siguiente prueba {next_test_name}: {e}"); state["stage"] = "menu"; visibility_updates = get_stage_visibility("menu"); menu_info_update = update_menu_info(state); dummy_instr = [gr.update()] * 2; dummy_test_ui = [gr.update()] * len(test_block_ui_components); dummy_results = [gr.update()] * len(results_display_outputs); dummy_history = [gr.update()] * len(history_display_outputs) return [state] + visibility_updates + [menu_info_update] + dummy_instr + dummy_test_ui + dummy_results + dummy_history initial_flex_rule_for_next_test = random.randint(0, 1) if next_test_name == "Flexibilidad" else 0 # Update state for the next test instructions state.update({ "stage": next_stage, "current_test": next_test_name, "current_test_index": next_test_index, "test_params": next_params, # Reset variables used within a single test run "test_sequence": [], "test_expected_response": {}, "test_trial_index": 0, "current_stimulus_index": -1, "last_processed_index": -99, "test_stimulus": None, "test_feedback": " ", "current_trial_results": [], "awaiting_input": False, "test_last_stimulus": '', "positive_score": 0, "negative_score": 0, # Reset scores for next test # Carry over Flex initial rule if set, or determine new one "initial_flex_rule": state.get("initial_flex_rule", 0) if next_test_name != "Flexibilidad" else initial_flex_rule_for_next_test, "test_current_rule_idx": initial_flex_rule_for_next_test, # Set based on new rule for next test "test_current_rule_idx_snapshot": initial_flex_rule_for_next_test, "flex_active_interference_key": flex_int_key, }) visibility_updates = get_stage_visibility(next_stage); menu_info_update = update_menu_info(state) instr_title_update = gr.update(value=f"#### {TEST_ICONS.get(next_test_name, '')} Briefing: {next_test_name} [Nivel {current_level}]") instr_text_update = gr.update(value=instruction_text) # Clear/hide other blocks test_block_dummies = [gr.update()] * len(test_block_ui_components); results_dummies = [gr.update()] * len(results_display_outputs); history_dummies = [gr.update()] * len(history_display_outputs) log_message(f"Showing instructions for next test: {next_test_name}") return ([state] + visibility_updates + [menu_info_update] + [instr_title_update, instr_text_update] + test_block_dummies + results_dummies + history_dummies) else: # --- All Tests in Round Completed - Show Results --- log_message("All tests in round completed. Calculating final results.") next_stage = "results"; avg_precision = sum(current_scores.values()) / len(current_scores) if current_scores else 0.0 can_advance = avg_precision >= ADVANCE_THRESHOLD_PERCENT; new_level = level_at_start_of_round; level_msg = "" if can_advance and level_at_start_of_round < MAX_DIFFICULTY_LEVEL: new_level += 1; level_msg = f"Análisis no disponible para {test}.
") all_analysis_parts.append(f"No hay análisis detallados.
" # Save results if alias exists if alias: log_message(f"Attempting to save results for {alias}, Level {level_at_start_of_round}...") save_success = guardar_resultado(alias, level_at_start_of_round, current_scores, test_order) log_message(f"Save successful: {save_success}") # Update state for results screen state.update({ "stage": next_stage, "level": new_level, # Update level for next round # Clear test-specific variables "current_test": None, "current_test_index": -1, "test_params": {}, "test_sequence": [], "test_expected_response": {}, "test_trial_index": 0, "current_stimulus_index": -1, "last_processed_index": -99, "test_stimulus": None, "awaiting_input": False, "current_trial_results": [], "test_last_stimulus": '', "test_feedback": " ", "_distraction_active": False, "positive_score": 0, "negative_score": 0, # Store final results summary "round_results": {"scores": current_scores, "avg_precision": avg_precision, "analysis_text": full_analysis_html, "level_message": level_msg, "summary_html": summary_html} }) visibility_updates = get_stage_visibility(next_stage); menu_info_update = update_menu_info(state); instr_dummies = [gr.update()] * 2; test_ui_dummies = [gr.update()] * len(test_block_ui_components) results_title_update = gr.update(value=f"### Reporte: {alias or 'Agente'} [Nivel {level_at_start_of_round}]"); results_summary_update = gr.HTML(summary_html); results_level_update = gr.HTML(level_msg) results_analysis_title_update = gr.update(visible=True); results_analysis_update = gr.HTML(full_analysis_html); history_dummies = [gr.update()] * len(history_display_outputs) log_message("Displaying final results screen.") # Output: state(1) + blocks(7) + menu(1) + instr(2) + test_ui(17) + results(5) + history(2) = 35 outputs return ([state] + visibility_updates + [menu_info_update] + instr_dummies + test_ui_dummies + [results_title_update, results_summary_update, results_level_update, results_analysis_title_update, results_analysis_update] + history_dummies) def view_history_wrapper(current_state_dict): state = current_state_dict # Modify directly state["stage"] = "history"; log_message("Viewing history.") visibility_updates = get_stage_visibility("history"); menu_info_update = update_menu_info(state) df_update = gr.update(value=None, visible=False); html_update = gr.update(value="Cargando historial...
", visible=not PANDAS_AVAILABLE) try: history_data = leer_historial_df() except Exception as e: log_message(f"ERROR reading history file: {e}\n{traceback.format_exc()}"); history_data = None; html_update = gr.update(value="Error crítico al leer historial.
", visible=True) if history_data is not None and len(history_data) > 0: log_message(f"History loaded. Records: {len(history_data)}. Pandas available: {PANDAS_AVAILABLE}") if PANDAS_AVAILABLE and isinstance(history_data, pd.DataFrame) and not history_data.empty: df_update = gr.update(value=history_data, visible=True); html_update = gr.update(visible=False) elif isinstance(history_data, list): # Fallback to HTML if pandas failed or not available headers = ['Alias', 'Lvl', 'Prec%', 'Fecha']; html_table = "| {h} | " for h in headers]) + "|||
|---|---|---|---|
| {alias_val} | {lvl_val} | {prec_val} | {ts_val} |
No hay registros de simulaciones previas.
", visible=True); df_update = gr.update(visible=False) else: # leer_historial_df returned None or empty list initially log_message("No history file found or file was empty.") if html_update.value == "Cargando historial...
": # Check if error message wasn't already set html_update = gr.update(value="No hay registros de simulaciones previas.
", visible=True) df_update = gr.update(visible=False) # Prepare full output list instr_dummies = [gr.update()] * 2; test_ui_dummies = [gr.update()] * len(test_block_ui_components); results_dummies = [gr.update()] * len(results_display_outputs) return ([state] + visibility_updates + [menu_info_update] + instr_dummies + test_ui_dummies + results_dummies + [df_update, html_update]) def reset_level_wrapper(current_state_dict): state = current_state_dict # Modify directly state["level"] = 1; gr.Info("Nivel de acceso reseteado a 1.") log_message(f"Level reset to 1 for alias {state.get('alias')}.") # Optionally clear scores/results if desired upon reset state["current_scores"] = {}; state["round_results"] = None; menu_info_update = update_menu_info(state) return [state, menu_info_update] def change_stage_wrapper(current_state_dict, target_stage): state = current_state_dict # Modify directly valid_stages = ["welcome", "set_alias", "menu", "history"] next_stage = target_stage if target_stage in valid_stages else "menu" # Default to menu if invalid target log_message(f"Changing stage to: {next_stage}") state["stage"] = next_stage # If moving back to a non-test stage, reset transient test variables if next_stage in ["welcome", "set_alias", "menu", "history"]: fields_to_reset = [ "current_test", "current_test_index", "test_params", "test_sequence", "test_expected_response", "test_trial_index", "current_stimulus_index", "last_processed_index", "test_stimulus", "test_user_response", "test_feedback", "test_last_stimulus", "test_start_time", "test_stimulus_show_time", "current_trial_timeout", "current_trial_iti", "awaiting_input", "initial_flex_rule", "test_current_rule_idx", "test_current_rule_idx_snapshot", "is_switch_trial", "flex_active_interference_key", "current_trial_results", "_distraction_active", # Keep round_results and current_scores until a new simulation starts "positive_score", "negative_score" ] for field in fields_to_reset: if field in state: state[field] = deepcopy(initial_state[field]) # Use deepcopy for reset values block_visibility_updates = get_stage_visibility(next_stage) menu_info_update = update_menu_info(state) feedback_clear = gr.update(value=" ") # Clear test feedback display alias_feedback_clear = gr.update(value="") if next_stage != "set_alias" else gr.update() # Clear alias feedback unless on that screen return [ state, *block_visibility_updates, menu_info_update, feedback_clear, alias_feedback_clear ] # --- Event Listeners --- # Define output lists based on the number of components they update # Base change stage (state + blocks + menu + test_feedback + alias_feedback = 1 + 7 + 1 + 1 + 1 = 11) base_outputs_change_stage = [ game_state, *all_blocks, agent_info, feedback_display, alias_feedback ] # Alias confirmation (state + blocks + menu + alias_feedback = 1 + 7 + 1 + 1 = 10) alias_outputs = [ game_state, *all_blocks, agent_info, alias_feedback ] # Start Simulation -> Instructions (state + blocks + menu + instr_title + instr_text = 1 + 7 + 1 + 1 + 1 = 11) start_sim_outputs = [ game_state, *all_blocks, agent_info, instructions_title, instructions_text ] # Full UI (state + blocks + menu + instr + test_ui + results + history = 1 + 7 + 1 + 2 + 17 + 5 + 2 = 35) full_ui_outputs = [ game_state, *all_blocks, agent_info, instructions_title, instructions_text, *test_block_ui_components, *results_display_outputs, *history_display_outputs ] # Trial Flow Yield (state + test_trial_ui = 1 + 17 = 18) trial_flow_yield_outputs = [game_state] + test_trial_ui_updates # Reset Level (state + menu_info = 1 + 1 = 2) reset_outputs = [game_state, agent_info] accept_button.click( lambda s: change_stage_wrapper(s, "set_alias" if s.get("alias") is None else "menu"), inputs=[game_state], outputs=base_outputs_change_stage, show_progress="hidden" ) alias_submit_button.click(confirm_alias_wrapper, [game_state, alias_input], alias_outputs, show_progress="hidden") alias_input.submit(confirm_alias_wrapper, [game_state, alias_input], alias_outputs, show_progress="hidden") start_sim_button.click(start_simulation_wrapper, [game_state], start_sim_outputs, show_progress="minimal") change_alias_button.click(lambda s: change_stage_wrapper(s, "set_alias"), [game_state], base_outputs_change_stage, show_progress="hidden") view_history_button.click(view_history_wrapper, [game_state], full_ui_outputs, show_progress="minimal") reset_level_button.click(reset_level_wrapper, [game_state], reset_outputs, show_progress="hidden") # Chain: Start Test -> Run Flow -> Finish Test/Show Next Instr/Show Results start_test_button.click( fn=start_test_wrapper, inputs=[game_state], outputs=full_ui_outputs, # Updates all UI elements for test start show_progress="minimal" ).then( fn=run_trial_flow, inputs=[game_state], outputs=trial_flow_yield_outputs, # Yields state and test UI updates during flow show_progress="hidden" ).then( fn=finish_current_test_wrapper, inputs=[game_state], outputs=full_ui_outputs, # Updates all UI for next instr or results show_progress="minimal" ) # Button click processing button_key_map = { attn_target_btn: ATTN_TARGET.lower(), attn_distractor_btn: ATTN_SIMILAR_DISTRACTOR_KEY.lower(), inhib_correct_btn: INHIB_CORRECT_KEY.lower(), inhib_incorrect_btn: INHIB_INCORRECT_KEY.lower(), mem_s_btn: MEM_MATCH_KEY.lower(), mem_n_btn: MEM_NOMATCH_KEY.lower(), fx_p_btn: FLEX_RULE_KEYS[0][0].lower(), fx_i_btn: FLEX_RULE_KEYS[0][1].lower(), fx_a_btn: FLEX_RULE_KEYS[1][0].lower(), fx_b_btn: FLEX_RULE_KEYS[1][1].lower(), fx_int_btn: 'placeholder_flex_int' # Special placeholder } for btn, key_or_placeholder in button_key_map.items(): btn.click( fn=process_click_wrapper, # Pass current state and the key associated with this button inputs=[game_state, gr.State(value=key_or_placeholder)], # Outputs are the state + test UI elements outputs=trial_flow_yield_outputs, show_progress="hidden" ) # Back buttons results_back_button.click(lambda s: change_stage_wrapper(s, "menu"), [game_state], base_outputs_change_stage, show_progress="hidden") hist_back_btn.click(lambda s: change_stage_wrapper(s, "menu"), [game_state], base_outputs_change_stage, show_progress="hidden") # --- App Launch --- if __name__ == "__main__": log_message("Application starting...") try: # Ensure CSV file exists with header with csv_lock: fieldnames = ['Alias','Timestamp','Level','AvgPrec'] + AVAILABLE_TESTS file_exists = os.path.isfile(ARCHIVO_RESULTADOS) needs_header = (not file_exists) or (os.path.getsize(ARCHIVO_RESULTADOS) == 0) if file_exists else True if needs_header: try: with open(ARCHIVO_RESULTADOS, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=fieldnames); writer.writeheader() log_message(f"Created/Cleared results file: {ARCHIVO_RESULTADOS}") except IOError as e: log_message(f"ERROR CRITICAL: Could not create results file: {e}"); sys.exit(1) except Exception as e: log_message(f"ERROR CRITICAL: Checking results file: {e}"); sys.exit(1) # Set queue concurrency high enough to handle rapid yields/clicks? # Default is 1, maybe increase slightly if needed, but be wary of race conditions demo.queue(max_size=20) log_message("Starting Gradio server...") try: # Use share=True for testing across devices if needed, otherwise False demo.launch(server_name="0.0.0.0", server_port=7860, share=False, debug=False) # Set debug=True for more Gradio logs if needed except Exception as e: log_message(f"ERROR FATAL during demo.launch(): {e}\n{traceback.format_exc()}") finally: log_message("--- Gradio Server Terminated or Launch Failed ---") # Optionally write log buffer to a file on exit try: log_file_path = APP_DIR / "matrix_app_log.txt" with open(log_file_path, "a", encoding="utf-8") as log_file: log_file.write(f"\n--- Session Log End: {datetime.now()} ---\n") log_buffer.seek(0) log_file.write(log_buffer.read()) log_message(f"Log saved to {log_file_path}") except Exception as log_e: print(f"Failed to write log buffer to file: {log_e}")