2. Python 数据采集库：打通硬件与数据的管道

数据采集这活儿，说白了就是跟各种硬件设备打交道，把物理世界的信号变成电脑里能处理的数据。我做了这么多年，发现很多新手上来就写采集逻辑，结果串口打不开、仪器连不上、数据读出来全是乱码——嗯，这些坑我都踩过。今天咱们就把三个最常用的库捋一遍：pySerial、pyvisa 和 pandas。

本章核心脉络：串口通信 → 仪器控制 → 数据整理。这三步走通了，你就能把传感器、示波器、万用表的数据乖乖收进 DataFrame 里。

Python 数据采集库知识体系 Python 数据采集库知识体系 数据采集入口 pySerial 串口通信 pyvisa 仪器控制 pandas 数据读取 端口扫描 波特率配置 读写超时 数据解析 资源管理 SCPI 指令 GPIB/USB 波形读取 CSV/Excel 时间序列 缺失值处理 数据合并 协同输出：结构化采集数据

2.1 pySerial：串口通信的瑞士军刀

串口通信在工业现场、嵌入式开发里太常见了。我最早接触 pySerial 是在一个温湿度监控项目里，传感器每 200ms 吐一串数据，结果我忘了设超时，程序直接卡死——嗯，教训深刻。

2.1.1 快速上手

安装就一行命令，没什么好说的：

pip install pyserial

打开串口的基本姿势是这样的：

import serial

ser = serial.Serial(
    port='COM3',        # Windows 用 COM 口，Linux 用 /dev/ttyUSB0
    baudrate=115200,    # 波特率，跟设备保持一致
    timeout=1           # 读超时，单位秒。不设的话 read() 会一直等
)

# 写数据
ser.write(b'AT\r\n')

# 读数据
response = ser.read(100)   # 读 100 字节
line = ser.readline()      # 读一行，直到换行符

ser.close()

我的习惯：用 with serial.Serial(...) as ser: 上下文管理器，省得忘了 close()。特别是调试时频繁中断，资源泄漏很烦人。

2.1.2 端口扫描与避坑

你想想看，设备插上电脑，端口号是多少？总不能一个个试吧。pySerial 自带了扫描工具：

import serial.tools.list_ports

ports = serial.tools.list_ports.comports()
for p in ports:
    print(f"{p.device}: {p.description}")

我曾经踩过的坑：USB 转串口线质量参差不齐，有些廉价线在 Windows 上会识别为 COM 口但实际无法通信。建议先用串口调试助手（比如 Putty）手动发一条数据确认链路通不通，再写代码。

2.1.3 数据解析实战

串口数据通常是二进制或文本混合的。我习惯先读原始字节，再按协议解析：

import serial

ser = serial.Serial('COM3', 9600, timeout=0.5)
buffer = b''

while True:
    chunk = ser.read(64)
    if not chunk:
        break
    buffer += chunk
    # 假设协议是：帧头 0xAA + 数据长度 + 数据 + 校验和
    while len(buffer) >= 4:
        if buffer[0] != 0xAA:
            buffer = buffer[1:]  # 丢无效字节
            continue
        length = buffer[1]
        if len(buffer) < 2 + length:
            break  # 数据还没收全
        frame = buffer[:2 + length]
        # 校验和验证...
        data = frame[2:2+length]
        print(f"收到数据: {data.hex()}")
        buffer = buffer[2 + length:]

关键点：串口是流式协议，没有固定帧边界。一定要自己做帧同步和校验，别指望 read() 一次返回完整的一帧。

2.2 pyvisa：让仪器开口说话

示波器、万用表、信号发生器这些仪器，通常用 GPIB、USB 或以太网连接。pyvisa 就是它们的翻译官。我个人觉得，pyvisa 比串口更「高级」一点——你发的是 SCPI 指令，仪器能听懂。

2.2.1 安装与资源发现

pip install pyvisa pyvisa-py

先看看电脑上连了哪些仪器：

import pyvisa

rm = pyvisa.ResourceManager()
resources = rm.list_resources()
print(resources)
# 输出示例：['USB0::0x1234::0x5678::INSTR', 'GPIB0::1::INSTR']

2.2.2 发送指令与读取数据

SCPI 指令是仪器界的通用语言。比如读万用表的电压：

import pyvisa

rm = pyvisa.ResourceManager()
inst = rm.open_resource('USB0::0x1234::0x5678::INSTR')

