1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
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})
|
