Software triggerd VNA measurement

qutil/measurement/VNA/inst_E5071C.py

+# -*- coding: utf-8 -*-
+"""
+Created on Sun Oct  2 13:27:14 2022
+
+@author: Ran Xue
+
+modified according to Python Example Retrieve the Data as .csv for all the Traces at the Screen on a E5071C
+https://support.keysight.com/KeysightdCX/s/knowledge-article-detail?language=en_US&keyid=ka05a000000V6MY
+"""
+
+import numpy as np
+import pandas as pd
+import matplotlib.pyplot as plt
+import time
+import visa
+import csv
+
+class VNA:
+    def __init__(self, remote_address):
+        self.remote_address = remote_address
+        self.inst = None
+        self.timeout=50000
+
+    def connect_to_VNA(self):
+        rm=visa.ResourceManager()
+        # print(rm.list_resources())
+        self.inst = rm.open_resource(self.remote_address)
+
+    def identify_inst(self):
+        self.inst.write('*IDN?')
+        IDN = self.inst.read()
+        return IDN
+
+    def get_traces(self):
+
+        MatHeader=[]
+        MatAllData=[]
+        param, getDispMaxChannel = self.get_disp()
+
+        for Channel in range(1,getDispMaxChannel+1):
+            #print(':CALCulate'+str(Channel)+':PARameter1:SELect')
+            self.inst.write(':FORMat:DATA %s' % ('ASCii'))
+            self.inst.write(':CALCulate'+str(Channel)+':PARameter1:SELect')
+            MaxNumTraces = int(self.inst.query(':CALCulate'+str(Channel)+':PARameter:COUNt?'))
+
+            self.inst.write(':FORMat:DATA %s' % ('ASCii'))
+            Frequency = (self.inst.query_ascii_values(':SENSe'+str(Channel)+':FREQuency:DATA?'))
+            MatAllData.append(Frequency)
+
+            for Trace in range(1,MaxNumTraces+1):
+                #print(Trace)
+                param1 = self.inst.query(':SENSe1:SWEep:TYPE?')
+
+                self.inst.write(':FORMat:DATA %s' % ('ASCii'))
+                Data = (self.inst.query_ascii_values(':CALCulate'+str(Channel)+':TRACe'+str(Trace)+':DATA:FDATa?'))
+                #print(Data)
+                #print('Channel'+str(Channel)+'_Trace'+str(Trace)+'amp'+'_Trace'+str(Trace)+'phase')
+                MatHeader.append('Frequency_Channel'+str(Channel)+'_Trace'+str(Trace)+'amp'+'_Trace'+str(Trace)+'phase')
+
+                MatAllData.append(Data[0::2])
+                MatAllData.append(Data[1::2])
+
+        A=MatAllData
+        A=np.array(A).transpose()
+        Headers=' '.join([str(item) for item in MatHeader])
+        return Headers, A
+
+    def setup_measurement(self):
+        # 1 channel 4 traces S11, S12, S21, S22
+        self.inst.write(':CALC1:PAR:COUN 4')
+
+        self.inst.write(':CALC1:PAR1:DEF S11')
+        self.inst.write(':CALC1:PAR2:DEF S12')
+        self.inst.write(':CALC1:PAR3:DEF S21')
+        self.inst.write(':CALC1:PAR4:DEF S22')
+
+        sweep_type = self.inst.query(':SENSe1:SWEep:TYPE?')
+        if sweep_type[:-1] == 'LIN':
+            pass
+        else:
+            self.inst.write(':SENS1:SWE:TYPE LIN\n')
+
+    def make_single_trace(self):
+        # Trigger a measurement
+        trigger_mode = self.inst.query(':TRIG:SOUR?')
+        if trigger_mode[:-1] != 'BUS':
+            self.inst.write(':TRIG:SOUR BUS')
+
+        self.inst.write(':INIT1')  # initiate channel 1 to the idel state (wait for single trigger)
+        time.sleep(1)
+        self.inst.write(':TRIG:SING')  # trigger stimulator
+
+    def get_disp(self):
+        # Get the Shape of the display to know how many channels are needed to be scanned
+        param = self.inst.query(':DISPlay:SPLit?')
+        param = param[0:-1]#remove the \n at the end
+
+        if param=="D1":
+            getDispMaxChannel = 1
+        elif param=="1X1":
+            getDispMaxChannel = 1
+        elif param=="D12":
+            getDispMaxChannel = 2
+        elif param=="D1_2":
+            getDispMaxChannel = 2
+        elif param=="D112":
+            getDispMaxChannel = 2
+        elif param=="D1_1_2":
+            getDispMaxChannel = 2
+        elif param=="D1X2":
+            getDispMaxChannel = 2
+        elif param=="D2X1":
+            getDispMaxChannel = 2
+
+        elif param=="D123":
+            getDispMaxChannel = 3
+        elif param=="D1_2_3":
+            getDispMaxChannel = 3
+        elif param=="D3X1":
+            getDispMaxChannel = 3
+        elif param=="D12_33":
+            getDispMaxChannel = 3
+        elif param=="D11_23":
+            getDispMaxChannel = 3
+        elif param=="D13_23":
+            getDispMaxChannel = 3
+        elif param=="D12_13":
+            getDispMaxChannel = 3
+        elif param=="D1X3":
+            getDispMaxChannel = 3
+
+        elif param=="D1234":
+            getDispMaxChannel = 4
+        elif param=="D1_2_3_4":
+            getDispMaxChannel = 4
+        elif param=="D12_34":
+            getDispMaxChannel = 4
+        elif param=="D1X4":
+            getDispMaxChannel = 4
+        elif param=="D4X1":
+            getDispMaxChannel = 4
+        elif param=="D2X2":
+            getDispMaxChannel = 4
+
+        elif param=="D123_456":
+            getDispMaxChannel = 6
+        elif param=="D12_34_56":
+            getDispMaxChannel = 6
+        elif param=="D2X3":
+            getDispMaxChannel = 6
+        elif param=="D3X2":
+            getDispMaxChannel = 6
+
+        elif param=="D1234_5678":
+            getDispMaxChannel = 8
+        elif param=="D12_34_56_78":
+            getDispMaxChannel = 8
+        elif param=="D2X4":
+            getDispMaxChannel = 8
+        elif param=="D4X2":
+            getDispMaxChannel = 8
+
+        elif param=="D123_456_789":
+            getDispMaxChannel = 9
+        elif param=="D3X3":
+            getDispMaxChannel = 9
+
+        elif param=="D123__ABC":
+            getDispMaxChannel = 12
+        elif param=="D1234__9ABC":
+            getDispMaxChannel = 12
+        elif param=="D3X4":
+            getDispMaxChannel = 12
+        elif param=="D4X3":
+            getDispMaxChannel = 12
+
+        elif param=="D1234_DEFG":
+            getDispMaxChannel = 16
+        elif param=="D1234__CDEF":
+            getDispMaxChannel = 16
+        elif param=="D4X4":
+            getDispMaxChannel = 16
+
+        elif param=="D4X5":
+            getDispMaxChannel = 20
+
+        elif param=="D4X6":
+            getDispMaxChannel = 24
+
+        elif param=="D4X7":
+            getDispMaxChannel = 28
+
+        elif param=="D4X8":
+            getDispMaxChannel = 32
+
+        elif param=="D4X9":
+           getDispMaxChannel = 36
+
+        elif param=="D8X9":
+           getDispMaxChannel = 72
+        elif param=="D6X12":
+           getDispMaxChannel = 72
+
+        elif param=="D8X12":
+           getDispMaxChannel = 96
+
+        elif param=="D10X16":
+           getDispMaxChannel = 160
+
+        else:
+            getDispMaxChannel=1
+
+        return param, getDispMaxChannel
+
+    def disconnect_inst(self):
+        self.inst.close()
+        rm.close()

