Add test for pipeline saving.

Author: christian-oreilly

pylossless/conftest.py

@@ -1,11 +1,14 @@
 """Pytest fixtures that can be reused across our unit tests."""
 # Author: Scott Huberty <[email protected]>
+#         Christian O'Reilly <[email protected]>
 #
 # License: MIT
 
 from pathlib import Path
 import shutil
 
+import mne
+import numpy as np
 from mne import Annotations
 
 import pylossless as ll
@@ -48,3 +51,30 @@ def pipeline_fixture():
     Path("test_config.yaml").unlink()  # delete config file
     shutil.rmtree(bids_path.root)
     return pipeline
+
+
+@pytest.fixture(scope="session")
+@pytest.mark.filterwarnings("ignore:Converting data files to EDF format")
+def bids_dataset_fixture(tmpdir_factory):
+    """Make sure MNE's annotate_break function can run."""
+    def edf_import_fct(path_in):
+        # read in a file
+        raw = mne.io.read_raw_edf(path_in, preload=True)
+        match_alias = {ch_name: ch_name.strip(".") for ch_name in raw.ch_names}
+        raw.set_montage("standard_1005", match_alias=match_alias, match_case=False)
+        return raw, np.array([[0, 0, 0]]), {"test": 0, "T0": 1, "T1": 2, "T2": 3}
+
+    tmp_path = tmpdir_factory.mktemp('bids_dataset')
+    testing_path = mne.datasets.testing.data_path()
+    fname = testing_path / "EDF" / "test_edf_overlapping_annotations.edf"
+    import_args = [{"path_in": fname}]
+    bids_path_args = [{'subject': '001', 'run': '01', 'session': '01',
+                       "task": "test"}]
+    bids_path = ll.bids.convert_dataset_to_bids(
+        edf_import_fct,
+        import_args,
+        bids_path_args,
+        bids_root=tmp_path,
+        overwrite=True
+    )[0]
+    return bids_path

pylossless/pipeline.py

@@ -457,6 +457,7 @@ def __init__(self, config_path=None, config=None):
         config : pylossless.config.Config | None
             :class:`pylossless.config.Config` object for the pipeline.
         """
+        self.bids_path = None
         self.flags = {
             "ch": FlaggedChs(self),
             "epoch": FlaggedEpochs(self),
@@ -1060,7 +1061,7 @@ def flag_noisy_ics(self):
 
         # icsd_epoch_flags=padflags(raw, icsd_epoch_flags,1,'value',.5);
 
-    def save(self, derivatives_path, overwrite=False, format="EDF", event_id=None):
+    def save(self, derivatives_path=None, overwrite=False, format="EDF", event_id=None):
         """Save the file at the end of the pipeline.
 
         Parameters
@@ -1075,6 +1076,10 @@ def save(self, derivatives_path, overwrite=False, format="EDF", event_id=None):
         event_id : dict | None (default None)
             Dictionary mapping annotation descriptions to event codes.
         """
+
+        if derivatives_path is None:
+            derivatives_path = self.get_derivative_path(self.bids_path)
+
         mne_bids.write_raw_bids(
             self.raw,
             derivatives_path,
@@ -1121,22 +1126,17 @@ def filter(self):
         # 5.a. Filter lowpass/highpass
         self.raw.filter(**self.config["filtering"]["filter_args"])
 
+        # 5.b. Filter notch
         if "notch_filter_args" in self.config["filtering"]:
             notch_args = self.config["filtering"]["notch_filter_args"]
-            # in raw.notch_filter, freqs=None is ok if method=spectrum_fit
-            if not notch_args["freqs"] and "method" not in notch_args:
-                logger.info("No notch filter arguments provided. Skipping")
-            else:
+            spectrum_fit_method = (
+                "method" in notch_args and notch_args["method"] == "spectrum_fit"
+            )
+            if notch_args["freqs"] or spectrum_fit_method:
+                # in raw.notch_filter, freqs=None is ok if method=='spectrum_fit'
                 self.raw.notch_filter(**notch_args)
-
-        # 5.b. Filter notch
-        notch_args = self.config["filtering"]["notch_filter_args"]
-        spectrum_fit_method = (
-            "method" in notch_args and notch_args["method"] == "spectrum_fit"
-        )
-        if notch_args["freqs"] or spectrum_fit_method:
-            # in raw.notch_filter, freqs=None is ok if method=='spectrum_fit'
-            self.raw.notch_filter(**notch_args)
+            else:
+                logger.info("No notch filter arguments provided. Skipping")
         else:
             logger.info("No notch filter arguments provided. Skipping")
 

pylossless/tests/test_pipeline.py

@@ -1,11 +1,10 @@
 from pathlib import Path
-
+import mne
 import pytest
+import shutil
 
 import pylossless as ll
 
-import mne
-
 
 def test_empty_repr(tmp_path):
     """Test the __repr__ method for a pipeline that hasn't run."""
@@ -25,6 +24,25 @@ def test_pipeline_run(pipeline_fixture):
     assert pipeline_fixture.flags["ch"].__repr__()
 
 
+@pytest.mark.filterwarnings("ignore:Converting data files to EDF format")
+@pytest.mark.filterwarnings("ignore:The provided Epochs instance is not filtered between 1 and 100 Hz.")
+def test_pipeline_save(bids_dataset_fixture, tmp_path):
+    """Test running the pipeline."""
+    config = ll.config.Config()
+    config.load_default()
+    config["filtering"]["filter_args"]["h_freq"] = 40
+    del config["filtering"]["notch_filter_args"]
+
+    pipeline = ll.LosslessPipeline(config=config)
+    pipeline.run(bids_dataset_fixture, save=True)
+
+    with pytest.raises(FileExistsError):
+        pipeline.save(overwrite=False, format="EDF")
+    pipeline.save(overwrite=True, format="EDF")
+
+    shutil.rmtree(bids_dataset_fixture.root)
+
+
 @pytest.mark.parametrize("logging", [True, False])
 def test_find_breaks(logging):
     """Make sure MNE's annotate_break function can run."""