from pymongo import MongoClient from pprint import pprint from collections import defaultdict from pprint import pformat from datetime import timedelta from .comap_data import get_latest_comap_data from .generator_outputs import generator_is_on client = MongoClient("mongodb://localhost:27017") db = client["NANOOX_TS"] class SimpleLogger: def info(self, msg): print(f"[INFO] {msg}") # print(msg) def warning(self, msg): print(f"[WARNING] {msg}") #print(msg) def error(self, msg): print(f"[ERROR] {msg}") def debug(self, msg): # print(f"[DEBUG] {msg}") return logger = SimpleLogger() def get_tail_historial_control(db, center_id, n=30): """ Versión con eliminación de duplicados que respeta EXACTAMENTE el mismo orden que `original_get_tail_historial_control`: - Primero se ordenan todos los logs por _log_ts desc, _id desc (en Mongo). - Luego se recorre en Python y se van tomando solo los logs únicos según (ubicacion, fecha, accion), hasta tener n elementos. """ n = int(n) # Pipeline IDÉNTICO al original, PERO SIN el $limit pipeline = [ {"$match": {"centro_id": center_id}}, {"$unwind": "$logs"}, {"$set": { "_log_ts": { "$let": { "vars": { "ts": "$logs.timestamp", "ts_is_date": {"$eq": [{"$type": "$logs.timestamp"}, "date"]}, "ts_to_date": { # Intenta castear (si es string ISO/epoch). Si falla, queda null. "$convert": { "input": "$logs.timestamp", "to": "date", "onError": None, "onNull": None } }, "parsed_fecha": { "$dateFromString": { "dateString": { "$replaceAll": { "input": {"$trim": {"input": "$logs.fecha", "chars": "[]"}}, "find": " - ", "replacement": " " } }, "format": "%Y/%m/%d %H:%M:%S", "timezone": "America/Santiago", "onError": None, "onNull": None } } }, "in": { "$ifNull": [ {"$cond": ["$$ts_is_date", "$$ts", "$$ts_to_date"]}, "$$parsed_fecha" ] } } } }}, {"$sort": {"_log_ts": -1, "_id": -1}}, # OJO: aquí NO ponemos $limit {"$replaceRoot": {"newRoot": {"$mergeObjects": ["$logs", {"_log_ts": "$_log_ts"}]}}} ] # 1) Obtenemos TODOS los logs ya ordenados (igual que antes, pero sin limitar n) all_logs = list(db["historial_control"].aggregate(pipeline, allowDiskUse=True)) # 2) Deduplicamos manteniendo el orden tal cual vienen seen = set() result = [] # Usa al menos 3 campos para considerar un log duplicado fields_for_key = ("ubicacion", "fecha", "accion") for log in all_logs: key = tuple(log.get(f) for f in fields_for_key) if key in seen: # Duplicado → lo saltamos continue seen.add(key) result.append(log) if len(result) >= n: break return result def get_last_state_bulk(collection, id_field_name, id_list, centro_id=None): if not id_list: return {} # OPTIMIZACIÓN CRÍTICA: Límite de ventana de tiempo. # Solo buscamos en los últimos 7 días. Esto permite que MongoDB use # el índice de timestamp de forma mucho más agresiva y evita Full Scans. fecha_limite = datetime.now() - timedelta(days=7) match_stage = { id_field_name: {"$in": id_list}, # Asumiendo que 'timestamp' es objeto Date. Si es string, conviértelo. "timestamp": {"$gte": fecha_limite} } if centro_id: match_stage["centro_id"] = centro_id pipeline = [ {"$match": match_stage}, {"$sort": {"timestamp": -1}}, {"$group": { "_id": f"${id_field_name}", "latest_doc": {"$first": "$$ROOT"} }} ] results = {} # Agregamos allowDiskUse=True por seguridad si la memoria se llena, # aunque con el filtro de fecha no debería ser necesario. for doc in collection.aggregate(pipeline, allowDiskUse=True): data = doc["latest_doc"] if "_id" in data: del data["_id"] results[doc["_id"]] = data return results def get_latest_data_ts(db, centro_id=None): region, area, centro = centro_id.split("-") generators_comap = get_latest_comap_data(db, centro_id) historial_control = [] generators_latest_snap = [] def build_response(data_status, data_status_detail, **extra): response = { "modulos": [], "generadores": generators_latest_snap, "historial_control": historial_control, "generators_comap": generators_comap, "data_status": data_status, "data_status_detail": data_status_detail, } response.update(extra) return response try: # 1. Obtener Historial (se mantiene igual, asumiendo que es ligero) try: historial_control = get_tail_historial_control(db, centro_id) except Exception: historial_control = [] # 2. Referencias a Colecciones metadata_col = db[f"{centro}_metadata"] estructura_col = db[f"{centro}_estructura_logica"] sensores_ts_col = db[f"{centro}_sensores"] can_ts_col = db[f"{centro}_cans"] # 3. Obtener TODOS los metadatos de una vez (1 Query) # Filtramos por centro_id si es compartido, o traemos todo si la col es dedicada all_metadata = list(metadata_col.find({"centro_id": centro_id}) if centro_id else metadata_col.find()) # Separar IDs para hacer consultas masivas bomba_ids = [] can_ids = [] metadata_map = {} # Para acceder rápido a la info estática (nombre, alias) for doc in all_metadata: doc_id = doc.get("_id") metadata_map[doc_id] = doc if doc.get("tipo") == "bomba": bomba_ids.append(doc_id) elif doc.get("tipo") == "can": can_ids.append(doc_id) # 4. CONSULTAS MASIVAS (El arreglo del problema de rendimiento) # En lugar de 50 consultas, hacemos solo 2. # Diccionario: {bomba_id: {datos_timeseries...}} latest_bombas_ts = get_last_state_bulk(sensores_ts_col, "bomba_id", bomba_ids, centro_id) # Diccionario: {can_id: {datos_timeseries...}} latest_cans_ts = get_last_state_bulk(can_ts_col, "can_id", can_ids, centro_id) # 5. Obtener "Multisensor" (Si esto trae datos diferentes a sensores_ts, se mantiene) mediciones_por_bomba = {} if bomba_ids: mediciones_por_bomba = obtener_ultimas_medidas_multisensor( db, bomba_ids, sensores_coll=sensores_ts_col, N=360, step=1 ) # 6. Procesar Generadores (Optimizado) metadata_gen_col = db[f"{centro}_metadata_generadores"] ts_gen_col = db[f"{centro}_generador_timeseries"] generadores_meta = list(metadata_gen_col.find({"centro_id": centro_id})) gen_names = [g.get("generator_name") for g in generadores_meta] # Bulk query para generadores también latest_gen_ts = get_last_state_bulk(ts_gen_col, "generador_name", gen_names) # o "generador_id" según tu esquema generators_latest_snap = [] generators_by_quadrant = {} generators_state = {} for gen in generadores_meta: name = gen.get("generator_name") quadrants = gen.get("quadrants", []) ultimo_dato = latest_gen_ts.get(name) # Obtenido de memoria, no DB if ultimo_dato: generators_latest_snap.append(ultimo_dato) # Encendido/apagado. Con control por falla activo el generador # en marcha reporta patrones mixtos (110 linea principal, 011 # respaldo): all(c == "1") los daba por apagados, y ese valor # llega a cada jaula del cuadrante via can_final["generator"]. current_state = generator_is_on(ultimo_dato) if current_state is None: current_state = ultimo_dato.get("STATE") generators_state[name] = current_state for q in quadrants: generators_by_quadrant[q] = name # 7. Reconstrucción de la Estructura Lógica (En memoria) # Ya tenemos todos los datos en diccionarios (latest_bombas_ts, latest_cans_ts) estructura_cans_doc = estructura_col.find_one({"_id": "cans"}) if not estructura_cans_doc: # Fallback a buscar por "id" si "_id" falla, como tenías comentado estructura_cans_doc = estructura_col.find_one({"id": "cans"}) if not estructura_cans_doc: has_comap_data = bool(generators_comap.get("generators")) if has_comap_data: return build_response( "comap_only", "El centro reporta generadores COMAP, pero no tiene estructura lógica de módulos.", ) return build_response( "sin_estructura", "Estructura lógica no encontrada.", error="Estructura lógica no encontrada", ) cans_final_data = {} for can_node in estructura_cans_doc.get("cans", []): can_id = can_node.get("can_id") # Recuperamos datos del CAN del diccionario bulk can_ts_data = latest_cans_ts.get(can_id) if not can_ts_data: continue # Si no hay datos TS, saltamos (o puedes poner default) # Inyectamos metadata si falta en TS can_final = can_ts_data.copy() # Logica de generador asociada hf_name = can_final.get("can", "") quadrant_name = hf_name.replace("HF", "quadrant_") gen_name = generators_by_quadrant.get(quadrant_name) can_final["generator"] = generators_state.get(gen_name) # Procesar bombas hijas can_final["bombas"] = [] for bomba_id in can_node.get("bombas", []): # Recuperar datos Bomba del diccionario bulk bomba_ts = latest_bombas_ts.get(bomba_id) if bomba_ts: bomba_final = bomba_ts.copy() # Inyectar mediciones multisensor si existen bomba_final["latest_measurements"] = mediciones_por_bomba.get(bomba_id, {}) can_final["bombas"].append(bomba_final) cans_final_data[can_id] = can_final # 8. Agrupar por Módulo (Output final) modulos_map = defaultdict(lambda: {"cans": []}) for can_id, can_data in cans_final_data.items(): modulo_id = can_data.get("modulo") if modulo_id is None: try: modulo_id = int(can_data.get("modulo")) except: continue modulos_map[modulo_id]["id"] = modulo_id modulos_map[modulo_id]["cans"].append(can_data) modulos = list(modulos_map.values()) if not modulos: return build_response( "sin_datos_recientes", "La estructura lógica existe, pero no hay datos recientes de módulos, CAN o jaulas.", ) return build_response( "ok", "Datos del centro disponibles.", modulos=modulos, ) except Exception as e: print(f"Error en get_latest_data_ts linea {e.__traceback__.tb_lineno}: {e}") # Retornar un dict seguro para que no explote el endpoint. return build_response( "error", "Ocurrió un error al obtener los datos del centro.", error="Error al obtener los datos del centro", ) from datetime import datetime def serialize_timestamp(timestamp): """ @brief Verifica si un timestamp es un objeto datetime y lo convierte a ISO 8601. @param timestamp Puede ser un objeto datetime o una cadena. @return string con el timestamp en formato ISO 8601 si es datetime. Si ya es string, se devuelve sin cambios. @throws ValueError si el timestamp no es un datetime o string válido. """ if isinstance(timestamp, datetime): # Si es un objeto datetime (ISODate en MongoDB) return timestamp.isoformat() if isinstance(timestamp, str): # Si ya es un string, lo retorna directamente return timestamp raise ValueError(f"Formato de timestamp inválido: {timestamp} ({type(timestamp)})") def obtener_ultimas_medidas_sensor(sensores_ts, bomba_id: int, N: int = 100, step: int = 10, centro_id = None): # print("en obtener ultimas mediciones") # Consulta de los últimos N documentos query = { "bomba_id": bomba_id} if centro_id: query.update({"centro_id": centro_id}) cursor = sensores_ts.find( query, {"_id": 0, "O2": 1, "SO2": 1, "timestamp": 1,} ).sort("timestamp", -1).limit(N) docs = list(cursor) # Invertir para ordenar por tiempo ascendente docs.reverse() # Extraer valores O2_values = [doc.get("O2", None) for doc in docs] SO2_values = [doc.get("SO2", None) for doc in docs] TS = [serialize_timestamp(doc.get("timestamp")) for doc in docs] # Resultado con muestreo opcional return { "O2": O2_values[-N:][::step], "SO2": SO2_values[-N:][::step], "TS": TS[-N:][::step] } def obtener_ultimas_medidas_multisensor( db, bomba_ids, sensores_coll, # o2_data, # {id: O2} N: int = 100, step: int = 10, centro_id = None ): """ Devuelve un dict con key=bomba_id y value={ "O2": [...], "SO2": [...], "TS": [...] }, todas las arrays ya ordenadas de más antiguo a más nuevo y muestreadas con `step`. """ try: logger.debug("en obtener_ultimas_medidas_multisensor") sensores_ts = sensores_coll#db[sensores_coll] # 1) Límites del día de hoy ahora = datetime.now() inicio_hoy = ahora.replace(hour=0, minute=0, second=0, microsecond=0) inicio_manana = inicio_hoy + timedelta(days=1) # 2) Match global por todos los sensores en una sola consulta match = { "bomba_id": {"$in": list(bomba_ids)}, #"O2": {"$gt": 0}, #"SO2": {"$gt": 0}, "timestamp": {"$gte": inicio_hoy, "$lt": inicio_manana}, **({"centro_id": centro_id} if centro_id is not None else {}) } pipeline = [ {"$match": match}, {"$sort": {"timestamp": -1}}, # 1) Pillar los N más recientes *globalmente* {"$group": { "_id": "$bomba_id", "docs": {"$push": { "O2": "$O2", "SO2": "$SO2", "O2_backup": {"$ifNull": ["$O2_bckp", None]}, "SO2_backup": {"$ifNull": ["$SO2_bckp", None]}, "Inyecting_O2": {"$ifNull": ["$Inyecting_O2", None]}, # inyecting puede faltar, así que le aplicamos $ifNull "Temp": {"$ifNull": ["$TEMP", None]}, # Puede no existir (?) - Ocurrió # "$TEMP", "Temp_backup": {"$ifNull": ["$TEMP_bckp", None]}, # Puede no existir (?) - Ocurrió # "$TEMP", "SetPoint": {"$ifNull": ["$SetPoint", None]}, "SetPointOFF": {"$ifNull": ["$SetPointOFF", None]}, "alarms": {"$ifNull": ["$alarms", None]}, "timestamp": "$timestamp" }} }}, {"$project": { # 2) Por cada grupo, slice de tamaño N "docs": {"$slice": ["$docs", N]} }} ] raw = list(sensores_ts.aggregate(pipeline)) print("RAW") # pprint(raw) # Detectar cambios en SetPoint y SetPointOFF resultado = {} for grupo in raw: bomba_id = grupo["_id"] docs = grupo["docs"] setpoint_values = [] setpointoff_values = [] last_sp_on = None last_sp_off = None # 3) Invertir para que queden asc y aplicar step/serialización docs_asc = list(reversed(docs)) docs_asc_step = docs_asc[::step] # aplicar step solo una vez use_backup = False#not bool( o2_data.get(bomba_id) ) # is None for doc in docs_asc: ts = serialize_timestamp(doc["timestamp"]) sp_on = doc.get("SetPoint") sp_off = doc.get("SetPointOFF") # Agregado de SetPoint if sp_on is not None and sp_on != last_sp_on: setpoint_values.append({ "value": sp_on, "time": ts }) last_sp_on = sp_on # Agregado de SetPointOFF if sp_off is not None and sp_off != last_sp_off: setpointoff_values.append({ "value": sp_off, "time": ts }) last_sp_off = sp_off #O2_vals = [d["O2"] for d in docs_asc_step][::step] #SO2_vals = [d["SO2"] for d in docs_asc_step][::step] # Decidir si obtener desde backup o de dato real O2_vals = [(d.get("O2_backup") if use_backup else d.get("O2")) for d in docs_asc_step if d] SO2_vals = [(d.get("SO2_backup") if use_backup else d.get("SO2")) for d in docs_asc_step if d] TS_vals = [serialize_timestamp(d.get("timestamp")) for d in docs_asc_step if d] temp_vals = [(d.get("Temp_backup") if use_backup else d.get("Temp")) for d in docs_asc_step if d] inyect_vals = [d.get("Inyecting_O2") for d in docs_asc_step if d] alarms = [d.get("alarms") for d in docs_asc_step if d] inyect_vals = [ d["Inyecting_O2"] for d in docs_asc_step ] resultado[bomba_id] = { "O2": O2_vals, "SO2": SO2_vals, "Temp": temp_vals, "TS": TS_vals, "Inyecting_O2": inyect_vals, "DataSetPoint": setpoint_values, "DataSetPointOFF": setpointoff_values, "alarms" : alarms } # pprint(resultado) return resultado except Exception as e: logger.error(f'Error {e} en {e.__traceback__.tb_lineno}') return {} #--Legacy Code----- def legacy_get_latest_data_ts(db, centro_id = None): region,area,centro = centro_id.split("-") try: try: historial_control = get_tail_historial_control(db, centro_id) print("Historial de control obtenido") # pprint(historial_control) except Exception as e: print(f'Error {e} obteniendo histiorial de control en {e.__traceback__.tb_lineno}') historial_control = [] metadata_col = f"{centro}_metadata" estructura_col = f"{centro}_estructura_logica" sensores_ts = f"{centro}_sensores" can_ts = f"{centro}_cans" metadata_col = db[metadata_col] estructura_col = db[estructura_col] sensores_ts = db[sensores_ts] can_ts = db[can_ts] # Indexar metadata bombas_data = {} cans_data = {} # Obtener todos los _id de documentos tipo "bomba" bombas_ids = [doc["_id"] for doc in metadata_col.find({"tipo": "bomba"}, {"_id": 1})] print(f"{bombas_ids=}") mediciones_por_bomba = obtener_ultimas_medidas_multisensor( db, bombas_ids, # o2_data, sensores_coll = sensores_ts, N=360, step=1, # centro_id=centro_id ) for doc in metadata_col.find(): if "tipo" not in doc: continue tipo = doc.get("tipo") _id = doc.get("_id") # _id = doc.get("id") if tipo == "bomba": # latest_measurement = obtener_ultimas_medidas_sensor(sensores_ts, bomba_id= _id, N=360, step=1, centro_id = centro_id) # pprint(latest_measurement) latest_measurement = mediciones_por_bomba.get(_id, {}) query = {"bomba_id": _id} if centro_id:query.update({ "centro_id": centro_id}) last_bomb_data = sensores_ts.find_one(query, sort=[("timestamp", -1)]) last_bomb_data.update({"latest_measurements": latest_measurement}) last_bomb_data.pop("_id") if ts: bombas_data[_id] = last_bomb_data elif tipo == "can": query = {"can_id": _id} if centro_id:query.update({ "centro_id": centro_id}) ts = can_ts.find_one(query, sort=[("timestamp", -1)]) if ts: cans_data[_id] = ts # pprint(bombas_data) #TODO: agregar un if para saber si agregar o no #obtencion de informacion de generadores en formato nuevo try: # Colecciones involucradas metadata_col_name = f"{centro}_metadata_generadores" ts_col_name = f"{centro}_generador_timeseries" metadata_col = db[metadata_col_name] ts_col = db[ts_col_name] resultado = [] generators_by_quadrant = {} generators_state = {} # Obtener metadata generadores = list(metadata_col.find({"centro_id": centro_id})) generators_latest_snap = [] for gen in generadores: generator_name = gen.get("generator_name") quadrants = gen.get("quadrants", []) # Buscar último dato en timeseries ultimo_dato = ts_col.find_one( {"generador_name": generator_name}, sort=[("timestamp", -1)] ) # print(ultimo_dato) del ultimo_dato["_id"] generators_latest_snap.append(ultimo_dato) resultado.append({ "centro_id": centro, "generator_name": generator_name, "quadrants": quadrants, "ultimo_estado": ultimo_dato }) # Construir mapping quadrant → generator for quadrant in quadrants: generators_by_quadrant[quadrant] = generator_name # Guardar estado ON/OFF del generador if ultimo_dato: # if centro.lower() == "puntacola": # pprint(ultimo_dato) # Aquí asumo que el estado se calcula a partir de STATE o digital_outputs_state state_raw = ultimo_dato.get("STATE") # Ejemplo: si tienes digital_outputs_state como string con "0"/"1" if "digital_outputs_state" in ultimo_dato: digital_outputs_state = ultimo_dato.get("digital_outputs_state","") print(f'{digital_outputs_state=}') if digital_outputs_state: current_state = all(c == "1" for c in digital_outputs_state) else: current_state = None print(f'{current_state=}') else: current_state = state_raw # usa tal cual si solo existe STATE generators_state[generator_name] = current_state # pprint(generators_by_quadrant) # print() # # Construir mapping final quadrant → estado actual # quadrants_state = {} # for quadrant, generator_name in generators_by_quadrant.items(): # quadrants_state[quadrant] = generators_state.get(generator_name) # pprint( { # "generadores": resultado, # "generators_by_quadrant": generators_by_quadrant, # "generators_state": generators_state, # "quadrants_state": quadrants_state # }) except Exception as e: print(f'Error {e} en {e.__traceback__.tb_lineno}') # # Reconstruir estructura lógica estructura_cans = estructura_col.find_one({"_id": "cans"}) # estructura_cans = estructura_col.find_one({"id": "cans"}) if not estructura_cans: # raise ValueError("❌ No se encontró la estructura lógica con _id='cans'") return "No se encontró la estructura lógica con _id='cans'" for can in estructura_cans.get("cans", []): can_id = can.get("can_id") if can_id not in cans_data: continue if centro.lower() == "puntacola": print(f"\n{' generators_state ':*^60}\n") pprint(generators_state) corresponding_can_data = cans_data[can_id] hf_name = corresponding_can_data.get("can") # print(f'{hf_name=}') # Convertir CAN -> quadrant_x quadrant_name = hf_name.replace("HF", "quadrant_") corresponding_generator = generators_by_quadrant.get(quadrant_name) corresponding_generator_state = generators_state.get(corresponding_generator) cans_data[can_id]["generator"] = corresponding_generator_state cans_data[can_id]["bombas"] = [] for bomba_id in can.get("bombas", []): bomba = bombas_data.get(bomba_id) if bomba: cans_data[can_id]["bombas"].append(bomba) # pprint(cans_data[can_id]) # pprint(estructura_cans) # return estructura_cans # Agrupar por módulo modulos_map = defaultdict(lambda: {"cans": []}) for can_id, can in cans_data.items(): modulo_id = int(can.get("modulo")) if modulo_id is None: continue can_obj = can.copy() # print(f'{can_obj=}') can_obj.pop("_id", None) modulos_map[modulo_id]["id"] = modulo_id modulos_map[modulo_id]["cans"].append(can_obj) latest_data_dict = { "modulos": list(modulos_map.values()), 'generadores': generators_latest_snap, "historial_control": historial_control } # pprint(historial_control) # {"modulos": modulos , "historial_control": historial_control, "hora": hora, 'generadores': generator, 'umbrales': umbrales} # return(list(modulos_map.values())) return latest_data_dict except Exception as e: print(f"Error en linea {e.__traceback__.tb_lineno} {e}") return [] def legacy__get_tail_historial_control(db, center_id, n=30): n = int(n) pipeline = [ {"$match": {"centro_id": center_id}}, {"$unwind": "$logs"}, {"$set": { "_log_ts": { "$let": { "vars": { "ts": "$logs.timestamp", "ts_is_date": {"$eq": [{"$type": "$logs.timestamp"}, "date"]}, "ts_to_date": { # Intenta castear (si es string ISO/epoch). Si falla, queda null. "$convert": { "input": "$logs.timestamp", "to": "date", "onError": None, "onNull": None } }, "parsed_fecha": { "$dateFromString": { "dateString": { "$replaceAll": { "input": {"$trim": {"input": "$logs.fecha", "chars": "[]"}}, "find": " - ", "replacement": " " } }, "format": "%Y/%m/%d %H:%M:%S", "timezone": "America/Santiago", "onError": None, "onNull": None } } }, "in": { "$ifNull": [ {"$cond": ["$$ts_is_date", "$$ts", "$$ts_to_date"]}, "$$parsed_fecha" ] } } } }}, {"$sort": {"_log_ts": -1, "_id": -1}}, {"$limit": n}, {"$replaceRoot": {"newRoot": {"$mergeObjects": ["$logs", {"_log_ts": "$_log_ts"}]}}} ] return list(db["historial_control"].aggregate(pipeline, allowDiskUse=True)) # # print(get_data(db, "centro")) # pprint(get_data(db, "Loslagos-Puertomontt-ITG"))