"""
Convert the data from a directory to be compatible with a newer version of prsctrl
"""
import logging
log = logging.getLogger(__name__)
import os
import numpy as np
import re
from ..data.prsdata import ABSORPTION, PrsData, REFLECTION, TRANSMISSION
from ..data.plot import plot_data
from mqutil.dict_util import get_subdict
from ..version import version as prsctrl_current_version
[docs]
def convert_00_to_03(data: dict):
"""
0.0: wl keys are
raw = array with [dR, R]
theta - list
Convert data from prsctrl 0.0 to 0.3
"""
data["phase_offset_deg_before"] = 0.0 # avoid issues
data["R_offset_volt_before"] = 0.0
phase_offset = data["phase_offset_deg_before"]
R_offset = data["R_offset_volt_before"]
wavelengths = [key for key in data.keys() if type(key) == int]
metadata = {k:v for k,v in data.items() if type(k) == str}
new_data = {}
for wl in wavelengths:
new_data[wl] = {}
# convert phase
new_data[wl]["lock-in-theta_raw"] = np.array(data[wl]["theta_raw"]) + phase_offset
new_data[wl]["lock-in-R_raw"] = data[wl]["raw"][0,:] + R_offset # TODO: did it use RMS or PP?
new_data[wl]["lock-in-aux_raw"] = data[wl]["raw"][1,:]
new_prsdata = PrsData(data=new_data, metadata=metadata)
new_prsdata.get_spectrum_data()
return new_prsdata
[docs]
def convert_01_to_03(data: PrsData) -> PrsData:
"""
Convert data from prsctrl 0.1 to 0.3
"""
data.remove_calculated_values()
if not "phase_offset_deg_before" in data.metadata or data.metadata["phase_offset_deg_before"] is None:
data.metadata["phase_offset_deg_before"] = 0.0
data.metadata["phase_offset_deg_after"] = 0.0
phase_offset = data.metadata["phase_offset_deg_before"]
if not "R_offset_volt_before" in data.metadata or data.metadata["R_offset_volt_before"] is None:
data.metadata["R_offset_volt_before"] = 0.0
data.metadata["R_offset_volt_after"] = 0.0
R_offset = data.metadata["R_offset_volt_before"]
for wl in data.wavelengths:
# convert phase
data.data[wl]["lock-in-theta_raw"] = data.data[wl]["theta_raw"] + phase_offset
del data.data[wl]["theta_raw"]
data.data[wl]["lock-in-R_raw"] = data.data[wl]["dR_raw"] / (2*np.sqrt(2)) + R_offset
del data.data[wl]["dR_raw"]
data.data[wl]["lock-in-aux_raw"] = data.data[wl]["R_raw"]
del data.data[wl]["R_raw"]
# initialize all other fields
data.get_spectrum_data([wl])
return data
[docs]
def convert_02_to_03(data: PrsData) -> PrsData:
"""
Convert data from prsctrl 0.2 to 0.3
Offsets are no longer integrated into the raw values
"""
data.remove_calculated_values()
if not "phase_offset_deg_before" in data.metadata or data.metadata["phase_offset_deg_before"] is None:
data.metadata["phase_offset_deg_before"] = 0.0
data.metadata["phase_offset_deg_after"] = 0.0
phase_offset = data.metadata["phase_offset_deg_before"]
if not "R_offset_volt_before" in data.metadata or data.metadata["R_offset_volt_before"] is None:
data.metadata["R_offset_volt_before"] = 0.0
data.metadata["R_offset_volt_after"] = 0.0
R_offset = data.metadata["R_offset_volt_before"]
for wl in data.wavelengths:
# convert phase
data.data[wl]["lock-in-theta_raw"] = data.data[wl]["lock-in-theta_raw"] + phase_offset
data.data[wl]["lock-in-R_raw"] = data.data[wl]["lock-in-R_raw"] / (2*np.sqrt(2)) + R_offset
# initialize all other fields
data.get_spectrum_data([wl])
return data
[docs]
def convert_03_to_04(data: PrsData) -> PrsData:
"""
Convert data from prsctrl 0.3 to 0.4
Offsets are recorded same as any wavelength, and raw data are stored in the same data dictionary
"""
data.remove_calculated_values()
new_names = {"before": "begin", "after": "end"}
for offset in ["before", "after"]:
# get the offsets, use zero if none are present
if not f"phase_offset_deg_{offset}" in data.metadata or data.metadata[f"phase_offset_deg_{offset}"] is None:
data.metadata[f"phase_offset_deg_{offset}"] = 0.0
phase_offset = data.metadata[f"phase_offset_deg_{offset}"]
del data.metadata[f"phase_offset_deg_{offset}"]
if not f"R_offset_volt_{offset}" in data.metadata or data.metadata[f"R_offset_volt_{offset}"] is None:
data.metadata[f"R_offset_volt_{offset}"] = 0.0
R_offset = data.metadata[f"R_offset_volt_{offset}"]
del data.metadata[f"R_offset_volt_{offset}"]
DC_offset = 0.0
new_name = f"offset_{new_names[offset]}"
data.metadata[new_name] = {
"R": R_offset,
"theta": phase_offset,
}
data.data[new_name] = {}
data.data[new_name]["lock-in-theta_raw"] = np.array([phase_offset])
data.data[new_name]["lock-in-R_raw"] = np.array([R_offset])
data.data[new_name]["lock-in-aux_raw"] = np.array([DC_offset])
# initialize all other fields
for wl in data.data.keys(): # .wavelengths:
data.get_spectrum_data([wl])
return data
[docs]
def convert_04_to_05(data: PrsData) -> PrsData:
"""
Introduced the 'mode' in metadata which indicates the measurement mode (reflection or transmission)
"""
if not "mode" in data.metadata:
if "tra" in data.metadata["name"].lower():
data.metadata["mode"] = TRANSMISSION
else:
data.metadata["mode"] = REFLECTION
log.info(f"Assuming mode={data.metadata['mode']} for data with name '{data.metadata['name']}'")
return data
[docs]
def convert_05_to_06(data: PrsData) -> PrsData:
"""
Changed metadata/settings format:
Each device has its own settings dict
"""
md = data.metadata
k = "lock-in_settings"
if k in md:
md["lock-in-amp"] = md[k]
del md[k]
else:
log.info(f"Did not find {k}")
# pre-amp settings string
k = "amplifier_settings"
if k in md:
if data.mode == REFLECTION: # amp settings never recorded for transmission
# string like this 'Gain: 10^7, Mode: Low Noise, Coupling: DC'
match = re.match(r"Gain: 10\^(\d), Mode: ([^,]+), Coupling: ([A-Z]+)", md[k])
if match:
md["pre-amp"] = {
"gain": match.groups(0),
"mode": match.groups(1),
"coupling": match.groups(2),
}
else:
log.warning(f"Could not match '{md[k]}'")
del md[k]
else:
log.info(f"Did not find {k}")
k = "led"
if k in md:
if type(k) != dict: # led name, not settings dict
led = md[k]
del md[k]
ld = get_subdict(md, "led")
ld["name_led"] = led
k = "measurement_parameters"
if k in md:
mp = md[k]
# pre-amp settings in mp, prefer settings string over this
kk = "amplifier_gain"
if kk in mp:
amp = get_subdict(md, "pre-amp")
if not "gain" in amp:
amp["gain"] = mp[kk]
del mp[kk]
kk = "amplifier_coupling"
if kk in mp:
amp = get_subdict(md, "pre-amp")
if not "coupling" in amp and data.mode == REFLECTION:
amp["coupling"] = mp[kk]
del mp[kk]
kk = "monochromator_bandwidth_nm"
if kk in mp:
mc = get_subdict(md, "monochromator")
mc["bandwidth_nm"] = mp[kk]
del mp[kk]
kk = "led_current_A"
if kk in mp:
mc = get_subdict(md, "led")
mc["current_A"] = mp[kk]
del mp[kk]
# srat and f are now in lock-in settings
kk = "sample_rate_Hz"
if kk in mp:
mc = get_subdict(md, "lock-in-amp")
mc["sample_rate_Hz"] = mp[kk]
del mp[kk]
kk = "frequency_Hz"
if kk in mp:
mc = get_subdict(md, "lock-in-amp")
mc["frequency_Hz"] = mp[kk]
del mp[kk]
# pre-amp explicit metadata
k = "femto_gain_ref"
if k in md:
if data.mode == REFLECTION:
amp = get_subdict(md, "pre-amp")
amp["gain"] = md[k]
del md[k]
k = "femto_gain_tra"
if k in md:
if data.mode == TRANSMISSION:
amp = get_subdict(md, "pre-amp")
amp["gain"] = md[k]
del md[k]
# device names
k = "device_lock-in"
if k in md:
ld = get_subdict(md, "lock-in-amp")
ld["name"] = md[k]
del md[k]
k = "device_monochromator"
if k in md:
ld = get_subdict(md, "monochromator")
ld["name"] = md[k]
del md[k]
k = "amplifier"
if k in md:
ld = get_subdict(md, "pre-amp")
if data.mode == REFLECTION:
ld["name"] = md[k]
del md[k]
return data
[docs]
def convert_06_to_07(data: PrsData) -> PrsData:
"""
Renamed keys from R to I
"""
keys = list(data.data.keys())
for key in keys:
if key in PrsData.DEPRECATED_KEYS:
nkey = PrsData.DEPRECATED_KEYS[key]
data.data[nkey] = data.data[key]
del data.data[key]
return data
[docs]
def convert_07_to_08(data: PrsData) -> PrsData:
"""
Changed offset calculation
"""
data.remove_calculated_values()
return data
[docs]
def convert_08_to_09(data: PrsData) -> PrsData:
"""
Every measurement now needs a timestamp.
For old data, calculate timestamps between 0 and 1 by assuming each measurement took the same amount of time (reasonable assumption).
Also assumes wavelengths were measured in ascending order.
"""
def hacky_sort(val):
# old versions only used offset_begin and offset_end, so we map those to -1 and 10^6 and then just sort by wavelength
if type(val) == str:
if val == "offset_begin":
return -1.0
elif val == "offset_end":
return 1e6
else:
raise ValueError(f"Unexpected string key '{val}")
else:
return val
sorted_keys = sorted(data.data.keys(), key=hacky_sort)
if len(sorted_keys) == 1:
offset_times = [0]
else:
offset_times = [i/(len(sorted_keys)-1) for i in range(len(sorted_keys))]
for i, k in enumerate(sorted_keys):
data.data[k]["timestamp_start"] = offset_times[i]
return data
[docs]
def convert_09_to_10(data: PrsData) -> PrsData:
"""
Changed plot keys
"""
data.remove_calculated_values()
return data
[docs]
def convert_data(data_dir: str, out_dir_base:str, from_version=None, to_version=None, write_partials=False, refresh_if_current=False):
"""
Convert prsdata a newer prsctrl version.
:param data_dir: PrsData directory to load the data from
:param from_version: The version of the data to convert. Set to None for automatic detection
:param to_version: The version of the data to convert. If set to None, convert to latest version
:param out_dir_base: The base directory to save the converted data to. A new subdirectory for the data will be created
with the same name as data_dir
:param write_partials: Whether to write the PART files
"""
old_data = PrsData(data_path=data_dir)
try:
mode = old_data.mode
if mode == ABSORPTION:
log.info(f"Can not upgrade absorption data, upgrade reflection and transmission and recalculate absorption later")
return
except:
pass
if from_version is None:
if "prsctrl_version" in old_data.metadata:
from_version = old_data.metadata["prsctrl_version"]
else:
wls = old_data.wavelengths
wl_data = old_data.data[wls[0]]
if "dR_raw" in wl_data:
from_version = "0.1"
if not from_version:
raise RuntimeError(f"Could not determine version of data. Metadata keys: {old_data.metadata.keys()}")
max_upgrade_for_version = {
"0.0": ("0.3", convert_00_to_03),
"0.1": ("0.3", convert_01_to_03),
"0.2": ("0.3", convert_02_to_03),
"0.3": ("0.4", convert_03_to_04),
"0.4": ("0.5", convert_04_to_05),
"0.5": ("0.6", convert_05_to_06),
"0.6": ("0.7", convert_06_to_07),
"0.7": ("0.8", convert_07_to_08),
"0.8": ("0.9", convert_08_to_09),
"0.9": ("0.10", convert_09_to_10),
}
ddirname = os.path.basename(data_dir.removesuffix("/"))
data = old_data
if to_version is None:
to_version = prsctrl_current_version
original_version = from_version
if from_version == to_version: # refresh all calculated values
if refresh_if_current:
new_data = old_data
new_data.remove_calculated_values()
new_data.get_spectrum_data()
else:
log.info(f"d={ddirname}: Already at version {to_version}")
return
while from_version != to_version:
if not from_version in max_upgrade_for_version:
raise RuntimeError(f"Can not upgrade from version {from_version} to {to_version}")
new_version = max_upgrade_for_version[from_version][0]
log.info(f"d={ddirname}: Upgrading from version {from_version} to {new_version}")
data.metadata[f"prsctrl_version-converted-from-{from_version}-to-{new_version}"] = True
data = max_upgrade_for_version[from_version][1](data)
from_version = new_version
data.metadata["prsctrl_version"] = to_version
# set the new_data object up for writing
data.dirpath = os.path.join(out_dir_base, ddirname)
data.dirname = ddirname
data.write_mode = True
if os.path.exists(data.dirpath):
raise FileExistsError(f"Directory {data.dirname} already exists")
data._assert_directory_exists()
if write_partials:
# write and plot data
data.write_metadata()
for wl in data.wavelengths:
data.write_partial_file(wl)
data.write_full_file()
data.save_csv()
plot_data(data, data.dirname, data.dirpath, error_shade=True)
