bambuprinter

Classes:

Name	Description
`BambuPrinter`	`BambuPrinter` is the main class within `bambu-printer-manager`

BambuPrinter

Python

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 BambuConfig instance that provides configuration information for the printer connection. If not provided, a default BambuConfig instance 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 `dict` (json document) of all `.3mf` files on the printer's SD card.
`get_sdcard_contents`	Returns a `dict` (json document) of ALL files on the printer's SD card.
`jsonSerializer`	Helper method used by `toJson()` to serialize this object.
`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 `BambuPrinter` session is it is active. Under the covers this
`print_3mf_file`	Submits a request to execute the `name` 3mf file on the printer's SDCard.
`quit`	Shuts down all threads. Your `BambuPrinter` instance should probably be
`refresh`	Triggers a full data refresh from the printer (if it is connected). You should use this
`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 `path` specified
`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 `AMSUserSetting` options
`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_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 `PrintOption` options
`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.
`target_spool`
`toJson`	Returns a `dict` (json document) representing this object's private class
`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`		returns the current bed temperature
`bed_temp_target`		returns or sets the bed temperature target
`chamber_temp`		returns or sets the nozzle temperature target
`chamber_temp_target`		returns or sets the chamber temperature target
`fan_speed`		returns the part fan speed in percent
`fan_speed_target`		returns or sets the part cooling fan target speed in percent
`heatbreak_fan_speed`		returns the heatbreak fan's current speed in percent
`on_update`		sets or returns the callback function that is called when the printer state is updated
`tool_temp`		returns or sets the tool temperature
`tool_temp_target`		returns or sets the nozzle temperature target

Source code in src/bpm/bambuprinter.py

Python

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._bed_temp = 0.0
    self._bed_temp_target = 0
    self._bed_temp_target_time = 0

    self._tool_temp = 0.0
    self._tool_temp_target = 0
    self._tool_temp_target_time = 0

    self._chamber_temp = 0.0
    self._chamber_temp_target = 0
    self._chamber_temp_target_time = 0

    self._fan_gear = 0
    self._heatbreak_fan_speed = 0
    self._fan_speed = 0
    self._fan_speed_target = 0
    self._fan_speed_target_time = 0

    self._light_state = ""
    self._wifi_signal = ""
    self._speed_level = 0

    self._gcode_state = ""
    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._percent_complete = 0
    self._time_remaining = 0
    self._start_time = 0
    self._elapsed_time = 0
    self._layer_count = 0
    self._current_layer = 0

    self._current_stage = 0
    self._current_stage_text = ""

    self._spools = []
    self._printer_state = BambuState()
    self._target_spool = 255
    self._active_spool = 255
    self._spool_state = ""
    self._ams_status = None
    self._ams_exists = False
    self._ams_rfid_status = None

    self._sdcard_contents = None
    self._sdcard_3mf_files = None

    self._hms_data = None
    self._hms_message = ""
    self._last_hms_message = ""

    self._print_type = ""
    self._skipped_objects = []

    self._nozzle_type = ""
    self._nozzle_diameter = 0.0

bed_temp `property`

Python

bed_temp

returns the current bed temperature

Deprecated

This property is deprecated (v1.0.0). Use BambuState.bed_temp instead.

bed_temp_target `property` `writable`

Python

bed_temp_target

returns or sets the bed temperature target

Deprecated

This property is deprecated (v1.0.0). Use BambuState.bed_temp_target and set_bed_temp_target.

chamber_temp `property` `writable`

Python

chamber_temp

returns or sets the nozzle temperature target

Deprecated

This property is deprecated (v1.0.0). Use BambuState.chamber_temp and set_chamber_temp.

chamber_temp_target `property` `writable`

Python

chamber_temp_target

returns or sets the chamber temperature target

Deprecated

This property is deprecated (v1.0.0). Use BambuState.chamber_temp_target and set_chamber_temp_target.

fan_speed `property`

Python

fan_speed

returns the part fan speed in percent

Deprecated

This property is deprecated (v1.0.0). Use BambuState.part_cooling_fan_speed_percent.

fan_speed_target `property` `writable`

Python

fan_speed_target

returns or sets the part cooling fan target speed in percent

Deprecated

This property is deprecated (v1.0.0). Use BambuState.part_cooling_fan_speed_target_percent and set_part_cooling_fan_speed_target_percent.

heatbreak_fan_speed `property`

Python

heatbreak_fan_speed

returns the heatbreak fan's current speed in percent

Deprecated

This property is deprecated (v1.0.0). Use BambuState.heatbreak_fan_speed_percent.

on_update `property` `writable`

Python

on_update

sets or returns the callback function that is called when the printer state is updated

Parameters

value : method - (setter) The method to call.

tool_temp `property`

Python

tool_temp

returns or sets the tool temperature

Deprecated

This property is deprecated (v1.0.0). Use BambuState.nozzle_temp.

tool_temp_target `property` `writable`

Python

tool_temp_target

returns or sets the nozzle temperature target

Deprecated

This property is deprecated (v1.0.0). Use BambuState.nozzle_temp.

delete_sdcard_file

Python

delete_sdcard_file(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

Source code in src/bpm/bambuprinter.py

Python

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

Python

delete_sdcard_folder(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

Source code in src/bpm/bambuprinter.py

Python

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

Python

download_sdcard_file(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

Source code in src/bpm/bambuprinter.py

Python

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

Python

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.

Source code in src/bpm/bambuprinter.py

Python

@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.
    """
    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

Python

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

Python

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

Python

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

Python

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")
    return self._sdcard_3mf_files

get_sdcard_contents

Python

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

Python

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

Python

jsonSerializer(obj)

Helper method used by toJson() to serialize this object.

Source code in src/bpm/bambuprinter.py

Python

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

Python

load_filament(slot: int)

Requests the printer to load filament into the extruder using the requested spool (slot #)

Parameters

slot : int

0 - AMS Spool #1
1 - AMS Spool #2
2 - AMS Spool #3
3 - AMS Spool #4
254 - External Spool

Source code in src/bpm/bambuprinter.py

Python

def load_filament(self, slot: int):
    """
    Requests the printer to load filament into the extruder using the requested spool (slot #)

    Parameters
    ----------
    slot : 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 = AMS_FILAMENT_CHANGE
    msg["print"]["target"] = int(slot)
    self.client.publish(
        f"device/{self.config.serial_number}/request", json.dumps(msg)
    )
    logger.debug(
        f"load_filament - published AMS_FILAMENT_CHANGE to [device/{self.config.serial_number}/request] - target: [{slot}], bambu_msg: [{msg}]"
    )

make_sdcard_directory

Python

make_sdcard_directory(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

Source code in src/bpm/bambuprinter.py

Python

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

Python

pause_printing()

Pauses the current print job if one is running.

Source code in src/bpm/bambuprinter.py

Python

def pause_printing(self):
    """
    Pauses the current print job if one is running.
    """
    self.client.publish(
        f"device/{self.config.serial_number}/request",
        json.dumps(PAUSE_PRINT),
    )
    logger.debug(
        f"pause_printing - published PAUSE_PRINT to [device/{self.config.serial_number}/request]"
    )

pause_session

Python

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

Python

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

Python

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 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

Source code in src/bpm/bambuprinter.py

Python

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

Python

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

Python

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

Python

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

Python

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),
        )

rename_sdcard_file

Python

rename_sdcard_file(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

Source code in src/bpm/bambuprinter.py

Python

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

Python

resume_printing()

Resumes the current print job if one is paused.

Source code in src/bpm/bambuprinter.py

Python

def resume_printing(self):
    """
    Resumes the current print job if one is paused.
    """
    self.client.publish(
        f"device/{self.config.serial_number}/request",
        json.dumps(RESUME_PRINT),
    )
    logger.debug(
        f"resume_printing - published RESUME_PRINT to [device/{self.config.serial_number}/request]"
    )

resume_session

Python

resume_session()

Resumes a previously paused session by re-subscribing to the /report topic.

Source code in src/bpm/bambuprinter.py

Python

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

Python

sdcard_file_exists(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

Source code in src/bpm/bambuprinter.py

Python

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

Python

select_extrusion_calibration_profile(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)

Source code in src/bpm/bambuprinter.py

Python

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

Python

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

Python

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}]"
    )

send_anything

Python

send_anything(anything: str)

puts an arbritary string on the request topic

Source code in src/bpm/bambuprinter.py

Python

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

Python

send_gcode(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

Source code in src/bpm/bambuprinter.py

Python

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

Python

set_active_tool(id: int)

sets the current active tool / extruder for machines (H2 series) that have multiple extruders

Source code in src/bpm/bambuprinter.py

Python

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

Python

set_ams_user_setting(setting: AMSUserSetting, enabled: bool, ams_id: int | None = 0)

Enable or disable one of the AMSUserSetting options

Source code in src/bpm/bambuprinter.py

Python

def set_ams_user_setting(
    self, setting: AMSUserSetting, enabled: bool, ams_id: int | None = 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_bed_temp_target

Python

set_bed_temp_target(value: int)

Sets the bed temperature target.

Parameters

value : float - The target bed temperature.

Source code in src/bpm/bambuprinter.py

Python

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

Python

set_buildplate_marker_detector(enabled: bool)

Enables or disables the buildplate_marker_detector

Source code in src/bpm/bambuprinter.py

Python

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

Python

set_chamber_temp(value: float)

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

Python

def set_chamber_temp(self, value: float):
    """
    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.
    """
    self._printer_state.climate.chamber_temp = value

set_chamber_temp_target

Python

set_chamber_temp_target(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?

Source code in src/bpm/bambuprinter.py

Python

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.printer_state.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}]"
        )
    else:
        self._printer_state.climate.chamber_temp_target = value
    self._chamber_temp_target_time = round(time.time())

set_nozzle_details

Python

set_nozzle_details(nozzle_diameter: NozzleDiameter, nozzle_type: NozzleType)

Sets the nozzle details.

Source code in src/bpm/bambuprinter.py

Python

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

Python

set_nozzle_temp_target(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).

Source code in src/bpm/bambuprinter.py

Python

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

Python

set_part_cooling_fan_speed_target_percent(value: int)

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

Python

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}\nM106 P2 S{speed}\nM106 P3 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

Python

set_print_option(option: PrintOption, enabled: bool)

Enable or disable one of the PrintOption options

Source code in src/bpm/bambuprinter.py

Python

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

Python

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,
)

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

Python

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,
):
    """
    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`

