update: firmware and provisioning now supports bootloader and partition tables

This commit is contained in:
2026-03-16 08:52:58 +02:00
parent 360725c93f
commit 4381a6681d
15 changed files with 776 additions and 49 deletions

View File

@@ -15,7 +15,7 @@ class BoardType(str, Enum):
BOARD_TYPE_LABELS = {
"vesper": "Vesper",
"vesper_plus": "Vesper+",
"vesper_plus": "Vesper Plus",
"vesper_pro": "Vesper Pro",
"chronos": "Chronos",
"chronos_pro": "Chronos Pro",
@@ -23,6 +23,28 @@ BOARD_TYPE_LABELS = {
"agnus": "Agnus",
}
# Family codes (BS + 4 chars = segment 1 of serial number)
BOARD_FAMILY_CODES = {
"vesper": "VSPR",
"vesper_plus": "VSPR",
"vesper_pro": "VSPR",
"agnus": "AGNS",
"agnus_mini": "AGNS",
"chronos": "CRNS",
"chronos_pro": "CRNS",
}
# Variant codes (3 chars = first part of segment 3 of serial number)
BOARD_VARIANT_CODES = {
"vesper": "STD",
"vesper_plus": "PLS",
"vesper_pro": "PRO",
"agnus": "STD",
"agnus_mini": "MIN",
"chronos": "STD",
"chronos_pro": "PRO",
}
class MfgStatus(str, Enum):
manufactured = "manufactured"

View File

@@ -1,4 +1,4 @@
from fastapi import APIRouter, Depends, Query, HTTPException
from fastapi import APIRouter, Depends, Query, HTTPException, UploadFile, File, Form
from fastapi.responses import Response
from fastapi.responses import RedirectResponse
from typing import Optional
@@ -15,6 +15,9 @@ from manufacturing import service
from manufacturing import audit
from shared.exceptions import NotFoundError
VALID_FLASH_ASSETS = {"bootloader.bin", "partitions.bin"}
VALID_HW_TYPES_MFG = {"vesper", "vesper_plus", "vesper_pro", "agnus", "agnus_mini", "chronos", "chronos_pro"}
router = APIRouter(prefix="/api/manufacturing", tags=["manufacturing"])
@@ -175,3 +178,68 @@ def redirect_firmware(
"""
url = service.get_firmware_url(sn)
return RedirectResponse(url=url, status_code=302)
# ─────────────────────────────────────────────────────────────────────────────
# Flash assets — bootloader.bin and partitions.bin per hw_type
# These are the binaries that must be flashed at fixed addresses during full
# provisioning (0x1000 bootloader, 0x8000 partition table).
# They are NOT flashed during OTA updates — only during initial provisioning.
# Upload once per hw_type after each PlatformIO build that changes the layout.
# ─────────────────────────────────────────────────────────────────────────────
@router.post("/flash-assets/{hw_type}/{asset}", status_code=204)
async def upload_flash_asset(
hw_type: str,
asset: str,
file: UploadFile = File(...),
_user: TokenPayload = Depends(require_permission("manufacturing", "add")),
):
"""Upload a bootloader.bin or partitions.bin for a given hw_type.
These are build artifacts from PlatformIO (.pio/build/{env}/bootloader.bin
and .pio/build/{env}/partitions.bin). Upload them once per hw_type after
each PlatformIO build that changes the partition layout.
"""
if hw_type not in VALID_HW_TYPES_MFG:
raise HTTPException(status_code=400, detail=f"Invalid hw_type. Must be one of: {', '.join(sorted(VALID_HW_TYPES_MFG))}")
if asset not in VALID_FLASH_ASSETS:
raise HTTPException(status_code=400, detail=f"Invalid asset. Must be one of: {', '.join(sorted(VALID_FLASH_ASSETS))}")
data = await file.read()
service.save_flash_asset(hw_type, asset, data)
@router.get("/devices/{sn}/bootloader.bin")
def download_bootloader(
sn: str,
_user: TokenPayload = Depends(require_permission("manufacturing", "view")),
):
"""Return the bootloader.bin for this device's hw_type (flashed at 0x1000)."""
item = service.get_device_by_sn(sn)
try:
data = service.get_flash_asset(item.hw_type, "bootloader.bin")
except NotFoundError as e:
raise HTTPException(status_code=404, detail=str(e))
return Response(
content=data,
media_type="application/octet-stream",
headers={"Content-Disposition": f'attachment; filename="bootloader_{item.hw_type}.bin"'},
)
@router.get("/devices/{sn}/partitions.bin")
def download_partitions(
sn: str,
_user: TokenPayload = Depends(require_permission("manufacturing", "view")),
):
"""Return the partitions.bin for this device's hw_type (flashed at 0x8000)."""
item = service.get_device_by_sn(sn)
try:
data = service.get_flash_asset(item.hw_type, "partitions.bin")
except NotFoundError as e:
raise HTTPException(status_code=404, detail=str(e))
return Response(
content=data,
media_type="application/octet-stream",
headers={"Content-Disposition": f'attachment; filename="partitions_{item.hw_type}.bin"'},
)

View File

@@ -2,9 +2,11 @@ import logging
import random
import string
from datetime import datetime, timezone
from pathlib import Path
logger = logging.getLogger(__name__)
from config import settings
from shared.firebase import get_db
from shared.exceptions import NotFoundError
from utils.serial_number import generate_serial
@@ -270,6 +272,29 @@ def delete_unprovisioned_devices() -> list[str]:
return deleted
def _flash_asset_path(hw_type: str, asset: str) -> Path:
"""Return path to a flash asset (bootloader.bin or partitions.bin) for a given hw_type."""
return Path(settings.flash_assets_storage_path) / hw_type / asset
def save_flash_asset(hw_type: str, asset: str, data: bytes) -> Path:
"""Persist a flash asset binary. asset must be 'bootloader.bin' or 'partitions.bin'."""
if asset not in ("bootloader.bin", "partitions.bin"):
raise ValueError(f"Unknown flash asset: {asset}")
path = _flash_asset_path(hw_type, asset)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
return path
def get_flash_asset(hw_type: str, asset: str) -> bytes:
"""Load a flash asset binary. Raises NotFoundError if not uploaded yet."""
path = _flash_asset_path(hw_type, asset)
if not path.exists():
raise NotFoundError(f"Flash asset '{asset}' for hw_type '{hw_type}' — upload it first via POST /api/manufacturing/flash-assets/{{hw_type}}/{{asset}}")
return path.read_bytes()
def get_firmware_url(sn: str) -> str:
"""Return the FastAPI download URL for the latest stable firmware for this device's hw_type."""
from firmware.service import get_latest