threemf_tools.py 40 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965
  1. """3MF file parsing utilities for filament tracking.
  2. This module provides functions to parse Bambu Lab 3MF files and extract
  3. per-layer filament usage data from the embedded G-code. This enables
  4. accurate partial usage reporting for multi-material prints.
  5. """
  6. import hashlib
  7. import json
  8. import logging
  9. import math
  10. import re
  11. import zipfile
  12. from pathlib import Path
  13. import defusedxml.ElementTree as ET
  14. logger = logging.getLogger(__name__)
  15. # Default filament properties
  16. DEFAULT_FILAMENT_DIAMETER = 1.75 # mm
  17. DEFAULT_FILAMENT_DENSITY = 1.24 # g/cm³ (PLA)
  18. def parse_gcode_layer_filament_usage(gcode_content: str) -> dict[int, dict[int, float]]:
  19. """Parse G-code to extract per-layer, per-filament cumulative extrusion in mm.
  20. This function tracks filament extrusion across layers and tool changes,
  21. building a cumulative usage map that can be used to calculate partial
  22. usage at any layer.
  23. Args:
  24. gcode_content: The raw G-code content as a string
  25. Returns:
  26. A nested dictionary mapping layer numbers to filament usage:
  27. {layer: {filament_id: cumulative_mm}, ...}
  28. Example:
  29. {0: {0: 125.5}, 1: {0: 250.0, 1: 50.0}, 2: {0: 375.0, 1: 150.0}}
  30. This shows:
  31. - Layer 0: filament 0 used 125.5mm cumulative
  32. - Layer 1: filament 0 used 250mm cumulative, filament 1 used 50mm
  33. - Layer 2: filament 0 used 375mm cumulative, filament 1 used 150mm
  34. G-code commands parsed:
  35. - M73 L<layer>: Layer change marker
  36. - M620 S<filament>: Filament/tool change (S255 = unload)
  37. - G0/G1/G2/G3 E<amount>: Extrusion moves
  38. """
  39. layer_filaments: dict[int, dict[int, float]] = {}
  40. current_layer = 0
  41. active_filament: int | None = None
  42. cumulative_extrusion: dict[int, float] = {} # filament_id -> total mm
  43. for line in gcode_content.splitlines():
  44. line = line.strip()
  45. if not line:
  46. continue
  47. # Handle comments - skip but check for layer markers
  48. if line.startswith(";"):
  49. # Some slicers use comment-based layer markers
  50. # e.g., "; CHANGE_LAYER" or ";LAYER_CHANGE"
  51. continue
  52. # Split line into command and inline comment
  53. if ";" in line:
  54. line = line.split(";")[0].strip()
  55. # Extract command and parameters
  56. parts = line.split()
  57. if not parts:
  58. continue
  59. cmd = parts[0].upper()
  60. # Layer change: M73 L<layer>
  61. # Bambu printers use M73 with L parameter for layer indication
  62. if cmd == "M73":
  63. for part in parts[1:]:
  64. part_upper = part.upper()
  65. if part_upper.startswith("L"):
  66. try:
  67. new_layer = int(part[1:])
  68. # Save current state before layer change
  69. if cumulative_extrusion:
  70. layer_filaments[current_layer] = cumulative_extrusion.copy()
  71. current_layer = new_layer
  72. except ValueError:
  73. pass # Skip G-code lines with unparseable layer numbers
  74. # Filament change: M620 S<filament>
  75. # Bambu uses M620 for AMS filament switching
  76. # S255 means full unload (no active filament)
  77. elif cmd == "M620":
  78. for part in parts[1:]:
  79. part_upper = part.upper()
  80. if part_upper.startswith("S"):
  81. filament_str = part[1:]
  82. if filament_str == "255":
  83. # Full unload - no active filament
  84. active_filament = None
  85. else:
  86. try:
  87. # Extract digits (e.g., "0A" -> 0, "1" -> 1)
  88. match = re.match(r"(\d+)", filament_str)
  89. if match:
  90. active_filament = int(match.group(1))
  91. except (ValueError, AttributeError):
  92. pass # Skip unparseable filament switch commands
  93. # Extrusion moves: G0/G1/G2/G3 with E parameter
  94. # Only G1 typically has extrusion, but check all for safety
  95. elif cmd in ("G0", "G1", "G2", "G3"):
  96. if active_filament is None:
  97. continue
  98. for part in parts[1:]:
  99. part_upper = part.upper()
  100. if part_upper.startswith("E"):
  101. try:
  102. extrusion = float(part[1:])
  103. # Only count positive extrusion (not retractions)
  104. if extrusion > 0:
  105. current = cumulative_extrusion.get(active_filament, 0)
  106. cumulative_extrusion[active_filament] = current + extrusion
  107. except ValueError:
  108. pass # Skip G-code lines with unparseable extrusion values
  109. # Save final layer state
  110. if cumulative_extrusion:
  111. layer_filaments[current_layer] = cumulative_extrusion.copy()
  112. return layer_filaments
  113. def mm_to_grams(
  114. length_mm: float,
  115. diameter_mm: float = DEFAULT_FILAMENT_DIAMETER,
  116. density_g_cm3: float = DEFAULT_FILAMENT_DENSITY,
  117. ) -> float:
  118. """Convert filament length in mm to weight in grams.
  119. Uses the formula: mass = volume × density
  120. where volume = π × r² × length
  121. Args:
  122. length_mm: Length of filament in millimeters
  123. diameter_mm: Filament diameter in millimeters (default: 1.75)
  124. density_g_cm3: Material density in g/cm³ (default: 1.24 for PLA)
  125. Returns:
  126. Weight in grams
  127. """
  128. radius_cm = (diameter_mm / 2) / 10 # Convert mm to cm
  129. length_cm = length_mm / 10 # Convert mm to cm
  130. volume_cm3 = math.pi * radius_cm * radius_cm * length_cm
  131. return volume_cm3 * density_g_cm3
  132. def extract_layer_filament_usage_from_3mf(file_path: Path) -> dict[int, dict[int, float]] | None:
  133. """Extract per-layer filament usage from a 3MF file's embedded G-code.
  134. Args:
  135. file_path: Path to the 3MF file
  136. Returns:
  137. Dictionary mapping layers to filament usage, or None if parsing fails.
  138. Format: {layer: {filament_id: cumulative_mm}, ...}
  139. """
  140. try:
  141. with zipfile.ZipFile(file_path, "r") as zf:
  142. # Find G-code file(s) - usually plate_1.gcode or Metadata/plate_1.gcode
  143. gcode_files = [f for f in zf.namelist() if f.endswith(".gcode")]
  144. if not gcode_files:
  145. return None
  146. # Use the first G-code file (typically only one per 3MF export)
  147. gcode_path = gcode_files[0]
  148. gcode_content = zf.read(gcode_path).decode("utf-8", errors="ignore")
  149. return parse_gcode_layer_filament_usage(gcode_content)
  150. except Exception:
  151. return None
  152. def get_cumulative_usage_at_layer(
  153. layer_usage: dict[int, dict[int, float]],
  154. target_layer: int,
  155. ) -> dict[int, float]:
  156. """Get cumulative filament usage (in mm) up to and including target_layer.
  157. Args:
  158. layer_usage: The output from parse_gcode_layer_filament_usage()
  159. target_layer: The layer number to get usage for
  160. Returns:
  161. Dictionary of {filament_id: cumulative_mm} for each filament used
  162. up to target_layer. Returns empty dict if no data available.
  163. """
  164. if not layer_usage:
  165. return {}
  166. # Find the highest recorded layer <= target_layer
  167. # (we store snapshots at layer changes, so we need the closest one)
  168. relevant_layers = [layer for layer in layer_usage if layer <= target_layer]
  169. if not relevant_layers:
  170. return {}
  171. max_layer = max(relevant_layers)
  172. return layer_usage.get(max_layer, {})
  173. def extract_filament_properties_from_3mf(file_path: Path) -> dict[int, dict]:
  174. """Extract filament properties (density, diameter, type) from 3MF metadata.
  175. Args:
  176. file_path: Path to the 3MF file
  177. Returns:
  178. Dictionary mapping filament IDs to their properties:
  179. {filament_id: {"diameter": 1.75, "density": 1.24, "type": "PLA"}, ...}
  180. Note: filament_id is 1-based (matches slot_id in slice_info.config)
  181. """
  182. properties: dict[int, dict] = {}
  183. try:
  184. with zipfile.ZipFile(file_path, "r") as zf:
  185. # Try slice_info.config first for filament types
  186. if "Metadata/slice_info.config" in zf.namelist():
  187. content = zf.read("Metadata/slice_info.config").decode()
  188. root = ET.fromstring(content)
  189. for f in root.findall(".//filament"):
  190. try:
  191. # id is 1-based in slice_info.config
  192. fid = int(f.get("id", 0))
  193. properties[fid] = {
  194. "type": f.get("type", "PLA"),
  195. "diameter": DEFAULT_FILAMENT_DIAMETER,
  196. "density": DEFAULT_FILAMENT_DENSITY,
  197. }
  198. except ValueError:
  199. pass # Skip filament entries with unparseable IDs
  200. # Try project_settings.config for density values
  201. if "Metadata/project_settings.config" in zf.namelist():
  202. content = zf.read("Metadata/project_settings.config").decode()
  203. try:
  204. data = json.loads(content)
  205. densities = data.get("filament_density", [])
  206. for i, density in enumerate(densities):
  207. # project_settings uses 0-based indexing, convert to 1-based
  208. fid = i + 1
  209. if fid not in properties:
  210. properties[fid] = {
  211. "type": "",
  212. "diameter": DEFAULT_FILAMENT_DIAMETER,
  213. }
  214. try:
  215. properties[fid]["density"] = float(density)
  216. except (ValueError, TypeError):
  217. properties[fid]["density"] = DEFAULT_FILAMENT_DENSITY
  218. except json.JSONDecodeError:
  219. pass # Skip malformed project_settings.config JSON
  220. except Exception:
  221. pass # Return whatever properties were collected before the error
  222. return properties
  223. def _first_settings_id(value: object) -> str | None:
  224. """A ``*_settings_id`` value is usually a string, occasionally a list (one
  225. entry per extruder). Return the first non-empty string, else None."""
  226. if isinstance(value, str):
  227. return value.strip() or None
  228. if isinstance(value, list):
  229. for item in value:
  230. if isinstance(item, str) and item.strip():
  231. return item.strip()
  232. return None
  233. def extract_embedded_presets_from_3mf(zf: zipfile.ZipFile) -> dict[str, str | None]:
  234. """Read the printer / process preset names a 3MF project was prepared with.
  235. BambuStudio / OrcaSlicer write the chosen preset names into
  236. ``Metadata/project_settings.config`` (``printer_settings_id`` and
  237. ``print_settings_id``). The SliceModal uses them to default its printer
  238. and process dropdowns to what the file was sliced for (#1325) instead of
  239. blindly taking the first listed preset.
  240. Returns ``{"printer": <name|None>, "process": <name|None>}``. Every failure
  241. mode (missing config, malformed JSON, unexpected shape) yields ``None``
  242. values so the modal falls back to its own defaults.
  243. """
  244. result: dict[str, str | None] = {"printer": None, "process": None}
  245. try:
  246. if "Metadata/project_settings.config" not in zf.namelist():
  247. return result
  248. data = json.loads(zf.read("Metadata/project_settings.config").decode())
  249. except (KeyError, ValueError, OSError):
  250. return result
  251. if not isinstance(data, dict):
  252. return result
  253. result["printer"] = _first_settings_id(data.get("printer_settings_id"))
  254. result["process"] = _first_settings_id(data.get("print_settings_id"))
  255. return result
  256. def extract_nozzle_mapping_from_3mf(zf: zipfile.ZipFile) -> dict[int, int] | None:
  257. """Extract per-slot nozzle/extruder mapping from a 3MF file.
  258. On dual-nozzle printers (H2D, H2D Pro), each filament slot is assigned to a
  259. specific nozzle. The slicer may override user preferences when using "Auto For
  260. Flush" mode, so the actual assignment comes from slice_info.config group_id
  261. attributes, not from the user's filament_nozzle_map preference.
  262. Priority:
  263. 1. group_id on <filament> elements in slice_info.config (actual assignment)
  264. 2. filament_nozzle_map in project_settings.config (user preference fallback)
  265. Both are mapped through physical_extruder_map to get MQTT extruder IDs (0=right, 1=left).
  266. Args:
  267. zf: An open ZipFile of the 3MF archive
  268. Returns:
  269. Dictionary mapping {slot_id: extruder_id} for dual-nozzle files,
  270. or None if single-nozzle, missing data, or parse error.
  271. """
  272. try:
  273. if "Metadata/project_settings.config" not in zf.namelist():
  274. return None
  275. content = zf.read("Metadata/project_settings.config").decode()
  276. data = json.loads(content)
  277. physical_extruder_map = data.get("physical_extruder_map")
  278. if not physical_extruder_map or len(physical_extruder_map) <= 1:
  279. return None # Single-nozzle printer
  280. # Check if only one extruder is active.
  281. # If so, we can skip the mapping and just assign all slots to that extruder.
  282. # extruder_nozzle_stats format: ["Standard#0|High Flow#0", "Standard#1"]
  283. # Each entry = one extruder. Format: <NozzleVolumeType>#<count>[|...]
  284. # #N is the count of physical nozzles of that type (0 = none installed).
  285. # Types: Standard, High Flow, Hybrid, TPU High Flow
  286. active_extruders = []
  287. for stats_str in data.get("extruder_nozzle_stats") or []:
  288. nozzle_counts = [n.partition("#")[2] for n in stats_str.split("|")]
  289. active_extruders.append(1 if any(c not in ("0", "") for c in nozzle_counts) else 0)
  290. if sum(active_extruders) == 1:
  291. nozzle_mapping: dict[int, int] = {}
  292. active_idx = active_extruders.index(1)
  293. target_extruder = int(physical_extruder_map[active_idx])
  294. if "Metadata/slice_info.config" in zf.namelist():
  295. si_content = zf.read("Metadata/slice_info.config").decode()
  296. si_root = ET.fromstring(si_content)
  297. for filament_elem in si_root.findall(".//filament"):
  298. try:
  299. nozzle_mapping[int(filament_elem.get("id"))] = target_extruder
  300. except (ValueError, TypeError):
  301. pass
  302. return nozzle_mapping or None
  303. # Priority 1: Use group_id from slice_info filament elements.
  304. # This reflects the actual slicer assignment (respects "Auto For Flush").
  305. nozzle_mapping: dict[int, int] = {}
  306. if "Metadata/slice_info.config" in zf.namelist():
  307. si_content = zf.read("Metadata/slice_info.config").decode()
  308. si_root = ET.fromstring(si_content)
  309. for filament_elem in si_root.findall(".//filament"):
  310. group_id_str = filament_elem.get("group_id")
  311. filament_id_str = filament_elem.get("id")
  312. if group_id_str is not None and filament_id_str:
  313. try:
  314. group_id = int(group_id_str)
  315. slot_id = int(filament_id_str)
  316. if group_id < len(physical_extruder_map):
  317. nozzle_mapping[slot_id] = int(physical_extruder_map[group_id])
  318. except (ValueError, TypeError, IndexError):
  319. pass
  320. if nozzle_mapping:
  321. return nozzle_mapping
  322. # Priority 2: Fall back to filament_nozzle_map (user preference).
  323. # This is correct when the user manually assigned nozzles, but may be
  324. # wrong when the slicer overrides via "Auto For Flush".
  325. filament_nozzle_map = data.get("filament_nozzle_map")
  326. if not filament_nozzle_map:
  327. return None
  328. for i, slicer_ext_str in enumerate(filament_nozzle_map):
  329. slot_id = i + 1
  330. try:
  331. slicer_ext = int(slicer_ext_str)
  332. if slicer_ext < len(physical_extruder_map):
  333. nozzle_mapping[slot_id] = int(physical_extruder_map[slicer_ext])
  334. except (ValueError, TypeError, IndexError):
  335. pass
  336. return nozzle_mapping if nozzle_mapping else None
  337. except Exception:
  338. return None
  339. def extract_filament_usage_from_3mf(file_path: Path, plate_id: int | None = None) -> list[dict]:
  340. """Extract per-filament total usage from 3MF slice_info.config.
  341. This extracts the slicer-estimated total usage per filament slot,
  342. not the per-layer breakdown.
  343. Args:
  344. file_path: Path to the 3MF file
  345. plate_id: Optional plate index to filter for (for multi-plate files)
  346. Returns:
  347. List of filament usage dictionaries:
  348. [{"slot_id": 1, "used_g": 50.5, "type": "PLA", "color": "#FF0000"}, ...]
  349. """
  350. filament_usage = []
  351. try:
  352. with zipfile.ZipFile(file_path, "r") as zf:
  353. if "Metadata/slice_info.config" not in zf.namelist():
  354. return []
  355. content = zf.read("Metadata/slice_info.config").decode()
  356. root = ET.fromstring(content)
  357. if plate_id is not None:
  358. # Find the plate element with matching index
  359. for plate_elem in root.findall(".//plate"):
  360. plate_index = None
  361. for meta in plate_elem.findall("metadata"):
  362. if meta.get("key") == "index":
  363. try:
  364. plate_index = int(meta.get("value", "0"))
  365. except ValueError:
  366. pass
  367. break
  368. if plate_index == plate_id:
  369. for f in plate_elem.findall("filament"):
  370. filament_id = f.get("id")
  371. used_g = f.get("used_g", "0")
  372. try:
  373. used_amount = float(used_g)
  374. if filament_id:
  375. filament_usage.append(
  376. {
  377. "slot_id": int(filament_id),
  378. "used_g": used_amount,
  379. "type": f.get("type", ""),
  380. "color": f.get("color", ""),
  381. }
  382. )
  383. except (ValueError, TypeError):
  384. pass
  385. break
  386. else:
  387. # No plate_id specified - extract all filaments
  388. for f in root.findall(".//filament"):
  389. filament_id = f.get("id")
  390. used_g = f.get("used_g", "0")
  391. try:
  392. used_amount = float(used_g)
  393. if filament_id:
  394. filament_usage.append(
  395. {
  396. "slot_id": int(filament_id),
  397. "used_g": used_amount,
  398. "type": f.get("type", ""),
  399. "color": f.get("color", ""),
  400. }
  401. )
  402. except (ValueError, TypeError):
  403. pass # Skip filament entries with unparseable usage values
  404. except Exception:
  405. pass # Return whatever usage data was collected before the error
  406. return filament_usage
  407. def extract_bed_type_from_3mf(file_path: Path, plate_id: int | None = None) -> str | None:
  408. """Extract the build plate type (`curr_bed_type`) for a specific plate (#1281).
  409. ``archive.bed_type`` is captured at ingest time but is one value per archive
  410. (the first plate's `curr_bed_type` — see services/archive.py:235). For a
  411. multi-plate 3MF where different plates target different beds (e.g. a 40-plate
  412. file mixing PEI + Engineering), the archive-level value lies. When a queue
  413. item or print modal targets a specific plate, this re-reads the 3MF and
  414. returns that plate's actual bed type.
  415. Args:
  416. file_path: Path to the 3MF file
  417. plate_id: Plate index to filter for; if None, returns the first plate's
  418. ``curr_bed_type`` (matches the archive-level capture).
  419. Returns:
  420. Bed type string (e.g. "Textured PEI Plate"), or None if not found.
  421. """
  422. try:
  423. with zipfile.ZipFile(file_path, "r") as zf:
  424. if "Metadata/slice_info.config" not in zf.namelist():
  425. return None
  426. content = zf.read("Metadata/slice_info.config").decode()
  427. root = ET.fromstring(content)
  428. for plate_elem in root.findall(".//plate"):
  429. plate_index = None
  430. bed_value: str | None = None
  431. for meta in plate_elem.findall("metadata"):
  432. key = meta.get("key")
  433. if key == "index":
  434. try:
  435. plate_index = int(meta.get("value", "0"))
  436. except ValueError:
  437. pass # Skip plate with unparseable index
  438. elif key == "curr_bed_type" and meta.get("value"):
  439. bed_value = (meta.get("value") or "").strip()
  440. if plate_id is None:
  441. # First plate wins when no plate_id is requested.
  442. return bed_value
  443. if plate_index == plate_id:
  444. return bed_value
  445. except Exception:
  446. pass # Return None on any failure rather than raising — caller decides
  447. return None
  448. # Header values exposed as `{placeholder}` substitutions inside snippets.
  449. # Aliases let users write Prusa-style names (`{max_layer_z}`) that map onto
  450. # Bambu/Orca header keys (`max_z_height`).
  451. _HEADER_PLACEHOLDER_ALIASES = {
  452. "max_layer_z": "max_z_height",
  453. "max_print_height": "max_z_height",
  454. "total_layers": "total_layer_number",
  455. }
  456. _HEADER_KEY_RE = re.compile(r"^;\s*([^:]+?)\s*:\s*(.+?)\s*$")
  457. _PLACEHOLDER_RE = re.compile(r"\{([a-zA-Z_][a-zA-Z0-9_]*)\}")
  458. _START_GCODE_END_MARKER = "; MACHINE_START_GCODE_END"
  459. _EXECUTABLE_BLOCK_END_MARKER = "; EXECUTABLE_BLOCK_END"
  460. def _parse_3mf_gcode_header(content: str) -> dict[str, str]:
  461. """Parse the `; HEADER_BLOCK_START..END` block into a normalised dict.
  462. Keys are lowercased, ` [units]` suffixes stripped, and spaces converted
  463. to underscores so callers can look up `total_layer_number` regardless of
  464. whether the source line is `; total layer number: 80` or
  465. `; total filament length [mm] : 12155.34`.
  466. """
  467. header: dict[str, str] = {}
  468. in_header = False
  469. for raw_line in content.splitlines():
  470. line = raw_line.strip()
  471. if line == "; HEADER_BLOCK_START":
  472. in_header = True
  473. continue
  474. if line == "; HEADER_BLOCK_END":
  475. break
  476. if not in_header:
  477. continue
  478. m = _HEADER_KEY_RE.match(line)
  479. if not m:
  480. continue
  481. key, value = m.group(1), m.group(2)
  482. key = re.sub(r"\s*\[[^\]]*\]\s*$", "", key)
  483. key = key.strip().lower().replace(" ", "_")
  484. header[key] = value
  485. return header
  486. def _substitute_placeholders(snippet: str, header: dict[str, str]) -> str:
  487. """Replace `{var}` placeholders with header values, leaving unknowns intact."""
  488. def repl(m: re.Match) -> str:
  489. name = m.group(1)
  490. value = header.get(name)
  491. if value is None:
  492. alias = _HEADER_PLACEHOLDER_ALIASES.get(name)
  493. if alias is not None:
  494. value = header.get(alias)
  495. if value is None:
  496. logger.warning(
  497. "G-code injection: placeholder {%s} not found in 3MF header; leaving as-is",
  498. name,
  499. )
  500. return m.group(0)
  501. return value
  502. return _PLACEHOLDER_RE.sub(repl, snippet)
  503. def _inject_start_at_marker(content: str, snippet: str) -> str:
  504. """Insert snippet immediately before `; MACHINE_START_GCODE_END`.
  505. The marker sits at the bottom of the printer's startup block — bed heat,
  506. homing, and nozzle prime are already done, so injected snippets land in
  507. the same place a slicer-side custom-start-gcode would. Falls back to
  508. prepending if the marker isn't present (older files / non-Bambu slicers).
  509. """
  510. marker_idx = content.find(_START_GCODE_END_MARKER)
  511. if marker_idx == -1:
  512. logger.warning(
  513. "G-code injection: '%s' not found, prepending start snippet to whole file",
  514. _START_GCODE_END_MARKER,
  515. )
  516. return snippet.rstrip("\n") + "\n" + content
  517. line_start = content.rfind("\n", 0, marker_idx)
  518. line_start = 0 if line_start == -1 else line_start + 1
  519. return content[:line_start] + snippet.rstrip("\n") + "\n" + content[line_start:]
  520. def _inject_end_before_marker(content: str, snippet: str) -> str:
  521. """Insert snippet immediately before `; EXECUTABLE_BLOCK_END`.
  522. The end snippet must run *inside* the executable block. Bambu firmware
  523. (verified on a P1S) does not execute G-code that sits after
  524. `; EXECUTABLE_BLOCK_END`, so appending to the file end silently drops the
  525. snippet — auto-eject / plate-clear moves never fire. Inserting before the
  526. marker places the snippet after the printer's own machine-end sequence but
  527. still within the executed block. Falls back to appending at the file end if
  528. the marker isn't present.
  529. """
  530. marker_idx = content.find(_EXECUTABLE_BLOCK_END_MARKER)
  531. if marker_idx == -1:
  532. logger.warning(
  533. "G-code injection: '%s' not found, appending end snippet to file end",
  534. _EXECUTABLE_BLOCK_END_MARKER,
  535. )
  536. return content.rstrip("\n") + "\n" + snippet.rstrip("\n") + "\n"
  537. line_start = content.rfind("\n", 0, marker_idx)
  538. line_start = 0 if line_start == -1 else line_start + 1
  539. return content[:line_start] + snippet.rstrip("\n") + "\n" + content[line_start:]
  540. def inject_gcode_into_3mf(
  541. source_path: Path,
  542. plate_id: int,
  543. start_gcode: str | None,
  544. end_gcode: str | None,
  545. ):
  546. """Create a temp copy of a 3MF with G-code injected at start/end.
  547. Snippets support `{placeholder}` substitution against values parsed from
  548. the 3MF G-code header block (e.g. `{max_layer_z}` → `16.00`). Start
  549. snippets are anchored to the `; MACHINE_START_GCODE_END` marker so they
  550. run after the printer's own startup (#422). End snippets are inserted just
  551. before `; EXECUTABLE_BLOCK_END` so they run inside the executable block —
  552. Bambu firmware (P1S) ignores g-code placed after that marker.
  553. The plate's `.gcode.md5` sidecar is recomputed so firmware that validates
  554. it against the gcode (e.g. P1S) still accepts the modified file.
  555. Args:
  556. source_path: Path to the original 3MF file.
  557. plate_id: Plate number (1-indexed) to inject into.
  558. start_gcode: G-code to insert after printer startup, or None.
  559. end_gcode: G-code to append, or None.
  560. Returns:
  561. Path to temp file with injected G-code, or None if injection failed.
  562. Caller is responsible for cleaning up the temp file.
  563. """
  564. import tempfile
  565. if not start_gcode and not end_gcode:
  566. return None
  567. try:
  568. # Find the target gcode file inside the 3MF
  569. with zipfile.ZipFile(source_path, "r") as zf:
  570. all_gcode = [f for f in zf.namelist() if f.endswith(".gcode")]
  571. if not all_gcode:
  572. return None
  573. # Try plate-specific gcode file first
  574. target_gcode = None
  575. plate_pattern = f"plate_{plate_id}.gcode"
  576. for f in all_gcode:
  577. if f.endswith(plate_pattern):
  578. target_gcode = f
  579. break
  580. # Fall back to first gcode file
  581. if target_gcode is None:
  582. target_gcode = all_gcode[0]
  583. # Read and modify gcode content
  584. gcode_content = zf.read(target_gcode).decode("utf-8", errors="ignore")
  585. header = _parse_3mf_gcode_header(gcode_content)
  586. if start_gcode:
  587. resolved = _substitute_placeholders(start_gcode, header)
  588. # Log the post-substitution snippet so the actually-injected G-code
  589. # (placeholders like {max_layer_z} already resolved) is visible at DEBUG.
  590. logger.debug("G-code injection [%s]: resolved START snippet:\n%s", target_gcode, resolved)
  591. gcode_content = _inject_start_at_marker(gcode_content, resolved)
  592. if end_gcode:
  593. resolved = _substitute_placeholders(end_gcode, header)
  594. logger.debug("G-code injection [%s]: resolved END snippet:\n%s", target_gcode, resolved)
  595. gcode_content = _inject_end_before_marker(gcode_content, resolved)
  596. # The printer validates the plate gcode against an embedded
  597. # `<plate>.gcode.md5` sidecar (uppercase hex, no trailing newline).
  598. # Rewriting the gcode without refreshing this hash makes firmware
  599. # reject the file at load (P1S: HMS 0500-4003 "unable to parse"),
  600. # so recompute it from the exact bytes we're about to write.
  601. gcode_bytes = gcode_content.encode("utf-8")
  602. md5_name = target_gcode + ".md5"
  603. # Not a security hash — this reproduces Bambu's `.gcode.md5` sidecar
  604. # format, so flag it as non-security for the linters (ruff S324 / bandit B324).
  605. md5_value = hashlib.md5(gcode_bytes, usedforsecurity=False).hexdigest().upper().encode("ascii")
  606. # Write modified 3MF to temp file
  607. with tempfile.NamedTemporaryFile(delete=False, suffix=".3mf") as tmp:
  608. tmp_path = Path(tmp.name)
  609. with zipfile.ZipFile(tmp_path, "w", zipfile.ZIP_DEFLATED) as zf_write:
  610. for item in zf.namelist():
  611. info = zf.getinfo(item)
  612. if item == target_gcode:
  613. zf_write.writestr(info, gcode_bytes)
  614. elif item == md5_name:
  615. zf_write.writestr(info, md5_value)
  616. else:
  617. zf_write.writestr(info, zf.read(item))
  618. return tmp_path
  619. except Exception:
  620. # Clean up temp file on error
  621. if "tmp_path" in locals() and tmp_path.exists():
  622. tmp_path.unlink(missing_ok=True)
  623. return None
  624. def extract_project_filaments_from_3mf(zf: zipfile.ZipFile) -> list[dict]:
  625. """Project-wide AMS slot config from ``Metadata/project_settings.config``.
  626. Returns one dict per configured AMS slot in slot order (1-indexed), with
  627. ``type`` and ``color`` populated from the project's ``filament_type`` and
  628. ``filament_colour`` arrays. ``used_grams`` / ``used_meters`` are 0 because
  629. project_settings carries the configuration, not per-print usage — the
  630. fields exist for shape compatibility with the slice_info-derived list.
  631. The SliceModal needs this on **unsliced** project files: slice_info.config
  632. is empty until Bambu Studio has actually sliced the project, but the user
  633. can still pick filament profiles for a slice we're about to perform.
  634. """
  635. if "Metadata/project_settings.config" not in zf.namelist():
  636. return []
  637. try:
  638. proj = json.loads(zf.read("Metadata/project_settings.config").decode())
  639. except (ValueError, OSError):
  640. return []
  641. if not isinstance(proj, dict):
  642. return []
  643. types_arr = proj.get("filament_type") or []
  644. colors_arr = proj.get("filament_colour") or []
  645. slot_count = max(
  646. len(types_arr) if isinstance(types_arr, list) else 0, len(colors_arr) if isinstance(colors_arr, list) else 0
  647. )
  648. out: list[dict] = []
  649. for i in range(slot_count):
  650. out.append(
  651. {
  652. "slot_id": i + 1,
  653. "type": types_arr[i] if i < len(types_arr) and isinstance(types_arr[i], str) else "",
  654. "color": colors_arr[i] if i < len(colors_arr) and isinstance(colors_arr[i], str) else "",
  655. "used_grams": 0,
  656. "used_meters": 0,
  657. }
  658. )
  659. return out
  660. _PAINT_COLOR_ATTR_RE = re.compile(rb'paint_color="([0-9A-Fa-f]+)"')
  661. # Painted-face quadtree leaves include both real filament assignments and
  662. # tiny edit artifacts (single-leaf accidents from "tried a colour, undid,
  663. # repainted with a different one"). The threshold's only job is dropping
  664. # accidents — anything the user spent meaningful effort on must survive.
  665. # 5% of an object's painted triangles is well below any 60/40 / 70/30 /
  666. # 33/33/33 split a real two- or three-colour print would hit, so all
  667. # intentional colours are kept; one-off single-leaf paints (typically
  668. # 0.1-1.5% in observed projects) are filtered. Note that this fallback
  669. # path runs ONLY when the preview-slice path can't reach the sidecar; in
  670. # the normal flow the slicer's own pruning produces the canonical list and
  671. # this threshold isn't reached.
  672. _PAINT_NOISE_THRESHOLD = 0.05
  673. def extract_plate_extruder_set_from_3mf(zf: zipfile.ZipFile, plate_id: int) -> set[int]:
  674. """Extruder/AMS slot indices (1-indexed) used by objects on ``plate_id``.
  675. Three sources are unioned because Bambu Studio splits per-object extruder
  676. info across THREE places depending on how the user assigned colours:
  677. 1. ``model_settings.config`` — top-level ``<metadata key="extruder">``
  678. on each ``<object>`` (the "default extruder" for the whole object).
  679. 2. ``model_settings.config`` — per-``<part>`` ``<metadata key="extruder">``
  680. overrides (used when the user split an object into multiple parts
  681. with distinct filaments).
  682. 3. ``3D/Objects/object_*.model`` — ``paint_color`` attributes on
  683. individual ``<triangle>`` elements (used when the user "painted" a
  684. face with a different filament). The encoding is a hex string where
  685. each nibble is a TriangleSelector tree node: ``0`` = unpainted leaf,
  686. ``F`` = branch (4 children follow), ``1``..``E`` = leaf painted with
  687. extruder N. We don't decode the tree — every leaf-paint nibble in
  688. the string IS the extruder number, so a flat scan over hex chars
  689. yields the correct set without recursive parsing.
  690. Without (3) the painted-face data is invisible: model_settings says
  691. every object on a multi-color plate uses extruder 1 by default but the
  692. actual print uses 3, 4, 12 etc. via face paint, so the SliceModal would
  693. render only one filament dropdown for what's clearly a multi-colour
  694. print (#1150 follow-up).
  695. """
  696. if "Metadata/model_settings.config" not in zf.namelist():
  697. return set()
  698. try:
  699. root = ET.fromstring(zf.read("Metadata/model_settings.config").decode())
  700. except (ET.ParseError, OSError):
  701. return set()
  702. # Pass 1: object → set of extruders from XML metadata (sources 1 + 2)
  703. # plus the per-object .model file path so we can later scan source 3.
  704. object_extruders: dict[str, set[int]] = {}
  705. object_model_paths: dict[str, list[str]] = {}
  706. for obj_elem in root.findall(".//object"):
  707. obj_id = obj_elem.get("id")
  708. if not obj_id:
  709. continue
  710. extruders: set[int] = set()
  711. top = obj_elem.find("metadata[@key='extruder']")
  712. if top is not None:
  713. try:
  714. v = int(top.get("value", "0"))
  715. if v > 0:
  716. extruders.add(v)
  717. except (ValueError, TypeError):
  718. pass
  719. for part_elem in obj_elem.findall(".//part"):
  720. part_ext = part_elem.find("metadata[@key='extruder']")
  721. if part_ext is None:
  722. continue
  723. try:
  724. v = int(part_ext.get("value", "0"))
  725. if v > 0:
  726. extruders.add(v)
  727. except (ValueError, TypeError):
  728. pass
  729. object_extruders[obj_id] = extruders
  730. # Pass 2: 3dmodel.model maps each <object id="N"> to its component
  731. # .model file path(s). Bambu wraps object IDs that match
  732. # model_settings.config IDs around <components><component
  733. # path="/3D/Objects/object_K.model" objectid="..." /></components>.
  734. # Strip xmlns prefixes on attributes so ElementTree can find them
  735. # without namespace gymnastics — `p:path` becomes `path` etc.
  736. if "3D/3dmodel.model" in zf.namelist():
  737. try:
  738. raw = zf.read("3D/3dmodel.model").decode()
  739. stripped = re.sub(r'xmlns:?\w*="[^"]*"', "", raw)
  740. stripped = re.sub(r"<(/?)\w+:", r"<\1", stripped)
  741. stripped = re.sub(r" \w+:(\w+=)", r" \1", stripped)
  742. model_root = ET.fromstring(stripped)
  743. for obj_elem in model_root.findall(".//object"):
  744. oid = obj_elem.get("id")
  745. if not oid:
  746. continue
  747. comps = obj_elem.find("components")
  748. if comps is None:
  749. continue
  750. paths = []
  751. for c in comps.findall("component"):
  752. p = c.get("path")
  753. if p:
  754. paths.append(p.lstrip("/"))
  755. if paths:
  756. object_model_paths[oid] = paths
  757. except (ET.ParseError, OSError):
  758. pass # No 3dmodel — paint scan just won't apply
  759. # Pass 3: scan paint_color attrs in each per-object .model file. Cache
  760. # by file path because two objects often share the same component tree.
  761. paint_cache: dict[str, set[int]] = {}
  762. def _scan_paint(path: str) -> set[int]:
  763. if path in paint_cache:
  764. return paint_cache[path]
  765. out: set[int] = set()
  766. if path not in zf.namelist():
  767. paint_cache[path] = out
  768. return out
  769. try:
  770. data = zf.read(path)
  771. except OSError:
  772. paint_cache[path] = out
  773. return out
  774. # Per-extruder triangle coverage. Each painted triangle may have
  775. # multiple leaf nibbles (the quadtree subdivides the face into
  776. # painted regions); we count one triangle per unique extruder per
  777. # match so the resulting fraction is "what share of painted
  778. # triangles include at least one leaf with extruder N". Noise from
  779. # one-off edit artifacts is filtered out at the threshold below.
  780. extruder_triangles: dict[int, int] = {}
  781. total_painted = 0
  782. for match in _PAINT_COLOR_ATTR_RE.finditer(data):
  783. total_painted += 1
  784. seen: set[int] = set()
  785. for ch in match.group(1):
  786. # Hex digit → 4-bit value. 0 = unpainted leaf, F = branch
  787. # (decoded recursively but children are encoded inline, so
  788. # we'll see them on later iterations). 1-E = leaf painted
  789. # with extruder N.
  790. if ch in b"123456789":
  791. seen.add(ch - 0x30)
  792. elif ch in b"ABCDEabcde":
  793. seen.add((ch & 0x4F) - 0x37)
  794. for e in seen:
  795. extruder_triangles[e] = extruder_triangles.get(e, 0) + 1
  796. if total_painted > 0:
  797. cutoff = max(1, int(total_painted * _PAINT_NOISE_THRESHOLD))
  798. for ext, count in extruder_triangles.items():
  799. if count >= cutoff:
  800. out.add(ext)
  801. paint_cache[path] = out
  802. return out
  803. # Walk plates — collect extruders for objects on the requested plate.
  804. used: set[int] = set()
  805. for plate_elem in root.findall(".//plate"):
  806. plater_id = None
  807. for meta in plate_elem.findall("metadata"):
  808. if meta.get("key") == "plater_id":
  809. try:
  810. plater_id = int(meta.get("value", ""))
  811. except (ValueError, TypeError):
  812. pass
  813. break
  814. if plater_id != plate_id:
  815. continue
  816. for inst in plate_elem.findall("model_instance"):
  817. for inst_meta in inst.findall("metadata"):
  818. if inst_meta.get("key") != "object_id":
  819. continue
  820. obj_id = inst_meta.get("value")
  821. if not obj_id:
  822. continue
  823. used.update(object_extruders.get(obj_id, set()))
  824. for path in object_model_paths.get(obj_id, []):
  825. used.update(_scan_paint(path))
  826. break
  827. return used