333 lines
11 KiB
Python
333 lines
11 KiB
Python
# module for comparing different experiments
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import plotly.express as px
|
|
import plotly.graph_objects as go
|
|
from plotly.subplots import make_subplots
|
|
from pathlib import Path
|
|
from scipy.optimize import curve_fit
|
|
|
|
|
|
|
|
##########################################
|
|
# Auxiliary Functions
|
|
##########################################
|
|
|
|
|
|
def get_data_subset(savefolder_path, exp_name, parameter):
|
|
"""Returns subset of data from instrument of experiment exp_name or from file (with path). If no data found, return an empty dummy dataframe."""
|
|
|
|
instrument = parameter[:parameter.find("-")]
|
|
param = parameter[parameter.find("-")+1:]
|
|
|
|
savefolder = Path(savefolder_path) / exp_name # name of the folder to save the data and summary of the experiment
|
|
datafile = f"{savefolder}/Processed_Data/{exp_name}_Data_{instrument}.csv" # file containing the processed data
|
|
|
|
try:
|
|
data = pd.read_csv(datafile, sep=';')
|
|
mask = (
|
|
(data["Time since Irradiation [min]"] >= -30) &
|
|
(data["Time since Irradiation [min]"] <= 60)
|
|
)
|
|
df_subset = data.copy().loc[mask]
|
|
|
|
except: # no data available
|
|
if param == "CO2":
|
|
unit = "ppm"
|
|
else:
|
|
unit = "ppb"
|
|
|
|
d = {"Time since Irradiation [min]": [None, None], f"{param} [{unit}]": [None, None]}
|
|
df_subset = pd.DataFrame(data=d)
|
|
|
|
return df_subset
|
|
|
|
|
|
def find_closest_indices(t1, t2): # by chatgpt
|
|
"""
|
|
For each time in t1, find the index of the closest time in t2.
|
|
|
|
Parameters
|
|
----------
|
|
t1 : array-like
|
|
Reference time points (fewer points)
|
|
t2 : array-like
|
|
Target time points (more points)
|
|
|
|
Returns
|
|
-------
|
|
indices : np.ndarray
|
|
ინდices in t2 corresponding to closest times to t1
|
|
"""
|
|
t1 = np.asarray(t1)
|
|
t2 = np.asarray(t2)
|
|
|
|
# Find insertion positions
|
|
idx = np.searchsorted(t2, t1)
|
|
|
|
# Clip to valid range
|
|
idx = np.clip(idx, 1, len(t2) - 1)
|
|
|
|
# Compare neighbors (left vs right)
|
|
left = t2[idx - 1]
|
|
right = t2[idx]
|
|
|
|
idx -= (t1 - left < right - t1)
|
|
|
|
return idx
|
|
|
|
|
|
def reformat_title(text):
|
|
# reformat title from YYYYMMDD_Comp1_..._CompN_R1_..._RN_RHXX(_extra) to Comp1:...:CompN R1:...:RN (XX% RH)
|
|
|
|
text_s = text[9:]
|
|
firstnum = next((i for i, c in enumerate(text_s) if c.isdigit()), -1)
|
|
RH_ind = text_s.find("RH")
|
|
compounds = text_s[:firstnum-1].replace("_", ":")
|
|
ratios = text_s[firstnum:RH_ind-1].replace("_", ":")
|
|
rh = text_s[RH_ind+2:]
|
|
|
|
title = compounds + " " + ratios + " (" + rh + "% RH)"
|
|
return title
|
|
|
|
|
|
|
|
##########################################
|
|
# Comparison Plots
|
|
##########################################
|
|
|
|
|
|
def compare_3_plots(savefolder_path, exp_list, param_list, color_dict):
|
|
"""Plots the (processed) data of the chosen instrument and parameter for all experiments in exp_list (list of savenames)"""
|
|
|
|
n_exp = len(exp_list)
|
|
#fig = go.Figure()
|
|
fig = make_subplots(
|
|
rows=1,
|
|
cols=n_exp,
|
|
specs=[[{"secondary_y": True}, {"secondary_y": True}, {"secondary_y": True}]], # currently only works for 3 subplots
|
|
subplot_titles=[reformat_title(e) for e in exp_list], # ["CO2", "NO", "HONO", "NO2"],
|
|
horizontal_spacing = 0.1,
|
|
vertical_spacing = 0.05
|
|
)
|
|
fig.update_layout(template="seaborn", font=dict(size=20))
|
|
fig.update_annotations(font_size=20) # font size of subplot titles
|
|
|
|
# if colors: # custom colors
|
|
# cols = colors
|
|
# else: # default colors
|
|
# cols = ['#033198', '#2064AE', '#3292C2', '#5DC1D3', '#A4E5D3', '#D0E3A3', '#C8B455', '#B0802B', '#995215', '#7E1700']
|
|
|
|
for n, experiment in enumerate(exp_list):
|
|
|
|
savefolder = Path(savefolder_path) / exp_list[n] # name of the folder to save the data and summary of the experiment
|
|
|
|
for p in param_list:
|
|
|
|
instrument = p[:p.find("-")]
|
|
param = p[p.find("-")+1:]
|
|
|
|
if param == "CO2":
|
|
unit = "ppm"
|
|
sec_y = True
|
|
else:
|
|
unit = "ppb"
|
|
sec_y = False
|
|
|
|
datafile = f"{savefolder}/Processed_Data/{exp_list[n]}_Data_{instrument}.csv" # file containing the processed data
|
|
|
|
try:
|
|
data = pd.read_csv(datafile, sep=';')
|
|
mask = (
|
|
(data["Time since Irradiation [min]"] >= -30) &
|
|
(data["Time since Irradiation [min]"] <= 60)
|
|
)
|
|
df_subset = data.copy().loc[mask]
|
|
except: # no data available
|
|
d = {"Time since Irradiation [min]": [None, None], f"{param} [{unit}]": [None, None]}
|
|
df_subset = pd.DataFrame(data=d)
|
|
|
|
fig.add_trace(go.Scatter(x=df_subset["Time since Irradiation [min]"], y=df_subset[f"{param} [{unit}]"],
|
|
mode='lines',
|
|
#name=p,
|
|
line=dict(color=color_dict[instrument][param]),
|
|
name=param,
|
|
legendgroup=param,
|
|
showlegend=(n == 1),
|
|
),
|
|
row=1, col=n+1,
|
|
secondary_y=sec_y,
|
|
)
|
|
fig.update_yaxes(title_text=f"NOx, HONO Concentration [ppb]",
|
|
title_font=dict(color=color_dict["Ecophysics_NOx_899"]["NOx"]),
|
|
tickfont=dict(color=color_dict["Ecophysics_NOx_899"]["NOx"]),
|
|
)
|
|
fig.update_yaxes(title_text=f"CO2 Concentration [ppm]",
|
|
title_font=dict(color="black"),
|
|
tickfont=dict(color="black"),
|
|
secondary_y=True,
|
|
)
|
|
|
|
fig.update_xaxes(title_text="Time since Irradiation [min]", range=(-10,50))
|
|
|
|
fig.add_shape(
|
|
type="rect",
|
|
x0=0,
|
|
x1=35,
|
|
y0=0,
|
|
y1=1,
|
|
xref=f"x{n+1}", #"x1",
|
|
yref="paper",
|
|
fillcolor="gold",
|
|
opacity=0.3,
|
|
line_width=0,
|
|
layer="below",
|
|
)
|
|
|
|
|
|
return fig
|
|
|
|
|
|
def compare_parameter(savefolder_path, exp_list, param_list, colors=None):
|
|
"""Plots the (processed) data of the chosen instrument-parameter list (!) for all experiments in exp_list (list of savenames)"""
|
|
# idea: if a list is given for the parameter: calculate the sum, using x axis of first parameter as basis.
|
|
|
|
if colors: # custom colors
|
|
cols = colors
|
|
else: # default colors
|
|
cols = px.colors.qualitative.Plotly
|
|
|
|
fig = go.Figure()
|
|
fig.update_layout(template="plotly_white")
|
|
|
|
#print(param_list) # for debugging
|
|
|
|
for n, experiment in enumerate(exp_list):
|
|
|
|
for k, parameter in enumerate(param_list):
|
|
|
|
instrument = parameter[:parameter.find("-")]
|
|
param = parameter[parameter.find("-")+1:]
|
|
|
|
if param == "CO2":
|
|
unit = "ppm"
|
|
else:
|
|
unit = "ppb"
|
|
|
|
df_subset = get_data_subset(savefolder_path, experiment, parameter)
|
|
|
|
if k==0:
|
|
data_x = np.asarray(df_subset["Time since Irradiation [min]"])
|
|
data_y = np.asarray(df_subset[f"{param} [{unit}]"])
|
|
else:
|
|
data_xn = np.asarray(df_subset["Time since Irradiation [min]"])
|
|
data_yn = np.asarray(df_subset[f"{param} [{unit}]"])
|
|
idx = find_closest_indices(data_x, data_xn)
|
|
data_y = data_y + data_yn[idx]
|
|
|
|
if param == "CO2" and len(param_list)>1:
|
|
data_y = data_y * 1000 # convert to ppb
|
|
|
|
fig.add_trace(go.Scatter(x=data_x, y=data_y,
|
|
mode='lines',
|
|
name=experiment,
|
|
line=dict(color=cols[n])
|
|
)
|
|
)
|
|
|
|
fig.update_xaxes(title_text="Time since Irradiation [min]", range=(-10,50))
|
|
fig.update_yaxes(title_text="Concentration [ppb]")
|
|
|
|
if len(param_list)==1:
|
|
p = param_list[0]
|
|
fig.update_layout(title_text = p[p.find("-")+1:] + " (" + p[:p.find("-")] + ")")
|
|
else:
|
|
sum_params = [p[p.find("-")+1:] + " (" + p[:p.find("-")] + ")" for p in param_list]
|
|
fig.update_layout(title_text = " + ".join(sum_params))
|
|
|
|
return fig
|
|
|
|
|
|
def compare_n_params(savefolder_path, exp_list, param_list, layout_dict, colors, subtitles=None, title=None):
|
|
"""Plots the (processed) data of the chosen instrument and parameter for all experiments in exp_list (list of savenames). Subplot titles (other than default) & overall title can be provided optionally."""
|
|
|
|
# make a plot for each individual parameter
|
|
fig_list = []
|
|
for p in param_list:
|
|
fig_list.append(compare_parameter(savefolder_path, exp_list, p if isinstance(p, list) else [p], colors))
|
|
|
|
if subtitles==None:
|
|
subts = [p[p.find("-")+1:] + " (" + p[:p.find("-")] + ")" for p in param_list]
|
|
else:
|
|
subts = subtitles
|
|
|
|
# place the individual plots in one plot
|
|
fig = make_subplots(
|
|
rows=layout_dict["nrows"],
|
|
cols=layout_dict["ncols"],
|
|
subplot_titles=subts,
|
|
#subplot_titles=[reformat_title(e) for e in exp_list], # ["CO2", "NO", "HONO", "NO2"],
|
|
horizontal_spacing = layout_dict["hspace"],
|
|
vertical_spacing = layout_dict["vspace"]
|
|
)
|
|
fig.update_layout(template="seaborn", font=dict(size=layout_dict["font_size"]))
|
|
fig.update_annotations(font_size=layout_dict["title_size"]) # font size of subplot titles
|
|
|
|
# draw the subplots
|
|
for i, f in enumerate(fig_list):
|
|
|
|
if i == 0:
|
|
legend = True
|
|
else:
|
|
legend = False
|
|
|
|
for trace in f.data:
|
|
trace.legendgroup = trace.name
|
|
trace.showlegend = legend
|
|
fig.add_trace(trace, row=i//layout_dict["ncols"]+1, col=i%layout_dict["ncols"]+1)
|
|
|
|
param = param_list[i]
|
|
if isinstance(param, list): # if list (sum)
|
|
fig.update_yaxes(title_text="Concentration [ppb]", row=i//layout_dict["ncols"]+1, col=i%layout_dict["ncols"]+1)
|
|
else:
|
|
par = param[param.find("-")+1:]
|
|
if par == "CO2":
|
|
unit = "ppm"
|
|
else:
|
|
unit = "ppb"
|
|
fig.update_yaxes(title_text=f"{par} [{unit}]", row=i//layout_dict["ncols"]+1, col=i%layout_dict["ncols"]+1)
|
|
|
|
fig.add_shape(
|
|
type="rect",
|
|
x0=0,
|
|
x1=35,
|
|
y0=-20,
|
|
y1=500,
|
|
xref=f"x{i+1}", #"x1",
|
|
yref= f"y{i+1}",
|
|
fillcolor="rgba(255, 215, 0, 0.5)",
|
|
line_width=0,
|
|
layer="below",
|
|
)
|
|
|
|
|
|
fig.update_xaxes(range=(-10,50))
|
|
for n in range(layout_dict["ncols"]):
|
|
fig.update_xaxes(title_text="Time since Irradiation [min]", row=layout_dict["nrows"], col=n+1)
|
|
|
|
if title:
|
|
fig.update_layout(
|
|
title=dict(
|
|
text=title,
|
|
x=0.125,
|
|
y=0.95,
|
|
xanchor="center",
|
|
yanchor="top"
|
|
),
|
|
)
|
|
|
|
return fig
|
|
|
|
|