|
|
@@ -0,0 +1,207 @@
|
|
|
+"""An X2D print from the right-hand external spool, sent without ams_mapping2 (#3405).
|
|
|
+
|
|
|
+The X2D has two external holders and lists both in ``vt_tray``, loaded or not.
|
|
|
+A print command that carries no ``ams_mapping2`` leaves the external spool as -1
|
|
|
+in the flat mapping (#3166 resolves it only from ams_mapping2). Until #3166's
|
|
|
+follow-up, Spoolman mode read that -1 as the first external holder present,
|
|
|
+254, and charged the empty left holder; since then it charged nothing. The
|
|
|
+built-in inventory drops a mapping that names no tray for any used slot and
|
|
|
+lets the printer's ``tray_now`` decide. Spoolman mode now does the same, and
|
|
|
+reads ``tray_now`` 255 as the right-hand holder on a printer that has one, as
|
|
|
+the built-in inventory does.
|
|
|
+
|
|
|
+Values are the reporter's ``/status``: 254 empty, 255 loaded with red PLA,
|
|
|
+``tray_now`` 255.
|
|
|
+"""
|
|
|
+
|
|
|
+from types import SimpleNamespace
|
|
|
+from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
+
|
|
|
+import pytest
|
|
|
+
|
|
|
+from backend.app.services import spoolman_tracking
|
|
|
+from backend.app.services.spoolman_tracking import _single_slot_tray_from_state
|
|
|
+from backend.app.services.usage_tracker import has_second_external_holder, mapping_names_no_used_tray
|
|
|
+
|
|
|
+X2D_VT_TRAY = [
|
|
|
+ {"id": 254, "tray_type": "", "tray_color": "00000000", "tray_info_idx": ""},
|
|
|
+ {"id": 255, "tray_type": "PLA", "tray_color": "F72323FF", "tray_info_idx": "GFA01"},
|
|
|
+]
|
|
|
+USAGE = [{"slot_id": 1, "used_g": 12.5, "type": "PLA", "color": "#F72323"}]
|
|
|
+RIGHT_HOLDER_SPOOL = 77
|
|
|
+
|
|
|
+
|
|
|
+def _x2d_state(tray_now=255, vt_tray=X2D_VT_TRAY, last_loaded_tray=-1, tray_change_log=(), **raw):
|
|
|
+ return SimpleNamespace(
|
|
|
+ raw_data={"vt_tray": vt_tray, **raw},
|
|
|
+ layer_num=0,
|
|
|
+ total_layers=0,
|
|
|
+ tray_change_log=list(tray_change_log),
|
|
|
+ tray_now=tray_now,
|
|
|
+ last_loaded_tray=last_loaded_tray,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+class TestTheRules:
|
|
|
+ def test_a_mapping_of_only_minus_one_names_no_used_tray(self):
|
|
|
+ assert mapping_names_no_used_tray([-1], USAGE)
|
|
|
+ assert not mapping_names_no_used_tray([255], USAGE)
|
|
|
+ # Another slot is mapped: the -1 is a slot the print does not feed.
|
|
|
+ two = [*USAGE, {"slot_id": 2, "used_g": 3.0}]
|
|
|
+ assert not mapping_names_no_used_tray([-1, 5], two)
|
|
|
+ assert not mapping_names_no_used_tray(None, USAGE)
|
|
|
+
|
|
|
+ def test_255_is_a_tray_only_on_a_printer_with_two_holders(self):
|
|
|
+ assert has_second_external_holder({"vt_tray": X2D_VT_TRAY})
|
|
|
+ assert not has_second_external_holder({"vt_tray": [X2D_VT_TRAY[0]]})
|
|
|
+ assert not has_second_external_holder(None)
|
|
|
+
|
|
|
+ def test_tray_now_255_is_the_right_holder(self):
|
|
|
+ assert _single_slot_tray_from_state(_x2d_state(), USAGE, 255) == (1, 255)
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("stale", [3, 254])
|
|
|
+ def test_the_right_holder_beats_a_tray_an_earlier_print_loaded(self, stale):
|
|
|
+ """A print from 255 never updates last_loaded_tray: it still names the
|
|
|
+ AMS tray, or the left holder, an earlier print used."""
|
|
|
+ state = _x2d_state(last_loaded_tray=stale)
|
|
|
+ assert _single_slot_tray_from_state(state, USAGE, 255) == (1, 255)
|
|
|
+
|
|
|
+ def test_a_tray_loaded_during_the_print_still_wins_over_255(self):
|
|
|
+ state = _x2d_state(last_loaded_tray=3, tray_change_log=[(3, 2)])
|
|
|
+ assert _single_slot_tray_from_state(state, USAGE, 255) == (1, 3)
|
|
|
+
|
|
|
+ def test_last_loaded_254_is_no_survivor(self):
|
|
|
+ """Same cap as the built-in inventory: 0-253 only."""
|
|
|
+ state = _x2d_state(vt_tray=[X2D_VT_TRAY[0]], last_loaded_tray=254)
|
|
|
+ assert _single_slot_tray_from_state(state, USAGE, 255) is None
|
|
|
+
|
|
|
+ def test_tray_now_255_on_a_one_holder_printer_is_nothing(self):
|
|
|
+ assert _single_slot_tray_from_state(_x2d_state(vt_tray=[X2D_VT_TRAY[0]]), USAGE, 255) is None
|
|
|
+
|
|
|
+
|
|
|
+class _AsyncCtx:
|
|
|
+ def __init__(self, db):
|
|
|
+ self._db = db
|
|
|
+
|
|
|
+ async def __aenter__(self):
|
|
|
+ return self._db
|
|
|
+
|
|
|
+ async def __aexit__(self, *exc):
|
|
|
+ return False
|
|
|
+
|
|
|
+
|
|
|
+def _slot_assignment(_printer_id, ams_id, tray_id):
|
|
|
+ # The reporter's spool, assigned to the holder it sits on: 255 is (255, 1).
|
|
|
+ return RIGHT_HOLDER_SPOOL if (ams_id, tray_id) == (255, 1) else None
|
|
|
+
|
|
|
+
|
|
|
+def _client():
|
|
|
+ client = AsyncMock()
|
|
|
+ client.find_spool_by_tag = AsyncMock(return_value=None)
|
|
|
+ client.get_spool = AsyncMock(
|
|
|
+ return_value={"id": RIGHT_HOLDER_SPOOL, "filament": {"color_hex": "F72323", "material": "PLA"}}
|
|
|
+ )
|
|
|
+ client.use_spool = AsyncMock()
|
|
|
+ return client
|
|
|
+
|
|
|
+
|
|
|
+def _patches(client, state, db):
|
|
|
+ pm = MagicMock()
|
|
|
+ pm.get_status.return_value = state
|
|
|
+ return (
|
|
|
+ patch.object(spoolman_tracking, "async_session", lambda: _AsyncCtx(db)),
|
|
|
+ patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
|
|
|
+ patch.object(spoolman_tracking, "_get_spoolman_client_with_fallback", AsyncMock(return_value=client)),
|
|
|
+ patch.object(spoolman_tracking, "_get_printer_serial", AsyncMock(return_value="SER")),
|
|
|
+ patch.object(
|
|
|
+ spoolman_tracking, "_resolve_spool_id_via_slot_assignment", AsyncMock(side_effect=_slot_assignment)
|
|
|
+ ),
|
|
|
+ patch("backend.app.services.printer_manager.printer_manager", pm),
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _tracking(slot_to_tray):
|
|
|
+ return SimpleNamespace(
|
|
|
+ filament_usage=list(USAGE),
|
|
|
+ ams_trays={
|
|
|
+ "254": {"tray_uuid": "", "tag_uid": "", "tray_type": ""},
|
|
|
+ "255": {"tray_uuid": "", "tag_uid": "", "tray_type": "PLA"},
|
|
|
+ },
|
|
|
+ slot_to_tray=slot_to_tray,
|
|
|
+ tray_remain_start=None,
|
|
|
+ layer_usage=None,
|
|
|
+ filament_properties=None,
|
|
|
+ tray_now_at_start=255,
|
|
|
+ archive_id=12,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+class TestReportUsage:
|
|
|
+ @staticmethod
|
|
|
+ async def _run(slot_to_tray, state=None):
|
|
|
+ archive = SimpleNamespace(filament_color="#F72323", filament_type="PLA")
|
|
|
+ rows = iter([_tracking(slot_to_tray)])
|
|
|
+
|
|
|
+ def _next_row(*_args, **_kwargs):
|
|
|
+ result = MagicMock()
|
|
|
+ result.scalar_one_or_none.return_value = next(rows, archive)
|
|
|
+ return result
|
|
|
+
|
|
|
+ db = AsyncMock()
|
|
|
+ db.execute = AsyncMock(side_effect=_next_row)
|
|
|
+ db.delete = AsyncMock()
|
|
|
+ db.commit = AsyncMock()
|
|
|
+ client = _client()
|
|
|
+ p = _patches(client, state or _x2d_state(), db)
|
|
|
+ with p[0], p[1], p[2], p[3], p[4], p[5]:
|
|
|
+ await spoolman_tracking.report_usage(printer_id=1, archive_id=12)
|
|
|
+ return client
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_the_loaded_right_holder_is_charged(self):
|
|
|
+ client = await self._run([-1])
|
|
|
+
|
|
|
+ client.use_spool.assert_awaited_once_with(RIGHT_HOLDER_SPOOL, 12.5)
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_same_coloured_ams_tray_is_not_charged_instead(self):
|
|
|
+ """The colour match knows no external holder; the built-in inventory
|
|
|
+ never runs it for a mapping that named no tray."""
|
|
|
+ ams = [{"id": 0, "tray": [{"id": 0, "tray_color": "F72323FF", "tray_type": "PLA"}]}]
|
|
|
+ client = await self._run([-1], state=_x2d_state(ams=ams))
|
|
|
+
|
|
|
+ client.use_spool.assert_awaited_once_with(RIGHT_HOLDER_SPOOL, 12.5)
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_resolved_mapping_is_unchanged(self):
|
|
|
+ """With ams_mapping2 the captured mapping already says 255 (#3166)."""
|
|
|
+ client = await self._run([255])
|
|
|
+
|
|
|
+ client.use_spool.assert_awaited_once_with(RIGHT_HOLDER_SPOOL, 12.5)
|
|
|
+
|
|
|
+
|
|
|
+class TestPartialUsage:
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_an_aborted_print_charges_the_right_holder_too(self):
|
|
|
+ tracking = _tracking([-1])
|
|
|
+ tracking.layer_usage = {}
|
|
|
+ # A same-coloured AMS tray must not take it either, as on completion.
|
|
|
+ ams = [{"id": 0, "tray": [{"id": 0, "tray_color": "F72323FF", "tray_type": "PLA"}]}]
|
|
|
+ state = _x2d_state(ams=ams)
|
|
|
+ state.layer_num, state.total_layers = 50, 100
|
|
|
+ report = AsyncMock(return_value=1)
|
|
|
+ p = _patches(_client(), state, AsyncMock())
|
|
|
+ with (
|
|
|
+ p[0],
|
|
|
+ p[1],
|
|
|
+ p[2],
|
|
|
+ p[3],
|
|
|
+ p[4],
|
|
|
+ p[5],
|
|
|
+ patch.object(spoolman_tracking, "_report_spool_usage_for_slots", report),
|
|
|
+ ):
|
|
|
+ await spoolman_tracking._report_partial_usage(1, tracking, last_progress=50)
|
|
|
+
|
|
|
+ report.assert_awaited()
|
|
|
+ slot_to_tray = report.await_args.args[3]
|
|
|
+ assert spoolman_tracking._resolve_global_tray_id(1, slot_to_tray, {254: {}, 255: {}}) == 255
|
maziggy