132 lines
6.4 KiB
Python
132 lines
6.4 KiB
Python
# Клиент A/B-загрузчика ProtoCAN с оконной передачей блоков по восемь байт. Ответы принимаются
|
||
# только для целевого адреса и текущей сессии; продвижение по этапам зависит от подтверждения,
|
||
# а не от факта отправки.
|
||
|
||
"""Legacy ProtoCAN Boot client ported from Gui_Android CanFirmwareProtocol.kt."""
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
import struct
|
||
|
||
from .protocan import ProtoCanId
|
||
from .transport import build_frame
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CanBootTarget:
|
||
device_type: int = 0
|
||
device: int = 13
|
||
product_type: int = 0x166
|
||
hardware_min: int = 0
|
||
hardware_max: int = 255
|
||
session_id: int = 1
|
||
|
||
def __post_init__(self):
|
||
if not 0 <= self.device_type <= 7 or not 0 <= self.device <= 15:
|
||
raise ValueError("Адрес ProtoCAN: Device Type 0…7, Device 0…15")
|
||
if not 0 <= self.product_type <= 65535:
|
||
raise ValueError("Product Type вне диапазона u16")
|
||
if not 0 <= self.hardware_min <= self.hardware_max <= 255:
|
||
raise ValueError("Неверный диапазон аппаратных ревизий")
|
||
if not 1 <= self.session_id <= 255:
|
||
raise ValueError("Session ID должен быть 1…255")
|
||
|
||
|
||
class CanBootTransfer:
|
||
"""Windowed 8-byte blocks; control 0x9, slots 0xA/0xB, status 0xC."""
|
||
def __init__(self, image, target: CanBootTarget):
|
||
if not 0 < len(image.data) <= 512 * 1024:
|
||
raise ValueError("ProtoCAN Boot: размер образа 1…512 КиБ")
|
||
if not 0 <= image.version <= 0xFFFFFFFF:
|
||
raise ValueError("Версия образа вне диапазона u32")
|
||
self.image, self.target = image, target
|
||
self.stage = "enter"
|
||
self.next_block = 0
|
||
self.slot = None
|
||
self.finished = False
|
||
self._window_end = 0
|
||
|
||
@property
|
||
def percent(self):
|
||
return min(100, self.next_block * 8 * 100 // len(self.image.data))
|
||
|
||
def _frame(self, kind, body, data=b""):
|
||
t = self.target
|
||
return build_frame(ProtoCanId.build(1, 0, t.device_type, t.device, kind, body), data, to_can=True)
|
||
|
||
def _control(self, command, data=b""):
|
||
return [self._frame(9, self.target.session_id << 8 | command, data)]
|
||
|
||
def start(self):
|
||
return self._control(2)
|
||
|
||
def abort(self):
|
||
return self._control(10)
|
||
|
||
def accepts(self, frame):
|
||
p = ProtoCanId.parse(frame.can_id)
|
||
t = self.target
|
||
return (not frame.to_can and frame.ide and not frame.rtr and not frame.is_error
|
||
and p.msg_type == 12 and p.pm == 1 and p.device_type == t.device_type
|
||
and p.device == t.device and p.body >> 8 == t.session_id and len(frame.data) == 8)
|
||
|
||
# Окно содержит до 16 блоков. Последний неполный блок заполняется 0xFF,
|
||
# но метаданные образа сохраняют исходную длину без этого дополнения.
|
||
def _window(self):
|
||
total = (len(self.image.data) + 7) // 8
|
||
self._window_end = min(total, self.next_block + 16)
|
||
return [self._frame(10 + self.slot, i, self.image.data[i * 8:(i + 1) * 8].ljust(8, b"\xff"))
|
||
for i in range(self.next_block, self._window_end)]
|
||
|
||
def handle_status(self, frame):
|
||
if not self.accepts(frame):
|
||
return [], ""
|
||
command = ProtoCanId.parse(frame.can_id).body & 255
|
||
expected_command = {"enter": 2, "image": 3, "compat": 4, "erase": 5,
|
||
"data": 0, "verify": 6, "commit": 7, "reboot": 9}.get(self.stage)
|
||
# Позднее подтверждение предыдущего этапа нельзя применять к новому:
|
||
# оно не продвигает автомат и не считается ошибкой текущего запроса.
|
||
if command != expected_command:
|
||
return [], ""
|
||
status, slot, expected = struct.unpack_from("<BBH", frame.data)
|
||
if status == 1:
|
||
return [], "Загрузчик занят"
|
||
if status not in (0, 8) or (status == 8 and self.stage != "data"):
|
||
raise RuntimeError(f"ProtoCAN Boot: ошибка 0x{status:02X}")
|
||
if self.stage == "enter":
|
||
self.stage = "image"
|
||
return self._control(3, struct.pack("<II", len(self.image.data), self.image.crc32)), "Передача метаданных"
|
||
if self.stage == "image":
|
||
self.stage = "compat"
|
||
t = self.target
|
||
return self._control(4, struct.pack("<HBBI", t.product_type, t.hardware_min, t.hardware_max, self.image.version)), "Проверка совместимости"
|
||
if self.stage == "compat":
|
||
if slot not in (0, 1):
|
||
raise RuntimeError("ProtoCAN Boot: неверный слот")
|
||
self.slot, self.stage = slot, "erase"
|
||
return self._control(5), "Стирание неактивного слота"
|
||
if self.stage in ("erase", "data"):
|
||
total = (len(self.image.data) + 7) // 8
|
||
# The final index of a 512 KiB image wraps the u16 status field.
|
||
if self.stage == "data" and expected == 0 and self._window_end == 65536 and status == 0:
|
||
expected = 65536
|
||
limit = total if self.stage == "erase" else self._window_end
|
||
if expected > limit:
|
||
raise RuntimeError("ProtoCAN Boot: подтверждён непереданный блок")
|
||
if self.stage == "data" and status == 0 and expected < self.next_block:
|
||
return [], ""
|
||
self.next_block = expected
|
||
if expected == total:
|
||
self.stage = "verify"
|
||
return self._control(6), "Проверка CRC32"
|
||
self.stage = "data"
|
||
return self._window(), "Передача блоков ProtoCAN Boot"
|
||
if self.stage == "verify":
|
||
self.stage = "commit"
|
||
return self._control(7), "Активация образа"
|
||
if self.stage == "commit":
|
||
self.stage = "reboot"
|
||
return self._control(9), "Перезапуск прибора"
|
||
self.stage, self.finished = "finished", True
|
||
return [], "Прошивка ProtoCAN Boot завершена"
|