#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Fitting and quantification of a single X-ray spectrum.
For spectrum-level analysis of fitting and quantification performance.
Import this module in your own code and call the
`fit_and_quantify_spectrum_from_ledger()` function, passing your desired 'sample_ID', 'spectrum_ID' and
options as arguments. This enables integration into larger workflows or pipelines.
Workflow:
- Loads sample configurations and spectral data from `ledger.json`. Results saved in the
legacy `Data.csv` format are converted to a `ledger.json` automatically
- Fits and (optionally) quantifies the selected spectrum
- Optionally plots the fitted spectrum
Notes
-----
- Only the `sample_ID` and 'spectrum_ID' are required if acquisition output is saved in the default Results directory;
otherwise, specify `results_path`.
Created on Tue Jul 29 13:18:16 2025
@author: Andrea
"""
import os
import warnings
import logging
from pathlib import Path
from typing import Any, Dict, Optional, List
import numpy as np
import autoemx.utils.helper as utils
import autoemx.utils.constants as cnst
import autoemx.config.defaults as dflt
from .fit_and_quantify_spectrum import fit_and_quantify_spectrum
from autoemx.config.ledger_io import ledger_runtime_configs, load_sample_ledger
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
__all__ = [
"fit_and_quantify_spectrum_from_ledger",
"fit_and_quantify_spectrum_from_datacsv",
"fit_and_quantify_spectrum_fromDatacsv",
]
[docs]
def fit_and_quantify_spectrum_from_ledger(
sample_ID: str,
spectrum_ID: int,
els_sample: Optional[List[str]] = None,
els_w_frs: Optional[Dict[str, float]] = None,
els_substrate: Optional[List[str]] = None,
is_standard: bool = False,
spectrum_lims: Optional[tuple] = None,
results_path: Optional[str] = None,
use_instrument_background: bool = dflt.use_instrument_background,
quantify: bool = True,
plot_signal: bool = True,
zoom_plot: bool = False,
line_to_plot: str = '',
fit_tol: float = dflt.fit_tolerance,
is_particle: bool = True,
max_undetectable_w_fr: float = dflt.max_undetectable_w_fr,
force_single_iteration: bool = False,
interrupt_fits_bad_spectra: bool = False,
standards_dict: Optional[dict] = None,
print_results: bool = True,
quant_verbose: bool = True,
fitting_verbose: bool = True,
free_area_el_lines: Optional[List[str]] = None,
free_peak_shape_els: Optional[List[str]] = None,
free_peak_shape_params: Optional[List[str]] = None,
free_background_params: Optional[List[str]] = None,
peaks_to_annotate: str = 'main',
ledger: Optional[Any] = None,
identify_missing_elements: Optional[bool] = None,
quantify_plot: Optional[bool] = None,
):
"""
Fit and (optionally) quantify a single spectrum.
Parameters
----------
sample_ID : str
Sample identifier.
spectrum_ID : int
Value reported in the ledger-backed Spectrum ID sequence (legacy-compatible
with the historical 'Spectrum #' identifier from Data.csv).
els_sample : list(str), optional
List of elements in the sample. If the first entry is "" or None, the rest of the list is appended to the
list loaded from the ledger; otherwise, the provided list replaces it.
els_substrate : list(str), optional
List of substrate elements. If the first entry is "" or None, the rest of the list is appended to the
list loaded from the ledger; otherwise, the provided list replaces it.
is_standard : bool
Defines whether measurement is from an experimental standard (i.e., sample of known composition)
results_path : str, optional
Base directory where results are stored. Default: autoemx/Results
use_instrument_background : bool, optional
Whether to use instrument background if present. Default: False
quantify : bool, optional
If True (default), fit and quantify the spectrum. If False, only fit it (e.g. to measure the P/B ratios of a standard).
plot_signal : bool, optional
Whether to plot the fitted spectrum.
zoom_plot : bool, optional
Whether to zoom on a specific line.
line_to_plot : str, optional
Line to zoom on.
peaks_to_annotate : str, optional
Which peaks to annotate in `plot_quantified_spectrum` ('all', 'main', 'most).
fit_tol : float, optional
scipy fit tolerance. Defines conditions of fit convergence
is_particle : bool, optional
If True, treats sample as particle (powder). Uses particle geometry fitting parameters
max_undetectable_w_fr : float, optional
Maximum allowed weight fraction for undetectable elements (default: 0). Total mass fraction of fitted
elements is forced to be between [1-max_undetectable_w_fr, 1]
force_single_iteration : bool, optional
If True, quantification will be run for a single iteration only (default: False).
interrupt_fits_bad_spectra : bool, optional
If True, interrupt fitting if spectrum is detected to lead to poor quantification (default: False).
print_results : bool, optional
If True, prints all fitted parameters and their values (default: True).
quant_verbose : bool, optional
If True, prints quantification operations
fitting_verbose : bool, optional
If True, prints fitting operations
identify_missing_elements : bool, optional
If True, look for elements missing from the fit before fitting (see XSp_Quantifier). Elements confirmed
by the fit are added (and quantified if a standard is available); weaker detections are reported.
If None (default), uses the setting of the sample's active quantification run (off for runs saved
before this option existed).
quantify_plot : bool, optional
Deprecated, use `quantify`.
Returns
-------
quantifier : XSp_Quantifier
The quantifier object containing the results, fit parameters, and methods for further analysis and plotting.
"""
if quantify_plot is not None:
warnings.warn("'quantify_plot' is deprecated, use 'quantify' instead.", FutureWarning, stacklevel=2)
quantify = quantify_plot
if results_path is None:
results_path = os.path.join(os.getcwd(), cnst.RESULTS_DIR)
try:
sample_dir = utils.get_sample_dir(results_path, sample_ID)
except Exception as e:
logging.warning("Failed to get sample directory for %s: %s", sample_ID, e)
return
ledger_path = os.path.join(sample_dir, f"{cnst.LEDGER_FILENAME}{cnst.LEDGER_FILEEXT}")
spectral_info_f_path = ledger_path
utils.print_double_separator(f"Sample '{sample_ID}', spectrum {spectrum_ID}")
try:
if ledger is None:
ledger = load_sample_ledger(ledger_path)
ledger_cfgs = ledger_runtime_configs(ledger)
except FileNotFoundError:
logging.warning(f"Could not find {spectral_info_f_path}. Skipping sample '{sample_ID}'.")
return
except Exception as e:
logging.warning(f"Error loading {spectral_info_f_path}. Skipping sample '{sample_ID}': {e}")
return
# Retrieve configuration objects
microscope_cfg = ledger_cfgs.microscope_cfg
sample_cfg = ledger_cfgs.sample_cfg
measurement_cfg = ledger_cfgs.measurement_cfg
sample_substrate_cfg = ledger_cfgs.sample_substrate_cfg
quant_cfg = ledger_cfgs.quant_cfg
# Resolve requested spectrum directly from ledger entries.
target_entry = None
for idx, spectrum_entry in enumerate(ledger.spectra):
entry_id = spectrum_entry.spectrum_id
if entry_id is None:
if idx == int(spectrum_ID):
target_entry = spectrum_entry
break
continue
try:
if int(float(entry_id)) == int(spectrum_ID):
target_entry = spectrum_entry
break
except (TypeError, ValueError):
if str(entry_id) == str(spectrum_ID):
target_entry = spectrum_entry
break
if target_entry is None:
logging.warning(
f"Spectrum ID {spectrum_ID} not found in ledger-backed data for sample '{sample_ID}'."
)
return
spectrum_relpath = target_entry.spectrum_relpath
if not spectrum_relpath:
logging.warning(
f"Missing spectrum pointer path for spectrum ID {spectrum_ID} in sample '{sample_ID}'."
)
return
spectrum_pointer = Path(sample_dir) / spectrum_relpath
try:
spectrum = np.asarray(ledger._load_counts_from_pointer_file(spectrum_pointer), dtype=float)
except Exception as e:
logging.warning(
f"Could not load spectrum data for spectrum ID {spectrum_ID} in sample '{sample_ID}': {e}"
)
return
background = None
if use_instrument_background:
bkg_relpath = target_entry.instrument_background_relpath
if bkg_relpath:
bkg_path = Path(sample_dir) / bkg_relpath
try:
background = np.asarray(np.load(bkg_path, allow_pickle=False), dtype=float)
except Exception:
background = None
if background is None:
warnings.warn(
"Instrument background not found or empty for this spectrum. "
"Spectral background will be computed instead."
)
sp_collection_time = target_entry.live_acquisition_time
# Calibration and configuration parameters
try:
beam_energy = measurement_cfg.beam_energy_keV
emergence_angle = measurement_cfg.emergence_angle
el_to_quantify = sample_cfg.elements
offset = microscope_cfg.energy_zero
scale = microscope_cfg.bin_width
except Exception as e:
logging.error(f"Error extracting calibration/configuration parameters: {e}")
return
# Sample elements
if els_sample is None:
els_sample = el_to_quantify
elif (els_sample[0] == "" or els_sample[0] is None):
els_sample = el_to_quantify + els_sample[1:]
# Substrate elements
if els_substrate is None:
els_substrate = sample_substrate_cfg.elements
elif (els_substrate[0] == "" or els_substrate[0] is None):
els_substrate = sample_substrate_cfg.elements + els_substrate[1:]
# Spectral limits
if spectrum_lims is None:
spectrum_lims = quant_cfg.spectrum_lims
# Ensure spectrum_lims are integers for array slicing
if spectrum_lims is not None:
try:
spectrum_lims = (int(spectrum_lims[0]), int(spectrum_lims[1]))
except (TypeError, ValueError, IndexError):
logging.error(f"Invalid spectrum_lims: {spectrum_lims}")
return
else:
logging.error("spectrum_lims could not be determined")
return
# Keep static typing explicit for downstream function signature.
if els_sample is None:
els_sample = []
if els_substrate is None:
els_substrate = []
# Weight fractions of a standard, stored in the sample config
if els_w_frs is None and is_standard:
els_w_frs = sample_cfg.w_frs
if els_w_frs is None:
raise ValueError(
f"No 'w_frs' (element weight fractions) found for standard '{sample_ID}' in the ledger "
f"(sample_cfg). Pass 'els_w_frs' directly, or add 'w_frs' to the ledger."
)
quantifier = fit_and_quantify_spectrum(
spectrum_vals = spectrum,
spectrum_lims = spectrum_lims,
microscope_ID = microscope_cfg.ID,
meas_type = measurement_cfg.type,
meas_mode = measurement_cfg.mode,
det_ch_offset=offset,
det_ch_width=scale,
beam_energy = beam_energy,
emergence_angle = emergence_angle,
sp_collection_time = sp_collection_time,
sample_ID = sample_ID,
els_sample = els_sample,
els_w_frs = els_w_frs,
els_substrate = els_substrate,
background_vals=background,
fit_tol = fit_tol,
is_particle = is_particle,
quantify = quantify,
max_undetectable_w_fr = max_undetectable_w_fr,
force_single_iteration = force_single_iteration,
interrupt_fits_bad_spectra = interrupt_fits_bad_spectra,
standards_dict = standards_dict,
plot_signal = plot_signal,
plot_title = f"{sample_ID}_#{spectrum_ID}",
zoom_plot = zoom_plot,
line_to_plot = line_to_plot,
print_results = print_results,
quant_verbose = quant_verbose,
fitting_verbose = fitting_verbose,
free_area_el_lines=free_area_el_lines,
free_peak_shape_els=free_peak_shape_els,
free_peak_shape_params=free_peak_shape_params,
free_background_params=free_background_params,
peaks_to_annotate=peaks_to_annotate,
identify_missing_elements=(bool(getattr(quant_cfg, 'identify_missing_elements', False))
if identify_missing_elements is None else identify_missing_elements),
)
return quantifier
[docs]
def fit_and_quantify_spectrum_fromDatacsv(*args, **kwargs):
"""Deprecated camelCase alias; use fit_and_quantify_spectrum_from_ledger."""
warnings.warn(
"fit_and_quantify_spectrum_fromDatacsv is deprecated; "
"use fit_and_quantify_spectrum_from_ledger instead.",
DeprecationWarning,
stacklevel=2,
)
return fit_and_quantify_spectrum_from_ledger(*args, **kwargs)
[docs]
def fit_and_quantify_spectrum_from_datacsv(*args, **kwargs):
"""Deprecated snake_case alias; use fit_and_quantify_spectrum_from_ledger."""
warnings.warn(
"fit_and_quantify_spectrum_from_datacsv is deprecated; "
"use fit_and_quantify_spectrum_from_ledger instead.",
DeprecationWarning,
stacklevel=2,
)
return fit_and_quantify_spectrum_from_ledger(*args, **kwargs)