mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
b0ecb8fd88
The forward fix reads the array the fitted plate points at, but only for archives made after it. Everything already in the library stays blank, and preheat keeps falling back to the keep-warm bed temperature whenever one of those jobs is reprinted from the queue - 0 of 455 real 3MFs had resolved. A one-shot pass re-reads the 3MF already on disk, gated by a settings flag the way #2614's repair is: the rows it cannot fill are exactly the ones it would reopen every boot. It fills NULLs only. Nothing recorded is overwritten, an archive whose file is gone stays NULL, and a corrupted 3MF is skipped rather than failing startup. The plate mapping moves to threemf_tools.bed_temperature_from_config so the ingest path and the repair cannot read a 3MF differently - the same drift move. _extract_settings_from_content is deleted. Nothing called it anywhere in the repo, and it carried the old bed_temperature mapping this issue fixed.
1737 lines
74 KiB
Python
1737 lines
74 KiB
Python
"""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]
|
||
|
||
|
||
# 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
|