bambuddy/backend/app/services/virtual_printer/mqtt_bridge.py

"""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/<vp>_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