Skip to content

Commit

Permalink
include more test cases for record_duration
Browse files Browse the repository at this point in the history
  • Loading branch information
skjerns committed Jan 23, 2025
1 parent 8ecbae6 commit 3b0490e
Show file tree
Hide file tree
Showing 3 changed files with 62 additions and 8 deletions.
13 changes: 8 additions & 5 deletions pyedflib/edfreader.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
REPAIR_FILE_SIZE_IF_WRONG = 2


def _debug_parse_header(filename: str) -> None: # pragma: no cover
def _debug_parse_header(filename: str, printout=True) -> None: # pragma: no cover
"""
A debug function that reads a header and outputs everything that
is contained in the header
Expand All @@ -55,8 +55,9 @@ def _debug_parse_header(filename: str) -> None: # pragma: no cover
header["record_duration"] = f.read(8).decode()
header["n_signals"] = f.read(4).decode()

print("\n##### Header")
print(json.dumps(header, indent=2))
if printout:
print("\n##### Header")
print(json.dumps(header, indent=2))

nsigs = int(header["n_signals"])
label = [f.read(16).decode() for i in range(nsigs)]
Expand Down Expand Up @@ -95,8 +96,10 @@ def _debug_parse_header(filename: str) -> None: # pragma: no cover
"reserved",
]
sheaders = [{field: values[field][i] for field in fields} for i in range(nsigs)]
print("\n##### Signal Headers")
print(json.dumps(sheaders, indent=2))
if printout:
print("\n##### Signal Headers")
print(json.dumps(sheaders, indent=2))
return header, sheaders


class EdfReader(CyEdfReader):
Expand Down
6 changes: 4 additions & 2 deletions pyedflib/edfwriter.py
Original file line number Diff line number Diff line change
Expand Up @@ -257,7 +257,7 @@ def all_divisors(num):
candidate = L / d
if candidate <= max_val:
# Build a short string
c_str = str(candidate).rstrip('0').rstrip('.')
c_str = f'{candidate:f}'.rstrip('0').rstrip('.')
if len(c_str) <= max_str_len:
return float(candidate)

Expand Down Expand Up @@ -602,7 +602,7 @@ def setGender(self, gender: Union[int, str, None]) -> None:
def setDatarecordDuration(self, record_duration: Union[float, int]) -> None:
"""
Sets the datarecord duration. The default value is 1 second.
The datarecord duration must be in the range 0.00001 to 60 seconds.
The datarecord duration must be in the range 0.001 to 60 seconds.
Usually, the datarecord duration is calculated automatically to
ensure that all sample frequencies are representable, nevertheless,
you can overwrite the datarecord duration manually. This can, however,
Expand All @@ -626,6 +626,8 @@ def setDatarecordDuration(self, record_duration: Union[float, int]) -> None:
"""
warnings.warn('Forcing a specific record_duration might alter calculated sample_frequencies when reading the file')
self._enforce_record_duration = True
if 0.001 > record_duration or record_duration > 60:
raise ValueError('record_duration must be between 0.001 and 60 seconds')
self.record_duration = record_duration
self.update_header()

Expand Down
51 changes: 50 additions & 1 deletion pyedflib/tests/test_edfwriter.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
import numpy as np

import pyedflib
from pyedflib.edfreader import EdfReader
from pyedflib.edfreader import EdfReader, _debug_parse_header
from pyedflib.edfwriter import ChannelDoesNotExist, EdfWriter, WrongInputSize
from pyedflib.edfwriter import _calculate_record_duration

Expand Down Expand Up @@ -1192,6 +1192,55 @@ def test_find_record_duration(self):
duration = _calculate_record_duration(freqs)
assert duration == 1

def test_record_durations(self):
"""use different record durations and look in the raw header if all seems right"""
for record_duration in [0.001, 0.01, 0.1, 1, 10, 60]:
channel_count = 1
sample_frequency = 1/record_duration

f = pyedflib.EdfWriter(self.edf_data_file, 1,
file_type=pyedflib.FILETYPE_EDF)
f.setDatarecordDuration(record_duration)

digMax = 32767
digMin = -digMax

f.setSignalHeaders([{
'label': 'test_label',
'sample_frequency': sample_frequency,
'dimension': 'mV',
'physical_min': 0,
'physical_max': 1,
'digital_min': digMin,
'digital_max': digMax,
'transducer': '',
'prefilter': ''
}])

f.writeSamples(np.random.rand(channel_count, 1000))
f.close()
del f

header, sheader = _debug_parse_header(self.edf_data_file, printout=False)
self.assertEqual(float(header['record_duration']), record_duration)

with pyedflib.EdfReader(self.edf_data_file) as f:
self.assertEqual(f.datarecord_duration, record_duration)
f.close()


for record_duration in [0.0001, 61]:
channel_count = 1
sample_frequency = 1/record_duration

with pyedflib.EdfWriter(self.edf_data_file, 1,
file_type=pyedflib.FILETYPE_EDF) as f:
with self.assertRaises(ValueError):
f.setDatarecordDuration(record_duration)




if __name__ == '__main__':
# run_module_suite(argv=sys.argv)
unittest.main()

0 comments on commit 3b0490e

Please sign in to comment.