1096 lines
66 KiB
Python
1096 lines
66 KiB
Python
from PyQt6.QtGui import *
|
|
from PyQt6.QtWidgets import *
|
|
from PyQt6.QtCore import *
|
|
|
|
import multiprocessing
|
|
import multiprocessing.managers
|
|
|
|
import time
|
|
from timeit import default_timer as timer
|
|
from datetime import datetime
|
|
import collections
|
|
from simple_pid import PID
|
|
import traceback, sys, os
|
|
import numpy as np
|
|
import pyqtgraph as pg
|
|
from design_files.Main_design import Ui_MainWindow
|
|
|
|
# Get the current script's directory
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
# Get the parent directory by going one level up
|
|
parent_dir = os.path.dirname(current_dir)
|
|
# Add the parent directory to sys.path
|
|
sys.path.append(parent_dir)
|
|
from scripts import import_txt
|
|
from scripts import analyse_cooldown_tools
|
|
|
|
|
|
class WorkerSignals(QObject):
|
|
'''
|
|
Defines the signals available from a running worker thread.
|
|
Supported signals are:
|
|
finished: No data
|
|
error: tuple (exctype, value, traceback.format_exc() )
|
|
result: object data returned from processing, anything
|
|
progress: int indicating % progress
|
|
'''
|
|
finished = pyqtSignal()
|
|
error = pyqtSignal(tuple)
|
|
result = pyqtSignal(object)
|
|
progress = pyqtSignal(list)
|
|
|
|
|
|
class Worker(QRunnable):
|
|
'''
|
|
Worker thread
|
|
Inherits from QRunnable to handler worker thread setup, signals and wrap-up.
|
|
:param callback: The function callback to run on this worker thread. Supplied args and
|
|
kwargs will be passed through to the runner.
|
|
:type callback: function
|
|
:param args: Arguments to pass to the callback function
|
|
:param kwargs: Keywords to pass to the callback function
|
|
'''
|
|
|
|
def __init__(self, fn, *args, **kwargs):
|
|
super(Worker, self).__init__()
|
|
|
|
# Store constructor arguments (re-used for processing)
|
|
self.fn = fn
|
|
self.args = args
|
|
self.kwargs = kwargs
|
|
self.signals = WorkerSignals()
|
|
|
|
# Add the callback to our kwargs
|
|
self.kwargs['progress_callback'] = self.signals.progress
|
|
|
|
@pyqtSlot()
|
|
def run(self):
|
|
'''
|
|
Initialise the runner function with passed args, kwargs.
|
|
'''
|
|
|
|
# Retrieve args/kwargs here; and fire processing using them
|
|
try:
|
|
result = self.fn(*self.args, **self.kwargs)
|
|
except:
|
|
traceback.print_exc()
|
|
exctype, value = sys.exc_info()[:2]
|
|
self.signals.error.emit((exctype, value, traceback.format_exc()))
|
|
else:
|
|
self.signals.result.emit(result) # Return the result of the processing
|
|
finally:
|
|
self.signals.finished.emit() # Done
|
|
|
|
def get_float(Qline,default = 0): #gets value from QLineEdit and converts it to float. If text is empty or cannot be converted, it returns "default" which is 0, if not specified
|
|
try:
|
|
out = float(Qline.text())
|
|
except:
|
|
out = default
|
|
return(out)
|
|
|
|
def update_single_entry(dict,name,ind,val):
|
|
'''updates a single value in global vars. To do so it gets the current value of "dict('name')", replaces "val" at indec "ind" and sends this back'''
|
|
data = dict.get(f"{name}") #get data
|
|
data[ind] = val #replace entry
|
|
dict.update({f"{name}":data}) #send data back
|
|
|
|
class MainWindow(QMainWindow, Ui_MainWindow):
|
|
def __init__(self, *args, **kwargs):
|
|
# Get the current script's directory
|
|
self.current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
# Get the parent directory by going one level up
|
|
self.parent_dir = os.path.dirname(current_dir)
|
|
|
|
#establish connection to global variables}
|
|
try: #try to connect to global variables
|
|
manager = multiprocessing.managers.BaseManager(address=('localhost',5001), authkey=b'')
|
|
manager.connect()
|
|
manager.register('syncdict')
|
|
manager.register('sync_BK_9174B')
|
|
manager.register('sync_BK_9174B_2')
|
|
manager.register('sync_BK_9132B_2')
|
|
manager.register('sync_BK_9132B')
|
|
manager.register('sync_BK_9131B')
|
|
manager.register('sync_Keithley_2230G')
|
|
manager.register('sync_LS_336')
|
|
manager.register('sync_LS_218')
|
|
manager.register('sync_imc')
|
|
manager.register('sync_converted')
|
|
manager.register('sync_main')
|
|
self.syncdict = manager.syncdict()
|
|
self.sync_BK_9174B = manager.sync_BK_9174B()
|
|
self.sync_BK_9174B_2 = manager.sync_BK_9174B_2()
|
|
self.sync_BK_9132B_2 = manager.sync_BK_9132B_2()
|
|
self.sync_BK_9132B = manager.sync_BK_9132B()
|
|
self.sync_BK_9131B = manager.sync_BK_9131B()
|
|
self.sync_Keithley_2230G = manager.sync_Keithley_2230G()
|
|
self.sync_LS_336 = manager.sync_LS_336()
|
|
self.sync_LS_218 = manager.sync_LS_218()
|
|
self.sync_imc = manager.sync_imc()
|
|
self.sync_converted = manager.sync_converted()
|
|
self.sync_main = manager.sync_main()
|
|
except: #open global variables, if no connection can be made (i.e. it is not running). Then connect to it
|
|
# subprocess.call(['D:\\Python instrument drivers\\env\\Scripts\\python.exe', 'D:\\Python instrument drivers\\StandAlones\\global_variables.py'])
|
|
self.global_vars = QProcess()
|
|
self.global_vars.start(self.current_dir+"\\.venv\\Scripts\\python.exe", [self.current_dir+'\\global_variables_TF.py'])
|
|
manager.connect()
|
|
manager.register('syncdict')
|
|
manager.register('sync_BK_9174B')
|
|
manager.register('sync_BK_9174B_2')
|
|
manager.register('sync_BK_9132B_2')
|
|
manager.register('sync_BK_9132B')
|
|
manager.register('sync_BK_9131B')
|
|
manager.register('sync_Keithley_2230G')
|
|
manager.register('sync_LS_336')
|
|
manager.register('sync_LS_218')
|
|
manager.register('sync_imc')
|
|
manager.register('sync_converted')
|
|
manager.register('sync_main')
|
|
self.syncdict = manager.syncdict()
|
|
self.sync_BK_9174B = manager.sync_BK_9174B()
|
|
self.sync_BK_9174B_2 = manager.sync_BK_9174B_2()
|
|
self.sync_BK_9132B_2 = manager.sync_BK_9132B_2()
|
|
self.sync_BK_9132B = manager.sync_BK_9132B()
|
|
self.sync_BK_9131B = manager.sync_BK_9131B()
|
|
self.sync_Keithley_2230G = manager.sync_Keithley_2230G()
|
|
self.sync_LS_336 = manager.sync_LS_336()
|
|
self.sync_LS_218 = manager.sync_LS_218()
|
|
self.sync_imc = manager.sync_imc()
|
|
self.sync_converted = manager.sync_converted()
|
|
self.sync_main = manager.sync_main()
|
|
print('!!!\nI opened global variables myself. If you close me, global variables will shut down too. Consider starting global variables in own instance for more security\n!!!')
|
|
|
|
#fill in variables, in case they are not defined in global variables
|
|
self.sync_LS_336.update({'T':[0,0,0,0]})
|
|
self.sync_LS_218.update({'T':[0,0,0,0,0,0,0,0]})
|
|
self.sync_imc.update({'FG':[0,0,0]})
|
|
self.sync_converted.update({'T':[0,0,0,0,0,0,0,0]})
|
|
self.sync_converted.update({'AMR_x':[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]})
|
|
self.sync_converted.update({'AMR_y':[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]})
|
|
self.sync_converted.update({'AMR_z':[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]})
|
|
self.sync_BK_9132B.update({'P':[0,0,0], 'U':[0,0,0], 'I': [0,0,0]})
|
|
self.sync_BK_9132B_2.update({'P':[0,0,0], 'U':[0,0,0], 'I': [0,0,0]})
|
|
self.sync_BK_9131B.update({'P':[0,0,0], 'U':[0,0,0], 'I': [0,0,0]})
|
|
# self.sync_main.update({'Measurement_Running':False, 'T':[0,0], 'Points_Path':'', 'T_set':[0,0], 'T_reset':False, 'Cooldown_Speed':0, 'Start_ramp':False, 'Stop_ramp':False})
|
|
|
|
#import Gui from QT designer file
|
|
super(MainWindow, self).__init__(*args, **kwargs)
|
|
self.setupUi(self)
|
|
|
|
#setup plot
|
|
self.plot_colors = ['#a6cee3', '#1f78b4','#b2df8a','#33a02c','#fb9a99','#e31a1c','#fdbf6f','#ff7f00','#cab2d6','#6a3d9a','#e7298a','#b15928','#016c59','#d9d9d9','#000000'] #list of colors for plots
|
|
self.plot_widgets = [self.graphWidget_T,self.graphWidget_Fluxgate,self.graphWidget_AMR_x,self.graphWidget_AMR_y,self.graphWidget_AMR_z] #list of all plot widgets for easier handling
|
|
self.plots_T = [i for i in range(11)] #will contain all plots in temperature grapgh widget. Is already the correct lengths, so the right amount of plots are created for the widget. Contains numbers from 0 to the number of plots so the elements can also be used as index below.
|
|
self.plots_fluxgate = [i for i in range(4)] #will contain all plots in temperature grapgh widget. Is already the correct lengths, so the right amount of plots are created for the widget. Contains numbers from 0 to the number of plots so the elements can also be used as index below.
|
|
self.plots_AMR_x = [i for i in range(15)] #will contain all plots in temperature grapgh widget. Is already the correct lengths, so the right amount of plots are created for the widget. Contains numbers from 0 to the number of plots so the elements can also be used as index below.
|
|
self.plots_AMR_y = [i for i in range(15)] #will contain all plots in temperature grapgh widget. Is already the correct lengths, so the right amount of plots are created for the widget. Contains numbers from 0 to the number of plots so the elements can also be used as index below.
|
|
self.plots_AMR_z = [i for i in range(15)] #will contain all plots in temperature grapgh widget. Is already the correct lengths, so the right amount of plots are created for the widget. Contains numbers from 0 to the number of plots so the elements can also be used as index below.
|
|
self.plot_sets = [self.plots_T, self.plots_fluxgate, self.plots_AMR_x, self.plots_AMR_y, self.plots_AMR_z]#list of all plot-sets. Each plot-set contains all plots of one graph widget
|
|
self.pens = [pg.mkPen(color = c, width = 2) for c in self.plot_colors]
|
|
self.titles = ['Temperature', 'Fluxgates', 'AMR x', 'AMR y', 'AMR z'] #Plot titles
|
|
self.ylabel = ['Temperature [K]', 'B [µT]', 'B [µT]', 'B [µT]', 'B [µT]'] #y axis labels for plots
|
|
self.axis = [] #list of axis. They are needed to set x axis to time. List is filled below
|
|
|
|
for i in range(5): #fill list with axis
|
|
self.axis.append(pg.DateAxisItem())
|
|
|
|
for widget, title, label, axis, plot_set in zip(self.plot_widgets, self.titles, self.ylabel, self.axis, self.plot_sets):
|
|
#setup background, title, labels, grid, and time axis
|
|
widget.setBackground('w')
|
|
widget.setTitle(title)
|
|
widget.setLabel('left', label)
|
|
widget.setLabel('bottom', 'Temperature')
|
|
widget.showGrid(x= True, y= True, alpha = 0.5)
|
|
widget.setAxisItems({'bottom':axis})
|
|
|
|
temp = [time.time(),time.time()-1]
|
|
|
|
for i in plot_set: #since plot_set so far just contains numbers from 0 to the number of desired plots, the elements can also be used as index
|
|
plot_set[i] = (widget.plot(temp,[1,0],pen = self.pens[i]))
|
|
plot_set[i].clear()
|
|
|
|
|
|
#define signals and slots
|
|
self.action_save_default.triggered.connect(self.save_default)
|
|
self.action_load_default.triggered.connect(self.load_default)
|
|
self.checkBox_monitoring.stateChanged.connect(self.set_monitoring)
|
|
self.SB_plot_points.valueChanged.connect(self.set_Npoints)
|
|
self.checkBox_plots_disable.stateChanged.connect(self.disable_plots)
|
|
self.dSB_timing_main.valueChanged.connect(self.set_timing_Main)
|
|
self.dSB_timing_monitoring.valueChanged.connect(self.set_timing_monitoring)
|
|
self.Button_change_Buffer.clicked.connect(self.change_buffer_size)
|
|
self.Button_imc.clicked.connect(self.set_imc)
|
|
self.Button_LS.clicked.connect(self.set_LS)
|
|
self.list_T.itemSelectionChanged.connect(self.plot_List_T_changed)
|
|
self.list_FG.itemSelectionChanged.connect(self.plot_List_FG_changed)
|
|
self.list_AMR_x.itemSelectionChanged.connect(self.plot_List_AMR_x_changed)
|
|
self.list_AMR_y.itemSelectionChanged.connect(self.plot_List_AMR_y_changed)
|
|
self.list_AMR_z.itemSelectionChanged.connect(self.plot_List_AMR_z_changed)
|
|
self.checkBox_autoscale_T.stateChanged.connect(self.set_autoscale)
|
|
self.checkBox_autoscale_FG.stateChanged.connect(self.set_autoscale)
|
|
self.checkBox_autoscale_AMR_x.stateChanged.connect(self.set_autoscale)
|
|
self.checkBox_autoscale_AMR_y.stateChanged.connect(self.set_autoscale)
|
|
self.checkBox_autoscale_AMR_z.stateChanged.connect(self.set_autoscale)
|
|
self.SB_T_channel_top.valueChanged.connect(self.set_TB_channel)
|
|
self.SB_T_channel_bot.valueChanged.connect(self.set_TB_channel)
|
|
self.Button_PID_on.clicked.connect(self.activate_PID)
|
|
self.Button_start_ramp.clicked.connect(self.start_ramp)
|
|
self.Button_stop_ramp.clicked.connect(self.stop_ramp_pushed)
|
|
self.Button_reset_T_set.clicked.connect(self.reset_set_T)
|
|
self.dSB_PID_P.valueChanged.connect(self.update_PIDs)
|
|
self.dSB_PID_I.valueChanged.connect(self.update_PIDs)
|
|
self.dSB_PID_D.valueChanged.connect(self.update_PIDs)
|
|
self.dSB_PID_P_2.valueChanged.connect(self.update_PIDs)
|
|
self.dSB_PID_I_2.valueChanged.connect(self.update_PIDs)
|
|
self.dSB_PID_D_2.valueChanged.connect(self.update_PIDs)
|
|
self.dSB_P_max_top.valueChanged.connect(self.update_PIDs)
|
|
self.dSB_P_max_bot.valueChanged.connect(self.update_PIDs)
|
|
self.dSB_R_top.valueChanged.connect(self.update_PIDs)
|
|
self.dSB_R_bot.valueChanged.connect(self.update_PIDs)
|
|
self.Button_set_I0.clicked.connect(self.set_I0)
|
|
self.Button_set_I1.clicked.connect(self.set_I1)
|
|
self.Button_set_I2.clicked.connect(self.set_I2)
|
|
self.Button_get_I0.clicked.connect(self.get_I0)
|
|
self.Button_get_I1.clicked.connect(self.get_I1)
|
|
self.Button_get_I2.clicked.connect(self.get_I2)
|
|
self.Button_Coils_off.clicked.connect(self.turn_off_coils)
|
|
self.Button_set_B.clicked.connect(self.set_B)
|
|
self.checkBox_LED.stateChanged.connect(self.set_LED)
|
|
self.checkBox_He_heater.stateChanged.connect(self.switch_he_heater)
|
|
self.Button_he_heater_measure_R.clicked.connect(self.measure_R_pressed)
|
|
self.dSB_He_heater_P_set.valueChanged.connect(self.switch_he_heater)
|
|
self.checkBox_AMR_supply.stateChanged.connect(self.switch_AMR_supply)
|
|
self.checkBox_AMR_flip.stateChanged.connect(self.AMR_flip)
|
|
self.checkBox_Relais_Fan.stateChanged.connect(self.switch_fan)
|
|
|
|
|
|
|
|
#define constants
|
|
self.running = True #true while app is running
|
|
self.timing_Main = 0.1 #wait time in function "update_Data".
|
|
self.monitor = False # Write monitoring data yes or no
|
|
self.timing_monitoring = 1 #Save intervall of monitoring function
|
|
self.T_imc = False #use imc as source for temperature data
|
|
self.buffer_size = 36000 #size of ringbuffers for plotting. 36000 is roughly 1 hour, if timng_main is 0.1 s
|
|
self.buffer = collections.deque(maxlen=self.buffer_size)
|
|
self.new_Data = []
|
|
self.plot_data_T = [[] for i in range(12)] #list of lists which will contain the selected points for plotting for T plots.
|
|
self.plot_data_FG = [[] for i in range(4)] #list of lists which will contain the selected points for plotting for Fluxgate plots.
|
|
self.plot_data_AMR_x = [[] for i in range(15)] #list of lists which will contain the selected points for plotting for AMR_x plots.
|
|
self.plot_data_AMR_y = [[] for i in range(15)] #list of lists which will contain the selected points for plotting for AMR_y plots.
|
|
self.plot_data_AMR_z = [[] for i in range(15)] #list of lists which will contain the selected points for plotting for AMR_z plots.
|
|
self.show_plot_T = [i for i in range(12)] #list of all T_plots that are active. This is needed for autoscaling. It is changed by self.plot_List_T_changed.
|
|
self.show_plot_FG = [i for i in range(4)] #list of all FG_plots that are active. This is needed for autoscaling. It is changed by self.plot_List_FG_changed.
|
|
self.show_plot_AMR_x = [i for i in range(15)] #list of all AMR_x_plots that are active. This is needed for autoscaling. It is changed by self.plot_List_AMR_x_changed.
|
|
self.show_plot_AMR_y = [i for i in range(15)] #list of all AMR_y_plots that are active. This is needed for autoscaling. It is changed by self.plot_List_AMR_y_changed.
|
|
self.show_plot_AMR_z = [i for i in range(15)] #list of all AMR_z_plots that are active. This is needed for autoscaling. It is changed by self.plot_List_AMR_z_changed.
|
|
self.show_plot_all = [self.show_plot_T,self.show_plot_FG,self.show_plot_AMR_x,self.show_plot_AMR_y,self.show_plot_AMR_z] #list containing all show plot list. Also needed for autoscaling
|
|
self.buffer_t = collections.deque(maxlen=self.buffer_size) #ringbuffer for times
|
|
self.Npoints = 200 #number of points to plot
|
|
self.ind_plot = [-i for i in range(1,self.Npoints+1)] #list containing all indices which are to be plotted
|
|
self.yrange = [[] for i in range(5)] #list of lists which will contain the y-axis ranges for all plots
|
|
self.autoscale = [True,True,True,True,True] #Wether to scale the plots or not
|
|
self.TB_channels = [0,7] #index of cernox used for PID controllers([top, bottom]).
|
|
self.PID_top = PID(1, 0.1, 0.0, setpoint=1) #PID controller for TOP edge
|
|
self.PID_bot = PID(1, 0.1, 0.0, setpoint=1) #PID controller for Bottom edge
|
|
self.PID_on = False #if set to false, the PID thred (self.PID_thread) stops.
|
|
self.ramp = False #is set to True if temperature ramp is activated
|
|
self.rate = 0 #ramprate of cooldown
|
|
self.set_T = [0,0] #set Temperatures of PID controllers([top,bottom])
|
|
self.powers = [] #current power values of heaters will be stored here
|
|
self.Automation_Parameters_old = [0,0,0,0,0,0,0,0,0]
|
|
self.Automation_Parameters_new = [0,0,0,0,0,1,0,0,0]
|
|
|
|
self.labels_T = [self.lab_T_1,self.lab_T_2,self.lab_T_3,self.lab_T_4,self.lab_T_5,self.lab_T_6,self.lab_T_7,self.lab_T_8,self.lab_T_9,self.lab_T_10,self.lab_T_11,self.lab_T_12] #List of all Labels containing temperaure data. Is needed for "update_gui"
|
|
self.labels_B = [self.lab_Bx_is,self.lab_By_is,self.lab_Bz_is] #List of all Labels containing B data. Is needed for "update_gui"
|
|
self.SB_all = [self.dSB_timing_main,self.dSB_timing_monitoring,self.SB_plot_points,self.SB_Buffer_Size,self.dSB_P_max_top,self.dSB_P_max_bot,self.dSB_R_top,self.dSB_R_bot,
|
|
self.dSB_T_reset_top,self.dSB_T_reset_bot,self.dSB_T_c,self.SB_T_channel_top,self.SB_T_channel_bot,self.dSB_ramprate,self.dSB_ramprate_t_const,self.dSB_PID_P,self.dSB_PID_I,self.dSB_PID_D,
|
|
self.dSB_PID_P_2,self.dSB_PID_I_2,self.dSB_PID_D_2,self.dSB_Bx_set,self.dSB_By_set,self.dSB_Bz_set,self.dSB_Ix_set0,self.dSB_Iy_set0,self.dSB_Iz_set0,
|
|
self.dSB_Ix_set1,self.dSB_Iy_set1,self.dSB_Iz_set1,self.dSB_Ix_set2,self.dSB_Iy_set2,self.dSB_Iz_set2] #list of all Spinboxes, helps for saving and loading of default values
|
|
self.lines_all = [self.line_Path_points,self.line_Path_Mon,self.line_Path_AMR_calib] #List of all lineedits, helps for saving and loading default values
|
|
self.checkBoxes_all = [self.checkBox_monitoring, self.checkBox_sens_ana_1, self.checkBox_sens_ana_2, self.checkBox_sens_ana_3,
|
|
self.checkBox_sens_ana_4, self.checkBox_sens_ana_5, self.checkBox_sens_ana_6, self.checkBox_sens_ana_7, self.checkBox_sens_ana_8,
|
|
self.checkBox_autoscale_T, self.checkBox_autoscale_FG, self.checkBox_autoscale_T, self.checkBox_autoscale_AMR_x,
|
|
self.checkBox_autoscale_AMR_y, self.checkBox_autoscale_AMR_z, self.checkBox_I0x, self.checkBox_I0y, self.checkBox_I0z,
|
|
self.checkBox_I1x, self.checkBox_I1y, self.checkBox_I1z, self.checkBox_I2x, self.checkBox_I2y, self.checkBox_I2z]
|
|
|
|
#set up pyQT threadpool
|
|
self.threadpool = QThreadPool()
|
|
|
|
#read default values from config and set values in gui
|
|
self.load_default()
|
|
self.set_LS() #make LS 2018 green, since this is the default
|
|
|
|
#Print process id of this script for priority adjustments in task manager
|
|
print(f"Prozess-ID: {os.getpid()}")
|
|
|
|
#enable all plots and set background color of items to the corresponding plot color
|
|
self.list_AMR_x.selectAll()
|
|
self.list_AMR_y.selectAll()
|
|
self.list_AMR_z.selectAll()
|
|
self.list_T.selectAll()
|
|
self.list_FG.selectAll()
|
|
|
|
for list_widget in [self.list_T, self.list_FG, self.list_AMR_x, self.list_AMR_y, self.list_AMR_z]:
|
|
items = list_widget.selectedItems()
|
|
for item,col in zip(items,self.plot_colors):
|
|
item.setBackground(QColor(f"{col}"))
|
|
|
|
#call update PIDs to make shure the PID parameters correspond to the ones in the gui
|
|
self.update_PIDs
|
|
|
|
|
|
#start standard threads
|
|
worker = Worker(self.update_Data)
|
|
worker.signals.progress.connect(self.update_gui) #The values from update_Data must be transmitted via a signal, so that the gui is set in the main thread. If the plots are not updated in the main thread, they freeze.
|
|
self.threadpool.start(worker)
|
|
|
|
def update_Data(self, progress_callback):#gets data from global variables and triggers update_gui
|
|
scale_data_T = [[] for i in range(12)] #list of lists which will contain the data for scaling the T_plot
|
|
scale_data_FG = [[] for i in range(4)] #list of lists which will contain the data for scaling the FG-Plot
|
|
scale_data_AMR_x = [[] for i in range(15)] #list of lists which will contain the data for scaling the AMR_x plot
|
|
scale_data_AMR_y = [[] for i in range(15)] #list of lists which will contain the data for scaling the AMR_y plot
|
|
scale_data_AMR_z = [[] for i in range(15)] #list of lists which will contain the data for scaling the AMR_z plot
|
|
|
|
while self.running == True:
|
|
start = timer()
|
|
|
|
#get power data of heaters. This is not appended to Ringbuffer, since it is not plotter
|
|
self.powers = [self.sync_BK_9132B_2.get('P')[0],self.sync_BK_9132B_2.get('P')[1], self.sync_BK_9132B.get('P')[0]]
|
|
#sort data in T, FG, AMR
|
|
#get lakeshore data
|
|
LS336 = self.sync_LS_336.get('T')[0:4]
|
|
LS218 = self.sync_converted.get('T')
|
|
self.T = LS218 + LS336
|
|
|
|
#FG data
|
|
self.FG = [i*100 for i in self.sync_imc.get('FG')]
|
|
squares = [i**2 for i in self.FG]
|
|
mag = np.sqrt(sum(squares))
|
|
self.FG.append(mag)
|
|
|
|
#AMR data
|
|
self.AMR_x = self.sync_converted.get('AMR_x')
|
|
self.AMR_y = self.sync_converted.get('AMR_y')
|
|
self.AMR_z = self.sync_converted.get('AMR_z')
|
|
|
|
#Write Top / Bottom Temperatures into sync_main so that the Automation Program gets T_top and T_bottom temperatures
|
|
T_top = self.T[self.TB_channels[0]]
|
|
T_bot = self.T[self.TB_channels[1]]
|
|
self.sync_main.update({'T':[T_top, T_bot]})
|
|
|
|
#Check if automated measurement is running (i.e. True). If True: Check for changes in the automation parameters
|
|
if self.sync_main.get('Measurement_Running') is None:
|
|
self.Measurement_Running = False
|
|
else:
|
|
self.Measurement_Running = self.sync_main.get('Measurement_Running')
|
|
|
|
if self.Measurement_Running == True:
|
|
for i,n in enumerate(['Points_Path','T_set','T_reset','Cooldown_Speed','Start_ramp','Stop_ramp','0-Field','1-Field','Set_1field']): #get new parameters from global variables and compare to old ones.
|
|
self.Automation_Parameters_new[i] = self.sync_main.get(n)
|
|
|
|
if self.Automation_Parameters_new != self.Automation_Parameters_old: #if a parameter has been changed, check which parameter has been changed.
|
|
if self.Automation_Parameters_new[0] != self.Automation_Parameters_old[0]: #if Points_Path has been changed
|
|
self.line_Path_points.setText(self.Automation_Parameters_new[0]) # Set line_Path_points
|
|
|
|
if self.Automation_Parameters_new[1] != self.Automation_Parameters_old[1]: #if T_set has been changed
|
|
self.dSB_T_reset_top.setValue(float(self.Automation_Parameters_new[1][0])) #set T_reset_top
|
|
self.dSB_T_reset_bot.setValue(float(self.Automation_Parameters_new[1][1])) #set T_reset_bot
|
|
|
|
if self.Automation_Parameters_new[2] != self.Automation_Parameters_old[2]: #if T_reset has been changed
|
|
if self.Automation_Parameters_new[2] == True: #Check if T_reset is True --> Run same function as if T_reset button has been pressed
|
|
self.reset_set_T()
|
|
self.sync_main.update({'T_reset':False}) #Change T_reset to false
|
|
|
|
if self.Automation_Parameters_new[3] != self.Automation_Parameters_old[3]: #Cooldown_Speed has been changed
|
|
self.dSB_ramprate.setValue(float(self.Automation_Parameters_new[3])) #set ramprate
|
|
|
|
if self.Automation_Parameters_new[4] != self.Automation_Parameters_old[4]: #if Start_ramp has been changed
|
|
if self.Automation_Parameters_new[4] == True: #Check if Start_Ramp is True --> Run same function as if Start_Ramp button has been pressed
|
|
self.start_ramp()
|
|
self.sync_main.update({'Start_ramp':False}) #Change Start_Ramp to false
|
|
|
|
if self.Automation_Parameters_new[5] != self.Automation_Parameters_old[5]: #if Stop_ramp has been changed
|
|
if self.Automation_Parameters_new[5] == True: #Check if Stop_Ramp is True --> Run same function as if Stop_Ramp button has been pressed
|
|
self.stop_ramp_pushed()
|
|
self.sync_main.update({'Stop_ramp':False}) #Change Stop_Ramp to false
|
|
|
|
if self.Automation_Parameters_new[6] != self.Automation_Parameters_old[6]: #if 0-field parameters have been changed
|
|
I = self.Automation_Parameters_new[6][0] #[Ix, Iy, Iy2, Iz]
|
|
Flip = self.Automation_Parameters_new[6][1] #[Flipx, Flipy, Flipz]
|
|
self.dSB_Ix_set0.setValue(I[0])
|
|
self.dSB_Iy_set0.setValue(I[1])
|
|
self.dSB_Iy2_set0.setValue(I[2])
|
|
self.dSB_Iz_set0.setValue(I[3])
|
|
for box,state in zip([self.checkBox_I0x,self.checkBox_I0y,self.checkBox_I0z], Flip):
|
|
box.setChecked(state)
|
|
|
|
if self.Automation_Parameters_new[7] != self.Automation_Parameters_old[7]: #if 1-field parameters have been changed
|
|
I = self.Automation_Parameters_new[7][0] #[Ix, Iy, Iy2, Iz]
|
|
Flip = self.Automation_Parameters_new[7][1] #[Flipx, Flipy, Flipz]
|
|
self.dSB_Ix_set1.setValue(I[0])
|
|
self.dSB_Iy_set1.setValue(I[1])
|
|
self.dSB_Iy2_set1.setValue(I[2])
|
|
self.dSB_Iz_set1.setValue(I[3])
|
|
for box,state in zip([self.checkBox_I1x,self.checkBox_I1y,self.checkBox_I1z], Flip):
|
|
box.setChecked(state)
|
|
|
|
if self.Automation_Parameters_new[8] != self.Automation_Parameters_old[8]: #if Set_1field has been changed
|
|
if self.Automation_Parameters_new[8] == True: #Check if Set_1field is True --> Run same function as if Set_1field button has been pressed
|
|
self.set_I1()
|
|
self.sync_main.update({'Set_1field':False}) #Change Set_1field to false
|
|
|
|
|
|
|
|
#combine data and append Ringbuffer
|
|
self.new_Data = [time.time()] + self.T + self.FG + self.AMR_x + self.AMR_y + self.AMR_z
|
|
self.buffer.append(self.new_Data)
|
|
|
|
#Prepare data for plotting. This is done here, to keep the main thread as lean as possible for better performance (espacially due to high demands of plots)
|
|
#get subset of ringbuffers which is used for plotting
|
|
ind_plot = self.ind_plot #get a "local" copy of self.ind_plot because if Npoints is set to a value larger than len(buffer) in gui while the plot lists are collected the index might be out of range.
|
|
if self.Npoints >= len(self.buffer): #as long as the ringbuffers are too short, the indices of the plotted points must be adjusted to the length of the buffer. Otherwise an "index is out of range" will occur
|
|
self.downsample(len(self.buffer))
|
|
ind_plot = self.ind_plot
|
|
len_ind = len(ind_plot)
|
|
len_scale = len(self.buffer) #is needed to build "lists_scale"
|
|
else:
|
|
len_scale = self.Npoints
|
|
|
|
lists_plot = [self.buffer[i] for i in ind_plot] #get all lists in self.buffer that will be used for plotting
|
|
len_ind = len(lists_plot)
|
|
plot_data_t = [lists_plot[n][0] for n in range(len_ind)]#sort out times
|
|
for i in range(12):#sort out temperature data
|
|
self.plot_data_T[i] = [lists_plot[n][i+1] for n in range(len_ind)]
|
|
for i in range(4):#sort out Fluxgate data
|
|
self.plot_data_FG[i] = [lists_plot[n][i+13] for n in range(len_ind)]
|
|
for i in range(15):#sort out AMR_x data
|
|
self.plot_data_AMR_x[i] = [lists_plot[n][i+17] for n in range(len_ind)]
|
|
for i in range(15):#sort out AMR_y data
|
|
self.plot_data_AMR_y[i] = [lists_plot[n][i+32] for n in range(len_ind)]
|
|
for i in range(15):#sort out AMR_z data
|
|
self.plot_data_AMR_z[i] = [lists_plot[n][i+47] for n in range(len_ind)]
|
|
|
|
plot_data = [self.plot_data_T,self.plot_data_FG,self.plot_data_AMR_x,self.plot_data_AMR_y,self.plot_data_AMR_z,plot_data_t]
|
|
|
|
|
|
#get the start and stop time for the plots. Is used to scale x-axis
|
|
self.t0 = plot_data_t[0]
|
|
self.t1 = plot_data_t[-1]
|
|
#get min and max values of the plotted data to scale y axis. Get the data from the whole data set of self.Npoints. Otherwise the scales keep jumping when a peak is exluded due to downsampling
|
|
lists_scale = [self.buffer[-i] for i in range(1,len_scale+1)] #get all lists in self.buffer that will be used for scaling
|
|
for i in range(12):#sort out temperature data
|
|
scale_data_T[i] = [lists_scale[n][i+1] for n in range(len_scale)]
|
|
for i in range(4):#sort out Fluxgate data
|
|
scale_data_FG[i] = [lists_scale[n][i+13] for n in range(len_scale)]
|
|
for i in range(15):#sort out AMR_x data
|
|
scale_data_AMR_x[i] = [lists_scale[n][i+17] for n in range(len_scale)]
|
|
for i in range(15):#sort out AMR_y data
|
|
scale_data_AMR_y[i] = [lists_scale[n][i+32] for n in range(len_scale)]
|
|
for i in range(15):#sort out AMR_z data
|
|
scale_data_AMR_z[i] = [lists_scale[n][i+47] for n in range(len_scale)]
|
|
|
|
scale_data = [scale_data_T,scale_data_FG,scale_data_AMR_x,scale_data_AMR_y,scale_data_AMR_z]
|
|
for i,data in enumerate(scale_data):
|
|
active_data = [data[n] for n in self.show_plot_all[i]]
|
|
if active_data == []:
|
|
y_max = 0
|
|
y_min = 0
|
|
else:
|
|
y_max = max([max(d) for d in active_data])
|
|
y_min = min([min(d) for d in active_data])
|
|
self.yrange[i] = [y_min,y_max]
|
|
|
|
progress_callback.emit(plot_data) #emit signal, so that "update_gui" starts. "update_gui" uses the emitted lists as data. If data is transmitted via "global" variables (self....) is can happen that update_data "overtakes" update gui which leads to size differences in the plot data.
|
|
end = timer()
|
|
# print(end-start)
|
|
try:
|
|
time.sleep(self.timing_Main-(end-start))
|
|
except:
|
|
print(end-start)
|
|
|
|
def update_gui(self,plot_data):#sets labels to new values and updates plots
|
|
# start = timer()
|
|
#Update labels
|
|
for lab,val in zip(self.labels_T,self.T):
|
|
lab.setText(f"{val:4.3f}")
|
|
|
|
for lab,val in zip(self.labels_B,self.FG):
|
|
lab.setText(f"{val:.3f}")
|
|
|
|
self.lab_T_top.setText(f"{self.T[self.TB_channels[0]]:4.3f}")
|
|
self.lab_T_bot.setText(f"{self.T[self.TB_channels[1]]:4.3f}")
|
|
self.lab_T_set_top.setText(f"{self.set_T[0]:.3f}")
|
|
self.lab_T_set_bot.setText(f"{self.set_T[1]:.3f}")
|
|
self.lab_P_top.setText(f"{self.powers[0]:2.3f}")
|
|
self.lab_P_bot.setText(f"{self.powers[1]:2.3f}")
|
|
self.lab_He_heater_P_is.setText(f"{self.powers[2]:2.2f}")
|
|
|
|
#update plots
|
|
for plot,data in zip(self.plots_T,plot_data[0]):
|
|
plot.setData(plot_data[5],data)
|
|
if self.autoscale[0] == True:
|
|
self.graphWidget_T.setXRange(self.t0,self.t1)
|
|
self.graphWidget_T.setYRange(self.yrange[0][0],self.yrange[0][1])
|
|
|
|
for plot,data in zip(self.plots_fluxgate,plot_data[1]):
|
|
plot.setData(plot_data[5],data)
|
|
if self.autoscale[1] == True:
|
|
self.graphWidget_Fluxgate.setXRange(self.t0,self.t1)
|
|
self.graphWidget_Fluxgate.setYRange(self.yrange[1][0],self.yrange[1][1])
|
|
|
|
for plot,data in zip(self.plots_AMR_x,plot_data[2]):
|
|
plot.setData(plot_data[5],data)
|
|
if self.autoscale[2] == True:
|
|
self.graphWidget_AMR_x.setXRange(self.t0,self.t1)
|
|
self.graphWidget_AMR_x.setYRange(self.yrange[2][0],self.yrange[2][1])
|
|
|
|
for plot,data in zip(self.plots_AMR_y,plot_data[3]):
|
|
plot.setData(plot_data[5],data)
|
|
if self.autoscale[3] == True:
|
|
self.graphWidget_AMR_y.setXRange(self.t0,self.t1)
|
|
self.graphWidget_AMR_y.setYRange(self.yrange[3][0],self.yrange[3][1])
|
|
|
|
for plot,data in zip(self.plots_AMR_z,plot_data[4]):
|
|
plot.setData(plot_data[5],data)
|
|
if self.autoscale[4] == True:
|
|
self.graphWidget_AMR_z.setXRange(self.t0,self.t1)
|
|
self.graphWidget_AMR_z.setYRange(self.yrange[4][0],self.yrange[4][1])
|
|
|
|
# end = timer()
|
|
# print(end-start)
|
|
|
|
def PID_thread(self,progress_callback): #function is called in own thread in self.activate_PIDs. Keeps PIDs running in loop, as long self.PID_on == True
|
|
V = [0,0] #store values to send to the powersupplies
|
|
|
|
#bring power supplies in the correct state
|
|
update_single_entry(self.sync_BK_9132B_2,'OutputOn',0,True)
|
|
update_single_entry(self.sync_BK_9132B_2,'OutputOn',1,True)
|
|
update_single_entry(self.sync_BK_9132B_2,'setI',0,1.5)
|
|
update_single_entry(self.sync_BK_9132B_2,'setI',1,1.5)
|
|
|
|
while self.PID_on == True:
|
|
start = timer()
|
|
#lower setpoint if self.ramp == True
|
|
if self.ramp == True:
|
|
self.set_T = [i-self.rate/10 for i in self.set_T] #rate is devided by 10 because timing of PID loop is 0.1 s and rate is given in K/s.
|
|
|
|
self.PID_top.setpoint = self.set_T[0]
|
|
self.PID_bot.setpoint = self.set_T[1]
|
|
T_is = [self.T[i] for i in [self.SB_T_channel_top.value(),self.SB_T_channel_bot.value()]] #get current values
|
|
P = [self.PID_top(T_is[0]),self.PID_bot(T_is[1])]
|
|
i=0
|
|
for r,p in zip(self.R,P): #transform power values to voltage values and send them to global vars
|
|
update_single_entry(self.sync_BK_9132B_2,'setU',i,np.sqrt(p*r))
|
|
i=i+1
|
|
end = timer()
|
|
try:
|
|
time.sleep(0.1-(end-start)) #timing of PID loop is always 0.1 s
|
|
except:
|
|
print(f"PID loop took:{end-start} s")
|
|
|
|
#turn supplies off
|
|
update_single_entry(self.sync_BK_9132B_2,'OutputOn',0,False)
|
|
update_single_entry(self.sync_BK_9132B_2,'OutputOn',1,False)
|
|
|
|
|
|
def plot_List_T_changed(self):#is triggered when the plot selection for Temperature plot changes. Sets all entries in show_plot_T to false and then sets only the selected ones to true again. It then shows and hides the corresponding plots
|
|
show_plot_T = [False for i in range(12)]
|
|
indexes =self.list_T.selectedIndexes() #the return indeces are qt objects. To get a "normal" index the function row() must be used.
|
|
for ind in indexes: #set selected indeces to true
|
|
row = ind.row()
|
|
show_plot_T[row] = True
|
|
|
|
for show,p in zip(show_plot_T,self.plots_T): #show or hide the plots
|
|
if show == True:
|
|
p.show()
|
|
else:
|
|
p.hide()
|
|
self.show_plot_T = [i for i,x in enumerate(show_plot_T) if x == True] #create list with indexes of active plots. This is needed for autoscaling
|
|
self.show_plot_all = [self.show_plot_T,self.show_plot_FG,self.show_plot_AMR_x,self.show_plot_AMR_y,self.show_plot_AMR_z] #update list with new show_plot_T
|
|
|
|
def plot_List_FG_changed(self):#is triggered when the plot selection for Fluxgate plot changes. Sets all entries in show_plot_FG to false and then sets only the selected ones to true again. It then shows and hides the corresponding plots
|
|
show_plot_FG = [False for i in range(4)]
|
|
indexes =self.list_FG.selectedIndexes()
|
|
for ind in indexes:
|
|
row = ind.row()
|
|
show_plot_FG[row] = True
|
|
|
|
for show,p in zip(show_plot_FG,self.plots_fluxgate):
|
|
if show == True:
|
|
p.show()
|
|
else:
|
|
p.hide()
|
|
self.show_plot_FG = [i for i,x in enumerate(show_plot_FG) if x == True] #generate list with indixes of active plots
|
|
self.show_plot_all = [self.show_plot_T,self.show_plot_FG,self.show_plot_AMR_x,self.show_plot_AMR_y,self.show_plot_AMR_z] #update list with new show_plot_FG
|
|
|
|
def plot_List_AMR_x_changed(self):#is triggered when the plot selection for AMR_x plot changes. Sets all entries in show_plot_AMR_x to false and then sets only the selected ones to true again. It then shows and hides the corresponding plots
|
|
show_plot_AMR_x = [False for i in range(15)]
|
|
indexes =self.list_AMR_x.selectedIndexes()
|
|
for ind in indexes:
|
|
row = ind.row()
|
|
show_plot_AMR_x[row] = True
|
|
|
|
for show,p in zip(show_plot_AMR_x,self.plots_AMR_x):
|
|
if show == True:
|
|
p.show()
|
|
else:
|
|
p.hide()
|
|
self.show_plot_AMR_x = [i for i,x in enumerate(show_plot_AMR_x) if x == True]#generate list with indixes of active plots
|
|
self.show_plot_all = [self.show_plot_T,self.show_plot_FG,self.show_plot_AMR_x,self.show_plot_AMR_y,self.show_plot_AMR_z] #update list with new show_plot_AMR_x
|
|
|
|
def plot_List_AMR_y_changed(self):#is triggered when the plot selection for AMR_y plot changes. Sets all entries in show_plot_AMR_y to false and then sets only the selected ones to true again. It then shows and hides the corresponding plots
|
|
show_plot_AMR_y = [False for i in range(15)]
|
|
indexes =self.list_AMR_y.selectedIndexes()
|
|
for ind in indexes:
|
|
row = ind.row()
|
|
show_plot_AMR_y[row] = True
|
|
|
|
for show,p in zip(show_plot_AMR_y,self.plots_AMR_y):
|
|
if show == True:
|
|
p.show()
|
|
else:
|
|
p.hide()
|
|
self.show_plot_AMR_y = [i for i,x in enumerate(show_plot_AMR_y) if x == True]#generate list with indixes of active plots
|
|
self.show_plot_all = [self.show_plot_T,self.show_plot_FG,self.show_plot_AMR_x,self.show_plot_AMR_y,self.show_plot_AMR_z] #update list with new show_plot_AMR_y
|
|
|
|
def plot_List_AMR_z_changed(self):#is triggered when the plot selection for AMR_z plot changes. Sets all entries in show_plot_AMR_x to false and then sets only the selected ones to true again. It then shows and hides the corresponding plots
|
|
show_plot_AMR_z = [False for i in range(15)]
|
|
indexes =self.list_AMR_z.selectedIndexes()
|
|
for ind in indexes:
|
|
row = ind.row()
|
|
show_plot_AMR_z[row] = True
|
|
|
|
for show,p in zip(show_plot_AMR_z,self.plots_AMR_z):
|
|
if show == True:
|
|
p.show()
|
|
else:
|
|
p.hide()
|
|
self.show_plot_AMR_z = [i for i,x in enumerate(show_plot_AMR_z) if x == True]#generate list with indixes of active plots
|
|
self.show_plot_all = [self.show_plot_T,self.show_plot_FG,self.show_plot_AMR_x,self.show_plot_AMR_y,self.show_plot_AMR_z] #update list with new show_plot_AMR_z
|
|
|
|
def downsample(self,N): #downsamples N down to 200. It creates list with 200 entries equally spaced from -1 to -N. self.ind_plot is set to this list and is subsequently used to downsample the plot data to 200 points.
|
|
if N <= 200: #list of indices to start at the end of the ringbuffer up N
|
|
self.ind_plot = [-i for i in range(1,N+1)]
|
|
else: #list of indices which downsamples to 200. Indices are again negative, because it starts at the end of the ringbuffer
|
|
q = N/(N-200+1) #every q-th index needs to be deleted
|
|
|
|
ind = [i for i in range(1,N+1)] #list containing all indices
|
|
|
|
ind_pop = []#empty list. Is filled with indices that will be excluded in the for-loop
|
|
for n in range(1,N-200+1): #fill list with indiced
|
|
ind_pop.append(round(n*q))
|
|
self.ind_plot = [-x for i, x in enumerate(ind) if i not in ind_pop] #exclude indices in "ind_pop" from "ind"
|
|
|
|
def set_Npoints(self,N): #is triggered when number of points to plot is changed. Sets variable self.Npoints to the new value. If Npoints is larger than 200 self.downsample() is called
|
|
self.Npoints = N
|
|
self.downsample(N)
|
|
|
|
def disable_plots(self): #is triggered when disable all plots checkbox is changed. Hides, or shows all plots
|
|
for plots in self.plot_sets:
|
|
for plot in plots:
|
|
if self.checkBox_plots_disable.isChecked() == True:
|
|
plot.hide()
|
|
else:
|
|
plot.show()
|
|
|
|
def set_autoscale(self): #is triggered when any of the autoscale checkboxes are changed. Sets self.autoscale to the new value
|
|
checkboxes = [self.checkBox_autoscale_T,self.checkBox_autoscale_FG,self.checkBox_autoscale_AMR_x,self.checkBox_autoscale_AMR_y,self.checkBox_autoscale_AMR_z]
|
|
for i,box in enumerate(checkboxes):
|
|
self.autoscale[i] = box.isChecked()
|
|
|
|
def set_timing_Main(self,t): #is triggered when timing Main is changed. Sets variable self.timin_Main to the new value
|
|
self.timing_Main = t
|
|
|
|
def change_buffer_size(self): #is triggered when self.button_change_Buffer is clicked. Since maxlen cannot be changed after initialization it initializes new buffers with the correct maxlen.
|
|
new_len = self.SB_Buffer_Size.value()
|
|
for i,buffer in enumerate(self.buffers_T): #replace old buffer with new one with new length. In order to keep the newest data points the data is stored in a list, and the newest points are used for the new buffer
|
|
data = list(buffer)
|
|
self.buffers_T[i] = collections.deque(data[-new_len:], maxlen = new_len)
|
|
|
|
for i,buffer in enumerate(self.buffers_FG): #replace old buffer with new one with new length. In order to keep the newest data points the data is stored in a list, and the newest points are used for the new buffer
|
|
data = list(buffer)
|
|
self.buffers_FG[i] = collections.deque(data[-new_len:], maxlen = new_len)
|
|
|
|
for i,buffer in enumerate(self.buffers_AMR_x): #replace old buffer with new one with new length. In order to keep the newest data points the data is stored in a list, and the newest points are used for the new buffer
|
|
data = list(buffer)
|
|
self.buffers_AMR_x[i] = collections.deque(data[-new_len:], maxlen = new_len)
|
|
|
|
for i,buffer in enumerate(self.buffers_AMR_y): #replace old buffer with new one with new length. In order to keep the newest data points the data is stored in a list, and the newest points are used for the new buffer
|
|
data = list(buffer)
|
|
self.buffers_AMR_y[i] = collections.deque(data[-new_len:], maxlen = new_len)
|
|
|
|
for i,buffer in enumerate(self.buffers_AMR_z): #replace old buffer with new one with new length. In order to keep the newest data points the data is stored in a list, and the newest points are used for the new buffer
|
|
data = list(buffer)
|
|
self.buffers_AMR_z[i] = collections.deque(data[-new_len:], maxlen = new_len)
|
|
|
|
#replace time buffer
|
|
data = list(self.buffer_t)
|
|
self.buffer_t = collections.deque(data[-new_len:], maxlen = new_len)
|
|
|
|
def set_monitoring(self): #is triggered when monitoring checkbox is clicked. If the new value is True, a new thread is started in which self.monitoring is running
|
|
if self.checkBox_monitoring.isChecked() == True:
|
|
self.monitor = True
|
|
self.worker_mon = Worker(self.monitoring)
|
|
self.threadpool.start(self.worker_mon)
|
|
else:
|
|
self.monitor = False
|
|
|
|
def set_timing_monitoring(self,t): #is triggered when monitoring timing is changed. Sets self.timing_monitoring to new value.
|
|
self.timing_monitoring = t
|
|
|
|
def set_imc(self): #is called when button "imc" is pressed. Sets self.T_imc = True and sets colors of buttons accordingly
|
|
self.T_imc = True
|
|
self.syncdict.update({"T_imc":True})
|
|
self.Button_imc.setStyleSheet("background-color: green; color: white;")
|
|
self.Button_LS.setStyleSheet("background-color: '#e1e1e1'; color: black;")
|
|
|
|
def set_LS(self): #is called when button "LS 2018" is pressed. Sets self.T_imc = False and sets colors of buttons accordingly
|
|
self.T_imc = False
|
|
self.syncdict.update({"T_imc":False})
|
|
self.Button_imc.setStyleSheet("background-color: '#e1e1e1'; color: black;")
|
|
self.Button_LS.setStyleSheet("background-color: green; color: white;")
|
|
|
|
def set_TB_channel(self): #is called when either channel of Top or Bottom cernox is changed. Updates the varable self.TB_channels according to the values
|
|
self.TB_channels = [self.SB_T_channel_top.value(), self.SB_T_channel_bot.value()]
|
|
|
|
def activate_PID(self): #is called when "button_PID_on" is clicked. Changes button text and sets self.PID_on to True or False. If it changed to True, is starts a thread with the PID controllers
|
|
if self.Button_PID_on.text() == 'on/OFF': #PID was OFF and is now turned ON
|
|
self.Button_PID_on.setText('ON/off')
|
|
self.Button_PID_on.setStyleSheet("background-color: green")
|
|
self.PID_on = True
|
|
#start thread
|
|
worker_PID = Worker(self.PID_thread)
|
|
self.threadpool.start(worker_PID)
|
|
else: #PID was On and is now turned off
|
|
self.Button_PID_on.setText('on/OFF')
|
|
self.Button_PID_on.setStyleSheet("background-color: red")
|
|
self.PID_on = False
|
|
|
|
def update_PIDs(self): #is called when any PID parameter, except setvalue, is changed. Updates the PID parameters
|
|
self.PID_top.tunings = (self.dSB_PID_P.value(),self.dSB_PID_I.value(),self.dSB_PID_D.value())
|
|
self.PID_bot.tunings = (self.dSB_PID_P_2.value(),self.dSB_PID_I.value(),self.dSB_PID_D.value())
|
|
self.PID_top.output_limits = (0,self.dSB_P_max_top.value())
|
|
self.PID_bot.output_limits = (0,self.dSB_P_max_bot.value())
|
|
self.R = [self.dSB_R_top.value(),self.dSB_R_bot.value()]
|
|
|
|
def reset_set_T(self): #is called when button "reset" is pushed. Sets self.set_T to the values in the reset double spin boxes. The values of self.set_T are then set in the labels "set" labes by self.update_gui
|
|
self.set_T = [self.dSB_T_reset_top.value(),self.dSB_T_reset_bot.value()]
|
|
|
|
def start_ramp(self): #is called when "start ramp" button is pushed. #If transition time spin box is not equal 0, it calculates the corresponding cooldown rate. Otherwise the cooldown rate is the correspoing dSB is used. Also sets self.ramp = True so self.set_T is lowered in self.PID_thread. Also stores time when the button is pushed. This will be used when save point button is pushed.
|
|
if self.dSB_ramprate_t_const.value() != 0:
|
|
self.rate = (self.set_T[0] - self.set_T[1])/self.dSB_ramprate_t_const.value()
|
|
else:
|
|
self.rate = self.dSB_ramprate.value()
|
|
self.ramp = True
|
|
self.start_time_ramp = self.new_Data[0] #get time stamp from current data point
|
|
self.B_start = self.FG
|
|
self.lab_ramp_status.setText("started")
|
|
|
|
def stop_ramp_pushed(self): #is called when stop ramp button is pushed. Since the programm needs to wait a few seconds in this procesure it is done in a seperate thread. This function only start this thread
|
|
worker_stop_ramp = Worker(self.stop_ramp)
|
|
self.threadpool.start(worker_stop_ramp)
|
|
self.lab_ramp_status.setText("stopped")
|
|
|
|
def stop_ramp(self,progress_callback): #is started by self.stop_ramp_pushed. Turnes off heaters, takes data when
|
|
self.ramp = 0 #stop ramping
|
|
self.set_T = [0,0] #setpoint to 0 so heaters are turned off
|
|
self.stop_time_ramp = self.new_Data[0] #get time stamp from current data point
|
|
time.sleep(3)
|
|
self.B_expelled = self.AMR_x + self.AMR_y + self.AMR_z #get data when B-Field is still active
|
|
time.sleep(0.5)
|
|
self.set_I0() #set 0-Field
|
|
self.lab_ramp_status.setText("0-field")
|
|
time.sleep(4)
|
|
self.save_point()
|
|
|
|
|
|
def save_point(self): #is called at the end of stop_ramp. Start a new thread with self.analyse_cooldown
|
|
self.save_time = self.new_Data[0] #get time stamp from current data point
|
|
self.B_trapped = self.AMR_x + self.AMR_y + self.AMR_z
|
|
worker_analyse = Worker(self.analyse_cooldown)
|
|
self.threadpool.start(worker_analyse)
|
|
self.lab_ramp_status.setText("finished")
|
|
|
|
|
|
def analyse_cooldown(self,progress_callback): #analyses the cooldown and writes it to file
|
|
print("start_analyse")
|
|
#find indices of start and stop time of ramp
|
|
for i in range(len(self.buffer)): #iterate through buffer from the back. Check the first item in the list to see if it equals the start or stop time
|
|
t = self.buffer[-i][0]
|
|
if t == self.start_time_ramp:
|
|
I_start = -i
|
|
break #stop time is found first, because iteration is done from back to front. After I_start is found the loop can be broken.
|
|
elif t == self.stop_time_ramp:
|
|
I_stop = -i
|
|
|
|
#prepare cernox and time data
|
|
print(self.start_time_ramp)
|
|
print(time.strftime("%Y-%m-%d_%H-%M-%S",time.gmtime(self.start_time_ramp))+"\t")
|
|
# print(time.strftime("%Y-%m-%d_%H-%M-%S",time.gmtime(self.stop_time_ramp))+"\t")
|
|
print(time.strftime("%Y-%m-%d_%H-%M-%S",time.gmtime(self.save_time))+"\t")
|
|
print([I_start,I_stop])
|
|
lists_ramp = [self.buffer[i] for i in range(I_start,I_stop)] #get all lists in self.buffer that are recorded during ramping
|
|
print(len(lists_ramp))
|
|
len_ind = len(lists_ramp)
|
|
print(lists_ramp[0][0])
|
|
time_data = [lists_ramp[n][0] for n in range(len_ind)]#sort out times
|
|
temp_data = [i for i in range(8)] #create "empty" list in which the cernox data will be stored
|
|
for i in range(8):#sort out temperature data
|
|
temp_data[i] = [lists_ramp[n][i+1] for n in range(len_ind)]
|
|
|
|
#preapre list of active cernox sensors for analysis
|
|
act_sens = [] #list of active sensors
|
|
for i,c in enumerate([self.checkBox_sens_ana_1, self.checkBox_sens_ana_2, self.checkBox_sens_ana_3, self.checkBox_sens_ana_4, self.checkBox_sens_ana_5, self.checkBox_sens_ana_6,
|
|
self.checkBox_sens_ana_7, self.checkBox_sens_ana_8]):
|
|
if c.isChecked() == True:
|
|
act_sens.append(i+1)
|
|
#analyse data
|
|
try:
|
|
results = analyse_cooldown_tools.calc_gradient(temp_data,time_data,act_sens,self.dSB_T_c.value())
|
|
print(f'T_c = {self.dSB_T_c.value()}')
|
|
except:
|
|
print('could not analyse data')
|
|
return
|
|
|
|
#open file
|
|
path = self.line_Path_points.text()
|
|
try: #if no valid path is given a file "points.txt" is created in the folder
|
|
if os.path.isfile(path) == False:
|
|
with open(path,'a') as file:
|
|
file.write('t_start\tt_stop\tt_save\tB_start x\tB_start y\tB_start z\tB_start mag\tgrad_glob\tav_grad_glob\te_av_grad_glob\ttrans_time\tav_rate_glob\te_av_rate_glob\tgrad_loc\trate_loc\t15*AMR_x_expelled\t15*AMR_y_expelled\t15*AMR_z_expelled\t45*AMR_trapped\n')
|
|
file = open(path,'a')
|
|
except:
|
|
path = self.current_dir+'\\points.txt'
|
|
if os.path.isfile(path) == False:
|
|
with open(path,'a') as file:
|
|
file.write('t_start\tt_stop\tt_save\tB_start x\tB_start y\tB_start x\tgrad_glob\tav_grad_glob\te_av_grad_glob\ttrans_time\tav_rate_glob\te_av_rate_glob\tgrad_loc\trate_loc\t15*AMR_x_expelled\t15*AMR_y_expelled\t15*AMR_z_expelled\t45*AMR_trapped\n')
|
|
file = open(path,'a')
|
|
|
|
#write data to file
|
|
file.write(time.strftime("%Y-%m-%d_%H-%M-%S",time.localtime(self.start_time_ramp))+"\t")
|
|
file.write(time.strftime("%Y-%m-%d_%H-%M-%S",time.localtime(self.stop_time_ramp))+"\t")
|
|
file.write(time.strftime("%Y-%m-%d_%H-%M-%S",time.localtime(self.save_time))+"\t")
|
|
#write contents of all the lists
|
|
for L in [self.B_start,results,self.B_expelled,self.B_trapped]:
|
|
for i in L:
|
|
try:
|
|
file.write(f"{i}\t")
|
|
except:
|
|
file.write("No Value\t")
|
|
file.write("\n")#new line
|
|
file.close()
|
|
|
|
|
|
def set_I0(self):#sets the current values and relais status of 0-Field
|
|
#turn on outputs
|
|
self.sync_BK_9174B.update({'OutputOn':[True,True]})
|
|
self.sync_BK_9174B_2.update({'OutputOn':[True,True]})
|
|
self.sync_BK_9174B.update({'setU':[70,70]})
|
|
self.sync_BK_9174B_2.update({'setU':[70,70]})
|
|
#set currents
|
|
self.sync_BK_9174B.update({'setI':[self.dSB_Ix_set0.value(),self.dSB_Iy_set0.value()]})
|
|
self.sync_BK_9174B_2.update({'setI':[self.dSB_Iz_set0.value(),self.dSB_Iy2_set0.value()]})
|
|
#set relais
|
|
self.sync_BK_9131B.update({'SetU':[5,5,5]})
|
|
self.sync_BK_9131B.update({'SetI':[1,1,1]})
|
|
self.sync_BK_9131B.update({'OutputOn':[self.checkBox_I0x.isChecked(),self.checkBox_I0y.isChecked(),self.checkBox_I0z.isChecked()]})
|
|
|
|
def set_I1(self):#sets the current values and relais status of Field 1
|
|
#turn on outputs
|
|
self.sync_BK_9174B.update({'OutputOn':[True,True]})
|
|
self.sync_BK_9174B_2.update({'OutputOn':[True,True]})
|
|
self.sync_BK_9174B.update({'setU':[70,70]})
|
|
self.sync_BK_9174B_2.update({'setU':[70,70]})
|
|
#set currents
|
|
self.sync_BK_9174B.update({'setI':[self.dSB_Ix_set1.value(),self.dSB_Iy_set1.value()]})
|
|
self.sync_BK_9174B_2.update({'setI':[self.dSB_Iz_set1.value(),self.dSB_Iy2_set1.value()]})
|
|
#set relais
|
|
self.sync_BK_9131B.update({'SetU':[5,5,5]})
|
|
self.sync_BK_9131B.update({'SetI':[1,1,1]})
|
|
self.sync_BK_9131B.update({'OutputOn':[self.checkBox_I1x.isChecked(),self.checkBox_I1y.isChecked(),self.checkBox_I1z.isChecked()]})
|
|
|
|
def set_I2(self):#sets the current values and relais status of Field 2
|
|
#turn on outputs
|
|
self.sync_BK_9174B.update({'OutputOn':[True,True]})
|
|
self.sync_BK_9174B_2.update({'OutputOn':[True,True]})
|
|
self.sync_BK_9174B.update({'setU':[70,70]})
|
|
self.sync_BK_9174B_2.update({'setU':[70,70]})
|
|
#set currents
|
|
self.sync_BK_9174B.update({'setI':[self.dSB_Ix_set2.value(),self.dSB_Iy_set2.value()]})
|
|
self.sync_BK_9174B_2.update({'setI':[self.dSB_Iz_set2.value(),self.dSB_Iy2_set2.value()]})
|
|
#set relais
|
|
self.sync_BK_9131B.update({'SetU':[5,5,5]})
|
|
self.sync_BK_9131B.update({'SetI':[1,1,1]})
|
|
self.sync_BK_9131B.update({'OutputOn':[self.checkBox_I2x.isChecked(),self.checkBox_I2y.isChecked(),self.checkBox_I2z.isChecked()]})
|
|
|
|
def get_I0(self): #gets momentary current values and relais status and saves them in 0-Field boxes
|
|
self.dSB_Ix_set0.setValue(self.sync_BK_9174B.get('setI')[0])
|
|
self.dSB_Iy_set0.setValue(self.sync_BK_9174B.get('setI')[1])
|
|
self.dSB_Iy2_set0.setValue(self.sync_BK_9174B_2.get('setI')[1])
|
|
self.dSB_Iz_set0.setValue(self.sync_BK_9174B_2.get('setI')[0])
|
|
|
|
|
|
for box,state in zip([self.checkBox_I0x,self.checkBox_I0y,self.checkBox_I0z], self.sync_BK_9131B.get('OutputOn')):
|
|
box.setChecked(state)
|
|
|
|
def get_I1(self): #gets momentary current values and relais status and saves them in Field 1 boxes
|
|
self.dSB_Ix_set1.setValue(self.sync_BK_9174B.get('setI')[0])
|
|
self.dSB_Iy_set1.setValue(self.sync_BK_9174B.get('setI')[1])
|
|
self.dSB_Iy2_set1.setValue(self.sync_BK_9174B_2.get('setI')[1])
|
|
self.dSB_Iz_set1.setValue(self.sync_BK_9174B_2.get('setI')[0])
|
|
|
|
for box,state in zip([self.checkBox_I1x,self.checkBox_I1y,self.checkBox_I1z], self.sync_BK_9131B.get('OutputOn')):
|
|
box.setChecked(state)
|
|
|
|
def get_I2(self): #gets momentary current values and relais status and saves them in Field 2 boxes
|
|
self.dSB_Ix_set2.setValue(self.sync_BK_9174B.get('setI')[0])
|
|
self.dSB_Iy_set2.setValue(self.sync_BK_9174B.get('setI')[1])
|
|
self.dSB_Iy2_set2.setValue(self.sync_BK_9174B_2.get('setI')[1])
|
|
self.dSB_Iz_set2.setValue(self.sync_BK_9174B_2.get('setI')[0])
|
|
|
|
for box,state in zip([self.checkBox_I2x,self.checkBox_I2y,self.checkBox_I2z], self.sync_BK_9131B.get('OutputOn')):
|
|
box.setChecked(state)
|
|
|
|
def turn_off_coils(self): #turns off all coils
|
|
#turn on outputs
|
|
self.sync_BK_9174B.update({'OutputOn':[False,False]})
|
|
self.sync_BK_9174B_2.update({'OutputOn':[False,False]})
|
|
#set relais
|
|
self.sync_BK_9131B.update({'OutputOn':[False,False,False]})
|
|
|
|
def set_B(self): #set global variables B_set and Start_Compensation to start magnetic field compensation
|
|
B_set = [self.dSB_Bx_set.value(), self.dSB_By_set.value(), self.dSB_Bz_set.value()]
|
|
self.syncdict.update({"B_set":B_set})
|
|
self.syncdict.update({"Start_Compensation":True})
|
|
|
|
def set_LED(self):
|
|
'''turns LED on or off'''
|
|
if self.checkBox_LED.isChecked():
|
|
update_single_entry(self.sync_BK_9132B_2,'setU',2,5)
|
|
update_single_entry(self.sync_BK_9132B_2,'setI',2,1)
|
|
update_single_entry(self.sync_BK_9132B_2,'OutputOn',2,True)
|
|
else:
|
|
update_single_entry(self.sync_BK_9132B_2,'OutputOn',2,False)
|
|
|
|
def switch_fan(self):
|
|
'''turn relais fan on or off'''
|
|
if self.checkBox_Relais_Fan.isChecked():
|
|
update_single_entry(self.sync_BK_9132B,'setU',1,9)
|
|
update_single_entry(self.sync_BK_9132B,'setI',1,1)
|
|
update_single_entry(self.sync_BK_9132B,'OutputOn',1,True)
|
|
else:
|
|
update_single_entry(self.sync_BK_9132B,'OutputOn',1,False)
|
|
|
|
def switch_AMR_supply(self):
|
|
'''turn AMR supply voltage of 9 V on or off'''
|
|
if self.checkBox_AMR_supply.isChecked():
|
|
update_single_entry(self.sync_Keithley_2230G,'setU',0,9)
|
|
update_single_entry(self.sync_Keithley_2230G,'setI',0,1)
|
|
update_single_entry(self.sync_Keithley_2230G,'OutputOn',0,True)
|
|
else:
|
|
update_single_entry(self.sync_Keithley_2230G,'OutputOn',0,False)
|
|
|
|
def AMR_flip(self):
|
|
'''turn AMR flip voltage on or off'''
|
|
if self.checkBox_AMR_flip.isChecked():
|
|
update_single_entry(self.sync_Keithley_2230G,'setU',1,20)
|
|
update_single_entry(self.sync_Keithley_2230G,'setI',1,0.2)
|
|
update_single_entry(self.sync_Keithley_2230G,'OutputOn',1,True)
|
|
else:
|
|
update_single_entry(self.sync_Keithley_2230G,'OutputOn',1,False)
|
|
|
|
def switch_he_heater(self):
|
|
'''turn he heater on or with the desired power. The corresponding voltage is calculated only once when the heater is turned on or the power is changed'''
|
|
if self.checkBox_He_heater.isChecked():
|
|
#calculate voltage
|
|
U = np.sqrt(self.dSB_He_heater_P_set.value()*self.dSB_He_heater_R.value())
|
|
update_single_entry(self.sync_BK_9132B,'setU',0,U)
|
|
update_single_entry(self.sync_BK_9132B,'setI',0,2)
|
|
update_single_entry(self.sync_BK_9132B,'OutputOn',0,True)
|
|
else:
|
|
update_single_entry(self.sync_BK_9132B,'OutputOn',0,False)
|
|
|
|
def measure_R_pressed(self):
|
|
'''starts thread with self.measure_R. Is triggered by button "measure R"'''
|
|
worker_measure_R = Worker(self.measure_R)
|
|
self.threadpool.start(worker_measure_R)
|
|
|
|
def measure_R(self):
|
|
'''sets a small voltage across the he heater to measure the resistance and sets in gui, turn off the heater after measurement'''
|
|
update_single_entry(self.sync_BK_9132B,'setU',0,1)
|
|
update_single_entry(self.sync_BK_9132B,'setI',0,0.2)
|
|
update_single_entry(self.sync_BK_9132B,'OutputOn',0,True)
|
|
time.sleep(3)
|
|
U = self.sync_BK_9132B.get('U')[0]
|
|
I = self.sync_BK_9132B.get('I')[0]
|
|
R = U/I
|
|
update_single_entry(self.sync_BK_9132B,'OutputOn',0,False)
|
|
self.dSB_He_heater_R.setValue(R)
|
|
|
|
def monitoring(self, progress_callback):
|
|
#if save checkbox is checked it writes measurement values to file specified in line.filePath. There the full path including file extension must be given.
|
|
while self.monitor == True:
|
|
time.sleep(self.timing_monitoring) #wait is at beginning so first point is not corrupted when app just started.
|
|
path = self.line_Path_Mon.text()
|
|
if os.path.isfile(path) == False:
|
|
with open(path,'a') as file:
|
|
file.write('date\tLS218 CH1[K]\tLS218 CH2[K]\tLS218 CH3[K]\tLS218 CH4[K]\tLS218 CH5[K]\tLS218 CH6[K]\tLS218 CH7[K]\tLS218 CH8[K]\tLS336 CHA[K]\tLS336 CHB[K]\tLS336 CHC[K]\tLS336 CHD[K]\n')
|
|
file = open(path,'a')
|
|
curr_time = datetime.now()
|
|
file.write(curr_time.strftime("%Y-%m-%d_%H-%M-%S-%f")+'\t')
|
|
for d in self.T:
|
|
file.write(f"{d}\t")
|
|
file.write('\n')
|
|
file.close
|
|
|
|
def save_default(self):
|
|
#saves current set values to txt file in subdirectory configs. Saves values from al spin boxes and text lines.
|
|
#Overwrites old values in config file.
|
|
path = self.current_dir+'\\configs\\Main_config.txt' #To make shure the config file is at the right place, independent from where the program is started the location of the file is retrieved
|
|
file = open(path,'w')
|
|
for SB in self.SB_all:
|
|
temp = f"{SB.value()}"
|
|
file.write(temp+'\t')
|
|
for l in self.lines_all:
|
|
file.write(l.text()+'\t')
|
|
for c in self.checkBoxes_all:
|
|
file.write(str(c.isChecked())+'\t')
|
|
|
|
file.write('\n')
|
|
file.close
|
|
|
|
def load_default(self):
|
|
#reads default values from config file in subdirectory config and sets the values in gui. (If no config file exists, it does nothing.)
|
|
path = self.current_dir+'\\configs\\Main_config.txt' #To make shure the config file is read from the right place, independent from where the program is started the location of the file is retrieved
|
|
try: #exit function if config file does not exist
|
|
vals = import_txt.read_raw(path)
|
|
except:
|
|
return
|
|
|
|
|
|
for SB,v in zip(self.SB_all,vals[0]):
|
|
if type(SB) == QDoubleSpinBox:
|
|
v = float(v) #convert string in txt to float, so number can be set in dSB
|
|
else:
|
|
v = int(v) #convert string in txt to int, so number can be set in SB
|
|
SB.setValue(v)
|
|
|
|
for l,v in zip(self.lines_all,vals[0][len(self.SB_all):]):
|
|
l.setText(v)
|
|
|
|
for c,v in zip(self.checkBoxes_all,vals[0][len(self.SB_all)+len(self.lines_all):]):
|
|
c.setChecked(v == 'True')
|
|
|
|
def closeEvent(self,event): #when window is closed self.running and self.monitor are set to False, so all threads stop
|
|
self.running = False
|
|
self.monitor = False
|
|
self.PID_on = False
|
|
time.sleep(1)
|
|
event.accept()
|
|
|
|
app = QApplication(sys.argv)
|
|
|
|
window = MainWindow()
|
|
window.show()
|
|
app.exec() |