Handle all-NaN channels in calc_adc_params (fixes #485)

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()

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(