BIDS MEG CTF importer

python/lib/db/queries/meg_ctf_head_shape.py

@@ -0,0 +1,17 @@
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session as Database
+
+from lib.db.models.meg_ctf_head_shape_file import DbMegCtfHeadShapeFile
+
+
+def try_get_meg_ctf_head_shape_file_with_path(db: Database, path: Path) -> DbMegCtfHeadShapeFile | None:
+    """
+    Get a MEG CTF head shape file from the database using its path, or return `None` if no file was
+    found.
+    """
+
+    return db.execute(select(DbMegCtfHeadShapeFile)
+        .where(DbMegCtfHeadShapeFile.path == path)
+    ).scalar_one_or_none()

python/lib/physio/ctf/head_shape.py

@@ -0,0 +1,41 @@
+from decimal import Decimal
+from pathlib import Path
+
+from loris_bids_utils.meg.ctf.head_shape import MegCtfHeadShapeFile
+from loris_utils.crypto import compute_file_blake2b_hash
+
+from lib.db.models.meg_ctf_head_shape_file import DbMegCtfHeadShapeFile
+from lib.db.models.meg_ctf_head_shape_point import DbMegCtfHeadShapePoint
+from lib.env import Env
+
+
+def insert_meg_ctf_head_shape_file(
+    env: Env,
+    head_shape_file: MegCtfHeadShapeFile,
+    loris_head_shape_file_path: Path,
+) -> DbMegCtfHeadShapeFile:
+    """
+    Insert a MEG CTF head shape file into the LORIS database.
+    """
+
+    blake2b_hash = compute_file_blake2b_hash(head_shape_file.path)
+
+    db_head_shape_file = DbMegCtfHeadShapeFile(
+        path         = loris_head_shape_file_path,
+        blake2b_hash = blake2b_hash,
+    )
+
+    env.db.add(db_head_shape_file)
+    env.db.flush()
+
+    for name, point in head_shape_file.points.items():
+        env.db.add(DbMegCtfHeadShapePoint(
+            file_id = db_head_shape_file.id,
+            name    = name,
+            x       = Decimal(point.x),
+            y       = Decimal(point.y),
+            z       = Decimal(point.z),
+        ))
+
+    env.db.flush()
+    return db_head_shape_file

python/lib/physio/file.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 
 from lib.db.models.imaging_file_type import DbImagingFileType
+from lib.db.models.meg_ctf_head_shape_file import DbMegCtfHeadShapeFile
 from lib.db.models.physio_file import DbPhysioFile
 from lib.db.models.physio_modality import DbPhysioModality
 from lib.db.models.physio_output_type import DbPhysioOutputType
