bambuddy/backend/app/utils/threemf_tools.py

"""3MF file parsing utilities for filament tracking.

This module provides functions to parse Bambu Lab 3MF files and extract
per-layer filament usage data from the embedded G-code. This enables
accurate partial usage reporting for multi-material prints.
"""

import hashlib
import json
import logging
import math
import re
import zipfile
from collections import OrderedDict
from dataclasses import dataclass, field
from pathlib import Path
from threading import Lock

# Parsing goes through defusedxml; the element type it hands back is the stdlib
# one, and defusedxml does not re-export it, so annotations name it directly.
from xml.etree.ElementTree import Element as XmlElement

import defusedxml.ElementTree as ET

logger = logging.getLogger(__name__)

# Default filament properties
DEFAULT_FILAMENT_DIAMETER = 1.75  # mm
DEFAULT_FILAMENT_DENSITY = 1.24  # g/cm³ (PLA)


def parse_gcode_layer_filament_usage(gcode_content: str) -> dict[int, dict[int, float]]:
    """Parse G-code to extract per-layer, per-filament cumulative extrusion in mm.

    This function tracks filament extrusion across layers and tool changes,
    building a cumulative usage map that can be used to calculate partial
    usage at any layer.

    Args:
        gcode_content: The raw G-code content as a string

    Returns:
        A nested dictionary mapping layer numbers to filament usage:
        {layer: {filament_id: cumulative_mm}, ...}

    Example:
        {0: {0: 125.5}, 1: {0: 250.0, 1: 50.0}, 2: {0: 375.0, 1: 150.0}}

        This shows:
        - Layer 0: filament 0 used 125.5mm cumulative
        - Layer 1: filament 0 used 250mm cumulative, filament 1 used 50mm
        - Layer 2: filament 0 used 375mm cumulative, filament 1 used 150mm

    G-code commands parsed:
        - M73 L<layer>: Layer change marker
        - M620 S<filament>: Filament/tool change (S255 = unload)
        - G0/G1/G2/G3 E<amount>: Extrusion moves
    """
    layer_filaments: dict[int, dict[int, float]] = {}
    current_layer = 0
    active_filament: int | None = None
    cumulative_extrusion: dict[int, float] = {}  # filament_id -> total mm

    for line in gcode_content.splitlines():
        line = line.strip()
        if not line:
            continue

        # Handle comments - skip but check for layer markers
        if line.startswith(";"):
            # Some slicers use comment-based layer markers
            # e.g., "; CHANGE_LAYER" or ";LAYER_CHANGE"
            continue

        # Split line into command and inline comment
        if ";" in line:
            line = line.split(";")[0].strip()

        # Extract command and parameters
        parts = line.split()
        if not parts:
            continue
        cmd = parts[0].upper()

        # Layer change: M73 L<layer>
        # Bambu printers use M73 with L parameter for layer indication
        if cmd == "M73":
            for part in parts[1:]:
                part_upper = part.upper()
                if part_upper.startswith("L"):
                    try:
                        new_layer = int(part[1:])
                        # Save current state before layer change
                        if cumulative_extrusion:
                            layer_filaments[current_layer] = cumulative_extrusion.copy()
                        current_layer = new_layer
                    except ValueError:
                        pass  # Skip G-code lines with unparseable layer numbers

        # Filament change: M620 S<filament>
        # Bambu uses M620 for AMS filament switching
        # S255 means full unload (no active filament)
        elif cmd == "M620":
            for part in parts[1:]:
                part_upper = part.upper()
                if part_upper.startswith("S"):
                    filament_str = part[1:]
                    if filament_str == "255":
                        # Full unload - no active filament
                        active_filament = None
                    else:
                        try:
                            # Extract digits (e.g., "0A" -> 0, "1" -> 1)
                            match = re.match(r"(\d+)", filament_str)
                            if match:
                                active_filament = int(match.group(1))
                        except (ValueError, AttributeError):
                            pass  # Skip unparseable filament switch commands

        # Extrusion moves: G0/G1/G2/G3 with E parameter
        # Only G1 typically has extrusion, but check all for safety
        elif cmd in ("G0", "G1", "G2", "G3"):
            if active_filament is None:
                continue
            for part in parts[1:]:
                part_upper = part.upper()
                if part_upper.startswith("E"):
                    try:
                        extrusion = float(part[1:])
                        # Only count positive extrusion (not retractions)
                        if extrusion > 0:
                            current = cumulative_extrusion.get(active_filament, 0)
                            cumulative_extrusion[active_filament] = current + extrusion
                    except ValueError:
                        pass  # Skip G-code lines with unparseable extrusion values

    # Save final layer state
    if cumulative_extrusion:
        layer_filaments[current_layer] = cumulative_extrusion.copy()

    return layer_filaments


def mm_to_grams(
    length_mm: float,
    diameter_mm: float = DEFAULT_FILAMENT_DIAMETER,
    density_g_cm3: float = DEFAULT_FILAMENT_DENSITY,
) -> float:
    """Convert filament length in mm to weight in grams.

    Uses the formula: mass = volume × density
    where volume = π × r² × length

    Args:
        length_mm: Length of filament in millimeters
        diameter_mm: Filament diameter in millimeters (default: 1.75)
        density_g_cm3: Material density in g/cm³ (default: 1.24 for PLA)

    Returns:
        Weight in grams
    """
    radius_cm = (diameter_mm / 2) / 10  # Convert mm to cm
    length_cm = length_mm / 10  # Convert mm to cm
    volume_cm3 = math.pi * radius_cm * radius_cm * length_cm
    return volume_cm3 * density_g_cm3


def extract_layer_filament_usage_from_3mf(
    file_path: Path, plate_id: int | None = None
) -> dict[int, dict[int, float]] | None:
    """Extract per-layer filament usage from a 3MF file's embedded G-code.

    Args:
        file_path: Path to the 3MF file
        plate_id: Plate to read. Required for multi-plate files — zip member
            order is whatever the slicer wrote, and Bambu Studio stores
            ``plate_2.gcode`` ahead of ``plate_1.gcode``, so the old
            "first member" behaviour read a different plate's layers than
            the one that printed. Returns None rather than silently falling
            back to another plate when the requested plate isn't in the
            file; callers degrade to linear scaling, which is bounded.

    Returns:
        Dictionary mapping layers to filament usage, or None if parsing fails.
        Format: {layer: {filament_id: cumulative_mm}, ...}
    """
    try:
        with zipfile.ZipFile(file_path, "r") as zf:
            names = zf.namelist()
            gcode_path = select_plate_gcode_name(names, plate_id)
            if gcode_path is None:
                # No plate asked for, or a file whose single G-code member
                # doesn't follow the plate_N naming convention (non-Bambu
                # slicers) — the lone member is unambiguous either way.
                gcode_files = [f for f in names if f.endswith(".gcode")]
                if plate_id is None or len(gcode_files) == 1:
                    gcode_path = default_plate_gcode_name(names)
            if gcode_path is None:
                return None

            gcode_content = zf.read(gcode_path).decode("utf-8", errors="ignore")

            return parse_gcode_layer_filament_usage(gcode_content)
    except Exception:
        return None


def get_cumulative_usage_at_layer(
    layer_usage: dict[int, dict[int, float]],
    target_layer: int,
) -> dict[int, float]:
    """Get cumulative filament usage (in mm) up to and including target_layer.

    Args:
        layer_usage: The output from parse_gcode_layer_filament_usage()
        target_layer: The layer number to get usage for

    Returns:
        Dictionary of {filament_id: cumulative_mm} for each filament used
        up to target_layer. Returns empty dict if no data available.
    """
    if not layer_usage:
        return {}

    # Find the highest recorded layer <= target_layer
    # (we store snapshots at layer changes, so we need the closest one)
    relevant_layers = [layer for layer in layer_usage if layer <= target_layer]
    if not relevant_layers:
        return {}

    max_layer = max(relevant_layers)
    return layer_usage.get(max_layer, {})


