bambuddy/backend/app/services/spoolman.py

"""Spoolman integration service for syncing AMS filament data."""

import asyncio
import logging
import weakref
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Literal

import httpx

from backend.app.utils.color_utils import color_match_key, spoolman_color_hex

logger = logging.getLogger(__name__)

BAMBU_RFID_TAG_LENGTH = 32


@dataclass
class SpoolmanSpool:
    """Represents a spool in Spoolman."""

    id: int
    filament_id: int | None
    remaining_weight: float | None
    used_weight: float
    first_used: str | None
    last_used: str | None
    location: str | None
    lot_nr: str | None
    comment: str | None
    extra: dict | None  # Contains tag_uid in extra.tag


@dataclass
class SpoolmanFilament:
    """Represents a filament type in Spoolman."""

    id: int
    name: str
    vendor_id: int | None
    material: str | None
    color_hex: str | None
    weight: float | None  # Net weight in grams


@dataclass
class AMSTray:
    """Represents an AMS tray with filament data from Bambu printer."""

    ams_id: int  # 0-3 for regular AMS, 128-135 for AMS-HT, 254+ for external spool
    tray_id: int  # 0-3
    tray_type: str  # PLA, PETG, ABS, etc.
    tray_sub_brands: str  # Full name like "PLA Basic", "PETG HF"
    tray_color: str  # Hex color like "FEC600FF"
    remain: int  # Remaining percentage (0-100)
    tag_uid: str  # RFID tag UID
    tray_uuid: str  # Spool UUID
    tray_info_idx: str  # Bambu filament preset ID like "GFA00"
    tray_weight: int  # Spool weight in grams (usually 1000)


class SpoolmanNotFoundError(Exception):
    """Raised when a spool ID does not exist in Spoolman (HTTP 404)."""


class SpoolmanUnavailableError(Exception):
    """Raised when Spoolman is unreachable or returns a server/network error."""


class SpoolmanClientError(Exception):
    """Raised when Spoolman returns a 4xx client error (not 404)."""

    def __init__(self, message: str, status_code: int, response_text: str = "") -> None:
        super().__init__(message)
        self.status_code = status_code
        self.response_text = response_text


def _filament_subtype_part(name: str, material: str) -> str:
    """Return the subtype portion of a filament name, lowercased.

    Mirrors the read-side derivation in
    ``backend/app/api/routes/_spoolman_helpers.py::_map_spoolman_spool``:
    if the filament name starts with the material prefix (e.g. ``"PLA Glow"``
    when material is ``"PLA"``), strip it; otherwise return the name as-is.

    Used by ``find_or_create_filament`` so that an existing filament saved by
    the AMS-sync path with name ``"Glow"`` still matches a user-driven edit
    that composes ``"PLA Glow"`` (#1357).
    """
    s = (name or "").strip()
    m = (material or "").strip()
    if m and s.upper().startswith(m.upper() + " "):
        return s[len(m) + 1 :].strip().lower()
    return s.lower()


