allow calibration and ctc files to be None

docs/source/dev.md.inc

@@ -6,6 +6,7 @@
 - New config option [`allow_missing_sessions`][mne_bids_pipeline._config.allow_missing_sessions] allows to continue when not all sessions are present for all subjects. (#1000 by @drammock)
 - New config option [`mf_extra_kws`][mne_bids_pipeline._config.mf_extra_kws] passes additional keyword arguments to `mne.preprocessing.maxwell_filter`. (#1038 by @drammock)
 - New value `"twa"` for config option [`mf_destination`][mne_bids_pipeline._config.mf_destination], to use the time-weighted average head position across runs as the destination position. (#1043 and #1055 by @drammock)
+- New config options [`mf_cal_missing`][mne_bids_pipeline._config.mf_cal_missing] and [`mf_ctc_missing`][mne_bids_pipeline._config.mf_ctc_missing] for handling missing calibration and cross-talk files (#1057 by @harrisonritz)
 
 ### :warning: Behavior changes
 
@@ -29,6 +30,7 @@
 - Fix bug where the ``config.proc`` parameter was not used properly during forward model creation (#1014 by @larsoner)
 - Fix bug where emptyroom recordings containing EEG channels would crash the pipeline during maxwell filtering (#1040 by @drammock)
 
+
 ### :books: Documentation
 
 - Choose the theme (dark of light) automatically based on the user's operating system setting (#979 by @hoechenberger)

mne_bids_pipeline/_config.py

@@ -704,6 +704,13 @@
     ```
 """  # noqa : E501
 
+mf_cal_missing: Literal["ignore", "warn", "raise"] = "raise"
+"""
+How to handle the situation where the MEG device's fine calibration file is missing.
+Possible options are to ignore the missing file (as may be appropriate for OPM data),
+issue a warning, or raise an error.
+"""
+
 mf_ctc_fname: str | None = None
 """
 Path to the Maxwell Filter cross-talk file. If `None`, the recommended
@@ -719,6 +726,13 @@
     ```
 """  # noqa : E501
 
+mf_ctc_missing: Literal["ignore", "warn", "raise"] = "raise"
+"""
+How to handle the situation where the MEG device's cross-talk file is missing. Possible
+options are to ignore the missing file (as may be appropriate for OPM data), issue a
+warning, or raise an error (appropriate for data from Electa/Neuromag/MEGIN systems).
+"""
+
 mf_esss: int = 0
 """
 Number of extended SSS (eSSS) basis projectors to use from empty-room data.

mne_bids_pipeline/_config_utils.py

@@ -452,54 +452,80 @@ def sanitize_cond_name(cond: str) -> str:
 
 def get_mf_cal_fname(
     *, config: SimpleNamespace, subject: str, session: str | None
-) -> pathlib.Path:
+) -> pathlib.Path | None:
+    msg = "Could not find Maxwell Filter calibration file {where}."
     if config.mf_cal_fname is None:
         bids_path = BIDSPath(
             subject=subject,
             session=session,
             suffix="meg",
             datatype="meg",
             root=config.bids_root,
-        ).match()[0]
-        mf_cal_fpath = bids_path.meg_calibration_fpath
-        if mf_cal_fpath is None:
-            raise ValueError(
-                "Could not determine Maxwell Filter Calibration file from BIDS "
-                f"definition for file {bids_path}."
-            )
+        ).match()
+        if len(bids_path) > 0:
+            mf_cal_fpath = bids_path[0].meg_calibration_fpath
+        else:
+            msg = msg.format(where=f"from BIDSPath {bids_path}")
+            if config.mf_cal_missing == "raise":
+                raise ValueError(msg)
+            else:
+                mf_cal_fpath = None
+                if config.mf_cal_missing == "warn":
+                    msg = f"WARNING: {msg} Set to None."
+                    logger.info(**gen_log_kwargs(message=msg))
     else:
         mf_cal_fpath = pathlib.Path(config.mf_cal_fname).expanduser().absolute()
         if not mf_cal_fpath.exists():
-            raise ValueError(
-                f"Could not find Maxwell Filter Calibration "
-                f"file at {str(mf_cal_fpath)}."
-            )
-
-    assert isinstance(mf_cal_fpath, pathlib.Path), type(mf_cal_fpath)
+            msg = msg.format(where=f"at {str(config.mf_cal_fname)}")
+            if config.mf_cal_missing == "raise":
+                raise ValueError(msg)
+            else:
+                mf_cal_fpath = None
+                if config.mf_cal_missing == "warn":
+                    msg = f"WARNING: {msg} Set to None."
+                    logger.info(**gen_log_kwargs(message=msg))
+
+    assert isinstance(mf_cal_fpath, pathlib.Path | None), type(mf_cal_fpath)
     return mf_cal_fpath
 
 
 def get_mf_ctc_fname(
     *, config: SimpleNamespace, subject: str, session: str | None
-) -> pathlib.Path:
+) -> pathlib.Path | None:
+    msg = "Could not find Maxwell Filter cross-talk file {where}."
     if config.mf_ctc_fname is None:
-        mf_ctc_fpath = BIDSPath(
+        bids_path = BIDSPath(
             subject=subject,
             session=session,
             suffix="meg",
             datatype="meg",
             root=config.bids_root,
-        ).meg_crosstalk_fpath
-        if mf_ctc_fpath is None:
-            raise ValueError("Could not find Maxwell Filter cross-talk file.")
+        ).match()
+        if len(bids_path) > 0:
+            mf_ctc_fpath = bids_path[0].meg_crosstalk_fpath
+        else:
+            msg = msg.format(where=f"from BIDSPath {bids_path}")
+            if config.mf_ctc_missing == "raise":
+                raise ValueError(msg)
+            else:
+                mf_ctc_fpath = None
+                if config.mf_ctc_missing == "warn":
+                    msg = f"WARNING: {msg} Set to None."
+                    logger.info(**gen_log_kwargs(message=msg))
+
     else:
         mf_ctc_fpath = pathlib.Path(config.mf_ctc_fname).expanduser().absolute()
         if not mf_ctc_fpath.exists():
-            raise ValueError(
-                f"Could not find Maxwell Filter cross-talk file at {str(mf_ctc_fpath)}."
-            )
-
-    assert isinstance(mf_ctc_fpath, pathlib.Path), type(mf_ctc_fpath)
+            msg = msg.format(where=f"at {str(config.mf_ctc_fname)}")
+            if config.mf_ctc_missing == "raise":
+                raise ValueError(msg)
+            else:
+                mf_ctc_fpath = None
+                if config.mf_ctc_missing == "warn":
+                    msg = f"WARNING: {msg} Set to None."
+                    logger.info(**gen_log_kwargs(message=msg))
+
+    assert isinstance(mf_ctc_fpath, pathlib.Path | None), type(mf_ctc_fpath)
     return mf_ctc_fpath
 
 

mne_bids_pipeline/steps/preprocessing/_01_data_quality.py

@@ -60,6 +60,14 @@ def get_input_fnames_data_quality(
                 add_bads=False,
             )
         )
+
+    # set calibration and crosstalk files (if provided)
+    if _do_mf_autobad(cfg=cfg):
+        if cfg.mf_cal_fname is not None:
+            in_files["mf_cal_fname"] = cfg.mf_cal_fname
+        if cfg.mf_ctc_fname is not None:
+            in_files["mf_ctc_fname"] = cfg.mf_ctc_fname
+
     return in_files
 
 
@@ -88,6 +96,7 @@ def assess_data_quality(
         bids_path_ref_in = None
     msg, _ = _read_raw_msg(bids_path_in=bids_path_in, run=run, task=task)
     logger.info(**gen_log_kwargs(message=msg))
+
     if run is None and task == "noise":
         raw = import_er_data(
             cfg=cfg,
@@ -111,6 +120,10 @@ def assess_data_quality(
     auto_noisy_chs: list[str] | None = None
     auto_flat_chs: list[str] | None = None
     if _do_mf_autobad(cfg=cfg):
+        # use calibration and crosstalk files (if provided)
+        cfg.mf_cal_fname = in_files.pop("mf_cal_fname", None)
+        cfg.mf_ctc_fname = in_files.pop("mf_ctc_fname", None)
+
         (
             auto_noisy_chs,
             auto_flat_chs,

mne_bids_pipeline/steps/preprocessing/_03_maxfilter.py

@@ -281,9 +281,12 @@ def get_input_fnames_maxwell_filter(
         )
         _update_for_splits(in_files, key, single=True)
 
-    # standard files
-    in_files["mf_cal_fname"] = cfg.mf_cal_fname
-    in_files["mf_ctc_fname"] = cfg.mf_ctc_fname
+    # set calibration and crosstalk files (if provided)
+    if cfg.mf_cal_fname is not None:
+        in_files["mf_cal_fname"] = cfg.mf_cal_fname
+    if cfg.mf_ctc_fname is not None:
+        in_files["mf_ctc_fname"] = cfg.mf_ctc_fname
+
     return in_files
 
 
@@ -380,8 +383,8 @@ def run_maxwell_filter(
     apply_msg += " to"
 
     mf_kws = dict(
-        calibration=in_files.pop("mf_cal_fname"),
-        cross_talk=in_files.pop("mf_ctc_fname"),
+        calibration=in_files.pop("mf_cal_fname", None),
+        cross_talk=in_files.pop("mf_ctc_fname", None),
         st_duration=cfg.mf_st_duration,
         st_correlation=cfg.mf_st_correlation,
         origin=cfg.mf_head_origin,

mne_bids_pipeline/tests/configs/config_ds000248_mf.py

@@ -0,0 +1,61 @@
+"""MNE Sample Data: M/EEG combined processing."""
+
+import mne
+import mne_bids
+
+bids_root = "~/mne_data/ds000248"
+deriv_root = "~/mne_data/derivatives/mne-bids-pipeline/ds000248_mf"
+subjects_dir = f"{bids_root}/derivatives/freesurfer/subjects"
+
+subjects = ["01"]
+rename_events = {"Smiley": "Emoji", "Button": "Switch"}
+conditions = ["Auditory", "Visual", "Auditory/Left", "Auditory/Right"]
+epochs_metadata_query = "index > 0"  # Just for testing!
+contrasts = [("Visual", "Auditory"), ("Auditory/Right", "Auditory/Left")]
+
+time_frequency_conditions = ["Auditory", "Visual"]
+
+ch_types = ["meg", "eeg"]
+mf_reference_run = "01"
+find_flat_channels_meg = True
+find_noisy_channels_meg = True
+use_maxwell_filter = True
+
+# ADJUST THESE TO TEST ERROR HANDLING ------------------------------------------
+mf_cal_fname = None
+mf_cal_missing = "warn"
+
+mf_ctc_fname = f"{bids_root}/wrong_ctc.fif"
+mf_ctc_missing = "warn"
+# -----------------------------------------------------------------------------
+
+
+def noise_cov(bp: mne_bids.BIDSPath) -> mne.Covariance:
+    """Estimate the noise covariance."""
+    # Use pre-stimulus period as noise source
+    bp = bp.copy().update(suffix="epo")
+    if not bp.fpath.exists():
+        bp.update(split="01")
+    epo = mne.read_epochs(bp)
+    cov = mne.compute_covariance(epo, rank="info", tmax=0)
+    return cov
+
+
+spatial_filter = "ssp"
+n_proj_eog = dict(n_mag=1, n_grad=1, n_eeg=1)
+n_proj_ecg = dict(n_mag=1, n_grad=1, n_eeg=0)
+ssp_meg = "combined"
+ecg_proj_from_average = True
+eog_proj_from_average = False
+epochs_decim = 4
+
+bem_mri_images = "FLASH"
+recreate_bem = True
+
+n_jobs = 2
+
+
+def mri_t1_path_generator(bids_path: mne_bids.BIDSPath) -> mne_bids.BIDSPath:
+    """Return the path to a T1 image."""
+    # don't really do any modifications – just for testing!
+    return bids_path

mne_bids_pipeline/tests/test_run.py

@@ -88,6 +88,9 @@ class _TestOptionsT(TypedDict, total=False):
     "ds000248_no_mri": {
         "steps": ("preprocessing", "sensor", "source"),
     },
+    "ds000248_mf": {
+        "steps": ("preprocessing"),
+    },
     "ds001810": {
         "steps": ("preprocessing", "preprocessing", "sensor"),
     },