diff --git a/dandi/cli/cmd_validate.py b/dandi/cli/cmd_validate.py index 5241c5eb4..c716925c8 100644 --- a/dandi/cli/cmd_validate.py +++ b/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. diff --git a/dandi/cli/tests/test_cmd_validate.py b/dandi/cli/tests/test_cmd_validate.py index 1da5c16ca..e3f4186af 100644 --- a/dandi/cli/tests/test_cmd_validate.py +++ b/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. diff --git a/dandi/files/bases.py b/dandi/files/bases.py index f2d0f22cd..7c4f14c69 100644 --- a/dandi/files/bases.py +++ b/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 diff --git a/dandi/files/bids.py b/dandi/files/bids.py index 371f7cd78..27823b296 100644 --- a/dandi/files/bids.py +++ b/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 diff --git a/dandi/pynwb_utils.py b/dandi/pynwb_utils.py index b0dda71c0..c065dad77 100644 --- a/dandi/pynwb_utils.py +++ b/dandi/pynwb_utils.py @@ -611,8 +611,9 @@ def rename_if_matched(_name: str, obj: Any) -> None: f.visititems(rename_if_matched) -@validate_cache.memoize_path -def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResult]: +def validate( + path: str | Path, devel_debug: bool = False, *, readable: Readable | None = None +) -> list[ValidationResult]: """Run validation on a file and return errors In case of an exception being thrown, an error message added to the @@ -621,8 +622,38 @@ def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResu Parameters ---------- path: str or Path + Path of the file, as reported in the returned results + devel_debug: bool, optional + Whether to re-raise exceptions instead of reporting them as errors + readable: Readable, optional + If given, the file's content is read from this `Readable` instead of from + ``path`` (e.g., to stream the content of an annexed file which is not + present locally); the results are then cached only if it is an + `AnnexedReadableFile` (see `annex_fingerprint`). + """ + source: str | Readable = readable if readable is not None else str(path) + # fscacher is untyped, so the memoized _validate returns Any for mypy + return cast( + "list[ValidationResult]", + _validate(source, str(path), devel_debug=devel_debug), + ) + + +@validate_cache.memoize_path(custom_fingerprint=annex_fingerprint) +def _validate( + source: str | Readable, path: str, devel_debug: bool = False +) -> list[ValidationResult]: + """`validate` proper, with the content source as first argument for caching + + Parameters + ---------- + source: str or Readable + What to read the file's content from: its path or a `Readable` + path: str + Path of the file, as reported in the returned results + devel_debug: bool + Whether to re-raise exceptions instead of reporting them as errors """ - path = str(path) # Might come in as pathlib's PATH errors: list[ValidationResult] = [] # To overcome @@ -634,7 +665,7 @@ def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResu ) version = None try: - version = get_nwb_version(path, sanitize=False) + version = get_nwb_version(source, sanitize=False) except Exception: # we just will not remove any errors, it is required so should be some pass @@ -680,9 +711,15 @@ def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResu ) try: - # Validates against the namespaces cached in the file; pynwb falls back - # to its own namespaces if the file has none cached - error_outputs = pynwb.validate(path=path) + # Either way, validates against the namespaces cached in the file; pynwb + # falls back to its own namespaces if the file has none cached + if isinstance(source, Readable): + with open_readable(source) as fp, h5py.File(fp, "r") as h5, NWBHDF5IO( + file=h5, mode="r", load_namespaces=True + ) as reader: + error_outputs = pynwb.validate(io=reader) + else: + error_outputs = pynwb.validate(path=source) except Exception as exc: if devel_debug: raise diff --git a/dandi/validate/_core.py b/dandi/validate/_core.py index 371ce5a30..ec90b30f8 100644 --- a/dandi/validate/_core.py +++ b/dandi/validate/_core.py @@ -9,9 +9,18 @@ from __future__ import annotations +from collections import deque from collections.abc import Iterator +from concurrent.futures import Future, ProcessPoolExecutor +from copy import copy +from functools import partial +import logging +from logging.handlers import QueueHandler, QueueListener +from multiprocessing import get_context import os from pathlib import Path +import shutil +from threading import Lock from typing import Any from ._types import ( @@ -25,10 +34,25 @@ ValidationResult, Validator, ) +from .. import get_logger from ..consts import dandiset_metadata_file -from ..files import find_dandi_files +from ..files import ( + BIDSAsset, + DandiFile, + LocalAsset, + LocalFileAsset, + NWBAsset, + find_dandi_files, +) +from ..support.annex_stream import ( + AnnexedReadableFile, + can_stream, + get_annexed_readable, +) from ..utils import find_parent_directory_containing +lgr = get_logger() + BIDS_TO_DANDI = { "subject": "subject_id", "session": "session_id", @@ -155,7 +179,8 @@ def _is_broken_symlink(filepath: Path) -> bool: # BIDS error codes that require reading file content (headers, pixel data). -# When ``only-non-data`` is active these are suppressed for broken-symlink files. +# When ``only-non-data`` or ``stream`` is active these are suppressed for +# broken-symlink files (the BIDS validator cannot stream content). _BIDS_CONTENT_DEPENDENT_CODES = frozenset( { "BIDS.NIFTI_HEADER_UNREADABLE", @@ -170,6 +195,7 @@ def validate( devel_debug: bool = False, allow_any_path: bool = False, missing_file_content: MissingFileContent = MissingFileContent.error, + jobs: int = 1, ) -> Iterator[ValidationResult]: """Validate content @@ -181,13 +207,25 @@ def validate( Policy for files whose content is unavailable (e.g. broken symlinks in a datalad dataset without fetched data). ``error`` emits a concise error, ``skip`` skips the file with a warning, ``only-non-data`` skips - content-dependent validators but still validates path layout. + content-dependent validators but still validates path layout, and + ``stream`` streams the content of annexed files (see + `dandi.support.annex_stream`) so that + content-dependent validators run without the content being present + locally. + jobs : int + Number of processes in which to validate NWB files (other than those of + BIDS datasets) in parallel; the results are yielded in the same order + as with 1 (the default), with which everything is validated in this + process. Yields ------ path, errors errors for a path """ + if missing_file_content == MissingFileContent.stream: + _check_streaming_requirements() + # Archive of unique `ValidationResult` objects obtained through # `DandiFile.get_validation_errors()` # Note: This is needed to hold on to the unique `ValidationResult` objects @@ -198,75 +236,319 @@ def validate( # The ids of the objects in `df_results` obtain through the `id()` built-in function df_result_ids: set[int] = set() + def unique(results: list[ValidationResult]) -> Iterator[ValidationResult]: + for r in results: + r_id = id(r) + if r_id not in df_result_ids: + df_results.append(r) + df_result_ids.add(r_id) + yield r + + # The results of the files, in order, either computed or being computed + # in `pool` + pending: deque[list[ValidationResult] | Future[list[ValidationResult]]] + pending = deque() + # The files to validate (and the results for paths not in a Dandiset), all + # found first so that the progress can be reported as a count + items: list[DandiFile | ValidationResult] = [] + # The `dataset_description.json` files of BIDS datasets, which the + # `BIDSAsset` files found refer to only weakly + bids_descriptions = [] for p in paths: p = os.path.abspath(p) dandiset_path = find_parent_directory_containing(dandiset_metadata_file, p) if dandiset_path is None: - yield ValidationResult( - id="DANDI.NO_DANDISET_FOUND", - origin=ORIGIN_VALIDATION_DANDI_LAYOUT, - severity=Severity.ERROR, - scope=Scope.DANDISET, - path=Path(p), - message="Path is not inside a Dandiset", + items.append( + ValidationResult( + id="DANDI.NO_DANDISET_FOUND", + origin=ORIGIN_VALIDATION_DANDI_LAYOUT, + severity=Severity.ERROR, + scope=Scope.DANDISET, + path=Path(p), + message="Path is not inside a Dandiset", + ) ) for df in find_dandi_files( p, dandiset_path=dandiset_path, allow_all=allow_any_path ): + if isinstance(df, BIDSAsset): + bids_descriptions.append(df.bids_dataset_description) + items.append(df) + progress = _Progress(sum(isinstance(i, DandiFile) for i in items)) + + pool = _ValidationPool(jobs) if jobs > 1 else None + try: + for item in items: + if isinstance(item, ValidationResult): + pending.append([item]) + continue + df = item # Handle broken symlinks (missing file content) - if _is_broken_symlink(df.filepath): - r = _handle_missing_content(df, missing_file_content) - if r is not None: - r_id = id(r) - if r_id not in df_result_ids: - df_results.append(r) - df_result_ids.add(r_id) - yield r - if missing_file_content in ( - MissingFileContent.skip, - MissingFileContent.error, - ): - continue - # only-non-data: fall through but pass the flag to validators - is_broken = _is_broken_symlink(df.filepath) - for r in df.get_validation_errors( - schema_version=schema_version, - devel_debug=devel_debug, - missing_file_content=(missing_file_content if is_broken else None), - ): - # For broken-symlink files under only-non-data, suppress - # BIDS errors that require reading file content (e.g. - # NIFTI_HEADER_UNREADABLE). The validator ran in full so - # real files still get those checks. - if ( - is_broken - and missing_file_content == MissingFileContent.only_non_data - and r.id in _BIDS_CONTENT_DEPENDENT_CODES - ): + if is_broken: + if missing_file_content == MissingFileContent.stream: + readable = _prepare_streaming(df) + # A file whose content cannot be streamed gets an error + # and is otherwise skipped + skip_file = readable is None + else: + readable = None + skip_file = missing_file_content in ( + MissingFileContent.skip, + MissingFileContent.error, + ) + pending.append( + [_handle_missing_content(df, missing_file_content, readable)] + ) + if skip_file: + progress.done(df, None) continue - r_id = id(r) - if r_id not in df_result_ids: - df_results.append(r) - df_result_ids.add(r_id) - yield r + # only-non-data & stream: fall through but pass the policy + # to the validators + + file_policy = missing_file_content if is_broken else None + if pool is not None and type(df) is NWBAsset: + future = pool.submit(df, schema_version, devel_debug, file_policy) + future.add_done_callback(partial(progress.future_done, df)) + pending.append(future) + else: + results = _validate_file(df, schema_version, devel_debug, file_policy) + progress.done(df, results) + pending.append(results) + # Yield what is ready, or wait (for the oldest file) if enough + # files are being validated + while pending and ( + isinstance(pending[0], list) + or pending[0].done() + or (pool is not None and len(pending) > 2 * jobs) + ): + head = pending.popleft() + yield from unique(head if isinstance(head, list) else head.result()) + while pending: + head = pending.popleft() + yield from unique(head if isinstance(head, list) else head.result()) + finally: + if pool is not None: + pool.shutdown() + + +class _Progress: + """Report the progress of validating ``total`` files with log messages""" + + def __init__(self, total: int) -> None: + self.total = total + self.count = 0 + # Files may be done in the threads of a `_ValidationPool` + self.lock = Lock() + if total: + lgr.info("Validating %d file%s", total, "" if total == 1 else "s") + + def done( + self, + df: DandiFile, + results: list[ValidationResult] | None, + outcome: str = "skipped", + ) -> None: + """ + Report that validating ``df`` is done, with the given results, or + `None` if it was not validated, for the reason ``outcome`` + """ + with self.lock: + self.count += 1 + if results is None: + pass + elif results: + outcome = f"{len(results)} issue{'' if len(results) == 1 else 's'}" + else: + outcome = "OK" + path = df.path if isinstance(df, LocalAsset) else df.filepath + lgr.info("[%d/%d] %s: %s", self.count, self.total, path, outcome) + + def future_done( + self, df: DandiFile, future: Future[list[ValidationResult]] + ) -> None: + """Report that validating ``df`` in a `_ValidationPool` is done""" + if future.cancelled() or future.exception() is not None: + self.done(df, None, outcome="failed") + else: + self.done(df, future.result()) + + +def _validate_file( + df: DandiFile, + schema_version: str | None, + devel_debug: bool, + missing_file_content: MissingFileContent | None, +) -> list[ValidationResult]: + """ + Return the results of validating ``df``; ``missing_file_content`` is the + policy for its missing content, if it is missing + """ + results = [] + for r in df.get_validation_errors( + schema_version=schema_version, + devel_debug=devel_debug, + missing_file_content=missing_file_content, + ): + # For broken-symlink files under only-non-data and stream, suppress + # BIDS errors that require reading file content (e.g. + # NIFTI_HEADER_UNREADABLE). The validator ran in full so real files + # still get those checks. + if ( + missing_file_content + in (MissingFileContent.only_non_data, MissingFileContent.stream) + and r.id in _BIDS_CONTENT_DEPENDENT_CODES + ): + continue + results.append(r) + return results + + +def _validate_file_in_worker( + df: LocalFileAsset, + schema_version: str | None, + devel_debug: bool, + missing_file_content: MissingFileContent | None, +) -> list[ValidationResult]: + """`_validate_file()` in a worker process of a `_ValidationPool`""" + if missing_file_content == MissingFileContent.stream: + # The streamed content is opened in this process + _prepare_streaming(df) + return _validate_file(df, schema_version, devel_debug, missing_file_content) + + +def _init_worker(log_queue: Any, levels: dict[str, int]) -> None: + """ + Set up a worker process of a `_ValidationPool`: send the log records to + the parent process, and ignore the warnings that it ignores + """ + # Avoid heavy import by importing within function: + from ..pynwb_utils import ignore_benign_pynwb_warnings + + root = logging.getLogger() + root.handlers = [QueueHandler(log_queue)] + for name, level in levels.items(): + logging.getLogger(name).setLevel(level) + ignore_benign_pynwb_warnings() + + +class _ValidationPool: + """ + Processes in which files are validated in parallel (h5py serializes all + reading of HDF5 files in a process), whose log records are handled by the + handlers of this process + """ + + def __init__(self, jobs: int) -> None: + # Not "fork": this process may have started threads (e.g., fsspec's + # for streaming content) + context = get_context("spawn") + self.log_queue = context.Queue() + self.listener = QueueListener( + self.log_queue, *logging.getLogger().handlers, respect_handler_level=True + ) + self.listener.start() + levels = {name: logging.getLogger(name or None).level for name in ("", "dandi")} + self.executor = ProcessPoolExecutor( + max_workers=jobs, + mp_context=context, + initializer=_init_worker, + initargs=(self.log_queue, levels), + ) + + def submit( + self, + df: LocalFileAsset, + schema_version: str | None, + devel_debug: bool, + missing_file_content: MissingFileContent | None, + ) -> Future[list[ValidationResult]]: + # The source of streamed content cannot be sent to another process; + # it is set up again there + df = copy(df) + df.content_source = None + return self.executor.submit( + _validate_file_in_worker, + df, + schema_version, + devel_debug, + missing_file_content, + ) + + def shutdown(self) -> None: + self.executor.shutdown(cancel_futures=True) + self.listener.stop() + self.log_queue.close() + + +def _check_streaming_requirements() -> None: + """Raise an informative error if what is needed for streaming file content + is not installed""" + if not can_stream(): + raise RuntimeError( + "Streaming file content requires datalad-fuse with remfile; install " + "them with `pip install 'dandi[datalad]'`" + ) + if shutil.which("git-annex") is None: + raise RuntimeError("Streaming file content requires git-annex") + + +def _prepare_streaming(df: DandiFile) -> AnnexedReadableFile | None: + """ + Set up streaming of the content of the annexed file represented by ``df``, + and return the `Readable` that will be used for it, or `None` if the + content cannot be streamed (see `get_annexed_readable()`) + """ + if not isinstance(df, LocalFileAsset): + return None + readable = get_annexed_readable(df.filepath) + if readable is not None: + df.content_source = readable + return readable def _handle_missing_content( - df: Any, + df: DandiFile, policy: MissingFileContent, -) -> ValidationResult | None: + readable: AnnexedReadableFile | None = None, +) -> ValidationResult: """Produce a single :class:`ValidationResult` for a file with missing content. - Returns ``None`` when *policy* is ``only-non-data`` (a warning is not - needed because validation still proceeds on the non-data aspects). + For the ``stream`` policy, ``readable`` is the `Readable` streaming the + file's content, or `None` if the content cannot be streamed. """ - from ..files import DandiFile - - assert isinstance(df, DandiFile) filepath = df.filepath - if policy == MissingFileContent.error: + if policy == MissingFileContent.stream: + if readable is not None: + return ValidationResult( + id="DANDI.FILE_CONTENT_STREAMED", + origin=ORIGIN_VALIDATION_DANDI_LAYOUT, + severity=Severity.INFO, + scope=Scope.FILE, + path=filepath, + dandiset_path=df.dandiset_path, + message=( + f"File content is not present locally (git-annex key " + f"{readable.key}); content-dependent validation streams " + f"it from {readable.url}" + ), + ) + return ValidationResult( + id="DANDI.FILE_CONTENT_MISSING", + origin=ORIGIN_VALIDATION_DANDI_LAYOUT, + severity=Severity.ERROR, + scope=Scope.FILE, + path=filepath, + dandiset_path=df.dandiset_path, + message=( + f"File content is not available (broken symlink: " + f"{filepath} -> {os.readlink(filepath)}) and cannot be " + f"streamed: not an annexed file in a git-annex repository " + f"with a URL known to git-annex." + ), + ) + elif policy == MissingFileContent.error: return ValidationResult( id="DANDI.FILE_CONTENT_MISSING", origin=ORIGIN_VALIDATION_DANDI_LAYOUT, @@ -277,8 +559,9 @@ def _handle_missing_content( message=( f"File content is not available (broken symlink: " f"{filepath} -> {os.readlink(filepath)}). " - f"Use --missing-file-content=skip or " - f"--missing-file-content=only-non-data to handle gracefully." + f"Use --missing-file-content=skip, " + f"--missing-file-content=only-non-data, or " + f"--missing-file-content=stream to handle gracefully." ), ) elif policy == MissingFileContent.skip: diff --git a/dandi/validate/_types.py b/dandi/validate/_types.py index f41c8402e..56e2e1f62 100644 --- a/dandi/validate/_types.py +++ b/dandi/validate/_types.py @@ -26,6 +26,12 @@ class MissingFileContent(StrEnum): skip = auto() """Skip the file entirely; emit a WARNING noting that validation was skipped.""" + stream = auto() + """Stream the content of annexed files (see + `dandi.support.annex_stream`) so that content-dependent validators run + without the content being present locally. An INFO result is emitted + for each streamed file, and an ERROR for each file that cannot be streamed.""" + lgr = logging.getLogger(__name__) diff --git a/dandi/validate/tests/test_core.py b/dandi/validate/tests/test_core.py index 7d331bb61..967142296 100644 --- a/dandi/validate/tests/test_core.py +++ b/dandi/validate/tests/test_core.py @@ -1,5 +1,8 @@ import json +import logging +import os from pathlib import Path +import shutil from typing import Any import pytest @@ -17,7 +20,12 @@ ) from ... import __version__ from ...consts import dandiset_metadata_file -from ...tests.fixtures import BIDS_TESTDATA_SELECTION +from ...files import NWBAsset, dandi_file +from ...pynwb_utils import validate as pynwb_validate +from ...support.annex_stream import get_annexed_readable +from ...tests.fixtures import BIDS_TESTDATA_SELECTION, make_git_annex_dandiset +from ...tests.range_http_server import RangeHTTPServer +from ...tests.skip import skipif def test_validate_nwb_error(simple3_nwb: Path) -> None: @@ -310,3 +318,247 @@ def test_validate_broken_symlink_real_file_still_validated(tmp_path: Path) -> No assert ( len(broken_pynwb) == 0 ), f"policy={policy.value}: pynwb should not run on the broken symlink" + + +# ---- Tests for streaming the content of annexed files (missing_file_content=stream) ---- + + +@pytest.fixture +def streamable_dandiset( + range_http_server: tuple[RangeHTTPServer, str], simple3_nwb: Path, tmp_path: Path +) -> tuple[Path, RangeHTTPServer]: + """A git-annex dandiset looking like a DataLad clone without fetched content. + + ``sub-001/sub-001.nwb`` is an annexed copy of *simple3_nwb* whose content + is served by the HTTP server, and ``sub-001/sub-001_video.mp4`` an annexed + video registered at a URL that is never read (only the size recorded in + its key is needed). + """ + skipif.no_git_annex() + pytest.importorskip("datalad_fuse.adapter") + pytest.importorskip("remfile") + server, base_url = range_http_server + shutil.copy(simple3_nwb, tmp_path / "served" / "content.nwb") + video = tmp_path / "video.mp4" + video.write_bytes(b"not really a video") + ds = tmp_path / "ds" + make_git_annex_dandiset( + ds, + { + "sub-001/sub-001.nwb": (simple3_nwb, f"{base_url}/content.nwb"), + "sub-001/sub-001_video.mp4": (video, f"{base_url}/never-read.mp4"), + }, + ) + return ds, server + + +def _content_results(results: list[ValidationResult], name: str) -> list[tuple]: + """Path-independent summary of the results for the file called *name*.""" + return sorted( + (r.id, r.severity, r.origin.validator, r.message) + for r in results + if r.path is not None + and r.path.name == name + and r.id != "DANDI.FILE_CONTENT_STREAMED" + ) + + +@pytest.mark.ai_generated +def test_validate_stream( + streamable_dandiset: tuple[Path, RangeHTTPServer], + simple3_nwb: Path, + tmp_path: Path, +) -> None: + """stream policy validates the content of annexed files by streaming it.""" + ds, server = streamable_dandiset + results = list(validate(ds, missing_file_content=MissingFileContent.stream)) + + streamed = [r for r in results if r.id == "DANDI.FILE_CONTENT_STREAMED"] + assert sorted(r.path.name for r in streamed if r.path is not None) == [ + "sub-001.nwb", + "sub-001_video.mp4", + ] + assert all(r.severity == Severity.INFO for r in streamed) + assert not [r for r in results if r.id.startswith("DANDI.FILE_CONTENT_MISSING")] + assert server.ranged_requests > 0 + + # Validating the video only needs the size recorded in its key + assert _content_results(results, "sub-001_video.mp4") == [] + + # Content-dependent validation of the NWB file gives the same results as + # for a regular dandiset containing the file itself + local = tmp_path / "local" + (local / "sub-001").mkdir(parents=True) + shutil.copy(ds / dandiset_metadata_file, local / dandiset_metadata_file) + shutil.copy(simple3_nwb, local / "sub-001" / "sub-001.nwb") + expected = list(validate(local)) + # simple3_nwb lacks a subject_id, which only content-based checks notice + assert any(r.origin.validator == Validator.nwbinspector for r in expected) + assert _content_results(results, "sub-001.nwb") == _content_results( + expected, "sub-001.nwb" + ) + + +@pytest.mark.ai_generated +@pytest.mark.skipif( + os.environ.get("DANDI_CACHE") == "ignore", reason="the validation cache is disabled" +) +def test_validate_stream_cached_by_key( + streamable_dandiset: tuple[Path, RangeHTTPServer], +) -> None: + """pynwb validation results of a streamed file are cached under its annex key.""" + ds, server = streamable_dandiset + nwb = ds / "sub-001" / "sub-001.nwb" + readable = get_annexed_readable(nwb) + assert readable is not None + results = pynwb_validate(nwb, readable=readable) + assert not [r for r in results if r.id == "pynwb.GENERIC"] + requests = server.ranged_requests + assert requests > 0 + # Served from the cache, without the content being read again + assert pynwb_validate(nwb, readable=get_annexed_readable(nwb)) == results + assert server.ranged_requests == requests + + +def _streamed_nwb_asset(ds: Path) -> NWBAsset: + nwb = ds / "sub-001" / "sub-001.nwb" + df = dandi_file(nwb, ds) + assert isinstance(df, NWBAsset) + df.content_source = get_annexed_readable(nwb) + assert df.content_source is not None + return df + + +@pytest.mark.ai_generated +@pytest.mark.skipif( + os.environ.get("DANDI_CACHE") == "ignore", reason="the validation cache is disabled" +) +def test_validate_stream_inspection_cached_by_key( + streamable_dandiset: tuple[Path, RangeHTTPServer], +) -> None: + """NWB Inspector results of a streamed file are cached under its annex key.""" + ds, server = streamable_dandiset + results = _streamed_nwb_asset(ds).get_validation_errors() + inspector = [r for r in results if r.origin.validator == Validator.nwbinspector] + # simple3_nwb lacks a subject_id + assert inspector + requests = server.ranged_requests + # Served from the cache, without the content being read again + assert _streamed_nwb_asset(ds).get_validation_errors() == results + assert server.ranged_requests == requests + + +@pytest.mark.ai_generated +def test_validate_stream_inspection_failure_not_cached( + streamable_dandiset: tuple[Path, RangeHTTPServer], tmp_path: Path +) -> None: + """A failure to read streamed content for NWB Inspector is not cached.""" + ds, _ = streamable_dandiset + served = tmp_path / "served" / "content.nwb" + content = served.read_bytes() + served.unlink() + df = _streamed_nwb_asset(ds) + [msg] = df._inspect_nwbfile() + assert msg.importance.name == "ERROR" + assert msg.check_function_name.startswith("During io.read()") + assert msg.file_path == str(df.filepath) + served.write_bytes(content) + assert all( + m.check_function_name != msg.check_function_name + for m in _streamed_nwb_asset(ds)._inspect_nwbfile() + ) + + +@pytest.mark.ai_generated +@pytest.mark.parametrize("jobs", [1, 2]) +def test_validate_progress( + streamable_dandiset: tuple[Path, RangeHTTPServer], + simple3_nwb: Path, + caplog: pytest.LogCaptureFixture, + jobs: int, +) -> None: + """The progress of validation is logged, one message per file.""" + ds, _ = streamable_dandiset + (ds / "sub-002").mkdir() + shutil.copy(simple3_nwb, ds / "sub-002" / "sub-002.nwb") + caplog.set_level(logging.INFO, logger="dandi") + list(validate(ds, missing_file_content=MissingFileContent.stream, jobs=jobs)) + messages = [r.getMessage() for r in caplog.records] + messages = messages[messages.index("Validating 3 files") :] + done = [m for m in messages if m.startswith("[")] + assert sorted(m.split("] ")[0] for m in done) == ["[1/3", "[2/3", "[3/3"] + assert any(m.endswith("] sub-001/sub-001_video.mp4: OK") for m in done) + assert any("sub-002/sub-002.nwb: " in m and "issue" in m for m in done) + + +@pytest.mark.ai_generated +def test_validate_jobs( + streamable_dandiset: tuple[Path, RangeHTTPServer], simple3_nwb: Path +) -> None: + """Validating in parallel processes gives the same results, in order.""" + ds, server = streamable_dandiset + for sub in ("sub-002", "sub-003"): + (ds / sub).mkdir() + shutil.copy(simple3_nwb, ds / sub / f"{sub}.nwb") + # In parallel first, so that the streamed content is read in a worker + # process rather than the results taken from the cache + parallel = list( + validate(ds, missing_file_content=MissingFileContent.stream, jobs=2) + ) + assert server.ranged_requests > 0 + serial = list(validate(ds, missing_file_content=MissingFileContent.stream)) + assert any(r.origin.validator == Validator.nwbinspector for r in serial) + + def summary(results: list[ValidationResult]) -> list[tuple]: + return [(r.id, r.path, r.origin.validator, r.message) for r in results] + + assert summary(parallel) == summary(serial) + + +@pytest.mark.ai_generated +def test_validate_stream_not_streamable(tmp_path: Path) -> None: + """stream policy emits an error for a broken symlink that cannot be streamed.""" + pytest.importorskip("datalad_fuse.adapter") + pytest.importorskip("remfile") + skipif.no_git_annex() + ds = _make_dandiset_with_broken_symlink(tmp_path) + results = list(validate(ds, missing_file_content=MissingFileContent.stream)) + errs = [r for r in results if r.id == "DANDI.FILE_CONTENT_MISSING"] + assert len(errs) == 1 + assert errs[0].severity == Severity.ERROR + assert errs[0].message is not None + assert "cannot be streamed" in errs[0].message + assert not [r for r in results if r.id == "DANDI.FILE_CONTENT_STREAMED"] + assert not [ + r + for r in results + if r.origin.validator in (Validator.pynwb, Validator.nwbinspector) + ] + + +@pytest.mark.ai_generated +def test_validate_stream_unreadable_url( + streamable_dandiset: tuple[Path, RangeHTTPServer], tmp_path: Path +) -> None: + """A registered URL that cannot be read yields errors, not an exception.""" + ds, _ = streamable_dandiset + (tmp_path / "served" / "content.nwb").unlink() + results = list(validate(ds, missing_file_content=MissingFileContent.stream)) + nwb = ds / "sub-001" / "sub-001.nwb" + assert [ + r for r in results if r.path == nwb and r.id == "DANDI.FILE_CONTENT_STREAMED" + ] + errs = [r for r in results if r.path == nwb and r.severity == Severity.ERROR] + assert {r.origin.validator for r in errs} == { + Validator.pynwb, + Validator.nwbinspector, + } + + +@pytest.mark.ai_generated +def test_validate_stream_requires_datalad_fuse( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr("dandi.validate._core.can_stream", lambda: False) + with pytest.raises(RuntimeError, match=r"dandi\[datalad\]"): + list(validate(tmp_path, missing_file_content=MissingFileContent.stream)) diff --git a/docs/source/cmdline/validate.rst b/docs/source/cmdline/validate.rst index 3a07d7e7a..541956959 100644 --- a/docs/source/cmdline/validate.rst +++ b/docs/source/cmdline/validate.rst @@ -56,7 +56,7 @@ Options Limit the number of results shown per group (or in total when not grouping); the excess is replaced by a count of omitted results -.. option:: --missing-file-content [error|only-non-data|skip] +.. option:: --missing-file-content [error|only-non-data|skip|stream] How to handle files whose content is unavailable, such as the broken symbolic links of a DataLad_ dataset (a git-annex_ repository) whose @@ -73,6 +73,13 @@ Options Skip content-dependent validators (pynwb, nwbinspector, ...) for each such file but still validate its path layout + ``stream`` + Stream the content of each such file from the URLs git-annex knows + for it so that + content-dependent validators run without the file having to be + downloaded; see `Validating DataLad Dandisets Remotely`_ + below. + .. option:: --load Instead of running validation, load previously saved results from the @@ -84,6 +91,66 @@ Options .. _git-annex: https://git-annex.branchable.com +Validating DataLad Dandisets Remotely +------------------------------------- + +Every Dandiset on the DANDI Archive is mirrored as a DataLad dataset at +https://github.com/dandisets (with https://github.com/dandisets/dandisets as +the superdataset containing all of them). In such a dataset, the assets are +annexed files: symbolic links that remain broken until the content is fetched, +which for some Dandisets would mean downloading terabytes of data. The +``stream`` policy of :option:`--missing-file-content` lets ``dandi validate`` +run the content-dependent validators (pynwb and nwbinspector for NWB files) on +such files by streaming their content on demand: it asks git-annex where the +content is (``git annex whereis``) and reads only the parts of each file that +the validators need. This is done with datalad-fuse_ (its adapter, without +mounting anything), which reads HDF5-based files such as NWB with remfile_, +optimized for their access pattern. This requires git-annex and ``pip install +"dandi[datalad]"``. + +For example, to produce a JSON Lines record of *all* validation results for +a Dandiset:: + + datalad clone https://github.com/dandisets/000003 + dandi validate --missing-file-content=stream --min-severity=INFO \ + --format=json_lines --output=000003.jsonl 000003 + +Each streamed file yields an ``INFO``-level ``DANDI.FILE_CONTENT_STREAMED`` +result naming the first URL known for its content (for a Dandiset, the DANDI +Archive's API download URL, which redirects to S3); a file that cannot be streamed +(not an annexed file, or no URL known to git-annex) yields a +``DANDI.FILE_CONTENT_MISSING`` error instead. + +Notes: + +- git-annex must be initialized in the clone, which ``datalad clone`` does + (after ``git clone``, run ``git annex init``): ``dandi validate`` does not + run any ``git annex`` command in a clone in which it is not, as that would + initialize it, modifying the clone. +- Some nwbinspector checks read data arrays (e.g., timestamps), so the amount + of data streamed for a file depends on its content; it is nevertheless + usually a small fraction of the file. +- The pynwb validation results and the metadata of a streamed file are cached + under its git-annex key (a digest of its content), just as those of a local + file are cached under its modification time and size, so re-running the + command skips that work for files whose key has not changed since. As for + local files, set the :envvar:`DANDI_CACHE` environment variable to ``clear`` + to start afresh or to ``ignore`` to bypass the cache. +- Zarr assets are stored as separate subdatasets (https://github.com/dandizarrs) + and are not streamed yet: an uninstalled Zarr subdataset is an empty directory + that is not validated at all. Streaming them is a follow-up for when NWB Zarr + support has matured across the ecosystem. +- The BIDS validator cannot stream content, so BIDS errors that require reading + a file (e.g., unreadable NIfTI headers) are suppressed for annexed files under + the ``stream`` and ``only-non-data`` policies. For NWB datasets, the primary + use case, nothing is lost: everything the BIDS validator needs is either + present (the non-annexed sidecar files, which are kept in git) or encoded in + the file and folder names of the annexed files themselves. + +.. _datalad-fuse: https://github.com/datalad/datalad-fuse +.. _remfile: https://github.com/flatironinstitute/remfile + + Development Options ------------------- diff --git a/tox.ini b/tox.ini index 63dbff4f9..9974aab59 100644 --- a/tox.ini +++ b/tox.ini @@ -88,6 +88,9 @@ filterwarnings = ignore:.*unclosed.*:ResourceWarning:vcr # addressed in joblib 0.8.2-826-g05caf07 ignore:(ast.Num|Attribute n) is deprecated.*:DeprecationWarning:joblib + # Advisory warning of joblib (used by fscacher) when saving the arguments of + # a cached call takes over 0.5 s, e.g., on a slow filesystem + ignore:Persisting input arguments took:UserWarning:joblib [coverage:run] parallel = True