diff --git a/.gitignore b/.gitignore index 99ef791..b96917b 100644 --- a/.gitignore +++ b/.gitignore @@ -8,6 +8,7 @@ config.cfg config.dev.cfg *.egg-info/ backups/ +firmware/ dist/ build/ .vscode/ diff --git a/app.py b/app.py index 244a2c7..aeb2a76 100644 --- a/app.py +++ b/app.py @@ -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(): diff --git a/templates/devices.html b/templates/devices.html index 151eabb..639d8b6 100644 --- a/templates/devices.html +++ b/templates/devices.html @@ -16,6 +16,29 @@ +{% if request.query_params.get('fw_uploaded') %} +
+ Firmware hochgeladen. + +
+{% endif %} +{% if request.query_params.get('fw_updated') %} +
+ Update auf {{ request.query_params.get('fw_updated') }} gestartet. Geraet startet neu. + +
+{% endif %} +{% if request.query_params.get('fw_error') %} +
+ + {% if request.query_params.get('fw_error') == 'format' %}Nur .bin Dateien erlaubt. + {% elif request.query_params.get('fw_error') == 'nofile' %}Firmware-Datei nicht gefunden. + {% elif request.query_params.get('fw_error') == 'failed' %}Update fehlgeschlagen. Config-Backup wurde erstellt. + {% else %}Firmware-Fehler.{% endif %} + +
+{% endif %} +
@@ -64,6 +87,12 @@ + {% if firmwares %} + + {% endif %}
+ + {% if firmwares %} + + {% endif %} {% endfor %} {% if not devices %} @@ -159,6 +222,39 @@ + + +
+
+
Firmware
+
+
+
+
+ + +
+ +
+ {% if firmwares %} +
+ Verfuegbare Firmwares: + {% for fw in firmwares %} +
+ {{ fw.name }} ({{ fw.size }}) +
+ +
+
+ {% endfor %} + {% endif %} +
+
{% endblock %} diff --git a/wled_client.py b/wled_client.py index c506ad6..f6c23b9 100644 --- a/wled_client.py +++ b/wled_client.py @@ -69,3 +69,19 @@ def restore_wled_config(hostname, config_json): return r.status_code == 200, f"{r.status_code} {r.reason}" except requests.exceptions.RequestException as e: return False, str(e) + + +def ota_update_wled(hostname, firmware_path): + """Push firmware binary to WLED device via OTA. Returns (success, message).""" + try: + with open(firmware_path, "rb") as f: + r = requests.post( + f"http://{hostname}/update", + files={"update": ("firmware.bin", f, "application/octet-stream")}, + timeout=120, + ) + if r.status_code == 200: + return True, "Update erfolgreich, Geraet startet neu" + return False, f"{r.status_code} {r.reason}" + except requests.exceptions.RequestException as e: + return False, str(e)