Merge branch '0.1.9b' into feature/home-assistant-env-vars

This commit is contained in:
MartinNYHC
2026-02-07 19:55:11 +01:00
committed by GitHub
31 changed files with 3072 additions and 850 deletions
+28
View File
@@ -2,6 +2,34 @@
All notable changes to Bambuddy will be documented in this file.
## [0.1.9b] - Not released
### Documentation
- **Proxy Mode Security Warning** — Added FTP data channel security warning to wiki, README, and website. Bambu Studio does not encrypt the FTP data channel despite negotiating PROT P; MQTT and FTP control channels are fully TLS-encrypted. VPN (Tailscale/WireGuard) recommended for full data encryption.
- **Docker Proxy Mode Ports** — Documented FTP passive data ports 50000-50100 required for proxy mode in Docker bridge mode. Updated port mappings in wiki virtual-printer and docker guides.
- **SSDP Discovery Limitations** — Added table showing when SSDP discovery works (same LAN, dual-homed, Docker host mode) vs when manual IP entry is required (VPN, Docker bridge, port forwarding). Updated wiki, README, and website.
- **Firewall Rules Updated** — Added port 50000-50100/tcp to all UFW, firewalld, and iptables examples for proxy mode FTP passive data.
### Added
- **Hostname Support for Printers** ([#290](https://github.com/maziggy/bambuddy/issues/290)) — Printers can now be added using hostnames (e.g., `printer.local`, `my-printer.home.lan`) in addition to IPv4 addresses. Updated backend validation, frontend forms, and all locale labels.
- **Camera View Controls** ([#291](https://github.com/maziggy/bambuddy/issues/291)) — Added chamber light toggle and skip objects buttons to both embedded camera viewer and standalone camera page. Extracted skip objects modal into a reusable `SkipObjectsModal` component shared across PrintersPage and both camera views.
### Fixed
- **Camera Stop 401 When Auth Enabled** — Camera stop requests (`sendBeacon`) failed with 401 Unauthorized when authentication was enabled because `sendBeacon` cannot send auth headers. Replaced with `fetch` + `keepalive: true` which supports Authorization headers while remaining reliable during page unload.
- **Filament Usage Charts Inflated by Quantity Multiplier** ([#229](https://github.com/maziggy/bambuddy/issues/229)) — Daily, weekly, and filament-type charts were multiplying `filament_used_grams` by print quantity, even though the value already represents the total for the entire job. A 26-object print using 126g was counted as 3,276g. Removed the erroneous multiplier from three aggregations in `FilamentTrends.tsx`.
- **Energy Cost Shows 0.00 in "Total Consumption" Mode** ([#284](https://github.com/maziggy/bambuddy/issues/284)) — Statistics Quick Stats showed 0.00 energy cost when Energy Display Mode was set to "Total Consumption" with Home Assistant smart plugs. The `homeassistant_service` was not configured with HA URL/token before querying plug energy data, causing it to silently return nothing.
### Testing
- **Mock FTPS Server & Comprehensive FTP Test Suite** — Added 67 automated test cases against a real implicit FTPS mock server, covering every known FTP failure mode from 0.1.8+:
- Mock server (`mock_ftp_server.py`) implements implicit TLS, custom AVBL command, and per-command failure injection
- Connection tests: auth, SSL modes (prot_p/prot_c), timeout, cache, disconnect edge cases
- Upload tests: chunked transfer via `transfercmd()`, progress callbacks, 553/550/552 error handling
- Download tests: bytes, to-file, 0-byte regression, large files, missing file cleanup
- Model-specific tests: X1C session reuse, A1/A1 Mini prot_c fallback, P1S, unknown model defaults
- Async wrapper tests: upload/download/list/delete with A1 fallback and multi-path download
- Failure injection tests: regressions for `error_perm` hierarchy, `diagnose_storage` CWD propagation, injection count decrement
- Added `pyOpenSSL` to `requirements-dev.txt` for Docker test image compatibility
## [0.1.8.1] - 2026-02-07
### Fixed
+3
View File
@@ -47,6 +47,9 @@ ENV LOG_DIR=/app/logs
ENV PORT=8000
EXPOSE 8000
EXPOSE 8883
EXPOSE 9990
EXPOSE 50000-50100
# Health check (uses PORT env var via shell)
HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \
+3 -2
View File
@@ -39,7 +39,8 @@
**Print from anywhere in the world** — Bambuddy's new Proxy Mode acts as a secure relay between your slicer and printer:
- 🔒 **End-to-end TLS encryption** — Your print data is encrypted from slicer to printer
- 🔒 **TLS-encrypted control channels** — MQTT and FTP control fully encrypted
- 🛡️ **VPN recommended** — Use Tailscale/WireGuard for full data encryption ([details](https://wiki.bambuddy.cool/features/virtual-printer/))
- 🌍 **No cloud dependency** — Direct connection through your own Bambuddy server
- 🔑 **Uses printer's access code** — No additional credentials needed
- ⚡ **Full-speed printing** — FTP and MQTT protocols proxied transparently
@@ -163,7 +164,7 @@ Perfect for remote print farms, traveling makers, or accessing your home printer
- Send prints directly from Bambu Studio/Orca Slicer
- Configurable printer model (X1C, P1S, A1, H2D, etc.)
- Archive mode, Review mode, Queue mode, or Proxy mode
- SSDP discovery (appears in slicer automatically)
- SSDP discovery (same LAN) or manual IP entry (VPN/remote)
- Secure TLS/MQTT/FTP communication
### 🛠️ Maintenance & Support
+5
View File
@@ -545,6 +545,11 @@ async def get_archive_stats(
plugs_result = await db.execute(select(SmartPlug))
plugs = list(plugs_result.scalars().all())
# Configure HA service once (needed for homeassistant-type plugs)
ha_url = await get_setting(db, "ha_url") or ""
ha_token = await get_setting(db, "ha_token") or ""
homeassistant_service.configure(ha_url, ha_token)
total_energy_kwh = 0.0
for plug in plugs:
if plug.plug_type == "tasmota":
+1 -1
View File
@@ -5,7 +5,7 @@ from pathlib import Path
from pydantic_settings import BaseSettings
# Application version - single source of truth
APP_VERSION = "0.1.8.1"
APP_VERSION = "0.1.9b"
GITHUB_REPO = "maziggy/bambuddy"
# App directory - where the application is installed (for static files)
+1 -1
View File
@@ -12,7 +12,7 @@ class Printer(Base):
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
serial_number: Mapped[str] = mapped_column(String(50), unique=True)
ip_address: Mapped[str] = mapped_column(String(45))
ip_address: Mapped[str] = mapped_column(String(253))
access_code: Mapped[str] = mapped_column(String(20))
model: Mapped[str | None] = mapped_column(String(50))
location: Mapped[str | None] = mapped_column(String(100)) # Group/location name
+10 -2
View File
@@ -6,7 +6,11 @@ from pydantic import BaseModel, Field
class PrinterBase(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
serial_number: str = Field(..., min_length=1, max_length=50)
ip_address: str = Field(..., pattern=r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$")
ip_address: str = Field(
...,
max_length=253,
pattern=r"^(\d{1,3}(\.\d{1,3}){3}|[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)*)$",
)
access_code: str = Field(..., min_length=1, max_length=20)
model: str | None = None
location: str | None = None # Group/location name
@@ -31,7 +35,11 @@ class PlateDetectionROI(BaseModel):
class PrinterUpdate(BaseModel):
name: str | None = None
ip_address: str | None = None
ip_address: str | None = Field(
default=None,
max_length=253,
pattern=r"^(\d{1,3}(\.\d{1,3}){3}|[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)*)$",
)
access_code: str | None = None
model: str | None = None
location: str | None = None
@@ -193,13 +193,39 @@ class CertificateService:
return ca_key, ca_cert
def generate_certificates(self) -> tuple[Path, Path]:
def _build_san_entries(self, local_ip: str, additional_ips: list[str] | None) -> list[x509.GeneralName]:
"""Build Subject Alternative Name entries for the printer certificate."""
entries: list[x509.GeneralName] = [
x509.DNSName("localhost"),
x509.DNSName("bambuddy"),
x509.DNSName(self.serial),
x509.IPAddress(IPv4Address(local_ip)),
x509.IPAddress(IPv4Address("127.0.0.1")),
]
seen_ips = {local_ip, "127.0.0.1"}
if additional_ips:
for ip in additional_ips:
if ip and ip not in seen_ips:
try:
entries.append(x509.IPAddress(IPv4Address(ip)))
seen_ips.add(ip)
logger.info("Added additional SAN IP: %s", ip)
except ValueError:
logger.warning("Skipping invalid additional SAN IP: %s", ip)
return entries
def generate_certificates(self, additional_ips: list[str] | None = None) -> tuple[Path, Path]:
"""Generate printer certificate (reusing existing CA if available).
Creates a certificate chain mimicking real Bambu printers:
- CA certificate (reused if exists and valid, otherwise generated)
- Printer certificate (CN=serial, signed by CA)
Args:
additional_ips: Extra IP addresses to include in certificate SAN.
Used in proxy mode to include the remote interface IP so the
slicer's TLS handshake succeeds when connecting to the proxy.
Returns:
Tuple of (cert_path, key_path)
"""
@@ -245,15 +271,7 @@ class CertificateService:
critical=True,
)
.add_extension(
x509.SubjectAlternativeName(
[
x509.DNSName("localhost"),
x509.DNSName("bambuddy"),
x509.DNSName(self.serial),
x509.IPAddress(IPv4Address(local_ip)),
x509.IPAddress(IPv4Address("127.0.0.1")),
]
),
x509.SubjectAlternativeName(self._build_san_entries(local_ip, additional_ips)),
critical=False,
)
.add_extension(
@@ -9,6 +9,7 @@ immediately upon connection, before any FTP commands are exchanged.
import asyncio
import logging
import os
import random
import ssl
from collections.abc import Callable
@@ -31,6 +32,8 @@ class FTPSession:
access_code: str,
ssl_context: ssl.SSLContext,
on_file_received: Callable[[Path, str], None] | None,
passive_port_range: tuple[int, int] = (50000, 50100),
pasv_address: str = "",
):
self.reader = reader
self.writer = writer
@@ -38,6 +41,8 @@ class FTPSession:
self.access_code = access_code
self.ssl_context = ssl_context
self.on_file_received = on_file_received
self.passive_port_range = passive_port_range
self.pasv_address = pasv_address
self.authenticated = False
self.username: str | None = None
@@ -159,6 +164,7 @@ class FTPSession:
features = [
"211-Features:",
" PASV",
" EPSV",
" UTF8",
" SIZE",
"211 End",
@@ -196,6 +202,28 @@ class FTPSession:
else:
await self.send(504, "Type not supported")
async def _bind_passive_port(self) -> bool:
"""Try to bind a passive data port with retries.
Returns True if a port was successfully bound, False otherwise.
Sets self.data_server and self.data_port on success.
"""
port_min, port_max = self.passive_port_range
for attempt in range(10):
port = random.randint(port_min, port_max)
try:
self.data_server = await asyncio.start_server(
self._handle_data_connection,
"0.0.0.0", # nosec B104
port,
ssl=self.ssl_context,
)
self.data_port = port
return True
except OSError:
logger.debug("FTP passive port %s in use, retrying (%s/10)", port, attempt + 1)
return False
async def cmd_EPSV(self, arg: str) -> None:
"""Handle EPSV command - Extended Passive Mode (IPv6 compatible)."""
if not self.authenticated:
@@ -210,24 +238,12 @@ class FTPSession:
self._data_reader = None
self._data_writer = None
# Find a free port for passive data connection
self.data_port = random.randint(50000, 60000)
try:
# Create data server with TLS - use same context for session reuse
self.data_server = await asyncio.start_server(
self._handle_data_connection,
"0.0.0.0", # nosec B104
self.data_port,
ssl=self.ssl_context,
)
if await self._bind_passive_port():
# EPSV response format: 229 Entering Extended Passive Mode (|||port|)
await self.send(229, f"Entering Extended Passive Mode (|||{self.data_port}|)")
logger.info("FTP EPSV listening on port %s", self.data_port)
except Exception as e:
logger.error("Failed to create EPSV data connection: %s", e)
else:
logger.error("Failed to bind any passive port for EPSV")
await self.send(425, "Cannot open data connection")
async def cmd_PASV(self, arg: str) -> None:
@@ -244,22 +260,18 @@ class FTPSession:
self._data_reader = None
self._data_writer = None
# Find a free port for passive data connection
self.data_port = random.randint(50000, 60000)
try:
# Create data server with TLS
self.data_server = await asyncio.start_server(
self._handle_data_connection,
"0.0.0.0", # nosec B104
self.data_port,
ssl=self.ssl_context,
)
# Get server's IP for response
# Use the IP the client connected to
sockname = self.writer.get_extra_info("sockname")
ip = sockname[0] if sockname else "127.0.0.1"
if await self._bind_passive_port():
# Determine the IP to advertise in PASV response
if self.pasv_address:
# Explicit override (e.g., for Docker bridge mode behind NAT)
ip = self.pasv_address
else:
# Use the local IP of the control connection
sockname = self.writer.get_extra_info("sockname")
ip = sockname[0] if sockname else "127.0.0.1"
# 0.0.0.0 is not routable — fall back to control connection IP
if ip == "0.0.0.0":
ip = "127.0.0.1"
# Format IP and port for PASV response
ip_parts = ip.split(".")
@@ -270,10 +282,9 @@ class FTPSession:
227,
f"Entering Passive Mode ({ip_parts[0]},{ip_parts[1]},{ip_parts[2]},{ip_parts[3]},{port_hi},{port_lo})",
)
logger.info("FTP PASV listening on port %s", self.data_port)
except Exception as e:
logger.error("Failed to create passive data connection: %s", e)
logger.info("FTP PASV listening on %s:%s", ip, self.data_port)
else:
logger.error("Failed to bind any passive port for PASV")
await self.send(425, "Cannot open data connection")
async def _handle_data_connection(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
@@ -458,6 +469,9 @@ class FTPSession:
class VirtualPrinterFTPServer:
"""Implicit FTPS server that accepts uploads from slicers."""
PASSIVE_PORT_MIN = 50000
PASSIVE_PORT_MAX = 50100
def __init__(
self,
upload_dir: Path,
@@ -487,6 +501,8 @@ class VirtualPrinterFTPServer:
self._running = False
self._ssl_context: ssl.SSLContext | None = None
self._active_sessions: list[asyncio.Task] = []
# Override PASV response IP for Docker bridge mode / NAT environments
self._pasv_address = os.environ.get("VIRTUAL_PRINTER_PASV_ADDRESS", "")
async def start(self) -> None:
"""Start the implicit FTPS server."""
@@ -521,6 +537,13 @@ class VirtualPrinterFTPServer:
self._running = True
logger.info("Implicit FTPS server started on port %s", self.port)
logger.info(
"FTP passive data port range: %s-%s",
self.PASSIVE_PORT_MIN,
self.PASSIVE_PORT_MAX,
)
if self._pasv_address:
logger.info("FTP PASV address override: %s", self._pasv_address)
async with self._server:
await self._server.serve_forever()
@@ -549,6 +572,8 @@ class VirtualPrinterFTPServer:
access_code=self.access_code,
ssl_context=self._ssl_context,
on_file_received=self.on_file_received,
passive_port_range=(self.PASSIVE_PORT_MIN, self.PASSIVE_PORT_MAX),
pasv_address=self._pasv_address,
)
# Track the session task so we can cancel it on stop
@@ -296,8 +296,12 @@ class VirtualPrinterManager:
self._cert_service.serial = proxy_serial
# Regenerate printer cert if needed (CA is preserved)
# Include remote interface IP in SAN so slicer TLS succeeds
additional_ips = []
if self._remote_interface_ip:
additional_ips.append(self._remote_interface_ip)
self._cert_service.delete_printer_certificate()
cert_path, key_path = self._cert_service.generate_certificates()
cert_path, key_path = self._cert_service.generate_certificates(additional_ips=additional_ips or None)
logger.info("Generated certificate for proxy serial: %s", proxy_serial)
# Initialize TLS proxy with our certificates
@@ -359,9 +363,11 @@ class VirtualPrinterManager:
)
logger.info(
f"Virtual printer proxy started: "
f"FTP 0.0.0.0:{SlicerProxyManager.LOCAL_FTP_PORT} -> {self._target_printer_ip}:{SlicerProxyManager.PRINTER_FTP_PORT}, "
f"MQTT 0.0.0.0:{SlicerProxyManager.LOCAL_MQTT_PORT} -> {self._target_printer_ip}:{SlicerProxyManager.PRINTER_MQTT_PORT}"
"Virtual printer proxy target: FTP %s:%d, MQTT %s:%d",
self._target_printer_ip,
SlicerProxyManager.PRINTER_FTP_PORT,
self._target_printer_ip,
SlicerProxyManager.PRINTER_MQTT_PORT,
)
def _start_fallback_ssdp(self, proxy_serial: str, run_with_logging) -> None:
@@ -500,6 +506,11 @@ class VirtualPrinterManager:
# "review" mode (or legacy "queue" mode)
await self._queue_file(file_path, source_ip)
# Reset MQTT status back to IDLE after file processing
# This tells the slicer the printer is done with the file
if self._mqtt and file_path.suffix.lower() == ".3mf":
self._mqtt.set_gcode_state("IDLE")
async def _on_print_command(self, filename: str, data: dict) -> None:
"""Handle print command from MQTT.
@@ -584,7 +595,12 @@ class VirtualPrinterManager:
# Only queue 3MF files
if file_path.suffix.lower() != ".3mf":
logger.warning("Skipping non-3MF file: %s", file_path.name)
logger.debug("Skipping non-3MF file: %s", file_path.name)
self._pending_files.pop(file_path.name, None)
try:
file_path.unlink()
except OSError:
pass # Best-effort removal of non-3MF file; may already be gone
return
try:
@@ -181,6 +181,11 @@ class SimpleMQTTServer:
self._status_push_task: asyncio.Task | None = None
self._sequence_id = 0
# Dynamic state for status reports
self._gcode_state = "IDLE"
self._current_file = ""
self._prepare_percent = "0"
async def start(self) -> None:
"""Start the MQTT server."""
if self._running:
@@ -521,10 +526,10 @@ class SimpleMQTTServer:
"sequence_id": str(self._sequence_id),
"command": "push_status",
"msg": 0,
"gcode_state": "IDLE",
"gcode_file": "",
"gcode_file_prepare_percent": "0",
"subtask_name": "",
"gcode_state": self._gcode_state,
"gcode_file": self._current_file,
"gcode_file_prepare_percent": self._prepare_percent,
"subtask_name": self._current_file.replace(".3mf", "") if self._current_file else "",
"mc_print_stage": "",
"mc_percent": 0,
"mc_remaining_time": 0,
@@ -589,38 +594,7 @@ class SimpleMQTTServer:
}
}
topic = f"device/{self.serial}/report"
message = json.dumps(status)
# Build MQTT PUBLISH packet
topic_bytes = topic.encode("utf-8")
message_bytes = message.encode("utf-8")
# Calculate remaining length
remaining = 2 + len(topic_bytes) + len(message_bytes)
# Build packet
packet = bytes([0x30]) # PUBLISH, QoS 0
# Encode remaining length
while remaining > 0:
byte = remaining % 128
remaining //= 128
if remaining > 0:
byte |= 0x80
packet += bytes([byte])
# Topic length and topic
packet += bytes([len(topic_bytes) >> 8, len(topic_bytes) & 0xFF])
packet += topic_bytes
# Message payload
packet += message_bytes
writer.write(packet)
await writer.drain()
logger.info("Sent initial status report on %s", topic)
await self._publish_to_report(writer, status)
except OSError as e:
logger.error("Failed to send status report: %s", e)
@@ -684,42 +658,75 @@ class SimpleMQTTServer:
}
}
topic = f"device/{self.serial}/report"
message = json.dumps(version_info)
# Build MQTT PUBLISH packet
topic_bytes = topic.encode("utf-8")
message_bytes = message.encode("utf-8")
# Calculate remaining length
remaining = 2 + len(topic_bytes) + len(message_bytes)
# Build packet
packet = bytes([0x30]) # PUBLISH, QoS 0
# Encode remaining length
while remaining > 0:
byte = remaining % 128
remaining //= 128
if remaining > 0:
byte |= 0x80
packet += bytes([byte])
# Topic length and topic
packet += bytes([len(topic_bytes) >> 8, len(topic_bytes) & 0xFF])
packet += topic_bytes
# Message payload
packet += message_bytes
writer.write(packet)
await writer.drain()
logger.info("Sent version response on %s", topic)
await self._publish_to_report(writer, version_info)
logger.info("Sent version response")
except OSError as e:
logger.error("Failed to send version response: %s", e)
def set_gcode_state(self, state: str, filename: str = "", prepare_percent: str = "0") -> None:
"""Update the gcode state reported to connected slicers.
Called by the manager to reflect FTP upload progress/completion.
"""
self._gcode_state = state
self._current_file = filename
self._prepare_percent = prepare_percent
async def _publish_to_report(self, writer: asyncio.StreamWriter, payload: dict) -> None:
"""Publish a message on the device report topic."""
topic = f"device/{self.serial}/report"
message = json.dumps(payload)
topic_bytes = topic.encode("utf-8")
message_bytes = message.encode("utf-8")
remaining = 2 + len(topic_bytes) + len(message_bytes)
packet = bytes([0x30]) # PUBLISH, QoS 0
while remaining > 0:
byte = remaining % 128
remaining //= 128
if remaining > 0:
byte |= 0x80
packet += bytes([byte])
packet += bytes([len(topic_bytes) >> 8, len(topic_bytes) & 0xFF])
packet += topic_bytes
packet += message_bytes
writer.write(packet)
await writer.drain()
async def _send_print_response(self, writer: asyncio.StreamWriter, sequence_id: str, filename: str) -> None:
"""Send project_file acknowledgment matching real Bambu printer behavior."""
# Update state so periodic status pushes reflect preparation
self._gcode_state = "PREPARE"
self._current_file = filename
self._prepare_percent = "0"
try:
# Send command acknowledgment — slicer expects to see
# command: "project_file" echoed back before starting FTP upload
subtask_name = filename.replace(".3mf", "") if filename else ""
response = {
"print": {
"command": "project_file",
"sequence_id": sequence_id,
"param": "Metadata/plate_1.gcode",
"subtask_name": subtask_name,
"gcode_state": "PREPARE",
"gcode_file": filename,
"gcode_file_prepare_percent": "0",
"result": "SUCCESS",
"msg": 0,
}
}
await self._publish_to_report(writer, response)
logger.info("Sent project_file acknowledgment for %s", filename)
except OSError as e:
logger.error("Failed to send print response: %s", e)
async def _handle_publish(self, header: int, payload: bytes, writer: asyncio.StreamWriter) -> None:
"""Handle MQTT PUBLISH packet."""
try:
@@ -776,11 +783,17 @@ class SimpleMQTTServer:
print_data = data["print"]
command = print_data.get("command", "")
filename = print_data.get("subtask_name", "")
sequence_id = print_data.get("sequence_id", "0")
logger.info("MQTT print command: %s for %s", command, filename)
if self.on_print_command and command == "project_file":
await self._notify_print_command(filename, print_data)
if command == "project_file":
# Respond with PREPARE status so slicer proceeds with FTP upload
file_3mf = print_data.get("file", filename)
await self._send_print_response(writer, sequence_id, file_3mf)
if self.on_print_command:
await self._notify_print_command(filename, print_data)
except json.JSONDecodeError:
pass # Non-JSON payloads on request topic are safely ignored
@@ -328,8 +328,13 @@ class SSDPProxy:
pass # Return partial headers if parsing fails; malformed packets are common
return headers
def _rewrite_ssdp_location(self, data: bytes) -> bytes:
"""Rewrite SSDP message with Bambuddy's remote IP as Location."""
def _rewrite_ssdp(self, data: bytes) -> bytes:
"""Rewrite SSDP message for proxy re-broadcast.
- Location: changed to Bambuddy's remote interface IP
- DevBind: forced to 'free' so the slicer treats the proxy as a
LAN-only printer (avoids cloud auth requirement for sending prints)
"""
try:
text = data.decode("utf-8", errors="ignore")
original = text
@@ -340,11 +345,25 @@ class SSDPProxy:
text,
flags=re.IGNORECASE,
)
# Force DevBind to 'free' - ensures slicer uses LAN mode for
# both monitoring AND sending prints through the proxy
text = re.sub(
r"(DevBind\.bambu\.com:\s*)\S+",
r"\g<1>free",
text,
flags=re.IGNORECASE,
)
# Append " - Proxy" to printer name so it's distinguishable
text = re.sub(
r"(DevName\.bambu\.com:\s*)(.+)",
r"\g<1>\g<2> - Proxy",
text,
flags=re.IGNORECASE,
)
if text != original:
logger.debug("Rewrote SSDP Location to %s", self.remote_interface_ip)
logger.debug("Rewritten SSDP packet:\n%s", text)
logger.debug("Rewrote SSDP for proxy:\n%s", text)
else:
logger.warning("SSDP Location rewrite had no effect. Packet:\n%s", original)
logger.warning("SSDP rewrite had no effect. Packet:\n%s", original)
return text.encode("utf-8")
except Exception as e:
logger.error("Failed to rewrite SSDP: %s", e)
@@ -453,13 +472,26 @@ class SSDPProxy:
self._remote_socket = None
async def _handle_local_packet(self, data: bytes, addr: tuple[str, int]) -> None:
"""Handle SSDP packet received on local interface (LAN A)."""
"""Handle SSDP packet received on local interface (LAN A).
Processes two types of traffic:
- NOTIFY from the real printer → cache and re-broadcast on LAN B
- M-SEARCH from slicers on LAN B → respond with cached printer info
"""
sender_ip = addr[0]
# Only process packets from the target printer
if sender_ip != self.target_printer_ip:
# Ignore packets from our own interfaces (prevent loops)
if sender_ip in (self.local_interface_ip, self.remote_interface_ip):
return
# Handle M-SEARCH from slicers (any IP that's not the target printer)
if sender_ip != self.target_printer_ip:
if b"M-SEARCH" in data:
await self._respond_to_msearch(data, addr)
return
# Below: NOTIFY handling from the real printer
# Check if it's a NOTIFY message
if b"NOTIFY" not in data and b"HTTP/1.1 200" not in data:
return
@@ -478,6 +510,44 @@ class SSDPProxy:
self._last_printer_ssdp = data
await self._broadcast_to_remote()
async def _respond_to_msearch(self, data: bytes, addr: tuple[str, int]) -> None:
"""Respond to M-SEARCH from a slicer with cached, rewritten printer info.
When Bambu Studio sends an M-SEARCH (e.g., before sending a print),
we respond with the cached printer info, rewritten to point to the
proxy's LAN B IP. Without this, the slicer thinks the printer is
offline and shows a 'connect to printer' modal.
"""
# Check if it's a relevant M-SEARCH
if b"bambulab-com:device:3dprinter" not in data and b"ssdp:all" not in data.lower():
return
if not self._last_printer_ssdp:
logger.debug("M-SEARCH from %s but no cached printer SSDP yet", addr[0])
return
logger.debug("Received M-SEARCH from slicer %s", addr[0])
# Rewrite the cached printer SSDP (Location → proxy IP, DevBind → free)
rewritten = self._rewrite_ssdp(self._last_printer_ssdp)
text = rewritten.decode("utf-8", errors="ignore")
# Convert NOTIFY format to M-SEARCH response format:
# "NOTIFY * HTTP/1.1" → "HTTP/1.1 200 OK"
# NT: → ST: (Notification Type → Search Target)
# Remove NTS: header (only in NOTIFY)
text = re.sub(r"^NOTIFY \* HTTP/1\.1", "HTTP/1.1 200 OK", text)
text = re.sub(r"^NT:", "ST:", text, flags=re.MULTILINE)
text = re.sub(r"^NTS:.*\r\n", "", text, flags=re.MULTILINE)
# Send unicast response directly to the slicer via remote socket
if self._remote_socket:
try:
self._remote_socket.sendto(text.encode("utf-8"), addr)
logger.info("Sent SSDP M-SEARCH response to %s", addr[0])
except OSError as e:
logger.debug("Failed to send M-SEARCH response to %s: %s", addr[0], e)
async def _broadcast_to_remote(self) -> None:
"""Broadcast cached printer SSDP on remote interface (LAN B)."""
if not self._remote_socket or not self._last_printer_ssdp:
@@ -485,7 +555,7 @@ class SSDPProxy:
try:
# Rewrite Location to point to Bambuddy's remote interface
rewritten = self._rewrite_ssdp_location(self._last_printer_ssdp)
rewritten = self._rewrite_ssdp(self._last_printer_ssdp)
# Calculate broadcast address for remote network
# Use 255.255.255.255 for simplicity (works across subnets)
@@ -12,13 +12,58 @@ Unlike a transparent TCP proxy, this terminates TLS on both ends:
import asyncio
import logging
import random
import re
import ssl
import subprocess
from collections.abc import Callable
from pathlib import Path
logger = logging.getLogger(__name__)
def detect_port_redirect(port: int) -> int | None:
"""Detect if iptables redirects a port to another port.
When iptables NAT REDIRECT rules exist (e.g. 990→9990), connections
to the original port never reach our socket because iptables intercepts
them in PREROUTING. We must listen on the redirect target instead.
Returns the redirect target port, or None if no redirect is active.
"""
# Method 1: Read persistent rules file (doesn't require root)
for rules_path in ("/etc/iptables/rules.v4", "/etc/iptables.rules"):
try:
with open(rules_path) as f:
content = f.read()
match = re.search(rf"--dport {port}\b.*?--to-ports\s+(\d+)", content)
if match:
target = int(match.group(1))
if target != port:
return target
except (FileNotFoundError, PermissionError, OSError):
continue
# Method 2: Query live iptables rules (may require root)
try:
result = subprocess.run( # noqa: S603, S607
["iptables-save", "-t", "nat"],
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0:
match = re.search(rf"--dport {port}\b.*?--to-ports\s+(\d+)", result.stdout)
if match:
target = int(match.group(1))
if target != port:
return target
except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
pass
return None
class TLSProxy:
"""TLS terminating proxy that forwards data between client and target.
@@ -115,6 +160,17 @@ class TLSProxy:
except OSError as e:
if e.errno == 98: # Address already in use
logger.error("%s proxy port %s is already in use", self.name, self.listen_port)
elif e.errno == 13: # Permission denied
logger.error(
"%s proxy: cannot bind to port %s (permission denied). "
"Port %s requires root or CAP_NET_BIND_SERVICE. "
"Docker: add 'cap_add: [NET_BIND_SERVICE]' to docker-compose.yml. "
"Native: use 'sudo setcap cap_net_bind_service=+ep $(which python3)' "
"or redirect with iptables.",
self.name,
self.listen_port,
self.listen_port,
)
else:
logger.error("%s proxy error: %s", self.name, e)
except asyncio.CancelledError:
@@ -284,6 +340,503 @@ class TLSProxy:
logger.debug("%s proxy %s: total %s bytes", self.name, direction, total_bytes)
class FTPTLSProxy(TLSProxy):
"""FTP-aware TLS proxy that handles passive data connections.
Extends TLSProxy to intercept PASV/EPSV responses on the FTP control
channel, dynamically create TLS data proxies on local ports, and rewrite
the responses so the slicer connects to the proxy instead of the printer.
Without this, FTP passive data connections bypass the proxy and go directly
to the printer, which fails when the slicer can't reach the printer's IP.
"""
PASV_PORT_MIN = 50000
PASV_PORT_MAX = 50100
async def stop(self) -> None:
"""Stop proxy and clean up data connection servers."""
# Close all data servers first
for server in list(self._data_servers):
try:
server.close()
await server.wait_closed()
except OSError:
pass # Best-effort cleanup of data proxy servers
self._data_servers.clear()
await super().stop()
async def start(self) -> None:
"""Start the FTP TLS proxy."""
self._data_servers: list[asyncio.Server] = []
await super().start()
async def _handle_client(
self,
client_reader: asyncio.StreamReader,
client_writer: asyncio.StreamWriter,
) -> None:
"""Handle FTP client with PASV/EPSV-aware response forwarding."""
peername = client_writer.get_extra_info("peername")
client_id = f"{peername[0]}:{peername[1]}" if peername else "unknown"
logger.info("%s proxy: client connected from %s", self.name, client_id)
if self.on_connect:
try:
self.on_connect(client_id)
except Exception:
pass # Ignore connect callback errors; connection proceeds regardless
# Determine our local IP from the control connection socket
sockname = client_writer.get_extra_info("sockname")
local_ip = sockname[0] if sockname else "0.0.0.0"
if local_ip in ("0.0.0.0", "::"):
local_ip = "127.0.0.1"
# Connect to target printer with TLS
try:
printer_reader, printer_writer = await asyncio.wait_for(
asyncio.open_connection(
self.target_host,
self.target_port,
ssl=self._client_ssl_context,
),
timeout=10.0,
)
logger.info("%s proxy: connected to printer %s:%s", self.name, self.target_host, self.target_port)
except TimeoutError:
logger.error("%s proxy: timeout connecting to %s:%s", self.name, self.target_host, self.target_port)
client_writer.close()
await client_writer.wait_closed()
return
except ssl.SSLError as e:
logger.error(
"%s proxy: SSL error connecting to %s:%s: %s", self.name, self.target_host, self.target_port, e
)
client_writer.close()
await client_writer.wait_closed()
return
except OSError as e:
logger.error("%s proxy: failed to connect to %s:%s: %s", self.name, self.target_host, self.target_port, e)
client_writer.close()
await client_writer.wait_closed()
return
# Track data channel protection level per session.
# PROT C = cleartext data, PROT P = TLS data.
# Default to cleartext — many Bambu printers (A1, H2D) use PROT C.
# If the slicer sends PROT P, we switch to TLS for data connections.
session_state: dict[str, str] = {"prot": "C"}
# Client→Printer: intercept EPSV and replace with PASV
# EPSV responses only contain a port (no IP), so the slicer reuses
# the control connection IP. If that IP is the real printer (via
# iptables REDIRECT), the data connection bypasses the proxy.
# PASV responses include an explicit IP that we can rewrite.
client_to_printer = asyncio.create_task(
self._forward_ftp_commands(client_reader, printer_writer, f"{client_id}→printer", session_state),
name=f"{self.name}_c2p_{client_id}",
)
# Printer→Client: intercept PASV/EPSV responses
printer_to_client = asyncio.create_task(
self._forward_ftp_control(printer_reader, client_writer, f"printer→{client_id}", local_ip, session_state),
name=f"{self.name}_p2c_{client_id}",
)
self._active_connections[client_id] = (client_to_printer, printer_to_client)
try:
done, pending = await asyncio.wait(
[client_to_printer, printer_to_client],
return_when=asyncio.FIRST_COMPLETED,
)
for task in pending:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass # Expected when cancelling the other forwarding direction
except Exception as e:
logger.debug("%s proxy connection error: %s", self.name, e)
finally:
self._active_connections.pop(client_id, None)
for writer in [client_writer, printer_writer]:
try:
writer.close()
await writer.wait_closed()
except OSError:
pass # Best-effort connection cleanup; peer may have disconnected
logger.info("%s proxy: client %s disconnected", self.name, client_id)
if self.on_disconnect:
try:
self.on_disconnect(client_id)
except Exception:
pass # Ignore disconnect callback errors; cleanup continues
async def _forward_ftp_commands(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
direction: str,
session_state: dict[str, str],
) -> None:
"""Forward FTP client commands, replacing EPSV with PASV.
EPSV responses only contain a port number — the client reuses the
control connection IP for data. When the control IP is the real
printer (due to iptables REDIRECT), EPSV data connections bypass
the proxy. PASV responses include an explicit IP that the proxy
can rewrite to its own address.
Also tracks PROT P/C commands to know whether data connections
should use TLS or cleartext.
"""
buffer = b""
total_bytes = 0
try:
while self._running:
data = await reader.read(65536)
if not data:
break
total_bytes += len(data)
buffer += data
output = b""
while b"\r\n" in buffer:
idx = buffer.index(b"\r\n")
line = buffer[:idx]
buffer = buffer[idx + 2 :]
cmd_upper = line.strip().upper()
# Replace EPSV with PASV so response includes an IP
if cmd_upper == b"EPSV":
line = b"PASV"
logger.info("FTP command rewrite: EPSV → PASV")
# Track PROT level for data channel encryption
elif cmd_upper == b"PROT P":
session_state["prot"] = "P"
logger.info("FTP data protection: PROT P (TLS)")
elif cmd_upper == b"PROT C":
session_state["prot"] = "C"
logger.info("FTP data protection: PROT C (cleartext)")
output += line + b"\r\n"
if output:
writer.write(output)
await writer.drain()
logger.debug("%s proxy %s: %s bytes", self.name, direction, len(data))
except asyncio.CancelledError:
pass # Expected when the other forwarding direction closes first
except ConnectionResetError:
logger.debug("%s proxy %s: connection reset", self.name, direction)
except BrokenPipeError:
logger.debug("%s proxy %s: broken pipe", self.name, direction)
except OSError as e:
logger.debug("%s proxy %s error: %s", self.name, direction, e)
if buffer:
try:
writer.write(buffer)
await writer.drain()
except OSError:
pass # Best-effort flush of remaining FTP command data
logger.debug("%s proxy %s: total %s bytes", self.name, direction, total_bytes)
async def _forward_ftp_control(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
direction: str,
local_ip: str,
session_state: dict[str, str],
) -> None:
"""Forward FTP control channel responses, rewriting PASV/EPSV.
FTP control channel is line-based (\\r\\n terminated). We buffer data
and process complete lines, intercepting 227 (PASV) and 229 (EPSV)
responses to create local data proxies.
"""
buffer = b""
total_bytes = 0
try:
while self._running:
data = await reader.read(65536)
if not data:
break
total_bytes += len(data)
buffer += data
output = b""
# Process all complete lines
while b"\r\n" in buffer:
idx = buffer.index(b"\r\n")
line = buffer[:idx]
buffer = buffer[idx + 2 :]
rewritten = await self._maybe_rewrite_pasv(line, local_ip, session_state)
output += rewritten + b"\r\n"
if output:
writer.write(output)
await writer.drain()
logger.debug("%s proxy %s: %s bytes", self.name, direction, len(data))
except asyncio.CancelledError:
pass # Expected when the other forwarding direction closes first
except ConnectionResetError:
logger.debug("%s proxy %s: connection reset", self.name, direction)
except BrokenPipeError:
logger.debug("%s proxy %s: broken pipe", self.name, direction)
except OSError as e:
logger.debug("%s proxy %s error: %s", self.name, direction, e)
# Flush any remaining buffered data
if buffer:
try:
writer.write(buffer)
await writer.drain()
except OSError:
pass # Best-effort flush of remaining FTP control data
logger.debug("%s proxy %s: total %s bytes", self.name, direction, total_bytes)
async def _maybe_rewrite_pasv(self, line: bytes, local_ip: str, session_state: dict[str, str]) -> bytes:
"""Rewrite PASV/EPSV response to point to a local data proxy."""
try:
text = line.decode("utf-8")
except UnicodeDecodeError:
return line
# 227 Entering Passive Mode (h1,h2,h3,h4,p1,p2)
if text.startswith("227 "):
match = re.search(r"\(\s*(\d+)\s*,\s*(\d+)\s*,\s*(\d+)\s*,\s*(\d+)\s*,\s*(\d+)\s*,\s*(\d+)\s*\)", text)
if match:
h1, h2, h3, h4, p1, p2 = (int(x) for x in match.groups())
printer_ip = f"{h1}.{h2}.{h3}.{h4}"
printer_port = p1 * 256 + p2
local_port = await self._create_data_proxy(printer_ip, printer_port, session_state)
if local_port:
ip_parts = local_ip.split(".")
lp1 = local_port // 256
lp2 = local_port % 256
rewritten = (
f"227 Entering Passive Mode "
f"({ip_parts[0]},{ip_parts[1]},{ip_parts[2]},{ip_parts[3]},{lp1},{lp2})"
)
logger.info("FTP PASV rewrite: %s:%s → %s:%s", printer_ip, printer_port, local_ip, local_port)
return rewritten.encode("utf-8")
else:
logger.error("FTP PASV: failed to create data proxy for %s:%s", printer_ip, printer_port)
else:
logger.warning("FTP PASV: 227 response didn't match expected format: %s", text[:100])
# 229 Entering Extended Passive Mode (|||port|)
elif text.startswith("229 "):
match = re.search(r"\(\|\|\|(\d+)\|\)", text)
if match:
printer_port = int(match.group(1))
local_port = await self._create_data_proxy(self.target_host, printer_port, session_state)
if local_port:
rewritten = f"229 Entering Extended Passive Mode (|||{local_port}|)"
logger.info("FTP EPSV rewrite: port %s → %s", printer_port, local_port)
return rewritten.encode("utf-8")
else:
logger.error("FTP EPSV: failed to create data proxy for port %s", printer_port)
else:
logger.warning("FTP EPSV: 229 response didn't match expected format: %s", text[:100])
return line
async def _create_data_proxy(self, printer_ip: str, printer_port: int, session_state: dict[str, str]) -> int | None:
"""Create a one-shot proxy for an FTP data connection.
Prefers the printer's original passive port so the port number stays
the same in the rewritten PASV/EPSV response. This is critical when
the slicer's FTP bounce-attack protection overrides the IP in the PASV
response: the slicer connects to <control_IP>:<port>, and if iptables
REDIRECT maps that port to the local machine, the data proxy must be
listening on the *same* port number.
Falls back to a random port if the original is unavailable.
Uses TLS or cleartext based on the session's PROT level:
- PROT P: TLS on both slicer and printer data connections
- PROT C: cleartext on both sides (common for A1/H2D printers)
Returns the local port number, or None if binding failed.
"""
use_tls = session_state.get("prot") == "P"
logger.info(
"FTP data proxy: creating data proxy for %s:%s (printer-side %s)",
printer_ip,
printer_port,
"TLS" if use_tls else "cleartext",
)
# Try the printer's original port first — this ensures the port
# matches even when bounce protection or iptables REDIRECT is in play.
try:
await self._start_data_proxy_server(printer_port, printer_ip, printer_port, use_tls)
logger.info("FTP data proxy: using printer's port %s", printer_port)
return printer_port
except OSError as e:
logger.debug(
"FTP data proxy: printer port %s unavailable (%s), trying random",
printer_port,
e,
)
for _attempt in range(10):
port = random.randint(self.PASV_PORT_MIN, self.PASV_PORT_MAX)
try:
await self._start_data_proxy_server(port, printer_ip, printer_port, use_tls)
logger.info("FTP data proxy: using random port %s", port)
return port
except OSError:
continue
logger.error("Failed to bind FTP data proxy port after 10 attempts")
return None
async def _start_data_proxy_server(self, port: int, printer_ip: str, printer_port: int, use_tls: bool) -> None:
"""Start a one-shot server for one FTP data connection.
The slicer-side listener is ALWAYS cleartext. Even when the slicer
sends PROT P on the control channel, Bambu Studio does not perform
a TLS handshake on the data connection — it relies on the implicit
FTPS control channel for authentication and sends data unencrypted.
The printer-side outbound connection follows the PROT level:
- PROT P (use_tls=True): TLS to the printer's data port
- PROT C (use_tls=False): cleartext to the printer's data port
This mirrors the control channel's TLS-termination architecture.
Raises OSError if the port is already in use.
"""
connected = asyncio.Event()
server_holder: list[asyncio.Server] = []
# Slicer side: ALWAYS cleartext — Bambu Studio does not do TLS on
# the data channel even after sending PROT P.
# Printer side: TLS if PROT P, cleartext if PROT C.
client_ssl = self._client_ssl_context if use_tls else None
printer_mode = "TLS" if use_tls else "cleartext"
async def handle_data(
client_reader: asyncio.StreamReader,
client_writer: asyncio.StreamWriter,
) -> None:
"""Handle one FTP data connection, then close the server."""
peername = client_writer.get_extra_info("peername")
data_client = f"{peername[0]}:{peername[1]}" if peername else "unknown"
logger.info(
"FTP data proxy port %s (slicer=cleartext, printer=%s): client connected from %s, bridging to %s:%s",
port,
printer_mode,
data_client,
printer_ip,
printer_port,
)
connected.set()
# One-shot: close server after accepting first connection
if server_holder:
server_holder[0].close()
printer_writer = None
try:
# Connect to printer's data port
printer_reader, printer_writer = await asyncio.wait_for(
asyncio.open_connection(
printer_ip,
printer_port,
ssl=client_ssl,
),
timeout=10.0,
)
logger.info(
"FTP data proxy port %s (printer=%s): connected to printer %s:%s",
port,
printer_mode,
printer_ip,
printer_port,
)
# Bidirectional data forwarding
c2p = asyncio.create_task(self._forward(client_reader, printer_writer, "data_c2p"))
p2c = asyncio.create_task(self._forward(printer_reader, client_writer, "data_p2c"))
done, pending = await asyncio.wait([c2p, p2c], return_when=asyncio.FIRST_COMPLETED)
for task in pending:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass # Expected when other data direction closes
except TimeoutError:
logger.error("FTP data proxy port %s: timeout connecting to printer", port)
except ssl.SSLError as e:
logger.error("FTP data proxy port %s: SSL error to printer: %s", port, e)
except Exception as e:
logger.error("FTP data proxy port %s: error: %s", port, e)
finally:
for w in [client_writer, printer_writer]:
if w:
try:
w.close()
await w.wait_closed()
except OSError:
pass # Best-effort data connection cleanup
logger.info("FTP data proxy port %s: connection closed", port)
server = await asyncio.start_server(
handle_data,
"0.0.0.0", # nosec B104
port,
# No TLS on slicer side — Bambu Studio doesn't do TLS on data
# channel even after PROT P. The proxy terminates TLS only on
# the printer side (inside handle_data).
)
server_holder.append(server)
self._data_servers.append(server)
# Auto-close after 60s if no connection arrives
async def auto_close() -> None:
try:
await asyncio.wait_for(connected.wait(), timeout=60.0)
except TimeoutError:
logger.debug("FTP data proxy on port %s timed out, closing", port)
try:
server.close()
await server.wait_closed()
except OSError:
pass # Best-effort timeout cleanup
finally:
if server in self._data_servers:
self._data_servers.remove(server)
asyncio.create_task(auto_close(), name=f"ftp_data_timeout_{port}")
logger.debug("FTP data proxy: port %s → %s:%s", port, printer_ip, printer_port)
class SlicerProxyManager:
"""Manages FTP and MQTT TLS proxies for a single printer target."""
@@ -324,10 +877,24 @@ class SlicerProxyManager:
"""Start FTP and MQTT TLS proxies."""
logger.info("Starting slicer TLS proxy to %s", self.target_host)
# Create proxies with TLS
self._ftp_proxy = TLSProxy(
# Detect iptables port redirect (e.g. 990→9990 for non-root installs).
# If active, connections to port 990 get intercepted by iptables PREROUTING
# and sent to the redirect target — our socket on 990 never sees them.
ftp_listen_port = self.LOCAL_FTP_PORT
redirect_target = detect_port_redirect(self.LOCAL_FTP_PORT)
if redirect_target:
logger.info(
"Detected iptables redirect: port %d → %d. FTP proxy will listen on %d.",
self.LOCAL_FTP_PORT,
redirect_target,
redirect_target,
)
ftp_listen_port = redirect_target
# Create FTP proxy with PASV/EPSV awareness for data connections
self._ftp_proxy = FTPTLSProxy(
name="FTP",
listen_port=self.LOCAL_FTP_PORT,
listen_port=ftp_listen_port,
target_host=self.target_host,
target_port=self.PRINTER_FTP_PORT,
server_cert_path=self.cert_path,
@@ -63,6 +63,58 @@ class TestPrintersAPI:
assert result["serial_number"] == "00M09A111111111"
assert result["model"] == "X1C"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_printer_with_hostname(self, async_client: AsyncClient):
"""Verify printer can be created with a hostname instead of IP address."""
data = {
"name": "DNS Printer",
"serial_number": "00M09A555555555",
"ip_address": "printer.local",
"access_code": "12345678",
"model": "P1S",
}
response = await async_client.post("/api/v1/printers/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "DNS Printer"
assert result["ip_address"] == "printer.local"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_printer_with_fqdn(self, async_client: AsyncClient):
"""Verify printer can be created with a fully qualified domain name."""
data = {
"name": "FQDN Printer",
"serial_number": "00M09A666666666",
"ip_address": "my-printer.home.lan",
"access_code": "12345678",
"model": "X1C",
}
response = await async_client.post("/api/v1/printers/", json=data)
assert response.status_code == 200
result = response.json()
assert result["ip_address"] == "my-printer.home.lan"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_printer_invalid_hostname(self, async_client: AsyncClient):
"""Verify invalid hostnames are rejected."""
data = {
"name": "Bad Printer",
"serial_number": "00M09A777777777",
"ip_address": "-invalid",
"access_code": "12345678",
}
response = await async_client.post("/api/v1/printers/", json=data)
assert response.status_code == 422
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_printer_duplicate_serial(self, async_client: AsyncClient, printer_factory, db_session):
+99
View File
@@ -0,0 +1,99 @@
"""Test fixtures for FTP service tests.
Provides a real implicit FTPS server (via mock_ftp_server) and client factory
for integration-style testing of BambuFTPClient against a live server.
"""
import socket
from unittest.mock import patch
import pytest
from backend.app.services.bambu_ftp import BambuFTPClient
from backend.app.services.virtual_printer.certificate import CertificateService
from backend.tests.unit.services.mock_ftp_server import MockBambuFTPServer
@pytest.fixture(scope="session")
def ftp_certs(tmp_path_factory):
"""Generate self-signed TLS certificates once per test session."""
cert_dir = tmp_path_factory.mktemp("ftp_certs")
svc = CertificateService(cert_dir, serial="TEST_FTP_SERVER")
cert_path, key_path = svc.generate_certificates()
return str(cert_path), str(key_path)
def _find_free_port() -> int:
"""Find a free TCP port on localhost."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
@pytest.fixture()
def ftp_root(tmp_path):
"""Create temp directory with standard Bambu printer directory structure."""
for d in ("cache", "timelapse", "model", "data", "data/Metadata"):
(tmp_path / d).mkdir(parents=True, exist_ok=True)
return tmp_path
@pytest.fixture()
def ftp_server(ftp_certs, ftp_root):
"""Start a mock implicit FTPS server, yield it, stop on cleanup."""
cert_path, key_path = ftp_certs
port = _find_free_port()
server = MockBambuFTPServer(
host="127.0.0.1",
port=port,
root_dir=str(ftp_root),
cert_path=cert_path,
key_path=key_path,
access_code="12345678",
)
server.start()
yield server
server.stop()
@pytest.fixture()
def ftp_client_factory(ftp_server):
"""Factory that creates BambuFTPClient instances pointed at the mock server."""
def _make_client(
printer_model: str = "X1C",
force_prot_c: bool = False,
access_code: str = "12345678",
timeout: float = 10.0,
) -> BambuFTPClient:
client = BambuFTPClient(
ip_address="127.0.0.1",
access_code=access_code,
timeout=timeout,
printer_model=printer_model,
force_prot_c=force_prot_c,
)
# Override port to point at mock server
client.FTP_PORT = ftp_server.port
return client
return _make_client
@pytest.fixture(autouse=True)
def clear_ftp_mode_cache():
"""Clear BambuFTPClient mode cache before and after each test."""
BambuFTPClient._mode_cache.clear()
yield
BambuFTPClient._mode_cache.clear()
@pytest.fixture()
def patch_ftp_port(ftp_server):
"""Patch FTP_PORT at class level for async wrapper tests.
Async wrappers create their own BambuFTPClient instances internally,
so we need to patch the class-level default port.
"""
with patch.object(BambuFTPClient, "FTP_PORT", ftp_server.port):
yield ftp_server
@@ -0,0 +1,240 @@
"""Mock implicit FTPS server for testing BambuFTPClient.
Built on pyftpdlib with implicit TLS support to match Bambu printer behavior.
Supports failure injection, custom AVBL command, and filesystem inspection.
"""
import logging
import os
import threading
import time
from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import TLS_FTPHandler
from pyftpdlib.servers import FTPServer
class ImplicitTLS_FTPHandler(TLS_FTPHandler):
"""FTP handler that wraps the socket in TLS before sending the 220 banner.
This implements implicit FTPS (port 990 style) where the TLS handshake
happens immediately on connect, before any FTP protocol exchange.
pyftpdlib only natively supports explicit FTPS (AUTH TLS after connect).
"""
# Per-class failure injection map: command -> (code, message, remaining_count)
# -1 remaining_count = permanent failure
_failure_map: dict = {}
# AVBL command response (bytes available)
_avbl_bytes: int = 1073741824 # 1 GB default
# Register AVBL as a recognized FTP command (pyftpdlib requires this)
proto_cmds = {
**TLS_FTPHandler.proto_cmds,
"AVBL": {
"perm": None,
"auth": True,
"arg": None,
"help": "Syntax: AVBL (get available bytes).",
},
}
def handle(self):
"""Wrap socket in TLS immediately, then send 220 banner."""
self.secure_connection(self.get_ssl_context())
super().handle()
def ftp_PROT(self, line):
"""Override PROT to auto-set _pbsz for implicit FTPS.
In implicit FTPS the connection is already TLS-secured, so requiring
a separate PBSZ command is unnecessary. Python's ftplib prot_c()
doesn't send PBSZ first (unlike prot_p()), causing 503 errors.
Real Bambu printers don't enforce this for implicit FTPS either.
"""
self._pbsz = True
return super().ftp_PROT(line)
def _check_failure(self, command: str, line: str):
"""Check if a failure is injected for this command.
Returns True if a failure response was sent, False otherwise.
"""
if command in self._failure_map:
code, message, remaining = self._failure_map[command]
if remaining != 0:
if remaining > 0:
self._failure_map[command] = (code, message, remaining - 1)
if remaining - 1 == 0:
del self._failure_map[command]
self.respond(f"{code} {message}")
return True
return False
def ftp_AVBL(self, line):
"""Handle custom AVBL command (available bytes on storage)."""
self.respond(f"213 {self._avbl_bytes}")
def ftp_RETR(self, file):
if self._check_failure("RETR", file):
return
return super().ftp_RETR(file)
def ftp_STOR(self, file):
if self._check_failure("STOR", file):
return
return super().ftp_STOR(file)
def ftp_DELE(self, line):
if self._check_failure("DELE", line):
return
return super().ftp_DELE(line)
def ftp_CWD(self, path):
if self._check_failure("CWD", path):
return
return super().ftp_CWD(path)
def ftp_LIST(self, path=""):
if self._check_failure("LIST", path):
return
return super().ftp_LIST(path)
def ftp_SIZE(self, path):
if self._check_failure("SIZE", path):
return
# Override to allow SIZE in ASCII mode (real Bambu printers allow it,
# and BambuFTPClient.get_file_size() doesn't set TYPE I first)
if not self.fs.isfile(self.fs.realpath(path)):
self.respond(f"550 {self.fs.fs2ftp(path)} is not retrievable.")
return
try:
size = self.run_as_current_user(self.fs.getsize, path)
except OSError as err:
self.respond(f"550 {err}.")
else:
self.respond(f"213 {size}")
def ftp_PASS(self, line):
if self._check_failure("PASS", line):
return
return super().ftp_PASS(line)
class MockBambuFTPServer:
"""Manages a mock implicit FTPS server in a background thread.
Simulates a Bambu printer FTP server with:
- Implicit TLS (like real printers on port 990)
- Standard Bambu directory structure
- AVBL command support
- Per-command failure injection for testing error paths
"""
def __init__(
self,
host: str,
port: int,
root_dir: str,
cert_path: str,
key_path: str,
access_code: str = "12345678",
):
self.host = host
self.port = port
self.root_dir = root_dir
self.cert_path = cert_path
self.key_path = key_path
self.access_code = access_code
self._server: FTPServer | None = None
self._thread: threading.Thread | None = None
# Create a unique handler class per instance so _failure_map is isolated
self._handler_class = type(
"TestFTPHandler",
(ImplicitTLS_FTPHandler,),
{
"_failure_map": {},
"_avbl_bytes": 1073741824,
},
)
def start(self):
"""Start the FTP server in a background daemon thread."""
authorizer = DummyAuthorizer()
authorizer.add_user("bblp", self.access_code, self.root_dir, perm="elradfmwMT")
handler = self._handler_class
handler.authorizer = authorizer
handler.certfile = self.cert_path
handler.keyfile = self.key_path
handler.passive_ports = range(60000, 60101)
handler.tls_control_required = False
handler.tls_data_required = False
# Reset ssl_context so it picks up our cert/key
handler.ssl_context = None
# Suppress pyftpdlib's noisy logging (startup/shutdown banners)
# to avoid "I/O operation on closed file" errors when xdist
# workers tear down while the daemon thread is still logging.
logging.getLogger("pyftpdlib").setLevel(logging.CRITICAL)
self._server = FTPServer((self.host, self.port), handler)
self._server.max_cons = 10
self._server.max_cons_per_ip = 5
self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
self._thread.start()
# Brief wait for server to be ready
time.sleep(0.1)
def stop(self):
"""Stop the FTP server and wait for thread to exit."""
if self._server:
self._server.close_all()
if self._thread:
self._thread.join(timeout=5)
self._server = None
self._thread = None
def inject_failure(self, command: str, code: int, message: str, count: int = -1):
"""Inject a failure response for a specific FTP command.
Args:
command: FTP command name (RETR, STOR, DELE, CWD, LIST, SIZE, PASS)
code: FTP response code (e.g. 550, 553)
message: Response message
count: Number of times to fail (-1 = permanent)
"""
self._handler_class._failure_map[command] = (code, message, count)
def clear_failures(self):
"""Remove all injected failures."""
self._handler_class._failure_map.clear()
def set_avbl_bytes(self, n: int):
"""Set the response value for the AVBL command."""
self._handler_class._avbl_bytes = n
def add_file(self, relative_path: str, content: bytes = b""):
"""Add a file to the server's filesystem."""
full_path = os.path.join(self.root_dir, relative_path.lstrip("/"))
os.makedirs(os.path.dirname(full_path), exist_ok=True)
with open(full_path, "wb") as f:
f.write(content)
def add_directory(self, relative_path: str):
"""Create a directory in the server's filesystem."""
full_path = os.path.join(self.root_dir, relative_path.lstrip("/"))
os.makedirs(full_path, exist_ok=True)
def file_exists(self, relative_path: str) -> bool:
"""Check if a file exists on the server."""
full_path = os.path.join(self.root_dir, relative_path.lstrip("/"))
return os.path.isfile(full_path)
def read_file(self, relative_path: str) -> bytes:
"""Read file content from the server's filesystem."""
full_path = os.path.join(self.root_dir, relative_path.lstrip("/"))
with open(full_path, "rb") as f:
return f.read()
@@ -0,0 +1,864 @@
"""Comprehensive FTP test suite for BambuFTPClient.
Tests against a real mock implicit FTPS server, covering:
- Connection (auth, SSL modes, timeout, caching)
- File listing
- Download (bytes, to_file, 0-byte regression)
- Upload (chunked transfer, progress, error codes)
- Delete
- File size
- Storage info (AVBL, directory scan, diagnose_storage)
- Model-specific behavior (X1C prot_p, A1 prot_c fallback)
- Async wrappers
- Failure injection scenarios (regressions for 0.1.8 bugs)
"""
import time
from pathlib import Path
import pytest
from backend.app.services.bambu_ftp import (
BambuFTPClient,
delete_file_async,
download_file_async,
download_file_try_paths_async,
list_files_async,
upload_file_async,
)
# Brief delay to allow pyftpdlib to flush uploaded files to disk.
# Needed because upload_file() skips voidresp() for A1 compatibility,
# so the server may still be processing the data channel close event.
_UPLOAD_FLUSH_DELAY = 0.3
# ---------------------------------------------------------------------------
# TestConnection
# ---------------------------------------------------------------------------
class TestConnection:
"""Tests for FTP connect/disconnect behavior."""
def test_connect_success(self, ftp_client_factory):
"""Successful implicit FTPS connection and login."""
client = ftp_client_factory()
assert client.connect() is True
client.disconnect()
def test_connect_wrong_access_code(self, ftp_client_factory):
"""Wrong access code returns False."""
client = ftp_client_factory(access_code="wrongcode")
assert client.connect() is False
def test_connect_unreachable_host(self, ftp_server):
"""Unreachable host returns False."""
client = BambuFTPClient(
ip_address="192.0.2.1", # TEST-NET, guaranteed unreachable
access_code="12345678",
timeout=1.0,
printer_model="X1C",
)
client.FTP_PORT = ftp_server.port
assert client.connect() is False
def test_connect_timeout(self, ftp_server):
"""Very short timeout triggers timeout error."""
client = BambuFTPClient(
ip_address="192.0.2.1",
access_code="12345678",
timeout=0.001, # Extremely short
printer_model="X1C",
)
client.FTP_PORT = ftp_server.port
assert client.connect() is False
def test_disconnect_clean(self, ftp_client_factory):
"""Clean disconnect after successful connect."""
client = ftp_client_factory()
client.connect()
client.disconnect()
assert client._ftp is None
def test_disconnect_without_connect(self, ftp_client_factory):
"""Disconnect without connect does not raise."""
client = ftp_client_factory()
client.disconnect() # Should not raise
assert client._ftp is None
def test_disconnect_after_server_gone(self, ftp_certs, ftp_root):
"""Disconnect after server has stopped raises EOFError.
Note: The current disconnect() catches (OSError, ftplib.Error) but
EOFError is neither. This documents actual behavior — a future fix
could add EOFError to the except clause.
"""
from backend.tests.unit.services.mock_ftp_server import (
MockBambuFTPServer,
)
from .conftest import _find_free_port
cert_path, key_path = ftp_certs
port = _find_free_port()
server = MockBambuFTPServer("127.0.0.1", port, str(ftp_root), cert_path, key_path)
server.start()
client = BambuFTPClient("127.0.0.1", "12345678", timeout=5.0)
client.FTP_PORT = port
client.connect()
server.stop()
with pytest.raises(EOFError):
client.disconnect()
def test_x1c_uses_prot_p(self, ftp_client_factory):
"""X1C model connects with prot_p (protected data channel)."""
client = ftp_client_factory(printer_model="X1C")
assert client.connect() is True
assert client._should_use_prot_c() is False
client.disconnect()
def test_a1_defaults_prot_p(self, ftp_client_factory):
"""A1 model defaults to prot_p when no cache exists."""
client = ftp_client_factory(printer_model="A1")
assert client._should_use_prot_c() is False
assert client.connect() is True
client.disconnect()
def test_a1_force_prot_c(self, ftp_client_factory):
"""A1 model with force_prot_c uses clear data channel."""
client = ftp_client_factory(printer_model="A1", force_prot_c=True)
assert client._should_use_prot_c() is True
assert client.connect() is True
client.disconnect()
def test_cached_mode_respected(self, ftp_client_factory):
"""Cached mode is used on subsequent connections."""
BambuFTPClient.cache_mode("127.0.0.1", "prot_c")
client = ftp_client_factory(printer_model="A1")
assert client._should_use_prot_c() is True
assert client.connect() is True
client.disconnect()
# ---------------------------------------------------------------------------
# TestListFiles
# ---------------------------------------------------------------------------
class TestListFiles:
"""Tests for directory listing."""
def test_list_empty_directory(self, ftp_client_factory):
"""Listing an empty directory returns empty list."""
client = ftp_client_factory()
client.connect()
files = client.list_files("/cache")
assert files == []
client.disconnect()
def test_list_directory_with_files(self, ftp_client_factory, ftp_server):
"""Files in directory are listed correctly."""
ftp_server.add_file("cache/test.3mf", b"x" * 1024)
ftp_server.add_file("cache/test2.gcode", b"y" * 512)
client = ftp_client_factory()
client.connect()
files = client.list_files("/cache")
names = {f["name"] for f in files}
assert "test.3mf" in names
assert "test2.gcode" in names
client.disconnect()
def test_directories_marked(self, ftp_client_factory, ftp_server):
"""Subdirectories are identified with is_directory=True."""
ftp_server.add_directory("model/subdir")
client = ftp_client_factory()
client.connect()
files = client.list_files("/model")
dirs = [f for f in files if f["is_directory"]]
assert len(dirs) >= 1
assert dirs[0]["name"] == "subdir"
client.disconnect()
def test_nonexistent_path_returns_empty(self, ftp_client_factory):
"""Listing a nonexistent path returns empty list."""
client = ftp_client_factory()
client.connect()
files = client.list_files("/nonexistent/path")
assert files == []
client.disconnect()
def test_file_sizes_and_paths(self, ftp_client_factory, ftp_server):
"""File sizes and full paths are parsed correctly."""
ftp_server.add_file("cache/sized.bin", b"a" * 2048)
client = ftp_client_factory()
client.connect()
files = client.list_files("/cache")
sized = [f for f in files if f["name"] == "sized.bin"]
assert len(sized) == 1
assert sized[0]["size"] == 2048
assert sized[0]["path"] == "/cache/sized.bin"
client.disconnect()
# ---------------------------------------------------------------------------
# TestDownload
# ---------------------------------------------------------------------------
class TestDownload:
"""Tests for file download operations."""
def test_download_file_returns_bytes(self, ftp_client_factory, ftp_server):
"""download_file() returns file content as bytes."""
content = b"Hello FTP World!"
ftp_server.add_file("cache/hello.txt", content)
client = ftp_client_factory()
client.connect()
result = client.download_file("/cache/hello.txt")
assert result == content
client.disconnect()
def test_download_file_missing(self, ftp_client_factory):
"""download_file() returns None for missing file."""
client = ftp_client_factory()
client.connect()
result = client.download_file("/cache/does_not_exist.txt")
assert result is None
client.disconnect()
def test_download_to_file_writes_to_disk(self, ftp_client_factory, ftp_server, tmp_path):
"""download_to_file() writes content to local filesystem."""
content = b"Downloaded content"
ftp_server.add_file("cache/dl.bin", content)
local = tmp_path / "output" / "dl.bin"
client = ftp_client_factory()
client.connect()
result = client.download_to_file("/cache/dl.bin", local)
assert result is True
assert local.read_bytes() == content
client.disconnect()
def test_download_to_file_creates_parent_dirs(self, ftp_client_factory, ftp_server, tmp_path):
"""download_to_file() creates parent directories automatically."""
ftp_server.add_file("cache/nested.txt", b"nested content")
local = tmp_path / "deep" / "nested" / "path" / "nested.txt"
client = ftp_client_factory()
client.connect()
result = client.download_to_file("/cache/nested.txt", local)
assert result is True
assert local.exists()
client.disconnect()
def test_zero_byte_download_returns_false(self, ftp_client_factory, ftp_server, tmp_path):
"""0-byte download returns False and cleans up (regression test)."""
ftp_server.add_file("cache/empty.bin", b"")
local = tmp_path / "empty.bin"
client = ftp_client_factory()
client.connect()
result = client.download_to_file("/cache/empty.bin", local)
assert result is False
assert not local.exists()
client.disconnect()
def test_download_to_file_missing_returns_false(self, ftp_client_factory, tmp_path):
"""Missing file returns False."""
local = tmp_path / "missing.bin"
client = ftp_client_factory()
client.connect()
result = client.download_to_file("/cache/no_such_file.bin", local)
assert result is False
client.disconnect()
def test_download_large_file(self, ftp_client_factory, ftp_server):
"""Large file download (>1MB) works correctly."""
large_content = b"X" * (1024 * 1024 + 500) # ~1MB + 500 bytes
ftp_server.add_file("cache/large.bin", large_content)
client = ftp_client_factory()
client.connect()
result = client.download_file("/cache/large.bin")
assert result == large_content
client.disconnect()
def test_download_not_connected(self):
"""download_file() returns None when not connected."""
client = BambuFTPClient("127.0.0.1", "12345678")
assert client.download_file("/cache/test.bin") is None
# ---------------------------------------------------------------------------
# TestUpload
# ---------------------------------------------------------------------------
class TestUpload:
"""Tests for file upload operations."""
def test_upload_success(self, ftp_client_factory, ftp_server, tmp_path):
"""Successful upload via transfercmd (not storbinary)."""
content = b"Upload test content"
local = tmp_path / "upload.3mf"
local.write_bytes(content)
client = ftp_client_factory()
client.connect()
result = client.upload_file(local, "/cache/upload.3mf")
assert result is True
client.disconnect()
# Verify via fresh connection (upload_file skips voidresp()
# so the original session can't be reused for download)
time.sleep(_UPLOAD_FLUSH_DELAY)
client2 = ftp_client_factory()
client2.connect()
downloaded = client2.download_file("/cache/upload.3mf")
assert downloaded == content
client2.disconnect()
def test_upload_progress_callback(self, ftp_client_factory, ftp_server, tmp_path):
"""Progress callback receives updates during upload."""
content = b"P" * 2048
local = tmp_path / "progress.bin"
local.write_bytes(content)
progress_calls = []
def on_progress(uploaded, total):
progress_calls.append((uploaded, total))
client = ftp_client_factory()
client.connect()
client.upload_file(local, "/cache/progress.bin", on_progress)
assert len(progress_calls) >= 1
# Last call should report full file uploaded
assert progress_calls[-1][0] == len(content)
assert progress_calls[-1][1] == len(content)
client.disconnect()
def test_upload_not_connected(self, tmp_path):
"""Upload when not connected returns False."""
local = tmp_path / "test.bin"
local.write_bytes(b"data")
client = BambuFTPClient("127.0.0.1", "12345678")
assert client.upload_file(local, "/cache/test.bin") is False
def test_upload_553_no_sd_card(self, ftp_client_factory, ftp_server, tmp_path):
"""553 error (no SD card) returns False."""
ftp_server.inject_failure("STOR", 553, "Could not create file.")
local = tmp_path / "test.bin"
local.write_bytes(b"data")
client = ftp_client_factory()
client.connect()
result = client.upload_file(local, "/cache/test.bin")
assert result is False
client.disconnect()
def test_upload_550_permission_denied(self, ftp_client_factory, ftp_server, tmp_path):
"""550 error (permission denied) returns False."""
ftp_server.inject_failure("STOR", 550, "Permission denied.")
local = tmp_path / "test.bin"
local.write_bytes(b"data")
client = ftp_client_factory()
client.connect()
result = client.upload_file(local, "/cache/test.bin")
assert result is False
client.disconnect()
def test_upload_552_storage_full(self, ftp_client_factory, ftp_server, tmp_path):
"""552 error (storage full) returns False."""
ftp_server.inject_failure("STOR", 552, "Storage quota exceeded.")
local = tmp_path / "test.bin"
local.write_bytes(b"data")
client = ftp_client_factory()
client.connect()
result = client.upload_file(local, "/cache/test.bin")
assert result is False
client.disconnect()
def test_upload_bytes_success(self, ftp_client_factory, ftp_server):
"""upload_bytes() writes data to server."""
data = b"Bytes upload content"
client = ftp_client_factory()
client.connect()
result = client.upload_bytes(data, "/cache/bytes.bin")
assert result is True
client.disconnect()
# Verify via fresh connection
time.sleep(_UPLOAD_FLUSH_DELAY)
client2 = ftp_client_factory()
client2.connect()
downloaded = client2.download_file("/cache/bytes.bin")
assert downloaded == data
client2.disconnect()
def test_upload_bytes_failure(self, ftp_client_factory, ftp_server):
"""upload_bytes() returns False on STOR failure."""
ftp_server.inject_failure("STOR", 553, "No space.")
client = ftp_client_factory()
client.connect()
result = client.upload_bytes(b"data", "/cache/fail.bin")
assert result is False
client.disconnect()
def test_upload_large_chunked(self, ftp_client_factory, ftp_server, tmp_path):
"""Large file upload in chunks completes without error.
Uses 2.5MB to trigger multiple chunks with 1MB CHUNK_SIZE.
Content verification skipped because upload_file() doesn't call
voidresp() (for A1 compatibility), so the server may still be
flushing when we check. The upload result=True confirms the
client sent all chunks without error.
"""
content = b"C" * (1024 * 1024 * 2 + 512 * 1024)
local = tmp_path / "large.bin"
local.write_bytes(content)
progress_calls = []
def on_progress(uploaded, total):
progress_calls.append((uploaded, total))
client = ftp_client_factory()
client.connect()
result = client.upload_file(local, "/cache/large.bin", on_progress)
assert result is True
# Verify multiple chunks were sent
assert len(progress_calls) >= 3 # 2.5MB / 1MB = at least 3 chunks
assert progress_calls[-1][0] == len(content)
client.disconnect()
# ---------------------------------------------------------------------------
# TestDelete
# ---------------------------------------------------------------------------
class TestDelete:
"""Tests for file deletion."""
def test_delete_success(self, ftp_client_factory, ftp_server):
"""Successful file deletion."""
ftp_server.add_file("cache/to_delete.bin", b"delete me")
client = ftp_client_factory()
client.connect()
result = client.delete_file("/cache/to_delete.bin")
assert result is True
assert not ftp_server.file_exists("cache/to_delete.bin")
client.disconnect()
def test_delete_not_found(self, ftp_client_factory):
"""Deleting a nonexistent file returns False."""
client = ftp_client_factory()
client.connect()
result = client.delete_file("/cache/no_such_file.bin")
assert result is False
client.disconnect()
def test_delete_not_connected(self):
"""Delete when not connected returns False."""
client = BambuFTPClient("127.0.0.1", "12345678")
assert client.delete_file("/cache/test.bin") is False
# ---------------------------------------------------------------------------
# TestFileSize
# ---------------------------------------------------------------------------
class TestFileSize:
"""Tests for get_file_size."""
def test_file_size_correct(self, ftp_client_factory, ftp_server):
"""Returns correct file size."""
ftp_server.add_file("cache/sized.bin", b"a" * 4096)
client = ftp_client_factory()
client.connect()
size = client.get_file_size("/cache/sized.bin")
assert size == 4096
client.disconnect()
def test_file_size_missing(self, ftp_client_factory):
"""Returns None for missing file."""
client = ftp_client_factory()
client.connect()
size = client.get_file_size("/cache/no_file.bin")
assert size is None
client.disconnect()
def test_file_size_not_connected(self):
"""Returns None when not connected."""
client = BambuFTPClient("127.0.0.1", "12345678")
assert client.get_file_size("/cache/test.bin") is None
# ---------------------------------------------------------------------------
# TestStorageInfo
# ---------------------------------------------------------------------------
class TestStorageInfo:
"""Tests for storage info and diagnostics."""
def test_avbl_parsed(self, ftp_client_factory, ftp_server):
"""AVBL response is parsed for free_bytes."""
ftp_server.set_avbl_bytes(5000000000)
client = ftp_client_factory()
client.connect()
info = client.get_storage_info()
assert info is not None
assert info["free_bytes"] == 5000000000
client.disconnect()
def test_used_bytes_from_scan(self, ftp_client_factory, ftp_server):
"""used_bytes calculated from directory scan."""
ftp_server.add_file("cache/file1.bin", b"a" * 1000)
ftp_server.add_file("cache/file2.bin", b"b" * 2000)
client = ftp_client_factory()
client.connect()
info = client.get_storage_info()
assert info is not None
assert info["used_bytes"] >= 3000 # At least these two files
client.disconnect()
def test_storage_info_not_connected(self):
"""Returns None when not connected."""
client = BambuFTPClient("127.0.0.1", "12345678")
assert client.get_storage_info() is None
def test_diagnose_storage_success(self, ftp_client_factory, ftp_server):
"""diagnose_storage() returns connected=True with working diagnostics."""
client = ftp_client_factory()
client.connect()
diag = client.diagnose_storage()
assert diag["connected"] is True
assert diag["can_list_root"] is True
assert diag["can_list_cache"] is True
assert diag["pwd"] is not None
assert diag["storage_info"] is not None
client.disconnect()
def test_diagnose_storage_not_connected(self):
"""diagnose_storage() reports not connected."""
client = BambuFTPClient("127.0.0.1", "12345678")
diag = client.diagnose_storage()
assert diag["connected"] is False
assert "FTP not connected" in diag["errors"]
# ---------------------------------------------------------------------------
# TestModelSpecificBehavior
# ---------------------------------------------------------------------------
class TestModelSpecificBehavior:
"""Tests for printer model-specific FTP behavior."""
def test_x1c_upload(self, ftp_client_factory, ftp_server, tmp_path):
"""X1C upload with session reuse succeeds."""
content = b"X1C upload data"
local = tmp_path / "x1c.3mf"
local.write_bytes(content)
client = ftp_client_factory(printer_model="X1C")
client.connect()
result = client.upload_file(local, "/cache/x1c.3mf")
assert result is True
client.disconnect()
# Verify via fresh connection
time.sleep(_UPLOAD_FLUSH_DELAY)
client2 = ftp_client_factory(printer_model="X1C")
client2.connect()
downloaded = client2.download_file("/cache/x1c.3mf")
assert downloaded == content
client2.disconnect()
def test_a1_upload_prot_c(self, ftp_client_factory, ftp_server, tmp_path):
"""A1 model upload with prot_c succeeds."""
content = b"A1 upload data"
local = tmp_path / "a1.3mf"
local.write_bytes(content)
client = ftp_client_factory(printer_model="A1", force_prot_c=True)
client.connect()
result = client.upload_file(local, "/cache/a1.3mf")
assert result is True
client.disconnect()
# Verify via fresh connection
time.sleep(_UPLOAD_FLUSH_DELAY)
client2 = ftp_client_factory(printer_model="A1", force_prot_c=True)
client2.connect()
downloaded = client2.download_file("/cache/a1.3mf")
assert downloaded == content
client2.disconnect()
def test_a1_mini_upload(self, ftp_client_factory, ftp_server, tmp_path):
"""A1 Mini model upload succeeds."""
content = b"A1 Mini data"
local = tmp_path / "a1mini.3mf"
local.write_bytes(content)
client = ftp_client_factory(printer_model="A1 Mini", force_prot_c=True)
client.connect()
result = client.upload_file(local, "/cache/a1mini.3mf")
assert result is True
client.disconnect()
def test_p1s_upload(self, ftp_client_factory, ftp_server, tmp_path):
"""P1S model upload with session reuse succeeds."""
content = b"P1S upload data"
local = tmp_path / "p1s.3mf"
local.write_bytes(content)
client = ftp_client_factory(printer_model="P1S")
client.connect()
result = client.upload_file(local, "/cache/p1s.3mf")
assert result is True
client.disconnect()
def test_unknown_model_defaults_prot_p(self, ftp_client_factory):
"""Unknown model defaults to prot_p."""
client = ftp_client_factory(printer_model="FuturePrinter3000")
assert client._is_a1_model() is False
assert client._should_use_prot_c() is False
assert client.connect() is True
client.disconnect()
def test_mode_cache_persists_and_clears(self, ftp_client_factory):
"""Mode cache works within a test and clears between tests."""
# Cache should be empty at start (autouse fixture clears it)
assert BambuFTPClient._mode_cache == {}
# Connect and cache a mode
BambuFTPClient.cache_mode("127.0.0.1", "prot_p")
assert BambuFTPClient._mode_cache["127.0.0.1"] == "prot_p"
# New client for same IP uses cached mode
client = ftp_client_factory(printer_model="A1")
assert client._get_cached_mode() == "prot_p"
assert client._should_use_prot_c() is False
client.disconnect()
# ---------------------------------------------------------------------------
# TestAsyncWrappers
# ---------------------------------------------------------------------------
class TestAsyncWrappers:
"""Tests for async wrapper functions using patch_ftp_port fixture."""
@pytest.mark.asyncio
async def test_upload_file_async_success(self, patch_ftp_port, tmp_path):
"""upload_file_async succeeds for X1C."""
content = b"async upload"
local = tmp_path / "async_up.3mf"
local.write_bytes(content)
result = await upload_file_async(
"127.0.0.1",
"12345678",
local,
"/cache/async_up.3mf",
timeout=30.0,
printer_model="X1C",
)
assert result is True
@pytest.mark.asyncio
async def test_upload_file_async_a1_fallback(self, patch_ftp_port, tmp_path):
"""upload_file_async tries prot_p then falls back to prot_c for A1."""
content = b"a1 async upload"
local = tmp_path / "a1_async.3mf"
local.write_bytes(content)
# For A1 models, if prot_p succeeds we get True.
# If prot_p fails, it tries prot_c. Either way should succeed
# against our mock server which accepts both.
result = await upload_file_async(
"127.0.0.1",
"12345678",
local,
"/cache/a1_async.3mf",
timeout=30.0,
printer_model="A1",
)
assert result is True
@pytest.mark.asyncio
async def test_download_file_async_success(self, patch_ftp_port, tmp_path):
"""download_file_async succeeds."""
server = patch_ftp_port
content = b"async download content"
server.add_file("cache/async_dl.bin", content)
local = tmp_path / "async_dl.bin"
result = await download_file_async(
"127.0.0.1",
"12345678",
"/cache/async_dl.bin",
local,
timeout=30.0,
printer_model="X1C",
)
assert result is True
assert local.read_bytes() == content
@pytest.mark.asyncio
async def test_download_file_async_a1_fallback(self, patch_ftp_port, tmp_path):
"""download_file_async falls back for A1 models."""
server = patch_ftp_port
server.add_file("cache/a1_dl.bin", b"a1 data")
local = tmp_path / "a1_dl.bin"
result = await download_file_async(
"127.0.0.1",
"12345678",
"/cache/a1_dl.bin",
local,
timeout=30.0,
printer_model="A1",
)
assert result is True
@pytest.mark.asyncio
async def test_download_file_try_paths_first_succeeds(self, patch_ftp_port, tmp_path):
"""download_file_try_paths_async succeeds on first path."""
server = patch_ftp_port
server.add_file("cache/try1.bin", b"first path")
local = tmp_path / "try.bin"
result = await download_file_try_paths_async(
"127.0.0.1",
"12345678",
["/cache/try1.bin", "/cache/try2.bin"],
local,
printer_model="X1C",
)
assert result is True
assert local.read_bytes() == b"first path"
@pytest.mark.asyncio
async def test_download_file_try_paths_fallback(self, patch_ftp_port, tmp_path):
"""download_file_try_paths_async falls back to second path."""
server = patch_ftp_port
server.add_file("cache/second.bin", b"second path")
local = tmp_path / "fallback.bin"
result = await download_file_try_paths_async(
"127.0.0.1",
"12345678",
["/cache/missing.bin", "/cache/second.bin"],
local,
printer_model="X1C",
)
assert result is True
assert local.read_bytes() == b"second path"
@pytest.mark.asyncio
async def test_list_files_async_success(self, patch_ftp_port):
"""list_files_async returns file list."""
server = patch_ftp_port
server.add_file("cache/listed.bin", b"data")
result = await list_files_async(
"127.0.0.1",
"12345678",
"/cache",
timeout=30.0,
printer_model="X1C",
)
names = {f["name"] for f in result}
assert "listed.bin" in names
@pytest.mark.asyncio
async def test_delete_file_async_success(self, patch_ftp_port):
"""delete_file_async deletes a file."""
server = patch_ftp_port
server.add_file("cache/to_async_del.bin", b"delete me")
result = await delete_file_async(
"127.0.0.1",
"12345678",
"/cache/to_async_del.bin",
printer_model="X1C",
)
assert result is True
assert not server.file_exists("cache/to_async_del.bin")
# ---------------------------------------------------------------------------
# TestFailureScenarios
# ---------------------------------------------------------------------------
class TestFailureScenarios:
"""Regression tests for known FTP failure modes."""
def test_550_caught_by_broad_except(self, ftp_client_factory, ftp_server, tmp_path):
"""550 error_perm is caught by (OSError, ftplib.Error) handler.
Regression: error_perm is a subclass of ftplib.Error, so the
broad except clause in upload_file catches it correctly.
"""
ftp_server.inject_failure("STOR", 550, "Permission denied.")
local = tmp_path / "test.bin"
local.write_bytes(b"data")
client = ftp_client_factory()
client.connect()
result = client.upload_file(local, "/cache/test.bin")
assert result is False
client.disconnect()
def test_zero_byte_download_detected(self, ftp_client_factory, ftp_server, tmp_path):
"""0-byte download is detected and file is cleaned up.
Regression: Prior to fix, 0-byte downloads were reported as success.
"""
ftp_server.add_file("cache/zero.bin", b"")
local = tmp_path / "zero.bin"
client = ftp_client_factory()
client.connect()
result = client.download_to_file("/cache/zero.bin", local)
assert result is False
assert not local.exists()
client.disconnect()
def test_connection_refused_handled(self):
"""Connection refused is handled gracefully."""
client = BambuFTPClient("127.0.0.1", "12345678", timeout=2.0)
client.FTP_PORT = 1 # Almost certainly not listening
assert client.connect() is False
def test_auth_failure_530(self, ftp_client_factory, ftp_server):
"""530 authentication failure returns False."""
ftp_server.inject_failure("PASS", 530, "Login incorrect.")
client = ftp_client_factory()
result = client.connect()
assert result is False
def test_retr_550_handled(self, ftp_client_factory, ftp_server):
"""RETR 550 (file not found) returns None."""
ftp_server.inject_failure("RETR", 550, "File not found.")
ftp_server.add_file("cache/exists.bin", b"data")
client = ftp_client_factory()
client.connect()
result = client.download_file("/cache/exists.bin")
assert result is None
client.disconnect()
def test_cwd_550_handled(self, ftp_client_factory, ftp_server):
"""CWD 550 is handled in list_files."""
ftp_server.inject_failure("CWD", 550, "Directory not found.")
client = ftp_client_factory()
client.connect()
result = client.list_files("/nonexistent")
assert result == []
client.disconnect()
def test_stor_553_handled(self, ftp_client_factory, ftp_server, tmp_path):
"""STOR 553 (no SD card) handled gracefully."""
ftp_server.inject_failure("STOR", 553, "Could not create file.")
local = tmp_path / "test.bin"
local.write_bytes(b"test")
client = ftp_client_factory()
client.connect()
result = client.upload_file(local, "/cache/test.bin")
assert result is False
client.disconnect()
def test_diagnose_storage_cwd_failure_doesnt_propagate(self, ftp_client_factory, ftp_server):
"""diagnose_storage CWD failure doesn't crash the whole operation.
Regression: diagnose_storage() was called in the upload path and
a CWD failure would propagate and crash the upload.
"""
ftp_server.inject_failure("CWD", 550, "No such directory.", count=2)
client = ftp_client_factory()
client.connect()
diag = client.diagnose_storage()
# Should still return results (with errors noted)
assert diag["connected"] is True
assert len(diag["errors"]) > 0
client.disconnect()
def test_failure_injection_count_decrements(self, ftp_client_factory, ftp_server):
"""Failure injection with count decrements and eventually succeeds."""
ftp_server.add_file("cache/retry.bin", b"data after retry")
ftp_server.inject_failure("RETR", 550, "Temporary error.", count=1)
client = ftp_client_factory()
client.connect()
# First attempt fails
result1 = client.download_file("/cache/retry.bin")
assert result1 is None
# Second attempt succeeds (failure count exhausted)
result2 = client.download_file("/cache/retry.bin")
assert result2 == b"data after retry"
client.disconnect()
@@ -507,7 +507,7 @@ class TestSSDPProxy:
"""Verify SSDP Location header is rewritten to remote interface IP."""
original_packet = b"NOTIFY * HTTP/1.1\r\nLocation: 192.168.1.50\r\nDevName.bambu.com: TestPrinter\r\n\r\n"
rewritten = ssdp_proxy._rewrite_ssdp_location(original_packet)
rewritten = ssdp_proxy._rewrite_ssdp(original_packet)
# Location should be changed to remote interface IP
assert b"Location: 10.0.0.100" in rewritten
@@ -519,7 +519,7 @@ class TestSSDPProxy:
"""Verify SSDP Location rewrite is case insensitive."""
original_packet = b"NOTIFY * HTTP/1.1\r\nlocation: 192.168.1.50\r\n\r\n"
rewritten = ssdp_proxy._rewrite_ssdp_location(original_packet)
rewritten = ssdp_proxy._rewrite_ssdp(original_packet)
assert b"10.0.0.100" in rewritten
@@ -527,10 +527,10 @@ class TestSSDPProxy:
"""Verify packet without Location header is returned unchanged."""
original_packet = b"NOTIFY * HTTP/1.1\r\nDevName.bambu.com: Test\r\n\r\n"
rewritten = ssdp_proxy._rewrite_ssdp_location(original_packet)
rewritten = ssdp_proxy._rewrite_ssdp(original_packet)
# Should be unchanged (no Location header to rewrite)
assert rewritten == original_packet
# No Location header, but _rewrite_ssdp logs a warning and returns as-is
assert b"DevName.bambu.com: Test" in rewritten
def test_parse_ssdp_message(self, ssdp_proxy):
"""Verify SSDP message parsing extracts headers."""
+12
View File
@@ -10,6 +10,11 @@ services:
# Override with: PUID=$(id -u) PGID=$(id -g) docker compose up -d
user: "${PUID:-1000}:${PGID:-1000}"
#
# Proxy mode: allow binding to port 990 (FTP) as non-root user.
# Without this, the FTP proxy silently fails and sending prints won't work.
cap_add:
- NET_BIND_SERVICE
#
# LINUX: Use host mode for printer discovery and camera streaming
network_mode: host
#
@@ -18,6 +23,9 @@ services:
# Note: Printer discovery won't work - add printers manually by IP.
#ports:
# - "${PORT:-8000}:8000"
# - "8883:8883" # Virtual printer MQTT
# - "9990:9990" # Virtual printer FTP control
# - "50000-50100:50000-50100" # Virtual printer FTP passive data
volumes:
- bambuddy_data:/app/data
- bambuddy_logs:/app/logs
@@ -30,6 +38,10 @@ services:
# Port BamBuddy runs on (default: 8000)
# Usage: PORT=8080 docker compose up -d
- PORT=${PORT:-8000}
# Virtual printer: Set to the Docker host's IP when using bridge mode (ports:).
# Required for FTP passive mode to work behind NAT.
# Example: VIRTUAL_PRINTER_PASV_ADDRESS=192.168.1.100
#- VIRTUAL_PRINTER_PASV_ADDRESS=
restart: unless-stopped
volumes:
@@ -11,6 +11,7 @@ import { MemoryRouter, Route, Routes } from 'react-router-dom';
import { QueryClient, QueryClientProvider } from '@tanstack/react-query';
import { ThemeProvider } from '../../contexts/ThemeContext';
import { ToastProvider } from '../../contexts/ToastContext';
import { AuthProvider } from '../../contexts/AuthContext';
import { I18nextProvider } from 'react-i18next';
import i18n from '../../i18n';
@@ -44,11 +45,13 @@ function renderCameraPage(printerId: number) {
<I18nextProvider i18n={i18n}>
<MemoryRouter initialEntries={[`/cameras/${printerId}`]}>
<ThemeProvider>
<ToastProvider>
<Routes>
<Route path="/cameras/:printerId" element={<CameraPage />} />
</Routes>
</ToastProvider>
<AuthProvider>
<ToastProvider>
<Routes>
<Route path="/cameras/:printerId" element={<CameraPage />} />
</Routes>
</ToastProvider>
</AuthProvider>
</ThemeProvider>
</MemoryRouter>
</I18nextProvider>
@@ -1,7 +1,12 @@
import { useState, useEffect, useRef, useCallback } from 'react';
import { useQuery } from '@tanstack/react-query';
import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query';
import { useTranslation } from 'react-i18next';
import { X, RefreshCw, AlertTriangle, Maximize2, Minimize2, GripVertical, WifiOff, ZoomIn, ZoomOut, Fullscreen, Minimize } from 'lucide-react';
import { api } from '../api/client';
import { api, getAuthToken } from '../api/client';
import { useToast } from '../contexts/ToastContext';
import { useAuth } from '../contexts/AuthContext';
import { ChamberLight } from './icons/ChamberLight';
import { SkipObjectsModal, SkipObjectsIcon } from './SkipObjectsModal';
interface EmbeddedCameraViewerProps {
printerId: number;
@@ -31,6 +36,11 @@ const DEFAULT_STATE: CameraState = {
};
export function EmbeddedCameraViewer({ printerId, printerName, viewerIndex = 0, onClose }: EmbeddedCameraViewerProps) {
const { t } = useTranslation();
const queryClient = useQueryClient();
const { showToast } = useToast();
const { hasPermission } = useAuth();
// Printer-specific storage key
const storageKey = `${STORAGE_KEY_PREFIX}${printerId}`;
@@ -87,6 +97,8 @@ export function EmbeddedCameraViewer({ printerId, printerName, viewerIndex = 0,
const countdownIntervalRef = useRef<NodeJS.Timeout | null>(null);
const stallCheckIntervalRef = useRef<NodeJS.Timeout | null>(null);
const [showSkipObjectsModal, setShowSkipObjectsModal] = useState(false);
// Fetch printer info
const { data: printer } = useQuery({
queryKey: ['printer', printerId],
@@ -94,6 +106,39 @@ export function EmbeddedCameraViewer({ printerId, printerName, viewerIndex = 0,
enabled: printerId > 0,
});
// Fetch printer status for light toggle and skip objects
const { data: status } = useQuery({
queryKey: ['printerStatus', printerId],
queryFn: () => api.getPrinterStatus(printerId),
refetchInterval: 30000,
enabled: printerId > 0,
});
// Chamber light mutation with optimistic update
const chamberLightMutation = useMutation({
mutationFn: (on: boolean) => api.setChamberLight(printerId, on),
onMutate: async (on) => {
await queryClient.cancelQueries({ queryKey: ['printerStatus', printerId] });
const previousStatus = queryClient.getQueryData(['printerStatus', printerId]);
queryClient.setQueryData(['printerStatus', printerId], (old: typeof status) => ({
...old,
chamber_light: on,
}));
return { previousStatus };
},
onSuccess: (_, on) => {
showToast(`Chamber light ${on ? 'on' : 'off'}`);
},
onError: (error: Error, _, context) => {
if (context?.previousStatus) {
queryClient.setQueryData(['printerStatus', printerId], context.previousStatus);
}
showToast(error.message || t('printers.toast.failedToControlChamberLight'), 'error');
},
});
const isPrintingWithObjects = (status?.state === 'RUNNING' || status?.state === 'PAUSE' || status?.state === 'PAUSED') && (status?.printable_objects_count ?? 0) >= 2;
// Save state to localStorage (printer-specific)
useEffect(() => {
const saveTimeout = setTimeout(() => {
@@ -111,7 +156,10 @@ export function EmbeddedCameraViewer({ printerId, printerName, viewerIndex = 0,
const sendStopOnce = () => {
if (printerId > 0 && !stopSentRef.current) {
stopSentRef.current = true;
navigator.sendBeacon(stopUrl);
const headers: Record<string, string> = {};
const token = getAuthToken();
if (token) headers['Authorization'] = `Bearer ${token}`;
fetch(stopUrl, { method: 'POST', keepalive: true, headers }).catch(() => {});
}
};
@@ -403,7 +451,10 @@ export function EmbeddedCameraViewer({ printerId, printerName, viewerIndex = 0,
if (reconnectTimerRef.current) clearTimeout(reconnectTimerRef.current);
if (countdownIntervalRef.current) clearInterval(countdownIntervalRef.current);
fetch(`/api/v1/printers/${printerId}/camera/stop`).catch(() => {});
const stopHeaders: Record<string, string> = {};
const stopToken = getAuthToken();
if (stopToken) stopHeaders['Authorization'] = `Bearer ${stopToken}`;
fetch(`/api/v1/printers/${printerId}/camera/stop`, { method: 'POST', headers: stopHeaders }).catch(() => {});
if (imgRef.current) imgRef.current.src = '';
setTimeout(() => setImageKey(Date.now()), 100);
@@ -482,6 +533,28 @@ export function EmbeddedCameraViewer({ printerId, printerName, viewerIndex = 0,
<span className="truncate">{printer?.name || printerName}</span>
</div>
<div className="flex items-center gap-1 no-drag">
<button
onClick={() => chamberLightMutation.mutate(!status?.chamber_light)}
disabled={!status?.connected || chamberLightMutation.isPending || !hasPermission('printers:control')}
className={`p-1 rounded disabled:opacity-50 ${status?.chamber_light ? 'bg-yellow-500/20 hover:bg-yellow-500/30' : 'hover:bg-bambu-dark-tertiary'}`}
title={!hasPermission('printers:control') ? t('printers.permission.noControl') : t('camera.chamberLight')}
>
<ChamberLight on={status?.chamber_light ?? false} className="w-3.5 h-3.5" />
</button>
<button
onClick={() => setShowSkipObjectsModal(true)}
disabled={!isPrintingWithObjects || !hasPermission('printers:control')}
className={`p-1 rounded disabled:opacity-50 ${isPrintingWithObjects && hasPermission('printers:control') ? 'hover:bg-bambu-dark-tertiary' : ''}`}
title={
!hasPermission('printers:control')
? t('printers.permission.noControl')
: !isPrintingWithObjects
? t('printers.skipObjects.onlyWhilePrinting')
: t('printers.skipObjects.tooltip')
}
>
<SkipObjectsIcon className="w-3.5 h-3.5 text-bambu-gray" />
</button>
<button
onClick={refresh}
disabled={streamLoading || isReconnecting}
@@ -625,6 +698,12 @@ export function EmbeddedCameraViewer({ printerId, printerName, viewerIndex = 0,
)}
</div>
)}
{/* Skip Objects Modal */}
<SkipObjectsModal
printerId={printerId}
isOpen={showSkipObjectsModal}
onClose={() => setShowSkipObjectsModal(false)}
/>
</div>
);
}
+5 -8
View File
@@ -61,10 +61,9 @@ export function FilamentTrends({ archives, currency = '$' }: FilamentTrendsProps
const key = `${date.getFullYear()}-${String(date.getMonth() + 1).padStart(2, '0')}-${String(date.getDate()).padStart(2, '0')}`;
const existing = dataMap.get(key) || { date: key, filament: 0, cost: 0, prints: 0 };
const qty = archive.quantity || 1;
existing.filament += (archive.filament_used_grams || 0) * qty;
existing.filament += archive.filament_used_grams || 0;
existing.cost += archive.cost || 0;
existing.prints += qty;
existing.prints += archive.quantity || 1;
dataMap.set(key, existing);
});
@@ -90,10 +89,9 @@ export function FilamentTrends({ archives, currency = '$' }: FilamentTrendsProps
const key = `${weekStart.getFullYear()}-${String(weekStart.getMonth() + 1).padStart(2, '0')}-${String(weekStart.getDate()).padStart(2, '0')}`;
const existing = dataMap.get(key) || { week: key, filament: 0, cost: 0, prints: 0 };
const qty = archive.quantity || 1;
existing.filament += (archive.filament_used_grams || 0) * qty;
existing.filament += archive.filament_used_grams || 0;
existing.cost += archive.cost || 0;
existing.prints += qty;
existing.prints += archive.quantity || 1;
dataMap.set(key, existing);
});
@@ -112,11 +110,10 @@ export function FilamentTrends({ archives, currency = '$' }: FilamentTrendsProps
filteredArchives.forEach(archive => {
const type = archive.filament_type || 'Unknown';
const qty = archive.quantity || 1;
// Handle multiple types (e.g., "PLA, PETG")
const types = type.split(', ');
types.forEach(t => {
const grams = ((archive.filament_used_grams || 0) * qty) / types.length;
const grams = (archive.filament_used_grams || 0) / types.length;
dataMap.set(t, (dataMap.get(t) || 0) + grams);
});
});
@@ -0,0 +1,271 @@
import { useQuery, useMutation } from '@tanstack/react-query';
import { useTranslation } from 'react-i18next';
import { X, Loader2, Monitor, AlertCircle, Box } from 'lucide-react';
import { api } from '../api/client';
import { useToast } from '../contexts/ToastContext';
import { useAuth } from '../contexts/AuthContext';
// Custom Skip Objects icon - arrow jumping over boxes
export const SkipObjectsIcon = ({ className }: { className?: string }) => (
<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="2" strokeLinecap="round" strokeLinejoin="round" className={className}>
{/* Three boxes at the bottom */}
<rect x="2" y="15" width="5" height="5" rx="0.5" />
<rect x="9.5" y="15" width="5" height="5" rx="0.5" fill="currentColor" opacity="0.3" />
<rect x="17" y="15" width="5" height="5" rx="0.5" />
{/* Curved arrow jumping over first box */}
<path d="M4 12 C4 6, 14 6, 14 12" />
<polyline points="12,10 14,12 12,14" />
</svg>
);
interface SkipObjectsModalProps {
printerId: number;
isOpen: boolean;
onClose: () => void;
}
export function SkipObjectsModal({ printerId, isOpen, onClose }: SkipObjectsModalProps) {
const { t } = useTranslation();
const { showToast } = useToast();
const { hasPermission } = useAuth();
const { data: status } = useQuery({
queryKey: ['printerStatus', printerId],
queryFn: () => api.getPrinterStatus(printerId),
refetchInterval: 30000,
enabled: isOpen,
});
const { data: objectsData, refetch: refetchObjects } = useQuery({
queryKey: ['printableObjects', printerId],
queryFn: () => api.getPrintableObjects(printerId),
enabled: isOpen,
refetchInterval: isOpen ? 5000 : false,
});
const skipObjectsMutation = useMutation({
mutationFn: (objectIds: number[]) => api.skipObjects(printerId, objectIds),
onSuccess: (data) => {
showToast(data.message || t('printers.skipObjects.objectsSkipped'));
refetchObjects();
},
onError: (error: Error) => showToast(error.message || t('printers.toast.failedToSkipObjects'), 'error'),
});
if (!isOpen) return null;
return (
<div
className="fixed inset-0 z-50 flex items-center justify-center"
onClick={onClose}
onKeyDown={(e) => e.key === 'Escape' && onClose()}
tabIndex={-1}
ref={(el) => el?.focus()}
>
{/* Backdrop */}
<div className="absolute inset-0 bg-black/50 z-0" />
{/* Modal */}
<div
className="relative z-10 bg-white dark:bg-bambu-dark border border-gray-200 dark:border-bambu-dark-tertiary rounded-xl shadow-2xl w-[560px] max-h-[85vh] flex flex-col overflow-hidden"
onClick={(e) => e.stopPropagation()}
>
{/* Header */}
<div className="flex items-center justify-between px-4 py-3 border-b border-gray-200 dark:border-bambu-dark-tertiary bg-gray-50 dark:bg-bambu-dark">
<div className="flex items-center gap-2">
<SkipObjectsIcon className="w-4 h-4 text-bambu-green" />
<span className="text-sm font-medium text-gray-900 dark:text-white">{t('printers.skipObjects.title')}</span>
</div>
<button
onClick={onClose}
className="p-1 text-gray-500 dark:text-bambu-gray hover:text-gray-900 dark:hover:text-white rounded transition-colors"
>
<X className="w-4 h-4" />
</button>
</div>
{!objectsData ? (
<div className="flex items-center justify-center py-12">
<Loader2 className="w-5 h-5 animate-spin text-bambu-gray" />
</div>
) : objectsData.objects.length === 0 ? (
<div className="text-center py-8 px-4 text-bambu-gray">
<p className="text-sm">{t('printers.noObjectsFound')}</p>
<p className="text-xs mt-1 opacity-70">{t('printers.objectsLoadedOnPrintStart')}</p>
</div>
) : (
<div className="flex flex-col overflow-hidden">
{/* Info Banner */}
<div className="flex items-center gap-3 px-4 py-2.5 bg-blue-50 dark:bg-blue-500/10 border-b border-gray-200 dark:border-bambu-dark-tertiary">
<div className="flex-shrink-0 w-8 h-8 rounded-lg bg-blue-100 dark:bg-blue-500/20 flex items-center justify-center">
<Monitor className="w-4 h-4 text-blue-500 dark:text-blue-400" />
</div>
<div className="flex-1 min-w-0">
<p className="text-xs text-blue-600 dark:text-blue-300">{t('printers.skipObjects.matchIdsInfo')}</p>
<p className="text-[10px] text-blue-500/70 dark:text-blue-300/60">{t('printers.skipObjects.printerShowsIds')}</p>
</div>
<div className="flex-shrink-0 text-xs text-gray-500 dark:text-bambu-gray">
{objectsData.skipped_count}/{objectsData.total} {t('printers.skipObjects.skipped')}
</div>
</div>
{/* Layer Warning */}
{(status?.layer_num ?? 0) <= 1 && (
<div className="flex items-center gap-2 px-4 py-2 bg-amber-50 dark:bg-amber-500/10 border-b border-gray-200 dark:border-bambu-dark-tertiary">
<AlertCircle className="w-4 h-4 text-amber-500 dark:text-amber-400 flex-shrink-0" />
<p className="text-xs text-amber-600 dark:text-amber-400">
{t('printers.skipObjects.waitForLayer', { layer: status?.layer_num ?? 0 })}
</p>
</div>
)}
{/* Content: Image + List side by side */}
<div className="flex flex-1 overflow-hidden">
{/* Left: Preview Image with object markers */}
<div className="w-52 flex-shrink-0 p-4 border-r border-gray-200 dark:border-bambu-dark-tertiary bg-gray-50 dark:bg-bambu-dark-secondary overflow-y-auto">
<div className="relative">
{status?.cover_url ? (
<img
src={`${status.cover_url}?view=top`}
alt={t('printers.printPreview')}
className="w-full aspect-square object-contain rounded-lg bg-gray-900 dark:bg-gray-900 border border-gray-300 dark:border-gray-600"
/>
) : (
<div className="w-full aspect-square rounded-lg bg-gray-100 dark:bg-bambu-dark flex items-center justify-center">
<Box className="w-8 h-8 text-gray-300 dark:text-bambu-gray/30" />
</div>
)}
{/* Object ID markers overlay - positioned based on object data */}
{objectsData.objects.length > 0 && (
<div className="absolute inset-0 pointer-events-none">
{objectsData.objects.map((obj, idx) => {
let x: number, y: number;
// Use position data if available, otherwise fall back to grid
if (obj.x != null && obj.y != null && objectsData.bbox_all) {
// bbox_all defines the visible area in the top_N.png image
// Format: [x_min, y_min, x_max, y_max] in mm
const [xMin, yMin, xMax, yMax] = objectsData.bbox_all;
const bboxWidth = xMax - xMin;
const bboxHeight = yMax - yMin;
// The image shows bbox_all area with some padding (~5-10%)
const padding = 8;
const contentArea = 100 - (padding * 2);
// Map object position to image percentage
x = padding + ((obj.x - xMin) / bboxWidth) * contentArea;
// Y axis: image Y increases downward, but 3D Y increases toward back
y = padding + ((yMax - obj.y) / bboxHeight) * contentArea;
// Clamp to valid range
x = Math.max(5, Math.min(95, x));
y = Math.max(5, Math.min(95, y));
} else if (obj.x != null && obj.y != null) {
// Fallback: use full build plate (256mm)
const buildPlate = 256;
x = (obj.x / buildPlate) * 100;
y = 100 - (obj.y / buildPlate) * 100;
x = Math.max(5, Math.min(95, x));
y = Math.max(5, Math.min(95, y));
} else {
// Fallback: arrange in a grid pattern over the build plate area
const cols = Math.ceil(Math.sqrt(objectsData.objects.length));
const row = Math.floor(idx / cols);
const col = idx % cols;
const rows = Math.ceil(objectsData.objects.length / cols);
x = 15 + (col * (70 / cols)) + (35 / cols);
y = 15 + (row * (70 / rows)) + (35 / rows);
}
return (
<div
key={obj.id}
className={`absolute flex items-center justify-center w-6 h-6 rounded-full text-[10px] font-bold shadow-lg ${
obj.skipped
? 'bg-red-500 text-white line-through'
: 'bg-bambu-green text-black'
}`}
style={{
left: `${x}%`,
top: `${y}%`,
transform: 'translate(-50%, -50%)'
}}
title={obj.name}
>
{obj.id}
</div>
);
})}
</div>
)}
{/* Object count overlay */}
<div className="absolute bottom-2 right-2 px-2 py-1 bg-white/90 dark:bg-black/80 rounded text-[10px] text-gray-700 dark:text-white shadow-sm">
{t('printers.skipObjects.activeCount', { count: objectsData.objects.filter(o => !o.skipped).length })}
</div>
</div>
</div>
{/* Right: Object List with prominent IDs */}
<div className="flex-1 min-w-0 overflow-y-auto">
{objectsData.objects.map((obj) => (
<div
key={obj.id}
className={`
flex items-center gap-3 px-4 py-3 border-b border-gray-200 dark:border-bambu-dark-tertiary/50 last:border-0
${obj.skipped ? 'bg-red-50 dark:bg-red-500/10' : 'hover:bg-gray-50 dark:hover:bg-bambu-dark/50'}
`}
>
{/* Large prominent ID badge */}
<div className={`
w-12 h-12 flex-shrink-0 rounded-lg flex flex-col items-center justify-center
${obj.skipped
? 'bg-red-100 dark:bg-red-500/20 border border-red-300 dark:border-red-500/40'
: 'bg-green-100 dark:bg-bambu-green/20 border border-green-300 dark:border-bambu-green/40'}
`}>
<span className={`text-lg font-mono font-bold ${obj.skipped ? 'text-red-500 dark:text-red-400' : 'text-green-600 dark:text-bambu-green'}`}>
{obj.id}
</span>
<span className={`text-[8px] uppercase tracking-wider ${obj.skipped ? 'text-red-400/60' : 'text-green-500/60 dark:text-bambu-green/60'}`}>
ID
</span>
</div>
{/* Object name and status */}
<div className="flex-1 min-w-0">
<span className={`block text-sm truncate ${obj.skipped ? 'text-red-500 dark:text-red-400 line-through' : 'text-gray-900 dark:text-white'}`}>
{obj.name}
</span>
{obj.skipped && (
<span className="text-[10px] text-red-400/60">{t('printers.willBeSkipped')}</span>
)}
</div>
{/* Skip button */}
{!obj.skipped ? (
<button
onClick={() => skipObjectsMutation.mutate([obj.id])}
disabled={skipObjectsMutation.isPending || (status?.layer_num ?? 0) <= 1 || !hasPermission('printers:control')}
className={`px-4 py-2 text-xs font-medium rounded-lg transition-colors ${
(status?.layer_num ?? 0) <= 1 || !hasPermission('printers:control')
? 'bg-gray-100 dark:bg-bambu-dark text-gray-400 dark:text-bambu-gray/50 cursor-not-allowed'
: 'bg-red-100 dark:bg-red-500/20 text-red-600 dark:text-red-400 hover:bg-red-200 dark:hover:bg-red-500/30 border border-red-300 dark:border-red-500/30'
}`}
title={!hasPermission('printers:control') ? t('printers.permission.noControl') : ((status?.layer_num ?? 0) <= 1 ? t('printers.skipObjects.waitForLayer', { layer: status?.layer_num ?? 0 }) : t('printers.skipObjects.skip'))}
>
{t('printers.skipObjects.skip')}
</button>
) : (
<span className="px-4 py-2 text-xs text-red-500 dark:text-red-400/70 bg-red-100 dark:bg-red-500/10 rounded-lg">
{t('printers.skipObjects.skipped')}
</span>
)}
</div>
))}
</div>
</div>
</div>
)}
</div>
</div>
);
}
+2 -1
View File
@@ -115,7 +115,7 @@ export default {
deletePrinter: 'Drucker löschen',
printerName: 'Druckername',
serialNumber: 'Seriennummer',
ipAddress: 'IP-Adresse',
ipAddress: 'IP-Adresse / Hostname',
accessCode: 'Zugangscode',
model: 'Modell',
nozzleCount: 'Düsenanzahl',
@@ -1510,6 +1510,7 @@ export default {
recording: 'Aufnahme',
startRecording: 'Aufnahme starten',
stopRecording: 'Aufnahme stoppen',
chamberLight: 'Kammerbeleuchtung umschalten',
},
// Groups management
+2 -1
View File
@@ -115,7 +115,7 @@ export default {
deletePrinter: 'Delete Printer',
printerName: 'Printer Name',
serialNumber: 'Serial Number',
ipAddress: 'IP Address',
ipAddress: 'IP Address / Hostname',
accessCode: 'Access Code',
model: 'Model',
nozzleCount: 'Nozzle Count',
@@ -1510,6 +1510,7 @@ export default {
recording: 'Recording',
startRecording: 'Start Recording',
stopRecording: 'Stop Recording',
chamberLight: 'Toggle chamber light',
},
// Groups management
+3 -2
View File
@@ -95,7 +95,7 @@ export default {
deletePrinter: 'プリンターを削除',
printerName: 'プリンター名',
serialNumber: 'シリアル番号',
ipAddress: 'IPアドレス',
ipAddress: 'IPアドレス / ホスト名',
accessCode: 'アクセスコード',
model: 'モデル',
nozzleCount: 'ノズル数',
@@ -240,7 +240,7 @@ export default {
form: {
name: '名前',
namePlaceholder: 'マイプリンター',
ipAddress: 'IPアドレス',
ipAddress: 'IPアドレス / ホスト名',
serialNumber: 'シリアル番号',
serialCannotChange: 'シリアル番号は変更できません',
accessCode: 'アクセスコード',
@@ -2642,6 +2642,7 @@ export default {
skipForward: '5秒進む',
refreshStream: 'ストリームを更新',
dragToResize: 'ドラッグしてリサイズ',
chamberLight: 'チャンバーライト切替',
},
// アーカイブカードラベル
+81 -5
View File
@@ -1,9 +1,13 @@
import { useState, useEffect, useRef, useCallback } from 'react';
import { useParams } from 'react-router-dom';
import { useQuery } from '@tanstack/react-query';
import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query';
import { useTranslation } from 'react-i18next';
import { RefreshCw, AlertTriangle, Camera, Maximize, Minimize, WifiOff, ZoomIn, ZoomOut } from 'lucide-react';
import { api } from '../api/client';
import { api, getAuthToken } from '../api/client';
import { useToast } from '../contexts/ToastContext';
import { useAuth } from '../contexts/AuthContext';
import { ChamberLight } from '../components/icons/ChamberLight';
import { SkipObjectsModal, SkipObjectsIcon } from '../components/SkipObjectsModal';
const MAX_RECONNECT_ATTEMPTS = 5;
const INITIAL_RECONNECT_DELAY = 2000; // 2 seconds
@@ -12,10 +16,14 @@ const STALL_CHECK_INTERVAL = 5000; // Check every 5 seconds
export function CameraPage() {
const { t } = useTranslation();
const queryClient = useQueryClient();
const { showToast } = useToast();
const { hasPermission } = useAuth();
const { printerId } = useParams<{ printerId: string }>();
const id = parseInt(printerId || '0', 10);
const [streamMode, setStreamMode] = useState<'stream' | 'snapshot'>('stream');
const [showSkipObjectsModal, setShowSkipObjectsModal] = useState(false);
const [streamError, setStreamError] = useState(false);
const [streamLoading, setStreamLoading] = useState(true);
const [imageKey, setImageKey] = useState(Date.now());
@@ -43,6 +51,39 @@ export function CameraPage() {
enabled: id > 0,
});
// Fetch printer status for light toggle and skip objects
const { data: status } = useQuery({
queryKey: ['printerStatus', id],
queryFn: () => api.getPrinterStatus(id),
refetchInterval: 30000,
enabled: id > 0,
});
// Chamber light mutation with optimistic update
const chamberLightMutation = useMutation({
mutationFn: (on: boolean) => api.setChamberLight(id, on),
onMutate: async (on) => {
await queryClient.cancelQueries({ queryKey: ['printerStatus', id] });
const previousStatus = queryClient.getQueryData(['printerStatus', id]);
queryClient.setQueryData(['printerStatus', id], (old: typeof status) => ({
...old,
chamber_light: on,
}));
return { previousStatus };
},
onSuccess: (_, on) => {
showToast(`Chamber light ${on ? 'on' : 'off'}`);
},
onError: (error: Error, _, context) => {
if (context?.previousStatus) {
queryClient.setQueryData(['printerStatus', id], context.previousStatus);
}
showToast(error.message || t('printers.toast.failedToControlChamberLight'), 'error');
},
});
const isPrintingWithObjects = (status?.state === 'RUNNING' || status?.state === 'PAUSE' || status?.state === 'PAUSED') && (status?.printable_objects_count ?? 0) >= 2;
// Update document title
useEffect(() => {
if (printer) {
@@ -64,11 +105,14 @@ export function CameraPage() {
const sendStopOnce = () => {
if (id > 0 && !stopSentRef.current) {
stopSentRef.current = true;
navigator.sendBeacon(stopUrl);
const headers: Record<string, string> = {};
const token = getAuthToken();
if (token) headers['Authorization'] = `Bearer ${token}`;
fetch(stopUrl, { method: 'POST', keepalive: true, headers }).catch(() => {});
}
};
// Handle page unload/close with sendBeacon (more reliable than fetch on unload)
// Handle page unload/close with keepalive fetch (more reliable than sendBeacon, supports auth)
const handleBeforeUnload = () => {
sendStopOnce();
};
@@ -303,7 +347,10 @@ export function CameraPage() {
const stopStream = () => {
if (id > 0) {
fetch(`/api/v1/printers/${id}/camera/stop`).catch(() => {});
const headers: Record<string, string> = {};
const token = getAuthToken();
if (token) headers['Authorization'] = `Bearer ${token}`;
fetch(`/api/v1/printers/${id}/camera/stop`, { method: 'POST', headers }).catch(() => {});
}
};
@@ -577,6 +624,28 @@ export function CameraPage() {
{t('camera.snapshot')}
</button>
</div>
<button
onClick={() => chamberLightMutation.mutate(!status?.chamber_light)}
disabled={!status?.connected || chamberLightMutation.isPending || !hasPermission('printers:control')}
className={`p-1.5 rounded disabled:opacity-50 ${status?.chamber_light ? 'bg-yellow-500/20 hover:bg-yellow-500/30' : 'hover:bg-bambu-dark-tertiary'}`}
title={!hasPermission('printers:control') ? t('printers.permission.noControl') : t('camera.chamberLight')}
>
<ChamberLight on={status?.chamber_light ?? false} className="w-4 h-4" />
</button>
<button
onClick={() => setShowSkipObjectsModal(true)}
disabled={!isPrintingWithObjects || !hasPermission('printers:control')}
className={`p-1.5 rounded disabled:opacity-50 ${isPrintingWithObjects && hasPermission('printers:control') ? 'hover:bg-bambu-dark-tertiary' : ''}`}
title={
!hasPermission('printers:control')
? t('printers.permission.noControl')
: !isPrintingWithObjects
? t('printers.skipObjects.onlyWhilePrinting')
: t('printers.skipObjects.tooltip')
}
>
<SkipObjectsIcon className="w-4 h-4 text-bambu-gray" />
</button>
<button
onClick={refresh}
disabled={isDisabled}
@@ -700,6 +769,13 @@ export function CameraPage() {
</div>
</div>
</div>
{/* Skip Objects Modal */}
<SkipObjectsModal
printerId={id}
isOpen={showSkipObjectsModal}
onClose={() => setShowSkipObjectsModal(false)}
/>
</div>
);
}
+12 -243
View File
@@ -33,7 +33,6 @@ import {
Pause,
Play,
X,
Monitor,
Fan,
Wind,
AirVent,
@@ -45,18 +44,6 @@ import {
Home,
} from 'lucide-react';
// Custom Skip Objects icon - arrow jumping over boxes
const SkipObjectsIcon = ({ className }: { className?: string }) => (
<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="2" strokeLinecap="round" strokeLinejoin="round" className={className}>
{/* Three boxes at the bottom */}
<rect x="2" y="15" width="5" height="5" rx="0.5" />
<rect x="9.5" y="15" width="5" height="5" rx="0.5" fill="currentColor" opacity="0.3" />
<rect x="17" y="15" width="5" height="5" rx="0.5" />
{/* Curved arrow jumping over first box */}
<path d="M4 12 C4 6, 14 6, 14 12" />
<polyline points="12,10 14,12 12,14" />
</svg>
);
import { useNavigate } from 'react-router-dom';
import { api, discoveryApi, firmwareApi } from '../api/client';
import { formatDateOnly } from '../utils/date';
@@ -75,6 +62,7 @@ import { LinkSpoolModal } from '../components/LinkSpoolModal';
import { ConfigureAmsSlotModal } from '../components/ConfigureAmsSlotModal';
import { useToast } from '../contexts/ToastContext';
import { ChamberLight } from '../components/icons/ChamberLight';
import { SkipObjectsModal, SkipObjectsIcon } from '../components/SkipObjectsModal';
// Complete Bambu Lab filament color mapping by tray_id_name
// Source: https://github.com/queengooborg/Bambu-Lab-RFID-Library
@@ -1265,23 +1253,13 @@ function PrinterCard({
// Query for printable objects (for skip functionality)
// Fetch when printing with 2+ objects OR when modal is open
const isPrintingWithObjects = (status?.state === 'RUNNING' || status?.state === 'PAUSE' || status?.state === 'PAUSED') && (status?.printable_objects_count ?? 0) >= 2;
const { data: objectsData, refetch: refetchObjects } = useQuery({
const { data: objectsData } = useQuery({
queryKey: ['printableObjects', printer.id],
queryFn: () => api.getPrintableObjects(printer.id),
enabled: showSkipObjectsModal || isPrintingWithObjects,
refetchInterval: showSkipObjectsModal ? 5000 : (isPrintingWithObjects ? 30000 : false), // 5s when modal open, 30s otherwise
});
// Skip objects mutation
const skipObjectsMutation = useMutation({
mutationFn: (objectIds: number[]) => api.skipObjects(printer.id, objectIds),
onSuccess: (data) => {
showToast(data.message || t('printers.skipObjects.objectsSkipped'));
refetchObjects();
},
onError: (error: Error) => showToast(error.message || t('printers.toast.failedToSkipObjects'), 'error'),
});
// State for tracking which AMS slot is being refreshed
const [refreshingSlot, setRefreshingSlot] = useState<{ amsId: number; slotId: number } | null>(null);
// Track if we've seen the printer enter "busy" state (ams_status_main !== 0)
@@ -3250,221 +3228,12 @@ function PrinterCard({
/>
)}
{/* Skip Objects Popup */}
{showSkipObjectsModal && (
<div
className="fixed inset-0 z-50 flex items-center justify-center"
onClick={() => setShowSkipObjectsModal(false)}
onKeyDown={(e) => e.key === 'Escape' && setShowSkipObjectsModal(false)}
tabIndex={-1}
ref={(el) => el?.focus()}
>
{/* Backdrop */}
<div className="absolute inset-0 bg-black/50 z-0" />
{/* Modal */}
<div
className="relative z-10 bg-white dark:bg-bambu-dark border border-gray-200 dark:border-bambu-dark-tertiary rounded-xl shadow-2xl w-[560px] max-h-[85vh] flex flex-col overflow-hidden"
onClick={(e) => e.stopPropagation()}
>
{/* Header */}
<div className="flex items-center justify-between px-4 py-3 border-b border-gray-200 dark:border-bambu-dark-tertiary bg-gray-50 dark:bg-bambu-dark">
<div className="flex items-center gap-2">
<SkipObjectsIcon className="w-4 h-4 text-bambu-green" />
<span className="text-sm font-medium text-gray-900 dark:text-white">{t('printers.skipObjects.title')}</span>
</div>
<button
onClick={() => setShowSkipObjectsModal(false)}
className="p-1 text-gray-500 dark:text-bambu-gray hover:text-gray-900 dark:hover:text-white rounded transition-colors"
>
<X className="w-4 h-4" />
</button>
</div>
{!objectsData ? (
<div className="flex items-center justify-center py-12">
<Loader2 className="w-5 h-5 animate-spin text-bambu-gray" />
</div>
) : objectsData.objects.length === 0 ? (
<div className="text-center py-8 px-4 text-bambu-gray">
<p className="text-sm">{t('printers.noObjectsFound')}</p>
<p className="text-xs mt-1 opacity-70">{t('printers.objectsLoadedOnPrintStart')}</p>
</div>
) : (
<div className="flex flex-col overflow-hidden">
{/* Info Banner */}
<div className="flex items-center gap-3 px-4 py-2.5 bg-blue-50 dark:bg-blue-500/10 border-b border-gray-200 dark:border-bambu-dark-tertiary">
<div className="flex-shrink-0 w-8 h-8 rounded-lg bg-blue-100 dark:bg-blue-500/20 flex items-center justify-center">
<Monitor className="w-4 h-4 text-blue-500 dark:text-blue-400" />
</div>
<div className="flex-1 min-w-0">
<p className="text-xs text-blue-600 dark:text-blue-300">{t('printers.skipObjects.matchIdsInfo')}</p>
<p className="text-[10px] text-blue-500/70 dark:text-blue-300/60">{t('printers.skipObjects.printerShowsIds')}</p>
</div>
<div className="flex-shrink-0 text-xs text-gray-500 dark:text-bambu-gray">
{objectsData.skipped_count}/{objectsData.total} {t('printers.skipObjects.skipped')}
</div>
</div>
{/* Layer Warning */}
{(status?.layer_num ?? 0) <= 1 && (
<div className="flex items-center gap-2 px-4 py-2 bg-amber-50 dark:bg-amber-500/10 border-b border-gray-200 dark:border-bambu-dark-tertiary">
<AlertCircle className="w-4 h-4 text-amber-500 dark:text-amber-400 flex-shrink-0" />
<p className="text-xs text-amber-600 dark:text-amber-400">
{t('printers.skipObjects.waitForLayer', { layer: status?.layer_num ?? 0 })}
</p>
</div>
)}
{/* Content: Image + List side by side */}
<div className="flex flex-1 overflow-hidden">
{/* Left: Preview Image with object markers */}
<div className="w-52 flex-shrink-0 p-4 border-r border-gray-200 dark:border-bambu-dark-tertiary bg-gray-50 dark:bg-bambu-dark-secondary overflow-y-auto">
<div className="relative">
{status?.cover_url ? (
<img
src={`${status.cover_url}?view=top`}
alt={t('printers.printPreview')}
className="w-full aspect-square object-contain rounded-lg bg-gray-900 dark:bg-gray-900 border border-gray-300 dark:border-gray-600"
/>
) : (
<div className="w-full aspect-square rounded-lg bg-gray-100 dark:bg-bambu-dark flex items-center justify-center">
<Box className="w-8 h-8 text-gray-300 dark:text-bambu-gray/30" />
</div>
)}
{/* Object ID markers overlay - positioned based on object data */}
{objectsData.objects.length > 0 && (
<div className="absolute inset-0 pointer-events-none">
{objectsData.objects.map((obj, idx) => {
let x: number, y: number;
// Use position data if available, otherwise fall back to grid
if (obj.x != null && obj.y != null && objectsData.bbox_all) {
// bbox_all defines the visible area in the top_N.png image
// Format: [x_min, y_min, x_max, y_max] in mm
const [xMin, yMin, xMax, yMax] = objectsData.bbox_all;
const bboxWidth = xMax - xMin;
const bboxHeight = yMax - yMin;
// The image shows bbox_all area with some padding (~5-10%)
const padding = 8;
const contentArea = 100 - (padding * 2);
// Map object position to image percentage
x = padding + ((obj.x - xMin) / bboxWidth) * contentArea;
// Y axis: image Y increases downward, but 3D Y increases toward back
y = padding + ((yMax - obj.y) / bboxHeight) * contentArea;
// Clamp to valid range
x = Math.max(5, Math.min(95, x));
y = Math.max(5, Math.min(95, y));
} else if (obj.x != null && obj.y != null) {
// Fallback: use full build plate (256mm)
const buildPlate = 256;
x = (obj.x / buildPlate) * 100;
y = 100 - (obj.y / buildPlate) * 100;
x = Math.max(5, Math.min(95, x));
y = Math.max(5, Math.min(95, y));
} else {
// Fallback: arrange in a grid pattern over the build plate area
const cols = Math.ceil(Math.sqrt(objectsData.objects.length));
const row = Math.floor(idx / cols);
const col = idx % cols;
const rows = Math.ceil(objectsData.objects.length / cols);
x = 15 + (col * (70 / cols)) + (35 / cols);
y = 15 + (row * (70 / rows)) + (35 / rows);
}
return (
<div
key={obj.id}
className={`absolute flex items-center justify-center w-6 h-6 rounded-full text-[10px] font-bold shadow-lg ${
obj.skipped
? 'bg-red-500 text-white line-through'
: 'bg-bambu-green text-black'
}`}
style={{
left: `${x}%`,
top: `${y}%`,
transform: 'translate(-50%, -50%)'
}}
title={obj.name}
>
{obj.id}
</div>
);
})}
</div>
)}
{/* Object count overlay */}
<div className="absolute bottom-2 right-2 px-2 py-1 bg-white/90 dark:bg-black/80 rounded text-[10px] text-gray-700 dark:text-white shadow-sm">
{t('printers.skipObjects.activeCount', { count: objectsData.objects.filter(o => !o.skipped).length })}
</div>
</div>
</div>
{/* Right: Object List with prominent IDs */}
<div className="flex-1 min-w-0 overflow-y-auto">
{objectsData.objects.map((obj) => (
<div
key={obj.id}
className={`
flex items-center gap-3 px-4 py-3 border-b border-gray-200 dark:border-bambu-dark-tertiary/50 last:border-0
${obj.skipped ? 'bg-red-50 dark:bg-red-500/10' : 'hover:bg-gray-50 dark:hover:bg-bambu-dark/50'}
`}
>
{/* Large prominent ID badge */}
<div className={`
w-12 h-12 flex-shrink-0 rounded-lg flex flex-col items-center justify-center
${obj.skipped
? 'bg-red-100 dark:bg-red-500/20 border border-red-300 dark:border-red-500/40'
: 'bg-green-100 dark:bg-bambu-green/20 border border-green-300 dark:border-bambu-green/40'}
`}>
<span className={`text-lg font-mono font-bold ${obj.skipped ? 'text-red-500 dark:text-red-400' : 'text-green-600 dark:text-bambu-green'}`}>
{obj.id}
</span>
<span className={`text-[8px] uppercase tracking-wider ${obj.skipped ? 'text-red-400/60' : 'text-green-500/60 dark:text-bambu-green/60'}`}>
ID
</span>
</div>
{/* Object name and status */}
<div className="flex-1 min-w-0">
<span className={`block text-sm truncate ${obj.skipped ? 'text-red-500 dark:text-red-400 line-through' : 'text-gray-900 dark:text-white'}`}>
{obj.name}
</span>
{obj.skipped && (
<span className="text-[10px] text-red-400/60">{t('printers.willBeSkipped')}</span>
)}
</div>
{/* Skip button */}
{!obj.skipped ? (
<button
onClick={() => skipObjectsMutation.mutate([obj.id])}
disabled={skipObjectsMutation.isPending || (status?.layer_num ?? 0) <= 1 || !hasPermission('printers:control')}
className={`px-4 py-2 text-xs font-medium rounded-lg transition-colors ${
(status?.layer_num ?? 0) <= 1 || !hasPermission('printers:control')
? 'bg-gray-100 dark:bg-bambu-dark text-gray-400 dark:text-bambu-gray/50 cursor-not-allowed'
: 'bg-red-100 dark:bg-red-500/20 text-red-600 dark:text-red-400 hover:bg-red-200 dark:hover:bg-red-500/30 border border-red-300 dark:border-red-500/30'
}`}
title={!hasPermission('printers:control') ? t('printers.permission.noControl') : ((status?.layer_num ?? 0) <= 1 ? t('printers.skipObjects.waitForLayer', { layer: status?.layer_num ?? 0 }) : t('printers.skipObjects.skip'))}
>
{t('printers.skipObjects.skip')}
</button>
) : (
<span className="px-4 py-2 text-xs text-red-500 dark:text-red-400/70 bg-red-100 dark:bg-red-500/10 rounded-lg">
{t('printers.skipObjects.skipped')}
</span>
)}
</div>
))}
</div>
</div>
</div>
)}
</div>
</div>
)}
{/* Skip Objects Modal */}
<SkipObjectsModal
printerId={printer.id}
isOpen={showSkipObjectsModal}
onClose={() => setShowSkipObjectsModal(false)}
/>
{/* HMS Error Modal */}
{showHMSModal && (
@@ -3846,11 +3615,11 @@ function AddPrinterModal({
<input
type="text"
required
pattern="\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}"
pattern="(\d{1,3}(\.\d{1,3}){3}|[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)*)"
className="w-full px-3 py-2 bg-bambu-dark border border-bambu-dark-tertiary rounded-lg text-white focus:border-bambu-green focus:outline-none"
value={form.ip_address}
onChange={(e) => setForm({ ...form, ip_address: e.target.value })}
placeholder="192.168.1.100"
placeholder="192.168.1.100 or printer.local"
/>
</div>
<div>
@@ -4223,11 +3992,11 @@ function EditPrinterModal({
<input
type="text"
required
pattern="\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}"
pattern="(\d{1,3}(\.\d{1,3}){3}|[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)*)"
className="w-full px-3 py-2 bg-bambu-dark border border-bambu-dark-tertiary rounded-lg text-white focus:border-bambu-green focus:outline-none"
value={form.ip_address}
onChange={(e) => setForm({ ...form, ip_address: e.target.value })}
placeholder="192.168.1.100"
placeholder="192.168.1.100 or printer.local"
/>
</div>
<div>
+3
View File
@@ -6,6 +6,9 @@ pytest-xdist>=3.5.0
httpx>=0.27.0
ruff>=0.8.0
# Required by pyftpdlib TLS_FTPHandler for mock FTP server tests
pyOpenSSL>=24.0.0
# Security scanning
bandit[sarif]>=1.7.0
pip-audit>=2.7.0
File diff suppressed because one or more lines are too long
+1 -1
View File
@@ -23,7 +23,7 @@
<!-- Splash screens for iOS -->
<link rel="apple-touch-startup-image" href="/img/android-chrome-512x512.png" />
<script type="module" crossorigin src="/assets/index-BZQD54OI.js"></script>
<script type="module" crossorigin src="/assets/index-ueO9Fu53.js"></script>
<link rel="stylesheet" crossorigin href="/assets/index-togsBDt6.css">
</head>
<body>