Fix Record.wrsamp() when d_signal is missing

changelog.d/wrsamp-physical.rst

@@ -0,0 +1,3 @@
+**Fix writing physical-only records**
+  ``Record.wrsamp`` now temporarily converts physical signals to digital samples
+  when the corresponding digital signals are missing, then restores the record.

src/wfdb/io/record.py

@@ -902,38 +902,56 @@ def wrsamp(self, expanded=False, write_dir="", wfdb_archive=None):
         N/A
 
         """
-        # A record whose signals have no files, such as the layout header
-        # of a variable layout multi-segment record, has no samples:
-        # only its header is written.
-        has_signal_files = self.n_sig > 0 and (
-            self.file_name is None or any(f != "~" for f in self.file_name)
-        )
-        if has_signal_files and self.file_name is not None and "~" in self.file_name:
-            raise ValueError(
-                "Writing a record with both null signals (file name '~') and "
-                "signals with samples is not supported"
+        converted = None
+        if not expanded and self.d_signal is None and self.p_signal is not None:
+            self.d_signal = self.adc()
+            converted = "d_signal"
+        elif expanded and self.e_d_signal is None and self.e_p_signal is not None:
+            self.e_d_signal = self.adc(expanded=True)
+            converted = "e_d_signal"
+
+        try:
+            # A record whose signals have no files, such as the layout header
+            # of a variable layout multi-segment record, has no samples:
+            # only its header is written.
+            has_signal_files = self.n_sig > 0 and (
+                self.file_name is None or any(f != "~" for f in self.file_name)
             )
+            if (
+                has_signal_files
+                and self.file_name is not None
+                and "~" in self.file_name
+            ):
+                raise ValueError(
+                    "Writing a record with both null signals (file name '~') and "
+                    "signals with samples is not supported"
+                )
 
-        # Update the checksum field (except for channels that did not have
-        # a checksum to begin with, or where the checksum was already
-        # valid.)
-        if self.checksum is not None and has_signal_files:
-            checksums = self.calc_checksum(expanded=expanded)
-            for ch, old_val in enumerate(self.checksum):
-                if old_val is None or (checksums[ch] - old_val) % 65536 == 0:
-                    checksums[ch] = old_val
-            self.checksum = checksums
-
-        with archive_for_writing(wfdb_archive, write_dir) as archive:
-            # Perform field validity and cohesion checks, and write the
-            # header file.
-            self.wrheader(write_dir=write_dir, expanded=expanded, wfdb_archive=archive)
-            if has_signal_files:
-                # Perform signal validity and cohesion checks, and write the
-                # associated dat files.
-                self.wr_dats(
-                    expanded=expanded, write_dir=write_dir, wfdb_archive=archive
+            # Update the checksum field (except for channels that did not have
+            # a checksum to begin with, or where the checksum was already
+            # valid.)
+            if self.checksum is not None and has_signal_files:
+                checksums = self.calc_checksum(expanded=expanded)
+                for ch, old_val in enumerate(self.checksum):
+                    if old_val is None or (checksums[ch] - old_val) % 65536 == 0:
+                        checksums[ch] = old_val
+                self.checksum = checksums
+
+            with archive_for_writing(wfdb_archive, write_dir) as archive:
+                # Perform field validity and cohesion checks, and write the
+                # header file.
+                self.wrheader(
+                    write_dir=write_dir, expanded=expanded, wfdb_archive=archive
                 )
+                if has_signal_files:
+                    # Perform signal validity and cohesion checks, and write the
+                    # associated dat files.
+                    self.wr_dats(
+                        expanded=expanded, write_dir=write_dir, wfdb_archive=archive
+                    )
+        finally:
+            if converted:
+                setattr(self, converted, None)
 
     def _arrange_fields(self, channels, sampfrom, smooth_frames):
         """Arrange/edit object fields to reflect user channel and/or signal range input.

tests/test_record.py

@@ -600,6 +600,61 @@ def test_2b(self, source, temp_path):
         assert record.__eq__(record_named)
         assert record.__eq__(record_write)
 
+    def test_wrsamp_physical_only_record(self, temp_path):
+        record = wfdb.rdrecord("sample-data/100", sampto=1000)
+        expected_d_signal = wfdb.rdrecord(
+            "sample-data/100", sampto=1000, physical=False
+        ).d_signal
+
+        assert record.d_signal is None
+
+        record.wrsamp(write_dir=temp_path)
+        record_write = wfdb.rdrecord(
+            os.path.join(temp_path, "100"),
+            physical=False,
+        )
+
+        np.testing.assert_array_equal(record_write.d_signal, expected_d_signal)
+        assert record.d_signal is None
+
+    def test_wrsamp_physical_only_record_second_write(self, temp_path):
+        record = wfdb.rdrecord("sample-data/100", sampto=1000)
+        record.wrsamp(write_dir=temp_path)
+
+        assert record.d_signal is None
+
+        record.p_signal[:] = 0
+        record.wrsamp(write_dir=temp_path)
+        record_write = wfdb.rdrecord(
+            os.path.join(temp_path, "100"),
+            physical=False,
+        )
+
+        expected_d_signal = np.tile(record.baseline, (record.sig_len, 1))
+        np.testing.assert_array_equal(record_write.d_signal, expected_d_signal)
+        assert record.d_signal is None
+
+    def test_wrsamp_expanded_physical_only_record(self, temp_path):
+        record = wfdb.rdrecord("sample-data/test01_00s_frame", smooth_frames=False)
+        expected_e_d_signal = wfdb.rdrecord(
+            "sample-data/test01_00s_frame",
+            smooth_frames=False,
+            physical=False,
+        ).e_d_signal
+
+        assert record.e_d_signal is None
+
+        record.wrsamp(write_dir=temp_path, expanded=True)
+        record_write = wfdb.rdrecord(
+            os.path.join(temp_path, "test01_00s_frame"),
+            smooth_frames=False,
+            physical=False,
+        )
+
+        for written, expected in zip(record_write.e_d_signal, expected_e_d_signal):
+            np.testing.assert_array_equal(written, expected)
+        assert record.e_d_signal is None
+
     def test_2c(self, temp_path):
         """Format 212, entire signal, physical, odd sampled record.