"""不依赖pyserial的Windows原生串口Modbus RTU客户端。""" from __future__ import annotations import ctypes import queue import threading import time import winreg from ctypes import wintypes from dataclasses import dataclass from typing import List, Optional, Tuple GENERIC_READ = 0x80000000 GENERIC_WRITE = 0x40000000 OPEN_EXISTING = 3 FILE_ATTRIBUTE_NORMAL = 0x80 PURGE_TXABORT = 0x0001 PURGE_RXABORT = 0x0002 PURGE_TXCLEAR = 0x0004 PURGE_RXCLEAR = 0x0008 INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value class DCB(ctypes.Structure): _fields_ = [ ("DCBlength", wintypes.DWORD), ("BaudRate", wintypes.DWORD), ("flags", wintypes.DWORD), ("wReserved", wintypes.WORD), ("XonLim", wintypes.WORD), ("XoffLim", wintypes.WORD), ("ByteSize", wintypes.BYTE), ("Parity", wintypes.BYTE), ("StopBits", wintypes.BYTE), ("XonChar", ctypes.c_char), ("XoffChar", ctypes.c_char), ("ErrorChar", ctypes.c_char), ("EofChar", ctypes.c_char), ("EvtChar", ctypes.c_char), ("wReserved1", wintypes.WORD), ] class COMMTIMEOUTS(ctypes.Structure): _fields_ = [ ("ReadIntervalTimeout", wintypes.DWORD), ("ReadTotalTimeoutMultiplier", wintypes.DWORD), ("ReadTotalTimeoutConstant", wintypes.DWORD), ("WriteTotalTimeoutMultiplier", wintypes.DWORD), ("WriteTotalTimeoutConstant", wintypes.DWORD), ] def list_serial_ports() -> List[str]: """从Windows注册表枚举COM端口,不需要pyserial。""" ports = set() path = r"HARDWARE\DEVICEMAP\SERIALCOMM" try: with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, path) as key: index = 0 while True: try: _, value, _ = winreg.EnumValue(key, index) ports.add(str(value)) index += 1 except OSError: break except OSError: pass def sort_key(name: str) -> Tuple[str, int]: prefix = "".join(ch for ch in name if not ch.isdigit()) digits = "".join(ch for ch in name if ch.isdigit()) return prefix, int(digits or 0) return sorted(ports, key=sort_key) class WinSerialPort: """使用同步Win32 API访问串口;由后台线程调用。""" def __init__(self) -> None: self._kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) self._handle: Optional[int] = None self._configure_signatures() def _configure_signatures(self) -> None: kernel = self._kernel32 kernel.CreateFileW.argtypes = [ wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE, ] kernel.CreateFileW.restype = wintypes.HANDLE kernel.CloseHandle.argtypes = [wintypes.HANDLE] kernel.CloseHandle.restype = wintypes.BOOL kernel.BuildCommDCBW.argtypes = [wintypes.LPCWSTR, ctypes.POINTER(DCB)] kernel.BuildCommDCBW.restype = wintypes.BOOL kernel.SetCommState.argtypes = [wintypes.HANDLE, ctypes.POINTER(DCB)] kernel.SetCommState.restype = wintypes.BOOL kernel.SetCommTimeouts.argtypes = [ wintypes.HANDLE, ctypes.POINTER(COMMTIMEOUTS), ] kernel.SetCommTimeouts.restype = wintypes.BOOL kernel.SetupComm.argtypes = [wintypes.HANDLE, wintypes.DWORD, wintypes.DWORD] kernel.SetupComm.restype = wintypes.BOOL kernel.PurgeComm.argtypes = [wintypes.HANDLE, wintypes.DWORD] kernel.PurgeComm.restype = wintypes.BOOL kernel.ReadFile.argtypes = [ wintypes.HANDLE, wintypes.LPVOID, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), wintypes.LPVOID, ] kernel.ReadFile.restype = wintypes.BOOL kernel.WriteFile.argtypes = [ wintypes.HANDLE, wintypes.LPCVOID, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), wintypes.LPVOID, ] kernel.WriteFile.restype = wintypes.BOOL @property def is_open(self) -> bool: return self._handle is not None def open(self, port_name: str, baud_rate: int = 115200) -> None: self.close() device_name = rf"\\.\{port_name}" handle = self._kernel32.CreateFileW( device_name, GENERIC_READ | GENERIC_WRITE, 0, None, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, None, ) if handle == INVALID_HANDLE_VALUE: self._raise_last_error(f"打开{port_name}失败") self._handle = handle try: dcb = DCB() dcb.DCBlength = ctypes.sizeof(DCB) config = f"baud={baud_rate} parity=e data=8 stop=1" if not self._kernel32.BuildCommDCBW(config, ctypes.byref(dcb)): self._raise_last_error("生成串口参数失败") if not self._kernel32.SetCommState(self._handle, ctypes.byref(dcb)): self._raise_last_error("设置串口参数失败") timeouts = COMMTIMEOUTS( ReadIntervalTimeout=20, ReadTotalTimeoutMultiplier=0, ReadTotalTimeoutConstant=20, WriteTotalTimeoutMultiplier=0, WriteTotalTimeoutConstant=300, ) if not self._kernel32.SetCommTimeouts( self._handle, ctypes.byref(timeouts) ): self._raise_last_error("设置串口超时失败") if not self._kernel32.SetupComm(self._handle, 4096, 4096): self._raise_last_error("设置串口缓冲区失败") self.purge() except Exception: self.close() raise def close(self) -> None: if self._handle is not None: self._kernel32.CloseHandle(self._handle) self._handle = None def purge(self) -> None: self._require_open() flags = PURGE_TXABORT | PURGE_RXABORT | PURGE_TXCLEAR | PURGE_RXCLEAR if not self._kernel32.PurgeComm(self._handle, flags): self._raise_last_error("清理串口缓冲区失败") def write(self, data: bytes) -> None: self._require_open() buffer = ctypes.create_string_buffer(data) written = wintypes.DWORD(0) if not self._kernel32.WriteFile( self._handle, buffer, len(data), ctypes.byref(written), None, ): self._raise_last_error("串口发送失败") if written.value != len(data): raise OSError(f"串口只发送了{written.value}/{len(data)}字节") def read(self, maximum: int = 256) -> bytes: self._require_open() buffer = ctypes.create_string_buffer(maximum) received = wintypes.DWORD(0) if not self._kernel32.ReadFile( self._handle, buffer, maximum, ctypes.byref(received), None, ): self._raise_last_error("串口接收失败") return buffer.raw[: received.value] def _require_open(self) -> None: if self._handle is None: raise OSError("串口尚未打开") @staticmethod def _raise_last_error(prefix: str) -> None: code = ctypes.get_last_error() raise OSError(code, f"{prefix}:{ctypes.FormatError(code).strip()}") def crc16(data: bytes) -> int: crc = 0xFFFF for byte in data: crc ^= byte for _ in range(8): crc = ((crc >> 1) ^ 0xA001) if (crc & 1) else (crc >> 1) return crc & 0xFFFF def add_crc(data: bytes) -> bytes: crc = crc16(data) return data + bytes((crc & 0xFF, (crc >> 8) & 0xFF)) def valid_crc(frame: bytes) -> bool: if len(frame) < 4: return False received = frame[-2] | (frame[-1] << 8) return crc16(frame[:-2]) == received def u16(value: int) -> bytes: return bytes(((value >> 8) & 0xFF, value & 0xFF)) def read_u16(data: bytes, offset: int) -> int: return (data[offset] << 8) | data[offset + 1] def build_read_holding(slave: int, address: int, quantity: int) -> bytes: if not 1 <= slave <= 247: raise ValueError("从站地址必须在1到247之间") if not 1 <= quantity <= 125: raise ValueError("读取数量必须在1到125之间") return add_crc(bytes((slave, 0x03)) + u16(address) + u16(quantity)) def build_write_single(slave: int, address: int, value: int) -> bytes: if not 1 <= slave <= 247: raise ValueError("从站地址必须在1到247之间") return add_crc(bytes((slave, 0x06)) + u16(address) + u16(value)) def build_write_multiple(slave: int, address: int, values: List[int]) -> bytes: if not 1 <= slave <= 247: raise ValueError("从站地址必须在1到247之间") if not 1 <= len(values) <= 123: raise ValueError("写入数量必须在1到123之间") payload = b"".join(u16(value) for value in values) body = ( bytes((slave, 0x10)) + u16(address) + u16(len(values)) + bytes((len(payload),)) + payload ) return add_crc(body) @dataclass(frozen=True) class Request: frame: bytes function: int address: int quantity_or_value: int context: str class ModbusClient: """单事务后台客户端;GUI通过events队列接收结果。""" # STM32从站由RTOS任务轮询。连续请求之间留出恢复时间,避免上一帧刚发送 # 完成时下一帧已经进入USART,导致从站状态机漏掉请求。 INTER_REQUEST_DELAY_S = 0.03 RESPONSE_TIMEOUT_S = 0.8 MAX_ATTEMPTS = 2 def __init__(self) -> None: self.events: "queue.Queue[tuple]" = queue.Queue() self._requests: "queue.Queue[Optional[Request]]" = queue.Queue() self._serial = WinSerialPort() self._thread: Optional[threading.Thread] = None self._stop_event = threading.Event() self._busy_event = threading.Event() self.slave_address = 1 self.port_name = "" self._last_request_finished = 0.0 @property def is_open(self) -> bool: return self._serial.is_open @property def is_busy(self) -> bool: return self._busy_event.is_set() or not self._requests.empty() def open(self, port_name: str, baud_rate: int, slave_address: int) -> None: if not 1 <= slave_address <= 247: raise ValueError("从站地址必须在1到247之间") self.close() self._serial.open(port_name, baud_rate) self.slave_address = slave_address self.port_name = port_name self._stop_event.clear() self._thread = threading.Thread( target=self._worker_loop, name="ModbusRtuWorker", daemon=True, ) self._thread.start() def close(self) -> None: self._stop_event.set() if self._thread is not None and self._thread.is_alive(): self._requests.put(None) self._thread.join(timeout=1.0) self._thread = None self._busy_event.clear() while not self._requests.empty(): try: self._requests.get_nowait() except queue.Empty: break self._serial.close() self._last_request_finished = 0.0 def read_holding(self, address: int, quantity: int, context: str = "") -> None: frame = build_read_holding(self.slave_address, address, quantity) self._requests.put(Request(frame, 0x03, address, quantity, context)) def write_single(self, address: int, value: int, context: str = "") -> None: frame = build_write_single(self.slave_address, address, value) self._requests.put(Request(frame, 0x06, address, value, context)) def write_multiple( self, address: int, values: List[int], context: str = "" ) -> None: frame = build_write_multiple(self.slave_address, address, values) self._requests.put(Request(frame, 0x10, address, len(values), context)) def _worker_loop(self) -> None: while not self._stop_event.is_set(): try: request = self._requests.get(timeout=0.1) except queue.Empty: continue if request is None: break self._busy_event.set() try: for attempt in range(1, self.MAX_ATTEMPTS + 1): try: self._wait_inter_request_gap() self._execute(request) break except TimeoutError: if attempt >= self.MAX_ATTEMPTS: raise self.events.put(("retry", request.context, attempt + 1)) except Exception as exc: # 将后台异常送回GUI线程 self.events.put(("error", request.context, str(exc))) finally: self._last_request_finished = time.monotonic() self._busy_event.clear() def _wait_inter_request_gap(self) -> None: remaining = ( self._last_request_finished + self.INTER_REQUEST_DELAY_S - time.monotonic() ) if remaining > 0: time.sleep(remaining) def _execute(self, request: Request) -> None: self._serial.purge() self._serial.write(request.frame) self.events.put(("tx", request.frame)) deadline = time.monotonic() + self.RESPONSE_TIMEOUT_S response = bytearray() expected_length: Optional[int] = None while time.monotonic() < deadline and not self._stop_event.is_set(): chunk = self._serial.read(256) if chunk: response.extend(chunk) if len(response) >= 2: function = response[1] if function == (request.function | 0x80): expected_length = 5 elif function == 0x03 and len(response) >= 3: expected_length = 5 + response[2] elif function in (0x06, 0x10): expected_length = 8 if expected_length is not None and len(response) >= expected_length: break if expected_length is None or len(response) < expected_length: raise TimeoutError("等待从站响应超时") frame = bytes(response[:expected_length]) self.events.put(("rx", frame)) if not valid_crc(frame): raise ValueError("响应CRC校验失败") if frame[0] != self.slave_address: raise ValueError("响应从站地址不匹配") function = frame[1] if function == (request.function | 0x80): self.events.put( ("exception", request.function, frame[2], request.context) ) return if function != request.function: raise ValueError("响应功能码不匹配") if function == 0x03: byte_count = frame[2] if byte_count != request.quantity_or_value * 2: raise ValueError("读取响应数据长度不匹配") values = [ read_u16(frame, offset) for offset in range(3, 3 + byte_count, 2) ] self.events.put( ("read", request.address, values, request.context) ) return echoed_address = read_u16(frame, 2) echoed_value = read_u16(frame, 4) if ( echoed_address != request.address or echoed_value != request.quantity_or_value ): raise ValueError("写响应回显与请求不一致") self.events.put( ( "write", function, echoed_address, echoed_value, request.context, ) )