"""Test reading of BrainVision format."""

# Authors: The MNE-Python contributors.
# License: BSD-3-Clause
# Copyright the MNE-Python contributors.

import datetime
import re
import shutil
from pathlib import Path

import numpy as np
import pytest
from numpy.testing import assert_allclose, assert_array_equal

from mne import concatenate_raws, pick_types, read_annotations
from mne._fiff.constants import FIFF
from mne.annotations import events_from_annotations
from mne.datasets import testing
from mne.io import read_raw_brainvision, read_raw_fif
from mne.io.tests.test_raw import _test_raw_reader
from mne.utils import _record_warnings, _stamp_to_dt, object_diff

data_dir = Path(__file__).parent / "data"
vhdr_path = data_dir / "test.vhdr"
vmrk_path = data_dir / "test.vmrk"
eeg_path = data_dir / "test.eeg"

vhdr_partially_disabled_hw_filter_path = (
    data_dir / "test_partially_disabled_hw_filter.vhdr"
)

vhdr_old_path = data_dir / "test_old_layout_latin1_software_filter.vhdr"
vhdr_old_longname_path = (
    data_dir / "test_old_layout_latin1_software_filter_longname.vhdr"
)
vhdr_v2_path = data_dir / "testv2.vhdr"

vhdr_highpass_path = data_dir / "test_highpass.vhdr"
vhdr_mixed_highpass_path = data_dir / "test_mixed_highpass.vhdr"
vhdr_highpass_hz_path = data_dir / "test_highpass_hz.vhdr"
vhdr_mixed_highpass_hz_path = data_dir / "test_mixed_highpass_hz.vhdr"

# Not a typo: we can reuse the highpass file for the lowpass (Hz) test
vhdr_lowpass_path = data_dir / "test_highpass.vhdr"
vhdr_mixed_lowpass_path = data_dir / "test_mixed_lowpass.vhdr"
vhdr_lowpass_s_path = data_dir / "test_lowpass_s.vhdr"
vhdr_mixed_lowpass_s_path = data_dir / "test_mixed_lowpass_s.vhdr"

# VHDR exported with neuroone
data_path = testing.data_path(download=False)
neuroone_vhdr = data_path / "Brainvision" / "test_NO.vhdr"

# AHDR exported with VAmp
data_path = testing.data_path(download=False)
vamp_ahdr = data_path / "Brainvision" / "test_VAmp.ahdr"

# Test for nanovolts as unit
vhdr_units_path = data_dir / "test_units.vhdr"

eeg_bin = data_dir / "test_bin_raw.fif"
eog = ["HL", "HR", "Vb"]

# XXX: BUG we cannot parse test.hpts FastSCAN file to create a DigMontage
#      (plus I've removed montage from all the read_raw_brainvision and nothing
#       broke, so we were not testing that set_montage in brainvision was
#       working)
#      This should be amend in its own PR.
montage = data_dir / "test.hpts"

_no_dig = pytest.warns(RuntimeWarning, match="No info on DataPoints")


def test_orig_units(recwarn):
    """Test exposure of original channel units."""
    raw = read_raw_brainvision(vhdr_path)
    orig_units = raw._orig_units
    assert len(orig_units) == 32
    assert orig_units["FP1"] == "µV"

    # no unit specified in the vhdr, ensure we default to µV here
    assert orig_units["FP2"] == "µV"
    assert orig_units["F3"] == "µV"

    sum([v == "µV" for v in orig_units.values()]) == 26

    assert orig_units["CP5"] == "n/a"  # originally BS, not a valid unit
    assert orig_units["CP6"] == "µS"
    assert orig_units["HL"] == "n/a"  # originally ARU, not a valid unit
    assert orig_units["HR"] == "n/a"  # originally uS ...
    assert orig_units["Vb"] == "S"
    assert orig_units["ReRef"] == "C"


DATE_TEST_CASES = np.array(
    [
        (
            "Mk1=New Segment,,1,1,0,20131113161403794232\n",  # content
            [1384359243, 794232],  # meas_date internal representation
            "2013-11-13 16:14:03 UTC",
        ),  # meas_date representation
        (
            (
                "Mk1=New Segment,,1,1,0,20070716122240937454\n"
                "Mk2=New Segment,,2,1,0,20070716122240937455\n"
            ),
            [1184588560, 937454],
            "2007-07-16 12:22:40 UTC",
        ),
        (
            "Mk1=New Segment,,1,1,0,\nMk2=New Segment,,2,1,0,20070716122240937454\n",
            [1184588560, 937454],
            "2007-07-16 12:22:40 UTC",
        ),
        ("Mk1=STATUS,,1,1,0\n", None, "unspecified"),
        ("Mk1=New Segment,,1,1,0,\n", None, "unspecified"),
        ("Mk1=New Segment,,1,1,0\n", None, "unspecified"),
        ("Mk1=New Segment,,1,1,0,00000000000304125000", None, "unspecified"),
    ],
    dtype=np.dtype(
        {
            "names": ["content", "meas_date", "meas_date_repr"],
            "formats": [object, object, "U22"],
        }
    ),
)


