diff --git a/backend/app/api/routes/archives.py b/backend/app/api/routes/archives.py index 4cf82d67d..5ed61ed85 100644 --- a/backend/app/api/routes/archives.py +++ b/backend/app/api/routes/archives.py @@ -2389,7 +2389,8 @@ async def get_archive_plates( for gf in gcode_files: # "Metadata/plate_5.gcode" -> 5 try: - plate_str = gf[15:-6] # Remove "Metadata/plate_" and ".gcode" + # Remove "Metadata/plate_" and ".gcode" + plate_str = gf[15:-6] plate_indices.append(int(plate_str)) except ValueError: pass # Skip gcode file with non-numeric plate index @@ -2792,14 +2793,9 @@ async def reprint_archive( ) ), ): - """Send an archived 3MF file to a printer and start printing.""" - from backend.app.main import register_expected_print + """Dispatch an archived 3MF file for send/start on a printer.""" from backend.app.models.printer import Printer - from backend.app.services.bambu_ftp import ( - get_ftp_retry_settings, - upload_file_async, - with_ftp_retry, - ) + from backend.app.services.background_dispatch import DispatchEnqueueRejected, background_dispatch from backend.app.services.printer_manager import printer_manager user, can_modify_all = auth_result @@ -2829,139 +2825,54 @@ async def reprint_archive( if not printer_manager.is_connected(printer_id): raise HTTPException(400, "Printer is not connected") - # Get the sliced 3MF file path if not archive.file_path: raise HTTPException( 404, "No 3MF file available for this archive. " "The file could not be downloaded from the printer when the print was recorded.", ) + + # Validate archive file exists file_path = settings.base_dir / archive.file_path if not file_path.is_file(): raise HTTPException(404, "Archive file not found") - # Upload file to printer via FTP - from backend.app.services.bambu_ftp import delete_file_async + plate_name = body.plate_name + if not plate_name and body.plate_id is not None: + plate_name = f"Plate {body.plate_id}" - # Use a clean filename to avoid issues with double extensions like .gcode.3mf - # The printer might reject filenames with unusual extensions - base_name = archive.filename - if base_name.endswith(".gcode.3mf"): - base_name = base_name[:-10] # Remove .gcode.3mf - elif base_name.endswith(".3mf"): - base_name = base_name[:-4] # Remove .3mf - remote_filename = f"{base_name}.3mf" - remote_path = f"/{remote_filename}" + dispatch_source_name = archive.filename + if plate_name: + dispatch_source_name = f"{archive.filename} • {plate_name}" - # Get FTP retry settings - ftp_retry_enabled, ftp_retry_count, ftp_retry_delay, ftp_timeout = await get_ftp_retry_settings() + try: + dispatch_result = await background_dispatch.dispatch_reprint_archive( + archive_id=archive_id, + archive_name=dispatch_source_name, + printer_id=printer_id, + printer_name=printer.name, + options=body.model_dump(exclude_none=True), + requested_by_user_id=user.id if user else None, + requested_by_username=user.username if user else None, + ) + except DispatchEnqueueRejected as e: + raise HTTPException(status_code=409, detail=str(e)) from e logger.info( - f"Reprint FTP upload starting: printer={printer.name} ({printer.model}), " - f"ip={printer.ip_address}, file={remote_filename}, local_path={file_path}, " - f"retry_enabled={ftp_retry_enabled}, retry_count={ftp_retry_count}, timeout={ftp_timeout}" - ) - - # Delete existing file if present (avoids 553 error) - logger.debug("Deleting existing file %s if present...", remote_path) - delete_result = await delete_file_async( - printer.ip_address, - printer.access_code, - remote_path, - socket_timeout=ftp_timeout, - printer_model=printer.model, - ) - logger.debug("Delete result: %s", delete_result) - - if ftp_retry_enabled: - uploaded = await with_ftp_retry( - upload_file_async, - printer.ip_address, - printer.access_code, - file_path, - remote_path, - socket_timeout=ftp_timeout, - printer_model=printer.model, - max_retries=ftp_retry_count, - retry_delay=ftp_retry_delay, - operation_name=f"Upload for reprint to {printer.name}", - ) - else: - uploaded = await upload_file_async( - printer.ip_address, - printer.access_code, - file_path, - remote_path, - socket_timeout=ftp_timeout, - printer_model=printer.model, - ) - - if not uploaded: - logger.error( - f"FTP upload failed for reprint: printer={printer.name}, model={printer.model}, " - f"ip={printer.ip_address}, file={remote_filename}. " - "Check logs above for storage diagnostics and specific error codes." - ) - raise HTTPException( - 500, - "Failed to upload file to printer. Check if SD card is inserted and properly formatted (FAT32/exFAT). " - "See server logs for detailed diagnostics.", - ) - - # Register this as an expected print so we don't create a duplicate archive - register_expected_print(printer_id, remote_filename, archive_id, ams_mapping=body.ams_mapping) - - # Use plate_id from request if provided, otherwise auto-detect from 3MF file - if body.plate_id is not None: - plate_id = body.plate_id - else: - # Auto-detect plate ID from 3MF file (legacy behavior for single-plate files) - plate_id = 1 - try: - with zipfile.ZipFile(file_path, "r") as zf: - for name in zf.namelist(): - if name.startswith("Metadata/plate_") and name.endswith(".gcode"): - # Extract plate number from "Metadata/plate_X.gcode" - plate_str = name[15:-6] # Remove "Metadata/plate_" and ".gcode" - plate_id = int(plate_str) - break - except (ValueError, zipfile.BadZipFile, OSError): - pass # Default to plate 1 if detection fails - - logger.info( - f"Reprint archive {archive_id}: plate_id={plate_id}, " - f"ams_mapping={body.ams_mapping}, bed_levelling={body.bed_levelling}, " - f"flow_cali={body.flow_cali}, vibration_cali={body.vibration_cali}, " - f"layer_inspect={body.layer_inspect}, timelapse={body.timelapse}" - ) - - # Start the print with options - started = printer_manager.start_print( + "Dispatched reprint archive %s for printer %s (dispatch_job_id=%s, dispatch_position=%s)", + archive_id, printer_id, - remote_filename, - plate_id, - ams_mapping=body.ams_mapping, - timelapse=body.timelapse, - bed_levelling=body.bed_levelling, - flow_cali=body.flow_cali, - vibration_cali=body.vibration_cali, - layer_inspect=body.layer_inspect, - use_ams=body.use_ams, + dispatch_result["dispatch_job_id"], + dispatch_result["dispatch_position"], ) - if not started: - raise HTTPException(500, "Failed to start print") - - # Track who started this print (Issue #206) - if user: - printer_manager.set_current_print_user(printer_id, user.id, user.username) - logger.info("Reprint started by user: %s", user.username) - return { - "status": "printing", + "status": "dispatched", "printer_id": printer_id, "archive_id": archive_id, "filename": archive.filename, + "dispatch_job_id": dispatch_result["dispatch_job_id"], + "dispatch_position": dispatch_result["dispatch_position"], } diff --git a/backend/app/api/routes/background_dispatch.py b/backend/app/api/routes/background_dispatch.py new file mode 100644 index 000000000..53c0ca3da --- /dev/null +++ b/backend/app/api/routes/background_dispatch.py @@ -0,0 +1,32 @@ +from fastapi import APIRouter, HTTPException + +from backend.app.core.auth import RequirePermissionIfAuthEnabled +from backend.app.core.permissions import Permission +from backend.app.models.user import User +from backend.app.services.background_dispatch import background_dispatch + +router = APIRouter(prefix="/background-dispatch", tags=["background-dispatch"]) + + +@router.delete("/{job_id}") +async def cancel_dispatch_job( + job_id: int, + _: User | None = RequirePermissionIfAuthEnabled(Permission.PRINTERS_CONTROL), +): + """Cancel a background-dispatch job. + + Queued jobs are cancelled immediately. Active jobs are marked for + cooperative cancellation and will stop at the next cancellation checkpoint. + """ + result = await background_dispatch.cancel_job(job_id) + + if not result["cancelled"]: + raise HTTPException(status_code=404, detail="Dispatch job not found") + + return { + "status": "cancelling" if result.get("pending") else "cancelled", + "job_id": result["job_id"], + "source_name": result["source_name"], + "printer_id": result["printer_id"], + "printer_name": result["printer_name"], + } diff --git a/backend/app/api/routes/library.py b/backend/app/api/routes/library.py index 69d7f5233..e9a7a3dbf 100644 --- a/backend/app/api/routes/library.py +++ b/backend/app/api/routes/library.py @@ -54,7 +54,7 @@ from backend.app.schemas.library import ( ZipExtractResponse, ZipExtractResult, ) -from backend.app.services.archive import ArchiveService, ThreeMFParser +from backend.app.services.archive import ThreeMFParser from backend.app.services.stl_thumbnail import generate_stl_thumbnail from backend.app.utils.threemf_tools import extract_nozzle_mapping_from_3mf @@ -1737,23 +1737,15 @@ async def print_library_file( db: AsyncSession = Depends(get_db), _: User | None = Depends(require_permission_if_auth_enabled(Permission.PRINTERS_CONTROL)), ): - """Print a library file directly. + """Dispatch a library file for send/start on a printer. - This endpoint: - 1. Creates an archive from the library file - 2. Uploads the file to the printer - 3. Starts the print + The actual send/start work is handled asynchronously by background + dispatch so the UI can continue immediately. Only sliced files (.gcode or .gcode.3mf) can be printed. """ - from backend.app.main import register_expected_print from backend.app.models.printer import Printer - from backend.app.services.bambu_ftp import ( - delete_file_async, - get_ftp_retry_settings, - upload_file_async, - with_ftp_retry, - ) + from backend.app.services.background_dispatch import DispatchEnqueueRejected, background_dispatch from backend.app.services.printer_manager import printer_manager # Use defaults if no body provided @@ -1790,131 +1782,34 @@ async def print_library_file( if not printer_manager.is_connected(printer_id): raise HTTPException(status_code=400, detail="Printer is not connected") - # Create archive from the library file - archive_service = ArchiveService(db) - archive = await archive_service.archive_print( - printer_id=printer_id, - source_file=file_path, - original_filename=lib_file.filename, - ) + plate_name = body.plate_name + if not plate_name and body.plate_id is not None: + plate_name = f"Plate {body.plate_id}" - if not archive: - raise HTTPException(status_code=500, detail="Failed to create archive") + dispatch_source_name = lib_file.filename + if plate_name: + dispatch_source_name = f"{lib_file.filename} • {plate_name}" - await db.flush() - - # Prepare remote filename - base_name = lib_file.filename - if base_name.endswith(".gcode.3mf"): - base_name = base_name[:-10] - elif base_name.endswith(".3mf"): - base_name = base_name[:-4] - remote_filename = f"{base_name}.3mf" - remote_path = f"/{remote_filename}" - - # Get FTP retry settings - ftp_retry_enabled, ftp_retry_count, ftp_retry_delay, ftp_timeout = await get_ftp_retry_settings() - - logger.info( - f"Library print FTP upload starting: printer={printer.name} ({printer.model}), " - f"ip={printer.ip_address}, file={remote_filename}, local_path={file_path}, " - f"retry_enabled={ftp_retry_enabled}, retry_count={ftp_retry_count}, timeout={ftp_timeout}" - ) - - # Delete existing file if present (avoids 553 error) - logger.debug("Deleting existing file %s if present...", remote_path) - delete_result = await delete_file_async( - printer.ip_address, - printer.access_code, - remote_path, - socket_timeout=ftp_timeout, - printer_model=printer.model, - ) - logger.debug("Delete result: %s", delete_result) - - # Upload file to printer - if ftp_retry_enabled: - uploaded = await with_ftp_retry( - upload_file_async, - printer.ip_address, - printer.access_code, - file_path, - remote_path, - socket_timeout=ftp_timeout, - printer_model=printer.model, - max_retries=ftp_retry_count, - retry_delay=ftp_retry_delay, - operation_name=f"Upload for print to {printer.name}", + try: + dispatch_result = await background_dispatch.dispatch_print_library_file( + file_id=file_id, + filename=dispatch_source_name, + printer_id=printer_id, + printer_name=printer.name, + options=body.model_dump(exclude_none=True), + requested_by_user_id=None, + requested_by_username=None, ) - else: - uploaded = await upload_file_async( - printer.ip_address, - printer.access_code, - file_path, - remote_path, - socket_timeout=ftp_timeout, - printer_model=printer.model, - ) - - if not uploaded: - logger.error( - f"FTP upload failed for library print: printer={printer.name}, model={printer.model}, " - f"ip={printer.ip_address}, file={remote_filename}. " - "Check logs above for storage diagnostics and specific error codes." - ) - raise HTTPException( - status_code=500, - detail="Failed to upload file to printer. Check if SD card is inserted and properly formatted (FAT32/exFAT). " - "See server logs for detailed diagnostics.", - ) - - # Register this as an expected print so we don't create a duplicate archive - register_expected_print(printer_id, remote_filename, archive.id, ams_mapping=body.ams_mapping) - - # Determine plate ID - if body.plate_id is not None: - plate_id = body.plate_id - else: - plate_id = 1 - try: - with zipfile.ZipFile(file_path, "r") as zf: - for name in zf.namelist(): - if name.startswith("Metadata/plate_") and name.endswith(".gcode"): - plate_str = name[15:-6] - plate_id = int(plate_str) - break - except (ValueError, zipfile.BadZipFile, OSError): - pass # Default plate_id=1 if archive is unreadable or has no gcode - - logger.info( - f"Print library file {file_id}: archive_id={archive.id}, plate_id={plate_id}, " - f"ams_mapping={body.ams_mapping}, bed_levelling={body.bed_levelling}" - ) - - # Start the print - started = printer_manager.start_print( - printer_id, - remote_filename, - plate_id, - ams_mapping=body.ams_mapping, - timelapse=body.timelapse, - bed_levelling=body.bed_levelling, - flow_cali=body.flow_cali, - vibration_cali=body.vibration_cali, - layer_inspect=body.layer_inspect, - use_ams=body.use_ams, - ) - - if not started: - raise HTTPException(status_code=500, detail="Failed to start print") - - await db.commit() + except DispatchEnqueueRejected as e: + raise HTTPException(status_code=409, detail=str(e)) from e return { - "status": "printing", + "status": "dispatched", "printer_id": printer_id, - "archive_id": archive.id, + "archive_id": None, "filename": lib_file.filename, + "dispatch_job_id": dispatch_result["dispatch_job_id"], + "dispatch_position": dispatch_result["dispatch_position"], } diff --git a/backend/app/api/routes/websocket.py b/backend/app/api/routes/websocket.py index d6d6f3a6d..f79e2e2b6 100644 --- a/backend/app/api/routes/websocket.py +++ b/backend/app/api/routes/websocket.py @@ -3,6 +3,7 @@ import logging from fastapi import APIRouter, WebSocket, WebSocketDisconnect from backend.app.core.websocket import ws_manager +from backend.app.services.background_dispatch import background_dispatch from backend.app.services.printer_manager import printer_manager, printer_state_to_dict logger = logging.getLogger(__name__) @@ -27,6 +28,15 @@ async def websocket_endpoint(websocket: WebSocket): "data": printer_state_to_dict(state, printer_id, printer_manager.get_model(printer_id)), } ) + + dispatch_state = await background_dispatch.get_state() + if (dispatch_state.get("dispatched", 0) + dispatch_state.get("processing", 0)) > 0: + await websocket.send_json( + { + "type": "background_dispatch", + "data": dispatch_state, + } + ) logger.info("Sent initial status for %s printers", len(statuses)) # Keep connection alive and handle incoming messages diff --git a/backend/app/main.py b/backend/app/main.py index 86a9960cf..3467efb27 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -5,6 +5,75 @@ from contextlib import asynccontextmanager from datetime import datetime, timedelta, timezone from logging.handlers import RotatingFileHandler +from fastapi import FastAPI +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, + auth, + background_dispatch as background_dispatch_routes, + camera, + cloud, + discovery, + external_links, + filaments, + firmware, + github_backup, + groups, + kprofiles, + library, + local_presets, + maintenance, + metrics, + notification_templates, + notifications, + pending_uploads, + print_queue, + printers, + projects, + settings as settings_routes, + smart_plugs, + spoolman, + support, + system, + updates, + users, + webhook, + websocket, +) +from backend.app.api.routes.maintenance import _get_printer_maintenance_internal, ensure_default_types +from backend.app.api.routes.support import init_debug_logging +from backend.app.core.config import APP_VERSION, settings as app_settings +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.background_dispatch import background_dispatch +from backend.app.services.bambu_ftp import download_file_async, get_ftp_retry_settings, with_ftp_retry +from backend.app.services.bambu_mqtt import PrinterState +from backend.app.services.github_backup import github_backup_service +from backend.app.services.homeassistant import homeassistant_service +from backend.app.services.mqtt_relay import mqtt_relay +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.spoolman_tracking import ( + cleanup_tracking as _cleanup_spoolman_tracking, + report_usage as _report_spoolman_usage, + store_print_data as _store_spoolman_print_data, +) +from backend.app.services.tasmota import tasmota_service + # ============================================================================= # Dependency Check - runs before other imports to give helpful error messages @@ -126,10 +195,8 @@ def check_dependencies(): check_dependencies() # ============================================================================= -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 @@ -169,9 +236,6 @@ if not app_settings.debug: logging.getLogger("paho.mqtt").setLevel(logging.WARNING) logging.info("Bambuddy starting - debug=%s, log_level=%s", app_settings.debug, 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, @@ -320,7 +384,8 @@ def register_expected_print(printer_id: int, filename: str, archive_id: int, ams _last_status_broadcast: dict[int, str] = {} -_nozzle_count_updated: set[int] = set() # Track printers where we've updated nozzle_count +# Track printers where we've updated nozzle_count +_nozzle_count_updated: set[int] = set() async def on_printer_status_change(printer_id: int, state: PrinterState): @@ -2848,7 +2913,8 @@ _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) +# Track alarm cooldowns (printer_id:ams_id:type -> last_alarm_time) +_ams_alarm_cooldown: dict[str, datetime] = {} AMS_ALARM_COOLDOWN_MINUTES = 60 # Don't send same alarm more than once per hour @@ -3232,6 +3298,9 @@ async def lifespan(app: FastAPI): # Start the print scheduler asyncio.create_task(print_scheduler.run()) + # Start background dispatch worker for send/start operations + await background_dispatch.start() + # Start the smart plug scheduler for time-based on/off smart_plug_manager.start_scheduler() @@ -3264,6 +3333,7 @@ async def lifespan(app: FastAPI): # Shutdown print_scheduler.stop() + await background_dispatch.stop() smart_plug_manager.stop_scheduler() notification_service.stop_digest_scheduler() github_backup_service.stop_scheduler() @@ -3297,7 +3367,8 @@ PUBLIC_API_ROUTES = { "/api/v1/auth/status", "/api/v1/auth/login", "/api/v1/auth/setup", # Needed for initial setup and recovery - "/api/v1/auth/advanced-auth/status", # Advanced auth status needed for login page + # Advanced auth status needed for login page + "/api/v1/auth/advanced-auth/status", "/api/v1/auth/forgot-password", # Password reset for advanced auth # Version check for updates (no sensitive data) "/api/v1/updates/version", @@ -3449,6 +3520,7 @@ app.include_router(local_presets.router, prefix=app_settings.api_prefix) app.include_router(smart_plugs.router, prefix=app_settings.api_prefix) app.include_router(print_log.router, prefix=app_settings.api_prefix) app.include_router(print_queue.router, prefix=app_settings.api_prefix) +app.include_router(background_dispatch_routes.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) diff --git a/backend/app/schemas/archive.py b/backend/app/schemas/archive.py index e42b807cc..4ba8da812 100644 --- a/backend/app/schemas/archive.py +++ b/backend/app/schemas/archive.py @@ -11,13 +11,15 @@ class ArchiveBase(BaseModel): cost: float | None = None failure_reason: str | None = None quantity: int | None = None # Number of items printed - external_url: str | None = None # User-defined link (Printables, Thingiverse, etc.) + # User-defined link (Printables, Thingiverse, etc.) + external_url: str | None = None class ArchiveUpdate(ArchiveBase): printer_id: int | None = None project_id: int | None = None - status: str | None = None # Allow changing status (e.g., clearing failed flag) + # Allow changing status (e.g., clearing failed flag) + status: str | None = None class ArchiveDuplicate(BaseModel): @@ -53,7 +55,8 @@ class ArchiveResponse(BaseModel): print_name: str | None print_time_seconds: int | None # Estimated time from slicer actual_time_seconds: int | None = None # Computed from started_at/completed_at - time_accuracy: float | None = None # Percentage: 100 = perfect, >100 = faster than estimated + # Percentage: 100 = perfect, >100 = faster than estimated + time_accuracy: float | None = None filament_used_grams: float | None filament_type: str | None filament_color: str | None @@ -73,7 +76,8 @@ class ArchiveResponse(BaseModel): makerworld_url: str | None designer: str | None - external_url: str | None = None # User-defined link (Printables, Thingiverse, etc.) + # User-defined link (Printables, Thingiverse, etc.) + external_url: str | None = None is_favorite: bool tags: str | None @@ -116,7 +120,8 @@ class ArchiveStats(BaseModel): prints_by_filament_type: dict prints_by_printer: dict # Time accuracy stats - average_time_accuracy: float | None = None # Average across all prints with data + # Average across all prints with data + average_time_accuracy: float | None = None time_accuracy_by_printer: dict | None = None # Per-printer accuracy # Energy stats total_energy_kwh: float = 0.0 @@ -181,6 +186,7 @@ class ReprintRequest(BaseModel): # Plate selection for multi-plate 3MF files # If not specified, auto-detects from file (legacy behavior for single-plate files) plate_id: int | None = None + plate_name: str | None = None # AMS slot mapping: list of tray IDs for each filament slot in the 3MF # Global tray ID = (ams_id * 4) + slot_id, external = 254 diff --git a/backend/app/schemas/library.py b/backend/app/schemas/library.py index c1048b056..1db1a8227 100644 --- a/backend/app/schemas/library.py +++ b/backend/app/schemas/library.py @@ -181,6 +181,7 @@ class FilePrintRequest(BaseModel): # Print options (same as archive reprint) plate_id: int | None = None + plate_name: str | None = None ams_mapping: list[int] | None = None bed_levelling: bool = True flow_cali: bool = False diff --git a/backend/app/services/background_dispatch.py b/backend/app/services/background_dispatch.py new file mode 100644 index 000000000..d8fca2ea5 --- /dev/null +++ b/backend/app/services/background_dispatch.py @@ -0,0 +1,857 @@ +"""Background dispatch for print/reprint jobs. + +This service is separate from the app's print queue feature. It exists only to +decouple "send/start print" operations (FTP upload + start command) from API +request latency so the UI can continue immediately after dispatch. +""" + +from __future__ import annotations + +import asyncio +import logging +import time +import zipfile +from collections import deque +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Literal + +from sqlalchemy import select + +from backend.app.core.config import settings +from backend.app.core.database import async_session +from backend.app.core.websocket import ws_manager +from backend.app.models.library import LibraryFile +from backend.app.models.printer import Printer +from backend.app.services.archive import ArchiveService +from backend.app.services.bambu_ftp import ( + delete_file_async, + get_ftp_retry_settings, + upload_file_async, + with_ftp_retry, +) +from backend.app.services.printer_manager import printer_manager + +logger = logging.getLogger(__name__) + + +class DispatchJobCancelled(Exception): + """Raised when a dispatch job is cancelled by the user.""" + + +class DispatchEnqueueRejected(Exception): + """Raised when a dispatch job should not be accepted.""" + + +@dataclass(slots=True) +class PrintDispatchJob: + id: int + kind: Literal["reprint_archive", "print_library_file"] + source_id: int + source_name: str + printer_id: int + printer_name: str + options: dict[str, Any] = field(default_factory=dict) + requested_by_user_id: int | None = None + requested_by_username: str | None = None + + +@dataclass(slots=True) +class ActiveDispatchState: + job: PrintDispatchJob + message: str + upload_bytes: int | None = None + upload_total_bytes: int | None = None + + +class BackgroundDispatchService: + def __init__(self): + self._queued_jobs: deque[PrintDispatchJob] = deque() + self._dispatcher_task: asyncio.Task | None = None + self._running_tasks: dict[int, asyncio.Task] = {} + self._lock = asyncio.Lock() + self._job_event = asyncio.Event() + self._next_job_id = 1 + self._active_jobs: dict[int, ActiveDispatchState] = {} + self._cancel_requested_job_ids: set[int] = set() + + # Progress for the current "batch" (since queue became non-empty) + self._batch_total = 0 + self._batch_completed = 0 + self._batch_failed = 0 + + @staticmethod + def _printer_is_busy_printing(printer_id: int) -> bool: + state = printer_manager.get_status(printer_id) + if not state: + return False + return state.state in ("RUNNING", "PAUSE", "PAUSED") and bool(state.gcode_file) + + async def start(self): + async with self._lock: + if self._dispatcher_task and not self._dispatcher_task.done(): + return + self._dispatcher_task = asyncio.create_task(self._dispatcher_loop(), name="background-dispatch-dispatcher") + logger.info("Background dispatch dispatcher started") + + async def stop(self): + dispatcher: asyncio.Task | None = None + running_tasks: list[asyncio.Task] = [] + async with self._lock: + dispatcher = self._dispatcher_task + self._dispatcher_task = None + running_tasks = list(self._running_tasks.values()) + self._running_tasks.clear() + self._active_jobs.clear() + self._queued_jobs.clear() + self._cancel_requested_job_ids.clear() + self._job_event.set() + + if dispatcher: + dispatcher.cancel() + for task in running_tasks: + task.cancel() + + if dispatcher: + try: + await dispatcher + except asyncio.CancelledError: + pass + + if running_tasks: + await asyncio.gather(*running_tasks, return_exceptions=True) + + logger.info("Background dispatch dispatcher stopped") + + async def dispatch_reprint_archive( + self, + *, + archive_id: int, + archive_name: str, + printer_id: int, + printer_name: str, + options: dict[str, Any], + requested_by_user_id: int | None, + requested_by_username: str | None, + ) -> dict[str, Any]: + return await self._dispatch( + kind="reprint_archive", + source_id=archive_id, + source_name=archive_name, + printer_id=printer_id, + printer_name=printer_name, + options=options, + requested_by_user_id=requested_by_user_id, + requested_by_username=requested_by_username, + ) + + async def get_state(self) -> dict[str, Any]: + """Get current dispatch queue state snapshot for newly connected clients.""" + async with self._lock: + return self._build_state_payload_unlocked() + + async def dispatch_print_library_file( + self, + *, + file_id: int, + filename: str, + printer_id: int, + printer_name: str, + options: dict[str, Any], + requested_by_user_id: int | None, + requested_by_username: str | None, + ) -> dict[str, Any]: + return await self._dispatch( + kind="print_library_file", + source_id=file_id, + source_name=filename, + printer_id=printer_id, + printer_name=printer_name, + options=options, + requested_by_user_id=requested_by_user_id, + requested_by_username=requested_by_username, + ) + + async def cancel_job(self, job_id: int) -> dict[str, Any]: + """Cancel a queued dispatch job. + + Queued jobs are removed immediately. Active jobs are cancelled + cooperatively and will stop at the next cancellation checkpoint. + """ + active_cancel_payload: dict[str, Any] | None = None + active_cancel_result: dict[str, Any] | None = None + + async with self._lock: + active_state = self._active_jobs.get(job_id) + if active_state is not None: + logger.info("Cancel requested for active dispatch job %s", job_id) + self._cancel_requested_job_ids.add(job_id) + active_job = active_state.job + active_cancel_payload = self._build_state_payload_unlocked( + recent_event={ + "status": "cancelling", + "job_id": active_job.id, + "source_name": active_job.source_name, + "printer_id": active_job.printer_id, + "printer_name": active_job.printer_name, + "message": "Cancelling current dispatch...", + } + ) + active_cancel_result = { + "cancelled": True, + "pending": True, + "job_id": active_job.id, + "source_name": active_job.source_name, + "printer_id": active_job.printer_id, + "printer_name": active_job.printer_name, + } + + if active_cancel_payload and active_cancel_result: + await ws_manager.broadcast({"type": "background_dispatch", "data": active_cancel_payload}) + return active_cancel_result + + async with self._lock: + cancelled_job: PrintDispatchJob | None = None + for job in self._queued_jobs: + if job.id == job_id: + cancelled_job = job + break + + if not cancelled_job: + logger.info("Cancel requested for unknown dispatch job %s", job_id) + return {"cancelled": False, "reason": "not_found"} + + self._queued_jobs.remove(cancelled_job) + logger.info("Cancelled queued dispatch job %s", cancelled_job.id) + self._batch_total = max(0, self._batch_total - 1) + + if self._batch_total == 0 and len(self._queued_jobs) == 0 and len(self._active_jobs) == 0: + self._batch_completed = 0 + self._batch_failed = 0 + + payload = self._build_state_payload_unlocked( + recent_event={ + "status": "cancelled", + "job_id": cancelled_job.id, + "source_name": cancelled_job.source_name, + "printer_id": cancelled_job.printer_id, + "printer_name": cancelled_job.printer_name, + "message": "Cancelled from queue", + } + ) + + await ws_manager.broadcast({"type": "background_dispatch", "data": payload}) + return { + "cancelled": True, + "pending": False, + "job_id": cancelled_job.id, + "source_name": cancelled_job.source_name, + "printer_id": cancelled_job.printer_id, + "printer_name": cancelled_job.printer_name, + } + + async def _dispatch( + self, + *, + kind: Literal["reprint_archive", "print_library_file"], + source_id: int, + source_name: str, + printer_id: int, + printer_name: str, + options: dict[str, Any], + requested_by_user_id: int | None, + requested_by_username: str | None, + ) -> dict[str, Any]: + async with self._lock: + has_pending_for_printer = any(job.printer_id == printer_id for job in self._queued_jobs) + has_active_for_printer = any(active.job.printer_id == printer_id for active in self._active_jobs.values()) + + if has_pending_for_printer or has_active_for_printer: + raise DispatchEnqueueRejected(f"Printer {printer_name} already has a background dispatch in progress") + + if self._printer_is_busy_printing(printer_id): + raise DispatchEnqueueRejected(f"Printer {printer_name} is currently busy printing") + + dispatch_position = len(self._queued_jobs) + len(self._active_jobs) + 1 + job = PrintDispatchJob( + id=self._next_job_id, + kind=kind, + source_id=source_id, + source_name=source_name, + printer_id=printer_id, + printer_name=printer_name, + options=options, + requested_by_user_id=requested_by_user_id, + requested_by_username=requested_by_username, + ) + self._next_job_id += 1 + self._batch_total += 1 + self._queued_jobs.append(job) + self._job_event.set() + + payload = self._build_state_payload_unlocked( + recent_event={ + "status": "dispatched", + "job_id": job.id, + "source_name": source_name, + "printer_id": printer_id, + "printer_name": printer_name, + "message": f"Dispatched to {printer_name}", + } + ) + + await ws_manager.broadcast({"type": "background_dispatch", "data": payload}) + + return { + "dispatch_job_id": job.id, + "dispatch_position": dispatch_position, + "status": "dispatched", + "printer_id": printer_id, + "source_id": source_id, + "source_name": source_name, + } + + async def _dispatcher_loop(self): + while True: + await self._job_event.wait() + self._job_event.clear() + + while True: + payload: dict[str, Any] | None = None + job_to_start: PrintDispatchJob | None = None + async with self._lock: + busy_printer_ids = {state.job.printer_id for state in self._active_jobs.values()} + start_index = next( + ( + idx + for idx, queued_job in enumerate(self._queued_jobs) + if queued_job.printer_id not in busy_printer_ids + ), + None, + ) + + if start_index is None: + break + + job_to_start = self._queued_jobs[start_index] + del self._queued_jobs[start_index] + self._active_jobs[job_to_start.id] = ActiveDispatchState( + job=job_to_start, + message="Preparing background dispatch...", + ) + + task = asyncio.create_task( + self._run_active_job(job_to_start), name=f"background-dispatch-job-{job_to_start.id}" + ) + self._running_tasks[job_to_start.id] = task + + payload = self._build_state_payload_unlocked( + recent_event={ + "status": "processing", + "job_id": job_to_start.id, + "source_name": job_to_start.source_name, + "printer_id": job_to_start.printer_id, + "printer_name": job_to_start.printer_name, + "message": "Preparing background dispatch...", + } + ) + + if payload: + await ws_manager.broadcast({"type": "background_dispatch", "data": payload}) + + async def _run_active_job(self, job: PrintDispatchJob): + try: + await self._process_job(job) + await self._mark_job_finished(job, failed=False, message="Background dispatch complete") + except DispatchJobCancelled: + await self._mark_job_cancelled(job) + except asyncio.CancelledError: + raise + except Exception as e: + logger.error("Background dispatch job %s failed: %s", job.id, e, exc_info=True) + await self._mark_job_finished(job, failed=True, message=str(e)) + finally: + self._job_event.set() + + async def _set_active_message(self, job: PrintDispatchJob, message: str): + async with self._lock: + active = self._active_jobs.get(job.id) + if not active: + return + active.message = message + payload = self._build_state_payload_unlocked( + recent_event={ + "status": "processing", + "job_id": active.job.id, + "source_name": active.job.source_name, + "printer_id": active.job.printer_id, + "printer_name": active.job.printer_name, + "message": message, + } + ) + await ws_manager.broadcast({"type": "background_dispatch", "data": payload}) + + async def _set_active_upload_progress(self, job: PrintDispatchJob, uploaded: int, total: int): + async with self._lock: + active = self._active_jobs.get(job.id) + if not active: + return + + active.upload_bytes = max(0, int(uploaded)) + active.upload_total_bytes = max(0, int(total)) + payload = self._build_state_payload_unlocked( + recent_event={ + "status": "processing", + "job_id": active.job.id, + "source_name": active.job.source_name, + "printer_id": active.job.printer_id, + "printer_name": active.job.printer_name, + "message": active.message, + } + ) + await ws_manager.broadcast({"type": "background_dispatch", "data": payload}) + + async def _mark_job_finished(self, job: PrintDispatchJob, *, failed: bool, message: str): + async with self._lock: + if failed: + self._batch_failed += 1 + else: + self._batch_completed += 1 + + self._active_jobs.pop(job.id, None) + self._running_tasks.pop(job.id, None) + self._cancel_requested_job_ids.discard(job.id) + + payload = self._build_state_payload_unlocked( + recent_event={ + "status": "failed" if failed else "completed", + "job_id": job.id, + "source_name": job.source_name, + "printer_id": job.printer_id, + "printer_name": job.printer_name, + "message": message, + } + ) + should_reset_batch = len(self._queued_jobs) == 0 and len(self._active_jobs) == 0 + + await ws_manager.broadcast({"type": "background_dispatch", "data": payload}) + + if should_reset_batch: + async with self._lock: + self._batch_total = 0 + self._batch_completed = 0 + self._batch_failed = 0 + + async def _mark_job_cancelled(self, job: PrintDispatchJob): + async with self._lock: + self._active_jobs.pop(job.id, None) + self._running_tasks.pop(job.id, None) + self._cancel_requested_job_ids.discard(job.id) + self._batch_total = max(0, self._batch_total - 1) + + if self._batch_total == 0 and len(self._queued_jobs) == 0 and len(self._active_jobs) == 0: + self._batch_completed = 0 + self._batch_failed = 0 + + payload = self._build_state_payload_unlocked( + recent_event={ + "status": "cancelled", + "job_id": job.id, + "source_name": job.source_name, + "printer_id": job.printer_id, + "printer_name": job.printer_name, + "message": "Cancelled during dispatch", + } + ) + + await ws_manager.broadcast({"type": "background_dispatch", "data": payload}) + + def _is_cancel_requested(self, job_id: int) -> bool: + return job_id in self._cancel_requested_job_ids + + def _raise_if_cancel_requested(self, job: PrintDispatchJob): + if self._is_cancel_requested(job.id): + raise DispatchJobCancelled(f"Dispatch job {job.id} cancelled") + + def _build_state_payload_unlocked(self, recent_event: dict[str, Any] | None = None) -> dict[str, Any]: + processing = len(self._active_jobs) + dispatched = len(self._queued_jobs) + + dispatched_jobs = [ + { + "job_id": job.id, + "kind": job.kind, + "source_id": job.source_id, + "source_name": job.source_name, + "printer_id": job.printer_id, + "printer_name": job.printer_name, + } + for job in list(self._queued_jobs) + ] + + active_jobs: list[dict[str, Any]] = [] + for active in self._active_jobs.values(): + upload_progress_pct = None + if active.upload_total_bytes and active.upload_total_bytes > 0 and active.upload_bytes is not None: + upload_progress_pct = round( + max(0.0, min(100.0, (active.upload_bytes / active.upload_total_bytes) * 100.0)), 1 + ) + + active_jobs.append( + { + "job_id": active.job.id, + "kind": active.job.kind, + "source_id": active.job.source_id, + "source_name": active.job.source_name, + "printer_id": active.job.printer_id, + "printer_name": active.job.printer_name, + "message": active.message, + "upload_bytes": active.upload_bytes, + "upload_total_bytes": active.upload_total_bytes, + "upload_progress_pct": upload_progress_pct, + } + ) + + active_jobs.sort(key=lambda item: int(item["job_id"])) + active_job = active_jobs[0] if active_jobs else None + + return { + "total": self._batch_total, + "dispatched": dispatched, + "processing": processing, + "completed": self._batch_completed, + "failed": self._batch_failed, + "dispatched_jobs": dispatched_jobs, + "active_jobs": active_jobs, + "active_job": active_job, + "recent_event": recent_event, + } + + async def _process_job(self, job: PrintDispatchJob): + if job.kind == "reprint_archive": + await self._run_reprint_archive(job) + return + if job.kind == "print_library_file": + await self._run_print_library_file(job) + return + raise RuntimeError(f"Unknown dispatch job kind: {job.kind}") + + async def _run_reprint_archive(self, job: PrintDispatchJob): + from backend.app.main import register_expected_print + + async with async_session() as db: + service = ArchiveService(db) + archive = await service.get_archive(job.source_id) + if not archive: + raise RuntimeError("Archive not found") + + printer = await db.scalar(select(Printer).where(Printer.id == job.printer_id)) + if not printer: + raise RuntimeError("Printer not found") + + printer_name = printer.name + printer_ip = printer.ip_address + printer_access_code = printer.access_code + printer_model = printer.model + archive_filename = archive.filename + + if not printer_manager.is_connected(job.printer_id): + raise RuntimeError("Printer is not connected") + + file_path = settings.base_dir / archive.file_path + if not file_path.exists(): + raise RuntimeError("Archive file not found") + + base_name = archive.filename + if base_name.endswith(".gcode.3mf"): + base_name = base_name[:-10] + elif base_name.endswith(".3mf"): + base_name = base_name[:-4] + remote_filename = f"{base_name}.3mf" + remote_path = f"/{remote_filename}" + + ftp_retry_enabled, ftp_retry_count, ftp_retry_delay, ftp_timeout = await get_ftp_retry_settings() + self._raise_if_cancel_requested(job) + + await self._set_active_message(job, f"Preparing upload to {printer_name}...") + await delete_file_async( + printer_ip, + printer_access_code, + remote_path, + socket_timeout=ftp_timeout, + printer_model=printer_model, + ) + + self._raise_if_cancel_requested(job) + + def upload_progress_callback(_uploaded: int, _total: int): + if self._is_cancel_requested(job.id): + raise DispatchJobCancelled(f"Dispatch job {job.id} cancelled during upload") + + try: + await self._set_active_message(job, f"Uploading {archive_filename} to {printer_name}...") + loop = asyncio.get_running_loop() + progress_state = {"last_emit": 0.0, "last_bytes": 0} + + def upload_progress_callback(uploaded: int, total: int): + if self._is_cancel_requested(job.id): + raise DispatchJobCancelled(f"Dispatch job {job.id} cancelled during upload") + + now = time.monotonic() + should_emit = ( + uploaded >= total + or now - progress_state["last_emit"] >= 0.2 + or uploaded - progress_state["last_bytes"] >= 256 * 1024 + ) + + if should_emit: + progress_state["last_emit"] = now + progress_state["last_bytes"] = uploaded + loop.call_soon_threadsafe( + lambda u=uploaded, t=total: asyncio.create_task(self._set_active_upload_progress(job, u, t)) + ) + + if ftp_retry_enabled: + uploaded = await with_ftp_retry( + upload_file_async, + printer_ip, + printer_access_code, + file_path, + remote_path, + progress_callback=upload_progress_callback, + socket_timeout=ftp_timeout, + printer_model=printer_model, + max_retries=ftp_retry_count, + retry_delay=ftp_retry_delay, + operation_name=f"Upload for reprint to {printer_name}", + non_retry_exceptions=(DispatchJobCancelled,), + ) + else: + uploaded = await upload_file_async( + printer_ip, + printer_access_code, + file_path, + remote_path, + progress_callback=upload_progress_callback, + socket_timeout=ftp_timeout, + printer_model=printer_model, + ) + + if uploaded: + await self._set_active_upload_progress(job, 1, 1) + + if not uploaded: + raise RuntimeError( + "Failed to upload file to printer. Check if SD card is inserted and properly formatted (FAT32/exFAT)." + ) + + register_expected_print(job.printer_id, remote_filename, job.source_id) + + plate_id = self._resolve_plate_id(file_path, job.options.get("plate_id")) + + self._raise_if_cancel_requested(job) + + await self._set_active_message(job, f"Starting print on {printer_name}...") + started = printer_manager.start_print( + job.printer_id, + remote_filename, + plate_id, + ams_mapping=job.options.get("ams_mapping"), + timelapse=job.options.get("timelapse", False), + bed_levelling=job.options.get("bed_levelling", True), + flow_cali=job.options.get("flow_cali", False), + vibration_cali=job.options.get("vibration_cali", False), + layer_inspect=job.options.get("layer_inspect", False), + use_ams=job.options.get("use_ams", True), + ) + + if not started: + raise RuntimeError("Failed to start print") + + if job.requested_by_user_id and job.requested_by_username: + printer_manager.set_current_print_user( + job.printer_id, + job.requested_by_user_id, + job.requested_by_username, + ) + except DispatchJobCancelled: + await self._set_active_message(job, f"Cancelled upload on {printer_name}.") + raise + + async def _run_print_library_file(self, job: PrintDispatchJob): + from backend.app.main import register_expected_print + + async with async_session() as db: + lib_file = await db.scalar(select(LibraryFile).where(LibraryFile.id == job.source_id)) + if not lib_file: + raise RuntimeError("File not found") + + if not self._is_sliced_file(lib_file.filename): + raise RuntimeError("Not a sliced file. Only .gcode or .gcode.3mf files can be printed.") + + file_path = Path(settings.base_dir) / lib_file.file_path + if not file_path.exists(): + raise RuntimeError("File not found on disk") + + printer = await db.scalar(select(Printer).where(Printer.id == job.printer_id)) + if not printer: + raise RuntimeError("Printer not found") + + printer_name = printer.name + printer_ip = printer.ip_address + printer_access_code = printer.access_code + printer_model = printer.model + library_filename = lib_file.filename + + if not printer_manager.is_connected(job.printer_id): + raise RuntimeError("Printer is not connected") + + await self._set_active_message(job, f"Creating archive for {lib_file.filename}...") + archive_service = ArchiveService(db) + archive = await archive_service.archive_print( + printer_id=job.printer_id, + source_file=file_path, + ) + if not archive: + raise RuntimeError("Failed to create archive") + + await db.flush() + + base_name = lib_file.filename + if base_name.endswith(".gcode.3mf"): + base_name = base_name[:-10] + elif base_name.endswith(".3mf"): + base_name = base_name[:-4] + remote_filename = f"{base_name}.3mf" + remote_path = f"/{remote_filename}" + + ftp_retry_enabled, ftp_retry_count, ftp_retry_delay, ftp_timeout = await get_ftp_retry_settings() + self._raise_if_cancel_requested(job) + + await self._set_active_message(job, f"Preparing upload to {printer_name}...") + await delete_file_async( + printer_ip, + printer_access_code, + remote_path, + socket_timeout=ftp_timeout, + printer_model=printer_model, + ) + + self._raise_if_cancel_requested(job) + + def upload_progress_callback(_uploaded: int, _total: int): + if self._is_cancel_requested(job.id): + raise DispatchJobCancelled(f"Dispatch job {job.id} cancelled during upload") + + try: + await self._set_active_message(job, f"Uploading {library_filename} to {printer_name}...") + loop = asyncio.get_running_loop() + progress_state = {"last_emit": 0.0, "last_bytes": 0} + + def upload_progress_callback(uploaded: int, total: int): + if self._is_cancel_requested(job.id): + raise DispatchJobCancelled(f"Dispatch job {job.id} cancelled during upload") + + now = time.monotonic() + should_emit = ( + uploaded >= total + or now - progress_state["last_emit"] >= 0.2 + or uploaded - progress_state["last_bytes"] >= 256 * 1024 + ) + + if should_emit: + progress_state["last_emit"] = now + progress_state["last_bytes"] = uploaded + loop.call_soon_threadsafe( + lambda u=uploaded, t=total: asyncio.create_task(self._set_active_upload_progress(job, u, t)) + ) + + if ftp_retry_enabled: + uploaded = await with_ftp_retry( + upload_file_async, + printer_ip, + printer_access_code, + file_path, + remote_path, + progress_callback=upload_progress_callback, + socket_timeout=ftp_timeout, + printer_model=printer_model, + max_retries=ftp_retry_count, + retry_delay=ftp_retry_delay, + operation_name=f"Upload for print to {printer_name}", + non_retry_exceptions=(DispatchJobCancelled,), + ) + else: + uploaded = await upload_file_async( + printer_ip, + printer_access_code, + file_path, + remote_path, + progress_callback=upload_progress_callback, + socket_timeout=ftp_timeout, + printer_model=printer_model, + ) + + if uploaded: + await self._set_active_upload_progress(job, 1, 1) + + if not uploaded: + await db.rollback() + raise RuntimeError( + "Failed to upload file to printer. Check if SD card is inserted and properly formatted (FAT32/exFAT)." + ) + + register_expected_print(job.printer_id, remote_filename, archive.id) + + plate_id = self._resolve_plate_id(file_path, job.options.get("plate_id")) + + self._raise_if_cancel_requested(job) + + await self._set_active_message(job, f"Starting print on {printer_name}...") + started = printer_manager.start_print( + job.printer_id, + remote_filename, + plate_id, + ams_mapping=job.options.get("ams_mapping"), + timelapse=job.options.get("timelapse", False), + bed_levelling=job.options.get("bed_levelling", True), + flow_cali=job.options.get("flow_cali", False), + vibration_cali=job.options.get("vibration_cali", False), + layer_inspect=job.options.get("layer_inspect", False), + use_ams=job.options.get("use_ams", True), + ) + + if not started: + await db.rollback() + raise RuntimeError("Failed to start print") + + await db.commit() + except DispatchJobCancelled: + await db.rollback() + await self._set_active_message(job, f"Cancelled upload on {printer_name}.") + raise + + @staticmethod + def _resolve_plate_id(file_path: Path, requested_plate_id: int | None) -> int: + if requested_plate_id is not None: + return requested_plate_id + + plate_id = 1 + try: + with zipfile.ZipFile(file_path, "r") as zf: + for name in zf.namelist(): + if name.startswith("Metadata/plate_") and name.endswith(".gcode"): + plate_str = name[15:-6] + plate_id = int(plate_str) + break + except (ValueError, zipfile.BadZipFile, OSError): + pass + return plate_id + + @staticmethod + def _is_sliced_file(filename: str) -> bool: + lower = filename.lower() + return lower.endswith(".gcode") or lower.endswith(".gcode.3mf") + + +background_dispatch = BackgroundDispatchService() diff --git a/backend/app/services/bambu_ftp.py b/backend/app/services/bambu_ftp.py index 5dfe559f9..cecc4ba2f 100644 --- a/backend/app/services/bambu_ftp.py +++ b/backend/app/services/bambu_ftp.py @@ -76,13 +76,16 @@ class BambuFTPClient: """FTP client for retrieving files from Bambu Lab printers.""" FTP_PORT = 990 - DEFAULT_TIMEOUT = 30 # Default timeout in seconds (increased for A1 printers) + # Default timeout in seconds (increased for A1 printers) + DEFAULT_TIMEOUT = 30 # Models that may need SSL mode fallback (try prot_p first, fall back to prot_c) # These models have varying FTP SSL behavior depending on firmware version A1_MODELS = ("A1", "A1 Mini") - # Chunk size for manual upload transfer (1MB) - # Larger chunks reduce overhead and work better with A1 printers - CHUNK_SIZE = 1024 * 1024 + # Chunk size for manual upload transfer (256KB) + # Smaller chunks increase cancellation responsiveness during uploads. + CHUNK_SIZE = 256 * 1024 + # Per-chunk data socket timeout during upload. + UPLOAD_CHUNK_TIMEOUT = 15 # Cache for working FTP modes per printer IP # Maps IP -> "prot_p" or "prot_c" @@ -359,6 +362,8 @@ class BambuFTPClient: logger.info("FTP uploading %s (%s bytes) to %s", local_path, file_size, remote_path) uploaded = 0 + callback_exception: Exception | None = None + transfer_response_ok = True # Use manual transfer instead of storbinary() for A1 compatibility # A1 printers have issues with storbinary's voidresp() hanging after transfer @@ -368,7 +373,7 @@ class BambuFTPClient: # Set explicit socket options for reliable transfer conn.setblocking(True) - conn.settimeout(120) # 2 minute timeout per chunk + conn.settimeout(self.UPLOAD_CHUNK_TIMEOUT) try: while True: @@ -382,14 +387,51 @@ class BambuFTPClient: logger.debug("FTP upload progress: %s/%s bytes", uploaded, file_size) if progress_callback: - progress_callback(uploaded, file_size) + try: + progress_callback(uploaded, file_size) + except Exception as e: + callback_exception = e + logger.info( + "FTP upload callback requested stop for %s at %s/%s bytes: %s", + remote_path, + uploaded, + file_size, + e, + ) + break except OSError as e: logger.error("FTP connection lost during upload: %s", e) - conn.close() raise + finally: + try: + conn.close() + except OSError: + pass - conn.close() + try: + self._ftp.voidresp() + except (OSError, ftplib.Error) as e: + transfer_response_ok = False + logger.debug("FTP upload final response for %s was not clean: %s", remote_path, e) + + if callback_exception is not None: + cleanup_ok = False + try: + cleanup_ok = self.delete_file(remote_path) + except Exception as cleanup_error: + logger.warning("FTP cancel cleanup failed for %s: %s", remote_path, cleanup_error) + + if cleanup_ok: + logger.info("FTP cancel cleanup succeeded for %s", remote_path) + raise callback_exception + + raise RuntimeError( + f"Upload cancelled but failed to remove partial file {remote_path} from printer" + ) from callback_exception + + if not transfer_response_ok: + return False logger.info("FTP upload complete: %s", remote_path) return True @@ -421,7 +463,7 @@ class BambuFTPClient: # Use manual transfer instead of storbinary() for A1 compatibility conn = self._ftp.transfercmd(f"STOR {remote_path}") conn.setblocking(True) - conn.settimeout(120) + conn.settimeout(self.UPLOAD_CHUNK_TIMEOUT) try: # Send data in chunks @@ -432,10 +474,12 @@ class BambuFTPClient: offset += len(chunk) except OSError as e: logger.error("FTP connection lost during upload_bytes: %s", e) - conn.close() raise - - conn.close() + finally: + try: + conn.close() + except OSError: + pass return True except (OSError, ftplib.Error): return False @@ -827,6 +871,7 @@ async def with_ftp_retry( max_retries: int = 3, retry_delay: float = 2.0, operation_name: str = "FTP operation", + non_retry_exceptions: tuple[type[BaseException], ...] = (), **kwargs, ) -> T | None: """Execute FTP operation with retry logic. @@ -837,6 +882,7 @@ async def with_ftp_retry( max_retries: Number of retry attempts (default: 3) retry_delay: Seconds to wait between retries (default: 2.0) operation_name: Name for logging purposes + non_retry_exceptions: Exception types that should immediately abort retries **kwargs: Keyword arguments for the operation Returns: @@ -856,6 +902,8 @@ async def with_ftp_retry( if attempt > 0: logger.info("%s attempt %s/%s returned failure", operation_name, attempt + 1, max_retries + 1) except Exception as e: + if non_retry_exceptions and isinstance(e, non_retry_exceptions): + raise last_error = e logger.warning("%s attempt %s/%s failed: %s", operation_name, attempt + 1, max_retries + 1, e) diff --git a/backend/tests/integration/test_background_dispatch_api.py b/backend/tests/integration/test_background_dispatch_api.py new file mode 100644 index 000000000..bdee42e35 --- /dev/null +++ b/backend/tests/integration/test_background_dispatch_api.py @@ -0,0 +1,243 @@ +"""Integration tests for background dispatch API behavior.""" + +from unittest.mock import AsyncMock, patch + +import pytest +from httpx import AsyncClient + +from backend.app.services.background_dispatch import DispatchEnqueueRejected + + +class TestBackgroundDispatchArchivesAPI: + """Tests for archive reprint dispatch endpoint.""" + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_reprint_returns_dispatched_payload( + self, async_client: AsyncClient, archive_factory, printer_factory, db_session, tmp_path + ): + """Reprint endpoint returns background dispatch metadata.""" + printer = await printer_factory() + archive = await archive_factory( + printer.id, + filename="widget.gcode.3mf", + file_path="archives/test/widget.gcode.3mf", + ) + + archive_file = tmp_path / archive.file_path + archive_file.parent.mkdir(parents=True, exist_ok=True) + archive_file.write_bytes(b"3mf-data") + + with ( + patch("backend.app.api.routes.archives.settings.base_dir", tmp_path), + patch("backend.app.services.printer_manager.printer_manager.is_connected", return_value=True), + patch( + "backend.app.services.background_dispatch.background_dispatch.dispatch_reprint_archive", + new=AsyncMock(return_value={"dispatch_job_id": 15, "dispatch_position": 1}), + ) as mock_dispatch, + ): + response = await async_client.post( + f"/api/v1/archives/{archive.id}/reprint?printer_id={printer.id}", + json={"plate_id": 2}, + ) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "dispatched" + assert data["dispatch_job_id"] == 15 + assert data["dispatch_position"] == 1 + assert data["filename"] == "widget.gcode.3mf" + + mock_dispatch.assert_awaited_once() + kwargs = mock_dispatch.await_args.kwargs + assert kwargs["archive_name"].endswith("• Plate 2") + assert kwargs["options"]["plate_id"] == 2 + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_reprint_returns_409_when_enqueue_rejected( + self, async_client: AsyncClient, archive_factory, printer_factory, db_session, tmp_path + ): + """Reprint endpoint maps enqueue rejection to HTTP 409.""" + printer = await printer_factory() + archive = await archive_factory( + printer.id, + filename="widget2.gcode.3mf", + file_path="archives/test/widget2.gcode.3mf", + ) + + archive_file = tmp_path / archive.file_path + archive_file.parent.mkdir(parents=True, exist_ok=True) + archive_file.write_bytes(b"3mf-data") + + with ( + patch("backend.app.api.routes.archives.settings.base_dir", tmp_path), + patch("backend.app.services.printer_manager.printer_manager.is_connected", return_value=True), + patch( + "backend.app.services.background_dispatch.background_dispatch.dispatch_reprint_archive", + new=AsyncMock(side_effect=DispatchEnqueueRejected("already has a background dispatch")), + ), + ): + response = await async_client.post( + f"/api/v1/archives/{archive.id}/reprint?printer_id={printer.id}", + json={"plate_id": 1}, + ) + + assert response.status_code == 409 + assert "already has a background dispatch" in response.json()["detail"] + + +class TestBackgroundDispatchLibraryAPI: + """Tests for library print dispatch endpoint.""" + + @pytest.fixture + async def library_file_factory(self, db_session): + """Factory to create library files.""" + + async def _create_file(**kwargs): + from backend.app.models.library import LibraryFile + + defaults = { + "filename": "library_part.gcode.3mf", + "file_path": "library/files/library_part.gcode.3mf", + "file_type": "gcode", + "file_size": 1024, + } + defaults.update(kwargs) + lib_file = LibraryFile(**defaults) + db_session.add(lib_file) + await db_session.commit() + await db_session.refresh(lib_file) + return lib_file + + return _create_file + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_library_print_returns_dispatched_payload( + self, async_client: AsyncClient, library_file_factory, printer_factory, db_session, tmp_path + ): + """Library print endpoint returns dispatch job metadata.""" + printer = await printer_factory() + lib_file = await library_file_factory() + + disk_path = tmp_path / lib_file.file_path + disk_path.parent.mkdir(parents=True, exist_ok=True) + disk_path.write_bytes(b"library data") + + with ( + patch("backend.app.api.routes.library.app_settings.base_dir", tmp_path), + patch("backend.app.services.printer_manager.printer_manager.is_connected", return_value=True), + patch( + "backend.app.services.background_dispatch.background_dispatch.dispatch_print_library_file", + new=AsyncMock(return_value={"dispatch_job_id": 21, "dispatch_position": 2}), + ) as mock_dispatch, + ): + response = await async_client.post( + f"/api/v1/library/files/{lib_file.id}/print?printer_id={printer.id}", + json={"plate_id": 4}, + ) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "dispatched" + assert data["dispatch_job_id"] == 21 + assert data["dispatch_position"] == 2 + assert data["archive_id"] is None + + mock_dispatch.assert_awaited_once() + kwargs = mock_dispatch.await_args.kwargs + assert kwargs["filename"].endswith("• Plate 4") + assert kwargs["options"]["plate_id"] == 4 + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_library_print_returns_409_when_enqueue_rejected( + self, async_client: AsyncClient, library_file_factory, printer_factory, db_session, tmp_path + ): + """Library print endpoint maps enqueue rejection to HTTP 409.""" + printer = await printer_factory() + lib_file = await library_file_factory(filename="another_part.gcode") + + disk_path = tmp_path / lib_file.file_path + disk_path.parent.mkdir(parents=True, exist_ok=True) + disk_path.write_bytes(b"library data") + + with ( + patch("backend.app.api.routes.library.app_settings.base_dir", tmp_path), + patch("backend.app.services.printer_manager.printer_manager.is_connected", return_value=True), + patch( + "backend.app.services.background_dispatch.background_dispatch.dispatch_print_library_file", + new=AsyncMock(side_effect=DispatchEnqueueRejected("queue conflict")), + ), + ): + response = await async_client.post( + f"/api/v1/library/files/{lib_file.id}/print?printer_id={printer.id}", + json={"plate_id": 1}, + ) + + assert response.status_code == 409 + assert "queue conflict" in response.json()["detail"] + + +class TestBackgroundDispatchCancelAPI: + """Tests for /background-dispatch cancel endpoint.""" + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_cancel_job_returns_cancelled(self, async_client: AsyncClient): + """Cancel endpoint returns cancelled for queued job.""" + with patch( + "backend.app.services.background_dispatch.background_dispatch.cancel_job", + new=AsyncMock( + return_value={ + "cancelled": True, + "pending": False, + "job_id": 9, + "source_name": "cube.gcode.3mf", + "printer_id": 1, + "printer_name": "Printer A", + } + ), + ): + response = await async_client.delete("/api/v1/background-dispatch/9") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "cancelled" + assert data["job_id"] == 9 + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_cancel_job_returns_cancelling_for_active_job(self, async_client: AsyncClient): + """Cancel endpoint returns cancelling while active upload is being interrupted.""" + with patch( + "backend.app.services.background_dispatch.background_dispatch.cancel_job", + new=AsyncMock( + return_value={ + "cancelled": True, + "pending": True, + "job_id": 10, + "source_name": "cube.gcode.3mf", + "printer_id": 1, + "printer_name": "Printer A", + } + ), + ): + response = await async_client.delete("/api/v1/background-dispatch/10") + + assert response.status_code == 200 + assert response.json()["status"] == "cancelling" + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_cancel_job_returns_404_when_not_found(self, async_client: AsyncClient): + """Cancel endpoint returns 404 for unknown job id.""" + with patch( + "backend.app.services.background_dispatch.background_dispatch.cancel_job", + new=AsyncMock(return_value={"cancelled": False, "reason": "not_found"}), + ): + response = await async_client.delete("/api/v1/background-dispatch/999") + + assert response.status_code == 404 + assert response.json()["detail"] == "Dispatch job not found" diff --git a/backend/tests/unit/services/test_background_dispatch.py b/backend/tests/unit/services/test_background_dispatch.py new file mode 100644 index 000000000..d72b38d34 --- /dev/null +++ b/backend/tests/unit/services/test_background_dispatch.py @@ -0,0 +1,158 @@ +"""Unit tests for background dispatch service.""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest + +from backend.app.services.background_dispatch import ( + ActiveDispatchState, + BackgroundDispatchService, + DispatchEnqueueRejected, + PrintDispatchJob, +) + + +@pytest.mark.asyncio +async def test_dispatch_rejects_when_printer_busy_printing(): + """Reject enqueue when target printer is already printing.""" + service = BackgroundDispatchService() + + with ( + patch( + "backend.app.services.background_dispatch.printer_manager.get_status", + return_value=SimpleNamespace(state="RUNNING", gcode_file="active.gcode.3mf"), + ), + pytest.raises(DispatchEnqueueRejected, match="currently busy printing"), + ): + await service.dispatch_reprint_archive( + archive_id=1, + archive_name="Test Archive", + printer_id=10, + printer_name="Printer A", + options={}, + requested_by_user_id=None, + requested_by_username=None, + ) + + +@pytest.mark.asyncio +async def test_dispatch_enqueues_job_and_broadcasts_state(): + """Enqueue succeeds and emits websocket queue update.""" + service = BackgroundDispatchService() + + with ( + patch("backend.app.services.background_dispatch.printer_manager.get_status", return_value=None), + patch( + "backend.app.services.background_dispatch.ws_manager.broadcast", new_callable=AsyncMock + ) as mock_broadcast, + ): + result = await service.dispatch_print_library_file( + file_id=22, + filename="cube.gcode.3mf", + printer_id=7, + printer_name="Printer B", + options={"plate_id": 2}, + requested_by_user_id=5, + requested_by_username="tester", + ) + + assert result["status"] == "dispatched" + assert result["dispatch_job_id"] == 1 + assert result["dispatch_position"] == 1 + assert len(service._queued_jobs) == 1 + + mock_broadcast.assert_awaited_once() + payload = mock_broadcast.await_args.args[0] + assert payload["type"] == "background_dispatch" + assert payload["data"]["recent_event"]["status"] == "dispatched" + + +@pytest.mark.asyncio +async def test_cancel_queued_job_removes_it_and_broadcasts(): + """Cancelling queued job removes it immediately.""" + service = BackgroundDispatchService() + + with ( + patch("backend.app.services.background_dispatch.printer_manager.get_status", return_value=None), + patch( + "backend.app.services.background_dispatch.ws_manager.broadcast", new_callable=AsyncMock + ) as mock_broadcast, + ): + result = await service.dispatch_reprint_archive( + archive_id=1, + archive_name="benchy.gcode.3mf", + printer_id=1, + printer_name="Printer 1", + options={}, + requested_by_user_id=None, + requested_by_username=None, + ) + mock_broadcast.reset_mock() + + cancel_result = await service.cancel_job(result["dispatch_job_id"]) + + assert cancel_result["cancelled"] is True + assert cancel_result["pending"] is False + assert len(service._queued_jobs) == 0 + assert service._batch_total == 0 + + mock_broadcast.assert_awaited_once() + payload = mock_broadcast.await_args.args[0] + assert payload["data"]["recent_event"]["status"] == "cancelled" + + +@pytest.mark.asyncio +async def test_cancel_active_job_marks_pending_and_sets_cancel_flag(): + """Cancelling active job marks it as pending cancellation.""" + service = BackgroundDispatchService() + job = PrintDispatchJob( + id=42, + kind="reprint_archive", + source_id=100, + source_name="gearbox.gcode.3mf", + printer_id=3, + printer_name="Printer C", + ) + service._active_jobs[job.id] = ActiveDispatchState(job=job, message="Uploading...") + + with patch( + "backend.app.services.background_dispatch.ws_manager.broadcast", new_callable=AsyncMock + ) as mock_broadcast: + result = await service.cancel_job(job.id) + + assert result["cancelled"] is True + assert result["pending"] is True + assert job.id in service._cancel_requested_job_ids + + mock_broadcast.assert_awaited_once() + payload = mock_broadcast.await_args.args[0] + assert payload["data"]["recent_event"]["status"] == "cancelling" + + +def test_resolve_plate_id_uses_request_value_when_provided(tmp_path): + """Explicit plate_id wins over auto-detection.""" + file_path = tmp_path / "dummy.3mf" + file_path.write_text("not-a-zip") + + plate_id = BackgroundDispatchService._resolve_plate_id(file_path, requested_plate_id=9) + assert plate_id == 9 + + +def test_resolve_plate_id_auto_detects_from_3mf(tmp_path): + """Auto-detect plate from Metadata/plate_X.gcode entry.""" + import zipfile + + file_path = tmp_path / "multi.3mf" + with zipfile.ZipFile(file_path, "w") as zf: + zf.writestr("Metadata/plate_7.gcode", b"G1 X0 Y0") + + plate_id = BackgroundDispatchService._resolve_plate_id(file_path, requested_plate_id=None) + assert plate_id == 7 + + +def test_is_sliced_file_recognizes_supported_extensions(): + """Only .gcode and .gcode.3mf should be accepted.""" + assert BackgroundDispatchService._is_sliced_file("part.gcode") is True + assert BackgroundDispatchService._is_sliced_file("part.gcode.3mf") is True + assert BackgroundDispatchService._is_sliced_file("part.3mf") is False diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 2e75c3ebe..82c220228 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -1630,6 +1630,15 @@ export interface NotificationTestResponse { message: string; } +export interface BackgroundDispatchResponse { + status: 'dispatched' | string; + printer_id: number; + archive_id?: number | null; + filename: string; + dispatch_job_id: number; + dispatch_position: number; +} + // Provider-specific config types for reference export interface CallMeBotConfig { phone: string; @@ -2937,6 +2946,7 @@ export const api = { printerId: number, options?: { plate_id?: number; + plate_name?: string; ams_mapping?: number[]; timelapse?: boolean; bed_levelling?: boolean; @@ -2946,7 +2956,7 @@ export const api = { use_ams?: boolean; } ) => - request<{ status: string; printer_id: number; archive_id: number; filename: string }>( + request( `/archives/${archiveId}/reprint?printer_id=${printerId}`, { method: 'POST', @@ -4008,6 +4018,7 @@ export const api = { printerId: number, options?: { plate_id?: number; + plate_name?: string; ams_mapping?: number[]; bed_levelling?: boolean; flow_cali?: boolean; @@ -4017,13 +4028,23 @@ export const api = { use_ams?: boolean; } ) => - request<{ status: string; printer_id: number; archive_id: number; filename: string }>( + request( `/library/files/${fileId}/print?printer_id=${printerId}`, { method: 'POST', body: options ? JSON.stringify(options) : undefined, } ), + cancelBackgroundDispatchJob: (jobId: number) => + request<{ + status: 'cancelled' | 'cancelling'; + job_id: number; + source_name: string; + printer_id: number; + printer_name: string; + }>(`/background-dispatch/${jobId}`, { + method: 'DELETE', + }), getLibraryFilePlates: (fileId: number) => request(`/library/files/${fileId}/plates`), getLibraryFileFilamentRequirements: (fileId: number, plateId?: number) => diff --git a/frontend/src/components/PrintModal/index.tsx b/frontend/src/components/PrintModal/index.tsx index 54cdb1a8e..5fd285f6e 100644 --- a/frontend/src/components/PrintModal/index.tsx +++ b/frontend/src/components/PrintModal/index.tsx @@ -1,27 +1,27 @@ -import { useState, useEffect, useMemo } from 'react'; -import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'; +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'; +import { AlertCircle, AlertTriangle, Calendar, Loader2, Pencil, Printer, X } from 'lucide-react'; +import { useEffect, useMemo, useState } from 'react'; import { useTranslation } from 'react-i18next'; -import { X, Printer, Loader2, Calendar, Pencil, AlertCircle, AlertTriangle } from 'lucide-react'; -import { api } from '../../api/client'; import type { PrintQueueItemCreate, PrintQueueItemUpdate } from '../../api/client'; -import { Card, CardContent } from '../Card'; -import { Button } from '../Button'; +import { api } from '../../api/client'; import { useToast } from '../../contexts/ToastContext'; import { useFilamentMapping } from '../../hooks/useFilamentMapping'; import { useMultiPrinterFilamentMapping, type PerPrinterConfig } from '../../hooks/useMultiPrinterFilamentMapping'; import { isPlaceholderDate } from '../../utils/amsHelpers'; import { toDateTimeLocalValue } from '../../utils/date'; -import { PrinterSelector } from './PrinterSelector'; -import { PlateSelector } from './PlateSelector'; +import { Button } from '../Button'; +import { Card, CardContent } from '../Card'; import { FilamentMapping } from './FilamentMapping'; +import { PlateSelector } from './PlateSelector'; +import { PrinterSelector } from './PrinterSelector'; import { PrintOptionsPanel } from './PrintOptions'; import { ScheduleOptionsPanel } from './ScheduleOptions'; import type { + AssignmentMode, PrintModalProps, PrintOptions, ScheduleOptions, ScheduleType, - AssignmentMode, } from './types'; import { DEFAULT_PRINT_OPTIONS, DEFAULT_SCHEDULE_OPTIONS } from './types'; @@ -230,6 +230,12 @@ export function PrintModal({ // Combine filament requirements from either source const effectiveFilamentReqs = isLibraryFile ? libraryFilamentReqs : archiveFilamentReqs; + const selectedPlateName = useMemo(() => { + if (selectedPlate === null || !platesData?.plates?.length) { + return undefined; + } + return platesData.plates.find((plate) => plate.index === selectedPlate)?.name || undefined; + }, [platesData, selectedPlate]); // Only fetch printer status when single printer selected (for filament mapping) const { data: printerStatus } = useQuery({ @@ -450,12 +456,15 @@ export function PrintModal({ const printerMapping = getMappingForPrinter(printerId); if (isLibraryFile) { await api.printLibraryFile(libraryFileId!, printerId, { + plate_id: selectedPlate ?? undefined, + plate_name: selectedPlateName, ams_mapping: printerMapping, ...printOptions, }); } else { await api.reprintArchive(archiveId!, printerId, { plate_id: selectedPlate ?? undefined, + plate_name: selectedPlateName, ams_mapping: printerMapping, ...printOptions, }); @@ -498,11 +507,12 @@ export function PrintModal({ if (assignmentMode === 'model') { showToast(mode === 'edit-queue-item' ? 'Queue item updated' : `Queued for any ${targetModel}`); } else { - const action = mode === 'reprint' ? 'sent to' : (mode === 'edit-queue-item' ? 'updated/queued for' : 'queued for'); - if (results.success === 1) { - showToast(mode === 'edit-queue-item' ? 'Queue item updated' : `Print ${action} printer`); + if (mode === 'edit-queue-item') { + showToast('Queue item updated'); + } else if (results.success === 1) { + showToast('Print queued for printer'); } else { - showToast(`Print ${action} ${results.success} printers`); + showToast(`Print queued for ${results.success} printers`); } } queryClient.invalidateQueries({ queryKey: ['queue'] }); @@ -743,4 +753,5 @@ export function PrintModal({ } // Re-export types for convenience -export type { PrintModalProps, PrintModalMode } from './types'; +export type { PrintModalMode, PrintModalProps } from './types'; + diff --git a/frontend/src/contexts/ToastContext.tsx b/frontend/src/contexts/ToastContext.tsx index c72439e0b..601a34ffe 100644 --- a/frontend/src/contexts/ToastContext.tsx +++ b/frontend/src/contexts/ToastContext.tsx @@ -1,5 +1,7 @@ -import { createContext, useContext, useState, useCallback, useRef, useEffect, type ReactNode } from 'react'; -import { CheckCircle, XCircle, AlertCircle, Info, X, Loader2 } from 'lucide-react'; +import { AlertCircle, CheckCircle, ChevronDown, ChevronUp, Info, Loader2, X, XCircle } from 'lucide-react'; +import { createContext, useCallback, useContext, useEffect, useRef, useState, type ReactNode } from 'react'; +import { useTranslation } from 'react-i18next'; +import { api } from '../api/client'; type ToastType = 'success' | 'error' | 'warning' | 'info' | 'loading'; @@ -8,6 +10,29 @@ interface Toast { message: string; type: ToastType; persistent?: boolean; + dispatchData?: DispatchToastData; +} + +type DispatchJobStatus = 'dispatched' | 'processing' | 'completed' | 'failed' | 'cancelled'; + +interface DispatchToastJob { + jobId: number; + sourceName: string; + printerName: string; + status: DispatchJobStatus; + message?: string; + uploadBytes?: number; + uploadTotalBytes?: number; + uploadProgressPct?: number; +} + +interface DispatchToastData { + total: number; + dispatched: number; + processing: number; + completed: number; + failed: number; + jobs: DispatchToastJob[]; } interface ToastContextType { @@ -43,8 +68,21 @@ const bgColors = { }; export function ToastProvider({ children }: { children: ReactNode }) { + const { t } = useTranslation(); const [toasts, setToasts] = useState([]); + const [isDispatchCollapsed, setIsDispatchCollapsed] = useState(false); + const [cancellingDispatchJobIds, setCancellingDispatchJobIds] = useState>(new Set()); const timeoutRefs = useRef>>(new Map()); + const dispatchToastId = 'background-dispatch'; + const lastDispatchSummaryRef = useRef(null); + + const formatBytes = useCallback((bytes: number) => { + if (!Number.isFinite(bytes) || bytes < 0) return '0 B'; + if (bytes < 1024) return `${bytes} B`; + if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`; + if (bytes < 1024 * 1024 * 1024) return `${(bytes / (1024 * 1024)).toFixed(1)} MB`; + return `${(bytes / (1024 * 1024 * 1024)).toFixed(1)} GB`; + }, []); // Clean up all timeouts on unmount useEffect(() => { @@ -88,6 +126,279 @@ export function ToastProvider({ children }: { children: ReactNode }) { setToasts((prev) => prev.filter((t) => t.id !== id)); }, []); + const cancelDispatchJob = useCallback(async (jobId: number) => { + setCancellingDispatchJobIds((prev) => { + const next = new Set(prev); + next.add(jobId); + return next; + }); + + try { + const result = await api.cancelBackgroundDispatchJob(jobId); + showToast( + result.status === 'cancelling' + ? t('backgroundDispatch.toast.cancellingUpload') + : t('backgroundDispatch.toast.cancelled'), + 'info' + ); + } catch (error) { + const message = error instanceof Error ? error.message : t('backgroundDispatch.toast.cancelFailed'); + showToast(message, 'error'); + } finally { + setCancellingDispatchJobIds((prev) => { + const next = new Set(prev); + next.delete(jobId); + return next; + }); + } + }, [showToast, t]); + + useEffect(() => { + interface DispatchEventDetail { + total?: number; + dispatched?: number; + processing?: number; + completed?: number; + failed?: number; + dispatched_jobs?: Array<{ + job_id: number; + source_name?: string; + printer_name?: string; + }>; + active_job?: { + job_id?: number; + printer_name?: string; + source_name?: string; + message?: string; + upload_bytes?: number; + upload_total_bytes?: number; + upload_progress_pct?: number; + } | null; + active_jobs?: Array<{ + job_id?: number; + printer_name?: string; + source_name?: string; + message?: string; + upload_bytes?: number; + upload_total_bytes?: number; + upload_progress_pct?: number; + }>; + recent_event?: { + status?: string; + job_id?: number; + source_name?: string; + printer_name?: string; + message?: string; + }; + } + + const updateJob = ( + jobs: DispatchToastJob[], + jobId: number, + next: Partial & { + status: DispatchJobStatus; + sourceName: string; + printerName: string; + } + ) => { + const index = jobs.findIndex((job) => job.jobId === jobId); + if (index === -1) { + return [...jobs, { jobId, ...next }]; + } + const copy = [...jobs]; + copy[index] = { + ...copy[index], + ...next, + }; + return copy; + }; + + const statusWeight = (status: DispatchJobStatus) => { + switch (status) { + case 'failed': + return 0; + case 'processing': + return 1; + case 'dispatched': + return 2; + case 'completed': + return 3; + case 'cancelled': + return 4; + } + }; + + const onDispatchEvent = (event: Event) => { + const detail = (event as CustomEvent).detail || {}; + const total = detail.total ?? 0; + const dispatched = detail.dispatched ?? 0; + const processing = detail.processing ?? 0; + const completed = detail.completed ?? 0; + const failed = detail.failed ?? 0; + + const hasActiveWork = dispatched + processing > 0; + const allDone = total > 0 && completed + failed >= total && !hasActiveWork; + + if (hasActiveWork) { + setToasts((prev) => { + const existing = prev.find((t) => t.id === dispatchToastId); + const existingJobs = existing?.dispatchData?.jobs || []; + + const dispatchedJobs: DispatchToastJob[] = (detail.dispatched_jobs || []).map((job) => ({ + jobId: job.job_id, + sourceName: job.source_name || t('backgroundDispatch.unknownFile'), + printerName: job.printer_name || t('backgroundDispatch.unknownPrinter'), + status: 'dispatched', + })); + + const activeJobsPayload = + detail.active_jobs && detail.active_jobs.length > 0 + ? detail.active_jobs + : detail.active_job?.job_id + ? [detail.active_job] + : []; + + const activeJobs: DispatchToastJob[] = activeJobsPayload + .filter((job) => typeof job.job_id === 'number') + .map((job) => ({ + jobId: job.job_id as number, + sourceName: job.source_name || t('backgroundDispatch.unknownFile'), + printerName: job.printer_name || t('backgroundDispatch.unknownPrinter'), + status: 'processing', + message: job.message, + uploadBytes: job.upload_bytes, + uploadTotalBytes: job.upload_total_bytes, + uploadProgressPct: job.upload_progress_pct, + })); + + const activeIds = new Set([...dispatchedJobs, ...activeJobs].map((job) => job.jobId)); + const historicalJobs = existingJobs.filter( + (job) => !activeIds.has(job.jobId) && ['completed', 'failed', 'cancelled'].includes(job.status) + ); + + let jobs = [...dispatchedJobs, ...activeJobs, ...historicalJobs]; + + if (detail.recent_event?.job_id && detail.recent_event?.status) { + const rawStatus = detail.recent_event.status; + const eventStatus = ( + rawStatus === 'cancelled' ? 'cancelled' : rawStatus === 'cancelling' ? 'processing' : rawStatus + ) as DispatchJobStatus; + const sourceName = detail.recent_event.source_name || t('backgroundDispatch.unknownFile'); + const printerName = detail.recent_event.printer_name || t('backgroundDispatch.unknownPrinter'); + jobs = updateJob(jobs, detail.recent_event.job_id, { + status: eventStatus, + sourceName, + printerName, + message: detail.recent_event.message, + }); + } + + activeJobs.forEach((activeJob) => { + jobs = updateJob(jobs, activeJob.jobId, { + status: 'processing', + sourceName: activeJob.sourceName, + printerName: activeJob.printerName, + message: activeJob.message, + uploadBytes: activeJob.uploadBytes, + uploadTotalBytes: activeJob.uploadTotalBytes, + uploadProgressPct: activeJob.uploadProgressPct, + }); + }); + + const dispatchData: DispatchToastData = { + total, + dispatched, + processing, + completed, + failed, + jobs: [...jobs].sort((a, b) => { + const byStatus = statusWeight(a.status) - statusWeight(b.status); + if (byStatus !== 0) { + return byStatus; + } + return a.jobId - b.jobId; + }), + }; + + const exists = prev.find((t) => t.id === dispatchToastId); + if (exists) { + return prev.map((t) => + t.id === dispatchToastId + ? { + ...t, + message: t('backgroundDispatch.startingPrints'), + type: 'loading', + persistent: true, + dispatchData, + } + : t + ); + } + return [ + ...prev, + { + id: dispatchToastId, + message: t('backgroundDispatch.startingPrints'), + type: 'loading', + persistent: true, + dispatchData, + }, + ]; + }); + return; + } + + const recentStatus = detail.recent_event?.status; + if (!hasActiveWork && recentStatus && ['cancelled', 'failed', 'completed', 'idle'].includes(recentStatus)) { + setToasts((prev) => prev.filter((t) => t.id !== dispatchToastId)); + } + + if (allDone) { + const summaryKey = `${completed}:${failed}`; + if (lastDispatchSummaryRef.current === summaryKey) { + return; + } + lastDispatchSummaryRef.current = summaryKey; + + setToasts((prev) => prev.filter((t) => t.id !== dispatchToastId)); + const doneMessage = failed > 0 + ? t('backgroundDispatch.toast.completeWithFailures', { completed, failed }) + : t('backgroundDispatch.toast.completeSuccess', { completed }); + const id = Math.random().toString(36).substr(2, 9); + setToasts((prev) => [...prev, { id, message: doneMessage, type: failed > 0 ? 'warning' : 'success' }]); + const timeout = setTimeout(() => { + setToasts((prev) => prev.filter((t) => t.id !== id)); + timeoutRefs.current.delete(id); + }, 4000); + timeoutRefs.current.set(id, timeout); + } + + if (detail.recent_event?.status === 'idle' && !hasActiveWork) { + setToasts((prev) => prev.filter((t) => t.id !== dispatchToastId)); + } + + if (!hasActiveWork) { + setCancellingDispatchJobIds(new Set()); + } + + if (detail.dispatched_jobs) { + const dispatchedIds = new Set(detail.dispatched_jobs.map((job) => job.job_id)); + setCancellingDispatchJobIds((prev) => { + const next = new Set(); + prev.forEach((id) => { + if (dispatchedIds.has(id)) { + next.add(id); + } + }); + return next; + }); + } + }; + + window.addEventListener('background-dispatch', onDispatchEvent); + return () => window.removeEventListener('background-dispatch', onDispatchEvent); + }, [t]); + return ( {children} @@ -97,16 +408,134 @@ export function ToastProvider({ children }: { children: ReactNode }) { {toasts.map((toast) => (
- {icons[toast.type]} - {toast.message} - + {toast.dispatchData ? ( + <> +
+
+ {icons[toast.type]} +
+

{t('backgroundDispatch.startingPrints')}

+

+ {t('backgroundDispatch.progressSummary', { + complete: toast.dispatchData.completed + toast.dispatchData.failed, + total: toast.dispatchData.total, + dispatched: toast.dispatchData.dispatched, + processing: toast.dispatchData.processing, + })} +

+
+
+
+ + +
+
+ + {!isDispatchCollapsed && ( +
+ {toast.dispatchData.jobs.map((job) => { + const progressByStatus: Record = { + dispatched: 15, + processing: 60, + completed: 100, + failed: 100, + cancelled: 100, + }; + const barColorByStatus: Record = { + dispatched: 'bg-bambu-gray/60', + processing: 'bg-bambu-green', + completed: 'bg-green-500', + failed: 'bg-red-500', + cancelled: 'bg-yellow-500', + }; + return ( +
+
+ + {job.sourceName} + +
+ {(job.status === 'dispatched' || job.status === 'processing') && ( + + )} + + {t(`backgroundDispatch.status.${job.status}`)} + +
+
+
+ {job.printerName} +
+ {job.message && ( +
+ {job.message} +
+ )} + {job.status === 'processing' && typeof job.uploadBytes === 'number' && typeof job.uploadTotalBytes === 'number' && job.uploadTotalBytes > 0 && ( +
+ {formatBytes(job.uploadBytes)} / {formatBytes(job.uploadTotalBytes)} + {typeof job.uploadProgressPct === 'number' ? ` (${job.uploadProgressPct.toFixed(1)}%)` : ''} +
+ )} +
+
+
+
+ ); + })} +
+ )} + + ) : ( + <> + {icons[toast.type]} + {toast.message} + + + )}
))}
diff --git a/frontend/src/hooks/useWebSocket.ts b/frontend/src/hooks/useWebSocket.ts index 8933984c7..33720e0cc 100644 --- a/frontend/src/hooks/useWebSocket.ts +++ b/frontend/src/hooks/useWebSocket.ts @@ -1,5 +1,5 @@ -import { useEffect, useRef, useCallback, useState } from 'react'; import { useQueryClient } from '@tanstack/react-query'; +import { useCallback, useEffect, useRef, useState } from 'react'; interface WebSocketMessage { type: string; @@ -249,6 +249,13 @@ export function useWebSocket() { tray_uuid: (message as unknown as { tray_uuid?: string }).tray_uuid, } })); + + case 'background_dispatch': + window.dispatchEvent( + new CustomEvent('background-dispatch', { + detail: (message as unknown as { data?: Record }).data || {}, + }) + ); break; } }, [queryClient, debouncedInvalidate, throttledPrinterStatusUpdate]); diff --git a/frontend/src/i18n/locales/de.ts b/frontend/src/i18n/locales/de.ts index f8e1de252..8cf974cac 100644 --- a/frontend/src/i18n/locales/de.ts +++ b/frontend/src/i18n/locales/de.ts @@ -890,6 +890,33 @@ export default { }, }, + backgroundDispatch: { + unknownFile: 'Unbekannte Datei', + unknownPrinter: 'Unbekannter Drucker', + startingPrints: 'Starte Drucke', + progressSummary: '{{complete}}/{{total}} abgeschlossen • Geplant: {{dispatched}} • In Bearbeitung: {{processing}}', + expandDetails: 'Dispatch-Details ausklappen', + collapseDetails: 'Dispatch-Details einklappen', + dismissToast: 'Dispatch-Hinweis schließen', + cancelDispatchJob: 'Dispatch-Job abbrechen', + cancel: 'Abbrechen', + cancelling: 'Wird abgebrochen…', + status: { + dispatched: 'Geplant', + processing: 'In Bearbeitung', + completed: 'Abgeschlossen', + failed: 'Fehlgeschlagen', + cancelled: 'Abgebrochen', + }, + toast: { + cancellingUpload: 'Upload wird abgebrochen...', + cancelled: 'Dispatch abgebrochen', + cancelFailed: 'Dispatch konnte nicht abgebrochen werden', + completeWithFailures: 'Background Dispatch abgeschlossen: {{completed}} erfolgreich, {{failed}} fehlgeschlagen', + completeSuccess: 'Background Dispatch abgeschlossen: {{completed}} erfolgreich', + }, + }, + // Statistics page stats: { title: 'Dashboard', diff --git a/frontend/src/i18n/locales/en.ts b/frontend/src/i18n/locales/en.ts index 4606a940b..58c7664a0 100644 --- a/frontend/src/i18n/locales/en.ts +++ b/frontend/src/i18n/locales/en.ts @@ -890,6 +890,33 @@ export default { }, }, + backgroundDispatch: { + unknownFile: 'Unknown file', + unknownPrinter: 'Unknown printer', + startingPrints: 'Starting prints', + progressSummary: '{{complete}}/{{total}} complete • Dispatched: {{dispatched}} • Processing: {{processing}}', + expandDetails: 'Expand dispatch details', + collapseDetails: 'Collapse dispatch details', + dismissToast: 'Dismiss dispatch toast', + cancelDispatchJob: 'Cancel dispatch job', + cancel: 'Cancel', + cancelling: 'Cancelling…', + status: { + dispatched: 'Dispatched', + processing: 'Processing', + completed: 'Completed', + failed: 'Failed', + cancelled: 'Cancelled', + }, + toast: { + cancellingUpload: 'Cancelling upload...', + cancelled: 'Dispatch cancelled', + cancelFailed: 'Failed to cancel dispatch', + completeWithFailures: 'Background dispatch complete: {{completed}} succeeded, {{failed}} failed', + completeSuccess: 'Background dispatch complete: {{completed}} succeeded', + }, + }, + // Statistics page stats: { title: 'Dashboard', diff --git a/frontend/src/i18n/locales/it.ts b/frontend/src/i18n/locales/it.ts index 8c624e666..8412ae909 100644 --- a/frontend/src/i18n/locales/it.ts +++ b/frontend/src/i18n/locales/it.ts @@ -869,6 +869,33 @@ export default { }, }, + backgroundDispatch: { + unknownFile: 'File sconosciuto', + unknownPrinter: 'Stampante sconosciuta', + startingPrints: 'Avvio stampe', + progressSummary: '{{complete}}/{{total}} completati • Inviati: {{dispatched}} • In elaborazione: {{processing}}', + expandDetails: 'Espandi dettagli dispatch', + collapseDetails: 'Comprimi dettagli dispatch', + dismissToast: 'Chiudi notifica dispatch', + cancelDispatchJob: 'Annulla job dispatch', + cancel: 'Annulla', + cancelling: 'Annullamento…', + status: { + dispatched: 'Inviato', + processing: 'In elaborazione', + completed: 'Completato', + failed: 'Fallito', + cancelled: 'Annullato', + }, + toast: { + cancellingUpload: 'Annullamento upload...', + cancelled: 'Dispatch annullato', + cancelFailed: 'Impossibile annullare il dispatch', + completeWithFailures: 'Dispatch in background completato: {{completed}} riusciti, {{failed}} falliti', + completeSuccess: 'Dispatch in background completato: {{completed}} riusciti', + }, + }, + // Statistics page stats: { title: 'Dashboard', diff --git a/frontend/src/i18n/locales/ja.ts b/frontend/src/i18n/locales/ja.ts index a0aa76673..02070b8d5 100644 --- a/frontend/src/i18n/locales/ja.ts +++ b/frontend/src/i18n/locales/ja.ts @@ -957,6 +957,32 @@ export default { noCancelItems: 'キューアイテムをキャンセルする権限がありません', }, }, + backgroundDispatch: { + unknownFile: '不明なファイル', + unknownPrinter: '不明なプリンター', + startingPrints: '印刷開始中', + progressSummary: '{{complete}}/{{total}} 完了 • 配信済み: {{dispatched}} • 処理中: {{processing}}', + expandDetails: '配信詳細を展開', + collapseDetails: '配信詳細を折りたたむ', + dismissToast: '配信トーストを閉じる', + cancelDispatchJob: '配信ジョブをキャンセル', + cancel: 'キャンセル', + cancelling: 'キャンセル中…', + status: { + dispatched: '配信済み', + processing: '処理中', + completed: '完了', + failed: '失敗', + cancelled: 'キャンセル済み', + }, + toast: { + cancellingUpload: 'アップロードをキャンセル中...', + cancelled: '配信をキャンセルしました', + cancelFailed: '配信のキャンセルに失敗しました', + completeWithFailures: 'バックグラウンド配信完了: {{completed}} 件成功、{{failed}} 件失敗', + completeSuccess: 'バックグラウンド配信完了: {{completed}} 件成功', + }, + }, stats: { title: '統計', overview: '概要',