| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441 |
- import asyncio
- import ftplib # nosec B402
- import logging
- import os
- import socket
- import ssl
- import threading
- import time
- import weakref
- from collections.abc import Awaitable, Callable
- from concurrent.futures import ThreadPoolExecutor
- from enum import Enum
- from ftplib import FTP, FTP_TLS # nosec B402
- from io import BytesIO
- from pathlib import Path
- from typing import TypeVar
- logger = logging.getLogger(__name__)
- T = TypeVar("T")
- # Every FTP call below is blocking ftplib work handed to a thread. They used to
- # run on asyncio's *default* executor, which is sized min(32, cpu_count + 4) —
- # six threads on a 2-core NAS — and is shared with every other ``to_thread`` /
- # ``run_in_executor`` caller in the app. That was survivable only because the
- # scheduler uploaded to exactly one printer at a time. Dispatching to several
- # printers at once (#2555) would park one thread per in-flight upload for
- # minutes at a stretch (a 41 MB 3MF at the ~150 KB/s a Bambu printer sustains
- # takes ~4 min), starving the default pool and stalling unrelated work.
- #
- # A dedicated pool keeps that blast radius inside the FTP layer: the scheduler's
- # own concurrency cap is what limits parallel uploads, and it can never exhaust
- # the executor everything else depends on. Threads are created lazily, so an
- # idle pool costs nothing.
- #
- # Sized well above `queue_max_concurrent_uploads` (max 16), because uploads are
- # not the only traffic here: SD browsing, timelapse/recording listing, cover
- # downloads, deletes and storage checks all run through this pool too, and on a
- # farm they fan out across every printer at once. The pool's work queue is
- # unbounded, so exceeding it does not fail — it queues. But `asyncio.wait_for`
- # starts its clock at submission, not at thread start, so a task that sits in the
- # queue can burn its whole timeout without ever running, and `list_files_async`
- # reports a timeout as an empty listing — a silent "this printer has no files".
- # Keep the headroom.
- _FTP_MAX_WORKERS = 48
- _ftp_executor = ThreadPoolExecutor(max_workers=_FTP_MAX_WORKERS, thread_name_prefix="bambu-ftp")
- # Overall upload deadline (#2529). A flat wall-clock cap punishes big files on
- # slow links rather than catching broken ones: a 96 MB 3MF at the ~75 KB/s an A1
- # sustains over WiFi legitimately needs ~20 minutes, and the old flat 600 s
- # declared it dead at ~70 MB. The deadline is therefore derived from the file
- # size against a deliberately pessimistic floor rate. This is a backstop, not the
- # failure detector — a link that has actually died is caught within
- # ``socket_timeout`` by the blocking ``sendall``, long before this fires.
- _UPLOAD_FLOOR_BYTES_PER_SEC = 25 * 1024
- _UPLOAD_MIN_TIMEOUT = 600.0
- # How long to give the worker thread to notice the cancel flag, unwind, and
- # delete its partial file. It checks the flag once per CHUNK_SIZE, so on a link
- # slow enough to have hit the deadline this is one chunk plus the delete.
- _UPLOAD_CANCEL_GRACE = 60.0
- class UploadCancelled(Exception):
- """Raised inside the upload worker to abort an in-flight transfer.
- ``upload_file`` treats any exception from its progress callback as "stop
- now": it breaks out of the send loop, deletes the partial file from the
- printer, and re-raises. That is the only way to stop a transfer — an
- executor thread cannot be cancelled from the event loop, so a bare
- ``asyncio.wait_for`` leaves it streaming (see ``upload_file_async``).
- """
- class DeleteResult(Enum):
- """Outcome of an FTP delete attempt.
- Distinguishes "file isn't on the printer" (550, recovery impossible by
- retrying) from "delete failed for some other reason" (network, auth,
- transient FTP error — worth retrying). The post-print SD-card cleanup in
- main.py used to flatten both into ``False`` and log a "may linger" WARNING
- on every successful print where the printer self-cleaned its SD card
- before our cleanup ran (#1721 reporter's A1).
- """
- DELETED = "deleted"
- NOT_FOUND = "not_found"
- FAILED = "failed"
- class FileNotOnPrinterError(Exception):
- """Raised when a remote FTP path returns 550 (file not found).
- 550 means the file does not exist at that path — retrying the same path
- will never succeed. Callers use this sentinel with with_ftp_retry's
- non_retry_exceptions to immediately move on to the next candidate path
- instead of burning the full retry budget (up to 11 × 30s per path) on
- a lookup that cannot recover.
- """
- class ImplicitFTP_TLS(FTP_TLS):
- """FTP_TLS subclass for implicit FTPS (port 990) with model-specific SSL handling.
- X1C/P1S printers (vsFTPd) require SSL with session reuse on the data channel.
- A1/A1 Mini printers have issues with SSL on the data channel entirely and
- timeout waiting for transfer completion. Set skip_session_reuse=True for A1
- printers to skip SSL on the data channel (control channel remains encrypted).
- Optionally caps the SSL context's maximum TLS version to v1.2 (P2S firmware
- 01.02.00.00 needs this — see :mod:`ftp_profiles` and #1401).
- """
- def __init__(self, *args, skip_session_reuse: bool = False, cap_tls_v1_2: bool = False, **kwargs):
- super().__init__(*args, **kwargs)
- self._sock = None
- self.skip_session_reuse = skip_session_reuse
- self.ssl_context = ssl.create_default_context()
- self.ssl_context.check_hostname = False
- self.ssl_context.verify_mode = ssl.CERT_NONE
- # ``create_default_context()`` does NOT guarantee a protocol floor: it
- # leaves ``minimum_version`` at ``MINIMUM_SUPPORTED``, and what that
- # resolves to is a property of the OpenSSL build, not of this code.
- # Measured on identical OpenSSL 3.5.6: python:3.13-slim-trixie (our
- # Docker base) reports TLSv1_2, a bare-metal venv reports
- # MINIMUM_SUPPORTED. Docker users have therefore always been floored at
- # 1.2 — every Bambu model is reachable under that floor — while
- # bare-metal and appliance installs could silently negotiate TLS 1.0.
- # State the floor rather than inheriting it.
- self.ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
- if cap_tls_v1_2:
- # With the floor above this pins the connection to exactly TLS 1.2.
- self.ssl_context.maximum_version = ssl.TLSVersion.TLSv1_2
- def connect(self, host="", port=990, timeout=-999, source_address=None):
- """Connect to host, wrapping socket in TLS immediately (implicit FTPS)."""
- if host:
- self.host = host
- if port > 0:
- self.port = port
- if timeout != -999:
- self.timeout = timeout
- if source_address:
- self.source_address = source_address
- # Create and wrap socket immediately (implicit TLS)
- self.sock = socket.create_connection((self.host, self.port), self.timeout, source_address=self.source_address)
- self.sock = self.ssl_context.wrap_socket(self.sock, server_hostname=self.host)
- self.af = self.sock.family
- self.file = self.sock.makefile("r", encoding=self.encoding)
- self.welcome = self.getresp()
- return self.welcome
- def ntransfercmd(self, cmd, rest=None):
- """Override to wrap data connection in SSL for X1C/P1S only.
- X1C/P1S printers (vsFTPd) require SSL session reuse on the data channel.
- A1/A1 Mini printers have issues with SSL on the data channel entirely -
- they timeout waiting for the transfer completion response. For A1, we
- skip SSL wrapping on the data channel (control channel remains encrypted).
- """
- conn, size = FTP.ntransfercmd(self, cmd, rest)
- if self._prot_p and not self.skip_session_reuse:
- # X1C/P1S: Wrap data channel with SSL session reuse (required by vsFTPd)
- conn = self.ssl_context.wrap_socket(
- conn,
- server_hostname=self.host,
- session=self.sock.session,
- )
- # A1/A1 Mini (skip_session_reuse=True): Don't wrap data channel in SSL
- # The control channel remains encrypted via implicit FTPS
- return conn, size
- class BambuFTPClient:
- """FTP client for retrieving files from Bambu Lab printers."""
- FTP_PORT = 990
- # 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 (64KB)
- # Smaller chunks provide smoother progress reporting — at typical printer FTP
- # speeds (~50-100KB/s) this gives a progress update roughly every second.
- CHUNK_SIZE = 64 * 1024
- # Cache for working FTP modes per printer IP
- # Maps IP -> "prot_p" or "prot_c"
- _mode_cache: dict[str, str] = {}
- def __init__(
- self,
- ip_address: str,
- access_code: str,
- timeout: float | None = None,
- printer_model: str | None = None,
- force_prot_c: bool = False,
- ):
- self.ip_address = ip_address
- self.access_code = access_code
- self.timeout = timeout if timeout is not None else self.DEFAULT_TIMEOUT
- self.printer_model = printer_model
- self.force_prot_c = force_prot_c
- self._ftp: ImplicitFTP_TLS | None = None
- def _is_a1_model(self) -> bool:
- """Check if this is an A1 series printer."""
- if not self.printer_model:
- return False
- return self.printer_model in self.A1_MODELS
- def _get_cached_mode(self) -> str | None:
- """Get cached FTP mode for this printer."""
- return self._mode_cache.get(self.ip_address)
- @classmethod
- def cache_mode(cls, ip_address: str, mode: str):
- """Cache the working FTP mode for a printer."""
- cls._mode_cache[ip_address] = mode
- logger.info("FTP mode cached for %s: %s", ip_address, mode)
- def _should_use_prot_c(self) -> bool:
- """Determine if we should use prot_c (clear) mode."""
- # If explicitly forced, use prot_c
- if self.force_prot_c:
- return True
- # Check cache first
- cached = self._get_cached_mode()
- if cached:
- return cached == "prot_c"
- # Default: try prot_p first (will fall back if needed)
- return False
- def connect(self) -> bool:
- """Connect to the printer FTP server (implicit FTPS on port 990)."""
- try:
- use_prot_c = self._should_use_prot_c()
- from backend.app.services.ftp_profiles import get_ftp_profile
- profile = get_ftp_profile(self.printer_model)
- logger.debug(
- f"FTP connecting to {self.ip_address}:{self.FTP_PORT} "
- f"(timeout={self.timeout}s, model={self.printer_model}, prot_c={use_prot_c}, "
- f"cap_tls_v1_2={profile.cap_tls_v1_2})"
- )
- self._ftp = ImplicitFTP_TLS(
- skip_session_reuse=use_prot_c,
- cap_tls_v1_2=profile.cap_tls_v1_2,
- )
- self._ftp.connect(self.ip_address, self.FTP_PORT, timeout=self.timeout)
- logger.debug("FTP connected, logging in as bblp")
- self._ftp.login("bblp", self.access_code)
- if use_prot_c:
- # Use clear (unencrypted) data channel
- logger.debug("FTP logged in, setting prot_c (clear) and passive mode")
- self._ftp.prot_c()
- else:
- # Use protected (encrypted) data channel with session reuse
- logger.debug("FTP logged in, setting prot_p (protected) and passive mode")
- self._ftp.prot_p()
- self._ftp.set_pasv(True)
- # Log welcome message for debugging
- if hasattr(self._ftp, "welcome") and self._ftp.welcome:
- logger.debug("FTP server welcome: %s", self._ftp.welcome)
- logger.info(
- f"FTP connected successfully to {self.ip_address} (model={self.printer_model}, prot_c={use_prot_c})"
- )
- return True
- except ftplib.error_perm as e:
- logger.warning("FTP connection permission error to %s: %s", self.ip_address, e)
- self._ftp = None
- return False
- except TimeoutError as e:
- logger.warning("FTP connection timed out to %s: %s", self.ip_address, e)
- self._ftp = None
- return False
- except ssl.SSLError as e:
- logger.warning("FTP SSL error connecting to %s: %s", self.ip_address, e)
- self._ftp = None
- return False
- except (OSError, ftplib.Error) as e:
- logger.warning("FTP connection failed to %s: %s (type: %s)", self.ip_address, e, type(e).__name__)
- self._ftp = None
- return False
- def disconnect(self):
- """Disconnect from the FTP server."""
- if self._ftp:
- try:
- self._ftp.quit()
- except (OSError, ftplib.Error, EOFError):
- pass # Best-effort FTP cleanup; connection may already be closed
- self._ftp = None
- def list_files(self, path: str = "/") -> list[dict]:
- """List files in a directory."""
- if not self._ftp:
- return []
- files = []
- try:
- self._ftp.cwd(path)
- items = []
- self._ftp.retrlines("LIST", items.append)
- for item in items:
- parts = item.split()
- if len(parts) >= 9:
- name = " ".join(parts[8:])
- is_dir = item.startswith("d")
- size = int(parts[4]) if not is_dir else 0
- # Parse modification time from FTP listing
- # Format: "Nov 30 10:15" or "Nov 30 2024"
- mtime = None
- try:
- from datetime import datetime
- month = parts[5]
- day = parts[6]
- time_or_year = parts[7]
- # Determine if it's time (HH:MM) or year
- if ":" in time_or_year:
- # Recent file: "Nov 30 10:15" - assume current year
- year = datetime.now().year
- time_str = f"{month} {day} {year} {time_or_year}"
- mtime = datetime.strptime(time_str, "%b %d %Y %H:%M")
- # If parsed date is in the future, use last year
- if mtime > datetime.now():
- mtime = mtime.replace(year=year - 1)
- else:
- # Older file: "Nov 30 2024" - no time, just date
- time_str = f"{month} {day} {time_or_year}"
- mtime = datetime.strptime(time_str, "%b %d %Y")
- except (ValueError, IndexError):
- pass # Non-critical: mtime parsing is best-effort; file entry works without it
- file_entry = {
- "name": name,
- "is_directory": is_dir,
- "size": size,
- "path": f"{path.rstrip('/')}/{name}",
- }
- if mtime:
- file_entry["mtime"] = mtime
- files.append(file_entry)
- logger.debug("Listed %s files in %s", len(files), path)
- except (OSError, ftplib.Error) as e:
- logger.info("FTP list_files failed for %s: %s", path, e)
- return files
- def download_file(self, remote_path: str) -> bytes | None:
- """Download a file from the printer."""
- if not self._ftp:
- return None
- try:
- buffer = BytesIO()
- self._ftp.retrbinary(f"RETR {remote_path}", buffer.write)
- return buffer.getvalue()
- except (OSError, ftplib.Error):
- return None
- def download_to_file(self, remote_path: str, local_path: Path) -> bool:
- """Download a file from the printer to local filesystem."""
- if not self._ftp:
- logger.warning("download_to_file called but FTP not connected")
- return False
- try:
- local_path.parent.mkdir(parents=True, exist_ok=True)
- with open(local_path, "wb") as f:
- self._ftp.retrbinary(f"RETR {remote_path}", f.write)
- f.flush()
- os.fsync(f.fileno())
- file_size = local_path.stat().st_size if local_path.exists() else 0
- if file_size == 0:
- logger.warning("FTP download returned 0 bytes for %s", remote_path)
- if local_path.exists():
- local_path.unlink()
- return False
- logger.info("Successfully downloaded %s to %s (%s bytes)", remote_path, local_path, file_size)
- return True
- except (OSError, ftplib.Error) as e:
- # Clean up partial file if it exists
- if local_path.exists():
- try:
- local_path.unlink()
- except OSError:
- pass # Best-effort partial file cleanup; not critical if removal fails
- # 550 means the file is not at this path. Surface as a sentinel so
- # with_ftp_retry can abandon this path immediately and the caller
- # can advance to the next candidate instead of retrying 11× at
- # 30s intervals (the pattern that cost #972's reporter ~48min).
- if isinstance(e, ftplib.error_perm) and str(e).startswith("550"):
- logger.info("FTP download failed for %s: %s (not on printer)", remote_path, e)
- raise FileNotOnPrinterError(f"{remote_path}: {e}") from e
- # Log at INFO level so we can see failures in normal logs
- logger.info("FTP download failed for %s: %s", remote_path, e)
- return False
- def diagnose_storage(self) -> dict:
- """Run storage diagnostics and return results. For debugging upload issues."""
- results = {
- "connected": self._ftp is not None,
- "can_list_root": False,
- "root_files": [],
- "can_list_cache": False,
- "storage_info": None,
- "pwd": None,
- "errors": [],
- }
- if not self._ftp:
- results["errors"].append("FTP not connected")
- return results
- # Try to get current directory
- try:
- results["pwd"] = self._ftp.pwd()
- logger.debug("FTP current directory: %s", results["pwd"])
- except (OSError, ftplib.Error) as e:
- results["errors"].append(f"PWD failed: {e}")
- logger.debug("FTP PWD failed: %s", e)
- # Try to list root directory
- try:
- self._ftp.cwd("/")
- items = []
- self._ftp.retrlines("LIST", items.append)
- results["can_list_root"] = True
- results["root_files"] = items[:10] # First 10 entries
- logger.debug("FTP root listing (%s items): %s", len(items), items[:5])
- except (OSError, ftplib.Error) as e:
- results["errors"].append(f"LIST / failed: {e}")
- logger.debug("FTP LIST / failed: %s", e)
- # Try to list /cache (should exist on all printers)
- try:
- self._ftp.cwd("/cache")
- items = []
- self._ftp.retrlines("LIST", items.append)
- results["can_list_cache"] = True
- logger.debug("FTP /cache listing: %s items", len(items))
- except (OSError, ftplib.Error) as e:
- results["errors"].append(f"LIST /cache failed: {e}")
- logger.debug("FTP LIST /cache failed: %s", e)
- # Try to get storage info
- try:
- results["storage_info"] = self.get_storage_info()
- logger.debug("FTP storage info: %s", results["storage_info"])
- except (OSError, ftplib.Error) as e:
- results["errors"].append(f"Storage info failed: {e}")
- return results
- def upload_file(
- self,
- local_path: Path,
- remote_path: str,
- progress_callback: Callable[[int, int], None] | None = None,
- ) -> bool:
- """Upload a file to the printer with optional progress callback."""
- if not self._ftp:
- logger.warning("upload_file: FTP not connected")
- return False
- try:
- file_size = local_path.stat().st_size if local_path.exists() else 0
- logger.info("FTP uploading %s (%s bytes) to %s", local_path, file_size, remote_path)
- uploaded = 0
- callback_exception: Exception | None = None
- # Use manual transfer instead of storbinary() for A1 compatibility
- # A1 printers have issues with storbinary's voidresp() hanging after transfer
- with open(local_path, "rb") as f:
- logger.debug("FTP STOR command starting for %s", remote_path)
- t0 = time.monotonic()
- conn = self._ftp.transfercmd(f"STOR {remote_path}")
- logger.info(
- "FTP data channel ready in %.1fs (PASV + TLS handshake)",
- time.monotonic() - t0,
- )
- # Set explicit socket options for reliable transfer
- conn.setblocking(True)
- conn.settimeout(self.timeout)
- try:
- while True:
- chunk = f.read(self.CHUNK_SIZE)
- if not chunk:
- logger.debug("FTP upload: final chunk reached")
- break
- conn.sendall(chunk)
- uploaded += len(chunk)
- logger.debug("FTP upload progress: %s/%s bytes", uploaded, file_size)
- if progress_callback:
- 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)
- raise
- finally:
- try:
- conn.close()
- except OSError:
- pass
- # Wait for the server's 226 "Transfer complete" response to confirm
- # the file has been flushed to the SD card. Without this, the printer
- # may try to read an incomplete file when the print command is sent,
- # causing 0500-C010 "MicroSD Card read/write exception" errors.
- # See: https://bugs.python.org/issue25458 (ftplib response desync)
- try:
- old_timeout = self._ftp.sock.gettimeout()
- # Use a generous timeout — H2D printers can take 30+ seconds
- # to send the 226 after the data channel closes.
- self._ftp.sock.settimeout(max(self.timeout, 60))
- try:
- resp = self._ftp.voidresp()
- logger.info("FTP STOR confirmed for %s: %s", remote_path, resp.strip())
- finally:
- self._ftp.sock.settimeout(old_timeout)
- except ftplib.Error as e:
- # Some P2S firmware revisions return ftplib.Error (e.g. 426
- # "Failure reading network stream") on voidresp() even when
- # the file landed fully on the SD card — the TLS data
- # channel close races the 226 confirmation (#1417 follow-up).
- # Verify via SIZE: if the server-side file size matches what
- # we just uploaded, the file is intact and we proceed with
- # a warning. If not — or SIZE itself fails — the transfer
- # was genuinely truncated and we must fail so the print
- # command doesn't go out for a partial 3MF (the original
- # reason this catch was tightened in the previous round).
- try:
- server_size = self._ftp.size(remote_path)
- except (OSError, ftplib.Error) as size_err:
- logger.debug("Post-error SIZE check failed: %s", size_err)
- server_size = None
- if server_size is not None and server_size == file_size:
- logger.warning(
- "FTP STOR returned %s for %s but file is intact on the "
- "printer (%s bytes match) — proceeding: %s",
- type(e).__name__,
- remote_path,
- file_size,
- e,
- )
- else:
- logger.error(
- "FTP STOR rejected by printer for %s: %s (%s); server size=%s expected=%s",
- remote_path,
- e,
- type(e).__name__,
- server_size,
- file_size,
- )
- raise
- except Exception as e:
- # Timeout or socket-level error reading 226 — the data was sent
- # on our side and the printer may still have written the file.
- # H2D can take 30+ seconds to send 226 after the data channel
- # closes, so we proceed with a warning rather than failing here.
- logger.warning(
- "FTP STOR confirmation not received for %s (proceeding): %s (%s)",
- remote_path,
- e,
- type(e).__name__,
- )
- if callback_exception is not None:
- cleanup_result: DeleteResult = DeleteResult.FAILED
- try:
- cleanup_result = self.delete_file(remote_path)
- except Exception as cleanup_error:
- logger.warning("FTP cancel cleanup failed for %s: %s", remote_path, cleanup_error)
- # NOT_FOUND is success here — the partial file is gone (printer
- # may have already swept on cancel), which is the goal.
- if cleanup_result in (DeleteResult.DELETED, DeleteResult.NOT_FOUND):
- logger.info("FTP cancel cleanup succeeded for %s (%s)", remote_path, cleanup_result.value)
- raise callback_exception
- raise RuntimeError(
- f"Upload cancelled but failed to remove partial file {remote_path} from printer"
- ) from callback_exception
- elapsed = time.monotonic() - t0
- speed_kbs = (file_size / 1024) / elapsed if elapsed > 0 else 0
- logger.info(
- "FTP upload complete: %s (%s bytes in %.1fs, %.0f KB/s)",
- remote_path,
- file_size,
- elapsed,
- speed_kbs,
- )
- return True
- except ftplib.error_perm as e:
- # Permanent FTP error (4xx/5xx response)
- error_code = str(e)[:3] if str(e) else "unknown"
- logger.error("FTP upload failed for %s: %s (error code: %s)", remote_path, e, error_code)
- if error_code == "553":
- logger.error(
- "FTP 553 error - Could not create file. Possible causes: "
- "1) No SD card inserted, 2) SD card full, 3) SD card not formatted correctly (needs FAT32/exFAT), "
- "4) Printer busy/not ready, 5) File path issue"
- )
- elif error_code == "550":
- logger.error("FTP 550 error - File/directory not found or permission denied")
- elif error_code == "552":
- logger.error("FTP 552 error - Storage quota exceeded (SD card full?)")
- return False
- except (OSError, ftplib.Error) as e:
- logger.error("FTP upload failed for %s: %s (type: %s)", remote_path, e, type(e).__name__)
- return False
- def upload_bytes(self, data: bytes, remote_path: str) -> bool:
- """Upload bytes to the printer."""
- if not self._ftp:
- return False
- try:
- # Use manual transfer instead of storbinary() for A1 compatibility
- conn = self._ftp.transfercmd(f"STOR {remote_path}")
- conn.setblocking(True)
- conn.settimeout(self.timeout)
- try:
- # Send data in chunks
- offset = 0
- while offset < len(data):
- chunk = data[offset : offset + self.CHUNK_SIZE]
- conn.sendall(chunk)
- offset += len(chunk)
- except OSError as e:
- logger.error("FTP connection lost during upload_bytes: %s", e)
- raise
- finally:
- try:
- conn.close()
- except OSError:
- pass
- # Wait for 226 confirmation (see upload_file for rationale).
- # ftplib.Error subclasses (e.g. 426 error_temp) mean the server
- # rejected the transfer and the file is partial — fail. Other
- # exceptions (timeout, socket-level) are tolerated as in upload_file.
- try:
- old_timeout = self._ftp.sock.gettimeout()
- self._ftp.sock.settimeout(max(self.timeout, 60))
- try:
- self._ftp.voidresp()
- finally:
- self._ftp.sock.settimeout(old_timeout)
- except ftplib.Error as e:
- # Same SIZE-verify path as upload_file (#1417 follow-up):
- # tolerate a transient 426 if the bytes are actually on the
- # printer, fail loudly if they aren't.
- try:
- server_size = self._ftp.size(remote_path)
- except (OSError, ftplib.Error) as size_err:
- logger.debug("Post-error SIZE check failed: %s", size_err)
- server_size = None
- if server_size is not None and server_size == len(data):
- logger.warning(
- "FTP STOR returned %s for %s but file is intact on the "
- "printer (%s bytes match) — proceeding: %s",
- type(e).__name__,
- remote_path,
- len(data),
- e,
- )
- else:
- logger.error(
- "FTP STOR rejected by printer for %s: %s (%s); server size=%s expected=%s",
- remote_path,
- e,
- type(e).__name__,
- server_size,
- len(data),
- )
- return False
- except Exception:
- pass # Timeout / socket-level — proceed, data was sent.
- return True
- except (OSError, ftplib.Error):
- return False
- def delete_file(self, remote_path: str) -> DeleteResult:
- """Delete a file from the printer.
- Returns :class:`DeleteResult` distinguishing the file-not-found case
- (550) from network / auth / transient FTP failure. Callers that just
- want "did it work" should check ``result == DeleteResult.DELETED``.
- """
- if not self._ftp:
- return DeleteResult.FAILED
- try:
- self._ftp.delete(remote_path)
- return DeleteResult.DELETED
- except ftplib.error_perm as e:
- if str(e).startswith("550"):
- logger.debug("FTP delete: %s not on printer (550)", remote_path)
- return DeleteResult.NOT_FOUND
- logger.warning("Failed to delete %s: %s", remote_path, e)
- return DeleteResult.FAILED
- except (OSError, ftplib.Error) as e:
- logger.warning("Failed to delete %s: %s", remote_path, e)
- return DeleteResult.FAILED
- def get_file_size(self, remote_path: str) -> int | None:
- """Get the size of a file."""
- if not self._ftp:
- return None
- try:
- return self._ftp.size(remote_path)
- except (OSError, ftplib.Error):
- return None
- def get_storage_info(self) -> dict | None:
- """Get storage information from the printer."""
- if not self._ftp:
- return None
- result = {}
- # Try AVBL command (available space) - some FTP servers support this
- try:
- response = self._ftp.sendcmd("AVBL")
- logger.debug("AVBL response: %s", response)
- # Response format: "213 <bytes available>"
- if response.startswith("213"):
- parts = response.split()
- if len(parts) >= 2:
- result["free_bytes"] = int(parts[1])
- except (OSError, ftplib.Error) as e:
- logger.debug("AVBL command not supported: %s", e)
- # Try STAT command as fallback
- try:
- response = self._ftp.sendcmd("STAT")
- logger.debug("STAT response: %s", response)
- except (OSError, ftplib.Error):
- pass # Both AVBL and STAT unsupported; storage info will rely on directory scan
- # Calculate used space by listing root directories
- try:
- total_used = 0
- dirs_to_scan = ["/cache", "/timelapse", "/model", "/data", "/data/Metadata", "/"]
- for dir_path in dirs_to_scan:
- try:
- self._ftp.cwd(dir_path)
- items = []
- self._ftp.retrlines("LIST", items.append)
- for item in items:
- parts = item.split()
- if len(parts) >= 5 and not item.startswith("d"):
- try:
- total_used += int(parts[4])
- except ValueError:
- pass # Skip entries with non-numeric size fields
- except (OSError, ftplib.Error):
- pass # Directory may not exist on this printer model; skip it
- result["used_bytes"] = total_used
- except (OSError, ftplib.Error):
- pass # Storage scan failed; return whatever info was collected above
- return result if result else None
- # Shared 3MF download cache (#972).
- #
- # Both the cover thumbnail endpoint (api/routes/printers.py) and the archive
- # metadata flow (main.py) fetch the same 3MF file over FTP during a print.
- # On slow / contended links (A1 Wi-Fi, large files) the duplicate transfers
- # compete for the printer's single FTP socket and trigger 425 "can't open
- # data channel" errors, feeding back into cause-2's retry storm.
- #
- # This cache stores the local path of a successfully-downloaded 3MF keyed
- # by (printer_id, normalized_name). Whichever flow downloads first populates
- # the cache; the other flow reuses the file read-only. Evicted on print
- # completion so a later print with the same name re-downloads fresh bytes.
- _threemf_path_cache: dict[tuple[int, str], Path] = {}
- def normalize_3mf_name(name: str) -> str:
- """Collapse various 3MF filename variants to a cache key.
- Bambu tooling produces names as bare subtask ("Part"), with .3mf, with
- .gcode.3mf, or (Studio-normalized) with spaces → underscores. All of
- these refer to the same print job on the same printer, so they must
- hash to the same cache key.
- """
- # Lowercase first so .3MF / .GCODE.3MF variants strip cleanly — a
- # real-world case since Windows-side tooling sometimes uppercases
- # extensions.
- cleaned = name.strip().lower().replace(".gcode.3mf", "").replace(".gcode", "").replace(".3mf", "")
- return cleaned.replace(" ", "_")
- def cache_3mf_download(printer_id: int, name: str, local_path: Path) -> None:
- """Record a successfully-downloaded 3MF so a sibling flow can reuse it."""
- _threemf_path_cache[(printer_id, normalize_3mf_name(name))] = local_path
- def get_cached_3mf(printer_id: int, name: str) -> Path | None:
- """Return a cached 3MF path for this printer/name if the file still exists."""
- key = (printer_id, normalize_3mf_name(name))
- cached = _threemf_path_cache.get(key)
- if cached and cached.exists() and cached.stat().st_size > 0:
- return cached
- # Evict dead entry — the file was cleaned up (temp dir clean, manual
- # deletion, restart) so the cache value is no longer usable.
- if cached:
- _threemf_path_cache.pop(key, None)
- return None
- def clear_3mf_cache(printer_id: int | None = None, delete_files: bool = True) -> None:
- """Drop cache entries for one printer (or all with None).
- When ``delete_files`` is True (default) the on-disk 3MF is removed as well
- — called from on_print_complete so temp files don't accumulate across
- prints. Tests that want to inspect the cache contents disable this.
- Only paths inside ``archive_dir/temp`` are unlinked. The dispatch sites
- added in #1166 also cache the live archive copy and library file bytes
- so /cover can skip FTP — those are *user data*, never the cache's to
- delete. Pre-fix this branch silently removed archive 3mfs on every print
- completion (#1212 + private reports of "file disappeared overnight").
- """
- from backend.app.core.config import settings as _config_settings
- temp_root = _config_settings.archive_dir / "temp"
- def _is_temp_path(path: Path) -> bool:
- try:
- return path.is_relative_to(temp_root)
- except (OSError, ValueError):
- return False
- def _maybe_unlink(path: Path) -> None:
- if not delete_files or not path.exists():
- return
- if not _is_temp_path(path):
- return
- try:
- path.unlink()
- except OSError as exc:
- logger.debug("3MF cache cleanup skipped %s: %s", path, exc)
- if printer_id is None:
- for path in list(_threemf_path_cache.values()):
- _maybe_unlink(path)
- _threemf_path_cache.clear()
- return
- for key in [k for k in _threemf_path_cache if k[0] == printer_id]:
- _maybe_unlink(_threemf_path_cache[key])
- _threemf_path_cache.pop(key, None)
- async def download_file_async(
- ip_address: str,
- access_code: str,
- remote_path: str,
- local_path: Path,
- timeout: float = 60.0,
- socket_timeout: float | None = None,
- printer_model: str | None = None,
- ) -> bool:
- """Async wrapper for downloading a file with timeout.
- For A1/A1 Mini printers, automatically tries prot_p first, then falls back
- to prot_c if the download fails. The working mode is cached for future operations.
- Args:
- ip_address: Printer IP address
- access_code: Printer access code
- remote_path: Remote file path on printer
- local_path: Local path to save file
- timeout: Overall operation timeout (asyncio)
- socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
- printer_model: Printer model for A1-specific workarounds
- """
- loop = asyncio.get_event_loop()
- is_a1 = printer_model in BambuFTPClient.A1_MODELS if printer_model else False
- # Per-attempt completion state: asyncio.wait_for cannot cancel
- # run_in_executor threads, so on timeout the executor may still complete
- # the download after we stop waiting. The thread flips `success` to True
- # ONLY after the file is fully written — a post-timeout check lets us
- # salvage the download without mistaking an in-progress partial write
- # for a completed one. Each attempt gets its own dict and event so a
- # zombie from an earlier attempt can't flip the flag for a later one.
- # The event is set in `_download`'s finally block so the post-timeout
- # path can wait for genuine thread completion instead of a fixed sleep.
- def _download(force_prot_c: bool, completion: dict, done: threading.Event) -> bool:
- mode_str = "prot_c" if force_prot_c else "prot_p"
- try:
- client = BambuFTPClient(
- ip_address,
- access_code,
- timeout=socket_timeout,
- printer_model=printer_model,
- force_prot_c=force_prot_c,
- )
- if client.connect():
- try:
- result = client.download_to_file(remote_path, local_path)
- if result:
- BambuFTPClient.cache_mode(ip_address, mode_str)
- completion["success"] = True
- return result
- finally:
- client.disconnect()
- return False
- finally:
- done.set()
- async def _run(force_prot_c: bool) -> bool:
- completion = {"success": False}
- done = threading.Event()
- try:
- return await asyncio.wait_for(
- loop.run_in_executor(_ftp_executor, _download, force_prot_c, completion, done), timeout=timeout
- )
- except TimeoutError:
- # Slow WiFi links commonly overshoot ftp_timeout by 10–30 s without
- # actually being stuck, so starting attempt 2 now would just contend
- # with the still-progressing RETR on attempt 1 and produce the
- # zombie-write race reported in #1014 (file landed on disk minutes
- # after the retry loop had already given up). Wait for the worker
- # thread to genuinely finish — capped at 30 s so a truly stuck
- # connection can't stall a whole attempt indefinitely, with a 0.5 s
- # floor so artificially small test timeouts still give zombies a
- # realistic window to finish.
- grace = max(min(timeout, 30.0), 0.5)
- # Deliberately the DEFAULT executor, not `_ftp_executor`: this thread
- # blocks waiting on `_download`, which is itself an `_ftp_executor`
- # worker. Parking waiters in the same bounded pool as the workers they
- # wait for is how you build a deadlock — with enough concurrent
- # timeouts the waiters would occupy every slot and the downloads they
- # are waiting for could never be scheduled.
- await loop.run_in_executor(None, done.wait, grace)
- if completion["success"] and local_path.exists() and local_path.stat().st_size > 0:
- logger.info(
- "FTP download wait_for timed out after %ss for %s, but thread completed within %ss grace (%s bytes) — salvaging",
- timeout,
- remote_path,
- grace,
- local_path.stat().st_size,
- )
- return True
- logger.warning(
- "FTP download timed out after %ss (plus %ss grace) for %s",
- timeout,
- grace,
- remote_path,
- )
- return False
- # Check if we have a cached mode for this printer
- cached_mode = BambuFTPClient._mode_cache.get(ip_address)
- if cached_mode:
- force_prot_c = cached_mode == "prot_c"
- return await _run(force_prot_c)
- # No cached mode - try prot_p first
- if await _run(False):
- return True
- # Download failed - for A1 models, try prot_c fallback
- if is_a1:
- logger.info("FTP download failed with prot_p for A1 model, trying prot_c fallback...")
- return await _run(True)
- return False
- async def download_file_try_paths_async(
- ip_address: str,
- access_code: str,
- remote_paths: list[str],
- local_path: Path,
- socket_timeout: float | None = None,
- printer_model: str | None = None,
- timeout: float = 90.0,
- ) -> bool:
- """Try downloading a file from multiple paths using a single connection.
- Args:
- socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
- printer_model: Printer model for A1-specific workarounds
- timeout: overall async cap. The per-socket timeout only bounds an
- in-flight worker; it does NOT bound how long this coroutine waits
- for a free slot in the fixed-size ``_ftp_executor``. On a large
- farm where offline printers keep every worker busy on dead
- connects, that queue wait is otherwise unbounded — and any caller
- holding a DB connection while awaiting this would pin it until the
- pool is exhausted (#2572). The cap converts that into a bounded
- wait; the orphaned worker finishes and its result is discarded.
- """
- loop = asyncio.get_event_loop()
- def _download():
- client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
- if not client.connect():
- return False
- try:
- # FileNotOnPrinterError signals "try the next path", not "give up" —
- # this function's whole purpose is to walk a list of candidates
- # over one connection. Only a real transport error should bubble.
- for remote_path in remote_paths:
- try:
- if client.download_to_file(remote_path, local_path):
- return True
- except FileNotOnPrinterError:
- continue
- return False
- finally:
- client.disconnect()
- try:
- return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _download), timeout=timeout)
- except TimeoutError:
- logger.warning("FTP download_try_paths exceeded its %ss cap for %s (#2572)", timeout, ip_address)
- return False
- def _upload_deadline(local_path: Path) -> float:
- """Derive an upload deadline from the file size (#2529).
- See ``_UPLOAD_FLOOR_BYTES_PER_SEC``. An unstat-able file falls back to the
- floor timeout — ``upload_file`` will fail on the open() anyway.
- """
- try:
- size = local_path.stat().st_size
- except OSError:
- return _UPLOAD_MIN_TIMEOUT
- return max(_UPLOAD_MIN_TIMEOUT, size / _UPLOAD_FLOOR_BYTES_PER_SEC)
- # One upload at a time per printer. Two concurrent STOR commands for the same
- # remote path leave a corrupt file on the SD card, and the printer reads as
- # flaky rather than busy (#2529). Held for the duration of a transfer, so a
- # second dispatch to the same printer queues behind the first instead of racing
- # it. Keyed per event loop: an asyncio.Lock binds to the loop that first awaits
- # it, and the test suite runs each case on a fresh loop.
- _upload_locks: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[str, asyncio.Lock]] = (
- weakref.WeakKeyDictionary()
- )
- def _upload_lock(loop: asyncio.AbstractEventLoop, ip_address: str) -> asyncio.Lock:
- per_loop = _upload_locks.setdefault(loop, {})
- lock = per_loop.get(ip_address)
- if lock is None:
- lock = asyncio.Lock()
- per_loop[ip_address] = lock
- return lock
- async def upload_file_async(
- ip_address: str,
- access_code: str,
- local_path: Path,
- remote_path: str,
- timeout: float | None = None,
- progress_callback: Callable[[int, int], None] | None = None,
- socket_timeout: float | None = None,
- printer_model: str | None = None,
- ) -> bool:
- """Async wrapper for uploading a file with timeout and progress callback.
- For A1/A1 Mini printers, automatically tries prot_p first, then falls back
- to prot_c if the upload fails. The working mode is cached for future uploads.
- Args:
- ip_address: Printer IP address
- access_code: Printer access code
- local_path: Local file path to upload
- remote_path: Remote path on printer
- timeout: Overall deadline. ``None`` (the default) derives it from the
- file size — see ``_upload_deadline``. A caller that passes a number
- gets exactly that, which is what the tests rely on.
- progress_callback: Optional callback for progress updates
- socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
- printer_model: Printer model for A1-specific workarounds
- """
- loop = asyncio.get_event_loop()
- is_a1 = printer_model in BambuFTPClient.A1_MODELS if printer_model else False
- deadline = _upload_deadline(local_path) if timeout is None else timeout
- # Set when the deadline expires. The worker checks it once per chunk.
- cancel = threading.Event()
- def _guarded_progress(uploaded: int, total: int) -> None:
- if cancel.is_set():
- raise UploadCancelled(f"upload of {remote_path} exceeded its {deadline:.0f}s deadline")
- if progress_callback:
- progress_callback(uploaded, total)
- def _upload(force_prot_c: bool = False) -> bool:
- mode_str = "prot_c" if force_prot_c else "prot_p"
- logger.info(
- f"FTP connecting to {ip_address} for upload (model={printer_model}, "
- f"mode={mode_str}, socket_timeout={socket_timeout}s, deadline={deadline:.0f}s)..."
- )
- client = BambuFTPClient(
- ip_address, access_code, timeout=socket_timeout, printer_model=printer_model, force_prot_c=force_prot_c
- )
- if client.connect():
- logger.info("FTP connected to %s", ip_address)
- try:
- result = client.upload_file(local_path, remote_path, _guarded_progress)
- if result:
- # Cache the working mode
- BambuFTPClient.cache_mode(ip_address, mode_str)
- return result
- finally:
- client.disconnect()
- logger.warning("FTP connection failed to %s", ip_address)
- return False
- async def _attempt(force_prot_c: bool) -> bool:
- """Run one upload attempt, and make a timeout actually stop the transfer.
- ``asyncio.wait_for`` cancels the *future*, never the executor thread
- behind it. Before #2529 a slow-but-healthy upload that overran the
- deadline left that thread streaming: it kept pushing bytes, kept firing
- the progress callback, and the retry above put a *second* STOR of the
- same file onto the same printer. The reporter's 96 MB job ran four
- concurrent transfers and never landed. So on timeout we signal the
- worker (it raises ``UploadCancelled`` from the progress callback, which
- breaks the send loop and deletes the partial file) and wait for it to
- actually go.
- """
- fut = loop.run_in_executor(_ftp_executor, lambda: _upload(force_prot_c))
- try:
- return await asyncio.wait_for(asyncio.shield(fut), timeout=deadline)
- except TimeoutError:
- cancel.set()
- logger.warning(
- "FTP upload of %s exceeded its %.0fs deadline — cancelling the transfer",
- remote_path,
- deadline,
- )
- try:
- await asyncio.wait_for(asyncio.shield(fut), timeout=_UPLOAD_CANCEL_GRACE)
- except UploadCancelled:
- logger.info("FTP upload of %s cancelled; partial file removed from the printer", remote_path)
- except TimeoutError:
- # The thread is wedged somewhere that never reaches the callback
- # (a blocked sendall, say). Nothing more we can do from here —
- # but consume the eventual result so asyncio doesn't log the
- # future's exception as unretrieved when it is garbage-collected.
- logger.error(
- "FTP upload thread for %s did not stop within %.0fs of the cancel signal",
- remote_path,
- _UPLOAD_CANCEL_GRACE,
- )
- fut.add_done_callback(_swallow_future_result)
- except Exception as e:
- logger.warning("FTP upload of %s errored while cancelling: %s", remote_path, e)
- # Raise rather than return False: a deadline expiry means the link
- # sustained less than the floor rate for the whole transfer, and a
- # retry would only spend another full deadline finding that out
- # again — with check_queue serialized, four of those block the
- # entire print queue for hours. ``with_ftp_retry`` never retries it.
- raise UploadCancelled(
- f"Upload of {remote_path} to {ip_address} exceeded its {deadline:.0f}s deadline "
- f"(link sustained less than {_UPLOAD_FLOOR_BYTES_PER_SEC // 1024} KB/s)"
- ) from None
- async with _upload_lock(loop, ip_address):
- # Check if we have a cached mode for this printer
- cached_mode = BambuFTPClient._mode_cache.get(ip_address)
- if cached_mode:
- # Use cached mode
- return await _attempt(cached_mode == "prot_c")
- # No cached mode - try prot_p first
- if await _attempt(False):
- return True
- # Upload failed - for A1 models, try prot_c fallback
- if is_a1:
- logger.info("FTP upload failed with prot_p for A1 model, trying prot_c fallback...")
- return await _attempt(True)
- return False
- def _swallow_future_result(fut: asyncio.Future) -> None:
- """Retrieve a future's exception so asyncio doesn't log it as unhandled."""
- if not fut.cancelled():
- fut.exception()
- async def list_files_async(
- ip_address: str,
- access_code: str,
- path: str = "/",
- timeout: float = 30.0,
- socket_timeout: float | None = None,
- printer_model: str | None = None,
- ) -> list[dict]:
- """Async wrapper for listing files with timeout.
- Args:
- socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
- printer_model: Printer model for A1-specific workarounds
- """
- loop = asyncio.get_event_loop()
- def _list():
- client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
- if client.connect():
- try:
- return client.list_files(path)
- finally:
- client.disconnect()
- return []
- try:
- return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _list), timeout=timeout)
- except TimeoutError:
- logger.warning("FTP list_files timed out after %ss for %s", timeout, path)
- return []
- async def delete_file_async(
- ip_address: str,
- access_code: str,
- remote_path: str,
- socket_timeout: float | None = None,
- printer_model: str | None = None,
- timeout: float = 60.0,
- ) -> DeleteResult:
- """Async wrapper for deleting a file.
- Returns :class:`DeleteResult` so callers can distinguish ``NOT_FOUND``
- (550 — file isn't on the printer, no retry value) from ``FAILED``
- (network / auth / transient — worth retrying or surfacing).
- Args:
- socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
- printer_model: Printer model for A1-specific workarounds
- timeout: overall async cap so a saturated ``_ftp_executor`` can't pin
- the caller (and any DB connection it holds) indefinitely (#2572).
- """
- loop = asyncio.get_event_loop()
- def _delete() -> DeleteResult:
- client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
- if client.connect():
- try:
- return client.delete_file(remote_path)
- finally:
- client.disconnect()
- return DeleteResult.FAILED
- try:
- return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _delete), timeout=timeout)
- except TimeoutError:
- logger.warning("FTP delete_file exceeded its %ss cap for %s (#2572)", timeout, ip_address)
- return DeleteResult.FAILED
- async def download_file_bytes_async(
- ip_address: str,
- access_code: str,
- remote_path: str,
- socket_timeout: float | None = None,
- printer_model: str | None = None,
- timeout: float = 300.0,
- ) -> bytes | None:
- """Async wrapper for downloading file as bytes.
- Args:
- socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
- printer_model: Printer model for A1-specific workarounds
- timeout: overall async cap so a saturated ``_ftp_executor`` can't pin
- the caller (and any DB connection it holds) indefinitely (#2572).
- Generous by default because this pulls whole files (timelapse
- video, gcode) which can legitimately take minutes over slow Wi-Fi —
- the cap only guards against a permanently-starved pool, not a
- slow-but-progressing transfer.
- """
- loop = asyncio.get_event_loop()
- def _download():
- client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
- if client.connect():
- try:
- return client.download_file(remote_path)
- finally:
- client.disconnect()
- return None
- try:
- return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _download), timeout=timeout)
- except TimeoutError:
- logger.warning("FTP download_bytes exceeded its %ss cap for %s (#2572)", timeout, ip_address)
- return None
- async def get_storage_info_async(
- ip_address: str,
- access_code: str,
- socket_timeout: float | None = None,
- printer_model: str | None = None,
- timeout: float = 60.0,
- ) -> dict | None:
- """Async wrapper for getting storage info.
- Args:
- socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
- printer_model: Printer model for A1-specific workarounds
- timeout: overall async cap so a saturated ``_ftp_executor`` can't pin
- the caller (and any DB connection it holds) indefinitely (#2572).
- """
- loop = asyncio.get_event_loop()
- def _get_storage():
- client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
- if client.connect():
- try:
- return client.get_storage_info()
- finally:
- client.disconnect()
- return None
- try:
- return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _get_storage), timeout=timeout)
- except TimeoutError:
- logger.warning("FTP get_storage_info exceeded its %ss cap for %s (#2572)", timeout, ip_address)
- return None
- async def get_ftp_retry_settings() -> tuple[bool, int, float, float]:
- """Get FTP retry settings from database.
- Returns:
- Tuple of (retry_enabled, retry_count, retry_delay, timeout)
- """
- from backend.app.api.routes.settings import get_setting
- from backend.app.core.database import async_session
- async with async_session() as db:
- enabled = (await get_setting(db, "ftp_retry_enabled") or "true") == "true"
- count = int(await get_setting(db, "ftp_retry_count") or "3")
- delay = float(await get_setting(db, "ftp_retry_delay") or "2")
- timeout = float(await get_setting(db, "ftp_timeout") or "30")
- return enabled, count, delay, timeout
- async def with_ftp_retry(
- operation: Callable[..., Awaitable[T]],
- *args,
- 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.
- Args:
- operation: Async function to execute
- *args: Positional arguments for the operation
- 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:
- Result of the operation, or None if all attempts fail
- ``UploadCancelled`` is never retried, whatever the caller passes: it means
- the transfer overran its size-derived deadline, so a retry would spend
- another full deadline reaching the same conclusion (#2529).
- """
- last_error = None
- for attempt in range(max_retries + 1):
- try:
- result = await operation(*args, **kwargs)
- # Check for "falsy" success indicators
- if result not in (False, None, []):
- if attempt > 0:
- logger.info("%s succeeded on attempt %s/%s", operation_name, attempt + 1, max_retries + 1)
- return result
- # Operation returned failure indicator
- if attempt > 0:
- logger.info("%s attempt %s/%s returned failure", operation_name, attempt + 1, max_retries + 1)
- except UploadCancelled:
- raise
- 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)
- # Don't wait after the last attempt
- if attempt < max_retries:
- logger.info("%s will retry in %ss...", operation_name, retry_delay)
- await asyncio.sleep(retry_delay)
- logger.error("%s failed after %s attempts", operation_name, max_retries + 1)
- if last_error:
- logger.debug("Last error: %s", last_error)
- return None
|