test for epoched recording

Author: dominikwelke

mne/io/curry/curry.py

@@ -191,13 +191,27 @@ def _get_curry_epoch_info(fname):
 def _get_curry_meg_normals(fname):
     fname_hdr = _check_curry_header_filename(fname)
     normals_str = fname_hdr.read_text().split("\n")
-    i_start, i_stop = [
+    # i_start, i_stop = [
+    #    i
+    #    for i, ll in enumerate(normals_str)
+    #    if ("NORMALS" in ll and "START_LIST" in ll)
+    #    or ("NORMALS" in ll and "END_LIST" in ll)
+    # ]
+    # normals_str = [nn.split("\t") for nn in normals_str[i_start + 1 : i_stop]]
+    i_list = [
         i
         for i, ll in enumerate(normals_str)
         if ("NORMALS" in ll and "START_LIST" in ll)
         or ("NORMALS" in ll and "END_LIST" in ll)
     ]
-    normals_str = [nn.split("\t") for nn in normals_str[i_start + 1 : i_stop]]
+    assert len(i_list) % 2 == 0
+    i_start_list = i_list[::2]
+    i_stop_list = i_list[1::2]
+    normals_str = [
+        nn.split("\t")
+        for i_start, i_stop in zip(i_start_list, i_stop_list)
+        for nn in normals_str[i_start + 1 : i_stop]
+    ]
     return np.array([[float(nnn.strip()) for nnn in nn] for nn in normals_str])
 
 
@@ -296,10 +310,18 @@ def _extract_curry_info(fname):
         ch_names = [
             ch_names_full[i] for i in np.argsort(chaninfile_full) if i not in i_drop
         ]
+        ch_pos = np.array(
+            [
+                ch_pos[i]
+                for i in np.argsort(chaninfile_full)
+                if (i not in i_drop) and (i < len(ch_pos))
+            ]
+        )
         ch_types = [ch_types[i] for i in np.argsort(chaninfile_full) if i not in i_drop]
         units = [units[i] for i in np.argsort(chaninfile_full) if i not in i_drop]
 
     assert len(ch_types) == len(units) == len(ch_names) == n_ch
+    assert len(ch_pos) == ch_types.count("eeg") + ch_types.count("mag")
 
     # finetune channel types (e.g. stim, eog etc might be identified by name)
     # TODO?
@@ -370,8 +392,9 @@ def _make_curry_montage(ch_names, ch_types, ch_pos, landmarks, landmarkslabels):
     ch_pos /= 1000.0
     landmarks /= 1000.0
     # channel locations
-    # TODO - what about misc without pos? can they mess things up if unordered?
+    # what about misc without pos? can they mess things up if unordered?
     assert len(ch_pos) >= (ch_types.count("mag") + ch_types.count("eeg"))
+    assert len(ch_pos) == (ch_types.count("mag") + ch_types.count("eeg"))
     ch_pos_eeg = {
         ch_names[i]: ch_pos[i, :3] for i, t in enumerate(ch_types) if t == "eeg"
     }
@@ -417,8 +440,9 @@ def _set_chanloc_curry(inst, ch_types, ch_pos, landmarks, landmarkslabels):
     ch_pos /= 1000.0
     landmarks /= 1000.0
     # channel locations
-    # TODO - what about misc without pos? can they mess things up if unordered?
+    # what about misc without pos? can they mess things up if unordered?
     assert len(ch_pos) >= (ch_types.count("mag") + ch_types.count("eeg"))
+    assert len(ch_pos) == (ch_types.count("mag") + ch_types.count("eeg"))
     ch_pos_meg = {
         ch_names[i]: ch_pos[i, :3] for i, t in enumerate(ch_types) if t == "mag"
     }
@@ -433,24 +457,24 @@ def _set_chanloc_curry(inst, ch_types, ch_pos, landmarks, landmarkslabels):
             landmark_dict[k] = None
     if len(landmarkslabels) > 0:
         hpi_pos = landmarks[
-            [i for i, n in enumerate(landmarkslabels) if re.match("HPI[1-99]", n)],
+            [i for i, n in enumerate(landmarkslabels) if re.match("HPI.?[1-99]", n)],
             :,
         ]
     else:
         hpi_pos = None
     if len(landmarkslabels) > 0:
         hsp_pos = landmarks[
-            [i for i, n in enumerate(landmarkslabels) if re.match("H[1-99]", n)], :
+            [i for i, n in enumerate(landmarkslabels) if re.match("H.?[1-99]", n)], :
         ]
     else:
         hsp_pos = None
 
     add_missing_fiducials = (
         True
         if (
-            not landmark_dict["Nas"]
-            and not landmark_dict["LPA"]
-            and not landmark_dict["LPA"]
+            isinstance(landmark_dict["Nas"], type(None))
+            and isinstance(landmark_dict["LPA"], type(None))
+            and isinstance(landmark_dict["RPA"], type(None))
         )
         else False  # raises otherwise
     )

mne/io/curry/tests/test_curry.py

@@ -3,7 +3,6 @@
 # License: BSD-3-Clause
 # Copyright the MNE-Python contributors.
 
-
 import numpy as np
 import pytest
 from numpy import empty
@@ -12,9 +11,14 @@
 from mne.annotations import events_from_annotations, read_annotations
 from mne.channels import DigMontage
 from mne.datasets import testing
+from mne.epochs import Epochs
 from mne.event import find_events
 from mne.io.curry import read_impedances_curry, read_montage_curry, read_raw_curry
-from mne.io.curry.curry import _check_curry_filename, _check_curry_header_filename
+from mne.io.curry.curry import (
+    RawCurry,
+    _check_curry_filename,
+    _check_curry_header_filename,
+)
 from mne.io.edf import read_raw_bdf
 from mne.io.tests.test_raw import _test_raw_reader
 from mne.transforms import Transform
@@ -30,6 +34,7 @@
 curry8_bdf_ascii_file = curry_dir / "test_bdf_stim_channel Curry 8 ASCII.cdt"
 Ref_chan_omitted_file = curry_dir / "Ref_channel_omitted Curry7.dat"
 Ref_chan_omitted_reordered_file = curry_dir / "Ref_channel_omitted reordered Curry7.dat"
+epoched_file = curry_dir / "Epoched.cdt"
 
 
 @pytest.fixture(scope="session")
@@ -72,7 +77,19 @@ def test_read_raw_curry(fname, tol, preload, bdf_curry_ref):
         bdf_curry_ref.get_data(picks=picks, start=start, stop=stop),
         rtol=tol,
     )
-    # assert raw.info["dev_head_t"] is None  # TODO do we need this value?
+    assert raw.info["dev_head_t"] == Transform("meg", "head")
+
+
+@testing.requires_testing_data
+def test_read_raw_curry_epoched():
+    """Test reading epoched file."""
+    ep = read_raw_curry(epoched_file)
+    assert isinstance(ep, Epochs)
+    assert len(ep.events) == 26
+    assert len(ep.annotations) == 0
+    raw = read_raw_curry(epoched_file, import_epochs_as_events=True)
+    assert isinstance(raw, RawCurry)
+    assert len(raw.annotations) == 26
 
 
 @testing.requires_testing_data
@@ -117,7 +134,6 @@ def test_read_events_curry_are_same_as_bdf(fname):
     """Test events from curry annotations recovers the right events."""
     EVENT_ID = {str(ii): ii for ii in range(5)}
     REF_EVENTS = find_events(read_raw_bdf(bdf_file, preload=True))
-    # TODO!
     raw = read_raw_curry(fname)
     events, _ = events_from_annotations(raw, event_id=EVENT_ID)
     assert_allclose(events, REF_EVENTS)