#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ =============================================================================== Archivo: NOMBRE_DEL_ARCHIVO.py Autor: TU_NOMBRE (TU_EMPRESA) Fecha creación: AAAA-MM-DD Versión: VERSIÓN_INICIAL =============================================================================== Descripción: Breve descripción del propósito del archivo y su funcionalidad =============================================================================== Copyright: © AÑO TU_EMPRESA. Todos los derechos reservados. Este archivo forma parte de NOMBRE_PROYECTO y se distribuye bajo los términos de la licencia TIPO_LICENCIA. =============================================================================== Interface Change List: Versión Fecha Autor Descripción del cambio --------- ---------- ------------------ --------------------------- vX.X.X AAAA-MM-DD TU_NOMBRE Descripción breve vX.X.X AAAA-MM-DD TU_NOMBRE Descripción breve =============================================================================== """ __all__ = ["ExampleClass", "main"] # ============================================================================= # Importación de librerías # ============================================================================= # === Importes estándar === from pathlib import Path import sys import os import time import math import numpy as np from datetime import datetime, timedelta from pprint import pprint from typing import Optional, List, Dict # === Importes de terceros === from pymongo import MongoClient # 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 from io import BytesIO import matplotlib.pyplot as plt import matplotlib.dates as mdates from openpyxl.drawing.image import Image as ExcelImg from openpyxl.utils import column_index_from_string # === Importes internos === from .reports_utils import ( logger, obtener_datos_por_fecha_por_alias, get_escritorio_path, add_markwater_images, get_styles, asegurar_indices_sensores_timeseries, get_historial_control_as_dict, insert_matplot_graphs, apply_cell_styles, ) # ============================================================================= # Constantes y configuraciones globales # ============================================================================= BASE_COLUMNS = [ "Time", "SP_ON", "SP_OFF", "O₂ mg/L", "Saturation %", "Temperature °C", "Pressure bar", "Alarms", "Voltage", # "Current", ] VERTICAL_OFFSET: int = 22 HORIZONTAL_OFFSET: int = 2 SHEET_PASSWORD: str = "itg123chile" # Extractores por título de columna. añadir/quitar títulos en BASE_COLUMNS # Si un título no está en este dict, se intentará record.get() como fallback. COLUMN_EXTRACTORS = { "Time": lambda rec: ( rec.get("timestamp").strftime("%H:%M:%S") if rec.get("timestamp") else None ), "SP_ON": lambda rec: rec.get("SetPoint"), "SP_OFF": lambda rec: rec.get("SetPointOFF"), "O₂ mg/L": lambda rec: rec.get("O2"), "Saturation %": lambda rec: rec.get("SO2"), "Temperature °C": lambda rec: rec.get("TEMP"), "Pressure bar": lambda rec: rec.get("Presion"), "Alarms": lambda rec: ( ",".join(rec.get("alarms", [])) if rec.get("alarms") else "-" ), "Voltage": lambda rec: rec.get("Vo"), } _DEFAULT_THRESHOLDS = {"voltage_high": 415.0} _COLUMN_ALIASES = { "Time": ["Time", "Hora"], "SP_ON": ["SP_ON", "SetPoint", "SPON"], "SP_OFF": ["SP_OFF", "SetPointOFF", "SPOFF"], "O2": ["O2", "O₂ mg/L", "Oxígeno mg/L"], "SO2": ["SO2", "Saturación %"], "TEMP": ["TEMP", "Temperatura °C", "Temperature °C"], "Presion": ["Presion", "Presión bar", "Pressure bar"], "Alarms": ["Alarms", "Alarmas"], "Vo": ["Vo", "Voltaje", "Voltage"], # "Io": ["Io", "Corriente", "Current"], } # ============================================================================= # Definiciones locales # ============================================================================= if getattr(sys, 'frozen', False): # Modo congelado (PyInstaller): utilizar sys._MEIPASS como base, # ya que allí se encuentran los archivos de recursos. current_dir = Path(getattr(sys, '_MEIPASS')) else: # Ejecución normal: la ruta base es la del script principal current_dir = Path(__file__).resolve().parent def get_desktop_path() -> Path: """ Retorna la ruta al escritorio del usuario independientemente del sistema operativo. Para Windows, prueba 'Desktop' y 'Escritorio' bajo USERPROFILE. Para Linux/mac, prueba 'Desktop' y 'Escritorio' bajo home. Si no encuentra ninguno, retorna el directorio actual. """ home = Path.home() candidates = [] env = os.environ.get('DESKTOP_PATH') if env and Path(env).exists(): logger.info(f"Escritorio identificado: {Path(env)}\n") return Path(env) # Windows paths if os.name == 'nt': userprofile = os.environ.get('USERPROFILE') if userprofile: candidates.extend([ Path(userprofile) / 'Desktop', Path(userprofile) / 'Escritorio', Path(userprofile) / 'OneDrive - ITG Chile' / 'Desktop', Path(userprofile) / 'OneDrive - ITG Chile' / 'Escritorio', ]) else: # Unix-like candidates.extend([ home / 'Desktop', home / 'Escritorio', ]) print() for desk in candidates: if desk.exists(): logger.info(f"Escritorio identificado: {desk}\n") return desk else: logger.info(f"Ruta Escritorio errónea: {desk}") cwd = Path.cwd() logger.warning(f"No fue posible identificar escritorio. Usando directorio actual: {cwd}\n") return cwd desktop = get_desktop_path() img_path = current_dir / "images" / "Logo_Reportes.png" save_dir = desktop # current_dir # Todo: CAMBIAR A ESCRITORIO # ============================================================================= # Clases # ============================================================================= # ============================================================================= # Métodos públicos # ============================================================================= # ============================================================================= # Métodos privados # ============================================================================= # ============================================================================= # Funciones Públicas # ============================================================================= def gen_data_report( rango_fechas, control_type, db_host="localhost", db_name="NANOOX_TS", coll_name="sensores_timeseries", centro_name=None, fileobj=None, db_port=27017, excel_save_path = '/' ): """ Genera un reporte Excel con datos de sensores y estado de inyección de O2, en el rango de fechas indicado. """ time_total = time.time() if centro_name != None: capitalized_center_name = centro_name.capitalize() sensors_coll_name = f"{capitalized_center_name}_sensores" metadata_coll_name = f"{capitalized_center_name}_metadata" else: sensors_coll_name = "sensores_timeseries", metadata_coll_name = "can_timeseries", try: print(f'{sensors_coll_name=}') # 1 Conexión y setup de base de datos db = _prepare_database(db_host, db_name, db_port, sensors_coll_name) pprint(db) # 2 Crear workbook y estilos workbook, styles = _prepare_workbook() # 3 Calcular rango de fechas start = datetime.strptime(rango_fechas['startDate'], "%d/%m/%Y") end = datetime.strptime(rango_fechas['endDate'], "%d/%m/%Y") days_between = (end - start).days + 1 dates = [(start + timedelta(days=i)).date() for i in range(days_between)] logger.info(f"Generando reporte de tipo {control_type} para {len(dates)} días") # 4 Procesar día por día # lista para almacenar informacion resultante del procesamiento por dia day_contexts: list[dict] = [] print(day_contexts) for date in dates: is_successful, result = _process_day_data(db, date, control_type, workbook, sensors_coll_name, metadata_coll_name) if is_successful: day_contexts.append(result) # store context for styling else: logger.warning(result) ok_chart, msg_chart = insert_matplot_graphs_from_context(result) if not ok_chart: logger.warning(msg_chart) if not day_contexts: return False, "No data available in the selected range." # # 5) Apply styles & protect sheets (cuando implementemos) ok, msg = style_all_days_modern( day_contexts, styles, # lo que retorna tu _prepare_workbook() thresholds={"voltage_high": 415.0}, # opcional watermark_img_path="", # opcional protect=True, sheet_password="itg123chile", ) if not ok: return False, msg # 6) Save workbook is_successful, msg = _save_workbook(workbook, dates, centro_name, fileobj, time_total, excel_save_path) return is_successful, msg except Exception as e: error_msg = f"General error: {e} (line {e.__traceback__.tb_lineno})" logger.error(error_msg) return False, error_msg def main(): """Punto de entrada principal del script.""" print("Ejecutando script principal...") DB_HOST = '10.0.0.56' DB_NAME = 'NANOOX_TS_ANEC' start_date = "07/10/2025" end_date = "07/10/2025" """ current_dir = os.path.dirname(os.path.abspath(__file__)) print(f"{current_dir=}") img_source_dir = os.path.abspath(os.path.join(current_dir, '../images')) # print(f"{root_dir=}") ExcelIMG_import = os.path.join('../images', 'Logo_Reportes.png') sys.path.append(img_source_dir) """ CONTROL = "PS" # rango_fechas = {"startDate": "06/07/2025", "endDate": "06/07/2025" } rango_fechas = {"startDate": start_date, "endDate": end_date} print(rango_fechas) gen_data_report(rango_fechas=rango_fechas, control_type=CONTROL, db_name= DB_NAME, db_host=DB_HOST, centro_name= None) # Lógica principal aquí # ============================================================================= # Funciones Privadas # ============================================================================= def get_db( db_host, db_name, db_port = 27017 ): mongo_uri = f"mongodb://{db_host}:{db_port}" logger.info(f"Buscando datos en colección {db_name} de {mongo_uri}.") client = MongoClient(mongo_uri) db = client[db_name] return db def _prepare_database(db_host: str, db_name: str, db_port: int, coll_name: str): """ Conecta a la base de datos MongoDB y asegura los índices necesarios para consultas de sensores. Parámetros: - db_host: dirección IP o hostname del servidor de BD. - db_name: nombre de la base de datos. - db_port: puerto TCP de conexión (por defecto 10200). - coll_name: nombre de la colección donde se almacenan los datos de sensores. Retorna: - Objeto de base de datos (db) listo para consultas. Lanza excepción si ocurre un error de conexión o configuración. """ try: t0 = time.time() db = get_db(db_host, db_name, db_port=db_port) try: asegurar_indices_sensores_timeseries(db, coll_name) logger.debug("Índices asegurados correctamente.") except Exception as e: logger.warning(f"Error al asegurar índices: {e} {e.__traceback__.tb_lineno}") logger.info(f"Base de datos '{db_name}' conectada en {(time.time()-t0)*1e3:.2f} ms") return db except Exception as e: logger.error(f"Error al conectar con la BD: {e} {e.__traceback__.tb_lineno}") raise def _prepare_workbook(): workbook = openpyxl.Workbook() workbook.remove(workbook.active) font, fill, fill_iny, fill_iny_man, fill_vdf, border, fill_alarm = get_styles() styles = { "font": font, "fill": fill, "fill_iny": fill_iny, "fill_iny_man": fill_iny_man, "fill_vdf": fill_vdf, "border": border, "fill_alarm": fill_alarm, } return workbook, styles def _process_day_data( db, current_date: datetime.date, control_type: str, workbook, sensors_coll_name: str, metadata_coll_name: str, ) : """ Genera la hoja del día (YYYY-MM-DD) replicando exactamente el patrón original y, ADICIONALMENTE, recolecta datos para gráficos por bomba sin alterar el layout. """ try: date_str = current_date.isoformat() logger.debug(f"Processing data for date: {date_str}") sensor_data = obtener_datos_por_fecha_por_alias(db, date_str, sensors_coll_name=sensors_coll_name, metadata_coll_name= metadata_coll_name) if not sensor_data: logger.info(f"No data found for date {date_str}") return False, f"No data found for {date_str}" ws = workbook.create_sheet(title=date_str) bomb_names = sorted(sensor_data.keys()) print(f'{bomb_names=}') add_markwater_images(ws, img_path, start_column="C", start_row=21, repeats= len(bomb_names), step=7) v_offset = VERTICAL_OFFSET h_offset = HORIZONTAL_OFFSET base_columns = list(BASE_COLUMNS) print(f'{BASE_COLUMNS=}') cant_keys_tabla = len(base_columns) # 9 tablas_dist = cant_keys_tabla + 1 # 10 space_titles = cant_keys_tabla - h_offset - 1 # 9 - 2 - 1 = 6 tablas_space = [""] * (tablas_dist - cant_keys_tabla) # 10 - 9 = 1 # ---- Arrays de análisis por bomba (como tu versión) ---- rangos_iny_on = {b: [] for b in bomb_names} rangos_bomba_on = {b: [] for b in bomb_names} rangos_iny_on_manual = {b: [] for b in bomb_names} rangos_iny_on_ts = {b: [] for b in bomb_names} rangos_bomba_on_ts = {b: [] for b in bomb_names} rangos_iny_on_manual_ts = {b: [] for b in bomb_names} # ---- NUEVO: datos para gráficos por bomba (no afecta layout) ---- values_for_charts = {b: {"ts": [], "O2": [], "SO2": []} for b in bomb_names} null_data = 0 sandwich_case = 0 total_data = 0 # ---- Padding superior ---- for _ in range(v_offset): ws.append([]) # ---- Fila 1: títulos por bomba (exacto patrón original) ---- fila_titulos = [] for alias_bomba in bomb_names: fila_titulos += ([""] * h_offset) + [f"Bomba {alias_bomba}"] + ([""] * space_titles) + tablas_space ws.append(fila_titulos) # ---- Fila 2: encabezados de columnas (exacto patrón original) ---- estructura = ([""] * h_offset) + base_columns if len(bomb_names) > 1: estructura += (([""] * (h_offset - 1)) + base_columns) * (len(bomb_names) - 1) ws.append(estructura) # ---- RANGOS por bloque EXACTOS ---- Rangos = {} # Usaremos el número máximo de filas de cualquier bomba para celdaFinal por bomba max_rows_by_pump = {b: len(sensor_data[b]) for b in bomb_names} max_rows_global = max(max_rows_by_pump.values()) if max_rows_by_pump else 0 for idx, channel in enumerate(bomb_names): L_inicial = get_column_letter(idx * tablas_dist + h_offset + 1) L_final = get_column_letter(idx * tablas_dist + h_offset + cant_keys_tabla) # Primera fila de datos = v_offset + 3 row_start = v_offset + 3 row_end = row_start + max_rows_by_pump[channel] - 1 if max_rows_by_pump[channel] > 0 else row_start Rangos[channel] = { "L_Inicial": L_inicial, "L_Final": L_final, "L_Graph": get_column_letter(idx * tablas_dist + h_offset + 1), # ancla gráfica "celdaInicial": f"{L_inicial}{row_start}", "celdaFinal": f"{L_final}{row_end}", } # ---- Longitud máxima entre bombas ---- max_column = max_rows_global # ---- Filas de datos (exacto patrón original) ---- for i in range(max_column): # ¡OJO!: tu patrón usa (h_offset - 1) al inicio de CADA fila de datos nueva_fila = [""] * (# The above code is defining a variable named `h_offset` # in Python. h_offset - 1) for idx, channel in enumerate(bomb_names): rows = sensor_data[channel] data_fila = rows[i] if i < len(rows) else {} # Copias locales ts_obj = data_fila.get("timestamp") ts = ts_obj.strftime("%H:%M:%S") if ts_obj else None dato_o2 = data_fila.get("O2") dato_so2 = data_fila.get("SO2") if dato_o2 is None and dato_so2 is None: null_data += 1 else: try: if isinstance(dato_o2, float) and math.isnan(dato_o2): dato_o2 = None if isinstance(dato_so2, float) and math.isnan(dato_so2): dato_so2 = None except Exception: pass InStart = data_fila.get("InStart") RLY = data_fila.get("RLY") inyection_o2 = bool(RLY) if RLY is not None else False in_sel = data_fila.get("InSel") tcb_auto = bool(int(in_sel)) if in_sel is not None else False corriente = data_fila.get("Io") voltaje = data_fila.get("Vo") alarmas = data_fila.get("alarms", []) status = data_fila.get("Status") sp_on = data_fila.get("SetPoint") sp_off = data_fila.get("SetPointOFF") total_data += 1 if data_fila else 0 # Selección de arrays (auto -> bomba_on; manual -> iny_manual) if tcb_auto: data_array = rangos_bomba_on[channel] graph_array = rangos_bomba_on_ts[channel] else: data_array = rangos_iny_on_manual[channel] graph_array = rangos_iny_on_manual_ts[channel] prev_graph_ts = graph_array[-1] if graph_array else None last_ts = rows[i - 1]["timestamp"] if i >= 1 and i < len(rows) and rows[i - 1].get("timestamp") else None last_last_ts = rows[i - 2]["timestamp"] if i >= 2 and i < len(rows) and rows[i - 2].get("timestamp") else None correct_this_cell = (status == 9) inyecting_vdf = bool(correct_this_cell or (control_type == "VDF" and InStart == "01")) inyecting_ps = bool(correct_this_cell or (control_type == "PS" and (InStart is not None and bool(int(InStart))))) pump_running = (inyecting_vdf or inyecting_ps) and (not alarmas) prev_is_fill = (prev_graph_ts == last_ts) if (prev_graph_ts and last_ts) else False prev_prev_is_fill = (prev_graph_ts == last_last_ts) if (prev_graph_ts and last_last_ts) else False if prev_prev_is_fill and not prev_is_fill and pump_running and last_ts: graph_array.append(last_ts) data_array.append(last_ts.strftime("%H:%M:%S")) sandwich_case += 1 if pump_running and ts and ts_obj: data_array.append(ts) graph_array.append(ts_obj) if tcb_auto and inyection_o2 and pump_running and ts and ts_obj: rangos_iny_on[channel].append(ts) rangos_iny_on_ts[channel].append(ts_obj) # ===== Construcción EXACTA por bloque ===== # padding de bloque (1 celda) nueva_fila.append("") # columnas en el mismo orden que espera apply_cell_styles nueva_fila.extend([ ts, sp_on, sp_off, data_fila.get("O2"), data_fila.get("SO2"), data_fila.get("TEMP"), ]) # Presión con redondeo y None si < -0.5 presion_val = data_fila.get("Presion") if presion_val is not None: try: rounded_press = round(presion_val, 2) nueva_fila.append(rounded_press if rounded_press > -0.5 else None) except Exception: nueva_fila.append(None) else: nueva_fila.append(None) # Alarmas y Voltaje nueva_fila.append(",".join(alarmas) if alarmas else "-") nueva_fila.append(voltaje) # nueva_fila.append("") # horizontal_fill = [""]* (cant_keys_tabla +1 - len(nueva_fila)) # print(f'{horizontal_fill=}') # ===== NUEVO: Recolección para gráficos (no toca el layout) ===== try: if ts_obj is not None and isinstance(data_fila.get("O2"), (int, float)) and not math.isnan(data_fila["O2"]): values_for_charts[channel]["ts"].append(ts_obj) values_for_charts[channel]["O2"].append(float(data_fila["O2"])) so2_val = data_fila.get("SO2") if isinstance(so2_val, (int, float)) and not math.isnan(so2_val): values_for_charts[channel]["SO2"].append(float(so2_val)) else: # Continuidad de trazo: 0.0 si no hay dato numérico values_for_charts[channel]["SO2"].append(0.0) except Exception: pass ws.append(nueva_fila) if null_data or sandwich_case: try: info = f"{date_str}: Total={total_data}" if null_data: info += f" | Nulls={null_data} ({(null_data/total_data)*100:.2f}%)" if sandwich_case: info += f" | Sandwich={sandwich_case} ({(sandwich_case/total_data)*100:.2f}%)" logger.info(info) except Exception: pass # ---- Contexto para estilos legacy + gráficos ---- context = { "worksheet": ws, "date": date_str, "pump_names": bomb_names, "ranges": { b: { "col_start": Rangos[b]["L_Inicial"], "col_end": Rangos[b]["L_Final"], "row_start": v_offset + 3, "graph_col": Rangos[b]["L_Graph"], # NUEVO: ancla de gráfico } for b in bomb_names }, "v_offset": v_offset, "h_offset": h_offset, "columns": base_columns, "control_type": control_type, "rangos_bomba_on": rangos_bomba_on, "rangos_bomba_on_ts": rangos_bomba_on_ts, "rangos_iny_on": rangos_iny_on, "rangos_iny_on_ts": rangos_iny_on_ts, "rangos_iny_on_manual": rangos_iny_on_manual, "rangos_iny_on_manual_ts": rangos_iny_on_manual_ts, # NUEVO: datos para gráficos "chart_data": values_for_charts, "stats": {"total": total_data, "nulls": null_data, "sandwich": sandwich_case}, } return True, context except Exception as e: error_msg = f"Error processing date {current_date}: {e} (line {e.__traceback__.tb_lineno})" logger.error(error_msg) return False, error_msg def _save_workbook( workbook, dates, centro_name, fileobj, time_total, excel_save_path ) : """Save workbook to disk or a file-like object.""" try: if "Sheet" in workbook.sheetnames: workbook.remove(workbook["Sheet"]) first = dates[0].isoformat() last = dates[-1].isoformat() identifier = f"{first}_{last}" if first != last else first if centro_name: identifier = f"{centro_name}_{identifier}" if fileobj is not None: workbook.save(fileobj) if hasattr(fileobj, "seek"): fileobj.seek(0) return True, f"Reporte_{identifier}.xlsx" # Ruta por defecto (ajusta a tu util si tienes get_escritorio_path/save_dir) dest_path = os.path.join( excel_save_path, f"Reporte_{identifier}_.xlsx" ) logger.info(f"Guardado archivo Excel en: {dest_path}...") current_dir = Path(os.path.dirname(os.path.abspath(__file__))) # dest_path = Path.home() / f"Reporte_{identifier}.xlsx" # dest_path = dest_path / f"Reporte_{identifier}_.xlsx" workbook.save(dest_path) if hasattr(workbook, "close"): workbook.close() logger.info(f"Excel saved at: {dest_path}") logger.info(f"Total time: {(time.time() - time_total) * 1e3:.2f} ms") return True, str(dest_path) except Exception as e: return False, f"Error saving workbook: {e} {e.__traceback__.tb_lineno}" # == estilos == def _apply_border_rect(ws, start_row: int, end_row: int, start_col: int, end_col: int, border): """Aplica 'border' a todas las celdas del rectángulo (inclusive).""" for r in range(start_row, end_row + 1): for c in range(start_col, end_col + 1): ws.cell(row=r, column=c).border = border def _style_headers_blockwise(ctx: dict, styles: dict) -> None: """ Deja los encabezados exactamente según el layout: - Merge + valor 'Bomba XXX' en fila v_offset+1 por bloque - Subtítulos (nombres de columnas) estilados en fila v_offset+2 - Bordes en ambas filas por bloque Requiere en ctx: worksheet, pump_names, v_offset, h_offset, columns """ ws = ctx["worksheet"] pump_names = ctx["pump_names"] v_offset = ctx["v_offset"] h_offset = ctx["h_offset"] columns = ctx["columns"] font = styles["font"] # blanco + bold fill = styles["fill"] # naranja border = styles["border"] cant_keys_tabla = len(columns) tablas_dist = cant_keys_tabla + 1 header_row = v_offset + 1 columns_row = v_offset + 2 if not pump_names: return # Título por bloque: merge + valor + estilo + borde for idx, pump in enumerate(pump_names): col_start_idx = idx * tablas_dist + h_offset + 1 col_end_idx = col_start_idx + cant_keys_tabla - 1 cell_title = ws.cell(row=header_row, column=col_start_idx) if cell_title.value in (None, ""): cell_title.value = f"Bomba {pump}" cell_title.font = font cell_title.fill = fill cell_title.alignment = Alignment(horizontal="center", vertical="center") ws.merge_cells( start_row=header_row, start_column=col_start_idx, end_row=header_row, end_column=col_end_idx ) _apply_border_rect(ws, header_row, header_row, col_start_idx, col_end_idx, border) # Subtítulos por bloque: estilo + borde for idx, _ in enumerate(pump_names): col_start_idx = idx * tablas_dist + h_offset + 1 col_end_idx = col_start_idx + cant_keys_tabla - 1 for c in range(col_start_idx, col_end_idx + 1): cell = ws.cell(row=columns_row, column=c) cell.font = font cell.fill = fill cell.alignment = Alignment(horizontal="center", vertical="center") _apply_border_rect(ws, columns_row, columns_row, col_start_idx, col_end_idx, border) def _find_col_index_in_block(columns, logical_name) -> int: wanted = set([logical_name] + _COLUMN_ALIASES.get(logical_name, [])) for i, name in enumerate(columns): if name in wanted: return i raise KeyError(f"Column '{logical_name}' not found in {columns}") def _style_day_modern( ctx: dict, styles: dict, thresholds, watermark_img_path, protect: bool = True, sheet_password: str = "itg123chile", ): try: # Validaciones mínimas for k in ["worksheet","pump_names","ranges","v_offset","h_offset","columns"]: if k not in ctx: return False, f"Missing ctx['{k}']" for sk in ["font","fill","fill_iny","fill_iny_man","fill_vdf","border","fill_alarm"]: if sk not in styles: return False, f"Missing styles['{sk}']" ws = ctx["worksheet"] pump_names = ctx["pump_names"] ranges = ctx["ranges"] v_offset = ctx["v_offset"] h_offset = ctx["h_offset"] columns = ctx["columns"] fill_iny = styles["fill_iny"] fill_iny_man = styles["fill_iny_man"] fill_vdf = styles["fill_vdf"] border = styles["border"] fill_alarm = styles["fill_alarm"] red_font = Font(color="FF0000") th = dict(_DEFAULT_THRESHOLDS) if thresholds: th.update(thresholds) # Freeze panes ws.freeze_panes = ws.cell(row=v_offset + 3, column=1) # (Opcional) Marca de agua if watermark_img_path: try: add_markwater_images(ws, watermark_img_path, start_column="C", start_row=21, repeats=len(pump_names), step=7) except Exception as ex: logger.warning(f"Watermark error on {ws.title}: {ex}") # === Encabezados: una sola función === _style_headers_blockwise(ctx, styles) # Índices relativos dentro del bloque try: idx_time = _find_col_index_in_block(columns, "Time") idx_sp_on = _find_col_index_in_block(columns, "SP_ON") idx_o2 = _find_col_index_in_block(columns, "O2") idx_alarms = _find_col_index_in_block(columns, "Alarms") idx_vo = _find_col_index_in_block(columns, "Vo") except KeyError as e: return False, f"Styling error: {e}" # Datos (bordes + reglas + rellenos por estado) last_row = ws.max_row columns_row = v_offset + 2 for pump in pump_names: col_start_letter = ranges[pump]["col_start"] col_end_letter = ranges[pump]["col_end"] col_start_idx = column_index_from_string(col_start_letter) col_end_idx = column_index_from_string(col_end_letter) data_start_row = ranges[pump]["row_start"] data_end_row = last_row # Bordes para todo el bloque de datos if data_start_row <= data_end_row: _apply_border_rect(ws, data_start_row, data_end_row, col_start_idx, col_end_idx, border) # Conjuntos de tiempos HH:MM:SS bomba_on_times = set(ctx.get("rangos_bomba_on", {}).get(pump, [])) iny_on_times = set(ctx.get("rangos_iny_on", {}).get(pump, [])) iny_manual_times = set(ctx.get("rangos_iny_on_manual", {}).get(pump, [])) for r in range(data_start_row, data_end_row + 1): # fila del bloque (solo columnas del bloque) row_cells = [ws.cell(row=r, column=c) for c in range(col_start_idx, col_end_idx + 1)] # Alineación (borde ya aplicado arriba) for cell in row_cells: cell.alignment = Alignment(horizontal="center", vertical="center", shrink_to_fit=True) # Estados por tiempo time_val = row_cells[idx_time].value bomb_active = (time_val in bomba_on_times) active_inject = (time_val in iny_on_times) and bomb_active manual_inject = (time_val in iny_manual_times) # Prioridad: manual > auto > vdf if manual_inject: for c in row_cells: c.fill = fill_iny_man elif active_inject: for c in row_cells: c.fill = fill_iny elif bomb_active: for c in row_cells: c.fill = fill_vdf # Reglas (solo datos, no headers) if r > columns_row: # O2 < SP_ON → O2 rojo try: sp_on_val = row_cells[idx_sp_on].value o2_val = row_cells[idx_o2].value sp_on_f = float(str(sp_on_val).replace(",", ".")) if sp_on_val is not None else None o2_f = float(str(o2_val).replace(",", ".")) if o2_val is not None else None if sp_on_f is not None and o2_f is not None and o2_f < sp_on_f: row_cells[idx_o2].font = red_font except Exception: pass # Voltaje > umbral → rojo try: vo_val = row_cells[idx_vo].value vo_f = float(str(vo_val).replace(",", ".")) if vo_val is not None else None if vo_f is not None and vo_f > th["voltage_high"]: row_cells[idx_vo].font = red_font except Exception: pass # Alarmas con 'A' → fill_alarm try: alarms_val = row_cells[idx_alarms].value if isinstance(alarms_val, str) and "A" in alarms_val: row_cells[idx_alarms].fill = fill_alarm except Exception: pass # Auto-ancho for col in ws.columns: max_len = 0 letter = col[0].column_letter for cell in col: v = cell.value if v is not None: l = len(str(v)) if l > max_len: max_len = l if max_len: ws.column_dimensions[letter].width = max(4, min(40, max_len + 2)) # Protección if protect: ws.protection.sheet = True ws.protection.objects = True ws.protection.scenarios = True ws.protection.password = sheet_password wb = ws.parent wb.security.workbookPassword = sheet_password wb.security.lockStructure = True return True, "" except Exception as e: return False, f"Styling failed on sheet {ctx.get('date','')}: {e}" def style_all_days_modern( day_contexts, styles: dict, thresholds, watermark_img_path, protect: bool = True, sheet_password: str = "itg123chile", ) : try: for ctx in day_contexts: ok, msg = _style_day_modern( ctx=ctx, styles=styles, thresholds=thresholds, watermark_img_path=watermark_img_path, protect=protect, sheet_password=sheet_password, ) if not ok: return False, msg return True, "" except Exception as e: return False, f"Bulk styling failed: {e}" def insert_matplot_graphs_from_context(ctx: dict) : """ Inserta gráficos por bomba en la hoja usando el context devuelto por _process_day_data. Requiere en ctx: - worksheet, pump_names, ranges{pump:{graph_col}}, v_offset - chart_data[pump]: {"ts":[], "O2":[], "SO2":[]} - rangos_*_ts[pump]: listas de datetime """ try: worksheet = ctx["worksheet"] pump_names = ctx["pump_names"] ranges = ctx["ranges"] v_offset = ctx["v_offset"] values_for_charts = ctx.get("chart_data", {}) rangos_iny_on_ts = ctx.get("rangos_iny_on_ts", {}) rangos_bomba_on_ts = ctx.get("rangos_bomba_on_ts", {}) rangos_iny_on_manual_ts = ctx.get("rangos_iny_on_manual_ts", {}) # estilos matplotlib (como en tu función) plt.rcParams.update({ 'font.size': 22, 'axes.titlesize': 25, 'xtick.labelsize': 20, 'ytick.labelsize': 20, 'legend.fontsize': 25, 'figure.titlesize': 25 }) for pump in pump_names: try: # datos base series = values_for_charts.get(pump, {"ts": [], "O2": [], "SO2": []}) ts_list = series.get("ts", []) o2_list = series.get("O2", []) so2_list = series.get("SO2", []) if not ts_list or not o2_list: # nada que graficar continue x = mdates.date2num(ts_list) o2 = o2_list so2 = so2_list if so2_list else [0] * len(o2) ts_inyecting = [ ts for ts in rangos_iny_on_ts.get(pump, []) if ts in set(rangos_bomba_on_ts.get(pump, [])) ] ts_iny_manual = rangos_iny_on_manual_ts.get(pump, []) # figura fig = plt.figure(figsize=(14, 7.3), dpi=100) left, right, top = 0.10, 0.95, 0.95 # margen inferior dinámico (similar a tu cálculo) total_graphs = 2 + (1 if ts_inyecting else 0) + (1 if ts_iny_manual else 0) bottom_margin = 0.30 # constante funciona bien width = right - left height = top - bottom_margin ax1 = fig.add_axes([left, bottom_margin, width, height]) # eje izquierdo (O2) + barras de estado (sombreado) 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, max(12, math.ceil(max(o2) + 1))) 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))) 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_injm = mdates.date2num(ts_iny_manual) dxm = (np.min(np.diff(np.sort(x_injm))) if len(x_injm) > 1 else 1/(24*60)) ax1.bar(x_injm, [y_max]*len(x_injm), width=dxm, bottom=y_min, color="#92DC97", alpha=0.3, label="Inyección O2 Manual") ax1.set_ylim(y_min, y_max) # eje derecho (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 if any(so2) else 100) ax2.set_ylabel("Saturación %", color="#ab0202", fontsize=16) ax2.tick_params(axis='y', labelcolor="#ab0202") # formato de tiempo en X ax1.xaxis.set_major_locator(mdates.MinuteLocator(interval=60)) ax1.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M')) fig.autofmt_xdate() for lbl in ax1.get_xticklabels(): lbl.set_rotation(45) lbl.set_ha('right') ax1.set_title(f"Sensor {pump}", fontsize=20) ax1.grid(True) fig.subplots_adjust(bottom=bottom_margin) # leyenda combinada 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, ncol=2, loc='upper center', bbox_to_anchor=(0.5, -0.2), bbox_transform=ax1.transAxes) # exportar imagen e insertar en Excel img_data = BytesIO() fig.savefig(img_data, format='png', dpi=100) plt.close(fig) img_data.seek(0) img = ExcelImg(img_data) # img.width, img.height = 480, 292 img.width, img.height = 620, 390 # ancla: usa la columna 'graph_col' del bloque, en la fila 2 (como en tu versión) graph_col_letter = ranges[pump]["graph_col"] anchor = f"{graph_col_letter}2" worksheet.add_image(img, anchor) except Exception as e: logger.error(f"Error generando gráfico para bomba {pump}: {e}") return True, "" except Exception as e: return False, f"Error insertando gráficos: {e}" # ============================================================================= # Ejecución como script # ============================================================================= if __name__ == "__main__": main()