mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
fix(notifications): scope completion notification to printed plate on multi-plate 3MFs (#1785)
The 3MF parser sums prediction + weight across every plate (#1593) so the archive card can headline the whole project — correct for the card, wrong for the completion notification of a single plate. The queue UI already re-reads the 3MF per-plate at print_queue.py:272-285; mirror that for the notification path so Discord / Pushover / email show the plate's actual duration and grams instead of the project sum. Helper fails open on every error path so a missing or corrupt 3MF can't block the notification.
This commit is contained in:
File diff suppressed because one or more lines are too long
@@ -38,7 +38,11 @@ from backend.app.schemas.print_queue import (
|
||||
from backend.app.services.filament_deficit import compute_deficit_for_queue_item
|
||||
from backend.app.services.notification_service import notification_service
|
||||
from backend.app.utils.printer_models import normalize_printer_model, normalize_printer_model_id
|
||||
from backend.app.utils.threemf_tools import extract_bed_type_from_3mf, extract_filament_usage_from_3mf
|
||||
from backend.app.utils.threemf_tools import (
|
||||
extract_bed_type_from_3mf,
|
||||
extract_filament_usage_from_3mf,
|
||||
extract_print_time_from_3mf,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -106,56 +110,10 @@ def _extract_filament_types_from_3mf(file_path: Path, plate_id: int | None = Non
|
||||
return sorted(types)
|
||||
|
||||
|
||||
def _extract_print_time_from_3mf(file_path: Path, plate_id: int | None = None) -> int | None:
|
||||
"""Extract print time (prediction) from a 3MF file.
|
||||
|
||||
Args:
|
||||
file_path: Path to the 3MF file
|
||||
plate_id: Optional plate index to filter for (for multi-plate files)
|
||||
|
||||
Returns:
|
||||
Print time in seconds, or None if not found
|
||||
"""
|
||||
try:
|
||||
with zipfile.ZipFile(file_path, "r") as zf:
|
||||
if "Metadata/slice_info.config" not in zf.namelist():
|
||||
return None
|
||||
|
||||
content = zf.read("Metadata/slice_info.config").decode()
|
||||
root = ET.fromstring(content)
|
||||
|
||||
if plate_id is not None:
|
||||
for plate_elem in root.findall(".//plate"):
|
||||
plate_index = None
|
||||
for meta in plate_elem.findall("metadata"):
|
||||
if meta.get("key") == "index":
|
||||
try:
|
||||
plate_index = int(meta.get("value", "0"))
|
||||
except ValueError:
|
||||
pass # Skip plate with unparseable index
|
||||
break
|
||||
|
||||
if plate_index == plate_id:
|
||||
for meta in plate_elem.findall("metadata"):
|
||||
if meta.get("key") == "prediction":
|
||||
try:
|
||||
return int(meta.get("value", "0"))
|
||||
except ValueError:
|
||||
return None
|
||||
break
|
||||
else:
|
||||
plate_elem = root.find(".//plate")
|
||||
if plate_elem is not None:
|
||||
for meta in plate_elem.findall("metadata"):
|
||||
if meta.get("key") == "prediction":
|
||||
try:
|
||||
return int(meta.get("value", "0"))
|
||||
except ValueError:
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.warning("Failed to extract print time from %s: %s", file_path, e)
|
||||
|
||||
return None
|
||||
# Local alias kept so existing call sites stay compact; the implementation lives
|
||||
# in utils/threemf_tools.py so the notification path (main.py) can reuse it
|
||||
# without importing from a routes module (#1785).
|
||||
_extract_print_time_from_3mf = extract_print_time_from_3mf
|
||||
|
||||
|
||||
def _enrich_response(item: PrintQueueItem) -> PrintQueueItemResponse:
|
||||
|
||||
+95
-3
@@ -691,6 +691,82 @@ def _get_start_plate_id(archive_id: int | None) -> int | None:
|
||||
return _print_plate_ids.get(archive_id)
|
||||
|
||||
|
||||
def _partial_progress_scale(progress: int | float | None) -> float:
|
||||
"""Clamp ``progress / 100`` into [0.0, 1.0] for partial-print scaling.
|
||||
|
||||
Used by every site that multiplies a "would-have-used" slicer estimate
|
||||
down to "actually-used" for failed / cancelled / stopped prints. Centralised
|
||||
so the three sites in ``_background_notifications`` (and the per-plate
|
||||
override helper) can't drift apart on the coercion shape.
|
||||
"""
|
||||
return max(0.0, min((progress or 0) / 100.0, 1.0))
|
||||
|
||||
|
||||
def _scope_notification_archive_data_to_plate(
|
||||
archive_data: dict,
|
||||
archive_file_path: str | None,
|
||||
plate_id: int | None,
|
||||
print_status: str,
|
||||
progress: int | float | None,
|
||||
base_dir: Path,
|
||||
) -> dict:
|
||||
"""Override summed-across-plates totals in ``archive_data`` with the values
|
||||
for ``plate_id`` so the completion notification reports what was actually
|
||||
printed, not the whole project (#1785).
|
||||
|
||||
The 3MF parser at services/archive.py:200-264 sums ``prediction`` and
|
||||
``weight`` across every plate of a multi-plate file (#1593) — correct for
|
||||
the archive card's "whole project" headline, wrong for the completion
|
||||
notification of a single-plate print. The queue UI already re-reads the
|
||||
3MF per-plate at print_queue.py:272-285; this helper mirrors that for the
|
||||
notification payload (filament grams, time estimate, per-slot breakdown).
|
||||
|
||||
No-ops when ``plate_id`` is None, the file is missing, or the 3MF carries
|
||||
no per-plate values — in every fail case the original ``archive_data`` is
|
||||
returned unchanged so the notification still sends.
|
||||
"""
|
||||
if plate_id is None or not archive_file_path:
|
||||
return archive_data
|
||||
|
||||
from backend.app.utils.threemf_tools import (
|
||||
extract_filament_usage_from_3mf,
|
||||
extract_print_time_from_3mf,
|
||||
)
|
||||
|
||||
archive_path = base_dir / archive_file_path
|
||||
if not archive_path.exists():
|
||||
return archive_data
|
||||
|
||||
plate_slots = extract_filament_usage_from_3mf(archive_path, plate_id)
|
||||
plate_grams = sum(f.get("used_g", 0) for f in plate_slots)
|
||||
plate_time = extract_print_time_from_3mf(archive_path, plate_id)
|
||||
|
||||
scale = 1.0 if print_status == "completed" else _partial_progress_scale(progress)
|
||||
|
||||
if plate_time:
|
||||
archive_data["print_time_seconds"] = plate_time
|
||||
|
||||
# Gate both the grams headline AND the per-slot breakdown on the same
|
||||
# `plate_grams > 0` signal: if the 3MF carries per-plate filament rows but
|
||||
# they all sum to zero (slicer bug / re-slice without estimate), drop back
|
||||
# to the project-level grams the archive columns already provide rather
|
||||
# than ship a project-level headline next to an all-zero per-plate
|
||||
# breakdown.
|
||||
if plate_grams > 0:
|
||||
archive_data["actual_filament_grams"] = round(plate_grams * scale, 1)
|
||||
archive_data["filament_slots"] = [
|
||||
{
|
||||
"slot_id": s.get("slot_id"),
|
||||
"used_g": round((s.get("used_g") or 0) * scale, 1),
|
||||
"type": s.get("type", ""),
|
||||
"color": s.get("color", ""),
|
||||
}
|
||||
for s in plate_slots
|
||||
]
|
||||
|
||||
return archive_data
|
||||
|
||||
|
||||
def _extract_filament_data_from_mqtt(data: dict, ams_mapping: list[int] | None = None) -> dict[str, str]:
|
||||
"""Best-effort filament metadata from the MQTT print-start snapshot.
|
||||
|
||||
@@ -4179,8 +4255,12 @@ async def on_print_complete(printer_id: int, data: dict):
|
||||
# Always drain the plate_id register on completion — the session already
|
||||
# consumed it at print-start injection; leaving it would leak into the next
|
||||
# print on the same archive_id (rare but possible with reprints) (#1697).
|
||||
# Capture the popped value so the completion notification can scope the
|
||||
# archive-level (summed-across-plates per #1593) filament + time totals
|
||||
# down to the single plate that was actually printed (#1785).
|
||||
notify_plate_id: int | None = None
|
||||
if archive_id:
|
||||
_print_plate_ids.pop(archive_id, None)
|
||||
notify_plate_id = _print_plate_ids.pop(archive_id, None)
|
||||
|
||||
# Internal inventory: track AMS remain% deltas (skip if Spoolman handles usage)
|
||||
try:
|
||||
@@ -4733,7 +4813,7 @@ async def on_print_complete(printer_id: int, data: dict):
|
||||
# Scale filament usage for partial prints
|
||||
if print_status != "completed" and archive.filament_used_grams:
|
||||
progress = data.get("progress") or 0
|
||||
scale = max(0.0, min(progress / 100.0, 1.0))
|
||||
scale = _partial_progress_scale(progress)
|
||||
archive_data["actual_filament_grams"] = round(archive.filament_used_grams * scale, 1)
|
||||
archive_data["progress"] = progress
|
||||
|
||||
@@ -4741,10 +4821,22 @@ async def on_print_complete(printer_id: int, data: dict):
|
||||
if archive.extra_data and archive.extra_data.get("filament_slots"):
|
||||
slots = archive.extra_data["filament_slots"]
|
||||
if print_status != "completed":
|
||||
scale = max(0.0, min((data.get("progress") or 0) / 100.0, 1.0))
|
||||
scale = _partial_progress_scale(data.get("progress"))
|
||||
slots = [{**s, "used_g": round(s["used_g"] * scale, 1)} for s in slots]
|
||||
archive_data["filament_slots"] = slots
|
||||
|
||||
# Scope project-summed totals down to the plate that was
|
||||
# actually printed — see _scope_notification_archive_data_to_plate
|
||||
# for the why (#1785).
|
||||
archive_data = _scope_notification_archive_data_to_plate(
|
||||
archive_data,
|
||||
archive.file_path,
|
||||
notify_plate_id,
|
||||
print_status,
|
||||
data.get("progress"),
|
||||
app_settings.base_dir,
|
||||
)
|
||||
|
||||
# Enrich filament_grams from usage_results when archive has no 3MF data
|
||||
if not archive_data.get("actual_filament_grams") and usage_results:
|
||||
total_from_usage = sum(r.get("weight_used", 0) for r in usage_results)
|
||||
|
||||
@@ -485,6 +485,64 @@ def extract_filament_usage_from_3mf(file_path: Path, plate_id: int | None = None
|
||||
return filament_usage
|
||||
|
||||
|
||||
def extract_print_time_from_3mf(file_path: Path, plate_id: int | None = None) -> int | None:
|
||||
"""Extract the slicer's predicted print time from a 3MF's slice_info.config.
|
||||
|
||||
Multi-plate 3MFs carry one ``<plate><metadata key="prediction" .../></plate>``
|
||||
per plate. The archive-level `print_time_seconds` is the sum across all plates
|
||||
(see services/archive.py:200-264, #1593). For per-plate UI / notifications,
|
||||
callers re-read the 3MF and request the specific plate's value via this helper.
|
||||
|
||||
Args:
|
||||
file_path: Path to the 3MF file
|
||||
plate_id: Plate index to filter for; if None, returns the first plate's
|
||||
``prediction`` (matches the legacy single-plate read).
|
||||
|
||||
Returns:
|
||||
Predicted print time in seconds, or None if not found / unparseable.
|
||||
"""
|
||||
try:
|
||||
with zipfile.ZipFile(file_path, "r") as zf:
|
||||
if "Metadata/slice_info.config" not in zf.namelist():
|
||||
return None
|
||||
|
||||
content = zf.read("Metadata/slice_info.config").decode()
|
||||
root = ET.fromstring(content)
|
||||
|
||||
if plate_id is not None:
|
||||
for plate_elem in root.findall(".//plate"):
|
||||
plate_index = None
|
||||
for meta in plate_elem.findall("metadata"):
|
||||
if meta.get("key") == "index":
|
||||
try:
|
||||
plate_index = int(meta.get("value", "0"))
|
||||
except ValueError:
|
||||
pass # Skip plate with unparseable index
|
||||
break
|
||||
|
||||
if plate_index == plate_id:
|
||||
for meta in plate_elem.findall("metadata"):
|
||||
if meta.get("key") == "prediction":
|
||||
try:
|
||||
return int(meta.get("value", "0"))
|
||||
except ValueError:
|
||||
return None
|
||||
break
|
||||
else:
|
||||
plate_elem = root.find(".//plate")
|
||||
if plate_elem is not None:
|
||||
for meta in plate_elem.findall("metadata"):
|
||||
if meta.get("key") == "prediction":
|
||||
try:
|
||||
return int(meta.get("value", "0"))
|
||||
except ValueError:
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.warning("Failed to extract print time from %s: %s", file_path, e)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def extract_bed_type_from_3mf(file_path: Path, plate_id: int | None = None) -> str | None:
|
||||
"""Extract the build plate type (`curr_bed_type`) for a specific plate (#1281).
|
||||
|
||||
|
||||
@@ -0,0 +1,326 @@
|
||||
"""Unit tests for `_scope_notification_archive_data_to_plate` (#1785).
|
||||
|
||||
The 3MF parser at services/archive.py:200-264 sums per-plate `prediction` and
|
||||
`weight` into archive-level totals (#1593) — correct for the archive card's
|
||||
"whole project" headline, wrong for the completion notification of a single
|
||||
plate. The helper under test mirrors what the queue UI does at
|
||||
print_queue.py:272-285: re-read the 3MF and substitute the plate's actual
|
||||
values for filament grams, time estimate, and per-slot breakdown.
|
||||
"""
|
||||
|
||||
import io
|
||||
import zipfile
|
||||
|
||||
from backend.app.main import _scope_notification_archive_data_to_plate
|
||||
|
||||
|
||||
def _write_multi_plate_3mf(tmp_path, name="multi.3mf") -> "tuple":
|
||||
"""Create a 3-plate 3MF with distinct prediction + weight per plate.
|
||||
|
||||
Plate 1: 30 min, 50g PLA
|
||||
Plate 2: 60 min, 120g PETG
|
||||
Plate 3: 90 min, 200g PLA
|
||||
"""
|
||||
xml_content = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<config>
|
||||
<plate>
|
||||
<metadata key="index" value="1"/>
|
||||
<metadata key="prediction" value="1800"/>
|
||||
<metadata key="weight" value="50"/>
|
||||
<filament id="1" used_g="50.0" type="PLA" color="#FF0000"/>
|
||||
</plate>
|
||||
<plate>
|
||||
<metadata key="index" value="2"/>
|
||||
<metadata key="prediction" value="3600"/>
|
||||
<metadata key="weight" value="120"/>
|
||||
<filament id="1" used_g="80.0" type="PETG" color="#00FF00"/>
|
||||
<filament id="2" used_g="40.0" type="PETG" color="#0000FF"/>
|
||||
</plate>
|
||||
<plate>
|
||||
<metadata key="index" value="3"/>
|
||||
<metadata key="prediction" value="5400"/>
|
||||
<metadata key="weight" value="200"/>
|
||||
<filament id="1" used_g="200.0" type="PLA" color="#FF0000"/>
|
||||
</plate>
|
||||
</config>
|
||||
"""
|
||||
buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(buffer, "w") as zf:
|
||||
zf.writestr("Metadata/slice_info.config", xml_content)
|
||||
buffer.seek(0)
|
||||
|
||||
file_path = tmp_path / name
|
||||
file_path.write_bytes(buffer.read())
|
||||
return file_path, "multi.3mf"
|
||||
|
||||
|
||||
def _project_totals_archive_data() -> dict:
|
||||
"""Pre-fix archive_data as `_background_notifications` constructs it: the
|
||||
summed-across-plates totals from PrintArchive's columns and extra_data."""
|
||||
return {
|
||||
# Summed: 30 + 60 + 90 min = 180 min = 10800s
|
||||
"print_time_seconds": 10800,
|
||||
"actual_time_seconds": None,
|
||||
# Summed: 50 + 120 + 200 = 370g
|
||||
"actual_filament_grams": 370.0,
|
||||
# Summed across all 3 plates' filament rows
|
||||
"filament_slots": [
|
||||
{"slot_id": 1, "used_g": 330.0, "type": "PLA", "color": "#FF0000"},
|
||||
{"slot_id": 2, "used_g": 40.0, "type": "PETG", "color": "#0000FF"},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
class TestScopeNotificationArchiveDataToPlate:
|
||||
def test_completed_plate_replaces_summed_totals(self, tmp_path):
|
||||
# The bug: notification shows project totals (370g, 3h) when only
|
||||
# plate 2 was printed. Expected after fix: plate 2's 120g and 60 min.
|
||||
file_path, rel = _write_multi_plate_3mf(tmp_path)
|
||||
archive_data = _project_totals_archive_data()
|
||||
|
||||
result = _scope_notification_archive_data_to_plate(
|
||||
archive_data,
|
||||
rel,
|
||||
plate_id=2,
|
||||
print_status="completed",
|
||||
progress=100,
|
||||
base_dir=tmp_path,
|
||||
)
|
||||
|
||||
assert result["actual_filament_grams"] == 120.0
|
||||
assert result["print_time_seconds"] == 3600
|
||||
assert result["filament_slots"] == [
|
||||
{"slot_id": 1, "used_g": 80.0, "type": "PETG", "color": "#00FF00"},
|
||||
{"slot_id": 2, "used_g": 40.0, "type": "PETG", "color": "#0000FF"},
|
||||
]
|
||||
|
||||
def test_plate_1_scoping_works(self, tmp_path):
|
||||
file_path, rel = _write_multi_plate_3mf(tmp_path)
|
||||
archive_data = _project_totals_archive_data()
|
||||
|
||||
result = _scope_notification_archive_data_to_plate(
|
||||
archive_data,
|
||||
rel,
|
||||
plate_id=1,
|
||||
print_status="completed",
|
||||
progress=100,
|
||||
base_dir=tmp_path,
|
||||
)
|
||||
|
||||
assert result["actual_filament_grams"] == 50.0
|
||||
assert result["print_time_seconds"] == 1800
|
||||
|
||||
def test_plate_3_scoping_works(self, tmp_path):
|
||||
file_path, rel = _write_multi_plate_3mf(tmp_path)
|
||||
archive_data = _project_totals_archive_data()
|
||||
|
||||
result = _scope_notification_archive_data_to_plate(
|
||||
archive_data,
|
||||
rel,
|
||||
plate_id=3,
|
||||
print_status="completed",
|
||||
progress=100,
|
||||
base_dir=tmp_path,
|
||||
)
|
||||
|
||||
assert result["actual_filament_grams"] == 200.0
|
||||
assert result["print_time_seconds"] == 5400
|
||||
|
||||
def test_partial_print_scales_plate_values(self, tmp_path):
|
||||
# Plate 2 cancelled at 50%: expect half the plate's grams + per-slot
|
||||
# values scaled, but full slicer estimate kept (callers display this
|
||||
# alongside the partial actual_filament_grams).
|
||||
file_path, rel = _write_multi_plate_3mf(tmp_path)
|
||||
archive_data = _project_totals_archive_data()
|
||||
|
||||
result = _scope_notification_archive_data_to_plate(
|
||||
archive_data,
|
||||
rel,
|
||||
plate_id=2,
|
||||
print_status="cancelled",
|
||||
progress=50,
|
||||
base_dir=tmp_path,
|
||||
)
|
||||
|
||||
assert result["actual_filament_grams"] == 60.0
|
||||
assert result["print_time_seconds"] == 3600
|
||||
assert result["filament_slots"] == [
|
||||
{"slot_id": 1, "used_g": 40.0, "type": "PETG", "color": "#00FF00"},
|
||||
{"slot_id": 2, "used_g": 20.0, "type": "PETG", "color": "#0000FF"},
|
||||
]
|
||||
|
||||
def test_no_plate_id_returns_unchanged(self, tmp_path):
|
||||
# Single-plate prints or non-plate-scoped completions take the
|
||||
# project-level archive values as-is.
|
||||
file_path, rel = _write_multi_plate_3mf(tmp_path)
|
||||
archive_data = _project_totals_archive_data()
|
||||
before = {**archive_data, "filament_slots": list(archive_data["filament_slots"])}
|
||||
|
||||
result = _scope_notification_archive_data_to_plate(
|
||||
archive_data,
|
||||
rel,
|
||||
plate_id=None,
|
||||
print_status="completed",
|
||||
progress=100,
|
||||
base_dir=tmp_path,
|
||||
)
|
||||
|
||||
assert result["actual_filament_grams"] == before["actual_filament_grams"]
|
||||
assert result["print_time_seconds"] == before["print_time_seconds"]
|
||||
assert result["filament_slots"] == before["filament_slots"]
|
||||
|
||||
def test_no_file_path_returns_unchanged(self, tmp_path):
|
||||
archive_data = _project_totals_archive_data()
|
||||
before = {**archive_data}
|
||||
|
||||
result = _scope_notification_archive_data_to_plate(
|
||||
archive_data,
|
||||
None,
|
||||
plate_id=2,
|
||||
print_status="completed",
|
||||
progress=100,
|
||||
base_dir=tmp_path,
|
||||
)
|
||||
|
||||
assert result["actual_filament_grams"] == before["actual_filament_grams"]
|
||||
assert result["print_time_seconds"] == before["print_time_seconds"]
|
||||
|
||||
def test_missing_3mf_returns_unchanged(self, tmp_path):
|
||||
# Archive's file may have been deleted (manual cleanup) between print
|
||||
# completion and the notification firing — must not blow up the
|
||||
# notification, just send the project-level numbers we already have.
|
||||
archive_data = _project_totals_archive_data()
|
||||
before = {**archive_data}
|
||||
|
||||
result = _scope_notification_archive_data_to_plate(
|
||||
archive_data,
|
||||
"missing.3mf",
|
||||
plate_id=2,
|
||||
print_status="completed",
|
||||
progress=100,
|
||||
base_dir=tmp_path,
|
||||
)
|
||||
|
||||
assert result["actual_filament_grams"] == before["actual_filament_grams"]
|
||||
assert result["print_time_seconds"] == before["print_time_seconds"]
|
||||
|
||||
def test_corrupt_3mf_returns_unchanged(self, tmp_path):
|
||||
# Invalid file at the right path: helper falls back gracefully.
|
||||
bad_path = tmp_path / "bad.3mf"
|
||||
bad_path.write_text("not a zip file")
|
||||
|
||||
archive_data = _project_totals_archive_data()
|
||||
before = {**archive_data}
|
||||
|
||||
result = _scope_notification_archive_data_to_plate(
|
||||
archive_data,
|
||||
"bad.3mf",
|
||||
plate_id=2,
|
||||
print_status="completed",
|
||||
progress=100,
|
||||
base_dir=tmp_path,
|
||||
)
|
||||
|
||||
assert result["actual_filament_grams"] == before["actual_filament_grams"]
|
||||
assert result["print_time_seconds"] == before["print_time_seconds"]
|
||||
|
||||
def test_plate_id_outside_range_returns_unchanged(self, tmp_path):
|
||||
# Defensive: if plate_id doesn't match any plate in the 3MF, leave the
|
||||
# project-level numbers alone rather than emitting zeros.
|
||||
file_path, rel = _write_multi_plate_3mf(tmp_path)
|
||||
archive_data = _project_totals_archive_data()
|
||||
before = {**archive_data}
|
||||
|
||||
result = _scope_notification_archive_data_to_plate(
|
||||
archive_data,
|
||||
rel,
|
||||
plate_id=99,
|
||||
print_status="completed",
|
||||
progress=100,
|
||||
base_dir=tmp_path,
|
||||
)
|
||||
|
||||
assert result["actual_filament_grams"] == before["actual_filament_grams"]
|
||||
assert result["print_time_seconds"] == before["print_time_seconds"]
|
||||
|
||||
def test_zero_grams_plate_keeps_project_level_breakdown(self, tmp_path):
|
||||
# Defensive: a 3MF that emits per-plate filament rows summing to zero
|
||||
# (slicer bug / re-slice without estimate) must NOT clobber the
|
||||
# project-level grams + per-slot breakdown the archive columns already
|
||||
# provide — otherwise the notification would headline "370 g" next to
|
||||
# an all-zero per-slot breakdown.
|
||||
xml_content = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<config>
|
||||
<plate>
|
||||
<metadata key="index" value="1"/>
|
||||
<metadata key="prediction" value="1800"/>
|
||||
<metadata key="weight" value="0"/>
|
||||
<filament id="1" used_g="0" type="PLA" color="#FF0000"/>
|
||||
</plate>
|
||||
</config>
|
||||
"""
|
||||
buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(buffer, "w") as zf:
|
||||
zf.writestr("Metadata/slice_info.config", xml_content)
|
||||
buffer.seek(0)
|
||||
file_path = tmp_path / "zero.3mf"
|
||||
file_path.write_bytes(buffer.read())
|
||||
|
||||
archive_data = _project_totals_archive_data()
|
||||
before_slots = list(archive_data["filament_slots"])
|
||||
before_grams = archive_data["actual_filament_grams"]
|
||||
|
||||
result = _scope_notification_archive_data_to_plate(
|
||||
archive_data,
|
||||
"zero.3mf",
|
||||
plate_id=1,
|
||||
print_status="completed",
|
||||
progress=100,
|
||||
base_dir=tmp_path,
|
||||
)
|
||||
|
||||
# Time still scopes (prediction parsed cleanly).
|
||||
assert result["print_time_seconds"] == 1800
|
||||
# Grams + per-slot breakdown stay on project-level so the notification
|
||||
# doesn't ship an inconsistent headline.
|
||||
assert result["actual_filament_grams"] == before_grams
|
||||
assert result["filament_slots"] == before_slots
|
||||
|
||||
def test_single_plate_file_with_plate_id_1(self, tmp_path):
|
||||
# Single-plate 3MF where queue still has plate_id=1 set: the parser's
|
||||
# "sum across plates" already collapses to plate 1's values, so the
|
||||
# helper just confirms (no double-scaling, no field clobber).
|
||||
xml_content = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<config>
|
||||
<plate>
|
||||
<metadata key="index" value="1"/>
|
||||
<metadata key="prediction" value="2400"/>
|
||||
<metadata key="weight" value="75"/>
|
||||
<filament id="1" used_g="75.0" type="PLA" color="#0000FF"/>
|
||||
</plate>
|
||||
</config>
|
||||
"""
|
||||
buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(buffer, "w") as zf:
|
||||
zf.writestr("Metadata/slice_info.config", xml_content)
|
||||
buffer.seek(0)
|
||||
file_path = tmp_path / "single.3mf"
|
||||
file_path.write_bytes(buffer.read())
|
||||
|
||||
archive_data = {
|
||||
"print_time_seconds": 2400,
|
||||
"actual_time_seconds": None,
|
||||
"actual_filament_grams": 75.0,
|
||||
}
|
||||
|
||||
result = _scope_notification_archive_data_to_plate(
|
||||
archive_data,
|
||||
"single.3mf",
|
||||
plate_id=1,
|
||||
print_status="completed",
|
||||
progress=100,
|
||||
base_dir=tmp_path,
|
||||
)
|
||||
|
||||
assert result["actual_filament_grams"] == 75.0
|
||||
assert result["print_time_seconds"] == 2400
|
||||
@@ -14,6 +14,7 @@ from backend.app.utils.threemf_tools import (
|
||||
extract_embedded_presets_from_3mf,
|
||||
extract_filament_usage_from_3mf,
|
||||
extract_plate_extruder_set_from_3mf,
|
||||
extract_print_time_from_3mf,
|
||||
extract_project_filaments_from_3mf,
|
||||
get_cumulative_usage_at_layer,
|
||||
mm_to_grams,
|
||||
@@ -841,3 +842,102 @@ class TestExtractBedTypeFrom3mf:
|
||||
file_path.write_bytes(create_mock_3mf(xml_content).read())
|
||||
|
||||
assert extract_bed_type_from_3mf(file_path) == "Textured PEI Plate"
|
||||
|
||||
|
||||
class TestExtractPrintTimeFrom3mf:
|
||||
"""Tests for extract_print_time_from_3mf — the per-plate `prediction` reader
|
||||
used by the completion notification path to scope the archive-level (summed)
|
||||
total down to the actually-printed plate (#1785)."""
|
||||
|
||||
def test_returns_plate_prediction_when_plate_id_matches(self, tmp_path):
|
||||
xml_content = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<config>
|
||||
<plate>
|
||||
<metadata key="index" value="1"/>
|
||||
<metadata key="prediction" value="3600"/>
|
||||
</plate>
|
||||
<plate>
|
||||
<metadata key="index" value="2"/>
|
||||
<metadata key="prediction" value="7200"/>
|
||||
</plate>
|
||||
<plate>
|
||||
<metadata key="index" value="3"/>
|
||||
<metadata key="prediction" value="10800"/>
|
||||
</plate>
|
||||
</config>
|
||||
"""
|
||||
file_path = tmp_path / "test.3mf"
|
||||
file_path.write_bytes(create_mock_3mf(xml_content).read())
|
||||
|
||||
assert extract_print_time_from_3mf(file_path, plate_id=2) == 7200
|
||||
|
||||
def test_returns_first_plate_when_no_plate_id(self, tmp_path):
|
||||
xml_content = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<config>
|
||||
<plate>
|
||||
<metadata key="index" value="1"/>
|
||||
<metadata key="prediction" value="900"/>
|
||||
</plate>
|
||||
<plate>
|
||||
<metadata key="index" value="2"/>
|
||||
<metadata key="prediction" value="1800"/>
|
||||
</plate>
|
||||
</config>
|
||||
"""
|
||||
file_path = tmp_path / "test.3mf"
|
||||
file_path.write_bytes(create_mock_3mf(xml_content).read())
|
||||
|
||||
assert extract_print_time_from_3mf(file_path) == 900
|
||||
|
||||
def test_returns_none_when_plate_id_missing(self, tmp_path):
|
||||
xml_content = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<config>
|
||||
<plate>
|
||||
<metadata key="index" value="1"/>
|
||||
<metadata key="prediction" value="3600"/>
|
||||
</plate>
|
||||
</config>
|
||||
"""
|
||||
file_path = tmp_path / "test.3mf"
|
||||
file_path.write_bytes(create_mock_3mf(xml_content).read())
|
||||
|
||||
assert extract_print_time_from_3mf(file_path, plate_id=5) is None
|
||||
|
||||
def test_returns_none_when_prediction_unparseable(self, tmp_path):
|
||||
xml_content = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<config>
|
||||
<plate>
|
||||
<metadata key="index" value="1"/>
|
||||
<metadata key="prediction" value="not-a-number"/>
|
||||
</plate>
|
||||
</config>
|
||||
"""
|
||||
file_path = tmp_path / "test.3mf"
|
||||
file_path.write_bytes(create_mock_3mf(xml_content).read())
|
||||
|
||||
assert extract_print_time_from_3mf(file_path, plate_id=1) is None
|
||||
|
||||
def test_returns_none_when_slice_info_missing(self, tmp_path):
|
||||
buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(buffer, "w") as zf:
|
||||
zf.writestr("other_file.txt", "content")
|
||||
buffer.seek(0)
|
||||
|
||||
file_path = tmp_path / "test.3mf"
|
||||
file_path.write_bytes(buffer.read())
|
||||
|
||||
assert extract_print_time_from_3mf(file_path) is None
|
||||
assert extract_print_time_from_3mf(file_path, plate_id=1) is None
|
||||
|
||||
def test_returns_none_when_file_invalid(self, tmp_path):
|
||||
file_path = tmp_path / "invalid.3mf"
|
||||
file_path.write_text("not a zip file")
|
||||
|
||||
assert extract_print_time_from_3mf(file_path) is None
|
||||
assert extract_print_time_from_3mf(file_path, plate_id=1) is None
|
||||
|
||||
def test_returns_none_when_file_missing(self, tmp_path):
|
||||
file_path = tmp_path / "nonexistent.3mf"
|
||||
|
||||
assert extract_print_time_from_3mf(file_path) is None
|
||||
assert extract_print_time_from_3mf(file_path, plate_id=2) is None
|
||||
|
||||
Reference in New Issue
Block a user