"""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.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): if self._word_running: self._report("Дождитесь завершения текущего обмена") return 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 self._word_generation += 1 generation = self._word_generation self._set_word_running(True) self._report("Запись и проверка чтением…" if write else "Чтение…") self.protocol.request_async(address, value, controller, lambda success, result: self._word_reply(generation, address, value, controller, success, result)) def _word_reply(self, generation, address, expected, controller, success, result): if generation != self._word_generation or not self._word_running: return if not success: self._set_word_running(False) self._report(str(result)) return if expected is not None and "value" not in result: self.protocol.request_async(address, None, controller, lambda ok, reply: self._word_reply(generation, address, expected, controller, ok, reply)) return value = result["value"] self.set_read_value(address, value) self._set_word_running(False) 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 self.byte_write_button.setEnabled(not running) self.byte_read_button.setEnabled(not running) def stop(self): """Cancel this page's scheduled work before closing its connection.""" self.bus_widget.stop_macro() if self._word_running: self._word_generation += 1 self._set_word_running(False) self.protocol.cancel_async() self._report("Обмен прерван; переданные команды могли быть выполнены") 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)