docs/builtin.rst

@@ -285,6 +285,14 @@ Available
        | unveiled with functional connectivity gradients.
        | Nature Neuroscience, Volume 23, Pages 1421–1432 (2020).
        | https://doi.org/10.1038/s41593-020-00711-6
+   * - AICHA
+     - ``version``
+     - ``AICHA_v1``, ``AICHA_v2``
+     - 0.0.3
+     - | Joliot, M., Jobard, G., Naveau, M. et al.
+       | AICHA: An atlas of intrinsic connectivity of homotopic areas.
+       | Journal of Neuroscience Methods, Volume 254, Pages 46-59 (2015).
+       | https://doi.org/10.1016/j.jneumeth.2015.07.013
 
 
 Planned

docs/changes/newsfragments/173.feature

@@ -0,0 +1 @@
+Add ``AICHA v1`` and ``AICHA v2`` parcellations to ``junifer.data`` by `Synchon Mandal`_

junifer/api/cli.py

@@ -137,6 +137,7 @@ def run(filepath: click.Path, element: str, verbose: Union[str, int]) -> None:
     """Run command for CLI.
 
     \f
+
     Parameters
     ----------
     filepath : click.Path
@@ -185,6 +186,7 @@ def collect(filepath: click.Path, verbose: Union[str, int]) -> None:
     """Collect command for CLI.
 
     \f
+
     Parameters
     ----------
     filepath : click.Path
