From ce56bcaf24f326f7bc5a97d8a6e06fc7dca0435f Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 27 Oct 2026 10:09:05 -0600 Subject: [PATCH 1/3] Skip IDEX low-rate waveform fits below 1.5 sigma --- imap_processing/idex/idex_l2a.py | 55 ++++- imap_processing/tests/idex/test_idex_l2a.py | 20 +- .../tests/idex/test_idex_signal_threshold.py | 203 ++++++++++++++++++ 3 files changed, 267 insertions(+), 11 deletions(-) create mode 100644 imap_processing/tests/idex/test_idex_signal_threshold.py diff --git a/imap_processing/idex/idex_l2a.py b/imap_processing/idex/idex_l2a.py index 0c3eff6b7a..d8b4734b0c 100644 --- a/imap_processing/idex/idex_l2a.py +++ b/imap_processing/idex/idex_l2a.py @@ -1007,6 +1007,10 @@ def estimate_dust_mass( """ Filter and fit the target or ion grid signals to get the total dust impact charge. + Fit only when a sample after the first 3 microseconds exceeds 1.5 times the + baseline RMS noise. Target excursions must be positive; Ion Grid excursions + may have either polarity. Skipped fits return NaN products. + Parameters ---------- low_sampling_time : xarray.DataArray @@ -1070,9 +1074,7 @@ def estimate_dust_mass( amplitude = float("nan") if channel_name != "Ion_Grid" and (not np.isfinite(amplitude) or amplitude <= 0.0): amplitude = float(np.max(signal) - baseline_mean) - if not np.isfinite(amplitude): - amplitude = float(np.max(signal)) - if channel_name != "Ion_Grid" and amplitude <= 0.0: + if not np.isfinite(amplitude) or (channel_name != "Ion_Grid" and amplitude <= 0.0): amplitude = float(np.max(signal)) rise_time_0 = 0.371 # How fast the signal rises (us) @@ -1094,6 +1096,15 @@ def estimate_dust_mass( [np.inf, np.inf, amplitude_upper_bound, np.inf, np.inf], ) + if not _has_low_rate_signal(time, signal, channel_name): + return ( + np.full(5, np.nan), + np.nan, + np.nan, + np.nan, + np.full(time.shape, np.nan), + ) + try: with np.errstate(invalid="ignore", over="ignore"): param, _ = curve_fit( @@ -1183,6 +1194,44 @@ def fit_impact( ) +def _has_low_rate_signal(time: NDArray, signal: NDArray, channel_name: str) -> bool: + """ + Check for a sample above 1.5 times the baseline RMS noise. + + Parameters + ---------- + time : numpy.ndarray + Sample times in microseconds. + signal : numpy.ndarray + Target or Ion Grid waveform in engineering units. + channel_name : str + Waveform name. Ion Grid permits either polarity; targets require a + positive excursion from the baseline. + + Returns + ------- + bool + Whether a finite sample after the first 3 microseconds exceeds the + threshold. False when no finite baseline samples are available. + """ + finite_time = np.isfinite(time) + window_start = float(np.min(time[finite_time], initial=np.inf)) + noise_window = finite_time & (time < window_start + 3.0) + finite_signal = np.isfinite(signal) + noise_samples = signal[noise_window & finite_signal] + if not noise_samples.size: + return False + + # Population RMS matches the baseline statistics; only samples outside that + # baseline window can establish the presence of a pulse. + noise_mean = float(np.mean(noise_samples)) + noise_rms = float(np.std(noise_samples, ddof=0)) + excursions = signal[finite_time & ~noise_window & finite_signal] - noise_mean + if channel_name == "Ion_Grid": + excursions = np.abs(excursions) + return bool(np.any(excursions > 1.5 * noise_rms)) + + def remove_signal_noise( time: np.ndarray, signal: np.ndarray, good_mask: np.ndarray ) -> NDArray: diff --git a/imap_processing/tests/idex/test_idex_l2a.py b/imap_processing/tests/idex/test_idex_l2a.py index dba158713c..d9c7baef73 100644 --- a/imap_processing/tests/idex/test_idex_l2a.py +++ b/imap_processing/tests/idex/test_idex_l2a.py @@ -737,9 +737,9 @@ def test_estimate_dust_mass_remove_noise_logs_debug(caplog): ) -def test_estimate_dust_mass_nonfinite_signal_fallbacks(): +def test_estimate_dust_mass_skips_nonfinite_signal(): """ - Test fallback handling when the input signal is entirely non-finite. + Test that an entirely non-finite waveform is skipped without fitting. """ time = xr.DataArray(np.linspace(-60, 60, 16)) signal = xr.DataArray(np.full(16, np.nan)) @@ -748,14 +748,16 @@ def test_estimate_dust_mass_nonfinite_signal_fallbacks(): "imap_processing.idex.idex_l2a.curve_fit", return_value=(np.array([0.0, 0.0, 1.0, 0.371, 37.1]), None), ) as mocked_curve_fit: - estimate_dust_mass(time, signal) + result = estimate_dust_mass(time, signal) - assert np.isnan(mocked_curve_fit.call_args.kwargs["p0"][2]) + mocked_curve_fit.assert_not_called() + for value in result: + assert np.all(np.isnan(value)) -def test_estimate_dust_mass_non_ion_grid_negative_amplitude_fallback(): +def test_estimate_dust_mass_skips_constant_negative_signal(): """ - Test the non-Ion_Grid negative-amplitude fallback path. + Test that a constant negative baseline does not qualify as a signal. """ time = xr.DataArray(np.linspace(-60, 60, 16)) signal = xr.DataArray(np.full(16, -2.0)) @@ -764,9 +766,11 @@ def test_estimate_dust_mass_non_ion_grid_negative_amplitude_fallback(): "imap_processing.idex.idex_l2a.curve_fit", return_value=(np.array([0.0, -2.0, -2.0, 0.371, 37.1]), None), ) as mocked_curve_fit: - estimate_dust_mass(time, signal) + result = estimate_dust_mass(time, signal) - assert mocked_curve_fit.call_args.kwargs["p0"][2] == -2.0 + mocked_curve_fit.assert_not_called() + for value in result: + assert np.all(np.isnan(value)) def test_estimate_dust_mass_ion_grid_negative_amplitude_bounds(): diff --git a/imap_processing/tests/idex/test_idex_signal_threshold.py b/imap_processing/tests/idex/test_idex_signal_threshold.py new file mode 100644 index 0000000000..9f1996a6c0 --- /dev/null +++ b/imap_processing/tests/idex/test_idex_signal_threshold.py @@ -0,0 +1,203 @@ +"""Test the signal threshold for Target and Ion Grid waveform fits.""" + +from unittest import mock + +import numpy as np +import pytest +import xarray as xr + +from imap_processing.idex.idex_l2a import ( + estimate_dust_mass, + idex_l2a, + log_smooth_powerlaw, +) + + +@pytest.mark.parametrize("waveform", ["Target_High", "Target_Low", "Ion_Grid"]) +@pytest.mark.parametrize("excursion", [0.0, 1.49, 1.5, 1.51, -1.49, -1.5, -1.51]) +@pytest.mark.parametrize("pulse_index", [6, -1]) +def test_low_rate_fit_requires_sample_above_noise(waveform, excursion, pulse_index): + """One sample must strictly exceed 1.5 sigma, with channel-specific polarity.""" + time = xr.DataArray(np.arange(-6.0, 6.0, 0.5)) + signal = np.full(time.size, 10.0) + # The first 3 microseconds have baseline 10 and population sigma 1. + signal[:6] += [-1.0, 1.0, -1.0, 1.0, -1.0, 1.0] + signal[pulse_index] += excursion + target = xr.DataArray(signal, name=waveform) + parameters = np.array([0.0, 10.0, excursion, 0.371, 37.1]) + + with mock.patch( + "imap_processing.idex.idex_l2a.curve_fit", return_value=(parameters, None) + ) as optimizer: + result = estimate_dust_mass(time, target) + + should_fit = abs(excursion) > 1.5 if waveform == "Ion_Grid" else excursion > 1.5 + if should_fit: + optimizer.assert_called_once() + np.testing.assert_array_equal(result[0], parameters) + assert np.all(np.isfinite(result[4])) + else: + optimizer.assert_not_called() + assert result[0].shape == (5,) + assert result[4].shape == signal.shape + for value in result: + assert np.all(np.isnan(value)) + + +@pytest.mark.parametrize("waveform", ["Target_High", "Target_Low", "Ion_Grid"]) +@pytest.mark.parametrize("excursion", [0.0, 1.0, -1.0]) +def test_low_rate_fit_zero_noise(waveform, excursion): + """A flat baseline requires a nonzero excursion without dividing by zero.""" + time = xr.DataArray(np.arange(-6.0, 6.0, 0.5)) + signal = np.full(time.size, -2.0) + signal[-1] += excursion + with mock.patch( + "imap_processing.idex.idex_l2a.curve_fit", + return_value=(np.array([0.0, -2.0, excursion, 0.371, 37.1]), None), + ) as optimizer: + result = estimate_dust_mass(time, xr.DataArray(signal), waveform_name=waveform) + + should_fit = excursion != 0.0 if waveform == "Ion_Grid" else excursion > 0.0 + assert optimizer.call_count == int(should_fit) + if not should_fit: + for value in result: + assert np.all(np.isnan(value)) + + +@pytest.mark.parametrize("waveform", ["Target_High", "Target_Low", "Ion_Grid"]) +def test_low_rate_fit_does_not_count_baseline_excursions(waveform): + """An excursion confined to the noise window is not evidence of a pulse.""" + time = xr.DataArray(np.arange(-6.0, 6.0, 0.5)) + signal = np.full(time.size, 4.0 / 6.0) + signal[:6] = [0.0, 0.0, 0.0, 0.0, 0.0, 4.0] + with mock.patch("imap_processing.idex.idex_l2a.curve_fit") as optimizer: + result = estimate_dust_mass(time, xr.DataArray(signal), waveform_name=waveform) + + optimizer.assert_not_called() + for value in result: + assert np.all(np.isnan(value)) + + +@pytest.mark.parametrize("missing", ["signal", "time", "baseline"]) +def test_low_rate_fit_requires_finite_noise_samples(missing): + """Missing baseline information cannot establish a significant signal.""" + time = np.arange(-6.0, 6.0, 0.5) + signal = np.zeros(time.size) + signal[-1] = 10.0 + if missing == "signal": + signal[:] = np.nan + elif missing == "time": + time[:] = np.nan + else: + signal[:6] = np.nan + + with mock.patch("imap_processing.idex.idex_l2a.curve_fit") as optimizer: + result = estimate_dust_mass( + xr.DataArray(time), xr.DataArray(signal), waveform_name="Target_High" + ) + + optimizer.assert_not_called() + for value in result: + assert np.all(np.isnan(value)) + + +@pytest.mark.parametrize("waveform", ["Target_High", "Target_Low", "Ion_Grid"]) +def test_low_rate_signal_gate_is_per_event(waveform): + """Vectorized fits skip weak events and keep an independent strong event.""" + time = xr.DataArray(np.arange(-6.0, 6.0, 0.5), dims="sample") + signal = np.zeros((2, time.size)) + signal[:, :6] = [-1.0, 1.0, -1.0, 1.0, -1.0, 1.0] + signal[0, -1] = 1.5 + signal[1, -1] = -2.0 if waveform == "Ion_Grid" else 2.0 + with mock.patch( + "imap_processing.idex.idex_l2a.curve_fit", + return_value=(np.array([0.0, 0.0, 2.0, 0.371, 37.1]), None), + ) as optimizer: + results = xr.apply_ufunc( + estimate_dust_mass, + time, + xr.DataArray(signal, dims=("epoch", "sample")), + input_core_dims=[["sample"], ["sample"]], + output_core_dims=[["parameter"], [], [], [], ["sample"]], + output_dtypes=[np.float64] * 5, + vectorize=True, + kwargs={"waveform_name": waveform}, + ) + + optimizer.assert_called_once() + for values in results: + assert np.all(np.isnan(values.isel(epoch=0))) + assert np.all(np.isfinite(values.isel(epoch=1))) + + +def test_l2a_signal_gate_preserves_independent_channels(): + """Skipped fits propagate NaNs while other channels retain their products.""" + time = np.tile(np.arange(-6.0, 6.0, 0.5), (2, 1)) + dimensions = ("epoch", "time_low_sample_rate_index") + high_dimensions = ("epoch", "time_high_sample_rate_index") + dataset = xr.Dataset( + { + "time_low_sample_rate": (dimensions, time), + "time_high_sample_rate": (high_dimensions, time - 1.0), + "TOF_High": (high_dimensions, np.tile([-1.0, 1.0], (2, 12))), + "science_event_flag": ("epoch", [1, 1]), + "tof_high_saturation_flag": ("epoch", [0, 0]), + }, + coords={"epoch": [0, 1]}, + attrs={"Logical_source": "imap_idex_l1b_sci-10days"}, + ) + strong_events = {"Target_High": [0], "Target_Low": [1], "Ion_Grid": [0, 1]} + for waveform, events in strong_events.items(): + signal = np.zeros_like(time) + signal[:, :6] = [-1.0, 1.0, -1.0, 1.0, -1.0, 1.0] + signal[events, -1] = 10.0 + if waveform == "Ion_Grid": + signal[0, -1] = -10.0 + dataset[waveform] = xr.DataArray(signal, dims=dimensions) + dataset[f"{waveform.lower()}_saturation_flag"] = ("epoch", [0, 0]) + + t_rise_params = np.array([1.27, -0.2, -2.1, -0.37, 5.3, 13.3, 13.3, 0.28]) + rise_time = 10 ** log_smooth_powerlaw(1.0, t_rise_params[0], t_rise_params[1:]) + + def fit_waveform(function, time, signal, p0, **kwargs): + parameters = np.asarray(p0) + parameters[3] = rise_time + return parameters, None + + with ( + mock.patch( + "imap_processing.idex.idex_l2a.load_calibration_files", + return_value=( + t_rise_params, + np.array([0.06, 2.8, 5.9, 4.1, 13.0, 22.7, 8.2, 0.40]), + ), + ), + mock.patch( + "imap_processing.idex.idex_l2a.time_to_mass", + return_value=(None, None, np.zeros_like(time)), + ), + mock.patch( + "imap_processing.idex.idex_l2a.analyze_peaks", + return_value=(np.zeros((2, 3)), np.ones(2), np.zeros(2), np.zeros(2)), + ), + mock.patch( + "imap_processing.idex.idex_l2a.curve_fit", side_effect=fit_waveform + ) as optimizer, + ): + result = idex_l2a(dataset, {}) + + assert optimizer.call_count == 4 + for waveform, events in strong_events.items(): + for product in ( + "fit_parameters", + "impact_charge", + "chi_squared", + "reduced_chi_squared", + "fit_results", + "velocity_estimate", + "dust_mass_estimate", + ): + values = result[f"{waveform.lower()}_{product}"] + assert np.all(np.isfinite(values.isel(epoch=events))) + weak_events = [event for event in range(2) if event not in events] + assert np.all(np.isnan(values.isel(epoch=weak_events))) From 2782cb827bae1d7f2bed5e2cdfe8616dca3a07ea Mon Sep 17 00:00:00 2001 From: Alex Date: Mon, 2 Nov 2026 13:33:42 -0700 Subject: [PATCH 2/3] Raise IDEX low-rate fit threshold to 3.5 RMS --- imap_processing/idex/idex_l2a.py | 6 +++--- .../tests/idex/test_idex_signal_threshold.py | 21 ++++++++++++------- 2 files changed, 16 insertions(+), 11 deletions(-) diff --git a/imap_processing/idex/idex_l2a.py b/imap_processing/idex/idex_l2a.py index d8b4734b0c..7d99995295 100644 --- a/imap_processing/idex/idex_l2a.py +++ b/imap_processing/idex/idex_l2a.py @@ -1007,7 +1007,7 @@ def estimate_dust_mass( """ Filter and fit the target or ion grid signals to get the total dust impact charge. - Fit only when a sample after the first 3 microseconds exceeds 1.5 times the + Fit only when a sample after the first 3 microseconds exceeds 3.5 times the baseline RMS noise. Target excursions must be positive; Ion Grid excursions may have either polarity. Skipped fits return NaN products. @@ -1196,7 +1196,7 @@ def fit_impact( def _has_low_rate_signal(time: NDArray, signal: NDArray, channel_name: str) -> bool: """ - Check for a sample above 1.5 times the baseline RMS noise. + Check for a sample above 3.5 times the baseline RMS noise. Parameters ---------- @@ -1229,7 +1229,7 @@ def _has_low_rate_signal(time: NDArray, signal: NDArray, channel_name: str) -> b excursions = signal[finite_time & ~noise_window & finite_signal] - noise_mean if channel_name == "Ion_Grid": excursions = np.abs(excursions) - return bool(np.any(excursions > 1.5 * noise_rms)) + return bool(np.any(excursions > 3.5 * noise_rms)) def remove_signal_noise( diff --git a/imap_processing/tests/idex/test_idex_signal_threshold.py b/imap_processing/tests/idex/test_idex_signal_threshold.py index 9f1996a6c0..62151b8e56 100644 --- a/imap_processing/tests/idex/test_idex_signal_threshold.py +++ b/imap_processing/tests/idex/test_idex_signal_threshold.py @@ -14,10 +14,13 @@ @pytest.mark.parametrize("waveform", ["Target_High", "Target_Low", "Ion_Grid"]) -@pytest.mark.parametrize("excursion", [0.0, 1.49, 1.5, 1.51, -1.49, -1.5, -1.51]) +@pytest.mark.parametrize( + "excursion", + [0.0, 1.5, 3.375, 3.49, 3.5, 3.51, 3.88, -1.5, -3.375, -3.49, -3.5, -3.51, -3.88], +) @pytest.mark.parametrize("pulse_index", [6, -1]) def test_low_rate_fit_requires_sample_above_noise(waveform, excursion, pulse_index): - """One sample must strictly exceed 1.5 sigma, with channel-specific polarity.""" + """One sample must strictly exceed 3.5 sigma, with channel-specific polarity.""" time = xr.DataArray(np.arange(-6.0, 6.0, 0.5)) signal = np.full(time.size, 10.0) # The first 3 microseconds have baseline 10 and population sigma 1. @@ -31,7 +34,7 @@ def test_low_rate_fit_requires_sample_above_noise(waveform, excursion, pulse_ind ) as optimizer: result = estimate_dust_mass(time, target) - should_fit = abs(excursion) > 1.5 if waveform == "Ion_Grid" else excursion > 1.5 + should_fit = abs(excursion) > 3.5 if waveform == "Ion_Grid" else excursion > 3.5 if should_fit: optimizer.assert_called_once() np.testing.assert_array_equal(result[0], parameters) @@ -67,9 +70,11 @@ def test_low_rate_fit_zero_noise(waveform, excursion): @pytest.mark.parametrize("waveform", ["Target_High", "Target_Low", "Ion_Grid"]) def test_low_rate_fit_does_not_count_baseline_excursions(waveform): """An excursion confined to the noise window is not evidence of a pulse.""" - time = xr.DataArray(np.arange(-6.0, 6.0, 0.5)) - signal = np.full(time.size, 4.0 / 6.0) - signal[:6] = [0.0, 0.0, 0.0, 0.0, 0.0, 4.0] + time = xr.DataArray(np.arange(60) / 5.0 - 6.0) + signal = np.full(time.size, 4.0 / 15.0) + # The baseline outlier exceeds 3.5 sigma but must not trigger a fit. + signal[:15] = 0.0 + signal[14] = 4.0 with mock.patch("imap_processing.idex.idex_l2a.curve_fit") as optimizer: result = estimate_dust_mass(time, xr.DataArray(signal), waveform_name=waveform) @@ -107,8 +112,8 @@ def test_low_rate_signal_gate_is_per_event(waveform): time = xr.DataArray(np.arange(-6.0, 6.0, 0.5), dims="sample") signal = np.zeros((2, time.size)) signal[:, :6] = [-1.0, 1.0, -1.0, 1.0, -1.0, 1.0] - signal[0, -1] = 1.5 - signal[1, -1] = -2.0 if waveform == "Ion_Grid" else 2.0 + signal[0, -1] = 3.5 + signal[1, -1] = -4.0 if waveform == "Ion_Grid" else 4.0 with mock.patch( "imap_processing.idex.idex_l2a.curve_fit", return_value=(np.array([0.0, 0.0, 2.0, 0.371, 37.1]), None), From a5f4b97b550513870160be32b5e03cc3f30f0e1e Mon Sep 17 00:00:00 2001 From: Alex Date: Mon, 2 Nov 2026 14:00:35 -0700 Subject: [PATCH 3/3] Raise IDEX low-rate fit threshold to 4 RMS --- imap_processing/idex/idex_l2a.py | 6 ++-- .../tests/idex/test_idex_signal_threshold.py | 36 +++++++++++++------ 2 files changed, 29 insertions(+), 13 deletions(-) diff --git a/imap_processing/idex/idex_l2a.py b/imap_processing/idex/idex_l2a.py index 7d99995295..d5440d3889 100644 --- a/imap_processing/idex/idex_l2a.py +++ b/imap_processing/idex/idex_l2a.py @@ -1007,7 +1007,7 @@ def estimate_dust_mass( """ Filter and fit the target or ion grid signals to get the total dust impact charge. - Fit only when a sample after the first 3 microseconds exceeds 3.5 times the + Fit only when a sample after the first 3 microseconds exceeds 4 times the baseline RMS noise. Target excursions must be positive; Ion Grid excursions may have either polarity. Skipped fits return NaN products. @@ -1196,7 +1196,7 @@ def fit_impact( def _has_low_rate_signal(time: NDArray, signal: NDArray, channel_name: str) -> bool: """ - Check for a sample above 3.5 times the baseline RMS noise. + Check for a sample above 4 times the baseline RMS noise. Parameters ---------- @@ -1229,7 +1229,7 @@ def _has_low_rate_signal(time: NDArray, signal: NDArray, channel_name: str) -> b excursions = signal[finite_time & ~noise_window & finite_signal] - noise_mean if channel_name == "Ion_Grid": excursions = np.abs(excursions) - return bool(np.any(excursions > 3.5 * noise_rms)) + return bool(np.any(excursions > 4.0 * noise_rms)) def remove_signal_noise( diff --git a/imap_processing/tests/idex/test_idex_signal_threshold.py b/imap_processing/tests/idex/test_idex_signal_threshold.py index 62151b8e56..adf6a1bb9a 100644 --- a/imap_processing/tests/idex/test_idex_signal_threshold.py +++ b/imap_processing/tests/idex/test_idex_signal_threshold.py @@ -16,11 +16,27 @@ @pytest.mark.parametrize("waveform", ["Target_High", "Target_Low", "Ion_Grid"]) @pytest.mark.parametrize( "excursion", - [0.0, 1.5, 3.375, 3.49, 3.5, 3.51, 3.88, -1.5, -3.375, -3.49, -3.5, -3.51, -3.88], + [ + 0.0, + 1.5, + 3.375, + 3.5, + 3.88, + 3.99, + 4.0, + 4.01, + -1.5, + -3.375, + -3.5, + -3.88, + -3.99, + -4.0, + -4.01, + ], ) @pytest.mark.parametrize("pulse_index", [6, -1]) def test_low_rate_fit_requires_sample_above_noise(waveform, excursion, pulse_index): - """One sample must strictly exceed 3.5 sigma, with channel-specific polarity.""" + """One sample must strictly exceed 4 sigma, with channel-specific polarity.""" time = xr.DataArray(np.arange(-6.0, 6.0, 0.5)) signal = np.full(time.size, 10.0) # The first 3 microseconds have baseline 10 and population sigma 1. @@ -34,7 +50,7 @@ def test_low_rate_fit_requires_sample_above_noise(waveform, excursion, pulse_ind ) as optimizer: result = estimate_dust_mass(time, target) - should_fit = abs(excursion) > 3.5 if waveform == "Ion_Grid" else excursion > 3.5 + should_fit = abs(excursion) > 4.0 if waveform == "Ion_Grid" else excursion > 4.0 if should_fit: optimizer.assert_called_once() np.testing.assert_array_equal(result[0], parameters) @@ -70,11 +86,11 @@ def test_low_rate_fit_zero_noise(waveform, excursion): @pytest.mark.parametrize("waveform", ["Target_High", "Target_Low", "Ion_Grid"]) def test_low_rate_fit_does_not_count_baseline_excursions(waveform): """An excursion confined to the noise window is not evidence of a pulse.""" - time = xr.DataArray(np.arange(60) / 5.0 - 6.0) - signal = np.full(time.size, 4.0 / 15.0) - # The baseline outlier exceeds 3.5 sigma but must not trigger a fit. - signal[:15] = 0.0 - signal[14] = 4.0 + time = xr.DataArray(np.arange(96) / 8.0 - 6.0) + signal = np.full(time.size, 4.0 / 24.0) + # The baseline outlier exceeds 4 sigma but must not trigger a fit. + signal[:24] = 0.0 + signal[23] = 4.0 with mock.patch("imap_processing.idex.idex_l2a.curve_fit") as optimizer: result = estimate_dust_mass(time, xr.DataArray(signal), waveform_name=waveform) @@ -112,8 +128,8 @@ def test_low_rate_signal_gate_is_per_event(waveform): time = xr.DataArray(np.arange(-6.0, 6.0, 0.5), dims="sample") signal = np.zeros((2, time.size)) signal[:, :6] = [-1.0, 1.0, -1.0, 1.0, -1.0, 1.0] - signal[0, -1] = 3.5 - signal[1, -1] = -4.0 if waveform == "Ion_Grid" else 4.0 + signal[0, -1] = 4.0 + signal[1, -1] = -5.0 if waveform == "Ion_Grid" else 5.0 with mock.patch( "imap_processing.idex.idex_l2a.curve_fit", return_value=(np.array([0.0, 0.0, 2.0, 0.371, 37.1]), None),