read only for now

Author: bruAristimunha

mne_bids/tests/test_read.py

@@ -6,13 +6,14 @@
 import contextlib
 import json
 import logging
+import multiprocessing as mp
 import os
 import os.path as op
 import re
 import shutil as sh
 from collections import OrderedDict
 from contextlib import nullcontext
-from datetime import datetime, timezone
+from datetime import date, datetime, timezone
 from pathlib import Path
 
 import mne
@@ -99,6 +100,49 @@ def fn(fname, *args, **kwargs):
 _read_raw_edf = _wrap_read_raw(mne.io.read_raw_edf)
 
 
+def _make_parallel_raw(subject, *, seed=None):
+    """Generate a lightweight Raw instance for parallel-reading tests."""
+    rng_seed = seed if seed is not None else sum(ord(ch) for ch in subject)
+    rng = np.random.default_rng(rng_seed)
+    info = mne.create_info(["MEG0113"], 100, ch_types="mag")
+    data = rng.standard_normal((1, 100)) * 1e-12
+    raw = mne.io.RawArray(data, info)
+    raw.set_meas_date(datetime(2020, 1, 1, tzinfo=timezone.utc))
+    raw.info["line_freq"] = 60
+    raw.info["subject_info"] = {
+        "his_id": subject,
+        "sex": 1,
+        "hand": 2,
+        "birthday": date(1990, 1, 1),
+    }
+    return raw
+
+
+def _write_parallel_dataset(root, *, subject, run):
+    """Write a minimal dataset using write_raw_bids."""
+    root = Path(root)
+    raw = _make_parallel_raw(subject)
+    bids_path = BIDSPath(
+        subject=subject, task="rest", run=run, datatype="meg", root=root
+    )
+    write_raw_bids(raw, bids_path, allow_preload=True, format="FIF", verbose=False)
+
+
+def _parallel_read_participants(root, expected_ids):
+    """Read participants.tsv in a multiprocessing worker."""
+    participants_path = Path(root) / "participants.tsv"
+    participants = _from_tsv(participants_path)
+    assert set(participants["participant_id"]) == set(expected_ids)
+
+
+def _parallel_read_scans(root, expected_filenames):
+    """Read scans.tsv in a multiprocessing worker."""
+    scans_path = BIDSPath(subject="01", root=root, suffix="scans", extension=".tsv")
+    scans = _from_tsv(scans_path.fpath)
+    filenames = {str(filename) for filename in scans["filename"]}
+    assert filenames == set(expected_filenames)
+
+
 def test_read_raw():
     """Test the raw reading."""
     # Use a file ending that does not exist
@@ -133,6 +177,71 @@ def test_read_correct_inputs():
         read_raw_bids(bids_path)
 
 
+@pytest.mark.filterwarnings(
+    "ignore:No events found or provided:RuntimeWarning",
+    "ignore:Found no extension for raw file.*:RuntimeWarning",
+)
+def test_parallel_participants_multiprocess(tmp_path):
+    """Ensure parallel reads keep all participants entries visible."""
+    bids_root = tmp_path / "parallel_multiprocess"
+    subjects = [f"{i:02d}" for i in range(1, 50)]
+
+    for subject in subjects:
+        _write_parallel_dataset(str(bids_root), subject=subject, run="01")
+
+    expected_ids = [f"sub-{subject}" for subject in subjects]
+    processes = []
+    for _ in range(len(subjects) // 10):  # spawn a few processes
+        proc = mp.Process(
+            target=_parallel_read_participants, args=(str(bids_root), expected_ids)
+        )
+        proc.start()
+        processes.append(proc)
+
+    for proc in processes:
+        proc.join()
+        assert proc.exitcode == 0
+
+    participants_path = bids_root / "participants.tsv"
+    assert participants_path.exists()
+    participants = _from_tsv(participants_path)
+    assert set(participants["participant_id"]) == set(expected_ids)
+    sh.rmtree(bids_root, ignore_errors=True)
+
+
+@pytest.mark.filterwarnings(
+    "ignore:No events found or provided:RuntimeWarning",
+    "ignore:Found no extension for raw file.*:RuntimeWarning",
+)
+def test_parallel_scans_multiprocessing(tmp_path):
+    """Ensure multiprocessing reads see all runs in scans.tsv."""
+    bids_root = tmp_path / "parallel_multiprocessing"
+    runs = [f"{i:02d}" for i in range(1, 50)]
+
+    for run in runs:
+        _write_parallel_dataset(str(bids_root), subject="01", run=run)
+
+    expected = {f"meg/sub-01_task-rest_run-{run}_meg.fif" for run in runs}
+    processes = []
+    for _ in range(4):
+        proc = mp.Process(target=_parallel_read_scans, args=(str(bids_root), expected))
+        proc.start()
+        processes.append(proc)
+
+    for proc in processes:
+        proc.join()
+        assert proc.exitcode == 0
+
+    scans_path = BIDSPath(
+        subject="01", root=bids_root, suffix="scans", extension=".tsv"
+    )
+    assert scans_path.fpath.exists()
+    scans = _from_tsv(scans_path.fpath)
+    filenames = {str(filename) for filename in scans["filename"]}
+    assert filenames == expected
+    sh.rmtree(bids_root, ignore_errors=True)
+
+
 @pytest.mark.filterwarnings(warning_str["channel_unit_changed"])
 @testing.requires_testing_data
 def test_read_participants_data(tmp_path):

mne_bids/tests/test_write.py

@@ -8,7 +8,6 @@
 
 import codecs
 import json
-import multiprocessing as mp
 import os
 import os.path as op
 import shutil as sh
@@ -100,44 +99,6 @@
 )
 
 
