mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-21 10:15:04 +02:00
fix(camera): take the finish photo when the print ends, not when its last layer starts (#2547)
The photo fired the moment layer_num reached total_layer_num. That edge is where the printer *starts* its final layer, not where it finishes it: the reporter's H2C capture shows it arriving at 92% with mc_remaining_time=2, three minutes and seventeen seconds and one filament change before the print actually ended, so the frame caught the toolhead mid-print over the model. The trigger also latched _finish_photo_captured, which locked out both the stage-22 and FINISH triggers for the rest of the print — so on firmware that never reports an end-of-print filament unload (H2C and A1 Mini confirmed) nothing could replace the bad frame. Remove the last-layer trigger. The photo is now taken at the FINISH-state trigger, which every model sends and which lands after the toolhead parks. Since Bambu's end G-code drops the plate ~100mm just before that, restore the framing before capturing: absolute G90/G1 Z to max_z_height + 10mm clearance, settle, capture, then drop it back so the print is as reachable as the printer left it. Absolute is the safety argument — that Z is a height the toolhead occupied seconds earlier, so it is inside the travel limits by construction and leaves the nozzle above the part, and it is unambiguous across model families because Z is the nozzle-to-bed gap whether the bed moves or the toolhead does. M211 is never touched (#2579). This is what #1145, #1397 and #1565 asked for. The height is only trusted when two independent sources agree: the archive is matched by the finished print's subtask_name by equality (not LIKE, so "Cube" cannot resolve to "Cube v2"), and its layer count from the 3MF must match the layer count the printer reported over MQTT. Matching on "most recent archive for this printer" was not safe — on_print_complete pops the _active_prints binding concurrently, and a print Bambuddy failed to archive would have resolved to its predecessor. A wrong height is the one failure that could drive the nozzle into the model. The move is additionally skipped when the print height is unknown, when a queue item is pending for the printer, when the printer has left FINISH, and when the new finish_photo_restore_plate setting is off. for every FINISH-state capture — which is what shipped the mid-print photo — the bank is used only when the dispatcher recorded that it injected End G-code into this print, since a SwapMod snippet may have ejected the plate. The flag is handed over in two steps (mark_pending at dispatch, adopt at print start) so it can never outlive its print: a job started from the slicer or SD card adopts False rather than inheriting its predecessor's answer. Those prints also skip the plate move outright, bank or no bank. The bank now refreshes on mc_percent advances as well as layer changes, via a new on_print_progress callback. Layer changes stop the instant the final layer begins, which left the #1867 fallback frame stale by the whole length of that layer; progress keeps ticking there and freezes before the End G-code runs, so a swapped plate still cannot reach the bank. The last-layer throttle exemption is dropped, since it would now fire a grab on every percent tick. On the timelapse path the moment producer returns early, so the consumer does the restore itself before its live-grab fallback — the documented usual outcome on P1-series, where the video has not transferred by the time the notification goes out and the shipped photo was of an already-dropped plate. The two waits are now derived from the settle window and the video poll timeout rather than hardcoded; at the old flat 75s that fallback was guaranteed to be cut off mid-settle. extract_max_z_height_from_3mf reads only a bounded prefix of the plate G-code, since a sliced plate is routinely tens of megabytes and the header is ~40 lines. It returns None for missing, unparseable, zero and negative values so callers must treat "don't know" as such rather than defaulting.
This commit is contained in:
maziggy
@@ -6327,14 +6327,19 @@ class TestTrayNowH2SExternalSpoolOverride:
|
||||
assert mqtt_client.state.tray_now == 255
|
||||
|
||||
|
||||
class TestLastLayerFinishPhotoTrigger:
|
||||
"""Tests for #1867: layer_num→total_layer_num edge fires the finish-photo
|
||||
moment before user End G-code (e.g. SwapMod) executes.
|
||||
class TestNoLastLayerFinishPhotoTrigger:
|
||||
"""#2547: the layer_num→total_layer_num edge must NOT trigger a photo.
|
||||
|
||||
A1 Mini firmware skips stg_cur=22 entirely, so the FINISH-state fallback
|
||||
fires after end G-code has already moved the plate. The last-layer edge
|
||||
is the earliest reliable "print finished" signal available across all
|
||||
Bambu printer variants.
|
||||
That edge is the moment the printer *starts* the final layer. On the H2C
|
||||
capture that closed #2547 it arrived at 92% with `mc_remaining_time=2`,
|
||||
three minutes and one filament change before the print actually ended, so
|
||||
the photo showed the toolhead mid-print over the part. It also latched
|
||||
`_finish_photo_captured`, which locked out the two triggers that fire at a
|
||||
real end-of-print — so these tests pin both halves: the edge is silent, and
|
||||
the later triggers still work after it has passed.
|
||||
|
||||
#1867 (End G-code ejecting the plate before FINISH) is handled in
|
||||
`on_finish_photo_moment` via `print_dispatch_context`, not here.
|
||||
"""
|
||||
|
||||
@pytest.fixture
|
||||
@@ -6351,79 +6356,31 @@ class TestLastLayerFinishPhotoTrigger:
|
||||
client.state.layer_num = 99
|
||||
return client
|
||||
|
||||
def test_fires_when_layer_reaches_total(self, mqtt_client):
|
||||
def test_reaching_the_last_layer_fires_nothing(self, mqtt_client):
|
||||
events = []
|
||||
mqtt_client.on_finish_photo_moment = lambda data: events.append(data)
|
||||
|
||||
mqtt_client._process_message({"print": {"layer_num": 100}})
|
||||
|
||||
assert len(events) == 1
|
||||
assert events[0]["trigger"] == "last_layer"
|
||||
assert mqtt_client._finish_photo_captured is True
|
||||
|
||||
def test_does_not_fire_when_layer_still_below_total(self, mqtt_client):
|
||||
events = []
|
||||
mqtt_client.on_finish_photo_moment = lambda data: events.append(data)
|
||||
|
||||
mqtt_client._process_message({"print": {"layer_num": 99}})
|
||||
|
||||
assert events == []
|
||||
assert mqtt_client._finish_photo_captured is False
|
||||
|
||||
def test_edge_only_no_double_fire(self, mqtt_client):
|
||||
"""Once fired, subsequent messages at layer_num == total must not
|
||||
re-fire (the guard flips _finish_photo_captured to True)."""
|
||||
def test_stage_22_still_fires_after_the_last_layer_started(self, mqtt_client):
|
||||
"""The regression the removed trigger caused: stage 22 is the good
|
||||
moment on firmware that emits it, and it arrives *after* the last-layer
|
||||
edge. The old latch swallowed it."""
|
||||
events = []
|
||||
mqtt_client.on_finish_photo_moment = lambda data: events.append(data)
|
||||
|
||||
mqtt_client._process_message({"print": {"layer_num": 100}})
|
||||
mqtt_client._process_message({"print": {"layer_num": 100}})
|
||||
mqtt_client._process_message({"print": {"layer_num": 100}})
|
||||
|
||||
assert len(events) == 1
|
||||
|
||||
def test_does_not_fire_when_not_running(self, mqtt_client):
|
||||
"""If the print never went through RUNNING (Bambuddy restart mid-print,
|
||||
firmware replay), _was_running is False and no photo trigger fires."""
|
||||
mqtt_client._was_running = False
|
||||
events = []
|
||||
mqtt_client.on_finish_photo_moment = lambda data: events.append(data)
|
||||
|
||||
mqtt_client._process_message({"print": {"layer_num": 100}})
|
||||
|
||||
assert events == []
|
||||
|
||||
def test_does_not_fire_when_total_layers_unknown(self, mqtt_client):
|
||||
"""total=0 (before slicer metadata arrives) must never satisfy the
|
||||
`new_layer >= total` condition."""
|
||||
mqtt_client.state.total_layers = 0
|
||||
mqtt_client.state.layer_num = 0
|
||||
events = []
|
||||
mqtt_client.on_finish_photo_moment = lambda data: events.append(data)
|
||||
|
||||
mqtt_client._process_message({"print": {"layer_num": 0}})
|
||||
|
||||
assert events == []
|
||||
|
||||
def test_stage_22_skipped_after_last_layer_already_fired(self, mqtt_client):
|
||||
"""Once the last-layer trigger has set _finish_photo_captured, the
|
||||
stage-22 hook that runs later on AMS printers must be a no-op."""
|
||||
events = []
|
||||
mqtt_client.on_finish_photo_moment = lambda data: events.append(data)
|
||||
|
||||
mqtt_client._process_message({"print": {"layer_num": 100}})
|
||||
assert len(events) == 1
|
||||
|
||||
mqtt_client.state.progress = 100
|
||||
mqtt_client._process_message({"print": {"stg_cur": 22}})
|
||||
|
||||
assert len(events) == 1
|
||||
assert [e["trigger"] for e in events] == ["stage_22"]
|
||||
|
||||
def test_finish_state_fallback_skipped_after_last_layer_fired(self, mqtt_client):
|
||||
"""The gcode_state=FINISH fallback (which fires after end G-code on
|
||||
every printer) must be suppressed once the last-layer edge fired.
|
||||
This is the #1867 regression check — SwapMod plate must be captured
|
||||
by last_layer, NOT by the post-End-G-code FINISH fallback."""
|
||||
def test_finish_state_still_fires_after_the_last_layer_started(self, mqtt_client):
|
||||
"""H2C/A1 Mini never emit stage 22, so FINISH is their only moment —
|
||||
and it is now reachable, where the latch used to block it."""
|
||||
events = []
|
||||
completion_events = []
|
||||
mqtt_client.on_finish_photo_moment = lambda data: events.append(data)
|
||||
@@ -6431,13 +6388,92 @@ class TestLastLayerFinishPhotoTrigger:
|
||||
mqtt_client._previous_gcode_state = "RUNNING"
|
||||
|
||||
mqtt_client._process_message({"print": {"layer_num": 100}})
|
||||
assert events[0]["trigger"] == "last_layer"
|
||||
|
||||
mqtt_client._process_message({"print": {"gcode_state": "FINISH"}})
|
||||
|
||||
assert len(events) == 1
|
||||
assert [e["trigger"] for e in events] == ["finish_state"]
|
||||
assert len(completion_events) == 1
|
||||
|
||||
def test_no_photo_trigger_fires_repeatedly_across_the_last_layer(self, mqtt_client):
|
||||
"""A three-minute last layer publishes many frames at layer_num ==
|
||||
total. None of them may produce a moment."""
|
||||
events = []
|
||||
mqtt_client.on_finish_photo_moment = lambda data: events.append(data)
|
||||
|
||||
for percent in (92, 93, 94, 95, 97, 98, 99):
|
||||
mqtt_client._process_message({"print": {"layer_num": 100, "mc_percent": percent}})
|
||||
|
||||
assert events == []
|
||||
|
||||
|
||||
class TestPrintProgressCallback:
|
||||
"""#2547: `on_print_progress` keeps the finish-photo frame bank fresh.
|
||||
|
||||
Layer changes stop firing the instant the final layer begins, so the bank
|
||||
would otherwise stay stale for the whole length of that layer. Progress is
|
||||
the field that keeps advancing there — and it freezes before the End G-code
|
||||
runs, which is what keeps a swapped plate out of the bank (#1867).
|
||||
"""
|
||||
|
||||
@pytest.fixture
|
||||
def mqtt_client(self):
|
||||
from backend.app.services.bambu_mqtt import BambuMQTTClient
|
||||
|
||||
client = BambuMQTTClient(
|
||||
ip_address="192.168.1.100",
|
||||
serial_number="TEST123",
|
||||
access_code="12345678",
|
||||
)
|
||||
client._was_running = True
|
||||
return client
|
||||
|
||||
def test_fires_on_each_advance(self, mqtt_client):
|
||||
seen = []
|
||||
mqtt_client.on_print_progress = seen.append
|
||||
|
||||
for percent in (92, 93, 94):
|
||||
mqtt_client._process_message({"print": {"mc_percent": percent}})
|
||||
|
||||
assert seen == [92, 93, 94]
|
||||
|
||||
def test_does_not_fire_when_progress_is_unchanged(self, mqtt_client):
|
||||
"""Most frames repeat the same percent; each one would otherwise cost a
|
||||
camera grab that contends with the live view."""
|
||||
seen = []
|
||||
mqtt_client.on_print_progress = seen.append
|
||||
|
||||
mqtt_client._process_message({"print": {"mc_percent": 92}})
|
||||
mqtt_client._process_message({"print": {"mc_percent": 92}})
|
||||
mqtt_client._process_message({"print": {"mc_percent": 92}})
|
||||
|
||||
assert seen == [92]
|
||||
|
||||
def test_does_not_fire_when_progress_goes_backwards(self, mqtt_client):
|
||||
"""Firmware resets progress to 0 on cancel — that is not the print
|
||||
advancing, and banking a frame there would be banking a cancelled bed."""
|
||||
seen = []
|
||||
mqtt_client.on_print_progress = seen.append
|
||||
|
||||
mqtt_client._process_message({"print": {"mc_percent": 92}})
|
||||
mqtt_client._process_message({"print": {"mc_percent": 0}})
|
||||
|
||||
assert seen == [92]
|
||||
|
||||
def test_does_not_fire_when_no_print_is_running(self, mqtt_client):
|
||||
mqtt_client._was_running = False
|
||||
seen = []
|
||||
mqtt_client.on_print_progress = seen.append
|
||||
|
||||
mqtt_client._process_message({"print": {"mc_percent": 92}})
|
||||
|
||||
assert seen == []
|
||||
|
||||
def test_absent_callback_is_not_an_error(self, mqtt_client):
|
||||
mqtt_client.on_print_progress = None
|
||||
|
||||
mqtt_client._process_message({"print": {"mc_percent": 92}})
|
||||
|
||||
assert mqtt_client.state.progress == 92
|
||||
|
||||
|
||||
class TestPresumedPowerOffRecovery:
|
||||
"""#2629: a smart-plug turn-off marks the printer offline optimistically.
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
"""Tests for the injected-End-G-code flag the finish photo depends on (#2547).
|
||||
|
||||
The flag decides whether the finish photo comes from the camera (the print is
|
||||
still on the plate) or from the in-print frame bank (a SwapMod snippet ejected
|
||||
it — #1867). Getting it wrong in either direction ships the wrong photo, so the
|
||||
two-step pending/adopt handoff exists to guarantee the flag can never outlive
|
||||
the print it was recorded for.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from backend.app.services import print_dispatch_context
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean():
|
||||
for printer_id in (1, 2):
|
||||
print_dispatch_context.clear(printer_id)
|
||||
yield
|
||||
for printer_id in (1, 2):
|
||||
print_dispatch_context.clear(printer_id)
|
||||
|
||||
|
||||
def test_unknown_printer_reports_no_injection():
|
||||
assert print_dispatch_context.end_gcode_injected(1) is False
|
||||
|
||||
|
||||
def test_pending_flag_only_counts_once_the_print_starts():
|
||||
"""Dispatch can fail between upload and start. Until the printer confirms a
|
||||
print running, the flag must not affect anything."""
|
||||
print_dispatch_context.mark_pending(1)
|
||||
|
||||
assert print_dispatch_context.end_gcode_injected(1) is False
|
||||
|
||||
assert print_dispatch_context.adopt(1) is True
|
||||
assert print_dispatch_context.end_gcode_injected(1) is True
|
||||
|
||||
|
||||
def test_adopting_consumes_the_pending_flag():
|
||||
"""A second print must not inherit the first print's snippet."""
|
||||
print_dispatch_context.mark_pending(1)
|
||||
print_dispatch_context.adopt(1)
|
||||
|
||||
assert print_dispatch_context.adopt(1) is False
|
||||
assert print_dispatch_context.end_gcode_injected(1) is False
|
||||
|
||||
|
||||
def test_a_print_we_did_not_dispatch_clears_the_previous_flag():
|
||||
"""The failure this two-step design exists to prevent: a print started from
|
||||
the slicer or SD card right after a SwapMod job would otherwise inherit its
|
||||
flag and get a mid-print banked frame instead of its own finish photo."""
|
||||
print_dispatch_context.mark_pending(1)
|
||||
print_dispatch_context.adopt(1)
|
||||
assert print_dispatch_context.end_gcode_injected(1) is True
|
||||
|
||||
# Next print start, with nothing pending — i.e. Bambuddy didn't send it.
|
||||
assert print_dispatch_context.adopt(1) is False
|
||||
assert print_dispatch_context.end_gcode_injected(1) is False
|
||||
|
||||
|
||||
def test_printers_do_not_share_flags():
|
||||
print_dispatch_context.mark_pending(1)
|
||||
print_dispatch_context.adopt(1)
|
||||
|
||||
assert print_dispatch_context.end_gcode_injected(1) is True
|
||||
assert print_dispatch_context.end_gcode_injected(2) is False
|
||||
|
||||
|
||||
def test_clear_forgets_pending_and_active():
|
||||
print_dispatch_context.mark_pending(1)
|
||||
print_dispatch_context.adopt(1)
|
||||
print_dispatch_context.mark_pending(1)
|
||||
|
||||
print_dispatch_context.clear(1)
|
||||
|
||||
assert print_dispatch_context.end_gcode_injected(1) is False
|
||||
assert print_dispatch_context.adopt(1) is False
|
||||
@@ -15,6 +15,7 @@ by the consumer. These tests pin the producer side of that contract.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from contextlib import asynccontextmanager
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
@@ -23,6 +24,7 @@ import pytest
|
||||
|
||||
from backend.app import main as main_module
|
||||
from backend.app.main import on_finish_photo_moment
|
||||
from backend.app.services import print_dispatch_context
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
@@ -54,11 +56,13 @@ def _clean_state():
|
||||
main_module._stage22_finish_frames.clear()
|
||||
main_module._inprint_frame_bank.clear()
|
||||
main_module._inprint_frame_bank_ts.clear()
|
||||
print_dispatch_context.clear(7)
|
||||
yield
|
||||
main_module._stage22_finish_in_flight.clear()
|
||||
main_module._stage22_finish_frames.clear()
|
||||
main_module._inprint_frame_bank.clear()
|
||||
main_module._inprint_frame_bank_ts.clear()
|
||||
print_dispatch_context.clear(7)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -78,6 +82,18 @@ def patched_env(fake_printer, monkeypatch):
|
||||
"backend.app.api.routes.camera.get_buffered_frame",
|
||||
lambda _pid: None,
|
||||
)
|
||||
|
||||
# #2547: default the plate restore to "print height unknown", so tests that
|
||||
# aren't about the restore never reach the G-code path. Tests that ARE about
|
||||
# it override these two.
|
||||
async def _no_height(_printer_id, _data, _logger):
|
||||
return None
|
||||
|
||||
async def _not_blocked(_printer_id):
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(main_module, "_max_z_for_current_print", _no_height)
|
||||
monkeypatch.setattr(main_module, "_plate_restore_is_blocked_by_queue", _not_blocked)
|
||||
return fake_printer
|
||||
|
||||
|
||||
@@ -212,10 +228,12 @@ async def test_consumer_wait_unblocked_when_producer_completes(patched_env, monk
|
||||
await producer
|
||||
|
||||
|
||||
async def test_finish_state_prefers_banked_frame(patched_env, monkeypatch):
|
||||
"""#1867: on the FINISH-state fallback (stage-22-less firmware, e.g. A1
|
||||
Mini) a live grab captures the post-swap plate. When a banked in-print
|
||||
frame exists it must be used instead, and the live grab must not run."""
|
||||
async def test_finish_state_prefers_banked_frame_when_end_gcode_was_injected(patched_env, monkeypatch):
|
||||
"""#1867: when Bambuddy injected End G-code, a SwapMod snippet may already
|
||||
have ejected the plate by FINISH — so the banked in-print frame is used and
|
||||
the live grab must not run."""
|
||||
print_dispatch_context.mark_pending(patched_env.id)
|
||||
print_dispatch_context.adopt(patched_env.id)
|
||||
main_module._inprint_frame_bank[patched_env.id] = b"\xff\xd8banked"
|
||||
|
||||
live_called = {"n": 0}
|
||||
@@ -232,10 +250,31 @@ async def test_finish_state_prefers_banked_frame(patched_env, monkeypatch):
|
||||
assert live_called["n"] == 0
|
||||
|
||||
|
||||
async def test_finish_state_falls_back_to_live_when_no_bank(patched_env, monkeypatch):
|
||||
"""No banked frame (feature just enabled, tiny print, capture failures) —
|
||||
the FINISH-state path still live-grabs so we degrade to the old behaviour
|
||||
rather than sending a text-only notification."""
|
||||
async def test_finish_state_grabs_live_when_no_end_gcode_was_injected(patched_env, monkeypatch):
|
||||
"""#2547: the ordinary case. Nothing moved the plate, the toolhead is
|
||||
parked, and the print is still sitting there — so the live frame is the
|
||||
finished print, and a banked mid-print frame must NOT win over it.
|
||||
|
||||
Preferring the bank here unconditionally, which is what this code used to
|
||||
do, is how the H2C shipped a photo with the toolhead over the part."""
|
||||
main_module._inprint_frame_bank[patched_env.id] = b"\xff\xd8banked-midprint"
|
||||
|
||||
async def _live(**_kwargs):
|
||||
return b"\xff\xd8live-finished-print"
|
||||
|
||||
monkeypatch.setattr("backend.app.services.camera.capture_camera_frame_bytes", _live)
|
||||
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
|
||||
|
||||
assert main_module._stage22_finish_frames[patched_env.id] == b"\xff\xd8live-finished-print"
|
||||
|
||||
|
||||
async def test_finish_state_falls_back_to_live_when_bank_is_empty(patched_env, monkeypatch):
|
||||
"""End G-code was injected but nothing was ever banked (feature just
|
||||
enabled, tiny print, capture failures). Degrade to a live grab rather than
|
||||
sending a text-only notification."""
|
||||
print_dispatch_context.mark_pending(patched_env.id)
|
||||
print_dispatch_context.adopt(patched_env.id)
|
||||
|
||||
async def _live(**_kwargs):
|
||||
return b"\xff\xd8live"
|
||||
@@ -247,10 +286,12 @@ async def test_finish_state_falls_back_to_live_when_no_bank(patched_env, monkeyp
|
||||
assert main_module._stage22_finish_frames[patched_env.id] == b"\xff\xd8live"
|
||||
|
||||
|
||||
async def test_last_layer_trigger_ignores_bank(patched_env, monkeypatch):
|
||||
"""The `last_layer` trigger fires before the swap and gives cleaner
|
||||
(parked-toolhead) framing via a live grab — the bank is only for the
|
||||
post-swap `finish_state` fallback, so it must be ignored here."""
|
||||
async def test_stage_22_trigger_ignores_bank(patched_env, monkeypatch):
|
||||
"""The `stage_22` trigger fires before any End G-code and gives cleaner
|
||||
(parked-toolhead, plate-still-up) framing via a live grab — the bank is only
|
||||
for the post-swap `finish_state` path, so it must be ignored here."""
|
||||
print_dispatch_context.mark_pending(patched_env.id)
|
||||
print_dispatch_context.adopt(patched_env.id)
|
||||
main_module._inprint_frame_bank[patched_env.id] = b"\xff\xd8banked"
|
||||
|
||||
async def _live(**_kwargs):
|
||||
@@ -258,7 +299,7 @@ async def test_last_layer_trigger_ignores_bank(patched_env, monkeypatch):
|
||||
|
||||
monkeypatch.setattr("backend.app.services.camera.capture_camera_frame_bytes", _live)
|
||||
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "last_layer"})
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "stage_22"})
|
||||
|
||||
assert main_module._stage22_finish_frames[patched_env.id] == b"\xff\xd8live"
|
||||
|
||||
@@ -299,10 +340,25 @@ async def test_bank_throttles_within_interval(monkeypatch):
|
||||
assert main_module._inprint_frame_bank[3] == b"frame-1"
|
||||
|
||||
|
||||
async def test_bank_always_refreshes_on_last_layer(monkeypatch):
|
||||
async def test_bank_throttles_on_the_last_layer_too(monkeypatch):
|
||||
"""#2547: the last layer used to bypass the throttle so it always got a
|
||||
fresh frame. Now that progress advances also drive banking, that exemption
|
||||
would fire a camera grab on every percent tick of a multi-minute last layer
|
||||
— and each grab contends with the live view for the single RTSP slot."""
|
||||
counter = _bank_env(monkeypatch, total_layers=10)
|
||||
await main_module._maybe_bank_inprint_frame(3, 5) # banks frame-1
|
||||
# Last layer bypasses the throttle for the best final framing.
|
||||
await main_module._maybe_bank_inprint_frame(3, 10) # last layer, within 25s
|
||||
assert counter["n"] == 1
|
||||
assert main_module._inprint_frame_bank[3] == b"frame-1"
|
||||
|
||||
|
||||
async def test_bank_refreshes_on_the_last_layer_once_the_throttle_elapses(monkeypatch):
|
||||
"""The point of banking on progress: a three-minute last layer keeps
|
||||
refreshing instead of freezing at the moment that layer began."""
|
||||
counter = _bank_env(monkeypatch, total_layers=10)
|
||||
await main_module._maybe_bank_inprint_frame(3, 10) # banks frame-1
|
||||
# Pretend the throttle window has passed, as it does mid-last-layer.
|
||||
main_module._inprint_frame_bank_ts[3] -= main_module._INPRINT_BANK_MIN_INTERVAL + 1
|
||||
await main_module._maybe_bank_inprint_frame(3, 10)
|
||||
assert counter["n"] == 2
|
||||
assert main_module._inprint_frame_bank[3] == b"frame-2"
|
||||
@@ -380,6 +436,9 @@ class TestStage22CacheHoldsExactlyOneRotation:
|
||||
does).
|
||||
"""
|
||||
monkeypatch.setattr(patched_env, "camera_rotation", 90, raising=False)
|
||||
# #2547: the bank is only preferred when End G-code was injected.
|
||||
print_dispatch_context.mark_pending(patched_env.id)
|
||||
print_dispatch_context.adopt(patched_env.id)
|
||||
already_rotated = self._jpeg(32, 64) # what one rotation of a 64x32 frame looks like
|
||||
main_module._inprint_frame_bank[patched_env.id] = already_rotated
|
||||
|
||||
@@ -458,3 +517,359 @@ def test_the_consumer_does_not_rotate_the_cached_frame():
|
||||
"Those bytes are already rotated by on_finish_photo_moment (#2708); rotating "
|
||||
"again returns a 180-degree print to upside-down."
|
||||
)
|
||||
|
||||
|
||||
class TestPlateRestore:
|
||||
"""#2547 / #1145 / #1397 / #1565: put the plate back into camera framing.
|
||||
|
||||
Bambu's end G-code drops the plate ~100mm as the last thing it does, so by
|
||||
FINISH the finished print sits well below where the camera frames it. The
|
||||
restore commands an ABSOLUTE Z back to just above the last printed layer.
|
||||
|
||||
Absolute is the safety argument, and these tests pin it: the target is a
|
||||
height the toolhead was physically at seconds earlier, so it is inside the
|
||||
travel limits and leaves the nozzle above the part. It is also unambiguous
|
||||
across model families — Z is the nozzle-to-bed gap whether the bed moves or
|
||||
the toolhead does — so there is no sign to get wrong the way the relative
|
||||
bed-jog path had (#1334).
|
||||
"""
|
||||
|
||||
@pytest.fixture
|
||||
def printer_client(self, monkeypatch):
|
||||
sent: list[str] = []
|
||||
client = SimpleNamespace(
|
||||
state=SimpleNamespace(state="FINISH"),
|
||||
send_gcode=lambda gcode: (sent.append(gcode), True)[1],
|
||||
)
|
||||
monkeypatch.setattr(main_module.printer_manager, "get_client", lambda _pid: client)
|
||||
monkeypatch.setattr(main_module.asyncio, "sleep", AsyncMock())
|
||||
client.sent = sent
|
||||
return client
|
||||
|
||||
async def test_commands_an_absolute_move_above_the_print(self, printer_client):
|
||||
ok = await main_module._restore_plate_for_finish_photo(7, 16.0, logging.getLogger(__name__))
|
||||
|
||||
assert ok is True
|
||||
assert printer_client.sent == ["G90\nG1 Z26.00 F600"]
|
||||
|
||||
async def test_never_touches_m211(self, printer_client):
|
||||
"""#2579: disabling soft endstops is what let a jog drive the nozzle
|
||||
into the bed. This path must not reintroduce it."""
|
||||
await main_module._restore_plate_for_finish_photo(7, 16.0, logging.getLogger(__name__))
|
||||
|
||||
assert not any("M211" in line for line in printer_client.sent)
|
||||
|
||||
async def test_skipped_when_the_printer_is_no_longer_in_finish(self, printer_client):
|
||||
"""The queue dispatches the next job the instant a print completes.
|
||||
Commanding a plate move into a starting print is not a race worth
|
||||
having, so state is re-read immediately before the move."""
|
||||
printer_client.state.state = "RUNNING"
|
||||
|
||||
ok = await main_module._restore_plate_for_finish_photo(7, 16.0, logging.getLogger(__name__))
|
||||
|
||||
assert ok is False
|
||||
assert printer_client.sent == []
|
||||
|
||||
async def test_skipped_when_the_printer_is_gone(self, monkeypatch):
|
||||
monkeypatch.setattr(main_module.printer_manager, "get_client", lambda _pid: None)
|
||||
|
||||
ok = await main_module._restore_plate_for_finish_photo(7, 16.0, logging.getLogger(__name__))
|
||||
|
||||
assert ok is False
|
||||
|
||||
async def test_reports_failure_when_the_send_fails(self, printer_client, monkeypatch):
|
||||
"""A failed send means the plate never moved — the caller must not go on
|
||||
to owe it a move back down."""
|
||||
monkeypatch.setattr(printer_client, "send_gcode", lambda _g: False)
|
||||
|
||||
ok = await main_module._restore_plate_for_finish_photo(7, 16.0, logging.getLogger(__name__))
|
||||
|
||||
assert ok is False
|
||||
|
||||
def test_park_lowers_the_plate_again(self, printer_client):
|
||||
main_module._park_plate_after_finish_photo(7, 16.0, logging.getLogger(__name__))
|
||||
|
||||
assert printer_client.sent == ["G90\nG1 Z116.00 F600"]
|
||||
|
||||
def test_park_skipped_once_the_next_print_has_started(self, printer_client):
|
||||
printer_client.state.state = "RUNNING"
|
||||
|
||||
main_module._park_plate_after_finish_photo(7, 16.0, logging.getLogger(__name__))
|
||||
|
||||
assert printer_client.sent == []
|
||||
|
||||
|
||||
class TestPlateRestoreWiring:
|
||||
"""The restore only runs in the one situation it is correct for, and the
|
||||
plate always comes back down afterwards."""
|
||||
|
||||
@pytest.fixture
|
||||
def restore_env(self, patched_env, monkeypatch):
|
||||
calls = {"restore": [], "park": [], "blocked": False, "height": 16.0}
|
||||
|
||||
async def _height(_printer_id, _data, _logger):
|
||||
return calls["height"]
|
||||
|
||||
async def _blocked(_printer_id):
|
||||
return calls["blocked"]
|
||||
|
||||
async def _restore(printer_id, max_z, _logger):
|
||||
calls["restore"].append((printer_id, max_z))
|
||||
return True
|
||||
|
||||
def _park(printer_id, max_z, _logger):
|
||||
calls["park"].append((printer_id, max_z))
|
||||
|
||||
monkeypatch.setattr(main_module, "_max_z_for_current_print", _height)
|
||||
monkeypatch.setattr(main_module, "_plate_restore_is_blocked_by_queue", _blocked)
|
||||
monkeypatch.setattr(main_module, "_restore_plate_for_finish_photo", _restore)
|
||||
monkeypatch.setattr(main_module, "_park_plate_after_finish_photo", _park)
|
||||
|
||||
async def _live(**_kwargs):
|
||||
return b"\xff\xd8live"
|
||||
|
||||
monkeypatch.setattr("backend.app.services.camera.capture_camera_frame_bytes", _live)
|
||||
return calls
|
||||
|
||||
async def test_restores_then_parks_on_the_finish_state_path(self, patched_env, restore_env):
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
|
||||
|
||||
assert restore_env["restore"] == [(patched_env.id, 16.0)]
|
||||
assert restore_env["park"] == [(patched_env.id, 16.0)]
|
||||
|
||||
async def test_not_restored_on_the_stage_22_path(self, patched_env, restore_env):
|
||||
"""Stage 22 fires before the end G-code drops the plate — it is already
|
||||
where we want it, and moving it would only cost the settle delay."""
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "stage_22"})
|
||||
|
||||
assert restore_env["restore"] == []
|
||||
assert restore_env["park"] == []
|
||||
|
||||
async def test_not_restored_when_the_banked_frame_is_used(self, patched_env, restore_env):
|
||||
"""The plate has been swapped out — no move brings the print back."""
|
||||
print_dispatch_context.mark_pending(patched_env.id)
|
||||
print_dispatch_context.adopt(patched_env.id)
|
||||
main_module._inprint_frame_bank[patched_env.id] = b"\xff\xd8banked"
|
||||
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
|
||||
|
||||
assert restore_env["restore"] == []
|
||||
assert restore_env["park"] == []
|
||||
|
||||
async def test_not_restored_when_end_gcode_was_injected_but_the_bank_is_empty(self, patched_env, restore_env):
|
||||
"""A plate-swap machine may have just ejected its plate. Even with no
|
||||
banked frame to fall back on, driving Z into whatever a swap mechanism
|
||||
is doing is not worth a photo of a bed we know may be bare."""
|
||||
print_dispatch_context.mark_pending(patched_env.id)
|
||||
print_dispatch_context.adopt(patched_env.id)
|
||||
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
|
||||
|
||||
assert restore_env["restore"] == []
|
||||
assert restore_env["park"] == []
|
||||
|
||||
async def test_not_restored_when_the_print_height_is_unknown(self, patched_env, restore_env):
|
||||
restore_env["height"] = None
|
||||
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
|
||||
|
||||
assert restore_env["restore"] == []
|
||||
assert restore_env["park"] == []
|
||||
|
||||
async def test_not_restored_when_another_job_is_queued(self, patched_env, restore_env):
|
||||
restore_env["blocked"] = True
|
||||
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
|
||||
|
||||
assert restore_env["restore"] == []
|
||||
assert restore_env["park"] == []
|
||||
|
||||
async def test_not_restored_when_the_setting_is_off(self, patched_env, restore_env, monkeypatch):
|
||||
async def _get_setting(_db, key):
|
||||
if key == "capture_finish_photo":
|
||||
return "true"
|
||||
if key == "finish_photo_restore_plate":
|
||||
return "false"
|
||||
return None
|
||||
|
||||
monkeypatch.setattr("backend.app.api.routes.settings.get_setting", _get_setting)
|
||||
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
|
||||
|
||||
assert restore_env["restore"] == []
|
||||
assert restore_env["park"] == []
|
||||
|
||||
async def test_plate_is_parked_even_when_the_capture_throws(self, patched_env, restore_env, monkeypatch):
|
||||
"""We raised it, so we owe the move back down — including when the grab
|
||||
between the two fails. Otherwise the user finds the print pinned under
|
||||
the nozzle."""
|
||||
|
||||
async def _boom(**_kwargs):
|
||||
raise RuntimeError("camera gone")
|
||||
|
||||
monkeypatch.setattr("backend.app.services.camera.capture_camera_frame_bytes", _boom)
|
||||
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
|
||||
|
||||
assert restore_env["restore"] == [(patched_env.id, 16.0)]
|
||||
assert restore_env["park"] == [(patched_env.id, 16.0)]
|
||||
|
||||
async def test_producer_event_is_still_set_after_a_restore(self, patched_env, restore_env):
|
||||
"""#1790: the consumer's bounded wait must be released on every exit,
|
||||
and the restore added a new path through the producer."""
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
|
||||
|
||||
assert main_module._stage22_finish_in_flight[patched_env.id].is_set()
|
||||
|
||||
|
||||
async def test_producer_wait_budget_covers_the_restore():
|
||||
"""The consumer's wait has to outlast settle + a worst-case RTSP grab, and
|
||||
still finish inside the notification's own photo budget — otherwise the
|
||||
restore path is cut off by a timeout somewhere above it."""
|
||||
assert main_module._FINISH_PHOTO_PRODUCER_WAIT_SECONDS > main_module._PLATE_RESTORE_SETTLE_SECONDS + 15
|
||||
|
||||
|
||||
class TestMaxZResolution:
|
||||
"""#2547 safety: the height that becomes a Z-move target must provably
|
||||
belong to the print that just finished.
|
||||
|
||||
A height from another print is the one failure mode that could drive the
|
||||
nozzle into the model — 20mm carried onto a 200mm print commands the plate
|
||||
up through the part. So the resolver refuses on every ambiguity rather than
|
||||
falling back to "whatever ran last on this printer".
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _archive(**overrides):
|
||||
base = {
|
||||
"id": 11,
|
||||
"file_path": "/data/archive/1/job/job.3mf",
|
||||
"plate_id": 1,
|
||||
"total_layers": 30,
|
||||
}
|
||||
base.update(overrides)
|
||||
return SimpleNamespace(**base)
|
||||
|
||||
@pytest.fixture
|
||||
def resolver_env(self, monkeypatch):
|
||||
env = {"archive": self._archive(), "reported_layers": 30, "height": 16.0, "where": None}
|
||||
|
||||
@asynccontextmanager
|
||||
async def _session():
|
||||
async def _execute(stmt):
|
||||
env["where"] = str(stmt)
|
||||
return SimpleNamespace(scalar_one_or_none=lambda: env["archive"])
|
||||
|
||||
yield SimpleNamespace(execute=_execute)
|
||||
|
||||
monkeypatch.setattr(main_module, "async_session", _session)
|
||||
monkeypatch.setattr(
|
||||
main_module.printer_manager,
|
||||
"get_client",
|
||||
lambda _pid: SimpleNamespace(state=SimpleNamespace(total_layers=env["reported_layers"])),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"backend.app.utils.threemf_tools.extract_max_z_height_from_3mf",
|
||||
lambda _path, _plate: env["height"],
|
||||
)
|
||||
return env
|
||||
|
||||
async def test_returns_the_height_when_name_and_layers_agree(self, resolver_env):
|
||||
height = await main_module._max_z_for_current_print(1, {"subtask_name": "job"}, logging.getLogger(__name__))
|
||||
assert height == 16.0
|
||||
|
||||
async def test_refuses_when_the_print_has_no_name_to_match_on(self, resolver_env):
|
||||
"""Without an identifier there is nothing to bind the archive to, and
|
||||
the query would degrade to 'the newest row for this printer'."""
|
||||
height = await main_module._max_z_for_current_print(1, {}, logging.getLogger(__name__))
|
||||
|
||||
assert height is None
|
||||
assert resolver_env["where"] is None # refused before touching the DB
|
||||
|
||||
async def test_refuses_when_no_archive_matches_the_name(self, resolver_env):
|
||||
resolver_env["archive"] = None
|
||||
|
||||
height = await main_module._max_z_for_current_print(1, {"subtask_name": "job"}, logging.getLogger(__name__))
|
||||
assert height is None
|
||||
|
||||
async def test_refuses_when_the_layer_counts_disagree(self, resolver_env):
|
||||
"""The corroboration check. The archive's layer count comes from the
|
||||
3MF; the printer's comes from MQTT. If two independent sources disagree,
|
||||
the row is not this print whatever its name says."""
|
||||
resolver_env["reported_layers"] = 240
|
||||
|
||||
height = await main_module._max_z_for_current_print(1, {"subtask_name": "job"}, logging.getLogger(__name__))
|
||||
assert height is None
|
||||
|
||||
async def test_proceeds_when_a_layer_count_is_simply_unknown(self, resolver_env):
|
||||
"""Absent is not the same as contradictory — a print Bambuddy has no
|
||||
layer count for still gets its height, because the name matched."""
|
||||
resolver_env["reported_layers"] = 0
|
||||
assert (
|
||||
await main_module._max_z_for_current_print(1, {"subtask_name": "job"}, logging.getLogger(__name__)) == 16.0
|
||||
)
|
||||
|
||||
resolver_env["reported_layers"] = 30
|
||||
resolver_env["archive"] = self._archive(total_layers=None)
|
||||
assert (
|
||||
await main_module._max_z_for_current_print(1, {"subtask_name": "job"}, logging.getLogger(__name__)) == 16.0
|
||||
)
|
||||
|
||||
async def test_matches_by_equality_not_substring(self, resolver_env):
|
||||
"""`LIKE %name%` would let "Cube" resolve to "Cube v2" — a different
|
||||
print, quite possibly a much taller one."""
|
||||
await main_module._max_z_for_current_print(1, {"subtask_name": "Cube"}, logging.getLogger(__name__))
|
||||
|
||||
assert "LIKE" not in resolver_env["where"].upper()
|
||||
|
||||
async def test_refuses_when_the_archive_has_no_file(self, resolver_env):
|
||||
resolver_env["archive"] = self._archive(file_path=None)
|
||||
|
||||
height = await main_module._max_z_for_current_print(1, {"subtask_name": "job"}, logging.getLogger(__name__))
|
||||
assert height is None
|
||||
|
||||
async def test_refuses_when_the_3mf_has_no_height(self, resolver_env):
|
||||
resolver_env["height"] = None
|
||||
|
||||
height = await main_module._max_z_for_current_print(1, {"subtask_name": "job"}, logging.getLogger(__name__))
|
||||
assert height is None
|
||||
|
||||
|
||||
class TestTimelapsePathPlateRestore:
|
||||
"""#2547: the timelapse path falls through to a live grab whenever the
|
||||
video hasn't landed yet — the documented usual outcome on P1-series.
|
||||
|
||||
`on_finish_photo_moment` returns early for those prints without raising the
|
||||
plate, so the photo that actually ships in the notification would be of an
|
||||
already-dropped plate. The consumer therefore does the restore itself, but
|
||||
only on that path — everywhere else the producer has already done it.
|
||||
"""
|
||||
|
||||
def test_notification_budget_outlasts_the_video_poll_plus_a_restore(self):
|
||||
"""The wait has to cover polling for the video AND the restore that
|
||||
follows when it doesn't arrive. At the old flat 75s the fallback was
|
||||
cut off mid-settle, so the plate would have moved for a photo nobody
|
||||
was still waiting for."""
|
||||
assert (
|
||||
main_module._FINISH_PHOTO_TIMELAPSE_POLL_TIMEOUT_SECONDS + main_module._FINISH_PHOTO_PRODUCER_WAIT_SECONDS
|
||||
> main_module._FINISH_PHOTO_TIMELAPSE_POLL_TIMEOUT_SECONDS + main_module._PLATE_RESTORE_SETTLE_SECONDS + 15
|
||||
)
|
||||
|
||||
async def test_producer_skips_the_restore_when_a_timelapse_was_recording(self, patched_env, monkeypatch):
|
||||
"""The producer returns before any of the restore code — the consumer
|
||||
owns it on this path, and doing it in both would move the plate twice."""
|
||||
moved = []
|
||||
|
||||
async def _restore(printer_id, max_z, _logger):
|
||||
moved.append((printer_id, max_z))
|
||||
return True
|
||||
|
||||
async def _height(_printer_id, _data, _logger):
|
||||
return 16.0
|
||||
|
||||
monkeypatch.setattr(main_module, "_restore_plate_for_finish_photo", _restore)
|
||||
monkeypatch.setattr(main_module, "_max_z_for_current_print", _height)
|
||||
|
||||
await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state", "timelapse_was_active": True})
|
||||
|
||||
assert moved == []
|
||||
|
||||
@@ -14,6 +14,7 @@ from backend.app.utils.threemf_tools import (
|
||||
extract_bed_type_from_3mf,
|
||||
extract_embedded_presets_from_3mf,
|
||||
extract_filament_usage_from_3mf,
|
||||
extract_max_z_height_from_3mf,
|
||||
extract_plate_extruder_set_from_3mf,
|
||||
extract_print_time_from_3mf,
|
||||
extract_project_filaments_from_3mf,
|
||||
@@ -1304,3 +1305,90 @@ class TestExtractPlateMetadataFrom3mf:
|
||||
assert meta.filament_usage == []
|
||||
# Missing file must not create a sticky cache entry (it may appear later).
|
||||
assert spy.call_count == 2
|
||||
|
||||
|
||||
def _make_plate_3mf(tmp_path, gcode_by_name: dict[str, str], name: str = "print.3mf"):
|
||||
"""Write a 3MF containing the given plate G-code members."""
|
||||
buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(buffer, "w") as zf:
|
||||
for member, content in gcode_by_name.items():
|
||||
zf.writestr(member, content)
|
||||
buffer.seek(0)
|
||||
path = tmp_path / name
|
||||
path.write_bytes(buffer.read())
|
||||
return path
|
||||
|
||||
|
||||
def _header(**values: str) -> str:
|
||||
lines = ["; HEADER_BLOCK_START"]
|
||||
lines += [f"; {key.replace('_', ' ')}: {value}" for key, value in values.items()]
|
||||
lines.append("; HEADER_BLOCK_END")
|
||||
lines.append("G1 X0 Y0")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
class TestExtractMaxZHeightFrom3mf:
|
||||
"""#2547: the print's top Z, used to command the plate back into camera
|
||||
framing before the finish photo.
|
||||
|
||||
This value becomes the target of a real Z move, so "don't know" has to be
|
||||
reported as None rather than defaulted — a wrong height would drive the
|
||||
nozzle into the part.
|
||||
"""
|
||||
|
||||
def test_reads_max_z_height_from_the_plate_header(self, tmp_path):
|
||||
path = _make_plate_3mf(
|
||||
tmp_path,
|
||||
{"Metadata/plate_1.gcode": _header(max_z_height="16.00", total_layer_number="80")},
|
||||
)
|
||||
assert extract_max_z_height_from_3mf(path, 1) == 16.0
|
||||
|
||||
def test_picks_the_requested_plate(self, tmp_path):
|
||||
path = _make_plate_3mf(
|
||||
tmp_path,
|
||||
{
|
||||
"Metadata/plate_1.gcode": _header(max_z_height="16.00"),
|
||||
"Metadata/plate_2.gcode": _header(max_z_height="42.50"),
|
||||
},
|
||||
)
|
||||
assert extract_max_z_height_from_3mf(path, 2) == 42.5
|
||||
|
||||
def test_falls_back_to_the_only_gcode_when_the_plate_name_does_not_match(self, tmp_path):
|
||||
"""Files from slicers that don't use Bambu's plate naming still resolve."""
|
||||
path = _make_plate_3mf(tmp_path, {"whatever.gcode": _header(max_z_height="7.25")})
|
||||
assert extract_max_z_height_from_3mf(path, 3) == 7.25
|
||||
|
||||
def test_missing_header_key_returns_none(self, tmp_path):
|
||||
path = _make_plate_3mf(tmp_path, {"Metadata/plate_1.gcode": _header(total_layer_number="80")})
|
||||
assert extract_max_z_height_from_3mf(path, 1) is None
|
||||
|
||||
def test_non_numeric_value_returns_none(self, tmp_path):
|
||||
path = _make_plate_3mf(tmp_path, {"Metadata/plate_1.gcode": _header(max_z_height="tall")})
|
||||
assert extract_max_z_height_from_3mf(path, 1) is None
|
||||
|
||||
def test_zero_and_negative_are_treated_as_unknown(self, tmp_path):
|
||||
"""Passed through, either would become a Z move toward the bed."""
|
||||
zero = _make_plate_3mf(tmp_path, {"Metadata/plate_1.gcode": _header(max_z_height="0")}, "z.3mf")
|
||||
negative = _make_plate_3mf(tmp_path, {"Metadata/plate_1.gcode": _header(max_z_height="-3")}, "n.3mf")
|
||||
assert extract_max_z_height_from_3mf(zero, 1) is None
|
||||
assert extract_max_z_height_from_3mf(negative, 1) is None
|
||||
|
||||
def test_no_gcode_member_returns_none(self, tmp_path):
|
||||
path = _make_plate_3mf(tmp_path, {"Metadata/slice_info.config": "<config/>"})
|
||||
assert extract_max_z_height_from_3mf(path, 1) is None
|
||||
|
||||
def test_unreadable_file_returns_none(self, tmp_path):
|
||||
path = tmp_path / "broken.3mf"
|
||||
path.write_text("not a zip")
|
||||
assert extract_max_z_height_from_3mf(path, 1) is None
|
||||
|
||||
def test_missing_file_returns_none(self, tmp_path):
|
||||
assert extract_max_z_height_from_3mf(tmp_path / "nope.3mf", 1) is None
|
||||
|
||||
def test_only_the_header_is_inflated(self, tmp_path):
|
||||
"""A sliced plate is routinely tens of MB; reading it whole to reach ~40
|
||||
header lines would stall the finish-photo path. The header is read from
|
||||
a bounded prefix, so a huge body must not change the answer."""
|
||||
gcode = _header(max_z_height="99.9") + "\n" + ("G1 X1 Y1 E0.1\n" * 400_000)
|
||||
path = _make_plate_3mf(tmp_path, {"Metadata/plate_1.gcode": gcode})
|
||||
assert extract_max_z_height_from_3mf(path, 1) == 99.9
|
||||
|
||||
Reference in New Issue
Block a user