viewit/KX-Bridge-Release
12
18
Fork 11
Code Issues 5 Pull Requests Actions Packages Projects Releases 83 Wiki Activity

refactor(bridge): extract AmsFilamentMixin (stage 3, mixin 3/6)

Browse Source
This commit is contained in:
viewit
2026-08-04 20:18:20 +02:00
parent 5311d74fd0
commit cda0e984ec
2 changed files with 864 additions and 840 deletions
Download Patch File Download Diff File Expand all files Collapse all files

862
bridge_ams.py Normal file
View File

@@ -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"),
}

842
kobrax_moonraker_bridge.py
View File

@@ -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: `<plate>_<material>_<layer>_<dur>.gcode`