diff --git a/spec/ndx-hed.extensions.yaml b/spec/ndx-hed.extensions.yaml index 9f62e48..a5c1954 100644 --- a/spec/ndx-hed.extensions.yaml +++ b/spec/ndx-hed.extensions.yaml @@ -1,18 +1,12 @@ datasets: -- neurodata_type_def: HedAnnotations +- neurodata_type_def: HedTags neurodata_type_inc: VectorData dtype: text doc: An extension of VectorData for Hierarchical Event Descriptor (HED) tags. If HED tags are used, the HED schema version must be specified in the NWB file using - the HedVersion type. - attributes: - - name: sub_name - dtype: text - doc: The smallest possible difference between two event times. Usually 1 divided - by the event time sampling rate on the data acquisition system. - required: false + the HedMetadata type. groups: -- neurodata_type_def: HedVersion +- neurodata_type_def: HedVersionAttr neurodata_type_inc: LabMetaData name: hed_version doc: An extension of LabMetaData to store the Hierarchical Event Descriptor (HED) @@ -21,7 +15,16 @@ groups: attributes: - name: version dtype: text - shape: - - null doc: The version of the HED schema used to validate the HED tags, e.g., '8.2.0'. Required if HED tags are used in the NWB file. +- neurodata_type_def: HedVersion + neurodata_type_inc: LabMetaData + name: hed_version + doc: An extension of LabMetaData to store the Hierarchical Event Descriptor (HED) + schema version. TODO When merged with core, this will no longer inherit from LabMetaData + but from NWBContainer and be placed optionally in /general. + datasets: + - name: version + dtype: text + doc: HED schema version to use for this dataset + quantity: '?' diff --git a/spec/ndx-hed.namespace.yaml b/spec/ndx-hed.namespace.yaml index 722403a..a4ad483 100644 --- a/spec/ndx-hed.namespace.yaml +++ b/spec/ndx-hed.namespace.yaml @@ -13,5 +13,8 @@ namespaces: name: ndx-hed schema: - namespace: core + neurodata_types: + - LabMetaData + - VectorData - source: ndx-hed.extensions.yaml version: 0.1.0 diff --git a/src/pynwb/README.md b/src/pynwb/README.md index e69de29..1b7b10e 100644 --- a/src/pynwb/README.md +++ b/src/pynwb/README.md @@ -0,0 +1 @@ +## ndx_hed NWB extension \ No newline at end of file diff --git a/src/pynwb/ndx_hed/__init__.py b/src/pynwb/ndx_hed/__init__.py index a5efd06..365e445 100644 --- a/src/pynwb/ndx_hed/__init__.py +++ b/src/pynwb/ndx_hed/__init__.py @@ -1,30 +1,58 @@ import os from pynwb import load_namespaces, get_class -try: - from importlib.resources import files -except ImportError: - # TODO: Remove when python 3.9 becomes the new minimum - from importlib_resources import files +# Set path of the namespace.yaml file to the expected install location +ndx_hed_specpath = os.path.join( + os.path.dirname(__file__), + 'spec', + 'ndx-events.namespace.yaml' +) -# Get path to the namespace.yaml file with the expected location when installed not in editable mode -__location_of_this_file = files(__name__) -__spec_path = __location_of_this_file / "spec" / "ndx-hed.namespace.yaml" - -# If that path does not exist, we are likely running in editable mode. Use the local path instead -if not os.path.exists(__spec_path): - __spec_path = __location_of_this_file.parent.parent.parent / "spec" / "ndx-hed.namespace.yaml" +# If the extension has not been installed yet but we are running directly from +# the git repo +if not os.path.exists(ndx_hed_specpath): + ndx_hed_specpath = os.path.abspath(os.path.join( + os.path.dirname(__file__), + '..', '..', '..', + 'spec', + 'ndx-hed.namespace.yaml' + )) # Load the namespace -load_namespaces(str(__spec_path)) - -# TODO: Define your classes here to make them accessible at the package level. -# Either have PyNWB generate a class from the spec using `get_class` as shown -# below or write a custom class and register it using the class decorator -# `@register_class("TetrodeSeries", "ndx-hed")` -# HedAnnotations = get_class("HedAnnotations", "ndx-hed") +load_namespaces(ndx_hed_specpath) +# +# from . import io as __io # noqa: E402,F401 +# from .hed_models import NWBHedVersion +# +# try: +# from importlib.resources import files +# except ImportError: +# # TODO: Remove when python 3.9 becomes the new minimum +# from importlib_resources import files +# +# +# +# # Get path to the namespace.yaml file with the expected location when installed not in editable mode +# __location_of_this_file = files(__name__) +# __spec_path = __location_of_this_file / "spec" / "ndx-hed.namespace.yaml" +# +# # If that path does not exist, we are likely running in editable mode. Use the local path instead +# if not os.path.exists(__spec_path): +# __spec_path = __location_of_this_file.parent.parent.parent / "spec" / "ndx-hed.namespace.yaml" +# +# # Load the namespace +# load_namespaces(str(__spec_path)) +# +# # TODO: Define your classes here to make them accessible at the package level. +# # Either have PyNWB generate a class from the spec using `get_class` as shown +# # below or write a custom class and register it using the class decorator +# # `@register_class("TetrodeSeries", "ndx-hed")` +# # HedAnnotations = get_class("HedAnnotations", "ndx-hed") # HedVersion = get_class("HedVersion", "ndx-hed") - -# Remove these functions from the package +# +# # Remove these functions from the package del load_namespaces, get_class -from .hed_annotations import HedVersion, HedAnnotations + +from .hed_version_attr import HedVersionAttr +from .hed_tags import HedTags +from .hed_version import HedVersion diff --git a/src/pynwb/ndx_hed/hed_annotations.py b/src/pynwb/ndx_hed/hed_annotations.py deleted file mode 100644 index 3d79778..0000000 --- a/src/pynwb/ndx_hed/hed_annotations.py +++ /dev/null @@ -1,104 +0,0 @@ -from collections.abc import Iterable -from hdmf.common import VectorData -from hdmf.utils import docval, getargs, get_docval, popargs -from hed.schema import HedSchema, HedSchemaGroup, load_schema_version -from pynwb import register_class -from pynwb.file import LabMetaData - - -@register_class('HedAnnotations', 'ndx-hed') -class HedAnnotations(VectorData): - """ - Column storing HED (Hierarchical Event Descriptors) annotations for a row. A HED string is a comma-separated, - and possibly parenthesized list of HED tags selected from a valid HED vocabulary as specified by the - NWBFile field HEDVersion. - - """ - - __nwbfields__ = ('sub_name') - - @docval(*get_docval(VectorData.__init__)) - def __init__(self, **kwargs): - # kwargs['name'] = 'HED' - super().__init__(**kwargs) - self._init_internal() - - def _init_internal(self): - """ - This finds the HED schema object of use in this NWBFile. - - TODO: How should errors be handled if this file doesn't have a HedVersion object in the LabMetaData? - - """ - self.sub_name = "HED" - root = self - # parent = root.parent - # while parent is not None: - # root = parent - # parent = root.parent - # hed_version = parent.get_lab_meta_data("HedVersion") - # if hed_version: - # self.hed_schema = hed_version.get_schema() - - @docval({'name': 'val', 'type': str, - 'doc': 'the value to add to this column. Should be a valid HED string.'}) - def add_row(self, **kwargs): - """Append a data value to this column.""" - val = getargs('val', kwargs) - # val.check_types() - # TODO how to validate - # - # if val is not None and self.validate(val): - # if self.term_set.validate(term=val): - # self.append(val) - # else: - # msg = ("%s is not in the term set." % val) - # raise ValueError(msg) - # - # else: - # self.append(val) - super().append(val) - - @docval({'name': 'key', 'type': 'str', 'doc': 'the value to add to this column'}) - def get(self, key): - """ - Retrieve elements from this object. - - """ - # TODO: Can key be more than a single value? Do we need to check validity of anything? - vals = super().get(key) - return vals - - @docval({'name': 'val', 'type': 'str', 'doc': 'the value to validate'}, - {'name': 'return', 'type': 'list', 'doc': 'list of issues or none'}) - def validate(self, **kwargs): - """Validate this HED string""" - val = getargs('val', kwargs) - return True - - -@register_class("HedVersion", "ndx-hed") -class HedVersion(LabMetaData): - """ - The class containing the HED versions and HED schema used in this data file. - - """ - - __nwbfields__ = ('name', 'description', 'version') - - @docval({'name': 'version', 'type': (str, list), 'doc': 'HED strings of type str'}) - def __init__(self, version): - kwargs = {'name': 'hed_version'} - super().__init__(**kwargs) - self.version = version - self._init_internal() - - def _init_internal(self): - """ - Create a HedSchema or HedSchemaGroup object from the HED Versions - """ - self._hed_schema = load_schema_version(self.version) - - @docval(returns='The HED schema or schema group object for this version', rtype=(HedSchema, HedSchemaGroup)) - def get_hed_schema(self): - return self._hed_schema diff --git a/src/pynwb/ndx_hed/hed_tags.py b/src/pynwb/ndx_hed/hed_tags.py new file mode 100644 index 0000000..5d0c4e6 --- /dev/null +++ b/src/pynwb/ndx_hed/hed_tags.py @@ -0,0 +1,98 @@ +from collections.abc import Iterable +from hdmf.common import VectorData +from hdmf.utils import docval, getargs, get_docval, popargs +from hed.errors import HedFileError, get_printable_issue_string +from hed.schema import HedSchema, HedSchemaGroup, load_schema_version, from_string +from hed.models import HedString +from pynwb import register_class +from pynwb.file import LabMetaData, NWBFile +from ndx_hed import HedVersionAttr + + +@register_class('HedTags', 'ndx-hed') +class HedTags(VectorData): + """ + Column storing HED (Hierarchical Event Descriptors) annotations for a row. A HED string is a comma-separated, + and possibly parenthesized list of HED tags selected from a valid HED vocabulary as specified by the + NWBFile field HedVersion. + + """ + + __nwbfields__ = ('_hed_schema') + + @docval({'name': 'name', 'type': 'str', 'doc': 'Must be HED', 'default': 'HED'}, + {'name': 'description', 'type': 'str', 'doc': 'Description of the HED annotations', + 'default': 'Hierarchical Event Descriptors (HED) annotations'}, + *get_docval(VectorData.__init__, 'data')) + def __init__(self, **kwargs): + kwargs['name'] = 'HED' + super().__init__(**kwargs) + self._init_internal() + + def _init_internal(self): + """ + This finds the HED schema object of use in this NWBFile. + + TODO: How should errors be handled if this file doesn't have a HedVersion object in the LabMetaData? + + """ + self._hed_schema = None + + @docval({'name': 'val', 'type': str, + 'doc': 'the value to add to this column. Should be a valid HED string -- just forces string.'}) + def add_row(self, **kwargs): + """Append a data value to this column.""" + val = getargs('val', kwargs) + super().append(val) + + # @docval({'name': 'schema', 'type': (HedSchema, None), 'doc': 'HedSchema to use to validate.', 'default': None}, + # {'name': 'return', 'type': 'list', 'doc': 'list of issues or none'}) + def validate(self, schema): + """Validate this VectorData. This is assuming a list --- where is the general iterator.""" + if not schema: + raise HedFileError('HedSchemaMissing', "Must provide a valid HedSchema", "") + issues = [] + for index in range(len(self.data)): + hed_obj = HedString(self.get(index), schema) + these_issues = hed_obj.validate() + if these_issues: + issues.append(f"line {str(index)}: {get_printable_issue_string(these_issues)}") + return "\n".join(issues) + + def get_hed_schema(self): + if not self._hed_schema: + root = self._get_root() + if isinstance(root, NWBFile): + self._hed_schema = root.get_lab_meta_data("hed_version").get_schema() + return self._hed_schema + + def _get_root(self): + root = self + while hasattr(root, 'parent') and root.parent: + root = root.parent + if root == self: + return None + return root + + + # root = parent + # parent = root.parent + # if parent: + # hed_version = parent.get_lab_meta_data("HedVersion") + # else: + # hed_version = None + # if hed_version: + # self.hed_schema = hed_version.get_schema() + + # root = self + # parent = root.parent + # while parent is not None: + # root = parent + # parent = root.parent + # if parent: + # hed_version = parent.get_lab_meta_data("HedVersion") + # else: + # hed_version = None + # if hed_version: + # self.hed_schema = hed_version.get_schema() + diff --git a/src/pynwb/ndx_hed/hed_version.py b/src/pynwb/ndx_hed/hed_version.py new file mode 100644 index 0000000..006c822 --- /dev/null +++ b/src/pynwb/ndx_hed/hed_version.py @@ -0,0 +1,34 @@ +from hdmf.utils import docval, popargs +from pynwb import register_class +from pynwb.file import LabMetaData +from hed.schema import HedSchema, HedSchemaGroup, load_schema_version, from_string + + +@register_class("HedVersion", "ndx-hed") +class HedVersion(LabMetaData): + """ The class containing the HED versions and HED schema used in this data file. """ + + __nwbfields__ = ('name', 'version', 'schema_string') + + @docval({'name': 'version', 'type': str, 'doc': 'HED strings of type str'}) + def __init__(self, **kwargs): + version = popargs('version', kwargs) + kwargs['name'] = 'hed_version' + super().__init__(**kwargs) + self.version = version + self._init_internal() + + def _init_internal(self): + """ Create a HED schema string """ + hed_schema = load_schema_version(self.version) + self.schema_string = hed_schema.get_as_xml_string() + + @docval(returns='The HED schema version', rtype=str) + def get_version(self): + """ Return the schema version. """ + return self.version + + @docval(returns='The HED schema or schema group object for this version', rtype=(HedSchema, HedSchemaGroup)) + def get_schema(self): + """ Return the HEDSchema object for this version.""" + return from_string(self.schema_string) diff --git a/src/pynwb/ndx_hed/hed_version_attr.py b/src/pynwb/ndx_hed/hed_version_attr.py new file mode 100644 index 0000000..bd9bb77 --- /dev/null +++ b/src/pynwb/ndx_hed/hed_version_attr.py @@ -0,0 +1,37 @@ +from hdmf.utils import docval, getargs, get_docval, popargs +from pynwb import register_class +from pynwb.file import LabMetaData +from hed.schema import HedSchema, HedSchemaGroup, load_schema_version, from_string + + +@register_class("HedVersionAttr", "ndx-hed") +class HedVersionAttr(LabMetaData): + """ + The class containing the HED versions and HED schema used in this data file. + + """ + + __nwbfields__ = ('name', 'version', 'schema_string') + + @docval({'name': 'version', 'type': str, 'doc': 'HED strings of type str'}) + def __init__(self, **kwargs): + version = popargs('version', kwargs) + kwargs['name'] = 'hed_version' + super().__init__(**kwargs) + self.version = version + self._init_internal() + + def _init_internal(self): + """ + Create a HED schema string + """ + hed_schema = load_schema_version(self.version) + self.schema_string = hed_schema.get_as_xml_string() + + @docval(returns='The HED schema version', rtype=str) + def get_version(self): + return self.version + + @docval(returns='The HED schema or schema group object for this version', rtype=(HedSchema, HedSchemaGroup)) + def get_schema(self): + return from_string(self.schema_string) diff --git a/src/pynwb/tests/test_hed_annotations.py b/src/pynwb/tests/test_hed_annotations.py deleted file mode 100644 index 723b5dd..0000000 --- a/src/pynwb/tests/test_hed_annotations.py +++ /dev/null @@ -1,208 +0,0 @@ -"""Unit and integration tests for ndx-hed.""" -from datetime import datetime -from dateutil.tz import tzlocal, tzutc -from hed.schema import HedSchema, HedSchemaGroup -from pynwb import NWBFile, NWBHDF5IO, get_manager # , NWBFile -from pynwb.testing.mock.file import mock_NWBFile -from pynwb.testing import TestCase, remove_test_file # , NWBH5IOFlexMixin -from src.pynwb.ndx_hed import HedAnnotations, HedVersion - - -class TestHedVersion(TestCase): - """Simple unit test for creating a HedMetadata.""" - - def test_constructor(self): - """Test setting HedNWBFile values using the constructor.""" - hed_version1 = HedVersion(["8.2.0"]) - print(f"HED version: {str(hed_version1.version)}") - print(f"{str(hed_version1)}") - self.assertIsInstance(hed_version1.version, list) - schema1 = hed_version1.get_hed_schema() - self.assertIsInstance(schema1, HedSchema) - hed_version2 = HedVersion(["8.2.0", "sc:score_1.0.0"]) - schema2 = hed_version2.get_hed_schema() - self.assertIsInstance(schema2, HedSchemaGroup) - - def test_add_to_nwbfile(self): - nwbfile = mock_NWBFile() - hed_version1 = HedVersion(["8.2.0"]) - nwbfile.add_lab_meta_data(hed_version1) - hed_version2 = nwbfile.get_lab_meta_data("hed_version") - self.assertIsInstance(hed_version2, HedVersion) - - -class TestHedNWBFileSimpleRoundtrip(TestCase): - """Simple roundtrip test for HedNWBFile.""" - - def setUp(self): - self.path = "test.nwb" - - def tearDown(self): - remove_test_file(self.path) - - def test_roundtrip(self): - """ - Create a HedMetadata, write it to file, read the file, and test that it matches the original HedNWBFile. - """ - nwbfile = mock_NWBFile() - hed_version = HedVersion(["8.2.0"]) - nwbfile.add_lab_meta_data(hed_version) - print(f"nwb: {str(nwbfile)}") - print(f"{str(hed_version)}") - - with NWBHDF5IO(self.path, mode="w") as io: - io.write(nwbfile) - - # with NWBHDF5IO(self.path, mode="r", load_namespaces=True) as io: - # read_nwbfile = io.read() - # print(f"nwb: {str(read_nwbfile)}") - # read_hed_version = read_nwbfile.get_lab_meta_data("hed_version") - # self.assertIsInstance(read_hed_version, HedVersion) - # self.assertEqual(read_hed_version.hed_version, "8.2.0") - - -# # class TestHedNWBFileRoundtripPyNWB(NWBH5IOFlexMixin, TestCase): -# # """Complex, more complete roundtrip test for HedNWBFile using pynwb.testing infrastructure.""" -# -# # def getContainerType(self): -# # return "HedNWBFile" -# -# # def addContainer(self): -# # self.nwbfile = HedNWBFile( -# # session_description="session_description", -# # identifier=str(uuid4()), -# # session_start_time=datetime(1970, 1, 1, tzinfo=tzlocal()), -# # hed_schema_version="8.2.0", -# # ) -# -# # def getContainer(self, nwbfile: NWBFile): -# # return nwbfile -# - -class TestHedAnnotationsConstructor(TestCase): - """Simple unit test for creating a HedTags.""" - - def test_constructor(self): - """Test setting HED values using the constructor.""" - hed_annotations = HedAnnotations( - name='HED', - description="description", - data=["animal_target, correct_response", "animal_target, incorrect_response"], - ) - print(f"{hed_annotations}") - self.assertEqual(hed_annotations.sub_name, "HED") - self.assertEqual(hed_annotations.description, "description") - self.assertEqual(hed_annotations.data,["animal_target, correct_response", "animal_target, incorrect_response"]) - - def test_add_to_trials_table(self): - """Test adding HED column and data to a trials table.""" - nwbfile = mock_NWBFile() - hed_version = HedVersion(["8.2.0"]) - nwbfile.add_lab_meta_data(hed_version) - #nwbfile.add_trial_column("HED", "HED annotations for each trial") - nwbfile.add_trial_column(name="HED", description="HED annotations for each trial", - col_cls=HedAnnotations, data=[]) - nwbfile.add_trial(start_time=0.0, stop_time=1.0, HED="animal_target, correct_response") - nwbfile.add_trial(start_time=2.0, stop_time=3.0, HED="animal_target, incorrect_response") - - self.assertIsInstance(nwbfile.trials["HED"], HedAnnotations) - print(f"{str(nwbfile)}") - print(f"{str(nwbfile.trials)}") - hed_col = nwbfile.trials["HED"] - print(f"hed_col: {str(hed_col.data)}") - self.assertEqual(nwbfile.trials["HED"].data[0], "animal_target, correct_response") - self.assertEqual(nwbfile.trials["HED"].data[1], "animal_target, incorrect_response") - -class TestHedTagsSimpleRoundtrip(TestCase): - """Simple roundtrip test for HedTags.""" - - def setUp(self): - self.path = "test.nwb" - self.start_time = datetime(1970, 1, 1, 12, tzinfo=tzutc()) - self.ref_time = datetime(1979, 1, 1, 0, tzinfo=tzutc()) - self.create_date = datetime(2017, 4, 15, 12, tzinfo=tzlocal()) - self.manager = get_manager() - self.filename = 'test_nwbfileio.h5' - self.nwbfile = NWBFile(session_description='a test NWB File', - identifier='TEST123', - session_start_time=self.start_time, - timestamps_reference_time=self.ref_time, - file_create_date=self.create_date, - experimenter='test experimenter', - stimulus_notes='test stimulus notes', - data_collection='test data collection notes', - experiment_description='test experiment description', - institution='nomad', - lab='nolab', - notes='nonotes', - pharmacology='nopharmacology', - protocol='noprotocol', - related_publications='nopubs', - session_id='007', - slices='noslices', - source_script='nosources', - surgery='nosurgery', - virus='novirus', - source_script_file_name='nofilename') - - def tearDown(self): - remove_test_file(self.path) - - def test_hed_version(self): - hed1 = HedVersion(["8.2.0"]) - self.assertIsInstance(hed1, HedVersion) - schema = hed1.get_hed_schema() - self.assertIsInstance(schema, HedSchema) - -# def test_roundtrip(self): -# """ -# Add a HedTags to an NWBFile, write it to file, read the file, and test that the HedTags from the -# file matches the original HedTags. -# """ -# hed_version = HedVersion(["8.2.0"]) -# self.nwbfile.add_lab_meta_data(hed_version) -# -# self.nwbfile.add_trial_column("HED", "HED annotations for each trial") -# self.nwbfile.add_trial(start_time=0.0, stop_time=1.0, HED="animal_target, correct_response") -# self.nwbfile.add_trial(start_time=2.0, stop_time=3.0, HED="animal_target, incorrect_response") -# print("to here") -# # nwbfile = mock_NWBFile() -# # hed_version = HedVersion(hed_schema_version="8.2.0") -# # nwbfile.add_lab_meta_data(hed_version) -# # -# # nwbfile.add_trial_column("HED", "HED annotations for each trial", col_cls=HedAnnotations) -# # nwbfile.add_trial(start_time=0.0, stop_time=1.0, HED="animal_target, correct_response") -# # nwbfile.add_trial(start_time=2.0, stop_time=3.0, HED="animal_target, incorrect_response") -# -# # with NWBHDF5IO(self.path, mode="w") as io: -# # io.write(nwbfile) -# # -# # with NWBHDF5IO(self.path, mode="r", load_namespaces=True) as io: -# # read_nwbfile = io.read() -# # read_hed_annotations = read_nwbfile.trials["HED"] -# # assert isinstance(read_hed_annotations, HedAnnotations) -# # # read_nwbfile.trials["hed_tags"][0] is read as a numpy array -# # assert read_hed_annotations[0] == "animal_target, correct_response" -# # assert read_hed_annotations[1] == "animal_target, incorrect_response" -# -# -# # class TestHedTagsRoundtripPyNWB(NWBH5IOFlexMixin, TestCase): -# # """Complex, more complete roundtrip test for HedTags using pynwb.testing infrastructure.""" -# -# # def getContainerType(self): -# # return "HedTags" -# -# # def addContainer(self): -# # self.nwbfile = HedNWBFile( -# # session_description="session_description", -# # identifier=str(uuid4()), -# # session_start_time=datetime(1970, 1, 1, tzinfo=tzlocal()), -# # hed_schema_version="8.2.0", -# # ) -# -# # self.nwbfile.add_trial_column("hed_tags", "HED tags for each trial", col_cls=HedTags, index=True) -# # self.nwbfile.add_trial(start_time=0.0, stop_time=1.0, hed_tags=["animal_target", "correct_response"]) -# # self.nwbfile.add_trial(start_time=2.0, stop_time=3.0, hed_tags=["animal_target", "incorrect_response"]) -# -# # def getContainer(self, nwbfile: NWBFile): -# # return nwbfile.trials["hed_tags"].target diff --git a/src/pynwb/tests/test_hed_tags.py b/src/pynwb/tests/test_hed_tags.py new file mode 100644 index 0000000..1c38c30 --- /dev/null +++ b/src/pynwb/tests/test_hed_tags.py @@ -0,0 +1,229 @@ +"""Unit and integration tests for ndx-hed.""" +from pandas import DataFrame +from datetime import datetime, timezone +from dateutil.tz import tzlocal, tzutc +from uuid import uuid4, UUID +from hed.schema import HedSchema, HedSchemaGroup, load_schema_version +from hdmf.common import VectorData +from hdmf.utils import docval, getargs, get_docval, popargs +from pynwb import NWBHDF5IO, get_manager, NWBFile +from pynwb.testing.mock.file import mock_NWBFile +from pynwb.testing import TestCase, remove_test_file, NWBH5IOFlexMixin +from src.pynwb.ndx_hed import HedVersion, HedTags + + +class TestHedTagsConstructor(TestCase): + """Simple unit test for creating a HedTags.""" + + def setUp(self): + self.tags = HedTags(data=["Correct-action", "Incorrect-action"]) + + def tearDown(self): + pass + + def test_constructor(self): + """Test setting HED values using the constructor.""" + self.assertEqual(self.tags.name, "HED") + self.assertTrue(self.tags.description) + self.assertEqual(self.tags.data, ["Correct-action", "Incorrect-action"]) + + def test_add_row(self): + """Testing adding a row to the HedTags. """ + self.assertEqual(len(self.tags.data), 2) + self.tags.add_row("Correct-action") + self.assertEqual(len(self.tags.data), 3) + + def test_get(self): + """Testing getting slices. """ + self.assertEqual(self.tags.get(0), "Correct-action") + self.assertEqual(self.tags.get([0, 1]), ['Correct-action', 'Incorrect-action']) + + def test_add_to_trials_table(self): + """Test adding HED column and data to a trials table.""" + nwbfile = mock_NWBFile() + nwbfile.add_trial_column(name="HED", description="temp", col_cls=HedTags, data=[]) + nwbfile.add_trial(start_time=0.0, stop_time=1.0, HED="Correct-action") + nwbfile.add_trial(start_time=2.0, stop_time=3.0, HED="Incorrect-action") + self.assertIsInstance(nwbfile.trials["HED"], HedTags) + hed_col = nwbfile.trials["HED"] + self.assertEqual(hed_col.name, "HED") + self.assertEqual(hed_col.description, "temp") + self.assertEqual(nwbfile.trials["HED"].data[0], "Correct-action") + self.assertEqual(nwbfile.trials["HED"].data[1], "Incorrect-action") + + def test_add_to_trials_table_force_HED(self): + """Trial table does not allow the forcing of the column name to be HED.""" + nwbfile = mock_NWBFile() + nwbfile.add_trial_column(name="Blech", description="temp", col_cls=HedTags, data=[]) + nwbfile.add_trial(start_time=0.0, stop_time=1.0, Blech="Correct-action") + nwbfile.add_trial(start_time=2.0, stop_time=3.0, Blech="Incorrect-action") + self.assertIsInstance(nwbfile.trials["Blech"], HedTags) + hed_col = nwbfile.trials["Blech"] + self.assertEqual(hed_col.name, "HED") + self.assertEqual(nwbfile.trials["Blech"].data[0], "Correct-action") + self.assertEqual(nwbfile.trials["Blech"].data[1], "Incorrect-action") + + def test_validate_hed_tags(self): + """Test the validate_hed_tags""" + schema = load_schema_version("8.2.0") + issues = self.tags.validate(schema) + self.assertFalse(issues) + self.tags.add_row("Blech") + self.tags.add_row("Sensory-event") + self.tags.add_row("") + self.tags.add_row("Agent-action") + self.tags.add_row("Red, (Blue, Green") + issues = self.tags.validate(schema) + self.assertEqual(7, len(self.tags.data)) + self.assertTrue(issues) + + def test_get_root(self): + root = self.tags._get_root() + self.assertFalse(root) + + def test_get_hed_version(self): + schema = self.tags.get_hed_schema() + self.assertFalse(schema) + + +class TestHedTagsSimpleRoundtrip(TestCase): + """Simple roundtrip test for HedNWBFile.""" + + def setUp(self): + self.path = "test.nwb" + nwb_mock = mock_NWBFile() + nwb_mock.add_lab_meta_data(HedVersion("8.2.0")) + nwb_mock.add_trial_column(name="HED", description="HED annotations for each trial", + col_cls=HedTags, data=[]) + nwb_mock.add_trial(start_time=0.0, stop_time=1.0, HED="Correct-action") + nwb_mock.add_trial(start_time=2.0, stop_time=3.0, HED="Incorrect-action") + self.nwb_mock = nwb_mock + + def tearDown(self): + remove_test_file(self.path) + + def test_get_root(self): + tags = self.nwb_mock.trials["HED"] + self.assertIsInstance(tags, HedTags) + root = tags._get_root() + self.assertIsInstance(root, NWBFile) + + def test_get_hed_schema(self): + tags = self.nwb_mock.trials["HED"] + schema = tags.get_hed_schema() + self.assertIsInstance(schema, HedSchema) + + def test_validate_hed_tags(self): + """Test the validate_hed_tags""" + schema = load_schema_version("8.2.0") + tags = self.nwb_mock.trials["HED"] + issues = tags.validate(schema) + self.assertFalse(issues) + tags.add_row("Blech") + tags.add_row("Sensory-event") + tags.add_row("") + tags.add_row("Agent-action") + tags.add_row("Red, (Blue, Green") + self.assertEqual(7, len(self.nwb_mock.trials["HED"])) + issues = tags.validate(schema) + self.assertEqual(7, len(tags.data)) + self.assertTrue(issues) + + def test_roundtrip(self): + """ Create a HedMetadata, write it to mock file, read file, and test that it matches the original HedNWBFile.""" + + with NWBHDF5IO(self.path, mode="w") as io: + io.write(self.nwb_mock) + + with NWBHDF5IO(self.path, mode="r", load_namespaces=True) as io: + read_nwbfile = io.read() + read_hed_version = read_nwbfile.get_lab_meta_data("hed_version") + self.assertIsInstance(read_hed_version, HedVersion) + self.assertEqual(read_hed_version.version, "8.2.0") + self.assertEqual(read_nwbfile.trials["HED"].data[0], "Correct-action") + self.assertEqual(read_nwbfile.trials["HED"].data[1], "Incorrect-action") + + +class TestHedTagsNWBFileRoundtrip(TestCase): + """Simple roundtrip test for HedTags.""" + + def setUp(self): + self.path = "test.nwb" + self.start_time = datetime(1970, 1, 1, 12, tzinfo=tzutc()) + self.ref_time = datetime(1979, 1, 1, 0, tzinfo=tzutc()) + self.create_date = datetime(2017, 4, 15, 12, tzinfo=tzlocal()) + self.manager = get_manager() + self.filename = 'test_nwbfileio.h5' + self.nwbfile = NWBFile(session_description='a test NWB File', + identifier='TEST123', + session_start_time=self.start_time, + timestamps_reference_time=self.ref_time, + file_create_date=self.create_date, + experimenter='test experimenter', + stimulus_notes='test stimulus notes', + data_collection='test data collection notes', + experiment_description='test experiment description', + institution='nomad', + lab='nolab', + notes='nonotes', + pharmacology='nopharmacology', + protocol='noprotocol', + related_publications='nopubs', + session_id='007', + slices='noslices', + source_script='nosources', + surgery='nosurgery', + virus='novirus', + source_script_file_name='nofilename') + hed_version = HedVersion("8.2.0") + self.nwbfile.add_lab_meta_data(hed_version) + self.nwbfile.add_trial_column(name="HED", description="HED annotations for each trial", + col_cls=HedTags, data=[]) + self.nwbfile.add_trial(start_time=0.0, stop_time=1.0, HED="Correct-action") + self.nwbfile.add_trial(start_time=2.0, stop_time=3.0, HED="Incorrect-action") + + def tearDown(self): + remove_test_file(self.path) + + def test_get_root(self): + tags = self.nwbfile.trials["HED"] + self.assertIsInstance(tags, HedTags) + root = tags._get_root() + self.assertIsInstance(root, NWBFile) + + def test_get_hed_schema(self): + tags = self.nwbfile.trials["HED"] + schema = tags.get_hed_schema() + self.assertIsInstance(schema, HedSchema) + + def test_validate_hed_tags(self): + """Test the validate_hed_tags""" + schema = load_schema_version("8.2.0") + tags = self.nwbfile.trials["HED"] + issues = tags.validate(schema) + self.assertFalse(issues) + tags.add_row("Blech") + tags.add_row("Sensory-event") + tags.add_row("") + tags.add_row("Agent-action") + tags.add_row("Red, (Blue, Green") + self.assertEqual(7, len(self.nwbfile.trials["HED"])) + issues = tags.validate(schema) + self.assertEqual(7, len(tags.data)) + self.assertTrue(issues) + def test_roundtrip(self): + """ + Add a HedTags to an NWBFile, write it to file, read the file, and test that the HedTags from the + file matches the original HedTags. + """ + + with NWBHDF5IO(self.path, mode="w") as io: + io.write(self.nwbfile) + + with NWBHDF5IO(self.path, mode="r", load_namespaces=True) as io: + read_nwbfile = io.read() + read_hed_version = read_nwbfile.get_lab_meta_data("hed_version") + self.assertIsInstance(read_hed_version, HedVersion) + self.assertEqual(read_hed_version.version, "8.2.0") + self.assertEqual(read_nwbfile.trials["HED"].data[0], "Correct-action") + self.assertEqual(read_nwbfile.trials["HED"].data[1], "Incorrect-action") diff --git a/src/pynwb/tests/test_hed_version.py b/src/pynwb/tests/test_hed_version.py new file mode 100644 index 0000000..2f750ef --- /dev/null +++ b/src/pynwb/tests/test_hed_version.py @@ -0,0 +1,137 @@ +"""Unit and integration tests for ndx-hed.""" +from datetime import datetime +from dateutil.tz import tzlocal, tzutc +from pynwb.file import LabMetaData +from pynwb import NWBHDF5IO, get_manager, NWBFile +from pynwb.testing.mock.file import mock_NWBFile +from pynwb.testing import TestCase, remove_test_file, NWBH5IOFlexMixin +from src.pynwb.ndx_hed import HedVersion +from hed.schema import HedSchema, HedSchemaGroup, load_schema_version +from hed.errors import HedFileError + + +class TestHedVersion(TestCase): + """Simple unit test for creating a HedMetadata.""" + + def test_constructor(self): + """Test setting HedNWBFile values using the constructor.""" + hed_version1 = HedVersion("8.2.0") + self.assertIsInstance(hed_version1, HedVersion) + self.assertIsInstance(hed_version1, LabMetaData) + self.assertEqual(hed_version1.name, "hed_version") + self.assertIsInstance(hed_version1.version, str) + self.assertEqual(hed_version1.version, "8.2.0") + self.assertIsInstance(hed_version1.schema_string, str) + # hed_version2 = HedVersion(["8.2.0"]) + # self.assertIsInstance(hed_version2.version, list) + # hed_version3 = HedVersion(["8.2.0", "sc:score_1.0.0"]) + + def test_constructor_bad_version(self): + with self.assertRaises(HedFileError): + HedVersion("bad-version") + + def test_get_schema(self): + hed_version1 = HedVersion("8.2.0") + hed_schema = hed_version1.get_schema() + self.assertIsInstance(hed_schema, HedSchema) + + def test_get_version(self): + hed_version1 = HedVersion("8.2.0") + version = hed_version1.get_version() + self.assertIsInstance(version, str) + self.assertEqual(version, "8.2.0") + + def test_add_to_nwbfile(self): + nwbfile = mock_NWBFile() + hed_version1 = HedVersion("8.2.0") + nwbfile.add_lab_meta_data(hed_version1) + hed_version2 = nwbfile.get_lab_meta_data("hed_version") + self.assertIsInstance(hed_version2, HedVersion) + + +class TestHedVersion1SimpleRoundtrip(TestCase): + """Simple roundtrip test for HedNWBFile.""" + + def setUp(self): + self.path = "test.nwb" + + def tearDown(self): + remove_test_file(self.path) + + def test_roundtrip(self): + """Create a HedMetadata, write it to file, read the file, and test that it matches the original HedNWBFile.""" + nwbfile = mock_NWBFile() + hed_version = HedVersion(version="8.2.0") + self.assertIsInstance(hed_version, HedVersion) + self.assertIsInstance(hed_version, LabMetaData) + nwbfile.add_lab_meta_data(lab_meta_data=hed_version) + meta = nwbfile.get_lab_meta_data("hed_version") + self.assertEqual(meta.name, "hed_version") + self.assertEqual(meta.version, "8.2.0") + self.assertIsInstance(meta, HedVersion) + self.assertIsInstance(meta, LabMetaData) + schema = meta.get_schema() + self.assertIsInstance(schema, HedSchema) + + with NWBHDF5IO(self.path, mode="w") as io: + io.write(nwbfile) + + with NWBHDF5IO(self.path, mode="r", load_namespaces=True) as io: + read_nwbfile = io.read() + read_hed_version = read_nwbfile.get_lab_meta_data(hed_version.name) + self.assertIsInstance(read_hed_version, HedVersion) + self.assertEqual(read_hed_version.version, "8.2.0") + + +class TestHedVersionNWBFileSimpleRoundtrip(TestCase): + """Simple roundtrip test for HedTags.""" + + def setUp(self): + self.path = "test.nwb" + self.start_time = datetime(1970, 1, 1, 12, tzinfo=tzutc()) + self.ref_time = datetime(1979, 1, 1, 0, tzinfo=tzutc()) + self.create_date = datetime(2017, 4, 15, 12, tzinfo=tzlocal()) + self.manager = get_manager() + self.filename = 'test_nwbfileio.h5' + self.nwbfile = NWBFile(session_description='a test NWB File', + identifier='TEST123', + session_start_time=self.start_time, + timestamps_reference_time=self.ref_time, + file_create_date=self.create_date, + experimenter='test experimenter', + stimulus_notes='test stimulus notes', + data_collection='test data collection notes', + experiment_description='test experiment description', + institution='nomad', + lab='nolab', + notes='nonotes', + pharmacology='nopharmacology', + protocol='noprotocol', + related_publications='nopubs', + session_id='007', + slices='noslices', + source_script='nosources', + surgery='nosurgery', + virus='novirus', + source_script_file_name='nofilename') + hed_version = HedVersion(version="8.2.0") + self.nwbfile.add_lab_meta_data(hed_version) + + def tearDown(self): + remove_test_file(self.path) + + def test_version(self): + hed_version = self.nwbfile.get_lab_meta_data("hed_version") + self.assertIsInstance(hed_version, HedVersion) + schema = hed_version.get_schema() + self.assertIsInstance(schema, HedSchema) + + def test_roundtrip(self): + with NWBHDF5IO(self.path, mode="w") as io: + io.write(self.nwbfile) + + with NWBHDF5IO(self.path, mode="r", load_namespaces=True) as io: + read_nwbfile = io.read() + read_hed_version = read_nwbfile.get_lab_meta_data("hed_version") + self.assertIsInstance(read_hed_version, HedVersion) + self.assertEqual(read_hed_version.version, "8.2.0") diff --git a/src/pynwb/tests/test_hed_version_attr.py b/src/pynwb/tests/test_hed_version_attr.py new file mode 100644 index 0000000..7fa7c54 --- /dev/null +++ b/src/pynwb/tests/test_hed_version_attr.py @@ -0,0 +1,139 @@ +"""Unit and integration tests for ndx-hed.""" +from datetime import datetime, timezone +from dateutil.tz import tzlocal, tzutc +from uuid import uuid4, UUID +from pynwb.file import LabMetaData +from pynwb import NWBHDF5IO, get_manager, NWBFile +from pynwb.testing.mock.file import mock_NWBFile +from pynwb.testing import TestCase, remove_test_file, NWBH5IOFlexMixin +from src.pynwb.ndx_hed import HedVersionAttr +from hed.schema import HedSchema, HedSchemaGroup + + +class TestHedVersionAttr(TestCase): + """Simple unit test for creating a HedMetadata.""" + + def test_constructor(self): + """Test setting HedNWBFile values using the constructor.""" + hed_version_example = HedVersionAttr("8.2.0") + self.assertIsInstance(hed_version_example, HedVersionAttr) + self.assertIsInstance(hed_version_example, LabMetaData) + self.assertEqual(hed_version_example.name, "hed_version") + self.assertIsInstance(hed_version_example.version, str) + self.assertEqual(hed_version_example.version, "8.2.0") + self.assertIsInstance(hed_version_example.schema_string, str) + # hed_version2 = HedVersion(["8.2.0"]) + # self.assertIsInstance(hed_version2.version, list) + # hed_version3 = HedVersion(["8.2.0", "sc:score_1.0.0"]) + + def test_get_schema(self): + hed_version_example = HedVersionAttr(version="8.2.0") + self.assertEqual(hed_version_example.name, "hed_version") + hed_schema = hed_version_example.get_schema() + self.assertIsInstance(hed_schema, HedSchema) + + def test_get_version(self): + hed_version_example = HedVersionAttr(version="8.2.0") + version = hed_version_example.get_version() + self.assertIsInstance(version, str) + self.assertEqual(version, "8.2.0") + + def test_add_to_nwbfile(self): + nwbfile = mock_NWBFile() + hed_version_example = HedVersionAttr(version="8.2.0") + nwbfile.add_lab_meta_data(lab_meta_data=hed_version_example) + hed_version_new = nwbfile.get_lab_meta_data(hed_version_example.name) + self.assertIsInstance(hed_version_new, HedVersionAttr) + # hed_version2 = HedVersion(["8.2.0"]) + + +class TestHedVersionSimpleRoundtrip(TestCase): + """Simple roundtrip test for HedNWBFile.""" + + def setUp(self): + self.path = "test.nwb" + + def tearDown(self): + remove_test_file(self.path) + + def test_roundtrip(self): + """ + Create a HedMetadata, write it to file, read the file, and test that it matches the original HedNWBFile. + """ + nwbfile = mock_NWBFile() + hed_version_example = HedVersionAttr(version="8.2.0") + nwbfile.add_lab_meta_data(lab_meta_data=hed_version_example) + hed_version_new = nwbfile.get_lab_meta_data("hed_version") + self.assertIsInstance(hed_version_new, HedVersionAttr) + self.assertEqual(hed_version_new.name, "hed_version") + self.assertEqual(hed_version_new.version, "8.2.0") + self.assertIsInstance(hed_version_new, HedVersionAttr) + self.assertIsInstance(hed_version_new, LabMetaData) + schema = hed_version_new.get_schema() + self.assertIsInstance(schema, HedSchema) + + with NWBHDF5IO(self.path, mode="w") as io: + io.write(nwbfile) + + with NWBHDF5IO(self.path, mode="r", load_namespaces=True) as io: + read_nwbfile = io.read() + read_hed_version = read_nwbfile.get_lab_meta_data(hed_version_example.name) + # self.assertIsInstance(read_hed_version, HedVersionAttr) + # self.assertEqual(read_hed_version.version, "8.2.0") + + +# class TestHedVersionNWBFileSimpleRoundtrip(TestCase): +# """Simple roundtrip test for HedTags.""" +# +# def setUp(self): +# self.path = "test.nwb" +# self.start_time = datetime(1970, 1, 1, 12, tzinfo=tzutc()) +# self.ref_time = datetime(1979, 1, 1, 0, tzinfo=tzutc()) +# self.create_date = datetime(2017, 4, 15, 12, tzinfo=tzlocal()) +# self.manager = get_manager() +# self.filename = 'test_nwbfileio.h5' +# self.nwbfile = NWBFile(session_description='a test NWB File', +# identifier='TEST123', +# session_start_time=self.start_time, +# timestamps_reference_time=self.ref_time, +# file_create_date=self.create_date, +# experimenter='test experimenter', +# stimulus_notes='test stimulus notes', +# data_collection='test data collection notes', +# experiment_description='test experiment description', +# institution='nomad', +# lab='nolab', +# notes='nonotes', +# pharmacology='nopharmacology', +# protocol='noprotocol', +# related_publications='nopubs', +# session_id='007', +# slices='noslices', +# source_script='nosources', +# surgery='nosurgery', +# virus='novirus', +# source_script_file_name='nofilename') +# hed_version = HedVersion(version="8.2.0") +# self.nwbfile.add_lab_meta_data(hed_version) +# print(f"Name:{hed_version.name}") +# +# def tearDown(self): +# remove_test_file(self.path) +# +# def test_version(self): +# hed_version = self.nwbfile.get_lab_meta_data("hed_version") +# self.assertIsInstance(hed_version, HedVersion) +# schema = hed_version.get_schema() +# self.assertIsInstance(schema, HedSchema) +# +# def test_roundtrip(self): +# print(self.nwbfile) +# with NWBHDF5IO(self.path, mode="w") as io: +# io.write(self.nwbfile) +# +# with NWBHDF5IO(self.path, mode="r", load_namespaces=True) as io: +# read_nwbfile = io.read() +# print(read_nwbfile) +# read_hed_version = read_nwbfile.get_lab_meta_data("hed_version") +# #self.assertIsInstance(read_hed_version, HedVersion) +# self.assertEqual(read_hed_version.version, "8.2.0") diff --git a/src/spec/create_extension_spec.py b/src/spec/create_extension_spec.py index 969e777..2d1fbb3 100644 --- a/src/spec/create_extension_spec.py +++ b/src/spec/create_extension_spec.py @@ -28,31 +28,21 @@ def main(): ) ns_builder.include_namespace("core") + ns_builder.include_type('LabMetaData', namespace='core') + ns_builder.include_type('VectorData', namespace='core') - # TODO: define your new data types - # see https://pynwb.readthedocs.io/en/latest/extensions.html#extending-nwb - # for more information - hed_annotations = NWBDatasetSpec( - neurodata_type_def="HedAnnotations", + hed_tags = NWBDatasetSpec( + neurodata_type_def="HedTags", neurodata_type_inc="VectorData", doc=("An extension of VectorData for Hierarchical Event Descriptor (HED) tags. If HED tags are used, " - "the HED schema version must be specified in the NWB file using the HedVersion type."), + "the HED schema version must be specified in the NWB file using the HedMetadata type."), dtype="text", - attributes=[ - NWBAttributeSpec( - name='sub_name', - dtype='text', - doc=('The smallest possible difference between two event times. Usually 1 divided by the event time ' - 'sampling rate on the data acquisition system.'), - required=False, - ), - ], ) - hed_version = NWBGroupSpec( - neurodata_type_def="HedVersion", - neurodata_type_inc="LabMetaData", + hed_version_attr = NWBGroupSpec( name="hed_version", # fixed name + neurodata_type_def="HedVersionAttr", + neurodata_type_inc="LabMetaData", doc=("An extension of LabMetaData to store the Hierarchical Event Descriptor (HED) schema version. " "TODO When merged with core, " "this will no longer inherit from LabMetaData but from NWBContainer and be placed " @@ -65,14 +55,31 @@ def main(): "Required if HED tags are used in the NWB file." ), dtype='text', - required=True, - shape=[None,] + required=True + #shape=[None,] ) - ], + ] + ) + + hed_version = NWBGroupSpec( + name="hed_version", # fixed name + neurodata_type_def="HedVersion", + neurodata_type_inc="LabMetaData", + doc=("An extension of LabMetaData to store the Hierarchical Event Descriptor (HED) schema version. " + "TODO When merged with core, " + "this will no longer inherit from LabMetaData but from NWBContainer and be placed " + "optionally in /general."), ) + hed_version.add_dataset( + name="version", + doc="HED schema version to use for this dataset", + dtype='text', + quantity='?' + ) + # TODO: add all of your new data types to this list - new_data_types = [hed_annotations, hed_version] + new_data_types = [hed_version_attr, hed_tags, hed_version] # export the spec to yaml files in the spec folder output_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "spec")) @@ -82,4 +89,4 @@ def main(): if __name__ == "__main__": # usage: python create_extension_spec.py - main() + main() \ No newline at end of file