MAINT: Replace manual PNS binary block reader in _read_segment_file with mffpy (#14030)

Author: PragnyaKhandelwal

doc/changes/dev/14030.other.rst

@@ -0,0 +1 @@
+Replace the manual PNS binary block reader in :func:`mne.io.read_raw_egi` with ``mffpy``-backed reading via ``get_physical_samples_from_epoch``, 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 (
-    _block_r,
     _get_blocks,
     _get_ep_info,
     _get_signalfname,
@@ -598,7 +597,6 @@ def __init__(
     def _read_segment_file(self, data, idx, fi, start, stop, cals, mult):
         """Read a chunk of data."""
         logger.debug(f"Reading MFF {start:6d} ... {stop:6d} ...")
-        dtype = "<f4"  # Data read in four byte floats.
 
         egi_info = self._raw_extras[fi]
         one = np.zeros((egi_info["kind_bounds"][-1], stop - start))
@@ -649,71 +647,17 @@ def _read_segment_file(self, data, idx, fi, start, stop, cals, mult):
                 one[eeg_one, cols] = eeg_chunk[eeg_in]
 
         if len(pns_one) > 0:
-            # PNS Data is present and should be read:
-            pns_filepath = egi_info["pns_filepath"]
-            pns_info = egi_info["pns_sample_blocks"]
-            n_channels = pns_info["n_channels"]
-            samples_block = pns_info["samples_block"]
-
-            # Get starting/stopping block/samples
-            block_samples_offset = np.cumsum(samples_block)
-            offset_blocks = np.sum(block_samples_offset < start)
-            offset_samples = start - (
-                block_samples_offset[offset_blocks - 1] if offset_blocks > 0 else 0
-            )
-
-            samples_to_read = stop - start
-            with open(pns_filepath, "rb", buffering=0) as fid:
-                # Check file size
-                fid.seek(0, 2)
-                file_size = fid.tell()
-                fid.seek(0)
-                # Go to starting block
-                current_block = 0
-                current_block_info = None
-                current_data_sample = 0
-                while current_block < offset_blocks:
-                    this_block_info = _block_r(fid)
-                    if this_block_info is not None:
-                        current_block_info = this_block_info
-                    fid.seek(current_block_info["block_size"], 1)
-                    current_block += 1
-
-                # Start reading samples
-                while samples_to_read > 0:
-                    if samples_to_read == 1 and fid.tell() == file_size:
-                        # We are in the presence of the EEG bug
-                        # fill with zeros and break the loop
-                        one[pns_one, -1] = 0
-                        break
-
-                    this_block_info = _block_r(fid)
-                    if this_block_info is not None:
-                        current_block_info = this_block_info
-
-                    to_read = current_block_info["nsamples"] * current_block_info["nc"]
-                    block_data = np.fromfile(fid, dtype, to_read)
-                    block_data = block_data.reshape(n_channels, -1, order="C")
-
-                    # Compute indexes
-                    samples_read = block_data.shape[1]
-                    if offset_samples > 0:
-                        # First block read, skip to the offset:
-                        block_data = block_data[:, offset_samples:]
-                        samples_read = samples_read - offset_samples
-                        offset_samples = 0
-
-                    if samples_to_read < samples_read:
-                        # Last block to read, skip the last samples
-                        block_data = block_data[:, :samples_to_read]
-                        samples_read = samples_to_read
-
-                    s_start = current_data_sample
-                    s_end = s_start + samples_read
-
-                    one[pns_one, disk_use_idx[s_start:s_end]] = block_data[pns_in]
-                    samples_to_read = samples_to_read - samples_read
-                    current_data_sample = current_data_sample + samples_read
+            mff_reader = _get_mff_reader(egi_info["mff_path"])
+            mff_epochs = mff_reader.epochs
+            for ei, t0, dt, out_start, out_stop in _disk_range_to_epochs(
+                egi_info, start, stop
+            ):
+                logger.debug(f"    Reading PNS from epoch {ei} t0={t0} dt={dt}")
+                pns_chunk, _ = mff_reader.get_physical_samples_from_epoch(
+                    mff_epochs[ei], t0=t0, dt=dt, channels=["PNSData"]
+                )["PNSData"]
+                cols = disk_use_idx[out_start:out_stop]
+                one[pns_one, cols[: pns_chunk.shape[1]]] = pns_chunk[pns_in]
 
         # do the calibration
         _mult_cal_one(data, one, idx, cals, mult)