from pymongo import MongoClient import yaml import time from datetime import datetime, timezone #from datetime import datetime # Módulos del Proyecto # from scapp_modules.Logger import main_logger, database_logger from .refactorization import (update_scapp_db_version, get_scapp_db_version, refactorice_indices_cameras, transform_to_timeseries, ensure_data_compose_index, ) from .utils.analytic_database_utils import ensure_data_analytic_collection from .utils.db_colection_utils import ensure_visitas_coll_and_index def main_logger(text): print(text) def database_logger(text): print(text) # ---- MongoDB ---- # # -- Usando MongoClient -- # client = MongoClient( 'mongodb://localhost/', 27017 ) database_names = { 'database' : 'ScappDB', 'cameras' : 'cameras', 'data' : 'data', 'analytics' : 'data_analytic', 'visitas' : 'data_visitas', 'parameters': 'parameters' } # ! Nuevas credenciales scapp_db = client[database_names['database']] # .ScappDB cameras_collection = scapp_db[database_names['cameras']] # .cameras data_collection = scapp_db[database_names['data']] # .data data_analytic = scapp_db[database_names['analytics']] # .data_analytic visitas_collection = scapp_db[database_names['visitas']] # .data_visitas parameters_collection = scapp_db[database_names['parameters']] # .parameters VERBOSE = False actual_db_versions = { 'cameras_collection': 1, 'data_collection': 1, # 1: TimeSeries DopamineLabs } def fetch_data_collection( start_of_day, end_of_day, verbose=False, convert_to_iso=True ): """ Retorna un diccionario que contiene toda la información de BD entre los días start_of_day y end_of_day El diccionario utiliza como llave los indices de las cámaras, teniendo como value un array con todos los datos de la BD Cada dato es un diccionario: { timestamp: 2024-06-27T12:03:26.000+00:00, camera_id = "1-A", # metadata: { -> Versión Obsoleta, ya no existe el campo # Dia: "2024-06-27", # Jaula: "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: "-" } } @Params - start_of_day (datetime): Fecha en formato datetime ( Puede obtenerse mediante datetime.now() ) - end_of_day (datetime): Fecha en formato datetime ( Puede obtenerse mediante start_of_day + timedelta(days=1) ) - verbose (bool): Flag para imprimir información adicional por consola """ documents_by_jaula = {} t1 = time.time() try: # Consulta para obtener todos los documentos del día de hoy query = { "timestamp": {"$gte": start_of_day, "$lt": end_of_day} } # Proyección para excluir el campo _id en los documentos retornados projection = {"_id": 0} # Se ordena la consulta por timestamp (ascendente o descendente según convenga) cursor = data_collection.find(query, projection).sort("timestamp", 1) parsed_ts = 0 # Agrupar los documentos en un diccionario usando como llave 'metadata.Jaula' for doc in cursor: # jaula = doc["metadata"]["Jaula"] jaula = doc["camera_id"] if jaula not in documents_by_jaula: documents_by_jaula[jaula] = [] # Si el documento tiene el campo 'timestamp' y es un objeto datetime, se convierte a formato ISO if convert_to_iso: if "timestamp" in doc and isinstance(doc["timestamp"], datetime): doc["timestamp"] = doc["timestamp"].replace(tzinfo=timezone.utc) doc["timestamp"] = doc["timestamp"].isoformat() parsed_ts += 1 # if '_id' in doc: # print(f"ID Encontrado en documento!!") documents_by_jaula[jaula].append(doc) except Exception as e: database_logger(f"Error leyendo documentos de día de hoy {e.__traceback__.tb_lineno} : {e}") if verbose: for jaula, data in documents_by_jaula.items(): database_logger(f"Cargados {len(data)} Datos para el canal {jaula}") print();database_logger(f"Se han convertido a formato iso {parsed_ts} elementos, tardando {(time.time()-t1)*1e3:.2f} ms.") return documents_by_jaula def fetch_visitas_collection( datetime_day, convert_to_iso=False ): """ ### ? Obtiene las visitas de todas las jaulas para una fecha dada en la colección data_visitas. @param datetime_day (datetime): Fecha para la cual recopilar datos @return: Diccionario con la estructura { 'jaula1': visitas_jaula1, 'jaula2': visitas_jaula2, ... } Donde cada visita tiene el formato: [ { duracion: 75.04, racion: 0, inicio: 2024-10-17T17:23:05.000+00:00, final: 2024-10-17T17:24:20.000+00:00 }, ... ] """ try: dia = datetime_day.strftime( "%Y-%m-%d" ) # Filtrar los documentos en base al campo 'metadata.Dia' # cursor = visitas_collection.find({"metadata.Dia": dia}) cursor = visitas_collection.find({"day": dia}) visitas_por_jaula = {} # Recorrer cada documento y extraer el nombre de la jaula y su lista de visitas for doc in cursor: # metadata = doc.get("metadata", {}) # jaula = metadata.get("Jaula") jaula = doc.get("camera_id") visitas = doc.get("Visitas", []) # Validar datos if not jaula or not visitas: continue # Actualizar el timezone de cada visita en el array """ for visita in visitas: if 'inicio' in visita and isinstance(visita['inicio'], datetime): # hasattr(visita['inicio'], 'replace'): visita['inicio'] = visita['inicio'].replace(tzinfo=timezone.utc) if 'final' in visita and isinstance(visita['final'], datetime): # hasattr(visita['final'], 'replace'): visita['final'] = visita['final'].replace(tzinfo=timezone.utc) """ if convert_to_iso: for visita in visitas: if 'inicio' in visita and isinstance(visita['inicio'], datetime): visita['inicio'] = visita['inicio'].isoformat() if 'final' in visita and isinstance(visita['final'], datetime): visita['final'] = visita['final'].isoformat() visitas_por_jaula[jaula] = visitas return visitas_por_jaula except Exception as e: database_logger(f"Error obteniendo visitas para el día {dia}: {e}") return {} def read_yaml( filename ): try: with open( filename, 'r', encoding='utf-8' ) as file: data = yaml.safe_load( file ) return data except: return {} def drop_collection( db, collection_name ): try: db[ collection_name ].drop() database_logger(f"Colección {collection_name} eliminada satisfactoriamente.\n") except Exception as e: database_logger(f"Error eliminando colección: {collection_name}: {e}") def Scapp_loadDB(): # Refactorizar db (Crear ScappDB) y Actualizar ScappDB.data_detailed -> # ! Antiguo, ya no existen clientes con esto. # Scapp_refactorice_db() # Leer si es necesario eliminar colección Analytics configs = read_yaml( 'configScapp.yaml' ) delete_analytics = configs.get( 'Analytic_delete_saved_data' , {}).get('value', False) if delete_analytics: print() ; database_logger(f"Se ha solicitado eliminar colección de analítica.") drop_collection( scapp_db, database_names['analytics'] ) # * MODO Cámaras Siempre activas # ? Fuerza detección siempre activa, no hace streaming de imágenes, no dibuja detecciones always_active = False readed_db_version = get_scapp_db_version(parameters_collection) # Revisar versión de colección cámaras if readed_db_version.get('cameras_collection', 0) < actual_db_versions['cameras_collection']: try: refactorice_indices_cameras(cameras_collection) update_scapp_db_version( parameters_collection, 'cameras_collection', actual_db_versions['cameras_collection']) main_logger(f"Actualización de base de datos de cámaras correcta.\n") except Exception as e: main_logger(f"Error refactorizando colección de cámaras en la Base de Datos: \n{e}\n") else: main_logger(f"Colección de cámaras ya actualizada.\n") # Revisar versión de data y data_visitas ensure_visitas_coll_and_index( scapp_db, database_names['visitas'] ) if readed_db_version.get('data_collection', 0) < actual_db_versions['data_collection']: try: args = { 'target_collection': database_names['data'], 'old_collection' : 'data_old', 'log_file': 'migration_log', 'verbose': False, } transform_to_timeseries(scapp_db, args) update_scapp_db_version(parameters_collection, 'data_collection', actual_db_versions['data_collection']) main_logger(f"Migración a series de tiempo exitosa.\n") except Exception as e: main_logger(f"Error migrando colección data a series de tiempo: {e}\n") else: ensure_data_compose_index( scapp_db[ database_names['data'] ] ) main_logger(f"Colección data ya actualizada a Series de Tiempo\n") # raise Exception(f"Cerrando Scapp anticipadamente") # Una migración pendiente de analítica no debe impedir el resto del servicio. try: ensure_data_analytic_collection( scapp_db, database_names['analytics'] ) except Exception as error: database_logger(f"No se pudo verificar la colección de analítica: {error}") return always_active always_active = Scapp_loadDB() # if __name__ == '__main__': """ # * Ejemplo de uso en consola: from datetime import datetime, timedelta from scapp_database import fetch_data_collection from scapp_database import fetch_data_collection, fetch_visitas_collection today = datetime.now() 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=True) visitas_today = fetch_visitas_collection( start_of_day ) """