Added deidentify hook to base class

.github/workflows/ci-cd.yml

@@ -25,6 +25,7 @@ jobs:
         shell: bash -l {0}
     env:
       PIP_BREAK_SYSTEM_PACKAGES: 1
+      MNE_DATA: ~/mne_data
     steps:
     - name: Checkout
       uses: actions/checkout@v2
@@ -49,18 +50,26 @@ jobs:
     - name: Install Package
       run: python3 -m pip install -e .[test] -e ./extras[test]
     - name: Cache MNE datasets
+      id: cache-mne-data
       uses: actions/cache@v4
       with:
         path: ~/mne_data
         key: mne-data-${{ hashFiles('**/requirements*.txt', '**/pyproject.toml') }}
-        restore-keys: |
-          mne-data-
+    - name: Download MNE datasets
+      if: steps.cache-mne-data.outputs.cache-hit != 'true'
+      run: |
+        mkdir -p $HOME/mne_data
+        python -c "import mne; mne.datasets.testing.data_path(); mne.datasets.sample.data_path()"
+    - name: Download EEGBCI data
+      if: steps.cache-mne-data.outputs.cache-hit != 'true'
+      continue-on-error: true
+      run: python -c "import mne; mne.datasets.eegbci.load_data(subjects=1, runs=[1], update_path=True)"
     # - name: MyPy
     #   run: mypy --install-types --non-interactive  --no-warn-unused-ignores .
     - name: Pytest
       run: pytest -vvs --cov fileformats --cov-config .coveragerc --cov-report xml .
     - name: Upload coverage to Codecov
-      uses: codecov/codecov-action@v2
+      uses: codecov/codecov-action@v5
       with:
         fail_ci_if_error: true
         token: ${{ secrets.CODECOV_TOKEN }}
