"""Qt transport over the common native TMS2812 codec.""" from PySide6.QtWidgets import * from PySide6.QtCore import * from PySide6.QtGui import * from PySide6.QtSerialPort import QSerialPort from protocan import tms_terminal as memory class RawProtocol: def __init__(self, host=None): self.host = host self.serial_port = QSerialPort() if host is None else None self.buffer = bytearray() self._loop = None self._callback = None self._command = None self._error = None self._packet = None self._raw_rx = bytearray() if host is None: self.serial_port.readyRead.connect(self._receive) @property def is_connected(self): return self.port.isOpen() @property def port(self): return self.host._serial if self.host else self.serial_port def connect(self, port_name, baudrate=115200): self.port.setPortName(port_name) self.port.setBaudRate(baudrate) return self.port.open(QIODevice.ReadWrite) def disconnect(self): self.port.close() def _receive(self): self.feed(bytes(self.port.readAll())) def feed(self, raw): self._raw_rx.extend(raw) del self._raw_rx[:-65536] if self._loop is None and self._callback is None: return self.buffer.extend(raw) try: packet = memory.memory_response(self.buffer, self._controller, self._command) if packet is not None: self._packet = packet if len(packet) == (10 if self._command == memory.PEEK else 6): self._settle.stop() self._finish_async() if self._callback else self._loop.quit() elif not self._settle.isActive(): self._settle.start(30) except ValueError as error: self._error = str(error) self._finish_async() if self._callback else self._loop.quit() def request_async(self, address, value, device_address, callback): """Begin one request and return immediately; completion runs on the Qt thread.""" if not self.is_connected: callback(False, "Подключите COM-порт") return if self._loop is not None or self._callback is not None or (self.host and self.host._pending_address is not None): callback(False, "Дождитесь завершения текущего обмена") return self._controller = self.host.address.value() if self.host else device_address self._command = memory.PEEK if value is None else memory.POKE try: request = memory.request(self._controller, address, value) except (ValueError, RuntimeError) as error: callback(False, str(error)) return self.buffer.clear() self._error = self._packet = None self._callback = callback self._settle = QTimer() self._settle.setSingleShot(True) self._settle.timeout.connect(self._finish_async) self._async_timer = QTimer() self._async_timer.setSingleShot(True) self._async_timer.timeout.connect(lambda: self.cancel_async("Нет ответа от устройства")) if self.host: self.host._pending_address = self._controller self.host._pending_kind = "pysetterminal" self.host._log("TX", request) self.port.readAll() self._async_timer.start(self.host.timeout_ms.value() if self.host else 1000) if self.port.write(request) != len(request): self.cancel_async("Ошибка передачи") def _finish_async(self): callback = self._callback if callback is None: return self._callback = None self._async_timer.stop() self._settle.stop() if self.host: self.host._pending_address = self.host._pending_kind = None if self._error or self._packet is None: callback(False, self._error or "Обмен прерван") return packet = self._packet result = {'address': self._controller, 'command': self._command, 'raw': packet} if self._command == memory.PEEK: result = {'value': memory.decode_reply(packet, self._controller, self._command), 'raw_response': result} callback(True, result) def cancel_async(self, reason="Макрос остановлен"): if self._callback is not None: self._error = reason self._finish_async() def _exchange(self, address, value, device_address): if not self.is_connected: return False, "Подключите COM-порт" if self._loop is not None or self._callback is not None or (self.host and self.host._pending_address is not None): return False, "Дождитесь завершения текущего обмена" self._controller = self.host.address.value() if self.host else device_address self._command = memory.PEEK if value is None else memory.POKE try: request = memory.request(self._controller, address, value) except (ValueError, RuntimeError) as error: return False, str(error) self.buffer.clear() self._error = self._packet = None loop = QEventLoop() timer = QTimer() timer.setSingleShot(True) timer.timeout.connect(loop.quit) self._loop = loop self._settle = QTimer() self._settle.setSingleShot(True) self._settle.timeout.connect(loop.quit) if self.host: self.host.byte_page.setEnabled(False) self.host._pending_address = self._controller self.host._pending_kind = "pysetterminal" self.host._log("TX", request) try: self.port.readAll() if self.port.write(request) != len(request): return False, "Ошибка передачи" if self._error is None and (self._packet is None or self._settle.isActive()): timer.start(self.host.timeout_ms.value() if self.host else 1000) loop.exec() if self._error or self._packet is None: return False, self._error or "Нет ответа от устройства" packet = self._packet result = {'address': self._controller, 'command': self._command, 'raw': packet} if value is None: return True, {'value': memory.decode_reply(packet, self._controller, self._command), 'raw_response': result} return True, result finally: timer.stop() self._settle.stop() self._loop = None if self.host: self.host._pending_address = self.host._pending_kind = None self.host.byte_page.setEnabled(True) def poke(self, mem_address, data_value, device_address=1): return self._exchange(mem_address, data_value, device_address) def peek(self, mem_address, device_address=1): return self._exchange(mem_address, None, device_address) def send_raw(self, hex_data): if not self.is_connected: return False, "Подключите COM-порт" if self._loop is not None or self._callback is not None or (self.host and self.host._pending_address is not None): return False, "Дождитесь завершения текущего обмена" if self.host and self.host._cycle_timer.isActive(): return False, "Остановите циклический обмен перед RAW" try: data = bytes.fromhex(hex_data) except ValueError: return False, "Некорректные HEX данные" if not data: return False, "Введите HEX данные" if self.port.write(data) != len(data): return False, "Ошибка передачи" if self.host: self.host._log("TX", data) return True, "TX " + data.hex(" ") def receive_raw(self, max_bytes=1024): data = bytes(self._raw_rx[:max_bytes]) del self._raw_rx[:max_bytes] return (True, data.hex(" ")) if data else (False, "Нет данных") class RawProtocolWidget(QWidget): """Виджет для отладки RAW протокола (всегда виден внизу окна)""" def __init__(self, raw_protocol): super().__init__() self.raw_protocol = raw_protocol self.init_ui() def init_ui(self): layout = QVBoxLayout() # Группа для ПРЯМОГО ДОСТУПА К ПАМЯТИ (PEEK/POKE) mem_group = QGroupBox("Прямой доступ к памяти (PEEK/POKE)") mem_layout = QGridLayout() mem_layout.addWidget(QLabel("Адрес памяти (hex):"), 0, 0) self.mem_addr_edit = QLineEdit("2000") self.mem_addr_edit.setMaximumWidth(100) mem_layout.addWidget(self.mem_addr_edit, 0, 1) mem_layout.addWidget(QLabel("Данные (hex):"), 1, 0) self.mem_data_edit = QLineEdit("0000") self.mem_data_edit.setMaximumWidth(100) mem_layout.addWidget(self.mem_data_edit, 1, 1) self.mem_write_btn = QPushButton("Записать (POKE)") self.mem_write_btn.clicked.connect(self.memory_write) mem_layout.addWidget(self.mem_write_btn, 0, 2) self.mem_read_btn = QPushButton("Прочитать (PEEK)") self.mem_read_btn.clicked.connect(self.memory_read) mem_layout.addWidget(self.mem_read_btn, 1, 2) self.mem_result_label = QLabel("---") mem_layout.addWidget(self.mem_result_label, 1, 3) mem_group.setLayout(mem_layout) layout.addWidget(mem_group) # Группа для RAW отправки raw_group = QGroupBox("RAW отправка/чтение") raw_layout = QVBoxLayout() self.raw_send_edit = QLineEdit() self.raw_send_edit.setPlaceholderText("Введите hex данные для отправки") raw_layout.addWidget(self.raw_send_edit) hbox = QHBoxLayout() self.raw_send_btn = QPushButton("Отправить RAW") self.raw_send_btn.clicked.connect(self.send_raw_data) hbox.addWidget(self.raw_send_btn) self.raw_read_btn = QPushButton("Прочитать RAW") self.raw_read_btn.clicked.connect(self.read_raw_data) hbox.addWidget(self.raw_read_btn) raw_layout.addLayout(hbox) self.raw_output = QTextEdit() self.raw_output.setReadOnly(True) self.raw_output.setMaximumHeight(100) raw_layout.addWidget(self.raw_output) self.raw_clear_btn = QPushButton("Очистить логи") self.raw_clear_btn.clicked.connect(self.clear_logs) raw_layout.addWidget(self.raw_clear_btn) raw_group.setLayout(raw_layout) layout.addWidget(raw_group) self.setLayout(layout) def memory_write(self): """Запись в память (POKE)""" try: device_addr = 10 mem_addr = int(self.mem_addr_edit.text(), 16) data = int(self.mem_data_edit.text(), 16) success, result = self.raw_protocol.poke(mem_addr, data, device_addr) if success: self.raw_output.append(f"POKE 0x{mem_addr:04X} = 0x{data:04X}") self.raw_output.append(f"Ответ: {result['raw'].hex()}") else: self.raw_output.append(f"Ошибка POKE: {result}") except ValueError as e: self.raw_output.append(f"Ошибка в данных: {e}") def memory_read(self): """Чтение из памяти (PEEK)""" try: device_addr = 10 mem_addr = int(self.mem_addr_edit.text(), 16) success, result = self.raw_protocol.peek(mem_addr, device_addr) if success: value = result['value'] self.mem_result_label.setText(f"0x{value:04X} ({value})") self.raw_output.append(f"PEEK 0x{mem_addr:04X} = 0x{value:04X}") self.raw_output.append(f"Ответ: {result['raw_response']['raw'].hex()}") else: self.mem_result_label.setText(f"Ошибка: {result}") self.raw_output.append(f"Ошибка PEEK: {result}") except ValueError as e: self.mem_result_label.setText(f"Ошибка: {e}") self.raw_output.append(f"Ошибка в данных: {e}") def send_raw_data(self): """Отправка сырых данных""" hex_str = self.raw_send_edit.text() success, result = self.raw_protocol.send_raw(hex_str) if success: self.raw_output.append(result) else: self.raw_output.append(f"Ошибка: {result}") def read_raw_data(self): """Чтение сырых данных""" success, result = self.raw_protocol.receive_raw() if success: self.raw_output.append(f"Получено: {result}") else: self.raw_output.append(f"Ошибка: {result}") def append_text(self, text): """Добавить текст в вывод (используется SerialTab)""" if text: self.raw_output.append(text) def clear_logs(self): """Очистить лог""" self.raw_output.clear()