public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-17 11:25:48 +02:00
Code Issues Packages Projects Releases Wiki Activity

- Refactored printer card's AMS section for better experience

Browse Source
This commit is contained in:
maziggy
2026-01-01 12:08:55 +01:00
parent 569dbc3d21
commit ccdb084f23
13 changed files with 2737 additions and 678 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/routes/cloud.py
+104 -29
View File
@@ -5,33 +5,36 @@ Handles authentication and profile management with Bambu Cloud.
"""
import json
import logging
from pathlib import Path
from typing import Literal
from fastapi import APIRouter, HTTPException, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import APIRouter, Body, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from backend.app.core.database import get_db
from backend.app.models.settings import Settings
from backend.app.services.bambu_cloud import (
get_cloud_service,
BambuCloudError,
BambuCloudAuthError,
)
from backend.app.schemas.cloud import (
CloudLoginRequest,
CloudVerifyRequest,
CloudLoginResponse,
CloudAuthStatus,
CloudTokenRequest,
SlicerSettingsResponse,
SlicerSetting,
CloudDevice,
CloudLoginRequest,
CloudLoginResponse,
CloudTokenRequest,
CloudVerifyRequest,
SlicerSetting,
SlicerSettingCreate,
SlicerSettingUpdate,
SlicerSettingDeleteResponse,
SlicerSettingsResponse,
SlicerSettingUpdate,
)
from backend.app.services.bambu_cloud import (
BambuCloudAuthError,
BambuCloudError,
get_cloud_service,
)
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/cloud", tags=["cloud"])
@@ -43,9 +46,7 @@ CLOUD_EMAIL_KEY = "bambu_cloud_email"
async def get_stored_token(db: AsyncSession) -> tuple[str | None, str | None]:
"""Get stored cloud token and email from database."""
result = await db.execute(
select(Settings).where(Settings.key.in_([CLOUD_TOKEN_KEY, CLOUD_EMAIL_KEY]))
)
result = await db.execute(select(Settings).where(Settings.key.in_([CLOUD_TOKEN_KEY, CLOUD_EMAIL_KEY])))
settings = {s.key: s.value for s in result.scalars().all()}
return settings.get(CLOUD_TOKEN_KEY), settings.get(CLOUD_EMAIL_KEY)
@@ -64,9 +65,7 @@ async def store_token(db: AsyncSession, token: str, email: str) -> None:
async def clear_token(db: AsyncSession) -> None:
"""Clear stored cloud token and email."""
result = await db.execute(
select(Settings).where(Settings.key.in_([CLOUD_TOKEN_KEY, CLOUD_EMAIL_KEY]))
)
result = await db.execute(select(Settings).where(Settings.key.in_([CLOUD_TOKEN_KEY, CLOUD_EMAIL_KEY])))
for setting in result.scalars().all():
await db.delete(setting)
await db.commit()
@@ -213,14 +212,16 @@ async def get_slicer_settings(
parsed = []
for s in all_settings:
parsed.append(SlicerSetting(
setting_id=s.get("setting_id", s.get("id", "")),
name=s.get("name", "Unknown"),
type=our_type,
version=s.get("version"),
user_id=s.get("user_id"),
updated_time=s.get("updated_time"),
))
parsed.append(
SlicerSetting(
setting_id=s.get("setting_id", s.get("id", "")),
name=s.get("name", "Unknown"),
type=our_type,
version=s.get("version"),
user_id=s.get("user_id"),
updated_time=s.get("updated_time"),
)
)
setattr(result, our_type, parsed)
return result
@@ -258,6 +259,80 @@ async def get_setting_detail(setting_id: str, db: AsyncSession = Depends(get_db)
raise HTTPException(status_code=500, detail=str(e))
# Cache for filament preset info (setting_id -> {name, k})
_filament_cache: dict[str, dict] = {}
_filament_cache_time: float = 0
FILAMENT_CACHE_TTL = 300 # 5 minutes
@router.post("/filament-info")
async def get_filament_info(setting_ids: list[str] = Body(...), db: AsyncSession = Depends(get_db)):
"""
Get filament preset info (name and K value) for multiple setting IDs.
Used to enrich AMS tray tooltips with cloud preset data.
"""
import time
logger.info(f"get_filament_info called with {len(setting_ids)} IDs: {setting_ids}")
global _filament_cache, _filament_cache_time
# Clear stale cache
if time.time() - _filament_cache_time > FILAMENT_CACHE_TTL:
_filament_cache = {}
_filament_cache_time = time.time()
token, _ = await get_stored_token(db)
if not token:
logger.info("get_filament_info: Not authenticated, returning empty")
# Return empty results if not authenticated (graceful degradation)
return {}
cloud = get_cloud_service()
cloud.set_token(token)
if not cloud.is_authenticated:
return {}
result = {}
for setting_id in setting_ids:
if not setting_id:
continue
# Check cache first
if setting_id in _filament_cache:
result[setting_id] = _filament_cache[setting_id]
continue
try:
data = await cloud.get_setting_detail(setting_id)
setting = data.get("setting", {})
# Extract name (e.g., "Bambu PLA Basic Jade White")
name = data.get("name", "")
# Extract K value (pressure_advance)
k_value = setting.get("pressure_advance")
if k_value is not None:
try:
k_value = float(k_value)
except (ValueError, TypeError):
k_value = None
info = {"name": name, "k": k_value}
_filament_cache[setting_id] = info
result[setting_id] = info
except Exception as e:
logger.warning(f"Failed to get cloud preset {setting_id}: {e}")
# Cache the failure to avoid repeated requests
_filament_cache[setting_id] = {"name": "", "k": None}
result[setting_id] = {"name": "", "k": None}
return result
@router.get("/devices", response_model=list[CloudDevice])
async def get_devices(db: AsyncSession = Depends(get_db)):
"""
@@ -425,7 +500,7 @@ def _load_fields(preset_type: str) -> dict:
if not file_path.exists():
raise HTTPException(status_code=404, detail=f"Field definitions not found for: {preset_type}")
with open(file_path, "r") as f:
with open(file_path) as f:
data = json.load(f)
_fields_cache[preset_type] = data
backend/app/api/routes/printers.py
+31 -3
View File
@@ -169,6 +169,16 @@ async def get_printer_status(printer_id: int, db: AsyncSession = Depends(get_db)
ams_exists = False
raw_data = state.raw_data or {}
# Build K-profile lookup map: cali_idx -> k_value
# This allows looking up the calibrated K value for each AMS slot
kprofile_map: dict[int, float] = {}
for kp in state.kprofiles or []:
if kp.slot_id is not None and kp.k_value:
try:
kprofile_map[kp.slot_id] = float(kp.k_value)
except (ValueError, TypeError):
pass
if "ams" in raw_data and isinstance(raw_data["ams"], list):
ams_exists = True
for ams_data in raw_data["ams"]:
@@ -184,6 +194,13 @@ async def get_printer_status(printer_id: int, db: AsyncSession = Depends(get_db)
tray_uuid = tray_data.get("tray_uuid", "")
if tray_uuid in ("", "00000000000000000000000000000000"):
tray_uuid = None
# Get K value: first try tray's k field, then lookup from K-profiles
k_value = tray_data.get("k")
cali_idx = tray_data.get("cali_idx")
if k_value is None and cali_idx is not None and cali_idx in kprofile_map:
k_value = kprofile_map[cali_idx]
trays.append(
AMSTray(
id=tray_data.get("id", 0),
@@ -193,7 +210,8 @@ async def get_printer_status(printer_id: int, db: AsyncSession = Depends(get_db)
tray_id_name=tray_data.get("tray_id_name"),
tray_info_idx=tray_data.get("tray_info_idx"),
remain=tray_data.get("remain", 0),
k=tray_data.get("k"),
k=k_value,
cali_idx=cali_idx,
tag_uid=tag_uid,
tray_uuid=tray_uuid,
nozzle_temp_min=tray_data.get("nozzle_temp_min"),
@@ -239,13 +257,23 @@ async def get_printer_status(printer_id: int, db: AsyncSession = Depends(get_db)
vt_tray_uuid = vt_data.get("tray_uuid", "")
if vt_tray_uuid in ("", "00000000000000000000000000000000"):
vt_tray_uuid = None
# Get K value: first try tray's k field, then lookup from K-profiles
vt_k_value = vt_data.get("k")
vt_cali_idx = vt_data.get("cali_idx")
if vt_k_value is None and vt_cali_idx is not None and vt_cali_idx in kprofile_map:
vt_k_value = kprofile_map[vt_cali_idx]
vt_tray = AMSTray(
id=254, # Virtual tray ID
tray_color=vt_data.get("tray_color"),
tray_type=vt_data.get("tray_type"),
tray_sub_brands=vt_data.get("tray_sub_brands"),
tray_id_name=vt_data.get("tray_id_name"),
tray_info_idx=vt_data.get("tray_info_idx"),
remain=vt_data.get("remain", 0),
k=vt_data.get("k"),
k=vt_k_value,
cali_idx=vt_cali_idx,
tag_uid=vt_tag_uid,
tray_uuid=vt_tray_uuid,
nozzle_temp_min=vt_data.get("nozzle_temp_min"),
@@ -977,7 +1005,7 @@ async def debug_simulate_print_complete(
"timelapse_was_active": False,
}
logger.info(f"[DEBUG] Simulating print complete for printer {printer_id}, archive {archive.id}")
logger.info(f"Simulating print complete for printer {printer_id}, archive {archive.id}")
# Call the actual on_print_complete handler
await on_print_complete(printer_id, data)