import sys sys.path.insert(0, './') from collections import defaultdict from datetime import datetime, timedelta from pathlib import Path # Rerportería - Excel import openpyxl from openpyxl.utils import get_column_letter from openpyxl.styles import Border, Side, PatternFill, Font, Alignment # GradientFill, from openpyxl.utils import get_column_letter, column_index_from_string # from openpyxl.chart import LineChart, Reference from openpyxl.drawing.image import Image as ExcelImg # from openpyxl.drawing.text import CharacterProperties from openpyxl.drawing.image import Image as ExcelImg import numpy as np from io import BytesIO # Matplotlib - Gráficos import matplotlib matplotlib.use('Agg') # Backend sin interfaz gráfica import matplotlib.pyplot as plt import matplotlib.dates as mdates class SimpleLogger: def info(self, msg): print(f"[INFO] {msg}") # print(msg) def warning(self, msg): print(f"[WARNING] {msg}") #print(msg) def error(self, msg): print(f"[ERROR] {msg}") def debug(self, msg): print(f"[DEBUG] {msg}") return logger = SimpleLogger() def get_escritorio_path(): # """Obtiene la ruta del escritorio utilizando funciones nativas de Windows.""" # buf = create_unicode_buffer(260) # MAX_PATH de Windows # windll.shell32.SHGetFolderPathW(0, 0x0000, 0, 0, buf) #return # buf.value """ Retorna la ruta al escritorio del usuario independientemente del sistema operativo. Usa Path.home() y asume que la carpeta se llama 'Desktop'. """ home = Path.home() desktop = home / "Desktop" return desktop def asegurar_indices_sensores_timeseries(db, coll_name="sensores_timeseries") -> None: logger.debug(f"En 'asegurar_indices_sensores_timeseries' {coll_name=}") """ Verifica y crea los índices en la colección time series 'sensores_timeseries' para los campos 'bomba_id' y 'centro_id', si no existen ya. """ ts_col = db[ coll_name ] # Obtenemos la información de los índices existentes existing_indexes = ts_col.index_information() # Definimos los índices que queremos tener desired_indexes = { "bomba_id_1": [("bomba_id", 1)], "centro_id_1": [("centro_id", 1)] } for name, spec in desired_indexes.items(): if name not in existing_indexes: # creamos el índice con nombre explícito para facilitar su identificación ts_col.create_index(spec, name=name) print(f"Índice '{name}' creado sobre {spec}") # else: # print(f"Índice '{name}' ya existe, se omite.") # -- Formateo de datos -- # def obtener_datos_por_fecha_por_alias( db, fecha_str, sensors_coll_name="sensores_timeseries", metadata_coll_name = "metadata"): """ Obtiene todas las lecturas de un día, agrupadas por alias de bomba, usando un aggregation pipeline para offload al servidor Mongo. Args: db: instancia de la base de datos. fecha_str: fecha en 'YYYY-MM-DD'. centro_id: opcional, filtra por centro. Returns: Dict donde la clave es el alias y el valor es la lista de documentos de timeseries. """ print() logger.debug(f"En 'obtener_datos_por_fecha_por_alias' {sensors_coll_name=}") logger.debug(f"En 'obtener_datos_por_fecha_por_alias' {metadata_coll_name=}") logger.debug(f"{fecha_str=}") sensores_ts_col = db[ sensors_coll_name ] # Parsear fecha de inicio y fin fecha = datetime.strptime(fecha_str, "%Y-%m-%d") siguiente_dia = fecha + timedelta(days=1) # Match inicial en timeseries match_ts = { "timestamp": {"$gte": fecha, "$lt": siguiente_dia} } logger.debug(match_ts) # Pipeline simplificado # pipeline = [ # {"$match": match_ts}, # {"$lookup": { # "from": metadata_coll_name, # "localField": "bomba_id", # "foreignField": "_id", # "as": "meta" # }}, # {"$unwind": "$meta"}, # # Filtramos solo metadatos de tipo 'bomba' # {"$match": {"meta.tipo": "bomba"}}, # {"$project": { # "_id": 0, # "alias": { # "$ifNull": [ # "$meta.alias", # {"$concat": ["bomba_", {"$toString": "$bomba_id"}]} # ] # }, # "timestamp": 1, # "O2": 1, "SO2": 1, "TEMP": 1, # "RLY": 1, "InStart": 1, "InSel":1, # "Control": 1, "Presion": 1, "Io":1, "Vo":1, # "alarms": 1, # "Status": 1 # }}, # {"$group": { # "_id": "$alias", # "lecturas": {"$push": { # "timestamp": "$timestamp", # "O2": "$O2", # "SO2": "$SO2", # "TEMP": "$TEMP", # "RLY": "$RLY", # "InStart": "$InStart", # "InSel": "$InSel", # "Control": "$Control", # "Presion": "$Presion", # "Io": "$Io", # "Vo": "$Vo", # "alarms": "$alarms", # "Status": "$Status" # }} # }} # ] # Pipeline simplificado pipeline = [ {"$match": match_ts}, {"$lookup": { "from": metadata_coll_name, "localField": "bomba_id", "foreignField": "_id", "as": "meta" }}, {"$unwind": "$meta"}, # Filtramos solo metadatos de tipo 'bomba' {"$match": {"meta.tipo": "bomba"}}, {"$project": { "_id": 0, "alias": { "$ifNull": [ "$meta.alias", {"$concat": ["bomba_", {"$toString": "$bomba_id"}]} ] }, "timestamp": 1, "O2": 1, "SO2": 1, # "SENSOR_SO2":1, # <--- Mapeo correcto "TEMP": 1, "SetPoint":1, "SetPointOFF":1, "RLY": 1, "InStart": 1, "InSel":1, "Control": 1, "Presion": 1, "Io":1, "Vo":1, "alarms": 1, "Status": 1 }}, {"$group": { "_id": "$alias", "lecturas": {"$push": { "timestamp": "$timestamp", "O2": "$O2", "SO2": "$SO2", "SENSOR_SO2": "$SENSOR_SO2", "TEMP": "$TEMP", "RLY": "$RLY", "InStart": "$InStart", "InSel": "$InSel", "Control": "$Control", "Presion": "$Presion", "Io": "$Io", "Vo": "$Vo", "SetPoint":"$SetPoint", "SetPointOFF":"$SetPointOFF", "alarms": "$alarms", "Status": "$Status" }} }} ] # Ejecuta la agregación (si hay mucha data, puedes activar allowDiskUse=True) cursor = sensores_ts_col.aggregate(pipeline, allowDiskUse=True) # Devuelve un dict alias → lista de lecturas print() return {doc["_id"]: doc["lecturas"] for doc in cursor} # Versión antigua def _obtener_datos_por_fecha_por_alias( db, fecha_str, sensors_coll_name="sensores_timeseries", metadata_coll_name = "metadata"): # client = MongoClient("mongodb://localhost:27017") print() logger.debug(f"En 'obtener_datos_por_fecha_por_alias' {sensors_coll_name=}") logger.debug(f"En 'obtener_datos_por_fecha_por_alias' {metadata_coll_name=}") metadata_col = db[metadata_coll_name] sensores_ts_col = db[sensors_coll_name] logger.debug(f"{metadata_col=}") # Parsear la fecha en formato 'YYYY-MM-DD' fecha = datetime.strptime(fecha_str, "%Y-%m-%d") siguiente_dia = fecha + timedelta(days=1) # Buscar bombas filtro_metadata = {"tipo": "bomba"} # if centro_id: # filtro_metadata["centro_id"] = centro_id # logger.debug(f"{filtro_metadata=}") alias_map = {} for doc in metadata_col.find(filtro_metadata): # print(doc) alias_map[doc["_id"]] = doc.get("alias", f"bomba_{doc['_id']}") # Filtro series de tiempo filtro_ts = { "bomba_id": {"$in": list(alias_map.keys())}, "timestamp": {"$gte": fecha, "$lt": siguiente_dia} } # if centro_id: # filtro_ts["centro_id"] = centro_id # logger.debug(f"{filtro_ts=}") # Campos a extraer campos = { "_id": 0, "bomba_id": 1, "timestamp": 1, "O2": 1, "SO2": 1, "TEMP": 1, "RLY": 1, "InStart": 1, "Control": 1, "Presion": 1 } resultados = defaultdict(list) # logger.debug(resultados) for doc in sensores_ts_col.find(filtro_ts, campos): alias = alias_map.get(doc["bomba_id"], str(doc["bomba_id"])) doc.pop("bomba_id") resultados[alias].append(doc) print() return dict(resultados) def get_historial_control_as_dict( db, coll_name="historial_control" ): """ 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['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: logger.error(f"Error al obtener el historial de 'historial_control': {e} {e.__traceback__.tb_lineno}") raise # -- Estilos -- # def get_styles(): try: font = Font(b=True, color="FFFFFF")# color="00FFFFFF" -> blanco fill = PatternFill(fill_type="solid", start_color='e84e0e',#'204D84', end_color='e84e0e')#'204D84') celeste = "92CDDC" verde = "92DC97" amarillo = 'FFFF00' gris = 'C8C8C8' rojo = 'EE6b6E' fill_iny = PatternFill(fill_type="solid", start_color=celeste, #'204D84', end_color=celeste ) #'204D84') fill_iny_man = PatternFill(fill_type="solid", start_color=verde , end_color=verde ) fill_vdf = PatternFill(fill_type="solid", start_color=gris, #'204D84', end_color=gris) #'204D84') fill_alarm = PatternFill(fill_type="solid", start_color=rojo, #'204D84', end_color=rojo) #'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")) except Exception as e: raise Exception(F"Error obteniendo estilos ({e.__traceback__.tb_lineno}): {e}") return font, fill, fill_iny, fill_iny_man, fill_vdf, border, fill_alarm 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) """ try: # 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): # Colocar imagen cada columna por medio if i%2 == 1: continue # 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: logger.error(f"La columna calculada ({col_index}) excede el límite de Excel (XFD). Deteniendo.") break if col_index < 1: logger.warning(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) logger.debug(f"Imagen insertada en {new_position}") logger.info("Marca de agua agregada (con expansión de columnas) satisfactoriamente.") except Exception as e: logger.error(f"Error insertando marcas de agua ({e.__traceback__.tb_lineno}): {e}.") def insert_matplot_graphs( worksheet, values_for_charts, rangos_iny_on_ts, rangos_bomba_on_ts, rangos_iny_on_manual_ts, Rangos ): try: plt.rcParams.update({ 'font.size': 22, 'axes.titlesize': 25, 'xtick.labelsize': 20, 'ytick.labelsize': 20, 'legend.fontsize': 25, 'figure.titlesize': 25 }) for idx, channel_data in enumerate(values_for_charts): try: # --- tus datos --- x = mdates.date2num(values_for_charts[channel_data]["ts"]) o2 = values_for_charts[channel_data]["O2"] so2 = values_for_charts[channel_data]["SO2"] if len(x) <= 0 or len(o2) <= 0 or len(so2) <= 0: logger.warning(f"{channel_data} sin datos para graficar.") continue ts_inyecting = [ ts for ts in rangos_iny_on_ts[channel_data] if ts in rangos_bomba_on_ts[channel_data] ] ts_iny_manual = rangos_iny_on_manual_ts[channel_data] # --- columnas y margen --- total_graphs = 2 \ + (1 if ts_inyecting else 0) \ + (1 if ts_iny_manual else 0) ncol = 2 bottom_margin = 0.3 # 0.30 if total_graphs > 2 else 0.40 # --- figura + ejes en caja fija --- ancho_fig, alto_fig = 14, 7.3 fig = plt.figure(figsize=(ancho_fig, alto_fig), dpi=100) left, right, top = 0.10, 0.95, 0.95 width = right - left height = top - bottom_margin ax1 = fig.add_axes([left, bottom_margin, width, height]) # --- eje izq (O2) + barras --- ax1.plot( x, o2, label="Nivel de 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=16) ax1.set_ylabel( "Nivel de O2 (mg/L)", color="#1e4882", fontsize=16 ) ax1.tick_params(axis='y', labelcolor="#1e4882") y_min, y_max = ax1.get_ylim() if ts_inyecting: x_inj = mdates.date2num(ts_inyecting) dx = ( np.min(np.diff(np.sort(x_inj))) # * 0.8 if len(x_inj) > 1 else 1/(24*60) ) ax1.bar( x_inj, [y_max]*len(x_inj), width=dx, bottom=y_min, color="#92CDDC", alpha=0.3, label="Inyección O2 - Automática" ) ax1.set_ylim(y_min, y_max) if ts_iny_manual: x_inj = mdates.date2num(ts_iny_manual) dx = ( np.min(np.diff(np.sort(x_inj))) # * 0.8 if len(x_inj) > 1 else 1/(24*60) ) ax1.bar( x_inj, [y_max]*len(x_inj), width=dx, bottom=y_min, color="#92DC97", alpha=0.3, label="Inyección O2 Manual" ) ax1.set_ylim(y_min, y_max) # --- eje der (SO2) --- ax2 = ax1.twinx() ax2.plot( x, so2, label="Saturación %", color="#ab0202", linestyle="--", linewidth=2 ) ax2.set_xlim(min(x), max(x)) ax2.set_ylim(0, max(so2)+10) ax2.set_ylabel( "Saturación %", color="#ab0202", fontsize=16 ) ax2.tick_params(axis='y', labelcolor="#ab0202") # --- formateo X + rotación --- ax1.xaxis.set_major_locator( mdates.MinuteLocator(interval=60) ) ax1.xaxis.set_major_formatter( mdates.DateFormatter('%H:%M') ) fig.autofmt_xdate() ax1.set_title(f"Sensor {channel_data}", fontsize=20) ax1.grid(True) # --- margen + leyenda EN AX1 --- fig.subplots_adjust(bottom=bottom_margin) # — RESTAURAR ROTACIÓN DIAGONAL DE LOS TICKS X — for lbl in ax1.get_xticklabels(): lbl.set_rotation(45) lbl.set_ha('right') lines_1, labels_1 = ax1.get_legend_handles_labels() lines_2, labels_2 = ax2.get_legend_handles_labels() handles = lines_1 + lines_2 labels = labels_1 + labels_2 # Aquí está la clave: bbox_transform=ax1.transAxes ax1.legend( handles,labels, ncol=ncol, loc='upper center', bbox_to_anchor=(0.5, -0.2), bbox_transform=ax1.transAxes ) # --- guardar sin recortes --- img_data = BytesIO() fig.savefig(img_data, format='png', dpi=100) plt.close(fig) # --- insertar en Excel --- img_data.seek(0) img = ExcelImg(img_data) img.width, img.height = 480, 292 celda = Rangos[channel_data]["L_Graph"] + "2" worksheet.add_image(img, celda) except Exception as e: logger.error( f"Error generando reporte en linea {e.__traceback__.tb_lineno} {e}" ) except Exception as e: logger.error( f"Error insertando gráficos ({e.__traceback__.tb_lineno}): {e}" ) def apply_cell_styles( worksheet, bomb_names, Rangos, cant_keys_tabla, tablas_dist, h_offset, v_offset, max_column, rangos_bomba_on, rangos_iny_on, rangos_iny_on_manual, border, font, fill, fill_iny, fill_vdf, fill_iny_man, fill_alarm ): try: 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 ) Letra5 = get_column_letter( idx*tablas_dist + h_offset + 5 ) Letra6 = get_column_letter( idx*tablas_dist + h_offset + 6 ) print( Letra1, Letra2, Letra3, Letra4, Letra5, Letra6,) #-- 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 )] cell_5 = worksheet[Letra5 + str( v_offset +2 )] cell_6 = worksheet[Letra6 + str( v_offset +2 )] # Font: cell_1.font = font cell_2.font = font cell_3.font = font cell_4.font = font cell_5.font = font cell_6.font = font # Fill: cell_1.fill = fill cell_2.fill = fill cell_3.fill = fill cell_4.fill = fill cell_5.fill = fill cell_6.fill = fill # * -- Combinar celdas 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 i,row in enumerate(cell_range): for j,cell in enumerate(row): cell.border = border bomb_active = row[0].value in rangos_bomba_on[channel] active_inyection = row[0].value in rangos_iny_on[channel] and bomb_active iny_manual = row[0].value in rangos_iny_on_manual[channel] # Pintar inyección automática activa if active_inyection: cell.fill = fill_iny elif bomb_active: cell.fill = fill_vdf # Pintar inyección manual activa if iny_manual: cell.fill = fill_iny_man alarm_cell_value = row[-1].value print(f"{ row[-1].value} row: {i} col: {j} offset: {v_offset +2} ") if (alarm_cell_value) and ("A" in alarm_cell_value) and i>=1: cell.fill = fill_alarm # * -- 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 except Exception as e: logger.error(f"Error aplicando estilos a celdas ({e.__traceback__.tb_lineno}): {e}")