from pymongo import MongoClient from datetime import datetime from pprint import pprint # from zoneinfo import ZoneInfo try: from zoneinfo import ZoneInfo # Python 3.9+ except ImportError: from backports.zoneinfo import ZoneInfo # Python 3.8 y menores from datetime import datetime dt = datetime.now(ZoneInfo("America/Santiago")) print(dt) # def get_db(db_name = "NANOOX_TS"): # client = MongoClient("mongodb://localhost:27017") # db = client[db_name] # return db # # # REEMPLAZAR con la estructura real si no se lee desde archivo # data = [{'id': 1, 'cans': [{'bombas': [{'aviot_status': True, 'alias': '105', 'nombre': 'VA1', 'ip_aviot': '10.0.0.44', 'Inputs': '101', 'InSel': 1, 'InStop': 0, 'InStart': 1, 'Control': 1, 'Start': '0000', 'Device_Modbus_Status': False, 'Vo': 0.0, 'Io': 0.0, 'Fault': None, 'Status': 13, 'last_measure_sensor': '2025-03-31 16:34:53', 'O2': 4.33, 'SO2': 47.77, 'TEMP': 12.11, 'O2_bckp': 5.64, 'O2_Modbus_Status': True, 'RLY': False, 'Start_Valv': '0000', 'SetPoint': 5.0, 'SetPointOFF': 6.0}, {'aviot_status': True, 'alias': '106', 'nombre': 'VA2', 'ip_aviot': '10.0.0.44', 'Inputs': '101', 'InSel': 1, 'InStop': 0, 'InStart': 1, 'Control': 1, 'Start': '0000', 'Device_Modbus_Status': False, 'Vo': 0.0, 'Io': 0.0, 'Fault': None, 'Status': 0, 'last_measure_sensor': '2025-03-31 16:34:53', 'O2': 5.37, 'SO2': 59.27, 'TEMP': 12.21, 'O2_bckp': 5.64, 'O2_Modbus_Status': True, 'RLY': False, 'Start_Valv': '0000', 'SetPoint': 5.0, 'SetPointOFF': 6.0}, {'aviot_status': True, 'alias': '107', 'nombre': 'VA3', 'ip_aviot': '10.0.0.44', 'Inputs': '101', 'InSel': 1, 'InStop': 0, 'InStart': 1, 'Control': 1, 'Start': '0000', 'Device_Modbus_Status': False, 'Vo': 397.58, 'Io': 17.4825, 'Fault': None, 'Status': 9, 'last_measure_sensor': '2025-03-31 14:11:38', 'O2': None, 'SO2': None, 'TEMP': None, 'O2_bckp': 5.64, 'O2_Modbus_Status': False, 'RLY': False, 'Start_Valv': '0000', 'SetPoint': 5.0, 'SetPointOFF': 6.0}], 'ipAddress': '192.168.10.62', 'can': 'HF1', 'modulo': 1, 'estado': True, 'source': False, 'presion': -2.5, 'sal': 0.0, 'voltaje': 397.58, 'corriente': 17.4825, 'aviot': '10.0.0.44'}, {'bombas': [{'aviot_status': True, 'alias': '108', 'nombre': 'VA5', 'ip_aviot': '10.0.0.44', 'Inputs': '101', 'InSel': 1, 'InStop': 0, 'InStart': 1, 'Control': 1, 'Start': '0000', 'Device_Modbus_Status': False, 'Vo': 0.0, 'Io': 0.0, 'Fault': None, 'Status': 0, 'last_measure_sensor': '2025-03-31 16:34:53', 'O2': 5.43, 'SO2': 60.07, 'TEMP': 12.31, 'O2_bckp': 5.64, 'O2_Modbus_Status': True, 'RLY': False, 'Start_Valv': '0000', 'SetPoint': 5.0, 'SetPointOFF': 6.0}, {'aviot_status': True, 'alias': '109', 'nombre': 'VA6', 'ip_aviot': '10.0.0.44', 'Inputs': '100', 'InSel': 1, 'InStop': 0, 'InStart': 0, 'Control': 1, 'Start': '0000', 'Device_Modbus_Status': False, 'Vo': 0.0, 'Io': 0.0, 'Fault': None, 'Status': 0, 'last_measure_sensor': '2025-03-31 16:34:54', 'O2': 6.26, 'SO2': 69.21, 'TEMP': 12.26, 'O2_bckp': 5.64, 'O2_Modbus_Status': True, 'RLY': False, 'Start_Valv': '0000', 'SetPoint': 5.0, 'SetPointOFF': 6.0}, {'aviot_status': True, 'alias': '110', 'nombre': 'VA7', 'ip_aviot': '10.0.0.44', 'Inputs': '100', 'InSel': 1, 'InStop': 0, 'InStart': 0, 'Control': 1, 'Start': '0000', 'Device_Modbus_Status': False, 'Vo': 0.0, 'Io': 0.0, 'Fault': None, 'Status': 0, 'last_measure_sensor': '2025-03-31 16:34:53', 'O2': 6.81, 'SO2': 75.13, 'TEMP': 12.11, 'O2_bckp': 5.64, 'O2_Modbus_Status': True, 'RLY': False, 'Start_Valv': '0000', 'SetPoint': 5.0, 'SetPointOFF': 6.0}], 'ipAddress': '192.168.10.68', 'can': 'HF2', 'modulo': 1, 'estado': True, 'source': False, 'presion': -2.5, 'sal': 0.0, 'voltaje': 0, 'corriente': 0.0, 'aviot': '10.0.0.44'}]}] def generate_centro_id(region_id, area_id, centro_id): return int(f"{region_id:02}{area_id:02}{centro_id:02}") def get_or_create_id(nombre, tipo, db): """ nombre: string (ej: 'LosLagos') tipo: uno de 'clientes', 'regiones', 'areas', 'centros' db: MongoClient().your_db """ try: col = db["catalogos"] doc = col.find_one({"_id": "mapas_id"}) # Si no existe el documento base, inicializarlo if not doc: doc = { "_id": "mapas_id", "clientes": {}, "regiones": {}, "areas": {}, "centros": {}, "contador": { "clientes": 0, "regiones": 0, "areas": 0, "centros": 0 } } col.insert_one(doc) nombre = nombre.strip().lower() if nombre in doc[tipo]: return doc[tipo][nombre] nuevo_id = doc["contador"][tipo] + 1 doc[tipo][nombre] = nuevo_id doc["contador"][tipo] = nuevo_id col.replace_one({"_id": "mapas_id"}, doc) return nuevo_id except Exception as e: print(f"Error {e} (line {e.__traceback__.tb_lineno})") def generate_numeric_id(modulo, aviot_idx, can_idx, bomba_pos): m = int(modulo) a = int(aviot_idx) c = int(can_idx) b = int(bomba_pos) # Ahora todos son ints, el :02 funciona sin problemas return int(f"{m:02}{a:02}{c:02}{b:02}") # def insert_metadata_si_no_existe(collection, doc): # try: # filtro = {"_id": doc["_id"], "centro_id": doc.get("centro_id")} # # filtro = {"id": doc["id"], "centro_id": doc.get("centro_id")} # #print(f"{filtro=}") # #print(doc) # #print(collection.count_documents(filtro)) # if collection.count_documents(filtro) == 0: # collection.insert_one(doc) # print(f"Insertando {doc['tipo'].upper()}") # # print(f"🟢 Insertado {doc['tipo'].upper()} {_id_centro_string(doc)}") # else: # print(f"🟡 {doc['tipo'].upper()} {_id_centro_string(doc)} ya existe, no se insertó") # except Exception as e: # print(f"Error en insert_metadata_si_no_existe: linea {e.__traceback__.tb_lineno} {e}") def insert_metadata_si_no_existe(collection, doc): # print("insert_metadata_si_no_existe".upper()) try: filtro = {"_id": doc["_id"], "centro_id": doc.get("centro_id")} existing_doc = collection.find_one(filtro) if existing_doc is None: # Si no existe → insertar collection.insert_one(doc) print(f"🟢 Insertado {doc.get('tipo', '').upper()} {_id_centro_string(doc)}") else: # Si ya existe → verificar si tiene el campo "control_version" if "control_version" not in existing_doc: # Si no tiene, actualizar documento agregando el campo desde el doc nuevo (si viene) control_version_value = doc.get("control_version", 1) result = collection.update_one( filtro, {"$set": {"control_version": control_version_value}} ) if result.modified_count: print(f"🟣 Campo 'control_version' agregado a {doc.get('tipo', '').upper()} {_id_centro_string(doc)}") else: print(f"🟠 No se modificó {doc.get('tipo', '').upper()}, ya tenía el campo o coincidía el valor.") else: print(f"🟡 {doc.get('tipo', '').upper()} {_id_centro_string(doc)} ya existe y tiene 'control_version'.") except Exception as e: print(f"❌ Error en insert_metadata_si_no_existe: línea {e.__traceback__.tb_lineno} — {e}") def _id_centro_string(doc): return f"{doc['_id']} ({doc.get('centro_id', 'sin centro_id')})" from datetime import timedelta, timezone from .generator_outputs import as_bool def insertar_dato_timeseries(collection, documento): res = collection.insert_one(documento) #print(f"➕ Insertado timeseries: {res.inserted_id}") #pprint(documento) def actualizar_estructura_logica(db, modulo_map, aviot_map, can_map, bombas_doc, centro_id=None): #print(f"{centro_id=}") region,area,centro = centro_id.split("-") estructura_col_name = f"{centro}_estructura_logica" estructura_col = db[estructura_col_name] def agregar_centro_id(lista_docs): for doc in lista_docs: doc["centro_id"] = centro_id return lista_docs modulos = list(modulo_map.values()) aviots = list(aviot_map.values()) cans = list(can_map.values()) bombas = bombas_doc["bombas"] if centro_id is not None: modulos = agregar_centro_id(modulos) aviots = agregar_centro_id(aviots) cans = agregar_centro_id(cans) bombas = agregar_centro_id(bombas) estructura_col.replace_one( {"_id": "modulos", "centro_id": centro_id} if centro_id else {"_id": "modulos"}, # {"id": "modulos", "centro_id": centro_id} if centro_id else {"id": "modulos"}, {"_id": "modulos", "modulos": modulos, "centro_id": centro_id} if centro_id else {"_id": "modulos", "modulos": modulos}, # {"id": "modulos", "modulos": modulos, "centro_id": centro_id} if centro_id else {"id": "modulos", "modulos": modulos}, upsert=True ) estructura_col.replace_one( {"_id": "aviots", "centro_id": centro_id} if centro_id else {"_id": "aviots"}, # {"id": "aviots", "centro_id": centro_id} if centro_id else {"id": "aviots"}, {"_id": "aviots", "aviots": aviots, "centro_id": centro_id} if centro_id else {"_id": "aviots", "aviots": aviots}, # {"id": "aviots", "aviots": aviots, "centro_id": centro_id} if centro_id else {"id": "aviots", "aviots": aviots}, upsert=True ) estructura_col.replace_one( {"_id": "cans", "centro_id": centro_id} if centro_id else {"_id": "cans"}, # {"id": "cans", "centro_id": centro_id} if centro_id else {"id": "cans"}, {"_id": "cans", "cans": cans, "centro_id": centro_id} if centro_id else {"_id": "cans", "cans": cans}, # {"id": "cans", "cans": cans, "centro_id": centro_id} if centro_id else {"id": "cans", "cans": cans}, upsert=True ) estructura_col.replace_one( {"_id": "bombas", "centro_id": centro_id} if centro_id else {"_id": "bombas"}, # {"id": "bombas", "centro_id": centro_id} if centro_id else {"id": "bombas"}, {"_id": "bombas", "bombas": bombas, "centro_id": centro_id} if centro_id else {"_id": "bombas", "bombas": bombas}, # {"id": "bombas", "bombas": bombas, "centro_id": centro_id} if centro_id else {"id": "bombas", "bombas": bombas}, upsert=True ) print(f"Estructura logica actualizada con {centro_id}." if centro_id else "Estructura logica actualizada.") def guardar_estructura_en_mongodb(db,data, generators_data = [] , centro_id=None, control = "PS", new_version = False, timestamp = None): print("en guardar_estructura_en_mongodb") try: region,area,centro = centro_id.split("-") # cliente_id = get_or_create_id("itg", "clientes", db) # region_id = get_or_create_id(region, "regiones", db) # area_id = get_or_create_id(area, "areas", db) # centro_id = get_or_create_id(centro, "centros", db) # centro_numeric_id = generate_centro_id(region_id= region_id , area_id= area_id, centro_id= centro_id) # print(f"{centro_id=} {centro_numeric_id=}") except Exception as e: print(f"Error {e} {e.__traceback__.tb_lineno}") # Parsear la hora original dt = datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S") # Convertir a formato ISO 8601 con zona horaria UTC # formatted_timestamp = dt.replace(tzinfo=timezone.utc).isoformat() formatted_timestamp = datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc) try: metadata_col_name = f"{centro}_metadata" sensores_col_name = f"{centro}_sensores" cans_col_name = f"{centro}_cans" generador_ts_col_name = f"{centro}_generador_timeseries" metadata_col = db[metadata_col_name] sensores_ts_col = db[sensores_col_name] can_ts_col = db[cans_col_name] generador_ts_col = db[generador_ts_col_name] # Crear colecciones si no existen if sensores_col_name not in db.list_collection_names(): db.create_collection(sensores_col_name, timeseries={"timeField": "timestamp", "metaField": "bomba_id", "granularity": "seconds"}) if cans_col_name not in db.list_collection_names(): db.create_collection(cans_col_name, timeseries={"timeField": "timestamp", "metaField": "can_id", "granularity": "seconds"}) if generador_ts_col_name not in db.list_collection_names(): db.create_collection( generador_ts_col_name, timeseries={"timeField": "timestamp", "metaField": "generador_id", "granularity": "seconds"}) modulo_map, aviot_map, can_map = {}, {}, {} bombas_doc = {"_id": "bombas", "bombas": []} for centro in data: for can_idx, can in enumerate(centro["cans"], start=1): try: # print("debuggeandp can") # print(can) ip_can = can["ipAddress"] ip_aviot = can["aviot"] modulo = can["modulo"] aviot_index = ip_aviot.split(".")[-1] can_id = generate_numeric_id(modulo, aviot_index, can_idx, 0) except Exception as e: print(f'Error {e} en {e.__traceback__.tb_lineno}') continue # Metadata CAN can_metadata = { "_id": can_id, # "id": can_id, "tipo": "can", "modulo": modulo, "ip_aviot": ip_aviot, "can": can["can"], "ip_can": ip_can, "control": control, "control_version" : new_version } if centro_id: can_metadata["centro_id"] = centro_id insert_metadata_si_no_existe(metadata_col, can_metadata) can_pressure = can.get("presion") # Timeseries CAN can_ts_doc = { "timestamp": formatted_timestamp, #datetime.now(), "can_id": can_id, "voltaje": can.get("voltaje"), "corriente": can.get("corriente"), "presion": can_pressure, "sal": can.get("sal"), "estado": can.get("estado"), "source": can.get("source"), "modulo": modulo, "aviot": ip_aviot, "can": can["can"] } if centro_id: can_ts_doc["centro_id"] = centro_id insertar_dato_timeseries(can_ts_col, can_ts_doc) # Estructura lógica modulo_map.setdefault(modulo, {"modulo_id": modulo, "nombre": f"Módulo {modulo}", "aviots": []}) if aviot_index not in modulo_map[modulo]["aviots"]: modulo_map[modulo]["aviots"].append(aviot_index) aviot_map.setdefault(aviot_index, { "aviot_id": aviot_index, "ip": ip_aviot, "modulo_id": modulo, "cans": [] }) if can_id not in aviot_map[aviot_index]["cans"]: aviot_map[aviot_index]["cans"].append(can_id) can_map.setdefault(can_id, {"can_id": can_id, "ip": ip_can, "aviot_id": aviot_index, "bombas": []}) for bomba_pos, bomba in enumerate(can["bombas"], start=1): bomba_id = generate_numeric_id(modulo, aviot_index, can_idx, bomba_pos) bomba_metadata = { "_id": bomba_id, # "id": bomba_id, "tipo": "bomba", "modulo": modulo, "ip_aviot": ip_aviot, "ip_can": ip_can, "posicion": bomba_pos, "nombre": bomba.get("nombre"), "alias": bomba.get("alias"), "control": control, "control_version" : new_version } if centro_id: bomba_metadata["centro_id"] = centro_id insert_metadata_si_no_existe(metadata_col, bomba_metadata) bomba_ts_doc = bomba.copy() bomba_ts_doc.update({ "timestamp": formatted_timestamp, #datetime.now(), "bomba_id": bomba_id, "Presion": can_pressure }) if centro_id: bomba_ts_doc["centro_id"] = centro_id insertar_dato_timeseries(sensores_ts_col, bomba_ts_doc) can_map[can_id]["bombas"].append(bomba_id) bombas_doc["bombas"].append({ "bomba_id": bomba_id, "nombre": bomba.get("nombre"), "alias": bomba.get("alias"), "can_id": can_id }) try: actualizar_estructura_logica(db, modulo_map, aviot_map, can_map, bombas_doc, centro_id) except Exception as er: print(f'Error {er} en {er.__traceback__.tb_lineno}') for generator_name, generator_data in generators_data: try: _save_one_generator_data( db, generador_ts_col, generator_name , generator_data) except Exception as e: print(f'Error {e} en {e.__traceback__.tb_lineno}') # if generador: # save_one_generator_data(db, centro_id, generador) except Exception as e: print(f"Error en guardar_estructura_en_mongodb: linea {e.__traceback__.tb_lineno} {e}") # print("fin funcion guardar_estructura_en_mongodb") # for i in range(10): # # # Llama a la función con tu estructura # guardar_estructura_en_mongodb(data, centro_id= "Loslagos-Puertomontt-ITG") def _comparison_key(snapshot): """ Clave que decide si un snapshot aporta informacion nueva. Comparar solo STATE descartaba la transicion entre linea principal ("110") y respaldo ("011"): con control por falla activo ocurre sin que STATE cambie, y es justo el evento que este contrato viene a comunicar. control_generator_back entra en la clave para que un cambio de modo quede registrado. Ausente y False se normalizan al mismo valor, asi que pasar de "no reportado" a False no genera fila. """ return ( snapshot.get("STATE"), snapshot.get("digital_outputs_state"), as_bool(snapshot.get("control_generator_back")) is True, ) def _save_one_generator_data(db, generators_col, generator_name, generator_data, centro_id=None): print("en _save_one_generator_data") try: if not generator_data or not isinstance(generator_data, dict): print(f"La información sobre generador no es válida: {generator_data}") return if all(value is None for value in generator_data.values()): print(f"Todos los datos sobre generador son inválidos: {generator_data}") return # Parsear timestamp a UTC datetime received_timestamp = generator_data.get("timestamp") parsed_to_isodate_timestamp = _to_utc_datetime(received_timestamp) generator_data["timestamp"] = parsed_to_isodate_timestamp # Construir documento con info adicional generator_doc = dict(generator_data) generator_doc.update({ "generador_name": generator_name, "generador_id": centro_id }) # Buscar último registro de ese generador existing_doc = generators_col.find_one( {"generador_name": generator_name}, {"STATE": 1, "digital_outputs_state": 1, "control_generator_back": 1, "_id": 0}, sort=[("timestamp", -1)] ) if existing_doc and _comparison_key(existing_doc) == _comparison_key(generator_data): print("Snapshot sin cambios (STATE/salidas/control por falla) -> no se guarda.") return # Insertar si no hay doc previo o si cambió algo de la clave de comparación. # Un reenvío desde 'pendientes' con el mismo snapshot sigue coincidiendo en # los tres campos, así que la idempotencia se mantiene. generators_col.insert_one(generator_doc) print("Nuevo documento insertado (cambió la clave de comparación).") except Exception as e: print(f"Error {e} en línea {e.__traceback__.tb_lineno}") # from datetime import datetime # from backports.zoneinfo import ZoneInfo # en tu Python 3.8 def _to_utc_datetime(ts, local_tz: str = "America/Santiago") -> datetime: """ Convierte un timestamp (str o datetime) a datetime UTC timezone-aware. Acepta: - datetime con o sin tz - string "%Y-%m-%d %H:%M:%S" - string ISO8601 con tz: "2025-10-02T18:57:11+00:00" """ if isinstance(ts, datetime): # Si ya es datetime dt_naive = ts if ts.tzinfo is None else ts.astimezone(ZoneInfo(local_tz)) elif isinstance(ts, str): try: dt_naive = datetime.fromisoformat(ts) except ValueError: # Fallback al formato clásico dt_naive = datetime.strptime(ts, "%Y-%m-%d %H:%M:%S") else: raise TypeError(f"Tipo no soportado para timestamp: {type(ts)}") # Si no trae tzinfo, se asume la zona local if dt_naive.tzinfo is None: dt_local = dt_naive.replace(tzinfo=ZoneInfo(local_tz)) else: dt_local = dt_naive # Convertir a UTC return dt_local.astimezone(ZoneInfo("UTC"))