Merge pull request #11 from dandi/gh-7

Implement basic helper tool

.gitignore

@@ -0,0 +1 @@
+__pycache__/

nwb_healthstatus/__main__.py

@@ -0,0 +1,57 @@
+from pathlib import Path
+import click
+import pynwb
+from .base import get_cases_in_namespace
+
+DEFAULT_SAMPLES_PATH = str(Path.home() / ".cache" / "nwb-healthstatus")
+
+@click.group()
+def main():
+    pass
+
+@main.group()
+def sample():
+    pass
+
+@sample.command()
+#@click.option("-e", "--environment")
+@click.option("--overwrite", is_flag=True)
+@click.option("--samples-path", type=click.Path(file_okay=False), default=DEFAULT_SAMPLES_PATH)
+@click.argument("casefile", type=click.Path(exists=True, dir_okay=False), nargs=-1)
+def create(casefile, overwrite, samples_path):
+    sample_dir = Path(samples_path)
+    for path in casefile:
+        p = Path(path)
+        producer = p.resolve().parent.name
+        namespace = {}
+        exec(p.read_text(), namespace)
+        for casecls in get_cases_in_namespace(namespace):
+            case = casecls()
+            filepath = sample_dir / producer / case.FILENAME
+            filepath.parent.mkdir(parents=True, exist_ok=True)
+            if overwrite or not filepath.exists():
+                nwbfile = case.create()
+                with pynwb.NWBHDF5IO(str(filepath), "w") as io:
+                    io.write(nwbfile) # , cache_spec=cache_spec) 
+
+@sample.command()
+#@click.option("-e", "--environment")
+@click.option("--samples-path", type=click.Path(exists=True, file_okay=False), default=DEFAULT_SAMPLES_PATH)
+@click.argument("casefile", type=click.Path(exists=True, dir_okay=False), nargs=-1)
+def test(casefile, samples_path):
+    sample_dir = Path(samples_path)
+    for path in casefile:
+        p = Path(path)
+        producer = p.resolve().parent.name
+        namespace = {}
+        exec(p.read_text(), namespace)
+        for casecls in get_cases_in_namespace(namespace):
+            case = casecls()
+            filepath = sample_dir / producer / case.FILENAME
+            with pynwb.NWBHDF5IO(str(filepath), mode='r') as io:
+                ## TODO: Capture and display possible warnings
+                obj = io.read()
+            case.test(obj)
+
+if __name__ == "__main__":
+    main()

nwb_healthstatus/base.py

@@ -0,0 +1,51 @@
+from abc import ABC, abstractmethod
+from functools import reduce
+from inspect import isclass
+from operator import or_
+from types import ModuleType
+from typing import ClassVar, Iterator, Set, Type
+from hdmf.container import Container
+import pynwb
+
+class SampleCase(ABC):
+    #: Set of extensions needed by the sample case
+    EXTENSIONS: ClassVar[Set[str]]
+
+    #: Basename for the sample file created by the case
+    FILENAME: ClassVar[str]
+
+    @abstractmethod
+    def create(self) -> pynwb.NWBFile:
+        """ Creates a sample NWB file """
+        ...
+
+    @abstractmethod
+    def test(self, data: Container) -> None:
+        """
+        Takes the data read from a sample file and asserts that it contains
+        what it should
+        """
+        ...
+
+    @classmethod
+    def __subclasshook__(cls, C):
+        if (
+            cls is SampleCase
+            and {"EXTENSIONS", "FILENAME", "create", "test"}
+            <= reduce(or_, (B.__dict__.keys() for B in C.__mro__))
+        ):
+                return True
+        return NotImplemented
+
+
+def get_cases_in_module(module: ModuleType) -> Iterator[Type[SampleCase]]:
+    for name in dir(module):
+        obj = getattr(module, name)
+        if isclass(obj) and issubclass(obj, SampleCase):
+            yield obj
+
+
+def get_cases_in_namespace(namespace: dict) -> Iterator[Type[SampleCase]]:
+    for obj in namespace.values():
+        if isclass(obj) and issubclass(obj, SampleCase):
+            yield obj

nwb_healthstatus/producers/core/simple1.py

@@ -1,6 +1,5 @@
-import hdmf
-import pynwb
 import datetime
+import pynwb
 
 metadata = dict(
     session_description= 'my first synthetic recording', 
@@ -13,36 +12,14 @@
     session_id='LONELYMTN'
 )
 
-def create():
-    return pynwb.NWBFile(**metadata)
-
-
-def test_basic(nwbfile):
-    # TODO: make it more specific to this example
-    for f, v in metadata.items():
-        assert getattr(nwbfile, f) == v, f"{f}: {getattr(nwbfile, f)!r} vs. {v!r}"
-
-
-if __name__ == '__main__':
-   base_filename = 'core_simple1'
-   env_details = {
-      'nwb': pynwb.__version__,
-      'hdmf': hdmf.__version__,
-   }
-   suffix = '_'.join('{}:{}'.format(*i) for i in env_details.items())
-
-   filename = f'/tmp/{base_filename}_{suffix}'
-
-   ### this would be executed once for some combinations of hdmf/pynwb
-   ### version and stored indefinetely somewhere
-   nwbfile = create()
-   with pynwb.NWBHDF5IO(filename + '.nwb', "w") as io:
-        io.write(nwbfile) # , cache_spec=cache_spec) 
-   # todo dump into '.yaml' the details of the spec
+class Simple1:
+    EXTENSIONS = set()
+    FILENAME = "simple1.nwb"
 
-   ### CI run would load the file and give it away for testing
-   with pynwb.NWBHDF5IO(filename + '.nwb', mode='r') as io:
-        ## capture and display possible warnings
-        obj = io.read()
+    def create(self):
+        return pynwb.NWBFile(**metadata)
 
-        test_basic(obj)
+    def test(self, nwbfile):
+        # TODO: make it more specific to this example
+        for f, v in metadata.items():
+            assert getattr(nwbfile, f) == v, f"{f}: {getattr(nwbfile, f)!r} vs. {v!r}"

requirements.txt

@@ -0,0 +1,3 @@
+click
+hdmf
+pynwb