-def _make_parallel_raw(subject, *, seed=None):
-    """Generate a lightweight Raw instance for parallel-writing tests."""
-    rng_seed = seed if seed is not None else sum(ord(ch) for ch in subject)
-    rng = np.random.default_rng(rng_seed)
-    info = mne.create_info(["MEG0113"], sfreq=100.0, ch_types="mag")
-    data = rng.standard_normal((1, 100)) * 1e-12
-    raw = mne.io.RawArray(data, info)
-    raw.set_meas_date(datetime(2025, 1, 1, tzinfo=timezone.utc))
-    raw.info["line_freq"] = 60
-    raw.info["subject_info"] = {
-        "his_id": subject,
-        "sex": 1,
-        "hand": 2,
-        "birthday": date(1990, 1, 1),
-    }
-    return raw
-
-
-def _write_parallel_dataset(root, *, subject, run):
-    """Write a minimal dataset using write_raw_bids."""
-    root = Path(root)
-    raw = _make_parallel_raw(subject)
-    bids_path = BIDSPath(
-        subject=subject, task="rest", run=run, datatype="meg", root=root
-    )
-    write_raw_bids(raw, bids_path, allow_preload=True, format="FIF", verbose=False)
-
-
-def _parallel_write_subject(root, subject):
-    """Handle write_raw_bids call in a multiprocessing worker."""
-    _write_parallel_dataset(root, subject=subject, run="01")
-
-
-def _multiprocessing_write_run(root, run):
-    """Handle write_raw_bids call in a multiprocessing worker."""
-    _write_parallel_dataset(root, subject="01", run=run)
-
-
 def _wrap_read_raw(read_raw):
     def fn(fname, *args, **kwargs):
         if Path(fname).suffix == ".mff":
@@ -356,65 +317,6 @@ def test_write_participants(_bids_validate, tmp_path):
     assert participants_tsv["age"][idx] == "n/a"
 
 
-@pytest.mark.filterwarnings(
-    "ignore:No events found or provided:RuntimeWarning",
-    "ignore:Found no extension for raw file.*:RuntimeWarning",
-)
-def test_parallel_participants_multiprocess(tmp_path):
-    """Ensure parallel writes keep all participants entries."""
-    bids_root = tmp_path / "parallel_multiprocess"
-    subjects = [f"{i:02d}" for i in range(1, 50)]
-
-    processes = []
-    for subject in subjects:
-        proc = mp.Process(
-            target=_parallel_write_subject, args=(str(bids_root), subject)
-        )
-        proc.start()
-        processes.append(proc)
-
-    for proc in processes:
-        proc.join()
-        assert proc.exitcode == 0
-
-    participants_path = bids_root / "participants.tsv"
-    assert participants_path.exists()
-    participants = _from_tsv(participants_path)
-    expected_ids = {f"sub-{subject}" for subject in subjects}
-    assert set(participants["participant_id"]) == expected_ids
-    sh.rmtree(bids_root, ignore_errors=True)
-
-
-@pytest.mark.filterwarnings(
-    "ignore:No events found or provided:RuntimeWarning",
-    "ignore:Found no extension for raw file.*:RuntimeWarning",
-)
-def test_parallel_scans_multiprocessing(tmp_path):
-    """Ensure multiprocessing writes add all runs to scans.tsv."""
-    bids_root = tmp_path / "parallel_multiprocessing"
-    runs = [f"{i:02d}" for i in range(1, 50)]
-
-    processes = []
-    for run in runs:
-        proc = mp.Process(target=_multiprocessing_write_run, args=(str(bids_root), run))
-        proc.start()
-        processes.append(proc)
-
-    for proc in processes:
-        proc.join()
-        assert proc.exitcode == 0
-
-    scans_path = BIDSPath(
-        subject="01", root=bids_root, suffix="scans", extension=".tsv"
-    )
-    assert scans_path.fpath.exists()
-    scans = _from_tsv(scans_path.fpath)
-    filenames = {str(filename) for filename in scans["filename"]}
-    expected = {f"meg/sub-01_task-rest_run-{run}_meg.fif" for run in runs}
-    assert filenames == expected
-    sh.rmtree(bids_root, ignore_errors=True)
-
-
 @testing.requires_testing_data
 def test_write_correct_inputs():
     """Test that inputs of write_raw_bids is correct."""