"""Minimal Spoolman REST client (/api/v1).""" from __future__ import annotations import json import re import httpx # Extra fields the bridge keeps on loaded spools (Spoolman stores values JSON-encoded). FIELD_PRINTER = "drucker" FIELD_SLOT = "slot" FIELD_DEFS = { FIELD_PRINTER: {"name": "Drucker", "field_type": "text", "order": 100}, FIELD_SLOT: {"name": "Slot", "field_type": "integer", "order": 101}, } class SpoolmanError(RuntimeError): pass class SpoolmanNotFound(SpoolmanError): pass class Spoolman: def __init__(self, base_url: str, timeout: float = 8.0, transport: httpx.BaseTransport | None = None): self.client = httpx.Client(base_url=base_url.rstrip("/") + "/api/v1", timeout=timeout, transport=transport) def close(self): self.client.close() def _request(self, method: str, path: str, **kw): try: r = self.client.request(method, path, **kw) except httpx.HTTPError as e: raise SpoolmanError(f"Spoolman nicht erreichbar ({type(e).__name__})") from None if r.status_code == 404: raise SpoolmanNotFound(f"Spoolman {path}: nicht gefunden") if r.status_code >= 400: raise SpoolmanError(f"Spoolman {method} {path}: HTTP {r.status_code}") try: return r.json() except ValueError: raise SpoolmanError(f"Spoolman {path}: keine JSON-Antwort") from None def spool(self, spool_id: int) -> dict: return self._request("GET", f"/spool/{int(spool_id)}") def spools(self, archived: bool = False) -> list[dict]: return self._request("GET", "/spool", params={"allow_archived": str(archived).lower()}) def update_spool(self, spool_id: int, body: dict) -> dict: return self._request("PATCH", f"/spool/{int(spool_id)}", json=body) def fields(self) -> list[dict]: return self._request("GET", "/field/spool") def add_field(self, key: str, params: dict) -> list[dict]: return self._request("POST", f"/field/spool/{key}", json=params) def use(self, spool_id: int, grams: float) -> dict: return self._request("PUT", f"/spool/{int(spool_id)}/use", json={"use_weight": round(float(grams), 3)}) def vendor_name(spool: dict) -> str: return ((spool.get("filament") or {}).get("vendor") or {}).get("name") or "" def label(spool: dict) -> str: """lot_nr like SUNLU001; otherwise built from vendor + id the same way.""" lot = (spool.get("lot_nr") or "").strip() if lot: return lot vendor = re.sub(r"[^A-Z0-9]", "", vendor_name(spool).upper()) or "SPOOL" return f"{vendor}{int(spool['id']):03d}" def color_hex(spool: dict) -> str | None: fil = spool.get("filament") or {} color = fil.get("color_hex") if not color and fil.get("multi_color_hexes"): color = str(fil["multi_color_hexes"]).split(",")[0] if not color: return None color = color.strip().lstrip("#").upper()[:6] return color if re.fullmatch(r"[0-9A-F]{6}", color) else None def extra_value(spool: dict, key: str): raw = (spool.get("extra") or {}).get(key) if raw is None: return None try: return json.loads(raw) except (TypeError, ValueError): return None def summary(spool: dict) -> dict: fil = spool.get("filament") or {} return { "id": spool.get("id"), "label": label(spool), "vendor": vendor_name(spool), "name": fil.get("name") or "", "material": fil.get("material") or "", "color": color_hex(spool), "diameter": fil.get("diameter"), "density": fil.get("density"), "remaining": spool.get("remaining_weight"), "initial": spool.get("initial_weight") or fil.get("weight"), "spool_weight": spool.get("spool_weight") or fil.get("spool_weight"), "extruder_temp": fil.get("settings_extruder_temp"), "location": spool.get("location") or "", "archived": bool(spool.get("archived")), "used_weight": spool.get("used_weight"), "printer": extra_value(spool, FIELD_PRINTER), "slot": extra_value(spool, FIELD_SLOT), }