python实现TCP Socket控制测试仪器

说明

本例是用python实现TCP Socket，最终目的是为了控制LCR电容测试仪器。

前提条件

• 测试仪器 配备了以太网接口，支持TCP/IP协议栈
  • 用户可以通过IP地址和端口号连接到仪器，建立TCP通信
• Python及socket包

代码

# 导入依赖包
import socket

# 获取socket套接字对象：供后续通信
def get_socket(ip_str, port):
    # 设置默认超时时间：避免网络通信卡死。若5秒内未能完成操作（如连接、读取数据），会抛出异常
    socket.setdefaulttimeout(5)
    # 创建一个基于 IPv4 的 TCP 套接字，支持点对点可靠连接
    tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # ip地址 和 端口 进行连接
    server_addr = (ip_str, port)    # 具体地址和端口得看仪器
    tcp_socket.connect(server_addr)

    # tcp_socket.recv_into()
    return tcp_socket

# 发送仪器指令：socket套接字对象， 仪器命令字符串
def socket_send_cmd(tcp_socket, cmd):
    # 构建待发送命令字符串，命令以"\r\n"结尾：LCR仪器通常以标准通信协议（如 SCPI）解析命令，要求命令以 \r\n（回车+换行）结束。
    send_str = cmd + "\r\n"
    # 使用sendall将数据完整发送至远端设备，即便数据量较大，也能确保发送完成
    tcp_socket.sendall(send_str.encode('utf-8'))
    # 发送空字节以确保发送结束
    tcp_socket.send(b"")

# 获取仪器返回结果：socket套接字对象
def socket_read_result(tcp_socket):
    while True:  # 无限循环，直到读取到完整结果
        try:
            global read_buf_g   # 声明全局变量，用于存储读取的缓冲数据
            result = tcp_socket.recv(256).decode('utf-8')   # 从套接字中读取最多256字节数据
            if "\n" in result:  # 判断是否接收到完整行数据符号\n（行以“\n”换行符结束）

                # 缓冲“\n”后的信息留给下一次拼接完整信息
                tmp = result[result.find("\n")+1:]  # +1：是为了去掉\n

                # 拼接完整返回信息：# 将前面缓冲的数据 跟 这次读取数据‘\n’ 前面的数据 拼接
                result = read_buf_g + result[:result.find("\n")]

                # 把上面缓冲的信息放在全局变量中，方便下次获取
                read_buf_g = tmp

                # 退出循环，返回结果
                break
            # 如果本次接收信息还没有完整信息标志\n的话，则把信息加入缓存当中
            read_buf_g += result
        except Exception:
            print("34 read timeout")    # 若发生异常（如超时），打印提示信息并继续尝试读取
            continue
   
    return result

# 本例LCR电容测试仪器的ip设为 169.254.1.22，端口为5025
ser_dev = get_socket("169.254.1.22",5025)
# 发生测量指令（SCPI指令）
socket_send_cmd(ser_dev, ":FETC?")
# 获取结果
rd_data = socket_read_result(ser_dev)
# 打印结果
print(rd_data)