diff --git a/changelog.d/wrsamp-physical.rst b/changelog.d/wrsamp-physical.rst new file mode 100644 index 00000000..a1ba1148 --- /dev/null +++ b/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. diff --git a/src/wfdb/io/record.py b/src/wfdb/io/record.py index 560c10b9..1c1f96fa 100644 --- a/src/wfdb/io/record.py +++ b/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. diff --git a/tests/test_record.py b/tests/test_record.py index 08a4a6c2..d66f265c 100644 --- a/tests/test_record.py +++ b/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.