Merge d1b18ef into a5aae08

README.md

@@ -72,6 +72,17 @@ for volume_file in ashs.HippocampalSubfieldsVolumeFile.find('/my/ashs/subjects')
     print(volume_file.read_volumes_dataframe())
 ```
 
+#### Intracranial Volume
+
+```python
+from freesurfer_volume_reader import ashs
+
+for volume_file in ashs.IntracranialVolumeFile.find('/my/ashs/subjects'):
+    print(volume_file.subject)
+    print(volume_file.read_volume_mm3())
+    print(volume_file.read_volume_series())
+```
+
 ### Freesurfer & ASHS
 
 ```sh

examples/compare_ashs_freesurfer_hipposf.ipynb

@@ -115,9 +115,9 @@
    ],
    "source": [
     "import os, pandas\n",
-    "from freesurfer_volume_reader import VolumeFile, ashs, freesurfer\n",
+    "from freesurfer_volume_reader import SubfieldVolumeFile, ashs, freesurfer\n",
     "\n",
-    "def read_volume_file(volume_file: VolumeFile) -> pandas.DataFrame:\n",
+    "def read_volume_file(volume_file: SubfieldVolumeFile) -> pandas.DataFrame:\n",
     "    volume_frame = volume_file.read_volumes_dataframe()\n",
     "    volume_frame['source_basename'] = os.path.basename(volume_file.absolute_path)\n",
     "    return volume_frame\n",

freesurfer_volume_reader/__init__.py

@@ -45,20 +45,23 @@ class VolumeFile(metaclass=abc.ABCMeta):
     def absolute_path(self):
         raise NotImplementedError()
 
-    @abc.abstractmethod
-    def read_volumes_mm3(self) -> typing.Dict[str, float]:
-        raise NotImplementedError()
-
-    @abc.abstractmethod
-    def read_volumes_dataframe(self) -> pandas.DataFrame:
-        raise NotImplementedError()
-
     @classmethod
     def find(cls, root_dir_path: str,
              filename_regex: typing.Optional[typing.Pattern] = None,
-             ) -> typing.Iterator['VolumeFile']:
+             ) -> typing.Iterator['SubfieldVolumeFile']:
         if not filename_regex:
             filename_regex = cls.FILENAME_REGEX
         for dirpath, _, filenames in os.walk(root_dir_path):
             for filename in filter(filename_regex.search, filenames):
                 yield cls(path=os.path.join(dirpath, filename))
+
+
+class SubfieldVolumeFile(VolumeFile):
+
+    @abc.abstractmethod
+    def read_volumes_mm3(self) -> typing.Dict[str, float]:
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def read_volumes_dataframe(self) -> pandas.DataFrame:
+        raise NotImplementedError()

freesurfer_volume_reader/ashs.py

@@ -20,7 +20,38 @@
 import freesurfer_volume_reader
 
 
-class HippocampalSubfieldsVolumeFile(freesurfer_volume_reader.VolumeFile):
+class IntracranialVolumeFile(freesurfer_volume_reader.VolumeFile):
+
+    FILENAME_REGEX = re.compile(r'^(?P<s>\w+)_icv.txt$')
+
+    def __init__(self, path: str):
+        self._absolute_path = os.path.abspath(path)
+        filename_match = self.FILENAME_REGEX.match(os.path.basename(path))
+        assert filename_match, self._absolute_path
+        self.subject = filename_match.groupdict()['s']
+
+    @property
+    def absolute_path(self):
+        return self._absolute_path
+
+    def read_volume_mm3(self) -> float:
+        with open(self.absolute_path, 'r') as volume_file:
+            subject, icv = volume_file.read().rstrip().split(' ')
+            assert subject == self.subject, (subject, self.subject)
+            return float(icv)
+
+    def read_volume_series(self) -> pandas.Series:
+        return pandas.Series(
+            data=[self.read_volume_mm3()],
+            name='volume_mm^3',
+            index=pandas.Index(
+                data=[self.subject],
+                name='subject',
+            ),
+        )
+
+
+class HippocampalSubfieldsVolumeFile(freesurfer_volume_reader.SubfieldVolumeFile):
 
     # https://sites.google.com/site/hipposubfields/tutorial#TOC-Viewing-ASHS-Segmentation-Results
     FILENAME_PATTERN = r'^(?P<s>\w+)_(?P<h>left|right)' \

freesurfer_volume_reader/freesurfer.py

@@ -20,7 +20,7 @@
 import freesurfer_volume_reader
 
 
-class HippocampalSubfieldsVolumeFile(freesurfer_volume_reader.VolumeFile):
+class HippocampalSubfieldsVolumeFile(freesurfer_volume_reader.SubfieldVolumeFile):
 
     # https://surfer.nmr.mgh.harvard.edu/fswiki/HippocampalSubfields
     FILENAME_PATTERN = r'^(?P<h>[lr])h\.hippoSfVolumes' \

tests/ashs_test.py

@@ -4,11 +4,142 @@
 import pandas
 import pytest
 
-from freesurfer_volume_reader.ashs import HippocampalSubfieldsVolumeFile
+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='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='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='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}),

tests/init_test.py

@@ -1,7 +1,8 @@
 import pytest
 
-from freesurfer_volume_reader import __version__, parse_version_string, \
-                                     remove_group_names_from_regex, VolumeFile
+from freesurfer_volume_reader import \
+    __version__, parse_version_string, remove_group_names_from_regex, \
+    VolumeFile, SubfieldVolumeFile
 
 
 def test_module_version():
@@ -45,15 +46,30 @@ class DummyVolumeFile(VolumeFile):
     def absolute_path(self):
         return super().absolute_path
 
+
+def test_volume_file_abstractmethod():
+    volume_file = DummyVolumeFile()
+    with pytest.raises(NotImplementedError):
+        assert volume_file.absolute_path
+
+
+class DummySubfieldVolumeFile(SubfieldVolumeFile):
+
+    # pylint: disable=useless-super-delegation
+
+    @property
+    def absolute_path(self):
+        return super().absolute_path
+
     def read_volumes_mm3(self):
         return super().read_volumes_mm3()
 
     def read_volumes_dataframe(self):
         return super().read_volumes_dataframe()
 
 
-def test_volume_file_abstractmethod():
-    volume_file = DummyVolumeFile()
+def test_subfield_volume_file_abstractmethod():
+    volume_file = DummySubfieldVolumeFile()
     with pytest.raises(NotImplementedError):
         assert volume_file.absolute_path
     with pytest.raises(NotImplementedError):

tests/subjects/alice/final/alice_icv.txt

@@ -0,0 +1 @@
+alice 1.5432e+06

tests/subjects/bert/final/bert_icv.txt

@@ -0,0 +1 @@
+bert 1.23456e+06