def extract_filament_properties_from_3mf(file_path: Path) -> dict[int, dict]:
    """Extract filament properties (density, diameter, type) from 3MF metadata.

    Args:
        file_path: Path to the 3MF file

    Returns:
        Dictionary mapping filament IDs to their properties:
        {filament_id: {"diameter": 1.75, "density": 1.24, "type": "PLA"}, ...}

        Note: filament_id is 1-based (matches slot_id in slice_info.config)
    """
    properties: dict[int, dict] = {}
    try:
        with zipfile.ZipFile(file_path, "r") as zf:
            # Try slice_info.config first for filament types
            if "Metadata/slice_info.config" in zf.namelist():
                content = zf.read("Metadata/slice_info.config").decode()
                root = ET.fromstring(content)
                for f in root.findall(".//filament"):
                    try:
                        # id is 1-based in slice_info.config
                        fid = int(f.get("id", 0))
                        properties[fid] = {
                            "type": f.get("type", "PLA"),
                            "diameter": DEFAULT_FILAMENT_DIAMETER,
                            "density": DEFAULT_FILAMENT_DENSITY,
                        }
                    except ValueError:
                        pass  # Skip filament entries with unparseable IDs

            # Try project_settings.config for density values
            if "Metadata/project_settings.config" in zf.namelist():
                content = zf.read("Metadata/project_settings.config").decode()
                try:
                    data = json.loads(content)
                    densities = data.get("filament_density", [])
                    for i, density in enumerate(densities):
                        # project_settings uses 0-based indexing, convert to 1-based
                        fid = i + 1
                        if fid not in properties:
                            properties[fid] = {
                                "type": "",
                                "diameter": DEFAULT_FILAMENT_DIAMETER,
                            }
                        try:
                            properties[fid]["density"] = float(density)
                        except (ValueError, TypeError):
                            properties[fid]["density"] = DEFAULT_FILAMENT_DENSITY
                except json.JSONDecodeError:
                    pass  # Skip malformed project_settings.config JSON
    except Exception:
        pass  # Return whatever properties were collected before the error

    return properties


def _first_settings_id(value: object) -> str | None:
    """A ``*_settings_id`` value is usually a string, occasionally a list (one
    entry per extruder). Return the first non-empty string, else None."""
    if isinstance(value, str):
        return value.strip() or None
    if isinstance(value, list):
        for item in value:
            if isinstance(item, str) and item.strip():
                return item.strip()
    return None


def extract_embedded_presets_from_3mf(zf: zipfile.ZipFile) -> dict[str, str | None]:
    """Read the printer / process preset names a 3MF project was prepared with.

    BambuStudio / OrcaSlicer write the chosen preset names into
    ``Metadata/project_settings.config`` (``printer_settings_id`` and
    ``print_settings_id``). The SliceModal uses them to default its printer
    and process dropdowns to what the file was sliced for (#1325) instead of
    blindly taking the first listed preset.

    Returns ``{"printer": <name|None>, "process": <name|None>}``. Every failure
    mode (missing config, malformed JSON, unexpected shape) yields ``None``
    values so the modal falls back to its own defaults.
    """
    result: dict[str, str | None] = {"printer": None, "process": None}
    try:
        if "Metadata/project_settings.config" not in zf.namelist():
            return result
        data = json.loads(zf.read("Metadata/project_settings.config").decode())
    except (KeyError, ValueError, OSError):
        return result
    if not isinstance(data, dict):
        return result
    result["printer"] = _first_settings_id(data.get("printer_settings_id"))
    result["process"] = _first_settings_id(data.get("print_settings_id"))
    return result


# Ceiling on the dense per-slot form below. Deliberately larger than the 32
# entries a print command carries, so a legitimate file is never silently
# truncated at the limit -- it is either usable or rejected outright.
_MAX_DENSE_FILAMENT_SLOTS = 64


def extract_slot_extruders_from_3mf(file_path: Path, plate_id: int | None = None) -> list[int] | None:
    """Per-slot extruder assignment as a dense list, or None (#2800).

    Same data as :func:`extract_nozzle_mapping_from_3mf`, reshaped for the
    dispatcher: index 0 is filament slot 1, and a slot this file does not
    print is ``-1``. Nozzle-rack printers (H2C) need it to build the physical
    ``nozzle_mapping`` the firmware expects — without one they fall back to
    picking a nozzle themselves, which can level with one hotend and print
    with another, several millimetres off the bed.

    ``plate_id`` scopes the answer to the plate actually being dispatched. A
    multi-plate 3MF carries one filament list per plate and they need not
    agree, so without it a slot can take its extruder from a plate this print
    is not going to run.

    Takes a path rather than an open archive because the dispatcher is
    handling the file, not the zip, and a broken file there must not take the
    print down: an unreadable or non-3MF path returns None, and the caller
    dispatches exactly as it did before this existed.
    """
    try:
        with zipfile.ZipFile(file_path) as zf:
            by_slot = extract_nozzle_mapping_from_3mf(zf, plate_id=plate_id)
    except (zipfile.BadZipFile, OSError) as exc:
        logger.warning("Failed to read nozzle mapping from %s: %s", file_path, exc)
        return None

    if not by_slot:
        return None

    # The slot IDs are whatever the file says, so the dense form has to be
    # bounded before it is built: a corrupt or hostile 3MF declaring
    # `filament id="50000000"` would otherwise allocate a fifty-million-entry
    # list here, on the dispatch path. Nothing above 32 is usable anyway --
    # that is the length of the array the printer is sent.
    highest_slot = max(by_slot)
    if highest_slot < 1 or highest_slot > _MAX_DENSE_FILAMENT_SLOTS:
        logger.warning(
            "Ignoring nozzle mapping from %s: highest filament slot %s is out of range",
            file_path,
            highest_slot,
        )
        return None
    return [by_slot.get(slot, -1) for slot in range(1, highest_slot + 1)]


@dataclass(frozen=True)
class RackGroup:
    """One filament group on a nozzle-rack plate, and what hotend it needs.

    A group is the slicer's *logical* nozzle. On an H2C the rack carriage hosts
    six of them, so several groups share one extruder index -- which is exactly
    the case ``extract_nozzle_mapping_from_3mf`` refuses to answer, because the
    physical rack position per group is the operator's choice and is stated
    nowhere in the file.
    """

    group_id: int
    on_rack: bool
    nozzle_diameter: str
    volume_type: str
    # Only a hint, for preferring a rack position already loaded with this
    # colour. Excluded from equality on purpose: two filaments may share a
    # group and differ in colour without the group being contradictory, and
    # the agreement check below must not reject that file.
    filament_color: str = field(default="", compare=False)


@dataclass(frozen=True)
class RackPlan:
    """Everything a rack dispatch needs from the 3MF, short of the choice itself.

    ``slot_groups`` is dense: index 0 is filament slot 1, and a slot the plate
    does not print is ``-1``, matching :func:`extract_slot_extruders_from_3mf`.
    ``groups`` is keyed by group id.
    """

    slot_groups: list[int]
    groups: dict[int, RackGroup]

    @property
    def rack_group_ids(self) -> list[int]:
        """Groups needing a rack position, lowest first, for stable assignment."""
        return sorted(gid for gid, group in self.groups.items() if group.on_rack)

    def group_dicts(self) -> dict[int, dict]:
        """The groups as plain dicts, the form the resolver and the API take.

        Keeps one definition of the shape rather than two that can drift: the
        dispatcher resolves against it and the print dialog renders from it.
        """
        return {
            gid: {
                "on_rack": group.on_rack,
                "nozzle_diameter": group.nozzle_diameter,
                "volume_type": group.volume_type,
                "filament_color": group.filament_color,
            }
            for gid, group in self.groups.items()
        }


def extract_rack_plan_from_3mf(file_path: Path, plate_id: int | None = None) -> RackPlan | None:
    """What a nozzle-rack plate needs per group, or None (#1784).

    :func:`extract_nozzle_mapping_from_3mf` answers "which carriage" and
    withholds the whole mapping when a plate needs several hotends off one
    rack. This answers the question underneath it -- which groups exist, which
    of them are rack-bound, and what nozzle each one wants -- so the caller can
    pair it with a chosen rack position and build a mapping the other function
    cannot.

    Measured basis (maintainer's H2C, 2026-08-14): the same plate was sent
    twice with different rack picks and every member of the two 3MFs was
    identical bar float noise -- ``group_id`` values, the toolchange stream and
    the ``NOZZLE_CHANGE`` markers included. The pick lives only in the
    dispatched ``nozzle_mapping``, so nothing here can or should derive it.

    Returns None whenever the plate cannot be described completely: a partial
    plan would place some slots and leave others at "not printed", which is the
    contradiction the firmware rejects as HMS 0500-4047.

    Takes a path rather than an open archive for the same reason
    :func:`extract_slot_extruders_from_3mf` does -- the dispatcher is holding
    the file, and a broken one must not take the print down.
    """
    try:
        with zipfile.ZipFile(file_path) as zf:
            return _rack_plan(zf, plate_id=plate_id)
    except (zipfile.BadZipFile, OSError) as exc:
        logger.warning("Failed to read rack plan from %s: %s", file_path, exc)
        return None
    except Exception:
        logger.exception("Unreadable rack plan in %s", file_path)
        return None


