MEG BIDS importer

pyproject.toml

@@ -55,11 +55,14 @@ packages = [
 members = ["python/loris_*"]
 
 [tool.uv.sources]
-loris_bids_importer  = { workspace = true }
-loris_bids_utils     = { workspace = true }
-loris_dicom_importer = { workspace = true }
-loris_ephys_chunker  = { workspace = true }
-loris_utils          = { workspace = true }
+loris_bids_importer           = { workspace = true }
+loris_bids_utils              = { workspace = true }
+loris_dicom_importer          = { workspace = true }
+loris_ephys_chunker           = { workspace = true }
+loris_ephys_visualizer_module = { workspace = true }
+loris_meegqc_module           = { workspace = true }
+loris_server                  = { workspace = true }
+loris_utils                   = { workspace = true }
 
 [tool.ruff]
 src = ["python"]
@@ -68,15 +71,16 @@ line-length = 120
 preview = true
 
 [tool.ruff.lint]
-ignore = ["E202", "E203", "E221", "E241", "E251", "E272"]
-select = ["E", "EXE", "F", "I", "N", "RUF", "UP", "W"]
+ignore = ["E202", "E203", "E221", "E241", "E251", "E272", "FAST003"]
+select = ["E", "EXE", "F", "FAST", "I", "N", "RUF", "UP", "W"]
 
 [tool.ruff.lint.pycodestyle]
 max-doc-length = 100
 
 [tool.ruff.lint.per-file-ignores]
 # ORM models often have very long lines.
 "python/lib/db/models/*.py" = ["E501"]
+"python/loris_meegqc_module/src/loris_meegqc_module/database/models/*.py" = ["E501"]
 
 # The strict type checking configuration is used to type check only the modern (typed) modules. An
 # additional basic type checking configuration to type check legacy modules can be found in the

python/lib/config.py

@@ -7,6 +7,14 @@
 from lib.logging import log_error_exit
 
 
+def get_jwt_secret_key_config(env: Env) -> str:
+    """
+    Get the LORIS JWT secret key from the in-database configuration.
+    """
+
+    return _get_config_value(env, 'JWTKey')
+
+
 def get_patient_id_dicom_header_config(env: Env) -> Literal['PatientID', 'PatientName']:
     """
     Get the DICOM header in which to look for the patient ID from the in-database configuration, or

python/lib/db/queries/meg_ctf_head_shape.py

@@ -0,0 +1,12 @@
+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:
+    return db.execute(select(DbMegCtfHeadShapeFile)
+        .where(DbMegCtfHeadShapeFile.path == path)
+    ).scalar_one_or_none()

python/lib/db/queries/user.py

@@ -0,0 +1,24 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session as Database
+
+from lib.db.models.user import DbUser
+
+
+def try_get_user_with_id(db: Database, user_id: int) -> DbUser | None:
+    """
+    Get a user from the database using its ID, or return `None` if no user is found.
+    """
+
+    return db.execute(select(DbUser)
+        .where(DbUser.id == user_id)
+    ).scalar_one_or_none()
+
+
+def try_get_user_with_username(db: Database, username: str) -> DbUser | None:
+    """
+    Get a user from the database using its username, or return `None` if no user is found.
+    """
+
+    return db.execute(select(DbUser)
+        .where(DbUser.username == username)
+    ).scalar_one_or_none()

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/lib/user.py

@@ -0,0 +1,29 @@
+from lib.db.models.project import DbProject
+from lib.db.models.session import DbSession
+from lib.db.models.site import DbSite
+from lib.db.models.user import DbUser
+from lib.env import Env
+
+
+def can_user_access_project(_: Env, user: DbUser, project: DbProject) -> bool:
+    """
+    Check whether a user has access to a project.
+    """
+
+    return project in user.projects
+
+
+def can_user_access_site(_: Env, user: DbUser, site: DbSite) -> bool:
+    """
+    Check whether a user has access to a site.
+    """
+
+    return site in user.sites
+
+
+def can_user_access_session(env: Env, user: DbUser, session: DbSession) -> bool:
+    """
+    Check whether a user has access to a session.
+    """
+
+    return can_user_access_site(env, user, session.site) and can_user_access_project(env, user, session.project)

python/loris_bids_importer/src/loris_bids_importer/acquisitions.py

@@ -5,14 +5,14 @@
 from lib.logging import log, log_error
 from loris_bids_utils.info import BidsAcquisitionInfo
 
-from loris_bids_importer.env import BidsImportEnv
+from loris_bids_importer.importer import BidsImporter
 
 T = TypeVar('T')
 
 
 def import_bids_acquisitions(
     env: Env,
-    import_env: BidsImportEnv,
+    import_env: BidsImporter,
     acquisitions: list[tuple[T, BidsAcquisitionInfo]],
     importer: Callable[[T, BidsAcquisitionInfo], None]
 ):
@@ -32,6 +32,7 @@ def import_bids_acquisitions(
             log(env, f"Successfully imported acquisition '{bids_info.name}'.")
             import_env.imported_acquisitions_count += 1
         except Exception as exception:
+            import traceback
             log_error(
                 env,
                 (
@@ -40,4 +41,5 @@ def import_bids_acquisitions(
                     "Skipping."
                 )
             )
+            print(traceback.format_exc())
             import_env.failed_acquisitions_count += 1

python/loris_bids_importer/src/loris_bids_importer/channels.py

@@ -14,12 +14,12 @@
 from loris_utils.error import group_errors, group_errors_tuple
 
 from loris_bids_importer.copy_files import get_loris_bids_file_path
-from loris_bids_importer.env import BidsImportEnv
+from loris_bids_importer.importer import BidsImporter
 
 
 def insert_bids_channels_file(
     env: Env,
-    import_env: BidsImportEnv,
+    import_env: BidsImporter,
     physio_file: DbPhysioFile,
     session: DbSession,
     acquisition: BidsAcquisitionInfo,

python/loris_bids_importer/src/loris_bids_importer/copy_files.py

@@ -9,7 +9,7 @@
 from loris_bids_utils.files.participants import BidsParticipantsTsvFile
 from loris_bids_utils.files.scans import BidsScansTsvFile
 
-from loris_bids_importer.env import BidsImportEnv
+from loris_bids_importer.importer import BidsImporter
 
 
 def get_loris_bids_dataset_path(env: Env, dataset_description: BidsDatasetDescriptionJsonFile) -> Path:
@@ -31,21 +31,21 @@ def get_loris_bids_dataset_path(env: Env, dataset_description: BidsDatasetDescri
     return loris_bids_path
 
 
-def get_loris_bids_root_file_path(import_env: BidsImportEnv, file_path: Path) -> Path:
+def get_loris_bids_root_file_path(importer: BidsImporter, file_path: Path) -> Path:
     """
     Get the path of a BIDS file relative to the LORIS data directory, maintaining the same relative
     path in the LORIS BIDS dataset as within the source BIDS dataset.
     """
 
     # In the import is run in no-copy mode, return the original file path.
-    if import_env.loris_bids_path is None:
-        return file_path.relative_to(import_env.data_dir_path)
+    if importer.loris_bids_path is None:
+        return file_path.relative_to(importer.data_dir_path)
 
-    return import_env.loris_bids_path / file_path.relative_to(import_env.source_bids_path)
+    return importer.loris_bids_path / file_path.relative_to(importer.args.source_bids_path)
 
 
 def get_loris_bids_file_path(
-    import_env: BidsImportEnv,
+    importer: BidsImporter,
     session: DbSession,
     data_type: str,
     file_path: Path,
@@ -56,20 +56,20 @@ def get_loris_bids_file_path(
     """
 
     # In the import is run in no-copy mode, return the original file path.
-    if import_env.loris_bids_path is None:
-        return file_path.relative_to(import_env.data_dir_path)
+    if importer.loris_bids_path is None:
+        return file_path.relative_to(importer.data_dir_path)
 
     # If the file is a derivative, the path is unpredictable, so return a copy of that path in the
     # LORIS BIDS dataset.
     if derivative:
-        return import_env.loris_bids_path / file_path.relative_to(import_env.source_bids_path)
+        return importer.loris_bids_path / file_path.relative_to(importer.args.source_bids_path)
 
     # Otherwise, normalize the subject and session directory names using the LORIS session
     # information.
     loris_file_name = get_loris_bids_file_name(file_path.name, session)
 
     return (
-        import_env.loris_bids_path
+        importer.loris_bids_path
         / f'sub-{session.candidate.psc_id}'
         / f'ses-{session.visit_label}'
         / data_type
@@ -91,34 +91,43 @@ def get_loris_bids_file_name(file_name: str, session: DbSession) -> str:
     return f'sub-{session.candidate.psc_id}_ses-{session.visit_label}_{file_name}'
 
 
-def get_loris_scans_path(import_env: BidsImportEnv, scans_file: BidsScansTsvFile, session: DbSession) -> Path:
+def get_loris_scans_path(importer: BidsImporter, scans_file: BidsScansTsvFile, session: DbSession) -> Path:
     """
     Get the path of a `scans.tsv` file in LORIS, relative to the LORIS data directory.
     """
 
     # In the import is run in no-copy mode, return the original file path.
-    if import_env.loris_bids_path is None:
-        return scans_file.path.relative_to(import_env.data_dir_path)
+    if importer.loris_bids_path is None:
+        return scans_file.path.relative_to(importer.data_dir_path)
 
     loris_file_name = get_loris_bids_file_name(scans_file.path.name, session)
     return (
-        import_env.loris_bids_path
+        importer.loris_bids_path
         / f'sub-{session.candidate.psc_id}'
         / f'ses-{session.visit_label}'
         / loris_file_name
     )
 
 
-def copy_loris_bids_file(import_env: BidsImportEnv, file_path: Path, loris_file_path: Path):
+def copy_loris_bids_file(importer: BidsImporter, file_path: Path, loris_file_path: Path):
     """
     Copy a BIDS file to the LORIS data directory, unless the no-copy mode is enabled.
     """
 
+    input_bids_file_path = file_path.relative_to(importer.args.source_bids_path)
+
+    if importer.loris_bids_path is not None:
+        output_bids_file_path = loris_file_path.relative_to(importer.loris_bids_path)
+    else:
+        output_bids_file_path = input_bids_file_path
+
+    importer.files_dict[input_bids_file_path] = output_bids_file_path
+
     # Do not copy the file in no-copy mode.
-    if import_env.loris_bids_path is None:
+    if importer.loris_bids_path is None:
         return
 
-    full_loris_file_path = import_env.data_dir_path / loris_file_path
+    full_loris_file_path = importer.data_dir_path / loris_file_path
 
     if full_loris_file_path.exists():
         raise Exception(f"File '{loris_file_path}' already exists in the LORIS data directory.")
@@ -130,31 +139,31 @@ def copy_loris_bids_file(import_env: BidsImportEnv, file_path: Path, loris_file_
         shutil.copytree(file_path, full_loris_file_path)
 
 
-def copy_bids_static_files(import_env: BidsImportEnv):
+def copy_bids_static_files(importer: BidsImporter):
     """
     Copy the static files of the source BIDS dataset to the LORIS BIDS dataset.
     """
 
     # Do not copy files in no-copy mode.
-    if import_env.loris_bids_path is None:
+    if importer.loris_bids_path is None:
         return
 
     for file_name in ['README', 'dataset_description.json']:
-        source_file_path = import_env.source_bids_path / file_name
+        source_file_path = importer.args.source_bids_path / file_name
         if not source_file_path.is_file():
             continue
 
-        loris_file_path = import_env.loris_bids_path / file_name
+        loris_file_path = importer.loris_bids_path / file_name
 
         # Do not copy the file if it is already present during an incremental import.
-        if (import_env.data_dir_path / loris_file_path).is_file():
+        if (importer.data_dir_path / loris_file_path).is_file():
             continue
 
-        copy_loris_bids_file(import_env, source_file_path, loris_file_path)
+        copy_loris_bids_file(importer, source_file_path, loris_file_path)
 
 
 def copy_bids_participants_file(
-    import_env: BidsImportEnv,
+    importer: BidsImporter,
     participants_file: BidsParticipantsTsvFile,
     loris_participants_path: Path,
 ):
@@ -164,27 +173,27 @@ def copy_bids_participants_file(
     """
 
     # Do not copy the file in no-copy mode.
-    if import_env.loris_bids_path is None:
+    if importer.loris_bids_path is None:
         return
 
-    participants_path = import_env.data_dir_path / loris_participants_path
+    participants_path = importer.data_dir_path / loris_participants_path
     if participants_path.exists():
         participants_file.merge(BidsParticipantsTsvFile(participants_path))
 
     participants_path.parent.mkdir(parents=True, exist_ok=True)
     participants_file.write(participants_path)
 
 
-def copy_bids_scans_file(import_env: BidsImportEnv, scans_file: BidsScansTsvFile, loris_scans_path: Path):
+def copy_bids_scans_file(importer: BidsImporter, scans_file: BidsScansTsvFile, loris_scans_path: Path):
     """
     Copy some `scans.tsv` rows into a LORIS `scans.tsv` file, creating it if necessary.
     """
 
     # Do not copy the file in no-copy mode.
-    if import_env.loris_bids_path is None:
+    if importer.loris_bids_path is None:
         return
 
-    scans_path = import_env.data_dir_path / loris_scans_path
+    scans_path = importer.data_dir_path / loris_scans_path
     if scans_path.exists():
         scans_file.merge(BidsScansTsvFile(scans_path))