@@ -92,7 +101,7 @@ jobs:
       run: python3 -m build ${{ matrix.pkg[1] }}
     - name: Check distributions
       run: twine check ${{ matrix.pkg[1] }}/dist/*
-    - uses: actions/upload-artifact@v3
+    - uses: actions/upload-artifact@v7
       with:
         name: built-${{ matrix.pkg[0] }}
         path: ${{ matrix.pkg[1] }}/dist
@@ -102,7 +111,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: Download build
-      uses: actions/download-artifact@v3
+      uses: actions/download-artifact@v7
       with:
         name: built-main
         path: dist
@@ -124,7 +133,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: Download build
-      uses: actions/download-artifact@v3
+      uses: actions/download-artifact@v7
       with:
         name: built-extras
         path: dist

conftest.py

@@ -50,25 +50,28 @@ def testing_data_path() -> Path:
 
 
 @pytest.fixture(scope="session")
-def fif_path(sample_data_path) -> Path:
+def fif_path(sample_data_path: Path) -> Path:
     return sample_data_path / "MEG" / "sample" / "sample_audvis_raw.fif"
 
 
 @pytest.fixture(scope="session")
-def edf_path() -> Path:
-    return Path(mne.datasets.eegbci.load_data(subject=1, runs=[1])[0])
+def edf_plus_path() -> Path:
+    try:
+        return Path(mne.datasets.eegbci.load_data(subjects=1, runs=[1])[0])
+    except Exception as e:
+        pytest.skip(f"EEGBCI data unavailable (physionet.org unreachable?): {e}")
 
 
 @pytest.fixture(scope="session")
-def bv_vhdr_path(testing_data_path) -> Path:
-    return testing_data_path / "BrainVision" / "test.vhdr"
+def bv_vhdr_path(testing_data_path: Path) -> Path:
+    return testing_data_path / "Brainvision" / "test_NO.vhdr"
 
 
 @pytest.fixture(scope="session")
-def ctf_ds_path(testing_data_path) -> Path:
+def ctf_ds_path(testing_data_path: Path) -> Path:
     return testing_data_path / "CTF" / "testdata_ctf.ds"
 
 
 @pytest.fixture(scope="session")
-def kit_sqd_path(testing_data_path) -> Path:
-    return testing_data_path / "KIT" / "test.sqd"
+def kit_sqd_path(testing_data_path: Path) -> Path:
+    return testing_data_path / "KIT" / "MQKIT_125_2sec.con"

extras/fileformats/extras/biosig/__init__.py

@@ -1 +1,5 @@
 from ._version import __version__
+from . import eeg
+from . import meg
+
+__all__ = ["__version__", "eeg", "meg"]

extras/fileformats/extras/biosig/eeg.py

@@ -5,69 +5,73 @@
 from pathlib import Path
 
 import mne.io
+import mne.export
 
 from fileformats.core import extra_implementation, FileSet
-from fileformats.biosig import Biosig, BrainVision, Edf, EdfPlus, Fif, FifGz
+from fileformats.biosig import Biosig, BrainVision, Edf, EdfPlus
 
-from .utils import _info_to_metadata
+from .utils import mne_deidentify
 
 
 @extra_implementation(FileSet.read_metadata)
-def fif_read_metadata(fif: Fif) -> dict[str, ty.Any]:
-    raw = mne.io.read_raw_fif(fif.fspath, preload=False, verbose=False)
-    return _info_to_metadata(raw.info)
-
-
-@extra_implementation(FileSet.read_metadata)
-def fif_gz_read_metadata(fif: FifGz) -> dict[str, ty.Any]:
-    raw = mne.io.read_raw_fif(fif.fspath, preload=False, verbose=False)
-    return _info_to_metadata(raw.info)
-
-
-@extra_implementation(FileSet.read_metadata)
-def edf_read_metadata(edf: Edf) -> dict[str, ty.Any]:
-    raw = mne.io.read_raw_edf(edf.fspath, preload=False, verbose=False)
+def edf_read_metadata(edf: Edf, **kwargs: ty.Any) -> ty.Mapping[str, ty.Any]:
+    raw = mne.io.read_raw_edf(edf, preload=False, verbose=False)
     return {
-        **_info_to_metadata(raw.info),
-        **_parse_edf_header(edf.fspath),
+        **raw.info.to_json_dict(),
+        **_parse_edf_header(edf),
     }
 
 
 @extra_implementation(FileSet.read_metadata)
-def edf_plus_read_metadata(edf: EdfPlus) -> dict[str, ty.Any]:
-    raw = mne.io.read_raw_edf(edf.fspath, preload=False, verbose=False)
+def edf_plus_read_metadata(edf: EdfPlus, **kwargs: ty.Any) -> ty.Mapping[str, ty.Any]:
+    raw = mne.io.read_raw_edf(edf, preload=False, verbose=False)
     return {
-        **_info_to_metadata(raw.info),
-        **_parse_edf_header(edf.fspath),
+        **raw.info.to_json_dict(),
+        **_parse_edf_header(edf),
     }
 
 
 @extra_implementation(FileSet.read_metadata)
-def brain_vision_read_metadata(bv: BrainVision) -> dict[str, ty.Any]:
+def brain_vision_read_metadata(
+    bv: BrainVision, **kwargs: ty.Any
+) -> ty.Mapping[str, ty.Any]:
     raw = mne.io.read_raw_brainvision(bv.header_file, preload=False, verbose=False)
     return {
-        **_info_to_metadata(raw.info),
+        **raw.info.to_json_dict(),
         **_parse_vhdr(bv.header_file),
     }
 
 
+@extra_implementation(Biosig.deidentify)
+def edf_deidentify(
+    edf: Edf,
+    spec: ty.Any = None,
+    out_dir: os.PathLike[str] | None = None,
+) -> tuple[Edf, dict[str, ty.Any]]:
+    out_dir = Path(tempfile.mkdtemp() if out_dir is None else out_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    raw = mne.io.read_raw_edf(edf, preload=True, verbose=False)
+    deidentified_info, reid = mne_deidentify(raw, spec)
+    raw.info = deidentified_info
+    deid_fspath = out_dir / "eeg.edf"
+    mne.export.export_raw(deid_fspath, raw, fmt="edf", overwrite=True)
+    return type(edf)(deid_fspath), reid
+
+
 @extra_implementation(Biosig.deidentify)
 def brain_vision_deidentify(
-    brain_vision: BrainVision,
-    out_dir: ty.Optional[Path] = None,
-    new_stem: ty.Optional[str] = None,
-    copy_mode: FileSet.CopyMode = FileSet.CopyMode.copy,
-) -> BrainVision:
-    if out_dir is None:
-        out_dir = Path(tempfile.mkdtemp())
+    bv: BrainVision,
+    spec: ty.Any = None,
+    out_dir: os.PathLike[str] | None = None,
+) -> tuple[BrainVision, dict[str, ty.Any]]:
+    out_dir = Path(tempfile.mkdtemp() if out_dir is None else out_dir)
     out_dir.mkdir(parents=True, exist_ok=True)
-    deidentified = brain_vision.copy(out_dir, new_stem=new_stem, mode=copy_mode)
-    raise NotImplementedError(
-        "need to implemnent deidentification techniques and a save method. If there is a standard "
-        "form to load the data into (e.g. MNE) it would be best to implement FileSet.load and FileSet.save"
-        "methods"
-    )
-    return deidentified
+    raw = mne.io.read_raw_brainvision(bv.header_file, preload=True, verbose=False)
+    deidentified_info, reid = mne_deidentify(raw, spec)
+    raw.info = deidentified_info
+    deid_vhdr = out_dir / "eeg.vhdr"
+    mne.export.export_raw(deid_vhdr, raw, fmt="brainvision", overwrite=True)
+    return BrainVision(out_dir / "eeg.eeg"), reid
 
 
 def _parse_edf_header(path: os.PathLike[str]) -> dict[str, ty.Any]:
@@ -117,7 +121,7 @@ def _parse_vhdr(path: os.PathLike[str]) -> dict[str, ty.Any]:
     """
     parser = configparser.RawConfigParser()
     # vhdr files start with a magic line before the first INI section — skip it
-    with open(path, encoding="utf-8", errors="replace") as f:
+    with open(path, encoding="utf-8-sig", errors="replace") as f:
         lines = f.readlines()
     ini_lines = [line for line in lines if not line.startswith("Brain Vision")]
     parser.read_string("".join(ini_lines))

extras/fileformats/extras/biosig/meg.py

@@ -1,56 +1,43 @@
+import os
 import typing as ty
-import xml.etree.ElementTree as ET
 from pathlib import Path
+import tempfile
 
 import mne.io
 
 from fileformats.core import extra_implementation, FileSet
-from fileformats.biosig import Ctf, Kit
-from .utils import _info_to_metadata
+from fileformats.biosig import Ctf, Kit, Fif
+from fileformats.biosig.base import Biosig
+
+from .utils import mne_deidentify
 
 
 @extra_implementation(FileSet.read_metadata)
-def ctf_read_metadata(ctf: Ctf) -> dict[str, ty.Any]:
-    raw = mne.io.read_raw_ctf(ctf.fspath, preload=False, verbose=False)
-    return {
-        **_info_to_metadata(raw.info),
-        **_parse_infods(Path(ctf.fspath)),
-    }
+def ctf_read_metadata(ctf: Ctf, **kwargs: ty.Any) -> ty.Mapping[str, ty.Any]:
+    return mne.io.read_raw_ctf(ctf, preload=False, verbose=False).info.to_json_dict()  # type: ignore[no-any-return]
 
 
-# elif ext in [".sqd", ".con"]:  # KIT/RIKEN main data files
-#                 # For KIT format, we need to find the marker file (.mrk) in the same directory
+@extra_implementation(FileSet.read_metadata)
+def kit_read_metadata(kit: Kit, **kwargs: ty.Any) -> ty.Mapping[str, ty.Any]:
+    return mne.io.read_raw_kit(kit, mrk=kit.mark_file, verbose=False).info.to_json_dict()  # type: ignore[no-any-return]
 
 
 @extra_implementation(FileSet.read_metadata)
-def kit_read_metadata(kit: Kit) -> dict[str, ty.Any]:
-    mrk_path = kit._find_kit_mrk_file()
-    return mne.io.read_raw_kit(kit, mrk=mrk_path, verbose=False)
-
-
-def _parse_infods(ds_path: Path) -> dict[str, ty.Any]:
-    """
-    Parse the .infods XML sidecar in a CTF .ds directory for metadata that
-    MNE does not surface via raw.info (subject name, operator, study description).
-    Returns an empty dict if no .infods file is present.
-    """
-    infods_files = list(ds_path.glob("*.infods"))
-    if not infods_files:
-        return {}
-    tree = ET.parse(infods_files[0])
-    root = tree.getroot()
-
-    def find(tag: str) -> str | None:
-        el = root.find(f".//{tag}")
-        return el.text.strip() if el is not None and el.text else None
-
-    return {
-        "subject_id": find("SUBJECTID"),
-        "subject_name": find("SUBJECTNAME"),
-        "operator": find("OPERATOR"),
-        "institution": find("INSTITUTION"),
-        "study_description": find("STUDYDESCRIPTION"),
-        "run_description": find("RUNDESCRIPTION"),
-        "acquisition_datetime": find("ACQUISITIONDATETIME"),
-        "date": find("DATE"),
-    }
+def fif_read_metadata(fif: Fif, **kwargs: ty.Any) -> ty.Mapping[str, ty.Any]:
+    return mne.io.read_raw_fif(fif, preload=False, verbose=False).info.to_json_dict()  # type: ignore[no-any-return]
+
+
+@extra_implementation(Biosig.deidentify)
+def fif_deidentify(
+    fif: Fif,
+    spec: ty.Any = None,
+    out_dir: os.PathLike[str] | None = None,
+) -> tuple[Fif, dict[str, ty.Any]]:
+    out_dir = Path(tempfile.mkdtemp() if out_dir is None else out_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    raw = mne.io.read_raw_fif(fif, preload=True, verbose=False)
+    deidentified_info, reid = mne_deidentify(raw, spec)
+    raw.info = deidentified_info
+    deid_fspath = out_dir / "meg-signals.fif"
+    raw.save(deid_fspath, overwrite=True)
+    return Fif(deid_fspath), reid

extras/fileformats/extras/biosig/tests/test_eeg_extras.py

@@ -12,28 +12,16 @@
 
 from fileformats.biosig import (
     BrainVision,
-    Edf,
-    Fif,
+    EdfPlus,
 )
 
-# ------------------------------
-# EEG: FIF
-# ------------------------------
-
-
-def test_fif_read_metadata(fif_path):
-    metadata = Fif(fif_path).metadata
-    assert isinstance(metadata, dict)
-    assert metadata["sfreq"] is not None
-
-
 # ------------------------------
 # EEG: EDF
 # ------------------------------
 
 
-def test_edf_read_metadata(edf_path):
-    metadata = Edf(edf_path).metadata
+def test_edf_plus_read_metadata(edf_plus_path):
+    metadata = EdfPlus(edf_plus_path).metadata
     assert isinstance(metadata, dict)
     assert metadata["sfreq"] is not None
     assert "edf_patient_code" in metadata
@@ -45,7 +33,7 @@ def test_edf_read_metadata(edf_path):
 
 
 def test_brainvision_read_metadata(bv_vhdr_path):
-    metadata = BrainVision(bv_vhdr_path.iterdir()).metadata
+    metadata = BrainVision(bv_vhdr_path.with_suffix(".eeg")).metadata
     assert isinstance(metadata, dict)
     assert metadata["sfreq"] is not None
     assert "bv_n_channels" in metadata