bambu_ftp.py 67 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585
  1. import asyncio
  2. import ftplib # nosec B402
  3. import logging
  4. import os
  5. import socket
  6. import ssl
  7. import threading
  8. import time
  9. import weakref
  10. from collections.abc import Awaitable, Callable
  11. from concurrent.futures import ThreadPoolExecutor
  12. from enum import Enum
  13. from ftplib import FTP, FTP_TLS # nosec B402
  14. from io import BytesIO
  15. from pathlib import Path
  16. from typing import TypeVar
  17. logger = logging.getLogger(__name__)
  18. T = TypeVar("T")
  19. # Every FTP call below is blocking ftplib work handed to a thread. They used to
  20. # run on asyncio's *default* executor, which is sized min(32, cpu_count + 4) —
  21. # six threads on a 2-core NAS — and is shared with every other ``to_thread`` /
  22. # ``run_in_executor`` caller in the app. That was survivable only because the
  23. # scheduler uploaded to exactly one printer at a time. Dispatching to several
  24. # printers at once (#2555) would park one thread per in-flight upload for
  25. # minutes at a stretch (a 41 MB 3MF at the ~150 KB/s a Bambu printer sustains
  26. # takes ~4 min), starving the default pool and stalling unrelated work.
  27. #
  28. # A dedicated pool keeps that blast radius inside the FTP layer: the scheduler's
  29. # own concurrency cap is what limits parallel uploads, and it can never exhaust
  30. # the executor everything else depends on. Threads are created lazily, so an
  31. # idle pool costs nothing.
  32. #
  33. # Sized well above `queue_max_concurrent_uploads` (max 16), because uploads are
  34. # not the only traffic here: SD browsing, timelapse/recording listing, cover
  35. # downloads, deletes and storage checks all run through this pool too, and on a
  36. # farm they fan out across every printer at once. The pool's work queue is
  37. # unbounded, so exceeding it does not fail — it queues. But `asyncio.wait_for`
  38. # starts its clock at submission, not at thread start, so a task that sits in the
  39. # queue can burn its whole timeout without ever running, and `list_files_async`
  40. # reports a timeout as an empty listing — a silent "this printer has no files".
  41. # Keep the headroom.
  42. _FTP_MAX_WORKERS = 48
  43. _ftp_executor = ThreadPoolExecutor(max_workers=_FTP_MAX_WORKERS, thread_name_prefix="bambu-ftp")
  44. # Overall upload deadline (#2529). A flat wall-clock cap punishes big files on
  45. # slow links rather than catching broken ones: a 96 MB 3MF at the ~75 KB/s an A1
  46. # sustains over WiFi legitimately needs ~20 minutes, and the old flat 600 s
  47. # declared it dead at ~70 MB. The deadline is therefore derived from the file
  48. # size against a deliberately pessimistic floor rate. This is a backstop, not the
  49. # failure detector — a link that has actually died is caught within
  50. # ``socket_timeout`` by the blocking ``sendall``, long before this fires.
  51. _UPLOAD_FLOOR_BYTES_PER_SEC = 25 * 1024
  52. _UPLOAD_MIN_TIMEOUT = 600.0
  53. # How long to give the worker thread to notice the cancel flag, unwind, and
  54. # delete its partial file. It checks the flag once per CHUNK_SIZE, so on a link
  55. # slow enough to have hit the deadline this is one chunk plus the delete.
  56. _UPLOAD_CANCEL_GRACE = 60.0
  57. class UploadCancelled(Exception):
  58. """Raised inside the upload worker to abort an in-flight transfer.
  59. ``upload_file`` treats any exception from its progress callback as "stop
  60. now": it breaks out of the send loop, deletes the partial file from the
  61. printer, and re-raises. That is the only way to stop a transfer — an
  62. executor thread cannot be cancelled from the event loop, so a bare
  63. ``asyncio.wait_for`` leaves it streaming (see ``upload_file_async``).
  64. """
  65. class DeleteResult(Enum):
  66. """Outcome of an FTP delete attempt.
  67. Distinguishes "file isn't on the printer" (550, recovery impossible by
  68. retrying) from "delete failed for some other reason" (network, auth,
  69. transient FTP error — worth retrying). The post-print SD-card cleanup in
  70. main.py used to flatten both into ``False`` and log a "may linger" WARNING
  71. on every successful print where the printer self-cleaned its SD card
  72. before our cleanup ran (#1721 reporter's A1).
  73. """
  74. DELETED = "deleted"
  75. NOT_FOUND = "not_found"
  76. FAILED = "failed"
  77. class FileNotOnPrinterError(Exception):
  78. """Raised when a remote FTP path returns 550 (file not found).
  79. 550 means the file does not exist at that path — retrying the same path
  80. will never succeed. Callers use this sentinel with with_ftp_retry's
  81. non_retry_exceptions to immediately move on to the next candidate path
  82. instead of burning the full retry budget (up to 11 × 30s per path) on
  83. a lookup that cannot recover.
  84. """
  85. class ImplicitFTP_TLS(FTP_TLS):
  86. """FTP_TLS subclass for implicit FTPS (port 990) with model-specific SSL handling.
  87. X1C/P1S printers (vsFTPd) require SSL with session reuse on the data channel.
  88. A1/A1 Mini printers have issues with SSL on the data channel entirely and
  89. timeout waiting for transfer completion. Set skip_session_reuse=True for A1
  90. printers to skip SSL on the data channel (control channel remains encrypted).
  91. Optionally caps the SSL context's maximum TLS version to v1.2 (P2S firmware
  92. 01.02.00.00 needs this — see :mod:`ftp_profiles` and #1401).
  93. """
  94. def __init__(self, *args, skip_session_reuse: bool = False, cap_tls_v1_2: bool = False, **kwargs):
  95. super().__init__(*args, **kwargs)
  96. self._sock = None
  97. self.skip_session_reuse = skip_session_reuse
  98. self.ssl_context = ssl.create_default_context()
  99. self.ssl_context.check_hostname = False
  100. self.ssl_context.verify_mode = ssl.CERT_NONE
  101. # ``create_default_context()`` does NOT guarantee a protocol floor: it
  102. # leaves ``minimum_version`` at ``MINIMUM_SUPPORTED``, and what that
  103. # resolves to is a property of the OpenSSL build, not of this code.
  104. # Measured on identical OpenSSL 3.5.6: python:3.13-slim-trixie (our
  105. # Docker base) reports TLSv1_2, a bare-metal venv reports
  106. # MINIMUM_SUPPORTED. Docker users have therefore always been floored at
  107. # 1.2 — every Bambu model is reachable under that floor — while
  108. # bare-metal and appliance installs could silently negotiate TLS 1.0.
  109. # State the floor rather than inheriting it.
  110. self.ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
  111. if cap_tls_v1_2:
  112. # With the floor above this pins the connection to exactly TLS 1.2.
  113. self.ssl_context.maximum_version = ssl.TLSVersion.TLSv1_2
  114. def connect(self, host="", port=990, timeout=-999, source_address=None):
  115. """Connect to host, wrapping socket in TLS immediately (implicit FTPS)."""
  116. if host:
  117. self.host = host
  118. if port > 0:
  119. self.port = port
  120. if timeout != -999:
  121. self.timeout = timeout
  122. if source_address:
  123. self.source_address = source_address
  124. # Create and wrap socket immediately (implicit TLS)
  125. self.sock = socket.create_connection((self.host, self.port), self.timeout, source_address=self.source_address)
  126. self.sock = self.ssl_context.wrap_socket(self.sock, server_hostname=self.host)
  127. self.af = self.sock.family
  128. self.file = self.sock.makefile("r", encoding=self.encoding)
  129. self.welcome = self.getresp()
  130. return self.welcome
  131. def ntransfercmd(self, cmd, rest=None):
  132. """Override to wrap data connection in SSL for X1C/P1S only.
  133. X1C/P1S printers (vsFTPd) require SSL session reuse on the data channel.
  134. A1/A1 Mini printers have issues with SSL on the data channel entirely -
  135. they timeout waiting for the transfer completion response. For A1, we
  136. skip SSL wrapping on the data channel (control channel remains encrypted).
  137. """
  138. conn, size = FTP.ntransfercmd(self, cmd, rest)
  139. if self._prot_p and not self.skip_session_reuse:
  140. # X1C/P1S: Wrap data channel with SSL session reuse (required by vsFTPd)
  141. conn = self.ssl_context.wrap_socket(
  142. conn,
  143. server_hostname=self.host,
  144. session=self.sock.session,
  145. )
  146. # A1/A1 Mini (skip_session_reuse=True): Don't wrap data channel in SSL
  147. # The control channel remains encrypted via implicit FTPS
  148. return conn, size
  149. class BambuFTPClient:
  150. """FTP client for retrieving files from Bambu Lab printers."""
  151. FTP_PORT = 990
  152. # Default timeout in seconds (increased for A1 printers)
  153. DEFAULT_TIMEOUT = 30
  154. # Models that may need SSL mode fallback (try prot_p first, fall back to prot_c)
  155. # These models have varying FTP SSL behavior depending on firmware version
  156. A1_MODELS = ("A1", "A1 Mini")
  157. # Chunk size for manual upload transfer (64KB)
  158. # Smaller chunks provide smoother progress reporting — at typical printer FTP
  159. # speeds (~50-100KB/s) this gives a progress update roughly every second.
  160. CHUNK_SIZE = 64 * 1024
  161. # Cache for working FTP modes per printer IP
  162. # Maps IP -> "prot_p" or "prot_c"
  163. _mode_cache: dict[str, str] = {}
  164. def __init__(
  165. self,
  166. ip_address: str,
  167. access_code: str,
  168. timeout: float | None = None,
  169. printer_model: str | None = None,
  170. force_prot_c: bool = False,
  171. ):
  172. self.ip_address = ip_address
  173. self.access_code = access_code
  174. self.timeout = timeout if timeout is not None else self.DEFAULT_TIMEOUT
  175. self.printer_model = printer_model
  176. self.force_prot_c = force_prot_c
  177. self._ftp: ImplicitFTP_TLS | None = None
  178. def _is_a1_model(self) -> bool:
  179. """Check if this is an A1 series printer."""
  180. if not self.printer_model:
  181. return False
  182. return self.printer_model in self.A1_MODELS
  183. def _get_cached_mode(self) -> str | None:
  184. """Get cached FTP mode for this printer."""
  185. return self._mode_cache.get(self.ip_address)
  186. @classmethod
  187. def cache_mode(cls, ip_address: str, mode: str):
  188. """Cache the working FTP mode for a printer."""
  189. cls._mode_cache[ip_address] = mode
  190. logger.info("FTP mode cached for %s: %s", ip_address, mode)
  191. def _should_use_prot_c(self) -> bool:
  192. """Determine if we should use prot_c (clear) mode."""
  193. # If explicitly forced, use prot_c
  194. if self.force_prot_c:
  195. return True
  196. # Check cache first
  197. cached = self._get_cached_mode()
  198. if cached:
  199. return cached == "prot_c"
  200. # Default: try prot_p first (will fall back if needed)
  201. return False
  202. def connect(self) -> bool:
  203. """Connect to the printer FTP server (implicit FTPS on port 990)."""
  204. try:
  205. use_prot_c = self._should_use_prot_c()
  206. from backend.app.services.ftp_profiles import get_ftp_profile
  207. profile = get_ftp_profile(self.printer_model)
  208. logger.debug(
  209. f"FTP connecting to {self.ip_address}:{self.FTP_PORT} "
  210. f"(timeout={self.timeout}s, model={self.printer_model}, prot_c={use_prot_c}, "
  211. f"cap_tls_v1_2={profile.cap_tls_v1_2})"
  212. )
  213. self._ftp = ImplicitFTP_TLS(
  214. skip_session_reuse=use_prot_c,
  215. cap_tls_v1_2=profile.cap_tls_v1_2,
  216. )
  217. self._ftp.connect(self.ip_address, self.FTP_PORT, timeout=self.timeout)
  218. logger.debug("FTP connected, logging in as bblp")
  219. self._ftp.login("bblp", self.access_code)
  220. if use_prot_c:
  221. # Use clear (unencrypted) data channel
  222. logger.debug("FTP logged in, setting prot_c (clear) and passive mode")
  223. self._ftp.prot_c()
  224. else:
  225. # Use protected (encrypted) data channel with session reuse
  226. logger.debug("FTP logged in, setting prot_p (protected) and passive mode")
  227. self._ftp.prot_p()
  228. self._ftp.set_pasv(True)
  229. # Log welcome message for debugging
  230. if hasattr(self._ftp, "welcome") and self._ftp.welcome:
  231. logger.debug("FTP server welcome: %s", self._ftp.welcome)
  232. logger.info(
  233. f"FTP connected successfully to {self.ip_address} (model={self.printer_model}, prot_c={use_prot_c})"
  234. )
  235. return True
  236. except ftplib.error_perm as e:
  237. logger.warning("FTP connection permission error to %s: %s", self.ip_address, e)
  238. self._ftp = None
  239. return False
  240. except TimeoutError as e:
  241. logger.warning("FTP connection timed out to %s: %s", self.ip_address, e)
  242. self._ftp = None
  243. return False
  244. except ssl.SSLError as e:
  245. logger.warning("FTP SSL error connecting to %s: %s", self.ip_address, e)
  246. self._ftp = None
  247. return False
  248. except (OSError, ftplib.Error) as e:
  249. logger.warning("FTP connection failed to %s: %s (type: %s)", self.ip_address, e, type(e).__name__)
  250. self._ftp = None
  251. return False
  252. def disconnect(self):
  253. """Disconnect from the FTP server."""
  254. if self._ftp:
  255. try:
  256. self._ftp.quit()
  257. except (OSError, ftplib.Error, EOFError):
  258. pass # Best-effort FTP cleanup; connection may already be closed
  259. self._ftp = None
  260. def list_files(self, path: str = "/") -> list[dict]:
  261. """List files in a directory."""
  262. if not self._ftp:
  263. return []
  264. files = []
  265. try:
  266. self._ftp.cwd(path)
  267. items = []
  268. self._ftp.retrlines("LIST", items.append)
  269. for item in items:
  270. parts = item.split()
  271. if len(parts) >= 9:
  272. name = " ".join(parts[8:])
  273. is_dir = item.startswith("d")
  274. size = int(parts[4]) if not is_dir else 0
  275. # Parse modification time from FTP listing
  276. # Format: "Nov 30 10:15" or "Nov 30 2024"
  277. mtime = None
  278. try:
  279. from datetime import datetime
  280. month = parts[5]
  281. day = parts[6]
  282. time_or_year = parts[7]
  283. # Determine if it's time (HH:MM) or year
  284. if ":" in time_or_year:
  285. # Recent file: "Nov 30 10:15" - assume current year
  286. year = datetime.now().year
  287. time_str = f"{month} {day} {year} {time_or_year}"
  288. mtime = datetime.strptime(time_str, "%b %d %Y %H:%M")
  289. # If parsed date is in the future, use last year
  290. if mtime > datetime.now():
  291. mtime = mtime.replace(year=year - 1)
  292. else:
  293. # Older file: "Nov 30 2024" - no time, just date
  294. time_str = f"{month} {day} {time_or_year}"
  295. mtime = datetime.strptime(time_str, "%b %d %Y")
  296. except (ValueError, IndexError):
  297. pass # Non-critical: mtime parsing is best-effort; file entry works without it
  298. file_entry = {
  299. "name": name,
  300. "is_directory": is_dir,
  301. "size": size,
  302. "path": f"{path.rstrip('/')}/{name}",
  303. }
  304. if mtime:
  305. file_entry["mtime"] = mtime
  306. files.append(file_entry)
  307. logger.debug("Listed %s files in %s", len(files), path)
  308. except (OSError, ftplib.Error) as e:
  309. logger.info("FTP list_files failed for %s: %s", path, e)
  310. return files
  311. def download_file(self, remote_path: str, expected_size: int | None = None) -> bytes | None:
  312. """Download a file from the printer.
  313. ``expected_size`` is the byte count the directory listing reported for
  314. this file. Pass it whenever a short read must not be mistaken for a
  315. successful download: an FTPS data connection that closes early does
  316. not always raise, so ``retrbinary`` can hand back a partial buffer that
  317. looks like a perfectly good file to everything downstream. That is
  318. tolerable when the printer keeps its copy, and not tolerable when the
  319. caller goes on to delete the source (#2704).
  320. A zero-byte result is always treated as a failure, matching
  321. :meth:`download_to_file` — no caller has a use for an empty file.
  322. """
  323. if not self._ftp:
  324. return None
  325. try:
  326. buffer = BytesIO()
  327. self._ftp.retrbinary(f"RETR {remote_path}", buffer.write)
  328. data = buffer.getvalue()
  329. except (OSError, ftplib.Error):
  330. return None
  331. if not data:
  332. logger.warning("FTP download returned 0 bytes for %s", remote_path)
  333. return None
  334. if expected_size is not None and len(data) != expected_size:
  335. logger.warning(
  336. "FTP download of %s is short: got %s bytes, listing reported %s — treating as failed",
  337. remote_path,
  338. len(data),
  339. expected_size,
  340. )
  341. return None
  342. return data
  343. def download_to_file(self, remote_path: str, local_path: Path) -> bool:
  344. """Download a file from the printer to local filesystem."""
  345. if not self._ftp:
  346. logger.warning("download_to_file called but FTP not connected")
  347. return False
  348. try:
  349. local_path.parent.mkdir(parents=True, exist_ok=True)
  350. with open(local_path, "wb") as f:
  351. self._ftp.retrbinary(f"RETR {remote_path}", f.write)
  352. f.flush()
  353. os.fsync(f.fileno())
  354. file_size = local_path.stat().st_size if local_path.exists() else 0
  355. if file_size == 0:
  356. logger.warning("FTP download returned 0 bytes for %s", remote_path)
  357. if local_path.exists():
  358. local_path.unlink()
  359. return False
  360. logger.info("Successfully downloaded %s to %s (%s bytes)", remote_path, local_path, file_size)
  361. return True
  362. except (OSError, ftplib.Error) as e:
  363. # Clean up partial file if it exists
  364. if local_path.exists():
  365. try:
  366. local_path.unlink()
  367. except OSError:
  368. pass # Best-effort partial file cleanup; not critical if removal fails
  369. # 550 means the file is not at this path. Surface as a sentinel so
  370. # with_ftp_retry can abandon this path immediately and the caller
  371. # can advance to the next candidate instead of retrying 11× at
  372. # 30s intervals (the pattern that cost #972's reporter ~48min).
  373. if isinstance(e, ftplib.error_perm) and str(e).startswith("550"):
  374. logger.info("FTP download failed for %s: %s (not on printer)", remote_path, e)
  375. raise FileNotOnPrinterError(f"{remote_path}: {e}") from e
  376. # Log at INFO level so we can see failures in normal logs
  377. logger.info("FTP download failed for %s: %s", remote_path, e)
  378. return False
  379. def diagnose_storage(self) -> dict:
  380. """Run storage diagnostics and return results. For debugging upload issues."""
  381. results = {
  382. "connected": self._ftp is not None,
  383. "can_list_root": False,
  384. "root_files": [],
  385. "can_list_cache": False,
  386. "storage_info": None,
  387. "pwd": None,
  388. "errors": [],
  389. }
  390. if not self._ftp:
  391. results["errors"].append("FTP not connected")
  392. return results
  393. # Try to get current directory
  394. try:
  395. results["pwd"] = self._ftp.pwd()
  396. logger.debug("FTP current directory: %s", results["pwd"])
  397. except (OSError, ftplib.Error) as e:
  398. results["errors"].append(f"PWD failed: {e}")
  399. logger.debug("FTP PWD failed: %s", e)
  400. # Try to list root directory
  401. try:
  402. self._ftp.cwd("/")
  403. items = []
  404. self._ftp.retrlines("LIST", items.append)
  405. results["can_list_root"] = True
  406. results["root_files"] = items[:10] # First 10 entries
  407. logger.debug("FTP root listing (%s items): %s", len(items), items[:5])
  408. except (OSError, ftplib.Error) as e:
  409. results["errors"].append(f"LIST / failed: {e}")
  410. logger.debug("FTP LIST / failed: %s", e)
  411. # Try to list /cache (should exist on all printers)
  412. try:
  413. self._ftp.cwd("/cache")
  414. items = []
  415. self._ftp.retrlines("LIST", items.append)
  416. results["can_list_cache"] = True
  417. logger.debug("FTP /cache listing: %s items", len(items))
  418. except (OSError, ftplib.Error) as e:
  419. results["errors"].append(f"LIST /cache failed: {e}")
  420. logger.debug("FTP LIST /cache failed: %s", e)
  421. # Try to get storage info
  422. try:
  423. results["storage_info"] = self.get_storage_info()
  424. logger.debug("FTP storage info: %s", results["storage_info"])
  425. except (OSError, ftplib.Error) as e:
  426. results["errors"].append(f"Storage info failed: {e}")
  427. return results
  428. def upload_file(
  429. self,
  430. local_path: Path,
  431. remote_path: str,
  432. progress_callback: Callable[[int, int], None] | None = None,
  433. ) -> bool:
  434. """Upload a file to the printer with optional progress callback."""
  435. if not self._ftp:
  436. logger.warning("upload_file: FTP not connected")
  437. return False
  438. try:
  439. file_size = local_path.stat().st_size if local_path.exists() else 0
  440. logger.info("FTP uploading %s (%s bytes) to %s", local_path, file_size, remote_path)
  441. uploaded = 0
  442. callback_exception: Exception | None = None
  443. # Use manual transfer instead of storbinary() for A1 compatibility
  444. # A1 printers have issues with storbinary's voidresp() hanging after transfer
  445. with open(local_path, "rb") as f:
  446. logger.debug("FTP STOR command starting for %s", remote_path)
  447. t0 = time.monotonic()
  448. conn = self._ftp.transfercmd(f"STOR {remote_path}")
  449. logger.info(
  450. "FTP data channel ready in %.1fs (PASV + TLS handshake)",
  451. time.monotonic() - t0,
  452. )
  453. # Set explicit socket options for reliable transfer
  454. conn.setblocking(True)
  455. conn.settimeout(self.timeout)
  456. try:
  457. while True:
  458. chunk = f.read(self.CHUNK_SIZE)
  459. if not chunk:
  460. logger.debug("FTP upload: final chunk reached")
  461. break
  462. conn.sendall(chunk)
  463. uploaded += len(chunk)
  464. logger.debug("FTP upload progress: %s/%s bytes", uploaded, file_size)
  465. if progress_callback:
  466. try:
  467. progress_callback(uploaded, file_size)
  468. except Exception as e:
  469. callback_exception = e
  470. logger.info(
  471. "FTP upload callback requested stop for %s at %s/%s bytes: %s",
  472. remote_path,
  473. uploaded,
  474. file_size,
  475. e,
  476. )
  477. break
  478. except OSError as e:
  479. logger.error("FTP connection lost during upload: %s", e)
  480. raise
  481. finally:
  482. try:
  483. conn.close()
  484. except OSError:
  485. pass
  486. # Wait for the server's 226 "Transfer complete" response to confirm
  487. # the file has been flushed to the SD card. Without this, the printer
  488. # may try to read an incomplete file when the print command is sent,
  489. # causing 0500-C010 "MicroSD Card read/write exception" errors.
  490. # See: https://bugs.python.org/issue25458 (ftplib response desync)
  491. try:
  492. old_timeout = self._ftp.sock.gettimeout()
  493. # Use a generous timeout — H2D printers can take 30+ seconds
  494. # to send the 226 after the data channel closes.
  495. self._ftp.sock.settimeout(max(self.timeout, 60))
  496. try:
  497. resp = self._ftp.voidresp()
  498. logger.info("FTP STOR confirmed for %s: %s", remote_path, resp.strip())
  499. finally:
  500. self._ftp.sock.settimeout(old_timeout)
  501. except ftplib.Error as e:
  502. # Some P2S firmware revisions return ftplib.Error (e.g. 426
  503. # "Failure reading network stream") on voidresp() even when
  504. # the file landed fully on the SD card — the TLS data
  505. # channel close races the 226 confirmation (#1417 follow-up).
  506. # Verify via SIZE: if the server-side file size matches what
  507. # we just uploaded, the file is intact and we proceed with
  508. # a warning. If not — or SIZE itself fails — the transfer
  509. # was genuinely truncated and we must fail so the print
  510. # command doesn't go out for a partial 3MF (the original
  511. # reason this catch was tightened in the previous round).
  512. try:
  513. server_size = self._ftp.size(remote_path)
  514. except (OSError, ftplib.Error) as size_err:
  515. logger.debug("Post-error SIZE check failed: %s", size_err)
  516. server_size = None
  517. if server_size is not None and server_size == file_size:
  518. logger.warning(
  519. "FTP STOR returned %s for %s but file is intact on the "
  520. "printer (%s bytes match) — proceeding: %s",
  521. type(e).__name__,
  522. remote_path,
  523. file_size,
  524. e,
  525. )
  526. else:
  527. logger.error(
  528. "FTP STOR rejected by printer for %s: %s (%s); server size=%s expected=%s",
  529. remote_path,
  530. e,
  531. type(e).__name__,
  532. server_size,
  533. file_size,
  534. )
  535. raise
  536. except Exception as e:
  537. # Timeout or socket-level error reading 226 — the data was sent
  538. # on our side and the printer may still have written the file.
  539. # H2D can take 30+ seconds to send 226 after the data channel
  540. # closes, so we proceed with a warning rather than failing here.
  541. logger.warning(
  542. "FTP STOR confirmation not received for %s (proceeding): %s (%s)",
  543. remote_path,
  544. e,
  545. type(e).__name__,
  546. )
  547. if callback_exception is not None:
  548. cleanup_result: DeleteResult = DeleteResult.FAILED
  549. try:
  550. cleanup_result = self.delete_file(remote_path)
  551. except Exception as cleanup_error:
  552. logger.warning("FTP cancel cleanup failed for %s: %s", remote_path, cleanup_error)
  553. # NOT_FOUND is success here — the partial file is gone (printer
  554. # may have already swept on cancel), which is the goal.
  555. if cleanup_result in (DeleteResult.DELETED, DeleteResult.NOT_FOUND):
  556. logger.info("FTP cancel cleanup succeeded for %s (%s)", remote_path, cleanup_result.value)
  557. raise callback_exception
  558. raise RuntimeError(
  559. f"Upload cancelled but failed to remove partial file {remote_path} from printer"
  560. ) from callback_exception
  561. elapsed = time.monotonic() - t0
  562. speed_kbs = (file_size / 1024) / elapsed if elapsed > 0 else 0
  563. logger.info(
  564. "FTP upload complete: %s (%s bytes in %.1fs, %.0f KB/s)",
  565. remote_path,
  566. file_size,
  567. elapsed,
  568. speed_kbs,
  569. )
  570. return True
  571. except ftplib.error_perm as e:
  572. # Permanent FTP error (4xx/5xx response)
  573. error_code = str(e)[:3] if str(e) else "unknown"
  574. logger.error("FTP upload failed for %s: %s (error code: %s)", remote_path, e, error_code)
  575. if error_code == "553":
  576. logger.error(
  577. "FTP 553 error - Could not create file. Possible causes: "
  578. "1) No SD card inserted, 2) SD card full, 3) SD card not formatted correctly (needs FAT32/exFAT), "
  579. "4) Printer busy/not ready, 5) File path issue"
  580. )
  581. elif error_code == "550":
  582. logger.error("FTP 550 error - File/directory not found or permission denied")
  583. elif error_code == "552":
  584. logger.error("FTP 552 error - Storage quota exceeded (SD card full?)")
  585. return False
  586. except (OSError, ftplib.Error) as e:
  587. logger.error("FTP upload failed for %s: %s (type: %s)", remote_path, e, type(e).__name__)
  588. return False
  589. def upload_bytes(self, data: bytes, remote_path: str) -> bool:
  590. """Upload bytes to the printer."""
  591. if not self._ftp:
  592. return False
  593. try:
  594. # Use manual transfer instead of storbinary() for A1 compatibility
  595. conn = self._ftp.transfercmd(f"STOR {remote_path}")
  596. conn.setblocking(True)
  597. conn.settimeout(self.timeout)
  598. try:
  599. # Send data in chunks
  600. offset = 0
  601. while offset < len(data):
  602. chunk = data[offset : offset + self.CHUNK_SIZE]
  603. conn.sendall(chunk)
  604. offset += len(chunk)
  605. except OSError as e:
  606. logger.error("FTP connection lost during upload_bytes: %s", e)
  607. raise
  608. finally:
  609. try:
  610. conn.close()
  611. except OSError:
  612. pass
  613. # Wait for 226 confirmation (see upload_file for rationale).
  614. # ftplib.Error subclasses (e.g. 426 error_temp) mean the server
  615. # rejected the transfer and the file is partial — fail. Other
  616. # exceptions (timeout, socket-level) are tolerated as in upload_file.
  617. try:
  618. old_timeout = self._ftp.sock.gettimeout()
  619. self._ftp.sock.settimeout(max(self.timeout, 60))
  620. try:
  621. self._ftp.voidresp()
  622. finally:
  623. self._ftp.sock.settimeout(old_timeout)
  624. except ftplib.Error as e:
  625. # Same SIZE-verify path as upload_file (#1417 follow-up):
  626. # tolerate a transient 426 if the bytes are actually on the
  627. # printer, fail loudly if they aren't.
  628. try:
  629. server_size = self._ftp.size(remote_path)
  630. except (OSError, ftplib.Error) as size_err:
  631. logger.debug("Post-error SIZE check failed: %s", size_err)
  632. server_size = None
  633. if server_size is not None and server_size == len(data):
  634. logger.warning(
  635. "FTP STOR returned %s for %s but file is intact on the "
  636. "printer (%s bytes match) — proceeding: %s",
  637. type(e).__name__,
  638. remote_path,
  639. len(data),
  640. e,
  641. )
  642. else:
  643. logger.error(
  644. "FTP STOR rejected by printer for %s: %s (%s); server size=%s expected=%s",
  645. remote_path,
  646. e,
  647. type(e).__name__,
  648. server_size,
  649. len(data),
  650. )
  651. return False
  652. except Exception:
  653. pass # Timeout / socket-level — proceed, data was sent.
  654. return True
  655. except (OSError, ftplib.Error):
  656. return False
  657. def delete_file(self, remote_path: str) -> DeleteResult:
  658. """Delete a file from the printer.
  659. Returns :class:`DeleteResult` distinguishing the file-not-found case
  660. (550) from network / auth / transient FTP failure. Callers that just
  661. want "did it work" should check ``result == DeleteResult.DELETED``.
  662. """
  663. if not self._ftp:
  664. return DeleteResult.FAILED
  665. try:
  666. self._ftp.delete(remote_path)
  667. return DeleteResult.DELETED
  668. except ftplib.error_perm as e:
  669. if str(e).startswith("550"):
  670. logger.debug("FTP delete: %s not on printer (550)", remote_path)
  671. return DeleteResult.NOT_FOUND
  672. logger.warning("Failed to delete %s: %s", remote_path, e)
  673. return DeleteResult.FAILED
  674. except (OSError, ftplib.Error) as e:
  675. logger.warning("Failed to delete %s: %s", remote_path, e)
  676. return DeleteResult.FAILED
  677. def get_file_size(self, remote_path: str) -> int | None:
  678. """Get the size of a file."""
  679. if not self._ftp:
  680. return None
  681. try:
  682. return self._ftp.size(remote_path)
  683. except (OSError, ftplib.Error):
  684. return None
  685. def get_storage_info(self) -> dict | None:
  686. """Get storage information from the printer."""
  687. if not self._ftp:
  688. return None
  689. result = {}
  690. # Try AVBL command (available space) - some FTP servers support this
  691. try:
  692. response = self._ftp.sendcmd("AVBL")
  693. logger.debug("AVBL response: %s", response)
  694. # Response format: "213 <bytes available>"
  695. if response.startswith("213"):
  696. parts = response.split()
  697. if len(parts) >= 2:
  698. result["free_bytes"] = int(parts[1])
  699. except (OSError, ftplib.Error) as e:
  700. logger.debug("AVBL command not supported: %s", e)
  701. # Try STAT command as fallback
  702. try:
  703. response = self._ftp.sendcmd("STAT")
  704. logger.debug("STAT response: %s", response)
  705. except (OSError, ftplib.Error):
  706. pass # Both AVBL and STAT unsupported; storage info will rely on directory scan
  707. # Calculate used space by listing root directories
  708. try:
  709. total_used = 0
  710. dirs_to_scan = ["/cache", "/timelapse", "/model", "/data", "/data/Metadata", "/"]
  711. for dir_path in dirs_to_scan:
  712. try:
  713. self._ftp.cwd(dir_path)
  714. items = []
  715. self._ftp.retrlines("LIST", items.append)
  716. for item in items:
  717. parts = item.split()
  718. if len(parts) >= 5 and not item.startswith("d"):
  719. try:
  720. total_used += int(parts[4])
  721. except ValueError:
  722. pass # Skip entries with non-numeric size fields
  723. except (OSError, ftplib.Error):
  724. pass # Directory may not exist on this printer model; skip it
  725. result["used_bytes"] = total_used
  726. except (OSError, ftplib.Error):
  727. pass # Storage scan failed; return whatever info was collected above
  728. return result if result else None
  729. # Shared 3MF download cache (#972).
  730. #
  731. # Both the cover thumbnail endpoint (api/routes/printers.py) and the archive
  732. # metadata flow (main.py) fetch the same 3MF file over FTP during a print.
  733. # On slow / contended links (A1 Wi-Fi, large files) the duplicate transfers
  734. # compete for the printer's single FTP socket and trigger 425 "can't open
  735. # data channel" errors, feeding back into cause-2's retry storm.
  736. #
  737. # This cache stores the local path of a successfully-downloaded 3MF keyed
  738. # by (printer_id, normalized_name). Whichever flow downloads first populates
  739. # the cache; the other flow reuses the file read-only. Evicted on print
  740. # completion so a later print with the same name re-downloads fresh bytes.
  741. _threemf_path_cache: dict[tuple[int, str], Path] = {}
  742. def normalize_3mf_name(name: str) -> str:
  743. """Collapse various 3MF filename variants to a cache key.
  744. Bambu tooling produces names as bare subtask ("Part"), with .3mf, with
  745. .gcode.3mf, or (Studio-normalized) with spaces → underscores. All of
  746. these refer to the same print job on the same printer, so they must
  747. hash to the same cache key.
  748. """
  749. # Lowercase first so .3MF / .GCODE.3MF variants strip cleanly — a
  750. # real-world case since Windows-side tooling sometimes uppercases
  751. # extensions.
  752. cleaned = name.strip().lower().replace(".gcode.3mf", "").replace(".gcode", "").replace(".3mf", "")
  753. return cleaned.replace(" ", "_")
  754. def cache_3mf_download(printer_id: int, name: str, local_path: Path) -> None:
  755. """Record a successfully-downloaded 3MF so a sibling flow can reuse it."""
  756. _threemf_path_cache[(printer_id, normalize_3mf_name(name))] = local_path
  757. def get_cached_3mf(printer_id: int, name: str) -> Path | None:
  758. """Return a cached 3MF path for this printer/name if the file still exists."""
  759. key = (printer_id, normalize_3mf_name(name))
  760. cached = _threemf_path_cache.get(key)
  761. if cached and cached.exists() and cached.stat().st_size > 0:
  762. return cached
  763. # Evict dead entry — the file was cleaned up (temp dir clean, manual
  764. # deletion, restart) so the cache value is no longer usable.
  765. if cached:
  766. _threemf_path_cache.pop(key, None)
  767. return None
  768. def clear_3mf_cache(printer_id: int | None = None, delete_files: bool = True) -> None:
  769. """Drop cache entries for one printer (or all with None).
  770. When ``delete_files`` is True (default) the on-disk 3MF is removed as well
  771. — called from on_print_complete so temp files don't accumulate across
  772. prints. Tests that want to inspect the cache contents disable this.
  773. Only paths inside ``archive_dir/temp`` are unlinked. The dispatch sites
  774. added in #1166 also cache the live archive copy and library file bytes
  775. so /cover can skip FTP — those are *user data*, never the cache's to
  776. delete. Pre-fix this branch silently removed archive 3mfs on every print
  777. completion (#1212 + private reports of "file disappeared overnight").
  778. """
  779. from backend.app.core.config import settings as _config_settings
  780. temp_root = _config_settings.archive_dir / "temp"
  781. def _is_temp_path(path: Path) -> bool:
  782. try:
  783. return path.is_relative_to(temp_root)
  784. except (OSError, ValueError):
  785. return False
  786. def _maybe_unlink(path: Path) -> None:
  787. if not delete_files or not path.exists():
  788. return
  789. if not _is_temp_path(path):
  790. return
  791. try:
  792. path.unlink()
  793. except OSError as exc:
  794. logger.debug("3MF cache cleanup skipped %s: %s", path, exc)
  795. if printer_id is None:
  796. for path in list(_threemf_path_cache.values()):
  797. _maybe_unlink(path)
  798. _threemf_path_cache.clear()
  799. return
  800. for key in [k for k in _threemf_path_cache if k[0] == printer_id]:
  801. _maybe_unlink(_threemf_path_cache[key])
  802. _threemf_path_cache.pop(key, None)
  803. async def download_file_async(
  804. ip_address: str,
  805. access_code: str,
  806. remote_path: str,
  807. local_path: Path,
  808. timeout: float = 60.0,
  809. socket_timeout: float | None = None,
  810. printer_model: str | None = None,
  811. ) -> bool:
  812. """Async wrapper for downloading a file with timeout.
  813. For A1/A1 Mini printers, automatically tries prot_p first, then falls back
  814. to prot_c if the download fails. The working mode is cached for future operations.
  815. Args:
  816. ip_address: Printer IP address
  817. access_code: Printer access code
  818. remote_path: Remote file path on printer
  819. local_path: Local path to save file
  820. timeout: Overall operation timeout (asyncio)
  821. socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
  822. printer_model: Printer model for A1-specific workarounds
  823. """
  824. loop = asyncio.get_event_loop()
  825. is_a1 = printer_model in BambuFTPClient.A1_MODELS if printer_model else False
  826. # Per-attempt completion state: asyncio.wait_for cannot cancel
  827. # run_in_executor threads, so on timeout the executor may still complete
  828. # the download after we stop waiting. The thread flips `success` to True
  829. # ONLY after the file is fully written — a post-timeout check lets us
  830. # salvage the download without mistaking an in-progress partial write
  831. # for a completed one. Each attempt gets its own dict and event so a
  832. # zombie from an earlier attempt can't flip the flag for a later one.
  833. # The event is set in `_download`'s finally block so the post-timeout
  834. # path can wait for genuine thread completion instead of a fixed sleep.
  835. def _download(force_prot_c: bool, completion: dict, done: threading.Event) -> bool:
  836. mode_str = "prot_c" if force_prot_c else "prot_p"
  837. try:
  838. client = BambuFTPClient(
  839. ip_address,
  840. access_code,
  841. timeout=socket_timeout,
  842. printer_model=printer_model,
  843. force_prot_c=force_prot_c,
  844. )
  845. if client.connect():
  846. try:
  847. result = client.download_to_file(remote_path, local_path)
  848. if result:
  849. BambuFTPClient.cache_mode(ip_address, mode_str)
  850. completion["success"] = True
  851. return result
  852. finally:
  853. client.disconnect()
  854. return False
  855. finally:
  856. done.set()
  857. async def _run(force_prot_c: bool) -> bool:
  858. completion = {"success": False}
  859. done = threading.Event()
  860. try:
  861. return await asyncio.wait_for(
  862. loop.run_in_executor(_ftp_executor, _download, force_prot_c, completion, done), timeout=timeout
  863. )
  864. except TimeoutError:
  865. # Slow WiFi links commonly overshoot ftp_timeout by 10–30 s without
  866. # actually being stuck, so starting attempt 2 now would just contend
  867. # with the still-progressing RETR on attempt 1 and produce the
  868. # zombie-write race reported in #1014 (file landed on disk minutes
  869. # after the retry loop had already given up). Wait for the worker
  870. # thread to genuinely finish — capped at 30 s so a truly stuck
  871. # connection can't stall a whole attempt indefinitely, with a 0.5 s
  872. # floor so artificially small test timeouts still give zombies a
  873. # realistic window to finish.
  874. grace = max(min(timeout, 30.0), 0.5)
  875. # Deliberately the DEFAULT executor, not `_ftp_executor`: this thread
  876. # blocks waiting on `_download`, which is itself an `_ftp_executor`
  877. # worker. Parking waiters in the same bounded pool as the workers they
  878. # wait for is how you build a deadlock — with enough concurrent
  879. # timeouts the waiters would occupy every slot and the downloads they
  880. # are waiting for could never be scheduled.
  881. await loop.run_in_executor(None, done.wait, grace)
  882. if completion["success"] and local_path.exists() and local_path.stat().st_size > 0:
  883. logger.info(
  884. "FTP download wait_for timed out after %ss for %s, but thread completed within %ss grace (%s bytes) — salvaging",
  885. timeout,
  886. remote_path,
  887. grace,
  888. local_path.stat().st_size,
  889. )
  890. return True
  891. logger.warning(
  892. "FTP download timed out after %ss (plus %ss grace) for %s",
  893. timeout,
  894. grace,
  895. remote_path,
  896. )
  897. return False
  898. # Check if we have a cached mode for this printer
  899. cached_mode = BambuFTPClient._mode_cache.get(ip_address)
  900. if cached_mode:
  901. force_prot_c = cached_mode == "prot_c"
  902. return await _run(force_prot_c)
  903. # No cached mode - try prot_p first
  904. if await _run(False):
  905. return True
  906. # Download failed - for A1 models, try prot_c fallback
  907. if is_a1:
  908. logger.info("FTP download failed with prot_p for A1 model, trying prot_c fallback...")
  909. return await _run(True)
  910. return False
  911. async def download_file_try_paths_async(
  912. ip_address: str,
  913. access_code: str,
  914. remote_paths: list[str],
  915. local_path: Path,
  916. socket_timeout: float | None = None,
  917. printer_model: str | None = None,
  918. timeout: float = 90.0,
  919. ) -> bool:
  920. """Try downloading a file from multiple paths using a single connection.
  921. Args:
  922. socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
  923. printer_model: Printer model for A1-specific workarounds
  924. timeout: overall async cap. The per-socket timeout only bounds an
  925. in-flight worker; it does NOT bound how long this coroutine waits
  926. for a free slot in the fixed-size ``_ftp_executor``. On a large
  927. farm where offline printers keep every worker busy on dead
  928. connects, that queue wait is otherwise unbounded — and any caller
  929. holding a DB connection while awaiting this would pin it until the
  930. pool is exhausted (#2572). The cap converts that into a bounded
  931. wait; the orphaned worker finishes and its result is discarded.
  932. """
  933. loop = asyncio.get_event_loop()
  934. def _download():
  935. client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
  936. if not client.connect():
  937. return False
  938. try:
  939. # FileNotOnPrinterError signals "try the next path", not "give up" —
  940. # this function's whole purpose is to walk a list of candidates
  941. # over one connection. Only a real transport error should bubble.
  942. for remote_path in remote_paths:
  943. try:
  944. if client.download_to_file(remote_path, local_path):
  945. return True
  946. except FileNotOnPrinterError:
  947. continue
  948. return False
  949. finally:
  950. client.disconnect()
  951. try:
  952. return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _download), timeout=timeout)
  953. except TimeoutError:
  954. logger.warning("FTP download_try_paths exceeded its %ss cap for %s (#2572)", timeout, ip_address)
  955. return False
  956. def _upload_deadline(local_path: Path) -> float:
  957. """Derive an upload deadline from the file size (#2529).
  958. See ``_UPLOAD_FLOOR_BYTES_PER_SEC``. An unstat-able file falls back to the
  959. floor timeout — ``upload_file`` will fail on the open() anyway.
  960. """
  961. try:
  962. size = local_path.stat().st_size
  963. except OSError:
  964. return _UPLOAD_MIN_TIMEOUT
  965. return max(_UPLOAD_MIN_TIMEOUT, size / _UPLOAD_FLOOR_BYTES_PER_SEC)
  966. # One upload at a time per printer. Two concurrent STOR commands for the same
  967. # remote path leave a corrupt file on the SD card, and the printer reads as
  968. # flaky rather than busy (#2529). Held for the duration of a transfer, so a
  969. # second dispatch to the same printer queues behind the first instead of racing
  970. # it. Keyed per event loop: an asyncio.Lock binds to the loop that first awaits
  971. # it, and the test suite runs each case on a fresh loop.
  972. _upload_locks: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[str, asyncio.Lock]] = (
  973. weakref.WeakKeyDictionary()
  974. )
  975. def _upload_lock(loop: asyncio.AbstractEventLoop, ip_address: str) -> asyncio.Lock:
  976. per_loop = _upload_locks.setdefault(loop, {})
  977. lock = per_loop.get(ip_address)
  978. if lock is None:
  979. lock = asyncio.Lock()
  980. per_loop[ip_address] = lock
  981. return lock
  982. async def upload_file_async(
  983. ip_address: str,
  984. access_code: str,
  985. local_path: Path,
  986. remote_path: str,
  987. timeout: float | None = None,
  988. progress_callback: Callable[[int, int], None] | None = None,
  989. socket_timeout: float | None = None,
  990. printer_model: str | None = None,
  991. ) -> bool:
  992. """Async wrapper for uploading a file with timeout and progress callback.
  993. For A1/A1 Mini printers, automatically tries prot_p first, then falls back
  994. to prot_c if the upload fails. The working mode is cached for future uploads.
  995. Args:
  996. ip_address: Printer IP address
  997. access_code: Printer access code
  998. local_path: Local file path to upload
  999. remote_path: Remote path on printer
  1000. timeout: Overall deadline. ``None`` (the default) derives it from the
  1001. file size — see ``_upload_deadline``. A caller that passes a number
  1002. gets exactly that, which is what the tests rely on.
  1003. progress_callback: Optional callback for progress updates
  1004. socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
  1005. printer_model: Printer model for A1-specific workarounds
  1006. """
  1007. loop = asyncio.get_event_loop()
  1008. is_a1 = printer_model in BambuFTPClient.A1_MODELS if printer_model else False
  1009. deadline = _upload_deadline(local_path) if timeout is None else timeout
  1010. # Set when the deadline expires. The worker checks it once per chunk.
  1011. cancel = threading.Event()
  1012. def _guarded_progress(uploaded: int, total: int) -> None:
  1013. if cancel.is_set():
  1014. raise UploadCancelled(f"upload of {remote_path} exceeded its {deadline:.0f}s deadline")
  1015. if progress_callback:
  1016. progress_callback(uploaded, total)
  1017. def _upload(force_prot_c: bool = False) -> bool:
  1018. mode_str = "prot_c" if force_prot_c else "prot_p"
  1019. logger.info(
  1020. f"FTP connecting to {ip_address} for upload (model={printer_model}, "
  1021. f"mode={mode_str}, socket_timeout={socket_timeout}s, deadline={deadline:.0f}s)..."
  1022. )
  1023. client = BambuFTPClient(
  1024. ip_address, access_code, timeout=socket_timeout, printer_model=printer_model, force_prot_c=force_prot_c
  1025. )
  1026. if client.connect():
  1027. logger.info("FTP connected to %s", ip_address)
  1028. try:
  1029. result = client.upload_file(local_path, remote_path, _guarded_progress)
  1030. if result:
  1031. # Cache the working mode
  1032. BambuFTPClient.cache_mode(ip_address, mode_str)
  1033. return result
  1034. finally:
  1035. client.disconnect()
  1036. logger.warning("FTP connection failed to %s", ip_address)
  1037. return False
  1038. async def _attempt(force_prot_c: bool) -> bool:
  1039. """Run one upload attempt, and make a timeout actually stop the transfer.
  1040. ``asyncio.wait_for`` cancels the *future*, never the executor thread
  1041. behind it. Before #2529 a slow-but-healthy upload that overran the
  1042. deadline left that thread streaming: it kept pushing bytes, kept firing
  1043. the progress callback, and the retry above put a *second* STOR of the
  1044. same file onto the same printer. The reporter's 96 MB job ran four
  1045. concurrent transfers and never landed. So on timeout we signal the
  1046. worker (it raises ``UploadCancelled`` from the progress callback, which
  1047. breaks the send loop and deletes the partial file) and wait for it to
  1048. actually go.
  1049. """
  1050. fut = loop.run_in_executor(_ftp_executor, lambda: _upload(force_prot_c))
  1051. try:
  1052. return await asyncio.wait_for(asyncio.shield(fut), timeout=deadline)
  1053. except TimeoutError:
  1054. cancel.set()
  1055. logger.warning(
  1056. "FTP upload of %s exceeded its %.0fs deadline — cancelling the transfer",
  1057. remote_path,
  1058. deadline,
  1059. )
  1060. try:
  1061. await asyncio.wait_for(asyncio.shield(fut), timeout=_UPLOAD_CANCEL_GRACE)
  1062. except UploadCancelled:
  1063. logger.info("FTP upload of %s cancelled; partial file removed from the printer", remote_path)
  1064. except TimeoutError:
  1065. # The thread is wedged somewhere that never reaches the callback
  1066. # (a blocked sendall, say). Nothing more we can do from here —
  1067. # but consume the eventual result so asyncio doesn't log the
  1068. # future's exception as unretrieved when it is garbage-collected.
  1069. logger.error(
  1070. "FTP upload thread for %s did not stop within %.0fs of the cancel signal",
  1071. remote_path,
  1072. _UPLOAD_CANCEL_GRACE,
  1073. )
  1074. fut.add_done_callback(_swallow_future_result)
  1075. except Exception as e:
  1076. logger.warning("FTP upload of %s errored while cancelling: %s", remote_path, e)
  1077. # Raise rather than return False: a deadline expiry means the link
  1078. # sustained less than the floor rate for the whole transfer, and a
  1079. # retry would only spend another full deadline finding that out
  1080. # again — with check_queue serialized, four of those block the
  1081. # entire print queue for hours. ``with_ftp_retry`` never retries it.
  1082. raise UploadCancelled(
  1083. f"Upload of {remote_path} to {ip_address} exceeded its {deadline:.0f}s deadline "
  1084. f"(link sustained less than {_UPLOAD_FLOOR_BYTES_PER_SEC // 1024} KB/s)"
  1085. ) from None
  1086. async with _upload_lock(loop, ip_address):
  1087. # Check if we have a cached mode for this printer
  1088. cached_mode = BambuFTPClient._mode_cache.get(ip_address)
  1089. if cached_mode:
  1090. # Use cached mode
  1091. return await _attempt(cached_mode == "prot_c")
  1092. # No cached mode - try prot_p first
  1093. if await _attempt(False):
  1094. return True
  1095. # Upload failed - for A1 models, try prot_c fallback
  1096. if is_a1:
  1097. logger.info("FTP upload failed with prot_p for A1 model, trying prot_c fallback...")
  1098. return await _attempt(True)
  1099. return False
  1100. def _swallow_future_result(fut: asyncio.Future) -> None:
  1101. """Retrieve a future's exception so asyncio doesn't log it as unhandled."""
  1102. if not fut.cancelled():
  1103. fut.exception()
  1104. async def list_files_async(
  1105. ip_address: str,
  1106. access_code: str,
  1107. path: str = "/",
  1108. timeout: float = 30.0,
  1109. socket_timeout: float | None = None,
  1110. printer_model: str | None = None,
  1111. ) -> list[dict]:
  1112. """Async wrapper for listing files with timeout.
  1113. Args:
  1114. socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
  1115. printer_model: Printer model for A1-specific workarounds
  1116. """
  1117. loop = asyncio.get_event_loop()
  1118. def _list():
  1119. client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
  1120. if client.connect():
  1121. try:
  1122. return client.list_files(path)
  1123. finally:
  1124. client.disconnect()
  1125. return []
  1126. try:
  1127. return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _list), timeout=timeout)
  1128. except TimeoutError:
  1129. logger.warning("FTP list_files timed out after %ss for %s", timeout, path)
  1130. return []
  1131. async def delete_file_async(
  1132. ip_address: str,
  1133. access_code: str,
  1134. remote_path: str,
  1135. socket_timeout: float | None = None,
  1136. printer_model: str | None = None,
  1137. timeout: float = 60.0,
  1138. ) -> DeleteResult:
  1139. """Async wrapper for deleting a file.
  1140. Returns :class:`DeleteResult` so callers can distinguish ``NOT_FOUND``
  1141. (550 — file isn't on the printer, no retry value) from ``FAILED``
  1142. (network / auth / transient — worth retrying or surfacing).
  1143. Args:
  1144. socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
  1145. printer_model: Printer model for A1-specific workarounds
  1146. timeout: overall async cap so a saturated ``_ftp_executor`` can't pin
  1147. the caller (and any DB connection it holds) indefinitely (#2572).
  1148. """
  1149. loop = asyncio.get_event_loop()
  1150. def _delete() -> DeleteResult:
  1151. client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
  1152. if client.connect():
  1153. try:
  1154. return client.delete_file(remote_path)
  1155. finally:
  1156. client.disconnect()
  1157. return DeleteResult.FAILED
  1158. try:
  1159. return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _delete), timeout=timeout)
  1160. except TimeoutError:
  1161. logger.warning("FTP delete_file exceeded its %ss cap for %s (#2572)", timeout, ip_address)
  1162. return DeleteResult.FAILED
  1163. async def download_file_bytes_async(
  1164. ip_address: str,
  1165. access_code: str,
  1166. remote_path: str,
  1167. socket_timeout: float | None = None,
  1168. printer_model: str | None = None,
  1169. timeout: float = 300.0,
  1170. expected_size: int | None = None,
  1171. ) -> bytes | None:
  1172. """Async wrapper for downloading file as bytes.
  1173. Args:
  1174. socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
  1175. printer_model: Printer model for A1-specific workarounds
  1176. timeout: overall async cap so a saturated ``_ftp_executor`` can't pin
  1177. the caller (and any DB connection it holds) indefinitely (#2572).
  1178. Generous by default because this pulls whole files (timelapse
  1179. video, gcode) which can legitimately take minutes over slow Wi-Fi —
  1180. the cap only guards against a permanently-starved pool, not a
  1181. slow-but-progressing transfer.
  1182. expected_size: size from the directory listing; a mismatch fails the
  1183. download instead of returning a truncated file. See
  1184. :meth:`BambuFTPClient.download_file`.
  1185. """
  1186. loop = asyncio.get_event_loop()
  1187. def _download():
  1188. client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
  1189. if client.connect():
  1190. try:
  1191. return client.download_file(remote_path, expected_size=expected_size)
  1192. finally:
  1193. client.disconnect()
  1194. return None
  1195. try:
  1196. return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _download), timeout=timeout)
  1197. except TimeoutError:
  1198. logger.warning("FTP download_bytes exceeded its %ss cap for %s (#2572)", timeout, ip_address)
  1199. return None
  1200. async def remote_file_settled(
  1201. ip_address: str,
  1202. access_code: str,
  1203. remote_path: str,
  1204. downloaded_bytes: int,
  1205. *,
  1206. printer_model: str | None = None,
  1207. ) -> bool:
  1208. """Confirm the printer has finished writing the file we just downloaded.
  1209. Matching the download against the size from the directory listing proves we
  1210. received what the listing *said*, not that the file was *finished*. The
  1211. timelapse scan's first look happens seconds after the print ends, which is
  1212. exactly when the printer is writing the video — so a file still growing can
  1213. be listed at a partial size, served at that size, and pass the length check
  1214. as a complete video (#2704).
  1215. That was survivable while the printer kept its copy. It isn't now that a
  1216. successful attach deletes the source, so re-list afterwards: if the file has
  1217. grown, what we hold is a prefix and the caller should discard it and try
  1218. again on the next round.
  1219. Returns True when the remote file can no longer differ from what we hold —
  1220. the size still matches, or the file is gone from the listing entirely and
  1221. so cannot grow any further. Returns False when it has changed size, and on
  1222. a listing failure, because "we could not check" must not read as "safe to
  1223. delete".
  1224. """
  1225. directory, _, name = remote_path.rpartition("/")
  1226. files = await list_files_async(ip_address, access_code, directory or "/", printer_model=printer_model)
  1227. if not files:
  1228. logger.warning("[TIMELAPSE] Could not re-list %s to confirm %s is complete", directory or "/", name)
  1229. return False
  1230. for f in files:
  1231. if f.get("name") == name:
  1232. size = f.get("size")
  1233. if size == downloaded_bytes:
  1234. return True
  1235. logger.info(
  1236. "[TIMELAPSE] %s is still being written (%s bytes now, %s when downloaded) — will retry",
  1237. name,
  1238. size,
  1239. downloaded_bytes,
  1240. )
  1241. return False
  1242. # Vanished between the download and now. Nothing left that could grow, and
  1243. # nothing left to delete either.
  1244. logger.debug("[TIMELAPSE] %s is no longer on the printer after download", name)
  1245. return True
  1246. async def delete_archived_timelapse(
  1247. ip_address: str,
  1248. access_code: str,
  1249. remote_path: str,
  1250. *,
  1251. verified: bool,
  1252. printer_model: str | None = None,
  1253. printer_name: str = "",
  1254. ) -> bool:
  1255. """Remove a timelapse from the printer once it is safely in the archive.
  1256. Call this only after the attach succeeded (#2704). Keeping ``/timelapse``
  1257. down to just the unclaimed videos is what makes the snapshot diff
  1258. unambiguous rather than merely usually-right, and it stops P1S cards
  1259. filling with AVIs.
  1260. ``verified`` must say whether the downloaded byte count was checked against
  1261. the size the directory listing reported. It is required rather than
  1262. defaulted because this is the one irreversible step in the flow: an FTPS
  1263. data connection that closes early does not always raise, so an unverified
  1264. transfer can be a partial file that looks complete, and deleting the source
  1265. would then destroy the only good copy. The check lives here rather than at
  1266. each call site so no future caller can omit it.
  1267. Best-effort otherwise: a printer that refuses the delete keeps its copy, the
  1268. diff still excludes that filename next time because it is attached to an
  1269. archive, and nothing else in the flow cares. Returns True only on an actual
  1270. delete or a 550 (already gone).
  1271. """
  1272. if not verified:
  1273. logger.warning(
  1274. "[TIMELAPSE] Not deleting %s from printer %s: the download was never size-checked",
  1275. remote_path,
  1276. printer_name,
  1277. )
  1278. return False
  1279. for attempt in range(1, 4):
  1280. try:
  1281. result = await delete_file_async(ip_address, access_code, remote_path, printer_model=printer_model)
  1282. except Exception as e:
  1283. result = DeleteResult.FAILED
  1284. logger.warning("[TIMELAPSE] Delete attempt %d/3 raised for %s: %s", attempt, remote_path, e)
  1285. if result == DeleteResult.DELETED:
  1286. logger.info("[TIMELAPSE] Deleted %s from printer %s after archiving", remote_path, printer_name)
  1287. return True
  1288. if result == DeleteResult.NOT_FOUND:
  1289. # 550 never recovers by waiting — the printer already cleaned up.
  1290. logger.debug("[TIMELAPSE] %s already gone from printer %s", remote_path, printer_name)
  1291. return True
  1292. if attempt < 3:
  1293. await asyncio.sleep(2)
  1294. logger.warning(
  1295. "[TIMELAPSE] Could not delete %s from printer %s (it stays on the card; the archive copy is unaffected)",
  1296. remote_path,
  1297. printer_name,
  1298. )
  1299. return False
  1300. async def get_storage_info_async(
  1301. ip_address: str,
  1302. access_code: str,
  1303. socket_timeout: float | None = None,
  1304. printer_model: str | None = None,
  1305. timeout: float = 60.0,
  1306. ) -> dict | None:
  1307. """Async wrapper for getting storage info.
  1308. Args:
  1309. socket_timeout: FTP socket timeout for slow connections (e.g., A1 printers)
  1310. printer_model: Printer model for A1-specific workarounds
  1311. timeout: overall async cap so a saturated ``_ftp_executor`` can't pin
  1312. the caller (and any DB connection it holds) indefinitely (#2572).
  1313. """
  1314. loop = asyncio.get_event_loop()
  1315. def _get_storage():
  1316. client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
  1317. if client.connect():
  1318. try:
  1319. return client.get_storage_info()
  1320. finally:
  1321. client.disconnect()
  1322. return None
  1323. try:
  1324. return await asyncio.wait_for(loop.run_in_executor(_ftp_executor, _get_storage), timeout=timeout)
  1325. except TimeoutError:
  1326. logger.warning("FTP get_storage_info exceeded its %ss cap for %s (#2572)", timeout, ip_address)
  1327. return None
  1328. async def get_ftp_retry_settings() -> tuple[bool, int, float, float]:
  1329. """Get FTP retry settings from database.
  1330. Returns:
  1331. Tuple of (retry_enabled, retry_count, retry_delay, timeout)
  1332. """
  1333. from backend.app.api.routes.settings import get_setting
  1334. from backend.app.core.database import async_session
  1335. async with async_session() as db:
  1336. enabled = (await get_setting(db, "ftp_retry_enabled") or "true") == "true"
  1337. count = int(await get_setting(db, "ftp_retry_count") or "3")
  1338. delay = float(await get_setting(db, "ftp_retry_delay") or "2")
  1339. timeout = float(await get_setting(db, "ftp_timeout") or "30")
  1340. return enabled, count, delay, timeout
  1341. async def with_ftp_retry(
  1342. operation: Callable[..., Awaitable[T]],
  1343. *args,
  1344. max_retries: int = 3,
  1345. retry_delay: float = 2.0,
  1346. operation_name: str = "FTP operation",
  1347. non_retry_exceptions: tuple[type[BaseException], ...] = (),
  1348. **kwargs,
  1349. ) -> T | None:
  1350. """Execute FTP operation with retry logic.
  1351. Args:
  1352. operation: Async function to execute
  1353. *args: Positional arguments for the operation
  1354. max_retries: Number of retry attempts (default: 3)
  1355. retry_delay: Seconds to wait between retries (default: 2.0)
  1356. operation_name: Name for logging purposes
  1357. non_retry_exceptions: Exception types that should immediately abort retries
  1358. **kwargs: Keyword arguments for the operation
  1359. Returns:
  1360. Result of the operation, or None if all attempts fail
  1361. ``UploadCancelled`` is never retried, whatever the caller passes: it means
  1362. the transfer overran its size-derived deadline, so a retry would spend
  1363. another full deadline reaching the same conclusion (#2529).
  1364. """
  1365. last_error = None
  1366. for attempt in range(max_retries + 1):
  1367. try:
  1368. result = await operation(*args, **kwargs)
  1369. # Check for "falsy" success indicators
  1370. if result not in (False, None, []):
  1371. if attempt > 0:
  1372. logger.info("%s succeeded on attempt %s/%s", operation_name, attempt + 1, max_retries + 1)
  1373. return result
  1374. # Operation returned failure indicator
  1375. if attempt > 0:
  1376. logger.info("%s attempt %s/%s returned failure", operation_name, attempt + 1, max_retries + 1)
  1377. except UploadCancelled:
  1378. raise
  1379. except Exception as e:
  1380. if non_retry_exceptions and isinstance(e, non_retry_exceptions):
  1381. raise
  1382. last_error = e
  1383. logger.warning("%s attempt %s/%s failed: %s", operation_name, attempt + 1, max_retries + 1, e)
  1384. # Don't wait after the last attempt
  1385. if attempt < max_retries:
  1386. logger.info("%s will retry in %ss...", operation_name, retry_delay)
  1387. await asyncio.sleep(retry_delay)
  1388. logger.error("%s failed after %s attempts", operation_name, max_retries + 1)
  1389. if last_error:
  1390. logger.debug("Last error: %s", last_error)
  1391. return None