diff --git a/CHANGELOG.md b/CHANGELOG.md index 21c57f834..630a51f13 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ All notable changes to Bambuddy will be documented in this file. ## [0.2.4b2] - Unreleased ### Added +- **Virtual Printer non-proxy modes now mirror the live target printer to the slicer** ([#1193](https://github.com/maziggy/bambuddy/issues/1193) follow-up) — Until now, Immediate / Review / Print Queue VPs looked like a stub Bambu Lab printer to the slicer: AMS dropdowns were empty, no live state, no camera, no per-filament k-profile lookup. The user could send a sliced file and that was it. With this change, the VP fans out the **target printer's live MQTT state** to the slicer (AMS units, FTS / dual-extruder routing, nozzle, temps, k-profiles, AMS load / dry / calibration commands) and proxies the **camera RTSPS stream** on port 322 — so the slicer treats the VP as a fully-functional Bambu printer while Bambuddy's queue / archive / dispatch features stay in the loop. **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, see #1164). The VP's `_send_status_report` returns a near-byte-identical copy of the real push with only the upload-state-machine fields (sequence_id, command, msg, gcode_state, gcode_file, prepare_percent, subtask_name) overridden under our control, so BambuStudio's Send pre-flight sees exactly the same shape as a direct-to-printer connection. Command responses (extrusion_cali_get, AMS write acks, xcam responses) are fanned out raw — they carry sequence_ids the slicer is waiting on. Slicer-issued commands forward to the real printer except `print.project_file` / `gcode_file`, which are still answered locally because the file lives on Bambuddy. **Field-shape gotchas worth remembering:** (1) Real Bambu printers wire-format push_status JSON with `indent=4` (32 254 bytes for an idle H2D push, vs 14 268 bytes compact) — BambuStudio's Send pre-flight rejects compact JSON silently, so `_publish_to_report` was switched to `json.dumps(payload, indent=4)`. (2) `net.info[*].ip` (little-endian uint32, e.g. 192.168.255.133 → 2248124608) is the FTP destination IP BambuStudio uses for "Send to Printer storage" — it overrides anything else, including the URL hosts the rest of MQTT advertises. The bridge rewrites this to the VP's bind IP on cache, otherwise the slicer FTPs straight to the real printer and bypasses Bambuddy entirely (symptom: "Failed to send" with zero inbound FTP connections on the VP — debug-by-tcpdump if anyone hits it again). (3) `upgrade_state.sn` and any other nested-dict `sn` matching the target serial are rewritten to the VP serial; AMS-hardware serials (`n3f/0.sn` etc.) are left alone — those identify physical AMS units, not the device. (4) `ipcam.rtsp_url` is left unchanged: BambuStudio overrides the URL host with the device IP it bound on (the VP), so the slicer hits the VP's :322 RTSPS port — not the printer's directly. (5) For the slicer's RTSPS to reach the printer, the VP gets a raw `TCPProxy` on `:322 → :322` (same approach proxy mode uses; `cap_net_bind_service` was already in the systemd unit for FTP :990). (6) `extrusion_cali_get` is forwarded — answering it locally hides the user's stored k-profiles. **Setup nuance for camera:** because the slicer authenticates against the printer's RTSPS with whatever access code is in its profile, the VP's access code must match the target printer's access code for the camera path to authenticate. This is a one-time configuration step (Settings → Virtual Printer → set access code = target printer's LAN code, then re-add the VP in Bambu Studio / Orca Slicer). MQTT and FTP work either way; only camera needs the match because RTSPS auth happens between the slicer and the real printer's broker. **Tested e2e** with both BambuStudio and OrcaSlicer against H2D (dual-nozzle, AMS 2 Pro + AMS HT) and X1C (single-nozzle, AMS) across all three non-proxy modes (Immediate / Review / Print Queue) — sync, send, k-profile lookup, AMS configuration from slicer, and live camera all work. **Files:** new `backend/app/services/virtual_printer/mqtt_bridge.py` (caches push_status / get_version, forwards slicer commands, fans out command responses, rewrites identity fields including `net.info[*].ip` LE uint32); `bambu_mqtt.py` gains `register_raw_message_handler` / `unregister_raw_message_handler` / `publish_raw` so the bridge can subscribe to Bambuddy's existing per-printer paho subscription without opening a second session; `mqtt_server.py` switches `_send_status_report` and `_send_version_response` to cached-as-base when the bridge has data, falls back to the original synthetic stubs otherwise; `manager.py` wires the bridge + a raw `TCPProxy` for RTSPS into `start_server` for non-proxy modes whenever a target printer is configured. 25 new tests in `test_vp_mqtt_bridge.py` pin the contract: lifecycle, push_status caching, serial / IP rewriting, get_version-modules cache, selective fan-out (only command responses, never push_status itself), wire format must use `indent=4`, routing of slicer-issued commands (project_file / gcode_file local; everything else forwarded), and the IP-encoding helper against captures from real H2D pushes. Proxy mode is untouched — `SlicerProxyManager` still owns its own MQTT/FTP/RTSP/Bind/Aux proxies in proxy mode and never instantiates `SimpleMQTTServer` or `MQTTBridge`. - **AMS slot Load / Unload from the printer card** ([#891](https://github.com/maziggy/bambuddy/issues/891), reported by @NNeerr00, +1 from @cadtoolbox) — The MQTT primitives for "load filament from a tray" and "unload the currently loaded tray" already existed in `bambu_mqtt.py` (reverse-engineered from BambuStudio captures, including the H2D dual-extruder right-external case captured fresh during this work) but were unused — there was no HTTP route and no UI. Net effect: every Load / Unload had to happen on the printer touchscreen, and external-spool users on dual-nozzle H2D had no way to drive Ext-R from the desktop at all. **Backend:** new `POST /printers/{id}/ams/load?tray_id={int}` and `POST /printers/{id}/ams/unload`, both gated on `Permission.PRINTERS_CONTROL`. The load route validates `tray_id ∈ {0..15, 254, 255}` (AMS slots, single-external/Ext-L, Ext-R respectively) and returns a human-readable target in the success message ("AMS 0 slot 1", "external spool", "Ext-R") so the UI toast tells the user which spool the printer is now feeding from. **MQTT primitive update:** `ams_load_filament` gains a third encoding branch for `tray_id=255` matching the BambuStudio capture verbatim — `ams_id=255, slot_id=0` (the right-extruder index, **not** a slot index — Bambu's load command on dual-extruder externals encodes the destination extruder, not the source slot), `target=255`, and `curr_temp = tar_temp = right-nozzle temp` (read from `state.temperatures["nozzle_2"]`, falling back to 215 °C if the right nozzle is cold or unknown — the printer rejects nonsensical temps, so a warm fallback is safer than `-1`). The existing `tray_id=254` branch is preserved verbatim (`slot_id=254, curr/tar=-1`) since that came from a single-extruder capture and is known to work; no risk of regression on existing single-external setups. **UI:** the existing AMS slot popover (the one with "Re-read RFID") gains two new entries — "Load" (posts `tray_id = ams.id * 4 + slotIdx`) and "Unload" (no params, global on the currently-loaded slot). The external spool slot — which had no popover at all before — gets one with the same Load + Unload entries, and on dual-nozzle H2D each external slot (Ext-L tray_id=254, Ext-R tray_id=255) drives its own extruder. The menu is hidden while `state === 'RUNNING'` (parallels the existing RFID re-read gating). **i18n:** `printers.ams.load`, `printers.ams.unload`, plus four new toast strings (`loadInitiated`, `unloadInitiated`, `failedToLoad`, `failedToUnload`) added to all 8 locales — English fully translated, German fully translated, the other 6 locales seeded with English copy pending native translation (matches the project's existing flow for newly-added user-facing features). 16 new tests pin the contract: 5 unit tests in `test_bambu_mqtt.py::TestAmsLoadFilamentEncoding` (AMS slot encoding, Ext-L preserves legacy capture, Ext-R uses the new captured shape with actual right-nozzle temp, Ext-R falls back to 215 °C when cold, disconnected client doesn't publish); 11 integration tests in `test_printers_api.py::TestAMSLoadUnloadAPI` (load: invalid tray_id 400, not-found 404, not-connected 400, AMS slot success with derived `ams_id*4+slot` math, Ext-L success, Ext-R success, MQTT failure 500; unload: not-found, not-connected, success, MQTT failure 500); 4 frontend tests in `PrintersPageAmsLoadUnload.test.tsx` (Load posts the right tray_id, Unload posts with no params, menu hidden while RUNNING, external spool's tray_id=254 round-trips through the route). - **API keys can read Bambu Cloud presets on the owner's behalf** ([#1182](https://github.com/maziggy/bambuddy/issues/1182), reported by @turulix) — Tim is building a fully automated headless slicing pipeline against Bambuddy's API and hit the wall flagged in the previous round of cloud-auth work ([#665](https://github.com/maziggy/bambuddy/issues/665)): `/cloud/*` routes resolve `cloud_token` per-user from `User.cloud_token`, but the auth gate (`require_permission_if_auth_enabled`, `auth.py:856`) returned `None` for API-keyed requests, so the route fell back to the global `Settings`-table token, which only carries a value in *auth-disabled* deployments. Net effect on auth-enabled deployments: API keys reached the gate just fine, then `/cloud/filaments` always saw `user=None`, called `get_stored_token(db, None)` against an empty Settings table, and returned 401 / empty results — no path to read the slicer presets, filament catalogue, or device list that a CLI workflow needs. The data model treated API keys as standalone tokens with no owner (`APIKey` had `id`, `name`, `key_hash`, scope flags, and `printer_ids` — no `user_id`), so even if the gate wanted to delegate the cloud lookup, there was no User to delegate to. **The fix:** make API keys carry an owner, route /cloud/* lookups through that owner, and gate the new capability behind an explicit opt-in scope so existing automation doesn't gain cloud-read access on upgrade. Concretely: (1) `APIKey` gains `user_id` (FK to `users.id`, ON DELETE CASCADE — Postgres enforces, SQLite plus an explicit `DELETE FROM api_keys WHERE user_id = ?` in the user-delete route since SQLite ships FK enforcement off; the project's existing pattern at `users.py:397-406` for `created_by_id` cleanup) and `can_access_cloud` (BOOLEAN DEFAULT 0 — opt-in, never set on legacy rows). (2) The auth gate now returns the owner User when it validates an API key with `user_id` set, so `/cloud/*` routes naturally resolve `user.cloud_token` the same way they do for JWT-authed sessions. Permission semantics are preserved — API keys still bypass the per-route permission check (their scopes live on the row itself), the User return is *only* so cloud-aware routes can read per-user state. Legacy ownerless keys (`user_id IS NULL`) keep returning None, stay anonymous, and continue working against every non-cloud route exactly as before. (3) A router-level dependency on the `/cloud/*` `APIRouter` enforces three independent fences for API-keyed callers: `user_id IS NOT NULL` (legacy keys → 401 with "recreate it from Settings → API Keys" — explicit recreate path rather than silently degrading), `can_access_cloud=True` (otherwise 403 with "Enable 'Allow cloud access' on the key"), and `build_authenticated_cloud` returning a service (otherwise 401 with the existing token-not-set error — unchanged for JWT flow). The router-level dep duplicates the API-key validation done by the regular auth gate (router-level deps run before route-level deps in FastAPI, so `request.state` isn't populated yet) — the cost is one extra `SELECT FROM api_keys` per cloud request, bounded and cheap with the `key_prefix` index. (4) The create route stamps `user_id = current_user.id` from the creator and rejects `can_access_cloud=True` when auth is disabled (no per-user `cloud_token` storage exists in that mode — fail loudly at create time rather than silently producing a non-functional key). PATCH route rejects flipping `can_access_cloud` to True on a legacy ownerless key for the same reason — force recreate. (5) `APIKeyResponse` exposes `user_id` so the UI can show ownership at a glance: a "Cloud" badge for cloud-enabled keys and a "Legacy" badge with hover tooltip ("Created before per-user ownership; recreate to use cloud access") for ownerless rows. The form gains an "Allow cloud access" checkbox, default off. **Migration:** two idempotent `ALTER TABLE api_keys ADD COLUMN` (`user_id INTEGER REFERENCES users(id) ON DELETE CASCADE` and `can_access_cloud BOOLEAN DEFAULT 0`) plus an index on `user_id` for the auth-gate's owner→keys lookup that runs on every API-keyed request. **i18n:** 5 new keys (`settings.cloudAccess`, `settings.cloudAccessDescription`, `settings.cloudBadge`, `settings.legacyKey`, `settings.legacyKeyTooltip`) added to all 8 locales — English fully translated, German fully translated, the other 6 locales seeded with English copies pending native translation (matches the project's existing flow for newly-added user-facing features). 9 backend integration tests in `test_api_key_cloud_access.py`: create stamps owner + cloud flag, defaults off when not asked for, rejected when auth disabled (no per-user storage), PATCH rejected on legacy keys; cloud router rejects legacy keys with the recreate copy, rejects owned-but-no-cloud-flag keys with the enable-cloud-access copy, lets owned-and-flagged keys through with owner's `cloud_token` in the response, JWT callers unaffected (gate is no-op for non-API-keyed); user-delete CASCADEs the API keys via the explicit DELETE in the route. 2 frontend SettingsPage tests pin the badge rendering matrix (Cloud badge present on `can_access_cloud=true`, Legacy badge present on `user_id=null`, neither rendered on a normal owned non-cloud key) and the create-form contract (toggling "Allow cloud access" results in `can_access_cloud=true` in the POST body). Permission semantics for the new fence are the only behavioural change for existing API keys: keys created before this release become "legacy" rows and are rejected at /cloud/* with the recreate message; every other endpoint they were used against — queue, status, control — is untouched. diff --git a/README.md b/README.md index 4cd9d69b6..8df28935a 100644 --- a/README.md +++ b/README.md @@ -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.) diff --git a/backend/app/main.py b/backend/app/main.py index cb3e18913..61f6c0053 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -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") diff --git a/backend/app/services/bambu_mqtt.py b/backend/app/services/bambu_mqtt.py index 648f46b86..6b03fd37a 100644 --- a/backend/app/services/bambu_mqtt.py +++ b/backend/app/services/bambu_mqtt.py @@ -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 ): diff --git a/backend/app/services/virtual_printer/manager.py b/backend/app/services/virtual_printer/manager.py index 9b543dfe6..cba314d4b 100644 --- a/backend/app/services/virtual_printer/manager.py +++ b/backend/app/services/virtual_printer/manager.py @@ -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 :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() diff --git a/backend/app/services/virtual_printer/mqtt_bridge.py b/backend/app/services/virtual_printer/mqtt_bridge.py new file mode 100644 index 000000000..da62177bc --- /dev/null +++ b/backend/app/services/virtual_printer/mqtt_bridge.py @@ -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 diff --git a/backend/app/services/virtual_printer/mqtt_server.py b/backend/app/services/virtual_printer/mqtt_server.py index c50c25ac9..dde9ae1f0 100644 --- a/backend/app/services/virtual_printer/mqtt_server.py +++ b/backend/app/services/virtual_printer/mqtt_server.py @@ -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) diff --git a/backend/tests/unit/test_vp_mqtt_bridge.py b/backend/tests/unit/test_vp_mqtt_bridge.py new file mode 100644 index 000000000..77f509a89 --- /dev/null +++ b/backend/tests/unit/test_vp_mqtt_bridge.py @@ -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") diff --git a/docker-compose.yml b/docker-compose.yml index 06b4ac0a1..7a2bb5dd3 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -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: