Checkpoint EEPROM, firmware analysis and instrument protocol updates

python/set_devices/firmware_identity.py

@@ -0,0 +1,110 @@
+"""Firmware-info v1 and the adjacent release passport (no device I/O)."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import hashlib
+import json
+from pathlib import Path
+
+from protocan.periph28335 import with_crc
+
+MAX_IMAGE_SIZE = 199993
+
+REGISTER_ADDRESS = 0x1200
+REGISTER_COUNT = 12
+
+
+def read_request(controller: int) -> bytes:
+    if not 1 <= controller <= 255:
+        raise ValueError("Для чтения версии нужен адрес контроллера 1..255")
+    return with_crc(bytes((controller, 3)) + REGISTER_ADDRESS.to_bytes(2, 'big')
+                    + REGISTER_COUNT.to_bytes(2, 'big'))
+
+
+@dataclass(frozen=True)
+class FirmwareIdentity:
+    version: tuple[int, int, int]
+    built_at: datetime
+    build_id: str
+
+    def __post_init__(self):
+        if (len(self.version) != 3 or any(type(v) is not int or not 0 <= v <= limit
+                for v, limit in zip(self.version, (255, 255, 999)))
+                or not 2000 <= self.built_at.year <= 2199
+                or self.built_at.utcoffset() is None
+                or self.built_at.microsecond
+                or len(self.build_id) != 8
+                or any(not 32 <= ord(c) <= 126 for c in self.build_id)):
+            raise ValueError("Некорректные данные версии прошивки")
+
+    def describe(self):
+        version = '.'.join(map(str, self.version))
+        return f"{version} · {self.built_at:%Y-%m-%d %H:%M:%S} UTC · Build ID {self.build_id}"
+
+
+def decode_identity(words):
+    if len(words) != REGISTER_COUNT or words[0] != 1:
+        raise ValueError("Неизвестный формат версии (требуется firmware-info v1)")
+    if any(type(w) is not int or not 0 <= w <= 65535 for w in words):
+        raise ValueError("Некорректные регистры версии")
+    try:
+        stamp = datetime(words[4], words[5] >> 8, words[5] & 255,
+                         words[6] >> 8, words[6] & 255, words[7], tzinfo=timezone.utc)
+        build_id = bytes(b for w in words[8:] for b in (w >> 8, w & 255)).decode('ascii')
+        return FirmwareIdentity(tuple(words[1:4]), stamp, build_id)
+    except (ValueError, UnicodeError) as error:
+        raise ValueError("Некорректные данные версии: " + str(error)) from error
+
+
+def inspect_image(path: Path, data: bytes | None = None, *, expected_eeprom=None,
+                  expected_product="Balsam-167-PM35", expected_target="TMS320F28335",
+                  max_image_size=MAX_IMAGE_SIZE):
+    """Return None for legacy images; reject a present but invalid passport.
+
+    The caller may supply the exact bytes it will upload, avoiding a second
+    read between hash verification and transmission.
+    """
+    if data is None:
+        with path.open('rb') as stream:
+            data = stream.read(max_image_size + 1)
+    if not 0 < len(data) <= max_image_size:
+        raise ValueError(f"Размер бинарного файла должен быть 1..{max_image_size} байт")
+    passport = Path(str(path) + '.manifest.json')
+    if not passport.exists():
+        return None
+    try:
+        with passport.open('r', encoding='utf-8-sig') as stream:
+            raw = stream.read(65537)
+        if len(raw) > 65536:
+            raise ValueError("Слишком большой паспорт прошивки")
+        meta = json.loads(raw)
+        if (meta['schema'] != 1 or meta['identity_contract'] != 1
+                or meta['identity_register_count'] != REGISTER_COUNT):
+            raise ValueError("Неизвестный формат паспорта")
+        if meta['product'] != expected_product or meta['target'] != expected_target:
+            raise ValueError(f"Паспорт предназначен для другого устройства (ожидается {expected_product} / {expected_target})")
+        if (type(meta['size_bytes']) is not int or meta['size_bytes'] != len(data)
+                or meta['sha256'] != hashlib.sha256(data).hexdigest()):
+            raise ValueError("BIN не соответствует паспорту: размер или SHA-256 не совпадает")
+        if expected_eeprom is not None and meta.get('eeprom') is not None:
+            if meta['eeprom'] != expected_eeprom:
+                raise ValueError(f"BIN для {meta['eeprom']}, а выбрана EEPROM {expected_eeprom}")
+        stamp = datetime.fromisoformat(meta['built_at_utc'].replace('Z', '+00:00'))
+        if stamp.utcoffset() is None or stamp.utcoffset().total_seconds() != 0:
+            raise ValueError("Дата паспорта должна быть в UTC")
+        return FirmwareIdentity(tuple(meta['version']), stamp, meta['build_id'])
+    except (KeyError, TypeError, ValueError, AttributeError) as error:
+        raise ValueError("Ошибка паспорта: " + str(error)) from error
+
+
+def compare_identity(device, image):
+    if device is None or image is None:
+        return "Сравнение недоступно: нужны версия с платы и проверенный паспорт BIN."
+    if device == image:
+        return "Версия, дата сборки и Build ID совпадают. Содержимое памяти платы этим не проверяется."
+    if image.version > device.version:
+        return "Выбранная версия новее установленной."
+    if image.version < device.version:
+        return "Выбранная версия старее установленной (понижение версии)."
+    return "Номер версии совпадает, но дата сборки или Build ID отличаются."