#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Batch fitting of experimental standards spectra (PB ratios) from existing ledgers.
This script is analogous to batch_acquire_experimental_stds.py, but instead of running collection and quantification,
it directly fits and updates standards from existing data using run_exp_std_fit.
Usage:
- Edit the 'std_list' list to define your standards (same as in batch_acquire_experimental_stds.py)
- Adjust configuration parameters as needed
- Run the script, or import and call batch_fit_experimental_stds() to fit/update standards for one or multiple samples
Created on May 20, 2026
@author: Andrea
"""
import logging
import os
import time
from pathlib import Path
from typing import List, Dict, Any, Optional, Tuple
from autoemx.core.composition_analysis import EMXSp_Composition_Analyzer
from autoemx.utils.helper import print_double_separator, get_sample_dir
from autoemx.config.ledger_io import ledger_runtime_configs, load_sample_ledger
import autoemx.utils.constants as cnst
import autoemx._paths as paths
import autoemx.config.defaults as dflts
from autoemx.config.schema_models.standards import EDSStandardsFile
from autoemx.config.runtime_configs import ExpStandardsConfig
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
__all__ = ["batch_fit_experimental_stds"]
[docs]
def batch_fit_experimental_stds(
std_ids: List[str],
spectrum_lims: Tuple[float, float] = dflts.spectrum_lims,
use_instrument_background: bool = dflts.use_instrument_background,
min_bckgrnd_cnts: Optional[float] = None,
update_std_library: bool = True,
force_refitting: bool = False,
els_sample: Optional[List[str]] = None,
els_w_frs: Optional[Dict[str, float]] = None,
els_substrate: Optional[List[str]] = None,
els_to_use_for_mean_PB_calc: Optional[List[str]] = None,
min_acceptable_PB_ratio: Optional[float] = None,
quant_flags_accepted: Optional[List[int]] = None,
generate_separate_std_dict: Optional[bool] = None,
output_filename_suffix: str = '',
verbose: bool = True,
stds_path: Optional["str | Path"] = None,
standards_dict_path: Optional["str | Path"] = None,
) -> List[Optional[EMXSp_Composition_Analyzer]]:
"""
Batch fitting of experimental standard spectra from existing ledgers.
Parameters
----------
std_ids : list of str
List of identifiers for the experimental standards.
spectrum_lims : tuple of float, optional
Lower and upper energy limits for spectrum fitting in eV.
use_instrument_background : bool, optional
Whether to use instrument background files during fitting.
min_bckgrnd_cnts : float, optional
Minimum background counts required for a spectrum not to be filtered out
(ClusteringConfig.min_bckgrnd_cnts). If None (default), keeps the value of the standard's active analysis.
update_std_library : bool, optional
If True, the experimental standard library will be updated with the newly fitted PB ratios.
force_refitting : bool, optional
If True, force refitting even if results with same fitting configurations already exist for a standard.
Default is False.
els_sample : Optional[List[str]], optional
List of elements in the sample. If None, defaults to all 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_w_frs : Optional[Dict[str, float]], optional
Dictionary of weight fractions for the elements in the sample to provide if they differ from those in the ledger. Keys should be element symbols. Used to compute PB correction.
els_substrate : Optional[List[str]], optional
List of elements in the substrate. If None, defaults to all 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_to_use_for_mean_PB_calc : list of str, optional
Elements whose P/B ratios are used for the mean reference values (ExpStandardsConfig.els_to_use_for_mean_PB_calc):
e.g. ["Fe", "O"]; ["all"] for all elements; [] or ["none"] for none.
min_acceptable_PB_ratio : float, optional
Minimum P/B ratio for a peak to be accepted as a standard (ExpStandardsConfig.min_acceptable_PB_ratio).
quant_flags_accepted : list of int, optional
Quantification flags of the spectra used for the standard (ExpStandardsConfig.quant_flags_accepted).
generate_separate_std_dict : bool, optional
Whether to update a separate standards file in the project folder instead of the main one
(ExpStandardsConfig.generate_separate_std_dict).
For these four parameters, None (default) keeps the value saved in each standard's ledger. Values given apply
to this run only, like the other parameters of this function.
output_filename_suffix : str, optional
Suffix for the output filename.
verbose : bool, optional
If True, print verbose output.
stds_path : Optional[str or pathlib.Path], optional
Directory containing experimental standard data.
standards_dict_path : Optional[str or pathlib.Path], optional
Path to a standards dictionary file.
Returns
-------
results : list(EMXSp_Composition_Analyzer)
A list of the composition analysis objects (one per sample).
"""
if stds_path is None:
stds_path = os.path.join(os.getcwd(), cnst.STDS_DIR)
standards_dict = None
if standards_dict_path is not None:
standards_dict_path = str(standards_dict_path)
try:
eds_file = EDSStandardsFile.from_json_file(
standards_dict_path, meas_type=None, beam_energy_keV=None
)
standards_dict = eds_file.to_standards_dict()
except Exception as e:
raise ValueError(
f"Could not load standards dictionary from {standards_dict_path}: {e}"
)
def _has_entries(x):
return x is not None and len(x) > 0
# count only non-empty sample IDs
n_samples = sum(1 for s in std_ids if s)
_warnings = [
(els_sample, "elements list", "els_sample"),
(els_w_frs, "element weight fractions", "els_w_frs"),
(els_substrate, "substrate elements list", "els_substrate"),
]
for value, label, argname in _warnings:
if _has_entries(value) and n_samples > 1:
logging.info(
f"⚠️ Warning: More than one sample ID provided. Using the same provided sample "
f"{label} for all samples: {value}.\nEnsure this behaviour is intended, or provide "
f"sample-specific {label} by leaving {argname} as None and specifying them in the ledgers."
)
results: List[Optional[EMXSp_Composition_Analyzer]] = []
comp_analyzer: Optional[EMXSp_Composition_Analyzer] = None
# Track values from the last successfully loaded ledger so we can generate
# a standards-coverage report at the end without re-opening files.
last_microscope_ID: Optional[str] = None
last_measurement_type: Optional[str] = None
last_beam_energy: Optional[float] = None
for std_id in std_ids:
try:
std_path = get_sample_dir(stds_path, std_id)
except Exception as e:
logging.warning("Failed to get sample directory for %s: %s", std_id, e)
continue
ledger_path = os.path.join(
std_path, f"{cnst.LEDGER_FILENAME}{cnst.LEDGER_FILEEXT}"
)
print("\n \n")
print_double_separator()
print_double_separator(f"Standard '{std_id}'")
try:
ledger = load_sample_ledger(ledger_path)
ledger_cfgs = ledger_runtime_configs(ledger)
except FileNotFoundError:
logging.warning(
f"Could not find {ledger_path}. Skipping sample '{std_id}'."
)
continue
except Exception as e:
logging.warning(
f"Error loading {ledger_path}. Skipping sample '{std_id}': {e}"
)
continue
sample_processing_time_start = time.time()
# Retrieve configuration objects for this sample
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
clustering_cfg = ledger_cfgs.clustering_cfg
# Sample elements
if els_sample is not None and (els_sample[0] == "" or els_sample[0] is None):
sample_cfg.elements = sample_cfg.elements + els_sample[1:]
elif els_sample is not None and len(els_sample) > 0:
sample_cfg.elements = els_sample
# Sample substrate elements
if els_substrate is not None and (els_substrate[0] == "" or els_substrate[0] is None):
sample_substrate_cfg.elements = sample_substrate_cfg.elements + els_substrate[1:]
elif els_substrate is not None and len(els_substrate) > 0:
sample_substrate_cfg.elements = els_substrate
exp_stds_cfg = getattr(
ledger.configs.measurement_cfg, "exp_stds_cfg", None
)
if exp_stds_cfg is None:
logging.warning(
f"No ExpStandardsConfig found in ledger for '{std_id}'. "
f"Skipping — cannot fit a standard without a reference formula."
)
continue
if els_w_frs is not None and len(els_w_frs) > 0:
sample_cfg.w_frs = els_w_frs
# Settings of the standard given here replace those saved in its ledger (validated again)
exp_stds_overrides = {
"els_to_use_for_mean_PB_calc": els_to_use_for_mean_PB_calc,
"min_acceptable_PB_ratio": min_acceptable_PB_ratio,
"quant_flags_accepted": quant_flags_accepted,
"generate_separate_std_dict": generate_separate_std_dict,
}
exp_stds_overrides = {key: value for key, value in exp_stds_overrides.items() if value is not None}
if exp_stds_overrides:
exp_stds_cfg = ExpStandardsConfig.model_validate({**exp_stds_cfg.model_dump(), **exp_stds_overrides})
if min_bckgrnd_cnts is not None:
clustering_cfg.min_bckgrnd_cnts = min_bckgrnd_cnts
if use_instrument_background is not None:
quant_cfg.use_instrument_background = use_instrument_background
if spectrum_lims is not None:
quant_cfg.spectrum_lims = spectrum_lims
# Track info for the post-loop standards report
try:
last_microscope_ID = getattr(microscope_cfg, "ID", last_microscope_ID)
last_measurement_type = getattr(measurement_cfg, "type", last_measurement_type)
last_beam_energy = getattr(measurement_cfg, "beam_energy_keV", last_beam_energy)
except Exception:
pass
# --- Run Composition Analyzer (single instantiation, fit mode)
comp_analyzer = EMXSp_Composition_Analyzer.from_ledger(
ledger,
ledger_cfgs,
exp_stds_cfg=exp_stds_cfg,
sample_id=std_id,
standards_dict=standards_dict,
output_filename_suffix=output_filename_suffix,
verbose=verbose,
)
try:
# Always run fitting (run_fitting=True) since this runner is for
# batch fitting from existing ledgers
comp_analyzer.run_exp_std_fit(
run_fitting=True,
force_refitting = force_refitting,
update_std_library=update_std_library,
)
results.append(comp_analyzer)
except Exception as e:
results.append(None)
logging.exception(
f"Sample '{std_id}': fitting failed: {e}"
)
continue
total_process_time = (time.time() - sample_processing_time_start) / 60
print_double_separator(f"Sample '{std_id}' successfully fitted in {total_process_time:.1f} min.")
# Generate a standards-coverage report only when standards library was updated.
if update_std_library and last_microscope_ID is not None \
and last_measurement_type is not None and last_beam_energy is not None:
try:
from autoemx.runners.extract_experimental_standards_details import ( # type: ignore
extract_experimental_standards_details,
)
if standards_dict_path is not None:
# standards_dict_path points to the standards file itself
standards_json_path = Path(standards_dict_path).expanduser().resolve()
standards_dir = standards_json_path.parent
else:
standards_json_path = paths.user_standards_file(
last_microscope_ID, last_measurement_type, last_beam_energy
)
standards_dir = standards_json_path.parent
report_kwargs: Dict[str, Any] = {
"microscope_ID": last_microscope_ID,
"voltage": last_beam_energy,
"report_output_dir": str(standards_dir),
"print_report": True,
"print_per_peak": False,
}
if standards_json_path.exists():
report_kwargs["standards_json_path"] = str(standards_json_path)
report_path = extract_experimental_standards_details(**report_kwargs)
logging.info(
"Standards details report generated at: %s", report_path
)
except Exception as e:
logging.warning(
"Could not generate standards details report after fitting: %s", e
)
return results