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与端口即可复用同一客户端类。