"""Tests for the silx fitting-engine helpers (ewoksxas.fit.silx_engine).
Includes the replacement Strip/Snip background theories (formerly in
test_backgrounds.py): silx's stock functions cache their last evaluation in
process-wide globals, so ``snip_bg`` crashes with a broadcast error when the
data length changes between fits (a region-of-interest change) and the cache is
shared by every FitManager. Those tests pin the crash regression, the cache
semantics, per-manager isolation, and exact parity with the stock silx output.
"""
import copy
import numpy as np
import pytest
from scipy.integrate import trapezoid
from silx.math.fit import bgtheories, functions
from silx.math.fit.fitmanager import FitManager
from ewoksxas.fit.silx_engine import (
CURATED_THEORIES,
SUPPORTED_BACKGROUNDS,
_anchor_indices,
area_from_height,
background_curve,
descriptors_from_fit_results,
make_fit_manager,
snip_theory,
strip_theory,
)
# A very wide, fine grid so the numerical integral captures the heavy tails.
_WIDE_X = np.linspace(-500.0, 500.0, 2_000_001, dtype=np.float64)
_H, _FWHM, _LOW, _HIGH, _ETA = 8.0, 4.0, 4.0, 6.0, 0.4
[docs]
def test_make_fit_manager_registers_curated_theories():
manager = make_fit_manager()
# Only the curated peak theories are exposed (no Hypermet/Step/...).
assert set(manager.theories) == set(CURATED_THEORIES)
# All silx backgrounds remain available on the headless manager.
assert set(SUPPORTED_BACKGROUNDS) <= set(manager.bgtheories)
[docs]
def test_make_fit_manager_curated_backgrounds():
manager = make_fit_manager(curated_backgrounds=True)
assert set(manager.bgtheories) == set(SUPPORTED_BACKGROUNDS)
[docs]
@pytest.mark.parametrize("curated", [False, True])
def test_make_fit_manager_registers_isolated_stripping_backgrounds(curated):
# Strip/Snip are replaced by the safely cached strip_theory/snip_theory on
# every manager (the stock silx ones crash on data-length changes and share
# their cache process-wide).
manager = make_fit_manager(curated_backgrounds=curated)
strip, snip = manager.bgtheories["Strip"], manager.bgtheories["Snip"]
assert strip.parameters == ["StripWidth", "StripIterations"]
assert snip.parameters == ["SnipWidth"]
# silx's estimate/configure are reused so the config keys keep working...
assert strip.estimate is bgtheories.estimate_strip
assert snip.estimate is bgtheories.estimate_snip
assert strip.configure is bgtheories.configure
assert snip.configure is bgtheories.configure
# ...but the functions (and so the caches) are per-manager, not silx's.
assert strip.function is not bgtheories.THEORY["Strip"].function
assert snip.function is not bgtheories.THEORY["Snip"].function
# The background names keep silx's combo order for the FitWidget.
stock_order = [
name for name in FitManager().bgtheories if name in manager.bgtheories
]
assert list(manager.bgtheories) == stock_order
def _fitted_manager(y, background="No Background"):
x = np.linspace(-20.0, 20.0, 1601, dtype=np.float64)
manager = make_fit_manager()
manager.settheory("Gaussians")
manager.setbackground(background)
manager.configure(Sensitivity=2.5, AutoFwhm=False, FwhmPoints=20)
manager.setdata(x=x, y=y(x))
manager.estimate()
manager.runfit()
return manager
[docs]
def test_descriptors_from_fit_results_single_gaussian():
manager = _fitted_manager(lambda x: functions.sum_gauss(x, _H, 2.0, _FWHM))
peaks = descriptors_from_fit_results(manager.fit_results, "gaussian", False)
assert len(peaks) == 1
peak = peaks[0]
assert peak["center"] == pytest.approx(2.0, abs=1e-3)
assert peak["height"] == pytest.approx(_H, rel=1e-3)
assert peak["fwhm"] == pytest.approx(_FWHM, abs=1e-2)
assert peak["area"] == pytest.approx(
area_from_height("gaussian", _H, fwhm=_FWHM), rel=1e-3
)
assert {"fwhm_low", "fwhm_high", "eta"}.isdisjoint(peak)
[docs]
def test_background_curve_constant():
manager = _fitted_manager(
lambda x: functions.sum_gauss(x, _H, 2.0, _FWHM) + 5.0, background="Constant"
)
background = background_curve(manager)
np.testing.assert_allclose(background, 5.0, atol=1e-3)
# The background parameters are excluded from the peak descriptors.
peaks = descriptors_from_fit_results(manager.fit_results, "gaussian", False)
assert len(peaks) == 1
# ---------------------------------------------------------------------------
# Replacement Strip/Snip background theories (strip_theory / snip_theory).
# ---------------------------------------------------------------------------
# (theory factory, args of a first call, args of a changed-settings call)
_THEORIES = [
pytest.param(strip_theory, (200, 5000), (100, 5000), id="strip"),
pytest.param(snip_theory, (300,), (200,), id="snip"),
]
@pytest.fixture(autouse=True)
def _restore_silx_config():
"""Snapshot and restore the shared silx background CONFIG around each test."""
snapshot = copy.deepcopy(bgtheories.CONFIG)
yield
bgtheories.CONFIG.clear()
bgtheories.CONFIG.update(snapshot)
[docs]
@pytest.fixture
def y(x, peak_factory):
peak, _ = peak_factory("gaussian")
return peak + 5.0 + 0.1 * x
[docs]
@pytest.mark.parametrize(("factory", "args", "_changed"), _THEORIES)
def test_function_survives_length_change(x, y, factory, args, _changed):
# Regression: silx's stock snip_bg raises "operands could not be broadcast
# together" when called with a shorter array than the cached one.
function = factory().function
full = function(x, y, *args)
short = function(x[:561], y[:561], *args)
assert full.shape == x.shape
assert short.shape == (561,)
[docs]
@pytest.mark.parametrize(("factory", "args", "changed"), _THEORIES)
def test_cache_hit_returns_cached_background(x, y, factory, args, changed):
function = factory().function
first = function(x, y, *args)
assert function(x, y, *args) is first
# A changed width (and a changed y) each invalidate the cache.
assert function(x, y, *changed) is not first
assert function(x, y + 1.0, *args) is not first
[docs]
def test_strip_cache_invalidated_by_threshold_factor(x, y):
# silx's stock cache key omits StripThresholdFactor; ours must not.
function = strip_theory().function
first = function(x, y, 2, 500)
bgtheories.CONFIG["StripThresholdFactor"] = 2.0
assert function(x, y, 2, 500) is not first
[docs]
@pytest.mark.parametrize(
("config", "offset"),
[
({}, 0.0),
({"SmoothingFlag": True, "SmoothingWidth": 5}, 0.5),
({"AnchorsFlag": True, "AnchorsList": [-5.0, 9.0]}, 1.0),
],
ids=["plain", "smoothing", "anchors"],
)
def test_functions_match_silx_stock_output(x, y, config, offset):
# Behaviour parity: the replacements are bit-identical to silx's stock
# strip_bg/snip_bg under the same CONFIG. The data is offset per case so
# the stock cache cannot hit: its smoothing/anchors cache globals are
# shared between strip and snip, so a preceding strip call would otherwise
# make snip return its stale (differently-configured) cached background.
bgtheories.CONFIG.update(config)
y = y + offset
np.testing.assert_array_equal(
strip_theory().function(x, y, 10, 100), bgtheories.strip_bg(x, y, 10, 100)
)
np.testing.assert_array_equal(
snip_theory().function(x, y, 30), bgtheories.snip_bg(x, y, 30)
)
[docs]
@pytest.mark.parametrize(
("flag", "anchors"),
[
(False, (1.0,)), # disabled -> None
(True, ()), # empty -> None
(True, (-25.0,)), # anchor at/before x[0] is skipped -> None
(True, (-5.0, 9.0)), # plain anchors
(True, (-25.0, 0.0)), # mixed: out-of-range anchor skipped
],
)
def test_anchor_indices_matches_silx(x, flag, anchors):
bgtheories.CONFIG["AnchorsFlag"] = flag
bgtheories.CONFIG["AnchorsList"] = list(anchors)
expected = bgtheories._convert_anchors_to_indices(x)
assert _anchor_indices(x, flag, anchors) == expected
[docs]
def test_cross_manager_cache_isolation(x, y):
# Each manager carries its own function objects (and so its own cache),
# distinct from silx's shared module-level theory.
first_manager, second_manager = make_fit_manager(), make_fit_manager()
first = first_manager.bgtheories["Snip"].function
second = second_manager.bgtheories["Snip"].function
assert first is not second
assert first is not bgtheories.THEORY["Snip"].function
# A different-length fit on one manager must not disturb the other.
reference = first(x, y, 300)
second(x[:561], y[:561], 300)
again = first(x, y, 300)
assert again.shape == x.shape
np.testing.assert_array_equal(again, reference)