Skip to content

Use raw data to blank instead of calculated OD #23

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dgruano wants to merge 2 commits into
base: rc
Choose a base branch
from
Open

Use raw data to blank instead of calculated OD #23

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ba7d532
New OD blank method
dgruano Aug 25, 2020
d4104e3
Major fixes
dgruano Aug 26, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
170 changes: 129 additions & 41 deletions experiment/template/eVOLVER.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
SAVE_PATH = os.path.dirname(os.path.realpath(__file__))
EXP_DIR = os.path.join(SAVE_PATH, EXP_NAME)
OD_CAL_PATH = os.path.join(SAVE_PATH, 'od_cal.json')
OD_RAW_ZERO_PATH = os.path.join(SAVE_PATH, 'od_raw_zero.json')
TEMP_CAL_PATH = os.path.join(SAVE_PATH, 'temp_cal.json')

SIGMOID = 'sigmoid'
Expand Down Expand Up @@ -69,22 +70,47 @@ def on_broadcast(self, data):
with open(TEMP_CAL_PATH) as f:
temp_cal = json.load(f)

# apply calibrations
# update temperatures if needed
data = self.transform_data(data, VIALS, od_cal, temp_cal)
if data is None:
logger.error('could not tranform raw data, skipping user-'
'defined functions')
return

# Store the OD blank depending on the options set (Raw blank, OD blank or none)
# should we "blank" the OD?
if self.use_blank and self.OD_initial is None:
logger.info('setting initial OD reading')
self.OD_initial = data['transformed']['od']
elif self.OD_initial is None:
self.OD_initial = np.zeros(len(VIALS))
data['transformed']['od'] = (data['transformed']['od'] -
self.OD_initial)
if self.OD_initial is None:
if self.use_blank and self.use_raw_blank:
logger.info('setting initial OD reading (raw_values)')
"""
Given Raw_cal_0, Raw_exp_0 and Raw_exp_t
We can calculate delta as:
delta = Raw_expt_t - Raw_exp_0
And therefore calculate the OD as:
OD = f(Raw_cal_0 + delta)
Which extended is:
OD = f(Raw_cal_0 - Raw_expt_0 + Raw_expt_t
So we can store "Raw_cal_0 - Raw_expt_0" in self.OD_initial
And add it to the measured Raw_expt_t before calculating the final OD
"""
# get calibration raw blank
with open(OD_RAW_ZERO_PATH, 'r') as f:
zero_cal_values = np.array(json.load(f))

self.OD_initial = zero_cal_values - np.array(
[float(x) for x in data['data']['od_135']]) # TODO: generalize for other od parameters

@zheins zheins Oct 14, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please do this - it's very important for other users with different setups.


elif self.use_blank: # This used to be the normal procedure
logger.info('setting initial OD reading (OD values)')
data = self.apply_OD_calibration(data, VIALS, od_cal)
self.OD_initial = data['transformed']['od']

else:
self.OD_initial = np.zeros(len(VIALS))

# Apply calibration and blank (If it's raw blank, before cal. If it's OD blank, after cal.)
data = self.apply_OD_calibration(data, VIALS, od_cal)

# save data
self.save_data(data['transformed']['od'], elapsed_time,
VIALS, 'OD')
Expand All @@ -100,7 +126,7 @@ def on_broadcast(self, data):
# run custom functions
self.custom_functions(data, VIALS, elapsed_time)
# save variables
self.save_variables(self.start_time, self.OD_initial)
self.save_variables(self.start_time, self.OD_initial, self.use_raw_blank)

def on_activecalibrations(self, data):
print('Calibrations recieved')
Expand All @@ -124,6 +150,25 @@ def on_activecalibrations(self, data):
x,
time.strftime("%c"))
self._create_file(x, param + '_raw', defaults=[exp_str])
try:
if calibration['calibrationType'] == 'od' and param == 'od_135': # TODO: generalize for other od parameters

@zheins zheins Oct 14, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please geenralize. It shouldn't be too much work. You basically just need to look at the incoming data and see what params exist and what type of calibration is being applied and which parameter is desired. Technically the od param can be named anything, and the calibration should guide you for what to look for.

# Fetch raw calibration values for OD = 0
zero_cal_raw = []
for x in calibration['raw']:
if x['param'] == 'od_135':
raw_cal_values = x['vialData']

for c, od_list in enumerate(calibration['measuredData']):
ind = od_list.index(0)
zero_cal_raw.append(sum(raw_cal_values[c][ind]) / 3) # Store the mean raw zero value

with open(OD_RAW_ZERO_PATH, 'w') as f:
json.dump(zero_cal_raw, f)
except Exception as e:
logger.error(f"Error '{e}' when calculating zero raw values of the calibration.")
logger.INFO("Changing to former OD blank method.")
print(e)
self.use_raw_blank = False
break

