update: added assets manager and extra nvs settings on cloudflash

backend/manufacturing/service.py

@@ -168,16 +168,23 @@ def update_device_status(sn: str, data: DeviceStatusUpdate, set_by: str | None =
     doc_data = docs[0].to_dict() or {}
     now = datetime.now(timezone.utc).isoformat()
 
-    history = doc_data.get("lifecycle_history") or []
+    history = list(doc_data.get("lifecycle_history") or [])
 
-    # Append new lifecycle entry
+    # Upsert lifecycle entry — overwrite existing entry for this status if present
     new_entry = {
         "status_id": data.status.value,
         "date": now,
         "note": data.note if data.note else None,
         "set_by": set_by,
     }
-    history.append(new_entry)
+    existing_idx = next(
+        (i for i, e in enumerate(history) if e.get("status_id") == data.status.value),
+        None,
+    )
+    if existing_idx is not None:
+        history[existing_idx] = new_entry
+    else:
+        history.append(new_entry)
 
     update = {
         "mfg_status": data.status.value,
@@ -379,11 +386,68 @@ def delete_unprovisioned_devices() -> list[str]:
     return deleted
 
 
+KNOWN_HW_TYPES = ["vesper", "vesper_plus", "vesper_pro", "agnus", "agnus_mini", "chronos", "chronos_pro"]
+FLASH_ASSET_FILES = ["bootloader.bin", "partitions.bin"]
+
+
 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 _flash_asset_info(hw_type: str) -> dict:
+    """Build the asset info dict for a single hw_type by inspecting the filesystem."""
+    base = Path(settings.flash_assets_storage_path) / hw_type
+    note_path = base / "note.txt"
+    note = note_path.read_text(encoding="utf-8").strip() if note_path.exists() else ""
+
+    files = {}
+    for fname in FLASH_ASSET_FILES:
+        p = base / fname
+        if p.exists():
+            stat = p.stat()
+            files[fname] = {
+                "exists": True,
+                "size_bytes": stat.st_size,
+                "uploaded_at": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(),
+            }
+        else:
+            files[fname] = {"exists": False, "size_bytes": None, "uploaded_at": None}
+
+    return {
+        "hw_type": hw_type,
+        "bootloader": files["bootloader.bin"],
+        "partitions": files["partitions.bin"],
+        "note": note,
+    }
+
+
+def list_flash_assets() -> list:
+    """Return asset status for all known board types plus any discovered bespoke directories."""
+    base = Path(settings.flash_assets_storage_path)
+    results = []
+
+    # Always include all known hw types, even if no files uploaded yet
+    seen = set(KNOWN_HW_TYPES)
+    for hw_type in KNOWN_HW_TYPES:
+        results.append(_flash_asset_info(hw_type))
+
+    # Discover bespoke directories (anything in storage/flash_assets/ not in known list)
+    if base.exists():
+        for entry in sorted(base.iterdir()):
+            if entry.is_dir() and entry.name not in seen:
+                seen.add(entry.name)
+                info = _flash_asset_info(entry.name)
+                info["is_bespoke"] = True
+                results.append(info)
+
+    # Mark known types
+    for r in results:
+        r.setdefault("is_bespoke", False)
+
+    return results
+
+
 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"):
@@ -394,6 +458,25 @@ def save_flash_asset(hw_type: str, asset: str, data: bytes) -> Path:
     return path
 
 
+def delete_flash_asset(hw_type: str, asset: str) -> None:
+    """Delete a flash asset file. Raises NotFoundError if not present."""
+    path = _flash_asset_path(hw_type, asset)
+    if not path.exists():
+        raise NotFoundError(f"Flash asset '{asset}' for '{hw_type}' not found")
+    path.unlink()
+
+
+def set_flash_asset_note(hw_type: str, note: str) -> None:
+    """Write (or clear) the note for a hw_type's flash asset directory."""
+    base = Path(settings.flash_assets_storage_path) / hw_type
+    base.mkdir(parents=True, exist_ok=True)
+    note_path = base / "note.txt"
+    if note.strip():
+        note_path.write_text(note.strip(), encoding="utf-8")
+    elif note_path.exists():
+        note_path.unlink()
+
+
 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)