import os.path
import shutil
from IPython.utils.terminal import toggle_set_term_title
from ..data.prsdata import *
from ..data.plot import *
from ..data.calc import calc_absorption_data
from devctrl.utility.file_io import listdir_abs, listdir_filter
from ..data.util import add_absorption_data, find_all_data_dirs, load_rta, find_all_rta_data_dirs, get_data_name_template, write_rta_csv
from ..data.check import sanity_check_data_gen_report
import time
from multiprocessing.pool import Pool
import logging
log = logging.getLogger(__name__)
[docs]
def setup_compare():
dirs = listdir_abs(r"Q:\photoreflectance\data-M-Ta3N5")
datas = [PrsData(dir) for dir in dirs if os.path.isdir(dir)]
wls = np.array(datas[0].wavelengths)
wls = wls[wls > 430]
fig, ax = plt.subplots(2, 1, figsize=(6, 7))
plot_spectrum_on_ax(datas[0].get_spectrum_data(wavelengths=wls)[0], ax[0], "dI_I", error_shade=True, label="Old Setup", color="#f00")
plot_spectrum_on_ax(datas[2].get_spectrum_data(wavelengths=wls)[0], ax[0], "dI_I", error_shade=True, label="New Setup", color="#52f")
ax[0].legend()
plot_spectrum_on_ax(datas[1].get_spectrum_data(wavelengths=wls)[0], ax[1], "dI_I", error_shade=True, label="Old Setup", color="#f00", mode=TRANSMISSION)
plot_spectrum_on_ax(datas[3].get_spectrum_data(wavelengths=wls)[0], ax[1], "dI_I", error_shade=True, label="New Setup", color="#52f", mode=TRANSMISSION)
ax[1].legend()
add_wl_axis(ax[0])
add_wl_axis(ax[1], label=False)
ax[1].set_xlabel(PrsData.get_column_tex_label_with_unit("E"))
fig.suptitle("Setup Comparison M-Ta3N5")
fig.tight_layout()
return fig, ax
[docs]
def move_rta_data_to_subdir(base_dir: str, move_files=True):
"""
Move reflection, transmission and absorption data directories to a subdirectory.
Eg:
data/measurement-1-ref/...
data/measurement-1-tra/...
data/measurement-1-abs/...
Will be moved to
data/measurement-1/measurement-1-ref
data/measurement-1/measurement-1-tra
data/measurement-1/measurement-1-abs
"""
rta_dirs = find_all_rta_data_dirs(base_dir)
print(rta_dirs)
for shared_dirpath, ref_dir, tra_dir, abs_dir in rta_dirs:
dirname = os.path.basename(shared_dirpath)
parent_dir = os.path.dirname(shared_dirpath)
if parent_dir.endswith(dirname):
log.debug(f"RTA data already in own subdir: {shared_dirpath}")
continue
shared_dir = shared_dirpath
if os.path.exists(shared_dir):
log.error(f"Path already exists: '{shared_dir}'")
continue
log.info(f"Creating {shared_dir}")
os.makedirs(shared_dir)
shutil.move(ref_dir, shared_dir)
shutil.move(tra_dir, shared_dir)
if abs_dir:
shutil.move(abs_dir, shared_dir)
if move_files:
files = listdir_abs(parent_dir)
for file in files:
if os.path.isfile(file) and file.startswith(dirname):
log.info(f"Moving file: '{file}'")
shutil.move(file, shared_dir)
[docs]
def merge_datas(first, second, combined, **plot_spectrum_kw):
data1 = PrsData(first)
data2 = PrsData(second)
datac = PrsData(combined, data={}, file_mode="w")
datac.metadata = data1.metadata.copy()
datac.metadata["combined_from"] = data1.name, data2.name
datac.data = data1.data.copy()
for k, v in data2.data.items():
if k in data1.data:
log.warning(f"Data key '{k}' exists in both datas - taking the one from first data")
continue
datac.data[k] = v
datac.update_wavelengths()
datac.write_full_file()
datac.save_csv()
plot_data(datac, **plot_spectrum_kw)
[docs]
def refresh_data(data: PrsData, plot_spectrum_kw):
try:
data.remove_calculated_values()
data.update_wavelengths()
# # TODO: remove later
# # remove offsets that are not real
# keys = list(data.data.keys())
# for k in keys:
# if type(k) != str: continue
# R = data.get_for_wl(k, "lock-in-aux")
# dR = data.get_for_wl(k, "lock-in-R")
# theta = data.get_for_wl(k, "lock-in-theta")
# if R == 0 and dR == 0 and theta == 0:
# log.warning(f"Offset {k} has all zero values and will be removed!")
# del data.data[k]
# data.remove_calculated_values()
data.save_csv()
if data.has_AC() and data.has_DC():
_ = data.get_spectrum_data() # force calculation of all values
elif data.has_DC():
_ = data.get_spectrum_data(columns=PrsData.spectrum_columns_reference) # force calculation of all values
# title = get_data_name_template(data.dirname, template="<sample_type> (<sample_batch>) [<mode>]")
title = get_data_name_template(data.dirname, template="<sample_type> <sample_batch> <info> <mode>")
# title = data.dirname.replace("_", " ")
plot_data(data, **(dict(title=title) | plot_spectrum_kw))
data.write_full_file() # write full file AFTER plot/save csv since then all necessary values are calculated
rep = sanity_check_data_gen_report(data)
if rep:
reppath = os.path.join(data.dirpath, data.dirname + "-report.md")
with open(reppath, "w") as file:
file.write(rep)
return rep, None
except Exception as e:
log.error(f"Error in refresh of '{data.data_path}': {type(e)}{e}")
import traceback
tb = traceback.format_exc()
return "", (data.data_path, e, tb)
[docs]
def refresh_rta_data(shared, r_dir, t_dir, a_dir, norm_data, plot_spectrum_kw):
log.info(f"Refreshing rta in base: '{shared}'")
try:
r = PrsData(r_dir, file_mode="rw")
t = PrsData(t_dir, file_mode="rw")
rep_r, err = refresh_data(r, plot_spectrum_kw)
if err is not None:
raise err[1]
rep_t, err = refresh_data(t, plot_spectrum_kw)
if err is not None:
raise err[1]
rep = rep_r + "\n\n" + rep_t + "\n\n"
a_dir = shared + "-" + ABSORPTION
a = add_absorption_data(r, t, a_dir, norm_data, replace_existing=True, replot_existing=True, **(dict(max_ylim=(-100, 100)) | plot_spectrum_kw))
if a:
rep += sanity_check_data_gen_report(a)
rep = rep.strip("\n")
if rep:
reppath = shared + "-report.md"
with open(reppath, "w") as file:
file.write(rep)
# csvpath = shared + ".csv"
# write_rta_csv(r, t, a, save_file=csvpath, columns=["dI_I"])
plt.close("all")
return rep, None
except Exception as e:
log.error(f"Error in rta refresh of '{shared}': {e}")
import traceback
tb = traceback.format_exc()
return "", (shared, e, tb)
[docs]
def refresh_all_datas(base_dir: str, norm_dir:str|None, **plot_spectrum_kw):
"""
For all datas in base_dir:
- recalculate all values from the raw data
- re-generate all plots
- re-generate csv
- re-new absorption data
"""
print(f"{'START refresh_all_datas':#^40}")
n = PrsData(norm_dir) if norm_dir else None
processed_data_dirs = []
rta_dirs = find_all_rta_data_dirs(base_dir)
all_dirs = find_all_data_dirs(base_dir)
res = []
# 1) RTA datas
with Pool() as pool:
res += pool.starmap(refresh_rta_data, [(shared, r_dir, t_dir, a_dir, n, plot_spectrum_kw) for shared, r_dir, t_dir, a_dir in rta_dirs])
for shared, r_dir, t_dir, a_dir in rta_dirs:
processed_data_dirs += [r_dir, t_dir, a_dir]
# 2) Remaining
# for ddir in all_dirs:
# if ddir in processed_data_dirs: continue
# log.info(f"Refreshing: '{ddir}'")
# d = PrsData(ddir, file_mode="rw")
# refresh_data(d)
with Pool() as pool:
res += pool.starmap(refresh_data, [(PrsData(ddir, file_mode="rw"), plot_spectrum_kw) for ddir in all_dirs if not ddir in processed_data_dirs])
print(f"{'REPORT for refresh_all_datas':#^40}")
n_fail = 0
for i, (_, err) in enumerate(res):
if err is None: continue
d, ex, tb = err
print(f"\n\nFAILED DATA: {d}")
print(tb)
print(ex)
n_fail += 1
print(f"\n\n{n_fail} failures")
def _plot_modulus(datadir, kw):
try:
data = PrsData(datadir)
plot_modulus_spectrum(data, **kw)
return (datadir, None, None)
except Exception as e:
import traceback
tb = traceback.format_exc()
return (datadir, e, tb)
[docs]
def add_all_modulus_spectra(base_dir: str, **kw):
"""
For all datas in base_dir:
- plot modulus spectrum into data dir
"""
all_dirs = find_all_data_dirs(base_dir)
all_dirs.sort()
# 1) RTA datas
with Pool() as pool:
errs = pool.starmap(_plot_modulus, [(data_dir, kw) for data_dir in all_dirs])
n_fail = 0
for (d, e, tb) in errs:
if e is None: continue
print(f"\n\nFAILED DATA: {d}")
print(tb)
print(e)
n_fail += 1
print(f"\n\n{n_fail} failures")
def _plot_XY(datadir, kw):
try:
data = PrsData(datadir)
if len(data.wavelengths) > 100:
plot_XY(data, save_fig=True, **kw)
plt.close()
return (datadir, None, None)
except Exception as e:
import traceback
tb = traceback.format_exc()
return (datadir, e, tb)
[docs]
def add_all_XY_spectra(base_dir: str, **kw):
"""
For all datas in base_dir:
- plot modulus spectrum into data dir
"""
all_dirs = find_all_data_dirs(base_dir)
all_dirs.sort()
# 1) RTA datas
with Pool() as pool:
errs = pool.starmap(_plot_XY, [(data_dir, kw) for data_dir in all_dirs])
n_fail = 0
for (d, e, tb) in errs:
if e is None: continue
print(f"\n\nFAILED DATA: {d}")
print(tb)
print(e)
n_fail += 1
print(f"\n\n{n_fail} failures")
