__author__ = "Gina Häußge " __license__ = "GNU Affero General Public License http://www.gnu.org/licenses/agpl.html" import collections import datetime import os import queue import re import threading import time from typing import Any, Dict, List, Optional import asyncio from pybambu import BambuClient, commands from serial import SerialTimeoutException from octoprint.util import RepeatedTimer, to_bytes, to_unicode, get_dos_filename from octoprint.util.files import unix_timestamp_to_m20_timestamp from .ftpsclient import IoTFTPSClient # noinspection PyBroadException class BambuPrinter: command_regex = re.compile(r"^([GM])(\d+)") def __init__( self, settings, printer_profile_manager, data_folder, seriallog_handler=None, read_timeout=5.0, write_timeout=10.0, faked_baudrate=115200, ): self._busyInterval = 2.0 self.tick_rate = 2.0 self._errors = { "checksum_mismatch": "Checksum mismatch", "checksum_missing": "Missing checksum", "lineno_mismatch": "expected line {} got {}", "lineno_missing": "No Line Number with checksum, Last Line: {}", "maxtemp": "MAXTEMP triggered!", "mintemp": "MINTEMP triggered!", "command_unknown": "Unknown command {}", } self._sendBusy = False self._ambient_temperature = 21.3 self.temp = [self._ambient_temperature] self.targetTemp = [0.0] self.bedTemp = self._ambient_temperature self.bedTargetTemp = 0.0 self._hasChamber = printer_profile_manager.get_current().get("heatedChamber") self.chamberTemp = self._ambient_temperature self.chamberTargetTemp = 0.0 self.lastTempAt = time.monotonic() self._firmwareName = "Bambu" self._m115FormatString = "FIRMWARE_NAME:{firmware_name} PROTOCOL_VERSION:1.0" self._received_lines = 0 self.extruderCount = 1 self._waitInterval = 5.0 self._killed = False self._heatingUp = False self.current_line = 0 self._writingToSd = False self._sdCardReady = True self._sdPrinter = None self._sdPrinting = False self._sdPrintStarting = False self._sdPrintingSemaphore = threading.Event() self._sdPrintingPausedSemaphore = threading.Event() self._sdFileListCache = {} self._selectedSdFile = None self._selectedSdFileSize = 0 self._selectedSdFilePos = 0 self._busy = None self._busy_loop = None import logging self._logger = logging.getLogger( "octoprint.plugins.bambu_printer.BambuPrinter" ) self._settings = settings self._printer_profile_manager = printer_profile_manager self._faked_baudrate = faked_baudrate self._plugin_data_folder = data_folder self._seriallog = logging.getLogger( "octoprint.plugins.bambu_printer.BambuPrinter.serial" ) self._seriallog.setLevel(logging.CRITICAL) self._seriallog.propagate = False if seriallog_handler is not None: import logging.handlers self._seriallog.addHandler(seriallog_handler) self._seriallog.setLevel(logging.INFO) self._seriallog.debug("-" * 78) self._read_timeout = read_timeout self._write_timeout = write_timeout self._rx_buffer_size = 64 self._incoming_lock = threading.RLock() self.incoming = CharCountingQueue(self._rx_buffer_size, name="RxBuffer") self.outgoing = queue.Queue() self.buffered = queue.Queue(maxsize=4) self._last_hms_errors = None self.bambu = None readThread = threading.Thread( target=self._processIncoming, name="octoprint.plugins.bambu_printer.wait_thread", daemon=True ) readThread.start() # bufferThread = threading.Thread( # target=self._processBuffer, # name="octoprint.plugins.bambu_printer.buffer_thread", # daemon=True # ) # bufferThread.start() # Move this into M110 command response? connectionThread = threading.Thread( target=self._create_connection, name="octoprint.plugins.bambu_printer.connection_thread", daemon=True ) connectionThread.start() def new_update(self, event_type): if event_type == "event_hms_errors": bambu_printer = self.bambu.get_device() if bambu_printer.hms.errors != self._last_hms_errors and bambu_printer.hms.errors["Count"] > 0: self._logger.debug(f"HMS Error: {bambu_printer.hms.errors}") for n in range(1, bambu_printer.hms.errors["Count"]+1): error = bambu_printer.hms.errors[f"{n}-Error"].strip() self._send(f"// action:notification {error}") self._last_hms_errors = bambu_printer.hms.errors elif event_type == "event_printer_data_update": device_data = self.bambu.get_device() ams = device_data.ams.__dict__ print_job = device_data.print_job.__dict__ temperatures = device_data.temperature.__dict__ lights = device_data.lights.__dict__ fans = device_data.fans.__dict__ speed = device_data.speed.__dict__ # self._logger.debug(device_data) self.temp[0] = temperatures.get("nozzle_temp", 0.0) self.targetTemp[0] = temperatures.get("target_nozzle_temp", 0.0) self.bedTemp = temperatures.get("bed_temp", 0.0) self.bedTargetTemp = temperatures.get("target_bed_temp", 0.0) self.chamberTemp = temperatures.get("chamber_temp", 0.0) if print_job.get("gcode_state") == "RUNNING" or print_job.get("gcode_state") == "PREPARE": if not self._sdPrintingSemaphore.is_set(): self._sdPrintingSemaphore.set() if self._sdPrintingPausedSemaphore.is_set(): self._sdPrintingPausedSemaphore.clear() self._sdPrintStarting = False if not self._sdPrinting: filename = print_job.get("subtask_name") if not self._sdFileListCache.get(filename.lower()): if self._sdFileListCache.get(f"{filename.lower()}.3mf"): filename = f"{filename.lower()}.3mf" elif self._sdFileListCache.get(f"{filename.lower()}.gcode.3mf"): filename = f"{filename.lower()}.gcode.3mf" self._selectSdFile(filename) self._startSdPrint(from_printer=True) # fuzzy math here to get print percentage to match BambuStudio self._selectedSdFilePos = int(self._selectedSdFileSize * ((print_job.get("print_percentage") + 1)/100)) if print_job.get("gcode_state") == "PAUSE": if not self._sdPrintingPausedSemaphore.is_set(): self._sdPrintingPausedSemaphore.set() if self._sdPrintingSemaphore.is_set(): self._sdPrintingSemaphore.clear() self._send("// action:paused") self._sendPaused() if ( print_job.get("gcode_state") == "FINISH" or print_job.get("gcode_state") == "FAILED" ): if self._sdPrintStarting is False: self._sdPrinting = False if self._sdPrintingSemaphore.is_set(): self._selectedSdFilePos = self._selectedSdFileSize self._finishSdPrint() def _create_connection(self): if (self._settings.get(["device_type"]) != "" and self._settings.get(["serial"]) != "" and self._settings.get(["serial"]) != "" and self._settings.get(["username"]) != "" and self._settings.get(["access_code"]) != "" ): asyncio.run(self._create_connection_async()) async def _create_connection_async(self): self.bambu = BambuClient(device_type=self._settings.get(["device_type"]), serial=self._settings.get(["serial"]), host=self._settings.get(["host"]), username="bblp" if self._settings.get_boolean(["local_mqtt"]) else self._settings.get(["username"]), access_code=self._settings.get(["access_code"]), local_mqtt=self._settings.get_boolean(["local_mqtt"]), region=self._settings.get(["region"]), email=self._settings.get(["email"]), auth_token=self._settings.get(["auth_token"]) ) self.bambu.connect(callback=self.new_update) self._logger.info(f"bambu connection status: {self.bambu.connected}") self._sendOk() # while True: # await asyncio.sleep(self.tick_rate) # self._processTemperatureQuery() def __str__(self): return "BAMBU(read_timeout={read_timeout},write_timeout={write_timeout},options={options})".format( read_timeout=self._read_timeout, write_timeout=self._write_timeout, options={"device_type": self._settings.get(["device_type"]), "host": self._settings.get(["host"])}, ) def _calculate_resend_every_n(self, resend_ratio): self._resend_every_n = (100 // resend_ratio) if resend_ratio else 0 def _reset(self): with self._incoming_lock: self._relative = True self._lastX = 0.0 self._lastY = 0.0 self._lastZ = 0.0 self._lastE = [0.0] * self.extruderCount self._lastF = 200 self._unitModifier = 1 self._feedrate_multiplier = 100 self._flowrate_multiplier = 100 self._sdCardReady = True self._sdPrinting = False self._sdPrintStarting = False if self._sdPrinter: self._sdPrinting = False self._sdPrintingSemaphore.clear() self._sdPrintingPausedSemaphore.clear() self._sdPrinter = None self._selectedSdFile = None self._selectedSdFileSize = None self._selectedSdFilePos = None if self._writingToSdHandle: try: self._writingToSdHandle.close() except Exception: pass self._writingToSd = False self._writingToSdHandle = None self._writingToSdFile = None self._newSdFilePos = None self._heatingUp = False self.current_line = 0 self.lastN = 0 self._debug_awol = False self._debug_sleep = 0 # self._sleepAfterNext.clear() # self._sleepAfter.clear() self._dont_answer = False self._broken_klipper_connection = False self._debug_drop_connection = False self._killed = False if self._sdstatus_reporter is not None: self._sdstatus_reporter.cancel() self._sdstatus_reporter = None self._clearQueue(self.incoming) self._clearQueue(self.outgoing) # self._clearQueue(self.buffered) if self._settings.get_boolean(["simulateReset"]): for item in self._settings.get(["resetLines"]): self._send(item + "\n") self._locked = self._settings.get_boolean(["locked"]) @property def timeout(self): return self._read_timeout @timeout.setter def timeout(self, value): self._logger.debug(f"Setting read timeout to {value}s") self._read_timeout = value @property def write_timeout(self): return self._write_timeout @write_timeout.setter def write_timeout(self, value): self._logger.debug(f"Setting write timeout to {value}s") self._write_timeout = value @property def port(self): return "BAMBU" @property def baudrate(self): return self._faked_baudrate # noinspection PyMethodMayBeStatic def _clearQueue(self, q): try: while q.get(block=False): q.task_done() continue except queue.Empty: pass def _processIncoming(self): linenumber = 0 next_wait_timeout = 0 def recalculate_next_wait_timeout(): nonlocal next_wait_timeout next_wait_timeout = time.monotonic() + self._waitInterval recalculate_next_wait_timeout() data = None buf = b"" while self.incoming is not None and not self._killed: try: data = self.incoming.get(timeout=0.01) data = to_bytes(data, encoding="ascii", errors="replace") self.incoming.task_done() except queue.Empty: continue except Exception: if self.incoming is None: # just got closed break if data is not None: buf += data nl = buf.find(b"\n") + 1 if nl > 0: data = buf[:nl] buf = buf[nl:] else: continue recalculate_next_wait_timeout() if data is None: continue self._received_lines += 1 # strip checksum if b"*" in data: checksum = int(data[data.rfind(b"*") + 1 :]) data = data[: data.rfind(b"*")] if not checksum == self._calculate_checksum(data): self._triggerResend(expected=self.current_line + 1) continue self.current_line += 1 elif self._settings.get_boolean(["forceChecksum"]): self._send(self._error("checksum_missing")) continue # track N = N + 1 if data.startswith(b"N") and b"M110" in data: linenumber = int(re.search(b"N([0-9]+)", data).group(1)) self.lastN = linenumber self.current_line = linenumber self._sendOk() continue elif data.startswith(b"N"): linenumber = int(re.search(b"N([0-9]+)", data).group(1)) expected = self.lastN + 1 if linenumber != expected: self._triggerResend(actual=linenumber) continue else: self.lastN = linenumber data = data.split(None, 1)[1].strip() data += b"\n" data = to_unicode(data, encoding="ascii", errors="replace").strip() # actual command handling command_match = BambuPrinter.command_regex.match(data) if command_match is not None: command = command_match.group(0) letter = command_match.group(1) try: # if we have a method _gcode_G, _gcode_M or _gcode_T, execute that first letter_handler = f"_gcode_{letter}" if hasattr(self, letter_handler): code = command_match.group(2) handled = getattr(self, letter_handler)(code, data) if handled: self._sendOk() continue # then look for a method _gcode_ and execute that if it exists command_handler = f"_gcode_{command}" if hasattr(self, command_handler): handled = getattr(self, command_handler)(data) if handled: self._sendOk() continue else: self._sendOk() if self.bambu.connected: GCODE_COMMAND = commands.SEND_GCODE_TEMPLATE GCODE_COMMAND['print']['param'] = data + "\n" if self.bambu.publish(GCODE_COMMAND): self._logger.info("command sent successfully") self._sendOk() continue finally: self._logger.debug(f"{data}") self._logger.debug("Closing down read loop") ##~~ command implementations # noinspection PyUnusedLocal def _gcode_M20(self, data: str) -> bool: if self._sdCardReady: self._listSd(incl_long="L" in data, incl_timestamp="T" in data) return True # noinspection PyUnusedLocal def _gcode_M21(self, data: str) -> bool: self._sdCardReady = True self._send("SD card ok") return True # noinspection PyUnusedLocal def _gcode_M22(self, data: str) -> bool: self._logger.debug("ignoring M22 command.") self._send("M22 disabled for Bambu") return True def _gcode_M23(self, data: str) -> bool: if self._sdCardReady: filename = data.split(None, 1)[1].strip() self._selectSdFile(filename) return True # noinspection PyUnusedLocal def _gcode_M24(self, data: str) -> bool: if self._sdCardReady: self._startSdPrint() return True # noinspection PyUnusedLocal def _gcode_M25(self, data: str) -> bool: if self._sdCardReady: self._pauseSdPrint() return True def _gcode_M524(self, data: str) -> bool: if self._sdCardReady: return self._cancelSdPrint() return False def _gcode_M26(self, data: str) -> bool: if data == "M26 S0": if self._sdCardReady: return self._cancelSdPrint() return False else: self._logger.debug("ignoring M26 command.") self._send("M26 disabled for Bambu") return True def _gcode_M27(self, data: str) -> bool: def report(): if self._sdCardReady: self._reportSdStatus() matchS = re.search(r"S([0-9]+)", data) if matchS: interval = int(matchS.group(1)) if self._sdstatus_reporter is not None: self._sdstatus_reporter.cancel() if interval > 0: self._sdstatus_reporter = RepeatedTimer(interval, report) self._sdstatus_reporter.start() else: self._sdstatus_reporter = None report() return True def _gcode_M28(self, data: str) -> bool: self._logger.debug("ignoring M28 command.") self._send("M28 disabled for Bambu") return True # noinspection PyUnusedLocal def _gcode_M29(self, data: str) -> bool: self._logger.debug("ignoring M29 command.") self._send("M29 disabled for Bambu") return True def _gcode_M30(self, data: str) -> bool: if self._sdCardReady: filename = data.split(None, 1)[1].strip() self._deleteSdFile(filename) return True def _gcode_M33(self, data: str) -> bool: self._logger.debug("ignoring M33 command.") self._send("M33 disabled for Bambu") return True # noinspection PyUnusedLocal def _gcode_M105(self, data: str) -> bool: self._processTemperatureQuery() return True # noinspection PyUnusedLocal def _gcode_M115(self, data: str) -> bool: self._send("Bambu Printer Integration") self._send("Cap:EXTENDED_M20:1") self._send("Cap:LFN_WRITE:1") self._send("Cap:LFN_WRITE:1") return True def _gcode_M117(self, data: str) -> bool: # we'll just use this to echo a message, to allow playing around with pause triggers result = re.search(r"M117\s+(.*)", data).group(1) self._send(f"echo:{result}") return False def _gcode_M118(self, data: str) -> bool: match = re.search(r"M118 (?:(?PA1|E1|Pn[012])\s)?(?P.*)", data) if not match: self._send("Unrecognized command parameters for M118") else: result = match.groupdict() text = result["text"] parameter = result["parameter"] if parameter == "A1": self._send(f"//{text}") elif parameter == "E1": self._send(f"echo:{text}") else: self._send(text) return True # noinspection PyUnusedLocal def _gcode_M400(self, data: str) -> bool: return True @staticmethod def _check_param_letters(letters, data): # Checks if any of the params (letters) are included in data # Purely for saving typing :) for param in list(letters): if param in data: return True ##~~ further helpers # noinspection PyMethodMayBeStatic def _calculate_checksum(self, line: bytes) -> int: checksum = 0 for c in bytearray(line): checksum ^= c return checksum def _kill(self): self._killed = True if self.bambu.connected: self.bambu.disconnect() self._send("echo:EMERGENCY SHUTDOWN DETECTED. KILLED.") def _triggerResend( self, expected: int = None, actual: int = None, checksum: int = None ) -> None: with self._incoming_lock: if expected is None: expected = self.lastN + 1 else: self.lastN = expected - 1 if actual is None: if checksum: self._send(self._error("checksum_mismatch")) else: self._send(self._error("checksum_missing")) else: self._send(self._error("lineno_mismatch", expected, actual)) def request_resend(): self._send("Resend:%d" % expected) # if not self._brokenResend: self._sendOk() request_resend() def _listSd(self, incl_long=False, incl_timestamp=False): line = "{dosname} {size} {timestamp} \"{name}\"" self._send("Begin file list") for item in map(lambda x: line.format(**x), self._getSdFiles()): self._send(item) self._send("End file list") def _mappedSdList(self) -> Dict[str, Dict[str, Any]]: result = {} host = self._settings.get(["host"]) access_code = self._settings.get(["access_code"]) ftp = IoTFTPSClient(f"{host}", 990, "bblp", f"{access_code}", ssl_implicit=True) filelist = ftp.list_files("", ".3mf") or [] for entry in filelist: if entry.startswith("/"): filename = entry[1:] else: filename = entry filesize = ftp.ftps_session.size(entry) date_str = ftp.ftps_session.sendcmd(f"MDTM {entry}").replace("213 ", "") filedate = datetime.datetime.strptime(date_str, "%Y%m%d%H%M%S").replace(tzinfo=datetime.timezone.utc).timestamp() dosname = get_dos_filename(filename, existing_filenames=list(result.keys())).lower() data = { "dosname": dosname, "name": filename, "path": filename, "size": filesize, "timestamp": unix_timestamp_to_m20_timestamp(int(filedate)) } result[dosname.lower()] = filename.lower() result[filename.lower()] = data filelistcache = ftp.list_files("cache/", ".3mf") or [] for entry in filelistcache: if entry.startswith("/"): filename = entry[1:].replace("cache/", "") else: filename = entry.replace("cache/", "") filesize = ftp.ftps_session.size(f"cache/{filename}") date_str = ftp.ftps_session.sendcmd(f"MDTM cache/{filename}").replace("213 ", "") filedate = datetime.datetime.strptime(date_str, "%Y%m%d%H%M%S").replace(tzinfo=datetime.timezone.utc).timestamp() dosname = get_dos_filename(filename, existing_filenames=list(result.keys())).lower() data = { "dosname": dosname, "name": filename, "path": "cache/"+filename, "size": filesize, "timestamp": unix_timestamp_to_m20_timestamp(int(filedate)) } result[dosname.lower()] = filename.lower() result[filename.lower()] = data return result def _getSdFileData(self, filename: str) -> Optional[Dict[str, Any]]: data = self._sdFileListCache.get(filename.lower()) if isinstance(data, str): data = self._sdFileListCache.get(data.lower()) return data def _getSdFiles(self) -> List[Dict[str, Any]]: self._sdFileListCache = self._mappedSdList() return [x for x in self._sdFileListCache.values() if isinstance(x, dict)] def _selectSdFile(self, filename: str, check_already_open: bool = False) -> None: if filename.startswith("/"): filename = filename[1:] file = self._getSdFileData(filename) if file is None: self._listSd(incl_long=True, incl_timestamp=True) self._sendOk() file = self._getSdFileData(filename) if file is None: self._send(f"{filename} open failed") return if self._selectedSdFile == file["path"] and check_already_open: return self._selectedSdFile = file["path"] self._selectedSdFileSize = file["size"] self._send(f"File opened: {file['name']} Size: {self._selectedSdFileSize}") self._send("File selected") def _startSdPrint(self, from_printer: bool = False) -> None: if self._selectedSdFile is not None: if self._sdPrinter is None: self._sdPrinting = True self._sdPrintStarting = True self._sdPrinter = threading.Thread(target=self._sdPrintingWorker, kwargs={"from_printer": from_printer}) self._sdPrinter.start() # self._sdPrintingSemaphore.set() if self._sdPrinter is not None: if self.bambu.connected: if self.bambu.publish(commands.RESUME): self._logger.info("print resumed") # if not self._sdPrintingSemaphore.is_set(): # self._sdPrintingSemaphore.set() else: self._logger.info("print resume failed") def _pauseSdPrint(self): if self.bambu.connected: if self.bambu.publish(commands.PAUSE): self._logger.info("print paused") else: self._logger.info("print pause failed") def _cancelSdPrint(self) -> bool: if self.bambu.connected: if self.bambu.publish(commands.STOP): self._logger.info("print cancelled") self._finishSdPrint() return True else: self._logger.info("print cancel failed") return False def _setSdPos(self, pos): self._newSdFilePos = pos def _reportSdStatus(self): if ( self._sdPrinter is not None or self._sdPrintStarting is True ) and self._selectedSdFileSize > 0: self._send(f"SD printing byte {self._selectedSdFilePos}/{self._selectedSdFileSize}") else: self._send("Not SD printing") def _generateTemperatureOutput(self) -> str: template = "{heater}:{actual:.2f}/ {target:.2f}" temps = collections.OrderedDict() temps["T"] = (self.temp[0], self.targetTemp[0]) temps["B"] = (self.bedTemp, self.bedTargetTemp) if self._hasChamber: temps["C"] = (self.chamberTemp, self.chamberTargetTemp) output = " ".join( map( lambda x: template.format(heater=x[0], actual=x[1][0], target=x[1][1]), temps.items(), ) ) output += " @:64\n" return output def _processTemperatureQuery(self): # includeOk = not self._okBeforeCommandOutput output = self._generateTemperatureOutput() self._send(output) def _writeSdFile(self, filename: str) -> None: self._send(f"Writing to file: {filename}") def _finishSdFile(self): try: self._writingToSdHandle.close() except Exception: pass finally: self._writingToSdHandle = None self._writingToSd = False self._selectedSdFile = None # Most printers don't have RTC and set some ancient date # by default. Emulate that using 2000-01-01 01:00:00 # (taken from prusa firmware behaviour) st = os.stat(self._writingToSdFile) os.utime(self._writingToSdFile, (st.st_atime, 946684800)) self._writingToSdFile = None self._send("Done saving file") def _sdPrintingWorker(self, from_printer: bool = False): self._selectedSdFilePos = 0 try: if not from_printer and self.bambu.connected: print_command = {"print": {"sequence_id": 0, "command": "project_file", "param": "Metadata/plate_1.gcode", "subtask_name": f"{self._selectedSdFile}", "url": f"file:///mnt/sdcard/{self._selectedSdFile}" if self._settings.get_boolean(["device_type"]) in ["X1", "X1C"] else f"file:///sdcard/{self._selectedSdFile}", "timelapse": self._settings.get_boolean(["timelapse"]), "bed_leveling": self._settings.get_boolean(["bed_leveling"]), "flow_cali": self._settings.get_boolean(["flow_cali"]), "vibration_cali": self._settings.get_boolean(["vibration_cali"]), "layer_inspect": self._settings.get_boolean(["layer_inspect"]), "use_ams": self._settings.get_boolean(["use_ams"]) } } self.bambu.publish(print_command) while self._selectedSdFilePos < self._selectedSdFileSize: if self._killed or not self._sdPrinting: break # if we are paused, wait for resuming self._sdPrintingSemaphore.wait() self._reportSdStatus() time.sleep(3) self._logger.debug(f"SD File Print: {self._selectedSdFile}") except AttributeError: if self.outgoing is not None: raise self._finishSdPrint() def _finishSdPrint(self): if not self._killed: self._sdPrintingSemaphore.clear() self._sdPrintingPausedSemaphore.clear() self._send("Done printing file") self._selectedSdFilePos = 0 self._selectedSdFileSize = 0 self._sdPrinting = False self._sdPrintStarting = False self._sdPrinter = None def _deleteSdFile(self, filename: str) -> None: host = self._settings.get(["host"]) access_code = self._settings.get(["access_code"]) if filename.startswith("/"): filename = filename[1:] file = self._getSdFileData(filename) if file is not None: ftp = IoTFTPSClient(f"{host}", 990, "bblp", f"{access_code}", ssl_implicit=True) try: if ftp.delete_file(file["path"]): self._logger.debug(f"{filename} deleted") else: raise Exception("delete failed") except Exception as e: self._logger.debug(f"Error deleting file {filename}") def _setBusy(self, reason="processing"): if not self._sendBusy: return def loop(): while self._busy: self._send(f"echo:busy {self._busy}") time.sleep(self._busyInterval) self._sendOk() self._busy = reason self._busy_loop = threading.Thread(target=loop) self._busy_loop.daemon = True self._busy_loop.start() def _setUnbusy(self): self._busy = None # def _processBuffer(self): # while self.buffered is not None: # try: # line = self.buffered.get(timeout=0.5) # except queue.Empty: # continue # # if line is None: # continue # # self.buffered.task_done() # # self._logger.debug("Closing down buffer loop") def _showPrompt(self, text, choices): self._hidePrompt() self._send(f"//action:prompt_begin {text}") for choice in choices: self._send(f"//action:prompt_button {choice}") self._send("//action:prompt_show") def _hidePrompt(self): self._send("//action:prompt_end") def write(self, data: bytes) -> int: data = to_bytes(data, errors="replace") u_data = to_unicode(data, errors="replace") with self._incoming_lock: if self.incoming is None or self.outgoing is None: return 0 if b"M112" in data: self._seriallog.debug(f"<<< {u_data}") self._kill() return len(data) try: written = self.incoming.put(data, timeout=self._write_timeout, partial=True) self._seriallog.debug(f"<<< {u_data}") return written except queue.Full: self._logger.info( "Incoming queue is full, raising SerialTimeoutException" ) raise SerialTimeoutException() def readline(self) -> bytes: timeout = self._read_timeout try: # fetch a line from the queue, wait no longer than timeout line = to_unicode(self.outgoing.get(timeout=timeout), errors="replace") self._seriallog.debug(f">>> {line.strip()}") self.outgoing.task_done() return to_bytes(line) except queue.Empty: # queue empty? return empty line return b"" def close(self): if self.bambu.connected: self.bambu.disconnect() self._killed = True self.incoming = None self.outgoing = None self.buffered = None def _sendOk(self): if self.outgoing is None: return ok = self._ok() if ok: self._send(ok) def _isPaused(self): return self._sdPrintingPausedSemaphore.is_set() def _sendPaused(self): paused_timer = RepeatedTimer(interval=3.0, function=self._send, args=[f"SD printing byte {self._selectedSdFilePos}/{self._selectedSdFileSize}"], daemon=True, run_first=True, condition=self._isPaused) paused_timer.start() def _send(self, line: str) -> None: if self.outgoing is not None: self.outgoing.put(line) def _ok(self): return "ok" def _error(self, error: str, *args, **kwargs) -> str: return f"Error: {self._errors.get(error).format(*args, **kwargs)}" # noinspection PyUnresolvedReferences class CharCountingQueue(queue.Queue): def __init__(self, maxsize, name=None): queue.Queue.__init__(self, maxsize=maxsize) self._size = 0 self._name = name def clear(self): with self.mutex: self.queue.clear() def put(self, item, block=True, timeout=None, partial=False) -> int: self.not_full.acquire() try: if not self._will_it_fit(item) and partial: space_left = self.maxsize - self._qsize() if space_left: item = item[:space_left] if not block: if not self._will_it_fit(item): raise queue.Full elif timeout is None: while not self._will_it_fit(item): self.not_full.wait() elif timeout < 0: raise ValueError("'timeout' must be a positive number") else: endtime = time.monotonic() + timeout while not self._will_it_fit(item): remaining = endtime - time.monotonic() if remaining <= 0: raise queue.Full self.not_full.wait(remaining) self._put(item) self.unfinished_tasks += 1 self.not_empty.notify() return self._len(item) finally: self.not_full.release() # noinspection PyMethodMayBeStatic def _len(self, item): return len(item) def _qsize(self, l=len): # noqa: E741 return self._size # Put a new item in the queue def _put(self, item): self.queue.append(item) self._size += self._len(item) # Get an item from the queue def _get(self): item = self.queue.popleft() self._size -= self._len(item) return item def _will_it_fit(self, item): return self.maxsize - self._qsize() >= self._len(item)