Merge pull request #166 from NeurodataWithoutBorders/inspect_on_dandisets

[Feature]: Direct inspection on Dandisets

Author: CodyCBakerPhD

.github/workflows/testing.yml

@@ -10,18 +10,23 @@ jobs:
         os: ["ubuntu-latest", "macos-latest", "windows-latest"]
         python-version: ["3.7", "3.10"]
     steps:
-      - uses: actions/checkout@v2
-      - run: git fetch --prune --unshallow --tags
-      - name: Setup Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v2
+      - uses: s-weigand/setup-conda@v1
         with:
+          update-conda: true
           python-version: ${{ matrix.python-version }}
+          conda-channels: conda-forge
+      - uses: actions/checkout@v2
+      - run: git fetch --prune --unshallow --tags
       - name: Install pytest
         run: |
           pip install pytest
           pip install pytest-cov
       - name: Install package
         run: pip install -e .
+      - name: Uninstall h5py
+        run: pip uninstall -y h5py
+      - name: Install ROS3
+        run: conda install -c conda-forge h5py
       - name: Run pytest with coverage
         run: pytest --cov=./ --cov-report xml:./nwbinspector/nwbinspector/coverage.xml
       - if: ${{ matrix.python-version == '3.10' && matrix.os == 'ubuntu-latest' }}

nwbinspector/nwbinspector.py

@@ -1,5 +1,6 @@
 """Primary functions for inspecting NWBFiles."""
 import os
+import re
 import importlib
 import traceback
 import json
@@ -10,7 +11,7 @@
 from typing import Optional, List
 from concurrent.futures import ProcessPoolExecutor, as_completed
 from types import FunctionType
-from warnings import filterwarnings
+from warnings import filterwarnings, warn
 from distutils.util import strtobool
 
 import click
@@ -26,6 +27,7 @@
     save_report,
 )
 from .register_checks import InspectorMessage, Importance
+from .tools import get_s3_urls_and_dandi_paths
 from .utils import FilePathType, PathType, OptionalListOfStrings
 
 INTERNAL_CONFIGS = dict(dandi=Path(__file__).parent / "internal_configs" / "dandi.inspector_config.yaml")
@@ -186,6 +188,21 @@ def configure_checks(
     is_flag=True,
 )
 @click.option("--progress-bar", help="Set this flag to False to disable display of the progress bar.")