class SpoolmanClient:
    """Client for interacting with Spoolman API."""

    def __init__(self, base_url: str):
        """Initialize the Spoolman client."""
        self.base_url = base_url.rstrip("/")
        self.api_url = f"{self.base_url}/api/v1"
        self._client: httpx.AsyncClient | None = None
        self._connected = False
        # Per-spool locks for atomic read-modify-write in merge_spool_extra.
        # WeakValueDictionary: locks are GC'd once no coroutine holds a reference.
        self._extra_locks: weakref.WeakValueDictionary[int, asyncio.Lock] = weakref.WeakValueDictionary()
        # Extra-field names this client has already registered with Spoolman.
        # Bounded by the number of distinct keys Bambuddy writes, so it never
        # grows with spool count; scoped to the instance so a client pointed at
        # a different Spoolman starts over.
        self._ensured_extra_fields: set[str] = set()
        # Whether Spoolman's extra-field listing has been read once. Separate
        # from the set above because the two answer different questions: the
        # set is "which fields are known to exist", this is "have we asked".
        # Without it a client registering three brand-new fields re-read the
        # whole listing before each one.
        self._extra_fields_listed = False
        self._ensure_extra_lock = asyncio.Lock()

    async def _get_client(self) -> httpx.AsyncClient:
        """Get or create the HTTP client with connection pooling limits."""
        if self._client is None:
            self._client = httpx.AsyncClient(
                timeout=httpx.Timeout(connect=5.0, read=10.0, write=5.0, pool=5.0),
                follow_redirects=False,
                verify=True,
                limits=httpx.Limits(
                    max_keepalive_connections=5,
                    max_connections=10,
                    keepalive_expiry=30.0,
                ),
            )
        return self._client

    async def close(self):
        """Close the HTTP client."""
        if self._client:
            await self._client.aclose()
            self._client = None

    async def health_check(self) -> bool:
        """Check if Spoolman server is reachable; returns True if healthy."""
        try:
            client = await self._get_client()
            response = await client.get(f"{self.api_url}/health")
            self._connected = response.status_code == 200
            return self._connected
        except Exception as e:
            logger.warning(
                "Spoolman health check failed (url=%s, type=%s): %s",
                self.api_url,
                type(e).__name__,
                e,
            )
            self._connected = False
            return False

    @property
    def is_connected(self) -> bool:
        """Check if client is connected to Spoolman."""
        return self._connected

    async def get_spools(self) -> list[dict]:
        """Fetch all spools from Spoolman with up to 3 retries on connection errors."""
        max_attempts = 3
        retry_delay = 0.5  # 500ms

        for attempt in range(1, max_attempts + 1):
            try:
                client = await self._get_client()
                response = await client.get(f"{self.api_url}/spool")
                response.raise_for_status()
                spools = response.json()
                if attempt > 1:
                    logger.info("Successfully fetched %d spools on attempt %d", len(spools), attempt)
                return spools
            except (httpx.ReadError, httpx.RemoteProtocolError, httpx.ConnectError) as e:
                # Connection-related errors - close and recreate client for next attempt
                if attempt < max_attempts:
                    logger.warning(
                        "Connection error getting spools (attempt %d/%d): %s. Recreating client and retrying in %dms...",
                        attempt,
                        max_attempts,
                        e,
                        int(retry_delay * 1000),
                    )
                    # Close the stale client and recreate it
                    await self.close()
                    await asyncio.sleep(retry_delay)
                else:
                    logger.error("Failed to get spools from Spoolman after %d attempts: %s", max_attempts, e)
                    raise SpoolmanUnavailableError("Cannot reach Spoolman") from e
            except Exception as e:
                # Other errors (HTTP errors, JSON decode errors, etc.)
                if attempt < max_attempts:
                    logger.warning(
                        "Failed to get spools from Spoolman (attempt %d/%d): %s. Retrying in %dms...",
                        attempt,
                        max_attempts,
                        e,
                        int(retry_delay * 1000),
                    )
                    await asyncio.sleep(retry_delay)
                else:
                    logger.error("Failed to get spools from Spoolman after %d attempts: %s", max_attempts, e)
                    raise SpoolmanUnavailableError("Cannot reach Spoolman") from e

    async def _get_with_retry(self, path: str, params: dict | None = None) -> list[dict]:
        """GET a Spoolman JSON list endpoint with up to 3 retries on connection errors."""
        max_attempts = 3
        retry_delay = 0.5
        url = f"{self.api_url}/{path.lstrip('/')}"

        for attempt in range(1, max_attempts + 1):
            try:
                client = await self._get_client()
                response = await client.get(url, params=params or None)
                response.raise_for_status()
                return response.json()
            except (httpx.ReadError, httpx.RemoteProtocolError, httpx.ConnectError) as e:
                if attempt < max_attempts:
                    logger.warning(
                        "Connection error fetching %s (attempt %d/%d): %s. Recreating client and retrying in %dms...",
                        path,
                        attempt,
                        max_attempts,
                        e,
                        int(retry_delay * 1000),
                    )
                    await self.close()
                    await asyncio.sleep(retry_delay)
                else:
                    logger.error("Failed to fetch %s from Spoolman after %d attempts: %s", path, max_attempts, e)
                    raise SpoolmanUnavailableError("Cannot reach Spoolman") from e
            except Exception as e:
                if attempt < max_attempts:
                    logger.warning(
                        "Failed to fetch %s from Spoolman (attempt %d/%d): %s. Retrying in %dms...",
                        path,
                        attempt,
                        max_attempts,
                        e,
                        int(retry_delay * 1000),
                    )
                    await asyncio.sleep(retry_delay)
                else:
                    logger.error("Failed to fetch %s from Spoolman after %d attempts: %s", path, max_attempts, e)
                    raise SpoolmanUnavailableError("Cannot reach Spoolman") from e

    async def get_filaments(self) -> list[dict]:
        """Fetch all internal filaments from Spoolman."""
        try:
            client = await self._get_client()
            response = await client.get(f"{self.api_url}/filament")
            response.raise_for_status()
            return response.json()
        except Exception as e:
            logger.error("Failed to get filaments from Spoolman: %s", e)
            raise SpoolmanUnavailableError("Cannot reach Spoolman") from e

    async def get_filament(self, filament_id: int) -> dict:
        """Fetch a single filament by ID from Spoolman."""
        if filament_id <= 0:
            raise ValueError(f"Invalid filament_id: {filament_id}")
        response = await self._request_filament("GET", filament_id, operation="get_filament")
        return response.json()

    async def get_external_filaments(self) -> list[dict]:
        """Fetch external/library filaments from Spoolman."""
        try:
            client = await self._get_client()
            response = await client.get(f"{self.api_url}/external/filament")
            response.raise_for_status()
            return response.json()
        except Exception as e:
            logger.error("Failed to get external filaments from Spoolman: %s", e)
            raise SpoolmanUnavailableError("Cannot reach Spoolman") from e

    async def get_vendors(self) -> list[dict]:
        """Fetch all vendors from Spoolman."""
        try:
            client = await self._get_client()
            response = await client.get(f"{self.api_url}/vendor")
            response.raise_for_status()
            return response.json()
        except Exception as e:
            logger.error("Failed to get vendors from Spoolman: %s", e)
            raise SpoolmanUnavailableError("Cannot reach Spoolman") from e

    async def create_vendor(self, name: str) -> dict:
        """Create a new vendor in Spoolman."""
        try:
            client = await self._get_client()
            response = await client.post(f"{self.api_url}/vendor", json={"name": name})
            if 400 <= response.status_code < 500:
                raise SpoolmanClientError(
                    f"Spoolman rejected vendor creation (HTTP {response.status_code})",
                    response.status_code,
                )
            response.raise_for_status()
            return response.json()
        except SpoolmanClientError:
            raise
        except Exception as e:
            logger.error("Failed to create vendor in Spoolman: %s", e)
            raise SpoolmanUnavailableError("Cannot reach Spoolman") from e

    def _get_material_density(self, material: str | None) -> float:
        """Return typical density (g/cm³) for the given filament material; defaults to PLA (1.24)."""
        # Typical densities for common filament materials
        densities = {
            "PLA": 1.24,
            "PLA-CF": 1.29,
            "PLA-S": 1.24,
            "PETG": 1.27,
            "ABS": 1.04,
            "ASA": 1.07,
            "TPU": 1.21,
            "PA": 1.14,  # Nylon
            "PA-CF": 1.20,
            "PC": 1.20,
            "PVA": 1.23,
            "HIPS": 1.04,
            "PP": 0.90,
            "PET": 1.38,
        }
        if material:
            # Try exact match first, then uppercase
            mat_upper = material.upper()
            for key, density in densities.items():
                if key.upper() == mat_upper or mat_upper.startswith(key.upper()):
                    return density
        return 1.24  # Default to PLA density

    async def create_filament(
        self,
        name: str,
        vendor_id: int | None = None,
        material: str | None = None,
        color_hex: str | None = None,
        color_name: str | None = None,
        weight: float | None = None,
        diameter: float = 1.75,
        density: float | None = None,
    ) -> dict:
        """Create a new filament in Spoolman."""
        if not name or not name.strip():
            raise ValueError("Filament name is required")

        if density is None:
            density = self._get_material_density(material)

        data: dict = {
            "name": name.strip(),
            "diameter": diameter,
            "density": density,
        }
        if vendor_id:
            data["vendor_id"] = vendor_id
        if material:
            data["material"] = material
        if color_hex:
            # Every create funnels through here, so this is where the stored shape
            # is decided: six characters for an opaque spool, eight only when the
            # alpha byte says the filament is translucent. See #2912.
            data["color_hex"] = spoolman_color_hex(color_hex) or color_hex
        if color_name:
            data["color_name"] = color_name
        if weight:
            data["weight"] = weight

        logger.debug("Creating filament in Spoolman: %s", data)
        try:
            client = await self._get_client()
            response = await client.post(f"{self.api_url}/filament", json=data)
            if 400 <= response.status_code < 500:
                raise SpoolmanClientError(
                    f"Spoolman rejected filament creation (HTTP {response.status_code})",
                    response.status_code,
                )
            response.raise_for_status()
            return response.json()
        except SpoolmanClientError:
            raise
        except Exception as e:
            logger.error("Failed to create filament in Spoolman: %s", e)
            raise SpoolmanUnavailableError("Cannot reach Spoolman") from e

    async def patch_filament(self, filament_id: int, data: dict) -> dict:
        """PATCH a filament entry in Spoolman (e.g. update name or spool_weight)."""
        if filament_id <= 0:
            raise ValueError(f"Invalid filament_id: {filament_id}")
        response = await self._request_filament("PATCH", filament_id, json_body=data, operation="patch_filament")
        return response.json()

    async def create_spool(
        self,
        filament_id: int,
        remaining_weight: float | None = None,
        location: str | None = None,
        lot_nr: str | None = None,
        comment: str | None = None,
        extra: dict | None = None,
    ) -> dict:
        """Create a new spool in Spoolman."""
        data: dict = {"filament_id": filament_id}
        if remaining_weight is not None:
            data["remaining_weight"] = remaining_weight
        if location:
            data["location"] = location
        if lot_nr:
            data["lot_nr"] = lot_nr
        if comment:
            data["comment"] = comment
        if extra:
            data["extra"] = extra
            await self._ensure_extra_fields(extra)

        logger.debug("Creating spool in Spoolman: %s", data)
        try:
            client = await self._get_client()
            response = await client.post(f"{self.api_url}/spool", json=data)
            if response.status_code == 404:
                raise SpoolmanNotFoundError(f"Filament {filament_id} not found in Spoolman")
            if 400 <= response.status_code < 500:
                raise SpoolmanClientError(
                    f"Spoolman rejected spool creation (HTTP {response.status_code})",
                    response.status_code,
                )
            response.raise_for_status()
            result = response.json()
            logger.info("Created spool %s in Spoolman", result.get("id"))
            return result
        except (SpoolmanNotFoundError, SpoolmanClientError):
            raise
        except Exception as e:
            logger.error("Failed to create spool in Spoolman: %s", e)
            raise SpoolmanUnavailableError("Cannot reach Spoolman") from e

    async def update_spool(
        self,
        spool_id: int,
        remaining_weight: float | None = None,
        location: str | None = None,
        clear_location: bool = False,
        extra: dict | None = None,
    ) -> dict:
        """Update an existing spool in Spoolman, always setting last_used."""
        data: dict = {}
        if remaining_weight is not None:
            data["remaining_weight"] = remaining_weight
        if clear_location:
            data["location"] = None
        elif location:
            data["location"] = location
        if extra:
            data["extra"] = extra
            await self._ensure_extra_fields(extra)
        data["last_used"] = datetime.now(timezone.utc).isoformat()

        response = await self._request_spool("PATCH", spool_id, json_body=data, operation="update")
        return response.json()

    async def _request_spool(
        self,
        method: Literal["GET", "PATCH", "DELETE"],
        spool_id: int,
        *,
        json_body: dict | None = None,
        operation: str,
    ) -> httpx.Response:
        """Perform a spool-scoped HTTP request, translating 404 and errors to named exceptions."""
        try:
            client = await self._get_client()
            response = await client.request(
                method,
                f"{self.api_url}/spool/{spool_id}",
                json=json_body,
            )
            if response.status_code == 404:
                raise SpoolmanNotFoundError(f"Spool {spool_id} not found in Spoolman")
            response.raise_for_status()
            return response
        except SpoolmanNotFoundError:
            raise
        except httpx.HTTPStatusError as e:
            if 400 <= e.response.status_code < 500:
                logger.warning(
                    "Spoolman returned %d for %s spool %s",
                    e.response.status_code,
                    operation,
                    spool_id,
                )
                raise SpoolmanClientError(
                    f"Spoolman rejected {operation} for spool {spool_id} (HTTP {e.response.status_code})",
                    e.response.status_code,
                    e.response.text[:500],
                ) from e
            else:
                logger.error("Failed to %s spool %s in Spoolman: %s", operation, spool_id, e)
                raise SpoolmanUnavailableError(f"Failed to {operation} spool {spool_id}") from e
        except Exception as e:
            logger.error("Failed to %s spool %s in Spoolman: %s", operation, spool_id, e)
            raise SpoolmanUnavailableError(f"Failed to {operation} spool {spool_id}") from e

    async def _request_filament(
        self,
        method: Literal["GET", "PATCH"],
        filament_id: int,
        *,
        json_body: dict | None = None,
        operation: str,
    ) -> httpx.Response:
        """Perform a filament-scoped HTTP request, translating 404 and errors to named exceptions."""
        try:
            client = await self._get_client()
            response = await client.request(
                method,
                f"{self.api_url}/filament/{filament_id}",
                json=json_body,
            )
            if response.status_code == 404:
                raise SpoolmanNotFoundError(f"Filament {filament_id} not found in Spoolman")
            response.raise_for_status()
            return response
        except SpoolmanNotFoundError:
            raise
        except httpx.HTTPStatusError as e:
            if 400 <= e.response.status_code < 500:
                logger.warning(
                    "Spoolman returned %d for %s filament %s",
                    e.response.status_code,
                    operation,
                    filament_id,
                )
                raise SpoolmanClientError(
                    f"Spoolman rejected {operation} for filament {filament_id} (HTTP {e.response.status_code})",
                    e.response.status_code,
                    e.response.text[:500],
                ) from e
            else:
                logger.error("Failed to %s filament %s in Spoolman: %s", operation, filament_id, e)
                raise SpoolmanUnavailableError(f"Failed to {operation} filament {filament_id}") from e
        except Exception as e:
            logger.error("Failed to %s filament %s in Spoolman: %s", operation, filament_id, e)
            raise SpoolmanUnavailableError(f"Failed to {operation} filament {filament_id}") from e

    async def get_spool(self, spool_id: int) -> dict:
        """Fetch a single spool by ID from Spoolman."""
        response = await self._request_spool("GET", spool_id, operation="get")
        return response.json()

    async def get_all_spools(self, allow_archived: bool = False) -> list[dict]:
        """Fetch all spools from Spoolman with retry, optionally including archived ones."""
        params: dict = {}
        if allow_archived:
            params["allow_archived"] = "true"
        return await self._get_with_retry("/spool", params=params or None)

    async def get_distinct_locations(self) -> list[str]:
        """Return distinct location strings currently assigned to Spoolman spools.

        Spoolman's `/location` endpoint shape varies across versions: older
        releases return `list[str]`, newer ones return `list[dict]` with a
        `name` field. Normalize to `list[str]` so callers can iterate without
        runtime shape checks.
        """
        raw = await self._get_with_retry("/location")
        if not isinstance(raw, list):
            return []
        names: list[str] = []
        for entry in raw:
            if isinstance(entry, str):
                names.append(entry)
            elif isinstance(entry, dict):
                name = entry.get("name")
                if isinstance(name, str):
                    names.append(name)
        return names

    async def rename_location(self, current_name: str, new_name: str) -> int:
        """Bulk-rename a location string on all Spoolman spools.

        Tries the bulk `PATCH /location/{name}` endpoint first. Spoolman
        versions older than ~0.16 don't expose it and respond 404/405 — in
        that case fall back to iterating every spool currently at
        ``current_name`` and PATCHing each one's ``location`` field directly.
        Returns the number of spools renamed (or 0 if the bulk endpoint
        succeeded without enumerating).
        """
        from urllib.parse import quote

        encoded = quote(current_name, safe="")
        client = await self._get_client()
        try:
            response = await client.patch(
                f"{self.api_url}/location/{encoded}",
                json={"name": new_name},
            )
            response.raise_for_status()
            return 0
        except httpx.HTTPStatusError as exc:
            if exc.response.status_code not in (404, 405):
                raise
            logger.info(
                "Spoolman bulk-rename endpoint unavailable (status %d); falling back to per-spool PATCH",
                exc.response.status_code,
            )

        # Per-spool fallback: enumerate every spool currently at the old name
        # and PATCH each. Keep going on individual failures so a single
        # already-deleted spool doesn't strand the rest at the old name —
        # collect errors and re-raise as a single SpoolmanClientError if any
        # leftover survives.
        spools = await self.get_all_spools(allow_archived=True)
        renamed = 0
        failures: list[str] = []
        for spool in spools:
            if (spool.get("location") or "").strip() != current_name:
                continue
            try:
                await self._request_spool(
                    "PATCH",
                    spool["id"],
                    json_body={"location": new_name},
                    operation="rename-location",
                )
                renamed += 1
            except SpoolmanNotFoundError:
                continue
            except Exception as exc:  # noqa: BLE001 — accumulate and re-raise below
                failures.append(f"spool {spool.get('id')}: {exc}")
        if failures:
            raise SpoolmanClientError(
                f"Spoolman rename fallback failed for {len(failures)} spool(s): {'; '.join(failures[:3])}",
                status_code=502,
            )
        return renamed

    async def delete_spool(self, spool_id: int) -> None:
        """Delete a spool from Spoolman."""
        await self._request_spool("DELETE", spool_id, operation="delete")

    async def is_filament_shared(self, filament_id: int, exclude_spool_id: int) -> bool:
        """True if any spool other than ``exclude_spool_id`` is linked to ``filament_id``.

        Used by the spool-edit path to decide between PATCHing the existing
        filament in place (singleton) and falling back to find_or_create
        (shared — re-linking the spool is the only safe option). Includes
        archived spools so a shared link doesn't suddenly look singleton just
        because the sibling spool was archived.
        """
        spools = await self.get_all_spools(allow_archived=True)
        for s in spools:
            if s.get("id") == exclude_spool_id:
                continue
            if ((s.get("filament") or {}).get("id")) == filament_id:
                return True
        return False

    async def set_spool_archived(self, spool_id: int, archived: bool) -> dict:
        """Archive or restore a spool in Spoolman."""
        response = await self._request_spool(
            "PATCH",
            spool_id,
            json_body={"archived": archived},
            operation="archive/restore",
        )
        return response.json()

    async def reset_spool_usage(self, spool_id: int) -> dict:
        """Reset a spool's used_weight to 0 in Spoolman.

        Used by the per-spool / bulk "Reset usage to 0" actions on the
        Inventory page so the Total Consumed stat can be cleared without
        touching the rest of the spool's data.
        """
        response = await self._request_spool(
            "PATCH",
            spool_id,
            json_body={"used_weight": 0},
            operation="reset-usage",
        )
        return response.json()

    async def update_spool_full(
        self,
        spool_id: int,
        *,
        filament_id: int | None = None,
        remaining_weight: float | None = None,
        comment: str | None = None,
        price: float | None = None,
        location: str | None = None,
        clear_location: bool = False,
        extra: dict | None = None,
        spool_weight: float | None = None,
        clear_spool_weight: bool = False,
    ) -> dict:
        """Update a spool with full field support; unlike update_spool, does not auto-set last_used."""
        data: dict = {}
        if filament_id is not None:
            data["filament_id"] = filament_id
        if remaining_weight is not None:
            data["remaining_weight"] = remaining_weight
        if comment is not None:
            data["comment"] = comment if comment else None
        if price is not None:
            data["price"] = price
        if clear_location:
            data["location"] = None
        elif location is not None:
            data["location"] = location
        if extra is not None:
            data["extra"] = extra
            await self._ensure_extra_fields(extra)
        if clear_spool_weight:
            data["spool_weight"] = None
        elif spool_weight is not None:
            data["spool_weight"] = spool_weight

        response = await self._request_spool("PATCH", spool_id, json_body=data, operation="update")
        return response.json()

    def extra_lock(self, spool_id: int) -> asyncio.Lock:
        """Return (creating if needed) the per-spool asyncio.Lock used by merge_spool_extra."""
        lock = self._extra_locks.get(spool_id)
        if lock is None:
            lock = asyncio.Lock()
            self._extra_locks[spool_id] = lock
        return lock

    async def merge_spool_extra(self, spool_id: int, new_fields: dict) -> dict:
        """Fetch the spool's extra dict, merge new_fields into it, then PATCH back — serialised per spool."""
        async with self.extra_lock(spool_id):
            current = await self.get_spool(spool_id)  # raises on error
            current_extra: dict = current.get("extra") or {}
            merged = {**current_extra, **new_fields}
            return await self.update_spool_full(spool_id=spool_id, extra=merged)

    async def find_or_create_vendor(self, name: str) -> int:
        """Return the Spoolman vendor ID for the given name, creating the vendor if absent."""
        vendors = await self.get_vendors()
        name_lower = name.strip().lower()
        for vendor in vendors:
            if vendor.get("name", "").strip().lower() == name_lower:
                return vendor["id"]
        created = await self.create_vendor(name.strip())
        vendor_id = created.get("id")
        if not vendor_id:
            raise SpoolmanUnavailableError(f"Spoolman returned vendor without id field: {list(created.keys())}")
        return vendor_id

    async def find_or_create_filament(
        self,
        material: str,
        subtype: str,
        brand: str | None,
        color_hex: str,
        label_weight: int,
        color_name: str | None = None,
    ) -> int:
        """Return the filament ID matching material/name/brand/color, creating it if absent."""
        name = f"{material} {subtype}".strip() if subtype else material
        # One value in both roles. `color_match_key` returns the shape the colour
        # would be stored as, so the key the loop below compares on and the value
        # a new filament is created with are the same string by construction: an
        # opaque spool keys and stores as six characters, a translucent one as
        # eight, and neither can be conflated with the other (#2912).
        color = color_match_key(color_hex)

        vendor_id: int | None = None
        if brand:
            vendor_id = await self.find_or_create_vendor(brand)

        # Normalised match keys (case-insensitive). Computed once outside the
        # loop so the inner comparison stays simple.
        composed_subtype = _filament_subtype_part(name, material)
        material_norm = material.upper()
        brand_norm = (brand or "").strip().lower()

        filaments = await self.get_filaments()
        for f in filaments:
            f_material = (f.get("material") or "").upper()
            f_color = color_match_key(f.get("color_hex"))
            f_vendor = f.get("vendor") or {}
            f_vendor_name = (f_vendor.get("name") or "").strip().lower()

            material_match = f_material == material_norm
            # Match on the subtype portion of the filament name. AMS-sync
            # auto-create (the underscore-prefixed `_find_or_create_filament`
            # used during MQTT tray import) stores the filament as just
            # ``tray.tray_sub_brands`` — e.g. ``"Glow"`` — while the
            # user-driven edit path here composes ``"<material> <subtype>"``
            # — ``"PLA Glow"``. The old literal equality `f_name == name`
            # failed to bridge the two shapes, so every edit fell through to
            # `create_filament`, leaving a trail of duplicate filaments AND
            # leaving the spool either still pointed at the old filament
            # whose `color_name` never got patched, or pointed at a new
            # filament with the colour while the inventory list kept
            # showing the synth fallback from the old one (#1357).
            f_subtype_part = _filament_subtype_part(f.get("name") or "", material)
            name_match = f_subtype_part == composed_subtype
            color_match = f_color == color
            vendor_match = (not brand) or f_vendor_name == brand_norm

            if material_match and name_match and color_match and vendor_match:
                # color_name is intentionally not part of the match key and
                # is no longer patched onto the filament here: Spoolman 0.23.1
                # has no `color_name` field on Filament (#1357 — confirmed
                # against the FilamentUpdateParameters schema). The earlier
                # #1319 fix tried to patch it and Spoolman silently dropped
                # the key, which is exactly why the user's edit looked "not
                # saved". The route now persists color_name via
                # spool.extra.bambu_color_name (see _map_spoolman_spool for
                # the read side); find_or_create_filament's only job is to
                # resolve the right filament_id for the spool link.
                return f["id"]

        # color_name omitted: Spoolman has no such field on Filament (#1357);
        # the user's color_name lands in spool.extra.bambu_color_name via the
        # route after find_or_create_filament returns the new id.
        filament = await self.create_filament(
            name=name,
            vendor_id=vendor_id,
            material=material,
            color_hex=color,
            weight=float(label_weight),
        )
        filament_id = filament.get("id")
        if not filament_id:
            raise SpoolmanUnavailableError(f"Spoolman returned filament without id field: {list(filament.keys())}")
        return filament_id

    async def use_spool(self, spool_id: int, used_weight: float) -> dict:
        """Record filament usage for a spool via the Spoolman /use endpoint."""
        try:
            client = await self._get_client()
            response = await client.put(
                f"{self.api_url}/spool/{spool_id}/use",
                json={"use_weight": used_weight},
            )
            if response.status_code == 404:
                raise SpoolmanNotFoundError(f"Spool {spool_id} not found in Spoolman")
            if 400 <= response.status_code < 500:
                raise SpoolmanClientError(
                    f"Spoolman rejected use_spool for spool {spool_id} (HTTP {response.status_code})",
                    response.status_code,
                )
            response.raise_for_status()
            return response.json()
        except (SpoolmanNotFoundError, SpoolmanClientError):
            raise
        except Exception as e:
            logger.error("Failed to record spool usage in Spoolman: %s", e)
            raise SpoolmanUnavailableError(f"Failed to record usage for spool {spool_id}") from e

    async def find_spool_by_tag(self, tag_uid: str, cached_spools: list[dict] | None = None) -> dict | None:
        """Return the spool matching the given RFID tag UID, or None if not found."""
        # Use cached spools if provided, otherwise fetch from API
        spools = cached_spools if cached_spools is not None else await self.get_spools()
        # Normalize tag_uid for comparison (uppercase, strip quotes)
        search_tag = tag_uid.strip('"').upper()

        for spool in spools:
            extra = spool.get("extra", {})
            if extra:
                stored_tag = extra.get("tag", "")
                # Normalize stored tag (strip quotes, uppercase)
                if stored_tag:
                    normalized_tag = stored_tag.strip('"').upper()
                    if normalized_tag == search_tag:
                        logger.debug("Found spool %s matching tag %s", spool["id"], tag_uid)
                        return spool
        return None

    def _find_spool_by_location(self, location: str, cached_spools: list[dict] | None) -> dict | None:
        """Return the spool at the exact location string, or None; fallback when RFID is unavailable."""
        if not cached_spools:
            return None
        for spool in cached_spools:
            if spool.get("location") == location:
                return spool
        return None

    async def find_spools_by_location_prefix(
        self, location_prefix: str, cached_spools: list[dict] | None = None
    ) -> list[dict]:
        """Return all spools whose location starts with location_prefix."""
        # Use cached spools if provided, otherwise fetch from API
        spools = cached_spools if cached_spools is not None else await self.get_spools()
        matching = []
        for spool in spools:
            location = spool.get("location", "")
            if location and location.startswith(location_prefix):
                matching.append(spool)
        return matching

    async def clear_location_for_removed_spools(
        self,
        printer_name: str,
        current_tray_uuids: set[str],
        cached_spools: list[dict] | None = None,
        synced_spool_ids: set[int] | None = None,
    ) -> int:
        """Clear location for Bambu Lab spools at this printer whose tray_uuid is no longer in the AMS."""
        location_prefix = f"{printer_name} - "
        spools_at_printer = await self.find_spools_by_location_prefix(location_prefix, cached_spools=cached_spools)
        cleared_count = 0

        for spool in spools_at_printer:
            spool_id = spool.get("id")

            # Skip spools that were just synced (matched by location or tag)
            if synced_spool_ids and spool_id in synced_spool_ids:
                continue

            # Get the tray_uuid (stored as "tag" in extra field)
            extra = spool.get("extra", {}) or {}
            stored_tag = extra.get("tag", "")
            if stored_tag:
                # Normalize: strip quotes and uppercase
                spool_uuid = stored_tag.strip('"').upper()
            else:
                spool_uuid = ""

            # Only clear location for Bambu Lab spools (those with a stored 32-character RFID tag).
            if len(spool_uuid) != BAMBU_RFID_TAG_LENGTH:
                continue

            # If this spool's UUID is not in the current AMS, clear its location
            if spool_uuid not in current_tray_uuids:
                logger.info(
                    f"Clearing location for spool {spool_id} "
                    f"(was: {spool.get('location')}, uuid: {spool_uuid[:16] if spool_uuid else 'none'}...)"
                )
                result = await self.update_spool(spool_id=spool_id, clear_location=True)
                if result:
                    cleared_count += 1

        return cleared_count

    async def ensure_bambu_vendor(self) -> int | None:
        """Return the Bambu Lab vendor ID in Spoolman, creating the vendor if absent."""
        vendors = await self.get_vendors()
        for vendor in vendors:
            if vendor.get("name", "").lower() == "bambu lab":
                return vendor["id"]

        # Create Bambu Lab vendor if not exists
        vendor = await self.create_vendor("Bambu Lab")
        return vendor["id"] if vendor else None

    async def ensure_tag_extra_field(self) -> bool:
        """Register the 'tag' extra field in Spoolman if not present; returns True on success."""
        return await self.ensure_extra_field("tag")

    async def _load_existing_extra_field_keys(self) -> set[str] | None:
        """Keys of the spool extra fields Spoolman already has, or ``None`` when
        the listing could not be read.

        ``None`` and ``set()`` mean different things and the caller acts on the
        difference: an empty set is "Spoolman has no extra fields", which means
        every field Bambuddy needs must be created; ``None`` is "we could not
        find out", where the only safe move is to fall back to attempting the
        write blind.
        """
        try:
            client = await self._get_client()
            response = await client.get(f"{self.api_url}/field/spool")
            if response.status_code != 200:
                logger.debug(
                    "Spoolman extra-field listing returned %s; falling back to blind registration",
                    response.status_code,
                )
                return None
            fields = response.json()
        except Exception as e:  # noqa: BLE001 — registration is best-effort, see _ensure_extra_fields
            logger.debug("Could not read Spoolman extra-field listing: %s", e)
            return None
        if not isinstance(fields, list):
            return None
        # Match on `key`, not `name`: `key` is the identifier the extra dict is
        # written under and the one Bambuddy cares about, while `name` is the
        # free-text label a user is free to change in Spoolman's UI.
        return {f["key"] for f in fields if isinstance(f, dict) and isinstance(f.get("key"), str)}

    async def ensure_extra_field(self, name: str, field_type: str = "text") -> bool:
        """Register a custom extra field in Spoolman if not present.

        Spoolman rejects PATCH requests that include unknown extra-dict keys
        with HTTP 400 ('Unknown extra field <name>.'), so any custom field
        Bambuddy persists alongside spools needs to be pre-registered.
        Idempotent — returns True if the field already exists.

        Existence is read from ``GET /field/spool``, the whole-listing endpoint.
        This used to probe ``GET /field/spool/{name}`` for one field at a time,
        which Spoolman has never served: it declares only POST and DELETE at
        that path, so the probe answered 405 every time and the check could
        never succeed (issue #2983, reported by @ngreatorex).

        Falling through to the POST on every call was worse than a wasted
        request, because that endpoint is an upsert rather than a create. It
        answered 200 whether or not the field was already there, so a field a
        user had renamed, retyped or given a default in Spoolman's own UI was
        silently reset to Bambuddy's version of it on every restart. Reading
        the listing first is what lets an existing field be left alone.
        """
        try:
            if name in self._ensured_extra_fields:
                return True

            if not self._extra_fields_listed:
                existing = await self._load_existing_extra_field_keys()
                if existing is not None:
                    # Bank the whole listing: the caller registers several
                    # fields in a row, and each one it already has is a request
                    # not sent and a user customisation not overwritten. Read
                    # once per client — every field created after this point is
                    # added to the set as it is created, so re-reading would
                    # only ever confirm what we already know.
                    self._ensured_extra_fields |= existing
                    self._extra_fields_listed = True
                    if name in existing:
                        logger.debug("Spoolman extra field %r already exists", name)
                        return True

            client = await self._get_client()
            field_data = {
                "name": name,
                "field_type": field_type,
                "default_value": None,
            }
            response = await client.post(f"{self.api_url}/field/spool/{name}", json=field_data)
            if response.status_code in (200, 201):
                logger.info("Created Spoolman extra field %r", name)
                self._ensured_extra_fields.add(name)
                return True

            logger.warning(
                "Failed to create Spoolman extra field %r: %s - %s",
                name,
                response.status_code,
                response.text,
            )
            return False

        except Exception as e:
            logger.warning("Failed to ensure Spoolman extra field %r exists: %s", name, e)
            return False

    async def _ensure_extra_fields(self, extra: dict | None) -> None:
        """Register every extra key an outgoing write declares, once per client.

        Spoolman answers HTTP 400 "Unknown extra field <name>." for any extra
        key that was not registered first, so registration has to happen before
        the write, not before the feature. It used to happen before the feature:
        three hand-maintained lists (the connect route, startup, and two inline
        blocks in the inventory routes) each named the fields they expected to
        be written later. Enabling Spoolman from Settings reaches none of them,
        so the first AMS sync on a fresh Spoolman failed on every slot -- and
        the Connect button that would have registered them is hidden by then,
        because saving the settings initialises the client and the status
        endpoint reads that as "connected" (issue #2903).

        Keying off the payload instead removes the chance to forget: a write
        that carries a key is a write that registers it. ``bambu_color_name``
        is the cautionary case -- it never made it into the connect or startup
        lists, and only works today because two call sites remembered to
        register it by hand.

        Best-effort by design. ``ensure_extra_field`` logs and returns False
        rather than raising, and a failure here must not turn a write that
        might still succeed into one that never happens -- the caller's own
        error handling stays exactly as it was.
        """
        names = [name for name in (extra or {}) if name not in self._ensured_extra_fields]
        if not names:
            return

        async with self._ensure_extra_lock:
            for name in names:
                # Re-check under the lock: a concurrent write may have just
                # registered this one, and two syncs racing to POST the same
                # field is how one of them gets a needless warning logged.
                if name not in self._ensured_extra_fields:
                    await self.ensure_extra_field(name)

    def parse_ams_tray(self, ams_id: int, tray_data: dict) -> AMSTray | None:
        """Parse raw MQTT tray data into an AMSTray; returns None for empty or invalid trays."""
        # Skip empty trays - check for valid tray_type
        tray_type = tray_data.get("tray_type", "")
        if not tray_type or tray_type.strip() == "":
            return None

        # Need valid color to create filament
        tray_color = tray_data.get("tray_color", "")
        if not tray_color or tray_color.strip() == "":
            logger.debug("Skipping tray with empty color")
            return None

        # Transparent filament (alpha=00) used to be rewritten to a cream
        # "natural PLA" colour before being stored, because the swatch
        # renderer couldn't show alpha. The swatch now paints a checkerboard
        # underlay for translucent rgbas (see filamentSwatchHelpers.ts), so
        # we pass `00000000` through verbatim — the inventory row keeps the
        # AMS-reported colour and the frontend resolves the name to "Clear"
        # via getColorName (#1545).

        # Get sub_brands, falling back to tray_type
        tray_sub_brands = tray_data.get("tray_sub_brands", "")
        if not tray_sub_brands or tray_sub_brands.strip() == "":
            tray_sub_brands = tray_type

        # Get tag_uid and tray_uuid, filtering out empty/invalid values
        tag_uid = tray_data.get("tag_uid", "")
        if tag_uid in ("", "0000000000000000"):
            tag_uid = ""
        tray_uuid = tray_data.get("tray_uuid", "")
        if tray_uuid in ("", "00000000000000000000000000000000"):
            tray_uuid = ""

        # Get tray_info_idx (Bambu filament preset ID like "GFA00")
        tray_info_idx = tray_data.get("tray_info_idx", "") or ""

        # Get remaining percentage (-1 means unknown/not read by AMS)
        remain = int(tray_data.get("remain", -1))

        return AMSTray(
            ams_id=ams_id,
            tray_id=int(tray_data.get("id", 0)),
            tray_type=tray_type.strip(),
            tray_sub_brands=tray_sub_brands.strip(),
            tray_color=tray_color,
            remain=remain,
            tag_uid=tag_uid,
            tray_uuid=tray_uuid,
            tray_info_idx=tray_info_idx.strip(),
            tray_weight=int(tray_data.get("tray_weight", 1000)),
        )

    def convert_ams_slot_to_location(self, ams_id: int, tray_id: int) -> str:
        """Return a human-readable location string (e.g. "AMS A1") for the given AMS slot."""
        if ams_id >= 254:
            return "External Spool"

        if 128 <= ams_id <= 135:
            # AMS-HT units use IDs 128-135
            ht_letter = chr(ord("A") + (ams_id - 128))
            return f"AMS-HT {ht_letter}{tray_id + 1}"

        ams_letter = chr(ord("A") + ams_id)
        return f"AMS {ams_letter}{tray_id + 1}"

    def is_bambu_lab_spool(self, tray_uuid: str, tag_uid: str = "", tray_info_idx: str = "") -> bool:
        """Return True if tray_uuid or tag_uid identifies a Bambu Lab spool; tray_info_idx is ignored."""
        # Check tray_uuid (preferred - consistent across printer models)
        if tray_uuid:
            uuid = tray_uuid.strip()
            if len(uuid) == 32 and uuid != "00000000000000000000000000000000":
                try:
                    int(uuid, 16)
                    return True
                except ValueError:
                    pass

        # Fallback: check tag_uid (RFID tag - varies between printer readers)
        # Bambu Lab RFID tags are 16 hex characters (8 bytes)
        if tag_uid:
            tag = tag_uid.strip()
            if len(tag) == 16 and tag != "0000000000000000":
                try:
                    int(tag, 16)
                    logger.debug("Identified Bambu Lab spool via tag_uid fallback: %s", tag)
                    return True
                except ValueError:
                    pass

        return False

    def calculate_remaining_weight(self, remain_percent: int, spool_weight: int) -> float:
        """Return remaining filament weight in grams given a percentage and total spool weight."""
        return (remain_percent / 100.0) * spool_weight

    async def sync_ams_tray(
        self,
        tray: AMSTray,
        printer_name: str,
        disable_weight_sync: bool = False,
        cached_spools: list[dict] | None = None,
        inventory_remaining: float | None = None,
        spoolman_spool_id_hint: int | None = None,
        auto_add_unknown_rfid: bool = True,
    ) -> dict | None:
        """Sync one AMS tray to Spoolman; creates the spool on first sight, updates weight otherwise."""
        logger.debug(
            f"Processing {printer_name} AMS {tray.ams_id} tray {tray.tray_id}: "
            f"type={tray.tray_type}, idx={tray.tray_info_idx or 'none'}, "
            f"uuid={tray.tray_uuid[:16] if tray.tray_uuid else 'none'}, "
            f"tag={tray.tag_uid[:8] if tray.tag_uid else 'none'}..."
        )

        # Determine which identifier to use for Spoolman (prefer tray_uuid, fallback to tag_uid)
        # Zero-filled values mean the AMS hasn't read the RFID tag — treat as no tag
        zero_uuid = "00000000000000000000000000000000"
        zero_tag = "0000000000000000"
        spool_tag = None
        if tray.tray_uuid and tray.tray_uuid != zero_uuid:
            spool_tag = tray.tray_uuid
        elif tray.tag_uid and tray.tag_uid != zero_tag:
            spool_tag = tray.tag_uid

        # Calculate remaining weight
        # Primary: AMS MQTT data (remain percentage + tray_weight)
        # Fallback: Built-in inventory tracked weight (when firmware sends invalid remain/tray_weight)
        if tray.remain >= 0 and tray.tray_weight > 0:
            remaining = self.calculate_remaining_weight(tray.remain, tray.tray_weight)
        elif inventory_remaining is not None:
            remaining = inventory_remaining
            logger.debug(
                "Using inventory weight fallback for %s AMS %s tray %s: %.1fg",
                printer_name,
                tray.ams_id,
                tray.tray_id,
                remaining,
            )
        else:
            remaining = None

        if spool_tag:
            # Primary path: match by RFID tag
            existing = await self.find_spool_by_tag(spool_tag, cached_spools=cached_spools)
            if existing:
                logger.info("Updating existing spool %s for tag %s...", existing["id"], spool_tag[:16])
                return await self.update_spool(
                    spool_id=existing["id"],
                    remaining_weight=None if disable_weight_sync else remaining,
                )

            # Spool not found by tag - auto-create it, unless the user has
            # opted out of auto-adding unknown RFIDs (settings.auto_add_unknown_rfid).
            # Caller broadcasts unknown_tag on the resulting None so the UI can
            # surface a "+ Add to inventory" affordance on the slot.
            if not auto_add_unknown_rfid:
                logger.info(
                    "Auto-add disabled; skipping Spoolman spool create for %s (tag: %s...)",
                    tray.tray_sub_brands,
                    spool_tag[:16],
                )
                return None

            logger.info("Creating new spool in Spoolman for %s (tag: %s...)", tray.tray_sub_brands, spool_tag[:16])
            if self.is_bambu_lab_spool(tray.tray_uuid, tray.tag_uid, tray.tray_info_idx):
                filament = await self._find_or_create_filament(tray)
                filament_id = filament["id"] if filament else None
            else:
                # Non-BL spool with custom RFID: use generic vendor lookup
                brand = tray.tray_sub_brands if tray.tray_sub_brands != tray.tray_type else None
                try:
                    filament_id = await self.find_or_create_filament(
                        material=tray.tray_type,
                        subtype="",
                        brand=brand,
                        color_hex=tray.tray_color,
                        label_weight=tray.tray_weight,
                    )
                except (SpoolmanNotFoundError, SpoolmanUnavailableError, SpoolmanClientError):
                    logger.warning("Could not find or create filament for non-BL spool %s", tray.tray_sub_brands)
                    return None

            if not filament_id:
                logger.error("Failed to find or create filament for %s", tray.tray_sub_brands)
                return None

            import json

            return await self.create_spool(
                filament_id=filament_id,
                remaining_weight=remaining,
                comment="Created by Bambuddy",
                extra={"tag": json.dumps(spool_tag)},
            )

        # No-RFID fallback: use the spool ID resolved from the local slot-assignment table.
        # Never create new spools without a tag to avoid duplicates.
        if spoolman_spool_id_hint is not None:
            existing = next((s for s in (cached_spools or []) if s.get("id") == spoolman_spool_id_hint), None)
            if existing is None:
                try:
                    existing = await self.get_spool(spoolman_spool_id_hint)
                except (SpoolmanNotFoundError, SpoolmanUnavailableError):
                    existing = None
            if existing:
                logger.info(
                    "Updating spool %s by slot-assignment hint (no RFID tag available)",
                    existing["id"],
                )
                return await self.update_spool(
                    spool_id=existing["id"],
                    remaining_weight=None if disable_weight_sync else remaining,
                )

        logger.info(
            "%s AMS %s tray %s — skipping (no RFID tag and no slot-assignment hint)",
            printer_name,
            tray.ams_id,
            tray.tray_id,
        )
        return None

    async def _find_or_create_filament(self, tray: AMSTray) -> dict | None:
        """Return a Bambu Lab filament matching the tray's material/color, creating it if absent."""
        bambu_vendor_id = await self.ensure_bambu_vendor()
        material_upper = tray.tray_type.upper()
        # Same single value as the user-driven path: the match key is the stored
        # shape. That is what lets an opaque tray still find the six-character
        # filaments every existing instance is full of, while a clear tray keys
        # to eight and gets its own record (#2912).
        color = color_match_key(tray.tray_color)

        # Search internal filaments - only match Bambu Lab vendor
        filaments = await self.get_filaments()
        for filament in filaments:
            fil_vendor_id = filament.get("vendor_id") or filament.get("vendor", {}).get("id")
            if fil_vendor_id != bambu_vendor_id:
                continue
            fil_material = filament.get("material") or ""
            if fil_material.upper() == material_upper and color_match_key(filament.get("color_hex")) == color:
                return filament

        # Search external filaments (SpoolmanDB) — restrict to Bambu Lab only.
        # The /api/v1/external/filament endpoint returns the full multi-vendor catalog
        # with no server-side filter, so without a manufacturer check the first PLA/black
        # hit is typically 3DJAKE or 3DXTECH, not Bambu Lab.
        external = await self.get_external_filaments()
        sub_brand = (tray.tray_sub_brands or "").strip().lower()
        bambu_candidates = []
        for filament in external:
            manufacturer = (filament.get("manufacturer") or "").strip().lower()
            ext_id = (filament.get("id") or "").strip().lower()
            if manufacturer != "bambu lab" and not ext_id.startswith("bambulab_"):
                continue
            fil_material = filament.get("material") or ""
            if fil_material.upper() == material_upper and color_match_key(filament.get("color_hex")) == color:
                bambu_candidates.append(filament)

        if bambu_candidates:
            # Prefer the entry whose `name` matches the AMS `tray_sub_brands`
            # (e.g. "PLA Basic", "Support for PLA/PETG Black") so the more specific
            # variant wins over a generic "Black" entry when both are present.
            chosen = next(
                (f for f in bambu_candidates if (f.get("name") or "").strip().lower() == sub_brand),
                bambu_candidates[0],
            )
            return await self._create_filament_from_external(chosen, tray)

        # Not found in either source - create a new Bambu Lab filament from scratch.
        return await self.create_filament(
            name=tray.tray_sub_brands or tray.tray_type,
            vendor_id=bambu_vendor_id,
            material=tray.tray_type,
            color_hex=color,
            weight=tray.tray_weight,
        )

    async def _create_filament_from_external(self, external: dict, tray: AMSTray) -> dict | None:
        """Create an internal Spoolman filament from an external library entry."""
        vendor_id = await self.ensure_bambu_vendor()
        return await self.create_filament(
            name=external.get("name", tray.tray_sub_brands),
            vendor_id=vendor_id,
            material=external.get("material", tray.tray_type),
            # `or`, not a two-argument get: an entry that carries the key with an
            # explicit null would hand None to create_filament rather than reach
            # the tray fallback. Only a candidate when the tray colour is empty
            # too, so this is a correctness tidy, not a fix for a live path.
            color_hex=external.get("color_hex") or color_match_key(tray.tray_color),
            weight=external.get("weight", tray.tray_weight),
            density=external.get("density"),
        )


# Global client instance (initialized when settings are loaded)
_spoolman_client: SpoolmanClient | None = None


async def get_spoolman_client() -> SpoolmanClient | None:
    """Return the global SpoolmanClient, or None if not configured."""
    return _spoolman_client


async def init_spoolman_client(url: str) -> SpoolmanClient:
    """Initialise (or reinitialise) the global SpoolmanClient; raises ValueError if url fails SSRF guard."""
    from backend.app.api.routes._spoolman_helpers import assert_safe_spoolman_url

    assert_safe_spoolman_url(url)

    global _spoolman_client
    if _spoolman_client:
        await _spoolman_client.close()

    _spoolman_client = SpoolmanClient(url)
    return _spoolman_client


async def close_spoolman_client():
    """Close the global Spoolman client."""
    global _spoolman_client
    if _spoolman_client:
        await _spoolman_client.close()
        _spoolman_client = None