From f1a7bcd8bc52fed90ec51b77afb4a0c526e21418 Mon Sep 17 00:00:00 2001 From: Andrey Date: Fri, 9 Oct 2026 16:41:47 +0300 Subject: [PATCH] Support shared serial ports and optional verification in TMS flashing --- python/protocan/tms_firmware.py | 3 ++ python/set_devices/qt_ports/tms_bootloader.py | 46 +++++++++++++++++-- 2 files changed, 45 insertions(+), 4 deletions(-) diff --git a/python/protocan/tms_firmware.py b/python/protocan/tms_firmware.py index 2531ca9..081d896 100644 --- a/python/protocan/tms_firmware.py +++ b/python/protocan/tms_firmware.py @@ -73,6 +73,7 @@ class TmsTarget: kind: str = "tms" board: int = 0 load_only: bool = False + verify: bool = True def validate(self, size: int) -> None: if not 0 <= self.controller <= 255: @@ -116,6 +117,8 @@ def programming_steps(data: bytes, target: TmsTarget): raise ValueError(f"Проверка EEPROM не пройдена: код {error}, обработано {words} слов") return yield tflash(t.controller, t.ram, t.flash, len(data)), 6, 180000, "CMD_TFLASH: запись Flash", 85 + if not t.verify: + return for offset in range(0, len(data), 256): block = data[offset:offset + 256] count = (len(block) + 1) & ~1 diff --git a/python/set_devices/qt_ports/tms_bootloader.py b/python/set_devices/qt_ports/tms_bootloader.py index 510dffe..d338f1a 100644 --- a/python/set_devices/qt_ports/tms_bootloader.py +++ b/python/set_devices/qt_ports/tms_bootloader.py @@ -11,12 +11,15 @@ from protocan import tms_firmware as wire class TmsBootloader(QObject): progress = Signal(int, str) finished = Signal(bool, str) + request_sent = Signal(bytes) def __init__(self, parent=None): super().__init__(parent) self._serial = QSerialPort(self) self._serial.readyRead.connect(self._read) self._serial.errorOccurred.connect(self._error) + self._owned_serial = self._serial + self._owns_serial = True self._timeout = QTimer(self) self._timeout.setSingleShot(True) self._timeout.timeout.connect(lambda: self._finish(False, f"Тайм-аут {self._stage}")) @@ -31,6 +34,7 @@ class TmsBootloader(QObject): self._active = False self._waiting = False self._rx = bytearray() + self.readback = bytearray() self._stage = "" def start(self, image, port: str, baud: int, target: wire.TmsTarget): @@ -45,6 +49,8 @@ class TmsBootloader(QObject): return self._image, self._target = image, target self._active = True + self._serial = self._owned_serial + self._owns_serial = True self._serial.setPortName(port) self._serial.setBaudRate(baud) self._serial.setDataBits(QSerialPort.DataBits.Data8) @@ -54,8 +60,29 @@ class TmsBootloader(QObject): if not self._serial.open(QSerialPort.OpenModeFlag.ReadWrite): self._finish(False, self._serial.errorString()) return + self._begin() + + def start_connected(self, image, serial, target: wire.TmsTarget): + """Use an existing port; its owner must forward RX and handle errors.""" + if self._active: + return + try: + target.validate(len(image.data)) + if not serial.isOpen() or serial.baudRate() <= 0: + raise ValueError("Подключите COM-порт в настройках 2812") + except ValueError as error: + self.finished.emit(False, str(error)) + return + self._image, self._target = image, target + self._serial = serial + self._owns_serial = False + self._active = True + self._begin() + + def _begin(self): + self.readback.clear() self._response = None - self._steps = wire.programming_steps(image.data, target) + self._steps = wire.programming_steps(self._image.data, self._target) self._advance() def _advance(self): @@ -66,6 +93,7 @@ class TmsBootloader(QObject): except StopIteration: self._finish(True, "Образ загружен в RAM TMS" if self._target.load_only else "Запись Spartan-6 подтверждена платой" if self._target.kind == "spartan6" + else "Прошивка записана; проверка отключена" if not self._target.verify else "Прошивка завершена, записанные данные проверены") return except (ValueError, RuntimeError) as error: @@ -79,11 +107,15 @@ class TmsBootloader(QObject): if self._serial.write(request) != len(request): self._finish(False, "Ошибка передачи " + self._stage) return + self.request_sent.emit(request) if self._active: - self._timeout.start(timeout + int(len(request) * 10000 / self._serial.baudRate())) + self._timeout.start(timeout + int((len(request) + self._size) * 10000 / self._serial.baudRate())) def _read(self): - chunk = bytes(self._serial.readAll()) + self.receive(bytes(self._serial.readAll())) + + def receive(self, chunk): + """Forward fragments from a shared port without reading it twice.""" if not self._waiting: return self._rx.extend(chunk) @@ -116,6 +148,10 @@ class TmsBootloader(QObject): self._timeout.stop() self._settle.stop() self._waiting = False + if self._command == wire.CMD_UPLOAD and self._target.kind == "tms": + data = wire.decode_upload_reply(reply, self._target.controller, self._size - 8) + remaining = len(self._image.data) - len(self.readback) + self.readback.extend(data[:remaining]) self._response = reply self._delay.start() @@ -133,8 +169,10 @@ class TmsBootloader(QObject): self._active = self._waiting = False for timer in (self._timeout, self._settle, self._delay): timer.stop() - if self._serial.isOpen(): + if self._owns_serial and self._serial.isOpen(): self._serial.close() + if not success: + self.readback.clear() if success: self.progress.emit(100, message) self.finished.emit(success, message)