+@click.option(
+    "--stream",
+    help=(
+        "Stream data from the DANDI archive. If the 'path' is a local copy of the target DANDISet, specifying this "
+        "flag will still force the data to be streamed instead of using the local copy. To use the local copy, simply "
+        "remove this flag. Requires the Read Only S3 (ros3) driver to be installed with h5py."
+    ),
+    is_flag=True,
+)
+@click.option(
+    "--version-id",
+    help=(
+        "When 'path' is a six-digit DANDISet ID, this further specifies which version of " "the DANDISet to inspect."
+    ),
+)
 def inspect_all_cli(
     path: str,
     modules: Optional[str] = None,
@@ -203,18 +220,34 @@ def inspect_all_cli(
     skip_validate: bool = False,
     detailed: bool = False,
     progress_bar: Optional[str] = None,
+    stream: bool = False,
+    version_id: Optional[str] = None,
 ):
     """
     Run the NWBInspector via the command line.
 
     path :
-    Path to either a local NWBFile, a local folder containing NWBFiles.
+    Path to either a local NWBFile, a local folder containing NWBFiles, a link to a dataset on
+    DANDI archive (i.e., https://dandiarchive.org/dandiset/{dandiset_id}/{version_id}), or a six-digit Dandiset ID.
     """
     levels = ["importance", "file_path"] if levels is None else levels.split(",")
     reverse = [False] * len(levels) if reverse is None else [strtobool(x) for x in reverse.split(",")]
     progress_bar = strtobool(progress_bar) if progress_bar is not None else True
     if config is not None:
         config = load_config(filepath_or_keyword=config)
+    if stream:
+        url_path = path if path.startswith("https://") else None
+        if url_path:
+            dandiset_id, version_id = url_path.split("/")[-2:]
+            path = dandiset_id
+        assert url_path or re.fullmatch(
+            pattern="^[0-9]{6}$", string=path
+        ), "'--stream' flag was enabled, but 'path' is neither a full link to the DANDI archive nor a DANDISet ID."
+        if Path(path).is_dir():
+            warn(
+                f"The local DANDISet '{path}' exists, but the '--stream' flag was used. "
+                "NWBInspector will use S3 streaming from DANDI. To use local data, remove the '--stream' flag."
+            )
     messages = list(
         inspect_all(
             path=path,
@@ -226,6 +259,8 @@ def inspect_all_cli(
             n_jobs=n_jobs,
             skip_validate=skip_validate,
             progress_bar=progress_bar,
+            stream=stream,
+            version_id=version_id,
         )
     )
     if json_file_path is not None:
@@ -254,14 +289,17 @@ def inspect_all(
     skip_validate: bool = False,
     progress_bar: bool = True,
     progress_bar_options: Optional[dict] = None,
+    stream: bool = False,
+    version_id: Optional[str] = None,
 ):
     """
     Inspect a local NWBFile or folder of NWBFiles and return suggestions for improvements according to best practices.
 
     Parameters
     ----------
     path : PathType
-        File path to an NWBFile, or folder path to iterate over recursively and scan all NWBFiles present.
+        File path to an NWBFile, folder path to iterate over recursively and scan all NWBFiles present, or a
+        six-digit identifier of the DANDISet.
     modules : list of strings, optional
         List of external module names to load; examples would be namespace extensions.
         These modules may also contain their own custom checks for their extensions.
@@ -294,22 +332,46 @@ def inspect_all(
         Defaults to True.
     progress_bar_options : dict, optional
         Dictionary of keyword arguments to pass directly to tqdm.
+    stream : bool, optional
+        Stream data from the DANDI archive. If the 'path' is a local copy of the target DANDISet, setting this
+        argument to True will force the data to be streamed instead of using the local copy.
+        Requires the Read Only S3 (ros3) driver to be installed with h5py.
+        Defaults to False.
+    version_id : str, optional
+        If the path is a DANDISet ID, version_id additionally specifies which version of the dataset to read from.
+        Common options are 'draft' or 'published'.
+        Defaults to the most recent published version, or if not published then the most recent draft version.
     """
     modules = modules or []
     if progress_bar_options is None:
-        progress_bar_options = dict(desc="Inspecting NWBFiles...")
-    in_path = Path(path)
-    if in_path.is_dir():
-        nwbfiles = list(in_path.rglob("*.nwb"))
-    elif in_path.is_file():
-        nwbfiles = [in_path]
+        progress_bar_options = dict(position=0, leave=False)
+        if stream:
+            progress_bar_options.update(desc="Inspecting NWBFiles with ROS3...")
+        else:
+            progress_bar_options.update(desc="Inspecting NWBFiles...")
+    if stream:
+        assert (
+            re.fullmatch(pattern="^[0-9]{6}$", string=str(path)) is not None
+        ), "'--stream' flag was enabled, but 'path' is not a DANDISet ID."
+        driver = "ros3"
+        nwbfiles = get_s3_urls_and_dandi_paths(dandiset_id=path, version_id=version_id, n_jobs=n_jobs)
     else:
-        raise ValueError(f"{in_path} should be a directory or an NWB file.")
+        driver = None
+        in_path = Path(path)
+        if in_path.is_dir():
+            nwbfiles = list(in_path.rglob("*.nwb"))
+        elif in_path.is_file():
+            nwbfiles = [in_path]
+        else:
+            raise ValueError(f"{in_path} should be a directory or an NWB file.")
     for module in modules:
         importlib.import_module(module)
     # Filtering of checks should apply after external modules are imported, in case those modules have their own checks
     checks = configure_checks(config=config, ignore=ignore, select=select, importance_threshold=importance_threshold)
 
+    nwbfiles_iterable = nwbfiles
+    if progress_bar:
+        nwbfiles_iterable = tqdm(nwbfiles_iterable, **progress_bar_options)
     if n_jobs != 1:
         progress_bar_options.update(total=len(nwbfiles))
         futures = []
@@ -318,26 +380,34 @@ def inspect_all(
             for nwbfile_path in nwbfiles:
                 futures.append(
                     executor.submit(
-                        _pickle_inspect_nwb, nwbfile_path=nwbfile_path, checks=checks, skip_validate=skip_validate
+                        _pickle_inspect_nwb,
+                        nwbfile_path=nwbfile_path,
+                        checks=checks,
+                        skip_validate=skip_validate,
+                        driver=driver,
                     )
                 )
-            completed_futures = as_completed(futures)
+            nwbfiles_iterable = as_completed(futures)
             if progress_bar:
-                completed_futures = tqdm(completed_futures, **progress_bar_options)
-            for future in completed_futures:
+                nwbfiles_iterable = tqdm(nwbfiles_iterable, **progress_bar_options)
+            for future in nwbfiles_iterable:
                 for message in future.result():
+                    if stream:
+                        message.file_path = nwbfiles[message.file_path]
                     yield message
     else:
-        if progress_bar:
-            nwbfiles = tqdm(nwbfiles, **progress_bar_options)
-        for nwbfile_path in nwbfiles:
-            for message in inspect_nwb(nwbfile_path=nwbfile_path, checks=checks):
+        for nwbfile_path in nwbfiles_iterable:
+            for message in inspect_nwb(nwbfile_path=nwbfile_path, checks=checks, driver=driver):
+                if stream:
+                    message.file_path = nwbfiles[message.file_path]
                 yield message
 
 
-def _pickle_inspect_nwb(nwbfile_path: str, checks: list = available_checks, skip_validate: bool = False):
+def _pickle_inspect_nwb(
+    nwbfile_path: str, checks: list = available_checks, skip_validate: bool = False, driver: Optional[str] = None
+):
     """Auxilliary function for inspect_all to run in parallel using the ProcessPoolExecutor."""
-    return list(inspect_nwb(nwbfile_path=nwbfile_path, checks=checks, skip_validate=skip_validate))
+    return list(inspect_nwb(nwbfile_path=nwbfile_path, checks=checks, skip_validate=skip_validate, driver=driver))
 
 
 def inspect_nwb(

nwbinspector/register_checks.py

@@ -78,9 +78,7 @@ class InspectorMessage:
     file_path: str = None
 
     def __repr__(self):
-        return ["InspectorMessage(\n"] + "\n,".join(
-            +[f"    {k}={v.__repr__()}" for k, v in self.__dict__.items()] + [")"]
-        )
+        return "InspectorMessage(\n" + ",\n".join([f"    {k}={v.__repr__()}" for k, v in self.__dict__.items()]) + "\n)"
 
 
 # TODO: neurodata_type could have annotation hdmf.utils.ExtenderMeta, which seems to apply to all currently checked
@@ -175,7 +173,6 @@ def parse_location(neurodata_object) -> Optional[str]:
     for key, val in known_locations.items():
         if isinstance(neurodata_object, key):
             return val
-
     """Infer the human-readable path of the object within an NWBFile by tracing its parents."""
     if neurodata_object.parent is None:
         return "/"

nwbinspector/tools.py

@@ -1,8 +1,12 @@
 """Helper functions for internal use that rely on external dependencies (i.e., pynwb)."""
+import re
 from uuid import uuid4
 from datetime import datetime
+from typing import Optional, Dict
+from concurrent.futures import ProcessPoolExecutor, as_completed
 
 from pynwb import NWBFile
+from dandi.dandiapi import DandiAPIClient
 
 
 def make_minimal_nwbfile():
@@ -22,3 +26,43 @@ def get_nwbfile_path_from_internal_object(obj):
     if isinstance(obj, NWBFile):
         return obj.container_source
     return obj.get_ancestor("NWBFile").container_source
+
+
+def get_s3_urls_and_dandi_paths(dandiset_id: str, version_id: Optional[str] = None, n_jobs: int = 1) -> Dict[str, str]:
+    """
+    Collect S3 URLS from a DANDISet ID.
+
+    Returns dictionary that maps each S3 url to the displayed file path on the DANDI archive content page.
+    """
+    assert re.fullmatch(
+        pattern="^[0-9]{6}$", string=dandiset_id
+    ), "The specified 'path' is not a proper DANDISet ID. It should be a six-digit numeric identifier."
+
+    s3_urls_to_dandi_paths = dict()
+    if n_jobs != 1:
+        with DandiAPIClient() as client:
+            dandiset = client.get_dandiset(dandiset_id=dandiset_id, version_id=version_id)
+            max_workers = n_jobs if n_jobs > 0 else None
+            with ProcessPoolExecutor(max_workers=max_workers) as executor:
+                futures = []
+                for asset in dandiset.get_assets():
+                    if asset.path.split(".")[-1] == "nwb":
+                        futures.append(
+                            executor.submit(
+                                _get_content_url_and_path, asset=asset, follow_redirects=1, strip_query=True
+                            )
+                        )
+                    for future in as_completed(futures):
+                        s3_urls_to_dandi_paths.update(future.result())
+    else:
+        with DandiAPIClient() as client:
+            dandiset = client.get_dandiset(dandiset_id=dandiset_id, version_id=version_id)
+            for asset in dandiset.get_assets():
+                if asset.path.split(".")[-1] == "nwb":
+                    s3_urls_to_dandi_paths.update(_get_content_url_and_path(asset=asset))
+    return s3_urls_to_dandi_paths
+
+
+def _get_content_url_and_path(asset, follow_redirects: int = 1, strip_query: bool = True) -> Dict[str, str]:
+    """Private helper function for parallelization in 'get_s3_urls_and_dandi_paths'."""
+    return {asset.get_content_url(follow_redirects=1, strip_query=True): asset.path}

setup.py

@@ -14,7 +14,7 @@
     packages=find_packages(),
     include_package_data=True,
     url="https://github.com/NeurodataWithoutBorders/nwbinspector",
-    install_requires=["pynwb", "natsort", "click", "PyYAML", "jsonschema", "tqdm"],
+    install_requires=["pynwb", "natsort", "click", "PyYAML", "jsonschema", "dandi", "tqdm"],
     entry_points={"console_scripts": ["nwbinspector=nwbinspector.nwbinspector:inspect_all_cli"]},
     license="BSD-3-Clause",
 )