from .prsdata import *
import itertools
[docs]
def list_to_ranges(l: list[float|int], step=1) -> list[float|int|tuple[float|int, float|int]]:
"""
Convert a list of integer values to a list of start and end points
Eg:
[1, 2, 3, 4, 5] -> [(1, 5)]
[1, 2, 3, 5, 7, 8] -> [(1, 3), 5, (7, 8)]
"""
ranges = []
range_start_val = None
for i in range(len(l)):
if i<len(l)-1 and l[i+1] == l[i] + step: # continuous value follows
if range_start_val is None:
range_start_val = l[i]
else:
pass
else:
if range_start_val is None: # separate value
ranges.append(l[i])
else: # end of range
ranges.append((range_start_val, l[i]))
range_start_val = None
return ranges
[docs]
def ranges_to_string(r: list[float|int|tuple[int|float,int|float]], sep=", "):
"""
Convert the result of list_to_ranges to a string:
Eg
[(1, 3), 5, (7, 8)] -> "1-3, 5, 7-8"
"""
s = ""
for v in r:
if type(v) == tuple:
s += f"{v[0]}-{v[1]}{sep}"
else:
s += f"{v}{sep}"
return s.removesuffix(sep)
[docs]
def sanity_check_data(data: PrsData) -> dict:
"""
Check the data for problematic measurements, like invalid, too small or too large DC values.
"""
wls = np.array(data.wavelengths)
msgs = {
"I_invalid": [], # DC value < 0 invalid
"I_large": [], # DC value larger than 10V (overload might have occured)
"I_small": [], # DC smaller than 0.020 (binning)
"offsets_mismatch": [],
"theta_large": [],
"offset_bad_phase": [],
}
# 1) Check I
try:
sdata = data.get_spectrum_data(columns=["I"])[0]
for i, wl in enumerate(wls):
val = sdata[i,0]
if val <= 0:
msgs["I_invalid"].append((wl, val))
elif val <= 0.020:
msgs["I_small"].append((wl, val))
elif val >= 10:
msgs["I_large"].append((wl, val))
except Exception as e:
log.debug(f"Get I data: {type(e)}: {e}")
# 2) Check Theta
# Commented out with 0.10.0:
# We now understand that large phases may occur when the rise and fall are exponentials
# and the time constants are not much smaller than the modulation frequency. However, this may not be problematic so there is little use inlcuding it in the report.
# try:
# sdata = data.get_spectrum_data(columns=["theta"])[0]
# for i, wl in enumerate(wls):
# theta = sdata[i,0]
# if 60 < np.abs(theta) % 180 < 120:
# msgs["theta_large"].append((wl, theta))
# except Exception as e:
# log.debug(f"Get theta data: {type(e)}: {e}")
# 3) Check offsets
offsets = [k for k in data.data.keys() if type(k) == str and k.startswith("offset")]
# check if any of the offsets are not within the uncertainty of one another
for o1, o2 in itertools.combinations(offsets, 2):
U1, sU1 = data.get_spectrum_data(wavelengths=[o1], columns=["dI", "sdI"])[0][0,:]
U2, sU2 = data.get_spectrum_data(wavelengths=[o2], columns=["dI", "sdI"])[0][0,:]
log.debug(f"Checking offsets: {o1}={U1}pm{sU1}, {o2}={U2}pm{sU2}")
# U1 = data[o1]["dI"]
# sU1 = data[o1]["sdI"]
# U2 = data[o2]["dI"]
# sU2 = data[o2]["sdI"]
if U1-sU1 > U2+sU2 or U1+sU1 < U2-sU2:
msgs["offsets_mismatch"].append((o1, U1, sU1, o2, U2, sU2))
for o in offsets:
stheta = data.get_for_wl(o, "stheta")
if stheta >= 1.0:
msgs["offset_bad_phase"].append((o, stheta))
return msgs
[docs]
def sanity_check_data_gen_report(data: PrsData, msgs:dict|None=None) -> str:
"""
Generate a report of all the problematic values/measurements of a data object.
:return: Report as string in markdown format
"""
report = ""
if len(data.wavelengths) >= 2:
wl_step = data.wavelengths[1] - data.wavelengths[0]
else:
wl_step = 1
formatters = {
# "I_invalid": lambda msg_a: "## DC Values < 0 V (invalid):\n\n" + "\n".join([f"- wl={msg[0]:4} nm: val={msg[1]:7.4f} V" for msg in msg_a]),
# "I_large": lambda msg_a: "## DC Values > 10 V (might overload):\n\n" + "\n".join([f"- wl={msg[0]:4} nm: val={msg[1]:7.4f} V" for msg in msg_a]),
# "I_small": lambda msg_a: "## DC Values < 20 mV (might be binning):\n\n" + "\n".join([f"- wl={msg[0]:4} nm: val={msg[1]:7.4f} V" for msg in msg_a]),
"I_invalid": lambda msg_a: f"## DC Values < 0 V (invalid): {len(msg_a)}\n\nAt wavelengths (nm): {ranges_to_string(list_to_ranges([msg[0] for msg in msg_a], wl_step))}",
"I_large": lambda msg_a: f"## DC Values > 10 V (might overload): {len(msg_a)}\n\nAt wavelengths (nm): {ranges_to_string(list_to_ranges([msg[0] for msg in msg_a], wl_step))}",
"I_small": lambda msg_a: f"## DC Values < 20 mV (might be binning): {len(msg_a)}\n\nAt wavelengths (nm): {ranges_to_string(list_to_ranges([msg[0] for msg in msg_a], wl_step))}",
"offsets_mismatch": lambda msg_a: f"## Mismatching offsets (might be drift during measurment): {len(msg_a)}\n\n" + "\n".join([f"- (offset={msg[0]}: MAG={msg[1]:10.4e} pm {msg[2]:10.4e} V) and (offset={msg[3]}: MAG={msg[4]:10.4e} pm {msg[5]:10.4e} V)" for msg in msg_a]),
"offset_bad_phase": lambda msg_a: f"## Bad Offset Theta sigma(theta) >= 1° (Lock-In might not lock onto offset)\n\n: " + "\n".join([f"- (offset={msg[0]}: sigma(theta)={msg[1]}°" for msg in msg_a]),
"theta_large": lambda msg_a: f"## Theta > 60 (frequency might be too fast): {len(msg_a)}\n\nAt wavelengths (nm): {ranges_to_string(list_to_ranges([msg[0] for msg in msg_a], wl_step))}",
}
if msgs is None:
msgs = sanity_check_data(data)
for k, vals in msgs.items():
if vals:
report += formatters[k](vals) + "\n\n"
if report:
header = f"# Report for {data.dirname}\n\n"
# if data.dirname: header += f"- Directory: {data.dirname}\n"
if data.mode: header += f"- Measurement mode: {data.mode}\n"
if "sample" in data.metadata: header += f"- Sample: {data.metadata['sample']}\n"
return header + "\n" + report.strip("\n")
return ""
