migrate bids importer from str to path

Author: MaximeBICMTL

python/lib/imaging_lib/bids/dataset.py

@@ -1,14 +1,14 @@
-import os
 import re
 from collections.abc import Iterator
 from functools import cached_property
+from pathlib import Path
 
 from bids import BIDSLayout
 
 from lib.imaging_lib.bids.dataset_description import BidsDatasetDescription
 from lib.imaging_lib.bids.tsv_participants import BidsTsvParticipant, read_bids_participants_tsv_file
 from lib.imaging_lib.bids.tsv_scans import BidsTsvScan, read_bids_scans_tsv_file
-from lib.imaging_lib.nifti import find_dir_nifti_names
+from lib.imaging_lib.nifti import find_dir_nifti_files
 from lib.util.fs import replace_file_extension, search_dir_file_with_regex
 from lib.util.iter import find
 
@@ -18,10 +18,10 @@
 
 
 class BIDSDataset:
-    path: str
+    path: Path
     validate: bool
 
-    def __init__(self, bids_path: str, validate: bool):
+    def __init__(self, bids_path: Path, validate: bool):
         self.path     = bids_path
         self.validate = validate
 
@@ -48,12 +48,12 @@ def subjects(self) -> list['BIDSSubject']:
 
         subjects: list[BIDSSubject] = []
 
-        for file in os.scandir(self.path):
+        for file in self.path.iterdir():
             subject_match = re.match(r'sub-([a-zA-Z0-9]+)', file.name)
             if subject_match is None:
                 continue
 
-            if not os.path.isdir(file):
+            if not file.is_dir():
                 continue
 
             subject_label = subject_match.group(1)
@@ -68,8 +68,8 @@ def get_dataset_description(self) -> 'BidsDatasetDescription | None':
         does contains incorrect data.
         """
 
-        dataset_description_path = os.path.join(self.path, 'dataset_description.json')
-        if not os.path.exists(dataset_description_path):
+        dataset_description_path = self.path / 'dataset_description.json'
+        if not dataset_description_path.exists():
             return None
 
         return BidsDatasetDescription(dataset_description_path)
@@ -81,8 +81,8 @@ def tsv_participants(self) -> dict[str, BidsTsvParticipant] | None:
         present. This property might raise an exception if the file is present but incorrect.
         """
 
-        tsv_participants_path = os.path.join(self.path, 'participants.tsv')
-        if not os.path.exists(tsv_participants_path):
+        tsv_participants_path = self.path / 'participants.tsv'
+        if not tsv_participants_path.exists():
             return None
 
         return read_bids_participants_tsv_file(tsv_participants_path)
@@ -143,13 +143,13 @@ def layout(self) -> BIDSLayout:
 
 class BIDSSubject:
     root_dataset: BIDSDataset
+    path: Path
     label: str
-    path: str
 
     def __init__(self, root_dataset: BIDSDataset, label: str):
         self.root_dataset = root_dataset
         self.label = label
-        self.path = os.path.join(self.root_dataset.path, f'sub-{self.label}')
+        self.path  = self.root_dataset.path / f'sub-{self.label}'
 
     @property
     def data_types(self) -> Iterator['BIDSDataType']:
@@ -169,8 +169,8 @@ def sessions(self) -> list['BIDSSession']:
 
         sessions: list[BIDSSession] = []
 
-        for file in os.scandir(self.path):
-            if not os.path.isdir(file):
+        for file in self.path.iterdir():
+            if not file.is_dir():
                 continue
 
             session_match = re.match(r'ses-([a-zA-Z0-9]+)', file.name)
@@ -195,23 +195,19 @@ def get_session(self, session_label: str) -> 'BIDSSession | None':
 
 class BIDSSession:
     subject: BIDSSubject
+    path: Path
     label: str | None
-    path: str
-    tsv_scans_path: str | None
+    tsv_scans_path: Path | None
 
     def __init__(self, subject: BIDSSubject, label: str | None):
         self.subject = subject
         self.label = label
-        if label is None:
-            self.path = self.subject.path
+        if label is not None:
+            self.path = subject.path / f'ses-{self.label}'
         else:
-            self.path = os.path.join(self.subject.path, f'ses-{self.label}')
+            self.path = subject.path
 
-        tsv_scans_name = search_dir_file_with_regex(self.path, r'scans.tsv$')
-        if tsv_scans_name is not None:
-            self.tsv_scans_path = os.path.join(self.path, tsv_scans_name)
-        else:
-            self.tsv_scans_path = None
+        self.tsv_scans_path = search_dir_file_with_regex(self.path, r'scans.tsv$')
 
     @property
     def root_dataset(self) -> BIDSDataset:
@@ -230,8 +226,8 @@ def data_types(self) -> list['BIDSDataType']:
 
         data_types: list[BIDSDataType] = []
 
-        for file in os.scandir(self.path):
-            if not os.path.isdir(file):
+        for file in self.path.iterdir():
+            if not file.is_dir():
                 continue
 
             data_types.append(BIDSDataType(self, file.name))
