""" Sistema de Lectura Automática de Medidores Este script implementa un backend para la lectura automática de medidores de energía eléctrica. Soporta múltiples conexiones simultáneas y puede manejar diferentes protocolos de medidores. Instrucciones de uso: 1. Asegúrese de tener Python 3.6 o superior instalado. 2. Modifique la sección 'if __name__ == "__main__":' al final del archivo para configurar sus conexiones. 3. Ejecute el script con: python3 ev_ami_backend.py Configuración de conexiones: backend.add_connection(host, port, mode, meter_type, interrogation_enabled, interrogation_frequency) - host: dirección IP del medidor - port: puerto del medidor - mode: "client" para conexiones salientes - meter_type: tipo de protocolo del medidor (por ejemplo, "IEC62056") - interrogation_enabled: True para habilitar la interrogación activa, False para modo pasivo - interrogation_frequency: frecuencia de interrogación en segundos (si está habilitada) Ejemplo: backend.add_connection("192.168.1.100", 5000, "client", "IEC62056", True, 30) Los datos recibidos se guardarán en el archivo 'meter_readings.txt' en el mismo directorio. """ import socket import json import threading import time import logging from abc import ABC, abstractmethod from datetime import datetime import re # Configuración de logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') class MeterProtocol(ABC): """Clase base abstracta para protocolos de medidores.""" @abstractmethod def process_message(self, message): """Procesa un mensaje recibido del medidor.""" pass class IEC62056Protocol(MeterProtocol): """Implementación del protocolo IEC62056 para medidores.""" def process_message(self, message): """ Procesa un mensaje en formato IEC62056. Args: message (str): Mensaje recibido del medidor. Returns: dict: Datos procesados del medidor. """ reception_time = datetime.now().isoformat() if message.startswith("EV IM:"): return self.process_ev_im_message(message, reception_time) else: return self.process_standard_message(message, reception_time) def process_ev_im_message(self, message, reception_time): """Procesa mensajes que comienzan con 'EV IM:'.""" lines = message.strip().split('\n') ev_im_line = lines[0] obis_data = '|'.join(lines) ev_im_match = re.match(r'EV IM:([0-9A-F:]+)', ev_im_line) ev_im = ev_im_match.group(1) if ev_im_match else "" return { "fecha": reception_time, "EV IM": ev_im, "OBIS": obis_data } def process_standard_message(self, message, reception_time): """Procesa mensajes estándar que no comienzan con 'EV IM:'.""" lines = message.strip().split('\n') obis_data = '|'.join(lines) return { "fecha": reception_time, "EV IM": "", "OBIS": obis_data } class MeterConnection: """Maneja la conexión y comunicación con un medidor específico.""" def __init__(self, host, port, mode, meter_type, interrogation_enabled=False, interrogation_frequency=60): """ Inicializa una conexión de medidor. Args: host (str): Dirección IP del medidor. port (int): Puerto del medidor. mode (str): Modo de conexión ("client" o "server"). meter_type (str): Tipo de protocolo del medidor. interrogation_enabled (bool): Si se debe interrogar activamente al medidor. interrogation_frequency (int): Frecuencia de interrogación en segundos. """ self.host = host self.port = port self.mode = mode self.meter_type = meter_type self.sock = None self.connected = False self.protocol = self.get_protocol(meter_type) self.logger = logging.getLogger(f"MeterConnection-{host}:{port}") self.interrogation_enabled = interrogation_enabled self.interrogation_frequency = interrogation_frequency def get_protocol(self, meter_type): """Selecciona el protocolo adecuado basado en el tipo de medidor.""" protocols = { "IEC62056": IEC62056Protocol(), # Añadir más protocolos aquí si es necesario } return protocols.get(meter_type, IEC62056Protocol()) # Default to IEC62056 if unknown def connect(self): """Establece la conexión con el medidor.""" while not self.connected: try: self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.connect((self.host, self.port)) self.logger.info(f"Cliente conectado a {self.host}:{self.port}") self.connected = True if self.interrogation_enabled: threading.Thread(target=self.interrogate_meter, daemon=True).start() except Exception as e: self.logger.error(f"Error de conexión: {e}. Reintentando en 5 segundos...") time.sleep(5) def interrogate_meter(self): """Envía periódicamente solicitudes de estado al medidor.""" while self.connected: try: self.sock.sendall(b"estado") self.logger.info(f"Enviada solicitud de estado a {self.host}:{self.port}") time.sleep(self.interrogation_frequency) except Exception as e: self.logger.error(f"Error al enviar solicitud de estado: {e}") self.connected = False self.sock.close() self.connect() def handle_connection(self): """Maneja la recepción de datos del medidor.""" while True: try: data = self.sock.recv(4096) if not data: raise Exception("Conexión cerrada por el otro extremo") processed_data = self.protocol.process_message(data.decode('utf-8')) self.save_data(processed_data) except Exception as e: self.logger.error(f"Error en la conexión: {e}") self.connected = False self.sock.close() self.connect() def save_data(self, data): """Guarda los datos procesados en un archivo.""" timestamp = datetime.now().isoformat() json_data = json.dumps({ "timestamp": timestamp, "meter": f"{self.host}:{self.port}", "data": data }, indent=2) with open("meter_readings.txt", "a") as f: f.write(json_data + "\n") self.logger.info(f"Datos guardados: {json_data}") class MeterReadingBackend: """Gestiona múltiples conexiones de medidores.""" def __init__(self): self.connections = [] def add_connection(self, host, port, mode, meter_type, interrogation_enabled=False, interrogation_frequency=60): """ Añade una nueva conexión de medidor al backend. Args: host (str): Dirección IP del medidor. port (int): Puerto del medidor. mode (str): Modo de conexión ("client" o "server"). meter_type (str): Tipo de protocolo del medidor. interrogation_enabled (bool): Si se debe interrogar activamente al medidor. interrogation_frequency (int): Frecuencia de interrogación en segundos. """ connection = MeterConnection(host, port, mode, meter_type, interrogation_enabled, interrogation_frequency) self.connections.append(connection) def start(self): """Inicia todas las conexiones de medidores.""" threads = [] for connection in self.connections: thread = threading.Thread(target=self.run_connection, args=(connection,)) thread.start() threads.append(thread) for thread in threads: thread.join() def run_connection(self, connection): """Ejecuta una conexión de medidor individual.""" connection.connect() connection.handle_connection() if __name__ == "__main__": # Crear una instancia del backend backend = MeterReadingBackend() # Configurar las conexiones de los medidores # Formato: backend.add_connection(host, port, mode, meter_type, interrogation_enabled, interrogation_frequency) backend.add_connection("35.247.249.153", 48306, "client", "IEC62056", interrogation_enabled=True, interrogation_frequency=600) backend.add_connection("35.247.249.153", 48316, "client", "IEC62056", interrogation_enabled=True, interrogation_frequency=600) backend.add_connection("35.247.249.153", 48307, "client", "IEC62056", interrogation_enabled=True, interrogation_frequency=60) # Iniciar el backend backend.start()