def _rack_plan(zf: zipfile.ZipFile, plate_id: int | None) -> RackPlan | None:
    """Body of :func:`extract_rack_plan_from_3mf`, on an already-open archive."""
    names = zf.namelist()
    if "Metadata/project_settings.config" not in names:
        return None
    if "Metadata/slice_info.config" not in names:
        return None

    data = json.loads(zf.read("Metadata/project_settings.config").decode())
    physical_extruder_map = data.get("physical_extruder_map")
    if not physical_extruder_map or len(physical_extruder_map) <= 1:
        return None

    # Which extruder index is the rack, taken from the file rather than a
    # constant: the rack is the carriage that can address more than one nozzle.
    # `extruder_max_nozzle_count` is ['1', '6'] on an H2C, and reading it here
    # means a future rack of a different size needs no change.
    rack_indices: set[int] = set()
    for index, count in enumerate(data.get("extruder_max_nozzle_count") or []):
        try:
            if int(count) > 1:
                rack_indices.add(index)
        except (TypeError, ValueError):
            return None
    if not rack_indices:
        return None

    si_root = ET.fromstring(zf.read("Metadata/slice_info.config").decode())
    plates = _plates_in_scope(si_root, plate_id)
    group_extruders = _group_extruder_indices(plates)
    if not group_extruders:
        return None

    filament_elems = [elem for plate in plates for elem in plate.findall(".//filament")]
    if not filament_elems:
        return None

    slot_groups: dict[int, int] = {}
    groups: dict[int, RackGroup] = {}
    for elem in filament_elems:
        group_id_str = elem.get("group_id")
        slot_id_str = elem.get("id")
        if group_id_str is None or not slot_id_str:
            # One ungrouped filament makes the plan partial, and a partial plan
            # dispatches the ungrouped slot as unprinted.
            return None
        try:
            group_id = int(group_id_str)
            slot_id = int(slot_id_str)
        except (TypeError, ValueError):
            return None

        extruder_index = group_extruders.get(group_id)
        if extruder_index is None or not 0 <= extruder_index < len(physical_extruder_map):
            return None

        # Two plates in scope may name the same slot; they must agree, or the
        # dispatched plate is ambiguous.
        if slot_groups.setdefault(slot_id, group_id) != group_id:
            return None

        group = RackGroup(
            group_id=group_id,
            on_rack=extruder_index in rack_indices,
            nozzle_diameter=(elem.get("nozzle_diameter") or "").strip(),
            volume_type=(elem.get("volume_type") or "").strip(),
            filament_color=(elem.get("color") or "").strip(),
        )
        # Filaments sharing a group must want the same hotend, or "the group is
        # one nozzle" is not true and no single position can serve them.
        if groups.setdefault(group_id, group) != group:
            return None

    highest_slot = max(slot_groups)
    if highest_slot < 1 or highest_slot > _MAX_DENSE_FILAMENT_SLOTS:
        return None

    return RackPlan(
        slot_groups=[slot_groups.get(slot, -1) for slot in range(1, highest_slot + 1)],
        groups=groups,
    )


def _plates_in_scope(si_root: XmlElement, plate_id: int | None) -> list[XmlElement]:
    """The ``<plate>`` elements a lookup should read, narrowed to one if asked.

    A 3MF holds every plate in the project, each with its own filament list, and
    two plates may assign the same slot to different extruders. Falls back to
    every plate when no id is given or none matches, which is what this module
    did before plates were distinguished at all.
    """
    plates = si_root.findall(".//plate")
    if not plates:
        return [si_root]
    if plate_id is None:
        return plates
    for plate in plates:
        for metadata in plate.findall("metadata"):
            if metadata.get("key") != "index":
                continue
            try:
                if int(metadata.get("value") or "") == plate_id:
                    return [plate]
            except (TypeError, ValueError):
                pass
    return plates


def _group_extruder_indices(plates: list[XmlElement]) -> dict[int, int] | None:
    """Map each filament group to the slicer extruder index it prints on.

    ``slice_info.config`` states this directly, as ``<nozzle id="<group>"
    extruder_id="<1-based extruder>"/>``. Reading it matters on nozzle-rack
    printers, where the group id is *not* an extruder index: the H2C's rack
    carriage can host six hotends (``extruder_max_nozzle_count`` is ``['1',
    '6']``), so the slicer emits more groups than the machine has extruders and
    several groups share one carriage. A plate of the reporter's carried groups
    0, 1 and 2 against a two-entry ``physical_extruder_map``.

    Returns None when the file states no table, or when two plates in scope
    disagree about a group — in which case the caller keeps treating the group
    id as the extruder index, which is what every H2D file in practice wants
    and what this module has always done.
    """
    table: dict[int, int] = {}
    for plate in plates:
        for nozzle in plate.findall(".//nozzle"):
            try:
                group_id = int(nozzle.get("id") or "")
                extruder_index = int(nozzle.get("extruder_id") or "") - 1
            except (TypeError, ValueError):
                return None
            if extruder_index < 0:
                return None
            if table.setdefault(group_id, extruder_index) != extruder_index:
                return None
    return table or None


