From 1a9937faf13725dcec3ce15b788cba4d7e125e71 Mon Sep 17 00:00:00 2001 From: Brian Gow Date: Wed, 16 Oct 2024 17:54:33 -0400 Subject: [PATCH 1/9] update wfdb.io.wrsamp to allow writing a signal with unique samps_per_frame --- wfdb/io/record.py | 120 +++++++++++++++++++++++++++++++--------------- 1 file changed, 81 insertions(+), 39 deletions(-) diff --git a/wfdb/io/record.py b/wfdb/io/record.py index 4b900b17..bae9b6e4 100644 --- a/wfdb/io/record.py +++ b/wfdb/io/record.py @@ -1,5 +1,4 @@ import datetime -import multiprocessing.dummy import posixpath import os import re @@ -2822,6 +2821,9 @@ def wrsamp( sig_name, p_signal=None, d_signal=None, + e_p_signal=None, + e_d_signal=None, + samps_per_frame=None, fmt=None, adc_gain=None, baseline=None, @@ -2860,6 +2862,14 @@ def wrsamp( file(s). The dtype must be an integer type. Either p_signal or d_signal must be set, but not both. In addition, if d_signal is set, fmt, gain and baseline must also all be set. + e_p_signal : ndarray, optional + The expanded physical conversion of the signal. Either a 2d numpy + array or a list of 1d numpy arrays. + e_d_signal : ndarray, optional + The expanded digital conversion of the signal. Either a 2d numpy + array or a list of 1d numpy arrays. + samps_per_frame : int or list of ints, optional + The total number of samples per frame. fmt : list, optional A list of strings giving the WFDB format of each file used to store each channel. Accepted formats are: '80','212','16','24', and '32'. There are @@ -2911,59 +2921,91 @@ def wrsamp( if "." in record_name: raise Exception("Record name must not contain '.'") # Check input field combinations - if p_signal is not None and d_signal is not None: + signal_list = [p_signal, d_signal, e_p_signal, e_d_signal] + signals_set = sum(1 for var in signal_list if var is not None) + if signals_set != 1: raise Exception( - "Must only give one of the inputs: p_signal or d_signal" + "Must provide one and only one input signal: p_signal, d_signal, e_p_signal, or e_d_signal" ) - if d_signal is not None: + if d_signal is not None or e_d_signal is not None: if fmt is None or adc_gain is None or baseline is None: raise Exception( - "When using d_signal, must also specify 'fmt', 'gain', and 'baseline' fields." + "When using d_signal or e_d_signal, must also specify 'fmt', 'gain', and 'baseline' fields" ) - # Depending on whether d_signal or p_signal was used, set other - # required features. + + # If samps_per_frame is a list, check that it aligns as expected with the channels in the signal + if len(samps_per_frame) > 1: + # Get properties of the signal being passed + non_none_signal = next(signal for signal in signal_list if signal is not None) + if isinstance(non_none_signal, np.ndarray): + num_sig_channels = non_none_signal.shape[1] + channel_samples = [non_none_signal.shape[0]] * non_none_signal.shape[1] + elif isinstance(non_none_signal, list): + num_sig_channels = len(non_none_signal) + channel_samples = [len(channel) for channel in non_none_signal] + else: + raise TypeError("Unsupported signal format. Must be ndarray or list of lists.") + + # Check that the number of channels matches the number of samps_per_frame entries + if num_sig_channels != len(samps_per_frame): + raise Exception( + "When passing samps_per_frame as a list, it must have the same number of entries as the signal has channels" + ) + + # Check that the number of frames is the same across all channels + frames = [a / b for a, b in zip(channel_samples, samps_per_frame)] + if len(set(frames)) > 1: + raise Exception( + "The number of samples in a channel divided by the corresponding samples_per_frame entry must be uniform" + ) + + # Create the Record object + record = Record( + record_name=record_name, + p_signal=p_signal, + d_signal=d_signal, + e_p_signal=e_p_signal, + e_d_signal=e_d_signal, + samps_per_frame=samps_per_frame, + fs=fs, + fmt=fmt, + units=units, + sig_name=sig_name, + adc_gain=adc_gain, + baseline=baseline, + comments=comments, + base_time=base_time, + base_date=base_date, + base_datetime=base_datetime, + ) + + # Depending on which signal was used, set other required fields. if p_signal is not None: - # Create the Record object - record = Record( - record_name=record_name, - p_signal=p_signal, - fs=fs, - fmt=fmt, - units=units, - sig_name=sig_name, - adc_gain=adc_gain, - baseline=baseline, - comments=comments, - base_time=base_time, - base_date=base_date, - base_datetime=base_datetime, - ) # Compute optimal fields to store the digital signal, carry out adc, # and set the fields. record.set_d_features(do_adc=1) - else: - # Create the Record object - record = Record( - record_name=record_name, - d_signal=d_signal, - fs=fs, - fmt=fmt, - units=units, - sig_name=sig_name, - adc_gain=adc_gain, - baseline=baseline, - comments=comments, - base_time=base_time, - base_date=base_date, - base_datetime=base_datetime, - ) + elif d_signal is not None: # Use d_signal to set the fields directly record.set_d_features() + elif e_p_signal is not None: + # Compute optimal fields to store the digital signal, carry out adc, + # and set the fields. + record.set_d_features(do_adc=1, expanded=True) + elif e_d_signal is not None: + # Use e_d_signal to set the fields directly + record.set_d_features(expanded=True) # Set default values of any missing field dependencies record.set_defaults() + + # Determine whether the signal is expanded + if (e_d_signal or e_p_signal) is not None: + expanded = True + else: + expanded = False + # Write the record files - header and associated dat - record.wrsamp(write_dir=write_dir) + record.wrsamp(write_dir=write_dir, expanded=expanded) def dl_database( From d7b5e93f63aae54eac62c1ef0526591d15e205a9 Mon Sep 17 00:00:00 2001 From: Brian Gow Date: Wed, 16 Oct 2024 17:54:52 -0400 Subject: [PATCH 2/9] add tests for wfdb.io.wrsamp for a signal with unique samps_per_frame --- tests/test_record.py | 111 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 111 insertions(+) diff --git a/tests/test_record.py b/tests/test_record.py index 3459897b..dcf2d762 100644 --- a/tests/test_record.py +++ b/tests/test_record.py @@ -20,6 +20,11 @@ class TestRecord(unittest.TestCase): """ + + wrsamp_params = ['record_name', 'fs', 'units', 'sig_name', 'p_signal', 'd_signal', 'e_p_signal', 'e_d_signal', + 'samps_per_frame', 'fmt', 'adc_gain', 'baseline', 'comments', 'base_time', 'base_date', + 'base_datetime'] + # ----------------------- 1. Basic Tests -----------------------# def test_1a(self): @@ -286,6 +291,112 @@ def test_read_write_flac_multifrequency(self): ) assert record == record_write + def test_unique_samps_per_frame_e_p_signal(self): + """ + Test writing an e_p_signal with wfdb.io.wrsamp where the signals have different samples per frame. All other + parameters which overlap between a Record object and wfdb.io.wrsamp are also checked. + """ + # Read in a record with different samples per frame + record = wfdb.rdrecord( + "sample-data/mixedsignals", + smooth_frames=False, + ) + + # Write the signals + wfdb.io.wrsamp('mixedsignals', fs=record.fs, units=record.units, sig_name=record.sig_name, + base_date=record.base_date, base_time=record.base_time, comments=record.comments, + p_signal=record.p_signal, d_signal=record.d_signal, e_p_signal=record.e_p_signal, + e_d_signal=record.e_d_signal, samps_per_frame=record.samps_per_frame, baseline=record.baseline, + adc_gain=record.adc_gain, fmt=record.fmt, write_dir=self.temp_path) + + # Check that the written record matches the original + # Read in the original and written records + record = wfdb.rdrecord("sample-data/mixedsignals", smooth_frames=False) + record_write = wfdb.rdrecord( + os.path.join(self.temp_path, "mixedsignals"), + smooth_frames=False, + ) + + # Check that the signals match + for n, name in enumerate(record.sig_name): + np.testing.assert_array_equal( + record.e_p_signal[n], record_write.e_p_signal[n], f"Mismatch in {name}" + ) + + # Filter out the signal + record_filtered = { + k: getattr(record, k) + for k in self.wrsamp_params + if not (isinstance(getattr(record, k), np.ndarray) or + (isinstance(getattr(record, k), list) and all( + isinstance(item, np.ndarray) for item in getattr(record, k)))) + } + + record_write_filtered = { + k: getattr(record_write, k) + for k in self.wrsamp_params + if not (isinstance(getattr(record_write, k), np.ndarray) or + (isinstance(getattr(record_write, k), list) and all( + isinstance(item, np.ndarray) for item in getattr(record_write, k)))) + } + + # Check that the arguments beyond the signals also match + assert record_filtered == record_write_filtered + + def test_unique_samps_per_frame_e_d_signal(self): + """ + Test writing an e_d_signal with wfdb.io.wrsamp where the signals have different samples per frame. All other + parameters which overlap between a Record object and wfdb.io.wrsamp are also checked. + """ + # Read in a record with different samples per frame + record = wfdb.rdrecord( + "sample-data/mixedsignals", + physical=False, + smooth_frames=False, + ) + + # Write the signals + wfdb.io.wrsamp('mixedsignals', fs=record.fs, units=record.units, sig_name=record.sig_name, + base_date=record.base_date, base_time=record.base_time, comments=record.comments, + p_signal=record.p_signal, d_signal=record.d_signal, e_p_signal=record.e_p_signal, + e_d_signal=record.e_d_signal, samps_per_frame=record.samps_per_frame, baseline=record.baseline, + adc_gain=record.adc_gain, fmt=record.fmt, write_dir=self.temp_path) + + # Check that the written record matches the original + # Read in the original and written records + record = wfdb.rdrecord("sample-data/mixedsignals", physical=False, smooth_frames=False) + record_write = wfdb.rdrecord( + os.path.join(self.temp_path, "mixedsignals"), + physical=False, + smooth_frames=False, + ) + + # Check that the signals match + for n, name in enumerate(record.sig_name): + np.testing.assert_array_equal( + record.e_d_signal[n], record_write.e_d_signal[n], f"Mismatch in {name}" + ) + + # Filter out the signal + record_filtered = { + k: getattr(record, k) + for k in self.wrsamp_params + if not (isinstance(getattr(record, k), np.ndarray) or + (isinstance(getattr(record, k), list) and all( + isinstance(item, np.ndarray) for item in getattr(record, k)))) + } + + record_write_filtered = { + k: getattr(record_write, k) + for k in self.wrsamp_params + if not (isinstance(getattr(record_write, k), np.ndarray) or + (isinstance(getattr(record_write, k), list) and all( + isinstance(item, np.ndarray) for item in getattr(record_write, k)))) + } + + # Check that the arguments beyond the signals also match + assert record_filtered == record_write_filtered + def test_read_write_flac_many_channels(self): """ Check we can read and write to format 516 with more than 8 channels. From 0e322090830c6b546e34f7ca982e717e1e9c2b8f Mon Sep 17 00:00:00 2001 From: Brian Gow Date: Fri, 18 Oct 2024 13:51:47 -0400 Subject: [PATCH 3/9] address failing tests and feedback --- wfdb/io/record.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/wfdb/io/record.py b/wfdb/io/record.py index bae9b6e4..9c300617 100644 --- a/wfdb/io/record.py +++ b/wfdb/io/record.py @@ -2934,15 +2934,15 @@ def wrsamp( ) # If samps_per_frame is a list, check that it aligns as expected with the channels in the signal - if len(samps_per_frame) > 1: + if isinstance(samps_per_frame, list): # Get properties of the signal being passed - non_none_signal = next(signal for signal in signal_list if signal is not None) - if isinstance(non_none_signal, np.ndarray): - num_sig_channels = non_none_signal.shape[1] - channel_samples = [non_none_signal.shape[0]] * non_none_signal.shape[1] - elif isinstance(non_none_signal, list): - num_sig_channels = len(non_none_signal) - channel_samples = [len(channel) for channel in non_none_signal] + first_valid_signal = next(signal for signal in signal_list if signal is not None) + if isinstance(first_valid_signal, np.ndarray): + num_sig_channels = first_valid_signal.shape[1] + channel_samples = [first_valid_signal.shape[0]] * first_valid_signal.shape[1] + elif isinstance(first_valid_signal, list): + num_sig_channels = len(first_valid_signal) + channel_samples = [len(channel) for channel in first_valid_signal] else: raise TypeError("Unsupported signal format. Must be ndarray or list of lists.") From def62e96f4a8e303faff7dac08a97ec7781ac37d Mon Sep 17 00:00:00 2001 From: Brian Gow Date: Fri, 18 Oct 2024 13:51:11 -0400 Subject: [PATCH 4/9] correct signal name being set --- wfdb/io/_signal.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/wfdb/io/_signal.py b/wfdb/io/_signal.py index e3df065e..4687aece 100644 --- a/wfdb/io/_signal.py +++ b/wfdb/io/_signal.py @@ -433,7 +433,7 @@ def set_d_features(self, do_adc=False, single_fmt=True, expanded=False): self.check_field("baseline", "all") # All required fields are present and valid. Perform ADC - self.d_signal = self.adc(expanded) + self.e_d_signal = self.adc(expanded) # Use e_d_signal to set fields self.check_field("e_d_signal", "all") From 11e87562c61c9204afe99067c2658d9c77d31dc3 Mon Sep 17 00:00:00 2001 From: Brian Gow Date: Fri, 18 Oct 2024 14:17:16 -0400 Subject: [PATCH 5/9] add missing import --- wfdb/io/record.py | 1 + 1 file changed, 1 insertion(+) diff --git a/wfdb/io/record.py b/wfdb/io/record.py index 9c300617..26fb0fc7 100644 --- a/wfdb/io/record.py +++ b/wfdb/io/record.py @@ -1,4 +1,5 @@ import datetime +import multiprocessing.dummy import posixpath import os import re From f8084d31044d031f00e6d3bd9c320561e203da88 Mon Sep 17 00:00:00 2001 From: Brian Gow Date: Sat, 19 Oct 2024 10:01:51 -0400 Subject: [PATCH 6/9] reformatted with black package --- wfdb/io/record.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/wfdb/io/record.py b/wfdb/io/record.py index 26fb0fc7..b640613f 100644 --- a/wfdb/io/record.py +++ b/wfdb/io/record.py @@ -2937,15 +2937,21 @@ def wrsamp( # If samps_per_frame is a list, check that it aligns as expected with the channels in the signal if isinstance(samps_per_frame, list): # Get properties of the signal being passed - first_valid_signal = next(signal for signal in signal_list if signal is not None) + first_valid_signal = next( + signal for signal in signal_list if signal is not None + ) if isinstance(first_valid_signal, np.ndarray): num_sig_channels = first_valid_signal.shape[1] - channel_samples = [first_valid_signal.shape[0]] * first_valid_signal.shape[1] + channel_samples = [ + first_valid_signal.shape[0] + ] * first_valid_signal.shape[1] elif isinstance(first_valid_signal, list): num_sig_channels = len(first_valid_signal) channel_samples = [len(channel) for channel in first_valid_signal] else: - raise TypeError("Unsupported signal format. Must be ndarray or list of lists.") + raise TypeError( + "Unsupported signal format. Must be ndarray or list of lists." + ) # Check that the number of channels matches the number of samps_per_frame entries if num_sig_channels != len(samps_per_frame): From 1e93bee6aa61d3289cdf648e558c82c7da3b7fbf Mon Sep 17 00:00:00 2001 From: Brian Gow Date: Sat, 19 Oct 2024 10:06:36 -0400 Subject: [PATCH 7/9] reformatted with black package --- tests/test_record.py | 132 +++++++++++++++++++++++++++++++++---------- 1 file changed, 103 insertions(+), 29 deletions(-) diff --git a/tests/test_record.py b/tests/test_record.py index dcf2d762..313eeb23 100644 --- a/tests/test_record.py +++ b/tests/test_record.py @@ -20,10 +20,24 @@ class TestRecord(unittest.TestCase): """ - - wrsamp_params = ['record_name', 'fs', 'units', 'sig_name', 'p_signal', 'd_signal', 'e_p_signal', 'e_d_signal', - 'samps_per_frame', 'fmt', 'adc_gain', 'baseline', 'comments', 'base_time', 'base_date', - 'base_datetime'] + wrsamp_params = [ + "record_name", + "fs", + "units", + "sig_name", + "p_signal", + "d_signal", + "e_p_signal", + "e_d_signal", + "samps_per_frame", + "fmt", + "adc_gain", + "baseline", + "comments", + "base_time", + "base_date", + "base_datetime", + ] # ----------------------- 1. Basic Tests -----------------------# @@ -303,11 +317,24 @@ def test_unique_samps_per_frame_e_p_signal(self): ) # Write the signals - wfdb.io.wrsamp('mixedsignals', fs=record.fs, units=record.units, sig_name=record.sig_name, - base_date=record.base_date, base_time=record.base_time, comments=record.comments, - p_signal=record.p_signal, d_signal=record.d_signal, e_p_signal=record.e_p_signal, - e_d_signal=record.e_d_signal, samps_per_frame=record.samps_per_frame, baseline=record.baseline, - adc_gain=record.adc_gain, fmt=record.fmt, write_dir=self.temp_path) + wfdb.io.wrsamp( + "mixedsignals", + fs=record.fs, + units=record.units, + sig_name=record.sig_name, + base_date=record.base_date, + base_time=record.base_time, + comments=record.comments, + p_signal=record.p_signal, + d_signal=record.d_signal, + e_p_signal=record.e_p_signal, + e_d_signal=record.e_d_signal, + samps_per_frame=record.samps_per_frame, + baseline=record.baseline, + adc_gain=record.adc_gain, + fmt=record.fmt, + write_dir=self.temp_path, + ) # Check that the written record matches the original # Read in the original and written records @@ -320,24 +347,40 @@ def test_unique_samps_per_frame_e_p_signal(self): # Check that the signals match for n, name in enumerate(record.sig_name): np.testing.assert_array_equal( - record.e_p_signal[n], record_write.e_p_signal[n], f"Mismatch in {name}" + record.e_p_signal[n], + record_write.e_p_signal[n], + f"Mismatch in {name}", ) # Filter out the signal record_filtered = { k: getattr(record, k) for k in self.wrsamp_params - if not (isinstance(getattr(record, k), np.ndarray) or - (isinstance(getattr(record, k), list) and all( - isinstance(item, np.ndarray) for item in getattr(record, k)))) + if not ( + isinstance(getattr(record, k), np.ndarray) + or ( + isinstance(getattr(record, k), list) + and all( + isinstance(item, np.ndarray) + for item in getattr(record, k) + ) + ) + ) } record_write_filtered = { k: getattr(record_write, k) for k in self.wrsamp_params - if not (isinstance(getattr(record_write, k), np.ndarray) or - (isinstance(getattr(record_write, k), list) and all( - isinstance(item, np.ndarray) for item in getattr(record_write, k)))) + if not ( + isinstance(getattr(record_write, k), np.ndarray) + or ( + isinstance(getattr(record_write, k), list) + and all( + isinstance(item, np.ndarray) + for item in getattr(record_write, k) + ) + ) + ) } # Check that the arguments beyond the signals also match @@ -356,15 +399,30 @@ def test_unique_samps_per_frame_e_d_signal(self): ) # Write the signals - wfdb.io.wrsamp('mixedsignals', fs=record.fs, units=record.units, sig_name=record.sig_name, - base_date=record.base_date, base_time=record.base_time, comments=record.comments, - p_signal=record.p_signal, d_signal=record.d_signal, e_p_signal=record.e_p_signal, - e_d_signal=record.e_d_signal, samps_per_frame=record.samps_per_frame, baseline=record.baseline, - adc_gain=record.adc_gain, fmt=record.fmt, write_dir=self.temp_path) + wfdb.io.wrsamp( + "mixedsignals", + fs=record.fs, + units=record.units, + sig_name=record.sig_name, + base_date=record.base_date, + base_time=record.base_time, + comments=record.comments, + p_signal=record.p_signal, + d_signal=record.d_signal, + e_p_signal=record.e_p_signal, + e_d_signal=record.e_d_signal, + samps_per_frame=record.samps_per_frame, + baseline=record.baseline, + adc_gain=record.adc_gain, + fmt=record.fmt, + write_dir=self.temp_path, + ) # Check that the written record matches the original # Read in the original and written records - record = wfdb.rdrecord("sample-data/mixedsignals", physical=False, smooth_frames=False) + record = wfdb.rdrecord( + "sample-data/mixedsignals", physical=False, smooth_frames=False + ) record_write = wfdb.rdrecord( os.path.join(self.temp_path, "mixedsignals"), physical=False, @@ -374,24 +432,40 @@ def test_unique_samps_per_frame_e_d_signal(self): # Check that the signals match for n, name in enumerate(record.sig_name): np.testing.assert_array_equal( - record.e_d_signal[n], record_write.e_d_signal[n], f"Mismatch in {name}" + record.e_d_signal[n], + record_write.e_d_signal[n], + f"Mismatch in {name}", ) # Filter out the signal record_filtered = { k: getattr(record, k) for k in self.wrsamp_params - if not (isinstance(getattr(record, k), np.ndarray) or - (isinstance(getattr(record, k), list) and all( - isinstance(item, np.ndarray) for item in getattr(record, k)))) + if not ( + isinstance(getattr(record, k), np.ndarray) + or ( + isinstance(getattr(record, k), list) + and all( + isinstance(item, np.ndarray) + for item in getattr(record, k) + ) + ) + ) } record_write_filtered = { k: getattr(record_write, k) for k in self.wrsamp_params - if not (isinstance(getattr(record_write, k), np.ndarray) or - (isinstance(getattr(record_write, k), list) and all( - isinstance(item, np.ndarray) for item in getattr(record_write, k)))) + if not ( + isinstance(getattr(record_write, k), np.ndarray) + or ( + isinstance(getattr(record_write, k), list) + and all( + isinstance(item, np.ndarray) + for item in getattr(record_write, k) + ) + ) + ) } # Check that the arguments beyond the signals also match From 7bc468b22a9079d2f9d2fb32ba95946c19d70642 Mon Sep 17 00:00:00 2001 From: Brian Gow Date: Mon, 28 Oct 2024 15:02:53 -0400 Subject: [PATCH 8/9] add wfdb.io.wrsamp signal checks --- wfdb/io/record.py | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/wfdb/io/record.py b/wfdb/io/record.py index b640613f..969b9792 100644 --- a/wfdb/io/record.py +++ b/wfdb/io/record.py @@ -2933,9 +2933,15 @@ def wrsamp( raise Exception( "When using d_signal or e_d_signal, must also specify 'fmt', 'gain', and 'baseline' fields" ) + if (e_p_signal is not None or e_d_signal is not None) and samps_per_frame is None: + raise Exception( + "When passing e_p_signal or e_d_signal, you also need to specify samples per frame for each channel" + ) - # If samps_per_frame is a list, check that it aligns as expected with the channels in the signal - if isinstance(samps_per_frame, list): + # If samps_per_frame is provided, check that it aligns as expected with the channels in the signal + if samps_per_frame: + # Get the number of elements being passed in samps_per_frame + samps_per_frame_length = len(samps_per_frame) if isinstance(samps_per_frame, list) else 1 # Get properties of the signal being passed first_valid_signal = next( signal for signal in signal_list if signal is not None @@ -2952,13 +2958,11 @@ def wrsamp( raise TypeError( "Unsupported signal format. Must be ndarray or list of lists." ) - # Check that the number of channels matches the number of samps_per_frame entries - if num_sig_channels != len(samps_per_frame): + if num_sig_channels != samps_per_frame_length: raise Exception( - "When passing samps_per_frame as a list, it must have the same number of entries as the signal has channels" + "When passing samps_per_frame, it must have the same number of entries as the signal has channels" ) - # Check that the number of frames is the same across all channels frames = [a / b for a, b in zip(channel_samples, samps_per_frame)] if len(set(frames)) > 1: From 6b0d3175a4c720fb3bc8aca4a53ec66b69975821 Mon Sep 17 00:00:00 2001 From: Brian Gow Date: Mon, 28 Oct 2024 15:45:02 -0400 Subject: [PATCH 9/9] reformatted with black package --- wfdb/io/record.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/wfdb/io/record.py b/wfdb/io/record.py index 969b9792..1a8855ed 100644 --- a/wfdb/io/record.py +++ b/wfdb/io/record.py @@ -2933,15 +2933,19 @@ def wrsamp( raise Exception( "When using d_signal or e_d_signal, must also specify 'fmt', 'gain', and 'baseline' fields" ) - if (e_p_signal is not None or e_d_signal is not None) and samps_per_frame is None: - raise Exception( - "When passing e_p_signal or e_d_signal, you also need to specify samples per frame for each channel" - ) + if ( + e_p_signal is not None or e_d_signal is not None + ) and samps_per_frame is None: + raise Exception( + "When passing e_p_signal or e_d_signal, you also need to specify samples per frame for each channel" + ) # If samps_per_frame is provided, check that it aligns as expected with the channels in the signal if samps_per_frame: # Get the number of elements being passed in samps_per_frame - samps_per_frame_length = len(samps_per_frame) if isinstance(samps_per_frame, list) else 1 + samps_per_frame_length = ( + len(samps_per_frame) if isinstance(samps_per_frame, list) else 1 + ) # Get properties of the signal being passed first_valid_signal = next( signal for signal in signal_list if signal is not None