feat(vp): mirror live target printer state to slicer in non-proxy modes

In non-proxy VP modes (Immediate / Review / Print Queue), the slicer now
sees real AMS / FTS / nozzle / k-profile state from the target printer
and streams the live camera — full slicer-as-remote functionality without
giving up Bambuddy's queue / archive / dispatch features.

Architecture (cached-as-base, single source of truth). The bridge caches
the latest real push_status and info.get_version response from Bambuddy's
existing per-printer MQTT subscription — no second session on the printer,
firmware in-flight budget unaffected (#1164). _send_status_report serves
a near-byte-identical copy of the cached push with only the upload-state-
machine fields overridden. Command responses (extrusion_cali_get, AMS
write acks, xcam) fan out raw — they carry sequence_ids the slicer is
waiting on. Slicer-issued commands forward to the printer except
project_file / gcode_file, which still terminate locally because the file
lives on Bambuddy. Camera is a raw TCPProxy on bind_ip:322 → printer:322,
same approach proxy mode uses.

Field-shape gotchas pinned in the bridge module's docstring and the
new test file:
  - Real Bambu pushes use json.dumps(indent=4) wire format. Compact JSON
    fails BambuStudio's Send pre-flight silently.
  - net.info[*].ip is the FTP destination IP (little-endian uint32).
    Without rewriting to the VP bind IP, the slicer FTPs straight to
    the real printer.
  - upgrade_state.sn rewritten to VP serial; AMS-hardware sn fields
    (n3f/0.sn etc.) left alone.
  - ipcam.rtsp_url passes through unchanged; BambuStudio overrides the
    URL host with the device IP it bound on, so :322 lands on the VP's
    TCPProxy.
  - extrusion_cali_get must forward; answering it locally hides the
    user's stored per-filament k-profiles.

Setup nuance for camera: the VP's access code must match the target
printer's because the slicer authenticates RTSPS with whatever access
code is in its profile. MQTT and FTP work either way.

Tested e2e with BambuStudio and OrcaSlicer against H2D (dual-nozzle,
AMS 2 Pro + AMS HT) and X1C across all three non-proxy modes — sync,
send, k-profile lookup, AMS configuration from slicer, and live camera
all work. Proxy mode is untouched: SlicerProxyManager owns its own
proxies and never instantiates SimpleMQTTServer or MQTTBridge.

25 new tests in backend/tests/unit/test_vp_mqtt_bridge.py cover lifecycle,
caching, identity / IP rewriting, wire format, slicer→printer routing,
and the LE-uint32 IP encoder against the real H2D capture value.
This commit is contained in:
maziggy 2026-05-03 13:43:54 +02:00
parent 66a3604d9b
commit 7dea33d0d8
9 changed files with 1175 additions and 16 deletions

File diff suppressed because one or more lines are too long

View file

@ -236,7 +236,8 @@ Perfect for remote print farms, traveling makers, or accessing your home printer
- Interactive API browser with live testing
### 🖨️ Virtual Printer & Remote Printing
- **🌐 Proxy Mode (NEW!)** — Print remotely from anywhere via secure TLS relay
- **🌐 Proxy Mode** — Print remotely from anywhere via secure TLS relay
- **🪞 Live target-printer mirror in non-proxy modes (NEW!)** — Immediate / Review / Queue VPs now mirror their target printer's live state to the slicer: AMS slot contents, FTS / dual-extruder routing, k-profiles, AMS load / dry / calibration commands, and the camera stream all flow through the VP. Use the slicer as a full remote for the printer behind the VP without giving up Bambuddy's queue / archive / dispatch features.
- Emulates a Bambu Lab printer on your network
- Send prints directly from Bambu Studio/Orca Slicer
- Configurable printer model (X1C, P1S, A1, H2D, etc.)

View file

@ -4433,6 +4433,7 @@ async def lifespan(app: FastAPI):
from backend.app.services.virtual_printer import virtual_printer_manager
virtual_printer_manager.set_session_factory(async_session)
virtual_printer_manager.set_printer_manager(printer_manager)
try:
await virtual_printer_manager.sync_from_db()
logging.info("Virtual printer manager synced from database")

View file

@ -358,6 +358,10 @@ class BambuMQTTClient:
self._message_log: deque[MQTTLogEntry] = deque(maxlen=100)
self._logging_enabled: bool = False
self._last_message_time: float = 0.0 # Track when we last received a message
# Raw-message fan-out for VP MQTT bridge (non-proxy modes republish the
# printer's pushes verbatim to slicers connected to a virtual printer).
# Handlers receive (topic, payload_bytes) before JSON parsing.
self._raw_message_handlers: list[Callable[[str, bytes], None]] = []
self._disconnection_event: threading.Event | None = None
self._previous_ams_hash: str | None = None # Track AMS changes
@ -685,6 +689,15 @@ class BambuMQTTClient:
self.on_state_change(self.state)
def _on_message(self, client, userdata, msg):
for handler in self._raw_message_handlers:
try:
handler(msg.topic, msg.payload)
except Exception:
logger.exception(
"[%s] raw-message handler crashed for topic=%s",
self.serial_number,
msg.topic,
)
try:
try:
raw = msg.payload.decode()
@ -3536,6 +3549,39 @@ class BambuMQTTClient:
"""Check if logging is enabled."""
return self._logging_enabled
def register_raw_message_handler(self, handler: Callable[[str, bytes], None]) -> None:
"""Register a handler invoked for every incoming MQTT message.
Used by the VP MQTT bridge to republish the printer's report pushes to
slicers connected to a virtual printer in non-proxy mode. Handlers run
on paho's network thread and must not block; exceptions are caught.
"""
if handler not in self._raw_message_handlers:
self._raw_message_handlers.append(handler)
def unregister_raw_message_handler(self, handler: Callable[[str, bytes], None]) -> None:
"""Unregister a previously-registered raw-message handler."""
try:
self._raw_message_handlers.remove(handler)
except ValueError:
pass
def publish_raw(self, topic: str, payload: bytes | str, qos: int = 1) -> bool:
"""Publish a pre-formed payload directly to the printer's MQTT broker.
Used by the VP MQTT bridge to forward slicer-originated commands without
going through send_command's sequence-id mangling. Returns False if the
underlying paho client isn't ready.
"""
if self._client is None:
return False
try:
info = self._client.publish(topic, payload, qos=qos)
return info.rc == mqtt.MQTT_ERR_SUCCESS
except Exception:
logger.exception("[%s] publish_raw failed for topic=%s", self.serial_number, topic)
return False
def send_drying_command(
self, ams_id: int, temp: int, duration: int, mode: int = 1, filament: str = "", rotate_tray: bool = False
):

View file

@ -9,15 +9,20 @@ import logging
from collections.abc import Callable
from datetime import datetime, timezone
from pathlib import Path
from typing import TYPE_CHECKING
from backend.app.core.config import settings as app_settings
from backend.app.services.virtual_printer.bind_server import BindServer
from backend.app.services.virtual_printer.certificate import CertificateService
from backend.app.services.virtual_printer.ftp_server import VirtualPrinterFTPServer
from backend.app.services.virtual_printer.mqtt_bridge import MQTTBridge
from backend.app.services.virtual_printer.mqtt_server import SimpleMQTTServer
from backend.app.services.virtual_printer.ssdp_server import SSDPProxy, VirtualPrinterSSDPServer
from backend.app.services.virtual_printer.tailscale import tailscale_service
from backend.app.services.virtual_printer.tcp_proxy import SlicerProxyManager
from backend.app.services.virtual_printer.tcp_proxy import SlicerProxyManager, TCPProxy
if TYPE_CHECKING:
from backend.app.services.printer_manager import PrinterManager
logger = logging.getLogger(__name__)
@ -117,6 +122,7 @@ class VirtualPrinterInstance:
tailscale_disabled: bool = True,
base_dir: Path,
session_factory: Callable | None = None,
printer_manager: "PrinterManager | None" = None,
):
self.id = vp_id
self.name = name
@ -133,6 +139,7 @@ class VirtualPrinterInstance:
self.remote_interface_ip = remote_interface_ip
self.tailscale_disabled = tailscale_disabled
self._session_factory = session_factory
self._printer_manager = printer_manager
# Directories
self.upload_dir = base_dir / "uploads" / str(vp_id)
@ -161,6 +168,8 @@ class VirtualPrinterInstance:
self._proxy: SlicerProxyManager | None = None
self._ftp: VirtualPrinterFTPServer | None = None
self._mqtt: SimpleMQTTServer | None = None
self._mqtt_bridge: MQTTBridge | None = None
self._rtsp_proxy: TCPProxy | None = None
self._bind: BindServer | None = None
self._ssdp: VirtualPrinterSSDPServer | None = None
self._ssdp_proxy: SSDPProxy | None = None
@ -621,6 +630,44 @@ class VirtualPrinterInstance:
)
)
# MQTT bridge — fans out the target printer's pushes to slicers connected
# to this VP and forwards their commands back to the printer. Only meaningful
# when a target printer is configured AND printer_manager was injected (it
# always is at runtime; tests may omit it).
if self.target_printer_id is not None and self._printer_manager is not None:
self._mqtt_bridge = MQTTBridge(
vp_id=self.id,
vp_name=self.name,
vp_serial=self.serial,
target_printer_id=self.target_printer_id,
mqtt_server=self._mqtt,
printer_manager=self._printer_manager,
)
self._mqtt.set_bridge(self._mqtt_bridge)
await self._mqtt_bridge.start()
# RTSPS camera passthrough on port 322. BambuStudio's camera button
# connects to the device IP it bound on (the VP), not the IP in
# `ipcam.rtsp_url`. Without a listener on <bind_ip>:322 the slicer
# gets connection refused → "LAN connection failed". Same raw TCP
# pass-through used by SlicerProxyManager in proxy mode.
target_client = self._printer_manager.get_client(self.target_printer_id)
target_ip = getattr(target_client, "ip_address", None) if target_client else None
if target_ip:
self._rtsp_proxy = TCPProxy(
name="RTSP",
listen_port=322,
target_host=target_ip,
target_port=322,
bind_address=bind_addr,
)
self._tasks.append(
asyncio.create_task(
run_with_logging(self._rtsp_proxy.start(), "RTSP"),
name=f"vp_{self.id}_rtsp",
)
)
# Bind server
self._bind = BindServer(
serial=self.serial,
@ -663,6 +710,20 @@ class VirtualPrinterInstance:
"""Stop server-mode services."""
await self._cancel_renewal_task()
await self._cancel_restart_task()
if self._mqtt_bridge:
try:
await self._mqtt_bridge.stop()
except Exception:
logger.exception("[VP %s] MQTT bridge stop failed", self.name)
if self._mqtt:
self._mqtt.set_bridge(None)
self._mqtt_bridge = None
if self._rtsp_proxy:
try:
await self._rtsp_proxy.stop()
except Exception:
logger.exception("[VP %s] RTSP proxy stop failed", self.name)
self._rtsp_proxy = None
if self._ftp:
await self._ftp.stop()
self._ftp = None
@ -803,6 +864,7 @@ class VirtualPrinterManager:
def __init__(self):
self._session_factory: Callable | None = None
self._printer_manager: PrinterManager | None = None
self._instances: dict[int, VirtualPrinterInstance] = {}
# Directories
@ -827,6 +889,10 @@ class VirtualPrinterManager:
"""Set the database session factory."""
self._session_factory = session_factory
def set_printer_manager(self, printer_manager: "PrinterManager") -> None:
"""Inject the global printer_manager so non-proxy VPs can mirror their target's MQTT stream."""
self._printer_manager = printer_manager
@property
def is_enabled(self) -> bool:
"""Check if any virtual printer is running."""
@ -939,6 +1005,7 @@ class VirtualPrinterManager:
tailscale_disabled=vp.tailscale_disabled,
base_dir=self._base_dir,
session_factory=self._session_factory,
printer_manager=self._printer_manager,
)
self._instances[vp.id] = instance
await instance.start_server()

