bambustate provides a unified, thread-safe representation of a Bambu Lab printer's
operational state, synchronized via MQTT telemetry.
Classes:
| Name | Description |
|---|---|
AMSDryerState |
State of an AMS unit's filament dryer (AMS 2 Pro and AMS HT only). |
AMSUnitState |
State information about an individual AMS unit. |
BambuClimate |
Contains all climate related attributes |
BambuState |
Representation of the Bambu printer state synchronized via MQTT. |
ExtensionToolState |
State of the toolhead extension-tool interface, parsed from |
ExtruderState |
State for an individual physical extruder toolhead. |
NozzleCharacteristics |
Normalized nozzle characteristics across telemetry and encoded identifiers. |
Attributes:
| Name | Type | Description |
|---|---|---|
DRYER_MODELS |
AMS models that can dry filament; only these carry an |
module-attribute
AMS models that can dry filament; only these carry an AMSDryerState.
dataclass
AMSDryerState(
state: AMSHeatingState = OFF,
sub_status: AMSDryerSubStatus = OFF,
fan1_status: AMSDryerFanStatus = OFF,
fan2_status: AMSDryerFanStatus = OFF,
remaining_minutes: int = 0,
temp_target: int = -1,
duration_target_hours: int = -1,
filament_target: str = "",
refusals: list[int] = list(),
refusal_message: str = "",
fail_code: str = "",
fail_message: str = "",
fail_count: int = 0,
)
State of an AMS unit's filament dryer (AMS 2 Pro and AMS HT only).
Measured values are plain; values the printer was ordered to use end in
_target.
Attributes:
| Name | Type | Description |
|---|---|---|
duration_target_hours |
int
|
Duration the running dry was ordered for, in hours ( |
fail_code |
str
|
HMS code of the last refused drying command reply (e.g. |
fail_count |
int
|
Counts refused drying command replies, so a consumer can tell a new |
fail_message |
str
|
Text for |
fan1_status |
AMSDryerFanStatus
|
Drying fan 1 status (bits 18-19 of |
fan2_status |
AMSDryerFanStatus
|
Drying fan 2 status (bits 20-21 of |
filament_target |
str
|
Filament type the running dry was ordered for ( |
refusal_message |
str
|
BambuStudio's text for |
refusals |
list[int]
|
Why a dry cannot start now: raw |
remaining_minutes |
int
|
Minutes left in the running dry ( |
state |
AMSHeatingState
|
The computed state of the dryer's heater. |
sub_status |
AMSDryerSubStatus
|
Drying phase (bits 22-25 of |
temp_target |
int
|
Temperature the running dry was ordered at, in °C ( |
class-attribute
instance-attribute
Duration the running dry was ordered for, in hours (dry_setting); -1
when idle.
class-attribute
instance-attribute
HMS code of the last refused drying command reply (e.g. HMS_0500-C04B).
Cleared by the next accepted one.
class-attribute
instance-attribute
Counts refused drying command replies, so a consumer can tell a new refusal from an old one.
class-attribute
instance-attribute
Text for fail_code, from HMS_STATUS.
class-attribute
instance-attribute
fan1_status: AMSDryerFanStatus = OFF
Drying fan 1 status (bits 18-19 of ams_info).
class-attribute
instance-attribute
fan2_status: AMSDryerFanStatus = OFF
Drying fan 2 status (bits 20-21 of ams_info).
class-attribute
instance-attribute
Filament type the running dry was ordered for (dry_setting); empty
when idle.
class-attribute
instance-attribute
BambuStudio's text for refusals; empty when a dry can start.
class-attribute
instance-attribute
Why a dry cannot start now: raw dry_sf_reason values (AMSDryerRefusal).
Empty when a dry can start; [6] while one runs. Unknown values are kept.
class-attribute
instance-attribute
Minutes left in the running dry (dry_time).
class-attribute
instance-attribute
state: AMSHeatingState = OFF
The computed state of the dryer's heater.
class-attribute
instance-attribute
sub_status: AMSDryerSubStatus = OFF
Drying phase (bits 22-25 of ams_info).
dataclass
AMSUnitState(
ams_id: int,
chip_id: str = "",
model: AMSModel = UNKNOWN,
temp_actual: float = 0.0,
humidity_index: int = 0,
humidity_raw: int = 0,
ams_info: int = 0,
dryer: AMSDryerState | None = None,
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. |
dryer |
AMSDryerState | None
|
The unit's dryer. |
humidity_index |
int
|
Humidity index. |
humidity_raw |
int
|
Raw humidity. |
model |
AMSModel
|
|
temp_actual |
float
|
Actual temp. |
tray_exists |
list[bool]
|
Slot presence, one entry per slot: 4, or 1 for an AMS HT. |
class-attribute
instance-attribute
assigned_to_extruder: ActiveTool = SINGLE_EXTRUDER
Target tool computed from raw_extruder_id
class-attribute
instance-attribute
dryer: AMSDryerState | None = None
The unit's dryer. None with model UNKNOWN means the unit has not
reported yet; None with any other model means it has no dryer.
dataclass
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,
enhanced_cooling_fan_target_percent: int = 0,
)
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. |
enhanced_cooling_fan_target_percent |
int
|
STICKY commanded state for the Toolhead Enhanced Cooling Fan (M106 P9). |
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
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
STICKY commanded state for the Toolhead Enhanced Cooling Fan (M106 P9).
UNLIKE every other fan field, this is NEVER updated from telemetry — the
printer publishes no run state for this fan (verified fw 01.03.00.00), so
the last commanded value is the only state that exists. Written by
BambuPrinter.set_enhanced_cooling_fan_speed_target_percent(); reset to 0
by the parser only when the extension tool leaves the MOUNTED state
(unplugged/removed — the commanded value no longer describes reality).
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
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 = "",
filament_step: FilamentStep = IDLE,
filament_step_type: FilamentStepType = LOAD,
filament_step_name: str = name,
filament_step_text: str = "",
hw_switch_state: int = -1,
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(),
extension_tool: ExtensionToolState = ExtensionToolState(),
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 |
extension_tool |
ExtensionToolState
|
Toolhead extension-tool interface state (Enhanced Cooling Fan / cutting / laser). |
extruders |
list[ExtruderState]
|
Extruder details. |
filament_step |
FilamentStep
|
The running load or unload step ( |
filament_step_name |
str
|
|
filament_step_text |
str
|
Bambu Studio's label for |
filament_step_type |
FilamentStepType
|
Whether the running change is an AMS load, an unload or an external-holder load. |
gcode_state |
str
|
Execution state. |
hms_errors |
list[dict]
|
HMS list. |
hw_switch_state |
int
|
The printer's filament-at-extruder switch: 1 filament, 0 none, -1 not reported. |
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
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
active_tool: ActiveTool = SINGLE_EXTRUDER
Active toolhead.
class-attribute
instance-attribute
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
ams_units: list[AMSUnitState] = field(default_factory=list)
Unit details.
class-attribute
instance-attribute
climate: BambuClimate = field(default_factory=BambuClimate)
Contains all climate related attributes
class-attribute
instance-attribute
extension_tool: ExtensionToolState = field(default_factory=ExtensionToolState)
Toolhead extension-tool interface state (Enhanced Cooling Fan / cutting / laser).
class-attribute
instance-attribute
extruders: list[ExtruderState] = field(default_factory=list)
Extruder details.
class-attribute
instance-attribute
filament_step: FilamentStep = IDLE
The running load or unload step (parseFilamentStep); IDLE when none runs.
class-attribute
instance-attribute
filament_step as a string.
class-attribute
instance-attribute
Bambu Studio's label for filament_step; empty when idle.
class-attribute
instance-attribute
filament_step_type: FilamentStepType = LOAD
Whether the running change is an AMS load, an unload or an external-holder load.
class-attribute
instance-attribute
HMS list.
class-attribute
instance-attribute
The printer's filament-at-extruder switch: 1 filament, 0 none, -1 not reported.
class-attribute
instance-attribute
Ext spool flag.
class-attribute
instance-attribute
spools: list[BambuSpool] = field(default_factory=list)
All spools associated with this printer
class-attribute
instance-attribute
Wi-Fi signal strength in dBm
classmethod
fromJson(data: dict[str, Any], printer: BambuPrinter) -> BambuState
Parses root MQTT payloads into a hierachical BambuState object.
Source code in src/bpm/bambustate.py
@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":
ams_id = p.get("ams_id", -1)
ams = next((u for u in base.ams_units if u.ams_id == ams_id), None)
dryer = ams.dryer if ams else None
result = str(p.get("result", "")).lower()
if dryer and result == "success":
# a start (mode 1) orders a temperature; a stop (mode 0) leaves none
on = int(p.get("mode", 1)) == 1
dryer.temp_target = int(p.get("temp", 0)) if on else -1
dryer.fail_code = ""
dryer.fail_message = ""
elif dryer and result == "fail":
err = decodeError(int(p.get("err_code", 0) or 0))
dryer.fail_code = err.get("code", "")
dryer.fail_message = (
err.get("msg", "") or "The printer refused the drying command."
)
dryer.fail_count += 1
# A refused command reply (e.g. a load while the AMS dries) is an error like any HMS
# entry: it joins hms_errors as a `command_error` and stays there until that command
# is sent again or the printer replies that it accepted it. `timestamp` tells a
# repeat refusal of the same code from the one before it.
command_errors = [e for e in base.hms_errors if e.get("type") == "command_error"]
if p.get("command", "") == "ams_change_filament":
command_errors = [
e for e in command_errors if e.get("command") != "ams_change_filament"
]
if str(p.get("result", "")).lower() == "fail":
err_code = int(p.get("err_code", 0) or 0)
err = decodeError(err_code) if err_code else {}
command_errors.append(
{
**err,
"code": err.get("code", ""),
"msg": err.get("msg", "")
or "The printer refused the filament change.",
"url": err.get("url", ""),
"type": "command_error",
"command": "ams_change_filament",
"ams_id": p.get("ams_id", -1),
"timestamp": int(time.time()),
}
)
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
caps["has_auto_switch_filament_support"] = 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
)
# EXTENSION TOOL (device.ext_tool — Enhanced Cooling Fan / cutting / laser)
ext_tool_root = device.get("ext_tool", {})
if ext_tool_root:
type_raw = str(ext_tool_root.get("type", ""))
new_extension_tool = ExtensionToolState(
tool_type=ExtensionToolType(type_raw),
mount_state=ExtensionToolMountState(
int(ext_tool_root.get("mount_3d", -1))
),
calibration_raw=int(ext_tool_root.get("calib", -1)),
type_raw=type_raw,
)
updates["extension_tool"] = new_extension_tool
# Sticky-state invalidation: the enhanced cooling fan publishes no
# run-state telemetry, so its commanded value survives every parse —
# EXCEPT when the tool leaves MOUNTED (unplugged/removed), where the
# commanded value no longer describes reality.
if (
base.extension_tool.mount_state == ExtensionToolMountState.MOUNTED
and new_extension_tool.mount_state != ExtensionToolMountState.MOUNTED
):
updates["climate"].enhanced_cooling_fan_target_percent = 0
else:
updates["extension_tool"] = base.extension_tool
# 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)
# only an extruder with the same id carries its state over: a partial
# first frame leaves a single-extruder placeholder (id -1, UNLOADED)
prev_ext = next((e for e in base.extruders if e.id == ext.id), None)
if prev_ext is not None:
base_tray_state = prev_ext.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_tray_state
new_extruders.append(ext)
elif len(base.extruders) > 1:
# A partial frame on a multi-extruder printer (a gcode_line ACK, for
# example) carries no device.extruder block at all. There is nothing
# to re-derive, so keep the last known extruders rather than
# collapsing them into the single-extruder model below.
new_extruders = list(base.extruders)
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)
# an unlisted serial prefix must not forget a model already known
model = getAMSModelBySerial(u.chip_id)
if model != AMSModel.UNKNOWN:
u.model = model
_fit_ams_unit_to_model(u)
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))
_hIdx = int(float(ams_u.get("humidity", 0)))
if 1 <= _hIdx <= 5:
u.humidity_index = _hIdx
_hRaw = int(float(ams_u.get("humidity_raw", 0)))
if 1 <= _hRaw <= 100:
u.humidity_raw = _hRaw
# resolve the model before any dryer field, so a unit's first
# report keeps its dryer values
p_ams = parseAMSInfo(ams_u["info"]) if "info" in ams_u else None
if p_ams and u.model == AMSModel.UNKNOWN:
u.model = p_ams["ams_type"]
_fit_ams_unit_to_model(u)
dryer = u.dryer
if dryer:
dryer.remaining_minutes = int(
float(ams_u.get("dry_time", dryer.remaining_minutes))
)
if isinstance(ams_u.get("dry_sf_reason"), list):
dryer.refusals = [int(r) for r in ams_u["dry_sf_reason"]]
dryer.refusal_message = dryerRefusalMessage(dryer.refusals)
dry_setting = ams_u.get("dry_setting")
if isinstance(dry_setting, dict):
dryer.temp_target = int(dry_setting.get("dry_temperature", -1))
dryer.duration_target_hours = int(dry_setting.get("dry_duration", -1))
dryer.filament_target = str(dry_setting.get("dry_filament", ""))
if p_ams:
u.ams_info = int(ams_u["info"], 16)
if dryer:
dryer.state = p_ams["heater_state"]
dryer.fan1_status = p_ams["dry_fan1_status"]
dryer.fan2_status = p_ams["dry_fan2_status"]
dryer.sub_status = p_ams["dry_sub_status"]
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 has a single slot
u.tray_exists = [bool((eb >> shift) & 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"])
updates["hw_switch_state"] = int(p.get("hw_switch_state", base.hw_switch_state))
step, step_type = parseFilamentStep(
updates["ams_status_raw"],
updates.get("target_tray_id", base.target_tray_id),
updates["hw_switch_state"] != 0,
base.filament_step,
)
updates["filament_step"] = step
updates["filament_step_type"] = step_type
updates["filament_step_name"] = step.name
updates["filament_step_text"] = FILAMENT_STEP_TEXT[step]
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"])
decoded_error["actions"] = hmsActions(
config.serial_number, updates["print_error"]
)
else:
decoded_error = {}
# A frame with no "hms" key (a gcode_line ACK, a partial push_status) says nothing about
# HMS, so the previous entries stay. Only the print_error entry is re-derived every frame.
if "hms" in p:
raw_hms = p["hms"]
else:
raw_hms = [
e
for e in base.hms_errors
if e.get("type") not in ("device_error", "command_error")
]
updates["hms_errors"] = decodeHMS(raw_hms)
if decoded_error and decoded_error not in updates["hms_errors"]:
updates["hms_errors"].insert(0, decoded_error)
updates["hms_errors"].extend(command_errors)
# capabilities mapped to BambuConfig
config.capabilities = new_caps
return replace(base, **updates)
dataclass
ExtensionToolState(
tool_type: ExtensionToolType = NONE,
mount_state: ExtensionToolMountState = UNKNOWN,
calibration_raw: int = -1,
type_raw: str = "",
)
State of the toolhead extension-tool interface, parsed from
device.ext_tool telemetry (H2-series). The Toolhead Enhanced Cooling
Fan, cutting module, and laser module all report through this block.
Attributes:
| Name | Type | Description |
|---|---|---|
calibration_raw |
int
|
Raw calib value (0=none, 1=first, 2=mount per BambuStudio CalibState). |
is_enhanced_cooling_fan_mounted |
bool
|
True when the Toolhead Enhanced Cooling Fan is mounted with its cable connected. |
mount_state |
ExtensionToolMountState
|
Mount state (mount_3d). MOUNTED is the only healthy attached state. |
tool_type |
ExtensionToolType
|
Attached tool type. NOTE: on cable loss the printer reports type as |
type_raw |
str
|
Verbatim type code from telemetry, preserved for unrecognized tools. |
class-attribute
instance-attribute
Raw calib value (0=none, 1=first, 2=mount per BambuStudio CalibState).
property
True when the Toolhead Enhanced Cooling Fan is mounted with its cable connected.
class-attribute
instance-attribute
mount_state: ExtensionToolMountState = UNKNOWN
Mount state (mount_3d). MOUNTED is the only healthy attached state.
class-attribute
instance-attribute
tool_type: ExtensionToolType = NONE
Attached tool type. NOTE: on cable loss the printer reports type as empty ("" = NONE) with mount_state NO_CABLE — key presence checks on mount_state, not tool_type alone.
dataclass
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
nozzle: NozzleCharacteristics = field(default_factory=NozzleCharacteristics)
Normalized nozzle characteristics.
class-attribute
instance-attribute
state: ExtruderInfoState = NO_NOZZLE
Filament status.
class-attribute
instance-attribute
The target tray for this extruder.
class-attribute
instance-attribute
tray_state: TrayState = LOADED
The tray state of this extruder
dataclass
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
flow: NozzleFlowType = UNKNOWN
Optional nozzle flow family (standard/high-flow/TPU high-flow).
class-attribute
instance-attribute
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
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
@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
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)