ENH: replace _get_blocks binary reader with mffpy Reader API

doc/changes/dev/14043.other.rst

@@ -0,0 +1 @@
+Replace manual binary block reader ``_get_blocks`` with ``mffpy.Reader`` API in ``_read_mff_header``, removing low-level EGI binary parsing in favour of the existing mffpy dependency, by `Pragnya Khandelwal`_.

mne/io/egi/egimff.py

@@ -25,7 +25,6 @@
 from ..base import BaseRaw
 from .events import _combine_triggers, _read_events, _triage_include_exclude
 from .general import (
-    _get_blocks,
     _get_ep_info,
     _get_signalfname,
 )
@@ -70,6 +69,7 @@ def _disk_range_to_epochs(egi_info, disk_start, disk_stop):
 def _read_mff_header(filepath):
     """Read mff header."""
     _soft_import("mffpy", "reading EGI MFF data")
+    from mffpy import Reader
     from mffpy.xml_files import XML
 
     all_files = _get_signalfname(filepath)
@@ -88,8 +88,14 @@ def _read_mff_header(filepath):
     rt_elem = info_obj.find("recordTime")
     record_time = str(rt_elem.text) if rt_elem is not None else ""
 
-    fname = op.join(filepath, eeg_file)
-    signal_blocks = _get_blocks(fname)
+    reader = Reader(filepath)
+    signal_blocks = dict(
+        n_channels=reader.num_channels["EEG"],
+        sfreq=reader.sampling_rates["EEG"],
+        n_blocks=len(reader.block_sample_counts["EEG"]),
+        samples_block=np.array(reader.block_sample_counts["EEG"]),
+        header_sizes=[],
+    )
     epochs = _get_ep_info(filepath)
     summaryinfo = dict(eeg_fname=eeg_file, info_fname=eeg_info_file)
     summaryinfo.update(signal_blocks)
@@ -169,9 +175,10 @@ def _read_mff_header(filepath):
 
     pns_names = []
     if "PNS" in all_files:
-        pns_fpath = op.join(filepath, all_files["PNS"]["signal"])
-        pns_blocks = _get_blocks(pns_fpath)
-        pns_samples = pns_blocks["samples_block"]
+        pns_sample_blocks = dict(
+            samples_block=np.array(reader.block_sample_counts["PNSData"])
+        )
+        pns_samples = pns_sample_blocks["samples_block"]
         signal_samples = signal_blocks["samples_block"]
         same_blocks = np.array_equal(
             pns_samples[:-1], signal_samples[:-1]
@@ -210,7 +217,7 @@ def _read_mff_header(filepath):
             pns_types=pns_types,
             pns_units=pns_units,
             pns_fname=all_files["PNS"]["signal"],
-            pns_sample_blocks=pns_blocks,
+            pns_sample_blocks=pns_sample_blocks,
         )
 
     summaryinfo.update(

mne/io/egi/general.py

@@ -6,8 +6,6 @@
 import os
 import re
 
-import numpy as np
-
 from ...utils import _pl
 
 
@@ -28,62 +26,6 @@ def _get_ep_info(filepath):
     return epoch_info
 
 
-def _get_blocks(filepath):
-    """Get info from meta data blocks."""
-    binfile = os.path.join(filepath)
-    n_blocks = 0
-    samples_block = []
-    header_sizes = []
-    n_channels = []
-    sfreq = []
-    # Meta data consists of:
-    # * 1 byte of flag (1 for meta data, 0 for data)
-    # * 1 byte of header size
-    # * 1 byte of block size
-    # * 1 byte of n_channels
-    # * n_channels bytes of offsets
-    # * n_channels bytes of sigfreqs?
-    with open(binfile, "rb") as fid:
-        fid.seek(0, 2)  # go to end of file
-        file_length = fid.tell()
-        block_size = file_length
-        fid.seek(0)
-        position = 0
-        while position < file_length:
-            block = _block_r(fid)
-            if block is None:
-                samples_block.append(samples_block[n_blocks - 1])
-                n_blocks += 1
-                fid.seek(block_size, 1)
-                position = fid.tell()
-                continue
-            block_size = block["block_size"]
-            header_size = block["header_size"]
-            header_sizes.append(header_size)
-            samples_block.append(block["nsamples"])
-            n_blocks += 1
-            fid.seek(block_size, 1)
-            sfreq.append(block["sfreq"])
-            n_channels.append(block["nc"])
-            position = fid.tell()
-
-    if any([n != n_channels[0] for n in n_channels]):
-        raise RuntimeError("All the blocks don't have the same amount of channels.")
-    if any([f != sfreq[0] for f in sfreq]):
-        raise RuntimeError("All the blocks don't have the same sampling frequency.")
-    if len(samples_block) < 1:
-        raise RuntimeError("There seems to be no data")
-    samples_block = np.array(samples_block)
-    signal_blocks = dict(
-        n_channels=n_channels[0],
-        sfreq=sfreq[0],
-        n_blocks=n_blocks,
-        samples_block=samples_block,
-        header_sizes=header_sizes,
-    )
-    return signal_blocks
-
-
 def _get_signalfname(filepath):
     """Get filenames."""
     from mffpy.xml_files import XML
@@ -113,31 +55,3 @@ def _get_signalfname(filepath):
             f"found in {filepath}:\n{infofiles_str}"
         )
     return all_files
-
-
-def _block_r(fid):
-    """Read meta data."""
-    if np.fromfile(fid, dtype=np.dtype("i4"), count=1).item() != 1:  # not meta
-        return None
-    header_size = np.fromfile(fid, dtype=np.dtype("i4"), count=1).item()
-    block_size = np.fromfile(fid, dtype=np.dtype("i4"), count=1).item()
-    hl = int(block_size / 4)
-    nc = np.fromfile(fid, dtype=np.dtype("i4"), count=1).item()
-    nsamples = int(hl / nc)
-    np.fromfile(fid, dtype=np.dtype("i4"), count=nc)  # sigoffset
-    sigfreq = np.fromfile(fid, dtype=np.dtype("i4"), count=nc)
-    depth = sigfreq[0] & 0xFF
-    if depth != 32:
-        raise ValueError("I do not know how to read this MFF (depth != 32)")
-    sfreq = sigfreq[0] >> 8
-    count = int(header_size / 4 - (4 + 2 * nc))
-    np.fromfile(fid, dtype=np.dtype("i4"), count=count)  # sigoffset
-    block = dict(
-        nc=nc,
-        hl=hl,
-        nsamples=nsamples,
-        block_size=block_size,
-        header_size=header_size,
-        sfreq=sfreq,
-    )
-    return block