qutil/measurement/VNA/scheduled_measurement.py

+# -*- coding: utf-8 -*-
+"""
+Created on Sun Oct  2 13:14:22 2022
+
+@author: Ran Xue
+"""
+
+#For scheduling task execution
+import schedule
+import time
+import inst_E5071C
+from pathlib import Path
+import numpy as np
+
+timeformat = "%Y_%m_%d_%H%M%S"
+datapath = Path(r'C:\Users\lablocal\Documents\Measurement_QuBus\setup_characterization\VNA_measurement')
+
+def make_scheduled_single_trace():
+    VNA = inst_E5071C.VNA('TCPIP::169.254.40.9::INSTR') # instantiate a VNA
+    VNA.connect_to_VNA()
+
+    if VNA.identify_inst() is None:
+        try:
+            VNA.connect_to_VNA()
+            idn = VNA.identify_inst()
+            print(f'refresh connection to VNA: {idn}')
+        except:
+            raise ConnectionError('No such an instrument found! Please check the connection first.')
+
+    VNA.setup_measurement()
+    VNA.make_single_trace()
+    headers, data = VNA.get_traces()
+
+    timestamp = time.strftime(timeformat)
+    filename = datapath / f'{timestamp}_data.csv'
+    np.savetxt(filename, data, delimiter=',', fmt='%5s', header= headers)
+    print(f'[{timestamp}] - VNA measurement saved.')
+
+print('>>> Start a scheduled measurement: VNA trace during cooling down <<<')
+schedule.every(10).minutes.do(make_scheduled_single_trace)
+print('Sleep for 10 min')
+
+while True:
+    schedule.run_pending()
+    time.sleep(1)
\ No newline at end of file