def extract_nozzle_mapping_from_3mf(zf: zipfile.ZipFile, plate_id: int | None = None) -> dict[int, int] | None:
    """Extract per-slot nozzle/extruder mapping from a 3MF file.

    On dual-nozzle printers (H2D, H2D Pro), each filament slot is assigned to a
    specific nozzle. The slicer may override user preferences when using "Auto For
    Flush" mode, so the actual assignment comes from slice_info.config group_id
    attributes, not from the user's filament_nozzle_map preference.

    Priority:
        1. group_id on <filament> elements in slice_info.config (actual assignment),
           resolved through the file's own group-to-extruder table
        2. filament_nozzle_map in project_settings.config (user preference fallback)

    Both are mapped through physical_extruder_map to get MQTT extruder IDs (0=right, 1=left).

    Returns None rather than a partial answer whenever a filament the plate
    prints cannot be placed. The gap does not stay a gap downstream: the dense
    form fills it with -1, which already means "slot not printed", and
    dispatching that against an ams_mapping that *does* name a tray for the slot
    is a contradiction the firmware rejects outright with HMS 0500-4047, "the
    available hotend quantity or model does not match the sliced file". Giving
    no mapping at all costs only the firmware's own nozzle pick.

    Args:
        zf: An open ZipFile of the 3MF archive
        plate_id: 1-based plate to read, or None for every plate in the file

    Returns:
        Dictionary mapping {slot_id: extruder_id} for dual-nozzle files,
        or None if single-nozzle, missing data, unplaceable, or parse error.
    """
    try:
        if "Metadata/project_settings.config" not in zf.namelist():
            return None

        content = zf.read("Metadata/project_settings.config").decode()
        data = json.loads(content)

        physical_extruder_map = data.get("physical_extruder_map")
        if not physical_extruder_map or len(physical_extruder_map) <= 1:
            return None  # Single-nozzle printer

        # Check if only one extruder is active.
        # If so, we can skip the mapping and just assign all slots to that extruder.
        # extruder_nozzle_stats format: ["Standard#0|High Flow#0", "Standard#1"]
        # Each entry = one extruder. Format: <NozzleVolumeType>#<count>[|...]
        # #N is the count of physical nozzles of that type (0 = none installed).
        # Types: Standard, High Flow, Hybrid, TPU High Flow

        active_extruders = []
        for stats_str in data.get("extruder_nozzle_stats") or []:
            nozzle_counts = [n.partition("#")[2] for n in stats_str.split("|")]
            active_extruders.append(1 if any(c not in ("0", "") for c in nozzle_counts) else 0)

        # Parse slice_info once: needed by both the single-active shortcut
        # (to verify the slice is actually single-group, #1825) and Priority 1.
        si_root: XmlElement | None = None
        filament_elems: list[XmlElement] = []
        group_extruders: dict[int, int] | None = None
        distinct_group_ids: set[int] = set()
        if "Metadata/slice_info.config" in zf.namelist():
            si_content = zf.read("Metadata/slice_info.config").decode()
            si_root = ET.fromstring(si_content)
            plates = _plates_in_scope(si_root, plate_id)
            group_extruders = _group_extruder_indices(plates)
            filament_elems = [elem for plate in plates for elem in plate.findall(".//filament")]
            for filament_elem in filament_elems:
                gid = filament_elem.get("group_id")
                if gid is not None:
                    try:
                        distinct_group_ids.add(int(gid))
                    except (ValueError, TypeError):
                        pass

            # Two groups on one extruder means that extruder is a nozzle rack,
            # and the plate wants a *different* hotend from it per group. Which
            # physical rack slot each group takes is the slicer's own choice
            # against the rack's live contents and is stated nowhere in the
            # file -- on the plate that prompted this, both rack groups carry
            # identical nozzle_diameter and volume_type, and BambuStudio still
            # dispatched them to positions 16 and 18 (captured 2026-08-13
            # 17:20; that print completed). Nothing here can reproduce that
            # choice, and answering anyway is what printed in mid-air, so the
            # whole mapping is withheld and the firmware picks for itself.
            if group_extruders and len(set(group_extruders.values())) < len(group_extruders):
                logger.warning(
                    "Ignoring nozzle mapping: groups %s share extruders %s, so the plate "
                    "needs more than one nozzle from a rack and the physical positions "
                    "are not derivable from the file",
                    sorted(group_extruders),
                    sorted(set(group_extruders.values())),
                )
                return None

        # Single-active shortcut: only safe when the slice actually uses one
        # group. extruder_nozzle_stats can under-report a second installed
        # nozzle when its volume-type differs from the profile's enumerated
        # types (HT-AMS / High-Flow asymmetry on H2D, #1825); without this
        # guard the shortcut collapses a real multi-extruder slice onto one
        # nozzle and the group_id mapping below is skipped.
        if sum(active_extruders) == 1 and len(distinct_group_ids) <= 1:
            nozzle_mapping: dict[int, int] = {}
            active_idx = active_extruders.index(1)
            target_extruder = int(physical_extruder_map[active_idx])
            for filament_elem in filament_elems:
                try:
                    nozzle_mapping[int(filament_elem.get("id"))] = target_extruder
                except (ValueError, TypeError):
                    pass
            return nozzle_mapping or None

        # Priority 1: Use group_id from slice_info filament elements.
        # This reflects the actual slicer assignment (respects "Auto For Flush").
        nozzle_mapping: dict[int, int] = {}
        ungrouped = 0
        for filament_elem in filament_elems:
            group_id_str = filament_elem.get("group_id")
            filament_id_str = filament_elem.get("id")
            if not filament_id_str:
                continue
            if group_id_str is None:
                # Counted rather than returned on: a file where *no* filament
                # carries a group falls through to Priority 2 as it always has.
                # Only a file that groups some and not others is unplaceable.
                ungrouped += 1
                continue
            try:
                group_id = int(group_id_str)
                slot_id = int(filament_id_str)
            except (ValueError, TypeError):
                logger.warning(
                    "Ignoring nozzle mapping: unreadable filament id=%r group_id=%r",
                    filament_id_str,
                    group_id_str,
                )
                return None
            # The group id is an extruder index only where the file states no
            # table of its own — true of every H2D slice, not of an H2C one.
            extruder_index = group_id if group_extruders is None else group_extruders.get(group_id)
            if extruder_index is None or not 0 <= extruder_index < len(physical_extruder_map):
                logger.warning(
                    "Ignoring nozzle mapping: filament slot %s is in group %s, which "
                    "resolves to extruder %r outside physical_extruder_map %r",
                    slot_id,
                    group_id,
                    extruder_index,
                    physical_extruder_map,
                )
                return None
            nozzle_mapping[slot_id] = int(physical_extruder_map[extruder_index])

        if nozzle_mapping and ungrouped:
            logger.warning(
                "Ignoring nozzle mapping: %d filament(s) carry no group_id while %d do, "
                "so the ungrouped slots would dispatch as unprinted",
                ungrouped,
                len(nozzle_mapping),
            )
            return None

        if nozzle_mapping:
            return nozzle_mapping

        # Priority 2: Fall back to filament_nozzle_map (user preference).
        # This is correct when the user manually assigned nozzles, but may be
        # wrong when the slicer overrides via "Auto For Flush".
        filament_nozzle_map = data.get("filament_nozzle_map")
        if not filament_nozzle_map:
            return None

        for i, slicer_ext_str in enumerate(filament_nozzle_map):
            slot_id = i + 1
            try:
                slicer_ext = int(slicer_ext_str)
                if slicer_ext < len(physical_extruder_map):
                    nozzle_mapping[slot_id] = int(physical_extruder_map[slicer_ext])
            except (ValueError, TypeError, IndexError):
                pass

        return nozzle_mapping if nozzle_mapping else None
    except Exception:
        return None


@dataclass(frozen=True)
class PlateMetadata:
    """Combined per-plate slice_info.config values from a single 3MF parse.

    Bundles the three fields the queue listing needs so a queue poll opens and
    parses each 3MF once instead of three times (#2573). ``filament_usage`` is
    the full per-filament list (other callers — usage tracking, Spoolman — need
    it); ``filament_used_grams`` is its ``used_g`` sum, precomputed here so the
    queue path doesn't re-sum on every hit.
    """

    print_time_seconds: int | None = None
    filament_usage: list[dict] = field(default_factory=list)
    bed_type: str | None = None
    filament_used_grams: float = 0.0


_EMPTY_PLATE_METADATA = PlateMetadata()

# Revision-keyed cache for parsed per-plate metadata. Queue polling re-lists the
# same unchanged 3MFs every few seconds per connected client (#2573); without a
# cache each row costs a ZIP open + XML parse. The key includes the file's
# mtime_ns and size so a replaced or edited file transparently gets a fresh
# entry — no manual invalidation needed. Bounded LRU + lock so it stays small
# and is safe to touch from worker threads.
_PLATE_METADATA_CACHE: "OrderedDict[tuple, PlateMetadata]" = OrderedDict()
_PLATE_METADATA_CACHE_LOCK = Lock()
_PLATE_METADATA_CACHE_MAX = 512


def clear_plate_metadata_cache() -> None:
    """Drop all cached per-plate metadata (used by tests)."""
    with _PLATE_METADATA_CACHE_LOCK:
        _PLATE_METADATA_CACHE.clear()


def _parse_plate_metadata_uncached(file_path: Path, plate_id: int | None) -> PlateMetadata:
    """Open the 3MF once and pull print time, filament usage and bed type.

    Replicates the per-field ``plate_id=None`` behaviour of the three legacy
    helpers exactly: usage collects every ``<filament>`` in the file, while
    print time and bed type come from the first ``<plate>``.
    """
    try:
        with zipfile.ZipFile(file_path, "r") as zf:
            if "Metadata/slice_info.config" not in zf.namelist():
                return _EMPTY_PLATE_METADATA
            content = zf.read("Metadata/slice_info.config").decode()
            root = ET.fromstring(content)
    except Exception as e:
        logger.warning("Failed to read plate metadata from %s: %s", file_path, e)
        return _EMPTY_PLATE_METADATA

    def _plate_index(plate_elem) -> int | None:
        for meta in plate_elem.findall("metadata"):
            if meta.get("key") == "index":
                try:
                    return int(meta.get("value", "0"))
                except ValueError:
                    return None
        return None

    def _collect_filaments(plate_elem) -> list[dict]:
        out: list[dict] = []
        for f in plate_elem.findall("filament"):
            filament_id = f.get("id")
            # Both the used_g float() and the id int() must stay inside the guard:
            # a non-numeric id or used_g is silently skipped (matches the legacy
            # helpers, which tolerated garbage rows rather than raising — a raise
            # here would 500 the whole queue listing).
            try:
                used_amount = float(f.get("used_g", "0"))
                if filament_id:
                    out.append(
                        {
                            "slot_id": int(filament_id),
                            "used_g": used_amount,
                            "type": f.get("type", ""),
                            "color": f.get("color", ""),
                        }
                    )
            except (ValueError, TypeError):
                continue
        return out

    print_time: int | None = None
    bed_type: str | None = None
    filament_usage: list[dict] = []
    matched_plate = None

    if plate_id is not None:
        for plate_elem in root.findall(".//plate"):
            if _plate_index(plate_elem) == plate_id:
                matched_plate = plate_elem
                break
    else:
        matched_plate = root.find(".//plate")

    if matched_plate is not None:
        for meta in matched_plate.findall("metadata"):
            key = meta.get("key")
            if key == "prediction" and print_time is None:
                try:
                    print_time = int(meta.get("value", "0"))
                except ValueError:
                    print_time = None
            elif key == "curr_bed_type" and meta.get("value"):
                bed_type = (meta.get("value") or "").strip()

    if plate_id is not None:
        if matched_plate is not None:
            filament_usage = _collect_filaments(matched_plate)
    else:
        # Legacy plate_id=None usage: every filament in the file, not just plate 1.
        for f in root.findall(".//filament"):
            filament_id = f.get("id")
            # int()/float() both guarded — a garbage id/used_g row is skipped, not raised.
            try:
                used_amount = float(f.get("used_g", "0"))
                if filament_id:
                    filament_usage.append(
                        {
                            "slot_id": int(filament_id),
                            "used_g": used_amount,
                            "type": f.get("type", ""),
                            "color": f.get("color", ""),
                        }
                    )
            except (ValueError, TypeError):
                continue

    return PlateMetadata(
        print_time_seconds=print_time,
        filament_usage=filament_usage,
        bed_type=bed_type,
        filament_used_grams=sum(f["used_g"] for f in filament_usage),
    )


