from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Sequence
from typing import Tuple
from typing import Union
from pydantic import BaseModel
from . import defaults
from . import detectors as detectors_module
[docs]
class XrfMapPaths(BaseModel):
filename: str
output_root_uri: str
convert_destination: str
workflow_path: str
config_filenames: List[str]
[docs]
class MosaicXrfMapPaths(BaseModel):
filenames: List[str]
output_root_uri: str
concat_bliss_scan_uri: str
convert_destination: str
workflow_path: str
config_filenames: List[str]
[docs]
class FluoXasPaths(BaseModel):
filenames: List[str]
output_root_uri: str
convert_destination: str
workflow_path: str
config_filenames: List[str]
[docs]
class MosaicFluoXasPaths(BaseModel):
filenames: List[List[str]]
output_root_uri: str
concat_output_root_uri: str
convert_destination: str
workflow_path: str
config_filenames: List[str]
[docs]
class CommonParameters(BaseModel):
instrument_name: Optional[str]
detector_names: List[str]
detector_normalization_template: Optional[str]
counter_name: Optional[str]
counter_normalization_template: Optional[str]
config_filenames: List[str]
energy_name: Optional[str]
energy_multiplier: Optional[float]
fast_fitting: bool
quantification: bool
diagnostics: bool
no_fit: bool
fit_single_detector: Optional[bool]
sum_spectra: Optional[bool]
norm_identifier: str
fit_identifier: str
real_axes: Optional[List[str]]
virtual_axes: Optional[Dict[str, str]]
ignore_axes: Optional[List[str]]
axis_units: Optional[Dict[str, str]]
@property
def regrid_positioners(self) -> List[str]:
return sorted(self.virtual_axes)[::-1]
[docs]
class XrfMapParameters(CommonParameters):
filename: str
scan_number: int
output_root_uri: str
resolution: Optional[Dict[str, Tuple[Union[int, float], str]]]
[docs]
class MosaicXrfMapParameters(CommonParameters):
filenames: List[str]
scan_ranges: List[Tuple[int, int]]
exclude_scans: List[List[int]]
output_root_uri: str
concat_bliss_scan_uri: str
resolution: Optional[Dict[str, Tuple[Union[int, float], str]]]
[docs]
class BaseNdRegParameters(BaseModel):
align_counter: Optional[str]
align_crop: Optional[bool]
fast_align_counter_selection: Optional[bool]
interpolation_order: int = 1
transformation_type: str = "translation"
block_size: int = 10
mapper: str = "Optimization-SimpleITK"
mapper_options: dict = {
"metric": "ANTSNeighborhoodCorrelation",
"optimizer": "Powell",
}
@property
def skip_pre_evaluation(self) -> bool:
return not self.align_counter and not self.fast_align_counter_selection
@property
def skip_post_evaluation(self) -> bool:
return not self.align_counter and self.fast_align_counter_selection
[docs]
class FluoXasParameters(CommonParameters, BaseNdRegParameters):
filenames: List[str]
scan_ranges: List[Tuple[int, int]]
exclude_scans: List[List[int]]
output_root_uri: str
resolution: Optional[Dict[str, Tuple[Union[int, float], str]]]
stack_axis: Optional[str]
[docs]
class MosaicFluoXasParameters(CommonParameters, BaseNdRegParameters):
filenames: List[List[str]]
scan_ranges: List[List[Tuple[int, int]]]
exclude_scans: List[List[List[int]]]
group_by_index: bool
output_root_uri: str
concat_output_root_uri: str
resolution: Optional[Dict[str, Tuple[Union[int, float], str]]]
stack_axis: Optional[str]
[docs]
def common_parameters_model(
detectors: Optional[Sequence[Union[int, str]]],
detector_numbers: Optional[Sequence[int]],
detector_names: Optional[Sequence[str]],
config_filenames: Sequence[str],
energy_name: Optional[str],
energy_multiplier: Optional[float],
counter_name: Optional[str],
instrument_name: Optional[str],
fast_fitting: bool,
quantification: bool,
diagnostics: bool,
livetime_ref_value: Union[str, int, float, None],
counter_ref_value: Union[str, int, float, None],
stack: bool,
real_axes: Optional[List[str]],
virtual_axes: Optional[Dict[str, str]],
ignore_axes: Optional[List[str]],
axis_units: Optional[Dict[str, str]],
) -> CommonParameters:
detector_names = detectors_module.get_detector_names(
detectors=detectors,
detector_numbers=detector_numbers,
detector_names=detector_names,
instrument_name=instrument_name,
)
# Default strings
energy_name = (
energy_name
if energy_name is not None
else defaults.ENERGY_COUNTER.get(instrument_name)
)
if energy_name == "":
energy_name = None
counter_name = (
counter_name
if counter_name is not None
else defaults.I0_COUNTER.get(instrument_name)
)
if counter_name == "":
counter_name = None
# Default dicts
axis_units = _merge_dict(axis_units, defaults.AXES_UNITS.get(instrument_name))
ignore_axes = (
ignore_axes
if ignore_axes is not None
else defaults.IGNORE_AXES.get(instrument_name)
)
virtual_axes = (
virtual_axes
if virtual_axes is not None
else defaults.VIRTUAL_AXES.get(instrument_name)
)
# Normalization templates: None gives the default, "" gives None
detector_normalization_template = _build_normalization_template(
livetime_ref_value,
defaults.DEFAULT_LIVETIME_REF_VALUE.get(instrument_name),
"live_time",
)
counter_normalization_template = _build_normalization_template(
counter_ref_value,
defaults.DEFAULT_COUNTER_REF_VALUE.get(instrument_name),
"data",
)
if counter_name is None:
counter_normalization_template = None
elif counter_normalization_template is None:
counter_name = None
# Number of detectors vs. number of configuration files
if len(detector_names) == 0 or len(config_filenames) == 0:
no_fit = True
sum_spectra = None
fit_single_detector = None
elif len(detector_names) == 1:
# Only one detector
if len(config_filenames) != 1:
raise ValueError("Only one pymca configuration is needed for one detector")
no_fit = False
sum_spectra = None
fit_single_detector = True
elif len(config_filenames) == 1:
# More than one detector and fit the sum
no_fit = False
sum_spectra = True
fit_single_detector = True
else:
# More than one detector and fit each detector separately
if len(detector_names) != len(config_filenames):
raise ValueError(
f"{len(detector_names)} pymca configurations are needed for {len(detector_names)} detectors"
)
no_fit = False
sum_spectra = False
fit_single_detector = False
if stack:
norm_identifier = "NormalizeXrfResultsStack"
if fit_single_detector:
fit_identifier = "FitStackSingleDetector"
else:
fit_identifier = "FitStackMultiDetector"
else:
norm_identifier = "NormalizeXrfResults"
if fit_single_detector:
fit_identifier = "FitSingleScanSingleDetector"
else:
fit_identifier = "FitSingleScanMultiDetector"
return CommonParameters(
instrument_name=instrument_name,
detector_names=detector_names,
detector_normalization_template=detector_normalization_template,
counter_name=counter_name,
counter_normalization_template=counter_normalization_template,
energy_name=energy_name,
energy_multiplier=energy_multiplier,
fast_fitting=fast_fitting,
quantification=quantification,
diagnostics=diagnostics,
no_fit=no_fit,
fit_single_detector=fit_single_detector,
config_filenames=config_filenames,
sum_spectra=sum_spectra,
norm_identifier=norm_identifier,
fit_identifier=fit_identifier,
real_axes=real_axes,
virtual_axes=virtual_axes,
ignore_axes=ignore_axes,
axis_units=axis_units,
)
def _build_normalization_template(
value: Union[str, int, float, None],
default_value: Union[str, int, float, None],
dataset: str,
) -> Optional[str]:
if value is None:
value = default_value
if value is not None and value != "":
return f"{value}/<instrument/{{}}/{dataset}>"
return None
def _merge_dict(
adict: Optional[Dict[str, Any]], default: Optional[Dict[str, Any]]
) -> Optional[Dict[str, Any]]:
if not adict and not default:
return None
result = {}
if default:
result.update(default)
if adict:
result.update(adict)
return result