diff --git a/python/lib/dcm2bids_imaging_pipeline_lib/nifti_insertion_pipeline.py b/python/lib/dcm2bids_imaging_pipeline_lib/nifti_insertion_pipeline.py index c9a4d75a4..e39bdb951 100644 --- a/python/lib/dcm2bids_imaging_pipeline_lib/nifti_insertion_pipeline.py +++ b/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): """ diff --git a/python/loris_bids_importer/src/loris_bids_importer/copy_files.py b/python/loris_bids_importer/src/loris_bids_importer/copy_files.py index 649741eda..43142e777 100644 --- a/python/loris_bids_importer/src/loris_bids_importer/copy_files.py +++ b/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, ) diff --git a/python/loris_bids_utils/src/loris_bids_utils/path.py b/python/loris_bids_utils/src/loris_bids_utils/path.py new file mode 100644 index 000000000..aa00419ac --- /dev/null +++ b/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, + ])) diff --git a/python/loris_utils/src/loris_utils/iter.py b/python/loris_utils/src/loris_utils/iter.py index e67414b60..d5d8535c8 100644 --- a/python/loris_utils/src/loris_utils/iter.py +++ b/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 diff --git a/python/loris_utils/src/loris_utils/path.py b/python/loris_utils/src/loris_utils/path.py index 27a48eafc..7f95fb54c 100644 --- a/python/loris_utils/src/loris_utils/path.py +++ b/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] diff --git a/python/loris_utils/src/loris_utils/sort.py b/python/loris_utils/src/loris_utils/sort.py new file mode 100644 index 000000000..9596e96ce --- /dev/null +++ b/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)))