146 lines
4.8 KiB
Python
146 lines
4.8 KiB
Python
"""Reusable helpers for publishing the shared firmware release catalog."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from .firmware_catalog import (
|
|
MAX_MANIFEST_BYTES,
|
|
SUPPORTED_TRANSPORTS,
|
|
parse_firmware_catalog,
|
|
)
|
|
|
|
MAX_FIRMWARE_BYTES = 128 * 1024 * 1024
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FirmwarePublication:
|
|
"""Metadata required to publish one firmware image."""
|
|
|
|
path: Path
|
|
product: str
|
|
version_name: str
|
|
version_code: int
|
|
transport: str
|
|
base_address: int | None = None
|
|
notes: str = ""
|
|
|
|
def validate(self) -> None:
|
|
if not self.path.is_file():
|
|
raise ValueError(f"Firmware file is missing: {self.path}")
|
|
if self.path.suffix.lower() not in {".bin", ".hex"}:
|
|
raise ValueError("Firmware file must have a .bin or .hex extension")
|
|
size = self.path.stat().st_size
|
|
if size <= 0:
|
|
raise ValueError("Firmware file is empty")
|
|
if size > MAX_FIRMWARE_BYTES:
|
|
raise ValueError("Firmware file exceeds the maximum size")
|
|
if not self.product.strip():
|
|
raise ValueError("Firmware product is empty")
|
|
if not self.version_name.strip():
|
|
raise ValueError("Firmware version name is empty")
|
|
if not 0 <= self.version_code <= 0x7FFFFFFF:
|
|
raise ValueError(
|
|
"Firmware version code must be between 0 and 2147483647"
|
|
)
|
|
if self.transport not in SUPPORTED_TRANSPORTS:
|
|
raise ValueError("Unsupported firmware transport")
|
|
if (
|
|
self.base_address is not None
|
|
and not 0 <= self.base_address <= 0xFFFFFFFF
|
|
):
|
|
raise ValueError("Firmware base address is outside the uint32 range")
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as source:
|
|
for chunk in iter(lambda: source.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def safe_release_tag_part(value: str) -> str:
|
|
source = value.strip()
|
|
part = re.sub(r"[^A-Za-z0-9._-]+", "-", source).strip("-.")
|
|
digest = hashlib.sha256(source.encode("utf-8")).hexdigest()[:8]
|
|
if not part:
|
|
return digest
|
|
return part if part == source else f"{part}-{digest}"
|
|
|
|
|
|
def firmware_release_tag(publication: FirmwarePublication) -> str:
|
|
return "firmware-%s-v%s" % (
|
|
safe_release_tag_part(publication.product),
|
|
safe_release_tag_part(publication.version_name),
|
|
)
|
|
|
|
|
|
def firmware_release_entry(
|
|
publication: FirmwarePublication, image_url: str, sha256: str
|
|
) -> dict:
|
|
result = {
|
|
"product": publication.product.strip(),
|
|
"versionCode": publication.version_code,
|
|
"versionName": publication.version_name.strip(),
|
|
"imageUrl": image_url,
|
|
"fileName": publication.path.name,
|
|
"sha256": sha256,
|
|
"transport": publication.transport,
|
|
"notes": publication.notes.strip(),
|
|
}
|
|
if publication.base_address is not None:
|
|
result["baseAddress"] = f"0x{publication.base_address:08X}"
|
|
return result
|
|
|
|
|
|
def firmware_entry_identity(entry: dict) -> tuple[str, int, str]:
|
|
return (
|
|
str(entry.get("product", entry.get("device", ""))).strip().casefold(),
|
|
int(entry.get("versionCode", 0)),
|
|
str(entry.get("transport", "rs485")).strip().lower(),
|
|
)
|
|
|
|
|
|
def update_firmware_manifest(manifest: dict, entry: dict) -> dict:
|
|
"""Insert or replace one release without disturbing other manifest data."""
|
|
result = dict(manifest)
|
|
existing = manifest.get("firmware")
|
|
firmware = dict(existing) if isinstance(existing, dict) else {}
|
|
rows = firmware.get("releases") if isinstance(existing, dict) else existing
|
|
releases = (
|
|
[dict(row) for row in rows if isinstance(row, dict)]
|
|
if isinstance(rows, list)
|
|
else []
|
|
)
|
|
identity = firmware_entry_identity(entry)
|
|
releases = [
|
|
row for row in releases if firmware_entry_identity(row) != identity
|
|
]
|
|
releases.append(dict(entry))
|
|
releases.sort(
|
|
key=lambda row: (
|
|
str(row.get("product", row.get("device", ""))).casefold(),
|
|
-int(row.get("versionCode", 0)),
|
|
str(row.get("transport", "rs485")),
|
|
)
|
|
)
|
|
previous_rows = firmware.get("releases")
|
|
changed = releases != previous_rows
|
|
firmware["catalogVersion"] = (
|
|
int(firmware.get("catalogVersion", 0)) + int(changed)
|
|
)
|
|
firmware["releases"] = releases
|
|
result["firmware"] = firmware
|
|
|
|
encoded = (json.dumps(result, ensure_ascii=False) + "\n").encode("utf-8")
|
|
if len(encoded) > MAX_MANIFEST_BYTES:
|
|
raise ValueError("Updated update.json exceeds the maximum size")
|
|
parse_firmware_catalog(encoded, "https://catalog.invalid/update.json")
|
|
return result
|
|
|