# Последовательный транспорт Lawicel/SLCAN для CAN-адаптера. Текстовое представление кодирует # ID, длину и данные кадра; окончания строк разделяют команды и требуют накопления неполной # строки. """Serial Line CAN (Lawicel/SLCAN) transport.""" from __future__ import annotations from dataclasses import dataclass from collections import deque # Храним ограниченную историю FD для отдельного просмотра диагностики. from ..can_adapter_settings import SlcanOptions, FD_LENGTHS # Профиль прошивки определяет команды и допустимые длины CAN FD. import re # Поиск ближайшего служебного байта за один проход сохраняет скорость при пакетном приёме большого числа CAN-кадров. from .qt_compat import QObject, Signal, QElapsedTimer, QTimer # Таймеры ограничивают ожидание ответов и откладывают выдачу кадров до окончания настройки. from .qt_compat import QSerialPort, QSerialPortInfo class SlcanError(RuntimeError): pass @dataclass(frozen=True) class SlcanChannel: path: str channel: int = 0 description: str = "" manufacturer: str = "" @property def label(self) -> str: details = self.description or self.manufacturer or "COM-порт" return f"{self.path} — {details}" @property def adapter_kind(self) -> str: return "slcan" _RECORD_END = re.compile(rb"[\r\n\x07]") # CR, LF и BEL имеют коды ASCII 13, 10 и 7; любой из них завершает запись протокола. def encode_frame(can_id: int, data: bytes, *, extended: bool | None = None) -> bytes: """Encode one classic-CAN frame in Lawicel ASCII format.""" if len(data) > 8: raise SlcanError("Classic CAN поддерживает не более 8 байт") if not 0 <= can_id <= 0x1FFFFFFF: raise SlcanError("CAN ID вне диапазона") if extended is None: extended = can_id > 0x7FF if not extended: if can_id > 0x7FF: raise SlcanError("Стандартный CAN ID вне диапазона") return f"t{can_id:03X}{len(data):X}{data.hex().upper()}\r".encode("ascii") return f"T{can_id:08X}{len(data):X}{data.hex().upper()}\r".encode("ascii") def decode_frame(line: bytes) -> tuple[int, bytes] | None: """Decode a received SLCAN data frame; ignore ACK/status/RTR records.""" if not line or line[:1] not in (b"t", b"T"): return None extended = line[:1] == b"T" id_size = 8 if extended else 3 try: can_id = int(line[1:1 + id_size], 16) size = int(line[1 + id_size:2 + id_size], 16) start = 2 + id_size payload = bytes.fromhex(line[start:start + size * 2].decode("ascii")) except (ValueError, UnicodeError): return None maximum = 0x1FFFFFFF if extended else 0x7FF if can_id > maximum or size > 8 or len(payload) != size: return None return can_id, payload class SlcanAdapter(QObject): """One SLCAN COM port using the usual 115200 8N1 host connection.""" frame_received = Signal(int, bytes) capture_received = Signal(int, bytes, bool, bool) connection_lost = Signal(str) fd_received = Signal(int, bytes, bool, int, int) # Отдельный маршрут FD исключает ошибочный разбор кадра как Classic CAN. def __init__(self, parent: QObject | None = None) -> None: super().__init__(parent) self._serial = QSerialPort(self) self._serial.readyRead.connect(self._read) self._serial.errorOccurred.connect(self._error) self._buffer = bytearray() self._listen_only = False self.options = SlcanOptions() # Без расширенных параметров остаётся прежний подтверждаемый протокол Lawicel. self.fd_history = deque(maxlen=500) # Не более 500 последних FD-кадров сохраняется между открытиями окна просмотра. self._query_prefix = None # Диагностические ответы отличаем от подтверждений настройки и CAN-кадров. self._query_reply = None # Полная строка ответа сохраняется только во время явного диагностического запроса. self._initializing = False # Открытие COM-порта ещё не подтверждает настройку CAN-контроллера. self._waiting_command = False # Отличаем подтверждение текущей команды от служебных строк рабочего потока. self._command_reply: bool | None = None # None означает ожидание, True — CR/ACK, False — BEL/отказ прошивки. self._after_cr = False # Следующий LF после CR относится к тому же окончанию строки, даже при раздельных USB-пакетах. self._pending_frames: list[tuple[int, bytes, bool]] = [] # Кадры при открытии CAN сохраняются до готовности подписчиков монитора. @property def connected(self) -> bool: return self._serial.isOpen() and not self._initializing # Считаем соединение готовым только после подтверждения всех команд настройки. def scan(self) -> list[SlcanChannel]: # Windows не предоставляет надёжного признака протокола SLCAN. # Показываем COM-кандидаты только внутри явно выбранного режима SLCAN. return [SlcanChannel(info.portName(), description=info.description(), manufacturer=info.manufacturer()) for info in QSerialPortInfo.availablePorts()] def connect_channel(self, channel: SlcanChannel, bitrate: int, host_baudrate: int = 115200, listen_only: bool = False, *, options: SlcanOptions | None = None) -> None: # Расширенные параметры передаются отдельно от прежнего контракта подключения. selected = options or SlcanOptions() # Дефолт сохраняет C, Sn и O/L с ожиданием CR/BEL. commands = selected.commands(bitrate, listen_only) # Проверяем весь профиль до закрытия прежнего соединения и открытия COM. self.disconnect_channel() self.options = selected # Настройка ожидания команд нужна уже во время инициализации порта. self._serial.setPortName(channel.path) if host_baudrate not in (9600, 19200, 38400, 57600, 115200, 230400, 460800, 921600): raise SlcanError(f"Недопустимая скорость COM: {host_baudrate}") self._serial.setBaudRate(host_baudrate) 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): raise SlcanError(f"Не удалось открыть {channel.path}: {self._serial.errorString()}") self._buffer.clear() self.fd_history.clear() # Журнал нового подключения не содержит сообщений прежнего устройства. self._listen_only = listen_only self._initializing = True # Во время настройки не выдаём ложный статус готовности и первые кадры с прежней скоростью CAN. try: # Любой отказ настройки должен закрыть порт и разрешить повторную попытку подключения. self._command("C", allow_rejection=True) # Закрываем прежний канал; BEL допустим, если канал уже закрыт по правилам Lawicel. for command in commands: # Профиль определяет порядок скорости, режима, фильтров, повторов и открытия. self._command(command) # Любой отказ прекращает настройку без опасного возврата к активному режиму. except SlcanError: # Отказ прошивки, неполная запись и отсутствие ответа одинаково означают неудачное подключение. self._serial.close() # Освобождаем COM-порт, чтобы пользователь мог исправить скорость или тип адаптера. self._buffer.clear() # Остатки неудачного обмена не должны повреждать строки следующего подключения. self._pending_frames.clear() # Не отправляем в монитор данные сеанса, который не удалось открыть. self._listen_only = False # Сбрасываем ограничение режима вместе с неудачным подключением. raise # Сохраняем точную причину отказа для статусной строки вызывающей вкладки. finally: # Признак настройки должен сбрасываться как при успехе, так и при исключении. self._initializing = False # После возврата вызывающая вкладка сможет установить правильную скорость и статус. QTimer.singleShot(0, self._read) # Выдаём накопленные первые кадры в следующем цикле Qt, после обновления настроек монитора. def _command(self, command: str, *, allow_rejection: bool = False) -> None: # Последовательный обмен сохраняет синхронный контракт подключения всех вкладок. self._read() # Убираем ранее пришедшие ответы до начала ожидания новой команды. self._command_reply = None # Старое подтверждение не может подтверждать следующую настройку. self._waiting_command = True # Пустая строка CR или отдельный BEL теперь относятся к текущей команде. packet = (command + "\r").encode("ascii") # Каждая команда Lawicel передаётся отдельной строкой с завершающим ASCII CR. elapsed = QElapsedTimer() # Монотонное время не зависит от перевода системных часов Windows. elapsed.start() # Общий срок ожидания включает запись команды и все фрагменты ответа. try: # Флаг ожидания обязательно очищается при любом результате обмена. if self._serial.write(packet) != len(packet): # Частичная локальная запись также не гарантирует передачу полной команды адаптеру. raise SlcanError(f"SLCAN: не удалось передать {command}: {self._serial.errorString()}") # Указываем неотправленную команду и причину драйвера. if self.options.profile in ("canable", "canable2"): # Эти прошивки не предоставляют CR/BEL для подтверждения конфигурации. if self._serial.bytesToWrite() and not self._serial.waitForBytesWritten(self.options.command_timeout_ms): # Ждём фактическую передачу очереди Qt в драйвер, а не выдуманный ACK. raise SlcanError(f"CANable: не удалось отправить {command}") # Ошибка локального драйвера не считается успешным подключением. self._serial.waitForReadyRead(20) # Пауза до 20 мс даёт прошивке обработать отдельную команду USB CDC. self._read() # Сохраняем уже пришедшие CAN-кадры, не требуя неподдерживаемого служебного ответа. if self._command_reply is False or not self._serial.isOpen(): # Не игнорируем явный BEL и отключение даже в профиле без ACK. raise SlcanError(f"CANable: отказ при выполнении {command}") # Пользователь получает конкретный этап неудачной настройки. return # Статус интерфейса отдельно сообщает, что профиль не подтверждает команды. while self._command_reply is None and self._serial.isOpen(): # CAN-кадры между служебными ответами не заменяют подтверждение команды. remaining = self.options.command_timeout_ms - elapsed.elapsed() # Общий выбранный тайм-аут включает все фрагменты ответа команды. if remaining <= 0: # Поток посторонних байтов не должен удерживать подключение в ожидании навсегда. break # Далее сформируем понятную ошибку отсутствия SLCAN-ответа. if self._serial.bytesAvailable() or self._serial.waitForReadyRead(remaining): # Используем уже поступившие байты или ждём только в пределах общего срока. self._read() # Тот же обработчик разбирает ответы CR/BEL и сохраняет встреченные CAN-кадры. else: # Драйвер завершил ожидание без новых данных либо порт был отключён. break # Не запускаем повторный полный тайм-аут на ту же команду. if not self._serial.isOpen(): # Физическое отключение во время настройки не является подтверждением команды. raise SlcanError(f"SLCAN: порт закрыт при выполнении {command}: {self._serial.errorString()}") # Сообщаем о потере порта на конкретном этапе. if self._command_reply is None: # Успешное открытие COM и запись в драйвер сами по себе не доказывают поддержку Lawicel. raise SlcanError(f"SLCAN: нет ответа на {command}; проверьте тип адаптера и скорость COM") # Направляем пользователя к настройкам, влияющим на обмен ASCII-командами. if not self._command_reply and not allow_rejection: # Только отказ повторного C при уже закрытом канале допускается протоколом настройки. raise SlcanError(f"SLCAN: адаптер отклонил команду {command} (BEL)") # Отказ скорости или режима CAN больше не маскируется статусом подключения. finally: # После команды подтверждения рабочего потока больше не меняют результат настройки. self._waiting_command = False # Возвращаем обработчик приёма в обычный режим. def disconnect_channel(self) -> None: if self._serial.isOpen(): # Команда закрытия нужна только для реально открытого COM-порта. self._serial.write(b"C\r") # Просим прошивку остановить CAN до освобождения порта. self._serial.waitForBytesWritten(100) # Ограничиваем ожидание отправки закрытия ста миллисекундами, чтобы не задерживать выход. self._serial.close() # Освобождаем порт независимо от доступности прошивки. self._buffer.clear() # Частичная строка прежнего сеанса не должна приклеиваться к первому кадру нового подключения. self._pending_frames.clear() # Отложенный обработчик Qt не должен выдавать кадры после отключения или переподключения. self._after_cr = False # Новый сеанс начинает разбор окончаний строк без состояния предыдущего CR. self._listen_only = False # Отключённый транспорт не сохраняет ограничение старого режима. def send(self, can_id: int, data: bytes, *, extended: bool = True) -> None: if not self.connected: raise SlcanError("SLCAN-адаптер не подключён") if self._listen_only: raise SlcanError("SLCAN-адаптер открыт в режиме только приёма") # The adapter contract is used by ProtoCAN, whose IDs are always EXT, # including the numerically small values that would also fit in 11 bits. packet = encode_frame(can_id, data, extended=extended) if self._serial.write(packet) != len(packet): raise SlcanError(f"SLCAN не передал CAN-кадр: {self._serial.errorString()}") def _read(self) -> None: if not self._serial.isOpen(): # Отложенное чтение после закрытия не должно обращаться к неоткрытому драйверу. return # Отключение уже очистило очередь и неполную строку. self._buffer.extend(bytes(self._serial.readAll())) # Накапливаем фрагменты USB/COM до полного кадра или служебного ответа. while self._buffer: # За одно чтение драйвер может вернуть несколько кадров и подтверждений команд. if self._after_cr: # CR+LF считается одним окончанием, даже если LF пришёл в следующем чтении. self._after_cr = False # Состояние относится только к ближайшему байту после CR. if self._buffer[0] == 10: # ASCII LF имеет код 10 и не является началом следующего SLCAN-кадра. del self._buffer[0] # Удаляем дополнительный LF, чтобы он не мешал распознаванию t/T и не давал второй ACK. continue # Продолжаем с первого байта следующей записи, если он уже поступил. boundary = _RECORD_END.search(self._buffer) # Ищем первый разделитель, не сканируя весь остаток пакета отдельно для отсутствующих LF и BEL. if boundary is None: # Неполную строку нельзя декодировать до поступления окончания. break # Сохраняем её в буфере для следующего readyRead. end = boundary.start() # Ближайшая граница сохраняет исходный порядок кадров и подтверждений команд. raw = bytes(self._buffer[:end]) # Получаем строку без служебного разделителя. separator = self._buffer[end] # Тип разделителя отличает подтверждение команды от BEL. del self._buffer[:end + 1] # Удаляем обработанную запись, не теряя оставшуюся часть потока. self._after_cr = separator == 13 # ASCII CR имеет код 13; следующий LF принадлежит этому же окончанию. if separator == 7: # ASCII BEL имеет код 7 и сообщает об отклонении последней команды. if self._waiting_command and self._command_reply is None: # Принимаем только первый ответ текущего этапа настройки. self._command_reply = False # Сохраняем отказ для проверки в последовательном обмене командами. continue # BEL не должен приклеиваться к следующему CAN-кадру. if self.options.fd and raw[:1] in (b"d", b"D", b"b", b"B") and self._receive_fd(raw): # Только корректная FD-строка забирается отдельным маршрутом; версия CANable может начинаться с HEX-буквы. continue # Даже FD с данными до восьми байтов не меняет тип при передаче подписчикам. frame = decode_frame(raw) # Кадры STD/EXT декодируются прежним публичным кодеком. if frame is not None: # Служебные строки и ответы прошивки не попадают в общий монитор CAN. self._pending_frames.append((frame[0], frame[1], raw[:1] == b"T")) # Сохраняем данные и признак EXT, включая первые кадры во время открытия. elif raw and self._query_prefix is not None and raw.startswith(self._query_prefix): # Диагностические строки распознаются только после исключения CAN-кадров. self._query_reply = raw.decode("ascii", errors="replace") # Версия CANable не имеет префикса V, поэтому допускается пустой ожидаемый префикс. elif not raw and self._waiting_command and self._command_reply is None: # Пустая строка CR означает ACK по Lawicel; строка с CAN-кадром не подтверждает команду. self._command_reply = True # Разрешаем отправку следующей команды только после реального подтверждения. if not self._initializing: # Первые кадры выдаются после возврата подключения, когда вкладка уже установила скорость CAN. pending, self._pending_frames = self._pending_frames, [] # Отделяем очередь до вызова подписчиков, чтобы повторное чтение не дублировало кадры. for can_id, data, extended in pending: # Выдаём все накопленные кадры в исходном порядке. self.capture_received.emit(can_id, data, extended, False) # Полные метаданные наполняют общий CAN-монитор. self.frame_received.emit(can_id, data) # Совместимый сигнал сохраняет работу существующих клиентов ProtoCAN. def _error(self, error: QSerialPort.SerialPortError) -> None: if error in (QSerialPort.SerialPortError.NoError, # Сигнал NoError не означает потерю соединения. QSerialPort.SerialPortError.NotOpenError, # Обращение после закрытия не требует повторного отключения. QSerialPort.SerialPortError.TimeoutError): # Тайм-аут синхронного ожидания проверяет сама команда; порт при этом остаётся исправным. return # Не превращаем отсутствие ответа на команду в ошибку физического соединения. if self._serial.isOpen(): message = self._serial.errorString() self._serial.close() self._listen_only = False if not self._initializing: # При настройке исключение возвращает connect_channel, избегая повторного отключения из обработчика GUI. self.connection_lost.emit(f"Связь с SLCAN прервана: {message}") # Во время обычного приёма уведомляем вкладку о настоящем обрыве. def close(self) -> None: self.disconnect_channel() def _receive_fd(self, raw: bytes) -> bool: # Результат отличает корректный FD от диагностической строки с похожей первой буквой. extended = raw[:1] in (b"D", b"B") # Заглавная буква означает идентификатор длиной восемь HEX-цифр. width = 8 if extended else 3 # Стандартный ID занимает три HEX-цифры в ASCII-протоколе. try: # Сохраняем устойчивость потока при повреждённых USB/COM-строках. identifier = int(raw[1:1 + width], 16) # Извлекаем числовой ID без префикса команды. length = FD_LENGTHS[int(raw[1 + width:2 + width], 16)] # Четырёхбитный DLC переводится в фактическое число байтов. payload = bytes.fromhex(raw[2 + width:].decode("ascii")) # Для CANable FD не документирован дополнительный суффикс временной метки. except (ValueError, UnicodeError, IndexError): # Неполный ID, DLC или HEX-пара отклоняют только текущую запись. return False # Неподходящая строка ещё может оказаться ответом версии CANable. if len(payload) != length or not 0 <= identifier <= (0x1FFFFFFF if extended else 0x7FF): # Не обрезаем лишние байты и не принимаем отрицательный ID или идентификатор другого формата. return False # Повреждённые данные не должны выглядеть успешным приёмом. record = (identifier, payload, extended, 2 | (4 if raw[:1] in (b"b", b"B") else 0), 0) # CANable не предоставляет аппаратную метку; ноль явно обозначает её отсутствие. self.fd_history.append(("RX", record)) # История ограничена размером deque и доступна после открытия диагностики. self.fd_received.emit(*record) # В мост и прикладной декодер Classic этот кадр не поступает. return True # Вызывающий обработчик не должен повторно трактовать этот кадр как служебный ответ. def send_fd(self, can_id: int, data: bytes, *, extended: bool = True, brs: bool = True) -> None: # Используем документированные команды расширения CANable 2.0. if not self.connected or self._listen_only or not self.options.fd: # Проверка защищает передачу независимо от состояния кнопок GUI. raise SlcanError("Передача FD требует активного подключения CAN FD") # Пассивный режим не допускает обхода через API. if len(data) not in FD_LENGTHS or not 0 <= can_id <= (0x1FFFFFFF if extended else 0x7FF): # Разрешаем только точные длины DLC без скрытого дополнения. raise SlcanError("Неверный CAN ID или длина CAN FD (0…8, 12, 16, 20, 24, 32, 48, 64)") # Сообщаем фактический набор допустимых размеров. prefix = ("B" if extended else "b") if brs else ("D" if extended else "d") # BRS выбирается буквой команды, а не полем ASCII-пакета. packet = f"{prefix}{can_id:0{8 if extended else 3}X}{FD_LENGTHS.index(len(data)):X}{data.hex().upper()}\r".encode("ascii") # ID, DLC и данные образуют одну завершённую строку SLCAN. if self._serial.write(packet) != len(packet): # Частичная запись не гарантирует доставку последнего CR в прошивку. raise SlcanError("SLCAN не передал CAN FD") # Причина остаётся видимой в диагностическом окне отправки. self.fd_history.append(("TX queued", (can_id, bytes(data), extended, 2 | (4 if brs else 0), 0))) # Локальная запись не выдаётся за подтверждение передачи на CAN. def diagnostics(self) -> str: # Явная диагностика не меняет режим CAN и не выполняется фоновым таймером. if not self.connected: # Чтение версии не должно захватывать ещё не выбранный COM-порт. raise SlcanError("Сначала подключите адаптер") # Настройка и жизненный цикл подключения остаются под управлением пользователя. lines = [] # Ответы V и N читают сведения, F дополнительно сбрасывает индикатор ошибки согласно Lawicel. canable = self.options.profile != "lawicel" # CANable возвращает V без буквенного префикса, а регистр E иногда без завершающего CR. if canable: # Отсутствие подтверждения настроек сообщается независимо от успешного чтения версии. lines.append(f"Профиль {self.options.profile}: команды конфигурации без ACK/NACK; применение настроек не подтверждается прошивкой.") # Диагностика не превращает ответ V в подтверждение предыдущих M/A/Y. for command in (("V", "E") if canable else ("V", "N", "F")): # Набор фиксирован спецификацией выбранной прошивки. self._read() # Сначала выдаём уже накопленные CAN-кадры, не смешивая их с ответом запроса. self._query_prefix, self._query_reply = (b"" if canable else command.encode("ascii")), None # CANable не использует префикс V в строке версии. elapsed = QElapsedTimer() # Ограничиваем полный запрос монотонным временем, а не числом фрагментов. elapsed.start() # Тайм-аут включает отправку и все поступившие строки CAN. try: # Состояние запроса очищается также при ошибке записи или обрыве порта. packet = (command + "\r").encode("ascii") # Все команды завершаются одиночным CR протокола Lawicel. if self._serial.write(packet) != len(packet): # Локальная ошибка передачи должна быть видна вызывающему коду. raise SlcanError(f"Не удалось отправить {command}") # Указываем конкретный отказавший диагностический запрос. while self._query_reply is None and self.connected: # Приём CAN-кадров продолжается тем же обработчиком. remaining = self.options.command_timeout_ms - elapsed.elapsed() # Поток кадров не должен бесконечно продлевать ожидание версии. if remaining <= 0 or not (self._serial.bytesAvailable() or self._serial.waitForReadyRead(remaining)): # Отсутствие ответа завершает только текущий запрос. break # Следующие строки диагностики покажут, какая команда не поддерживается. self._read() # Ответ распознаётся по префиксу, а CAN продолжает поступать обычным подписчикам. if command == "E" and self._buffer.startswith(b"CANable Error Register:"): # Документированная исходная прошивка может возвращать E без CR. self._query_reply = bytes(self._buffer).decode("ascii", errors="replace") # Такой ответ принимаем только после завершения ожидания, не обрезая поток CAN. self._buffer.clear() # Нетерминированная служебная строка не должна приклеиваться к следующему CAN-кадру. lines.append(self._query_reply or f"{command}: нет ответа") # Отсутствие данных не заменяем выдуманными версией или нулевыми ошибками. finally: # Следующий обычный приём не должен сохранять посторонние строки как ответ старого запроса. self._query_prefix = None # Завершаем явную диагностическую транзакцию. return "\n".join(lines) # Текст можно скопировать для анализа состояния устройства.