2. SCPI命令发送与响应解析
任务说明
本案例实现一个基于TCP/IP的SCPI命令发送与响应解析工具,用于控制支持LAN接口的仪器(如示波器、信号源、万用表等)。工具需完成:建立连接、发送SCPI命令、读取并解析响应(含数值与字符串)、错误处理与连接关闭。通过一个模拟SCPI服务器验证完整流程。
实现方案
使用Python标准库socket实现底层通信,定义SCPIClient类封装连接管理与命令交互。响应解析支持三种常见格式:纯数值(如1.23E-5)、带单位数值(如1.23E-5 V)、带引号字符串(如"Waveform1")。为便于独立验证,同时提供一个基于socketserver的模拟SCPI服务器,监听本地端口并回显固定响应。
代码实现
import socket
import re
import threading
import socketserver
class SCPIClient:
"""基于TCP的SCPI仪器客户端"""
def __init__(self, host, port, timeout=5):
self.host = host
self.port = port
self.timeout = timeout
self.sock = None
def connect(self):
"""建立TCP连接"""
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.settimeout(self.timeout)
self.sock.connect((self.host, self.port))
def send_command(self, cmd):
"""发送SCPI命令(自动追加换行符)"""
if not self.sock:
raise ConnectionError("Not connected")
if not cmd.endswith('\n'):
cmd += '\n'
self.sock.sendall(cmd.encode('ascii'))
def read_response(self):
"""读取完整响应直到换行符"""
data = b''
while not data.endswith(b'\n'):
chunk = self.sock.recv(4096)
if not chunk:
break
data += chunk
return data.decode('ascii').strip()
def query(self, cmd):
"""发送查询命令并返回原始响应字符串"""
self.send_command(cmd)
return self.read_response()
def query_number(self, cmd):
"""发送查询并解析为浮点数(支持单位后缀)"""
resp = self.query(cmd)
return parse_number(resp)
def query_string(self, cmd):
"""发送查询并解析为字符串(去除引号)"""
resp = self.query(cmd)
return parse_quoted_string(resp)
def close(self):
"""关闭连接"""
if self.sock:
self.sock.close()
self.sock = None
def parse_number(resp):
"""解析数值响应,支持科学计数法和单位后缀"""
m = re.search(r'[-+]?\d*\.?\d+(?:[eE][-+]?\d+)?', resp)
if m:
return float(m.group())
raise ValueError(f"Cannot parse number from: {resp}")
def parse_quoted_string(resp):
"""解析带引号字符串响应"""
m = re.search(r'"([^"]*)"', resp)
if m:
return m.group(1)
return resp # 无引号时返回原字符串
# ---------- 模拟SCPI服务器(用于验证) ----------
class MockSCPIHandler(socketserver.StreamRequestHandler):
def handle(self):
while True:
cmd = self.rfile.readline().decode('ascii').strip()
if not cmd:
break
if cmd == '*IDN?':
response = '"MockScope,DS1000Z,SN12345,1.01"\n'
elif cmd == 'MEAS:VPP?':
response = '1.234E+0 V\n'
elif cmd == 'FREQ?':
response = '1.000000E+6\n'
elif cmd == 'SYST:ERR?':
response = '0,"No error"\n'
else:
response = 'ERROR: Unknown command\n'
self.wfile.write(response.encode('ascii'))
class MockSCPIServer:
"""启动模拟服务器(线程方式)"""
def __init__(self, host='127.0.0.1', port=5025):
self.server = socketserver.TCPServer((host, port), MockSCPIHandler)
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
def start(self):
self.thread.start()
def stop(self):
self.server.shutdown()
self.server.server_close()
# ---------- 主流程演示 ----------
if __name__ == '__main__':
# 启动模拟服务器
mock = MockSCPIServer()
mock.start()
# 创建客户端并连接
client = SCPIClient('127.0.0.1', 5025)
client.connect()
# 发送查询并解析
idn = client.query_string('*IDN?')
vpp = client.query_number('MEAS:VPP?')
freq = client.query_number('FREQ?')
err_code, err_msg = client.query('SYST:ERR?').split(',', 1)
# 输出结果
print(f"IDN: {idn}")
print(f"VPP: {vpp:.3f} V")
print(f"FREQ: {freq:.0f} Hz")
print(f"ERR: {err_code} - {err_msg.strip('\"')}")
# 清理
client.close()
mock.stop()
关键设计说明
- 响应解析:
parse_number使用正则提取首个浮点数(含科学计数法),忽略单位后缀;parse_quoted_string提取引号内内容,无引号时返回原始字符串。 - 通信协议:每条SCPI命令以换行符
\n结尾,响应同样以换行符结束,符合大多数LAN仪器的SCPI-over-TCP约定。 - 错误处理:连接超时、未连接时发送命令、解析失败均抛出明确异常,便于上层捕获。
验证方法
直接运行本脚本(python scpi_client.py),观察控制台输出。预期输出如下(与模拟服务器逻辑严格对应):
IDN: MockScope,DS1000Z,SN12345,1.01
VPP: 1.234 V
FREQ: 1000000 Hz
ERR: 0 - No error
判定标准:程序无异常退出,四行输出与上述内容一致,说明SCPI命令发送、响应读取、数值/字符串解析及连接管理均工作正常。若连接真实仪器,只需修改host和port为仪器实际IP与端口即可复用同一客户端类。