def extract_plate_metadata_from_3mf(file_path: Path, plate_id: int | None = None) -> PlateMetadata:
    """Return combined per-plate metadata, cached by file revision (#2573).

    The result is keyed by ``(path, plate_id, mtime_ns, size)`` so an unchanged
    file is parsed at most once; a replaced/edited file re-parses automatically.
    The returned ``PlateMetadata`` is shared and MUST be treated as read-only —
    callers that need a mutable filament list get a copy from the wrappers below.
    """
    file_path = Path(file_path)
    try:
        stat = file_path.stat()
    except OSError:
        # File missing/unreadable: parse (which will return empty) but don't
        # cache — the file may appear later and we don't want a sticky miss.
        return _parse_plate_metadata_uncached(file_path, plate_id)

    key = (str(file_path), plate_id, stat.st_mtime_ns, stat.st_size)
    with _PLATE_METADATA_CACHE_LOCK:
        cached = _PLATE_METADATA_CACHE.get(key)
        if cached is not None:
            _PLATE_METADATA_CACHE.move_to_end(key)
            return cached

    metadata = _parse_plate_metadata_uncached(file_path, plate_id)

    with _PLATE_METADATA_CACHE_LOCK:
        _PLATE_METADATA_CACHE[key] = metadata
        _PLATE_METADATA_CACHE.move_to_end(key)
        while len(_PLATE_METADATA_CACHE) > _PLATE_METADATA_CACHE_MAX:
            _PLATE_METADATA_CACHE.popitem(last=False)
    return metadata


def extract_filament_usage_from_3mf(file_path: Path, plate_id: int | None = None) -> list[dict]:
    """Extract per-filament total usage from 3MF slice_info.config.

    This extracts the slicer-estimated total usage per filament slot,
    not the per-layer breakdown.

    Args:
        file_path: Path to the 3MF file
        plate_id: Optional plate index to filter for (for multi-plate files)

    Returns:
        List of filament usage dictionaries:
        [{"slot_id": 1, "used_g": 50.5, "type": "PLA", "color": "#FF0000"}, ...]
    """
    # Delegate to the cached combined parse (#2573). Return fresh dicts so callers
    # that mutate the list don't corrupt the shared cached PlateMetadata.
    return [dict(f) for f in extract_plate_metadata_from_3mf(file_path, plate_id).filament_usage]


def extract_print_time_from_3mf(file_path: Path, plate_id: int | None = None) -> int | None:
    """Extract the slicer's predicted print time from a 3MF's slice_info.config.

    Multi-plate 3MFs carry one ``<plate><metadata key="prediction" .../></plate>``
    per plate. The archive-level `print_time_seconds` is the sum across all plates
    (see services/archive.py:200-264, #1593). For per-plate UI / notifications,
    callers re-read the 3MF and request the specific plate's value via this helper.

    Args:
        file_path: Path to the 3MF file
        plate_id: Plate index to filter for; if None, returns the first plate's
            ``prediction`` (matches the legacy single-plate read).

    Returns:
        Predicted print time in seconds, or None if not found / unparseable.
    """
    return extract_plate_metadata_from_3mf(file_path, plate_id).print_time_seconds


def extract_bed_type_from_3mf(file_path: Path, plate_id: int | None = None) -> str | None:
    """Extract the build plate type (`curr_bed_type`) for a specific plate (#1281).

    ``archive.bed_type`` is captured at ingest time but is one value per archive
    (the first plate's `curr_bed_type` — see services/archive.py:235). For a
    multi-plate 3MF where different plates target different beds (e.g. a 40-plate
    file mixing PEI + Engineering), the archive-level value lies. When a queue
    item or print modal targets a specific plate, this re-reads the 3MF and
    returns that plate's actual bed type.

    Args:
        file_path: Path to the 3MF file
        plate_id: Plate index to filter for; if None, returns the first plate's
            ``curr_bed_type`` (matches the archive-level capture).

    Returns:
        Bed type string (e.g. "Textured PEI Plate"), or None if not found.
    """
    return extract_plate_metadata_from_3mf(file_path, plate_id).bed_type


# Bed temperature is not one key in a BambuStudio project. Every plate type has
# its own per-filament array, and the plate actually fitted is named separately
# in ``curr_bed_type`` -- so reading a bed temperature means picking the array
# the plate points at. Keys and mapping are BambuStudio's own
# ``get_bed_temp_1st_layer_key`` / ``get_bed_temp_key`` (PrintConfig.hpp), and
# the plate names are the ``curr_bed_type`` enum values (PrintConfig.cpp).
# First-layer temperature first: that is what the printer heats to before the
# print starts, which is what preheat is trying to reach.
#
# ``Default Plate`` is deliberately absent -- BambuStudio maps it to no key at
# all, so there is nothing to read and guessing a plate would invent a bed
# temperature the slice never specified.
_BED_TEMP_KEYS: dict[str, tuple[str, str]] = {
    "Cool Plate": ("cool_plate_temp_initial_layer", "cool_plate_temp"),
    "Engineering Plate": ("eng_plate_temp_initial_layer", "eng_plate_temp"),
    "High Temp Plate": ("hot_plate_temp_initial_layer", "hot_plate_temp"),
    "Textured PEI Plate": ("textured_plate_temp_initial_layer", "textured_plate_temp"),
    "Supertack Plate": ("supertack_plate_temp_initial_layer", "supertack_plate_temp"),
}

# Fallback for a config that names no plate: the Orca/PrusaSlicer spelling,
# which is a single value rather than a per-plate array.
_GENERIC_BED_TEMP_KEYS = ("bed_temperature_initial_layer", "bed_temperature")


def _plate_temperature(val) -> int | None:
    """Bed temperature from one plate-temperature entry, or None.

    The plate arrays carry one entry per filament in the project, and a 0 means
    that filament cannot print on this plate. The bed only has one temperature,
    so the print runs at the highest its filaments ask for -- taking entry 0 the
    way the neighbouring scalar settings do would store a 0 for any project
    whose first filament is not one this plate is heated for.
    """
    values = val if isinstance(val, list) else [val]
    temps = []
    for entry in values:
        if isinstance(entry, bool) or not isinstance(entry, (int, float, str)):
            continue
        try:
            temps.append(int(float(entry)))
        except (TypeError, ValueError):
            continue
    return max(temps) if temps else None


def bed_temperature_from_config(data: dict) -> int | None:
    """Bed temperature for the plate *data* is sliced for, or None (#2989).

    *data* is a parsed ``Metadata/project_settings.config``. Lives here rather
    than beside the archive parser so the ingest path and the one-shot backfill
    that repairs archives written before the fix read it exactly the same way.
    """
    bed_type = str(data.get("curr_bed_type") or "").strip()
    for key in (*_BED_TEMP_KEYS.get(bed_type, ()), *_GENERIC_BED_TEMP_KEYS):
        if key not in data:
            continue
        temperature = _plate_temperature(data[key])
        # A plate array of all zeros means no filament in the project prints on
        # this plate, which is not a bed temperature -- keep looking rather than
        # recording a 0 that reads as "cold bed".
        if temperature:
            return temperature
    return None


def extract_bed_temperature_from_3mf(file_path: Path) -> int | None:
    """Read a 3MF's bed temperature straight off disk, or None.

    For the backfill, which has a ``file_path`` and nothing else. Opens only
    ``Metadata/project_settings.config`` -- the archive parser reads thumbnails,
    the model and the slice info as well, and none of that is wanted here.

    Every failure is None. Deliberately broader than the handful of exceptions a
    malformed zip is expected to raise: the caller runs inside the startup
    migration, which has no handler above it, so anything unlisted escaping here
    does not skip one archive -- it stops Bambuddy from booting, and keeps
    stopping it, because the one-shot flag is written in the same transaction
    that just rolled back. A bed temperature is not worth that.
    """
    try:
        with zipfile.ZipFile(file_path, "r") as zf:
            if "Metadata/project_settings.config" not in zf.namelist():
                return None
            data = json.loads(zf.read("Metadata/project_settings.config").decode())
        return bed_temperature_from_config(data) if isinstance(data, dict) else None
    except Exception:
        return None


# Header values exposed as `{placeholder}` substitutions inside snippets.
# Aliases let users write Prusa-style names (`{max_layer_z}`) that map onto
# Bambu/Orca header keys (`max_z_height`).
_HEADER_PLACEHOLDER_ALIASES = {
    "max_layer_z": "max_z_height",
    "max_print_height": "max_z_height",
    "total_layers": "total_layer_number",
}