@pytest.fixture(scope="session")
def _mocked_meas_date_data(tmp_path_factory):
    """Prepare files for mocked_meas_date_file fixture."""
    # Prepare the files
    tmp_path = tmp_path_factory.mktemp("brainvision_mocked_meas_date")
    vhdr_fname, vmrk_fname, eeg_fname = (
        tmp_path / ff.name for ff in [vhdr_path, vmrk_path, eeg_path]
    )
    for orig, dest in zip([vhdr_path, eeg_path], [vhdr_fname, eeg_fname]):
        shutil.copyfile(orig, dest)

    # Get the marker information
    with open(vmrk_path) as fin:
        lines = fin.readlines()

    return vhdr_fname, vmrk_fname, lines


@pytest.fixture(scope="session", params=[tt for tt in DATE_TEST_CASES])
def mocked_meas_date_file(_mocked_meas_date_data, request):
    """Prepare a generator for use in test_meas_date."""
    MEAS_DATE_LINE = 11  # see test.vmrk file
    vhdr_fname, vmrk_fname, lines = _mocked_meas_date_data

    lines[MEAS_DATE_LINE] = request.param["content"]
    with open(vmrk_fname, "w") as fout:
        fout.writelines(lines)
    meas_date = request.param["meas_date"]
    if meas_date is not None:
        meas_date = _stamp_to_dt(meas_date)

    yield vhdr_fname, meas_date, request.param["meas_date_repr"]


def test_meas_date(mocked_meas_date_file):
    """Test successful extraction of measurement date."""
    vhdr_f, expected_meas, expected_meas_repr = mocked_meas_date_file
    raw = read_raw_brainvision(vhdr_f)
    assert expected_meas_repr in repr(raw.info)
    if expected_meas is None:
        assert raw.info["meas_date"] is None
    else:
        assert raw.info["meas_date"] == expected_meas


def test_vhdr_codepage_ansi(tmp_path):
    """Test BV reading with ANSI codepage."""
    raw_init = read_raw_brainvision(vhdr_path)
    data_expected, times_expected = raw_init[:]
    ansi_vhdr_path = tmp_path / vhdr_path.name
    ansi_vmrk_path = tmp_path / vmrk_path.name
    ansi_eeg_path = tmp_path / eeg_path.name
    # copy data file
    shutil.copy(eeg_path, ansi_eeg_path)
    # modify header file
    with open(ansi_vhdr_path, "wb") as fout:
        with open(vhdr_path, "rb") as fin:
            for line in fin:
                # Common Infos section
                if line.startswith(b"Codepage"):
                    line = b"Codepage=ANSI\n"
                fout.write(line)
    # modify marker file
    with open(ansi_vmrk_path, "wb") as fout:
        with open(vmrk_path, "rb") as fin:
            for line in fin:
                # Common Infos section
                if line.startswith(b"Codepage"):
                    line = b"Codepage=ANSI\n"
                fout.write(line)

    raw = read_raw_brainvision(ansi_vhdr_path)
    data_new, times_new = raw[:]

    assert raw_init.ch_names == raw.ch_names
    assert_allclose(data_new, data_expected, atol=1e-15)
    assert_allclose(times_new, times_expected, atol=1e-15)


