Adding test for flags after loading.

Author: christian-oreilly

pylossless/datasets/datasets.py

@@ -6,7 +6,7 @@
 import pylossless as ll
 
 
-def load_openneuro_bids(subject="pd6"):
+def load_openneuro_bids(subject="pd6", timeout=20):
     """Download and Load BIDS dataset ds002778 from OpenNeuro.
 
     Parameters
@@ -72,8 +72,14 @@ def load_openneuro_bids(subject="pd6"):
         root=bids_root,
     )
 
-    while not bids_path.fpath.with_suffix(".bdf").exists():
-        print(list(bids_path.fpath.glob("*")))
+    for _ in range(timeout):
+        if bids_path.fpath.with_suffix(".bdf").exists():
+            break
+        print("Waiting for .bdf files to be created. Current files available:",
+              list(bids_path.fpath.glob("*")))
         sleep(1)
+    else:
+        raise TimeoutError("OpenNeuro failed to create the .bdf files.")
+
     raw = mne_bids.read_raw_bids(bids_path, verbose="ERROR")
     return raw, config, bids_path

pylossless/flagging.py

@@ -52,6 +52,11 @@ def __init__(self, ll, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.ll = ll
 
+    @property
+    def valid_keys(self):
+        """Return the valid keys for FlaggedChs objects."""
+        return ('ch_sd', 'bridge', 'low_r', 'rank')
+
     def __repr__(self):
         """Return a string representation of the FlaggedChs object."""
         return (
@@ -62,13 +67,20 @@ def __repr__(self):
             f"  Rank: {self.get('rank', None)}\n"
         )
 
+    def __eq__(self, other):
+        for key in self.valid_keys:
+            if not np.array_equal(self.get(key, np.array([])),
+                                  other.get(key, np.array([]))):
+                return False
+        return True
+
     def add_flag_cat(self, kind, bad_ch_names, *args):
         """Store channel names that have been flagged by pipeline.
 
         Parameters
         ----------
         kind : str
-            Should be one of ``'outlier'``, ``'ch_sd'``, ``'low_r'``,
+            Should be one of ``'ch_sd'``, ``'low_r'``,
             ``'bridge'``, ``'rank'``.
         bad_ch_names : list | tuple
             Channel names. Will be the values corresponding to the ``kind``
@@ -175,6 +187,27 @@ def __init__(self, ll, *args, **kwargs):
 
         self.ll = ll
 
+    @property
+    def valid_keys(self):
+        """Return the valid keys for FlaggedEpochs objects."""
+        return ('noisy', 'uncorrelated', 'noisy_ICs')
+
+    def __repr__(self):
+        """Return a string representation of the FlaggedEpochs object."""
+        return (
+            f"Flagged channels: |\n"
+            f"  Noisy: {self.get('noisy', None)}\n"
+            f"  Noisy ICs: {self.get('noisy_ICs', None)}\n"
+            f"  Uncorrelated: {self.get('uncorrelated', None)}\n"
+        )
+
+    def __eq__(self, other):
+        for key in self.valid_keys:
+            if not np.array_equal(self.get(key, np.array([])),
+                                  other.get(key, np.array([]))):
+                return False
+        return True
+
     def add_flag_cat(self, kind, bad_epoch_inds, epochs):
         """Add information on time periods flagged by pyLossless.
 

pylossless/pipeline.py

@@ -1216,9 +1216,20 @@ def run_dataset(self, paths):
         for path in paths:
             self.run(path)
 
-    # TODO: Finish docstring
     def load_ll_derivative(self, derivatives_path):
-        """Load a completed pylossless derivative state."""
+        """Load a completed pylossless derivative state.
+
+        Parameters
+        ----------
+        derivatives_path : str | mne_bids.BIDSPath
+            Path to a saved pylossless derivatives.
+
+        Returns
+        -------
+        :class:`~pylossless.pipeline.LosslessPipeline`
+            Returns an instance of :class:`~pylossless.pipeline.LosslessPipeline`
+            for the loaded pylossless derivative state.
+        """
         if not isinstance(derivatives_path, BIDSPath):
             derivatives_path = get_bids_path_from_fname(derivatives_path)
         self.raw = mne_bids.read_raw_bids(derivatives_path)

pylossless/tests/test_pipeline.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-
+import mne_bids
 import pytest
 
 import pylossless as ll
@@ -54,3 +54,37 @@ def test_deprecation():
     # with pytest.raises(DeprecationWarning, match=f"config_fname is deprecated"):
     # DeprecationWarning are currently ignored by pytest given our toml file
     pipeline.config_fname = pipeline.config_fname
+
+
+@pytest.mark.filterwarnings("ignore:Converting data files to EDF format")
+def test_load_flags(pipeline_fixture, tmp_path):
+    """Test running the pipeline."""
+    bids_root = tmp_path / "derivatives" / "pylossless"
+    print(bids_root)
+
+    subject = "pd6"
+    datatype = "eeg"
+    session = "off"
+    task = "rest"
+    suffix = "eeg"
+    bids_path = mne_bids.BIDSPath(
+        subject=subject,
+        session=session,
+        task=task,
+        suffix=suffix,
+        datatype=datatype,
+        root=bids_root
+    )
+
+    pipeline_fixture.save(bids_path,
+                          overwrite=False, format="EDF", event_id=None)
+    pipeline = ll.LosslessPipeline().load_ll_derivative(bids_path)
+    assert pipeline_fixture.flags['ch'] == pipeline.flags['ch']
+    pipeline.flags['ch']["bridge"] = ["xx"]
+    assert pipeline_fixture.flags['ch'] != pipeline.flags['ch']
+
+    assert pipeline_fixture.flags['epoch'] == pipeline.flags['epoch']
+    pipeline.flags['epoch']["bridge"] = ["noisy"]
+    assert pipeline_fixture.flags['epoch'] == pipeline.flags['epoch']
+
+    assert pipeline_fixture.flags['ic'] == pipeline.flags['ic']