Add firmware update helper for LAN-only printers

Enables checking and uploading firmware updates for printers operating
  in LAN-only mode without Bambu Cloud connectivity.

  Features:
  - Automatic firmware version checking against Bambu Lab servers
  - Orange "Update" badge on printer cards when updates available
  - Firmware update modal with version info and release notes
  - One-click firmware upload to printer SD card via FTP
  - Real-time upload progress (actual bytes transferred)
  - Step-by-step instructions for triggering update from printer
  - Local firmware caching for faster re-uploads
  - Supports all Bambu Lab printer models

  New files:
  - backend/app/services/firmware_check.py - Version checking service
  - backend/app/services/firmware_update.py - Upload orchestration
  - backend/app/api/routes/firmware.py - REST API endpoints

  Also includes:
  - FTP upload progress callback support
  - 10-minute upload timeout protection
  - Firmware cache directory in .gitignore
This commit is contained in:
maziggy
2026-01-04 18:41:24 +01:00
parent b0c576c8a8
commit 6fb71de6a2
21 changed files with 10220 additions and 8674 deletions
+3
View File
@@ -32,6 +32,9 @@ npm-debug.log*
# Archive files (user data)
archive/
# Firmware cache (downloaded firmware files)
firmware/
# Virtual printer (auto-generated certs and uploads)
virtual_printer/
+13
View File
@@ -2,6 +2,19 @@
All notable changes to Bambuddy will be documented in this file.
## [0.1.6b7] - 2026-01-04
### Added
- **Firmware update helper** - Check and upload firmware updates for LAN-only printers:
- Automatic firmware update checking against Bambu Lab's servers
- Orange "Update" badge on printer cards when updates are available
- Click badge to open firmware update modal with version info and release notes
- One-click firmware upload to printer's SD card via FTP
- Real-time upload progress tracking with actual bytes transferred
- Step-by-step instructions for triggering update from printer screen
- Supports all Bambu Lab printer models (X1C, X1, X1E, P1S, P1P, P2S, A1, A1 Mini, H2D, H2C, H2S)
- Firmware files cached locally for faster re-uploads
## [0.1.6b6] - 2026-01-04
### Added
+1
View File
@@ -113,6 +113,7 @@
- Interval reminders (hours/days)
- Print time accuracy stats
- File manager for printer storage
- Firmware update helper (LAN-only printers)
</td>
</tr>
+78
View File
@@ -22,6 +22,8 @@ from backend.app.schemas.cloud import (
CloudLoginResponse,
CloudTokenRequest,
CloudVerifyRequest,
FirmwareUpdateInfo,
FirmwareUpdatesResponse,
SlicerSetting,
SlicerSettingCreate,
SlicerSettingDeleteResponse,
@@ -371,6 +373,82 @@ async def get_devices(db: AsyncSession = Depends(get_db)):
raise HTTPException(status_code=500, detail=str(e))
@router.get("/firmware-updates", response_model=FirmwareUpdatesResponse)
async def get_firmware_updates(db: AsyncSession = Depends(get_db)):
"""
Check for firmware updates for all bound devices.
Returns firmware version info for each device including:
- Current installed version
- Latest available version
- Whether an update is available
- Release notes for the latest version
Requires cloud authentication.
"""
token, _ = await get_stored_token(db)
if not token:
raise HTTPException(status_code=401, detail="Not authenticated")
cloud = get_cloud_service()
cloud.set_token(token)
if not cloud.is_authenticated:
raise HTTPException(status_code=401, detail="Not authenticated")
try:
# First get list of bound devices
devices_data = await cloud.get_devices()
devices = devices_data.get("devices", [])
updates = []
updates_available = 0
# Check firmware for each device
for device in devices:
device_id = device.get("dev_id", "")
device_name = device.get("name", "Unknown")
try:
firmware_info = await cloud.get_firmware_version(device_id)
update_available = firmware_info.get("update_available", False)
if update_available:
updates_available += 1
updates.append(
FirmwareUpdateInfo(
device_id=device_id,
device_name=device_name,
current_version=firmware_info.get("current_version"),
latest_version=firmware_info.get("latest_version"),
update_available=update_available,
release_notes=firmware_info.get("release_notes"),
)
)
except BambuCloudError as e:
logger.warning(f"Failed to get firmware info for {device_name}: {e}")
# Still include device but with unknown firmware status
updates.append(
FirmwareUpdateInfo(
device_id=device_id,
device_name=device_name,
current_version=None,
latest_version=None,
update_available=False,
release_notes=None,
)
)
return FirmwareUpdatesResponse(updates=updates, updates_available=updates_available)
except BambuCloudAuthError:
await clear_token(db)
raise HTTPException(status_code=401, detail="Authentication expired")
except BambuCloudError as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/settings")
async def create_setting(request: SlicerSettingCreate, db: AsyncSession = Depends(get_db)):
"""
+298
View File
@@ -0,0 +1,298 @@
"""
Firmware Update API Routes
Check for firmware updates from Bambu Lab.
Also provides endpoints for uploading firmware to printers via SD card.
"""
import logging
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from backend.app.core.database import get_db
from backend.app.models.printer import Printer
from backend.app.services.firmware_check import get_firmware_service
from backend.app.services.firmware_update import (
FirmwareUploadStatus,
get_firmware_update_service,
get_upload_state,
)
from backend.app.services.printer_manager import printer_manager
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/firmware", tags=["firmware"])
class FirmwareUpdateInfo(BaseModel):
"""Firmware update information for a printer."""
printer_id: int
printer_name: str
model: str | None
current_version: str | None
latest_version: str | None
update_available: bool
download_url: str | None = None
release_notes: str | None = None
class FirmwareUpdatesResponse(BaseModel):
"""Response containing firmware updates for all printers."""
updates: list[FirmwareUpdateInfo] = Field(default_factory=list)
updates_available: int = Field(0, description="Number of printers with updates available")
class LatestFirmwareInfo(BaseModel):
"""Latest firmware version info for a model."""
model_key: str
version: str
download_url: str
release_notes: str | None = None
@router.get("/updates", response_model=FirmwareUpdatesResponse)
async def check_firmware_updates(
db: AsyncSession = Depends(get_db),
):
"""
Check for firmware updates for all connected printers.
Compares each printer's current firmware version against the latest
available version from Bambu Lab's official firmware download page.
Note: This does not require cloud authentication - it uses public
firmware information from bambulab.com.
"""
firmware_service = get_firmware_service()
# Get all printers from database
result = await db.execute(select(Printer).where(Printer.is_active.is_(True)))
printers = result.scalars().all()
updates = []
updates_available = 0
for printer in printers:
# Get current firmware version from MQTT state
current_version = None
mqtt_client = printer_manager.get_client(printer.id)
if mqtt_client and mqtt_client.state:
current_version = mqtt_client.state.firmware_version
# Check for update
model = printer.model or "Unknown"
update_info = await firmware_service.check_for_update(model, current_version or "")
if update_info["update_available"]:
updates_available += 1
updates.append(
FirmwareUpdateInfo(
printer_id=printer.id,
printer_name=printer.name,
model=model,
current_version=current_version,
latest_version=update_info["latest_version"],
update_available=update_info["update_available"],
download_url=update_info["download_url"],
release_notes=update_info["release_notes"],
)
)
return FirmwareUpdatesResponse(updates=updates, updates_available=updates_available)
@router.get("/updates/{printer_id}", response_model=FirmwareUpdateInfo)
async def check_printer_firmware(
printer_id: int,
db: AsyncSession = Depends(get_db),
):
"""
Check for firmware update for a specific printer.
"""
firmware_service = get_firmware_service()
# Get printer from database
result = await db.execute(select(Printer).where(Printer.id == printer_id))
printer = result.scalar_one_or_none()
if not printer:
raise HTTPException(status_code=404, detail="Printer not found")
# Get current firmware version from MQTT state
current_version = None
mqtt_client = printer_manager.get_client(printer.id)
if mqtt_client and mqtt_client.state:
current_version = mqtt_client.state.firmware_version
# Check for update
model = printer.model or "Unknown"
update_info = await firmware_service.check_for_update(model, current_version or "")
return FirmwareUpdateInfo(
printer_id=printer.id,
printer_name=printer.name,
model=model,
current_version=current_version,
latest_version=update_info["latest_version"],
update_available=update_info["update_available"],
download_url=update_info["download_url"],
release_notes=update_info["release_notes"],
)
@router.get("/latest", response_model=list[LatestFirmwareInfo])
async def get_all_latest_firmware():
"""
Get the latest firmware versions for all Bambu Lab printer models.
This endpoint fetches the latest available firmware versions from
Bambu Lab's official firmware download page.
"""
firmware_service = get_firmware_service()
versions = await firmware_service.get_all_latest_versions()
return [
LatestFirmwareInfo(
model_key=key,
version=info.version,
download_url=info.download_url,
release_notes=info.release_notes,
)
for key, info in versions.items()
]
# ============================================================================
# Firmware Upload Endpoints (for LAN-only firmware updates)
# ============================================================================
class FirmwareUploadPrepareResponse(BaseModel):
"""Response from firmware upload preparation check."""
can_proceed: bool
sd_card_present: bool
sd_card_free_space: int = Field(-1, description="Free space in bytes, -1 if unknown")
firmware_size: int = Field(0, description="Estimated firmware size in bytes")
space_sufficient: bool
update_available: bool
current_version: str | None = None
latest_version: str | None = None
firmware_filename: str | None = None
errors: list[str] = Field(default_factory=list)
class FirmwareUploadStatusResponse(BaseModel):
"""Response containing firmware upload status."""
status: str # idle, preparing, downloading, uploading, complete, error
progress: int = Field(0, ge=0, le=100)
message: str = ""
error: str | None = None
firmware_filename: str | None = None
firmware_version: str | None = None
class FirmwareUploadStartResponse(BaseModel):
"""Response when starting a firmware upload."""
started: bool
message: str
@router.get("/updates/{printer_id}/prepare", response_model=FirmwareUploadPrepareResponse)
async def prepare_firmware_upload(
printer_id: int,
db: AsyncSession = Depends(get_db),
):
"""
Check prerequisites for uploading firmware to a printer.
This performs pre-flight checks including:
- SD card presence
- Available storage space
- Update availability
Call this before starting a firmware upload to ensure the operation
can succeed.
"""
update_service = get_firmware_update_service()
result = await update_service.prepare_update(printer_id, db)
return FirmwareUploadPrepareResponse(**result)
@router.post("/updates/{printer_id}/upload", response_model=FirmwareUploadStartResponse)
async def start_firmware_upload(
printer_id: int,
db: AsyncSession = Depends(get_db),
):
"""
Start uploading firmware to a printer's SD card.
This initiates a background process that:
1. Downloads the firmware from Bambu Lab
2. Uploads it to the printer's SD card via FTP
Progress is broadcast via WebSocket with type "firmware_upload_progress".
Use GET /firmware/updates/{printer_id}/upload/status for polling fallback.
After upload completes, the user must trigger the update from the
printer's screen (Settings > Firmware).
"""
# First check prerequisites
update_service = get_firmware_update_service()
prepare_result = await update_service.prepare_update(printer_id, db)
if not prepare_result["can_proceed"]:
errors = prepare_result.get("errors", ["Cannot proceed with firmware upload"])
raise HTTPException(
status_code=400,
detail="; ".join(errors),
)
# Start the upload
started = await update_service.start_upload(printer_id, db)
if not started:
state = get_upload_state(printer_id)
if state.status == FirmwareUploadStatus.DOWNLOADING:
return FirmwareUploadStartResponse(
started=False,
message="Firmware upload already in progress",
)
raise HTTPException(
status_code=500,
detail=state.error or "Failed to start firmware upload",
)
return FirmwareUploadStartResponse(
started=True,
message="Firmware upload started. Progress will be broadcast via WebSocket.",
)
@router.get("/updates/{printer_id}/upload/status", response_model=FirmwareUploadStatusResponse)
async def get_firmware_upload_status(printer_id: int):
"""
Get the current status of a firmware upload operation.
This is a polling fallback for clients that don't use WebSocket.
For real-time updates, connect to WebSocket and listen for
"firmware_upload_progress" messages.
"""
state = get_upload_state(printer_id)
return FirmwareUploadStatusResponse(
status=state.status.value,
progress=state.progress,
message=state.message,
error=state.error,
firmware_filename=state.firmware_filename,
firmware_version=state.firmware_version,
)
+5
View File
@@ -362,6 +362,11 @@ async def get_printer_status(printer_id: int, db: AsyncSession = Depends(get_db)
mc_print_sub_stage=state.mc_print_sub_stage,
last_ams_update=state.last_ams_update,
printable_objects_count=len(state.printable_objects),
cooling_fan_speed=state.cooling_fan_speed,
big_fan1_speed=state.big_fan1_speed,
big_fan2_speed=state.big_fan2_speed,
heatbreak_fan_speed=state.heatbreak_fan_speed,
firmware_version=state.firmware_version,
)
+2
View File
@@ -59,6 +59,7 @@ from backend.app.api.routes import (
discovery,
external_links,
filaments,
firmware,
kprofiles,
maintenance,
notification_templates,
@@ -1814,6 +1815,7 @@ app.include_router(system.router, prefix=app_settings.api_prefix)
app.include_router(websocket.router, prefix=app_settings.api_prefix)
app.include_router(discovery.router, prefix=app_settings.api_prefix)
app.include_router(pending_uploads.router, prefix=app_settings.api_prefix)
app.include_router(firmware.router, prefix=app_settings.api_prefix)
# Serve static files (React build)
+18
View File
@@ -106,3 +106,21 @@ class SlicerSettingDeleteResponse(BaseModel):
success: bool
message: str
class FirmwareUpdateInfo(BaseModel):
"""Firmware update information for a device."""
device_id: str = Field(..., description="Device ID")
device_name: str = Field(..., description="Device name")
current_version: str | None = Field(None, description="Currently installed firmware version")
latest_version: str | None = Field(None, description="Latest available firmware version")
update_available: bool = Field(False, description="Whether an update is available")
release_notes: str | None = Field(None, description="Release notes for the latest version")
class FirmwareUpdatesResponse(BaseModel):
"""Response containing firmware updates for all devices."""
updates: list[FirmwareUpdateInfo] = Field(default_factory=list)
updates_available: int = Field(0, description="Total number of devices with updates available")
+2
View File
@@ -158,3 +158,5 @@ class PrinterStatus(BaseModel):
big_fan1_speed: int | None = None # Auxiliary fan
big_fan2_speed: int | None = None # Chamber/exhaust fan
heatbreak_fan_speed: int | None = None # Hotend heatbreak fan
# Firmware version (from info.module[name="ota"].sw_ver)
firmware_version: str | None = None
+30
View File
@@ -378,6 +378,36 @@ class BambuCloudService:
except httpx.RequestError as e:
raise BambuCloudError(f"Request failed: {e}")
async def get_firmware_version(self, device_id: str) -> dict:
"""
Get firmware version info for a device.
Returns dict with:
- current_version: Installed firmware version
- latest_version: Latest available firmware version
- update_available: Boolean indicating if update is available
- release_notes: Release notes for latest version
"""
if not self.is_authenticated:
raise BambuCloudAuthError("Not authenticated")
try:
response = await self._client.get(
f"{self.base_url}/v1/iot-service/api/user/device/version",
headers=self._get_headers(),
params={"device_id": device_id},
)
if response.status_code == 200:
data = response.json()
# API wraps response in 'data' field
return data.get("data", data)
raise BambuCloudError(f"Failed to get firmware version: {response.status_code}")
except httpx.RequestError as e:
raise BambuCloudError(f"Request failed: {e}")
async def close(self):
"""Close the HTTP client."""
await self._client.aclose()
+27 -6
View File
@@ -2,6 +2,7 @@ import asyncio
import logging
import socket
import ssl
from collections.abc import Callable
from ftplib import FTP, FTP_TLS
from io import BytesIO
from pathlib import Path
@@ -183,8 +184,13 @@ class BambuFTPClient:
pass
return False
def upload_file(self, local_path: Path, remote_path: str) -> bool:
"""Upload a file to the printer."""
def upload_file(
self,
local_path: Path,
remote_path: str,
progress_callback: Callable[[int, int], None] | None = None,
) -> bool:
"""Upload a file to the printer with optional progress callback."""
if not self._ftp:
logger.warning("upload_file: FTP not connected")
return False
@@ -192,8 +198,17 @@ class BambuFTPClient:
try:
file_size = local_path.stat().st_size if local_path.exists() else 0
logger.info(f"FTP uploading {local_path} ({file_size} bytes) to {remote_path}")
uploaded = 0
def on_block(block: bytes):
nonlocal uploaded
uploaded += len(block)
if progress_callback:
progress_callback(uploaded, file_size)
with open(local_path, "rb") as f:
self._ftp.storbinary(f"STOR {remote_path}", f)
self._ftp.storbinary(f"STOR {remote_path}", f, callback=on_block)
logger.info(f"FTP upload complete: {remote_path}")
return True
except Exception as e:
@@ -340,8 +355,10 @@ async def upload_file_async(
access_code: str,
local_path: Path,
remote_path: str,
timeout: float = 600.0,
progress_callback: Callable[[int, int], None] | None = None,
) -> bool:
"""Async wrapper for uploading a file."""
"""Async wrapper for uploading a file with timeout and progress callback."""
loop = asyncio.get_event_loop()
def _upload():
@@ -350,13 +367,17 @@ async def upload_file_async(
if client.connect():
logger.info(f"FTP connected to {ip_address}")
try:
return client.upload_file(local_path, remote_path)
return client.upload_file(local_path, remote_path, progress_callback)
finally:
client.disconnect()
logger.warning(f"FTP connection failed to {ip_address}")
return False
return await loop.run_in_executor(None, _upload)
try:
return await asyncio.wait_for(loop.run_in_executor(None, _upload), timeout=timeout)
except TimeoutError:
logger.warning(f"FTP upload timed out after {timeout}s for {remote_path}")
return False
async def list_files_async(
+56
View File
@@ -158,6 +158,8 @@ class PrinterState:
big_fan1_speed: int | None = None # Auxiliary fan
big_fan2_speed: int | None = None # Chamber/exhaust fan
heatbreak_fan_speed: int | None = None # Hotend heatbreak fan
# Firmware version info (from info.module[name="ota"].sw_ver)
firmware_version: str | None = None
# Stage name mapping from BambuStudio DeviceManager.cpp
@@ -323,6 +325,8 @@ class BambuMQTTClient:
client.subscribe(self.topic_subscribe)
# Request full status update (includes nozzle info in push_status response)
self._request_push_all()
# Request firmware version info
self._request_version()
# Note: get_accessories returns stale nozzle data on H2D, so we don't use it.
# The correct nozzle data comes from push_status.
# Prime K-profile request (Bambu printers often ignore first request)
@@ -397,6 +401,12 @@ class BambuMQTTClient:
logger.info(f"[{self.serial_number}] Received system data: {system_data}")
self._handle_system_response(system_data)
# Handle info responses (firmware version info from get_version command)
if "info" in payload:
info_data = payload["info"]
if isinstance(info_data, dict) and info_data.get("command") == "get_version":
self._handle_version_info(info_data)
# Parse WiFi signal at top level (some printers send it here)
if "wifi_signal" in payload:
wifi_signal = payload["wifi_signal"]
@@ -487,6 +497,39 @@ class BambuMQTTClient:
# actual nozzle is 'HH01' hardened steel high-flow)
logger.info(f"[{self.serial_number}] Accessories response (not used for nozzle data): {data}")
def _handle_version_info(self, data: dict):
"""Handle version info response from get_version command.
Parses firmware version from the 'ota' module in the module list.
Message format:
{
"command": "get_version",
"module": [
{"name": "ota", "sw_ver": "01.08.05.00"},
{"name": "rv1126", "sw_ver": "00.00.14.74"},
...
]
}
"""
modules = data.get("module", [])
if not isinstance(modules, list):
return
for module in modules:
if not isinstance(module, dict):
continue
if module.get("name") == "ota":
version = module.get("sw_ver")
if version:
old_version = self.state.firmware_version
self.state.firmware_version = version
if old_version != version:
logger.info(f"[{self.serial_number}] Firmware version: {version}")
# Trigger state change callback
if self.on_state_change:
self.on_state_change(self.state)
break
def _parse_xcam_data(self, xcam_data):
"""Parse xcam data for camera settings and AI detection options."""
if not isinstance(xcam_data, dict):
@@ -1766,6 +1809,19 @@ class BambuMQTTClient:
message = {"pushing": {"command": "pushall"}}
self._client.publish(self.topic_publish, json.dumps(message), qos=1)
def _request_version(self):
"""Request firmware version info from printer."""
if self._client:
self._sequence_id += 1
message = {
"info": {
"sequence_id": str(self._sequence_id),
"command": "get_version",
}
}
logger.debug(f"[{self.serial_number}] Requesting firmware version info")
self._client.publish(self.topic_publish, json.dumps(message), qos=1)
def request_status_update(self) -> bool:
"""Request a full status update from the printer (public API).
+382
View File
@@ -0,0 +1,382 @@
"""
Firmware Check Service
Checks for firmware updates by fetching from Bambu Lab's official firmware download page.
Also provides firmware download functionality for offline updates.
"""
import logging
import re
import time
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
import httpx
from backend.app.core.config import _data_dir
logger = logging.getLogger(__name__)
# Bambu Lab firmware download page
BAMBU_FIRMWARE_BASE = "https://bambulab.com"
FIRMWARE_PAGE = "/en/support/firmware-download/all"
# Cache TTL in seconds (1 hour)
CACHE_TTL = 3600
# Map Bambuddy model names to Bambu Lab API keys
MODEL_TO_API_KEY = {
"X1": "x1",
"X1C": "x1",
"X1-Carbon": "x1",
"X1 Carbon": "x1",
"P1P": "p1",
"P1S": "p1",
"A1": "a1",
"A1 Mini": "a1-mini",
"A1-Mini": "a1-mini",
"A1mini": "a1-mini",
"H2D": "h2d",
"H2C": "h2d", # H2C uses same firmware as H2D
"H2S": "h2s",
"P2S": "p2s",
"X1E": "x1e",
"H2D Pro": "h2d-pro",
"H2D-Pro": "h2d-pro",
}
# Reverse mapping: API key to model codes
API_KEY_TO_DEV_MODEL = {
"x1": "BL-P001",
"p1": "C11",
"a1": "N2S",
"a1-mini": "N1",
"h2d": "O1D",
"h2s": "O1S",
"p2s": "N7",
"x1e": "C13",
"h2d-pro": "O1E",
}
@dataclass
class FirmwareVersion:
"""Firmware version information."""
version: str
download_url: str
release_notes: str | None = None
release_time: str | None = None
class FirmwareCheckService:
"""Service for checking firmware updates from Bambu Lab."""
def __init__(self):
self._build_id: str | None = None
self._build_id_time: float = 0
self._version_cache: dict[str, FirmwareVersion] = {}
self._cache_time: float = 0
self._client = httpx.AsyncClient(
timeout=30.0,
headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
},
)
async def _get_build_id(self) -> str | None:
"""Fetch the Next.js build ID from Bambu Lab's firmware page."""
# Use cached build ID if still valid (cache for 1 hour)
if self._build_id and (time.time() - self._build_id_time) < CACHE_TTL:
return self._build_id
try:
response = await self._client.get(f"{BAMBU_FIRMWARE_BASE}{FIRMWARE_PAGE}")
if response.status_code == 200:
# Extract buildId from the page
match = re.search(r'"buildId":"([^"]+)"', response.text)
if match:
self._build_id = match.group(1)
self._build_id_time = time.time()
logger.info(f"Got Bambu Lab build ID: {self._build_id}")
return self._build_id
logger.warning(f"Failed to get Bambu Lab page: {response.status_code}")
except Exception as e:
logger.error(f"Error fetching Bambu Lab build ID: {e}")
return self._build_id # Return cached value if available
async def _fetch_firmware_versions(self, api_key: str) -> FirmwareVersion | None:
"""Fetch firmware versions for a specific printer from Bambu Lab API."""
build_id = await self._get_build_id()
if not build_id:
logger.warning("No build ID available, cannot fetch firmware versions")
return None
try:
url = f"{BAMBU_FIRMWARE_BASE}/_next/data/{build_id}/en/support/firmware-download/{api_key}.json"
response = await self._client.get(url)
if response.status_code == 200:
data = response.json()
page_props = data.get("pageProps", {})
printer_map = page_props.get("printerMap", {})
printer_data = printer_map.get(api_key, {})
versions = printer_data.get("versions", [])
if versions:
latest = versions[0]
return FirmwareVersion(
version=latest.get("version", ""),
download_url=latest.get("url", ""),
release_notes=latest.get("release_notes_en"),
release_time=latest.get("release_time"),
)
else:
logger.warning(f"Failed to fetch firmware for {api_key}: {response.status_code}")
except Exception as e:
logger.error(f"Error fetching firmware for {api_key}: {e}")
return None
async def get_latest_version(self, model: str) -> FirmwareVersion | None:
"""
Get the latest firmware version for a printer model.
Args:
model: Bambuddy printer model name (e.g., "X1C", "P1S", "H2D")
Returns:
FirmwareVersion if found, None otherwise
"""
# Normalize model name
model_upper = model.upper().replace(" ", "").replace("-", "")
# Find the API key for this model
api_key = None
for model_name, key in MODEL_TO_API_KEY.items():
if model_name.upper().replace(" ", "").replace("-", "") == model_upper:
api_key = key
break
if not api_key:
# Try direct lookup with original model
api_key = MODEL_TO_API_KEY.get(model)
if not api_key:
logger.debug(f"Unknown printer model: {model}")
return None
# Check cache
cache_key = api_key
if cache_key in self._version_cache and (time.time() - self._cache_time) < CACHE_TTL:
return self._version_cache[cache_key]
# Fetch from API
version = await self._fetch_firmware_versions(api_key)
if version:
self._version_cache[cache_key] = version
self._cache_time = time.time()
return version
async def check_for_update(self, model: str, current_version: str) -> dict:
"""
Check if a firmware update is available for a printer.
Args:
model: Printer model name
current_version: Currently installed firmware version
Returns:
Dict with update info:
- update_available: bool
- current_version: str
- latest_version: str or None
- download_url: str or None
- release_notes: str or None
"""
result = {
"update_available": False,
"current_version": current_version,
"latest_version": None,
"download_url": None,
"release_notes": None,
}
if not current_version:
return result
latest = await self.get_latest_version(model)
if not latest:
return result
result["latest_version"] = latest.version
result["download_url"] = latest.download_url
result["release_notes"] = latest.release_notes
# Compare versions (format: XX.XX.XX.XX)
try:
current_parts = [int(x) for x in current_version.split(".")]
latest_parts = [int(x) for x in latest.version.split(".")]
# Pad to same length
while len(current_parts) < 4:
current_parts.append(0)
while len(latest_parts) < 4:
latest_parts.append(0)
result["update_available"] = latest_parts > current_parts
except (ValueError, AttributeError):
logger.warning(f"Could not compare versions: {current_version} vs {latest.version}")
return result
async def get_all_latest_versions(self) -> dict[str, FirmwareVersion]:
"""
Fetch latest firmware versions for all known printer models.
Returns:
Dict mapping API key to FirmwareVersion
"""
results = {}
for api_key in API_KEY_TO_DEV_MODEL:
version = await self._fetch_firmware_versions(api_key)
if version:
results[api_key] = version
return results
def _get_firmware_cache_dir(self) -> Path:
"""Get the firmware cache directory, creating it if needed."""
cache_dir = _data_dir / "firmware"
cache_dir.mkdir(parents=True, exist_ok=True)
return cache_dir
def _get_cached_firmware_path(self, model: str, version: str) -> Path:
"""Get the path where a firmware file would be cached."""
# Normalize model name for filename
model_safe = model.upper().replace(" ", "-").replace("/", "-")
version_safe = version.replace(".", "_")
filename = f"{model_safe}_{version_safe}.bin"
return self._get_firmware_cache_dir() / filename
async def get_firmware_file_info(self, model: str) -> dict | None:
"""
Get information about the firmware file for a model.
Returns:
Dict with download_url, version, filename, and estimated_size (if available)
"""
latest = await self.get_latest_version(model)
if not latest or not latest.download_url:
return None
# Extract filename from URL
url_parts = latest.download_url.split("/")
filename = url_parts[-1] if url_parts else f"firmware_{model}.bin"
return {
"download_url": latest.download_url,
"version": latest.version,
"filename": filename,
"release_notes": latest.release_notes,
}
async def download_firmware(
self,
model: str,
progress_callback: Callable[[int, int, str], None] | None = None,
) -> Path | None:
"""
Download firmware file for a printer model.
Args:
model: Printer model name (e.g., "X1C", "P1S", "H2D")
progress_callback: Optional callback(bytes_downloaded, total_bytes, status_message)
Returns:
Path to downloaded firmware file, or None on failure
"""
latest = await self.get_latest_version(model)
if not latest or not latest.download_url:
logger.warning(f"No firmware download URL available for model: {model}")
return None
# Check if already cached
cached_path = self._get_cached_firmware_path(model, latest.version)
if cached_path.exists():
logger.info(f"Using cached firmware: {cached_path}")
return cached_path
# Extract original filename from URL (must preserve for SD card update)
url_parts = latest.download_url.split("/")
original_filename = url_parts[-1] if url_parts else f"firmware_{model}.bin"
# Download to temp file first
temp_path = self._get_firmware_cache_dir() / f".downloading_{original_filename}"
try:
logger.info(f"Downloading firmware from {latest.download_url}")
if progress_callback:
progress_callback(0, 0, "Starting download...")
async with self._client.stream("GET", latest.download_url) as response:
if response.status_code != 200:
logger.error(f"Firmware download failed with status {response.status_code}")
return None
total_size = int(response.headers.get("content-length", 0))
downloaded = 0
with open(temp_path, "wb") as f:
async for chunk in response.aiter_bytes(chunk_size=65536):
f.write(chunk)
downloaded += len(chunk)
if progress_callback:
progress_callback(downloaded, total_size, "Downloading firmware...")
# Also save a copy with the original filename for SD card
original_path = self._get_firmware_cache_dir() / original_filename
if original_path.exists():
original_path.unlink()
# Move temp to both cached path and original filename path
import shutil
shutil.copy2(temp_path, cached_path)
temp_path.rename(original_path)
logger.info(f"Firmware downloaded successfully: {original_path}")
if progress_callback:
progress_callback(downloaded, total_size, "Download complete")
return original_path
except Exception as e:
logger.error(f"Firmware download failed: {e}")
if temp_path.exists():
try:
temp_path.unlink()
except Exception:
pass
return None
async def close(self):
"""Close the HTTP client."""
await self._client.aclose()
# Singleton instance
_firmware_service: FirmwareCheckService | None = None
def get_firmware_service() -> FirmwareCheckService:
"""Get the singleton firmware check service instance."""
global _firmware_service
if _firmware_service is None:
_firmware_service = FirmwareCheckService()
return _firmware_service
+353
View File
@@ -0,0 +1,353 @@
"""
Firmware Update Service
Orchestrates firmware updates for Bambu Lab printers:
1. Check prerequisites (SD card, space, update available)
2. Download firmware from Bambu Lab
3. Upload to printer's SD card via FTP
4. Notify user to trigger update from printer screen
"""
import asyncio
import logging
from dataclasses import dataclass
from enum import Enum
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from backend.app.core.websocket import ws_manager
from backend.app.models.printer import Printer
from backend.app.services.bambu_ftp import get_storage_info_async, upload_file_async
from backend.app.services.firmware_check import get_firmware_service
from backend.app.services.printer_manager import printer_manager
logger = logging.getLogger(__name__)
class FirmwareUploadStatus(str, Enum):
"""Status of a firmware upload operation."""
IDLE = "idle"
PREPARING = "preparing"
DOWNLOADING = "downloading"
UPLOADING = "uploading"
COMPLETE = "complete"
ERROR = "error"
@dataclass
class FirmwareUploadState:
"""State of a firmware upload operation for a printer."""
status: FirmwareUploadStatus = FirmwareUploadStatus.IDLE
progress: int = 0 # 0-100
message: str = ""
error: str | None = None
firmware_filename: str | None = None
firmware_version: str | None = None
# Track upload state per printer
_upload_states: dict[int, FirmwareUploadState] = {}
def get_upload_state(printer_id: int) -> FirmwareUploadState:
"""Get the current upload state for a printer."""
if printer_id not in _upload_states:
_upload_states[printer_id] = FirmwareUploadState()
return _upload_states[printer_id]
def reset_upload_state(printer_id: int):
"""Reset the upload state for a printer."""
_upload_states[printer_id] = FirmwareUploadState()
class FirmwareUpdateService:
"""Service for managing firmware updates."""
# Minimum free space required (100MB buffer)
MIN_FREE_SPACE_BYTES = 100 * 1024 * 1024
async def prepare_update(
self,
printer_id: int,
db: AsyncSession,
) -> dict:
"""
Check prerequisites for firmware update.
Returns:
Dict with:
- can_proceed: bool
- sd_card_present: bool
- sd_card_free_space: int (bytes, -1 if unknown)
- firmware_size: int (bytes, estimated)
- space_sufficient: bool
- update_available: bool
- current_version: str | None
- latest_version: str | None
- firmware_filename: str | None
- errors: list[str]
"""
result = {
"can_proceed": False,
"sd_card_present": False,
"sd_card_free_space": -1,
"firmware_size": 0,
"space_sufficient": False,
"update_available": False,
"current_version": None,
"latest_version": None,
"firmware_filename": None,
"errors": [],
}
# Get printer from database
stmt = select(Printer).where(Printer.id == printer_id)
db_result = await db.execute(stmt)
printer = db_result.scalar_one_or_none()
if not printer:
result["errors"].append("Printer not found")
return result
# Check printer is connected
mqtt_client = printer_manager.get_client(printer_id)
if not mqtt_client or not mqtt_client.state:
result["errors"].append("Printer not connected")
return result
state = mqtt_client.state
# Get current firmware version
result["current_version"] = state.firmware_version
# Check SD card
result["sd_card_present"] = state.sdcard
if not state.sdcard:
result["errors"].append("No SD card inserted in printer")
# Get storage info via FTP
if state.sdcard:
try:
storage_info = await get_storage_info_async(
printer.ip_address,
printer.access_code,
)
if storage_info and "free_bytes" in storage_info:
result["sd_card_free_space"] = storage_info["free_bytes"]
except Exception as e:
logger.warning(f"Could not get storage info: {e}")
# Check for firmware update
firmware_service = get_firmware_service()
model = printer.model or "Unknown"
if state.firmware_version:
update_info = await firmware_service.check_for_update(model, state.firmware_version)
result["update_available"] = update_info["update_available"]
result["latest_version"] = update_info["latest_version"]
else:
# If we don't know current version, just get latest
latest = await firmware_service.get_latest_version(model)
if latest:
result["latest_version"] = latest.version
result["update_available"] = True # Assume update needed
if not result["update_available"]:
result["errors"].append("Firmware is already up to date")
# Get firmware file info
file_info = await firmware_service.get_firmware_file_info(model)
if file_info:
result["firmware_filename"] = file_info["filename"]
# Estimate size (typical firmware is 50-150MB)
# We'll get actual size during download
result["firmware_size"] = 100 * 1024 * 1024 # 100MB estimate
# Check space
if result["sd_card_free_space"] > 0:
# Need firmware size + buffer
required = result["firmware_size"] + self.MIN_FREE_SPACE_BYTES
result["space_sufficient"] = result["sd_card_free_space"] >= required
if not result["space_sufficient"]:
result["errors"].append(
f"Insufficient SD card space. Need {required // (1024*1024)}MB, "
f"have {result['sd_card_free_space'] // (1024*1024)}MB"
)
elif result["sd_card_present"]:
# Couldn't determine space, assume sufficient
result["space_sufficient"] = True
# Final check
result["can_proceed"] = (
result["sd_card_present"]
and result["space_sufficient"]
and result["update_available"]
and len(result["errors"]) == 0
)
return result
async def start_upload(
self,
printer_id: int,
db: AsyncSession,
) -> bool:
"""
Start the firmware upload process.
This runs asynchronously and broadcasts progress via WebSocket.
Returns True if upload started successfully.
"""
state = get_upload_state(printer_id)
# Check if already in progress
if state.status in (FirmwareUploadStatus.DOWNLOADING, FirmwareUploadStatus.UPLOADING):
logger.warning(f"Firmware upload already in progress for printer {printer_id}")
return False
# Get printer
stmt = select(Printer).where(Printer.id == printer_id)
db_result = await db.execute(stmt)
printer = db_result.scalar_one_or_none()
if not printer:
state.status = FirmwareUploadStatus.ERROR
state.error = "Printer not found"
return False
# Get printer model
model = printer.model or "Unknown"
# Reset state
reset_upload_state(printer_id)
state = get_upload_state(printer_id)
state.status = FirmwareUploadStatus.PREPARING
state.message = "Preparing firmware update..."
await self._broadcast_progress(printer_id, state)
# Run the upload in background
asyncio.create_task(
self._do_upload(
printer_id=printer_id,
ip_address=printer.ip_address,
access_code=printer.access_code,
model=model,
)
)
return True
async def _do_upload(
self,
printer_id: int,
ip_address: str,
access_code: str,
model: str,
):
"""Perform the actual firmware download and upload."""
state = get_upload_state(printer_id)
firmware_service = get_firmware_service()
try:
# Download firmware (quick, usually cached)
state.status = FirmwareUploadStatus.DOWNLOADING
state.progress = 0
state.message = "Preparing firmware..."
await self._broadcast_progress(printer_id, state)
firmware_path = await firmware_service.download_firmware(model)
if not firmware_path:
raise Exception("Failed to download firmware")
state.firmware_filename = firmware_path.name
# Get firmware version for state
latest = await firmware_service.get_latest_version(model)
if latest:
state.firmware_version = latest.version
# Upload to printer (0-100% progress shown here)
state.status = FirmwareUploadStatus.UPLOADING
state.progress = 0
state.message = f"Uploading {firmware_path.name} to printer..."
await self._broadcast_progress(printer_id, state)
# Upload to root of SD card (where printer expects firmware)
remote_path = f"/{firmware_path.name}"
logger.info(f"Uploading firmware to printer {printer_id}: {remote_path}")
# Track real progress via FTP callback
loop = asyncio.get_event_loop()
last_progress = 0
def on_upload_progress(uploaded: int, total: int):
nonlocal last_progress
if total > 0:
progress = int((uploaded / total) * 100)
# Only broadcast every 1% to avoid flooding
if progress > last_progress:
last_progress = progress
state.progress = min(99, progress) # Cap at 99 until complete
asyncio.run_coroutine_threadsafe(self._broadcast_progress(printer_id, state), loop)
success = await upload_file_async(
ip_address,
access_code,
firmware_path,
remote_path,
progress_callback=on_upload_progress,
)
if not success:
raise Exception("Failed to upload firmware to printer")
# Complete
state.status = FirmwareUploadStatus.COMPLETE
state.progress = 100
state.message = (
f"Firmware {state.firmware_version or ''} uploaded successfully! "
"Please go to printer screen and trigger the update from Settings > Firmware."
)
await self._broadcast_progress(printer_id, state)
logger.info(f"Firmware upload complete for printer {printer_id}")
except Exception as e:
logger.error(f"Firmware upload failed for printer {printer_id}: {e}")
state.status = FirmwareUploadStatus.ERROR
state.error = str(e)
state.message = f"Firmware upload failed: {e}"
await self._broadcast_progress(printer_id, state)
async def _broadcast_progress(self, printer_id: int, state: FirmwareUploadState):
"""Broadcast firmware upload progress via WebSocket."""
await ws_manager.broadcast(
{
"type": "firmware_upload_progress",
"printer_id": printer_id,
"status": state.status.value,
"progress": state.progress,
"message": state.message,
"error": state.error,
"firmware_filename": state.firmware_filename,
"firmware_version": state.firmware_version,
}
)
# Singleton instance
_firmware_update_service: FirmwareUpdateService | None = None
def get_firmware_update_service() -> FirmwareUpdateService:
"""Get the singleton firmware update service instance."""
global _firmware_update_service
if _firmware_update_service is None:
_firmware_update_service = FirmwareUpdateService()
return _firmware_update_service
+54
View File
@@ -2526,3 +2526,57 @@ export const pendingUploadsApi = {
discardAll: () =>
request<{ discarded: number }>('/pending-uploads/discard-all', { method: 'DELETE' }),
};
// Firmware API Types
export interface FirmwareUpdateInfo {
printer_id: number;
printer_name: string;
model: string | null;
current_version: string | null;
latest_version: string | null;
update_available: boolean;
download_url: string | null;
release_notes: string | null;
}
export interface FirmwareUploadPrepare {
can_proceed: boolean;
sd_card_present: boolean;
sd_card_free_space: number;
firmware_size: number;
space_sufficient: boolean;
update_available: boolean;
current_version: string | null;
latest_version: string | null;
firmware_filename: string | null;
errors: string[];
}
export interface FirmwareUploadStatus {
status: 'idle' | 'preparing' | 'downloading' | 'uploading' | 'complete' | 'error';
progress: number;
message: string;
error: string | null;
firmware_filename: string | null;
firmware_version: string | null;
}
// Firmware API
export const firmwareApi = {
checkUpdates: () =>
request<{ updates: FirmwareUpdateInfo[]; updates_available: number }>('/firmware/updates'),
checkPrinterUpdate: (printerId: number) =>
request<FirmwareUpdateInfo>(`/firmware/updates/${printerId}`),
prepareUpload: (printerId: number) =>
request<FirmwareUploadPrepare>(`/firmware/updates/${printerId}/prepare`),
startUpload: (printerId: number) =>
request<{ started: boolean; message: string }>(`/firmware/updates/${printerId}/upload`, {
method: 'POST',
}),
getUploadStatus: (printerId: number) =>
request<FirmwareUploadStatus>(`/firmware/updates/${printerId}/upload/status`),
};
+232 -2
View File
@@ -38,6 +38,7 @@ import {
Wind,
AirVent,
Minus,
Download,
} from 'lucide-react';
// Custom Skip Objects icon - arrow jumping over boxes
@@ -53,8 +54,8 @@ const SkipObjectsIcon = ({ className }: { className?: string }) => (
</svg>
);
import { useNavigate } from 'react-router-dom';
import { api, discoveryApi } from '../api/client';
import type { Printer, PrinterCreate, AMSUnit, DiscoveredPrinter } from '../api/client';
import { api, discoveryApi, firmwareApi } from '../api/client';
import type { Printer, PrinterCreate, AMSUnit, DiscoveredPrinter, FirmwareUpdateInfo, FirmwareUploadStatus } from '../api/client';
import { Card, CardContent } from '../components/Card';
import { Button } from '../components/Button';
import { ConfirmModal } from '../components/ConfirmModal';
@@ -903,6 +904,7 @@ function PrinterCard({
trayUuid: string;
trayInfo: { type: string; color: string; location: string };
} | null>(null);
const [showFirmwareModal, setShowFirmwareModal] = useState(false);
const { data: status } = useQuery({
queryKey: ['printerStatus', printer.id],
@@ -910,6 +912,14 @@ function PrinterCard({
refetchInterval: 30000, // Fallback polling, WebSocket handles real-time
});
// Check for firmware updates (cached for 5 minutes)
const { data: firmwareInfo } = useQuery({
queryKey: ['firmwareUpdate', printer.id],
queryFn: () => firmwareApi.checkPrinterUpdate(printer.id),
staleTime: 5 * 60 * 1000,
refetchInterval: 5 * 60 * 1000,
});
// Collect unique tray_info_idx values for cloud filament info lookup
const trayInfoIds = useMemo(() => {
const ids = new Set<string>();
@@ -1391,6 +1401,17 @@ function PrinterCard({
{queueCount}
</button>
)}
{/* Firmware Update Badge */}
{firmwareInfo?.update_available && (
<button
onClick={() => setShowFirmwareModal(true)}
className="flex items-center gap-1 px-2 py-1 rounded-full text-xs bg-orange-500/20 text-orange-400 hover:opacity-80 transition-opacity"
title={`Firmware update available: ${firmwareInfo.current_version} → ${firmwareInfo.latest_version}`}
>
<Download className="w-3 h-3" />
Update
</button>
)}
</div>
)}
</div>
@@ -2697,6 +2718,15 @@ function PrinterCard({
/>
)}
{/* Firmware Update Modal */}
{showFirmwareModal && firmwareInfo && (
<FirmwareUpdateModal
printer={printer}
firmwareInfo={firmwareInfo}
onClose={() => setShowFirmwareModal(false)}
/>
)}
{/* AMS Slot Menu Backdrop - closes menu when clicking outside */}
{amsSlotMenu && (
<div
@@ -3087,6 +3117,206 @@ function AddPrinterModal({
);
}
function FirmwareUpdateModal({
printer,
firmwareInfo,
onClose,
}: {
printer: Printer;
firmwareInfo: FirmwareUpdateInfo;
onClose: () => void;
}) {
const queryClient = useQueryClient();
const { showToast } = useToast();
const [uploadStatus, setUploadStatus] = useState<FirmwareUploadStatus | null>(null);
const [isUploading, setIsUploading] = useState(false);
const [pollInterval, setPollInterval] = useState<NodeJS.Timeout | null>(null);
// Prepare check query
const { data: prepareInfo, isLoading: isPreparing } = useQuery({
queryKey: ['firmwarePrepare', printer.id],
queryFn: () => firmwareApi.prepareUpload(printer.id),
staleTime: 30000,
});
// Start upload mutation
const uploadMutation = useMutation({
mutationFn: () => firmwareApi.startUpload(printer.id),
onSuccess: () => {
setIsUploading(true);
// Start polling for status
const interval = setInterval(async () => {
try {
const status = await firmwareApi.getUploadStatus(printer.id);
setUploadStatus(status);
if (status.status === 'complete' || status.status === 'error') {
clearInterval(interval);
setPollInterval(null);
setIsUploading(false);
if (status.status === 'complete') {
showToast('Firmware uploaded! Trigger update from printer screen.', 'success');
queryClient.invalidateQueries({ queryKey: ['firmwareUpdate', printer.id] });
}
}
} catch {
// Ignore errors during polling
}
}, 2000);
setPollInterval(interval);
},
onError: (error: Error) => {
showToast(`Failed to start upload: ${error.message}`, 'error');
setIsUploading(false);
},
});
// Cleanup on unmount
useEffect(() => {
return () => {
if (pollInterval) clearInterval(pollInterval);
};
}, [pollInterval]);
const handleStartUpload = () => {
setUploadStatus(null);
uploadMutation.mutate();
};
return (
<div className="fixed inset-0 bg-black/50 flex items-center justify-center z-50">
<Card className="w-full max-w-md mx-4">
<CardContent>
<div className="flex items-start gap-3 mb-4">
<div className="p-2 rounded-full bg-orange-500/20">
<Download className="w-5 h-5 text-orange-400" />
</div>
<div className="flex-1">
<h3 className="text-lg font-semibold text-white">Firmware Update</h3>
<p className="text-sm text-bambu-gray mt-1">
{printer.name}
</p>
</div>
</div>
{/* Version Info */}
<div className="bg-bambu-dark rounded-lg p-3 mb-4">
<div className="flex justify-between items-center text-sm">
<span className="text-bambu-gray">Current:</span>
<span className="text-white font-mono">{firmwareInfo.current_version || 'Unknown'}</span>
</div>
<div className="flex justify-between items-center text-sm mt-1">
<span className="text-bambu-gray">Latest:</span>
<span className="text-orange-400 font-mono">{firmwareInfo.latest_version}</span>
</div>
{firmwareInfo.release_notes && (
<details className="mt-3 text-sm">
<summary className="text-orange-400 cursor-pointer hover:underline">
Release Notes
</summary>
<div className="mt-2 text-bambu-gray text-xs max-h-40 overflow-y-auto whitespace-pre-wrap">
{firmwareInfo.release_notes}
</div>
</details>
)}
</div>
{/* Status / Progress */}
{isPreparing ? (
<div className="flex items-center gap-2 text-bambu-gray text-sm mb-4">
<Loader2 className="w-4 h-4 animate-spin" />
Checking prerequisites...
</div>
) : prepareInfo && !isUploading && !uploadStatus ? (
<div className="mb-4">
{prepareInfo.can_proceed ? (
<div className="flex items-center gap-2 text-bambu-green text-sm">
<Box className="w-4 h-4" />
SD card ready. Click below to upload firmware.
</div>
) : (
<div className="space-y-1">
{prepareInfo.errors.map((error, i) => (
<div key={i} className="flex items-center gap-2 text-red-400 text-sm">
<AlertCircle className="w-4 h-4 flex-shrink-0" />
{error}
</div>
))}
</div>
)}
</div>
) : null}
{/* Upload Progress */}
{(isUploading || uploadStatus) && uploadStatus && (
<div className="mb-4">
<div className="flex items-center justify-between text-sm mb-1">
<span className="text-bambu-gray capitalize">{uploadStatus.status}</span>
<span className="text-white">{uploadStatus.progress}%</span>
</div>
<div className="w-full bg-bambu-dark-tertiary rounded-full h-2">
<div
className={`h-2 rounded-full transition-all ${
uploadStatus.status === 'error' ? 'bg-red-500' :
uploadStatus.status === 'complete' ? 'bg-bambu-green' : 'bg-orange-500'
} ${uploadStatus.status === 'uploading' ? 'animate-pulse' : ''}`}
style={{ width: `${uploadStatus.progress}%` }}
/>
</div>
<p className="text-xs text-bambu-gray mt-1">{uploadStatus.message}</p>
{uploadStatus.error && (
<p className="text-xs text-red-400 mt-1">{uploadStatus.error}</p>
)}
</div>
)}
{/* Success Message */}
{uploadStatus?.status === 'complete' && (
<div className="bg-bambu-green/10 border border-bambu-green/30 rounded-lg p-3 mb-4">
<p className="text-sm text-bambu-green font-medium mb-2">
Firmware uploaded to SD card!
</p>
<p className="text-xs text-bambu-gray">
To apply the update on your printer:
</p>
<ol className="text-xs text-bambu-gray mt-1 list-decimal list-inside space-y-1">
<li>On the printer's touchscreen, go to <strong className="text-white">Settings</strong></li>
<li>Navigate to <strong className="text-white">Firmware</strong></li>
<li>Select <strong className="text-white">Update from SD card</strong></li>
<li>The update will take 10-20 minutes</li>
</ol>
</div>
)}
{/* Buttons */}
<div className="flex gap-2 justify-end">
<Button variant="secondary" onClick={onClose}>
{uploadStatus?.status === 'complete' ? 'Done' : 'Cancel'}
</Button>
{prepareInfo?.can_proceed && !isUploading && uploadStatus?.status !== 'complete' && (
<Button
onClick={handleStartUpload}
disabled={uploadMutation.isPending}
>
{uploadMutation.isPending ? (
<>
<Loader2 className="w-4 h-4 animate-spin mr-2" />
Starting...
</>
) : (
<>
<Download className="w-4 h-4 mr-2" />
Upload Firmware
</>
)}
</Button>
)}
</div>
</CardContent>
</Card>
</div>
);
}
function EditPrinterModal({
printer,
onClose,
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
+2 -2
View File
@@ -23,8 +23,8 @@
<!-- Splash screens for iOS -->
<link rel="apple-touch-startup-image" href="/img/android-chrome-512x512.png" />
<script type="module" crossorigin src="/assets/index-BaxJ1N11.js"></script>
<link rel="stylesheet" crossorigin href="/assets/index-Ds1sabci.css">
<script type="module" crossorigin src="/assets/index-DPwHFDrf.js"></script>
<link rel="stylesheet" crossorigin href="/assets/index-DYSmHndB.css">
</head>
<body>
<div id="root"></div>