@pytest.mark.parametrize(
    "header",
    [
        b"BrainVision Data Exchange %s File Version 1.0\n",
        # 2.0, space, core, comma
        b"Brain Vision Core Data Exchange %s File, Version 2.0\n",
        # unsupported version
        b"Brain Vision Core Data Exchange %s File, Version 3.0\n",
        # missing header
        b"\n",
    ],
)
def test_vhdr_versions(tmp_path, header):
    """Test BV reading with different header variants."""
    raw_init = read_raw_brainvision(vhdr_path)
    data_expected, times_expected = raw_init[:]
    use_vhdr_path = tmp_path / vhdr_path.name
    use_vmrk_path = tmp_path / vmrk_path.name
    use_eeg_path = tmp_path / eeg_path.name
    shutil.copy(eeg_path, use_eeg_path)
    with open(use_vhdr_path, "wb") as fout:
        with open(vhdr_path, "rb") as fin:
            for line in fin:
                # Common Infos section
                if line.startswith(b"Brain"):
                    if header != b"\n":
                        line = header % b"Header"
                    else:
                        line = header
                fout.write(line)
    with open(use_vmrk_path, "wb") as fout:
        with open(vmrk_path, "rb") as fin:
            for line in fin:
                # Common Infos section
                if line.startswith(b"Brain"):
                    if header != b"\n":
                        line = header % b"Marker"
                    else:
                        line = header
                fout.write(line)

    if b"3.0" in header:  # unsupported version
        with pytest.warns(RuntimeWarning, match=r"3\.0.*Contact MNE-Python"):
            read_raw_brainvision(use_vhdr_path)
        return
    elif header == b"\n":  # no version header
        with pytest.warns(RuntimeWarning, match="Missing header"):
            read_raw_brainvision(use_vhdr_path)
        return
    else:
        raw = read_raw_brainvision(use_vhdr_path)
        data_new, _ = raw[:]
        assert_allclose(data_new, data_expected, atol=1e-15)


@pytest.mark.parametrize("data_sep", (b" ", b",", b"+"))
def test_ascii(tmp_path, data_sep):
    """Test ASCII BV reading."""
    raw = read_raw_brainvision(vhdr_path)
    ascii_vhdr_path = tmp_path / vhdr_path.name
    # copy marker file
    shutil.copy(
        vhdr_path.with_suffix(".vmrk"),
        ascii_vhdr_path.with_suffix(".vmrk"),
    )
    # modify header file
    skipping = False
    with open(ascii_vhdr_path, "wb") as fout:
        with open(vhdr_path, "rb") as fin:
            for line in fin:
                # Common Infos section
                if line.startswith(b"DataFormat"):
                    line = b"DataFormat=ASCII\n"
                elif line.startswith(b"DataFile="):
                    line = b"DataFile=test.dat\n"
                # Replace the "'Binary Infos'" section
                elif line.startswith(b"[Binary Infos]"):
                    skipping = True
                    fout.write(
                        b"[ASCII Infos]\nDecimalSymbol=.\nSkipLines=1\n"
                        b"SkipColumns=0\n\n"
                    )
                elif skipping and line.startswith(b"["):
                    skipping = False
                if not skipping:
                    fout.write(line)
    # create the .dat file
    data, times = raw[:]
    with open(ascii_vhdr_path.with_suffix(".dat"), "wb") as fid:
        fid.write(
            data_sep.join(ch_name.encode("ASCII") for ch_name in raw.ch_names) + b"\n"
        )
        fid.write(
            b"\n".join(b" ".join(b"%.3f" % dd for dd in d) for d in data.T / raw._cals)
        )

    if data_sep == b";":
        with pytest.raises(RuntimeError, match="Unknown.*data format"):
            read_raw_brainvision(ascii_vhdr_path)
        return

    raw = read_raw_brainvision(ascii_vhdr_path)
    assert isinstance(raw.orig_format, str)
    data_new, times_new = raw[:]
    assert_allclose(data_new, data, atol=1e-15)
    assert_allclose(times_new, times)


def test_ch_names_comma(tmp_path):
    """Test that channel names containing commas are properly read."""
    # commas in BV are encoded as \1
    replace_dict = {
        r"^Ch4=F4,": r"Ch4=F4\\1foo,",
        r"^4\s\s\s\s\sF4": "4     F4,foo ",
    }

    # Copy existing vhdr file to tmp_path and manipulate to contain
    # a channel with comma
    for src, dest in zip(
        (vhdr_path, vmrk_path, eeg_path), ("test.vhdr", "test.vmrk", "test.eeg")
    ):
        shutil.copyfile(src, tmp_path / dest)

    comma_vhdr = tmp_path / "test.vhdr"
    with open(comma_vhdr) as fin:
        lines = fin.readlines()

    new_lines = []
    nperformed_replacements = 0
    for line in lines:
        for to_replace, replacement in replace_dict.items():
            match = re.search(to_replace, line)
            if match is not None:
                new = re.sub(to_replace, replacement, line)
                new_lines.append(new)
                nperformed_replacements += 1
                break
        else:
            new_lines.append(line)
    assert nperformed_replacements == len(replace_dict)

    with open(comma_vhdr, "w") as fout:
        fout.writelines(new_lines)

    # Read the line containing a "comma channel name"
    raw = read_raw_brainvision(comma_vhdr)
    assert "F4,foo" in raw.ch_names


