"""Shared spool ``slicer_filament`` → ``(tray_info_idx, setting_id)`` resolver. The internal-inventory and Spoolman-inventory routes both need to translate a spool's stored slicer-preset reference (cloud preset ID / local preset ID / GF-prefix Bambu filament ID / free-text material name) into the two MQTT fields ``ams_filament_setting`` consumes: the printer-side ``tray_info_idx`` (filament_id) and the slicer-side ``setting_id``. The two routes were drifting in lockstep before #1713 — internal mode resolved everything, Spoolman mode silently dropped slicer_filament on the floor and only the generic-material fallback fired. This module is the single chokepoint so the two flows can't diverge again. Resolver outcomes: - Returns ``("", "", None)`` when ``slicer_filament`` is empty, unresolvable, or sanitised away as a slicer-rejected value (literal material name, PFUS / PFCN cloud setting_id). The caller is responsible for the generic-material fallback when this happens. - Returns ``(tray_info_idx, setting_id, sub_brand_override)`` otherwise. The third element is non-empty when a cloud-detail lookup or a local- preset name provides a more specific brand label than the spool's own ``" "`` concatenation — the caller should prefer it over its computed default. The resolver is async because the GFS / PFUS / PFCN branches need cloud authentication and the local-preset branch reads ``LocalPreset`` from the DB. Pass ``current_user=None`` to skip cloud auth (the on_ams_change replay path uses this); cloud-prefix presets then fall back to a static ``normalize_slicer_filament`` parse, which is correct when the slot was already configured by an earlier authenticated assign and the printer's calibration table preserves the real filament_id. """ from __future__ import annotations import json import logging import re from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from backend.app.core.permissions import Permission from backend.app.models.user import User from backend.app.utils.filament_ids import ( GENERIC_FILAMENT_IDS, filament_id_to_setting_id, normalize_slicer_filament, ) from backend.app.utils.filament_types import is_material_name logger = logging.getLogger(__name__) # Orca Cloud profile ids are UUIDs, the one preset reference in this codebase # with no letter prefix to key off. A spool stores the bare id (the spool form # persists ``preset.setting_id`` verbatim), so shape is all there is to go on. _ORCA_PROFILE_ID = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}") async def _orca_filament_id( db: AsyncSession, current_user: User | None, profile_id: str, ) -> tuple[str, str | None, str | None]: """Look up an Orca Cloud profile's own filament_id. Returns ``(filament_id, name, filament_type)`` -- all empty/None when the profile cannot be fetched or carries no id of its own, which leaves the caller on its generic fallback. Best-effort by construction: this runs inside spool assignment, not a user request, so a missing pairing, a revoked token or a lapsed permission must degrade to the fallback rather than fail the assignment. That is also why ``clear_on_auth_failure=False`` -- Orca reports every refresh rejection with one composite reason, so a background caller cannot tell a real revocation from a lost rotation race and must not wipe a working pairing on it. The route path hits the same failure in front of a user and clears there. """ if current_user is not None and not current_user.has_permission(Permission.ORCA_CLOUD_AUTH.value): logger.debug("Orca filament lookup skipped for %r: caller lacks orca_cloud:auth", profile_id) return ("", None, None) svc = None try: from backend.app.api.routes.orca_cloud import _build_authenticated_service svc = await _build_authenticated_service(db, current_user, clear_on_auth_failure=False) profile = await svc.get_profile(profile_id) except Exception as e: logger.debug("Orca filament lookup failed for %r: %s", profile_id, e) return ("", None, None) finally: # A raise in `finally` escapes the `except` above, so guard it: closing # an httpx client must never be what fails a spool assignment. if svc is not None: try: await svc.close() except Exception as e: # noqa: BLE001 - close() is best-effort logger.debug("Orca client close failed after lookup of %r: %s", profile_id, e) content = profile.get("content") if isinstance(profile, dict) else None if not isinstance(content, dict): return ("", None, None) raw_fid = content.get("filament_id") filament_id = raw_fid.strip() if isinstance(raw_fid, str) else "" name = profile.get("name") if isinstance(profile, dict) else None return ( filament_id, name if isinstance(name, str) and name else None, _preset_filament_type(content.get("filament_type")), ) def _preset_filament_type(raw: object) -> str | None: """Read a slicer preset's ``filament_type`` field. Bambu Studio and OrcaSlicer both store it as a one-element array (``["PLA"]``); some hand-written and older profiles store a bare string. ``orca_profiles._extract_filament_fields`` accepts both and this has to agree with it, since that is what fills ``LocalPreset.filament_type``. The value is still run through ``printer_filament_type`` by the caller. That is a no-op for every type the app knows -- ``TestTheMaterialsBambuddyOffers`` pins exactly that -- and a pass-through for a type it does not, so nothing the slicer says is discarded. It only bites on a hand-edited profile whose ``filament_type`` is a product line, which is the case this whole module exists to keep out of an AMS slot. """ if isinstance(raw, list): raw = raw[0] if raw else None if isinstance(raw, str) and raw.strip(): return raw.strip() return None async def resolve_slicer_filament( *, db: AsyncSession, current_user: User | None, slicer_filament: str | None, slicer_filament_name: str | None, material: str | None, ) -> tuple[str, str, str | None, str | None]: """Resolve a spool's slicer-preset reference to printer-side ids. ``slicer_filament``: the spool's stored reference (e.g. ``"GFA01"``, ``"PFUS990b6e19965353"``, ``"38"`` for a numeric LocalPreset id, or free-text). May be empty or None — returns the empty tuple in that case. ``slicer_filament_name``: optional builtin-name realignment hint. When set and the resolved tray_info_idx maps to a different builtin name, the resolver swaps to the builtin whose name matches (e.g. user picked "Bambu PLA Matte" but the cloud lookup landed on "Bambu PLA Basic"). ``material``: spool material string for the local-preset fallback branch when the LocalPreset's setting JSON doesn't carry a filament_id. Returns ``(tray_info_idx, setting_id, sub_brand_override, type_override)`` — all empty when nothing resolved. ``sub_brand_override`` is non-None when a more specific brand label is available (cloud detail name or local preset name); ``None`` means the caller should use its own default. ``type_override`` is the preset's own ``filament_type`` when the preset carries one — the slicer's answer to what the material is, rather than one parsed out of the spool's material column. It is what the caller should write into ``tray_type``. ``None`` means no preset said, and the caller falls back to reducing the spool's material (``printer_filament_type``). Raised in the #2902 thread by @doncaruana: a preset has to be chosen from a list the slicer defines, so its type needs no interpreting. It cannot be the only source, though — ``slicer_filament`` is nullable on a spool while ``material`` is required, and the spool this issue was reported for had no preset at all. """ sf = (slicer_filament or "").strip() if not sf: return ("", "", None, None) tray_info_idx = "" setting_id = "" sub_brand_override: str | None = None type_override: str | None = None base_sf = sf.split("_")[0] if "_" in sf else sf # Cloud-side preset IDs in three known shapes: # GFS… — Bambu official cloud preset # PFUS… — cloud user-created preset # PFCN… — cloud shared / partner preset (e.g. Polymaker's "(Custom)" # Bambu Lab H2D variant, #1648) # All three need a cloud-detail lookup to extract the underlying # filament_id; without it the raw cloud id ends up in tray_info_idx # and the printer's calibration table can't resolve it. # Source order is Orca Cloud, Bambu Cloud, local import, generic fallback. # Orca goes first because its ids are the only ones identified by shape # rather than prefix -- and because, before #3003, a UUID fell through every # branch below into ``normalize_slicer_filament``, which passes anything it # does not recognise straight through. A 36-character UUID then went into # tray_info_idx, an 8-character field, and the slot ended up pointing at the # first 8 characters of a UUID: the same failure the PFUS guard at the # bottom of this function exists for. if _ORCA_PROFILE_ID.fullmatch(base_sf): tray_info_idx, orca_name, orca_type = await _orca_filament_id(db, current_user, base_sf) if orca_type: type_override = orca_type if orca_name: sub_brand_override = orca_name.split("@")[0].strip() # setting_id is left empty here: the UUID is what the slicer cannot # resolve, and unlike a PFUS there is no cloud id form it accepts # instead. All three callers then derive one from the filament_id # (`filament_id_to_setting_id`), which is what keeps the slot from # going out half configured -- the same path a local import takes. elif base_sf.startswith("GFS") or base_sf.startswith("PFUS") or base_sf.startswith("PFCN"): setting_id = base_sf try: from backend.app.api.routes.cloud import build_authenticated_cloud cloud = await build_authenticated_cloud(db, current_user) if cloud is not None and cloud.is_authenticated: try: detail = await cloud.get_setting_detail(base_sf) # The preset's own type, straight from the slicer's own # profile -- no parsing of a product name (#2902). The # preset JSON is nested under ``setting``; some responses # carry it at the top level instead, the same shape spread # ``preset_resolver`` documents. cloud_setting = detail.get("setting") type_override = _preset_filament_type( (cloud_setting if isinstance(cloud_setting, dict) else detail).get("filament_type") ) # A custom preset's OWN filament_id is the only thing that # gets it into an AMS slot as itself: the printer stores # that id, the slicer matches its presets against it, and # the 8-character field fits it exactly ("P" + 7 hex). # # Bambu Cloud normally returns it on the envelope, which is # what the captures in #1053 show for a Studio-created # preset (filament_id: "Pbd31b30"). The `setting` fallback # here is belt-and-braces for a response that carries it in # the preset JSON instead, the same spread `filament_type` # above has to handle -- no captured response has needed it # yet, and it costs a dict lookup to be ready for one. # # An Orca-created preset has no filament_id anywhere: the # envelope says null and `setting` is a delta from the base # (#1053 again). Those legitimately fall to base_id below # and reach the slicer as the profile they inherit from -- # an OrcaSlicer preset-format gap, filed upstream as # OrcaSlicer PR #13315, not something resolvable here. own_filament_id = detail.get("filament_id") or ( cloud_setting.get("filament_id") if isinstance(cloud_setting, dict) else None ) if own_filament_id: tray_info_idx = own_filament_id cloud_name = detail.get("name", "") if cloud_name: sub_brand_override = cloud_name.replace(r"@.*$", "").split("@")[0].strip() elif detail.get("base_id"): bid = detail["base_id"].split("_")[0] if bid.startswith("GFS") and len(bid) >= 5: tray_info_idx = f"GF{bid[3:]}" else: tray_info_idx = bid finally: await cloud.close() elif cloud is not None: await cloud.close() except Exception as e: logger.warning("Slicer-filament resolve: cloud lookup failed for %r: %s", sf, e) if not tray_info_idx: tray_info_idx, setting_id = normalize_slicer_filament(sf) elif base_sf.startswith("GF"): tray_info_idx, setting_id = normalize_slicer_filament(sf) else: try: local_id = int(sf) from backend.app.models.local_preset import LocalPreset as LP lp_result = await db.execute(select(LP).where(LP.id == local_id, LP.preset_type == "filament")) lp = lp_result.scalar_one_or_none() if lp: # The slicer's own answer, extracted from the profile at import # time by ``orca_profiles``. Preferred over anything parsed out # of the spool's material column (#2902). type_override = _preset_filament_type(lp.filament_type) # Local preset's setting JSON carries the printer-recognized # filament_id (e.g. "P4d64437") — use that directly so the # slicer can resolve the specific preset. Falls through to # generic material id only when the JSON doesn't carry one. lp_filament_id = "" if lp.setting: try: setting_data = json.loads(lp.setting) raw_fid = setting_data.get("filament_id") if isinstance(raw_fid, str) and raw_fid: lp_filament_id = raw_fid except (json.JSONDecodeError, AttributeError): pass if lp_filament_id: tray_info_idx = lp_filament_id setting_id = filament_id_to_setting_id(lp_filament_id) else: # Deliberately not widened to cover product-line materials # ("PLA+", "HTPLA") the way the callers' own fallbacks were # (#2902). Returning an id here rather than "" would skip # the caller's whole no-id block, and with it the slot-reuse # branch that keeps a printer's calibrated preset -- so the # widening belongs there, after reuse has had its turn. mat = (material or lp.filament_type or "").upper().strip() tray_info_idx = ( GENERIC_FILAMENT_IDS.get(mat) or GENERIC_FILAMENT_IDS.get(mat.split("-")[0].split(" ")[0]) or "" ) if lp.name: sub_brand_override = lp.name.split("@")[0].strip() except (ValueError, TypeError): tray_info_idx, setting_id = normalize_slicer_filament(sf) # Realign tray_info_idx to a builtin whose name matches slicer_filament_name # when the current resolution lands on a builtin with a different name # (e.g. cloud detail returned PLA Basic but the spool was labelled PLA Matte). if tray_info_idx and slicer_filament_name: from backend.app.api.routes.cloud import _BUILTIN_FILAMENT_NAMES expected_name = _BUILTIN_FILAMENT_NAMES.get(tray_info_idx, "") if expected_name and expected_name != slicer_filament_name: for fid, fname in _BUILTIN_FILAMENT_NAMES.items(): if fname == slicer_filament_name: tray_info_idx = fid setting_id = filament_id_to_setting_id(fid) break # Defend against tray_info_idx values the slicer cannot resolve. Three # shapes leak through and must be discarded so the caller's generic- # material fallback can rescue the slot: # 1. Literal material names ("PLA", "PETG-CF") that pass through # normalize_slicer_filament unchanged when the spool's slicer_filament # is free-text rather than a real preset ID. Product lines ("PLA+", # "HTPLA") count as material names too -- see is_material_name, which # is shared with the slot-reuse check that must agree with this. # 2. PFUS-prefix cloud setting_ids — valid as setting_id but rejected # by the slicer as tray_info_idx (the printer's calibration table # indexes by filament_id, and a PFUS isn't one). This normally gets # realigned to a P-prefix local id via the caller's printer_kp # lookup, but on the replay path in main.py.on_ams_change # current_user=None skips cloud auth and leaves the raw PFUS in # tray_info_idx — overwriting the correctly-configured slot from # the original assign. # 3. PFCN-prefix cloud shared / partner presets (e.g. Polymaker's # "(Custom)" H2D variants, #1648) — same shape problem as PFUS. # 4. Orca Cloud profile UUIDs, when the branch above could not reach the # profile to trade one for its filament_id (#3003). Worst of the four # at 36 characters against an 8-character field. # Valid tray_info_idx values: "GF" + letter + digits (Bambu official) or # "P" followed by hex (user/local presets, NOT "PFUS" or "PFCN"). if tray_info_idx and ( is_material_name(tray_info_idx) or tray_info_idx.startswith("PFUS") or tray_info_idx.startswith("PFCN") or _ORCA_PROFILE_ID.fullmatch(tray_info_idx) ): tray_info_idx = "" # Preserve setting_id when it's still a valid slicer reference # (PFUS / PFCN cloud user/shared preset, or GFS Bambu official # preset). The slicer accepts these as setting_id even though # they're rejected as tray_info_idx; without preservation the # slicer falls back to whatever generic filament the caller's # tray_info_idx fallback produces and shows "Generic " # instead of the user's actual custom preset (#1815). Material-name # leaks (e.g. setting_id="PETG") are still cleared — those are # never valid slicer references. if not ( setting_id and (setting_id.startswith("PFUS") or setting_id.startswith("PFCN") or setting_id.startswith("GFS")) ): setting_id = "" return (tray_info_idx, setting_id, sub_brand_override, type_override)