Stream validation in parallel

dandi/cli/cmd_validate.py

@@ -61,6 +61,7 @@ def _collect_results(
     devel_debug: bool,
     allow_any_path: bool,
     missing_file_content: MissingFileContent = MissingFileContent.error,
+    jobs: int = 1,
 ) -> list[ValidationResult]:
     """Run validation and collect all results into a list."""
     # Avoid heavy import by importing within function:
@@ -87,6 +88,7 @@ def _collect_results(
             devel_debug=devel_debug,
             allow_any_path=allow_any_path,
             missing_file_content=missing_file_content,
+            jobs=jobs,
         )
     )
 
@@ -170,10 +172,25 @@ def _filter_results(
     "in a datalad dataset without fetched data). 'error' (default) emits a "
     "concise error per file, 'skip' skips each such file with a warning, "
     "'only-non-data' skips content-dependent validators but still validates "
-    "path layout.",
-    type=click.Choice(["error", "only-non-data", "skip"], case_sensitive=True),
+    "path layout, 'stream' streams the content of annexed files from the URLs "
+    "git-annex knows for it, so that content-dependent validators run without "
+    "the files having to be downloaded (requires git-annex and `pip install "
+    "'dandi[datalad]'`).",
+    type=click.Choice(
+        ["error", "only-non-data", "skip", "stream"], case_sensitive=True
+    ),
     default="error",
 )
