bambustate provides a unified, thread-safe representation of a Bambu Lab printer's
operational state, synchronized via MQTT telemetry.
Classes:
| Name | Description |
|---|---|
AMSUnitState |
State information about an individual AMS unit. |
BambuClimate |
Contains all climate related attributes |
BambuState |
Representation of the Bambu printer state synchronized via MQTT. |
ExtruderState |
State for an individual physical extruder toolhead. |
NozzleCharacteristics |
Normalized nozzle characteristics across telemetry and encoded identifiers. |
dataclass
Python
AMSUnitState(
ams_id: int,
chip_id: str = "",
model: AMSModel = UNKNOWN,
temp_actual: float = 0.0,
temp_target: int = 0,
humidity_index: int = 0,
humidity_raw: int = 0,
ams_info: int = 0,
heater_state: AMSHeatingState = OFF,
dry_fan1_status: AMSDryFanStatus = OFF,
dry_fan2_status: AMSDryFanStatus = OFF,
dry_sub_status: AMSDrySubStatus = OFF,
dry_time: int = 0,
tray_exists: list[bool] = (lambda: [False] * 4)(),
assigned_to_extruder: ActiveTool = SINGLE_EXTRUDER,
)
State information about an individual AMS unit.
Attributes:
| Name | Type | Description |
|---|---|---|
ams_id |
int
|
Unique ID. |
ams_info |
int
|
Underlying ams info value |
assigned_to_extruder |
ActiveTool
|
Target tool computed from raw_extruder_id |
chip_id |
str
|
Hardware serial. |
dry_fan1_status |
AMSDryFanStatus
|
Drying fan 1 status (bits 18-19 of ams_info) |
dry_fan2_status |
AMSDryFanStatus
|
Drying fan 2 status (bits 20-21 of ams_info) |
dry_sub_status |
AMSDrySubStatus
|
Drying sub-status phase (bits 22-25 of ams_info) |
dry_time |
int
|
Minutes left. |
heater_state |
AMSHeatingState
|
The computed state of the AMS's heater |
humidity_index |
int
|
Humidity index. |
humidity_raw |
int
|
Raw humidity. |
model |
AMSModel
|
|
temp_actual |
float
|
Actual temp. |
temp_target |
int
|
Target drying temp. |
tray_exists |
list[bool]
|
Slot presence. |
class-attribute
instance-attribute
Python
assigned_to_extruder: ActiveTool = SINGLE_EXTRUDER
Target tool computed from raw_extruder_id
class-attribute
instance-attribute
Python
dry_fan1_status: AMSDryFanStatus = OFF
Drying fan 1 status (bits 18-19 of ams_info)
class-attribute
instance-attribute
Python
dry_fan2_status: AMSDryFanStatus = OFF
Drying fan 2 status (bits 20-21 of ams_info)
class-attribute
instance-attribute
Python
dry_sub_status: AMSDrySubStatus = OFF
Drying sub-status phase (bits 22-25 of ams_info)
class-attribute
instance-attribute
Python
heater_state: AMSHeatingState = OFF
The computed state of the AMS's heater
dataclass
Python
BambuClimate(
bed_temp: float = 0.0,
bed_temp_target: int = 0,
airduct_mode: int = -1,
airduct_sub_mode: int = -1,
chamber_temp: float = 0.0,
chamber_temp_target: int = 0,
air_conditioning_mode: AirConditioningMode = NOT_SUPPORTED,
part_cooling_fan_speed_percent: int = 0,
part_cooling_fan_speed_target_percent: int = 0,
aux_fan_speed_percent: int = 0,
exhaust_fan_speed_percent: int = 0,
heatbreak_fan_speed_percent: int = 0,
zone_intake_open: bool = False,
zone_part_fan_percent: int = 0,
zone_aux_percent: int = 0,
zone_exhaust_percent: int = 0,
zone_top_vent_open: bool = False,
is_chamber_door_open: bool = False,
is_chamber_lid_open: bool = False,
)
Contains all climate related attributes
Attributes:
| Name | Type | Description |
|---|---|---|
air_conditioning_mode |
AirConditioningMode
|
The mode the printer's AC is in if equipped with one. |
airduct_mode |
int
|
Raw current mode. |
airduct_sub_mode |
int
|
Raw sub mode. |
aux_fan_speed_percent |
int
|
aux fan %. |
bed_temp |
float
|
Bed temp. |
bed_temp_target |
int
|
Bed target. |
chamber_temp |
float
|
Chamber temp. |
chamber_temp_target |
int
|
Chamber target. |
exhaust_fan_speed_percent |
int
|
Exhaust (chamber) fan %. |
heatbreak_fan_speed_percent |
int
|
Heatbreak fan %. |
is_chamber_door_open |
bool
|
For printers that support it (see |
is_chamber_lid_open |
bool
|
For printers that support it (see |
part_cooling_fan_speed_percent |
int
|
Part fan %. |
part_cooling_fan_speed_target_percent |
int
|
Part target %. |
zone_aux_percent |
int
|
aux %. |
zone_exhaust_percent |
int
|
Exhaust %. |
zone_intake_open |
bool
|
Heater power. |
zone_part_fan_percent |
int
|
Internal %. |
zone_top_vent_open |
bool
|
Top vent status - derived from exhaust fan on and cooling ac mode. |
class-attribute
instance-attribute
Python
air_conditioning_mode: AirConditioningMode = NOT_SUPPORTED
The mode the printer's AC is in if equipped with one.
class-attribute
instance-attribute
aux fan %.
class-attribute
instance-attribute
Chamber target.
class-attribute
instance-attribute
Exhaust (chamber) fan %.
class-attribute
instance-attribute
Heatbreak fan %.
class-attribute
instance-attribute
For printers that support it (see PrinterCapabilities.has_chamber_door_sensor), reports whether the chamber door is open
class-attribute
instance-attribute
For printers that support it (see PrinterCapabilities.has_chamber_door_sensor), reports whether the chamber lid is open
class-attribute
instance-attribute
Part fan %.
class-attribute
instance-attribute
Part target %.
class-attribute
instance-attribute
Exhaust %.
class-attribute
instance-attribute
Heater power.
class-attribute
instance-attribute
Internal %.
dataclass
Python
BambuState(
gcode_state: str = "IDLE",
active_ams_id: int = -1,
active_tray_id: int = 255,
active_tray_state: TrayState = UNLOADED,
active_tray_state_name: str = name,
target_tray_id: int = -1,
active_tool: ActiveTool = SINGLE_EXTRUDER,
is_external_spool_active: bool = False,
active_nozzle_temp: float = 0.0,
active_nozzle_temp_target: int = 0,
active_nozzle: NozzleCharacteristics = NozzleCharacteristics(),
ams_status_raw: int = 0,
ams_status_text: str = "",
ams_exist_bits: int = 0,
ams_connected_count: int = 0,
ams_units: list[AMSUnitState] = list(),
extruders: list[ExtruderState] = list(),
spools: list[BambuSpool] = list(),
print_error: int = 0,
hms_errors: list[dict] = list(),
wifi_signal_strength: str = "",
climate: BambuClimate = BambuClimate(),
stat: str = "0",
fun: str = "0",
)
Representation of the Bambu printer state synchronized via MQTT.
Methods:
| Name | Description |
|---|---|
fromJson |
Parses root MQTT payloads into a hierachical BambuState object. |
Attributes:
| Name | Type | Description |
|---|---|---|
active_ams_id |
int
|
Current active AMS unit id |
active_nozzle |
NozzleCharacteristics
|
Normalized characteristics of the currently active nozzle. |
active_nozzle_temp |
float
|
Nozzle temp. |
active_nozzle_temp_target |
int
|
Nozzle target. |
active_tool |
ActiveTool
|
Active toolhead. |
active_tray_id |
int
|
Current tray. |
active_tray_state |
TrayState
|
Loading enum. |
active_tray_state_name |
str
|
Loading string. |
ams_connected_count |
int
|
AMS count. |
ams_exist_bits |
int
|
AMS mask. |
ams_status_raw |
int
|
Raw AMS status. |
ams_status_text |
str
|
Human AMS status. |
ams_units |
list[AMSUnitState]
|
Unit details. |
climate |
BambuClimate
|
Contains all climate related attributes |
extruders |
list[ExtruderState]
|
Extruder details. |
gcode_state |
str
|
Execution state. |
hms_errors |
list[dict]
|
HMS list. |
is_external_spool_active |
bool
|
Ext spool flag. |
print_error |
int
|
Main error. |
spools |
list[BambuSpool]
|
All spools associated with this printer |
target_tray_id |
int
|
Next tray. |
wifi_signal_strength |
str
|
Wi-Fi signal strength in dBm |
class-attribute
instance-attribute
Current active AMS unit id
class-attribute
instance-attribute
Python
active_nozzle: NozzleCharacteristics = field(default_factory=NozzleCharacteristics)
Normalized characteristics of the currently active nozzle.
class-attribute
instance-attribute
Nozzle temp.
class-attribute
instance-attribute
Nozzle target.
class-attribute
instance-attribute
Python
active_tool: ActiveTool = SINGLE_EXTRUDER
Active toolhead.
class-attribute
instance-attribute
Python
active_tray_state: TrayState = UNLOADED
Loading enum.
class-attribute
instance-attribute
Loading string.
class-attribute
instance-attribute
AMS count.
class-attribute
instance-attribute
Human AMS status.
class-attribute
instance-attribute
Python
ams_units: list[AMSUnitState] = field(default_factory=list)
Unit details.
class-attribute
instance-attribute
Python
climate: BambuClimate = field(default_factory=BambuClimate)
Contains all climate related attributes
class-attribute
instance-attribute
Python
extruders: list[ExtruderState] = field(default_factory=list)
Extruder details.
class-attribute
instance-attribute
HMS list.
class-attribute
instance-attribute
Ext spool flag.
class-attribute
instance-attribute
Python
spools: list[BambuSpool] = field(default_factory=list)
All spools associated with this printer
class-attribute
instance-attribute
Wi-Fi signal strength in dBm
classmethod
Python
fromJson(data: dict[str, Any], printer: BambuPrinter) -> BambuState
Parses root MQTT payloads into a hierachical BambuState object.
Source code in src/bpm/bambustate.py
Python
@classmethod
def fromJson(cls, data: dict[str, Any], printer: "BambuPrinter") -> "BambuState":
"""Parses root MQTT payloads into a hierachical BambuState object."""
current_state = printer.printer_state
config = printer.config
aji = printer.active_job_info
base = current_state if current_state else cls()
info = data.get("info", {})
p = data.get("print", {})
if (
p.get("command", "") == "ams_filament_drying"
and p.get("result", "") == "success"
):
ams_id = p.get("ams_id", -1)
ams = next((u for u in base.ams_units if u.ams_id == ams_id), None)
if ams:
ams.temp_target = int(p.get("temp", 0))
ams_root = p.get("ams", {})
device = p.get("device", {})
extruder_root = device.get("extruder", {})
nozzle_root = device.get("nozzle", {})
ctc_root = device.get("ctc", {})
airduct_root = device.get("airduct", {})
modules = info.get("module", [])
updates = {}
# CAPABILITIES
caps = asdict(config.capabilities)
if ctc_root:
caps["has_chamber_temp"] = True
if "ams" in ams_root or "ams" in p:
caps["has_ams"] = True
if airduct_root:
caps["has_air_filtration"] = True
if len(extruder_root.get("info", [])) > 1:
caps["has_dual_extruder"] = True
caps["has_camera"] = True
xcam_data = p.get("xcam", None)
if xcam_data:
caps["has_lidar"] = xcam_data.get("first_layer_inspector", False)
else:
caps["has_lidar"] = config.capabilities.has_lidar
new_caps = PrinterCapabilities(**caps)
climate = asdict(base.climate)
updates["climate"] = BambuClimate(**climate)
# STATUS & PROGRESS
updates["gcode_state"] = p.get("gcode_state", base.gcode_state)
updates["fun"] = p.get("fun", base.fun)
fun = int(updates["fun"], 16)
new_caps.has_chamber_door_sensor = bool((fun >> 12) & 0x01)
new_caps.has_spaghetti_detector_support = bool((fun >> 42) & 0x01)
new_caps.has_purgechutepileup_detector_support = bool((fun >> 43) & 0x01)
new_caps.has_nozzleclumping_detector_support = bool((fun >> 44) & 0x01)
new_caps.has_airprinting_detector_support = bool((fun >> 45) & 0x01)
if new_caps.has_chamber_door_sensor:
updates["stat"] = p.get("stat", base.stat)
stat = int(updates["stat"], 16)
updates["climate"].is_chamber_door_open = bool((stat >> 23) & 0x01)
updates["climate"].is_chamber_lid_open = bool((stat >> 24) & 0x01)
# AIRDUCT
if airduct_root:
updates["climate"].airduct_mode = int(
airduct_root.get("modeCur", base.climate.airduct_mode)
)
updates["climate"].airduct_sub_mode = int(
airduct_root.get("subMode", base.climate.airduct_sub_mode)
)
if updates["climate"].airduct_mode == 1:
updates["climate"].air_conditioning_mode = AirConditioningMode.HEAT_MODE
elif updates["climate"].airduct_mode == 0:
updates["climate"].air_conditioning_mode = AirConditioningMode.COOL_MODE
base.climate.chamber_temp_target = 0
else:
updates[
"climate"
].air_conditioning_mode = AirConditioningMode.NOT_SUPPORTED
parts = {part["id"]: part["state"] for part in airduct_root.get("parts", [])}
updates["climate"].zone_part_fan_percent = parts.get(
16, base.climate.zone_part_fan_percent
)
updates["climate"].zone_aux_percent = parts.get(
32, base.climate.zone_aux_percent
)
updates["climate"].zone_exhaust_percent = parts.get(
48, base.climate.zone_exhaust_percent
)
zone_intake_open = parts.get(96, -1)
updates["climate"].zone_intake_open = zone_intake_open not in (-1, 0)
updates["climate"].zone_top_vent_open = bool(
updates["climate"].zone_exhaust_percent > 0
and not updates["climate"].zone_intake_open
)
# THERMALS & CTC DECODING
updates["climate"].bed_temp = float(p.get("bed_temper", base.climate.bed_temp))
updates["climate"].bed_temp_target = int(
p.get("bed_target_temper", base.climate.bed_temp_target)
)
ctc_temp_target = 0
if ctc_root:
ctc_temp_raw = unpackTemperature(ctc_root.get("info", {}).get("temp", 0.0))
ctc_temp = ctc_temp_raw[0]
ctc_temp_target = int(ctc_temp_raw[1])
updates["climate"].chamber_temp = ctc_temp
updates["climate"].chamber_temp_target = ctc_temp_target
elif not config.external_chamber:
chamber_temp = int(p.get("chamber_temper", base.climate.chamber_temp))
if chamber_temp != 5:
updates["climate"].chamber_temp = chamber_temp
if (
ctc_temp_target == 0
and updates["climate"].air_conditioning_mode != AirConditioningMode.HEAT_MODE
):
updates["climate"].chamber_temp_target = base.climate.chamber_temp_target
if p.get("command", "") == "set_ctt" and p.get("result", "") == "success":
ctc_temp_target = int(p.get("ctt_val", -1))
updates["climate"].chamber_temp_target = ctc_temp_target
if ctc_temp_target < 45:
updates["climate"].air_conditioning_mode = AirConditioningMode.COOL_MODE
# EXTRUDERS
new_extruders = []
if "info" in extruder_root:
nozzle_by_id: dict[int, dict[str, Any]] = {}
nozzle_info = nozzle_root.get("info", [])
for nozzle_item in nozzle_info:
raw_nozzle_id = nozzle_item.get("id")
nozzle_id = int(raw_nozzle_id) & 0xFF
nozzle_by_id[nozzle_id] = nozzle_item
for new_ext in extruder_root["info"]:
raw_t = int(new_ext.get("temp", 0))
act_t, tar_t = unpackTemperature(raw_t)
sn = int(new_ext.get("snow", -1))
hn = int(new_ext.get("hnow", -1))
st = int(new_ext.get("star", -1))
ht = int(new_ext.get("htar", -1))
ext = ExtruderState()
ext.id = int(new_ext.get("id", 0))
ext.temp = act_t
ext.temp_target = int(tar_t)
ext.info_bits = int(new_ext.get("info", 0))
ext.state = parseExtruderInfo(ext.info_bits)
ext.status = parseExtruderStatus(int(new_ext.get("stat", 0)))
nozzle_id_key = int(new_ext.get("hnow", ext.id))
nozzle_info = nozzle_by_id.get(nozzle_id_key)
if nozzle_info is None:
nozzle_info = nozzle_by_id.get(ext.id)
ext.nozzle = NozzleCharacteristics.from_telemetry(
nozzle_type=(
str(nozzle_info.get("type"))
if nozzle_info is not None and nozzle_info.get("type") is not None
else None
),
nozzle_diameter=(
nozzle_info.get("diameter")
if nozzle_info is not None
and nozzle_info.get("diameter") is not None
else None
),
nozzle_id=(
str(nozzle_info.get("id"))
if nozzle_info is not None and nozzle_info.get("id") is not None
else None
),
)
ext.active_tray_id = parseExtruderTrayState(ext.id, hn, sn)
ext.target_tray_id = parseExtruderTrayState(ext.id, ht, st)
if base.extruders and len(base.extruders) > ext.id:
base_tray_state = base.extruders[ext.id].tray_state
else:
base_tray_state = (
TrayState.LOADED
if ext.state != ExtruderInfoState.EMPTY
else TrayState.UNLOADED
)
if (
ext.state == ExtruderInfoState.LOADED
and ext.status == ExtruderStatus.ACTIVE
):
ext.tray_state = TrayState.LOADED
elif (
ext.state == ExtruderInfoState.EMPTY
and ext.status == ExtruderStatus.IDLE
):
ext.tray_state = TrayState.UNLOADED
elif (
ext.state == ExtruderInfoState.LOADED
and ext.status == ExtruderStatus.HEATING
and base_tray_state not in (TrayState.LOADING, TrayState.UNLOADED)
):
ext.tray_state = TrayState.UNLOADING
elif ext.status is not ExtruderStatus.IDLE:
ext.tray_state = TrayState.LOADING
else:
ext.tray_state = base.active_tray_state
new_extruders.append(ext)
else:
ext = base.extruders[0] if base.extruders else ExtruderState()
ext.id = ActiveTool.SINGLE_EXTRUDER
ext.temp = float(p.get("nozzle_temper", base.active_nozzle_temp))
ext.temp_target = int(
p.get("nozzle_target_temper", base.active_nozzle_temp_target)
)
ext.state = ExtruderInfoState.NOT_AVAILABLE
ext.status = ExtruderStatus.NOT_AVAILABLE
ext.active_tray_id = base.active_tray_id
ext.target_tray_id = base.target_tray_id
if "tray_now" in ams_root:
raw = int(ams_root["tray_now"])
ext.active_tray_id = -1 if raw == 255 else raw
if "tray_tar" in ams_root:
raw = int(ams_root["tray_tar"])
ext.target_tray_id = -1 if raw == 255 else raw
ext.assigned_to_ams_id = (
ext.active_tray_id >> 2
if ext.active_tray_id not in (-1, 254, 255)
else -1
)
if aji.stage_id == 24:
ext.tray_state = TrayState.LOADING
elif aji.stage_id == 22:
ext.tray_state = TrayState.UNLOADING
elif ext.active_tray_id not in (-1, 254, 255):
ext.tray_state = TrayState.LOADED
else:
ext.tray_state = TrayState.UNLOADED
ext.nozzle = NozzleCharacteristics.from_telemetry(
nozzle_type=p.get(
"nozzle_type",
ext.nozzle.telemetry_type_raw,
),
nozzle_diameter=p.get("nozzle_diameter", ext.nozzle.diameter_mm),
nozzle_id=p.get("nozzle_id", ext.nozzle.encoded_id),
flow_type=NozzleFlowType.STANDARD,
)
new_extruders.append(ext)
updates["extruders"] = new_extruders if new_extruders else base.extruders
# TOOL SELECTION
if "state" in extruder_root:
raw_t_idx = (int(extruder_root["state"]) >> 4) & 0xF
if new_caps.has_dual_extruder:
updates["active_tool"] = ActiveTool(raw_t_idx)
else:
updates["active_tool"] = ActiveTool.SINGLE_EXTRUDER
else:
updates["active_tool"] = base.active_tool
# AMS UNITS
cur_ams = {u.ams_id: u for u in base.ams_units}
for m in modules:
if (
m.get("name", "").startswith("n3f/")
or m.get("name", "").startswith("n3s/")
or m.get("name", "").startswith("ams")
):
ams_id = int(m["name"].split("/")[-1])
u = cur_ams.get(ams_id, AMSUnitState(ams_id=ams_id))
u.chip_id = m.get("sn", u.chip_id)
u.model = getAMSModelBySerial(u.chip_id)
cur_ams[ams_id] = u
for ams_u in ams_root.get("ams", []):
id = int(ams_u.get("id", 0))
u = cur_ams.get(id, AMSUnitState(ams_id=id))
u.temp_actual = float(ams_u.get("temp", u.temp_actual))
u.humidity_index = int(float(ams_u.get("humidity", u.humidity_index)))
u.humidity_raw = int(float(ams_u.get("humidity_raw", u.humidity_raw)))
u.dry_time = int(float(ams_u.get("dry_time", u.dry_time)))
# ugly hack for capturing target temp
if u.dry_time > 0 and u.temp_target < int(u.temp_actual) - 1:
u.temp_target = int(u.temp_actual)
elif u.dry_time == 0:
u.temp_target = 0
if "info" in ams_u:
u.ams_info = int(ams_u["info"], 16)
p_ams = parseAMSInfo(ams_u["info"])
u.heater_state = p_ams["heater_state"]
u.dry_fan1_status = p_ams["dry_fan1_status"]
u.dry_fan2_status = p_ams["dry_fan2_status"]
u.dry_sub_status = p_ams["dry_sub_status"]
# Update AMS model from parsed info if not already set
if u.model == AMSModel.UNKNOWN:
u.model = p_ams["ams_type"]
if new_caps.has_dual_extruder:
u.assigned_to_extruder = ActiveTool(p_ams.get("extruder_id", 15))
updates["extruders"][
u.assigned_to_extruder.value
].assigned_to_ams_id = u.ams_id
rb = ams_root.get("tray_exist_bits")
if rb is not None:
eb = int(rb, 16) if isinstance(rb, str) else int(rb)
# Calculate the bit shift based on the unit ID
# Standard AMS: 0, 1, 2, 3 -> shift 0, 4, 8, 12
# AMS-HT: 128, 129, 130, 131 -> shift 16, 20, 24, 28
if id >= 128:
shift = 16 + (4 * (id - 128))
# AMS-HT is a 1-slot unit, so we check only range(1)
u.tray_exists = [bool((eb >> shift) & (1 << j)) for j in range(1)]
else:
shift = 4 * id
# Standard AMS is a 4-slot unit, so we check range(4)
u.tray_exists = [bool((eb >> shift) & (1 << j)) for j in range(4)]
cur_ams[id] = u
updates["ams_units"] = list(cur_ams.values())
# ACTIVE / TARGET TRAYS AND TOOL TEMP
# if multi-extruder return the active one
a_ext = next(
(e for e in updates["extruders"] if e.id == updates["active_tool"].value),
None,
)
if a_ext:
if a_ext.active_tray_id not in (254, 255, -1):
updates["active_ams_id"] = (
a_ext.assigned_to_ams_id
if a_ext.assigned_to_ams_id != -1
else a_ext.active_tray_id >> 2
)
else:
updates["active_ams_id"] = -1
updates["active_tray_id"] = a_ext.active_tray_id
updates["target_tray_id"] = a_ext.target_tray_id
updates["active_tray_state"] = a_ext.tray_state
updates["active_nozzle_temp"] = a_ext.temp
updates["active_nozzle_temp_target"] = a_ext.temp_target
updates["active_nozzle"] = a_ext.nozzle
else:
# otherwise process a single extruder printer update
updates["active_nozzle_temp"] = float(
p.get("nozzle_temper", base.active_nozzle_temp)
)
updates["active_nozzle_temp_target"] = int(
p.get("nozzle_target_temper", base.active_nozzle_temp_target)
)
updates["active_nozzle"] = (
updates["extruders"][0].nozzle
if updates.get("extruders")
else base.active_nozzle
)
updates["active_tray_id"] = int(ams_root.get("tray_now", base.active_tray_id))
if updates["active_tray_id"] == 255:
updates["active_tray_id"] = -1
updates["active_tray_state"] = TrayState.UNLOADED
elif aji.stage_id == 24:
updates["active_tray_state"] = TrayState.LOADING
elif aji.stage_id == 22:
updates["active_tray_state"] = TrayState.UNLOADING
else:
updates["active_tray_state"] = TrayState.LOADED
updates["active_ams_id"] = (
updates["active_tray_id"] >> 2
if updates["active_tray_id"] not in (254, 255)
else base.active_ams_id
)
if "active_tray_id" in updates:
updates["is_external_spool_active"] = updates["active_tray_id"] in [254, 255]
else:
updates["is_external_spool_active"] = False
if "active_tray_state" in updates:
updates["active_tray_state_name"] = updates["active_tray_state"].name
# GLOBAL METADATA & FANS
raw_exist = ams_root.get("ams_exist_bits", base.ams_exist_bits)
updates["ams_exist_bits"] = (
int(raw_exist, 16) if isinstance(raw_exist, str) else int(raw_exist)
)
updates["ams_connected_count"] = bin(updates["ams_exist_bits"]).count("1")
updates["ams_status_raw"] = int(p.get("ams_status", base.ams_status_raw))
updates["ams_status_text"] = parseAMSStatus(updates["ams_status_raw"])
part_cooling_fan_speed_percent = -1
if not config.capabilities.has_chamber_temp:
part_cooling_fan_speed_percent = (
scaleFanSpeed(p.get("cooling_fan_speed"))
if p.get("cooling_fan_speed", -1) != -1
else -1
)
else:
part_cooling_fan_speed_percent = updates["climate"].zone_part_fan_percent
if part_cooling_fan_speed_percent == -1:
part_cooling_fan_speed_percent = base.climate.part_cooling_fan_speed_percent
updates["climate"].part_cooling_fan_speed_percent = part_cooling_fan_speed_percent
updates["climate"].part_cooling_fan_speed_target_percent = updates[
"climate"
].part_cooling_fan_speed_percent
heatbreak_fan_speed_percent = scaleFanSpeed(p.get("heatbreak_fan_speed", -1))
if heatbreak_fan_speed_percent == -1:
heatbreak_fan_speed_percent = base.climate.heatbreak_fan_speed_percent
updates["climate"].heatbreak_fan_speed_percent = heatbreak_fan_speed_percent
exhaust_fan_speed_percent = -1
if not config.capabilities.has_chamber_temp:
exhaust_fan_speed_percent = scaleFanSpeed(p.get("big_fan2_speed", -1))
else:
exhaust_fan_speed_percent = updates["climate"].zone_exhaust_percent
if exhaust_fan_speed_percent == -1:
exhaust_fan_speed_percent = base.climate.exhaust_fan_speed_percent
updates["climate"].exhaust_fan_speed_percent = exhaust_fan_speed_percent
aux_fan_speed_percent = -1
if not config.capabilities.has_chamber_temp:
aux_fan_speed_percent = scaleFanSpeed(p.get("big_fan1_speed", -1))
else:
aux_fan_speed_percent = updates["climate"].zone_aux_percent
if aux_fan_speed_percent == -1:
aux_fan_speed_percent = base.climate.aux_fan_speed_percent
updates["climate"].aux_fan_speed_percent = aux_fan_speed_percent
updates["wifi_signal_strength"] = p.get("wifi_signal", base.wifi_signal_strength)
# ERROR HANDLING
updates["print_error"] = int(p.get("print_error", base.print_error))
if updates["print_error"] != 0:
decoded_error = decodeError(updates["print_error"])
else:
decoded_error = {}
base.hms_errors = []
updates["hms_errors"] = decodeHMS(p.get("hms", base.hms_errors))
if decoded_error and decoded_error not in updates["hms_errors"]:
updates["hms_errors"].insert(0, decoded_error)
# capabilities mapped to BambuConfig
config.capabilities = new_caps
return replace(base, **updates)
dataclass
Python
ExtruderState(
id: int = 0,
temp: float = 0.0,
temp_target: int = 0,
info_bits: int = 0,
state: ExtruderInfoState = NO_NOZZLE,
status: ExtruderStatus = IDLE,
nozzle: NozzleCharacteristics = NozzleCharacteristics(),
active_tray_id: int = -1,
target_tray_id: int = -1,
tray_state: TrayState = LOADED,
assigned_to_ams_id: int = -1,
)
State for an individual physical extruder toolhead.
Attributes:
| Name | Type | Description |
|---|---|---|
active_tray_id |
int
|
The active tray for this extruder. |
assigned_to_ams_id |
int
|
The id of the ams associated with this extruder |
id |
int
|
Physical ID. |
info_bits |
int
|
Raw bitmask. |
nozzle |
NozzleCharacteristics
|
Normalized nozzle characteristics. |
state |
ExtruderInfoState
|
Filament status. |
status |
ExtruderStatus
|
Op state. |
target_tray_id |
int
|
The target tray for this extruder. |
temp |
float
|
Current Temp. |
temp_target |
int
|
Target Temp. |
tray_state |
TrayState
|
The tray state of this extruder |
class-attribute
instance-attribute
The active tray for this extruder.
class-attribute
instance-attribute
The id of the ams associated with this extruder
class-attribute
instance-attribute
Python
nozzle: NozzleCharacteristics = field(default_factory=NozzleCharacteristics)
Normalized nozzle characteristics.
class-attribute
instance-attribute
Python
state: ExtruderInfoState = NO_NOZZLE
Filament status.
class-attribute
instance-attribute
The target tray for this extruder.
class-attribute
instance-attribute
Python
tray_state: TrayState = LOADED
The tray state of this extruder
dataclass
Python
NozzleCharacteristics(
material: NozzleType = UNKNOWN,
diameter_mm: float = 0.0,
flow: NozzleFlowType = UNKNOWN,
encoded_id: str = "",
telemetry_type_raw: str = "",
)
Normalized nozzle characteristics across telemetry and encoded identifiers.
This dataclass captures nozzle material, diameter, and optional flow-family
metadata. It is intended to provide a single canonical representation that
can be built from telemetry fields (for example nozzle_type,
nozzle_diameter) and optional encoded nozzle identifiers (for example
HS00-0.4).
Methods:
| Name | Description |
|---|---|
from_telemetry |
Build a |
to_identifier |
Return the best available encoded nozzle identifier. |
Attributes:
| Name | Type | Description |
|---|---|---|
diameter_mm |
float
|
Nozzle diameter in millimeters. |
encoded_id |
str
|
Raw encoded identifier such as |
flow |
NozzleFlowType
|
Optional nozzle flow family (standard/high-flow/TPU high-flow). |
material |
NozzleType
|
Canonical nozzle material/type parsed from telemetry or encoded ID. |
telemetry_type_raw |
str
|
Original raw |
class-attribute
instance-attribute
Nozzle diameter in millimeters.
class-attribute
instance-attribute
Raw encoded identifier such as HS00-0.4 when available.
class-attribute
instance-attribute
Python
flow: NozzleFlowType = UNKNOWN
Optional nozzle flow family (standard/high-flow/TPU high-flow).
class-attribute
instance-attribute
Python
material: NozzleType = UNKNOWN
Canonical nozzle material/type parsed from telemetry or encoded ID.
class-attribute
instance-attribute
Original raw nozzle_type string from telemetry payloads.
classmethod
Python
from_telemetry(
nozzle_type: str | None,
nozzle_diameter: str | float | int | None,
nozzle_id: str | None = None,
flow_type: NozzleFlowType = UNKNOWN,
) -> Self
Build a NozzleCharacteristics instance from telemetry fields.
Parameters
nozzle_type : str | None
Raw nozzle type string (for example hardened_steel).
nozzle_diameter : str | float | int | None
Raw nozzle diameter value in millimeters.
nozzle_id : str | None
Optional encoded identifier (for example HS00-0.4).
Returns
Self Normalized nozzle characteristics instance.
Source code in src/bpm/bambustate.py
Python
@classmethod
def from_telemetry(
cls,
nozzle_type: str | None,
nozzle_diameter: str | float | int | None,
nozzle_id: str | None = None,
flow_type: NozzleFlowType = NozzleFlowType.UNKNOWN,
) -> Self:
"""Build a `NozzleCharacteristics` instance from telemetry fields.
Parameters
----------
nozzle_type : str | None
Raw nozzle type string (for example `hardened_steel`).
nozzle_diameter : str | float | int | None
Raw nozzle diameter value in millimeters.
nozzle_id : str | None
Optional encoded identifier (for example `HS00-0.4`).
Returns
-------
Self
Normalized nozzle characteristics instance.
"""
material = parse_nozzle_type(nozzle_type)
diameter = float(nozzle_diameter) if nozzle_diameter is not None else 0.0
flow = flow_type
telemetry_type = (nozzle_type or "").strip()
if telemetry_type:
parsed_flow_from_type, parsed_material_from_type, _ = parse_nozzle_identifier(
telemetry_type
)
if parsed_flow_from_type != NozzleFlowType.UNKNOWN:
flow = parsed_flow_from_type
if (
material == NozzleType.UNKNOWN
and parsed_material_from_type != NozzleType.UNKNOWN
):
material = parsed_material_from_type
encoded = (nozzle_id or "").strip()
if encoded:
parsed_flow, parsed_material, _ = parse_nozzle_identifier(encoded)
if parsed_flow != NozzleFlowType.UNKNOWN:
flow = parsed_flow
if material == NozzleType.UNKNOWN and parsed_material != NozzleType.UNKNOWN:
material = parsed_material
return cls(
material=material,
diameter_mm=diameter,
flow=flow,
encoded_id=encoded,
telemetry_type_raw=(nozzle_type or ""),
)
Return the best available encoded nozzle identifier.
Returns the original encoded_id when present. Otherwise, it attempts
to build one from normalized material/flow/diameter values.
Source code in src/bpm/bambustate.py
Python
def to_identifier(self) -> str:
"""Return the best available encoded nozzle identifier.
Returns the original `encoded_id` when present. Otherwise, it attempts
to build one from normalized material/flow/diameter values.
"""
if self.encoded_id:
return self.encoded_id
if self.flow == NozzleFlowType.UNKNOWN:
return ""
return build_nozzle_identifier(self.flow, self.material, self.diameter_mm)