_HEADER_KEY_RE = re.compile(r"^;\s*([^:]+?)\s*:\s*(.+?)\s*$")
_PLACEHOLDER_RE = re.compile(r"\{([a-zA-Z_][a-zA-Z0-9_]*)\}")
_START_GCODE_END_MARKER = "; MACHINE_START_GCODE_END"
_EXECUTABLE_BLOCK_END_MARKER = "; EXECUTABLE_BLOCK_END"


def _parse_3mf_gcode_header(content: str) -> dict[str, str]:
    """Parse the `; HEADER_BLOCK_START..END` block into a normalised dict.

    Keys are lowercased, ` [units]` suffixes stripped, and spaces converted
    to underscores so callers can look up `total_layer_number` regardless of
    whether the source line is `; total layer number: 80` or
    `; total filament length [mm] : 12155.34`.
    """
    header: dict[str, str] = {}
    in_header = False
    for raw_line in content.splitlines():
        line = raw_line.strip()
        if line == "; HEADER_BLOCK_START":
            in_header = True
            continue
        if line == "; HEADER_BLOCK_END":
            break
        if not in_header:
            continue
        m = _HEADER_KEY_RE.match(line)
        if not m:
            continue
        key, value = m.group(1), m.group(2)
        key = re.sub(r"\s*\[[^\]]*\]\s*$", "", key)
        key = key.strip().lower().replace(" ", "_")
        header[key] = value
    return header


def _plate_number_of(name: str) -> int | None:
    """Plate index encoded in a ``…/plate_<n>.gcode`` member, or None.

    Parsed as an int rather than string-matched so a zero-padded
    ``plate_01.gcode`` resolves to the same 1 the plates endpoint reports.
    """
    marker = "plate_"
    idx = name.rfind(marker)
    if idx < 0 or not name.endswith(".gcode"):
        return None
    try:
        return int(name[idx + len(marker) : -len(".gcode")])
    except ValueError:
        return None


def select_plate_gcode_name(names: list[str], plate_id: int | None) -> str | None:
    """The ``.gcode`` member for exactly ``plate_id``, or None if it isn't there.

    Returns None for a ``plate_id`` the file doesn't hold — callers that want a
    fallback compose this with ``default_plate_gcode_name``; callers serving a
    user's explicit plate choice want the None so they can 404 instead of
    quietly rendering a different plate.
    """
    if plate_id is None:
        return None
    for name in names:
        if name.endswith(".gcode") and _plate_number_of(name) == plate_id:
            return name
    return None


def default_plate_gcode_name(names: list[str]) -> str | None:
    """The ``.gcode`` member to show when no plate was asked for.

    The lowest plate number, NOT the first member in the archive: zip order is
    whatever the slicer happened to write, and Bambu Studio does not write
    plates in order — a two-plate file measured here stores ``plate_2.gcode``
    ahead of ``plate_1.gcode``, so taking the first member opened plate 2. Files
    from slicers that don't use the plate naming convention keep the old
    first-member behaviour, since there is no numbering to sort by.
    """
    gcodes = [n for n in names if n.endswith(".gcode")]
    if not gcodes:
        return None
    numbered = [(num, n) for n in gcodes if (num := _plate_number_of(n)) is not None]
    if numbered:
        return min(numbered)[1]
    return gcodes[0]


def names_carry_gcode(names: list[str]) -> bool:
    """Is this 3MF a sliced file — does it carry printer-executable G-code?

    One definition, because several of them is the bug (#2993). The archive
    side judged a file by what it holds -- the card's GCODE badge reads the
    layer count and print time parsed out of the plate G-code, and
    ``/archives/{id}/capabilities`` scanned the zip -- while the library judged
    it by its filename. So a sliced 3MF stored as ``Foo.3mf`` rather than
    ``Foo.gcode.3mf`` carried the badge and still re-imported as a source-only
    project. This is the answer for anything asking the zip directly.

    Defers to ``default_plate_gcode_name`` rather than testing for
    ``Metadata/plate_<n>.gcode``, so a slicer that lays its output out some
    other way is judged by the same rule everywhere.
    """
    return default_plate_gcode_name(names) is not None


def carries_gcode(file_path: Path | str) -> bool:
    """``names_carry_gcode`` for a file on disk. False for anything unreadable.

    Only the zip's central directory is read — no member is decompressed — so
    this is cheap enough to run on every ingested file.
    """
    try:
        with zipfile.ZipFile(file_path, "r") as zf:
            return names_carry_gcode(zf.namelist())
    except (OSError, zipfile.BadZipFile):
        # Not a zip, gone, or unreadable. Callers treat that as "no G-code
        # visible", which is what they did before this check existed.
        return False


# The header block sits at the very top of the plate G-code. Read only that
# much: a sliced plate is routinely tens of megabytes and `ZipFile.read()`
# would inflate all of it to reach ~40 lines.
_HEADER_READ_LIMIT_BYTES = 64 * 1024


def extract_max_z_height_from_3mf(file_path: Path, plate_id: int | None = None) -> float | None:
    """Return the plate's ``max_z_height`` in mm, or None if not knowable.

    This is the Z the toolhead sat at for the final layer — the same value
    Bambu's own end G-code adds its bed-drop offset to (``G1 Z{max_layer_z +
    100}``). #2547 uses it to put the plate back into camera framing before the
    finish photo, which is only safe because it is a height the printer was
    physically at seconds earlier.

    None means "don't know" and callers must treat it as such rather than
    substituting a default: the file may be unreadable, carry no plate G-code,
    or come from a slicer that writes no ``max_z_height`` header. Guessing a
    height here would command a Z move to somewhere the nozzle has never been.
    """
    try:
        with zipfile.ZipFile(file_path, "r") as zf:
            names = zf.namelist()
            target = select_plate_gcode_name(names, plate_id) or default_plate_gcode_name(names)
            if target is None:
                return None
            with zf.open(target, "r") as fh:
                head = fh.read(_HEADER_READ_LIMIT_BYTES)
    except (OSError, zipfile.BadZipFile, KeyError) as e:
        logger.debug("max_z_height: cannot read %s: %s", file_path, e)
        return None

    raw = _parse_3mf_gcode_header(head.decode("utf-8", errors="ignore")).get("max_z_height")
    if raw is None:
        return None
    try:
        value = float(raw)
    except ValueError:
        logger.debug("max_z_height: unusable value %r in %s", raw, file_path)
        return None
    # Zero or negative means the header key is present but meaningless. Passed
    # on as a height it would become a move *toward* the bed, so drop it.
    return value if value > 0 else None


def _substitute_placeholders(snippet: str, header: dict[str, str]) -> str:
    """Replace `{var}` placeholders with header values, leaving unknowns intact."""

    def repl(m: re.Match) -> str:
        name = m.group(1)
        value = header.get(name)
        if value is None:
            alias = _HEADER_PLACEHOLDER_ALIASES.get(name)
            if alias is not None:
                value = header.get(alias)
        if value is None:
            logger.warning(
                "G-code injection: placeholder {%s} not found in 3MF header; leaving as-is",
                name,
            )
            return m.group(0)
        return value

    return _PLACEHOLDER_RE.sub(repl, snippet)


def _inject_start_at_marker(content: str, snippet: str) -> str:
    """Insert snippet immediately before `; MACHINE_START_GCODE_END`.

    The marker sits at the bottom of the printer's startup block — bed heat,
    homing, and nozzle prime are already done, so injected snippets land in
    the same place a slicer-side custom-start-gcode would. Falls back to
    prepending if the marker isn't present (older files / non-Bambu slicers).
    """
    marker_idx = content.find(_START_GCODE_END_MARKER)
    if marker_idx == -1:
        logger.warning(
            "G-code injection: '%s' not found, prepending start snippet to whole file",
            _START_GCODE_END_MARKER,
        )
        return snippet.rstrip("\n") + "\n" + content
    line_start = content.rfind("\n", 0, marker_idx)
    line_start = 0 if line_start == -1 else line_start + 1
    return content[:line_start] + snippet.rstrip("\n") + "\n" + content[line_start:]


def _inject_end_before_marker(content: str, snippet: str) -> str:
    """Insert snippet immediately before `; EXECUTABLE_BLOCK_END`.

    The end snippet must run *inside* the executable block. Bambu firmware
    (verified on a P1S) does not execute G-code that sits after
    `; EXECUTABLE_BLOCK_END`, so appending to the file end silently drops the
    snippet — auto-eject / plate-clear moves never fire. Inserting before the
    marker places the snippet after the printer's own machine-end sequence but
    still within the executed block. Falls back to appending at the file end if
    the marker isn't present.
    """
    marker_idx = content.find(_EXECUTABLE_BLOCK_END_MARKER)
    if marker_idx == -1:
        logger.warning(
            "G-code injection: '%s' not found, appending end snippet to file end",
            _EXECUTABLE_BLOCK_END_MARKER,
        )
        return content.rstrip("\n") + "\n" + snippet.rstrip("\n") + "\n"
    line_start = content.rfind("\n", 0, marker_idx)
    line_start = 0 if line_start == -1 else line_start + 1
    return content[:line_start] + snippet.rstrip("\n") + "\n" + content[line_start:]


