当 AI 遇上锁相放大器:探索 zhinst-toolkit 编程的新工作流
在精密测量领域,构建高效的自动化测试脚本是提升科研产出的关键。Zurich Instruments 提供的 Python 开发包 zhinst-toolkit 以模块化架构为核心,极大地简化了锁相放大器的控制逻辑。然而,面对成百上千的硬件节点和复杂的嵌套数据结构,新用户常会困惑:该如何快速上手?怎样才能最高效地编写自动化测量脚本?
近年来,生成式人工智能(AI)在代码辅助方面展现出巨大的应用潜力。为验证其是否可作为新手入门的辅助工具,我们开展了一项探索性实践:使用免费的大语言模型(如 Google Gemini),不用智能体(Agent),结合 LabOne,在真实场景下从零开始构建基于 MFLI 锁相放大器的自动化测控流程。本文由易到难展示了三个示例实验,其核心区别在于数据获取模式的不同。本文中的示例代码及其迭代修正均由 AI 自动生成。
需要强调的是,本文旨在探讨 AI 在学习阶段的辅助潜力及其使用边界,并非推荐将其用于生产级开发。您会发现,AI 一方面能够提供一些合理的框架与蓝本,另一方面也会不时掉入实际运行时才会暴露的陷阱——路径错误、语法过时、数据结构误解等。这些正是我们希望通过本文揭示的重点,希望能够帮助大家避开这些陷阱。
1. 从单“点”采集开始:复制-粘贴-对话
zhinst-toolkit 采用了与硬件节点(Nodes)高度对应的层级化 API 逻辑。通过引用官方文档的标准代码块和示例,开发者可以快速实现设备的初始化。然后,开发者可以把基础代码交给 AI,进行完善。当然,也可让 AI 从零开始编写代码,但这种方式出错率较高,通常不推荐。我们的第一个实验内容是与锁相放大器建立连接,进行基础设置,获得一个解调结果数据点。本文中我们使用的 zhinst-toolkit 和 LabOne 软件版本均为 25.10。
AI 协作提示词: “我正在使用 Zurich Instruments 锁相放大器及其 API zhinst-toolkit。请写一个 Python 基础模板:连接到设备 ID 为 'DEV5418' 的锁相放大器,并将第一个解调器的采样率设为 1 kSa/s,低通滤波器设为 4 阶,最后读取一次当前的 X 和 Y 值。之前我已经给过你一些已经完成的代码,请在那个基础上完善。”
AI 的回答有随机性,且依赖于对话上下文,可能给出这样的代码和注释。
from zhinst.toolkit import Session
# 1. 建立与 LabOne 软件的通信会话 (Session)
session = Session('localhost')
# 2. 连接物理设备
# 如果您在 LabOne 界面能看到该设备,即可通过其 ID 连接
device = session.connect_device('DEV5418')
# 3. 配置解调器 (Demodulator) 参数
# 锁相放大器采用节点(Nodes)控制逻辑:[0] 代表第一个通道
with device.set_transaction():
device.demods[0].rate(1000) # 设置采样率 (Rate) 为 1 kSa/s
device.demods[0].filter_order(4) # 设置低通滤波器阶数 (Order) 为 4 阶 (24 dB/oct)
device.sigouts[0].on(1) # 确保信号输出开启,以便进行同步检测
# 4. 获取实时数据快照
# .sample.get() 类似于按下快门,获取当前解调器所有分量的瞬时值
sample = device.demods[0].sample.get()
# 5. 解析并打印 X 和 Y 值
# 返回的 sample 是一个嵌套字典,包含硬件时间戳以及所有测量向量
x_val = sample['x']
y_val = sample['y']
print(f"--- 设备 DEV5418 实时采样 ---")
print(f"解调器 1 - X 分量: {x_val:.6e} V")
print(f"解调器 1 - Y 分量: {y_val:.6e} V")
乍看之下,AI 生成的代码逻辑清晰,注释完整,但直接运行时往往会暴露出一些典型问题
- 路径“幻觉”:AI 可能会猜错节点名,例如将滤波器阶数误写为 .filter_order(),而正确写法是 .order()。
- API 版本不兼容:AI 习惯性地使用 .get() 方法。但在最新 zhinst-toolkit 中,sample 节点应直接调用。
- 逻辑遗漏:代码往往忘记执行 .enable(1)。如果解调器未开启,读取操作会直接导致超时(Timeout)。
- 数据维度误判:sample['x'] 返回 NumPy 数组,直接用于格式化字符串会引发 TypeError。
面对上述报错,您无需去翻阅几百页的用户手册。一个高效的策略是:通过结合 LabOne 的 API Log、运行时报错信息及数据结构快照,形成一个“反馈闭环”,从而有效引导 AI 逐步修正错误。
第一步:利用 API Log 精准定位
获取正确路径和设置命令的最快方式,是使用 LabOne 界面自带的 API Log 功能。在 LabOne 网页界面的 API Log 中,记录格式选择 Python Toolkit。然后,在界面上用鼠标手动修改一下您想要的参数,比如把低通滤波器阶数改为 4 阶。API Log 日志窗口会记录对应的规范代码,例如 ...demods[0].order(4)。最后,将这行原始代码直接粘贴给 AI,要求其以此为准修正设置命令。我们一贯建议开发者优先熟悉 LabOne 图形界面。通过在界面上进行直观配置,API Log 能够提供最权威、最即时的参考代码。此外,LabOne 集成的示波器、数值绘图 仪及频谱分析仪等可视化工具,不仅是自动化脚本的“参考答案”,更是实验调试中不可或缺的利器。
第二步:AI 化身“数据结构翻译官”
当代码能跑通但解析报错时,您需要利用 AI 的 “数据结构解析” 能力。推荐的做法是:直接在代码中打印 print(sample),将终端输出的完整内容(通常是包含字典、数组的数据快照)粘贴给 AI,并询问如何提取所需的纯数值。当前的数据结构很简单,您可能会认为没有必要,但在随后的例子中,您会发现数据结构复杂时 AI 带来的便利。
经过这几步排错后,我们要求 AI 给出修正后的代码并标注修改之处,最终生成的代码和注释可能如下所示。
from zhinst.toolkit import Session
# 1. 建立会话
session = Session('localhost')
device = session.connect_device('DEV5418')
# 2. 基础配置:使用 Transaction 确保设置同步
with device.set_transaction():
device.demods[0].rate(1000)
device.demods[0].order(4) # 修正 1:路径缩写为 order
device.demods[0].enable(1) # 修正 2:必须 Enable 才能上传数据
device.sigouts[0].on(True)
# 3. 获取实时数据快照
# 修正 3:节点对象直接调用,无需 .get()
sample = device.demods[0].sample()
# 4. 解析数据:结构化提取
# 修正 4:利用 [0] 索引从 NumPy 数组中拆包提取标量
x_val = sample['x'][0]
y_val = sample['y'][0]
print(f"--- 设备 DEV5418 实时采样 ---")
print(f"解调器 1 - X 分量: {x_val:.6e} V")
print(f"解调器 1 - Y 分量: {y_val:.6e} V")
2. 时域数据采集方案:poll 模式
当测量需求从单个数据点扩展到一段时间中的数据流时,poll 指令提供了基础的数据段读取能力。它通过轮询方式读取指定时间窗口内累积的原始数据包。现在,我们尝试第二个实验,用 poll 抓取解调器结果的一段连续数据。
AI 协作提示词:我正在使用 zhinst-toolkit 的 session.poll 方法。请帮我写一段代码:订阅解调器 1 的数据节点,抓取 0.5 秒的数据流。
AI 可能给出这样的代码。代码中的错误标注并非 AI 初始输出,而是我们在修正后要求其补充的说明。
from zhinst.toolkit import Session
import time
# 1. 基础连接
session = Session('localhost')
device = session.connect_device('DEV5418')
# 2. 配置与启用(这里已经吸取教训,修正了之前的坑)
with device.set_transaction():
device.demods[0].rate(1000)
device.demods[0].order(4) # ✅ 修正点:不再使用 filter_order
device.demods[0].enable(1) # ✅ 修正点:记得开启传输了
# 3. 订阅数据路径
sample_path = device.demods[0].sample
# ❌ 错误 A:依然使用 session.subscribe(path) 这种旧逻辑
# 在新版 Toolkit 中,这会提示 session 没有该属性或无法正确注册
session.subscribe(sample_path)
# 4. 执行 Poll 采集
data = session.poll(0.5)
# 5. 提取数据
if sample_path in data:
# ❌ 错误 B:AI 认为 data[sample_path]['x'] 拿到的直接就是数字
x_data = data[sample_path]['x']
y_data = data[sample_path]['y']
# ❌ 错误 C:维度崩溃。由于 x_data 是 numpy.ndarray,直接格式化打印会抛出 TypeError
print(f"解调结果: X = {x_data:.6e}, Y = {y_data:.6e}")
# 6. 取消订阅
# ❌ 错误 D:语法错误,session 无法直接处理节点对象的取消订阅
session.unsubscribe(sample_path)
通过之前的纠错交互,AI 在后续输出中体现出了一定的“学习”效果(例如记住了正确的节点名和 enable 设置),但在处理 zhinst-toolkit 特有的对象化语法以及 NumPy 结构化数据的解析时,依然会掉进坑里。
错误 A & D:对象化语法的“最后一步”
- 现象:AttributeError: 'Session' object has no attribute 'subscribe'。
- 真相:在 zhinst-toolkit 中,订阅操作属于节点,而不是 Session。
- 纠正方案:直接告诉 AI “用节点对象自己订阅。”
- 最终形态:sample_path.subscribe() 和 sample_path.unsubscribe()。
错误 B & C:数据结构的“深层解析”
- 现象:程序崩溃,提示 TypeError: unsupported format string passed to numpy.ndarray。
- 真相:这是 AI 对硬件数据最典型的误解。poll 返回的是一个包含 500 个点(0.5s @ 1kHz 数据采样率)的数组。不能像格式化单个数字那样直接格式化整个数组。
- 纠正方案:利用 AI 识别 data 的结构。print(data[sample_path]),把打印输出给 AI, 请 AI 解析。您会发现 data[sample_path]['x'] 实际上是一个 NumPy 数组。
- 最终形态:如果您要拿第一个点:x_data[0]。如果要算平均值:np.mean(x_data)。
修正后的代码如下。
from zhinst.toolkit import Session
import numpy as np
# 1. 基础连接
session = Session('localhost')
device = session.connect_device('DEV5418')
# 2. 配置与启用
with device.set_transaction():
device.demods[0].rate(1000)
device.demods[0].order(4)
device.demods[0].enable(1)
# 3. 订阅数据路径
sample_path = device.demods[0].sample
# ✅ 修正 A:由节点对象本身发起订阅,而不是 session
sample_path.subscribe()
# 4. 执行 Poll 采集
# poll 返回的是一个字典,键是节点的完整路径字符串
data = session.poll(0.5)
# ✅ 修正 B:在字典中匹配时,需确保 Key 是路径字符串
# 或者直接使用 sample_path 作为键(toolkit 内部会处理对象到字符串的映射)
if sample_path in data:
# 提取数据块
sample = data[sample_path]
# ✅ 修正 C:处理维度崩溃
# sample['x'] 返回的是 array([val1, val2, ...])
# 如果只想打印第一个点的值,必须使用索引 [0] 提取标量
x_val = sample['x'][0]
y_val = sample['y'][0]
print(f"解调结果(第一个样本点): X = {x_val:.6e} V, Y = {y_val:.6e} V")
print(f"本次 Poll 共采集到 {len(sample['x'])} 个点")
else:
print("未获取到数据,请检查设备连接或解调器是否已 Enable。")
# 5. 取消订阅
# ✅ 修正 D:同样由节点对象发起取消订阅
sample_path.unsubscribe()
3. 高性能数据采集方案:DAQ 模块
相比基础的 poll 指令,DAQ (Data Acquisition) 模块功能更加强大。它支持直接订阅幅度(R)和相位(Phase)等数据节点,支持硬件级触发,并且可以进行无限次连续采集。现在,我们进行第三个实验,用 DAQ 连续采集解调数据,并将数据保存下来。
AI 协作提示词:“我想用更专业的 DAQ 模块进行连续采集。请写一个脚本:订阅幅值 R 路径,以连续模式采集,每个数据块包含 100 个点,并将结果(包含 Trigger_ID、时间轴、R 值)及时写入 CSV 文件。
AI 可能会给出这样的代码。代码中的错误标注并非 AI 初始输出,而是我们在修正后要求其补充的说明。
import numpy as np
import time
import csv
from zhinst.toolkit import Session
# ==========================================
# 1. 初始化设备连接
# ==========================================
session = Session("localhost")
device = session.connect_device("DEV5418") # 替换为你的设备ID
# --- 仪器配置 ---
with device.set_transaction():
device.demods[0].enable(1)
device.demods[0].harmonic(1)
device.demods[0].rate(10)
device.sigouts[0].amplitudes[0](0.01)
device.sigins[0].range(1)
path = device.demods[0].sample.r
# ==========================================
# 2. 配置 DAQ 模块
# ==========================================
daq = session.modules.daq
daq.device(device)
daq.subscribe(path)
daq.type(0) # Continuous 模式
daq.grid.mode(2) # 人工注释,这是错误的,但是不影响运行。后文中会做说明。
daq.grid.cols(100) # 建议设大一点,比如 100,提高保存效率
daq.endless(1) # 无限采集
# ==========================================
# 3. 连续采集并保存(带 Header 信息)
# ==========================================
filename = "thz_data_with_headers.csv"
with open(filename, mode='w', newline='') as f:
writer = csv.writer(f)
# 更新表头:增加 Trigger_ID 和硬件原始 Ticks
writer.writerow(['Trigger_ID', 'Hardware_Ticks', 'Relative_Time_s', 'Amplitude_R'])
print(f"正在保存数据至 {filename}... 按 Ctrl+C 停止")
daq.execute()
# ==========================================
# 4. 错误的数据解析逻辑(仅展示 while 循环内部)
# ==========================================
try:
while True:
data = daq.read()
if path in data:
# ❌ 典型错误 1:忽略了数据是一个列表(List of Chunks)
# 报错:AttributeError: 'list' object has no attribute 'header'
# 因为 data[path] 返回的是 [DAQResult, DAQResult, ...]
block = data[path]
# ❌ 典型错误 2:错误地提取 Header 信息
# 开发者认为 triggernumber 是个数字,实际上它在 header 字典里也是数组
trig_id = block.header['triggernumber']
# ❌ 典型错误 3:维度坍塌(Dimension Mismatch)
# chunk.value 通常是 (1, 100) 的二维数组,直接 row 写入会导致 CSV 格式错乱
t_points = block.time
v_points = block.value
# ❌ 典型错误 4:错误的 zip 尝试
# 试图直接将数组和标量 zip,会导致每一行只写入一个字母或直接报错
rows = zip(trig_id, t_points, v_points)
writer.writerows(rows)
f.flush()
time.sleep(0.5)
except KeyboardInterrupt:
daq.finish()
DAQ 模块返回的数据结构层级丰富、嵌套复杂,即便是经验丰富的开发者在初次解析时也可能感到棘手。在 AI 协作模式下,最有效的做法是将 daq.read() 返回的数据快照直接交给 AI 分析,避免因嵌套层级深而产生的逻辑错误。
优化后的核心实现逻辑:
- 开启 endless 模式:通过 daq.endless(1) 实现真正的无限循环采集。
- 解析 Chunk 列表:读取结果是一个包含多个数据块的列表,必须通过循环逐一解析。
- 数据扁平化:利用 flatten() 处理 NumPy 的多维输出,配合 zip 函数将时间轴与数值一一对应,确保 CSV 每行一个点。
- 及时刷新:定期执行 f.flush() 确保即使程序意外中断,数据也能安全存盘。
在最终实现中,我们将 daq.grid.mode 从 AI 建议的 2 手动修正为 4。这是一个极易被忽视的“暗坑”:
- Mode 2(线性插值模式): 即使程序运行不报错,DAQ 也会根据 daq.grid.duration(默认 0.01s)对原始数据进行重采样。这意味着即便设置了每行 100 个点,这些点也是在固定时间窗口内经过线性插值计算出来的,而非真实的原始采样点。
- Mode 4(精准对齐采样): 这是实现精准采集的关键。在该模式下,DAQ 会严格按照 device.demods[0].rate() 的采样率,实打实地采集 daq.grid.cols 指定的点数并作为一行输出。
通过这一改动,我们确保了数据的物理真实性 ,避免了因自动插值带来的波形失真。
import numpy as np
import time
import csv
from zhinst.toolkit import Session
# ==========================================
# 1. 初始化设备连接
# ==========================================
session = Session("localhost")
device = session.connect_device("DEV5418") # 替换为你的设备ID
# --- 仪器配置 ---
with device.set_transaction():
device.demods[0].enable(1)
device.demods[0].harmonic(1)
device.demods[0].rate(10)
device.sigouts[0].amplitudes[0](0.01)
device.sigins[0].range(1)
path = device.demods[0].sample.r
# ==========================================
# 2. 配置 DAQ 模块
# ==========================================
daq = session.modules.daq
daq.device(device)
daq.subscribe(path)
daq.type(0) # Continuous 模式
daq.grid.mode(4) # 手动修改,4=exact(on-grid)
daq.grid.cols(100) # 建议设大一点,比如 100,提高保存效率
daq.endless(1) # 无限采集
# ==========================================
# 3. 连续采集并保存(带 Header 信息)
# ==========================================
filename = "thz_data_with_headers.csv"
with open(filename, mode='w', newline='') as f:
writer = csv.writer(f)
# 更新表头:增加 Trigger_ID 和硬件原始 Ticks
writer.writerow(['Trigger_ID', 'Hardware_Ticks', 'Relative_Time_s', 'Amplitude_R'])
print(f"正在保存数据至 {filename}... 按 Ctrl+C 停止")
daq.execute()
try:
while True:
data = daq.read()
# data 是字典,path 是键,chunks 是 DAQResult 对象列表
if path in data:
for chunk in data[path]:
# --- 提取 Header 信息 ---
trig_id = chunk.header['triggernumber'][0]
hw_ticks = chunk.header['createdtimestamp'][0]
# --- 提取并平整化数据 ---
t_points = chunk.time # (N,) 数组
v_points = chunk.value.flatten() # (1, N) -> (N,)
# --- 准备写入行 ---
# 为该 chunk 的每一个点都打上相同的 ID 和 Ticks 标签
# zip 的时候,我们手动把 trig_id 和 hw_ticks 复制成同样长度
rows = zip(
[trig_id] * len(v_points),
[hw_ticks] * len(v_points),
t_points,
v_points
)
writer.writerows(rows)
f.flush()
print(f"已保存 Chunk #{trig_id},包含 {len(v_points)} 个点")
time.sleep(10) # 手动修改为10,避免 CPU 过载
except KeyboardInterrupt:
print("\n正在停止采集并保存...")
daq.finish()
daq.unsubscribe('*')
结语:AI 辅助开发的边界与最佳实践
通过以上实验,我们看到了“AI + zhinst-toolkit”这一模式在降低初始学习成本和加速代码原型构建方面的潜力。AI 可以作为辅助工具,快速生成代码框架或帮助理解复杂的数据结构。我们建议可以尝试这样的协作模式:
- 从示例开始:利用官方模板和示例代码建立坚实基础,不要重复造轮子。如果您遇到的需求和文中示例类似,不妨直接以文中的修正代码作为起点。
- 拥抱 API Log:用图形界面操作生成的标准代码,确保硬件配置的准确性。
- AI 闭环调试:将运行时的报错信息和复杂的数据结构快照反馈给 AI,让其协助诊断。
然而,工具在进化,专业洞察的核心价值始终不可替代。当前 AI 仍可能产生“幻觉”,即使生成的代码语法正确,也无法替代对物理测量原理的深入理解。当您遇到 AI 无法解决的硬件报错、性能瓶颈,或对测量数据心存疑虑时,请随时联系我们的应用支持团队。 我们的应用科学家将为您提供从底层硬件到软件逻辑的专业指导——毕竟,对物理实验边界的把握与仪器极限的洞察,目前依然是人类专家的领域。我们的应用工程师团队始终是您探索之路的可靠伙伴。本文的撰写与修订得益于 AI 工具的辅助,以及我们中国区应用科学家团队的支持。
