Skip to content
Snippets Groups Projects

[Generic] Injection code for DynamicFF corrections

Merged Philipp Schmidt requested to merge feat/shimadzu-injection into feat/shimadzu-correction
All threads resolved!
Files
3
from cal_tools.step_timing import StepTimer
from cal_tools.step_timing import StepTimer
from cal_tools.tools import (
from cal_tools.tools import (
get_dir_creation_date,
get_dir_creation_date,
 
run_prop_seq_from_path,
save_dict_to_hdf5
save_dict_to_hdf5
)
)
from cal_tools.restful_config import calibration_client, extra_calibration_client
from cal_tools.restful_config import calibration_client, extra_calibration_client
from cal_tools.shimadzu import ShimadzuHPVX2
from cal_tools.shimadzu import ShimadzuHPVX2
 
from cal_tools.constants import write_ccv, inject_ccv
import dynflatfield as dffc
import dynflatfield as dffc
from dynflatfield.draw import plot_images, plot_camera_image
from dynflatfield.draw import plot_images, plot_camera_image
module_constants = constants.setdefault(meta["db_module"], {})
module_constants = constants.setdefault(meta["db_module"], {})
module_constants["Offset"] = dict(
module_constants["Offset"] = dict(
conditions=conditions, data=dark, pdu_no=meta["pdu_no"],
conditions=conditions, data=dark, pdu_no=meta["pdu_no"],
creation_time=dark_creation_time
creation_time=dark_creation_time, dims=['ss', 'fs']
)
)
step_timer.done_step("Process dark images")
step_timer.done_step("Process dark images")
display()
display()
module_constants = constants.setdefault(meta["db_module"], {})
module_constants = constants.setdefault(meta["db_module"], {})
module_constants["DynamicFF"] = dict(
module_constants["DynamicFF"] = dict(
conditions=conditions, data=flat_data, pdu_no=meta["pdu_no"],
conditions=conditions, data=flat_data, pdu_no=meta["pdu_no"],
creation_time=flat_creation_time
creation_time=flat_creation_time, dims=['component', 'ss', 'fs']
)
)
step_timer.done_step("Process flat-field images")
step_timer.done_step("Process flat-field images")
``` python
``` python
step_timer.start()
step_timer.start()
 
_, proposal, _ = run_prop_seq_from_path(in_folder)
 
# Output Folder Creation:
# Output Folder Creation:
if local_output:
if local_output:
os.makedirs(out_folder, exist_ok=True)
os.makedirs(out_folder, exist_ok=True)
def inject_ccv(in_folder, metadata_folder, runs, calibration, cond, pdu, const_input, begin_at):
print("* Send to db:", const_input)
print(" - in folder:", in_folder)
print(" - metadata folder:", metadata_folder)
print(" - runs:", runs)
print(" -", calibration)
print(" -", cond)
print(" -", begin_at)
for db_module, module_constants in constants.items():
for db_module, module_constants in constants.items():
for constant_name, constant in module_constants.items():
for constant_name, constant in module_constants.items():
conditions = constant["conditions"]
conditions = constant["conditions"]
conditions_dict = conditions.make_dict(
pdu = pdus["data"][constant["pdu_no"]]
conditions.calibration_types[constant_name])
data_to_store = {db_module: {constant_name: {'0': {
'conditions': conditions_dict,
'data': constant["data"],
}}}}
with NamedTemporaryFile() as tempf:
with NamedTemporaryFile() as tempf:
save_dict_to_hdf5(data_to_store, tempf)
ccv_root = write_ccv(
 
tempf.name,
 
pdu['physical_name'], pdu['uuid'], pdu['detector_type']['name'],
 
constant_name, conditions, constant['creation_time'],
 
proposal, [dark_run, flat_run],
 
constant["data"], constant['dims'])
if db_output:
if db_output:
inject_ccv(
inject_ccv(tempf.name, ccv_root, metadata_folder)
in_folder, metadata_folder, [dark_run, flat_run],
constant_name, conditions, pdus["data"][constant["pdu_no"]],
ofile, constant["creation_time"]
)
if local_output:
if local_output:
ofile = f"{out_folder}/const_{constant_name}_{db_module}.h5"
ofile = f"{out_folder}/const_{constant_name}_{db_module}.h5"
Loading