import json import logging from models.medicion import Medicion from database.connection import DatabaseConnection class MessageProcessor: """Procesador principal de mensajes IoT""" def __init__(self): self.db = DatabaseConnection() def process_single_message(self, message_line): """Procesa una sola línea de mensaje JSON (método original)""" try: # Parsear JSON message = json.loads(message_line.strip()) # Verificar que tiene objectJSON (datos del medidor) if 'objectJSON' not in message or not message['objectJSON']: logging.warning("Mensaje sin objectJSON, omitiendo") return False # Crear objeto Medicion medicion = Medicion(message) # Buscar ID del medidor usando devEUI convertido a hex device_identifier = medicion.get_device_identifier(message) if not device_identifier: logging.warning("No se pudo obtener identificador del dispositivo") return False logging.info(f"Buscando medidor con identificador: {device_identifier}") id_medidor = self._get_medidor_id(device_identifier) if id_medidor is None: logging.warning(f"Medidor no encontrado: {device_identifier}") return False medicion.id_medidor = id_medidor # Insertar en base de datos usando el método simplificado result = self._insert_medicion_and_tiempo_real(medicion) return result['success'] except json.JSONDecodeError as e: logging.error(f"Error decodificando JSON: {e}") return False except Exception as e: logging.error(f"Error procesando mensaje: {e}") return False def process_single_message_detailed(self, message_line): """Procesa una sola línea de mensaje JSON con información detallada de errores""" try: # Parsear JSON try: message = json.loads(message_line.strip()) except json.JSONDecodeError as e: return { 'success': False, 'error_type': 'data_error', 'error_reason': f'JSON inválido: {str(e)}' } # Verificar que tiene objectJSON (datos del medidor) if 'objectJSON' not in message or not message['objectJSON']: return { 'success': False, 'error_type': 'data_error', 'error_reason': 'Mensaje sin objectJSON' } # Verificar que objectJSON contiene datos válidos try: object_data = json.loads(message['objectJSON']) # Verificar que no está vacío y tiene datos esenciales if not object_data or not isinstance(object_data, dict): return { 'success': False, 'error_type': 'data_error', 'error_reason': 'objectJSON vacío o inválido' } # Verificar que tiene al menos algunos campos esenciales required_fields = ['address', 'cumulativeValue', 'innerTime'] missing_fields = [field for field in required_fields if field not in object_data] if missing_fields: return { 'success': False, 'error_type': 'data_error', 'error_reason': f'objectJSON no tiene campos esenciales: {", ".join(missing_fields)}' } except (json.JSONDecodeError, TypeError) as e: return { 'success': False, 'error_type': 'data_error', 'error_reason': f'objectJSON inválido: {str(e)}' } # Crear objeto Medicion medicion = Medicion(message) # Buscar ID del medidor usando devEUI convertido a hex device_identifier = medicion.get_device_identifier(message) if not device_identifier: return { 'success': False, 'error_type': 'data_error', 'error_reason': 'No se pudo obtener identificador del dispositivo' } logging.info(f"Buscando medidor con identificador: {device_identifier}") id_medidor_result = self._get_medidor_id_detailed(device_identifier) if not id_medidor_result['success']: if id_medidor_result['error_type'] == 'not_found': return { 'success': False, 'error_type': 'data_error', 'error_reason': f'Medidor no encontrado: {device_identifier}' } else: return { 'success': False, 'error_type': 'system_error', 'error_reason': f'Error de base de datos: {id_medidor_result["error_reason"]}' } medicion.id_medidor = id_medidor_result['id_medidor'] # Calcular delta acumulado last_cumul_value = self._get_last_cumul_flow_value(medicion.id_medidor) if last_cumul_value is not None: # Convertir a float para evitar conflictos de tipos decimal.Decimal vs float last_cumul_float = float(last_cumul_value) if last_cumul_value is not None else 0.0 medicion.delta_acumulado = round(medicion.cumul_flow_value - last_cumul_float, 3) logging.info(f"Delta calculado: {medicion.cumul_flow_value} - {last_cumul_float} = {medicion.delta_acumulado}") else: medicion.delta_acumulado = None # Primera medición del medidor logging.info("Primera medición del medidor, delta = None") # Procesar ambas inserciones en una sola transacción insert_result = self._insert_medicion_and_tiempo_real(medicion) return insert_result except Exception as e: return { 'success': False, 'error_type': 'system_error', 'error_reason': f'Error inesperado: {str(e)}' } def _get_medidor_id(self, device_identifier): """Busca el ID del medidor por eui (comparación en mayúsculas)""" if not self.db.connect(): return None try: cursor = self.db.get_cursor() # Buscar por eui comparando en mayúsculas (excluir eliminados) query = """ SELECT idMedidor FROM medidores WHERE UPPER(eui) = UPPER(%s) AND deleted_at IS NULL LIMIT 1 """ cursor.execute(query, (device_identifier,)) result = cursor.fetchone() if result: logging.info(f"Medidor encontrado con ID: {result[0]}") return result[0] else: logging.warning(f"Medidor no encontrado en DB: {device_identifier}") return None except Exception as e: logging.error(f"Error buscando medidor: {e}") return None finally: self.db.disconnect() def _get_medidor_id_detailed(self, device_identifier): """Busca el ID del medidor con información detallada de errores""" if not self.db.connect(): return { 'success': False, 'error_type': 'connection_error', 'error_reason': 'No se pudo conectar a la base de datos' } try: cursor = self.db.get_cursor() # Buscar por eui comparando en mayúsculas (excluir eliminados) query = """ SELECT idMedidor FROM medidores WHERE UPPER(eui) = UPPER(%s) AND deleted_at IS NULL LIMIT 1 """ cursor.execute(query, (device_identifier,)) result = cursor.fetchone() if result: logging.info(f"Medidor encontrado con ID: {result[0]}") return { 'success': True, 'id_medidor': result[0], 'error_type': None, 'error_reason': None } else: logging.warning(f"Medidor no encontrado en DB: {device_identifier}") return { 'success': False, 'error_type': 'not_found', 'error_reason': f'Medidor no encontrado: {device_identifier}' } except Exception as e: logging.error(f"Error buscando medidor: {e}") return { 'success': False, 'error_type': 'database_error', 'error_reason': str(e) } finally: self.db.disconnect() def _insert_medicion_and_tiempo_real(self, medicion): """Inserta medición en ambas tablas usando una sola transacción""" if not self.db.connect(): return { 'success': False, 'error_type': 'system_error', 'error_reason': 'No se pudo conectar a la base de datos' } try: cursor = self.db.get_cursor() # Insertar en tabla mediciones mediciones_query = """ INSERT INTO mediciones (idMedidor, nserie, fecha_recepcion, meter_time, rssi, cumul_flow_value, daily_flow_value, reverse_flow_value, flow_rate_value, temperatura, delta_acumulado, status) VALUES (%(idMedidor)s, %(nserie)s, %(fecha_recepcion)s, %(meter_time)s, %(rssi)s, %(cumul_flow_value)s, %(daily_flow_value)s, %(reverse_flow_value)s, %(flow_rate_value)s, %(temperatura)s, %(delta_acumulado)s, %(status)s) """ cursor.execute(mediciones_query, medicion.to_dict()) logging.info(f"Medición insertada en tabla mediciones para medidor ID: {medicion.id_medidor}") # Debug: Verificar que la inserción fue exitosa logging.info(f"Datos insertados: {medicion.to_dict()}") # Insertar/actualizar en tabla tiempo_real tiempo_real_query = """ INSERT INTO tiempo_real (idMedidor, lastReporte, estado, acumulado, diario, rssi, temperatura, cumulFlowUnit, dailyFlowUnit) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE lastReporte = VALUES(lastReporte), estado = VALUES(estado), acumulado = VALUES(acumulado), diario = VALUES(diario), rssi = VALUES(rssi), temperatura = VALUES(temperatura), cumulFlowUnit = VALUES(cumulFlowUnit), dailyFlowUnit = VALUES(dailyFlowUnit) """ cursor.execute(tiempo_real_query, ( medicion.id_medidor, medicion.fecha_recepcion, medicion.status, medicion.cumul_flow_value if medicion.cumul_flow_value else 0, medicion.daily_flow_value if medicion.daily_flow_value else 0, medicion.rssi, medicion.temperatura, medicion.cumul_flow_unit, medicion.daily_flow_unit )) logging.info(f"Tiempo real actualizado para medidor ID: {medicion.id_medidor}") # Hacer commit de ambas operaciones self.db.commit() return { 'success': True, 'error_type': None, 'error_reason': None } except Exception as e: # Rollback en caso de error if self.db.connection: self.db.connection.rollback() logging.error(f"Error insertando medición y tiempo real: {e}") return { 'success': False, 'error_type': 'system_error', 'error_reason': str(e) } finally: self.db.disconnect() def _get_last_cumul_flow_value(self, id_medidor): """Obtiene el último cumul_flow_value del medidor especificado""" if not self.db.connect(): return None try: cursor = self.db.get_cursor() query = """ SELECT cumul_flow_value FROM mediciones WHERE idMedidor = %s ORDER BY fecha_recepcion DESC, id DESC LIMIT 1 """ cursor.execute(query, (id_medidor,)) result = cursor.fetchone() if result: return result[0] else: return None except Exception as e: logging.error(f"Error obteniendo último cumul_flow_value: {e}") return None finally: self.db.disconnect()