Python

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

Python

@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

Python

skip_objects(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

Python

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

Python

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

Python

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

Python

stop_printing()

Requests the printer to stop printing if a job is currently running.

Source code in src/bpm/bambuprinter.py

Python

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]"
    )

target_spool `deprecated`

Python

target_spool()

Deprecated

This property is deprecated (v1.0.0). Use BambuState.target_tray_id.

Source code in src/bpm/bambuprinter.py

Python

@deprecated("This property is deprecated (v1.0.0). Use `BambuState.target_tray_id`.")
def target_spool(self):
    return self._printer_state.target_tray_id

toJson

Python

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

Python

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

Python

turn_off_ams_dryer(ams_id: int = 0)

Sends a command to the printer to turn off the AMS dryer.

Source code in src/bpm/bambuprinter.py

Python

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 = self.printer_state.ams_units[ams_id]
    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

Python

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

Python

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 = self.printer_state.ams_units[ams_id]
    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

Python

unload_filament()

Requests the printer to unload whatever filament / spool may be currently loaded.

Source code in src/bpm/bambuprinter.py

Python

def unload_filament(self):
    """
    Requests the printer to unload whatever filament / spool may be currently loaded.
    """
    self.client.publish(
        f"device/{self.config.serial_number}/request",
        json.dumps(UNLOAD_FILAMENT),
    )
    logger.debug(
        f"unload_filament - published UNLOAD_FILAMENT to [device/{self.config.serial_number}/request]"
    )

