diff --git a/python/pyproject.toml b/python/pyproject.toml index 72f8fab..37f1664 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -15,7 +15,7 @@ saleae = ["logic2-automation>=1.0.7,<2"] [tool.setuptools.packages.find] where = ["."] -include = ["firmware_image*", "altera_logic*", "logic_analyzer*", "set_devices*", "protocan*", "setprotocol*"] +include = ["firmware_image*", "altera_logic*", "logic_analyzer*", "set_devices*", "protocan*", "setprotocol*", "pysetterminal*"] [tool.setuptools.package-data] set_devices = ["gas_logger_maps/*.json"] diff --git a/python/pysetterminal/README.md b/python/pysetterminal/README.md index a415f35..f15c4f3 100644 --- a/python/pysetterminal/README.md +++ b/python/pysetterminal/README.md @@ -3,7 +3,7 @@ Shared PySide6 terminal migrated from PySetTerminal-master/pysetterminal. - Standalone entry point: `python -m pysetterminal` with templates/python on PYTHONPATH. -- SETGUI embeds `pysetterminal.panel.build_byte_page` in TMS2812 / Работа с байтом. +- SETGUI embeds `pysetterminal.panel.ByteWorkPanel` in TMS2812 / Работа с байтом. - Serial bus, macro editor, asynchronous macro execution with cancellation, PEEK/POKE and RAW console are included. - COM transport uses QtSerialPort. Embedded operations reuse the TMS2812 connection and configured controller address. - PEEK/POKE packet construction and response validation use the existing protocan native C ABI, with 16-bit TMS2812 values. @@ -11,3 +11,78 @@ Shared PySide6 terminal migrated from PySetTerminal-master/pysetterminal. - The standalone launcher uses PySide6; SETGUI Windows 7 uses its existing PySide2/Qt5 compatibility layer. Frozen SETGUI builds bundle Python and Qt. Pyserial is not required. A native set-protocol runtime remains required. `build_and_clean.py` is the migrated original standalone packaging helper. + +## Подключение «Работы с байтом» в другую вкладку + +Вся панель — самостоятельный `ByteWorkPanel`: чтение и запись слова, HEX/BIN, +операции с шиной, макросы, PEEK/POKE и RAW. Она не требует `Tms2812Tab`. +Данные — 16 бит, адрес слова — 32 бита; протокол по умолчанию — TMS2812. + +```python +from pysetterminal.panel import ByteWorkPanel +from pysetterminal.rawprotocol import RawProtocol + +# Сохраните протокол в объекте, который владеет соединением. +self.byte_protocol = RawProtocol() +self.byte_page = ByteWorkPanel( + self.byte_protocol, + device_address=lambda: self.controller_address.value(), +) +self.tabs.addTab(self.byte_page, "Работа с байтом") + +# Подключение выполняется явным действием пользователя. +# self.byte_protocol.connect("COM3", 57600) + +# Перед закрытием соединения: +# self.byte_page.stop() +# self.byte_protocol.disconnect() +``` + +В существующую компоновку панель добавляется через +`layout.addWidget(self.byte_page)`. Фабрика +`build_byte_page(protocol, **options)` возвращает такой же виджет. + +Параметры: + +- `protocol` — существующий протокол с методами `request_async`, + `cancel_async`, `peek`, `poke`, `send_raw`, `receive_raw`. + Асинхронный ответ передаётся в `callback(success, result)`: для чтения + `result["value"]`, для записи `result["raw"]`; ошибки — строкой. + Синхронные `peek/poke` возвращают `(success, result)`; чтение дополнительно + содержит `result["raw_response"]["raw"]` для журнала. +- `device_address` — адрес контроллера или функция, возвращающая текущий адрес; + используется всеми блоками панели. Адрес фиксируется на время записи с + проверкой и выполнения макроса. +- `macros_file` — отдельный JSON-файл макросов для конкретной вкладки. + По умолчанию сохраняется существующее хранилище приложения. +- `default_address`, `hint`, `initial_status` — начальный адрес и тексты. +- `start_word(write, address, value)` — необязательный обработчик для вкладок, + у которых уже есть очередь памяти. При чтении `value=None`. + Обработчик отвечает за чтение, проверку записи и остановку своей очереди. + +Без `start_word` панель выполняет асинхронный запрос, а запись подтверждает +последующим чтением. Пример подключения к существующей очереди: + +```python +self.byte_page = ByteWorkPanel( + self.protocol, + device_address=lambda: self.address.value(), + start_word=lambda write, address, value: self.start_memory( + write=write, address=address, value=value), + macros_file=self.storage_dir / "byte_macros.json", +) +self.tabs.addTab(self.byte_page, "Работа с байтом") + +# В обработчике ответа очереди: +self.byte_page.set_read_value(address, value) +self.byte_page.set_status("Чтение завершено") +``` + +`set_status(text)` показывает внешний статус. Сигнал `status_changed(str)` +передаёт ошибки ввода и результаты операций панели владельцу. +`stop()` отменяет её асинхронный обмен и макрос; внешнюю очередь памяти +останавливает владелец вкладки. + +Каждому независимому соединению нужен собственный протокол. Если вкладки +используют одно соединение, владелец транспорта должен запрещать одновременные +запросы и маршрутизировать ответы. В SETGUI это делает существующая очередь TMS2812. diff --git a/python/pysetterminal/panel.py b/python/pysetterminal/panel.py index 2f5a3f9..4fa8540 100644 --- a/python/pysetterminal/panel.py +++ b/python/pysetterminal/panel.py @@ -1,55 +1,183 @@ -from PySide6.QtCore import Qt -from PySide6.QtWidgets import QWidget, QVBoxLayout, QLabel, QHBoxLayout, QGroupBox, QGridLayout, QLineEdit, QPushButton, QScrollArea, QSizePolicy +"""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, +) -def build_byte_page(self): - self.byte_page = QScrollArea() - self.byte_page.setWidgetResizable(True) - content = QWidget() - self.byte_page.setWidget(content) - layout = QVBoxLayout(content) - hint = QLabel("TMS2812: адрес слова — 32 бита; значение — 16 бит (0000…FFFF). " - "Адрес: десятичный, $HEX или 0xHEX. Запись проверяется чтением.") - hint.setWordWrap(True) - layout.addWidget(hint) - panels = QHBoxLayout() - for write in (True, False): - group = QGroupBox("Запись" if write else "Чтение") - group.setSizePolicy(QSizePolicy.Maximum, QSizePolicy.Fixed) - form = QGridLayout(group) - address = QLineEdit("$1000") - 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 - 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)) +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.byte_read_address, self.byte_read_value, self.byte_read_binary = address, value, binary - button.clicked.connect(lambda: self._start_byte(False)) - panels.addWidget(group) - panels.addStretch() - layout.addLayout(panels) - status = QLabel("Нет данных от контроллера") - status.setWordWrap(True) - self.status_changed.connect(status.setText) - layout.addWidget(status) - from pysetterminal.rawprotocol import RawProtocol, RawProtocolWidget - from pysetterminal.serialbus import SerialBus, SerialTab - self.pyset_protocol = RawProtocol(self) - self.pyset_raw_widget = RawProtocolWidget(self.pyset_protocol) - self.pyset_bus_widget = SerialTab(SerialBus(self.pyset_protocol), self.pyset_protocol, self.pyset_raw_widget) - layout.addWidget(self.pyset_bus_widget) - layout.addWidget(self.pyset_raw_widget) - layout.addStretch() + 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) diff --git a/python/pysetterminal/rawprotocol.py b/python/pysetterminal/rawprotocol.py index 510dbba..4738e79 100644 --- a/python/pysetterminal/rawprotocol.py +++ b/python/pysetterminal/rawprotocol.py @@ -66,7 +66,8 @@ class RawProtocol: 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._controller = (device_address if device_address is not None else + self.host.address.value() if self.host else 1) self._command = memory.PEEK if value is None else memory.POKE try: request = memory.request(self._controller, address, value) @@ -119,7 +120,8 @@ class RawProtocol: 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._controller = (device_address if device_address is not None else + self.host.address.value() if self.host else 1) self._command = memory.PEEK if value is None else memory.POKE try: request = memory.request(self._controller, address, value) @@ -162,10 +164,10 @@ class RawProtocol: 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): + def poke(self, mem_address, data_value, device_address=None): return self._exchange(mem_address, data_value, device_address) - def peek(self, mem_address, device_address=1): + def peek(self, mem_address, device_address=None): return self._exchange(mem_address, None, device_address) def send_raw(self, hex_data): @@ -196,9 +198,10 @@ class RawProtocol: class RawProtocolWidget(QWidget): """Виджет для отладки RAW протокола (всегда виден внизу окна)""" - def __init__(self, raw_protocol): - super().__init__() + def __init__(self, raw_protocol, parent=None, *, device_address=10): + super().__init__(parent) self.raw_protocol = raw_protocol + self.device_address = device_address self.init_ui() def init_ui(self): @@ -272,7 +275,7 @@ class RawProtocolWidget(QWidget): def memory_write(self): """Запись в память (POKE)""" try: - device_addr = 10 + device_addr = self.device_address() if callable(self.device_address) else self.device_address mem_addr = int(self.mem_addr_edit.text(), 16) data = int(self.mem_data_edit.text(), 16) @@ -290,7 +293,7 @@ class RawProtocolWidget(QWidget): def memory_read(self): """Чтение из памяти (PEEK)""" try: - device_addr = 10 + device_addr = self.device_address() if callable(self.device_address) else self.device_address mem_addr = int(self.mem_addr_edit.text(), 16) success, result = self.raw_protocol.peek(mem_addr, device_addr) diff --git a/python/pysetterminal/serialbus.py b/python/pysetterminal/serialbus.py index 54db527..27dc9c3 100644 --- a/python/pysetterminal/serialbus.py +++ b/python/pysetterminal/serialbus.py @@ -322,8 +322,10 @@ class SerialTab(QWidget): MACROS_FILE = "macros.json" # Файл для сохранения макросов - def __init__(self, serial_bus, raw_protocol, raw_widget): - super().__init__() + def __init__(self, serial_bus, raw_protocol, raw_widget, parent=None, *, + device_address=10, macros_file=None): + super().__init__(parent) + self.device_address = device_address self.serial_bus = serial_bus self.raw_protocol = raw_protocol self.raw_widget = raw_widget @@ -333,7 +335,8 @@ class SerialTab(QWidget): self._macro_generation = 0 from pathlib import Path - self.MACROS_FILE = str(Path(QStandardPaths.writableLocation(QStandardPaths.AppDataLocation)) / "pysetterminal" / "macros.json") + self.MACROS_FILE = str(macros_file if macros_file is not None else + Path(QStandardPaths.writableLocation(QStandardPaths.AppDataLocation)) / "pysetterminal" / "macros.json") Path(self.MACROS_FILE).parent.mkdir(parents=True, exist_ok=True) # Загружаем сохраненные макросы self.load_macros() @@ -455,7 +458,7 @@ class SerialTab(QWidget): def write_to_bus(self): """Запись в шину через регистры""" try: - device_addr = 10 + device_addr = self.device_address() if callable(self.device_address) else self.device_address data = int(self.bus_data_edit.text(), 16) tetrad = int(self.bus_tetrad_edit.text()) register_num = int(self.bus_register_edit.text()) @@ -477,7 +480,7 @@ class SerialTab(QWidget): def read_from_bus(self): """Чтение из шины через регистры""" try: - device_addr = 10 + device_addr = self.device_address() if callable(self.device_address) else self.device_address tetrad = int(self.bus_tetrad_edit.text()) register_num = int(self.bus_register_edit.text()) @@ -652,6 +655,7 @@ class SerialTab(QWidget): return from protocan.tms_terminal import bus_steps try: + controller = self.device_address() if callable(self.device_address) else self.device_address jobs = [] for command in commands: if command['type'] not in ('write', 'read'): @@ -664,6 +668,7 @@ class SerialTab(QWidget): if not jobs: self.raw_widget.append_text("Макрос пуст") return + self._macro_controller = controller self._macro_jobs = jobs self._macro_index = 0 self._macro_running = True @@ -686,7 +691,7 @@ class SerialTab(QWidget): return generation = self._macro_generation address, value = self._macro_jobs[self._macro_index] - self.raw_protocol.request_async(address, value, 10, + self.raw_protocol.request_async(address, value, self._macro_controller, lambda success, result: self._macro_reply(generation, success, result)) def _macro_reply(self, generation, success, result): diff --git a/python/tests/test_byte_work_panel.py b/python/tests/test_byte_work_panel.py new file mode 100644 index 0000000..3fda8c2 --- /dev/null +++ b/python/tests/test_byte_work_panel.py @@ -0,0 +1,213 @@ +"""The reusable terminal must work without a Tms2812Tab or its private state.""" +import os +from pathlib import Path +import subprocess +import sys + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +import pytest +from PySide6.QtTest import QTest +from PySide6.QtWidgets import QApplication, QTabWidget +from pysetterminal.panel import ByteWorkPanel, build_byte_page + + +class Protocol: + def __init__(self): + self.requests = [] + self.callback = None + self.cancelled = 0 + + def request_async(self, address, value, controller, callback): + self.requests.append((address, value, controller)) + self.callback = callback + + def reply(self, success, result): + callback, self.callback = self.callback, None + callback(success, result) + + def cancel_async(self): + self.cancelled += 1 + if self.callback is not None: + self.reply(False, "cancelled") + + def poke(self, address, value, controller): + self.requests.append((address, value, controller)) + return True, {"raw": b"ack"} + + def peek(self, address, controller): + self.requests.append((address, None, controller)) + return True, {"value": 0x1234, "raw_response": {"raw": b"reply"}} + + def send_raw(self, text): + return True, text + + def receive_raw(self): + return False, "Нет данных" + + +@pytest.fixture(scope="module") +def app(): + return QApplication.instance() or QApplication([]) + + +@pytest.fixture +def page(app, tmp_path): + protocol = Protocol() + widget = ByteWorkPanel(protocol, device_address=7, macros_file=tmp_path / "macros.json") + yield widget, protocol + widget.close() + app.processEvents() + + +def test_shared_panel_imports_and_constructs_without_desktop_application(tmp_path): + templates = Path(__file__).resolve().parents[1] + code = """ +import importlib.abc, sys +class BlockDesktop(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path=None, target=None): + if fullname.split('.')[0] == 'gui_desktop': + raise ImportError('Unexpected desktop dependency') +sys.meta_path.insert(0, BlockDesktop()) +from PySide6.QtWidgets import QApplication, QTabWidget +from pysetterminal.panel import build_byte_page +from pysetterminal.rawprotocol import RawProtocol +app = QApplication([]) +tabs = QTabWidget() +page = build_byte_page(RawProtocol(), macros_file='macros.json') +tabs.addTab(page, 'Работа с байтом') +assert tabs.widget(0) is page +assert 'gui_desktop' not in sys.modules +assert page.bus_widget and page.raw_widget +page.close() +""" + result = subprocess.run([sys.executable, "-c", code], cwd=tmp_path, + env=dict(os.environ, PYTHONPATH=str(templates)), + capture_output=True, text=True, timeout=30) + assert result.returncode == 0, result.stderr + + +def test_multiple_tabs_keep_controls_status_and_macros_separate(app, tmp_path): + calls = [] + tabs = QTabWidget() + first = build_byte_page(Protocol(), start_word=lambda *args: calls.append(args), + macros_file=tmp_path / "first.json") + second = ByteWorkPanel(Protocol(), macros_file=tmp_path / "second.json") + tabs.addTab(first, "Первый") + tabs.addTab(second, "Второй") + first.byte_write_value.setText("0301") + first.byte_write_value.textEdited.emit("0301") + first.byte_write_address.setText("0x1234") + first.byte_write_button.click() + first.set_read_value(0x1234, 0x0301) + first.set_status("Готово") + first.bus_widget.macros = [{"name": "Первый", "commands": []}] + first.bus_widget.save_macros() + assert calls == [(True, 0x1234, 0x0301)] + assert first.byte_write_binary.text() == "0000 0011 0000 0001" + assert second.byte_write_value.text() == "0000" + assert second.byte_read_value.text() == "" + assert second.status.text() == "Нет данных от контроллера" + second.bus_widget.load_macros() + assert second.bus_widget.macros == [] + first.close() + second.close() + tabs.close() + + +@pytest.mark.parametrize("address", ["4096", "$1000", "0x1000"]) +def test_read_accepts_address_formats_and_displays_both_values(page, address): + widget, protocol = page + widget.byte_read_address.setText(address) + widget.byte_read_button.click() + assert protocol.requests == [(0x1000, None, 7)] + assert not widget.byte_read_button.isEnabled() + protocol.reply(True, {"value": 0xABCD}) + assert widget.byte_read_address.text() == "$1000" + assert widget.byte_read_value.text() == "ABCD" + assert widget.byte_read_binary.text() == "1010 1011 1100 1101" + assert widget.byte_read_button.isEnabled() + + +@pytest.mark.parametrize("address,value,binary", [ + ("-1", "0000", "0"), + ("$100000000", "0000", "0"), + ("invalid", "0000", "0"), + ("$1000", "10000", "10000000000000000"), + ("$1000", "0001", "0"), + ("$1000", "GGGG", "0"), + ("$1000", "0000", "invalid"), +]) +def test_invalid_inputs_do_not_send(page, address, value, binary): + widget, protocol = page + widget.byte_write_address.setText(address) + widget.byte_write_value.setText(value) + widget.byte_write_binary.setText(binary) + widget.byte_write_button.click() + assert protocol.requests == [] + assert "Проверьте" in widget.status.text() + + +@pytest.mark.parametrize("readback,confirmed", [(0x0301, True), (0x0302, False)]) +def test_write_is_verified_and_controller_is_captured_for_whole_operation(page, readback, confirmed): + widget, protocol = page + controller = [13] + widget.device_address = lambda: controller[0] + widget.byte_write_value.setText("0301") + widget.sync_byte_value("0301", 16) + widget.byte_write_button.click() + assert protocol.requests == [(0x1000, 0x0301, 13)] + controller[0] = 21 + protocol.reply(True, {"raw": b"ack"}) + assert protocol.requests[-1] == (0x1000, None, 13) + protocol.reply(True, {"value": readback}) + assert widget.byte_read_value.text() == f"{readback:04X}" + assert ("Запись проверена" in widget.status.text()) == confirmed + if not confirmed: + assert "не подтверждена" in widget.status.text() + assert widget.byte_write_button.isEnabled() + + +def test_failure_and_cancellation_restore_controls_and_ignore_late_reply(page): + widget, protocol = page + widget.byte_read_button.click() + protocol.reply(False, "CRC ответа не совпадает") + assert widget.status.text() == "CRC ответа не совпадает" + assert widget.byte_read_button.isEnabled() + widget.byte_read_button.click() + late_reply = protocol.callback + widget.stop() + assert protocol.cancelled == 1 + widget.byte_read_button.click() + late_reply(True, {"value": 0x9999}) + assert widget.byte_read_value.text() == "" + assert not widget.byte_read_button.isEnabled() + protocol.reply(True, {"value": 0x1234}) + assert widget.byte_read_value.text() == "1234" + assert widget.byte_read_button.isEnabled() + + +def test_controller_provider_is_used_by_bus_raw_memory_and_macros(app, tmp_path): + protocol = Protocol() + controller = [21] + widget = ByteWorkPanel(protocol, device_address=lambda: controller[0], + macros_file=tmp_path / "address.json") + try: + widget.raw_widget.memory_write() + widget.raw_widget.memory_read() + assert protocol.requests == [(0x2000, 0, 21), (0x2000, None, 21)] + widget.bus_widget.read_from_bus() + assert protocol.requests[-2:] == [(0x200B, 0, 21), (0x200F, None, 21)] + widget.bus_widget.execute_macro_commands([{"type": "read", "tetrad": 3, "register": 5}]) + controller[0] = 22 + QTest.qWait(20) + assert protocol.requests[-1] == (0x200B, 0x35, 21) + protocol.reply(True, {"raw": b"ack"}) + QTest.qWait(20) + assert protocol.requests[-1] == (0x200F, None, 21) + protocol.reply(True, {"value": 0x4321, "raw_response": {"raw": b"reply"}}) + QTest.qWait(20) + assert not widget.bus_widget._macro_running + assert widget.bus_widget.bus_read_value_label.text() == "0x4321 (17185)" + finally: + widget.close()