以Cura插件USBPrinting为例解读3D打印机USB联机打印实现方案

众所周知，对3D打印机感兴趣的小伙伴来说，都清楚Cura是3D打印机的切片软件，它的UI部分是基于QT来开发的。而Cura中很多功能其实是基于插件的形式来开发，其中，用于实现Cura的USB转串口联机打印的逻辑就是一个插件，它是使用Python语言来实现的，具体代码位于：

https://github.com/Ultimaker/Cura/tree/main/plugins/USBPrinting

 

之前我也做了一些3D打印机的联机打印的开源项目：

Anycubic Vyper 3D打印机串口屏改造开源项目之串口屏项目启动篇(一)

Anycubic Vyper 3D打印机串口屏改造开源项目之QT温度曲线显示(二)

而我前阵子参加开放原子基金会组织的开发者成长激励活动的作品其实也算是联机打印的一种，只是实现的方式不同而已罢了：

开发者成长激励计划-基于TencentOS Tiny FDM 3D打印机云控制系统方案

说到Cura中的USB转串口联机打印，核心逻辑可以梳理下为以下几点：

(1)查找串口设备列表并获取对应的打印机设备端口号，这部分的代码是在USBPrinterOutputDeviceManager.py这个文件里实现的。

(2)设置串口设备参数并连接设备、启动更新线程来处理串口数据接收

具体的代码实现如下：

 

def connect(self):
self._firmware_name = None  # after each connection ensure that the firmware name is removed

if self._baud_rate is None:
  if self._use_auto_detect:
     auto_detect_job = AutoDetectBaudJob(self._serial_port)
     auto_detect_job.start()
     auto_detect_job.finished.connect(self._autoDetectFinished)
  return
  if self._serial is None:
     try:
     # 设置串口参数
        self._serial = Serial(str(self._serial_port), self._baud_rate, timeout=self._timeout, writeTimeout=self._timeout)
     except SerialException:
         Logger.warning("An exception occurred while trying to create serial connection.")
        return
     except OSError as e:
         Logger.warning("The serial device is suddenly unavailable while trying to create a serial connection: {err}".format(err = str(e)))
         return
     CuraApplication.getInstance().globalContainerStackChanged.connect(self._onGlobalContainerStackChanged)
     self._onGlobalContainerStackChanged()
     self.setConnectionState(ConnectionState.Connected)
     # 启动更新线程
     self._update_thread.start()

 

(3)启动更新任务线程，更新任务线程的作用是处理以下几件事情：

以readline()的方式去接收打印机回复的数据，然后处理数据，例如接收到了ok或者温度信息等。

处理接收的数据，并接着发下一条Gcode指令，直到没有得发为止。

处理打印过程中发生的异常事件

发送M105获取温度命令，这里Cura是做了一些处理的，发送该条命令的前提是打印机不处于忙状态并且温度到了设定的固件超时时间才会进行发送。Cura的超时设置为3s。

Gcode重发机制的实现

具体的代码实现如下：

 

