""" Gestor de conexiones para Serial, TCP y UDP """ import serial import socket import time class ConnectionManager: def __init__(self): self.connection = None self.connection_type = None self.dest_address = None # Para UDP def open_connection(self, conn_type, conn_params): """Abre una conexión según el tipo especificado""" try: if conn_type == "Serial": self.connection = serial.Serial( port=conn_params['port'], baudrate=conn_params['baudrate'], # Standard pyserial parameter name timeout=conn_params.get('timeout', 1), bytesize=conn_params.get('bytesize', serial.EIGHTBITS), # Use provided or default parity=conn_params.get('parity', serial.PARITY_NONE), # Use provided or default stopbits=conn_params.get('stopbits', serial.STOPBITS_ONE), # Use provided or default xonxoff=conn_params.get('xonxoff', False), rtscts=conn_params.get('rtscts', False), dsrdtr=conn_params.get('dsrdtr', False) ) self.connection_type = "Serial" elif conn_type == "TCP": sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(5.0) sock.connect((conn_params['ip'], conn_params['port'])) self.connection = sock self.connection_type = "TCP" elif conn_type == "UDP": sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.settimeout(1.0) self.dest_address = (conn_params['ip'], conn_params['port']) self.connection = sock self.connection_type = "UDP" return self.connection except Exception as e: raise Exception(f"Error al abrir conexión {conn_type}: {e}") def close_connection(self): """Cierra la conexión actual""" try: if self.connection_type == "Serial": if self.connection and self.connection.is_open: self.connection.close() elif self.connection_type in ["TCP", "UDP"]: if self.connection: self.connection.close() except Exception as e: print(f"Error al cerrar conexión: {e}") finally: self.connection = None self.connection_type = None self.dest_address = None def send_data(self, data_bytes): """Envía datos por la conexión actual""" if not self.connection: raise Exception("No hay conexión activa") data_to_send = None if isinstance(data_bytes, str): # Esto no debería suceder si el llamador (NetComTab, SimulatorTab) funciona como se espera. # Loguear una advertencia e intentar codificar como último recurso. print(f"ADVERTENCIA: ConnectionManager.send_data recibió str, se esperaba bytes. Intentando codificar a ASCII. Datos: {data_bytes!r}") try: data_to_send = data_bytes.encode('ascii') except UnicodeEncodeError as uee: print(f"ERROR CRÍTICO: No se pudo codificar la cadena (str) a ASCII antes de enviar: {uee}. Datos: {data_bytes!r}") # Elevar una excepción clara porque no se puede continuar si la codificación falla. raise Exception(f"Error al enviar datos: la cadena no pudo ser codificada a ASCII: {uee}") from uee elif isinstance(data_bytes, (bytes, bytearray)): data_to_send = data_bytes # Ya es bytes o bytearray (que .write/.send aceptan) else: # Si no es ni str ni bytes/bytearray, es un error de tipo fundamental. print(f"ERROR CRÍTICO: ConnectionManager.send_data recibió un tipo inesperado: {type(data_bytes)}. Se esperaba bytes. Datos: {data_bytes!r}") raise TypeError(f"Error al enviar datos: se esperaba un objeto tipo bytes, pero se recibió {type(data_bytes)}") try: if self.connection_type == "Serial": self.connection.write(data_to_send) elif self.connection_type == "TCP": self.connection.send(data_to_send) elif self.connection_type == "UDP": self.connection.sendto(data_to_send, self.dest_address) except Exception as e: raise Exception(f"Error al enviar datos: {e}") def read_response(self, timeout=0.5): """Intenta leer una respuesta del dispositivo""" if not self.connection: return None try: response = None if self.connection_type == "Serial": # Guardar timeout original original_timeout = self.connection.timeout self.connection.timeout = timeout # Esperar un poco para que llegue la respuesta time.sleep(0.05) # Leer todos los bytes disponibles response_bytes = b"" start_time = time.time() while (time.time() - start_time) < timeout: if self.connection.in_waiting > 0: response_bytes += self.connection.read(self.connection.in_waiting) # Si encontramos un terminador, salir if b'\r' in response_bytes or b'\n' in response_bytes: break else: time.sleep(0.01) if response_bytes: response = response_bytes.decode('ascii', errors='ignore') self.connection.timeout = original_timeout elif self.connection_type == "TCP": self.connection.settimeout(timeout) try: response = self.connection.recv(1024).decode('ascii', errors='ignore') except socket.timeout: pass elif self.connection_type == "UDP": self.connection.settimeout(timeout) try: response, addr = self.connection.recvfrom(1024) response = response.decode('ascii', errors='ignore') except socket.timeout: pass return response except Exception as e: print(f"Error al leer respuesta: {e}") return None def read_data_non_blocking(self): """Lee datos disponibles sin bloquear (para modo trace y netcom)""" if not self.connection: return None try: data = None if self.connection_type == "Serial": if self.connection.in_waiting > 0: data = self.connection.read(self.connection.in_waiting) # Returns bytes elif self.connection_type == "TCP": self.connection.settimeout(0.1) try: data = self.connection.recv(1024) # Returns bytes if not data: # Conexión cerrada return None except socket.timeout: pass elif self.connection_type == "UDP": self.connection.settimeout(0.1) try: data, addr = self.connection.recvfrom(1024) # data is already bytes except socket.timeout: pass return data except Exception as e: print(f"Error al leer datos: {e}") return None def is_connected(self): """Verifica si hay una conexión activa""" if self.connection_type == "Serial": return self.connection and self.connection.is_open else: return self.connection is not None