From b0c0c05cfa6cc7de6c814ca9c79a5ace2b436a5d Mon Sep 17 00:00:00 2001 From: Andrey Date: Thu, 8 Oct 2026 18:34:41 +0300 Subject: [PATCH] Add shared PySetTerminal with asynchronous cancellable macros --- c/set-protocol/include/pcan_abi.h | 3 + c/set-protocol/include/tms2812.h | 3 + c/set-protocol/src/pcan_abi.c | 6 + c/set-protocol/src/tms2812.c | 13 + c/set-protocol/tests/test_tms2812.c | 13 + python/protocan/tms_terminal.py | 19 + python/pysetterminal/.gitignore | 5 + python/pysetterminal/README.md | 13 + python/pysetterminal/__init__.py | 1 + python/pysetterminal/__main__.py | 2 + python/pysetterminal/build_and_clean.py | 124 +++++ python/pysetterminal/mainterm.py | 209 +++++++ python/pysetterminal/panel.py | 49 ++ python/pysetterminal/rawprotocol.py | 335 ++++++++++++ python/pysetterminal/serialbus.py | 689 ++++++++++++++++++++++++ python/tests/test_pysetterminal.py | 71 +++ 16 files changed, 1555 insertions(+) create mode 100644 python/pysetterminal/.gitignore create mode 100644 python/pysetterminal/README.md create mode 100644 python/pysetterminal/__init__.py create mode 100644 python/pysetterminal/__main__.py create mode 100644 python/pysetterminal/build_and_clean.py create mode 100644 python/pysetterminal/mainterm.py create mode 100644 python/pysetterminal/panel.py create mode 100644 python/pysetterminal/rawprotocol.py create mode 100644 python/pysetterminal/serialbus.py create mode 100644 python/tests/test_pysetterminal.py diff --git a/c/set-protocol/include/pcan_abi.h b/c/set-protocol/include/pcan_abi.h index 9ec1055..f921a22 100644 --- a/c/set-protocol/include/pcan_abi.h +++ b/c/set-protocol/include/pcan_abi.h @@ -119,6 +119,9 @@ PCAN_ABI_API const char *pcan_abi_periph28335_command_name( /* PM67/TMS320F2812 legacy memory upload. */ PCAN_ABI_API size_t pcan_abi_tms2812_build_fixed(uint8_t controller, uint8_t command, const uint8_t *payload, size_t payload_size, uint8_t *output, size_t output_size); +PCAN_ABI_API int pcan_abi_tms2812_bus_step(int write, unsigned step, unsigned tetrad, unsigned reg, + uint16_t data, uint32_t *address, uint16_t *value, uint8_t *command); + PCAN_ABI_API size_t pcan_abi_tms2812_build_memory(uint8_t controller, uint8_t command, uint32_t word_address, uint16_t value, uint8_t *output, size_t output_size); PCAN_ABI_API size_t pcan_abi_tms2812_build_test(uint8_t controller, uint16_t duration, diff --git a/c/set-protocol/include/tms2812.h b/c/set-protocol/include/tms2812.h index 68a5631..859ab62 100644 --- a/c/set-protocol/include/tms2812.h +++ b/c/set-protocol/include/tms2812.h @@ -36,6 +36,9 @@ typedef enum { TMS2812_ERROR_CAPACITY = -6 } tms2812_status_t; +int tms2812_bus_step(int write, unsigned step, unsigned tetrad, unsigned reg, + uint16_t data, uint32_t *address, uint16_t *value, uint8_t *command); + uint16_t tms2812_crc16(const uint8_t *data, size_t size); size_t tms2812_build_fixed_request(uint8_t controller, uint8_t command, diff --git a/c/set-protocol/src/pcan_abi.c b/c/set-protocol/src/pcan_abi.c index 0ad7c48..ceefa92 100644 --- a/c/set-protocol/src/pcan_abi.c +++ b/c/set-protocol/src/pcan_abi.c @@ -379,3 +379,9 @@ int pcan_abi_gui_parser_stats(const void *parser_storage, uint32_t *frames, if (stray_bytes != NULL) *stray_bytes = parser->stats.stray_bytes; return 1; } + +int pcan_abi_tms2812_bus_step(int write, unsigned step, unsigned tetrad, unsigned reg, + uint16_t data, uint32_t *address, uint16_t *value, uint8_t *command) +{ + return tms2812_bus_step(write, step, tetrad, reg, data, address, value, command); +} diff --git a/c/set-protocol/src/tms2812.c b/c/set-protocol/src/tms2812.c index 3222bda..415be8e 100644 --- a/c/set-protocol/src/tms2812.c +++ b/c/set-protocol/src/tms2812.c @@ -137,3 +137,16 @@ int tms2812_decode_upload_response( memcpy(output, &data[2], byte_count); return TMS2812_OK; } + +int tms2812_bus_step(int write, unsigned step, unsigned tetrad, unsigned reg, + uint16_t data, uint32_t *address, uint16_t *value, uint8_t *command) +{ + if (!address || !value || !command) return TMS2812_ERROR_ARGUMENT; + if ((write != 0 && write != 1) || step > 1U || tetrad > 14U || reg > 15U) + return TMS2812_ERROR_RANGE; + *command = (!write && step == 1U) ? TMS2812_CMD_PEEK : TMS2812_CMD_POKE; + *address = write ? (step == 0U ? 0x200AU : 0x200BU) : (step == 0U ? 0x200BU : 0x200FU); + *value = write && step == 0U ? data : (!write && step == 1U ? 0U : + (uint16_t)((write ? 0x8000U : 0U) | (tetrad << 4) | reg)); + return TMS2812_OK; +} diff --git a/c/set-protocol/tests/test_tms2812.c b/c/set-protocol/tests/test_tms2812.c index 864f8c2..66a7f76 100644 --- a/c/set-protocol/tests/test_tms2812.c +++ b/c/set-protocol/tests/test_tms2812.c @@ -18,6 +18,19 @@ static int test_terminal(void) uint8_t reply[] = {0x0a, 0x38, 0x65, 0xa5, 0, 0, 0xaf, 0x9b, 0, 0}; const uint8_t test_prefix[] = {7, 66, 25, 0, 18, 0}; size_t size; + uint32_t bus_address; + uint16_t bus_value; + uint8_t bus_command; + if (tms2812_bus_step(1, 0, 3, 5, 0x1234, &bus_address, &bus_value, &bus_command) || + bus_address != 0x200AU || bus_value != 0x1234 || bus_command != 57) return 20; + if (tms2812_bus_step(1, 1, 3, 5, 0x1234, &bus_address, &bus_value, &bus_command) || + bus_address != 0x200BU || bus_value != 0x8035 || bus_command != 57) return 21; + if (tms2812_bus_step(0, 0, 3, 5, 0, &bus_address, &bus_value, &bus_command) || + bus_address != 0x200BU || bus_value != 0x35 || bus_command != 57) return 22; + if (tms2812_bus_step(0, 1, 3, 5, 0, &bus_address, &bus_value, &bus_command) || + bus_address != 0x200FU || bus_command != 56) return 23; + if (tms2812_bus_step(0, 0, 15, 0, 0, &bus_address, &bus_value, &bus_command) != TMS2812_ERROR_RANGE || + tms2812_bus_step(0, 2, 0, 0, 0, &bus_address, &bus_value, &bus_command) != TMS2812_ERROR_RANGE) return 24; if (tms2812_build_memory_request(10, 0, 0, 0, output, sizeof output) != 8U || memcmp(output, ping, sizeof ping) != 0) return 10; if (tms2812_build_memory_request(10, TMS2812_CMD_POKE, 0x12345678UL, 0xABCD, diff --git a/python/protocan/tms_terminal.py b/python/protocan/tms_terminal.py index b9d8bc8..d9bd3e8 100644 --- a/python/protocan/tms_terminal.py +++ b/python/protocan/tms_terminal.py @@ -136,3 +136,22 @@ def memory_response(raw, controller, command): return None return data return None + + +def bus_steps(write, tetrad, register, data=0): + """Two bus register operations planned by the shared C core.""" + _range(tetrad, 14, "Адрес шины") + _range(register, 15, "Регистр шины") + _range(data, 0xFFFF, "Данные шины") + function = _api("bus_step", [ct.c_int, ct.c_uint, ct.c_uint, ct.c_uint, + ct.c_uint16, ct.POINTER(ct.c_uint32), ct.POINTER(ct.c_uint16), + ct.POINTER(ct.c_uint8)], ct.c_int) + result = [] + for step in range(2): + address, value, command = ct.c_uint32(), ct.c_uint16(), ct.c_uint8() + status = function(bool(write), step, tetrad, register, data, + ct.byref(address), ct.byref(value), ct.byref(command)) + if status: + raise ValueError(f"Ошибка планирования шины: {status}") + result.append((address.value, None if command.value == PEEK else value.value)) + return result diff --git a/python/pysetterminal/.gitignore b/python/pysetterminal/.gitignore new file mode 100644 index 0000000..29c5c60 --- /dev/null +++ b/python/pysetterminal/.gitignore @@ -0,0 +1,5 @@ + +/.vscode +/__pycache__ +/macros.json +/BusTerminal.exe diff --git a/python/pysetterminal/README.md b/python/pysetterminal/README.md new file mode 100644 index 0000000..96c93a2 --- /dev/null +++ b/python/pysetterminal/README.md @@ -0,0 +1,13 @@ +# PySetTerminal + +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 / Работа с байтом. +- 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. +- Macro files are stored under the application data directory, outside the source checkout. +- PySide2 and pyserial are no longer required. A native set-protocol runtime remains required. + +`build_and_clean.py` is the migrated original standalone packaging helper. diff --git a/python/pysetterminal/__init__.py b/python/pysetterminal/__init__.py new file mode 100644 index 0000000..1177be3 --- /dev/null +++ b/python/pysetterminal/__init__.py @@ -0,0 +1 @@ +"""Shared PySide6 PySetTerminal project; protocol codecs use protocan native ABI.""" diff --git a/python/pysetterminal/__main__.py b/python/pysetterminal/__main__.py new file mode 100644 index 0000000..344fd03 --- /dev/null +++ b/python/pysetterminal/__main__.py @@ -0,0 +1,2 @@ +from .mainterm import main +main() diff --git a/python/pysetterminal/build_and_clean.py b/python/pysetterminal/build_and_clean.py new file mode 100644 index 0000000..9491753 --- /dev/null +++ b/python/pysetterminal/build_and_clean.py @@ -0,0 +1,124 @@ +import subprocess +import shutil +import os +from pathlib import Path +import PySide6 +from PyInstaller.utils.hooks import collect_data_files +# install: pip install PySide6 lxml nuitka pyinstaller +# - PyInstaller +# - nuitka +# - PySide6 +# - clang + +# === Конфигурация === +USE_NUITKA = True # True — сборка через Nuitka, False — через PyInstaller +MAIN_SCRIPT_NAME = "mainterm" +OUTPUT_NAME = "BusTerminal" +USE_ICON = False # True — использовать иконку, False — без иконки + + +SRC_PATH = Path("./") +SCRIPT_PATH = SRC_PATH / (MAIN_SCRIPT_NAME + ".py") + +DIST_PATH = Path("./").resolve() +WORK_PATH = Path("./build_temp").resolve() +SPEC_PATH = WORK_PATH +ICON_PATH = SRC_PATH / "icon.png" +ICON_ICO_PATH = SRC_PATH / "icon.ico" + +TEMP_FOLDERS = [ + "build_temp", + "__pycache__", + MAIN_SCRIPT_NAME + ".build", + MAIN_SCRIPT_NAME + ".onefile-build", + MAIN_SCRIPT_NAME + ".dist" +] +# === Пути к DLL и прочим зависимостям === +LIBS = { +} + +# === PySide6 плагины === +PySide6_path = Path(PySide6.__file__).parent +datas = [] +datas += collect_data_files('PySide6', includes=['plugins/platforms/*']) +datas += collect_data_files('PySide6', includes=['plugins/styles/*']) +datas += collect_data_files('PySide6', includes=['plugins/imageformats/*']) + +add_data_list = [f"{src};{dest}" for src, dest in datas] + +# Проверка наличия DLL и добавление +add_binary_list = [] +for name, path in LIBS.items(): + if path.exists(): + add_binary_list.append(f"{str(path)};{name}") + else: + print(f"WARNING: {path.name} не найден — он не будет включён в сборку") + +def clean_temp(): + for folder in TEMP_FOLDERS: + path = Path(folder) + if path.exists(): + shutil.rmtree(path, ignore_errors=True) + +if USE_NUITKA: + # Формируем include-data-file только для DLL + include_data_files = [f"--include-data-file={str(path)}={name}" for name, path in LIBS.items() if path.exists()] + + # Добавляем опциональную иконку как встроенный ресурс + if USE_ICON and ICON_ICO_PATH.exists(): + include_data_files.append(f"--include-data-file={ICON_ICO_PATH}=icon.ico") + + cmd = [ + "python", "-m", "nuitka", + "--standalone", + "--onefile", + "--enable-plugin=pyside2", + "--windows-console-mode=disable", + f"--output-dir={DIST_PATH}", + f"--output-filename={OUTPUT_NAME}.exe", + ] + + # Добавляем параметр иконки только если она нужна и существует + if USE_ICON and ICON_ICO_PATH.exists(): + cmd.append(f"--windows-icon-from-ico={ICON_ICO_PATH}") + + cmd.extend(include_data_files) + cmd.append(str(SCRIPT_PATH)) + +else: + # PyInstaller + cmd = [ + "pyinstaller", + "--name", OUTPUT_NAME, + "--distpath", str(DIST_PATH), + "--workpath", str(WORK_PATH), + "--specpath", str(SPEC_PATH), + "--windowed", + "--hidden-import=PySide6.QtWidgets", + "--hidden-import=PySide6.QtGui", + "--hidden-import=PySide6.QtCore", + *[arg for b in add_binary_list for arg in ("--add-binary", b)], + *[arg for d in add_data_list for arg in ("--add-data", d)], + ] + + # Добавляем иконку только если она нужна и существует + if USE_ICON and ICON_ICO_PATH.exists(): + cmd.append(f"--icon={ICON_ICO_PATH}") + else: + print("WARNING: Иконка не используется или не найдена") + + cmd.append(str(SCRIPT_PATH)) + +# === Запуск сборки === +print("Выполняется сборка с помощью " + ("Nuitka" if USE_NUITKA else "PyInstaller")) +print(" ".join(cmd)) + +try: + subprocess.run(cmd, check=True) + print("\nСборка успешно завершена!") + + # Удаление временных папок после сборки + clean_temp() + +except subprocess.CalledProcessError: + print("\nОшибка при сборке.") diff --git a/python/pysetterminal/mainterm.py b/python/pysetterminal/mainterm.py new file mode 100644 index 0000000..720e88b --- /dev/null +++ b/python/pysetterminal/mainterm.py @@ -0,0 +1,209 @@ +import sys +from PySide6.QtSerialPort import QSerialPortInfo +from PySide6.QtWidgets import * +from PySide6.QtCore import * +from PySide6.QtGui import * + +from pysetterminal.rawprotocol import RawProtocol, RawProtocolWidget +from pysetterminal.serialbus import SerialBus, SerialTab + + + +class MainWindow(QMainWindow): + """Главное окно приложения""" + + def __init__(self): + super().__init__() + # Создаем низкоуровневый протокол + self.raw_protocol = RawProtocol() + # Создаем высокоуровневый протокол, передавая ему low-level + self.serial_bus = SerialBus(self.raw_protocol) + self.init_ui() + + def init_ui(self): + self.setWindowTitle("Терминал") + self.setGeometry(100, 100, 800, 600) + + # Главный виджет и компоновка + main_widget = QWidget() + main_layout = QVBoxLayout() + + # Центральный виджет с вкладками + self.tab_widget = QTabWidget() + + # Виджет для RAW протокола (всегда внизу) + self.raw_widget = RawProtocolWidget(self.raw_protocol) + + # Вкладка для серийной шины + self.serial_tab = SerialTab(self.serial_bus, self.raw_protocol, self.raw_widget) + self.tab_widget.addTab(self.serial_tab, "Серийная шина") + + main_layout.addWidget(self.tab_widget) + + main_layout.addWidget(self.raw_widget) + + main_widget.setLayout(main_layout) + self.setCentralWidget(main_widget) + + # Создание меню + self.create_menu() + + # Создание панели инструментов + self.create_toolbar() + + def create_menu(self): + menubar = self.menuBar() + + # Меню Файл + file_menu = menubar.addMenu("Файл") + + exit_action = QAction("Выход", self) + exit_action.triggered.connect(self.close) + file_menu.addAction(exit_action) + + # Меню Помощь + help_menu = menubar.addMenu("Помощь") + + protocol_action = QAction("Информация о протоколе", self) + protocol_action.triggered.connect(self.show_protocol_info) + help_menu.addAction(protocol_action) + + def create_toolbar(self): + toolbar = self.addToolBar("Подключение") + + # Выбор COM порта + toolbar.addWidget(QLabel("COM порт:")) + + self.port_combo = QComboBox() + self.port_combo.setMinimumWidth(150) + self.refresh_ports() + toolbar.addWidget(self.port_combo) + + # Кнопка обновления портов + refresh_btn = QToolButton() + refresh_btn.setText("⟳") + refresh_btn.setToolTip("Обновить список портов") + refresh_btn.clicked.connect(self.refresh_ports) + toolbar.addWidget(refresh_btn) + + # Кнопка подключения/отключения + self.connect_btn = QPushButton("Подключить") + self.connect_btn.clicked.connect(self.toggle_connection) + toolbar.addWidget(self.connect_btn) + + # Индикатор статуса + self.status_label = QLabel("Отключено") + self.status_label.setStyleSheet("color: red;") + toolbar.addWidget(self.status_label) + + toolbar.addSeparator() + + # Настройки скорости + toolbar.addWidget(QLabel("Скорость:")) + self.baud_combo = QComboBox() + self.baud_combo.addItems(["9600", "19200", "38400", "57600", "115200", "230400", "460800", "921600"]) + self.baud_combo.setCurrentText("115200") + toolbar.addWidget(self.baud_combo) + + def refresh_ports(self): + """Обновление списка доступных COM портов""" + self.port_combo.clear() + ports = QSerialPortInfo.availablePorts() + for port in ports: + self.port_combo.addItem(f"{port.portName()} - {port.description()}") + + if not ports: + self.port_combo.addItem("Порты не найдены") + + def toggle_connection(self): + """Подключение/отключение от COM порта""" + if self.raw_protocol.is_connected: + # Отключение + self.raw_protocol.disconnect() + self.connect_btn.setText("Подключить") + self.status_label.setText("Отключено") + self.status_label.setStyleSheet("color: red;") + + # Блокировка элементов управления + self.serial_tab.bus_write_btn.setEnabled(False) + self.serial_tab.bus_read_btn.setEnabled(False) + self.raw_widget.raw_send_btn.setEnabled(False) + self.raw_widget.raw_read_btn.setEnabled(False) + else: + # Подключение + port_text = self.port_combo.currentText() + if " - " in port_text: + port_name = port_text.split(" - ")[0] + else: + port_name = port_text + + try: + baudrate = int(self.baud_combo.currentText()) + if self.raw_protocol.connect(port_name, baudrate): + self.connect_btn.setText("Отключить") + self.status_label.setText("Подключено") + self.status_label.setStyleSheet("color: green;") + + # Разблокировка элементов управления + self.serial_tab.bus_write_btn.setEnabled(True) + self.serial_tab.bus_read_btn.setEnabled(True) + self.raw_widget.raw_send_btn.setEnabled(True) + self.raw_widget.raw_read_btn.setEnabled(True) + else: + QMessageBox.warning(self, "Ошибка", "Не удалось подключиться к порту") + except Exception as e: + QMessageBox.warning(self, "Ошибка", f"Ошибка подключения: {e}") + + def closeEvent(self, event): + self.serial_tab.stop_macro() + self.raw_protocol.cancel_async() + self.raw_protocol.disconnect() + super().closeEvent(event) + + def show_protocol_info(self): + """Показать информацию о протоколе""" + info = """ + ИНФОРМАЦИЯ О ПРОТОКОЛЕ: + + На основе кода контроллера: + + 1. ФОРМАТ СООБЩЕНИЙ: + [адрес][команда][данные][CRC16-IBM] + + 2. КЛЮЧЕВЫЕ АДРЕСА: + - 0x200A: Регистр данных шины (запись) + - 0x200B: Регистр управления шиной + - 0x200F: Регистр данных шины (чтение) + + 3. ПРОТОКОЛ ШИНЫ: + ЗАПИСЬ: + 1. POKE 0x200A <данные> + 2. POKE 0x200B 0x80A0 | (адрес << 4) + + ЧТЕНИЕ: + 1. POKE 0x200B 0x00A0 | (адрес << 4) + 2. PEEK 0x200F + + 4. ЧТО НУЖНО УТОЧНИТЬ: + - Точные коды команд CMD_RS232_PEEK и CMD_RS232_POKE + - Длину полей в сообщениях + - Формат ответов от контроллера + """ + + QMessageBox.information(self, "Информация о протоколе", info) + + +def main(): + app = QApplication(sys.argv) + + # Установка стиля + app.setStyle("Fusion") + + window = MainWindow() + window.show() + + sys.exit(app.exec()) + + +if __name__ == "__main__": + main() diff --git a/python/pysetterminal/panel.py b/python/pysetterminal/panel.py new file mode 100644 index 0000000..616c9db --- /dev/null +++ b/python/pysetterminal/panel.py @@ -0,0 +1,49 @@ +from PySide6.QtWidgets import QWidget, QVBoxLayout, QLabel, QHBoxLayout, QGroupBox, QGridLayout, QLineEdit, QPushButton, QScrollArea + +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 "Чтение") + form = QGridLayout(group) + address = QLineEdit("$1000") + value = QLineEdit("0000" if write else "") + binary = QLineEdit("0000 0000 0000 0000" if write else "") + binary.setMinimumWidth(220) + value.setReadOnly(not write) + binary.setReadOnly(not write) + for row, (title, widget) in enumerate((("Адрес", address), ("Байт (hex)", value), ("Байт (bin)", binary))): + form.addWidget(QLabel(title), row * 2, 0) + form.addWidget(widget, row * 2 + 1, 0) + button = QPushButton("Записать" if write else "Прочитать") + form.addWidget(button, 6, 0) + 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)) + 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) + 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() diff --git a/python/pysetterminal/rawprotocol.py b/python/pysetterminal/rawprotocol.py new file mode 100644 index 0000000..39d129a --- /dev/null +++ b/python/pysetterminal/rawprotocol.py @@ -0,0 +1,335 @@ +"""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() diff --git a/python/pysetterminal/serialbus.py b/python/pysetterminal/serialbus.py new file mode 100644 index 0000000..465e655 --- /dev/null +++ b/python/pysetterminal/serialbus.py @@ -0,0 +1,689 @@ +import json +import os +from functools import partial +from copy import deepcopy +from PySide6.QtWidgets import * +from PySide6.QtCore import * +from PySide6.QtGui import * + +class SerialBus: + """ВЫСОКОУРОВНЕВЫЙ протокол для работы с серийной шиной (только через регистры)""" + + def __init__(self, raw_protocol): + self.raw = raw_protocol # Используем низкоуровневый протокол + + def _execute(self, write, tetrad, register, data, device_address): + from protocan.tms_terminal import bus_steps + try: + steps = bus_steps(write, tetrad, register, data) + except (ValueError, RuntimeError) as error: + return False, str(error) + result = None + for address, value in steps: + success, result = (self.raw.peek(address, device_address) if value is None + else self.raw.poke(address, value, device_address)) + if not success: + return False, result + return True, "Запись в шину подтверждена" if write else result + + def write_bus(self, data_word, address_tetrad, register_num=0, device_address=1): + return self._execute(True, address_tetrad, register_num, data_word, device_address) + + def read_bus(self, address_tetrad, register_num=0, device_address=1): + return self._execute(False, address_tetrad, register_num, 0, device_address) + + +class MacroCommandDialog(QDialog): + """Диалог для добавления/редактирования команды макроса""" + + def __init__(self, parent=None, command=None): + super().__init__(parent) + self.command = command if command else {} + self.init_ui() + + def init_ui(self): + self.setWindowTitle("Команда макроса") + self.setMinimumWidth(400) + + layout = QVBoxLayout() + + # Тип команды + type_layout = QHBoxLayout() + type_layout.addWidget(QLabel("Тип команды:")) + self.cmd_type = QComboBox() + self.cmd_type.addItems(["Запись", "Чтение"]) + type_layout.addWidget(self.cmd_type) + layout.addLayout(type_layout) + + # Адрес устройства + addr_layout = QHBoxLayout() + addr_layout.addWidget(QLabel("Адрес устройства (0-14):")) + self.device_addr = QSpinBox() + self.device_addr.setRange(0, 14) + addr_layout.addWidget(self.device_addr) + layout.addLayout(addr_layout) + + # Регистр + reg_layout = QHBoxLayout() + reg_layout.addWidget(QLabel("Регистр (0-15):")) + self.register_num = QSpinBox() + self.register_num.setRange(0, 15) + reg_layout.addWidget(self.register_num) + layout.addLayout(reg_layout) + + # Данные (только для записи) + self.data_label = QLabel("Данные (hex):") + self.data_edit = QLineEdit("0000") + self.data_edit.setMaximumWidth(100) + data_layout = QHBoxLayout() + data_layout.addWidget(self.data_label) + data_layout.addWidget(self.data_edit) + layout.addLayout(data_layout) + + # Загружаем данные команды, если она передана + if self.command: + if self.command.get('type') == 'write': + self.cmd_type.setCurrentText("Запись") + self.device_addr.setValue(self.command.get('tetrad', 0)) + self.register_num.setValue(self.command.get('register', 0)) + self.data_edit.setText(f"{self.command.get('data', 0):04X}") + else: + self.cmd_type.setCurrentText("Чтение") + self.device_addr.setValue(self.command.get('tetrad', 0)) + self.register_num.setValue(self.command.get('register', 0)) + + # Подключаем изменение типа команды + self.cmd_type.currentTextChanged.connect(self.on_cmd_type_changed) + self.on_cmd_type_changed(self.cmd_type.currentText()) + + # Кнопки + button_layout = QHBoxLayout() + self.ok_btn = QPushButton("OK") + self.ok_btn.clicked.connect(self._accept_command) + self.cancel_btn = QPushButton("Отмена") + self.cancel_btn.clicked.connect(self.reject) + button_layout.addWidget(self.ok_btn) + button_layout.addWidget(self.cancel_btn) + layout.addLayout(button_layout) + + self.setLayout(layout) + + def on_cmd_type_changed(self, text): + """Показать/скрыть поле данных в зависимости от типа команды""" + if text == "Запись": + self.data_label.show() + self.data_edit.show() + else: + self.data_label.hide() + self.data_edit.hide() + + def _accept_command(self): + if self.cmd_type.currentText() == "Запись": + try: + value = int(self.data_edit.text(), 16) + if not 0 <= value <= 0xFFFF: + raise ValueError() + except ValueError: + QMessageBox.warning(self, "Ошибка", "Данные должны быть в диапазоне 0000…FFFF") + return + self.accept() + + def get_command(self): + """Получить команду из диалога""" + cmd_type = 'write' if self.cmd_type.currentText() == "Запись" else 'read' + command = { + 'type': cmd_type, + 'tetrad': self.device_addr.value(), + 'register': self.register_num.value() + } + + if cmd_type == 'write': + try: + command['data'] = int(self.data_edit.text(), 16) + except ValueError: + command['data'] = 0 + + return command + + +class MacroEditorDialog(QDialog): + """Диалог для создания/редактирования макроса""" + + def __init__(self, parent=None, macro=None): + super().__init__(parent) + self.macro = deepcopy(macro) if macro else {'name': '', 'commands': []} + self.init_ui() + + def init_ui(self): + self.setWindowTitle("Редактор макроса") + self.setMinimumSize(500, 400) + + layout = QVBoxLayout() + + # Имя макроса + name_layout = QHBoxLayout() + name_layout.addWidget(QLabel("Имя макроса:")) + self.macro_name = QLineEdit(self.macro.get('name', 'Новый макрос')) + name_layout.addWidget(self.macro_name) + layout.addLayout(name_layout) + + # Список команд + layout.addWidget(QLabel("Команды:")) + self.commands_list = QListWidget() + layout.addWidget(self.commands_list) + + # Кнопки управления командами + cmd_buttons_layout = QHBoxLayout() + self.add_cmd_btn = QPushButton("Добавить команду") + self.add_cmd_btn.clicked.connect(self.add_command) + cmd_buttons_layout.addWidget(self.add_cmd_btn) + + self.edit_cmd_btn = QPushButton("Редактировать") + self.edit_cmd_btn.clicked.connect(self.edit_command) + cmd_buttons_layout.addWidget(self.edit_cmd_btn) + + self.remove_cmd_btn = QPushButton("Удалить") + self.remove_cmd_btn.clicked.connect(self.remove_command) + cmd_buttons_layout.addWidget(self.remove_cmd_btn) + + self.move_up_btn = QPushButton("Вверх") + self.move_up_btn.clicked.connect(self.move_command_up) + cmd_buttons_layout.addWidget(self.move_up_btn) + + self.move_down_btn = QPushButton("Вниз") + self.move_down_btn.clicked.connect(self.move_command_down) + cmd_buttons_layout.addWidget(self.move_down_btn) + + layout.addLayout(cmd_buttons_layout) + + # Загружаем команды, если они есть + self.load_commands() + + # Кнопки сохранения/отмены + button_layout = QHBoxLayout() + self.save_btn = QPushButton("Сохранить") + self.save_btn.clicked.connect(self.accept) + self.cancel_btn = QPushButton("Отмена") + self.cancel_btn.clicked.connect(self.reject) + button_layout.addWidget(self.save_btn) + button_layout.addWidget(self.cancel_btn) + layout.addLayout(button_layout) + + self.setLayout(layout) + + def load_commands(self): + """Загрузить команды в список""" + self.commands_list.clear() + for cmd in self.macro['commands']: + if cmd['type'] == 'write': + text = f"Запись: устр.{cmd['tetrad']}, рег.{cmd['register']}, 0x{cmd['data']:04X}" + else: + text = f"Чтение: устр.{cmd['tetrad']}, рег.{cmd['register']}" + self.commands_list.addItem(text) + + def add_command(self): + """Добавить новую команду""" + dialog = MacroCommandDialog(self) + if dialog.exec(): + command = dialog.get_command() + self.macro['commands'].append(command) + self.load_commands() + + def edit_command(self): + """Редактировать выбранную команду""" + current_row = self.commands_list.currentRow() + if current_row >= 0 and current_row < len(self.macro['commands']): + dialog = MacroCommandDialog(self, self.macro['commands'][current_row]) + if dialog.exec(): + self.macro['commands'][current_row] = dialog.get_command() + self.load_commands() + + def remove_command(self): + """Удалить выбранную команду""" + current_row = self.commands_list.currentRow() + if current_row >= 0 and current_row < len(self.macro['commands']): + del self.macro['commands'][current_row] + self.load_commands() + + def move_command_up(self): + """Переместить команду вверх""" + current_row = self.commands_list.currentRow() + if current_row > 0: + self.macro['commands'][current_row], self.macro['commands'][current_row-1] = \ + self.macro['commands'][current_row-1], self.macro['commands'][current_row] + self.load_commands() + self.commands_list.setCurrentRow(current_row-1) + + def move_command_down(self): + """Переместить команду вниз""" + current_row = self.commands_list.currentRow() + if current_row >= 0 and current_row < len(self.macro['commands']) - 1: + self.macro['commands'][current_row], self.macro['commands'][current_row+1] = \ + self.macro['commands'][current_row+1], self.macro['commands'][current_row] + self.load_commands() + self.commands_list.setCurrentRow(current_row+1) + + def get_macro(self): + """Получить макрос из диалога""" + return { + 'name': self.macro_name.text(), + 'commands': self.macro['commands'].copy() + } + + + +class SerialTab(QWidget): + """Вкладка для работы с серийной шиной (ТОЛЬКО операции с шиной)""" + + MACROS_FILE = "macros.json" # Файл для сохранения макросов + + def __init__(self, serial_bus, raw_protocol, raw_widget): + super().__init__() + self.serial_bus = serial_bus + self.raw_protocol = raw_protocol + self.raw_widget = raw_widget + self.macros = [] # Список макросов + self.macro_buttons = [] # Список кнопок макросов + self._macro_running = False + self._macro_generation = 0 + + from pathlib import Path + self.MACROS_FILE = str(Path(QStandardPaths.writableLocation(QStandardPaths.AppDataLocation)) / "pysetterminal" / "macros.json") + Path(self.MACROS_FILE).parent.mkdir(parents=True, exist_ok=True) + # Загружаем сохраненные макросы + self.load_macros() + + self.init_ui() + + + def init_ui(self): + layout = QVBoxLayout() + + # Группа для операций с шиной + bus_group = QGroupBox("Операции с шиной (через регистры 0x200A/0x200B/0x200F)") + bus_layout = QGridLayout() + + bus_layout.addWidget(QLabel("Данные для шины (hex):"), 0, 0) + self.bus_data_edit = QLineEdit("0000") + self.bus_data_edit.setMaximumWidth(100) + bus_layout.addWidget(self.bus_data_edit, 0, 1) + + bus_layout.addWidget(QLabel("Адреса (0-14):"), 0, 2) + self.bus_tetrad_edit = QLineEdit("0") + self.bus_tetrad_edit.setMaximumWidth(50) + bus_layout.addWidget(self.bus_tetrad_edit, 0, 3) + + bus_layout.addWidget(QLabel("Регистр (0-15):"), 0, 4) + self.bus_register_edit = QLineEdit("0") + self.bus_register_edit.setMaximumWidth(50) + bus_layout.addWidget(self.bus_register_edit, 0, 5) + + self.bus_write_btn = QPushButton("Записать в шину") + self.bus_write_btn.clicked.connect(self.write_to_bus) + bus_layout.addWidget(self.bus_write_btn, 1, 0, 1, 3) + + self.bus_read_btn = QPushButton("Прочитать из шину") + self.bus_read_btn.clicked.connect(self.read_from_bus) + bus_layout.addWidget(self.bus_read_btn, 1, 3, 1, 3) + + bus_group.setLayout(bus_layout) + layout.addWidget(bus_group) + + # --- Группа для макросов --- + macros_group = QGroupBox("Макросы") + macros_layout = QVBoxLayout() + + # Кнопка создания нового макроса + self.create_macro_btn = QPushButton("Создать новый макрос") + self.create_macro_btn.clicked.connect(self.create_macro) + macros_layout.addWidget(self.create_macro_btn) + self.stop_macro_btn = QPushButton("Остановить макрос") + self.stop_macro_btn.setEnabled(False) + self.stop_macro_btn.clicked.connect(self.stop_macro) + macros_layout.addWidget(self.stop_macro_btn) + + # Прокручиваемая область для кнопок макросов + self.macros_scroll_area = QScrollArea() + self.macros_scroll_area.setWidgetResizable(True) + # Устанавливаем фиксированную высоту, чтобы вместить 3-4 строки макросов + self.macros_scroll_area.setMinimumHeight(150) # Высота для 3-4 строк + + self.macros_scroll_content = QWidget() + self.macros_scroll_layout = QGridLayout() + self.macros_scroll_content.setLayout(self.macros_scroll_layout) + self.macros_scroll_area.setWidget(self.macros_scroll_content) + self.macros_scroll_layout.setAlignment(Qt.AlignTop) + macros_layout.addWidget(self.macros_scroll_area) + + macros_group.setLayout(macros_layout) + layout.addWidget(macros_group) + + # --- ВНЕ ГРУППЫ: Результаты чтения --- + results_layout = QGridLayout() + + # Считанное слово + results_layout.addWidget(QLabel("Считанное слово:"), 0, 0) + self.bus_read_value_label = QLabel("---") + self.bus_read_value_label.setStyleSheet("font-weight: bold; font-size: 14px; border: 1px solid #ccc; padding: 5px;") + results_layout.addWidget(self.bus_read_value_label, 0, 1) + + # Ответ контроллера + results_layout.addWidget(QLabel("Ответ контроллера:"), 1, 0) + self.bus_result_label = QLabel("---") + self.bus_result_label.setStyleSheet("border: 1px solid #ccc; padding: 5px; font-family: monospace;") + self.bus_result_label.setWordWrap(True) + results_layout.addWidget(self.bus_result_label, 1, 1) + + # Создаем фрейм для результатов + results_frame = QFrame() + results_frame.setFrameShape(QFrame.StyledPanel) + results_frame.setLayout(results_layout) + layout.addWidget(results_frame) + + layout.addStretch() + self.setLayout(layout) + + # После создания интерфейса создаем кнопки для загруженных макросов + # ВАЖНО: Используем enumerate для правильных индексов + for i, macro in enumerate(self.macros): + self.add_macro_button_with_index(macro, i) + + def write_to_bus(self): + """Запись в шину через регистры""" + try: + device_addr = 10 + data = int(self.bus_data_edit.text(), 16) + tetrad = int(self.bus_tetrad_edit.text()) + register_num = int(self.bus_register_edit.text()) + + success, result = self.serial_bus.write_bus(data, tetrad, register_num, device_addr) + + if success: + self.raw_widget.append_text(f"Запись в шину: адрес={tetrad}, рег={register_num}, данные=0x{data:04X} - OK") + self.last_result = f"Запись в шину: данные=0x{data:04X}, адрес=0x{tetrad:X}, регистр=0x{register_num:X}" + self.last_result += f"\n{result}" + else: + self.raw_widget.append_text(f"Запись в шину: адрес={tetrad}, рег={register_num} - Ошибка: {result}") + self.last_result = f"Ошибка записи в шину: {result}" + + except ValueError as e: + self.raw_widget.append_text(f"Ошибка в данных: {e}") + self.last_result = f"Ошибка в данных: {e}" + + def read_from_bus(self): + """Чтение из шины через регистры""" + try: + device_addr = 10 + tetrad = int(self.bus_tetrad_edit.text()) + register_num = int(self.bus_register_edit.text()) + + success, result = self.serial_bus.read_bus(tetrad, register_num, device_addr) + + if success: + # result теперь словарь с данными + hex_str = ' '.join(f'{b:02X}' for b in result['raw_response']['raw']) + self.bus_result_label.setText(f"{hex_str}") + + # Извлекаем считанное слово + read_word = result['value'] + self.bus_read_value_label.setText(f"0x{read_word:04X} ({read_word})") + + self.raw_widget.append_text(f"Чтение из шины: адрес={tetrad}, рег={register_num} = 0x{read_word:04X}") + + self.last_result = f"Чтение из шины: адрес=0x{tetrad:X}, регистр=0x{register_num:X}" + self.last_result += f"\nСчитано: 0x{read_word:04X} ({read_word})" + self.last_result += f"\nОтвет контроллера: {hex_str}" + else: + self.bus_result_label.setText(f"Ошибка: {result}") + self.bus_read_value_label.setText("---") + self.raw_widget.append_text(f"Чтение из шины: адрес={tetrad}, рег={register_num} - Ошибка: {result}") + self.last_result = f"Ошибка чтения из шины: {result}" + + except ValueError as e: + self.bus_result_label.setText(f"Ошибка: {e}") + self.bus_read_value_label.setText("---") + self.raw_widget.append_text(f"Ошибка в данных: {e}") + self.last_result = f"Ошибка в данных: {e}" + + except ValueError as e: + self.bus_result_label.setText(f"Ошибка: {e}") + self.bus_read_value_label.setText("---") + error_text = f"Ошибка в данных: {e}" + self.raw_widget.append_text(f"✗ {error_text}") + self.last_result = error_text + + def create_macro(self): + """Создать новый макрос""" + dialog = MacroEditorDialog(self) + if dialog.exec(): + macro = dialog.get_macro() + if not macro['name']: + macro['name'] = f"Макрос {len(self.macros) + 1}" + + self.macros.append(macro) + self.add_macro_button(macro) + self.save_macros() # Сохраняем после добавления + + dialog.deleteLater() + QApplication.processEvents() # Обрабатываем события + + + def add_macro_button_with_index(self, macro, index): + """Добавить кнопку макроса в интерфейс с указанным индексом""" + # Изменяем расчет строк и колонок для 3 колонок + items_per_row = 3 # Теперь 3 кнопки в строке + row = len(self.macro_buttons) // items_per_row + col = len(self.macro_buttons) % items_per_row + + # Создаем контейнер для кнопки и кнопки редактирования + container = QWidget() + container_layout = QHBoxLayout() + container_layout.setContentsMargins(0, 0, 0, 0) + + # Основная кнопка макроса + macro_btn = QPushButton(macro['name']) + macro_btn.setMinimumHeight(40) + macro_btn.setStyleSheet(""" + QPushButton { + font-weight: bold; + text-align: left; + padding: 10px; + } + QPushButton:hover { + background-color: #e0e0e0; + } + """) + + # Используем partial для фиксации индекса + macro_btn.clicked.connect(partial(self.execute_macro_by_index, index)) + + # Кнопка редактирования + edit_btn = QPushButton("✎") + edit_btn.setFixedSize(30, 30) + edit_btn.setToolTip("Редактировать макрос") + edit_btn.clicked.connect(partial(self.edit_macro_by_index, index)) + + # Кнопка удаления + delete_btn = QPushButton("✕") + delete_btn.setFixedSize(30, 30) + delete_btn.setToolTip("Удалить макрос") + delete_btn.clicked.connect(partial(self.delete_macro, index)) + + container_layout.addWidget(macro_btn) + container_layout.addWidget(edit_btn) + container_layout.addWidget(delete_btn) + container.setLayout(container_layout) + + self.macros_scroll_layout.addWidget(container, row, col, 1, 1) # span изменен на 1 + self.macro_buttons.append(container) + + + def add_macro_button(self, macro): + """Добавить кнопку макроса в интерфейс (при создании нового)""" + # При создании нового макроса используем последний индекс + index = len(self.macros) - 1 + self.add_macro_button_with_index(macro, index) + + def execute_macro_by_index(self, macro_index): + """Выполнить макрос по индексу""" + if 0 <= macro_index < len(self.macros): + self.execute_macro_commands(self.macros[macro_index]['commands']) + + def edit_macro_by_index(self, macro_index): + """Редактировать макрос по индексу""" + if 0 <= macro_index < len(self.macros): + dialog = MacroEditorDialog(self, self.macros[macro_index]) + if dialog.exec(): + edited_macro = dialog.get_macro() + self.macros[macro_index] = edited_macro + + # Обновляем кнопку - находим ее по индексу + if macro_index < len(self.macro_buttons): + container = self.macro_buttons[macro_index] + macro_btn = container.layout().itemAt(0).widget() + if macro_btn: + macro_btn.setText(edited_macro['name']) + + self.save_macros() + + dialog.deleteLater() + QApplication.processEvents() # Обрабатываем события + + # Добавим метод для удаления макроса + def delete_macro(self, macro_index): + """Удалить макрос""" + if 0 <= macro_index < len(self.macros): + # Спросим подтверждение + reply = QMessageBox.question( + self, + "Удаление макроса", + f"Удалить макрос '{self.macros[macro_index]['name']}'?", + QMessageBox.Yes | QMessageBox.No, + QMessageBox.No + ) + + if reply == QMessageBox.Yes: + # Удаляем макрос из списка + del self.macros[macro_index] + + # Полностью перестраиваем layout + self.rebuild_macros_layout() + + # Сохраняем изменения + self.save_macros() + + def execute_macro_commands(self, commands): + """Run a macro as Qt callbacks, without loops or waiting in the UI thread.""" + if self._macro_running: + self.raw_widget.append_text("Дождитесь завершения макроса") + return + from protocan.tms_terminal import bus_steps + try: + jobs = [] + for command in commands: + if command['type'] not in ('write', 'read'): + raise ValueError("Неизвестная команда макроса") + jobs.extend(bus_steps(command['type'] == 'write', command['tetrad'], + command['register'], command.get('data', 0))) + except (KeyError, ValueError, RuntimeError, TypeError) as error: + self.raw_widget.append_text(f"Макрос остановлен: {error}") + return + if not jobs: + self.raw_widget.append_text("Макрос пуст") + return + self._macro_jobs = jobs + self._macro_index = 0 + self._macro_running = True + self._macro_generation += 1 + self._set_macro_controls(True) + generation = self._macro_generation + QTimer.singleShot(0, lambda: self._next_macro_command(generation)) + + def _set_macro_controls(self, running): + self.stop_macro_btn.setEnabled(running) + for widget in (self.bus_write_btn, self.bus_read_btn, self.create_macro_btn, + self.macros_scroll_area, self.raw_widget): + widget.setEnabled(not running) + + def _next_macro_command(self, generation): + if not self._macro_running or generation != self._macro_generation: + return + if self._macro_index >= len(self._macro_jobs): + self._finish_macro("Макрос выполнен") + return + generation = self._macro_generation + address, value = self._macro_jobs[self._macro_index] + self.raw_protocol.request_async(address, value, 10, + lambda success, result: self._macro_reply(generation, success, result)) + + def _macro_reply(self, generation, success, result): + if not self._macro_running or generation != self._macro_generation: + return + if not success: + self._finish_macro(f"Макрос остановлен: {result}") + return + if 'value' in result: + value = result['value'] + self.bus_read_value_label.setText(f"0x{value:04X} ({value})") + self.bus_result_label.setText(result['raw_response']['raw'].hex(" ")) + self.raw_widget.append_text(f"Макрос: прочитано 0x{value:04X}") + else: + self.raw_widget.append_text("Макрос: запись подтверждена") + self._macro_index += 1 + QTimer.singleShot(0, lambda: self._next_macro_command(generation)) + + def _finish_macro(self, message): + self._macro_running = False + self._set_macro_controls(False) + self.last_result = message + self.raw_widget.append_text(message) + + def stop_macro(self): + if not self._macro_running: + return + self._macro_generation += 1 + self._finish_macro("Макрос остановлен пользователем; переданные команды могли быть выполнены") + self.raw_protocol.cancel_async() + + def load_macros(self): + """Загрузить макросы из файла""" + if os.path.exists(self.MACROS_FILE): + try: + with open(self.MACROS_FILE, 'r', encoding='utf-8') as f: + self.macros = json.load(f) + print(f"Загружено {len(self.macros)} макросов") + except Exception as e: + print(f"Ошибка загрузки макросов: {e}") + self.macros = [] + else: + self.macros = [] + + def save_macros(self): + """Сохранить макросы в файл""" + try: + with open(self.MACROS_FILE, 'w', encoding='utf-8') as f: + json.dump(self.macros, f, ensure_ascii=False, indent=2) + print(f"Сохранено {len(self.macros)} макросов") + except Exception as e: + print(f"Ошибка сохранения макросов: {e}") + + def rebuild_macros_layout(self): + """Перестроить layout макросов""" + # Очищаем layout + for i in reversed(range(self.macros_scroll_layout.count())): + item = self.macros_scroll_layout.itemAt(i) + if item and item.widget(): + item.widget().setParent(None) + + # Очищаем список кнопок + self.macro_buttons.clear() + + # Пересоздаем все кнопки с правильными индексами + for i, macro in enumerate(self.macros): + self.add_macro_button_with_index(macro, i) + + + def get_last_result(self): + """Получить последний результат операции""" + result = getattr(self, 'last_result', '') + return result diff --git a/python/tests/test_pysetterminal.py b/python/tests/test_pysetterminal.py new file mode 100644 index 0000000..fa99218 --- /dev/null +++ b/python/tests/test_pysetterminal.py @@ -0,0 +1,71 @@ +"""Shared terminal macro lifecycle, without physical UART hardware.""" +import os +import unittest +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") +from PySide6.QtWidgets import QApplication +from PySide6.QtTest import QTest +from pysetterminal.rawprotocol import RawProtocolWidget +from pysetterminal.serialbus import SerialBus, SerialTab + + +class PendingProtocol: + def __init__(self): + self.requests = [] + self.callback = None + + def request_async(self, address, value, device_address, callback): + self.requests.append((address, value)) + self.callback = callback + + def cancel_async(self): + callback, self.callback = self.callback, None + if callback: + callback(False, "cancelled") + + +class MacroTests(unittest.TestCase): + def setUp(self): + self.app = QApplication.instance() or QApplication([]) + self.protocol = PendingProtocol() + self.raw = RawProtocolWidget(self.protocol) + self.tab = SerialTab(SerialBus(self.protocol), self.protocol, self.raw) + self.command = {'type': 'read', 'tetrad': 3, 'register': 5} + + def tearDown(self): + self.tab.stop_macro() + self.tab.close() + self.raw.close() + + def test_scheduled_execution_returns_and_can_be_cancelled(self): + self.tab.execute_macro_commands([self.command]) + self.assertEqual(self.protocol.requests, []) + QTest.qWait(20) + self.assertEqual(self.protocol.requests, [(0x200B, 0x35)]) + self.assertTrue(self.tab.stop_macro_btn.isEnabled()) + self.tab.stop_macro() + QTest.qWait(20) + self.assertFalse(self.tab._macro_running) + self.assertTrue(self.tab.bus_read_btn.isEnabled()) + self.assertEqual(len(self.protocol.requests), 1) + + def test_cancelled_scheduled_work_does_not_enter_new_macro(self): + self.tab.execute_macro_commands([self.command]) + self.tab.stop_macro() + self.tab.execute_macro_commands([self.command]) + QTest.qWait(20) + self.assertEqual(len(self.protocol.requests), 1) + self.protocol.callback(False, "CRC error") + QTest.qWait(20) + self.assertFalse(self.tab._macro_running) + self.assertEqual(len(self.protocol.requests), 1) + + def test_completion_runs_next_step_and_updates_value(self): + self.tab.execute_macro_commands([self.command]) + QTest.qWait(20) + self.protocol.callback(True, {'raw': b"ack"}) + QTest.qWait(20) + self.assertEqual(self.protocol.requests[-1], (0x200F, None)) + self.protocol.callback(True, {'value': 0x1234, 'raw_response': {'raw': b"reply"}}) + QTest.qWait(20) + self.assertFalse(self.tab._macro_running) + self.assertEqual(self.tab.bus_read_value_label.text(), "0x1234 (4660)")