Skip IDEX low-rate waveform fits below 4.0 rms

imap_processing/idex/idex_l2a.py

@@ -1003,6 +1003,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 4 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
@@ -1036,6 +1040,13 @@ def estimate_dust_mass(
     """
     signal = np.array(target_signal.data)
     time = np.array(low_sampling_time.data)
+    null_result = (
+        np.full(5, np.nan),
+        np.nan,
+        np.nan,
+        np.nan,
+        np.full(time.shape, np.nan),
+    )
     if baseline_mean is None:
         baseline, _ = calculate_baseline_and_rms(signal, time)
         baseline_mean = float(baseline)
@@ -1060,9 +1071,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)
@@ -1084,6 +1093,9 @@ 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 null_result
+
     try:
         with np.errstate(invalid="ignore", over="ignore"):
             param, _ = curve_fit(
@@ -1101,13 +1113,7 @@ def estimate_dust_mass(
             f"Signal range: {min(signal):.2f} to {max(signal):.2f}\n"
             "Returning None."
         )
-        return (
-            np.full(len(p0), np.nan),
-            np.nan,
-            np.nan,
-            np.nan,
-            np.full_like(time, np.nan),
-        )
+        return null_result
 
     impact_fit = fit_impact(time, *param)
     # Evaluate the analytic extremum of the fitted pulse instead of relying on the
@@ -1173,6 +1179,41 @@ def fit_impact(
     )
 
 
+def _has_low_rate_signal(time: NDArray, signal: NDArray, channel_name: str) -> bool:
+    """
+    Check for a sample above 4 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.
+    """
+    noise_mean, noise_rms = calculate_baseline_and_rms(signal, time)
+    if not (np.isfinite(noise_mean) and np.isfinite(noise_rms)):
+        return False
+
+    # Only finite samples outside the shared statistics' first 3 microseconds
+    # can establish the presence of a pulse.
+    finite_time = np.isfinite(time)
+    window_start = float(np.min(time[finite_time]))
+    signal_window = finite_time & (time >= window_start + 3.0) & np.isfinite(signal)
+    excursions = signal[signal_window] - noise_mean
+    if channel_name == "Ion_Grid":
+        excursions = np.abs(excursions)
+    return bool(np.any(excursions > 4.0 * noise_rms))
+
+
 def remove_signal_noise(
     time: np.ndarray, signal: np.ndarray, good_mask: np.ndarray
 ) -> NDArray:

imap_processing/tests/idex/test_idex_l2a.py

@@ -804,6 +804,10 @@ def test_estimate_dust_mass_reuses_baseline():
     time = xr.DataArray(np.arange(-6.0, 6.0))
     signal = xr.DataArray(np.arange(12.0))
     with (
+        # Isolate fit initialization; the threshold gate computes its own RMS.
+        mock.patch(
+            "imap_processing.idex.idex_l2a._has_low_rate_signal", return_value=True
+        ),
         mock.patch(
             "imap_processing.idex.idex_l2a.calculate_baseline_and_rms",
             side_effect=AssertionError("L1b baseline must not be recalculated"),
@@ -843,9 +847,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))
@@ -854,14 +858,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))
@@ -870,9 +876,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():