def inject_gcode_into_3mf(
    source_path: Path,
    plate_id: int,
    start_gcode: str | None,
    end_gcode: str | None,
):
    """Create a temp copy of a 3MF with G-code injected at start/end.

    Snippets support `{placeholder}` substitution against values parsed from
    the 3MF G-code header block (e.g. `{max_layer_z}` → `16.00`). Start
    snippets are anchored to the `; MACHINE_START_GCODE_END` marker so they
    run after the printer's own startup (#422). End snippets are inserted just
    before `; EXECUTABLE_BLOCK_END` so they run inside the executable block —
    Bambu firmware (P1S) ignores g-code placed after that marker.

    The plate's `.gcode.md5` sidecar is recomputed so firmware that validates
    it against the gcode (e.g. P1S) still accepts the modified file.

    Args:
        source_path: Path to the original 3MF file.
        plate_id: Plate number (1-indexed) to inject into.
        start_gcode: G-code to insert after printer startup, or None.
        end_gcode: G-code to append, or None.

    Returns:
        Path to temp file with injected G-code, or None if injection failed.
        Caller is responsible for cleaning up the temp file.
    """
    import tempfile

    if not start_gcode and not end_gcode:
        return None

    try:
        # Find the target gcode file inside the 3MF
        with zipfile.ZipFile(source_path, "r") as zf:
            # Plate-specific gcode first, else the lowest-numbered plate.
            names = zf.namelist()
            target_gcode = select_plate_gcode_name(names, plate_id) or default_plate_gcode_name(names)
            if target_gcode is None:
                return None

            # Read and modify gcode content
            gcode_content = zf.read(target_gcode).decode("utf-8", errors="ignore")
            header = _parse_3mf_gcode_header(gcode_content)

            if start_gcode:
                resolved = _substitute_placeholders(start_gcode, header)
                # Log the post-substitution snippet so the actually-injected G-code
                # (placeholders like {max_layer_z} already resolved) is visible at DEBUG.
                logger.debug("G-code injection [%s]: resolved START snippet:\n%s", target_gcode, resolved)
                gcode_content = _inject_start_at_marker(gcode_content, resolved)
            if end_gcode:
                resolved = _substitute_placeholders(end_gcode, header)
                logger.debug("G-code injection [%s]: resolved END snippet:\n%s", target_gcode, resolved)
                gcode_content = _inject_end_before_marker(gcode_content, resolved)

            # The printer validates the plate gcode against an embedded
            # `<plate>.gcode.md5` sidecar (uppercase hex, no trailing newline).
            # Rewriting the gcode without refreshing this hash makes firmware
            # reject the file at load (P1S: HMS 0500-4003 "unable to parse"),
            # so recompute it from the exact bytes we're about to write.
            gcode_bytes = gcode_content.encode("utf-8")
            md5_name = target_gcode + ".md5"
            # Not a security hash — this reproduces Bambu's `.gcode.md5` sidecar
            # format, so flag it as non-security for the linters (ruff S324 / bandit B324).
            md5_value = hashlib.md5(gcode_bytes, usedforsecurity=False).hexdigest().upper().encode("ascii")

            # Write modified 3MF to temp file
            with tempfile.NamedTemporaryFile(delete=False, suffix=".3mf") as tmp:
                tmp_path = Path(tmp.name)

            with zipfile.ZipFile(tmp_path, "w", zipfile.ZIP_DEFLATED) as zf_write:
                for item in zf.namelist():
                    info = zf.getinfo(item)
                    if item == target_gcode:
                        zf_write.writestr(info, gcode_bytes)
                    elif item == md5_name:
                        zf_write.writestr(info, md5_value)
                    else:
                        zf_write.writestr(info, zf.read(item))

        return tmp_path

    except Exception:
        # Clean up temp file on error
        if "tmp_path" in locals() and tmp_path.exists():
            tmp_path.unlink(missing_ok=True)
        return None


def extract_project_filaments_from_3mf(zf: zipfile.ZipFile) -> list[dict]:
    """Project-wide AMS slot config from ``Metadata/project_settings.config``.

    Returns one dict per configured AMS slot in slot order (1-indexed), with
    ``type`` and ``color`` populated from the project's ``filament_type`` and
    ``filament_colour`` arrays. ``used_grams`` / ``used_meters`` are 0 because
    project_settings carries the configuration, not per-print usage — the
    fields exist for shape compatibility with the slice_info-derived list.

    The SliceModal needs this on **unsliced** project files: slice_info.config
    is empty until Bambu Studio has actually sliced the project, but the user
    can still pick filament profiles for a slice we're about to perform.
    """
    if "Metadata/project_settings.config" not in zf.namelist():
        return []
    try:
        proj = json.loads(zf.read("Metadata/project_settings.config").decode())
    except (ValueError, OSError):
        return []
    if not isinstance(proj, dict):
        return []
    types_arr = proj.get("filament_type") or []
    colors_arr = proj.get("filament_colour") or []
    slot_count = max(
        len(types_arr) if isinstance(types_arr, list) else 0, len(colors_arr) if isinstance(colors_arr, list) else 0
    )
    out: list[dict] = []
    for i in range(slot_count):
        out.append(
            {
                "slot_id": i + 1,
                "type": types_arr[i] if i < len(types_arr) and isinstance(types_arr[i], str) else "",
                "color": colors_arr[i] if i < len(colors_arr) and isinstance(colors_arr[i], str) else "",
                "used_grams": 0,
                "used_meters": 0,
            }
        )
    return out


def expand_to_project_slots(zf: zipfile.ZipFile, used: list[dict]) -> list[dict]:
    """Widen a used-only filament list to one entry per project slot.

    ``used`` is the slice_info-derived list: only the slots whose G-code
    actually consumed filament, each carrying real usage figures. That is the
    right answer for print-time AMS matching, and the wrong one for the slice
    modal, because the list the modal builds is **positional** — index 0 is
    slot 1 all the way down to the ``filament_N.json`` parts handed to the CLI.
    A source whose only used slot is 4 therefore produced a single dropdown
    whose pick the CLI bound to slot 1, leaving slot 4 — the one the model
    prints with — on whatever the source had baked in (#2712).

    Returns the project's slots in slot order, each flagged ``used_in_plate``.
    Rows present in ``used`` are kept whole, so their usage figures, resolved
    type/colour and ``tray_info_idx`` survive; the rest come from the project
    configuration with zero usage. A used slot beyond the project's slot count
    is appended rather than dropped — the caller asked for a superset, and
    silently losing the one slot that prints would be the original bug again.

    ``used`` is returned unchanged when the file carries no project settings
    to widen against: a narrower-than-ideal list still prints correctly, an
    invented one might not.
    """
    project = extract_project_filaments_from_3mf(zf)
    if not project:
        return used

    by_slot = {f["slot_id"]: f for f in used}
    out: list[dict] = []
    for slot in project:
        known = by_slot.pop(slot["slot_id"], None)
        if known is not None:
            known["used_in_plate"] = True
            out.append(known)
        else:
            slot["used_in_plate"] = False
            out.append(slot)
    # Anything slice_info reported that the project doesn't declare.
    for leftover in by_slot.values():
        leftover["used_in_plate"] = True
        out.append(leftover)
    out.sort(key=lambda f: f["slot_id"])
    return out


# BambuStudio serialises bool config options as string "1"/"0" in
# project_settings.config, but forks / older versions occasionally write real
# booleans or ints — accept anything that isn't unambiguously falsy. A missing
# key counts as off: a 3MF that never declares `enable_support` gives us no
# support intent to act on.
_SUPPORTS_DISABLED_VALUES = (False, 0, "0", "false", "False", "", None)


def supports_enabled_in_config(cfg: dict[str, object]) -> bool:
    """Whether a 3MF's ``project_settings.config`` has supports switched on.

    Shared by the callers that read support intent out of a source file so
    they agree on what "on" means: the slot extractor below and the slice
    route's process-preset support carry-over (#1881 / #2820).
    """
    return cfg.get("enable_support") not in _SUPPORTS_DISABLED_VALUES