@@ -228,6 +230,7 @@ def queue(
     """Queue command for CLI.
 
     \f
+
     Parameters
     ----------
     filepath : click.Path
@@ -266,6 +269,7 @@ def wtf(long_: bool) -> None:
     """Wtf command for CLI.
 
     \f
+
     Parameters
     ----------
     long_ : bool
@@ -288,6 +292,7 @@ def selftest(subpkg: str) -> None:
     """Selftest command for CLI.
 
     \f
+
     Parameters
     ----------
     subpkg : {"all", "api", "configs", "data", "datagrabber", "datareader",

junifer/data/parcellations.py

@@ -7,6 +7,7 @@
 
 import io
 import shutil
+import tarfile
 import tempfile
 import zipfile
 from pathlib import Path
@@ -17,6 +18,7 @@ import numpy as np
 import pandas as pd
 import requests
 from nilearn import datasets, image
+from requests.exceptions import ConnectionError, HTTPError, ReadTimeout
 
 from ..utils.logging import logger, raise_error, warn_with_log
 from .utils import closest_resolution
@@ -74,6 +76,12 @@ for scale in range(1, 5):
         "magneticfield": "3T",
         "space": "MNInonlinear2009cAsym",
     }
+# Add AICHA parcellation info
+for version in (1, 2):
+    _available_parcellations[f"AICHA_v{version}"] = {
+        "family": "AICHA",
+        "version": version,
+    }
 
 
 def register_parcellation(
@@ -151,7 +159,6 @@ def list_parcellations() -> List[str]:
 #     return resolution
 
 
-# TODO: keyword arguments are not passed, check
 def load_parcellation(
     name: str,
     parcellations_dir: Union[str, Path, None] = None,
@@ -245,8 +252,8 @@ def _retrieve_parcellation(
 
     Parameters
     ----------
-    family : str
-        The name of the parcellation's family, e.g. 'Schaefer'.
+    family : {"Schaefer", "SUIT", "Tian", "AICHA"}
+        The name of the parcellation family.
     parcellations_dir : str or pathlib.Path, optional
         Path where the retrieved parcellations file are stored. The default
         location is "$HOME/junifer/data/parcellations" (default None).
@@ -278,6 +285,9 @@ def _retrieve_parcellation(
         ``space`` : {"MNI", "SUIT"}, optional
             Space of parcellation (default "MNI"). (For more information
             see http://www.diedrichsenlab.org/imaging/suit.htm).
+    * AICHA :
+        ``version`` : {1, 2}, optional
+            Version of parcellation (default 2).
 
     Returns
     -------
@@ -306,19 +316,25 @@ def _retrieve_parcellation(
 
     # Retrieval details per family
     if family == "Schaefer":
-        parcellation_fname, parcellation_labesl = _retrieve_schaefer(
+        parcellation_fname, parcellation_labels = _retrieve_schaefer(
             parcellations_dir=parcellations_dir,
             resolution=resolution,
             **kwargs,
         )
     elif family == "SUIT":
-        parcellation_fname, parcellation_labesl = _retrieve_suit(
+        parcellation_fname, parcellation_labels = _retrieve_suit(
             parcellations_dir=parcellations_dir,
             resolution=resolution,
             **kwargs,
         )
     elif family == "Tian":
-        parcellation_fname, parcellation_labesl = _retrieve_tian(
+        parcellation_fname, parcellation_labels = _retrieve_tian(
+            parcellations_dir=parcellations_dir,
+            resolution=resolution,
+            **kwargs,
+        )
+    elif family == "AICHA":
+        parcellation_fname, parcellation_labels = _retrieve_aicha(
             parcellations_dir=parcellations_dir,
             resolution=resolution,
             **kwargs,
@@ -328,7 +344,7 @@ def _retrieve_parcellation(
             f"The provided parcellation name {family} cannot be retrieved."
         )
 
-    return parcellation_fname, parcellation_labesl
+    return parcellation_fname, parcellation_labels
 
 
 def _retrieve_schaefer(
@@ -342,7 +358,7 @@ def _retrieve_schaefer(
     Parameters
     ----------
     parcellations_dir : pathlib.Path
-        The path to the parcallations' data directory.
+        The path to the parcellation data directory.
     resolution : float, optional
         The desired resolution of the parcellation to load. If it is not
         available, the closest resolution will be loaded. Preferably, use a
@@ -364,7 +380,7 @@ def _retrieve_schaefer(
     Raises
     ------
     ValueError
-        If invalid value is provided for `n_rois` or `yeo_networks` or if
+        If invalid value is provided for ``n_rois`` or ``yeo_networks`` or if
         there is a problem fetching the parcellation.
 
     """
@@ -447,7 +463,6 @@ def _retrieve_tian(
     ----------
     parcellations_dir : pathlib.Path
         The path to the parcellation data directory.
-    resolution : float, optional
     resolution : float, optional
         The desired resolution of the parcellation to load. If it is not
         available, the closest resolution will be loaded. Preferably, use a
@@ -472,8 +487,8 @@ def _retrieve_tian(
     Raises
     ------
     ValueError
-        If invalid value is provided for `scale` or `magneticfield` or `space`
-        or if there is a problem fetching the parcellation.
+        If invalid value is provided for ``scale`` or ``magneticfield`` or
+        ``space`` or if there is a problem fetching the parcellation.
 
     """
     # show parameters to user
@@ -632,7 +647,7 @@ def _retrieve_suit(
     Raises
     ------
     ValueError
-        If invalid value is provided for `space` or if there is a problem
+        If invalid value is provided for ``space`` or if there is a problem
         fetching the parcellation.
 
     """
@@ -707,6 +722,143 @@ def _retrieve_suit(
     return parcellation_fname, labels
 
 
+def _retrieve_aicha(
+    parcellations_dir: Path,
+    resolution: Optional[float] = None,
+    version: int = 2,
+) -> Tuple[Path, List[str]]:
+    """Retrieve AICHA parcellation.
+
+    Parameters
+    ----------
+    parcellations_dir : pathlib.Path
+        The path to the parcellation data directory.
+    resolution : float, optional
+        The desired resolution of the parcellation to load. If it is not
+        available, the closest resolution will be loaded. Preferably, use a
+        resolution higher than the desired one. By default, will load the
+        highest one (default None). Available resolution for this
+        parcellation is 2mm.
+    version : {1, 2}, optional
+        The version of the parcellation to use (default 2).
+
+    Returns
+    -------
+    pathlib.Path
+        File path to the parcellation image.
+    list of str
+        Parcellation labels.
+
+    Raises
+    ------
+    ValueError
+        If invalid value is provided for ``version`` or if there is a problem
+        fetching the parcellation.
+
+    Notes
+    -----
+    The resolution of the parcellation is 2mm and although v2 provides
+    1mm, it is only for display purpose as noted in the release document.
+
+    """
+    # show parameters to user
+    logger.info("Parcellation parameters:")
+    logger.info(f"\tversion: {version}")
+
+    # Check version value
+    _valid_version = (1, 2)
+    if version not in _valid_version:
+        raise_error(
+            f"The parameter `version` ({version}) needs to be one of the "
+            f"following: {_valid_version}"
+        )
+
+    _valid_resolutions = [1]
+    resolution = closest_resolution(resolution, _valid_resolutions)
+
+    # Define image file
+    parcellation_fname = (
+        parcellations_dir / f"AICHA_v{version}" / "AICHA" / "AICHA.nii"
+    )
+
+    # Define label file name according to version
+    if version == 1:
+        parcellation_lname = (
+            parcellations_dir
+            / f"AICHA_v{version}"
+            / "AICHA"
+            / "AICHA_vol1.txt"
+        )
+    elif version == 2:
+        parcellation_lname = (
+            parcellations_dir
+            / f"AICHA_v{version}"
+            / "AICHA"
+            / "AICHA_vol3.txt"
+        )
+
+    # Check existence of parcellation
+    if not (parcellation_fname.exists() and parcellation_lname.exists()):
+        logger.info(
+            "At least one of the parcellation files are missing, fetching."
+        )
+
+        # Set file name on server according to version
+        if version == 1:
+            server_filename = "aicha_v1.zip"
+        elif version == 2:
+            server_filename = "AICHA_v2.tar.zip"
+
+        # Set URL
+        url = f"http://www.gin.cnrs.fr/wp-content/uploads/{server_filename}"
+
+        logger.info(f"Downloading AICHA v{version} from {url}")
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # Make HTTP request
+            try:
+                resp = requests.get(url)
+                resp.raise_for_status()
+            except (ConnectionError, ReadTimeout, HTTPError) as err:
+                raise_error(
+                    f"Failed to download AICHA v{version} due to: {err}"
+                )
+            else:
+                parcellation_zip_path = Path(tmpdir) / server_filename
+                with open(parcellation_zip_path, "wb") as f:
+                    f.write(resp.content)
+
+            # Extract zipfile
+            with zipfile.ZipFile(parcellation_zip_path, "r") as zip_ref:
+                if version == 1:
+                    zip_ref.extractall(
+                        (parcellations_dir / "AICHA_v1").as_posix()
+                    )
+                elif version == 2:
+                    zip_ref.extractall(Path(tmpdir).as_posix())
+                    # Extract tarfile for v2
+                    with tarfile.TarFile(
+                        Path(tmpdir) / "aicha_v2.tar", "r"
+                    ) as tar_ref:
+                        tar_ref.extractall(
+                            (parcellations_dir / "AICHA_v2").as_posix()
+                        )
+
+            # Cleanup after unzipping
+            if (parcellations_dir / f"AICHA_v{version}" / "__MACOSX").exists():
+                shutil.rmtree(
+                    (
+                        parcellations_dir / f"AICHA_v{version}" / "__MACOSX"
+                    ).as_posix()
+                )
+
+    # Load labels
+    labels = pd.read_csv(
+        parcellation_lname, sep="\t", header=None, skiprows=[0]  # type: ignore
+    )[0].to_list()
+
+    return parcellation_fname, labels
+
+
 def merge_parcellations(
     parcellations_list: List["Nifti1Image"],
     parcellations_names: List[str],

junifer/data/tests/test_parcellations.py

@@ -15,6 +15,7 @@ from nilearn.image import new_img_like
 from numpy.testing import assert_array_almost_equal, assert_array_equal
 
 from junifer.data.parcellations import (
+    _retrieve_aicha,
     _retrieve_parcellation,
     _retrieve_schaefer,
     _retrieve_suit,
@@ -168,6 +169,8 @@ def test_register_parcellation(
 @pytest.mark.parametrize(
     "parcellation_name",
     [
+        "AICHA_v1",
+        "AICHA_v2",
         "SUITxSUIT",
         "SUITxMNI",
         "Schaefer100x7",
@@ -552,6 +555,46 @@ def test_retrieve_tian_incorrect_scale(tmp_path: Path) -> None:
         )
 
 
+@pytest.mark.parametrize("version", [1, 2])
+def test_aicha(tmp_path: Path, version: int) -> None:
+    """Test AICHA parcellation.
+
+    Parameters
+    ----------
+    tmp_path : pathlib.Path
+        The path to the test directory.
+    version : int
+        The parametrized version values.
+
+    """
+    parcellations = list_parcellations()
+    assert f"AICHA_v{version}" in parcellations
+    # Load parcellation
+    img, label, img_path = load_parcellation(
+        name=f"AICHA_v{version}", parcellations_dir=tmp_path
+    )
+    assert img is not None
+    assert img_path.name == "AICHA.nii"
+    assert len(label) == 384
+    assert_array_equal(img.header["pixdim"][1:4], [2, 2, 2])  # type: ignore
+
+
+def test_retrieve_aicha_incorrect_version(tmp_path: Path) -> None:
+    """Test retrieve AICHA with incorrect version.
+
+    Parameters
+    ----------
+    tmp_path : pathlib.Path
+        The path to the test directory.
+
+    """
+    with pytest.raises(ValueError, match="The parameter `version`"):
+        _retrieve_aicha(
+            parcellations_dir=tmp_path,
+            version=100,
+        )
+
+
 def test_merge_parcellations() -> None:
     """Test merging parcellations."""
     # load some parcellations for testing