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()}") return response 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_group_194_response(self, response): values = {} logging.debug(f"Respuesta recibida: {response.hex()}") if len(response) >= 36: # Verificamos que la longitud sea suficiente try: for i in range(3): start = 12 + i * 6 # Ajustamos el inicio de los datos # Extraemos corriente (3 bytes, little endian) current = int.from_bytes(response[start:start+3], 'little') # Extraemos tensión (3 bytes, little endian) voltage = int.from_bytes(response[start+3:start+6], 'little') values[f'current_l{i+1}'] = current / 10.0 # Convertimos décimas de A a A values[f'voltage_l{i+1}'] = voltage / 10.0 # Convertimos décimas de V a V # Parseamos el timestamp (los últimos 5 bytes antes del checksum y byte de cierre) timestamp_data = response[-7:-2] values['vi_timestamp'] = self._parse_timestamp(timestamp_data) logging.debug(f"Datos parseados: {values}") except Exception as e: logging.error(f"Error al parsear datos del grupo 194: {e}") else: logging.warning(f"Respuesta demasiado corta para el grupo 194: {response.hex()}") return values def _parse_timestamp(self, timestamp_bytes): try: year = 2000 + timestamp_bytes[4] month = timestamp_bytes[3] day = timestamp_bytes[2] hour = timestamp_bytes[1] minute = timestamp_bytes[0] return datetime(year, month, day, hour, minute) except Exception as e: logging.error(f"Error al parsear timestamp: {e}") return None ''' def parse_group_194_response(self, response): values = {} logging.debug(f"Respuesta recibida: {response.hex()}") if len(response) >= 36: # Ajustamos la longitud mínima esperada try: for i in range(3): start = 14 + i * 8 current = struct.unpack('>i', response[start:start+4])[0] voltage = struct.unpack('>i', response[start+4:start+8])[0] values[f'current_l{i+1}'] = current / 1000.0 # Convertimos mA a A values[f'voltage_l{i+1}'] = voltage / 10.0 # Convertimos decivolts a V # Parseamos el timestamp (los últimos 5 bytes antes del byte de cierre) timestamp_data = response[-6:-1] values['vi_timestamp'] = self._parse_timestamp(timestamp_data) logging.debug(f"Datos parseados: {values}") except Exception as e: logging.error(f"Error al parsear datos del grupo 194: {e}") else: logging.warning(f"Respuesta demasiado corta para el grupo 194: {response.hex()}") return values def _parse_timestamp(self, data): try: logging.debug(f"Datos de timestamp recibidos: {data.hex()}") minute = data[0] hour = data[1] day = data[2] month = data[3] year = data[4] + 2000 # Asumimos que los años empiezan desde 2000 logging.debug(f"Timestamp parseado - Año: {year}, Mes: {month}, Día: {day}, Hora: {hour}, Minuto: {minute}") return datetime(year, month, day, hour, minute).isoformat() except Exception as e: logging.error(f"Error 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: group_194_values = reader.read_group_194() print("Valores del Grupo 194 (Tensión y Corriente):") for key, value in group_194_values.items(): if 'current' in key: print(f"{key}: {value:.3f} A") elif 'voltage' in key: print(f"{key}: {value:.1f} V") else: print(f"{key}: {value}") except Exception as e: print(f"Error: {e}") finally: reader.disconnect()