Przeglądaj źródła

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.
maziggy 2 miesięcy temu
rodzic
commit
2f6007a148

Plik diff jest za duży
+ 0 - 0
CHANGELOG.md


+ 9 - 51
backend/app/api/routes/print_queue.py

@@ -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
backend/app/main.py

@@ -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)

+ 58 - 0
backend/app/utils/threemf_tools.py

@@ -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).
 

+ 326 - 0
backend/tests/unit/test_notification_plate_scope.py

@@ -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

+ 100 - 0
backend/tests/unit/test_threemf_tools.py

@@ -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

Niektóre pliki nie zostały wyświetlone z powodu dużej ilości zmienionych plików