在信号发生器脚本中实现信号的实时监控,通常需要结合硬件接口、数据采集和可视化技术。以下是分步骤的实现方法及代码示例(以Python为例),涵盖从硬件控制到实时显示的完整流程:
一、核心实现步骤
1. 硬件连接与控制
- 信号发生器控制:通过SCPI命令(如Keysight、Rigol设备)或专用API(如NI-DAQ、ADI库)控制信号输出。
- 数据采集:使用示波器、ADC或数据采集卡实时采集输出信号。
2. 数据流处理
- 同步触发:确保信号发生器输出与采集设备同步(如通过外部触发)。
- 缓冲机制:采用循环缓冲区(Circular Buffer)存储实时数据,避免内存溢出。
3. 实时可视化
- 动态绘图:使用
matplotlib的FuncAnimation或pyqtgraph实现高效刷新。 - 多线程处理:分离数据采集与绘图线程,避免阻塞。
4. 关键参数监控
- 频谱分析:通过FFT计算实时频谱,监控杂散和谐波。
- 统计指标:计算幅度、频率的实时均值/标准差。
二、代码实现示例
1. 基于PyVISA的信号发生器控制
| import pyvisa |
|
| # 连接信号发生器 |
| rm = pyvisa.ResourceManager() |
| scope = rm.open_resource("TCPIP0::192.168.1.100::inst0::INSTR") |
|
| # 设置信号参数(示例:1MHz正弦波,1Vpp) |
| scope.write("SOUR1:FUNC SIN") |
| scope.write("SOUR1:FREQ 1e6") |
| scope.write("SOUR1:VOLT 1") |
| scope.write("OUTP1 ON") |
2. 实时数据采集(模拟数据)
| import numpy as np |
| import time |
| from collections import deque |
|
| # 初始化缓冲区(存储最近1000个采样点) |
| buffer_size = 1000 |
| data_buffer = deque(maxlen=buffer_size) |
| timestamps = deque(maxlen=buffer_size) |
|
| # 模拟数据采集(实际应替换为硬件读取,如scope.query_binary_values) |
| def acquire_data(): |
| # 示例:生成带噪声的1MHz信号 |
| t = time.time() |
| x = np.linspace(0, 1, 100) |
| signal = np.sin(2 * np.pi * 1e6 * x) + 0.1 * np.random.randn(100) |
| return t, signal |
|
| # 数据采集线程(简化版,实际建议用threading) |
| def data_collection_loop(): |
| while True: |
| t, signal = acquire_data() |
| data_buffer.extend(signal) |
| timestamps.extend([t] * len(signal)) |
| time.sleep(0.01) |
3. 实时绘图(使用Matplotlib)
| import matplotlib.pyplot as plt |
| from matplotlib.animation import FuncAnimation |
|
| fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 6)) |
| line1, = ax1.plot([], [], 'b-') |
| line2, = ax2.plot([], [], 'r-') |
| ax1.set_ylim(-1.5, 1.5) |
| ax2.set_ylim(-50, 10) |
|
| def update_plot(frame): |
| if len(data_buffer) > 0: |
| # 时域绘图 |
| line1.set_data(range(len(data_buffer)), data_buffer) |
| ax1.set_xlim(0, buffer_size) |
|
| # 频域绘图(FFT) |
| fft_result = np.abs(np.fft.fft(data_buffer))[:buffer_size//2] |
| freqs = np.fft.fftfreq(buffer_size, d=1e-6)[:buffer_size//2] |
| line2.set_data(freqs, 20 * np.log10(fft_result)) |
| ax2.set_xlim(0, 5e6) |
|
| return line1, line2 |
|
| ani = FuncAnimation(fig, update_plot, interval=100) |
| plt.tight_layout() |
| plt.show() |
4. 完整脚本(多线程版)
| import threading |
|
| # 启动数据采集线程 |
| thread = threading.Thread(target=data_collection_loop, daemon=True) |
| thread.start() |
|
| # 主线程运行绘图 |
| plt.show() |
三、关键优化点
1. 硬件同步
- 触发信号:通过信号发生器的触发输出连接采集设备的外部触发输入。
- 时间戳对齐:使用高精度时钟(如
time.perf_counter())同步数据采集与显示。
2. 性能优化
- 减少绘图开销:
- 使用
pyqtgraph替代matplotlib(速度更快,适合高频更新)。 - 限制绘图点数(如每帧仅显示最近1000点)。
- 异步处理:
3. 异常处理
- 设备断开检测:捕获
pyvisa.VisaIOError并重连。 - 数据完整性检查:验证采样点数是否符合预期。
四、扩展功能
1. 频谱监控
| from scipy.fft import fft, fftfreq |
|
| def compute_spectrum(data, sample_rate=1e6): |
| n = len(data) |
| yf = fft(data) |
| xf = fftfreq(n, 1/sample_rate)[:n//2] |
| return xf, 2.0/n * np.abs(yf[:n//2]) |
2. 报警机制
pythondef check_threshold(data, threshold=1.2):if np.max(np.abs(data)) > threshold:print("⚠️ 信号幅度超限!")
3. 数据记录
| import pandas as pd |
|
| def save_to_csv(timestamps, data): |
| df = pd.DataFrame({"timestamp": timestamps, "signal": data}) |
| df.to_csv("signal_log.csv", index=False) |
五、工具与库推荐
- 硬件控制:
- PyVISA(SCPI设备)
- NI-DAQmx(National Instruments设备)
- ADI库(ADI相关硬件)
- 实时绘图:
pyqtgraph(高性能)bokeh(Web交互式)
- 信号处理:
scipy.signal(滤波、FFT)numpy(数值计算)
六、常见问题解决
- 绘图卡顿:
- 降低更新频率(如从100Hz降至30Hz)。
- 使用硬件加速(如OpenGL后端)。
- 数据丢失:
- 同步错误:
通过上述方法,可实现信号发生器输出信号的实时监控,适用于调试、自动化测试或闭环控制场景。根据实际硬件接口调整代码中的通信协议(如USBTMC、VXI-11等)即可适配不同设备。