View file

@ -0,0 +1,334 @@
"""MQTT bridge for non-proxy virtual printers.
Mirrors the target printer's state to slicers connected to a virtual printer
without opening a second MQTT session on the printer (reuses Bambuddy's
existing subscription firmware inflight budget unaffected, see PR #1164).
Architecture (cached-as-base, not a separate fan-out stream):
- **push_status** snapshots from the printer are CACHED here. The VP's
`SimpleMQTTServer._send_status_report` consults that cache and sends
a near-byte-identical copy of the real push to the slicer (with
sequence_id / gcode_state / etc. overridden). Single source of truth
keeps BambuStudio's Send pre-flight happy.
- **info.get_version** responses are also cached so the synthetic version
response can include the real AMS module list (n3f/n3s/ams entries).
Without this BambuStudio's Prepare tab labels every AMS as "unknown".
- **Other command responses** (extrusion_cali_get, AMS write acks,
xcam responses, ) are fanned out raw to the slicer they carry
sequence_ids the slicer is waiting on; the slicer matches and ignores
unrelated ones.
Identity rewriting at cache time:
- `upgrade_state.sn` (and any other nested dict's `sn` matching the real
serial) VP serial
- `net.info[*].ip` little-endian uint32 VP bind IP. BambuStudio reads
this as the FTP destination IP. Without this the slicer FTPs straight
to the real printer and bypasses Bambuddy.
- `ipcam.rtsp_url` is left unchanged: BambuStudio overrides the URL host
with the device IP it bound to (the VP), so the slicer hits the VP's
own RTSPS proxy on port 322.
"""
from __future__ import annotations
import asyncio
import copy
import json
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from backend.app.services.bambu_mqtt import BambuMQTTClient
from backend.app.services.printer_manager import PrinterManager
from backend.app.services.virtual_printer.mqtt_server import SimpleMQTTServer
logger = logging.getLogger(__name__)
REFRESH_INTERVAL_SECONDS = 30.0
def _ip_to_uint32_le(ip_str: str) -> int:
"""Encode dotted-quad IPv4 as little-endian uint32 (Bambu MQTT's `net.info[].ip` shape)."""
parts = [int(x) for x in ip_str.split(".")]
if len(parts) != 4 or any(p < 0 or p > 255 for p in parts):
raise ValueError(f"invalid IPv4: {ip_str!r}")
return parts[0] | (parts[1] << 8) | (parts[2] << 16) | (parts[3] << 24)
class MQTTBridge:
"""Per-VP MQTT fan-out between a real printer and slicers connected to a VP."""
def __init__(
self,
*,
vp_id: int,
vp_name: str,
vp_serial: str,
target_printer_id: int,
mqtt_server: SimpleMQTTServer,
printer_manager: PrinterManager,
):
self.vp_id = vp_id
self.vp_name = vp_name
self.vp_serial = vp_serial
self.target_printer_id = target_printer_id
self._mqtt_server = mqtt_server
self._printer_manager = printer_manager
self._target_client: BambuMQTTClient | None = None
self._target_serial: str | None = None
self._target_ip_uint32_le: int | None = None
self._vp_ip_uint32_le: int | None = None
self._loop: asyncio.AbstractEventLoop | None = None
self._refresh_task: asyncio.Task | None = None
self._stopping = False
self._latest_print_state: dict | None = None
self._latest_version_modules: list | None = None
@property
def is_active(self) -> bool:
"""True iff a target client is bound and currently connected."""
client = self._target_client
return bool(client is not None and getattr(client, "state", None) and client.state.connected)
async def start(self) -> None:
"""Bind to the target printer (if connected) and start the refresh loop."""
self._loop = asyncio.get_running_loop()
self._stopping = False
self._resolve_client()
self._refresh_task = asyncio.create_task(self._refresh_loop())
async def stop(self) -> None:
"""Detach from the target printer and stop the refresh loop."""
self._stopping = True
if self._refresh_task is not None:
self._refresh_task.cancel()
try:
await self._refresh_task
except asyncio.CancelledError:
pass
self._refresh_task = None
self._unbind_client()
self._loop = None
async def _refresh_loop(self) -> None:
"""Re-resolve the target client periodically — paho clients can be replaced.
BambuMQTTClient is destroyed and recreated on PrinterManager.connect_printer
(e.g. printer config update). Without periodic refresh the bridge would lose
fan-out after such a churn until the VP itself restarts.
"""
try:
while not self._stopping:
await asyncio.sleep(REFRESH_INTERVAL_SECONDS)
self._resolve_client()
except asyncio.CancelledError:
raise
except Exception:
logger.exception("[%s] MQTT bridge refresh loop crashed", self.vp_name)
def _resolve_client(self) -> None:
"""Look up the current client for target_printer_id and rebind if it changed."""
try:
current = self._printer_manager.get_client(self.target_printer_id)
except Exception:
logger.exception("[%s] MQTT bridge: get_client failed", self.vp_name)
return
if current is self._target_client:
return
# Client identity changed — unregister from the old, register on the new.
self._unbind_client()
if current is None:
return
try:
current.register_raw_message_handler(self._on_printer_raw)
except Exception:
logger.exception("[%s] MQTT bridge: register_raw_message_handler failed", self.vp_name)
return
self._target_client = current
self._target_serial = getattr(current, "serial_number", None)
# Cache printer IP and VP bind IP encoded as little-endian uint32, so we
# can rewrite `net.info[*].ip` in cached push_status. BambuStudio reads
# that field for the FTP destination IP — without rewriting, the slicer
# bypasses the VP and FTPs straight to the real printer.
target_ip = getattr(current, "ip_address", None)
vp_ip = getattr(self._mqtt_server, "bind_address", None)
if target_ip and vp_ip and vp_ip not in ("0.0.0.0", "", None): # nosec B104
try:
self._target_ip_uint32_le = _ip_to_uint32_le(target_ip)
self._vp_ip_uint32_le = _ip_to_uint32_le(vp_ip)
except ValueError:
self._target_ip_uint32_le = None
self._vp_ip_uint32_le = None
logger.info(
"[%s] MQTT bridge bound to printer %s (serial=%s)",
self.vp_name,
self.target_printer_id,
self._target_serial,
)
# Trigger a fresh get_version + pushall against the printer so the bridge
# cache populates immediately. Bambuddy itself queries these on connect,
# but that fires before the bridge attaches as a raw-message consumer,
# so without this nudge the cache stays empty until the next periodic
# query (which can be minutes away).
request_fn = getattr(current, "_request_version", None)
if callable(request_fn):
try:
request_fn()
except Exception:
logger.exception("[%s] MQTT bridge: _request_version failed", self.vp_name)
request_status_fn = getattr(current, "request_status_update", None)
if callable(request_status_fn):
try:
request_status_fn()
except Exception:
logger.exception("[%s] MQTT bridge: request_status_update failed", self.vp_name)
def _unbind_client(self) -> None:
if self._target_client is None:
return
try:
self._target_client.unregister_raw_message_handler(self._on_printer_raw)
except Exception:
logger.exception("[%s] MQTT bridge: unregister_raw_message_handler failed", self.vp_name)
logger.info("[%s] MQTT bridge unbound from printer %s", self.vp_name, self.target_printer_id)
self._target_client = None
self._target_serial = None
def _on_printer_raw(self, topic: str, payload: bytes) -> None:
"""Paho-thread callback — cache the latest push_status for synthetic replay.
Instead of fanning out a second stream of MQTT messages to the slicer
(which trips BambuStudio's Send pre-flight consistency checks), we cache
the latest real printer push_status here. The VP's existing 1 Hz
synthetic push (which is what Send is built around) consults this cache
and replaces its stub fields with real values when available.
"""
if self._stopping:
return
target_serial = self._target_serial
if not target_serial:
return
prefix = f"device/{target_serial}/"
if not topic.startswith(prefix):
return
suffix = topic[len(prefix) :]
if not suffix.startswith("report"):
return
try:
data = json.loads(payload)
except json.JSONDecodeError:
return
# Race-free by construction: `json.loads` returns a fresh dict tree per
# call so paho-thread mutations below cannot collide with prior cached
# state held by the asyncio thread. `_send_status_report`'s shallow
# `dict(cached)` is also safe because nothing else writes to the cached
# tree after assignment. The defensive deep-copy on store below removes
# any future risk if a maintainer later re-enters the cached dict to
# mutate it.
# push_status snapshots → cache the print dict for the periodic 1 Hz
# cached-as-base delivery. We do NOT fan these out separately (the
# 1 Hz cached-as-base IS the slicer-facing push_status stream).
print_data = data.get("print")
if isinstance(print_data, dict) and print_data.get("command") == "push_status":
for value in print_data.values():
if isinstance(value, dict) and value.get("sn") == target_serial:
value["sn"] = self.vp_serial
# Note: `ipcam.rtsp_url` carries the real printer's IP. We pass it
# through unchanged — the slicer uses it to fetch the live camera
# stream directly from the printer. On the same LAN this works as
# long as the slicer's stored access code matches the printer's
# (i.e. configure the VP with the same access code as its target).
# Rewrite real printer IP → VP bind IP in `net.info[*].ip` so the
# slicer's FTP destination resolves to the VP, not the real printer.
if self._target_ip_uint32_le is not None and self._vp_ip_uint32_le is not None:
net = print_data.get("net")
if isinstance(net, dict):
info = net.get("info")
if isinstance(info, list):
for entry in info:
if isinstance(entry, dict) and entry.get("ip") == self._target_ip_uint32_le:
entry["ip"] = self._vp_ip_uint32_le
# Defensive deep copy on store so the cache is fully decoupled from
# the freshly-parsed tree and from any reader's reference.
self._latest_print_state = copy.deepcopy(print_data)
return
# info.get_version responses → cache the module list so the synthetic
# version response can include the real AMS modules.
info_data = data.get("info")
if isinstance(info_data, dict) and info_data.get("command") == "get_version":
modules = info_data.get("module")
if isinstance(modules, list):
rewritten: list = []
for module in modules:
if isinstance(module, dict):
module = dict(module)
if module.get("sn") == target_serial:
module["sn"] = self.vp_serial
rewritten.append(module)
self._latest_version_modules = rewritten
# Don't fan out get_version — the slicer's request (when it issues
# one) is intercepted locally and answered from the cached modules.
return
# Everything else (extrusion_cali_get response, AMS write acks, xcam
# responses, …): fan out to the slicer. These are responses to commands
# the slicer (or Bambuddy) issued; the slicer matches by sequence_id and
# ignores responses to commands it didn't send. Without this, slicer-
# initiated queries like extrusion_cali_get hang forever and BambuStudio
# blocks Send waiting for the response.
loop = self._loop
if loop is None:
return
target_bytes = target_serial.encode("ascii")
if target_bytes in payload:
payload = payload.replace(target_bytes, self.vp_serial.encode("ascii"))
vp_topic = f"device/{self.vp_serial}/{suffix}"
try:
asyncio.run_coroutine_threadsafe(
self._mqtt_server.push_raw_to_clients(vp_topic, payload),
loop,
)
except RuntimeError:
pass
def get_latest_print_state(self) -> dict | None:
"""Return the most recent real printer push_status `print` dict, or None."""
return self._latest_print_state
def get_latest_version_modules(self) -> list | None:
"""Return the most recent real printer get_version `module` list, or None."""
return self._latest_version_modules
def forward_to_printer(self, payload: dict) -> bool:
"""Publish a slicer-originated command to the real printer's request topic.
Returns False if no printer client is currently bound.
"""
client = self._target_client
target_serial = self._target_serial
if client is None or target_serial is None:
logger.debug(
"[%s] forward_to_printer dropped (printer %s not bound): %s",
self.vp_name,
self.target_printer_id,
list(payload.keys()),
)
return False
topic = f"device/{target_serial}/request"
try:
return client.publish_raw(topic, json.dumps(payload), qos=1)
except Exception:
logger.exception("[%s] forward_to_printer publish failed", self.vp_name)
return False

