3D-Druck/OctoPrint-BambuPrinter
1
0
Fork 0
Code Pull Requests Actions 1 Packages Releases 1 Activity

WIP Refactor printer logic into states and subsystem objects.

Browse Source
This commit is contained in:
Anton Skrypnyk
2024-07-24 17:15:46 +03:00
parent 75b0a11fef
commit 155f3d2bd3
16 changed files with 1059 additions and 903 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
octoprint_bambu_printer/printer/__init__.py Normal file
View File

521
octoprint_bambu_printer/printer/bambu_virtual_printer.py Normal file
View File

@ -0,0 +1,521 @@
__author__ = "Gina Häußge <osd@foosel.net>"
__license__ = "GNU Affero General Public License http://www.gnu.org/licenses/agpl.html"
import collections
from dataclasses import dataclass, field
import math
import os
import queue
import re
import threading
import time
import asyncio
from pybambu import BambuClient, commands
import logging
import logging.handlers
from octoprint.util import RepeatedTimer
from octoprint_bambu_printer.printer.states.a_printer_state import APrinterState
from octoprint_bambu_printer.printer.states.idle_state import IdleState
from .printer_serial_io import PrinterSerialIO
from .states.print_finished_state import PrintFinishedState
from .states.paused_state import PausedState
from .states.printing_state import PrintingState
from .gcode_executor import GCodeExecutor
from .remote_sd_card_file_list import RemoteSDCardFileList
AMBIENT_TEMPERATURE: float = 21.3
@dataclass
class BambuPrinterTelemetry:
temp: list[float] = field(default_factory=lambda: [AMBIENT_TEMPERATURE])
targetTemp: list[float] = field(default_factory=lambda: [0.0])
bedTemp: float = AMBIENT_TEMPERATURE
bedTargetTemp = 0.0
hasChamber: bool = False
chamberTemp: float = AMBIENT_TEMPERATURE
chamberTargetTemp: float = 0.0
lastTempAt: float = time.monotonic()
firmwareName: str = "Bambu"
extruderCount: int = 1
# noinspection PyBroadException
class BambuVirtualPrinter:
gcode_executor = GCodeExecutor()
def __init__(
self,
settings,
printer_profile_manager,
data_folder,
serial_log_handler=None,
faked_baudrate=115200,
):
self._log = logging.getLogger("octoprint.plugins.bambu_printer.BambuPrinter")
self._state_idle = IdleState(self)
self._state_printing = PrintingState(self)
self._state_paused = PausedState(self)
self._state_finished = PrintFinishedState(self)
self._current_state = self._state_idle
self._serial_io = PrinterSerialIO(
self._process_gcode_serial_command,
serial_log_handler,
read_timeout=5.0,
write_timeout=10.0,
)
self.tick_rate = 2.0
self._telemetry = BambuPrinterTelemetry()
self._telemetry.hasChamber = printer_profile_manager.get_current().get(
"heatedChamber"
)
self._running = True
self.file_system = RemoteSDCardFileList(settings)
self._busy_reason = None
self._busy_loop = None
self._busy_interval = 2.0
self._settings = settings
self._printer_profile_manager = printer_profile_manager
self._faked_baudrate = faked_baudrate
self._plugin_data_folder = data_folder
self._last_hms_errors = None
self._serial_io.start()
self._bambu_client: BambuClient = None
asyncio.get_event_loop().run_until_complete(self._create_connection_async())
@property
def bambu_client(self):
if self._bambu_client is None:
raise ValueError("No connection to Bambulab was established")
return self._bambu_client
@property
def is_running(self):
return self._running
@property
def current_print_job(self):
if isinstance(self._current_state, PrintingState):
return self._current_state.print_job
return None
def change_state(self, new_state: APrinterState):
if self._current_state == new_state:
return
self._log.debug(
f"Changing state from {self._current_state.__class__.__name__} to {new_state.__class__.__name__}"
)
self._current_state.finalize()
self._current_state = new_state
self._current_state.init()
def new_update(self, event_type):
if event_type == "event_hms_errors":
self._update_hms_errors()
elif event_type == "event_printer_data_update":
self._update_printer_info()
def _update_printer_info(self):
device_data = self.bambu_client.get_device()
ams = device_data.ams.__dict__
print_job = device_data.print_job
temperatures = device_data.temperature.__dict__
lights = device_data.lights.__dict__
fans = device_data.fans.__dict__
speed = device_data.speed.__dict__
self.lastTempAt = time.monotonic()
self._telemetry.temp[0] = temperatures.get("nozzle_temp", 0.0)
self._telemetry.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.gcode_state == "RUNNING":
self.change_state(self._state_printing)
self._state_printing.set_print_job_info(print_job)
if print_job.gcode_state == "PAUSE":
self.change_state(self._state_paused)
if print_job.gcode_state == "FINISH" or print_job.gcode_state == "FAILED":
self.change_state(self._state_finished)
def _update_hms_errors(self):
bambu_printer = self.bambu_client.get_device()
if (
bambu_printer.hms.errors != self._last_hms_errors
and bambu_printer.hms.errors["Count"] > 0
):
self._log.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.sendIO(f"// action:notification {error}")
self._last_hms_errors = bambu_printer.hms.errors
def on_disconnect(self, on_disconnect):
self._log.debug(f"on disconnect called")
return on_disconnect
def on_connect(self, on_connect):
self._log.debug(f"on connect called")
return on_connect
async def _create_connection_async(self):
if (
self._settings.get(["device_type"]) == ""
or self._settings.get(["serial"]) == ""
or self._settings.get(["username"]) == ""
or self._settings.get(["access_code"]) == ""
):
self._log.debug("invalid settings to start connection with Bambu Printer")
return
self._log.debug(
f"connecting via local mqtt: {self._settings.get_boolean(['local_mqtt'])}"
)
self._bambu_client = 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_client.on_disconnect = self.on_disconnect(
self._bambu_client.on_disconnect
)
self._bambu_client.on_connect = self.on_connect(self._bambu_client.on_connect)
self._bambu_client.connect(callback=self.new_update)
self._log.info(f"bambu connection status: {self._bambu_client.connected}")
self._serial_io.sendOk()
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 _reset(self):
with self._serial_io.incoming_lock:
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._running = False
if self._sdstatus_reporter is not None:
self._sdstatus_reporter.cancel()
self._sdstatus_reporter = None
if self._settings.get_boolean(["simulateReset"]):
for item in self._settings.get(["resetLines"]):
self.sendIO(item + "\n")
self._locked = self._settings.get_boolean(["locked"])
self._serial_io.reset()
@property
def timeout(self):
return self._read_timeout
@timeout.setter
def timeout(self, value):
self._log.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._log.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
def write(self, data: bytes) -> int:
return self._serial_io.write(data)
def readline(self) -> bytes:
return self._serial_io.readline()
def sendIO(self, line: str):
self.sendIO(line)
def sendOk(self):
self._serial_io.sendOk()
##~~ command implementations
def run_gcode_handler(self, gcode, data):
self.gcode_executor.execute(self, gcode, data)
@gcode_executor.register("M23")
def _gcode_M23(self, data: str) -> bool:
filename = data.split(maxsplit=1)[1].strip()
self.file_system.select_file(filename)
return True
@gcode_executor.register("M26")
def _gcode_M26(self, data: str) -> bool:
if data == "M26 S0":
return self._cancelSdPrint()
else:
self._log.debug("ignoring M26 command.")
self.sendIO("M26 disabled for Bambu")
return True
@gcode_executor.register("M27")
def _gcode_M27(self, data: str) -> bool:
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, self.report_print_job_status
)
self._sdstatus_reporter.start()
else:
self._sdstatus_reporter = None
self.report_print_job_status()
return True
@gcode_executor.register("M30")
def _gcode_M30(self, data: str) -> bool:
filename = data.split(None, 1)[1].strip()
self.file_system.delete_file(filename)
return True
@gcode_executor.register("M105")
def _gcode_M105(self, data: str) -> bool:
return self._processTemperatureQuery()
# noinspection PyUnusedLocal
@gcode_executor.register("M115")
def _gcode_M115(self, data: str) -> bool:
self.sendIO("Bambu Printer Integration")
self.sendIO("Cap:EXTENDED_M20:1")
self.sendIO("Cap:LFN_WRITE:1")
self.sendIO("Cap:LFN_WRITE:1")
return True
@gcode_executor.register("M117")
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.sendIO(f"echo:{result}")
return False
@gcode_executor.register("M118")
def _gcode_M118(self, data: str) -> bool:
match = re.search(r"M118 (?:(?P<parameter>A1|E1|Pn[012])\s)?(?P<text>.*)", data)
if not match:
self.sendIO("Unrecognized command parameters for M118")
else:
result = match.groupdict()
text = result["text"]
parameter = result["parameter"]
if parameter == "A1":
self.sendIO(f"//{text}")
elif parameter == "E1":
self.sendIO(f"echo:{text}")
else:
self.sendIO(text)
return True
# noinspection PyUnusedLocal
@gcode_executor.register("M220")
def _gcode_M220(self, data: str) -> bool:
if self.bambu_client.connected:
gcode_command = commands.SEND_GCODE_TEMPLATE
percent = int(data[1:])
if percent is None or percent < 1 or percent > 166:
return True
speed_fraction = 100 / percent
acceleration = math.exp((speed_fraction - 1.0191) / -0.814)
feed_rate = (
2.1645 * (acceleration**3)
- 5.3247 * (acceleration**2)
+ 4.342 * acceleration
- 0.181
)
speed_level = 1.539 * (acceleration**2) - 0.7032 * acceleration + 4.0834
speed_command = f"M204.2 K${acceleration:.2f} \nM220 K${feed_rate:.2f} \nM73.2 R${speed_fraction:.2f} \nM1002 set_gcode_claim_speed_level ${speed_level:.0f}\n"
gcode_command["print"]["param"] = speed_command
if self.bambu_client.publish(gcode_command):
self._log.info(f"{percent}% speed adjustment command sent successfully")
return True
def _process_gcode_serial_command(self, gcode_letter: str, gcode: str, data: bytes):
self._log.debug(f"processing gcode command {gcode_letter} {gcode} {data}")
if gcode_letter in self.gcode_executor:
handled = self.run_gcode_handler(gcode_letter, data)
else:
handled = self.run_gcode_handler(gcode, data)
if handled:
self._serial_io.sendOk()
return
# post gcode to printer otherwise
if self.bambu_client.connected:
GCODE_COMMAND = commands.SEND_GCODE_TEMPLATE
GCODE_COMMAND["print"]["param"] = data + "\n"
if self.bambu_client.publish(GCODE_COMMAND):
self._log.info("command sent successfully")
self._serial_io.sendOk()
##~~ further helpers
@gcode_executor.register_no_data("M112")
def _kill(self):
self._running = True
if self.bambu_client.connected:
self.bambu_client.disconnect()
self.sendIO("echo:EMERGENCY SHUTDOWN DETECTED. KILLED.")
self._serial_io.stop()
@gcode_executor.register_no_data("M20")
def _listSd(self):
self.sendIO("Begin file list")
for item in map(lambda f: f.get_log_info(), self.file_system.get_all_files()):
self.sendIO(item)
self.sendIO("End file list")
@gcode_executor.register_no_data("M24")
def _startSdPrint(self, from_printer: bool = False) -> bool:
self._log.debug(f"_startSdPrint: from_printer={from_printer}")
self.change_state(self._state_printing)
@gcode_executor.register_no_data("M25")
def _pauseSdPrint(self):
if self.bambu_client.connected:
if self.bambu_client.publish(commands.PAUSE):
self._log.info("print paused")
else:
self._log.info("print pause failed")
@gcode_executor.register("M524")
def _cancelSdPrint(self) -> bool:
self._current_state.cancel()
def report_print_job_status(self):
print_job = self.current_print_job
if print_job is not None:
self.sendIO(
f"SD printing byte {print_job.file_position}/{print_job.file_info.size}"
)
else:
self.sendIO("Not SD printing")
def _generateTemperatureOutput(self) -> str:
template = "{heater}:{actual:.2f}/ {target:.2f}"
temps = collections.OrderedDict()
temps["T"] = (self._telemetry.temp[0], self._telemetry.targetTemp[0])
temps["B"] = (self.bedTemp, self.bedTargetTemp)
if self._telemetry.hasChamber:
temps["C"] = (self.chamberTemp, self._telemetry.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) -> bool:
# includeOk = not self._okBeforeCommandOutput
if self.bambu_client.connected:
output = self._generateTemperatureOutput()
self.sendIO(output)
return True
else:
return False
def _writeSdFile(self, filename: str) -> None:
self.sendIO(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.sendIO("Done saving file")
def _setMainThreadBusy(self, reason="processing"):
def loop():
while self._busy_reason is not None:
self.sendIO(f"echo:busy {self._busy_reason}")
time.sleep(self._busy_interval)
self._serial_io.sendOk()
self._busy_reason = reason
self._busy_loop = threading.Thread(target=loop)
self._busy_loop.daemon = True
self._busy_loop.start()
def _setMainThreadIdle(self):
self._busy_reason = None
def close(self):
if self.bambu_client.connected:
self.bambu_client.disconnect()
self._serial_io.stop()

67
octoprint_bambu_printer/printer/char_counting_queue.py Normal file
View File

@ -0,0 +1,67 @@
import queue
import time
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)

