COM端口控制注射泵的循环运行一段时间后停止发送命令求助

核心问题分析 #

代码在长时间运行后出现异常停止、仅发送0流速命令，主要源于以下关键问题：

1. 串口命令重复发送导致解析混乱
   每个命令被重复调用3次ser.write，且前两次未添加命令终止符\r\n，注射泵无法正确解析这些冗余且格式错误的命令，长时间运行后会进入异常状态，最终停止响应正常流速指令。

2. 浮点数精度误判
   使用rate == 0判断流速是否为0，由于numpy生成的正弦值存在浮点精度误差（如理论值为0的点可能是-1.2246e-16），会导致误触发stop命令或进入错误的流速分支，破坏正常运行逻辑。

3. 串口缓冲区堆积阻塞通信
   代码未读取注射泵返回的响应数据，长时间运行后串口接收缓冲区被填满，导致新命令无法发送，系统陷入通信阻塞。

4. 异常捕获被禁用
   注释掉了SerialException捕获逻辑，串口通信中出现的小错误无法被处理，会逐步积累引发整体异常。

5. 无限循环写法不规范
   使用while i<=2:实现无限循环虽然能运行，但逻辑不清晰，建议改为while True:增强可读性。

修复后的代码实现 #

import numpy as np
import serial
import time
import logging

# 配置日志，便于追踪长时间运行状态
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 串口配置参数
SERIAL_CONFIG = {
    'port': 'COM3',
    'baudrate': 115200,
    'bytesize': serial.EIGHTBITS,
    'parity': serial.PARITY_NONE,
    'stopbits': 1,
    'xonxoff': 0,
    'rtscts': 0,
    'dsrdtr': 0,
    'timeout': 1
}

def init_serial():
    """初始化并返回串口连接"""
    ser = serial.Serial(**SERIAL_CONFIG)
    try:
        if ser.isOpen():
            ser.close()
        ser.open()
        logging.info(f"串口 {SERIAL_CONFIG['port']} 连接成功")
        return ser
    except serial.SerialException as e:
        logging.error(f"串口初始化失败: {str(e)}")
        raise

# 生成正弦流速序列
steps_per_cycle = 20
pump_rates = 0.5 * (np.sin(2 * np.pi * np.arange(steps_per_cycle) / steps_per_cycle))

def send_command(ser, cmd):
    """统一处理命令发送与响应读取"""
    try:
        # 发送带终止符的完整命令
        full_cmd = f"{cmd}\r\n"
        ser.write(full_cmd.encode('utf-8'))
        logging.info(f"发送命令: {cmd}")
        # 读取并记录响应，避免缓冲区堆积
        response = ser.read_all().decode('utf-8').strip()
        if response:
            logging.info(f"收到响应: {response}")
        return response
    except Exception as e:
        logging.error(f"命令发送失败: {str(e)}")
        raise

if __name__ == "__main__":
    ser = None
    try:
        ser = init_serial()
        # 真正的无限循环
        while True:
            for rate in pump_rates:
                # 用浮点近似判断代替精确等于，避免精度误差
                if np.isclose(rate, 0, atol=1e-6):
                    send_command(ser, 'stop')
                elif rate > 0:
                    # 注入流程
                    send_command(ser, f'irate {"%.5f" % rate} ml/min')
                    send_command(ser, 'irun')
                    send_command(ser, 'ctime')
                else:
                    # 抽取流程
                    send_command(ser, f'wrate {"%.5f" % -rate} ml/min')
                    send_command(ser, 'wrun')
                    send_command(ser, 'ctime')
                time.sleep(1)
    except Exception as e:
        logging.error(f"程序异常终止: {str(e)}")
    finally:
        # 确保串口最终被关闭
        if ser and ser.isOpen():
            ser.close()
            logging.info("串口已关闭")

额外排查建议 #

1. 核对注射泵手册：确认命令格式（如终止符是否为\r\n）、命令执行顺序是否符合泵的要求。
2. 监控系统资源：长时间运行时查看Python进程的内存、CPU占用，排查是否存在内存泄漏。
3. 添加心跳检测：每隔若干周期发送查询命令（如查询当前流速），确认泵的状态，异常时尝试重新初始化串口。
4. 缩短测试周期：临时调小steps_per_cycle值，加速复现问题，便于快速验证修复效果。

内容的提问来源于stack exchange，提问作者pedwards1