4、数据采集基础:使用pyvisa与仪器通信、读取电压/电流/温度数据、数据格式解析

各位同学,欢迎来到第四章。前面我们聊了电池化成的工艺流程,也搭建了软件环境。今天终于要动真格的了——让电脑和仪器真正「说上话」。

说实话,我当年第一次用 pyvisa 连上仪器时,看到屏幕上跳出第一组电压数据,那种兴奋感至今还记得。但别急,这条路没那么顺。我踩过的坑,今天都给你标出来。

4.1 为什么是 pyvisa?

你可能会问:市面上那么多通信库,为什么偏偏选 pyvisa?

原因很简单:它是工业界的通用语言。无论你用的是是德科技、泰克、吉时利还是国产的艾德克斯,只要仪器支持 VISA 标准,pyvisa 就能一把抓。

我在项目中遇到过最头疼的事——客户给了三台不同品牌的电源,要求统一采集。用 pyvisa,一套代码全搞定。换做其他库,你得写三套驱动。

核心概念:VISA(Virtual Instrument Software Architecture)是仪器通信的抽象层。它把 GPIB、USB、以太网、串口这些物理接口统一成一个编程接口。你写代码时不用关心仪器到底插在哪个口上。

4.2 环境准备与连接测试

先别急着写代码。咱们一步步来。

4.2.1 安装 pyvisa 和后端

打开终端,敲两行命令:

pip install pyvisa
pip install pyvisa-py

嗯,这里要注意:pyvisa-py 是纯 Python 的后端实现。如果你装了 NI-VISA(National Instruments 的官方驱动),也可以用它。我个人习惯用 pyvisa-py,因为不用装几百兆的 NI 软件,轻量多了。

4.2.2 查找仪器

仪器连上电脑后,第一件事是确认它被系统识别了。写个脚本扫一扫:

import pyvisa

rm = pyvisa.ResourceManager()
resources = rm.list_resources()
print("找到的仪器资源:")
for res in resources:
    print(res)

跑完这段,你会看到类似这样的输出:

找到的仪器资源:
USB0::0x2A8D::0x1301::MY59001234::INSTR
TCPIP0::192.168.1.100::inst0::INSTR
ASRL3::INSTR

USB 开头的就是 USB 连接,TCPIP 是网口,ASRL 是串口。如果你啥也没看到——别慌,先检查驱动和线缆。我当年第一次用 USB 转 GPIB 线,折腾了半小时才发现驱动没装对。

小技巧:如果 list_resources() 返回空列表,试试指定后端:ResourceManager('@py')。有时候默认后端找不到设备。

4.3 读取电压、电流、温度数据

找到仪器了,接下来就是真正的数据采集。不同仪器的 SCPI 命令不同,但套路是一样的。

4.3.1 建立连接与发送命令

import pyvisa

rm = pyvisa.ResourceManager()
inst = rm.open_resource('USB0::0x2A8D::0x1301::MY59001234::INSTR')

# 查询仪器身份
idn = inst.query('*IDN?')
print(f"仪器型号:{idn}")

# 读取电压
voltage = inst.query('MEAS:VOLT?')
print(f"当前电压:{voltage} V")

# 读取电流
current = inst.query('MEAS:CURR?')
print(f"当前电流:{current} A")

# 读取温度(有些仪器支持)
try:
    temp = inst.query('MEAS:TEMP?')
    print(f"当前温度:{temp} °C")
except:
    print("该仪器不支持温度读取")

看到没?核心就两个函数:query() 是「发命令+读回复」,write() 是「只发命令不读回复」。我建议能用 query 就别用 write+read 分开写,少出错。

4.3.2 连续采集与时间戳

单次读取没意思,我们要的是连续采集。下面这个例子每 1 秒读一次,持续 10 秒:

import time
import csv
from datetime import datetime

data = []
for i in range(10):
    v = float(inst.query('MEAS:VOLT?'))
    c = float(inst.query('MEAS:CURR?'))
    t = datetime.now().strftime('%H:%M:%S.%f')[:-3]
    data.append([t, v, c])
    print(f"{t} - 电压:{v:.4f}V  电流:{c:.4f}A")
    time.sleep(1)

