from .save_data import guardar_estructura_en_mongodb from .comap_data import save_comap_data #INVERMAR SCRIPT from datetime import datetime, timedelta import pymongo # CRUD Sensores import copy import re import os from typing import Optional # 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 get_control_by_centro_id(nanoox_db , centro_id): print(f"EN get_control_by_centro_id {centro_id}" ) metadata_col = nanoox_db["metadata"] print(f"{list(metadata_col.find({'centro_id': centro_id}))=}") for doc in metadata_col.find({"centro_id": centro_id}): print(doc) if "control" in doc: print("ENCONTRE EL CONTROL") return doc.get("control") else: return "VDF" return "PS" # def get_control_version_by_centro_id(nanoox_db , centro_id): # print(f"EN get_control_version_by_centro_id {centro_id}" ) # metadata_col = nanoox_db["metadata"] # for doc in metadata_col.find({"centro_id": centro_id}): # print(doc.keys()) # if "control_version" in doc: # print("ENCONTRE EL control_version en el documento") # return doc.get("control_version") # else: # return False # return True def get_control_version_by_centro_id(nanoox_db, metadata_col_name, centro_id): """Devuelve control_version para el centro dado o None si no existe.""" metadata_col = nanoox_db[metadata_col_name] doc = metadata_col.find_one( {"centro_id": centro_id}, projection={"control_version": 1, "_id": 0} ) print(f"EN get_control_version_by_centro_id {centro_id}" ) print(doc) return doc.get("control_version") if doc and "control_version" in doc else None 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, new_version = False): """ Procesa un único registro de datos extrayendo cada bloque y guardándolo en la BD. """ print(f"\n{'estamos en process_sensor_data'.upper():*^100}\n") print(f"{data.keys()} {centro_id}, {new_version=}") #print(centro_id) # print(f"{data.keys()}") timestamp = data.get("hora") print(f"{timestamp=}") control = data.get("control", "PS") try: # Un payload sin "modulos" (por ejemplo un centro COMAP puro) no debe # abortar el resto del bloque legacy con un KeyError. modulos = data.get("modulos") if modulos: guardar_estructura_en_mongodb(db, modulos, data.get("generators", []), centro_id, control, new_version, timestamp= timestamp) else: print(f"Sin bloque 'modulos' para {centro_id}: no se actualiza estructura logica") # Procesar historial de control print(f"\n{' # Procesar historial de control ':*^60}\n") print(f"{data.get('historial_control')=}") logs = data.get('historial_control', []) if logs: save_one_historial_data(db, logs, centro_id) print("antes de revisar generadores") if not new_version: #versino anterior --- # Procesar generador (si existe) generador = data.get('generador') print(f"{generador=}") if generador: save_one_generator_data(db, centro_id, generador) except Exception as e: print(f"error en linea {e.__traceback__.tb_lineno} {e}") # COMAP se procesa de forma independiente para no perderlo si falla # cualquiera de los bloques legacy anteriores. try: comap_block = data.get("generators_comap") or {} if comap_block: save_comap_data(db, centro_id, comap_block) else: print(f"Sin datos COMAP para {centro_id}") except Exception as e: print(f"Error procesando COMAP linea {e.__traceback__.tb_lineno}: {e}") # *----------- CRUD DB Sensores ---------* # #-- Generacion de reportes --# 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 #--------------------- Seccion historial control -----------------------------# def original_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") date_from_log = data['fecha'] formatted_current_date = __extract_date(date_from_log) # Construir la consulta para buscar el documento de la fecha actual query = {"date": formatted_current_date} # Preparar la operación de actualización o inserción update = { "$setOnInsert": {"date": formatted_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 {formatted_current_date}.") else: print(f"Documento creado para la fecha {formatted_current_date} y log agregado.") return result except Exception as e: print(f"Error al agregar log a 'historial_control': {e}") raise def _add_to_historial_control(db, centro_id, logs_list): """ Inserta logs usando BulkWrite y $addToSet con un límite de seguridad. """ if not logs_list: return # 1. Agrupar logs por fecha (por si vienen logs de días distintos en el mismo paquete) logs_por_fecha = {} for log in logs_list: # Usamos tu función existente para extraer la fecha fecha_doc = __extract_date(log.get('fecha')) or log.get('date', '').split(' ')[0] if fecha_doc not in logs_por_fecha: logs_por_fecha[fecha_doc] = [] logs_por_fecha[fecha_doc].append(log) operaciones = [] for fecha, logs in logs_por_fecha.items(): # LÍMITE DE SEGURIDAD: # Si el array ya es muy grande, no intentamos insertar más en ese documento. # Un documento de 16MB aguanta aprox 15,000 - 20,000 logs de este tipo. # Ponemos un límite conservador de 10,000 logs por día/documento. doc_actual = db.historial_control.find_one({"date": fecha, "centro_id": centro_id}, {"logs": 1}) if doc_actual and len(doc_actual.get("logs", [])) > 10000: print(f"AVISO: Documento de {fecha} para {centro_id} está cerca del límite de 16MB. Saltando inserción.") continue # Usamos $addToSet con $each para insertar múltiples logs evitando duplicados exactos operaciones.append( pymongo.UpdateOne( {"date": fecha, "centro_id": centro_id}, { "$setOnInsert": {"date": fecha, "centro_id": centro_id}, "$addToSet": {"logs": {"$each": logs}} }, upsert=True ) ) if operaciones: try: db.historial_control.bulk_write(operaciones, ordered=False) print(f"BulkWrite exitoso para {centro_id}") except pymongo.errors.BulkWriteError as bwe: # Captura errores de tamaño específicamente for error in bwe.details.get('writeErrors', []): if error.get('code') == 10334 or "larger than 16777216" in error.get('errmsg'): print(f"CRÍTICO: El documento de {centro_id} alcanzó el límite físico de 16MB.") except Exception as e: print(f"Error en BulkWrite: {e}") def __extract_date(timestamp_str: str) -> Optional[str]: """ Validates and extracts the date from a timestamp string in the format "[YYYY/MM/DD - HH:MM:SS]". Returns the date as a string in "YYYY-MM-DD" format if valid, otherwise returns None. """ pattern = r"^\[(\d{4})/(\d{2})/(\d{2}) - (\d{2}):(\d{2}):(\d{2})\]$" match = re.match(pattern, timestamp_str) if not match: return None # Format is not valid try: # Remove square brackets and parse the inner string parsed_dt = datetime.strptime(timestamp_str[1:-1], "%Y/%m/%d - %H:%M:%S") return parsed_dt.date().isoformat() # Return as "YYYY-MM-DD" except ValueError: return None # Invalid date (e.g. Feb 30) """ # Example usage example = "[2025/08/06 - 11:12:05]" result = extract_date(example) if result: print(f"Valid date: {result}") else: print("Invalid format or non-existent date.") """ def save_one_historial_data( db, historial_data, centro_id): try: _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 get_documentos_historial(collection, aviot_id): filtro = {"aviot_id": aviot_id} documento = collection.find(filtro) return list(documento) 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 get_historial_control(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: # Conexión a la base de datos MongoDB collection = db.historial_control fecha = datetime.now().strftime("%Y-%m-%d") # Obtener todos los documentos de la colección documents = collection.find({"centro_id":centro_id, "date":fecha}) # 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 historial_dict = logs return historial_dict 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_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 #----------------------------------------------------------------------------------------------# #-- Extraccion de info generadores --# def legacy_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. """ print("En GET generator version antigua") 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 # --- OPTIMIZACIÓN DE GENERADOR LEGACY --- def get_generator(db, centro_id): """ Optimizado para no iterar cursor innecesariamente. """ try: collection = db.generadores date = datetime.now().strftime("%Y-%m-%d") # Usamos find_one con ordenamiento inverso para sacar solo el último array # Esto evita traer todo el documento si es muy grande, o iterar doc = collection.find_one( {"centro_id": centro_id, "date": date}, {"generador": {"$slice": -1}} # Proyección: Solo el último elemento del array ) if doc and "generador" in doc and doc["generador"]: return doc["generador"][0] return None except Exception as e: print(f"Error get_generator: {e}") 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 def save_one_generator_data(db, centro_id, data): #fecha, try: if data: # if set(data.values()) == {None}: # return print(f"save_one_generator_data {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(f"{centro_id}: Datos de generador sin cambios, no se actualiza BD.") 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(f"No hay datos de generadores: {data}") return except Exception as e: print(e.__traceback__.tb_lineno)