saving state

Author: Cody Baker

nwbinspector/nwbinspector.py

@@ -13,7 +13,7 @@
 from types import FunctionType
 from warnings import filterwarnings, warn
 from distutils.util import strtobool
-from time import sleep
+from collections import defaultdict
 
 import click
 import pynwb
@@ -154,10 +154,7 @@ def configure_checks(
 @click.argument("path")
 @click.option("--modules", help="Modules to import prior to reading the file(s).")
 @click.option(
-    "--report-file-path",
-    default=None,
-    help="Save path for the report file.",
-    type=click.Path(writable=True),
+    "--report-file-path", default=None, help="Save path for the report file.", type=click.Path(writable=True),
 )
 @click.option("--overwrite", help="Overwrite an existing report file at the location.", is_flag=True)
 @click.option("--levels", help="Comma-separated names of InspectorMessage attributes to organize by.")
@@ -370,6 +367,30 @@ def inspect_all(
     # 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)
 
+    # Manual identifier check over all files in the folder path
+    identifiers = defaultdict(list)
+    for nwbfile_path in nwbfiles:
+        with pynwb.NWBHDF5IO(path=nwbfile_path, mode="r", driver=driver) as io:
+            nwbfile = robust_s3_read(io.read)
+            identifiers[nwbfile.identifier].append(nwbfile_path)
+    if len(identifiers) != len(nwbfiles):
+        for identifier, nwbfiles_with_identifier in identifiers.items():
+            if len(nwbfiles_with_identifier) > 1:
+                yield InspectorMessage(
+                    message=(
+                        f"The identifier '{identifier}' is used across the .nwb files: "
+                        f"{[str(x) for x in nwbfiles_with_identifier]}\n"
+                        "The identifier of any NWBFile should be a completely unique value - "
+                        "we recommend using uuid4 to achieve this."
+                    ),
+                    importance=Importance.CRITICAL,
+                    check_function_name="check_unique_identifiers",
+                    object_type="NWBFile",
+                    object_name="root",
+                    location="/",
+                    file_path=str(Path(nwbfiles_with_identifier[0]).parent),
+                )
+
     nwbfiles_iterable = nwbfiles
     if progress_bar:
         nwbfiles_iterable = tqdm(nwbfiles_iterable, **progress_bar_options)

tests/test_inspector.py

@@ -4,6 +4,7 @@
 from tempfile import mkdtemp
 from pathlib import Path
 from unittest import TestCase
+from datetime import datetime
 
 import numpy as np
 from pynwb import NWBFile, NWBHDF5IO, TimeSeries
@@ -144,7 +145,7 @@ def assertLogFileContentsEqual(
                     if ".nwb" in test_line:
                         # Transform temporary testing path and formatted to hardcoded fake path
                         str_loc = test_line.find(".nwb")
-                        correction_str = test_line.replace(test_line[5 : str_loc - 8], "./")
+                        correction_str = test_line.replace(test_line[5 : str_loc - 8], "./")  # noqa E203 (black)
                         test_file_lines[line_number] = correction_str
                 self.assertEqual(first=test_file_lines[skip_first_n_lines:-1], second=true_file_lines)
 
@@ -561,3 +562,75 @@ def test_dandiset_streaming_cli_parallel(self):
             f"> {console_output_file}"
         )
         self.assertFileExists(path=self.tempdir / "test_nwbinspector_streaming_report_7.txt")
+
+
+class TestCheckUniqueIdentifiersPass(TestCase):
+    maxDiff = None
+
+    @classmethod
+    def setUpClass(cls):
+        cls.tempdir = Path(mkdtemp())
+        num_nwbfiles = 3
+        unique_id_nwbfiles = list()
+        for j in range(num_nwbfiles):
+            unique_id_nwbfiles.append(make_minimal_nwbfile())
+
+        cls.unique_id_nwbfile_paths = [str(cls.tempdir / f"unique_id_testing{j}.nwb") for j in range(num_nwbfiles)]
+        for nwbfile_path, nwbfile in zip(cls.unique_id_nwbfile_paths, unique_id_nwbfiles):
+            with NWBHDF5IO(path=nwbfile_path, mode="w") as io:
+                io.write(nwbfile)
+
+    @classmethod
+    def tearDownClass(cls):
+        rmtree(cls.tempdir)
+
+    def test_check_unique_identifiers_pass(self):
+        assert list(inspect_all(path=self.tempdir, select=["check_data_orientation"])) == []
+
+
+class TestCheckUniqueIdentifiersFail(TestCase):
+    maxDiff = None
+
+    @classmethod
+    def setUpClass(cls):
+        cls.tempdir = Path(mkdtemp())
+        num_nwbfiles = 3
+        non_unique_id_nwbfiles = list()
+        for j in range(num_nwbfiles):
+            non_unique_id_nwbfiles.append(
+                NWBFile(
+                    session_description="",
+                    identifier="not a unique identifier!",
+                    session_start_time=datetime.now().astimezone(),
+                )
+            )
+
+        cls.non_unique_id_nwbfile_paths = [
+            str(cls.tempdir / f"non_unique_id_testing{j}.nwb") for j in range(num_nwbfiles)
+        ]
+        for nwbfile_path, nwbfile in zip(cls.non_unique_id_nwbfile_paths, non_unique_id_nwbfiles):
+            with NWBHDF5IO(path=nwbfile_path, mode="w") as io:
+                io.write(nwbfile)
+
+    @classmethod
+    def tearDownClass(cls):
+        rmtree(cls.tempdir)
+
+    def test_check_unique_identifiers_fail(self):
+        print(list(inspect_all(path=self.tempdir, select=["check_data_orientation"])))
+        assert list(inspect_all(path=self.tempdir, select=["check_data_orientation"])) == [
+            InspectorMessage(
+                message=(
+                    "The identifier 'not a unique identifier!' is used across the .nwb files: "
+                    f"{[str(x) for x in self.non_unique_id_nwbfile_paths]}\n"
+                    "The identifier of any NWBFile should be a completely unique value - "
+                    "we recommend using uuid4 to achieve this."
+                ),
+                importance=Importance.CRITICAL,
+                check_function_name="check_unique_identifiers",
+                object_type="NWBFile",
+                object_name="root",
+                location="/",
+                file_path=str(Path(self.non_unique_id_nwbfile_paths[0]).parent),
+            )
+        ]