def extract_support_filament_slots_from_3mf(zf: zipfile.ZipFile) -> set[int]:
    """Slots referenced by the process settings for support material.

    Supports aren't attached to object geometry — they're generated by
    the slicer's process pass — so :func:`extract_plate_extruder_set_from_3mf`,
    which walks per-object extruder metadata + paint_color triangles,
    doesn't see them. Callers that need the complete set of slots a
    plate print will exercise (e.g. the SliceModal's filament-
    substitution logic) must union this in — otherwise a support-only
    slot (typical PLA-model + PVA-support setup) looks "unused" and its
    user-picked profile gets silently overwritten with slot 1's,
    producing a single-material print (#1881).

    Returns the empty set when supports are disabled, ``support_filament``
    / ``support_interface_filament`` are 0 (== "same as model"), the
    project has no embedded settings, or the file isn't a valid 3MF.
    """
    if "Metadata/project_settings.config" not in zf.namelist():
        return set()
    try:
        cfg = json.loads(zf.read("Metadata/project_settings.config").decode("utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError, OSError):
        return set()
    if not isinstance(cfg, dict):
        return set()
    if not supports_enabled_in_config(cfg):
        return set()
    out: set[int] = set()
    for key in ("support_filament", "support_interface_filament"):
        raw = cfg.get(key)
        if raw is None:
            continue
        try:
            slot = int(raw)
        except (ValueError, TypeError):
            continue
        # Slot 0 means "same as model" — no dedicated slot to preserve.
        if slot > 0:
            out.add(slot)
    return out


_PAINT_COLOR_ATTR_RE = re.compile(rb'paint_color="([0-9A-Fa-f]+)"')

# Painted-face quadtree leaves include both real filament assignments and
# tiny edit artifacts (single-leaf accidents from "tried a colour, undid,
# repainted with a different one"). The threshold's only job is dropping
# accidents — anything the user spent meaningful effort on must survive.
# 5% of an object's painted triangles is well below any 60/40 / 70/30 /
# 33/33/33 split a real two- or three-colour print would hit, so all
# intentional colours are kept; one-off single-leaf paints (typically
# 0.1-1.5% in observed projects) are filtered. Note that this fallback
# path runs ONLY when the preview-slice path can't reach the sidecar; in
# the normal flow the slicer's own pruning produces the canonical list and
# this threshold isn't reached.
_PAINT_NOISE_THRESHOLD = 0.05


def extract_plate_extruder_set_from_3mf(zf: zipfile.ZipFile, plate_id: int) -> set[int]:
    """Extruder/AMS slot indices (1-indexed) used by objects on ``plate_id``.

    Three sources are unioned because Bambu Studio splits per-object extruder
    info across THREE places depending on how the user assigned colours:

    1. ``model_settings.config`` — top-level ``<metadata key="extruder">``
       on each ``<object>`` (the "default extruder" for the whole object).
    2. ``model_settings.config`` — per-``<part>`` ``<metadata key="extruder">``
       overrides (used when the user split an object into multiple parts
       with distinct filaments).
    3. ``3D/Objects/object_*.model`` — ``paint_color`` attributes on
       individual ``<triangle>`` elements (used when the user "painted" a
       face with a different filament). The encoding is a hex string where
       each nibble is a TriangleSelector tree node: ``0`` = unpainted leaf,
       ``F`` = branch (4 children follow), ``1``..``E`` = leaf painted with
       extruder N. We don't decode the tree — every leaf-paint nibble in
       the string IS the extruder number, so a flat scan over hex chars
       yields the correct set without recursive parsing.

    Without (3) the painted-face data is invisible: model_settings says
    every object on a multi-color plate uses extruder 1 by default but the
    actual print uses 3, 4, 12 etc. via face paint, so the SliceModal would
    render only one filament dropdown for what's clearly a multi-colour
    print (#1150 follow-up).
    """
    if "Metadata/model_settings.config" not in zf.namelist():
        return set()
    try:
        root = ET.fromstring(zf.read("Metadata/model_settings.config").decode())
    except (ET.ParseError, OSError):
        return set()

    # Pass 1: object → set of extruders from XML metadata (sources 1 + 2)
    # plus the per-object .model file path so we can later scan source 3.
    object_extruders: dict[str, set[int]] = {}
    object_model_paths: dict[str, list[str]] = {}
    for obj_elem in root.findall(".//object"):
        obj_id = obj_elem.get("id")
        if not obj_id:
            continue
        extruders: set[int] = set()
        top = obj_elem.find("metadata[@key='extruder']")
        if top is not None:
            try:
                v = int(top.get("value", "0"))
                if v > 0:
                    extruders.add(v)
            except (ValueError, TypeError):
                pass
        for part_elem in obj_elem.findall(".//part"):
            part_ext = part_elem.find("metadata[@key='extruder']")
            if part_ext is None:
                continue
            try:
                v = int(part_ext.get("value", "0"))
                if v > 0:
                    extruders.add(v)
            except (ValueError, TypeError):
                pass
        object_extruders[obj_id] = extruders

    # Pass 2: 3dmodel.model maps each <object id="N"> to its component
    # .model file path(s). Bambu wraps object IDs that match
    # model_settings.config IDs around <components><component
    # path="/3D/Objects/object_K.model" objectid="..." /></components>.
    # Strip xmlns prefixes on attributes so ElementTree can find them
    # without namespace gymnastics — `p:path` becomes `path` etc.
    if "3D/3dmodel.model" in zf.namelist():
        try:
            raw = zf.read("3D/3dmodel.model").decode()
            stripped = re.sub(r'xmlns:?\w*="[^"]*"', "", raw)
            stripped = re.sub(r"<(/?)\w+:", r"<\1", stripped)
            stripped = re.sub(r" \w+:(\w+=)", r" \1", stripped)
            model_root = ET.fromstring(stripped)
            for obj_elem in model_root.findall(".//object"):
                oid = obj_elem.get("id")
                if not oid:
                    continue
                comps = obj_elem.find("components")
                if comps is None:
                    continue
                paths = []
                for c in comps.findall("component"):
                    p = c.get("path")
                    if p:
                        paths.append(p.lstrip("/"))
                if paths:
                    object_model_paths[oid] = paths
        except (ET.ParseError, OSError):
            pass  # No 3dmodel — paint scan just won't apply

    # Pass 3: scan paint_color attrs in each per-object .model file. Cache
    # by file path because two objects often share the same component tree.
    paint_cache: dict[str, set[int]] = {}

    def _scan_paint(path: str) -> set[int]:
        if path in paint_cache:
            return paint_cache[path]
        out: set[int] = set()
        if path not in zf.namelist():
            paint_cache[path] = out
            return out
        try:
            data = zf.read(path)
        except OSError:
            paint_cache[path] = out
            return out
        # Per-extruder triangle coverage. Each painted triangle may have
        # multiple leaf nibbles (the quadtree subdivides the face into
        # painted regions); we count one triangle per unique extruder per
        # match so the resulting fraction is "what share of painted
        # triangles include at least one leaf with extruder N". Noise from
        # one-off edit artifacts is filtered out at the threshold below.
        extruder_triangles: dict[int, int] = {}
        total_painted = 0
        for match in _PAINT_COLOR_ATTR_RE.finditer(data):
            total_painted += 1
            seen: set[int] = set()
            for ch in match.group(1):
                # Hex digit → 4-bit value. 0 = unpainted leaf, F = branch
                # (decoded recursively but children are encoded inline, so
                # we'll see them on later iterations). 1-E = leaf painted
                # with extruder N.
                if ch in b"123456789":
                    seen.add(ch - 0x30)
                elif ch in b"ABCDEabcde":
                    seen.add((ch & 0x4F) - 0x37)
            for e in seen:
                extruder_triangles[e] = extruder_triangles.get(e, 0) + 1
        if total_painted > 0:
            cutoff = max(1, int(total_painted * _PAINT_NOISE_THRESHOLD))
            for ext, count in extruder_triangles.items():
                if count >= cutoff:
                    out.add(ext)
        paint_cache[path] = out
        return out

    # Walk plates — collect extruders for objects on the requested plate.
    used: set[int] = set()
    for plate_elem in root.findall(".//plate"):
        plater_id = None
        for meta in plate_elem.findall("metadata"):
            if meta.get("key") == "plater_id":
                try:
                    plater_id = int(meta.get("value", ""))
                except (ValueError, TypeError):
                    pass
                break
        if plater_id != plate_id:
            continue
        for inst in plate_elem.findall("model_instance"):
            for inst_meta in inst.findall("metadata"):
                if inst_meta.get("key") != "object_id":
                    continue
                obj_id = inst_meta.get("value")
                if not obj_id:
                    continue
                used.update(object_extruders.get(obj_id, set()))
                for path in object_model_paths.get(obj_id, []):
                    used.update(_scan_paint(path))
        break
    return used