@pytest.mark.filterwarnings("ignore:.*different.*:RuntimeWarning")
def test_brainvision_data_highpass_filters():
    """Test reading raw Brain Vision files with amplifier filter settings."""
    # Homogeneous highpass in seconds (default measurement unit)
    raw = _test_raw_reader(read_raw_brainvision, vhdr_fname=vhdr_highpass_path, eog=eog)

    assert raw.info["highpass"] == 1.0 / (2 * np.pi * 10)
    assert raw.info["lowpass"] == 250.0

    # Heterogeneous highpass in seconds (default measurement unit)
    with pytest.warns(RuntimeWarning, match="different .*pass filters") as w:
        raw = _test_raw_reader(
            read_raw_brainvision, vhdr_fname=vhdr_mixed_highpass_path, eog=eog
        )

    w = [str(ww.message) for ww in w]
    assert not any("different lowpass filters" in ww for ww in w), w
    assert any("different highpass filters" in ww for ww in w), w

    assert raw.info["highpass"] == 1.0 / (2 * np.pi * 10)
    assert raw.info["lowpass"] == 250.0

    # Homogeneous highpass in Hertz
    raw = _test_raw_reader(
        read_raw_brainvision, vhdr_fname=vhdr_highpass_hz_path, eog=eog
    )

    assert raw.info["highpass"] == 10.0
    assert raw.info["lowpass"] == 250.0

    # Heterogeneous highpass in Hertz
    with pytest.warns(RuntimeWarning, match="different .*pass filters") as w:
        raw = _test_raw_reader(
            read_raw_brainvision, vhdr_fname=vhdr_mixed_highpass_hz_path, eog=eog
        )

    w = [str(ww.message) for ww in w]
    assert not any("will be dropped" in ww for ww in w), w
    assert not any("different lowpass filters" in ww for ww in w), w
    assert any("different highpass filters" in ww for ww in w), w

    assert raw.info["highpass"] == 5.0
    assert raw.info["lowpass"] == 250.0


def test_brainvision_data_lowpass_filters():
    """Test files with amplifier LP filter settings."""
    # Homogeneous lowpass in Hertz (default measurement unit)
    raw = _test_raw_reader(read_raw_brainvision, vhdr_fname=vhdr_lowpass_path, eog=eog)

    assert raw.info["highpass"] == 1.0 / (2 * np.pi * 10)
    assert raw.info["lowpass"] == 250.0

    # Heterogeneous lowpass in Hertz (default measurement unit)
    with pytest.warns(RuntimeWarning) as w:  # event parsing
        raw = _test_raw_reader(
            read_raw_brainvision, vhdr_fname=vhdr_mixed_lowpass_path, eog=eog
        )

    lowpass_warning = ["different lowpass filters" in str(ww.message) for ww in w]
    highpass_warning = ["different highpass filters" in str(ww.message) for ww in w]

    expected_warnings = zip(lowpass_warning, highpass_warning)

    assert any(any([lp, hp]) for lp, hp in expected_warnings)

    assert raw.info["highpass"] == 1.0 / (2 * np.pi * 10)
    assert raw.info["lowpass"] == 250.0

    # Homogeneous lowpass in seconds
    raw = _test_raw_reader(
        read_raw_brainvision, vhdr_fname=vhdr_lowpass_s_path, eog=eog
    )

    assert raw.info["highpass"] == 1.0 / (2 * np.pi * 10)
    assert raw.info["lowpass"] == 1.0 / (2 * np.pi * 0.004)

    # Heterogeneous lowpass in seconds
    with pytest.warns(RuntimeWarning) as w:  # filter settings
        raw = _test_raw_reader(
            read_raw_brainvision, vhdr_fname=vhdr_mixed_lowpass_s_path, eog=eog
        )

    lowpass_warning = ["different lowpass filters" in str(ww.message) for ww in w]
    highpass_warning = ["different highpass filters" in str(ww.message) for ww in w]

    expected_warnings = zip(lowpass_warning, highpass_warning)

    assert any(any([lp, hp]) for lp, hp in expected_warnings)

    assert raw.info["highpass"] == 1.0 / (2 * np.pi * 10)
    assert raw.info["lowpass"] == 1.0 / (2 * np.pi * 0.004)


