106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
"""打印机发送模块：通过局域网将 3MF 发送到 Bambu A1 并开始打印。"""
|
||
|
||
from __future__ import annotations
|
||
from pathlib import Path
|
||
from dataclasses import dataclass
|
||
|
||
|
||
@dataclass
|
||
class PrinterConfig:
|
||
"""打印机连接配置。"""
|
||
ip: str = "" # 打印机 IP 地址
|
||
access_code: str = "" # 局域网访问码
|
||
serial: str = "" # 打印机序列号
|
||
|
||
|
||
def check_printer_ready(config: PrinterConfig) -> bool:
|
||
"""检查打印机是否空闲可接收新任务。
|
||
|
||
返回 True 表示打印机空闲，False 表示正忙或离线。
|
||
"""
|
||
from bambulabs_api import Printer as BambuPrinter
|
||
from bambulabs_api.states_info import GcodeState
|
||
printer = BambuPrinter(config.ip, config.access_code, config.serial)
|
||
try:
|
||
printer.connect()
|
||
state = printer.get_state()
|
||
return state == GcodeState.IDLE
|
||
except Exception:
|
||
return False
|
||
finally:
|
||
printer.disconnect()
|
||
|
||
|
||
def _start_print_for_a1(printer, file_name: str, plate_number: int = 1) -> None:
|
||
"""发送打印机开始打印的 MQTT 指令，使用 A1 专用的 sdcard URL 格式。
|
||
|
||
bambulabs_api 的 start_print 方法硬编码了 ftp:/// 前缀，
|
||
这对 A1 不生效。A1 需要 file:///sdcard/ 前缀。
|
||
本函数绕过 start_print，直接构造正确的 MQTT 载荷。
|
||
"""
|
||
plate_location = f"Metadata/plate_{int(plate_number)}.gcode"
|
||
payload = {
|
||
"print": {
|
||
"command": "project_file",
|
||
"param": plate_location,
|
||
"file": file_name,
|
||
"bed_leveling": True,
|
||
"bed_type": "textured_plate",
|
||
"flow_cali": True,
|
||
"vibration_cali": True,
|
||
"url": f"file:///sdcard/{file_name}",
|
||
"layer_inspect": False,
|
||
"sequence_id": "10000000",
|
||
"use_ams": True,
|
||
"ams_mapping": [0],
|
||
}
|
||
}
|
||
# 调用名称改写后的私有方法发送原始 JSON 荷载
|
||
printer.mqtt_client._PrinterMQTTClient__publish_command(payload)
|
||
|
||
|
||
def send_to_printer(file_path: Path, config: PrinterConfig) -> bool:
|
||
"""将 3MF 文件上传至 A1 打印机并通过局域网开始打印。
|
||
|
||
发送前会检查打印机状态；正忙则拒绝发送。
|
||
⚠️ 仅支持本地局域网，电脑与打印机必须在同一网络下。
|
||
|
||
Args:
|
||
file_path: 要发送的 3MF 文件路径
|
||
config: 打印机连接配置
|
||
|
||
Returns:
|
||
True 表示成功
|
||
|
||
Raises:
|
||
ImportError: bambulabs_api 未安装
|
||
RuntimeError: 打印机正忙或上传/打印失败
|
||
"""
|
||
from bambulabs_api import Printer as BambuPrinter
|
||
from bambulabs_api.states_info import GcodeState
|
||
|
||
printer = BambuPrinter(config.ip, config.access_code, config.serial)
|
||
try:
|
||
printer.connect()
|
||
|
||
# 检查打印机是否空闲
|
||
state = printer.get_state()
|
||
if state != GcodeState.IDLE:
|
||
raise RuntimeError(
|
||
f"打印机当前状态为 {state.value}，"
|
||
f"无法开始新任务。请等待当前打印完成。"
|
||
)
|
||
|
||
file_name = file_path.name
|
||
with open(file_path, "rb") as f:
|
||
import io
|
||
result = printer.upload_file(io.BytesIO(f.read()), file_name)
|
||
if "226" not in result:
|
||
raise RuntimeError(f"上传文件失败: {result}")
|
||
|
||
# 使用 A1 专用的 MQTT 指令开始打印，而非内置的 start_print
|
||
_start_print_for_a1(printer, file_name, plate_number=1)
|
||
return True
|
||
finally:
|
||
printer.disconnect()
|