Firmware CI releases + Gitea auto-update in the server

New Gitea Actions workflow builds both board variants and publishes them as release assets whenever firmware/version.txt is bumped. The server can now poll that repo's releases (next to the existing manual upload) and either surface an "Update frame" button or, with "Automatically apply updates" checked, stage the new build itself -- the frame still only updates on its own next wake either way.
2026-07-21 19:25:08 -04:00
parent 4cf75b4b11
commit 94d67f7767
8 changed files with 493 additions and 42 deletions
@@ -16,8 +16,9 @@ from fastapi.templating import Jinja2Templates
 from PIL import Image
 from pydantic import BaseModel

-from . import config, photo_queue
+from . import config, gitea_releases, photo_queue
 from .face_labels import compute_face_labels
+from .firmware import firmware_path, parse_app_version
 from .image_pipeline import render_frame
 from .immich_client import ImmichClient

@@ -274,11 +275,15 @@ def api_config_save(
    quiet_hours_start: str = Form("22:00"),
    quiet_hours_end: str = Form("07:00"),
    timezone: str = Form("UTC"),
+    firmware_update_repo_url: str = Form(""),
+    firmware_board_variant: str = Form("xiao"),
+    firmware_auto_update: bool = Form(False),
 ):
-    # Immich URL/API key are env-var only (IMMICH_URL/IMMICH_API_KEY, see
-    # docker-compose.yml.example) -- config.load() already applies them,
-    # and this handler doesn't touch cfg.immich_url/immich_api_key at all,
-    # so there's nothing here that could overwrite or clear them.
+    # Immich URL/API key/Gitea token are env-var only (IMMICH_URL/
+    # IMMICH_API_KEY/GITEA_FIRMWARE_TOKEN, see docker-compose.yml.example)
+    # -- config.load() already applies them, and this handler doesn't touch
+    # cfg.immich_url/immich_api_key/firmware_update_token at all, so
+    # there's nothing here that could overwrite or clear them.
    with config.locked():
        cfg = config.load()
        if album_id != cfg.album_id:
