mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
maziggy
76fd8c5c48
The firmware check only mapped display names (e.g., "H2D Pro") to API keys but not raw SSDP device codes (e.g., "O1E", "O2D"). If a printer's model was stored as the raw SSDP code, the firmware check either failed or matched the wrong model — H2D Pro matched against H2D firmware, showing a false update-available badge. Add all known SSDP device codes to MODEL_TO_API_KEY so firmware versions resolve correctly regardless of how the model was stored.
451 lines
15 KiB
Python
451 lines
15 KiB
Python
"""
|
|
Firmware Check Service
|
|
|
|
Checks for firmware updates by fetching from Bambu Lab's official wiki and firmware
|
|
download page. The wiki is used as the primary version source (always up-to-date),
|
|
while the download page provides firmware file URLs 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 (for download URLs)
|
|
BAMBU_FIRMWARE_BASE = "https://bambulab.com"
|
|
FIRMWARE_PAGE = "/en/support/firmware-download/all"
|
|
|
|
# Bambu Lab wiki (primary source for latest version detection)
|
|
BAMBU_WIKI_BASE = "https://wiki.bambulab.com"
|
|
|
|
# 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": "h2c",
|
|
"H2S": "h2s",
|
|
"P2S": "p2s",
|
|
"X1E": "x1e",
|
|
"H2D Pro": "h2d-pro",
|
|
"H2D-Pro": "h2d-pro",
|
|
"H2DPRO": "h2d-pro",
|
|
# SSDP model codes (DevModel header) — in case raw codes are stored
|
|
"O1D": "h2d",
|
|
"O1E": "h2d-pro",
|
|
"O2D": "h2d-pro",
|
|
"O1C": "h2c",
|
|
"O1C2": "h2c",
|
|
"O1S": "h2s",
|
|
"BL-P001": "x1",
|
|
"BL-P002": "x1",
|
|
"BL-P003": "x1e",
|
|
"C11": "p1",
|
|
"C12": "p1",
|
|
"C13": "p2s",
|
|
"N2S": "a1",
|
|
"N1": "a1-mini",
|
|
"N7": "p2s",
|
|
}
|
|
|
|
# Reverse mapping: API key to model codes
|
|
API_KEY_TO_DEV_MODEL = {
|
|
"x1": "BL-P001",
|
|
"p1": "C11",
|
|
"a1": "N2S",
|
|
"a1-mini": "N1",
|
|
"h2d": "O1D",
|
|
"h2c": "O1C",
|
|
"h2s": "O1S",
|
|
"p2s": "N7",
|
|
"x1e": "C13",
|
|
"h2d-pro": "O1E",
|
|
}
|
|
|
|
# Wiki firmware release history pages (primary version source)
|
|
API_KEY_TO_WIKI_PATH = {
|
|
"x1": "/en/x1/manual/X1-X1C-firmware-release-history",
|
|
"x1e": "/en/x1/manual/X1E-firmware-release-history",
|
|
"p1": "/en/p1/manual/p1p-firmware-release-history",
|
|
"a1": "/en/a1/manual/a1-firmware-release-history",
|
|
"a1-mini": "/en/a1-mini/manual/a1-mini-firmware-release-history",
|
|
"h2d": "/en/h2d/manual/h2d-firmware-release-history",
|
|
"h2c": "/en/h2c/manual/h2c-firmware-release-history",
|
|
"h2s": "/en/h2s/manual/h2s-firmware-release-history",
|
|
"p2s": "/en/p2s/manual/p2s-firmware-release-history",
|
|
"h2d-pro": "/en/h2d-pro/manual/firmware-release-history",
|
|
}
|
|
|
|
|
|
@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("Got Bambu Lab build ID: %s", self._build_id)
|
|
return self._build_id
|
|
logger.warning("Failed to get Bambu Lab page: %s", response.status_code)
|
|
except Exception as e:
|
|
logger.error("Error fetching Bambu Lab build ID: %s", e)
|
|
|
|
return self._build_id # Return cached value if available
|
|
|
|
async def _fetch_version_from_wiki(self, api_key: str) -> str | None:
|
|
"""Fetch the latest firmware version from Bambu Lab's wiki release history page."""
|
|
wiki_path = API_KEY_TO_WIKI_PATH.get(api_key)
|
|
if not wiki_path:
|
|
return None
|
|
|
|
try:
|
|
url = f"{BAMBU_WIKI_BASE}{wiki_path}"
|
|
response = await self._client.get(url, follow_redirects=True)
|
|
|
|
if response.status_code == 200:
|
|
# Extract version strings (format: XX.XX.XX.XX), first match is the latest
|
|
versions = re.findall(r"(\d{2}\.\d{2}\.\d{2}\.\d{2})", response.text)
|
|
if versions:
|
|
logger.debug("Wiki firmware for %s: %s", api_key, versions[0])
|
|
return versions[0]
|
|
else:
|
|
logger.debug("Wiki firmware page for %s returned %s", api_key, response.status_code)
|
|
|
|
except Exception as e:
|
|
logger.debug("Error fetching wiki firmware for %s: %s", api_key, e)
|
|
|
|
return None
|
|
|
|
async def _fetch_from_download_page(self, api_key: str) -> FirmwareVersion | None:
|
|
"""Fetch firmware info from Bambu Lab's download page (has download URLs)."""
|
|
build_id = await self._get_build_id()
|
|
if not build_id:
|
|
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"),
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.debug("Error fetching download page firmware for %s: %s", api_key, e)
|
|
|
|
return None
|
|
|
|
async def _fetch_firmware_versions(self, api_key: str) -> FirmwareVersion | None:
|
|
"""Fetch firmware version info, using wiki as primary source and download page as fallback."""
|
|
# Try wiki first (always has the latest version)
|
|
wiki_version = await self._fetch_version_from_wiki(api_key)
|
|
|
|
# Try download page (has download URLs, may lag behind wiki)
|
|
download_info = await self._fetch_from_download_page(api_key)
|
|
|
|
if wiki_version:
|
|
# Wiki has the latest version — use it, attach download URL if available
|
|
download_url = ""
|
|
release_notes = None
|
|
if download_info and download_info.version == wiki_version:
|
|
download_url = download_info.download_url
|
|
release_notes = download_info.release_notes
|
|
return FirmwareVersion(
|
|
version=wiki_version,
|
|
download_url=download_url,
|
|
release_notes=release_notes,
|
|
)
|
|
|
|
if download_info:
|
|
return download_info
|
|
|
|
logger.warning("Could not fetch firmware info for %s from wiki or download page", api_key)
|
|
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("Unknown printer model: %s", 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("Could not compare versions: %s vs %s", current_version, 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
|
|
|
|
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("No firmware download URL available for model: %s", model)
|
|
return None
|
|
|
|
# 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"
|
|
|
|
# Check if already cached (using original filename so SD card gets the right name)
|
|
cached_path = self._get_firmware_cache_dir() / original_filename
|
|
if cached_path.exists():
|
|
logger.info("Using cached firmware: %s", cached_path)
|
|
return cached_path
|
|
|
|
# Download to temp file first
|
|
temp_path = self._get_firmware_cache_dir() / f".downloading_{original_filename}"
|
|
|
|
try:
|
|
logger.info("Downloading firmware from %s", 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("Firmware download failed with status %s", 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...")
|
|
|
|
# Move temp to final path, preserving original filename
|
|
temp_path.rename(cached_path)
|
|
|
|
logger.info("Firmware downloaded successfully: %s", cached_path)
|
|
if progress_callback:
|
|
progress_callback(downloaded, total_size, "Download complete")
|
|
|
|
return cached_path
|
|
|
|
except Exception as e:
|
|
logger.error("Firmware download failed: %s", e)
|
|
if temp_path.exists():
|
|
try:
|
|
temp_path.unlink()
|
|
except OSError:
|
|
pass # Best-effort cleanup of failed download temp file
|
|
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
|