Skip to content
1 change: 1 addition & 0 deletions imap_processing/tests/external_test_data_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,7 @@
("status_test_data_repoint00047.csv", "ultra/data/l1/"),
("voltage_culling_results_repoint00047.csv", "ultra/data/l1/"),
("validate_high_energy_culling_results_repoint00047_v2.csv", "ultra/data/l1/"),
("validate_stat_culling_results_repoint00047.csv", "ultra/data/l1/"),
("de_test_data_repoint00047.csv", "ultra/data/l1/"),
("FM45_UltraFM45Extra_TV_Tests_2024-01-22T0930_20240122T093008.CCSDS", "ultra/data/l0/"),
("ultra45_raw_sc_rawnrgevnt_19840122_00.csv", "ultra/data/l0/"),
Expand Down
44 changes: 44 additions & 0 deletions imap_processing/tests/ultra/unit/test_ultra_l1b_culling.py
Original file line number Diff line number Diff line change
Expand Up @@ -844,6 +844,50 @@ def test_get_poisson_stats():
assert outlier_mask[-1]


@pytest.mark.external_test_data
def test_validate_stat_cull():
"""Validate that statistical-outlier quality flags match expected results."""
# read test data from csv files
xspin = pd.read_csv(TEST_PATH / "extendedspin_test_data_repoint00047.csv")
results_df = pd.read_csv(
TEST_PATH / "validate_stat_culling_results_repoint00047.csv"
)
de_df = pd.read_csv(TEST_PATH / "de_test_data_repoint00047.csv")
de_ds = xr.Dataset(
{
"de_event_met": ("epoch", de_df.event_times.values),
"energy_spacecraft": ("epoch", de_df.energy_spacecraft.values),
"quality_outliers": ("epoch", de_df.quality_outliers.values),
"quality_scattering": ("epoch", de_df.quality_scattering.values),
"ebin": ("epoch", de_df.ebin.values),
}
)
# Use constants from the code to ensure consistency with the actual culling code
spin_bin_size = UltraConstants.SPIN_BIN_SIZE
spin_tbin_edges = get_binned_spins_edges(
xspin.spin_number.values,
xspin.spin_period.values,
xspin.spin_start_time.values,
spin_bin_size,
)
intervals, _, _ = build_energy_bins()
# Get the energy ranges
energy_ranges = get_binned_energy_ranges(intervals)
mask = np.zeros((len(energy_ranges) - 1, len(spin_tbin_edges) - 1), dtype=bool)
flags, con, it, std = flag_statistical_outliers(
de_ds, spin_tbin_edges, energy_ranges, mask, 90
)
expected_qf = results_df.iloc[:, :-3].values.astype(bool)
converge = results_df["converge"].values
iterations = results_df["iterations"].values
std_diff = results_df["std_diff"].values
# check that the flags match the expected results
np.testing.assert_array_equal(flags, ~expected_qf.astype(bool))
np.testing.assert_array_equal(con, converge)
np.testing.assert_array_equal(it, iterations)
np.testing.assert_allclose(std, std_diff, rtol=1e-10)


def test_get_energy_range_flags():
"""Tests get_binned_energy_range_flags function."""
# Get energy bins used at l1c
Expand Down
1 change: 0 additions & 1 deletion imap_processing/ultra/l1b/ultra_l1b_culling.py
Original file line number Diff line number Diff line change
Expand Up @@ -880,7 +880,6 @@ def flag_statistical_outliers(
# Select counts that have not been flagged in any channel.
counts = count_summary[e_idx, ~combined_mask]
std_ratio, _ = get_poisson_stats(counts)
std_diff[e_idx] = std_ratio
if std_ratio < std_threshold:
convergence[e_idx] = True

Expand Down