""" bridge_moonraker.py - MoonrakerCompatMixin for KobraXBridge. The Moonraker/Klipper-compatible HTTP surface (/server/*, /printer/*, /machine/*) that Mainsail/Fluidd/OrcaSlicer/moonraker-obico talk to: server/printer info, printer.objects query/list/subscribe, files list + metadata, history, webcams, access api-key + update-manager stubs. Mixed into KobraXBridge. ──────────────────────────────────────────────────────────────────────────── Copyright (C) 2026 viewit (KX-Bridge contributors) Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md. """ import sys import time import logging from aiohttp import web from bridge_constants import MOONRAKER_VERSION, KLIPPER_VERSION log = logging.getLogger("bridge") class MoonrakerCompatMixin: async def handle_server_info(self, request): return web.json_response({ "result": { "klippy_connected": True, "klippy_state": "ready", "components": ["file_manager", "job_state", "virtual_sdcard"], "failed_components":[], "registered_directories": ["gcodes"], "warnings": [], "websocket_count": len(self.ws_clients), "moonraker_version": MOONRAKER_VERSION, "api_version": [1, 3, 0], "api_version_string": "1.3.0", } }) async def handle_printer_info(self, request): s = self._state return web.json_response({ "result": { "state": "ready", "state_message": "Printer is ready", "hostname": "kobrax-bridge", "klipper_path": "/home/pi/klipper", "python_path": "/home/pi/klippy-env/bin/python", "log_file": "/tmp/klippy.log", "config_file": "/home/pi/printer.cfg", "software_version": KLIPPER_VERSION, "cpu_info": s["printer_name"], } }) async def handle_machine_system_info(self, request): return web.json_response({ "result": { "system_info": { "cpu_info": {"cpu_count": 4, "bits": "64bit", "processor": "armv7l", "cpu_desc": "Anycubic Kobra X Bridge", "serial_number": "", "hardware_desc": "", "model": "Kobra X Bridge", "total_memory": 524288, "memory_units": "kB"}, "sd_info": {}, "distribution": {"name": "Linux", "id": "linux", "version": "1.0", "version_parts": {}, "like": "", "codename": ""}, "available_services": [], "service_state": {}, "python": {"version": list(sys.version_info[:3]), "version_string": sys.version}, "network": {}, "canbus": {}, } } }) async def handle_objects_query(self, request): objects = self._build_printer_objects() requested = [] query = request.rel_url.query if "objects" in query: requested = [x.strip() for x in str(query.get("objects", "")).split(",") if x.strip()] elif query: requested = [k for k in query.keys() if k] filtered = {k: objects[k] for k in requested if k in objects} if requested else objects return web.json_response({"result": {"status": filtered, "eventtime": time.time()}}) async def handle_objects_list(self, request): return web.json_response({ "result": { "objects": list(self._build_printer_objects().keys()) } }) async def handle_objects_subscribe(self, request): return web.json_response({ "result": { "status": self._build_printer_objects(), "eventtime": time.time(), } }) async def handle_files_list(self, request): filename = self._state.get("filename", "") files = [] if filename: files.append({ "path": filename, "modified": time.time(), "size": 0, "permissions": "rw", }) return web.json_response({"result": files}) def _build_file_metadata(self, filename: str) -> dict: """Builds the Moonraker file metadata for a file. Shared source for HTTP /server/files/metadata AND the WS RPC server.files.metadata (previously the WS path had its own broken logic with a non-existent existierenden Store-Methode → leere Antwort → Mobileraker fragte in endless loop, app hung on refresh, Issue #48). Liefert Mobileraker-kompatible Pflichtfelder: `filename`, `size`, `modified` are non-nullable in GCodeFile; `print_start_time` and the Slicer-Felder optional.""" s = self._state # Live _state values are only relevant for the currently/last tracked # job's own file - using them as a starting point for a DIFFERENT # filename leaked the tracked job's layer count/time into unrelated # metadata queries (Issue #102). For any other filename, rely solely # on that file's own GCodeStore row. is_tracked_file = bool(filename) and filename == s.get("filename") layer_h = float(s.get("layer_height") or 0.0) if is_tracked_file else 0.0 first_h = float(s.get("first_layer_height") or 0.0) if is_tracked_file else 0.0 total_layers = int(s.get("total_layers") or 0) if is_tracked_file else 0 est_time = int(s.get("slicer_time") or 0) if is_tracked_file else 0 size_bytes = 0 try: gf = self._store.get_file_by_name(filename) or {} if not layer_h: layer_h = float(gf.get("layer_height") or 0.0) first_h = float(gf.get("first_layer_height") or layer_h) if not total_layers: total_layers = int(gf.get("layer_count") or 0) if not est_time: est_time = int(gf.get("est_print_time_sec") or 0) size_bytes = int(gf.get("size_bytes") or 0) except Exception: pass # Third fallback: the printer's own buried/report analytics event # (fires once per print start regardless of slicer), for files that # are neither the currently-tracked job nor in our own GCodeStore - # e.g. printed directly via Anycubic Slicer Next (Issue #102). buried = self._buried_cache if buried and buried.get("task_name") == filename: if not total_layers: total_layers = buried.get("total_layers") or total_layers if not est_time: est_time = buried.get("estimate_duration") or est_time if not size_bytes: size_bytes = buried.get("gcode_size") or size_bytes if not layer_h: layer_h = self._layer_height_from_filename(filename) if layer_h and not first_h: first_h = layer_h object_height = round(first_h + max(0, total_layers - 1) * layer_h, 3) if (layer_h and total_layers) else 0.0 return { "filename": filename, # GCodeFile (Mobileraker) requires size as a non-nullable int. "size": size_bytes or 1, "modified": time.time(), "estimated_time": est_time or None, "layer_height": layer_h or None, "first_layer_height": first_h or None, "layer_count": total_layers or None, "object_height": object_height or None, "thumbnails": [], } async def handle_files_metadata(self, request): """Moonraker /server/files/metadata — moonraker-obico + Mobileraker holen Datei-Metadaten (Slicer-Zeit, Layer, object_height). Logic in _build_file_metadata (shared with WS RPC).""" filename = request.rel_url.query.get("filename", "") or self._state.get("filename", "") if not filename: return web.json_response({"result": {}}) return web.json_response({"result": self._build_file_metadata(filename)}) # -- Moonraker stubs for moonraker-obico ---------------------------------- async def handle_access_api_key(self, request): """Moonraker /access/api_key - we have no auth, return a dummy. moonraker-obico logs a WARNING otherwise.""" return web.json_response({"result": "kx-bridge-no-auth-required"}) async def handle_machine_update_status(self, request): """Moonraker /machine/update/status - Obico uses this to show installed plugins.""" return web.json_response({ "result": { "busy": False, "github_rate_limit": 60, "github_requests_remaining": 60, "github_limit_reset_time": time.time() + 3600, "version_info": {}, } }) async def handle_history_list(self, request): """Moonraker /server/history/list - job history from the GCodeStore. moonraker-obico only uses the last element (limit=1, order=desc).""" try: limit = int(request.rel_url.query.get("limit", "50")) except ValueError: limit = 50 try: jobs = self._store.list_jobs(limit=limit) or [] except Exception: jobs = [] # Mapping to the Moonraker schema. Moonraker returns start_time as a Unix # timestamp (float), not an ISO string - moonraker-obico parses it with # int(start_time) and crashes otherwise. def _to_unix_ts(iso: str | None) -> float: if not iso: return 0.0 try: from datetime import datetime # Format from GCodeStore: "2026-05-27T21:22:25Z" dt = datetime.strptime(iso, "%Y-%m-%dT%H:%M:%SZ") return dt.replace(tzinfo=__import__("datetime").timezone.utc).timestamp() except Exception: return 0.0 result_jobs = [] for j in jobs: start_ts = _to_unix_ts(j.get("started_at")) dur = j.get("duration_sec") or 0 result_jobs.append({ "job_id": j.get("id"), "exists": True, "end_time": (start_ts + dur) if start_ts and dur else None, "filament_used": 0.0, "filename": j.get("filename", ""), "metadata": {}, "print_duration": dur, "status": j.get("status") or "completed", "start_time": start_ts, "total_duration": dur, }) return web.json_response({"result": {"count": len(result_jobs), "jobs": result_jobs}}) async def handle_webcams_list(self, request): """Moonraker /server/webcams/list - Obico fetches the webcam URLs here. When the client comes from another host (e.g. moonraker-obico on a separate server), it needs absolute URLs to reach the stream. A Host header with localhost/127.0.0.1 is replaced by the real LAN IP.""" host_hdr = request.headers.get("Host", "") if request else "" host_name = (host_hdr or "").split(":")[0] port_part = f":{host_hdr.split(':')[1]}" if ":" in (host_hdr or "") else f":{self._args.port}" local_ip = getattr(self, "_local_ip", None) or host_name if host_name in ("localhost", "127.0.0.1", ""): host_name = local_ip base = f"http://{host_name}{port_part}" stream_url = f"{base}/api/camera/stream" snapshot_url = f"{base}/api/camera/snapshot" return web.json_response({ "result": { "webcams": [ { "name": "KX-Bridge", "location": "printer", "service": "mjpegstreamer", "enabled": True, "icon": "mdiWebcam", "target_fps": 5, "target_fps_idle": 2, "stream_url": stream_url, "snapshot_url": snapshot_url, "flip_horizontal": False, "flip_vertical": False, "rotation": 0, "aspect_ratio": "16:9", "extra_data": {}, } ] } })