View file

@ -10,6 +10,10 @@ import logging
import ssl
from collections.abc import Callable
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from backend.app.services.virtual_printer.mqtt_bridge import MQTTBridge
logger = logging.getLogger(__name__)
@ -218,6 +222,12 @@ class SimpleMQTTServer:
self._current_file = ""
self._prepare_percent = "0"
# MQTT bridge for non-proxy modes — set by VirtualPrinterInstance after start().
# When the bridge is_active, real printer pushes are fanned out to slicers and
# the synthetic 1s push is suspended. When the target printer goes offline the
# synthetic fallback resumes automatically.
self._bridge: MQTTBridge | None = None
async def start(self) -> None:
"""Start the MQTT server."""
if self._running:
@ -346,14 +356,17 @@ class SimpleMQTTServer:
return None
return rest[:slash]
def set_bridge(self, bridge: "MQTTBridge | None") -> None:
"""Attach (or detach) the MQTT bridge that mirrors the target printer."""
self._bridge = bridge
async def _periodic_status_push(self) -> None:
"""Send periodic status updates to all connected clients."""
"""Send periodic status updates to all connected clients (1 Hz, exact pre-bridge behaviour)."""
logger.info("Starting periodic status push task")
while self._running:
try:
await asyncio.sleep(1) # Push every 1 second like real printers
# Send status to all connected clients
disconnected = []
for client_id, writer in list(self._clients.items()):
try:
@ -378,6 +391,48 @@ class SimpleMQTTServer:
logger.info("Periodic status push task stopped")
async def push_raw_to_clients(self, topic: str, payload: bytes) -> None:
"""Publish a pre-serialized MQTT payload on `topic` to every connected slicer.
Called by MQTTBridge from the asyncio loop (scheduled via
run_coroutine_threadsafe from paho's network thread).
"""
topic_bytes = topic.encode("utf-8")
# MQTT remaining-length: 2-byte topic length prefix + topic + message body.
remaining = 2 + len(topic_bytes) + len(payload)
packet = bytearray([0x30]) # PUBLISH, QoS 0
while True:
byte = remaining % 128
remaining //= 128
if remaining > 0:
byte |= 0x80
packet.append(byte)
if remaining == 0:
break
packet.extend([len(topic_bytes) >> 8, len(topic_bytes) & 0xFF])
packet.extend(topic_bytes)
packet.extend(payload)
frame = bytes(packet)
disconnected = []
for client_id, writer in list(self._clients.items()):
try:
if writer.is_closing():
disconnected.append(client_id)
continue
writer.write(frame)
try:
await asyncio.wait_for(writer.drain(), timeout=5)
except TimeoutError:
logger.debug("MQTT drain timeout pushing bridge frame to %s", client_id)
except OSError as e:
logger.debug("Failed to push bridge frame to %s: %s", client_id, e)
disconnected.append(client_id)
for client_id in disconnected:
self._clients.pop(client_id, None)
self._client_serials.pop(client_id, None)
async def _handle_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
"""Handle an MQTT client connection."""
addr = writer.get_extra_info("peername")
@ -575,10 +630,40 @@ class SimpleMQTTServer:
logger.debug("MQTT SUBSCRIBE error: %s", e)
async def _send_status_report(self, writer: asyncio.StreamWriter, serial: str | None = None) -> None:
"""Send a status report to the slicer after connection."""
"""Send a status report to the slicer after connection.
When a bridge is active and has cached the real printer's latest
push_status, send a copy of the real push with only the upload-state-
machine fields we own (gcode_state, gcode_file, prepare_percent,
subtask_name) overridden. BambuStudio's Send pre-flight checks the
push_status shape against what it expects from the printer model, and
the synthetic stub introduced fields the real H2D doesn't have (storage,
the wrong chamber_temper shape, etc.) which trip the check.
"""
try:
# Build status message matching Bambu printer format
self._sequence_id += 1
cached = self._bridge.get_latest_print_state() if self._bridge is not None else None
if isinstance(cached, dict):
# Real-printer-shaped response. Copy the cache, then replace the
# protocol / upload-state fields with values under our control.
print_block = dict(cached)
print_block["sequence_id"] = str(self._sequence_id)
print_block["command"] = "push_status"
print_block["msg"] = 0
print_block["gcode_state"] = self._gcode_state
print_block["gcode_file"] = self._current_file
print_block["gcode_file_prepare_percent"] = self._prepare_percent
if self._current_file:
print_block["subtask_name"] = self._current_file.replace(".3mf", "")
else:
# Don't override real subtask_name with empty if no upload pending.
print_block.setdefault("subtask_name", "")
status = {"print": print_block}
await self._publish_to_report(writer, status, serial or self.serial)
return
# No bridge / no cache yet — fall back to the synthetic stub.
status = {
"print": {
"sequence_id": str(self._sequence_id),
@ -724,6 +809,15 @@ class SimpleMQTTServer:
}
}
# Overlay real version modules from the bridge cache when available
# (specifically the AMS modules ams/0, n3f/0, n3s/128 etc. that
# BambuStudio's Prepare tab uses to identify AMS hardware — without
# them every AMS unit shows as "unknown" in the Prepare panel).
if self._bridge is not None:
cached_modules = self._bridge.get_latest_version_modules()
if isinstance(cached_modules, list) and cached_modules:
version_info["info"]["module"] = cached_modules
await self._publish_to_report(writer, version_info, serial)
logger.info("Sent version response (product_name=%s)", product_name)
@ -740,9 +834,15 @@ class SimpleMQTTServer:
self._prepare_percent = prepare_percent
async def _publish_to_report(self, writer: asyncio.StreamWriter, payload: dict, serial: str = "") -> None:
"""Publish a message on the device report topic."""
"""Publish a message on the device report topic.
Real Bambu printers wire-format push_status JSON with 4-space indentation
(32254 bytes for an idle H2D push vs 14268 bytes compact). BambuStudio's
Send pre-flight rejects compact JSON without matching the on-wire
format the slicer never proceeds to FTP upload.
"""
topic = f"device/{serial or self.serial}/report"
message = json.dumps(payload)
message = json.dumps(payload, indent=4)
topic_bytes = topic.encode("utf-8")
message_bytes = message.encode("utf-8")
@ -857,6 +957,15 @@ class SimpleMQTTServer:
)
return
# The synthetic flow below is the original (pre-bridge) behaviour and is
# what the proven-working FTP "Send" depends on. Do NOT replace any
# synthetic response with a forward — only ADD forwarding alongside,
# at the bottom, for commands the synthetic flow doesn't handle
# (AMS write / xcam / system / etc., which need to actually reach
# the real printer).
handled_locally = False
# Handle pushing command (status request)
if "pushing" in data:
pushing_data = data["pushing"]
@ -864,13 +973,13 @@ class SimpleMQTTServer:
logger.info("MQTT pushing command: %s", command)
if command == "pushall":
# Slicer is requesting full status - send response
logger.info("Sending status report in response to pushall")
await self._send_status_report(writer, serial=client_serial)
handled_locally = True
elif command == "start":
# Slicer wants periodic status updates - send one now
logger.info("Starting status push stream")
await self._send_status_report(writer, serial=client_serial)
handled_locally = True
# Handle info commands (get_version, etc.)
if "info" in data:
@ -881,6 +990,7 @@ class SimpleMQTTServer:
if command == "get_version":
await self._send_version_response(writer, sequence_id, serial=client_serial)
handled_locally = True
# Handle print commands
if "print" in data:
@ -891,13 +1001,18 @@ class SimpleMQTTServer:
logger.info("MQTT print command: %s for %s", command, filename)
if command == "project_file":
# Respond with PREPARE status so slicer proceeds with FTP upload
if command in ("project_file", "gcode_file"):
# File lives on Bambuddy, not the printer — synthetic only.
file_3mf = print_data.get("file", filename)
await self._send_print_response(writer, sequence_id, file_3mf, serial=client_serial)
if self.on_print_command:
await self._notify_print_command(filename, print_data)
handled_locally = True
# Forward anything the synthetic flow didn't handle to the real
# printer. AMS load / dry / xcam / system / extrusion_cali_get etc.
if not handled_locally and self._bridge is not None and self._bridge.is_active:
self._bridge.forward_to_printer(data)
except (IndexError, ValueError, OSError) as e:
logger.debug("MQTT PUBLISH error: %s", e)

