From 673bfb0deedc070d86eb8afdb43f98da501bb465 Mon Sep 17 00:00:00 2001 From: Andrew <159703254+andrewsoonqn@users.noreply.github.com> Date: Thu, 16 Jul 2026 16:36:55 +0800 Subject: [PATCH] Track BBLPrinterAgentPlugin.py --- .../orca_plugins/BBLPrinterAgentPlugin.py | 1243 +++++++++++++++++ 1 file changed, 1243 insertions(+) create mode 100644 resources/orca_plugins/BBLPrinterAgentPlugin.py diff --git a/resources/orca_plugins/BBLPrinterAgentPlugin.py b/resources/orca_plugins/BBLPrinterAgentPlugin.py new file mode 100644 index 0000000000..a7edd8747b --- /dev/null +++ b/resources/orca_plugins/BBLPrinterAgentPlugin.py @@ -0,0 +1,1243 @@ +# /// script +# requires-python = ">=3.10" +# dependencies = ["paho-mqtt"] +# +# [tool.orcaslicer.plugin] +# name = "BBL Printer Agent" +# description = "Barebones Bambu Lab printer-agent plugin." +# author = "OrcaSlicer" +# version = "0.0.1" +# type = "printer-connection" +# /// +import json +import ftplib +import queue +import select +import socket +import ssl +import sys +import threading +import uuid +from pathlib import Path + +import orca + +from typing import Any, Callable + +try: + import paho.mqtt.client as mqtt +except ModuleNotFoundError: + mqtt = None + + +def _log(msg): + print(f"[BBLPrinterAgentPlugin] {msg}", file=sys.stderr, flush=True) + + +MQTT_CONNECT_TIMEOUT_SECONDS = 10 +MQTT_KEEPALIVE_SECONDS = 60 +MQTT_PORT = 1883 +MQTT_SSL_PORT = 8883 +FTP_SSL_PORT = 990 +SSDP_GROUP = "239.255.255.250" +SSDP_PORTS = (1990, 2021) +SSDP_RECV_SIZE = 8192 +BAMBU_DEFAULT_USERNAME = "bblp" +MQTT_CONNACK_MESSAGES = { + 0: "accepted", + 1: "unacceptable protocol version", + 2: "identifier rejected", + 3: "server unavailable", + 4: "bad username or password", + 5: "not authorized", +} +CONNECT_STATUS_OK = 0 +CONNECT_STATUS_FAILED = 1 +CONNECT_STATUS_LOST = 2 +BAMBU_NETWORK_ERR_INVALID_HANDLE = -1 +BAMBU_NETWORK_ERR_SEND_MSG_FAILED = -4 +BAMBU_NETWORK_ERR_BIND_FAILED = -5 +BAMBU_NETWORK_ERR_UNBIND_FAILED = -6 +BAMBU_NETWORK_ERR_FILE_NOT_EXIST = -14 +BAMBU_NETWORK_ERR_CANCELED = -18 +BAMBU_NETWORK_ERR_PRINT_WR_UPLOAD_FTP_FAILED = -2130 +BAMBU_NETWORK_ERR_PRINT_LP_UPLOAD_FTP_FAILED = -4020 +BAMBU_NETWORK_ERR_PRINT_LP_PUBLISH_MSG_FAILED = -4030 +BAMBU_NETWORK_ERR_PRINT_SG_UPLOAD_FTP_FAILED = -5010 +PRINTING_STAGE_CREATE = 0 +PRINTING_STAGE_UPLOAD = 1 +PRINTING_STAGE_SENDING = 3 +PRINTING_STAGE_FINISHED = 6 +PRINTING_STAGE_ERROR = 7 +INITIAL_PUSH_PAYLOAD = { + "pushing": {"command": "pushall"}, + "info": {"command": "get_version"}, + "upgrade": {"command": "get_history"}, +} + + +class _ImplicitFTP_TLS(ftplib.FTP_TLS): + def __init__(self, *args, unwrap: bool = False, **kwargs): + super().__init__(*args, **kwargs) + self._sock = None + self.unwrap = unwrap + + def ntransfercmd(self, cmd, rest=None): + conn, size = ftplib.FTP.ntransfercmd(self, cmd, rest) + if self._prot_p: + conn = self.context.wrap_socket(conn, server_hostname=self.host, session=self.sock.session) + return conn, size + + @property + def sock(self): + return self._sock + + @sock.setter + def sock(self, value): + if value is not None and not isinstance(value, ssl.SSLSocket): + value = self.context.wrap_socket(value) + self._sock = value + + def storbinary(self, cmd, fp, blocksize=8192, callback=None, rest=None): + self.voidcmd("TYPE I") + conn = self.transfercmd(cmd, rest) + try: + while True: + buf = fp.read(blocksize) + if not buf: + break + conn.sendall(buf) + if callback: + callback(buf) + if isinstance(conn, ssl.SSLSocket) and self.unwrap: + conn.unwrap() + finally: + conn.close() + + # Bambu's FTPS server may leave the control channel without the usual + # final 226 response after the data socket has been closed. At this + # point transfercmd/sendall has already succeeded, so accept the known + # non-standard completion forms instead of reporting a false upload + # failure to the UI. + old_timeout = self.sock.gettimeout() if self.sock else None + try: + if self.sock: + self.sock.settimeout(2) + return self.voidresp() + except TimeoutError: + _log(f"FTP upload completed for {cmd}; timed out waiting for final response") + return "226 Transfer complete" + except ftplib.error_reply as exc: + if str(exc).startswith("200"): + _log(f"FTP upload completed for {cmd}; accepted non-standard response: {exc}") + return str(exc) + raise + finally: + if self.sock and old_timeout is not None: + self.sock.settimeout(old_timeout) + + +class BBLPrinterAgentPlugin(orca.printer_agent.PrinterAgentBase): + def on_load(self): + self.on_server_err_fn: Callable[..., None] | None = None + self.on_printer_connected_fn: Callable[..., None] | None = None + self.on_subscribe_failure_fn: Callable[..., None] | None = None + self.on_message_fn: Callable[..., None] | None = None + self.on_user_message_fn: Callable[..., None] | None = None + self.on_local_connect_fn: Callable[..., None] | None = None + self.on_local_message_fn: Callable[..., None] | None = None + self.on_ssdp_msg_fn: Callable[..., None] | None = None + self.queue_on_main_fn: Callable[..., None] | None = None + self._selected_machine = "" + self._mqtt_lock = threading.RLock() + self._mqtt_client = None + self._mqtt_connected = threading.Event() + self._mqtt_dev_id = "" + self._mqtt_host = "" + self._mqtt_username = BAMBU_DEFAULT_USERNAME + self._mqtt_password = "" + self._mqtt_command_topic = "" + self._mqtt_manual_disconnect = False + self._printer_data: dict[str, Any] = {} + self._last_error = "" + self._last_code = BAMBU_NETWORK_ERR_INVALID_HANDLE + self._sequence_id = 10000000 + self._mqtt_command_queue = queue.Queue() + self._mqtt_command_worker_stop = threading.Event() + self._mqtt_command_worker = threading.Thread(target=self._mqtt_command_loop, daemon=True) + self._mqtt_command_worker.start() + self._ssdp_stop = threading.Event() + self._ssdp_thread = None + self._ssdp_sockets = [] + self._ssdp_seen = {} + + def get_name(self): + return "BBL Printer Agent" + + def set_server_callback(self, on_server_err_fn): + self.on_server_err_fn = on_server_err_fn + return 0 + + def set_on_printer_connected_fn(self, on_printer_connected_fn): + self.on_printer_connected_fn = on_printer_connected_fn + return 0 + + def set_on_subscribe_failure_fn(self, on_subscribe_failure_fn): + self.on_subscribe_failure_fn = on_subscribe_failure_fn + return 0 + + def set_on_message_fn(self, on_message_fn): + self.on_message_fn = on_message_fn + return 0 + + def set_on_user_message_fn(self, on_user_message_fn): + self.on_user_message_fn = on_user_message_fn + return 0 + + def set_on_local_connect_fn(self, on_connect_fn): + self.on_local_connect_fn = on_connect_fn + return 0 + + def set_on_local_message_fn(self, on_message_fn): + self.on_local_message_fn = on_message_fn + return 0 + + def set_on_ssdp_msg_fn(self, on_ssdp_msg_fn): + self.on_ssdp_msg_fn = on_ssdp_msg_fn + return 0 + + def set_queue_on_main_fn(self, queue_on_main_fn): + self.queue_on_main_fn = queue_on_main_fn + return 0 + + def connect_printer(self, dev_id, dev_ip, username, password, use_ssl) -> int: + self._ensure_state() + dev_id = str(dev_id or "").strip() + dev_ip = str(dev_ip or "").strip() + username = str(username or BAMBU_DEFAULT_USERNAME).strip() + password = str(password or "").strip() + _log( + "connect_printer requested " + f"dev_id={dev_id or ''} host={dev_ip or ''} " + f"username={username or BAMBU_DEFAULT_USERNAME} ssl={bool(use_ssl)} " + f"password_length={len(password)}" + ) + + if not dev_ip: + return self._connection_failed(dev_id, "Missing printer IP address") + if mqtt is None: + return self._connection_failed(dev_id, "paho-mqtt is required for MQTT printer connections") + if not dev_id: + return self._connection_failed(dev_id, "Missing printer serial/device id") + if not password: + return self._connection_failed(dev_id, "Missing printer access code/password") + + port = MQTT_SSL_PORT if use_ssl else MQTT_PORT + connected = threading.Event() + connect_result = {"rc": None} + client = self._create_mqtt_client(dev_id) + command_topic = f"device/{dev_id}/request" + report_topic = f"device/{dev_id}/report" + + def on_connect(client, userdata, flags, rc, *extra): + connect_result["rc"] = self._mqtt_result_code(rc) + try: + _log( + f"MQTT on_connect dev_id={dev_id} rc={connect_result['rc']} " + f"message={MQTT_CONNACK_MESSAGES.get(connect_result['rc'], 'unknown')}" + ) + if connect_result["rc"] == 0: + self._mqtt_connected.set() + client.subscribe(report_topic) + client.publish(command_topic, json.dumps(INITIAL_PUSH_PAYLOAD)) + _log(f"MQTT subscribed dev_id={dev_id} topic={report_topic}; requested initial push") + finally: + connected.set() + + def on_disconnect(client, userdata, *args): + reason = self._disconnect_reason(args) + with self._mqtt_lock: + is_current_client = client is self._mqtt_client + if is_current_client: + self._mqtt_connected.clear() + current_dev_id = self._mqtt_dev_id + manual_disconnect = self._mqtt_manual_disconnect + else: + current_dev_id = "" + manual_disconnect = True + _log(f"MQTT on_disconnect dev_id={current_dev_id or dev_id} reason={reason} manual={manual_disconnect}") + if self.on_local_connect_fn and current_dev_id and not manual_disconnect: + self.on_local_connect_fn(CONNECT_STATUS_LOST, current_dev_id, f"MQTT disconnected: {reason}") + + def on_message(client, userdata, message): + with self._mqtt_lock: + if client is not self._mqtt_client: + return + payload = message.payload.decode("utf-8", errors="replace") + self._handle_message_payload(payload) + if self.on_local_message_fn: + self.on_local_message_fn(dev_id, payload) + + def on_log(client, userdata, level, buf): + text = str(buf) + if any(token in text.lower() for token in ("connect", "ssl", "tls", "fail", "error", "refused")): + _log(f"MQTT paho log dev_id={dev_id} level={level}: {text}") + + client.on_connect = on_connect + client.on_disconnect = on_disconnect + client.on_message = on_message + client.on_log = on_log + + username = username or BAMBU_DEFAULT_USERNAME + client.username_pw_set(username, password) + if use_ssl: + client.tls_set(tls_version=ssl.PROTOCOL_TLS, cert_reqs=ssl.CERT_NONE) + client.tls_insecure_set(True) + + with self._mqtt_lock: + old_client = self._detach_mqtt_locked() + if old_client is not None: + _log("connect_printer replacing existing MQTT client") + self._stop_mqtt_client(old_client) + + with self._mqtt_lock: + self._mqtt_client = client + self._mqtt_dev_id = dev_id + self._mqtt_host = dev_ip + self._mqtt_username = username + self._mqtt_password = password + self._mqtt_command_topic = command_topic + self._mqtt_manual_disconnect = False + self._printer_data = {} + self._last_error = "" + self._selected_machine = dev_id + + try: + _log(f"MQTT connecting dev_id={dev_id} host={dev_ip} port={port}") + client.connect_async(dev_ip, port, keepalive=MQTT_KEEPALIVE_SECONDS) + client.loop_start() + _log(f"MQTT loop started dev_id={dev_id}; waiting up to {MQTT_CONNECT_TIMEOUT_SECONDS}s for CONNACK") + except Exception as exc: + client_to_stop = None + with self._mqtt_lock: + if self._mqtt_client is client: + client_to_stop = self._detach_mqtt_locked() + self._stop_mqtt_client(client_to_stop) + return self._connection_failed(dev_id, str(exc)) + + if not connected.wait(MQTT_CONNECT_TIMEOUT_SECONDS): + client_to_stop = None + with self._mqtt_lock: + if self._mqtt_client is client: + client_to_stop = self._detach_mqtt_locked() + self._stop_mqtt_client(client_to_stop) + return self._connection_failed(dev_id, f"Timed out connecting to MQTT broker at {dev_ip}:{port}") + + with self._mqtt_lock: + is_current_client = client is self._mqtt_client + if not is_current_client: + return self._connection_failed(dev_id, "MQTT client was replaced before connection completed") + + rc = connect_result.get("rc") + if rc == 0: + if self.on_local_connect_fn: + self.on_local_connect_fn(CONNECT_STATUS_OK, dev_id, "0") + if self.on_printer_connected_fn: + self.on_printer_connected_fn(report_topic) + _log(f"connect_printer succeeded dev_id={dev_id} host={dev_ip} port={port}") + return 0 + + reason = MQTT_CONNACK_MESSAGES.get(rc, "unknown") + message = ( + f"MQTT broker rejected connection with code {rc} ({reason}); " + f"check LAN access code, username={username}, ssl={bool(use_ssl)}" + ) + client_to_stop = None + with self._mqtt_lock: + if self._mqtt_client is client: + client_to_stop = self._detach_mqtt_locked() + self._stop_mqtt_client(client_to_stop) + return self._connection_failed(dev_id, message, callback_message=str(rc)) + + def disconnect_printer(self) -> int: + self._ensure_state() + with self._mqtt_lock: + client = self._detach_mqtt_locked() + self._stop_mqtt_client(client) + return 0 + + def start_discovery(self, start=True, sending=False) -> bool: + self._ensure_state() + if start: + if self._ssdp_thread and self._ssdp_thread.is_alive(): + _log("SSDP discovery already running") + return True + self._ssdp_stop.clear() + self._ssdp_seen = {} + _log(f"Starting SSDP discovery sending={bool(sending)} ports={SSDP_PORTS}") + self._ssdp_thread = threading.Thread( + target=self._ssdp_loop, + args=(bool(sending),), + daemon=True, + ) + self._ssdp_thread.start() + return True + + if not self._ssdp_sockets and not (self._ssdp_thread and self._ssdp_thread.is_alive()): + return True + + _log("Stopping SSDP discovery") + self._ssdp_stop.set() + for sock in list(self._ssdp_sockets): + try: + sock.close() + except OSError: + pass + self._ssdp_sockets = [] + return True + + def install_device_cert(self, dev_id, lan_only=True): + return + + def get_agent_info(self): + return orca.printer_agent.AgentInfo( + "pybbl", + "Bambu Lab", + "0.0.1", + "Barebones Bambu Lab printer-agent plugin", + ) + + def get_user_selected_machine(self): + self._ensure_state() + return self._selected_machine + + def set_user_selected_machine(self, dev_id): + self._ensure_state() + self._selected_machine = str(dev_id or "") + return 0 + + def send_message(self, dev_id, json_str, qos=0, flag=0): + return self.send_message_to_printer(dev_id, json_str, qos, flag) + + def send_message_to_printer(self, dev_id, json_str, qos=0, flag=0): + self._ensure_state() + del dev_id, flag + if self._publish_payload(json_str, qos=max(0, min(2, self._int_or_default(qos, 0)))): + return 0 + self._last_code = BAMBU_NETWORK_ERR_SEND_MSG_FAILED + return self._last_code + + def check_cert(self): + return 0 + + def ping_bind(self, ping_code): + self._last_error = "Printer binding is not supported by this plugin" + self._last_code = BAMBU_NETWORK_ERR_BIND_FAILED + return self._last_code + + def bind_detect(self, dev_ip, sec_link, detect): + detect.result_msg = "Printer binding is not supported by this plugin" + detect.command = "bind_detect" + detect.dev_id = "" + detect.model_id = "" + detect.dev_name = "" + detect.version = "" + detect.bind_state = "unsupported" + detect.connect_type = "lan" + self._last_error = detect.result_msg + self._last_code = BAMBU_NETWORK_ERR_BIND_FAILED + return self._last_code + + def bind(self, dev_ip, dev_id, sec_link, timezone, improved, update_fn): + del dev_ip, dev_id, sec_link, timezone, improved + self._last_error = "Printer binding is not supported by this plugin" + self._last_code = BAMBU_NETWORK_ERR_BIND_FAILED + self._update_progress(update_fn, PRINTING_STAGE_ERROR, self._last_code, self._last_error) + return self._last_code + + def unbind(self, dev_id): + del dev_id + self._last_error = "Printer unbinding is not supported by this plugin" + self._last_code = BAMBU_NETWORK_ERR_UNBIND_FAILED + return self._last_code + + def request_bind_ticket(self): + return (BAMBU_NETWORK_ERR_BIND_FAILED, "") + + def start_print(self, params=None, update_fn=None, cancel_fn=None, wait_fn=None): + del wait_fn + return self._start_local_print(params, BAMBU_NETWORK_ERR_PRINT_WR_UPLOAD_FTP_FAILED, update_fn, cancel_fn) + + def start_local_print(self, params=None, update_fn=None, cancel_fn=None): + return self._start_local_print(params, BAMBU_NETWORK_ERR_PRINT_LP_UPLOAD_FTP_FAILED, update_fn, cancel_fn) + + def start_local_print_with_record(self, params=None, update_fn=None, cancel_fn=None, wait_fn=None): + del wait_fn + return self._start_local_print(params, BAMBU_NETWORK_ERR_PRINT_WR_UPLOAD_FTP_FAILED, update_fn, cancel_fn) + + def start_sdcard_print(self, params=None, update_fn=None, cancel_fn=None): + self._ensure_state() + del update_fn + if self._cancel_requested(cancel_fn): + self._last_error = "Cancelled" + self._last_code = BAMBU_NETWORK_ERR_CANCELED + return self._last_code + if params is None: + self._last_error = "Missing print params" + self._last_code = BAMBU_NETWORK_ERR_INVALID_HANDLE + return self._last_code + try: + payload = self._build_sdcard_project_file_payload(params) + except Exception as exc: + self._last_error = str(exc) + self._last_code = BAMBU_NETWORK_ERR_INVALID_HANDLE + return self._last_code + if self._queue_payload(payload): + return 0 + self._last_code = BAMBU_NETWORK_ERR_PRINT_LP_PUBLISH_MSG_FAILED + return self._last_code + + def start_send_gcode_to_sdcard(self, params=None, update_fn=None, cancel_fn=None, wait_fn=None): + self._ensure_state() + del wait_fn + if params is None: + self._last_error = "Missing print params" + self._last_code = BAMBU_NETWORK_ERR_INVALID_HANDLE + return self._last_code + if self._upload_print_file(params, update_fn, cancel_fn): + return 0 + return self._last_code or BAMBU_NETWORK_ERR_PRINT_SG_UPLOAD_FTP_FAILED + + def get_filament_sync_mode(self): + return orca.printer_agent.FilamentSyncMode.Subscription + + def fetch_filament_info(self, dev_id): + del dev_id + return self._publish_payload({"pushing": {"command": "pushall"}}, require_connected=False) + + def send_gcode(self, gcode, sequence_id=None): + self._ensure_state() + gcode = str(gcode or "").strip() + if not gcode: + self._last_error = "Missing G-code command" + self._last_code = BAMBU_NETWORK_ERR_INVALID_HANDLE + return self._last_code + payload = { + "print": { + "command": "gcode_line", + "param": gcode, + "sequence_id": str(sequence_id or self._next_sequence_id()), + } + } + if self._queue_payload(payload): + return 0 + self._last_code = BAMBU_NETWORK_ERR_PRINT_LP_PUBLISH_MSG_FAILED + return self._last_code + + def send_command(self, request_json): + return self._send_command_impl(request_json, None, None) + + def send_command_with_progress(self, request_json, update_fn, cancel_fn): + return self._send_command_impl(request_json, update_fn, cancel_fn) + + def _send_command_impl(self, request_json, update_fn=None, cancel_fn=None): + try: + request = json.loads(request_json) + except json.JSONDecodeError: + return json.dumps({ + "status": "error", + "message": "Invalid printer-agent request", + }) + + command = request.get("command") + payload = request.get("payload") or {} + if not isinstance(payload, dict): + return json.dumps({ + "status": "error", + "message": "Invalid printer-agent payload", + }) + + if command == "connect_printer": + dev_id = request.get("dev_id") or payload.get("dev_id") or "" + result = self.connect_printer( + dev_id, + payload.get("dev_ip") or "", + payload.get("username") or "", + payload.get("password") or "", + self._as_bool(payload.get("use_ssl")), + ) + if result == 0: + return json.dumps({"status": "ok"}) + return json.dumps({"status": "error", "message": self._last_error or "MQTT connection failed"}) + + if command == "disconnect_printer": + self.disconnect_printer() + return json.dumps({"status": "ok"}) + + if command == "get_filament_info": + self._queue_payload({"pushing": {"command": "pushall"}}, require_connected=False) + return json.dumps(self._build_filament_info()) + + if command in ("send_message", "send_message_to_printer"): + outbound = payload.get("message") if "message" in payload else payload + qos = self._int_or_default(payload.get("qos"), 0) + qos = max(0, min(2, qos)) + return self._status_response(self._queue_payload(outbound, qos=qos)) + + if command == "start_local_print_with_record": + return self._status_response( + self._start_local_print(payload, BAMBU_NETWORK_ERR_PRINT_WR_UPLOAD_FTP_FAILED, update_fn, cancel_fn) + ) + + if command == "start_local_print": + return self._status_response( + self._start_local_print(payload, BAMBU_NETWORK_ERR_PRINT_LP_UPLOAD_FTP_FAILED, update_fn, cancel_fn) + ) + + if command == "start_send_gcode_to_sdcard": + result = 0 if self._upload_print_file(payload, update_fn, cancel_fn) else self._last_code + if result is None: + result = BAMBU_NETWORK_ERR_PRINT_SG_UPLOAD_FTP_FAILED + return self._status_response(result) + + if command == "start_sdcard_print": + return self._status_response(self.start_sdcard_print(payload)) + + if command == "send_gcode": + gcode = payload.get("gcode") or payload.get("param") or payload.get("message") + return self._status_response(self.send_gcode(gcode, payload.get("sequence_id"))) + + if self._looks_like_bambu_payload(request): + return self._status_response(self._queue_payload(request, qos=0)) + + return json.dumps({ + "status": "error", + "message": f"Unsupported printer-agent command: {command or ''}", + }) + + def on_unload(self): + self.start_discovery(False, False) + self._mqtt_command_worker_stop.set() + self.disconnect_printer() + if self._mqtt_command_worker and self._mqtt_command_worker.is_alive(): + self._mqtt_command_worker.join(timeout=1.0) + self.on_server_err_fn: Callable[..., None] | None = None + self.on_printer_connected_fn: Callable[..., None] | None = None + self.on_subscribe_failure_fn: Callable[..., None] | None = None + self.on_message_fn: Callable[..., None] | None = None + self.on_user_message_fn: Callable[..., None] | None = None + self.on_local_connect_fn: Callable[..., None] | None = None + self.on_local_message_fn: Callable[..., None] | None = None + self.on_ssdp_msg_fn: Callable[..., None] | None = None + self.queue_on_main_fn: Callable[..., None] | None = None + + def _ensure_state(self): + if hasattr(self, "_mqtt_lock"): + return + self.on_load() + + def _create_mqtt_client(self, dev_id): + client_id = f"OrcaSlicer-{uuid.uuid4().hex[:12]}" + _log(f"MQTT creating client dev_id={dev_id or ''} client_id={client_id}") + kwargs = { + "client_id": client_id, + "clean_session": True, + "protocol": mqtt.MQTTv311, + } + callback_api_version = getattr(mqtt, "CallbackAPIVersion", None) + if callback_api_version is not None: + kwargs["callback_api_version"] = callback_api_version.VERSION1 + return mqtt.Client(**kwargs) + + def _detach_mqtt_locked(self): + client = self._mqtt_client + self._mqtt_manual_disconnect = True + self._mqtt_client = None + self._mqtt_dev_id = "" + self._mqtt_host = "" + self._mqtt_command_topic = "" + self._mqtt_connected.clear() + return client + + def _stop_mqtt_client(self, client): + if client is None: + return + try: + client.disconnect() + except Exception: + pass + try: + client.loop_stop() + except Exception: + pass + + def _stop_mqtt_client_async(self, client): + if client is None: + return + threading.Thread(target=self._stop_mqtt_client, args=(client,), daemon=True).start() + + def _ssdp_loop(self, sending=False): + sockets = [] + try: + for port in SSDP_PORTS: + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + if hasattr(socket, "SO_REUSEPORT"): + try: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) + except OSError: + pass + sock.bind(("", port)) + membership = socket.inet_aton(SSDP_GROUP) + socket.inet_aton("0.0.0.0") + sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, membership) + sockets.append(sock) + _log(f"SSDP listening on {SSDP_GROUP}:{port}") + except OSError as exc: + self._last_error = f"SSDP listen failed on port {port}: {exc}" + _log(self._last_error) + + self._ssdp_sockets = sockets + if not sockets: + _log("SSDP discovery has no listening sockets") + return + + if sending: + _log("Sending SSDP M-SEARCH") + for port in SSDP_PORTS: + request = ( + "M-SEARCH * HTTP/1.1\r\n" + f"HOST: {SSDP_GROUP}:{port}\r\n" + 'MAN: "ssdp:discover"\r\n' + "MX: 1\r\n" + "ST: ssdp:all\r\n" + "\r\n" + ).encode("utf-8") + for sock in sockets: + try: + sock.sendto(request, (SSDP_GROUP, port)) + except OSError as exc: + _log(f"SSDP M-SEARCH send failed port={port}: {exc}") + + while not self._ssdp_stop.is_set(): + try: + readable, _, _ = select.select(sockets, [], [], 1.0) + except OSError as exc: + _log(f"SSDP select failed: {exc}") + break + + for sock in readable: + try: + packet, addr = sock.recvfrom(SSDP_RECV_SIZE) + except OSError as exc: + _log(f"SSDP recv failed: {exc}") + continue + + text = packet.decode("utf-8", errors="replace") + headers = {} + for line in text.replace("\r\n", "\n").split("\n"): + if ":" not in line: + continue + key, value = line.split(":", 1) + key = "".join(ch for ch in key.lower() if ch.isalnum()) + value = value.strip() + headers[key] = value + for suffix in ("bambucom", "bambulabcom"): + if key.endswith(suffix): + headers.setdefault(key[:-len(suffix)], value) + + def header(*names, default=""): + for name in names: + key = "".join(ch for ch in name.lower() if ch.isalnum()) + value = headers.get(key) + if value: + return value + return default + + dev_id = header("dev_id", "devid", "serial", "serial_number", "usn") + if dev_id.startswith("uuid:"): + dev_id = dev_id[5:] + if "::" in dev_id: + dev_id = dev_id.split("::", 1)[0] + if not dev_id: + continue + + dev_name = header("dev_name", "devname", "friendly_name", "name", default=dev_id) + device = { + "dev_name": dev_name, + "dev_id": dev_id, + "dev_ip": addr[0], + "dev_type": header("dev_type", "dev_model", "devmodel", "model", "printer_type", default=""), + "dev_signal": header("dev_signal", "devsignal", "signal", default="0"), + "connect_type": header("connect_type", "connection_type", "dev_connect", "devconnect", default="lan"), + "bind_state": header("bind_state", "dev_bind", "devbind", "bind", default="free"), + "sec_link": header("sec_link", "dev_sec_link", "devseclink", "secure_link", default=""), + "ssdp_version": header("ssdp_version", "dev_version", "devversion", "version", default=""), + "connection_name": header("connection_name", default=dev_name), + } + seen_key = device["dev_id"] + seen_value = ( + device["dev_ip"], + device["dev_name"], + device["dev_type"], + device["connect_type"], + device["bind_state"], + device["sec_link"], + device["ssdp_version"], + ) + if self._ssdp_seen.get(seen_key) != seen_value: + self._ssdp_seen[seen_key] = seen_value + _log(f"SSDP discovered printer {json.dumps(device, sort_keys=True)}") + if self.on_ssdp_msg_fn: + self.on_ssdp_msg_fn(json.dumps(device)) + finally: + for sock in sockets: + try: + sock.close() + except OSError: + pass + self._ssdp_sockets = [] + _log("SSDP discovery loop exited") + + def _connection_failed(self, dev_id, message, callback_message=None) -> int: + self._last_error = message + _log(f"connect_printer failed dev_id={dev_id or ''}: {message}") + if self.on_local_connect_fn: + self.on_local_connect_fn(CONNECT_STATUS_FAILED, dev_id, callback_message or message) + return -1 + + def _queue_payload(self, payload, qos=1, require_connected=True) -> bool: + try: + self._mqtt_command_queue.put_nowait((payload, qos, require_connected)) + return True + except Exception as exc: + self._last_error = str(exc) + return False + + def _mqtt_command_loop(self): + while not self._mqtt_command_worker_stop.is_set(): + try: + payload, qos, require_connected = self._mqtt_command_queue.get(timeout=0.25) + except queue.Empty: + continue + + try: + if not self._publish_payload(payload, qos=qos, require_connected=require_connected): + _log(f"MQTT queued command publish failed: {self._last_error}") + finally: + self._mqtt_command_queue.task_done() + + def _publish_payload(self, payload, qos=1, require_connected=True) -> bool: + if isinstance(payload, str): + try: + payload = json.loads(payload) + except json.JSONDecodeError: + pass + + with self._mqtt_lock: + client = self._mqtt_client + topic = self._mqtt_command_topic + connected = self._mqtt_connected.is_set() + + if client is None or not topic: + self._last_error = "MQTT client is not connected" + return False + if require_connected and not connected: + self._last_error = "MQTT client is not connected" + return False + + try: + message = payload if isinstance(payload, str) else json.dumps(payload) + info = client.publish(topic, message, qos=qos) + if getattr(info, "rc", 0) != 0: + self._last_error = f"MQTT publish failed with code {info.rc}" + return False + return True + except Exception as exc: + self._last_error = str(exc) + return False + + def _handle_message_payload(self, payload: str): + try: + doc = json.loads(payload) + except json.JSONDecodeError: + return + if not isinstance(doc, dict): + return + with self._mqtt_lock: + self._merge_printer_data(doc) + + def _merge_printer_data(self, doc: dict[str, Any]): + for key, value in doc.items(): + if isinstance(value, dict) and isinstance(self._printer_data.get(key), dict): + self._printer_data[key].update(value) + else: + self._printer_data[key] = value + + def _build_filament_info(self) -> dict[str, Any]: + with self._mqtt_lock: + data = json.loads(json.dumps(self._printer_data)) + + print_data = data.get("print", {}) + if not isinstance(print_data, dict): + return {"units": [], "external": []} + + return { + "units": self._build_ams_units(print_data.get("ams", {})), + "external": self._build_external_filaments(print_data.get("vt_tray")), + } + + def _build_ams_units(self, ams_data) -> list[dict[str, Any]]: + if not isinstance(ams_data, dict): + return [] + units = [] + for unit_index, unit in enumerate(ams_data.get("ams", []) or []): + if not isinstance(unit, dict): + continue + unit_id = str(unit.get("id", unit_index)) + slots = [] + for slot_index, tray in enumerate(unit.get("tray", []) or []): + if not isinstance(tray, dict): + continue + slots.append(self._convert_filament_tray(tray, slot_index)) + units.append({ + "id": unit_id, + "type": self._ams_type_name(unit), + "extruder": self._int_or_default(unit.get("ext_id"), 0), + "temperature": self._float_or_default(unit.get("temp"), 0.0), + "humidity_percent": self._int_or_default(unit.get("humidity_raw"), -1), + "dry_time_min": self._int_or_default(unit.get("dry_time"), 0), + "slots": slots, + }) + return units + + def _build_external_filaments(self, vt_tray) -> list[dict[str, Any]]: + if not isinstance(vt_tray, dict): + return [] + return [self._convert_filament_tray(vt_tray, 0, extruder=0)] + + def _convert_filament_tray(self, tray, index, extruder=None) -> dict[str, Any]: + color = self._normalize_color(tray.get("tray_color") or tray.get("color") or "00000000") + material = str(tray.get("tray_type") or tray.get("type") or "") + preset_id = str(tray.get("tray_info_idx") or tray.get("preset_id") or "") + converted = { + "index": self._int_or_default(tray.get("id"), index), + "loaded": bool(material or preset_id or tray.get("n")), + "material": material, + "preset_id": preset_id, + "color": color, + "nozzle_temp_min": self._int_or_default(tray.get("nozzle_temp_min"), 0), + "nozzle_temp_max": self._int_or_default(tray.get("nozzle_temp_max"), 0), + "remain_percent": self._int_or_default(tray.get("remain"), -1), + "k": self._float_or_default(tray.get("k"), 0.0), + } + if extruder is not None: + converted["extruder"] = extruder + diameter = self._float_or_none(tray.get("tray_diameter") or tray.get("diameter")) + if diameter is not None: + converted["diameter_mm"] = diameter + weight = self._int_or_none(tray.get("tray_weight") or tray.get("weight")) + if weight is not None: + converted["weight_g"] = weight + return converted + + def _start_local_print(self, params, upload_error_code, update_fn=None, cancel_fn=None): + self._ensure_state() + if params is None: + self._last_error = "Missing print params" + self._last_code = BAMBU_NETWORK_ERR_INVALID_HANDLE + return self._last_code + if self._cancel_requested(cancel_fn): + self._last_error = "Cancelled" + self._last_code = BAMBU_NETWORK_ERR_CANCELED + return self._last_code + + self._update_progress(update_fn, PRINTING_STAGE_CREATE, 0, "Preparing...") + if not self._upload_print_file(params, update_fn, cancel_fn, upload_error_code): + self._update_progress(update_fn, PRINTING_STAGE_ERROR, self._last_code, self._last_error) + return self._last_code + if self._cancel_requested(cancel_fn): + self._last_error = "Cancelled" + self._last_code = BAMBU_NETWORK_ERR_CANCELED + return self._last_code + + try: + payload = self._build_project_file_payload(params) + except Exception as exc: + self._last_error = str(exc) + self._last_code = BAMBU_NETWORK_ERR_INVALID_HANDLE + self._update_progress(update_fn, PRINTING_STAGE_ERROR, self._last_code, self._last_error) + return self._last_code + + self._update_progress(update_fn, PRINTING_STAGE_SENDING, 0, "Starting print...") + if not self._queue_payload(payload): + self._last_code = BAMBU_NETWORK_ERR_PRINT_LP_PUBLISH_MSG_FAILED + self._update_progress(update_fn, PRINTING_STAGE_ERROR, self._last_code, self._last_error) + return self._last_code + + self._update_progress(update_fn, PRINTING_STAGE_FINISHED, 100, "1") + return 0 + + def _build_project_file_payload(self, params) -> dict[str, Any]: + filename = self._remote_print_name(params) + if not filename: + raise ValueError("Missing print filename") + + plate = self._param(params, "plate_index", 1) + try: + plate = int(plate) + except (TypeError, ValueError): + plate = 1 + + ams_mapping = self._json_or_default(self._param(params, "ams_mapping"), [0]) + if not isinstance(ams_mapping, list): + ams_mapping = [ams_mapping] + + return { + "print": { + "command": "project_file", + "param": self._param(params, "plate_location") or f"Metadata/plate_{plate}.gcode", + "file": filename, + "bed_leveling": self._as_bool(self._param(params, "task_bed_leveling", True)), + "bed_type": self._param(params, "task_bed_type", "textured_plate"), + "flow_cali": self._as_bool(self._param(params, "task_flow_cali", True)), + "vibration_cali": self._as_bool(self._param(params, "task_vibration_cali", True)), + "url": self._param(params, "url") or f"ftp:///{filename}", + "layer_inspect": self._as_bool(self._param(params, "task_layer_inspect", False)), + "sequence_id": str(self._param(params, "sequence_id") or self._next_sequence_id()), + "timelapse": self._as_bool(self._param(params, "task_record_timelapse", False)), + "use_ams": self._as_bool(self._param(params, "task_use_ams", True)), + "ams_mapping": ams_mapping, + "skip_objects": self._json_or_default(self._param(params, "skip_objects"), None), + } + } + + def _build_sdcard_project_file_payload(self, params) -> dict[str, Any]: + remote = ( + self._param(params, "dst_file") + or self._param(params, "ftp_file") + or self._param(params, "file") + or self._param(params, "filename") + ) + if not remote: + raise ValueError("Missing printer storage file") + remote = str(remote) + if remote.startswith("file://"): + file_url = remote + elif remote.startswith("/"): + file_url = f"file://{remote}" + else: + file_url = f"file:///{remote}" + payload_params = dict(params) if isinstance(params, dict) else {} + payload_params["ftp_file"] = Path(remote).name + payload_params["url"] = self._param(params, "url") or file_url + return self._build_project_file_payload(payload_params) + + def _upload_print_file(self, params, update_fn=None, cancel_fn=None, error_code=BAMBU_NETWORK_ERR_PRINT_SG_UPLOAD_FTP_FAILED) -> bool: + source = ( + self._param(params, "local_path") + or self._param(params, "source") + or self._param(params, "filename") + or self._param(params, "dst_file") + ) + if not source: + self._last_error = "Missing local file path" + self._last_code = BAMBU_NETWORK_ERR_FILE_NOT_EXIST + return False + path = Path(source) + if not path.is_file(): + self._last_error = f"Local file does not exist: {source}" + self._last_code = BAMBU_NETWORK_ERR_FILE_NOT_EXIST + return False + + remote = self._remote_print_name(params, path) + host = self._param(params, "dev_ip") or self._mqtt_host + username = self._param(params, "username") or self._mqtt_username or BAMBU_DEFAULT_USERNAME + password = self._param(params, "password") or self._mqtt_password + use_ssl = self._as_bool(self._param(params, "use_ssl_for_ftp", True)) + port = FTP_SSL_PORT if use_ssl else 21 + if not host or not password: + self._last_error = "Missing FTP host or password" + self._last_code = error_code + return False + + try: + self._update_progress(update_fn, PRINTING_STAGE_UPLOAD, 0, "Uploading...") + ftp = _ImplicitFTP_TLS() if use_ssl else ftplib.FTP() + _log( + "FTP upload starting " + f"host={host} port={port} ssl={use_ssl} source={path} remote={remote} " + f"size={path.stat().st_size}" + ) + ftp.connect(host=host, port=port, timeout=30) + ftp.login(username, password) + if use_ssl: + ftp.prot_p() + total = path.stat().st_size + uploaded = 0 + + def on_upload(block): + nonlocal uploaded + if self._cancel_requested(cancel_fn): + raise InterruptedError("Cancelled") + uploaded += len(block) + if total > 0: + self._update_progress(update_fn, PRINTING_STAGE_UPLOAD, min(100, int(uploaded * 100 / total)), "Uploading...") + + with path.open("rb") as fh: + ftp.storbinary(f"STOR {remote}", fh, blocksize=32768, callback=on_upload) + ftp.close() + self._update_progress(update_fn, PRINTING_STAGE_UPLOAD, 100, "Uploading...") + _log(f"FTP upload succeeded remote={remote} bytes={uploaded}") + return True + except InterruptedError as exc: + self._last_error = str(exc) + self._last_code = BAMBU_NETWORK_ERR_CANCELED + _log(f"FTP upload cancelled remote={remote}: {exc}") + return False + except Exception as exc: + self._last_error = str(exc) + self._last_code = error_code + _log(f"FTP upload failed remote={remote} code={error_code}: {type(exc).__name__}: {exc}") + return False + + def _remote_print_name(self, params, source_path: Path | None = None) -> str: + remote = self._param(params, "ftp_file") or self._param(params, "file") + if remote: + return Path(str(remote)).name + candidate = self._param(params, "filename") or self._param(params, "dst_file") or self._param(params, "source") or self._param(params, "local_path") + if candidate: + return Path(str(candidate)).name + project_name = self._param(params, "project_name") + if project_name: + return Path(str(project_name)).name + return source_path.name if source_path else "" + + def _status_response(self, result): + if isinstance(result, bool): + result = 0 if result else (self._last_code or BAMBU_NETWORK_ERR_INVALID_HANDLE) + if result == 0: + return json.dumps({"status": "ok"}) + status = "cancelled" if result == BAMBU_NETWORK_ERR_CANCELED else "error" + return json.dumps({ + "status": status, + "code": result, + "message": self._last_error or "operation failed", + }) + + @staticmethod + def _looks_like_bambu_payload(request) -> bool: + return isinstance(request, dict) and any(key in request for key in ("print", "system", "pushing", "info", "upgrade", "xcam", "camera")) + + @staticmethod + def _disconnect_reason(args) -> int: + if not args: + return 0 + if len(args) >= 2: + return BBLPrinterAgentPlugin._mqtt_result_code(args[1]) + return BBLPrinterAgentPlugin._mqtt_result_code(args[0]) + + @staticmethod + def _ams_type_name(unit) -> str: + ams_type = str(unit.get("ams_type", unit.get("type", "ams"))).lower() + if ams_type in {"2", "ams_lite", "ams-lite"}: + return "ams_lite" + if ams_type in {"3", "n3f"}: + return "n3f" + if ams_type in {"4", "n3s"}: + return "n3s" + return "ams" + + @staticmethod + def _normalize_color(value) -> str: + color = str(value or "").strip().lstrip("#") + if not color: + return "00000000" + if len(color) == 6: + color += "FF" + return color.upper() + + @staticmethod + def _param(params, key, default=None): + if isinstance(params, dict): + return params.get(key, default) + return getattr(params, key, default) + + def _next_sequence_id(self): + with self._mqtt_lock: + self._sequence_id += 1 + return self._sequence_id + + @staticmethod + def _update_progress(update_fn, stage, code, info): + if not update_fn: + return + try: + update_fn(stage, code, info) + except Exception as exc: + _log(f"progress callback failed: {exc}") + + @staticmethod + def _cancel_requested(cancel_fn) -> bool: + if not cancel_fn: + return False + try: + return bool(cancel_fn()) + except Exception as exc: + _log(f"cancel callback failed: {exc}") + return False + + @staticmethod + def _json_or_default(value, default): + if value in (None, ""): + return default + if isinstance(value, str): + try: + return json.loads(value) + except json.JSONDecodeError: + return default + return value + + @staticmethod + def _int_or_default(value, default): + result = BBLPrinterAgentPlugin._int_or_none(value) + return default if result is None else result + + @staticmethod + def _int_or_none(value): + try: + return int(value) + except (TypeError, ValueError): + return None + + @staticmethod + def _float_or_default(value, default): + result = BBLPrinterAgentPlugin._float_or_none(value) + return default if result is None else result + + @staticmethod + def _float_or_none(value): + try: + return float(value) + except (TypeError, ValueError): + return None + + @staticmethod + def _mqtt_result_code(rc) -> int: + try: + return int(rc) + except (TypeError, ValueError): + value = getattr(rc, "value", None) + return int(value) if value is not None else -1 + + @staticmethod + def _as_bool(value) -> bool: + if isinstance(value, str): + return value.strip().lower() in {"1", "true", "yes", "on"} + return bool(value) + + +@orca.plugin +class BBLPrinterAgentPackage(orca.base): + def register_capabilities(self): + orca.register_capability(BBLPrinterAgentPlugin)