bambuddy/backend/app/services/slot_nozzle.py

"""Which nozzle does this AMS slot feed, and how wide is it?

Every path that configures a slot needs the same two facts: the extruder the
slot feeds, and that nozzle's diameter. Both the filament preset and the K
profile are stored per nozzle diameter, so getting the diameter wrong silently
selects the wrong preset *and* the wrong K value -- and before this module the
answer was worked out independently in seven places, each with ``nozzles[0]``
hard-coded as the diameter for every slot on the machine.

``nozzles[0]`` is correct on a single-nozzle printer and correct on a
dual-nozzle printer with the same size fitted both sides, which is why it has
survived. It is wrong the moment someone fits a 0.4 and a 0.2, which is exactly
the machine this feature exists for.

## Which array index belongs to which extruder

``PrinterState.nozzles`` is filled by two different MQTT parsers that use
opposite conventions, and this module is where that is resolved once:

* The **H2/X2 path** (``bambu_mqtt`` ~5100) writes ``nozzles[nozzle["id"]]``
  straight from ``device.nozzle.info``, i.e. indexed by physical nozzle id.
* The **legacy path** (~5013) writes left -> ``nozzles[0]``, right ->
  ``nozzles[1]``, which is the reverse of the extruder ids (extruder 0 is the
  RIGHT hotend).

**MEASURED 2026-08-27 on an H2D with 0.4 high flow LEFT and 0.6 high flow
RIGHT: ``nozzles[0]`` read 0.6 -- the right hotend, which is extruder 0.** So
the array is indexed by extruder id, and the H2 convention (physical nozzle id N
sits on extruder N) is the one that holds.

The legacy branch cannot govern a real dual-nozzle machine anyway: every model
in ``DUAL_NOZZLE_MODELS`` is H2-series or X2D, all of which report
``device.nozzle.info``, and ``left_nozzle_diameter`` appears nowhere in any
captured log or wire trace. On a single-nozzle printer both conventions agree
that index 0 is the only nozzle.

The distinction is invisible on a machine with matching nozzles, since both
conventions then return the same string -- which is why it went unnoticed for so
long, and why this is the single place to change if a future model contradicts
it.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass

from backend.app.utils.fts_routing import slot_extruder
from backend.app.utils.printer_models import is_dual_nozzle_model
from backend.app.utils.tray_ids import TrayLayout

logger = logging.getLogger(__name__)

# What a printer that has told us nothing is assumed to have fitted. Matches
# the default every call site used before this module existed.
DEFAULT_NOZZLE_DIAMETER = "0.4"


@dataclass(frozen=True)
class SlotNozzle:
    """The nozzle an AMS slot feeds."""

    # None when the printer has not said which extruder this slot feeds. Callers
    # that must have a number use ``extruder_or_default``; callers that store a
    # row keep the None so "unknown" is not written as "the right-hand nozzle".
    extruder: int | None
    diameter: str
    # "HH" (high flow), "HS" (standard), or None when the printer has not said.
    flow: str | None = None

    @property
    def extruder_or_default(self) -> int:
        """0 when unknown -- correct on a single-nozzle machine, a guess on a dual."""
        return 0 if self.extruder is None else self.extruder

    def flow_matches(self, stored_flow: str | None) -> bool:
        """Whether a stored K profile's flow type applies to this nozzle.

        Unknown on either side matches anything, and that is the load-bearing
        case rather than a nicety:

        * Every K profile stored before this existed has NULL here, so a strict
          comparison would stop applying all of them at once.
        * An X1C declares no flow on any calibration entry -- measured: all
          eight come back with ``nozzle_id: ''`` -- so profiles saved from one
          have nothing truthful to store. Treating "no answer" as "Standard"
          and then filtering on it would break the moment a high-flow nozzle is
          fitted to a machine whose table never mentioned flow.

        Once BOTH sides do declare one, they have to agree: a K value measured
        on a high-flow nozzle is not a fact about a standard one, the same way
        a 0.6 measurement says nothing about a 0.4.
        """
        if not stored_flow or not self.flow:
            return True
        return normalise_flow(stored_flow) == self.flow


def normalise_flow(raw: str | None) -> str | None:
    """The flow-type code in a nozzle id or type string, or None.

    Both spellings reduce to the same two letters, which is the whole point:
    a calibration entry files its nozzle as ``HH00-0.4`` / ``HS00-0.4`` while
    the fitted nozzle reports its type as ``HH01`` -- measured on an H2D, and
    the reason this compares two characters rather than four. The trailing
    digits are a hardware variant the calibration table normalises to ``00``.
    """
    text = (raw or "").strip().upper()
    return text[:2] if text[:2] in ("HH", "HS") else None


def reports_two_nozzles(state) -> bool:
    """Whether the live state shows a dual-nozzle printer.

    Decides which extruder an external spool feeds (254 = left = extruder 1,
    255 = right = extruder 0). ``ams_extruder_map`` alone is not enough: it is
    derived from AMS info bits, so a dual-nozzle printer with no AMS reports an
    empty map, and its external spools then matched no nozzle (#2771).

    ``nozzles`` is always a two-entry list (the state seeds two empty stubs), so
    its length proves nothing; only a populated diameter on the second entry
    means real hardware. The other two signals cover firmware revisions that
    surface one but not the other: a populated ``ams_extruder_map`` is
    dual-nozzle by construction, and so is more than one ``vt_tray`` entry,
    since single-nozzle printers expose exactly one external feed.
    """
    raw_data = getattr(state, "raw_data", None) or {}
    nozzles = getattr(state, "nozzles", None) or []
    vt_trays = raw_data.get("vt_tray") or []
    return bool(
        (len(nozzles) > 1 and getattr(nozzles[1], "nozzle_diameter", ""))
        or raw_data.get("ams_extruder_map")
        # isinstance, because a dict here would count its keys as trays. An
        # H3C's one input per hotend is not a second extruder.
        or (isinstance(vt_trays, list) and len(vt_trays) > 1 and not TrayLayout.from_units(None, vt_trays).rack)
    )


def nozzle_flow_for_extruder(state, extruder: int | None, model: str | None = None) -> str | None:
    """The flow type fitted to ``extruder``, or None when the printer is silent.

    Read from the same array as the diameter and indexed the same way. A
    printer that reports no nozzle type -- an X1C sends none at all -- yields
    None, which ``flow_matches`` treats as "applies to anything" rather than
    inventing Standard.
    """
    nozzles = getattr(state, "nozzles", None) or []
    if not nozzles:
        return None

    index = 0
    if extruder is not None and extruder > 0 and is_dual_nozzle_model(model):
        index = extruder

    for candidate in (index, 0):
        if candidate < len(nozzles):
            flow = normalise_flow(getattr(nozzles[candidate], "nozzle_type", ""))
            if flow:
                return flow
    return None


def nozzle_diameter_for_extruder(state, extruder: int | None, model: str | None = None) -> str:
    """The diameter fitted to ``extruder``, or the printer's only nozzle.

    Falls back to index 0, and then to 0.4, whenever the printer has not
    reported the entry -- an absent nozzle must not make this raise, since it is
    called on every assign.
    """
    nozzles = getattr(state, "nozzles", None) or []
    if not nozzles:
        return DEFAULT_NOZZLE_DIAMETER

    index = 0
    if extruder is not None and extruder > 0 and is_dual_nozzle_model(model):
        # Physical nozzle id N sits on extruder N -- see the module docstring
        # for why the legacy left/right convention cannot apply here.
        index = extruder

    for candidate in (index, 0):
        if candidate < len(nozzles):
            diameter = (getattr(nozzles[candidate], "nozzle_diameter", "") or "").strip()
            if diameter:
                return diameter
    return DEFAULT_NOZZLE_DIAMETER


def resolve_slot_nozzle(state, ams_id: int, tray_id: int, model: str | None = None) -> SlotNozzle:
    """The extruder an AMS slot feeds and that nozzle's diameter.

    ``state`` is the live ``PrinterState`` (or None when the printer is not
    connected, which yields the defaults rather than an error).
    """
    if state is None:
        return SlotNozzle(extruder=None, diameter=DEFAULT_NOZZLE_DIAMETER)

    extruder = slot_extruder(
        ams_id,
        tray_id,
        getattr(state, "ams_extruder_map", None),
        getattr(state, "ams_switch_inlet", None),
    )
    return SlotNozzle(
        extruder=extruder,
        diameter=nozzle_diameter_for_extruder(state, extruder, model),
        flow=nozzle_flow_for_extruder(state, extruder, model),
    )