diff --git a/phys2bids/heuristics/heur_test_multifreq.py b/phys2bids/heuristics/heur_test_multifreq.py new file mode 100644 index 000000000..c6bdbbd5d --- /dev/null +++ b/phys2bids/heuristics/heur_test_multifreq.py @@ -0,0 +1,71 @@ +import fnmatch + + +def heur(physinfo): + """ + Set of if .. elif statements to fill BIDS names. + + It requires the user (you!) to adjust it accordingly! + It needs an ``if`` or ``elif`` statement for each file that + needs to be processed. + The statement will test if the ``physinfo``: + - is similar to a string (first case), or + - exactly matches a string (second case). + + Parameters + ---------- + physinfo: str + Name of an input file that should be bidsified (See Notes) + + Returns + ------- + info: dictionary of str + Dictionary containing BIDS keys + + Notes + ----- + The `if ..` structure should always be similar to + ``` + if physinfo == 'somepattern': + info['var'] = 'somethingelse' + ``` + or, in case it's a partial match + ``` + if fnmatch.fnmatchcase(physinfo, '*somepattern?'): + info['var'] = 'somethingelse' + ``` + Where: + - `physinfo` and `info` are dedicated keywords, + - 'somepattern' is the name of the file, + - 'var' is a bids key in the list below + - 'somethingelse' is the value of the key + """ + info = {} + # ################################# # + # ## Modify here! ## # + # ## ## # + # ## Possible variables are: ## # + # ## -info['task'] (required) ## # + # ## -info['run'] ## # + # ## -info['rec'] ## # + # ## -info['acq'] ## # + # ## -info['dir'] ## # + # ## ## # + # ## Remember that they are ## # + # ## dictionary keys ## # + # ## See example below ## # + # ################################# # + + if fnmatch.fnmatchcase(physinfo, '*onescan*'): + info['task'] = 'test' + info['run'] = '01' + info['rec'] = 'biopac' + elif physinfo == 'Example': + info['task'] = 'rest' + info['run'] = '01' + info['acq'] = 'resp' + + # ############################## # + # ## Don't modify below this! ## # + # ############################## # + return info diff --git a/phys2bids/tests/test_integration.py b/phys2bids/tests/test_integration.py index ebb6cb6df..6ba565157 100644 --- a/phys2bids/tests/test_integration.py +++ b/phys2bids/tests/test_integration.py @@ -6,6 +6,7 @@ import shutil import subprocess from csv import reader +from os.path import join as opj from pkg_resources import resource_filename from phys2bids._version import get_versions @@ -23,92 +24,6 @@ def check_string(str_container, str_to_find, str_expected, is_num=True): return str_expected in str_found -def test_logger(multifreq_lab_file): - """ - Tests the logger - """ - - test_path, test_filename = os.path.split(multifreq_lab_file) - - # Input arguments - test_chtrig = 3 - test_ntp = 1 - test_outdir = test_path - conversion_path = os.path.join(test_path, 'code/conversion') - # Phys2bids call through terminal - subprocess.run(f'phys2bids -in {test_filename} -indir {test_path} ' - f'-chtrig {test_chtrig} -ntp {test_ntp} -outdir {test_outdir}', - shell=True, check=True) - - assert os.path.isfile(os.path.join(conversion_path, 'call.sh')) - - # Read logger file - logger_file = glob.glob(os.path.join(conversion_path, '*phys2bids*'))[0] - with open(logger_file) as logger_info: - logger_info = logger_info.readlines() - - # Get version info - current_version = get_versions() - assert check_string(logger_info, 'phys2bids version', current_version['version'], is_num=False) - - # Removes generated files - os.remove(os.path.join(test_path, logger_file)) - shutil.rmtree(conversion_path) - - -def test_integration_txt(samefreq_short_txt_file): - """ - Does an integration test with the tutorial file - """ - - test_path, test_filename = os.path.split(samefreq_short_txt_file) - test_chtrig = 2 - conversion_path = os.path.join(test_path, 'code/conversion') - - 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 ['.json', '.tsv.gz']: - assert os.path.isfile(os.path.join(test_path, 'Test_belt_pulse_samefreq_short' + suffix)) - - # Check files in extra are generated - for suffix in ['.log', '_trigger_time.png']: - assert os.path.isfile(os.path.join(conversion_path, - 'Test_belt_pulse_samefreq_short' + suffix)) - - # Read log file (note that this file is not the logger file) - with open(os.path.join(conversion_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', '1') - # Check timepoints found - assert check_string(log_info, 'Timepoints found', '60') - # Check sampling frequency - 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, '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'], 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) - shutil.rmtree(conversion_path) - - def test_integration_acq(samefreq_full_acq_file): """ Does the integration test for an acq file @@ -116,21 +31,21 @@ def test_integration_acq(samefreq_full_acq_file): test_path, test_filename = os.path.split(samefreq_full_acq_file) test_chtrig = 3 - conversion_path = os.path.join(test_path, 'code/conversion') + conversion_path = opj(test_path, 'code/conversion') 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 ['.json', '.tsv.gz']: - assert os.path.isfile(os.path.join(test_path, 'Test_belt_pulse_samefreq' + suffix)) + assert os.path.isfile(opj(test_path, 'Test_belt_pulse_samefreq' + suffix)) # Check files in extra are generated for suffix in ['.log', '_trigger_time.png']: - assert os.path.isfile(os.path.join(conversion_path, 'Test_belt_pulse_samefreq' + suffix)) + assert os.path.isfile(opj(conversion_path, 'Test_belt_pulse_samefreq' + suffix)) # Read log file (note that this file is not the logger file) - with open(os.path.join(conversion_path, 'Test_belt_pulse_samefreq.log')) as log_info: + with open(opj(conversion_path, 'Test_belt_pulse_samefreq.log')) as log_info: log_info = log_info.readlines() # Check timepoints expected @@ -145,7 +60,7 @@ def test_integration_acq(samefreq_full_acq_file): 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_samefreq.json')) as json_file: + with open(opj(test_path, 'Test_belt_pulse_samefreq.json')) as json_file: json_data = json.load(json_file) # Compares values in json file with ground truth @@ -155,271 +70,127 @@ def test_integration_acq(samefreq_full_acq_file): 'MR TRIGGER - Custom, HLT100C - A 5', 'PPG100C', 'CO2', 'O2'] # Remove generated files - for filename in glob.glob(os.path.join(conversion_path, 'phys2bids*')): + for filename in glob.glob(opj(conversion_path, 'phys2bids*')): os.remove(filename) - for filename in glob.glob(os.path.join(test_path, 'Test_belt_pulse_samefreq*')): + for filename in glob.glob(opj(test_path, 'Test_belt_pulse_samefreq*')): os.remove(filename) shutil.rmtree(conversion_path) -def test_integration_multifreq(multifreq_lab_file): +def test_integration_heuristic(multifreq_lab_file): """ - Does the integration test for a multi-frequency file + Does integration test of tutorial file with heurositics """ test_path, test_filename = os.path.split(multifreq_lab_file) + test_full_path = opj(test_path, test_filename) test_chtrig = 3 - conversion_path = os.path.join(test_path, 'code/conversion') - - 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 ['.json', '.tsv.gz']: - assert os.path.isfile(os.path.join(test_path, - 'Test1_multifreq_onescan_40.0' + suffix)) - for suffix in ['.json', '.tsv.gz']: - assert os.path.isfile(os.path.join(test_path, - 'Test1_multifreq_onescan_100.0' + suffix)) - for suffix in ['.json', '.tsv.gz']: - assert os.path.isfile(os.path.join(test_path, - 'Test1_multifreq_onescan_500.0' + suffix)) - for suffix in ['.json', '.tsv.gz']: - assert os.path.isfile(os.path.join(test_path, - 'Test1_multifreq_onescan_1000.0' + suffix)) - for freq in ['40', '100', '500', '1000']: - assert os.path.isfile(os.path.join(conversion_path, - 'Test1_multifreq_onescan_' + freq + '.log')) - assert os.path.isfile(os.path.join(conversion_path, - 'Test1_multifreq_onescan_trigger_time.png')) - - """ - Checks 40 Hz output - """ - # Read log file of frequency 625 (note that this file is not the logger file) - with open(os.path.join(conversion_path, 'Test1_multifreq_onescan_40.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', '40.0') - # Check sampling frequency - 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, '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'], 40.0) - assert math.isclose(json_data['StartTime'], -157.8535) - assert json_data['Columns'] == ['O2'] - - """ - Checks 100 Hz output - """ - # Read log file of frequency 625 (note that this file is not the logger file) - with open(os.path.join(conversion_path, 'Test1_multifreq_onescan_100.log')) as log_info: - log_info = log_info.readlines() + test_outdir = test_path + conversion_path = opj(test_path, 'code/conversion') + test_ntp = 158 + test_tr = 1.2 + test_thr = 0.735 + heur_path = resource_filename('phys2bids', 'heuristics') + test_heur = opj(heur_path, 'heur_test_multifreq.py') - # 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', '100.0') - # Check sampling started - assert check_string(log_info, 'Sampling started', '-0.3057') - # Check start time - assert check_string(log_info, 'first trigger', 'Time 0', is_num=False) + # Move into folder + subprocess.run(f'cd {test_path}', shell=True, check=True) + # Phys2bids call through terminal + command_str = (f'phys2bids -in {test_full_path} ', + f'-chtrig {test_chtrig} -outdir {test_outdir} ', + f'-tr {test_tr} -ntp {test_ntp} -thr {test_thr} ', + f'-sub 006 -ses 01 -heur {test_heur}') + command_str = ''.join(command_str) + subprocess.run(command_str, shell=True, check=True) - # Checks json file - with open(os.path.join(test_path, 'Test1_multifreq_onescan_100.0.json')) as json_file: - json_data = json.load(json_file) + # Check that call.sh is generated + assert os.path.isfile(opj(conversion_path, 'call.sh')) - # Compares values in json file with ground truth - assert math.isclose(json_data['SamplingFrequency'], 100.0) - assert math.isclose(json_data['StartTime'], -0.3057) - assert json_data['Columns'] == ['CO2'] + # Read logger file + logger_file = glob.glob(opj(conversion_path, '*phys2bids*'))[0] + with open(logger_file) as logger_info: + logger_info = logger_info.readlines() - """ - Checks 500 Hz output - """ - # Read log file of frequency 625 (note that this file is not the logger file) - with open(os.path.join(conversion_path, 'Test1_multifreq_onescan_500.log')) as log_info: - log_info = log_info.readlines() + # Get version info + current_version = get_versions() + assert check_string(logger_info, 'phys2bids version', current_version['version'], is_num=False) - # 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) + assert check_string(logger_info, '01. Trigger; sampled at', '1000.0') + assert check_string(logger_info, '04. Belt; sampled at', '500.0') - # 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) + # Check that files are generated in conversion path + for freq in ['40', '100', '500', '1000']: + assert os.path.isfile(opj(conversion_path, + 'sub-006_ses-01_task-test_rec-biopac_run-01_recording-' + + freq + '.log')) + assert os.path.isfile(opj(conversion_path, + 'Test1_multifreq_onescan_sub-006_ses-01_trigger_time.png')) + assert os.path.isfile(opj(conversion_path, 'Test1_multifreq_onescan.png')) + assert os.path.isfile(opj(conversion_path, 'heur_test_multifreq.py')) + test_path_output = opj(test_path, 'sub-006/ses-01/func') - # 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'] + # Check that files are generated + base_filename = 'sub-006_ses-01_task-test_rec-biopac_run-01_recording-' + for suffix in ['.json', '.tsv.gz']: + for freq in ['40.0', '100.0', '500.0', '1000.0']: + assert os.path.isfile(opj(test_path_output, base_filename + freq + '_physio' + suffix)) - """ - Checks 1000 Hz output - """ - # Read log file of frequency 625 (note that this file is not the logger file) - with open(os.path.join(conversion_path, 'Test1_multifreq_onescan_1000.log')) as log_info: + ###### Checks for 40 Hz files + # Read log file (note that this file is not the logger file) + log_filename = 'sub-006_ses-01_task-test_rec-biopac_run-01_recording-40.log' + with open(opj(conversion_path, log_filename)) as log_info: log_info = log_info.readlines() # Check timepoints expected - assert check_string(log_info, 'Timepoints expected', '1') + assert check_string(log_info, 'Timepoints expected', '158') # Check timepoints found - assert check_string(log_info, 'Timepoints found', '0') + assert check_string(log_info, 'Timepoints found', '1') # Check sampling frequency - assert check_string(log_info, 'Sampling Frequency', '1000.0') + assert check_string(log_info, 'Sampling Frequency', '40.0') # Check sampling started - assert check_string(log_info, 'Sampling started', '-1.0000') - # Check start time + assert check_string(log_info, 'Sampling started', '-157.8535') + # Check first trigger 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_filename = 'sub-006_ses-01_task-test_rec-biopac_run-01_recording-40.0_physio.json' + with open(opj(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'], -1.0000) - assert json_data['Columns'] == ['time', 'Trigger'] - - # Remove generated files - for filename in glob.glob(os.path.join(conversion_path, 'phys2bids*')): - os.remove(filename) - for filename in glob.glob(os.path.join(test_path, 'Test_belt_pulse_multifreq*')): - os.remove(filename) - shutil.rmtree(conversion_path) - - -def test_integration_heuristic(samefreq_short_txt_file): - """ - Does integration test of tutorial file with heurositics - """ - - 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 - conversion_path = os.path.join(test_path, 'code/conversion') - test_ntp = 158 - test_tr = 1.2 - test_thr = 0.735 - heur_path = resource_filename('phys2bids', 'heuristics') - 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) - - test_path_output = os.path.join(test_path, 'sub-006/ses-01/func') + assert math.isclose(json_data['SamplingFrequency'], 40.0,) + assert math.isclose(json_data['StartTime'], -157.8535,) + assert json_data['Columns'] == ['O2'] - # Check that files are generated - base_filename = 'sub-006_ses-01_task-test_rec-biopac_run-01_physio' - for suffix in ['.json', '.tsv.gz']: - assert os.path.isfile(os.path.join(test_path_output, base_filename + suffix)) - assert os.path.isfile(os.path.join(conversion_path, base_filename + '.log')) + ###### Checks for 100 Hz files # Read log file (note that this file is not the logger file) - log_filename = 'sub-006_ses-01_task-test_rec-biopac_run-01_physio.log' - with open(os.path.join(conversion_path, log_filename)) as log_info: + log_filename = 'sub-006_ses-01_task-test_rec-biopac_run-01_recording-100.log' + with open(opj(conversion_path, 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', '0') + assert check_string(log_info, 'Timepoints found', '1') # 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', '-189.6000') + assert check_string(log_info, 'Sampling started', '-0.3057') # 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-biopac_run-01_physio.json' - with open(os.path.join(test_path_output, json_filename)) as json_file: + json_filename = 'sub-006_ses-01_task-test_rec-biopac_run-01_recording-100.0_physio.json' + with open(opj(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'], 10000.0,) - assert math.isclose(json_data['StartTime'], -189.6,) - assert json_data['Columns'] == ['time', 'RESP - RSP100C', 'MR TRIGGER - Custom, HLT100C - A 5'] - - # Check that participant.tsv gets updated - phys2bids(filename=test_full_path, chtrig=test_chtrig, outdir=test_outdir, - num_timepoints_expected=test_ntp, tr=test_tr, thr=test_thr, sub='002', - ses='01', heur_file=test_heur) - - counter = 0 - subject_list = ['participant_id', '006', '002'] - with open(os.path.join(test_path, 'participants.tsv')) as pf: - tsvreader = reader(pf, delimiter="\t") - for line in tsvreader: - assert subject_list[counter] in line[0] - counter += 1 + assert math.isclose(json_data['SamplingFrequency'], 100.0,) + assert math.isclose(json_data['StartTime'], -0.3057,) + assert json_data['Columns'] == ['CO2'] # Remove generated files - for filename in glob.glob(os.path.join(conversion_path, 'phys2bids*')): - os.remove(filename) - for filename in glob.glob(os.path.join(test_path, 'Test_belt_pulse_samefreq*')): - os.remove(filename) - for filename in glob.glob(os.path.join(test_path_output, '*')): - os.remove(filename) + shutil.rmtree(test_path_output) shutil.rmtree(conversion_path) - - -def test_integration_info(samefreq_short_txt_file): - """ - Tests the info option - """ - - test_path, test_filename = os.path.split(samefreq_short_txt_file) - test_chtrig = 1 - test_outdir = test_path - test_ntp = 158 - test_tr = 1.2 - test_thr = 0.735 - conversion_path = os.path.join(test_path, 'code/conversion') - # Move into folder - subprocess.run(f'cd {test_path}', shell=True, check=True) - # Phys2bids call through terminal - command_str = (f'phys2bids -in {test_filename} -indir {test_path} ', - f'-chtrig {test_chtrig} -outdir {test_outdir} ', - f'-tr {test_tr} -ntp {test_ntp} -thr {test_thr} ', - f'-info') - command_str = ''.join(command_str) - subprocess.run(command_str, shell=True, check=True) - - # Check that plot all file is generated - assert os.path.isfile(os.path.join(conversion_path, - 'Test_belt_pulse_samefreq_short.png')) - - # Read logger file - logger_file = glob.glob(os.path.join(conversion_path, '*phys2bids*'))[0] - with open(logger_file) as logger_info: - logger_info = logger_info.readlines() - - 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(conversion_path, 'phys2bids*')): + for filename in glob.glob(opj(test_path, 'Test1_multifreq_onescan*')): os.remove(filename) - shutil.rmtree(conversion_path)