""" kobrax_moonraker_bridge.py - Moonraker-compatible HTTP/WebSocket bridge for the Anycubic Kobra X Emulates the Moonraker/Klipper API so OrcaSlicer can control the Kobra X directly. Verwendung: python kobrax_moonraker_bridge.py --printer-ip 192.168.178.94 OrcaSlicer-Konfiguration: Drucker-Typ: Klipper | Host: 127.0.0.1 | Port: 7125 ──────────────────────────────────────────────────────────────────────────── Copyright (C) 2026 viewit (KX-Bridge contributors) This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License v3.0 as published by the Free Software Foundation. See the LICENSE file in the project root or for the full text. This program is distributed WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. Reverse-engineering of the Anycubic Kobra X MQTT protocol was carried out for interoperability purposes (§69e UrhG / EU Software Directive Art. 6). This project is not affiliated with Anycubic. See NOTICE.md for details. """ import argparse try: import config_loader as env_loader except ImportError: import env_loader import asyncio import copy import json import logging import os import re import sys import time import threading # Must run before any extracted-module import below: bridge_endpoints pulls in # bridge_logging, which attaches its browser-log handler to the root logger at # import time. If that happens first, basicConfig() becomes a no-op (it only # configures the root logger when it has no handlers yet) and every log.info/ # warning call silently stops reaching stdout/stderr - "docker logs" then only # ever shows whatever fired before this point (e.g. an early config warning). logging.basicConfig(level=logging.INFO, format="[%(asctime)s] %(levelname)-5s %(name)s: %(message)s", datefmt="%H:%M:%S") log = logging.getLogger("bridge") # aiohttp logs one INFO line per HTTP request (access log) — with 2s frontend # polling that drowns out the bridge's own logs by default. Toggleable at # runtime via the verbose_http_log setting (see handle_api_settings_post). logging.getLogger("aiohttp.access").setLevel(logging.WARNING) # For PyInstaller binaries everything sits next to sys.executable, otherwise next to __file__ _BASE = os.path.dirname(sys.executable) if getattr(sys, "frozen", False) else os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, _BASE) # Read-only web assets (themes) are embedded in the onefile binary via --add-data under # sys._MEIPASS entpackt; im Script-/Docker-Modus liegen sie neben dieser Datei. _WEB_BASE = getattr(sys, "_MEIPASS", _BASE) from kobrax_client import KobraXClient # Extracted modules, re-exported here so existing imports (tests, callers) # keep working against kobrax_moonraker_bridge unchanged. from spoolman_client import SpoolmanClient from kxgauge_client import KXGaugeClient from gcode_store import GCodeStore from gcode_meta import ( _parse_gcode_estimated_time, _parse_gcode_layer_heights, _extract_thumbnail, _extract_filament_info, ) from camera import CameraCache, _find_ffmpeg from credentials import _kx_fetch_credentials, _kx_generate_signature, _kx_decrypt_info from bridge_spoolman import SpoolmanMixin from bridge_mqtt import MqttCallbacksMixin from bridge_ams import AmsFilamentMixin from bridge_moonraker import MoonrakerCompatMixin from bridge_endpoints import EndpointsMixin try: from aiohttp import web import aiohttp except ImportError: print("Error: aiohttp is not installed. Run: pip install aiohttp") sys.exit(1) # UI theme-name validation (used in __init__); the /kx/ui asset-serving # constants live in bridge_endpoints alongside their only consumers. _UI_THEME_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$") # Browser log stream (ring buffer + SSE queues + handler) lives in # bridge_logging; import the shared buffer/queues so the log-stream and # log-download endpoints below operate on the same objects the handler writes. from bridge_logging import ( _set_verbose_http_log, _log_buffer, _log_sse_queues, _BrowserLogHandler, _browser_handler, ) from bridge_constants import ( KOBRA_TO_KLIPPER_STATE, MOONRAKER_VERSION, KLIPPER_VERSION, DEFAULT_KXGAUGE_MAPPING, ) class KobraXBridge(SpoolmanMixin, MqttCallbacksMixin, AmsFilamentMixin, MoonrakerCompatMixin, EndpointsMixin): def _kxgauge_notify_state(self, kobra_state: str) -> None: """Fires the mapped KXGauge emotion for a state set outside of _on_print (offline transitions in _poll_loop/run_bridge, which set self._state["kobra_state"] directly rather than through the MQTT print/report callback).""" if not self._kxgauge: return emotion = self._kxgauge_mapping.get(kobra_state) if emotion: self._kxgauge.set_emotion(emotion) def _load_kxgauge_mapping_config(self) -> dict[str, str]: """Reads [kxgauge_mapping] (kobra_state -> KXGauge emotion name) from config.ini. Falls back to DEFAULT_KXGAUGE_MAPPING for any state not present in the file, or entirely if the section is missing.""" import configparser cfg_path = self._find_config_path() if not cfg_path.is_file(): return dict(DEFAULT_KXGAUGE_MAPPING) cfg = configparser.ConfigParser(interpolation=None) cfg.read(cfg_path, encoding="utf-8") sec = "kxgauge_mapping" if not cfg.has_section(sec): return dict(DEFAULT_KXGAUGE_MAPPING) out = dict(DEFAULT_KXGAUGE_MAPPING) for key, value in cfg.items(sec): if value.strip(): out[key] = value.strip().lower() return out def __init__(self, client: KobraXClient, args=None, store=None, printer_id: str = "1", all_bridges=None): self.client = client self._args = args self._printer_id = printer_id self._all_bridges = all_bridges if all_bridges is not None else {} self.ws_clients: set[web.WebSocketResponse] = set() # In-memory KV store for Moonraker /server/database/item (moonraker-obico, # mainsail presets etc.). Not persistent - does not survive a restart. self._moonraker_kv_store: dict[str, dict] = {} # Slot -> Orca filament profile mapping (from config.ini [filament_profiles]). # Format: {slot_idx: {"id": "OGFL01", "vendor": "Polymaker"}}. # Used in _build_lane_data so OrcaSlicer shows the concrete # brand ("PolyTerra PLA - Polymaker") instead of just "Generic PLA". try: import config_loader as _cl self._filament_profiles: dict[int, dict] = _cl.list_filament_profiles(self._printer_id) except Exception: self._filament_profiles = {} # Vendor visibility filter for the slot profile dropdown (Issue #41 option A). # Empty list = all vendors visible (backwards compatible). try: import config_loader as _cl self._visible_vendors: list[str] = _cl.list_visible_vendors(self._printer_id) except Exception: self._visible_vendors = [] self._last_state: dict = {} self._last_ams_set_request: dict | None = None self._state = { "nozzle_temp": 0.0, "nozzle_target": 0.0, "bed_temp": 0.0, "bed_target": 0.0, "print_state": "standby", "kobra_state": "free", "filename": "", "slicer_time": 0, "progress": 0.0, "print_duration": 0, "remain_time": 0, "curr_layer": 0, "total_layers": 0, # Layer heights for the currently running file (parsed from the # GCode header). Set in the upload path + in _fetch_from_store. # Obico uses currentZ from gcode_position[2] - the bridge computes # currentZ from curr_layer + these values in build_print_payload. "layer_height": 0.0, "first_layer_height": 0.0, "printer_name": env_loader.get("BRIDGE_PRINTER_NAME", "Anycubic Kobra X"), "firmware_version": "unknown", "upload_url": "", "camera_url": "", "fan_speed": 0, "light_on": False, "light_brightness": 80, "taskid": "-1", "print_speed_mode": 2, "connection_error": "", "file_ready": "", "filament_mismatch": None, "print_start_dialog": getattr(args, "print_start_dialog", 1), "filament_mode": "toolhead", "supplies_usage": 0, "ace_drying": {"status": 0, "target_temp": 0, "duration": 0, "remain_time": 0, "humidity": None, "current_temp": None}, "error_code": 0, "pause_msg": "", "storage_total_mb": 0, "storage_used_mb": 0, } self._ams_slots: list[dict] = [] # flat global list; each entry has global_index + box_id self._ams_loaded_slot: int = -1 # global slot index of currently loaded slot self._pending_load_slot: int = -1 # global slot index requested via /api/ams/feed type=1 self._ace_box_ids: list[int] = [] # detected ACE unit IDs (0..3) self._ace_auto_feed: dict[int, int] = {} # per-box auto_feed state (0/1) self._head_tools_model: int = -1 self._filament_mode: str = "toolhead" self._last_uploaded_file: str = "" # Pending waiters for a specific file/report `action` (e.g. "listLocal", # "deleteBatch"). publish()'s own return value for these actions is just # a generic immediate ACK skeleton (code=0, all fields empty) - the real # answer arrives later via the file/report callback (_on_file), same # as the existing fileDetails fire-and-forget pattern. Format: # {action: {"event": threading.Event(), "result": dict|None}}. self._file_action_waiters: dict[str, dict] = {} # Thumbnail cache for files on the printer's own storage (filename -> # base64 PNG string, "" if the file has no embedded thumbnail). # In-memory only - not persisted, cleared on restart. self._printer_thumbnail_cache: dict[str, str] = {} # Last buried/report payload (printer's own analytics event, fired once # per print start regardless of slicer - see reference_buried_report_trigger # memory). Carries gcode_size/estimate_duration/total_layers that are # otherwise unavailable for files not uploaded through the bridge itself # (Issue #102). Single entry only - just the most recent print. self._buried_cache: dict | None = None self._store = store if store is not None else GCodeStore(args.data_dir) self._serve_dir_path: str = self._store._gcode_dir self._current_job_id: str = "" # Filename of the file backing _current_job_id, kept alongside it so # the "finished" handler can still delete it from the printer's own # storage (Issue: delete-after-print) after self._state["filename"] # has already been cleared as part of the terminal-state reset below. self._current_job_filename: str = "" self._camera_autostarted: bool = False self._camera_user_stopped: bool = False # user manually stopped the camera during a print self.camera_cache: CameraCache = CameraCache() self._thumbnail_b64: str = "" self._ace_dry_presets: dict[str, dict] = self._load_ace_dry_presets_config() # Part skip: most recent skip list reported by the printer (v0.9.10) self._skip_state: dict = {"objects": [], "skipped": [], "ts": 0} # Pre-Print-Skip: pending until printer enters printing state self._pending_preprint_skip: list[str] = [] self._pending_preprint_skip_deadline: float = 0.0 # Spoolman filament tracking _sm_url = (getattr(args, "spoolman_server", "") or "").strip() self._spoolman: SpoolmanClient | None = ( SpoolmanClient(_sm_url, getattr(args, "spoolman_sync_rate", 0)) if _sm_url else None ) # Persistierte Spool-Zuordnung (AMS-Slot → Spoolman-Spool) je Drucker laden. # Fix: this used to reference `config_loader`, but the module alias is # `env_loader` (line 32) -> NameError swallowed by the bare `except`, # so persistence never loaded. Now via the local import + per printer. try: import config_loader as _cl self._spoolman_slot_spools: dict[int, int] = _cl.list_spool_map(self._printer_id) except Exception as _e: log.warning("Spoolman: failed to load slot map: %s", _e) self._spoolman_slot_spools = {} # {ams_slot_idx: spoolman_spool_id} self._spoolman_slot_usage: dict[int, float] = {} # per-slot accumulated mm this print self._spoolman_slot_reported: dict[int, float] = {} # per-slot mm already sent to Spoolman self._spoolman_last_usage: float = 0.0 # supplies_usage at last attribution tick self._spoolman_last_sync: float = 0.0 # KXGauge display (https://gitea.it-drui.de/viewit/kxgauge) _kg_url = (getattr(args, "kxgauge_url", "") or "").strip() _kg_enabled = bool(int(getattr(args, "kxgauge_enabled", 0) or 0)) try: _kg_heat_peak = float(getattr(args, "kxgauge_heat_peak", 250) or 250) except (TypeError, ValueError): _kg_heat_peak = 250.0 self._kxgauge: KXGaugeClient | None = ( KXGaugeClient(_kg_url, _kg_heat_peak) if (_kg_enabled and _kg_url) else None ) self._kxgauge_mapping: dict[str, str] = self._load_kxgauge_mapping_config() # Validate theme name (no special characters or umlauts) raw_theme = (getattr(args, "ui_theme", None) or "default").strip() if not _UI_THEME_NAME_RE.match(raw_theme): log.warning("Invalid UI theme name %r – using default", raw_theme) raw_theme = "default" self._ui_theme = raw_theme self._index_tpl_cache: str | None = None self._index_tpl_cache_key: tuple[str, float] | None = None # Register MQTT push callbacks client.callbacks["tempature/report"] = self._on_temp client.callbacks["print/report"] = self._on_print client.callbacks["info/report"] = self._on_info client.callbacks["file/report"] = self._on_file client.callbacks["buried/report"] = self._on_buried client.callbacks["multiColorBox/report"] = self._on_multicolor_box client.callbacks["light/report"] = self._on_light client.callbacks["skip/report"] = self._on_skip # Reachability is rechecked periodically (not just once at boot) so the # UI status dot reflects the printer's/Spoolman's actual current state # instead of freezing on the boot-time result. self._spoolman_reachable: bool = False self._spoolman_last_health_check: float = 0.0 if self._spoolman: def _check(): ok = self._spoolman.health_check() self._spoolman_reachable = ok self._spoolman_last_health_check = time.time() log.info(f"Spoolman: {'OK' if ok else 'unreachable'} at {self._spoolman.server_url}") threading.Thread(target=_check, daemon=True, name="spoolman-health").start() @staticmethod def _layer_height_from_filename(fname: str) -> float: """OrcaSlicer-Default-Filename-Pattern: `___.gcode` z.B. `adapter_e27_plate(01)_PLA_0.2_41m1s.gcode` → 0.2. Fallback when the GCode header was not parsed (e.g. file started directly on the slicer, or uploaded before v0.9.18). Returns 0.0 when the pattern does not match.""" import re if not fname: return 0.0 m = re.search(r"_(0\.\d+)_(\d+[hms])", fname) if not m: return 0.0 try: return float(m.group(1)) except Exception: return 0.0 def _estimate_current_z(self) -> float: """Estimates the current Z height from curr_layer + layer heights. The printer provides no real Z position via MQTT, but Obico (moonraker-obico/printer.py:267) reads currentZ from `gcode_position[2]`. We back-compute it with the layer_height from the GCode header: z = first_layer_height + (curr_layer - 1) * layer_height Values are set in the upload path and only reset on print cancel/end (slot/color changes do not affect them). If the values are missing (e.g. because the print was started directly on the slicer without an upload through the bridge), they are reloaded once from the GCode store. Returns 0.0 when nothing is known - Obico then shows keinen Z-Wert.""" s = self._state layer_h = float(s.get("layer_height") or 0.0) first_h = float(s.get("first_layer_height") or 0.0) fname = s.get("filename", "") if not layer_h and fname: try: gf = self._store.get_file_by_name(fname) if gf: layer_h = float(gf.get("layer_height") or 0.0) first_h = float(gf.get("first_layer_height") or layer_h) except Exception: pass if not layer_h and fname: # Last fallback: the OrcaSlicer default filename contains the layer height layer_h = self._layer_height_from_filename(fname) if layer_h and not first_h: first_h = layer_h if layer_h: # cache in state so not every build queries the store again s["layer_height"] = layer_h s["first_layer_height"] = first_h if not layer_h: return 0.0 curr = int(s.get("curr_layer") or 0) if curr <= 0: return 0.0 # Layer 1 = first_layer_height, Layer 2 = first + layer_h, … return round(first_h + max(0, curr - 1) * layer_h, 3) # ------------------------------------------------------------------------- # WebSocket push # ------------------------------------------------------------------------- # Static objects that never change at runtime. They are delivered once # via objects.query/subscribe, but NOT included in every # notify_status_update - otherwise Mobileraker's # ConfigFile.parse (expensive + strict) runs on every status tick and the app # hangs/crashes on refresh (Issue #48). _STATIC_STATUS_OBJECTS = ("configfile", "webhooks", "heaters", "history") def _push_status_update(self): if not self.ws_clients: return objs = self._build_printer_objects() live = {k: v for k, v in objs.items() if k not in self._STATIC_STATUS_OBJECTS} msg = { "jsonrpc": "2.0", "method": "notify_status_update", "params": [live, time.time()], } text = json.dumps(msg) dead = set() for ws in self.ws_clients: try: asyncio.run_coroutine_threadsafe(ws.send_str(text), ws._loop) except Exception: dead.add(ws) self.ws_clients -= dead def _build_mmu_object(self) -> dict: # POSITIONSTREU: ein Gate je physischem Slot, in Reihenfolge. Leere Slots # get gate_status=0 (instead of being omitted) - otherwise the # Farben in OrcaSlicer auf falsche Gates (Slot 1=gelb, 2=leer, 3=rot → # red must not land on gate 1). gate_status 0=empty, 1=available. slots = sorted( ((int(s.get("global_index", i)), s) for i, s in enumerate(self._ams_slots)), key=lambda item: item[0], ) if not slots: return {} _TEMP = {"PLA": 210, "PETG": 230, "ABS": 240, "ASA": 250, "TPU": 220, "PA": 260, "PC": 270, "HIPS": 220} num_gates = len(slots) gate_status, gate_material, gate_color, gate_temperature, gate_color_rgb = [], [], [], [], [] gate_filament_name = [] gate_spool_id = [] for _global_index, slot in slots: occupied = slot.get("status") == 5 gate_status.append(1 if occupied else 0) material = self._normalize_material(slot.get("type") or "PLA") if occupied else "" gate_material.append(material) c = slot.get("color", [0, 0, 0]) if occupied else [0, 0, 0] # Happy Hare expects gate_color as RRGGBB WITHOUT '#' (Klipper limitation). # Leerer Gate: leerer String + RGB [0,0,0]. gate_color.append("{:02X}{:02X}{:02X}".format(*c[:3]) if occupied else "") gate_color_rgb.append([round(c[0]/255, 3), round(c[1]/255, 3), round(c[2]/255, 3)] if occupied else [0.0, 0.0, 0.0]) gate_temperature.append(_TEMP.get(material, 210) if occupied else 0) # gate_filament_name from user override or material default for the # HH-Pfad in OrcaSlicer (fetch_hh_filament_info). Wenn Orca den # HH path (MMU detection), PR #13719 evaluates this field as a # preset name -> 'Anycubic PLA' matches the printer-specific # preset; an empty string previously led to Generic PLA. if occupied: # Stale-profile guard (see _effective_slot_profile): only apply the # override while its material family still matches the loaded filament. user_profile = self._effective_slot_profile(_global_index, material) fila_name = user_profile.get("name") or self._default_filament_name(material) gate_filament_name.append(fila_name) else: gate_filament_name.append("") # Spoolman spool ID per gate from the (printer-specific) slot map so # Happy Hare/OrcaSlicer can show the bound spool (-1 = none). gate_spool_id.append(self._spoolman_slot_spools.get(_global_index, -1) if occupied else -1) loaded_index_map = {global_index: idx for idx, (global_index, _) in enumerate(slots)} active_gate = loaded_index_map.get(int(self._ams_loaded_slot), -1) return { "num_gates": num_gates, "enabled": True, "gate_status": gate_status, "gate_material": gate_material, "gate_color": gate_color, "gate_temperature": gate_temperature, "gate_color_rgb": gate_color_rgb, "gate_filament_name": gate_filament_name, "gate_spool_id": gate_spool_id, "ttg_map": list(range(num_gates)), "tool": active_gate, "gate": active_gate, } def _default_filament_name(self, material: str) -> str: """Default name for `gate_filament_name`/`name` in lane_data when no user override is set. Deliberate design decision: **always Generic ** as the default - the library profile is `compatible_printers:[]` (= compatible with every printer) and therefore guaranteed to be visible. OrcaSlicer then matches the neutral generic preset and the user can set a concrete brand per slot if they want to.""" if not material: return "" mat = self._normalize_material(material) profs = self._load_orca_filaments() # Varianten-Mapping: Drucker meldet z.B. "PLA SILK", OrcaSlicer speichert # all variants under type=PLA with the variant name in the name field. _VARIANT_NAME = { "PLA SILK": "Generic PLA Silk", "PLA MATTE": "Generic PLA Matte", "PLA+": "Generic PLA", "PLA-CF": "Generic PLA-CF", "PETG-CF": "Generic PETG-CF", } if mat in _VARIANT_NAME: target = _VARIANT_NAME[mat] for p in profs: if p.get("vendor") == "Generic" and p.get("name") == target: return p["name"] def _match_type(p: dict) -> bool: pt = (p.get("type") or "").upper() return pt == mat or pt.startswith(mat + "-") or pt.startswith(mat + " ") # Library-Generic-Profil (immer is_visible+is_compatible) for p in profs: if p.get("vendor") == "Generic" and p.get("name", "").startswith("Generic ") and _match_type(p): return p.get("name", "") # If the library generic for this exotic material type is missing, # we return nothing - OrcaSlicer falls back to filament_id_by_type. return "" def _build_printer_objects(self) -> dict: s = self._state return { "extruder": { "temperature": s["nozzle_temp"], "target": s["nozzle_target"], "power": 0.0, }, "heater_bed": { "temperature": s["bed_temp"], "target": s["bed_target"], "power": 0.0, }, "print_stats": { "state": s["print_state"], "filename": s["filename"], "print_duration": s["print_duration"], "total_duration": s["print_duration"], "remain_time": s["remain_time"], "info": { "current_layer": s["curr_layer"], "total_layer": s["total_layers"], }, }, "display_status": { "progress": s["progress"], "message": "", }, "virtual_sdcard": { "progress": s["progress"], "is_active": s["print_state"] == "printing", "file_path": s["filename"], # file_position approximiert: fraction × est_total_size. # The printer does not provide an exact value; Obico only uses it for display. "file_position": int(s["progress"] * 1_000_000) if s["progress"] else 0, }, "toolhead": { "position": [0, 0, 0, 0], "homed_axes": "xyz", "print_time": s["print_duration"], "estimated_print_time": s["print_duration"], }, "mmu": self._build_mmu_object(), # -- Moonraker compatibility for moonraker-obico -- "heaters": { "available_heaters": ["extruder", "heater_bed"], "available_sensors": [], "available_monitors": [], }, "webhooks": { "state": "ready", "state_message": "Printer is ready", }, # speed_factor: 1=silent(0.5) / 2=standard(1.0) / 3=high(1.3) / 4=ultra(1.5) # Estimate the current Z height for Obico from curr_layer + layer heights # (the printer provides no real Z position via MQTT). gcode_position[2] # is the value moonraker-obico reads as currentZ in printer.py. "gcode_move": { "speed_factor": {1: 0.5, 2: 1.0, 3: 1.3, 4: 1.5}.get(int(s.get("print_speed_mode") or 2), 1.0), "extrude_factor": 1.0, "speed": 0, "gcode_position": [0, 0, self._estimate_current_z(), 0], "absolute_coordinates": True, "absolute_extrude": True, "homing_origin": [0, 0, 0, 0], "position": [0, 0, self._estimate_current_z(), 0], }, # motion_report: Mobileraker reads the live velocity here # (live_velocity). The Kobra X MQTT provides NO real mm/s, only # a print_speed_mode (1-4). live_velocity therefore stays 0 - but the # object must exist, otherwise Mobileraker displays nothing # (motion_report used to be null). live_position mirrors the # estimated Z height (like gcode_move). "motion_report": { "live_position": [0, 0, self._estimate_current_z(), 0], "live_velocity": 0.0, "live_extruder_velocity": 0.0, }, "fan": { "speed": (int(s.get("fan_speed") or 0)) / 100.0, "rpm": None, }, # history (object): Obico subscribes to it as an object; the actual # /server/history/list endpoint delivers the real list separately. "history": { "job_totals": { "total_jobs": 0, "total_time": 0, "total_print_time": 0, "total_filament_used": 0.0, "longest_job": 0, "longest_print": 0, }, "current_job": None, }, # Pseudo Klipper macros for moonraker-obico: # - _OBICO_LAYER_CHANGE reports the current layer number. Obico uses this # for "first layer scan" triggers and layer-aligned time-lapse frames. # We feed this from the MQTT stream (s["curr_layer"]). # - TIMELAPSE_TAKE_FRAME signals that the current pause comes from the # time-lapse (otherwise Obico would interpret the pause as a user # pause). We set is_paused=False because our pauses are # never time-lapse pauses. "gcode_macro _OBICO_LAYER_CHANGE": { "current_layer": int(s.get("curr_layer") or 0), "first_layer_scanning": False, "first_layer_scan_enabled": False, }, "gcode_macro TIMELAPSE_TAKE_FRAME": { "is_paused": False, }, # configfile stub - Mobileraker and other clients crash without # this object (Missing field: configFile). Values from the # decrypted avata_main.conf (ACCFG1.0 - Kobra X firmware). # Mobileraker (Issue #48) parses BOTH branches config + settings via # denselben ConfigFile.parse → ConfigExtruder.fromJson; ein leeres # config:{} crashed the non-nullable Dart parser. Therefore # config identisch zu settings gespiegelt. "configfile": self._klipper_configfile_stub(), } def _klipper_configfile_stub(self) -> dict: """Minimal Klipper configfile stub for Mobileraker/OctoApp (Issue #48). Mobileraker parses BOTH branches `config` and `settings` through the same ConfigFile.parse → ConfigExtruder.fromJson. Ein leeres `config: {}` crashed the non-nullable Dart parser, therefore `config` is mirrored identically to `settings`. Values from the decrypted avata_main.conf (ACCFG1.0 — Kobra X Firmware). """ settings = { "printer": { "kinematics": "cartesian", "max_velocity": 450, "max_accel": 10000, "max_z_velocity": 12, "max_z_accel": 100, "square_corner_velocity": 20.0, }, "extruder": { "nozzle_diameter": 0.4, "filament_diameter": 1.75, "sensor_type": "ATC Semitec 104GT-2", "min_temp": 0, "max_temp": 320, "min_extrude_temp": 10, # Mobileraker ConfigExtruder erwartet diese Felder non-nullable # (max_extrude_only_distance, max_power) or present as a key # (max_extrude_only_velocity/accel may be null). Missing = # Crash in ConfigExtruder.fromJson (Issue #48). "max_extrude_only_distance": 100.0, "max_power": 1.0, "max_extrude_only_velocity": None, "max_extrude_only_accel": None, }, "heater_bed": { # Mobileraker ConfigHeaterBed: heater_pin, sensor_type, control # are non-nullable. Values are placeholders (the bridge does not know # the real pins - Anycubic firmware, no Klipper printer.cfg). "heater_pin": "PA0", "sensor_type": "ATC Semitec 104GT-2", "control": "pid", "min_temp": 0, "max_temp": 120, }, # Fill stepper_* with non-nullable required fields (step_pin, dir_pin, # rotation_distance), otherwise ConfigStepper.fromJson crashes. "stepper_x": {"step_pin": "PA1", "dir_pin": "PA2", "rotation_distance": 40, "position_min": -18.5, "position_max": 280}, "stepper_y": {"step_pin": "PA3", "dir_pin": "PA4", "rotation_distance": 40, "position_min": -6.5, "position_max": 272.5}, "stepper_z": {"step_pin": "PA5", "dir_pin": "PA6", "rotation_distance": 8, "position_min": -4, "position_max": 262}, "virtual_sdcard": {"path": "/data/gcodes"}, "pause_resume": {}, "display_status": {}, } # config + settings must contain the same fields - Mobileraker # parses both. deepcopy so no client is affected by a shared reference # versehentlich beide Zweige mutiert. return { "config": copy.deepcopy(settings), "settings": settings, "warnings": [], "save_config_pending": False, "save_config_pending_items": {}, } # ------------------------------------------------------------------------- # /kx/ API handlers (GCode Store, History, Filament) # ------------------------------------------------------------------------- _CORS = { "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Methods": "GET, POST, DELETE, OPTIONS", "Access-Control-Allow-Headers": "Content-Type", } def _json_cors(self, data, status=200): return web.json_response(data, status=status, headers=self._CORS) # ------------------------------------------------------------------------- # WebSocket handler # ------------------------------------------------------------------------- async def handle_websocket(self, request): ws = web.WebSocketResponse(heartbeat=30) await ws.prepare(request) ws._loop = asyncio.get_event_loop() self.ws_clients.add(ws) log.info(f"WS client connected ({len(self.ws_clients)} total)") # Send klippy_ready notification await ws.send_str(json.dumps({ "jsonrpc": "2.0", "method": "notify_klippy_ready", "params": [], })) # Send initial status await ws.send_str(json.dumps({ "jsonrpc": "2.0", "method": "notify_status_update", "params": [self._build_printer_objects(), time.time()], })) async for msg in ws: if msg.type == aiohttp.WSMsgType.TEXT: await self._handle_ws_rpc(ws, msg.data) elif msg.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSE): break self.ws_clients.discard(ws) log.info(f"WS client disconnected ({len(self.ws_clients)} remaining)") return ws async def _handle_ws_rpc(self, ws: web.WebSocketResponse, raw: str): try: req = json.loads(raw) except Exception: return rpc_id = req.get("id") method = req.get("method", "") log.info(f"WS RPC: {method} params={str(req.get('params',''))[:120]}") params = req.get("params") or {} if isinstance(params, list): params = params[0] if params else {} result = None error = None try: if method in ("printer.info", "printer_info"): result = { "state": "ready", "state_message": "Printer is ready", "hostname": "kobrax-bridge", "software_version": KLIPPER_VERSION, "cpu_info": self._state["printer_name"], "klipper_path": "/home/pi/klipper", "python_path": "/home/pi/klippy-env/bin/python", } elif method in ("server.info", "server_info"): result = { "klippy_connected": True, "klippy_state": "ready", "moonraker_version": MOONRAKER_VERSION, "components": [], "failed_components": [], "registered_directories": ["gcodes"], "warnings": [], } elif method in ("printer.objects.list",): result = {"objects": list(self._build_printer_objects().keys())} elif method in ("printer.objects.query", "printer.objects.get"): objects = params.get("objects", {}) all_objs = self._build_printer_objects() if objects: filtered = {k: all_objs.get(k, {}) for k in objects} else: filtered = all_objs result = {"status": filtered, "eventtime": time.time()} elif method == "printer.objects.subscribe": objects = params.get("objects", {}) all_objs = self._build_printer_objects() if objects: filtered = {k: all_objs.get(k, {}) for k in objects} else: filtered = all_objs result = {"status": filtered, "eventtime": time.time()} elif method == "printer.print.start": filename = params.get("filename", self._last_uploaded_file) loop = asyncio.get_event_loop() resp = await loop.run_in_executor( None, lambda: self.client.publish("print", "start", {"filename": filename, "use_ams": False}, timeout=15.0) ) result = "ok" if resp else "timeout" elif method == "printer.print.pause": loop = asyncio.get_event_loop() await loop.run_in_executor(None, self.client.pause_print) result = "ok" elif method == "printer.print.resume": loop = asyncio.get_event_loop() await loop.run_in_executor(None, self.client.resume_print) result = "ok" elif method == "printer.print.cancel": loop = asyncio.get_event_loop() await loop.run_in_executor(None, self.client.stop_print) result = "ok" elif method == "machine.system_info": result = {"system_info": {"cpu_info": {"cpu_desc": "Kobra X Bridge"}}} elif method == "server.files.list": result = [] # ── moonraker-obico passthru-Targets ── elif method == "printer.gcode.script": script = (params.get("script") or "").strip().upper() if isinstance(params, dict) else "" result = await self._exec_gcode_script(script) elif method in ("server.connection.identify",): # Obico identifies itself on connect. Connection ID doesn't matter. result = {"connection_id": 1} elif method == "connection.register_remote_method": # Obico registriert obico_remote_event-Callback. Wir akzeptieren leer. result = "ok" elif method == "server.webcams.list": # WS variant: absolute URL with the real LAN IP instead of localhost _lip = getattr(self, "_local_ip", None) or "127.0.0.1" _base = f"http://{_lip}:{self._args.port}" result = {"webcams": [{ "name": "KX-Bridge", "location": "printer", "service": "mjpegstreamer", "enabled": True, "stream_url": f"{_base}/api/camera/stream", "snapshot_url": f"{_base}/api/camera/snapshot", "flip_horizontal": False, "flip_vertical": False, "rotation": 0, "target_fps": 5, "aspect_ratio": "16:9", }]} elif method == "server.history.list": # Reuse the HTTP handler logic (Moonraker schema with Unix TS). try: jobs = self._store.list_jobs(limit=50) or [] except Exception: jobs = [] from datetime import datetime, timezone as _tz def _ts(iso): try: return datetime.strptime(iso, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=_tz.utc).timestamp() except Exception: return 0.0 result = {"count": len(jobs), "jobs": [ {"job_id": j.get("id"), "exists": True, "filename": j.get("filename",""), "status": j.get("status") or "completed", "print_duration": j.get("duration_sec") or 0, "total_duration": j.get("duration_sec") or 0, "start_time": _ts(j.get("started_at")), "end_time": (_ts(j.get("started_at")) + (j.get("duration_sec") or 0)) if j.get("started_at") and j.get("duration_sec") else None, "filament_used": 0.0, "metadata": {}} for j in jobs ]} elif method == "machine.update.status": result = {"busy": False, "version_info": {}} elif method == "server.files.metadata": # Obico + Mobileraker request metadata for a file. Same # logic as the HTTP endpoint (previously a separate broken path with # a non-existent store method -> empty response -> # Mobileraker-Endlosschleife, Issue #48). fname = (params or {}).get("filename") if isinstance(params, dict) else None fname = fname or self._state.get("filename", "") result = self._build_file_metadata(fname) if fname else {} else: log.debug(f"Unbekannte RPC-Methode: {method}") result = {} except Exception as e: log.error(f"RPC error for {method}: {e}") error = {"code": -32603, "message": str(e)} if rpc_id is not None: response = {"jsonrpc": "2.0", "id": rpc_id} if error: response["error"] = error else: response["result"] = result await ws.send_str(json.dumps(response)) # ------------------------------------------------------------------------- # Poll loop (sync, runs in executor) # ------------------------------------------------------------------------- def _printer_reachable(self) -> bool: """TCP probe on the MQTT port - no ICMP needed, no root required.""" import socket as _socket try: with _socket.create_connection( (self._args.printer_ip, self._args.mqtt_port), timeout=2.0 ): return True except OSError: return False def _poll_loop(self, stop_event: threading.Event): _offline = self._state["kobra_state"] == "offline" _probe_interval = 10.0 # Sekunden zwischen TCP-Probes im Offline-Modus while not stop_event.is_set(): # ── Offline-Modus: warten bis Drucker wieder erreichbar ────────── if _offline: if self._printer_reachable(): log.info("Printer reachable - establishing MQTT connection...") try: self.client.connect() _offline = False self._state["print_state"] = "standby" self._state["kobra_state"] = "free" self._kxgauge_notify_state("free") self._state["connection_error"] = "" log.info("MQTT connection re-established") except Exception as e: err = _mqtt_error_msg(e) self._state["connection_error"] = err log.warning(f"Connection attempt failed: {err}") stop_event.wait(_probe_interval) continue else: stop_event.wait(_probe_interval) continue # ── Online-Modus: normaler Poll ────────────────────────────────── try: info = self.client.query_info() if info: self._on_info(info) elif not self.client.is_connected(): # publish() swallows send/reconnect failures internally and # just returns None (Issue #105) - a falsy `info` alone # doesn't distinguish "printer sent nothing this tick" from # "the MQTT session itself is dead". Check is_connected() # explicitly so a dead session gets routed into the same # clean offline/reconnect path as a TCP-unreachable printer, # instead of silently retrying every poll_interval forever. log.warning("MQTT connection lost (query returned no response) - switching to offline mode") self._state["print_state"] = "error" self._state["kobra_state"] = "offline" self._kxgauge_notify_state("offline") self._state["connection_error"] = f"MQTT connection lost ({self._args.printer_ip})" try: self.client.disconnect() except Exception: pass _offline = True stop_event.wait(getattr(self._args, "poll_interval", 3)) continue # While printing: query print/report directly if self._state["print_state"] in ("printing", "preheating", "auto_leveling", "checking", "init"): print_r = self.client.publish("print", "query", timeout=3.0) if print_r: self._on_print(print_r) # Spoolman mid-print sync if (self._spoolman and self._spoolman.sync_rate > 0 and self._spoolman_slot_spools and self._state.get("print_state") == "printing"): now = time.time() if now - self._spoolman_last_sync >= self._spoolman.sync_rate: self._spoolman_sync_midprint() self._spoolman_last_sync = now box = self.client.query_multicolor_box() if box: data = box.get("data") or {} self._head_tools_model = int(data.get("head_tools_model", self._head_tools_model)) boxes = data.get("multi_color_box") or [] if boxes: self._update_ace_drying_state(data, boxes) self._filament_mode = self._detect_filament_mode(boxes, self._head_tools_model) self._state["filament_mode"] = self._filament_mode global_slots, global_loaded = self._aggregate_slots(boxes, self._filament_mode) activity_map = self._slot_activity_map(boxes, global_loaded) for s in global_slots: s["activity"] = activity_map.get(s.get("global_index"), "") if global_slots: self._ams_slots = global_slots self._ams_loaded_slot = global_loaded self._spoolman_attribute_tick(activity_map) else: # No multiColorBox data — still attribute (no transitions to skip) self._spoolman_attribute_tick({}) # Recheck Spoolman reachability periodically so the UI status # dot reflects the current state, not just the boot-time result. if self._spoolman and time.time() - self._spoolman_last_health_check >= 30.0: self._spoolman_reachable = self._spoolman.health_check() self._spoolman_last_health_check = time.time() except Exception as e: log.warning(f"Poll error: {e}") # Check whether the printer is really gone if not self._printer_reachable(): log.info("Printer unreachable - switching to offline mode") self._state["print_state"] = "error" self._state["kobra_state"] = "offline" self._kxgauge_notify_state("offline") self._state["connection_error"] = f"Printer unreachable ({self._args.printer_ip})" try: self.client.disconnect() except Exception: pass _offline = True stop_event.wait(getattr(self._args, "poll_interval", 3)) # --------------------------------------------------------------------------- # App factory + main # --------------------------------------------------------------------------- def _mqtt_error_msg(exc: Exception) -> str: msg = str(exc) if "20020005" in msg: return "Wrong MQTT credentials (username, password or device ID incorrect)" return msg @web.middleware async def cors_middleware(request, handler): if request.method == "OPTIONS": return web.Response(status=204, headers={ "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Methods": "GET, POST, DELETE, OPTIONS", "Access-Control-Allow-Headers": "Content-Type", }) resp = await handler(request) resp.headers["Access-Control-Allow-Origin"] = "*" return resp def build_app(bridge: KobraXBridge) -> web.Application: app = web.Application( client_max_size=256 * 1024 * 1024, middlewares=[cors_middleware], ) r = app.router # Moonraker API r.add_get("/server/info", bridge.handle_server_info) r.add_get("/printer/info", bridge.handle_printer_info) r.add_get("/machine/system_info", bridge.handle_machine_system_info) r.add_get("/printer/objects/list", bridge.handle_objects_list) r.add_get("/printer/objects/query", bridge.handle_objects_query) r.add_get("/printer/objects/subscribe", bridge.handle_objects_subscribe) r.add_post("/printer/objects/subscribe", bridge.handle_objects_subscribe) r.add_get("/server/files/list", bridge.handle_files_list) r.add_get("/server/files/metadata", bridge.handle_files_metadata) r.add_post("/server/files/upload", bridge.handle_file_upload) r.add_post("/printer/print/start", bridge.handle_print_start) r.add_post("/printer/print/pause", bridge.handle_print_pause) r.add_post("/printer/print/resume", bridge.handle_print_resume) r.add_post("/printer/print/cancel", bridge.handle_print_cancel) # Moonraker stubs for moonraker-obico r.add_get("/access/api_key", bridge.handle_access_api_key) r.add_get("/machine/update/status", bridge.handle_machine_update_status) r.add_get("/server/history/list", bridge.handle_history_list) r.add_get("/server/webcams/list", bridge.handle_webcams_list) r.add_post("/printer/gcode/script", bridge.handle_printer_gcode_script) # OctoPrint compatibility (OrcaSlicer probes this + uploads here) r.add_get("/api/version", bridge.handle_octoprint_version) r.add_post("/api/files/local", bridge.handle_file_upload) r.add_post("/api/files/{path:.*}", bridge.handle_file_upload) # Moonraker database (OrcaSlicer AMS-Sync) r.add_get("/server/database/item", bridge.handle_moonraker_database) r.add_post("/server/database/item", bridge.handle_moonraker_database_post) r.add_get("/server/database/list", bridge.handle_database_list) # New API endpoints r.add_post("/api/light", bridge.handle_api_light) r.add_post("/api/fan", bridge.handle_api_fan) r.add_post("/api/connect", bridge.handle_api_connect) r.add_post("/api/disconnect", bridge.handle_api_disconnect) r.add_post("/api/restart", bridge.handle_api_restart) r.add_post("/api/kxgauge/test", bridge.handle_kx_kxgauge_test) r.add_post("/api/speed", bridge.handle_api_speed) r.add_post("/api/ams/feed", bridge.handle_api_ams_feed) r.add_post("/api/ams/set_slot", bridge.handle_api_ams_set_slot) r.add_post("/api/ace/auto_feed", bridge.handle_api_ace_auto_feed) r.add_post("/api/ace/dry", bridge.handle_api_ace_dry) r.add_post("/api/axis", bridge.handle_api_axis) r.add_post("/api/temperature", bridge.handle_api_temperature) r.add_get("/api/camera", bridge.handle_api_camera) r.add_get("/api/camera/stream", bridge.handle_camera_stream) r.add_get("/api/camera/h264", bridge.handle_camera_h264) r.add_get("/api/camera/snapshot", bridge.handle_api_camera_snapshot) r.add_post("/api/camera/start", bridge.handle_api_camera_start) r.add_post("/api/camera/stop", bridge.handle_api_camera_stop) r.add_post("/api/camera/reset", bridge.handle_api_camera_reset) r.add_get("/api/state", bridge.handle_api_state) r.add_get("/api/settings", bridge.handle_api_settings_get) r.add_post("/api/settings", bridge.handle_api_settings_post) r.add_get("/api/update/check", bridge.handle_api_update_check) r.add_post("/api/update/apply", bridge.handle_api_update_apply) r.add_post("/api/file_ready/clear", bridge.handle_api_file_ready_clear) r.add_get("/api/log/stream", bridge.handle_api_log_stream) r.add_get("/api/log/download", bridge.handle_api_log_download) r.add_get("/serve/{filename}", bridge.handle_serve_file) # /kx/ GCode Store + History + Filament r.add_get("/kx/printers", bridge.handle_kx_printers) r.add_post("/kx/printers/add", bridge.handle_kx_printer_add) r.add_delete("/kx/printers/{pid}", bridge.handle_kx_printer_remove) r.add_post("/kx/printers/{pid}/power", bridge.handle_kx_printer_power) r.add_get("/kx/printers/{pid}/power-status", bridge.handle_kx_printer_power_status) r.add_post("/kx/print", bridge.handle_kx_print) r.add_get("/kx/files", bridge.handle_kx_files) r.add_delete("/kx/files/{file_id}", bridge.handle_kx_file_delete) r.add_get("/kx/files/{file_id}/download", bridge.handle_kx_file_download) r.add_post("/kx/files/{file_id}/verify", bridge.handle_kx_file_verify) r.add_get("/kx/printer-files", bridge.handle_kx_printer_files) r.add_post("/kx/printer-files/delete", bridge.handle_kx_printer_file_delete) r.add_get("/kx/printer-files/{filename}/thumbnail", bridge.handle_kx_printer_file_thumbnail) r.add_get("/kx/filament/slots", bridge.handle_kx_filament_slots) r.add_get("/kx/filament/profiles", bridge.handle_kx_filament_profiles) r.add_post("/kx/filament/slots/{idx}/profile", bridge.handle_kx_filament_slot_profile) r.add_get("/kx/filament/visible_vendors", bridge.handle_kx_visible_vendors) r.add_post("/kx/filament/visible_vendors", bridge.handle_kx_visible_vendors) # Custom profile import (Issue #41) - the user uploads their own Orca filament # profiles as ZIP/JSON (e.g. from ~/.config/OrcaSlicer/user//filament/), # because the bridge typically does not run on the same host as OrcaSlicer. r.add_get("/kx/filament/profiles/user", bridge.handle_kx_filament_profiles_user_list) r.add_post("/kx/filament/profiles/user", bridge.handle_kx_filament_profiles_import) r.add_delete("/kx/filament/profiles/user", bridge.handle_kx_filament_profiles_user_delete) r.add_get("/kx/history", bridge.handle_kx_history) r.add_get("/kx/ui/{name:.*}", bridge.handle_kx_ui_asset) r.add_get("/kx/files/{id}/objects", bridge.handle_kx_file_objects) r.add_post("/kx/skip", bridge.handle_kx_skip) r.add_post("/kx/skip/query", bridge.handle_kx_skip_query) r.add_get("/kx/skip/state", bridge.handle_kx_skip_state) r.add_get("/kx/spoolman/status", bridge.handle_kx_spoolman_status) r.add_get("/kx/spoolman/spools", bridge.handle_kx_spoolman_spools) r.add_post("/kx/spoolman/active-spool", bridge.handle_kx_spoolman_set_active) r.add_route("OPTIONS", "/kx/{path:.*}", bridge.handle_kx_options) # Root + Printer-Routen (Single-Page, JS liest Pathname) r.add_get("/", bridge.handle_index) r.add_get(r"/printer{num:\d+}", bridge.handle_index) r.add_get("/favicon.ico", bridge.handle_favicon) # WebSocket r.add_get("/websocket", bridge.handle_websocket) # Catch-all: log all unknown requests instead of 404 r.add_route("*", "/{path:.*}", bridge.handle_catchall) return app def _build_per_printer_args(base_args, p: dict): """Copy CLI args, override with the printer entry from config.ini.""" import copy a = copy.copy(base_args) a.printer_ip = p.get("printer_ip") or base_args.printer_ip a.mqtt_port = int(p.get("mqtt_port") or base_args.mqtt_port) a.username = p.get("username") or base_args.username a.password = p.get("password") or base_args.password a.mode_id = p.get("mode_id") or base_args.mode_id a.device_id = p.get("device_id") or base_args.device_id a.port = int(p.get("http_port") or base_args.port) a.power_on_url = p.get("power_on_url") or getattr(base_args, "power_on_url", "") or "" a.power_off_url = p.get("power_off_url") or getattr(base_args, "power_off_url", "") or "" a.power_status_url = p.get("power_status_url") or getattr(base_args, "power_status_url", "") or "" a.power_status_inverted = int(p.get("power_status_inverted") or getattr(base_args, "power_status_inverted", 0) or 0) a.kxgauge_url = p.get("kxgauge_url") or getattr(base_args, "kxgauge_url", "") or "" a.kxgauge_enabled = p.get("kxgauge_enabled") or getattr(base_args, "kxgauge_enabled", 0) or 0 a.kxgauge_heat_peak = p.get("kxgauge_heat_peak") or getattr(base_args, "kxgauge_heat_peak", 250) or 250 return a async def run_bridge(args): _set_verbose_http_log(bool(getattr(args, "verbose_http_log", 0))) printers = env_loader.list_printers() multi_mode = bool(printers) if not printers: printers = [{ "id": "1", "name": getattr(args, "printer_name", None) or "Anycubic Kobra X", "printer_ip": args.printer_ip, "mqtt_port": args.mqtt_port, "username": args.username, "password": args.password, "mode_id": args.mode_id, "device_id": args.device_id, "http_port": args.port, }] store = GCodeStore(args.data_dir) all_bridges: dict = {} runners = [] stop_event = threading.Event() loop = asyncio.get_event_loop() for idx, p in enumerate(printers): pid = str(p.get("id") or (idx + 1)) per_args = _build_per_printer_args(args, p) # Default port convention: 7125 + (id-1) when no http_port is set if not p.get("http_port") and multi_mode: try: per_args.port = 7125 + (int(pid) - 1) except ValueError: per_args.port = 7125 + idx client = KobraXClient( host=per_args.printer_ip, port=per_args.mqtt_port, username=per_args.username, password=per_args.password, mode_id=per_args.mode_id, device_id=per_args.device_id, client_id=f"kobrax_bridge_{pid}", ) bridge = KobraXBridge( client, args=per_args, store=store, printer_id=pid, all_bridges=all_bridges, ) # Adopt printer_name from config.ini if set if p.get("name"): bridge._state["printer_name"] = p["name"] bridge._name_locked = True all_bridges[pid] = bridge log.info(f"[Printer {pid}] Connecting to {per_args.printer_ip}:{per_args.mqtt_port}...") try: await loop.run_in_executor(None, client.connect) log.info(f"[Printer {pid}] MQTT connected") except Exception as e: err = _mqtt_error_msg(e) log.warning(f"[Printer {pid}] Connection failed: {err} - offline mode") bridge._state["print_state"] = "error" bridge._state["kobra_state"] = "offline" bridge._kxgauge_notify_state("offline") bridge._state["connection_error"] = err threading.Thread( target=bridge._poll_loop, args=(stop_event,), daemon=True, name=f"poll-{pid}", ).start() app = build_app(bridge) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, args.host, per_args.port) await site.start() runners.append((runner, client, pid)) import socket as _socket _in_docker = os.path.exists("/.dockerenv") _host_ip_override = env_loader.BRIDGE_HOST_IP.strip() if _host_ip_override: _local_ip = _host_ip_override else: try: with _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM) as _s: _s.connect(("8.8.8.8", 80)) _local_ip = _s.getsockname()[0] except Exception: _local_ip = args.host # Propagate to all bridge instances - used for absolute webcam URLs for _b in all_bridges.values(): _b._local_ip = _local_ip ports = ", ".join(str(getattr(b._args, 'port', 0)) for b in all_bridges.values()) if _in_docker and not _host_ip_override: # In a container the UDP trick only yields the Docker-internal IP - don't show it log.info(f"OrcaSlicer → Klipper → http://:{ports}") log.info("Running in Docker — set BRIDGE_HOST_IP to show the exact address") else: log.info(f"OrcaSlicer → Klipper → http://{_local_ip}:{ports}") log.info("Press Ctrl-C to stop") try: while True: await asyncio.sleep(3600) except (KeyboardInterrupt, asyncio.CancelledError): pass finally: stop_event.set() for runner, client, pid in runners: try: await runner.cleanup() except Exception: pass try: client.disconnect() except Exception: pass log.info("Bridge stopped") def _default_data_dir() -> str: """Persistenz-Verzeichnis: Docker setzt KX_DATA_DIR, Binary nutzt /data, Dev script uses /data (or /app/data if present).""" if os.environ.get("KX_DATA_DIR"): return os.environ["KX_DATA_DIR"] if getattr(sys, "frozen", False): return os.path.join(os.path.dirname(sys.executable), "data") if os.path.isdir("/app"): return "/app/data" return os.path.normpath(os.path.join(_BASE, "..", "data")) def main(): parser = argparse.ArgumentParser(description="Moonraker bridge for the Anycubic Kobra X") parser.add_argument("--printer-ip", default=env_loader.PRINTER_IP, help="IP-Adresse des Druckers") parser.add_argument("--mqtt-port", type=int, default=env_loader.MQTT_PORT) parser.add_argument("--username", default=env_loader.USERNAME) parser.add_argument("--password", default=env_loader.PASSWORD) parser.add_argument("--mode-id", default=env_loader.MODE_ID) parser.add_argument("--device-id", default=env_loader.DEVICE_ID) parser.add_argument("--power-on-url", default=env_loader.POWER_ON_URL, help="HTTP GET URL to power the printer on (e.g. a Tasmota smart plug)") parser.add_argument("--power-off-url", default=env_loader.POWER_OFF_URL, help="HTTP GET URL to power the printer off") parser.add_argument("--power-status-url", default=env_loader.POWER_STATUS_URL, help="HTTP GET URL returning the smart plug's current on/off state") parser.add_argument("--power-status-inverted", type=int, default=env_loader.POWER_STATUS_INVERTED, help="Show the power switch's current state instead of the toggle action") parser.add_argument("--default-ams-slot",default=env_loader.DEFAULT_AMS_SLOT) parser.add_argument("--auto-leveling", type=int, default=env_loader.AUTO_LEVELING) parser.add_argument("--vibration-compensation", type=int, default=env_loader.VIBRATION_COMPENSATION) parser.add_argument("--camera-on-print", type=int, default=env_loader.CAMERA_ON_PRINT) parser.add_argument("--web-upload-warning", type=int, default=env_loader.WEB_UPLOAD_WARNING) parser.add_argument("--delete-printer-file-after-print", type=int, default=env_loader.DELETE_PRINTER_FILE_AFTER_PRINT, help="After a successful print, delete the file from the printer's " "own storage if it's also in the bridge's own GCode store") parser.add_argument("--print-start-dialog", dest="print_start_dialog", type=int, default=env_loader.PRINT_START_DIALOG) parser.add_argument("--file-ready-dialog", dest="print_start_dialog", type=int) parser.add_argument("--spoolman-server", default=env_loader.SPOOLMAN_SERVER, help="Spoolman URL (e.g. http://192.168.x.x:7912); leave empty to disable") parser.add_argument("--spoolman-sync-rate", type=int, default=env_loader.SPOOLMAN_SYNC_RATE, help="Mid-print filament sync interval in seconds (0 = only on print end)") parser.add_argument("--kxgauge-url", default=env_loader.KXGAUGE_URL, help="KXGauge display base URL (e.g. http://192.168.x.x); leave empty to disable") parser.add_argument("--kxgauge-enabled", type=int, default=env_loader.KXGAUGE_ENABLED) parser.add_argument("--kxgauge-heat-peak", type=float, default=env_loader.KXGAUGE_HEAT_PEAK, help="Target temperature (°C) the KXGauge heat-ring scale maxes out at") parser.add_argument("--poll-interval", type=int, default=env_loader.POLL_INTERVAL, help="Printer poll interval in seconds") parser.add_argument("--verbose-http-log", type=int, default=env_loader.VERBOSE_HTTP_LOG, help="Log every HTTP request (aiohttp access log)") parser.add_argument("--host", default="0.0.0.0", help="Bind address for the bridge server") parser.add_argument("--port", type=int, default=7125, help="HTTP/WS-Port (Moonraker-Standard: 7125)") parser.add_argument("--data-dir", default=_default_data_dir(), help="Persistence directory for the GCode store and DB") parser.add_argument( "--ui-theme", default=os.environ.get("KX_UI_THEME", "default"), metavar="NAME", help="Web-UI-Theme (Ordner web/themes/NAME/, Standard: default). " "Alternativ: Umgebungsvariable KX_UI_THEME.", ) args = parser.parse_args() if args.printer_ip and ":" in args.printer_ip: args.printer_ip = args.printer_ip.split(":")[0] # Windows needs ProactorEventLoop for asyncio.create_subprocess_exec if sys.platform == "win32": asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy()) asyncio.run(run_bridge(args)) if __name__ == "__main__": main()