@@ -264,13 +260,15 @@ def get_tsv_scan(self, file_name: str) -> 'BidsTsvScan | None':
 
 class BIDSDataType:
     session: BIDSSession
-    name: str
-    path: str
+    path: Path
 
     def __init__(self, session: BIDSSession, name: str):
         self.session = session
-        self.name = name
-        self.path = os.path.join(self.session.path, self.name)
+        self.path    = session.path / name
+
+    @property
+    def name(self) -> str:
+        return self.path.name
 
     @property
     def root_dataset(self) -> BIDSDataset:
@@ -288,29 +286,31 @@ def niftis(self) -> list['BIDSNifti']:
 
         niftis: list[BIDSNifti] = []
 
-        for nifti_name in find_dir_nifti_names(self.path):
-            niftis.append(BIDSNifti(self, nifti_name))
+        for nifti_path in find_dir_nifti_files(self.path):
+            niftis.append(BIDSNifti(self, nifti_path.name))
 
         return niftis
 
 
 class BIDSNifti:
     data_type: BIDSDataType
-    name: str
-    path: str
+    path: Path
     suffix: str | None
 
     def __init__(self, data_type: BIDSDataType, name: str):
         self.data_type = data_type
-        self.path = os.path.join(self.data_type.path, name)
-        self.name = name
+        self.path      = data_type.path / name
 
         suffix_match = re.search(r'_([a-zA-Z0-9]+)\.nii(\.gz)?$', self.name)
         if suffix_match is not None:
             self.suffix = suffix_match.group(1)
         else:
             self.suffix = None
 
+    @property
+    def name(self) -> str:
+        return self.path.name
+
     @property
     def root_dataset(self) -> BIDSDataset:
         return self.data_type.root_dataset
@@ -323,38 +323,38 @@ def subject(self) -> BIDSSubject:
     def session(self) -> BIDSSession:
         return self.data_type.session
 
-    def get_json_path(self) -> str | None:
+    def get_json_path(self) -> Path | None:
         """
         Get the JSON sidecar file path of this NIfTI file if it exists.
         """
 
         json_name = replace_file_extension(self.name, 'json')
-        json_path = os.path.join(self.data_type.path, json_name)
-        if not os.path.exists(json_path):
+        json_path = self.data_type.path / json_name
+        if not json_path.exists():
             return None
 
         return json_path
 
-    def get_bval_path(self) -> str | None:
+    def get_bval_path(self) -> Path | None:
         """
         Get the BVAL file path of this NIfTI file if it exists.
         """
 
         bval_name = replace_file_extension(self.name, 'bval')
-        bval_path = os.path.join(self.data_type.path, bval_name)
-        if not os.path.exists(bval_path):
+        bval_path = self.data_type.path / bval_name
+        if not bval_path.exists():
             return None
 
         return bval_path
 
-    def get_bvec_path(self) -> str | None:
+    def get_bvec_path(self) -> Path | None:
         """
         Get the BVEC file path of this NIfTI file if it exists.
         """
 
         bvec_name = replace_file_extension(self.name, 'bvec')
-        bvec_path = os.path.join(self.data_type.path, bvec_name)
-        if not os.path.exists(bvec_path):
+        bvec_path = self.data_type.path / bvec_name
+        if not bvec_path.exists():
             return None
 
         return bvec_path

python/lib/imaging_lib/bids/dataset_description.py

@@ -1,4 +1,5 @@
 import json
+from pathlib import Path
 from typing import Any
 
 
@@ -31,7 +32,7 @@ class BidsDatasetDescription:
     The BIDS dataset description JSON data.
     """
 
-    def __init__(self, dataset_descrption_path: str):
+    def __init__(self, dataset_descrption_path: Path):
         """
         Read a BIDS dataset description file, or raise an exception if that file contains incorrect
         data.

python/lib/imaging_lib/bids/json.py

@@ -1,4 +1,5 @@
 import json
+from pathlib import Path
 from typing import Any
 
 from lib.config import get_patient_id_dicom_header_config
@@ -41,7 +42,7 @@ def get_bids_json_session_info(env: Env, bids_json: dict[str, Any]) -> SessionIn
     return get_session_info(env, patient_id, scanner_info)
 
 
-def add_bids_json_file_parameters(env: Env, bids_json_path: str, rel_json_path: str, file_parameters: dict[str, Any]):
+def add_bids_json_file_parameters(env: Env, bids_json_path: Path, rel_json_path: str, file_parameters: dict[str, Any]):
     """
     Read a BIDS JSON sidecar file and add its parameters to a LORIS file parameters dictionary.
     """

python/lib/imaging_lib/bids/tsv_participants.py

@@ -1,6 +1,7 @@
 import csv
 import re
 from dataclasses import dataclass
+from pathlib import Path
 
 from dateutil.parser import ParserError, parse
 
@@ -21,7 +22,7 @@ class BidsTsvParticipant:
     project:    str | None = None
 
 
