#!/usr/bin/env python3 """Mirror Klipper/Moonraker print jobs into Google Calendar. The watcher is read-only with respect to Moonraker. It never sends G-code or printer-control commands. It polls print_stats for each configured Moonraker instance and sends idempotent start/end updates to a Google Apps Script web app. Compatible with Python 3.7 and newer; no third-party Python packages required. """ import argparse import json import logging import logging.handlers import os import signal import ssl import stat import sys import time import urllib.error import urllib.parse import urllib.request import uuid ACTIVE_STATES = {"printing", "paused"} TERMINAL_STATES = {"complete", "cancelled", "error"} def as_float(value, default=0.0): try: return float(value) except (TypeError, ValueError): return default def atomic_write_json(path, value): parent = os.path.dirname(path) if parent: os.makedirs(parent, exist_ok=True) temporary = path + ".tmp" with open(temporary, "w") as output: json.dump(value, output, indent=2, sort_keys=True) output.write("\n") output.flush() os.fsync(output.fileno()) os.chmod(temporary, 0o600) os.replace(temporary, path) def json_request(url, timeout, api_key=None, method="GET", payload=None): headers = {"Accept": "application/json"} data = None if api_key: headers["X-Api-Key"] = api_key if payload is not None: headers["Content-Type"] = "application/json" data = json.dumps(payload).encode("utf-8") request = urllib.request.Request(url, data=data, headers=headers, method=method) context = ssl.create_default_context() if url.lower().startswith("https://") else None with urllib.request.urlopen(request, timeout=timeout, context=context) as response: raw = response.read().decode("utf-8") parsed = json.loads(raw) if not isinstance(parsed, dict): raise ValueError("expected a JSON object from %s" % url) return parsed class PersistentState: def __init__(self, path, logger): self.path = path self.logger = logger self.data = {"version": 1, "printers": {}, "outbox": []} self.load() def load(self): try: with open(self.path, "r") as input_file: loaded = json.load(input_file) if not isinstance(loaded, dict): raise ValueError("root is not an object") if not isinstance(loaded.get("printers"), dict): loaded["printers"] = {} if not isinstance(loaded.get("outbox"), list): loaded["outbox"] = [] loaded["version"] = 1 self.data = loaded except FileNotFoundError: return except Exception as exc: self.logger.error("Could not load %s; starting with empty state: %s", self.path, exc) def save(self): atomic_write_json(self.path, self.data) def printer(self, printer_id): return self.data["printers"].setdefault( printer_id, {"last_state": None, "active_job": None}, ) def enqueue(self, payload): action_key = "%s:%s:%s" % ( payload.get("job_id"), payload.get("action"), payload.get("status"), ) for entry in self.data["outbox"]: if entry.get("key") == action_key: return False self.data["outbox"].append( { "key": action_key, "payload": payload, "attempts": 0, "next_attempt": 0, } ) return True class MoonrakerClient: def __init__(self, printer, timeout): self.printer = printer self.base_url = printer["moonraker_url"].rstrip("/") self.api_key = str(printer.get("api_key", "")).strip() or None self.timeout = timeout def get_status(self): response = json_request( self.base_url + "/printer/objects/query?webhooks&virtual_sdcard&print_stats", self.timeout, api_key=self.api_key, ) result = response.get("result") if not isinstance(result, dict): raise ValueError("Moonraker response has no result object") status = result.get("status") if not isinstance(status, dict): raise ValueError("Moonraker response has no status object") if not isinstance(status.get("print_stats"), dict): raise ValueError("Moonraker did not return print_stats") if not isinstance(status.get("virtual_sdcard"), dict): raise ValueError("Moonraker did not return virtual_sdcard") return status def get_metadata(self, filename): encoded = urllib.parse.urlencode({"filename": filename}) response = json_request( self.base_url + "/server/files/metadata?" + encoded, self.timeout, api_key=self.api_key, ) result = response.get("result") return result if isinstance(result, dict) else {} class CalendarWatcher: def __init__(self, config, logger): self.config = config self.logger = logger self.poll_seconds = float(config.get("poll_seconds", 2)) self.http_timeout = float(config.get("http_timeout_seconds", 10)) self.endpoint = config["calendar_endpoint"] self.token = config["calendar_token"] self.update_completed = bool(config.get("update_completed_events", True)) self.state = PersistentState(config["state_file"], logger) self.printers = config["printers"] self.clients = { printer["id"]: MoonrakerClient(printer, self.http_timeout) for printer in self.printers } self.stop_requested = False self.last_errors = {} def request_stop(self, signum=None, frame=None): self.stop_requested = True def check(self): all_ok = True for printer in self.printers: try: status = self.clients[printer["id"]].get_status() webhooks = status.get("webhooks", {}) print_stats = status["print_stats"] self.logger.info( "[%s] Moonraker=%s, Klipper=%s, print_state=%s, filename=%s", printer["label"], printer["moonraker_url"], webhooks.get("state", "unknown"), print_stats.get("state", "unknown"), print_stats.get("filename", ""), ) except Exception as exc: all_ok = False self.logger.error("[%s] Check failed: %s", printer["label"], exc) return all_ok def run(self): self.logger.info("Watching %d Moonraker instances", len(self.printers)) while not self.stop_requested: cycle_started = time.monotonic() for printer in self.printers: self.poll_printer(printer) self.flush_outbox(max_items=10) elapsed = time.monotonic() - cycle_started remaining = max(0.1, self.poll_seconds - elapsed) end_wait = time.monotonic() + remaining while not self.stop_requested and time.monotonic() < end_wait: sleep_seconds = end_wait - time.monotonic() if sleep_seconds <= 0: break time.sleep(min(0.25, sleep_seconds)) self.logger.info("Watcher stopped") def poll_printer(self, printer): printer_id = printer["id"] label = printer["label"] try: status = self.clients[printer_id].get_status() webhooks = status.get("webhooks", {}) if webhooks.get("state") not in (None, "ready"): raise RuntimeError("Klipper is %s: %s" % ( webhooks.get("state"), webhooks.get("state_message", "") )) self.last_errors.pop(printer_id, None) self.process_status(printer, status) except Exception as exc: message = str(exc) if self.last_errors.get(printer_id) != message: self.logger.error("[%s] Moonraker query failed: %s", label, message) self.last_errors[printer_id] = message def process_status(self, printer, status): printer_id = printer["id"] printer_state = self.state.printer(printer_id) print_stats = status["print_stats"] virtual_sdcard = status["virtual_sdcard"] current_state = str(print_stats.get("state", "standby")).lower() filename = str(print_stats.get("filename", "")).strip() active_job = printer_state.get("active_job") changed = False if current_state in ACTIVE_STATES: if active_job is None or active_job.get("filename") != filename: if active_job is not None: self.queue_end(printer, active_job, "stopped", "A different job became active") active_job = self.build_job(printer, print_stats, virtual_sdcard) printer_state["active_job"] = active_job self.queue_start(printer, active_job) changed = True self.logger.info( "[%s] Print started: %s (calendar end estimate %s)", printer["label"], active_job["filename"], time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(active_job["estimated_end"])), ) elif current_state in TERMINAL_STATES and active_job is not None: status_name = { "complete": "completed", "cancelled": "cancelled", "error": "failed", }[current_state] if status_name != "completed" or self.update_completed: reason = str(print_stats.get("message", "")).strip() self.queue_end(printer, active_job, status_name, reason, print_stats) printer_state["active_job"] = None changed = True self.logger.info( "[%s] Print ended: %s (%s)", printer["label"], active_job["filename"], status_name, ) elif current_state == "standby" and active_job is not None: # This covers a watcher restart or a Moonraker reset that caused us # to miss the brief terminal state. It still closes the calendar # event instead of leaving it at the predicted end forever. self.queue_end(printer, active_job, "stopped", "Printer returned to standby") printer_state["active_job"] = None changed = True self.logger.warning( "[%s] Active job disappeared; calendar event marked stopped", printer["label"], ) if printer_state.get("last_state") != current_state: printer_state["last_state"] = current_state changed = True if changed: self.state.save() def build_job(self, printer, print_stats, virtual_sdcard): now = time.time() filename = str(print_stats.get("filename", "")).strip() or "Unknown file" total_duration = max(0.0, as_float(print_stats.get("total_duration"))) print_duration = max(0.0, as_float(print_stats.get("print_duration"))) approximate_start = now - max(total_duration, print_duration) metadata = {} try: metadata = self.clients[printer["id"]].get_metadata(filename) except Exception as exc: self.logger.warning("[%s] Could not read G-code metadata: %s", printer["label"], exc) # File metadata can retain print_start_time and job_id from a previous # run of the same G-code. In particular, an immediate reprint after a # cancellation can otherwise be mistaken for a retry of the cancelled # run. Use the live print counters for timing and assign a fresh ID to # each newly observed transition into an active state. start_time = approximate_start estimated_seconds = max(0.0, as_float(metadata.get("estimated_time"))) progress = max(0.0, min(1.0, as_float(virtual_sdcard.get("progress")))) if estimated_seconds <= 0 and progress > 0.005 and print_duration > 0: estimated_seconds = print_duration / progress estimated_seconds = max(estimated_seconds, print_duration + 60.0, 60.0) estimated_seconds = min(estimated_seconds, 30.0 * 24.0 * 60.0 * 60.0) identity = "%s:run:%d:%s" % ( printer["id"], int(now * 1000), uuid.uuid4().hex[:16], ) return { "job_id": identity, "filename": filename, "start": int(start_time), "estimated_end": int(start_time + estimated_seconds), } def queue_start(self, printer, job): payload = { "token": self.token, "action": "start", "status": "printing", "job_id": job["job_id"], "printer": printer["label"], "file": job["filename"], "start": job["start"], "end": job["estimated_end"], } self.state.enqueue(payload) def queue_end(self, printer, job, status_name, reason="", print_stats=None): now = time.time() end_time = now if isinstance(print_stats, dict): total_duration = max(0.0, as_float(print_stats.get("total_duration"))) calculated_end = job["start"] + total_duration if job["start"] + 1 <= calculated_end <= now + 60: end_time = calculated_end payload = { "token": self.token, "action": "end", "status": status_name, "job_id": job["job_id"], "printer": printer["label"], "file": job["filename"], "start": job["start"], "end": int(max(job["start"] + 1, end_time)), "reason": reason, } self.state.enqueue(payload) def flush_outbox(self, max_items): sent = 0 while sent < max_items and self.state.data["outbox"]: entry = self.state.data["outbox"][0] if as_float(entry.get("next_attempt")) > time.time(): return try: response = json_request( self.endpoint, self.http_timeout, method="POST", payload=entry["payload"], ) if response.get("ok") is not True: raise RuntimeError(response.get("error", "calendar endpoint rejected request")) payload = entry["payload"] self.logger.info( "[%s] Calendar %s accepted for %s " "(status=%s, event_id=%s, unchanged=%s)", payload["printer"], payload["action"], payload["file"], response.get("status", "unknown"), response.get("event_id", "unknown"), response.get("unchanged", False), ) self.state.data["outbox"].pop(0) self.state.save() sent += 1 except Exception as exc: entry["attempts"] = int(entry.get("attempts", 0)) + 1 delay = min(300, 5 * (2 ** min(entry["attempts"] - 1, 6))) entry["next_attempt"] = int(time.time() + delay) self.state.save() self.logger.error( "Calendar request failed (attempt %d, retry in %ds): %s", entry["attempts"], delay, exc, ) return def load_config(path): with open(path, "r") as input_file: config = json.load(input_file) if not isinstance(config, dict): raise ValueError("config root must be an object") endpoint = str(config.get("calendar_endpoint", "")).strip() if not endpoint.startswith("https://script.google.com/"): raise ValueError("calendar_endpoint must be the HTTPS Apps Script web-app URL") token = str(config.get("calendar_token", "")).strip() if len(token) < 20 or token.startswith("REPLACE_"): raise ValueError("calendar_token must be replaced with a secret of at least 20 characters") base_dir = os.path.dirname(os.path.abspath(path)) config["calendar_endpoint"] = endpoint config["calendar_token"] = token config["state_file"] = os.path.abspath( config.get("state_file", os.path.join(base_dir, "state.json")) ) config["log_file"] = os.path.abspath( config.get("log_file", os.path.join(base_dir, "watcher.log")) ) poll_seconds = as_float(config.get("poll_seconds", 2), 2) if poll_seconds < 1 or poll_seconds > 60: raise ValueError("poll_seconds must be between 1 and 60") printers = config.get("printers") if not isinstance(printers, list) or not printers: raise ValueError("printers must be a non-empty array") ids = set() urls = set() for index, printer in enumerate(printers): if not isinstance(printer, dict): raise ValueError("printer %d must be an object" % (index + 1)) for key in ("id", "label", "moonraker_url"): if not str(printer.get(key, "")).strip(): raise ValueError("printer %d is missing %s" % (index + 1, key)) printer["id"] = str(printer["id"]).strip() printer["label"] = str(printer["label"]).strip() printer["moonraker_url"] = str(printer["moonraker_url"]).strip().rstrip("/") if not printer["moonraker_url"].startswith(("http://", "https://")): raise ValueError("printer %s has an invalid moonraker_url" % printer["id"]) if printer["id"] in ids: raise ValueError("duplicate printer id: %s" % printer["id"]) if printer["moonraker_url"] in urls: raise ValueError("duplicate moonraker_url: %s" % printer["moonraker_url"]) ids.add(printer["id"]) urls.add(printer["moonraker_url"]) return config def setup_logging(path, verbose=False): parent = os.path.dirname(path) if parent: os.makedirs(parent, exist_ok=True) logger = logging.getLogger("moonraker-calendar-watcher") logger.setLevel(logging.DEBUG if verbose else logging.INFO) formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") stream = logging.StreamHandler(sys.stdout) stream.setFormatter(formatter) logger.addHandler(stream) rotating = logging.handlers.RotatingFileHandler( path, maxBytes=2 * 1024 * 1024, backupCount=2 ) rotating.setFormatter(formatter) logger.addHandler(rotating) return logger def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--config", required=True, help="Path to config.json") parser.add_argument("--check", action="store_true", help="Test Moonraker only; do not contact Google") parser.add_argument("--verbose", action="store_true") args = parser.parse_args() config_path = os.path.abspath(args.config) config = load_config(config_path) logger = setup_logging(config["log_file"], args.verbose) try: mode = stat.S_IMODE(os.stat(config_path).st_mode) if mode & 0o077: logger.warning("Config contains a secret; run: chmod 600 %s", config_path) except OSError: pass watcher = CalendarWatcher(config, logger) if args.check: return 0 if watcher.check() else 1 signal.signal(signal.SIGINT, watcher.request_stop) signal.signal(signal.SIGTERM, watcher.request_stop) watcher.run() return 0 if __name__ == "__main__": sys.exit(main())