126 lines
4.9 KiB
Python
126 lines
4.9 KiB
Python
"""@file firmware.py
|
||
@brief Проверка образов и переносимая машина блочной передачи прошивки.
|
||
|
||
Реализует загрузку и валидацию образов прошивки из различных форматов файлов,
|
||
проверку целостности данных, и логику передачи образов на устройство.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
from set_devices.protocol import crc32_ieee
|
||
|
||
|
||
class FirmwareImageError(ValueError):
|
||
"""@brief Файл нельзя безопасно представить как непрерывный бинарный образ."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class FirmwareImage:
|
||
"""@brief Загруженный образ прошивки с метаданными.
|
||
|
||
@param path Путь к исходному файлу образа
|
||
@param data Бинарные данные образа
|
||
@param base_address Адрес начала загрузки (для Intel HEX формата)
|
||
@param version Номер версии прошивки
|
||
"""
|
||
path: Path
|
||
data: bytes
|
||
base_address: int = 0
|
||
version: int = 0
|
||
|
||
@property
|
||
def crc32(self) -> int:
|
||
return crc32_ieee(self.data)
|
||
|
||
@property
|
||
def sha256(self) -> str:
|
||
return hashlib.sha256(self.data).hexdigest()
|
||
|
||
@classmethod
|
||
def load(cls, path: str | Path, maximum_size: int = 2 * 1024 * 1024) -> "FirmwareImage":
|
||
file_path = Path(path)
|
||
raw = file_path.read_bytes()
|
||
if file_path.suffix.lower() == ".hex":
|
||
data, base = _parse_intel_hex(raw.decode("ascii"))
|
||
elif file_path.suffix.lower() == ".bin":
|
||
data, base = raw, 0
|
||
else:
|
||
raise FirmwareImageError("Поддерживаются только .bin и Intel HEX")
|
||
if not data:
|
||
raise FirmwareImageError("Образ пуст")
|
||
if len(data) > maximum_size:
|
||
raise FirmwareImageError(f"Размер образа превышает {maximum_size} байт")
|
||
return cls(file_path, data, base)
|
||
|
||
|
||
def _parse_intel_hex(text: str) -> tuple[bytes, int]:
|
||
memory: dict[int, int] = {}
|
||
upper = 0
|
||
eof = False
|
||
for line_number, source in enumerate(text.splitlines(), 1):
|
||
line = source.strip()
|
||
if not line:
|
||
continue
|
||
if not line.startswith(":"):
|
||
raise FirmwareImageError(f"Intel HEX: строка {line_number} без ':'")
|
||
try:
|
||
record = bytes.fromhex(line[1:])
|
||
except ValueError as error:
|
||
raise FirmwareImageError(f"Intel HEX: неверный HEX в строке {line_number}") from error
|
||
if len(record) < 5 or len(record) != record[0] + 5 or sum(record) & 0xFF:
|
||
raise FirmwareImageError(f"Intel HEX: неверная длина/CRC строки {line_number}")
|
||
count = record[0]
|
||
address = int.from_bytes(record[1:3], "big")
|
||
kind = record[3]
|
||
payload = record[4 : 4 + count]
|
||
if kind == 0x00:
|
||
absolute = upper + address
|
||
for index, value in enumerate(payload):
|
||
key = absolute + index
|
||
if key in memory and memory[key] != value:
|
||
raise FirmwareImageError("Intel HEX содержит конфликтующие диапазоны")
|
||
memory[key] = value
|
||
elif kind == 0x01:
|
||
eof = True
|
||
break
|
||
elif kind == 0x04 and count == 2:
|
||
upper = int.from_bytes(payload, "big") << 16
|
||
elif kind == 0x02 and count == 2:
|
||
upper = int.from_bytes(payload, "big") << 4
|
||
elif kind not in (0x03, 0x05):
|
||
raise FirmwareImageError(f"Intel HEX: тип записи {kind:#x} не поддержан")
|
||
if not eof or not memory:
|
||
raise FirmwareImageError("Intel HEX не содержит данных или EOF")
|
||
start, end = min(memory), max(memory) + 1
|
||
image = bytearray(b"\xFF" * (end - start))
|
||
for address, value in memory.items():
|
||
image[address - start] = value
|
||
return bytes(image), start
|
||
|
||
|
||
@dataclass
|
||
class FirmwareTransfer:
|
||
image: FirmwareImage
|
||
block_size: int = 480
|
||
offset: int = 0
|
||
|
||
def next_block(self) -> tuple[int, bytes] | None:
|
||
if self.offset >= len(self.image.data):
|
||
return None
|
||
offset = self.offset
|
||
return offset, self.image.data[offset : offset + self.block_size]
|
||
|
||
def acknowledge(self, next_offset: int) -> None:
|
||
current = self.next_block()
|
||
if current is None or next_offset != current[0] + len(current[1]):
|
||
raise FirmwareImageError("МК подтвердил неожиданный offset")
|
||
self.offset = next_offset
|
||
|
||
@property
|
||
def percent(self) -> int:
|
||
return int(self.offset * 100 / len(self.image.data))
|