diff --git a/CHANGELOG.md b/CHANGELOG.md index 363f5980..ae89a7f9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,24 @@ 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 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 + DAQ system -- on by default (`ShowProgress=False` turns it off; + `ProgressBar=` / `ProgressBarTag=` reuse a window). The per-epoch reader and + metadata-reader work moves to the new `ndi.daq.fun.ingestEpochDocuments`, + which runs it on spawned worker processes when NDI's parallel preferences + allow; database writes stay serial and in epoch order. To support this: + `ProgressBarWindow` gains MATLAB's silent (headless) mode -- `Silent`, + `isHeadless`, `silentModeDefault`, `resolveSilentMode` -- so a run with no + display or no PySide6 keeps the bookkeeping and builds no window instead of + Qt aborting the process, and `addBar` gains MATLAB's per-bar `Timeout`; and + the `Parallel.OpenPoolIfNone`, `Parallel.UseDefaultNumberOfWorkers` and + `Parallel.NumberOfWorkers` preferences are registered (shared with + NDI-matlab) and read by the new `ndi.fun.parallelWorkers`. Python has no + pool that outlives a call, so with the default `OpenPoolIfNone=false` + ingestion is serial. + - **Navigator: DAQ system browser, cloud dataset id, Assign to workspace, and faster DAQ system loading** (mirroring VH-Lab/NDI-matlab#1058). "Session > DAQ Systems..." opens the new `ndi.gui.nav.DaqSystemBrowser`, a diff --git a/src/ndi/daq/fun/__init__.py b/src/ndi/daq/fun/__init__.py new file mode 100644 index 00000000..d14e6b8f --- /dev/null +++ b/src/ndi/daq/fun/__init__.py @@ -0,0 +1,14 @@ +""" +ndi.daq.fun - Helper functions for ndi.daq systems. + +MATLAB equivalent: +ndi/+daq/+fun/ + +Provides :func:`~ndi.daq.fun.ingest_epoch_documents.ingestEpochDocuments`, +the per-epoch work of ingestion, run serially or on worker processes. +""" + +from __future__ import annotations + +from .ingest_epoch_documents import ingest_epoch_documents, ingestEpochDocuments + +__all__ = ["ingestEpochDocuments", "ingest_epoch_documents"] diff --git a/src/ndi/daq/fun/ingest_epoch_documents.py b/src/ndi/daq/fun/ingest_epoch_documents.py new file mode 100644 index 00000000..235897fe --- /dev/null +++ b/src/ndi/daq/fun/ingest_epoch_documents.py @@ -0,0 +1,190 @@ +"""ndi.daq.fun.ingest_epoch_documents - build the ingested-data documents for a set of epochs. + +MATLAB counterpart: ``+ndi/+daq/+fun/ingestEpochDocuments.m`` + +This is the per-epoch part of :meth:`ndi.daq.system.ndi_daq_system.ingest`: +each epoch's reader and metadata-reader ``ingest_epochfiles`` documents. No +database is touched here, so the epochs can be computed on worker processes +while the caller keeps every database write serial. + +WHY READERS ARE DETACHED FROM THEIR SESSION FOR THE WORKERS +MATLAB's readers are value objects that hold no session, so a parfor copies +them to the workers as they are. The Python readers keep a ``_session`` +reference (used only on the ingested-READ paths), and a session -- with its +open SQLite database -- cannot and must not be shipped to another process. +Each worker therefore gets a shallow copy of each reader with ``_session`` +cleared; ``ingest_epochfiles`` never uses it. +""" + +from __future__ import annotations + +import copy +import multiprocessing +import pickle +import warnings +from collections.abc import Callable +from concurrent.futures import ProcessPoolExecutor, as_completed +from typing import Any + +from pydantic import ConfigDict, validate_call + +__all__ = ["ingestEpochDocuments", "ingest_epoch_documents"] + + +@validate_call(config=ConfigDict(arbitrary_types_allowed=True)) +def ingestEpochDocuments( # noqa: N802 (MATLAB mirror) + daqreader: Any, + daqmetadatareaders: list[Any], + epochFiles: list[list[str]], + epochIds: list[str], + ProgressFcn: Callable[[int, int], Any] | None = None, + NumberOfWorkers: int | None = None, +) -> tuple[list[list[Any]], bool]: + """Build the ingested-data documents for a set of epochs. + + [DOCSPEREPOCH, USEDPARALLEL] = ndi.daq.fun.ingestEpochDocuments(DAQREADER, ... + DAQMETADATAREADERS, EPOCHFILES, EPOCHIDS, ...) + + Runs DAQREADER.ingest_epochfiles and each DAQMETADATAREADERS[j].ingest_epochfiles + for every epoch described by EPOCHFILES[k] (a list of file names) and + EPOCHIDS[k] (the epoch id). + + DOCSPEREPOCH[k] is the list of ndi_document objects created for epoch k, in + the same order as a serial ingest produces them (the daqreader document + followed by one document per metadata reader). + + No database is touched here: the documents are NOT added to any database, + so the caller remains responsible for serializing all database writes. + + Whether the epochs are computed on worker processes follows NDI's parallel + preferences, through ndi.fun.parallelWorkers. Results do not depend on the + number of workers. + + Args: + daqreader: The ndi.daq.reader whose ingest_epochfiles makes the main + document for each epoch. + daqmetadatareaders: The ndi.daq.metadatareader objects to run per epoch. + 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. + NumberOfWorkers: Overrides the preferences for this call (passed to + ndi.fun.parallelWorkers); 0 forces serial. + + Returns: + ``(docsPerEpoch, usedParallel)``; usedParallel is True if the epochs + were actually computed on worker processes. + + Python-specific Notes: + Workers are ``concurrent.futures.ProcessPoolExecutor`` processes started + with the ``spawn`` method -- fresh interpreters, like MATLAB's pool + 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. + """ + 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 + if ProgressFcn is not None: + try: + ProgressFcn(done, num_epochs) + except Exception as exc: # noqa: BLE001 - progress must never break an ingest + warnings.warn(f"Progress reporting failed: {exc}", stacklevel=3) + + number_of_workers = 0 + if num_epochs > 1: + number_of_workers = _resolve_workers(NumberOfWorkers) + + payload = None + if number_of_workers > 0: + payload = _detached_readers(daqreader, daqmetadatareaders) + if payload is None: + number_of_workers = 0 + + 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 + ) + 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() + return docs_per_epoch, False + + +#: Snake-case alias for MATLAB's name; one function under two names. +ingest_epoch_documents = ingestEpochDocuments + + +def _ingestOneEpoch( # noqa: N802 (mirrors MATLAB's local function) + daqreader: Any, daqmetadatareaders: list[Any], ef: list[str], epoch_id: str +) -> 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) + docs.extend(new_d if isinstance(new_d, list) else [new_d]) + return docs + + +def _resolve_workers(override: int | None) -> int: + """Worker count for this call, following NDI's parallel preferences.""" + from ndi.fun.utils import parallelWorkers + + if override is None: + return parallelWorkers() + if override < 1: + return 0 # forced serial + return parallelWorkers(numberOfWorkers=override) + + +def _detach(obj: Any) -> Any: + """A shallow copy of *obj* with its session reference cleared.""" + if obj is None: + return None + clone = copy.copy(obj) + if hasattr(clone, "_session"): + clone._session = None + return clone + + +def _detached_readers(daqreader: Any, daqmetadatareaders: list[Any]) -> tuple[Any, list] | None: + """Session-free, picklable copies of the readers, or None to run serially.""" + reader = _detach(daqreader) + mreaders = [_detach(m) for m in daqmetadatareaders] + try: + pickle.dumps((reader, mreaders)) + except Exception as exc: # noqa: BLE001 - any pickling failure means "run serially" + warnings.warn( + f"The DAQ readers cannot be sent to worker processes ({exc}). Ingesting serially.", + stacklevel=3, + ) + return None + return reader, mreaders diff --git a/src/ndi/daq/fun/ndi_matlab_python_bridge.yaml b/src/ndi/daq/fun/ndi_matlab_python_bridge.yaml new file mode 100644 index 00000000..3617093c --- /dev/null +++ b/src/ndi/daq/fun/ndi_matlab_python_bridge.yaml @@ -0,0 +1,57 @@ +# ndi_matlab_python_bridge.yaml — src/ndi/daq/fun/ +# The Primary Contract for the ndi.daq.fun namespace. + +project_metadata: + bridge_version: "1.1" + naming_policy: "Strict MATLAB Mirror" + indexing_policy: "Semantic Parity (1-based for user concepts, 0-based for internal data)" + +# ========================================================================= +# Standalone functions +# ========================================================================= +functions: + + - name: ingestEpochDocuments + type: function + matlab_path: "+ndi/+daq/+fun/ingestEpochDocuments.m" + matlab_last_sync_hash: "8feb9166e" + python_path: "ndi/daq/fun/ingest_epoch_documents.py" + input_arguments: + - name: daqreader + type_matlab: "ndi.daq.reader" + type_python: "Any" + - name: daqmetadatareaders + type_matlab: "cell array of ndi.daq.metadatareader" + type_python: "list[Any]" + - name: epochFiles + type_matlab: "cell array of cell arrays of char" + type_python: "list[list[str]]" + - name: epochIds + type_matlab: "cell array of char" + type_python: "list[str]" + - name: ProgressFcn + type_matlab: "function handle or []" + type_python: "Callable[[int, int], Any] | None" + default: "None" + - name: NumberOfWorkers + type_matlab: "double ([]: use the preferences)" + type_python: "int | None" + default: "None" + output_arguments: + - name: docsPerEpoch + type_python: "list[list[Any]]" + - name: usedParallel + type_python: "bool" + decision_log: > + Ported (VH-Lab/NDI-matlab#1079). Runs each epoch's reader and + metadata-reader ingest_epochfiles, serially or on workers as + ndi.fun.parallelWorkers allows, with ProgressFcn(numDone, numTotal) + called in the calling process after each epoch and the result always in + epoch order. MATLAB's parfor + parallel.pool.DataQueue becomes a spawn + ProcessPoolExecutor + as_completed. Two Python-specific points: the + readers keep a _session reference that MATLAB's value-class readers do + not have, so workers get shallow copies with it cleared + (ingest_epochfiles never uses it); and a reader that cannot be pickled + 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. diff --git a/src/ndi/daq/ndi_matlab_python_bridge.yaml b/src/ndi/daq/ndi_matlab_python_bridge.yaml index f57146f7..f50dfe3e 100644 --- a/src/ndi/daq/ndi_matlab_python_bridge.yaml +++ b/src/ndi/daq/ndi_matlab_python_bridge.yaml @@ -756,7 +756,7 @@ classes: - name: system type: class matlab_path: "+ndi/+daq/system.m" - matlab_last_sync_hash: "e066e8d" + matlab_last_sync_hash: "375a0ff7e" python_path: "ndi/daq/system.py" python_class: "ndi_daq_system" inherits: "ndi.ido" @@ -764,6 +764,19 @@ classes: 2026-10-24: NDI-matlab 989a620 added an optional prefetched-documents input to the constructor and e066e8d reverted it (VH-Lab/NDI-matlab#1058); the file is unchanged from fe64a9f5 in net, so nothing to follow. + 2026-10-27: NDI-matlab 96430ab81 and 8feb9166e (VH-Lab/NDI-matlab#1079) + give ingest a progress bar and parallel per-epoch work. Ported: ingest + takes ShowProgress (default True) and ProgressBar, collects the + uningested epochs, makes the file navigator documents serially, hands + the per-epoch reader/metadata-reader work to + ndi.daq.fun.ingestEpochDocuments (worker processes when the Parallel.* + preferences allow), and keeps the database add serial and in epoch + order. The hidden static progress helpers addIngestProgressBar, + updateIngestProgressBar and removeIngestProgressBar are ported as static + methods. Python ingest keeps its existing guard for a missing + filenavigator or session. + NDI-matlab 375a0ff7e only preallocates in ingest's loops (Code Analyzer); + nothing to follow. properties: - name: name @@ -936,13 +949,81 @@ classes: Semantic Parity: epoch and channel are 1-indexed (user concepts). - name: ingest - input_arguments: [] + input_arguments: + - name: ShowProgress + type_matlab: "logical" + type_python: "bool" + default: "True" + - name: ProgressBar + type_matlab: "ndi.gui.component.ProgressBarWindow or []" + type_python: "ndi_gui_component_ProgressBarWindow | None" + default: "None" output_arguments: - name: success type_python: "bool" - name: documents type_python: "list[Any]" - decision_log: "Exact match. Returns 2-tuple." + decision_log: > + Exact match. Returns 2-tuple. The name/value options are + keyword-only arguments. Parallel use follows the Parallel.* + preferences; there is no per-call parallel option, as in MATLAB. + + - name: addIngestProgressBar + kind: static + input_arguments: + - name: progressBar + type_matlab: "ndi.gui.component.ProgressBarWindow or []" + type_python: "ndi_gui_component_ProgressBarWindow | None" + - name: label + type_matlab: "char" + type_python: "str" + - name: Tag + type_matlab: "char" + type_python: "str" + default: "'' (a new unique id)" + - name: Auto + type_matlab: "logical" + type_python: "bool" + default: "True" + output_arguments: + - name: progressBar + type_python: "ndi_gui_component_ProgressBarWindow | None" + - name: barTag + type_python: "str" + decision_log: > + Exact match (Hidden in MATLAB). Opens or reuses an 'NDI Ingest' + window when none is given; the bar never times out; any failure + warns and returns (None, ""). + + - name: updateIngestProgressBar + kind: static + input_arguments: + - name: progressBar + type_matlab: "ndi.gui.component.ProgressBarWindow or []" + type_python: "ndi_gui_component_ProgressBarWindow | None" + - name: barTag + type_matlab: "char" + type_python: "str" + - name: fraction + type_matlab: "double" + type_python: "float" + output_arguments: [] + decision_log: > + Exact match (Hidden in MATLAB). A bar the user closed is skipped. + MATLAB also skips a bar in state 'Closed'; Python's removeBar drops + the row instead, so "not found" covers that case. + + - name: removeIngestProgressBar + kind: static + input_arguments: + - name: progressBar + type_matlab: "ndi.gui.component.ProgressBarWindow or []" + type_python: "ndi_gui_component_ProgressBarWindow | None" + - name: barTag + type_matlab: "char" + type_python: "str" + output_arguments: [] + decision_log: "Exact match (Hidden in MATLAB). Used on error; never raises." - name: deleteepoch input_arguments: diff --git a/src/ndi/daq/system.py b/src/ndi/daq/system.py index 471f001a..7a11b89f 100644 --- a/src/ndi/daq/system.py +++ b/src/ndi/daq/system.py @@ -605,55 +605,171 @@ def getmetadata( return reader.readmetadata(epochfiles) - def ingest(self) -> tuple[bool, list[Any]]: + def ingest( + self, + *, + ShowProgress: bool = True, + ProgressBar: Any = None, + ) -> tuple[bool, list[Any]]: """ Ingest data from this DAQ system into the database. + MATLAB equivalent: ndi.daq.system/ingest + + [B,D] = INGEST(NDI_DAQSYSTEM_OBJ, ...) + + Ingest any uningested epochs. The per-epoch work runs on worker + processes when NDI's parallel preferences allow it (see + ndi.fun.parallelWorkers; by default it is serial). All database writes + are made serially in the calling process after the epochs are computed, + and the documents are added in the same order as a serial ingest. + + Args: + ShowProgress: Show a progress bar using + ndi.gui.component.ProgressBarWindow (silent automatically when + there is no display). + ProgressBar: An existing ProgressBarWindow to add the bar to. None + opens/reuses one titled 'NDI Ingest'. + Returns: Tuple of (success, documents): - success: True if successful - documents: List of created documents """ + from .fun import ingestEpochDocuments + et = self.epochtable() - docs = [] - filenavigator_ingested = False + docs: list[Any] = [] + # find the epochs that still need to be ingested + epoch_files: list[list[str]] = [] + epoch_ids: list[str] = [] for entry in et: underlying = entry.get("underlying_epochs", {}) epochfiles = underlying.get("underlying", []) + if not self._is_ingested(epochfiles): + epoch_files.append(list(epochfiles)) + epoch_ids.append(entry["epoch_id"]) + + progress_bar, bar_tag = None, "" + if ShowProgress and epoch_files: + progress_bar, bar_tag = self.addIngestProgressBar( + ProgressBar, f"Ingesting {self.name} ({len(epoch_files)} epochs)" + ) + # the database add is one more step after all of the epochs + num_steps = len(epoch_files) + 1 + + try: + if epoch_files: + # the file navigator documents are made once, serially; they + # need the session and its database + if self._filenavigator is not None: + docs.extend(self._filenavigator.ingest()) + + docs_per_epoch, _ = ingestEpochDocuments( + self._daqreader, + list(self._daqmetadatareaders), + epoch_files, + epoch_ids, + ProgressFcn=lambda num_done, num_total: self.updateIngestProgressBar( + progress_bar, bar_tag, num_done / num_steps + ), + ) + for epoch_docs in docs_per_epoch: + docs.extend(epoch_docs) - # Check if already ingested - if self._is_ingested(epochfiles): - continue + # Set session IDs and add to database + if self.session is not None: + session_id = self.session.id() + for doc in docs: + doc.set_session_id(session_id) + self.session.database_add(docs) + except BaseException: + self.removeIngestProgressBar(progress_bar, bar_tag) + raise - # Ingest file navigator first - if not filenavigator_ingested and self._filenavigator is not None: - nav_docs = self._filenavigator.ingest() - docs.extend(nav_docs) - filenavigator_ingested = True + self.updateIngestProgressBar(progress_bar, bar_tag, 1.0) + return True, docs - # Ingest DAQ reader data - if self._daqreader is not None: - reader_docs = self._daqreader.ingest_epochfiles(epochfiles, entry["epoch_id"]) - if not isinstance(reader_docs, list): - reader_docs = [reader_docs] - docs.extend(reader_docs) + # -- Ingestion progress helpers (MATLAB: methods (Static, Hidden)) ------- + # Shared with ndi_session.ingest. Progress reporting must never break an + # ingest, so every call is guarded. + + @staticmethod + def addIngestProgressBar( + progressBar: Any, + label: str, + Tag: str = "", + Auto: bool = True, + ) -> tuple[Any, str]: + """Add an ingestion bar to a ProgressBarWindow. + + MATLAB equivalent: ndi.daq.system.addIngestProgressBar + + Adds a bar labeled *label* to *progressBar*. If *progressBar* is None, a + window titled 'NDI Ingest' is opened (or reused). Returns the window and + the tag of the new bar; on any failure both are empty and a warning is + issued. The bar never times out (ingesting one epoch can take longer + than the window's default timeout). + """ + import math + import warnings + + bar_tag = Tag or ndi_ido.unique_id() + try: + if progressBar is None: + from ..gui.component.ProgressBarWindow import ( + ndi_gui_component_ProgressBarWindow, + ) - # Ingest metadata - for mreader in self._daqmetadatareaders: - m_docs = mreader.ingest_epochfiles(epochfiles, entry["epoch_id"]) - if not isinstance(m_docs, list): - m_docs = [m_docs] - docs.extend(m_docs) + progressBar = ndi_gui_component_ProgressBarWindow("NDI Ingest") + progressBar.addBar(Label=label, Tag=bar_tag, Auto=Auto, Timeout=math.inf) + except Exception as exc: # noqa: BLE001 - no bar is not a failed ingest + warnings.warn(f"Could not create an ingestion progress bar: {exc}", stacklevel=2) + return None, "" + return progressBar, bar_tag - # Set session IDs and add to database - if self.session is not None: - session_id = self.session.id() - for doc in docs: - doc.set_session_id(session_id) - self.session.database_add(docs) + #: Snake-case alias for MATLAB's name; one method under two names. + add_ingest_progress_bar = addIngestProgressBar - return True, docs + @staticmethod + def updateIngestProgressBar(progressBar: Any, barTag: str, fraction: float) -> None: + """Set an ingestion bar's progress; no-op if the bar is gone. + + MATLAB equivalent: ndi.daq.system.updateIngestProgressBar + """ + import warnings + + if progressBar is None or not barTag: + return + try: + idx, _ = progressBar.getBarNum(barTag) + if idx is None: + return # closed by the user, or removed on completion + progressBar.updateBar(barTag, max(0.0, min(1.0, float(fraction)))) + except Exception as exc: # noqa: BLE001 - progress must never break an ingest + warnings.warn(f"Could not update the ingestion progress bar: {exc}", stacklevel=2) + + #: Snake-case alias for MATLAB's name; one method under two names. + update_ingest_progress_bar = updateIngestProgressBar + + @staticmethod + def removeIngestProgressBar(progressBar: Any, barTag: str) -> None: + """Remove an ingestion bar (used on error). + + MATLAB equivalent: ndi.daq.system.removeIngestProgressBar + """ + if progressBar is None or not barTag: + return + try: + idx, _ = progressBar.getBarNum(barTag) + if idx is not None: + progressBar.removeBar(barTag) + except Exception: # noqa: BLE001 - the ingest error is what matters + pass + + #: Snake-case alias for MATLAB's name; one method under two names. + remove_ingest_progress_bar = removeIngestProgressBar def _is_ingested(self, epochfiles: list[str]) -> bool: """Check if epochfiles indicate an ingested epoch.""" diff --git a/src/ndi/fun/__init__.py b/src/ndi/fun/__init__.py index 335a36f6..3b79a1af 100644 --- a/src/ndi/fun/__init__.py +++ b/src/ndi/fun/__init__.py @@ -26,6 +26,8 @@ identifier, name2variable_name, name2variableName, + parallel_workers, + parallelWorkers, pseudorandomint, timestamp, ) @@ -37,6 +39,8 @@ "plot_extracellular_spikeshapes", "name2variableName", "name2variable_name", + "parallelWorkers", + "parallel_workers", "parseText", "parse_text", "probe", diff --git a/src/ndi/fun/ndi_matlab_python_bridge.yaml b/src/ndi/fun/ndi_matlab_python_bridge.yaml index 52582357..903d3e39 100644 --- a/src/ndi/fun/ndi_matlab_python_bridge.yaml +++ b/src/ndi/fun/ndi_matlab_python_bridge.yaml @@ -3348,18 +3348,43 @@ not_applicable: MATLAB's library paths. - name: parallelWorkers + type: function matlab_path: "+ndi/+fun/parallelWorkers.m" - matlab_last_sync_hash: "9a9150e7" - status: matlab_only + matlab_last_sync_hash: "9a9150e7c" + python_path: "ndi/fun/utils.py" + input_arguments: + - name: numberOfWorkers + type_matlab: "double ([]: use the preference)" + type_python: "int | float | None" + default: "None" + - name: useDefaultNumberOfWorkers + type_matlab: "logical ([]: use the preference)" + type_python: "bool | None" + default: "None" + - name: openPoolIfNone + type_matlab: "logical ([]: use the preference)" + type_python: "bool | None" + default: "None" + output_arguments: + - name: numberOfWorkers + type_python: "int" decision_log: > - Not applicable. Returns the worker limit to pass as the second - argument of a MATLAB `parfor`, reading the Parallel.OpenPoolIfNone and - Parallel.NumberOfWorkers preferences. Python has no parfor: parallel - code here uses multiprocessing or joblib, which take a worker count - directly, so there is no argument for this function to compute. The - preferences themselves stay meaningful on the MATLAB side; a Python - caller that grows a worker pool should read them through - ndi.preferences rather than reviving this signature. + Ported 2026-10-27, reversing the earlier matlab_only decision because a + Python caller now grows a worker pool: ndi_session.ingest / + ndi.daq.fun.ingestEpochDocuments, mirroring NDI-matlab's parallel ingest + (VH-Lab/NDI-matlab#1079). The preference reading, the per-call + overrides (giving numberOfWorkers implies the default is not wanted), + the bad-count warning and the fallback to the shipped defaults when the + preferences cannot be read all match. ONE DIFFERENCE, FROM THE + LANGUAGES RATHER THAN THE PORT: a MATLAB parpool outlives the call, so + MATLAB uses a pool the user opened earlier; Python workers live only + for the call that starts them, so there is never "a pool already open". + With OpenPoolIfNone false the answer is therefore always 0 (serial); + with it true it is NumberOfWorkers when UseDefaultNumberOfWorkers is + off, else the number of CPUs available to the process (what a default + MATLAB pool gets). The Parallel Computing Toolbox checks have no + counterpart: concurrent.futures is in the standard library. A + snake_case alias parallel_workers is bound to the same function. - name: assertAddonOnPath matlab_path: "+ndi/+fun/assertAddonOnPath.m" diff --git a/src/ndi/fun/utils.py b/src/ndi/fun/utils.py index 7afc079f..56e1b79f 100644 --- a/src/ndi/fun/utils.py +++ b/src/ndi/fun/utils.py @@ -3,11 +3,12 @@ MATLAB equivalents: +ndi/+fun/channelname2prefixnumber.m, name2variableName.m, pseudorandomint.m, timestamp.m, - find_calc_directories.m + find_calc_directories.m, parallelWorkers.m """ from __future__ import annotations +import os import random import re import warnings @@ -219,3 +220,98 @@ def find_calc_directories() -> list[Path]: if directory not in d: d.append(directory) return d + + +@validate_call +def parallelWorkers( + numberOfWorkers: int | float | None = None, + useDefaultNumberOfWorkers: bool | None = None, + openPoolIfNone: bool | None = None, +) -> int: + """How many workers a parallel loop should be allowed. + + MATLAB equivalent: +ndi/+fun/parallelWorkers.m + + N = parallelWorkers() + N = parallelWorkers(numberOfWorkers=N, openPoolIfNone=TF) + + Returns the number of worker processes a parallel loop may use, so that a + calculation honours the user's parallel preferences. N is 0 when the loop + should run serially in the calling process. Results never depend on N; + only the time taken and the memory used do. + + The preferences, shared with NDI-matlab through ~/.ndi/NDI_Preferences.json + and editable in the preferences editor: + + * ``Parallel.OpenPoolIfNone`` -- start workers when none are running. + Default False, so NDI never starts workers on its own. + * ``Parallel.UseDefaultNumberOfWorkers`` -- let NDI decide how many + workers (one per available core). Default True. + * ``Parallel.NumberOfWorkers`` -- how many workers to allow when + UseDefaultNumberOfWorkers is off. Ignored while it is on. + + Any of the three may be overridden for a single call with the keyword of + the same name. Giving numberOfWorkers implies that the default is not + wanted, so useDefaultNumberOfWorkers does not also have to be given. + + Python-specific Notes + --------------------- + MATLAB's parpool stays open between calls, so with the default + preferences a MATLAB calculation uses a pool the user opened earlier. + Python has no such pool: workers live only for the duration of the call + that starts them (e.g. a ``concurrent.futures.ProcessPoolExecutor``). There + is therefore never "a pool already open", and with OpenPoolIfNone False + this returns 0 (serial). With it True, the result is NumberOfWorkers when + UseDefaultNumberOfWorkers is off, otherwise the number of CPUs available to + this process -- what a MATLAB pool opened with the default size gets. + """ + try: + from ndi import preferences as _preferences + + open_pool = bool(_preferences.get("Parallel.OpenPoolIfNone")) + use_default = bool(_preferences.get("Parallel.UseDefaultNumberOfWorkers")) + preferred = _preferences.get("Parallel.NumberOfWorkers") + except Exception: # noqa: BLE001 - an unreadable preferences file must not fail a calculation + open_pool = False + use_default = True + preferred = 0 + + if openPoolIfNone is not None: + open_pool = bool(openPoolIfNone) + if useDefaultNumberOfWorkers is not None: + use_default = bool(useDefaultNumberOfWorkers) + if numberOfWorkers is not None: + # Naming a number is itself a statement that the default is not wanted. + preferred = numberOfWorkers + use_default = False + + # max_workers is this function's own shorthand, never a preference value: + # 0 means impose no limit. + if use_default: + max_workers = 0 + else: + try: + max_workers = round(float(preferred)) + except (TypeError, ValueError): + max_workers = 0 + if max_workers < 1: + warnings.warn( + f"Parallel.NumberOfWorkers is {preferred}, which is not a number of " + "workers. Using the default instead.", + stacklevel=2, + ) + max_workers = 0 + + # No pool outlives a call in Python, so there is never one to reuse. + if not open_pool: + return 0 + + if max_workers > 0: + return max_workers + if hasattr(os, "sched_getaffinity"): + return max(len(os.sched_getaffinity(0)), 1) + return max(os.cpu_count() or 1, 1) + + +#: Snake-case alias for MATLAB's name; one function under two names. +parallel_workers = parallelWorkers diff --git a/src/ndi/gui/component/ProgressBarWindow.py b/src/ndi/gui/component/ProgressBarWindow.py index 7e9f4442..d262a39b 100644 --- a/src/ndi/gui/component/ProgressBarWindow.py +++ b/src/ndi/gui/component/ProgressBarWindow.py @@ -7,6 +7,16 @@ Each bar shows a label, a coloured progress bar, a percentage, an estimated-time-remaining label, and a close button. +SILENT (HEADLESS) MODE +Mirrors MATLAB's silent mode. Building a Qt window needs PySide6 and a usable +display; without a display Qt typically aborts the whole process, which no +try/except can catch. So, as in MATLAB, the window runs silently whenever it +detects a headless context: no widgets are created and every graphics +touchpoint is skipped, while the bookkeeping (addBar, updateBar, removeBar, +getBarNum, getState, timeouts) behaves exactly as it does with a window. The +decision is made once per construction by resolveSilentMode, in this order: +the ``Silent`` argument, the process-wide silentModeDefault, then isHeadless. + CROSS-LANGUAGE NAMING Methods mirroring MATLAB keep MATLAB's exact name and also carry a snake_case alias bound to the same function -- one method under two names, @@ -16,11 +26,14 @@ from __future__ import annotations +import math +import os import random +import sys import time import warnings -from ndi.gui._qt_helpers import get_or_create_app, require_qt +from ndi.gui._qt_helpers import _QT_AVAILABLE, get_or_create_app, require_qt try: from PySide6 import QtCore, QtWidgets @@ -30,6 +43,13 @@ # Module-level registry of existing windows (mirrors MATLAB findall) _ACTIVE_WINDOWS: dict[str, ndi_gui_component_ProgressBarWindow] = {} +# Process-wide silent-mode override (mirrors MATLAB's persistent variable in +# silentModeDefault). None means "decide by detection". +_SILENT_OVERRIDE: bool | None = None + +# Sentinel distinguishing "no argument" from None in silentModeDefault. +_NOT_GIVEN = object() + class _BarRecord: """Internal bookkeeping for one progress bar row.""" @@ -47,14 +67,24 @@ class _BarRecord: "color", "clock_start", "clock_last", + # Per-bar timeout in seconds (MATLAB addBar's Timeout). None inherits + # the window's Timeout; math.inf never times out. + "Timeout", # The row's container widget, so removeBar can take the whole row # out. It is in __slots__ because addBar assigns it: without it, # every addBar raised AttributeError and no bar could be drawn. "_frame", ) - def __init__(self, tag: str, auto: bool, color: tuple[float, float, float]): + def __init__( + self, + tag: str, + auto: bool, + color: tuple[float, float, float], + timeout: float | None = None, + ): self.Tag = tag + self.Timeout = timeout self.Progress: float = 0.0 self.State: str = "Open" self.Auto = auto @@ -82,6 +112,11 @@ class ndi_gui_component_ProgressBarWindow: If ``True``, close any existing window with the same *title*. AutoDelete : bool If ``True``, auto-close the window when all bars are closed. + Silent : bool | None + Forces silent (headless) mode on or off for this window, bypassing + detection. ``None`` (the default) decides automatically -- see + :meth:`resolveSilentMode`. In silent mode no Qt widgets are created + (PySide6 is not even required) and the bars exist as bookkeeping only. """ def __init__( @@ -90,7 +125,25 @@ def __init__( *, Overwrite: bool = False, AutoDelete: bool = True, + Silent: bool | None = None, ) -> None: + # Decide once, here, whether to run headless; every method below asks + # the object rather than re-detecting. + self.Silent: bool = self.resolveSilentMode(Silent) + + if self.Silent: + # No window to find or reuse (MATLAB: openFigs is empty when + # silent), so a silent window is always a fresh object. + self.ScreenFrac = 0.025 + self.Timeout = 60.0 # seconds + self.AutoDelete = AutoDelete + self._title = title + self._bars = [] + self._app = None + self._window = None + self._layout = None + return + require_qt() self._app = get_or_create_app() @@ -135,6 +188,7 @@ def addBar( Tag: str = "", Color: tuple[float, float, float] = (1.0, 1.0, 1.0), Auto: bool = False, + Timeout: float | None = None, ) -> None: """Add a new progress bar row. @@ -148,6 +202,11 @@ def addBar( RGB colour for the bar. Auto : bool Auto-remove bar on completion or timeout. + Timeout : float | None + Per-bar timeout in seconds, overriding the window's ``Timeout`` + for this bar only. ``None`` (the default) inherits the window's + value; ``math.inf`` opts the bar out of timing out -- appropriate + for a bar that ticks less often than the window timeout. """ if not Tag: Tag = Label @@ -176,7 +235,12 @@ def addBar( if 1.5 <= s <= 2.8: break - rec = _BarRecord(Tag, Auto, Color) + rec = _BarRecord(Tag, Auto, Color, Timeout) + + if self.Silent: + # Bookkeeping only; no widgets. + self._bars.append(rec) + return # Build widgets row_frame = QtWidgets.QFrame() @@ -259,11 +323,13 @@ def updateBar(self, barID: int | str, progress: float) -> None: rec = self._bars[idx] try: rec.Progress = progress - rec.bar_widget.setValue(int(progress * 1000)) - rec.pct_widget.setText(f"{progress * 100:.0f}%") + if rec.bar_widget is not None: + rec.bar_widget.setValue(int(progress * 1000)) + if rec.pct_widget is not None: + rec.pct_widget.setText(f"{progress * 100:.0f}%") rec.clock_last = time.monotonic() - if 0 < progress < 1: + if 0 < progress < 1 and rec.timer_widget is not None: elapsed = rec.clock_last - rec.clock_start remaining = elapsed * (1.0 - progress) / progress rec.timer_widget.setText(f"Estimated time: {self._fmt_time(remaining)}") @@ -407,7 +473,10 @@ def _check_timeout(self) -> list[int]: continue if rec.Progress >= 1.0: continue - if now - rec.clock_last >= self.Timeout: + timeout = self.Timeout if rec.Timeout is None else rec.Timeout + if math.isinf(timeout): + continue + if now - rec.clock_last >= timeout: rec.State = "Timeout" if rec.btn_widget is not None: rec.btn_widget.setText("\u26a0") # ⚠ @@ -435,11 +504,15 @@ def _on_button(self, tag: str) -> None: self.removeBar(tag) def _resize_window(self) -> None: + if self._window is None: + return n = max(len(self._bars), 1) self._window.resize(500, 60 * n + 20) def _deleteIfNoOpenBars(self) -> None: if not self._bars: + if self._window is None: + return self._window.close() if self._title in _ACTIVE_WINDOWS: del _ACTIVE_WINDOWS[self._title] @@ -451,3 +524,67 @@ def _fmt_time(seconds: float) -> str: if seconds < 7200: return f"{seconds / 60:.0f} minutes" return f"{seconds / 3600:.0f} hours" + + # -- Silent (headless) mode ------------------------------------------ + + @staticmethod + def isHeadless() -> bool: + """Is there no usable display for a Qt window? + + Mirrors MATLAB ``isHeadless``. MATLAB asks whether it was started + with ``-batch`` or, on Linux, has an empty ``DISPLAY``. Python has no + ``-batch``; the Python-specific signals are: + + * PySide6 is not installed -- there is nothing to draw with; + * on Linux, neither ``DISPLAY`` nor ``WAYLAND_DISPLAY`` is set, unless + ``QT_QPA_PLATFORM`` names a platform that needs no display + (``offscreen`` or ``minimal``), in which case Qt can still build + widgets safely. + """ + if not _QT_AVAILABLE: + return True + if sys.platform.startswith("linux"): + platform = os.environ.get("QT_QPA_PLATFORM", "").strip().lower() + if platform in ("offscreen", "minimal"): + return False + return not (os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY")) + return False + + #: Snake-case alias for MATLAB's name; one method under two names. + is_headless = isHeadless + + @staticmethod + def silentModeDefault(newValue: object = _NOT_GIVEN) -> bool | None: + """Get or set the process-wide silent-mode override. + + Mirrors MATLAB ``silentModeDefault``. Called with no argument, it + returns the current override (``None`` means "decide by detection"). + Called with ``True``/``False`` it forces silent mode on/off for every + window constructed afterwards; called with ``None`` it clears the + override. + """ + global _SILENT_OVERRIDE + if newValue is not _NOT_GIVEN: + _SILENT_OVERRIDE = None if newValue is None else bool(newValue) + return _SILENT_OVERRIDE + + #: Snake-case alias for MATLAB's name; one method under two names. + silent_mode_default = silentModeDefault + + @classmethod + def resolveSilentMode(cls, explicitValue: bool | None = None) -> bool: + """Decide whether a new window runs silently. + + Mirrors MATLAB ``resolveSilentMode``: the explicit value when given, + else the process-wide :meth:`silentModeDefault`, else + :meth:`isHeadless`. + """ + if explicitValue is not None: + return bool(explicitValue) + override = cls.silentModeDefault() + if override is not None: + return bool(override) + return cls.isHeadless() + + #: Snake-case alias for MATLAB's name; one method under two names. + resolve_silent_mode = resolveSilentMode diff --git a/src/ndi/gui/component/ndi_matlab_python_bridge.yaml b/src/ndi/gui/component/ndi_matlab_python_bridge.yaml index b3720930..04fe7d65 100644 --- a/src/ndi/gui/component/ndi_matlab_python_bridge.yaml +++ b/src/ndi/gui/component/ndi_matlab_python_bridge.yaml @@ -523,6 +523,23 @@ classes: 2026-09-25: NDI-matlab b367283b added a per-bar Timeout name-value on addBar so a long-running bar can opt out of the global auto-timeout sweep. No Python UI counterpart; no behavioural change here. 2026-09-28: NDI-matlab 2b23dd30 relaxed the addBar Timeout argument-block size constraint (dropped `(1,:)`, which was rejecting `duration.empty` under some MATLAB releases). Pure MATLAB argument-validation adjustment; no behavioural change here. 2026-10-14: NDI-matlab 1e8fc581, d367f621 and 175b08af (on main via merge aa96851e) made the bar fill its row on R2026a: addBar now parents the bar's uiaxes to a borderless uipanel (AutoResizeChildren off) and pins the axes' InnerPosition BarMargin (5) px inside it through the new static fitBarAxes, run once and as the panel's SizeChangedFcn; removeBar deletes and re-rows that panel via Axes.Parent. This works around uiaxes reserving tick-label/title room in a grid cell, which left R2026a a 2.7 px bar. The Qt port draws a QProgressBar that its layout already sizes to the row, so there is nothing to follow; BarMargin and fitBarAxes join the uifigure plumbing above that has no Qt counterpart. No behavioural change here. + 2026-10-27: the behavioural gap above is closed, at the maintainer's + direction, so that ndi_session.ingest can show progress by default + without risking a headless run (Qt aborts the whole process when it + cannot reach a display, which no try/except catches). Silent mode is + ported: the Silent constructor argument and property, and the static + isHeadless, silentModeDefault and resolveSilentMode, with MATLAB's + decision order (explicit argument, process-wide override, detection). + A silent window builds no widgets -- PySide6 is not even imported -- and + keeps all bar bookkeeping. isHeadless has Python-specific signals: + MATLAB's batchStartupOptionUsed has no Python counterpart, so it is + "PySide6 is missing, or on Linux neither DISPLAY nor WAYLAND_DISPLAY is + set" (QT_QPA_PLATFORM=offscreen/minimal counts as a usable display, + since Qt then needs none). The per-bar Timeout of NDI-matlab b367283b, + recorded above as having no counterpart, is ported too: ingestion bars + need to opt out of the 60 s window timeout, because one epoch can take + longer than that. Examined against NDI-matlab aa96851e; the MATLAB file + has not changed. NDI-python#295: the MATLAB properties this port does not expose under MATLAB's name are enumerated in tests/bridge_property_allowlist.yaml, @@ -549,6 +566,13 @@ classes: type_python: "bool" default: "True" decision_log: "Exact match." + - name: Silent + type_matlab: "logical" + type_python: "bool" + decision_log: > + Exact match. Decided once at construction by resolveSilentMode; + True means no window or widgets exist and the bars are bookkeeping + only. - name: ProgressBars type_matlab: "struct array" type_python: "list[dict[str, Any]]" @@ -573,6 +597,10 @@ classes: type_matlab: "logical" type_python: "bool" default: "True" + - name: Silent + type_matlab: "logical or [] (empty: decide automatically)" + type_python: "bool | None" + default: "None" output_arguments: - name: app type_python: "ndi_gui_component_ProgressBarWindow" @@ -599,8 +627,15 @@ classes: type_matlab: "logical" type_python: "bool" default: "False" + - name: Timeout + type_matlab: "duration (duration.empty: inherit the window's)" + type_python: "float | None" + default: "None" output_arguments: [] - decision_log: "Exact match." + decision_log: > + Exact match. Timeout is in seconds rather than a duration, like the + window's Timeout; None inherits the window's value and math.inf + (MATLAB minutes(Inf)) never times out. - name: updateBar input_arguments: @@ -665,3 +700,42 @@ classes: - name: barNum type_python: "list[int]" decision_log: "Exact match." + + - name: isHeadless + kind: static + input_arguments: [] + output_arguments: + - name: tf + type_python: "bool" + decision_log: > + Ported with Python-specific signals: True when PySide6 is missing, or + on Linux when neither DISPLAY nor WAYLAND_DISPLAY is set (unless + QT_QPA_PLATFORM is offscreen/minimal). MATLAB's -batch check has no + Python counterpart. + + - name: silentModeDefault + kind: static + input_arguments: + - name: newValue + type_matlab: "logical or [] (clear)" + type_python: "bool | None" + default: "(omitted: read only)" + output_arguments: + - name: value + type_python: "bool | None" + decision_log: > + Exact match. A module-level variable stands in for MATLAB's + persistent; None plays the part of MATLAB's [] both as "no override" + and, when passed, as "clear it". + + - name: resolveSilentMode + kind: static + input_arguments: + - name: explicitValue + type_matlab: "logical or []" + type_python: "bool | None" + default: "None" + output_arguments: + - name: tf + type_python: "bool" + decision_log: "Exact match. Explicit value, then silentModeDefault, then isHeadless." diff --git a/src/ndi/ndi_matlab_python_bridge.yaml b/src/ndi/ndi_matlab_python_bridge.yaml index f6f51d38..20001b3d 100644 --- a/src/ndi/ndi_matlab_python_bridge.yaml +++ b/src/ndi/ndi_matlab_python_bridge.yaml @@ -1207,6 +1207,16 @@ classes: under the same keys, so one setting governs both languages. The third is MATLAB's own: Python registers Update.Python.LastCheckTime instead, since the two check different repositories and one must not hold the other off. + 2026-10-27: the three Parallel items (OpenPoolIfNone, + UseDefaultNumberOfWorkers, NumberOfWorkers) are now ported, reversing + the earlier decision above, because there is now Python code that reads + them: ndi_session.ingest computes its epochs on worker processes when + they allow (through ndi.fun.utils.parallelWorkers), mirroring + NDI-matlab's ingest (VH-Lab/NDI-matlab#1079). Same keys and defaults as + MATLAB, so one setting in the shared file governs both languages. + NumberOfWorkers is registered as "int" (a worker count). The + descriptions say how each language reads them: Python has no pool that + outlives a call, so with OpenPoolIfNone false NDI-python is serial. properties: - name: Items diff --git a/src/ndi/preferences.py b/src/ndi/preferences.py index 9dd855a0..9255ad61 100644 --- a/src/ndi/preferences.py +++ b/src/ndi/preferences.py @@ -240,6 +240,48 @@ def _register_defaults(self) -> None: "NDI_FROM_MATLAB=1, the console script itself drops MATLAB's " "library paths before loading anything.", ) + # The Parallel items are the same keys NDI-matlab registers, so one + # setting in the shared file governs both languages. Python has no + # parpool that stays open between calls, so "a pool you have already + # opened" never exists here: NDI-python starts worker processes only + # when OpenPoolIfNone is true (see ndi.fun.utils.parallelWorkers). + self._add_item( + "Parallel", + "", + "OpenPoolIfNone", + False, + "bool", + "Start parallel workers when a calculation reaches a parallel " + "loop and no pool is running. When false, NDI never starts a pool " + "of its own: NDI-matlab uses a pool you have already opened, and " + "otherwise runs serially; NDI-python, which has no pool that stays " + "open, runs serially. Each worker is a separate process, so a pool " + "nobody asked for costs both start-up time and a multiple of the " + "memory.", + ) + self._add_item( + "Parallel", + "", + "UseDefaultNumberOfWorkers", + True, + "bool", + "Let NDI decide how many workers a parallel loop may use: an open " + "MATLAB pool is used at its own size, and a pool opened by NDI " + "takes the default, normally one worker per core. Turn this off to " + "set the number yourself in NumberOfWorkers, which is worth doing " + "on a machine with limited memory.", + ) + self._add_item( + "Parallel", + "", + "NumberOfWorkers", + 2, + "int", + "How many workers a parallel loop may use, when " + "UseDefaultNumberOfWorkers is off. Ignored while that is on. This " + "is an upper limit: an open MATLAB pool smaller than this is used " + "at its own size, not enlarged.", + ) # Update.CheckOnStartup and Update.CheckIntervalDays are the same keys # NDI-matlab registers, so one setting in the shared file governs # both languages. The last-check time is per language (MATLAB's is diff --git a/src/ndi/session/ndi_matlab_python_bridge.yaml b/src/ndi/session/ndi_matlab_python_bridge.yaml index 25267ea4..f987f936 100644 --- a/src/ndi/session/ndi_matlab_python_bridge.yaml +++ b/src/ndi/session/ndi_matlab_python_bridge.yaml @@ -33,7 +33,7 @@ classes: - name: session type: class matlab_path: "+ndi/session.m" - matlab_last_sync_hash: "10bad94" + matlab_last_sync_hash: "375a0ff7e" python_path: "ndi/session/session_base.py" python_class: "ndi_session" @@ -289,13 +289,32 @@ classes: decision_log: "Exact match." - name: ingest - input_arguments: [] + input_arguments: + - name: ShowProgress + type_matlab: "logical" + type_python: "bool" + default: "True" + - name: ProgressBar + type_matlab: "ndi.gui.component.ProgressBarWindow or []" + type_python: "ndi_gui_component_ProgressBarWindow | None" + default: "None" + - name: ProgressBarTag + type_matlab: "char" + type_python: "str" + default: "''" output_arguments: - name: b type_python: "bool" - name: errmsg type_python: "str" - decision_log: "Exact match. Returns (bool, str) tuple." + decision_log: > + Exact match. Returns (bool, str) tuple. NDI-matlab 96430ab81 and + 8feb9166e (VH-Lab/NDI-matlab#1079) add a session-level progress bar + (not auto-removed; it stays at 100% like MATLAB's), one bar per DAQ + system, and parallel per-epoch work governed by the Parallel.* + preferences; all ported, with the options as keyword-only + arguments. Python ingest keeps its existing behaviour of catching a + DAQ system's exception into errmsg rather than raising. - name: get_ingested_docs input_arguments: [] diff --git a/src/ndi/session/session_base.py b/src/ndi/session/session_base.py index 5ba4ca76..d4ba5f9c 100644 --- a/src/ndi/session/session_base.py +++ b/src/ndi/session/session_base.py @@ -729,18 +729,42 @@ def _update_syncgraph_in_db(self) -> None: # Ingest Methods # ========================================================================= - def ingest(self) -> tuple[bool, str]: + def ingest( + self, + *, + ShowProgress: bool = True, + ProgressBar: Any = None, + ProgressBarTag: str = "", + ) -> tuple[bool, str]: """ Ingest all raw data and sync info into the database. + MATLAB equivalent: ndi.session/ingest + + [B,ERRMSG] = INGEST(NDI_SESSION_OBJ, ...) + + The reading of each epoch runs on worker processes when NDI's parallel + preferences allow it (Parallel.OpenPoolIfNone, Parallel.UseDefault- + NumberOfWorkers, Parallel.NumberOfWorkers; see ndi.fun.parallelWorkers). + With the default preferences ingestion is serial. All database writes + stay serial. Progress is shown with + ndi.gui.component.ProgressBarWindow: one bar for the session and one for + each DAQ system as it is ingested; with no display the window is + silent automatically. + + Args: + ShowProgress: Show ingestion progress bars. + ProgressBar: An existing ProgressBarWindow to use. None opens/reuses + one titled 'NDI Ingest'. + ProgressBarTag: Tag for the session-level bar. Empty makes a new + unique id. + Returns: Tuple of (success, error_message) """ - errmsg = "" + from ..daq.system import ndi_daq_system - # Ingest syncgraph - if self._syncgraph is not None: - d_syncgraph = self._syncgraph.new_document() + errmsg = "" # Get all DAQ systems daqs = self.daqsystem_load(name="(.*)") @@ -749,12 +773,25 @@ def ingest(self) -> tuple[bool, str]: elif not isinstance(daqs, list): daqs = [daqs] + progress_bar, bar_tag = None, "" + if ShowProgress: + progress_bar, bar_tag = ndi_daq_system.addIngestProgressBar( + ProgressBar, f"Ingesting session {self.reference}", Tag=ProgressBarTag, Auto=False + ) + # steps: the syncgraph, each daq system, and the final syncgraph add + num_steps = len(daqs) + 2 + + # Ingest syncgraph + if self._syncgraph is not None: + d_syncgraph = self._syncgraph.new_document() + ndi_daq_system.updateIngestProgressBar(progress_bar, bar_tag, 1 / num_steps) + # Ingest each DAQ system daq_docs = [] success = True - for daq in daqs: + for i, daq in enumerate(daqs, start=1): try: - b, docs = daq.ingest() + b, docs = daq.ingest(ShowProgress=ShowProgress, ProgressBar=progress_bar) daq_docs.append(docs) if not b: success = False @@ -762,6 +799,7 @@ def ingest(self) -> tuple[bool, str]: except Exception as e: success = False errmsg = str(e) + ndi_daq_system.updateIngestProgressBar(progress_bar, bar_tag, (1 + i) / num_steps) if not success: # Clean up on failure @@ -769,12 +807,14 @@ def ingest(self) -> tuple[bool, str]: if docs: for doc in docs if isinstance(docs, list) else [docs]: self.database_rm(doc) + ndi_daq_system.removeIngestProgressBar(progress_bar, bar_tag) else: # Add syncgraph documents if self._syncgraph is not None: for doc in d_syncgraph: doc = doc.set_session_id(self.id()) self.database_add(doc) + ndi_daq_system.updateIngestProgressBar(progress_bar, bar_tag, 1.0) return success, errmsg diff --git a/tests/bridge_property_allowlist.yaml b/tests/bridge_property_allowlist.yaml index af9730d3..ef88624f 100644 --- a/tests/bridge_property_allowlist.yaml +++ b/tests/bridge_property_allowlist.yaml @@ -260,12 +260,6 @@ not_exposed: reason: > Not ported. The Qt port sets its border in a stylesheet rather than exposing a width for MATLAB's hand-drawn axes. - - python_class: ndi_gui_component_ProgressBarWindow - property: Silent - reason: > - Not ported. MATLAB suppresses the window when it cannot open a figure; - the Qt port raises at construction if Qt is unavailable, so there is no - silent mode to configure. - python_class: ndi_gui_component_ProgressBarWindow property: ProgressFigure reason: "MATLAB uifigure handle; the Qt port owns its own QWidget." diff --git a/tests/test_ingest_progress_parallel.py b/tests/test_ingest_progress_parallel.py new file mode 100644 index 00000000..fc22d196 --- /dev/null +++ b/tests/test_ingest_progress_parallel.py @@ -0,0 +1,412 @@ +"""Ingestion progress bars and parallel per-epoch work. + +MATLAB counterpart: tests/+ndi/+unittest/+ingestion/ingestionProgressParallel.m +(VH-Lab/NDI-matlab#1079) + +Covers the pieces ported together for that change: + +* ndi_gui_component_ProgressBarWindow's silent (headless) mode and per-bar + Timeout -- needed so ingest can show progress by default without a display; +* the Parallel.* preferences and ndi.fun.parallelWorkers; +* ndi.daq.fun.ingestEpochDocuments, serially and on worker processes; +* ndi_session.ingest / ndi_daq_system.ingest driving the bars, end to end on + the example Intan session given a second epoch. + +Every window here is forced silent, so no display or PySide6 is needed. +""" + +from __future__ import annotations + +import math +import shutil +import threading +import warnings + +import pytest + +import ndi.preferences +from ndi.daq.fun import ingestEpochDocuments +from ndi.fun.utils import parallelWorkers +from ndi.gui.component.ProgressBarWindow import ndi_gui_component_ProgressBarWindow as PBW + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _no_silent_override(): + """Leave the process-wide silent override as it was found.""" + previous = PBW.silentModeDefault() + yield + PBW.silentModeDefault(previous) + + +def _prefs(monkeypatch, **values): + """Pin Parallel.* preference values for one test (nothing is written).""" + real_get = ndi.preferences.get + + def fake_get(path): + key = path.split(".")[-1] + if path.startswith("Parallel.") and key in values: + return values[key] + return real_get(path) + + monkeypatch.setattr(ndi.preferences, "get", fake_get) + + +class _FakeDoc: + def __init__(self, kind, epoch_id): + self.kind = kind + self.epoch_id = epoch_id + + def __eq__(self, other): + return (self.kind, self.epoch_id) == (other.kind, other.epoch_id) + + +class _FakeReader: + """A picklable reader; its _session is deliberately unpicklable.""" + + def __init__(self, kind="reader"): + self.kind = kind + self._session = threading.Lock() # a live session cannot be pickled either + + def ingest_epochfiles(self, epochfiles, epoch_id): + return _FakeDoc(self.kind, epoch_id) + + +class _UnpicklableReader(_FakeReader): + def __init__(self): + super().__init__() + self.handle = threading.Lock() # not the session: cannot be detached + + +# --------------------------------------------------------------------------- +# ProgressBarWindow silent mode +# --------------------------------------------------------------------------- + + +class TestSilentProgressBarWindow: + def test_silent_window_keeps_bookkeeping(self): + w = PBW("silent test", Silent=True, AutoDelete=False) + assert w.Silent + w.addBar(Label="one", Tag="a", Auto=False) + assert w.getState("a") == "Open" + w.updateBar("a", 0.5) + assert w._bars[0].Progress == 0.5 + w.updateBar("a", 1.0) + assert w.getState("a") == "Complete" + w.removeBar("a") + assert w.getBarNum("a")[0] is None + + def test_auto_bar_is_removed_on_completion(self): + w = PBW("silent auto", Silent=True, AutoDelete=False) + w.addBar(Label="auto", Tag="a", Auto=True) + w.updateBar("a", 1.0) + assert w.getBarNum("a")[0] is None + + def test_per_bar_timeout(self): + w = PBW("silent timeout", Silent=True, AutoDelete=False) + w.setTimeout(0.0) + w.addBar(Label="inherits", Tag="inherit", Auto=False) + w.addBar(Label="never", Tag="never", Auto=False, Timeout=math.inf) + w.updateBar("never", 0.5) + assert w.getState("inherit") == "Timeout" + assert w.getState("never") == "Open" + + def test_silent_mode_default_overrides_detection(self): + PBW.silentModeDefault(True) + assert PBW.resolveSilentMode() is True + assert PBW.resolveSilentMode(False) is False # explicit wins + PBW.silentModeDefault(False) + assert PBW.resolveSilentMode() is False + PBW.silentModeDefault(None) + assert PBW.silentModeDefault() is None + + def test_is_headless_rules(self, monkeypatch): + import ndi.gui.component.ProgressBarWindow as mod + + monkeypatch.setattr(mod, "_QT_AVAILABLE", False) + assert PBW.isHeadless() is True + + monkeypatch.setattr(mod, "_QT_AVAILABLE", True) + monkeypatch.setattr(mod.sys, "platform", "linux") + monkeypatch.delenv("DISPLAY", raising=False) + monkeypatch.delenv("WAYLAND_DISPLAY", raising=False) + monkeypatch.delenv("QT_QPA_PLATFORM", raising=False) + assert PBW.isHeadless() is True + monkeypatch.setenv("QT_QPA_PLATFORM", "offscreen") + assert PBW.isHeadless() is False + monkeypatch.delenv("QT_QPA_PLATFORM") + monkeypatch.setenv("DISPLAY", ":0") + assert PBW.isHeadless() is False + + +# --------------------------------------------------------------------------- +# Parallel preferences and parallelWorkers +# --------------------------------------------------------------------------- + + +class TestParallelWorkers: + def test_preferences_are_registered_with_matlab_defaults(self): + prefs = ndi.preferences.ndi_preferences(filename=None) + items = {(i.category, i.name): i for i in prefs._items} + assert items[("Parallel", "OpenPoolIfNone")].default_value is False + assert items[("Parallel", "UseDefaultNumberOfWorkers")].default_value is True + assert items[("Parallel", "NumberOfWorkers")].default_value == 2 + + def test_default_is_serial(self, monkeypatch): + _prefs(monkeypatch, OpenPoolIfNone=False, UseDefaultNumberOfWorkers=True) + assert parallelWorkers() == 0 + # naming a number does not start workers either, as in MATLAB + assert parallelWorkers(numberOfWorkers=4) == 0 + + def test_open_pool_uses_number_of_workers(self, monkeypatch): + _prefs( + monkeypatch, + OpenPoolIfNone=True, + UseDefaultNumberOfWorkers=False, + NumberOfWorkers=3, + ) + assert parallelWorkers() == 3 + assert parallelWorkers(numberOfWorkers=5) == 5 + + def test_open_pool_default_size_is_cpu_count(self, monkeypatch): + _prefs(monkeypatch, OpenPoolIfNone=True, UseDefaultNumberOfWorkers=True) + assert parallelWorkers() >= 1 + + def test_bad_worker_count_warns_and_uses_default(self, monkeypatch): + _prefs( + monkeypatch, + OpenPoolIfNone=True, + UseDefaultNumberOfWorkers=False, + NumberOfWorkers=0, + ) + with pytest.warns(UserWarning, match="not a number of workers"): + n = parallelWorkers() + assert n >= 1 + + +# --------------------------------------------------------------------------- +# ingestEpochDocuments +# --------------------------------------------------------------------------- + +_EPOCH_FILES = [[f"/data/e{k}.rhd"] for k in range(1, 5)] +_EPOCH_IDS = [f"epoch_{k}" for k in range(1, 5)] + + +class TestIngestEpochDocuments: + def test_serial_order_and_progress(self, monkeypatch): + _prefs(monkeypatch, OpenPoolIfNone=False) + calls = [] + docs, used_parallel = ingestEpochDocuments( + _FakeReader(), + [_FakeReader("meta")], + _EPOCH_FILES, + _EPOCH_IDS, + ProgressFcn=lambda done, total: calls.append((done, total)), + ) + assert not used_parallel + assert calls == [(k, 4) for k in range(1, 5)] + assert [d[0].epoch_id for d in docs] == _EPOCH_IDS + assert [d[1].kind for d in docs] == ["meta"] * 4 + + def test_forced_serial_override(self, monkeypatch): + _prefs(monkeypatch, OpenPoolIfNone=True, UseDefaultNumberOfWorkers=True) + _, used_parallel = ingestEpochDocuments( + _FakeReader(), [], _EPOCH_FILES, _EPOCH_IDS, NumberOfWorkers=0 + ) + assert not used_parallel + + def test_parallel_matches_serial(self, monkeypatch): + _prefs(monkeypatch, OpenPoolIfNone=False) + serial, _ = ingestEpochDocuments( + _FakeReader(), [_FakeReader("meta")], _EPOCH_FILES, _EPOCH_IDS + ) + + _prefs( + monkeypatch, + OpenPoolIfNone=True, + UseDefaultNumberOfWorkers=False, + NumberOfWorkers=2, + ) + calls = [] + reader = _FakeReader() + parallel, used_parallel = ingestEpochDocuments( + reader, + [_FakeReader("meta")], + _EPOCH_FILES, + _EPOCH_IDS, + ProgressFcn=lambda done, total: calls.append((done, total)), + ) + assert used_parallel + assert parallel == serial + assert sorted(calls) == [(k, 4) for k in range(1, 5)] + # the caller's reader keeps its session; only the workers' copies lose it + assert reader._session is not None + + def test_unpicklable_reader_falls_back_to_serial(self, monkeypatch): + _prefs( + monkeypatch, + OpenPoolIfNone=True, + UseDefaultNumberOfWorkers=False, + NumberOfWorkers=2, + ) + with pytest.warns(UserWarning, match="Ingesting serially"): + docs, used_parallel = ingestEpochDocuments( + _UnpicklableReader(), [], _EPOCH_FILES, _EPOCH_IDS + ) + assert not used_parallel + assert [d[0].epoch_id for d in docs] == _EPOCH_IDS + + def test_worker_errors_propagate(self, monkeypatch): + _prefs(monkeypatch, OpenPoolIfNone=False) + + class Boom(_FakeReader): + def ingest_epochfiles(self, epochfiles, epoch_id): + raise RuntimeError("bad epoch") + + with pytest.raises(RuntimeError, match="bad epoch"): + ingestEpochDocuments(Boom(), [], _EPOCH_FILES, _EPOCH_IDS) + + def test_size_mismatch_raises(self): + with pytest.raises(ValueError, match="same number"): + ingestEpochDocuments(_FakeReader(), [], _EPOCH_FILES, _EPOCH_IDS[:2]) + + +# --------------------------------------------------------------------------- +# ndi_session.ingest end to end +# --------------------------------------------------------------------------- + + +def _intan_session(session_dir): + """The example Intan session (as in the ingestion symmetry test) with a + second epoch, so there is more than one epoch to distribute. + + The second epoch is a copy in a subdirectory: this port's + find_file_groups still makes one epoch per DIRECTORY (see the comment + there), where MATLAB's findfilegroups makes one per '#' match, which is + why the MATLAB test can keep both copies side by side.""" + from ndi.session.dir import ndi_session_dir + from tests.symmetry.make_artifacts.session._ingestion_helpers import ( + setup_intan_session, + ) + + session_dir.mkdir() + info = setup_intan_session(session_dir, reader_class="intan") + stem = info["rhd_file"][: -len(".rhd")] + second = session_dir / "t00002" + second.mkdir() + shutil.copy2(session_dir / info["rhd_file"], second / info["rhd_file"]) + shutil.copy2( + session_dir / f"{stem}.epochprobemap.ndi", + second / f"{stem}.epochprobemap.ndi", + ) + + session = ndi_session_dir("exp1", session_dir) + session.cache.clear() + fn_doc = session.newdocument( + "daq/filenavigator", + **{ + "base.name": "unknown", + "filenavigator.ndi_filenavigator_class": "ndi.file.navigator", + "filenavigator.fileparameters": "{ '#.rhd', '#.epochprobemap.ndi' }", + "filenavigator.epochprobemap_class": "ndi.epoch.epochprobemap_daqsystem", + "filenavigator.epochprobemap_fileparameters": "{ '(.*?)epochprobemap.ndi' }", + }, + ) + session.database_add(fn_doc) + dr_doc = session.newdocument( + "daq/daqreader", + **{ + "base.name": "intan_reader", + "daqreader.ndi_daqreader_class": "ndi.daq.reader.mfdaq.intan", + }, + ) + session.database_add(dr_doc) + daq_doc = session.newdocument( + "daq/daqsystem", + **{"base.name": "intan1", "daqsystem.ndi_daqsystem_class": "ndi.daq.system.mfdaq"}, + ) + daq_doc = daq_doc.set_dependency_value("filenavigator_id", fn_doc.id, error_if_not_found=False) + daq_doc = daq_doc.set_dependency_value("daqreader_id", dr_doc.id, error_if_not_found=False) + session.database_add(daq_doc) + return session + + +def _ingested_epoch_docs(session): + from ndi.query import ndi_query + + return session.database_search(ndi_query("").isa("daqreader_epochdata_ingested")) + + +def _ingested_summary(session): + """What each ingested epoch says, independent of the (random) epoch ids.""" + import json + + return sorted( + json.dumps(d.document_properties["daqreader_epochdata_ingested"], sort_keys=True) + for d in _ingested_epoch_docs(session) + ) + + +def _have_intan(): + try: + from tests.symmetry.make_artifacts.session._ingestion_helpers import _find_intan_rhd + + return _find_intan_rhd() is not None + except Exception: # noqa: BLE001 + return False + + +@pytest.mark.skipif(not _have_intan(), reason="Intan example data (.rhd) not found") +class TestSessionIngest: + def test_default_ingest_still_works(self, tmp_path, monkeypatch): + _prefs(monkeypatch, OpenPoolIfNone=False) + PBW.silentModeDefault(True) + session = _intan_session(tmp_path / "exp1") + ok, msg = session.ingest() + assert ok, msg + assert len(_ingested_epoch_docs(session)) == 2 + + def test_progress_bars_reach_completion(self, tmp_path, monkeypatch): + _prefs(monkeypatch, OpenPoolIfNone=False) + session = _intan_session(tmp_path / "exp1") + w = PBW("ingest test", Silent=True, AutoDelete=False) + ok, msg = session.ingest(ProgressBar=w, ProgressBarTag="sessionBar") + assert ok, msg + idx, _ = w.getBarNum("sessionBar") + assert idx is not None + assert w._bars[idx].Progress == 1.0 + assert w.getState("sessionBar") == "Complete" + # the DAQ system's auto bar completed and removed itself + assert len(w._bars) == 1 + + def test_show_progress_false_adds_no_bars(self, tmp_path, monkeypatch): + _prefs(monkeypatch, OpenPoolIfNone=False) + session = _intan_session(tmp_path / "exp1") + w = PBW("ingest test", Silent=True, AutoDelete=False) + ok, msg = session.ingest(ShowProgress=False, ProgressBar=w) + assert ok, msg + assert w._bars == [] + + def test_parallel_matches_serial(self, tmp_path, monkeypatch): + PBW.silentModeDefault(True) + _prefs(monkeypatch, OpenPoolIfNone=False) + s1 = _intan_session(tmp_path / "serial") + ok, msg = s1.ingest() + assert ok, msg + + _prefs( + monkeypatch, + OpenPoolIfNone=True, + UseDefaultNumberOfWorkers=False, + NumberOfWorkers=2, + ) + s2 = _intan_session(tmp_path / "parallel") + with warnings.catch_warnings(): + warnings.simplefilter("error", UserWarning) # no fallback to serial + ok, msg = s2.ingest() + assert ok, msg + assert len(_ingested_epoch_docs(s2)) == 2 + assert _ingested_summary(s2) == _ingested_summary(s1)