def test_brainvision_data_partially_disabled_hw_filters():
    """Test heterogeneous filter settings including non-numeric values."""
    with pytest.warns(RuntimeWarning) as w:  # event parsing
        raw = _test_raw_reader(
            read_raw_brainvision,
            vhdr_fname=vhdr_partially_disabled_hw_filter_path,
            eog=eog,
        )

    trigger_warning = ["will be dropped" in str(ww.message) for ww in w]
    lowpass_warning = ["different lowpass filters" in str(ww.message) for ww in w]
    highpass_warning = ["different highpass filters" in str(ww.message) for ww in w]

    expected_warnings = zip(trigger_warning, lowpass_warning, highpass_warning)

    assert any(any([trg, lp, hp]) for trg, lp, hp in expected_warnings)

    assert raw.info["highpass"] == 0.0
    assert raw.info["lowpass"] == 500.0


def test_brainvision_data_software_filters_latin1_global_units():
    """Test reading raw Brain Vision files."""
    with _no_dig, pytest.warns(RuntimeWarning, match="software filter"):
        raw = _test_raw_reader(
            read_raw_brainvision,
            vhdr_fname=vhdr_old_path,
            eog=("VEOGo", "VEOGu", "HEOGli", "HEOGre"),
            misc=("A2",),
        )

    assert raw.info["highpass"] == 1.0 / (2 * np.pi * 0.9)
    assert raw.info["lowpass"] == 50.0

    # test sensor name with spaces (#9299)
    with _no_dig, pytest.warns(RuntimeWarning, match="software filter"):
        raw = _test_raw_reader(
            read_raw_brainvision,
            vhdr_fname=vhdr_old_longname_path,
            eog=("VEOGo", "VEOGu", "HEOGli", "HEOGre"),
            misc=("A2",),
        )

    assert raw.info["highpass"] == 1.0 / (2 * np.pi * 0.9)
    assert raw.info["lowpass"] == 50.0


def test_brainvision_data():
    """Test reading raw Brain Vision files."""
    pytest.raises(OSError, read_raw_brainvision, vmrk_path)
    pytest.raises(
        ValueError, read_raw_brainvision, vhdr_path, preload=True, scale="foo"
    )

    raw_py = _test_raw_reader(
        read_raw_brainvision, vhdr_fname=vhdr_path, eog=eog, misc="auto"
    )

    assert "RawBrainVision" in repr(raw_py)

    assert raw_py.info["highpass"] == 0.0
    assert raw_py.info["lowpass"] == 250.0

    picks = pick_types(raw_py.info, meg=False, eeg=True, exclude="bads")
    data_py, times_py = raw_py[picks]

    # compare with a file that was generated using MNE-C
    raw_bin = read_raw_fif(eeg_bin, preload=True)
    picks = pick_types(raw_py.info, meg=False, eeg=True, exclude="bads")
    data_bin, times_bin = raw_bin[picks]

    assert_allclose(data_py, data_bin)
    assert_allclose(times_py, times_bin)

    # Make sure EOG channels are marked correctly
    for ch in raw_py.info["chs"]:
        if ch["ch_name"] in eog:
            assert ch["kind"] == FIFF.FIFFV_EOG_CH
        elif ch["ch_name"] == "STI 014":
            assert ch["kind"] == FIFF.FIFFV_STIM_CH
        elif ch["ch_name"] in ("CP5", "CP6"):
            assert ch["kind"] == FIFF.FIFFV_MISC_CH
            assert ch["unit"] == FIFF.FIFF_UNIT_NONE
        elif ch["ch_name"] == "ReRef":
            assert ch["kind"] == FIFF.FIFFV_MISC_CH
            assert ch["unit"] == FIFF.FIFF_UNIT_CEL
        else:
            assert ch["ch_name"] in raw_py.info["ch_names"], f"Unknown: {ch['ch_name']}"
            assert ch["kind"] == FIFF.FIFFV_EEG_CH
            assert ch["unit"] == FIFF.FIFF_UNIT_V

    # test loading v2
    read_raw_brainvision(vhdr_v2_path, eog=eog, preload=True, verbose="error")
    # test different units with alternative header file
    raw_units = _test_raw_reader(
        read_raw_brainvision, vhdr_fname=vhdr_units_path, eog=eog, misc="auto"
    )
    assert raw_units.info["chs"][0]["ch_name"] == "FP1"
    assert raw_units.info["chs"][0]["kind"] == FIFF.FIFFV_EEG_CH
    data_units, _ = raw_units[0]
    assert_allclose(data_py[0, :], data_units.squeeze())

    assert raw_units.info["chs"][1]["ch_name"] == "FP2"
    assert raw_units.info["chs"][1]["kind"] == FIFF.FIFFV_EEG_CH
    data_units, _ = raw_units[1]
    assert_allclose(data_py[1, :], data_units.squeeze())

    assert raw_units.info["chs"][2]["ch_name"] == "F3"
    assert raw_units.info["chs"][2]["kind"] == FIFF.FIFFV_EEG_CH
    data_units, _ = raw_units[2]
    assert_allclose(data_py[2, :], data_units.squeeze())


