[ENH] Implement dataset description for CAPS datasets

clinica/iotools/bids_dataset_description.py

@@ -1,12 +1,21 @@
-from enum import Enum
+import json
+from pathlib import Path
 from typing import IO
 
 from attrs import define, fields
 from cattr.gen import make_dict_unstructure_fn, override
 from cattr.preconf.json import make_converter
+from packaging.version import InvalidVersion, Version
 
 from clinica.utils.bids import BIDS_VERSION
+from clinica.utils.exceptions import ClinicaBIDSError
 from clinica.utils.inputs import DatasetType
+from clinica.utils.stream import log_and_raise
+
+__all__ = [
+    "BIDSDatasetDescription",
+    "get_bids_version",
+]
 
 
 @define
@@ -17,7 +26,7 @@ class BIDSDatasetDescription:
     """
 
     name: str
-    bids_version: str = BIDS_VERSION
+    bids_version: Version = BIDS_VERSION
     dataset_type: DatasetType = DatasetType.RAW
 
     def write(self, to: IO[str]):
@@ -39,7 +48,7 @@ def _rename(name: str) -> str:
 
 # Register a JSON converter for the BIDS dataset description model.
 converter = make_converter()
-
+converter.register_unstructure_hook(Version, lambda dt: str(dt))
 converter.register_unstructure_hook(
     BIDSDatasetDescription,
     make_dict_unstructure_fn(
@@ -51,3 +60,40 @@ def _rename(name: str) -> str:
         },
     ),
 )
+
+
+def get_bids_version(dataset_folder: Path) -> Version:
+    """Returns the BIDS version number of a BIDS or CAPS dataset."""
+    try:
+        with open(dataset_folder / "dataset_description.json", "r") as fp:
+            bids_metadata = json.load(fp)
+        return Version(bids_metadata["BIDSVersion"])
+    except InvalidVersion as e:
+        log_and_raise(
+            (
+                f"File {dataset_folder / 'dataset_description.json'} has a "
+                f"BIDS version number not properly formatted:\n{e}"
+            ),
+            ClinicaBIDSError,
+        )
+    except FileNotFoundError:
+        log_and_raise(
+            (
+                f"File {dataset_folder / 'dataset_description.json'} is missing "
+                "while it is mandatory for a BIDS/CAPS dataset."
+            ),
+            ClinicaBIDSError,
+        )
+    except KeyError:
+        log_and_raise(
+            (
+                f"File {dataset_folder / 'dataset_description.json'} is missing a "
+                "'BIDSVersion' key while it is mandatory."
+            ),
+            ClinicaBIDSError,
+        )
+    except json.JSONDecodeError as e:
+        log_and_raise(
+            f"File {dataset_folder / 'dataset_description.json'} is not formatted correctly:\n{e}.",
+            ClinicaBIDSError,
+        )

clinica/pipelines/engine.py

@@ -423,6 +423,11 @@ def __init__(
         from pathlib import Path
         from tempfile import mkdtemp
 
+        from clinica.utils.caps import (
+            build_caps_dataset_description,
+            write_caps_dataset_description,
+        )
+        from clinica.utils.exceptions import ClinicaCAPSError
         from clinica.utils.inputs import check_bids_folder, check_caps_folder
 
         self._is_built: bool = False
@@ -461,11 +466,37 @@ def __init__(
                     f"The {self._name} pipeline does not contain "
                     "BIDS nor CAPS directory at the initialization."
                 )
-            check_caps_folder(self._caps_directory)
+            try:
+                check_caps_folder(self._caps_directory)
+            except ClinicaCAPSError as e:
+                desc = build_caps_dataset_description(
+                    self._caps_directory,
+                    self._caps_directory,
+                    self._name,
+                    f"subjects/*/*/{self._name.replace('-', '_')}",
+                )
+                raise ClinicaCAPSError(
+                    f"{e}\nYou might want to create a 'dataset_description.json' "
+                    f"file with the following content:\n{desc}"
+                )
             self.is_bids_dir = False
         else:
             check_bids_folder(self._bids_directory)
             self.is_bids_dir = True
+            if self._caps_directory is not None:
+                if (
+                    not self._caps_directory.exists()
+                    or len([f for f in self._caps_directory.iterdir()]) == 0
+                ):
+                    self._caps_directory.mkdir(parents=True, exist_ok=True)
+        if self._caps_directory:
+            write_caps_dataset_description(
+                self.input_dir,
+                self._caps_directory,
+                self._name,
+                f"subjects/*/*/{self._name.replace('-', '_')}",
+            )
+            check_caps_folder(self._caps_directory)
         self._compute_subjects_and_sessions()
         self._init_nodes()
 

clinica/utils/bids.py

@@ -4,6 +4,8 @@
 from pathlib import Path
 from typing import Dict, Tuple, Union
 
+from packaging.version import Version
+
 __all__ = [
     "BIDSLabel",
     "BIDSFileName",
@@ -12,7 +14,7 @@
     "Suffix",
 ]
 
-BIDS_VERSION = "1.7.0"
+BIDS_VERSION = Version("1.7.0")
 
 
 class Extension(str, Enum):