bambuprinter
Classes:
| Name | Description |
|---|---|
BambuPrinter |
|
BambuPrinter
BambuPrinter(config: BambuConfig | None = None)
BambuPrinter is the main class within bambu-printer-manager
It is responsible for interacting with and managing your Bambu Lab 3d printer.
It provides an object oriented abstraction layer between your project and the
mqtt and ftps based mechanisms in place for communicating with your printer.
Parameters
- config : Optional[BambuConfig] = None
An optional
BambuConfiginstance that provides configuration information for the printer connection. If not provided, a defaultBambuConfiginstance will be created.
Methods:
| Name | Description |
|---|---|
delete_sdcard_file |
Delete the specified file on the printer's SDCard and returns an updated dict of all files on the printer |
delete_sdcard_folder |
Delete the specified folder on the printer's SDCard and returns an updated dict of all files on the printer |
download_sdcard_file |
Downloads a file from the printer |
ftp_connection |
Opens a connection to the printer's FTP server to support file management. |
get_current_bind_list |
Generates the manual_ams_bind list based on current AMS toolhead assignments. |
get_sdcard_3mf_files |
Returns a |
get_sdcard_contents |
Returns a |
jsonSerializer |
Helper method used by |
load_filament |
Requests the printer to load filament into the extruder using the requested spool (slot #) |
make_sdcard_directory |
Creates the specified directory on the printer and returns an updated dict of all files on the printer |
pause_printing |
Pauses the current print job if one is running. |
pause_session |
Pauses the |
print_3mf_file |
Submits a request to execute the |
quit |
Shuts down all threads. Your |
refresh |
Triggers a full data refresh from the printer (if it is connected). You should use this |
refresh_spool_rfid |
read the rfid tag for the selected ams and spool (slot) id. |
rename_sdcard_file |
Renames the specified file on the printer and returns an updated dict of all files on the printer |
resume_printing |
Resumes the current print job if one is paused. |
resume_session |
Resumes a previously paused session by re-subscribing to the /report topic. |
sdcard_file_exists |
Checks to see if a file exists on the printer at the |
select_extrusion_calibration_profile |
Sets the k factor profile for the specified tray. |
send_ams_control_command |
Send an AMS Control Command - will pause, resume, or reset the AMS. |
send_anything |
puts an arbritary string on the request topic |
send_gcode |
Submit one, or more, gcode commands to the printer. To submit multiple gcode commands, separate them with a newline (\n) character. |
set_active_tool |
sets the current active tool / extruder for machines (H2 series) |
set_ams_user_setting |
Enable or disable one of the |
set_aux_fan_speed_target_percent |
sets the aux (chamber recirculation) fan speed target represented in percent |
set_bed_temp_target |
Sets the bed temperature target. |
set_buildplate_marker_detector |
Enables or disables the buildplate_marker_detector |
set_chamber_temp |
for printers that do not have managed chambers, this enables you to inject |
set_chamber_temp_target |
set chamber temperature target if printer supports it, otherwise just |
set_exhaust_fan_speed_target_percent |
sets the exhaust (chamber) fan speed target represented in percent |
set_nozzle_details |
Sets the nozzle details. |
set_nozzle_temp_target |
Sets the nozzle temperature target. |
set_part_cooling_fan_speed_target_percent |
sets the part cooling fan speed target represented in percent |
set_print_option |
Enable or disable one of the |
set_spool_details |
Sets spool / tray details such as filament type, color, and nozzle min/max temperature. |
set_spool_k_factor |
Sets the linear advance k factor for a specific spool / tray |
skip_objects |
skip a list of objects extracted from the 3mf's plate_x.json file |
start_session |
Initiates a connection to the Bambu Lab printer and provides a stateful |
stop_printing |
Requests the printer to stop printing if a job is currently running. |
toJson |
Returns a |
turn_off_ams_dryer |
Sends a command to the printer to turn off the AMS dryer. |
turn_on_ams_dryer |
Sends a command to the printer to turn on the AMS dryer with specified parameters. |
unload_filament |
Requests the printer to unload whatever filament / spool may be currently loaded. |
upload_sdcard_file |
Uploads the local filesystem file to the printer and returns an updated dict of all files on the printer |
Attributes:
| Name | Type | Description |
|---|---|---|
bed_temp_target_time |
Timestamp of the last change to the heatbed target temperature. |
|
cached_sd_card_3mf_files |
A list of only the 3mf files found on the SD card. |
|
cached_sd_card_contents |
A list of all files and folders found on the printer's SD card. |
|
chamber_temp_target_time |
Timestamp of the last change to the chamber target temperature. |
|
client |
Client
|
The networking client used to communicate with the printer. |
config |
BambuConfig
|
The settings used to connect to and configure the printer's behavior. |
current_3mf_file |
The path to the 3mf file currently being printed. |
|
current_3mf_file_md5 |
The checksum used to verify the current project file. |
|
current_plate_num |
The plate number selected for the current job. |
|
current_plate_type |
The type of build plate required for the current print. |
|
fan_speed_target_time |
Timestamp of the last change to the part fan target speed. |
|
gcode_file |
The name of the G-code file currently in use. |
|
internalException |
The last error captured during printer communication. |
|
light_state |
The status of the printer lights. Toggling this will update all lights on the machine. |
|
nozzle_diameter |
NozzleDiameter
|
The diameter of the nozzle currently installed on the printer. |
nozzle_type |
NozzleType
|
The type of nozzle currently installed on the printer. |
on_update |
The callback function executed whenever the printer's state is updated. |
|
print_type |
Indicates how the current print job was initiated. |
|
printer_state |
BambuState
|
The current status and sensor data for the printer. |
recent_update |
Indicates if the printer's state has been updated recently. |
|
service_state |
The current service connection state. |
|
skipped_objects |
A list of objects that have been excluded from the current print. |
|
speed_level |
The active print speed mode. Updating this will change the printer's execution speed. |
|
subtask_name |
The name of the current printing task. |
|
tool_temp_target_time |
Timestamp of the last change to the nozzle target temperature. |
Source code in src/bpm/bambuprinter.py
def __init__(self, config: BambuConfig | None = None):
"""
Sets up all internal storage attributes for `BambuPrinter` and bootstraps the
logging engine.
Parameters
----------
* config : Optional[BambuConfig] = None
An optional `BambuConfig` instance that provides configuration information
for the printer connection. If not provided, a default `BambuConfig` instance
will be created.
"""
self._mqtt_client_thread = None
self._watchdog_thread = None
self._internalException = None
self._lastMessageTime = None
self._recent_update = False
if config is None:
config = BambuConfig("", "", "")
self._config = config
self._service_state = ServiceState.NO_STATE
self._client = None
self._on_update = None
self._tool_temp_target_time = 0
self._bed_temp_target_time = 0
self._chamber_temp_target_time = 0
self._fan_speed_target_time = 0
self._light_state = ""
self._speed_level = 0
self._gcode_file = ""
self._3mf_file = ""
self._3mf_file_md5 = ""
self._plate_num = 0
self._plate_type = PlateType.NONE
self._subtask_name = ""
self._print_type = ""
self._printer_state = BambuState()
self._sdcard_contents = None
self._sdcard_3mf_files = None
self._print_type = ""
self._skipped_objects = []
self._nozzle_type = ""
self._nozzle_diameter = 0.0
bed_temp_target_time
property
Timestamp of the last change to the heatbed target temperature.
cached_sd_card_3mf_files
property
A list of only the 3mf files found on the SD card.
cached_sd_card_contents
property
A list of all files and folders found on the printer's SD card.
chamber_temp_target_time
property
Timestamp of the last change to the chamber target temperature.
client
property
writable
The networking client used to communicate with the printer.
This is a private property
config
property
writable
config: BambuConfig
The settings used to connect to and configure the printer's behavior.
current_3mf_file
property
The path to the 3mf file currently being printed.
Deprecated
This property is deprecated (v1.0.0). No replacement yet.
current_3mf_file_md5
property
The checksum used to verify the current project file.
Deprecated
This property is deprecated (v1.0.0). No replacement yet.
current_plate_num
property
The plate number selected for the current job.
Deprecated
This property is deprecated (v1.0.0). No replacement yet.
current_plate_type
property
The type of build plate required for the current print.
Deprecated
This property is deprecated (v1.0.0). No replacement yet.
fan_speed_target_time
property
Timestamp of the last change to the part fan target speed.
gcode_file
property
writable
The name of the G-code file currently in use.
Deprecated
This property is deprecated (v1.0.0). No replacement yet.
internalException
property
The last error captured during printer communication.
light_state
property
writable
The status of the printer lights. Toggling this will update all lights on the machine.
nozzle_diameter
property
nozzle_diameter: NozzleDiameter
The diameter of the nozzle currently installed on the printer.
Deprecated
This property is deprecated (v1.0.0). No replacement yet.
nozzle_type
property
nozzle_type: NozzleType
The type of nozzle currently installed on the printer.
Deprecated
This property is deprecated (v1.0.0). No replacement yet.
on_update
property
writable
The callback function executed whenever the printer's state is updated.
print_type
property
Indicates how the current print job was initiated.
Deprecated
This property is deprecated (v1.0.0). No replacement yet.
printer_state
property
printer_state: BambuState
The current status and sensor data for the printer.
recent_update
property
Indicates if the printer's state has been updated recently.
skipped_objects
property
A list of objects that have been excluded from the current print.
speed_level
property
writable
The active print speed mode. Updating this will change the printer's execution speed.
subtask_name
property
The name of the current printing task.
Deprecated
This property is deprecated No replacement yet.
tool_temp_target_time
property
Timestamp of the last change to the nozzle target temperature.
delete_sdcard_file
Delete the specified file on the printer's SDCard and returns an updated dict of all files on the printer
Parameters
- file : str - the full path filename to be deleted
Source code in src/bpm/bambuprinter.py
def delete_sdcard_file(self, file: str):
"""
Delete the specified file on the printer's SDCard and returns an updated dict of all files on the printer
Parameters
----------
* file : str - the full path filename to be deleted
"""
logger.debug(f"deleting remote file: [{file}]")
with self.ftp_connection() as ftps:
ftps.delete_file(file)
def search_for_and_remove_file(file: str, entry: dict):
if "children" in entry:
entry["children"] = list(
filter(lambda i: i["id"] != file, entry["children"])
)
for child in entry["children"]:
search_for_and_remove_file(file, child)
if self._sdcard_contents:
search_for_and_remove_file(file, self._sdcard_contents)
if self._sdcard_3mf_files:
search_for_and_remove_file(file, self._sdcard_3mf_files)
logger.debug(f"delete_sdcard_file - deleted file [{file}] from sdcard")
return self._sdcard_contents
delete_sdcard_folder
Delete the specified folder on the printer's SDCard and returns an updated dict of all files on the printer
Parameters
- path : str - the full path to folder to be deleted
Source code in src/bpm/bambuprinter.py
def delete_sdcard_folder(self, path: str):
"""
Delete the specified folder on the printer's SDCard and returns an updated dict of all files on the printer
Parameters
----------
* path : str - the full path to folder to be deleted
"""
logger.debug(f"delete_sdcard_folder - deleting remote folder: [{path}]")
def delete_all_contents(ftps: IoTFTPSClient, path: str):
fs = ftps.list_files_ex(path)
if fs is not None:
for item in fs:
if item.is_dir:
delete_all_contents(ftps, item.path)
else:
ftps.delete_file(item.path)
ftps.delete_folder(path)
with self.ftp_connection() as ftps:
delete_all_contents(ftps, path)
def search_for_and_remove_folder(path: str, entry: dict):
if not path.endswith("/"):
path = f"{path}/"
if "children" in entry:
entry["children"] = list(
filter(lambda i: not i["id"].startswith(path), entry["children"])
)
for child in entry["children"]:
search_for_and_remove_folder(path, child)
if self._sdcard_contents is not None:
search_for_and_remove_folder(path, self._sdcard_contents)
if self._sdcard_3mf_files is not None:
search_for_and_remove_folder(path, self._sdcard_3mf_files)
return self._sdcard_contents
download_sdcard_file
Downloads a file from the printer
Parameters
- src : str - the full path filename on the printer to be downloaded to the host
- dest : str - the full path filename on the host to store the downloaded file
Source code in src/bpm/bambuprinter.py
def download_sdcard_file(self, src: str, dest: str):
"""
Downloads a file from the printer
Parameters
----------
* src : str - the full path filename on the printer to be downloaded to the host
* dest : str - the full path filename on the host to store the downloaded file
"""
logger.debug(
f"download_sdcard_file - downloading file src: [{src}] dest: [{dest}]"
)
with self.ftp_connection() as ftps:
ftps.download_file(src, dest)
ftp_connection
Opens a connection to the printer's FTP server to support file management.
Acts as a context manager, so can be used with the with statement. In that case, the connection will be closed automatically, otherwise the calling code has to make sure to close it again.
This is an internal method.
Source code in src/bpm/bambuprinter.py
@contextlib.contextmanager
def ftp_connection(self):
"""
Opens a connection to the printer's FTP server to support file management.
Acts as a context manager, so can be used with the with statement. In that case,
the connection will be closed automatically, otherwise the calling code has to make
sure to close it again.
This is an internal method.
"""
ftps = None
try:
ftps = IoTFTPSClient(
self._config.hostname,
990,
self._config.mqtt_username,
self._config.access_code,
ssl_implicit=True,
)
yield ftps
finally:
if ftps and ftps.ftps_session:
ftps.disconnect()
get_current_bind_list
get_current_bind_list(state: BambuState) -> list[dict[str, Any]]
Generates the manual_ams_bind list based on current AMS toolhead assignments.
This method implements the H2D cross-over hardware targeting: - RIGHT_EXTRUDER (0) -> Hardware Index 1 - LEFT_EXTRUDER (1) -> Hardware Index 0
Includes sentinel placeholder logic for single-AMS configurations to satisfy dual-extruder firmware array requirements.
Source code in src/bpm/bambuprinter.py
def get_current_bind_list(self, state: "BambuState") -> list[dict[str, Any]]:
"""
Generates the manual_ams_bind list based on current AMS toolhead assignments.
This method implements the H2D cross-over hardware targeting:
- RIGHT_EXTRUDER (0) -> Hardware Index 1
- LEFT_EXTRUDER (1) -> Hardware Index 0
Includes sentinel placeholder logic for single-AMS configurations to satisfy
dual-extruder firmware array requirements.
"""
bind_list = []
# 1. Map physical units based on current logical assignments
for unit in state.ams_units:
# Hardware register inversion logic
hw_target = 1 if unit.assigned_to_extruder == ActiveTool.RIGHT_EXTRUDER else 0
bind_list.append(
{"ams_f_bind": int(unit.ams_id), "ams_s_bind": 0, "extruder": hw_target}
)
# 2. Dual-Extruder Array Completeness Requirement
# If only one AMS is connected, firmware requires a placeholder for the unused register.
if state.ams_connected_count == 1 and len(bind_list) == 1:
existing_assignment = bind_list[0]
# Determine the unassigned hardware register
placeholder_hw_target = 0 if existing_assignment["extruder"] == 1 else 1
# Add sentinel placeholder (Unit ID 1) to complete the dual-path mapping
bind_list.append(
{
"ams_f_bind": 1, # Placeholder Unit ID
"ams_s_bind": 0,
"extruder": placeholder_hw_target,
}
)
return bind_list
get_sdcard_3mf_files
Returns a dict (json document) of all .3mf files on the printer's SD card.
Usage
The return value of this method is very useful for binding to things like a clientside TreeView
Source code in src/bpm/bambuprinter.py
def get_sdcard_3mf_files(self):
"""
Returns a `dict` (json document) of all `.3mf` files on the printer's SD card.
Usage
-----
The return value of this method is very useful for binding to things like a clientside `TreeView`
"""
logger.debug("get_sdcard_3mf_files - returning sdcard_3mf_files")
self.get_sdcard_contents()
return self._sdcard_3mf_files
get_sdcard_contents
Returns a dict (json document) of ALL files on the printer's SD card.
The private class level _sdcard_contents attribute is also populated.
Usage
The return value of this method is very useful for binding to things like a clientside TreeView
Source code in src/bpm/bambuprinter.py
def get_sdcard_contents(self):
"""
Returns a `dict` (json document) of ALL files on the printer's SD card.
The private class level `_sdcard_contents` attribute is also populated.
Usage
-----
The return value of this method is very useful for binding to things like a clientside `TreeView`
"""
with self.ftp_connection() as ftps:
fs = self._get_sftp_files(ftps, "/")
if fs is None:
logger.warning("get_sdcard_contents - failed to retrieve files from sdcard")
self._sdcard_contents = None
self._sdcard_3mf_files = None
return None
self._sdcard_contents = sortFileTreeAlphabetically(fs)
def search_for_and_remove_all_other_files(mask: str, entry: dict):
if "children" in entry:
entry["children"] = [
x
for x in entry["children"]
if x["id"].endswith(mask) or "children" in x
]
for child in entry["children"]:
search_for_and_remove_all_other_files(mask, child)
self._sdcard_3mf_files = json.loads(json.dumps(self._sdcard_contents))
search_for_and_remove_all_other_files(".3mf", self._sdcard_3mf_files)
logger.debug("get_sdcard_contents - retrieved all files from sdcard")
return fs
jsonSerializer
Helper method used by toJson() to serialize this object.
Source code in src/bpm/bambuprinter.py
def jsonSerializer(self, obj):
"""
Helper method used by `toJson()` to serialize this object.
"""
try:
if isinstance(obj, mqtt.Client) or isinstance(obj, Thread):
return "these are not the droids you are looking for"
if (
str(obj.__class__).replace("<class '", "").replace("'>", "")
== "mappingproxy"
):
return "this space intentionally left blank"
return obj.__dict__
except Exception:
logger.warning(
f"jsonSerializer - unable to serialize object - 'obj': [{obj}]"
)
return "not available"
load_filament
Requests the printer to load filament into the extruder using the requested spool (slot #)
Parameters
slot_id : int
0- AMS Spool #11- AMS Spool #22- AMS Spool #33- AMS Spool #4254- External Spool
Source code in src/bpm/bambuprinter.py
def load_filament(self, slot_id: int, ams_id: int = 0):
"""
Requests the printer to load filament into the extruder using the requested spool (slot #)
Parameters
----------
slot_id : int
* `0` - AMS Spool #1
* `1` - AMS Spool #2
* `2` - AMS Spool #3
* `3` - AMS Spool #4
* `254` - External Spool
"""
# TODO: refactor to support multiple AMSs
msg = copy.deepcopy(AMS_CHANGE_FILAMENT)
msg["print"]["ams_id"] = ams_id
msg["print"]["target"] = slot_id
msg["print"]["slot_id"] = slot_id
msg["print"]["soft_temp"] = 0
msg["print"]["tar_temp"] = -1
msg["print"]["curr_temp"] = -1
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(msg)
)
logger.debug(
f"load_filament - published AMS_CHANGE_FILAMENT to [device/{self.config.serial_number}/request] - target: [{slot_id}], bambu_msg: [{msg}]"
)
make_sdcard_directory
Creates the specified directory on the printer and returns an updated dict of all files on the printer
Parameters
- dir : str - the full path directory name to be created
Source code in src/bpm/bambuprinter.py
def make_sdcard_directory(self, dir: str):
"""
Creates the specified directory on the printer and returns an updated dict of all files on the printer
Parameters
----------
* dir : str - the full path directory name to be created
"""
logger.debug(f"make_sdcard_directory - creating remote directory [{dir}]")
with self.ftp_connection() as ftps:
ftps.mkdir(dir)
return self.get_sdcard_contents()
pause_printing
Pauses the current print job if one is running.
Source code in src/bpm/bambuprinter.py
pause_session
Pauses the BambuPrinter session is it is active. Under the covers this
method unsubscribes from the /report topic, essentially disabling all
printer data refreshes.
Source code in src/bpm/bambuprinter.py
def pause_session(self):
"""
Pauses the `BambuPrinter` session is it is active. Under the covers this
method unsubscribes from the `/report` topic, essentially disabling all
printer data refreshes.
"""
if self.service_state != ServiceState.PAUSED:
self.client.unsubscribe(f"device/{self.config.serial_number}/report")
logger.debug(
f"pause_session - unsubscribed from [device/{self.config.serial_number}/report]"
)
self.service_state = ServiceState.PAUSED
print_3mf_file
print_3mf_file(
name: str,
plate: int,
bed: PlateType,
use_ams: bool,
ams_mapping: str | None = "",
bedlevel: bool | None = True,
flow: bool | None = True,
timelapse: bool | None = False,
)
Submits a request to execute the name 3mf file on the printer's SDCard.
Parameters
- name : str - path, filename, and extension to execute
- plate : int - the plate # from your slicer to use (usually 1)
- bed : PlateType - the bambutools.PlateType to use
- use_ams : bool - Use the AMS for this print
- ams_mapping : Optional[str] - an
AMS Mappingthat specifies which AMS spools to use (external spool is used if blank) - bedlevel : Optional[bool] = True - boolean value indicates whether or not the printer should auto-level the bed
- flow : Optional[bool] = True - boolean value indicates if the printer should perform an extrusion flow calibration
- timelapse : Optional[bool] = False - boolean value indicates if printer should take timelapse photos during the job
Example
print_3mf_file("/jobs/my_project.3mf", 1, PlateType.HOT_PLATE, False, "")- Print the my_project.3mf file in the SDCard /jobs directory using the external spool with bed leveling and extrusion flow calibration enabled and timelapse disabledprint_3mf_file("/jobs/my_project.gcode.3mf", 1, PlateType.HOT_PLATE, True, "[-1,-1,2,-1]")- Same as above but use AMS spool #3
AMS Mapping
[0,-1,-1,-1]- use AMS spool #1 only[-1,1,-1,-1]- use AMS spool #2 only[0,-1,-1,3]- use AMS spools #1 and #4[0,1,2,3]- use all 4 AMS spools
Source code in src/bpm/bambuprinter.py
def print_3mf_file(
self,
name: str,
plate: int,
bed: PlateType,
use_ams: bool,
ams_mapping: str | None = "",
bedlevel: bool | None = True,
flow: bool | None = True,
timelapse: bool | None = False,
):
"""
Submits a request to execute the `name` 3mf file on the printer's SDCard.
Parameters
----------
* name : str - path, filename, and extension to execute
* plate : int - the plate # from your slicer to use (usually 1)
* bed : PlateType - the bambutools.PlateType to use
* use_ams : bool - Use the AMS for this print
* ams_mapping : Optional[str] - an `AMS Mapping` that specifies which AMS spools to use (external spool is used if blank)
* bedlevel : Optional[bool] = True - boolean value indicates whether or not the printer should auto-level the bed
* flow : Optional[bool] = True - boolean value indicates if the printer should perform an extrusion flow calibration
* timelapse : Optional[bool] = False - boolean value indicates if printer should take timelapse photos during the job
Example
-------
* `print_3mf_file("/jobs/my_project.3mf", 1, PlateType.HOT_PLATE, False, "")` - Print the my_project.3mf file in the SDCard /jobs directory using the external spool with bed leveling and extrusion flow calibration enabled and timelapse disabled
* `print_3mf_file("/jobs/my_project.gcode.3mf", 1, PlateType.HOT_PLATE, True, "[-1,-1,2,-1]")` - Same as above but use AMS spool #3
AMS Mapping
-----------
* `[0,-1,-1,-1]` - use AMS spool #1 only
* `[-1,1,-1,-1]` - use AMS spool #2 only
* `[0,-1,-1,3]` - use AMS spools #1 and #4
* `[0,1,2,3]` - use all 4 AMS spools
"""
self._3mf_file = f"{name}"
self._plate_num = int(plate)
self._plate_type = bed
file = copy.deepcopy(PRINT_3MF_FILE)
subtask = name[name.rindex("/") + 1 : :] if "/" in name else name
subtask = (
subtask[::-1].replace(".3mf"[::-1], "", 1)[::-1]
if subtask.endswith(".3mf")
else subtask
)
subtask = (
subtask[::-1].replace(".gcode"[::-1], "", 1)[::-1]
if subtask.endswith(".gcode")
else subtask
)
file["print"]["file"] = self._3mf_file
file["print"]["url"] = f"file:///sdcard{self._3mf_file}"
file["print"]["subtask_name"] = subtask
file["print"]["bed_type"] = bed.name.lower()
file["print"]["param"] = file["print"]["param"].replace("#", str(self._plate_num))
file["print"]["use_ams"] = use_ams
if ams_mapping and len(ams_mapping) > 0:
file["print"]["ams_mapping"] = json.loads(ams_mapping)
file["print"]["bed_leveling"] = bedlevel
file["print"]["flow_cali"] = flow
file["print"]["timelapse"] = timelapse
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(file)
)
logger.debug(
f"print_3mf_file - published PRINT_3MF_FILE to [device/{self.config.serial_number}/request] print_command: [{file}]"
)
quit
Shuts down all threads. Your BambuPrinter instance should probably be
considered dead after making this call although you may be able to restart a
session with start_session().
Source code in src/bpm/bambuprinter.py
def quit(self):
"""
Shuts down all threads. Your `BambuPrinter` instance should probably be
considered dead after making this call although you may be able to restart a
session with [start_session](#bpm.bambuprinter.BambuPrinter.start_session)().
"""
if self.client and self.client.is_connected():
self.client.disconnect()
logger.debug("quit - mqtt client was connected and is now disconnected")
else:
logger.debug("quit - mqtt client was already disconnected")
self._service_state = ServiceState.QUIT
self._notify_update()
if self._mqtt_client_thread and self._mqtt_client_thread.is_alive():
self._mqtt_client_thread.join()
if self._watchdog_thread and self._watchdog_thread.is_alive():
self._watchdog_thread.join()
logger.debug("quit - all threads have terminated")
refresh
Triggers a full data refresh from the printer (if it is connected). You should use this method sparingly as resorting to it indicates something is not working properly.
Source code in src/bpm/bambuprinter.py
def refresh(self):
"""
Triggers a full data refresh from the printer (if it is connected). You should use this
method sparingly as resorting to it indicates something is not working properly.
"""
if self.service_state == ServiceState.CONNECTED:
logger.debug(
f"refresh - publishing ANNOUNCE_VERSION to [device/{self.config.serial_number}/request]"
)
self.client.publish(
f"device/{self.config.serial_number}/request",
json.dumps(ANNOUNCE_VERSION),
)
logger.debug(
f"refresh - publishing ANNOUNCE_PUSH to [device/{self.config.serial_number}/request]"
)
self.client.publish(
f"device/{self.config.serial_number}/request",
json.dumps(ANNOUNCE_PUSH),
)
refresh_spool_rfid
read the rfid tag for the selected ams and spool (slot) id.
Source code in src/bpm/bambuprinter.py
def refresh_spool_rfid(self, slot_id: int, ams_id: int = 0):
"""
read the rfid tag for the selected ams and spool (slot) id.
"""
cmd = copy.deepcopy(AMS_GET_RFID)
cmd["print"]["ams_id"] = ams_id
cmd["print"]["slot_id"] = slot_id
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"refresh_spool_rfid - published AMS_GET_RFID to [device/{self.config.serial_number}/request] command: [{cmd}]"
)
rename_sdcard_file
Renames the specified file on the printer and returns an updated dict of all files on the printer
Parameters
- src : str - the full path name to be renamed
- dest : str - the full path name to be renamed to
Source code in src/bpm/bambuprinter.py
def rename_sdcard_file(self, src: str, dest: str):
"""
Renames the specified file on the printer and returns an updated dict of all files on the printer
Parameters
----------
* src : str - the full path name to be renamed
* dest : str - the full path name to be renamed to
"""
logger.debug(f"rename_sdcard_file - renaming printer file [{src}] to [{dest}]")
with self.ftp_connection() as ftps:
ftps.move_file(src, dest)
return self.get_sdcard_contents()
resume_printing
Resumes the current print job if one is paused.
Source code in src/bpm/bambuprinter.py
resume_session
Resumes a previously paused session by re-subscribing to the /report topic.
Source code in src/bpm/bambuprinter.py
def resume_session(self):
"""
Resumes a previously paused session by re-subscribing to the /report topic.
"""
if (
self.client
and self.client.is_connected()
and self.service_state == ServiceState.PAUSED
):
self.client.subscribe(f"device/{self.config.serial_number}/report")
logger.debug(
f"resume_session - subscribed to [device/{self.config.serial_number}/report]"
)
self._lastMessageTime = time.monotonic()
self.service_state = ServiceState.CONNECTED
return
self.service_state = ServiceState.QUIT
sdcard_file_exists
Checks to see if a file exists on the printer at the path specified
Parameters
- path : str - the full path name to check
Source code in src/bpm/bambuprinter.py
def sdcard_file_exists(self, path: str) -> bool:
"""
Checks to see if a file exists on the printer at the `path` specified
Parameters
----------
* path : str - the full path name to check
"""
logger.debug(f"sdcard_file_exists - checking if printer file [{path}] exists")
with self.ftp_connection() as ftps:
return ftps.fexists(path)
select_extrusion_calibration_profile
Sets the k factor profile for the specified tray.
Parameters
tray_id : int - tray id cali_idx : calibration index , optional, defaults to -1 (the default profile)
Source code in src/bpm/bambuprinter.py
def select_extrusion_calibration_profile(self, tray_id: int, cali_idx: int = -1):
"""
Sets the k factor profile for the specified tray.
Parameters
----------
tray_id : int - tray id
cali_idx : calibration index , optional, defaults to -1 (the default profile)
"""
cmd = copy.deepcopy(EXTRUSION_CALI_SEL)
ams_id = math.floor(tray_id / 4)
if tray_id in (254, 255):
ams_id = tray_id
tray_id = 0
cmd["print"]["ams_id"] = ams_id
cmd["print"]["tray_id"] = tray_id
cmd["print"]["slot_id"] = tray_id % 4
cmd["print"]["cali_idx"] = cali_idx
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"select_extrusion_calibration_profile - published EXTRUSION_CALI_SEL to [device/{self.config.serial_number}/request] cmd: [{cmd}]"
)
send_ams_control_command
send_ams_control_command(ams_control_cmd: AMSControlCommand)
Send an AMS Control Command - will pause, resume, or reset the AMS.
Source code in src/bpm/bambuprinter.py
def send_ams_control_command(self, ams_control_cmd: AMSControlCommand):
"""
Send an AMS Control Command - will pause, resume, or reset the AMS.
"""
ams_cmd = ams_control_cmd.name.lower()
cmd = copy.deepcopy(AMS_CONTROL)
cmd["print"]["param"] = ams_cmd
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"send_ams_control_commandpublished AMS_CONTROL to [device/{self.config.serial_number}/request] bambu_msg: [{cmd}]"
)
# trigger resume print for good measure
if ams_control_cmd == AMSControlCommand.RESUME:
self.resume_printing()
send_anything
puts an arbritary string on the request topic
Source code in src/bpm/bambuprinter.py
def send_anything(self, anything: str):
"""
puts an arbritary string on the request topic
"""
self.client.publish(
f"device/{self.config.serial_number}/request",
json.dumps(json.loads(anything)),
)
logger.debug(
f"send_anything - published message to [device/{self.config.serial_number}/request] message: [{anything}]"
)
send_gcode
Submit one, or more, gcode commands to the printer. To submit multiple gcode commands, separate them with a newline (\n) character.
Parameters
gcode : str
Examples
send_gcode("G91\nG0 X0\nG0 X50")- queues 3 gcode commands on the printer for processingsend_gcode("G28")- queues 1 gcode command on the printer for processing
Source code in src/bpm/bambuprinter.py
def send_gcode(self, gcode: str):
"""
Submit one, or more, gcode commands to the printer. To submit multiple gcode commands, separate them with a newline (\\n) character.
Parameters
----------
gcode : str
Examples
--------
* `send_gcode("G91\\nG0 X0\\nG0 X50")` - queues 3 gcode commands on the printer for processing
* `send_gcode("G28")` - queues 1 gcode command on the printer for processing
"""
cmd = copy.deepcopy(SEND_GCODE_TEMPLATE)
cmd["print"]["param"] = f"{gcode} \n"
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"send_gcode - published SEND_GCODE_TEMPLATE to [device/{self.config.serial_number}/request] gcode: [{gcode}]"
)
set_active_tool
sets the current active tool / extruder for machines (H2 series) that have multiple extruders
Source code in src/bpm/bambuprinter.py
def set_active_tool(self, id: int):
"""
sets the current active tool / extruder for machines (H2 series)
that have multiple extruders
"""
cmd = copy.deepcopy(SET_ACTIVE_TOOL)
cmd["print"]["extruder_index"] = id
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"set_active_tool - published SET_ACTIVE_TOOL to [device/{self.config.serial_number}/request] command: [{cmd}]"
)
set_ams_user_setting
set_ams_user_setting(setting: AMSUserSetting, enabled: bool, ams_id: int = 0)
Enable or disable one of the AMSUserSetting options
Source code in src/bpm/bambuprinter.py
def set_ams_user_setting(
self, setting: AMSUserSetting, enabled: bool, ams_id: int = 0
):
"""
Enable or disable one of the `AMSUserSetting` options
"""
cmd = copy.deepcopy(AMS_USER_SETTING)
cmd["print"]["ams_id"] = ams_id
cmd["print"][AMSUserSetting.CALIBRATE_REMAIN_FLAG.name.lower()] = (
self.config.calibrate_remain_flag
)
cmd["print"][AMSUserSetting.STARTUP_READ_OPTION.name.lower()] = (
self.config.startup_read_option
)
cmd["print"][AMSUserSetting.TRAY_READ_OPTION.name.lower()] = (
self.config.tray_read_option
)
cmd["print"][setting.name.lower()] = enabled
if setting == AMSUserSetting.STARTUP_READ_OPTION:
self.config.startup_read_option = enabled
elif setting == AMSUserSetting.TRAY_READ_OPTION:
self.config.tray_read_option = enabled
elif setting == AMSUserSetting.CALIBRATE_REMAIN_FLAG:
self.config.calibrate_remain_flag = enabled
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"set_ams_user_setting - published AMS_USER_SETTING to [device/{self.config.serial_number}/request] bambu_msg: [{cmd}]"
)
set_aux_fan_speed_target_percent
sets the aux (chamber recirculation) fan speed target represented in percent
Parameters
- value : int - The target speed in percent
Source code in src/bpm/bambuprinter.py
def set_aux_fan_speed_target_percent(self, value: int):
"""
sets the aux (chamber recirculation) fan speed target represented in percent
Parameters
----------
* value : int - The target speed in percent
"""
if value < 0:
value = 0
self._printer_state.climate.zone_aux_percent = value
speed = round(value * 2.55, 0)
gcode = SEND_GCODE_TEMPLATE
gcode["print"]["param"] = f"M106 P2 S{speed}\n"
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(gcode)
)
logger.debug(
f"set_aux_fan_speed_target_percent - published SEND_GCODE_TEMPLATE to [device/{self.config.serial_number}/request] command: [{gcode}]"
)
set_bed_temp_target
Sets the bed temperature target.
Parameters
- value : float - The target bed temperature.
Source code in src/bpm/bambuprinter.py
def set_bed_temp_target(self, value: int):
"""
Sets the bed temperature target.
Parameters
----------
* value : float - The target bed temperature.
"""
if value < 0:
value = 0
gcode = SEND_GCODE_TEMPLATE
gcode["print"]["param"] = f"M140 S{value}\n"
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(gcode)
)
logger.debug(
f"set_bed_temp_target - published SEND_GCODE_TEMPLATE to [device/{self.config.serial_number}/request] command: [{gcode}]"
)
self._bed_temp_target_time = round(time.time())
set_buildplate_marker_detector
Enables or disables the buildplate_marker_detector
Source code in src/bpm/bambuprinter.py
def set_buildplate_marker_detector(self, enabled: bool):
"""
Enables or disables the buildplate_marker_detector
"""
cmd = copy.deepcopy(XCAM_CONTROL_SET)
cmd["xcam"]["control"] = enabled
cmd["xcam"]["enable"] = enabled
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"set_buildplate_marker_detector - published XCAM_CONTROL_SET to [device/{self.config.serial_number}/request] bambu_msg: [{cmd}]"
)
set_chamber_temp
for printers that do not have managed chambers, this enables you to inject a chamber temperature value from an external source
Parameters
- value : float - The chamber temperature.
Source code in src/bpm/bambuprinter.py
set_chamber_temp_target
set chamber temperature target if printer supports it, otherwise just store the value
Parameters
- value : float - The target chamber temperature.
- temper_check : OPTIONAL bool - perform a temperature check?
Source code in src/bpm/bambuprinter.py
def set_chamber_temp_target(self, value: int, temper_check: bool = True):
"""
set chamber temperature target if printer supports it, otherwise just
store the value
Parameters
----------
* value : float - The target chamber temperature.
* temper_check : OPTIONAL bool - perform a temperature check?
"""
if self.config.capabilities.has_chamber_temp:
cmd = copy.deepcopy(SET_CHAMBER_TEMP_TARGET)
cmd["print"]["ctt_val"] = value
cmd["print"]["temper_check"] = temper_check
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"set_chamber_temp_target - published SET_CHAMBER_TEMP_TARGET to [device/{self.config.serial_number}/request] command: [{cmd}]"
)
cmd = copy.deepcopy(SET_CHAMBER_AC_MODE)
if value < 40:
cmd["print"]["modeId"] = 0
else:
cmd["print"]["modeId"] = 1
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"set_chamber_temp_target - published SET_CHAMBER_AC_MODE to [device/{self.config.serial_number}/request] command: [{cmd}]"
)
self._printer_state.climate.chamber_temp_target = value
self._chamber_temp_target_time = round(time.time())
set_exhaust_fan_speed_target_percent
sets the exhaust (chamber) fan speed target represented in percent
Parameters
- value : int - The target speed in percent
Source code in src/bpm/bambuprinter.py
def set_exhaust_fan_speed_target_percent(self, value: int):
"""
sets the exhaust (chamber) fan speed target represented in percent
Parameters
----------
* value : int - The target speed in percent
"""
if value < 0:
value = 0
self._printer_state.climate.zone_exhaust_percent = value
speed = round(value * 2.55, 0)
gcode = SEND_GCODE_TEMPLATE
gcode["print"]["param"] = f"M106 P3 S{speed}\n"
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(gcode)
)
logger.debug(
f"set_exhaust_fan_speed_target_percent - published SEND_GCODE_TEMPLATE to [device/{self.config.serial_number}/request] command: [{gcode}]"
)
set_nozzle_details
set_nozzle_details(nozzle_diameter: NozzleDiameter, nozzle_type: NozzleType)
Sets the nozzle details.
Source code in src/bpm/bambuprinter.py
def set_nozzle_details(
self, nozzle_diameter: NozzleDiameter, nozzle_type: NozzleType
):
"""
Sets the nozzle details.
"""
cmd = copy.deepcopy(SET_ACCESSORIES)
cmd["system"]["nozzle_diameter"] = nozzle_diameter.value
cmd["system"]["nozzle_type"] = nozzle_type.name.lower()
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"set_nozzle_details - published SET_ACCESSORIES to [device/{self.config.serial_number}/request] bambu_msg: [{cmd}]"
)
set_nozzle_temp_target
Sets the nozzle temperature target.
Parameters
- value : float - The target nozzle temperature.
- tool_num : int - The tool number (default is 0).
Source code in src/bpm/bambuprinter.py
def set_nozzle_temp_target(self, value: int, tool_num: int = -1):
"""
Sets the nozzle temperature target.
Parameters
----------
* value : float - The target nozzle temperature.
* tool_num : int - The tool number (default is 0).
"""
if value < 0:
value = 0
gcode = SEND_GCODE_TEMPLATE
gcode["print"]["param"] = (
f"M104 S{value}{'' if tool_num == -1 else ' T' + str(tool_num)}\n"
)
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(gcode)
)
logger.debug(
f"set_nozzle_temp_target - published SEND_GCODE_TEMPLATE to [device/{self.config.serial_number}/request] command: [{gcode}]"
)
self._tool_temp_target_time = round(time.time())
set_part_cooling_fan_speed_target_percent
sets the part cooling fan speed target represented in percent
Parameters
- value : int - The target speed in percent
Source code in src/bpm/bambuprinter.py
def set_part_cooling_fan_speed_target_percent(self, value: int):
"""
sets the part cooling fan speed target represented in percent
Parameters
----------
* value : int - The target speed in percent
"""
if value < 0:
value = 0
self._printer_state.climate.part_cooling_fan_speed_target_percent = value
speed = round(value * 2.55, 0)
gcode = SEND_GCODE_TEMPLATE
gcode["print"]["param"] = f"M106 P1 S{speed}\n"
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(gcode)
)
logger.debug(
f"set_part_cooling_fan_speed_target_percent - published SEND_GCODE_TEMPLATE to [device/{self.config.serial_number}/request] command: [{gcode}]"
)
self._fan_speed_target_time = round(time.time())
set_print_option
set_print_option(option: PrintOption, enabled: bool)
Enable or disable one of the PrintOption options
Source code in src/bpm/bambuprinter.py
def set_print_option(self, option: PrintOption, enabled: bool):
"""
Enable or disable one of the `PrintOption` options
"""
cmd = PRINT_OPTION_COMMAND
cmd["print"][option.name.lower()] = "true" if enabled else "false"
if option == PrintOption.AUTO_RECOVERY:
cmd["print"]["option"] = "1" if enabled else "0"
self.config.auto_recovery = enabled
elif option == PrintOption.AUTO_SWITCH_FILAMENT:
self.config.auto_switch_filament = enabled
elif option == PrintOption.FILAMENT_TANGLE_DETECT:
self.config.filament_tangle_detect = enabled
elif option == PrintOption.SOUND_ENABLE:
self.config.sound_enable = enabled
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"set_print_option - published PRINT_OPTION_COMMAND to [device/{self.config.serial_number}/request] bambu_msg: [{cmd}]"
)
set_spool_details
set_spool_details(
tray_id: int,
tray_info_idx: str,
tray_id_name: str | None = "",
tray_type: str | None = "",
tray_color: str | None = "",
nozzle_temp_min: int | None = -1,
nozzle_temp_max: int | None = -1,
ams_id: int | None = 0,
)
Sets spool / tray details such as filament type, color, and nozzle min/max temperature.
For the external tray (254), send no_filament as the tray_info_idx value to empty the tray.
Source code in src/bpm/bambuprinter.py
def set_spool_details(
self,
tray_id: int,
tray_info_idx: str,
tray_id_name: str | None = "",
tray_type: str | None = "",
tray_color: str | None = "",
nozzle_temp_min: int | None = -1,
nozzle_temp_max: int | None = -1,
ams_id: int | None = 0,
):
"""
Sets spool / tray details such as filament type, color, and nozzle min/max temperature.
For the external tray (254), send `no_filament` as the `tray_info_idx` value to empty the tray.
"""
cmd = copy.deepcopy(AMS_FILAMENT_SETTING)
ams_id = math.floor(tray_id / 4)
if tray_id == 254 or tray_id == 255:
ams_id = tray_id
tray_id = 0
cmd["print"]["ams_id"] = ams_id
cmd["print"]["tray_id"] = tray_id
cmd["print"]["slot_id"] = tray_id % 4
if tray_info_idx == "no_filament":
tray_info_idx = ""
tray_id_name = ""
tray_type = ""
tray_color = "FFFFFF00"
nozzle_temp_min = 0
nozzle_temp_max = 0
cmd["print"]["tray_info_idx"] = tray_info_idx
cmd["print"]["tray_id_name"] = tray_id_name
cmd["print"]["tray_type"] = tray_type
if tray_color and tray_color != "":
color = ""
try:
color = f"{name_to_hex(tray_color)}FF".replace("#", "").upper()
except Exception:
color = tray_color
cmd["print"]["tray_color"] = color
cmd["print"]["nozzle_temp_min"] = nozzle_temp_min
cmd["print"]["nozzle_temp_max"] = nozzle_temp_max
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"set_spool_details - published AMS_FILAMENT_SETTING to [device/{self.config.serial_number}/request] bambu_msg: [{cmd}]"
)
set_spool_k_factor
deprecated
set_spool_k_factor(
tray_id: int,
k_value: float,
n_coef: float | None = 1.399999976158142,
nozzle_temp: int | None = -1,
bed_temp: int | None = -1,
max_volumetric_speed: int | None = -1,
)
Deprecated
This method is deprecated. The closest alternative is select_extrusion_calibration_profile.
Sets the linear advance k factor for a specific spool / tray
broken in recent firmware -- will require implementing k factor list management
Source code in src/bpm/bambuprinter.py
@deprecated(
"This method is deprecated. The closest alternative is `select_extrusion_calibration_profile`."
)
def set_spool_k_factor(
self,
tray_id: int,
k_value: float,
n_coef: float | None = 1.399999976158142,
nozzle_temp: int | None = -1,
bed_temp: int | None = -1,
max_volumetric_speed: int | None = -1,
):
"""
Sets the linear advance k factor for a specific spool / tray
broken in recent firmware -- will require implementing
k factor list management
"""
cmd = copy.deepcopy(EXTRUSION_CALI_SET)
cmd["print"]["tray_id"] = tray_id
cmd["print"]["slot_id"] = tray_id % 4
cmd["print"]["k_value"] = k_value
cmd["print"]["n_coef"] = n_coef
if nozzle_temp != -1:
cmd["print"]["nozzle_temp"] = nozzle_temp
if bed_temp != -1:
cmd["print"]["bed_temp"] = bed_temp
if max_volumetric_speed != -1:
cmd["print"]["max_volumetric_speed"] = max_volumetric_speed
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"set_spool_k_factor - published EXTRUSION_CALI_SET to [device/{self.config.serial_number}/request] bambu_msg: [{cmd}]"
)
skip_objects
skip a list of objects extracted from the 3mf's plate_x.json file
Parameters
objects : list
Source code in src/bpm/bambuprinter.py
def skip_objects(self, objects):
"""
skip a list of objects extracted from the 3mf's plate_x.json file
Parameters
----------
objects : list
"""
objs = []
for obj in objects:
objs.append(int(obj))
cmd = copy.deepcopy(SKIP_OBJECTS)
cmd["print"]["obj_list"] = objs
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
logger.debug(
f"skip_objects - published SKIP_OBJECTS to [device/{self.config.serial_number}/request] bambu_msg: [{cmd}]"
)
start_session
Initiates a connection to the Bambu Lab printer and provides a stateful
session, with built-in recovery in the event BambuPrinter
becomes disconnected from the machine.
This method is required to be called before any commands or data collection / callbacks can take place with the machine.
Source code in src/bpm/bambuprinter.py
def start_session(self):
"""
Initiates a connection to the Bambu Lab printer and provides a stateful
session, with built-in recovery in the event `BambuPrinter`
becomes disconnected from the machine.
This method is required to be called before any commands or data
collection / callbacks can take place with the machine.
"""
logger.debug("start_session - starting session")
if (
self.config.hostname is None
or self.config.access_code is None
or self.config.serial_number is None
):
raise Exception("hostname, access_code, and serial_number are required")
if self.client and self.client.is_connected():
raise Exception("a session is already active")
def on_connect(client, userdata, flags, reason_code, properties):
logger.debug("on_connect - session on_connect")
if self.service_state != ServiceState.PAUSED:
self.service_state = ServiceState.CONNECTED
client.subscribe(f"device/{self.config.serial_number}/report")
logger.debug(f"subscribed to [device/{self.config.serial_number}/report]")
def on_disconnect(client, userdata, flags, reason_code, properties):
logger.debug("on_disconnect - session on_disconnect")
if self._internalException:
logger.exception("on_disconnect - an internal exception occurred")
self.service_state = ServiceState.QUIT
raise self._internalException
if self.service_state != ServiceState.PAUSED:
self.service_state = ServiceState.DISCONNECTED
def on_message(client, userdata, msg):
# logger.debug(f"on_message - topic: [{msg.topic}]")
if self._lastMessageTime and self._recent_update:
self._lastMessageTime = time.monotonic()
self._on_message(msg.payload.decode("utf-8"))
def loop_forever(printer: BambuPrinter):
logger.debug("loop_forever - starting mqtt loop_forever")
try:
printer.client.loop_forever(retry_first_connection=True)
except Exception as e:
logger.exception("loop_forever - an internal exception occurred")
printer._internalException = e
if printer.client and printer.client.is_connected():
printer.client.disconnect()
printer.service_state = ServiceState.QUIT
self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2) # type: ignore
self.client.on_connect = on_connect
self.client.on_disconnect = on_disconnect
self.client.on_message = on_message
self.client.tls_set(tls_version=ssl.PROTOCOL_TLS, cert_reqs=ssl.CERT_NONE)
self.client.tls_insecure_set(True)
self.client.reconnect_delay_set(min_delay=1, max_delay=1)
self.client.username_pw_set(
self.config.mqtt_username, password=self.config.access_code
)
self.client.user_data_set(self.config.mqtt_client_id)
try:
self.client.connect(self.config.hostname, self.config.mqtt_port, 60)
except Exception as e:
self._internalException = e
logger.warning(
f"start_session - unable to connect to printer - reason: [{e}] stacktrace: [{traceback.format_exc()}]"
)
self.service_state = ServiceState.QUIT
return
self._mqtt_client_thread = threading.Thread(
target=loop_forever, name="bambuprinter-session", args=(self,)
)
self._mqtt_client_thread.start()
self._start_watchdog()
stop_printing
Requests the printer to stop printing if a job is currently running.
Source code in src/bpm/bambuprinter.py
def stop_printing(self):
"""
Requests the printer to stop printing if a job is currently running.
"""
self.client.publish(
f"device/{self.config.serial_number}/request",
json.dumps(STOP_PRINT),
)
logger.debug(
f"stop_printing - published STOP_PRINT to [device/{self.config.serial_number}/request]"
)
toJson
Returns a dict (json document) representing this object's private class
level attributes that are serializable (most are).
Source code in src/bpm/bambuprinter.py
def toJson(self):
"""
Returns a `dict` (json document) representing this object's private class
level attributes that are serializable (most are).
"""
response = json.dumps(self, default=self.jsonSerializer, indent=4, sort_keys=True)
# logger.debug(f"toJson - json: [{response}]")
return json.loads(response)
turn_off_ams_dryer
Sends a command to the printer to turn off the AMS dryer.
Source code in src/bpm/bambuprinter.py
def turn_off_ams_dryer(self, ams_id: int = 0):
"""
Sends a command to the printer to turn off the AMS dryer.
"""
cmd = copy.deepcopy(AMS_FILAMENT_DRYING)
cmd["print"]["ams_id"] = ams_id
cmd["print"]["mode"] = 0 # Turn off drying mode
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
ams = next((u for u in self._printer_state.ams_units if u.ams_id == ams_id), None)
if not ams:
raise Exception("invalid ams_id provided")
ams.temp_target = 0
logger.debug(
f"turn_off_ams_dryer - published AMS_FILAMENT_DRYING to [device/{self.config.serial_number}/request] command: [{cmd}]"
)
turn_on_ams_dryer
turn_on_ams_dryer(
target_temp: int,
duration: int,
target_humidity: int = 0,
cooling_temp: int = 45,
rotate_tray: bool = False,
ams_id: int = 0,
)
Sends a command to the printer to turn on the AMS dryer with specified parameters.
Parameters
- target_temp : int - The target drying temperature.
- duration : int - The drying duration in minutes.
- target_humidity : int - The target humidity level.
- cooling_temp : int - The cooling temperature after drying (default is 45).
- rotate_tray : bool - Whether to rotate the tray during drying (default is False).
- ams_id : int - The AMS ID to control (default is 0).
Source code in src/bpm/bambuprinter.py
def turn_on_ams_dryer(
self,
target_temp: int,
duration: int,
target_humidity: int = 0,
cooling_temp: int = 45,
rotate_tray: bool = False,
ams_id: int = 0,
):
"""
Sends a command to the printer to turn on the AMS dryer with specified parameters.
Parameters
----------
* target_temp : int - The target drying temperature.
* duration : int - The drying duration in minutes.
* target_humidity : int - The target humidity level.
* cooling_temp : int - The cooling temperature after drying (default is 45).
* rotate_tray : bool - Whether to rotate the tray during drying (default is False).
* ams_id : int - The AMS ID to control (default is 0).
"""
cmd = copy.deepcopy(AMS_FILAMENT_DRYING)
cmd["print"]["ams_id"] = ams_id
cmd["print"]["mode"] = 1 # Turn on drying mode
cmd["print"]["temp"] = target_temp
cmd["print"]["duration"] = duration
cmd["print"]["humidity"] = target_humidity
cmd["print"]["cooling_temp"] = cooling_temp
cmd["print"]["rotate_tray"] = rotate_tray
self.client.publish(
f"device/{self.config.serial_number}/request", json.dumps(cmd)
)
ams = next((u for u in self._printer_state.ams_units if u.ams_id == ams_id), None)
if not ams:
raise Exception("invalid ams_id provided")
ams.temp_target = target_temp
logger.debug(
f"turn_on_ams_dryer - published AMS_FILAMENT_DRYING to [device/{self.config.serial_number}/request] command: [{cmd}]"
)
unload_filament
Requests the printer to unload whatever filament / spool may be currently loaded.
Source code in src/bpm/bambuprinter.py
def unload_filament(self, ams_id: int = 0):
"""
Requests the printer to unload whatever filament / spool may be currently loaded.
"""
msg = copy.deepcopy(AMS_CHANGE_FILAMENT)
msg["print"]["ams_id"] = ams_id
self.client.publish(
f"device/{self.config.serial_number}/request",
json.dumps(msg),
)
logger.debug(
f"unload_filament - published AMS_CHANGE_FILAMENT to [device/{self.config.serial_number}/request] - bambu_msg: [{msg}]"
)
upload_sdcard_file
Uploads the local filesystem file to the printer and returns an updated dict of all files on the printer
Parameters
- src : str - the full path filename on the host to be uploaded to the printer
- dest : str - the full path filename on the printer to upload to
Source code in src/bpm/bambuprinter.py
def upload_sdcard_file(self, src: str, dest: str):
"""
Uploads the local filesystem file to the printer and returns an updated dict of all files on the printer
Parameters
----------
* src : str - the full path filename on the host to be uploaded to the printer
* dest : str - the full path filename on the printer to upload to
"""
logger.debug(f"upload_sdcard_file - uploading file src: [{src}] dest: [{dest}]")
with self.ftp_connection() as ftps:
ftps.upload_file(src, dest)
return self.get_sdcard_contents()