diff options
Diffstat (limited to 'deckit')
| -rw-r--r-- | deckit/modules/k1c.py | 147 |
1 files changed, 147 insertions, 0 deletions
diff --git a/deckit/modules/k1c.py b/deckit/modules/k1c.py new file mode 100644 index 0000000..bc7c439 --- /dev/null +++ b/deckit/modules/k1c.py @@ -0,0 +1,147 @@ +# deck-it - a Klipper printer (Creality K1C) through Moonraker +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# +# 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&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 _snapshot(webcams) -> str | None: + """The first enabled webcam's snapshot URL, if it is plain http(s).""" + for cam in webcams or []: + url = cam.get("snapshot_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 summarize(status: dict, snapshot: 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"), + "fan": (status.get("fan") or {}).get("speed"), + "position": ((status.get("toolhead") or {}).get("position") or [None] * 3)[:3], + "speed_factor": (status.get("gcode_move") or {}).get("speed_factor"), + "snapshot": snapshot, + } + + +def make(cfg: dict) -> Module: + base = str(cfg.get("k1c", {}).get("url", "")).rstrip("/") + http = cfg.get("http", _http) + down = {"until": 0.0, "snapshot": 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["snapshot"] is None: + down["snapshot"] = _snapshot(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["snapshot"] 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}) |