import logging
from pathlib import Path
from typing import Dict, List, Union
from ewokstools.submit import save_and_execute, wait_to_finish_queue
from ewoksutils.task_utils import task_inputs
from pyFAI.io.integration_config import WorkerConfig
from ._utils import (
SLURM_JOB_PARAMETERS_BACKGROUND,
SLURM_JOB_PARAMETERS_CNMF,
SLURM_JOB_PARAMETERS_INFERENCE,
SLURM_JOB_PARAMETERS_INTEGRATE,
EwoksWorkflow,
_confirm_submission,
_get_destination_filename,
_get_nb_components,
_get_pyfai_config_filename,
_get_pyfai_wavelength,
_make_nxprocess_name,
_normalize_external_output_filenames,
_normalize_lima_url_template_id13,
_normalize_pyfai_configurations,
_normalize_pyfai_tag,
_normalize_slurm_job_parameters,
_normalize_tag,
_print_checking_template_format,
_print_job_progress,
_print_number_of_jobs_to_submit,
_print_template_invalid,
_print_template_valid,
_transfer_cif_files,
_validate_yaml_template_model,
_warning_dry_run_mode,
_warning_if_too_many_jobs,
get_list_scan_info_from_bliss_filenames_id13,
)
from .resources import REPROCESS_WORKFLOW
from .resources.models import ReprocessModel
logger = logging.getLogger(__name__)
[docs]
def main_reprocess(args):
for file in args.FILES:
if file.endswith((".yaml", ".yml")):
_main_reprocess_from_template(filename=file)
elif file.endswith(".h5"):
kwargs = vars(args)
for key in ("FILES", "command"):
kwargs.pop(key, None)
_main_reprocess_from_cli(filename=file, **kwargs)
else:
logger.warning(
f"File {file} has an unsupported extension. Skipping it. Supported extensions are .yaml, .yml and .h5."
)
def _main_reprocess_from_template(filename: Union[str, Path]):
_print_checking_template_format()
is_valid, result = _validate_yaml_template_model(filename, ReprocessModel)
if not is_valid:
_print_template_invalid(filename, result)
return
_print_template_valid()
_main_reprocess(**result)
def _main_reprocess_from_cli(filename: Union[str, Path], **kwargs):
if kwargs.get("pyfai_configurations") is None:
logger.error("PyFAI configuration file is required for integration.")
return
integration_params = {
"bliss_filenames": [filename],
"detectors": kwargs.pop("detectors"),
"scans": kwargs.pop("scans"),
"save_separate_files": kwargs.pop("save_separate_files"),
"output_root_folder": kwargs.pop("output_root_folder"),
"tag": kwargs.pop("tag"),
}
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),
}
pyfai_configurations = [
{"filename": file} for file in kwargs.pop("pyfai_configurations")
]
for pyfai_config in pyfai_configurations:
pyfai_config.update({k: v for k, v in kwargs.items() if v is not None})
_main_reprocess(
pyfai_configurations=pyfai_configurations,
submit_parameters=submit_parameters,
**integration_params,
**kwargs,
)
def _main_reprocess(
bliss_filenames: Union[str, List[str]],
pyfai_configurations: Union[str, List[str]],
detectors: Union[str, List[str]] = None,
scans: Union[int, List[int]] = None,
subscans: Union[int, List[int]] = None,
save_separate_files: bool = False,
output_root_folder: str = None,
tag: str = "",
submit_parameters: Dict = None,
**kwargs,
):
"""
Main function to reprocess scans from ID13.
param: bliss_filenames: str or list of str
One or several bliss files containing the scans to reprocess. Can be provided as a list of filenames or as a single string.
param: pyfai_configurations:
One or several pyFAI configurations to apply for the reprocessing.
Can be provided as a list of dictionaries or as a single dictionary.
Each configuration should contain:
- "filename": the path to the .poni/.json file with the pyFAI configuration.
- "tag" (optional): a tag to differentiate the output files if several pyFAI configurations are provided.
- **additional_parameters for pyFAI worker
param: detectors: str or list of str, optional
The detector name(s) to reprocess. If not provided, all 2D detectors will be reprocessed.
param: scans: int or list of int, optional
The scan number(s) to reprocess. If not provided, all scans will be reprocessed.
param: subscans: int or list of int, optional
The subscan number(s) to reprocess. If not provided, all subscans will be reprocessed.
param: save_separate_files: bool, optional
Whether to save integrated data in the dataset file or linked to separated files.
param: save_pyfai_config: bool, optional
Whether to save the pyFAI configuration file used for the integration in the same folder as
param: output_root_folder: str, optional
The root output folder where the reprocessed files will be saved. If not provided, the
param: tag: str, optional
A tag to add to the output files and job names to differentiate them from other reprocessings.
param: submit_parameters: dict, optional
Additional parameters for job submission, such as whether to actually submit the jobs or just do a dry run, and parameters for slurm job submission.
**kwargs:
All the kwargs should be dictionaries whose keys are the id of the nodes in the workflow.
Hence, you can pass any input for any task through this dictionary, for example:
"diffmap" : {
"normalization_counter": value1,
}
"""
workflow_reprocess = EwoksWorkflow(REPROCESS_WORKFLOW)
submit_parameters = submit_parameters or {}
dry_run = not submit_parameters.get("submit")
list_scans_info = get_list_scan_info_from_bliss_filenames_id13(
bliss_filenames=bliss_filenames,
detectors=detectors,
scans=scans,
subscans=subscans,
output_root_folder=output_root_folder,
**kwargs,
)
pyfai_configurations = _normalize_pyfai_configurations(pyfai_configurations)
nb_pyfai_configs = len(pyfai_configurations)
nb_scans = len(list_scans_info)
nb_total_jobs = nb_pyfai_configs * nb_scans
_print_number_of_jobs_to_submit(nb_total_jobs)
_confirm_submission()
if dry_run:
_warning_dry_run_mode()
else:
_warning_if_too_many_jobs(nb_total_jobs)
submitted_jobs = []
tag = _normalize_tag(tag, dry_run)
for index_scan, scan_info in enumerate(list_scans_info):
output_filename, external_output_filename = (
_normalize_external_output_filenames(
scan_info=scan_info,
save_separate_files=save_separate_files,
)
)
for index_pyfai, pyfai_config in enumerate(pyfai_configurations):
index_job = index_scan * nb_pyfai_configs + index_pyfai + 1
inputs_ewoks_reprocess = []
suffix = f"{tag}{_normalize_pyfai_tag(pyfai_config, index_pyfai)}"
nxprocess_name_integrate = _make_nxprocess_name(
scan_info.detector_name, workflow_reprocess.integrate.id, suffix
)
nxprocess_name_diffmap = _make_nxprocess_name(
scan_info.detector_name, workflow_reprocess.diffmap.id, suffix
)
# workerconfig object
pyfai_workerconfig_dict = WorkerConfig.from_file(
pyfai_config.get("filename")
).as_dict()
pyfai_workerconfig_dict.update(pyfai_config)
# Inputs for pyFAI config generation
inputs_ewoks_reprocess += task_inputs(
id=workflow_reprocess.pyfai_config.id,
task_identifier=workflow_reprocess.pyfai_config.task_identifier,
inputs={
"filename": pyfai_config.get("filename"),
"integration_options": pyfai_config,
},
)
# Inputs for Saving pyFAI config file
inputs_ewoks_reprocess += task_inputs(
id=workflow_reprocess.save_pyfai_config.id,
task_identifier=workflow_reprocess.save_pyfai_config.task_identifier,
inputs={
"output_filename": _get_pyfai_config_filename(
output_filename,
scan_info.scan_nb,
scan_info.detector_name,
suffix,
),
},
)
# Inputs for IntegrateBlissScan task
inputs_ewoks_reprocess += task_inputs(
id=workflow_reprocess.integrate.id,
task_identifier=workflow_reprocess.integrate.task_identifier,
inputs={
"filename": scan_info.filename_raw_dataset,
"detector_name": scan_info.detector_name,
"scan": scan_info.scan_nb,
"subscan": scan_info.subscan,
"lima_url_template": _normalize_lima_url_template_id13(
lima_url_template=scan_info.lima_url_template,
**kwargs.get("integrate", {}),
),
"lima_url_template_args": scan_info.lima_url_template_args,
"nxprocess_name": nxprocess_name_integrate,
"output_filename": output_filename,
"external_output_filename": external_output_filename,
**kwargs.get("integrate", {}),
},
)
inputs_ewoks_reprocess += task_inputs(
id=workflow_reprocess.diffmap.id,
task_identifier=workflow_reprocess.diffmap.task_identifier,
inputs={
"nxdata_url": f"{external_output_filename}::{scan_info.scan_nb}.{scan_info.subscan}/{nxprocess_name_integrate}/integrated",
"nxprocess_name": nxprocess_name_diffmap,
"master_filename": scan_info.filename_raw_dataset,
"scan_nb": scan_info.scan_nb,
"lima_name": scan_info.detector_name,
"output_filename": output_filename,
"external_output_filename": external_output_filename,
"integration_options": pyfai_workerconfig_dict,
**kwargs.get("diffmap", {}),
},
)
# Inputs for average task
inputs_ewoks_reprocess += task_inputs(
id=workflow_reprocess.average.id,
task_identifier=workflow_reprocess.average.task_identifier,
inputs={
"nxprocess_name": _make_nxprocess_name(
scan_info.detector_name,
workflow_reprocess.average.id,
suffix,
),
"detector_name": scan_info.detector_name,
"scan_nb": scan_info.scan_nb,
"output_filename": output_filename,
"external_output_filename": external_output_filename,
**kwargs.get("average", {}),
},
)
# Inputs for edf task
if save_separate_files:
_replace_edf = f"_{scan_info.detector_name}{suffix}.edf"
else:
_replace_edf = (
f"_{scan_info.scan_nb:05}_{scan_info.detector_name}{suffix}.edf"
)
inputs_ewoks_reprocess += task_inputs(
id=workflow_reprocess.toedf.id,
task_identifier=workflow_reprocess.toedf.task_identifier,
inputs={
"external_output_filename": external_output_filename.replace(
".h5",
_replace_edf,
),
**kwargs.get("edf", {}),
},
)
# Inputs for neural network tasks
wavelength_A = _get_pyfai_wavelength(pyfai_config.get("filename")) * 1e10
radial_limits = kwargs.get("cnmf", {}).get("radial_limits")
background_parameters = kwargs.get("background", {}).copy()
nxprocess_name_background = _make_nxprocess_name(
scan_info.detector_name, workflow_reprocess.background.id, suffix
)
inputs_ewoks_reprocess += task_inputs(
id=workflow_reprocess.background.id,
task_identifier=workflow_reprocess.background.task_identifier,
inputs={
"nxdata_url": f"{external_output_filename}::{scan_info.scan_nb}.{scan_info.subscan}/{nxprocess_name_diffmap}/diffmap",
"wavelength": wavelength_A,
"nxprocess_name": nxprocess_name_background,
"radial_limits": radial_limits,
"use_neuralnetwork": False,
"destination_file": _get_destination_filename(
output_filename, scan_info.scan_nb, nxprocess_name_background
),
"worker_module": "scattering",
"wait_to_finish": kwargs.get("cnmf", {}).get(
"do_matrix_factorization"
),
"output_filename": output_filename,
"external_output_filename": external_output_filename,
"slurm_job_parameters": _normalize_slurm_job_parameters(
SLURM_JOB_PARAMETERS_BACKGROUND,
background_parameters.pop("slurm_job_parameters", {}),
),
"submit_to_slurm": True,
**background_parameters,
},
)
cnmf_parameters = kwargs.get("cnmf", {}).copy()
nxprocess_name_cnmf = _make_nxprocess_name(
scan_info.detector_name, workflow_reprocess.cnmf.id, suffix
)
references_directory = cnmf_parameters.pop("references_directory")
if cnmf_parameters.get("do_matrix_factorization", True):
target_cif_directory = _transfer_cif_files(
references_directory=str(Path(references_directory).resolve()),
target_directory=Path(external_output_filename).parent
/ f".{nxprocess_name_cnmf}_cif",
)
nb_components = _get_nb_components(
target_cif_directory=target_cif_directory,
nb_components=cnmf_parameters.pop("nb_components", None),
nb_residues=cnmf_parameters.pop("nb_residues", 1),
)
else:
target_cif_directory = None
nb_components = None
inputs_ewoks_reprocess += task_inputs(
id=workflow_reprocess.cnmf.id,
task_identifier=workflow_reprocess.cnmf.task_identifier,
inputs={
"nxdata_url": f"{external_output_filename}::{scan_info.scan_nb}.{scan_info.subscan}/{nxprocess_name_background}/filtered",
"wavelength": wavelength_A,
"nxprocess_name": nxprocess_name_cnmf,
"radial_limits": radial_limits,
"destination_file": _get_destination_filename(
output_filename, scan_info.scan_nb, nxprocess_name_cnmf
),
"worker_module": "scattering",
"wait_to_finish": False,
"output_filename": output_filename,
"external_output_filename": external_output_filename,
"slurm_job_parameters": _normalize_slurm_job_parameters(
SLURM_JOB_PARAMETERS_CNMF,
cnmf_parameters.pop("slurm_job_parameters", {}),
),
"submit_to_slurm": True,
"references_directory": target_cif_directory,
"nb_components": nb_components,
**cnmf_parameters,
},
)
inference_parameters = kwargs.get("inference", {}).copy()
nxprocess_name_inference = _make_nxprocess_name(
scan_info.detector_name, workflow_reprocess.inference.id, suffix
)
inputs_ewoks_reprocess += task_inputs(
id=workflow_reprocess.inference.id,
task_identifier=workflow_reprocess.inference.task_identifier,
inputs={
"nxdata_url": f"{external_output_filename}::{scan_info.scan_nb}.{scan_info.subscan}/{nxprocess_name_background}/filtered",
"wavelength": wavelength_A,
"references_directory": target_cif_directory,
"nxprocess_name": nxprocess_name_inference,
"radial_limits": radial_limits,
"destination_file": _get_destination_filename(
output_filename, scan_info.scan_nb, nxprocess_name_inference
),
"worker_module": "scattering",
"wait_to_finish": False,
"output_filename": output_filename,
"external_output_filename": external_output_filename,
"slurm_job_parameters": _normalize_slurm_job_parameters(
SLURM_JOB_PARAMETERS_INFERENCE,
inference_parameters.pop("slurm_job_parameters", {}),
),
"submit_to_slurm": True,
**inference_parameters,
},
)
submit_parameters_ = submit_parameters.copy()
submitted_jobs.append(
save_and_execute(
dry_run=dry_run,
workflow=REPROCESS_WORKFLOW,
inputs=inputs_ewoks_reprocess,
destination_filename=_get_destination_filename(
output_filename,
scan_info.scan_nb,
f"{scan_info.detector_name}{suffix}_reprocess",
),
slurm_job_parameters=_normalize_slurm_job_parameters(
SLURM_JOB_PARAMETERS_INTEGRATE,
submit_parameters_.pop("slurm_job_parameters", {}),
),
**submit_parameters_,
)
)
_print_job_progress(index_job, nb_total_jobs, dry_run)
wait_to_finish_queue(submitted_jobs)