test_scheduler_ams_mapping.py 57 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448
  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 TestPreferLowestFilament:
  342. """Test prefer_lowest_filament sorting in _match_filaments_to_slots."""
  343. @pytest.fixture
  344. def scheduler(self):
  345. return PrintScheduler()
  346. def test_prefer_lowest_picks_lower_remain(self, scheduler):
  347. """When enabled, should pick the spool with lower remaining filament."""
  348. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  349. loaded = [
  350. {"type": "PLA", "color": "#FF0000", "global_tray_id": 0, "remain": 80},
  351. {"type": "PLA", "color": "#FF0000", "global_tray_id": 1, "remain": 30},
  352. ]
  353. result = scheduler._match_filaments_to_slots(required, loaded, prefer_lowest=True)
  354. assert result == [1] # Should pick tray 1 (30% remaining)
  355. def test_prefer_lowest_disabled_picks_first(self, scheduler):
  356. """When disabled, should pick the first matching spool (default behavior)."""
  357. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  358. loaded = [
  359. {"type": "PLA", "color": "#FF0000", "global_tray_id": 0, "remain": 80},
  360. {"type": "PLA", "color": "#FF0000", "global_tray_id": 1, "remain": 30},
  361. ]
  362. result = scheduler._match_filaments_to_slots(required, loaded, prefer_lowest=False)
  363. assert result == [0] # Should pick tray 0 (first match)
  364. def test_prefer_lowest_unknown_remain_sorted_last(self, scheduler):
  365. """Spools with remain=-1 (unknown) should be sorted to end."""
  366. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  367. loaded = [
  368. {"type": "PLA", "color": "#FF0000", "global_tray_id": 0, "remain": -1},
  369. {"type": "PLA", "color": "#FF0000", "global_tray_id": 1, "remain": 50},
  370. ]
  371. result = scheduler._match_filaments_to_slots(required, loaded, prefer_lowest=True)
  372. assert result == [1] # Should pick tray 1 (known 50%) over unknown
  373. def test_prefer_lowest_missing_remain_sorted_last(self, scheduler):
  374. """Spools without remain field should be sorted to end."""
  375. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  376. loaded = [
  377. {"type": "PLA", "color": "#FF0000", "global_tray_id": 0}, # No remain field
  378. {"type": "PLA", "color": "#FF0000", "global_tray_id": 1, "remain": 50},
  379. ]
  380. result = scheduler._match_filaments_to_slots(required, loaded, prefer_lowest=True)
  381. assert result == [1] # Should pick tray 1 (known 50%) over missing
  382. def test_prefer_lowest_multiple_slots(self, scheduler):
  383. """Should pick lowest remain for each slot independently."""
  384. required = [
  385. {"slot_id": 1, "type": "PLA", "color": "#FF0000"},
  386. {"slot_id": 2, "type": "PLA", "color": "#FF0000"},
  387. ]
  388. loaded = [
  389. {"type": "PLA", "color": "#FF0000", "global_tray_id": 0, "remain": 80},
  390. {"type": "PLA", "color": "#FF0000", "global_tray_id": 1, "remain": 30},
  391. {"type": "PLA", "color": "#FF0000", "global_tray_id": 2, "remain": 60},
  392. ]
  393. result = scheduler._match_filaments_to_slots(required, loaded, prefer_lowest=True)
  394. # Slot 1 gets tray 1 (30%), slot 2 gets tray 2 (60%) — tray 0 (80%) unused
  395. assert result == [1, 2]
  396. def test_prefer_lowest_with_tray_info_idx(self, scheduler):
  397. """Should sort within tray_info_idx subset too."""
  398. required = [{"slot_id": 1, "type": "PLA", "color": "#FFFFFF", "tray_info_idx": "GFA00"}]
  399. loaded = [
  400. {"type": "PLA", "color": "#FFFFFF", "global_tray_id": 0, "tray_info_idx": "GFA00", "remain": 80},
  401. {"type": "PLA", "color": "#FFFFFF", "global_tray_id": 1, "tray_info_idx": "GFA00", "remain": 20},
  402. ]
  403. result = scheduler._match_filaments_to_slots(required, loaded, prefer_lowest=True)
  404. assert result == [1] # Should pick tray 1 (20%) within idx subset
  405. def test_prefer_lowest_external_spool(self, scheduler):
  406. """External spool with low remain should be preferred over AMS spool."""
  407. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  408. loaded = [
  409. {"type": "PLA", "color": "#FF0000", "global_tray_id": 0, "remain": 80, "is_external": False},
  410. {"type": "PLA", "color": "#FF0000", "global_tray_id": 254, "remain": 10, "is_external": True},
  411. ]
  412. result = scheduler._match_filaments_to_slots(required, loaded, prefer_lowest=True)
  413. assert result == [254] # Should pick external spool (10%) over AMS (80%)
  414. class TestPreferLowestInventoryOverride:
  415. """Tests for the #1508 inventory-aware sort: when the user has bound a
  416. Bambuddy inventory spool to an AMS slot, that spool's remaining weight
  417. becomes the sort signal instead of the MQTT ``remain`` percentage.
  418. The fix is two-tier: inventory-tracked spools always sort before
  419. MQTT-only ones, then ascending by remaining within each tier, then
  420. ascending by AMS slot position. See
  421. ``print_scheduler._prefer_lowest_sort_key`` for the rationale.
  422. """
  423. @pytest.fixture
  424. def scheduler(self):
  425. return PrintScheduler()
  426. def test_inventory_override_beats_mqtt_remain(self, scheduler):
  427. """Slot 4's inventory shows 50 g remaining; slot 1's clone has 950 g.
  428. MQTT ``remain`` is -1 for both (non-RFID spools), so without the
  429. override the sort collapses to AMS-slot order and slot 1 wins.
  430. With the override slot 4 (the original, nearly empty) wins. This is
  431. the literal reporter scenario in #1508.
  432. """
  433. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  434. loaded = [
  435. {
  436. "type": "PLA",
  437. "color": "#FF0000",
  438. "ams_id": 0,
  439. "tray_id": 0,
  440. "global_tray_id": 0,
  441. "remain": -1,
  442. },
  443. {
  444. "type": "PLA",
  445. "color": "#FF0000",
  446. "ams_id": 0,
  447. "tray_id": 3,
  448. "global_tray_id": 3,
  449. "remain": -1,
  450. },
  451. ]
  452. # Slot 1 (gtid 0) is the fresh clone at 950 g; slot 4 (gtid 3) is the
  453. # nearly-empty original at 50 g.
  454. overrides = {0: 950.0, 3: 50.0}
  455. result = scheduler._match_filaments_to_slots(
  456. required, loaded, prefer_lowest=True, inventory_remain_overrides=overrides
  457. )
  458. assert result == [3]
  459. def test_inventory_override_with_zero_grams_still_wins(self, scheduler):
  460. """An inventory-tracked spool at 0 g must still sort first within
  461. its tier — the user wants to finish what's left (or be told there's
  462. a deficit) rather than skip to the fresh one. The clamp-to-101
  463. legacy logic only fires on negative values, so 0 g stays 0.0.
  464. """
  465. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  466. loaded = [
  467. {
  468. "type": "PLA",
  469. "color": "#FF0000",
  470. "ams_id": 0,
  471. "tray_id": 0,
  472. "global_tray_id": 0,
  473. "remain": -1,
  474. },
  475. {
  476. "type": "PLA",
  477. "color": "#FF0000",
  478. "ams_id": 0,
  479. "tray_id": 1,
  480. "global_tray_id": 1,
  481. "remain": -1,
  482. },
  483. ]
  484. overrides = {0: 500.0, 1: 0.0}
  485. result = scheduler._match_filaments_to_slots(
  486. required, loaded, prefer_lowest=True, inventory_remain_overrides=overrides
  487. )
  488. assert result == [1]
  489. def test_inventory_tier_beats_mqtt_tier_regardless_of_value(self, scheduler):
  490. """Mixed mode: one slot is inventory-tracked at 800 g (high), the
  491. other has only a MQTT remain of 10 (very low). The inventory tier
  492. wins because the user has explicitly told us to manage that slot —
  493. unit-mixing across tiers is intentional and resolved by the tier
  494. flag, not value comparison.
  495. """
  496. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  497. loaded = [
  498. {
  499. "type": "PLA",
  500. "color": "#FF0000",
  501. "ams_id": 0,
  502. "tray_id": 0,
  503. "global_tray_id": 0,
  504. "remain": 10,
  505. },
  506. {
  507. "type": "PLA",
  508. "color": "#FF0000",
  509. "ams_id": 0,
  510. "tray_id": 1,
  511. "global_tray_id": 1,
  512. "remain": -1,
  513. },
  514. ]
  515. overrides = {1: 800.0} # only slot 2 has an inventory binding
  516. result = scheduler._match_filaments_to_slots(
  517. required, loaded, prefer_lowest=True, inventory_remain_overrides=overrides
  518. )
  519. assert result == [1]
  520. def test_two_tracked_spools_tied_at_same_grams_lower_slot_wins(self, scheduler):
  521. """Two inventory-tracked spools with genuinely equal remaining
  522. weight — slot tie-breaker decides. Lower AMS slot wins to match
  523. the user's mental model of "use the lower slot first when equal."
  524. """
  525. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  526. loaded = [
  527. {
  528. "type": "PLA",
  529. "color": "#FF0000",
  530. "ams_id": 0,
  531. "tray_id": 1,
  532. "global_tray_id": 1,
  533. "remain": -1,
  534. },
  535. {
  536. "type": "PLA",
  537. "color": "#FF0000",
  538. "ams_id": 0,
  539. "tray_id": 0,
  540. "global_tray_id": 0,
  541. "remain": -1,
  542. },
  543. ]
  544. overrides = {0: 500.0, 1: 500.0}
  545. result = scheduler._match_filaments_to_slots(
  546. required, loaded, prefer_lowest=True, inventory_remain_overrides=overrides
  547. )
  548. assert result == [0]
  549. def test_no_override_falls_back_to_mqtt_remain(self, scheduler):
  550. """When no slots are inventory-bound the override map is empty
  551. and behaviour is identical to the pre-#1508 MQTT-only sort.
  552. Regression guard for the un-tracked-spool case.
  553. """
  554. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  555. loaded = [
  556. {
  557. "type": "PLA",
  558. "color": "#FF0000",
  559. "ams_id": 0,
  560. "tray_id": 0,
  561. "global_tray_id": 0,
  562. "remain": 80,
  563. },
  564. {
  565. "type": "PLA",
  566. "color": "#FF0000",
  567. "ams_id": 0,
  568. "tray_id": 1,
  569. "global_tray_id": 1,
  570. "remain": 30,
  571. },
  572. ]
  573. result = scheduler._match_filaments_to_slots(
  574. required, loaded, prefer_lowest=True, inventory_remain_overrides={}
  575. )
  576. assert result == [1]
  577. def test_no_override_unknown_remain_sorts_after_known(self, scheduler):
  578. """In the no-binding case, ``remain = -1`` (non-RFID, unknown) must
  579. still sort *after* a slot the printer knows something about — the
  580. legacy sentinel-101 behaviour is preserved.
  581. """
  582. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  583. loaded = [
  584. {
  585. "type": "PLA",
  586. "color": "#FF0000",
  587. "ams_id": 0,
  588. "tray_id": 0,
  589. "global_tray_id": 0,
  590. "remain": -1,
  591. },
  592. {
  593. "type": "PLA",
  594. "color": "#FF0000",
  595. "ams_id": 0,
  596. "tray_id": 1,
  597. "global_tray_id": 1,
  598. "remain": 50,
  599. },
  600. ]
  601. result = scheduler._match_filaments_to_slots(
  602. required, loaded, prefer_lowest=True, inventory_remain_overrides=None
  603. )
  604. assert result == [1]
  605. def test_external_with_negative_ams_id_does_not_outrank_ams_on_tie(self, scheduler):
  606. """``_build_loaded_filaments`` emits external/VT trays with
  607. ``ams_id = -1``. A naive ``ams_id * 4 + tray_id`` slot-priority
  608. formula would compute -4 for an external and 0 for AMS slot 0 —
  609. flipping the legacy stable-sort baseline (which kept AMS first
  610. because it's emitted before externals). When ``remain`` ties
  611. between the two, AMS slot 0 must still win.
  612. """
  613. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  614. loaded = [
  615. {
  616. "type": "PLA",
  617. "color": "#FF0000",
  618. "ams_id": 0,
  619. "tray_id": 0,
  620. "global_tray_id": 0,
  621. "remain": -1,
  622. "is_external": False,
  623. },
  624. {
  625. "type": "PLA",
  626. "color": "#FF0000",
  627. "ams_id": -1,
  628. "tray_id": 0,
  629. "global_tray_id": 254,
  630. "remain": -1,
  631. "is_external": True,
  632. },
  633. ]
  634. result = scheduler._match_filaments_to_slots(
  635. required, loaded, prefer_lowest=True, inventory_remain_overrides=None
  636. )
  637. assert result == [0]
  638. def test_ams_ht_does_not_outrank_regular_ams_on_tie(self, scheduler):
  639. """AMS-HT units use ``ams_id`` >= 128 with a single tray. On a tied
  640. ``remain`` value, regular AMS slot 0 (slot_priority 0) must beat
  641. AMS-HT (slot_priority 1000+).
  642. """
  643. required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
  644. loaded = [
  645. {
  646. "type": "PLA",
  647. "color": "#FF0000",
  648. "ams_id": 0,
  649. "tray_id": 0,
  650. "global_tray_id": 0,
  651. "remain": 50,
  652. },
  653. {
  654. "type": "PLA",
  655. "color": "#FF0000",
  656. "ams_id": 128,
  657. "tray_id": 0,
  658. "global_tray_id": 128,
  659. "remain": 50,
  660. },
  661. ]
  662. result = scheduler._match_filaments_to_slots(
  663. required, loaded, prefer_lowest=True, inventory_remain_overrides=None
  664. )
  665. assert result == [0]
  666. class TestBuildLoadedFilamentsTrayInfoIdx:
  667. """Test tray_info_idx extraction in _build_loaded_filaments."""
  668. @pytest.fixture
  669. def scheduler(self):
  670. return PrintScheduler()
  671. def test_build_loaded_filaments_includes_tray_info_idx(self, scheduler):
  672. """Should extract tray_info_idx from AMS trays."""
  673. class MockStatus:
  674. raw_data = {
  675. "ams": [
  676. {
  677. "id": 0,
  678. "tray": [
  679. {"id": 0, "tray_type": "PLA", "tray_color": "000000", "tray_info_idx": "GFA00"},
  680. {"id": 1, "tray_type": "PLA", "tray_color": "000000", "tray_info_idx": "GFA01"},
  681. ],
  682. }
  683. ]
  684. }
  685. result = scheduler._build_loaded_filaments(MockStatus())
  686. assert len(result) == 2
  687. assert result[0]["tray_info_idx"] == "GFA00"
  688. assert result[1]["tray_info_idx"] == "GFA01"
  689. def test_build_loaded_filaments_empty_tray_info_idx(self, scheduler):
  690. """Missing tray_info_idx should default to empty string."""
  691. class MockStatus:
  692. raw_data = {
  693. "ams": [
  694. {
  695. "id": 0,
  696. "tray": [
  697. {"id": 0, "tray_type": "PLA", "tray_color": "FF0000"}, # No tray_info_idx
  698. ],
  699. }
  700. ]
  701. }
  702. result = scheduler._build_loaded_filaments(MockStatus())
  703. assert len(result) == 1
  704. assert result[0]["tray_info_idx"] == ""
  705. def test_build_loaded_filaments_external_spool_tray_info_idx(self, scheduler):
  706. """Should extract tray_info_idx from external spool."""
  707. class MockStatus:
  708. raw_data = {"vt_tray": [{"tray_type": "TPU", "tray_color": "0000FF", "tray_info_idx": "P4d64437"}]}
  709. result = scheduler._build_loaded_filaments(MockStatus())
  710. assert len(result) == 1
  711. assert result[0]["tray_info_idx"] == "P4d64437"
  712. assert result[0]["is_external"] is True
  713. def _make_3mf_zip(
  714. project_settings: dict | None = None,
  715. slice_info_xml: str | None = None,
  716. ) -> zipfile.ZipFile:
  717. """Create an in-memory ZipFile mimicking a 3MF with project_settings.config."""
  718. buf = io.BytesIO()
  719. with zipfile.ZipFile(buf, "w") as zf:
  720. if project_settings is not None:
  721. zf.writestr("Metadata/project_settings.config", json.dumps(project_settings))
  722. if slice_info_xml is not None:
  723. zf.writestr("Metadata/slice_info.config", slice_info_xml)
  724. buf.seek(0)
  725. return zipfile.ZipFile(buf, "r")
  726. class TestExtractNozzleMappingFrom3mf:
  727. """Test the extract_nozzle_mapping_from_3mf utility."""
  728. def test_group_id_priority_over_filament_nozzle_map(self):
  729. """group_id from slice_info should override filament_nozzle_map from project_settings.
  730. Real-world scenario: "Auto For Flush" mode sets filament_nozzle_map all to 0
  731. (user preference) but the actual assignment in slice_info has different group_ids.
  732. """
  733. # filament_nozzle_map says all on slicer ext 0 → MQTT ext 1 (LEFT)
  734. # But slice_info group_id says slot 6 → group 0 (LEFT), slot 12 → group 1 (RIGHT)
  735. slice_info = """<?xml version="1.0" encoding="UTF-8"?>
  736. <config>
  737. <plate>
  738. <filament id="6" type="PLA" color="#56B7E6" used_g="1.84" group_id="0"/>
  739. <filament id="12" type="PLA" color="#B39B84" used_g="1.76" group_id="1"/>
  740. </plate>
  741. </config>"""
  742. zf = _make_3mf_zip(
  743. {
  744. "filament_nozzle_map": ["0"] * 12,
  745. "physical_extruder_map": ["1", "0"],
  746. },
  747. slice_info_xml=slice_info,
  748. )
  749. result = extract_nozzle_mapping_from_3mf(zf)
  750. # group_id 0 → physical_extruder_map[0] = 1 (LEFT)
  751. # group_id 1 → physical_extruder_map[1] = 0 (RIGHT)
  752. assert result == {6: 1, 12: 0}
  753. zf.close()
  754. def test_fallback_to_filament_nozzle_map_without_group_id(self):
  755. """Should fall back to filament_nozzle_map when slice_info has no group_id."""
  756. slice_info = """<?xml version="1.0" encoding="UTF-8"?>
  757. <config>
  758. <plate>
  759. <filament id="1" type="PLA" color="#FF0000" used_g="5.0"/>
  760. </plate>
  761. </config>"""
  762. zf = _make_3mf_zip(
  763. {
  764. "filament_nozzle_map": ["0", "1", "0"],
  765. "physical_extruder_map": ["0", "1"],
  766. },
  767. slice_info_xml=slice_info,
  768. )
  769. result = extract_nozzle_mapping_from_3mf(zf)
  770. assert result == {1: 0, 2: 1, 3: 0}
  771. zf.close()
  772. def test_fallback_to_filament_nozzle_map_without_slice_info(self):
  773. """Should fall back to filament_nozzle_map when no slice_info.config exists."""
  774. zf = _make_3mf_zip(
  775. {
  776. "filament_nozzle_map": ["0", "1", "0"],
  777. "physical_extruder_map": ["0", "1"],
  778. }
  779. )
  780. result = extract_nozzle_mapping_from_3mf(zf)
  781. assert result == {1: 0, 2: 1, 3: 0}
  782. zf.close()
  783. def test_single_nozzle_returns_none(self):
  784. """Single physical_extruder_map entry should return None (single-nozzle)."""
  785. zf = _make_3mf_zip(
  786. {
  787. "filament_nozzle_map": ["0", "0", "0"],
  788. "physical_extruder_map": ["0"],
  789. }
  790. )
  791. result = extract_nozzle_mapping_from_3mf(zf)
  792. assert result is None
  793. zf.close()
  794. def test_missing_project_settings_returns_none(self):
  795. """Missing project_settings.config should return None."""
  796. zf = _make_3mf_zip(None)
  797. result = extract_nozzle_mapping_from_3mf(zf)
  798. assert result is None
  799. zf.close()
  800. def test_missing_fields_returns_none(self):
  801. """Missing physical_extruder_map should return None."""
  802. zf = _make_3mf_zip({"some_other_key": "value"})
  803. result = extract_nozzle_mapping_from_3mf(zf)
  804. assert result is None
  805. zf.close()
  806. def test_physical_extruder_map_remapping(self):
  807. """Should apply physical_extruder_map to remap slicer extruder to MQTT extruder."""
  808. # Slicer ext 0 -> MQTT ext 1, slicer ext 1 -> MQTT ext 0
  809. zf = _make_3mf_zip(
  810. {
  811. "filament_nozzle_map": ["0", "1"],
  812. "physical_extruder_map": ["1", "0"],
  813. }
  814. )
  815. result = extract_nozzle_mapping_from_3mf(zf)
  816. assert result == {1: 1, 2: 0}
  817. zf.close()
  818. def test_single_active_extruder_maps_all_slots(self):
  819. """When only one extruder has nozzles installed, all filaments map to it.
  820. H2C scenario: left extruder has no nozzles (Standard#0|High Flow#0),
  821. right extruder has one Standard nozzle (Standard#1). Print uses only
  822. the right extruder, so all filaments should map to physical extruder 0.
  823. """
  824. slice_info = """<?xml version="1.0" encoding="UTF-8"?>
  825. <config>
  826. <plate>
  827. <filament id="1" type="PLA" color="#FF0000" used_g="5.0" group_id="0"/>
  828. <filament id="2" type="PLA" color="#00FF00" used_g="3.0" group_id="0"/>
  829. </plate>
  830. </config>"""
  831. zf = _make_3mf_zip(
  832. {
  833. "physical_extruder_map": ["1", "0"],
  834. "extruder_nozzle_stats": ["Standard#0|High Flow#0", "Standard#1"],
  835. },
  836. slice_info_xml=slice_info,
  837. )
  838. result = extract_nozzle_mapping_from_3mf(zf)
  839. # Only extruder index 1 is active → physical_extruder_map[1] = 0 (RIGHT)
  840. assert result == {1: 0, 2: 0}
  841. zf.close()
  842. def test_two_active_extruders_falls_through(self):
  843. """When both extruders have nozzles, the single-active shortcut is skipped.
  844. Should fall through to the normal group_id-based mapping.
  845. """
  846. slice_info = """<?xml version="1.0" encoding="UTF-8"?>
  847. <config>
  848. <plate>
  849. <filament id="1" type="PLA" color="#FF0000" used_g="5.0" group_id="0"/>
  850. <filament id="2" type="PLA" color="#00FF00" used_g="3.0" group_id="1"/>
  851. </plate>
  852. </config>"""
  853. zf = _make_3mf_zip(
  854. {
  855. "physical_extruder_map": ["1", "0"],
  856. "extruder_nozzle_stats": ["Standard#1", "High Flow#1"],
  857. },
  858. slice_info_xml=slice_info,
  859. )
  860. result = extract_nozzle_mapping_from_3mf(zf)
  861. # Both active → normal group_id mapping: group 0 → phys 1, group 1 → phys 0
  862. assert result == {1: 1, 2: 0}
  863. zf.close()
  864. def test_missing_extruder_nozzle_stats_falls_through(self):
  865. """When extruder_nozzle_stats is absent, the single-active shortcut is skipped.
  866. Should fall through to the normal group_id-based mapping.
  867. """
  868. slice_info = """<?xml version="1.0" encoding="UTF-8"?>
  869. <config>
  870. <plate>
  871. <filament id="1" type="PLA" color="#FF0000" used_g="5.0" group_id="0"/>
  872. <filament id="2" type="PLA" color="#00FF00" used_g="3.0" group_id="1"/>
  873. </plate>
  874. </config>"""
  875. zf = _make_3mf_zip(
  876. {
  877. "physical_extruder_map": ["1", "0"],
  878. },
  879. slice_info_xml=slice_info,
  880. )
  881. result = extract_nozzle_mapping_from_3mf(zf)
  882. # No extruder_nozzle_stats → normal group_id mapping
  883. assert result == {1: 1, 2: 0}
  884. zf.close()
  885. def test_single_active_extruder_no_slice_info_returns_none(self):
  886. """Single active extruder but no slice_info should return None."""
  887. zf = _make_3mf_zip(
  888. {
  889. "physical_extruder_map": ["1", "0"],
  890. "extruder_nozzle_stats": ["Standard#0", "Standard#1"],
  891. },
  892. )
  893. result = extract_nozzle_mapping_from_3mf(zf)
  894. assert result is None
  895. zf.close()
  896. class TestNozzleAwareMapping:
  897. """Test nozzle-aware filament matching in the print scheduler."""
  898. @pytest.fixture
  899. def scheduler(self):
  900. return PrintScheduler()
  901. def test_dual_nozzle_matching(self, scheduler):
  902. """Filaments assigned to different nozzles should match to correct AMS units."""
  903. required = [
  904. {"slot_id": 1, "type": "PLA", "color": "#FF0000", "nozzle_id": 0}, # Right nozzle
  905. {"slot_id": 2, "type": "PLA", "color": "#00FF00", "nozzle_id": 1}, # Left nozzle
  906. ]
  907. loaded = [
  908. {"type": "PLA", "color": "#00FF00", "global_tray_id": 0, "extruder_id": 0}, # AMS0 on right
  909. {"type": "PLA", "color": "#FF0000", "global_tray_id": 4, "extruder_id": 1}, # AMS1 on left
  910. ]
  911. # Without nozzle filtering, slot 1 (red, right) would match tray 4 (red, left) by color.
  912. # With nozzle filtering, slot 1 (right nozzle) can only use tray 0 (right extruder),
  913. # and slot 2 (left nozzle) can only use tray 4 (left extruder).
  914. result = scheduler._match_filaments_to_slots(required, loaded)
  915. assert result == [0, 4]
  916. def test_nozzle_hard_filter_no_fallback(self, scheduler):
  917. """Hard filter: no fallback to wrong nozzle when target nozzle has no trays."""
  918. required = [
  919. {"slot_id": 1, "type": "PLA", "color": "#FF0000", "nozzle_id": 0}, # Right nozzle
  920. ]
  921. loaded = [
  922. # Only a tray on the left nozzle, none on right
  923. {"type": "PLA", "color": "#FF0000", "global_tray_id": 4, "extruder_id": 1},
  924. ]
  925. # No trays on extruder 0 — hard filter returns -1, no cross-nozzle fallback
  926. result = scheduler._match_filaments_to_slots(required, loaded)
  927. assert result == [-1]
  928. def test_no_nozzle_id_skips_filtering(self, scheduler):
  929. """When nozzle_id is None, no nozzle filtering should be applied."""
  930. required = [
  931. {"slot_id": 1, "type": "PLA", "color": "#FF0000"}, # No nozzle_id
  932. ]
  933. loaded = [
  934. {"type": "PLA", "color": "#FF0000", "global_tray_id": 0, "extruder_id": 0},
  935. {"type": "PLA", "color": "#FF0000", "global_tray_id": 4, "extruder_id": 1},
  936. ]
  937. # Should match first available (tray 0) regardless of extruder
  938. result = scheduler._match_filaments_to_slots(required, loaded)
  939. assert result == [0]
  940. def test_extruder_id_in_loaded_filaments(self, scheduler):
  941. """_build_loaded_filaments should include extruder_id from ams_extruder_map."""
  942. class MockStatus:
  943. raw_data = {
  944. "ams": [
  945. {"id": 0, "tray": [{"id": 0, "tray_type": "PLA", "tray_color": "FF0000"}]},
  946. {"id": 1, "tray": [{"id": 0, "tray_type": "PLA", "tray_color": "00FF00"}]},
  947. ],
  948. "ams_extruder_map": {"0": 0, "1": 1},
  949. }
  950. result = scheduler._build_loaded_filaments(MockStatus())
  951. assert len(result) == 2
  952. assert result[0]["extruder_id"] == 0
  953. assert result[1]["extruder_id"] == 1
  954. def test_extruder_id_none_without_map(self, scheduler):
  955. """extruder_id should be None when ams_extruder_map is absent."""
  956. class MockStatus:
  957. raw_data = {
  958. "ams": [
  959. {"id": 0, "tray": [{"id": 0, "tray_type": "PLA", "tray_color": "FF0000"}]},
  960. ]
  961. }
  962. result = scheduler._build_loaded_filaments(MockStatus())
  963. assert len(result) == 1
  964. assert result[0]["extruder_id"] is None
  965. def test_external_spool_extruder_id(self, scheduler):
  966. """External spool 254 (Ext-L) should have extruder_id=1 (LEFT) when ams_extruder_map exists."""
  967. class MockStatus:
  968. raw_data = {
  969. "vt_tray": [{"tray_type": "TPU", "tray_color": "0000FF"}],
  970. "ams_extruder_map": {"0": 0},
  971. }
  972. result = scheduler._build_loaded_filaments(MockStatus())
  973. assert len(result) == 1
  974. # Default vt_tray id=254 → Ext-L → LEFT nozzle (extruder 1)
  975. assert result[0]["extruder_id"] == 1
  976. assert result[0]["is_external"] is True
  977. def test_external_spool_no_extruder_map(self, scheduler):
  978. """External spool extruder_id should be None without ams_extruder_map."""
  979. class MockStatus:
  980. raw_data = {"vt_tray": [{"tray_type": "TPU", "tray_color": "0000FF"}]}
  981. result = scheduler._build_loaded_filaments(MockStatus())
  982. assert len(result) == 1
  983. assert result[0]["extruder_id"] is None
  984. def test_dual_nozzle_with_tray_info_idx(self, scheduler):
  985. """Nozzle filtering should work together with tray_info_idx matching."""
  986. required = [
  987. {"slot_id": 1, "type": "PLA", "color": "#000000", "tray_info_idx": "GFA00", "nozzle_id": 0},
  988. {"slot_id": 2, "type": "PLA", "color": "#000000", "tray_info_idx": "GFA01", "nozzle_id": 1},
  989. ]
  990. loaded = [
  991. {"type": "PLA", "color": "#000000", "global_tray_id": 0, "tray_info_idx": "GFA00", "extruder_id": 0},
  992. {"type": "PLA", "color": "#000000", "global_tray_id": 4, "tray_info_idx": "GFA01", "extruder_id": 1},
  993. ]
  994. result = scheduler._match_filaments_to_slots(required, loaded)
  995. assert result == [0, 4]
  996. # ============================================================================
  997. # MODEL-SPECIFIC TESTS: Real data from actual printers
  998. # ============================================================================
  999. def _h2d_raw_data():
  1000. """H2D real data fixture (from live API response 2026-02-18).
  1001. Configuration:
  1002. LEFT nozzle (extruder 1): AMS 0 (4-slot), AMS 2 (4-slot)
  1003. RIGHT nozzle (extruder 0): AMS 1 (4-slot), AMS-HT 128 (1-slot, empty)
  1004. External: 254 (Ext-L, LEFT), 255 (Ext-R, RIGHT, empty)
  1005. ams_extruder_map: {"0": 1, "1": 0, "2": 1, "128": 0}
  1006. """
  1007. return {
  1008. "ams": [
  1009. {
  1010. "id": 0,
  1011. "tray": [
  1012. {"id": 0, "tray_type": "PETG", "tray_color": "FFFFFFFF", "tray_info_idx": "GFG02"},
  1013. {"id": 1, "tray_type": "PLA", "tray_color": "C8C8C8FF", "tray_info_idx": "GFA06"},
  1014. {"id": 2, "tray_type": "PETG", "tray_color": "875718FF", "tray_info_idx": "GFG02"},
  1015. {"id": 3, "tray_type": "PLA", "tray_color": "000000FF", "tray_info_idx": "GFA00"},
  1016. ],
  1017. },
  1018. {
  1019. "id": 1,
  1020. "tray": [
  1021. {"id": 0, "tray_type": "PLA", "tray_color": "FFFFFFFF", "tray_info_idx": "GFA00"},
  1022. {"id": 1, "tray_type": "PETG", "tray_color": "000000FF", "tray_info_idx": "GFG02"},
  1023. {"id": 2, "tray_type": "PLA", "tray_color": "5F6367FF", "tray_info_idx": "GFA06"},
  1024. {"id": 3, "tray_type": "PLA", "tray_color": "B39B84FF", "tray_info_idx": "GFA02"},
  1025. ],
  1026. },
  1027. {
  1028. "id": 128,
  1029. "tray": [{"id": 0}], # AMS-HT, empty
  1030. },
  1031. {
  1032. "id": 2,
  1033. "tray": [
  1034. {"id": 0, "tray_type": "PLA-S", "tray_color": "FFFFFFFF", "tray_info_idx": "P8aa1726"},
  1035. {"id": 1, "tray_type": "PLA", "tray_color": "56B7E6FF", "tray_info_idx": "PFUS9924"},
  1036. {"id": 2, "tray_type": "PETG", "tray_color": "6EE53CFF", "tray_info_idx": "GFG02"},
  1037. {"id": 3, "tray_type": "PLA", "tray_color": "FF0000FF", "tray_info_idx": "PFUS9ac9"},
  1038. ],
  1039. },
  1040. ],
  1041. "vt_tray": [
  1042. {"id": 254, "tray_type": "PLA", "tray_color": "000000FF", "tray_info_idx": "P4d64437"},
  1043. {"id": 255, "tray_type": "", "tray_color": "00000000"}, # empty
  1044. ],
  1045. "ams_extruder_map": {"0": 1, "1": 0, "2": 1, "128": 0},
  1046. }
  1047. def _x1c_raw_data():
  1048. """X1C real data fixture (from live API response 2026-02-18).
  1049. Configuration:
  1050. Single nozzle (extruder 0): AMS 0 (4-slot, all empty), AMS 1 (4-slot, 3 loaded)
  1051. External: 254 (single, empty)
  1052. ams_extruder_map: {"0": 0, "1": 0} ← NOT empty, all on extruder 0
  1053. """
  1054. return {
  1055. "ams": [
  1056. {
  1057. "id": 0,
  1058. "tray": [
  1059. {"id": 0}, # empty
  1060. {"id": 1}, # empty
  1061. {"id": 2}, # empty
  1062. {"id": 3}, # empty
  1063. ],
  1064. },
  1065. {
  1066. "id": 1,
  1067. "tray": [
  1068. {"id": 0}, # empty
  1069. {"id": 1, "tray_type": "PLA", "tray_color": "EBCFA6FF", "tray_info_idx": "PFUS22b2"},
  1070. {"id": 2, "tray_type": "PLA", "tray_color": "FCECD6FF", "tray_info_idx": "P4d64437"},
  1071. {"id": 3, "tray_type": "PLA", "tray_color": "0066FFFF", "tray_info_idx": "P4d64437"},
  1072. ],
  1073. },
  1074. ],
  1075. "vt_tray": [
  1076. {"id": 254, "tray_type": "", "tray_color": "00000000"}, # empty
  1077. ],
  1078. "ams_extruder_map": {"0": 0, "1": 0},
  1079. }
  1080. class TestH2DModel:
  1081. """H2D-specific tests with real printer data (dual nozzle, AMS-HT)."""
  1082. @pytest.fixture
  1083. def scheduler(self):
  1084. return PrintScheduler()
  1085. def test_build_loaded_filaments_h2d(self, scheduler):
  1086. """H2D: correct extruder_id, global_tray_id, AMS-HT handling."""
  1087. class MockStatus:
  1088. raw_data = _h2d_raw_data()
  1089. result = scheduler._build_loaded_filaments(MockStatus())
  1090. # Should have 13 loaded filaments (4 + 4 + 0 + 4 + 1 external)
  1091. assert len(result) == 13
  1092. # AMS 0 trays → extruder 1 (LEFT)
  1093. ams0 = [f for f in result if f["ams_id"] == 0]
  1094. assert len(ams0) == 4
  1095. assert all(f["extruder_id"] == 1 for f in ams0)
  1096. assert [f["global_tray_id"] for f in ams0] == [0, 1, 2, 3]
  1097. # AMS 1 trays → extruder 0 (RIGHT)
  1098. ams1 = [f for f in result if f["ams_id"] == 1]
  1099. assert len(ams1) == 4
  1100. assert all(f["extruder_id"] == 0 for f in ams1)
  1101. assert [f["global_tray_id"] for f in ams1] == [4, 5, 6, 7]
  1102. # AMS-HT 128 → empty, should not appear
  1103. ams_ht = [f for f in result if f["ams_id"] == 128]
  1104. assert len(ams_ht) == 0
  1105. # AMS 2 trays → extruder 1 (LEFT)
  1106. ams2 = [f for f in result if f["ams_id"] == 2]
  1107. assert len(ams2) == 4
  1108. assert all(f["extruder_id"] == 1 for f in ams2)
  1109. assert [f["global_tray_id"] for f in ams2] == [8, 9, 10, 11]
  1110. def test_external_spool_extruder_h2d(self, scheduler):
  1111. """H2D: Ext-L (254) = LEFT (extruder 1), Ext-R (255) = RIGHT (extruder 0)."""
  1112. class MockStatus:
  1113. raw_data = _h2d_raw_data()
  1114. result = scheduler._build_loaded_filaments(MockStatus())
  1115. ext = [f for f in result if f["is_external"]]
  1116. assert len(ext) == 1 # Only 254 has filament
  1117. assert ext[0]["global_tray_id"] == 254
  1118. # Ext-L (254) should be LEFT nozzle (extruder 1)
  1119. assert ext[0]["extruder_id"] == 1
  1120. def test_match_left_nozzle_only(self, scheduler):
  1121. """H2D: left-nozzle requirement only matches left-nozzle AMS."""
  1122. class MockStatus:
  1123. raw_data = _h2d_raw_data()
  1124. loaded = scheduler._build_loaded_filaments(MockStatus())
  1125. required = [
  1126. {"slot_id": 1, "type": "PLA", "color": "#000000", "nozzle_id": 1}, # LEFT
  1127. ]
  1128. result = scheduler._match_filaments_to_slots(required, loaded)
  1129. # Black PLA on LEFT: AMS 0 T4 (global 3)
  1130. assert result == [3]
  1131. def test_match_right_nozzle_only(self, scheduler):
  1132. """H2D: right-nozzle requirement only matches right-nozzle AMS."""
  1133. class MockStatus:
  1134. raw_data = _h2d_raw_data()
  1135. loaded = scheduler._build_loaded_filaments(MockStatus())
  1136. required = [
  1137. {"slot_id": 1, "type": "PLA", "color": "#FFFFFF", "nozzle_id": 0}, # RIGHT
  1138. ]
  1139. result = scheduler._match_filaments_to_slots(required, loaded)
  1140. # White PLA on RIGHT: AMS 1 T1 (global 4)
  1141. assert result == [4]
  1142. def test_reject_cross_nozzle(self, scheduler):
  1143. """H2D: hard filter rejects cross-nozzle assignment."""
  1144. class MockStatus:
  1145. raw_data = _h2d_raw_data()
  1146. loaded = scheduler._build_loaded_filaments(MockStatus())
  1147. # PLA-S only exists on AMS 2 T1 (LEFT), require on RIGHT
  1148. required = [
  1149. {"slot_id": 1, "type": "PLA-S", "color": "#FFFFFF", "nozzle_id": 0},
  1150. ]
  1151. result = scheduler._match_filaments_to_slots(required, loaded)
  1152. assert result == [-1] # No fallback to wrong nozzle
  1153. def test_dual_nozzle_multi_filament(self, scheduler):
  1154. """H2D: multi-filament print maps to correct nozzles."""
  1155. class MockStatus:
  1156. raw_data = _h2d_raw_data()
  1157. loaded = scheduler._build_loaded_filaments(MockStatus())
  1158. required = [
  1159. {"slot_id": 1, "type": "PETG", "color": "#FFFFFF", "nozzle_id": 1, "tray_info_idx": "GFG02"},
  1160. {"slot_id": 2, "type": "PLA", "color": "#FFFFFF", "nozzle_id": 0, "tray_info_idx": "GFA00"},
  1161. ]
  1162. result = scheduler._match_filaments_to_slots(required, loaded)
  1163. # PETG white on LEFT: AMS 0 T1 (global 0)
  1164. # PLA white on RIGHT: AMS 1 T1 (global 4)
  1165. assert result == [0, 4]
  1166. def test_external_spool_matches_on_correct_nozzle(self, scheduler):
  1167. """H2D: external spool on left nozzle matches left-nozzle requirement."""
  1168. class MockStatus:
  1169. raw_data = _h2d_raw_data()
  1170. loaded = scheduler._build_loaded_filaments(MockStatus())
  1171. required = [
  1172. {"slot_id": 1, "type": "PLA", "color": "#000000", "nozzle_id": 1, "tray_info_idx": "P4d64437"},
  1173. ]
  1174. result = scheduler._match_filaments_to_slots(required, loaded)
  1175. assert result == [254] # External spool on left nozzle
  1176. class TestX1CModel:
  1177. """X1C-specific tests with real printer data (single nozzle, 2x regular AMS)."""
  1178. @pytest.fixture
  1179. def scheduler(self):
  1180. return PrintScheduler()
  1181. def test_build_loaded_filaments_x1c(self, scheduler):
  1182. """X1C: all filaments on extruder 0, correct global_tray_id."""
  1183. class MockStatus:
  1184. raw_data = _x1c_raw_data()
  1185. result = scheduler._build_loaded_filaments(MockStatus())
  1186. # Only 3 loaded (AMS 1 trays 1-3)
  1187. assert len(result) == 3
  1188. # All on extruder 0
  1189. assert all(f["extruder_id"] == 0 for f in result)
  1190. # Correct global tray IDs
  1191. assert [f["global_tray_id"] for f in result] == [5, 6, 7]
  1192. def test_single_nozzle_no_filtering(self, scheduler):
  1193. """X1C: single-nozzle 3MF has no nozzle_id, all trays available."""
  1194. class MockStatus:
  1195. raw_data = _x1c_raw_data()
  1196. loaded = scheduler._build_loaded_filaments(MockStatus())
  1197. required = [
  1198. {"slot_id": 1, "type": "PLA", "color": "#0066FF"}, # No nozzle_id
  1199. ]
  1200. result = scheduler._match_filaments_to_slots(required, loaded)
  1201. # Blue PLA → AMS 1 T4 (global 7)
  1202. assert result == [7]
  1203. def test_tray_info_idx_matching_x1c(self, scheduler):
  1204. """X1C: tray_info_idx matching works across AMS units."""
  1205. class MockStatus:
  1206. raw_data = _x1c_raw_data()
  1207. loaded = scheduler._build_loaded_filaments(MockStatus())
  1208. required = [
  1209. {"slot_id": 1, "type": "PLA", "color": "#EBCFA6", "tray_info_idx": "PFUS22b2"},
  1210. ]
  1211. result = scheduler._match_filaments_to_slots(required, loaded)
  1212. # Unique tray_info_idx → AMS 1 T2 (global 5)
  1213. assert result == [5]
  1214. def test_non_unique_tray_info_idx_color_match_x1c(self, scheduler):
  1215. """X1C: non-unique tray_info_idx falls back to color matching."""
  1216. class MockStatus:
  1217. raw_data = _x1c_raw_data()
  1218. loaded = scheduler._build_loaded_filaments(MockStatus())
  1219. # P4d64437 appears in AMS 1 T3 and T4
  1220. required = [
  1221. {"slot_id": 1, "type": "PLA", "color": "#FCECD6", "tray_info_idx": "P4d64437"},
  1222. ]
  1223. result = scheduler._match_filaments_to_slots(required, loaded)
  1224. # Should pick AMS 1 T3 (global 6, color FCECD6) over T4 (0066FF)
  1225. assert result == [6]
  1226. def test_multi_filament_x1c(self, scheduler):
  1227. """X1C: multi-filament print matches freely across AMS units."""
  1228. class MockStatus:
  1229. raw_data = _x1c_raw_data()
  1230. loaded = scheduler._build_loaded_filaments(MockStatus())
  1231. required = [
  1232. {"slot_id": 1, "type": "PLA", "color": "#EBCFA6"},
  1233. {"slot_id": 2, "type": "PLA", "color": "#0066FF"},
  1234. ]
  1235. result = scheduler._match_filaments_to_slots(required, loaded)
  1236. assert result == [5, 7]