Skip to content
Snippets Groups Projects
Commit 0279bbc0 authored by Martin Killenberg's avatar Martin Killenberg
Browse files

Refactor measurement loops 

- Outer loop for sweep file
- Inner loop for measurement (can be called as a separate function)
- Each measurements has its own csv file
parent 3af53a63
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
Python_script/prototype.py
+ 145
− 132
View file @ 0279bbc0
......@@ -35,7 +35,7 @@ class MeasurementData:
class Measurements:
def __init__(self, chamber_address, vna_address, sweep_file, output_file, standby, config_data):
def __init__(self, chamber_address, vna_address, sweep_file, output_basename, standby, config_data):
self.max_delta_temp = config_data['delta_temp']
self.max_delta_hum = config_data['delta_hum']
self.max_delta_mag = config_data['delta_mag']
......@@ -48,7 +48,7 @@ class Measurements:
self.vna = VNA.create_vna(vna_address, target_accuracy)
self.sweep_file = sweep_file
self.standby = standby
self.output = output_file
self.output_basename = output_basename
self.clock = virtual_time.get_clock(chamber_address, target_accuracy)
self.temperature_stable = False
self.humidity_stable = False
......@@ -65,7 +65,33 @@ class Measurements:
self.data_collection = []
def perform_measurements(self):
with open(self.output, mode='w', newline='') as csv_file:
with open(self.sweep_file) as file:
try:
measurement_number = 0
while line := file.readline().rstrip():
list_of_values = line.split()
next_temp = float(list_of_values[0]) # target_temperature
next_hum = float(list_of_values[1]) # target_humidity
next_soaking = int(list_of_values[2]) # soaking_time
next_reads = int(list_of_values[3]) # number of stable readings
self.perform_single_measurement(self.output_basename+'_'+str(measurement_number)+'.csv', next_temp, next_hum,
next_soaking, next_reads)
measurement_number += 1
except KeyboardInterrupt:
pass
plt.close()
return measurement_number
if self.standby:
standby_response = self.chamber.set_mode('STANDBY')
print(standby_response)
def perform_single_measurement(self, output, target_temp, target_hum, soaking_time, n_stable_reads):
with open(output, mode='w', newline='') as csv_file:
fieldnames = ['TIMESTAMP', 'TARGET_TEMPERATURE', 'READBACK_TEMPERATURE', 'TARGET_HUMIDITY',
'READBACK_HUMIDITY', 'RF_POWER', 'RF_FREQUENCY', 'DUT_IDENTIFIER', 'RUN_ID',
'EQUILIBRIUM_INDICATOR', 'S11_MAGNITUDE', 'S11_PHASE', 'S12_MAGNITUDE', 'S12_PHASE',
......@@ -76,130 +102,112 @@ class Measurements:
self.data_collection = []
plt.ion()
with open(self.sweep_file) as file:
try:
while line := file.readline().rstrip():
list_of_values = line.split()
next_temp = float(list_of_values[0]) # target_temperature
next_hum = float(list_of_values[1]) # target_humidity
next_soaking = int(list_of_values[2]) # soaking_time
next_reads = int(list_of_values[3]) # number of stable readings
set_const_response = self.chamber.set_const((next_temp, next_hum))
print(set_const_response)
set_mode_response = self.chamber.set_mode('CONSTANT')
print(set_mode_response)
number_of_soaking_reads = next_soaking / self.sleep_time + 1
do_another_measurement = True
#next_read_time is the starttime of the second read (i.e. read after this one). The time of
#this read (i.e. the first read in the measurement) is now().
next_read_time = self.clock.time() + self.sleep_time
while do_another_measurement:
# wait until set point is reached (+soaking time)
magnitudes_queue = []
phase_queue = []
while True:
data = self.read_data()
# if it is not within the target range reset start time
self.temperature_stable = False
self.humidity_stable = False
self.temperature_stable = self.calculate_temperature_stability(next_temp, float(data.temp))
self.humidity_stable = self.calculate_humidity_stability(next_hum, float(data.hum))
# The queue must not be longer than the max number of soaking reads.
# If the queue is already full, we have to pop the first element before we can add the
# current measurement.
if len(magnitudes_queue) >= number_of_soaking_reads:
magnitudes_queue.pop(0)
if self.temperature_stable and self.humidity_stable:
magnitudes_queue.append(self.calculate_mean_magnitude(data.s21))
else:
magnitudes_queue.clear()
# check cable stability parameters
self.magnitude_stable = False
if len(magnitudes_queue) >= number_of_soaking_reads:
spread = max(magnitudes_queue) - min(magnitudes_queue)
if spread < 2*self.max_delta_mag:
self.magnitude_stable = True
if len(phase_queue) >= number_of_soaking_reads:
phase_queue.pop(0)
if self.temperature_stable and self.humidity_stable:
phase_queue.append(self.calculate_mean_phase(data.s21))
else:
phase_queue.clear()
self.phase_stable = False
if len(phase_queue) >= number_of_soaking_reads:
spread = max(phase_queue) - min(phase_queue)
if spread < 2*self.max_delta_phase:
self.phase_stable = True
print('Setpoint: ' + str(next_temp) + ' ' + str(next_hum) + ' | Temp: ' + data.temp +
' °C' + ' | Humid: ' + data.hum + '%'
+ ' | soaking read nr' + str(len(magnitudes_queue)))
self.store_and_plot_data(next_temp, next_hum, data, self.cook_up_equi_indicator())
writer.writerow(self.data_collection[-1])
if self.temperature_stable and self.humidity_stable and self.magnitude_stable and\
self.phase_stable:
reference_magnitude = magnitudes_queue[-1]
reference_phase = phase_queue[-1]
print('SOAKING FINISHED!')
break
else:
self.sleep_until(next_read_time)
next_read_time += self.sleep_time
# perform the configured number of measurements and check that they are really stable
# It started running after everything become stable
supposedly_stable_measurements = []
all_measurements_stable = True
for i in range(0, next_reads):
data = self.read_data()
self.temperature_stable = self.calculate_temperature_stability(next_temp, float(data.temp))
self.humidity_stable = self.calculate_humidity_stability(next_hum, float(data.hum))
mag = self.calculate_mean_magnitude(data.s21)
phase = self.calculate_mean_phase(data.s21)
self.magnitude_stable = (reference_magnitude-self.max_delta_mag <= mag) and\
(mag <= reference_magnitude+self.max_delta_mag)
self.phase_stable = (reference_phase-self.max_delta_phase <= phase) and\
(phase <= reference_phase+self.max_delta_phase)
self.store_and_plot_data(next_temp, next_hum, data, self.cook_up_equi_indicator())
supposedly_stable_measurements.append(self.data_collection[-1])
set_const_response = self.chamber.set_const((target_temp, target_hum))
print(set_const_response)
set_mode_response = self.chamber.set_mode('CONSTANT')
print(set_mode_response)
number_of_soaking_reads = soaking_time / self.sleep_time + 1
do_another_measurement = True
#next_read_time is the starttime of the second read (i.e. read after this one). The time of
#this read (i.e. the first read in the measurement) is now().
next_read_time = self.clock.time() + self.sleep_time
while do_another_measurement:
# wait until set point is reached (+soaking time)
magnitudes_queue = []
phase_queue = []
while True:
data = self.read_data()
# if it is not within the target range reset start time
self.temperature_stable = False
self.humidity_stable = False
self.temperature_stable = self.calculate_temperature_stability(target_temp, float(data.temp))
self.humidity_stable = self.calculate_humidity_stability(target_hum, float(data.hum))
# The queue must not be longer than the max number of soaking reads.
# If the queue is already full, we have to pop the first element before we can add the
# current measurement.
if len(magnitudes_queue) >= number_of_soaking_reads:
magnitudes_queue.pop(0)
if self.temperature_stable and self.humidity_stable:
magnitudes_queue.append(self.calculate_mean_magnitude(data.s21))
else:
magnitudes_queue.clear()
# check cable stability parameters
self.magnitude_stable = False
if len(magnitudes_queue) >= number_of_soaking_reads:
spread = max(magnitudes_queue) - min(magnitudes_queue)
if spread < 2*self.max_delta_mag:
self.magnitude_stable = True
if len(phase_queue) >= number_of_soaking_reads:
phase_queue.pop(0)
if self.temperature_stable and self.humidity_stable:
phase_queue.append(self.calculate_mean_phase(data.s21))
else:
phase_queue.clear()
self.phase_stable = False
if len(phase_queue) >= number_of_soaking_reads:
spread = max(phase_queue) - min(phase_queue)
if spread < 2*self.max_delta_phase:
self.phase_stable = True
print('Setpoint: ' + str(target_temp) + ' ' + str(target_hum) + ' | Temp: ' + data.temp +
' °C' + ' | Humid: ' + data.hum + '%'
+ ' | soaking read nr' + str(len(magnitudes_queue)))
self.store_and_plot_data(target_temp, target_hum, data, self.cook_up_equi_indicator())
writer.writerow(self.data_collection[-1])
if self.temperature_stable and self.humidity_stable and self.magnitude_stable and\
self.phase_stable:
reference_magnitude = magnitudes_queue[-1]
reference_phase = phase_queue[-1]
print('SOAKING FINISHED!')
break
else:
self.sleep_until(next_read_time)
next_read_time += self.sleep_time
# perform the configured number of measurements and check that they are really stable
# It started running after everything become stable
supposedly_stable_measurements = []
all_measurements_stable = True
for i in range(0, n_stable_reads):
data = self.read_data()
self.temperature_stable = self.calculate_temperature_stability(target_temp, float(data.temp))
self.humidity_stable = self.calculate_humidity_stability(target_hum, float(data.hum))
mag = self.calculate_mean_magnitude(data.s21)
phase = self.calculate_mean_phase(data.s21)
self.magnitude_stable = (reference_magnitude-self.max_delta_mag <= mag) and\
(mag <= reference_magnitude+self.max_delta_mag)
self.phase_stable = (reference_phase-self.max_delta_phase <= phase) and\
(phase <= reference_phase+self.max_delta_phase)
self.store_and_plot_data(target_temp, target_hum, data, self.cook_up_equi_indicator())
supposedly_stable_measurements.append(self.data_collection[-1])
if (self.temperature_stable and self.humidity_stable and self.magnitude_stable and
self.phase_stable):
print('Stable measurement ' + str(i+1) + '/' + str(next_reads))
self.sleep_until(next_read_time)
next_read_time += self.sleep_time
else:
all_measurements_stable = False
print('Measurement not stable. Retrying.')
break
for measurement in supposedly_stable_measurements:
if all_measurements_stable:
measurement['EQUILIBRIUM_INDICATOR'] = TEMPERATURE_STABLE | HUMIDITY_STABLE |\
MAGNITUDE_STABLE | PHASE_STABLE |\
MEASUREMENT_STABLE
do_another_measurement = False
writer.writerow(measurement)
plt.close()
except KeyboardInterrupt:
pass
if self.standby:
standby_response = self.chamber.set_mode('STANDBY')
print(standby_response)
if (self.temperature_stable and self.humidity_stable and self.magnitude_stable and
self.phase_stable):
print('Stable measurement ' + str(i+1) + '/' + str(n_stable_reads))
self.sleep_until(next_read_time)
next_read_time += self.sleep_time
else:
all_measurements_stable = False
print('Measurement not stable. Retrying.')
break
for measurement in supposedly_stable_measurements:
if all_measurements_stable:
measurement['EQUILIBRIUM_INDICATOR'] = TEMPERATURE_STABLE | HUMIDITY_STABLE |\
MAGNITUDE_STABLE | PHASE_STABLE |\
MEASUREMENT_STABLE
do_another_measurement = False
writer.writerow(measurement)
def sleep_until(self, wakeup_time):
remaining_sleep_time = wakeup_time - self.clock.time()
......@@ -298,12 +306,17 @@ class Measurements:
return (target_hum-self.max_delta_hum <= float(readback_hum)) and \
(float(readback_hum) <= target_hum+self.max_delta_hum)
def plot_output(output_basename, show_blocking_plot, save_pdf=True):
data_frame = pd.read_csv(output_basename+'.csv')
def plot_output(output_basename, n_measurements, show_blocking_plot):
list_of_frames = []
for m in range(n_measurements):
measurement_name = output_basename+'_'+str(m)
list_of_frames.append(pd.read_csv(measurement_name+'.csv'))
combined_data_frame = pd.concat(list_of_frames, ignore_index=True, sort=False)
if show_blocking_plot:
plt.ioff()
plot = MeasurementPlot.MeasurementPlot()
plot.draw(data_frame, output_basename + '_graph.pdf')
plot.draw(combined_data_frame, output_basename + '_graph.pdf')
if __name__ == '__main__':
parser = ArgumentParser()
......@@ -341,11 +354,11 @@ if __name__ == '__main__':
with open('test_stand_parameter.json', 'r') as f:
config_data = json.load(f)
mes = Measurements(args.chamber, args.vna, args.file, output_basename+'.csv', args.standby, config_data)
mes = Measurements(args.chamber, args.vna, args.file, output_basename, args.standby, config_data)
try:
mes.perform_measurements()
plot_output(output_basename, args.plot)
n_measurements = mes.perform_measurements()
plot_output(output_basename, n_measurements, args.plot)
finally:
mes.chamber.close()
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

imprint | privacy | Conditions of Use