import gradio as gr import numpy as np import pandas as pd from sklearn.linear_model import LinearRegression from sklearn.metrics import r2_score import os import tempfile import threading import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import io # ───────────────────────────────────────────── # CONSTANTES FÍSICAS (capacidad pulmonar vía globo) # ───────────────────────────────────────────── # El globo se modela como un elipsoide de revolución (esferoide), a partir de # dos circunferencias medidas con cinta métrica: # - Circunferencia ECUATORIAL: la parte más ancha del globo. # - Circunferencia POLAR: de polo a polo, pasando por la boquilla. # # El volumen medido en campo (a la temperatura y presión ambiente del momento) # se normaliza a una condición estándar de referencia -similar al criterio # BTPS usado en espirometría real (Body Temperature, Pressure, Saturated)- # para que las mediciones tomadas en el Chimborazo (frío, muy baja presión) # sean comparables con las tomadas en otras condiciones (por ejemplo, en el # salón de clases antes de la expedición). # # Simplificación pedagógica: se usa la ley combinada de los gases ideales # (P·V/T = constante) sin corregir por presión de vapor de agua, ya que esa # corrección requiere datos de humedad que esta práctica de campo no captura. STANDARD_PRESSURE_HPA = 1013.25 # presión estándar a nivel del mar (atmósfera estándar internacional) STANDARD_TEMP_K = 310.15 # 37 °C — temperatura corporal, referencia tipo BTPS def presion_estandar_por_altitud(altitud_m): """Fórmula barométrica de la Atmósfera Estándar Internacional (ISA). Estima la presión atmosférica esperada a una altitud dada, para no depender de que el equipo lleve un barómetro: basta con conocer la altitud (GPS, altímetro o dato de referencia del lugar).""" altitud_m = max(0.0, float(altitud_m)) return STANDARD_PRESSURE_HPA * (1 - 2.25577e-5 * altitud_m) ** 5.25588 def calcular_capacidad_pulmonar_globo(circ_ecuatorial_cm, circ_polar_cm, temperatura_c, altitud_m=0.0): """ Devuelve (volumen_L_medido, volumen_L_estandar, presion_ambiente_hPa). volumen_L_medido: volumen del globo en las condiciones reales de campo. volumen_L_estandar: mismo volumen, normalizado a 37°C y a la presión de referencia a nivel del mar, para comparar mediciones tomadas en distintas condiciones de temperatura/altitud. """ r_ecuatorial = circ_ecuatorial_cm / (2 * np.pi) r_polar = circ_polar_cm / (2 * np.pi) volumen_cm3 = (4 / 3) * np.pi * (r_ecuatorial ** 2) * r_polar volumen_L_medido = volumen_cm3 / 1000.0 presion_ambiente_hpa = presion_estandar_por_altitud(altitud_m) temp_ambiente_k = temperatura_c + 273.15 # Ley combinada de los gases: P1V1/T1 = P2V2/T2 → V2 = V1·(P1/P2)·(T2/T1) volumen_L_estandar = volumen_L_medido * (presion_ambiente_hpa / STANDARD_PRESSURE_HPA) * (STANDARD_TEMP_K / temp_ambiente_k) return volumen_L_medido, volumen_L_estandar, presion_ambiente_hpa def _circunferencias_desde_volumen_estandar(volumen_L_estandar, temperatura_c, altitud_m, k=1.3): """Función inversa, usada SOLO para construir el set de datos sintético base a partir de las capacidades pulmonares originales de la demo. Dado un volumen estándar deseado, calcula qué circunferencias ecuatorial y polar debió tener el globo en esas condiciones, asumiendo una razón fija k = radio_polar / radio_ecuatorial (forma típica de un globo de fiesta).""" presion_amb = presion_estandar_por_altitud(altitud_m) temp_amb_k = temperatura_c + 273.15 volumen_L_medido = volumen_L_estandar / ((presion_amb / STANDARD_PRESSURE_HPA) * (STANDARD_TEMP_K / temp_amb_k)) volumen_cm3 = volumen_L_medido * 1000.0 r_ecuatorial = (3 * volumen_cm3 / (4 * np.pi * k)) ** (1 / 3) r_polar = k * r_ecuatorial return round(2 * np.pi * r_ecuatorial, 1), round(2 * np.pi * r_polar, 1) # ───────────────────────────────────────────── # ESTADO DE ÁNIMO — Modelo Circunflejo del Afecto (Russell, 1980) # ───────────────────────────────────────────── # El estado de ánimo se captura en dos dimensiones independientes: # - Valencia: 1 (muy negativa) a 5 (muy positiva) — agrado/desagrado. # - Activación: 1 (muy baja) a 5 (muy alta) — intensidad fisiológica. # La combinación de ambas clasifica el perfil afectivo en 6 categorías, # siguiendo el criterio: si la valencia es neutra (3) domina el criterio de # "neutro"; si no, y la activación es moderada (3), domina "activación # moderada"; en cualquier otro caso se usa el cuadrante correspondiente. VALENCIA_LABELS = { "1 - Muy negativa": 1, "2 - Negativa": 2, "3 - Neutral": 3, "4 - Positiva": 4, "5 - Muy positiva": 5, } ACTIVACION_LABELS = { "1 - Muy baja": 1, "2 - Baja": 2, "3 - Moderada": 3, "4 - Alta": 4, "5 - Muy alta": 5, } VALENCIA_NOMBRE = {1: "Muy negativa", 2: "Negativa", 3: "Neutral", 4: "Positiva", 5: "Muy positiva"} COLOR_CATEGORIA = { "Positivo alta activación": "#27ae60", "Positivo baja activación": "#3498db", "Neutro": "#95a5a6", "Activación moderada": "#9b59b6", "Negativo alta activación": "#e74c3c", "Negativo baja activación": "#7f8c8d", } def categorizar_estado_animo(valencia, activacion): """Devuelve (categoria, descripcion) según el Modelo Circunflejo del Afecto.""" valencia = int(round(valencia)) activacion = int(round(activacion)) if valencia == 3: return "Neutro", "No hay predominio claro de emociones positivas o negativas." if activacion == 3: return "Activación moderada", "La intensidad emocional permanece en un nivel intermedio." if valencia >= 4 and activacion >= 4: return "Positivo alta activación", "Entusiasmo, motivación y energía elevada." if valencia >= 4 and activacion <= 2: return "Positivo baja activación", "Tranquilidad, relajación, satisfacción y bienestar." if valencia <= 2 and activacion >= 4: return "Negativo alta activación", "Ansiedad, estrés, irritabilidad o enojo." return "Negativo baja activación", "Tristeza, apatía, desmotivación o fatiga emocional." # ───────────────────────────────────────────── # ESQUEMA DE DATOS # ───────────────────────────────────────────── COLUMNS = [ "ID", "Edad", "Talla_m", "Sexo", "Ejercicio", "Circ_Ecu_Antes_cm", "Circ_Pol_Antes_cm", "Circ_Ecu_Despues_cm", "Circ_Pol_Despues_cm", "Temp_Antes_C", "Temp_Despues_C", "Altitud_m", "SpO2_Antes", "SpO2_Despues", "Valencia_Antes", "Activacion_Antes", "Valencia_Despues", "Activacion_Despues", ] COLUMNAS_CAPTURA = COLUMNS + [ "CapPulm_Antes", "CapPulm_Despues", "Delta_CapPulm", "Delta_SpO2", "Delta_Valencia", "Delta_Activacion", "Categoria_Antes", "Categoria_Despues", ] EJERCICIO_LABELS = { "1 - Sedentario (no hace ejercicio)": 1, "2 - Ocasional (1 vez/semana)": 2, "3 - Moderado (2-3 veces/semana)": 3, "4 - Activo (4-5 veces/semana)": 4, "5 - Muy activo (diario)": 5, } EJER_NOMBRE = {1: "Sedentario", 2: "Ocasional", 3: "Moderado", 4: "Activo", 5: "Muy activo"} COLORES_EJER = {1: "#e74c3c", 2: "#e67e22", 3: "#f1c40f", 4: "#2ecc71", 5: "#27ae60"} CSV_PATH = "datos_animo_ejercicio.csv" # Nota metodológica (variable Sexo): # El modelo usa "Sexo" como variable binaria (Mujer=0, Hombre=1) por simplicidad # estadística en esta demostración pedagógica de regresión lineal múltiple con un # grupo pequeño de participantes. No representa una postura sobre identidad de # género; es una simplificación técnica típica de un ejercicio introductorio de # regresión con muestras pequeñas. # Candado de escritura para evitar condiciones de carrera cuando varios # participantes guardan datos casi simultáneamente desde sus celulares (FIX #1). csv_lock = threading.Lock() estado = { "modelo": None, "df": None, "entrenado": False, "r2": None, "coefs": None, "intercept": None, } # ───────────────────────────────────────────── # DATOS BASE SINTÉTICOS (calibración en el salón, antes de la expedición) # ───────────────────────────────────────────── # Se asume que estos 24 registros de calibración se tomaron en Quito # (altitud ≈ 2,850 m, ≈15 °C), ANTES de subir al Chimborazo. Los datos reales # de la expedición (con la altitud y temperatura real del volcán) se capturan # en la Sección 3 de la app y se combinan con estos en el reentrenamiento. _SEED_ROWS = [ # ID, Edad, Talla, Sexo, Ejercicio, CapPulm_Antes(L), CapPulm_Despues(L), SpO2_A, SpO2_D, Animo_A, Animo_D [1, 17, 1.62, 0, 3, 2.8, 3.1, 97, 98, 3, 5], [2, 16, 1.75, 1, 5, 3.5, 3.9, 96, 99, 2, 5], [3, 18, 1.58, 0, 1, 2.2, 2.3, 95, 96, 3, 3], [4, 17, 1.70, 1, 4, 3.2, 3.6, 97, 98, 3, 5], [5, 16, 1.55, 0, 2, 2.5, 2.7, 96, 97, 2, 4], [6, 18, 1.80, 1, 5, 4.0, 4.5, 98, 99, 3, 5], [7, 17, 1.60, 0, 1, 2.1, 2.2, 94, 95, 3, 3], [8, 16, 1.65, 0, 3, 2.7, 3.0, 97, 98, 2, 4], [9, 18, 1.78, 1, 4, 3.6, 4.0, 96, 98, 3, 5], [10, 17, 1.72, 1, 3, 3.3, 3.5, 97, 97, 3, 4], [11, 16, 1.50, 0, 1, 2.0, 2.1, 95, 95, 2, 2], [12, 18, 1.68, 0, 2, 2.6, 2.8, 96, 97, 3, 4], [13, 17, 1.74, 1, 5, 3.8, 4.3, 97, 99, 2, 5], [14, 16, 1.57, 0, 2, 2.3, 2.5, 96, 96, 3, 4], [15, 18, 1.82, 1, 4, 3.9, 4.2, 98, 99, 3, 5], [16, 17, 1.61, 0, 1, 2.1, 2.1, 95, 95, 2, 2], [17, 16, 1.69, 0, 3, 2.8, 3.1, 97, 98, 3, 4], [18, 18, 1.76, 1, 4, 3.5, 3.8, 96, 98, 2, 5], [19, 17, 1.63, 0, 2, 2.4, 2.6, 96, 97, 3, 4], [20, 16, 1.71, 1, 3, 3.1, 3.3, 97, 98, 3, 4], [21, 18, 1.59, 0, 1, 2.0, 2.0, 94, 94, 2, 2], [22, 17, 1.77, 1, 5, 4.1, 4.6, 98, 99, 3, 5], [23, 16, 1.64, 0, 2, 2.5, 2.7, 96, 97, 2, 3], [24, 18, 1.73, 1, 4, 3.4, 3.7, 97, 98, 3, 5], ] _ALTITUD_BASE = 2850.0 # Quito, m s.n.m. _TEMP_BASE = 15.0 # °C promedio de Quito def _construir_base_data(): filas = [] for (id_, edad, talla, sexo, ejercicio, cap_a, cap_d, spo2_a, spo2_d, animo_a, animo_d) in _SEED_ROWS: ce_a, cp_a = _circunferencias_desde_volumen_estandar(cap_a, _TEMP_BASE, _ALTITUD_BASE) ce_d, cp_d = _circunferencias_desde_volumen_estandar(cap_d, _TEMP_BASE, _ALTITUD_BASE) activ_a = min(5, max(1, ejercicio)) activ_d = min(5, max(1, ejercicio + 1)) filas.append([ id_, edad, talla, sexo, ejercicio, ce_a, cp_a, ce_d, cp_d, _TEMP_BASE, _TEMP_BASE, _ALTITUD_BASE, spo2_a, spo2_d, animo_a, activ_a, animo_d, activ_d, ]) return filas BASE_DATA = _construir_base_data() # ───────────────────────────────────────────── # LÓGICA DEL MODELO # ───────────────────────────────────────────── def calcular_deltas(df): df = df.copy() cap_a_list, cap_d_list = [], [] for _, row in df.iterrows(): _, v_a, _ = calcular_capacidad_pulmonar_globo( row["Circ_Ecu_Antes_cm"], row["Circ_Pol_Antes_cm"], row["Temp_Antes_C"], row["Altitud_m"]) _, v_d, _ = calcular_capacidad_pulmonar_globo( row["Circ_Ecu_Despues_cm"], row["Circ_Pol_Despues_cm"], row["Temp_Despues_C"], row["Altitud_m"]) cap_a_list.append(v_a) cap_d_list.append(v_d) df["CapPulm_Antes"] = cap_a_list df["CapPulm_Despues"] = cap_d_list df["Delta_CapPulm"] = df["CapPulm_Despues"] - df["CapPulm_Antes"] df["Delta_SpO2"] = df["SpO2_Despues"] - df["SpO2_Antes"] df["Delta_Valencia"] = df["Valencia_Despues"] - df["Valencia_Antes"] df["Delta_Activacion"] = df["Activacion_Despues"] - df["Activacion_Antes"] df["Categoria_Antes"] = df.apply(lambda r: categorizar_estado_animo(r["Valencia_Antes"], r["Activacion_Antes"])[0], axis=1) df["Categoria_Despues"] = df.apply(lambda r: categorizar_estado_animo(r["Valencia_Despues"], r["Activacion_Despues"])[0], axis=1) return df def entrenar_modelo(df): df_delta = calcular_deltas(df) X = df_delta[["Ejercicio", "Delta_CapPulm", "Delta_SpO2", "Edad", "Talla_m", "Sexo"]].values y = df_delta["Delta_Valencia"].values modelo = LinearRegression().fit(X, y) r2 = r2_score(y, modelo.predict(X)) return modelo, r2, df_delta def _construir_salidas_modelo(df, df_delta, modelo, r2, fuente): estado["modelo"] = modelo estado["df"] = df_delta estado["entrenado"] = True estado["r2"] = r2 estado["coefs"] = modelo.coef_ estado["intercept"] = modelo.intercept_ c = modelo.coef_ b0 = modelo.intercept_ ecuacion = ( f"**Ecuación del modelo ({fuente}):**\n\n" f"`ΔValencia = {b0:.4f}`\n" f"` + {c[0]:.4f} × Ejercicio`\n" f"` + {c[1]:.4f} × ΔCapacidadPulmonar (L, estandarizada)`\n" f"` + {c[2]:.4f} × ΔOximetría`\n" f"` + {c[3]:.4f} × Edad`\n" f"` + {c[4]:.4f} × Talla`\n" f"` + {c[5]:.4f} × Sexo`\n\n" f"**R² = {r2:.4f}** — El modelo explica el **{r2*100:.1f}%** de la varianza en el cambio de " f"valencia (agrado/desagrado) *dentro de esta muestra* (demostración pedagógica, no un " f"resultado inferencial generalizable).\n\n" f"**Nota sobre 'Sexo':** se codifica como variable binaria (Mujer=0, Hombre=1) por simplicidad " f"estadística de esta demo introductoria; no busca representar la diversidad de identidad de género.\n\n" f"**Nota sobre ΔCapacidadPulmonar:** proviene del volumen del globo (elipsoide, a partir de las " f"circunferencias ecuatorial y polar), normalizado por temperatura y presión atmosférica " f"(estimada por altitud) a una condición estándar de referencia — así una medición en el " f"Chimborazo es comparable con una tomada a nivel del mar.\n\n" f"**Interpretación de β₁ (Ejercicio) = {c[0]:.4f}:**\n" f"Por cada nivel adicional de ejercicio, la valencia cambia en promedio " f"{'↑' if c[0] > 0 else '↓'} {abs(c[0]):.4f} puntos Likert dentro de esta muestra, " f"manteniendo fijas las demás variables del modelo." ) coef_md = ( "| Variable | Coeficiente (β) |\n" "|---|---|\n" f"| Constante (β₀) | {b0:.4f} |\n" f"| Ejercicio (β₁) | {c[0]:.4f} |\n" f"| ΔCap. Pulmonar estandarizada (β₂) | {c[1]:.4f} |\n" f"| ΔOximetría (β₃) | {c[2]:.4f} |\n" f"| Edad (β₄) | {c[3]:.4f} |\n" f"| Talla (β₅) | {c[4]:.4f} |\n" f"| Sexo (β₆, binaria: M=0/H=1) | {c[5]:.4f} |\n" f"| **R²** | **{r2:.4f}** |" ) tabla = df_delta[[ "ID", "Edad", "Talla_m", "Sexo", "Ejercicio", "Altitud_m", "CapPulm_Antes", "CapPulm_Despues", "Delta_CapPulm", "SpO2_Antes", "SpO2_Despues", "Delta_SpO2", "Valencia_Antes", "Activacion_Antes", "Categoria_Antes", "Valencia_Despues", "Activacion_Despues", "Categoria_Despues", "Delta_Valencia", "Delta_Activacion", ]] return ( f"Modelo entrenado con {len(df)} participantes. Fuente: {fuente}", len(df), tabla, ecuacion, coef_md, ) def _csv_capturado_a_df(): if not os.path.exists(CSV_PATH): return None, "No hay datos capturados en el CSV todavía." try: with csv_lock: df_csv = pd.read_csv(CSV_PATH) if df_csv.empty: return None, "El CSV está vacío." missing = [c for c in COLUMNS if c not in df_csv.columns] if missing: return None, f"Al CSV le faltan columnas: {missing}" df_out = df_csv[COLUMNS].copy() for col in COLUMNS[1:]: df_out[col] = pd.to_numeric(df_out[col], errors="coerce") df_out = df_out.dropna() if len(df_out) < 3: return None, f"Se necesitan al menos 3 participantes para entrenar. Hay {len(df_out)}." return df_out, None except Exception as e: return None, f"Error leyendo el CSV: {e}" def cargar_y_entrenar(file_obj=None): df = pd.DataFrame(BASE_DATA, columns=COLUMNS) if file_obj is not None: try: ext = os.path.splitext(file_obj.name)[1].lower() df_new = pd.read_csv(file_obj.name) if ext == ".csv" else pd.read_excel(file_obj.name) if all(c in df_new.columns for c in COLUMNS): df_new = df_new[COLUMNS].copy() else: df_new = df_new.iloc[:, :len(COLUMNS)] df_new.columns = COLUMNS[:df_new.shape[1]] for col in df_new.columns: df_new[col] = pd.to_numeric(df_new[col], errors="coerce") df_new = df_new.dropna() df = pd.concat([df, df_new], ignore_index=True) fuente = f"datos base (24 reg.) + archivo subido ({len(df_new)} reg.)" except Exception as e: return f"Error al procesar el archivo: {e}", 0, pd.DataFrame(), "", "" else: fuente = "datos base únicamente" modelo, r2, df_delta = entrenar_modelo(df) return _construir_salidas_modelo(df, df_delta, modelo, r2, fuente) def reentrenar_con_csv(): df_base = pd.DataFrame(BASE_DATA, columns=COLUMNS) df_csv, err = _csv_capturado_a_df() if err: return ( f"Error: {err}", len(df_base), pd.DataFrame(), "— Entrena primero el modelo —", "", ) df_csv = df_csv.reindex(columns=COLUMNS) df_combined = pd.concat([df_base, df_csv], ignore_index=True) modelo, r2, df_delta = entrenar_modelo(df_combined) fuente = f"datos base (24 reg.) + CSV capturado ({len(df_csv)} reg.)" return _construir_salidas_modelo(df_combined, df_delta, modelo, r2, fuente) # ───────────────────────────────────────────── # PREDICCIÓN # ───────────────────────────────────────────── def predecir(ejercicio_label, edad, talla, sexo, circ_ec_antes, circ_pol_antes, temp_antes, circ_ec_despues, circ_pol_despues, temp_despues, altitud_m, spo2_antes, spo2_despues): if not estado["entrenado"]: return "El modelo no ha sido entrenado. Haz clic en 'Entrenar'." ejer_val = EJERCICIO_LABELS.get(ejercicio_label, 1) sexo_val = 1 if sexo == "Hombre" else 0 v_med_a, v_std_a, p_amb = calcular_capacidad_pulmonar_globo(circ_ec_antes, circ_pol_antes, temp_antes, altitud_m) v_med_d, v_std_d, _ = calcular_capacidad_pulmonar_globo(circ_ec_despues, circ_pol_despues, temp_despues, altitud_m) delta_cap = v_std_d - v_std_a delta_spo2 = spo2_despues - spo2_antes X_new = np.array([[ejer_val, delta_cap, delta_spo2, edad, talla, sexo_val]]) pred = estado["modelo"].predict(X_new)[0] c = estado["coefs"] b0 = estado["intercept"] if pred >= 1.5: interp = "Mejora notable en la valencia (agrado) — se espera un cambio hacia un estado más positivo." elif pred >= 0.5: interp = "Ligera mejora en la valencia — cambio positivo moderado." elif pred >= -0.5: interp = "Sin cambio relevante en la valencia." else: interp = "Posible descenso en la valencia — revisar variables de contexto." direccion = ( "El coeficiente de Ejercicio es positivo en este modelo: dentro de esta " "muestra, mayor nivel de ejercicio se asocia con mayor ΔValencia." if c[0] > 0 else "El coeficiente de Ejercicio es negativo en este modelo: dentro de esta " "muestra, mayor nivel de ejercicio se asocia con menor ΔValencia." ) return ( f"## Resultado de la Predicción\n\n" f"**Perfil ingresado:**\n" f"- Ejercicio: {ejercicio_label} → valor numérico = {ejer_val}\n" f"- Edad: {edad} años | Talla: {talla} m | Sexo: {sexo}\n" f"- Altitud: {altitud_m:.0f} m → presión atmosférica estimada: **{p_amb:.1f} hPa**\n\n" f"**Capacidad pulmonar (globo, elipsoide):**\n" f"- Antes: medida en campo = {v_med_a:.2f} L → estandarizada (37°C, nivel del mar) = **{v_std_a:.2f} L**\n" f"- Después: medida en campo = {v_med_d:.2f} L → estandarizada = **{v_std_d:.2f} L**\n" f"- ΔCapacidad pulmonar estandarizada: **{delta_cap:+.2f} L**\n" f"- Oximetría: {spo2_antes}% → {spo2_despues}% (Δ = {delta_spo2:+.1f}%)\n\n" f"---\n\n" f"**ΔValencia predicho: `{pred:+.3f}` puntos Likert** (dimensión de agrado/desagrado del Modelo " f"Circunflejo del Afecto)\n\n" f"{interp}\n\n" f"---\n\n" f"**Ecuación aplicada:**\n" f"`ΔValencia = {b0:.4f} + {c[0]:.4f}×{ejer_val} + {c[1]:.4f}×{delta_cap:.2f} " f"+ {c[2]:.4f}×{delta_spo2:.1f} + {c[3]:.4f}×{edad} " f"+ {c[4]:.4f}×{talla} + {c[5]:.4f}×{sexo_val}`\n\n" f"**R² del modelo = {estado['r2']:.4f}** (ajuste dentro de esta muestra, no una medida " f"de significancia estadística)\n\n" f"**Dirección observada del efecto:** {direccion}\n\n" f"*Esta es una demostración pedagógica del funcionamiento de un modelo de regresión, " f"no una prueba de hipótesis formal (no incluye p-valores ni pruebas de significancia). " f"Predice el cambio en la dimensión de VALENCIA; no predice la activación.*" ) # ───────────────────────────────────────────── # CAPTURA DE PARTICIPANTES # ───────────────────────────────────────────── def obtener_tabla_csv(): with csv_lock: if os.path.exists(CSV_PATH): try: return pd.read_csv(CSV_PATH) except Exception: pass return pd.DataFrame(columns=COLUMNAS_CAPTURA) def limpiar_csv(): with csv_lock: if os.path.exists(CSV_PATH): os.remove(CSV_PATH) return "Registros eliminados.", pd.DataFrame(columns=COLUMNAS_CAPTURA) def guardar_participante(edad, talla, sexo, ejercicio_label, circ_ec_antes, circ_pol_antes, temp_antes, circ_ec_despues, circ_pol_despues, temp_despues, altitud_m, spo2_antes, spo2_despues, valencia_antes_label, activacion_antes_label, valencia_despues_label, activacion_despues_label): errores = [] if not (10 <= edad <= 100): errores.append("Edad fuera de rango (10-100).") if not (1.0 <= talla <= 2.5): errores.append("Talla fuera de rango (1.0-2.5 m).") if not (50 <= spo2_antes <= 100): errores.append("SpO2 Antes fuera de rango (50-100).") if not (50 <= spo2_despues <= 100): errores.append("SpO2 Después fuera de rango (50-100).") if not (10 <= circ_ec_antes <= 150): errores.append("Circunferencia ecuatorial Antes fuera de rango (10-150 cm).") if not (10 <= circ_pol_antes <= 150):errores.append("Circunferencia polar Antes fuera de rango (10-150 cm).") if not (10 <= circ_ec_despues <= 150): errores.append("Circunferencia ecuatorial Después fuera de rango (10-150 cm).") if not (10 <= circ_pol_despues <= 150): errores.append("Circunferencia polar Después fuera de rango (10-150 cm).") if not (-40 <= temp_antes <= 50): errores.append("Temperatura Antes fuera de rango (-40 a 50°C).") if not (-40 <= temp_despues <= 50): errores.append("Temperatura Después fuera de rango (-40 a 50°C).") if not (0 <= altitud_m <= 9000): errores.append("Altitud fuera de rango (0-9000 m).") if errores: return "Error: " + " | ".join(errores), obtener_tabla_csv() sexo_val = 1 if sexo == "Hombre" else 0 ejer_val = EJERCICIO_LABELS.get(ejercicio_label, 1) val_a = VALENCIA_LABELS.get(valencia_antes_label, 3) act_a = ACTIVACION_LABELS.get(activacion_antes_label, 3) val_d = VALENCIA_LABELS.get(valencia_despues_label, 3) act_d = ACTIVACION_LABELS.get(activacion_despues_label, 3) v_med_a, v_std_a, _ = calcular_capacidad_pulmonar_globo(circ_ec_antes, circ_pol_antes, temp_antes, altitud_m) v_med_d, v_std_d, _ = calcular_capacidad_pulmonar_globo(circ_ec_despues, circ_pol_despues, temp_despues, altitud_m) cat_a, _ = categorizar_estado_animo(val_a, act_a) cat_d, _ = categorizar_estado_animo(val_d, act_d) with csv_lock: siguiente_id = 1 if os.path.exists(CSV_PATH): try: df_ex = pd.read_csv(CSV_PATH) if not df_ex.empty and "ID" in df_ex.columns: siguiente_id = int(df_ex["ID"].max()) + 1 except Exception: pass nueva_fila = { "ID": siguiente_id, "Edad": int(edad), "Talla_m": talla, "Sexo": sexo_val, "Ejercicio": ejer_val, "Circ_Ecu_Antes_cm": circ_ec_antes, "Circ_Pol_Antes_cm": circ_pol_antes, "Circ_Ecu_Despues_cm": circ_ec_despues, "Circ_Pol_Despues_cm": circ_pol_despues, "Temp_Antes_C": temp_antes, "Temp_Despues_C": temp_despues, "Altitud_m": altitud_m, "SpO2_Antes": spo2_antes, "SpO2_Despues": spo2_despues, "Valencia_Antes": val_a, "Activacion_Antes": act_a, "Valencia_Despues": val_d, "Activacion_Despues": act_d, "CapPulm_Antes": round(v_std_a, 3), "CapPulm_Despues": round(v_std_d, 3), "Delta_CapPulm": round(v_std_d - v_std_a, 3), "Delta_SpO2": round(spo2_despues - spo2_antes, 1), "Delta_Valencia": val_d - val_a, "Delta_Activacion": act_d - act_a, "Categoria_Antes": cat_a, "Categoria_Despues": cat_d, } df_nuevo = pd.DataFrame([nueva_fila], columns=COLUMNAS_CAPTURA) try: modo = "a" if os.path.exists(CSV_PATH) else "w" with open(CSV_PATH, mode=modo, newline="", encoding="utf-8") as f: df_nuevo.to_csv(f, header=(modo == "w"), index=False) tabla_actual = pd.read_csv(CSV_PATH) except Exception as e: print(f"Error guardando CSV: {e}") return f"Error al guardar: {e}", pd.read_csv(CSV_PATH) if os.path.exists(CSV_PATH) else pd.DataFrame(columns=COLUMNAS_CAPTURA) return ( f"Participante {nueva_fila['ID']} guardado. " f"Cap. pulmonar estandarizada: {v_std_a:.2f} L → {v_std_d:.2f} L. " f"Ánimo: {cat_a} → {cat_d}.", tabla_actual ) # ───────────────────────────────────────────── # GRÁFICAS # ───────────────────────────────────────────── def _buf_a_tempfile(buf): if buf is None: return None tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".png") tmp.write(buf.read()) tmp.flush() tmp.close() return tmp.name def generar_graficas(): VACIO = (None, None, None, "No hay datos capturados todavía. Registra participantes en la Sección 3.") if not os.path.exists(CSV_PATH): return VACIO try: with csv_lock: df = pd.read_csv(CSV_PATH) except Exception as e: return None, None, None, f"Error leyendo CSV: {e}" if df.empty or "Valencia_Antes" not in df.columns: return VACIO df = df.dropna(subset=["Valencia_Antes", "Valencia_Despues", "Delta_Valencia", "Ejercicio"]) if len(df) < 2: return None, None, None, "Se necesitan al menos 2 participantes para graficar." def save_fig(fig): buf = io.BytesIO() fig.savefig(buf, format="png", dpi=130, bbox_inches="tight") buf.seek(0) plt.close(fig) return _buf_a_tempfile(buf) # Gráfica 1: Barras Antes vs Después (Valencia) por participante fig1, ax1 = plt.subplots(figsize=(max(8, len(df) * 0.65), 5)) ids = [f"P{int(i)}" for i in df["ID"]] x, w = range(len(df)), 0.35 ax1.bar([i - w / 2 for i in x], df["Valencia_Antes"], w, label="Antes", color="#5b8dd9", alpha=0.85) bars_d = ax1.bar([i + w / 2 for i in x], df["Valencia_Despues"], w, label="Después", color="#27ae60", alpha=0.85) for bar, delta in zip(bars_d, df["Delta_Valencia"]): col = "#27ae60" if delta > 0 else ("#e74c3c" if delta < 0 else "#888888") ax1.text(bar.get_x() + bar.get_width() / 2, bar.get_height() + 0.06, f"{delta:+.0f}", ha="center", va="bottom", fontsize=8, color=col, fontweight="bold") ax1.set_xticks(list(x)) ax1.set_xticklabels(ids, rotation=45, ha="right", fontsize=8) ax1.set_yticks([1, 2, 3, 4, 5]) ax1.set_yticklabels([f"{v} - {VALENCIA_NOMBRE[v]}" for v in [1, 2, 3, 4, 5]], fontsize=8) ax1.set_ylim(0, 6.2) ax1.set_title("Valencia (agrado): Antes vs Después por Participante", fontsize=13, fontweight="bold", pad=12) ax1.set_xlabel("Participante") ax1.set_ylabel("Valencia (Likert 1-5)") ax1.legend(loc="upper left") ax1.spines[["top", "right"]].set_visible(False) ax1.grid(axis="y", alpha=0.3) fig1.tight_layout() path1 = save_fig(fig1) # Gráfica 2: Boxplot ΔValencia por nivel de ejercicio fig2, ax2 = plt.subplots(figsize=(7, 5)) niveles = sorted(df["Ejercicio"].dropna().unique().astype(int)) data_box = [df[df["Ejercicio"] == n]["Delta_Valencia"].values for n in niveles] etiquetas = [f"{n}\n{EJER_NOMBRE.get(n, '')}" for n in niveles] bp = ax2.boxplot(data_box, patch_artist=True, widths=0.45, medianprops=dict(color="black", linewidth=2)) for patch, niv in zip(bp["boxes"], niveles): patch.set_facecolor(COLORES_EJER.get(niv, "#aaaaaa")) patch.set_alpha(0.72) for i, (niv, datos) in enumerate(zip(niveles, data_box), 1): jitter = np.random.uniform(-0.15, 0.15, size=len(datos)) ax2.scatter([i + j for j in jitter], datos, color=COLORES_EJER.get(niv, "#aaaaaa"), s=55, zorder=5, edgecolors="white", linewidths=0.6) ax2.axhline(0, color="#e74c3c", linestyle="--", linewidth=1.2, alpha=0.7, label="Sin cambio (Δ = 0)") ax2.set_xticks(range(1, len(niveles) + 1)) ax2.set_xticklabels(etiquetas, fontsize=9) ax2.set_title("ΔValencia según Nivel de Ejercicio", fontsize=13, fontweight="bold", pad=12) ax2.set_xlabel("Nivel de Ejercitación") ax2.set_ylabel("ΔValencia (puntos Likert)") ax2.legend(fontsize=9) ax2.spines[["top", "right"]].set_visible(False) ax2.grid(axis="y", alpha=0.3) fig2.tight_layout() path2 = save_fig(fig2) # Gráfica 3: Dona con la distribución de categorías de ánimo (Después) fig3, ax3 = plt.subplots(figsize=(6.5, 5)) conteo = df["Categoria_Despues"].value_counts() vals = conteo.values.tolist() labels_pie = [f"{cat}\n{n} ({n/conteo.sum()*100:.0f}%)" for cat, n in conteo.items()] colors = [COLOR_CATEGORIA.get(cat, "#aaaaaa") for cat in conteo.index] wedges, _ = ax3.pie(vals, colors=colors, startangle=90, wedgeprops=dict(width=0.55, edgecolor="white", linewidth=2)) ax3.legend(wedges, labels_pie, loc="lower center", bbox_to_anchor=(0.5, -0.35), ncol=min(2, len(vals)), fontsize=8) ax3.set_title("Categorías de Ánimo Después (Modelo Circunflejo)", fontsize=13, fontweight="bold", pad=12) ax3.text(0, 0, f"n={conteo.sum()}", ha="center", va="center", fontsize=14, fontweight="bold", color="#333333") fig3.tight_layout() path3 = save_fig(fig3) total = len(df) mejora = int((df["Delta_Valencia"] > 0).sum()) igual = int((df["Delta_Valencia"] == 0).sum()) descenso = int((df["Delta_Valencia"] < 0).sum()) pct = mejora / total * 100 if total > 0 else 0 resumen = ( f"Resumen: {total} participantes analizados — " f"**{mejora} mejoraron su valencia** ({pct:.0f}%), " f"{igual} sin cambio, {descenso} con descenso. " f"ΔValencia promedio: **{df['Delta_Valencia'].mean():+.2f} puntos**." ) return path1, path2, path3, resumen def guardar_y_graficar(edad, talla, sexo, ejercicio_label, circ_ec_antes, circ_pol_antes, temp_antes, circ_ec_despues, circ_pol_despues, temp_despues, altitud_m, spo2_antes, spo2_despues, valencia_antes_label, activacion_antes_label, valencia_despues_label, activacion_despues_label): msg, tabla = guardar_participante( edad, talla, sexo, ejercicio_label, circ_ec_antes, circ_pol_antes, temp_antes, circ_ec_despues, circ_pol_despues, temp_despues, altitud_m, spo2_antes, spo2_despues, valencia_antes_label, activacion_antes_label, valencia_despues_label, activacion_despues_label ) p1, p2, p3, resumen = generar_graficas() return msg, tabla, resumen, p1, p2, p3 # ───────────────────────────────────────────── # INICIALIZACIÓN # ───────────────────────────────────────────── init_msg, init_count, init_df, init_ec, init_coef = cargar_y_entrenar() # ───────────────────────────────────────────── # INTERFAZ # ───────────────────────────────────────────── with gr.Blocks(title="Signos Vitales de Misión | SEEC 2027 | HEJAU") as app: gr.Markdown("# Experimento: ¿Cómo afecta el ejercicio físico al estado de ánimo de un astronauta?") gr.Markdown( "**¿Cómo sabes si un ser humano está listo para una misión extrema?**\n\n" "**Pregunta exploratoria:** ¿El nivel de ejercicio físico se asocia con el cambio en la " "*valencia* del estado de ánimo (Modelo Circunflejo del Afecto, Russell, 1980), dentro de una " "misión de altura extrema, al controlar capacidad pulmonar (estandarizada por temperatura y " "presión atmosférica), oximetría, edad, talla y sexo?\n\n" "**Variable dependiente:** ΔValencia = Valencia_después − Valencia_antes (escala Likert 1-5)\n\n" "**Práctica en campo:** Volcán Chimborazo (Ecuador). La capacidad pulmonar se estima inflando " "un globo y midiendo su circunferencia ecuatorial y polar (se modela como un elipsoide), y el " "volumen se normaliza por la temperatura ambiente y la presión atmosférica de la altitud del " "punto de medición.\n\n" "*Esta app es una demostración pedagógica de un modelo de regresión lineal múltiple, " "pensada para que docentes vean el flujo completo de datos → modelo → predicción. " "No sustituye un análisis estadístico formal con muestras grandes ni una espirometría clínica.*" ) # SECCIÓN 1: MODELO gr.Markdown("---") gr.Markdown("## 1. Modelo de Regresión Lineal Múltiple") with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Carga de datos") file_input = gr.File(label="Subir CSV/Excel externo (opcional)", file_types=[".csv", ".xlsx"]) load_btn = gr.Button("Entrenar con datos base (+ archivo si se subió)", variant="primary") gr.Markdown("---") gr.Markdown( "**¿Ya capturaste participantes?**\n\n" "Usa el botón de abajo para reentrenar el modelo con los datos base " "**más** todos los participantes registrados en la Sección 3." ) retrain_btn = gr.Button("Reentrenar con datos base + CSV capturado", variant="secondary") status_out = gr.Textbox(label="Estado del modelo", value=init_msg, interactive=False) n_part_out = gr.Number(label="Total de participantes usados", value=init_count, interactive=False) with gr.Column(scale=2): gr.Markdown("### Ecuación y coeficientes") ecuacion_out = gr.Markdown(value=init_ec) coef_out = gr.Markdown(value=init_coef) gr.Markdown("### Datos de entrenamiento") data_table = gr.DataFrame(value=init_df, label="Conjunto de datos (con deltas y categorías calculadas)", interactive=False, wrap=True) load_btn.click( cargar_y_entrenar, inputs=file_input, outputs=[status_out, n_part_out, data_table, ecuacion_out, coef_out] ) retrain_btn.click( reentrenar_con_csv, inputs=None, outputs=[status_out, n_part_out, data_table, ecuacion_out, coef_out] ) # SECCIÓN 2: PREDICCIÓN gr.Markdown("---") gr.Markdown("## 2. Realizar una Predicción") gr.Markdown("Ingresa el perfil de un participante para predecir su cambio esperado en la valencia del ánimo.") with gr.Row(): with gr.Column(): gr.Markdown("#### Variables del perfil") ejercicio_pred = gr.Dropdown( choices=list(EJERCICIO_LABELS.keys()), label="Nivel de ejercitación habitual", value=list(EJERCICIO_LABELS.keys())[2] ) with gr.Row(): edad_pred = gr.Number(label="Edad (años)", value=17, step=1) talla_pred = gr.Number(label="Talla (m)", value=1.65, step=0.01) sexo_pred = gr.Radio(["Mujer", "Hombre"], label="Sexo", info="Variable binaria simplificada para esta demo estadística.", value="Mujer") altitud_pred = gr.Number(label="Altitud del punto de medición (m s.n.m.)", value=5000, step=50, info="Ej. Chimborazo (refugio/zona de práctica) ≈ 5000 m") gr.Markdown("#### Globo — Capacidad pulmonar (elipsoide) ANTES") with gr.Row(): ce_a_pred = gr.Number(label="Circunf. ecuatorial ANTES (cm)", value=55.0, step=0.5) cp_a_pred = gr.Number(label="Circunf. polar ANTES (cm)", value=70.0, step=0.5) temp_a_pred = gr.Number(label="Temperatura ambiente ANTES (°C)", value=5.0, step=0.5) gr.Markdown("#### Globo — Capacidad pulmonar (elipsoide) DESPUÉS") with gr.Row(): ce_d_pred = gr.Number(label="Circunf. ecuatorial DESPUÉS (cm)", value=60.0, step=0.5) cp_d_pred = gr.Number(label="Circunf. polar DESPUÉS (cm)", value=76.0, step=0.5) temp_d_pred = gr.Number(label="Temperatura ambiente DESPUÉS (°C)", value=5.0, step=0.5) gr.Markdown("#### Oximetría") with gr.Row(): spo2_a_pred = gr.Number(label="Oximetría ANTES (%)", value=90.0, step=0.5) spo2_d_pred = gr.Number(label="Oximetría DESPUÉS (%)", value=92.0, step=0.5) btn_pred = gr.Button("Calcular ΔValencia Predicho", variant="primary") with gr.Column(): salida_pred = gr.Markdown(label="Resultado") btn_pred.click( predecir, inputs=[ejercicio_pred, edad_pred, talla_pred, sexo_pred, ce_a_pred, cp_a_pred, temp_a_pred, ce_d_pred, cp_d_pred, temp_d_pred, altitud_pred, spo2_a_pred, spo2_d_pred], outputs=salida_pred ) # SECCIÓN 3: CAPTURA gr.Markdown("---") gr.Markdown("## 3. Registro de Participantes en Campo") gr.Markdown( "Captura los datos de cada participante antes y después de la actividad física, " "midiendo el globo con cinta métrica y anotando la temperatura ambiente y la altitud." ) with gr.Row(): with gr.Column(): gr.Markdown("#### Datos generales") with gr.Row(): edad_cap = gr.Number(label="Edad (años)", value=17, step=1, minimum=10, maximum=100) talla_cap = gr.Number(label="Talla (m)", value=1.65, step=0.01, minimum=1.0, maximum=2.5) sexo_cap = gr.Radio(["Mujer", "Hombre"], label="Sexo", info="Variable binaria simplificada para esta demo estadística.", value="Mujer") ejercicio_cap = gr.Dropdown( choices=list(EJERCICIO_LABELS.keys()), label="Nivel de ejercitación habitual", value=list(EJERCICIO_LABELS.keys())[0] ) altitud_cap = gr.Number(label="Altitud del punto de medición (m s.n.m.)", value=5000, step=50, info="Misma altitud para Antes y Después (mismo punto de la ruta).") gr.Markdown("#### Globo — Capacidad pulmonar (elipsoide): mide circunferencia ecuatorial y polar") with gr.Row(): ce_a_cap = gr.Number(label="Circunf. ecuatorial ANTES (cm)", value=50.0, step=0.5, minimum=10, maximum=150) cp_a_cap = gr.Number(label="Circunf. polar ANTES (cm)", value=65.0, step=0.5, minimum=10, maximum=150) temp_a_cap = gr.Number(label="Temperatura ANTES (°C)", value=5.0, step=0.5, minimum=-40, maximum=50) with gr.Row(): ce_d_cap = gr.Number(label="Circunf. ecuatorial DESPUÉS (cm)", value=55.0, step=0.5, minimum=10, maximum=150) cp_d_cap = gr.Number(label="Circunf. polar DESPUÉS (cm)", value=70.0, step=0.5, minimum=10, maximum=150) temp_d_cap = gr.Number(label="Temperatura DESPUÉS (°C)", value=5.0, step=0.5, minimum=-40, maximum=50) gr.Markdown("#### Oximetría (ANTES y DESPUÉS)") with gr.Row(): spo2_a_cap = gr.Number(label="SpO2 ANTES (%)", value=92.0, step=0.5, minimum=50, maximum=100) spo2_d_cap = gr.Number(label="SpO2 DESPUÉS (%)", value=90.0, step=0.5, minimum=50, maximum=100) gr.Markdown("#### Estado de ánimo — Valencia y Activación (Modelo Circunflejo, Russell 1980)") with gr.Row(): valencia_a_cap = gr.Dropdown(choices=list(VALENCIA_LABELS.keys()), label="Valencia ANTES", value=list(VALENCIA_LABELS.keys())[2]) activacion_a_cap = gr.Dropdown(choices=list(ACTIVACION_LABELS.keys()), label="Activación ANTES", value=list(ACTIVACION_LABELS.keys())[2]) with gr.Row(): valencia_d_cap = gr.Dropdown(choices=list(VALENCIA_LABELS.keys()), label="Valencia DESPUÉS", value=list(VALENCIA_LABELS.keys())[3]) activacion_d_cap = gr.Dropdown(choices=list(ACTIVACION_LABELS.keys()), label="Activación DESPUÉS", value=list(ACTIVACION_LABELS.keys())[3]) with gr.Row(): btn_guardar = gr.Button("Guardar Participante", variant="primary") btn_limpiar = gr.Button("Borrar Todos los Registros") msg_cap = gr.Textbox(label="Estado del registro", interactive=False) with gr.Column(): gr.Markdown("#### Participantes capturados") tabla_cap = gr.DataFrame( value=obtener_tabla_csv(), label="Registros guardados", interactive=False, wrap=True ) btn_dl = gr.Button("Descargar CSV") archivo_dl = gr.File(label="Archivo generado", visible=False) # SECCIÓN 4: GRÁFICAS gr.Markdown("---") gr.Markdown("## 4. Gráficas de Mejora en el Estado de Ánimo") gr.Markdown( "Las gráficas se actualizan automáticamente al guardar cada participante. " "También puedes generarlas manualmente con el botón." ) btn_graficas = gr.Button("Generar / Actualizar Gráficas", variant="secondary") resumen_graf = gr.Markdown() with gr.Row(): graf1 = gr.Image(label="Valencia: Antes vs Después por participante", type="filepath") graf2 = gr.Image(label="ΔValencia por nivel de ejercicio", type="filepath") graf3 = gr.Image(label="Categorías de ánimo (Después) — Modelo Circunflejo", type="filepath") # EVENTOS def descargar_csv(): if os.path.exists(CSV_PATH): try: with csv_lock: df = pd.read_csv(CSV_PATH) if not df.empty: tmp = tempfile.NamedTemporaryFile( delete=False, suffix=".csv", prefix="datos_animo_ejercicio_" ) df.to_csv(tmp.name, index=False) tmp.close() return gr.File(value=tmp.name, visible=True) except Exception as e: print(f"Error descarga: {e}") return gr.File(visible=False) btn_guardar.click( guardar_y_graficar, inputs=[edad_cap, talla_cap, sexo_cap, ejercicio_cap, ce_a_cap, cp_a_cap, temp_a_cap, ce_d_cap, cp_d_cap, temp_d_cap, altitud_cap, spo2_a_cap, spo2_d_cap, valencia_a_cap, activacion_a_cap, valencia_d_cap, activacion_d_cap], outputs=[msg_cap, tabla_cap, resumen_graf, graf1, graf2, graf3] ) btn_limpiar.click( limpiar_csv, inputs=None, outputs=[msg_cap, tabla_cap] ) btn_dl.click( descargar_csv, inputs=None, outputs=archivo_dl ) btn_graficas.click( generar_graficas, inputs=None, outputs=[graf1, graf2, graf3, resumen_graf] ) gr.Markdown("---") gr.Markdown( "*Basado en: Ruiz Cruz, E. H. (2026). Pensamiento científico auténtico mediante STEAM, " "u-learning y modelos predictivos en educación técnica. CONALEP. Clasificación del estado " "de ánimo según el Modelo Circunflejo del Afecto (Russell, 1980).*" ) if __name__ == "__main__": # concurrency_count limita cuántos eventos corren en paralelo dentro de la # cola de Gradio; junto con csv_lock, evita que 25-30 celulares guardando # casi al mismo tiempo corrompan el archivo CSV compartido. app.queue(default_concurrency_limit=4) app.launch(server_name="0.0.0.0", server_port=7860)