213 lines
9.3 KiB
Python
213 lines
9.3 KiB
Python
"""@file serial_port.py
|
|
@brief Неблокирующий QSerialPort и потоковый parser GUI transport.
|
|
|
|
Реализует асинхронный интерфейс к последовательному порту с полностью
|
|
неблокирующей обработкой данных и автоматическим разбором кадров протокола.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from set_devices.protocol_capture import uart_event
|
|
|
|
from .qt_compat import QObject, Signal
|
|
from .qt_compat import QSerialPort, QSerialPortInfo
|
|
|
|
from set_devices.protocol import Frame, MessageType, build_frame
|
|
from set_devices.protocol_router import ProtocolMode, ProtocolRouter
|
|
from setprotocol.core import (
|
|
Frame as SetFrame,
|
|
FrameFlag as SetFrameFlag,
|
|
MessageType as SetMessageType,
|
|
build_frame as build_set_frame,
|
|
)
|
|
|
|
|
|
class SerialDevicePort(QObject):
|
|
"""@brief Реальный COM-порт без блокирующих ожиданий в GUI thread.
|
|
|
|
Управляет связью с устройством через последовательный порт, обеспечивая
|
|
асинхронный обмен данными и передачу кадров по протоколу GUI.
|
|
"""
|
|
|
|
connected_changed = Signal(bool, str)
|
|
frame_received = Signal(object)
|
|
tx_logged = Signal(str)
|
|
capture_event = Signal(object)
|
|
rx_logged = Signal(str)
|
|
error_occurred = Signal(str)
|
|
protocol_changed = Signal(object)
|
|
|
|
def __init__(self, parent: QObject | None = None) -> None:
|
|
super().__init__(parent)
|
|
self._serial = QSerialPort(self)
|
|
self._router = ProtocolRouter()
|
|
self._sequence = 0
|
|
self._serial.readyRead.connect(self._read_available)
|
|
self._serial.errorOccurred.connect(self._handle_error)
|
|
|
|
@staticmethod
|
|
def available_ports() -> list[str]:
|
|
"""Возвращает стабильный отсортированный список имён COM-портов."""
|
|
return [name for name, _description in SerialDevicePort.available_port_infos()]
|
|
|
|
@staticmethod
|
|
def available_port_infos() -> list[tuple[str, str]]:
|
|
"""@brief Перечисляет COM-порты вместе с описанием драйвера.
|
|
|
|
Порядок естественный: COM3 идёт перед COM10, поэтому список не
|
|
перестраивается при подключении новых преобразователей.
|
|
|
|
@return Пары «имя порта, описание устройства».
|
|
"""
|
|
infos = [
|
|
(info.portName(), info.description() or info.manufacturer())
|
|
for info in QSerialPortInfo.availablePorts()
|
|
]
|
|
return sorted(infos, key=lambda item: SerialDevicePort._sort_key(item[0]))
|
|
|
|
@staticmethod
|
|
def _sort_key(port_name: str) -> tuple[str, int, str]:
|
|
"""Разделяет имя порта на буквенный префикс и числовой индекс."""
|
|
digits = "".join(char for char in port_name if char.isdigit())
|
|
prefix = port_name[: len(port_name) - len(digits)] if digits else port_name
|
|
return (prefix.upper(), int(digits) if digits else 0, port_name)
|
|
|
|
def open(self, port_name: str, baud_rate: int) -> bool:
|
|
"""Открывает 8N1 без flow control; результат сообщает синхронную ошибку."""
|
|
if not port_name:
|
|
self.error_occurred.emit("COM-порт не выбран")
|
|
return False
|
|
self.close()
|
|
self._serial.setPortName(port_name)
|
|
self._serial.setBaudRate(baud_rate)
|
|
self._serial.setDataBits(QSerialPort.DataBits.Data8)
|
|
self._serial.setParity(QSerialPort.Parity.NoParity)
|
|
self._serial.setStopBits(QSerialPort.StopBits.OneStop)
|
|
self._serial.setFlowControl(QSerialPort.FlowControl.NoFlowControl)
|
|
if not self._serial.open(QSerialPort.OpenModeFlag.ReadWrite):
|
|
self.error_occurred.emit(self._serial.errorString())
|
|
return False
|
|
self.set_protocol_mode(ProtocolMode.AUTO)
|
|
self.connected_changed.emit(True, port_name)
|
|
return True
|
|
|
|
@property
|
|
def is_open(self) -> bool:
|
|
"""Признак открытого физического COM-порта."""
|
|
return self._serial.isOpen()
|
|
|
|
@property
|
|
def port_name(self) -> str:
|
|
"""Имя текущего COM-порта, например ``COM11``."""
|
|
return self._serial.portName()
|
|
|
|
@property
|
|
def baud_rate(self) -> int:
|
|
"""Фактически установленная скорость последовательного порта."""
|
|
return self._serial.baudRate()
|
|
|
|
def close(self) -> None:
|
|
"""Закрывает порт и не оставляет активных Qt-операций чтения."""
|
|
if not self._serial.isOpen():
|
|
return
|
|
name = self._serial.portName()
|
|
self._serial.close()
|
|
self.connected_changed.emit(False, name)
|
|
|
|
def send(self, frame: Frame) -> bool:
|
|
"""Ставит целый кадр в буфер Qt без ожидания физической передачи."""
|
|
if not self._serial.isOpen():
|
|
self.error_occurred.emit("Последовательный порт не подключён")
|
|
return False
|
|
packet = build_frame(frame)
|
|
written = self._serial.write(packet)
|
|
if written > 0:
|
|
self.capture_event.emit(uart_event('SET', 'TX', packet[:written], self.baud_rate))
|
|
if written != len(packet):
|
|
self.error_occurred.emit(self._serial.errorString())
|
|
return False
|
|
self.tx_logged.emit(packet.hex(" ").upper())
|
|
return True
|
|
|
|
def send_set(self, frame: SetFrame) -> bool:
|
|
"""Передаёт SETProtocol v2 кадр, в том числе пробный кадр AUTO."""
|
|
if not self._serial.isOpen():
|
|
self.error_occurred.emit("Последовательный порт не подключён")
|
|
return False
|
|
if self.protocol_mode is ProtocolMode.GUI_V1:
|
|
self.error_occurred.emit("Канал уже работает в режиме GUI protocol v1")
|
|
return False
|
|
packet = build_set_frame(frame)
|
|
written = self._serial.write(packet)
|
|
if written > 0:
|
|
self.capture_event.emit(uart_event('SET', 'TX', packet[:written], self.baud_rate))
|
|
if written != len(packet):
|
|
self.error_occurred.emit(self._serial.errorString())
|
|
return False
|
|
self.tx_logged.emit(packet.hex(" ").upper())
|
|
return True
|
|
|
|
def next_sequence(self) -> int:
|
|
"""Выделяет очередной sequence для адаптера транзакций."""
|
|
self._sequence = (self._sequence + 1) & 0xFFFF
|
|
return self._sequence
|
|
|
|
def send_message(self, message_type: MessageType, payload: bytes = b"") -> int | None:
|
|
"""Передаёт запрос и возвращает его sequence либо ``None``."""
|
|
sequence = self.next_sequence()
|
|
return sequence if self.send(Frame(message_type, sequence, payload)) else None
|
|
|
|
def send_set_message(
|
|
self,
|
|
message_type: int | SetMessageType,
|
|
payload: bytes = b"",
|
|
*,
|
|
flags: SetFrameFlag = SetFrameFlag.ACK_REQUIRED,
|
|
destination: int = 0,
|
|
) -> int | None:
|
|
"""Передаёт запрос SETProtocol v2 и возвращает sequence."""
|
|
sequence = self.next_sequence()
|
|
frame = SetFrame(
|
|
message_type,
|
|
sequence,
|
|
payload,
|
|
flags=flags,
|
|
destination=destination,
|
|
)
|
|
return sequence if self.send_set(frame) else None
|
|
|
|
@property
|
|
def protocol_mode(self) -> ProtocolMode:
|
|
return self._router.mode
|
|
|
|
def set_protocol_mode(self, mode: ProtocolMode) -> None:
|
|
"""Сбрасывает parser-ы и выбирает режим текущего соединения."""
|
|
changed = self._router.mode is not mode
|
|
self._router.select(mode)
|
|
if changed or mode is ProtocolMode.AUTO:
|
|
self.protocol_changed.emit(mode)
|
|
|
|
def ping(self) -> bool:
|
|
"""Отправляет совместимый PING с новым 16-битным sequence."""
|
|
return self.send_message(MessageType.PING) is not None
|
|
|
|
def _read_available(self) -> None:
|
|
"""Выгружает доступный фрагмент и передаёт parser-у без блокировки."""
|
|
data = bytes(self._serial.readAll())
|
|
if not data:
|
|
return
|
|
self.rx_logged.emit(data.hex(" ").upper())
|
|
self.capture_event.emit(uart_event('SET', 'RX', data, self.baud_rate))
|
|
previous_mode = self._router.mode
|
|
frames = self._router.feed(data)
|
|
if self._router.mode is not previous_mode:
|
|
self.protocol_changed.emit(self._router.mode)
|
|
for frame in frames:
|
|
self.frame_received.emit(frame)
|
|
|
|
def _handle_error(self, error: QSerialPort.SerialPortError) -> None:
|
|
"""Публикует только реальные ошибки, игнорируя NoError."""
|
|
if error == QSerialPort.SerialPortError.NoError:
|
|
return
|
|
self.error_occurred.emit(self._serial.errorString())
|