From cda0e984ecd8287b9f2b16c60b49b0c050c0832e Mon Sep 17 00:00:00 2001 From: viewit Date: Tue, 4 Aug 2026 20:18:20 +0200 Subject: [PATCH] refactor(bridge): extract AmsFilamentMixin (stage 3, mixin 3/6) --- bridge_ams.py | 862 +++++++++++++++++++++++++++++++++++++ kobrax_moonraker_bridge.py | 842 +----------------------------------- 2 files changed, 864 insertions(+), 840 deletions(-) create mode 100644 bridge_ams.py diff --git a/bridge_ams.py b/bridge_ams.py new file mode 100644 index 0000000..7c731c0 --- /dev/null +++ b/bridge_ams.py @@ -0,0 +1,862 @@ +""" +bridge_ams.py - AmsFilamentMixin for KobraXBridge. + +AMS/ACE slot topology + aggregation, per-slot filament-profile resolution +(material normalization, RFID vendor matching, effective slot profile), +ACE dryer presets, the lane-data builder for OrcaSlicer sync, and the +_on_multicolor_box / _on_light MQTT callbacks (which live here because they +are all about AMS state). Mixed into KobraXBridge. + +──────────────────────────────────────────────────────────────────────────── +Copyright (C) 2026 viewit (KX-Bridge contributors) + +Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md. +""" + +import re +import time +import logging +import threading + +log = logging.getLogger("bridge") + + +class AmsFilamentMixin: + def _default_ace_dry_presets(self) -> dict[str, dict]: + return { + "pla": {"temp": 45, "duration_sec": 4 * 3600}, + "pla_plus": {"temp": 45, "duration_sec": 4 * 3600}, + "petg": {"temp": 50, "duration_sec": 4 * 3600}, + "tpu": {"temp": 55, "duration_sec": 4 * 3600}, + "abs_asa": {"temp": 45, "duration_sec": 8 * 3600}, + "pa_pc": {"temp": 55, "duration_sec": 12 * 3600}, + "custom_1": {"name": "Custom 1", "temp": 45, "duration_sec": 4 * 3600}, + "custom_2": {"name": "Custom 2", "temp": 45, "duration_sec": 4 * 3600}, + "custom_3": {"name": "Custom 3", "temp": 45, "duration_sec": 4 * 3600}, + } + + def _sanitize_ace_dry_presets(self, presets: dict) -> dict[str, dict]: + out = self._default_ace_dry_presets() + for key in list(out.keys()): + src = presets.get(key) if isinstance(presets, dict) else None + if not isinstance(src, dict): + continue + try: + t = int(src.get("temp", out[key]["temp"])) + except Exception: + t = out[key]["temp"] + try: + d = int(src.get("duration_sec", out[key]["duration_sec"])) + except Exception: + d = out[key]["duration_sec"] + out[key]["temp"] = max(30, min(80, t)) + out[key]["duration_sec"] = max(10 * 60, min(24 * 3600, d)) + if key.startswith("custom_"): + name = str(src.get("name", out[key].get("name", key.replace("_", " ").title()))).strip() + out[key]["name"] = name or out[key].get("name", "Custom") + return out + + def _load_ace_dry_presets_config(self) -> dict[str, dict]: + import configparser + defaults = self._default_ace_dry_presets() + cfg_path = self._find_config_path() + if not cfg_path.is_file(): + return defaults + cfg = configparser.ConfigParser(interpolation=None) + cfg.read(cfg_path, encoding="utf-8") + sec = "ace_dry_presets" + if not cfg.has_section(sec): + return defaults + out = {} + for key, d in defaults.items(): + temp_k = f"{key}_temp" + dur_k = f"{key}_duration_sec" + try: + temp = int(cfg.get(sec, temp_k, fallback=str(d["temp"]))) + except Exception: + temp = d["temp"] + try: + dur = int(cfg.get(sec, dur_k, fallback=str(d["duration_sec"]))) + except Exception: + dur = d["duration_sec"] + out[key] = { + "temp": max(30, min(80, temp)), + "duration_sec": max(10 * 60, min(24 * 3600, dur)), + } + if key.startswith("custom_"): + name_k = f"{key}_name" + name = cfg.get(sec, name_k, fallback=str(d.get("name", key.replace("_", " ").title()))).strip() + out[key]["name"] = name or str(d.get("name", "Custom")) + return out + + @staticmethod + def _detect_filament_mode(boxes: list, head_tools_model: int = -1) -> str: + """Detect active filament topology mode. + + Modes: + - toolhead: only toolhead slots + - ace_direct: ACE channels directly mapped, no toolhead box present. + Covers one unit (Kobra X) as well as multiple daisy-chained units + (Kobra S1 with 2+ ACE Pro, Issue #95) — each unit contributes a + block of 4 global slots at box_id * 4. + - ace_hub: toolhead + ACE via hub (slot 4 as hub path) + """ + toolhead = any(b.get("id") == -1 for b in boxes) + ace = any(b.get("id", -1) >= 0 for b in boxes) + if ace and toolhead: + return "ace_hub" + if ace: + return "ace_direct" + return "toolhead" + + @staticmethod + def _aggregate_slots(boxes: list, mode: str = "toolhead") -> tuple: + """Aggregate multi_color_box list into a flat global slot list.""" + toolhead = next((b for b in boxes if b.get("id") == -1), None) + ace_boxes = sorted( + [b for b in boxes if b.get("id", -1) >= 0], + key=lambda b: b["id"] + ) + + global_slots: list = [] + global_loaded: int = -1 + + if mode == "toolhead": + if toolhead: + for local_idx, s in enumerate(toolhead.get("slots") or []): + s = dict(s) + s["global_index"] = local_idx + s["box_id"] = -1 + global_slots.append(s) + loaded = toolhead.get("loaded_slot", -1) + if loaded >= 0: + global_loaded = loaded + return global_slots, global_loaded + + if mode == "ace_direct": + # One or more ACE units, no toolhead buffer (Kobra X: 1 unit, + # Kobra S1: up to 2+ units, Issue #95). Global index = + # box_id * 4 + local slot, so the numbering matches + # _global_to_box_slot's //4-%4 fallback and stays stable + # regardless of report order. + for ace in ace_boxes: + ace_id = int(ace["id"]) + base = ace_id * 4 + for local_idx, s in enumerate((ace.get("slots") or [])[:4]): + s = dict(s) + s["global_index"] = base + local_idx + s["box_id"] = ace_id + global_slots.append(s) + ace_loaded = ace.get("loaded_slot", -1) + if 0 <= ace_loaded < 4: + global_loaded = base + ace_loaded + return global_slots, global_loaded + + # ace_hub + if toolhead: + for local_idx, s in enumerate((toolhead.get("slots") or [])[:3]): + s = dict(s) + s["global_index"] = local_idx + s["box_id"] = -1 + global_slots.append(s) + th_loaded = toolhead.get("loaded_slot", -1) + if 0 <= th_loaded <= 2: + global_loaded = th_loaded + + for ace in ace_boxes: + ace_id = ace["id"] + base = 3 + ace_id * 4 + for local_idx, s in enumerate(ace.get("slots") or []): + s = dict(s) + s["global_index"] = base + local_idx + s["box_id"] = ace_id + global_slots.append(s) + ace_loaded = ace.get("loaded_slot", -1) + if ace_loaded >= 0: + global_loaded = base + ace_loaded + + return global_slots, global_loaded + + def _global_to_box_slot(self, global_index: int) -> tuple: + """Convert a global slot index to (box_id, local_slot_index).""" + for s in self._ams_slots: + if s.get("global_index") == global_index: + return s.get("box_id", -1), s.get("index", global_index) + + ace_present = any(s.get("box_id", -1) >= 0 for s in self._ams_slots) + if self._filament_mode == "ace_direct" and ace_present: + return global_index // 4, global_index % 4 + if not ace_present or global_index < 3: + return -1, global_index + offset = global_index - 3 + return offset // 4, offset % 4 + + def _slot_to_print_ams_index(self, global_index: int) -> int: + """Convert UI/global slot index to printer print/start ams_index. + + In ace_hub mode, print/start uses global channel numbering where + toolhead channels occupy 1..3 and ACE0 starts at index 4. + """ + idx = int(global_index) + if self._filament_mode == "ace_hub": + box_id, local_slot = self._global_to_box_slot(idx) + if box_id >= 0: + return 4 + box_id * 4 + int(local_slot) + return idx + return idx + + def _slot_usable_for_print(self, global_index: int) -> bool: + """Whether a global slot can be used for current filament mode.""" + slot = next((s for s in self._ams_slots if int(s.get("global_index", -1)) == int(global_index)), None) + if not slot: + return False + if int(slot.get("status", 0)) != 5: + return False + + box_id = int(slot.get("box_id", -1)) + if self._filament_mode == "ace_hub": + # In hub mode, toolhead channels (0..2) and ACE channels are both printable. + return box_id == -1 or box_id >= 0 + if self._filament_mode == "ace_direct": + return box_id >= 0 + return box_id == -1 + + def _loaded_slots_for_print(self) -> list[tuple[int, dict]]: + """Loaded slots filtered for current filament mode.""" + loaded = [ + (int(s.get("global_index", i)), s) + for i, s in enumerate(self._ams_slots) + if s.get("status") == 5 and self._slot_usable_for_print(int(s.get("global_index", i))) + ] + return loaded + + def _select_loaded_slots_for_print(self, warn_on_empty_default: bool = False) -> list[tuple[int, dict]]: + """Return loaded slots, honoring default_ams_slot when configured.""" + default_slot = getattr(self._args, "default_ams_slot", "auto") + all_loaded = self._loaded_slots_for_print() + if default_slot == "auto": + return all_loaded + + try: + slot_idx = int(default_slot) + except ValueError: + return all_loaded + + selected = [(i, s) for i, s in all_loaded if i == slot_idx] + if selected: + return selected + + if warn_on_empty_default: + log.warning(f"Default slot {slot_idx} is empty - falling back to auto") + return all_loaded + + @staticmethod + def _slot_color_rgba(slot: dict) -> list[int]: + color = slot.get("color", [255, 255, 255]) + if isinstance(color, list) and len(color) >= 3: + return [int(color[0]), int(color[1]), int(color[2]), 255] + return [255, 255, 255, 255] + + def _build_auto_ams_box_mapping( + self, + warn_on_empty_default: bool = False, + loaded_slots: list[tuple[int, dict]] | None = None, + ) -> list[dict]: + """Build print mapping from currently loaded slots (no explicit dialog assignments).""" + loaded = loaded_slots + if loaded is None: + loaded = self._select_loaded_slots_for_print(warn_on_empty_default=warn_on_empty_default) + if not loaded: + return [] + loaded_map = {gidx: s for gidx, s in loaded} + max_idx = max(loaded_map.keys()) + # The printer interprets ams_box_mapping as an ordered list (entry N = TN). + # Missing slots must be inserted as placeholders, otherwise everything shifts. + # A placeholder must NOT reference a physically empty tray: the printer + # rejects such an entry even for a tool the GCode never calls (printing + # Filament 4 with the slot below it empty fails; all-full works). Point + # gap placeholders at a definitely-loaded tray instead of the gap's own + # (empty) index. + fallback_gidx = max_idx # highest loaded slot -> loaded + printable + fallback_slot = loaded_map[fallback_gidx] + fallback_ams = self._slot_to_print_ams_index(fallback_gidx) + result = [] + for i in range(max_idx + 1): + if i in loaded_map: + s = loaded_map[i] + result.append({ + "paint_index": i, + "ams_index": self._slot_to_print_ams_index(i), + "paint_color": [255, 255, 255, 255], + "ams_color": self._slot_color_rgba(s), + "material_type": s.get("type", "PLA"), + }) + else: + result.append({ + "paint_index": i, + "ams_index": fallback_ams, + "paint_color": [255, 255, 255, 255], + "ams_color": self._slot_color_rgba(fallback_slot), + "material_type": fallback_slot.get("type", "PLA"), + }) + return result + + def _build_assigned_ams_box_mapping(self, assignments: list) -> tuple[list[dict], int, int]: + """Build print mapping from UI filament assignments. + + Returns (mapping, unused_count, invalid_count). + """ + slot_by_global_index = { + int(s.get("global_index", i)): s + for i, s in enumerate(self._ams_slots) + } + ams_box_mapping: list[dict] = [] + unused_count = 0 + invalid_count = 0 + + for i, a in enumerate(assignments): + try: + if a.get("is_used") is False: + unused_count += 1 + continue + global_slot = int(a["slot_index"]) + except (ValueError, TypeError, KeyError): + invalid_count += 1 + continue + + if global_slot < 0: + unused_count += 1 + continue + if not self._slot_usable_for_print(global_slot): + invalid_count += 1 + continue + + slot = slot_by_global_index.get(global_slot, {}) + ams_box_mapping.append({ + # Preserve slicer paint indices (can be sparse when paint 0 is unused). + "paint_index": a.get("paint_index", i), + "ams_index": self._slot_to_print_ams_index(global_slot), + "paint_color": a.get("paint_color", [255, 255, 255, 255]), + "ams_color": self._slot_color_rgba(slot), + "material_type": slot.get("type", a.get("material", "PLA")), + }) + + return ams_box_mapping, unused_count, invalid_count + + def _box_local_to_global(self, box_id: int, local_slot: int, boxes: list) -> int: + """Convert (box_id, local slot) to global slot index for current topology.""" + if box_id == -1: + return local_slot + if self._filament_mode == "ace_direct": + # Multi-ACE (Issue #95): each unit occupies its own block of 4. + # Identical to the old `return local_slot` for a single unit (id 0). + return box_id * 4 + local_slot + return 3 + box_id * 4 + local_slot + + def _slot_activity_map(self, boxes: list, global_loaded: int = -1) -> dict: + """Build {global_slot_index: loading|unloading} from feed_status data.""" + # Note: all boxes are considered — the old primary_ace_id filter (skip + # every ACE box except the first in ace_direct mode) is gone since the + # slot aggregation now handles multiple ACE units (Issue #95). + activity: dict = {} + for box in boxes: + fs = box.get("feed_status") or {} + current_status = int(fs.get("current_status", -1)) + local_slot = int(fs.get("slot_index", -1)) + feed_type = int(fs.get("type", -1)) + if current_status in (-1, 10, 11) or local_slot < 0: + continue + box_slots = box.get("slots") or [] + if local_slot >= len(box_slots) or (box_slots[local_slot] or {}).get("status") != 5: + continue + if feed_type == 1: + act = "loading" + elif feed_type == 2: + act = "unloading" + else: + continue + global_slot = self._box_local_to_global(int(box.get("id", -1)), local_slot, boxes) + if feed_type == 1 and self._pending_load_slot >= 0 and global_slot != self._pending_load_slot: + # Ignore transient firmware-reported loading slots that differ from the requested target. + if global_loaded >= 0 and global_loaded != self._pending_load_slot: + activity[global_loaded] = "unloading" + continue + if feed_type == 1 and global_loaded >= 0 and global_slot != global_loaded: + # During a slot swap the firmware reports the target slot immediately, + # while the previously loaded slot is still being unloaded first. + activity[global_loaded] = "unloading" + activity[global_slot] = act + return activity + + def _on_multicolor_box(self, payload: dict): + if payload.get("state") == "failed": + req = getattr(self, "_last_ams_set_request", None) + log.warning( + f"multiColorBox setInfo rejected by printer: request={req} raw_response={payload.get('data')}" + ) + self._state["last_ams_set_error"] = True + return + data = payload.get("data") or {} + if not isinstance(data, dict): + log.warning(f"multiColorBox/report: unexpected data shape: {data!r}") + return + boxes = data.get("multi_color_box") or [] + if not boxes: + return + self._state["last_ams_set_error"] = False + self._head_tools_model = int(data.get("head_tools_model", self._head_tools_model)) + self._filament_mode = self._detect_filament_mode(boxes, self._head_tools_model) + self._state["filament_mode"] = self._filament_mode + + global_slots, global_loaded = self._aggregate_slots(boxes, self._filament_mode) + self._ams_loaded_slot = global_loaded + self._update_ace_drying_state(data, boxes) + for box in boxes: + bid = int(box.get("id", -1)) + if 0 <= bid <= 3 and "auto_feed" in box: + self._ace_auto_feed[bid] = int(box["auto_feed"]) + if self._pending_load_slot >= 0 and global_loaded == self._pending_load_slot: + self._pending_load_slot = -1 + activity_map = self._slot_activity_map(boxes, global_loaded) + for s in global_slots: + s["activity"] = activity_map.get(s.get("global_index"), "") + + # Tip forming: after feed-in (status=10) or feed-out (status=11) + # the original slicer automatically sends type=3 (extruder retract). + # Check ALL boxes so ACE-triggered events are handled correctly. + for box in boxes: + fs = box.get("feed_status") or {} + current_status = fs.get("current_status") + slot_index = fs.get("slot_index", 0) + box_id = box.get("id", -1) + if current_status in (10, 11): + def _tip_form(bi=box_id, si=slot_index, cs=current_status): + import time; time.sleep(2) + self.client.publish( + "multiColorBox", "feedFilament", + {"multi_color_box": [{"id": bi, "feed_status": {"slot_index": si, "type": 3}}]}, + timeout=0 + ) + log.info(f"Tip forming (type=3) after status={cs} box={bi} slot={si}") + threading.Thread(target=_tip_form, daemon=True).start() + + if global_slots: + self._ams_slots = global_slots + log.info(f"AMS slots received: {len(global_slots)}, loaded_slot={self._ams_loaded_slot}") + self._push_status_update() + + def _update_ace_drying_state(self, data: dict, boxes: list): + """Extract ACE drying state from multiColorBox report/getInfo payloads.""" + ace_ids = sorted({int(b.get("id", -1)) for b in boxes if int(b.get("id", -1)) >= 0}) + self._ace_box_ids = [i for i in ace_ids if 0 <= i <= 3] + + def _num_from(src: dict, keys: tuple[str, ...], default=None): + for k in keys: + v = src.get(k) + if v is not None: + try: + return float(v) + except Exception: + return default + return default + + def _humidity_from(src: dict, default=None): + return _num_from(src, ("humidity", "current_humidity", "cur_humidity", "relative_humidity", "humidity_value"), default) + + def _current_temp_from(src: dict, default=None): + return _num_from(src, ("current_temp", "cur_temp", "temperature", "temp", "drying_temp", "chamber_temp"), default) + + def _minutes_from(src: dict, key: str, default=0): + raw = src.get(key, default) + try: + value = int(float(raw)) + except Exception: + return int(default) + # Some firmware payloads report dryer times in seconds while the UI uses minutes. + if value > (24 * 60): + return max(0, int(round(value / 60.0))) + return max(0, value) + + per_unit: list[dict] = [] + for box in boxes: + bid = int(box.get("id", -1)) + if bid < 0: + continue + + bs = box.get("drying_status") or box.get("drying_settings") + bs = bs if isinstance(bs, dict) else {} + hu = _humidity_from(bs, _humidity_from(box)) + ct = _current_temp_from(bs, _current_temp_from(box)) + + if bs or hu is not None or ct is not None: + per_unit.append({ + "id": bid, + "status": int(bs.get("status", 0)), + "target_temp": int(bs.get("target_temp", 0)), + "duration": _minutes_from(bs, "duration", 0), + "remain_time": _minutes_from(bs, "remain_time", 0), + "humidity": hu, + "current_temp": ct, + }) + + src = data.get("drying_status") or data.get("drying_settings") + if not isinstance(src, dict): + for box in boxes: + if int(box.get("id", -1)) < 0: + continue + cand = box.get("drying_status") or box.get("drying_settings") + if isinstance(cand, dict): + src = cand + break + + if isinstance(src, dict): + cur = self._state.get("ace_drying") or {} + active = [u for u in per_unit if u.get("status", 0)] + primary = active[0] if active else (per_unit[0] if per_unit else {}) + self._state["ace_drying"] = { + "status": int(src.get("status", cur.get("status", 0))), + "target_temp": int(src.get("target_temp", cur.get("target_temp", 0))), + "duration": _minutes_from(src, "duration", cur.get("duration", 0)), + "remain_time": _minutes_from(src, "remain_time", cur.get("remain_time", 0)), + "humidity": _humidity_from(src, primary.get("humidity", cur.get("humidity"))), + "current_temp": _current_temp_from(src, primary.get("current_temp", cur.get("current_temp"))), + "units": per_unit, + } + elif per_unit: + active = [u for u in per_unit if u.get("status", 0)] + primary = active[0] if active else per_unit[0] + self._state["ace_drying"] = { + "status": int(primary.get("status", 0)), + "target_temp": int(primary.get("target_temp", 0)), + "duration": int(primary.get("duration", 0)), + "remain_time": int(primary.get("remain_time", 0)), + "humidity": primary.get("humidity"), + "current_temp": primary.get("current_temp"), + "units": per_unit, + } + + def _on_light(self, payload: dict): + d = payload.get("data") or {} + self._state["light_on"] = bool(d.get("status", 0)) + self._state["light_brightness"] = int(d.get("brightness", 80)) + self._push_status_update() + + # OrcaSlicer filament preset IDs (MoonrakerPrinterAgent.cpp mapping) + # Default mapping per material type when the user has not set a slot + # profile override. For the Kobra X we prefer Anycubic's own + # filament IDs from the `@Anycubic Kobra X 0.4 nozzle` profiles - those + # are printer-specific is_compatible and are picked up by OrcaSlicer directly + # matched. Library fallbacks (OGF*) only for material types without + # Kobra X-specific Anycubic profile - their @system profiles have + # `compatible_printers: []` (= compatible with all printers). + _TRAY_INFO_IDX = { + # Anycubic-eigene Kobra-X-Profile + "PLA": "GFPLA", + "PLA+": "GFPLA+", + "PLA SILK": "GFPLA Silk", + "PLA-SILK": "GFPLA Silk", + "PLASILK": "GFPLA Silk", + "SILK PLA": "GFPLA Silk", + "PLA MATTE": "GFPLA", + "PLA-MATTE": "GFPLA", + "PLA MARBLE": "GFPLA", + "PLA WOOD": "GFPLA", + "PETG": "GFPETG", + "PETG+": "GFPETG", + "ABS": "GFABS", + "ASA": "GFASA", + "TPU": "GFTPU 95A", + "TPE": "GFTPU 95A", + "PVA": "GFPVA", + # Kein Anycubic-Kobra-X-Profil → Library-Fallback + "PLA-CF": "OGFL98", + "PLA CF": "OGFL98", + "PETG-CF": "OGFG98", + "PETG CF": "OGFG98", + "PA": "OGFN99", + "PA-CF": "OGFN98", + "PA CF": "OGFN98", + "PC": "OGFC99", + "HIPS": "OGFS98", + } + + # Normalizes material type strings to the canonical key for _TRAY_INFO_IDX + # and _default_filament_name. PLA variants without an exact match fall + # back to their base family (PLA+ -> PLA+, PLA Matte -> PLA, etc.). + @staticmethod + def _normalize_material(mat: str) -> str: + m = mat.upper().strip().replace("-", " ").replace("_", " ") + # Bekannte Varianten normalisieren + _ALIASES = { + "PLAPLUS": "PLA+", "PLA PLUS": "PLA+", + "SILK PLA": "PLA SILK", "PLASILK": "PLA SILK", + "PLA MATTE": "PLA MATTE", "PLA MARBLE": "PLA MARBLE", + "PLA WOOD": "PLA WOOD", + "TPE": "TPU", + "PETG PLUS": "PETG+", + "PA6": "PA", "PA12": "PA", "PA66": "PA", + } + if m in _ALIASES: + return _ALIASES[m] + return m + + @staticmethod + def _material_family(mat: str) -> str: + """Reduce a material to its base polymer family. + + PLA / PLA+ / PLA SILK / PLA MATTE -> "PLA"; PETG / PETG+ -> "PETG"; etc. + Used by the stale-profile guard: only a change of *family* (e.g. PETG -> + PLA) invalidates a saved slot profile — a change within the family + (PLA -> PLA SILK) must not discard an otherwise valid profile. + """ + if not mat: + return "" + m = AmsFilamentMixin._normalize_material(mat) + # Longer prefixes first so "PETG" is not swallowed by "PET". + for fam in ("PETG", "PLA", "ABS", "ASA", "TPU", "PVA", "HIPS", "PA", "PC", "PET"): + if m.startswith(fam): + return fam + return m + + def _parse_combined_rfid_type(self, raw_type: str) -> tuple[str, str]: + """Split a combined ACE-RFID "VENDOR TYPE SERIAL" string (e.g. + "GEEETECH PLA Bas", written via third-party RFID tools) into + (vendor, material_family). + + Anycubic's ACE RFID system concatenates vendor + material + a + truncated serial/variant into one `type` string for custom tags - + unlike a normal spool report where `type` is just "PLA"/"PETG"/etc. + Returns ("", "") when the first token isn't a known vendor (from the + merged system+user filament library), which leaves plain type + strings like "PLA" completely unaffected (Issue #101). + """ + tokens = raw_type.split() + if len(tokens) < 2: + return "", "" + first = tokens[0].strip().lower() + vendors = {p.get("vendor", "").lower(): p.get("vendor", "") for p in self._load_orca_filaments()} + vendor = vendors.get(first) + if not vendor: + return "", "" + family = self._material_family(" ".join(tokens[1:])) + if not family: + return "", "" + return vendor, family + + @staticmethod + def _rfid_variant_tokens(raw_type: str) -> list[str]: + """Tokens after "VENDOR TYPE" in a combined ACE-RFID string (e.g. + ["bas"] for "GEEETECH PLA Bas") - the truncated variant/serial that + distinguishes multiple profiles of the same (vendor, material family), + e.g. "Basic" vs. "Matte". Kept separate from _parse_combined_rfid_type() + so that function's 2-tuple signature (and its existing callers/tests) + stay unchanged (Issue #101).""" + tokens = raw_type.split() + return [t.lower() for t in tokens[2:]] + + def _match_profile_by_vendor_family(self, vendor: str, family: str, + variant_tokens: list[str] | None = None) -> dict: + """Find an imported/system filament profile by (vendor, material + family) - used to auto-resolve a combined ACE-RFID type string to + the user's already-imported OrcaSlicer profile (Issue #101), since + the exact profile `name` never appears verbatim in the truncated + RFID string. + + When multiple profiles share the same (vendor, family) - e.g. "Geeetech + PLA Basic" and "Geeetech PLA Matte" both matching (Geeetech, PLA) - + variant_tokens (the RFID string's remaining tokens, e.g. ["bas"] for + "Basic") are scored against each candidate's name: a word-prefix match + scores higher than a plain substring match, so "bas" prefers "Basic" + over "Matte" or an unrelated profile name containing "bas" as noise. + Falls back to the first match when nothing disambiguates.""" + matches = [ + p for p in self._load_orca_filaments() + if p.get("vendor", "").lower() == vendor.lower() + and self._material_family(p.get("type", "")) == family + ] + if not matches: + return {} + if len(matches) == 1 or not variant_tokens: + return matches[0] + + best = matches[0] + best_score = -1 + for p in matches: + name_words = p.get("name", "").lower().split() + score = 0 + for tok in variant_tokens: + if any(w.startswith(tok) for w in name_words): + score += 2 + elif tok in p.get("name", "").lower(): + score += 1 + if score > best_score: + best_score = score + best = p + log.debug( + f"_match_profile_by_vendor_family: {len(matches)} profiles match " + f"vendor={vendor!r} family={family!r}, variant_tokens={variant_tokens!r} " + f"-> {best.get('name')!r} (score={best_score})" + ) + return best + + def _profile_material(self, profile: dict) -> str: + """Material type (e.g. "PETG") of a saved slot profile, resolved by + (vendor, name) from the Orca filament library. Returns "" when the + profile is not in the library — we do NOT guess in that case.""" + name = (profile or {}).get("name", "") + if not name: + return "" + vendor = profile.get("vendor", "") + for p in self._load_orca_filaments(): + if p.get("vendor") == vendor and p.get("name") == name: + return p.get("type", "") or "" + return "" + + def _effective_slot_profile(self, global_idx: int, ams_material: str) -> dict: + """Saved slot-profile override — but only while its material *family* + still matches the material currently loaded in the AMS. Falls back to + auto-resolving a combined ACE-RFID type string (Issue #101) when there + is no (usable) manual override. + + Non-destructive suppression (Option A): when the family no longer matches + (e.g. a PETG profile but PLA loaded) the override is skipped → falls + through to the RFID auto-match / generic default. The override stays in + config.ini and reactivates as soon as the matching material is loaded + again. When the profile's family is unknown we do NOT suppress (fail-safe). + + Centralized here (rather than duplicated per caller) so every consumer - + the dashboard's /kx/filament/slots, Happy-Hare gate data, and the + OrcaSlicer lane-data sync - benefits from RFID auto-matching identically, + instead of only the one call site that happened to also call + _parse_combined_rfid_type() directly.""" + # A combined ACE-RFID string ("GEEETECH PLA Bas") carries a vendor + # prefix that _material_family() alone can't see past (it only + # strips known polymer prefixes, so "GEEETECH PLA BAS" resolves to + # itself, not "PLA") - resolve the plain material family through the + # RFID parser first so the stale-profile guard below compares against + # the actual polymer family, not the raw combined string. + vendor, family = self._parse_combined_rfid_type(ams_material) + plain_material = family or ams_material + + profile = self._filament_profiles.get(global_idx) or {} + if profile.get("name"): + prof_fam = self._material_family(self._profile_material(profile)) + ams_fam = self._material_family(plain_material) + if not (prof_fam and ams_fam and prof_fam != ams_fam): + return profile + + if vendor: + variant_tokens = self._rfid_variant_tokens(ams_material) + auto = self._match_profile_by_vendor_family(vendor, family, variant_tokens) + if auto.get("name"): + return auto + + return {} + + def _build_lane_data(self) -> dict: + """Builds BBL AMS JSON for OrcaSlicer DevFilaSystemParser::ParseV1_0. + + POSITION-FAITHFUL: every physical slot keeps its position (tray id = + slot position). Empty slots are reported as placeholder trays, NOT + filtered out/compacted - otherwise colors shift to wrong positions + (e.g. slot 1=yellow, 2=empty, 3=red -> red must not land on position 2). + """ + slots = self._ams_slots + total = len(slots) + if total == 0: + return {"ams": [], "ams_exist_bits": "0", "tray_exist_bits": "0"} + + ams_count = (total + 3) // 4 + ams_exist_bits = 0 + tray_exist_bits = 0 + ams_array = [] + + for ams_id in range(ams_count): + ams_exist_bits |= (1 << ams_id) + tray_array = [] + max_slot = min(3, total - ams_id * 4 - 1) + for slot_id in range(max_slot + 1): + slot_index = ams_id * 4 + slot_id + slot = slots[slot_index] if slot_index < total else {} + occupied = slot.get("status") == 5 + + if occupied: + tray_exist_bits |= (1 << slot_index) + color_raw = slot.get("color", [255, 255, 255]) + if isinstance(color_raw, list) and len(color_raw) >= 3: + color_hex = "{:02X}{:02X}{:02X}FF".format( + int(color_raw[0]), int(color_raw[1]), int(color_raw[2]) + ) + elif isinstance(color_raw, str) and len(color_raw) >= 6: + color_hex = color_raw[:6].upper() + "FF" + else: + color_hex = "FFFFFFFF" + material = self._normalize_material(slot.get("type", "PLA")) + # User override from config.ini [filament_profiles].slot_N_id + # takes precedence over the default mapping by material type. + # The vendor is sent along (tray_sub_brands + filament_vendor), + # so a patched OrcaSlicer can match by brand + type + + # color (analogous to SnapmakerPrinterAgent). + # Three-layer resolution for the filament hint sent to OrcaSlicer, + # all handled inside _effective_slot_profile() (Issue #101): + # 1. User-Wahl (config.ini [filament_profiles]) — exakte Kontrolle + # 2. Combined ACE-RFID "VENDOR TYPE SERIAL" string (e.g. + # "GEEETECH PLA Bas") auto-matched against the user's + # already-imported profile library. Not persisted to + # config.ini - re-derives on every call, so a differently + # tagged spool loaded later isn't stuck with a stale match. + # 3. Generic fallback (_TRAY_INFO_IDX) per material type - no + # vendor hint; OrcaSlicer then picks its own generic preset + user_profile = self._effective_slot_profile(slot_index, material) + if user_profile.get("name"): + material = self._material_family(user_profile.get("type", material)) or material + vendor = user_profile.get("vendor", "") + fila_name = user_profile.get("name", "") + tray_info_idx = user_profile.get("id") or self._TRAY_INFO_IDX.get(material, "OGFL99") + else: + # Default: Library-Generic-Profil (siehe _default_filament_name) — + # is compatible with all printers and guaranteed to be visible. + # The user deliberately picks a concrete brand per slot if they + # want one; the default stays neutral. + fila_name = self._default_filament_name(material) + vendor = "Generic" if fila_name.startswith("Generic ") else "" + tray_info_idx = self._lookup_filament_id(vendor, fila_name) or self._TRAY_INFO_IDX.get(material, "OGFL99") + tray_array.append({ + "id": str(slot_id), + "tag_uid": "0000000000000000", + "tray_info_idx": tray_info_idx, + "tray_type": material, + "tray_color": color_hex, + "tray_sub_brands": vendor, + # OrcaSlicer-Empfangs-Patch PR #13719 erwartet `name` + + # `vendor_name` pro Lane (Stufen-Matching: Vendor+Name → Name → + # filament_id_by_type). We send both spellings so that + # older patch variants + future upstream PRs are both + # covered. + "name": fila_name, + "vendor_name": vendor, + # Aliases for older patch variants (variant 2, + # MoonrakerPrinterAgent.cpp): filament_id direkt (exakt), + # otherwise resolve the preset name via find_preset(). + "filament_id": tray_info_idx, + "filament_vendor": vendor, + "filament_name": fila_name, + "preset": fila_name, + }) + else: + tray_array.append({ + "id": str(slot_id), + "tag_uid": "0000000000000000", + "tray_info_idx": "", + "tray_type": "", + "tray_color": "00000000", + "tray_slot_placeholder": "1", + }) + + ams_array.append({"id": str(ams_id), "info": "0002", "tray": tray_array}) + + return { + "ams": ams_array, + "ams_exist_bits": format(ams_exist_bits, "X"), + "tray_exist_bits": format(tray_exist_bits, "X"), + } diff --git a/kobrax_moonraker_bridge.py b/kobrax_moonraker_bridge.py index 98c398a..7ff2a39 100644 --- a/kobrax_moonraker_bridge.py +++ b/kobrax_moonraker_bridge.py @@ -69,6 +69,7 @@ from camera import CameraCache, _find_ffmpeg from credentials import _kx_fetch_credentials, _kx_generate_signature, _kx_decrypt_info from bridge_spoolman import SpoolmanMixin from bridge_mqtt import MqttCallbacksMixin +from bridge_ams import AmsFilamentMixin try: @@ -116,7 +117,7 @@ from bridge_logging import ( from bridge_constants import KOBRA_TO_KLIPPER_STATE, MOONRAKER_VERSION, KLIPPER_VERSION -class KobraXBridge(SpoolmanMixin, MqttCallbacksMixin): +class KobraXBridge(SpoolmanMixin, MqttCallbacksMixin, AmsFilamentMixin): def __init__(self, client: KobraXClient, args=None, store=None, printer_id: str = "1", all_bridges=None): self.client = client self._args = args @@ -284,845 +285,6 @@ class KobraXBridge(SpoolmanMixin, MqttCallbacksMixin): log.info(f"Spoolman: {'OK' if ok else 'unreachable'} at {self._spoolman.server_url}") threading.Thread(target=_check, daemon=True, name="spoolman-health").start() - def _default_ace_dry_presets(self) -> dict[str, dict]: - return { - "pla": {"temp": 45, "duration_sec": 4 * 3600}, - "pla_plus": {"temp": 45, "duration_sec": 4 * 3600}, - "petg": {"temp": 50, "duration_sec": 4 * 3600}, - "tpu": {"temp": 55, "duration_sec": 4 * 3600}, - "abs_asa": {"temp": 45, "duration_sec": 8 * 3600}, - "pa_pc": {"temp": 55, "duration_sec": 12 * 3600}, - "custom_1": {"name": "Custom 1", "temp": 45, "duration_sec": 4 * 3600}, - "custom_2": {"name": "Custom 2", "temp": 45, "duration_sec": 4 * 3600}, - "custom_3": {"name": "Custom 3", "temp": 45, "duration_sec": 4 * 3600}, - } - - def _sanitize_ace_dry_presets(self, presets: dict) -> dict[str, dict]: - out = self._default_ace_dry_presets() - for key in list(out.keys()): - src = presets.get(key) if isinstance(presets, dict) else None - if not isinstance(src, dict): - continue - try: - t = int(src.get("temp", out[key]["temp"])) - except Exception: - t = out[key]["temp"] - try: - d = int(src.get("duration_sec", out[key]["duration_sec"])) - except Exception: - d = out[key]["duration_sec"] - out[key]["temp"] = max(30, min(80, t)) - out[key]["duration_sec"] = max(10 * 60, min(24 * 3600, d)) - if key.startswith("custom_"): - name = str(src.get("name", out[key].get("name", key.replace("_", " ").title()))).strip() - out[key]["name"] = name or out[key].get("name", "Custom") - return out - - def _load_ace_dry_presets_config(self) -> dict[str, dict]: - import configparser - defaults = self._default_ace_dry_presets() - cfg_path = self._find_config_path() - if not cfg_path.is_file(): - return defaults - cfg = configparser.ConfigParser(interpolation=None) - cfg.read(cfg_path, encoding="utf-8") - sec = "ace_dry_presets" - if not cfg.has_section(sec): - return defaults - out = {} - for key, d in defaults.items(): - temp_k = f"{key}_temp" - dur_k = f"{key}_duration_sec" - try: - temp = int(cfg.get(sec, temp_k, fallback=str(d["temp"]))) - except Exception: - temp = d["temp"] - try: - dur = int(cfg.get(sec, dur_k, fallback=str(d["duration_sec"]))) - except Exception: - dur = d["duration_sec"] - out[key] = { - "temp": max(30, min(80, temp)), - "duration_sec": max(10 * 60, min(24 * 3600, dur)), - } - if key.startswith("custom_"): - name_k = f"{key}_name" - name = cfg.get(sec, name_k, fallback=str(d.get("name", key.replace("_", " ").title()))).strip() - out[key]["name"] = name or str(d.get("name", "Custom")) - return out - - @staticmethod - def _detect_filament_mode(boxes: list, head_tools_model: int = -1) -> str: - """Detect active filament topology mode. - - Modes: - - toolhead: only toolhead slots - - ace_direct: ACE channels directly mapped, no toolhead box present. - Covers one unit (Kobra X) as well as multiple daisy-chained units - (Kobra S1 with 2+ ACE Pro, Issue #95) — each unit contributes a - block of 4 global slots at box_id * 4. - - ace_hub: toolhead + ACE via hub (slot 4 as hub path) - """ - toolhead = any(b.get("id") == -1 for b in boxes) - ace = any(b.get("id", -1) >= 0 for b in boxes) - if ace and toolhead: - return "ace_hub" - if ace: - return "ace_direct" - return "toolhead" - - @staticmethod - def _aggregate_slots(boxes: list, mode: str = "toolhead") -> tuple: - """Aggregate multi_color_box list into a flat global slot list.""" - toolhead = next((b for b in boxes if b.get("id") == -1), None) - ace_boxes = sorted( - [b for b in boxes if b.get("id", -1) >= 0], - key=lambda b: b["id"] - ) - - global_slots: list = [] - global_loaded: int = -1 - - if mode == "toolhead": - if toolhead: - for local_idx, s in enumerate(toolhead.get("slots") or []): - s = dict(s) - s["global_index"] = local_idx - s["box_id"] = -1 - global_slots.append(s) - loaded = toolhead.get("loaded_slot", -1) - if loaded >= 0: - global_loaded = loaded - return global_slots, global_loaded - - if mode == "ace_direct": - # One or more ACE units, no toolhead buffer (Kobra X: 1 unit, - # Kobra S1: up to 2+ units, Issue #95). Global index = - # box_id * 4 + local slot, so the numbering matches - # _global_to_box_slot's //4-%4 fallback and stays stable - # regardless of report order. - for ace in ace_boxes: - ace_id = int(ace["id"]) - base = ace_id * 4 - for local_idx, s in enumerate((ace.get("slots") or [])[:4]): - s = dict(s) - s["global_index"] = base + local_idx - s["box_id"] = ace_id - global_slots.append(s) - ace_loaded = ace.get("loaded_slot", -1) - if 0 <= ace_loaded < 4: - global_loaded = base + ace_loaded - return global_slots, global_loaded - - # ace_hub - if toolhead: - for local_idx, s in enumerate((toolhead.get("slots") or [])[:3]): - s = dict(s) - s["global_index"] = local_idx - s["box_id"] = -1 - global_slots.append(s) - th_loaded = toolhead.get("loaded_slot", -1) - if 0 <= th_loaded <= 2: - global_loaded = th_loaded - - for ace in ace_boxes: - ace_id = ace["id"] - base = 3 + ace_id * 4 - for local_idx, s in enumerate(ace.get("slots") or []): - s = dict(s) - s["global_index"] = base + local_idx - s["box_id"] = ace_id - global_slots.append(s) - ace_loaded = ace.get("loaded_slot", -1) - if ace_loaded >= 0: - global_loaded = base + ace_loaded - - return global_slots, global_loaded - - def _global_to_box_slot(self, global_index: int) -> tuple: - """Convert a global slot index to (box_id, local_slot_index).""" - for s in self._ams_slots: - if s.get("global_index") == global_index: - return s.get("box_id", -1), s.get("index", global_index) - - ace_present = any(s.get("box_id", -1) >= 0 for s in self._ams_slots) - if self._filament_mode == "ace_direct" and ace_present: - return global_index // 4, global_index % 4 - if not ace_present or global_index < 3: - return -1, global_index - offset = global_index - 3 - return offset // 4, offset % 4 - - def _slot_to_print_ams_index(self, global_index: int) -> int: - """Convert UI/global slot index to printer print/start ams_index. - - In ace_hub mode, print/start uses global channel numbering where - toolhead channels occupy 1..3 and ACE0 starts at index 4. - """ - idx = int(global_index) - if self._filament_mode == "ace_hub": - box_id, local_slot = self._global_to_box_slot(idx) - if box_id >= 0: - return 4 + box_id * 4 + int(local_slot) - return idx - return idx - - def _slot_usable_for_print(self, global_index: int) -> bool: - """Whether a global slot can be used for current filament mode.""" - slot = next((s for s in self._ams_slots if int(s.get("global_index", -1)) == int(global_index)), None) - if not slot: - return False - if int(slot.get("status", 0)) != 5: - return False - - box_id = int(slot.get("box_id", -1)) - if self._filament_mode == "ace_hub": - # In hub mode, toolhead channels (0..2) and ACE channels are both printable. - return box_id == -1 or box_id >= 0 - if self._filament_mode == "ace_direct": - return box_id >= 0 - return box_id == -1 - - def _loaded_slots_for_print(self) -> list[tuple[int, dict]]: - """Loaded slots filtered for current filament mode.""" - loaded = [ - (int(s.get("global_index", i)), s) - for i, s in enumerate(self._ams_slots) - if s.get("status") == 5 and self._slot_usable_for_print(int(s.get("global_index", i))) - ] - return loaded - - def _select_loaded_slots_for_print(self, warn_on_empty_default: bool = False) -> list[tuple[int, dict]]: - """Return loaded slots, honoring default_ams_slot when configured.""" - default_slot = getattr(self._args, "default_ams_slot", "auto") - all_loaded = self._loaded_slots_for_print() - if default_slot == "auto": - return all_loaded - - try: - slot_idx = int(default_slot) - except ValueError: - return all_loaded - - selected = [(i, s) for i, s in all_loaded if i == slot_idx] - if selected: - return selected - - if warn_on_empty_default: - log.warning(f"Default slot {slot_idx} is empty - falling back to auto") - return all_loaded - - @staticmethod - def _slot_color_rgba(slot: dict) -> list[int]: - color = slot.get("color", [255, 255, 255]) - if isinstance(color, list) and len(color) >= 3: - return [int(color[0]), int(color[1]), int(color[2]), 255] - return [255, 255, 255, 255] - - def _build_auto_ams_box_mapping( - self, - warn_on_empty_default: bool = False, - loaded_slots: list[tuple[int, dict]] | None = None, - ) -> list[dict]: - """Build print mapping from currently loaded slots (no explicit dialog assignments).""" - loaded = loaded_slots - if loaded is None: - loaded = self._select_loaded_slots_for_print(warn_on_empty_default=warn_on_empty_default) - if not loaded: - return [] - loaded_map = {gidx: s for gidx, s in loaded} - max_idx = max(loaded_map.keys()) - # The printer interprets ams_box_mapping as an ordered list (entry N = TN). - # Missing slots must be inserted as placeholders, otherwise everything shifts. - # A placeholder must NOT reference a physically empty tray: the printer - # rejects such an entry even for a tool the GCode never calls (printing - # Filament 4 with the slot below it empty fails; all-full works). Point - # gap placeholders at a definitely-loaded tray instead of the gap's own - # (empty) index. - fallback_gidx = max_idx # highest loaded slot -> loaded + printable - fallback_slot = loaded_map[fallback_gidx] - fallback_ams = self._slot_to_print_ams_index(fallback_gidx) - result = [] - for i in range(max_idx + 1): - if i in loaded_map: - s = loaded_map[i] - result.append({ - "paint_index": i, - "ams_index": self._slot_to_print_ams_index(i), - "paint_color": [255, 255, 255, 255], - "ams_color": self._slot_color_rgba(s), - "material_type": s.get("type", "PLA"), - }) - else: - result.append({ - "paint_index": i, - "ams_index": fallback_ams, - "paint_color": [255, 255, 255, 255], - "ams_color": self._slot_color_rgba(fallback_slot), - "material_type": fallback_slot.get("type", "PLA"), - }) - return result - - def _build_assigned_ams_box_mapping(self, assignments: list) -> tuple[list[dict], int, int]: - """Build print mapping from UI filament assignments. - - Returns (mapping, unused_count, invalid_count). - """ - slot_by_global_index = { - int(s.get("global_index", i)): s - for i, s in enumerate(self._ams_slots) - } - ams_box_mapping: list[dict] = [] - unused_count = 0 - invalid_count = 0 - - for i, a in enumerate(assignments): - try: - if a.get("is_used") is False: - unused_count += 1 - continue - global_slot = int(a["slot_index"]) - except (ValueError, TypeError, KeyError): - invalid_count += 1 - continue - - if global_slot < 0: - unused_count += 1 - continue - if not self._slot_usable_for_print(global_slot): - invalid_count += 1 - continue - - slot = slot_by_global_index.get(global_slot, {}) - ams_box_mapping.append({ - # Preserve slicer paint indices (can be sparse when paint 0 is unused). - "paint_index": a.get("paint_index", i), - "ams_index": self._slot_to_print_ams_index(global_slot), - "paint_color": a.get("paint_color", [255, 255, 255, 255]), - "ams_color": self._slot_color_rgba(slot), - "material_type": slot.get("type", a.get("material", "PLA")), - }) - - return ams_box_mapping, unused_count, invalid_count - - def _box_local_to_global(self, box_id: int, local_slot: int, boxes: list) -> int: - """Convert (box_id, local slot) to global slot index for current topology.""" - if box_id == -1: - return local_slot - if self._filament_mode == "ace_direct": - # Multi-ACE (Issue #95): each unit occupies its own block of 4. - # Identical to the old `return local_slot` for a single unit (id 0). - return box_id * 4 + local_slot - return 3 + box_id * 4 + local_slot - - def _slot_activity_map(self, boxes: list, global_loaded: int = -1) -> dict: - """Build {global_slot_index: loading|unloading} from feed_status data.""" - # Note: all boxes are considered — the old primary_ace_id filter (skip - # every ACE box except the first in ace_direct mode) is gone since the - # slot aggregation now handles multiple ACE units (Issue #95). - activity: dict = {} - for box in boxes: - fs = box.get("feed_status") or {} - current_status = int(fs.get("current_status", -1)) - local_slot = int(fs.get("slot_index", -1)) - feed_type = int(fs.get("type", -1)) - if current_status in (-1, 10, 11) or local_slot < 0: - continue - box_slots = box.get("slots") or [] - if local_slot >= len(box_slots) or (box_slots[local_slot] or {}).get("status") != 5: - continue - if feed_type == 1: - act = "loading" - elif feed_type == 2: - act = "unloading" - else: - continue - global_slot = self._box_local_to_global(int(box.get("id", -1)), local_slot, boxes) - if feed_type == 1 and self._pending_load_slot >= 0 and global_slot != self._pending_load_slot: - # Ignore transient firmware-reported loading slots that differ from the requested target. - if global_loaded >= 0 and global_loaded != self._pending_load_slot: - activity[global_loaded] = "unloading" - continue - if feed_type == 1 and global_loaded >= 0 and global_slot != global_loaded: - # During a slot swap the firmware reports the target slot immediately, - # while the previously loaded slot is still being unloaded first. - activity[global_loaded] = "unloading" - activity[global_slot] = act - return activity - - def _on_multicolor_box(self, payload: dict): - if payload.get("state") == "failed": - req = getattr(self, "_last_ams_set_request", None) - log.warning( - f"multiColorBox setInfo rejected by printer: request={req} raw_response={payload.get('data')}" - ) - self._state["last_ams_set_error"] = True - return - data = payload.get("data") or {} - if not isinstance(data, dict): - log.warning(f"multiColorBox/report: unexpected data shape: {data!r}") - return - boxes = data.get("multi_color_box") or [] - if not boxes: - return - self._state["last_ams_set_error"] = False - self._head_tools_model = int(data.get("head_tools_model", self._head_tools_model)) - self._filament_mode = self._detect_filament_mode(boxes, self._head_tools_model) - self._state["filament_mode"] = self._filament_mode - - global_slots, global_loaded = self._aggregate_slots(boxes, self._filament_mode) - self._ams_loaded_slot = global_loaded - self._update_ace_drying_state(data, boxes) - for box in boxes: - bid = int(box.get("id", -1)) - if 0 <= bid <= 3 and "auto_feed" in box: - self._ace_auto_feed[bid] = int(box["auto_feed"]) - if self._pending_load_slot >= 0 and global_loaded == self._pending_load_slot: - self._pending_load_slot = -1 - activity_map = self._slot_activity_map(boxes, global_loaded) - for s in global_slots: - s["activity"] = activity_map.get(s.get("global_index"), "") - - # Tip forming: after feed-in (status=10) or feed-out (status=11) - # the original slicer automatically sends type=3 (extruder retract). - # Check ALL boxes so ACE-triggered events are handled correctly. - for box in boxes: - fs = box.get("feed_status") or {} - current_status = fs.get("current_status") - slot_index = fs.get("slot_index", 0) - box_id = box.get("id", -1) - if current_status in (10, 11): - def _tip_form(bi=box_id, si=slot_index, cs=current_status): - import time; time.sleep(2) - self.client.publish( - "multiColorBox", "feedFilament", - {"multi_color_box": [{"id": bi, "feed_status": {"slot_index": si, "type": 3}}]}, - timeout=0 - ) - log.info(f"Tip forming (type=3) after status={cs} box={bi} slot={si}") - threading.Thread(target=_tip_form, daemon=True).start() - - if global_slots: - self._ams_slots = global_slots - log.info(f"AMS slots received: {len(global_slots)}, loaded_slot={self._ams_loaded_slot}") - self._push_status_update() - - def _update_ace_drying_state(self, data: dict, boxes: list): - """Extract ACE drying state from multiColorBox report/getInfo payloads.""" - ace_ids = sorted({int(b.get("id", -1)) for b in boxes if int(b.get("id", -1)) >= 0}) - self._ace_box_ids = [i for i in ace_ids if 0 <= i <= 3] - - def _num_from(src: dict, keys: tuple[str, ...], default=None): - for k in keys: - v = src.get(k) - if v is not None: - try: - return float(v) - except Exception: - return default - return default - - def _humidity_from(src: dict, default=None): - return _num_from(src, ("humidity", "current_humidity", "cur_humidity", "relative_humidity", "humidity_value"), default) - - def _current_temp_from(src: dict, default=None): - return _num_from(src, ("current_temp", "cur_temp", "temperature", "temp", "drying_temp", "chamber_temp"), default) - - def _minutes_from(src: dict, key: str, default=0): - raw = src.get(key, default) - try: - value = int(float(raw)) - except Exception: - return int(default) - # Some firmware payloads report dryer times in seconds while the UI uses minutes. - if value > (24 * 60): - return max(0, int(round(value / 60.0))) - return max(0, value) - - per_unit: list[dict] = [] - for box in boxes: - bid = int(box.get("id", -1)) - if bid < 0: - continue - - bs = box.get("drying_status") or box.get("drying_settings") - bs = bs if isinstance(bs, dict) else {} - hu = _humidity_from(bs, _humidity_from(box)) - ct = _current_temp_from(bs, _current_temp_from(box)) - - if bs or hu is not None or ct is not None: - per_unit.append({ - "id": bid, - "status": int(bs.get("status", 0)), - "target_temp": int(bs.get("target_temp", 0)), - "duration": _minutes_from(bs, "duration", 0), - "remain_time": _minutes_from(bs, "remain_time", 0), - "humidity": hu, - "current_temp": ct, - }) - - src = data.get("drying_status") or data.get("drying_settings") - if not isinstance(src, dict): - for box in boxes: - if int(box.get("id", -1)) < 0: - continue - cand = box.get("drying_status") or box.get("drying_settings") - if isinstance(cand, dict): - src = cand - break - - if isinstance(src, dict): - cur = self._state.get("ace_drying") or {} - active = [u for u in per_unit if u.get("status", 0)] - primary = active[0] if active else (per_unit[0] if per_unit else {}) - self._state["ace_drying"] = { - "status": int(src.get("status", cur.get("status", 0))), - "target_temp": int(src.get("target_temp", cur.get("target_temp", 0))), - "duration": _minutes_from(src, "duration", cur.get("duration", 0)), - "remain_time": _minutes_from(src, "remain_time", cur.get("remain_time", 0)), - "humidity": _humidity_from(src, primary.get("humidity", cur.get("humidity"))), - "current_temp": _current_temp_from(src, primary.get("current_temp", cur.get("current_temp"))), - "units": per_unit, - } - elif per_unit: - active = [u for u in per_unit if u.get("status", 0)] - primary = active[0] if active else per_unit[0] - self._state["ace_drying"] = { - "status": int(primary.get("status", 0)), - "target_temp": int(primary.get("target_temp", 0)), - "duration": int(primary.get("duration", 0)), - "remain_time": int(primary.get("remain_time", 0)), - "humidity": primary.get("humidity"), - "current_temp": primary.get("current_temp"), - "units": per_unit, - } - - def _on_light(self, payload: dict): - d = payload.get("data") or {} - self._state["light_on"] = bool(d.get("status", 0)) - self._state["light_brightness"] = int(d.get("brightness", 80)) - self._push_status_update() - - # OrcaSlicer filament preset IDs (MoonrakerPrinterAgent.cpp mapping) - # Default mapping per material type when the user has not set a slot - # profile override. For the Kobra X we prefer Anycubic's own - # filament IDs from the `@Anycubic Kobra X 0.4 nozzle` profiles - those - # are printer-specific is_compatible and are picked up by OrcaSlicer directly - # matched. Library fallbacks (OGF*) only for material types without - # Kobra X-specific Anycubic profile - their @system profiles have - # `compatible_printers: []` (= compatible with all printers). - _TRAY_INFO_IDX = { - # Anycubic-eigene Kobra-X-Profile - "PLA": "GFPLA", - "PLA+": "GFPLA+", - "PLA SILK": "GFPLA Silk", - "PLA-SILK": "GFPLA Silk", - "PLASILK": "GFPLA Silk", - "SILK PLA": "GFPLA Silk", - "PLA MATTE": "GFPLA", - "PLA-MATTE": "GFPLA", - "PLA MARBLE": "GFPLA", - "PLA WOOD": "GFPLA", - "PETG": "GFPETG", - "PETG+": "GFPETG", - "ABS": "GFABS", - "ASA": "GFASA", - "TPU": "GFTPU 95A", - "TPE": "GFTPU 95A", - "PVA": "GFPVA", - # Kein Anycubic-Kobra-X-Profil → Library-Fallback - "PLA-CF": "OGFL98", - "PLA CF": "OGFL98", - "PETG-CF": "OGFG98", - "PETG CF": "OGFG98", - "PA": "OGFN99", - "PA-CF": "OGFN98", - "PA CF": "OGFN98", - "PC": "OGFC99", - "HIPS": "OGFS98", - } - - # Normalizes material type strings to the canonical key for _TRAY_INFO_IDX - # and _default_filament_name. PLA variants without an exact match fall - # back to their base family (PLA+ -> PLA+, PLA Matte -> PLA, etc.). - @staticmethod - def _normalize_material(mat: str) -> str: - m = mat.upper().strip().replace("-", " ").replace("_", " ") - # Bekannte Varianten normalisieren - _ALIASES = { - "PLAPLUS": "PLA+", "PLA PLUS": "PLA+", - "SILK PLA": "PLA SILK", "PLASILK": "PLA SILK", - "PLA MATTE": "PLA MATTE", "PLA MARBLE": "PLA MARBLE", - "PLA WOOD": "PLA WOOD", - "TPE": "TPU", - "PETG PLUS": "PETG+", - "PA6": "PA", "PA12": "PA", "PA66": "PA", - } - if m in _ALIASES: - return _ALIASES[m] - return m - - @staticmethod - def _material_family(mat: str) -> str: - """Reduce a material to its base polymer family. - - PLA / PLA+ / PLA SILK / PLA MATTE -> "PLA"; PETG / PETG+ -> "PETG"; etc. - Used by the stale-profile guard: only a change of *family* (e.g. PETG -> - PLA) invalidates a saved slot profile — a change within the family - (PLA -> PLA SILK) must not discard an otherwise valid profile. - """ - if not mat: - return "" - m = KobraXBridge._normalize_material(mat) - # Longer prefixes first so "PETG" is not swallowed by "PET". - for fam in ("PETG", "PLA", "ABS", "ASA", "TPU", "PVA", "HIPS", "PA", "PC", "PET"): - if m.startswith(fam): - return fam - return m - - def _parse_combined_rfid_type(self, raw_type: str) -> tuple[str, str]: - """Split a combined ACE-RFID "VENDOR TYPE SERIAL" string (e.g. - "GEEETECH PLA Bas", written via third-party RFID tools) into - (vendor, material_family). - - Anycubic's ACE RFID system concatenates vendor + material + a - truncated serial/variant into one `type` string for custom tags - - unlike a normal spool report where `type` is just "PLA"/"PETG"/etc. - Returns ("", "") when the first token isn't a known vendor (from the - merged system+user filament library), which leaves plain type - strings like "PLA" completely unaffected (Issue #101). - """ - tokens = raw_type.split() - if len(tokens) < 2: - return "", "" - first = tokens[0].strip().lower() - vendors = {p.get("vendor", "").lower(): p.get("vendor", "") for p in self._load_orca_filaments()} - vendor = vendors.get(first) - if not vendor: - return "", "" - family = self._material_family(" ".join(tokens[1:])) - if not family: - return "", "" - return vendor, family - - @staticmethod - def _rfid_variant_tokens(raw_type: str) -> list[str]: - """Tokens after "VENDOR TYPE" in a combined ACE-RFID string (e.g. - ["bas"] for "GEEETECH PLA Bas") - the truncated variant/serial that - distinguishes multiple profiles of the same (vendor, material family), - e.g. "Basic" vs. "Matte". Kept separate from _parse_combined_rfid_type() - so that function's 2-tuple signature (and its existing callers/tests) - stay unchanged (Issue #101).""" - tokens = raw_type.split() - return [t.lower() for t in tokens[2:]] - - def _match_profile_by_vendor_family(self, vendor: str, family: str, - variant_tokens: list[str] | None = None) -> dict: - """Find an imported/system filament profile by (vendor, material - family) - used to auto-resolve a combined ACE-RFID type string to - the user's already-imported OrcaSlicer profile (Issue #101), since - the exact profile `name` never appears verbatim in the truncated - RFID string. - - When multiple profiles share the same (vendor, family) - e.g. "Geeetech - PLA Basic" and "Geeetech PLA Matte" both matching (Geeetech, PLA) - - variant_tokens (the RFID string's remaining tokens, e.g. ["bas"] for - "Basic") are scored against each candidate's name: a word-prefix match - scores higher than a plain substring match, so "bas" prefers "Basic" - over "Matte" or an unrelated profile name containing "bas" as noise. - Falls back to the first match when nothing disambiguates.""" - matches = [ - p for p in self._load_orca_filaments() - if p.get("vendor", "").lower() == vendor.lower() - and self._material_family(p.get("type", "")) == family - ] - if not matches: - return {} - if len(matches) == 1 or not variant_tokens: - return matches[0] - - best = matches[0] - best_score = -1 - for p in matches: - name_words = p.get("name", "").lower().split() - score = 0 - for tok in variant_tokens: - if any(w.startswith(tok) for w in name_words): - score += 2 - elif tok in p.get("name", "").lower(): - score += 1 - if score > best_score: - best_score = score - best = p - log.debug( - f"_match_profile_by_vendor_family: {len(matches)} profiles match " - f"vendor={vendor!r} family={family!r}, variant_tokens={variant_tokens!r} " - f"-> {best.get('name')!r} (score={best_score})" - ) - return best - - def _profile_material(self, profile: dict) -> str: - """Material type (e.g. "PETG") of a saved slot profile, resolved by - (vendor, name) from the Orca filament library. Returns "" when the - profile is not in the library — we do NOT guess in that case.""" - name = (profile or {}).get("name", "") - if not name: - return "" - vendor = profile.get("vendor", "") - for p in self._load_orca_filaments(): - if p.get("vendor") == vendor and p.get("name") == name: - return p.get("type", "") or "" - return "" - - def _effective_slot_profile(self, global_idx: int, ams_material: str) -> dict: - """Saved slot-profile override — but only while its material *family* - still matches the material currently loaded in the AMS. Falls back to - auto-resolving a combined ACE-RFID type string (Issue #101) when there - is no (usable) manual override. - - Non-destructive suppression (Option A): when the family no longer matches - (e.g. a PETG profile but PLA loaded) the override is skipped → falls - through to the RFID auto-match / generic default. The override stays in - config.ini and reactivates as soon as the matching material is loaded - again. When the profile's family is unknown we do NOT suppress (fail-safe). - - Centralized here (rather than duplicated per caller) so every consumer - - the dashboard's /kx/filament/slots, Happy-Hare gate data, and the - OrcaSlicer lane-data sync - benefits from RFID auto-matching identically, - instead of only the one call site that happened to also call - _parse_combined_rfid_type() directly.""" - # A combined ACE-RFID string ("GEEETECH PLA Bas") carries a vendor - # prefix that _material_family() alone can't see past (it only - # strips known polymer prefixes, so "GEEETECH PLA BAS" resolves to - # itself, not "PLA") - resolve the plain material family through the - # RFID parser first so the stale-profile guard below compares against - # the actual polymer family, not the raw combined string. - vendor, family = self._parse_combined_rfid_type(ams_material) - plain_material = family or ams_material - - profile = self._filament_profiles.get(global_idx) or {} - if profile.get("name"): - prof_fam = self._material_family(self._profile_material(profile)) - ams_fam = self._material_family(plain_material) - if not (prof_fam and ams_fam and prof_fam != ams_fam): - return profile - - if vendor: - variant_tokens = self._rfid_variant_tokens(ams_material) - auto = self._match_profile_by_vendor_family(vendor, family, variant_tokens) - if auto.get("name"): - return auto - - return {} - - def _build_lane_data(self) -> dict: - """Builds BBL AMS JSON for OrcaSlicer DevFilaSystemParser::ParseV1_0. - - POSITION-FAITHFUL: every physical slot keeps its position (tray id = - slot position). Empty slots are reported as placeholder trays, NOT - filtered out/compacted - otherwise colors shift to wrong positions - (e.g. slot 1=yellow, 2=empty, 3=red -> red must not land on position 2). - """ - slots = self._ams_slots - total = len(slots) - if total == 0: - return {"ams": [], "ams_exist_bits": "0", "tray_exist_bits": "0"} - - ams_count = (total + 3) // 4 - ams_exist_bits = 0 - tray_exist_bits = 0 - ams_array = [] - - for ams_id in range(ams_count): - ams_exist_bits |= (1 << ams_id) - tray_array = [] - max_slot = min(3, total - ams_id * 4 - 1) - for slot_id in range(max_slot + 1): - slot_index = ams_id * 4 + slot_id - slot = slots[slot_index] if slot_index < total else {} - occupied = slot.get("status") == 5 - - if occupied: - tray_exist_bits |= (1 << slot_index) - color_raw = slot.get("color", [255, 255, 255]) - if isinstance(color_raw, list) and len(color_raw) >= 3: - color_hex = "{:02X}{:02X}{:02X}FF".format( - int(color_raw[0]), int(color_raw[1]), int(color_raw[2]) - ) - elif isinstance(color_raw, str) and len(color_raw) >= 6: - color_hex = color_raw[:6].upper() + "FF" - else: - color_hex = "FFFFFFFF" - material = self._normalize_material(slot.get("type", "PLA")) - # User override from config.ini [filament_profiles].slot_N_id - # takes precedence over the default mapping by material type. - # The vendor is sent along (tray_sub_brands + filament_vendor), - # so a patched OrcaSlicer can match by brand + type + - # color (analogous to SnapmakerPrinterAgent). - # Three-layer resolution for the filament hint sent to OrcaSlicer, - # all handled inside _effective_slot_profile() (Issue #101): - # 1. User-Wahl (config.ini [filament_profiles]) — exakte Kontrolle - # 2. Combined ACE-RFID "VENDOR TYPE SERIAL" string (e.g. - # "GEEETECH PLA Bas") auto-matched against the user's - # already-imported profile library. Not persisted to - # config.ini - re-derives on every call, so a differently - # tagged spool loaded later isn't stuck with a stale match. - # 3. Generic fallback (_TRAY_INFO_IDX) per material type - no - # vendor hint; OrcaSlicer then picks its own generic preset - user_profile = self._effective_slot_profile(slot_index, material) - if user_profile.get("name"): - material = self._material_family(user_profile.get("type", material)) or material - vendor = user_profile.get("vendor", "") - fila_name = user_profile.get("name", "") - tray_info_idx = user_profile.get("id") or self._TRAY_INFO_IDX.get(material, "OGFL99") - else: - # Default: Library-Generic-Profil (siehe _default_filament_name) — - # is compatible with all printers and guaranteed to be visible. - # The user deliberately picks a concrete brand per slot if they - # want one; the default stays neutral. - fila_name = self._default_filament_name(material) - vendor = "Generic" if fila_name.startswith("Generic ") else "" - tray_info_idx = self._lookup_filament_id(vendor, fila_name) or self._TRAY_INFO_IDX.get(material, "OGFL99") - tray_array.append({ - "id": str(slot_id), - "tag_uid": "0000000000000000", - "tray_info_idx": tray_info_idx, - "tray_type": material, - "tray_color": color_hex, - "tray_sub_brands": vendor, - # OrcaSlicer-Empfangs-Patch PR #13719 erwartet `name` + - # `vendor_name` pro Lane (Stufen-Matching: Vendor+Name → Name → - # filament_id_by_type). We send both spellings so that - # older patch variants + future upstream PRs are both - # covered. - "name": fila_name, - "vendor_name": vendor, - # Aliases for older patch variants (variant 2, - # MoonrakerPrinterAgent.cpp): filament_id direkt (exakt), - # otherwise resolve the preset name via find_preset(). - "filament_id": tray_info_idx, - "filament_vendor": vendor, - "filament_name": fila_name, - "preset": fila_name, - }) - else: - tray_array.append({ - "id": str(slot_id), - "tag_uid": "0000000000000000", - "tray_info_idx": "", - "tray_type": "", - "tray_color": "00000000", - "tray_slot_placeholder": "1", - }) - - ams_array.append({"id": str(ams_id), "info": "0002", "tray": tray_array}) - - return { - "ams": ams_array, - "ams_exist_bits": format(ams_exist_bits, "X"), - "tray_exist_bits": format(tray_exist_bits, "X"), - } - @staticmethod def _layer_height_from_filename(fname: str) -> float: """OrcaSlicer-Default-Filename-Pattern: `___.gcode`