mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-06 06:02:05 +02:00
maziggy
d336bb2ad2
Groups can be limited to printers and whole locations, managed on the new Printer access page, and jobs can wait for staff review before they print. Merging onto the current dev: - The Printer Locations page keeps to the same access rules: a rename keeps its groups' access, deleting a granted location or moving printers into or out of one needs an admin and drops the stale grant, and a user limited to some printers sees and changes only their own locations. - The overlay logo checks its token without a printer, which the stricter overlay check now needs and the logo route doesn't have.
367 lines
12 KiB
Python
367 lines
12 KiB
Python
"""
|
|
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.auth import (
|
|
RequestPrinterScope,
|
|
RequirePermissionIfAuthEnabled,
|
|
RequirePrinterPermissionIfAuthEnabled,
|
|
)
|
|
from backend.app.core.database import get_db
|
|
from backend.app.core.permissions import Permission
|
|
from backend.app.core.printer_scope import PrinterScope
|
|
from backend.app.models.printer import Printer
|
|
from backend.app.models.settings import Settings
|
|
from backend.app.models.user import User
|
|
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 AvailableFirmwareVersion(BaseModel):
|
|
"""A single firmware version announced by Bambu Lab."""
|
|
|
|
version: str
|
|
file_available: bool
|
|
download_url: str | None = None
|
|
release_notes: str | None = None
|
|
release_time: str | None = None
|
|
|
|
|
|
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
|
|
available_versions: list[AvailableFirmwareVersion] = Field(default_factory=list)
|
|
|
|
|
|
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
|
|
|
|
|
|
async def _checks_enabled(db: AsyncSession) -> bool:
|
|
"""The "Check printer firmware" setting (default on).
|
|
|
|
Enforced here and not only in the UI: the printers page asks before its
|
|
settings have loaded, so a check switched off still went out to bambulab.com
|
|
once per printer on every page load.
|
|
"""
|
|
value = (
|
|
await db.execute(select(Settings.value).where(Settings.key == "check_printer_firmware"))
|
|
).scalar_one_or_none()
|
|
return (value or "true").lower() != "false"
|
|
|
|
|
|
def _current_version(printer_id: int) -> str | None:
|
|
mqtt_client = printer_manager.get_client(printer_id)
|
|
if mqtt_client and mqtt_client.state:
|
|
return mqtt_client.state.firmware_version
|
|
return None
|
|
|
|
|
|
def _not_checked(printer: Printer) -> FirmwareUpdateInfo:
|
|
return FirmwareUpdateInfo(
|
|
printer_id=printer.id,
|
|
printer_name=printer.name,
|
|
model=printer.model or "Unknown",
|
|
current_version=_current_version(printer.id),
|
|
latest_version=None,
|
|
update_available=False,
|
|
)
|
|
|
|
|
|
@router.get("/updates", response_model=FirmwareUpdatesResponse)
|
|
async def check_firmware_updates(
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User | None = RequirePermissionIfAuthEnabled(Permission.FIRMWARE_READ),
|
|
printer_scope: PrinterScope = RequestPrinterScope,
|
|
):
|
|
"""
|
|
Check for firmware updates for the connected printers the caller may see.
|
|
|
|
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 = [p for p in result.scalars().all() if printer_scope.allows(p.id)]
|
|
|
|
if not await _checks_enabled(db):
|
|
return FirmwareUpdatesResponse(updates=[_not_checked(p) for p in printers], updates_available=0)
|
|
|
|
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"],
|
|
available_versions=[AvailableFirmwareVersion(**v) for v in update_info.get("available_versions", [])],
|
|
)
|
|
)
|
|
|
|
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),
|
|
_: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.FIRMWARE_READ),
|
|
):
|
|
"""
|
|
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")
|
|
|
|
if not await _checks_enabled(db):
|
|
return _not_checked(printer)
|
|
|
|
current_version = _current_version(printer.id)
|
|
|
|
# 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"],
|
|
available_versions=[AvailableFirmwareVersion(**v) for v in update_info.get("available_versions", [])],
|
|
)
|
|
|
|
|
|
@router.get("/latest", response_model=list[LatestFirmwareInfo])
|
|
async def get_all_latest_firmware(
|
|
_: User | None = RequirePermissionIfAuthEnabled(Permission.FIRMWARE_READ),
|
|
):
|
|
"""
|
|
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
|
|
target_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,
|
|
version: str | None = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.FIRMWARE_READ),
|
|
):
|
|
"""
|
|
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, target_version=version)
|
|
return FirmwareUploadPrepareResponse(**result)
|
|
|
|
|
|
@router.post("/updates/{printer_id}/upload", response_model=FirmwareUploadStartResponse)
|
|
async def start_firmware_upload(
|
|
printer_id: int,
|
|
version: str | None = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.FIRMWARE_UPDATE),
|
|
):
|
|
"""
|
|
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, target_version=version)
|
|
|
|
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, target_version=version)
|
|
|
|
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,
|
|
_: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.FIRMWARE_READ),
|
|
):
|
|
"""
|
|
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,
|
|
)
|