def test_brainvision_vectorized_data():
    """Test reading BrainVision data files with vectorized data."""
    with _no_dig, pytest.warns(RuntimeWarning, match="software filter"):
        raw = read_raw_brainvision(vhdr_old_path, preload=True)

    assert_array_equal(raw._data.shape, (29, 251))

    first_two_samples_all_chs = np.array(
        [
            [+5.22000008e-06, +5.10000000e-06],
            [+2.10000000e-06, +2.27000008e-06],
            [+1.15000000e-06, +1.33000002e-06],
            [+4.00000000e-07, +4.00000000e-07],
            [-3.02999992e-06, -2.82000008e-06],
            [+2.71000004e-06, +2.45000000e-06],
            [+2.41000004e-06, +2.36000004e-06],
            [+1.01999998e-06, +1.18000002e-06],
            [-1.33999996e-06, -1.25000000e-06],
            [-2.60000000e-06, -2.46000004e-06],
            [+6.80000019e-07, +8.00000000e-07],
            [+1.48000002e-06, +1.48999996e-06],
            [+1.61000004e-06, +1.51000004e-06],
            [+7.19999981e-07, +8.60000038e-07],
            [-3.00000000e-07, -4.00000006e-08],
            [-1.20000005e-07, +6.00000024e-08],
            [+8.19999981e-07, +9.89999962e-07],
            [+1.13000002e-06, +1.28000002e-06],
            [+1.08000002e-06, +1.33999996e-06],
            [+2.20000005e-07, +5.69999981e-07],
            [-4.09999990e-07, +4.00000006e-08],
            [+5.19999981e-07, +9.39999962e-07],
            [+1.01000004e-06, +1.51999998e-06],
            [+1.01000004e-06, +1.55000000e-06],
            [-1.43000002e-06, -1.13999996e-06],
            [+3.65000000e-06, +3.65999985e-06],
            [+4.15999985e-06, +3.79000015e-06],
            [+9.26999969e-06, +8.95999985e-06],
            [-7.35999985e-06, -7.18000031e-06],
        ]
    )

    assert_allclose(raw._data[:, :2], first_two_samples_all_chs)


def test_coodinates_extraction():
    """Test reading of [Coordinates] section if present."""
    # vhdr 2 has a Coordinates section
    with (
        _record_warnings(),
        pytest.warns(RuntimeWarning, match="coordinate information"),
    ):
        raw = read_raw_brainvision(vhdr_v2_path)

    # Basic check of extracted coordinates
    assert raw.info["dig"] is not None
    diglist = raw.info["dig"]
    coords = np.array([dig["r"] for dig in diglist])
    EXPECTED_SHAPE = (
        # HL, HR, Vb, ReRef are not set in dig
        # but LPA, Nasion, RPA are estimated
        len(raw.ch_names) - 4 + 3,
        3,
    )
    assert coords.shape == EXPECTED_SHAPE

    # Make sure the scaling seems right
    # a coordinate more than 20cm away from origin is implausible
    assert coords.max() < 0.2

    # vhdr 1 does not have a Coordinates section
    raw2 = read_raw_brainvision(vhdr_path)
    assert raw2.info["dig"] is None


@testing.requires_testing_data
def test_brainvision_neuroone_export():
    """Test Brainvision file exported with neuroone system."""
    raw = read_raw_brainvision(neuroone_vhdr, verbose="error")
    assert raw.info["meas_date"] is None
    assert len(raw.info["chs"]) == 65
    assert raw.info["sfreq"] == 5000.0


@testing.requires_testing_data
def test_read_vmrk_annotations(tmp_path):
    """Test load brainvision annotations."""
    sfreq = 1000.0

    # Test vmrk file without annotations
    # delete=False is for Windows compatibility
    with open(vmrk_path) as myfile:
        head = [next(myfile) for x in range(6)]
    fname = tmp_path / "temp.vmrk"
    with open(str(fname), "w") as temp:
        for item in head:
            temp.write(item)
    read_annotations(fname, sfreq=sfreq)