# 设置万用表为直流电压档
inst.write('MEAS:VOLT:DC?')
# 读取结果
voltage = inst.read()
print(f"电压: {voltage} V")

inst.close()

我建议：用 inst.query('*IDN?') 先确认仪器身份。这步能帮你排除 90% 的通信问题——连不上时先看看仪器地址对不对。

2.2.3 批量采集波形数据

示波器采集波形时，数据量比较大。我一般用二进制块传输：

import pyvisa
import numpy as np

rm = pyvisa.ResourceManager()
scope = rm.open_resource('TCPIP0::192.168.1.100::INSTR')

# 设置时基和通道
scope.write(':TIM:SCAL 1e-3')
scope.write(':CHAN1:DISP ON')

# 获取波形数据（二进制格式）
scope.write(':WAV:DATA?')
raw_data = scope.read_raw()
# 跳过头部，解析为浮点数
header_len = int(raw_data[1:2]) + 2
waveform = np.frombuffer(raw_data[header_len:-1], dtype='>f4')

print(f"波形点数: {len(waveform)}")
scope.close()

注意：不同厂家的 SCPI 指令略有差异。Keysight、Tektronix、Rigol 的波形查询命令可能不一样。我当年被这个坑过——拿着 Agilent 的代码去控制 Tektronix，读出来的数据全是乱的。一定要看仪器手册。

2.3 pandas：把数据收拾得整整齐齐

数据采回来了，但往往是乱糟糟的文本、CSV 或者二进制。pandas 就是用来收拾残局的。说实话，没有 pandas 之前，我都是自己写循环解析，又慢又容易出错。

2.3.1 读取常见文件格式

文件格式 pandas 函数 常用参数
CSV pd.read_csv() sep, header, encoding
Excel pd.read_excel() sheet_name, usecols
JSON pd.read_json() orient
固定宽度文本 pd.read_fwf() widths

举个实际例子，从串口采集的温度数据存成了 CSV：

import pandas as pd

df = pd.read_csv('temperature_log.csv', 
                 parse_dates=['timestamp'],
                 index_col='timestamp')
print(df.head())

2.3.2 时间序列处理

采集数据几乎都带时间戳。pandas 的时间序列功能非常强大：

import pandas as pd

# 假设 df 有两列：timestamp 和 temperature
df['timestamp'] = pd.to_datetime(df['timestamp'])
df = df.set_index('timestamp')

# 重采样到 1 分钟均值
df_resampled = df.resample('1T').mean()

# 处理缺失值——传感器偶尔会丢包
df_filled = df_resampled.interpolate(method='linear')

print(df_filled.head())

核心思路：采集数据 → 原始 DataFrame → 清洗（去重、补缺、类型转换）→ 分析/可视化。pandas 把中间两步变得极其简单。

2.3.3 数据合并与导出

有时候多个传感器数据要合并。我习惯用 pd.concat 或 pd.merge：

import pandas as pd

df_temp = pd.read_csv('temperature.csv')
df_humidity = pd.read_csv('humidity.csv')

# 按时间合并
df_combined = pd.merge(df_temp, df_humidity, on='timestamp', how='outer')

# 导出为 Excel，方便给非技术人员看
df_combined.to_excel('sensor_data.xlsx', index=False)

我的习惯：采集完成后，第一时间用 df.info() 和 df.describe() 检查数据质量。数据类型对不对？有没有空值？范围合理吗？这些检查能省掉后面大量 debug 时间。

2.4 三个库的协同工作流

实际项目中，这三个库很少单独使用。我画了个典型流程：

  1. pySerial/pyvisa 采集原始数据
  2. Python 标准库（struct、json）解析成结构化数据
  3. pandas 整理、清洗、存储
  4. matplotlib/plotly 可视化（后续章节会讲）

举个完整的例子——从串口读取温湿度传感器数据，存到 DataFrame：

import serial
import pandas as pd
import time

ser = serial.Serial('COM3', 115200, timeout=1)
data_rows = []

for _ in range(100):  # 采集 100 次
    line = ser.readline().decode('utf-8', errors='ignore').strip()
    if line:
        # 假设数据格式: "temp=25.3,hum=60.1"
        parts = line.split(',')
        temp = float(parts[0].split('=')[1])
        hum = float(parts[1].split('=')[1])
        data_rows.append({
            'timestamp': time.time(),
            'temperature': temp,
            'humidity': hum
        })
    time.sleep(0.1)

ser.close()

df = pd.DataFrame(data_rows)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s')
print(df.head())
← 上一章 📖 返回目录 下一章 →