mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
1136 lines
47 KiB
Python
1136 lines
47 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
|
||
|
||
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) -> 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
|
||
|
||
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:
|
||
# Find G-code file(s) - usually plate_1.gcode or Metadata/plate_1.gcode
|
||
gcode_files = [f for f in zf.namelist() if f.endswith(".gcode")]
|
||
if not gcode_files:
|
||
return None
|
||
|
||
# Use the first G-code file (typically only one per 3MF export)
|
||
gcode_path = gcode_files[0]
|
||
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
|
||
|
||
|
||
def extract_nozzle_mapping_from_3mf(zf: zipfile.ZipFile) -> 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)
|
||
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).
|
||
|
||
Args:
|
||
zf: An open ZipFile of the 3MF archive
|
||
|
||
Returns:
|
||
Dictionary mapping {slot_id: extruder_id} for dual-nozzle files,
|
||
or None if single-nozzle, missing data, 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: ET.Element | 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)
|
||
for filament_elem in si_root.findall(".//filament"):
|
||
gid = filament_elem.get("group_id")
|
||
if gid is not None:
|
||
try:
|
||
distinct_group_ids.add(int(gid))
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# 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])
|
||
if si_root is not None:
|
||
for filament_elem in si_root.findall(".//filament"):
|
||
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] = {}
|
||
if si_root is not None:
|
||
for filament_elem in si_root.findall(".//filament"):
|
||
group_id_str = filament_elem.get("group_id")
|
||
filament_id_str = filament_elem.get("id")
|
||
if group_id_str is not None and filament_id_str:
|
||
try:
|
||
group_id = int(group_id_str)
|
||
slot_id = int(filament_id_str)
|
||
if group_id < len(physical_extruder_map):
|
||
nozzle_mapping[slot_id] = int(physical_extruder_map[group_id])
|
||
except (ValueError, TypeError, IndexError):
|
||
pass
|
||
|
||
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
|
||
|
||
|
||
# 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 _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:
|
||
all_gcode = [f for f in zf.namelist() if f.endswith(".gcode")]
|
||
if not all_gcode:
|
||
return None
|
||
|
||
# Try plate-specific gcode file first
|
||
target_gcode = None
|
||
plate_pattern = f"plate_{plate_id}.gcode"
|
||
for f in all_gcode:
|
||
if f.endswith(plate_pattern):
|
||
target_gcode = f
|
||
break
|
||
|
||
# Fall back to first gcode file
|
||
if target_gcode is None:
|
||
target_gcode = all_gcode[0]
|
||
|
||
# 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 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()
|
||
# 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.
|
||
enable = cfg.get("enable_support")
|
||
if enable in (False, 0, "0", "false", "False", "", None):
|
||
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
|