"""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 change marker - M620 S: Filament/tool change (S255 = unload) - G0/G1/G2/G3 E: 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 # 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 # 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": , "process": }``. 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 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: #[|...] # #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 ```` in the file, while print time and bed type come from the first ````. """ 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 ```` 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 _select_plate_gcode_name(names: list[str], plate_id: int | None) -> str | None: """Pick a plate's ``.gcode`` member out of a 3MF namelist. Prefers ``plate_.gcode``, then falls back to the first ``.gcode`` member so single-plate files — and files from slicers that don't use the plate naming convention — still resolve. """ gcodes = [n for n in names if n.endswith(".gcode")] if not gcodes: return None if plate_id is not None: suffix = f"plate_{plate_id}.gcode" for name in gcodes: if name.endswith(suffix): return name 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: target = _select_plate_gcode_name(zf.namelist(), plate_id) 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 first one in the file. target_gcode = _select_plate_gcode_name(zf.namelist(), plate_id) 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 # `.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 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 ```` on each ``