# 线程_update_thread->更新任务函数的实现
def _update(self):
  while self._connection_state == ConnectionState.Connected and self._serial is not None:
    try:
       line = self._serial.readline()
    except:
       continue

    # 获取固件信息
    # 如果是Marlin，则会输出类似如下所示的信息
    # FIRMWARE_NAME:Marlin 1.1.0 ....
    if not self._firmware_name_requested:
           self._firmware_name_requested = True
           self.sendCommand("M115")

    # 获取FIRMWARE_NAME并保存起来
    if b"FIRMWARE_NAME:" in line:
           self._setFirmwareName(line)

    # time()是获取时间戳，以秒作为时间间隔，这里的timeout是3，也就意味着，Cura发送获取温度的条件是：
    # 1、当前的打印机不处于忙状态
    # 2、超时，这里设置的时间是大于3s
    # 以上两个条件需要同时满足
    if self._last_temperature_request is None or time() > self._last_temperature_request + self._timeout:
          self.sendCommand("M105")
          self._last_temperature_request = time()

    # 使用正则表达式获取由打印机端上报的温度事件，其中T:开头的数据代表喷头温度，B:开头的数据代表热床温度
    if re.search(b"[B|Td*]: ?d+.?d*", line):  # Temperature message. 'T:' for extruder and 'B:' for bed
       extruder_temperature_matches = re.findall(b"T(d*): ?(d+.?d*)s*/?(d+.?d*)?", line)
    # Update all temperature values
    # 获取喷头当前/目标温度值并更新到前端显示
       matched_extruder_nrs = []
       for match in extruder_temperature_matches:
           extruder_nr = 0
           if match[0] != b"":
              extruder_nr = int(match[0])
           if extruder_nr in matched_extruder_nrs:
              continue
           matched_extruder_nrs.append(extruder_nr)
           if extruder_nr >= len(self._printers[0].extruders):
              Logger.log("w", "Printer reports more temperatures than the number of configured extruders")
              continue
              extruder = self._printers[0].extruders[extruder_nr]
           if match[1]:
              extruder.updateHotendTemperature(float(match[1]))
           if match[2]:
              extruder.updateTargetHotendTemperature(float(match[2]))

           # 获取热床当前/目标温度值并更新到前端显示
           bed_temperature_matches = re.findall(b"B: ?(d+.?d*)s*/?(d+.?d*)?", line)
           if bed_temperature_matches:
              match = bed_temperature_matches[0]
           if match[0]:
              self._printers[0].updateBedTemperature(float(match[0]))
           if match[1]:
              self._printers[0].updateTargetBedTemperature(float(match[1]))

            # 空行表示固件空闲
            # 多个空行可能意味着固件和 Cura 正在等待
            # 因为错过了“ok”，所以我们跟踪空行
            # 因为ok可能丢掉了，所以我们需要将空行记录下来
            if line == b"":
                # An empty line means that the firmware is idle
                # Multiple empty lines probably means that the firmware and Cura are waiting
                # for each other due to a missed "ok", so we keep track of empty lines
                self._firmware_idle_count += 1
            else:
                self._firmware_idle_count = 0

            # 检查到ok字串或者_firmware_idle_count > 1
            if line.startswith(b"ok") or self._firmware_idle_count > 1:
                # 此时打印机忙状态解除
                self._printer_busy = False
                # 设置接收事件为True
                self._command_received.set()
                # 如果当前命令队列不为空，则从队列取出一条命令往打印机串口继续发送
                if not self._command_queue.empty():
                    self._sendCommand(self._command_queue.get())
                # 如果处于正在打印中，则继续发送下一条Gcode命令
                # 如果此时暂停标志生效，则什么事情都不干
                elif self._is_printing:
                    if self._paused:
                        pass  # Nothing to do!
                    else:
                        self._sendNextGcodeLine()

            # 如果匹配到Marlin回复了"echo:busy"子串时，则设置打印机为忙状态
            if line.startswith(b"echo"):
                self._printer_busy = True

            # 如果在打印中接收到'!!'，则表示打印机发出致命错误，这个时候需要直接取消打印
            if self._is_printing:
                if line.startswith(b'!!'):
                    Logger.log('e', "Printer signals fatal error. Cancelling print. {}".format(line))
                    self.cancelPrint()
                # 如果在打印中接收到"resend"或者"rs"这样的字符串,则可以通过 Resend、resend 或 rs 请求重新发送。
                elif line.lower().startswith(b"resend") or line.startswith(b"rs"):
                    # A resend can be requested either by Resend, resend or rs.
                    try:
                        self._gcode_position = int(line.replace(b"N:", b" ").replace(b"N", b" ").replace(b":", b" ").split()[-1])
                    except:
                        if line.startswith(b"rs"):
                            # In some cases of the RS command it needs to be handled differently.
                            self._gcode_position = int(line.split()[1])

 

在USB转串口联机打印中，也实现了一些打印的基本业务，待后续分析和开源作品分享。