#INVERMAR SCRIPT from datetime import datetime, timedelta import pymongo # CRUD Sensores import copy import re import os # Rerportería - Excel import openpyxl from openpyxl.utils import get_column_letter from openpyxl.styles import Border, Side, PatternFill, Font, GradientFill, Alignment from openpyxl.chart import LineChart, Reference from openpyxl.drawing.image import Image as ExcelImg from openpyxl.drawing.text import CharacterProperties from openpyxl.utils import get_column_letter, column_index_from_string from openpyxl.workbook.protection import WorkbookProtection import ctypes from ctypes import create_unicode_buffer import matplotlib from pathlib import Path password='itg123chile' KEYS_INTERES =["hora", "O2", "SO2", "TEMP", "RLY", "Start"] KEYS_INTERES_AUX =["hora", "O2", "SO2", "Temp", "RLY", "Start"] # # ---- Conexion a bases de datos de usuarios, ciclos y servicios---- # # #Conexion a Servidor local de MongoDB # client = pymongo.MongoClient("mongodb://localhost:27017/") # # Bases de datos globales, compartidas entre clientes. Cada cliente es una coleccion # nanoox_database_name = "NANOOX" # nanoox_db = client[nanoox_database_name] # users_collection = nanoox_db.USERS HISTORIAL_CONTROL_ARRAY_LEN_TO_SEND = 50 #!----FUNCIONES NANOOX----! def editar_usuario(): """ Editar usuario: actualizacion de datos de usuario en la base de datos. Solo se permite modificar 'cargo', 'correo', y 'accesos'. :param nuevos_datos: Diccionario con los nuevos datos. Formato esperado: { "cargo": "nuevo cargo", "correo": "nuevo_correo@example.com", "accesos": {} } :param users_collection: Coleccion de usuarios en la base de datos. :param username: Nombre de usuario a actualizar. :return: Codigo de estado de la operacion. """ # Obtener los datos enviados en la solicitud POST nuevos_datos = request.json # Buscar el usuario actual en la base de datos current_user = get_jwt_identity() user = users_collection.find_one({"username": current_user}) if not user: return jsonify({'error': f'El usuario {current_user} no existe'}), 401 # Preparar el diccionario de actualización campos_a_actualizar = ['cargo', 'correo', 'accesos'] actualizacion = {"$set": {}} # Iterar sobre los campos permitidos y actualizar solo si están presentes en nuevos_datos for campo in campos_a_actualizar: if campo in nuevos_datos: actualizacion["$set"][campo] = nuevos_datos[campo] # Verificar si hay algo que actualizar if not actualizacion["$set"]: return jsonify({'error': 'No hay datos válidos para actualizar'}), 400 # Realizar la actualización resultado = users_collection.update_one({'username': current_user}, actualizacion) # Verificar el resultado de la actualización if resultado.modified_count > 0: return jsonify({'mensaje': f"Usuario '{current_user}' actualizado correctamente."}), 200 else: return jsonify({'error': f"Usuario '{current_user}' no encontrado o no se realizaron cambios."}), 400 #Comprueba si es la data recibida es un diccionario o un diccionario anidado #i! # Deprecated original # def tipo_de_datos(datos): # try: # llaves = datos.get('data').keys() # es_simple = 'bombas' in llaves # es_anidado = '0' in llaves # if es_simple: # tipo = "0" # elif es_anidado: # tipo = "1" # else: # tipo = "2" # return tipo # except: # return False def tipo_de_datos(datos): """ Determina el formato de 'data' dentro del JSON recibido. Retorna: - "0" si 'data' es un diccionario (formato simple). - "nested" si 'data' es una lista (formato anidado). - None en otros casos. """ data = datos.get('data') if isinstance(data, dict): return "dict" elif isinstance(data, list): return "nested" return None def process_sensor_data(db, data, centro_id): """ Procesa un único registro de datos extrayendo cada bloque y guardándolo en la BD. """ print("en process_sensor_data") print(data) # Procesar datos de CAN's for can in data.get('cans', []): save_one_can_data(db, can) # Procesar datos de Bombas for bomba in data.get('bombas', []): save_one_bomb_data(db, bomba) # Procesar historial de control for log in data.get('historial_control', []): save_one_historial_data(db, log, centro_id) # Procesar generador (si existe) generador = data.get('generador') if generador: save_one_generator_data(db, centro_id, generador) # Se elimnó el argumento salmonera #i! def get_center_ids(db, region, area, centro): """ Esta funcion recupera los identificadores de CAN, bombas y el historial de control para un centro especifico, dentro de una region, area; de una salmonera especifica. Args: db: Base de datos o sesion de conexion de donde se recuperan los metadatos. salmonera: El nombre de la salmonera de la cual recuperar los metadatos. region: La region especifica en la que se encuentra el centro. area: El area especifica dentro de la region. centro: El centro especifico dentro del area. Returns: Tuple (can_ids, bomb_ids, historial_control_ids) donde: - can_ids es un diccionario que mapea el nombre de cada modulo a una lista de IDs de CAN. - bomb_ids es un diccionario que mapea el nombre de cada modulo a una lista de IDs de bombas. - historial_control_ids es una lista de identificadores de historial de control obtenidos de cada aviot. Esta funcion asume que la estructura de 'existing_metadata' es consistente y que los campos requeridos estan siempre presentes. Si hay posibilidad de que algunos campos puedan no existir, deberian manejarse con comprobaciones adicionales. """ # Inicializamos los diccionarios que contendran los ids can_ids = {} bomb_ids = {} historial_control_ids = [] # Obtener los metadatos existentes existing_metadata = get_metadata(db) # Verificar si hay metadatos, si no los hay, retornamos diccionarios vacios if not existing_metadata: return can_ids, bomb_ids, historial_control_ids # Recorremos las regiones en busca de la solicitada for region_node in existing_metadata.get('regiones', []): if region_node['nombre'] == region: # Recorremos las areas de la region encontrada for area_node in region_node.get('areas', []): if area_node['nombre'] == area: # Recorremos los centros del area encontrada for centro_node in area_node.get('centros', []): if centro_node['nombre'] == centro: # Recorremos las Aviots del centro encontrado for aviot_node in centro_node.get('aviots', []): # Aqui obtenemos el historial de control directamente de aviot_node if aviot_node.get('historial_control'): historial_control_ids.append( aviot_node.get('historial_control') ) # Recorremos los modulos de cada Aviot for modulo_node in aviot_node.get('modulos', []): modulo_name = modulo_node['nombre'] # Recorremos las CANs y las agregamos a su respectivo modulo for can_node in modulo_node.get('cans', []): can_list = can_ids.setdefault(modulo_name, []) can_list.append(can_node['can_id']) # Recorremos las bombas y las agregamos a su respectivo modulo for bomba_node in modulo_node.get('bombas', []): bomb_list = bomb_ids.setdefault(modulo_name, []) bomb_list.append(bomba_node['bomba_id']) # Retornamos los tres resultados obtenidos return can_ids, bomb_ids, historial_control_ids # Se eliminó el argumento CLIENT_NAME #i! def _get_last_cand_and_bomb_data( nanoox_db, can_ids, bomb_ids ): """ Intenta obtener los datos de los dias anteriores, hasta un maximo de 'max_days'. """ # Días máximos a recorrer #max_days = 2 # Obtener fecha de hoy fecha = datetime.now() date = ( fecha ).strftime("%Y-%m-%d") # Diccionario de datos para enviar al front modulos = {} # Procesar CANs y obtener datos de sensores for modulo_id, can_id_array in can_ids.items(): # Si no existe el mdulo en 'modulos', inicializarlo if modulo_id not in modulos: modulos[modulo_id] = {'id': modulo_id, 'cans': [], 'bombas': []} for can_id in can_id_array: can_data = get_can_data(nanoox_db, can_id, date, date) # print(f'can_data: {can_data}') try: # Asumimos que queremos el ultimo sensor de la ultima CAN try: can_data[0]['sensores'][-1]['bombas'] = can_data[0]['bombas'][-1] except Exception as e: print(f"Error en nueva version de get_can_data: \n{e}") modulos[modulo_id]['cans'].append(can_data[0]['sensores'][-1]) except IndexError: # Error si 'can_data' est vacio o no tiene 'sensores' print(f"Error obteniendo datos de CAN: {can_id}") pass # Procesar Bombas y obtener datos de sensores for modulo_id, bomb_id_array in bomb_ids.items(): # Si no existe el modulo en 'modulos', inicializarlo if modulo_id not in modulos: modulos[modulo_id] = {'id': modulo_id, 'cans': [], 'bombas': []} # Anadir Bombas al mdulo correspondiente for bomb_id in bomb_id_array: bomb_data = get_bomb_data(nanoox_db, bomb_id, date, date) # print(f"{bomb_data=}") try: # Asumimos que queremos el ltimo sensor de la ltima Bomba modulos[modulo_id]['bombas'].append(bomb_data[0]['sensores'][-1]) except (IndexError, KeyError): # Error si 'bomb_data' est vaco o no tiene 'sensores' print(f"Error obteniendo datos de Bomba: {bomb_id}") pass return modulos # *----------- CRUD DB Sensores ---------* # # ? Almacenar datos #i! def save_one_bomb_data(db, sensor_data_json): # Extraer informacin relevante del JSON recibido region = sensor_data_json["region"] area = sensor_data_json["area"] centro = sensor_data_json["centro"] aviot = sensor_data_json["aviot"] modulo = sensor_data_json["modulo"] bomba = sensor_data_json["bomba"] fecha = sensor_data_json["fecha"] sensores = sensor_data_json["sensores"] control = sensor_data_json["control"] # print(f"") bomba_id = f"{region}-{area}-{centro}-{aviot}-{modulo}-{bomba}" # print(f" {bomba_id=}") if not all([fecha, sensores, area, centro, control, aviot, modulo, bomba]): print("Falta informacion importante en los datos recibidos.") # logging.error("Falta informacin importante en los datos recibidos.") return try: # Asegura que los metadatos existen y son correctos # Internamente, se guarda en la coleccion db[f'{salmonera}_metadata'] # _add_or_update_nested_bomb_structure(db, salmonera, region, area, centro, aviot, modulo, bomba, bomba_id) _add_or_update_nested_bomb_structure(db, region, area, centro, control, aviot, modulo, bomba, bomba_id) # Preparar el objeto sensor_data para agregar a la coleccion de sensores sensor_data = { "fecha": fecha, "bomba_id": bomba_id, "sensores": sensores } # Realiza la operacion de upsert directamente, sin necesidad de verificar si los datos existen previamente # upsert_result = _upsert_bomb_sensor_data(db, salmonera, sensor_data["fecha"], sensor_data["bomba_id"], sensor_data["sensores"]) upsert_result = _upsert_bomb_sensor_data(db, sensor_data["fecha"], sensor_data["bomba_id"], sensor_data["sensores"]) # print(f"Datos de sensores actualizados correctamente para {salmonera} / {bomba} en {fecha}.") # logging.info(f"Datos de sensores actualizados correctamente para {salmonera} / {bomba} en {fecha}.") print(f"Datos de sensores actualizados correctamente para {region}-{area}-{centro} / {bomba} en {fecha}.") # logging.info(f"Datos de sensores actualizados correctamente para {salmonera} / {bomba} en {fecha}.") except Exception as e: print(f"Error al actualizar los datos para {region}-{area}-{centro} / {bomba} en {fecha}.") print(f"Error al actualizar los datos en linea: {e.__traceback__.tb_lineno}: \n{e}") # logging.error(f"Error al actualizar los datos: {e}") return def save_one_can_data( db, sensor_data_json ): # Extraer informacin relevante del JSON recibido region = sensor_data_json.get("region") area = sensor_data_json.get("area") centro = sensor_data_json.get("centro") aviot = sensor_data_json.get("aviot") modulo = sensor_data_json.get("modulo") can = sensor_data_json.get("can") # IP bombas = sensor_data_json.get("bombas") # Lista de apodos de bombas fecha = sensor_data_json.get("fecha") sensores = sensor_data_json.get("sensores") control = sensor_data_json.get("control") source = sensor_data_json.get("source") print(f'{can=}') print(f'{source=}') print(f'raw {sensores=}') sensores.update({"source":source}) print(f'updated {sensores=}') # print(f'can {can=}') # Revisar que se obtuvieron todos los datos if not all([fecha, sensores, region, area, centro, aviot, modulo, can]): print("Falta informacion importante en los datos recibidos.") # logging.error("Falta informacin importante en los datos recibidos.") return # Construir id de can can_id = f"{region}-{area}-{centro}-{aviot}-{modulo}-{can}" # print(f'{can_id=}') try: # Asegura que los metadatos existen y son correctos # _add_or_update_nested_can_structure(db, salmonera, region, area, centro, aviot, modulo, can, can_id) _add_or_update_nested_can_structure(db, region, area, centro, control, aviot, modulo, can, can_id) # Crear o actualizar datos en coleccion # _upsert_can_sensor_data(db, salmonera, can_id, fecha, sensores, bombas) _upsert_can_sensor_data(db, can_id, fecha, sensores, bombas) # print(f"Datos de sensores actualizados correctamente para {salmonera} / CAN: {can} en {fecha}.") print(f"Datos de sensores actualizados correctamente para {region}-{area}-{centro} / CAN: {can} en {fecha}.") # logging.info(f"Datos de sensores actualizados correctamente para {salmonera} / {bomba} en {fecha}.") except Exception as e: print(f"Error al actualizar los datos: {e}") # logging.error(f"Error al actualizar los datos: {e}") #i! def save_one_historial_data( db, historial_data, centro_id): # region, area, centro = None, None, None # for sensor_data_json in all_bombs_data: # print(f"{sensor_data_json=}") # region = sensor_data_json.get("region") # area = sensor_data_json.get("area") # centro = sensor_data_json.get("centro") # # Se queda con el primer conjunto de elementos que existan # if all( [ region, area, centro] ): # break # # print(f"{region}- {area} - {centro}") # if not all( [region, area, centro] ): return try: # ip_aviot = historial_data.get('ip_aviot') # msg = historial_data.get('msg') # date = historial_data.get('date') # Revisar que se obtuvieron todos los datos # if not all([ip_aviot, msg, date]): # print("Falta informacion importante en los datos recibidos.") # return # Construir id de aviot # centro_id = f"{region}-{area}-{centro}" #-{ip_aviot} # data_to_add = { # 'date' : date, # 'msg' : msg, # } # Se asume que los metadatos existen y son correctos para la aviot # Crear o actualizar datos en coleccion # _upsert_historial_control_data( db, centro_id, historial_data ) _add_to_historial_control( db, centro_id, historial_data ) print(f"Datos de historial de control actualizados correctamente para Centro: {centro_id}.") except Exception as e: print(f"Error al actualizar los datos: {e}") # logging.error(f"Error al actualizar los datos: {e}") # def save_one_generator_data(db, centro_id, data): # date = datetime.now().strftime("%Y-%m-%d") # # Define la consulta para buscar si existen datos previos para la misma bomba_id y fecha # query = { # "centro_id": centro_id, # "date": date # } # # Define la actualizacion a realizar, estableciendo los nuevos datos si se inserta un documento nuevo, # # o agregando los datos del sensor al array existente si el documento ya existe. # update = { # "$setOnInsert": { # "date": date, # "centro_id": centro_id, # }, # "$push": { # "generador": data, # } # } # try: # result = db["generadores"].update_one(query, update, upsert=True) # except Exception as e: # print(f'Error al guardar datos de generador en centro {centro_id} {e.__traceback__.tb_lineno}: \n{e} ') # return result def save_one_generator_data(db, centro_id, data): #fecha, if data: if set(data.values()) == {None}: return print(f"{data=}") date = datetime.now().strftime("%Y-%m-%d") # Define la consulta para buscar si existen datos previos para la misma bomba_id y fecha query = { "centro_id": centro_id, "date": date } existing_doc = db.generadores.find_one(query,{'generador.timestamp': False}) print(f"{existing_doc=}") if existing_doc: alredy_data = existing_doc['generador'][-1] print(f"{alredy_data=}") new_data = {k: v for k, v in data.items() if k != 'timestamp'} print(f"{new_data=}") # Verificar si los datos son diferentes if alredy_data != new_data: # Preparar la operación de actualización (upsert) update_operation = pymongo.UpdateOne( {'_id': existing_doc['_id']}, {"$push": {"generador": data}}) try: # result = db["generadores"].update_one(query, update, upsert=True) result = db.generadores.bulk_write([update_operation]) print(f"Upsert realizado. Documentos modificados: {result.modified_count}, documentos insertados: {result.upserted_count}") except Exception as e: print(f'Error al guardar datos de generador en centro {centro_id} {e.__traceback__.tb_lineno}: \n{e} ') else: print("Los datos son idénticos, no se realizó ningún cambio.") else: # Si no hay documento existente, insertar uno nuevo db.generadores.insert_one({'centro_id': centro_id, "date":date, 'generador': [data]}) print("Nuevo documento insertado.") else: print("No hay datos de generadores: {data}") return def get_generator(db, centro_id): """ Obtiene todos los logs de la colección 'historial_control' y los organiza en un diccionario. Las llaves del diccionario son las fechas (en formato 'YYYY-MM-DD') y los valores son listas de logs. :return: Diccionario con las fechas como llaves y listas de logs como valores. """ try: collection = db.generadores date = datetime.now().strftime("%Y-%m-%d") documents = collection.find({"centro_id":centro_id,"date": date }) generador_data = None for doc in documents: generador_data = doc['generador'][-1] return generador_data except Exception as e: print(f"Error al obtener datos de generador para centro {centro_id}': {e} {e.__traceback__.tb_lineno}") raise # ? Crear Metadatas # Revisar y actualizar Metadata de Bomba # Se eliminó el argumento nombre salmonera def _add_or_update_nested_bomb_structure(db, region, area, centro, control, aviot, modulo, bomba, bomba_id): # Obtener los metadatos existentes # existing_metadata = get_metadata(db) # existing_metadata = verify_center_metadata(db, region, area, centro) print("+++++++DENTRO DE BOMBAS++++++++") # print(f"{}") existing_metadata = check_bomb_exists(db, region, area, centro, aviot, modulo, bomba) print(f"{existing_metadata=}") metadata_db = get_metadata( db) # print(f"{metadata_db=}") # bomb_check = tiene_bombas(metadata_db) # print(f"{bomb_check=}") # print("###############HASTA QUI TODO BIEN EN BOMBAS############") # if existing_metadata: if metadata_db: metadata_id = metadata_db["_id"] # Si no hay metadatos, crear nuevos y almacenarlos if existing_metadata is None and metadata_db is None: print("***ENTRANDO A CREAR BOMBAS METADATA ****") new_metadata = _make_new_bomb_metadata( region, area, centro, control, aviot, modulo, bomba, bomba_id) create_metadata(db, new_metadata) return elif existing_metadata is None and metadata_db: # elif existing_metadata is None and bomb_check: print("*****ENTRANDO A ACTUALIZAR DATOS DE BOMBAS YA CREADAS******") new_metadata =_make_new_bomb_metadata( region, area, centro, control, aviot, modulo, bomba, bomba_id) actualizar_metadata_existente(db, region, area, centro, new_metadata) return # print("###############SI VES ESTO EL ERROR ES DESPUES############") # Crear una copia profunda para realizar modificaciones sin alterar los originales modified_metadata = copy.deepcopy(metadata_db) # Encuentra o crea el nodo de 'region' region_node = _find_or_create_node(modified_metadata.setdefault('regiones', []), 'nombre', region) # Encuentra o crea el nodo de 'area' area_node = _find_or_create_node(region_node.setdefault('areas', []), 'nombre', area) # Encuentra o crea el nodo de 'centro' centro_node = _find_or_create_node(area_node.setdefault('centros', []), 'nombre', centro) # Encuentra o crea el nodo de 'aviot' aviot_node = _find_or_create_node(centro_node.setdefault('aviots', []), 'nombre', aviot, {'historial_control': f"{region}-{area}-{centro}-{aviot}"}) # Encuentra o crea el nodo de 'modulo' modulo_node = _find_or_create_node(aviot_node.setdefault('modulos', []), 'nombre', modulo) # Encuentra o crea el nodo de 'bomba' con bomba_id _ = _find_or_create_node(modulo_node.setdefault('bombas', []), 'nombre', bomba, {'bomba_id': bomba_id}) # Comprobar si la estructura se ha modificado y actualizar si es necesario # if modified_metadata != existing_metadata: # # Actualizar el documento en la base de datos # update_metadata(db, metadata_id, modified_metadata) # print(f"Metadata {bomba_id} añadida a cliente {region}-{area}-{centro}") return # def verify_center_metadata(db, region, area, centro): # metadata = get_metadata( db) # print(f"{metadata=}") # if metadata: # regiones = metadata.get("regiones") # region_names = [region_name['nombre'] for region_name in regiones] # print(f"{region_names=}") # # print(f'{regiones=}') # if region not in region_names: # return None # for region_data in regiones: # areas = region_data.get("areas") # areas_names = [areas_name['nombre'] for areas_name in areas] # print(f'{areas_names=}') # if area not in areas_names: # return None # centros_names = [area_data['centros'][0]['nombre'] for area_data in areas] # print(f'{centros_names=}') # if centro not in centros_names: # return None # else: # # print("TRUH!") # return True # else: # return None def verify_center_metadata(db, region, area, centro): metadata = get_metadata(db) if metadata: regiones = metadata.get("regiones") region_names = [region_name['nombre'] for region_name in regiones] print(f"{region_names=}") if region not in region_names: return None for region_data in regiones: if region_data['nombre'] == region: areas = region_data.get("areas") areas_names = [areas_name['nombre'] for areas_name in areas] print(f'{areas_names=}') if area not in areas_names: return None for area_data in areas: if area_data['nombre'] == area: centros_nombres = [centro_data['nombre'] for centro_data in area_data.get('centros', [])] print(f"{centros_nombres=}") if centro not in centros_nombres: return None else: return True return None def check_bomb_exists(db, region, area, centro, aviot, modulo, bomba_nombre): """ Verifica si existe una bomba con 'bomba_nombre' dentro de: regiones -> areas -> centros -> aviots -> modulos -> bombas Retorna True si la bomba existe, o None si no la encuentra. """ metadata = get_metadata(db) if not metadata: return None # 1) Listamos las regiones regiones = metadata.get("regiones", []) for region_data in regiones: if region_data["nombre"] == region: # 2) Buscar el área for area_data in region_data.get("areas", []): if area_data["nombre"] == area: # 3) Buscar el centro for centro_data in area_data.get("centros", []): if centro_data["nombre"] == centro: # 4) Buscar el aviot for aviot_data in centro_data.get("aviots", []): if aviot_data["nombre"] == aviot: # 5) Buscar el módulo for modulo_data in aviot_data.get("modulos", []): # Ojo: a veces 'módulo' es string, otras int. # Convertimos a string para coherencia: if str(modulo_data["nombre"]) == str(modulo): # 6) Finalmente, revisar si existe la bomba bombas = modulo_data.get("bombas", []) for bomba_data in bombas: if bomba_data["nombre"] == bomba_nombre: return True # Si el flujo termina sin encontrar la bomba, retornamos None return None def tiene_bombas(metadata_db): """ Retorna True si, dentro de la estructura de datos: metadata_db['regiones'][0]['areas'][0]['centros'][0]['aviots'][0]['modulos'][n] existe al menos un módulo que tenga la clave 'bombas'. De lo contrario, retorna False. """ # Usamos .get(key, default) y listas vacías como "fallback" para evitar errores # en caso de que alguna clave no exista o la lista esté vacía. modulos = ( metadata_db .get('regiones', [{}])[0] .get('areas', [{}])[0] .get('centros', [{}])[0] .get('aviots', [{}])[0] .get('modulos', []) ) # Revisamos cada módulo. Si alguno tiene el campo 'bombas', devolvemos True. for modulo in modulos: if 'bombas' in modulo: return True # Si terminamos el bucle sin encontrar 'bombas' en ningún módulo, devolvemos False. return False def check_can_exists(db, nombre_region, nombre_area, nombre_centro, nombre_aviot, nombre_modulo, can_ip): # Obtener la metadata desde la base de datos metadata = get_metadata(db) if metadata: # Iterar sobre las regiones for region_data in metadata.get("regiones", []): if region_data["nombre"] == nombre_region: # Iterar sobre las áreas de la región for area_data in region_data.get("areas", []): if area_data["nombre"] == nombre_area: # Iterar sobre los centros del área for centro_data in area_data.get("centros", []): if centro_data["nombre"] == nombre_centro: # Iterar sobre los aviots del centro for aviot_data in centro_data.get("aviots", []): if aviot_data["nombre"] == nombre_aviot: # Iterar sobre los módulos del aviot for modulo_data in aviot_data.get("modulos", []): if modulo_data["nombre"] == nombre_modulo: for can in modulo_data.get("cans", []): if can['can'] == can_ip: # Verificar si existe el campo "can" en el módulo # if "cans" in modulo_data and modulo_data["cans"]: return True else: return None return None # Revisar y actualizar Metadata de CAN # Se eliminó el argumento salmonera def _add_or_update_nested_can_structure(db, region, area, centro, control, aviot, modulo, can, can_id): # print(f'_add_or_can_id: {can_id=}') # Obtener los metadatos existentes # existing_metadata = get_metadata(db) # existing_metadata = verify_center_metadata(db, region, area, centro) existing_metadata = check_can_exists(db, region, area, centro, aviot, modulo, can) # print(f"{existing_metadata=}") metadata_db = get_metadata( db) # print(f"{metadata_db=}") # print(f"{metadata_db['_id']=}") if metadata_db: # if existing_metadata: metadata_id = metadata_db["_id"] # print(metadata_id) # Si no hay metadatos, crear nuevos if existing_metadata is None and metadata_db is None: print('------------entrando a crear nueva metadata-------------') new_metadata = _make_new_can_metadata( region, area, centro, control, aviot, modulo, can, can_id) create_metadata(db, new_metadata) # update_metadata(db, metadata_id ,new_metadata) return elif existing_metadata is None and metadata_db: print('------------entrando a actualizar con nueva metadata-------------') upsert_can_metadata(db.metadata, metadata_id, region, area, centro, control, aviot, modulo, can, can_id) # print("Upsert realizado con éxito.") # new_metadata = _make_new_can_metadata( region, area, centro, control, aviot, modulo, can, can_id) # new_can_info = new_metadata['regiones'][0]['areas'][0]['centros'][0]['aviots'][0]['modulos'][0]['cans'][0] # agregar_informacion_de_centro_can(db, region, area, centro, aviot, modulo, new_can_info) # actualizar_metadata_existente(db, region, area, centro, new_metadata) return # Crear una copia profunda para realizar modificaciones modified_metadata = copy.deepcopy(metadata_db) # Navegar o crear 'region' region_node = _find_or_create_node(modified_metadata.setdefault('regiones', []), 'nombre', region) # Navegar o crear 'area' area_node = _find_or_create_node(region_node.setdefault('areas', []), 'nombre', area) # Navegar o crear 'centro' centro_node = _find_or_create_node(area_node.setdefault('centros', []), 'nombre', centro) # Navegar o crear 'aviot' aviot_node = _find_or_create_node(centro_node.setdefault('aviots', []), 'nombre', aviot, {'historial_control': f"{region}-{area}-{centro}-{aviot}"}) # Navegar o crear 'modulo' modulo_node = _find_or_create_node(aviot_node.setdefault('modulos', []), 'nombre', modulo) # Navegar o crear 'can' con can_id _ = _find_or_create_node(modulo_node.setdefault('cans', []), 'can', can , {'can_id': can_id} ) # Comprobar si la estructura se ha modificado # if modified_metadata != existing_metadata: # # Actualizar el documento en la base de datos # update_metadata(db, metadata_id ,modified_metadata) # # print(f'Can Metadata de {salmonera} actualizada exitosamente') # print(f"Metadata {can_id} añadida a cliente {region}-{area}-{centro}") return # Funcion para insertar en DB nueva metadata # Se eliminó el argumento nombre_salmonera def create_metadata(db, metadata): """ Inserta metadatos en la coleccin de metadatos. """ # db[f'{nombre_salmonera}_metadata'].insert_one(metadata) db.metadata.insert_one(metadata) # Funcion para actualizar en DB metadata existente # Se eliminó el argumento nombre_salmonera #Se agregó el argumento metadata_id def update_metadata(db, metadata_id, new_document): """ Actualiza los metadatos en la coleccin de metadatos. """ # db[f'{nombre_salmonera}_metadata'].replace_one({'nombre': nombre_salmonera}, new_document) db.metadata.replace_one({'_id': metadata_id}, new_document) #db.metadata.replace_one(new_document) # Funcion de apoyo para crear un nodo especifico faltante def _find_or_create_node(node_list, key, value, extra_data=None): """ Encuentra un nodo en una lista de diccionarios o crea uno nuevo si no existe. """ node = next((item for item in node_list if item[key] == value), None) if node is None: new_node = {key: value} if extra_data: new_node.update(extra_data) node_list.append(new_node) return new_node return node def upsert_can_metadata( collection, doc_id, # El _id del documento donde se guardará todo region, area, centro, control, aviot, modulo, can, can_id ): """ Inserta o actualiza (upsert) la información en la topología dada: - region - area - centro (con control) - aviot (con historial_control) - modulo - can (con can_id) Solo se agrega lo que no existe, manteniendo la estructura original. """ # 1) Buscar el documento principal por _id (o el criterio que necesites). doc = collection.find_one({"_id": doc_id}) # Si no existe, iniciamos una estructura base. if not doc: doc = { "_id": doc_id, "regiones": [] } # Extraemos la lista de regiones. regiones = doc.get("regiones", []) # 2) Buscar/crear la región region_obj = next((r for r in regiones if r["nombre"] == region), None) if not region_obj: region_obj = { "nombre": region, "areas": [] } regiones.append(region_obj) # 3) Buscar/crear el área areas = region_obj.get("areas", []) area_obj = next((a for a in areas if a["nombre"] == area), None) if not area_obj: area_obj = { "nombre": area, "centros": [] } areas.append(area_obj) # 4) Buscar/crear el centro centros = area_obj.get("centros", []) centro_obj = next((c for c in centros if c["nombre"] == centro), None) if not centro_obj: centro_obj = { "nombre": centro, "control": control, # Guardamos el control si no existía. "aviots": [] } centros.append(centro_obj) else: # Si deseas sobreescribir o setear el "control" en caso de que no exista, # podrías hacer algo como: if "control" not in centro_obj: centro_obj["control"] = control # 5) Buscar/crear el aviot aviots = centro_obj.get("aviots", []) aviot_obj = next((av for av in aviots if av["nombre"] == aviot), None) if not aviot_obj: aviot_obj = { "nombre": aviot, "historial_control": f"{region}-{area}-{centro}-{aviot}", "modulos": [] } aviots.append(aviot_obj) # 6) Buscar/crear el módulo modulos = aviot_obj.get("modulos", []) # Ojo que en tu ejemplo el "nombre" del módulo a veces es string y otras veces int. # Para mayor coherencia, podrías unificarlo a string siempre, o int siempre. # Aquí lo convertimos a string para simplificar: modulo_str = str(modulo) modulo_obj = next((m for m in modulos if str(m["nombre"]) == modulo_str), None) if not modulo_obj: modulo_obj = { "nombre": modulo_str, # "bombas": [], # Si quisieras también manejar bombas en el upsert, lo harías aquí. "cans": [] } modulos.append(modulo_obj) # 7) Finalmente, buscar/crear el can cans = modulo_obj.get("cans", []) can_obj = next((c for c in cans if c["can"] == can), None) if not can_obj: can_obj = { "can": can, "can_id": can_id } print(f"{can_obj=}") cans.append(can_obj) # Guardamos la lista de regiones actualizada en el doc doc["regiones"] = regiones # 8) Hacemos upsert (replace_one con upsert=True) para actualizar el documento completo. collection.replace_one({"_id": doc_id}, doc, upsert=True) # ? Crear metadatos para Salmonera # Crear Metadata de Bomba # Se eliminó el argumento salmonera def _make_new_bomb_metadata(region, area, centro, control, aviot, modulo, bomba, bomba_id): new_metadata = { "regiones": [ { "nombre": region, "areas": [ { "nombre": area, "centros": [ { "nombre": centro, "control": control, "aviots": [ { "nombre": aviot, "historial_control": f"{region}-{area}-{centro}-{aviot}", "modulos": [ { "nombre": str(modulo), # "bombas": [{"nombre": bomba}] "bombas": [{"nombre": bomba, "bomba_id": bomba_id}] } ] } ] } ] } ] } ] } return new_metadata # Crear Metadata de CAN # Se eliminó el argumento salmonera def _make_new_can_metadata( region, area, centro, control, aviot, modulo, can, can_id): print(f"*/*/* ENTRANDO A CREAR CAN con AVIOT {aviot} */*/*") new_metadata = { "regiones": [ { "nombre": region, "areas": [ { "nombre": area, "centros": [ { "nombre": centro, "control": control, "aviots": [ { "nombre": aviot, "historial_control": f"{region}-{area}-{centro}-{aviot}", "modulos": [ { "nombre": str(modulo), "cans": [ {"can": can , "can_id": can_id} ] } ] } ] } ] } ] } ] } print(f'{new_metadata=}') return new_metadata # ? Obtener metadatos para Salmonera # Se eliminó argumento nombre_salmonera def get_metadata(db): """ Obtiene los metadatos de la coleccin de metadatos. """ # metadata = db[f'{nombre_salmonera}_metadata'].find_one() metadata = db.metadata.find_one() # if metadata is None: # print( f"No se encontraron metadatos para la salmonera: {nombre_salmonera}" ) # raise Exception(f"No se encontraron metadatos para la salmonera: {nombre_salmonera}") return metadata #? Actualizar metadata segun region, area y centro def coincidencias_metadata(db, region, area, centro): metadata = get_metadata( db) c_regiones = [] c_areas = [] c_centros= [] coincidencias = [] if metadata: regiones = metadata.get("regiones") region_names = [region_name['nombre'] for region_name in regiones] # print(f"{region_names=}") # print(f'{regiones=}') if region not in region_names: c_regiones.append(None) # return None else: c_regiones.append(True) for region_data in regiones: areas = region_data.get("areas") areas_names = [areas_name['nombre'] for areas_name in areas] # print(f'{areas_names=}') if area not in areas_names: c_areas.append(None) # return None else: c_areas.append(True) for area_data in areas: centros = area_data.get("centros") centros_names = [centros_name['nombre'] for centros_name in centros] # print(f'{centros_names=}') if centro not in centros_names: c_centros.append(None) # return None else: c_centros.append(True) # print("TRUH!") # return True coincidencias.append(any(c_regiones)) coincidencias.append(any(c_areas)) coincidencias.append(any(c_centros)) return coincidencias else: return [False, False, False] def agregar_informacion_de_centro(db, nombre_region, nombre_area, nueva_informacion): # Encuentra el documento que contiene el área query = { 'regiones.areas.nombre': nombre_area } # Define la actualización que quieres realizar update = { '$push': { 'regiones.$[reg].areas.$[area].centros': nueva_informacion } } # Especifica los parámetros de arrayFilters para buscar el área correcta array_filters = [ {'reg.nombre': nombre_region}, # Cambia este valor según la región correcta {'area.nombre': nombre_area} ] # Realiza la actualización result = db.metadata.update_one( query, update, array_filters=array_filters ) return result.modified_count def agregar_informacion_de_area(db, nombre_region, nueva_informacion_area): query = { 'regiones.nombre': nombre_region } update = { '$push': { 'regiones.$.areas' : nueva_informacion_area } } result = db.metadata.update_one( query, update, ) return result.modified_count def agregar_informacion_de_region(db, nueva_informacion_region): # Define la actualización que quieres realizar update = { '$push': { 'regiones': nueva_informacion_region } } # Realiza la actualización result = db.metadata.update_one( {}, update, ) return result.modified_count # def actualizar_metadata_existente(db, region, area, centro, new_metadata): # if coincidencias_metadata(db, region, area, centro)[0]: # print(f"Existe la region {region}") # else: # #Aqui la data se inserta tal cual make_metadata # # Se debe crear una nueva region e insertar los datos # print("No Existe la region.") # nueva_region = new_metadata['regiones'][0] # result = agregar_informacion_de_region(db, nueva_region) # if result == 1: # print(f"Se ha registrado la región {region} con éxito") # else: # print(f"No se ha registrado la región {region} con éxito") # if coincidencias_metadata(db, region, area, centro,)[1]: # print(f"Existe el área {area}") # else: # # Se debe crear una nueva area dentro de la region e insertar los datos # print("No Existe el área") # nueva_area = new_metadata['regiones'][0]['areas'][0] # result = agregar_informacion_de_area(db, region, nueva_area) # if result == 1: # print(f"Se ha registrado el área {area} con éxito") # else: # print(f"No se ha registrado el área {area} con éxito") # if coincidencias_metadata(db, region, area, centro,)[2]: # print(f"Existe el centro {centro}") # return # else: # nuevo_centro = new_metadata['regiones'][0]['areas'][0]['centros'][0] # # Se debe crear un nuevo centro dentro de la region-area e insertar los datos # print("No Existe el centro") # result = agregar_informacion_de_centro(db, region, area, nuevo_centro) # if result == 1: # print(f"Se ha registrado el centro {centro} con éxito") # else: # print(f"No se ha registrado el centro {centro} con éxito") # return def actualizar_metadata_existente(db, region, area, centro, new_metadata): # 1) Chequeo/creación de región if coincidencias_metadata(db, region, area, centro)[0]: print(f"Existe la región {region}") else: print("No Existe la región.") nueva_region = new_metadata['regiones'][0] result = agregar_informacion_de_region(db, nueva_region) if result == 1: print(f"Se ha registrado la región {region} con éxito") else: print(f"No se ha registrado la región {region} con éxito") # 2) Chequeo/creación de área if coincidencias_metadata(db, region, area, centro)[1]: print(f"Existe el área {area}") else: print("No Existe el área") nueva_area = new_metadata['regiones'][0]['areas'][0] result = agregar_informacion_de_area(db, region, nueva_area) if result == 1: print(f"Se ha registrado el área {area} con éxito") else: print(f"No se ha registrado el área {area} con éxito") # 3) Chequeo/creación de centro if coincidencias_metadata(db, region, area, centro)[2]: print(f"Existe el centro {centro}") else: print("No Existe el centro") nuevo_centro = new_metadata['regiones'][0]['areas'][0]['centros'][0] result = agregar_informacion_de_centro(db, region, area, nuevo_centro) if result == 1: print(f"Se ha registrado el centro {centro} con éxito") else: print(f"No se ha registrado el centro {centro} con éxito") # 4) [OPCIONAL] Insertar bombas sin perder 'cans' # ------------------------------------------------ # Supongamos que new_metadata trae: # 'regiones' -> [0] -> 'areas' -> [0] -> 'centros' -> [0] -> 'aviots' -> [0] -> 'modulos' -> [0] -> 'bombas' # Ejemplo: # new_metadata = { # "regiones": [{ # "nombre": region, # "areas": [{ # "nombre": area, # "centros": [{ # "nombre": centro, # "aviots": [{ # "nombre": "192.168.10.x", # "modulos": [{ # "nombre": "1", # "bombas": [ # {"nombre": "VA1", "bomba_id": "..."}, # {"nombre": "VA2", "bomba_id": "..."} # ] # ... # }] # }] # }] # }] # }] # } # Verificamos si hay aviots/modulos/bombas en new_metadata: try: aviot_data = new_metadata["regiones"][0]["areas"][0]["centros"][0]["aviots"][0] aviot_nombre = aviot_data["nombre"] modulo_data = aviot_data["modulos"][0] modulo_nombre = modulo_data["nombre"] nuevas_bombas = modulo_data.get("bombas", []) if nuevas_bombas: print(f"{nuevas_bombas=}") # Insertamos o "pusheamos" las nuevas bombas sin eliminar 'cans' result_bombas = agregar_bombas_al_modulo( db=db, region=region, area=area, centro=centro, aviot=aviot_nombre, modulo=modulo_nombre, nuevas_bombas=nuevas_bombas ) if result_bombas > 0: print(f"Se han agregado {result_bombas} actualización(es) de bombas con éxito.") else: print("No se agregaron bombas (o ya existían).") except (IndexError, KeyError): print("No se encontró info de aviots/modulos/bombas en new_metadata. Se omite la inserción de bombas.") return def agregar_bombas_al_modulo( db, region, area, centro, aviot, modulo, nuevas_bombas ): """ Inserta una o varias 'bombas' en la estructura: regiones -> areas -> centros -> aviots -> modulos -> bombas Sin eliminar o modificar 'cans' ni otros campos que pudieran existir en ese módulo. nuevas_bombas es una lista de objetos bomba, por ejemplo: [ {"nombre": "VA2", "bomba_id": "Region-Area-Centro-AVIOT-Modulo-VA2"}, {"nombre": "VA3", "bomba_id": "Region-Area-Centro-AVIOT-Modulo-VA3"} ] """ # 1) Construimos el query que ubica el modulo específico query = { "regiones.nombre": region, "regiones.areas.nombre": area, "regiones.areas.centros.nombre": centro, "regiones.areas.centros.aviots.nombre": aviot, "regiones.areas.centros.aviots.modulos.nombre": str(modulo) #str(modulo) } # 2) Usamos $push + $each para insertar las bombas en el array 'bombas' update = { "$push": { "regiones.$[r].areas.$[a].centros.$[c].aviots.$[av].modulos.$[m].bombas": { "$each": nuevas_bombas } } } # 3) Array filters para indicar en qué elemento(s) hacer $push array_filters = [ {"r.nombre": region}, {"a.nombre": area}, {"c.nombre": centro}, {"av.nombre": aviot}, {"m.nombre": str(modulo)} #str(modulo) ] # 4) Ejecutamos la actualización (update_one) con arrayFilters result = db.metadata.update_one( query, update, array_filters=array_filters ) return result.modified_count # def agregar_bombas_al_modulo( # db, # region, # area, # centro, # aviot, # modulo, # nuevas_bombas # ): # """ # Inserta una o varias 'bombas' en la estructura: # regiones -> areas -> centros -> aviots -> modulos -> bombas # Sin eliminar o modificar 'cans' ni otros campos que pudieran existir # en ese módulo. # nuevas_bombas es una lista de objetos bomba, por ejemplo: # [ # {"nombre": "VA2", "bomba_id": "Region-Area-Centro-AVIOT-Modulo-VA2"}, # {"nombre": "VA3", "bomba_id": "Region-Area-Centro-AVIOT-Modulo-VA3"} # ] # """ # # 1) Construimos el query que ubica el modulo específico # query = { # "regiones.nombre": region, # "regiones.areas.nombre": area, # "regiones.areas.centros.nombre": centro, # "regiones.areas.centros.aviots.nombre": aviot, # "regiones.areas.centros.aviots.modulos.nombre": str(modulo) # } # # 2) Usamos $push + $each para insertar las bombas en el array 'bombas' # update = { # "$push": { # "regiones.$[r].areas.$[a].centros.$[c].aviots.$[av].modulos.$[m].bombas": { # "$each": nuevas_bombas # } # } # } # # 3) Array filters para indicar en qué elemento(s) hacer $push # array_filters = [ # {"r.nombre": region}, # {"a.nombre": area}, # {"c.nombre": centro}, # {"av.nombre": aviot}, # {"m.nombre": str(modulo)} # ] # # ---------------------------- # # LOGS / DEBUG: # # ---------------------------- # print("========== DEBUG agregar_bombas_al_modulo ==========") # print("Query (para .find_one / .update_one):") # print(query) # print("Array Filters:") # print(array_filters) # print("Update (document):") # print(update) # print("-----------------------------------------------------") # # 4) Opcional: revisa si hay documentos que cumplan con ese query # documento_encontrado = db["metadata"].find_one(query) # if documento_encontrado is None: # print("No se encontró ningún documento que coincida con el query anterior.") # else: # print("Documento encontrado para ese query (mostrando solo _id):", documento_encontrado.get("_id")) # # 5) Ejecutamos la actualización (update_one) con arrayFilters # result = db["metadata"].update_one( # query, # update, # array_filters=array_filters # ) # # 6) Revisamos los resultados # print("matched_count :", result.matched_count) # print("modified_count:", result.modified_count) # if result.upserted_id: # print("upserted_id :", result.upserted_id) # print("=====================================================\n") # return result.modified_count def agregar_informacion_de_centro_can(db, nombre_region, nombre_area, nombre_centro, nombre_aviot, nombre_modulo, nueva_informacion): # Encuentra el documento que contiene el módulo específico query = { 'regiones.nombre': nombre_region, 'regiones.areas.nombre': nombre_area, 'regiones.areas.centros.nombre': nombre_centro, 'regiones.areas.centros.aviots.nombre': nombre_aviot, 'regiones.areas.centros.aviots.modulos.nombre': nombre_modulo } # Define la actualización que quieres realizar update = { '$push': { 'regiones.$[reg].areas.$[area].centros.$[centro].aviots.$[aviot].modulos.$[modulo].cans': nueva_informacion } } # Especifica los parámetros de arrayFilters para buscar el área correcta array_filters = [ {'reg.nombre': nombre_region}, {'area.nombre': nombre_area}, {'centro.nombre': nombre_centro}, {'aviot.nombre': nombre_aviot}, {'modulo.nombre': nombre_modulo} ] # Realiza la actualización result = db.metadata.update_one( query, update, array_filters=array_filters ) if result == 1: print(f"Se ha registrado la can del centro {nombre_centro} con éxito") else: print(f"No se ha registrado la can del centro {nombre_centro} ya que esta registrada") return # ? Crear o actualizar Data de sensores # Se elimnó el argumento nombre_salmonera def _upsert_bomb_sensor_data(db, fecha, bomba_id, sensores): """ Inserta o actualiza los datos de sensores en la coleccion de sensores para una bomba y fecha especificas. Utiliza una operacion de upsert para minimizar las interacciones con la base de datos y mejorar la eficiencia. Args: db: Base de datos o sesion de conexion de donde se manejan los datos. nombre_salmonera: El nombre de la salmonera a la que pertenecen los datos del sensor. fecha: La fecha correspondiente a los datos del sensor. bomba_id: El identificador de la bomba para la cual se estan insertando/actualizando los datos del sensor. sensores: Una lista de diccionarios que contiene los datos de los sensores a insertar/actualizar. Returns: Resultado de la operacion de insercion/actualizacion. """ result = None # Define la consulta para buscar si existen datos previos para la misma bomba_id y fecha query = { "bomba_id": bomba_id, "fecha": fecha } # Define la actualizacion a realizar, estableciendo los nuevos datos si se inserta un documento nuevo, # o agregando los datos del sensor al array existente si el documento ya existe. update = { "$setOnInsert": { "fecha": fecha, "bomba_id": bomba_id, }, "$push": { "sensores": { "$each": sensores if isinstance(sensores, list) else [sensores] # Suponiendo que 'sensores' es una lista de datos de sensores } } } # Realiza la operacion de upsert en la base de datos # result = db[f'{nombre_salmonera}_sensores'].update_one(query, update, upsert=True) try: result = db.sensores.update_one(query, update, upsert=True) except Exception as e: print(f'Error en upsert_bomb_sensor_data {e.__traceback__.tb_lineno}: \n{e} ') print(f'Error insertando datos en {bomba_id} con fecha {fecha}') # from pprint import pprint # print(f"{bomba_id=}") # pprint(f"{sensores=}") return result # Se elimnó el argumento nombre_salmonera def _upsert_can_sensor_data(db, can_id, fecha, sensores, bombas): """ Intenta actualizar los datos del sensor para un CAN, y si no existen, crea un nuevo documento. """ query = { "fecha": fecha, "can_id": can_id } update = { "$setOnInsert": { "can_id": can_id, "fecha": fecha, }, "$push": { "sensores": {"$each": sensores if isinstance(sensores, list) else [sensores]}, "bombas": {"$each": bombas if isinstance(bombas, list) else [bombas]} } } return db.sensores.update_one(query, update, upsert=True) # return db[f'{nombre_salmonera}_sensores'].update_one(query, update, upsert=True) #Se eliminó el argumento cliente def _upsert_historial_control_data(db, aviot_id, data_to_add): """ Intenta actualizar los datos del sensor para un CAN, y si no existen, crea un nuevo documento. """ query = { "aviot_id": aviot_id } update = { "$setOnInsert": { "aviot_id": aviot_id, }, "$push": { "historial_control": { "$each": data_to_add if isinstance(data_to_add, list) else [data_to_add] }, } } return db.historial_control.update_one(query, update, upsert=True) # return db[f'{cliente}_sensores'].update_one(query, update, upsert=True) def _add_to_historial_control(db, centro_id, data): """ Agrega un nuevo log al documento correspondiente a la fecha actual en la colección 'historial_control'. Si el documento de la fecha no existe, lo crea con el log como primer elemento en el arreglo 'logs'. :param db: Conexión a la base de datos MongoDB. :param data: Diccionario que contiene la información a guardar en el log. :return: Resultado de la operación de actualización/inserción. """ try: # data["enviado"] = False # Obtener la fecha actual en formato "YYYY-MM-DD" # current_date = datetime.now().strftime("%Y-%m-%d") current_date = data['date'].split(' ')[0] # Construir la consulta para buscar el documento de la fecha actual query = {"date": current_date} # Preparar la operación de actualización o inserción update = { "$setOnInsert": {"date": current_date}, # Si el documento no existe, inicializar la fecha "$setOnInsert": {"centro_id": centro_id}, "$push": {"logs": data} # Agregar el log al arreglo 'logs' } # Realizar la actualización con upsert result = db.historial_control.update_one(query, update, upsert=True) # Mensajes de resultado if result.matched_count > 0: print(f"Log agregado al documento existente de la fecha {current_date}.") else: print(f"Documento creado para la fecha {current_date} y log agregado.") return result except Exception as e: print(f"Error al agregar log a 'historial_control': {e}") raise # ? Obtener datos de Bombas / CAN # Obtener datos de sensores para una bomba especfica en un rango de fechas # Se eliminó el argumento nombre_salmonera def get_bomb_data(db, bomba_id, start_date, end_date): query = { "bomba_id": bomba_id, "fecha": {"$gte": start_date, "$lte": end_date} } return list(db.sensores.find(query)) # return list(db[f'{nombre_salmonera}_sensores'].find(query)) # Obtener datos de sensores para una CAN especfica en un rango de fechas # Se elimnó el argumento nombre_salmonera def get_can_data(db, can_id, start_date, end_date): """ Obtiene los datos de los sensores para una bomba especfica dentro de un rango de fechas. Args: db (object): Objeto de base de datos. nombre_salmonera (str): El nombre de la salmonera. can_id (str): La IP de la can. start_date (str): La fecha en el formato "YYYY-MM-DD". end_date (str): La fecha en el formato "YYYY-MM-DD". (Puede ser igual a start_date para obtener nico da) Returns: list: Lista de documentos que coinciden con la consulta. """ query = { "can_id": can_id, "fecha": {"$gte": start_date, "$lte": end_date} } return list(db.sensores.find(query)) # return list(db[f'{nombre_salmonera}_sensores'].find(query)) #Se elinió el argumento nombre_salmonera #i! # def get_historial_control(db, historial_control_id ): # # query = { # # "aviot_id": historial_control_id, # # } # query = { # "aviot_id": historial_control_id, # } # # Usar proyeccion con $slice para obtener solo los ultimos elementos # projection = { # "historial_control": {"$slice": -HISTORIAL_CONTROL_ARRAY_LEN_TO_SEND} # } # try: # # aviot_history_db = list(db[f'{nombre_salmonera}_sensores'].find(query, projection)) # aviot_history_db = list(db[f'sensores'].find(query, projection)) # if aviot_history_db and isinstance(aviot_history_db, list): # aviot_history = aviot_history_db[0].get('historial_control', []) # return aviot_history # except Exception as e: # print(f'Error obteniendo historial de control en linea {e.__traceback__.tb_lineno}: \n{e}') # return [] # ? No utilizado - Eliminar metadatos para Salmonera def delete_metadata(db, nombre_salmonera): """ Elimina todos los metadatos en la coleccin de metadatos. """ db[f'{nombre_salmonera}_metadata'].delete_many({}) # Actualizar datos de sensores para una bomba especfica def update_sensor_data(db, nombre_salmonera, bomba_id, date, sensor_updates): """ Actualiza los datos de sensores en la coleccin de sensores. """ query = { "bomba_id": bomba_id, "fecha": date } db[f'{nombre_salmonera}_sensores'].update_one(query, {"$push": {"sensores": sensor_updates}}) # * Eliminar datos de sensores e historial de control # Elimina todos los datos de un centro en especifico def eliminar_datos_centro(db, nombre_salmonera, region_nombre, area_nombre, centro_nombre): """ Elimina todos los datos de sensores asociados a un centro especifico, y actualiza la coleccion de metadata correspondientemente, eliminando el centro, el area si queda vacia, y la region si tambien queda vacia. Parametros: - db: La instancia de la base de datos de MongoDB. - nombre_salmonera: El nombre del cliente al cual acceder. - region_nombre: El nombre de la region del centro a eliminar. - area_nombre: El nombre del area del centro a eliminar. - centro_nombre: El nombre del centro a eliminar. """ # Buscar el documento de metadata metadata = get_metadata(db, nombre_salmonera) # Inicializar un array para almacenar los IDs de bombas y CANs a eliminar bomba_ids = [] can_ids = [] historial_control_ids = [] centro_encontrado = False area_vacia = False region_vacia = False try: # Buscar los IDs relevantes en la estructura de metadata for region in metadata['regiones']: if region['nombre'] == region_nombre: for area in region['areas']: if area['nombre'] == area_nombre: for centro in area['centros']: if centro['nombre'] == centro_nombre: centro_encontrado = True for aviot in centro['aviots']: try: historial_control_ids.append( aviot['historial_control'] ) except: print(f"Historial de control no encontrado") for modulo in aviot['modulos']: try: bomba_ids.extend([bomba['bomba_id'] for bomba in modulo['bombas']]) except: print(f"Bomba no encontrada") try: can_ids.extend([can['can_id'] for can in modulo['cans']]) except: print(f"CAN no encontrada") # Eliminar el centro de la estructura de metadata area['centros'].remove(centro) break # Verificar si el área queda vacía y eliminarla si es necesario if not area['centros']: print('Area queda vacia, se eliminara') area_vacia = True region['areas'].remove(area) # Verificar si la región queda vacía y eliminarla si es necesario if not region['areas']: print('Region queda vacia, se eliminara') region_vacia = True metadata['regiones'].remove(region) if not centro_encontrado: return False, 'Centro no encontrado' # Actualizar el documento de metadata update_metadata(db, nombre_salmonera, metadata) # Eliminar datos de sensores para cada bomba y CAN encontrado for bomba_id in bomba_ids: delete_all_sensor_data_for_bomba(db, nombre_salmonera, bomba_id) for can_id in can_ids: delete_all_sensor_data_for_can(db, nombre_salmonera, can_id) for historial_id in historial_control_ids: delete_historial_control(db, nombre_salmonera, historial_id) datos_eliminados = len(bomba_id) + len(can_id) + len(historial_id) msg = f'Se han eliminado {datos_eliminados} datos.' if area_vacia: msg += f' El area {area_nombre} quedo vacia y fue eliminada.' if region_vacia: msg += f' La region {region_nombre} quedo vacia y fue eliminada.' return True, msg except Exception as e: msg = f'Error eliminando datos: {e}' return False, msg # Eliminar datos de sensores para una bomba especfica def delete_all_sensor_data_for_bomba(db, nombre_salmonera, bomba_id): """ Elimina todos los datos de sensores para una bomba especifica, sin limitacion de fecha. Parametros: - db: La instancia de la base de datos de MongoDB. - nombre_salmonera: El nombre de la coleccion de sensores a acceder, que se asume es unica por salmonera. - bomba_id: El identificador unico de la bomba, siguiendo el formato 'region-area-centro-aviot-modulo-nombre'. """ # Construir la consulta para identificar los documentos a eliminar basada únicamente en el bomba_id query = { "bomba_id": bomba_id } # Ejecutar la operación de eliminación en la colección correspondiente db[f'{nombre_salmonera}_sensores'].delete_many(query) # Eliminar datos de sensores para una bomba especfica def delete_all_sensor_data_for_can(db, nombre_salmonera, can_id): """ Elimina todos los datos de sensores para una bomba especifica, sin limitacion de fecha. Parametros: - db: La instancia de la base de datos de MongoDB. - nombre_salmonera: El nombre de la coleccion de sensores a acceder, que se asume es unica por salmonera. - can_id: El identificador unico de la can, siguiendo el formato 'region-area-centro-aviot-modulo-nombre'. """ # Construir la consulta para identificar los documentos a eliminar basada únicamente en el bomba_id query = { "can_id": can_id } # Ejecutar la operación de eliminación en la colección correspondiente db[f'{nombre_salmonera}_sensores'].delete_many(query) # Eliminar datos de sensores para una bomba especfica en un rango de fechas def delete_sensor_data(db, nombre_salmonera, bomba_id, start_date, end_date): """ Elimina los datos de sensores para una bomba especifica dentro de un rango de fechas. Parametros: - db: La instancia de la base de datos de MongoDB. - nombre_salmonera: El nombre de la coleccion de sensores a acceder, que se asume es unica por salmonera. - bomba_id: El identificador unico de la bomba, siguiendo el formato 'region-area-centro-aviot-modulo-nombre'. - start_date: La fecha de inicio del rango de fechas para eliminar datos, como objeto datetime. - end_date: La fecha de fin del rango de fechas para eliminar datos, como objeto datetime. """ # Asegurar que start_date y end_date son objetos datetime de Python # Esto es crucial si las fechas se pasan como strings u otro formato if isinstance(start_date, str): start_date = datetime.strptime(start_date, "%Y-%m-%d") if isinstance(end_date, str): end_date = datetime.strptime(end_date, "%Y-%m-%d") # Construir la consulta para identificar los documentos a eliminar query = { "bomba_id": bomba_id, "fecha": {"$gte": start_date, "$lte": end_date} } # Ejecutar la operación de eliminación en la colección correspondiente db[f'{nombre_salmonera}_sensores'].delete_many(query) # Eliminar historial de control def get_all_historial_control_ids(db, nombre_salmonera): # Obtener metadata del cliente metadata = get_metadata(db, nombre_salmonera) # Definir array de IDs historial_control_ids = [] # Recorrer todos las regiones, areas, centros y aviots para encontrar las IDs for regiones in metadata['regiones']: for areas in regiones['areas']: for centros in areas['centros']: for aviots in centros['aviots']: try: historial_control_ids.append(aviots['historial_control']) except Exception as e: print(f'Error: {e}') # Eliminar los documentos de historiales de control for historial_control_id in historial_control_ids: delete_historial_control(db, nombre_salmonera, historial_control_id) def delete_historial_control(db, nombre_salmonera, historial_control_id): # Crear un query basado en el aviot_id query = { "aviot_id": historial_control_id } try: # Llamada para eliminar el documento result = db[f'{nombre_salmonera}_sensores'].delete_one(query) # Verificar si se eliminó algún documento if result.deleted_count > 0: print(f"Documento con aviot_id {historial_control_id} eliminado.") return True else: print(f"No se encontró el documento con aviot_id {historial_control_id}.") return False except Exception as e: print(f'Error eliminando historial de control en linea {e.__traceback__.tb_lineno}: \n{e}') return False # Generacion de reportes #funcion para obtener los id de historiales (aviot) de control registrados #los centros de todas las areas de todas las regiones de un cliente def get_aviots_id(db): metadata = get_metadata( db) regiones = metadata.get("regiones") ids_aviot = [] for region_data in regiones: nom_region = region_data.get("nombre") areas = region_data.get("areas") for area_data in areas: nom_area = area_data.get("nombre") centros = area_data.get("centros") for centro_data in centros: nom_centro = centro_data.get("nombre") aviots = centro_data.get("aviots") for aviot_data in aviots: id_historial = aviot_data.get("historial_control") ids_aviot.append(id_historial) return ids_aviot """ open docs docs: Filtra todos los documentos asociados a sensores que contengan una fecha en especifico @param db: (pymong_obj) base de datos de nanoox @param fecha: (str) "2024-01-17" @return documentos: (list) lista de documentos filtrados """ def get_documentos_fecha(collection,fecha, centro_id): # filtro = {"fecha": fecha} filtro = { "fecha": fecha, "bomba_id": { "$regex": re.escape(centro_id) #"^" + } } documentos = collection.find(filtro) filtered_documents = [] for documento in documentos: try: bomba_id = documento.get('bomba_id') fecha_doc = documento.get('fecha') sensores = documento.get('sensores') # Revisar que se obtuvieron todos los datos if not all([bomba_id, fecha_doc, sensores]): pass else: filtered_documents.append(documento) except Exception as e: print("Ocurrió un error:", e.__traceback__.tb_lineno, e) pass return filtered_documents def get_documentos_historial(collection, aviot_id): filtro = {"aviot_id": aviot_id} documento = collection.find(filtro) return list(documento) """ open docs docs: Genera un diccionario @param db: lista de documentos de base de datos mongo @return documentos: (dict) con llaves-> nombre bomba ("VA1") y valor: lista con datos sensores """ def get_sensores_bombas(documentos_fecha): # KEYS_INTERES_AUX =["hora", "O2", "SO2", "Temp", "RLY", "Start"] sensores_bombas = dict() for documento in documentos_fecha: bomb_data = documento.get('sensores') alias_bomba = bomb_data[0].get('alias') if alias_bomba not in sensores_bombas: sensores_bomba = documento.get('sensores', []) filtered_sensores_bomba = [] for datos_sensores in sensores_bomba: if 'TEMP' in datos_sensores: # print("Entramos al IF de TEMP") filtered_data = {key: datos_sensores[key] for key in KEYS_INTERES} else: # print("Entramos al ELSE de Temp") filtered_data = {key: datos_sensores[key] for key in KEYS_INTERES_AUX} filtered_sensores_bomba.append(filtered_data) sensores_bombas[alias_bomba] = filtered_sensores_bomba return sensores_bombas # gen_reporte_data # # def _gen_data_report(db, rango_fechas, path) # def gen_reporte_data(db, rango_fechas, save_path, file_name): # CONTROL = 'VDF' # try: # print("----------ENTRAMOS Al try _gen_data_report --------") # print(f"generando reporte de tipo {CONTROL}") # print(rango_fechas) # sensores_collection = db[f'sensores'] # # get_db()[f'sensores'] # # Supongamos que 'msg' es un diccionario que contiene las claves 'FechaInicio' y 'FechaFin' # # Convertir las fechas de cadena a objetos datetime # start = datetime.strptime(rango_fechas['startDate'], "%d/%m/%Y") # end = datetime.strptime(rango_fechas['endDate'], "%d/%m/%Y") # #-- Calculo de cantidad de dias a analizar --# # if len( str(end-start).split(" ") ) > 1: # diasEntre = int( str(end-start).split(" ")[0] ) + 1 # para que tome en cuenta ultimo dia # print("Se han seleccionado ",diasEntre," dias") # else: # diasEntre = 1 # #* -- Crear libro nuevo en Excel-- # # # workbook = openpyxl.Workbook() # # ws_names = [] # #-- Analisis por dia entre fechas del rango --# # print(f"{diasEntre=}") # for i in range(diasEntre): # # print("Comenzando a generar reporte...") # dia = str(start + timedelta(days=i)).split(" ")[0].split("-") # fecha_format = "-".join(dia) # #lista documentos donde se encuentra la fecha # documentos_fecha = get_documentos_fecha(sensores_collection, fecha_format) # print(f"{documentos_fecha=}") # if documentos_fecha: # #* -- Crear libro nuevo en Excel-- # # workbook = openpyxl.Workbook() # ws_names = [] # #dicc con llave-> nombre bomba ("VA1") y valor: lista con datos sensores #diccionario para almacenar informacion del documento # datos_sensores_bombas = get_sensores_bombas(documentos_fecha) # # Nombres de cámaras # bomb_names = list(datos_sensores_bombas.keys()) # rangos_iny = {alias_bomba: [] for alias_bomba in bomb_names } # rangos_vdf = {alias_bomba: [] for alias_bomba in bomb_names } # #cantidad de elementos por columna # max_column = len(datos_sensores_bombas[bomb_names[0]]) # #Creacion de pagina especifica para la fecha analizada # worksheet = workbook.create_sheet(fecha_format) # ws_names.append(fecha_format) # #* ----- especificaciones para paginas en el excel ---- * # v_offset = 18 #espacio vertical desde el inicio de la pagina # h_offset = 2 #espacio horizontal desde el inicio de la pagina # tablas_dist = 6 #espacio entre tablas # tablas_space = [""]*(tablas_dist - 4) # cant_keys_tabla = len(KEYS_INTERES) -2 #keys interes - auxiliares para definicino de color celda # # print(f"{KEYS_INTERES=}") # space_titles = cant_keys_tabla - h_offset - 1 # #* ------------------------------------------------------- # #* -------------------------------------------- Agregar filas --------------------------------------- # # for i in range(v_offset): # worksheet.append([]) # Dejar celdas vacias al inicio # #* -- Primera Fila : Nombres de las estanques -- # # fila = [] # for alias_bomba in bomb_names: #en formato fila para facilitar visualizacion # fila+= [""]*h_offset ; fila.append(f"Bomba {alias_bomba}") ; fila+= [""]*space_titles # fila += tablas_space # worksheet.append(fila) # #* -- Segunda Fila : información de las jaulas -- # # estructura = ([""]*h_offset + ["Hora" ,"Oxígeno mg/L", "Saturación %" , 'Temperatura °C'] )*len(bomb_names) # worksheet.append( estructura ) # #* -- Filas siguientes: -- # # #* Agrega iterativamente los datos directamente desde la base de datos # Rangos = {} # # Rango de celdas de cada canal # for idx,channel in enumerate(bomb_names): # cell_inicial_Letra = get_column_letter( idx*tablas_dist + h_offset +1 ) # El ultimo +1 es por desplazamiento lateral de las tablas # cell_final_Letra = get_column_letter( idx*tablas_dist + h_offset + cant_keys_tabla ) # cell_graph = get_column_letter( idx*tablas_dist + h_offset +1) # Rangos[channel] = { # 'L_Inicial':cell_inicial_Letra, # 'L_Final': cell_final_Letra, # 'L_Graph': cell_graph, # 'celdaInicial': str(cell_inicial_Letra + str(2 + v_offset)), # 'celdaFinal': str(cell_final_Letra + str( len(datos_sensores_bombas[channel]) + 2 + v_offset ) ), # } # values_for_charts = {} # for i in range(max_column): # nueva_fila = [] # # Para cada dato -> recorrer todos los canales # for idx,channel in enumerate(bomb_names): # if len(datos_sensores_bombas[channel]) > i: # data_fila = [datos_sensores_bombas[channel][i][key] for key in KEYS_INTERES] # # data_fila[0] = datetime.strptime( data_fila[0], '%H:%M:%S') # # Convertir horas a formato datetime si es necesario # if data_fila[2] == -1: # data_fila[1] = "-" # data_fila[2] = "-" # data_fila[3] = '-' # #busca instantes de inyeccion # if data_fila[-2] == True: # rangos_iny[channel].append(data_fila[0]) # # print(f"{data_fila=}") # if CONTROL == "VDF": # if data_fila[-1] == "01": # # print(f"{data_fila=}") # rangos_vdf[channel].append(data_fila[0]) # else: # if data_fila[-1] == "00aa": # # print(f"{data_fila=}") # rangos_vdf[channel].append(data_fila[0]) # # print(f"{data_fila=}") # nueva_fila.append( '' ) # nueva_fila.append( '' ) # nueva_fila+=data_fila[:-2] # nueva_fila += tablas_space[0:-h_offset] # if channel not in values_for_charts: # values_for_charts[channel] = [] # else: # values_for_charts[channel].append(data_fila) # # # Agregar a la planilla la fila con todas las columnas # worksheet.append( nueva_fila ) # #* -------------------------------------------- Graficos --------------------------------------- # # from io import BytesIO # matplotlib.use('Agg') # Backend sin interfaz gráfica # import matplotlib.pyplot as plt # import matplotlib.dates as mdates # from openpyxl.drawing.image import Image as ExcelImg # plt.rcParams.update({ # 'font.size': 20, # Tamaño general del texto # 'axes.titlesize': 20, # Tamaño del título del gráfico # 'axes.labelsize': 30, # Tamaño de etiquetas de los ejes # 'xtick.labelsize': 20, # Tamaño de las etiquetas del eje X # 'ytick.labelsize': 20, # Tamaño de las etiquetas del eje Y # 'legend.fontsize': 15, # Tamaño de la leyenda # 'figure.titlesize': 25 # Tamaño del título de la figura # }) # # Recorremos los datos por canal # for idx, channel_data in enumerate(values_for_charts): # print(f"{channel_data} {len(values_for_charts[channel_data])}") # # Extraer datos # horas_str = [fila[0] for fila in values_for_charts[channel_data]] # y_values = [fila[1] for fila in values_for_charts[channel_data]] # Nivel de O2 # saturacion_values = [fila[2] for fila in values_for_charts[channel_data]] # Saturación % # # Verificar datos # print(f"Cantidad de datos: {len(horas_str)}") # print("Primeros 5 valores de horas:", horas_str[:5]) # # Convertir a datetime (formato HH:MM:SS) # try: # horas = [datetime.strptime(h, "%H:%M:%S") for h in horas_str] # except ValueError as e: # print("Error en conversión de tiempo:", e) # # Convertir a formato numérico para matplotlib # x = mdates.date2num(horas) # # Dimensiones del gráfico # ancho_grafico = 10 # Anchura en pulgadas # alto_grafico = 7.3 # Altura en pulgadas # # Crear gráfico con eje principal # fig, ax1 = plt.subplots(figsize=(ancho_grafico, alto_grafico), dpi=100) # # Graficar Nivel de O2 en eje principal (izquierdo) # ax1.plot(x, y_values, label="Nivel de O2 (mg/L)", color="#1e4882", linewidth=2) # ax1.set_xlim(min(x), max(x)) # ax1.set_xlabel("Hora del día", fontsize=12) # ax1.set_ylabel("Nivel de O2 (mg/L)", color="#1e4882", fontsize=12) # ax1.set_ylim(0,12) # ax1.tick_params(axis='y', labelcolor="#1e4882") # # Crear segundo eje Y para saturación de oxígeno # ax2 = ax1.twinx() # ax2.set_xlim(min(x), max(x)) # ax2.plot(x, saturacion_values, label="Saturación %", color="#ab0202", linestyle="--", linewidth=2) # ax2.set_ylim(0,100) # ax2.set_ylabel("Saturación %", color="#ab0202", fontsize=12) # ax2.tick_params(axis='y', labelcolor="#ab0202") # # Configuración del eje X en escala de horas # interval = 60 # ax1.xaxis.set_major_locator(mdates.MinuteLocator(interval=interval)) # ax1.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M')) # # Rotar etiquetas del eje X # fig.autofmt_xdate() # # Título del gráfico # ax1.set_title(f"Sensor {channel_data}", fontsize=14) # # Leyendas combinadas # lines_1, labels_1 = ax1.get_legend_handles_labels() # lines_2, labels_2 = ax2.get_legend_handles_labels() # ax1.legend(lines_1 + lines_2, labels_1 + labels_2, loc='lower center', bbox_to_anchor=(0.5, -0.25), ncol=2) # # Ajustar diseño # plt.tight_layout(pad=0.5) # plt.grid(True) # # Guardar el gráfico en un buffer en memoria # img_data = BytesIO() # plt.savefig(img_data, format='png', dpi=100, bbox_inches='tight', pad_inches=0.1) # plt.close(fig) # # Insertar la imagen en la hoja de Excel # img_data.seek(0) # img = ExcelImg(img_data) # # Configurar tamaño de imagen # img.width = 400 # img.height = 292 # celda = Rangos[channel_data]["L_Graph"] + "2" # worksheet.add_image(img, celda) # # # Crear un gráfico de líneas # # chart = LineChart() # # chart.width = 10.3 # # chart.height = 7.3 # # chart.title = "Medición sensor O2" # # chart.style = 13 # # chart.x_axis.title = 'Hora' # # chart.y_axis.title = 'mg/L' # # letter_graph = Rangos[channel]["L_Graph"] + "2" # # categories = Reference(worksheet, # # min_col=idx*tablas_dist + h_offset + 1, # # max_col=idx*tablas_dist + h_offset + 1, # # min_row=15, # # max_row=max_column) # # data = Reference(worksheet, # # min_col=idx*tablas_dist + h_offset + 2, # # max_col=idx*tablas_dist + h_offset + 2, # # min_row=15, # # max_row=max_column) # # chart.add_data(data) # # chart.set_categories(categories) # # chart.legend = None # # chart.series[0].graphicalProperties.line.solidFill = "4a7ebb" # # chart.series[0].graphicalProperties.line.width = 20000 # # chart.x_axis.tickLblPos = 'low' # # chart.x_axis.delete = False # # chart.x_axis.label_rotation = 90 # # chart.x_axis.majorUnit = 15 # # # Configuración de los títulos de los ejes con separación # # chart.x_axis.title = "Hora" # # chart.x_axis.title.offset = 5000 # Mayor separación para el título del eje X # # chart.y_axis.title = "mg/L" # # chart.y_axis.title.offset = 5000 # Mayor separación para el título del eje Y # # # Configurar etiquetas del eje X y eje Y # # chart.x_axis.tickLblPos = 'low' # Coloca las etiquetas X más abajo # # chart.x_axis.label_rotation = 90 # Rotación de las etiquetas X para evitar sobreposición # # chart.y_axis.tickLblPos = 'nextTo' # Asegura que las etiquetas Y queden correctamente posicionadas # # # Configurar la escala del eje Y # # chart.y_axis.scaling.min = 0 # # chart.y_axis.scaling.max = 10 # # chart.y_axis.majorUnit = 1 # Etiquetas cada 1 unidad # # chart.y_axis.minorUnit = 0.5 # Marcas menores cada 0.5 # # chart.y_axis.delete = False # Asegura que el eje Y no se elimine # # worksheet.add_chart(chart, letter_graph) # #* -------------------------------------------- Estilos --------------------------------------- # # # Marca de Agua # # Obtener la ruta absoluta del archivo de imagen # try: # # img_path = os.path.join(os.path.dirname(__file__), 'Logo_Reportes.png') # Ruta relativa al script # base_dir = os.path.dirname(os.path.abspath(__file__)) # Directorio actual del script (database) # img_path = os.path.join(base_dir, "images", "Logo_Reportes.png") # Subir un nivel y entrar a 'images' #base_dir, "..", "images", "Logo_Reportes.png" # if not os.path.exists(img_path): # raise FileNotFoundError(f"No se encuentra la imagen en: {img_path}") # add_markwater_images(worksheet, img_path, start_column="C", start_row=21, repeats=2, step=7) # """ # # Cargar la imagen # img = ExcelImg(img_path) # # img.anchor = 'C21' # Posición inicial de la imagen en la hoja # # start_position = 'C21' # start_column = 'C' # start_row = 21 # # Ajustar el tamaño de la imagen # img.width = 600 # Ajusta el ancho (en píxeles) # img.height = 450 # Ajusta la altura (en píxeles) # # Agregar la imagen al libro # # worksheet.add_image(img, start_position) # # worksheet.add_image(img, start_column + str(start_row)) # # print("Marca de agua inicial agregada correctamente.") # # Repetir la imagen horizontalmente tres veces más # for i in range(0, 4): # Repetir tres veces (i=1, 2, 3) # img_copy = ExcelImg(img_path) # img_copy.width = 600 # img_copy.height = 450 # # Calcular nueva posición horizontal con separación (por ejemplo, 2 columnas) # new_column = chr(ord(start_column) + i * 7) # Ajusta el valor para la columna # new_position = new_column + str(start_row) # # new_column = chr(ord(start_position[0]) + i * 10) # Ajusta el valor para la columna # # new_position = new_column + start_position[1:] # worksheet.add_image(img_copy, new_position) # print("Marca de agua agregada correctamente.") # """ # except Exception as e: # print(f"Error al agregar la marca de agua: {e}") # font = Font(b=True, color="FFFFFF")# color="00FFFFFF" -> blanco # fill = PatternFill(fill_type="solid", # start_color='e84e0e',#'204D84', # end_color='e84e0e')#'204D84') # fill_iny = PatternFill(fill_type="solid", # start_color='FFFF00',#'204D84', # end_color='FFFF00')#'204D84') # fill_vdf = PatternFill(fill_type="solid", # start_color='C8C8C8',#'204D84', # end_color='C8C8C8')#'204D84') # # Estilo de bordes: # border = Border(left=Side(border_style="thin", color="000000"), # right=Side(border_style="thin", color="000000"), # top=Side(border_style="thin", color="000000"), # bottom=Side(border_style="thin", color="000000")) # for idx,channel in enumerate(bomb_names): # Letra1 = get_column_letter( idx*tablas_dist + h_offset + 1 ) # El +1 es por desplazamiento lateral de las tablas # Letra2 = get_column_letter( idx*tablas_dist + h_offset + 2 ) # Letra3 = get_column_letter( idx*tablas_dist + h_offset + 3 ) # Letra4 = get_column_letter( idx*tablas_dist + h_offset + 4 ) # #-- Diseño celdas -- # # cell_merge = worksheet[Letra1 + str(v_offset +1)] # cell_merge.font = font # cell_merge.fill = fill # # Definicion celdas # cell_1 = worksheet[Letra1 + str(v_offset + 2)] # cell_2 = worksheet[Letra2 + str( v_offset +2 )] # cell_3 = worksheet[Letra3 + str( v_offset +2 )] # cell_4 = worksheet[Letra4 + str( v_offset +2 )] # # Font: # cell_1.font = font # cell_2.font = font # cell_3.font = font # cell_4.font = font # # Fill: # cell_1.fill = fill # cell_2.fill = fill # cell_3.fill = fill # cell_4.fill = fill # #-- Combinar cheldas de titulos --# # for column in range( len(bomb_names) ): # worksheet.merge_cells(start_row=v_offset + 1, # start_column=column*tablas_dist + h_offset +1, # end_row=v_offset +1, # end_column=column*tablas_dist + h_offset +cant_keys_tabla # ) # fila 2 ya que se deja una vacía al inicio # #-- Aplicar el borde a un conjunto de celdas --# # for channel in bomb_names: # cell_range = worksheet[Rangos[channel]['celdaInicial'] : Rangos[channel]['celdaFinal'] ] # for row in cell_range: # for cell in row: # cell.border = border # if row[0].value in rangos_iny[channel] and row[0].value in rangos_vdf[channel]: # cell.fill = fill_iny # elif row[0].value in rangos_vdf[channel]: # cell.fill = fill_vdf # #-- alineacion de celdas -- # for row in worksheet.iter_rows(min_row=4, max_row= max_column + v_offset + 2): # for cell in row: # cell.alignment = Alignment(horizontal='center', # vertical='center', # shrink_to_fit=True, # wrap_text=False) # # Ajustar el ancho de las columnas solo para aquellas que contienen datos # for column in worksheet.columns: # max_length = 0 # for cell in column: # try: # if cell.value is not None and len(str(cell.value)) > max_length: # max_length = len(cell.value) # except: # pass # adjusted_width = (max_length + 2) # if adjusted_width > 2: # Ajustar solo si hay datos en la columna # worksheet.column_dimensions[column[0].column_letter].width = adjusted_width # if 'Sheet' in workbook.sheetnames: # workbook.remove(workbook['Sheet']) # try: # # Detectar la ruta del escritorio # path_escritorio = save_path # # path_escritorio = get_escritorio_path() # print(f"\n-------Ruta detectada del escritorio: {path_escritorio}-------- \n") # print(f"{file_name=}") # # Crear la ruta completa para el archivo Excel # # dest_path = os.path.join(path_escritorio, "Reporte.xlsx") # dest_path = os.path.join(path_escritorio, file_name) # # dest_path = str(path_escritorio + file_name) # print(f"{dest_path=}") # # Verificar que la ruta exista # if not os.path.exists(path_escritorio): #path_escritorio # raise FileNotFoundError(f"El escritorio no existe o no es accesible: {path_escritorio}") # print(f"{workbook.sheetnames=}") # if workbook.sheetnames == []: # print("No hay hojas creadas") # return False # else: # # Guardar el archivo Excel # workbook.save(dest_path) # if hasattr(workbook, 'close'): # workbook.close() # print(f"Archivo Excel guardado correctamente en: {dest_path}") # return True # except Exception as e: # print(f"Error al guardar el archivo: {e}") # print(e, e.__traceback__.tb_lineno) # import traceback # traceback.print_exc() # elif documentos_fecha == []: # print("No hay datos") # return False # # return True # except Exception as e: # print(f"Error: {e} {e.__traceback__.tb_lineno}") # return None # def get_documentos_fecha(sensores_collection, fecha_format): # """ # Dada una colección y una fecha en formato 'YYYY-MM-DD', # retorna la lista de documentos que coinciden con esa fecha. # """ # # Ejemplo mínimo (ajusta a tu Query real): # return list(sensores_collection.find({"fecha": fecha_format})) # def get_sensores_bombas(documentos_fecha): # """ # Dada la lista de documentos (documentos_fecha), # retorna un diccionario: # { 'VA1': [ {...}, {...}, ... ], # 'VA2': [ {...}, {...}, ... ], # ... # } # con los datos organizados por bomba. # """ # # Ejemplo básico: # from collections import defaultdict # datos_por_bomba = defaultdict(list) # for doc in documentos_fecha: # nombre_bomba = doc.get("bomba", "VAx") # datos_por_bomba[nombre_bomba].append(doc) # return dict(datos_por_bomba) # def add_markwater_images(worksheet, img_path, start_column="C", start_row=21, repeats=2, step=7): # """ # Inserta una marca de agua repetida en la hoja 'worksheet'. # Puedes personalizar repeticiones, separaciones, etc. # """ # try: # # Cargar la imagen # img_base = ExcelImg(img_path) # img_base.width = 600 # img_base.height = 450 # # Repetir la imagen en horizontal varias veces # for i in range(repeats): # new_img = ExcelImg(img_path) # new_img.width = img_base.width # new_img.height = img_base.height # new_col = chr(ord(start_column) + i * step) # position = f"{new_col}{start_row}" # worksheet.add_image(new_img, position) # print("Marca de agua agregada correctamente.") # except Exception as e: # print(f"Error al agregar la marca de agua: {e}") # ----------------------------------------------------------------- # Función principal de generación de reporte # ----------------------------------------------------------------- #original # def _gen_reporte_data(db, rango_fechas, save_path, file_name, control): #, control # """ # Genera un reporte en Excel dentro del rango de fechas (rango_fechas['startDate'], rango_fechas['endDate']). # Se apoya en la colección 'sensores' dentro de la BD 'db'. # Retorna: # - True => si al menos un día tuvo datos y se generó el archivo exitosamente. # - False => si no hubo datos en ningún día (no se genera archivo). # - None => si ocurre un error no controlado (excepción). # Ajusta variables según tu escenario (claves, flags, ruta de imagen, etc.). # """ # import os # import openpyxl # from openpyxl import Workbook # from openpyxl.utils import get_column_letter # from openpyxl.styles import PatternFill, Font, Border, Side, Alignment # from openpyxl.drawing.image import Image as ExcelImg # import matplotlib # matplotlib.use('Agg') # Backend sin interfaz gráfica # import matplotlib.pyplot as plt # import matplotlib.dates as mdates # from datetime import datetime, timedelta # from io import BytesIO # CONTROL = control #'VDF' #control # Puedes cambiarlo a "PS" u otro, según sea tu caso # print(f"REPORTE {CONTROL=}") # # Ajusta a las claves de tu doc: ej. "hora", "oxigeno_mg", etc. # # KEYS_INTERES = ["hora", "oxigeno_mg", "saturacion", "temperatura", "iny_bool", "control_flag"] # try: # print("---------- ENTRAMOS a gen_reporte_data ----------") # print("DEBUG: Rango_fechas recibido =", rango_fechas) # # Colección de sensores # sensores_collection = db.sensores # Ajustar nombre real # # 1) Convertir strings a datetime # start = datetime.strptime(rango_fechas['startDate'], "%d/%m/%Y") # end = datetime.strptime(rango_fechas['endDate'], "%d/%m/%Y") # # 2) Calcular días a analizar (inclusive) # diasEntre = (end - start).days + 1 # if diasEntre < 1: # print("DEBUG: Rango de fechas inválido (end < start).") # return False # print(f"DEBUG: Se han seleccionado {diasEntre} día(s)") # # # 3) Crear UN solo Workbook fuera del bucle # # workbook = openpyxl.Workbook() # # # Quitar la hoja 'Sheet' por defecto (opcional) # # if 'Sheet' in workbook.sheetnames: # # workbook.remove(workbook['Sheet']) # hay_hoja_con_datos = False # # 4) Recorrer cada día del rango # for i in range(diasEntre): # fecha_actual = start + timedelta(days=i) # fecha_format = fecha_actual.strftime("%Y-%m-%d") # Ajusta si tu DB usa otro formato # documentos_fecha = get_documentos_fecha(sensores_collection, fecha_format) # print(f"{documentos_fecha=}") # print(f"DEBUG: Fecha '{fecha_format}': documentos hallados = {len(documentos_fecha)}") # if documentos_fecha and len(documentos_fecha) > 0: # # Si sí hay datos # hay_hoja_con_datos = True # workbook = openpyxl.Workbook() # if 'Sheet' in workbook.sheetnames: # workbook.remove(workbook['Sheet']) # print(f"DEBUG: Creando hoja de Excel para {fecha_format}...") # worksheet = workbook.create_sheet(fecha_format) # datos_sensores_bombas = get_sensores_bombas(documentos_fecha) # bomb_names = list(datos_sensores_bombas.keys()) # # Diccionarios para marcar inyecciones (iny_bool) o VDF (control_flag) # rangos_iny = {b: [] for b in bomb_names} # rangos_vdf = {b: [] for b in bomb_names} # # Tomar la bomba que tenga más registros para saber cuántas filas # max_column = len(datos_sensores_bombas[bomb_names[0]]) # # Parámetros visuales # v_offset = 18 # h_offset = 2 # tablas_dist = 6 # tablas_space = [""] * (tablas_dist - 4) # cant_keys_tabla = len(KEYS_INTERES) - 2 # -2 => iny_bool y control_flag no se muestran en columnas # space_titles = cant_keys_tabla - h_offset - 1 # # Dejar filas vacías al inicio # for _ in range(v_offset): # worksheet.append([]) # # -------------- Fila 1 (nombres de bombas) -------------- # fila1 = [] # for alias_bomba in bomb_names: # fila1 += [""] * h_offset # fila1.append(f"Bomba {alias_bomba}") # fila1 += [""] * space_titles # fila1 += tablas_space # worksheet.append(fila1) # # -------------- Fila 2 (cabeceras) -------------- # fila2 = [] # for _ in bomb_names: # fila2 += ([""] * h_offset + ["Hora", "Oxígeno mg/L", "Saturación %", "Temperatura °C"]) # fila2 += tablas_space[0:-h_offset] # worksheet.append(fila2) # # -------------- Rangos para bordes y gráficos -------------- # Rangos = {} # for idx, channel in enumerate(bomb_names): # c_inicial = get_column_letter(idx * tablas_dist + h_offset + 1) # c_final = get_column_letter(idx * tablas_dist + h_offset + cant_keys_tabla) # c_graph = get_column_letter(idx * tablas_dist + h_offset + 1) # Columna donde poner la imagen # Rangos[channel] = { # 'L_Inicial': c_inicial, # 'L_Final': c_final, # 'L_Graph': c_graph, # 'celdaInicial': f"{c_inicial}{2 + v_offset}", # 'celdaFinal': f"{c_final}{(2 + v_offset + max_column)}", # } # # -------------- Llenar filas de datos -------------- # values_for_charts = {} # for fila_idx in range(max_column): # nueva_fila = [] # for idx, channel in enumerate(bomb_names): # datos_bomba = datos_sensores_bombas[channel] # if len(datos_bomba) > fila_idx: # # Extrae cada campo según KEYS_INTERES # data_fila = [datos_bomba[fila_idx].get(k) for k in KEYS_INTERES] # # Ejemplo: data_fila = [hora, oxigeno_mg, saturacion, temperatura, iny_bool, control_flag] # # -1 => sin dato # if data_fila[2] == -1: # data_fila[1] = "-" # data_fila[2] = "-" # data_fila[3] = "-" # # Recolectar instantes de inyección # if data_fila[-2] is True: # rangos_iny[channel].append(data_fila[0]) # # Recolectar instantes de VDF # if CONTROL == "VDF": # if data_fila[-1] == "01": # rangos_vdf[channel].append(data_fila[0]) # else: # if data_fila[-1] == "00aa": # rangos_vdf[channel].append(data_fila[0]) # # Añadir celdas vacías + [hora, O2, saturación, temp] # nueva_fila.append('') # nueva_fila.append('') # nueva_fila += data_fila[:-2] # omite iny_bool y control_flag # nueva_fila += tablas_space[0:-h_offset] # # Para gráficos # if channel not in values_for_charts: # values_for_charts[channel] = [] # values_for_charts[channel].append(data_fila) # worksheet.append(nueva_fila) # # -------------- Generar gráficos e insertarlos como imagen -------------- # for idx, channel_data in enumerate(values_for_charts): # datos_canal = values_for_charts[channel_data] # if not datos_canal: # continue # # horas_str = [fila[0] for fila in datos_canal] # horas_str = [] # saturacion_values = [] # # y_values = [fila[1] for fila in datos_canal] # O2 mg # y_values = [] # for fila in datos_canal: # if not isinstance(fila[1], float): # # print("/*/*/*/*/*/*/*/*/*/*/-/-*/-*/-*/-*/-*/-*/-//-*/*-/-*/-*/-*") # # print(f"{fila[1]=}, {fila[0]=}") # continue # y_values.append(fila[1]) # horas_str.append(fila[0]) # saturacion_values.append(fila[2]) # # saturacion_values = [fila[2] for fila in datos_canal] # Saturación % # try: # horas = [datetime.strptime(h, "%H:%M:%S") for h in horas_str if isinstance(h, str)] # except ValueError as e: # print("DEBUG: Error en conversión de tiempo:", e) # horas = [] # if not horas: # continue # x = mdates.date2num(horas) # fig, ax1 = plt.subplots(figsize=(10, 7.3), dpi=100) # ax1.plot(x, y_values, label="Nivel O2 (mg/L)", color="#1e4882", linewidth=2) # ax1.set_xlim(min(x), max(x)) # ax1.set_ylim(0, 12) # ax1.set_xlabel("Hora del día", fontsize=12) # ax1.set_ylabel("Nivel de O2 (mg/L)", color="#1e4882", fontsize=12) # ax1.tick_params(axis='y', labelcolor="#1e4882") # ax2 = ax1.twinx() # ax2.plot(x, saturacion_values, label="Saturación %", color="#ab0202", linestyle="--", linewidth=2) # # ax2.set_ylim(0, 100) # ax2.set_ylim(0, max(saturacion_values)) # ax2.set_ylabel("Saturación %", color="#ab0202") # ax2.tick_params(axis='y', labelcolor="#ab0202") # ax1.set_title(f"Sensor {channel_data}") # ax1.xaxis.set_major_locator(mdates.MinuteLocator(interval=60)) # ax1.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M')) # fig.autofmt_xdate() # # Leyenda unificada # lines_1, labels_1 = ax1.get_legend_handles_labels() # lines_2, labels_2 = ax2.get_legend_handles_labels() # ax1.legend(lines_1 + lines_2, labels_1 + labels_2, # loc='lower center', bbox_to_anchor=(0.5, -0.25), ncol=2) # plt.grid(True) # plt.tight_layout() # img_data = BytesIO() # plt.savefig(img_data, format='png', dpi=100, bbox_inches='tight', pad_inches=0.1) # plt.close(fig) # img_data.seek(0) # img = ExcelImg(img_data) # img.width = 400 # img.height = 292 # celda = Rangos[channel_data]["L_Graph"] + "2" # worksheet.add_image(img, celda) # # -------------- Marca de agua -------------- # try: # base_dir = os.path.dirname(os.path.abspath(__file__)) # img_path = os.path.join(base_dir, "images", "Logo_Reportes.png") # add_markwater_images(worksheet, img_path, # start_column="C", start_row=21, # repeats=2, step=7) # except Exception as e: # print(f"DEBUG: Error al agregar la marca de agua: {e}") # # -------------- Estilos (fuentes, relleno, bordes...) -------------- # font_blanco = Font(b=True, color="FFFFFF") # fill_naranja = PatternFill(fill_type="solid", start_color='e84e0e', end_color='e84e0e') # fill_iny = PatternFill(fill_type="solid", start_color='FFFF00', end_color='FFFF00') # fill_vdf = PatternFill(fill_type="solid", start_color='C8C8C8', end_color='C8C8C8') # border_thin = Border( # left=Side(border_style="thin", color="000000"), # right=Side(border_style="thin", color="000000"), # top=Side(border_style="thin", color="000000"), # bottom=Side(border_style="thin", color="000000") # ) # # Cabecera (fila 1 y fila 2) # for idx, channel in enumerate(bomb_names): # let1 = get_column_letter(idx * tablas_dist + h_offset + 1) # let2 = get_column_letter(idx * tablas_dist + h_offset + 2) # let3 = get_column_letter(idx * tablas_dist + h_offset + 3) # let4 = get_column_letter(idx * tablas_dist + h_offset + 4) # cell_merge = worksheet[f"{let1}{v_offset+1}"] # cell_merge.font = font_blanco # cell_merge.fill = fill_naranja # c1 = worksheet[f"{let1}{v_offset+2}"] # c2 = worksheet[f"{let2}{v_offset+2}"] # c3 = worksheet[f"{let3}{v_offset+2}"] # c4 = worksheet[f"{let4}{v_offset+2}"] # for c in [c1, c2, c3, c4]: # c.font = font_blanco # c.fill = fill_naranja # # Combinar celdas para los títulos # for col_idx in range(len(bomb_names)): # start_col = col_idx * tablas_dist + h_offset + 1 # end_col = col_idx * tablas_dist + h_offset + cant_keys_tabla # worksheet.merge_cells( # start_row=v_offset+1, start_column=start_col, # end_row=v_offset+1, end_column=end_col # ) # # Borde y rellenos por filas # for channel in bomb_names: # rango_celdas = worksheet[Rangos[channel]['celdaInicial'] : Rangos[channel]['celdaFinal']] # for row in rango_celdas: # for cell in row: # cell.border = border_thin # hora_val = row[0].value # if hora_val in rangos_iny[channel] and hora_val in rangos_vdf[channel]: # cell.fill = fill_iny # elif hora_val in rangos_vdf[channel]: # cell.fill = fill_vdf # # Alineación # for row in worksheet.iter_rows(min_row=4, max_row=(max_column + v_offset + 2)): # for cell in row: # cell.alignment = Alignment( # horizontal='center', # vertical='center', # shrink_to_fit=True, # wrap_text=False # ) # # Ajustar ancho de columnas # for col in worksheet.columns: # max_length = 0 # col_letter = col[0].column_letter # for cell in col: # val = cell.value # if val and isinstance(val, str) and len(val) > max_length: # max_length = len(val) # adjusted_width = max_length + 2 # if adjusted_width > 2: # worksheet.column_dimensions[col_letter].width = adjusted_width # worksheet.protection.sheet=True # worksheet.protection.objects=True # worksheet.protection.scenarios=True # worksheet.protection.password=password # else: # # documentos_fecha vacío => no creamos hoja # print(f"DEBUG: No se generará hoja para '{fecha_format}' (sin datos)") # # Fuera del for # print(f"DEBUG: Finalizado el for. hay_hoja_con_datos = {hay_hoja_con_datos}") # # Si no se creó ninguna hoja (ningún día tuvo datos): # if not hay_hoja_con_datos: # print("DEBUG: No hubo datos en todo el rango.") # dest_path = os.path.join(save_path, file_name) # if os.path.exists(dest_path): # os.remove(dest_path) # return False # else: # # 5) Guardar una sola vez si hubo al menos una hoja # dest_path = os.path.join(save_path, file_name) # print(f"DEBUG: A punto de guardar el Workbook en {dest_path}") # if not os.path.exists(save_path): # raise FileNotFoundError(f"No existe la carpeta destino: {save_path}") # workbook.security.workbookPassword=password # workbook.security.lockStructure=True # workbook.save(dest_path) # workbook.close() # print(f"DEBUG: Archivo Excel guardado correctamente en {dest_path}") # return True # except Exception as e: # print(f"DEBUG: Error en gen_reporte_data: {e} (línea {e.__traceback__.tb_lineno})") # return None def gen_reporte_data(db, rango_fechas, save_path, file_name, control, centro_id): """ Genera un reporte en Excel dentro del rango de fechas (rango_fechas['startDate'], rango_fechas['endDate']). Se apoya en la colección 'sensores' dentro de la BD 'db'. Retorna: - True => si al menos un día tuvo datos y se generó el archivo exitosamente. - False => si no hubo datos en ningún día (no se genera archivo). - None => si ocurre un error no controlado (excepción). Ajusta variables según tu escenario (claves, flags, ruta de imagen, etc.). """ import os import openpyxl from openpyxl import Workbook from openpyxl.utils import get_column_letter from openpyxl.styles import PatternFill, Font, Border, Side, Alignment from openpyxl.drawing.image import Image as ExcelImg import matplotlib matplotlib.use('Agg') # Backend sin interfaz gráfica import matplotlib.pyplot as plt import matplotlib.dates as mdates from datetime import datetime, timedelta from io import BytesIO CONTROL = control # "VDF" u otro # Ajusta las claves que quieras mostrar en columnas KEYS_INTERES_INTERNAS = ["hora", "oxigeno_mg", "saturacion", "temperatura", "iny_bool", "control_flag"] print("ENTRAMOS A DEGENERAR REPORTES-*-*-*-") # Se asume que hay funciones externas # get_documentos_fecha(coleccion, fecha_str) # get_sensores_bombas(documentos_fecha) # add_markwater_images(...) # password = '...' # Contraseña para proteger la hoja si lo deseas # password = "1234" # <-- EJEMPLO try: print("---------- ENTRAMOS a gen_reporte_data ----------") print("DEBUG: Rango_fechas recibido =", rango_fechas) # Colección de sensores # sensores_collection = db.sensores # Ajustar nombre real según tu BD sensores_collection = db.sensores # Ajustar nombre real según tu BD # 1) Convertir strings a datetime start = datetime.strptime(rango_fechas['startDate'], "%d/%m/%Y") end = datetime.strptime(rango_fechas['endDate'], "%d/%m/%Y") # 2) Calcular días a analizar (inclusive) diasEntre = (end - start).days + 1 if diasEntre < 1: print("DEBUG: Rango de fechas inválido (end < start).") return False print(f"DEBUG: Se han seleccionado {diasEntre} día(s)") ##### 1) CREAMOS UN SOLO WORKBOOK FUERA DEL FOR workbook = openpyxl.Workbook() # Quitar la hoja 'Sheet' por defecto (opcional) if 'Sheet' in workbook.sheetnames: workbook.remove(workbook['Sheet']) hay_hoja_con_datos = False # 4) Recorrer cada día del rango for i in range(diasEntre): fecha_actual = start + timedelta(days=i) fecha_format = fecha_actual.strftime("%Y-%m-%d") # Ejemplo de función que obtendría los documentos de la base documentos_fecha = get_documentos_fecha(sensores_collection, fecha_format, centro_id) print(f"DEBUG: Fecha '{fecha_format}': documentos hallados = {len(documentos_fecha)}") if documentos_fecha and len(documentos_fecha) > 0: # Si sí hay datos para este día, creamos una hoja dentro del workbook hay_hoja_con_datos = True print(f"DEBUG: Creando hoja de Excel para {fecha_format}...") worksheet = workbook.create_sheet(fecha_format) # Obtener datos estructurados por bomba datos_sensores_bombas = get_sensores_bombas(documentos_fecha) bomb_names = list(datos_sensores_bombas.keys()) # Diccionarios para marcar inyecciones (iny_bool) o VDF (control_flag) rangos_iny = {b: [] for b in bomb_names} rangos_vdf = {b: [] for b in bomb_names} # Tomar la bomba con más registros para saber cuántas filas max_column = len(datos_sensores_bombas[bomb_names[0]]) # Parámetros visuales v_offset = 18 h_offset = 2 tablas_dist = 6 tablas_space = [""] * (tablas_dist - 4) cant_keys_tabla = len(KEYS_INTERES_INTERNAS) - 2 # -2 => iny_bool y control_flag no se muestran space_titles = cant_keys_tabla - h_offset - 1 # Dejar filas vacías al inicio (para el gráfico y formato) for _ in range(v_offset): worksheet.append([]) # -------------- Fila 1 (nombres de bombas) -------------- fila1 = [] for alias_bomba in bomb_names: fila1 += [""] * h_offset fila1.append(f"Bomba {alias_bomba}") fila1 += [""] * space_titles fila1 += tablas_space worksheet.append(fila1) # -------------- Fila 2 (cabeceras) -------------- fila2 = [] for _ in bomb_names: fila2 += ([""] * h_offset + ["Hora", "Oxígeno mg/L", "Saturación %", "Temperatura °C"]) fila2 += tablas_space[0:-h_offset] worksheet.append(fila2) # -------------- Rangos para bordes y gráficos -------------- from openpyxl.utils import get_column_letter Rangos = {} for idx, channel in enumerate(bomb_names): c_inicial = get_column_letter(idx * tablas_dist + h_offset + 1) c_final = get_column_letter(idx * tablas_dist + h_offset + cant_keys_tabla) c_graph = get_column_letter(idx * tablas_dist + h_offset + 1) # Columna donde poner la imagen Rangos[channel] = { 'L_Inicial': c_inicial, 'L_Final': c_final, 'L_Graph': c_graph, 'celdaInicial': f"{c_inicial}{2 + v_offset}", 'celdaFinal': f"{c_final}{(2 + v_offset + max_column)}", } # -------------- Llenar filas de datos -------------- values_for_charts = {} for fila_idx in range(max_column): nueva_fila = [] for idx, channel in enumerate(bomb_names): datos_bomba = datos_sensores_bombas[channel] if len(datos_bomba) > fila_idx: # data_fila = [hora, oxigeno_mg, saturacion, temperatura, iny_bool, control_flag] data_fila = [datos_bomba[fila_idx].get(k) for k in KEYS_INTERES] # Ajustes si saturación = -1 (u otro control) if data_fila[2] == -1: data_fila[1] = "-" data_fila[2] = "-" data_fila[3] = "-" # Recolectar instantes de inyección if data_fila[-2] is True: rangos_iny[channel].append(data_fila[0]) # Recolectar instantes de VDF if CONTROL == "VDF": if data_fila[-1] == "01": rangos_vdf[channel].append(data_fila[0]) else: if data_fila[-1] == "00aa": rangos_vdf[channel].append(data_fila[0]) # Añadir celdas vacías + [hora, O2, saturación, temp] nueva_fila.append('') nueva_fila.append('') nueva_fila += data_fila[:-2] # omite iny_bool y control_flag nueva_fila += tablas_space[0:-h_offset] # Para gráficos if channel not in values_for_charts: values_for_charts[channel] = [] values_for_charts[channel].append(data_fila) else: # No hay datos en esta fila para esa bomba nueva_fila.append('') nueva_fila.append('') nueva_fila += ["", "", "", ""] nueva_fila += tablas_space[0:-h_offset] worksheet.append(nueva_fila) # -------------- Generar gráficos e insertarlos como imagen -------------- import matplotlib.dates as mdates import matplotlib.pyplot as plt from io import BytesIO for channel_data in values_for_charts: datos_canal = values_for_charts[channel_data] if not datos_canal: continue horas_str = [] y_values = [] saturacion_values = [] for fila in datos_canal: # fila = [hora, O2_mg, saturacion, temp, iny_bool, ctrl_flag] hora = fila[0] o2 = fila[1] sat = fila[2] # Filtramos los que tengan valores numéricos if isinstance(o2, (float, int)) and isinstance(sat, (float, int)): horas_str.append(hora) y_values.append(o2) saturacion_values.append(sat) # Convertimos horas a formato datetime try: horas = [datetime.strptime(h, "%H:%M:%S") for h in horas_str if isinstance(h, str)] except ValueError as e: print("DEBUG: Error en conversión de tiempo:", e) horas = [] if not horas: continue x = mdates.date2num(horas) fig, ax1 = plt.subplots(figsize=(10, 7.3), dpi=100) ax1.plot(x, y_values, label="Nivel O2 (mg/L)", color="#1e4882", linewidth=2) ax1.set_xlim(min(x), max(x)) ax1.set_ylim(0, 12) ax1.set_xlabel("Hora del día", fontsize=12) ax1.set_ylabel("Nivel de O2 (mg/L)", color="#1e4882", fontsize=12) ax1.tick_params(axis='y', labelcolor="#1e4882") ax2 = ax1.twinx() ax2.plot(x, saturacion_values, label="Saturación %", color="#ab0202", linestyle="--", linewidth=2) ax2.set_ylim(0, max(saturacion_values)) ax2.set_ylabel("Saturación %", color="#ab0202") ax2.tick_params(axis='y', labelcolor="#ab0202") ax1.set_title(f"Sensor {channel_data}") ax1.xaxis.set_major_locator(mdates.MinuteLocator(interval=60)) ax1.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M')) fig.autofmt_xdate() # Leyenda unificada lines_1, labels_1 = ax1.get_legend_handles_labels() lines_2, labels_2 = ax2.get_legend_handles_labels() ax1.legend(lines_1 + lines_2, labels_1 + labels_2, loc='lower center', bbox_to_anchor=(0.5, -0.25), ncol=2) plt.grid(True) plt.tight_layout() img_data = BytesIO() plt.savefig(img_data, format='png', dpi=100, bbox_inches='tight', pad_inches=0.1) plt.close(fig) img_data.seek(0) img = ExcelImg(img_data) img.width = 400 img.height = 292 celda = Rangos[channel_data]["L_Graph"] + "2" worksheet.add_image(img, celda) # -------------- Marca de agua (opcional) -------------- try: import sys # base_dir = os.path.dirname(os.path.abspath(sys.argv[0])) base_dir = os.path.dirname(os.path.abspath(__file__)) img_path = os.path.join(base_dir, "images", "Logo_Reportes.png") # img_path = "/var/www/itgmanager/itgmanager/nanoox/images/Logo_Reportes.png" print(f"{base_dir=}") print(f"{img_path=}") add_markwater_images(worksheet, img_path, start_column="C", start_row=21, repeats=2, step=7) except Exception as e: print(f"DEBUG: Error al agregar la marca de agua: {e}") # -------------- Estilos (fuentes, relleno, bordes...) -------------- font_blanco = Font(b=True, color="FFFFFF") fill_naranja = PatternFill(fill_type="solid", start_color='e84e0e', end_color='e84e0e') fill_iny = PatternFill(fill_type="solid", start_color='FFFF00', end_color='FFFF00') fill_vdf = PatternFill(fill_type="solid", start_color='C8C8C8', end_color='C8C8C8') border_thin = Border( left=Side(border_style="thin", color="000000"), right=Side(border_style="thin", color="000000"), top=Side(border_style="thin", color="000000"), bottom=Side(border_style="thin", color="000000") ) # Cabecera (fila 1 y fila 2) for idx, channel in enumerate(bomb_names): let1 = get_column_letter(idx * tablas_dist + h_offset + 1) let2 = get_column_letter(idx * tablas_dist + h_offset + 2) let3 = get_column_letter(idx * tablas_dist + h_offset + 3) let4 = get_column_letter(idx * tablas_dist + h_offset + 4) cell_merge = worksheet[f"{let1}{v_offset+1}"] cell_merge.font = font_blanco cell_merge.fill = fill_naranja c1 = worksheet[f"{let1}{v_offset+2}"] c2 = worksheet[f"{let2}{v_offset+2}"] c3 = worksheet[f"{let3}{v_offset+2}"] c4 = worksheet[f"{let4}{v_offset+2}"] for c in [c1, c2, c3, c4]: c.font = font_blanco c.fill = fill_naranja # Combinar celdas para los títulos for col_idx in range(len(bomb_names)): start_col = col_idx * tablas_dist + h_offset + 1 end_col = col_idx * tablas_dist + h_offset + cant_keys_tabla worksheet.merge_cells( start_row=v_offset+1, start_column=start_col, end_row=v_offset+1, end_column=end_col ) # Borde y rellenos for channel in bomb_names: rango_celdas = worksheet[Rangos[channel]['celdaInicial'] : Rangos[channel]['celdaFinal']] for row in rango_celdas: for cell in row: cell.border = border_thin # Columna de la hora hora_val = row[0].value # Si coincide con inyección y/o VDF if hora_val in rangos_iny[channel] and hora_val in rangos_vdf[channel]: cell.fill = fill_iny elif hora_val in rangos_vdf[channel]: cell.fill = fill_vdf # Alineación for row in worksheet.iter_rows(min_row=4, max_row=(max_column + v_offset + 2)): for cell in row: cell.alignment = Alignment( horizontal='center', vertical='center', shrink_to_fit=True, wrap_text=False ) # Ajustar ancho de columnas for col in worksheet.columns: max_length = 0 col_letter = col[0].column_letter for cell in col: val = cell.value if val and isinstance(val, str) and len(val) > max_length: max_length = len(val) adjusted_width = max_length + 2 if adjusted_width > 2: worksheet.column_dimensions[col_letter].width = adjusted_width # Protegemos la hoja si lo deseas worksheet.protection.sheet = True worksheet.protection.objects = True worksheet.protection.scenarios = True worksheet.protection.password = password else: # documentos_fecha vacío => no creamos hoja print(f"DEBUG: No se generará hoja para '{fecha_format}' (sin datos)") # Fuera del for print(f"DEBUG: Finalizado el for. hay_hoja_con_datos = {hay_hoja_con_datos}") # Si no se creó ninguna hoja (ningún día tuvo datos): if not hay_hoja_con_datos: print("DEBUG: No hubo datos en todo el rango.") dest_path = os.path.join(save_path, file_name) if os.path.exists(dest_path): os.remove(dest_path) return False else: # Guardamos el workbook una sola vez dest_path = os.path.join(save_path, file_name) print(f"DEBUG: A punto de guardar el Workbook en {dest_path}") if not os.path.exists(save_path): raise FileNotFoundError(f"No existe la carpeta destino: {save_path}") workbook.security.workbookPassword = password workbook.security.lockStructure = True workbook.save(dest_path) workbook.close() print(f"DEBUG: Archivo Excel guardado correctamente en {dest_path}") return True except Exception as e: import traceback print(f"DEBUG: Error en gen_reporte_data: {e}") print(traceback.format_exc()) return None def add_markwater_images(worksheet, img_path, start_column="C", start_row=21, repeats=4, step=7): """ Inserta repetidamente la imagen en la hoja, avanzando horizontalmente columnas que pueden exceder la 'Z' (soporta AA, AB, etc.). :param worksheet: Hoja de cálculo (openpyxl worksheet) :param img_path: Ruta al archivo de imagen :param start_column: Columna inicial (string), p.ej "C" :param start_row: Fila inicial (int) :param repeats: Cuántas veces quieres repetir la imagen :param step: Cuántas columnas saltar cada vez (int) """ # 1) Convertir la columna inicial a índice numérico (A=1, B=2, C=3, ... AA=27, etc.) start_col_num = column_index_from_string(start_column) for i in range(repeats): # 2) Cálculo de la siguiente columna col_index = start_col_num + i * step # Opcional: verificar que col_index no exceda el máximo de Excel (16384 columnas en XLSX) if col_index > 16384: print(f"La columna calculada ({col_index}) excede el límite de Excel (XFD). Deteniendo.") break if col_index < 1: print(f"Índice de columna no válido ({col_index}). Se ignora.") continue # 3) Convertir el índice numérico a la correspondiente letra(s) de columna new_col_str = get_column_letter(col_index) # 4) Construir la posición en la hoja (ej. "C21", "J21", "AA21", etc.) new_position = f"{new_col_str}{start_row}" # 5) Crear la imagen y setear tamaño img_copy = ExcelImg(img_path) img_copy.width = 600 img_copy.height = 450 # 6) Insertar la imagen en la hoja worksheet.add_image(img_copy, new_position) print(f"Imagen insertada en {new_position}") print("Marca de agua agregada (con expansión de columnas) satisfactoriamente.") def get_escritorio_path(): """Obtiene la ruta del escritorio utilizando funciones nativas de Windows.""" buf = create_unicode_buffer(260) # MAX_PATH de Windows ctypes.windll.shell32.SHGetFolderPathW(0, 0x0000, 0, 0, buf) return buf.value """ def gen_reporte_data(db, rango_fechas, save_path): try: print("----------ENTRAMOS Al try gen_reporte_data --------") KEYS_INTERES =["hora", "O2", "SO2", "Temp", "RLY", "Start"] print(rango_fechas) sensores_collection = db[f'sensores'] # Supongamos que 'msg' es un diccionario que contiene las claves 'FechaInicio' y 'FechaFin' # Convertir las fechas de cadena a objetos datetime start = datetime.strptime(rango_fechas['startDate'], "%d/%m/%Y") end = datetime.strptime(rango_fechas['endDate'], "%d/%m/%Y") #-- Calculo de cantidad de dias a analizar --# if len( str(end-start).split(" ") ) > 1: diasEntre = int( str(end-start).split(" ")[0] ) + 1 # para que tome en cuenta ultimo dia print("Se han seleccionado ",diasEntre," dias") else: diasEntre = 1 #* -- Crear libro nuevo en Excel-- # workbook = openpyxl.Workbook() ws_names = [] #-- Analisis por dia entre fechas del rango --# for i in range(diasEntre): # print("Comenzando a generar reporte...") dia = str(start + timedelta(days=i)).split(" ")[0].split("-") fecha_format = "-".join(dia) print(fecha_format) #lista documentos donde se encuentra la fecha documentos_fecha = get_documentos_fecha(sensores_collection, fecha_format) if documentos_fecha: #dicc con llave-> nombre bomba ("VA1") y valor: lista con datos sensores #diccionario para almacenar informacion del documento datos_sensores_bombas = get_sensores_bombas(documentos_fecha) print(fecha_format, len(documentos_fecha), datos_sensores_bombas.keys()) # Nombres de cámaras bomb_names = list(datos_sensores_bombas.keys()) print(bomb_names) rangos_iny = {alias_bomba: [] for alias_bomba in bomb_names } rangos_vdf = {alias_bomba: [] for alias_bomba in bomb_names } #cantidad de elementos por columna max_column = len(datos_sensores_bombas[bomb_names[0]]) print(f"creando pagina {fecha_format}") #Creacion de pagina especifica para la fecha analizada worksheet = workbook.create_sheet(fecha_format) ws_names.append(fecha_format) #* ----- especificaciones para paginas en el excel ---- * v_offset = 12 #espacio vertical desde el inicio de la pagina h_offset = 2 #espacio horizontal desde el inicio de la pagina tablas_dist = 6 #espacio entre tablas tablas_space = [""]*(tablas_dist - 4) cant_keys_tabla = len(KEYS_INTERES) -2 #keys interes - auxiliares para definicino de color celda space_titles = cant_keys_tabla - h_offset - 1 #* ------------------------------------------------------- #* -------------------------------------------- Agregar filas --------------------------------------- # for i in range(v_offset): worksheet.append([]) # Dejar celdas vacias al inicio #* -- Primera Fila : Nombres de las estanques -- # fila = [] for alias_bomba in bomb_names: #en formato fila para facilitar visualizacion fila+= [""]*h_offset ; fila.append(f"Bomba {alias_bomba}") ; fila+= [""]*space_titles fila += tablas_space worksheet.append(fila) print("SEGUNDA FILA") #* -- Segunda Fila : información de las jaulas -- # estructura = ([""]*h_offset + ["Hora" ,"Oxígeno mg/L", "Saturación %" , 'Temperatura °C'] )*len(bomb_names) worksheet.append( estructura ) #* -- Filas siguientes: -- # #* Agrega iterativamente los datos directamente desde la base de datos Rangos = {} # Rango de celdas de cada canal for idx,channel in enumerate(bomb_names): cell_inicial_Letra = get_column_letter( idx*tablas_dist + h_offset +1 ) # El ultimo +1 es por desplazamiento lateral de las tablas cell_final_Letra = get_column_letter( idx*tablas_dist + h_offset + cant_keys_tabla ) cell_graph = get_column_letter( idx*tablas_dist + h_offset +1) Rangos[channel] = { 'L_Inicial':cell_inicial_Letra, 'L_Final': cell_final_Letra, 'L_Graph': cell_graph, 'celdaInicial': str(cell_inicial_Letra + str(2 + v_offset)), 'celdaFinal': str(cell_final_Letra + str( len(datos_sensores_bombas[channel]) + 2 + v_offset ) ), } for i in range(max_column): nueva_fila = [] # Para cada dato -> recorrer todos los canales for idx,channel in enumerate(bomb_names): data_fila = [datos_sensores_bombas[channel][i][key] for key in KEYS_INTERES] if data_fila[2] == -1: data_fila[1] = '-' data_fila[2] = '-' data_fila[3] = '-' if data_fila[-2] == "1": rangos_iny[channel].append(data_fila[0]) if data_fila[-1] == "01": rangos_vdf[channel].append(data_fila[0]) nueva_fila.append( '' ) nueva_fila.append( '' ) nueva_fila+=data_fila[:-2] nueva_fila += tablas_space[0:-h_offset] # # Agregar a la planilla la fila con todas las columnas worksheet.append( nueva_fila ) #* -------------------------------------------- Graficos --------------------------------------- # print("GRAFICOS") for idx, channel in enumerate(bomb_names): # Crear un gráfico de líneas chart = LineChart() chart.width = 10.3 # Establece el ancho del gráfico en unidades de caracteres (por defecto es 15) chart.height = 5.3 chart.title = "Medición sensor O2" chart.style = 13 chart.x_axis.title = 'Hora' chart.y_axis.title = 'mg/L' letter_graph =Rangos[channel]["L_Graph"] + "2" categories = Reference(worksheet, min_col=idx*tablas_dist + h_offset + 1, max_col=idx*tablas_dist + h_offset + 1, min_row=14, max_row=max_column) # Referencia a los datos del gráfico data = Reference(worksheet, min_col=idx*tablas_dist + h_offset + 2, max_col=idx*tablas_dist + h_offset + 2, min_row=14, max_row=max_column) # Agregar datos al gráfico chart.add_data(data, titles_from_data=True) chart.set_categories(categories) # Posiciones de la leyenda chart.legend.position = 'b' # "t" para arriba , "b" para abajo , "l" para la izquierda , "r" para la derecha # Cambiar el color de la línea chart.series[0].graphicalProperties.line.solidFill = "4a7ebb" # Código de color RGB en formato hexadecimal # Establecer el grosor de la línea chart.series[0].graphicalProperties.line.width = 20000 # Ubicar el gráfico en la hoja de trabajo worksheet.add_chart(chart, letter_graph) # # #* -------------------------------------------- Estilos --------------------------------------- # # # # Marca de Agua # # try: # # img = ExcelImg(ExcelIMG_import) # # img.anchor = 'C21' # Posicion inicial de imagen # # img.width = 600 # 400 => 6 columnas # # img.height = 450 # 300 => 16 filas # # worksheet.add_image(img) # # except: # # pass # # print('Marca de agua no encontrada') font = Font(b=True, color="FFFFFF")# color="00FFFFFF" -> blanco fill = PatternFill(fill_type="solid", start_color='e84e0e',#'204D84', end_color='e84e0e')#'204D84') fill_iny = PatternFill(fill_type="solid", start_color='a2ff00',#'204D84', end_color='a2ff00')#'204D84') fill_vdf = PatternFill(fill_type="solid", start_color='C8C8C8',#'204D84', end_color='C8C8C8')#'204D84') # Estilo de bordes: border = Border(left=Side(border_style="thin", color="000000"), right=Side(border_style="thin", color="000000"), top=Side(border_style="thin", color="000000"), bottom=Side(border_style="thin", color="000000")) for idx,channel in enumerate(bomb_names): Letra1 = get_column_letter( idx*tablas_dist + h_offset + 1 ) # El +1 es por desplazamiento lateral de las tablas Letra2 = get_column_letter( idx*tablas_dist + h_offset + 2 ) Letra3 = get_column_letter( idx*tablas_dist + h_offset + 3 ) Letra4 = get_column_letter( idx*tablas_dist + h_offset + 4 ) #-- Diseño celdas -- # cell_merge = worksheet[Letra1 + str(v_offset +1)] cell_merge.font = font cell_merge.fill = fill # Definicion celdas cell_1 = worksheet[Letra1 + str(v_offset + 2)] cell_2 = worksheet[Letra2 + str( v_offset +2 )] cell_3 = worksheet[Letra3 + str( v_offset +2 )] cell_4 = worksheet[Letra4 + str( v_offset +2 )] # Font: cell_1.font = font cell_2.font = font cell_3.font = font cell_4.font = font # Fill: cell_1.fill = fill cell_2.fill = fill cell_3.fill = fill cell_4.fill = fill #-- Combinar cheldas de titulos --# for column in range( len(bomb_names) ): worksheet.merge_cells(start_row=v_offset + 1, start_column=column*tablas_dist + h_offset +1, end_row=v_offset +1, end_column=column*tablas_dist + h_offset +cant_keys_tabla ) # fila 2 ya que se deja una vacía al inicio #-- Aplicar el borde a un conjunto de celdas --# for channel in bomb_names: cell_range = worksheet[Rangos[channel]['celdaInicial'] : Rangos[channel]['celdaFinal'] ] for row in cell_range: for cell in row: cell.border = border if row[0].value in rangos_iny[channel] and row[0].value in rangos_vdf[channel]: cell.fill = fill_iny elif row[0].value in rangos_vdf[channel]: cell.fill = fill_vdf #-- alineacion de celdas -- for row in worksheet.iter_rows(min_row=4, max_row= max_column + v_offset + 2): for cell in row: cell.alignment = Alignment(horizontal='center', vertical='center', shrink_to_fit=True, wrap_text=False) # Ajustar el ancho de las columnas solo para aquellas que contienen datos for column in worksheet.columns: max_length = 0 for cell in column: try: if cell.value is not None and len(str(cell.value)) > max_length: max_length = len(cell.value) except: pass adjusted_width = (max_length + 2) if adjusted_width > 2: # Ajustar solo si hay datos en la columna worksheet.column_dimensions[column[0].column_letter].width = adjusted_width if 'Sheet' in workbook.sheetnames: workbook.remove(workbook['Sheet']) print("GUARDANDO") # Ruta completa del archivo Excel dest_path = os.path.join(save_path)#, "Reporte.xlsx") try: # Guardar el archivo Excel workbook.save(dest_path ) workbook.close() print("SE GENERO CORRECTAMENTE") except: print("NON PUDE") # return True return True except Exception as e: print(e, e.__traceback__.tb_lineno) # return False """ #original #deprecated def __gen_reporte_historial(db, rango_fechas, save_path): # Supongamos que 'msg' es un diccionario que contiene las claves 'FechaInicio' y 'FechaFin' # Convertir las fechas de cadena a objetos datetime start = datetime.strptime(rango_fechas['startDate'], "%d/%m/%Y") end = datetime.strptime(rango_fechas['endDate'], "%d/%m/%Y") #-- Calculo de cantidad de dias a analizar --# if len( str(end-start).split(" ") ) > 1: diasEntre = int( str(end-start).split(" ")[0] ) + 1 # para que tome en cuenta ultimo dia print("Se han seleccionado ",diasEntre," dias") else: diasEntre = 1 # if rango_fechas["Horas"]: # HoraInicio = rango_fechas["HoraInicio"] # Horafin = rango_fechas["HoraFin"] # print("Se ha escogido 1 dia y entre horas...") # else: # print("Se ha escogido 1 dia...") # sensores_collection = db[f'historial_control'] #[f'sensores'] sensores_collection = db.historial_control #[f'sensores'] aviots_id = get_aviots_id(db) print(aviots_id) #diccionario con el formato {date: [{"hora": hour, "msg": msg}, {"hora": hour, "msg": msg}, ...] } dicc_registros = dict() for aviot_id in aviots_id: #lista de registros con {"date": value, "msg": value} list_regs = get_documentos_historial(sensores_collection, aviot_id)[0]["historial_control"] for reg in list_regs: date = reg["date"] msg = reg["msg"] # color = reg["color"] day, hour = date.split(" ") if day not in dicc_registros: dicc_registros[day] = [] dicc_registros[day].append( {"hora": hour, "msg": msg})#, "color": color}) #* ----- especificaciones para paginas en el excel ---- * v_offset = 2 #espacio vertical desde el inicio de la pagina h_offset = 2 #espacio horizontal desde el inicio de la pagina tablas_dist = 6 #espacio entre tablas tablas_space = [""]*(tablas_dist - 4) cant_keys_tabla = 3 space_titles = cant_keys_tabla - h_offset - 1 #* -- Crear libro nuevo en Excel-- # workbook = openpyxl.Workbook() ws_names = [] #* ------------------------------------------------------- colors = dict() Rangos = {} # Rango de celdas de ca in enumerate(dicc_registros): for key,value in dicc_registros.items(): cell_inicial_Letra = get_column_letter( h_offset +1 ) # El ultimo +1 es por desplazamiento lateral de las tablas cell_final_Letra = get_column_letter( h_offset + cant_keys_tabla ) cell_graph = get_column_letter( h_offset +1) Rangos[key] = { 'L_Inicial':cell_inicial_Letra, 'L_Final': cell_final_Letra, 'L_Graph': cell_graph, 'celdaInicial': str(cell_inicial_Letra + str(1 + v_offset)), 'celdaFinal': str(cell_final_Letra + str( len(value) + 2 + v_offset ) ), } # for reg in value: # print(reg["color"], key, reg["msg"]) # if key not in colors: # colors[key] = [] # if reg["color"] != "#FFFFFF": # colors[key].append(reg["color"]) # else: # colors[key].append("#000000") # #-- Analisis por dia entre fechas del rango --# for i in range(diasEntre): # print("Comenzando a generar reporte...") dia = str(start + timedelta(days=i)).split(" ")[0].split("-") fecha_format = "-".join(dia) # print(fecha_format, dicc_registros.keys(), fecha_format in dicc_registros.keys() ) if fecha_format in dicc_registros.keys() : # #Creacion de pagina especifica para la fecha analizada worksheet = workbook.create_sheet(fecha_format) ws_names.append(fecha_format) #* -------------------------------------------- Agregar filas --------------------------------------- # for i in range(v_offset): worksheet.append([]) # Dejar celdas vacias al inicio #* -- Primera Fila : Nombres de las estanques -- # estructura = ([""]*h_offset + ["Historial de control"]) worksheet.append( estructura ) # #* -- Segunda Fila : información de las jaulas -- # fila= [""]*h_offset + ["Hora","Categoría" ,"Descripción"] worksheet.append(fila) # #* -- Filas siguientes: -- # # #* Agrega iterativamente los datos directamente desde la base de datos sheet_data_src = dicc_registros[fecha_format] #lista de diccionarios #cantidad de elementos por columna max_column = len(sheet_data_src) for i in range(len(sheet_data_src)): nueva_fila = [] nueva_fila.append( '' ) nueva_fila.append( '' ) print(sheet_data_src[i]) hora, msj = list(sheet_data_src[i].values()) try: idx = msj.find("]") cat = msj[:idx+1] dec = msj[idx+1:] nueva_fila += [hora, cat, dec] except Exception as e: print(e) # print(msj, e) # nueva_fila += [hora, msj, msj] worksheet.append( nueva_fila ) #* -------------------------------------------- Estilos --------------------------------------- # # Marca de Agua try: img = ExcelImg(ExcelIMG_import) img.anchor = 'C21' # Posicion inicial de imagen img.width = 600 # 400 => 6 columnas img.height = 450 # 300 => 16 filas worksheet.add_image(img) except: pass print('Marca de agua no encontrada') font = Font(b=True, color="FFFFFF")# color="00FFFFFF" -> blanco fill = PatternFill(fill_type="solid", start_color='e84e0e',#'204D84', end_color='e84e0e')#'204D84') fill_error = PatternFill(fill_type="solid", start_color='a2ff00',#'204D84', end_color='a2ff00')#'204D84') fill_vdf = PatternFill(fill_type="solid", start_color='C8C8C8',#'204D84', end_color='C8C8C8')#'204D84') # Estilo de bordes: border = Border(left=Side(border_style="thin", color="000000"), right=Side(border_style="thin", color="000000"), top=Side(border_style="thin", color="000000"), bottom=Side(border_style="thin", color="000000")) # #-- Diseño celdas -- # Letra1 = get_column_letter( h_offset + 1 ) # El +1 es por desplazamiento lateral de las tablas Letra2 = get_column_letter( h_offset + 2 ) Letra3 = get_column_letter( h_offset + 3 ) cell_merge = worksheet[Letra1 + str(v_offset +1)] cell_merge.font = font cell_merge.fill = fill # Definicion celdas cell_1 = worksheet[Letra1 + str(v_offset + 2)] cell_2 = worksheet[Letra2 + str( v_offset +2 )] cell_3 = worksheet[Letra3 + str( v_offset +2 )] # cell_4 = worksheet[Letra4 + str( v_offset +2 )] # Font: cell_1.font = font cell_2.font = font cell_3.font = font # cell_4.font = font # Fill: cell_1.fill = fill cell_2.fill = fill cell_3.fill = fill # cell_4.fill = fill #-- Combinar celdas de titulos --# worksheet.merge_cells(start_row=v_offset + 1, start_column= h_offset +1, end_row=v_offset +1, end_column= h_offset +3 ) # fila 2 ya que se deja una vacía al inicio # #-- Aplicar el borde a un conjunto de celdas --# cell_range = worksheet[Rangos[fecha_format]['celdaInicial'] : Rangos[fecha_format]['celdaFinal'] ] for row in cell_range: for cell in row: cell.border = border # if row[0].value in colors: # cell.fill = PatternFill(fill_type="solid", start_color=colors[row[0].value][:1] ,end_color= colors[row[0].value][1:] ) # elif row[0].value in rangos_vdf[channel]: # cell.fill = fill_vdf #-- color texto --# # for idx, row in enumerate( worksheet.iter_rows(min_row=v_offset +3, max_row= max_column + v_offset +2)): # for cell in row: # cell.font = font = Font(color=colors[fecha_format][idx][1:] )# color="00FFFFFF" -> blanco #-- alineacion de celdas -- for idx, row in enumerate( worksheet.iter_rows(min_row=v_offset +1, max_row= max_column + v_offset +2)): for cell in row: cell.alignment = Alignment(horizontal='center', vertical='center', shrink_to_fit=True, wrap_text=False) # cell.font = font = Font(color=colors[fecha_format][idx][1:] )# color="00FFFFFF" -> blanco # print(len( worksheet.iter_rows(min_row=v_offset, max_row= max_column + v_offset + 4))) # Ajustar el ancho de las columnas solo para aquellas que contienen datos for column in worksheet.columns: max_length = 0 for cell in column: try: if cell.value is not None and len(str(cell.value)) > max_length: max_length = len(cell.value) except: pass adjusted_width = (max_length + 2) if adjusted_width > 2: # Ajustar solo si hay datos en la columna worksheet.column_dimensions[column[0].column_letter].width = adjusted_width if 'Sheet' in workbook.sheetnames: workbook.remove(workbook['Sheet']) # path_escritorio = os.path.join(os.path.join(os.environ['USERPROFILE']), 'Desktop') # path_save= os.path.join("../reportes", os.getcwd()) # Ruta completa del archivo Excel # dest_path = os.path.join(path_save, "Reporte_historial.xlsx") # # Guardar el archivo Excel # workbook.save("Reporte_historial.xlsx") # workbook.close() # Ruta completa del archivo Excel dest_path = os.path.join(save_path)#, "Reporte.xlsx") # Guardar el archivo Excel workbook.save(dest_path ) workbook.close() return True def get_historial_control_as_dict(db): """ Obtiene todos los logs de la colección 'historial_control' y los organiza en un diccionario. Las llaves del diccionario son las fechas (en formato 'YYYY-MM-DD') y los valores son listas de logs. :return: Diccionario con las fechas como llaves y listas de logs como valores. """ try: # Conexión a la base de datos MongoDB # db = get_db() # collection = db['historial_control'] collection = db.historial_control # Obtener todos los documentos de la colección documents = collection.find({}) # Transformar los documentos en un diccionario historial_dict = {} for doc in documents: date = doc.get('date') # Obtener la fecha logs = doc.get('logs', []) # Obtener los logs (por defecto, lista vacía si no existe) historial_dict[date] = logs return historial_dict except Exception as e: print(f"Error al obtener el historial de 'historial_control': {e} {e.__traceback__.tb_lineno}") raise def get_historial_control_as_dict_sm(db, rango_fechas=None): """ Obtiene todos los logs de la colección 'historial_control' y los organiza en un diccionario. Las llaves del diccionario son las fechas (en formato 'YYYY-MM-DD') y los valores son listas de logs. Si se proporciona 'rango_fechas', se espera un diccionario con: {'startDate': '27/02/2025', 'endDate': '27/02/2025'} y se realiza la conversión para filtrar los documentos entre ambas fechas. :return: Diccionario con las fechas como llaves y listas de logs como valores, o None si no se encuentran datos. """ try: collection = db.historial_control query = {} if rango_fechas: start_date_input = rango_fechas.get('startDate') end_date_input = rango_fechas.get('endDate') if start_date_input: dt_start = datetime.strptime(start_date_input, "%d/%m/%Y") start_date_str = dt_start.strftime("%Y-%m-%d") if end_date_input: dt_end = datetime.strptime(end_date_input, "%d/%m/%Y") end_date_str = dt_end.strftime("%Y-%m-%d") query["date"] = {} if start_date_input: query["date"]["$gte"] = start_date_str if end_date_input: query["date"]["$lte"] = end_date_str documents = collection.find(query) historial_dict = {} for doc in documents: date = doc.get('date') # Se asume que en la base se guarda en formato "YYYY-MM-DD" logs = doc.get('logs', []) if logs: historial_dict[date] = logs # Si el diccionario quedó vacío, retornar None if not historial_dict: return None return historial_dict except Exception as e: print(f"Error al obtener el historial de 'historial_control': {e} {e.__traceback__.tb_lineno}") raise def dedupe_logs_full(logs): """ Elimina duplicados considerando TODO el contenido del log. Un log se considera duplicado si todos sus pares (clave, valor) coinciden. """ seen = set() unique_logs = [] for log in logs: if not isinstance(log, dict): continue # Crear una clave hashable a partir de todos los pares (clave, valor) # sorted(log.items()) asegura que el orden de las claves no afecte try: key = tuple(sorted(log.items())) except Exception as e: # Si por alguna razón hay valores no hasheables, lo dejamos pasar sin dedupe unique_logs.append(log) continue if key in seen: # Ya vimos un log exactamente igual → lo descartamos continue seen.add(key) unique_logs.append(log) return unique_logs def get_historial_control(db, centro_id): """ Obtiene los logs de la colección 'historial_control' para el centro y la fecha actual. Devuelve una lista de logs SIN duplicados exactos (por todo el contenido del log). """ try: collection = db.historial_control fecha = datetime.now().strftime("%Y-%m-%d") documents = collection.find({"centro_id": centro_id, "date": fecha}) all_logs = [] for doc in documents: logs = doc.get('logs', []) if isinstance(logs, list): all_logs.extend(logs) # DEBUG opcional: ver cuántos logs había antes print(f"[DEBUG] Logs totales antes de dedupe para {centro_id}: {len(all_logs)}") # Eliminar duplicados EXACTOS filtered_logs = dedupe_logs_full(all_logs) # DEBUG opcional: ver cuántos logs quedan después print(f"[DEBUG] Logs totales después de dedupe para {centro_id}: {len(filtered_logs)}") return filtered_logs except Exception as e: print(f"Error al obtener el historial de 'historial_control': {e} {e.__traceback__.tb_lineno}") raise #From Richi def gen_reporte_historial(db, rango_fechas, save_path, file_name): print("********ENTRANDO A GENERAR HISTORIAL ************") # Convertir las fechas de cadena a objetos datetime start = datetime.strptime(rango_fechas['startDate'], "%d/%m/%Y") end = datetime.strptime(rango_fechas['endDate'], "%d/%m/%Y") #-- Calculo de cantidad de dias a analizar --# if len( str(end-start).split(" ") ) > 1: diasEntre = int( str(end-start).split(" ")[0] ) + 1 # para que tome en cuenta ultimo dia print("Se han seleccionado ",diasEntre," dias") else: diasEntre = 1 hay_hoja_con_datos = False try: #diccionario con el formato {date: [{"hora": hour, "msg": msg}, {"hora": hour, "msg": msg}, ...] } dicc_logs = get_historial_control_as_dict(db) print(f"{dicc_logs=}") #* ----- especificaciones para paginas en el excel ---- * v_offset = 2 #espacio vertical desde el inicio de la pagina h_offset = 2 #espacio horizontal desde el inicio de la pagina tablas_dist = 6 #espacio entre tablas tablas_space = [""]*(tablas_dist - 4) cant_keys_tabla = 3 space_titles = cant_keys_tabla - h_offset - 1 #* -- Crear libro nuevo en Excel-- # workbook = openpyxl.Workbook() ws_names = [] # # #* ------------------------------------------------------- colors = dict() Rangos = {} # Rango de celdas de ca in enumerate(dicc_logs): for key,value in dicc_logs.items(): cell_inicial_Letra = get_column_letter( h_offset +1 ) # El ultimo +1 es por desplazamiento lateral de las tablas cell_final_Letra = get_column_letter( h_offset + cant_keys_tabla ) cell_graph = get_column_letter( h_offset +1) Rangos[key] = { 'L_Inicial':cell_inicial_Letra, 'L_Final': cell_final_Letra, 'L_Graph': cell_graph, 'celdaInicial': str(cell_inicial_Letra + str(1 + v_offset)), 'celdaFinal': str(cell_final_Letra + str( len(value) + 2 + v_offset ) ), } for reg in value: # print(f"{reg['color']} {key} {reg['msg']}") if key not in colors: colors[key] = [] if reg["color"] != "#FFFFFF": colors[key].append(reg["color"]) elif reg["color"] != "#E6E6FA": colors[key].append("#000000") else: colors[key].append("#000000") # # #-- Analisis por dia entre fechas del rango --# for i in range(diasEntre): # print("Comenzando a generar reporte...") dia = str(start + timedelta(days=i)).split(" ")[0].split("-") fecha_format = "-".join(dia) if fecha_format in dicc_logs.keys() : hay_hoja_con_datos = True #Creacion de pagina especifica para la fecha analizada worksheet = workbook.create_sheet(fecha_format) ws_names.append(fecha_format) #* -------------------------------------------- Agregar filas --------------------------------------- # for i in range(v_offset): worksheet.append([]) # Dejar celdas vacias al inicio #* -- Primera Fila : Nombres de las estanques -- # estructura = ([""]*h_offset + ["Historial de control"]) worksheet.append( estructura ) # #* -- Segunda Fila : información de las jaulas -- # fila= [""]*h_offset + ["Hora","Categoría" ,"Descripción"] worksheet.append(fila) # #* -- Filas siguientes: -- # # #* Agrega iterativamente los datos directamente desde la base de datos sheet_data_src = dicc_logs[fecha_format] #lista de diccionarios # print(f"{sheet_data_src=}") # #cantidad de elementos por columna max_column = len(sheet_data_src) for idx, log in enumerate(sheet_data_src): # for i in range(len(sheet_data_src)): nueva_fila = [] nueva_fila.append( '' ) nueva_fila.append( '' ) try: msg = log.get("msg", "") idx = msg.find("]") ts_hh_mm_ss = log.get("date").split(" ")[1] cat = msg[:idx+1] dec = msg[idx+1:] nueva_fila += [ts_hh_mm_ss, cat, dec] except Exception as e: print(f"Error: {e} {e.__traceback__.tb_lineno}") # print(msj, e) # nueva_fila += [hora, msj, msj] worksheet.append( nueva_fila ) #* -------------------------------------------- Estilos --------------------------------------- # # Marca de Agua # Construir la ruta absoluta a la imagen try: base_dir = os.path.dirname(os.path.abspath(__file__)) # Directorio actual del script (database) img_path = os.path.join(base_dir, "images", "Logo_Reportes.png") # Subir un nivel y entrar a 'images' # Verificar si la imagen existe if not os.path.exists(img_path): raise FileNotFoundError(f"No se encuentra la imagen en: {img_path}") add_markwater_images(worksheet, img_path, start_column="C", start_row=21, repeats=2, step=7) except Exception as e: print(f"Error al agregar la marca de agua: {e}") font = Font(b=True, color="FFFFFF")# color="00FFFFFF" -> blanco fill = PatternFill(fill_type="solid", start_color='e84e0e',#'204D84', end_color='e84e0e')#'204D84') fill_error = PatternFill(fill_type="solid", start_color='a2ff00',#'204D84', end_color='a2ff00')#'204D84') fill_vdf = PatternFill(fill_type="solid", start_color='C8C8C8',#'204D84', end_color='C8C8C8')#'204D84') # Estilo de bordes: border = Border(left=Side(border_style="thin", color="000000"), right=Side(border_style="thin", color="000000"), top=Side(border_style="thin", color="000000"), bottom=Side(border_style="thin", color="000000")) #-- Diseño celdas -- # Letra1 = get_column_letter( h_offset + 1 ) # El +1 es por desplazamiento lateral de las tablas Letra2 = get_column_letter( h_offset + 2 ) Letra3 = get_column_letter( h_offset + 3 ) cell_merge = worksheet[Letra1 + str(v_offset +1)] cell_merge.font = font cell_merge.fill = fill # Definicion celdas cell_1 = worksheet[Letra1 + str(v_offset + 2)] cell_2 = worksheet[Letra2 + str( v_offset +2 )] cell_3 = worksheet[Letra3 + str( v_offset +2 )] # cell_4 = worksheet[Letra4 + str( v_offset +2 )] # Font: cell_1.font = font cell_2.font = font cell_3.font = font # cell_4.font = font # Fill: cell_1.fill = fill cell_2.fill = fill cell_3.fill = fill # cell_4.fill = fill #-- Combinar celdas de titulos --# worksheet.merge_cells(start_row=v_offset + 1, start_column= h_offset +1, end_row=v_offset +1, end_column= h_offset +3 ) # fila 2 ya que se deja una vacía al inicio # #-- Aplicar el borde a un conjunto de celdas --# cell_range = worksheet[Rangos[fecha_format]['celdaInicial'] : Rangos[fecha_format]['celdaFinal'] ] for row in cell_range: for cell in row: cell.border = border # if row[0].value in colors: # cell.fill = PatternFill(fill_type="solid", start_color=colors[row[0].value][:1] ,end_color= colors[row[0].value][1:] ) # elif row[0].value in rangos_vdf[channel]: # cell.fill = fill_vdf #-- color texto --# for idx, row in enumerate( worksheet.iter_rows(min_row=v_offset +3, max_row= max_column + v_offset +2)): for cell in row: cell.font = font = Font(color=colors[fecha_format][idx][1:] )# color="00FFFFFF" -> blanco #-- alineacion de celdas -- for idx, row in enumerate( worksheet.iter_rows(min_row=v_offset +1, max_row= max_column + v_offset +2)): for cell in row: cell.alignment = Alignment(horizontal='center', vertical='center', shrink_to_fit=True, wrap_text=False) # cell.font = font = Font(color=colors[fecha_format][idx][1:] )# color="00FFFFFF" -> blanco # print(len( worksheet.iter_rows(min_row=v_offset, max_row= max_column + v_offset + 4))) # Ajustar el ancho de las columnas solo para aquellas que contienen datos for column in worksheet.columns: max_length = 0 for cell in column: try: if cell.value is not None and len(str(cell.value)) > max_length: max_length = len(cell.value) except: pass adjusted_width = (max_length + 2) if adjusted_width > 2: # Ajustar solo si hay datos en la columna worksheet.column_dimensions[column[0].column_letter].width = adjusted_width worksheet.protection.sheet=True worksheet.protection.objects=True worksheet.protection.scenarios=True worksheet.protection.password=password else: print(f"DEBUG: No se generará hoja para '{fecha_format}' (sin datos)") # Fuera del for print(f"DEBUG: Finalizado el for. hay_hoja_con_datos = {hay_hoja_con_datos}") # Si no se creó ninguna hoja (ningún día tuvo datos): if not hay_hoja_con_datos: print("DEBUG: No hubo datos en todo el rango.") return False else: # 5) Guardar una sola vez si hubo al menos una hoja dest_path = os.path.join(save_path, file_name) print(f"DEBUG: A punto de guardar el Workbook en {dest_path}") if not os.path.exists(save_path): raise FileNotFoundError(f"No existe la carpeta destino: {save_path}") if 'Sheet' in workbook.sheetnames: workbook.remove(workbook['Sheet']) workbook.security.workbookPassword=password workbook.security.lockStructure=True workbook.save(dest_path) workbook.close() print(f"DEBUG: Archivo Excel guardado correctamente en {dest_path}") return True except Exception as e: print(f"DEBUG: Error en gen_reporte_data: {e} (línea {e.__traceback__.tb_lineno})") return None # def _gen_reporte_historial(db, rango_fechas, save_path, file_name): # print("********ENTRANDO A GENERAR HISTORIAL ************") # # print(f"{rango_fechas=}") # # Convertir las fechas de cadena a objetos datetime # start = datetime.strptime(rango_fechas['startDate'], "%d/%m/%Y") # end = datetime.strptime(rango_fechas['endDate'], "%d/%m/%Y") # #-- Calculo de cantidad de dias a analizar --# # if len( str(end-start).split(" ") ) > 1: # diasEntre = int( str(end-start).split(" ")[0] ) + 1 # para que tome en cuenta ultimo dia # print("Se han seleccionado ",diasEntre," dias") # else: # diasEntre = 1 # hay_hoja_con_datos = False # try: # #diccionario con el formato {date: [{"hora": hour, "msg": msg}, {"hora": hour, "msg": msg}, ...] def gen_reporte_generadores(db, rango_fechas, save_path, file_name): """ Genera un reporte en Excel dentro del rango de fechas (rango_fechas['startDate'], rango_fechas['endDate']). Se apoya en la colección 'generadores' dentro de la BD 'db'. Retorna: - True => si al menos un día tuvo datos y se generó el archivo exitosamente. - False => si no hubo datos en ningún día (no se genera archivo). - None => si ocurre un error no controlado (excepción). Parámetros visuales: - La tabla de cada hoja comienza en la fila 18 - Deja siempre 2 columnas en blanco (inicio en la columna C) - Aplica bordes 'thin' a toda la tabla - Ajuste automático de columnas con un ancho máximo de 50 """ import os import openpyxl from openpyxl.utils import get_column_letter from openpyxl.styles import Alignment, Border, Side from datetime import datetime, timedelta from openpyxl.drawing.image import Image as ExcelImage # Ajusta aquí las claves que quieras usar LLAVES = ["timestamp", "plc_status", "hf_status", "partida", "detencion"] #KEYS_INTERES # Encabezados que se verán en el Excel HEADERS = ["Timestamp", "PLC Status", "HF Status", "Partida", "Detención"] # Identificamos el centro a partir del nombre del archivo (por si lo necesitas) centro_id = file_name.split("-reporte")[0] try: # Colección de generadores en tu BD generadores_collection = db.generadores # Convertir strings de fecha a datetime start = datetime.strptime(rango_fechas['startDate'], "%d/%m/%Y") end = datetime.strptime(rango_fechas['endDate'], "%d/%m/%Y") # Calcular días a analizar (inclusive) diasEntre = (end - start).days + 1 if diasEntre < 1: print("DEBUG: Rango de fechas inválido (end < start).") return False print(f"DEBUG: Se han seleccionado {diasEntre} día(s)") # Crear el Workbook workbook = openpyxl.Workbook() # Eliminar la hoja por defecto si existe if 'Sheet' in workbook.sheetnames: workbook.remove(workbook['Sheet']) hay_hoja_con_datos = False # Parámetros para la ubicación de la tabla START_ROW = 18 # La tabla comienza en la fila 18 BLANK_COLS = 2 # Dejar 2 columnas en blanco (la data comienza en la columna C) thin_border_side = Side(style='thin', color='000000') # Borde delgado thin_border = Border( left=thin_border_side, right=thin_border_side, top=thin_border_side, bottom=thin_border_side ) # Recorrer cada día del rango for i in range(diasEntre): fecha_actual = start + timedelta(days=i) fecha_format = fecha_actual.strftime("%Y-%m-%d") # Ajustar si tu DB usa otro formato # Tu función para extraer documentos (ajusta según sea necesario) # documentos_fecha = get_documentos_fecha_generadores(generadores_collection, fecha_format) documentos_fecha = get_documentos_fecha_generadores(db, centro_id, fecha_format)[0] print(f"DEBUG: Fecha '{fecha_format}': documentos hallados = {len(documentos_fecha)}") # Verificamos si hay datos para este día if documentos_fecha and len(documentos_fecha) > 0: hay_hoja_con_datos = True print(f"DEBUG: Creando hoja de Excel para {fecha_format}...") # Crear hoja worksheet = workbook.create_sheet(fecha_format) # 1) Agregamos el encabezado en la fila 18, con 2 columnas en blanco al inicio # Formamos la lista ["", ""] + HEADERS header_row = ([""] * BLANK_COLS) + HEADERS # Antes de escribir el encabezado, podemos insertar filas vacías # para posicionar el encabezado exactamente en la fila 18: for _ in range(START_ROW - 1): # 17 filas vacías worksheet.append([]) # Escribimos la fila de encabezado (ahora en la fila 18) worksheet.append(header_row) # 2) Agregar los datos for doc in documentos_fecha['generador']: # Construimos la fila con 2 columnas vacías + datos data_row = [""] * BLANK_COLS + [ doc.get("timestamp", ""), doc.get("plc_status", ""), doc.get("hf_status", ""), doc.get("partida", ""), doc.get("detencion", "") ] worksheet.append(data_row) img_data = create_boolean_plot(documentos_fecha['generador'], fecha_format) if img_data: img_for_excel = ExcelImage(img_data) worksheet.add_image(img_for_excel, "C2") # Determinar cuál es la primera y última fila con datos (incluyendo encabezado) total_filas = len(documentos_fecha['generador']) # sin contar el encabezado first_data_row = START_ROW last_data_row = START_ROW + total_filas # fila donde termina la data # Determinar las columnas mín. y máx. con datos first_data_col = BLANK_COLS + 1 # si BLANK_COLS=2 => comienza en columna 3 (C) last_data_col = BLANK_COLS + len(LLAVES) # 2 + 5 = 7 (col G si tienes 5 keys) # 3) Aplicar estilos (bordes + alineación centrada) for row_cells in worksheet.iter_rows( min_row=first_data_row, max_row=last_data_row, min_col=first_data_col, max_col=last_data_col ): for cell in row_cells: cell.border = thin_border cell.alignment = Alignment(horizontal='center', vertical='center') # 4) Ajustar el ancho de columnas donde hay datos # Iteramos de la primera a la última columna con datos for col_idx in range(first_data_col, last_data_col + 1): col_letter = get_column_letter(col_idx) max_length = 0 # Revisamos cada celda para calcular el ancho óptimo for row_idx in range(first_data_row, last_data_row + 1): cell_value = worksheet.cell(row=row_idx, column=col_idx).value if cell_value and len(str(cell_value)) > max_length: max_length = len(str(cell_value)) # Ajuste con un colchón de +2 y un máximo de 50 adjusted_width = min(max_length + 2, 50) worksheet.column_dimensions[col_letter].width = adjusted_width # 5) Aplicar formato condicional en la tabla según "PLC Status" from openpyxl.styles import PatternFill # Definir los rellenos (fill) para cada caso fill_green = PatternFill(start_color="FF296F4D", end_color="FF296F4D", fill_type="solid") # #296f4d FF296F4D 00FF00 fill_red = PatternFill(start_color="FF7F1D1D", end_color="FF7F1D1D", fill_type="solid") # #7f1d1d FF7F1D1D FF0000 fill_orange = PatternFill(start_color="FFD7592A", end_color="FFD7592A", fill_type="solid") # #d7592a FFD7592A FFA500 # Determinar la columna donde se encuentra "PLC Status". # Se calcula sumando las columnas en blanco más la posición de "plc_status" en KEYS_INTERES (contando desde 1) col_plc_status = BLANK_COLS + LLAVES.index("plc_status") + 1 col_partida_status = BLANK_COLS + LLAVES.index("partida") + 1 # Iterar sobre cada fila de datos (excluyendo la fila de encabezado, que está en START_ROW) for row in worksheet.iter_rows( min_row=START_ROW + 1, max_row=last_data_row, min_col=first_data_col, max_col=last_data_col): # Obtener la celda que corresponde a "PLC Status" plc_cell = row[col_plc_status - first_data_col] partida_cell = row[col_partida_status - first_data_col] try: plc_val = int(plc_cell.value) partida_val = int(partida_cell.value) except (ValueError, TypeError): plc_val = None partida_val = None # Seleccionar el relleno según el valor fill = None if plc_val == 60: if partida_val == 100: fill = fill_green elif plc_val == 100: fill = fill_red elif plc_val == -1: fill = fill_orange # else: # fill = None # Si se asignó un relleno, aplicarlo a todas las celdas de la fila de la tabla if fill: for cell in row: cell.fill = fill else: print(f"DEBUG: No se generará hoja para '{fecha_format}' (sin datos)") # Finalmente, guardamos el archivo solo si al menos un día tuvo datos if hay_hoja_con_datos: if not os.path.exists(save_path): raise FileNotFoundError(f"No existe la carpeta destino: {save_path}") # Construir la ruta absoluta a la imagen try: base_dir = os.path.dirname(os.path.abspath(__file__)) # Directorio actual del script (database) img_path = os.path.join(base_dir, "images", "Logo_Reportes.png") # Subir un nivel y entrar a 'images' # Verificar si la imagen existe if not os.path.exists(img_path): raise FileNotFoundError(f"No se encuentra la imagen en: {img_path}") add_markwater_images(worksheet, img_path, start_column="C", start_row=21, repeats=1, step=7) except Exception as e: print(f"Error al agregar la marca de agua: {e}") dest_path = os.path.join(save_path, file_name) # worksheet.add_image(img_for_excel, "C2") workbook.save(dest_path) workbook.close() print(f"DEBUG: Archivo Excel guardado correctamente en {dest_path}") return True else: print("DEBUG: No hubo datos en todo el rango.") return False except Exception as e: print(f"DEBUG: Error en gen_reporte_generadores: {e} (línea {e.__traceback__.tb_lineno})") return None def get_documentos_fecha_generadores(db,centro_id,fecha): filtro = {"centro_id": centro_id,"date": fecha} documentos = list(db.generadores.find(filtro)) print(documentos[0]) print(documentos[0]['generador']) filtered_documents = [] for documento in documentos: print(documento['generador']) print() try: # plc_status = documento.get('plc_status') # fecha_doc = documento.get('fecha') generador = documento.get('generador') # Revisar que se obtuvieron todos los datos if not all([ generador]): pass else: filtered_documents.append(documento) except Exception as e: print("Ocurrió un error:", e.__traceback__.tb_lineno, e) pass return filtered_documents def create_boolean_plot(documentos_fecha, fecha_str): """ Crea una gráfica de los valores booleanos 'detencion' y 'partida' a lo largo del tiempo (eje X). Parámetros: ----------- documentos_fecha : list Lista de documentos (diccionarios) que contienen los campos: - timestamp (str) - detencion (bool) - partida (bool) fecha_str : str String representando la fecha (por ejemplo, "YYYY-MM-DD") que se usará para el título de la gráfica. Retorna: -------- image_data : BytesIO Objeto BytesIO con la imagen en formato PNG. """ import matplotlib matplotlib.use('Agg') # Para generar la figura en segundo plano (sin ventana GUI) import matplotlib.pyplot as plt import matplotlib.dates as mdates from datetime import datetime from io import BytesIO # Prepara las listas para graficar times = [] # detencion_vals = [] partida_vals = [] # Parsear cada documento for doc in documentos_fecha: # documentos_fecha['generador'], por ejemplo # Ajusta el formato de timestamp según venga en tu DB # Ej: si viene en "2023-09-12 15:30:00", ajusta en el strptime: t = datetime.strptime(doc["timestamp"], "%Y-%m-%d %H:%M:%S") # times.append(t) if doc.get("plc_status")==60: if doc.get("partida")==100: partida_vals.append(1) times.append(t) else: partida_vals.append(0) times.append(t) elif doc.get("plc_status")==-1: if partida_vals != []: partida_vals.append(partida_vals[-1]) times.append(t) else: partida_vals.append(0) times.append(t) # Convertimos booleans a 0/1 (o manejar None) # detencion_vals.append(1 if doc.get("detencion") else 0) # partida_vals.append(1 if doc.get("partida") else 0) # # Crear la figura fig, ax = plt.subplots(figsize=(5, 3), dpi = 100) ax.set_title(f"Generador: Arranques día {fecha_str}") ax.set_xlabel("Tiempo") ax.set_ylabel("Estado Generador") # Formateo de la escala de tiempo en el eje X ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M')) # Solo horas y minutos ax.xaxis.set_major_locator(mdates.AutoDateLocator()) fig.autofmt_xdate() # Graficar los valores # Usamos 'step' o 'plot' según prefieras la visualización de valores booleanos # ax.step(times, detencion_vals, color="#ab0202",label="Detención", where='post') ax.step(times, partida_vals, color="#296f4d",label="Partida", where='post') #V: #1e823e, A: #1e4882, Vgpt: #128d12 ax.set_yticks([0,1]);ax.set_yticklabels(["OFF", "ON"]) ax.legend() plt.tight_layout() # Guardar la figura en un BytesIO image_data = BytesIO() plt.savefig(image_data, format='png', dpi=100, bbox_inches='tight', pad_inches=0.1) plt.close(fig) # Cerramos la figura para liberar memoria image_data.seek(0) # Regresar el puntero al inicio return image_data