wip

Author: MaximeBICMTL

pyproject.toml

@@ -24,6 +24,7 @@ dependencies = [
     "numpy",
     "protobuf>=3.0.0",
     "pybids==0.17.0",
+    "pydantic",
     "pydicom",
     "python-dateutil",
     "scikit-learn",
@@ -56,7 +57,7 @@ line-length = 120
 preview = true
 
 [tool.ruff.lint]
-ignore = ["E202", "E203", "E221", "E241", "E251", "E272"]
+ignore = ["E202", "E203", "E221", "E241", "E251", "E271", "E272"]
 select = ["E", "EXE", "F", "I", "N", "RUF", "UP", "W"]
 
 # The strict type checking configuration is used to type check only the modern (typed) modules. An

python/lib/imaging_lib/bids/dataset.py

@@ -230,10 +230,28 @@ def data_types(self) -> list['BIDSDataType']:
             if not file.is_dir():
                 continue
 
+            # TODO: Specialized code because only MEG has a specialized data type for now. However,
+            # this function should be reworked once every modality has specialized data types.
+            if file.name == 'meg' and self.meg is not None:
+                data_types.append(self.meg)
+                continue
+
             data_types.append(BIDSDataType(self, file.name))
 
         return data_types
 
+    @cached_property
+    def meg(self) -> 'BIDSMEGDataType | None':
+        """
+        The MEG data type directory found in this session directory, if there is one.
+        """
+
+        meg_data_type_path = self.path / 'meg'
+        if not meg_data_type_path.exists():
+            return None
+
+        return BIDSMEGDataType(self, 'meg')
+
     @cached_property
     def tsv_scans(self) -> dict[str, BidsTsvScan] | None:
         """
@@ -358,3 +376,39 @@ def get_bvec_path(self) -> Path | None:
             return None
 
         return bvec_path
+
+
+class BIDSMEGDataType(BIDSDataType):
+    @cached_property
+    def acquisitions(self) -> list['BIDSMEGAcquisition']:
+        """
+        The MEG acquisitions found in the MEG data type.
+        """
+
+        acquisitions: list[BIDSMEGAcquisition] = []
+        for acquisition_name in find_dir_meg_acquisition_names(self.path):
+            acquisitions.append(BIDSMEGAcquisition(self, acquisition_name))
+
+        return acquisitions
+
+
+class BIDSMEGAcquisition:
+    data_type: BIDSMEGDataType
+    name: str
+    sidecar_path: Path
+
+    def __init__(self, data_type: BIDSMEGDataType, name: str):
+        self.data_type    = data_type
+        self.name         = name
+        self.sidecar_path = (self.data_type.path / name).with_suffix('.json')
+
+
+def find_dir_meg_acquisition_names(dir_path: Path) -> Iterator[str]:
+    """
+    Iterate over the Path objects of the NIfTI files found in a directory.
+    """
+
+    for item_path in dir_path.iterdir():
+        name_match = re.search(r'(.+_meg)\.json', item_path.name)
+        if name_match is not None:
+            yield name_match.group(1)

python/lib/imaging_lib/bids/meg/sidecar.py

@@ -0,0 +1,81 @@
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+NA            = Literal['n/a']
+RecordingType = Literal['continuous', 'epoched', 'discontinuous']
+Manufacturer  = Literal['CTF', 'Neuromag/Elekta/MEGIN', 'BTi/4D', 'KIT/Yokogawa', 'ITAB', 'KRISS', 'Other']
+
+
+class BIDSMEGSidecar(BaseModel):
+    """
+    Model for the BIDS MEG sidecar JSON file.
+
+    Documentation: https://bids-specification.readthedocs.io/en/stable/modality-specific-files/magnetoencephalography.html#sidecar-json-_megjson
+    """
+
+    # REQUIRED fields
+    sampling_frequency    : float                          = Field(..., gt=0)
+    power_line_frequency  : float | NA                     = Field(...)
+    dewar_position        : str                            = Field(...)
+    software_filters      : dict[str, dict[str, Any]] | NA = Field(...)
+    digitized_landmarks   : bool                           = Field(...)
+    digitized_head_points : bool                           = Field(...)
+
+    # RECOMMENDED fields
+    meg_channel_count     : int | None = Field(None, ge=0)
+    meg_ref_channel_count : int | None = Field(None, ge=0)
+    eeg_channel_count     : int | None = Field(None, ge=0)
+    ecog_channel_count    : int | None = Field(None, ge=0)
+    seeg_channel_count    : int | None = Field(None, ge=0)
+    eog_channel_count     : int | None = Field(None, ge=0)
+    ecg_channel_count     : int | None = Field(None, ge=0)
+    emg_channel_count     : int | None = Field(None, ge=0)
+    misc_channel_count    : int | None = Field(None, ge=0)
+    trigger_channel_count : int | None = Field(None, ge=0)
+
+    # RECOMMENDED recording fields
+    recording_duration           : float | None                          = Field(None, ge=0)
+    recording_type               : RecordingType | None                  = Field(None)
+    epoch_length                 : float | None                          = Field(None, ge=0)
+    continuous_head_localization : bool | None                           = Field(None)
+    head_coil_frequency          : list[float] | float | None            = Field(None)
+    max_movement                 : float | None                          = Field(None, ge=0)
+    subject_artefact_description : str | NA | None                       = Field(None)
+    associated_empty_room        : list[str] | str | None                = Field(None)
+    hardware_filters             : dict[str, dict[str, Any]] | NA | None = Field(None)
+
+    # OPTIONAL electrical stimulation fields
+    electrical_stimulation            : bool | None = Field(None)
+    electrical_stimulation_parameters : str | None  = Field(None)
+
+    # RECOMMENDED hardware information fields
+    manufacturer             : Manufacturer | None = Field(None)
+    manufacturers_model_name : str | None          = Field(None)
+    software_versions        : str | None          = Field(None)
+    device_serial_number     : str | None          = Field(None)
+
+    # REQUIRED and RECOMMENDED task information fields
+    task_name        : str        = Field(...)
+    task_description : str | None = Field(None)
+    instructions     : str | None = Field(None)
+    cog_atlas_id     : str | None = Field(None)
+    cog_po_id        : str | None = Field(None)
+
+    # RECOMMENDED institution information fields
+    institution_name              : str | None = Field(None)
+    institution_address           : str | None = Field(None)
+    institutional_department_name : str | None = Field(None)
+
+    # OPTIONAL EEG-specific fields (if recorded with MEG)
+    eeg_placement_scheme         : str | None = Field(None)
+    cap_manufacturer             : str | None = Field(None)
+    cap_manufacturers_model_name : str | None = Field(None)
+    eeg_reference                : str | None = Field(None)
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        extra='forbid',
+        validate_assignment=True,
+        populate_by_name=True,
+    )

python/lib/import_bids_dataset/main.py

@@ -2,7 +2,7 @@
 import re
 import shutil
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 from lib.config import get_data_dir_path_config, get_default_bids_visit_label_config
 from lib.database import Database
@@ -11,7 +11,7 @@
 from lib.db.queries.session import try_get_session_with_cand_id_visit_label
 from lib.eeg import Eeg
 from lib.env import Env
-from lib.imaging_lib.bids.dataset import BIDSDataset, BIDSDataType, BIDSSession
+from lib.imaging_lib.bids.dataset import BIDSDataset, BIDSDataType, BIDSMEGDataType, BIDSSession
 from lib.imaging_lib.bids.dataset_description import BidsDatasetDescriptionError
 from lib.imaging_lib.bids.tsv_participants import (
     BidsTsvParticipant,
@@ -37,10 +37,6 @@
 from lib.logging import log, log_error, log_error_exit, log_warning
 from lib.util.iter import count
 
-BIDS_EEG_DATA_TYPES = ['eeg', 'ieeg']
-
-BIDS_MRI_DATA_TYPES = ['anat', 'dwi', 'fmap', 'func']
-
 
 def import_bids_dataset(env: Env, args: Args, legacy_db: Database):
     """
@@ -177,12 +173,15 @@ def import_bids_data_type_files(
     Read the provided BIDS data type directory and import it into LORIS.
     """
 
-    if data_type.name in BIDS_MRI_DATA_TYPES:
-        import_bids_mri_data_type_files(env, import_env, args, session, data_type)
-    elif data_type.name in BIDS_EEG_DATA_TYPES:
-        import_bids_eeg_data_type_files(env, import_env, args, session, data_type, events_metadata, legacy_db)
-    else:
-        log_warning(env, f"Unknown data type '{data_type.name}'. Skipping.")
+    match data_type.name:
+        case 'anat' | 'dwi' | 'fmap' | 'func':
+            import_bids_mri_data_type_files(env, import_env, args, session, data_type)
+        case 'eeg' | 'ieeg':
+            import_bids_eeg_data_type_files(env, import_env, args, session, data_type, events_metadata, legacy_db)
+        case 'meg':
+            import_bids_meg_data_type_files(env, import_env, args, session, cast(BIDSMEGDataType, data_type))
+        case _:
+            log_warning(env, f"Unknown data type '{data_type.name}'. Skipping.")
 
 
 def import_bids_mri_data_type_files(
@@ -249,6 +248,21 @@ def import_bids_eeg_data_type_files(
     )
 
 
+def import_bids_meg_data_type_files(
+    env: Env,
+    import_env: BIDSImportEnv,
+    args: Args,
+    session: DbSession,
+    data_type: BIDSMEGDataType,
+):
+    """
+    Read the BIDS MEG data type directory and import its files into LORIS.
+    """
+
+    for acquisition in data_type.acquisitions:
+        log(env, f"Found MEG acquisition '{acquisition.name}'.")
+
+
 def copy_bids_tsv_participants(tsv_participants: dict[str, BidsTsvParticipant], loris_participants_tsv_path: Path):
     """
     Copy some participants.tsv rows into the LORIS participants.tsv file, creating it if necessary.