# scapp_modules.Analytic.Scapp_analytics import copy from datetime import datetime, timedelta # from scapp_modules.Logger import data_analytic_logger from .utils import process_one_day_data from .scapp_database import fetch_data_collection, fetch_visitas_collection, data_analytic, cameras_collection # from .scapp_database.utils.analytic_database_utils import save_analytics_data, fetch_analytic_data, get_docs_by_date_and_area, fetch_analytic_area_data from .scapp_database.utils.analytic_database_utils import save_analytics_data, fetch_analytic_data, get_docs_by_date_and_area, fetch_analytic_area_data, get_docs_by_region, fetch_analytic_region_data, _init_value_summary, _init_tasa_summary, _postprocess_values, _postprocess_tasa, validar_y_sumar import time def data_analytic_logger(text): print(text) def compute_daily_analytic_data( scapp_configs:dict, ): """ Calcula y alamacena los datos del dia anterior. Se llama diariamente para siempre tener los datos del dia anterior ya almacenados en BD. """ today = datetime.today() start_day = datetime(today.year, today.month, today.day) yesterday = start_day - timedelta( days=1 ) t1 = datetime.now() _ = compute_multiple_analytic_days( yesterday, day_ranges=0, scapp_configs=scapp_configs, parse_data=False ) t_end = datetime.now() data_analytic_logger(f"Datos de Analitica del dia anterior calculados correctamente en {(t_end-t1).total_seconds()*1e3:.4f}ms.") def compute_multiple_analytic_days( center_id:str, _start_day:datetime, day_ranges: int = 7, scapp_configs:dict = {}, parse_data: bool = True, ) -> dict: """ Calcula los datos de analitica para un conjunto de dias, especificando su dia inicial y el rango de dias solicitado @params - center_id (str): ID de centro, en formato string, correspondiente al identificador unico de cada centro. - _start_day (datetime): Fecha, en formato datetime, correspondiente al dia inicial solicitado. - day_ranges (int): Cantidad de dias solicitados. - parse_data (bool): Flag que dicta si parsear los datos al formato esperado por front. - verbose (bool): Flag para entregar informacion detallada. - timed (bool): Imprime tiempos de ejecucion. @returns - analytic_data (dict): Diccionario agrupado por jaula y por dia con los datos procesados. Cada jaula contiene las llaves: { name, sfr_value, delivered_food, tasa_units, tasa_values, tasa_data, fasts_count, feeding_time_range, alarms_data, alarms_count } - parse_data=True: Cada llave es un array de la forma [{day, value}, ...] - parse_data=False: Cada llave es un diccionario, de la forma {day, value} - Llave 'name': Siempre contiene como valor el nombre de la Jaula. """ analytic_data = {} day_ranges = int(day_ranges) if day_ranges < 1: raise ValueError("La cantidad de días debe ser mayor a cero") # Asegurar que start_day tenga H:M:S 00:00:00 start_day = datetime( _start_day.year, _start_day.month, _start_day.day ) end_day = start_day + timedelta(days=day_ranges) print(f"LDRd {start_day=}") print(f"LDRd {end_day=}") # Leer parametros debug timed = scapp_configs.get( 'Analytic_debug_timed',{} ).get( 'value' , True ) verbose = scapp_configs.get( 'Analytic_debug_verbose',{} ).get( 'value' , False ) # Obtener datos ya calculados de analitica desde la coleccion correspondiente # loaded_data = [] # ! Old # loaded_data = fetch_analytic_data( data_analytic, start_day, end_day ) # days_with_data = identify_days_with_data( loaded_data ) # ! New: fetch_analytic_data ya entrega el diccionario formateado correctamente days_with_data = fetch_analytic_data( data_analytic, center_id, start_day, end_day ) # print(f"{ days_with_data=}") # Datos a guardar en BD docs_array = [] # Obtener el dia de hoy en formato d/m/YYYY _today = datetime.today() today = datetime( _today.year, _today.month, _today.day ) # today_formatted = today.strftime('%d/%m/%Y') t1 = datetime.now() # Obtener nombres de las camaras try: filter_cameras = {'_id': 'Cameras'} #, 'centro_id':center_id db_cam_data = (list( list( cameras_collection.find( filter_cameras )) ))[0]['Cams'] cam_names_dict = {cam['Index']: cam['Nombre'] for cam in db_cam_data} except Exception as e: data_analytic_logger(f"Error leyendo nombres de camaras desde BD ({e.__traceback__.tb_lineno}): {e}") cam_names_dict = {} computed_days = [] # Array de dias calculados loaded_days = [] # Array de dias cargados desde DB saved_days = [] # Array de dias almacenados en DB # Recorrer los dias solicitados for days in range(day_ranges): # Calcular dia inicial y final start_of_day = start_day + timedelta(days=days) end_of_day = start_of_day + timedelta(days=1) # Formatear dia a str formatted_day = start_of_day.strftime('%d/%m/%Y') # Y -> 2025 ; y -> 25 if (start_of_day > today): data_analytic_logger(f"Se han solicitados datos de fecha futura: {formatted_day}, omitiendo calculo.") continue # HASTA AQUi **PROCESA** SERVICE MANAGER # Procesar dias que no estan en base de datos if formatted_day not in days_with_data: one_day_data={} pass """ # # Obtener datos del dia seleccionado desde BD # today_data = fetch_data_collection(start_of_day, end_of_day, verbose=verbose) # visitas_today = fetch_visitas_collection( start_of_day ) # # Obtener los datos del dia # one_day_data = process_one_day_data( today_data, visitas_today, formatted_day, cam_names_dict, scapp_configs, verbose=False ) # computed_days.append( formatted_day ) # # Solo guardar datos de dias completos: anteriores a hoy o si es hoy y ya pasaron las 21:00 # if ((start_of_day < today) or (start_of_day == today and today.hour > 21)) and len(one_day_data) > 0: # # Guardar los datos del dia en BD: # # Al guardar, se crea un documento por cada jaula. # for jaula, jaula_data in one_day_data.items(): # doc = { # "timestamp": start_of_day, # "camera_id": jaula, # Se utiliza el nombre o identificador unico de la jaula # "data": jaula_data # Se guarda directamente el diccionario con los datos # } # docs_array.append(doc) # if formatted_day not in saved_days: # saved_days.append( formatted_day ) # elif not one_day_data: # # Omitir print si es fecha futura # if not (start_of_day > today): # data_analytic_logger(f"Dia {formatted_day} sin datos, no se almacena en BD.") # else: # data_analytic_logger(f"Se ha procesado el dia de hoy sin ser las 22:00 aun, no se guardan estos datos.") """ else: # Se cargaron los datos desde BD one_day_data = days_with_data[ formatted_day ] loaded_days.append( formatted_day ) # data_analytic_logger(f"Datos del {formatted_day} cargados correctamente.") # Añadir los datos del dia al diccionario que contiene todos los dias if parse_data: parse_one_day_data( analytic_data, one_day_data, formatted_day ) info_msg = 'Se han ' if len(loaded_days) > 0: info_msg += f'cargado los datos de {len(loaded_days)} dias ' if len(computed_days): info_msg += f'y procesado los datos de {len(computed_days)} dias: \n\tcargado: {loaded_days}\n\tprocesado: {computed_days}\n' else: info_msg += f"\n cargado: {loaded_days}\n" elif len(computed_days): info_msg += f'procesado los datos de {len(computed_days)} dias: \n\tprocesado: {computed_days}\n' else: info_msg = f"No se han obtenido datos de analitica.\n" data_analytic_logger(info_msg) # Guardar los documentos procesados en una transaccion atomica if docs_array: save_analytics_data( data_analytic, docs_array, saved_days ) t_end = datetime.now() if verbose or timed: computed_info = '' if len( docs_array ) == 0 else f"Se han procesado {len( docs_array )} dias.\n" # loaded_info = '' if len(loaded_data) == 0 else f"Se han cargado {len(loaded_data)} dias.\n" loaded_info = '' if len(days_with_data) == 0 else f"Se han cargado {len(days_with_data)} dias.\n" print(); data_analytic_logger(f"El proceso ha tardado {(t_end-t1).total_seconds()*1e3:.4f} ms.\n{computed_info}{loaded_info}") return analytic_data def parse_one_day_data( analytic_data, one_day_data, formatted_day ): """ Agrega los datos de un dia para cada jaula al diccionario que contiene todos los dias solicitados. @params - analytic_data (dict) : Diccionario por jaula que contiene los datos de multiples dias: { jaula: [{ ... } , ... ] } - one_day_data (dict): Diccionario de los datos del dia { jaula: { sfr_value, delivered_food, tasa_units, tasa_values, tasa_data, fasts_count, feeding_time_range, alarms_data, alarms_count } } - formatted_day (str): Dia formateado (ej. '25/04/2025'). """ for jaula, jaula_data in one_day_data.items(): if jaula not in analytic_data: analytic_data[jaula] = { } # 'nombre': cam_names_dict.get(jaula, jaula ) } for k,v in jaula_data.items(): if k not in analytic_data[jaula]: # Dejar llave 'name' sin ser un array if k == 'name': analytic_data[jaula][k] = v # Las demas llaves se transforman a array else: analytic_data[jaula][k] = [] if k != 'name': # Datos ya formateados en BD : { day, value } # analytic_data[jaula][k].append(v) # Formatear aca, no en BD { day, value } analytic_data[jaula][k].append( {'day': formatted_day , 'value': v} ) return def identify_days_with_data( loaded_data ): """ ### ! Obsoleto: ahora este trabajo lo realiza directamente fetch_analytic_data """ days_with_data = {} for data_day in loaded_data: try: y_m_d_data = data_day.get('metadata') if not y_m_d_data or not '/' in y_m_d_data: continue date_splitted = y_m_d_data.split('/') if len(date_splitted) != 3: continue chilean_date = f"{date_splitted[2]}/{date_splitted[1]}/{date_splitted[0]}" days_with_data[chilean_date] = copy.deepcopy( data_day['data'] ) except Exception as e: #data_analytic_logger(f"Error identificando datos") pass # print(f"dias obtenidos desde BD: {days_with_data.keys()}") return days_with_data #Funciones para Analitica aREA """ #ORIGINAL # old process_analytic_area_data def compute_analytic_area_data(area, start_day, range_days): start_time = time.time() summary = {} start_day = start_day end_day = start_day + timedelta(days=range_days) try: area_data = fetch_analytic_area_data( data_analytic, area, start_day, end_day) print(f"{area_data=}") # area_summary = compute_multiple_analytic_area_days(area_data) area_summary = compute_analytic_summary(area_data, key_funcs=[area_fn, centro_fn],show_area=False) summary[area] = area_summary except Exception as e: data_analytic_logger(f"Error generando analitica de area {area} por error: {e.__traceback__.tb_lineno} : {e}") area_summary = None end_time = time.time() elapsed_time = end_time - start_time data_analytic_logger(f"Tiempo de ejecucion para el area '{area}': {elapsed_time:.2f} segundos") return summary """ def compute_analytic_area_data(area_id: str, start_day: datetime, day_ranges: int) -> dict: """ - area_id: "REGION-AREA" - start_day: datetime de inicio (00:00) - day_ranges: nº de dias a procesar Devuelve: { area_id: { centro_id: { values_summary, tasa_summary, feeding_time_range } } } """ t0 = time.time() # 1) Leer datos crudos por area end_day = start_day + timedelta(days=day_ranges) area_data = fetch_analytic_area_data(data_analytic, area_id, start_day, end_day) # area_data: { "DD/MM/YYYY": { jaula_full_id: jaula_data, … }, … } # 2) Agregar valores y tasas como antes area_summary = compute_analytic_summary( area_data, key_funcs=[area_fn, centro_fn], show_area=False ) # area_summary: { centro_id: { values_summary: {...}, tasa_summary: {...} } } # 3) **Nuevo**: por cada centro y por cada dia, calcular feeding_time_range agregado for centro_id, centro_block in area_summary.items(): # recorremos cada fecha en el summary de valores for fecha, stats in centro_block["values_summary"].items(): # Datos crudos de ese dia raw_jaulas = area_data.get(fecha, {}) # Encontrar el inicio minimo y final maximo min_inicio = {"jaula": None, "hora": None} max_final = {"jaula": None, "hora": None} for jaula_full, jaula_data in raw_jaulas.items(): # solo las jaulas de este centro if not jaula_full.startswith(centro_id + "-"): continue fr = jaula_data.get("feeding_time_range", {}) ini = fr.get("inicio") fin = fr.get("final") if not ini or not fin: continue # convertir a datetime para comparar dt_ini = datetime.fromisoformat(ini) dt_fin = datetime.fromisoformat(fin) if min_inicio["hora"] is None or dt_ini < min_inicio["hora"]: min_inicio["hora"] = dt_ini min_inicio["jaula"] = jaula_full if max_final["hora"] is None or dt_fin > max_final["hora"]: max_final["hora"] = dt_fin max_final["jaula"] = jaula_full # Ajustar el formato ISO + nombre de jaula if min_inicio["jaula"] is not None: stats["feeding_time_range"] = { "inicio": { "jaula": min_inicio["jaula"], "hora": min_inicio["hora"].isoformat() }, "fin": { "jaula": max_final["jaula"], "hora": max_final["hora"].isoformat() } } total_time = time.time() - t0 print(f"[compute_analytic_area_data] Procesado en {total_time:.2f}s") return {area_id: area_summary} ## Versus de area ## def compute_analytic_area_versus_data(collection, area, start_day, range_days): t1 = datetime.now() # fecha_inicial = datetime.strptime(start_day, "%d-%m-%Y") fecha_inicial = start_day fecha_final = fecha_inicial + timedelta(days=range_days) # print(f"{fecha_final=}") try: area_data = fetch_analytic_area_data( collection, area, fecha_inicial, fecha_final) # print(f"{area_data =}") # area_summary = compute_area_summary_by_day(area_data) area_summary = compute_analytic_summary(area_data, key_funcs=[area_fn, centro_fn], show_area = False) except Exception as e: data_analytic_logger(f"Error generando analitica de area {area} por error: {e.__traceback__.tb_lineno} : {e}") area_summary = None t_end = datetime.now() data_analytic_logger(f"Datos de Analitica versus entre areas calculados correctamente en {(t_end-t1).total_seconds()*1e3:.4f}ms.") return area_summary #Funciones para Analitica REGION def compute_analytics_by_region(db, region_id, start_day, day_ranges): t1 = datetime.now() summary = {} # Calcular fecha final sumando num_dias a fecha_inicial end_day = start_day + timedelta(days=day_ranges) collection = db['data_analytic'] try: documents_by_region = get_docs_by_region(collection, region_id, start_day, end_day) computed_analytic_region_data = fetch_analytic_region_data(documents_by_region) # region_analytic_resume = compute_region_summary_by_day(computed_analytic_region_data) region_analytic_resume = compute_analytic_summary(computed_analytic_region_data, key_funcs=[area_fn, centro_fn], show_area = True) except Exception as e: print(f"No se ha generado correctamente a anaĺitica Scapp de la region {region_id} por error en linea {e.__traceback__.tb_lineno}: \n{e}") region_analytic_resume = {} summary = {"analytic_data": region_analytic_resume, "name": region_id} t_end = datetime.now() data_analytic_logger(f"Datos de Analitica de Region {region_id} calculados correctamente en {(t_end-t1).total_seconds()*1e3:.4f}ms.") return summary ## Funcion Auxiliar para unificacion de codigo ## def compute_analytic_summary(area_data, key_funcs, show_area=True): """ key_funcs: [ # por cada jaula, devuelve la ruta de claves donde acumular: # e.g. ["centro_id", jaula_id] o [] para solo por fecha """ final = {} for fecha, jaulas in area_data.items(): for jaula, datos in jaulas.items(): # extraer partes solo si hay key_funcs que lo necesiten partes = jaula.split("-") # construir ruta de keys ruta = [] for fn in key_funcs: # print(f"{fn =}") ruta.append(fn(partes, jaula)) # print(f"{ruta =}") # localizar/crear nodo en final: node = final # if per_date: # # primer nivel: fecha # node = node.setdefault(fecha, { # "values_summary": _init_value_summary(), # "tasa_summary": _init_tasa_summary() # }) # vs = node["values_summary"] # ts = node["tasa_summary"] # else: # ruta de claves + fecha for k in ruta: if show_area: node = node.setdefault(k, {}) else: if k.count('-') > 1: node = node.setdefault(k, {}) # print(f"{k =}, {len(k)}") vs_dict = node.setdefault("values_summary", {}) ts_dict = node.setdefault("tasa_summary", {}) vs = vs_dict.setdefault(fecha, _init_value_summary()) ts = ts_dict.setdefault(fecha, _init_tasa_summary()) # Acumular valores_summary for key, val in datos.get("values_summary", {}).items(): if key in vs: vs[key].append(val) # Acumular alarms_by_tasa tasas_niv = datos.get("alarms_by_tasa_levels") or {} alar_niv = datos.get("alarms_by_tasa") or {} for nivel in ("alta", "media", "baja"): # sumo siempre un int ts[nivel]["alarm"] += alar_niv.get(nivel, {}).get("value", 0) # actualizo las tasas: primero miro en alar_niv[nivel]["tasas"], sino en tasas_niv ts[nivel]["tasas"].update( alar_niv.get(nivel, {}).get("tasas", []) or tasas_niv.get(nivel, []) ) # for nivel in ("alta", "media", "baja"): # ts[nivel]["alarm"] += alar_niv.get(nivel, 0) # ts[nivel]["tasas"].update(tasas_niv.get(nivel, [])) # Post-procesar todo def recurse(node): if not isinstance(node, dict): return if "values_summary" in node and "tasa_summary" in node: # es un centro/fecha o dia if isinstance(node["values_summary"], dict) and all(isinstance(v, list) for v in node["values_summary"].values()): _postprocess_values(node["values_summary"]) _postprocess_tasa(node["tasa_summary"]) else: # caso anidado: values_summary y tasa_summary son dict de fechas for fecha2, vs in node["values_summary"].items(): _postprocess_values(vs) for fecha2, ts in node["tasa_summary"].items(): _postprocess_tasa(ts) else: for sub in node.values(): recurse(sub) recurse(final) return final def centro_fn(parts, jaula): return "-".join(parts[:3]) def jaula_fn(parts, jaula): return "-".join(parts[3:]) if len(parts)>=4 else None def area_fn(parts, jaula): return "-".join(parts[:2]) #deprecated def compute_multiple_analytic_area_days(area_data): final_results = {} for fecha, jaulas in area_data.items(): for jaula, datos in jaulas.items(): partes = jaula.split("-") if len(partes) < 4: continue centro_id = "-".join(partes[:3]) jaula_id = "-".join(partes[3:]) centro = final_results.setdefault(centro_id, {}) vs_por_fecha = centro.setdefault("values_summary", {}) ts_por_fecha = centro.setdefault("tasa_summary", {}) resumen = vs_por_fecha.setdefault(fecha, { "alarms_by_ton": [], "alarms_count": [], "alarms_threshold": [], "delivered_food": [], "fasts_count": [], "perdidos_count": [], "sfr_value": [], "tasa_units": [], "visitas_count": [] }) tasa_fecha = ts_por_fecha.setdefault(fecha, {}) tasa_jaula = tasa_fecha.setdefault(jaula_id, {}) # Agregar valores de resumen for key, val in datos.get("values_summary", {}).items(): if key in resumen: resumen[key].append(val) # Agregar alarmas por nivel con estructura {"alarm": int, "tasas": list} tasas_por_nivel = datos.get("alarms_by_tasa_levels") or {} alarmas_por_nivel = datos.get("alarms_by_tasa") or {} for nivel in ["alta", "media", "baja"]: tasas = tasas_por_nivel.get(nivel, []) alarm = alarmas_por_nivel.get(nivel, 0) registro = tasa_jaula.setdefault(nivel, {"alarm": 0, "tasas": set()}) # Solo actualizamos si tasas tiene datos if tasas: registro["tasas"].update(tasas) registro["alarm"] = alarm # Esto se actualiza incluso si tasas esta vacio # Postprocesamiento for centro in final_results.values(): for fecha, resumen in centro["values_summary"].items(): for key, valores in resumen.items(): if key in ["sfr_value", "alarms_threshold"]: resumen[key] = round(sum(valores) / len(valores),2) if valores else 0 elif key == 'tasa_units': resumen[key] = valores[0] if valores else None else: resumen[key] = validar_y_sumar(valores) for fecha, jaulas in centro["tasa_summary"].items(): for jaula_id, niveles in jaulas.items(): for nivel in niveles: niveles[nivel]["tasas"] = list(niveles[nivel]["tasas"]) return final_results #deprecated def compute_area_summary_by_day(area_data): final_results = {} for fecha, jaulas in area_data.items(): dia_resumen = final_results.setdefault(fecha, { "values_summary": { "alarms_by_ton": [], "alarms_count": [], "alarms_threshold": [], "delivered_food": [], "fasts_count": [], "perdidos_count": [], "sfr_value": [], "tasa_units": [], "visitas_count": [] }, "tasa_summary": { "alta": {"alarm": 0, "tasas": set()}, "media": {"alarm": 0, "tasas": set()}, "baja": {"alarm": 0, "tasas": set()} } }) resumen = dia_resumen["values_summary"] tasa_summary = dia_resumen["tasa_summary"] for jaula, datos in jaulas.items(): # Acumular valores del resumen for key, val in datos.get("values_summary", {}).items(): if key in resumen: resumen[key].append(val) # Acumular tasas y alarmas por nivel tasas_por_nivel = datos.get("alarms_by_tasa_levels") or {} alarmas_por_nivel = datos.get("alarms_by_tasa") or {} for nivel in ["alta", "media", "baja"]: tasas = tasas_por_nivel.get(nivel, []) alarm = alarmas_por_nivel.get(nivel, 0) tasa_summary[nivel]["alarm"] += alarm tasa_summary[nivel]["tasas"].update(tasas) # Postprocesamiento for fecha, datos in final_results.items(): resumen = datos["values_summary"] for key, valores in resumen.items(): if key in ["sfr_value", "alarms_threshold"]: resumen[key] = round(sum(valores) / len(valores),2) if valores else 0 elif key == 'tasa_units': resumen[key] = valores[0] if valores else None else: resumen[key] = validar_y_sumar(valores) for nivel in ["alta", "media", "baja"]: datos["tasa_summary"][nivel]["tasas"] = list(datos["tasa_summary"][nivel]["tasas"]) return final_results """ Ejemplos de uso # ? 1 - compute_multiple_analytic_days import pprint import sys from datetime import datetime, timedelta from scapp_database import fetch_data_collection from scapp_database import fetch_data_collection, fetch_visitas_collection from scapp_modules.Analytic.Scapp_analytics import compute_multiple_analytic_days, compute_daily_analytic_data day_range = 100 today = datetime.now() - timedelta(days=day_range) #today = datetime(2025,4,2) - timedelta(days=7) # datetime(2025,3,5) start_day = datetime(today.year, today.month, today.day) t1 = datetime.now() analytic_data = compute_multiple_analytic_days( start_day, day_ranges=day_range ) t_end = datetime.now() pp = pprint.PrettyPrinter(indent=2, width=100, compact=True) formatted_data = pp.pformat( analytic_data ) print(f"{formatted_data}") print(f"Obtenidos Peso de datos del dia: {sys.getsizeof( analytic_data )} bytes demorando {(t_end - t1).total_seconds()*1e3} milisegundos") # Guardar import json with open( 'analytic_data.json', 'w' , encoding='utf-8' ) as file: json.dump( analytic_data, file, indent=4 ) """