Compare commits

..

8 Commits

Author SHA1 Message Date
35c9693740 fix: stabilize OTA updates in v2.0.2
- Исправлен перезапуск приложения после замены файлов и добавлено подтверждение запуска новой версии.
- Обеспечены откат при ошибке и корректное завершение фоновых задач.
- Исключены посторонние DLL из Windows-сборки, обновлены тесты и инструкция ручного перехода с v2.0.1.
2026-09-17 10:30:18 +03:00
3a4491f0c4 feat: add ZIP package updates
- Сборки Windows и Linux теперь дополнительно формируют ZIP-архивы с бинарником и ресурсами.

- OTA-проверка выбирает ZIP-артефакт текущей платформы вместо отдельного исполняемого файла.

- Добавлена безопасная проверка архивов, защита от traversal, ссылок, повреждений и превышения лимитов.

- Обновление транзакционно заменяет все верхнеуровневые файлы и каталоги пакета с резервным копированием и откатом.

- Пользовательские файлы, отсутствующие в ZIP, включая settings.json, сохраняются.

- Версия приложения повышена до 2.0.1, документация и кроссплатформенные тесты обновлены.
2026-09-16 14:07:01 +03:00
add8aff0af feat: release version 2.0.0
- Добавлена обратная совместимость с прошивками, содержащими маркер no_auth.

- Повторная авторизация после команды 04 и успешного обновления прошивки приведена к требуемому порядку.

- В терминал команд добавлена ручная кнопка авторизации.

- Реализовано OTA-обновление приложения с загрузкой платформенного артефакта, безопасной заменой и перезапуском.

- Проверка обновлений учитывает только более новые версии по схеме major, minor и patch.

- Обновлены имена Windows и Linux сборок, документация и тестовое покрытие.
2026-09-16 10:01:35 +03:00
e163b63666 fix: authenticate after firmware update
- удалена повторная авторизация перед отправкой команды 04

- полный AES-256 handshake запускается только после успешного OTA и перезагрузки контроллера

- обновлены сообщения интерфейса и документация порядка обновления

- тесты теперь проверяют, что команда 04 предшествует команде авторизации 07

- версия приложения повышена до 1.5.1

- локальные каталоги релизных артефактов исключены из Git
2026-09-15 13:33:38 +03:00
c34cb0cd7e feat: add authenticated firmware update flow
- добавлена полная повторная авторизация перед командой 04

- добавлена авторизация после успешного обновления и перезагрузки платы

- общий OTA-сценарий подключён к базовому и PRO-интерфейсам

- в терминал команд добавлена кнопка ручной авторизации

- состояние авторизации синхронизировано с основной вкладкой

- добавлены тесты порядка авторизации и обработки ошибок OTA

- версия приложения повышена до 1.5.0

- обновлена документация
2026-09-15 10:49:51 +03:00
5513eb8a56 Improve controller startup handshake and builds
- Добавлен стартовый handshake: ожидание строки версии контроллера, пауза 2000 мс и подтверждение готовности только по READY!.

- Исправлена работа COM-буфера и аппаратного сброса через RTS, добавлены модульные тесты протокола.

- Добавлена нативная сборка для Debian 13 x86_64 и обновлена документация.

- Улучшена очистка Windows-сборки и кроссплатформенная обработка devnull.
2026-09-11 15:01:45 +03:00
914d8defe3 Add planning rule to AGENTS.md
- Добавлено правило в AGENTS.md, обязывающее агента составлять подробный план изменений перед началом выполнения любой нетривиальной задачи и согласовывать его с пользователем.
2026-07-17 15:40:45 +03:00
7f73d2814b Improve OTA update logging and add vertically resizable log panels
- Изменен процесс OTA-обновления (команда 04) в FirmwareDialog (PRO-режим): убрано всплывающее информационное окно, заменено на подробное фоновое логирование процесса (получение прогресса прошивки и перезапуска платы) в реальном времени.
- Внедрен QSplitter во все окна с логами (FirmwareDialog, FirmwareSection, CommandTerminalDialog) для обеспечения возможности ручного изменения высоты панели логов.
- Версия приложения обновлена до 1.4.2.
2026-07-17 15:30:58 +03:00
26 changed files with 3105 additions and 247 deletions

4
.gitignore vendored
View File

@ -14,6 +14,10 @@ env/
# ── PyInstaller build ─────────────────────────────────
build/
dist/
build-linux/
dist-linux/
.venv-linux/
release-builds/
*.manifest
*.spec.bak

View File

@ -12,11 +12,35 @@
## MCP Tool Preference
- При подходящих сценариях агенту следует по максимуму отдавать приоритет MCP-серверам `claude-core`, `claude-explorer` и `claude-rag` вместо встроенных инструментов.
- При подходящих сценариях агенту следует по максимуму отдавать приоритет MCP-серверам `claude-core` и `claude-explorer` вместо встроенных инструментов.
- `claude-core` следует предпочитать для чтения и редактирования файлов, поиска по проекту, работы с задачами и базовых shell-операций, если его возможностей достаточно.
- `claude-explorer` следует предпочитать для изучения архитектуры, исходников, команд и внутренних инструментов Claude Code.
- `claude-rag` следует предпочитать для семантического поиска по проекту, анализа больших объёмов кода и ситуаций, где локальный индекс даёт преимущество над обычным текстовым поиском.
- Память о проекте, его договорённостях, архитектурных решениях, важных гайдах, кодовой базе и ранее выявленном контексте следует по возможности сохранять и повторно использовать через `claude-rag`.
- Кодовую базу проекта также следует по возможности индексировать и использовать через `claude-rag`, чтобы агент мог опираться на локальную семантическую память о структуре, модулях, связях и ранее изученных участках кода.
- При повторяющихся задачах, исследовании крупных подсистем, возвращении к старым решениям, поиске проектного контекста и анализе существующей кодовой базы агенту следует сначала опираться на `claude-rag`, если для этого подходит локальный индекс.
- Если MCP-инструмент хуже подходит для конкретной задачи, недоступен или создаёт лишние ограничения, агент может использовать встроенные инструменты как запасной вариант.
## Project Memory
- **ОБЯЗАТЕЛЬНОЕ ПРАВИЛО**: MCP-сервер `agentmemory` является постоянной долговременной памятью проекта и должен использоваться непрерывно во всех видах работы: при анализе, планировании, программировании, исправлении ошибок, рефакторинге, тестировании, изменении ресурсов и документации.
- Канонический идентификатор проекта во всех вызовах `agentmemory`: `plate-config-app`. Нельзя создавать отдельные области памяти, используя вместо него путь к каталогу, имя приложения или название подсистемы.
- Перед любой задачей первым действием с инструментами необходимо вызвать `memory_recall` или `memory_smart_search` и восстановить относящиеся к задаче требования, архитектуру, прошлые решения, известные ошибки и сведения о затрагиваемых файлах. Перед изменением незнакомой подсистемы либо при обнаружении противоречия поиск в памяти нужно выполнить повторно.
- `agentmemory` следует использовать не только для решений и ошибок. Сохранять необходимо любую новую полезную и повторно используемую информацию о проекте, в том числе:
- архитектуру, зависимости, границы подсистем, потоки данных, состояния, контракты и принятые решения;
- актуальную структуру каталогов, файлов, ресурсов и модулей, их назначение, точки входа и связи;
- устройство кодовой базы: ключевые классы, функции, сигналы, конфигурационные схемы, протоколы, форматы данных, инварианты и тестовую инфраструктуру;
- требования пользователя, ограничения, запреты, договорённости, настройки, внешние интеграции и сведения о соседних проектах;
- содержание и изменения документации, инструкций агентов, дорожной карты, планов реализации и справочных материалов;
- обнаруженные дефекты, причины, способы исправления, опасные места, технический долг и ещё не закрытые вопросы;
- созданные, изменённые, перемещённые и удалённые файлы, затронутые символы и краткое описание фактического изменения;
- команды сборки и тестирования, результаты проверок, особенности среды и подтверждённое состояние реализации.
- Сохранять контекст нужно не только в конце большой задачи, но и после каждого содержательного открытия, архитектурного решения или законченного блока изменений, если эта информация может понадобиться после смены сессии либо другому агенту. Перед финальным ответом пользователю необходимо выполнить итоговый `memory_save`.
- Записи следует делать атомарными и разделять по смыслу и типу (`architecture`, `pattern`, `bug`, `preference`, `fact`). В каждой записи указывать `project: plate-config-app`, релевантные пути в `files`, понятные `concepts` и подробное описание на русском языке. Для кодовой базы сохранять назначение и поведение файлов и символов, а не бесконтрольно дублировать полный текст исходников.
- Память дополняет, но не заменяет репозиторий. Если данные в `agentmemory` расходятся с текущими файлами, конфигурацией или результатами запуска, источником истины считается проверенное текущее состояние проекта; устаревшую запись необходимо исправить либо сохранить новую запись, явно описывающую расхождение и актуальное состояние.
- Не сохранять в память случайный шум, огромные необработанные логи и секреты, отсутствующие в репозитории или явно не разрешённые пользователем. Любые значимые выводы из таких данных сохранять в кратком структурированном виде.
- Если `agentmemory` временно недоступен, необходимо сообщить об этом в рабочем комментарии, продолжить безопасную локальную работу и повторить сохранение при первой возможности. Недоступность памяти нельзя молча игнорировать.
- Если задача передаётся другому агенту, в поручении требуется обязать его сначала восстановить контекст через `agentmemory`, а по завершении сохранить открытия, изменения, проверки и незакрытые вопросы туда же.
- Для долговременной памяти, восстановления контекста и передачи знаний между агентами разрешён только MCP-сервер `agentmemory`. Не использовать параллельные memory-хранилища других MCP-серверов.
- Для фактической работы с репозиторием, приложением, сборкой и тестами использовать подходящие инструменты; это не заменяет обязательную синхронизацию результатов с `agentmemory`.
## Planning Before Task Execution
- Перед тем как приступать к выполнению любой задачи (кроме тривиальных исправлений опечаток или ответов на вопросы), агент обязан составить подробный план реализации изменений.
- План должен быть предоставлен пользователю на проверку и одобрение. Агент не должен приступать к написанию кода или выполнению команд до получения явного подтверждения от пользователя.

View File

@ -1,5 +1,24 @@
# Сборка EXE
## Linux (Debian 13 x86_64)
Для Linux используется нативная сборка: Windows-версия PyInstaller не может
создать рабочий Linux-бинарник. В Debian 13 установите Python и средства
сборки, затем запустите скрипт из корня проекта:
```bash
sudo apt-get update
sudo apt-get install -y python3 python3-venv python3-pip libglib2.0-0t64 libgl1 libegl1 libfontconfig1 libxkbcommon0 libfreetype6 libxcb-cursor0 libxcb-icccm4 libxcb-image0 libxcb-keysyms1 libxcb-render-util0 libxcb-shape0 libxcb-xkb1 libxcb-util1 libxkbcommon-x11-0
chmod +x build_linux.sh
./build_linux.sh
```
Результат: бинарник `dist-linux/plate_config_app-linux-x86_64`, папки
конфигураций рядом с ним и архив
`dist-linux/plate_config_app-linux-x86_64.zip`. Для автообновления публикуется
ZIP-архив. Для работы с USB-COM пользователь Debian должен состоять в группе
`dialout`.
## Предварительные требования
- Python ≥ 3.11 (виртуальное окружение `.venv`)
@ -12,7 +31,9 @@
.\build_exe.ps1 -InstallBuildDeps
```
Готовый файл: `dist\plate_config_app.exe`
Готовые файлы: `dist\plate_config_app-windows-x86_64.exe` и
`dist\plate_config_app-windows-x86_64.zip`. Для автообновления публикуется
ZIP-архив.
## Команды по отдельности
@ -32,8 +53,10 @@ python -m pip install -r requirements-build.txt
Скрипт автоматически:
- очищает `build/`, `dist/`, `__pycache__/`
- запускает PyInstaller по `plate_config_app.spec`
- переименовывает бинарник в `plate_config_app-windows-x86_64.exe`
- копирует папки `light_modes/`, `input_mappings/`, `commands/` в `dist/`
- выводит путь и размер готового EXE
- создаёт `plate_config_app-windows-x86_64.zip` с бинарником и ресурсами
- выводит пути и размеры готовых EXE и ZIP
### 3. Только очистка (без сборки)
@ -75,4 +98,5 @@ Hidden imports: `qtawesome`, `Crypto.Cipher._raw_aes`, `Crypto.Util._raw_api`
| `commands/` | Наборы команд (.csv) |
Эти файлы можно редактировать без пересборки — достаточно изменить их
рядом с EXE и перезапустить приложение.
рядом с EXE и перезапустить приложение. При установке OTA-обновления каталоги,
которые входят в новый ZIP-пакет, полностью заменяются их новыми версиями.

View File

@ -123,6 +123,11 @@ plate_config_app/
Сервисы загружают данные с диска, управляют COM-соединением,
выполняют авторизацию. Могут использоваться без GUI.
`ComConnection` хранит версию текущей прошивки и режим авторизации,
определённые по стартовой строке. Маркер `no_auth` проверяется без учёта
регистра; для неизвестной версии применяется строгий режим с AES-256
handshake. После переподключения и OTA режим определяется заново.
2. **ui/** -- визуальное оформление. `theme.py` содержит палитры
(`ThemePalette`) и QSS-шаблон с подстановкой переменных.
`icons.py` -- обёртка над `qtawesome` с fallback при отсутствии
@ -196,5 +201,6 @@ health-check) вынесены в `QThread`-подклассы. Связь с UI
.\build_exe.ps1 -InstallBuildDeps
```
Результат: `dist\plate_config_app.exe` + папки `dist\light_modes\`,
Результат: `dist\plate_config_app-windows-x86_64.exe`, ZIP-архив
`dist\plate_config_app-windows-x86_64.zip` и папки `dist\light_modes\`,
`dist\input_mappings\`, `dist\commands\`.

View File

@ -10,6 +10,7 @@
- [Системные требования](#системные-требования)
- [Установка и запуск](#установка-и-запуск)
- [Обновление приложения](#обновление-приложения)
- [Интерфейс приложения](#интерфейс-приложения)
- [Главная страница](#главная-страница)
- [Подключение к плате](#подключение-к-плате)
@ -23,6 +24,7 @@
## Системные требования
- Windows 10 / 11
- Debian 13 x86_64 (для Linux-сборки и запуска Linux-пакета)
- COM-порт (физический или виртуальный), к которому подключена плата
Для запуска из собранного EXE дополнительных зависимостей не требуется.
@ -31,10 +33,11 @@
## Установка и запуск
### Из готового EXE
### Из готовой сборки
1. Скопируйте `plate_config_app.exe` в любую папку.
2. Запустите двойным кликом.
1. Скопируйте сборку для своей платформы в каталог с папками ресурсов.
2. На Windows запустите `plate_config_app-windows-x86_64.exe`.
3. На Debian запустите `./plate_config_app-linux-x86_64`.
Файл настроек `settings.json` создается автоматически рядом с EXE при первом
изменении настроек.
@ -43,6 +46,38 @@
См. [DEVELOPMENT.md](DEVELOPMENT.md).
Для Debian 13 x86_64 используйте [build_linux.sh](build_linux.sh); подробности
сборки приведены в [BUILD_EXE.md](BUILD_EXE.md).
---
## Обновление приложения
При запуске приложение в фоне переходит по адресу
`https://git.robotail.pro/Robotail/plate-config-app/releases/latest` и сравнивает
тег последнего релиза с текущей версией. Кнопка «Скачать обновление» появляется
под настройками только тогда, когда удалённый тег формата `vX.Y.Z` численно
новее: сначала сравнивается major, затем minor и patch.
Приложение скачивает ZIP-пакет текущей платформы, проверяет его целостность и
безопасность, завершает старый процесс, заменяет бинарник и все каталоги из
пакета, а затем запускается заново. Перед заменой создаётся резервная копия;
она удаляется только после подтверждения запуска новой версии. Если запуск
не удался, предыдущая версия восстанавливается, а сообщение записывается
в `update-error.log`. Файл `settings.json` и другие файлы,
которых нет в архиве, сохраняются. Для работы проверки релизы репозитория должны
быть доступны без авторизации.
Имена release-артефактов:
- Windows x86_64: `plate_config_app-windows-x86_64.zip`;
- Linux x86_64: `plate_config_app-linux-x86_64.zip`.
Для перехода с уже установленной v2.0.1 на v2.0.2 обновите приложение вручную:
закройте его, скачайте ZIP-пакет своей платформы и распакуйте содержимое в
каталог приложения с заменой прежних файлов. `settings.json` удалять не нужно.
Начиная с v2.0.2 последующие обновления можно устанавливать кнопкой в приложении.
---
## Интерфейс приложения
@ -55,6 +90,8 @@
- **Команды платы** -- справочник поддерживаемых команд платы с поиском.
- **Настройки приложения** -- выбор цветовой темы (светлая / темная, зеленый /
синий акцент).
- **Скачать обновление** -- появляется под настройками, если версия последнего
релиза формата `vX.Y.Z` новее версии запущенного приложения.
Приложение поддерживает несколько одновременных подключений: каждое соединение
открывается в отдельной вкладке.
@ -114,6 +151,15 @@
- Зеленый — подключено, плата готова к работе.
Кнопка переподключения (↻) позволяет быстро восстановить связь при обрыве.
При подключении приложение определяет режим по версии в стартовой строке платы.
Если версия содержит маркер `no_auth` без учёта регистра, AES-256 авторизация
пропускается, а статус отображается как «Готово без авторизации». Для остальных
и неизвестных версий сохраняется строгая обязательная авторизация.
После успешного OTA и перезагрузки приложение повторно читает версию новой
прошивки. Полная AES-256 авторизация выполняется только для прошивок без маркера
`no_auth`; перед командой `04` повторная авторизация не выполняется. Кнопка
ручной авторизации в терминале использует то же правило совместимости.
---

View File

@ -1,3 +1,3 @@
"""Plate Config App package."""
__version__ = "1.4.1"
__version__ = "2.0.2"

View File

@ -2,11 +2,13 @@ from __future__ import annotations
import sys
from PySide6.QtCore import QTimer
from PySide6.QtGui import QFont
from PySide6.QtWidgets import QApplication
from .main_window import MainWindow
from .services import settings_service
from .services.update_service import signal_update_ready
from .ui.icons import ICON_COLOR_PRIMARY, icon
from .ui.theme import build_stylesheet, get_palette, create_qpalette
@ -27,4 +29,5 @@ def main() -> None:
if not app_icon.isNull():
window.setWindowIcon(app_icon)
window.show()
QTimer.singleShot(0, signal_update_ready)
sys.exit(app.exec())

View File

@ -119,6 +119,8 @@ class MainWindow(QMainWindow):
widget = self.pages.widget(i)
if isinstance(widget, ConnectionTab):
widget.close_serial()
elif isinstance(widget, StartTab):
widget.shutdown()
super().closeEvent(event)
def _attach_close_button(self, index: int) -> None:

View File

@ -1,17 +1,23 @@
"""AES-256 двойное рукопожатие (авторизация) с платой периферийного контроллера.
"""Запуск и AES-256 авторизация платы периферийного контроллера.
Алгоритм (из документации «Система команд платы v2.x»):
1. Компьютер → плата: 07<32 hex> (16 случайных байт)
2. Плата → компьютер: CHAL<32 hex> (встречный запрос, 16 байт)
3. Компьютер → плата: 08<32 hex> (зашифрованный ответ на CHAL)
4. Плата → компьютер: RAES<32 hex> (ответ на запрос из шага 1)
Стартовая последовательность:
1. Плата → компьютер: periphery controller ver_<версия и комментарий>
2. Компьютер ждёт 2000 мс.
3. Компьютер → плата: 07<32 hex> (16 случайных байт)
4. Плата → компьютер: CHAL<32 hex> (встречный запрос, 16 байт)
5. Компьютер → плата: 08<32 hex> (зашифрованный ответ на CHAL)
6. Плата → компьютер: RAES<32 hex> (ответ на запрос из шага 3)
7. Плата → компьютер: READY!
Компьютер проверяет RAES — расшифровывает и сравнивает с исходными 16 байт.
Авторизация считается завершённой только после точной строки ``READY!``.
"""
from __future__ import annotations
import hmac
import os
import time
from collections.abc import Callable
from dataclasses import dataclass
from enum import Enum, auto
@ -19,13 +25,21 @@ from Crypto.Cipher import AES
from .com_service import ComConnection
# ── Фиксированный AES-256 ключ (32 байта) ──────────────────────────────
_AES_KEY = bytes((
STARTUP_BANNER_PREFIX = "periphery controller ver_"
NO_AUTH_MARKER = "no_auth"
STARTUP_DELAY_SECONDS = 2.0
STARTUP_BANNER_TIMEOUT_SECONDS = 20.0
READY_TIMEOUT_SECONDS = 5.0
# Ключ хранится в том же обфусцированном виде, что и в рабочем клиенте.
_AES_KEY_MASK = 0x37
_AES_OBFUSCATED_KEY = bytes((
0x69, 0x40, 0xC7, 0x0E, 0xF2, 0xC7, 0x45, 0x70,
0x72, 0xCD, 0x28, 0x58, 0x99, 0x8F, 0x73, 0xA8,
0x5F, 0x5F, 0xED, 0xB4, 0xC1, 0x62, 0xB4, 0x24,
0xAC, 0xD7, 0xA3, 0xD6, 0xD8, 0x0B, 0x50, 0x37,
))
_AES_KEY = bytes(value ^ _AES_KEY_MASK for value in _AES_OBFUSCATED_KEY)
class AuthStatus(Enum):
@ -40,6 +54,23 @@ class AuthStatus(Enum):
class AuthResult:
status: AuthStatus
detail: str = ""
startup_ready: bool = False
firmware_version: str | None = None
auth_skipped: bool = False
def firmware_requires_auth(version: str) -> bool:
"""Определить режим авторизации по строке версии прошивки."""
return NO_AUTH_MARKER not in version.casefold()
def _remember_firmware(conn: ComConnection, banner: str) -> str:
version = banner[len(STARTUP_BANNER_PREFIX):].strip()
conn.set_firmware_version(
version,
auth_required=firmware_requires_auth(version),
)
return version
def _aes_encrypt(plaintext: bytes, key: bytes = _AES_KEY) -> bytes:
@ -53,14 +84,45 @@ def _aes_decrypt(ciphertext: bytes, key: bytes = _AES_KEY) -> bytes:
return cipher.decrypt(ciphertext)
def _read_until(
conn: ComConnection,
predicate: Callable[[str], bool],
timeout: float,
) -> str | None:
"""Читать полные строки до совпадения или истечения общего таймаута."""
deadline = time.monotonic() + timeout
while conn.is_connected:
remaining = deadline - time.monotonic()
if remaining <= 0:
return None
line = conn.read_response(timeout=remaining)
if line is None:
return None
if predicate(line.strip()):
return line.strip()
return None
def wait_for_startup_banner(
conn: ComConnection,
timeout: float = STARTUP_BANNER_TIMEOUT_SECONDS,
) -> str | None:
"""Дождаться ``periphery controller ver_<...>`` с непустым суффиксом."""
def is_banner(line: str) -> bool:
if not line.startswith(STARTUP_BANNER_PREFIX):
return False
return bool(line[len(STARTUP_BANNER_PREFIX):].strip())
return _read_until(conn, is_banner, timeout)
def check_status(conn: ComConnection, timeout: float = 3.0) -> AuthStatus:
"""Отправить команду 99 и проанализировать ответ.
* ``READY`` в ответе → плата авторизована и готова (``OK``).
* Любой другой *непустой* ответ → плата жива, авторизация
не требуется или уже пройдена (``OK``). Это покрывает
платы без системы авторизации.
* Полный таймаут (ни одного ответа) → ``NEED_AUTH``.
Для прошивки с авторизацией только точная строка ``READY!`` означает
готовность. Для обнаруженной прошивки ``no_auth`` достаточно любого
непустого ответа: отсутствие ответа считается потерей связи, а не
запросом авторизации.
"""
if not conn.is_connected:
return AuthStatus.DISCONNECTED
@ -70,26 +132,47 @@ def check_status(conn: ComConnection, timeout: float = 3.0) -> AuthStatus:
return AuthStatus.DISCONNECTED
try:
import time
deadline = time.monotonic() + timeout
got_any_response = False
while time.monotonic() < deadline:
remaining = max(deadline - time.monotonic(), 0.05)
resp = conn.read_response(timeout=min(remaining, 0.5))
if resp is not None:
got_any_response = True
if "READY" in resp.upper():
return AuthStatus.OK
# Не прерываемся на None — ждём до deadline
# Если хоть один ответ пришёл — плата жива (без авторизации)
if got_any_response:
return AuthStatus.OK
return AuthStatus.NEED_AUTH
if conn.auth_required is False:
response = _read_until(conn, bool, timeout)
return AuthStatus.OK if response is not None else AuthStatus.DISCONNECTED
ready = _read_until(conn, lambda line: line == "READY!", timeout)
return AuthStatus.OK if ready is not None else AuthStatus.NEED_AUTH
finally:
conn.release()
def perform_handshake(conn: ComConnection, timeout: float = 2.0) -> AuthResult:
def perform_required_handshake(
conn: ComConnection,
timeout: float = 2.0,
ready_timeout: float = READY_TIMEOUT_SECONDS,
) -> AuthResult:
"""Выполнить handshake только если он требуется текущей прошивке."""
if not conn.is_connected:
return AuthResult(AuthStatus.DISCONNECTED, "Порт не открыт")
if conn.auth_required is False:
version = conn.firmware_version
detail = "Авторизация не требуется: обнаружена прошивка no_auth"
if version:
detail = f"Прошивка {version}: авторизация не требуется"
return AuthResult(
AuthStatus.OK,
detail,
firmware_version=version,
auth_skipped=True,
)
return perform_handshake(
conn,
timeout=timeout,
ready_timeout=ready_timeout,
)
def perform_handshake(
conn: ComConnection,
timeout: float = 2.0,
ready_timeout: float = READY_TIMEOUT_SECONDS,
) -> AuthResult:
"""Выполнить полный цикл двойного рукопожатия.
Returns:
@ -108,61 +191,116 @@ def perform_handshake(conn: ComConnection, timeout: float = 2.0) -> AuthResult:
conn.release()
return AuthResult(AuthStatus.FAILED, f"Ошибка отправки 07: {exc}")
# ── Шаг 2: ждём CHAL<32 hex> от контроллера ──
resp = conn.read_response(timeout=timeout)
if resp is None:
conn.release()
return AuthResult(AuthStatus.TIMEOUT, "Таймаут ожидания CHAL от контроллера")
try:
# ── Шаг 2: ждём CHAL<32 hex> от контроллера ──
resp = _read_until(conn, lambda line: line.upper().startswith("CHAL"), timeout)
if resp is None:
return AuthResult(AuthStatus.TIMEOUT, "Таймаут ожидания CHAL от контроллера")
# Парсим CHAL — допускаем разделители (пробел, двоеточие)
if not resp.upper().startswith("CHAL"):
conn.release()
return AuthResult(AuthStatus.FAILED, f"Неожиданный ответ вместо CHAL: {resp}")
# Парсим CHAL — допускаем разделители (пробел, двоеточие)
board_challenge_hex = resp[4:].strip().lstrip(":").strip()
if len(board_challenge_hex) != 32:
return AuthResult(
AuthStatus.FAILED,
f"Некорректная длина CHAL: {board_challenge_hex}",
)
# Извлекаем hex-часть, убирая возможные разделители
board_challenge_hex = resp[4:].strip().lstrip(":").strip()
try:
board_challenge = bytes.fromhex(board_challenge_hex)
except ValueError:
return AuthResult(
AuthStatus.FAILED,
f"CHAL содержит не-hex символы: {board_challenge_hex}",
)
if len(board_challenge_hex) != 32:
# ── Шаг 3: шифруем board_challenge AES-256 и отправляем 08<hex> ──
encrypted_response = _aes_encrypt(board_challenge)
cmd_08 = "08" + encrypted_response.hex().upper()
try:
conn.write_raw(cmd_08)
except Exception as exc:
return AuthResult(AuthStatus.FAILED, f"Ошибка отправки 08: {exc}")
# ── Шаг 4: проверяем доказательство контроллера RAES ──
raes = _read_until(conn, lambda line: line.upper().startswith("RAES"), timeout)
if raes is None:
return AuthResult(AuthStatus.TIMEOUT, "Таймаут ожидания RAES от контроллера")
controller_proof_hex = raes[4:].strip().lstrip(":").strip()
try:
controller_proof = bytes.fromhex(controller_proof_hex)
except ValueError:
return AuthResult(AuthStatus.FAILED, "RAES содержит некорректные hex-данные")
expected_proof = _aes_encrypt(our_challenge)
if len(controller_proof) != 16 or not hmac.compare_digest(
controller_proof,
expected_proof,
):
return AuthResult(AuthStatus.FAILED, "Проверка RAES контроллера не пройдена")
# AUTH_PC_OK и другие промежуточные строки игнорируются. Успех —
# только точная отдельная строка READY! после RAES.
ready = _read_until(conn, lambda line: line == "READY!", ready_timeout)
if ready is None:
return AuthResult(AuthStatus.TIMEOUT, "Таймаут ожидания READY! от контроллера")
return AuthResult(AuthStatus.OK, "Авторизация подтверждена строкой READY!")
finally:
conn.release()
return AuthResult(AuthStatus.FAILED, f"Некорректная длина CHAL: {board_challenge_hex}")
def perform_startup_handshake(
conn: ComConnection,
*,
banner_timeout: float = STARTUP_BANNER_TIMEOUT_SECONDS,
startup_delay: float = STARTUP_DELAY_SECONDS,
handshake_timeout: float = 2.0,
ready_timeout: float = READY_TIMEOUT_SECONDS,
sleep_fn: Callable[[float], None] = time.sleep,
) -> AuthResult:
"""Дождаться загрузки и выполнить handshake, если он требуется."""
if not conn.is_connected:
return AuthResult(AuthStatus.DISCONNECTED, "Порт не открыт")
try:
board_challenge = bytes.fromhex(board_challenge_hex)
except ValueError:
conn.release()
return AuthResult(AuthStatus.FAILED, f"CHAL содержит не-hex символы: {board_challenge_hex}")
# ── Шаг 3: шифруем board_challenge AES-256 и отправляем 08<hex> ──
encrypted_response = _aes_encrypt(board_challenge)
cmd_08 = "08" + encrypted_response.hex().upper()
try:
conn.write_raw(cmd_08)
banner = wait_for_startup_banner(conn, timeout=banner_timeout)
except Exception as exc:
conn.release()
return AuthResult(AuthStatus.FAILED, f"Ошибка отправки 08: {exc}")
return AuthResult(AuthStatus.FAILED, f"Ошибка ожидания строки запуска: {exc}")
# ── Шаг 4: ждём RAES<32 hex> — зашифрованный ответ на наш запрос ──
resp2 = conn.read_response(timeout=timeout)
if resp2 is None:
conn.release()
return AuthResult(AuthStatus.TIMEOUT, "Таймаут ожидания RAES от контроллера")
if banner is None:
return AuthResult(
AuthStatus.TIMEOUT,
f"Таймаут ожидания {STARTUP_BANNER_PREFIX}<версия>",
)
upper2 = resp2.upper()
if upper2.startswith("RAES"):
# RAES получен — плата ответила на наш challenge.
# Верификацию RAES пропускаем (плата уже приняла наш 08).
pass
elif "READY" in upper2 or "AUTH_CNT_OK" in upper2:
conn.release()
return AuthResult(AuthStatus.OK, f"Авторизация пройдена ({resp2})")
else:
conn.release()
return AuthResult(AuthStatus.FAILED, f"Неожиданный ответ вместо RAES: {resp2}")
version = _remember_firmware(conn, banner)
sleep_fn(startup_delay)
if not conn.is_connected:
return AuthResult(
AuthStatus.DISCONNECTED,
"Порт закрыт до проверки режима авторизации",
startup_ready=True,
firmware_version=version,
)
# Ждём AUTH_CNT_OK / READY после RAES
ready_resp = conn.read_response(timeout=timeout)
conn.release()
if ready_resp:
return AuthResult(AuthStatus.OK, f"Авторизация пройдена ({ready_resp})")
if not firmware_requires_auth(version):
conn.release()
return AuthResult(
AuthStatus.OK,
f"{banner}\nАвторизация не требуется для прошивки no_auth",
startup_ready=True,
firmware_version=version,
auth_skipped=True,
)
return AuthResult(AuthStatus.OK, "Авторизация пройдена")
result = perform_handshake(
conn,
timeout=handshake_timeout,
ready_timeout=ready_timeout,
)
result.startup_ready = True
result.firmware_version = version
if result.status == AuthStatus.OK:
result.detail = f"{banner}\n{result.detail}"
return result

View File

@ -33,6 +33,7 @@ PARITY_MAP = {
"Space": PARITY_SPACE,
}
PARITY_SHORT_MAP = {"None": "N", "Even": "E", "Odd": "O", "Mark": "M", "Space": "S"}
STARTUP_RESET_PULSE_SECONDS = 0.12
def list_available_ports() -> list[str]:
@ -90,6 +91,8 @@ class ComConnection(QObject):
# Параметры последнего подключения (для reconnect)
self._last_port: str | None = None
self._last_params: SerialParams | None = None
self._firmware_version: str | None = None
self._auth_required: bool | None = None
# ── Properties ────────────────────────────────────────────────────
@ -101,6 +104,20 @@ class ComConnection(QObject):
def serial(self) -> Serial | None:
return self._serial
@property
def firmware_version(self) -> str | None:
return self._firmware_version
@property
def auth_required(self) -> bool | None:
"""Требуется ли авторизация по данным текущей прошивки."""
return self._auth_required
def set_firmware_version(self, version: str, *, auth_required: bool) -> None:
"""Сохранить версию и обнаруженный режим авторизации прошивки."""
self._firmware_version = version.strip()
self._auth_required = auth_required
@property
def seconds_since_last_command(self) -> float:
"""Сколько секунд прошло с последней отправки команды."""
@ -110,7 +127,15 @@ class ComConnection(QObject):
# ── Open / Close ──────────────────────────────────────────────────
def open(self, *, port: str, params: SerialParams, timeout: float = 0.1) -> None:
def open(
self,
*,
port: str,
params: SerialParams,
timeout: float = 0.1,
drain_paused: bool = False,
reset_device: bool = False,
) -> None:
self.close()
self._last_port = port
self._last_params = params
@ -124,9 +149,19 @@ class ComConnection(QObject):
rtscts=params.rtscts,
timeout=timeout,
)
# При старте авторизации reader должен не забрать загрузочный banner.
if drain_paused:
self._drain_paused.set()
else:
self._drain_paused.clear()
self._read_buf.clear()
self._event_buf.clear()
if reset_device:
self.reset_controller()
# Запускаем фоновый дренаж
self._drain_stop.clear()
self._event_buf.clear()
self._drain_thread = threading.Thread(
target=self._drain_loop, daemon=True, name="serial-drain"
)
@ -148,9 +183,17 @@ class ComConnection(QObject):
if self._serial and self._serial.is_open:
self._serial.close()
self._serial = None
self._read_buf.clear()
self._event_buf.clear()
self._firmware_version = None
self._auth_required = None
def reconnect(self) -> None:
def reconnect(
self,
*,
drain_paused: bool = False,
reset_device: bool = False,
) -> None:
"""Переподключиться с параметрами последнего open().
Raises:
@ -159,7 +202,28 @@ class ComConnection(QObject):
"""
if self._last_port is None or self._last_params is None:
raise RuntimeError("Нет сохранённых параметров для reconnect")
self.open(port=self._last_port, params=self._last_params)
self.open(
port=self._last_port,
params=self._last_params,
drain_paused=drain_paused,
reset_device=reset_device,
)
def reset_controller(self, pulse_seconds: float = STARTUP_RESET_PULSE_SECONDS) -> None:
"""Перезагрузить ESP32 импульсом RTS, не закрывая COM-порт."""
if not self.is_connected:
raise RuntimeError("Порт не открыт")
assert self._serial is not None
self._read_buf.clear()
self._event_buf.clear()
self._serial.reset_input_buffer()
self._serial.dtr = False
self._serial.rts = True
try:
_time.sleep(pulse_seconds)
finally:
self._serial.rts = False
# ── Send / Receive ────────────────────────────────────────────────
@ -240,23 +304,17 @@ class ComConnection(QObject):
if line is not None:
return line
old_timeout = self._serial.timeout
try:
while time.monotonic() < deadline:
remaining = max(deadline - time.monotonic(), 0.001)
self._serial.timeout = min(remaining, 0.05)
b = self._serial.read(1)
if not b:
continue
buf.extend(b)
line = self._extract_line(buf)
if line is not None:
return line
finally:
try:
self._serial.timeout = old_timeout
except Exception:
pass
# Не менять Serial.timeout в цикле: в Windows setter каждый раз
# перенастраивает драйвер COM и падает на длинном загрузочном логе.
while time.monotonic() < deadline:
available = self._serial.in_waiting
chunk = self._serial.read(available or 1)
if not chunk:
continue
buf.extend(chunk)
line = self._extract_line(buf)
if line is not None:
return line
# Таймаут — вернём что накопилось
if buf:

View File

@ -0,0 +1,100 @@
"""Запуск OTA с повторной авторизацией после обновления контроллера."""
from __future__ import annotations
import time
from collections.abc import Callable
from dataclasses import dataclass
from .auth_service import (
AuthResult,
AuthStatus,
READY_TIMEOUT_SECONDS,
STARTUP_BANNER_TIMEOUT_SECONDS,
STARTUP_DELAY_SECONDS,
perform_startup_handshake,
)
from .com_service import ComConnection
OTA_TIMEOUT_SECONDS = 60.0
@dataclass
class FirmwareUpdateResult:
success: bool
detail: str
post_auth: AuthResult | None = None
def perform_authenticated_ota(
conn: ComConnection,
*,
log_message: Callable[[str], None] | None = None,
ota_timeout: float = OTA_TIMEOUT_SECONDS,
banner_timeout: float = STARTUP_BANNER_TIMEOUT_SECONDS,
startup_delay: float = STARTUP_DELAY_SECONDS,
handshake_timeout: float = 2.0,
ready_timeout: float = READY_TIMEOUT_SECONDS,
sleep_fn: Callable[[float], None] = time.sleep,
) -> FirmwareUpdateResult:
"""Выполнить ``04`` и полностью авторизоваться после успешного OTA."""
emit = log_message or (lambda _message: None)
if not conn.is_connected:
return FirmwareUpdateResult(False, "Порт не подключён")
try:
conn.send_command("04")
emit("→ 04")
deadline = time.monotonic() + ota_timeout
update_confirmed = False
while time.monotonic() < deadline and conn.is_connected:
remaining = max(deadline - time.monotonic(), 0.1)
response = conn.read_response(timeout=min(remaining, 2.0))
if response is None:
continue
emit(f"← {response}")
if "successful update" in response.lower():
update_confirmed = True
break
if not update_confirmed:
return FirmwareUpdateResult(
False,
"Таймаут ожидания подтверждения успешного обновления",
)
emit("✓ Обновление подтверждено. Ожидание перезагрузки платы…")
post_auth = perform_startup_handshake(
conn,
banner_timeout=banner_timeout,
startup_delay=startup_delay,
handshake_timeout=handshake_timeout,
ready_timeout=ready_timeout,
sleep_fn=sleep_fn,
)
if post_auth.status != AuthStatus.OK:
return FirmwareUpdateResult(
False,
f"Авторизация после OTA не выполнена: {post_auth.detail}",
post_auth=post_auth,
)
if post_auth.auth_skipped:
emit("✓ Прошивка no_auth: авторизация после OTA не требуется.")
detail = "Прошивка обновлена, плата готова без авторизации"
else:
emit("✓ Авторизация после OTA выполнена.")
detail = "Прошивка обновлена, плата перезагружена и авторизована"
return FirmwareUpdateResult(
True,
detail,
post_auth=post_auth,
)
except Exception as exc:
return FirmwareUpdateResult(
False,
f"Ошибка OTA: {exc}",
)
finally:
conn.release()

View File

