# deck-it - a Klipper printer (Creality K1C) through Moonraker # Copyright (C) 2026 Danilo M. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 as # published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License along # with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. """A Klipper printer, read through Moonraker's HTTP API. Read-only apart from one action, the emergency stop (M112), which Klipper answers by shutting down at once. The printer is often switched off, so a failed poll backs off instead of costing every tick a timeout. """ import json import posixpath import time import urllib.parse import urllib.request from ..module import Module, ModuleError QUERY = ("/printer/objects/query?webhooks&print_stats&virtual_sdcard" "&extruder=temperature,target,power&heater_bed=temperature,target,power" "&fan=speed&output_pin%20fan0=value&toolhead=position,homed_axes&gcode_move=speed_factor") BACKOFF = 10 # seconds between tries while the printer is off def _http(url, data=None, timeout=2): req = urllib.request.Request(url, data=data, method="POST" if data is not None else "GET") with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read()) def _camera(webcams) -> str | None: """The first enabled webcam's live MJPEG stream URL, if it is plain http(s).""" for cam in webcams or []: url = cam.get("stream_url") or "" if cam.get("enabled", True) and url.startswith(("http://", "https://")): return url return None def _thumbnail(base: str, filename: str, meta: dict) -> str | None: """The largest slicer thumbnail's URL; relative to the gcode's own folder.""" thumbs = [t for t in meta.get("thumbnails") or [] if t.get("relative_path")] if not thumbs: return None best = max(thumbs, key=lambda t: (t.get("width") or 0) * (t.get("height") or 0)) path = posixpath.join(posixpath.dirname(filename), best["relative_path"]) return f"{base}/server/files/gcodes/{urllib.parse.quote(path)}" def _first(*values): return next((v for v in values if v is not None), None) def summarize(status: dict, camera: str | None = None) -> dict: """The panel's state from Moonraker's object status.""" ps = status.get("print_stats") or {} sd = status.get("virtual_sdcard") or {} hook = status.get("webhooks") or {} progress = sd.get("progress") or 0.0 elapsed = ps.get("print_duration") or 0.0 # Remaining from progress by file position, as Fluidd's "file" estimate. remaining = elapsed / progress - elapsed if progress > 0.01 and elapsed else None info = ps.get("info") or {} def heater(name): h = status.get(name) or {} return {"temp": h.get("temperature"), "target": h.get("target"), "power": h.get("power")} return { "offline": False, "klippy": hook.get("state"), # ready, startup, shutdown, error "message": hook.get("state_message"), "state": ps.get("state"), # standby, printing, paused, complete, cancelled, error "file": ps.get("filename") or None, "progress": progress, "elapsed": elapsed, "remaining": remaining, "layer": info.get("current_layer"), "layers": info.get("total_layer"), "filament": ps.get("filament_used"), "extruder": heater("extruder"), "bed": heater("heater_bed"), # Creality's firmware drives the part fan as `output_pin fan0` and # leaves the standard `fan` object empty. "fan": _first((status.get("fan") or {}).get("speed"), (status.get("output_pin fan0") or {}).get("value")), "position": ((status.get("toolhead") or {}).get("position") or [None] * 3)[:3], "speed_factor": (status.get("gcode_move") or {}).get("speed_factor"), "camera": camera, } def make(cfg: dict) -> Module: base = str(cfg.get("k1c", {}).get("url", "")).rstrip("/") http = cfg.get("http", _http) down = {"until": 0.0, "camera": None} # The current file's slicer metadata, fetched once per file. meta = {"file": None, "thumbnail": None, "estimate": None} def file_meta(filename): if filename == meta["file"]: return meta.update(file=filename, thumbnail=None, estimate=None) if not filename: return try: m = http(base + "/server/files/metadata?filename=" + urllib.parse.quote(filename))["result"] except (OSError, ValueError, KeyError, TypeError): meta["file"] = None # try again next poll return meta.update(thumbnail=_thumbnail(base, filename, m), estimate=m.get("estimated_time")) def state(): if not base: return {"offline": True, "unconfigured": True} if time.monotonic() < down["until"]: return {"offline": True} try: status = http(base + QUERY)["result"]["status"] if down["camera"] is None: down["camera"] = _camera(http(base + "/server/webcams/list")["result"]["webcams"]) or "" except (OSError, ValueError, KeyError, TypeError): down["until"] = time.monotonic() + BACKOFF return {"offline": True} st = summarize(status, down["camera"] or None) file_meta(st["file"]) st["thumbnail"] = meta["thumbnail"] # Until the file-position estimate settles, the slicer's is better. if st["remaining"] is None and meta["estimate"] and st["state"] == "printing": st["remaining"] = max(0.0, meta["estimate"] - st["elapsed"]) return st def emergency_stop(body): if not base: raise ModuleError("no printer configured") try: http(base + "/printer/emergency_stop", data=b"", timeout=3) except (OSError, ValueError) as e: raise ModuleError(f"emergency stop failed: {e}") from e return Module(name="k1c", title="K1C", state=state, actions={"emergency_stop": emergency_stop})