"""Reusable byte/word terminal page with an injected protocol or word handler.""" from PySide6.QtCore import Qt, Signal from PySide6.QtWidgets import ( QWidget, QVBoxLayout, QLabel, QHBoxLayout, QGroupBox, QGridLayout, QLineEdit, QPushButton, QScrollArea, QSizePolicy, ) from .rawprotocol import RawProtocolWidget from .serialbus import SerialBus, SerialTab class ByteWorkPanel(QScrollArea): """Embed the complete terminal page in any layout or QTabWidget. protocol supplies request_async/cancel_async, peek/poke and RAW methods. device_address can be an integer or a callable returning the current controller address. An optional start_word(write, address, value) handler lets an existing tab keep its own memory queue; it must report results with set_read_value() and set_status(). Otherwise this panel performs asynchronous reads and verifies writes by reading the same word back. """ status_changed = Signal(str) def __init__(self, protocol, parent=None, *, device_address=10, start_word=None, macros_file=None, default_address="$1000", hint=None, initial_status="Нет данных от контроллера"): super().__init__(parent) self.protocol = protocol self.device_address = device_address self._start_word = start_word self._word_running = False self._word_generation = 0 # Поколение сохраняет совместимость с прежними обработчиками результатов. self._word_keys = set() # Каждый принятый запрос слова получает независимый ключ отмены. self.setWidgetResizable(True) content = QWidget() self.setWidget(content) layout = QVBoxLayout(content) self.hint = QLabel(hint if hint is not None else "Адрес слова — 32 бита; значение — 16 бит (0000…FFFF). " "Адрес: десятичный, $HEX или 0xHEX. Запись проверяется чтением.") self.hint.setWordWrap(True) layout.addWidget(self.hint) panels = QHBoxLayout() for write in (True, False): group = QGroupBox("Запись" if write else "Чтение") group.setSizePolicy(QSizePolicy.Maximum, QSizePolicy.Fixed) form = QGridLayout(group) address = QLineEdit(default_address) value = QLineEdit("0000" if write else "") binary = QLineEdit("0000 0000 0000 0000" if write else "") address.setFixedWidth(150) value.setFixedWidth(100) binary.setFixedWidth(240) value.setReadOnly(not write) binary.setReadOnly(not write) for row, (title, widget) in enumerate((("Адрес", address), ("Байт (hex)", value), ("Байт (bin)", binary))): form.addWidget(QLabel(title), row, 0) form.addWidget(widget, row, 1) button = QPushButton("Записать" if write else "Прочитать") form.addWidget(button, 3, 1, Qt.AlignLeft) form.setColumnStretch(2, 1) if write: self.byte_write_address, self.byte_write_value, self.byte_write_binary = address, value, binary self.byte_write_button = button value.textEdited.connect(lambda text: self.sync_byte_value(text, 16)) binary.textEdited.connect(lambda text: self.sync_byte_value(text, 2)) button.clicked.connect(lambda: self.start_byte(True)) else: self.byte_read_address, self.byte_read_value, self.byte_read_binary = address, value, binary self.byte_read_button = button button.clicked.connect(lambda: self.start_byte(False)) panels.addWidget(group) panels.addStretch() layout.addLayout(panels) self.status = QLabel(initial_status) self.status.setWordWrap(True) layout.addWidget(self.status) self.raw_widget = RawProtocolWidget(protocol, device_address=device_address) self.bus_widget = SerialTab(SerialBus(protocol), protocol, self.raw_widget, device_address=device_address, macros_file=macros_file) layout.addWidget(self.bus_widget) layout.addWidget(self.raw_widget) layout.addStretch() def set_status(self, text): """Display external connection/operation status without emitting it again.""" self.status.setText(text) def _report(self, text): self.set_status(text) self.status_changed.emit(text) @staticmethod def _binary(value): bits = f"{value:016b}" return " ".join(bits[i:i + 4] for i in range(0, 16, 4)) def sync_byte_value(self, text, base): try: value = int(text.replace(" ", ""), base) if not 0 <= value <= 0xFFFF: return except ValueError: return target = self.byte_write_binary if base == 16 else self.byte_write_value target.setText(self._binary(value) if base == 16 else f"{value:04X}") def set_read_value(self, address, value): self.byte_read_address.setText(f"${address:X}") self.byte_read_value.setText(f"{value:04X}") self.byte_read_binary.setText(self._binary(value)) def start_byte(self, write): try: field = self.byte_write_address if write else self.byte_read_address text = field.text().strip() address = int(text[1:], 16) if text.startswith("$") else int( text, 16 if text.lower().startswith("0x") else 10) value = int(self.byte_write_value.text().replace(" ", ""), 16) if write else None if not 0 <= address <= 0xFFFFFFFF: raise ValueError() if write and (not 0 <= value <= 0xFFFF or int(self.byte_write_binary.text().replace(" ", ""), 2) != value): raise ValueError() except ValueError: self._report("Проверьте адрес и значение HEX/BIN") return if self._start_word is not None: self._start_word(write, address, value) return controller = self.device_address() if callable(self.device_address) else self.device_address # Адрес контроллера фиксируется до ожидания занятого соединения. key = object() # Повторное нажатие создаёт самостоятельное действие, а не заменяет предыдущее. self._word_keys.add(key) # Все ожидающие слова учитываются для остановки страницы. self._set_word_running(True) # Признак состояния не меняет доступность полей и кнопок. self.bus_widget.queue.submit(lambda: self._word_steps(address, value), controller, lambda ok, reply: self._word_done(key, address, value, ok, reply), key=key) # Запись и проверка чтением выполняются целиком без вмешательства другой шины. def _word_steps(self, address, value): # Существующий протокол сохраняет проверку записанного слова чтением того же адреса. if value is not None: # Чистое чтение не требует предварительного WriteMemory. yield address, value # Подтверждение записи должно прийти до запроса проверки. result = yield address, None # ReadMemory возвращает одно шестнадцатибитное слово по выбранному адресу. if not isinstance(result, dict) or not isinstance(result.get("value"), int) or not 0 <= result["value"] <= 0xFFFF: # Неверный ответ не выдаётся за прочитанное слово. raise ValueError("Некорректный ответ чтения памяти") # Ошибка формата остаётся доступна в статусе страницы. return result # Только финальный ответ используется для отображения HEX и BIN. def _word_done(self, key, address, expected, success, result): # Результат связан с параметрами конкретного нажатия, а не с изменёнными полями. if key not in self._word_keys: # Поздний callback закрытой страницы не перезаписывает её значения. return # Остановленное действие не продолжает обмен. self._word_keys.remove(key) # Этот запрос слова больше не ожидает соединение. self._set_word_running(bool(self._word_keys)) # Признак учитывает остальные принятые нажатия. if not success: # Таймаут, ошибка CRC или отключение отображаются как действительные ошибки связи. self._report(str(result)) # Пользователь видит причину отказа данного запроса. return # Не показываем недостоверное слово памяти. value = result["value"] # Проверенное слово принадлежит завершённому ReadMemory. self.set_read_value(address, value) # Заполняем имеющиеся поля HEX и BIN без пересоздания панели. if expected is not None and value != expected: # Прежняя проверка записи сохраняется для несовпадающего прочитанного значения. self._report(f"Запись 0x{address:08X} не подтверждена: прочитано {value:04X}, ожидалось {expected:04X}") # Указываем адрес, фактическое слово и ожидаемое значение. else: # Успешное чтение либо совпадающая проверка записи завершают действие. self._report("Запись проверена чтением" if expected is not None else "Чтение завершено") # Сохраняем понятный итог операции. def _set_word_running(self, running): # Состояние обмена хранится отдельно от доступности интерфейса. self._word_running = running # Кнопки и поля принимают новые действия даже во время текущего обмена. def stop(self): # Отключение страницы отменяет задания всего общего соединения. self._word_keys.clear() # Поздние результаты отключения не меняют поля закрываемой страницы. self._set_word_running(False) # Завершаем состояние ожидания собственных запросов слова. self.bus_widget.stop() # Общая очередь снимает неотправленные задания и отменяет текущий транспорт при закрытии соединения. def closeEvent(self, event): self.stop() super().closeEvent(event) def build_byte_page(protocol, **options): """Create a standalone page; add the returned widget to a tab or layout.""" return ByteWorkPanel(protocol, **options)