python自动化测试工具
·
车载控制器项目开发过程中，通常需要搭建一些压力测试台架，PC上位机软件需要控制外部设备比如CAN、程控电源、usb 继电器之类的。本文分享几个简单的python接口，这些接口基于外设厂商提供的dll，进行二次封装成python类，方便调用和移植。
1. 程控电源控制脚本
设备型号为IT6211B国产的程控电源
需要安装NI相关套件
Installation — PyVISA 1.13.1.dev31+g071fa6d documentation
代码如下：
#依赖库pyvisa 命令:pip install pyvisa
import pyvisa
import time
DEVICE_ID = 'USB0::0xFFFF::0x6122::800612011787070009::INSTR'
"""write 接口的返回值是写入字符的长度，加\r\n windows系统"""
class IT6122B(object):
def __init__(self, device_id=DEVICE_ID):
self.resourceManager = pyvisa.ResourceManager('c:/windows/system32/visa32.dll')
self.devcie_id = device_id
self.line_break_len = 2 #windows
print("识别到的设备号: ")
print(self.resourceManager.list_resources())
if self.resourceManager == None:
print("device couldn't be opened!")
def open(self):
"""打开设备"""
self.device_inst = self.resourceManager.open_resource(self.devcie_id)
def close(self):
"""关闭设备"""
self.device_inst.close()
def _write(self, str):
"""控制接口"""
if len(str) == 0:
return False
ret = self.device_inst.write(str)
if ret != len(str) + self.line_break_len:
print("设置错误")
return False
else:
return True
def base_info_print(self):
"""打印基本信息"""
print("设备版本号：")
print(self.device_inst.query_ascii_values("SYST:VERS?")) #查询版本号
def power_ctl(self, power_swt):
"""电源控制接口"""
if power_swt == 'ON':
return self._write("OUTPut ON")
if power_swt == 'OFF':
return self._write("OUTPut OFF")
def mode_set(self, mode=None):
"""it6122b 模式设置"""
if mode == None:
return False
return self._write(mode)
def vol_get(self):
"""电压读取,返回浮点数"""
return self.device_inst.query("MEASure:VOLTage?")
def cur_get(self):
"""获取当前电流值，返回浮点数"""
return self.device_inst.query("MEASure:CURRent?")
def vol_set(self, vol):
"""设置电压,设置完电压需要至少等待500ms让电源生效，如果有问题这个值可以适当放大一些"""
tmp_str = "VOLTage " + vol
return self._write(tmp_str)
def set_ref_vol(vol):
"""设置参考电压"""
global it6122b_obj
it6122b_obj.vol_set("{}".format(vol))
time.sleep(0.5)
ret = it6122b_obj.vol_get()
if __name__ == "__main__":
it6122b_obj = IT6122B(DEVICE_ID)
it6122b_obj.open()
it6122b_obj.mode_set("SYST:REM")
it6122b_obj.base_info_print()
it6122b_obj.vol_set("8.0")
time.sleep(0.5)
it6122b_obj.power_ctl("ON")
time.sleep(0.5)
ret = it6122b_obj.vol_get()
ret = float(ret) * 1000
print("数值： " + str(ret))
print("设置值 " + "8000")
print("获取的电压值 ：" + it6122b_obj.vol_get())
通信接口为统一的visa接口，有专门的python包 pyvisa包，这个开发也很简单，里面实现了一些基本的操作。其中device_id可以通过NI MAX-->设备和接口查看设备的id，这个固定的。
2. USB继电器
依赖的文件：
1.淘宝上买的usb接口的继电器，一般卖家会提供dll的开发文件和对应的头文件
2.根据头文件识别出支持的哪些接口
3.dll是32bit还是64bit的，需要使用对应的32bit python或64bit python运行
卖家提供的头文件：
#ifndef USB_RELAY_DEVICE_H__
#define USB_RELAY_DEVICE_H__
#pragma comment(lib, "setupapi.lib")
#ifdef __cplusplus
extern "C" {
#endif
//#pragma comment(lib, "setupapi.lib")
#ifdef _WIN32
#define EXPORT_API __declspec(dllexport)
#else
#define EXPORT_API
#endif
enum usb_relay_device_type
{
USB_RELAY_DEVICE_ONE_CHANNEL = 1,
USB_RELAY_DEVICE_TWO_CHANNEL = 2,
USB_RELAY_DEVICE_FOUR_CHANNEL = 4,
USB_RELAY_DEVICE_EIGHT_CHANNEL = 8
};
/*usb relay board info structure*/
struct usb_relay_device_info
{
unsigned char *serial_number;
char *device_path;
usb_relay_device_type type;
usb_relay_device_info* next;
};
/*init the USB Relay Libary
@returns: This function returns 0 on success and -1 on error.
*/
int EXPORT_API usb_relay_init(void);
/*Finalize the USB Relay Libary.
This function frees all of the static data associated with
USB Relay Libary. It should be called at the end of execution to avoid
memory leaks.
@returns:This function returns 0 on success and -1 on error.
*/
int EXPORT_API usb_relay_exit(void);
/*Enumerate the USB Relay Devices.*/
struct usb_relay_device_info EXPORT_API * usb_relay_device_enumerate(void);
/*Free an enumeration Linked List*/
void EXPORT_API usb_relay_device_free_enumerate(struct usb_relay_device_info*);
/*open device that serial number is serial_number*/
/*@return: This funcation returns a valid handle to the device on success or NULL on failure.*/
/*e.g: usb_relay_device_open_with_serial_number("abcde", 5")*/
int EXPORT_API usb_relay_device_open_with_serial_number(const char *serial_number, unsigned len);
/*open a usb relay device
@return: This funcation returns a valid handle to the device on success or NULL on failure.
*/
int EXPORT_API usb_relay_device_open(struct usb_relay_device_info* device_info);
/*close a usb relay device*/
void EXPORT_API usb_relay_device_close(int hHandle);
/*open a relay channel on the USB-Relay-Device
@paramter: index -- which channel your want to open
hHandle -- which usb relay device your want to operate
@returns: 0 -- success; 1 -- error; 2 -- index is outnumber the number of the usb relay device
*/
int EXPORT_API usb_relay_device_open_one_relay_channel(int hHandle, int index);
/*open all relay channel on the USB-Relay-Device
@paramter: hHandle -- which usb relay device your want to operate
@returns: 0 -- success; 1 -- error
*/
int EXPORT_API usb_relay_device_open_all_relay_channel(int hHandle);
/*close a relay channel on the USB-Relay-Device
@paramter: index -- which channel your want to close
hHandle -- which usb relay device your want to operate
@returns: 0 -- success; 1 -- error; 2 -- index is outnumber the number of the usb relay device
*/
int EXPORT_API usb_relay_device_close_one_relay_channel(int hHandle, int index);
/*close all relay channel on the USB-Relay-Device
@paramter: hHandle -- which usb relay device your want to operate
@returns: 0 -- success; 1 -- error
*/
int EXPORT_API usb_relay_device_close_all_relay_channel(int hHandle);
/*
status bit: High --> Low 0000 0000 0000 0000 0000 0000 0000 0000, one bit indicate a relay status.
the lowest bit 0 indicate relay one status, 1 -- means open status, 0 -- means closed status.
bit 0/1/2/3/4/5/6/7/8 indicate relay 1/2/3/4/5/6/7/8 status
@returns: 0 -- success; 1 -- error
*/
int EXPORT_API usb_relay_device_get_status(int hHandle, unsigned int *status);
#ifndef EXPORT_DLL
int EXPORT_API usb_relay_device_set_serial(int hHandle, char serial[5]);
#endif
#ifdef __cplusplus
}
#endif
#endif //end of ifdef __cplusplus需要将c接口转换成python接口，可以根据自己需要转换对应接口即可，不要全部转。下面是python的类。
import ctypes
import platform
from ctypes import *
import sys
import time
# 定义设备类型枚举
class usb_relay_device_type(c_int):
USB_RELAY_DEVICE_ONE_CHANNEL = 1
USB_RELAY_DEVICE_TWO_CHANNEL = 2
USB_RELAY_DEVICE_FOUR_CHANNEL = 4
USB_RELAY_DEVICE_EIGHT_CHANNEL = 8
# 定义设备信息结构体
class usb_relay_device_info(Structure):
pass # 前向声明
usb_relay_device_info._fields_ = [
("serial_number", POINTER(c_ubyte)), # unsigned char*
("device_path", c_char_p), # char*
("type", usb_relay_device_type), # 枚举值
("next", POINTER(usb_relay_device_info)) # 链表下一个节点
]
class USBRelayController:
def __init__(self):
"""初始化USB继电器控制器"""
# 检查是否为32位架构,dll为32bit的，需要32bit的python配合使用
arch, _ = platform.architecture()
if arch != '32bit':
print("错误: 只支持32位架构")
sys.exit(1)
self.usb_relay_lib = None
self.use_relay_fd = 0
self.device_info = None
self.dev_ptr = None
self.load_dll()
def load_dll(self):
"""加载DLL库"""
self.usb_relay_lib = ctypes.WinDLL("./usb_relay_device.dll")
# 设置函数原型
self.usb_relay_lib.usb_relay_device_enumerate.argtypes = []
self.usb_relay_lib.usb_relay_device_enumerate.restype = POINTER(usb_relay_device_info)
# 释放函数
self.usb_relay_lib.usb_relay_device_free_enumerate.argtypes = [POINTER(usb_relay_device_info)]
self.usb_relay_lib.usb_relay_device_free_enumerate.restype = None
self.usb_relay_lib.usb_relay_init.argtypes = []
self.usb_relay_lib.usb_relay_init.restype = c_int
self.usb_relay_lib.usb_relay_exit.argtypes = []
self.usb_relay_lib.usb_relay_exit.restype = None
self.usb_relay_lib.usb_relay_device_open_with_serial_number.argtypes = [c_char_p, c_uint]
self.usb_relay_lib.usb_relay_device_open_with_serial_number.restype = c_int
def enumerate_devices(self):
"""枚举所有USB继电器设备"""
devices = []
device_ptr = self.usb_relay_lib.usb_relay_device_enumerate()
if not device_ptr:
return devices
self.dev_ptr = device_ptr
current = device_ptr
while current:
info = current.contents
# 转换序列号(unsigned char*)为字符串
serial = []
p = info.serial_number
i = 0
while p[i] != 0:
serial.append(f"{p[i]:02X}")
i += 1
serial_str = "".join(serial)
devices.append({
"serial_number": serial_str,
"device_path": info.device_path.decode('utf-8') if info.device_path else "",
"type": info.type.value
})
current = info.next
if not devices:
print("未找到任何USB继电器设备")
else:
for i, dev in enumerate(devices):
print(f"设备 {i+1}:")
print(f" 序列号: {dev['serial_number']}")
print(f" 设备路径: {dev['device_path']}")
print(f" 类型: {dev['type']}通道继电器")
self.device_info = devices
def open_device(self, device_index=0):
"""打开指定设备"""
if not self.device_info:
self.enumerate_devices()
if not self.device_info or device_index >= len(self.device_info):
print("无效的设备索引")
return False
dev = self.device_info[device_index]
byte_data = bytes.fromhex(dev['serial_number'])
ascii_str = byte_data.decode('ascii')
self.use_relay_fd = self.usb_relay_lib.usb_relay_device_open_with_serial_number(byte_data, len(ascii_str))
if self.use_relay_fd == 0:
print("打开设备失败")
return False
return True
def control_relay(self, relay_index, relay_status):
"""控制继电器状态"""
if relay_status:
self.usb_relay_lib.usb_relay_device_open_one_relay_channel(self.use_relay_fd, relay_index)
else:
self.usb_relay_lib.usb_relay_device_close_one_relay_channel(self.use_relay_fd, relay_index)
def release(self):
"""释放资源"""
self.usb_relay_lib.usb_relay_device_free_enumerate(self.dev_ptr)
self.usb_relay_lib.usb_relay_exit()
if __name__ == '__main__':
# 示例用法
controller = USBRelayController()
controller.enumerate_devices()
if controller.open_device():
controller.control_relay(1, True)
time.sleep(2)
controller.control_relay(1, False)
controller.release()
1. 完成dll导入
2.c接口转成python接口
3.usb设备枚举
3. can 工具封装
基于周立功zcan-pro-4e-u提供的python库做二次封装，zlg的工具类比较齐全，我们稍微封装了下，方便我们调试和移植，
下面是参考代码：
依赖周立功的开发工具包，官网可以下载到。
#zlg zcan pro 4e-u
#基于zlg提供的py类做二次封装，存在的问题，连续打开设备会失败，此时需要插拔一次设备，即可恢复
#
import ctypes
import platform
from ctypes import *
import sys
import time
from zlgcan import *
PORT_CAN0 = 0
PORT_CAN1 = 0
PORT_CAN2 = 0
PORT_CAN3 = 0
class ZLGUsbCanControllers:
def __init__(self, port):
"""zlg 4eu 初始化"""
self.port = port
self.zcan_handler = ZCAN()
self.device_hanler = self.zcan_handler.OpenDevice(ZCAN_USBCAN_4E_U, 0, 0)
print("device handle:%d." %(self.device_hanler ))
if self.device_hanler == INVALID_DEVICE_HANDLE:
print("Open USBCAN-XE-U device failed!")
self.chn_handle = self.zcan_handler.can_E_U_start(self.zcan_handler, self.device_hanler, port)
print("Open USBCAN-XE-U device success!")
def can_deinit(self):
#Close CAN
ret = self.zcan_handler.ResetCAN(self.chn_handle)
if ret == 1:
print("ResetCAN success! ")
#Close Device
ret = self.zcan_handler.CloseDevice(self.chn_handle)
if ret == 1:
print("CloseDevice success! ")
def can_data_tx(self, can_id, data, num):
'''num为msg的条数，每个msg长度默认8字节'''
msgs = ZCAN_Transmit_Data()
msgs.transmit_type = 0 #0-正常发送，2-自发自收
msgs.frame.eff = 0 #0-标准帧，1-扩展帧
msgs.frame.rtr = 0 #0-数据帧，1-远程帧
msgs.frame.can_id = can_id
msgs.frame.can_dlc = 8
for i in range(8):
msgs.frame.data[i] = data[i]
self.zcan_handler.Transmit(self.chn_handle, msgs, num)
def can_data_rx(self):
"""can数据接收"""
rcv_num = self.zcan_handler.GetReceiveNum(self.chn_handle, ZCAN_TYPE_CAN) #返回值为接收缓存里面的CAN帧数
if rcv_num:
rcv_msg, rcv_num = self.zcan_handler.Receive(self.chn_handle, rcv_num)
return (rcv_msg, rcv_num)
return (None, None)
def zlg_init(port = 'can0'):
"""zlg can device usbcan 4e-u"""
global can_handler
global zcanlib
global chn_handle
global device_hanler
#ecan = ECAN(0, 0xFFF, log) #disable filter
zcanlib = ZCAN()
#stress_obj = StressTest(ecan, 1)
can_handler = zcanlib
#ecan.ReadThread()
device_hanler = zcanlib.OpenDevice(ZCAN_USBCAN_4E_U, 0, 0)
#print("handle = " + str(handle))
if device_hanler == INVALID_DEVICE_HANDLE:
print("Open USBCAN-XE-U device failed!")
print("Open USBCAN-XE-U device success!")
print("device handle:%d." %(device_hanler))
info = zcanlib.GetDeviceInf(device_hanler)
print("Device Information:\n%s" %(info))
#Start CAN
if port == 'can0':
chn_handle = zcanlib.can_E_U_start(zcanlib, device_hanler, 0)
if port == 'can1':
chn_handle = zcanlib.can_E_U_start(zcanlib, device_hanler, 1)
print("channel handle:%d." %(chn_handle))
if __name__ == '__main__':
# 示例用法
ZLGZCanPro = ZLGUsbCanControllers(PORT_CAN0)
while True:
rcv_msg, rcv_num = ZLGZCanPro.can_data_rx()
if rcv_num is not None:
for i in range(rcv_num):
print("[%d]:timestamps:%d,type:CAN, id:%s, dlc:%d, eff:%d, rtr:%d, data:%s" %(i, rcv_msg[i].timestamp,
hex(rcv_msg[i].frame.can_id), rcv_msg[i].frame.can_dlc,
rcv_msg[i].frame.eff, rcv_msg[i].frame.rtr,
''.join(hex(rcv_msg[i].frame.data[j])+ ' 'for j in range(rcv_msg[i].frame.can_dlc))))
time.sleep(0.5)
4.总结
有了这些工具结合中断用户使用场景可以做一些压力测试组合。
更多推荐
2
0- 0
已为社区贡献1条内容
所有评论(0)