test_scheduler_ams_mapping.py 40 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009
  1. """Tests for the AMS mapping computation in the print scheduler."""
  2. import io
  3. import json
  4. import zipfile
  5. import pytest
  6. from backend.app.services.print_scheduler import PrintScheduler
  7. from backend.app.utils.threemf_tools import extract_nozzle_mapping_from_3mf
  8. class TestSchedulerAmsMappingHelpers:
  9. """Test the AMS mapping helper methods in PrintScheduler."""
  10. @pytest.fixture
  11. def scheduler(self):
  12. return PrintScheduler()
  13. def test_normalize_color_with_hash(self, scheduler):
  14. """Color with hash should return #RRGGBB format."""
  15. result = scheduler._normalize_color("#FF5500")
  16. assert result == "#FF5500"
  17. def test_normalize_color_without_hash(self, scheduler):
  18. """Color without hash should add hash prefix."""
  19. result = scheduler._normalize_color("FF5500")
  20. assert result == "#FF5500"
  21. def test_normalize_color_with_alpha(self, scheduler):
  22. """Color with alpha channel should strip it."""
  23. result = scheduler._normalize_color("FF5500AA")
  24. assert result == "#FF5500"
  25. def test_normalize_color_none(self, scheduler):
  26. """None color should return default gray."""
  27. result = scheduler._normalize_color(None)
  28. assert result == "#808080"
  29. def test_normalize_color_empty(self, scheduler):
  30. """Empty color should return default gray."""
  31. result = scheduler._normalize_color("")
  32. assert result == "#808080"
  33. def test_normalize_color_for_compare(self, scheduler):
  34. """Color for compare should be lowercase without hash."""
  35. result = scheduler._normalize_color_for_compare("#FF5500")
  36. assert result == "ff5500"
  37. def test_normalize_color_for_compare_with_alpha(self, scheduler):
  38. """Alpha channel should be stripped for comparison."""
  39. result = scheduler._normalize_color_for_compare("#FF5500AA")
  40. assert result == "ff5500"
  41. def test_colors_are_similar_exact_match(self, scheduler):
  42. """Exact same colors should be similar."""
  43. assert scheduler._colors_are_similar("#FF5500", "#FF5500") is True
  44. def test_colors_are_similar_within_threshold(self, scheduler):
  45. """Colors within threshold should be similar."""
  46. # Red difference of 10, well within default threshold of 40
  47. assert scheduler._colors_are_similar("#FF5500", "#F55500") is True
  48. def test_colors_are_similar_outside_threshold(self, scheduler):
  49. """Colors outside threshold should not be similar."""
  50. # Red: FF (255) vs 00 (0) = 255 difference
  51. assert scheduler._colors_are_similar("#FF0000", "#00FF00") is False
  52. def test_colors_are_similar_none_colors(self, scheduler):
  53. """None colors should not be similar."""
  54. assert scheduler._colors_are_similar(None, "#FF5500") is False
  55. assert scheduler._colors_are_similar("#FF5500", None) is False
  56. class TestBuildLoadedFilaments:
  57. """Test the _build_loaded_filaments method."""
  58. @pytest.fixture
  59. def scheduler(self):
  60. return PrintScheduler()
  61. def test_build_loaded_filaments_empty_status(self, scheduler):
  62. """Empty status should return empty list."""
  63. class MockStatus:
  64. raw_data = {}
  65. result = scheduler._build_loaded_filaments(MockStatus())
  66. assert result == []
  67. def test_build_loaded_filaments_with_ams(self, scheduler):
  68. """Should extract filaments from AMS units."""
  69. class MockStatus:
  70. raw_data = {
  71. "ams": [
  72. {
  73. "id": 0,
  74. "tray": [
  75. {"id": 0, "tray_type": "PLA", "tray_color": "FF0000"},
  76. {"id": 1, "tray_type": "PETG", "tray_color": "00FF00"},
  77. ],
  78. }
  79. ]
  80. }
  81. result = scheduler._build_loaded_filaments(MockStatus())
  82. assert len(result) == 2
  83. # First filament
  84. assert result[0]["type"] == "PLA"
  85. assert result[0]["color"] == "#FF0000"
  86. assert result[0]["ams_id"] == 0
  87. assert result[0]["tray_id"] == 0
  88. assert result[0]["global_tray_id"] == 0 # 0 * 4 + 0
  89. # Second filament
  90. assert result[1]["type"] == "PETG"
  91. assert result[1]["global_tray_id"] == 1 # 0 * 4 + 1
  92. def test_build_loaded_filaments_with_ht_ams(self, scheduler):
  93. """AMS-HT (single tray) should be marked as is_ht."""
  94. class MockStatus:
  95. raw_data = {
  96. "ams": [
  97. {
  98. "id": 128,
  99. "tray": [{"id": 0, "tray_type": "PLA-CF", "tray_color": "000000"}],
  100. }
  101. ]
  102. }
  103. result = scheduler._build_loaded_filaments(MockStatus())
  104. assert len(result) == 1
  105. assert result[0]["is_ht"] is True
  106. assert result[0]["global_tray_id"] == 128 # AMS-HT uses ams_id directly
  107. def test_build_loaded_filaments_with_external(self, scheduler):
  108. """Should include external spool."""
  109. class MockStatus:
  110. raw_data = {"vt_tray": [{"tray_type": "TPU", "tray_color": "0000FF"}]}
  111. result = scheduler._build_loaded_filaments(MockStatus())
  112. assert len(result) == 1
  113. assert result[0]["type"] == "TPU"
  114. assert result[0]["is_external"] is True
  115. assert result[0]["global_tray_id"] == 254
  116. def test_build_loaded_filaments_skips_empty_trays(self, scheduler):
  117. """Trays without tray_type should be skipped."""
  118. class MockStatus:
  119. raw_data = {
  120. "ams": [
  121. {
  122. "id": 0,
  123. "tray": [
  124. {"id": 0, "tray_type": "PLA", "tray_color": "FF0000"},
  125. {"id": 1, "tray_type": "", "tray_color": ""}, # Empty
  126. {"id": 2}, # No tray_type key
  127. ],
  128. }
  129. ]
  130. }
  131. result = scheduler._build_loaded_filaments(MockStatus())
  132. assert len(result) == 1
  133. assert result[0]["type"] == "PLA"
  134. class TestMatchFilamentsToSlots:
  135. """Test the _match_filaments_to_slots method."""
  136. @pytest.fixture
  137. def scheduler(self):
  138. return PrintScheduler()
  139. def test_match_empty_required(self, scheduler):
  140. """Empty required list should return None."""
  141. result = scheduler._match_filaments_to_slots([], [])
  142. assert result is None
  143. def test_match_exact_color(self, scheduler):
  144. """Should prefer exact color match."""
  145. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  146. loaded = [
  147. {"type": "PLA", "color": "#00FF00", "global_tray_id": 0}, # Wrong color
  148. {"type": "PLA", "color": "#FF0000", "global_tray_id": 1}, # Exact match
  149. ]
  150. result = scheduler._match_filaments_to_slots(required, loaded)
  151. assert result == [1] # Should pick tray 1 (exact color match)
  152. def test_match_similar_color(self, scheduler):
  153. """Should match similar colors when no exact match."""
  154. required = [{"slot_id": 1, "type": "PLA", "color": "#FF5500"}]
  155. loaded = [
  156. {"type": "PLA", "color": "#FF5510", "global_tray_id": 0}, # Similar
  157. ]
  158. result = scheduler._match_filaments_to_slots(required, loaded)
  159. assert result == [0]
  160. def test_match_type_only(self, scheduler):
  161. """Should match by type when colors don't match."""
  162. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  163. loaded = [
  164. {"type": "PLA", "color": "#0000FF", "global_tray_id": 5}, # Type match, color way off
  165. ]
  166. result = scheduler._match_filaments_to_slots(required, loaded)
  167. assert result == [5]
  168. def test_match_no_match_returns_minus_one(self, scheduler):
  169. """Unmatched filaments should have -1 in mapping."""
  170. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  171. loaded = [
  172. {"type": "PETG", "color": "#FF0000", "global_tray_id": 0}, # Wrong type
  173. ]
  174. result = scheduler._match_filaments_to_slots(required, loaded)
  175. assert result == [-1]
  176. def test_match_multiple_filaments(self, scheduler):
  177. """Should match multiple filaments correctly."""
  178. required = [
  179. {"slot_id": 1, "type": "PLA", "color": "#FF0000"},
  180. {"slot_id": 2, "type": "PETG", "color": "#00FF00"},
  181. ]
  182. loaded = [
  183. {"type": "PLA", "color": "#FF0000", "global_tray_id": 0},
  184. {"type": "PETG", "color": "#00FF00", "global_tray_id": 1},
  185. ]
  186. result = scheduler._match_filaments_to_slots(required, loaded)
  187. assert result == [0, 1]
  188. def test_match_avoids_duplicate_assignment(self, scheduler):
  189. """Same tray should not be assigned to multiple slots."""
  190. required = [
  191. {"slot_id": 1, "type": "PLA", "color": "#FF0000"},
  192. {"slot_id": 2, "type": "PLA", "color": "#FF0000"}, # Same requirements
  193. ]
  194. loaded = [
  195. {"type": "PLA", "color": "#FF0000", "global_tray_id": 0}, # Only one PLA
  196. ]
  197. result = scheduler._match_filaments_to_slots(required, loaded)
  198. # First slot gets the match, second slot gets -1
  199. assert result == [0, -1]
  200. def test_match_h2d_pro_ams_ids(self, scheduler):
  201. """Should work with H2D Pro's high AMS IDs (128+)."""
  202. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  203. loaded = [
  204. {"type": "PLA", "color": "#FF0000", "global_tray_id": 512}, # AMS 128, slot 0
  205. ]
  206. result = scheduler._match_filaments_to_slots(required, loaded)
  207. assert result == [512]
  208. def test_match_external_spool(self, scheduler):
  209. """Should match external spool with ID 254."""
  210. required = [{"slot_id": 1, "type": "TPU", "color": "#0000FF"}]
  211. loaded = [
  212. {"type": "TPU", "color": "#0000FF", "global_tray_id": 254, "is_external": True},
  213. ]
  214. result = scheduler._match_filaments_to_slots(required, loaded)
  215. assert result == [254]
  216. def test_match_by_tray_info_idx_priority(self, scheduler):
  217. """tray_info_idx match should have highest priority over color match."""
  218. required = [{"slot_id": 1, "type": "PLA", "color": "#000000", "tray_info_idx": "GFA00"}]
  219. loaded = [
  220. {
  221. "type": "PLA",
  222. "color": "#000000",
  223. "global_tray_id": 0,
  224. "tray_info_idx": "GFB00",
  225. }, # Same color, different spool
  226. {
  227. "type": "PLA",
  228. "color": "#000000",
  229. "global_tray_id": 1,
  230. "tray_info_idx": "GFA00",
  231. }, # Same color, exact spool
  232. {
  233. "type": "PLA",
  234. "color": "#000000",
  235. "global_tray_id": 2,
  236. "tray_info_idx": "GFC00",
  237. }, # Same color, different spool
  238. ]
  239. result = scheduler._match_filaments_to_slots(required, loaded)
  240. assert result == [1] # Should pick tray 1 (exact tray_info_idx match)
  241. def test_match_by_tray_info_idx_with_different_colors(self, scheduler):
  242. """tray_info_idx match should work even if colors differ slightly."""
  243. required = [{"slot_id": 1, "type": "PLA", "color": "#000000", "tray_info_idx": "P4d64437"}]
  244. loaded = [
  245. {"type": "PLA", "color": "#000000", "global_tray_id": 0, "tray_info_idx": ""}, # No idx
  246. {
  247. "type": "PLA",
  248. "color": "#000010",
  249. "global_tray_id": 3,
  250. "tray_info_idx": "P4d64437",
  251. }, # Exact spool (slightly different color reported)
  252. ]
  253. result = scheduler._match_filaments_to_slots(required, loaded)
  254. assert result == [3] # Should pick tray 3 (exact tray_info_idx match)
  255. def test_match_fallback_to_color_when_no_tray_info_idx(self, scheduler):
  256. """Should fall back to color matching when tray_info_idx is empty."""
  257. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000", "tray_info_idx": ""}]
  258. loaded = [
  259. {"type": "PLA", "color": "#00FF00", "global_tray_id": 0, "tray_info_idx": "GFA00"},
  260. {"type": "PLA", "color": "#FF0000", "global_tray_id": 1, "tray_info_idx": "GFB00"}, # Color match
  261. ]
  262. result = scheduler._match_filaments_to_slots(required, loaded)
  263. assert result == [1] # Should pick tray 1 (color match)
  264. def test_match_fallback_to_color_when_no_matching_tray_info_idx(self, scheduler):
  265. """Should fall back to color when tray_info_idx doesn't match any loaded spool."""
  266. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000", "tray_info_idx": "OLD_SPOOL"}]
  267. loaded = [
  268. {
  269. "type": "PLA",
  270. "color": "#FF0000",
  271. "global_tray_id": 0,
  272. "tray_info_idx": "NEW_SPOOL",
  273. }, # Different idx but same color
  274. ]
  275. result = scheduler._match_filaments_to_slots(required, loaded)
  276. assert result == [0] # Should fall back to color match
  277. def test_match_multiple_same_color_with_tray_info_idx(self, scheduler):
  278. """Multiple identical filaments should be matched by tray_info_idx (H2D Pro scenario)."""
  279. # This is the exact scenario from issue #245 - 3 black PLA spools
  280. required = [
  281. {"slot_id": 1, "type": "PLA", "color": "#000000", "tray_info_idx": "GFA03"}, # Wants tray 3
  282. ]
  283. loaded = [
  284. {"type": "PLA", "color": "#000000", "global_tray_id": 0, "tray_info_idx": "GFA00"}, # Tray 0
  285. {"type": "PLA", "color": "#000000", "global_tray_id": 1, "tray_info_idx": "GFA01"}, # Tray 1
  286. {"type": "PLA", "color": "#000000", "global_tray_id": 2, "tray_info_idx": "GFA02"}, # Tray 2
  287. {
  288. "type": "PLA",
  289. "color": "#000000",
  290. "global_tray_id": 3,
  291. "tray_info_idx": "GFA03",
  292. }, # Tray 3 - the one we want
  293. ]
  294. result = scheduler._match_filaments_to_slots(required, loaded)
  295. assert result == [3] # Should pick tray 3, not tray 0
  296. def test_match_tray_info_idx_not_reused(self, scheduler):
  297. """tray_info_idx matched trays should not be reused for other slots."""
  298. required = [
  299. {"slot_id": 1, "type": "PLA", "color": "#000000", "tray_info_idx": "GFA00"},
  300. {"slot_id": 2, "type": "PLA", "color": "#000000", "tray_info_idx": "GFA01"},
  301. ]
  302. loaded = [
  303. {"type": "PLA", "color": "#000000", "global_tray_id": 0, "tray_info_idx": "GFA00"},
  304. {"type": "PLA", "color": "#000000", "global_tray_id": 1, "tray_info_idx": "GFA01"},
  305. ]
  306. result = scheduler._match_filaments_to_slots(required, loaded)
  307. assert result == [0, 1] # Each slot gets its specific tray
  308. def test_match_non_unique_tray_info_idx_uses_color(self, scheduler):
  309. """Non-unique tray_info_idx should fall back to color matching.
  310. This is the scenario where multiple trays have the same tray_info_idx
  311. (e.g., two spools of generic PLA both have GFA00). The color should
  312. be used as tiebreaker instead of just picking the first match.
  313. """
  314. # User sliced with green PLA (tray_info_idx=GFA00)
  315. # Two trays have GFA00: tray 3 (white) and tray 4 (green)
  316. # Should pick tray 4 because the color matches
  317. required = [
  318. {"slot_id": 2, "type": "PLA", "color": "#00FF00", "tray_info_idx": "GFA00"}, # Green PLA
  319. ]
  320. loaded = [
  321. {"type": "PLA", "color": "#FFFFFF", "global_tray_id": 3, "tray_info_idx": "GFA00"}, # White PLA
  322. {"type": "PLA", "color": "#00FF00", "global_tray_id": 4, "tray_info_idx": "GFA00"}, # Green PLA
  323. ]
  324. result = scheduler._match_filaments_to_slots(required, loaded)
  325. assert result == [-1, 4] # Should pick tray 4 (color match), not tray 3 (first match)
  326. def test_match_non_unique_tray_info_idx_same_color(self, scheduler):
  327. """Non-unique tray_info_idx with identical colors picks first match.
  328. When multiple trays have the same tray_info_idx AND same color,
  329. there's no way to differentiate, so first match is used.
  330. """
  331. required = [
  332. {"slot_id": 2, "type": "PLA", "color": "#FFFFFF", "tray_info_idx": "GFA00"},
  333. ]
  334. loaded = [
  335. {"type": "PLA", "color": "#FFFFFF", "global_tray_id": 3, "tray_info_idx": "GFA00"},
  336. {"type": "PLA", "color": "#FFFFFF", "global_tray_id": 4, "tray_info_idx": "GFA00"},
  337. ]
  338. result = scheduler._match_filaments_to_slots(required, loaded)
  339. # Both have same color, so first is used
  340. assert result == [-1, 3]
  341. class TestBuildLoadedFilamentsTrayInfoIdx:
  342. """Test tray_info_idx extraction in _build_loaded_filaments."""
  343. @pytest.fixture
  344. def scheduler(self):
  345. return PrintScheduler()
  346. def test_build_loaded_filaments_includes_tray_info_idx(self, scheduler):
  347. """Should extract tray_info_idx from AMS trays."""
  348. class MockStatus:
  349. raw_data = {
  350. "ams": [
  351. {
  352. "id": 0,
  353. "tray": [
  354. {"id": 0, "tray_type": "PLA", "tray_color": "000000", "tray_info_idx": "GFA00"},
  355. {"id": 1, "tray_type": "PLA", "tray_color": "000000", "tray_info_idx": "GFA01"},
  356. ],
  357. }
  358. ]
  359. }
  360. result = scheduler._build_loaded_filaments(MockStatus())
  361. assert len(result) == 2
  362. assert result[0]["tray_info_idx"] == "GFA00"
  363. assert result[1]["tray_info_idx"] == "GFA01"
  364. def test_build_loaded_filaments_empty_tray_info_idx(self, scheduler):
  365. """Missing tray_info_idx should default to empty string."""
  366. class MockStatus:
  367. raw_data = {
  368. "ams": [
  369. {
  370. "id": 0,
  371. "tray": [
  372. {"id": 0, "tray_type": "PLA", "tray_color": "FF0000"}, # No tray_info_idx
  373. ],
  374. }
  375. ]
  376. }
  377. result = scheduler._build_loaded_filaments(MockStatus())
  378. assert len(result) == 1
  379. assert result[0]["tray_info_idx"] == ""
  380. def test_build_loaded_filaments_external_spool_tray_info_idx(self, scheduler):
  381. """Should extract tray_info_idx from external spool."""
  382. class MockStatus:
  383. raw_data = {"vt_tray": [{"tray_type": "TPU", "tray_color": "0000FF", "tray_info_idx": "P4d64437"}]}
  384. result = scheduler._build_loaded_filaments(MockStatus())
  385. assert len(result) == 1
  386. assert result[0]["tray_info_idx"] == "P4d64437"
  387. assert result[0]["is_external"] is True
  388. def _make_3mf_zip(
  389. project_settings: dict | None = None,
  390. slice_info_xml: str | None = None,
  391. ) -> zipfile.ZipFile:
  392. """Create an in-memory ZipFile mimicking a 3MF with project_settings.config."""
  393. buf = io.BytesIO()
  394. with zipfile.ZipFile(buf, "w") as zf:
  395. if project_settings is not None:
  396. zf.writestr("Metadata/project_settings.config", json.dumps(project_settings))
  397. if slice_info_xml is not None:
  398. zf.writestr("Metadata/slice_info.config", slice_info_xml)
  399. buf.seek(0)
  400. return zipfile.ZipFile(buf, "r")
  401. class TestExtractNozzleMappingFrom3mf:
  402. """Test the extract_nozzle_mapping_from_3mf utility."""
  403. def test_group_id_priority_over_filament_nozzle_map(self):
  404. """group_id from slice_info should override filament_nozzle_map from project_settings.
  405. Real-world scenario: "Auto For Flush" mode sets filament_nozzle_map all to 0
  406. (user preference) but the actual assignment in slice_info has different group_ids.
  407. """
  408. # filament_nozzle_map says all on slicer ext 0 → MQTT ext 1 (LEFT)
  409. # But slice_info group_id says slot 6 → group 0 (LEFT), slot 12 → group 1 (RIGHT)
  410. slice_info = """<?xml version="1.0" encoding="UTF-8"?>
  411. <config>
  412. <plate>
  413. <filament id="6" type="PLA" color="#56B7E6" used_g="1.84" group_id="0"/>
  414. <filament id="12" type="PLA" color="#B39B84" used_g="1.76" group_id="1"/>
  415. </plate>
  416. </config>"""
  417. zf = _make_3mf_zip(
  418. {
  419. "filament_nozzle_map": ["0"] * 12,
  420. "physical_extruder_map": ["1", "0"],
  421. },
  422. slice_info_xml=slice_info,
  423. )
  424. result = extract_nozzle_mapping_from_3mf(zf)
  425. # group_id 0 → physical_extruder_map[0] = 1 (LEFT)
  426. # group_id 1 → physical_extruder_map[1] = 0 (RIGHT)
  427. assert result == {6: 1, 12: 0}
  428. zf.close()
  429. def test_fallback_to_filament_nozzle_map_without_group_id(self):
  430. """Should fall back to filament_nozzle_map when slice_info has no group_id."""
  431. slice_info = """<?xml version="1.0" encoding="UTF-8"?>
  432. <config>
  433. <plate>
  434. <filament id="1" type="PLA" color="#FF0000" used_g="5.0"/>
  435. </plate>
  436. </config>"""
  437. zf = _make_3mf_zip(
  438. {
  439. "filament_nozzle_map": ["0", "1", "0"],
  440. "physical_extruder_map": ["0", "1"],
  441. },
  442. slice_info_xml=slice_info,
  443. )
  444. result = extract_nozzle_mapping_from_3mf(zf)
  445. assert result == {1: 0, 2: 1, 3: 0}
  446. zf.close()
  447. def test_fallback_to_filament_nozzle_map_without_slice_info(self):
  448. """Should fall back to filament_nozzle_map when no slice_info.config exists."""
  449. zf = _make_3mf_zip(
  450. {
  451. "filament_nozzle_map": ["0", "1", "0"],
  452. "physical_extruder_map": ["0", "1"],
  453. }
  454. )
  455. result = extract_nozzle_mapping_from_3mf(zf)
  456. assert result == {1: 0, 2: 1, 3: 0}
  457. zf.close()
  458. def test_single_nozzle_returns_none(self):
  459. """Single physical_extruder_map entry should return None (single-nozzle)."""
  460. zf = _make_3mf_zip(
  461. {
  462. "filament_nozzle_map": ["0", "0", "0"],
  463. "physical_extruder_map": ["0"],
  464. }
  465. )
  466. result = extract_nozzle_mapping_from_3mf(zf)
  467. assert result is None
  468. zf.close()
  469. def test_missing_project_settings_returns_none(self):
  470. """Missing project_settings.config should return None."""
  471. zf = _make_3mf_zip(None)
  472. result = extract_nozzle_mapping_from_3mf(zf)
  473. assert result is None
  474. zf.close()
  475. def test_missing_fields_returns_none(self):
  476. """Missing physical_extruder_map should return None."""
  477. zf = _make_3mf_zip({"some_other_key": "value"})
  478. result = extract_nozzle_mapping_from_3mf(zf)
  479. assert result is None
  480. zf.close()
  481. def test_physical_extruder_map_remapping(self):
  482. """Should apply physical_extruder_map to remap slicer extruder to MQTT extruder."""
  483. # Slicer ext 0 -> MQTT ext 1, slicer ext 1 -> MQTT ext 0
  484. zf = _make_3mf_zip(
  485. {
  486. "filament_nozzle_map": ["0", "1"],
  487. "physical_extruder_map": ["1", "0"],
  488. }
  489. )
  490. result = extract_nozzle_mapping_from_3mf(zf)
  491. assert result == {1: 1, 2: 0}
  492. zf.close()
  493. class TestNozzleAwareMapping:
  494. """Test nozzle-aware filament matching in the print scheduler."""
  495. @pytest.fixture
  496. def scheduler(self):
  497. return PrintScheduler()
  498. def test_dual_nozzle_matching(self, scheduler):
  499. """Filaments assigned to different nozzles should match to correct AMS units."""
  500. required = [
  501. {"slot_id": 1, "type": "PLA", "color": "#FF0000", "nozzle_id": 0}, # Right nozzle
  502. {"slot_id": 2, "type": "PLA", "color": "#00FF00", "nozzle_id": 1}, # Left nozzle
  503. ]
  504. loaded = [
  505. {"type": "PLA", "color": "#00FF00", "global_tray_id": 0, "extruder_id": 0}, # AMS0 on right
  506. {"type": "PLA", "color": "#FF0000", "global_tray_id": 4, "extruder_id": 1}, # AMS1 on left
  507. ]
  508. # Without nozzle filtering, slot 1 (red, right) would match tray 4 (red, left) by color.
  509. # With nozzle filtering, slot 1 (right nozzle) can only use tray 0 (right extruder),
  510. # and slot 2 (left nozzle) can only use tray 4 (left extruder).
  511. result = scheduler._match_filaments_to_slots(required, loaded)
  512. assert result == [0, 4]
  513. def test_nozzle_hard_filter_no_fallback(self, scheduler):
  514. """Hard filter: no fallback to wrong nozzle when target nozzle has no trays."""
  515. required = [
  516. {"slot_id": 1, "type": "PLA", "color": "#FF0000", "nozzle_id": 0}, # Right nozzle
  517. ]
  518. loaded = [
  519. # Only a tray on the left nozzle, none on right
  520. {"type": "PLA", "color": "#FF0000", "global_tray_id": 4, "extruder_id": 1},
  521. ]
  522. # No trays on extruder 0 — hard filter returns -1, no cross-nozzle fallback
  523. result = scheduler._match_filaments_to_slots(required, loaded)
  524. assert result == [-1]
  525. def test_no_nozzle_id_skips_filtering(self, scheduler):
  526. """When nozzle_id is None, no nozzle filtering should be applied."""
  527. required = [
  528. {"slot_id": 1, "type": "PLA", "color": "#FF0000"}, # No nozzle_id
  529. ]
  530. loaded = [
  531. {"type": "PLA", "color": "#FF0000", "global_tray_id": 0, "extruder_id": 0},
  532. {"type": "PLA", "color": "#FF0000", "global_tray_id": 4, "extruder_id": 1},
  533. ]
  534. # Should match first available (tray 0) regardless of extruder
  535. result = scheduler._match_filaments_to_slots(required, loaded)
  536. assert result == [0]
  537. def test_extruder_id_in_loaded_filaments(self, scheduler):
  538. """_build_loaded_filaments should include extruder_id from ams_extruder_map."""
  539. class MockStatus:
  540. raw_data = {
  541. "ams": [
  542. {"id": 0, "tray": [{"id": 0, "tray_type": "PLA", "tray_color": "FF0000"}]},
  543. {"id": 1, "tray": [{"id": 0, "tray_type": "PLA", "tray_color": "00FF00"}]},
  544. ],
  545. "ams_extruder_map": {"0": 0, "1": 1},
  546. }
  547. result = scheduler._build_loaded_filaments(MockStatus())
  548. assert len(result) == 2
  549. assert result[0]["extruder_id"] == 0
  550. assert result[1]["extruder_id"] == 1
  551. def test_extruder_id_none_without_map(self, scheduler):
  552. """extruder_id should be None when ams_extruder_map is absent."""
  553. class MockStatus:
  554. raw_data = {
  555. "ams": [
  556. {"id": 0, "tray": [{"id": 0, "tray_type": "PLA", "tray_color": "FF0000"}]},
  557. ]
  558. }
  559. result = scheduler._build_loaded_filaments(MockStatus())
  560. assert len(result) == 1
  561. assert result[0]["extruder_id"] is None
  562. def test_external_spool_extruder_id(self, scheduler):
  563. """External spool 254 (Ext-L) should have extruder_id=1 (LEFT) when ams_extruder_map exists."""
  564. class MockStatus:
  565. raw_data = {
  566. "vt_tray": [{"tray_type": "TPU", "tray_color": "0000FF"}],
  567. "ams_extruder_map": {"0": 0},
  568. }
  569. result = scheduler._build_loaded_filaments(MockStatus())
  570. assert len(result) == 1
  571. # Default vt_tray id=254 → Ext-L → LEFT nozzle (extruder 1)
  572. assert result[0]["extruder_id"] == 1
  573. assert result[0]["is_external"] is True
  574. def test_external_spool_no_extruder_map(self, scheduler):
  575. """External spool extruder_id should be None without ams_extruder_map."""
  576. class MockStatus:
  577. raw_data = {"vt_tray": [{"tray_type": "TPU", "tray_color": "0000FF"}]}
  578. result = scheduler._build_loaded_filaments(MockStatus())
  579. assert len(result) == 1
  580. assert result[0]["extruder_id"] is None
  581. def test_dual_nozzle_with_tray_info_idx(self, scheduler):
  582. """Nozzle filtering should work together with tray_info_idx matching."""
  583. required = [
  584. {"slot_id": 1, "type": "PLA", "color": "#000000", "tray_info_idx": "GFA00", "nozzle_id": 0},
  585. {"slot_id": 2, "type": "PLA", "color": "#000000", "tray_info_idx": "GFA01", "nozzle_id": 1},
  586. ]
  587. loaded = [
  588. {"type": "PLA", "color": "#000000", "global_tray_id": 0, "tray_info_idx": "GFA00", "extruder_id": 0},
  589. {"type": "PLA", "color": "#000000", "global_tray_id": 4, "tray_info_idx": "GFA01", "extruder_id": 1},
  590. ]
  591. result = scheduler._match_filaments_to_slots(required, loaded)
  592. assert result == [0, 4]
  593. # ============================================================================
  594. # MODEL-SPECIFIC TESTS: Real data from actual printers
  595. # ============================================================================
  596. def _h2d_raw_data():
  597. """H2D real data fixture (from live API response 2026-02-18).
  598. Configuration:
  599. LEFT nozzle (extruder 1): AMS 0 (4-slot), AMS 2 (4-slot)
  600. RIGHT nozzle (extruder 0): AMS 1 (4-slot), AMS-HT 128 (1-slot, empty)
  601. External: 254 (Ext-L, LEFT), 255 (Ext-R, RIGHT, empty)
  602. ams_extruder_map: {"0": 1, "1": 0, "2": 1, "128": 0}
  603. """
  604. return {
  605. "ams": [
  606. {
  607. "id": 0,
  608. "tray": [
  609. {"id": 0, "tray_type": "PETG", "tray_color": "FFFFFFFF", "tray_info_idx": "GFG02"},
  610. {"id": 1, "tray_type": "PLA", "tray_color": "C8C8C8FF", "tray_info_idx": "GFA06"},
  611. {"id": 2, "tray_type": "PETG", "tray_color": "875718FF", "tray_info_idx": "GFG02"},
  612. {"id": 3, "tray_type": "PLA", "tray_color": "000000FF", "tray_info_idx": "GFA00"},
  613. ],
  614. },
  615. {
  616. "id": 1,
  617. "tray": [
  618. {"id": 0, "tray_type": "PLA", "tray_color": "FFFFFFFF", "tray_info_idx": "GFA00"},
  619. {"id": 1, "tray_type": "PETG", "tray_color": "000000FF", "tray_info_idx": "GFG02"},
  620. {"id": 2, "tray_type": "PLA", "tray_color": "5F6367FF", "tray_info_idx": "GFA06"},
  621. {"id": 3, "tray_type": "PLA", "tray_color": "B39B84FF", "tray_info_idx": "GFA02"},
  622. ],
  623. },
  624. {
  625. "id": 128,
  626. "tray": [{"id": 0}], # AMS-HT, empty
  627. },
  628. {
  629. "id": 2,
  630. "tray": [
  631. {"id": 0, "tray_type": "PLA-S", "tray_color": "FFFFFFFF", "tray_info_idx": "P8aa1726"},
  632. {"id": 1, "tray_type": "PLA", "tray_color": "56B7E6FF", "tray_info_idx": "PFUS9924"},
  633. {"id": 2, "tray_type": "PETG", "tray_color": "6EE53CFF", "tray_info_idx": "GFG02"},
  634. {"id": 3, "tray_type": "PLA", "tray_color": "FF0000FF", "tray_info_idx": "PFUS9ac9"},
  635. ],
  636. },
  637. ],
  638. "vt_tray": [
  639. {"id": 254, "tray_type": "PLA", "tray_color": "000000FF", "tray_info_idx": "P4d64437"},
  640. {"id": 255, "tray_type": "", "tray_color": "00000000"}, # empty
  641. ],
  642. "ams_extruder_map": {"0": 1, "1": 0, "2": 1, "128": 0},
  643. }
  644. def _x1c_raw_data():
  645. """X1C real data fixture (from live API response 2026-02-18).
  646. Configuration:
  647. Single nozzle (extruder 0): AMS 0 (4-slot, all empty), AMS 1 (4-slot, 3 loaded)
  648. External: 254 (single, empty)
  649. ams_extruder_map: {"0": 0, "1": 0} ← NOT empty, all on extruder 0
  650. """
  651. return {
  652. "ams": [
  653. {
  654. "id": 0,
  655. "tray": [
  656. {"id": 0}, # empty
  657. {"id": 1}, # empty
  658. {"id": 2}, # empty
  659. {"id": 3}, # empty
  660. ],
  661. },
  662. {
  663. "id": 1,
  664. "tray": [
  665. {"id": 0}, # empty
  666. {"id": 1, "tray_type": "PLA", "tray_color": "EBCFA6FF", "tray_info_idx": "PFUS22b2"},
  667. {"id": 2, "tray_type": "PLA", "tray_color": "FCECD6FF", "tray_info_idx": "P4d64437"},
  668. {"id": 3, "tray_type": "PLA", "tray_color": "0066FFFF", "tray_info_idx": "P4d64437"},
  669. ],
  670. },
  671. ],
  672. "vt_tray": [
  673. {"id": 254, "tray_type": "", "tray_color": "00000000"}, # empty
  674. ],
  675. "ams_extruder_map": {"0": 0, "1": 0},
  676. }
  677. class TestH2DModel:
  678. """H2D-specific tests with real printer data (dual nozzle, AMS-HT)."""
  679. @pytest.fixture
  680. def scheduler(self):
  681. return PrintScheduler()
  682. def test_build_loaded_filaments_h2d(self, scheduler):
  683. """H2D: correct extruder_id, global_tray_id, AMS-HT handling."""
  684. class MockStatus:
  685. raw_data = _h2d_raw_data()
  686. result = scheduler._build_loaded_filaments(MockStatus())
  687. # Should have 13 loaded filaments (4 + 4 + 0 + 4 + 1 external)
  688. assert len(result) == 13
  689. # AMS 0 trays → extruder 1 (LEFT)
  690. ams0 = [f for f in result if f["ams_id"] == 0]
  691. assert len(ams0) == 4
  692. assert all(f["extruder_id"] == 1 for f in ams0)
  693. assert [f["global_tray_id"] for f in ams0] == [0, 1, 2, 3]
  694. # AMS 1 trays → extruder 0 (RIGHT)
  695. ams1 = [f for f in result if f["ams_id"] == 1]
  696. assert len(ams1) == 4
  697. assert all(f["extruder_id"] == 0 for f in ams1)
  698. assert [f["global_tray_id"] for f in ams1] == [4, 5, 6, 7]
  699. # AMS-HT 128 → empty, should not appear
  700. ams_ht = [f for f in result if f["ams_id"] == 128]
  701. assert len(ams_ht) == 0
  702. # AMS 2 trays → extruder 1 (LEFT)
  703. ams2 = [f for f in result if f["ams_id"] == 2]
  704. assert len(ams2) == 4
  705. assert all(f["extruder_id"] == 1 for f in ams2)
  706. assert [f["global_tray_id"] for f in ams2] == [8, 9, 10, 11]
  707. def test_external_spool_extruder_h2d(self, scheduler):
  708. """H2D: Ext-L (254) = LEFT (extruder 1), Ext-R (255) = RIGHT (extruder 0)."""
  709. class MockStatus:
  710. raw_data = _h2d_raw_data()
  711. result = scheduler._build_loaded_filaments(MockStatus())
  712. ext = [f for f in result if f["is_external"]]
  713. assert len(ext) == 1 # Only 254 has filament
  714. assert ext[0]["global_tray_id"] == 254
  715. # Ext-L (254) should be LEFT nozzle (extruder 1)
  716. assert ext[0]["extruder_id"] == 1
  717. def test_match_left_nozzle_only(self, scheduler):
  718. """H2D: left-nozzle requirement only matches left-nozzle AMS."""
  719. class MockStatus:
  720. raw_data = _h2d_raw_data()
  721. loaded = scheduler._build_loaded_filaments(MockStatus())
  722. required = [
  723. {"slot_id": 1, "type": "PLA", "color": "#000000", "nozzle_id": 1}, # LEFT
  724. ]
  725. result = scheduler._match_filaments_to_slots(required, loaded)
  726. # Black PLA on LEFT: AMS 0 T4 (global 3)
  727. assert result == [3]
  728. def test_match_right_nozzle_only(self, scheduler):
  729. """H2D: right-nozzle requirement only matches right-nozzle AMS."""
  730. class MockStatus:
  731. raw_data = _h2d_raw_data()
  732. loaded = scheduler._build_loaded_filaments(MockStatus())
  733. required = [
  734. {"slot_id": 1, "type": "PLA", "color": "#FFFFFF", "nozzle_id": 0}, # RIGHT
  735. ]
  736. result = scheduler._match_filaments_to_slots(required, loaded)
  737. # White PLA on RIGHT: AMS 1 T1 (global 4)
  738. assert result == [4]
  739. def test_reject_cross_nozzle(self, scheduler):
  740. """H2D: hard filter rejects cross-nozzle assignment."""
  741. class MockStatus:
  742. raw_data = _h2d_raw_data()
  743. loaded = scheduler._build_loaded_filaments(MockStatus())
  744. # PLA-S only exists on AMS 2 T1 (LEFT), require on RIGHT
  745. required = [
  746. {"slot_id": 1, "type": "PLA-S", "color": "#FFFFFF", "nozzle_id": 0},
  747. ]
  748. result = scheduler._match_filaments_to_slots(required, loaded)
  749. assert result == [-1] # No fallback to wrong nozzle
  750. def test_dual_nozzle_multi_filament(self, scheduler):
  751. """H2D: multi-filament print maps to correct nozzles."""
  752. class MockStatus:
  753. raw_data = _h2d_raw_data()
  754. loaded = scheduler._build_loaded_filaments(MockStatus())
  755. required = [
  756. {"slot_id": 1, "type": "PETG", "color": "#FFFFFF", "nozzle_id": 1, "tray_info_idx": "GFG02"},
  757. {"slot_id": 2, "type": "PLA", "color": "#FFFFFF", "nozzle_id": 0, "tray_info_idx": "GFA00"},
  758. ]
  759. result = scheduler._match_filaments_to_slots(required, loaded)
  760. # PETG white on LEFT: AMS 0 T1 (global 0)
  761. # PLA white on RIGHT: AMS 1 T1 (global 4)
  762. assert result == [0, 4]
  763. def test_external_spool_matches_on_correct_nozzle(self, scheduler):
  764. """H2D: external spool on left nozzle matches left-nozzle requirement."""
  765. class MockStatus:
  766. raw_data = _h2d_raw_data()
  767. loaded = scheduler._build_loaded_filaments(MockStatus())
  768. required = [
  769. {"slot_id": 1, "type": "PLA", "color": "#000000", "nozzle_id": 1, "tray_info_idx": "P4d64437"},
  770. ]
  771. result = scheduler._match_filaments_to_slots(required, loaded)
  772. assert result == [254] # External spool on left nozzle
  773. class TestX1CModel:
  774. """X1C-specific tests with real printer data (single nozzle, 2x regular AMS)."""
  775. @pytest.fixture
  776. def scheduler(self):
  777. return PrintScheduler()
  778. def test_build_loaded_filaments_x1c(self, scheduler):
  779. """X1C: all filaments on extruder 0, correct global_tray_id."""
  780. class MockStatus:
  781. raw_data = _x1c_raw_data()
  782. result = scheduler._build_loaded_filaments(MockStatus())
  783. # Only 3 loaded (AMS 1 trays 1-3)
  784. assert len(result) == 3
  785. # All on extruder 0
  786. assert all(f["extruder_id"] == 0 for f in result)
  787. # Correct global tray IDs
  788. assert [f["global_tray_id"] for f in result] == [5, 6, 7]
  789. def test_single_nozzle_no_filtering(self, scheduler):
  790. """X1C: single-nozzle 3MF has no nozzle_id, all trays available."""
  791. class MockStatus:
  792. raw_data = _x1c_raw_data()
  793. loaded = scheduler._build_loaded_filaments(MockStatus())
  794. required = [
  795. {"slot_id": 1, "type": "PLA", "color": "#0066FF"}, # No nozzle_id
  796. ]
  797. result = scheduler._match_filaments_to_slots(required, loaded)
  798. # Blue PLA → AMS 1 T4 (global 7)
  799. assert result == [7]
  800. def test_tray_info_idx_matching_x1c(self, scheduler):
  801. """X1C: tray_info_idx matching works across AMS units."""
  802. class MockStatus:
  803. raw_data = _x1c_raw_data()
  804. loaded = scheduler._build_loaded_filaments(MockStatus())
  805. required = [
  806. {"slot_id": 1, "type": "PLA", "color": "#EBCFA6", "tray_info_idx": "PFUS22b2"},
  807. ]
  808. result = scheduler._match_filaments_to_slots(required, loaded)
  809. # Unique tray_info_idx → AMS 1 T2 (global 5)
  810. assert result == [5]
  811. def test_non_unique_tray_info_idx_color_match_x1c(self, scheduler):
  812. """X1C: non-unique tray_info_idx falls back to color matching."""
  813. class MockStatus:
  814. raw_data = _x1c_raw_data()
  815. loaded = scheduler._build_loaded_filaments(MockStatus())
  816. # P4d64437 appears in AMS 1 T3 and T4
  817. required = [
  818. {"slot_id": 1, "type": "PLA", "color": "#FCECD6", "tray_info_idx": "P4d64437"},
  819. ]
  820. result = scheduler._match_filaments_to_slots(required, loaded)
  821. # Should pick AMS 1 T3 (global 6, color FCECD6) over T4 (0066FF)
  822. assert result == [6]
  823. def test_multi_filament_x1c(self, scheduler):
  824. """X1C: multi-filament print matches freely across AMS units."""
  825. class MockStatus:
  826. raw_data = _x1c_raw_data()
  827. loaded = scheduler._build_loaded_filaments(MockStatus())
  828. required = [
  829. {"slot_id": 1, "type": "PLA", "color": "#EBCFA6"},
  830. {"slot_id": 2, "type": "PLA", "color": "#0066FF"},
  831. ]
  832. result = scheduler._match_filaments_to_slots(required, loaded)
  833. assert result == [5, 7]