Factorize BIDS path utils

python/lib/dcm2bids_imaging_pipeline_lib/nifti_insertion_pipeline.py

@@ -9,7 +9,9 @@
 from loris_bids_importer.file_type import get_check_bids_imaging_file_type_from_extension
 from loris_bids_importer.mri.sidecar import add_bids_mri_sidecar_file_parameters, get_bids_mri_sidecar_session_info
 from loris_bids_utils.mri.sidecar import BidsMriSidecarJsonFile
+from loris_bids_utils.path import build_bids_file_name, build_bids_modality_path
 from loris_utils.crypto import compute_file_blake2b_hash, compute_file_md5_hash
+from loris_utils.path import get_path_extension
 
 import lib.exitcode
 from lib.db.queries.dicom_archive import try_get_dicom_archive_series_with_series_uid_echo_time
@@ -46,7 +48,7 @@ def __init__(self, loris_getopt_obj, script_name):
          :type script_name: str
         """
         super().__init__(loris_getopt_obj, script_name)
-        self.nifti_path = self.options_dict["nifti_path"]["value"]
+        self.nifti_path = Path(self.options_dict["nifti_path"]["value"])
         self.nifti_s3_url = self.options_dict["nifti_path"]["s3_url"] \
             if 's3_url' in self.options_dict["nifti_path"].keys() else None
         self.nifti_blake2 = compute_file_blake2b_hash(self.nifti_path)
@@ -243,7 +245,7 @@ def init_session_info(self):
                 self.dicom_archive.patient_name,
                 self.dicom_archive.id,
                 self.json_file_dict,
-                self.nifti_path,
+                str(self.nifti_path),
                 str(error),
             )
 
@@ -293,7 +295,7 @@ def _validate_nifti_patient_name_with_dicom_patient_name(self):
                 nifti_pname,
                 self.dicom_archive.id,
                 self.json_file_dict,
-                self.nifti_path,
+                str(self.nifti_path),
                 err_msg
             )
 
@@ -365,7 +367,6 @@ def _determine_acquisition_protocol(self):
          :rtype: int
         """
 
-        nifti_name = os.path.basename(self.nifti_path)
         scan_param = self.json_file_dict
 
         # get the list of lines in the mri_protocol table that apply to the given scan based on the
@@ -379,7 +380,7 @@ def _determine_acquisition_protocol(self):
         )
 
         protocol_info = self.imaging_obj.get_acquisition_protocol_info(
-            protocols_list, nifti_name, scan_param, self.loris_scan_type
+            protocols_list, self.nifti_path.name, scan_param, self.loris_scan_type
         )
 
         log_verbose(self.env, protocol_info['error_message'])
@@ -448,8 +449,6 @@ def _determine_new_nifti_assembly_rel_path(self):
                 file_bids_entities_dict[key] = value
 
         # determine where the file should go
-        bids_cand_id = 'sub-' + str(self.session.candidate.cand_id)
-        bids_visit = 'ses-' + self.session.visit_label
         bids_subfolder = self.bids_categories_dict['BIDSCategoryName']
 
         # determine NIfTI file name
@@ -459,7 +458,14 @@ def _determine_new_nifti_assembly_rel_path(self):
             file_bids_entities_dict['run'] += 1
             new_nifti_name = self._construct_nifti_filename(file_bids_entities_dict)
 
-        return os.path.join('assembly_bids', bids_cand_id, bids_visit, bids_subfolder, new_nifti_name)
+        relative_path = build_bids_modality_path(
+            str(self.session.candidate.cand_id),
+            self.session.visit_label,
+            bids_subfolder,
+            new_nifti_name,
+        )
+
+        return os.path.join('assembly_bids', relative_path)
 
     def _construct_nifti_filename(self, file_bids_entities_dict):
         """
@@ -474,44 +480,19 @@ def _construct_nifti_filename(self, file_bids_entities_dict):
          :rtype: str
         """
 
