dzxj_dtu/backend/modules/serial_port.py

import serial
import serial.tools.list_ports
import threading
import time
import logging
import platform
import glob
from dataclasses import dataclass

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('serial_port')

@dataclass
class SerialConfig:
    """串口配置数据类"""
    port: str
    baudrate: int = 115200
    bytesize: int = serial.EIGHTBITS
    parity: str = serial.PARITY_NONE
    stopbits: int = serial.STOPBITS_ONE
    timeout: float = 1.0
    xonxoff: bool = False
    rtscts: bool = False
    dsrdtr: bool = False

class SerialPort:
    """串口通信类，提供串口连接、读写和状态管理功能"""
    
    def __init__(self):
        self.ser = None
        self.is_connected = False
        self.lock = threading.RLock()  # 使用可重入锁
        self.read_thread = None
        self.stop_event = threading.Event()
        self.data_callback = None
        self.status_callback = None
        self.error_callback = None
        self.current_config = None
        self.reconnect_attempts = 0
        self.max_reconnect_attempts = 3
    
    def list_ports(self):
        """列出系统中可用的串口"""
        ports = []
        
        try:
            # 首先尝试使用serial.tools.list_ports
            try:
                detected_ports = [port.device for port in serial.tools.list_ports.comports()]
                ports.extend(detected_ports)
            except Exception as e:
                logger.warning(f"使用serial.tools.list_ports失败: {str(e)}")
            
            # 根据不同平台进行补充查找
            system = platform.system()
            
            if system == 'Windows':
                try:
                    import winreg
                    # 在Windows系统中读取注册表
                    key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, 
                                        r'HARDWARE\DEVICEMAP\SERIALCOMM')
                    i = 0
                    while True:
                        try:
                            port, value, _ = winreg.EnumValue(key, i)
                            if value not in ports:
                                ports.append(value)
                            i += 1
                        except OSError:
                            break
                except Exception as e:
                    logger.error(f"读取Windows串口注册表失败: {str(e)}")
            
            elif system == 'Darwin':  # macOS
                # 使用glob查找/dev/tty.*设备
                darwin_ports = glob.glob('/dev/tty.*')
                # 过滤掉不需要的端口
                for port in darwin_ports:
                    if not ('Bluetooth' in port or 'debug' in port or 'com.apple' in port) and port not in ports:
                        ports.append(port)
            
            elif system == 'Linux':
                # 使用glob查找Linux系统中的串口
                linux_ports = glob.glob('/dev/ttyS*') + glob.glob('/dev/ttyUSB*') + glob.glob('/dev/ttyACM*')
                for port in linux_ports:
                    if port not in ports:
                        ports.append(port)
            
            logger.info(f"找到 {len(ports)} 个可用串口: {ports}")
        except Exception as e:
            logger.error(f"列出串口时出错: {str(e)}")
        
        return sorted(ports)  # 排序返回
    
    def connect(self, port, baudrate=9600, timeout=1, **kwargs):
        """连接到串口"""
        try:
            # 构建配置
            config = SerialConfig(
                port=port,
                baudrate=baudrate,
                timeout=timeout,
                **kwargs
            )
            
            with self.lock:
                if self.is_connected:
                    self.disconnect()
                
                logger.info(f"尝试连接串口: {port}, 波特率: {baudrate}")
                self.ser = serial.Serial(
                    port=config.port,
                    baudrate=config.baudrate,
                    bytesize=config.bytesize,
                    parity=config.parity,
                    stopbits=config.stopbits,
                    timeout=config.timeout,
                    xonxoff=config.xonxoff,
                    rtscts=config.rtscts,
                    dsrdtr=config.dsrdtr
                )
                
                # 检查连接是否成功
                if not self.ser.is_open:
                    raise Exception("串口打开失败")
                
                self.is_connected = True
                self.stop_event.clear()
                self.current_config = config
                self.reconnect_attempts = 0
                
                # 启动读取线程
                self.read_thread = threading.Thread(target=self._read_loop, daemon=True)
                self.read_thread.start()
                
                if self.status_callback:
                    self.status_callback(True)
                
                logger.info(f"已连接到 {port}，波特率 {baudrate}")
                return True, f"已连接到 {port}，波特率 {baudrate}"
                
        except Exception as e:
            error_msg = f"连接失败: {str(e)}"
            logger.error(error_msg)
            if self.status_callback:
                self.status_callback(False)
            if self.error_callback:
                self.error_callback(error_msg)
            return False, error_msg
    
    def disconnect(self):
        """断开串口连接"""
        try:
            with self.lock:
                logger.info("断开串口连接")
                self.stop_event.set()
                if self.read_thread and self.read_thread.is_alive():
                    self.read_thread.join(timeout=2.0)
                    if self.read_thread.is_alive():
                        logger.warning("读取线程未能正常终止")
                
                if self.ser and self.ser.is_open:
                    try:
                        self.ser.close()
                    except Exception as e:
                        logger.error(f"关闭串口时出错: {str(e)}")
                    finally:
                        self.ser = None
                
                self.is_connected = False
                self.current_config = None
                
                if self.status_callback:
                    self.status_callback(False)
                
            return True, "已断开连接"
            
        except Exception as e:
            error_msg = f"断开连接失败: {str(e)}"
            logger.error(error_msg)
            if self.error_callback:
                self.error_callback(error_msg)
            return False, error_msg
    
    def _read_loop(self):
        """读取串口数据的循环"""
        logger.info("启动串口读取线程")
        
        while not self.stop_event.is_set():
            try:
                if self.ser and self.ser.is_open:
                    # 使用in_waiting提高效率
                    if self.ser.in_waiting > 0:
                        data = self.ser.read(self.ser.in_waiting)
                        # 尝试解码，如果失败则返回原始数据
                        try:
                            decoded_data = data.decode('utf-8', errors='replace').strip()
                            if decoded_data and self.data_callback:
                                self.data_callback(decoded_data)
                        except Exception as e:
                            # 对于无法解码的二进制数据，以十六进制形式返回
                            hex_data = data.hex()
                            if hex_data and self.data_callback:
                                self.data_callback(hex_data)
                            logger.warning(f"收到无法解码的二进制数据，长度: {len(data)}字节")
                    time.sleep(0.001)
            except Exception as e:
                error_msg = f"读取串口数据错误: {str(e)}"
                logger.error(error_msg)
                if self.error_callback:
                    self.error_callback(error_msg)
                
                # 尝试重连
                if self._should_reconnect():
                    logger.warning(f"尝试重连串口... (第{self.reconnect_attempts}次)")
                    if self.current_config:
                        # 等待一段时间后重连
                        time.sleep(2)
                        self.connect(
                            port=self.current_config.port,
                            baudrate=self.current_config.baudrate,
                            timeout=self.current_config.timeout,
                            bytesize=self.current_config.bytesize,
                            parity=self.current_config.parity,
                            stopbits=self.current_config.stopbits,
                            xonxoff=self.current_config.xonxoff,
                            rtscts=self.current_config.rtscts,
                            dsrdtr=self.current_config.dsrdtr
                        )
                break
        
        # 线程结束时清理资源
        logger.info("串口读取线程结束")
        with self.lock:
            self.is_connected = False
            if self.ser:
                try:
                    self.ser.close()
                except:
                    pass
                self.ser = None
            if self.status_callback:
                self.status_callback(False)
    
    def send_data(self, data, encoding='utf-8'):
        """发送数据到串口"""
        try:
            with self.lock:
                if not self.is_connected or not self.ser or not self.ser.is_open:
                    return False, "串口未连接"
                
                # 确保数据以换行符结束
                if isinstance(data, str):
                    if not data.endswith('\n'):
                        data += '\n'
                    bytes_data = data.encode(encoding)
                elif isinstance(data, bytes):
                    if not data.endswith(b'\n'):
                        bytes_data = data + b'\n'
                    else:
                        bytes_data = data
                else:
                    raise TypeError("数据必须是字符串或字节类型")
                
                bytes_sent = self.ser.write(bytes_data)
                self.ser.flush()  # 确保数据被发送
                
                logger.debug(f"发送数据到串口: {bytes_data.hex()[:50]}... (共{bytes_sent}字节)")
                return True, "发送成功"
        except Exception as e:
            error_msg = f"发送失败: {str(e)}"
            logger.error(error_msg)
            if self.error_callback:
                self.error_callback(error_msg)
            return False, error_msg
    
    def set_data_callback(self, callback):
        """设置数据接收回调函数"""
        self.data_callback = callback
    
    def set_status_callback(self, callback):
        """设置状态变化回调函数"""
        self.status_callback = callback
    
    def set_error_callback(self, callback):
        """设置错误回调函数"""
        self.error_callback = callback
    
    def get_status(self):
        """获取当前连接状态"""
        with self.lock:
            return {
                'connected': self.is_connected,
                'config': self.current_config,
                'has_error': self.reconnect_attempts > 0
            }
    
    def _should_reconnect(self):
        """判断是否应该尝试重连"""
        self.reconnect_attempts += 1
        return self.reconnect_attempts <= self.max_reconnect_attempts
    
    def flush_input(self):
        """清空输入缓冲区"""
        try:
            with self.lock:
                if self.ser and self.ser.is_open:
                    self.ser.reset_input_buffer()
                    return True, "输入缓冲区已清空"
                return False, "串口未连接"
        except Exception as e:
            error_msg = f"清空缓冲区失败: {str(e)}"
            logger.error(error_msg)
            return False, error_msg
    
    def flush_output(self):
        """清空输出缓冲区"""
        try:
            with self.lock:
                if self.ser and self.ser.is_open:
                    self.ser.reset_output_buffer()
                    return True, "输出缓冲区已清空"
                return False, "串口未连接"
        except Exception as e:
            error_msg = f"清空缓冲区失败: {str(e)}"
            logger.error(error_msg)
            return False, error_msg

# 创建全局串口实例
global_serial = SerialPort()