@@ -303,6 +308,9 @@ def api_config_save(
            cfg.quiet_hours_end = quiet_hours_end
        if timezone in ALL_TIMEZONES:
            cfg.timezone = timezone
+        cfg.firmware_update_repo_url = firmware_update_repo_url.strip()
+        cfg.firmware_board_variant = firmware_board_variant if firmware_board_variant in ("xiao", "devkit") else "xiao"
+        cfg.firmware_auto_update = firmware_auto_update
        cfg.stats.config_saves += 1
        config.save(cfg)
    return {"status": "saved"}
@@ -448,36 +456,6 @@ def frame_battery(body: BatteryReport):
    return {"status": "saved"}


-# ESP-IDF app images embed an esp_app_desc_t at byte offset 32 (24-byte
-# image header + 8-byte first-segment header): magic word, then version
-# (32 bytes, NUL-padded) at +16 and project name (32 bytes) at +48 --
-# verified against this project's real build artifact.
-APP_DESC_OFFSET = 32
-APP_DESC_MAGIC = 0xABCD5432
-EXPECTED_PROJECT_NAME = "espresso_frame"
-
-
-def _firmware_path():
-    return config.CONFIG_PATH.parent / "firmware.bin"
-
-
-def _parse_app_version(data: bytes) -> str:
-    """Extracts the embedded version from an ESP-IDF app image, raising
-    HTTPException(400) for anything that isn't this project's firmware."""
-    if len(data) < APP_DESC_OFFSET + 80:
-        raise HTTPException(400, "File is too small to be a firmware image")
-    magic = int.from_bytes(data[APP_DESC_OFFSET : APP_DESC_OFFSET + 4], "little")
-    if magic != APP_DESC_MAGIC:
-        raise HTTPException(400, "Not an ESP-IDF application image")
-    version = data[APP_DESC_OFFSET + 16 : APP_DESC_OFFSET + 48].split(b"\x00")[0].decode(errors="replace")
-    project = data[APP_DESC_OFFSET + 48 : APP_DESC_OFFSET + 80].split(b"\x00")[0].decode(errors="replace")
-    if project != EXPECTED_PROJECT_NAME:
-        raise HTTPException(400, f"Image is for project '{project}', not '{EXPECTED_PROJECT_NAME}'")
-    if not version:
-        raise HTTPException(400, "Image has no embedded version")
-    return version
-
-
@app.post("/api/firmware", dependencies=[Depends(require_access_token)])
 def api_firmware_upload(file: UploadFile = File(...)):
    """Uploads a firmware image for OTA. The version is parsed out of the
@@ -485,8 +463,8 @@ def api_firmware_upload(file: UploadFile = File(...)):
    form field, and the project name is checked so an unrelated .bin
    can't be pushed to the frame by mistake."""
    data = file.file.read()
-    version = _parse_app_version(data)
-    _firmware_path().write_bytes(data)
+    version = parse_app_version(data)
+    firmware_path().write_bytes(data)
    with config.locked():
        cfg = config.load()
        cfg.firmware_available_version = version
@@ -499,12 +477,100 @@ def frame_firmware():
    """The uploaded OTA image, streamed to the device (esp_https_ota).
    404 until something has been uploaded."""
    _touch_last_seen()
-    path = _firmware_path()
+    path = firmware_path()
    if not path.exists():
        raise HTTPException(404, "No firmware uploaded")
    return FileResponse(path, media_type="application/octet-stream")


+def _fetch_latest_release(cfg: config.FrameConfig) -> dict | None:
+    try:
+        return gitea_releases.fetch_latest_release(cfg.firmware_update_repo_url, cfg.firmware_update_token)
+    except httpx.HTTPError as e:
+        raise HTTPException(502, f"Could not reach Gitea at {cfg.firmware_update_repo_url}: {e}") from e
+
+
+def _apply_gitea_update(cfg: config.FrameConfig) -> str:
+    """Downloads the configured Gitea repo's latest release asset for this
+    frame's board variant and stages it exactly like a manual
+    POST /api/firmware upload would. Network I/O happens before the lock
+    is taken, matching the load/mutate/save concurrency pattern used
+    elsewhere (see config.locked())."""
+    release = _fetch_latest_release(cfg)
+    if not release:
+        raise HTTPException(404, "No releases found in the configured Gitea repo")
+    asset_name = gitea_releases.asset_name_for_board(cfg.firmware_board_variant)
+    asset_url = release["assets"].get(asset_name)
+    if not asset_url:
+        raise HTTPException(404, f"Latest release has no '{asset_name}' asset")
+    try:
+        data = gitea_releases.download_asset(asset_url, cfg.firmware_update_token)
+    except httpx.HTTPError as e:
+        raise HTTPException(502, f"Could not download '{asset_name}' from Gitea: {e}") from e
+    version = parse_app_version(data)  # same validation the manual upload path applies
+    firmware_path().write_bytes(data)
+    with config.locked():
+        cfg = config.load()
+        cfg.firmware_available_version = version
+        cfg.firmware_gitea_latest_version = version
+        cfg.firmware_update_checked_at = time.time()
+        config.save(cfg)
+    return version
+
+
+@app.get("/api/firmware/check", dependencies=[Depends(require_access_token)])
+def api_firmware_check():
+    """Throttled check of the configured Gitea repo's latest release
+    (gitea_releases.UPDATE_CHECK_INTERVAL_S) -- cheap, since it only reads
+    the release's tag name, not its binaries. If firmware_auto_update is
+    on and a newer version is found, applies it immediately; otherwise
+    just reports it so the web UI can offer the "Update frame" button."""
+    cfg = config.load()
+    if not cfg.firmware_update_repo_url:
+        return {"enabled": False}
+
+    now = time.time()
+    if now - cfg.firmware_update_checked_at >= gitea_releases.UPDATE_CHECK_INTERVAL_S:
+        # Deliberately not updated on failure (see below) -- checked_at only
+        # advances on a successful reach, so a Gitea outage gets retried
+        # every poll instead of waiting out the full throttle interval.
+        release = _fetch_latest_release(cfg)
+        with config.locked():
+            cfg = config.load()  # re-read: state may have changed since the unlocked read above
+            cfg.firmware_update_checked_at = now
+            if release:
+                cfg.firmware_gitea_latest_version = release["version"]
+            config.save(cfg)
+        cfg = config.load()
+
+    update_available = bool(
+        cfg.firmware_gitea_latest_version
+    ) and cfg.firmware_gitea_latest_version != cfg.firmware_available_version
+    if update_available and cfg.firmware_auto_update:
+        _apply_gitea_update(cfg)
+        cfg = config.load()
+        update_available = False
+
+    return {
+        "enabled": True,
+        "latest_version": cfg.firmware_gitea_latest_version or None,
+        "staged_version": cfg.firmware_available_version or None,
+        "update_available": update_available,
+    }
+
+
+@app.post("/api/firmware/apply-latest", dependencies=[Depends(require_access_token)])
+def api_firmware_apply_latest():
+    """The "Update frame" button: applies the latest Gitea release right
+    now, bypassing the check throttle -- this is an explicit user action,
+    not a background poll."""
+    cfg = config.load()
+    if not cfg.firmware_update_repo_url:
+        raise HTTPException(400, "No Gitea firmware repo configured")
+    version = _apply_gitea_update(cfg)
+    return {"status": "saved", "version": version}
+
+
 def _battery_estimate_s(cfg: config.FrameConfig) -> int | None:
    """Linear remaining-time estimate from the current discharge cycle's
    observed rate, or None when there's not enough signal to be honest