def test_ignore_marker_types():
    """Test ignore marker types."""
    # default behavior (do not ignore marker types)
    raw = read_raw_brainvision(vhdr_path)
    expected_descriptions = [
        "New Segment/",
        "Stimulus/S253",
        "Stimulus/S255",
        "Event/254",
        "Stimulus/S255",
        "Event/254",
        "Stimulus/S255",
        "Stimulus/S253",
        "Stimulus/S255",
        "Response/R255",
        "Event/254",
        "Stimulus/S255",
        "SyncStatus/Sync On",
        "Optic/O  1",
    ]
    assert_array_equal(raw.annotations.description, expected_descriptions)

    # ignore marker types
    raw = read_raw_brainvision(vhdr_path, ignore_marker_types=True)
    expected_descriptions = [
        "",
        "S253",
        "S255",
        "254",
        "S255",
        "254",
        "S255",
        "S253",
        "S255",
        "R255",
        "254",
        "S255",
        "Sync On",
        "O  1",
    ]
    assert_array_equal(raw.annotations.description, expected_descriptions)


@testing.requires_testing_data
def test_read_vhdr_annotations_and_events(tmp_path):
    """Test load brainvision annotations and parse them to events."""
    # First we add a custom event that contains a comma in its description
    for src, dest in zip(
        (vhdr_path, vmrk_path, eeg_path), ("test.vhdr", "test.vmrk", "test.eeg")
    ):
        shutil.copyfile(src, tmp_path / dest)

    # Commas are encoded as "\1"
    with open(tmp_path / "test.vmrk", "a") as fout:
        fout.write(r"Mk15=Comma\1Type,CommaValue\11,7800,1,0\n")

    sfreq = 1000.0
    expected_orig_time = _stamp_to_dt((1384359243, 794232))
    expected_onset_latency = np.array(
        [
            0,
            486.0,
            496.0,
            1769.0,
            1779.0,
            3252.0,
            3262.0,
            4935.0,
            4945.0,
            5999.0,
            6619.0,
            6629.0,
            7629.0,
            7699.0,
            7799.0,
        ]
    )
    expected_annot_description = [
        "New Segment/",
        "Stimulus/S253",
        "Stimulus/S255",
        "Event/254",
        "Stimulus/S255",
        "Event/254",
        "Stimulus/S255",
        "Stimulus/S253",
        "Stimulus/S255",
        "Response/R255",
        "Event/254",
        "Stimulus/S255",
        "SyncStatus/Sync On",
        "Optic/O  1",
        "Comma,Type/CommaValue,1",
    ]
    expected_events = (
        np.stack(
            [
                expected_onset_latency,
                np.zeros_like(expected_onset_latency),
                [
                    99999,
                    253,
                    255,
                    254,
                    255,
                    254,
                    255,
                    253,
                    255,
                    1255,
                    254,
                    255,
                    99998,
                    2001,
                    10001,
                ],
            ]
        )
        .astype("int64")
        .T
    )
    expected_event_id = {
        "New Segment/": 99999,
        "Stimulus/S253": 253,
        "Stimulus/S255": 255,
        "Event/254": 254,
        "Response/R255": 1255,
        "SyncStatus/Sync On": 99998,
        "Optic/O  1": 2001,
        "Comma,Type/CommaValue,1": 10001,
    }

    raw = read_raw_brainvision(tmp_path / "test.vhdr", eog=eog)

    # validate annotations
    assert raw.annotations.orig_time == expected_orig_time
    assert_allclose(raw.annotations.onset, expected_onset_latency / sfreq)
    assert_array_equal(raw.annotations.description, expected_annot_description)

    # validate event extraction
    events, event_id = events_from_annotations(raw)
    assert_array_equal(events, expected_events)
    assert event_id == expected_event_id

    # validate that None gives us a sorted list
    expected_none_event_id = {
        desc: idx + 1 for idx, desc in enumerate(sorted(event_id.keys()))
    }
    events, event_id = events_from_annotations(raw, event_id=None)
    assert event_id == expected_none_event_id

    # Add some custom ones, plus a 2-digit one
    s_10 = "Stimulus/S 10"
    raw.annotations.append([1, 2, 3], 10, ["ZZZ", s_10, "YYY"])
    # others starting at 10001 ...
    # we already have "Comma,Type/CommaValue,1" as 10001
    expected_event_id.update(YYY=10002, ZZZ=10003)
    expected_event_id[s_10] = 10
    _, event_id = events_from_annotations(raw)
    assert event_id == expected_event_id

    # Concatenating two shouldn't change the resulting event_id
    # (BAD and EDGE should be ignored)
    with pytest.warns(RuntimeWarning, match="expanding outside"):
        raw_concat = concatenate_raws([raw.copy(), raw.copy()])
    _, event_id = events_from_annotations(raw_concat)
    assert event_id == expected_event_id


