from typing import Any
import numpy as np
from ewoksorange.tests.utils import execute_task
from Orange.data import Table
from silx.resources import resource_filename
from ewoksxas.converters.orange import Converter
from ewoksxas.tasks.normalize import Normalize
[docs]
def post_edge_avg(
table: Table, e0: float | None = None, decimals: int | None = None
) -> float:
energy, mu = Converter.from_table(table).features
if e0 is None:
e0: float = Converter.from_table(table).get_meta("e0").mean()
_indexes = np.argwhere(energy * 1000 > e0 + 15).flatten()
if decimals is not None:
return np.round(mu[:, _indexes].mean(), decimals=decimals)
return mu[:, _indexes].mean()
[docs]
def pre_edge_avg(
table: Table, e0: float | None = None, decimals: int | None = None
) -> float:
energy, mu = Converter.from_table(table).features
if e0 is None:
e0: float = Converter.from_table(table).get_meta("e0").mean()
_indexes = np.argwhere(energy * 1000 < e0 - 15).flatten()
if decimals is not None:
return np.round(mu[:, _indexes].mean(), decimals=decimals)
return mu[:, _indexes].mean()
[docs]
def test_raw_data(fe_foil_data):
"""Verify the expected dataset is loaded and is not already normalized.
Compare against a stored snapshot with a tolerance. ReadScans interpolates
repeat scans with np.interp, whose last-bit output varies across operating
systems, so a byte-exact identity check is not portable. Regenerate with
``python -m ewoksxas.resources.data.regenerate_reference fe_foil_raw_reference``
if the dataset legitimately changes.
"""
reference = np.load(
resource_filename("ewoksxas:data/snapshots/fe_foil_raw_reference.npz")
)
energy, mu = Converter.from_table(fe_foil_data).features
np.testing.assert_allclose(energy, reference["energy"], rtol=1e-6, atol=1e-8)
np.testing.assert_allclose(mu, reference["mu"], rtol=1e-6, atol=1e-8)
pre_edge_raw = pre_edge_avg(fe_foil_data, e0=7112, decimals=3)
post_edge_raw = post_edge_avg(fe_foil_data, e0=7112, decimals=3)
assert pre_edge_raw == 2.034
assert post_edge_raw == 2.355
[docs]
def test_normalize(fe_foil_data):
"""Test normalize task with default parameters."""
# Larch default parameters.
inputs: dict[str, Table | dict] = {"Data": fe_foil_data}
output = execute_task(Normalize, inputs=inputs)["Data"]
e0 = round(Converter.from_table(output).get_meta("e0").mean())
assert e0 == 7112
pre_edge_norm = pre_edge_avg(output, decimals=2)
post_edge_norm = post_edge_avg(output, decimals=2)
assert pre_edge_norm == 0.000
assert post_edge_norm == 1.000
[docs]
def test_preedge_params(fe_foil_data):
"""Test normalize task changing only pre1 and pre2."""
# Case 1: normal case (parameters as realistic choices).
params: dict[str, Any] = {"pre1": -100, "pre2": -10}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
output_1: Table = execute_task(Normalize, inputs=inputs)["Data"]
pre_edge_norm = pre_edge_avg(output_1, decimals=3)
post_edge_norm = post_edge_avg(output_1, decimals=3)
assert pre_edge_norm == 0.0
assert post_edge_norm == 1.0
# Case 2: extreme case (parameters skew pre-edge).
params: dict[str, Any] = {"pre1": -2, "pre2": 0}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
output_2: Table = execute_task(Normalize, inputs=inputs)["Data"]
pre_edge_norm = pre_edge_avg(output_2, decimals=3)
post_edge_norm = post_edge_avg(output_2, decimals=3)
assert pre_edge_norm == 3.762 # skewed pre-edge.
assert post_edge_norm == 1.0
# Case 3: accidental swap pre2 < pre1 (should be identical to case 1).
params: dict[str, Any] = {"pre1": -10, "pre2": -100}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
output_3: Table = execute_task(Normalize, inputs=inputs)["Data"]
np.testing.assert_array_equal(output_3.X, output_1.X)
[docs]
def test_postedge_params(fe_foil_data):
"""Test normalize task changing only norm1 and norm2."""
# Case 1: normal case (parameters as realistic choices).
params: dict[str, Any] = {"norm1": 45, "norm2": 1000}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
output_1: Table = execute_task(Normalize, inputs=inputs)["Data"]
pre_edge_norm = pre_edge_avg(output_1, decimals=2)
post_edge_norm = post_edge_avg(output_1, decimals=2)
assert pre_edge_norm == 0.0
assert post_edge_norm == 1.0
# Case 2: extreme case (parameters skew post-edge).
params: dict[str, Any] = {"norm1": 45, "norm2": 100}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
output_2: Table = execute_task(Normalize, inputs=inputs)["Data"]
pre_edge_norm = pre_edge_avg(output_2, decimals=2)
post_edge_norm = post_edge_avg(output_2, decimals=2)
assert pre_edge_norm == 0.0
assert post_edge_norm == 1.61 # skewed post-edge.
# Case 3: accidental swap norm1 > norm2 (should be identical to case 1).
params: dict[str, Any] = {"norm1": 1000, "norm2": 45}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
output_3: Table = execute_task(Normalize, inputs=inputs)["Data"]
np.testing.assert_array_equal(output_3.X, output_1.X)
[docs]
def test_make_flat_normal(fe_foil_data):
"""Test make_flat == False with normal / expected parameters."""
# All other parameters as default.
params: dict[str, Any] = {"make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
output_1: Table = execute_task(Normalize, inputs=inputs)["Data"]
pre_edge_norm = pre_edge_avg(output_1, decimals=2)
post_edge_norm = post_edge_avg(output_1, decimals=2)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.91 # off-set from flattened data.
# Normal pre1 and pre2.
params: dict[str, Any] = {"pre1": -100, "pre2": -10, "make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
output_2: Table = execute_task(Normalize, inputs=inputs)["Data"]
pre_edge_norm = pre_edge_avg(output_2, decimals=2)
post_edge_norm = post_edge_avg(output_2, decimals=2)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.87
# Normal norm1 and norm2.
params: dict[str, Any] = {"norm1": 45, "norm2": 1000, "make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
output_3: Table = execute_task(Normalize, inputs=inputs)["Data"]
pre_edge_norm = pre_edge_avg(output_3, decimals=2)
post_edge_norm = post_edge_avg(output_3, decimals=2)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.9
[docs]
def test_make_flat_extreme(fe_foil_data):
"""Test make_flat == False with bad / unrealistic / unexpected parameters."""
# Bad pre1 and pre2 amplified by make_flat == False.
params: dict[str, Any] = {"pre1": -2, "pre2": 0, "make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 5.8 # skewed pre-edge.
assert post_edge_norm == -64.8 # skewed post-edge.
# Bad norm1 and norm2 not affected by make_flat == False.
params: dict[str, Any] = {"norm1": 45, "norm2": 100, "make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.9 # skewed post-edge.
[docs]
def test_energy_exponent_doesnt_affect_flattened_array(fe_foil_data):
"""Skewed post-edge by nvict is accounted for by the make_flat subtraction."""
arrays = []
for exp in range(4):
params: dict[str, Any] = {"nvict": exp}
output = execute_task(
Normalize, inputs={"Data": fe_foil_data, "parameters": params}
)
arrays.append(output["Data"].X.mean(axis=0))
for i in range(len(arrays) - 1):
assert np.allclose(arrays[i], arrays[i + 1], atol=0.01)
[docs]
def test_energy_exponent_normal(fe_foil_data):
"""Test nvict with normal / expected parameters, make_flat == False."""
params: dict[str, Any] = {"nvict": 1, "make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.8 # off-set from flattened data.
params: dict[str, Any] = {"nvict": 2, "make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.7 # off-set from flattened data.
params: dict[str, Any] = {"nvict": 3, "make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.6 # off-set from flattened data.
# Normal pre1 and pre2.
params: dict[str, Any] = {"pre1": -100, "pre2": -10, "make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.9
# Normal norm1 and norm2.
params: dict[str, Any] = {"norm1": 45, "norm2": 1000, "make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.9
[docs]
def test_energy_exponent_extreme(fe_foil_data):
"""Test nvict with bad / unexpected parameters, make_flat == False."""
# Bad pre1 and pre2 amplified by make_flat == False and nvict > 0.
params: dict[str, Any] = {"pre1": -2, "pre2": 0, "make_flat": False, "nvict": 3}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 9.7 # skewed pre-edge, amplified by nvict.
assert post_edge_norm == -67.9 # skewed post-edge, amplified by nvict.
# Bad norm1 and norm2 with make_flat == False.
params: dict[str, Any] = {"norm1": 45, "norm2": 100, "make_flat": False, "nvict": 3}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.6
# Bad norm1 and norm2 with make_flat == True (default).
params: dict[str, Any] = {"norm1": 45, "norm2": 100, "nvict": 3}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 2.1
[docs]
def test_postedge_order_normal(fe_foil_data):
"""Test nnorm with normal / expected parameters."""
params: dict[str, Any] = {"nnorm": 0}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 1.0 # off-set from flattened data.
params: dict[str, Any] = {"nnorm": 1}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 1.1 # off-set from flattened data.
params: dict[str, Any] = {"nnorm": 2}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 1.0 # off-set from flattened data.
params: dict[str, Any] = {"nnorm": 0, "make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 1.0 # off-set from flattened data.
params: dict[str, Any] = {"nnorm": 1, "make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 1.0 # off-set from flattened data.
params: dict[str, Any] = {"nnorm": 2, "make_flat": False}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.9 # off-set from flattened data.
[docs]
def test_postedge_order_extreme(fe_foil_data):
"""Postedge order changes post-edge line-shape, strongly affected by bad limits."""
# Bad norm1 and norm2, nnorm=0.
params: dict[str, Any] = {"norm1": 45, "norm2": 100, "nnorm": 0}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 1.0
# Bad norm1 and norm2, nnorm=1.
params: dict[str, Any] = {"norm1": 45, "norm2": 100, "nnorm": 1}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 2.4
# Bad norm1 and norm2, nnorm=2.
params: dict[str, Any] = {"norm1": 45, "norm2": 100, "nnorm": 2}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == -19.7
# Bad norm1 and norm2 with make_flat == False, nnorm=0.
params: dict[str, Any] = {"norm1": 45, "norm2": 100, "make_flat": False, "nnorm": 0}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 1.0
# Bad norm1 and norm2 with make_flat == False, nnorm=1.
params: dict[str, Any] = {"norm1": 45, "norm2": 100, "make_flat": False, "nnorm": 1}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.9
# Bad norm1 and norm2 with make_flat == False, nnorm=2.
params: dict[str, Any] = {"norm1": 45, "norm2": 100, "make_flat": False, "nnorm": 2}
inputs: dict[str, Table | dict] = {"Data": fe_foil_data, "parameters": params}
outputs = execute_task(Normalize, inputs=inputs)
out_data: Table = outputs["Data"]
pre_edge_norm: np.ndarray = np.round(out_data.X[:, :100].mean(), decimals=1)
post_edge_norm: np.ndarray = np.round(out_data.X[:, -100:].mean(), decimals=1)
assert pre_edge_norm == 0.0
assert post_edge_norm == 0.8