-def read_bids_participants_tsv_file(participants_tsv_path: str) -> dict[str, BidsTsvParticipant]:
+def read_bids_participants_tsv_file(participants_tsv_path: Path) -> dict[str, BidsTsvParticipant]:
     """
     Read the `participants.tsv` file of a BIDS dataset and get the participant rows indexed by
     participant ID. Raise an exception if the `participants.tsv` file is incorrect.
@@ -42,7 +43,7 @@ def read_bids_participants_tsv_file(participants_tsv_path: str) -> dict[str, Bid
 
 def read_bids_participants_tsv_row(
     tsv_participant_row: dict[str, str],
-    participants_tsv_path: str,
+    participants_tsv_path: Path,
 ) -> BidsTsvParticipant:
     """
     Read a `participants.tsv` row, or raise an exception if that row is incorrect.
@@ -70,7 +71,7 @@ def read_bids_participants_tsv_row(
     )
 
 
-def write_bids_participants_tsv_file(tsv_participants: dict[str, BidsTsvParticipant], participants_file_path: str):
+def write_bids_participants_tsv_file(tsv_participants: dict[str, BidsTsvParticipant], participants_file_path: Path):
     """
     Write the `participants.tsv` file based from a set of participant rows.
     """

python/lib/imaging_lib/bids/tsv_scans.py

@@ -1,6 +1,7 @@
 import csv
 from dataclasses import dataclass
 from datetime import datetime
+from pathlib import Path
 from typing import Any
 
 from dateutil.parser import ParserError, parse
@@ -19,7 +20,7 @@ class BidsTsvScan:
     age_at_scan      : str | None
 
 
-def read_bids_scans_tsv_file(scans_tsv_path: str) -> dict[str, BidsTsvScan]:
+def read_bids_scans_tsv_file(scans_tsv_path: Path) -> dict[str, BidsTsvScan]:
     """
     Read a `scans.tsv` file of a BIDS dataset and get the scan rows indexed by file name. Raise an
     exception if the `scans.tsv` file is incorrect.
@@ -38,7 +39,7 @@ def read_bids_scans_tsv_file(scans_tsv_path: str) -> dict[str, BidsTsvScan]:
     return tsv_scans
 
 
-def read_bids_scans_tsv_row(tsv_scan_row: dict[str, str], scans_tsv_path: str) -> BidsTsvScan:
+def read_bids_scans_tsv_row(tsv_scan_row: dict[str, str], scans_tsv_path: Path) -> BidsTsvScan:
     """
     Read a `scans.tsv` row, or raise an exception if that row is incorrect.
     """
@@ -57,7 +58,7 @@ def read_bids_scans_tsv_row(tsv_scan_row: dict[str, str], scans_tsv_path: str) -
     )
 
 
-def write_bids_scans_tsv_file(tsv_scans: dict[str, BidsTsvScan], scans_tsv_path: str):
+def write_bids_scans_tsv_file(tsv_scans: dict[str, BidsTsvScan], scans_tsv_path: Path):
     """
     Write the `scans.tsv` file from a set of scan rows.
     """
@@ -115,7 +116,7 @@ def _read_age_at_scan(tsv_scan_row: dict[str, str]) -> str | None:
     return None
 
 
-def add_scan_tsv_file_parameters(scan_tsv: BidsTsvScan, scans_tsv_path: str, file_parameters: dict[str, Any]):
+def add_scan_tsv_file_parameters(scan_tsv: BidsTsvScan, scans_tsv_path: Path, file_parameters: dict[str, Any]):
     """
     Add a scans.tsv file and row parameters to a LORIS file parameters dictionary.
     """

python/lib/imaging_lib/nifti.py

@@ -1,11 +1,11 @@
-import os
 from collections.abc import Iterator
+from pathlib import Path
 from typing import Any, cast
 
 import nibabel as nib
 
 
-def add_nifti_file_parameters(nifti_path: str, nifti_file_hash: str, file_parameters: dict[str, Any]):
+def add_nifti_file_parameters(nifti_path: Path, nifti_file_hash: str, file_parameters: dict[str, Any]):
     """
     Read a NIfTI image and add some of its properties to the file parameters.
     """
@@ -34,11 +34,11 @@ def add_nifti_file_parameters(nifti_path: str, nifti_file_hash: str, file_parame
     file_parameters['file_blake2b_hash'] = nifti_file_hash
 
 
-def find_dir_nifti_names(dir_path: str) -> Iterator[str]:
+def find_dir_nifti_files(dir_path: Path) -> Iterator[Path]:
     """
-    Iterate over the names of the NIfTI files found in a directory.
+    Iterate over the Path objects of the NIfTI files found in a directory.
     """
 
-    for file_name in os.listdir(dir_path):
-        if file_name.endswith(('.nii', '.nii.gz')):
-            yield file_name
+    for item_path in dir_path.iterdir():
+        if item_path.name.endswith(('.nii', '.nii.gz')):
+            yield item_path