import logging
from pathlib import Path
from typing import Dict, Optional, Union
from ewokstools.submit import save_and_execute, wait_to_finish_queue
from ewoksutils.task_utils import task_inputs
from ._utils import (
SLURM_JOB_PARAMETERS_PYFAI_RECALIB,
_confirm_submission,
_normalize_slurm_job_parameters,
_print_checking_template_format,
_print_template_invalid,
_print_template_valid,
_validate_yaml_template_model,
_warning_dry_run_mode,
)
from .resources import WORKFLOW_PYFAI_RECALIB
from .resources.models import RecalibModel
logger = logging.getLogger(__name__)
# Copied from pyFAI version 2026.3.0
PYFAI_CALIBRANTS = {
"Si_SRM640a": "pyfai:Si_SRM640a",
"lysozyme": "pyfai:lysozyme",
"Pt": "pyfai:Pt",
"Si_SRM640e": "pyfai:Si_SRM640e",
"Si_SRM640b": "pyfai:Si_SRM640b",
"diamond": "pyfai:diamond",
"Si_SRM640d": "pyfai:Si_SRM640d",
"PBBA": "pyfai:PBBA",
"TiO2": "pyfai:TiO2",
"Cr2O3": "pyfai:Cr2O3",
"CuO": "pyfai:CuO",
"C14H30O": "pyfai:C14H30O",
"LaB6_SRM660c": "pyfai:LaB6_SRM660c",
"Si_SRM640": "pyfai:Si_SRM640",
"LaB6_SRM660a": "pyfai:LaB6_SRM660a",
"C60": "pyfai:C60",
"graphite": "pyfai:graphite",
"Si_SRM640c": "pyfai:Si_SRM640c",
"ZnO": "pyfai:ZnO",
"Au": "pyfai:Au",
"LaB6_SRM660b": "pyfai:LaB6_SRM660b",
"vanadinite": "pyfai:vanadinite",
"Al": "pyfai:Al",
"LaB6": "pyfai:LaB6",
"alpha_Al2O3": "pyfai:alpha_Al2O3",
"Si": "pyfai:Si",
"hydrocerussite": "pyfai:hydrocerussite",
"AgBh": "pyfai:AgBh",
"CeO2": "pyfai:CeO2",
"quartz": "pyfai:quartz",
"Ni": "pyfai:Ni",
"cristobaltite": "pyfai:cristobaltite",
"NaCl": "pyfai:NaCl",
"mock": "pyfai:mock",
"CrOx": "pyfai:CrOx",
}
RECALIB_SUFFIX_DEFAULT = "recalib"
def _get_filename_pyfaiconfig_output(input_filename: str) -> str:
recalib_suffix = f"_{RECALIB_SUFFIX_DEFAULT}"
input_filename = Path(input_filename)
stem = f"{input_filename.stem}{recalib_suffix}"
nb_recalib_files = len(
list(input_filename.parent.glob(f"{stem}*{input_filename.suffix}"))
)
if nb_recalib_files > 0:
stem = f"{stem}_{nb_recalib_files:02}"
return str(input_filename.with_stem(stem))
[docs]
def main_recalib(args):
file = args.FILE
if file.endswith((".yaml", ".yml")):
_main_recalib_from_template(filename=file)
elif file.endswith((".json", ".poni")):
kwargs = vars(args)
_main_recalib_from_cli(**kwargs)
else:
logger.warning(
f"File {file} has an unsupported extension. Skipping it. Supported extensions are .yaml, .yml and .h5."
)
def _main_recalib_from_cli(**kwargs):
for key in ("command",):
kwargs.pop(key, None)
kwargs["filename_pyfaiconfig"] = kwargs.pop("FILE")
kwargs["submit_parameters"] = {
"submit": not kwargs.pop("dry_run"),
"slurm_job_parameters": kwargs.pop("slurm_job_parameters") or {},
"queue": kwargs.pop("queue", None),
"local_execution": kwargs.pop("local_execution", False),
}
_main_recalib(**kwargs)
def _main_recalib_from_template(filename: Union[str, Path]):
_print_checking_template_format()
is_valid, result = _validate_yaml_template_model(filename, RecalibModel)
if not is_valid:
_print_template_invalid(filename, result)
return
_print_template_valid()
_main_recalib(**result)
def _main_recalib(
filename_pyfaiconfig: Optional[str],
filename_bliss_dataset: Optional[str],
scan_nb: Optional[int] = 1,
calibrant: str = "alpha_Al2O3",
detector_name: Optional[str] = "eiger",
filename_pyfaiconfig_output: Optional[str] = None,
submit_parameters: Dict = None,
**kwargs,
) -> None:
if calibrant not in PYFAI_CALIBRANTS:
raise ValueError(
f"Calibrant {calibrant} not recognized. Available calibrants are: {list(PYFAI_CALIBRANTS.keys())}"
)
if not Path(filename_bliss_dataset).is_file():
raise ValueError(
f"Bliss dataset file {filename_bliss_dataset} does not exist. Check full path."
)
dry_run = not submit_parameters.get("submit", True)
_confirm_submission()
if dry_run:
_warning_dry_run_mode()
filename_pyfaiconfig_output = (
filename_pyfaiconfig_output
or _get_filename_pyfaiconfig_output(filename_pyfaiconfig)
)
if filename_bliss_dataset.endswith(".h5"):
filename_data = (
f"{filename_bliss_dataset}::/{scan_nb}.1/measurement/{detector_name}"
)
elif filename_bliss_dataset.endswith((".edf", ".tiff", ".tif")):
filename_data = filename_bliss_dataset
else:
raise ValueError(
f"Bliss dataset file {filename_bliss_dataset} has an unsupported extension. Supported extensions are .h5, .edf, .tiff and .tif."
)
inputs_ewoks = (
task_inputs(
id="pyfai_config",
task_identifier="ewoksxrpd.tasks.pyfaiconfig.PyFaiConfig",
inputs={
"filename": filename_pyfaiconfig,
"calibrant": calibrant,
**kwargs.get("pyfai_config", {}),
},
)
+ task_inputs(
id="pyfai_calib",
task_identifier="ewoksxrpd.tasks.calibrate.CalibrateSingle",
inputs={
"image": filename_data,
"detector": detector_name,
"calibrant": calibrant,
**kwargs.get("pyfai_calib", {}),
},
)
+ task_inputs(
id="save_pyfai_config",
task_identifier="ewoksxrpd.tasks.pyfaiconfig.SavePyFaiConfig",
inputs={
"output_filename": filename_pyfaiconfig_output,
**kwargs.get("save_pyfai_config", {}),
},
)
)
directory_workflows = (
Path(filename_bliss_dataset.replace("RAW_DATA", "PROCESSED_DATA")).parent
/ "workflows"
)
filename_ewoks_workflow = (
directory_workflows
/ f"{Path(filename_bliss_dataset).stem}_pyfai_autocalib_{calibrant}.json"
)
submitted = save_and_execute(
python_package="ewoksid13",
dry_run=dry_run,
workflow=WORKFLOW_PYFAI_RECALIB,
inputs=inputs_ewoks,
destination_filename=filename_ewoks_workflow,
slurm_job_parameters=_normalize_slurm_job_parameters(
SLURM_JOB_PARAMETERS_PYFAI_RECALIB,
submit_parameters.pop("slurm_job_parameters", {}),
),
**submit_parameters,
)
wait_to_finish_queue([submitted])