-        # this list defined the order in which BIDS entities should appear in the filename
-        bids_entity_order = (
-            'sub',       # Subject
-            'ses',       # Session
-            'task',      # Task
-            'acq',       # Acquisition
-            'ce',        # Contrast Enhancing Agent
-            'rec',       # Reconstruction
-            'dir',       # Phase Encoding Direction
-            'run',       # Run
-            'mod',       # Corresponding Modality
-            'echo',      # Echo
-            'flip',      # Flip Angle
-            'inv',       # Inversion Time
-            'mt',        # Magnetization Transfer
-            'part',      # Part
-            'recording'  # Recording
-        )
-
-        nifti_filename = ''
-        for entity in bids_entity_order:
-            if entity == 'sub':
-                nifti_filename += f"{entity}-{file_bids_entities_dict[entity]}"
-            elif entity == "echo" and self.bids_categories_dict['BIDSScanType'] == 'magnitude':
-                self.bids_categories_dict['BIDSScanType'] = f"magnitude{file_bids_entities_dict[entity]}"
-            else:
-                if entity in file_bids_entities_dict.keys():
-                    nifti_filename += f"_{entity}-{file_bids_entities_dict[entity]}"
-
-        # add BIDS scan type to the NIfTI filename
-        nifti_filename += f"_{self.bids_categories_dict['BIDSScanType']}"
+        suffix = self.bids_categories_dict['BIDSScanType']
+        entities = dict(file_bids_entities_dict)
+        if suffix == 'magnitude' and 'echo' in entities:
+            suffix = f"magnitude{entities.pop('echo')}"
 
-        # determine NIfTI file extension and append it to filename
-        curr_nifti_path = self.nifti_path
-        nifti_ext = re.search(r"\.nii(\.gz)?$", curr_nifti_path).group()
-        nifti_filename += nifti_ext
+        nifti_ext = get_path_extension(self.nifti_path)
+        if nifti_ext is None:
+            log_error_exit(
+                self.env,
+                f"Missing NIfTI extension in file name: {self.nifti_path}"
+            )
 
-        return nifti_filename
+        return build_bids_file_name(entities, suffix, nifti_ext)
 
     def _move_to_trashbin(self):
         """

python/loris_bids_importer/src/loris_bids_importer/copy_files.py

@@ -8,6 +8,7 @@
 from loris_bids_utils.files.dataset_description import BidsDatasetDescriptionJsonFile
 from loris_bids_utils.files.participants import BidsParticipantsTsvFile
 from loris_bids_utils.files.scans import BidsScansTsvFile, BidsScanTsvRow
+from loris_bids_utils.path import build_bids_modality_path, build_bids_session_path
 
 from loris_bids_importer.env import BidsImportEnv
 
@@ -68,12 +69,11 @@ def get_loris_bids_file_path(
     # information.
     loris_file_name = get_loris_bids_file_name(file_path.name, session)
 
-    return (
-        import_env.loris_bids_path
-        / f'sub-{session.candidate.psc_id}'
-        / f'ses-{session.visit_label}'
-        / data_type
-        / loris_file_name
+    return import_env.loris_bids_path / build_bids_modality_path(
+        session.candidate.psc_id,
+        session.visit_label,
+        data_type,
+        loris_file_name,
     )
 
 
@@ -101,11 +101,11 @@ def get_loris_scans_path(import_env: BidsImportEnv, scans_file: BidsScansTsvFile
         return scans_file.path.relative_to(import_env.data_dir_path)
 
     loris_file_name = get_loris_bids_file_name(scans_file.path.name, session)
-    return (
-        import_env.loris_bids_path
-        / f'sub-{session.candidate.psc_id}'
-        / f'ses-{session.visit_label}'
-        / loris_file_name
+
+    return import_env.loris_bids_path / build_bids_session_path(
+        session.candidate.psc_id,
+        session.visit_label,
+        loris_file_name,
     )
 
 

python/loris_bids_utils/src/loris_bids_utils/path.py

@@ -0,0 +1,109 @@
+"""Small, dependency-free helpers for BIDS names and relative paths."""
+
+from pathlib import Path
+
+from loris_utils.iter import filter_non_none
+from loris_utils.path import remove_name_extension
+from loris_utils.sort import sort_dict_by_key_order
+
+# The standard order in which BIDS entities should appear in a file name.
+# https://bids-specification.readthedocs.io/en/stable/appendices/entity-table.html
+BIDS_ENTITY_ORDER = [
+    'sub',        # Subject
+    'ses',        # Session
+    'task',       # Task
+    'acq',        # Acquisition
+    'ce',         # Contrast Enhancing Agent
+    'rec',        # Reconstruction
+    'dir',        # Phase Encoding Direction
+    'run',        # Run
+    'mod',        # Corresponding Modality
+    'echo',       # Echo
+    'flip',       # Flip Angle
+    'inv',        # Inversion Time
+    'mt',         # Magnetization Transfer
+    'part',       # Part
+    'space',      # Coordinate System Space
+    'split',      # Split
+    'recording',  # Recording
+]
+
+
+def parse_bids_entities(name: str) -> dict[str, str]:
+    """
+    Parse the entities from a BIDS file name. This function assumes the provided BIDS file name is
+    formatted correctly.
+    """
+
+    stem = remove_name_extension(name)
+
+    parts = stem.split('_')
+
+    entities: dict[str, str] = {}
+
+    # Ignore the BIDS suffix.
+    for pair in parts[:-1]:
+        key, value = pair.split('-', 1)
+        entities[key] = value
+
+    return entities
+
+
+def build_bids_file_name(entities: dict[str, str], suffix: str, extension: str) -> str:
+    """
+    Build a BIDS file name from its entities, suffix, and extension.
+    """
+
+    # Sort the BIDS entities according to the BIDS entity order.
+    ordered_entities = sort_dict_by_key_order(entities, BIDS_ENTITY_ORDER)
+
+    # Build the BIDS key label pairs from the BIDS entities.
+    prefix = '_'.join(f'{key}-{label}' for key, label in ordered_entities.items())
+
+    # Build the final BIDS file name.
+    return f'{prefix}_{suffix}.{extension}'
+
+
+def build_bids_subject_path(
+    subject: str,
+    file_name: str,
+) -> Path:
+    """
+    Build a BIDS path for a file at the subject level.
+    """
+
+    return Path(f'sub-{subject}', file_name)
+
+
+def build_bids_session_path(
+    subject: str,
+    session: str | None,
+    file_name: str,
+) -> Path:
+    """
+    Build a BIDS path for a file at the session level.
+    """
+
+    return Path(*filter_non_none([
+        f'sub-{subject}',
+        f'ses-{session}' if session is not None else None,
+        file_name,
+    ]))
+
+
+def build_bids_modality_path(
+    subject: str,
+    session: str | None,
+    data_type: str,
+    file_name: str,
+) -> Path:
+    """
+    Build a BIDS path for a file at the modality level.
+    """
+
+    return Path(*filter_non_none([
+        f'sub-{subject}',
+        f'ses-{session}' if session is not None else None,
+        data_type,
+        file_name,
+    ]))

python/loris_utils/src/loris_utils/iter.py

@@ -85,3 +85,13 @@ def map_non_none(value: T | None, function: Callable[[T], U]) -> U | None:
     """
 
     return function(value) if value is not None else value