def request_calibrations(self):
Expand Down Expand Up @@ -162,36 +207,8 @@ def transform_data(self, data, vials, od_cal, temp_cal):
temp_set_data = np.genfromtxt(file_path, delimiter=',')
temp_set = temp_set_data[len(temp_set_data)-1][1]
temps.append(temp_set)
od_coefficients = od_cal['coefficients'][x]
temp_coefficients = temp_cal['coefficients'][x]
try:
if od_cal['type'] == SIGMOID:
#convert raw photodiode data into ODdata using calibration curve
od_data[x] = np.real(od_coefficients[2] -
((np.log10((od_coefficients[1] -
od_coefficients[0]) /
(float(od_data[x]) -
od_coefficients[0])-1)) /
od_coefficients[3]))
if not np.isfinite(od_data[x]):
od_data[x] = 'NaN'
logger.debug('OD from vial %d: %s' % (x, od_data[x]))
else:
logger.debug('OD from vial %d: %.3f' % (x, od_data[x]))
elif od_cal['type'] == THREE_DIMENSION:
od_data[x] = np.real(od_coefficients[0] +
(od_coefficients[1]*od_data[x]) +
(od_coefficients[2]*od_data_2[x]) +
(od_coefficients[3]*(od_data[x]**2)) +
(od_coefficients[4]*od_data[x]*od_data_2[x]) +
(od_coefficients[5]*(od_data_2[x]**2)))
else:
logger.error('OD calibration not of supported type!')
od_data[x] = 'NaN'
except ValueError:
print("OD Read Error")
logger.error('OD read error for vial %d, setting to NaN' % x)
od_data[x] = 'NaN'

try:
temp_data[x] = (float(temp_data[x]) *
temp_coefficients[0]) + temp_coefficients[1]
Expand Down Expand Up @@ -240,6 +257,69 @@ def transform_data(self, data, vials, od_cal, temp_cal):
data['transformed']['temp'] = temp_data
return data

def apply_OD_calibration(self, data, vials, od_cal):
od_data_2 = None
if od_cal['type'] == THREE_DIMENSION:
od_data_2 = data['data'].get(od_cal['params'][1], None)

od_data = data['data'].get(od_cal['params'][0], None)

if self.use_raw_blank:
zero_delta = self.OD_initial
od_blank = np.zeros(len(vials))
else:
zero_delta = np.zeros(len(vials))
od_blank = self.OD_initial

if od_data is None:
print('Incomplete data recieved, Error with measurement')
logger.error('Incomplete data received, error with measurements')
return None
if 'NaN' in od_data:
print('NaN recieved, Error with measurement')
logger.error('NaN received, error with measurements')
return None

od_data = np.array([float(x) for x in od_data])
if od_data_2:
od_data_2 = np.array([float(x) for x in od_data_2])

for x in vials:
od_coefficients = od_cal['coefficients'][x]
try:
if od_cal['type'] == SIGMOID:
#convert raw photodiode data into ODdata using calibration curve
od_data[x] = np.real(od_coefficients[2] -
((np.log10((od_coefficients[1] -
od_coefficients[0]) /
(zero_delta[x] + float(od_data[x]) -
od_coefficients[0])-1)) /
od_coefficients[3]))
if not np.isfinite(od_data[x]):
od_data[x] = 'NaN'
logger.debug('OD from vial %d: %s' % (x, od_data[x]))
else:
logger.debug('OD from vial %d: %.3f' % (x, od_data[x]))
elif od_cal['type'] == THREE_DIMENSION:
od_data[x] = np.real(od_coefficients[0] +
(od_coefficients[1]*od_data[x]) +
(od_coefficients[2]*od_data_2[x]) +
(od_coefficients[3]*(od_data[x]**2)) +
(od_coefficients[4]*od_data[x]*od_data_2[x]) +

@zheins zheins Oct 14, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is the zero-delta calc changes being applied to the other methods, or just sigmoid?

(od_coefficients[5]*(od_data_2[x]**2)))
else:
logger.error('OD calibration not of supported type!')
od_data[x] = 'NaN'
except ValueError as e:
print("OD Read Error")
#print(e)
logger.error('OD read error for vial %d, setting to NaN' % x)
od_data[x] = 'NaN'

# update od data in the data dictionary
data['transformed']['od'] = od_data - od_blank
return data

def update_stir_rate(self, stir_rates, immediate = False):
data = {'param': 'stir', 'value': stir_rates,
'immediate': immediate, 'recurring': True}
Expand Down Expand Up @@ -391,9 +471,16 @@ def initialize_exp(self, vials, always_yes=False):
if exp_blank == 'y':
# will do it with first broadcast
self.use_blank = True
logger.info('will use initial OD measurement as blank')
raw_blank = input('Use raw blank instead of OD blank? (y/n): ')
if raw_blank == 'y':
logger.info('will use initial raw measurement as blank')
self.use_raw_blank = True
else:
logger.info('will use initial OD measurement as blank')
self.use_raw_blank = False
else:
self.use_blank = False
self.use_raw_blank = False
self.OD_initial = np.zeros(len(vials))
else:
# load existing experiment
Expand All @@ -405,6 +492,7 @@ def initialize_exp(self, vials, always_yes=False):
x = loaded_var
start_time = x[0]
self.OD_initial = x[1]
self.use_raw_blank = x[2]

# copy current custom script to txt file
backup_filename = '{0}_{1}.txt'.format(EXP_NAME,
Expand Down Expand Up @@ -435,14 +523,14 @@ def save_data(self, data, elapsed_time, vials, parameter):
text_file.write("{0},{1}\n".format(elapsed_time, data[x]))
text_file.close()

def save_variables(self, start_time, OD_initial):
def save_variables(self, start_time, OD_initial, use_raw_blank):
# save variables needed for restarting experiment later
save_path = os.path.dirname(os.path.realpath(__file__))
pickle_name = "{0}.pickle".format(EXP_NAME)
pickle_path = os.path.join(EXP_DIR, pickle_name)
logger.debug('saving all variables: %s' % pickle_path)
with open(pickle_path, 'wb') as f:
pickle.dump([start_time, OD_initial], f)
pickle.dump([start_time, OD_initial, use_raw_blank], f)

def get_flow_rate(self):
file_path = os.path.join(SAVE_PATH, PUMP_CAL_FILE)
Expand Down