Ingest: write the epoch's data, ingest by segment in parallel, and compare tool (NDI-matlab#1089, #1090, #1091)

CHANGELOG.md

@@ -24,6 +24,34 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
   clones), and the `ndi.fun.epoch` and `ndi.setup` entries are re-examined
   against NDI-matlab main.
 
+- **Ingestion stores the data** (mirroring NDI-matlab's
+  `ndi.daq.reader.mfdaq/ingest_epochfiles`, with VH-Lab/NDI-matlab#1089's
+  within-epoch progress). Ingesting an epoch used to make its documents
+  without any data, so an ingested epoch could not be read (`channel_list.bin
+  not found`). `ndi_daq_reader_mfdaq.ingest_epochfiles` now writes the channel
+  list and the epoch's data in compressed segments as MATLAB does -- analog
+  data in the reader's underlying type (Intan: uint16, via the new
+  `ndi_daq_reader_mfdaq_intan.underlying_datatype`), digital, time, and
+  event/marker/text channels -- and metadata readers store their parameters
+  (`data.bin`), which `readmetadata_ingested` now reads back. The channel list
+  is written in MATLAB's `saveStructArray` format so NDI-matlab can read it.
+  `ingest_epochfiles` takes a `ProgressFcn`, and `ingestEpochDocuments`
+  reports fractional progress as segments are written, in worker processes
+  too. The Intan reader now reads auxiliary inputs at their own sample rate
+  and its digital channels as NDR numbers them (both as MATLAB), and no
+  longer pads reads that run past the end of the file. Needs ndi-compress with
+  Waltham-Data-Science/NDI-compress-python#4.
+
+- **Parallel ingestion by segment, and an ingest comparison tool** (mirroring
+  VH-Lab/NDI-matlab#1090 and #1091). `ingest_epochfiles` is split into
+  `ingest_epochfiles_plan`, `ingest_segment` and the static
+  `ingest_attach_segments`, and `ingestEpochDocuments` now hands single
+  segments to the worker processes, so every worker stays busy even when one
+  epoch is much longer than the others. The new `ndi.test.ingest.compare`
+  and `mfdaq_compare` compare a raw session with an ingested copy of it --
+  every epoch, channel list, sample and event -- and return a report of
+  every check, with no stopping or prompting on a mismatch.
+
 - **Ingestion progress bars and parallel per-epoch work** (mirroring
   VH-Lab/NDI-matlab#1079). `ndi_session.ingest()` now shows progress through
   `ndi_gui_component_ProgressBarWindow` -- one bar for the session and one per

src/ndi/daq/fun/ingest_epoch_documents.py

@@ -19,11 +19,12 @@
 from __future__ import annotations
 
 import copy
+import functools
 import multiprocessing
 import pickle
 import warnings
 from collections.abc import Callable
-from concurrent.futures import ProcessPoolExecutor, as_completed
+from concurrent.futures import FIRST_COMPLETED, ProcessPoolExecutor, wait
 from typing import Any
 
 from pydantic import ConfigDict, validate_call
@@ -37,7 +38,7 @@ def ingestEpochDocuments(  # noqa: N802 (MATLAB mirror)
     daqmetadatareaders: list[Any],
     epochFiles: list[list[str]],
     epochIds: list[str],
-    ProgressFcn: Callable[[int, int], Any] | None = None,
+    ProgressFcn: Callable[[float, int], Any] | None = None,
     NumberOfWorkers: int | None = None,
 ) -> tuple[list[list[Any]], bool]:
     """Build the ingested-data documents for a set of epochs.
@@ -67,7 +68,10 @@ def ingestEpochDocuments(  # noqa: N802 (MATLAB mirror)
         epochFiles: One list of file names per epoch.
         epochIds: One epoch id per epoch.
         ProgressFcn: Called as ``ProgressFcn(numDone, numTotal)`` in the calling
-            process after each epoch completes.
+            process as epochs progress. numDone counts epochs and is fractional
+            while an epoch is underway: readers whose ingest_epochfiles is
+            ndi_daq_reader_mfdaq's report each segment they write. It is called
+            only when numDone changes, so each whole number is reported once.
         NumberOfWorkers: Overrides the preferences for this call (passed to
             ndi.fun.parallelWorkers); 0 forces serial.
 
@@ -81,27 +85,44 @@ def ingestEpochDocuments(  # noqa: N802 (MATLAB mirror)
         workers -- and live only for this call. Readers are sent with their
         ``_session`` cleared (see the module docstring). A reader that cannot
         be pickled falls back to serial with a warning rather than failing.
-        Completed epochs are reported as they finish, so ProgressFcn may see
-        epochs out of order, but the returned list is always in epoch order.
+        On workers, a reader whose ingest_epochfiles is ndi_daq_reader_mfdaq's
+        is ingested one segment per task (MATLAB's ``parfeval`` scheduler):
+        every epoch is planned in the calling process, the segments of all
+        epochs (and each epoch's metadata readers) are handed to the workers
+        as they free up, and the files are attached in plan order afterwards.
+        Progress is reported as each task finishes, weighted by segment size,
+        so ProgressFcn may see epochs out of order, but the returned list is
+        always in epoch order. The pool is used even for a single epoch.
     """
     num_epochs = len(epochFiles)
     if len(epochIds) != num_epochs:
         raise ValueError("EPOCHFILES and EPOCHIDS must have the same number of entries.")
 
     docs_per_epoch: list[list[Any]] = [[] for _ in range(num_epochs)]
-    done = 0
-
-    def tick() -> None:
-        nonlocal done
-        done += 1
+    # per-epoch fractions done
+    frac = [0.0] * num_epochs
+
+    def update_progress(k: int, fraction: float) -> None:
+        # Fractions only move forward, and ProgressFcn is called only when the
+        # total changes, so a reader's own final report and the epoch's
+        # completion do not count the epoch twice.
+        new_fraction = max(frac[k], min(max(float(fraction), 0.0), 1.0))
+        if new_fraction == frac[k]:
+            return
+        frac[k] = new_fraction
         if ProgressFcn is not None:
             try:
-                ProgressFcn(done, num_epochs)
+                ProgressFcn(sum(frac), num_epochs)
             except Exception as exc:  # noqa: BLE001 - progress must never break an ingest
                 warnings.warn(f"Progress reporting failed: {exc}", stacklevel=3)
 
+    # readers using ndi_daq_reader_mfdaq's ingest_epochfiles take a ProgressFcn
+    # and can be ingested segment by segment; any other reader works (and
+    # reports) per epoch
+    plans_segments = _reader_plans_segments(daqreader)
+
     number_of_workers = 0
-    if num_epochs > 1:
+    if num_epochs > 0:
         number_of_workers = _resolve_workers(NumberOfWorkers)
 
     payload = None
@@ -112,48 +133,186 @@ def tick() -> None:
 
     if number_of_workers > 0:
         reader, mreaders = payload
-        ctx = multiprocessing.get_context("spawn")
-        executor = ProcessPoolExecutor(
-            max_workers=min(number_of_workers, num_epochs), mp_context=ctx
+        docs_per_epoch = _ingest_on_pool(
+            daqreader,
+            reader,
+            mreaders,
+            epochFiles,
+            epochIds,
+            number_of_workers,
+            plans_segments,
+            update_progress,
         )
-        try:
-            futures = {
-                executor.submit(_ingestOneEpoch, reader, mreaders, ef, eid): k
-                for k, (ef, eid) in enumerate(zip(epochFiles, epochIds))
-            }
-            for future in as_completed(futures):
-                docs_per_epoch[futures[future]] = future.result()
-                tick()
-        except BaseException:
-            executor.shutdown(wait=True, cancel_futures=True)
-            raise
-        executor.shutdown(wait=True)
         return docs_per_epoch, True
 
     for k, (ef, eid) in enumerate(zip(epochFiles, epochIds)):
-        docs_per_epoch[k] = _ingestOneEpoch(daqreader, daqmetadatareaders, ef, eid)
-        tick()
+        epoch_progress = None
+        if plans_segments and ProgressFcn is not None:
+            epoch_progress = functools.partial(update_progress, k)
+        docs_per_epoch[k] = _ingestOneEpoch(
+            daqreader, daqmetadatareaders, ef, eid, epoch_progress=epoch_progress
+        )
+        update_progress(k, 1)
     return docs_per_epoch, False
 
 
 #: Snake-case alias for MATLAB's name; one function under two names.
 ingest_epoch_documents = ingestEpochDocuments
 
 
+def _ingest_on_pool(
+    daqreader: Any,
+    reader: Any,
+    mreaders: list[Any],
+    epoch_files: list[list[str]],
+    epoch_ids: list[str],
+    number_of_workers: int,
+    plans_segments: bool,
+    update_progress: Callable[[int, float], None],
+) -> list[list[Any]]:
+    """Ingest the epochs on worker processes, one segment per task.
+
+    MATLAB counterpart: ``ingestOnPool``. Every epoch is planned here
+    (ndi_daq_reader_mfdaq.ingest_epochfiles_plan reads only file headers and
+    writes the channel list); each planned segment, and each epoch's
+    metadata readers, is a task. Readers that do not plan segments run one
+    task per epoch. Tasks are handed out as workers free up, so the pool stays
+    busy however unequal the epochs are, and the segment files are attached
+    in plan order afterwards, so the documents match a serial ingest.
+    """
+    from ..mfdaq import ndi_daq_reader_mfdaq
+
+    num_epochs = len(epoch_files)
+    epoch_docs: list[Any] = [None] * num_epochs
+    plans: list[list[dict[str, Any]]] = [[] for _ in range(num_epochs)]
+    segment_files: list[list[Any]] = [[] for _ in range(num_epochs)]
+    meta_docs: list[list[Any]] = [[] for _ in range(num_epochs)]
+    docs_per_epoch: list[list[Any]] = [[] for _ in range(num_epochs)]
+
+    # (epoch, kind, index into the plan, weight)
+    tasks: list[tuple[int, str, int, float]] = []
+    for k in range(num_epochs):
+        if plans_segments:
+            epoch_docs[k], plans[k] = daqreader.ingest_epochfiles_plan(epoch_files[k], epoch_ids[k])
+            segment_files[k] = [None] * len(plans[k])
+            tasks.extend((k, "segment", t, e["weight"]) for t, e in enumerate(plans[k]))
+            if mreaders:
+                tasks.append((k, "metadata", 0, 0.0))
+        else:
+            tasks.append((k, "epoch", 0, 1.0))
+    remaining = [0] * num_epochs
+    for k, *_ in tasks:
+        remaining[k] += 1
+    weight_done = [0.0] * num_epochs
+
+    ctx = multiprocessing.get_context("spawn")
+    executor = ProcessPoolExecutor(max_workers=number_of_workers, mp_context=ctx)
+    try:
+        futures = {}
+        for index, (k, kind, t, _) in enumerate(tasks):
+            entry = plans[k][t] if kind == "segment" else None
+            future = executor.submit(
+                _run_task,
+                reader,
+                mreaders if kind != "segment" else [],
+                epoch_files[k],
+                epoch_ids[k],
+                kind,
+                entry,
+            )
+            futures[future] = index
+        pending = set(futures)
+        while pending:
+            finished, pending = wait(pending, return_when=FIRST_COMPLETED)
+            for future in finished:
+                k, kind, t, weight = tasks[futures[future]]
+                out = future.result()
+                if kind == "segment":
+                    segment_files[k][t] = out
+                elif kind == "metadata":
+                    meta_docs[k] = out
+                else:
+                    docs_per_epoch[k] = out
+                remaining[k] -= 1
+                weight_done[k] += weight
+                update_progress(k, 1 if remaining[k] == 0 else min(weight_done[k], 0.999))
+    except BaseException:
+        executor.shutdown(wait=True, cancel_futures=True)
+        raise
+    executor.shutdown(wait=True)
+    for k in range(num_epochs):
+        update_progress(k, 1)  # e.g., an epoch with nothing to write
+
+    if plans_segments:
+        for k in range(num_epochs):
+            d = ndi_daq_reader_mfdaq.ingest_attach_segments(
+                epoch_docs[k], plans[k], segment_files[k]
+            )
+            docs_per_epoch[k] = [d, *meta_docs[k]]
+    return docs_per_epoch
+
+
+def _run_task(
+    daqreader: Any,
+    daqmetadatareaders: list[Any],
+    ef: list[str],
+    epoch_id: str,
+    kind: str,
+    plan_entry: dict[str, Any] | None,
+) -> Any:
+    """One worker task: a segment, an epoch's metadata, or a whole epoch."""
+    if kind == "segment":
+        return daqreader.ingest_segment(ef, plan_entry)
+    if kind == "metadata":
+        return _metadata_docs(daqmetadatareaders, ef, epoch_id)
+    return _ingestOneEpoch(daqreader, daqmetadatareaders, ef, epoch_id)
+
+
+def _metadata_docs(daqmetadatareaders: list[Any], ef: list[str], epoch_id: str) -> list[Any]:
+    """The metadata readers' documents for one epoch."""
+    docs: list[Any] = []
+    for mreader in daqmetadatareaders:
+        new_d = mreader.ingest_epochfiles(ef, epoch_id)
+        docs.extend(new_d if isinstance(new_d, list) else [new_d])
+    return docs
+
+
 def _ingestOneEpoch(  # noqa: N802 (mirrors MATLAB's local function)
-    daqreader: Any, daqmetadatareaders: list[Any], ef: list[str], epoch_id: str
+    daqreader: Any,
+    daqmetadatareaders: list[Any],
+    ef: list[str],
+    epoch_id: str,
+    epoch_progress: Callable[[float], Any] | None = None,
 ) -> list[Any]:
     """Create the documents for a single epoch (no database access)."""
     docs: list[Any] = []
     if daqreader is not None:
-        new_d = daqreader.ingest_epochfiles(ef, epoch_id)
-        docs.extend(new_d if isinstance(new_d, list) else [new_d])
-    for mreader in daqmetadatareaders:
-        new_d = mreader.ingest_epochfiles(ef, epoch_id)
+        if epoch_progress is None:
+            new_d = daqreader.ingest_epochfiles(ef, epoch_id)
+        else:
+            new_d = daqreader.ingest_epochfiles(ef, epoch_id, ProgressFcn=epoch_progress)
         docs.extend(new_d if isinstance(new_d, list) else [new_d])
+    docs.extend(_metadata_docs(daqmetadatareaders, ef, epoch_id))
     return docs
 
 
+def _reader_plans_segments(daqreader: Any) -> bool:
+    """Does DAQREADER use ndi_daq_reader_mfdaq's ingest_epochfiles?
+
+    MATLAB counterpart: ``readerPlansSegments``. Such a reader takes a
+    ProgressFcn and can be ingested segment by segment
+    (ingest_epochfiles_plan / ingest_segment). A subclass that overrides
+    ingest_epochfiles keeps its own signature and is called once per epoch,
+    without one.
+    """
+    if daqreader is None:
+        return False
+    from ..mfdaq import ndi_daq_reader_mfdaq
+
+    method = getattr(type(daqreader), "ingest_epochfiles", None)
+    return method is ndi_daq_reader_mfdaq.ingest_epochfiles
+
+
 def _resolve_workers(override: int | None) -> int:
     """Worker count for this call, following NDI's parallel preferences."""
     from ndi.fun.utils import parallelWorkers

src/ndi/daq/fun/ndi_matlab_python_bridge.yaml

@@ -14,7 +14,7 @@ functions:
   - name: ingestEpochDocuments
     type: function
     matlab_path: "+ndi/+daq/+fun/ingestEpochDocuments.m"
-    matlab_last_sync_hash: "8feb9166e"
+    matlab_last_sync_hash: "569fdab67"
     python_path: "ndi/daq/fun/ingest_epoch_documents.py"
     input_arguments:
       - name: daqreader
@@ -31,7 +31,7 @@ functions:
         type_python: "list[str]"
       - name: ProgressFcn
         type_matlab: "function handle or []"
-        type_python: "Callable[[int, int], Any] | None"
+        type_python: "Callable[[float, int], Any] | None"
         default: "None"
       - name: NumberOfWorkers
         type_matlab: "double ([]: use the preferences)"
@@ -55,3 +55,24 @@ functions:
       falls back to serial with a warning. MATLAB's thread-pool refusal has
       no counterpart (Python workers here are always processes). A snake_case
       alias ingest_epoch_documents is bound to the same function.
+
+      NDI-matlab b192ae605 (VH-Lab/NDI-matlab#1089): ProgressFcn's numDone is
+      fractional while an epoch is underway. A reader whose ingest_epochfiles
+      is ndi_daq_reader_mfdaq's own is passed a ProgressFcn and reports each
+      segment it writes; the total is reported only when it changes, so each
+      whole number is reported once. Ported. MATLAB's DataQueue sends from
+      the workers become a multiprocessing.Manager queue that the caller
+      drains while it waits on the futures; the DefiningClass check becomes
+      an identity check on the ingest_epochfiles function.
+
+      NDI-matlab 569fdab67 (VH-Lab/NDI-matlab#1090) schedules the parallel
+      work by segment instead of by epoch, so every worker stays busy even
+      with one long epoch: each epoch whose reader plans segments (its
+      ingest_epochfiles is ndi_daq_reader_mfdaq's own) is planned in the
+      caller, and each segment, each metadata reader's work and each other
+      epoch become one task; segments are attached in plan order as each
+      epoch's tasks finish. The pool is used whenever there is an epoch, as
+      in MATLAB. Ported: parfeval + fetchNext becomes a spawn
+      ProcessPoolExecutor + as_completed, and progress is counted in the
+      caller from the plan weights, so the Manager queue of #1089's port is
+      gone.