Improves functionality of ANT Neuro reader (#12810)

Co-authored-by: Mathieu Scheltienne <mathieu.scheltienne@gmail.com>
Co-authored-by: Eric Larson <larson.eric.d@gmail.com>

mne/datasets/config.py

@@ -87,7 +87,7 @@
 # update the checksum in the MNE_DATASETS dict below, and change version
 # here: ↓↓↓↓↓↓↓↓
 RELEASES = dict(
-    testing="0.154",
+    testing="0.155",
     misc="0.27",
     phantom_kit="0.2",
     ucl_opm_auditory="0.2",
@@ -115,7 +115,7 @@
 # Testing and misc are at the top as they're updated most often
 MNE_DATASETS["testing"] = dict(
     archive_name=f"{TESTING_VERSIONED}.tar.gz",
-    hash="md5:c079431e69ea1c30462a7c5fb4b5d8e0",
+    hash="md5:a3ddc1fbfd48830207112db13c3fdd6a",
     url=(
         "https://codeload.github.com/mne-tools/mne-testing-data/"
         f'tar.gz/{RELEASES["testing"]}'

mne/io/ant/ant.py

@@ -80,6 +80,7 @@ class RawANT(BaseRaw):
             Note that the impedance annotation will likely have a duration of ``0``.
             If the measurement marks a discontinuity, the duration should be modified to
             cover the discontinuity in its entirety.
+    %(preload)s
     %(verbose)s
     """
 
@@ -94,6 +95,7 @@ def __init__(
         bipolars: list[str] | tuple[str, ...] | None,
         impedance_annotation: str,
         *,
+        preload: bool | NDArray,
         verbose=None,
     ) -> None:
         logger.info("Reading ANT file %s", fname)
@@ -102,10 +104,16 @@ def __init__(
                 "Missing optional dependency 'antio'. Use pip or conda to install "
                 "'antio'."
             )
-        check_version("antio", "0.2.0")
+        check_version("antio", "0.3.0")
 
         from antio import read_cnt
-        from antio.parser import read_data, read_info, read_triggers
+        from antio.parser import (
+            read_device_info,
+            read_info,
+            read_meas_date,
+            read_subject_info,
+            read_triggers,
+        )
 
         fname = _check_fname(fname, overwrite="read", must_exist=True, name="fname")
         _validate_type(eog, (str, None), "eog")
@@ -132,14 +140,34 @@ def __init__(
         info = create_info(
             ch_names, sfreq=cnt.get_sample_frequency(), ch_types=ch_types
         )
+        info.set_meas_date(read_meas_date(cnt))
+        make, model, serial, site = read_device_info(cnt)
+        info["device_info"] = dict(type=make, model=model, serial=serial, site=site)
+        his_id, name, sex, birthday = read_subject_info(cnt)
+        info["subject_info"] = dict(
+            his_id=his_id, first_name=name, sex=sex, birthday=birthday
+        )
         if bipolars is not None:
             with info._unlock():
                 for idx in bipolars_idx:
                     info["chs"][idx]["coil_type"] = FIFF.FIFFV_COIL_EEG_BIPOLAR
-        # read and scale data array
-        data = read_data(cnt)
-        _scale_data(data, ch_units)
-        super().__init__(info, preload=data, filenames=[fname], verbose=verbose)
+        first_samps = np.array((0,))
+        last_samps = (cnt.get_sample_count() - 1,)
+        raw_extras = {
+            "orig_nchan": cnt.get_channel_count(),
+            "orig_ch_units": ch_units,
+            "first_samples": np.array(first_samps),
+            "last_samples": np.array(last_samps),
+        }
+        super().__init__(
+            info,
+            preload=preload,
+            first_samps=first_samps,
+            last_samps=last_samps,
+            filenames=[fname],
+            verbose=verbose,
+            raw_extras=[raw_extras],
+        )
         # look for annotations (called trigger by ant)
         onsets, durations, descriptions, impedances, disconnect = read_triggers(cnt)
         onsets, durations, descriptions = _prepare_annotations(
@@ -159,6 +187,27 @@ def impedances(self) -> list[dict[str, float]]:
         """List of impedance measurements."""
         return self._impedances
 
+    def _read_segment_file(self, data, idx, fi, start, stop, cals, mult):
+        from antio import read_cnt
+        from antio.parser import read_data
+
+        ch_units = self._raw_extras[0]["orig_ch_units"]
+        first_samples = self._raw_extras[0]["first_samples"]
+        n_times = self._raw_extras[0]["last_samples"] + 1
+        for first_samp, this_n_times in zip(first_samples, n_times):
+            i_start = max(start, first_samp)
+            i_stop = min(stop, this_n_times + first_samp)
+            # read and scale data array
+            cnt = read_cnt(str(self._filenames[fi]))
+            one = read_data(cnt, i_start, i_stop)
+            _scale_data(one, ch_units)
+            data_view = data[:, i_start - start : i_stop - start]
+            if isinstance(idx, slice):
+                data_view[:] = one[idx]
+            else:
+                # faster than doing one = one[idx]
+                np.take(one, idx, axis=0, out=data_view)
+
 
 def _handle_bipolar_channels(
     ch_names: list[str], ch_refs: list[str], bipolars: list[str] | tuple[str, ...]
@@ -273,6 +322,7 @@ def read_raw_ant(
     bipolars=None,
     impedance_annotation="impedance",
     *,
+    preload=False,
     verbose=None,
 ) -> RawANT:
     """
@@ -289,5 +339,6 @@ def read_raw_ant(
         misc=misc,
         bipolars=bipolars,
         impedance_annotation=impedance_annotation,
+        preload=preload,
         verbose=verbose,
     )