@@ -18,6 +19,7 @@ def insert_physio_file(
     modality: DbPhysioModality,
     output_type: DbPhysioOutputType,
     acquisition_time: datetime | None,
+    head_shape_file: DbMegCtfHeadShapeFile | None = None,
 ) -> DbPhysioFile:
     """
     Insert a physiological file into the database.
@@ -31,6 +33,7 @@ def insert_physio_file(
         output_type_id   = output_type.id,
         acquisition_time = acquisition_time,
         inserted_by_user = getpass.getuser(),
+        head_shape_file_id = head_shape_file.id if head_shape_file is not None else None,
     )
 
     env.db.add(file)

python/loris_bids_importer/src/loris_bids_importer/main.py

@@ -7,6 +7,7 @@
 from lib.db.queries.session import try_get_session_with_cand_id_visit_label
 from lib.env import Env
 from lib.logging import log, log_error, log_error_exit, log_warning
+from loris_bids_utils.meg.reader import BidsMegDataTypeReader
 from loris_bids_utils.mri.reader import BidsMriDataTypeReader
 from loris_bids_utils.reader import BidsDatasetReader, BidsDataTypeReader, BidsSessionReader
 
@@ -20,6 +21,7 @@
 from loris_bids_importer.eeg.main import Eeg
 from loris_bids_importer.env import BidsImportEnv
 from loris_bids_importer.events import import_bids_root_event_dict_file
+from loris_bids_importer.meg.ctf import import_bids_meg_data_type
 from loris_bids_importer.mri.main import import_bids_mri_data_type
 from loris_bids_importer.print import print_bids_import_summary, print_bids_info
 from loris_bids_importer.validation.sessions import validate_bids_sessions
@@ -177,11 +179,13 @@ def import_bids_data_type(
     match data_type:
         case BidsMriDataTypeReader():
             import_bids_mri_data_type(env, import_env, session, data_type)
+        case BidsMegDataTypeReader():
+            import_bids_meg_data_type(env, import_env, session, data_type)
         case BidsDataTypeReader():
-            import_bids_eeg_data_type_files(env, import_env, args, session, data_type, dataset_tag_dict, legacy_db)
+            import_bids_eeg_data_type(env, import_env, args, session, data_type, dataset_tag_dict, legacy_db)
 
 
-def import_bids_eeg_data_type_files(
+def import_bids_eeg_data_type(
     env: Env,
     import_env: BidsImportEnv,
     args: Args,

python/loris_bids_importer/src/loris_bids_importer/meg/ctf.py

@@ -0,0 +1,178 @@
+from pathlib import Path
+
+from lib.config import get_ephys_visualization_enabled_config
+from lib.db.models.meg_ctf_head_shape_file import DbMegCtfHeadShapeFile
+from lib.db.models.session import DbSession
+from lib.db.queries.hed_schema_node import get_all_hed_schema_nodes
+from lib.db.queries.meg_ctf_head_shape import try_get_meg_ctf_head_shape_file_with_path
+from lib.db.queries.physio_file import try_get_physio_file_with_path
+from lib.env import Env
+from lib.logging import log, log_warning
+from lib.physio.chunking import create_physio_channels_chunks
+from lib.physio.ctf.head_shape import insert_meg_ctf_head_shape_file
+from lib.physio.events import EventDictFileSource
+from lib.physio.file import insert_physio_file
+from lib.physio.parameters import insert_physio_file_parameter
+from loris_bids_utils.info import BidsAcquisitionInfo
+from loris_bids_utils.meg.ctf.acquisition import MegCtfAcquisition
+from loris_bids_utils.meg.ctf.copy_ds import copy_bids_meg_ctf_ds
+from loris_bids_utils.meg.reader import BidsMegDataTypeReader
+from loris_utils.error import group_errors_tuple
+
+from loris_bids_importer.acquisitions import BidsImportFileResult, BidsImportFileStatus, import_bids_acquisitions
+from loris_bids_importer.channels import insert_bids_channels_file
+from loris_bids_importer.copy_files import copy_loris_bids_file, get_loris_bids_file_path
+from loris_bids_importer.env import BidsImportEnv
+from loris_bids_importer.events import insert_bids_event_dict_file, insert_bids_events_file
+from loris_bids_importer.file_type import get_check_bids_imaging_file_type
+from loris_bids_importer.physio import (
+    get_check_bids_physio_file_hash,
+    get_check_bids_physio_modality,
+    get_check_bids_physio_output_type,
+)
+
+
+def import_bids_meg_data_type(
+    env: Env,
+    import_env: BidsImportEnv,
+    session: DbSession,
+    data_type: BidsMegDataTypeReader,
+):
+    if data_type.head_shape_file is not None:
+        head_shape_file_path = get_loris_bids_file_path(
+            import_env,
+            session,
+            data_type.name,
+            data_type.head_shape_file.path,
+        )
+
+        head_shape_file = try_get_meg_ctf_head_shape_file_with_path(env.db, head_shape_file_path)
+        if head_shape_file is None:
+            head_shape_file = insert_meg_ctf_head_shape_file(env, data_type.head_shape_file, head_shape_file_path)
+            copy_loris_bids_file(import_env, data_type.head_shape_file.path, head_shape_file_path)
+    else:
+        head_shape_file = None
+
+    import_bids_acquisitions(
+        env,
+        import_env,
+        session,
+        data_type.acquisitions,
+        lambda acquisition, bids_info: import_bids_meg_acquisition(
+            env,
+            import_env,
+            session,
+            acquisition,
+            bids_info,
+            head_shape_file,
+        ),
+    )
+
+
+def import_bids_meg_acquisition(
+    env: Env,
+    import_env: BidsImportEnv,
+    session: DbSession,
+    acquisition: MegCtfAcquisition,
+    bids_info: BidsAcquisitionInfo,
+    head_shape_file: DbMegCtfHeadShapeFile | None,
+) -> BidsImportFileResult:
+    loris_file_path = get_loris_bids_file_path(import_env, session, bids_info.data_type, acquisition.ctf_path)
+
+    loris_file = try_get_physio_file_with_path(env.db, loris_file_path)
+    if loris_file is not None:
+        return BidsImportFileResult(BidsImportFileStatus.IGNORE, loris_file_path)
+
+    modality, output_type, file_type, file_hash = group_errors_tuple(
+        f"Error while checking database information for MEG acquisition '{bids_info.name}'.",
+        lambda: get_check_bids_physio_modality(env, bids_info.data_type),
+        # TODO: Use real output type after rebasing with #1443
+        # Code: `importer.args.type or 'raw'`
+        lambda: get_check_bids_physio_output_type(env, 'raw'),
+        lambda: get_check_bids_imaging_file_type(env, 'ctf'),
+        lambda: get_check_bids_physio_file_hash(env, acquisition.ctf_path),
+    )
+
+    # The files to copy to LORIS, with the source path on the left and the LORIS path on the right.
+    files_to_copy: list[tuple[Path, Path]] = []
+
+    check_bids_meg_metadata_files(env, acquisition, bids_info)
+
+    physio_file = insert_physio_file(
+        env,
+        session,
+        loris_file_path,
+        file_type,
+        modality,
+        output_type,
+        bids_info.scan_row.get_acquisition_time() if bids_info.scan_row is not None else None,
+        head_shape_file,
+    )
+
+    insert_physio_file_parameter(env, physio_file, 'physiological_file_blake2b_hash', file_hash)
+    for name, value in acquisition.sidecar_file.data.items():
+        insert_physio_file_parameter(env, physio_file, name, value)
+
+    if acquisition.events_file is not None:
+        hed_union = get_all_hed_schema_nodes(env.db)
+
+        loris_events_file_path = get_loris_bids_file_path(
+            import_env, session, bids_info.data_type, acquisition.events_file.path
+        )
+
+        insert_bids_events_file(env, physio_file, acquisition.events_file, loris_events_file_path, {}, {}, hed_union)
+        files_to_copy.append((acquisition.events_file.path, loris_events_file_path))
+        if acquisition.events_file.dictionary is not None:
+            loris_event_dict_file_path = get_loris_bids_file_path(
+                import_env, session, bids_info.data_type, acquisition.events_file.dictionary.path
+            )
+
+            insert_bids_event_dict_file(
+                env,
+                EventDictFileSource.from_file(physio_file),
+                acquisition.events_file.dictionary,
+                loris_event_dict_file_path,
+            )
+
+            files_to_copy.append((acquisition.events_file.dictionary.path, loris_event_dict_file_path))
+
+    if acquisition.channels_file is not None:
+        insert_bids_channels_file(env, import_env, physio_file, session, bids_info, acquisition.channels_file)
+        loris_channels_file_path = get_loris_bids_file_path(
+            import_env, session, bids_info.data_type, acquisition.channels_file.path
+        )
+        files_to_copy.append((acquisition.channels_file.path, loris_channels_file_path))
+
+    if import_env.loris_bids_path is not None:
+        copy_bids_meg_ctf_ds(acquisition.ctf_path, import_env.data_dir_path / loris_file_path)
+
+    for source_path, destination_path in files_to_copy:
+        copy_loris_bids_file(import_env, source_path, destination_path)
+
+    env.db.commit()
+
+    log(env, f"MEG file successfully imported with ID: {physio_file.id}.")
+
+    if get_ephys_visualization_enabled_config(env):
+        log(env, "Creating visualization chunks...")
+        create_physio_channels_chunks(env, physio_file)
+
+    env.db.commit()
+
+    return BidsImportFileResult(BidsImportFileStatus.SUCCESS, loris_file_path)
+
+
+def check_bids_meg_metadata_files(env: Env, acquisition: MegCtfAcquisition, bids_info: BidsAcquisitionInfo):
+    """
+    Check for the presence of BIDS metadata files for the BIDS MEG acquisition and warn the user if
+    that is not the case.
+    """
+
+    if acquisition.channels_file is None:
+        log_warning(env, f"No channels file found for acquisition '{bids_info.name}'.")
+
+    if acquisition.events_file is None:
+        log_warning(env, f"No events file found for acquisition '{bids_info.name}'.")
+
+    if acquisition.events_file is not None and acquisition.events_file.dictionary is not None:
+        log_warning(env, f"No events dictionary file found for acquisition '{bids_info.name}'.")

python/loris_bids_importer/src/loris_bids_importer/physio.py

@@ -5,7 +5,7 @@
 from lib.db.queries.physio import try_get_physio_modality_with_name, try_get_physio_output_type_with_name
 from lib.db.queries.physio_file import try_get_physio_file_with_hash
 from lib.env import Env
-from loris_utils.crypto import compute_file_blake2b_hash
+from loris_utils.crypto import compute_directory_blake2b_hash, compute_file_blake2b_hash
 
 
 def get_check_bids_physio_modality(env: Env, data_type_name: str) -> DbPhysioModality:
@@ -40,7 +40,10 @@ def get_check_bids_physio_file_hash(env: Env, file_path: Path) -> str:
     registered in the database.
     """
 
