diff --git a/mne/datasets/config.py b/mne/datasets/config.py index 2cd937dbdee..ca3345acb78 100644 --- a/mne/datasets/config.py +++ b/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"]}' diff --git a/mne/io/ant/ant.py b/mne/io/ant/ant.py index 0a5b7865321..dd1fb79638f 100644 --- a/mne/io/ant/ant.py +++ b/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, ) diff --git a/mne/io/ant/tests/test_ant.py b/mne/io/ant/tests/test_ant.py index 170f17770c3..bf1e4556808 100644 --- a/mne/io/ant/tests/test_ant.py +++ b/mne/io/ant/tests/test_ant.py @@ -8,18 +8,20 @@ from pathlib import Path from typing import TYPE_CHECKING +import numpy as np import pytest from numpy.testing import assert_allclose from mne import Annotations from mne.datasets import testing -from mne.io import BaseRaw, read_raw_ant, read_raw_brainvision +from mne.io import BaseRaw, read_raw, read_raw_ant, read_raw_brainvision +from mne.io.ant.ant import RawANT if TYPE_CHECKING: from pathlib import Path - -data_path = testing.data_path(download=False) / "antio" / "CA_208" +pytest.importorskip("antio", minversion="0.3.0") +data_path = testing.data_path(download=False) / "antio" def read_raw_bv(fname: Path) -> BaseRaw: @@ -43,53 +45,179 @@ def read_raw_bv(fname: Path) -> BaseRaw: @pytest.fixture(scope="module") -def ca_208() -> dict[str, dict[str, Path]]: +def ca_208() -> dict[str, dict[str, Path] | str | int | dict[str, str | int]]: """Return the paths to the CA_208 dataset containing 64 channel gel recordings.""" - pytest.importorskip("antio", minversion="0.2.0") + cnt = { + "short": data_path / "CA_208" / "test_CA_208.cnt", + "amp-dc": data_path / "CA_208" / "test_CA_208_amp_disconnection.cnt", + "start-stop": data_path / "CA_208" / "test_CA_208_start_stop.cnt", + } + bv = {key: value.with_suffix(".vhdr") for key, value in cnt.items()} + return { + "cnt": cnt, + "bv": bv, + "n_eeg": 64, + "n_misc": 24, + "meas_date": "2024-08-14-10-44-47+0000", + "patient_info": { + "name": "antio test", + "his_id": "", + "birthday": "2024-08-14", + "sex": 0, + }, + "machine_info": ("eego", "EE_225", ""), + "hospital": "", + } + +@pytest.fixture(scope="module") +def andy_101() -> dict[str, dict[str, Path] | str | int | dict[str, str | int]]: + """Return the path and info to the andy_101 dataset.""" cnt = { - "short": data_path / "test_CA_208.cnt", - "amp-dc": data_path / "test_CA_208_amp_disconnection.cnt", - "start-stop": data_path / "test_CA_208_start_stop.cnt", + "short": data_path / "andy_101" / "Andy_101-raw.cnt", } bv = {key: value.with_suffix(".vhdr") for key, value in cnt.items()} - return {"cnt": cnt, "bv": bv} + return { + "cnt": cnt, + "bv": bv, + "n_eeg": 128, + "n_misc": 0, + "meas_date": "2024-08-19-16-17-07+0000", + "patient_info": { + "name": "Andy test_middle_name EEG_Exam", + "his_id": "test_subject_code", + "birthday": "2024-08-19", + "sex": 2, + }, + "machine_info": ("eego", "EE_226", ""), + "hospital": "", + } @testing.requires_testing_data -def test_io_data(ca_208: dict[str, dict[str, Path]]) -> None: +@pytest.mark.parametrize("dataset", ["ca_208", "andy_101"]) +def test_io_data(dataset, request): """Test loading of .cnt file.""" - raw_cnt = read_raw_ant(ca_208["cnt"]["short"]) - raw_bv = read_raw_bv(ca_208["bv"]["short"]) + dataset = request.getfixturevalue(dataset) + raw_cnt = read_raw_ant(dataset["cnt"]["short"]) # preload=False + raw_bv = read_raw_bv(dataset["bv"]["short"]) cnt = raw_cnt.get_data() bv = raw_bv.get_data() assert cnt.shape == bv.shape + assert_allclose(cnt, bv, atol=1e-8) + + # check preload=False and preload=False with raw.load_data() + raw_cnt.crop(0.05, 1.05) + raw_cnt2 = read_raw_ant(dataset["cnt"]["short"], preload=False) + raw_cnt2.crop(0.05, 1.05).load_data() + assert_allclose(raw_cnt.get_data(), raw_cnt2.get_data()) + + # check preload=False vs Brainvision file + raw_bv.crop(0.05, 1.05) assert_allclose(raw_cnt.get_data(), raw_bv.get_data(), atol=1e-8) + # check preload=False vs BrainVision file after dropping channels + raw_cnt.pick(raw_cnt.ch_names[::2]) + raw_bv.pick(raw_bv.ch_names[::2]) + assert_allclose(raw_cnt.get_data(), raw_bv.get_data(), atol=1e-8) + + # check after raw_cnt.load_data() + raw_cnt.load_data() + assert_allclose(raw_cnt.get_data(), raw_bv.get_data(), atol=1e-8) + + # check preload True vs False + raw_cnt = read_raw_ant(dataset["cnt"]["short"], preload=False) + raw_cnt2 = read_raw_ant(dataset["cnt"]["short"], preload=True) + bads = [raw_cnt.ch_names[idx] for idx in (1, 5, 10)] + assert_allclose( + raw_cnt.drop_channels(bads).get_data(), raw_cnt2.drop_channels(bads).get_data() + ) + raw_bv = read_raw_bv(dataset["bv"]["short"]).drop_channels(bads) + assert_allclose(raw_cnt.get_data(), raw_bv.get_data(), atol=1e-8) + assert_allclose(raw_cnt2.get_data(), raw_bv.get_data(), atol=1e-8) + @testing.requires_testing_data -def test_io_info(ca_208: dict[str, dict[str, Path]]) -> None: - """Test the info loaded from a .cnt file.""" - raw_cnt = read_raw_ant(ca_208["cnt"]["short"]) - raw_bv = read_raw_bv(ca_208["bv"]["short"]) +@pytest.mark.parametrize("dataset", ["ca_208", "andy_101"]) +def test_io_info(dataset, request): + """Test the ifo loaded from a .cnt file.""" + dataset = request.getfixturevalue(dataset) + raw_cnt = read_raw_ant(dataset["cnt"]["short"]) # preload=False + raw_bv = read_raw_bv(dataset["bv"]["short"]) assert raw_cnt.ch_names == raw_bv.ch_names assert raw_cnt.info["sfreq"] == raw_bv.info["sfreq"] - assert raw_cnt.get_channel_types() == ["eeg"] * 64 + ["misc"] * 24 + assert ( + raw_cnt.get_channel_types() + == ["eeg"] * dataset["n_eeg"] + ["misc"] * dataset["n_misc"] + ) + assert_allclose( + (raw_bv.info["meas_date"] - raw_cnt.info["meas_date"]).total_seconds(), + 0, + atol=1e-3, + ) + + +@testing.requires_testing_data +def test_io_info_parse_misc( + ca_208: dict[str, dict[str, Path] | str | int | dict[str, str | int]], +): + """Test parsing misc channels from a .cnt file.""" + raw_cnt = read_raw_ant(ca_208["cnt"]["short"]) with pytest.warns( RuntimeWarning, match="All EEG channels are not referenced to the same electrode.", ): raw_cnt = read_raw_ant(ca_208["cnt"]["short"], misc=None) + assert len(raw_cnt.ch_names) == ca_208["n_eeg"] + ca_208["n_misc"] assert raw_cnt.get_channel_types() == ["eeg"] * len(raw_cnt.ch_names) + + +@testing.requires_testing_data +def test_io_info_parse_eog( + ca_208: dict[str, dict[str, Path] | str | int | dict[str, str | int]], +): + """Test parsing EOG channels from a .cnt file.""" raw_cnt = read_raw_ant(ca_208["cnt"]["short"], eog="EOG") + assert len(raw_cnt.ch_names) == ca_208["n_eeg"] + ca_208["n_misc"] idx = raw_cnt.ch_names.index("EOG") - ch_types = ["eeg"] * 64 + ["misc"] * 24 + ch_types = ["eeg"] * ca_208["n_eeg"] + ["misc"] * ca_208["n_misc"] ch_types[idx] = "eog" assert raw_cnt.get_channel_types() == ch_types @testing.requires_testing_data -def test_io_amp_disconnection(ca_208: dict[str, dict[str, Path]]) -> None: +@pytest.mark.parametrize("dataset", ["andy_101", "ca_208"]) +def test_subject_info(dataset, request): + """Test reading the subject info.""" + dataset = request.getfixturevalue(dataset) + raw_cnt = read_raw_ant(dataset["cnt"]["short"]) + subject_info = raw_cnt.info["subject_info"] + assert subject_info["his_id"] == dataset["patient_info"]["his_id"] + assert subject_info["first_name"] == dataset["patient_info"]["name"] + assert subject_info["sex"] == dataset["patient_info"]["sex"] + assert ( + subject_info["birthday"].strftime("%Y-%m-%d") + == dataset["patient_info"]["birthday"] + ) + + +@testing.requires_testing_data +@pytest.mark.parametrize("dataset", ["andy_101", "ca_208"]) +def test_machine_info(dataset, request): + """Test reading the machine info.""" + dataset = request.getfixturevalue(dataset) + raw_cnt = read_raw_ant(dataset["cnt"]["short"]) + device_info = raw_cnt.info["device_info"] + make, model, serial = dataset["machine_info"] + assert device_info["type"] == make + assert device_info["model"] == model + assert device_info["serial"] == serial + + +@testing.requires_testing_data +def test_io_amp_disconnection( + ca_208: dict[str, dict[str, Path] | str | int | dict[str, str | int]], +): """Test loading of .cnt file with amplifier disconnection.""" raw_cnt = read_raw_ant(ca_208["cnt"]["amp-dc"]) raw_bv = read_raw_bv(ca_208["bv"]["amp-dc"]) @@ -121,7 +249,10 @@ def test_io_amp_disconnection(ca_208: dict[str, dict[str, Path]]) -> None: @testing.requires_testing_data @pytest.mark.parametrize("description", ["impedance", "test"]) -def test_io_impedance(ca_208: dict[str, dict[str, Path]], description: str) -> None: +def test_io_impedance( + ca_208: dict[str, dict[str, Path] | str | int | dict[str, str | int]], + description: str, +): """Test loading of impedances from a .cnt file.""" raw_cnt = read_raw_ant(ca_208["cnt"]["amp-dc"], impedance_annotation=description) assert isinstance(raw_cnt.impedances, list) @@ -136,8 +267,44 @@ def test_io_impedance(ca_208: dict[str, dict[str, Path]], description: str) -> N @testing.requires_testing_data -def test_io_segments(ca_208: dict[str, dict[str, Path]]) -> None: +def test_io_segments( + ca_208: dict[str, dict[str, Path] | str | int | dict[str, str | int]], +): """Test reading a .cnt file with segents (start/stop).""" raw_cnt = read_raw_ant(ca_208["cnt"]["start-stop"]) raw_bv = read_raw_bv(ca_208["bv"]["start-stop"]) assert_allclose(raw_cnt.get_data(), raw_bv.get_data(), atol=1e-8) + + +@testing.requires_testing_data +def test_annotations_and_preload( + ca_208: dict[str, dict[str, Path] | str | int | dict[str, str | int]], +): + """Test annotation loading with preload True/False.""" + raw_cnt_preloaded = read_raw_ant(ca_208["cnt"]["short"], preload=True) + assert len(raw_cnt_preloaded.annotations) == 2 # impedance measurements, start/end + raw_cnt = read_raw_ant(ca_208["cnt"]["short"], preload=False) + assert len(raw_cnt.annotations) == 2 + raw_cnt.crop(2, 3) + assert len(raw_cnt.annotations) == 0 + raw_cnt.load_data() + assert len(raw_cnt.annotations) == 0 + + raw_cnt_preloaded = read_raw_ant(ca_208["cnt"]["amp-dc"], preload=True) + assert len(raw_cnt_preloaded.annotations) == 5 # 4 impedances, 1 disconnection + raw_cnt = read_raw_ant(ca_208["cnt"]["amp-dc"], preload=False) + assert len(raw_cnt.annotations) == 5 + idx = np.where(raw_cnt.annotations.description == "BAD_disconnection")[0] + onset = raw_cnt.annotations.onset[idx][0] + raw_cnt.crop(0, onset - 1) + assert len(raw_cnt.annotations) == 1 # initial impedance measurement + assert raw_cnt.annotations.description[0] == "impedance" + + +@testing.requires_testing_data +def test_read_raw( + ca_208: dict[str, dict[str, Path] | str | int | dict[str, str | int]], +): + """Test loading through read_raw.""" + raw = read_raw(ca_208["cnt"]["short"]) + assert isinstance(raw, RawANT) diff --git a/tools/vulture_allowlist.py b/tools/vulture_allowlist.py index d612d0ec5ed..d06eac34285 100644 --- a/tools/vulture_allowlist.py +++ b/tools/vulture_allowlist.py @@ -73,6 +73,9 @@ _notebook_vtk_works _.drop_inds_ +# mne/io/ant/tests/test_ant.py +andy_101 + # mne/io/snirf/tests/test_snirf.py _.dataTimeSeries _.sourceIndex