""" bridge_mqtt.py - MqttCallbacksMixin for KobraXBridge. The MQTT reader-thread callbacks (_on_temp/_on_print/_on_info/_on_skip/ _on_buried/_on_file) and their helpers (_wait_for_file_action, _delete_printer_file_fire_and_forget, _apply_preprint_skip_after_start). Mixed into KobraXBridge; relies on shared bridge state (self._state, self.client, self._store, self._current_job_*, self._spoolman*, etc.). ──────────────────────────────────────────────────────────────────────────── Copyright (C) 2026 viewit (KX-Bridge contributors) Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md. """ import copy import time import asyncio import logging import threading try: import config_loader as env_loader except ImportError: import env_loader from bridge_constants import KOBRA_TO_KLIPPER_STATE log = logging.getLogger("bridge") class MqttCallbacksMixin: # ------------------------------------------------------------------------- # MQTT callbacks (called from reader thread) # ------------------------------------------------------------------------- def _on_temp(self, payload: dict): d = payload.get("data") or {} self._state["nozzle_temp"] = float(d.get("curr_nozzle_temp", 0)) self._state["nozzle_target"] = float(d.get("target_nozzle_temp", 0)) self._state["bed_temp"] = float(d.get("curr_hotbed_temp", 0)) self._state["bed_target"] = float(d.get("target_hotbed_temp", 0)) if self._kxgauge: self._kxgauge.set_heat_celsius(self._state["nozzle_temp"]) self._push_status_update() def _on_print(self, payload: dict): d = payload.get("data") or {} kobra_state = payload.get("state", "") self._state["print_state"] = KOBRA_TO_KLIPPER_STATE.get(kobra_state, "printing") if kobra_state: self._state["kobra_state"] = kobra_state if self._kxgauge: emotion = self._kxgauge_mapping.get(kobra_state) if emotion: self._kxgauge.set_emotion(emotion) # Automatically switch on the camera at print start (settings option). # Centralized here so it covers all print start paths (OrcaSlicer + UI). # _camera_autostarted verhindert Mehrfach-Trigger pro Druck. if kobra_state == "printing": if (getattr(self._args, "camera_on_print", 0) and not self._camera_autostarted and not self._camera_user_stopped): self._camera_autostarted = True try: self.client.start_camera() log.info("Camera switched on automatically at print start") except Exception as e: log.warning(f"Camera auto-start failed: {e}") elif kobra_state in ("free", "finished", "stoped", "canceled"): self._camera_autostarted = False self._camera_user_stopped = False # release for the next print if kobra_state in ("pause", "paused"): pause_msg = payload.get("msg", "") if pause_msg: error_code = payload.get("code", 0) self._state["error_code"] = error_code self._state["pause_msg"] = pause_msg log.warning(f"Printer paused: [{error_code}] {pause_msg}") elif kobra_state in ("resuming", "resumed", "printing", "finished", "stoped", "canceled"): self._state["error_code"] = 0 self._state["pause_msg"] = "" # Job-History: Druckstart erkennen if kobra_state == "printing" and not self._current_job_id: filename = d.get("filename", self._state.get("filename", "")) if filename: gf = self._store.get_file_by_name(filename) if gf: self._current_job_id = self._store.start_job( gcode_file_id=gf["id"], printer_id=self._printer_id, ) self._current_job_filename = filename log.info(f"Job started: {self._current_job_id} for {filename}") self._spoolman_slot_usage = {} self._spoolman_slot_reported = {} self._spoolman_last_usage = 0.0 self._spoolman_last_sync = 0.0 # Job-History: Druckende erkennen if kobra_state in ("finished",) and self._current_job_id: self._store.finish_job(self._current_job_id, status="completed") log.info(f"Job abgeschlossen: {self._current_job_id}") self._spoolman_notify_end() self._current_job_id = "" # Optional cleanup (Settings -> Print): only for files that are # also backed by the bridge's own GCode store - never for prints # started directly from the printer/Anycubic Slicer, which would # otherwise be deleted with no copy left anywhere (Issue: delete # printer file after successful print). Deliberately only on a # clean "finished" - stoped/canceled prints keep their file. if getattr(self._args, "delete_printer_file_after_print", 0) and self._current_job_filename: self._delete_printer_file_fire_and_forget(self._current_job_filename) self._current_job_filename = "" elif kobra_state in ("stoped", "canceled") and self._current_job_id: self._store.finish_job(self._current_job_id, status="cancelled") log.info(f"Job abgebrochen: {self._current_job_id}") self._spoolman_notify_end() self._current_job_id = "" self._current_job_filename = "" # Terminal states (successful finish AND stop/cancel) must leave the # same clean end state - a "finished" print used to only clear # file_ready (Issue #29), leaving progress/filename/duration/layer # fields stuck at the last job's values until the *next* print # happened to overwrite them (Issue #102). if kobra_state in ("finished", "stoped", "canceled"): self._state["progress"] = 0.0 self._state["filename"] = "" self._state["file_ready"] = "" self._state["print_duration"] = 0 self._state["remain_time"] = 0 self._state["slicer_time"] = 0 self._state["layer_height"] = 0.0 self._state["first_layer_height"] = 0.0 self._state["supplies_usage"] = 0 self._state["curr_layer"] = 0 self._state["total_layers"] = 0 self._thumbnail_b64 = "" else: # Only adopt the payload's filename outside terminal states - the # printer often still reports the just-finished job's filename in # the same "finished"/"stoped"/"canceled" message that triggered # the reset above, which would otherwise immediately undo it. self._state["filename"] = d.get("filename", self._state["filename"]) # Pre-print phases (leveling/preheating/checking) report their own # "progress" - passing it through would make display_status.progress/ # virtual_sdcard.progress jump non-monotonically once real printing # starts and the value resets (Issue #102). if "progress" in d and kobra_state not in ("preheating", "auto_leveling", "checking", "updated", "init"): self._state["progress"] = float(d["progress"]) / 100.0 if "print_time" in d: self._state["print_duration"] = int(d["print_time"]) * 60 if "remain_time" in d: self._state["remain_time"] = int(d["remain_time"]) * 60 if "curr_layer" in d: self._state["curr_layer"] = d["curr_layer"] if "total_layers" in d: self._state["total_layers"] = d["total_layers"] if "taskid" in d: self._state["taskid"] = str(d["taskid"]) if "supplies_usage" in d: self._state["supplies_usage"] = int(d["supplies_usage"]) settings = d.get("settings") or {} if "print_speed_mode" in settings: self._state["print_speed_mode"] = int(settings["print_speed_mode"]) self._push_status_update() def _on_info(self, payload: dict): d = payload.get("data") or {} # Only adopt the MQTT name if no custom name is set (env or per-printer config) if not env_loader.get("BRIDGE_PRINTER_NAME") and not getattr(self, "_name_locked", False): self._state["printer_name"] = d.get("printerName", self._state["printer_name"]) self._state["firmware_version"] = d.get("version", self._state["firmware_version"]) # The real print state lives in info/report inside the nested # project.state ("printing"/"paused"/...). The top-level data.state is only # the device state ("busy"/"free") and would swallow "paused". project = d.get("project") or {} proj_state = project.get("state", "") kobra_state = proj_state or d.get("state", "") if kobra_state: self._state["print_state"] = KOBRA_TO_KLIPPER_STATE.get(kobra_state, "standby") self._state["kobra_state"] = kobra_state # Hide the upload banner after the print ends (Issue #29) - the state also # arrives via info/report (project.state) depending on the printer, not only print/report. # Layer fields must reset here too (Issue #102) - info/report is the # only source for curr_layer/total_layers on some printers, and they # otherwise stay stuck at the last job's values indefinitely. if kobra_state in ("finished", "stoped", "canceled"): self._state["file_ready"] = "" self._state["curr_layer"] = 0 self._state["total_layers"] = 0 # Camera auto-start here as well (OrcaSlicer often reports the start via info/report). # The _camera_autostarted guard prevents a double start with _on_print. if kobra_state == "printing": if (getattr(self._args, "camera_on_print", 0) and not self._camera_autostarted and not self._camera_user_stopped): self._camera_autostarted = True try: self.client.start_camera() log.info("Camera switched on automatically at print start") except Exception as e: log.warning(f"Camera auto-start failed: {e}") elif kobra_state in ("free", "finished", "stoped", "canceled"): self._camera_autostarted = False self._camera_user_stopped = False # release for the next print if project: if "filename" in project: self._state["filename"] = project["filename"] # Same non-monotonic-progress guard as _on_print (Issue #102). if "progress" in project and kobra_state not in ("preheating", "auto_leveling", "checking", "updated", "init"): self._state["progress"] = float(project["progress"]) / 100.0 if "print_time" in project: self._state["print_duration"] = int(project["print_time"]) * 60 if "remain_time" in project: self._state["remain_time"] = int(project["remain_time"]) * 60 if "curr_layer" in project: self._state["curr_layer"] = project["curr_layer"] if "total_layers" in project: self._state["total_layers"] = project["total_layers"] t = d.get("temp") or {} if t: self._state["nozzle_temp"] = float(t.get("curr_nozzle_temp", 0)) self._state["nozzle_target"] = float(t.get("target_nozzle_temp", 0)) self._state["bed_temp"] = float(t.get("curr_hotbed_temp", 0)) self._state["bed_target"] = float(t.get("target_hotbed_temp", 0)) urls = d.get("urls") or {} if urls.get("fileUploadurl"): self._state["upload_url"] = urls["fileUploadurl"] if urls.get("rtspUrl"): self._state["camera_url"] = urls["rtspUrl"] self.camera_cache.set_url(urls["rtspUrl"]) fan = d.get("fan_speed_pct") if fan is not None: self._state["fan_speed"] = int(fan) speed_mode = d.get("print_speed_mode") if speed_mode is not None: self._state["print_speed_mode"] = int(speed_mode) self._push_status_update() def _on_skip(self, payload: dict): """skip/report-Callback (Part-Skip-Feature, v0.9.10). The printer ALWAYS reports the list of already-skipped objects here (objects_skip_parts), whether on query_obj or after skip/start. The full object list comes from file/report. """ d = payload.get("data") or {} skipped = d.get("objects_skip_parts") or d.get("skipped") or d.get("skipped_parts") or [] # While a pre-print skip is still pending, ignore empty early reports # so the UI doesn't snap back before the printer confirms the skip. now = time.time() if (not skipped and self._pending_preprint_skip and now <= self._pending_preprint_skip_deadline): return # During an active print, skip states are effectively monotonic. # Some firmware reports come back empty/partial in between; # those must not remove already-confirmed skip objects from the UI. existing_skipped = [str(n) for n in (self._skip_state.get("skipped") or []) if n] existing_set = set(existing_skipped) incoming_skipped = [str(n) for n in (skipped or []) if n] incoming_set = set(incoming_skipped) active_print = self._state.get("print_state") in ("printing", "paused") if active_print and existing_set: if not incoming_set: skipped = list(existing_skipped) elif not incoming_set.issuperset(existing_set): merged = list(existing_skipped) for n in incoming_skipped: if n not in existing_set: merged.append(n) skipped = merged # Release the pending lock once the printer confirms the requested objects if self._pending_preprint_skip and set(skipped) >= set(self._pending_preprint_skip): self._pending_preprint_skip = [] self._pending_preprint_skip_deadline = 0.0 self._skip_state = { "skipped": list(skipped), "ts": int(time.time()), } if payload.get("state") == "done" or payload.get("code") == 200: log.info(f"Skip response: state={payload.get('state')} code={payload.get('code')} skipped={skipped}") def _delete_printer_file_fire_and_forget(self, filename: str) -> None: """Deletes a file from the printer's own storage without waiting for the response - called from _on_print(), which runs on the MQTT reader thread itself, so blocking here (like _wait_for_file_action does) would deadlock: the file/report reply that would unblock it is dispatched from that same thread. Fire-and-forget is safe because the bridge's own copy in the GCode store is what matters for correctness here; a failed delete just leaves the printer's storage as it is (Settings -> Print -> "Delete file from printer after successful print").""" try: self.client.publish( "file", "deleteBatch", {"root": "local", "files": [{"path": "/", "filename": filename}]}, timeout=0, ) log.info(f"Requested printer-storage delete for {filename} after successful print") except Exception as e: log.warning(f"Delete-after-print request failed for {filename}: {e}") def _wait_for_file_action(self, action: str, send_fn, timeout: float = 8.0) -> dict | None: """Sends a file/* MQTT request (via send_fn, which must call self.client.publish(..., timeout=0) fire-and-forget) and blocks the calling thread until a matching file/report with this `action` arrives via _on_file, or the timeout elapses. Needed because the printer's publish() return value for actions like listLocal/deleteBatch is just a generic immediate ACK skeleton (code=0, empty fields) - the real response is a separate, later file/report message, same as the existing fileDetails pattern. Must be called from a worker thread (e.g. via run_in_executor), not the asyncio event loop, since it blocks on a threading.Event. """ event = threading.Event() waiter = {"event": event, "result": None} self._file_action_waiters[action] = waiter try: send_fn() event.wait(timeout) return waiter["result"] finally: if self._file_action_waiters.get(action) is waiter: del self._file_action_waiters[action] def _on_buried(self, payload: dict): """buried/report - the printer's own analytics event, fired once per print start (verified live against a real Kobra X: fires identically for prints started via Anycubic Slicer Next and via OrcaSlicer/the bridge). Carries gcode_size/estimate_duration/total_layers, which _build_file_metadata() falls back to for files not in our own GCodeStore (Issue #102), plus printer storage usage.""" d = payload.get("data") or {} task_name = d.get("task_name") or "" if not task_name: return self._buried_cache = { "task_name": task_name, "gcode_size": int(d.get("gcode_size") or 0), "estimate_duration": int(d.get("estimate_duration") or 0), "total_layers": int(d.get("total_layers") or 0), } self._state["storage_total_mb"] = int(d.get("storage_total") or 0) self._state["storage_used_mb"] = int(d.get("storage_used") or 0) log.info( f"buried/report: {task_name} size={d.get('gcode_size')} " f"est={d.get('estimate_duration')}s layers={d.get('total_layers')}" ) def _on_file(self, payload: dict): # Deliver to any pending listLocal/deleteBatch waiter first (see # _wait_for_file_action) - these actions carry no file_details/ # thumbnail payload of their own, so this doesn't interfere with the # handling below. action = payload.get("action") or "" waiter = self._file_action_waiters.get(action) if waiter is not None: waiter["result"] = payload waiter["event"].set() d = payload.get("data") or {} details = d.get("file_details") or {} thumb = details.get("thumbnail") or details.get("png_image") or "" file_name = d.get("filename") or details.get("filename") or self._last_uploaded_file active_print = self._state.get("print_state") in ("printing", "paused") current_print_file = self._state.get("filename") or "" # Uploads during a running print must not overwrite the active # progress preview. if thumb and (not active_print or (file_name and file_name == current_print_file)): self._thumbnail_b64 = thumb log.info(f"Thumbnail received: {len(thumb)} base64 chars") # Part-Skip: Objekt-Liste + optionales SVG (v0.9.10) objs = details.get("objects_skip_parts") or [] svg = details.get("svg_image") or "" if objs: filename = file_name if filename: try: self._store.update_file_objects(filename, objs, svg) log.info(f"Skip objects for {filename}: {len(objs)} ({'with SVG' if svg else 'no SVG'})") except Exception as e: log.warning(f"update_file_objects failed: {e}") self._push_status_update() def _apply_preprint_skip_after_start(self, names: list[str], retries: int = 20, delay_s: float = 0.75): """Sends the skip command only after the printer switched to the printing state. Before that, the command goes nowhere (no active print). """ wanted = [str(n) for n in (names or []) if isinstance(n, str) and n] if not wanted: return False for i in range(max(1, int(retries))): try: if self._state.get("print_state") not in ("printing", "paused"): time.sleep(max(0.1, float(delay_s))) continue resp = self.client.skip_objects(wanted) if resp is not None: log.info(f"Pre-Print skip applied ({len(wanted)} objects) on attempt {i+1}/{retries}") self._pending_preprint_skip = [] self._pending_preprint_skip_deadline = 0.0 return True except Exception as e: log.debug(f"Pre-Print skip attempt {i+1}/{retries} failed: {e}") time.sleep(max(0.1, float(delay_s))) log.warning(f"Pre-Print skip could not be confirmed after {retries} attempts") self._pending_preprint_skip = [] self._pending_preprint_skip_deadline = 0.0 return False