Charge every filament that shares a tray, not just the first (#3230)

The 3MF pass skipped a tray once it had charged it, so a second slicer
filament mapped to the same tray was never deducted, and a single-filament
print that switched trays and came back lost its last segment. Deduplicate
by slicer slot instead. Trays covered before the pass, and trays only
guessed by position that were already charged, are still skipped; charged
trays are still reported to the remain% fallback.

backend/app/services/usage_tracker.py

@@ -1632,12 +1632,23 @@ async def _track_from_3mf(
     # Collected rather than acted on inline so one notification covers the whole
     # print instead of one per slot (#2812).
     unassigned_global_trays: list[int] = []
+    # Deduplicated by slicer slot, not by tray (#3230). Several slots mapped to
+    # one tray, or a single-slot print that returns to an earlier tray, each
+    # consume filament of their own. A slot only repeats when the plate is
+    # unknown and every plate of the file is listed, and that must not charge
+    # the same slot once per plate. Only a tray the caller had already covered
+    # before this pass is skipped.
+    charged_slots: set[int] = set()
+    covered_before = frozenset(handled_trays)
 
     for usage in filament_usage:
         slot_id = usage.get("slot_id", 0)
         used_g = usage.get("used_g", 0)
         if used_g <= 0:
             continue
+        if slot_id in charged_slots:
+            continue
+        charged_slots.add(slot_id)
 
         # --- Mid-print tray switch: split weight across trays ---
         # Split math is shared with the Spoolman writer via
@@ -1698,7 +1709,7 @@ async def _track_from_3mf(
                     seg_tray_id = tray_global % 4
 
                 seg_key = (seg_ams_id, seg_tray_id)
-                if seg_key in handled_trays:
+                if seg_key in covered_before:
                     continue
 
                 seg_start_layer = tray_changes[seg_idx][1]
@@ -1787,6 +1798,7 @@ async def _track_from_3mf(
             continue  # Skip normal single-tray processing for this slot
 
         # Map 3MF slot_id to physical (ams_id, tray_id) using resolved mapping
+        tray_is_guessed = False
         if tray_now_override is not None:
             # Single-filament non-queue print: use actual tray from printer state
             global_tray_id = tray_now_override
@@ -1822,6 +1834,7 @@ async def _track_from_3mf(
             # never gets recorded. vt_tray entries are already filtered the
             # same way inside `build_ams_tray_lookup` (line 174 checks
             # `tray_type`), so this just mirrors that for the AMS side.
+            tray_is_guessed = global_tray_id is None
             if global_tray_id is None:
                 _state = printer_manager.get_status(printer_id)
                 _raw = getattr(_state, "raw_data", None) if _state else None
@@ -1858,7 +1871,12 @@ async def _track_from_3mf(
         )
 
         key = (ams_id, tray_id)
-        if key in handled_trays:
+        if key in covered_before:
+            continue
+        if tray_is_guessed and key in handled_trays:
+            # A guess by position can land a second slot on a tray another slot
+            # was charged for. That says nothing about which spool fed it, so
+            # it is not charged to that tray's spool as well.
             continue
 
         spool_id = await _resolve_spool_id_for_tray(

backend/tests/unit/test_shared_tray_usage_3230.py

@@ -0,0 +1,169 @@
+"""Several slicer slots charged to one tray (#3230).
+
+The 3MF pass skipped a tray once it had charged it, so a second slicer slot
+mapped to the same tray was dropped. The reporter's H2C printed filaments 1 and
+4 from one AMS-HT tray (mapping ``[128, -1, -1, 128, -1]``) and lost filament
+4's 2.13 g. The same skip dropped the last segment of a single-filament print
+that switched to another tray and back. A slot is now charged once and a tray
+as often as slots use it.
+"""
+
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.models.archive import PrintArchive
+from backend.app.models.spool import Spool
+from backend.app.services.usage_tracker import _track_from_3mf
+
+pytestmark = pytest.mark.unit
+
+
+def _spool(spool_id: int):
+    spool = MagicMock()
+    spool.id = spool_id
+    spool.label_weight = 1000
+    spool.weight_used = 0
+    spool.cost_per_kg = None
+    spool.material = "PLA"
+    spool.rgba = None
+    return spool
+
+
+def _db(spools: dict[int, MagicMock]):
+    archive = MagicMock()
+    archive.id = 3230
+    archive.file_path = "archives/3230/test.3mf"
+    archive.extra_data = None
+    archive.plate_id = None
+
+    async def execute(stmt, *args, **kwargs):
+        entity = stmt.column_descriptions[0].get("entity")
+        result = MagicMock()
+        value = None
+        if entity is PrintArchive:
+            value = archive
+        elif entity is Spool:
+            value = spools.get(stmt.whereclause.right.value)
+        result.scalar_one_or_none.return_value = value
+        result.scalars.return_value.first.return_value = None
+        result.scalar.return_value = None
+        return result
+
+    db = AsyncMock()
+    db.execute = execute
+    db.add = MagicMock()
+    return db
+
+
+# HT tray 128 holds spool 7; AMS 0 slot 1 holds spool 8.
+ASSIGNED = {(128, 0): 7, (0, 1): 8}
+
+
+async def _track(ams_mapping, filament_usage, tray_change_log=None, total_layers=0, handled_trays=None, raw_data=None):
+    spools = {7: _spool(7), 8: _spool(8)}
+    printer_manager = MagicMock()
+    printer_manager.get_status.return_value = SimpleNamespace(
+        progress=100,
+        layer_num=90,
+        total_layers=total_layers,
+        tray_now=128,
+        last_loaded_tray=-1,
+        tray_change_log=tray_change_log or [],
+        raw_data=raw_data or {},
+    )
+
+    async def resolve(printer_id, ams_id, tray_id, **kwargs):
+        return ASSIGNED.get((ams_id, tray_id))
+
+    with (
+        patch("backend.app.core.config.settings") as mock_settings,
+        patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
+        patch("backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf", return_value=None),
+        patch("backend.app.utils.threemf_tools.extract_filament_properties_from_3mf", return_value={}),
+        patch("backend.app.services.usage_tracker._resolve_spool_id_for_tray", side_effect=resolve),
+    ):
+        mock_path = MagicMock()
+        mock_path.exists.return_value = True
+        mock_settings.base_dir = MagicMock()
+        mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+        results = await _track_from_3mf(
+            printer_id=1,
+            archive_id=3230,
+            status="completed",
+            print_name="shared_tray",
+            handled_trays=handled_trays if handled_trays is not None else set(),
+            printer_manager=printer_manager,
+            db=_db(spools),
+            ams_mapping=ams_mapping,
+            print_started_at=datetime.now(timezone.utc),
+        )
+    return results, spools
+
+
+def _usage(slot_id: int, used_g: float) -> dict:
+    return {"slot_id": slot_id, "used_g": used_g, "type": "PLA", "color": "#FFFFFF"}
+
+
+class TestSharedTray:
+    @pytest.mark.asyncio
+    async def test_two_slots_on_one_tray_are_both_charged(self):
+        # The reporter's job, verbatim mapping and grams.
+        results, spools = await _track([128, -1, -1, 128, -1], [_usage(1, 1.05), _usage(4, 2.13)])
+        assert [(r["slot_id"], r["spool_id"], r["weight_used"]) for r in results] == [(1, 7, 1.1), (4, 7, 2.1)]
+        assert spools[7].weight_used == pytest.approx(3.18)
+
+    @pytest.mark.asyncio
+    async def test_slots_on_different_trays_are_unchanged(self):
+        results, spools = await _track([128, 1], [_usage(1, 1.0), _usage(2, 2.0)])
+        assert [r["spool_id"] for r in results] == [7, 8]
+        assert spools[7].weight_used == pytest.approx(1.0)
+        assert spools[8].weight_used == pytest.approx(2.0)
+
+    @pytest.mark.asyncio
+    async def test_a_shared_tray_is_still_handed_to_the_remain_fallback_as_covered(self):
+        handled: set = set()
+        await _track([128, -1, -1, 128, -1], [_usage(1, 1.05), _usage(4, 2.13)], handled_trays=handled)
+        assert handled == {(128, 0)}
+
+    @pytest.mark.asyncio
+    async def test_a_slot_listed_once_per_plate_is_charged_once(self):
+        # With the plate unknown, every plate of the file is listed, so slot 1
+        # appears once per plate. Charging it per plate would multiply the debit.
+        results, spools = await _track([128], [_usage(1, 4.0), _usage(1, 4.0), _usage(1, 4.0)])
+        assert len(results) == 1
+        assert spools[7].weight_used == pytest.approx(4.0)
+
+
+class TestReturningToATray:
+    @pytest.mark.asyncio
+    async def test_every_segment_is_charged_when_a_print_returns_to_an_earlier_tray(self):
+        # Single-filament print: HT tray, then AMS 0 slot 1, then back to the HT
+        # tray. Linear split over 90 layers gives a third per segment.
+        results, spools = await _track(
+            [128],
+            [_usage(1, 9.0)],
+            tray_change_log=[(128, 0), (1, 30), (128, 60)],
+            total_layers=90,
+        )
+        assert [r["spool_id"] for r in results] == [7, 8, 7]
+        assert spools[7].weight_used == pytest.approx(6.0)
+        assert spools[8].weight_used == pytest.approx(3.0)
+
+
+class TestGuessedTrays:
+    @pytest.mark.asyncio
+    async def test_a_guessed_tray_already_charged_is_not_charged_again(self):
+        # No mapping: slot 1 takes the only loaded tray (AMS 0 slot 1, global
+        # 1), slot 2 is past the loaded trays and falls back to slot_id - 1,
+        # the same tray. A position guess says nothing about which spool fed
+        # slot 2, so it is not added to slot 1's spool.
+        raw = {"ams": [{"id": "0", "tray": [{"id": "1", "tray_type": "PLA"}]}]}
+        with patch(
+            "backend.app.services.spoolman_tracking.build_ams_tray_lookup", return_value={1: {"tray_type": "PLA"}}
+        ):
+            results, spools = await _track(None, [_usage(1, 1.0), _usage(2, 2.0)], raw_data=raw)
+        assert [r["slot_id"] for r in results] == [1]
+        assert spools[8].weight_used == pytest.approx(1.0)