Caps generator

pyproject.toml

@@ -27,6 +27,8 @@ classifiers = [
 
 [tool.poetry.dependencies]
 python = ">=3.10,<3.13"
+nibabel = "^5"
+scipy = "^1.12"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.3.3"

src/clinicaio/utils/bids_entities.py

@@ -0,0 +1,104 @@
+from typing import Any
+from .caps import Space, Description, Resolution
+from .pet import Tracer, SUVRReferenceRegion
+
+
+class Entity:
+    """Base class for BIDS entities."""
+
+    key: str
+
+    def __new__(cls, value: Any) -> str:
+        """Returns key-value pair."""
+        return cls.key + "-" + cls._process_value(value)
+
+    @classmethod
+    def _process_value(cls, value: Any) -> str:
+        """Processes the value."""
+        return str(value)
+
+
+class SubjectEntity(Entity):
+    """Class to model the Subject entity ('sub-...')."""
+
+    key = "sub"
+
+    @classmethod
+    def _process_value(cls, value: int) -> str:
+        if not isinstance(value, int):
+            try:
+                value = int(value)
+            except TypeError as exc:
+                raise ValueError(f"Subject ID should be an int. Got {value}") from exc
+        if value < 1 or value > 999:
+            raise ValueError(f"Subject ID should be between 1 and 999. Got {value}")
+
+        return str(value).zfill(3)
+
+
+class SessionEntity(Entity):
+    """Class to model the Session entity ('ses-...')."""
+
+    key = "ses"
+
+    @classmethod
+    def _process_value(cls, value: int) -> str:
+        if not isinstance(value, int):
+            try:
+                value = int(value)
+            except TypeError as exc:
+                raise ValueError(f"Session ID should be an int. Got {value}") from exc
+        if value < 0 or value > 999:
+            raise ValueError(f"Session ID should be between 0 and 999. Got {value}")
+
+        return "M" + str(value).zfill(3)
+
+
+class SpaceEntity(Entity):
+    """Class to model the Space entity ('space-...')."""
+
+    key = "space"
+
+    @classmethod
+    def _process_value(cls, value: str) -> str:
+        return Space(value).value
+
+
+class DescriptionEntity(Entity):
+    """Class to model the Description entity ('desc-...')."""
+
+    key = "desc"
+
+    @classmethod
+    def _process_value(cls, value: str) -> str:
+        return Description(value).value
+
+
+class ResolutionEntity(Entity):
+    """Class to model the Resolution entity ('res-...')."""
+
+    key = "res"
+
+    @classmethod
+    def _process_value(cls, value: str) -> str:
+        return Resolution(value).value
+
+
+class TracerEntity(Entity):
+    """Class to model the Tracer entity ('trc-...')."""
+
+    key = "trc"
+
+    @classmethod
+    def _process_value(cls, value: str) -> str:
+        return Tracer(value).value
+
+
+class SUVREntity(Entity):
+    """Class to model the SUVR Reference Region entity ('suvr-...')."""
+
+    key = "suvr"
+
+    @classmethod
+    def _process_value(cls, value: str) -> str:
+        return SUVRReferenceRegion(value).value

src/clinicaio/utils/caps.py

@@ -0,0 +1,52 @@
+from enum import Enum
+
+
+class Pipeline(str, Enum):
+    """Preprocessing pipelines that can be found in CAPS."""
+
+    T1_LINEAR = "t1-linear"
+    FLAIR_LINEAR = "flair-linear"
+    PET_LINEAR = "pet-linear"
+
+
+class Extension(str, Enum):
+    """Possible extensions in BIDS file names."""
+
+    NIIGZ = ".nii.gz"
+    NII = ".nii"
+    JSON = ".json"
+    TSV = ".tsv"
+    MAT = ".mat"
+    BVAL = ".bval"
+    BVEC = ".bvec"
+
+
+class Suffix(str, Enum):
+    """Possible suffixes in CAPS file names."""
+
+    DWI = "dwi"
+    PET = "pet"
+    T1W = "T1w"
+    FLAIR = "FLAIR"
+    AFFINE = "affine"
+    RIGID = "rigid"
+
+
+class Space(str, Enum):
+    """Possible registration spaces."""
+
+    MNI = "MNI152NLin2009cSym"
+    IXI = "Ixi549Space"
+    T1W = "T1w"
+
+
+class Resolution(str, Enum):
+    """Resolutions that can be found in CAPS."""
+
+    ONE = "1x1x1"
+
+
+class Description(str, Enum):
+    """BIDS Description values that can be found in CAPS."""
+
+    CROP = "Crop"

src/clinicaio/utils/caps_generator/filename.py

@@ -0,0 +1,42 @@
+from pathlib import Path
+from typing import Optional
+
+from clinicaio.utils.bids_entities import (
+    DescriptionEntity,
+    ResolutionEntity,
+    SessionEntity,
+    SpaceEntity,
+    SubjectEntity,
+    SUVREntity,
+    TracerEntity,
+)
+from clinicaio.utils.caps import Description, Extension, Suffix
+
+
+def get_caps_filename(
+    subject: int,
+    session: int,
+    suffix: str,
+    extension: str,
+    tracer: Optional[str] = None,
+    space: Optional[str] = None,
+    crop: bool = False,
+    resolution: Optional[str] = None,
+    suvr_ref_region: Optional[str] = None,
+) -> Path:
+    """Returns a BIDS-compliant filename from entity values, a suffix and a extension."""
+    entities_list = []
+    for entity in [  # order matters
+        SubjectEntity(subject),
+        SessionEntity(session),
+        TracerEntity(tracer) if tracer else None,
+        SpaceEntity(space) if space else None,
+        DescriptionEntity(Description.CROP) if crop else None,
+        ResolutionEntity(resolution) if resolution else None,
+        SUVREntity(suvr_ref_region) if suvr_ref_region else None,
+        Suffix(suffix).value,
+    ]:
+        if entity is not None:
+            entities_list.append(entity)
+
+    return Path("_".join(entities_list)).with_suffix(Extension(extension).value)

src/clinicaio/utils/caps_generator/flair_linear.py

@@ -0,0 +1,67 @@
+from pathlib import Path
+from typing import Union
+
+import nibabel as nib
+import numpy as np
+from scipy.io import savemat
+
+from clinicaio.utils.bids_entities import SessionEntity, SubjectEntity
+from clinicaio.utils.caps import Extension, Resolution, Space, Suffix
+
+from .filename import get_caps_filename
+
+
+def build_flair_linear(
+    root: Union[str, Path], subject: int, session: int, crop: bool = True
+):
+    """
+    Simulates flair-linear by creating fake output files in `root`.
+    """
+    dummy_nifti_img = nib.Nifti1Image(np.ones((1, 1, 1)).astype(np.int8), np.eye(4))
+    dummy_mat = {
+        "AffineTransform_double_3_3": np.ones((1, 1)).astype(np.int8),
+        "fixed": np.ones((1, 1)).astype(np.int8),
+    }
+
+    space = Space.MNI
+    resolution = Resolution.ONE
+    directory = (
+        Path(root)
+        / "subjects"
+        / SubjectEntity(subject)
+        / SessionEntity(session)
+        / "flair_linear"
+    )
+    directory.mkdir(parents=True, exist_ok=True)
+
+    uncropped_file = directory / get_caps_filename(
+        subject,
+        session,
+        space=space,
+        resolution=resolution,
+        suffix=Suffix.FLAIR,
+        extension=Extension.NIIGZ,
+    )
+    nib.save(dummy_nifti_img, uncropped_file)
+
+    mat_file = directory / get_caps_filename(
+        subject,
+        session,
+        space=space,
+        resolution=resolution,
+        suffix=Suffix.AFFINE,
+        extension=Extension.MAT,
+    )
+    savemat(mat_file, dummy_mat)
+
+    if crop:
+        cropped_file = directory / get_caps_filename(
+            subject,
+            session,
+            space=space,
+            crop=True,
+            resolution=resolution,
+            suffix=Suffix.FLAIR,
+            extension=Extension.NIIGZ,
+        )
+        nib.save(dummy_nifti_img, cropped_file)

src/clinicaio/utils/caps_generator/generator.py

@@ -0,0 +1,109 @@
+import shutil
+from pathlib import Path
+from typing import List, Union
+
+from clinicaio.utils.bids_entities import SessionEntity, SubjectEntity
+from clinicaio.utils.caps import Pipeline
+
+from .flair_linear import build_flair_linear
+from .pet_linear import build_pet_linear
+from .t1_linear import build_t1_linear
+
+
+class CAPSGenerator:
+    """
+    To build fake CAPS. A CAPSGenerator simulates preprocessing pipelines
+    by creating fake output files.
+
+    Parameters
+    ----------
+    directory : Union[str, Path]
+        the directory of the CAPS.
+    """
+
+    def __init__(self, directory: Union[str, Path]) -> None:
+        self.dir = Path(directory)
+
+    def build_pipeline(
+        self,
+        pipeline: Union[str, Pipeline],
+        subjects: List[int],
+        sessions: List[int],
+        **kwargs,
+    ) -> None:
+        """
+        Simulates a preprocessing pipeline for every subject in `subjects` and every session
+        in `sessions`.
+
+        Parameters
+        ----------
+        pipeline : Union[str, Pipeline]
+            the pipeline to simulate. Supported pipelines are `t1-linear`,
+            `flair-linear` and `pet-linear`.
+        subjects : List[int]
+            the list of subject IDs
+        sessions : List[int]
+            the list of session IDs.
+        **kwargs
+            any argument accepted by the pipeline:
+            - `t1_linear`: `crop`;
+            - `flair_linear`: `crop`;
+            - `pet-linear`: `tracer`, `suvr_ref_region`, `crop` and `save_pet_in_t1w_space`.
+
+        Raises
+        ------
+        ValueError
+            if  `pipeline` is not in `t1-linear`, `flair-linear` and `pet-linear`.
+        """
+        pipeline = Pipeline(pipeline)
+        if pipeline == Pipeline.T1_LINEAR:
+            builder = build_t1_linear
+        elif pipeline == Pipeline.FLAIR_LINEAR:
+            builder = build_flair_linear
+        elif pipeline == Pipeline.PET_LINEAR:
+            builder = build_pet_linear
+        else:
+            raise ValueError(f"pipeline {pipeline} is not yet implemented.")
+
+        for subject in subjects:
+            for session in sessions:
+                builder(self.dir, subject, session, **kwargs)
+
+    def remove_pipeline(
+        self, pipeline: Union[str, Pipeline], subject: int, session: int
+    ) -> None:
+        """
+        Removes a preprocessing pipeline for a specific (subject, session).
+
+        Parameters
+        ----------
+        pipeline : Union[str, Pipeline]
+            the pipeline to remove. Supported pipelines are `t1-linear`,
+            `flair-linear` and `pet-linear`.
+        subject : int
+            the subject ID.
+        session : int
+            the session ID.
+
+        Raises
+        ------
+        ValueError
+            if  `pipeline` is not in `t1-linear`, `flair-linear` and `pet-linear`.
+        """
+        pipeline = Pipeline(pipeline)
+        if pipeline == Pipeline.T1_LINEAR:
+            directory = "t1_linear"
+        elif pipeline == Pipeline.FLAIR_LINEAR:
+            directory = "flair_linear"
+        elif pipeline == Pipeline.PET_LINEAR:
+            directory = "pet_linear"
+        else:
+            raise ValueError(f"pipeline {pipeline} is not yet implemented.")
+
+        full_dir = (
+            self.dir / "subjects" / SubjectEntity(subject) / SessionEntity(session) / directory
+        )
+        try:
+            shutil.rmtree(full_dir)
+        except FileNotFoundError:   # there is not such subject/session for this pipeline
+            pass