-    file_hash = compute_file_blake2b_hash(file_path)
+    if file_path.is_dir():
+        file_hash = compute_directory_blake2b_hash(file_path)
+    else:
+        file_hash = compute_file_blake2b_hash(file_path)
 
     file = try_get_physio_file_with_hash(env.db, file_hash)
     if file is not None:

python/loris_bids_utils/pyproject.toml

@@ -10,6 +10,10 @@ dependencies = [
     "pybids",
 ]
 
+[project.scripts]
+copy-bids-ctf-ds = "loris_bids_utils.scripts.copy_bids_ctf_ds:main"
+copy-ctf-ds      = "loris_bids_utils.scripts.copy_ctf_ds:main"
+
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"

python/loris_bids_utils/src/loris_bids_utils/meg/ctf/acquisition.py

@@ -0,0 +1,37 @@
+
+import re
+from pathlib import Path
+
+from loris_utils.path import remove_path_extension
+
+from loris_bids_utils.eeg.channels import BidsEegChannelsTsvFile
+from loris_bids_utils.files.events import BidsEventsTsvFile
+from loris_bids_utils.meg.ctf.head_shape import MegCtfHeadShapeFile
+from loris_bids_utils.meg.sidecar import BidsMegSidecarJsonFile
+
+
+class MegCtfAcquisition:
+    ctf_path: Path
+    sidecar_file: BidsMegSidecarJsonFile
+    channels_file: BidsEegChannelsTsvFile | None
+    events_file: BidsEventsTsvFile | None
+    head_shape_file: MegCtfHeadShapeFile | None
+
+    def __init__(self, ctf_path: Path, head_shape_file: MegCtfHeadShapeFile | None):
+        self.ctf_path = ctf_path
+
+        path = remove_path_extension(ctf_path)
+
+        sidecar_path = path.with_suffix('.json')
+        if not sidecar_path.exists():
+            raise Exception("No MEG JSON sidecar file.")
+
+        self.sidecar_file = BidsMegSidecarJsonFile(sidecar_path)
+
+        channels_path = path.parent / re.sub(r'_meg$', '_channels.tsv', path.name)
+        self.channels_file = BidsEegChannelsTsvFile(channels_path) if channels_path.exists() else None
+
+        events_path = path.parent / re.sub(r'_meg$', '_events.tsv', path.name)
+        self.events_file = BidsEventsTsvFile(events_path) if events_path.exists() else None
+
+        self.head_shape_file = head_shape_file