Merge branch 'dev' into negative_job_slicing

Author: CodyCBakerPhD

nwbinspector/nwbinspector.py

@@ -13,6 +13,7 @@
 from types import FunctionType
 from warnings import filterwarnings, warn
 from distutils.util import strtobool
+from time import sleep
 
 import click
 import pynwb
@@ -28,7 +29,7 @@
 )
 from .register_checks import InspectorMessage, Importance
 from .tools import get_s3_urls_and_dandi_paths
-from .utils import FilePathType, PathType, OptionalListOfStrings, calculate_number_of_cpu
+from .utils import FilePathType, PathType, OptionalListOfStrings, robust_s3_read, calculate_number_of_cpu
 
 INTERNAL_CONFIGS = dict(dandi=Path(__file__).parent / "internal_configs" / "dandi.inspector_config.yaml")
 
@@ -417,8 +418,9 @@ def inspect_nwb(
     ignore: OptionalListOfStrings = None,
     select: OptionalListOfStrings = None,
     importance_threshold: Union[str, Importance] = Importance.BEST_PRACTICE_SUGGESTION,
-    driver: str = None,
+    driver: Optional[str] = None,
     skip_validate: bool = False,
+    max_retries: int = 10,
 ) -> List[InspectorMessage]:
     """
     Inspect a NWBFile object and return suggestions for improvements according to best practices.
@@ -452,6 +454,11 @@ def inspect_nwb(
     skip_validate : bool
         Skip the PyNWB validation step. This may be desired for older NWBFiles (< schema version v2.10).
         The default is False, which is also recommended.
+    max_retries : int, optional
+        When using the ros3 driver to stream data from an s3 path, occasional curl issues can result.
+        AWS suggests using iterative retry with an exponential backoff of 0.1 * 2^retries.
+        This sets a hard bound on the number of times to attempt to retry the collection of messages.
+        Defaults to 10 (corresponds to 102.4s maximum delay on final attempt).
     """
     importance_threshold = (
         Importance[importance_threshold] if isinstance(importance_threshold, str) else importance_threshold
@@ -463,6 +470,7 @@ def inspect_nwb(
     nwbfile_path = str(nwbfile_path)
     filterwarnings(action="ignore", message="No cached namespaces found in .*")
     filterwarnings(action="ignore", message="Ignoring cached namespace .*")
+
     with pynwb.NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True, driver=driver) as io:
         if not skip_validate:
             validation_errors = pynwb.validate(io=io)
@@ -476,7 +484,7 @@ def inspect_nwb(
                 )
 
         try:
-            nwbfile = io.read()
+            nwbfile = robust_s3_read(command=io.read, max_retries=max_retries)
             for inspector_message in run_checks(nwbfile=nwbfile, checks=checks):
                 inspector_message.file_path = nwbfile_path
                 yield inspector_message
@@ -502,7 +510,7 @@ def run_checks(nwbfile: pynwb.NWBFile, checks: list):
         for nwbfile_object in nwbfile.objects.values():
             if check_function.neurodata_type is None or issubclass(type(nwbfile_object), check_function.neurodata_type):
                 try:
-                    output = check_function(nwbfile_object)
+                    output = robust_s3_read(command=check_function, command_args=[nwbfile_object])
                 # if an individual check fails, include it in the report and continue with the inspection
                 except Exception:
                     output = InspectorMessage(

nwbinspector/utils.py

@@ -3,10 +3,11 @@
 import re
 import json
 import numpy as np
-from typing import TypeVar, Optional, List
+from typing import TypeVar, Optional, List, Dict, Callable
 from pathlib import Path
 from importlib import import_module
 from packaging import version
+from time import sleep
 
 PathType = TypeVar("PathType", str, Path)  # For types that can be either files or folders
 FilePathType = TypeVar("FilePathType", str, Path)
@@ -116,6 +117,22 @@ def get_package_version(name: str) -> version.Version:
     return version.parse(package_version)
 
 
+def robust_s3_read(
+    command: Callable, max_retries: int = 10, command_args: Optional[list] = None, command_kwargs: Optional[Dict] = None
+):
+    """Attempt the command (usually acting on an S3 IO) up to the number of max_retries using exponential backoff."""
+    command_args = command_args or []
+    command_kwargs = command_kwargs or dict()
+    for retry in range(max_retries):
+        try:
+            return command(*command_args, **command_kwargs)
+        except OSError:  # cannot curl request
+            sleep(0.1 * 2**retry)
+        except Exception as exc:
+            raise exc
+    raise TimeoutError(f"Unable to complete the command ({command.__name__}) after {max_retries} attempts!")
+
+
 def calculate_number_of_cpu(requested_cpu: int = 1) -> int:
     """
     Calculate the number CPUs to use with respect to negative slicing and check against maximal available resources.

tests/unit_tests/test_time_series.py

@@ -1,4 +1,5 @@
 from packaging import version
+from time import sleep
 
 import numpy as np
 import pynwb
@@ -14,7 +15,7 @@
     check_missing_unit,
     check_resolution,
 )
-from nwbinspector.utils import get_package_version
+from nwbinspector.utils import get_package_version, robust_s3_read
 
 try:
     # Test ros3 on sub-YutaMouse54/sub-YutaMouse54_ses-YutaMouse54-160630_behavior+ecephys.nwb from #3
@@ -195,9 +196,12 @@ def test_check_none_matnwb_resolution_pass():
         load_namespaces=True,
         driver="ros3",
     ) as io:
-        nwbfile = io.read()
-        time_series = nwbfile.processing["video_files"]["video"].time_series["20170203_KIB_01_s1.1.h264"]
-        assert check_resolution(time_series) is None
+        nwbfile = robust_s3_read(command=io.read)
+        time_series = robust_s3_read(
+            "20170203_KIB_01_s1.1.h264",
+            command=nwbfile.processing["video_files"]["video"].time_series.get,
+        )
+    assert check_resolution(time_series) is None
 
 
 def test_check_resolution_fail():