+@click.option(
+    "-J",
+    "--jobs",
+    type=click.IntRange(min=1),
+    default=1,
+    show_default=True,
+    help="Number of NWB files to validate in parallel, in separate processes.  "
+    "Useful especially with '--missing-file-content stream', for which "
+    "validating a file mostly waits for its content to be downloaded.",
+)
 @click.option(
     "--load",
     help="Load validation results from JSONL file(s) instead of running validation.",
@@ -196,6 +213,7 @@ def validate(
     summary: bool = False,
     max_per_group: int | None = None,
     missing_file_content: str = "error",
+    jobs: int = 1,
     load: tuple[str, ...] = (),
     schema: str | None = None,
     devel_debug: bool = False,
@@ -239,7 +257,12 @@ def validate(
     else:
         mfc = MissingFileContent(missing_file_content)
         results = _collect_results(
-            paths, schema, devel_debug, allow_any_path, missing_file_content=mfc
+            paths,
+            schema,
+            devel_debug,
+            allow_any_path,
+            missing_file_content=mfc,
+            jobs=jobs,
         )
         # Auto-save companion right after collection, before filtering — so
         # all results are preserved regardless of display filters.

dandi/cli/tests/test_cmd_validate.py

@@ -1,6 +1,7 @@
 import json
 from pathlib import Path
-from typing import cast
+import shutil
+from typing import Any, cast
 
 from click.testing import CliRunner
 import pytest
@@ -998,6 +999,47 @@ def test_validate_missing_file_content_no_broken_symlinks(tmp_path: Path) -> Non
     assert "FILE_CONTENT_MISSING" not in r_skip.output
 
 
+@pytest.mark.ai_generated
+def test_validate_missing_file_content_stream(
+    tmp_path: Path, simple3_nwb: Path, range_http_server: tuple[Any, str]
+) -> None:
+    """--missing-file-content=stream validates annexed content by streaming it."""
+    from ...tests.fixtures import make_git_annex_dandiset
+    from ...tests.skip import skipif
+
+    skipif.no_git_annex()
+    pytest.importorskip("datalad_fuse.adapter")
+    pytest.importorskip("remfile")
+    _, base_url = range_http_server
+    shutil.copy(simple3_nwb, tmp_path / "served" / "content.nwb")
+    ds = tmp_path / "ds"
+    make_git_annex_dandiset(
+        ds, {"sub-001/sub-001.nwb": (simple3_nwb, f"{base_url}/content.nwb")}
+    )
+    out = tmp_path / "out.jsonl"
+    r = CliRunner().invoke(
+        validate,
+        [
+            "--missing-file-content",
+            "stream",
+            "--min-severity",
+            "INFO",
+            "-f",
+            "json_lines",
+            "-o",
+            str(out),
+            str(ds),
+        ],
+    )
+    assert "Traceback" not in r.output
+    ids = [rec.id for rec in load_validation_jsonl([str(out)])]
+    assert "DANDI.FILE_CONTENT_STREAMED" in ids
+    assert "DANDI.FILE_CONTENT_MISSING" not in ids
+    # simple3_nwb lacks a subject_id, which only content-based checks notice
+    assert "NWBI.check_subject_id_exists" in ids
+    assert r.exit_code == 1
+
+
 @pytest.mark.ai_generated
 def test_validate_broken_symlink_path_argument(tmp_path: Path) -> None:
     """A broken symlink can be given directly as a path to validate.

dandi/files/bases.py

@@ -2,15 +2,17 @@
 
 from abc import ABC, abstractmethod
 from collections import deque
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass
 from datetime import datetime
+from functools import cache
 import os
 from pathlib import Path
 import re
 from threading import Lock
-from typing import IO, Any, Generic
+import traceback
+from typing import IO, Any, Generic, cast
 from xml.etree.ElementTree import fromstring
 
 import dandischema
@@ -32,7 +34,8 @@
     set_asset_schema_key,
 )
 from dandi.metadata.core import get_default_metadata
-from dandi.misctypes import DUMMY_DANDI_ETAG, Digest, LocalReadableFile, P
+from dandi.misctypes import DUMMY_DANDI_ETAG, Digest, LocalReadableFile, P, Readable
+from dandi.support.annex_stream import annex_fingerprint
 from dandi.utils import post_upload_size_check, pre_upload_size_check, yaml_load
 from dandi.validate._types import (
     ORIGIN_INTERNAL_DANDI,
@@ -320,12 +323,22 @@ class LocalFileAsset(LocalAsset):
     an asset of a Dandiset
     """
 
+    #: An alternative source for the asset's content, used instead of the file
+    #: at `filepath` when set.  This is used to stream the content of an
+    #: annexed file (e.g., in a DataLad dataset) that is not present locally;
+    #: see `dandi.support.annex_stream`.
+    content_source: Readable | None = None
+
+    def _content(self) -> Path | Readable:
+        """The source to read the asset's content from"""
+        return self.content_source if self.content_source is not None else self.filepath
+
     def get_metadata(
         self,
         digest: Digest | None = None,
         ignore_errors: bool = True,
     ) -> BareAsset:
-        metadata = get_default_metadata(self.filepath, digest=digest)
+        metadata = get_default_metadata(self._content(), digest=digest)
         metadata.path = self.path
         return metadata
 
@@ -515,7 +528,7 @@ def get_metadata(
         from dandi.metadata.nwb import nwb2asset
 
         try:
-            metadata = nwb2asset(self.filepath, digest=digest)
+            metadata = nwb2asset(self._content(), digest=digest)
         except Exception as e:
             lgr.warning(
                 "Failed to extract NWB metadata from %s: %s: %s",
@@ -524,7 +537,7 @@ def get_metadata(
                 str(e),
             )
             if ignore_errors:
-                metadata = get_default_metadata(self.filepath, digest=digest)
+                metadata = get_default_metadata(self._content(), digest=digest)
             else:
                 raise
         metadata.path = self.path
@@ -555,12 +568,16 @@ def get_validation_errors(
             pass
         else:
             # Avoid heavy import by importing within function:
-            from nwbinspector import Importance, inspect_nwbfile, load_config
+            from nwbinspector import Importance, load_config
 
             # Avoid heavy import by importing within function:
             from dandi.pynwb_utils import validate as pynwb_validate
 
-            errors.extend(pynwb_validate(self.filepath, devel_debug=devel_debug))
+            errors.extend(
+                pynwb_validate(
+                    self.filepath, devel_debug=devel_debug, readable=self.content_source
+                )
+            )
             if schema_version is not None:
                 errors.extend(
                     super().get_validation_errors(
@@ -576,9 +593,7 @@ def get_validation_errors(
                         validator_version=str(_get_nwb_inspector_version()),
                     )
 
-                    for error in inspect_nwbfile(
-                        nwbfile_path=self.filepath,
-                        skip_validate=True,
+                    for error in self._inspect_nwbfile(
                         config=load_config(filepath_or_keyword="dandi"),
                         importance_threshold=Importance.BEST_PRACTICE_VIOLATION,
                     ):
@@ -623,6 +638,117 @@ def get_validation_errors(
             )
         return errors
 
+    def _inspect_nwbfile(
+        self, config: dict | None = None, importance_threshold: Any = None
+    ) -> list[Any]:
+        """
+        Return the messages from inspecting the file with nwbinspector, reading
+        the content from `content_source` if it is set (in which case what
+        `nwbinspector.inspect_nwbfile()` does for a local file is replicated
+        here for the streamed content)
+        """
+        # Avoid heavy import by importing within function:
+        from importlib.metadata import version
+
+        from nwbinspector import Importance, InspectorMessage
+
+        if importance_threshold is None:
+            importance_threshold = Importance.BEST_PRACTICE_SUGGESTION
+        source = self.content_source
+        try:
+            return _cached_inspect_nwbfile()(
+                source if source is not None else str(self.filepath),
+                str(self.filepath),
+                version("nwbinspector"),
+                config,
+                importance_threshold,
+            )
+        except _StreamedReadError as e:
+            # Report the failure the same way inspect_nwbfile() does for a
+            # local file that PyNWB cannot read
+            exc = e.__cause__
+            assert exc is not None
+            return [
+                InspectorMessage(
+                    message="".join(traceback.format_exception(exc)),
+                    importance=Importance.ERROR,
+                    check_function_name=(
+                        f"During io.read(), an error occurred: "
+                        f"{type(exc).__module__}.{type(exc).__name__}. "
+                        "This indicates that PyNWB was unable to read the file. "
+                        "See the traceback message for more details."
+                    ),
+                    file_path=str(self.filepath),
+                )
+            ]
+
+
+class _StreamedReadError(Exception):
+    """
+    Failure to read or inspect streamed content, which is not cached as a
+    result of `_inspect_nwbfile()`, as it might be transient (e.g., a network
+    error)
+    """
+
+
+def _inspect_nwbfile(
+    source: str | Readable,
+    filepath: str,
+    nwbinspector_version: str,
+    config: dict | None,
+    importance_threshold: Any,
+) -> list[Any]:
+    """
+    Inspect the NWB file with nwbinspector, reading its content from ``source``
+    (its path or a `Readable`), and return the messages, reporting
+    ``filepath`` as their file path
+
+    The results are cached under ``source`` (see `annex_fingerprint`) and the
+    version of nwbinspector (which the caller passes so that it is part of the
+    key).
+    """
+    # Avoid heavy import by importing within function:
+    from nwbinspector import inspect_nwbfile, inspect_nwbfile_object
+
+    del nwbinspector_version
+    kwargs = {"config": config, "importance_threshold": importance_threshold}
+    if isinstance(source, str):
+        return list(inspect_nwbfile(nwbfile_path=source, skip_validate=True, **kwargs))
+
+    # Avoid heavy import by importing within function:
+    import h5py
+    from pynwb import NWBHDF5IO
+
+    from dandi.pynwb_utils import open_readable
+
+    messages = []
+    try:
+        with open_readable(source) as fp, h5py.File(fp, "r") as h5, NWBHDF5IO(
+            file=h5, mode="r", load_namespaces=True
+        ) as io:
+            nwbfile = io.read()
+            for message in inspect_nwbfile_object(nwbfile_object=nwbfile, **kwargs):
+                if message is not None:
+                    message.file_path = filepath
+                    messages.append(message)
+    except Exception as e:
+        raise _StreamedReadError() from e
+    return messages
+
+
+@cache
+def _cached_inspect_nwbfile() -> Callable[..., list[Any]]:
+    """`_inspect_nwbfile()` memoized in the validation cache"""
+    # Avoid heavy import by importing within function:
+    from dandi.pynwb_utils import validate_cache
+
+    return cast(
+        "Callable[..., list[Any]]",
+        validate_cache.memoize_path(custom_fingerprint=annex_fingerprint)(
+            _inspect_nwbfile
+        ),
+    )
+
 
 class VideoAsset(LocalFileAsset):
     pass

dandi/files/bids.py

@@ -262,7 +262,7 @@ def get_bids_metadata(self, digest: Digest | None = None) -> BareAsset:
         """
         metadata = self.bids_dataset_description.get_asset_metadata(self)
         start_time = end_time = datetime.now().astimezone()
-        add_common_metadata(metadata, self.filepath, start_time, end_time, digest)
+        add_common_metadata(metadata, self._content(), start_time, end_time, digest)
         metadata.path = self.path
         return metadata