import asyncio import logging from contextlib import asynccontextmanager from datetime import datetime, timedelta from logging.handlers import RotatingFileHandler from fastapi import FastAPI # Import settings first for logging configuration from backend.app.core.config import APP_VERSION, settings as app_settings # Configure logging based on settings # DEBUG=true -> DEBUG level, else use LOG_LEVEL setting log_level_str = "DEBUG" if app_settings.debug else app_settings.log_level.upper() log_level = getattr(logging, log_level_str, logging.INFO) log_format = "%(asctime)s %(levelname)s [%(name)s] %(message)s" # Create root logger root_logger = logging.getLogger() root_logger.setLevel(log_level) # Console handler - always enabled console_handler = logging.StreamHandler() console_handler.setLevel(log_level) console_handler.setFormatter(logging.Formatter(log_format)) root_logger.addHandler(console_handler) # File handler - only in production or if explicitly enabled if app_settings.log_to_file: log_file = app_settings.log_dir / "bambuddy.log" file_handler = RotatingFileHandler( log_file, maxBytes=5 * 1024 * 1024, # 5MB backupCount=3, encoding="utf-8", ) file_handler.setLevel(log_level) file_handler.setFormatter(logging.Formatter(log_format)) root_logger.addHandler(file_handler) logging.info(f"Logging to file: {log_file}") # Reduce noise from third-party libraries in production if not app_settings.debug: logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING) logging.getLogger("httpcore").setLevel(logging.WARNING) logging.getLogger("httpx").setLevel(logging.WARNING) logging.info(f"Bambuddy starting - debug={app_settings.debug}, log_level={log_level_str}") from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles from sqlalchemy import delete, or_, select from backend.app.api.routes import ( ams_history, api_keys, archives, camera, cloud, discovery, external_links, filaments, kprofiles, maintenance, notification_templates, notifications, pending_uploads, print_queue, printers, projects, settings as settings_routes, smart_plugs, spoolman, system, updates, webhook, websocket, ) from backend.app.api.routes.maintenance import _get_printer_maintenance_internal, ensure_default_types from backend.app.core.database import async_session, init_db from backend.app.core.websocket import ws_manager from backend.app.models.smart_plug import SmartPlug from backend.app.services.archive import ArchiveService from backend.app.services.bambu_ftp import download_file_async from backend.app.services.bambu_mqtt import PrinterState from backend.app.services.notification_service import notification_service from backend.app.services.print_scheduler import scheduler as print_scheduler from backend.app.services.printer_manager import ( init_printer_connections, printer_manager, printer_state_to_dict, ) from backend.app.services.smart_plug_manager import smart_plug_manager from backend.app.services.spoolman import close_spoolman_client, get_spoolman_client, init_spoolman_client from backend.app.services.tasmota import tasmota_service from backend.app.services.telemetry import start_telemetry_loop # Track active prints: {(printer_id, filename): archive_id} _active_prints: dict[tuple[int, str], int] = {} # Track expected prints from reprint/scheduled (skip auto-archiving for these) # {(printer_id, filename): archive_id} _expected_prints: dict[tuple[int, str], int] = {} # Track starting energy for prints: {archive_id: starting_kwh} _print_energy_start: dict[int, float] = {} def register_expected_print(printer_id: int, filename: str, archive_id: int): """Register an expected print from reprint/scheduled so we don't create duplicate archives.""" # Store with multiple filename variations to catch different naming patterns _expected_prints[(printer_id, filename)] = archive_id # Also store without .3mf extension if present if filename.endswith(".3mf"): base = filename[:-4] _expected_prints[(printer_id, base)] = archive_id _expected_prints[(printer_id, f"{base}.gcode")] = archive_id logging.getLogger(__name__).info( f"Registered expected print: printer={printer_id}, file={filename}, archive={archive_id}" ) _last_status_broadcast: dict[int, str] = {} _nozzle_count_updated: set[int] = set() # Track printers where we've updated nozzle_count async def _report_spoolman_usage(printer_id: int, archive_id: int, logger): """Report filament usage to Spoolman after print completion. This finds the spool by RFID tag_uid from current AMS state and reports the filament_used_grams from the archive metadata. """ async with async_session() as db: from backend.app.api.routes.settings import get_setting from backend.app.models.archive import PrintArchive # Check if Spoolman is enabled spoolman_enabled = await get_setting(db, "spoolman_enabled") if not spoolman_enabled or spoolman_enabled.lower() != "true": return # Get Spoolman URL spoolman_url = await get_setting(db, "spoolman_url") if not spoolman_url: return # Get or create Spoolman client client = await get_spoolman_client() if not client: client = await init_spoolman_client(spoolman_url) # Check if Spoolman is reachable if not await client.health_check(): logger.warning("Spoolman not reachable for usage reporting") return # Get archive to find filament usage result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id)) archive = result.scalar_one_or_none() if not archive or not archive.filament_used_grams: logger.debug(f"No filament usage data for archive {archive_id}") return filament_used = archive.filament_used_grams logger.info(f"[SPOOLMAN] Archive {archive_id} used {filament_used}g of filament") # Get current AMS state from printer to find the active spool state = printer_manager.get_status(printer_id) if not state or not state.raw_data: logger.debug("No printer state available for usage reporting") return ams_data = state.raw_data.get("ams") if not ams_data: logger.debug("No AMS data available for usage reporting") return # Find spools with RFID tags in Spoolman and report usage # For now, we report usage to the first spool found with a matching tag # TODO: In future, track which specific trays were used during the print spools_updated = 0 for ams_unit in ams_data: trays = ams_unit.get("tray", []) for tray_data in trays: tag_uid = tray_data.get("tag_uid") if not tag_uid: continue # Find spool in Spoolman by tag spool = await client.find_spool_by_tag(tag_uid) if spool: # Report usage to Spoolman result = await client.use_spool(spool["id"], filament_used) if result: logger.info( f"[SPOOLMAN] Reported {filament_used}g usage to spool {spool['id']} (tag: {tag_uid})" ) spools_updated += 1 # Only report to one spool for single-material prints # Multi-material prints would need more sophisticated tracking return if spools_updated == 0: logger.debug(f"No matching Spoolman spools found for printer {printer_id}") async def on_printer_status_change(printer_id: int, state: PrinterState): """Handle printer status changes - broadcast via WebSocket.""" # Only broadcast if something meaningful changed (reduce WebSocket spam) # Include rounded temperatures to detect meaningful temp changes (within 1 degree) temps = state.temperatures or {} nozzle_temp = round(temps.get("nozzle", 0)) bed_temp = round(temps.get("bed", 0)) nozzle_2_temp = round(temps.get("nozzle_2", 0)) if "nozzle_2" in temps else "" chamber_temp = round(temps.get("chamber", 0)) if "chamber" in temps else "" # Auto-detect dual-nozzle printers from MQTT temperature data if "nozzle_2" in temps and printer_id not in _nozzle_count_updated: _nozzle_count_updated.add(printer_id) # Update nozzle_count in database async with async_session() as db: from backend.app.models.printer import Printer result = await db.execute(select(Printer).where(Printer.id == printer_id)) printer = result.scalar_one_or_none() if printer and printer.nozzle_count != 2: printer.nozzle_count = 2 await db.commit() logging.getLogger(__name__).info( f"Auto-detected dual-nozzle printer {printer_id}, updated nozzle_count=2" ) # Include target temps for heating phase detection bed_target = round(temps.get("bed_target", 0)) nozzle_target = round(temps.get("nozzle_target", 0)) status_key = ( f"{state.connected}:{state.state}:{state.progress}:{state.layer_num}:" f"{nozzle_temp}:{bed_temp}:{nozzle_2_temp}:{chamber_temp}:" f"{state.stg_cur}:{bed_target}:{nozzle_target}:" f"{state.cooling_fan_speed}:{state.big_fan1_speed}:{state.big_fan2_speed}" ) if _last_status_broadcast.get(printer_id) == status_key: return # No change, skip broadcast _last_status_broadcast[printer_id] = status_key await ws_manager.send_printer_status( printer_id, printer_state_to_dict(state, printer_id), ) async def on_ams_change(printer_id: int, ams_data: list): """Handle AMS data changes - sync to Spoolman if enabled and auto mode.""" import logging logger = logging.getLogger(__name__) try: async with async_session() as db: from backend.app.api.routes.settings import get_setting from backend.app.models.printer import Printer # Check if Spoolman is enabled spoolman_enabled = await get_setting(db, "spoolman_enabled") if not spoolman_enabled or spoolman_enabled.lower() != "true": return # Check sync mode sync_mode = await get_setting(db, "spoolman_sync_mode") if sync_mode and sync_mode != "auto": return # Only sync on auto mode # Get Spoolman URL spoolman_url = await get_setting(db, "spoolman_url") if not spoolman_url: return # Get or create Spoolman client client = await get_spoolman_client() if not client: client = await init_spoolman_client(spoolman_url) # Check if Spoolman is reachable if not await client.health_check(): logger.warning(f"Spoolman not reachable at {spoolman_url}") return # Get printer name for location result = await db.execute(select(Printer).where(Printer.id == printer_id)) printer = result.scalar_one_or_none() printer_name = printer.name if printer else f"Printer {printer_id}" # Sync each AMS tray synced = 0 for ams_unit in ams_data: ams_id = int(ams_unit.get("id", 0)) trays = ams_unit.get("tray", []) for tray_data in trays: tray = client.parse_ams_tray(ams_id, tray_data) if not tray: continue # Empty tray try: result = await client.sync_ams_tray(tray, printer_name) if result: synced += 1 except Exception as e: logger.error(f"Error syncing AMS {ams_id} tray {tray.tray_id}: {e}") if synced > 0: logger.info(f"Auto-synced {synced} AMS trays to Spoolman for printer {printer_id}") except Exception as e: import logging logging.getLogger(__name__).warning(f"Spoolman AMS sync failed: {e}") async def _send_print_start_notification( printer_id: int, data: dict, archive_data: dict | None = None, logger=None, ): """Helper to send print start notification with optional archive data.""" if logger is None: import logging logger = logging.getLogger(__name__) try: async with async_session() as db: from backend.app.models.printer import Printer result = await db.execute(select(Printer).where(Printer.id == printer_id)) printer = result.scalar_one_or_none() printer_name = printer.name if printer else f"Printer {printer_id}" await notification_service.on_print_start(printer_id, printer_name, data, db, archive_data=archive_data) except Exception as e: logger.warning(f"Notification on_print_start failed: {e}") def _load_objects_from_archive(archive, printer_id: int, logger) -> None: """Extract printable objects from an archive's 3MF file and store in printer state.""" try: from backend.app.services.archive import extract_printable_objects_from_3mf file_path = app_settings.base_dir / archive.file_path if file_path.exists() and str(file_path).endswith(".3mf"): with open(file_path, "rb") as f: threemf_data = f.read() # Extract with positions for UI overlay printable_objects, bbox_all = extract_printable_objects_from_3mf(threemf_data, include_positions=True) if printable_objects: client = printer_manager.get_client(printer_id) if client: client.state.printable_objects = printable_objects client.state.printable_objects_bbox_all = bbox_all client.state.skipped_objects = [] logger.info(f"Loaded {len(printable_objects)} printable objects for printer {printer_id}") except Exception as e: logger.debug(f"Failed to extract printable objects from archive: {e}") async def on_print_start(printer_id: int, data: dict): """Handle print start - archive the 3MF file immediately.""" import logging logger = logging.getLogger(__name__) logger.info(f"[CALLBACK] on_print_start called for printer {printer_id}, data keys: {list(data.keys())}") await ws_manager.send_print_start(printer_id, data) # Track if notification was sent (to avoid sending twice) notification_sent = False # Smart plug automation: turn on plug when print starts try: async with async_session() as db: await smart_plug_manager.on_print_start(printer_id, db) except Exception as e: logger.warning(f"Smart plug on_print_start failed: {e}") async with async_session() as db: from backend.app.models.printer import Printer from backend.app.services.bambu_ftp import list_files_async result = await db.execute(select(Printer).where(Printer.id == printer_id)) printer = result.scalar_one_or_none() if not printer or not printer.auto_archive: # Send notification without archive data (auto-archive disabled) logger.info( f"[CALLBACK] Skipping archive - printer: {printer is not None}, auto_archive: {printer.auto_archive if printer else 'N/A'}" ) if not notification_sent: await _send_print_start_notification(printer_id, data, logger=logger) return # Get the filename and subtask_name filename = data.get("filename", "") subtask_name = data.get("subtask_name", "") logger.info(f"[CALLBACK] Print start detected - filename: {filename}, subtask: {subtask_name}") if not filename and not subtask_name: # Send notification without archive data (no filename) logger.info("[CALLBACK] Skipping archive - no filename or subtask_name") if not notification_sent: await _send_print_start_notification(printer_id, data, logger=logger) return # Check if this is an expected print from reprint/scheduled # Build list of possible keys to check expected_keys = [] if subtask_name: expected_keys.append((printer_id, subtask_name)) expected_keys.append((printer_id, f"{subtask_name}.3mf")) expected_keys.append((printer_id, f"{subtask_name}.gcode.3mf")) if filename: fname = filename.split("/")[-1] if "/" in filename else filename expected_keys.append((printer_id, fname)) # Strip extensions to match base = fname.replace(".gcode", "").replace(".3mf", "") expected_keys.append((printer_id, base)) expected_keys.append((printer_id, f"{base}.3mf")) expected_archive_id = None for key in expected_keys: expected_archive_id = _expected_prints.pop(key, None) if expected_archive_id: # Clean up other possible keys for this print for other_key in expected_keys: _expected_prints.pop(other_key, None) break if expected_archive_id: # This is a reprint/scheduled print - use existing archive, don't create new one logger.info(f"Using expected archive {expected_archive_id} for print (skipping duplicate)") from datetime import datetime from backend.app.models.archive import PrintArchive result = await db.execute(select(PrintArchive).where(PrintArchive.id == expected_archive_id)) archive = result.scalar_one_or_none() if archive: # Update archive status to printing archive.status = "printing" archive.started_at = datetime.now() await db.commit() # Track as active print _active_prints[(printer_id, archive.filename)] = archive.id if subtask_name: _active_prints[(printer_id, f"{subtask_name}.3mf")] = archive.id # Set up energy tracking try: plug_result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id)) plug = plug_result.scalar_one_or_none() logger.info( f"[ENERGY] Print start - archive {archive.id}, printer {printer_id}, plug found: {plug is not None}" ) if plug: energy = await tasmota_service.get_energy(plug) logger.info(f"[ENERGY] Energy response from plug: {energy}") if energy and energy.get("total") is not None: _print_energy_start[archive.id] = energy["total"] logger.info( f"[ENERGY] Recorded starting energy for archive {archive.id}: {energy['total']} kWh" ) else: logger.warning(f"[ENERGY] No 'total' in energy response for archive {archive.id}") else: logger.info(f"[ENERGY] No smart plug found for printer {printer_id}") except Exception as e: logger.warning(f"Failed to record starting energy: {e}") await ws_manager.send_archive_updated( { "id": archive.id, "status": "printing", } ) # Send notification with archive data (reprint/scheduled) if not notification_sent: archive_data = {"print_time_seconds": archive.print_time_seconds} await _send_print_start_notification(printer_id, data, archive_data, logger) # Extract printable objects from the archived 3MF file _load_objects_from_archive(archive, printer_id, logger) return # Skip creating a new archive # Check if there's already a "printing" archive for this printer/file # This prevents duplicates when backend restarts during an active print from backend.app.models.archive import PrintArchive check_name = subtask_name or filename.split("/")[-1].replace(".gcode", "").replace(".3mf", "") existing = await db.execute( select(PrintArchive) .where(PrintArchive.printer_id == printer_id) .where(PrintArchive.status == "printing") .where(PrintArchive.print_name.ilike(f"%{check_name}%")) .order_by(PrintArchive.created_at.desc()) .limit(1) ) existing_archive = existing.scalar_one_or_none() if existing_archive: logger.info(f"Skipping duplicate - already have printing archive {existing_archive.id} for {check_name}") # Track this as the active print _active_prints[(printer_id, existing_archive.filename)] = existing_archive.id # Also set up energy tracking if not already tracked if existing_archive.id not in _print_energy_start: try: plug_result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id)) plug = plug_result.scalar_one_or_none() if plug: energy = await tasmota_service.get_energy(plug) if energy and energy.get("total") is not None: _print_energy_start[existing_archive.id] = energy["total"] logger.info( f"Recorded starting energy for existing archive {existing_archive.id}: {energy['total']} kWh" ) except Exception as e: logger.warning(f"Failed to record starting energy for existing archive: {e}") # Send notification with archive data (existing archive) if not notification_sent: archive_data = {"print_time_seconds": existing_archive.print_time_seconds} await _send_print_start_notification(printer_id, data, archive_data, logger) # Extract printable objects from the archived 3MF file _load_objects_from_archive(existing_archive, printer_id, logger) return # Build list of possible 3MF filenames to try possible_names = [] # Bambu printers typically store files as "Name.gcode.3mf" # The subtask_name is usually the best source for the filename if subtask_name: # Try common Bambu naming patterns possible_names.append(f"{subtask_name}.gcode.3mf") possible_names.append(f"{subtask_name}.3mf") # Try original filename with .3mf extension if filename: # Extract just the filename part, not the full path fname = filename.split("/")[-1] if "/" in filename else filename if fname.endswith(".3mf"): possible_names.append(fname) elif fname.endswith(".gcode"): base = fname.rsplit(".", 1)[0] possible_names.append(f"{base}.gcode.3mf") possible_names.append(f"{base}.3mf") else: possible_names.append(f"{fname}.gcode.3mf") possible_names.append(f"{fname}.3mf") # Remove duplicates while preserving order seen = set() possible_names = [x for x in possible_names if not (x in seen or seen.add(x))] logger.info(f"Trying filenames: {possible_names}") # Try to find and download the 3MF file temp_path = None downloaded_filename = None for try_filename in possible_names: if not try_filename.endswith(".3mf"): continue remote_paths = [ f"/cache/{try_filename}", f"/model/{try_filename}", f"/{try_filename}", ] temp_path = app_settings.archive_dir / "temp" / try_filename temp_path.parent.mkdir(parents=True, exist_ok=True) for remote_path in remote_paths: logger.debug(f"Trying FTP download: {remote_path}") try: if await download_file_async( printer.ip_address, printer.access_code, remote_path, temp_path, ): downloaded_filename = try_filename logger.info(f"Downloaded: {remote_path}") break except Exception as e: logger.debug(f"FTP download failed for {remote_path}: {e}") if downloaded_filename: break # If still not found, try listing /cache to find matching file if not downloaded_filename and (filename or subtask_name): search_term = (subtask_name or filename).lower().replace(".gcode", "").replace(".3mf", "") logger.info(f"Direct FTP download failed, listing /cache to find '{search_term}'") try: cache_files = await list_files_async(printer.ip_address, printer.access_code, "/cache") threemf_files = [f.get("name") for f in cache_files if f.get("name", "").endswith(".3mf")] logger.info( f"Found {len(threemf_files)} 3MF files in /cache: {threemf_files[:5]}{'...' if len(threemf_files) > 5 else ''}" ) for f in cache_files: if f.get("is_directory"): continue fname = f.get("name", "") if fname.endswith(".3mf") and search_term in fname.lower(): logger.info(f"Found matching file: {fname}") temp_path = app_settings.archive_dir / "temp" / fname temp_path.parent.mkdir(parents=True, exist_ok=True) if await download_file_async( printer.ip_address, printer.access_code, f"/cache/{fname}", temp_path, ): downloaded_filename = fname logger.info(f"Found and downloaded from cache: {fname}") break except Exception as e: logger.warning(f"Failed to list cache: {e}") if not downloaded_filename or not temp_path: logger.warning(f"Could not find 3MF file for print: {filename or subtask_name}") # Create a fallback archive without 3MF data so the print is still tracked # This commonly happens with P1S/A1 printers where FTP has file size limitations try: from backend.app.models.archive import PrintArchive # Derive print name from subtask_name or filename print_name = subtask_name or filename if print_name: # Clean up the name (remove extensions, path parts) print_name = print_name.split("/")[-1] print_name = print_name.replace(".gcode.3mf", "").replace(".gcode", "").replace(".3mf", "") else: print_name = "Unknown Print" # Create minimal archive entry fallback_archive = PrintArchive( printer_id=printer_id, filename=filename or f"{print_name}.3mf", file_path="", # Empty - no 3MF file available file_size=0, print_name=print_name, status="printing", started_at=datetime.now(), extra_data={"no_3mf_available": True, "original_subtask": subtask_name, "_print_data": data}, ) db.add(fallback_archive) await db.commit() await db.refresh(fallback_archive) logger.info(f"Created fallback archive {fallback_archive.id} for {print_name} (no 3MF available)") # Track as active print _active_prints[(printer_id, fallback_archive.filename)] = fallback_archive.id if filename: _active_prints[(printer_id, filename)] = fallback_archive.id if subtask_name: _active_prints[(printer_id, f"{subtask_name}.3mf")] = fallback_archive.id _active_prints[(printer_id, subtask_name)] = fallback_archive.id # Record starting energy if smart plug available try: plug_result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id)) plug = plug_result.scalar_one_or_none() if plug: energy = await tasmota_service.get_energy(plug) if energy and energy.get("total") is not None: _print_energy_start[fallback_archive.id] = energy["total"] logger.info( f"[ENERGY] Recorded starting energy for fallback archive {fallback_archive.id}: {energy['total']} kWh" ) except Exception as e: logger.warning(f"Failed to record starting energy for fallback: {e}") # Send WebSocket notification await ws_manager.send_archive_created( { "id": fallback_archive.id, "printer_id": fallback_archive.printer_id, "filename": fallback_archive.filename, "print_name": fallback_archive.print_name, "status": fallback_archive.status, } ) # Send notification without archive data (file not found) if not notification_sent: await _send_print_start_notification(printer_id, data, logger=logger) return except Exception as e: logger.error(f"Failed to create fallback archive: {e}") # Send notification without archive data (file not found) if not notification_sent: await _send_print_start_notification(printer_id, data, logger=logger) return try: # Archive the file with status "printing" service = ArchiveService(db) archive = await service.archive_print( printer_id=printer_id, source_file=temp_path, print_data={**data, "status": "printing"}, ) if archive: # Track this active print (use both original filename and downloaded filename) _active_prints[(printer_id, downloaded_filename)] = archive.id if filename and filename != downloaded_filename: _active_prints[(printer_id, filename)] = archive.id if subtask_name: _active_prints[(printer_id, f"{subtask_name}.3mf")] = archive.id logger.info(f"Created archive {archive.id} for {downloaded_filename}") # Record starting energy from smart plug if available try: plug_result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id)) plug = plug_result.scalar_one_or_none() logger.info( f"[ENERGY] Auto-archive print start - archive {archive.id}, printer {printer_id}, plug found: {plug is not None}" ) if plug: energy = await tasmota_service.get_energy(plug) logger.info(f"[ENERGY] Auto-archive energy response: {energy}") if energy and energy.get("total") is not None: _print_energy_start[archive.id] = energy["total"] logger.info( f"[ENERGY] Recorded starting energy for archive {archive.id}: {energy['total']} kWh" ) else: logger.warning(f"[ENERGY] No 'total' in energy response for archive {archive.id}") else: logger.info(f"[ENERGY] No smart plug found for printer {printer_id}") except Exception as e: logger.warning(f"Failed to record starting energy: {e}") await ws_manager.send_archive_created( { "id": archive.id, "printer_id": archive.printer_id, "filename": archive.filename, "print_name": archive.print_name, "status": archive.status, } ) # Send notification with archive data (new archive created) if not notification_sent: archive_data = {"print_time_seconds": archive.print_time_seconds} await _send_print_start_notification(printer_id, data, archive_data, logger) notification_sent = True # Extract printable objects for skip object functionality try: from backend.app.services.archive import extract_printable_objects_from_3mf with open(temp_path, "rb") as f: threemf_data = f.read() # Extract with positions for UI overlay printable_objects, bbox_all = extract_printable_objects_from_3mf( threemf_data, include_positions=True ) if printable_objects: # Store objects in printer state client = printer_manager.get_client(printer_id) if client: client.state.printable_objects = printable_objects client.state.printable_objects_bbox_all = bbox_all client.state.skipped_objects = [] # Reset skipped objects for new print logger.info(f"Loaded {len(printable_objects)} printable objects for printer {printer_id}") except Exception as e: logger.debug(f"Failed to extract printable objects: {e}") finally: if temp_path and temp_path.exists(): temp_path.unlink() async def _scan_for_timelapse_with_retries(archive_id: int): """ Scan for timelapse with retries. The printer encodes the timelapse quickly after print completion. We just need a short delay then grab the most recent file. Since we KNOW timelapse was active (from MQTT ipcam data), the most recent file in /timelapse is our target. Retries handle FTP connection issues. """ import logging logger = logging.getLogger(__name__) # Short delays - printer usually finishes encoding within seconds retry_delays = [5, 10, 20] for attempt, delay in enumerate(retry_delays, 1): logger.info( f"[TIMELAPSE] Attempt {attempt}/{len(retry_delays)}: waiting {delay}s before scanning for archive {archive_id}" ) await asyncio.sleep(delay) try: async with async_session() as db: from backend.app.models.printer import Printer from backend.app.services.bambu_ftp import download_file_bytes_async, list_files_async # Get archive (ArchiveService from module-level import) service = ArchiveService(db) archive = await service.get_archive(archive_id) if not archive: logger.warning(f"[TIMELAPSE] Archive {archive_id} not found, stopping retries") return if archive.timelapse_path: logger.info(f"[TIMELAPSE] Archive {archive_id} already has timelapse attached, stopping retries") return if not archive.printer_id: logger.warning(f"[TIMELAPSE] Archive {archive_id} has no printer, stopping retries") return # Get printer result = await db.execute(select(Printer).where(Printer.id == archive.printer_id)) printer = result.scalar_one_or_none() if not printer: logger.warning(f"[TIMELAPSE] Printer not found for archive {archive_id}, stopping retries") return # Scan timelapse directory on printer # H2D may store in different locations than X1C files = [] found_path = None for timelapse_path in ["/timelapse", "/timelapse/video", "/record", "/recording"]: try: found_files = await list_files_async(printer.ip_address, printer.access_code, timelapse_path) if found_files: files = found_files found_path = timelapse_path logger.info(f"[TIMELAPSE] Attempt {attempt}: Found {len(files)} files in {timelapse_path}") break except Exception as e: logger.debug(f"[TIMELAPSE] Path {timelapse_path} failed: {e}") continue if not files: logger.info(f"[TIMELAPSE] Attempt {attempt}: No timelapse files found on printer, will retry") continue mp4_files = [f for f in files if not f.get("is_directory") and f.get("name", "").endswith(".mp4")] # Log ALL mp4 files found for debugging logger.info(f"[TIMELAPSE] Attempt {attempt}: Found {len(mp4_files)} MP4 files in {found_path}") for f in mp4_files[:5]: # Log first 5 logger.info(f"[TIMELAPSE] - {f.get('name')}, mtime={f.get('mtime')}") if not mp4_files: logger.info(f"[TIMELAPSE] Attempt {attempt}: No MP4 files found, will retry") continue # Sort by mtime descending to get most recent file mp4_files_with_mtime = [f for f in mp4_files if f.get("mtime")] if not mp4_files_with_mtime: logger.info(f"[TIMELAPSE] Attempt {attempt}: No MP4 files with mtime found, will retry") continue mp4_files_with_mtime.sort(key=lambda x: x.get("mtime"), reverse=True) most_recent = mp4_files_with_mtime[0] file_name = most_recent.get("name") logger.info(f"[TIMELAPSE] Attempt {attempt}: Most recent file: {file_name}") # Since we KNOW timelapse was active (from MQTT), just grab the most recent file remote_path = most_recent.get("path") or f"/timelapse/{file_name}" logger.info(f"[TIMELAPSE] Downloading {file_name} for archive {archive_id}") timelapse_data = await download_file_bytes_async(printer.ip_address, printer.access_code, remote_path) if timelapse_data: success = await service.attach_timelapse(archive_id, timelapse_data, file_name) if success: logger.info(f"[TIMELAPSE] Successfully attached timelapse to archive {archive_id}") await ws_manager.send_archive_updated({"id": archive_id, "timelapse_attached": True}) return # Success! else: logger.warning(f"[TIMELAPSE] Failed to attach timelapse to archive {archive_id}") else: logger.warning(f"[TIMELAPSE] Attempt {attempt}: Failed to download, will retry") except Exception as e: logger.warning(f"[TIMELAPSE] Attempt {attempt} failed with error: {e}") logger.warning(f"[TIMELAPSE] All {len(retry_delays)} attempts exhausted for archive {archive_id}, giving up") async def on_print_complete(printer_id: int, data: dict): """Handle print completion - update the archive status.""" import logging import time logger = logging.getLogger(__name__) start_time = time.time() def log_timing(section: str): elapsed = time.time() - start_time logger.info(f"[TIMING] {section}: {elapsed:.3f}s elapsed") logger.info(f"[CALLBACK] on_print_complete started for printer {printer_id}") try: ws_data = { "status": data.get("status"), "filename": data.get("filename"), "subtask_name": data.get("subtask_name"), "timelapse_was_active": data.get("timelapse_was_active"), } await ws_manager.send_print_complete(printer_id, ws_data) log_timing("WebSocket send_print_complete") except Exception as e: logger.warning(f"[CALLBACK] WebSocket send_print_complete failed: {e}") filename = data.get("filename", "") subtask_name = data.get("subtask_name", "") if not filename and not subtask_name: logger.warning("Print complete without filename or subtask_name") return logger.info(f"Print complete - filename: {filename}, subtask: {subtask_name}, status: {data.get('status')}") # Build list of possible keys to try (matching how they were registered in on_print_start) possible_keys = [] # Try subtask_name variations first (most reliable for matching) if subtask_name: possible_keys.append((printer_id, f"{subtask_name}.3mf")) possible_keys.append((printer_id, f"{subtask_name}.gcode.3mf")) possible_keys.append((printer_id, subtask_name)) # Try filename variations if filename: # Extract just the filename if it's a path fname = filename.split("/")[-1] if "/" in filename else filename if fname.endswith(".3mf"): possible_keys.append((printer_id, fname)) elif fname.endswith(".gcode"): base_name = fname.rsplit(".", 1)[0] possible_keys.append((printer_id, f"{base_name}.gcode.3mf")) possible_keys.append((printer_id, f"{base_name}.3mf")) possible_keys.append((printer_id, fname)) else: possible_keys.append((printer_id, f"{fname}.gcode.3mf")) possible_keys.append((printer_id, f"{fname}.3mf")) possible_keys.append((printer_id, fname)) # Also try full path versions if filename.endswith(".3mf"): possible_keys.append((printer_id, filename)) elif filename.endswith(".gcode"): base_name = filename.rsplit(".", 1)[0] possible_keys.append((printer_id, f"{base_name}.3mf")) possible_keys.append((printer_id, filename)) else: possible_keys.append((printer_id, f"{filename}.3mf")) possible_keys.append((printer_id, filename)) # Find the archive for this print logger.info(f"Looking for archive in _active_prints, keys to try: {possible_keys[:5]}...") logger.info(f"Current _active_prints: {list(_active_prints.keys())}") archive_id = None for key in possible_keys: archive_id = _active_prints.pop(key, None) if archive_id: logger.info(f"Found archive {archive_id} with key {key}") # Also clean up any other keys pointing to this archive keys_to_remove = [k for k, v in _active_prints.items() if v == archive_id] for k in keys_to_remove: _active_prints.pop(k, None) break if not archive_id: # Try to find by filename or subtask_name if not tracked (for prints started before app) async with async_session() as db: from backend.app.models.archive import PrintArchive # Try matching by subtask_name (stored as print_name) first if subtask_name: result = await db.execute( select(PrintArchive) .where(PrintArchive.printer_id == printer_id) .where(PrintArchive.status == "printing") .where( or_( PrintArchive.print_name.ilike(f"%{subtask_name}%"), PrintArchive.filename.ilike(f"%{subtask_name}%"), ) ) .order_by(PrintArchive.created_at.desc()) .limit(1) ) archive = result.scalar_one_or_none() if archive: archive_id = archive.id logger.info(f"Found archive {archive_id} by subtask_name match: {subtask_name}") # Also try by filename if not archive_id and filename: result = await db.execute( select(PrintArchive) .where(PrintArchive.printer_id == printer_id) .where(PrintArchive.filename == filename) .where(PrintArchive.status == "printing") .order_by(PrintArchive.created_at.desc()) .limit(1) ) archive = result.scalar_one_or_none() if archive: archive_id = archive.id if not archive_id: logger.warning(f"Could not find archive for print complete: filename={filename}, subtask={subtask_name}") return log_timing("Archive lookup") # Update archive status logger.info(f"[ARCHIVE] Updating archive {archive_id} status...") try: async with async_session() as db: service = ArchiveService(db) status = data.get("status", "completed") # Auto-detect failure reason failure_reason = None if status == "aborted": failure_reason = "User cancelled" logger.info("[ARCHIVE] Print was aborted by user, setting failure_reason='User cancelled'") elif status == "failed": # Try to determine failure reason from HMS errors hms_errors = data.get("hms_errors", []) if hms_errors: logger.info(f"[ARCHIVE] HMS errors at failure: {hms_errors}") # Map known HMS error modules to failure reasons # Module 0x07 = Filament, 0x0C = MC (Motion Controller), etc. for err in hms_errors: module = err.get("module", 0) if module == 0x07: # Filament module failure_reason = "Filament runout" break elif module == 0x0C: # Motion controller failure_reason = "Layer shift" break elif module == 0x05: # Nozzle/extruder failure_reason = "Clogged nozzle" break if failure_reason: logger.info(f"[ARCHIVE] Detected failure_reason from HMS: {failure_reason}") else: logger.info("[ARCHIVE] No HMS errors available to determine failure reason") await service.update_archive_status( archive_id, status=status, completed_at=datetime.now() if status in ("completed", "failed", "aborted") else None, failure_reason=failure_reason, ) logger.info(f"[ARCHIVE] Archive {archive_id} status updated to {status}, failure_reason={failure_reason}") await ws_manager.send_archive_updated( { "id": archive_id, "status": status, } ) logger.info(f"[ARCHIVE] WebSocket notification sent for archive {archive_id}") except Exception as e: logger.error(f"[ARCHIVE] Failed to update archive {archive_id} status: {e}", exc_info=True) # Continue with other operations even if archive update fails log_timing("Archive status update") # Report filament usage to Spoolman if print completed successfully if data.get("status") == "completed": try: await _report_spoolman_usage(printer_id, archive_id, logger) log_timing("Spoolman usage report") except Exception as e: logger.warning(f"Spoolman usage reporting failed: {e}") # Run slow operations as background tasks to avoid blocking the event loop # These operations can take 5-10+ seconds and would freeze the UI if awaited starting_kwh = _print_energy_start.pop(archive_id, None) async def _background_energy_calculation(): """Calculate and save energy usage in background.""" try: logger.info(f"[ENERGY-BG] Starting energy calculation for archive {archive_id}") async with async_session() as db: plug_result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id)) plug = plug_result.scalar_one_or_none() if plug: energy = await tasmota_service.get_energy(plug) logger.info(f"[ENERGY-BG] Energy response: {energy}") energy_used = None if starting_kwh is not None and energy and energy.get("total") is not None: ending_kwh = energy["total"] energy_used = round(ending_kwh - starting_kwh, 4) logger.info(f"[ENERGY-BG] Per-print energy: {energy_used} kWh") if energy_used is not None and energy_used >= 0: from backend.app.api.routes.settings import get_setting energy_cost_per_kwh = await get_setting(db, "energy_cost_per_kwh") cost_per_kwh = float(energy_cost_per_kwh) if energy_cost_per_kwh else 0.15 energy_cost = round(energy_used * cost_per_kwh, 2) from backend.app.models.archive import PrintArchive result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id)) archive = result.scalar_one_or_none() if archive: archive.energy_kwh = energy_used archive.energy_cost = energy_cost await db.commit() logger.info(f"[ENERGY-BG] Saved: {energy_used} kWh, cost={energy_cost}") else: logger.info(f"[ENERGY-BG] No smart plug for printer {printer_id}") except Exception as e: logger.warning(f"[ENERGY-BG] Failed: {e}") async def _background_finish_photo(): """Capture finish photo in background.""" try: logger.info(f"[PHOTO-BG] Starting finish photo capture for archive {archive_id}") from backend.app.api.routes.camera import _active_streams, get_buffered_frame async with async_session() as db: from backend.app.api.routes.settings import get_setting capture_enabled = await get_setting(db, "capture_finish_photo") if capture_enabled is None or capture_enabled.lower() == "true": from backend.app.models.printer import Printer result = await db.execute(select(Printer).where(Printer.id == printer_id)) printer = result.scalar_one_or_none() if printer and archive_id: from backend.app.models.archive import PrintArchive result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id)) archive = result.scalar_one_or_none() if archive: import uuid from datetime import datetime from pathlib import Path archive_dir = app_settings.base_dir / Path(archive.file_path).parent photo_filename = None # Check if camera stream is active - use buffered frame to avoid freeze active_for_printer = [k for k in _active_streams if k.startswith(f"{printer_id}-")] buffered_frame = get_buffered_frame(printer_id) if active_for_printer and buffered_frame: # Use frame from active stream logger.info("[PHOTO-BG] Using buffered frame from active stream") photos_dir = archive_dir / "photos" photos_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") photo_filename = f"finish_{timestamp}_{uuid.uuid4().hex[:8]}.jpg" photo_path = photos_dir / photo_filename await asyncio.to_thread(photo_path.write_bytes, buffered_frame) logger.info(f"[PHOTO-BG] Saved buffered frame: {photo_filename}") else: # No active stream - capture new frame from backend.app.services.camera import capture_finish_photo photo_filename = await capture_finish_photo( printer_id=printer_id, ip_address=printer.ip_address, access_code=printer.access_code, model=printer.model, archive_dir=archive_dir, ) if photo_filename: photos = archive.photos or [] photos.append(photo_filename) archive.photos = photos await db.commit() logger.info(f"[PHOTO-BG] Saved: {photo_filename}") except Exception as e: logger.warning(f"[PHOTO-BG] Failed: {e}") asyncio.create_task(_background_energy_calculation()) asyncio.create_task(_background_finish_photo()) # Skips if camera stream active log_timing("Background tasks scheduled (energy, photo)") # Also run smart plug, notifications, and maintenance as background tasks print_status = data.get("status", "completed") async def _background_smart_plug(): """Handle smart plug automation in background.""" try: logger.info(f"[AUTO-OFF-BG] Starting smart plug automation for printer {printer_id}") async with async_session() as db: await smart_plug_manager.on_print_complete(printer_id, print_status, db) logger.info("[AUTO-OFF-BG] Completed") except Exception as e: logger.warning(f"[AUTO-OFF-BG] Failed: {e}") async def _background_notifications(): """Send print complete notifications in background.""" try: logger.info(f"[NOTIFY-BG] Starting notifications for printer {printer_id}") async with async_session() as db: from backend.app.models.archive import PrintArchive from backend.app.models.printer import Printer result = await db.execute(select(Printer).where(Printer.id == printer_id)) printer = result.scalar_one_or_none() printer_name = printer.name if printer else f"Printer {printer_id}" archive_data = None if archive_id: archive_result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id)) archive = archive_result.scalar_one_or_none() if archive: archive_data = { "print_time_seconds": archive.print_time_seconds, "actual_filament_grams": archive.filament_used_grams, "failure_reason": archive.failure_reason, } await notification_service.on_print_complete( printer_id, printer_name, print_status, data, db, archive_data=archive_data ) logger.info("[NOTIFY-BG] Completed") except Exception as e: logger.warning(f"[NOTIFY-BG] Failed: {e}") async def _background_maintenance_check(): """Check for maintenance due in background.""" if print_status != "completed": return try: logger.info(f"[MAINT-BG] Starting maintenance check for printer {printer_id}") async with async_session() as db: from backend.app.models.printer import Printer result = await db.execute(select(Printer).where(Printer.id == printer_id)) printer = result.scalar_one_or_none() printer_name = printer.name if printer else f"Printer {printer_id}" await ensure_default_types(db) overview = await _get_printer_maintenance_internal(printer_id, db, commit=True) items_needing_attention = [ {"name": item.maintenance_type_name, "is_due": item.is_due, "is_warning": item.is_warning} for item in overview.maintenance_items if item.enabled and (item.is_due or item.is_warning) ] if items_needing_attention: await notification_service.on_maintenance_due(printer_id, printer_name, items_needing_attention, db) logger.info(f"[MAINT-BG] Sent notification: {len(items_needing_attention)} items need attention") else: logger.info("[MAINT-BG] Completed (no items need attention)") except Exception as e: logger.warning(f"[MAINT-BG] Failed: {e}") asyncio.create_task(_background_smart_plug()) asyncio.create_task(_background_notifications()) asyncio.create_task(_background_maintenance_check()) log_timing("All background tasks scheduled") # Auto-scan for timelapse if recording was active during the print if archive_id and data.get("timelapse_was_active") and data.get("status") == "completed": logger.info(f"[TIMELAPSE] Timelapse was active during print, scheduling auto-scan for archive {archive_id}") # Schedule timelapse scan as background task with retries # The printer needs time to encode the video after print completion asyncio.create_task(_scan_for_timelapse_with_retries(archive_id)) log_timing("Timelapse scan scheduled") # Update queue item if this was a scheduled print try: async with async_session() as db: from backend.app.models.print_queue import PrintQueueItem # Note: SmartPlug is already imported at module level (line 56) # Do NOT import it here as it would shadow the module-level import # and cause "cannot access local variable" errors earlier in this function result = await db.execute( select(PrintQueueItem) .where(PrintQueueItem.printer_id == printer_id) .where(PrintQueueItem.status == "printing") ) queue_item = result.scalar_one_or_none() if queue_item: status = data.get("status", "completed") queue_item.status = status queue_item.completed_at = datetime.now() await db.commit() logger.info(f"Updated queue item {queue_item.id} status to {status}") # Handle auto_off_after - power off printer if requested (after cooldown) if queue_item.auto_off_after: result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id)) plug = result.scalar_one_or_none() if plug and plug.enabled: logger.info(f"Auto-off requested for printer {printer_id}, waiting for cooldown...") async def cooldown_and_poweroff(pid: int, plug_id: int): # Wait for nozzle to cool down await printer_manager.wait_for_cooldown(pid, target_temp=50.0, timeout=600) # Re-fetch plug in new session async with async_session() as new_db: result = await new_db.execute(select(SmartPlug).where(SmartPlug.id == plug_id)) p = result.scalar_one_or_none() if p and p.enabled: success = await tasmota_service.turn_off(p) if success: logger.info(f"Powered off printer {pid} via smart plug '{p.name}'") else: logger.warning(f"Failed to power off printer {pid} via smart plug") asyncio.create_task(cooldown_and_poweroff(printer_id, plug.id)) except Exception as e: import logging logging.getLogger(__name__).warning(f"Queue item update failed: {e}") log_timing("Queue item update") logger.info(f"[CALLBACK] on_print_complete finished for printer {printer_id}, archive {archive_id}") # AMS sensor history recording _ams_history_task: asyncio.Task | None = None AMS_HISTORY_INTERVAL = 300 # Record every 5 minutes AMS_HISTORY_RETENTION_DAYS = 30 # Keep data for 30 days _ams_cleanup_counter = 0 # Track recordings to trigger periodic cleanup _ams_alarm_cooldown: dict[str, datetime] = {} # Track alarm cooldowns (printer_id:ams_id:type -> last_alarm_time) AMS_ALARM_COOLDOWN_MINUTES = 60 # Don't send same alarm more than once per hour async def record_ams_history(): """Background task to record AMS humidity and temperature data.""" import logging logger = logging.getLogger(__name__) # Wait a short time for MQTT connections to establish on startup await asyncio.sleep(10) while True: try: from backend.app.models.ams_history import AMSSensorHistory from backend.app.models.printer import Printer from backend.app.models.settings import Settings async with async_session() as db: # Get all active printers result = await db.execute(select(Printer).where(Printer.is_active.is_(True))) printers = result.scalars().all() # Get alarm thresholds from settings humidity_threshold = 60.0 # Default: fair threshold temp_threshold = 35.0 # Default: fair threshold result = await db.execute(select(Settings).where(Settings.key == "ams_humidity_fair")) setting = result.scalar_one_or_none() if setting: try: humidity_threshold = float(setting.value) except (ValueError, TypeError): pass result = await db.execute(select(Settings).where(Settings.key == "ams_temp_fair")) setting = result.scalar_one_or_none() if setting: try: temp_threshold = float(setting.value) except (ValueError, TypeError): pass recorded_count = 0 for printer in printers: # Get current state from printer manager state = printer_manager.get_status(printer.id) if not state or not state.connected or not state.raw_data: continue # Skip disconnected printers - don't use stale data raw_data = state.raw_data if "ams" not in raw_data or not isinstance(raw_data["ams"], list): continue # Record data for each AMS unit for ams_data in raw_data["ams"]: ams_id = int(ams_data.get("id", 0)) # Get humidity (prefer humidity_raw) humidity_raw = ams_data.get("humidity_raw") humidity_idx = ams_data.get("humidity") humidity = None if humidity_raw is not None: try: humidity = float(humidity_raw) except (ValueError, TypeError): pass if humidity is None and humidity_idx is not None: try: humidity = float(humidity_idx) except (ValueError, TypeError): pass # Get temperature temperature = None temp_str = ams_data.get("temp") if temp_str is not None: try: temperature = float(temp_str) except (ValueError, TypeError): pass # Skip if no data if humidity is None and temperature is None: continue # Record the data point history = AMSSensorHistory( printer_id=printer.id, ams_id=ams_id, humidity=humidity, humidity_raw=float(humidity_raw) if humidity_raw else None, temperature=temperature, ) db.add(history) recorded_count += 1 # Generate AMS label and determine if it's AMS-HT (A, B, C, D or HT-A for AMS-Lite/Hub) is_ams_ht = ams_id >= 128 if is_ams_ht: ams_label = f"HT-{chr(65 + (ams_id - 128))}" else: ams_label = f"AMS-{chr(65 + ams_id)}" # Check humidity alarm (only if above threshold) if humidity is not None and humidity > humidity_threshold: cooldown_key = f"{printer.id}:{ams_id}:humidity" last_alarm = _ams_alarm_cooldown.get(cooldown_key) now = datetime.now() if ( last_alarm is None or (now - last_alarm).total_seconds() >= AMS_ALARM_COOLDOWN_MINUTES * 60 ): _ams_alarm_cooldown[cooldown_key] = now logger.info( f"Sending humidity alarm for {printer.name} {ams_label}: {humidity}% > {humidity_threshold}%" ) try: # Call different notification method based on AMS type if is_ams_ht: await notification_service.on_ams_ht_humidity_high( printer.id, printer.name, ams_label, humidity, humidity_threshold, db ) else: await notification_service.on_ams_humidity_high( printer.id, printer.name, ams_label, humidity, humidity_threshold, db ) except Exception as e: logger.warning(f"Failed to send humidity alarm: {e}") # Check temperature alarm (only if above threshold) if temperature is not None and temperature > temp_threshold: cooldown_key = f"{printer.id}:{ams_id}:temperature" last_alarm = _ams_alarm_cooldown.get(cooldown_key) now = datetime.now() if ( last_alarm is None or (now - last_alarm).total_seconds() >= AMS_ALARM_COOLDOWN_MINUTES * 60 ): _ams_alarm_cooldown[cooldown_key] = now logger.info( f"Sending temperature alarm for {printer.name} {ams_label}: {temperature}°C > {temp_threshold}°C" ) try: # Call different notification method based on AMS type if is_ams_ht: await notification_service.on_ams_ht_temperature_high( printer.id, printer.name, ams_label, temperature, temp_threshold, db ) else: await notification_service.on_ams_temperature_high( printer.id, printer.name, ams_label, temperature, temp_threshold, db ) except Exception as e: logger.warning(f"Failed to send temperature alarm: {e}") await db.commit() if recorded_count > 0: logger.info(f"Recorded {recorded_count} AMS sensor history entries") # Periodic cleanup of old data (every ~288 recordings = ~24 hours at 5min interval) global _ams_cleanup_counter _ams_cleanup_counter += 1 if _ams_cleanup_counter >= 288: _ams_cleanup_counter = 0 # Get retention days from settings from backend.app.models.settings import Settings result = await db.execute(select(Settings).where(Settings.key == "ams_history_retention_days")) setting = result.scalar_one_or_none() retention_days = int(setting.value) if setting else AMS_HISTORY_RETENTION_DAYS cutoff = datetime.now() - timedelta(days=retention_days) result = await db.execute(delete(AMSSensorHistory).where(AMSSensorHistory.recorded_at < cutoff)) await db.commit() if result.rowcount > 0: logger.info( f"Cleaned up {result.rowcount} old AMS sensor history entries (older than {retention_days} days)" ) # Wait until next recording interval await asyncio.sleep(AMS_HISTORY_INTERVAL) except asyncio.CancelledError: break except Exception as e: logger.warning(f"AMS history recording failed: {e}") await asyncio.sleep(60) # Wait a bit before retrying def start_ams_history_recording(): """Start the AMS history recording background task.""" global _ams_history_task if _ams_history_task is None: _ams_history_task = asyncio.create_task(record_ams_history()) logging.getLogger(__name__).info("AMS history recording started") def stop_ams_history_recording(): """Stop the AMS history recording background task.""" global _ams_history_task if _ams_history_task: _ams_history_task.cancel() _ams_history_task = None logging.getLogger(__name__).info("AMS history recording stopped") # Printer runtime tracking _runtime_tracking_task: asyncio.Task | None = None RUNTIME_TRACKING_INTERVAL = 30 # Update every 30 seconds async def track_printer_runtime(): """Background task to track printer active runtime (RUNNING/PAUSE states).""" import logging logger = logging.getLogger(__name__) # Wait for MQTT connections to establish on startup await asyncio.sleep(15) while True: try: from backend.app.models.printer import Printer async with async_session() as db: # Get all active printers result = await db.execute(select(Printer).where(Printer.is_active.is_(True))) printers = result.scalars().all() now = datetime.now() updated_count = 0 needs_commit = False for printer in printers: # Get current state from printer manager state = printer_manager.get_status(printer.id) if not state: logger.debug(f"[{printer.name}] Runtime tracking: no state available") continue if not state.connected: logger.debug(f"[{printer.name}] Runtime tracking: not connected") continue # Check if printer is in an active state (RUNNING or PAUSE) if state.state in ("RUNNING", "PAUSE"): # Calculate time since last update if printer.last_runtime_update: elapsed = (now - printer.last_runtime_update).total_seconds() # Sanity check: don't add more than 2x the interval (handles server restarts) if elapsed > 0 and elapsed < RUNTIME_TRACKING_INTERVAL * 2: printer.runtime_seconds += int(elapsed) updated_count += 1 needs_commit = True logger.debug( f"[{printer.name}] Runtime tracking: added {int(elapsed)}s, " f"total={printer.runtime_seconds}s ({printer.runtime_seconds / 3600:.2f}h)" ) else: logger.warning( f"[{printer.name}] Runtime tracking: skipped elapsed={elapsed:.1f}s " f"(outside valid range 0-{RUNTIME_TRACKING_INTERVAL * 2}s)" ) else: # First time seeing printer active - need to commit to save timestamp needs_commit = True logger.debug(f"[{printer.name}] Runtime tracking: first active detection") printer.last_runtime_update = now else: # Printer is idle/offline - clear last_runtime_update if printer.last_runtime_update is not None: logger.debug( f"[{printer.name}] Runtime tracking: state={state.state}, clearing last_runtime_update" ) printer.last_runtime_update = None needs_commit = True if needs_commit: await db.commit() if updated_count > 0: logger.debug(f"Updated runtime for {updated_count} printer(s)") except asyncio.CancelledError: logger.info("Runtime tracking cancelled") break except Exception as e: logger.warning(f"Runtime tracking failed: {e}") await asyncio.sleep(RUNTIME_TRACKING_INTERVAL) def start_runtime_tracking(): """Start the printer runtime tracking background task.""" global _runtime_tracking_task if _runtime_tracking_task is None: _runtime_tracking_task = asyncio.create_task(track_printer_runtime()) logging.getLogger(__name__).info("Printer runtime tracking started") def stop_runtime_tracking(): """Stop the printer runtime tracking background task.""" global _runtime_tracking_task if _runtime_tracking_task: _runtime_tracking_task.cancel() _runtime_tracking_task = None logging.getLogger(__name__).info("Printer runtime tracking stopped") @asynccontextmanager async def lifespan(app: FastAPI): # Startup await init_db() # Set up printer manager callbacks loop = asyncio.get_event_loop() printer_manager.set_event_loop(loop) printer_manager.set_status_change_callback(on_printer_status_change) printer_manager.set_print_start_callback(on_print_start) printer_manager.set_print_complete_callback(on_print_complete) printer_manager.set_ams_change_callback(on_ams_change) # Connect to all active printers async with async_session() as db: await init_printer_connections(db) # Auto-connect to Spoolman if enabled async with async_session() as db: from backend.app.api.routes.settings import get_setting spoolman_enabled = await get_setting(db, "spoolman_enabled") spoolman_url = await get_setting(db, "spoolman_url") if spoolman_enabled and spoolman_enabled.lower() == "true" and spoolman_url: try: client = await init_spoolman_client(spoolman_url) if await client.health_check(): logging.info(f"Auto-connected to Spoolman at {spoolman_url}") else: logging.warning(f"Spoolman at {spoolman_url} is not reachable") except Exception as e: logging.warning(f"Failed to auto-connect to Spoolman: {e}") # Start the print scheduler asyncio.create_task(print_scheduler.run()) # Start the smart plug scheduler for time-based on/off smart_plug_manager.start_scheduler() # Resume any pending auto-offs that were interrupted by restart await smart_plug_manager.resume_pending_auto_offs() # Start the notification digest scheduler notification_service.start_digest_scheduler() # Start AMS history recording start_ams_history_recording() # Start printer runtime tracking start_runtime_tracking() # Start anonymous telemetry (opt-out via settings) asyncio.create_task(start_telemetry_loop(async_session)) # Initialize virtual printer manager from backend.app.services.virtual_printer import virtual_printer_manager virtual_printer_manager.set_session_factory(async_session) # Auto-start virtual printer if enabled async with async_session() as db: from backend.app.api.routes.settings import get_setting vp_enabled = await get_setting(db, "virtual_printer_enabled") if vp_enabled and vp_enabled.lower() == "true": vp_access_code = await get_setting(db, "virtual_printer_access_code") or "" vp_mode = await get_setting(db, "virtual_printer_mode") or "immediate" if vp_access_code: try: await virtual_printer_manager.configure( enabled=True, access_code=vp_access_code, mode=vp_mode, ) logging.info("Virtual printer started") except Exception as e: logging.warning(f"Failed to start virtual printer: {e}") yield # Shutdown print_scheduler.stop() smart_plug_manager.stop_scheduler() notification_service.stop_digest_scheduler() stop_ams_history_recording() stop_runtime_tracking() printer_manager.disconnect_all() await close_spoolman_client() # Stop virtual printer if running if virtual_printer_manager.is_enabled: await virtual_printer_manager.configure(enabled=False) app = FastAPI( title=app_settings.app_name, description="Archive and manage Bambu Lab 3MF files", version=APP_VERSION, lifespan=lifespan, ) # API routes app.include_router(printers.router, prefix=app_settings.api_prefix) app.include_router(archives.router, prefix=app_settings.api_prefix) app.include_router(filaments.router, prefix=app_settings.api_prefix) app.include_router(settings_routes.router, prefix=app_settings.api_prefix) app.include_router(cloud.router, prefix=app_settings.api_prefix) app.include_router(smart_plugs.router, prefix=app_settings.api_prefix) app.include_router(print_queue.router, prefix=app_settings.api_prefix) app.include_router(kprofiles.router, prefix=app_settings.api_prefix) app.include_router(notifications.router, prefix=app_settings.api_prefix) app.include_router(notification_templates.router, prefix=app_settings.api_prefix) app.include_router(spoolman.router, prefix=app_settings.api_prefix) app.include_router(updates.router, prefix=app_settings.api_prefix) app.include_router(maintenance.router, prefix=app_settings.api_prefix) app.include_router(camera.router, prefix=app_settings.api_prefix) app.include_router(external_links.router, prefix=app_settings.api_prefix) app.include_router(projects.router, prefix=app_settings.api_prefix) app.include_router(api_keys.router, prefix=app_settings.api_prefix) app.include_router(webhook.router, prefix=app_settings.api_prefix) app.include_router(ams_history.router, prefix=app_settings.api_prefix) 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) # Serve static files (React build) if app_settings.static_dir.exists() and any(app_settings.static_dir.iterdir()): app.mount( "/assets", StaticFiles(directory=app_settings.static_dir / "assets"), name="assets", ) if (app_settings.static_dir / "img").exists(): app.mount( "/img", StaticFiles(directory=app_settings.static_dir / "img"), name="img", ) if (app_settings.static_dir / "icons").exists(): app.mount( "/icons", StaticFiles(directory=app_settings.static_dir / "icons"), name="icons", ) @app.get("/") async def serve_frontend(): """Serve the React frontend.""" index_file = app_settings.static_dir / "index.html" if index_file.exists(): return FileResponse(index_file) return { "message": "Bambuddy API", "docs": "/docs", "frontend": "Build and place React app in /static directory", } @app.get("/health") async def health_check(): """Health check endpoint.""" return {"status": "healthy"} @app.get("/manifest.json") async def serve_manifest(): """Serve PWA manifest.""" manifest_file = app_settings.static_dir / "manifest.json" if manifest_file.exists(): return FileResponse(manifest_file, media_type="application/manifest+json") return {"error": "Manifest not found"} @app.get("/sw.js") async def serve_service_worker(): """Serve service worker.""" sw_file = app_settings.static_dir / "sw.js" if sw_file.exists(): return FileResponse(sw_file, media_type="application/javascript") return {"error": "Service worker not found"} # Catch-all route for React Router (must be last) @app.get("/{full_path:path}") async def serve_spa(full_path: str): """Serve React app for client-side routing.""" # Don't intercept API routes if full_path.startswith("api/"): return {"error": "Not found"} index_file = app_settings.static_dir / "index.html" if index_file.exists(): return FileResponse(index_file) return {"error": "Frontend not built"}