Add shared waveform processing, generation and firmware image support
This commit is contained in:
31
python/firmware_image/README.md
Normal file
31
python/firmware_image/README.md
Normal file
@@ -0,0 +1,31 @@
|
||||
# Общие образы прошивки
|
||||
|
||||
SETGUI и климатический bridge используют один пакет и один C99-парсер
|
||||
`c/set-protocol/src/firmware_image.c`. Python преобразует результат C в
|
||||
`HexSegment`; алгоритмического fallback нет. Карта Flash, план записи,
|
||||
заполнение промежутков FF и стандартные CRC32/SHA256 доступны через `core`.
|
||||
Запуск программатора, файловый allowlist и журнал остаются в приложении.
|
||||
|
||||
Строгая политика HEX одинакова для обоих потребителей: checksum, обязательный
|
||||
EOF, запрет данных после EOF, запрет любых перекрытий, проверка длины и адресов.
|
||||
Непрерывный образ имеет заданный предел **до** выделения его буфера.
|
||||
Профиль STM32F407VE сохраняется для совместимости; другой MCU передаёт свой
|
||||
`FlashLayout` в функции планирования.
|
||||
|
||||
Из корня templates:
|
||||
|
||||
```powershell
|
||||
python c/set-protocol/tools/build_host.py --output python/protocan/native/setprotocol.dll
|
||||
python -m pip install -e ./python
|
||||
python -m unittest discover -s python/tests -p test_firmware_image_shared.py
|
||||
```
|
||||
|
||||
На Linux/macOS имя результата — `libsetprotocol.so`/`libsetprotocol.dylib`.
|
||||
Также поддерживается явный путь `SETPROTOCOL_LIBRARY`. Старая DLL без API
|
||||
образов выдаёт понятную ошибку с требованием пересборки.
|
||||
|
||||
```python
|
||||
from firmware_image.core import parse_intel_hex, contiguous_image
|
||||
segments = parse_intel_hex(':0100000001FE\n:00000001FF')
|
||||
image, base = contiguous_image(':0100000001FE\n:00000001FF', maximum_size=1024)
|
||||
```
|
||||
2
python/firmware_image/__init__.py
Normal file
2
python/firmware_image/__init__.py
Normal file
@@ -0,0 +1,2 @@
|
||||
"""Shared firmware images and flash layout plans."""
|
||||
from .core import *
|
||||
373
python/firmware_image/core.py
Normal file
373
python/firmware_image/core.py
Normal file
@@ -0,0 +1,373 @@
|
||||
"""Переносимое ядро проверки образа прошивки STM32.
|
||||
|
||||
Модуль не знает ни про HTTP, ни про Modbus, ни про файловую систему, ни про
|
||||
конкретный программатор: на вход подаются уже прочитанные байты, на выходе
|
||||
получается проверенный план записи. Благодаря этому те же правила проверяются
|
||||
host-тестом без платы и без bridge.
|
||||
|
||||
Основные границы:
|
||||
|
||||
* :func:`parse_intel_hex` — полный разбор Intel HEX с проверкой checksum каждой
|
||||
записи, поддержкой расширенной адресации и поиском пересечений;
|
||||
* :func:`build_raw_plan` / :func:`build_hex_plan` — план записи с проверкой
|
||||
выравнивания, границ приложения и защищённых секторов;
|
||||
* :func:`image_digest` — CRC32 и SHA-256 фактически записываемых байтов.
|
||||
|
||||
Геометрия Flash задаётся объектом :class:`FlashLayout`, поэтому ядро не привязано
|
||||
к STM32F407: значения по умолчанию описывают текущую плату и берутся из
|
||||
``MDK-ARM/flash_layout.sct`` и ``EEPROM_Emul/lib/flash_ring.h``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import re
|
||||
import zlib
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Iterable, Optional, Sequence
|
||||
|
||||
__all__ = [
|
||||
"FirmwareImageError",
|
||||
"FlashLayout",
|
||||
"FlashRegion",
|
||||
"FlashSector",
|
||||
"HexSegment",
|
||||
"ImagePlan",
|
||||
"STM32F407VE_LAYOUT",
|
||||
"RAW_EXTENSIONS",
|
||||
"HEX_EXTENSIONS",
|
||||
"build_hex_plan",
|
||||
"build_raw_plan",
|
||||
"detect_format",
|
||||
"format_address",
|
||||
"image_digest",
|
||||
"parse_flash_address",
|
||||
"parse_intel_hex",
|
||||
]
|
||||
|
||||
|
||||
class FirmwareImageError(ValueError):
|
||||
"""Единственный тип ошибки ядра: сообщение пригодно для показа оператору."""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Геометрия Flash
|
||||
# ---------------------------------------------------------------------------
|
||||
@dataclass(frozen=True)
|
||||
class FlashSector:
|
||||
index: int
|
||||
start: int
|
||||
size: int
|
||||
|
||||
@property
|
||||
def end(self) -> int:
|
||||
"""Последний байт сектора включительно."""
|
||||
return self.start + self.size - 1
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FlashRegion:
|
||||
name: str
|
||||
start: int
|
||||
end: int
|
||||
|
||||
def overlaps(self, start: int, end: int) -> bool:
|
||||
return start <= self.end and end >= self.start
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FlashLayout:
|
||||
"""Полное описание внутренней Flash целевого МК."""
|
||||
|
||||
name: str
|
||||
sectors: Sequence[FlashSector]
|
||||
application: FlashRegion
|
||||
reserved: Sequence[FlashRegion] = field(default_factory=tuple)
|
||||
write_alignment: int = 4
|
||||
min_image_bytes: int = 8
|
||||
|
||||
@property
|
||||
def max_image_bytes(self) -> int:
|
||||
return self.application.end - self.application.start + 1
|
||||
|
||||
def sector_at(self, address: int) -> Optional[FlashSector]:
|
||||
for sector in self.sectors:
|
||||
if sector.start <= address <= sector.end:
|
||||
return sector
|
||||
return None
|
||||
|
||||
def sectors_for_range(self, start: int, end: int) -> list[FlashSector]:
|
||||
return [sector for sector in self.sectors if start <= sector.end and end >= sector.start]
|
||||
|
||||
def reserved_hit(self, start: int, end: int) -> Optional[FlashRegion]:
|
||||
for region in self.reserved:
|
||||
if region.overlaps(start, end):
|
||||
return region
|
||||
return None
|
||||
|
||||
|
||||
# Секторы STM32F407VE: 4x16 КиБ, 64 КиБ, 3x128 КиБ.
|
||||
STM32F407VE_LAYOUT = FlashLayout(
|
||||
name="STM32F407VET6",
|
||||
sectors=(
|
||||
FlashSector(0, 0x08000000, 16 * 1024),
|
||||
FlashSector(1, 0x08004000, 16 * 1024),
|
||||
FlashSector(2, 0x08008000, 16 * 1024),
|
||||
FlashSector(3, 0x0800C000, 16 * 1024),
|
||||
FlashSector(4, 0x08010000, 64 * 1024),
|
||||
FlashSector(5, 0x08020000, 128 * 1024),
|
||||
FlashSector(6, 0x08040000, 128 * 1024),
|
||||
FlashSector(7, 0x08060000, 128 * 1024),
|
||||
),
|
||||
application=FlashRegion("Приложение", 0x08000000, 0x0803FFFF),
|
||||
# Секторы 6-7 отданы кольцу EEPROM: их стирание уничтожило бы настройки.
|
||||
reserved=(FlashRegion("Кольцо EEPROM (секторы 6-7)", 0x08040000, 0x0807FFFF),),
|
||||
)
|
||||
|
||||
RAW_EXTENSIONS = (".bin", ".fw")
|
||||
HEX_EXTENSIONS = (".hex",)
|
||||
|
||||
|
||||
def detect_format(name: str) -> str:
|
||||
"""Формат образа по расширению; исключение для всего остального."""
|
||||
lower = str(name or "").lower()
|
||||
if lower.endswith(RAW_EXTENSIONS):
|
||||
return "bin"
|
||||
if lower.endswith(HEX_EXTENSIONS):
|
||||
return "hex"
|
||||
raise FirmwareImageError("Поддерживаются только .bin/.hex/.fw")
|
||||
|
||||
|
||||
def format_address(address: int) -> str:
|
||||
return f"0x{int(address):08X}"
|
||||
|
||||
|
||||
def parse_flash_address(value: object) -> int:
|
||||
"""Адрес принимается только как явный decimal или HEX с префиксом 0x."""
|
||||
text = str(value if value is not None else "").strip().replace(" ", "")
|
||||
if not text:
|
||||
raise FirmwareImageError("Базовый адрес Flash не задан")
|
||||
if not re.fullmatch(r"0[xX][0-9a-fA-F]{1,8}|\d+", text):
|
||||
raise FirmwareImageError("Адрес должен быть decimal или HEX с префиксом 0x")
|
||||
address = int(text, 16 if text.lower().startswith("0x") else 10)
|
||||
if not 0 <= address <= 0xFFFFFFFF:
|
||||
raise FirmwareImageError("Адрес выходит за диапазон uint32")
|
||||
return address
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Intel HEX
|
||||
# ---------------------------------------------------------------------------
|
||||
@dataclass(frozen=True)
|
||||
class HexSegment:
|
||||
"""Непрерывный участок данных Intel HEX."""
|
||||
|
||||
start: int
|
||||
data: bytes
|
||||
|
||||
@property
|
||||
def end(self) -> int:
|
||||
return self.start + len(self.data) - 1
|
||||
|
||||
|
||||
RECORD_DATA = 0x00
|
||||
RECORD_EOF = 0x01
|
||||
RECORD_EXTENDED_SEGMENT = 0x02
|
||||
RECORD_START_SEGMENT = 0x03
|
||||
RECORD_EXTENDED_LINEAR = 0x04
|
||||
RECORD_START_LINEAR = 0x05
|
||||
_KNOWN_RECORDS = (RECORD_DATA, RECORD_EOF, RECORD_EXTENDED_SEGMENT,
|
||||
RECORD_START_SEGMENT, RECORD_EXTENDED_LINEAR, RECORD_START_LINEAR)
|
||||
|
||||
# Лимит защищает bridge от файла, который «раздувается» адресными записями.
|
||||
MAX_HEX_RECORDS = 1_000_000
|
||||
|
||||
|
||||
def parse_intel_hex(text: object, *, max_records: int = MAX_HEX_RECORDS) -> list[HexSegment]:
|
||||
"""Validate HEX with the shared C99 parser and return contiguous segments."""
|
||||
from .native import parse_segments
|
||||
return parse_segments(text, max_records, FirmwareImageError, HexSegment)
|
||||
|
||||
|
||||
def contiguous_image(text, maximum_size=2 * 1024 * 1024):
|
||||
"""Convert validated segments to a bounded image, filling gaps with FF."""
|
||||
segments = parse_intel_hex(text)
|
||||
base, end = segments[0].start, segments[-1].end + 1
|
||||
if end - base > maximum_size:
|
||||
raise FirmwareImageError('Размер образа превышает %d байт' % maximum_size)
|
||||
image = bytearray(b'\xff' * (end - base))
|
||||
for segment in segments:
|
||||
offset = segment.start - base
|
||||
image[offset:offset + len(segment.data)] = segment.data
|
||||
return bytes(image), base
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ImagePlan:
|
||||
"""Проверенный план: что именно и куда будет записано."""
|
||||
|
||||
image_format: str
|
||||
start: int
|
||||
end: int
|
||||
payload_bytes: int
|
||||
sectors: tuple[int, ...]
|
||||
erased_bytes: int
|
||||
segments: tuple[HexSegment, ...]
|
||||
gaps: tuple[tuple[int, int], ...]
|
||||
crc32: str
|
||||
sha256: str
|
||||
|
||||
@property
|
||||
def text(self) -> str:
|
||||
return f"{format_address(self.start)}..{format_address(self.end)}"
|
||||
|
||||
def as_dict(self) -> dict:
|
||||
"""Представление для JSON API и GUI."""
|
||||
return {
|
||||
"format": self.image_format,
|
||||
"start": format_address(self.start),
|
||||
"end": format_address(self.end),
|
||||
"targetRange": self.text,
|
||||
"payloadBytes": self.payload_bytes,
|
||||
"spanBytes": self.end - self.start + 1,
|
||||
"sectors": list(self.sectors),
|
||||
"erasedBytes": self.erased_bytes,
|
||||
"segments": [
|
||||
{"start": format_address(item.start), "end": format_address(item.end),
|
||||
"size": len(item.data)}
|
||||
for item in self.segments
|
||||
],
|
||||
"gaps": [
|
||||
{"start": format_address(low), "end": format_address(high), "size": high - low + 1}
|
||||
for low, high in self.gaps
|
||||
],
|
||||
"crc32": self.crc32,
|
||||
"sha256": self.sha256,
|
||||
}
|
||||
|
||||
|
||||
def image_digest(data: bytes) -> tuple[str, str]:
|
||||
"""CRC32 и SHA-256 фактически записываемых байтов."""
|
||||
return (
|
||||
f"{zlib.crc32(bytes(data)) & 0xFFFFFFFF:08X}",
|
||||
hashlib.sha256(bytes(data)).hexdigest().upper(),
|
||||
)
|
||||
|
||||
|
||||
def _check_range(layout: FlashLayout, start: int, end: int) -> None:
|
||||
reserved = layout.reserved_hit(start, end)
|
||||
if reserved is not None:
|
||||
raise FirmwareImageError(f"Диапазон пересекает защищённую область: {reserved.name}")
|
||||
if start < layout.application.start or end > layout.application.end:
|
||||
raise FirmwareImageError(
|
||||
"Диапазон должен помещаться в "
|
||||
f"{format_address(layout.application.start)}.."
|
||||
f"{format_address(layout.application.end)}"
|
||||
)
|
||||
|
||||
|
||||
def _sector_summary(layout: FlashLayout, start: int, end: int) -> tuple[tuple[int, ...], int]:
|
||||
sectors = layout.sectors_for_range(start, end)
|
||||
return tuple(item.index for item in sectors), sum(item.size for item in sectors)
|
||||
|
||||
|
||||
def build_raw_plan(size: object, base_address: object, *,
|
||||
layout: FlashLayout = STM32F407VE_LAYOUT,
|
||||
data: Optional[bytes] = None) -> ImagePlan:
|
||||
"""План для сырого образа.
|
||||
|
||||
Базовый адрес обязан совпадать с началом сектора: стирание STM32F4 идёт
|
||||
только целыми секторами, поэтому запись с середины сектора уничтожила бы
|
||||
соседние данные. ``data`` необязателен: список файлов SD знает лишь размер,
|
||||
и контрольные суммы считаются только тогда, когда содержимое действительно
|
||||
прочитано.
|
||||
"""
|
||||
try:
|
||||
length = int(size)
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise FirmwareImageError("Размер образа должен быть целым числом") from exc
|
||||
if data is not None and len(data) != length:
|
||||
raise FirmwareImageError("Размер образа не совпадает с числом прочитанных байт")
|
||||
if length < layout.min_image_bytes:
|
||||
raise FirmwareImageError(f"Размер образа меньше {layout.min_image_bytes} байт")
|
||||
if length > layout.max_image_bytes:
|
||||
raise FirmwareImageError(
|
||||
f"Образ больше области приложения ({layout.max_image_bytes // 1024} КиБ)"
|
||||
)
|
||||
start = parse_flash_address(base_address)
|
||||
if start % layout.write_alignment:
|
||||
raise FirmwareImageError(
|
||||
f"Базовый адрес должен быть выровнен на {layout.write_alignment} байта"
|
||||
)
|
||||
sector = layout.sector_at(start)
|
||||
if sector is None:
|
||||
raise FirmwareImageError(f"Базовый адрес вне внутренней Flash {layout.name}")
|
||||
if sector.start != start:
|
||||
raise FirmwareImageError(
|
||||
f"Базовый адрес должен совпадать с началом сектора ({format_address(sector.start)})"
|
||||
)
|
||||
end = start + length - 1
|
||||
_check_range(layout, start, end)
|
||||
sectors, erased = _sector_summary(layout, start, end)
|
||||
crc32, sha256 = image_digest(data) if data is not None else ("", "")
|
||||
segments = (HexSegment(start, bytes(data)),) if data is not None else ()
|
||||
return ImagePlan("bin", start, end, length, sectors, erased, segments, (), crc32, sha256)
|
||||
|
||||
|
||||
def build_hex_plan(segments: Iterable[HexSegment], *,
|
||||
layout: FlashLayout = STM32F407VE_LAYOUT) -> ImagePlan:
|
||||
"""План для Intel HEX: адреса берутся из файла, оператор их не задаёт.
|
||||
|
||||
Проверяется каждый сегмент отдельно, поэтому файл с одним «правильным» и
|
||||
одним посторонним диапазоном отклоняется целиком. Разрывы между сегментами
|
||||
сохраняются в плане: при записи они заполняются 0xFF, и оператор должен
|
||||
видеть, какая часть Flash будет стёрта без данных из файла.
|
||||
"""
|
||||
ordered = sorted(segments, key=lambda item: item.start)
|
||||
if not ordered:
|
||||
raise FirmwareImageError("Файл Intel HEX не содержит данных")
|
||||
for segment in ordered:
|
||||
if not segment.data:
|
||||
raise FirmwareImageError("Сегмент Intel HEX пуст")
|
||||
_check_range(layout, segment.start, segment.end)
|
||||
start = ordered[0].start
|
||||
end = ordered[-1].end
|
||||
if start % layout.write_alignment:
|
||||
raise FirmwareImageError(
|
||||
f"Начальный адрес Intel HEX не выровнен на {layout.write_alignment} байта"
|
||||
)
|
||||
sector = layout.sector_at(start)
|
||||
if sector is None or sector.start != start:
|
||||
raise FirmwareImageError(
|
||||
"Первый адрес Intel HEX должен совпадать с началом сектора "
|
||||
f"({format_address(sector.start) if sector else 'вне Flash'})"
|
||||
)
|
||||
# Пустоты заполняются 0xFF — стёртым состоянием NOR Flash.
|
||||
image = bytearray(b"\xFF" * (end - start + 1))
|
||||
gaps: list[tuple[int, int]] = []
|
||||
previous_end = None
|
||||
payload = 0
|
||||
for segment in ordered:
|
||||
if previous_end is not None and segment.start > previous_end + 1:
|
||||
gaps.append((previous_end + 1, segment.start - 1))
|
||||
image[segment.start - start:segment.end - start + 1] = segment.data
|
||||
payload += len(segment.data)
|
||||
previous_end = segment.end
|
||||
sectors, erased = _sector_summary(layout, start, end)
|
||||
crc32, sha256 = image_digest(bytes(image))
|
||||
return ImagePlan("hex", start, end, payload, sectors, erased,
|
||||
tuple(ordered), tuple(gaps), crc32, sha256)
|
||||
|
||||
|
||||
def build_plan(image_format: str, *, data: Optional[bytes] = None, size: Optional[int] = None,
|
||||
base_address: object = None,
|
||||
layout: FlashLayout = STM32F407VE_LAYOUT) -> ImagePlan:
|
||||
"""Единая точка входа: выбрать ветку разбора по формату образа."""
|
||||
if image_format == "hex":
|
||||
if data is None:
|
||||
raise FirmwareImageError("Для Intel HEX требуется содержимое файла")
|
||||
return build_hex_plan(parse_intel_hex(data), layout=layout)
|
||||
if image_format == "bin":
|
||||
length = len(data) if data is not None else size
|
||||
return build_raw_plan(length, base_address, layout=layout, data=data)
|
||||
raise FirmwareImageError("Неизвестный формат образа")
|
||||
48
python/firmware_image/native.py
Normal file
48
python/firmware_image/native.py
Normal file
@@ -0,0 +1,48 @@
|
||||
"""Models for the C99 Intel HEX parser. No Python parser fallback."""
|
||||
import ctypes as ct
|
||||
from functools import lru_cache
|
||||
from protocan.native import _load_library
|
||||
|
||||
class Cell(ct.Structure):
|
||||
_fields_ = [('address', ct.c_uint32), ('value', ct.c_uint8)]
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def library():
|
||||
lib = _load_library()
|
||||
try:
|
||||
lib.firmware_hex_parse.argtypes = [ct.c_char_p, ct.c_size_t, ct.POINTER(Cell),
|
||||
ct.c_size_t, ct.c_size_t, ct.POINTER(ct.c_size_t), ct.POINTER(ct.c_size_t)]
|
||||
lib.firmware_hex_parse.restype = ct.c_int
|
||||
lib.firmware_hex_segment_size.argtypes = [ct.POINTER(Cell), ct.c_size_t, ct.c_size_t]
|
||||
lib.firmware_hex_segment_size.restype = ct.c_size_t
|
||||
except AttributeError as exc:
|
||||
raise RuntimeError('Rebuild SETProtocol: native firmware image API is missing') from exc
|
||||
return lib
|
||||
|
||||
def parse_segments(text, max_records, error_type, segment_type):
|
||||
try:
|
||||
raw = bytes(text) if isinstance(text, (bytes, bytearray)) else str(text or '').encode('ascii')
|
||||
except UnicodeError as exc:
|
||||
raise error_type('Intel HEX должен состоять из ASCII') from exc
|
||||
if max_records < 1 or len(raw) > 64 * 1024 * 1024:
|
||||
raise error_type('Файл Intel HEX превышает допустимый размер')
|
||||
cells = (Cell * max(1, len(raw) // 2))()
|
||||
count, line = ct.c_size_t(), ct.c_size_t()
|
||||
lib = library()
|
||||
status = lib.firmware_hex_parse(raw, len(raw), cells, len(cells), max_records,
|
||||
ct.byref(count), ct.byref(line))
|
||||
if status:
|
||||
messages = {1: 'неверный формат', 2: 'неверная checksum записи',
|
||||
3: 'неизвестный тип или неверная длина записи', 4: 'нет данных/EOF или данные после записи EOF',
|
||||
5: 'адрес описан повторно', 6: 'адрес выходит за uint32',
|
||||
7: 'превышено допустимое число записей',
|
||||
8: "запись должна начинаться с ':'", 9: 'запись содержит не пары HEX-символов',
|
||||
10: 'длина записи не совпадает с полем count'}
|
||||
raise error_type('Intel HEX, строка %d: %s' % (line.value, messages.get(status, 'ошибка')))
|
||||
result, offset = [], 0
|
||||
while offset < count.value:
|
||||
size = lib.firmware_hex_segment_size(cells, count.value, offset)
|
||||
result.append(segment_type(cells[offset].address,
|
||||
bytes(cells[i].value for i in range(offset, offset + size))))
|
||||
offset += size
|
||||
return result
|
||||
Reference in New Issue
Block a user