View file

@ -0,0 +1,592 @@
"""Tests for the VP MQTT bridge — non-proxy mirror of target printer state to slicer."""
import asyncio
import json
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from backend.app.services.virtual_printer.mqtt_bridge import MQTTBridge, _ip_to_uint32_le
from backend.app.services.virtual_printer.mqtt_server import SimpleMQTTServer
H2D_SERIAL = "0948BB540200427"
VP_SERIAL = "09400A391800003"
H2D_IP = "192.168.255.133"
VP_IP = "192.168.255.16"
def _make_server(serial: str = VP_SERIAL, bind_address: str = VP_IP) -> SimpleMQTTServer:
return SimpleMQTTServer(
serial=serial,
access_code="deadbeef",
cert_path=Path("/tmp/unused.crt"), # nosec B108
key_path=Path("/tmp/unused.key"), # nosec B108
model="O1D",
bind_address=bind_address,
)
def _make_paho_client(
serial: str = H2D_SERIAL,
ip: str = H2D_IP,
*,
connected: bool = True,
) -> MagicMock:
"""Build a mock BambuMQTTClient that satisfies MQTTBridge's interface."""
client = MagicMock()
client.serial_number = serial
client.ip_address = ip
client.state = MagicMock()
client.state.connected = connected
client.publish_raw = MagicMock(return_value=True)
client._raw_handlers: list = []
def _register(handler):
client._raw_handlers.append(handler)
def _unregister(handler):
if handler in client._raw_handlers:
client._raw_handlers.remove(handler)
client.register_raw_message_handler.side_effect = _register
client.unregister_raw_message_handler.side_effect = _unregister
# No-op for _request_version / request_status_update so the post-bind nudge doesn't crash.
client._request_version = MagicMock()
client.request_status_update = MagicMock()
return client
def _make_printer_manager(client) -> MagicMock:
pm = MagicMock()
pm.get_client = MagicMock(return_value=client)
return pm
def _make_bridge(server: SimpleMQTTServer, target: MagicMock | None = None) -> MQTTBridge:
target = target if target is not None else _make_paho_client()
pm = _make_printer_manager(target)
return MQTTBridge(
vp_id=1,
vp_name="vp1",
vp_serial=VP_SERIAL,
target_printer_id=42,
mqtt_server=server,
printer_manager=pm,
)
# ---------------------------------------------------------------------------
# Lifecycle
# ---------------------------------------------------------------------------
class TestBridgeLifecycle:
@pytest.mark.asyncio
async def test_start_registers_handler_on_target_client(self):
target = _make_paho_client()
bridge = _make_bridge(_make_server(), target)
await bridge.start()
assert len(target._raw_handlers) == 1
assert bridge.is_active is True
await bridge.stop()
assert len(target._raw_handlers) == 0
@pytest.mark.asyncio
async def test_start_with_no_target_client_does_not_crash(self):
pm = MagicMock()
pm.get_client = MagicMock(return_value=None)
bridge = MQTTBridge(
vp_id=1,
vp_name="vp1",
vp_serial=VP_SERIAL,
target_printer_id=42,
mqtt_server=_make_server(),
printer_manager=pm,
)
await bridge.start()
assert bridge.is_active is False
await bridge.stop()
@pytest.mark.asyncio
async def test_resolve_rebinds_when_paho_client_replaced(self):
"""BambuMQTTClient is destroyed and recreated on connect_printer; bridge must rebind."""
old_client = _make_paho_client(serial="REAL_OLD")
new_client = _make_paho_client(serial="REAL_NEW")
pm = _make_printer_manager(old_client)
bridge = MQTTBridge(
vp_id=1,
vp_name="vp1",
vp_serial=VP_SERIAL,
target_printer_id=42,
mqtt_server=_make_server(),
printer_manager=pm,
)
await bridge.start()
assert len(old_client._raw_handlers) == 1
assert bridge._target_serial == "REAL_OLD"
pm.get_client.return_value = new_client
bridge._resolve_client()
assert len(old_client._raw_handlers) == 0
assert len(new_client._raw_handlers) == 1
assert bridge._target_serial == "REAL_NEW"
await bridge.stop()
@pytest.mark.asyncio
async def test_post_bind_nudge_requests_version_and_status(self):
target = _make_paho_client()
bridge = _make_bridge(_make_server(), target)
await bridge.start()
target._request_version.assert_called_once()
target.request_status_update.assert_called_once()
await bridge.stop()
# ---------------------------------------------------------------------------
# Caching: push_status
# ---------------------------------------------------------------------------
class TestPushStatusCache:
"""push_status snapshots feed `_send_status_report` via the cache, not a fan-out."""
@pytest.mark.asyncio
async def test_push_status_is_cached_not_fanned_out(self):
server = _make_server()
server.push_raw_to_clients = AsyncMock()
bridge = _make_bridge(server)
await bridge.start()
payload = json.dumps({"print": {"command": "push_status", "ams": {"ams": []}, "gcode_state": "IDLE"}}).encode()
bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload)
await asyncio.sleep(0.01)
server.push_raw_to_clients.assert_not_awaited()
cached = bridge.get_latest_print_state()
assert cached is not None
assert cached["command"] == "push_status"
assert cached["gcode_state"] == "IDLE"
await bridge.stop()
@pytest.mark.asyncio
async def test_serial_rewritten_in_cached_push(self):
server = _make_server()
bridge = _make_bridge(server)
await bridge.start()
payload = json.dumps(
{
"print": {
"command": "push_status",
"upgrade_state": {"sn": H2D_SERIAL, "status": "IDLE"},
}
}
).encode()
bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload)
await asyncio.sleep(0.01)
cached = bridge.get_latest_print_state()
assert cached["upgrade_state"]["sn"] == VP_SERIAL
await bridge.stop()
@pytest.mark.asyncio
async def test_net_info_ip_rewritten_to_vp_ip(self):
"""BambuStudio reads `net.info[].ip` (LE uint32) for the FTP destination —
must be rewritten to the VP's bind IP or the slicer bypasses the VP."""
server = _make_server(bind_address=VP_IP)
bridge = _make_bridge(server)
await bridge.start()
h2d_le = _ip_to_uint32_le(H2D_IP)
vp_le = _ip_to_uint32_le(VP_IP)
payload = json.dumps(
{
"print": {
"command": "push_status",
"net": {"info": [{"ip": h2d_le, "mask": 0xFFFFFF}, {"ip": 0, "mask": 0}]},
}
}
).encode()
bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload)
await asyncio.sleep(0.01)
cached = bridge.get_latest_print_state()
assert cached["net"]["info"][0]["ip"] == vp_le
assert cached["net"]["info"][1]["ip"] == 0 # untouched
await bridge.stop()
@pytest.mark.asyncio
async def test_request_topic_message_is_ignored(self):
server = _make_server()
bridge = _make_bridge(server)
await bridge.start()
payload = json.dumps({"print": {"command": "push_status"}}).encode()
bridge._on_printer_raw(f"device/{H2D_SERIAL}/request", payload)
await asyncio.sleep(0.01)
assert bridge.get_latest_print_state() is None
await bridge.stop()
# ---------------------------------------------------------------------------
# Caching: get_version response
# ---------------------------------------------------------------------------
class TestVersionCache:
@pytest.mark.asyncio
async def test_get_version_response_caches_modules(self):
server = _make_server()
bridge = _make_bridge(server)
await bridge.start()
payload = json.dumps(
{
"info": {
"command": "get_version",
"module": [
{"name": "ota", "sn": H2D_SERIAL, "sw_ver": "01.03.00.00"},
{"name": "n3f/0", "sn": "AMS_HW_1", "sw_ver": "04.00.21.87"},
],
}
}
).encode()
bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload)
await asyncio.sleep(0.01)
modules = bridge.get_latest_version_modules()
assert modules is not None
assert len(modules) == 2
# Device-level sn rewritten; AMS-hardware sn left alone.
assert modules[0]["sn"] == VP_SERIAL
assert modules[1]["sn"] == "AMS_HW_1"
await bridge.stop()
# ---------------------------------------------------------------------------
# Selective fan-out (everything that's not push_status / get_version)
# ---------------------------------------------------------------------------
class TestCommandResponseFanout:
@pytest.mark.asyncio
async def test_extrusion_cali_get_response_is_fanned_out(self):
"""Slicer's extrusion_cali_get goes to the printer; the printer's response
must reach the slicer or BambuStudio's pre-flight blocks Send."""
server = _make_server()
server.push_raw_to_clients = AsyncMock()
bridge = _make_bridge(server)
await bridge.start()
body = json.dumps({"print": {"command": "extrusion_cali_get", "filaments": []}}).encode()
bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", body)
await asyncio.sleep(0.01)
server.push_raw_to_clients.assert_awaited_once()
topic, _payload = server.push_raw_to_clients.await_args.args
assert topic == f"device/{VP_SERIAL}/report"
await bridge.stop()
# ---------------------------------------------------------------------------
# Forwarding: slicer → printer
# ---------------------------------------------------------------------------
class TestForwardToPrinter:
@pytest.mark.asyncio
async def test_forward_publishes_to_real_serial_request_topic(self):
target = _make_paho_client()
bridge = _make_bridge(_make_server(), target)
await bridge.start()
ok = bridge.forward_to_printer({"print": {"command": "stop"}})
assert ok is True
target.publish_raw.assert_called_once()
topic, payload = target.publish_raw.call_args.args
assert topic == f"device/{H2D_SERIAL}/request"
assert json.loads(payload) == {"print": {"command": "stop"}}
await bridge.stop()
@pytest.mark.asyncio
async def test_forward_returns_false_when_not_bound(self):
pm = MagicMock()
pm.get_client = MagicMock(return_value=None)
bridge = MQTTBridge(
vp_id=1,
vp_name="vp1",
vp_serial=VP_SERIAL,
target_printer_id=42,
mqtt_server=_make_server(),
printer_manager=pm,
)
await bridge.start()
assert bridge.forward_to_printer({"print": {"command": "stop"}}) is False
await bridge.stop()
# ---------------------------------------------------------------------------
# SimpleMQTTServer status response: cached-as-base
# ---------------------------------------------------------------------------
class TestStatusReportCachedAsBase:
"""`_send_status_report` sends near-byte-identical real data when bridge cache exists."""
def _capture_published(self, server: SimpleMQTTServer):
"""Wrap _publish_to_report to capture (topic, payload_dict)."""
published: list = []
async def _capture(writer, payload, serial=""):
published.append((serial or server.serial, payload))
server._publish_to_report = _capture # type: ignore[assignment]
return published
@pytest.mark.asyncio
async def test_uses_real_cache_when_bridge_active(self):
server = _make_server()
bridge = MagicMock()
bridge.get_latest_print_state.return_value = {
"command": "push_status",
"msg": 0,
"ams": {"ams": [{"id": "0"}]},
"device": {"extruder": {"info": [{"id": 0}, {"id": 1}]}},
"nozzle_diameter": "0.4",
"nozzle_type": "HH01", # real H2D value, not synthetic 'hardened_steel'
}
server.set_bridge(bridge)
published = self._capture_published(server)
await server._send_status_report(MagicMock())
assert len(published) == 1
_serial, payload = published[0]
# AMS / device / nozzle_type all from cache
assert payload["print"]["nozzle_type"] == "HH01"
assert payload["print"]["device"]["extruder"]["info"][1]["id"] == 1
# Protocol fields under our control
assert payload["print"]["command"] == "push_status"
assert payload["print"]["gcode_state"] == "IDLE"
@pytest.mark.asyncio
async def test_falls_back_to_synthetic_when_no_cache(self):
server = _make_server()
bridge = MagicMock()
bridge.get_latest_print_state.return_value = None
server.set_bridge(bridge)
published = self._capture_published(server)
await server._send_status_report(MagicMock())
assert len(published) == 1
_serial, payload = published[0]
# Synthetic baseline has stub fields like nozzle_type='hardened_steel'
# and a `storage` field that the real H2D doesn't push.
assert payload["print"]["nozzle_type"] == "hardened_steel"
assert "storage" in payload["print"]
@pytest.mark.asyncio
async def test_overrides_protocol_fields_even_when_cache_present(self):
"""Cached value's gcode_state must NOT win over our local upload-state-machine value."""
server = _make_server()
server._gcode_state = "PREPARE"
server._current_file = "foo.3mf"
bridge = MagicMock()
bridge.get_latest_print_state.return_value = {
"command": "push_status",
"gcode_state": "IDLE", # printer is idle; we are mid-FTP-upload
"gcode_file": "",
"gcode_file_prepare_percent": "0",
}
server.set_bridge(bridge)
published = self._capture_published(server)
await server._send_status_report(MagicMock())
_serial, payload = published[0]
assert payload["print"]["gcode_state"] == "PREPARE"
assert payload["print"]["gcode_file"] == "foo.3mf"
# ---------------------------------------------------------------------------
# Wire format
# ---------------------------------------------------------------------------
class TestWireFormat:
"""BambuStudio's Send pre-flight rejects compact JSON — must match real printer's
indented format (32K bytes for an idle H2D vs 14K compact)."""
@pytest.mark.asyncio
async def test_publish_uses_indent_4_json_format(self):
server = _make_server()
captured: list = []
async def _capture_drain():
pass
writer = MagicMock()
writer.write = lambda data: captured.append(data)
writer.drain = AsyncMock()
await server._publish_to_report(writer, {"print": {"command": "push_status", "ams": {}}})
body = b"".join(captured)
assert b'\n "print"' in body, "publish_to_report must use indent=4 JSON"
# ---------------------------------------------------------------------------
# Routing: _handle_publish
# ---------------------------------------------------------------------------
class TestPublishRouting:
"""Slicer-issued commands: project_file/gcode_file handled locally, everything
else forwarded to the real printer."""
def _build_publish_payload(self, topic: str, body: bytes) -> bytes:
topic_bytes = topic.encode("utf-8")
return bytes([len(topic_bytes) >> 8, len(topic_bytes) & 0xFF]) + topic_bytes + body
def _attach_active_bridge(self, server: SimpleMQTTServer) -> MagicMock:
bridge = MagicMock()
bridge.is_active = True
bridge.forward_to_printer = MagicMock(return_value=True)
server.set_bridge(bridge)
return bridge
@pytest.mark.asyncio
async def test_project_file_handled_locally_not_forwarded(self):
server = _make_server()
bridge = self._attach_active_bridge(server)
writer = MagicMock()
writer.write = MagicMock()
writer.drain = AsyncMock()
body = json.dumps({"print": {"command": "project_file", "subtask_name": "f", "sequence_id": "1"}}).encode()
payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body)
with patch.object(server, "_send_print_response", new=AsyncMock()) as mock_resp:
await server._handle_publish(0x30, payload, writer, "client1")
bridge.forward_to_printer.assert_not_called()
mock_resp.assert_awaited_once()
@pytest.mark.asyncio
async def test_gcode_file_handled_locally_not_forwarded(self):
server = _make_server()
bridge = self._attach_active_bridge(server)
writer = MagicMock()
writer.write = MagicMock()
writer.drain = AsyncMock()
body = json.dumps({"print": {"command": "gcode_file", "subtask_name": "f.gcode", "sequence_id": "1"}}).encode()
payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body)
with patch.object(server, "_send_print_response", new=AsyncMock()):
await server._handle_publish(0x30, payload, writer, "client1")
bridge.forward_to_printer.assert_not_called()
@pytest.mark.asyncio
async def test_pushall_handled_locally_not_forwarded(self):
server = _make_server()
bridge = self._attach_active_bridge(server)
writer = MagicMock()
writer.write = MagicMock()
writer.drain = AsyncMock()
body = json.dumps({"pushing": {"command": "pushall", "sequence_id": "0"}}).encode()
payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body)
with patch.object(server, "_send_status_report", new=AsyncMock()) as mock_status:
await server._handle_publish(0x30, payload, writer, "client1")
# Synthetic answer fires (fast, low latency); no forwarding (the
# cache already mirrors what the printer would respond with).
bridge.forward_to_printer.assert_not_called()
mock_status.assert_awaited_once()
@pytest.mark.asyncio
async def test_get_version_handled_locally_not_forwarded(self):
server = _make_server()
bridge = self._attach_active_bridge(server)
writer = MagicMock()
writer.write = MagicMock()
writer.drain = AsyncMock()
body = json.dumps({"info": {"command": "get_version", "sequence_id": "1"}}).encode()
payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body)
with patch.object(server, "_send_version_response", new=AsyncMock()) as mock_ver:
await server._handle_publish(0x30, payload, writer, "client1")
bridge.forward_to_printer.assert_not_called()
mock_ver.assert_awaited_once()
@pytest.mark.asyncio
async def test_extrusion_cali_get_is_forwarded(self):
"""extrusion_cali_get fetches per-filament k-profiles — must reach the printer."""
server = _make_server()
bridge = self._attach_active_bridge(server)
writer = MagicMock()
writer.write = MagicMock()
writer.drain = AsyncMock()
body = json.dumps(
{
"print": {
"command": "extrusion_cali_get",
"filament_id": "",
"nozzle_diameter": "0.4",
"sequence_id": "5",
}
}
).encode()
payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body)
await server._handle_publish(0x30, payload, writer, "client1")
bridge.forward_to_printer.assert_called_once()
forwarded = bridge.forward_to_printer.call_args.args[0]
assert forwarded["print"]["command"] == "extrusion_cali_get"
@pytest.mark.asyncio
async def test_print_stop_is_forwarded(self):
server = _make_server()
bridge = self._attach_active_bridge(server)
writer = MagicMock()
writer.write = MagicMock()
writer.drain = AsyncMock()
body = json.dumps({"print": {"command": "stop", "sequence_id": "5"}}).encode()
payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body)
await server._handle_publish(0x30, payload, writer, "client1")
bridge.forward_to_printer.assert_called_once()
# ---------------------------------------------------------------------------
# IP encoding helper
# ---------------------------------------------------------------------------
class TestIpEncoding:
def test_le_uint32_matches_real_h2d_capture(self):
# 192.168.255.133 captured from real H2D's net.info[0].ip = 2248124608
assert _ip_to_uint32_le("192.168.255.133") == 2248124608
def test_vp_ip_round_trip(self):
assert _ip_to_uint32_le("192.168.255.16") == 285190336
def test_invalid_ip_raises(self):
with pytest.raises(ValueError):
_ip_to_uint32_le("not.an.ip.actually")

View file

@ -10,8 +10,10 @@ services:
# Override with: PUID=$(id -u) PGID=$(id -g) docker compose up -d
user: "${PUID:-1000}:${PGID:-1000}"
#
# Proxy mode: allow binding to privileged ports (322, 990) as non-root user.
# Without this, the FTP and RTSP proxies silently fail.
# Allow binding to privileged ports (322, 990) as non-root user — required
# for FTPS in every VP mode and for the RTSPS camera proxy in proxy mode +
# non-proxy modes that have a target printer configured. Without this, the
# FTP and RTSP listeners silently fail.
cap_add:
- NET_BIND_SERVICE
#
@ -28,7 +30,7 @@ services:
# - "8883:8883" # Virtual printer MQTT
# - "990:990" # Virtual printer FTP control
# - "6000:6000" # Virtual printer file transfer tunnel
# - "322:322" # Virtual printer RTSP camera (X1/H2/P2)
# - "322:322" # Virtual printer RTSP camera (X1/H2/P2; proxy mode + non-proxy modes with a target printer)
# - "2024-2026:2024-2026" # Virtual printer proprietary ports (A1/P1S)
# - "50000-50100:50000-50100" # Virtual printer FTP passive data
volumes: