def demonstrate_serial_basics():
    """
    演示串口通信的基本參數(shù)和概念
    """
    serial_parameters = {
        'baudrate': '波特率 - 數(shù)據(jù)傳輸速度（bps）',
        'bytesize': '數(shù)據(jù)位 - 每個(gè)字節(jié)的位數(shù)（5, 6, 7, 8）',
        'parity': '校驗(yàn)位 - 錯(cuò)誤檢測(cè)機(jī)制（NONE, EVEN, ODD, MARK, SPACE）',
        'stopbits': '停止位 - 字節(jié)結(jié)束標(biāo)志（1, 1.5, 2）',
        'timeout': '超時(shí) - 讀操作等待時(shí)間',
        'xonxoff': '軟件流控制',
        'rtscts': '硬件流控制（RTS/CTS）',
        'dsrdtr': '硬件流控制（DSR/DTR）'
    }
    
    print("=== 串口通信基本參數(shù) ===")
    for param, description in serial_parameters.items():
        print(f"{param:10}: {description}")
    
    # 常見波特率標(biāo)準(zhǔn)
    common_baudrates = [9600, 19200, 38400, 57600, 115200, 230400, 460800, 921600]
    print(f"\n常見波特率: {common_baudrates}")
    
    # 數(shù)據(jù)格式示例
    print("\n數(shù)據(jù)格式示例:")
    print("  8N1: 8數(shù)據(jù)位, 無(wú)校驗(yàn), 1停止位")
    print("  7E1: 7數(shù)據(jù)位, 偶校驗(yàn), 1停止位")
    print("  8O2: 8數(shù)據(jù)位, 奇校驗(yàn), 2停止位")

demonstrate_serial_basics()

def compare_serial_libraries():
    """
    比較Python中不同的串口通信庫(kù)
    """
    libraries = [
        {
            'name': 'pySerial',
            'description': '最流行的串口庫(kù)，功能完整，跨平臺(tái)支持好',
            'features': ['支持所有主流操作系統(tǒng)', '豐富的API', '活躍的維護(hù)', '良好的文檔'],
            'installation': 'pip install pyserial',
            'usage': 'import serial'
        },
        {
            'name': 'serial.tools',
            'description': 'pySerial的附加工具集',
            'features': ['列出可用端口', '端口探測(cè)', '高級(jí)工具函數(shù)'],
            'installation': '包含在pySerial中',
            'usage': 'from serial.tools import list_ports'
        },
        {
            'name': 'miniterm',
            'description': 'pySerial提供的終端模擬器',
            'features': ['簡(jiǎn)單的串口終端',調(diào)試工具', '內(nèi)置在pySerial中'],
            'installation': 'python -m serial.tools.miniterm',
            'usage': '命令行工具'
        }
    ]
    
    print("=== Python串口庫(kù)比較 ===")
    for lib in libraries:
        print(f"\n{lib['name']}:")
        print(f"  描述: {lib['description']}")
        print(f"  特性: {', '.join(lib['features'])}")
        print(f"  安裝: {lib['installation']}")
        print(f"  導(dǎo)入: {lib['usage']}")

compare_serial_libraries()

import serial
import serial.tools.list_ports
from typing import Optional, List

class BasicSerialCommunicator:
    """
    基礎(chǔ)串口通信類
    """
    
    def __init__(self):
        self.ser = None
    
    def list_available_ports(self) -> List[str]:
        """
        列出所有可用串口
        """
        ports = serial.tools.list_ports.comports()
        port_list = []
        
        print("=== 可用串口列表 ===")
        for port in ports:
            info = f"{port.device}: {port.description} [{port.hwid}]"
            port_list.append(port.device)
            print(info)
        
        return port_list
    
    def connect(self, port: str, baudrate: int = 9600, 
               timeout: float = 1.0, **kwargs) -> bool:
        """
        連接到串口設(shè)備
        """
        try:
            # 默認(rèn)參數(shù)
            default_params = {
                'bytesize': serial.EIGHTBITS,
                'parity': serial.PARITY_NONE,
                'stopbits': serial.STOPBITS_ONE,
                'xonxoff': False,
                'rtscts': False,
                'dsrdtr': False
            }
            default_params.update(kwargs)
            
            self.ser = serial.Serial(
                port=port,
                baudrate=baudrate,
                timeout=timeout,
                **default_params
            )
            
            if self.ser.is_open:
                print(f"成功連接到 {port}，波特率 {baudrate}")
                return True
            else:
                print("連接失敗")
                return False
                
        except serial.SerialException as e:
            print(f"串口連接錯(cuò)誤: {e}")
            return False
        except Exception as e:
            print(f"連接異常: {e}")
            return False
    
    def send_data(self, data: bytes) -> bool:
        """
        發(fā)送數(shù)據(jù)
        """
        if not self.ser or not self.ser.is_open:
            print("串口未連接")
            return False
        
        try:
            if isinstance(data, str):
                data = data.encode('utf-8')
            
            bytes_written = self.ser.write(data)
            self.ser.flush()  # 確保數(shù)據(jù)發(fā)送完成
            print(f"發(fā)送 {bytes_written} 字節(jié): {data.hex(' ')}")
            return True
            
        except serial.SerialException as e:
            print(f"發(fā)送錯(cuò)誤: {e}")
            return False
    
    def receive_data(self, size: int = 1) -> Optional[bytes]:
        """
        接收數(shù)據(jù)
        """
        if not self.ser or not self.ser.is_open:
            print("串口未連接")
            return None
        
        try:
            data = self.ser.read(size)
            if data:
                print(f"接收 {len(data)} 字節(jié): {data.hex(' ')}")
                return data
            else:
                print("未接收到數(shù)據(jù)")
                return None
                
        except serial.SerialException as e:
            print(f"接收錯(cuò)誤: {e}")
            return None
    
    def close(self):
        """
        關(guān)閉串口連接
        """
        if self.ser and self.ser.is_open:
            self.ser.close()
            print("串口連接已關(guān)閉")

# 使用示例
def demo_basic_serial():
    """基礎(chǔ)串口通信演示"""
    communicator = BasicSerialCommunicator()
    
    # 列出可用端口
    ports = communicator.list_available_ports()
    
    if ports:
        # 嘗試連接第一個(gè)可用端口（在實(shí)際使用中應(yīng)選擇正確的端口）
        test_port = ports[0]
        print(f"\n嘗試連接: {test_port}")
        
        if communicator.connect(test_port, baudrate=9600):
            try:
                # 發(fā)送測(cè)試數(shù)據(jù)
                communicator.send_data(b'AT\r\n')  # 常見的AT命令
                
                # 接收響應(yīng)（如果有設(shè)備響應(yīng)）
                response = communicator.receive_data(64)
                if response:
                    print(f"設(shè)備響應(yīng): {response.decode('ascii', errors='ignore')}")
                
            finally:
                communicator.close()
    else:
        print("沒有找到可用串口")

# 注意：在實(shí)際環(huán)境中運(yùn)行需要真實(shí)的串口設(shè)備
# demo_basic_serial()

class AdvancedSerialHandler:
    """
    高級(jí)串口操作處理器
    """
    
    def __init__(self):
        self.ser = None
        self.read_buffer = bytearray()
        self.callbacks = {
            'on_data_received': None,
            'on_error': None,
            'on_connect': None,
            'on_disconnect': None
        }
    
    def connect_with_retry(self, port: str, baudrate: int = 9600, 
                          max_retries: int = 3, **kwargs) -> bool:
        """
        帶重試機(jī)制的連接
        """
        for attempt in range(max_retries):
            try:
                print(f"連接嘗試 {attempt + 1}/{max_retries}...")
                
                self.ser = serial.Serial(
                    port=port,
                    baudrate=baudrate,
                    **kwargs
                )
                
                if self.ser.is_open:
                    print(f"成功連接到 {port}")
                    if self.callbacks['on_connect']:
                        self.callbacks['on_connect'](port)
                    return True
                    
            except serial.SerialException as e:
                print(f"連接嘗試 {attempt + 1} 失敗: {e}")
                if attempt == max_retries - 1:
                    if self.callbacks['on_error']:
                        self.callbacks['on_error'](f"連接失敗: {e}")
                    return False
                
                # 等待后重試
                import time
                time.sleep(1)
        
        return False
    
    def configure_serial_port(self, **settings):
        """
        動(dòng)態(tài)配置串口參數(shù)
        """
        if not self.ser or not self.ser.is_open:
            print("串口未連接")
            return False
        
        try:
            # 保存當(dāng)前設(shè)置
            current_settings = {
                'baudrate': self.ser.baudrate,
                'bytesize': self.ser.bytesize,
                'parity': self.ser.parity,
                'stopbits': self.ser.stopbits,
                'timeout': self.ser.timeout
            }
            
            # 應(yīng)用新設(shè)置
            for key, value in settings.items():
                if hasattr(self.ser, key):
                    setattr(self.ser, key, value)
                    print(f"設(shè)置 {key} = {value}")
            
            print("串口配置已更新")
            return True
            
        except Exception as e:
            print(f"配置錯(cuò)誤: {e}")
            return False
    
    def read_line(self, terminator: bytes = b'\n', 
                 timeout: float = 5.0) -> Optional[bytes]:
        """
        讀取一行數(shù)據(jù)（帶超時(shí)）
        """
        if not self.ser:
            return None
        
        import time
        start_time = time.time()
        self.read_buffer.clear()
        
        while time.time() - start_time < timeout:
            if self.ser.in_waiting > 0:
                # 讀取可用數(shù)據(jù)
                data = self.ser.read(self.ser.in_waiting)
                self.read_buffer.extend(data)
                
                # 檢查終止符
                if terminator in self.read_buffer:
                    terminator_pos = self.read_buffer.find(terminator)
                    line = bytes(self.read_buffer[:terminator_pos + len(terminator)])
                    # 保留緩沖區(qū)中剩余的數(shù)據(jù)
                    self.read_buffer = self.read_buffer[terminator_pos + len(terminator):]
                    return line
            else:
                # 短暫等待以避免CPU占用過(guò)高
                time.sleep(0.01)
        
        print("讀取超時(shí)")
        return None
    
    def send_command(self, command: bytes, 
                    expected_response: bytes = None,
                    response_timeout: float = 2.0) -> bool:
        """
        發(fā)送命令并等待預(yù)期響應(yīng)
        """
        if not self.send_data(command):
            return False
        
        if expected_response:
            response = self.read_line(timeout=response_timeout)
            if response and expected_response in response:
                print("收到預(yù)期響應(yīng)")
                return True
            else:
                print("未收到預(yù)期響應(yīng)")
                return False
        
        return True
    
    def register_callback(self, event: str, callback):
        """
        注冊(cè)事件回調(diào)
        """
        if event in self.callbacks:
            self.callbacks[event] = callback
            print(f"已注冊(cè) {event} 回調(diào)")
        else:
            print(f"未知事件: {event}")

# 使用示例
def demo_advanced_serial():
    """高級(jí)串口操作演示"""
    handler = AdvancedSerialHandler()
    
    # 注冊(cè)回調(diào)
    handler.register_callback('on_connect', lambda port: print(f"連接到 {port}"))
    handler.register_callback('on_error', lambda error: print(f"錯(cuò)誤: {error}"))
    
    # 列出端口并嘗試連接
    ports = serial.tools.list_ports.comports()
    if ports:
        port_name = ports[0].device
        
        if handler.connect_with_retry(port_name, baudrate=9600, timeout=1):
            try:
                # 配置串口
                handler.configure_serial_port(
                    timeout=2.0,
                    parity=serial.PARITY_NONE
                )
                
                # 發(fā)送AT命令測(cè)試
                success = handler.send_command(
                    b'AT\r\n',
                    expected_response=b'OK',
                    response_timeout=1.0
                )
                
                print(f"命令執(zhí)行 {'成功' if success else '失敗'}")
                
            finally:
                handler.close()

# demo_advanced_serial()

import struct
import crcmod
from enum import Enum

class SerialProtocol:
    """
    自定義串口通信協(xié)議
    """
    
    class MessageType(Enum):
        COMMAND = 0x01
        RESPONSE = 0x02
        DATA = 0x03
        ERROR = 0x04
    
    def __init__(self):
        # 創(chuàng)建CRC16計(jì)算函數(shù)
        self.crc16_func = crcmod.mkCrcFun(0x18005, rev=True, initCrc=0xFFFF, xorOut=0x0000)
    
    def create_message(self, msg_type: MessageType, payload: bytes = b'') -> bytes:
        """
        創(chuàng)建協(xié)議消息
        """
        # 消息頭
        header = struct.pack('>BH', msg_type.value, len(payload))
        
        # 計(jì)算CRC
        crc_data = header + payload
        crc = self.crc16_func(crc_data)
        
        # 組裝完整消息
        message = b'\xAA\x55' + crc_data + struct.pack('>H', crc) + b'\x55\xAA'
        return message
    
    def parse_message(self, data: bytes) -> dict:
        """
        解析協(xié)議消息
        """
        # 檢查起始和結(jié)束標(biāo)志
        if not data.startswith(b'\xAA\x55') or not data.endswith(b'\x55\xAA'):
            raise ValueError("無(wú)效的消息格式")
        
        # 提取消息內(nèi)容
        message_content = data[2:-2]  # 去掉起始和結(jié)束標(biāo)志
        
        # 驗(yàn)證CRC
        received_crc = struct.unpack('>H', message_content[-2:])[0]
        calculated_crc = self.crc16_func(message_content[:-2])
        
        if received_crc != calculated_crc:
            raise ValueError("CRC校驗(yàn)失敗")
        
        # 解析消息頭
        msg_type, payload_length = struct.unpack('>BH', message_content[:3])
        payload = message_content[3:3 + payload_length]
        
        return {
            'type': SerialProtocol.MessageType(msg_type),
            'length': payload_length,
            'payload': payload,
            'crc_valid': True
        }
    
    def create_command_message(self, command_id: int, parameters: list = None) -> bytes:
        """
        創(chuàng)建命令消息
        """
        if parameters is None:
            parameters = []
        
        # 打包參數(shù)
        param_data = b''
        for param in parameters:
            if isinstance(param, int):
                param_data += struct.pack('>i', param)
            elif isinstance(param, float):
                param_data += struct.pack('>f', param)
            elif isinstance(param, str):
                param_data += param.encode('ascii')
            elif isinstance(param, bytes):
                param_data += param
        
        # 添加命令I(lǐng)D
        command_data = struct.pack('>B', command_id) + param_data
        
        return self.create_message(self.MessageType.COMMAND, command_data)
    
    def parse_response_message(self, data: bytes) -> dict:
        """
        解析響應(yīng)消息
        """
        message = self.parse_message(data)
        
        if message['type'] != self.MessageType.RESPONSE:
            raise ValueError("不是響應(yīng)消息")
        
        # 解析響應(yīng)數(shù)據(jù)
        if len(message['payload']) >= 1:
            status = message['payload'][0]
            response_data = message['payload'][1:]
            
            return {
                'status': status,
                'data': response_data,
                'original_message': message
            }
        
        raise ValueError("無(wú)效的響應(yīng)格式")

# 使用示例
def demo_protocol():
    """協(xié)議演示"""
    protocol = SerialProtocol()
    
    # 創(chuàng)建命令消息
    command_msg = protocol.create_command_message(
        0x10,
        [100, 3.14, "test"]
    )
    print(f"命令消息: {command_msg.hex(' ')}")
    
    # 解析消息（模擬接收）
    try:
        parsed = protocol.parse_message(command_msg)
        print(f"解析結(jié)果: {parsed}")
        
        # 創(chuàng)建響應(yīng)消息
        response_msg = protocol.create_message(
            SerialProtocol.MessageType.RESPONSE,
            b'\x00' + b'OK'
        )
        print(f"響應(yīng)消息: {response_msg.hex(' ')}")
        
        # 解析響應(yīng)
        response_parsed = protocol.parse_response_message(response_msg)
        print(f"響應(yīng)解析: {response_parsed}")
        
    except ValueError as e:
        print(f"協(xié)議錯(cuò)誤: {e}")

demo_protocol()

class FrameHandler:
    """
    數(shù)據(jù)幀處理與流控制
    """
    
    def __init__(self, ser_instance):
        self.ser = ser_instance
        self.buffer = bytearray()
        self.frame_delimiter = b'\x7E'  # 幀分隔符
        self.escape_character = b'\x7D'  # 轉(zhuǎn)義字符
        self.escape_map = {
            b'\x7E': b'\x7D\x5E',  # ~ -> }^
            b'\x7D': b'\x7D\x5D',  # } -> }]
            b'\x13': b'\x7D\x33',  # XON -> }3
            b'\x11': b'\x7D\x31'   # XOFF -> }1
        }
        self.reverse_escape_map = {v: k for k, v in self.escape_map.items()}
    
    def escape_data(self, data: bytes) -> bytes:
        """
        數(shù)據(jù)轉(zhuǎn)義處理
        """
        escaped_data = bytearray()
        
        for byte in data:
            byte_b = bytes([byte])
            if byte_b in self.escape_map:
                escaped_data.extend(self.escape_map[byte_b])
            else:
                escaped_data.append(byte)
        
        return bytes(escaped_data)
    
    def unescape_data(self, data: bytes) -> bytes:
        """
        數(shù)據(jù)反轉(zhuǎn)義
        """
        unescaped_data = bytearray()
        i = 0
        
        while i < len(data):
            current_byte = bytes([data[i]])
            
            if current_byte == self.escape_character and i + 1 < len(data):
                # 處理轉(zhuǎn)義序列
                escape_seq = bytes([data[i], data[i + 1]])
                if escape_seq in self.reverse_escape_map:
                    unescaped_data.extend(self.reverse_escape_map[escape_seq])
                    i += 2
                else:
                    # 無(wú)效的轉(zhuǎn)義序列
                    unescaped_data.append(data[i])
                    i += 1
            else:
                unescaped_data.append(data[i])
                i += 1
        
        return bytes(unescaped_data)
    
    def send_frame(self, data: bytes) -> bool:
        """
        發(fā)送數(shù)據(jù)幀
        """
        try:
            # 轉(zhuǎn)義數(shù)據(jù)
            escaped_data = self.escape_data(data)
            
            # 組裝幀
            frame = self.frame_delimiter + escaped_data + self.frame_delimiter
            
            # 發(fā)送幀
            self.ser.write(frame)
            self.ser.flush()
            
            print(f"發(fā)送幀: {frame.hex(' ')}")
            return True
            
        except Exception as e:
            print(f"發(fā)送幀錯(cuò)誤: {e}")
            return False
    
    def receive_frame(self, timeout: float = 5.0) -> Optional[bytes]:
        """
        接收數(shù)據(jù)幀
        """
        import time
        start_time = time.time()
        
        while time.time() - start_time < timeout:
            # 讀取可用數(shù)據(jù)
            if self.ser.in_waiting > 0:
                data = self.ser.read(self.ser.in_waiting)
                self.buffer.extend(data)
                
                # 查找完整幀
                frame_start = self.buffer.find(self.frame_delimiter)
                if frame_start != -1:
                    # 查找?guī)Y(jié)束
                    remaining_buffer = self.buffer[frame_start + 1:]
                    frame_end = remaining_buffer.find(self.frame_delimiter)
                    
                    if frame_end != -1:
                        # 提取幀數(shù)據(jù)
                        frame_data = bytes(remaining_buffer[:frame_end])
                        
                        # 從緩沖區(qū)移除已處理的數(shù)據(jù)
                        self.buffer = self.buffer[frame_start + frame_end + 2:]
                        
                        # 反轉(zhuǎn)義數(shù)據(jù)
                        try:
                            unescaped_data = self.unescape_data(frame_data)
                            print(f"接收幀: {frame_data.hex(' ')}")
                            print(f"解析數(shù)據(jù): {unescaped_data.hex(' ')}")
                            return unescaped_data
                        except Exception as e:
                            print(f"幀解析錯(cuò)誤: {e}")
                            continue
            
            # 短暫等待
            time.sleep(0.01)
        
        print("接收幀超時(shí)")
        return None

# 使用示例
def demo_frame_handler():
    """幀處理演示"""
    # 注意：需要真實(shí)的串口連接來(lái)測(cè)試
    print("幀處理功能演示 - 需要實(shí)際硬件測(cè)試")
    
    # 模擬數(shù)據(jù)
    test_data = b'\x7E\x7D\x11\x13Hello World\x7E'
    handler = FrameHandler(None)  # 傳入None僅用于演示
    
    # 測(cè)試轉(zhuǎn)義
    escaped = handler.escape_data(test_data)
    print(f"原始數(shù)據(jù): {test_data.hex(' ')}")
    print(f"轉(zhuǎn)義后: {escaped.hex(' ')}")
    
    # 測(cè)試反轉(zhuǎn)義
    unescaped = handler.unescape_data(escaped)
    print(f"反轉(zhuǎn)義后: {unescaped.hex(' ')}")
    print(f"數(shù)據(jù)恢復(fù): {test_data == unescaped}")

demo_frame_handler()

class GPSDataParser:
    """
    GPS數(shù)據(jù)解析器（NMEA協(xié)議）
    """
    
    def __init__(self):
        self.sentence_handlers = {
            'GPGGA': self._parse_gpgga,
            'GPRMC': self._parse_gprmc,
            'GPGSA': self._parse_gpgsa,
            'GPGSV': self._parse_gpgsv
        }
        self.current_data = {}
    
    def parse_nmea_sentence(self, sentence: str) -> Optional[dict]:
        """
        解析NMEA語(yǔ)句
        """
        # 檢查基本格式
        if not sentence.startswith('$') or '*' not in sentence:
            return None
        
        # 分離數(shù)據(jù)和校驗(yàn)和
        data_part, checksum_part = sentence[1:].split('*', 1)
        calculated_checksum = self._calculate_nmea_checksum(data_part)
        
        # 驗(yàn)證校驗(yàn)和
        try:
            received_checksum = int(checksum_part, 16)
            if calculated_checksum != received_checksum:
                print(f"校驗(yàn)和錯(cuò)誤: 計(jì)算值={calculated_checksum:02X}, 接收值={received_checksum:02X}")
                return None
        except ValueError:
            return None
        
        # 分離字段
        fields = data_part.split(',')
        sentence_type = fields[0]
        
        # 調(diào)用相應(yīng)的解析器
        if sentence_type in self.sentence_handlers:
            try:
                result = self.sentence_handlers[sentence_type](fields)
                result['sentence_type'] = sentence_type
                result['timestamp'] = time.time()
                return result
            except Exception as e:
                print(f"解析 {sentence_type} 錯(cuò)誤: {e}")
                return None
        
        return None
    
    def _calculate_nmea_checksum(self, data: str) -> int:
        """
        計(jì)算NMEA校驗(yàn)和
        """
        checksum = 0
        for char in data:
            checksum ^= ord(char)
        return checksum
    
    def _parse_gpgga(self, fields: list) -> dict:
        """解析GPGGA語(yǔ)句（全球定位系統(tǒng)定位數(shù)據(jù)）"""
        if len(fields) < 15:
            raise ValueError("無(wú)效的GPGGA格式")
        
        return {
            'type': 'position',
            'time': fields[1],
            'latitude': self._parse_latitude(fields[2], fields[3]),
            'longitude': self._parse_longitude(fields[4], fields[5]),
            'quality': int(fields[6]),
            'satellites': int(fields[7]),
            'hdop': float(fields[8]) if fields[8] else 0.0,
            'altitude': float(fields[9]) if fields[9] else 0.0,
            'geoidal_separation': float(fields[11]) if fields[11] else 0.0
        }
    
    def _parse_gprmc(self, fields: list) -> dict:
        """解析GPRMC語(yǔ)句（推薦最小定位信息）"""
        if len(fields) < 12:
            raise ValueError("無(wú)效的GPRMC格式")
        
        return {
            'type': 'navigation',
            'time': fields[1],
            'status': fields[2],
            'latitude': self._parse_latitude(fields[3], fields[4]),
            'longitude': self._parse_longitude(fields[5], fields[6]),
            'speed': float(fields[7]) if fields[7] else 0.0,
            'course': float(fields[8]) if fields[8] else 0.0,
            'date': fields[9],
            'variation': float(fields[10]) if fields[10] else 0.0
        }
    
    def _parse_latitude(self, value: str, direction: str) -> float:
        """解析緯度"""
        if not value or len(value) < 4:
            return 0.0
        
        degrees = float(value[:2])
        minutes = float(value[2:])
        latitude = degrees + minutes / 60.0
        
        if direction == 'S':
            latitude = -latitude
        
        return latitude
    
    def _parse_longitude(self, value: str, direction: str) -> float:
        """解析經(jīng)度"""
        if not value or len(value) < 5:
            return 0.0
        
        degrees = float(value[:3])
        minutes = float(value[3:])
        longitude = degrees + minutes / 60.0
        
        if direction == 'W':
            longitude = -longitude
        
        return longitude
    
    def _parse_gpgsa(self, fields: list) -> dict:
        """解析GPGSA語(yǔ)句（精度因子）"""
        return {
            'type': 'accuracy',
            'mode': fields[1],
            'fix_type': int(fields[2]),
            'pdop': float(fields[15]) if fields[15] else 0.0,
            'hdop': float(fields[16]) if fields[16] else 0.0,
            'vdop': float(fields[17]) if fields[17] else 0.0
        }
    
    def _parse_gpgsv(self, fields: list) -> dict:
        """解析GPGSV語(yǔ)句（衛(wèi)星狀態(tài)）"""
        return {
            'type': 'satellites',
            'total_messages': int(fields[1]),
            'message_number': int(fields[2]),
            'satellites_in_view': int(fields[3])
        }

# 使用示例
def demo_gps_parser():
    """GPS解析器演示"""
    parser = GPSDataParser()
    
    # 示例NMEA語(yǔ)句
    test_sentences = [
        "$GPGGA,092750.000,5321.6802,N,00630.3372,W,1,8,1.03,61.7,M,55.2,M,,*76",
        "$GPRMC,092750.000,A,5321.6802,N,00630.3372,W,0.02,31.66,280511,,,A*43",
        "$GPGSA,A,3,31,32,11,18,22,26,28,29,,,,,2.01,1.03,1.70 * 07",
        "$GPGSV,3,1,11,31,70,083,42,32,58,308,42,11,54,050,42,18,46,291,42 * 7D"
    ]
    
    for sentence in test_sentences:
        print(f"\n解析: {sentence}")
        result = parser.parse_nmea_sentence(sentence)
        if result:
            print(f"結(jié)果: {result}")
        else:
            print("解析失敗")

demo_gps_parser()

class IndustrialDeviceController:
    """
    工業(yè)設(shè)備控制器（Modbus RTU協(xié)議示例）
    """
    
    def __init__(self, serial_port):
        self.ser = serial_port
        self.device_address = 1
        self.transaction_id = 0
    
    def _create_modbus_frame(self, function_code: int, data: bytes) -> bytes:
        """
        創(chuàng)建Modbus RTU幀
        """
        # 事務(wù)ID（簡(jiǎn)單遞增）
        self.transaction_id = (self.transaction_id + 1) % 256
        
        # 組裝PDU
        pdu = struct.pack('>BB', self.device_address, function_code) + data
        
        # 計(jì)算CRC
        crc = self._calculate_crc16(pdu)
        
        return pdu + struct.pack('<H', crc)
    
    def _calculate_crc16(self, data: bytes) -> int:
        """
        計(jì)算Modbus CRC16
        """
        crc = 0xFFFF
        for byte in data:
            crc ^= byte
            for _ in range(8):
                if crc & 0x0001:
                    crc = (crc >> 1) ^ 0xA001
                else:
                    crc = crc >> 1
        return crc
    
    def read_holding_registers(self, start_address: int, count: int) -> Optional[list]:
        """
        讀取保持寄存器
        """
        try:
            # 創(chuàng)建請(qǐng)求
            request_data = struct.pack('>HH', start_address, count)
            frame = self._create_modbus_frame(0x03, request_data)
            
            # 發(fā)送請(qǐng)求
            self.ser.write(frame)
            self.ser.flush()
            
            # 讀取響應(yīng)
            response = self._read_modbus_response()
            if response and len(response) >= 5 + count * 2:
                # 解析響應(yīng)
                byte_count = response[2]
                register_data = response[3:3 + byte_count]
                
                # 提取寄存器值
                registers = []
                for i in range(0, byte_count, 2):
                    if i + 1 < byte_count:
                        value = struct.unpack('>H', register_data[i:i+2])[0]
                        registers.append(value)
                
                return registers
            
            return None
            
        except Exception as e:
            print(f"讀取寄存器錯(cuò)誤: {e}")
            return None
    
    def write_single_register(self, address: int, value: int) -> bool:
        """
        寫入單個(gè)寄存器
        """
        try:
            # 創(chuàng)建請(qǐng)求
            request_data = struct.pack('>HH', address, value)
            frame = self._create_modbus_frame(0x06, request_data)
            
            # 發(fā)送請(qǐng)求
            self.ser.write(frame)
            self.ser.flush()
            
            # 讀取響應(yīng)（應(yīng)該與請(qǐng)求相同）
            response = self._read_modbus_response()
            if response and len(response) >= 6:
                # 驗(yàn)證響應(yīng)
                resp_address = struct.unpack('>H', response[2:4])[0]
                resp_value = struct.unpack('>H', response[4:6])[0]
                
                return resp_address == address and resp_value == value
            
            return False
            
        except Exception as e:
            print(f"寫入寄存器錯(cuò)誤: {e}")
            return False
    
    def _read_modbus_response(self, timeout: float = 1.0) -> Optional[bytes]:
        """
        讀取Modbus響應(yīng)
        """
        import time
        start_time = time.time()
        buffer = bytearray()
        
        while time.time() - start_time < timeout:
            if self.ser.in_waiting > 0:
                data = self.ser.read(self.ser.in_waiting)
                buffer.extend(data)
                
                # 檢查最小長(zhǎng)度
                if len(buffer) >= 5:  # 地址+功能碼+字節(jié)數(shù)+最小數(shù)據(jù)+CRC
                    # 嘗試解析
                    if len(buffer) >= 5 + buffer[2] + 2:  # 完整幀
                        return bytes(buffer)
            
            time.sleep(0.01)
        
        return None

# 使用示例
def demo_industrial_control():
    """工業(yè)控制演示"""
    print("工業(yè)設(shè)備控制器演示 - 需要實(shí)際Modbus設(shè)備測(cè)試")
    
    # 模擬用法
    controller = IndustrialDeviceController(None)
    
    # 測(cè)試CRC計(jì)算
    test_data = b'\x01\x03\x00\x00\x00\x02'
    crc = controller._calculate_crc16(test_data)
    print(f"測(cè)試數(shù)據(jù): {test_data.hex(' ')}")
    print(f"CRC16: {crc:04X}")
    
    # 測(cè)試幀創(chuàng)建
    frame = controller._create_modbus_frame(0x03, b'\x00\x00\x00\x02')
    print(f"Modbus幀: {frame.hex(' ')}")

demo_industrial_control()

class OptimizedSerialHandler:
    """
    優(yōu)化的串口處理器
    """
    
    def __init__(self):
        self.ser = None
        self.read_thread = None
        self.running = False
        self.data_queue = []
        self.error_count = 0
        self.max_errors = 10
        self.last_activity = 0
    
    def start_async_reading(self):
        """
        啟動(dòng)異步讀取線程
        """
        if not self.ser or not self.ser.is_open:
            print("串口未連接")
            return False
        
        self.running = True
        self.read_thread = threading.Thread(target=self._read_loop, daemon=True)
        self.read_thread.start()
        print("異步讀取已啟動(dòng)")
        return True
    
    def _read_loop(self):
        """
        異步讀取循環(huán)
        """
        while self.running:
            try:
                if self.ser.in_waiting > 0:
                    data = self.ser.read(self.ser.in_waiting)
                    self.data_queue.append((time.time(), data))
                    self.last_activity = time.time()
                    
                    # 處理數(shù)據(jù)（在實(shí)際應(yīng)用中這里可以調(diào)用回調(diào)函數(shù)）
                    if len(self.data_queue) > 1000:  # 防止隊(duì)列過(guò)大
                        self.data_queue.pop(0)
                
                # 檢查超時(shí)
                if time.time() - self.last_activity > 30.0:  # 30秒無(wú)活動(dòng)
                    print("串口通信超時(shí)")
                    self.handle_error("通信超時(shí)")
                
                time.sleep(0.001)  # 減少CPU占用
                
            except Exception as e:
                self.handle_error(f"讀取循環(huán)錯(cuò)誤: {e}")
                time.sleep(1)  # 錯(cuò)誤后等待
    
    def handle_error(self, error_msg: str):
        """
        錯(cuò)誤處理
        """
        self.error_count += 1
        print(f"錯(cuò)誤 ({self.error_count}/{self.max_errors}): {error_msg}")
        
        if self.error_count >= self.max_errors:
            print("錯(cuò)誤次數(shù)過(guò)多，停止讀取")
            self.stop_async_reading()
    
    def stop_async_reading(self):
        """
        停止異步讀取
        """
        self.running = False
        if self.read_thread and self.read_thread.is_alive():
            self.read_thread.join(timeout=2.0)
        print("異步讀取已停止")
    
    def get_queued_data(self, clear: bool = True) -> list:
        """
        獲取隊(duì)列中的數(shù)據(jù)
        """
        data = self.data_queue.copy()
        if clear:
            self.data_queue.clear()
        return data
    
    def send_with_retry(self, data: bytes, max_retries: int = 3) -> bool:
        """
        帶重試的發(fā)送
        """
        for attempt in range(max_retries):
            try:
                if self.send_data(data):
                    return True
                else:
                    print(f"發(fā)送嘗試 {attempt + 1} 失敗")
            except Exception as e:
                print(f"發(fā)送異常: {e}")
            
            if attempt < max_retries - 1:
                time.sleep(0.1 * (attempt + 1))  # 遞增等待
        
        return False
    
    def monitor_serial_health(self):
        """
        監(jiān)控串口健康狀態(tài)
        """
        stats = {
            'error_count': self.error_count,
            'data_queue_size': len(self.data_queue),
            'last_activity': time.time() - self.last_activity,
            'is_running': self.running,
            'is_connected': self.ser and self.ser.is_open
        }
        
        print("串口健康狀態(tài):")
        for key, value in stats.items():
            print(f"  {key}: {value}")
        
        return stats

# 使用示例
def demo_optimized_handler():
    """優(yōu)化處理器演示"""
    print("優(yōu)化串口處理器演示")
    
    handler = OptimizedSerialHandler()
    
    # 模擬健康檢查
    stats = handler.monitor_serial_health()
    
    # 測(cè)試錯(cuò)誤處理
    handler.handle_error("測(cè)試錯(cuò)誤")
    handler.handle_error("另一個(gè)測(cè)試錯(cuò)誤")
    
    print(f"當(dāng)前錯(cuò)誤計(jì)數(shù): {handler.error_count}")

demo_optimized_handler()

def cross_platform_serial_considerations():
    """
    跨平臺(tái)串口兼容性考慮
    """
    considerations = {
        'Windows': [
            'COM端口命名: COM1, COM2, ... COM256',
            '需要管理員權(quán)限訪問(wèn)某些COM端口',
            'USB轉(zhuǎn)串口設(shè)備可能需要安裝驅(qū)動(dòng)程序',
            '注意COM端口號(hào)可能隨USB端口變化'
        ],
        'Linux': [
            '設(shè)備文件: /dev/ttyS0, /dev/ttyUSB0, /dev/ttyACM0',
            '用戶需要屬于dialout組或有相應(yīng)權(quán)限',
            'USB設(shè)備通常自動(dòng)檢測(cè)',
            '可能需要udev規(guī)則配置權(quán)限'
        ],
        'macOS': [
            '設(shè)備文件: /dev/tty.usbserial*, /dev/tty.usbmodem*',
            '權(quán)限管理通過(guò)系統(tǒng)偏好設(shè)置',
            'USB轉(zhuǎn)串口驅(qū)動(dòng)可能需要單獨(dú)安裝',
            '設(shè)備命名可能包含廠商信息'
        ],
        '通用最佳實(shí)踐': [
            '使用pySerial的自動(dòng)端口檢測(cè)功能',
            '實(shí)現(xiàn)自動(dòng)波特率檢測(cè)（如果協(xié)議支持）',
            '處理權(quán)限錯(cuò)誤并提供用戶指導(dǎo)',
            '實(shí)現(xiàn)設(shè)備熱插拔檢測(cè)',
            '使用跨平臺(tái)路徑處理'
        ]
    }
    
    print("=== 跨平臺(tái)串口兼容性 ===")
    for platform, notes in considerations.items():
        print(f"\n{platform}:")
        for note in notes:
            print(f"  ? {note}")
    
    # 演示跨平臺(tái)端口列表
    try:
        ports = serial.tools.list_ports.comports()
        print(f"\n檢測(cè)到的串口設(shè)備:")
        for port in ports:
            print(f"  {port.device}: {port.description}")
    except Exception as e:
        print(f"端口檢測(cè)錯(cuò)誤: {e}")

cross_platform_serial_considerations()