viewit
6ab5c31035
All checks were successful
Testing Build / build (push) Successful in 8m44s
Adds an optional per-printer integration with KXGauge (https://gitea.it-drui.de/viewit/kxgauge), a small ESP32 round-face display that shows printer status as an emotion and hotend temperature as a color ring. KXGauge only exposes a GET-only HTTP API with no push/websocket, so the bridge actively pushes to it from the existing MQTT callbacks (_on_temp for the heat ring, _on_print + offline transitions for the emotion) whenever state actually changes, with a built-in dedupe so it doesn't spam the device every poll tick. - kxgauge_client.py: thin synchronous HTTP client (mirrors spoolman_client.py's shape - called from the MQTT reader thread). - New [kxgauge]/[kxgauge_mapping] config.ini sections; the mapping (kobra_state -> KXGauge emotion) is user-editable with a sane default and falls back per-key if only partially configured. - Settings UI: new card under Integrations with enable/URL/target-temp fields, a per-state emotion mapping list, and a connection-test button (/api/kxgauge/test). - Multi-printer aware: kxgauge_url/enabled/heat_peak merge per [printer_N] like the existing power-switch settings. Also fixes a real bug found while testing this: _find_config_path() resolves to the live project config/config.ini, not a sandboxed path, so any test hitting /api/settings POST without stubbing it out will silently overwrite the real printer config. test_settings.py already guards against this - test_kxgauge.py now does too.
1377 lines
67 KiB
Python
1377 lines
67 KiB
Python
"""
|
||
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 <https://www.gnu.org/licenses/gpl-3.0.html> 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
|
||
|
||
# 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)
|
||
|
||
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)
|
||
|
||
|
||
# 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: `<plate>_<material>_<layer>_<dur>.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 <type>** 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/<id>/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.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://<IP of this Docker host>:{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 <exe-dir>/data,
|
||
Dev script uses <repo>/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("--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()
|