feat: OTA firmware update per device with auto config backup

- Upload .bin firmware files via web UI
- Update individual devices (button per device in table)
- Auto config backup before each update
- Shows current version vs available firmware
- Firmware file management (upload/delete)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-24 13:08:38 +02:00
co-authored by Claude Opus 4.6
parent 0d6c03049c
commit 9218d10540
4 changed files with 182 additions and 2 deletions
+69 -2
View File
@@ -39,9 +39,10 @@ async def lifespan(application: FastAPI):
except Exception as e:
logger.error("Graph-Client konnte nicht initialisiert werden: %s", e)
# Backup-Verzeichnis erstellen
# Verzeichnisse erstellen
os.makedirs(BACKUP_DIR, exist_ok=True)
os.makedirs(WLED_BACKUP_DIR, exist_ok=True)
os.makedirs(FIRMWARE_DIR, exist_ok=True)
# Start background polling
interval = int(database.get_setting("poll_interval", "15"))
@@ -61,6 +62,7 @@ async def lifespan(application: FastAPI):
BACKUP_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "backups")
WLED_BACKUP_DIR = os.path.join(BACKUP_DIR, "wled")
FIRMWARE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "firmware")
def auto_backup_db():
@@ -280,8 +282,15 @@ async def devices_page(request: Request):
).fetchone()
conn.close()
last_check = last_check_row["lc"] if last_check_row and last_check_row["lc"] else None
# Firmware files
firmwares = []
if os.path.exists(FIRMWARE_DIR):
for f in sorted(os.listdir(FIRMWARE_DIR), reverse=True):
if f.endswith(".bin"):
size = os.path.getsize(os.path.join(FIRMWARE_DIR, f)) / 1024
firmwares.append({"name": f, "size": f"{size:.0f} KB"})
return templates.TemplateResponse(*_ctx(request, "devices",
devices=devices, last_check=last_check))
devices=devices, last_check=last_check, firmwares=firmwares))
@app.post("/api/devices")
@@ -705,6 +714,64 @@ async def restore_wled_backup(filename: str, device_id: int = Form(...)):
return RedirectResponse(f"/backup?restored={'1' if success else '0'}", status_code=303)
# ── Firmware Update ─────────────────────────────────────────────────────────
@app.post("/api/firmware/upload")
async def upload_firmware(firmware: UploadFile = File(...)):
"""Upload a WLED firmware .bin file."""
if not firmware.filename.endswith(".bin"):
return RedirectResponse("/devices?fw_error=format", status_code=303)
path = os.path.join(FIRMWARE_DIR, firmware.filename)
with open(path, "wb") as f:
content = await firmware.read()
f.write(content)
logger.info("Firmware hochgeladen: %s (%d KB)", firmware.filename, len(content) // 1024)
return RedirectResponse("/devices?fw_uploaded=1", status_code=303)
@app.post("/api/firmware/{device_id}/update")
async def ota_update_device(device_id: int, firmware_file: str = Form(...)):
"""OTA update a single WLED device. Backs up config first."""
from wled_client import get_wled_config, ota_update_wled
conn = database.get_db()
dev = conn.execute("SELECT hostname, name FROM wled_devices WHERE id=?", (device_id,)).fetchone()
conn.close()
if not dev:
return RedirectResponse("/devices?fw_error=notfound", status_code=303)
fw_path = os.path.join(FIRMWARE_DIR, firmware_file)
if not os.path.exists(fw_path) or ".." in firmware_file:
return RedirectResponse("/devices?fw_error=nofile", status_code=303)
# Auto config backup before update
config = get_wled_config(dev["hostname"])
if config:
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
safe_name = dev["hostname"].replace(".", "_")
backup_path = os.path.join(WLED_BACKUP_DIR, f"{safe_name}_{ts}_pre_update.json")
with open(backup_path, "w") as f:
json.dump(config, f, indent=2)
logger.info("Pre-Update Backup: %s", backup_path)
# Push firmware
success, msg = ota_update_wled(dev["hostname"], fw_path)
logger.info("OTA Update %s: %s - %s", dev["hostname"], "OK" if success else "FEHLER", msg)
if success:
return RedirectResponse(f"/devices?fw_updated={dev['hostname']}", status_code=303)
else:
return RedirectResponse(f"/devices?fw_error=failed", status_code=303)
@app.post("/api/firmware/{filename}/delete")
async def delete_firmware(filename: str):
path = os.path.join(FIRMWARE_DIR, filename)
if os.path.exists(path) and ".." not in filename:
os.remove(path)
return RedirectResponse("/devices", status_code=303)
# ── CLI Auth Command ───────────────────────────────────────────────────────
def cli_auth():