## # @file ValidationFunctions.py # # @brief Definicion de las funciones utilizadas para validacion de los datos almacenados # en la nueva colección. # # @authors DopamineLabs LTDA # @date 2025 ## from bson.objectid import ObjectId from DataDefinitions import * MAX_DIFF_NUMERICAL = 1e-3 def reverse_visitas(visitas): """! Esta función convierte las estructuras de visitas desde la colección transformada al formato de visitas de la colección de origen. @param visitas El diccionario de visitas proveniente de la colección transformada. @return Retorna una lista de diccionarios con el formato de visitas de la colección de origen (no transformada). """ return_struct = [] for index, visita in enumerate(visitas): index_str = str(index) return_struct.append({ 'duracion': visita['duracion'], 'racion': visita['racion'], 'inicio': visita['inicio'].strftime("%Y-%m-%d %H:%M:%S"), 'final': visita['final'].strftime("%Y-%m-%d %H:%M:%S") }) return return_struct def reverse_document(transformed_document): '''! Esta función realiza la conversión de un documento transformado a la colección de serie de tiempo, de vuelta a un documento con el formato de la colección original. @details Known-issue: La conversión del campo 'Porc_entregado' a string, como detalla la especificación del proyecto, se dejó comentada, lo que hace que el valor quede como int. Esto se realiza para que coincida con los valores presente en la data sintética, que no se condicen con la especificación del proyecto. @param transformed_document Documento perteneciente a la colección transformada (serie de tiempo). @return Retorna un diccionario que contiene la información en el formato de la colección original. ''' # Default value reconstructed_doc = {'dia': None, 'Metodo_alimentacion': None} # Unpack data timestamp = transformed_document['timestamp'] metadata = transformed_document['metadata'] detecciones = transformed_document['Detecciones'] alimentacion = transformed_document['Alimentacion'] visitas = transformed_document['Visitas'] posicion = transformed_document['Posicion'] sensores = transformed_document['Sensores'] metodo_alimentacion = transformed_document['Metodo_alimentacion'] # Rebuild reconstructed_doc['dia'] = metadata['Dia'] reconstructed_doc['Metodo_alimentacion'] = metodo_alimentacion cage_name = metadata['Jaula'] # Parse cage data cage_data = { 'hora': timestamp.strftime("%Y-%m-%d %H:%M:%S"), 'perdidos': detecciones['perdidos'], 'detecciones': detecciones['detectados'], 'perdidos_propios': detecciones['perdidos_propios'], 'nivel_actividad': detecciones['nivel_actividad'], 'use_feeding_api': alimentacion['use_feeding_api'], 'detection_active': alimentacion['detection_active'], 'Comida_entregada': alimentacion['Comida_entregada'], 'Meta_dia': alimentacion['Meta_dia'], #'Porc_entregado': str(alimentacion['Porc_entregado']), 'Porc_entregado': alimentacion['Porc_entregado'], # Se deja como int para que coincida con la data sintetica 'Tasa': alimentacion['Tasa'], 'Dia': alimentacion['Dia'], 'Feeding_active': alimentacion['Feeding_active'], 'Unidad_Medida': alimentacion['Unidad_Medida'], 'Visitas': reverse_visitas(visitas), 'x_pos': posicion['x_pos'], 'y_pos': posicion['y_pos'], 'winche_connected': posicion['winche_connected'], 'temp': sensores['temp'] } if(cage_name not in reconstructed_doc.keys()): reconstructed_doc[cage_name] = [cage_data] else: reconstructed_doc[cage_name].append(cage_data) return reconstructed_doc def search_original_cage_entry(original, cage_name, hour_field): '''! Busca la entrada asociada a una jaula (nombre y hora), dentro de un documento de la colección original. @param original Documento de la colección original en la cual se buscará la entrada. @param cage_name Nombre de la jaula. @param hour_field Campo de fecha-hora utilizado para verificar la entrada de jaula buscada. @return Si la entrada está presente en el documento original, retorna un diccionario que contiene la información de la entrada. En caso contrario, retorna None. ''' try: cage_entries = original[cage_name] for cage_data in cage_entries: cage_hour = cage_data['hora'] if(cage_hour == hour_field): return cage_data return None except: # Could not find entry return None def generate_key_evaluation(): '''! Genera un diccionario de evaluación para una entrada. @return Retorna el diccionario de evaluación con valores por defecto. ''' return { 'valid': False, 'value': None, 'type': None, 'expected': None, 'expected_type': None, 'error_msg': '' } def validate_values(original, transformed): '''! Esta función realiza la validación entre dos valores. Da como resultado un documento de evaluación. @detail La función realiza los siguientes checks: -- Validación de tipo: Revisa que el tipo de dato entre ambos valores sea el mismo. -- Validación numérica: Si los valores son numéricos, realiza una diferencia para la validación. Si esta diferencia es menos a 1e-3, se considera como un par de datos iguales. -- Validación de igualdad: Si los valores no son numéricos, se comparan directamente para su validación. La función asigna un campo de validez y en caso de que haya error, una indicación de cual fue el error detectado. @param original Valor asociado al documento de la colección original. @param transformed Valor asociado al documento de la colección transformada. @return Retorna un diccionario de evaluación con los resultados de la comparación. ''' evaluation_dict = generate_key_evaluation() similarity_flag = True error_msg = "" # Common process evaluation_dict['value'] = transformed evaluation_dict['type'] = str(type(transformed)) evaluation_dict['expected'] = original evaluation_dict['expected_type'] = str(type(original)) # Check numerical values first if type(original) != type(transformed): similarity_flag = False error_msg = "Values dont have the same type" elif type(original) in (int, float): diff = abs(original - transformed) if(diff > MAX_DIFF_NUMERICAL): similarity_flag = False error_msg = "Difference between values larger than 1e-3" elif original != transformed: similarity_flag = False error_msg = "Values dont match" # Assign results evaluation_dict['valid'] = similarity_flag evaluation_dict['error_msg'] = error_msg return evaluation_dict def get_validation_result(evaluation_dict): '''! La función toma el diccionario de evaluación para un documento transformado, y genera como resultado un valor de integridad para el documento. @detail La función revisa todas las llaves y su integridad, si existe un único campo con integridad False, se determina que el documento no ha respetado la integridad de la data original. Known-issue: Este criterio estricto, hace que muchos documentos transformados sean marcados con integridad inválida, debido a las dierencias presentes en la data sintética con las especificaciones del proyecto, en las cuales se basaron los criterios de validación. @param evaluation_dict Diccionario de evaluación asociado al documento transformado. @return Retorna un valor booleano. True si el documento no presenta errores de integridad. False en caso contrario. ''' for eval_key, eval_value in evaluation_dict.items(): if(not eval_value['valid']): return False return True def compare_documents(original, associated_ids, transformed_docs, db_handle): '''! Esta función realiza la comparación entre un documento reconstruido (a partir de la colección transformada) y un documento de la colección original. @detail La función toma los ids de documento de la colección transformada, asociados a un documento de la colección original, para realizar la comparación. Cada uno de los documentos asociados, se convierte al formato de la colección original, para realizar una comparación llave a llave de los documentos. Los resultados de las comparaciones individuales entre los documentos son almacenados en la colección pasada. @param original Documento original a comparar. @param associated_ids IDs de documentos transformados, asociados al documento original. @transformed_docs Listado de documentos de la colección transformada. @db_handle Colección en la que se almacenarán los resultados de la evaluación. ''' for a_doc in transformed_docs: a_doc_id = str(a_doc['_id']) if a_doc_id not in associated_ids: # If the docs is not in the ids, skip continue evaluation_dictionary = {} # Revert the document for direct comparison reverted_doc = reverse_document(a_doc) # Start matching directly with original reference for reverted_key, reverted_value in reverted_doc.items(): if reverted_key not in original.keys(): # Check against default value evaluation_dictionary[reverted_key] = validate_values(OLD_TOP_LEVEL_STRUCTURE[reverted_key]['default'], reverted_value) elif isinstance(reverted_value, list): reverted_cage_data = reverted_value[0] # Check if entry exists on original doc reverted_hour = reverted_cage_data['hora'] original_cage_data = search_original_cage_entry(original, reverted_key, reverted_hour) if(original_cage_data == None): error_dict = generate_key_evaluation() error_dict['valid'] = False error_dict['error_msg'] = "Cage data not found in original document." evaluation_dictionary[reverted_key] = error_dict # If cage data exists, evalute every key else: for r_cage_key, r_cage_data in reverted_cage_data.items(): if r_cage_key not in original_cage_data.keys(): evaluation_dictionary[r_cage_key] = validate_values(OLD_CAGE_STRUCTURE[r_cage_key]['default'], r_cage_data) else: original_data_evaluation = original_cage_data[r_cage_key] evaluation_dictionary[r_cage_key] = validate_values(original_data_evaluation, r_cage_data) else: evaluation_dictionary[reverted_key] = validate_values(original[reverted_key], reverted_value) # Add evaluation result doc_evaluation = get_validation_result(evaluation_dictionary) temp_dict = {'integrity': doc_evaluation, 'reference': original['_id'], 'evaluation': evaluation_dictionary} # Save onto collection doc_dict = {'_id': a_doc['_id']} doc_dict.update(temp_dict) db_handle.insert_one(doc_dict)