Ignora Símbolo
" elif is_timeout: feedback_html = "T/O
" elif correct: feedback_html = "✓
" else: feedback_html = "❌
" next_state["test_feedback"] = feedback_html next_state["test_user_response"] = key if not is_timeout else "T/O" next_state["awaiting_input"] = False next_state["last_processed_index"] = stimulus_index trial_info = { 'idx': stimulus_index, 'stim': stimulus, 'resp': next_state["test_user_response"], 'ok': correct, 'rt': round(rt, 3) if rt != -1 else -1, 'to': is_timeout, 'prev_stim': next_state.get("test_last_stimulus", '') if current_test == "Atencion" else None, 'is_letter': is_letter_trial if current_test == "Memoria" else None, 'rule_idx': current_rule_index if current_test == "Flexibilidad" else None, 'is_switch': next_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(next_state.get("current_trial_results"), list): next_state["current_trial_results"] = [] next_state["current_trial_results"].append(trial_info) return next_state # --- Score Calculation --- 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 = (correct_trials / n * 100) if n > 0 else 0 timeouts = sum(1 for r in trial_results if r['to']) timeout_percent = (timeouts / n * 100) if n > 0 else 0 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 as e: # print(f"WARN: Pandas error calculating SD for RTs: {e}") rt_sd = 0 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 / avg_rt) > 0.5: rt_analysis += " - Alta Variabilidad TR" analysis = [f"Precisión Total: {precision:.1f}% ({correct_trials}/{n})"] 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.
") analysis_html = "".join(f"{line}
" for line in analysis if line 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 or stim_raw in string.ascii_uppercase: display_text = stim_raw else: raise ValueError(f"Invalid stimulus content for Memoria: {stim_raw}") 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 display_text = str(number) color_map = params.get('colors', FLEX_RULE_COLORS) style_color = color_map.get(color_code, "#00FF00") 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) display_text_safe = display_text.replace("<", "<").replace(">", ">") if not display_text_safe or not display_text_safe.strip(): display_text_safe = " " return f"
{display_text_safe}
" except Exception as e: print(f"ERROR formatting stimulus: Stim={stim_raw}, Test={test_name}\n{traceback.format_exc()}") return "ERR
" def get_test_buttons_visibility(state): visibility = [gr.update(visible=False)] * 9 stage = state.get("stage", "") test_name = state.get("current_test", "") params = state.get("test_params", {}) stimulus_raw = state.get("test_stimulus") if stage.startswith("test_") and state.get("awaiting_input", False) and test_name: try: if test_name == "Atencion": target_key = params.get('target', ATTN_TARGET) visibility[0] = gr.update(visible=True, value=target_key.upper()) elif test_name == "Inhibicion": visibility[1] = gr.update(visible=True) elif test_name == "Memoria": if isinstance(stimulus_raw, str) and (stimulus_raw in MEM_LETTERS or stimulus_raw in string.ascii_uppercase): match_key = params.get('match_key', MEM_MATCH_KEY) nomatch_key = params.get('nomatch_key', MEM_NOMATCH_KEY) visibility[2] = gr.update(visible=True, value=f"{match_key.upper()} (Igual)") visibility[3] = gr.update(visible=True, value=f"{nomatch_key.upper()} (Dif.)") elif test_name == "Flexibilidad": if not isinstance(stimulus_raw, tuple) or len(stimulus_raw) != 3: # print(f"WARN: Invalid stimulus for Flex button visibility: {stimulus_raw}") return visibility _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: visibility[8] = gr.update(visible=True, value=f"{active_int_key.upper()} (ROJO!)") else: 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: if len(rule_keys) > 0 and len(rule_keys[0]) == 2: key1, key2 = rule_keys[0] visibility[4] = gr.update(visible=True, value=f"{key1.upper()} (Par)") visibility[5] = gr.update(visible=True, value=f"{key2.upper()} (Impar)") elif current_rule_index == 1: if len(rule_keys) > 1 and len(rule_keys[1]) == 2: key1, key2 = rule_keys[1] visibility[6] = gr.update(visible=True, value=f"{key1.upper()} (Alto>5)") visibility[7] = gr.update(visible=True, value=f"{key2.upper()} (Bajo≤5)") except Exception as e: print(f"ERROR getting button visibility for {test_name}: {e}") traceback.print_exc() return [gr.update(visible=False)] * 9 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): if state.get("current_test") != "Flexibilidad": return state.get("test_current_rule_idx", 0) expected_info = state.get("test_expected_response", {}) switches = expected_info.get('switches', set()) initial_rule = state.get("initial_flex_rule", 0) num_switches_passed = sum(1 for switch_trial_index in switches if trial_index > switch_trial_index) 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"- NO PRESIONES NADA si ves '{target}' solo, '{cue}' solo, '{cue}' seguido de otra letra, o cualquier otra letra.")
lines.append("- Ignora las letras distractoras (algunas pueden ser visualmente similares).")
elif test_name == "Inhibicion":
target_word = params.get('target_word', INHIB_TARGET_WORD)
target_color = params.get('target_color_name', INHIB_TARGET_COLOR)
nogo_word = params.get('nogo_word', INHIB_NOGO_WORD)
nogo_color = params.get('nogo_color_name', INHIB_NOGO_COLOR)
target_display_word = INHIB_WORDS.get(target_word, target_word)
nogo_display_word = INHIB_WORDS.get(nogo_word, nogo_word)
target_hex = INHIB_COLORS.get(target_color, "#FFFFFF")
nogo_hex = INHIB_COLORS.get(nogo_color, "#FFFFFF")
lines.append("Objetivo: Control Inhibitorio (Stroop Cognitivo).")
lines.append(f"- Presiona ESPACIO SOLO si ves la palabra '{target_display_word}' escrita en color {target_color.upper()}.")
lines.append(f"- NO PRESIONES NADA si ves la palabra '{nogo_display_word}' escrita en color {nogo_color.upper()}.")
lines.append("- NO PRESIONES NADA para cualquier otra combinación de palabra/color.")
lines.append("- ¡CONCÉNTRATE en la condición exacta de palabra Y color!")
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}).")
lines.append("- IGNORA COMPLETAMENTE los símbolos distractores. NO respondas a ellos.")
else:
lines.append(f"- Aparecerán letras.")
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} (ya que no hay {n_back} letras previas para comparar).")
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.")
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 (Par/Impar o Alto/Bajo) y presiona la tecla de interferencia.")
if level >= key_switch_level:
lines.append(f"- ¡ALERTA NIVEL ALTO! La tecla de interferencia (para números rojos) empieza siendo {base_int_key}, pero puede CAMBIAR a {alt_int_key} durante la prueba. ¡Presta atención!")
else:
lines.append(f"- La tecla de interferencia (para números 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") 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_btn = gr.Button("X", elem_classes="btn-matrix-response", scale=1, visible=False, elem_id="attn-btn") inh_btn = gr.Button("ESPACIO", elem_classes="btn-matrix-response", scale=1, visible=False, elem_id="inh-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) ---", 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") all_blocks = [welcome_block, alias_block, menu_block, instructions_block, test_block, results_block, history_block] all_response_buttons = [attn_btn, inh_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, 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] # Define the list of components belonging to the test block area for dummy updates test_block_ui_components = [test_title, progress_indicator, stimulus_display, feedback_display, timer_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(state_dict, alias_str): alias = alias_str.strip()[:25] if isinstance(alias_str, str) else "" next_state = deepcopy(state_dict) 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 = "" else: feedback_msg = "Alias inválido. Debe tener al menos 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(state_dict): current_state = deepcopy(state_dict) if not current_state.get("alias"): gr.Warning("Se requiere alias para iniciar la simulación.") current_state["stage"] = "set_alias" return [current_state] + get_stage_visibility("set_alias") + [update_menu_info(current_state), gr.update(), gr.update()] current_level = current_state.get("level", 1) test_order = random.sample(AVAILABLE_TESTS, len(AVAILABLE_TESTS)) current_state["current_test_order"] = test_order # print(f"\n--- Iniciando Simulación Nivel {current_level} (V8 HARD) ---") # print(f"Orden de Pruebas: {', '.join(test_order)}") first_test_index = 0 first_test_name = test_order[first_test_index] try: # print(f"Preparando prueba inicial: {first_test_name}...") 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: gr.Error(f"Error Crítico preparando la prueba {first_test_name}: {e}") traceback.print_exc() current_state["stage"] = "menu" return [current_state] + get_stage_visibility("menu") + [update_menu_info(current_state), gr.update(), gr.update()] 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" ] for field in fields_to_reset: current_state[field] = deepcopy(initial_state[field]) initial_flex_rule_for_round = random.randint(0, 1) if first_test_name == "Flexibilidad" else 0 current_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}, "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(current_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) return [current_state] + visibility_updates + [menu_info_update, instr_title_update, instr_text_update] def start_test_wrapper(state_dict): current_state = deepcopy(state_dict) current_test = current_state.get("current_test") params = current_state.get("test_params", {}) level = current_state.get("level", 1) if not current_test or not params: gr.Error("Estado inválido: Falta información de la prueba para iniciar.") current_state["stage"] = "menu" visibility_updates = get_stage_visibility("menu") menu_info_update = update_menu_info(current_state) dummy_instr = [gr.update()] * 2 dummy_test_ui = [gr.update()] * len(test_block_ui_components) # Use full test block components list dummy_results = [gr.update()] * len(results_display_outputs) dummy_history = [gr.update()] * len(history_display_outputs) return [current_state] + visibility_updates + [menu_info_update] + dummy_instr + dummy_test_ui + dummy_results + dummy_history try: # print(f"Generando secuencia V8 para: {current_test} Nivel {level}...") sequence, expected_info = generate_sequence_for_test(current_test, params) params['trials'] = len(sequence) current_state['test_params'] = params except (RuntimeError, ValueError, Exception) as e: gr.Error(f"Error Crítico generando secuencia para {current_test}: {e}") traceback.print_exc() current_state["stage"] = "menu" visibility_updates = get_stage_visibility("menu") menu_info_update = update_menu_info(current_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 [current_state] + visibility_updates + [menu_info_update] + dummy_instr + dummy_test_ui + dummy_results + dummy_history next_stage = f"test_{current_test.lower()}" initial_flex_rule_for_round = current_state.get("initial_flex_rule", 0) current_state.update({ "stage": next_stage, "test_sequence": sequence, "test_expected_response": expected_info, "test_trial_index": 0, "current_stimulus_index": -1, "last_processed_index": -1, "test_start_time": time.time(), "test_feedback": " ", "test_last_stimulus": '', "test_current_rule_idx": initial_flex_rule_for_round, "test_current_rule_idx_snapshot": initial_flex_rule_for_round, "flex_active_interference_key": params.get('interference',{}).get('base_key', FLEX_INTERFERENCE_BASE_KEY) if current_test == "Flexibilidad" else "?", "current_trial_results": [], "awaiting_input": False, "_distraction_active": False, }) visibility_updates = get_stage_visibility(next_stage) menu_info_update = update_menu_info(current_state) instr_updates = [gr.update(), gr.update()] 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") button_updates = [gr.update(visible=False)] * len(all_response_buttons) distraction_update = gr.update(value="") results_dummies = [gr.update()] * len(results_display_outputs) history_dummies = [gr.update()] * len(history_display_outputs) # print(f"Iniciando prueba V8: {current_test}, {len(sequence)} trials.") return ([current_state] + visibility_updates + [menu_info_update] + instr_updates + [test_title_update, progress_update, stimulus_update, feedback_update, timer_update] + button_updates + [distraction_update] + results_dummies + history_dummies) def run_trial_flow(state_after_start): state = deepcopy(state_after_start) 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: raise ValueError("Secuencia de prueba vacía o test inválido en run_trial_flow.") 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', {}) flex_key_switch_point = state.get("test_expected_response", {}).get('interference_key_switch_point', -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) key_switched_flag = False # print(f"--- Iniciando Flujo de Trials: {current_test} ({test_duration} trials) ---") while state["test_trial_index"] < test_duration: current_trial_idx = state["test_trial_index"] 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 time.sleep(current_iti) distraction_update_html = "" if random.random() < distraction_prob: state["_distraction_active"] = True distraction_update_html = "" yield [state] + [gr.update()] * (len(test_trial_ui_updates) - 1) + [gr.update(value=distraction_update_html)] time.sleep(0.05) state["_distraction_active"] = False distraction_update_html = "" state["test_last_stimulus"] = state.get("test_stimulus", '') state["test_stimulus"] = sequence[current_trial_idx] state["current_stimulus_index"] = current_trial_idx 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 state["is_switch_trial"] = (current_trial_idx in state.get("test_expected_response", {}).get('switches', set())) current_active_int_key = flex_base_key if flex_key_switch_point != -1: if current_trial_idx >= flex_key_switch_point: current_active_int_key = flex_alt_key if not key_switched_flag: # print(f"*** Flex Interference Key switched to: {flex_alt_key} at trial index {current_trial_idx} ***") key_switched_flag = True state["flex_active_interference_key"] = current_active_int_key if isinstance(state["test_stimulus"], tuple) and len(state["test_stimulus"]) == 3: stim_num, stim_color, _ = state["test_stimulus"] state["test_stimulus"] = (stim_num, stim_color, current_active_int_key) else: # print(f"WARN: Reconstructing Flex stimulus tuple at trial {current_trial_idx}") try: state["test_stimulus"] = (int(state["test_stimulus"][0]), str(state["test_stimulus"][1]), current_active_int_key) except: print(f"ERROR: Failed to reconstruct Flex stimulus tuple.") 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 state["awaiting_input"] = True state["test_feedback"] = " " state["test_stimulus_show_time"] = time.time() stimulus_html = format_stimulus_html(state) feedback_html = gr.HTML(state["test_feedback"]) progress_html = f"Progreso: {current_trial_idx + 1}/{test_duration}" timer_html = gr.update(value=f"T-Max: {current_timeout:.2f}s") button_visibility = get_test_buttons_visibility(state) yield [state, gr.update(value=stimulus_html), feedback_html, gr.update(value=progress_html), timer_html] + button_visibility + [gr.update(value=distraction_update_html)] time.sleep(current_timeout) state_after_wait = state if state_after_wait.get("awaiting_input", False) and \ state_after_wait.get("last_processed_index", -1) != current_trial_idx: state = process_response(state_after_wait, None, is_timeout=True) stim_upd = gr.update(value="") fb_upd = gr.HTML(state["test_feedback"]) btn_upd = [gr.update(visible=False)] * len(all_response_buttons) distr_upd = gr.update(value="") yield [state, stim_upd, fb_upd, gr.update(), gr.update(value="")] + btn_upd + [distr_upd] time.sleep(feedback_delay) yield [state, gr.update(), gr.HTML(" "), gr.update(), gr.update()] + [gr.update()]*len(all_response_buttons) + [gr.update()] state["test_trial_index"] += 1 # print(f"--- Flujo de Trials Completado: {current_test} ---") state["awaiting_input"] = False yield [state, gr.update(value="
"), gr.HTML(" "), gr.update(), gr.update(value="")] + [gr.update(visible=False)]*len(all_response_buttons) + [gr.update(value="")] except Exception as e: print(f"ERROR FATAL durante run_trial_flow para {state.get('current_test', '??')}: {e}") traceback.print_exc() gr.Error(f"Error durante la prueba {state.get('current_test', '??')}. Volviendo al menú.") state["stage"] = "menu" state["awaiting_input"] = False yield [state, gr.update(value="
ERROR
"), gr.HTML("ERROR
"), gr.update(), gr.update()] + \ [gr.update(visible=False)]*len(all_response_buttons) + \ [gr.update(value="")] def process_click_wrapper(current_state, button_signal): state_at_click = deepcopy(current_state) 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", -1) if not is_awaiting or last_processed == stimulus_index_at_click: yield [current_state] + [gr.update()] * (len(test_trial_ui_updates)) return actual_key_to_process = None if button_signal == 'placeholder_flex_int': actual_key_to_process = state_at_click.get("flex_active_interference_key", FLEX_INTERFERENCE_BASE_KEY) else: actual_key_to_process = button_signal if actual_key_to_process is None: print("ERROR: Click detectado pero la tecla no pudo ser resuelta.") yield [current_state] + [gr.update()] * (len(test_trial_ui_updates)) return 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) stim_upd = gr.update(value="") fb_upd = gr.HTML(next_state["test_feedback"]) button_visibility_update = [gr.update(visible=False)] * len(all_response_buttons) distr_upd = gr.update(value="") immediate_updates = ([next_state, stim_upd, fb_upd, gr.update(), gr.update(value="")] + button_visibility_update + [distr_upd]) yield immediate_updates time.sleep(feedback_delay) yield [next_state, gr.update(), gr.HTML(" "), gr.update(), gr.update()] + [gr.update()]*len(all_response_buttons) + [gr.update()] def finish_current_test_wrapper(state_dict_from_flow): state = deepcopy(state_dict_from_flow) 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) if not current_test or current_test_index < 0 or not test_order: print(f"WARN: Estado inválido al finalizar prueba. Test={current_test}, Index={current_test_index}. Volviendo al menú.") state["stage"] = "menu" state.update({k: deepcopy(initial_state[k]) for k in ["current_test", "current_test_index", "test_params", "test_sequence", "awaiting_input", "round_results", "current_trial_results"]}) 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) # Use full list size 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 # print(f"Finalizando prueba V8: {current_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'] except Exception as e: print(f"ERROR calculando puntajes V8 para {current_test}: {e}") traceback.print_exc() precision = 0.0 analysis_html = "
Error al generar análisis detallado.
" 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 next_test_index = current_test_index + 1 if next_test_index < len(test_order): next_test_name = test_order[next_test_index] # print(f"Transición a la siguiente prueba V8: {next_test_name}") 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: gr.Error(f"Error Crítico preparando la siguiente prueba V8 {next_test_name}: {e}") traceback.print_exc() 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) # Use full list size 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 state.update({ "stage": next_stage, "current_test": next_test_name, "current_test_index": next_test_index, "test_params": next_params, "test_sequence": [], "test_expected_response": {}, "test_trial_index": 0, "current_stimulus_index": -1, "last_processed_index": -1, "test_stimulus": None, "test_feedback": " ", "current_trial_results": [], "awaiting_input": False, "initial_flex_rule": initial_flex_rule_for_next_test, "test_current_rule_idx": initial_flex_rule_for_next_test, "test_current_rule_idx_snapshot": initial_flex_rule_for_next_test, "test_last_stimulus": '', "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) # **FIX:** Create dummy updates for all test block components (15) + results (5) + history (2) = 22 test_block_dummies = [gr.update()] * len(test_block_ui_components) # 15 dummies results_dummies = [gr.update()] * len(results_display_outputs) # 5 dummies history_dummies = [gr.update()] * len(history_display_outputs) # 2 dummies # Total returned = 1(state) + 7(vis) + 1(agent) + 1(instr_title) + 1(instr_text) + 15(test_dummies) + 5(results) + 2(history) = 33 return ([state] + visibility_updates + [menu_info_update] + [instr_title_update, instr_text_update] + test_block_dummies + results_dummies + history_dummies) else: # print("Simulación V8 completada. Mostrando resultados.") 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 detallado no disponible para {test}.
") all_analysis_parts.append(f"No hay análisis detallados disponibles.
" final_results_data = { "scores": current_scores, "avg_precision": avg_precision, "analysis_text": full_analysis_html, "level_message": level_msg, "summary_html": summary_html } state["round_results"] = final_results_data if alias: # print(f"Guardando resultado V8 para {alias} (Nivel {level_at_start_of_round})...") save_success = guardar_resultado(alias, level_at_start_of_round, current_scores, test_order) # print("Resultado guardado." if save_success else "¡Fallo al guardar resultado!") # else: # print("INFO: No se guarda resultado (sin alias).") state.update({ "stage": next_stage, "level": new_level, "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": -1, "test_stimulus": None, "awaiting_input": False, "current_trial_results": [], "test_last_stimulus": '', "test_feedback": " ", "_distraction_active": False, }) 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) # Use full list size results_title_update = gr.update(value=f"### Reporte V8: {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) # Total returned = 1 + 7 + 1 + 2(instr) + 15(test) + 5(results) + 2(history) = 33 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(state_dict): state = deepcopy(state_dict) state["stage"] = "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: print(f"ERROR en view_history_wrapper al llamar leer_historial_df: {e}") traceback.print_exc() history_data = None html_update = gr.update(value="Error crítico al leer historial.
", visible=True) if history_data is not None: 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) and history_data: headers = ['Alias', 'Lvl', 'Prec%', 'Fecha'] + [t[:4] for t in AVAILABLE_TESTS] html_table = "| {h} | " for h in headers]) + "{row.get(test,'-'):.1f} | " for test in AVAILABLE_TESTS]) html_table += f"||
|---|---|---|---|
| {alias_val} | {lvl_val} | {prec_val} | {ts_val} | {test_vals}
No hay registros de simulaciones previas.
", visible=True) df_update = gr.update(visible=False) else: if html_update.value == "Cargando historial...
": html_update = gr.update(value="No se pudo cargar el historial.
", visible=True) df_update = gr.update(visible=False) instr_dummies = [gr.update()] * 2 test_ui_dummies = [gr.update()] * len(test_block_ui_components) # Use full list size results_dummies = [gr.update()] * len(results_display_outputs) # Total returned = 1 + 7 + 1 + 2(instr) + 15(test) + 5(results) + 2(history) = 33 return ([state] + visibility_updates + [menu_info_update] + instr_dummies + test_ui_dummies + results_dummies + [df_update, html_update]) def reset_level_wrapper(state_dict): state = deepcopy(state_dict) state["level"] = 1 gr.Info("Nivel de acceso reseteado a 1.") state["current_scores"] = {} state["round_results"] = None menu_info_update = update_menu_info(state) return [state, menu_info_update] def change_stage_wrapper(state_dict, target_stage): state = deepcopy(state_dict) valid_stages = ["welcome", "set_alias", "menu", "history"] next_stage = target_stage if target_stage in valid_stages else "menu" state["stage"] = next_stage if next_stage in ["menu", "set_alias", "history", "welcome"]: 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_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", "round_results", "_distraction_active", "test_feedback" ] if next_stage == "menu": fields_to_reset.extend(["current_scores"]) for field in fields_to_reset: if field in state: state[field] = deepcopy(initial_state[field]) output_updates = {} output_updates[game_state] = state block_visibility_updates = get_stage_visibility(next_stage) block_components = all_blocks if len(block_visibility_updates) == len(block_components): for i, block_comp in enumerate(block_components): output_updates[block_comp] = block_visibility_updates[i] else: print(f"ERROR: Mismatch visibility updates ({len(block_visibility_updates)}) vs blocks ({len(block_components)})") for block_comp in block_components: output_updates[block_comp] = gr.update(visible=False) output_updates[agent_info] = update_menu_info(state) output_updates[feedback_display] = gr.update(value=" ") if next_stage != "set_alias": output_updates[alias_feedback] = gr.update(value="") return output_updates base_outputs_change_stage = [ game_state, *all_blocks, agent_info, feedback_display, alias_feedback ] alias_outputs = [ game_state, *all_blocks, agent_info, alias_feedback ] start_sim_outputs = [ game_state, *all_blocks, agent_info, instructions_title, instructions_text ] start_test_initial_outputs = ([ game_state, *all_blocks, agent_info, instructions_title, instructions_text, *test_block_ui_components, # Use the explicit list of 15 components *results_display_outputs, *history_display_outputs ]) trial_flow_yield_outputs = [game_state] + test_trial_ui_updates finish_test_outputs = ([ game_state, *all_blocks, agent_info, instructions_title, instructions_text, *test_block_ui_components, # Use the explicit list of 15 components *results_display_outputs, *history_display_outputs ]) history_outputs = ([ game_state, *all_blocks, agent_info, instructions_title, instructions_text, *test_block_ui_components, # Use the explicit list of 15 components *results_display_outputs, *history_display_outputs ]) 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], history_outputs, show_progress="minimal") reset_level_button.click(reset_level_wrapper, [game_state], reset_outputs, show_progress="hidden") start_test_button.click( fn=start_test_wrapper, inputs=[game_state], outputs=start_test_initial_outputs, show_progress="minimal" ).then( fn=run_trial_flow, inputs=[game_state], outputs=trial_flow_yield_outputs, show_progress="hidden" ).then( fn=finish_current_test_wrapper, inputs=[game_state], outputs=finish_test_outputs, show_progress="minimal" ) button_key_map = { attn_btn: ATTN_TARGET.lower(), inh_btn: ' ', 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' } for btn, key_or_placeholder in button_key_map.items(): btn.click( fn=process_click_wrapper, inputs=[game_state, gr.State(value=key_or_placeholder)], outputs=trial_flow_yield_outputs, show_progress="hidden" ) 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") # --- Launch App --- if __name__ == "__main__": # print("Iniciando Servidor Gradio MC AR v8 HARD...") try: 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 needs_header: try: with open(ARCHIVO_RESULTADOS, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() # print(f"Archivo de resultados V8 HARD creado con cabecera: {ARCHIVO_RESULTADOS}") except IOError as e: print(f"ERROR CRÍTICO: No se pudo crear el archivo de resultados: {e}") traceback.print_exc() exit(1) except Exception as e: print(f"ERROR CRÍTICO: No se pudo verificar/crear archivo de resultados: {e}") traceback.print_exc() exit(1) # print("Configurando Gradio Queue y Launch...") demo.queue() try: # print("--- Servidor Listo para Conexiones (V8 HARD) ---") demo.launch(server_name="0.0.0.0", server_port=7860, share=False) except Exception as e: print(f"ERROR FATAL durante demo.launch(): {e}") traceback.print_exc() finally: print("--- Servidor Gradio Terminado o Fallo en Launch ---")