zebra/pyzebra
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Utility renames and hdf_param_study simplification

Browse Source
This commit is contained in:
usov_i 2022-01-27 16:11:52 +01:00
parent 10bcabd7f9
commit bcd594fa7e
8 changed files with 286 additions and 282 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
pyzebra/app/panel_ccl_compare.py
View File

@ -72,8 +72,8 @@ for (let i = 0; i < js_data.data['fname'].length; i++) {
def create():
doc = curdoc()
det_data1 = []
det_data2 = []
dataset1 = []
dataset2 = []
fit_params = {}
js_data = ColumnDataSource(data=dict(content=["", ""], fname=["", ""], ext=["", ""]))
@ -99,17 +99,17 @@ def create():
proposal_textinput.on_change("name", proposal_textinput_callback)
def _init_datatable():
# det_data2 should have the same metadata to det_data1
scan_list = [s["idx"] for s in det_data1]
hkl = [f'{s["h"]} {s["k"]} {s["l"]}' for s in det_data1]
export = [s["export"] for s in det_data1]
# dataset2 should have the same metadata as dataset1
scan_list = [s["idx"] for s in dataset1]
hkl = [f'{s["h"]} {s["k"]} {s["l"]}' for s in dataset1]
export = [s["export"] for s in dataset1]
twotheta = [np.median(s["twotheta"]) if "twotheta" in s else None for s in det_data1]
gamma = [np.median(s["gamma"]) if "gamma" in s else None for s in det_data1]
omega = [np.median(s["omega"]) if "omega" in s else None for s in det_data1]
chi = [np.median(s["chi"]) if "chi" in s else None for s in det_data1]
phi = [np.median(s["phi"]) if "phi" in s else None for s in det_data1]
nu = [np.median(s["nu"]) if "nu" in s else None for s in det_data1]
twotheta = [np.median(s["twotheta"]) if "twotheta" in s else None for s in dataset1]
gamma = [np.median(s["gamma"]) if "gamma" in s else None for s in dataset1]
omega = [np.median(s["omega"]) if "omega" in s else None for s in dataset1]
chi = [np.median(s["chi"]) if "chi" in s else None for s in dataset1]
phi = [np.median(s["phi"]) if "phi" in s else None for s in dataset1]
nu = [np.median(s["nu"]) if "nu" in s else None for s in dataset1]
scan_table_source.data.update(
scan=scan_list,
@ -163,9 +163,9 @@ def create():
new_data1 = new_data1[:min_len]
new_data2 = new_data2[:min_len]
nonlocal det_data1, det_data2
det_data1 = new_data1
det_data2 = new_data2
nonlocal dataset1, dataset2
dataset1 = new_data1
dataset2 = new_data2
_init_datatable()
file_open_button = Button(label="Open New", width=100, disabled=True)
@ -201,9 +201,9 @@ def create():
new_data1 = new_data1[:min_len]
new_data2 = new_data2[:min_len]
nonlocal det_data1, det_data2
det_data1 = new_data1
det_data2 = new_data2
nonlocal dataset1, dataset2
dataset1 = new_data1
dataset2 = new_data2
_init_datatable()
upload_div = Div(text="or upload 2 .ccl files:", margin=(5, 5, 0, 5))
@ -213,17 +213,17 @@ def create():
upload_button.on_change("filename", upload_button_callback)
def monitor_spinner_callback(_attr, old, new):
if det_data1 and det_data2:
pyzebra.normalize_dataset(det_data1, new)
pyzebra.normalize_dataset(det_data2, new)
if dataset1 and dataset2:
pyzebra.normalize_dataset(dataset1, new)
pyzebra.normalize_dataset(dataset2, new)
_update_plot()
monitor_spinner = Spinner(title="Monitor:", mode="int", value=100_000, low=1, width=145)
monitor_spinner.on_change("value", monitor_spinner_callback)
def _update_table():
fit_ok = [(1 if "fit" in scan else 0) for scan in det_data1]
export = [scan["export"] for scan in det_data1]
fit_ok = [(1 if "fit" in scan else 0) for scan in dataset1]
export = [scan["export"] for scan in dataset1]
scan_table_source.data.update(fit=fit_ok, export=export)
def _update_plot():
@ -382,7 +382,7 @@ def create():
def scan_table_source_callback(_attr, _old, new):
# unfortunately, we don't know if the change comes from data update or user input
# also `old` and `new` are the same for non-scalars
for scan1, scan2, export in zip(det_data1, det_data2, new["export"]):
for scan1, scan2, export in zip(dataset1, dataset2, new["export"]):
scan1["export"] = export
scan2["export"] = export
_update_preview()
@ -426,14 +426,14 @@ def create():
def _get_selected_scan():
ind = scan_table_source.selected.indices[0]
return det_data1[ind], det_data2[ind]
return dataset1[ind], dataset2[ind]
merge_from_select = Select(title="scan:", width=145)
def merge_button_callback():
scan_into1, scan_into2 = _get_selected_scan()
scan_from1 = det_data1[int(merge_from_select.value)]
scan_from2 = det_data2[int(merge_from_select.value)]
scan_from1 = dataset1[int(merge_from_select.value)]
scan_from2 = dataset2[int(merge_from_select.value)]
if scan_into1 is scan_from1:
print("WARNING: Selected scans for merging are identical")
@ -577,7 +577,7 @@ def create():
fit_output_textinput = TextAreaInput(title="Fit results:", width=750, height=200)
def proc_all_button_callback():
for scan in [*det_data1, *det_data2]:
for scan in [*dataset1, *dataset2]:
if scan["export"]:
pyzebra.fit_scan(
scan, fit_params, fit_from=fit_from_spinner.value, fit_to=fit_to_spinner.value
@ -628,7 +628,7 @@ def create():
temp_file = temp_dir + "/temp"
export_data1 = []
export_data2 = []
for scan1, scan2 in zip(det_data1, det_data2):
for scan1, scan2 in zip(dataset1, dataset2):
if scan1["export"]:
export_data1.append(scan1)
export_data2.append(scan2)

50
pyzebra/app/panel_ccl_integrate.py
View File

@ -72,7 +72,7 @@ for (let i = 0; i < js_data.data['fname'].length; i++) {
def create():
doc = curdoc()
det_data = []
dataset = []
fit_params = {}
js_data = ColumnDataSource(data=dict(content=["", ""], fname=["", ""], ext=["", ""]))
@ -100,16 +100,16 @@ def create():
proposal_textinput.on_change("name", proposal_textinput_callback)
def _init_datatable():
scan_list = [s["idx"] for s in det_data]
hkl = [f'{s["h"]} {s["k"]} {s["l"]}' for s in det_data]
export = [s["export"] for s in det_data]
scan_list = [s["idx"] for s in dataset]
hkl = [f'{s["h"]} {s["k"]} {s["l"]}' for s in dataset]
export = [s["export"] for s in dataset]
twotheta = [np.median(s["twotheta"]) if "twotheta" in s else None for s in det_data]
gamma = [np.median(s["gamma"]) if "gamma" in s else None for s in det_data]
omega = [np.median(s["omega"]) if "omega" in s else None for s in det_data]
chi = [np.median(s["chi"]) if "chi" in s else None for s in det_data]
phi = [np.median(s["phi"]) if "phi" in s else None for s in det_data]
nu = [np.median(s["nu"]) if "nu" in s else None for s in det_data]
twotheta = [np.median(s["twotheta"]) if "twotheta" in s else None for s in dataset]
gamma = [np.median(s["gamma"]) if "gamma" in s else None for s in dataset]
omega = [np.median(s["omega"]) if "omega" in s else None for s in dataset]
chi = [np.median(s["chi"]) if "chi" in s else None for s in dataset]
phi = [np.median(s["phi"]) if "phi" in s else None for s in dataset]
nu = [np.median(s["nu"]) if "nu" in s else None for s in dataset]
scan_table_source.data.update(
scan=scan_list,
@ -133,7 +133,7 @@ def create():
file_select = MultiSelect(title="Available .ccl/.dat files:", width=210, height=250)
def file_open_button_callback():
nonlocal det_data
nonlocal dataset
new_data = []
for f_path in file_select.value:
with open(f_path) as file:
@ -155,7 +155,7 @@ def create():
pyzebra.merge_datasets(new_data, file_data)
if new_data:
det_data = new_data
dataset = new_data
_init_datatable()
append_upload_button.disabled = False
@ -175,7 +175,7 @@ def create():
continue
pyzebra.normalize_dataset(file_data, monitor_spinner.value)
pyzebra.merge_datasets(det_data, file_data)
pyzebra.merge_datasets(dataset, file_data)
if file_data:
_init_datatable()
@ -184,7 +184,7 @@ def create():
file_append_button.on_click(file_append_button_callback)
def upload_button_callback(_attr, _old, _new):
nonlocal det_data
nonlocal dataset
new_data = []
for f_str, f_name in zip(upload_button.value, upload_button.filename):
with io.StringIO(base64.b64decode(f_str).decode()) as file:
@ -205,7 +205,7 @@ def create():
pyzebra.merge_datasets(new_data, file_data)
if new_data:
det_data = new_data
dataset = new_data
_init_datatable()
append_upload_button.disabled = False
@ -227,7 +227,7 @@ def create():
continue
pyzebra.normalize_dataset(file_data, monitor_spinner.value)
pyzebra.merge_datasets(det_data, file_data)
pyzebra.merge_datasets(dataset, file_data)
if file_data:
_init_datatable()
@ -239,16 +239,16 @@ def create():
append_upload_button.on_change("filename", append_upload_button_callback)
def monitor_spinner_callback(_attr, old, new):
if det_data:
pyzebra.normalize_dataset(det_data, new)
if dataset:
pyzebra.normalize_dataset(dataset, new)
_update_plot()
monitor_spinner = Spinner(title="Monitor:", mode="int", value=100_000, low=1, width=145)
monitor_spinner.on_change("value", monitor_spinner_callback)
def _update_table():
fit_ok = [(1 if "fit" in scan else 0) for scan in det_data]
export = [scan["export"] for scan in det_data]
fit_ok = [(1 if "fit" in scan else 0) for scan in dataset]
export = [scan["export"] for scan in dataset]
scan_table_source.data.update(fit=fit_ok, export=export)
def _update_plot():
@ -368,7 +368,7 @@ def create():
def scan_table_source_callback(_attr, _old, new):
# unfortunately, we don't know if the change comes from data update or user input
# also `old` and `new` are the same for non-scalars
for scan, export in zip(det_data, new["export"]):
for scan, export in zip(dataset, new["export"]):
scan["export"] = export
_update_preview()
@ -410,13 +410,13 @@ def create():
)
def _get_selected_scan():
return det_data[scan_table_source.selected.indices[0]]
return dataset[scan_table_source.selected.indices[0]]
merge_from_select = Select(title="scan:", width=145)
def merge_button_callback():
scan_into = _get_selected_scan()
scan_from = det_data[int(merge_from_select.value)]
scan_from = dataset[int(merge_from_select.value)]
if scan_into is scan_from:
print("WARNING: Selected scans for merging are identical")
@ -557,7 +557,7 @@ def create():
fit_output_textinput = TextAreaInput(title="Fit results:", width=750, height=200)
def proc_all_button_callback():
for scan in det_data:
for scan in dataset:
if scan["export"]:
pyzebra.fit_scan(
scan, fit_params, fit_from=fit_from_spinner.value, fit_to=fit_to_spinner.value
@ -602,7 +602,7 @@ def create():
with tempfile.TemporaryDirectory() as temp_dir:
temp_file = temp_dir + "/temp"
export_data = []
for scan in det_data:
for scan in dataset:
if scan["export"]:
export_data.append(scan)

60
pyzebra/app/panel_hdf_param_study.py
View File

@ -48,8 +48,7 @@ IMAGE_PLOT_H = int(IMAGE_H * 2) + 27
def create():
doc = curdoc()
zebra_data = []
det_data = {}
dataset = []
cami_meta = {}
num_formatter = NumberFormatter(format="0.00", nan_format="")
@ -108,15 +107,15 @@ def create():
def _init_datatable():
file_list = []
for scan in zebra_data:
for scan in dataset:
file_list.append(os.path.basename(scan["original_filename"]))
scan_table_source.data.update(
file=file_list,
param=[None] * len(zebra_data),
frame=[None] * len(zebra_data),
x_pos=[None] * len(zebra_data),
y_pos=[None] * len(zebra_data),
param=[None] * len(dataset),
frame=[None] * len(dataset),
x_pos=[None] * len(dataset),
y_pos=[None] * len(dataset),
)
scan_table_source.selected.indices = []
scan_table_source.selected.indices = [0]
@ -127,7 +126,7 @@ def create():
frame = []
x_pos = []
y_pos = []
for scan in zebra_data:
for scan in dataset:
if "fit" in scan:
framei = scan["fit"]["frame"]
x_posi = scan["fit"]["x_pos"]
@ -150,13 +149,13 @@ def create():
print("Could not read data from the file.")
return
zebra_data.extend(new_data)
dataset.extend(new_data)
_init_datatable()
def file_open_button_callback():
nonlocal zebra_data
zebra_data = []
nonlocal dataset
dataset = []
_file_open()
file_open_button = Button(label="Open New", width=100)
@ -170,8 +169,6 @@ def create():
# Scan select
def scan_table_select_callback(_attr, old, new):
nonlocal det_data
if not new:
# skip empty selections
return
@ -186,21 +183,21 @@ def create():
# skip unnecessary update caused by selection drop
return
det_data = zebra_data[new[0]]
scan = dataset[new[0]]
zebra_mode = det_data["zebra_mode"]
zebra_mode = scan["zebra_mode"]
if zebra_mode == "nb":
metadata_table_source.data.update(geom=["normal beam"])
else: # zebra_mode == "bi"
metadata_table_source.data.update(geom=["bisecting"])
if "mf" in det_data:
metadata_table_source.data.update(mf=[det_data["mf"][0]])
if "mf" in scan:
metadata_table_source.data.update(mf=[scan["mf"][0]])
else:
metadata_table_source.data.update(mf=[None])
if "temp" in det_data:
metadata_table_source.data.update(temp=[det_data["temp"][0]])
if "temp" in scan:
metadata_table_source.data.update(temp=[scan["temp"][0]])
else:
metadata_table_source.data.update(temp=[None])
@ -240,12 +237,15 @@ def create():
autosize_mode="none",
)
def _get_selected_scan():
return dataset[scan_table_source.selected.indices[0]]
def param_select_callback(_attr, _old, new):
if new == "user defined":
param = [None] * len(zebra_data)
param = [None] * len(dataset)
else:
# TODO: which value to take?
param = [scan[new][0] for scan in zebra_data]
param = [scan[new][0] for scan in dataset]
scan_table_source.data["param"] = param
_update_param_plot()
@ -259,7 +259,8 @@ def create():
param_select.on_change("value", param_select_callback)
def update_overview_plot():
h5_data = det_data["data"]
scan = _get_selected_scan()
h5_data = scan["data"]
n_im, n_y, n_x = h5_data.shape
overview_x = np.mean(h5_data, axis=1)
overview_y = np.mean(h5_data, axis=2)
@ -290,10 +291,10 @@ def create():
frame_range.reset_end = n_im
frame_range.bounds = (0, n_im)
scan_motor = det_data["scan_motor"]
scan_motor = scan["scan_motor"]
overview_plot_y.axis[1].axis_label = f"Scanning motor, {scan_motor}"
var = det_data[scan_motor]
var = scan[scan_motor]
var_start = var[0]
var_end = var[-1] + (var[-1] - var[0]) / (n_im - 1)
@ -470,7 +471,7 @@ def create():
x = []
y = []
fit_param = fit_param_select.value
for s, p in zip(zebra_data, scan_table_source.data["param"]):
for s, p in zip(dataset, scan_table_source.data["param"]):
if "fit" in s and fit_param:
x.append(p)
y.append(s["fit"][fit_param])
@ -498,7 +499,7 @@ def create():
fit_param_select.on_change("value", fit_param_select_callback)
def proc_all_button_callback():
for scan in zebra_data:
for scan in dataset:
pyzebra.fit_event(
scan,
int(np.floor(frame_range.start)),
@ -511,7 +512,7 @@ def create():
_update_table()
for scan in zebra_data:
for scan in dataset:
if "fit" in scan:
options = list(scan["fit"].keys())
fit_param_select.options = options
@ -524,8 +525,9 @@ def create():
proc_all_button.on_click(proc_all_button_callback)
def proc_button_callback():
scan = _get_selected_scan()
pyzebra.fit_event(
det_data,
scan,
int(np.floor(frame_range.start)),
int(np.ceil(frame_range.end)),
int(np.floor(det_y_range.start)),
@ -536,7 +538,7 @@ def create():
_update_table()
for scan in zebra_data:
for scan in dataset:
if "fit" in scan:
options = list(scan["fit"].keys())
fit_param_select.options = options

114
pyzebra/app/panel_hdf_viewer.py
View File

@ -52,7 +52,7 @@ IMAGE_PLOT_H = int(IMAGE_H * 2) + 27
def create():
doc = curdoc()
det_data = {}
scan = {}
cami_meta = {}
num_formatter = NumberFormatter(format="0.00", nan_format="")
@ -108,14 +108,14 @@ def create():
upload_cami_button.on_change("value", upload_cami_button_callback)
def _file_open(file, cami_meta):
nonlocal det_data
nonlocal scan
try:
det_data = pyzebra.read_detector_data(file, cami_meta)
scan = pyzebra.read_detector_data(file, cami_meta)
except KeyError:
print("Could not read data from the file.")
return
last_im_index = det_data["data"].shape[0] - 1
last_im_index = scan["data"].shape[0] - 1
index_spinner.value = 0
index_spinner.high = last_im_index
@ -125,7 +125,7 @@ def create():
index_slider.disabled = False
index_slider.end = last_im_index
zebra_mode = det_data["zebra_mode"]
zebra_mode = scan["zebra_mode"]
if zebra_mode == "nb":
metadata_table_source.data.update(geom=["normal beam"])
else: # zebra_mode == "bi"
@ -157,7 +157,7 @@ def create():
if index is None:
index = index_spinner.value
current_image = det_data["data"][index]
current_image = scan["data"][index]
proj_v_line_source.data.update(
x=np.arange(0, IMAGE_W) + 0.5, y=np.mean(current_image, axis=0)
)
@ -180,24 +180,24 @@ def create():
image_glyph.color_mapper.low = im_min
image_glyph.color_mapper.high = im_max
if "mf" in det_data:
metadata_table_source.data.update(mf=[det_data["mf"][index]])
if "mf" in scan:
metadata_table_source.data.update(mf=[scan["mf"][index]])
else:
metadata_table_source.data.update(mf=[None])
if "temp" in det_data:
metadata_table_source.data.update(temp=[det_data["temp"][index]])
if "temp" in scan:
metadata_table_source.data.update(temp=[scan["temp"][index]])
else:
metadata_table_source.data.update(temp=[None])
gamma, nu = calculate_pol(det_data, index)
omega = np.ones((IMAGE_H, IMAGE_W)) * det_data["omega"][index]
gamma, nu = calculate_pol(scan, index)
omega = np.ones((IMAGE_H, IMAGE_W)) * scan["omega"][index]
image_source.data.update(gamma=[gamma], nu=[nu], omega=[omega])
# update detector center angles
det_c_x = int(IMAGE_W / 2)
det_c_y = int(IMAGE_H / 2)
if det_data["zebra_mode"] == "nb":
if scan["zebra_mode"] == "nb":
gamma_c = gamma[det_c_y, det_c_x]
nu_c = nu[det_c_y, det_c_x]
omega_c = omega[det_c_y, det_c_x]
@ -205,13 +205,13 @@ def create():
phi_c = None
else: # zebra_mode == "bi"
wave = det_data["wave"]
ddist = det_data["ddist"]
gammad = det_data["gamma"][index]
om = det_data["omega"][index]
ch = det_data["chi"][index]
ph = det_data["phi"][index]
nud = det_data["nu"]
wave = scan["wave"]
ddist = scan["ddist"]
gammad = scan["gamma"][index]
om = scan["omega"][index]
ch = scan["chi"][index]
ph = scan["phi"][index]
nud = scan["nu"]
nu_c = 0
chi_c, phi_c, gamma_c, omega_c = pyzebra.ang_proc(
@ -223,7 +223,7 @@ def create():
)
def update_overview_plot():
h5_data = det_data["data"]
h5_data = scan["data"]
n_im, n_y, n_x = h5_data.shape
overview_x = np.mean(h5_data, axis=1)
overview_y = np.mean(h5_data, axis=2)
@ -254,10 +254,10 @@ def create():
frame_range.reset_end = n_im
frame_range.bounds = (0, n_im)
scan_motor = det_data["scan_motor"]
scan_motor = scan["scan_motor"]
overview_plot_y.axis[1].axis_label = f"Scanning motor, {scan_motor}"
var = det_data[scan_motor]
var = scan[scan_motor]
var_start = var[0]
var_end = var[-1] + (var[-1] - var[0]) / (n_im - 1) if n_im != 1 else var_start + 1
@ -374,9 +374,9 @@ def create():
# calculate hkl-indices of first mouse entry
def mouse_enter_callback(_event):
if det_data and np.array_equal(image_source.data["h"][0], np.zeros((1, 1))):
if scan and np.array_equal(image_source.data["h"][0], np.zeros((1, 1))):
index = index_spinner.value
h, k, l = calculate_hkl(det_data, index)
h, k, l = calculate_hkl(scan, index)
image_source.data.update(h=[h], k=[k], l=[l])
plot.on_event(MouseEnter, mouse_enter_callback)
@ -438,7 +438,7 @@ def create():
def box_edit_callback(_attr, _old, new):
if new["x"]:
h5_data = det_data["data"]
h5_data = scan["data"]
x_val = np.arange(h5_data.shape[0])
left = int(np.floor(new["x"][0]))
right = int(np.ceil(new["x"][0] + new["width"][0]))
@ -739,7 +739,7 @@ def create():
def add_event_button_callback():
pyzebra.fit_event(
det_data,
scan,
int(np.floor(frame_range.start)),
int(np.ceil(frame_range.end)),
int(np.floor(det_y_range.start)),
@ -748,21 +748,21 @@ def create():
int(np.ceil(det_x_range.end)),
)
wave = det_data["wave"]
ddist = det_data["ddist"]
cell = det_data["cell"]
wave = scan["wave"]
ddist = scan["ddist"]
cell = scan["cell"]
gamma = det_data["gamma"][0]
omega = det_data["omega"][0]
nu = det_data["nu"][0]
chi = det_data["chi"][0]
phi = det_data["phi"][0]
gamma = scan["gamma"][0]
omega = scan["omega"][0]
nu = scan["nu"][0]
chi = scan["chi"][0]
phi = scan["phi"][0]
scan_motor = det_data["scan_motor"]
var_angle = det_data[scan_motor]
scan_motor = scan["scan_motor"]
var_angle = scan[scan_motor]
snr_cnts = det_data["fit"]["snr"]
frC = det_data["fit"]["frame"]
snr_cnts = scan["fit"]["snr"]
frC = scan["fit"]["frame"]
var_F = var_angle[int(np.floor(frC))]
var_C = var_angle[int(np.ceil(frC))]
@ -781,11 +781,11 @@ def create():
elif scan_motor == "phi":
phi = var_p
intensity = det_data["fit"]["intensity"]
x_pos = det_data["fit"]["x_pos"]
y_pos = det_data["fit"]["y_pos"]
intensity = scan["fit"]["intensity"]
x_pos = scan["fit"]["x_pos"]
y_pos = scan["fit"]["y_pos"]
if det_data["zebra_mode"] == "nb":
if scan["zebra_mode"] == "nb":
chi = None
phi = None
@ -883,22 +883,22 @@ def create():
return Panel(child=tab_layout, title="hdf viewer")
def calculate_hkl(det_data, index):
def calculate_hkl(scan, index):
h = np.empty(shape=(IMAGE_H, IMAGE_W))
k = np.empty(shape=(IMAGE_H, IMAGE_W))
l = np.empty(shape=(IMAGE_H, IMAGE_W))
wave = det_data["wave"]
ddist = det_data["ddist"]
gammad = det_data["gamma"][index]
om = det_data["omega"][index]
nud = det_data["nu"]
ub = det_data["ub"]
geometry = det_data["zebra_mode"]
wave = scan["wave"]
ddist = scan["ddist"]
gammad = scan["gamma"][index]
om = scan["omega"][index]
nud = scan["nu"]
ub = scan["ub"]
geometry = scan["zebra_mode"]
if geometry == "bi":
chi = det_data["chi"][index]
phi = det_data["phi"][index]
chi = scan["chi"][index]
phi = scan["phi"][index]
elif geometry == "nb":
chi = 0
phi = 0
@ -914,10 +914,10 @@ def calculate_hkl(det_data, index):
return h, k, l
def calculate_pol(det_data, index):
ddist = det_data["ddist"]
gammad = det_data["gamma"][index]
nud = det_data["nu"]
def calculate_pol(scan, index):
ddist = scan["ddist"]
gammad = scan["gamma"][index]
nud = scan["nu"]
yi, xi = np.ogrid[:IMAGE_H, :IMAGE_W]
gamma, nu = pyzebra.det2pol(ddist, gammad, nud, xi, yi)

66
pyzebra/app/panel_param_study.py
View File

@ -83,7 +83,7 @@ def color_palette(n_colors):
def create():
doc = curdoc()
det_data = []
dataset = []
fit_params = {}
js_data = ColumnDataSource(data=dict(content=[""], fname=[""], ext=[""]))
@ -111,15 +111,15 @@ def create():
proposal_textinput.on_change("name", proposal_textinput_callback)
def _init_datatable():
scan_list = [s["idx"] for s in det_data]
export = [s["export"] for s in det_data]
scan_list = [s["idx"] for s in dataset]
export = [s["export"] for s in dataset]
if param_select.value == "user defined":
param = [None] * len(det_data)
param = [None] * len(dataset)
else:
param = [scan[param_select.value] for scan in det_data]
param = [scan[param_select.value] for scan in dataset]
file_list = []
for scan in det_data:
for scan in dataset:
file_list.append(os.path.basename(scan["original_filename"]))
scan_table_source.data.update(
@ -128,8 +128,8 @@ def create():
scan_table_source.selected.indices = []
scan_table_source.selected.indices = [0]
scan_motor_select.options = det_data[0]["scan_motors"]
scan_motor_select.value = det_data[0]["scan_motor"]
scan_motor_select.options = dataset[0]["scan_motors"]
scan_motor_select.value = dataset[0]["scan_motor"]
merge_options = [(str(i), f"{i} ({idx})") for i, idx in enumerate(scan_list)]
merge_from_select.options = merge_options
@ -138,7 +138,7 @@ def create():
file_select = MultiSelect(title="Available .ccl/.dat files:", width=210, height=250)
def file_open_button_callback():
nonlocal det_data
nonlocal dataset
new_data = []
for f_path in file_select.value:
with open(f_path) as file:
@ -160,7 +160,7 @@ def create():
pyzebra.merge_datasets(new_data, file_data)
if new_data:
det_data = new_data
dataset = new_data
_init_datatable()
append_upload_button.disabled = False
@ -180,7 +180,7 @@ def create():
continue
pyzebra.normalize_dataset(file_data, monitor_spinner.value)
pyzebra.merge_datasets(det_data, file_data)
pyzebra.merge_datasets(dataset, file_data)
if file_data:
_init_datatable()
@ -189,7 +189,7 @@ def create():
file_append_button.on_click(file_append_button_callback)
def upload_button_callback(_attr, _old, _new):
nonlocal det_data
nonlocal dataset
new_data = []
for f_str, f_name in zip(upload_button.value, upload_button.filename):
with io.StringIO(base64.b64decode(f_str).decode()) as file:
@ -210,7 +210,7 @@ def create():
pyzebra.merge_datasets(new_data, file_data)
if new_data:
det_data = new_data
dataset = new_data
_init_datatable()
append_upload_button.disabled = False
@ -232,7 +232,7 @@ def create():
continue
pyzebra.normalize_dataset(file_data, monitor_spinner.value)
pyzebra.merge_datasets(det_data, file_data)
pyzebra.merge_datasets(dataset, file_data)
if file_data:
_init_datatable()
@ -244,8 +244,8 @@ def create():
append_upload_button.on_change("filename", append_upload_button_callback)
def monitor_spinner_callback(_attr, _old, new):
if det_data:
pyzebra.normalize_dataset(det_data, new)
if dataset:
pyzebra.normalize_dataset(dataset, new)
_update_single_scan_plot()
_update_overview()
@ -253,8 +253,8 @@ def create():
monitor_spinner.on_change("value", monitor_spinner_callback)
def scan_motor_select_callback(_attr, _old, new):
if det_data:
for scan in det_data:
if dataset:
for scan in dataset:
scan["scan_motor"] = new
_update_single_scan_plot()
_update_overview()
@ -263,12 +263,12 @@ def create():
scan_motor_select.on_change("value", scan_motor_select_callback)
def _update_table():
fit_ok = [(1 if "fit" in scan else 0) for scan in det_data]
export = [scan["export"] for scan in det_data]
fit_ok = [(1 if "fit" in scan else 0) for scan in dataset]
export = [scan["export"] for scan in dataset]
if param_select.value == "user defined":
param = [None] * len(det_data)
param = [None] * len(dataset)
else:
param = [scan[param_select.value] for scan in det_data]
param = [scan[param_select.value] for scan in dataset]
scan_table_source.data.update(fit=fit_ok, export=export, param=param)
@ -322,7 +322,7 @@ def create():
par = []
for s, p in enumerate(scan_table_source.data["param"]):
if p is not None:
scan = det_data[s]
scan = dataset[s]
scan_motor = scan["scan_motor"]
xs.append(scan[scan_motor])
x.extend(scan[scan_motor])
@ -331,8 +331,8 @@ def create():
param.append(float(p))
par.extend(scan["counts"])
if det_data:
scan_motor = det_data[0]["scan_motor"]
if dataset:
scan_motor = dataset[0]["scan_motor"]
ov_plot.axis[0].axis_label = scan_motor
ov_param_plot.axis[0].axis_label = scan_motor
@ -371,7 +371,7 @@ def create():
y_lower = []
y_upper = []
fit_param = fit_param_select.value
for s, p in zip(det_data, scan_table_source.data["param"]):
for s, p in zip(dataset, scan_table_source.data["param"]):
if "fit" in s and fit_param:
x.append(p)
param_fit_val = s["fit"].params[fit_param].value
@ -534,7 +534,7 @@ def create():
def scan_table_source_callback(_attr, _old, new):
# unfortunately, we don't know if the change comes from data update or user input
# also `old` and `new` are the same for non-scalars
for scan, export in zip(det_data, new["export"]):
for scan, export in zip(dataset, new["export"]):
scan["export"] = export
_update_overview()
_update_param_plot()
@ -563,7 +563,7 @@ def create():
def merge_button_callback():
scan_into = _get_selected_scan()
scan_from = det_data[int(merge_from_select.value)]
scan_from = dataset[int(merge_from_select.value)]
if scan_into is scan_from:
print("WARNING: Selected scans for merging are identical")
@ -587,7 +587,7 @@ def create():
restore_button.on_click(restore_button_callback)
def _get_selected_scan():
return det_data[scan_table_source.selected.indices[0]]
return dataset[scan_table_source.selected.indices[0]]
def param_select_callback(_attr, _old, _new):
_update_table()
@ -720,7 +720,7 @@ def create():
fit_output_textinput = TextAreaInput(title="Fit results:", width=750, height=200)
def proc_all_button_callback():
for scan in det_data:
for scan in dataset:
if scan["export"]:
pyzebra.fit_scan(
scan, fit_params, fit_from=fit_from_spinner.value, fit_to=fit_to_spinner.value
@ -735,7 +735,7 @@ def create():
_update_overview()
_update_table()
for scan in det_data:
for scan in dataset:
if "fit" in scan:
options = list(scan["fit"].params.keys())
fit_param_select.options = options
@ -760,7 +760,7 @@ def create():
_update_overview()
_update_table()
for scan in det_data:
for scan in dataset:
if "fit" in scan:
options = list(scan["fit"].params.keys())
fit_param_select.options = options
@ -782,7 +782,7 @@ def create():
temp_file = temp_dir + "/temp"
export_data = []
param_data = []
for scan, param in zip(det_data, scan_table_source.data["param"]):
for scan, param in zip(dataset, scan_table_source.data["param"]):
if scan["export"] and param:
export_data.append(scan)
param_data.append(param)

104
pyzebra/ccl_io.py
View File

@ -93,9 +93,9 @@ def load_1D(filepath):
"""
with open(filepath, "r") as infile:
_, ext = os.path.splitext(filepath)
det_variables = parse_1D(infile, data_type=ext)
dataset = parse_1D(infile, data_type=ext)
return det_variables
return dataset
def parse_1D(fileobj, data_type):
@ -133,7 +133,7 @@ def parse_1D(fileobj, data_type):
metadata["zebra_mode"] = "nb"
# read data
scan = []
dataset = []
if data_type == ".ccl":
ccl_first_line = CCL_FIRST_LINE + CCL_ANGLES[metadata["zebra_mode"]]
ccl_second_line = CCL_SECOND_LINE
@ -143,47 +143,49 @@ def parse_1D(fileobj, data_type):
if not line or line.isspace():
continue
s = {}
s["export"] = True
scan = {}
scan["export"] = True
# first line
for param, (param_name, param_type) in zip(line.split(), ccl_first_line):
s[param_name] = param_type(param)
scan[param_name] = param_type(param)
# second line
next_line = next(fileobj)
for param, (param_name, param_type) in zip(next_line.split(), ccl_second_line):
s[param_name] = param_type(param)
scan[param_name] = param_type(param)
if s["scan_motor"] != "om":
if scan["scan_motor"] != "om":
raise Exception("Unsupported variable name in ccl file.")
# "om" -> "omega"
s["scan_motor"] = "omega"
s["scan_motors"] = ["omega", ]
scan["scan_motor"] = "omega"
scan["scan_motors"] = ["omega", ]
# overwrite metadata, because it only refers to the scan center
half_dist = (s["n_points"] - 1) / 2 * s["angle_step"]
s["omega"] = np.linspace(s["omega"] - half_dist, s["omega"] + half_dist, s["n_points"])
half_dist = (scan["n_points"] - 1) / 2 * scan["angle_step"]
scan["omega"] = np.linspace(
scan["omega"] - half_dist, scan["omega"] + half_dist, scan["n_points"]
)
# subsequent lines with counts
counts = []
while len(counts) < s["n_points"]:
while len(counts) < scan["n_points"]:
counts.extend(map(float, next(fileobj).split()))
s["counts"] = np.array(counts)
s["counts_err"] = np.sqrt(np.maximum(s["counts"], 1))
scan["counts"] = np.array(counts)
scan["counts_err"] = np.sqrt(np.maximum(scan["counts"], 1))
if s["h"].is_integer() and s["k"].is_integer() and s["l"].is_integer():
s["h"], s["k"], s["l"] = map(int, (s["h"], s["k"], s["l"]))
if scan["h"].is_integer() and scan["k"].is_integer() and scan["l"].is_integer():
scan["h"], scan["k"], scan["l"] = map(int, (scan["h"], scan["k"], scan["l"]))
scan.append({**metadata, **s})
dataset.append({**metadata, **scan})
elif data_type == ".dat":
# TODO: this might need to be adapted in the future, when "gamma" will be added to dat files
if metadata["zebra_mode"] == "nb":
metadata["gamma"] = metadata["twotheta"]
s = defaultdict(list)
s["export"] = True
scan = defaultdict(list)
scan["export"] = True
match = re.search("Scanning Variables: (.*), Steps: (.*)", next(fileobj))
motors = [motor.lower() for motor in match.group(1).split(", ")]
@ -192,8 +194,8 @@ def parse_1D(fileobj, data_type):
match = re.search("(.*) Points, Mode: (.*), Preset (.*)", next(fileobj))
if match.group(2) != "Monitor":
raise Exception("Unknown mode in dat file.")
s["n_points"] = int(match.group(1))
s["monitor"] = float(match.group(3))
scan["n_points"] = int(match.group(1))
scan["monitor"] = float(match.group(3))
col_names = list(map(str.lower, next(fileobj).split()))
@ -203,56 +205,56 @@ def parse_1D(fileobj, data_type):
break
for name, val in zip(col_names, line.split()):
s[name].append(float(val))
scan[name].append(float(val))
for name in col_names:
s[name] = np.array(s[name])
scan[name] = np.array(scan[name])
s["counts_err"] = np.sqrt(np.maximum(s["counts"], 1))
scan["counts_err"] = np.sqrt(np.maximum(scan["counts"], 1))
s["scan_motors"] = []
scan["scan_motors"] = []
for motor, step in zip(motors, steps):
if step == 0:
# it's not a scan motor, so keep only the median value
s[motor] = np.median(s[motor])
scan[motor] = np.median(scan[motor])
else:
s["scan_motors"].append(motor)
scan["scan_motors"].append(motor)
# "om" -> "omega"
if "om" in s["scan_motors"]:
s["scan_motors"][s["scan_motors"].index("om")] = "omega"
s["omega"] = s["om"]
del s["om"]
if "om" in scan["scan_motors"]:
scan["scan_motors"][scan["scan_motors"].index("om")] = "omega"
scan["omega"] = scan["om"]
del scan["om"]
# "tt" -> "temp"
if "tt" in s["scan_motors"]:
s["scan_motors"][s["scan_motors"].index("tt")] = "temp"
s["temp"] = s["tt"]
del s["tt"]
if "tt" in scan["scan_motors"]:
scan["scan_motors"][scan["scan_motors"].index("tt")] = "temp"
scan["temp"] = scan["tt"]
del scan["tt"]
# "mf" stays "mf"
# "phi" stays "phi"
s["scan_motor"] = s["scan_motors"][0]
scan["scan_motor"] = scan["scan_motors"][0]
if "h" not in s:
s["h"] = s["k"] = s["l"] = float("nan")
if "h" not in scan:
scan["h"] = scan["k"] = scan["l"] = float("nan")
for param in ("mf", "temp"):
if param not in metadata:
s[param] = 0
scan[param] = 0
s["idx"] = 1
scan["idx"] = 1
scan.append({**metadata, **s})
dataset.append({**metadata, **scan})
else:
print("Unknown file extention")
return scan
return dataset
def export_1D(data, path, export_target, hkl_precision=2):
def export_1D(dataset, path, export_target, hkl_precision=2):
"""Exports data in the .comm/.incomm format for fullprof or .col/.incol format for jana.
Scans with integer/real hkl values are saved in .comm/.incomm or .col/.incol files
@ -262,11 +264,11 @@ def export_1D(data, path, export_target, hkl_precision=2):
if export_target not in EXPORT_TARGETS:
raise ValueError(f"Unknown export target: {export_target}.")
zebra_mode = data[0]["zebra_mode"]
zebra_mode = dataset[0]["zebra_mode"]
exts = EXPORT_TARGETS[export_target]
file_content = {ext: [] for ext in exts}
for scan in data:
for scan in dataset:
if "fit" not in scan:
continue
@ -306,7 +308,7 @@ def export_1D(data, path, export_target, hkl_precision=2):
out_file.writelines(content)
def export_ccl_compare(data1, data2, path, export_target, hkl_precision=2):
def export_ccl_compare(dataset1, dataset2, path, export_target, hkl_precision=2):
"""Exports compare data in the .comm/.incomm format for fullprof or .col/.incol format for jana.
Scans with integer/real hkl values are saved in .comm/.incomm or .col/.incol files
@ -316,11 +318,11 @@ def export_ccl_compare(data1, data2, path, export_target, hkl_precision=2):
if export_target not in EXPORT_TARGETS:
raise ValueError(f"Unknown export target: {export_target}.")
zebra_mode = data1[0]["zebra_mode"]
zebra_mode = dataset1[0]["zebra_mode"]
exts = EXPORT_TARGETS[export_target]
file_content = {ext: [] for ext in exts}
for scan1, scan2 in zip(data1, data2):
for scan1, scan2 in zip(dataset1, dataset2):
if "fit" not in scan1:
continue
@ -363,9 +365,9 @@ def export_ccl_compare(data1, data2, path, export_target, hkl_precision=2):
out_file.writelines(content)
def export_param_study(data, param_data, path):
def export_param_study(dataset, param_data, path):
file_content = []
for scan, param in zip(data, param_data):
for scan, param in zip(dataset, param_data):
if "fit" not in scan:
continue

30
pyzebra/ccl_process.py
View File

@ -1,7 +1,7 @@
import os
import numpy as np
from lmfit.models import Gaussian2dModel, GaussianModel, LinearModel, PseudoVoigtModel, VoigtModel
from lmfit.models import GaussianModel, LinearModel, PseudoVoigtModel, VoigtModel
from scipy.integrate import simpson, trapezoid
from .ccl_io import CCL_ANGLES
@ -259,31 +259,3 @@ def get_area(scan, area_method, lorentz):
area_s = np.abs(area_s * corr_factor)
scan["area"] = (area_v, area_s)
def fit_event(scan, fr_from, fr_to, y_from, y_to, x_from, x_to):
data_roi = scan["data"][fr_from:fr_to, y_from:y_to, x_from:x_to]
model = GaussianModel()
fr = np.arange(fr_from, fr_to)
counts_per_fr = np.sum(data_roi, axis=(1, 2))
params = model.guess(counts_per_fr, fr)
result = model.fit(counts_per_fr, x=fr, params=params)
frC = result.params["center"].value
intensity = result.params["height"].value
counts_std = counts_per_fr.std()
counts_mean = counts_per_fr.mean()
snr = 0 if counts_std == 0 else counts_mean / counts_std
model = Gaussian2dModel()
xs, ys = np.meshgrid(np.arange(x_from, x_to), np.arange(y_from, y_to))
xs = xs.flatten()
ys = ys.flatten()
counts = np.sum(data_roi, axis=0).flatten()
params = model.guess(counts, xs, ys)
result = model.fit(counts, x=xs, y=ys, params=params)
xC = result.params["centerx"].value
yC = result.params["centery"].value
scan["fit"] = {"frame": frC, "x_pos": xC, "y_pos": yC, "intensity": intensity, "snr": snr}

86
pyzebra/h5.py
View File

@ -1,6 +1,6 @@
import h5py
import numpy as np
from lmfit.models import Gaussian2dModel, GaussianModel
META_MATRIX = ("UB")
META_CELL = ("cell")
@ -74,69 +74,97 @@ def read_detector_data(filepath, cami_meta=None):
n, cols, rows = data.shape
data = data.reshape(n, rows, cols)
det_data = {"data": data}
det_data["original_filename"] = filepath
scan = {"data": data}
scan["original_filename"] = filepath
if "/entry1/zebra_mode" in h5f:
det_data["zebra_mode"] = h5f["/entry1/zebra_mode"][0].decode()
scan["zebra_mode"] = h5f["/entry1/zebra_mode"][0].decode()
else:
det_data["zebra_mode"] = "nb"
scan["zebra_mode"] = "nb"
# overwrite zebra_mode from cami
if cami_meta is not None:
if "zebra_mode" in cami_meta:
det_data["zebra_mode"] = cami_meta["zebra_mode"][0]
scan["zebra_mode"] = cami_meta["zebra_mode"][0]
# om, sometimes ph
if det_data["zebra_mode"] == "nb":
det_data["omega"] = h5f["/entry1/area_detector2/rotation_angle"][:]
if scan["zebra_mode"] == "nb":
scan["omega"] = h5f["/entry1/area_detector2/rotation_angle"][:]
else: # bi
det_data["omega"] = h5f["/entry1/sample/rotation_angle"][:]
scan["omega"] = h5f["/entry1/sample/rotation_angle"][:]
det_data["gamma"] = h5f["/entry1/ZEBRA/area_detector2/polar_angle"][:] # gammad
det_data["nu"] = h5f["/entry1/ZEBRA/area_detector2/tilt_angle"][:] # nud
det_data["ddist"] = h5f["/entry1/ZEBRA/area_detector2/distance"][:]
det_data["wave"] = h5f["/entry1/ZEBRA/monochromator/wavelength"][:]
det_data["chi"] = h5f["/entry1/sample/chi"][:] # ch
det_data["phi"] = h5f["/entry1/sample/phi"][:] # ph
det_data["ub"] = h5f["/entry1/sample/UB"][:].reshape(3, 3)
det_data["name"] = h5f["/entry1/sample/name"][0].decode()
det_data["cell"] = h5f["/entry1/sample/cell"][:]
scan["gamma"] = h5f["/entry1/ZEBRA/area_detector2/polar_angle"][:] # gammad
scan["nu"] = h5f["/entry1/ZEBRA/area_detector2/tilt_angle"][:] # nud
scan["ddist"] = h5f["/entry1/ZEBRA/area_detector2/distance"][:]
scan["wave"] = h5f["/entry1/ZEBRA/monochromator/wavelength"][:]
scan["chi"] = h5f["/entry1/sample/chi"][:] # ch
scan["phi"] = h5f["/entry1/sample/phi"][:] # ph
scan["ub"] = h5f["/entry1/sample/UB"][:].reshape(3, 3)
scan["name"] = h5f["/entry1/sample/name"][0].decode()
scan["cell"] = h5f["/entry1/sample/cell"][:]
if n == 1:
# a default motor for a single frame file
det_data["scan_motor"] = "omega"
scan["scan_motor"] = "omega"
else:
for var in ("omega", "gamma", "nu", "chi", "phi"):
if abs(det_data[var][0] - det_data[var][-1]) > 0.1:
det_data["scan_motor"] = var
if abs(scan[var][0] - scan[var][-1]) > 0.1:
scan["scan_motor"] = var
break
else:
raise ValueError("No angles that vary")
# optional parameters
if "/entry1/sample/magnetic_field" in h5f:
det_data["mf"] = h5f["/entry1/sample/magnetic_field"][:]
scan["mf"] = h5f["/entry1/sample/magnetic_field"][:]
if "/entry1/sample/temperature" in h5f:
det_data["temp"] = h5f["/entry1/sample/temperature"][:]
scan["temp"] = h5f["/entry1/sample/temperature"][:]
# overwrite metadata from .cami
if cami_meta is not None:
if "crystal" in cami_meta:
cami_meta_crystal = cami_meta["crystal"]
if "name" in cami_meta_crystal:
det_data["name"] = cami_meta_crystal["name"]
scan["name"] = cami_meta_crystal["name"]
if "UB" in cami_meta_crystal:
det_data["ub"] = cami_meta_crystal["UB"]
scan["ub"] = cami_meta_crystal["UB"]
if "cell" in cami_meta_crystal:
det_data["cell"] = cami_meta_crystal["cell"]
scan["cell"] = cami_meta_crystal["cell"]
if "lambda" in cami_meta_crystal:
det_data["wave"] = cami_meta_crystal["lambda"]
scan["wave"] = cami_meta_crystal["lambda"]
if "detector parameters" in cami_meta:
cami_meta_detparam = cami_meta["detector parameters"]
if "dist2" in cami_meta_detparam:
det_data["ddist"] = cami_meta_detparam["dist2"]
scan["ddist"] = cami_meta_detparam["dist2"]
return det_data
return scan
def fit_event(scan, fr_from, fr_to, y_from, y_to, x_from, x_to):
data_roi = scan["data"][fr_from:fr_to, y_from:y_to, x_from:x_to]
model = GaussianModel()
fr = np.arange(fr_from, fr_to)
counts_per_fr = np.sum(data_roi, axis=(1, 2))
params = model.guess(counts_per_fr, fr)
result = model.fit(counts_per_fr, x=fr, params=params)
frC = result.params["center"].value
intensity = result.params["height"].value
counts_std = counts_per_fr.std()
counts_mean = counts_per_fr.mean()
snr = 0 if counts_std == 0 else counts_mean / counts_std
model = Gaussian2dModel()
xs, ys = np.meshgrid(np.arange(x_from, x_to), np.arange(y_from, y_to))
xs = xs.flatten()
ys = ys.flatten()
counts = np.sum(data_roi, axis=0).flatten()
params = model.guess(counts, xs, ys)
result = model.fit(counts, x=xs, y=ys, params=params)
xC = result.params["centerx"].value
yC = result.params["centery"].value
scan["fit"] = {"frame": frC, "x_pos": xC, "y_pos": yC, "intensity": intensity, "snr": snr}