"""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 → the address a slicer can reach Bambuddy on. BambuStudio reads this as the FTP destination IP. Without this the slicer FTPs straight to the real printer and bypasses Bambuddy. Normally that address is the VP bind IP; `VIRTUAL_PRINTER_ADVERTISE_ADDRESS` overrides it for NAT'd deployments (see `ADVERTISE_ADDRESS_ENV`). - `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 ipaddress import json import logging import os import socket from typing import TYPE_CHECKING from backend.app.services.bambu_mqtt import apply_tray_exist_bits from backend.app.services.virtual_printer._debug import append_event, dump_wire 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 # Opt-in override for the address written into `net.info[].ip`. Needed only # where the address a slicer has to use to reach Bambuddy is not one of the # container's own interfaces — Docker bridge networking being the case that # prompted it (#2930), where the bind address is a container-private IP like # `172.24.0.2` and a slicer that follows it opens an FTP connection to # nothing. Host and macvlan networking stay the supported modes and need # nothing set here. # # Deliberately an environment variable rather than a change to how the VP IP # is resolved: the alternative was to prefer the VP's "Network Interface # Override" (`remote_interface_ip`), which today feeds SSDP and the cert SANs # only. Reading it here would silently move the FTP destination for every # install that has it set — the multi-NIC, VLAN and Tailscale setups, i.e. # exactly the ones most likely to have been tuned by hand. Unset, this # variable changes nothing. Mirrors `VIRTUAL_PRINTER_PASV_ADDRESS`, which # exists for the same reason on the FTP side. ADVERTISE_ADDRESS_ENV = "VIRTUAL_PRINTER_ADVERTISE_ADDRESS" # Bambuddy's internal printer state in bambu_mqtt.py (around line 2686+) is # updated per-field — each `if "X" in data: self.state.X = ...` block leaves # every other field untouched, so the state accumulates everything the # printer has ever sent. The bridge cache below mirrors that pattern: when # the incoming push_status omits a field, the previous value is preserved # verbatim; only fields actually present in the new push overwrite. This # stops capability/lifecycle fields (cali_version, print_type, mc_print_stage, # device, ...) draining out of the cache between pushalls, which surfaced # as #1622 (BambuStudio's Device-tab UIs greying out on P1S after the # cache drained to a thin incremental snapshot). The `ams` field still # gets unit-/tray-level deep merge via `_merge_ams_dict` because firmware # sends partial `ams` blobs under the same key (#1387). 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) def _resolve_target_to_ipv4(target: str) -> str | None: """Return a dotted-quad IPv4 for `target`, resolving hostnames if needed. The printer client may be configured by IPv4 *or* by hostname/FQDN (e.g. `p1s.fritz.box`) — the latter is common on home LANs with a DNS-providing router. The downstream `net.info[].ip` field is a 32-bit little-endian integer though, so a hostname can't round-trip through it; we have to pick *one* concrete IPv4 to write in. Returns None if `target` is empty, not parseable as IPv4, and DNS resolution fails — caller logs that as the not-armed reason and re-tries on the next refresh tick (DHCP/DNS churn picks itself up). """ if not target: return None try: return str(ipaddress.IPv4Address(target)) except (ValueError, ipaddress.AddressValueError): pass try: # AF_INET filters to IPv4 only; the rewrite field is uint32 LE, # there's no IPv6 representation that fits. infos = socket.getaddrinfo(target, None, family=socket.AF_INET) except OSError: return None for info in infos: sockaddr = info[4] if sockaddr and isinstance(sockaddr[0], str): return sockaddr[0] return None def _resolve_host_interface_for_target(target_ip: str) -> str | None: """Pick a host-side IPv4 for `net.info[].ip` when the VP has no dedicated bind IP. Used when `mqtt_server.bind_address` is empty or 0.0.0.0 — the listener accepts on every interface but we still need ONE concrete IPv4 to write into the rewritten `net.info[].ip` field so the slicer's FTP target resolves to Bambuddy rather than the real printer. Returns the IPv4 of the host interface that shares a subnet with the printer (best fit because the slicer is typically on the same LAN as the printer), or None if no interface matches — in which case the bridge leaves encoding unarmed and the previous (still-leaky) behaviour stands. """ try: from backend.app.services.network_utils import find_interface_for_ip except Exception: # pragma: no cover - import shielding return None try: iface = find_interface_for_ip(target_ip) except Exception: logger.exception("MQTT bridge: find_interface_for_ip(%s) crashed", target_ip) return None if not iface: return None ip = iface.get("ip") return ip if isinstance(ip, str) and ip else None def _resolve_advertise_override(vp_name: str) -> str: """Return the validated `net.info[].ip` override from the environment, or "". Validated here rather than on each refresh tick for two reasons: a typo produces one warning instead of one every 30s, and an unusable value falls back to the bind address instead of leaving the rewrite unarmed. That second part matters — an unarmed rewrite puts the *real printer IP* back in front of the slicer (#1429), so a mistyped override must not be able to reopen the leak this whole path exists to close. """ raw = os.environ.get(ADVERTISE_ADDRESS_ENV, "").strip() if not raw: return "" try: _ip_to_uint32_le(raw) except ValueError: logger.warning( "[%s] %s=%r is not a dotted-quad IPv4 — ignoring it, using the VP bind address instead", vp_name, ADVERTISE_ADDRESS_ENV, raw, ) return "" return raw def _merge_ams_dict(prev_ams: dict, new_ams: dict) -> dict: """Merge a new ``ams`` blob from an incremental push onto the previous one. Bambu firmware sends three shapes for the ``ams`` field on push_status: 1. Full pushall (after a printer reconnect or explicit pushall request): ``{ams: [{id, tray: [{id, tray_type, ...}, ...]}, ...], ams_status, ams_exist_bits, ...}`` — every unit + every tray populated. 2. Status-only incremental: ``{ams_status: 1}`` or ``{humidity: 30}`` — no ``ams`` array at all. Bambuddy logs these as "AMS partial update (no tray data)" (#784 vintage). 3. Tray-targeted incremental during a print: ``{ams: [{id: 0, tray: [{id: 0, state: 11}]}]}`` — only the units / trays whose state changed. Replacing the cached ``ams`` wholesale on shapes (2) and (3) is what made the slicer "lose" AMS between pushalls and trip the symptom in #1387: the slicer would see a stripped ``ams_status``-only blob and fall back to its "no AMS" default render. This merge mirrors the deep-merge logic in ``bambu_mqtt.py::_handle_ams_data`` at the bridge layer so the slicer-facing cache always carries the latest known coherent state. Strategy: - Shallow-merge top-level scalars: keys in ``new`` win; keys only in ``prev`` are preserved. - For the ``ams`` array (list of units): match by ``id``. Units only in ``prev`` survive. Units in ``new`` overlay onto their ``prev`` counterpart; same recursion applies to each unit's ``tray`` array by tray ``id``. """ merged = dict(prev_ams) for k, v in new_ams.items(): if k != "ams": merged[k] = v prev_units = prev_ams.get("ams") if isinstance(prev_ams.get("ams"), list) else [] new_units = new_ams.get("ams") if isinstance(new_ams.get("ams"), list) else None if new_units is None: # Shape (2): no ``ams`` array in the incremental — keep prev's units. if prev_units: merged["ams"] = prev_units return merged prev_by_id = {u.get("id"): u for u in prev_units if isinstance(u, dict) and u.get("id") is not None} merged_units: list = [] seen_ids: set = set() for new_unit in new_units: if not isinstance(new_unit, dict): merged_units.append(new_unit) continue uid = new_unit.get("id") prev_unit = prev_by_id.get(uid) if uid is not None else None if prev_unit is None: merged_units.append(new_unit) if uid is not None: seen_ids.add(uid) continue # Shallow-merge unit fields; preserve prev's trays not present in new. merged_unit = dict(prev_unit) for k, v in new_unit.items(): if k != "tray": merged_unit[k] = v new_trays = new_unit.get("tray") if isinstance(new_unit.get("tray"), list) else None if new_trays is None: # Unit-level partial — keep prev's tray list intact. pass else: prev_trays = prev_unit.get("tray") if isinstance(prev_unit.get("tray"), list) else [] prev_trays_by_id = {t.get("id"): t for t in prev_trays if isinstance(t, dict) and t.get("id") is not None} merged_trays: list = [] seen_tray_ids: set = set() for new_tray in new_trays: if not isinstance(new_tray, dict): merged_trays.append(new_tray) continue tid = new_tray.get("id") prev_tray = prev_trays_by_id.get(tid) if tid is not None else None if prev_tray is None: merged_trays.append(new_tray) else: merged_tray = dict(prev_tray) merged_tray.update(new_tray) merged_trays.append(merged_tray) if tid is not None: seen_tray_ids.add(tid) # Preserve prev trays not mentioned in the incremental. for tid, prev_tray in prev_trays_by_id.items(): if tid not in seen_tray_ids: merged_trays.append(prev_tray) merged_unit["tray"] = merged_trays merged_units.append(merged_unit) if uid is not None: seen_ids.add(uid) # Preserve prev units not mentioned in the incremental. for uid, prev_unit in prev_by_id.items(): if uid not in seen_ids: merged_units.append(prev_unit) merged["ams"] = merged_units return merged 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 # Last reason `_refresh_ip_encoding` early-returned without arming. # Used to throttle the "NOT armed" diagnostic log to one line per # state change — refresh runs every 30s, so without throttling an # idle-but-unarmed bridge would emit one line per tick forever. Set # to None once arming succeeds so the next failure re-logs. #1429 # follow-up: makes silent early-returns visible without grepping the # source. self._not_armed_reason: str | None = None # NAT escape hatch for `net.info[].ip`, resolved once — the process # environment cannot change without a restart. "" means "use the VP # bind address", which is every install that has not set it. self._advertise_address = _resolve_advertise_override(vp_name) 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. On crash exit, the handler must be unbound — otherwise the registered ``_on_printer_raw`` keeps firing on every real-printer message even though the bridge is functionally dead (memory leak + behaviour leak across VP restart). """ 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) # Crash exit — unbind so the orphaned handler stops firing. # ``stop()`` won't be invoked because the task completes done-not-cancelled. self._unbind_client() 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: # Same client object — but `ip_address` can fill in *after* the # initial bind (e.g. DB row had a stale/empty value until the # client's first SSDP-driven IP refresh). The original code only # encoded `_target_ip_uint32_le` on client-identity change, so # that late-arriving IP was never picked up, the `net.info[*].ip` # rewrite stayed disabled, and the cache filled with the real # printer IP — #1429. Refresh the encoding every tick so it # self-heals once `ip_address` becomes valid. self._refresh_ip_encoding() 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) self._refresh_ip_encoding() 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). # # The bind frequently races the real printer's MQTT TLS handshake — a # slicer-side reconnect re-resolves the client before the underlying # session has reconnected, especially on A1 firmware where the bridge # cycles more aggressively (#1721). When that happens, the nudge is a # no-op — the next periodic pushall populates the cache anyway — but # `request_status_update` logs WARNING on the not-connected return path # and pollutes every support bundle with a benign line. # # Gate both nudges on the client being actually connected. The fall- # through path is unchanged: when the client comes up, the next # `_resolve_client` tick re-enters this branch on identity change OR # the periodic pushall in `bambu_mqtt.py` fills the cache. client_connected = bool(getattr(getattr(current, "state", None), "connected", False)) if not client_connected: logger.debug( "[%s] MQTT bridge: post-bind nudge skipped (printer client not connected yet)", self.vp_name, ) else: 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 _refresh_ip_encoding(self) -> None: """(Re-)encode `_target_ip_uint32_le` / `_vp_ip_uint32_le` from current values. Called on every refresh tick, not just on client-identity change, so a late-arriving printer IP (or a bind-address change) is picked up without restarting the VP. When the encoding becomes valid for the first time *after* the cache already received a push with the real printer IP, also sweep the existing cache so the slicer's next pull sees the rewritten value (#1429). Without this sweep the sticky-key preservation keeps the poisoned `net.info[].ip` alive forever. VP IP resolution, in order: the `VIRTUAL_PRINTER_ADVERTISE_ADDRESS` override if one is set (NAT'd deployments where no local interface carries the address slicers use — see `ADVERTISE_ADDRESS_ENV`), then `mqtt_server.bind_address`, then — when that is empty or `0.0.0.0`, the default for VPs never assigned a dedicated bind IP — the host interface sharing a subnet with the printer's IP. Without that last fallback the rewrite never arms on a default-config flat-LAN install and `net.info[].ip` leaks the real printer IP — the slicer follows it on Send (#1429 residual). """ def _log_not_armed(reason: str) -> None: # Throttle: only log when the reason changes, otherwise an idle # unarmed bridge would emit one INFO line every refresh tick # (~30s) forever. Cleared on arm so a regression re-logs. if reason != self._not_armed_reason: logger.info("[%s] MQTT bridge IP encoding NOT armed: %s", self.vp_name, reason) self._not_armed_reason = reason client = self._target_client if client is None: _log_not_armed("target_client is None (bridge not bound to a printer)") return configured_target = getattr(client, "ip_address", None) if not configured_target: _log_not_armed("printer client has no ip_address yet") return # Printers configured by hostname/FQDN (e.g. `p1s.fritz.box`) need to # be resolved to an IPv4 before encoding: net.info[*].ip is uint32 LE # and can't carry a hostname (#1429 follow-up). target_ip = _resolve_target_to_ipv4(configured_target) if not target_ip: _log_not_armed( f"could not resolve printer host {configured_target!r} to IPv4 (invalid address and DNS lookup failed)" ) return if self._advertise_address: vp_ip = self._advertise_address vp_ip_source = ADVERTISE_ADDRESS_ENV else: vp_ip = getattr(self._mqtt_server, "bind_address", None) vp_ip_source = "bind_address" if not vp_ip or vp_ip in ("0.0.0.0", ""): # nosec B104 resolved = _resolve_host_interface_for_target(target_ip) if not resolved: _log_not_armed( f"no host interface shares a subnet with printer IP {target_ip} " f"(and VP bind_address is 0.0.0.0/empty) — set {ADVERTISE_ADDRESS_ENV} " "to the address slicers reach Bambuddy on if this host is NAT'd" ) return vp_ip = resolved vp_ip_source = "auto-resolved" try: new_target_le = _ip_to_uint32_le(target_ip) new_vp_le = _ip_to_uint32_le(vp_ip) except ValueError as e: _log_not_armed(f"invalid IPv4 (target={target_ip!r}, vp={vp_ip!r}): {e}") return if new_target_le == self._target_ip_uint32_le and new_vp_le == self._vp_ip_uint32_le: return # No change — nothing to do. # Encoding either became valid for the first time or shifted (DHCP # renewal, bind_ip reconfigured, etc.). Update + sweep the cache. was_armed = self._target_ip_uint32_le is not None and self._vp_ip_uint32_le is not None self._target_ip_uint32_le = new_target_le self._vp_ip_uint32_le = new_vp_le # Clear the dedup so a future failure re-emits the diagnostic line. self._not_armed_reason = None target_display = target_ip if target_ip == configured_target else f"{configured_target}→{target_ip}" logger.info( "[%s] MQTT bridge IP encoding %s: target=%s vp=%s (%s)", self.vp_name, "updated" if was_armed else "armed", target_display, vp_ip, vp_ip_source, ) cached = self._latest_print_state if isinstance(cached, dict): n = self._rewrite_net_info_ips(cached) if n: logger.info( "[%s] MQTT bridge swept %d net.info[].ip entries in cached push", self.vp_name, n, ) def _rewrite_net_info_ips(self, print_state: dict) -> int: """Rewrite every non-zero `net.info[].ip` in `print_state` to the VP's IP. Returns the number of entries rewritten. Mutates `print_state` in place. Strategy: rewrite ALL entries with a non-zero `ip`, not only those matching `_target_ip_uint32_le`. Real printers (X1C, H2D Pro) can report multiple active interfaces (WiFi + Ethernet) with different IPs — only one matches the IP Bambuddy tracks, but the slicer may read any of them. Leaving non-matching entries pointing at real printer interfaces leaks an FTP fallback path that bypasses the VP (the #1429 / #1302 symptom). Entries with `ip == 0` are placeholders for unpopulated interfaces — leave them alone so the slicer's "active interface" detection still recognises them as absent. """ if self._vp_ip_uint32_le is None: return 0 net = print_state.get("net") if not isinstance(net, dict): return 0 info = net.get("info") if not isinstance(info, list): return 0 rewritten = 0 for entry in info: if not isinstance(entry, dict): continue ip_value = entry.get("ip") if not isinstance(ip_value, int) or ip_value == 0: continue if ip_value == self._vp_ip_uint32_le: continue entry["ip"] = self._vp_ip_uint32_le rewritten += 1 return rewritten 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 → the VP's IP in `net.info[*].ip` so the # slicer's FTP destination resolves to the VP, not the real printer. self._rewrite_net_info_ips(print_data) # Defensive deep copy on store so the cache is fully decoupled from # the freshly-parsed tree and from any reader's reference. new_state = copy.deepcopy(print_data) # Bambu firmware sends two kinds of push_status: full pushall # responses (on `pushall` requests / printer reconnect) which # include the full top-level field set (AMS, vt_tray, net, # cali_version, print_type, mc_print_stage, device, ...) — and # ~1 Hz incrementals with just the fields that changed (temps, # fan, wifi, status). Carry over every prev field the incoming # push doesn't overwrite, mirroring the per-field accumulate # pattern in bambu_mqtt.py's internal state handler — without # this the cache thins out to whatever the latest incremental # carried (~17 keys on P1S in #1622), and the slicer's Device- # tab capability gates (manage-calibration, AMS-assign dropdown, # …) flip off because their gating fields drained from the # cache. The deep-copy is defensive: without it the carried- # over nested dicts/lists are shared with the previous cache, # so any in-place mutation later would corrupt both. prev = self._latest_print_state if prev is not None: for prev_key, prev_value in prev.items(): if prev_key not in new_state: new_state[prev_key] = copy.deepcopy(prev_value) # Firmware sends partial `ams` blobs (status-only / unit- # targeted / tray-targeted) under the same key on # incremental updates, which would overwrite the cached # full blob and break the slicer's AMS render (#1387 / # #1371). Deep-merge mirrors what bambu_mqtt.py does # internally in `_handle_ams_data`. if isinstance(new_state.get("ams"), dict) and isinstance(prev.get("ams"), dict): new_state["ams"] = _merge_ams_dict(prev["ams"], new_state["ams"]) # Same per-field accumulate rule applied one level deeper for # other top-level dict-shaped fields. Firmware sends partial # `vt_tray` (external spool) updates right after a slicer # `ams_filament_setting` pick — typically just `{tray_info_idx, # tray_color}`, dropping the ~18 other fields (`tray_type`, # `state`, `remain`, `k`, `n`, `cali_idx`, `nozzle_temp_min/max`, # `tray_uuid`, `xcam_info`, ...) the slicer needs to render the # slot. Without overlay the next 1 Hz cached-as-base push # delivered the stripped dict and the slicer rendered the # external slot as "invalid" until a reload triggered a fresh # pushall (#1622 round 5, reported by @shaddowlink). AMS slots # didn't suffer because `_merge_ams_dict` deep-merges per tray. # Same shape covers `device`, `online`, `upgrade_state`, `ipcam`, # `upload`, `net`, ... against future firmware partials too. # `ams` is excluded — already deep-merged above. for key, new_value in list(new_state.items()): if key == "ams": continue prev_value = prev.get(key) if isinstance(prev_value, dict) and isinstance(new_value, dict): merged = dict(prev_value) merged.update(new_value) new_state[key] = merged # Apply empty-slot cleanup on the merged AMS so the slicer-facing # cache mirrors what Bambuddy's AMS card shows internally. Without # this the cached units carry stale per-tray filament fields for # slots whose `tray_exist_bits` bit is 0, and BambuStudio's Sync # paints those empty slots as phantom loaded filaments (#1726). # Runs whether or not a prev cache existed — fresh pushalls also # carry tray_exist_bits and benefit from the cleanup. # These units carry the RAW firmware ids — this cache is what the # slicer sees, and BambuStudio addresses the A2L's AMS-Lite as the # physical id 16 (it sends `ams_get_rfid {ams_id: 16}` through the # VP), so we must not normalise them to 6 the way Bambuddy's # internal state does. `apply_tray_exist_bits` folds 16 onto the # same bit base internally instead (#2697). merged_ams_dict = new_state.get("ams") if isinstance(merged_ams_dict, dict): units = merged_ams_dict.get("ams") apply_tray_exist_bits( units if isinstance(units, list) else [], merged_ams_dict.get("tray_exist_bits"), power_on_flag=merged_ams_dict.get("power_on_flag", True), log_label=self.vp_name, ) self._latest_print_state = new_state dump_wire(self.vp_name, "in", new_state) 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}" # Env-flagged command trace (#1622): every printer-originated response # that gets fanned to the slicer (extrusion_cali_get / ams write acks / # xcam / system / etc.) gets a line in vp_wire/_cmd.jsonl. Pair # with the slicer-side publishes captured in mqtt_server. Off by # default. Capture AFTER serial rewrite so the dump matches what the # slicer actually sees on the wire. append_event(self.vp_name, "printer_to_slicer", vp_topic, payload) 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