Danil Iarosh
add8aff0af
- Добавлена обратная совместимость с прошивками, содержащими маркер no_auth. - Повторная авторизация после команды 04 и успешного обновления прошивки приведена к требуемому порядку. - В терминал команд добавлена ручная кнопка авторизации. - Реализовано OTA-обновление приложения с загрузкой платформенного артефакта, безопасной заменой и перезапуском. - Проверка обновлений учитывает только более новые версии по схеме major, minor и patch. - Обновлены имена Windows и Linux сборок, документация и тестовое покрытие.
101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
"""Запуск 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()
|