@ -0,0 +1,648 @@
"""Проверка, загрузка и установка обновлений Plate Config App."""
from __future__ import annotations
import os
import platform
import re
import shutil
import stat
import subprocess
import sys
import uuid
import zipfile
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import BinaryIO
from urllib.parse import quote, unquote, urlparse
from urllib.request import Request, urlopen
LATEST_RELEASE_URL = (
"https://git.robotail.pro/Robotail/plate-config-app/releases/latest"
)
RELEASE_DOWNLOAD_BASE_URL = (
"https://git.robotail.pro/Robotail/plate-config-app/releases/download"
)
UPDATE_HOST = "git.robotail.pro"
HTTP_TIMEOUT_SECONDS = 10.0
DOWNLOAD_TIMEOUT_SECONDS = 30.0
DOWNLOAD_CHUNK_SIZE = 1024 * 1024
MAX_DOWNLOAD_BYTES = 300 * 1024 * 1024
MAX_UNPACKED_BYTES = 600 * 1024 * 1024
MAX_ARCHIVE_MEMBERS = 10_000
UPDATE_READY_ENV = "PLATE_CONFIG_UPDATE_READY_FILE"
VERSION_PATTERN = re.compile(
r"[vV]?(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)"
)
TOP_LEVEL_NAME_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]*")
class UpdateError(RuntimeError):
"""Ошибка проверки, загрузки или запуска обновления."""
class UpdateCancelled(UpdateError):
"""Загрузка обновления отменена пользователем или закрытием приложения."""
@dataclass(frozen=True)
class UpdateInfo:
tag: str
version: str
asset_name: str
download_url: str
def normalize_version(value: str) -> str:
normalized = value.strip()
if normalized[:1].casefold() == "v":
normalized = normalized[1:]
return normalized
def parse_version(value: str) -> tuple[int, int, int]:
"""Преобразовать версию X.Y.Z или vX.Y.Z в числовой кортеж."""
match = VERSION_PATTERN.fullmatch(value.strip())
if match is None:
raise UpdateError(
f"Некорректная версия {value!r}: ожидается формат vX.Y.Z"
)
return tuple(int(component) for component in match.groups())
def platform_executable_name(
*,
system: str | None = None,
machine: str | None = None,
) -> str:
system_name = (system or platform.system()).casefold()
machine_name = (machine or platform.machine()).casefold()
if machine_name not in {"amd64", "x86_64"}:
raise UpdateError(f"Автообновление не поддерживает архитектуру {machine_name}")
if system_name == "windows":
return "plate_config_app-windows-x86_64.exe"
if system_name == "linux":
return "plate_config_app-linux-x86_64"
raise UpdateError(f"Автообновление не поддерживает платформу {system_name}")
def platform_asset_name(
*,
system: str | None = None,
machine: str | None = None,
) -> str:
executable_name = platform_executable_name(system=system, machine=machine)
if executable_name.casefold().endswith(".exe"):
executable_name = executable_name[:-4]
return f"{executable_name}.zip"
def archive_executable_name(asset_name: str) -> str:
names = {
"plate_config_app-windows-x86_64.zip": (
"plate_config_app-windows-x86_64.exe"
),
"plate_config_app-linux-x86_64.zip": "plate_config_app-linux-x86_64",
}
try:
return names[asset_name]
except KeyError as exc:
raise UpdateError("Неизвестное имя архива обновления") from exc
def extract_release_tag(final_url: str) -> str:
parsed = urlparse(final_url)
if parsed.scheme != "https" or parsed.hostname != UPDATE_HOST:
raise UpdateError("Сервер релизов перенаправил запрос на неизвестный адрес")
parts = [unquote(part) for part in parsed.path.split("/") if part]
try:
release_index = parts.index("releases")
except ValueError as exc:
raise UpdateError("Не удалось определить тег последнего релиза") from exc
if parts[release_index + 1:release_index + 2] != ["tag"]:
raise UpdateError("Не удалось определить тег последнего релиза")
tag_parts = parts[release_index + 2:release_index + 3]
if not tag_parts or not tag_parts[0].strip():
raise UpdateError("Сервер релизов вернул пустой тег")
return tag_parts[0].strip()
def build_download_url(tag: str, asset_name: str) -> str:
encoded_tag = quote(tag, safe="")
encoded_asset = quote(asset_name, safe="")
return f"{RELEASE_DOWNLOAD_BASE_URL}/{encoded_tag}/{encoded_asset}"
def check_for_update(
current_version: str,
*,
timeout: float = HTTP_TIMEOUT_SECONDS,
opener: Callable[..., BinaryIO] | None = None,
system: str | None = None,
machine: str | None = None,
) -> UpdateInfo | None:
"""Перейти по latest URL и вернуть данные отличающегося релиза."""
request = Request(
LATEST_RELEASE_URL,
headers={
"Accept": "text/html,application/xhtml+xml",
"User-Agent": f"PlateConfigApp/{normalize_version(current_version)}",
},
)
open_request = opener or urlopen
try:
with open_request(request, timeout=timeout) as response:
final_url = response.geturl()
except Exception as exc:
raise UpdateError(f"Не удалось проверить обновления: {exc}") from exc
tag = extract_release_tag(final_url)
remote_version = parse_version(tag)
if remote_version <= parse_version(current_version):
return None
version = normalize_version(tag)
asset_name = platform_asset_name(system=system, machine=machine)
return UpdateInfo(
tag=tag,
version=version,
asset_name=asset_name,
download_url=build_download_url(tag, asset_name),
)
def _validate_executable_signature(signature: bytes, executable_name: str) -> None:
if executable_name.casefold().endswith(".exe"):
valid = signature.startswith(b"MZ")
else:
valid = signature == b"\x7fELF"
if not valid:
raise UpdateError("Сервер вернул файл, не похожий на исполняемый артефакт")
def _validated_archive_members(
archive: zipfile.ZipFile,
executable_name: str,
) -> list[zipfile.ZipInfo]:
members = archive.infolist()
if len(members) > MAX_ARCHIVE_MEMBERS:
raise UpdateError("В архиве обновления слишком много файлов")
total_size = 0
normalized_names: set[str] = set()
executable_member: zipfile.ZipInfo | None = None
for member in members:
raw_name = member.filename
if not raw_name or "\\" in raw_name or "\x00" in raw_name:
raise UpdateError("Архив обновления содержит небезопасный путь")
relative_path = PurePosixPath(raw_name.rstrip("/"))
if (
relative_path.is_absolute()
or not relative_path.parts
or any(part in {"", ".", ".."} for part in relative_path.parts)
):
raise UpdateError("Архив обновления содержит небезопасный путь")
if not TOP_LEVEL_NAME_PATTERN.fullmatch(relative_path.parts[0]):
raise UpdateError("Архив обновления содержит недопустимое имя")
normalized_name = relative_path.as_posix().casefold()
if normalized_name in normalized_names:
raise UpdateError("Архив обновления содержит повторяющиеся пути")
normalized_names.add(normalized_name)
unix_mode = member.external_attr >> 16
file_type = stat.S_IFMT(unix_mode)
if file_type not in {0, stat.S_IFREG, stat.S_IFDIR}:
raise UpdateError("Архив обновления содержит ссылку или специальный файл")
if member.flag_bits & 0x1:
raise UpdateError("Зашифрованные архивы обновления не поддерживаются")
if not member.is_dir():
total_size += member.file_size
if total_size > MAX_UNPACKED_BYTES:
raise UpdateError("Распакованный размер обновления превышает предел")
if relative_path.as_posix() == executable_name:
if member.is_dir():
raise UpdateError("Исполняемый файл обновления имеет неверный тип")
executable_member = member
if executable_member is None:
raise UpdateError("В архиве обновления не найден исполняемый файл")
with archive.open(executable_member, "r") as executable:
_validate_executable_signature(executable.read(4), executable_name)
return members
def _validate_update_archive(path: Path, executable_name: str) -> None:
try:
with zipfile.ZipFile(path, "r") as archive:
_validated_archive_members(archive, executable_name)
bad_member = archive.testzip()
except (OSError, zipfile.BadZipFile, RuntimeError) as exc:
raise UpdateError(f"Не удалось проверить ZIP-архив обновления: {exc}") from exc
if bad_member is not None:
raise UpdateError(f"Повреждён файл в архиве обновления: {bad_member}")
def _extract_update_archive(
archive_path: Path,
staging_directory: Path,
executable_name: str,
) -> Path:
try:
staging_directory.mkdir()
with zipfile.ZipFile(archive_path, "r") as archive:
members = _validated_archive_members(archive, executable_name)
for member in members:
relative_path = PurePosixPath(member.filename.rstrip("/"))
destination = staging_directory.joinpath(*relative_path.parts)
if member.is_dir():
destination.mkdir(parents=True, exist_ok=True)
continue
destination.parent.mkdir(parents=True, exist_ok=True)
with (
archive.open(member, "r") as source,
destination.open("xb") as output,
):
shutil.copyfileobj(source, output)
executable_path = staging_directory / executable_name
with executable_path.open("rb") as executable:
_validate_executable_signature(executable.read(4), executable_name)
if not executable_name.casefold().endswith(".exe"):
executable_path.chmod(0o755)
return executable_path
except Exception:
shutil.rmtree(staging_directory, ignore_errors=True)
raise
def download_update(
info: UpdateInfo,
target_directory: Path,
*,
timeout: float = DOWNLOAD_TIMEOUT_SECONDS,
opener: Callable[..., BinaryIO] | None = None,
progress: Callable[[int], None] | None = None,
cancelled: Callable[[], bool] | None = None,
) -> Path:
"""Скачать и проверить ZIP-пакет обновления рядом с приложением."""
target_directory = target_directory.resolve()
if not target_directory.is_dir():
raise UpdateError("Каталог приложения не найден")
if not os.access(target_directory, os.W_OK):
raise UpdateError("Нет прав на запись в каталог приложения")
temporary_path = target_directory / (
f".{info.asset_name}.download-{uuid.uuid4().hex}"
)
request = Request(
info.download_url,
headers={
"Accept": "application/octet-stream",
"User-Agent": f"PlateConfigAppUpdater/{info.version}",
},
)
open_request = opener or urlopen
received = 0
try:
with open_request(request, timeout=timeout) as response:
final_url = urlparse(response.geturl())
if final_url.scheme != "https" or final_url.hostname != UPDATE_HOST:
raise UpdateError("Загрузка перенаправлена на неизвестный сервер")
if final_url.path.rstrip("/") == "/user/login":
raise UpdateError("Release-артефакты недоступны без авторизации")
content_length = response.headers.get("Content-Length")
total = int(content_length) if content_length else 0
if total > MAX_DOWNLOAD_BYTES:
raise UpdateError("Размер обновления превышает допустимый предел")
with temporary_path.open("wb") as output:
while True:
if cancelled is not None and cancelled():
raise UpdateCancelled("Загрузка обновления отменена")
chunk = response.read(DOWNLOAD_CHUNK_SIZE)
if not chunk:
break
received += len(chunk)
if received > MAX_DOWNLOAD_BYTES:
raise UpdateError("Размер обновления превышает допустимый предел")
output.write(chunk)
if progress is not None and total > 0:
progress(min(received * 100 // total, 100))
output.flush()
os.fsync(output.fileno())
if received == 0:
raise UpdateError("Сервер вернул пустой файл обновления")
if total > 0 and received != total:
raise UpdateError("Файл обновления скачан не полностью")
executable_name = archive_executable_name(info.asset_name)
_validate_update_archive(temporary_path, executable_name)
if progress is not None:
progress(100)
return temporary_path
except Exception:
temporary_path.unlink(missing_ok=True)
raise
def current_executable() -> Path:
if not getattr(sys, "frozen", False):
raise UpdateError("Самообновление доступно только в собранной версии")
return Path(sys.executable).resolve()
def signal_update_ready() -> None:
ready_file = os.environ.pop(UPDATE_READY_ENV, None)
if ready_file:
try:
Path(ready_file).write_text(str(os.getpid()), encoding="ascii")
except OSError:
pass
def _write_windows_updater(script_path: Path) -> None:
script = r'''param(
[int]$ApplicationProcessId,
[string]$Staging,
[string]$Target,
[string]$Backup,
[string]$ReadyFile
)
$ErrorActionPreference = "Stop"
$applicationDirectory = Split-Path -Parent $Target
$errorLog = Join-Path $applicationDirectory "update-error.log"
$installed = [System.Collections.Generic.List[string]]::new()
$backedUp = [System.Collections.Generic.List[string]]::new()
$restarted = $null
try {
Wait-Process -Id $ApplicationProcessId -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
New-Item -ItemType Directory -Path $Backup | Out-Null
$entries = @(Get-ChildItem -LiteralPath $Staging -Force)
foreach ($entry in $entries) {
$destination = Join-Path $applicationDirectory $entry.Name
if (Test-Path -LiteralPath $destination) {
$deadline = (Get-Date).AddSeconds(30)
while ($true) {
try {
Move-Item -LiteralPath $destination -Destination $Backup -Force -ErrorAction Stop
break
} catch {
if ((Get-Date) -ge $deadline) { throw }
Start-Sleep -Milliseconds 250
}
}
$backedUp.Add($entry.Name)
}
Move-Item -LiteralPath $entry.FullName -Destination $destination -Force -ErrorAction Stop
$installed.Add($entry.Name)
}
if (-not (Test-Path -LiteralPath $Target -PathType Leaf)) {
throw "В пакете обновления отсутствует исполняемый файл"
}
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
$previousReadyFile = $env:PLATE_CONFIG_UPDATE_READY_FILE
$previousResetEnvironment = $env:PYINSTALLER_RESET_ENVIRONMENT
try {
$env:PLATE_CONFIG_UPDATE_READY_FILE = $ReadyFile
$env:PYINSTALLER_RESET_ENVIRONMENT = "1"
$restarted = Start-Process -FilePath $Target -WorkingDirectory $applicationDirectory -PassThru
} finally {
$env:PLATE_CONFIG_UPDATE_READY_FILE = $previousReadyFile
$env:PYINSTALLER_RESET_ENVIRONMENT = $previousResetEnvironment
}
$deadline = (Get-Date).AddSeconds(60)
while (-not (Test-Path -LiteralPath $ReadyFile -PathType Leaf)) {
if ($restarted.HasExited) { throw "Новая версия завершилась до запуска интерфейса" }
if ((Get-Date) -ge $deadline) { throw "Новая версия не подтвердила запуск" }
Start-Sleep -Milliseconds 250
}
if ($restarted.HasExited) { throw "Новая версия завершилась сразу после запуска" }
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $Staging -Recurse -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force
} catch {
$failure = $_
if ($restarted -and -not $restarted.HasExited) {
& taskkill.exe /PID $restarted.Id /T /F *> $null
Start-Sleep -Milliseconds 500
}
$rollbackErrors = [System.Collections.Generic.List[string]]::new()
foreach ($name in $installed) {
$destination = Join-Path $applicationDirectory $name
if (Test-Path -LiteralPath $destination) {
try {
Remove-Item -LiteralPath $destination -Recurse -Force -ErrorAction Stop
} catch {
$rollbackErrors.Add($_.ToString())
}
}
}
foreach ($name in $backedUp) {
$backupItem = Join-Path $Backup $name
$destination = Join-Path $applicationDirectory $name
if (Test-Path -LiteralPath $backupItem) {
try {
Move-Item -LiteralPath $backupItem -Destination $destination -Force -ErrorAction Stop
} catch {
$rollbackErrors.Add($_.ToString())
}
}
}
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $Staging -Recurse -Force -ErrorAction SilentlyContinue
if ($rollbackErrors.Count -eq 0) {
Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
}
@($failure.ToString()) + @($rollbackErrors) | Out-File -LiteralPath $errorLog -Encoding utf8
Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force -ErrorAction SilentlyContinue
exit 1
}
'''
script_path.write_text(script, encoding="utf-8-sig", newline="\r\n")
def _write_linux_updater(script_path: Path) -> None:
script = r'''#!/bin/sh
set -eu
application_pid="$1"
staging_path="$2"
target_path="$3"
backup_path="$4"
ready_path="$5"
application_dir="$(dirname -- "$target_path")"
error_log="$application_dir/update-error.log"
installed_file="$backup_path/.installed"
backed_up_file="$backup_path/.backed_up"
new_pid=""
rollback() {
exit_code="$?"
trap - EXIT HUP INT TERM
set +e
if [ -n "$new_pid" ]; then
kill "$new_pid" 2>/dev/null
wait "$new_pid" 2>/dev/null
fi
rollback_failed=0
if [ -f "$installed_file" ]; then
while IFS= read -r name; do
rm -rf -- "$application_dir/$name" || rollback_failed=1
done < "$installed_file"
fi
if [ -f "$backed_up_file" ]; then
while IFS= read -r name; do
if [ -e "$backup_path/$name" ]; then
mv -f -- "$backup_path/$name" "$application_dir/$name" || rollback_failed=1
fi
done < "$backed_up_file"
fi
rm -f -- "$ready_path"
rm -rf -- "$staging_path"
if [ "$rollback_failed" -eq 0 ]; then
rm -rf -- "$backup_path"
fi
printf '%s\n' "Не удалось заменить приложение" >"$error_log"
rm -f -- "$0"
if [ "$exit_code" -eq 0 ]; then
exit_code=1
fi
exit "$exit_code"
}
trap rollback EXIT HUP INT TERM
while kill -0 "$application_pid" 2>/dev/null; do
sleep 0.2
done
rm -rf -- "$backup_path"
mkdir -p -- "$backup_path"
: >"$installed_file"
: >"$backed_up_file"
for source_path in "$staging_path"/*; do
name="$(basename -- "$source_path")"
destination="$application_dir/$name"
if [ -e "$destination" ]; then
mv -f -- "$destination" "$backup_path/$name"
printf '%s\n' "$name" >>"$backed_up_file"
fi
mv -f -- "$source_path" "$destination"
printf '%s\n' "$name" >>"$installed_file"
done
test -f "$target_path"
chmod 755 -- "$target_path"
rm -f -- "$ready_path"
PLATE_CONFIG_UPDATE_READY_FILE="$ready_path" PYINSTALLER_RESET_ENVIRONMENT=1 "$target_path" >/dev/null 2>&1 &
new_pid=$!
attempts=0
while [ ! -f "$ready_path" ]; do
if ! kill -0 "$new_pid" 2>/dev/null; then
exit 1
fi
attempts=$((attempts + 1))
if [ "$attempts" -ge 240 ]; then
exit 1
fi
sleep 0.25
done
if ! kill -0 "$new_pid" 2>/dev/null; then
exit 1
fi
trap - EXIT HUP INT TERM
rm -f -- "$ready_path"
rm -rf -- "$backup_path" "$staging_path"
rm -f -- "$0"
'''
script_path.write_text(script, encoding="utf-8", newline="\n")
script_path.chmod(0o700)
def launch_update(
downloaded_path: Path,
*,
executable_path: Path | None = None,
process_id: int | None = None,
system: str | None = None,
) -> None:
"""Распаковать обновление и запустить helper для замены пакета."""
archive_path = downloaded_path.resolve()
target = (executable_path or current_executable()).resolve()
if not archive_path.is_file():
raise UpdateError("Скачанный архив обновления не найден")
if not os.access(target.parent, os.W_OK):
raise UpdateError("Нет прав на замену приложения")
updater_id = uuid.uuid4().hex
application_pid = process_id or os.getpid()
system_name = (system or platform.system()).casefold()
if system_name not in {"windows", "linux"}:
raise UpdateError(f"Автообновление не поддерживает платформу {system_name}")
staging_path = target.parent / f".plate_config_update-{updater_id}"
backup_path = target.parent / f".plate_config_backup-{updater_id}"
ready_path = target.parent / f".plate_config_ready-{updater_id}"
script_path: Path | None = None
try:
_extract_update_archive(archive_path, staging_path, target.name)
archive_path.unlink()
except Exception:
shutil.rmtree(staging_path, ignore_errors=True)
raise
if system_name == "windows":
script_path = target.parent / f".plate_config_updater-{updater_id}.ps1"
try:
_write_windows_updater(script_path)
creation_flags = (
getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
| getattr(subprocess, "CREATE_NO_WINDOW", 0)
)
subprocess.Popen(
[
"powershell.exe",
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-File",
str(script_path),
str(application_pid),
str(staging_path),
str(target),
str(backup_path),
str(ready_path),
],
close_fds=True,
creationflags=creation_flags,
)
except Exception:
script_path.unlink(missing_ok=True)
shutil.rmtree(staging_path, ignore_errors=True)
raise
return
if system_name == "linux":
script_path = target.parent / f".plate_config_updater-{updater_id}.sh"
try:
_write_linux_updater(script_path)
subprocess.Popen(
[
"/bin/sh",
str(script_path),
str(application_pid),
str(staging_path),
str(target),
str(backup_path),
str(ready_path),
],
cwd=str(target.parent),
close_fds=True,
start_new_session=True,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except Exception:
script_path.unlink(missing_ok=True)
shutil.rmtree(staging_path, ignore_errors=True)
raise
return

View File

@ -26,7 +26,13 @@ from ..services.com_service import (
list_available_ports,
params_from_ui,
)
from ..services.auth_service import AuthResult, AuthStatus, check_status, perform_handshake
from ..services.auth_service import (
AuthResult,
AuthStatus,
check_status,
perform_required_handshake,
perform_startup_handshake,
)
from ..services import settings_service
from ..ui.icons import get_icon_color, get_icon_muted, icon
from ..serial_config import SerialParams
@ -115,6 +121,7 @@ class ConnectionTab(QWidget):
self._health_running = False # идёт ли сейчас проверка
self._health_fail_count = 0 # подряд неудачных проверок
self._payment_observers_bound = False
self._startup_ready = False
self._setup_ui()
self._load_ports()
@ -362,6 +369,8 @@ class ConnectionTab(QWidget):
def _start_basic_polling(self) -> None:
"""Запустить подписку на входы и подключить PaymentSection к reader-событиям."""
if self._plate_state != PlateState.READY:
return
self._input_section.start_polling()
if not self._payment_observers_bound:
self._connection.input_received.connect(
@ -393,7 +402,7 @@ class ConnectionTab(QWidget):
for group in self._basic_groups:
group.setVisible(not checked)
# Управление poll-потоком InputTestSection
if not checked:
if not checked and self._plate_state == PlateState.READY:
self._start_basic_polling()
else:
self._input_section.stop_polling()
@ -433,8 +442,14 @@ class ConnectionTab(QWidget):
return
params = self._collect_params()
self._startup_ready = False
try:
self._connection.open(port=port, params=params)
self._connection.open(
port=port,
params=params,
drain_paused=True,
reset_device=True,
)
except Exception as exc:
QMessageBox.critical(self, "Ошибка подключения", f"Не удалось открыть {port}\n{exc}")
return
@ -448,21 +463,14 @@ class ConnectionTab(QWidget):
f"{PARITY_SHORT_MAP[self.parity_combo.currentText()]}-"
f"{self.stop_bits_combo.currentText()})"
)
self._set_status(self._conn_info, PlateState.NEED_AUTH)
self._update_plate_state(PlateState.NEED_AUTH)
tab_index = self._tab_index_getter()
if tab_index >= 0:
self._set_tab_title(tab_index, port)
# Проверяем авторизацию один раз при подключении
self._check_auth_on_connect()
# Запускаем периодический health-check (если включён в настройках)
self._apply_health_settings()
# Если PRO выключен — запускаем опрос входов в inline-секции
if not self._pro_toggle.isChecked():
self._start_basic_polling()
# До стартового banner и READY! никакие другие команды не отправляются.
self._start_startup_handshake()
def _close_this_tab(self) -> None:
index = self._tab_index_getter()
@ -485,12 +493,14 @@ class ConnectionTab(QWidget):
self._button_backlight_section.shutdown()
self._firmware_section.cancel_upload()
self._connection.close()
self._startup_ready = False
# Дождаться завершения всех фоновых потоков
for thread in self._active_threads:
if thread.isRunning():
thread.quit()
thread.wait(1000)
self._active_threads.clear()
self._set_board_actions_enabled(False)
self._update_plate_state(PlateState.DISCONNECTED)
self._reconnect_btn.hide()
@ -501,22 +511,26 @@ class ConnectionTab(QWidget):
# Остановить опрос / health
self._stop_health_timer()
self._input_section.stop_polling()
self._set_board_actions_enabled(False)
port = self._connection.serial.port if self._connection.serial else None
if port is None:
return
params = self._collect_params()
self._connection.close()
self._startup_ready = False
self._update_plate_state(PlateState.DISCONNECTED)
try:
self._connection.open(port=port, params=params)
self._connection.open(
port=port,
params=params,
drain_paused=True,
reset_device=True,
)
except Exception as exc:
QMessageBox.critical(self, "Переподключение", f"Не удалось открыть {port}\n{exc}")
return
self._set_status(self._conn_info, PlateState.NEED_AUTH)
self._check_auth_on_connect()
self._apply_health_settings()
if not self._pro_toggle.isChecked():
self._start_basic_polling()
self._update_plate_state(PlateState.NEED_AUTH)
self._start_startup_handshake()
# ── Открытие диалогов ───────────────────────────────────────────────
@ -616,11 +630,36 @@ class ConnectionTab(QWidget):
def _open_terminal(self) -> None:
from .dialogs.command_terminal_dialog import CommandTerminalDialog
def create_terminal() -> CommandTerminalDialog:
dialog = CommandTerminalDialog(self._connection, parent=self)
dialog.authorization_finished.connect(self._on_terminal_authorization)
return dialog
self._exec_dialog(
"terminal",
lambda: CommandTerminalDialog(self._connection, parent=self),
create_terminal,
)
def _on_terminal_authorization(self, result: AuthResult) -> None:
if result.status == AuthStatus.OK:
self._update_plate_state(PlateState.READY)
self._set_board_actions_enabled(True)
self._health_fail_count = 0
self._apply_health_settings()
if not self._pro_toggle.isChecked():
self._start_basic_polling()
return
state = (
PlateState.NEED_AUTH
if self._connection.is_connected
else PlateState.DISCONNECTED
)
self._input_section.stop_polling()
self._set_board_actions_enabled(False)
self._update_plate_state(state)
def _not_implemented(self) -> None:
QMessageBox.information(self, "В разработке", "Эта функция пока не реализована.")
@ -630,6 +669,8 @@ class ConnectionTab(QWidget):
PlateState.NEED_AUTH: "Требуется авторизация",
PlateState.READY: "Готово",
}
if state == PlateState.READY and self._connection.auth_required is False:
_STATE_LABELS[PlateState.READY] = "Готово без авторизации"
_STATE_CSS = {
PlateState.DISCONNECTED: "disconnected",
PlateState.NEED_AUTH: "need-auth",
@ -674,54 +715,79 @@ class ConnectionTab(QWidget):
else:
self.auth_btn.hide()
def _check_auth_on_connect(self) -> None:
"""Проверить статус авторизации сразу после подключения (в фоновом потоке)."""
self._run_status_check(timeout=3)
def _run_status_check(self, timeout: float = 0.5) -> None:
"""Запуск проверки статуса в фоновом потоке."""
thread = _StatusThread(self._connection, timeout, parent=self)
thread.result_ready.connect(self._on_status_checked)
thread.finished.connect(lambda: self._cleanup_thread(thread))
self._active_threads.append(thread)
thread.start()
def _cleanup_thread(self, thread: QThread) -> None:
"""Убрать завершённый поток из списка."""
if thread in self._active_threads:
self._active_threads.remove(thread)
thread.deleteLater()
def _on_status_checked(self, status: AuthStatus) -> None:
if status == AuthStatus.OK:
self._update_plate_state(PlateState.READY)
elif status == AuthStatus.NEED_AUTH:
self._update_plate_state(PlateState.NEED_AUTH)
else:
self._update_plate_state(PlateState.DISCONNECTED)
def _start_startup_handshake(self) -> None:
"""Дождаться загрузочной строки платы и автоматически авторизоваться."""
self._run_handshake(wait_for_startup=True)
def _set_board_actions_enabled(self, enabled: bool) -> None:
"""Запретить команды UI до подтверждённой строки READY!."""
for group in self._basic_groups:
group.setEnabled(enabled)
for button, _ in self._pro_buttons:
button.setEnabled(enabled)
self._command_input_btn.setEnabled(enabled)
def _do_handshake(self) -> None:
"""Выполнить авторизацию (двойное рукопожатие AES-256) в фоновом потоке."""
self._run_handshake(wait_for_startup=not self._startup_ready)
def _run_handshake(self, *, wait_for_startup: bool) -> None:
if not self._connection.is_connected:
QMessageBox.warning(self, "Авторизация", "Порт не подключён.")
return
self._stop_health_timer()
self._input_section.stop_polling()
self._connection.pause_drain()
self._set_board_actions_enabled(False)
self.auth_btn.setEnabled(False)
self.auth_btn.setText("Авторизация…")
if wait_for_startup:
self.auth_btn.setText("Ожидание платы…")
else:
self.auth_btn.setText("Авторизация…")
thread = _HandshakeThread(self._connection, parent=self)
thread.result_ready.connect(self._on_handshake_done)
thread = _HandshakeThread(
self._connection,
wait_for_startup=wait_for_startup,
parent=self,
)
thread.result_ready.connect(
lambda result, automatic=wait_for_startup: self._on_handshake_done(
result,
automatic=automatic,
)
)
thread.finished.connect(lambda: self._cleanup_thread(thread))
self._active_threads.append(thread)
thread.start()
def _on_handshake_done(self, result: AuthResult) -> None:
def _on_handshake_done(self, result: AuthResult, *, automatic: bool) -> None:
self._connection.resume_drain()
if automatic:
self._startup_ready = result.startup_ready
self.auth_btn.setEnabled(True)
self.auth_btn.setText("Handshake (Авторизация)")
if result.status == AuthStatus.OK:
self._update_plate_state(PlateState.READY)
QMessageBox.information(self, "Авторизация", f"Успешно!\n{result.detail}")
self._set_board_actions_enabled(True)
self._health_fail_count = 0
self._apply_health_settings()
if not self._pro_toggle.isChecked():
self._start_basic_polling()
if not automatic:
QMessageBox.information(self, "Авторизация", f"Успешно!\n{result.detail}")
else:
self._update_plate_state(PlateState.DISCONNECTED)
state = (
PlateState.NEED_AUTH
if self._connection.is_connected
else PlateState.DISCONNECTED
)
self._update_plate_state(state)
QMessageBox.warning(
self,
"Авторизация",
@ -794,13 +860,18 @@ class ConnectionTab(QWidget):
def _on_health_result(self, status: AuthStatus) -> None:
self._health_running = False
if status == AuthStatus.OK:
# Связь жива, плата авторизована (или не требует авторизацию)
# Только READY! подтверждает готовность платы.
self._health_fail_count = 0
if self._plate_state != PlateState.READY:
self._update_plate_state(PlateState.READY)
self._set_board_actions_enabled(True)
if not self._pro_toggle.isChecked():
self._start_basic_polling()
elif status == AuthStatus.NEED_AUTH:
# Связь жива, но нужна авторизация (полный таймаут на 99)
# Связь жива, но READY! не получен.
self._health_fail_count = 0
self._input_section.stop_polling()
self._set_board_actions_enabled(False)
if self._plate_state != PlateState.NEED_AUTH:
self._update_plate_state(PlateState.NEED_AUTH)
else:
@ -808,6 +879,8 @@ class ConnectionTab(QWidget):
self._health_fail_count += 1
if self._health_fail_count >= self._MAX_HEALTH_FAILURES:
self._health_fail_count = 0
self._set_board_actions_enabled(False)
self._startup_ready = False
self._update_plate_state(PlateState.DISCONNECTED)
self._connection.close()
@ -826,25 +899,12 @@ class ConnectionTab(QWidget):
self._health_running = False
if success:
self._health_fail_count = 0
# Порт снова доступен — проверяем авторизацию
self._check_auth_on_connect()
self._startup_ready = False
self._update_plate_state(PlateState.NEED_AUTH)
self._start_startup_handshake()
# Если не удалось — ничего не делаем, следующий тик попробует снова
class _StatusThread(QThread):
"""Фоновый поток для проверки статуса платы."""
result_ready = Signal(AuthStatus)
def __init__(self, connection: ComConnection, timeout: float, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._conn = connection
self._timeout = timeout
def run(self) -> None: # noqa: D401
status = check_status(self._conn, timeout=self._timeout)
self.result_ready.emit(status)
class _HealthCheckThread(QThread):
"""Быстрая проверка связи с платой (команда 99, таймаут 2 с)."""
result_ready = Signal(AuthStatus)
@ -871,21 +931,30 @@ class _ReconnectThread(QThread):
def run(self) -> None: # noqa: D401
try:
self._conn.reconnect()
self._conn.reconnect(drain_paused=True, reset_device=True)
self.result_ready.emit(True)
except Exception:
self.result_ready.emit(False)
class _HandshakeThread(QThread):
"""Фоновый поток для авторизации."""
"""Фоновый поток стартовой последовательности или повторной авторизации."""
result_ready = Signal(object)
def __init__(self, connection: ComConnection, parent: QWidget | None = None) -> None:
def __init__(
self,
connection: ComConnection,
*,
wait_for_startup: bool,
parent: QWidget | None = None,
) -> None:
super().__init__(parent)
self._conn = connection
self._wait_for_startup = wait_for_startup
def run(self) -> None: # noqa: D401
from ..services.auth_service import perform_handshake as _handshake
result = _handshake(self._conn, timeout=2.0)
if self._wait_for_startup:
result = perform_startup_handshake(self._conn)
else:
result = perform_required_handshake(self._conn, timeout=2.0)
self.result_ready.emit(result)

View File

@ -5,7 +5,7 @@
from __future__ import annotations
from PySide6.QtCore import Qt
from PySide6.QtCore import QThread, Qt, Signal
from PySide6.QtGui import QKeyEvent
from PySide6.QtWidgets import (
QHBoxLayout,
@ -13,17 +13,28 @@ from PySide6.QtWidgets import (
QPushButton,
QTextEdit,
QVBoxLayout,
QSplitter,
QWidget,
)
from ...services.com_service import ComConnection
from ...ui.icons import get_icon_muted, icon
from ...services.auth_service import (
AuthResult,
AuthStatus,
perform_required_handshake,
)
from ...ui.icons import get_icon_color, get_icon_muted, icon
from .styled_dialog import StyledDialog
class CommandTerminalDialog(StyledDialog):
authorization_finished = Signal(object)
def __init__(self, connection: ComConnection, parent=None) -> None:
super().__init__(parent)
self._conn = connection
self._auth_thread: _TerminalHandshakeThread | None = None
self._command_controls: list[QWidget] = []
self.setWindowTitle("Ввод команд")
self.setMinimumSize(540, 400)
@ -48,13 +59,22 @@ class CommandTerminalDialog(StyledDialog):
root = self.containerLayout()
root.setSpacing(8)
# Вертикальный сплиттер для изменения высоты лога
splitter = QSplitter(Qt.Orientation.Vertical)
# ── Лог ──
self._log = QTextEdit()
self._log.setReadOnly(True)
self._log.setStyleSheet(
"QTextEdit { background: #1e293b; color: #e2e8f0; font-family: 'Consolas', 'Courier New', monospace; font-size: 12px; border-radius: 8px; padding: 8px; }"
)
root.addWidget(self._log, 1)
splitter.addWidget(self._log)
# ── Нижняя панель управления ──
bottom_widget = QWidget()
bottom_layout = QVBoxLayout(bottom_widget)
bottom_layout.setContentsMargins(0, 0, 0, 0)
bottom_layout.setSpacing(8)
# ── Ввод ──
input_layout = QHBoxLayout()
@ -63,21 +83,27 @@ class CommandTerminalDialog(StyledDialog):
self._input.returnPressed.connect(self._send_command)
input_layout.addWidget(self._input, 1)
send_btn = QPushButton("Отправить")
send_btn.setProperty("role", "primary")
send_btn.setIcon(icon("fa5s.paper-plane", color="#ecfdf5"))
send_btn.clicked.connect(self._send_command)
input_layout.addWidget(send_btn)
self._send_btn = QPushButton("Отправить")
self._send_btn.setProperty("role", "primary")
self._send_btn.setIcon(icon("fa5s.paper-plane", color="#ecfdf5"))
self._send_btn.clicked.connect(self._send_command)
input_layout.addWidget(self._send_btn)
clear_btn = QPushButton("Очистить")
clear_btn.setIcon(icon("fa5s.eraser", color=get_icon_muted()))
clear_btn.clicked.connect(self._log.clear)
input_layout.addWidget(clear_btn)
root.addLayout(input_layout)
bottom_layout.addLayout(input_layout)
# ── Быстрые команды ──
quick_layout = QHBoxLayout()
self._auth_btn = QPushButton("Авторизация")
self._auth_btn.setMinimumHeight(30)
self._auth_btn.setIcon(icon("fa5s.key", color=get_icon_color()))
self._auth_btn.clicked.connect(self._start_authorization)
quick_layout.addWidget(self._auth_btn)
quick_commands = [
("Статус (99)", "99"),
("Версия (05)", "05"),
@ -91,8 +117,9 @@ class CommandTerminalDialog(StyledDialog):
btn.setStyleSheet("padding: 4px 10px;")
btn.clicked.connect(lambda checked=False, c=cmd: self._send(c))
quick_layout.addWidget(btn)
self._command_controls.append(btn)
quick_layout.addStretch()
root.addLayout(quick_layout)
bottom_layout.addLayout(quick_layout)
# ── Закрыть ──
btn_layout = QHBoxLayout()
@ -101,7 +128,63 @@ class CommandTerminalDialog(StyledDialog):
close_btn.setIcon(icon("fa5s.times", color=get_icon_muted()))
close_btn.clicked.connect(self.close)
btn_layout.addWidget(close_btn)
root.addLayout(btn_layout)
bottom_layout.addLayout(btn_layout)
splitter.addWidget(bottom_widget)
# Начальное распределение высоты (75% на лог, 25% на нижнюю панель)
splitter.setStretchFactor(0, 75)
splitter.setStretchFactor(1, 25)
root.addWidget(splitter, 1)
self._command_controls.extend((self._input, self._send_btn, self._auth_btn))
def _start_authorization(self) -> None:
if not self._conn.is_connected:
self._log.append('<span style="color:#f87171;">✗ Порт не подключён</span>')
return
if self._auth_thread is not None:
return
self._set_command_controls_enabled(False)
self._auth_btn.setText("Авторизация…")
self._log.append('<span style="color:#facc15;">Проверка авторизации…</span>')
self._conn.pause_drain()
thread = _TerminalHandshakeThread(self._conn, parent=self)
thread.result_ready.connect(self._on_authorization_done)
thread.finished.connect(self._cleanup_auth_thread)
self._auth_thread = thread
thread.start()
def _on_authorization_done(self, result: AuthResult) -> None:
self._conn.resume_drain()
self._set_command_controls_enabled(True)
self._auth_btn.setText("Авторизация")
if result.status == AuthStatus.OK:
if result.auth_skipped:
self._log.append(
'<span style="color:#4ade80;">✓ Прошивка no_auth: '
'авторизация не требуется</span>'
)
else:
self._log.append(
'<span style="color:#4ade80;">✓ Авторизация успешно завершена</span>'
)
else:
self._log.append(
f'<span style="color:#f87171;">✗ Авторизация: {result.detail}</span>'
)
self.authorization_finished.emit(result)
def _cleanup_auth_thread(self) -> None:
if self._auth_thread is not None:
self._auth_thread.deleteLater()
self._auth_thread = None
def _set_command_controls_enabled(self, enabled: bool) -> None:
for control in self._command_controls:
control.setEnabled(enabled)
def _send_command(self) -> None:
text = self._input.text().strip()
@ -167,8 +250,29 @@ class CommandTerminalDialog(StyledDialog):
self._log.append(f'<span style="color:#93c5fd;">← {stripped}</span>')
def closeEvent(self, event) -> None:
if self._auth_thread is not None and self._auth_thread.isRunning():
self._log.append(
'<span style="color:#facc15;">Дождитесь завершения авторизации.</span>'
)
event.ignore()
return
try:
self._conn.line_received.disconnect(self._on_line_received)
except (RuntimeError, TypeError):
pass
super().closeEvent(event)
class _TerminalHandshakeThread(QThread):
result_ready = Signal(object)
def __init__(self, connection: ComConnection, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._conn = connection
def run(self) -> None:
try:
result = perform_required_handshake(self._conn, timeout=2.0)
except Exception as exc:
result = AuthResult(AuthStatus.FAILED, str(exc))
self.result_ready.emit(result)

View File

@ -31,9 +31,12 @@ from PySide6.QtWidgets import (
QPushButton,
QTextEdit,
QVBoxLayout,
QSplitter,
QWidget,
)
from ...services.com_service import ComConnection
from ...services.firmware_service import perform_authenticated_ota
from ...ui.icons import get_icon_color, get_icon_muted, icon
from .styled_dialog import StyledDialog
@ -55,6 +58,15 @@ class FirmwareDialog(StyledDialog):
root = self.containerLayout()
root.setSpacing(12)
# Вертикальный сплиттер для изменения высоты лога
splitter = QSplitter(Qt.Orientation.Vertical)
# ── Верхняя панель (информация + загрузка) ──
top_widget = QWidget()
top_layout = QVBoxLayout(top_widget)
top_layout.setContentsMargins(0, 0, 0, 0)
top_layout.setSpacing(12)
# ── Информация о плате ──
info_group = QGroupBox("Информация о плате")
info_layout = QVBoxLayout(info_group)
@ -82,7 +94,7 @@ class FirmwareDialog(StyledDialog):
self._info_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
info_layout.addWidget(self._info_label)
root.addWidget(info_group)
top_layout.addWidget(info_group)
# ── Загрузка файла ──
upload_group = QGroupBox("Загрузка прошивки")
@ -117,13 +129,19 @@ class FirmwareDialog(StyledDialog):
upload_btn_layout.addWidget(self._flash_btn)
upload_layout.addLayout(upload_btn_layout)
root.addWidget(upload_group)
top_layout.addWidget(upload_group)
splitter.addWidget(top_widget)
# ── Лог ──
# ── Лог (нижняя часть сплиттера) ──
self._log = QTextEdit()
self._log.setReadOnly(True)
self._log.setMaximumHeight(200)
root.addWidget(self._log)
splitter.addWidget(self._log)
# Начальное распределение размеров (65% на верхнюю часть, 35% на лог)
splitter.setStretchFactor(0, 65)
splitter.setStretchFactor(1, 35)
root.addWidget(splitter, 1)
# ── Закрыть ──
btn_layout = QHBoxLayout()
@ -220,10 +238,27 @@ class FirmwareDialog(StyledDialog):
self._log.append("✗ Загрузка не удалась.")
def _flash_ota(self) -> None:
resp = self._query("04")
self._log.append(f"→ 04\n← {resp or 'таймаут'}")
if resp:
QMessageBox.information(self, "OTA", f"Ответ контроллера:\n{resp}")
if not self._conn.is_connected:
self._log.append("✗ Порт не подключён.")
return
self._flash_btn.setEnabled(False)
self._upload_btn.setEnabled(False)
self._log.append("Запуск OTA (04)…")
from PySide6.QtCore import QCoreApplication
def append_log(message: str) -> None:
self._log.append(message)
QCoreApplication.processEvents()
result = perform_authenticated_ota(self._conn, log_message=append_log)
if result.success:
self._log.append(f"✓ {result.detail}.")
else:
self._log.append(f"✗ {result.detail}")
self._flash_btn.setEnabled(True)
self._upload_btn.setEnabled(True)
def closeEvent(self, event) -> None:
if self._upload_thread is not None and self._upload_thread.isRunning():

View File

@ -7,7 +7,7 @@ from __future__ import annotations
import os
from PySide6.QtCore import QThread, Signal
from PySide6.QtCore import QThread, Signal, Qt
from PySide6.QtWidgets import (
QFileDialog,
QGroupBox,
@ -19,9 +19,11 @@ from PySide6.QtWidgets import (
QTextEdit,
QVBoxLayout,
QWidget,
QSplitter,
)
from ...services.com_service import ComConnection
from ...services.firmware_service import perform_authenticated_ota
from ...ui.icons import get_icon_color, icon
# Переиспользуем воркер из диалога
@ -50,6 +52,9 @@ class FirmwareSection(QWidget):
root.setContentsMargins(0, 0, 0, 0)
root.setSpacing(10)
# Вертикальный сплиттер для изменения высоты лога
splitter = QSplitter(Qt.Orientation.Vertical)
# ── Обновление прошивки ──
fw_group = QGroupBox("Обновление прошивки")
fw_layout = QVBoxLayout(fw_group)
@ -84,13 +89,18 @@ class FirmwareSection(QWidget):
self._update_btn.clicked.connect(self._start_update)
fw_layout.addWidget(self._update_btn)
root.addWidget(fw_group)
splitter.addWidget(fw_group)
# ── Лог ──
self._log = QTextEdit()
self._log.setReadOnly(True)
self._log.setMaximumHeight(160)
root.addWidget(self._log)
splitter.addWidget(self._log)
# Задаем начальное распределение размеров (70% на верхнюю часть, 30% на лог)
splitter.setStretchFactor(0, 70)
splitter.setStretchFactor(1, 30)
root.addWidget(splitter, 1)
# ── Public ────────────────────────────────────────────────────────
@ -186,36 +196,14 @@ class FirmwareSection(QWidget):
self._pick_btn.setEnabled(True)
self.upload_finished.emit()
return
try:
self._conn.send_command("04")
self._log.append("→ 04")
# Контроллер шлёт прогресс 1-100, затем «Successful update»
# и через ~4 сек перезагружается с «READY!».
# Ждём до 60 сек (перепрошивка + перезагрузка).
import time
deadline = time.time() + 60.0
success = False
while time.time() < deadline:
remaining = max(deadline - time.time(), 0.1)
resp = self._conn.read_response(timeout=min(remaining, 2.0))
if resp is None:
continue
self._log.append(f"← {resp}")
if "ready" in resp.lower():
success = True
break
self._conn.release()
if success:
self._log.append("✓ Прошивка обновлена, плата перезагружена.")
else:
self._log.append("← таймаут (нет ответа)")
except Exception as exc:
self._conn.release()
self._log.append(f"Ошибка OTA: {exc}")
result = perform_authenticated_ota(
self._conn,
log_message=self._log.append,
)
if result.success:
self._log.append(f"✓ {result.detail}.")
else:
self._log.append(f"✗ {result.detail}")
self._update_btn.setEnabled(True)
self._pick_btn.setEnabled(True)
self.upload_finished.emit()

View File

@ -1,17 +1,29 @@
from __future__ import annotations
from PySide6.QtCore import Qt
from pathlib import Path
from PySide6.QtCore import QThread, QTimer, Qt, Signal
from PySide6.QtGui import QColor
from PySide6.QtWidgets import (
QFrame,
QGraphicsDropShadowEffect,
QHBoxLayout,
QLabel,
QMessageBox,
QPushButton,
QVBoxLayout,
QWidget,
)
from .. import __version__
from ..services.update_service import (
UpdateCancelled,
UpdateInfo,
check_for_update,
current_executable,
download_update,
launch_update,
)
from ..ui.icons import ICON_COLOR_PRIMARY, get_icon_color, icon
from .dialogs.commands_dialog import CommandsDialog
from .dialogs.settings_dialog import SettingsDialog
@ -21,6 +33,9 @@ class StartTab(QWidget):
def __init__(self, on_new_connection) -> None:
super().__init__()
self._open_dialog_windows: dict[str, QWidget] = {}
self._available_update: UpdateInfo | None = None
self._update_check_thread: _UpdateCheckThread | None = None
self._update_download_thread: _UpdateDownloadThread | None = None
layout = QVBoxLayout(self)
layout.setContentsMargins(26, 26, 26, 26)
layout.addStretch(1)
@ -81,17 +96,33 @@ class StartTab(QWidget):
self._commands_btn.clicked.connect(self._open_commands)
bottom_row.addWidget(self._commands_btn)
settings_column = QWidget()
self._settings_column_layout = QVBoxLayout(settings_column)
self._settings_column_layout.setContentsMargins(0, 0, 0, 0)
self._settings_column_layout.setSpacing(8)
self._settings_btn = QPushButton("Настройки приложения")
self._settings_btn.setMinimumHeight(38)
self._settings_btn.setMinimumWidth(180)
self._settings_btn.setIcon(icon("fa5s.cog", color=get_icon_color()))
self._settings_btn.clicked.connect(self._open_settings)
bottom_row.addWidget(self._settings_btn)
self._settings_column_layout.addWidget(self._settings_btn)
self._update_btn = QPushButton("Скачать обновление")
self._update_btn.setProperty("role", "primary")
self._update_btn.setMinimumHeight(38)
self._update_btn.setMinimumWidth(180)
self._update_btn.setIcon(icon("fa5s.download", color=ICON_COLOR_PRIMARY))
self._update_btn.clicked.connect(self._start_update_download)
self._update_btn.hide()
self._settings_column_layout.addWidget(self._update_btn)
bottom_row.addWidget(settings_column, alignment=Qt.AlignmentFlag.AlignTop)
bottom_row.addStretch(1)
layout.addLayout(bottom_row)
layout.addStretch(2)
QTimer.singleShot(500, self._start_update_check)
# ── Public ────────────────────────────────────────────────────
@ -100,6 +131,14 @@ class StartTab(QWidget):
clr = get_icon_color()
self._commands_btn.setIcon(icon("fa5s.book", color=clr))
self._settings_btn.setIcon(icon("fa5s.cog", color=clr))
self._update_btn.setIcon(icon("fa5s.download", color=ICON_COLOR_PRIMARY))
def shutdown(self) -> None:
"""Остановить фоновые задачи перед закрытием приложения."""
for thread in (self._update_check_thread, self._update_download_thread):
if thread is not None and thread.isRunning():
thread.requestInterruption()
thread.wait(35_000)
# ── Handlers ──────────────────────────────────────────────────
@ -132,3 +171,121 @@ class StartTab(QWidget):
def _open_settings(self) -> None:
self._open_single_dialog("settings", lambda: SettingsDialog(self))
# ── Обновление приложения ─────────────────────────────────────
def _start_update_check(self) -> None:
if self._update_check_thread is not None:
return
thread = _UpdateCheckThread(parent=self)
thread.result_ready.connect(self._on_update_check_done)
thread.finished.connect(self._cleanup_update_check)
self._update_check_thread = thread
thread.start()
def _on_update_check_done(self, info: UpdateInfo | None) -> None:
self._available_update = info
if info is None:
self._update_btn.hide()
return
self._update_btn.setText("Скачать обновление")
self._update_btn.setToolTip(f"Доступна версия {info.tag}")
self._update_btn.setEnabled(True)
self._update_btn.show()
def _cleanup_update_check(self) -> None:
if self._update_check_thread is not None:
self._update_check_thread.deleteLater()
self._update_check_thread = None
def _start_update_download(self) -> None:
info = self._available_update
if info is None or self._update_download_thread is not None:
return
try:
target_directory = current_executable().parent
except Exception as exc:
QMessageBox.information(self, "Обновление приложения", str(exc))
return
self._update_btn.setEnabled(False)
self._update_btn.setText("Скачивание 0%")
thread = _UpdateDownloadThread(info, target_directory, parent=self)
thread.progress_changed.connect(self._on_update_progress)
thread.result_ready.connect(self._on_update_downloaded)
thread.failed.connect(self._on_update_failed)
thread.finished.connect(self._cleanup_update_download)
self._update_download_thread = thread
thread.start()
def _on_update_progress(self, percent: int) -> None:
self._update_btn.setText(f"Скачивание {percent}%")
def _on_update_downloaded(self, path: Path) -> None:
self._update_btn.setText("Перезапуск…")
try:
launch_update(path)
except Exception as exc:
path.unlink(missing_ok=True)
self._on_update_failed(str(exc))
return
self.window().close()
def _on_update_failed(self, detail: str) -> None:
self._update_btn.setText("Скачать обновление")
self._update_btn.setEnabled(True)
QMessageBox.warning(
self,
"Обновление приложения",
f"Не удалось установить обновление:\n{detail}",
)
def _cleanup_update_download(self) -> None:
if self._update_download_thread is not None:
self._update_download_thread.deleteLater()
self._update_download_thread = None
class _UpdateCheckThread(QThread):
result_ready = Signal(object)
def run(self) -> None:
try:
info = check_for_update(__version__)
except Exception:
info = None
if not self.isInterruptionRequested():
self.result_ready.emit(info)
class _UpdateDownloadThread(QThread):
progress_changed = Signal(int)
result_ready = Signal(object)
failed = Signal(str)
def __init__(
self,
info: UpdateInfo,
target_directory: Path,
parent: QWidget | None = None,
) -> None:
super().__init__(parent)
self._info = info
self._target_directory = target_directory
def run(self) -> None:
try:
path = download_update(
self._info,
self._target_directory,
progress=self.progress_changed.emit,
cancelled=self.isInterruptionRequested,
)
except UpdateCancelled:
return
except Exception as exc:
if not self.isInterruptionRequested():
self.failed.emit(str(exc))
return
if not self.isInterruptionRequested():
self.result_ready.emit(path)

View File

@ -19,11 +19,14 @@ function Clean-Build {
}
}
# __pycache__ в проекте (без .venv)
Get-ChildItem -Path $projectRoot -Directory -Filter "__pycache__" -Recurse |
Get-ChildItem -Path $projectRoot -Directory -Filter "__pycache__" -Recurse -ErrorAction SilentlyContinue |
Where-Object { $_.FullName -notlike "*\.venv\*" } |
Sort-Object { $_.FullName.Length } -Descending |
ForEach-Object {
Remove-Item $_.FullName -Recurse -Force
Write-Host " Удалено: $($_.FullName -replace [regex]::Escape($projectRoot), '.')"
if (Test-Path -LiteralPath $_.FullName) {
Remove-Item -LiteralPath $_.FullName -Recurse -Force
Write-Host " Удалено: $($_.FullName -replace [regex]::Escape($projectRoot), '.')"
}
}
Write-Host "Очистка завершена."
}
@ -51,16 +54,36 @@ if ($InstallBuildDeps) {
# ── Сборка ─────────────────────────────────────────────────────────
Write-Host ""
Write-Host "=== Сборка EXE ==="
& $venvPython -m PyInstaller --noconfirm --clean plate_config_app.spec
if ($LASTEXITCODE -ne 0) {
throw "PyInstaller завершился с ошибкой (код $LASTEXITCODE)."
$pythonHome = (& $venvPython -c "import sys; print(sys.base_prefix)").Trim()
if ($LASTEXITCODE -ne 0 -or -not (Test-Path -LiteralPath $pythonHome -PathType Container)) {
throw "Не удалось определить каталог Python для сборки."
}
$buildPath = @(
(Split-Path -Parent $venvPython),
$pythonHome,
(Join-Path $pythonHome "DLLs"),
(Join-Path $env:SystemRoot "System32"),
$env:SystemRoot
) -join ";"
$originalPath = $env:PATH
try {
$env:PATH = $buildPath
& $venvPython -m PyInstaller --noconfirm --clean plate_config_app.spec
$buildExitCode = $LASTEXITCODE
} finally {
$env:PATH = $originalPath
}
$exe = Join-Path $projectRoot "dist\plate_config_app.exe"
if (-not (Test-Path $exe)) {
if ($buildExitCode -ne 0) {
throw "PyInstaller завершился с ошибкой (код $buildExitCode)."
}
$builtExe = Join-Path $projectRoot "dist\plate_config_app.exe"
if (-not (Test-Path $builtExe)) {
throw "EXE не найден после сборки."
}
$exe = Join-Path $projectRoot "dist\plate_config_app-windows-x86_64.exe"
Move-Item -LiteralPath $builtExe -Destination $exe -Force
# ── Копирование ресурсных папок рядом с EXE ───────────────────────
Write-Host ""
@ -77,6 +100,18 @@ foreach ($folder in $resourceFolders) {
}
}
$archive = Join-Path $projectRoot "dist\plate_config_app-windows-x86_64.zip"
$archiveEntries = @($exe)
foreach ($folder in $resourceFolders) {
$resourcePath = Join-Path $projectRoot "dist\$folder"
if (Test-Path $resourcePath) {
$archiveEntries += $resourcePath
}
}
Compress-Archive -LiteralPath $archiveEntries -DestinationPath $archive -Force
$size = [math]::Round((Get-Item $exe).Length / 1MB, 1)
$archiveSize = [math]::Round((Get-Item $archive).Length / 1MB, 1)
Write-Host ""
Write-Host "Сборка завершена: $exe ($size MB)"
Write-Host "ZIP-архив: $archive ($archiveSize MB)"

59
build_linux.sh Normal file
View File

@ -0,0 +1,59 @@
#!/usr/bin/env bash
# Нативная сборка одного исполняемого файла для Debian x86_64.
set -euo pipefail
project_root="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
venv_dir="$project_root/.venv-linux"
build_dir="$project_root/build-linux"
dist_dir="$project_root/dist-linux"
artifact_name="plate_config_app-linux-x86_64"
archive_name="plate_config_app-linux-x86_64.zip"
cd "$project_root"
rm -rf "$build_dir" "$dist_dir"
python3 -m venv "$venv_dir"
"$venv_dir/bin/python" -m pip install --upgrade pip
"$venv_dir/bin/python" -m pip install -r requirements.txt -r requirements-build.txt
"$venv_dir/bin/python" -m unittest discover -s tests -v
"$venv_dir/bin/python" -m PyInstaller \
--noconfirm \
--clean \
--workpath "$build_dir" \
--distpath "$dist_dir" \
plate_config_app.spec
install -m 0755 "$dist_dir/plate_config_app" "$dist_dir/$artifact_name"
rm -f "$dist_dir/plate_config_app"
for resource_dir in light_modes input_mappings commands; do
cp -a "$resource_dir" "$dist_dir/$resource_dir"
done
cat > "$dist_dir/README.txt" <<'EOF'
Plate Config App для Debian 13 x86_64
Запуск:
./plate_config_app-linux-x86_64
Папки light_modes, input_mappings и commands можно изменять без пересборки.
Для доступа к USB-COM добавьте пользователя в группу dialout и войдите в систему заново:
sudo usermod -aG dialout "$USER"
EOF
(
cd "$dist_dir"
"$venv_dir/bin/python" -m zipfile -c \
"$archive_name" \
"$artifact_name" \
light_modes \
input_mappings \
commands \
README.txt
)
printf 'Готово: %s\n' "$dist_dir/$artifact_name"
sha256sum "$dist_dir/$artifact_name"
printf 'ZIP-архив: %s\n' "$dist_dir/$archive_name"
sha256sum "$dist_dir/$archive_name"

View File

@ -1,11 +1,12 @@
import os
import sys
# В windowed-сборке (PyInstaller --noconsole) stdout/stderr могут быть None.
# Подменяем их на devnull, чтобы print() и faulthandler не падали.
if sys.stderr is None:
sys.stderr = open("NUL", "w") # noqa: SIM115
sys.stderr = open(os.devnull, "w") # noqa: SIM115
if sys.stdout is None:
sys.stdout = open("NUL", "w") # noqa: SIM115
sys.stdout = open(os.devnull, "w") # noqa: SIM115
import faulthandler

262
tests/test_auth_service.py Normal file
View File

@ -0,0 +1,262 @@
from __future__ import annotations
import unittest
from collections import deque
from unittest.mock import patch
from app.services.auth_service import (
AuthStatus,
_aes_encrypt,
check_status,
firmware_requires_auth,
perform_handshake,
perform_required_handshake,
perform_startup_handshake,
wait_for_startup_banner,
)
class FakeConnection:
def __init__(self, responses: list[str | None]) -> None:
self.is_connected = True
self._responses = deque(responses)
self.events: list[str] = []
self.release_count = 0
self.firmware_version: str | None = None
self.auth_required: bool | None = None
def read_response(self, timeout: float | None = None) -> str | None:
response = self._responses.popleft() if self._responses else None
self.events.append(f"read:{response}")
return response
def send_command(self, command: str) -> None:
self.events.append(f"send:{command}")
def write_raw(self, command: str) -> None:
self.events.append(f"write:{command}")
def release(self) -> None:
self.release_count += 1
def set_firmware_version(self, version: str, *, auth_required: bool) -> None:
self.firmware_version = version
self.auth_required = auth_required
class StartupHandshakeTests(unittest.TestCase):
_HOST_CHALLENGE = bytes.fromhex("00112233445566778899AABBCCDDEEFF")
_CHAL = "CHAL00112233445566778899AABBCCDDEEFF"
_RAES = "RAES40E9A18C22C2CFCD764666C90B358F9D"
def _perform_handshake(
self,
conn: FakeConnection,
*,
timeout: float = 0.1,
ready_timeout: float = 0.1,
):
with patch(
"app.services.auth_service.os.urandom",
return_value=self._HOST_CHALLENGE,
):
return perform_handshake(
conn,
timeout=timeout,
ready_timeout=ready_timeout,
)
def test_waits_for_banner_and_two_seconds_before_sending_auth(self) -> None:
conn = FakeConnection([
"boot log",
"periphery controller ver_2.3.6 release",
self._CHAL,
self._RAES,
"AUTH_PC_OK",
"READY!",
])
sleep_calls: list[float] = []
def fake_sleep(seconds: float) -> None:
sleep_calls.append(seconds)
conn.events.append(f"sleep:{seconds}")
with patch(
"app.services.auth_service.os.urandom",
return_value=self._HOST_CHALLENGE,
):
result = perform_startup_handshake(
conn,
startup_delay=2.0,
sleep_fn=fake_sleep,
)
self.assertEqual(AuthStatus.OK, result.status)
self.assertTrue(result.startup_ready)
self.assertEqual([2.0], sleep_calls)
sleep_index = conn.events.index("sleep:2.0")
send_index = next(
index for index, event in enumerate(conn.events) if event.startswith("send:07")
)
self.assertLess(sleep_index, send_index)
self.assertEqual("2.3.6 release", conn.firmware_version)
self.assertTrue(conn.auth_required)
def test_no_auth_banner_skips_handshake(self) -> None:
conn = FakeConnection([
"boot log",
"periphery controller ver_2.3.6_No_Auth_prizes",
])
sleep_calls: list[float] = []
result = perform_startup_handshake(
conn,
startup_delay=2.0,
sleep_fn=sleep_calls.append,
)
self.assertEqual(AuthStatus.OK, result.status)
self.assertTrue(result.startup_ready)
self.assertTrue(result.auth_skipped)
self.assertEqual("2.3.6_No_Auth_prizes", result.firmware_version)
self.assertFalse(conn.auth_required)
self.assertEqual([2.0], sleep_calls)
self.assertFalse(any(event.startswith("send:07") for event in conn.events))
self.assertEqual(1, conn.release_count)
def test_auth_mode_marker_is_case_insensitive(self) -> None:
self.assertFalse(firmware_requires_auth("2.3.6_no_auth_prizes"))
self.assertFalse(firmware_requires_auth("2.3.6_NO_AUTH"))
self.assertTrue(firmware_requires_auth("2.3.6_release"))
def test_manual_handshake_is_skipped_for_known_no_auth_firmware(self) -> None:
conn = FakeConnection([])
conn.set_firmware_version("2.3.6_no_auth", auth_required=False)
result = perform_required_handshake(conn, timeout=0.1)
self.assertEqual(AuthStatus.OK, result.status)
self.assertTrue(result.auth_skipped)
self.assertFalse(any(event.startswith("send:") for event in conn.events))
def test_banner_requires_non_empty_version_suffix(self) -> None:
conn = FakeConnection(["periphery controller ver_", None])
banner = wait_for_startup_banner(conn, timeout=0.1)
self.assertIsNone(banner)
self.assertFalse(any(event.startswith("send:") for event in conn.events))
def test_startup_timeout_never_sends_auth_command(self) -> None:
conn = FakeConnection(["unrelated boot log", None])
result = perform_startup_handshake(
conn,
banner_timeout=0.1,
startup_delay=2.0,
sleep_fn=lambda _seconds: self.fail("Пауза не должна начинаться без banner"),
)
self.assertEqual(AuthStatus.TIMEOUT, result.status)
self.assertFalse(result.startup_ready)
self.assertFalse(any(event.startswith("send:") for event in conn.events))
def test_raes_does_not_complete_handshake(self) -> None:
conn = FakeConnection([self._CHAL, self._RAES, None])
result = self._perform_handshake(conn)
self.assertEqual(AuthStatus.TIMEOUT, result.status)
self.assertIn("READY!", result.detail)
def test_auth_pc_ok_does_not_complete_handshake(self) -> None:
conn = FakeConnection([self._CHAL, self._RAES, "AUTH_PC_OK", None])
result = self._perform_handshake(conn)
self.assertEqual(AuthStatus.TIMEOUT, result.status)
self.assertIn("READY!", result.detail)
def test_ready_without_exclamation_mark_is_rejected(self) -> None:
conn = FakeConnection([self._CHAL, self._RAES, "READY", None])
result = self._perform_handshake(conn)
self.assertEqual(AuthStatus.TIMEOUT, result.status)
def test_exact_ready_bang_completes_handshake(self) -> None:
conn = FakeConnection([
self._CHAL,
self._RAES,
"AUTH_PC_OK",
" READY! ",
])
result = self._perform_handshake(conn)
self.assertEqual(AuthStatus.OK, result.status)
self.assertEqual(1, conn.release_count)
def test_effective_aes_key_matches_controller_profile(self) -> None:
encrypted = _aes_encrypt(self._HOST_CHALLENGE)
self.assertEqual("40E9A18C22C2CFCD764666C90B358F9D", encrypted.hex().upper())
def test_invalid_raes_fails_controller_verification(self) -> None:
conn = FakeConnection([
self._CHAL,
"RAES00000000000000000000000000000000",
"READY!",
])
result = self._perform_handshake(conn)
self.assertEqual(AuthStatus.FAILED, result.status)
self.assertIn("RAES", result.detail)
def test_status_requires_exact_ready_bang(self) -> None:
ready_conn = FakeConnection(["READY!"])
not_ready_conn = FakeConnection(["READY", None])
self.assertEqual(AuthStatus.OK, check_status(ready_conn, timeout=0.1))
self.assertEqual(AuthStatus.NEED_AUTH, check_status(not_ready_conn, timeout=0.1))
def test_no_auth_status_accepts_any_non_empty_response(self) -> None:
conn = FakeConnection(["OK"])
conn.set_firmware_version("2.3.6_no_auth", auth_required=False)
self.assertEqual(AuthStatus.OK, check_status(conn, timeout=0.1))
def test_no_auth_status_timeout_is_connection_failure(self) -> None:
conn = FakeConnection([None])
conn.set_firmware_version("2.3.6_no_auth", auth_required=False)
self.assertEqual(AuthStatus.DISCONNECTED, check_status(conn, timeout=0.1))
def test_new_secure_banner_reenables_handshake(self) -> None:
conn = FakeConnection([
"periphery controller ver_2.0.0_release",
self._CHAL,
self._RAES,
"READY!",
])
conn.set_firmware_version("1.0.0_no_auth", auth_required=False)
with patch(
"app.services.auth_service.os.urandom",
return_value=self._HOST_CHALLENGE,
):
result = perform_startup_handshake(
conn,
startup_delay=0.0,
sleep_fn=lambda _seconds: None,
)
self.assertEqual(AuthStatus.OK, result.status)
self.assertFalse(result.auth_skipped)
self.assertTrue(conn.auth_required)
self.assertTrue(any(event.startswith("send:07") for event in conn.events))
if __name__ == "__main__":
unittest.main()

95
tests/test_com_service.py Normal file
View File

@ -0,0 +1,95 @@
from __future__ import annotations
import unittest
from collections import deque
from app.services.com_service import ComConnection
class FakeSerial:
def __init__(self, chunks: list[bytes] | None = None) -> None:
self.is_open = True
self._timeout = 0.1
self._chunks = deque(chunks or [])
self.events: list[str] = []
self._dtr = True
self._rts = False
@property
def timeout(self) -> float:
return self._timeout
@timeout.setter
def timeout(self, _value: float) -> None:
raise AssertionError("read_response не должен перенастраивать timeout COM-порта")
@property
def in_waiting(self) -> int:
return len(self._chunks[0]) if self._chunks else 0
@property
def dtr(self) -> bool:
return self._dtr
@dtr.setter
def dtr(self, value: bool) -> None:
self._dtr = value
self.events.append(f"dtr:{value}")
@property
def rts(self) -> bool:
return self._rts
@rts.setter
def rts(self, value: bool) -> None:
self._rts = value
self.events.append(f"rts:{value}")
def read(self, _size: int) -> bytes:
return self._chunks.popleft() if self._chunks else b""
def reset_input_buffer(self) -> None:
self.events.append("reset_input_buffer")
def close(self) -> None:
self.is_open = False
class ComConnectionTests(unittest.TestCase):
def test_read_response_keeps_serial_timeout_unchanged(self) -> None:
serial = FakeSerial([b"boot log\r\nperiphery controller ver_2.4.2_MDB\r\n"])
conn = ComConnection()
conn._serial = serial
self.assertEqual("boot log", conn.read_response(timeout=0.2))
self.assertEqual(
"periphery controller ver_2.4.2_MDB",
conn.read_response(timeout=0.2),
)
self.assertEqual(0.1, serial.timeout)
def test_reset_controller_uses_rts_pulse_with_dtr_inactive(self) -> None:
serial = FakeSerial()
conn = ComConnection()
conn._serial = serial
conn.reset_controller(pulse_seconds=0)
self.assertEqual(
["reset_input_buffer", "dtr:False", "rts:True", "rts:False"],
serial.events,
)
def test_close_clears_detected_firmware_mode(self) -> None:
conn = ComConnection()
conn._serial = FakeSerial()
conn.set_firmware_version("2.0.0_no_auth", auth_required=False)
conn.close()
self.assertIsNone(conn.firmware_version)
self.assertIsNone(conn.auth_required)
if __name__ == "__main__":
unittest.main()

View File

@ -0,0 +1,115 @@
from __future__ import annotations
import unittest
from collections import deque
from unittest.mock import patch
from app.services.auth_service import AuthStatus
from app.services.firmware_service import perform_authenticated_ota
class FakeConnection:
def __init__(self, responses: list[str | None]) -> None:
self.is_connected = True
self._responses = deque(responses)
self.events: list[str] = []
self.release_count = 0
self.firmware_version: str | None = None
self.auth_required: bool | None = None
def read_response(self, timeout: float | None = None) -> str | None:
response = self._responses.popleft() if self._responses else None
self.events.append(f"read:{response}")
return response
def send_command(self, command: str) -> None:
self.events.append(f"send:{command}")
def write_raw(self, command: str) -> None:
self.events.append(f"write:{command}")
def release(self) -> None:
self.release_count += 1
def set_firmware_version(self, version: str, *, auth_required: bool) -> None:
self.firmware_version = version
self.auth_required = auth_required
class AuthenticatedOtaTests(unittest.TestCase):
_HOST_CHALLENGE = bytes.fromhex("00112233445566778899AABBCCDDEEFF")
_CHAL = "CHAL00112233445566778899AABBCCDDEEFF"
_RAES = "RAES40E9A18C22C2CFCD764666C90B358F9D"
def _run(self, conn: FakeConnection):
with patch(
"app.services.auth_service.os.urandom",
return_value=self._HOST_CHALLENGE,
):
return perform_authenticated_ota(
conn,
ota_timeout=0.1,
banner_timeout=0.1,
startup_delay=0.0,
handshake_timeout=0.1,
ready_timeout=0.1,
sleep_fn=lambda _seconds: None,
)
def test_sends_04_before_authentication_after_successful_update(self) -> None:
conn = FakeConnection([
"Writing 100%",
"Successful update",
"periphery controller ver_2.4.0 release",
self._CHAL,
self._RAES,
"READY!",
])
result = self._run(conn)
self.assertTrue(result.success)
self.assertEqual(AuthStatus.OK, result.post_auth.status)
send_events = [event for event in conn.events if event.startswith("send:")]
self.assertEqual("send:04", send_events[0])
self.assertTrue(send_events[1].startswith("send:07"))
def test_sends_04_without_pre_authentication(self) -> None:
conn = FakeConnection([None])
result = self._run(conn)
self.assertFalse(result.success)
self.assertEqual("send:04", conn.events[0])
self.assertFalse(any(event.startswith("send:07") for event in conn.events))
def test_no_auth_firmware_skips_post_ota_handshake(self) -> None:
conn = FakeConnection([
"Successful update",
"periphery controller ver_2.0.0_no_auth",
])
result = self._run(conn)
self.assertTrue(result.success)
self.assertEqual(AuthStatus.OK, result.post_auth.status)
self.assertTrue(result.post_auth.auth_skipped)
self.assertFalse(conn.auth_required)
self.assertFalse(any(event.startswith("send:07") for event in conn.events))
self.assertIn("без авторизации", result.detail)
def test_reports_post_authentication_failure_after_update(self) -> None:
conn = FakeConnection([
"Successful update",
None,
])
result = self._run(conn)
self.assertFalse(result.success)
self.assertEqual(AuthStatus.TIMEOUT, result.post_auth.status)
self.assertIn("после OTA", result.detail)
if __name__ == "__main__":
unittest.main()

160
tests/test_start_tab.py Normal file
View File

@ -0,0 +1,160 @@
from __future__ import annotations
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication
from app.services.update_service import UpdateError, UpdateInfo
from app.widgets.start_tab import StartTab, _UpdateCheckThread, _UpdateDownloadThread
class StartTabUpdateTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls._app = QApplication.instance() or QApplication([])
def test_update_button_is_below_settings_and_hidden_initially(self) -> None:
tab = StartTab(lambda: None)
self.assertTrue(tab._update_btn.isHidden())
self.assertLess(
tab._settings_column_layout.indexOf(tab._settings_btn),
tab._settings_column_layout.indexOf(tab._update_btn),
)
tab.shutdown()
def test_available_update_shows_download_button(self) -> None:
tab = StartTab(lambda: None)
info = UpdateInfo(
tag="v2.1.0",
version="2.1.0",
asset_name="plate_config_app-windows-x86_64.zip",
download_url="https://git.robotail.pro/download/test",
)
tab._on_update_check_done(info)
self.assertFalse(tab._update_btn.isHidden())
self.assertEqual("Скачать обновление", tab._update_btn.text())
self.assertIn("v2.1.0", tab._update_btn.toolTip())
tab.shutdown()
def test_no_update_hides_previously_visible_button(self) -> None:
tab = StartTab(lambda: None)
tab._on_update_check_done(self._update_info())
tab._on_update_check_done(None)
self.assertTrue(tab._update_btn.isHidden())
self.assertIsNone(tab._available_update)
tab.shutdown()
def test_progress_updates_button_text(self) -> None:
tab = StartTab(lambda: None)
tab._on_update_progress(57)
self.assertEqual("Скачивание 57%", tab._update_btn.text())
tab.shutdown()
def test_download_failure_resets_button_and_warns(self) -> None:
tab = StartTab(lambda: None)
tab._update_btn.setEnabled(False)
tab._update_btn.setText("Скачивание 42%")
with patch("app.widgets.start_tab.QMessageBox.warning") as warning:
tab._on_update_failed("ошибка сети")
self.assertTrue(tab._update_btn.isEnabled())
self.assertEqual("Скачать обновление", tab._update_btn.text())
warning.assert_called_once()
self.assertIn("ошибка сети", warning.call_args.args[2])
tab.shutdown()
def test_launch_failure_removes_download_and_warns(self) -> None:
tab = StartTab(lambda: None)
with tempfile.TemporaryDirectory() as directory:
downloaded = Path(directory) / ".downloaded-update"
downloaded.write_bytes(b"MZupdate")
with (
patch(
"app.widgets.start_tab.launch_update",
side_effect=UpdateError("не удалось запустить helper"),
),
patch("app.widgets.start_tab.QMessageBox.warning") as warning,
):
tab._on_update_downloaded(downloaded)
self.assertFalse(downloaded.exists())
self.assertEqual("Скачать обновление", tab._update_btn.text())
self.assertTrue(tab._update_btn.isEnabled())
warning.assert_called_once()
tab.shutdown()
def test_successful_download_launches_update_and_closes_window(self) -> None:
tab = StartTab(lambda: None)
downloaded = Path("downloaded-update")
with (
patch("app.widgets.start_tab.launch_update") as launcher,
patch.object(tab, "close") as close_window,
):
tab._on_update_downloaded(downloaded)
launcher.assert_called_once_with(downloaded)
close_window.assert_called_once_with()
self.assertEqual("Перезапуск…", tab._update_btn.text())
tab.shutdown()
def test_update_check_thread_hides_network_errors(self) -> None:
thread = _UpdateCheckThread()
results: list[UpdateInfo | None] = []
thread.result_ready.connect(results.append)
with patch(
"app.widgets.start_tab.check_for_update",
side_effect=UpdateError("offline"),
):
thread.run()
self.assertEqual([None], results)
def test_download_thread_emits_progress_and_result(self) -> None:
info = self._update_info()
result_path = Path("downloaded-update")
thread = _UpdateDownloadThread(info, Path("."))
progress_values: list[int] = []
results: list[Path] = []
thread.progress_changed.connect(progress_values.append)
thread.result_ready.connect(results.append)
def fake_download(_info, _directory, *, progress, cancelled):
self.assertFalse(cancelled())
progress(75)
return result_path
with patch("app.widgets.start_tab.download_update", side_effect=fake_download):
thread.run()
self.assertEqual([75], progress_values)
self.assertEqual([result_path], results)
@staticmethod
def _update_info() -> UpdateInfo:
return UpdateInfo(
tag="v2.1.0",
version="2.1.0",
asset_name="plate_config_app-windows-x86_64.zip",
download_url="https://git.robotail.pro/download/test",
)
if __name__ == "__main__":
unittest.main()

View File

@ -0,0 +1,716 @@
from __future__ import annotations
import hashlib
import io
import os
import shutil
import stat
import subprocess
import sys
import tempfile
import time
import unittest
import zipfile
from pathlib import Path
from unittest.mock import patch
from app.services import update_service
from app.services.update_service import (
UpdateCancelled,
UpdateError,
UpdateInfo,
archive_executable_name,
build_download_url,
check_for_update,
download_update,
extract_release_tag,
launch_update,
normalize_version,
parse_version,
platform_asset_name,
platform_executable_name,
signal_update_ready,
)
class FakeResponse:
def __init__(
self,
url: str,
data: bytes = b"",
*,
content_length: int | None = None,
) -> None:
self._url = url
self._stream = io.BytesIO(data)
self.headers: dict[str, str] = {}
if content_length is not None:
self.headers["Content-Length"] = str(content_length)
def __enter__(self):
return self
def __exit__(self, *_args) -> None:
return None
def geturl(self) -> str:
return self._url
def read(self, size: int = -1) -> bytes:
return self._stream.read(size)
def make_update_archive(
executable_name: str,
executable_data: bytes,
*,
extra_files: dict[str, bytes] | None = None,
) -> bytes:
stream = io.BytesIO()
with zipfile.ZipFile(stream, "w", zipfile.ZIP_DEFLATED) as archive:
archive.writestr(executable_name, executable_data)
for name, data in (extra_files or {}).items():
archive.writestr(name, data)
return stream.getvalue()
class UpdateServiceTests(unittest.TestCase):
def test_normalizes_version_prefix_and_whitespace(self) -> None:
self.assertEqual("2.0.0", normalize_version(" v2.0.0 "))
self.assertEqual("2.0.0", normalize_version("V2.0.0"))
def test_parses_version_into_numeric_components(self) -> None:
self.assertEqual((12, 34, 56), parse_version("v12.34.56"))
self.assertEqual((2, 0, 0), parse_version(" 2.0.0 "))
def test_rejects_non_standard_versions(self) -> None:
invalid_versions = (
"v2.0",
"v2.0.0.1",
"v2.0.0-beta",
"v2.01.0",
"release-2.0.0",
)
for version in invalid_versions:
with self.subTest(version=version), self.assertRaises(UpdateError):
parse_version(version)
def test_extracts_tag_from_latest_redirect(self) -> None:
tag = extract_release_tag(
"https://git.robotail.pro/Robotail/plate-config-app/releases/tag/v2.1.0"
)
self.assertEqual("v2.1.0", tag)
def test_rejects_invalid_release_redirects(self) -> None:
invalid_urls = (
"https://git.robotail.pro/user/login",
"http://git.robotail.pro/Robotail/plate-config-app/releases/tag/v2.1.0",
"https://example.com/Robotail/plate-config-app/releases/tag/v2.1.0",
)
for url in invalid_urls:
with self.subTest(url=url), self.assertRaises(UpdateError):
extract_release_tag(url)
def test_build_download_url_escapes_tag_and_asset(self) -> None:
url = build_download_url("v2.1.0 rc/1", "plate config.zip")
self.assertTrue(url.endswith("/v2.1.0%20rc%2F1/plate%20config.zip"))
def test_equal_or_older_tag_does_not_offer_update(self) -> None:
cases = (
("2.0.0", "v2.0.0"),
("2.0.0", "v1.99.99"),
("2.5.0", "v2.4.99"),
("2.5.4", "v2.5.3"),
)
for current, remote in cases:
with self.subTest(current=current, remote=remote):
result = check_for_update(
current,
opener=self._release_opener(remote),
system="Windows",
machine="AMD64",
)
self.assertIsNone(result)
def test_newer_major_minor_and_patch_versions_offer_zip(self) -> None:
cases = (
("1.9.9", "v2.0.0"),
("2.4.9", "v2.5.0"),
("2.5.3", "v2.5.4"),
)
for current, remote in cases:
with self.subTest(current=current, remote=remote):
result = check_for_update(
current,
opener=self._release_opener(remote),
system="Windows",
machine="x86_64",
)
self.assertIsNotNone(result)
assert result is not None
self.assertEqual("plate_config_app-windows-x86_64.zip", result.asset_name)
self.assertTrue(result.download_url.endswith("/" + result.asset_name))
def test_rejects_invalid_current_or_remote_version(self) -> None:
for current, remote in (("development", "v2.1.0"), ("2.0.0", "v2.1")):
with self.subTest(current=current, remote=remote), self.assertRaises(
UpdateError
):
check_for_update(
current,
opener=self._release_opener(remote),
system="Linux",
machine="x86_64",
)
def test_wraps_network_error_during_update_check(self) -> None:
def opener(_request, timeout: float):
raise OSError("offline")
with self.assertRaisesRegex(UpdateError, "offline") as raised:
check_for_update("2.0.0", opener=opener)
self.assertIsInstance(raised.exception.__cause__, OSError)
def test_platform_names(self) -> None:
self.assertEqual(
"plate_config_app-windows-x86_64.exe",
platform_executable_name(system="Windows", machine="AMD64"),
)
self.assertEqual(
"plate_config_app-windows-x86_64.zip",
platform_asset_name(system="Windows", machine="AMD64"),
)
self.assertEqual(
"plate_config_app-linux-x86_64.zip",
platform_asset_name(system="Linux", machine="x86_64"),
)
self.assertEqual(
"plate_config_app-linux-x86_64",
archive_executable_name("plate_config_app-linux-x86_64.zip"),
)
def test_rejects_unsupported_platform_architecture_or_asset(self) -> None:
with self.assertRaisesRegex(UpdateError, "архитектуру"):
platform_asset_name(system="Linux", machine="arm64")
with self.assertRaisesRegex(UpdateError, "платформу"):
platform_asset_name(system="Darwin", machine="x86_64")
with self.assertRaisesRegex(UpdateError, "Неизвестное"):
archive_executable_name("update.zip")
def test_signals_restart_only_when_ready_file_is_requested(self) -> None:
with tempfile.TemporaryDirectory() as directory:
marker = Path(directory) / "ready"
with patch.dict(os.environ, {update_service.UPDATE_READY_ENV: str(marker)}):
signal_update_ready()
self.assertEqual(str(os.getpid()), marker.read_text(encoding="ascii"))
self.assertNotIn(update_service.UPDATE_READY_ENV, os.environ)
marker.unlink()
signal_update_ready()
self.assertFalse(marker.exists())
def test_downloads_and_validates_windows_zip(self) -> None:
info = self._update_info()
data = make_update_archive(
"plate_config_app-windows-x86_64.exe",
b"MZnew",
extra_files={"commands/default.csv": b"04;Update"},
)
progress_values: list[int] = []
with tempfile.TemporaryDirectory() as directory:
result = download_update(
info,
Path(directory),
opener=lambda _request, timeout: FakeResponse(
info.download_url,
data,
content_length=len(data),
),
progress=progress_values.append,
)
self.assertEqual(data, result.read_bytes())
self.assertEqual(100, progress_values[-1])
def test_downloads_and_validates_linux_zip(self) -> None:
info = self._update_info(system="linux")
data = make_update_archive("plate_config_app-linux-x86_64", b"\x7fELFnew")
with tempfile.TemporaryDirectory() as directory:
result = download_update(
info,
Path(directory),
opener=lambda _request, timeout: FakeResponse(info.download_url, data),
)
with zipfile.ZipFile(result) as archive:
self.assertEqual(
["plate_config_app-linux-x86_64"],
archive.namelist(),
)
def test_rejects_invalid_or_incomplete_download(self) -> None:
info = self._update_info()
valid_data = make_update_archive(
"plate_config_app-windows-x86_64.exe",
b"MZnew",
)
cases = (
(b"<html>login</html>", None),
(valid_data, len(valid_data) + 10),
(b"", None),
)
for data, content_length in cases:
with self.subTest(size=len(data)), tempfile.TemporaryDirectory() as directory:
target = Path(directory)
with self.assertRaises(UpdateError):
download_update(
info,
target,
opener=lambda _request, timeout, payload=data, length=content_length: (
FakeResponse(
info.download_url,
payload,
content_length=length,
)
),
)
self.assertEqual([], list(target.iterdir()))
def test_rejects_login_or_foreign_download_redirect(self) -> None:
info = self._update_info()
data = make_update_archive(
"plate_config_app-windows-x86_64.exe",
b"MZnew",
)
redirects = (
"https://git.robotail.pro/user/login",
"https://example.com/update.zip",
)
for redirect in redirects:
with self.subTest(redirect=redirect), tempfile.TemporaryDirectory() as directory:
with self.assertRaises(UpdateError):
download_update(
info,
Path(directory),
opener=lambda _request, timeout, url=redirect: FakeResponse(
url,
data,
),
)
def test_rejects_oversized_downloads(self) -> None:
info = self._update_info()
with tempfile.TemporaryDirectory() as directory, patch.object(
update_service,
"MAX_DOWNLOAD_BYTES",
4,
):
for content_length in (8, None):
with self.subTest(content_length=content_length), self.assertRaisesRegex(
UpdateError,
"превышает",
):
download_update(
info,
Path(directory),
opener=lambda _request, timeout, length=content_length: FakeResponse(
info.download_url,
b"12345678",
content_length=length,
),
)
def test_cancels_download_and_removes_partial_file(self) -> None:
info = self._update_info()
with tempfile.TemporaryDirectory() as directory:
target = Path(directory)
with self.assertRaises(UpdateCancelled):
download_update(
info,
target,
opener=lambda _request, timeout: FakeResponse(
info.download_url,
b"partial",
),
cancelled=lambda: True,
)
self.assertEqual([], list(target.iterdir()))
def test_rejects_missing_or_read_only_target_directory(self) -> None:
info = self._update_info()
with tempfile.TemporaryDirectory() as directory:
with self.assertRaisesRegex(UpdateError, "не найден"):
download_update(info, Path(directory) / "missing")
with patch("app.services.update_service.os.access", return_value=False):
with self.assertRaisesRegex(UpdateError, "Нет прав"):
download_update(info, Path(directory))
def test_rejects_archive_without_executable(self) -> None:
data = self._custom_archive({"commands/default.csv": b"04;Update"})
self._assert_invalid_archive(data, "исполняемый файл")
def test_rejects_archive_path_traversal(self) -> None:
data = self._custom_archive(
{
"plate_config_app-windows-x86_64.exe": b"MZnew",
"../outside.txt": b"unsafe",
}
)
self._assert_invalid_archive(data, "небезопасный путь")
def test_rejects_archive_symlink(self) -> None:
stream = io.BytesIO()
with zipfile.ZipFile(stream, "w") as archive:
archive.writestr("plate_config_app-windows-x86_64.exe", b"MZnew")
link = zipfile.ZipInfo("commands/link")
link.create_system = 3
link.external_attr = (stat.S_IFLNK | 0o777) << 16
archive.writestr(link, "../../outside")
self._assert_invalid_archive(stream.getvalue(), "ссылку")
def test_rejects_duplicate_archive_paths(self) -> None:
stream = io.BytesIO()
with zipfile.ZipFile(stream, "w") as archive:
archive.writestr("plate_config_app-windows-x86_64.exe", b"MZnew")
archive.writestr("commands/Test.csv", b"one")
archive.writestr("commands/test.csv", b"two")
self._assert_invalid_archive(stream.getvalue(), "повторяющиеся")
def test_rejects_excessive_unpacked_size(self) -> None:
data = make_update_archive(
"plate_config_app-windows-x86_64.exe",
b"MZnew",
extra_files={"commands/large.csv": b"12345678"},
)
with patch.object(update_service, "MAX_UNPACKED_BYTES", 8):
self._assert_invalid_archive(data, "Распакованный размер")
def test_prepares_windows_package_replacement_helper(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "plate_config_app-windows-x86_64.exe"
archive = root / ".downloaded-update.zip"
target.write_bytes(b"MZold")
archive.write_bytes(
make_update_archive(
target.name,
b"MZnew",
extra_files={"commands/default.csv": b"new"},
)
)
with patch("app.services.update_service.subprocess.Popen") as popen:
launch_update(
archive,
executable_path=target,
process_id=123,
system="Windows",
)
scripts = list(root.glob(".plate_config_updater-*.ps1"))
staging = list(root.glob(".plate_config_update-*"))
self.assertEqual(1, len(scripts))
self.assertEqual(1, len(staging))
self.assertFalse(archive.exists())
self.assertEqual(b"new", (staging[0] / "commands/default.csv").read_bytes())
script = scripts[0].read_text(encoding="utf-8-sig")
self.assertIn("$Staging", script)
self.assertIn("$backedUp", script)
self.assertIn("AddSeconds(30)", script)
self.assertIn("$restarted.HasExited", script)
self.assertIn("$ReadyFile", script)
self.assertIn('$env:PYINSTALLER_RESET_ENVIRONMENT = "1"', script)
popen.assert_called_once()
@unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux")
def test_prepares_linux_package_replacement_helper(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "plate_config_app-linux-x86_64"
archive = root / ".downloaded-update.zip"
target.write_bytes(b"\x7fELFold")
archive.write_bytes(
make_update_archive(
target.name,
b"\x7fELFnew",
extra_files={"commands/default.csv": b"new"},
)
)
with patch("app.services.update_service.subprocess.Popen") as popen:
launch_update(
archive,
executable_path=target,
process_id=123,
system="Linux",
)
scripts = list(root.glob(".plate_config_updater-*.sh"))
staging = list(root.glob(".plate_config_update-*"))
self.assertEqual(1, len(scripts))
self.assertEqual(1, len(staging))
self.assertTrue((staging[0] / target.name).stat().st_mode & stat.S_IXUSR)
script = scripts[0].read_text(encoding="utf-8")
self.assertIn("staging_path", script)
self.assertIn("rollback", script)
self.assertIn("PLATE_CONFIG_UPDATE_READY_FILE", script)
self.assertIn("PYINSTALLER_RESET_ENVIRONMENT=1", script)
popen.assert_called_once()
def test_extracts_external_archive_next_to_executable(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
downloads = root / "downloads"
application = root / "application"
downloads.mkdir()
application.mkdir()
target = application / "plate_config_app-linux-x86_64"
archive = downloads / "update.zip"
target.write_bytes(b"\x7fELFold")
archive.write_bytes(make_update_archive(target.name, b"\x7fELFnew"))
with patch("app.services.update_service.subprocess.Popen") as popen:
launch_update(
archive,
executable_path=target,
process_id=123,
system="Linux",
)
staging = list(application.glob(".plate_config_update-*"))
self.assertEqual(1, len(staging))
self.assertFalse(archive.exists())
self.assertIn(str(staging[0]), popen.call_args.args[0])
def test_failed_helper_launch_removes_staging_and_script(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "plate_config_app-windows-x86_64.exe"
archive = root / "update.zip"
target.write_bytes(b"MZold")
archive.write_bytes(make_update_archive(target.name, b"MZnew"))
with patch(
"app.services.update_service.subprocess.Popen",
side_effect=OSError("helper failed"),
), self.assertRaisesRegex(OSError, "helper failed"):
launch_update(
archive,
executable_path=target,
process_id=123,
system="Windows",
)
self.assertEqual([], list(root.glob(".plate_config_update-*")))
self.assertEqual([], list(root.glob(".plate_config_updater-*")))
def test_rejects_missing_archive_or_unsupported_platform(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "plate_config_app"
archive = root / "update.zip"
target.write_bytes(b"old")
with self.assertRaisesRegex(UpdateError, "не найден"):
launch_update(archive, executable_path=target, system="Linux")
archive.write_bytes(b"not-a-zip")
with self.assertRaisesRegex(UpdateError, "платформу"):
launch_update(archive, executable_path=target, system="Darwin")
def _assert_invalid_archive(self, data: bytes, message: str) -> None:
info = self._update_info()
with tempfile.TemporaryDirectory() as directory:
target = Path(directory)
with self.assertRaisesRegex(UpdateError, message):
download_update(
info,
target,
opener=lambda _request, timeout: FakeResponse(
info.download_url,
data,
),
)
self.assertEqual([], list(target.iterdir()))
@staticmethod
def _custom_archive(files: dict[str, bytes]) -> bytes:
stream = io.BytesIO()
with zipfile.ZipFile(stream, "w", zipfile.ZIP_DEFLATED) as archive:
for name, data in files.items():
archive.writestr(name, data)
return stream.getvalue()
@staticmethod
def _update_info(*, system: str = "windows") -> UpdateInfo:
asset_name = f"plate_config_app-{system}-x86_64.zip"
return UpdateInfo(
tag="v2.1.0",
version="2.1.0",
asset_name=asset_name,
download_url=(
"https://git.robotail.pro/Robotail/plate-config-app/releases/"
f"download/v2.1.0/{asset_name}"
),
)
@staticmethod
def _release_opener(tag: str):
def opener(_request, timeout: float):
return FakeResponse(
"https://git.robotail.pro/Robotail/plate-config-app/"
f"releases/tag/{tag}"
)
return opener
class ReplacementIntegrationTests(unittest.TestCase):
@staticmethod
def _digest(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
@staticmethod
def _wait_for_rollback(
target: Path,
expected_digest: str,
replaced_resource: Path,
) -> None:
deadline = time.monotonic() + 20
while time.monotonic() < deadline:
if (
target.is_file()
and ReplacementIntegrationTests._digest(target) == expected_digest
and replaced_resource.is_file()
and replaced_resource.read_text(encoding="utf-8") == "old"
and not list(target.parent.glob(".plate_config_updater-*"))
and not list(target.parent.glob(".plate_config_update-*"))
and not list(target.parent.glob(".plate_config_backup-*"))
and (target.parent / "update-error.log").is_file()
):
return
time.sleep(0.1)
raise AssertionError("Внешний helper не откатил пакет за 20 секунд")
@unittest.skipUnless(sys.platform == "win32", "Тест предназначен для Windows")
def test_windows_helper_rolls_back_when_new_app_exits_early(self) -> None:
system_root = Path(os.environ.get("SystemRoot", r"C:\Windows"))
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "plate_config_app-windows-x86_64.exe"
archive = root / "update.zip"
new_executable = (system_root / "System32" / "whoami.exe").read_bytes()
shutil.copy2(system_root / "System32" / "where.exe", target)
self._prepare_old_files(root)
archive.write_bytes(
make_update_archive(
target.name,
new_executable,
extra_files={"commands/version.txt": b"new"},
)
)
expected_digest = self._digest(target)
processes = self._launch_and_capture(archive, target, "Windows")
self._wait_for_rollback(
target,
expected_digest,
root / "commands/version.txt",
)
processes[0].wait(timeout=5)
self._assert_package_preserved(root)
self._wait_for_windows_unlock(target)
@unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux")
def test_linux_helper_rolls_back_when_new_app_exits_early(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "plate_config_app-linux-x86_64"
archive = root / "update.zip"
new_executable = Path("/bin/true").read_bytes()
shutil.copy2("/bin/false", target)
self._prepare_old_files(root)
archive.write_bytes(
make_update_archive(
target.name,
new_executable,
extra_files={"commands/version.txt": b"new"},
)
)
expected_digest = self._digest(target)
processes = self._launch_and_capture(archive, target, "Linux")
self._wait_for_rollback(
target,
expected_digest,
root / "commands/version.txt",
)
processes[0].wait(timeout=5)
self._assert_package_preserved(root)
self.assertTrue(target.stat().st_mode & stat.S_IXUSR)
@staticmethod
def _prepare_old_files(root: Path) -> None:
commands = root / "commands"
commands.mkdir()
(commands / "version.txt").write_text("old", encoding="utf-8")
(commands / "obsolete.txt").write_text("remove", encoding="utf-8")
(root / "settings.json").write_text("preserve", encoding="utf-8")
@staticmethod
def _assert_package_preserved(root: Path) -> None:
self_settings = (root / "settings.json").read_text(encoding="utf-8")
if self_settings != "preserve":
raise AssertionError("settings.json не сохранён")
if not (root / "commands/obsolete.txt").exists():
raise AssertionError("Старое содержимое каталога не восстановлено")
@staticmethod
def _launch_and_capture(
archive: Path,
target: Path,
system: str,
) -> list[subprocess.Popen]:
processes: list[subprocess.Popen] = []
real_popen = subprocess.Popen
def start_process(*args, **kwargs):
process = real_popen(*args, **kwargs)
processes.append(process)
return process
with patch(
"app.services.update_service.subprocess.Popen",
side_effect=start_process,
):
launch_update(
archive,
executable_path=target,
process_id=2_147_483_647,
system=system,
)
return processes
@staticmethod
def _wait_for_windows_unlock(target: Path) -> None:
deadline = time.monotonic() + 10
while True:
try:
target.unlink()
return
except PermissionError:
if time.monotonic() >= deadline:
raise
time.sleep(0.1)
if __name__ == "__main__":
unittest.main()

View File

@ -72,6 +72,12 @@ by: Паша Николаевич
После загрузки контроллер не принимает никаких команд, кроме запросов на авторизацию.
После открытия COM-порта программа выполняет штатный аппаратный сброс контроллера
коротким импульсом RTS и затем ожидает отдельную строку
`periphery controller ver_<номер версии и комментарий>`. До её получения программа
не отправляет контроллеру команды. После получения строки программа выдерживает паузу
`2000 мс` и только затем начинает авторизацию.
Запрос – требование зашифровать строку шестнадцатеричных символов с помощью AES256 алгоритма, используя известный ключ.
Компьютер генерирует случайную последовательность из 16 байт и отправляет ее контроллеру на шифрование. Тот ее шифрует и отвечает. Компьютер расшифровывает и проверяет на совпадение.
@ -93,7 +99,10 @@ by: Паша Николаевич
4. **Контроллер ответит на первый запрос компа**:
**RAES\<symbols\> (16 байт или 32 HEX-символа)**
***Пример:*** **RAES***99ED731648AB8C1709CC4C3C8C3BB0E1*
5. При успехе контроллер отправляет `AUTH_PC_OK` и статус. При ошибке - `AUTH_PC_FAIL`.
5. Контроллер может отправить промежуточные ответы `AUTH_PC_OK` и статус.
6. Авторизация считается полностью завершённой только после получения отдельной точной
строки `READY!`. Ответы `RAES`, `AUTH_PC_OK` и любые другие строки сами по себе не
означают успешного завершения авторизации.
# Команды