实现自动化控制,本次为大家讲解综测仪MT8862A的控制逻辑。
新建底层控制逻辑

在文件basis_contorl.py中写入仪器控制底层代码,代码如下:
import tkinter.messagebox
import pyvisa
class InstrumentControl(object):
    inst = None
    def __init__(self, equipment_name: str, equipment_visa: str, visa_dll='c:/Windows/System32/visa32.dll'):
        """
        建立仪器初始对象
        @param equipment_name: 仪器名
        @param equipment_visa: 仪器visa地址
        @param visa_dll: visa.all地址
        """
        try:
            self.rm = pyvisa.ResourceManager(visa_dll)
            print("打开visa32.dll文件成功")
            self.inst = self.open_by_name(equipment_name, equipment_visa)
        except Exception as e:
            tkinter.messagebox.showerror('错误', '在{}中找不到visa32.dll,请确认正确安装了NI-MAX和驱动')
            print(f"打开visa32.dll文件失败, 报错原因:{e}")
        pass
    def open_by_name(self, device_name, gpib_ip):
        """
        仪表通讯
        @param device_name:
        @param gpib_ip:
        @return: 仪器控制对象
        """
        self.inst = self.rm.open_resource(gpib_ip)
        print(self.inst)
        print("打开资源:" + gpib_ip)
        query_name_str = self.command_query("*IDN?")
        print("查询名称为:" + query_name_str)
        if device_name in query_name_str:
            print("打开仪表成功,打开的是:" + device_name)
            print("打开仪表成功,打开的是:" + device_name)
            return self.inst
        return self.inst
    def command_write(self, write_cmd):
        """
        命令写入
        @param write_cmd: 写入指令
        @return: 执行结果
        """
        print("执行了写入指令:" + write_cmd)
        ret_res = self.inst.write(write_cmd)
        print("返回了:" + str(ret_res))
        return ret_res
    def command_read(self, read_cmd):
        """
        命令读取
        @param read_cmd: 读取指令
        @return: 读取信息
        """
        print("执行了读取指令:" + read_cmd)
        ret_res = self.inst.read(read_cmd)
        print("返回了:" + str(ret_res))
        return ret_res
    def command_query(self, query_cmd):
        """
        命令查询
        @param query_cmd: 查询指令
        @return: 查询信息
        """
        print("执行了查询指令:" + query_cmd)
        ret_res = self.inst.query(query_cmd).replace('\n', '')
        print("返回了:" + str(ret_res))
        return ret_res
