freesurfer-volume-reader/tests/ashs_test.py

import os
import re

import pandas
import pytest

from freesurfer_volume_reader.ashs import IntracranialVolumeFile, HippocampalSubfieldsVolumeFile

from conftest import SUBJECTS_DIR, assert_volume_frames_equal


@pytest.mark.parametrize(('volume_file_path', 'expected_subject'), [
    ('bert_icv.txt', 'bert'),
    ('final/bert_icv.txt', 'bert'),
    ('ashs/subjects/bert/final/bert_icv.txt', 'bert'),
    ('ashs/subjects/alice/final/long_subject_name_42_icv.txt', 'long_subject_name_42'),
])
def test_intracranial_volume_file_init(volume_file_path, expected_subject):
    volume_file = IntracranialVolumeFile(path=volume_file_path)
    assert os.path.abspath(volume_file_path) == volume_file.absolute_path
    assert expected_subject == volume_file.subject


@pytest.mark.parametrize('volume_file_path', [
    '_icv.txt',
    'bert_ICV.txt',
    'bert_icv.csv',
    'bert_ICV.txt.zip',
])
def test_intracranial_volume_file_init_invalid_filename(volume_file_path):
    with pytest.raises(Exception):
        IntracranialVolumeFile(path=volume_file_path)


@pytest.mark.parametrize(('volume_file_path', 'expected_volume'), [
    (os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_icv.txt'), 1234560),
    (os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_icv.txt'), 1.23456e06),
    (os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_icv.txt'), 1.23456e+06),
    (os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_icv.txt'), float('1.23456e+06')),
    (os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_icv.txt'), 1543200),
])
def test_intracranial_volume_file_read_volume_mm3(volume_file_path, expected_volume):
    volume_file = IntracranialVolumeFile(path=volume_file_path)
    assert expected_volume == pytest.approx(volume_file.read_volume_mm3())


@pytest.mark.parametrize('volume_file_path', [
    os.path.join(SUBJECTS_DIR, 'noone', 'final', 'noone_icv.txt'),
])
def test_intracranial_volume_file_read_volume_mm3_not_found(volume_file_path):
    volume_file = IntracranialVolumeFile(path=volume_file_path)
    with pytest.raises(FileNotFoundError):
        volume_file.read_volume_mm3()


@pytest.mark.parametrize(('volume_file_path', 'expected_series'), [
    (os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_icv.txt'),
     pandas.Series(
         data=[1234560.0],
         name='intercranial_volume_mm^3',
         index=pandas.Index(data=['bert'], name='subject'),
     )),
    (os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_icv.txt'),
     pandas.Series(
         data=[1543200.0],
         name='intercranial_volume_mm^3',
         index=pandas.Index(data=['alice'], name='subject'),
     )),
])
def test_intracranial_volume_file_read_volume_series_single(volume_file_path, expected_series):
    volume_file = IntracranialVolumeFile(path=volume_file_path)
    pandas.testing.assert_series_equal(
        left=expected_series,
        right=volume_file.read_volume_series(),
        check_dtype=True,
        check_names=True,
    )


@pytest.mark.parametrize(('volume_file_paths', 'expected_series'), [
    ([os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_icv.txt'),
      os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_icv.txt')],
     pandas.Series(
         data=[1234560.0, 1543200.0],
         name='intercranial_volume_mm^3',
         index=pandas.Index(data=['bert', 'alice'], name='subject'),
     )),
])
def test_intracranial_volume_file_read_volume_series_concat(volume_file_paths, expected_series):
    volume_series = pandas.concat(
        IntracranialVolumeFile(path=p).read_volume_series()
        for p in volume_file_paths)
    pandas.testing.assert_series_equal(
        left=expected_series,
        right=volume_series,
        check_dtype=True,
        check_names=True,
    )


@pytest.mark.parametrize('volume_file_path', [
    os.path.join(SUBJECTS_DIR, 'bert', 'final', 'BERT_icv.txt'),
])
def test_intracranial_volume_file_read_volume_series_not_found(volume_file_path):
    volume_file = IntracranialVolumeFile(path=volume_file_path)
    with pytest.raises(FileNotFoundError):
        volume_file.read_volume_series()


@pytest.mark.parametrize(('root_dir_path', 'expected_file_paths'), [
    (os.path.join(SUBJECTS_DIR, 'bert'),
     {os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_icv.txt')}),
    (os.path.join(SUBJECTS_DIR, 'alice'),
     {os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_icv.txt')}),
    (SUBJECTS_DIR,
     {os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_icv.txt'),
      os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_icv.txt')}),
])
def test_intracranial_volume_file_find(root_dir_path, expected_file_paths):
    volume_files_iterator = IntracranialVolumeFile.find(
        root_dir_path=root_dir_path)
    assert expected_file_paths == set(
        f.absolute_path for f in volume_files_iterator)


@pytest.mark.parametrize(('root_dir_path', 'filename_pattern', 'expected_file_paths'), [
    (SUBJECTS_DIR,
     r'^\w{4,6}_icv.txt$',
     {os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_icv.txt'),
      os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_icv.txt')}),
    (SUBJECTS_DIR,
     r'^\w{5,6}_icv.txt$',
     {os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_icv.txt')}),
    (SUBJECTS_DIR,
     r'^\w{7,}_icv.txt$',
     set()),
])
def test_intracranial_volume_file_find_pattern(
        root_dir_path, filename_pattern, expected_file_paths):
    volume_files_iterator = IntracranialVolumeFile.find(
        root_dir_path=root_dir_path, filename_regex=re.compile(filename_pattern))
    assert expected_file_paths == set(
        f.absolute_path for f in volume_files_iterator)


@pytest.mark.parametrize(('volume_file_path', 'expected_attrs'), [
    ('ashs/final/bert_left_heur_volumes.txt',
     {'subject': 'bert', 'hemisphere': 'left', 'correction': None}),
    ('ashs/final/bert_left_corr_nogray_volumes.txt',
     {'subject': 'bert', 'hemisphere': 'left', 'correction': 'nogray'}),
    ('ashs/final/bert_left_corr_usegray_volumes.txt',
     {'subject': 'bert', 'hemisphere': 'left', 'correction': 'usegray'}),
    ('ashs/final/bert_right_heur_volumes.txt',
     {'subject': 'bert', 'hemisphere': 'right', 'correction': None}),
    ('ashs/final/bert_right_corr_nogray_volumes.txt',
     {'subject': 'bert', 'hemisphere': 'right', 'correction': 'nogray'}),
    ('ashs/final/bert_right_corr_usegray_volumes.txt',
     {'subject': 'bert', 'hemisphere': 'right', 'correction': 'usegray'}),
    ('somewhere/else/bert_right_heur_volumes.txt',
     {'subject': 'bert', 'hemisphere': 'right', 'correction': None}),
    ('somewhere/else/bert_right_corr_nogray_volumes.txt',
     {'subject': 'bert', 'hemisphere': 'right', 'correction': 'nogray'}),
    ('bert_right_heur_volumes.txt',
     {'subject': 'bert', 'hemisphere': 'right', 'correction': None}),
    ('/foo/bar/alice_20190503_right_corr_nogray_volumes.txt',
     {'subject': 'alice_20190503', 'hemisphere': 'right', 'correction': 'nogray'}),
])
def test_hippocampal_subfields_volume_file_init(volume_file_path, expected_attrs):
    volume_file = HippocampalSubfieldsVolumeFile(path=volume_file_path)
    assert os.path.abspath(volume_file_path) == volume_file.absolute_path
    for attr, value in expected_attrs.items():
        assert value == getattr(volume_file, attr)


@pytest.mark.parametrize('volume_file_path', [
    'bert_middle_heur_volumes.txt',
    'bert_right_hear_volumes.txt',
    'bert_right_heur_volumes.nii',
    'bert_left_lfseg_corr_usegray.nii.gz',
])
def test_hippocampal_subfields_volume_file_init_invalid(volume_file_path):
    with pytest.raises(Exception):
        HippocampalSubfieldsVolumeFile(path=volume_file_path)


@pytest.mark.parametrize(('volume_file_path', 'expected_volumes'), [
    (os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_left_corr_nogray_volumes.txt'),
     {'CA1': 678.901,
      'CA2+3': 123.456,
      'DG': 901.234,
      'ERC': 789.012,
      'PHC': 2345.876,
      'PRC': 2345.678,
      'SUB': 457.789}),
])
def test_hippocampal_subfields_volume_file_read_volumes_mm3(volume_file_path, expected_volumes):
    volume_file = HippocampalSubfieldsVolumeFile(path=volume_file_path)
    assert expected_volumes == volume_file.read_volumes_mm3()


def test_hippocampal_subfields_volume_file_read_volumes_mm3_not_found():
    volume_file = HippocampalSubfieldsVolumeFile(
        path=os.path.join(SUBJECTS_DIR, 'nobert', 'final', 'bert_left_corr_nogray_volumes.txt'))
    with pytest.raises(FileNotFoundError):
        volume_file.read_volumes_mm3()


@pytest.mark.parametrize(('volume_file_path', 'expected_dataframe'), [
    (os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_left_heur_volumes.txt'),
     pandas.DataFrame({
         'subfield': ['CA1', 'CA2+3', 'DG', 'ERC', 'PHC', 'PRC', 'SUB'],
         'volume_mm^3': [679.904, 124.459, 902.237, 789.012, 2346.879, 2346.671, 458.782],
         'subject': 'alice',
         'hemisphere': 'left',
         'correction': None,
     })),
])
def test_hippocampal_subfields_volume_file_read_volumes_dataframe(
        volume_file_path: str, expected_dataframe: pandas.DataFrame):
    volume_file = HippocampalSubfieldsVolumeFile(path=volume_file_path)
    assert_volume_frames_equal(left=expected_dataframe,
                               right=volume_file.read_volumes_dataframe())


def test_hippocampal_subfields_volume_file_read_volumes_dataframe_not_found():
    volume_file = HippocampalSubfieldsVolumeFile(
        path=os.path.join(SUBJECTS_DIR, 'nobert', 'final', 'bert_left_corr_nogray_volumes.txt'))
    with pytest.raises(FileNotFoundError):
        volume_file.read_volumes_dataframe()


@pytest.mark.parametrize(('root_dir_path', 'expected_file_paths'), [
    (os.path.join(SUBJECTS_DIR, 'alice'),
     {os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_left_heur_volumes.txt'),
      os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_left_corr_nogray_volumes.txt')}),
    (os.path.join(SUBJECTS_DIR, 'bert'),
     {os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_left_corr_nogray_volumes.txt'),
      os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_left_corr_usegray_volumes.txt'),
      os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_left_heur_volumes.txt'),
      os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_right_corr_nogray_volumes.txt')}),
    (SUBJECTS_DIR,
     {os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_left_heur_volumes.txt'),
      os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_left_corr_nogray_volumes.txt'),
      os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_left_corr_nogray_volumes.txt'),
      os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_left_corr_usegray_volumes.txt'),
      os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_left_heur_volumes.txt'),
      os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_right_corr_nogray_volumes.txt')}),
])
def test_hippocampal_subfields_volume_file_find(root_dir_path, expected_file_paths):
    volume_files_iterator = HippocampalSubfieldsVolumeFile.find(root_dir_path=root_dir_path)
    assert expected_file_paths == set(f.absolute_path for f in volume_files_iterator)


@pytest.mark.parametrize(('root_dir_path', 'filename_pattern', 'expected_file_paths'), [
    (SUBJECTS_DIR,
     r'^bert_right_',
     {os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_right_corr_nogray_volumes.txt')}),
    (SUBJECTS_DIR,
     r'_nogray_volumes.txt$',
     {os.path.join(SUBJECTS_DIR, 'alice', 'final', 'alice_left_corr_nogray_volumes.txt'),
      os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_left_corr_nogray_volumes.txt'),
      os.path.join(SUBJECTS_DIR, 'bert', 'final', 'bert_right_corr_nogray_volumes.txt')}),
])
def test_hippocampal_subfields_volume_file_find_pattern(root_dir_path, filename_pattern,
                                                        expected_file_paths):
    volume_files_iterator = HippocampalSubfieldsVolumeFile.find(
        root_dir_path=root_dir_path, filename_regex=re.compile(filename_pattern))
    assert expected_file_paths == set(f.absolute_path for f in volume_files_iterator)