diff --git a/phys2bids/tests/conftest.py b/phys2bids/tests/conftest.py index 16d66969f..a99625e5e 100644 --- a/phys2bids/tests/conftest.py +++ b/phys2bids/tests/conftest.py @@ -44,12 +44,6 @@ def samefreq_full_acq_file(testpath): 'Test_belt_pulse_samefreq.acq') -@pytest.fixture -def samefreq_full_txt_file(testpath): - return fetch_file('sdz4n', testpath, - 'Test_belt_pulse_samefreq.txt') - - @pytest.fixture def samefreq_short_txt_file(testpath): return fetch_file('4yudk', testpath, @@ -58,7 +52,7 @@ def samefreq_short_txt_file(testpath): @pytest.fixture def samefreq_noheader_txt_file(testpath): - return fetch_file('sre3h', testpath, + return fetch_file('xbwq9', testpath, 'Test_belt_pulse_samefreq_no_header.txt') @@ -70,11 +64,11 @@ def multifreq_acq_file(testpath): @pytest.fixture def multifreq_lab_file(testpath): - return fetch_file('q4x2f', testpath, - 'Test_2minRest_trig_multifreq_header_comment.txt') + return fetch_file('7se4t', testpath, + 'Test1_multifreq_onescan.txt') @pytest.fixture def notime_lab_file(testpath): - return fetch_file('5829m', testpath, - 'Test2_trigger_CO2_O2_pulse_1000Hz_534TRs_no_time.txt') + return fetch_file('cv5zr', testpath, + 'Test2_samefreq_onescan_notime.txt') diff --git a/phys2bids/tests/test_integration.py b/phys2bids/tests/test_integration.py index 1ce893b0a..cb50eda93 100644 --- a/phys2bids/tests/test_integration.py +++ b/phys2bids/tests/test_integration.py @@ -21,23 +21,22 @@ def check_string(str_container, str_to_find, str_expected, is_num=True): return str_expected in str_found -def test_logger(): +def test_logger(multifreq_lab_file): """ Tests the logger """ - # Move into folder - test_path = resource_filename('phys2bids', 'tests/data') - subprocess.run(f'cd {test_path}', shell=True, check=True) + test_path, test_filename = os.path.split(multifreq_lab_file) # Input arguments - test_filename = 'tutorial_file.txt' - test_chtrig = 1 + test_chtrig = 3 + test_ntp = 1 test_outdir = test_path # Phys2bids call through terminal subprocess.run(f'phys2bids -in {test_filename} -indir {test_path} ' - f'-chtrig {test_chtrig} -outdir {test_outdir}', shell=True, check=True) + f'-chtrig {test_chtrig} -ntp {test_ntp} -outdir {test_outdir}', + shell=True, check=True) # Read logger file logger_file = glob.glob(os.path.join(test_path, '*phys2bids*'))[0] @@ -52,52 +51,50 @@ def test_logger(): os.remove(os.path.join(test_path, logger_file)) -def test_integration_tutorial(): +def test_integration_txt(samefreq_short_txt_file): """ Does an integration test with the tutorial file """ - test_path = resource_filename('phys2bids', 'tests/data') - test_filename = 'tutorial_file.txt' - test_full_path = os.path.join(test_path, test_filename) - test_chtrig = 1 - test_outdir = test_path - test_ntp = 158 - test_tr = 1.2 - test_thr = 0.735 - phys2bids(filename=test_full_path, chtrig=test_chtrig, outdir=test_outdir, - num_timepoints_expected=test_ntp, tr=test_tr, thr=test_thr) + + test_path, test_filename = os.path.split(samefreq_short_txt_file) + test_chtrig = 2 + + phys2bids(filename=test_filename, indir=test_path, outdir=test_path, + chtrig=test_chtrig, num_timepoints_expected=1) # Check that files are generated for suffix in ['.log', '.json', '.tsv.gz', '_trigger_time.png']: - assert os.path.isfile(os.path.join(test_path, 'tutorial_file' + suffix)) + assert os.path.isfile(os.path.join(test_path, 'Test_belt_pulse_samefreq_short' + suffix)) # Read log file (note that this file is not the logger file) - with open(os.path.join(test_path, 'tutorial_file.log')) as log_info: + with open(os.path.join(test_path, 'Test_belt_pulse_samefreq_short.log')) as log_info: log_info = log_info.readlines() # Check timepoints expected - assert check_string(log_info, 'Timepoints expected', '158') + assert check_string(log_info, 'Timepoints expected', '1') # Check timepoints found - assert check_string(log_info, 'Timepoints found', '158') - # Check sampling frequency - assert check_string(log_info, 'Sampling Frequency', '1000.0') + assert check_string(log_info, 'Timepoints found', '60') # Check sampling frequency - assert check_string(log_info, 'Sampling started', '0.2450') + assert check_string(log_info, 'Sampling Frequency', '10000.0') + # Check sampling started + assert check_string(log_info, 'Sampling started', '10.4251') # Check start time assert check_string(log_info, 'first trigger', 'Time 0', is_num=False) # Checks json file - with open(os.path.join(test_path, 'tutorial_file.json')) as json_file: + with open(os.path.join(test_path, 'Test_belt_pulse_samefreq_short.json')) as json_file: json_data = json.load(json_file) # Compares values in json file with ground truth - assert math.isclose(json_data['SamplingFrequency'], 1000.0) - assert math.isclose(json_data['StartTime'], 0.2450) - assert json_data['Columns'] == ['time', 'Trigger', 'CO2', 'O2', 'Pulse'] + assert math.isclose(json_data['SamplingFrequency'], 10000.0) + assert math.isclose(json_data['StartTime'], 10.4251) + assert json_data['Columns'] == ['time', 'RESP - RSP100C', 'MR TRIGGER - Custom, HLT100C - A 5'] # Remove generated files for filename in glob.glob(os.path.join(test_path, 'phys2bids*')): os.remove(filename) + for filename in glob.glob(os.path.join(test_path, 'Test_belt_pulse_samefreq_short*')): + os.remove(filename) def test_integration_acq(samefreq_full_acq_file): @@ -147,12 +144,12 @@ def test_integration_acq(samefreq_full_acq_file): os.remove(filename) -def test_integration_multifreq(multifreq_acq_file): +def test_integration_multifreq(multifreq_lab_file): """ Does the integration test for a multi-frequency file """ - test_path, test_filename = os.path.split(multifreq_acq_file) + test_path, test_filename = os.path.split(multifreq_lab_file) test_chtrig = 3 phys2bids(filename=test_filename, indir=test_path, outdir=test_path, @@ -161,66 +158,125 @@ def test_integration_multifreq(multifreq_acq_file): # Check that files are generated for suffix in ['.log', '.json', '.tsv.gz']: assert os.path.isfile(os.path.join(test_path, - 'Test_belt_pulse_multifreq_625.0' + suffix)) + 'Test1_multifreq_onescan_40.0' + suffix)) + for suffix in ['.log', '.json', '.tsv.gz']: + assert os.path.isfile(os.path.join(test_path, + 'Test1_multifreq_onescan_100.0' + suffix)) + for suffix in ['.log', '.json', '.tsv.gz']: + assert os.path.isfile(os.path.join(test_path, + 'Test1_multifreq_onescan_500.0' + suffix)) for suffix in ['.log', '.json', '.tsv.gz']: assert os.path.isfile(os.path.join(test_path, - 'Test_belt_pulse_multifreq_10000.0' + suffix)) - assert os.path.isfile(os.path.join(test_path, 'Test_belt_pulse_multifreq_trigger_time.png')) + 'Test1_multifreq_onescan_1000.0' + suffix)) + assert os.path.isfile(os.path.join(test_path, 'Test1_multifreq_onescan_trigger_time.png')) """ - Checks 625 Hz output + Checks 40 Hz output """ # Read log file of frequency 625 (note that this file is not the logger file) - with open(os.path.join(test_path, 'Test_belt_pulse_multifreq_625.0.log')) as log_info: + with open(os.path.join(test_path, 'Test1_multifreq_onescan_40.0.log')) as log_info: log_info = log_info.readlines() # Check timepoints expected assert check_string(log_info, 'Timepoints expected', '1') # Check timepoints found - assert check_string(log_info, 'Timepoints found', '60') + assert check_string(log_info, 'Timepoints found', '0') # Check sampling frequency - assert check_string(log_info, 'Sampling Frequency', '625.0') + assert check_string(log_info, 'Sampling Frequency', '40.0') # Check sampling frequency - assert check_string(log_info, 'Sampling started', '0.2905') + assert check_string(log_info, 'Sampling started', '-157.8535') # Check start time assert check_string(log_info, 'first trigger', 'Time 0', is_num=False) # Checks json file - with open(os.path.join(test_path, 'Test_belt_pulse_multifreq_625.0.json')) as json_file: + with open(os.path.join(test_path, 'Test1_multifreq_onescan_40.0.json')) as json_file: json_data = json.load(json_file) # Compares values in json file with ground truth - assert math.isclose(json_data['SamplingFrequency'], 625.0) - assert math.isclose(json_data['StartTime'], 0.2905) - assert json_data['Columns'] == ['PULSE - Custom, DA100C'] + assert math.isclose(json_data['SamplingFrequency'], 40.0) + assert math.isclose(json_data['StartTime'], -157.8535) + assert json_data['Columns'] == ['O2'] """ - Checks 10000 Hz output + Checks 100 Hz output """ # Read log file of frequency 625 (note that this file is not the logger file) - with open(os.path.join(test_path, 'Test_belt_pulse_multifreq_10000.0.log')) as log_info: + with open(os.path.join(test_path, 'Test1_multifreq_onescan_100.0.log')) as log_info: log_info = log_info.readlines() # Check timepoints expected assert check_string(log_info, 'Timepoints expected', '1') # Check timepoints found - assert check_string(log_info, 'Timepoints found', '60') + assert check_string(log_info, 'Timepoints found', '0') # Check sampling frequency - assert check_string(log_info, 'Sampling Frequency', '10000.0') + assert check_string(log_info, 'Sampling Frequency', '100.0') # Check sampling started - assert check_string(log_info, 'Sampling started', '10.4251') + assert check_string(log_info, 'Sampling started', '-0.3057') # Check start time assert check_string(log_info, 'first trigger', 'Time 0', is_num=False) # Checks json file - with open(os.path.join(test_path, 'Test_belt_pulse_multifreq_10000.0.json')) as json_file: + with open(os.path.join(test_path, 'Test1_multifreq_onescan_100.0.json')) as json_file: json_data = json.load(json_file) # Compares values in json file with ground truth - assert math.isclose(json_data['SamplingFrequency'], 10000.0) - assert math.isclose(json_data['StartTime'], 10.4251) - assert json_data['Columns'] == ['time', 'RESP - RSP100C', - 'MR TRIGGER - Custom, HLT100C - A 5', 'PPG100C', 'CO2', 'O2'] + assert math.isclose(json_data['SamplingFrequency'], 100.0) + assert math.isclose(json_data['StartTime'], -0.3057) + assert json_data['Columns'] == ['CO2'] + + """ + Checks 500 Hz output + """ + # Read log file of frequency 625 (note that this file is not the logger file) + with open(os.path.join(test_path, 'Test1_multifreq_onescan_500.0.log')) as log_info: + log_info = log_info.readlines() + + # Check timepoints expected + assert check_string(log_info, 'Timepoints expected', '1') + # Check timepoints found + assert check_string(log_info, 'Timepoints found', '0') + # Check sampling frequency + assert check_string(log_info, 'Sampling Frequency', '500.0') + # Check sampling started + assert check_string(log_info, 'Sampling started', '-4.2019') + # Check start time + assert check_string(log_info, 'first trigger', 'Time 0', is_num=False) + + # Checks json file + with open(os.path.join(test_path, 'Test1_multifreq_onescan_500.0.json')) as json_file: + json_data = json.load(json_file) + + # Compares values in json file with ground truth + assert math.isclose(json_data['SamplingFrequency'], 500.0) + assert math.isclose(json_data['StartTime'], -4.2019) + assert json_data['Columns'] == ['Belt'] + + """ + Checks 100 Hz output + """ + # Read log file of frequency 625 (note that this file is not the logger file) + with open(os.path.join(test_path, 'Test1_multifreq_onescan_1000.0.log')) as log_info: + log_info = log_info.readlines() + + # Check timepoints expected + assert check_string(log_info, 'Timepoints expected', '1') + # Check timepoints found + assert check_string(log_info, 'Timepoints found', '0') + # Check sampling frequency + assert check_string(log_info, 'Sampling Frequency', '1000.0') + # Check sampling started + assert check_string(log_info, 'Sampling started', '-1.0000') + # Check start time + assert check_string(log_info, 'first trigger', 'Time 0', is_num=False) + + # Checks json file + with open(os.path.join(test_path, 'Test1_multifreq_onescan_1000.0.json')) as json_file: + json_data = json.load(json_file) + + # Compares values in json file with ground truth + assert math.isclose(json_data['SamplingFrequency'], 1000.0) + assert math.isclose(json_data['StartTime'], -1.0000) + assert json_data['Columns'] == ['time', 'Trigger'] # Remove generated files for filename in glob.glob(os.path.join(test_path, 'phys2bids*')): @@ -229,12 +285,12 @@ def test_integration_multifreq(multifreq_acq_file): os.remove(filename) -def test_integration_heuristic(): +def test_integration_heuristic(samefreq_short_txt_file): """ Does integration test of tutorial file with heurositics """ - test_path = resource_filename('phys2bids', 'tests/data') - test_filename = 'tutorial_file.txt' + + test_path, test_filename = os.path.split(samefreq_short_txt_file) test_full_path = os.path.join(test_path, test_filename) test_chtrig = 1 test_outdir = test_path @@ -242,7 +298,7 @@ def test_integration_heuristic(): test_tr = 1.2 test_thr = 0.735 heur_path = resource_filename('phys2bids', 'heuristics') - test_heur = os.path.join(heur_path, 'heur_tutorial.py') + test_heur = os.path.join(heur_path, 'heur_test_acq.py') phys2bids(filename=test_full_path, chtrig=test_chtrig, outdir=test_outdir, num_timepoints_expected=test_ntp, tr=test_tr, thr=test_thr, sub='006', ses='01', heur_file=test_heur) @@ -250,35 +306,35 @@ def test_integration_heuristic(): test_path_output = os.path.join(test_path, 'sub-006/ses-01/func') # Check that files are generated - base_filename = 'sub-006_ses-01_task-test_rec-labchart_run-01_physio' + base_filename = 'sub-006_ses-01_task-test_rec-biopac_run-01_physio' for suffix in ['.log', '.json', '.tsv.gz']: assert os.path.isfile(os.path.join(test_path_output, base_filename + suffix)) # Read log file (note that this file is not the logger file) - log_filename = 'sub-006_ses-01_task-test_rec-labchart_run-01_physio.log' + log_filename = 'sub-006_ses-01_task-test_rec-biopac_run-01_physio.log' with open(os.path.join(test_path_output, log_filename)) as log_info: log_info = log_info.readlines() # Check timepoints expected assert check_string(log_info, 'Timepoints expected', '158') # Check timepoints found - assert check_string(log_info, 'Timepoints found', '158') + assert check_string(log_info, 'Timepoints found', '0') # Check sampling frequency - assert check_string(log_info, 'Sampling Frequency', '1000.0') + assert check_string(log_info, 'Sampling Frequency', '10000.0') # Check sampling started - assert check_string(log_info, 'Sampling started', '0.2450') - # Check start time + assert check_string(log_info, 'Sampling started', '-189.6000') + # Check first trigger assert check_string(log_info, 'first trigger', 'Time 0', is_num=False) # Checks json file - json_filename = 'sub-006_ses-01_task-test_rec-labchart_run-01_physio.json' + json_filename = 'sub-006_ses-01_task-test_rec-biopac_run-01_physio.json' with open(os.path.join(test_path_output, json_filename)) as json_file: json_data = json.load(json_file) # Compares values in json file with ground truth - assert math.isclose(json_data['SamplingFrequency'], 1000.0) - assert math.isclose(json_data['StartTime'], 0.2450) - assert json_data['Columns'] == ['time', 'Trigger', 'CO2', 'O2', 'Pulse'] + assert math.isclose(json_data['SamplingFrequency'], 10000.0,) + assert math.isclose(json_data['StartTime'], -189.6,) + assert json_data['Columns'] == ['time', 'RESP - RSP100C', 'MR TRIGGER - Custom, HLT100C - A 5'] # Remove generated files for filename in glob.glob(os.path.join(test_path, 'phys2bids*')): @@ -289,12 +345,12 @@ def test_integration_heuristic(): os.remove(filename) -def test_integration_info(): +def test_integration_info(samefreq_short_txt_file): """ Tests the info option """ - test_path = resource_filename('phys2bids', 'tests/data') - test_filename = 'tutorial_file.txt' + + test_path, test_filename = os.path.split(samefreq_short_txt_file) test_chtrig = 1 test_outdir = test_path test_ntp = 158 @@ -312,21 +368,16 @@ def test_integration_info(): subprocess.run(command_str, shell=True, check=True) # Check that plot all file is generated - assert os.path.isfile('tutorial_file.png') + assert os.path.isfile('Test_belt_pulse_samefreq_short.png') # Read logger file logger_file = glob.glob(os.path.join(test_path, '*phys2bids*'))[0] with open(logger_file) as logger_info: logger_info = logger_info.readlines() - # Get trigger info - assert check_string(logger_info, 'Trigger; sampled at', '1000.0') - # Get CO2 info - assert check_string(logger_info, 'CO2; sampled at', '1000.0') - # Get O2 info - assert check_string(logger_info, 'O2; sampled at', '1000.0') - # Get pulse info - assert check_string(logger_info, 'Pulse; sampled at', '1000.0') + assert check_string(logger_info, '01. RESP - RSP100C; sampled at', '10000.0') + assert check_string(logger_info, + '02. MR TRIGGER - Custom, HLT100C - A 5; sampled at', '10000.0') # Remove generated files for filename in glob.glob(os.path.join(test_path, 'phys2bids*')): diff --git a/phys2bids/tests/test_txt.py b/phys2bids/tests/test_txt.py index 4952b9e11..b018d699e 100644 --- a/phys2bids/tests/test_txt.py +++ b/phys2bids/tests/test_txt.py @@ -26,7 +26,7 @@ def loaded_lab_file(multifreq_lab_file): header_lab, channels_lab = txt.read_header_and_channels(multifreq_lab_file, chtrig) # just a few quick checks to make sure the data loaded correctly - assert len(channels_lab[152109 - 9]) == 6 # check the comment has been eliminated + assert len(channels_lab[0]) == 5 assert 'Interval=' in header_lab[0] return header_lab, channels_lab, chtrig @@ -125,4 +125,4 @@ def test_multifreq(loaded_lab_file): header, channels, chtrig = loaded_lab_file phys_obj = txt.process_labchart(channels, chtrig=chtrig, header=header) new_freq = txt.check_multifreq(phys_obj.timeseries, [phys_obj.freq[0]] * len(phys_obj.freq)) - assert new_freq[-4:-1] == [100, 40, 500] + assert new_freq[-3:] == [100, 40, 500]