编写好底层控制代码,在文件instrument_MT8862A.py写入综测仪MT8862A的控制指令,代码如下:
from instrument_control.basis_contorl import InstrumentControl
class MT8862A(InstrumentControl):
    def __init__(self, equipment_name: str, equipment_visa: str):
        super(MT8862A, self).__init__(equipment_name, equipment_visa)
    def reset_inst(self):
        """
        初始化仪表
        """
        self.command_write('*RST;*OPC?')
        self.command_write('*CLS;*OPC?')
        self.command_write('NWTYPE AP;*OPC?')
        self.command_write('MODE RXMODE;*OPC?')
        self.command_write('ILVL 12;*OPC?')
        self.command_write('OLVL -50;*OPC?')
    def query_equipment(self):
        """
        查询设备的sn
        :return: 设备sn
        """
        result = self.command_query("*IDN?")
        return result
    def set_ppuu_type(self, ppuu_type: str, standard: str):
        """
        设置带宽
        @param ppuu_type: 20MHZ
        @param standard: 测试制式
        """
        self.command_write(f"PPDUTYPE_{standard.upper()} {ppuu_type};*ESR?")
    def query_ppuu_type(self, standard: str):
        """
        查询带宽
        @return: 查询结果
        @param standard: 测试制式
        """
        result = self.command_query(f"PPDUTYPE_{standard.upper()}?;*ESR?")
        return result
    def set_mcs_index(self, mcs_index: str, standard: str):
        """
        设置速率
        @param mcs_index: 0,1。。。。。
        @param standard: 测试制式
        """
        self.command_write(f"MCSINDEX_{standard.upper()} {mcs_index};*ESR?")
    def query_mcs_index(self, standard: str):
        """
        查询速率
        @return: 查询结果
        @param standard: 测试制式
        """
        result = self.command_query(f"MCSINDEX_{standard.upper()}?")
        return result
    def set_channel_band(self, channel_band: str, standard: str):
        """
        设置2/5g信道或6g信道
        @param channel_band: 6G,2PT4G_5G
        @param standard: 测试制式
        """
        self.command_write(f"CHANNELBAND_{standard.upper()} {channel_band};*ESR?")
    def query_channel_band(self, standard: str):
        """
        查询信道模式
        @return: 查询结果
        @param standard: 测试制式
        """
        result = self.command_query(f"CHANNELBAND_{standard.upper()}?")
        return result
    def set_wifi_ssid(self, wifi_ssid):
        """
        设置WiFi的ssid
        @param wifi_ssid: WiFi名字
        """
        self.command_write(f"NWSSID {wifi_ssid}")
    def query_wifi_ssid(self):
        """
        查询WiFi的ssid
        @return: 查询结果
        """
        result = self.command_query(f"NWSSID?")
        return result
    def set_security_type(self, security_type: str):
        """
        设置加密方式
        @param security_type:WPA3
        """
        self.command_write(f"SECURITY_TYPE_AP {security_type};*ESR?")
    def query_security_type(self):
        """
        查询加密方式
        @return: 查询结果
        """
        result = self.command_query("SECURITY_TYPE_AP?")
        return result
    def set_psk_passphrase(self, psk_passphrase: str):
        """
        设置密码
        @param psk_passphrase: 12345678
        """
        self.command_write(f'SECURITY_PSKPASSPHRASE_WPA3_AP "{psk_passphrase}";*ESR?')
    def query_psk_passphrase(self):
        """
        查询密码
        @return: 查询结果
        """
        result = self.command_query("SECURITY_PSKPASSPHRASE_WPA3_AP?")
        return result
    def set_country_state(self, country_state: str):
        """
        设置是否支持国家码
        @param country_state: ENABLE
        """
        self.command_write(f"COUNTRY_STATE {country_state};*ESR?")
    def query_country_state(self):
        """
        查询是否支持国家码
        @return:
        """
        result = self.command_query("COUNTRY_STATE?")
        return result
    def set_country_code_mode(self, country_code_mode: str):
        """
        设置国家码模式
        @param country_code_mode: STR
        """
        self.command_write(f"COUNTRY_REGION_MODE {country_code_mode};*ESR?")
    def query_country_code_mode(self):
        """
        查询国家码模式
        """
        result = self.command_query(f"COUNTRY_REGION_MODE?")
        return result
    def set_country_code_one(self, country_code: str):
        """
        设置国家码
        @param country_code: CN
        """
        self.command_write(f"COUNTRY_REGION {country_code};*ESR?")
    def query_country_code_one(self):
        """
        查询国家码
        """
        result = self.command_query(f"COUNTRY_REGION?")
        return result
    def set_loss_state(self, state: str):
        """
        设置线损开关状态
        @param state: 开:ON  关:OFF
        """
        self.command_write(f'EXTLOSSW {state}')
    def set_rate_11g(self, rate: int):
        """
        设置802.11g速率
        @param rate: 速率
        """
        self.command_write(f'DATARATE_G {rate}')
    def set_rate_11b(self, rate):
        """
        设置802.11b速率
        @param rate: 速率
        """
        self.command_write(f'DATARATE_B {rate}')
    def set_rate_11a(self, rate):
        """
        设置802.11a速率
        @param rate: 速率
        """
        self.command_write(f'DATARATE_A {rate}')
    def set_wifi_system(self, tech: str):  # 11a-5G 11b-2g
        """
        设置制式
        @param tech: 802.11a:A, 802.11b:B, 802.11g:G, 802.11n:N, 802.11ac:AC, 802.11ax:AX
        """
        cmd = 'WLANSTD {}'.format(str(tech).upper())
        self.command_write(cmd)
    def query_wifi_system(self):
        """
        查询设置制式
        @return: 返回制式
        """
        result = self.command_query("WLANSTD?")
        return result
    def set_rx_or_tx_mode(self, mode: str):
        """
        更改RX mode或TX mode
        @param mode: tx:TXMODE rx:RXMODE
        """
        self.command_write('MODE {}'.format(mode))
    def set_data(self):
        """
        更新目标参数
        """
        cmd = "NWUPDATE;*ESR?"
        self.command_write(cmd)
    def set_channel(self, ch: int):
        """
        频段设置
        @param ch: 测试频段
        """
        cmd = 'CHANNELNUM {}'.format(str(ch))
        self.command_write(cmd)
    def set_start_and_end_level(self, start_level: str, end_level: str):
        """
        设置start_level 和 end_level
        @param start_level: 起始电平值
        @param end_level: 结束电平值
        """
        self.command_write(f'RXSEARCH_LVL {start_level},{end_level}')
    def set_output_level_one(self, level: str):
        """
        设置输出电平
        @param level: 电平值
        """
        self.command_write(f"OLVL {level};*ESR?")
    def query_output_level_one(self):
        """
        查询电平值
        @return:
        """
        result = self.command_query("OLVL?;*ESR?")
        return result
    def set_input_level_one(self, level: str):
        """
        设置输入电平
        @param level: 电平值
        """
        self.command_write(f"ILVL {level};*ESR?")
    def query_input_level_one(self):
        """
        查询电平值
        @return:
        """
        result = self.command_query("ILVL?;*ESR?")
        return result
    def set_primary_channel(self, primary_channel: str):
        """
        设置测试信道
        @param primary_channel: 0,1.。。。
        """
        self.command_write(f"CHANNELNUM {primary_channel};*ESR?")
    def query_primary_channel(self):
        """
        查询测试信道
        @return: 查询结果
        """
        result = self.command_query("CHANNELNUM?;*ESR?")
        return result
    def set_assign_ip_address(self, assign_ip_address: str):
        """
        设置assign_ip_address
        @param assign_ip_address: MANUAL
        """
        self.command_write(f"IPPROP {assign_ip_address};*ESR?")
    def query_assign_ip_address(self):
        """
        查询assign_ip_address
        @return: 查询结果
        """
        result = self.command_query("IPPROP?")
        return result
    def set_dut_ip_address(self, dut_ip_address: str):
        self.command_write("")
    def set_anti_seize(self):
        """
        卡住重连
        """
        self.command_write("ABORT;*ESR?")
        self.command_write("DISCONNECT;*ESR?")
    def get_ack(self):
        """
        查询PER(丢包率)测量值
        @return: 丢包率
        """
        ret = self.command_query('MEAS_PER?').split(',')[1]
        return ret
    def check_connect(self):
        """
        查询wifi连接状态
        @return: Idle:0 Scanning:2 Connecting:6 Connected:7 Disable:9
        """
        try:
            ret = self.command_query('NWSTAT?').replace('\r', '')
        except:
            ret = None
        return ret
    def set_pkt(self, pkt: int):
        """
        设置执行PER测试时MT8862A传输的数据包数
        @param pkt: 数据包数
        """
        self.command_write('PERPKTS {}'.format(pkt))
    def single_on(self):
        """
        发射WIFI信号 (前提需要先连接上仪表WIFI 带8862A字样)
        """
        self.command_write('SNGLS')
    def query_pkt_statue(self):
        """
        查询测试状态
        @return: 测试状态
        """
        pkt_statue = self.command_query("MPSTAT?")
        return pkt_statue
    def query_test_statue(self):
        """
        查询测试状态
        @return: 测试状态
        """
        test_statue = self.command_query("MSTAT?")
        return test_statue
    def set_test_step(self, step: str):
        """
        设置测试步进
        @param step: 测试步进
        """
        self.command_write("RXSEARCH_STEP {}".format(step))
    def query_test_step(self):
        """
        查询测试步进
        @return: 测试步进
        """
        result = self.command_query("RXSEARCH_STEP?")
        return result
    def query_line_loss(self):
        """
        查询输出电损
        @return: 输出电损
        """
        result = self.command_query("LOSSTBLVAL_LIST?")
        return result
    def set_loss(self, loss_value: float, cent_freq: int, port: str):
        """
        设置对应频段线损
        :param loss_value: 线损值
        :param cent_freq: 测试频段
        :param port: 输出端口,Main1或Main2
        """
        if 'MAIN1' in port:
            self.command_write(f'LOSSTBLVAL_LIST {cent_freq}MHZ,{loss_value},{loss_value},0.00,0.00,0.00;*ESR?')
        else:
            self.command_write(f'LOSSTBLVAL_LIST {cent_freq}MHZ,0.00,0.00,0.00,{loss_value},{loss_value};*ESR?')
    def set_lvl_status(self, status: str):
        """
        设置lvl状态
        @param status: ON为开,OFF为关
        """
        self.command_write(f"LVL {status};*ESR?")
    def disconnect_wifi(self):
        """
        断开WiFi
        """
        self.command_write("DISCONNECT;*ESR?")
    def set_wifi_sign(self):
        self.command_write("TRGSRC COM;*ESR?")
        self.command_write("COMTRGMODE MODE3;*ESR?")
        self.command_write("TXDATARATE_CTRL MODEB;*ESR?")
    def set_output_port(self, port: str):
        """
        设置输出端口
        @param port:
        """
        self.command_write(f"PORT MAIN{port};*ESR?")
    def query_output_port(self):
        """
        查询输出端口
        @return: 返回端口结果
        """
        result = self.command_query(f"PORT?")
        return result
控制仪器时先设置控制对象,具体例子如下:
    set_equipment_name = "MT8862A"
    set_equipment_visa = "仪器的GPIB地址"
    d66319_object = MT8862A(equipment_name=set_equipment_name, equipment_visa=set_equipment_visa)
关于综测仪MT8862A的控制方法就到这里了,博主后续还会更新不同仪器的控制方法,各位敬请期待吧,有疑问欢迎找博主解答,我是活动的笑脸。



















