import socket import struct import time from datetime import datetime import logging logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') class CircutorReader: def __init__(self, ip, port): self.ip = ip self.port = port self.socket = None self.direccion = 1 # Dirección del medidor, ajustar según sea necesario self.clave = 1 # Clave de acceso, ajustar según sea necesario def connect(self): self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.connect((self.ip, self.port)) logging.info(f"Conectado a {self.ip}:{self.port}") def disconnect(self): if self.socket: self.socket.close() logging.info("Desconectado") def send_receive(self, data, expected_response=None): logging.debug(f"Enviando: {data.hex()}") self.socket.sendall(data) response = self.socket.recv(1024) logging.debug(f"Recibido: {response.hex()}") if len(response) < 6: raise Exception(f"Respuesta demasiado corta: {response.hex()}") if response[0] == 0x10: # Trama de longitud fija if expected_response is not None and response[1] != expected_response: logging.warning(f"Respuesta inesperada: esperaba 0x{expected_response:02x}, recibió 0x{response[1]:02x}") return response elif response[0] == 0x68: # Trama de longitud variable if expected_response is not None and response[4] != expected_response: logging.warning(f"Respuesta inesperada: esperaba 0x{expected_response:02x}, recibió 0x{response[4]:02x}") return response else: raise Exception(f"Formato de trama desconocido: {response.hex()}") def authenticate(self): try: # Paso 1: Iniciar comunicación init_request = self.create_fixed_frame(0x49) response = self.send_receive(init_request, 0x0B) logging.info("Paso 1 completado") # Paso 2: Reposición del enlace remoto reset_request = self.create_fixed_frame(0x40) response = self.send_receive(reset_request, 0x00) logging.info("Paso 2 completado") # Paso 3: Verificar estado del enlace status_request = self.create_fixed_frame(0x49) response = self.send_receive(status_request, 0x0B) logging.info("Paso 3 completado") # Paso 4: Enviar clave de acceso auth_request = self.create_auth_request() response = self.send_receive(auth_request, 0x00) logging.info("Paso 4 completado") # Paso 5: Solicitar confirmación de autenticación confirm_request = self.create_fixed_frame(0x5B) response = self.send_receive(confirm_request, 0x08) logging.info("Paso 5 completado") if response[0] == 0x68 and (response[7] != 0xB7 or response[9] != 0x07): raise Exception("Confirmación de autenticación fallida") logging.info("Autenticación completada con éxito") return True except Exception as e: logging.error(f"Error de autenticación: {e}") return False def create_fixed_frame(self, control): msg = bytearray([0x10, control, self.direccion & 0xFF, self.direccion >> 8]) crc = self.crc8(msg[1:]) msg.extend([crc, 0x16]) return msg def create_auth_request(self): msg = bytearray([0x68, 0x0D, 0x0D, 0x68, 0x73, self.direccion & 0xFF, self.direccion >> 8]) msg.extend([0xB7, 0x01, 0x06, 0x01, 0x00, 0x00]) msg.extend(struct.pack('> 8]) msg.extend([0xA2, 0x01, 0x05, 0x01, 0x00, 0x00, group]) crc = self.crc8(msg[4:]) msg.extend([crc, 0x16]) return msg def parse_instantaneous_response(self, response, group): values = {} if response[0] == 0x68 and response[7] == 0xA3 and response[9] == 0x05: try: if group == 192: # Energía values['active_import'] = struct.unpack('>I', response[14:18])[0] values['active_export'] = struct.unpack('>I', response[18:22])[0] values['reactive_q1'] = struct.unpack('>I', response[22:26])[0] values['reactive_q2'] = struct.unpack('>I', response[26:30])[0] values['reactive_q3'] = struct.unpack('>I', response[30:34])[0] values['reactive_q4'] = struct.unpack('>I', response[34:38])[0] values['energy_timestamp'] = self._parse_timestamp(response[38:43]) elif group == 193: # Potencia values['active_power_total'] = struct.unpack('>i', response[14:18])[0] / 10.0 values['reactive_power_total'] = struct.unpack('>i', response[18:22])[0] / 10.0 values['power_factor_total'] = struct.unpack('>h', response[22:24])[0] / 1000.0 for i in range(3): start = 24 + i * 8 values[f'active_power_l{i+1}'] = struct.unpack('>i', response[start:start+4])[0] / 10.0 values[f'reactive_power_l{i+1}'] = struct.unpack('>i', response[start+4:start+8])[0] / 10.0 values['power_timestamp'] = self._parse_timestamp(response[48:53]) elif group == 194: # Tensión y Corriente for i in range(3): start = 14 + i * 8 values[f'current_l{i+1}'] = struct.unpack('>i', response[start:start+4])[0] / 1000.0 values[f'voltage_l{i+1}'] = struct.unpack('>i', response[start+4:start+8])[0] / 10.0 values['vi_timestamp'] = self._parse_timestamp(response[38:43]) logging.debug(f"Datos parseados para el grupo {group}: {values}") except struct.error as e: logging.error(f"Error al desempaquetar datos para el grupo {group}: {e}") logging.debug(f"Datos recibidos: {response.hex()}") except ValueError as e: logging.error(f"Error al interpretar datos para el grupo {group}: {e}") logging.debug(f"Datos recibidos: {response.hex()}") except IndexError as e: logging.error(f"Error de índice al procesar datos para el grupo {group}: {e}") logging.debug(f"Datos recibidos: {response.hex()}") else: logging.warning(f"Respuesta inesperada para el grupo {group}: {response.hex()}") return values def _parse_timestamp(self, data): try: minute = data[0] & 0x3F hour = data[1] & 0x1F day = data[2] & 0x1F month = data[3] & 0x0F year = data[4] + 2000 return datetime(year, month, day, hour, minute).isoformat() except ValueError as e: logging.error(f"Error al parsear timestamp: {e}. Datos: {data.hex()}") return "Timestamp inválido" except IndexError as e: logging.error(f"Error de índice al parsear timestamp: {e}. Datos: {data.hex()}") return "Timestamp inválido" # Ejemplo de uso if __name__ == "__main__": reader = CircutorReader("35.247.249.153", 58111) reader.connect() try: instant_values = reader.read_instantaneous_values() print("Valores instantáneos:") for key, value in instant_values.items(): if isinstance(value, float): if 'energy' in key: print(f"{key}: {value:.3f} kWh/kVArh") elif 'power' in key and 'factor' not in key: print(f"{key}: {value:.3f} kW/kVAr") elif 'current' in key: print(f"{key}: {value:.3f} A") elif 'voltage' in key: print(f"{key}: {value:.1f} V") elif 'factor' in key: print(f"{key}: {value:.3f}") else: print(f"{key}: {value:.3f}") else: print(f"{key}: {value}") except Exception as e: print(f"Error: {e}") finally: reader.disconnect()