from flask import Flask, jsonify, request from flask_cors import CORS import sqlite3 import os import time import threading import subprocess # NUEVO: Librería para ejecutar comandos en la terminal app = Flask(__name__) CORS(app) # Permite que la página web consulte este servidor RUTA_ACTUAL = "/dev/shm/aql_actual.txt" RUTA_BD = "/home/pi/microestacion.db" # --- LÓGICA DE LA NOM-172-SEMARNAT-2019 --- def calcular_ica_individual(cp, gas): # Tablas de cortes (Breakpoints) según la NOM-172 # Estructura: [BP_low, BP_high, Index_low, Index_high] breakpoints = { "O3": [(0.0, 0.051, 0, 50), (0.052, 0.070, 51, 100), (0.071, 0.092, 101, 150), (0.093, 0.114, 151, 200), (0.115, 10.0, 201, 500)], "CO": [(0.0, 8.74, 0, 50), (8.75, 11.00, 51, 100), (11.01, 13.30, 101, 150), (13.31, 15.50, 151, 200), (15.51, 100.0, 201, 500)], "SO2": [(0.0, 0.008, 0, 50), (0.009, 0.110, 51, 100), (0.111, 0.165, 101, 150), (0.166, 0.220, 151, 200), (0.221, 10.0, 201, 500)], "NO2": [(0.0, 0.107, 0, 50), (0.108, 0.210, 51, 100), (0.211, 0.230, 101, 150), (0.231, 0.250, 151, 200), (0.251, 10.0, 201, 500)] } if gas not in breakpoints: return 0 for bp in breakpoints[gas]: bp_lo, bp_hi, i_lo, i_hi = bp if bp_lo <= cp <= bp_hi: # Fórmula de interpolación lineal oficial ica = ((i_hi - i_lo) / (bp_hi - bp_lo)) * (cp - bp_lo) + i_lo return round(ica) # Si sobrepasa el límite máximo establecido (Caso extremo) return 500 def obtener_banda_ica(ica): if ica <= 50: return "Buena" elif ica <= 100: return "Aceptable" elif ica <= 150: return "Mala" elif ica <= 200: return "Muy Mala" else: return "Extremadamente Mala" def procesador_ica_background(): """Hilo que corre en segundo plano calculando el ICA de los nuevos registros""" while True: try: conexion = sqlite3.connect(RUTA_BD) cursor = conexion.cursor() # Buscamos todas las filas que el programa C acaba de insertar (ica es NULL) cursor.execute("SELECT id, ozono, no2, co, so2 FROM mediciones WHERE ica IS NULL") filas_pendientes = cursor.fetchall() for fila in filas_pendientes: id_fila = fila[0] # Calculamos el ICA de cada contaminante por separado ica_o3 = calcular_ica_individual(fila[1], "O3") ica_no2 = calcular_ica_individual(fila[2], "NO2") ica_co = calcular_ica_individual(fila[3], "CO") ica_so2 = calcular_ica_individual(fila[4], "SO2") # La norma dicta que el ICA global es el valor máximo encontrado ica_global = max(ica_o3, ica_no2, ica_co, ica_so2) banda = obtener_banda_ica(ica_global) # Actualizamos la fila en la base de datos con el cálculo final cursor.execute("UPDATE mediciones SET ica = ?, banda_ica = ? WHERE id = ?", (ica_global, banda, id_fila)) conexion.commit() conexion.close() except Exception as e: print(f"Error procesando ICA: {e}") # Esperamos 30 segundos antes de volver a revisar la base de datos time.sleep(30) # Función 1: Obtener datos en tiempo real @app.route('/api/actual') def obtener_actual(): datos = {"estado": "ok"} # Validamos si el archivo existe if not os.path.exists(RUTA_ACTUAL): return jsonify({"estado": "error", "mensaje": "Archivo no encontrado"}) # Validamos si el programa en C se detuvo (Si el archivo tiene más de 20 segundos sin modificarse) tiempo_modificacion = os.path.getmtime(RUTA_ACTUAL) if (time.time() - tiempo_modificacion) > 20: return jsonify({"estado": "desconectado"}) # Esto activará el mensaje de "Desconectado" en la web # Leemos el archivo línea por línea with open(RUTA_ACTUAL, 'r') as f: for linea in f: if '=' in linea: clave, valor = linea.strip().split('=', 1) datos[clave] = valor.replace('"', '') # Limpiamos comillas de la fecha return jsonify(datos) # Función 2: Obtener historial para las gráficas @app.route('/api/historial') def obtener_historial(): rango = request.args.get('rango', '1_mes') # Por defecto lee el último mes conexion = sqlite3.connect(RUTA_BD) conexion.row_factory = sqlite3.Row # Permite acceder a las columnas por nombre cursor = conexion.cursor() # Si el usuario eligió fechas personalizadas if rango == 'personalizado': inicio = request.args.get('inicio') fin = request.args.get('fin') # Usamos ? para inyectar las variables de forma segura consulta = "SELECT fecha_hora, presion, temperatura, humedad, ozono, no2, co, so2 FROM mediciones WHERE fecha_hora BETWEEN ? AND ? ORDER BY fecha_hora ASC" # Agregamos las horas al inicio y al fin para abarcar los días completos cursor.execute(consulta, (f"{inicio} 00:00:00", f"{fin} 23:59:59")) # Si eligió los rangos predefinidos else: filtro_fecha = "" if rango == '1_mes': filtro_fecha = "date('now', '-1 month')" elif rango == '2_meses': filtro_fecha = "date('now', '-2 months')" elif rango == '1_anio': filtro_fecha = "date('now', '-1 year')" consulta = f"SELECT fecha_hora, presion, temperatura, humedad, ozono, no2, co, so2 FROM mediciones WHERE fecha_hora >= {filtro_fecha} ORDER BY fecha_hora ASC" cursor.execute(consulta) filas = cursor.fetchall() conexion.close() # Convertimos los resultados de SQLite a una lista de diccionarios (JSON) resultado = [dict(fila) for fila in filas] return jsonify(resultado) # Función 3: Terminal Remota (Ejecutar comandos) @app.route('/api/ejecutar', methods=['POST']) def ejecutar_comando(): datos = request.get_json() comando = datos.get('comando', '') if not comando: return jsonify({'salida': '', 'error': 'No se proporcionó ningún comando.'}) try: # Ejecutar el comando en el shell de la Raspberry Pi # Se captura la salida normal (stdout) y los errores (stderr) resultado = subprocess.run(comando, shell=True, capture_output=True, text=True, timeout=10) salida = resultado.stdout error = resultado.stderr # Combinar salida y error (Priorizar mostrar el error si existe) respuesta = error if error else salida return jsonify({'salida': respuesta, 'error': ''}) except subprocess.TimeoutExpired: return jsonify({'salida': '', 'error': 'El comando tardó demasiado en ejecutarse (Timeout).'}) except Exception as e: return jsonify({'salida': '', 'error': str(e)}) # --- ARRANQUE DEL SERVIDOR Y EL HILO --- if __name__ == '__main__': # Iniciamos el procesador de ICA en un hilo paralelo para que no bloquee el servidor web hilo_ica = threading.Thread(target=procesador_ica_background, daemon=True) hilo_ica.start() app.run(host='0.0.0.0', port=5000)