mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
fix(#422): start g-code anchor + slicer placeholder substitution
Auto-Print G-code Injection had two reviewer-reported bugs from the initial
ship:
1. Start snippets were prepended to the entire plate_X.gcode, landing
before the printer's own bed-heat / homing / nozzle-prime sequence —
so a Swapmod start snippet that assumed nozzle-at-temp ran on a cold
printer (pleite). Anchor injection at "; MACHINE_START_GCODE_END" so
snippets land where a slicer-side custom-start-gcode would. Files
without the marker keep prepend behaviour as a fallback with a
warning log.
2. Placeholders like "G1 Z{max_layer_z} F600" were written verbatim;
firmware parsed them as Z1 and crashed the head into the print on
tall models — real safety bug (DevScarabyte). Added a header parser
for the 3MF "; HEADER_BLOCK_START..END" block (lowercased keys,
[units] suffix stripped, spaces -> underscores) and a Prusa-style
{name} substitution pass over both start and end snippets before
injection. Supported placeholders: {max_layer_z} / {max_print_height},
{total_layer_number} / {total_layers}, {total_filament_weight},
{total_filament_length}, plus any other normalised header key.
Unknown placeholders are left verbatim with a warning — a typo never
silently expands to an empty string.
16 new regression tests across 4 new classes in test_gcode_injection.py
(anchored injection + missing-marker fallback, placeholder substitution
including alias resolution + unknown-pass-through, direct unit tests
for each new helper). All 2195 backend unit tests pass.
Wiki print-queue page updated with the supported placeholder list and a
{max_layer_z} safety callout for park moves.
This commit is contained in:
@@ -35,6 +35,8 @@ All notable changes to Bambuddy will be documented in this file.
|
||||
- **Per-request trace ID column on every log line, plumbed through HTTP access log + application logs + response headers** — Builds on the new uvicorn-access-log-into-bambuddy.log change below: the access line tells you *who* called an endpoint, but until now there was no way to tie that line to the application records emitted on the server side while handling that request. A new FastAPI middleware (`trace_id_middleware` in `main.py`, sourced from `backend.app.core.trace`) stamps each request with a fresh 8-char hex ID (or honours a sane inbound `X-Trace-Id` header for cross-system correlation), stores it in a `ContextVar` so any code in the request's call stack can read it, echoes it on the response as `X-Trace-Id`, and a new `TraceIDFilter` injects it into every `LogRecord` so the format string `[%(trace_id)s]` resolves to the right ID for the right request. ContextVars (rather than `request.state`) are the right plumbing here because asyncio copies the current context into every `asyncio.create_task`, so background work spawned from inside a request inherits the trace ID without explicit threading; the logging filter has no access to the FastAPI request object regardless. Records emitted outside any request scope (startup, MQTT callbacks, scheduler) get a stable `-` placeholder so the column stays visually aligned and missing values are obvious in `grep`. Inbound `X-Trace-Id` is hard-validated against a strict whitelist (`[A-Za-z0-9_-]+`, max 64 chars) before being honoured — a hostile or buggy caller cannot smuggle log-injection payloads (newlines, control chars, megabyte blobs) into `bambuddy.log` via the trace-ID column; values that fail the gate silently trigger a freshly minted server-side ID rather than failing the request. Middleware is decorated AFTER `auth_middleware` on purpose: Starlette stacks `@app.middleware` decorators LIFO so the last-decorated runs first inbound, making trace stamp the OUTERMOST layer — auth log lines and every record emitted on the way down to and back from the route handler all carry the same ID. Output now looks like `2026-04-26 09:51:39,152 INFO [uvicorn.access] [a4f3b1e7] 192.168.1.42:54812 - "POST /api/v1/printers/1/print/stop HTTP/1.1" 200` paired with the route handler's `2026-04-26 09:51:39,158 INFO [bambu_mqtt] [a4f3b1e7] [SERIAL] Sent stop print command` — one `grep a4f3b1e7` away from the full causality chain. 30 new tests across `tests/unit/test_trace.py` (placeholder when no request scope, filter copies ContextVar value onto records, ID propagates into spawned tasks via asyncio context copy, concurrent requests don't leak IDs into each other, generator produces unique hex IDs, hostile payloads rejected by validator, max-length boundary, dash/underscore variants accepted) plus `tests/integration/test_trace_middleware.py` (X-Trace-Id header echoed on response, body and header IDs match, each request gets a unique ID, generator format stays short hex, safe inbound IDs honoured, hostile inbound IDs replaced, overlong inbound IDs replaced, ContextVar reset cleanly after request).
|
||||
|
||||
### Fixed
|
||||
- **Auto-Print G-code Injection: start snippet landed before printer startup, and `{placeholder}` substitution was silently broken** ([#422](https://github.com/maziggy/bambuddy/issues/422) follow-up) — Two compounding bugs surfaced by @pleite (Swapmod) and @DevScarabyte (multi-height test prints) on the initial #422 ship: **(1)** Start snippets were prepended to the entire `plate_X.gcode` content, which placed them *before* the printer's bed-heat / homing / nozzle-prime sequence — so a Swapmod start snippet that assumed nozzle-at-temp ran on a cold printer. The injection now anchors at `; MACHINE_START_GCODE_END` (the marker sitting at the bottom of every Bambu/Orca slicer's `MACHINE_START_GCODE` block, after `M109` wait-for-temp), matching where a slicer-side custom-start-gcode would land. Files without the marker (older slicer versions) keep the prepend behaviour as a fallback with a warning log. **(2)** Slicer-style placeholders like `G1 Z{max_layer_z} F600` were written verbatim to the output gcode — the printer firmware then parsed `Z{max_layer_z}` as `Z1` and crashed the head into the print on a 60mm-tall model (a real safety issue: prints damaged, top glass + AMS pushed up off the printer when the model was taller than the hard-coded park height). Added a header parser that reads the 3MF's `; HEADER_BLOCK_START..END` block (lowercased keys, `[units]` suffix stripped, spaces → underscores) and a Prusa-style `{name}` substitution pass that runs over both start and end snippets before injection. Supported placeholders: `{max_layer_z}` / `{max_print_height}` (top-layer Z), `{total_layer_number}` / `{total_layers}`, `{total_filament_weight}`, `{total_filament_length}`, plus any other normalised header key from the source file. Unknown placeholders are left in the snippet verbatim with a warning log — a typo never silently expands to an empty string and the firmware never receives a malformed `Z` parameter. 16 new regression tests in `test_gcode_injection.py` covering: start snippet anchored to the marker (printer startup runs first, snippet sits between `M109 S220` and the marker, file head untouched), missing-marker fallback path, end snippet still appended at EOF, `{max_layer_z}` resolved through the alias map, direct-key substitution from the normalised header, unknown-placeholder pass-through, and direct unit tests for each new helper (`_parse_3mf_gcode_header`, `_substitute_placeholders`, `_inject_start_at_marker`). Wiki page documents the supported placeholder list with a safety warning specifically calling out `{max_layer_z}` for park moves.
|
||||
|
||||
- **Camera page ignored `?fps=N` URL parameter** ([#1131](https://github.com/maziggy/bambuddy/issues/1131) diagnostic) — `CameraPage.tsx` hard-coded `fps=15` in the stream URL and never read the URL query string, so `/camera/1?fps=5` (and similar diagnostic suggestions for the freeze report) were silent no-ops. The sibling `StreamOverlayPage` already honoured `?fps=` correctly; the bug was that `CameraPage` was the gap. Now reads `searchParams.get('fps')` via `useSearchParams`, parses it, falls back to 15 on missing/non-numeric, clamps to the backend's 1–30 range, and threads the resulting value into the stream URL. Backend `generate_rtsp_mjpeg_stream` already accepted the parameter and re-clamps per-model (chamber-image A1/P1 capped at 5, RTSP capped at 30). 5 new regression tests in `CameraPage.test.tsx::fps URL parameter (#1131)` cover default-15, honoured value, clamp-above-30, clamp-below-1, and non-numeric fallback — same matrix `StreamOverlayPage.test.tsx` already pins. Independent of the underlying freeze investigation in #1131; surfaced while triaging that report.
|
||||
- **Reprint-from-archive failed with `0500_4003` SD R/W errors after a stuck dispatch, fixable only by restarting the container** ([#1136](https://github.com/maziggy/bambuddy/issues/1136)) — Reported by @smandon: reprinting from archives sometimes fails immediately with MicroSD R/W exception errors, with the printer's MQTT push referencing a 3MF file from a *different unrelated* archive (`WARIO_Wall_decor_-_NO_AMS.3mf` while the user was actually trying to print `Cable_Organiser_Cable_Clip.3mf`). Once it starts happening, every subsequent reprint hits the same error until the container is restarted. Root cause traced from his support package log to paho-mqtt's client-side QoS 1 queue: when the printer's command channel goes half-broken (telemetry still flowing, publishes silently dropped — same #887/#936 pattern), Bambuddy's 15s dispatch deadline expires (`background_dispatch.py:993`) and calls `force_reconnect_stale_session()`. That function was force-closing the underlying socket so paho's auto-reconnect would kick in — but the same `mqtt.Client` instance, same `client_id`, and **same in-process QoS 1 queue** stayed alive across the reconnect. Any unacked publish from the broken session — typically the just-sent `project_file` for the new archive — got replayed verbatim on the new connection. And because the in-process queue accumulates across multiple stuck dispatches within one Python process, by the second or third stuck reprint there were several stale `project_file`/`resume`/`stop`/`clean_print_error` commands queued up and replaying together. The printer received the flood, tried to load whichever stale path the firmware latched onto last, found a file that no longer existed on its SD card → `0500_4003`. Container restart was the only thing that fixed it because it was the only thing that wiped paho's in-process queue. Replaced the socket-close with a context-aware reconnect: `force_reconnect_stale_session()` and `check_staleness()` now go through a routing helper `_reset_client_for_reconnect()` that picks the right teardown strategy based on caller context. **Async-context callers** (the dispatch deadline path — `background_dispatch.py:993` — which is the actual #1136 trigger, plus FastAPI route handlers via `check_staleness`) get the **hard-reset path**: `client.disconnect()` (broker sees DISCONNECT and drops the session immediately, since `clean_session=True`), `client.loop_stop()` (kills the paho network thread, taking its QoS 1 queue with it), nulls out `self._client`, and calls `self.connect()` to construct a fresh `mqtt.Client` with an incremented `client_id`. New connection starts genuinely empty, no replay possible. **Paho-network-thread callers** (the developer-mode probe and `ams_filament_setting` zombie detection inside `_update_state`, lines ~2604 and ~2623) keep the **socket-close fallback** — calling `loop_stop()` from inside the network thread would self-join and deadlock, so the safe pattern there remains "close the socket and let paho's own loop detect it and auto-reconnect on the same client". Theoretical queue replay is still possible on those paths but #1136 specifically traced through the dispatch path, and the legacy socket-close has been battle-tested for the zombie paths since #887. Routing decision is made via `asyncio.get_running_loop()` — paho's callback thread has no loop, every legitimate hard-reset caller does. 7 regression tests across two new test classes: `TestForceReconnectRouting` (3 tests pinning the sync-context → socket-close fallback, async-context → hard-reset path with mock-stubbed `connect()`, and the state-disconnected broadcast firing once on either path) and `TestHardResetClientDirect` (3 tests pinning the helper directly: old client receives `disconnect()` + `loop_stop()`, `_client` reference cleared, failing `disconnect()` doesn't propagate so the await chain in `background_dispatch.py` doesn't break). Existing `TestZombieSessionDetection::test_two_timeouts_force_reconnect` and `TestDeveloperModeProbeTimeout::test_second_timeout_forces_reconnect` updated to assert the socket-close path (matching their paho-thread context), preserving the legacy contract. All 2179 backend unit tests pass. Thanks to @smandon for the precise reproduction logs that made this diagnosable from a single support package.
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ accurate partial usage reporting for multi-material prints.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import math
|
||||
import re
|
||||
import zipfile
|
||||
@@ -13,6 +14,8 @@ from pathlib import Path
|
||||
|
||||
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)
|
||||
@@ -442,6 +445,90 @@ def extract_filament_usage_from_3mf(file_path: Path, plate_id: int | None = None
|
||||
return filament_usage
|
||||
|
||||
|
||||
# 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"
|
||||
|
||||
|
||||
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_gcode_into_3mf(
|
||||
source_path: Path,
|
||||
plate_id: int,
|
||||
@@ -450,10 +537,16 @@ def inject_gcode_into_3mf(
|
||||
):
|
||||
"""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 appended
|
||||
after the last line of the print.
|
||||
|
||||
Args:
|
||||
source_path: Path to the original 3MF file.
|
||||
plate_id: Plate number (1-indexed) to inject into.
|
||||
start_gcode: G-code to prepend, or None.
|
||||
start_gcode: G-code to insert after printer startup, or None.
|
||||
end_gcode: G-code to append, or None.
|
||||
|
||||
Returns:
|
||||
@@ -486,11 +579,14 @@ def inject_gcode_into_3mf(
|
||||
|
||||
# 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:
|
||||
gcode_content = start_gcode + "\n" + gcode_content
|
||||
resolved = _substitute_placeholders(start_gcode, header)
|
||||
gcode_content = _inject_start_at_marker(gcode_content, resolved)
|
||||
if end_gcode:
|
||||
gcode_content = gcode_content.rstrip("\n") + "\n" + end_gcode + "\n"
|
||||
resolved = _substitute_placeholders(end_gcode, header)
|
||||
gcode_content = gcode_content.rstrip("\n") + "\n" + resolved + "\n"
|
||||
|
||||
# Write modified 3MF to temp file
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=".3mf") as tmp:
|
||||
|
||||
@@ -4,9 +4,12 @@ import tempfile
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from backend.app.utils.threemf_tools import inject_gcode_into_3mf
|
||||
from backend.app.utils.threemf_tools import (
|
||||
_inject_start_at_marker,
|
||||
_parse_3mf_gcode_header,
|
||||
_substitute_placeholders,
|
||||
inject_gcode_into_3mf,
|
||||
)
|
||||
|
||||
|
||||
def _make_temp_path(suffix=".3mf") -> Path:
|
||||
@@ -205,3 +208,218 @@ class TestInjectGcodeInto3mf:
|
||||
source.unlink(missing_ok=True)
|
||||
if result:
|
||||
result.unlink(missing_ok=True)
|
||||
|
||||
|
||||
# Realistic Bambu / Orca header + startup block — the start-gcode marker is the
|
||||
# anchor point #422 reviewers (DevScarabyte, pleite) reported as the correct
|
||||
# injection point. Snippets injected before this should land *after* the bed
|
||||
# heat / homing / nozzle prime sequence, not before it.
|
||||
_BAMBU_GCODE_TEMPLATE = """\
|
||||
; HEADER_BLOCK_START
|
||||
; BambuStudio 02.06.00.51
|
||||
; total layer number: 80
|
||||
; total filament length [mm] : 12155.34
|
||||
; total filament weight [g] : 36.55
|
||||
; max_z_height: 16.00
|
||||
; HEADER_BLOCK_END
|
||||
; MACHINE_START_GCODE_BEGIN
|
||||
M104 S220 ; preheat
|
||||
G28 ; home
|
||||
M109 S220 ; wait for nozzle
|
||||
G92 E0 ; reset extruder
|
||||
; MACHINE_START_GCODE_END
|
||||
G1 X10 Y10 Z0.2
|
||||
G1 X100 Y100 E5
|
||||
M104 S0
|
||||
"""
|
||||
|
||||
|
||||
class TestStartAnchoredInjection:
|
||||
"""Tests for #422 follow-up: start g-code injected at MACHINE_START_GCODE_END."""
|
||||
|
||||
def test_start_lands_after_printer_startup(self):
|
||||
"""Start snippet sits immediately before MACHINE_START_GCODE_END, not at file head."""
|
||||
source = _make_test_3mf(_BAMBU_GCODE_TEMPLATE)
|
||||
try:
|
||||
result = inject_gcode_into_3mf(source, 1, "; SWAPMOD-START", None)
|
||||
assert result is not None
|
||||
|
||||
with zipfile.ZipFile(result, "r") as zf:
|
||||
gcode = zf.read("Metadata/plate_1.gcode").decode("utf-8")
|
||||
|
||||
# Original file head is preserved — snippet does NOT prepend.
|
||||
assert gcode.startswith("; HEADER_BLOCK_START\n")
|
||||
# Snippet sits right above the marker.
|
||||
marker_idx = gcode.index("; MACHINE_START_GCODE_END")
|
||||
snippet_idx = gcode.index("; SWAPMOD-START")
|
||||
assert snippet_idx < marker_idx
|
||||
# Nothing else between snippet and marker except the trailing newline.
|
||||
between = gcode[snippet_idx:marker_idx]
|
||||
assert between == "; SWAPMOD-START\n"
|
||||
# Printer's own startup commands still come BEFORE the snippet.
|
||||
startup_idx = gcode.index("M109 S220")
|
||||
assert startup_idx < snippet_idx
|
||||
finally:
|
||||
source.unlink(missing_ok=True)
|
||||
if result:
|
||||
result.unlink(missing_ok=True)
|
||||
|
||||
def test_no_marker_falls_back_to_prepend(self):
|
||||
"""Files without MACHINE_START_GCODE_END (older slicers) keep prepend behaviour."""
|
||||
source = _make_test_3mf("G28\nM400\n")
|
||||
try:
|
||||
result = inject_gcode_into_3mf(source, 1, "; LEGACY-START", None)
|
||||
assert result is not None
|
||||
|
||||
with zipfile.ZipFile(result, "r") as zf:
|
||||
gcode = zf.read("Metadata/plate_1.gcode").decode("utf-8")
|
||||
|
||||
assert gcode.startswith("; LEGACY-START\n")
|
||||
assert "G28" in gcode
|
||||
finally:
|
||||
source.unlink(missing_ok=True)
|
||||
if result:
|
||||
result.unlink(missing_ok=True)
|
||||
|
||||
def test_end_still_appended_at_eof(self):
|
||||
"""End g-code keeps the existing append-to-EOF behaviour even with marker present."""
|
||||
source = _make_test_3mf(_BAMBU_GCODE_TEMPLATE)
|
||||
try:
|
||||
result = inject_gcode_into_3mf(source, 1, None, "; SWAPMOD-END")
|
||||
assert result is not None
|
||||
|
||||
with zipfile.ZipFile(result, "r") as zf:
|
||||
gcode = zf.read("Metadata/plate_1.gcode").decode("utf-8")
|
||||
|
||||
assert gcode.endswith("; SWAPMOD-END\n")
|
||||
# Marker anchor is irrelevant for end snippets.
|
||||
assert gcode.index("; SWAPMOD-END") > gcode.index("; MACHINE_START_GCODE_END")
|
||||
finally:
|
||||
source.unlink(missing_ok=True)
|
||||
if result:
|
||||
result.unlink(missing_ok=True)
|
||||
|
||||
|
||||
class TestPlaceholderSubstitution:
|
||||
"""Tests for #422 follow-up: {placeholder} substitution from 3MF header values."""
|
||||
|
||||
def test_max_z_height_substituted_in_end_snippet(self):
|
||||
"""`G1 Z{max_layer_z}` resolves to the model's actual top-layer Z (DevScarabyte safety bug)."""
|
||||
source = _make_test_3mf(_BAMBU_GCODE_TEMPLATE)
|
||||
try:
|
||||
# Prusa-style alias: max_layer_z → max_z_height in the Bambu header
|
||||
result = inject_gcode_into_3mf(source, 1, None, "G1 Z{max_layer_z} F600")
|
||||
assert result is not None
|
||||
|
||||
with zipfile.ZipFile(result, "r") as zf:
|
||||
gcode = zf.read("Metadata/plate_1.gcode").decode("utf-8")
|
||||
|
||||
# max_z_height in the template is 16.00 — the dangerous Z1 fallback is gone.
|
||||
assert "G1 Z16.00 F600" in gcode
|
||||
assert "{max_layer_z}" not in gcode
|
||||
finally:
|
||||
source.unlink(missing_ok=True)
|
||||
if result:
|
||||
result.unlink(missing_ok=True)
|
||||
|
||||
def test_direct_header_key_lookup(self):
|
||||
"""Snippets can reference normalised header keys directly without going through aliases."""
|
||||
source = _make_test_3mf(_BAMBU_GCODE_TEMPLATE)
|
||||
try:
|
||||
result = inject_gcode_into_3mf(
|
||||
source, 1, None, "; layers={total_layer_number} weight={total_filament_weight}"
|
||||
)
|
||||
assert result is not None
|
||||
|
||||
with zipfile.ZipFile(result, "r") as zf:
|
||||
gcode = zf.read("Metadata/plate_1.gcode").decode("utf-8")
|
||||
|
||||
assert "; layers=80 weight=36.55" in gcode
|
||||
finally:
|
||||
source.unlink(missing_ok=True)
|
||||
if result:
|
||||
result.unlink(missing_ok=True)
|
||||
|
||||
def test_unknown_placeholder_left_intact(self):
|
||||
"""A typo or unsupported placeholder is preserved verbatim instead of becoming empty."""
|
||||
source = _make_test_3mf(_BAMBU_GCODE_TEMPLATE)
|
||||
try:
|
||||
result = inject_gcode_into_3mf(source, 1, None, "; nope={does_not_exist}")
|
||||
assert result is not None
|
||||
|
||||
with zipfile.ZipFile(result, "r") as zf:
|
||||
gcode = zf.read("Metadata/plate_1.gcode").decode("utf-8")
|
||||
|
||||
assert "; nope={does_not_exist}" in gcode
|
||||
finally:
|
||||
source.unlink(missing_ok=True)
|
||||
if result:
|
||||
result.unlink(missing_ok=True)
|
||||
|
||||
def test_no_placeholders_no_header_required(self):
|
||||
"""Snippets without placeholders inject correctly even when the header is absent."""
|
||||
source = _make_test_3mf("G28\nM400\n")
|
||||
try:
|
||||
result = inject_gcode_into_3mf(source, 1, "; PLAIN", None)
|
||||
assert result is not None
|
||||
|
||||
with zipfile.ZipFile(result, "r") as zf:
|
||||
gcode = zf.read("Metadata/plate_1.gcode").decode("utf-8")
|
||||
|
||||
assert gcode.startswith("; PLAIN\n")
|
||||
finally:
|
||||
source.unlink(missing_ok=True)
|
||||
if result:
|
||||
result.unlink(missing_ok=True)
|
||||
|
||||
|
||||
class TestHeaderParser:
|
||||
"""Direct tests for `_parse_3mf_gcode_header`."""
|
||||
|
||||
def test_parses_bambu_header_block(self):
|
||||
header = _parse_3mf_gcode_header(_BAMBU_GCODE_TEMPLATE)
|
||||
assert header["max_z_height"] == "16.00"
|
||||
assert header["total_layer_number"] == "80"
|
||||
# Units suffix is stripped from the key.
|
||||
assert header["total_filament_length"] == "12155.34"
|
||||
assert header["total_filament_weight"] == "36.55"
|
||||
|
||||
def test_ignores_lines_outside_header_block(self):
|
||||
content = "; HEADER_BLOCK_START\n; key: in\n; HEADER_BLOCK_END\n; key: out\n"
|
||||
header = _parse_3mf_gcode_header(content)
|
||||
assert header == {"key": "in"}
|
||||
|
||||
def test_returns_empty_when_no_header(self):
|
||||
assert _parse_3mf_gcode_header("G28\nG1 X0\n") == {}
|
||||
|
||||
|
||||
class TestPlaceholderHelper:
|
||||
"""Direct tests for `_substitute_placeholders`."""
|
||||
|
||||
def test_substitutes_known_keys(self):
|
||||
assert _substitute_placeholders("Z={a} F={b}", {"a": "10", "b": "600"}) == "Z=10 F=600"
|
||||
|
||||
def test_alias_resolves_to_underlying_key(self):
|
||||
assert _substitute_placeholders("Z={max_layer_z}", {"max_z_height": "16.00"}) == "Z=16.00"
|
||||
|
||||
def test_unknown_left_verbatim(self):
|
||||
assert _substitute_placeholders("{nope}", {}) == "{nope}"
|
||||
|
||||
|
||||
class TestStartMarkerHelper:
|
||||
"""Direct tests for `_inject_start_at_marker`."""
|
||||
|
||||
def test_inserts_before_marker_line(self):
|
||||
content = "first\nsecond\n; MACHINE_START_GCODE_END\ntail\n"
|
||||
result = _inject_start_at_marker(content, "INJECTED")
|
||||
assert result == "first\nsecond\nINJECTED\n; MACHINE_START_GCODE_END\ntail\n"
|
||||
|
||||
def test_marker_at_start_of_file(self):
|
||||
content = "; MACHINE_START_GCODE_END\nrest\n"
|
||||
result = _inject_start_at_marker(content, "INJECTED")
|
||||
assert result == "INJECTED\n; MACHINE_START_GCODE_END\nrest\n"
|
||||
|
||||
def test_missing_marker_falls_back_to_prepend(self):
|
||||
content = "G28\nG1 X0\n"
|
||||
result = _inject_start_at_marker(content, "INJECTED")
|
||||
assert result == "INJECTED\nG28\nG1 X0\n"
|
||||
|
||||
Reference in New Issue
Block a user