+
+
+def filter_non_none(iterable: Iterable[T | None]) -> Iterator[T]:
+    """
+    Filter the `None` elements out of an iterator.
+    """
+
+    for element in iterable:
+        if element is not None:
+            yield element

python/loris_utils/src/loris_utils/path.py

@@ -36,8 +36,7 @@ def remove_path_extension(path: Path) -> Path:
     Remove the extension (including multiple extensions) of a path.
     """
 
-    parts = path.name.split('.')
-    return path.with_name(parts[0])
+    return path.with_name(remove_name_extension(path.name))
 
 
 def replace_path_extension(path: Path, extension: str) -> Path:
@@ -47,3 +46,11 @@ def replace_path_extension(path: Path, extension: str) -> Path:
 
     parts = path.name.split('.')
     return path.with_name(f'{parts[0]}.{extension}')
+
+
+def remove_name_extension(name: str) -> str:
+    """
+    Remove the extension (including multiple extensions) of a file name.
+    """
+
+    return name.split('.')[0]

python/loris_utils/src/loris_utils/sort.py

@@ -0,0 +1,31 @@
+from collections.abc import Callable
+from typing import Any, TypeVar
+
+K = TypeVar('K')
+V = TypeVar('V')
+
+
+def sort_dict(
+    dictionary: dict[K, V],
+    key: Callable[[K], Any] | None = None,
+    reverse: bool = False,
+) -> dict[K, V]:
+    """
+    Sort a dictionary by keys and return a new sorted dictionary.
+    """
+
+    # FIXME: Typing the following line requires a full type hint for `key`.
+    return {k: dictionary[k] for k in sorted(dictionary, key=key, reverse=reverse)}  # type: ignore
+
+
+def sort_dict_by_key_order(dictionary: dict[K, V], key_order: list[K]) -> dict[K, V]:
+    """
+    Sort a dictionary based on a custom key order. Keys not in the key order go at the end in their
+    original order.
+    """
+
+    order_indexes = {k: i for i, k in enumerate(key_order)}
+    default_index = len(key_order)
+    dict_keys = list(dictionary.keys())
+
+    return sort_dict(dictionary, key=lambda k: (order_indexes.get(k, default_index), dict_keys.index(k)))