PyVISA串口连接仪器时如何获取命令执行控制台输出

问题场景 #

Python仪器控制开发中，需通过PyVISA以串口通信方式向仪器发送指令，触发仪器运行内部shell脚本，并将脚本在仪器自带Linux终端的执行日志完整回传到本地主机。

• 基于SSH+paramiko的实现可通过stdout.readlines()正常获取全量输出
• 基于PyVISA串口的实现仅单次调用read()时，无法拿到脚本完整控制台输出

参考：可正常运行的SSH实现代码 #

def create_connection_ssh(self):
        self.ssh_amp = paramiko.SSHClient()
        self.ssh_amp.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        self.ssh_amp.connect(self.ip_amp, username=self.username_amp, password=self.password_amp)

def amplifier_clock_set(self):
        stdin, stdout, stderr = self.ssh_amp.exec_command('../../opt/scripts/config/amplifier_clock_set.sh')

        day = datetime.today().strftime('%d')
        month = datetime.today().strftime('%m')
        year = datetime.today().strftime('%Y')
        hour = datetime.today().strftime('%H')
        minute = datetime.today().strftime('%M')
        weekday = datetime.today().strftime('%a')
        montname = datetime.today().strftime('%b')
        
        stdin.write(f'{day}\n')
        stdin.write(f'{month}\n')
        stdin.write(f'{year}\n')
        stdin.write(f'{hour}\n')
        stdin.write(f'{minute}\n')

        result = ' '.join(stdout.readlines())
        error = ' '.join(stderr.readlines())
        print('Output: '+result+error)

存在问题的初始串口实现代码 #

def create_connection_serial(self):

        RscManager = visa.ResourceManager()
        try:
            self.amplifier = RscManager.open_resource(self.amplifier_serial)
            self.amplifier.baud_rate = 115200
            self.amplifier.write_termination = '\n'
            self.amplifier.read_termination = '\n'
            time.sleep(0.5)

def amplifier_clock_set(self):
        self.amplifier.write('./amplifier_clock_set.sh')

        day = datetime.today().strftime('%d')
        month = datetime.today().strftime('%m')
        year = datetime.today().strftime('%Y')
        hour = datetime.today().strftime('%H')
        minute = datetime.today().strftime('%M')

        self.amplifier.write(f'{day}')
        self.amplifier.write(f'{month}')
        self.amplifier.write(f'{year}')
        self.amplifier.write(f'{hour}')
        self.amplifier.write(f'{minute}')

        print(self.amplifier.read()) # 此处无法获取完整的仪器终端输出

问题根因 #

• PyVISA串口默认的read()方法单次读取到预设终止符就直接返回，仅能获取第一行匹配终止规则的内容，不会自动持续读取直到脚本执行完成
• 串口是裸字节流传输，没有SSH封装的stdout/stderr流机制，不会自动等待命令执行结束、归集全量输出，需要手动实现读取循环和执行结束判定逻辑

修正实现 #

第一步：调整串口初始化配置 #

补充超时配置、清空缓冲区残留数据，避免读取阻塞或拿到无关内容：

def create_connection_serial(self):
    RscManager = visa.ResourceManager()
    try:
        self.amplifier = RscManager.open_resource(self.amplifier_serial)
        self.amplifier.baud_rate = 115200
        self.amplifier.write_termination = '\n'
        self.amplifier.read_termination = '\n'
        # 设置读取超时，单位毫秒，根据脚本最长执行时间调整，比如10秒设为10000
        self.amplifier.timeout = 10000
        time.sleep(0.5)
        # 清空连接建立时串口缓冲区残留的无关内容
        self.amplifier.flush(visa.constants.BufferOperation.discard_read_buffer)

第二步：实现循环读取+结束判定逻辑 #

两种可靠的结束判定方式二选一即可：

• 匹配Shell提示符（无侵入）：脚本执行完成后会回到Linux shell提示符，读到对应提示符即判定执行结束
• 自定义结束标记（兼容性更好）：执行脚本时追加输出自定义特殊字符串，读到该字符串即判定结束，不用适配不同设备的shell提示符

修正后的业务逻辑代码（采用自定义结束标记方案，无需适配提示符）：

def amplifier_clock_set(self):
    output_cache = []
    # 自定义结束标记，选不会在脚本正常输出里出现的字符串即可
    END_FLAG = "__SCRIPT_EXEC_FINISHED__"

    # 执行脚本时，脚本跑完自动输出结束标记
    self.amplifier.write(f'./amplifier_clock_set.sh && echo {END_FLAG}')

    day = datetime.today().strftime('%d')
    month = datetime.today().strftime('%m')
    year = datetime.today().strftime('%Y')
    hour = datetime.today().strftime('%H')
    minute = datetime.today().strftime('%M')

    self.amplifier.write(f'{day}')
    self.amplifier.write(f'{month}')
    self.amplifier.write(f'{year}')
    self.amplifier.write(f'{hour}')
    self.amplifier.write(f'{minute}')

    # 循环读取所有输出，直到碰到结束标记
    while True:
        try:
            line = self.amplifier.read()
            if END_FLAG in line:
                break
            output_cache.append(line)
        except visa.VisaIOError:
            # 触发超时异常时退出循环，可按需加超时错误处理
            break
    
    full_output = '\n'.join(output_cache)
    print('脚本执行输出：\n' + full_output)

注意事项 #

• 如果选择匹配Shell提示符的方案，需要先手动连接串口确认设备的实际提示符内容（比如常见的root@device:~#、[root@localhost ~]$），匹配错误会导致输出截断或读取卡住
• 超时时间需要根据脚本实际最长运行时间调整，避免脚本还没跑完就触发超时退出
• 不要用固定时长的time.sleep()等待输出，脚本执行时长受设备运行状态影响会波动，固定等待要么读不全内容，要么浪费不必要的等待时间

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