diff --git a/tests/test_record.py b/tests/test_record.py index 907a929b..ce6d3bee 100644 --- a/tests/test_record.py +++ b/tests/test_record.py @@ -3,6 +3,7 @@ import shutil import tempfile import unittest +import warnings import numpy as np import pandas as pd @@ -1435,6 +1436,72 @@ def test_adc_gain_min_boundary(self): expected_gain = (32767 - (-2147483648)) / (base_value + 1.5) np.testing.assert_allclose(record.adc_gain[0], expected_gain, rtol=1e-3) + def test_all_nan_single_channel(self): + """ + Writing a channel that is entirely NaN should not crash or emit an + "All-NaN slice encountered" warning. Every NaN sample maps to the + format's reserved invalid-sample value, so the channel reads back + as all-NaN. Regression test for + https://github.com/MIT-LCP/wfdb-python/issues/485. + """ + p_signal = np.array([[np.nan], [np.nan], [np.nan], [np.nan]]) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + wfdb.wrsamp( + "test_all_nan_single", + fs=250, + sig_name=["ECG"], + units=["mV"], + p_signal=p_signal, + fmt=["16"], + write_dir=self.temp_path, + ) + self.assertFalse( + [w for w in caught if issubclass(w.category, RuntimeWarning)] + ) + + record = wfdb.rdrecord( + os.path.join(self.temp_path, "test_all_nan_single"), + physical=True, + ) + self.assertTrue(np.all(np.isnan(record.p_signal))) + + def test_all_nan_mixed_channels(self): + """ + A multi-channel signal with one all-NaN channel and one normal + channel should write and read back without crashing or warning. + The all-NaN channel reads back as all-NaN; the normal channel + round-trips within quantization tolerance. Regression test for + https://github.com/MIT-LCP/wfdb-python/issues/485. + """ + normal = np.array([1.0, 2.0, 3.0, 4.0]) + p_signal = np.column_stack([normal, np.full(4, np.nan)]) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + wfdb.wrsamp( + "test_all_nan_mixed", + fs=250, + sig_name=["ECG", "EMPTY"], + units=["mV", "mV"], + p_signal=p_signal, + fmt=["16", "16"], + write_dir=self.temp_path, + ) + self.assertFalse( + [w for w in caught if issubclass(w.category, RuntimeWarning)] + ) + + record = wfdb.rdrecord( + os.path.join(self.temp_path, "test_all_nan_mixed"), + physical=True, + ) + np.testing.assert_allclose( + record.p_signal[:, 0], normal, rtol=1e-4, atol=1e-3 + ) + self.assertTrue(np.all(np.isnan(record.p_signal[:, 1]))) + @classmethod def setUpClass(cls): cls.temp_directory = tempfile.TemporaryDirectory() diff --git a/wfdb/io/_signal.py b/wfdb/io/_signal.py index 5bad70c1..93a1dae6 100644 --- a/wfdb/io/_signal.py +++ b/wfdb/io/_signal.py @@ -2,6 +2,7 @@ import os import posixpath import sys +import warnings import fsspec import numpy as np @@ -746,7 +747,7 @@ def calc_adc_gain_baseline(self, ch, minvals, maxvals): # Figure out digital samples used to store physical samples # If the entire signal is NAN, gain/baseline won't be used - if pmin == np.nan: + if np.isnan(pmin): adc_gain = 1 baseline = 1 # If the signal is just one value, store one digital value. @@ -827,9 +828,12 @@ def calc_adc_params(self): raise ValueError("Signal contains inf. Cannot perform adc.") # min and max ignoring nans, unless whole channel is NAN. - # Should suppress warning message. - minvals = np.nanmin(self.p_signal, axis=0) - maxvals = np.nanmax(self.p_signal, axis=0) + # Suppress the "All-NaN slice encountered" warning; channels + # that are entirely NAN are handled in calc_adc_gain_baseline. + with warnings.catch_warnings(): + warnings.simplefilter("ignore", category=RuntimeWarning) + minvals = np.nanmin(self.p_signal, axis=0) + maxvals = np.nanmax(self.p_signal, axis=0) for ch in range(np.shape(self.p_signal)[1]): adc_gain, baseline = self.calc_adc_gain_baseline(