@testing.requires_testing_data
def test_automatic_vmrk_sfreq_recovery():
    """Test proper sfreq inference by checking the onsets."""
    assert_array_equal(
        read_annotations(vmrk_path, sfreq="auto"),
        read_annotations(vmrk_path, sfreq=1000.0),
    )


@testing.requires_testing_data
def test_event_id_stability_when_save_and_fif_reload(tmp_path):
    """Test load events from brainvision annotations when read_raw_fif."""
    fname = tmp_path / "bv-raw.fif"
    raw = read_raw_brainvision(vhdr_path, eog=eog)
    original_events, original_event_id = events_from_annotations(raw)

    raw.save(fname)
    raw = read_raw_fif(fname)
    events, event_id = events_from_annotations(raw)

    assert event_id == original_event_id
    assert_array_equal(events, original_events)


def test_parse_impedance():
    """Test case for parsing the impedances from header."""
    expected_imp_meas_time = datetime.datetime(
        2013, 11, 13, 16, 12, 27, tzinfo=datetime.timezone.utc
    )
    expected_imp_unit = "kOhm"
    expected_electrodes = [
        "FP1",
        "FP2",
        "F3",
        "F4",
        "C3",
        "C4",
        "P3",
        "P4",
        "O1",
        "O2",
        "F7",
        "F8",
        "P7",
        "P8",
        "Fz",
        "FCz",
        "Cz",
        "CPz",
        "Pz",
        "POz",
        "FC1",
        "FC2",
        "CP1",
        "CP2",
        "FC5",
        "FC6",
        "CP5",
        "CP6",
        "HL",
        "HR",
        "Vb",
        "ReRef",
        "Ref",
        "Gnd",
    ]
    n_electrodes = len(expected_electrodes)
    expected_imps = [np.nan] * (n_electrodes - 2) + [0.0, 4.0]
    expected_imp_lower_bound = 0.0
    expected_imp_upper_bound = [100.0] * (n_electrodes - 2) + [10.0, 10.0]

    expected_impedances = {
        elec: {
            "imp": expected_imps[i],
            "imp_unit": expected_imp_unit,
            "imp_meas_time": expected_imp_meas_time,
            "imp_lower_bound": expected_imp_lower_bound,
            "imp_upper_bound": expected_imp_upper_bound[i],
            "imp_range_unit": expected_imp_unit,
        }
        for i, elec in enumerate(expected_electrodes)
    }

    raw = read_raw_brainvision(vhdr_path, eog=eog)
    assert object_diff(expected_impedances, raw.impedances) == ""

    # Test "Impedances Imported from actiCAP Control Software"
    expected_imp_meas_time = expected_imp_meas_time.replace(
        hour=10, minute=17, second=2
    )
    tmpidx = expected_electrodes.index("CP6")
    expected_electrodes = expected_electrodes[:tmpidx] + [
        "CP 6",
        "ECG+",
        "ECG-",
        "HEOG+",
        "HEOG-",
        "VEOG+",
        "VEOG-",
        "ReRef",
        "Ref",
        "Gnd",
    ]
    n_electrodes = len(expected_electrodes)
    expected_imps = [np.nan] * (n_electrodes - 9) + [
        35.0,
        46.0,
        6.0,
        8.0,
        3.0,
        4.0,
        0.0,
        8.0,
        2.5,
    ]
    expected_impedances = {
        elec: {
            "imp": expected_imps[i],
            "imp_unit": expected_imp_unit,
            "imp_meas_time": expected_imp_meas_time,
        }
        for i, elec in enumerate(expected_electrodes)
    }

    with pytest.warns(RuntimeWarning, match="different .*pass filters"):
        raw = read_raw_brainvision(
            vhdr_mixed_lowpass_path, eog=["HEOG", "VEOG"], misc=["ECG"]
        )
    assert object_diff(expected_impedances, raw.impedances) == ""


@testing.requires_testing_data
def test_ahdr_format():
    """Test case for parsing data in ahdr format."""
    expected_num_channels = 6
    expected_hp = 0.0
    expected_lp = 250.0

    raw = read_raw_brainvision(vamp_ahdr)
    assert raw.info["nchan"] == expected_num_channels
    assert raw.info["highpass"] == expected_hp
    assert raw.info["lowpass"] == expected_lp