# 保存到 CSV
with open('采集数据.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    writer.writerow(['时间', '电压(V)', '电流(A)'])
    writer.writerows(data)

我曾经在产线上遇到过一个问题:采集频率太高,仪器响应不过来,数据全是重复的。后来加了 time.sleep(0.5) 才解决。你想想看,仪器也是要喘气的。

注意:不要用 time.sleep(0) 来追求极限速度。大多数仪器每秒能稳定响应 10-20 次查询。超过这个频率,要么丢数据,要么仪器死机。

4.4 数据格式解析

仪器返回的数据,可不是每次都那么规整。我见过最离谱的——返回字符串里带换行、带空格、甚至带单位。所以解析这步,必须做。

4.4.1 常见格式与处理方法

返回格式 示例 处理方法
纯数字 4.2356 float(value.strip())
带单位 4.2356 V float(value.split()[0])
科学计数法 +4.2356E+00 float(value) 自动识别
多值返回 4.2356,0.1234,25.3 [float(x) for x in value.split(',')]
带换行符 4.2356\n value.strip()

写个健壮的解析函数,一劳永逸:

def parse_measurement(raw_value):
    """解析仪器返回的测量值,返回浮点数"""
    # 去除首尾空白和换行
    cleaned = raw_value.strip()
    # 去掉可能的单位(如 V, A, °C)
    parts = cleaned.split()
    if parts:
        try:
            return float(parts[0])
        except ValueError:
            # 如果是科学计数法,直接转
            return float(cleaned)
    return None

4.4.2 批量解析与异常处理

实际项目中,你不可能每次都手动解析。我习惯写一个采集类,把解析逻辑封装进去:

class BatteryChannel:
    def __init__(self, instrument, channel=1):
        self.inst = instrument
        self.ch = channel
    
    def read_voltage(self):
        raw = self.inst.query(f'MEAS:VOLT? (@{self.ch})')
        return parse_measurement(raw)
    
    def read_current(self):
        raw = self.inst.query(f'MEAS:CURR? (@{self.ch})')
        return parse_measurement(raw)
    
    def read_temperature(self):
        raw = self.inst.query(f'MEAS:TEMP? (@{self.ch})')
        return parse_measurement(raw)
    
    def read_all(self):
        """一次读取所有数据"""
        return {
            'voltage': self.read_voltage(),
            'current': self.read_current(),
            'temperature': self.read_temperature()
        }

用起来就清爽多了:

ch1 = BatteryChannel(inst, 1)
data = ch1.read_all()
print(f"通道1 - 电压:{data['voltage']:.4f}V  电流:{data['current']:.4f}A  温度:{data['temperature']:.1f}°C")

避坑指南:我曾经遇到过仪器返回 +4.2356E+00 这种格式,直接 float() 没问题。但有些老仪器会返回 4.2356E+00V,带单位!这时候 float() 会报错。所以先 split 再转,更安全。

4.5 实战:完整的数据采集流程

最后,咱们把今天学的串起来。写一个完整的采集脚本,包含连接、采集、解析、保存、异常处理:

import pyvisa
import time
import csv
from datetime import datetime

def collect_data(ip_address, duration_sec=60, interval=1):
    """
    从指定 IP 的仪器采集数据
    :param ip_address: 仪器 IP 地址
    :param duration_sec: 采集时长(秒)
    :param interval: 采集间隔(秒)
    """
    rm = pyvisa.ResourceManager()
    try:
        inst = rm.open_resource(f'TCPIP0::{ip_address}::inst0::INSTR')
        inst.timeout = 5000  # 5秒超时
        print(f"已连接:{inst.query('*IDN?').strip()}")
    except Exception as e:
        print(f"连接失败:{e}")
        return
    
    data = []
    start_time = time.time()
    samples = int(duration_sec / interval)
    
    print(f"开始采集,共 {samples} 个样本点...")
    for i in range(samples):
        try:
            v = float(inst.query('MEAS:VOLT?'))
            c = float(inst.query('MEAS:CURR?'))
            t = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
            data.append([t, v, c])
            print(f"[{i+1}/{samples}] {t}  V={v:.4f}  I={c:.4f}")
        except Exception as e:
            print(f"第 {i+1} 次采集异常:{e}")
            data.append([datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3], None, None])
        time.sleep(interval)
    
    # 保存数据
    filename = f'化成数据_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv'
    with open(filename, 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(['时间', '电压(V)', '电流(A)'])
        writer.writerows(data)
    print(f"数据已保存至:{filename}")
    
    inst.close()
    return data

# 使用示例
if __name__ == '__main__':
    collect_data('192.168.1.100', duration_sec=30, interval=2)

这个脚本,我在产线上跑了上千次。你拿去改改 IP 和参数,就能直接用。

好了,今天的内容就到这。下一章我们讲数据清洗——因为仪器采集回来的数据,往往没那么干净。你会看到各种毛刺、缺失值、异常点。怎么处理?下回分解。

本章要点回顾:

  • pyvisa 是仪器通信的通用方案,支持 USB、TCPIP、串口
  • query() 发命令并读取回复,用 write() 只发不读
  • 仪器返回的数据格式多样,必须做解析和异常处理
  • 封装成类或函数,提高代码复用性
  • 采集时注意频率控制,别把仪器逼死

公众号:蓝海资料掘金营,微信deep3321