# OBSOLETO: la nube espeja `data_analytic` calculada por Scapp y no ejecuta este pipeline. # No sincronizar contratos nuevos aquí; el productor vivo está en SW_SCAPP. import datetime import statistics # from scapp_modules.Logger import data_analytic_logger def data_analytic_logger(text): print(text) def process_one_day_data( day_data, day_visits, formatted_day, cam_names_dict, scapp_configs, verbose=True ): """ ### ? Recibe los datos de un dia completo de analisis y los formatea para mostrar en el módulo de analítica @params - day_data (dict): Diccionario con datos del día por jaula, de la forma { cam_index: [data...], ... } Cada 'data' es de la forma: [ { timestamp: 2024-06-27T12:03:26.000+00:00, camera_id: "1-A", Detecciones: { perdidos: 0, detectados: 0, nivel_actividad: 26.04, umbral_alarma: 3, perdidos_propios: 0, }, Alimentacion: { use_feeding_api: false, detection_active: false, Comida_entregada: 0, Meta_dia: 0, Porc_entregado: 0, Tasa: 0, Dia: 0, Feeding_active: false, Unidad_Medida: "kg/min", Metodo_alimentacion: "Saciedad", feeding_api_type: "AkvaControl", }, Posicion: { x_pos: "-", y_pos: "-", winche_connected: false, }, Sensores: { temp: "-" } } , ... ] @return - analytic_data (dict): Diccionario de los datos del día por jaula { jaula: { sfr_value, delivered_food, tasa_units, tasa_values, tasa_data, fasts_count, feeding_time_range, alarms_data, alarms_count } } Cada llave es un diccionario, con el formato: { 'day' : 'dd/mm/yyyy , 'value': ... } """ analytic_data = {} # Diccionario por jaula con los datos preparados para analitica have_feeding_data = False # Rangos horarios para obtener datos de alimentación str_start_activity = f"{formatted_day} 07:30:00" str_end_activity = f"{formatted_day} 22:30:00" formato = "%d/%m/%Y %H:%M:%S" start_activity = datetime.datetime.strptime(str_start_activity, formato) end_activity = datetime.datetime.strptime(str_end_activity, formato) seconds_to_save_activity = 899 # Un dato cada 15 min, considerando que datos suelen tener ms > 0 alarm_gap = scapp_configs.get( 'Analytic_alarm_gap_time', {} ).get( 'value' , 20 ) for cam_idx, cam_data in day_data.items(): try: cam_feeding_data = {} # { sfr_value, delivered_food, tasa_units, tasa_values } cam_detection_data = [] # [ {start, end, duration, detectados, nivel_actividad, tasa_value }, ... ] tasa_data = [] # [ { time, value}, ... ] -> Para gráfico last_detect_threshold = None total_perdidos = 0 # Datos de actividad activity_data = [] # [ { time, value}, ... ] -> Para gráfico activity_buffer = [] # Buffer para activar detecciones de a lo más 10min last_saved_activity = datetime.datetime.strptime(str_start_activity, formato) for document in cam_data: detection_doc = document['Detecciones'] feeding_doc = document['Alimentacion'] have_feeding_data = feeding_doc.get('use_feeding_api', False) timestamp = document['timestamp'] dt_timestamp = datetime.datetime.fromisoformat( timestamp.split('+')[0] ) last_detect_threshold = detection_doc['umbral_alarma'] # Determinar si hubo alarma y añadir el correspondiente evento a cam_detection_data last_detect_threshold, added_detection_data = _verify_alarm_event( cam_detection_data, detection_doc, timestamp, last_detect_threshold, alarm_gap ) # Actualizar perdidos perdidos = detection_doc.get('perdidos', 0) total_perdidos = max( perdidos, total_perdidos ) valid_feed_hour = (start_activity <= dt_timestamp <= end_activity) or feeding_doc.get('Feeding_active', False) # Acumular Actividad if valid_feed_hour: activity_buffer.append( detection_doc['nivel_actividad'] ) # Almacenar un dato cada 10min if (dt_timestamp - last_saved_activity).seconds >= seconds_to_save_activity: if len(activity_buffer) > 0: last_saved_activity = dt_timestamp activity_data.append( {'time': last_saved_activity.isoformat() , 'value': statistics.mean( activity_buffer ) } ) activity_buffer = [] # if not have_feeding_data: # have_feeding_data = feeding_doc.get('use_feeding_api', False) # Si no existen datos de alimentación simplemente continuar al siguiente dato # if not have_feeding_data: continue # Actualizar datos de Alimentación _update_feeding_data( tasa_data, cam_feeding_data, feeding_doc, timestamp, cam_detection_data, added_detection_data, valid_feed_hour, have_feeding_data ) # Corregir tasa_data (asegurar que su último valor sea 0) _verify_tasa_data( tasa_data ) # Procesar ayuno fed_metrics, quantized_alarms = feeding_metrics( day_visits.get(cam_idx, []), cam_feeding_data, cam_detection_data, scapp_configs ) # Actualizar analytic_data """ # Estructura anterior _analytic_data = { # sfr_value, delivered_food, tasa_units, tasa_values, tasa_data, fasts_count, feeding_time_range, alarms_data, alarms_count **cam_feeding_data, # { sfr_value, delivered_food, tasa_units, tasa_values } # **{ k: [{'day': formatted_day, 'value': v}] for k, v in cam_feeding_data.items() }, 'tasa_data' : tasa_data, 'fasts_count' : fed_metrics['today_fast'], # float 'feeding_time_range': fed_metrics['feeding_time_range'], 'alarms_data' : cam_detection_data, # [ {start, end, duration, detectados, nivel_actividad, tasa_value }, ... ] 'alarms_by_tasa' : fed_metrics['alarms_by_tasa'], # { baja, media, alta } 'alarms_by_ton' : fed_metrics['alarms_by_ton'], # float 'alarms_count' : fed_metrics['alarms_count'], # int } """ # Estructura nueva -> {values_summary tasa_values, tasa_data, feeding_time_range, alarms_data } _values_summary = { # { alarms_by_ton, alarms_count, fasts_count, delivered_food, sfr_value, tasa_units, } 'alarms_by_ton' : fed_metrics['alarms_by_ton'], # float 'alarms_count' : fed_metrics['alarms_count'], # int 'alarms_threshold' : last_detect_threshold, # int 'fasts_count' : fed_metrics['today_fast'], # float # { delivered_food, sfr_value, tasa_units } **{ k: v for k, v in cam_feeding_data.items() if k != 'tasa_values'}, 'perdidos_cout' : total_perdidos, # int 'visitas_cout' : len(day_visits.get(cam_idx, [])), # int } _analytic_data = { 'activity_data' : activity_data, # datos grafico actividad 'alarms_data' : cam_detection_data, # [ {start, end, duration, detectados, nivel_actividad, tasa_value }, ... ] 'feeding_time_range' : fed_metrics['feeding_time_range'], 'alarms_by_tasa' : fed_metrics['alarms_by_tasa'], # { baja, media, alta } 'alarms_by_tasa_levels' : quantized_alarms, # { baja, media, alta } 'tasa_data' : tasa_data, # datos grafico tasa **{ k: v for k, v in cam_feeding_data.items() if k == 'tasa_values'}, # tasa_values # 'values_summary' : _values_summary, # <- Llaves desordenadas 'values_summary' : { k: _values_summary[k] for k in sorted( _values_summary.keys() ) }, } analytic_data[ cam_idx ] = { # Formatear para BD: { day, value } # **{ k: {'day': formatted_day, 'value': v} for k, v in _analytic_data.items() }, # No formatear para BD, datos mas limpios en BD # **_analytic_data, # <- Llaves desordenadas 'name': cam_names_dict.get( cam_idx, cam_idx ) , **{ k: _analytic_data[k] for k in sorted( _analytic_data.keys() ) }, } except Exception as e: data_analytic_logger(f"Error calculando {cam_idx} del {formatted_day} ({e.__traceback__.tb_lineno}): {e} ") # analyzed_day = f"{day_datetime.day}/{day_datetime.month}/{day_datetime.year}" if verbose: data_analytic_logger(f"Se ha computado correctamente la Analítica del {formatted_day} ") return analytic_data # Datos de detecciones def _verify_alarm_event( cam_detection_data, detection_doc, timestamp, last_detect_threshold, alarm_gap ): """ ### ? Verifica si hubo alarma durante el instante entregado, si es así, añade el evento a cam_detection_data @params - cam_detection_data ([dict]) : Array de diccionario con eventos de alarmas. - detection_doc (dict) : Datos de detecciones correspondiente al instante analizado (con resolución máxima de 2s). - timestamp (datetime) : Timestamp como objeto datetime. - last_detect_threshold (int) : Último valor de treshold en BD, se actualiza en cada iteración. - added_detection_data (bool) : Flag que indica si se añadió un evento de alarma. - alarm_gap (int) : Tiempo, en segundos, para considerar 2 eventos como independientes. """ added_detection_data = False last_detect_threshold = detection_doc['umbral_alarma'] # Determinar si hay alarma activa if detection_doc['detectados'] >= last_detect_threshold: added_detection_data = True # data_analytic_logger(f"Creado evento de alarma\n") # Verificar que sea un evento independiente if len(cam_detection_data) == 0: _append_new_alarm_event( cam_detection_data, detection_doc, timestamp ) # data_analytic_logger(f"Creado nuevo evento de alarma\n") else: last_timestamp = cam_detection_data[-1]['end'] if _verify_independent_alarm_event( last_timestamp, timestamp, alarm_gap ): _append_new_alarm_event( cam_detection_data, detection_doc, timestamp ) else: _update_alarm_event( cam_detection_data, timestamp, detection_doc['detectados'], detection_doc['nivel_actividad'] ) return last_detect_threshold, added_detection_data def _verify_independent_alarm_event( last_timestamp, timestamp, alarm_gap ): """ ### ? Verifica si el evento de alarma se considera un evento independiente, o debe actualizarse el evento anterior. @params - last_timestamp (datetime) : Timestamp anterior, como objeto datetime. - timestamp (datetime) : Timestamp, como objeto datetime. - alarm_gap (int) : Tiempo, en segundos, para considerar 2 eventos como independientes. """ ts = datetime.datetime.fromisoformat(timestamp) last_ts = datetime.datetime.fromisoformat(last_timestamp) # print(f"Comparando tiempos de alarma: {timestamp} - {last_timestamp} = {(ts - last_ts).seconds} -> {( ts - last_ts ).seconds >= MIN_SECS_BETWEEN_ALARM_EVENTS}") return ( ts - last_ts ).seconds >= alarm_gap def _append_new_alarm_event( cam_detection_data, detection_doc, timestamp ): """ ### ? Añade un nuevo evento de detección @params - cam_detection_data ([dict]) : Array de diccionario con eventos de alarmas. - detection_doc (dict) : Datos de detecciones correspondiente al instante analizado (con resolución máxima de 2s). - timestamp (datetime) : Timestamp como objeto datetime. """ cam_detection_data.append({ 'start' : timestamp, 'end' : timestamp, 'duration' : 2, 'detectados' : detection_doc['detectados'], 'nivel_actividad' : detection_doc['nivel_actividad'], 'tasa_value' : 0, }) return def _update_alarm_event( cam_detection_data, timestamp, detectados, nivel_actividad ): """ ### ? Actualiza un nuevo evento de detección: - Modifica timestamp de término - Modifica duración del evento - Almacena como nivel de actividad el máximo entre los eventos. - Almacena como detectados el máximo entre los eventos. @params - cam_detection_data ([dict]) : Array de diccionario con eventos de alarmas. - timestamp (datetime) : Timestamp como objeto datetime. - detectados (int) : Valor de detecciones actual. - nivel_actividad (int) : Valor de nivel de actividad actual. """ start_time = cam_detection_data[-1]['start'] last_activity = cam_detection_data[-1]['nivel_actividad'] last_detectados = cam_detection_data[-1]['detectados'] start_ts = datetime.datetime.fromisoformat(start_time) actual_ts = datetime.datetime.fromisoformat(timestamp) cam_detection_data[-1].update( { 'end' : timestamp, 'duration' : (actual_ts - start_ts).seconds, 'nivel_actividad' : max(last_activity , nivel_actividad), 'detectados' : max(last_detectados , detectados), # No modificar tasa_value }) # data_analytic_logger(f"Actualizado cam_detection_data: \n{cam_detection_data}") return # Datos de Alimentación def _update_feeding_data( tasa_data, cam_feeding_data, feeding_doc, timestamp, cam_detection_data, added_detection_data, valid_feed_hour, have_feeding_data ): """ ### ? Actualiza los datos de alimentación: - Añade o modifica el timestamp de Tasa para el gráfico - Actualiza dato de tasa en evento de alarma, en caso de haber evento de alarma @params - tasa_data ([dict]) : Datos de tasa para gráfico, cada elemento contiene value y timestamp. - cam_feeding_data (dict) : Almacena los datos de alimentacion: { sfr_value, delivered_food, tasa_units, tasa_values } sfr_value (float) : Valor de sft del día, se actualiza en cada iteración, corresponde a un porcentaje, que puede y suele superar 100% delivered_food (float) : Valor de comida entregada, en kg tasa_units (str) : Unidades de medida de la tasa (kg/min , pellet/pez/minuto, ppv, etc) tasa_values ([float]): Valores de tasa de alimentación de la jornada. - feeding_doc (dict) : Datos de alimentación correspondiente al instante analizado (con resolución máxima de 2s). - timestamp (datetime) : Timestamp como objeto datetime. - cam_detection_data ([dict]) : Array de diccionario con eventos de alarmas. - added_detection_data (bool) : Flag que indica si se añadió un evento de alarma. - valid_feed_hour (bool) : Flag que indica si se está en un rango aceptado de horas """ # Añadir datos de Tasa actual_tasa = feeding_doc.get('Tasa') # feeding_doc['Tasa'] if have_feeding_data else 0 if actual_tasa is None: actual_tasa = 0 tasa_values = cam_feeding_data.get( 'tasa_values' , [] ) if actual_tasa not in tasa_values: tasa_values.append( actual_tasa ) # Solo actualizar la alimentación en horarios válidos, o cuando hay tasa de alimentación if (valid_feed_hour or actual_tasa > 0) and have_feeding_data: # * Actualizar datos del gráfico # Se debe almacenar los cambios de tasa, es decir, para cada tasa dejar punto inicial y final if len(tasa_data) == 0: tasa_data.append( {'value': actual_tasa , 'time': timestamp } ) else: last_graph_tasa = tasa_data[-1]['value'] """ # Si hubo cambio en la tasa, añadir el dato y un dato que marque el final del anterior if last_graph_tasa != actual_tasa: # Añadir dato que marca fin del valor anterior last_tasa = tasa_data[-1]['value'] if type(timestamp) == str: ts = datetime.datetime.fromisoformat(timestamp) else: ts = timestamp timestamp_un_segundo_antes = (ts - datetime.timedelta(seconds=1)).isoformat() tasa_data.append( {'value': last_tasa , 'time': timestamp_un_segundo_antes } ) # Añadir dato con nuevo valor tasa_data.append( {'value': actual_tasa , 'time': timestamp } ) """ if len( tasa_data ) < 2: tasa_data.append( {'value': actual_tasa , 'time': timestamp } ) # Si hubo cambio en la tasa, añadir el dato y un dato que marque el final del anterior elif last_graph_tasa != actual_tasa: # Añadir dato con nuevo valor tasa_data.append( {'value': actual_tasa , 'time': timestamp } ) # Hay más de dos datos, sin haber cambio en tasa -> asegurar dos datos iguales con la misma tasa (inicio y fin) else: # Si los dos últimos datos tienen la misma tasa, actualizar timestamp del último if tasa_data[-1]['value'] == tasa_data[-2]['value']: tasa_data[-1]['time'] = timestamp # Si no hay dos datos con la misma tasa, añadir el 2do que marca el fin de esta tasa else: tasa_data.append( {'value': actual_tasa , 'time': timestamp } ) # * Actualizar datos de tasa en la alarma if added_detection_data: # data_analytic_logger(f"Dato de tasa en alarma actualizado\n") last_det_tasa = cam_detection_data[-1].get('tasa_value', 0) cam_detection_data[-1].update( { 'tasa_value': max( last_det_tasa, actual_tasa ) } ) cam_feeding_data.update({ 'delivered_food' : feeding_doc.get( 'Comida_entregada', 0 ), 'sfr_value' : feeding_doc.get( 'Porc_entregado', 0 ), 'tasa_units' : feeding_doc.get( 'Unidad_Medida', 'kg/min' ), 'tasa_values' : tasa_values, }) return def _verify_tasa_data( tasa_data ): """ ### ? Asegura que la tasa siempre termine en 0, en caso de tener datos @params - tasa_data ([dict]) : Datos de tasa para gráfico, cada elemento contiene value y timestamp. """ if len( tasa_data ) > 0: last_value = tasa_data[-1]['value'] if last_value != 0: last_timestamp = tasa_data[-1]['time'] # Convertir timestamp a objeto datetime if type(last_timestamp) == str: ts = datetime.datetime.fromisoformat(last_timestamp) else: ts = last_timestamp one_second_later = (ts + datetime.timedelta(seconds=1)).isoformat() tasa_data.append( {'value': 0 , 'time': one_second_later } ) return def feeding_metrics( day_visits, cam_feeding_data, cam_detection_data, scapp_configs ): """ ### ? Calcula todas las métricas adicionales de alimentación @params: - day_visits [dict] : Array que contiene diccionarios de visitas, cada uno de la forma: day: "2024-10-17", camera_id: "13-A", Visitas: [ duracion: 75.04, racion: 0, inicio: 2024-10-17T17:23:05.000+00:00, -> datetime final: 2024-10-17T17:24:20.000+00:00 ] - cam_feeding_data (dict) : { sfr_value, delivered_food, tasa_units, tasa_values } - cam_detection_data ([dict]): [ {start, end, duration, detectados, nivel_actividad, tasa_value }, ... ] @return - today_fast (float) : Ayuno del día [0, 0.5, 1.0] - feeding_time_range {str(isoformat)} : Diccionario de timestamp de alimentación: {inicio, final} """ today_fast, feeding_time_range = _calculate_fasts_and_time_ranges( day_visits ) alarms_by_tasa, quantized_alarms = _calculate_alarms_by_feeding_rate( cam_feeding_data, cam_detection_data, scapp_configs ) # Calcular alarmas por tonelada alarms_count = len( cam_detection_data ) delivered_food = cam_feeding_data.get('delivered_food' , 0 ) alarms_by_ton = (alarms_count / delivered_food)*1e3 if delivered_food > 0 else 0 feeding_metrics = { 'today_fast': today_fast, 'feeding_time_range': feeding_time_range, 'alarms_count': alarms_count, 'alarms_by_ton': alarms_by_ton, 'alarms_by_tasa': alarms_by_tasa, } return feeding_metrics, quantized_alarms def _calculate_fasts_and_time_ranges( day_visits ): """ ### ? Calcula y retorna el ayuno del día @return - today_fast (float) : Ayuno del día [0, 0.5, 1.0] - feeding_time_range (dict) : { inicio: str(isoformat), final: str(isoformat) } """ ayuno = 1.0 feeding_time_range = {} # Si no hubieron visitas: día completo de ayuno if len(day_visits) > 0: # Obtener jornadas de mañana y tarde breakfast = day_visits[0] dinner = day_visits[-1] # Revisar jornada de mañana try: breakfast_timestamp = breakfast['inicio'] if isinstance( breakfast_timestamp, str ): breakfast_timestamp = datetime.datetime.fromisoformat(breakfast_timestamp) breakfast_time = breakfast_timestamp.hour if breakfast_time < 12: ayuno -= .5 feeding_time_range['inicio'] = breakfast_timestamp.isoformat() except Exception as e: data_analytic_logger(f"Error calculando ayuno de mañana ({e.__traceback__.tb_lineno}): {e}") # Revisar jornada de tarde try: dinner_timestamp = dinner['final'] if isinstance( dinner_timestamp, str ): dinner_timestamp = datetime.datetime.fromisoformat(dinner_timestamp) dinner_time = dinner_timestamp.hour if dinner_time > 12: ayuno -= .5 feeding_time_range['final'] = dinner_timestamp.isoformat() except Exception as e: data_analytic_logger(f"Error calculando ayuno de tarde ({e.__traceback__.tb_lineno}): {e}") return ayuno, feeding_time_range def _calculate_alarms_by_feeding_rate( cam_feeding_data, cam_detection_data, scapp_configs ): """ ### ? Calcula la cantidad de alarmas por tasa, separadas en 3 niveles: Alarmas a tasa [baja, media, alta] @params: - cam_feeding_data (dict) : { sfr_value, delivered_food, tasa_units, tasa_values } - cam_detection_data ([dict]): [ {start, end, duration, detectados, nivel_actividad, tasa_value }, ... ] @return - alarms_by_tasa (dict) : Cantidad de alarma por tasas cuantizadas: {baja, media, alta} """ alarms_by_tasa = { 'baja': 0, 'media': 0, 'alta': 0 } # Posicionar cada tasa de alimentación en niveles cuantizados quantized_alarms = _discretize_feeding_rates( cam_feeding_data, scapp_configs ) # Recorrer los eventos de detección for detect_event in cam_detection_data: # Obtener la tasa de alimentación asociada al evento alarma event_tasa = detect_event.get('tasa_value') if event_tasa is None: data_analytic_logger(f"Warning: Tasa no encontrada en evento de detección, seteando a 0: \n{detect_event} ") event_tasa = 0 # Definir donde se posiciona la alarma, según la tasa if event_tasa in quantized_alarms['alta']: alarms_by_tasa['alta'] += 1 elif event_tasa in quantized_alarms['media']: alarms_by_tasa['media'] += 1 else: alarms_by_tasa['baja'] += 1 return alarms_by_tasa, quantized_alarms def _discretize_feeding_rates( cam_feeding_data, scapp_configs ): """ ### ? Cuantizar las tasas de alimentación del día en 3 niveles: {baja, media, alta} Si existen más de 3 tasas de alimentación, los niveles se definen como: - baja: [ min_value , mean - std/2 ] - media: ] mean - std/2, mean + std/2 [ - alta: [ mean + std/2, max_value ] @return - quantized_alarms (dict): Diccionario con tasas cuantizadas: {baja: [], media: [], alta: [] } """ debug = False # Posicionar cada tasa de alimentación en niveles cuantizados quantized_alarms = { 'baja': [ 0, ], 'media': [], 'alta': [] } try: # debug and print(f"Discretizando valores de tasa: {cam_feeding_data.get('tasa_values')}") if 'tasa_values' not in cam_feeding_data: return quantized_alarms _tasa_values = cam_feeding_data['tasa_values'] tasa_values = [ t for t in _tasa_values if t > 0] tasa_values_len = len( tasa_values ) if tasa_values_len == 0 : return if tasa_values_len <= 3: if tasa_values_len == 2: # 3 valores sin tomar en cuenta 0, uno a cada tramo tasa_values.sort() min_tasa, mid_tasa, max_tasa = tasa_values[0], tasa_values[1], tasa_values[2] if min_tasa not in quantized_alarms['baja'] : quantized_alarms['baja'].append( min_tasa ) quantized_alarms['media'].append( mid_tasa ) quantized_alarms['alta'].append( max_tasa ) elif tasa_values_len == 2: # 2 valores sin tomar en cuenta 0, uno a alto y uno a medio min_tasa, max_tasa = min(tasa_values), max(tasa_values) quantized_alarms['media'].append( min_tasa ) quantized_alarms['alta'].append( max_tasa ) elif tasa_values_len == 1: # Un valor sin tomar en cuenta 0, se manda a alta tasa = tasa_values[0] quantized_alarms['alta'].append( tasa ) else: baja_array = [ 0 ] media_array = [] alta_array = [] # Corroborar que no haya un valor extremadamente grande que ensucie los demás datos _tasa_values = [ t for t in tasa_values if t < max(tasa_values) ] if max(tasa_values) > 10 * max(_tasa_values): data_analytic_logger(f"Warning: Tasa excesivamente grande detectada: {max(tasa_values)}") # Hay un máximo que ensucia los valores -> Añadir a alta y calcular con los demás alta_array.append( max(tasa_values) ) tasa_values = _tasa_values config_method = scapp_configs.get( 'Analytic_discr_method',{} ).get( 'value' , 'wide_medium' ) # ['equal_ranges', 'wide_medium', 'normal'] rango_bajo, rango_alto = compute_discrete_ranges( tasa_values, config_method ) # debug and print(f"Discretizando valores de tasa: {cam_feeding_data.get('tasa_values')}") debug and tasa_values.sort() ; print(f"Rango de tasas, hay más de 3 tasas: {tasa_values} -> {(round(rango_bajo,1), round(rango_alto,1))}") # Añadir cada valor de tasa a su grupo correspondiente for tasa_val in tasa_values: # Menor a rango_bajo -> añadir a baja if tasa_val <= rango_bajo: if tasa_val not in baja_array: baja_array.append( tasa_val ) # Mayor a rango_alto -> añadir a alta elif tasa_val >= rango_alto: if tasa_val not in alta_array: alta_array.append( tasa_val ) # Valor entre rangos else: if tasa_val not in media_array: media_array.append( tasa_val ) if len(baja_array) > 0 : baja_array.sort() if len(media_array) > 0: media_array.sort() if len(alta_array) > 0 : alta_array.sort() quantized_alarms = { 'baja': baja_array, 'media': media_array, 'alta': alta_array } if len(quantized_alarms['alta']) > 0: debug and print(f"\t{quantized_alarms=}\n") except Exception as e: data_analytic_logger(f"Error discretizando rangos de Tasa de Alimentación ({e.__traceback__.tb_lineno}): {e} ") return quantized_alarms # * Calcular rangos de los niveles bajo, medio y alto def compute_discrete_ranges( _tasa_values, method = 2 ): """ Calcula los cortes para dividir las tasas en 3 - Methods : ['equal_ranges', 'wide_medium', 'normal'] """ tasa_values = [ t for t in _tasa_values if t != 0 ] if method == 'normal': rangos = _asumming_normal_dist( tasa_values ) elif method == 'equal_ranges': rangos = _equal_ranges( tasa_values ) else: rangos = _wide_medium_ranges( tasa_values ) return rangos def _asumming_normal_dist( tasa_values ): """ Calcula rangos asumiendo distribución normal ( mu +- D.E. ) - Menor a DE - prom -> añadir a baja - Mayor a DE + prom -> añadir a alta """ stdev = statistics.stdev( tasa_values ) mean = statistics.mean( tasa_values ) std_plus_mean = mean + stdev / 2 std_minus_mean = mean - stdev / 2 return [ std_minus_mean, std_plus_mean ] def _equal_ranges( tasa_values ): """ Se divide el rango en 3 partes iguales """ min_value = min(tasa_values) rango = max(tasa_values) - min_value step = rango / 3 return [ min_value + step , min_value + step*2 ] def _wide_medium_ranges( tasa_values, medium_range_percentage = 0.45 ): """ Se le asigna al rango central un mayor porcentaje del rango, definido mediante medium_range_percentage """ min_value = min(tasa_values) max_value = max(tasa_values) rango = max_value - min_value # Se asume que % del rango correspondiente al máximo y mínimo son iguales min_max_percentage = 1-medium_range_percentage step = ((min_max_percentage) / 2) * rango return [ min_value + step , max_value - step ] """ Ejemplos de uso import pprint import sys import time 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.utils.Scapp_analytic_utils import process_one_day_data # today = datetime.now() today = datetime(2025,3,19) start_of_day = datetime(today.year, today.month, today.day) end_of_day = start_of_day + timedelta(days=1) today_data = fetch_data_collection(start_of_day, end_of_day, verbose=False) visitas_today = fetch_visitas_collection( start_of_day ) formatted_day = start_of_day.strftime('%d/%m/%Y') # Y -> 2025 ; y -> 25 t1 = datetime.now() analytic_data = process_one_day_data( today_data, visitas_today, formatted_day ) 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 día: {sys.getsizeof( analytic_data )} bytes demorando {(t_end - t1).total_seconds()*1e3} milisegundos") """ """ Modificación de datos Estaba pensando en modificar la estructura del json que te mando, crear una nueva llave: data_summary Que contenga los valores que son solo un numero o str, algo así: values_summary = { day : 'dia igual que ahora', value : { - sfr_value - delivered_food - tasa_units - fasts_count - alarms_count - alarms_by_ton - perdidos_count } Mantener como llave los que son un array y el rango de alimentacion - tasa_data (el grafico) - tasa_values - feeding_time_range - alarms_by_tasa - alarms_by_tasa_levels Tendrías que modificar un poco como obtener las primeras métricas pero creo que es un poco mas ordenado, te parece? """