upload_sdcard_file

Python

upload_sdcard_file(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

Source code in src/bpm/bambuprinter.py

Python

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()

bambuprinter

BambuPrinter

Parameters

bed_temp property

bed_temp_target property writable

chamber_temp property writable

chamber_temp_target property writable

fan_speed property

fan_speed_target property writable

heatbreak_fan_speed property

on_update property writable

Parameters

tool_temp property

tool_temp_target property writable

delete_sdcard_file

Parameters

delete_sdcard_folder

Parameters

download_sdcard_file

Parameters

ftp_connection

get_current_bind_list

get_sdcard_3mf_files

Usage

get_sdcard_contents

Usage

jsonSerializer

load_filament

Parameters

make_sdcard_directory

Parameters

pause_printing

pause_session

print_3mf_file

Parameters

Example

AMS Mapping

quit

refresh

rename_sdcard_file

Parameters

resume_printing

resume_session

sdcard_file_exists

Parameters

select_extrusion_calibration_profile

Parameters

send_ams_control_command

send_anything

send_gcode

Parameters

Examples

set_active_tool

set_ams_user_setting

set_bed_temp_target

Parameters

set_buildplate_marker_detector

set_chamber_temp

Parameters

set_chamber_temp_target

Parameters

set_nozzle_details

set_nozzle_temp_target

Parameters

set_part_cooling_fan_speed_target_percent

Parameters

set_print_option

set_spool_details

set_spool_k_factor deprecated

skip_objects

Parameters

start_session

stop_printing

target_spool deprecated

toJson

turn_off_ams_dryer

turn_on_ams_dryer

Parameters

unload_filament

upload_sdcard_file

bed_temp `property`

bed_temp_target `property` `writable`

chamber_temp `property` `writable`

chamber_temp_target `property` `writable`

fan_speed `property`

fan_speed_target `property` `writable`

heatbreak_fan_speed `property`

on_update `property` `writable`

tool_temp `property`

tool_temp_target `property` `writable`

set_spool_k_factor `deprecated`

target_spool `deprecated`