bonamin/bellsystems-cp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update: overhauled firmware ui. Added public flash page.

Browse Source
This commit is contained in:
Vasileios Kousaridas
2026-03-18 17:49:40 +02:00
parent 4381a6681d
commit d0ac4f1d91
45 changed files with 6798 additions and 1723 deletions
Download Patch File Download Diff File Expand all files Collapse all files

190
backend/firmware/service.py
View File

@@ -16,7 +16,7 @@ logger = logging.getLogger(__name__)
COLLECTION = "firmware_versions"
VALID_HW_TYPES = {"vesper", "vesper_plus", "vesper_pro", "chronos", "chronos_pro", "agnus", "agnus_mini"}
VALID_HW_TYPES = {"vesper", "vesper_plus", "vesper_pro", "chronos", "chronos_pro", "agnus", "agnus_mini", "bespoke"}
VALID_CHANNELS = {"stable", "beta", "alpha", "testing"}
@@ -43,8 +43,10 @@ def _doc_to_firmware_version(doc) -> FirmwareVersion:
update_type=data.get("update_type", UpdateType.mandatory),
min_fw_version=data.get("min_fw_version"),
uploaded_at=uploaded_str,
notes=data.get("notes"),
changelog=data.get("changelog"),
release_note=data.get("release_note"),
is_latest=data.get("is_latest", False),
bespoke_uid=data.get("bespoke_uid"),
)
@@ -65,7 +67,7 @@ def _fw_to_metadata_response(fw: FirmwareVersion) -> FirmwareMetadataResponse:
min_fw_version=fw.min_fw_version,
download_url=download_url,
uploaded_at=fw.uploaded_at,
notes=fw.notes,
release_note=fw.release_note,
)
@@ -76,33 +78,59 @@ def upload_firmware(
file_bytes: bytes,
update_type: UpdateType = UpdateType.mandatory,
min_fw_version: str | None = None,
notes: str | None = None,
changelog: str | None = None,
release_note: str | None = None,
bespoke_uid: str | None = None,
) -> FirmwareVersion:
if hw_type not in VALID_HW_TYPES:
raise HTTPException(status_code=400, detail=f"Invalid hw_type. Must be one of: {', '.join(sorted(VALID_HW_TYPES))}")
if channel not in VALID_CHANNELS:
raise HTTPException(status_code=400, detail=f"Invalid channel. Must be one of: {', '.join(sorted(VALID_CHANNELS))}")
if hw_type == "bespoke" and not bespoke_uid:
raise HTTPException(status_code=400, detail="bespoke_uid is required when hw_type is 'bespoke'")
db = get_db()
sha256 = hashlib.sha256(file_bytes).hexdigest()
now = datetime.now(timezone.utc)
# For bespoke firmware: if a firmware with the same bespoke_uid already exists,
# overwrite it (delete old doc + file, reuse same storage path keyed by uid).
if hw_type == "bespoke" and bespoke_uid:
existing_docs = list(
db.collection(COLLECTION)
.where("hw_type", "==", "bespoke")
.where("bespoke_uid", "==", bespoke_uid)
.stream()
)
for old_doc in existing_docs:
old_data = old_doc.to_dict() or {}
old_path = _storage_path("bespoke", old_data.get("channel", channel), old_data.get("version", version))
if old_path.exists():
old_path.unlink()
try:
old_path.parent.rmdir()
except OSError:
pass
old_doc.reference.delete()
dest = _storage_path(hw_type, channel, version)
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_bytes(file_bytes)
sha256 = hashlib.sha256(file_bytes).hexdigest()
now = datetime.now(timezone.utc)
doc_id = str(uuid.uuid4())
db = get_db()
# Mark previous latest for this hw_type+channel as no longer latest
prev_docs = (
db.collection(COLLECTION)
.where("hw_type", "==", hw_type)
.where("channel", "==", channel)
.where("is_latest", "==", True)
.stream()
)
for prev in prev_docs:
prev.reference.update({"is_latest": False})
# (skip for bespoke — each bespoke_uid is its own independent firmware)
if hw_type != "bespoke":
prev_docs = (
db.collection(COLLECTION)
.where("hw_type", "==", hw_type)
.where("channel", "==", channel)
.where("is_latest", "==", True)
.stream()
)
for prev in prev_docs:
prev.reference.update({"is_latest": False})
doc_ref = db.collection(COLLECTION).document(doc_id)
doc_ref.set({
@@ -115,8 +143,10 @@ def upload_firmware(
"update_type": update_type.value,
"min_fw_version": min_fw_version,
"uploaded_at": now,
"notes": notes,
"changelog": changelog,
"release_note": release_note,
"is_latest": True,
"bespoke_uid": bespoke_uid,
})
return _doc_to_firmware_version(doc_ref.get())
@@ -142,6 +172,8 @@ def list_firmware(
def get_latest(hw_type: str, channel: str, hw_version: str | None = None, current_version: str | None = None) -> FirmwareMetadataResponse:
if hw_type not in VALID_HW_TYPES:
raise HTTPException(status_code=400, detail=f"Invalid hw_type '{hw_type}'")
if hw_type == "bespoke":
raise HTTPException(status_code=400, detail="Bespoke firmware is not served via auto-update. Use the direct download URL.")
if channel not in VALID_CHANNELS:
raise HTTPException(status_code=400, detail=f"Invalid channel '{channel}'")
@@ -182,6 +214,52 @@ def get_version_info(hw_type: str, channel: str, version: str) -> FirmwareMetada
return _fw_to_metadata_response(_doc_to_firmware_version(docs[0]))
def get_latest_changelog(hw_type: str, channel: str) -> str:
if hw_type not in VALID_HW_TYPES:
raise HTTPException(status_code=400, detail=f"Invalid hw_type '{hw_type}'")
if channel not in VALID_CHANNELS:
raise HTTPException(status_code=400, detail=f"Invalid channel '{channel}'")
db = get_db()
docs = list(
db.collection(COLLECTION)
.where("hw_type", "==", hw_type)
.where("channel", "==", channel)
.where("is_latest", "==", True)
.limit(1)
.stream()
)
if not docs:
raise NotFoundError("Firmware")
fw = _doc_to_firmware_version(docs[0])
if not fw.changelog:
raise NotFoundError("Changelog")
return fw.changelog
def get_version_changelog(hw_type: str, channel: str, version: str) -> str:
if hw_type not in VALID_HW_TYPES:
raise HTTPException(status_code=400, detail=f"Invalid hw_type '{hw_type}'")
if channel not in VALID_CHANNELS:
raise HTTPException(status_code=400, detail=f"Invalid channel '{channel}'")
db = get_db()
docs = list(
db.collection(COLLECTION)
.where("hw_type", "==", hw_type)
.where("channel", "==", channel)
.where("version", "==", version)
.limit(1)
.stream()
)
if not docs:
raise NotFoundError("Firmware version")
fw = _doc_to_firmware_version(docs[0])
if not fw.changelog:
raise NotFoundError("Changelog")
return fw.changelog
def get_firmware_path(hw_type: str, channel: str, version: str) -> Path:
path = _storage_path(hw_type, channel, version)
if not path.exists():
@@ -205,6 +283,82 @@ def record_ota_event(event_type: str, payload: dict[str, Any]) -> None:
logger.warning("Failed to persist OTA event (%s): %s", event_type, exc)
def edit_firmware(
doc_id: str,
channel: str | None = None,
version: str | None = None,
update_type: UpdateType | None = None,
min_fw_version: str | None = None,
changelog: str | None = None,
release_note: str | None = None,
bespoke_uid: str | None = None,
file_bytes: bytes | None = None,
) -> FirmwareVersion:
db = get_db()
doc_ref = db.collection(COLLECTION).document(doc_id)
doc = doc_ref.get()
if not doc.exists:
raise NotFoundError("Firmware")
data = doc.to_dict() or {}
hw_type = data["hw_type"]
old_channel = data.get("channel", "")
old_version = data.get("version", "")
effective_channel = channel if channel is not None else old_channel
effective_version = version if version is not None else old_version
if channel is not None and channel not in VALID_CHANNELS:
raise HTTPException(status_code=400, detail=f"Invalid channel. Must be one of: {', '.join(sorted(VALID_CHANNELS))}")
updates: dict = {}
if channel is not None:
updates["channel"] = channel
if version is not None:
updates["version"] = version
if update_type is not None:
updates["update_type"] = update_type.value
if min_fw_version is not None:
updates["min_fw_version"] = min_fw_version if min_fw_version else None
if changelog is not None:
updates["changelog"] = changelog if changelog else None
if release_note is not None:
updates["release_note"] = release_note if release_note else None
if bespoke_uid is not None:
updates["bespoke_uid"] = bespoke_uid if bespoke_uid else None
if file_bytes is not None:
# Move binary if path changed
old_path = _storage_path(hw_type, old_channel, old_version)
new_path = _storage_path(hw_type, effective_channel, effective_version)
if old_path != new_path and old_path.exists():
old_path.unlink()
try:
old_path.parent.rmdir()
except OSError:
pass
new_path.parent.mkdir(parents=True, exist_ok=True)
new_path.write_bytes(file_bytes)
updates["sha256"] = hashlib.sha256(file_bytes).hexdigest()
updates["size_bytes"] = len(file_bytes)
elif (channel is not None and channel != old_channel) or (version is not None and version != old_version):
# Path changed but no new file — move existing binary
old_path = _storage_path(hw_type, old_channel, old_version)
new_path = _storage_path(hw_type, effective_channel, effective_version)
if old_path.exists() and old_path != new_path:
new_path.parent.mkdir(parents=True, exist_ok=True)
old_path.rename(new_path)
try:
old_path.parent.rmdir()
except OSError:
pass
if updates:
doc_ref.update(updates)
return _doc_to_firmware_version(doc_ref.get())
def delete_firmware(doc_id: str) -> None:
db = get_db()
doc_ref = db.collection(COLLECTION).document(doc_id)