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 += resultexcept Exception:print("34 read timeout")    # 若发生异常（如超时），打印提示信息并继续尝试读取continuereturn 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)