#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Peak overlap warnings
=====================
Identifies reference lines used to quantify sample elements that overlap with other lines whose area
is fitted freely, which may compromise the quantification of those elements.
Lines whose area is fitted freely are the reference lines of every fitted element (sample and substrate),
i.e. the lines that are their own weight reference in XSp_Fitter (Ka1, La1 or Ll when La1 is absent,
Ma1 for Z > 58, Mz1 for Z <= 58), and the lines listed in `free_area_el_lines`. All other lines have their
area tied to a reference line of the same element.
The overlap criteria are defined per measurement method in autoemx/data/Xray_lines/common_peak_overlaps.json
(generated by make_common_peak_overlaps_file.py, which also lists the candidate pairs of reference lines used
by the automated peak identification). Methods without an entry (anything other than 'EDS', currently)
produce no warnings. The pairs are computed here at runtime, as they also depend on `free_area_el_lines`. For EDS, the severity of an overlap depends on the separation of the two
lines in units of sigma_eff = sqrt((sigma1^2 + sigma2^2) / 2), with sigma the detector resolution at each
line energy, computed with the calibration of the microscope used (DetectorResponseFunction.det_sigma).
"""
import json
import os
from functools import lru_cache
from typing import Dict, Iterable, List, Optional, Sequence, Tuple
import numpy as np
import autoemx.calibrations as calibs
import autoemx.config.defaults as dflt
from autoemx._logging import get_logger
from autoemx.core.fitter.detector_response import DetectorResponseFunction
from autoemx.core.fitter.fitter import get_reference_xray_line
from autoemx.core.fitter.peaks import DEFAULT_FREE_AREA_EL_LINES
from autoemx.core.quantifier.quantifier import XSp_Quantifier
from autoemx.data.Xray_lines import SCRIPT_DIR as XRAY_LINES_DIR, get_el_xray_lines
from autoemx.utils.helper import print_double_separator
logger = get_logger(__name__)
PEAK_OVERLAPS_FILE = os.path.join(XRAY_LINES_DIR, "common_peak_overlaps.json")
# Lines are fitted only if the overvoltage exceeds this value (see XSp_Fitter._define_xray_lines)
MIN_FIT_OVERVOLTAGE = 1.2
@lru_cache(maxsize=None)
def _load_method_table(meas_type: str) -> Optional[dict]:
"""Return the overlap table (criteria and candidate pairs) of a measurement method, or None if it has none."""
with open(PEAK_OVERLAPS_FILE, "r", encoding="utf-8") as file:
methods = json.load(file)["methods"]
return methods.get(meas_type)
@lru_cache(maxsize=None)
def _load_detector_params(microscope_ID: str) -> Tuple[float, float, float]:
"""Detector resolution parameters (conv_eff, elec_noise, F) from the detector model of a microscope."""
model = calibs.load_detector_model(microscope_ID)
return model.conv_eff, model.elec_noise, model.F
def _separation_sigma(energy_a: float, energy_b: float, detector_params: Tuple[float, float, float]) -> float:
"""Separation of two lines (keV) in units of the RMS detector sigma at their energies."""
sigma_a, sigma_b = (DetectorResponseFunction.det_sigma(en, *detector_params) for en in (energy_a, energy_b))
return abs(energy_a - energy_b) / np.sqrt((sigma_a ** 2 + sigma_b ** 2) / 2)
def _severity(separation_sigma: float, thresholds: Dict[str, float]) -> Optional[str]:
"""Severity category of an overlap, from the thresholds in increasing order (None if not overlapping)."""
for label, threshold in thresholds.items():
if separation_sigma < threshold:
return label
return None
def _is_line_fitted(
energy: float,
beam_energy_keV: Optional[float],
energy_range_keV: Optional[Tuple[float, float]],
) -> bool:
"""Whether a line of given energy (keV) is included in the fit."""
if beam_energy_keV is not None and beam_energy_keV / energy <= MIN_FIT_OVERVOLTAGE:
return False
if energy_range_keV is not None and not energy_range_keV[0] <= energy <= energy_range_keV[1]:
return False
return True
def _get_free_area_lines(
el: str,
beam_energy_keV: Optional[float],
energy_range_keV: Optional[Tuple[float, float]],
free_area_el_lines: Sequence[str],
) -> Dict[str, float]:
"""Fitted lines of an element whose area is fitted freely, with their energy (keV)."""
try:
el_lines = get_el_xray_lines(el)
except (ValueError, KeyError):
return {}
free_lines = {}
for line, info in el_lines.items():
energy = float(info["energy (keV)"])
if not _is_line_fitted(energy, beam_energy_keV, energy_range_keV):
continue
el_line = f"{el}_{line}"
if (el_line in free_area_el_lines
or get_reference_xray_line(el, line, el_lines, XSp_Quantifier.xray_quant_ref_lines) == el_line):
free_lines[line] = energy
return free_lines
def _get_candidate_quant_lines(free_lines: Dict[str, float], beam_energy_keV: Optional[float]) -> List[str]:
"""
Lines that may be used to quantify an element, as selected by XSp_Quantifier among its fitted reference
lines (XSp_Quantifier.get_candidate_quant_lines). Only lines with a free area can have the largest area
among the candidates, as tied lines are weaker than their reference. The final choice among multiple
candidates depends on the fitted peak areas, so all candidates are returned.
"""
ref_lines = {line: en for line, en in free_lines.items() if line in XSp_Quantifier.xray_quant_ref_lines}
if beam_energy_keV is None:
return list(ref_lines)
return XSp_Quantifier.get_candidate_quant_lines(ref_lines, beam_energy_keV)
[docs]
def get_peak_overlaps(
meas_type: str,
els_sample: Iterable[str],
els_substrate: Iterable[str] = (),
beam_energy_keV: Optional[float] = None,
energy_range_keV: Optional[Tuple[float, float]] = None,
microscope_ID: Optional[str] = None,
free_area_el_lines: Optional[Sequence[str]] = None,
) -> List[dict]:
"""
Find the reference lines used to quantify sample elements that overlap with other lines whose area
is fitted freely, which may compromise the quantification of those elements.
Parameters
----------
meas_type : str
Measurement method (e.g. 'EDS'). Methods without overlap criteria return no overlaps.
els_sample, els_substrate : iterable of str
Element symbols of the sample (quantified) and of the substrate (fitted, not quantified).
beam_energy_keV : float, optional
Beam energy. If None, lines are not filtered by overvoltage and all reference lines of an
element are considered as possible quantification lines.
energy_range_keV : (float, float), optional
Energy range of the fitted spectrum. If None, lines are not filtered by energy.
microscope_ID : str, optional
Microscope whose detector calibration defines the resolution. Defaults to the default microscope.
free_area_el_lines : sequence of str, optional
Additional lines whose area is fitted freely (e.g. 'Ge_Lb1'). Defaults to the fitter's default.
Returns
-------
list of dict
One entry per overlapping quantification line, sorted by increasing separation in sigmas, with keys
'element' (sample element whose quantification may be inaccurate), 'line' (its quantification line,
a dict with 'element', 'line', 'energy_keV'), 'severity' (of the closest overlap) and 'overlapping'
(lines with free area overlapping with it, dicts with 'element', 'line', 'energy_keV',
'separation_eV', 'separation_sigma', 'severity', sorted by increasing separation).
"""
table = _load_method_table(meas_type)
if table is None:
return []
detector_params = _load_detector_params(microscope_ID or dflt.microscope_ID)
thresholds = table["severity_thresholds_sigma"]
if free_area_el_lines is None:
free_area_el_lines = DEFAULT_FREE_AREA_EL_LINES
els_sample = list(dict.fromkeys(el for el in els_sample if el))
els_fitted = list(dict.fromkeys(els_sample + [el for el in els_substrate if el]))
free_lines = {el: _get_free_area_lines(el, beam_energy_keV, energy_range_keV, free_area_el_lines)
for el in els_fitted}
overlaps = []
for el in els_sample:
for quant_line in _get_candidate_quant_lines(free_lines[el], beam_energy_keV):
quant_energy = free_lines[el][quant_line]
overlapping = []
for other_el, lines in free_lines.items():
for other_line, other_energy in lines.items():
if (other_el, other_line) == (el, quant_line):
continue
separation_sigma = _separation_sigma(quant_energy, other_energy, detector_params)
severity = _severity(separation_sigma, thresholds)
if severity is None:
continue
overlapping.append({
"element": other_el,
"line": other_line,
"energy_keV": round(other_energy, 4),
"separation_eV": round(abs(quant_energy - other_energy) * 1000, 1),
"separation_sigma": round(float(separation_sigma), 2),
"severity": severity,
})
if not overlapping:
continue
overlapping.sort(key=lambda o: o["separation_sigma"])
overlaps.append({
"element": el,
"line": {"element": el, "line": quant_line, "energy_keV": round(quant_energy, 4)},
"severity": overlapping[0]["severity"],
"overlapping": overlapping,
})
overlaps.sort(key=lambda o: o["overlapping"][0]["separation_sigma"])
return overlaps
[docs]
def affected_elements(overlaps: Sequence[dict]) -> List[str]:
"""Sample elements whose quantification may be compromised by the overlaps."""
return sorted({o["element"] for o in overlaps})
[docs]
def warn_peak_overlaps(
meas_type: str,
els_sample: Iterable[str],
els_substrate: Iterable[str] = (),
beam_energy_keV: Optional[float] = None,
energy_range_keV: Optional[Tuple[float, float]] = None,
microscope_ID: Optional[str] = None,
free_area_el_lines: Optional[Sequence[str]] = None,
) -> List[dict]:
"""Log a warning, enclosed in double separators, listing the overlaps found by get_peak_overlaps."""
overlaps = get_peak_overlaps(
meas_type, els_sample, els_substrate, beam_energy_keV, energy_range_keV, microscope_ID, free_area_el_lines
)
if overlaps:
print_double_separator()
logger.warning(format_peak_overlaps_warning(overlaps))
print_double_separator()
return overlaps