1
octoprint_bambu_printer/printer/ftpsclient/__init__.py Normal file
View File

@ -0,0 +1 @@
from .ftpsclient import IoTFTPSClient

228
octoprint_bambu_printer/printer/ftpsclient/ftpsclient.py Normal file
View File

@ -0,0 +1,228 @@
"""
Based on: <https://github.com/dgonzo27/py-iot-utils>
MIT License
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
wrapper for FTPS server interactions
"""
import ftplib
import os
import socket
import ssl
from typing import Optional, Union, List
from contextlib import redirect_stdout
import io
import re
class ImplicitTLS(ftplib.FTP_TLS):
"""ftplib.FTP_TLS sub-class to support implicit SSL FTPS"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._sock = None
@property
def sock(self):
"""return socket"""
return self._sock
@sock.setter
def sock(self, value):
"""wrap and set SSL socket"""
if value is not None and not isinstance(value, ssl.SSLSocket):
value = self.context.wrap_socket(value)
self._sock = value
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) # this is the fix
return conn, size
class IoTFTPSClient:
"""iot ftps ftpsclient"""
ftps_host: str
ftps_port: int
ftps_user: str
ftps_pass: str
ssl_implicit: bool
ftps_session: Union[ftplib.FTP, ImplicitTLS]
last_error: Optional[str] = None
welcome: str
def __init__(
self,
ftps_host: str,
ftps_port: Optional[int] = 21,
ftps_user: Optional[str] = "",
ftps_pass: Optional[str] = "",
ssl_implicit: Optional[bool] = False,
) -> None:
self.ftps_host = ftps_host
self.ftps_port = ftps_port
self.ftps_user = ftps_user
self.ftps_pass = ftps_pass
self.ssl_implicit = ssl_implicit
self.instantiate_ftps_session()
def __repr__(self) -> str:
return (
"IoT FTPS Client\n"
"--------------------\n"
f"host: {self.ftps_host}\n"
f"port: {self.ftps_port}\n"
f"user: {self.ftps_user}\n"
f"ssl: {self.ssl_implicit}"
)
def instantiate_ftps_session(self) -> None:
"""init ftps_session based on input params"""
self.ftps_session = ImplicitTLS() if self.ssl_implicit else ftplib.FTP()
self.ftps_session.set_debuglevel(0)
self.welcome = self.ftps_session.connect(
host=self.ftps_host, port=self.ftps_port)
if self.ftps_user and self.ftps_pass:
self.ftps_session.login(user=self.ftps_user, passwd=self.ftps_pass)
else:
self.ftps_session.login()
if self.ssl_implicit:
self.ftps_session.prot_p()
def disconnect(self) -> None:
"""disconnect the current session from the ftps server"""
self.ftps_session.close()
def download_file(self, source: str, dest: str):
"""download a file to a path on the local filesystem"""
with open(dest, "wb") as file:
self.ftps_session.retrbinary(f"RETR {source}", file.write)
def upload_file(self, source: str, dest: str, callback=None) -> bool:
"""upload a file to a path inside the FTPS server"""
file_size = os.path.getsize(source)
block_size = max(file_size // 100, 8192)
rest = None
try:
# Taken from ftplib.storbinary but with custom ssl handling
# due to the shitty bambu p1p ftps server TODO fix properly.
with open(source, "rb") as fp:
self.ftps_session.voidcmd('TYPE I')
with self.ftps_session.transfercmd(f"STOR {dest}", rest) as conn:
while 1:
buf = fp.read(block_size)
if not buf:
break
conn.sendall(buf)
if callback:
callback(buf)
# shutdown ssl layer
if ftplib._SSLSocket is not None and isinstance(conn, ftplib._SSLSocket):
# Yeah this is suposed to be conn.unwrap
# But since we operate in prot p mode
# we can close the connection always.
# This is cursed but it works.
if "vsFTPd" in self.welcome:
conn.unwrap()
else:
conn.shutdown(socket.SHUT_RDWR)
return True
except Exception as ex:
print(f"unexpected exception occurred: {ex}")
pass
return False
def delete_file(self, path: str) -> bool:
"""delete a file from under a path inside the FTPS server"""
try:
self.ftps_session.delete(path)
return True
except Exception as ex:
print(f"unexpected exception occurred: {ex}")
pass
return False
def move_file(self, source: str, dest: str):
"""move a file inside the FTPS server to another path inside the FTPS server"""
self.ftps_session.rename(source, dest)
def mkdir(self, path: str) -> str:
return self.ftps_session.mkd(path)
def list_files(self, path: str, file_pattern: Optional[str] = None) -> Union[List[str], None]:
"""list files under a path inside the FTPS server"""
try:
files = self.ftps_session.nlst(path)
if not files:
return
if file_pattern:
return [f for f in files if file_pattern in f]
return files
except Exception as ex:
print(f"unexpected exception occurred: {ex}")
pass
return
def list_files_ex(self, path: str) -> Union[list[str], None]:
"""list files under a path inside the FTPS server"""
try:
f = io.StringIO()
with redirect_stdout(f):
self.ftps_session.dir(path)
s = f.getvalue()
files = []
for row in s.split("\n"):
if len(row) <= 0: continue
attribs = row.split(" ")
match = re.search(r".*\ (\d\d\:\d\d|\d\d\d\d)\ (.*)", row)
name = ""
if match:
name = match.groups(1)[1]
else:
name = attribs[len(attribs) - 1]
file = ( attribs[0], name )
files.append(file)
return files
except Exception as ex:
print(f"unexpected exception occurred: [{ex}]")
pass
return

314
octoprint_bambu_printer/printer/gcode_executor.py Normal file
View File

@ -0,0 +1,314 @@
import itertools
import logging
from inspect import signature
GCODE_DOCUMENTATION = {
"G0": "Linear Move",
"G1": "Linear Move",
"G2": "Arc or Circle Move",
"G3": "Arc or Circle Move",
"G4": "Dwell",
"G5": "Bézier cubic spline",
"G6": "Direct Stepper Move",
"G10": "Retract",
"G11": "Recover",
"G12": "Clean the Nozzle",
"G17": "CNC Workspace Planes",
"G18": "CNC Workspace Planes",
"G19": "CNC Workspace Planes",
"G20": "Inch Units",
"G21": "Millimeter Units",
"G26": "Mesh Validation Pattern",
"G27": "Park toolhead",
"G28": "Auto Home",
"G29": "Bed Leveling",
"G29": "Bed Leveling (3-Point)",
"G29": "Bed Leveling (Linear)",
"G29": "Bed Leveling (Manual)",
"G29": "Bed Leveling (Bilinear)",
"G29": "Bed Leveling (Unified)",
"G30": "Single Z-Probe",
"G31": "Dock Sled",
"G32": "Undock Sled",
"G33": "Delta Auto Calibration",
"G34": "Z Steppers Auto-Alignment",
"G34": "Mechanical Gantry Calibration",
"G35": "Tramming Assistant",
"G38.2": "Probe target",
"G38.3": "Probe target",
"G38.4": "Probe target",
"G38.5": "Probe target",
"G42": "Move to mesh coordinate",
"G53": "Move in Machine Coordinates",
"G60": "Save Current Position",
"G61": "Return to Saved Position",
"G76": "Probe temperature calibration",
"G80": "Cancel Current Motion Mode",
"G90": "Absolute Positioning",
"G91": "Relative Positioning",
"G92": "Set Position",
"G425": "Backlash Calibration",
"M0": "Unconditional stop",
"M1": "Unconditional stop",
"M3": "Spindle CW / Laser On",
"M4": "Spindle CCW / Laser On",
"M5": "Spindle / Laser Off",
"M7": "Coolant Controls",
"M8": "Coolant Controls",
"M9": "Coolant Controls",
"M10": "Vacuum / Blower Control",
"M11": "Vacuum / Blower Control",
"M16": "Expected Printer Check",
"M17": "Enable Steppers",
"M18": "Disable steppers",
"M84": "Disable steppers",
"M20": "List SD Card",
"M21": "Init SD card",
"M22": "Release SD card",
"M23": "Select SD file",
"M24": "Start or Resume SD print",
"M25": "Pause SD print",
"M26": "Set SD position",
"M27": "Report SD print status",
"M28": "Start SD write",
"M29": "Stop SD write",
"M30": "Delete SD file",
"M31": "Print time",
"M32": "Select and Start",
"M33": "Get Long Path",
"M34": "SDCard Sorting",
"M42": "Set Pin State",
"M43": "Debug Pins",
"M48": "Probe Repeatability Test",
"M73": "Set Print Progress",
"M75": "Start Print Job Timer",
"M76": "Pause Print Job Timer",
"M77": "Stop Print Job Timer",
"M78": "Print Job Stats",
"M80": "Power On",
"M81": "Power Off",
"M82": "E Absolute",
"M83": "E Relative",
"M85": "Inactivity Shutdown",
"M86": "Hotend Idle Timeout",
"M87": "Disable Hotend Idle Timeout",
"M92": "Set Axis Steps-per-unit",
"M100": "Free Memory",
"M102": "Configure Bed Distance Sensor",
"M104": "Set Hotend Temperature",
"M105": "Report Temperatures",
"M106": "Set Fan Speed",
"M107": "Fan Off",
"M108": "Break and Continue",
"M109": "Wait for Hotend Temperature",
"M110": "Set / Get Line Number",
"M111": "Debug Level",
"M112": "Full Shutdown",
"M113": "Host Keepalive",
"M114": "Get Current Position",
"M115": "Firmware Info",
"M117": "Set LCD Message",
"M118": "Serial print",
"M119": "Endstop States",
"M120": "Enable Endstops",
"M121": "Disable Endstops",
"M122": "TMC Debugging",
"M123": "Fan Tachometers",
"M125": "Park Head",
"M126": "Baricuda 1 Open",
"M127": "Baricuda 1 Close",
"M128": "Baricuda 2 Open",
"M129": "Baricuda 2 Close",
"M140": "Set Bed Temperature",
"M141": "Set Chamber Temperature",
"M143": "Set Laser Cooler Temperature",
"M145": "Set Material Preset",
"M149": "Set Temperature Units",
"M150": "Set RGB(W) Color",
"M154": "Position Auto-Report",
"M155": "Temperature Auto-Report",
"M163": "Set Mix Factor",
"M164": "Save Mix",
"M165": "Set Mix",
"M166": "Gradient Mix",
"M190": "Wait for Bed Temperature",
"M191": "Wait for Chamber Temperature",
"M192": "Wait for Probe temperature",
"M193": "Set Laser Cooler Temperature",
"M200": "Set Filament Diameter",
"M201": "Print / Travel Move Limits",
"M203": "Set Max Feedrate",
"M204": "Set Starting Acceleration",
"M205": "Set Advanced Settings",
"M206": "Set Home Offsets",
"M207": "Set Firmware Retraction",
"M208": "Firmware Recover",
"M209": "Set Auto Retract",
"M211": "Software Endstops",
"M217": "Filament swap parameters",
"M218": "Set Hotend Offset",
"M220": "Set Feedrate Percentage",
"M221": "Set Flow Percentage",
"M226": "Wait for Pin State",
"M240": "Trigger Camera",
"M250": "LCD Contrast",
"M255": "LCD Sleep/Backlight Timeout",
"M256": "LCD Brightness",
"M260": "I2C Send",
"M261": "I2C Request",
"M280": "Servo Position",
"M281": "Edit Servo Angles",
"M282": "Detach Servo",
"M290": "Babystep",
"M300": "Play Tone",
"M301": "Set Hotend PID",
"M302": "Cold Extrude",
"M303": "PID autotune",
"M304": "Set Bed PID",
"M305": "User Thermistor Parameters",
"M306": "Model Predictive Temp. Control",
"M350": "Set micro-stepping",
"M351": "Set Microstep Pins",
"M355": "Case Light Control",
"M360": "SCARA Theta A",
"M361": "SCARA Theta-B",
"M362": "SCARA Psi-A",
"M363": "SCARA Psi-B",
"M364": "SCARA Psi-C",
"M380": "Activate Solenoid",
"M381": "Deactivate Solenoids",
"M400": "Finish Moves",
"M401": "Deploy Probe",
"M402": "Stow Probe",
"M403": "MMU2 Filament Type",
"M404": "Set Filament Diameter",
"M405": "Filament Width Sensor On",
"M406": "Filament Width Sensor Off",
"M407": "Filament Width",
"M410": "Quickstop",
"M412": "Filament Runout",
"M413": "Power-loss Recovery",
"M420": "Bed Leveling State",
"M421": "Set Mesh Value",
"M422": "Set Z Motor XY",
"M423": "X Twist Compensation",
"M425": "Backlash compensation",
"M428": "Home Offsets Here",
"M430": "Power Monitor",
"M486": "Cancel Objects",
"M493": "Fixed-Time Motion",
"M500": "Save Settings",
"M501": "Restore Settings",
"M502": "Factory Reset",
"M503": "Report Settings",
"M504": "Validate EEPROM contents",
"M510": "Lock Machine",
"M511": "Unlock Machine",
"M512": "Set Passcode",
"M524": "Abort SD print",
"M540": "Endstops Abort SD",
"M569": "Set TMC stepping mode",
"M575": "Serial baud rate",
"M592": "Nonlinear Extrusion Control",
"M593": "ZV Input Shaping",
"M600": "Filament Change",
"M603": "Configure Filament Change",
"M605": "Multi Nozzle Mode",
"M665": "Delta Configuration",
"M665": "SCARA Configuration",
"M666": "Set Delta endstop adjustments",
"M666": "Set dual endstop offsets",
"M672": "Duet Smart Effector sensitivity",
"M701": "Load filament",
"M702": "Unload filament",
"M710": "Controller Fan settings",
"M808": "Repeat Marker",
"M851": "XYZ Probe Offset",
"M852": "Bed Skew Compensation",
"M871": "Probe temperature config",
"M876": "Handle Prompt Response",
"M900": "Linear Advance Factor",
"M906": "Stepper Motor Current",
"M907": "Set Motor Current",
"M908": "Set Trimpot Pins",
"M909": "DAC Print Values",
"M910": "Commit DAC to EEPROM",
"M911": "TMC OT Pre-Warn Condition",
"M912": "Clear TMC OT Pre-Warn",
"M913": "Set Hybrid Threshold Speed",
"M914": "TMC Bump Sensitivity",
"M915": "TMC Z axis calibration",
"M916": "L6474 Thermal Warning Test",
"M917": "L6474 Overcurrent Warning Test",
"M918": "L6474 Speed Warning Test",
"M919": "TMC Chopper Timing",
"M928": "Start SD Logging",
"M951": "Magnetic Parking Extruder",
"M993": "Back up flash settings to SD",
"M994": "Restore flash from SD",
"M995": "Touch Screen Calibration",
"M997": "Firmware update",
"M999": "STOP Restart",
"M7219": "MAX7219 Control",
}
class GCodeExecutor:
def __init__(self):
self._log = logging.getLogger(
"octoprint.plugins.bambu_printer.BambuPrinter.gcode_executor"
)
self.handler_names = set()
self.gcode_handlers = {}
self.gcode_handlers_no_data = {}
def __contains__(self, item):
return item in self.gcode_handlers or item in self.gcode_handlers_no_data
def _get_required_args_count(self, func):
sig = signature(func)
required_count = sum(
1
for p in sig.parameters.values()
if (p.kind == p.POSITIONAL_OR_KEYWORD or p.kind == p.POSITIONAL_ONLY)
and p.default == p.empty
)
return required_count
def register(self, gcode):
def decorator(func):
required_count = self._get_required_args_count(func)
if required_count == 1:
self.gcode_handlers_no_data[gcode] = func
elif required_count == 2:
self.gcode_handlers[gcode] = func
else:
raise ValueError(
f"Cannot register function with {required_count} required parameters"
)
return func
return decorator
def register_no_data(self, gcode):
def decorator(func):
self.gcode_handlers_no_data[gcode] = func
return func
return decorator
def execute(self, printer, gcode, data):
gcode_info = self._gcode_with_info(gcode)
if gcode in self.gcode_handlers:
self._log.debug(f"Executing {gcode_info}")
return self.gcode_handlers[gcode](printer, data)
elif gcode in self.gcode_handlers_no_data:
self._log.debug(f"Executing {gcode_info}")
return self.gcode_handlers_no_data[gcode](printer)
else:
self._log.debug(f"ignoring {gcode_info} command.")
return True
def _gcode_with_info(self, gcode):
return f"{gcode} ({GCODE_DOCUMENTATION.get(gcode, 'Info not specified')})"

20
octoprint_bambu_printer/printer/print_job.py Normal file
View File

@ -0,0 +1,20 @@
from __future__ import annotations
from dataclasses import dataclass
from octoprint_bambu_printer.printer.remote_sd_card_file_list import FileInfo
@dataclass
class PrintJob:
file_info: FileInfo
file_position: int
@property
def progress(self):
if self.file_info.size is None:
return 100
return 100 * self.file_position / self.file_info.size
@progress.setter
def progress(self, value):
self.file_position = int(self.file_info.size * ((value + 1) / 100))

265
octoprint_bambu_printer/printer/printer_serial_io.py Normal file
View File

@ -0,0 +1,265 @@
import logging
import queue
import re
import threading
import time
from typing import Callable
from octoprint.util import to_bytes, to_unicode
from serial import SerialTimeoutException
from .char_counting_queue import CharCountingQueue
class PrinterSerialIO(threading.Thread):
command_regex = re.compile(r"^([GM])(\d+)")
def __init__(
self,
handle_command_callback: Callable[[str, str, bytes], None],
settings,
serial_log_handler=None,
read_timeout=5.0,
write_timeout=10.0,
) -> None:
super().__init__(
name="octoprint.plugins.bambu_printer.wait_thread", daemon=True
)
self._handle_command_callback = handle_command_callback
self._settings = settings
self._serial_log = logging.getLogger(
"octoprint.plugins.bambu_printer.BambuPrinter.serial"
)
self._serial_log.setLevel(logging.CRITICAL)
self._serial_log.propagate = False
if serial_log_handler is not None:
self._serial_log.addHandler(serial_log_handler)
self._serial_log.setLevel(logging.INFO)
self._serial_log.debug("-" * 78)
self._read_timeout = read_timeout
self._write_timeout = write_timeout
self._received_lines = 0
self._wait_interval = 5.0
self._running = True
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.command_queue = queue.Queue()
@property
def incoming_lock(self):
return self._incoming_lock
def run(self) -> None:
linenumber = 0
next_wait_timeout = 0
def recalculate_next_wait_timeout():
nonlocal next_wait_timeout
next_wait_timeout = time.monotonic() + self._wait_interval
recalculate_next_wait_timeout()
data = None
buf = b""
while self.incoming is not None and self._running:
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._format_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"
command = to_unicode(data, encoding="ascii", errors="replace").strip()
# actual command handling
command_match = self.command_regex.match(command)
if command_match is not None:
gcode = command_match.group(0)
gcode_letter = command_match.group(1)
self._handle_command_callback(gcode_letter, gcode, data)
self._serial_log.debug("Closing down read loop")
def stop(self):
self._running = False
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.is_closed():
return 0
try:
written = self.incoming.put(
data, timeout=self._write_timeout, partial=True
)
self._serial_log.debug(f"<<< {u_data}")
return written
except queue.Full:
self._serial_log.error(
"Incoming queue is full, raising SerialTimeoutException"
)
raise SerialTimeoutException()
def readline(self) -> bytes:
assert self.outgoing is not None
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._serial_log.debug(f">>> {line.strip()}")
self.outgoing.task_done()
return to_bytes(line)
except queue.Empty:
# queue empty? return empty line
return b""
def send(self, line: str) -> None:
if self.outgoing is not None:
self.outgoing.put(line)
def sendOk(self):
if self.outgoing is None:
return
self.send("ok")
def reset(self):
if self.incoming is not None:
self._clearQueue(self.incoming)
if self.outgoing is not None:
self._clearQueue(self.outgoing)
def close(self):
self.stop()
self.incoming = None
self.outgoing = None
def is_closed(self):
return self.incoming is None or self.outgoing is None
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._format_error("checksum_mismatch"))
else:
self.send(self._format_error("checksum_missing"))
else:
self.send(self._format_error("lineno_mismatch", expected, actual))
def request_resend():
self.send("Resend:%d" % expected)
self.sendOk()
request_resend()
def _calculate_checksum(self, line: bytes) -> int:
checksum = 0
for c in bytearray(line):
checksum ^= c
return checksum
def _format_error(self, error: str, *args, **kwargs) -> str:
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 {}",
}
return f"Error: {errors.get(error).format(*args, **kwargs)}"
def _clearQueue(self, q: queue.Queue):
try:
while q.get(block=False):
q.task_done()
continue
except queue.Empty:
pass

171
octoprint_bambu_printer/printer/remote_sd_card_file_list.py Normal file
View File

@ -0,0 +1,171 @@
from __future__ import annotations
from dataclasses import asdict, dataclass
import datetime
import itertools
from pathlib import Path
from typing import Any, Dict, Iterator, List, Optional
import logging.handlers
from octoprint.util import get_dos_filename
from octoprint.util.files import unix_timestamp_to_m20_timestamp
from .ftpsclient import IoTFTPSClient
@dataclass(frozen=True)
class FileInfo:
dosname: str
path: Path
size: int | None
timestamp: str
@property
def file_name(self):
return self.path.name.lower()
def get_log_info(self):
return f'{self.dosname} {self.size} {self.timestamp} "{self.file_name}"'
def to_dict(self):
return asdict(self)
class RemoteSDCardFileList:
def __init__(self, settings) -> None:
self._settings = settings
self._file_alias_cache = {}
self._file_data_cache = {}
self._selected_file_info: FileInfo | None = None
self._logger = logging.getLogger("octoprint.plugins.bambu_printer.BambuPrinter")
@property
def selected_file(self):
return self._selected_file_info
@property
def has_selected_file(self):
return self._selected_file_info is not None
def _get_ftp_file_info(
self, ftp: IoTFTPSClient, ftp_path, file_path: Path, existing_files: list[str]
):
file_size = ftp.ftps_session.size(ftp_path)
date_str = ftp.ftps_session.sendcmd(f"MDTM {ftp_path}").replace("213 ", "")
filedate = (
datetime.datetime.strptime(date_str, "%Y%m%d%H%M%S")
.replace(tzinfo=datetime.timezone.utc)
.timestamp()
)
file_name = file_path.name.lower()
dosname = get_dos_filename(file_name, existing_filenames=existing_files).lower()
return FileInfo(
dosname,
file_path,
file_size,
unix_timestamp_to_m20_timestamp(int(filedate)),
)
def _scan_ftp_file_list(
self, ftp, files: list[str], existing_files: list[str]
) -> Iterator[FileInfo]:
for entry in files:
ftp_path = Path(entry)
file_info = self._get_ftp_file_info(ftp, entry, ftp_path, existing_files)
yield file_info
existing_files.append(file_info.file_name)
def _get_existing_files_info(self):
host = self._settings.get(["host"])
access_code = self._settings.get(["access_code"])
ftp = IoTFTPSClient(str(host), 990, "bblp", str(access_code), ssl_implicit=True)
all_files_info: list[FileInfo] = []
existing_files = []
filelist = ftp.list_files("", ".3mf") or []
all_files_info.extend(self._scan_ftp_file_list(ftp, filelist, existing_files))
filelist_cache = ftp.list_files("cache/", ".3mf") or []
all_files_info.extend(
self._scan_ftp_file_list(ftp, filelist_cache, existing_files)
)
return all_files_info
def _get_file_data(self, file_path: str) -> FileInfo | None:
self._logger.debug(f"_getSdFileData: {file_path}")
file_name = Path(file_path).name.lower()
full_file_name = self._file_alias_cache.get(file_name, None)
if full_file_name is not None:
data = self._file_data_cache.get(file_name, None)
self._logger.debug(f"_getSdFileData: {data}")
return data
def get_all_files(self):
self._update_existing_files_info()
self._logger.debug(f"_getSdFiles return: {self._file_data_cache}")
return list(self._file_data_cache.values())
def _update_existing_files_info(self):
file_info_list = self._get_existing_files_info()
self._file_alias_cache = {
info.dosname: info.file_name for info in file_info_list
}
self._file_data_cache = {info.file_name: info for info in file_info_list}
def search_by_stem(self, file_stem: str, allowed_suffixes: list[str]):
for file_name in self._file_data_cache:
file_data = self._get_file_data(file_name)
if file_data is None:
continue
file_path = file_data.path
if file_path.stem == file_stem and any(
s in allowed_suffixes for s in file_path.suffixes
):
return file_data
return None
def select_file(self, file_path: str, check_already_open: bool = False) -> bool:
self._logger.debug(
f"_selectSdFile: {file_path}, check_already_open={check_already_open}"
)
file_name = Path(file_path).name
file_info = self._get_file_data(file_name)
if file_info is None:
file_info = self._get_file_data(file_name)
if file_info is None:
self._logger.error(f"{file_name} open failed")
return False
if (
self._selected_file_info is not None
and self._selected_file_info.path == file_info.path
and check_already_open
):
return True
self._selected_file_info = file_info
self._logger.info(
f"File opened: {self._selected_file_info.file_name} Size: {self._selected_file_info.size}"
)
return True
def delete_file(self, file_path: str) -> None:
host = self._settings.get(["host"])
access_code = self._settings.get(["access_code"])
file_info = self._get_file_data(file_path)
if file_info is not None:
ftp = IoTFTPSClient(
f"{host}", 990, "bblp", f"{access_code}", ssl_implicit=True
)
try:
if ftp.delete_file(str(file_info.path)):
self._logger.debug(f"{file_path} deleted")
else:
raise Exception("delete failed")
except Exception as e:
self._logger.debug(f"Error deleting file {file_path}")

0
octoprint_bambu_printer/printer/states/__init__.py Normal file
View File

39
octoprint_bambu_printer/printer/states/a_printer_state.py Normal file
View File

@ -0,0 +1,39 @@
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from octoprint_bambu_printer.printer.bambu_virtual_printer import (
BambuVirtualPrinter,
)
class APrinterState:
def __init__(self, printer: BambuVirtualPrinter) -> None:
self._log = logging.getLogger(
"octoprint.plugins.bambu_printer.BambuPrinter.states"
)
self._printer = printer
def init(self):
pass
def finalize(self):
pass
def handle_gcode(self, gcode):
self._log.debug(f"{self.__class__.__name__} gcode execution disabled")
def connect(self):
self._log_skip_state_transition("connect")
def pause(self):
self._log_skip_state_transition("pause")
def cancel(self):
self._log_skip_state_transition("cancel")
def resume(self):
self._log_skip_state_transition("resume")
def _log_skip_state_transition(self, method):
self._log.debug(f"skipping {self.__class__.__name__} state transition {method}")

7
octoprint_bambu_printer/printer/states/idle_state.py Normal file
View File

@ -0,0 +1,7 @@
from __future__ import annotations
from octoprint_bambu_printer.printer.states.a_printer_state import APrinterState
class IdleState(APrinterState):
pass

37
octoprint_bambu_printer/printer/states/paused_state.py Normal file
View File

@ -0,0 +1,37 @@
import threading
from octoprint.util import RepeatedTimer
from octoprint_bambu_printer.printer.bambu_virtual_printer import BambuVirtualPrinter
from octoprint_bambu_printer.printer.states.a_printer_state import APrinterState
class PausedState(APrinterState):
def __init__(self, printer: BambuVirtualPrinter) -> None:
super().__init__(printer)
self._pausedLock = threading.Event()
def init(self):
if not self._pausedLock.is_set():
self._pausedLock.set()
self._printer.sendIO("// action:paused")
self._sendPaused()
def finalize(self):
if self._pausedLock.is_set():
self._pausedLock.clear()
def _sendPaused(self):
if self._printer.current_print_job is None:
self._log.warn("job paused, but no print job available?")
return
paused_timer = RepeatedTimer(
interval=3.0,
function=self._printer.report_print_job_status,
daemon=True,
run_first=True,
condition=self._pausedLock.is_set,
)
paused_timer.start()

19
octoprint_bambu_printer/printer/states/print_finished_state.py Normal file
View File

@ -0,0 +1,19 @@
from __future__ import annotations
from octoprint_bambu_printer.printer.states.a_printer_state import APrinterState
class PrintFinishedState(APrinterState):
def init(self):
if self._printer.current_print_job is not None:
self._printer.current_print_job.progress = 100
self._finishSdPrint()
def _finishSdPrint(self):
if self._printer.is_running:
self._printer.sendIO("Done printing file")
self._selectedSdFilePos = 0
self._selectedSdFileSize = 0
self._sdPrinting = False
self._sdPrintStarting = False
self._sdPrinter = None

131
octoprint_bambu_printer/printer/states/printing_state.py Normal file
View File

@ -0,0 +1,131 @@
from __future__ import annotations
import threading
import pybambu
import pybambu.models
import pybambu.commands
from octoprint_bambu_printer.printer.bambu_virtual_printer import BambuVirtualPrinter
from octoprint_bambu_printer.printer.print_job import PrintJob
from octoprint_bambu_printer.printer.states.a_printer_state import APrinterState
class PrintingState(APrinterState):
def __init__(self, printer: BambuVirtualPrinter) -> None:
super().__init__(printer)
self._printingLock = threading.Event()
self._print_job: PrintJob | None = None
self._sd_printing_thread = None
@property
def print_job(self):
return self._print_job
def init(self):
if not self._printingLock.is_set():
self._printingLock.set()
def finalize(self):
if self._printingLock.is_set():
self._printingLock.clear()
def _start_worker_thread(self, from_printer: bool = False):
if self._sd_printing_thread is None:
self._sdPrinting = True
self._sdPrintStarting = True
self._sd_printing_thread = threading.Thread(
target=self._printing_worker, kwargs={"from_printer": from_printer}
)
self._sd_printing_thread.start()
def set_print_job_info(self, print_job_info):
filename: str = print_job_info.get("subtask_name")
project_file_info = self._printer.file_system.search_by_stem(
filename, [".3mf", ".gcode.3mf"]
)
if project_file_info is None:
self._log.debug(f"No 3mf file found for {print_job_info}")
return
if self._printer.file_system.select_file(filename):
self._printer.sendOk()
self.start_new_print(from_printer=True)
# fuzzy math here to get print percentage to match BambuStudio
progress = print_job_info.get("print_percentage")
self._print_job = PrintJob(project_file_info, 0)
self._print_job.progress =
def start_new_print(self, from_printer: bool = False):
if self._printer.file_system.selected_file is not None:
self._start_worker_thread(from_printer)
if self._sd_printing_thread is not None:
if self._printer.bambu_client.connected:
if self._printer.bambu_client.publish(pybambu.commands.RESUME):
self._log.info("print resumed")
else:
self._log.info("print resume failed")
return True
def _printing_worker(self, from_printer: bool = False):
try:
if not from_printer and self._printer.bambu_client.connected:
selected_file = self._printer.file_system.selected_file
print_command = {
"print": {
"sequence_id": 0,
"command": "project_file",
"param": "Metadata/plate_1.gcode",
"md5": "",
"profile_id": "0",
"project_id": "0",
"subtask_id": "0",
"task_id": "0",
"subtask_name": f"{selected_file}",
"file": f"{selected_file}",
"url": (
f"file:///mnt/sdcard/{selected_file}"
if self._printer._settings.get_boolean(["device_type"])
in ["X1", "X1C"]
else f"file:///sdcard/{selected_file}"
),
"timelapse": self._printer._settings.get_boolean(["timelapse"]),
"bed_leveling": self._printer._settings.get_boolean(["bed_leveling"]),
"flow_cali": self._printer._settings.get_boolean(["flow_cali"]),
"vibration_cali": self._printer._settings.get_boolean(
["vibration_cali"]
),
"layer_inspect": self._printer._settings.get_boolean(["layer_inspect"]),
"use_ams": self._printer._settings.get_boolean(["use_ams"]),
}
}
self._printer.bambu_client.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._log.debug(f"SD File Print: {self._selectedSdFile}")
except AttributeError:
if self.outgoing is not None:
raise
self._printer.change_state(self._printer._state_finished)
def cancel(self):
if self._printer.bambu_client.connected:
if self._printer.bambu_client.publish(pybambu.commands.STOP):
self._log.info("print cancelled")
self._printer.change_state(self._printer._state_finished)
return True
else:
self._log.info("print cancel failed")
return False
return False