Quick pass on duration

Author: IanCa

hed/models/model_constants.py

@@ -1,10 +1,4 @@
 """ Defined constants for definitions, def labels, and expanded labels. """
-COLUMN_TO_HED_TAGS = "column_to_hed_tags"
-ROW_HED_STRING = "HED"
-COLUMN_ISSUES = "column_issues"
-ROW_ISSUES = "row_issues"
-
-
 class DefTagNames:
     """ Source names for definitions, def labels, and expanded labels. """
 
@@ -19,9 +13,11 @@ class DefTagNames:
     ONSET_ORG_KEY = "Onset"
     OFFSET_ORG_KEY = "Offset"
     INSET_ORG_KEY = "Inset"
+    DURATION_ORG_KEY = "Duration"
 
     ONSET_KEY = ONSET_ORG_KEY.lower()
     OFFSET_KEY = OFFSET_ORG_KEY.lower()
     INSET_KEY = INSET_ORG_KEY.lower()
+    DURATION_KEY = DURATION_ORG_KEY.lower()
 
     TEMPORAL_KEYS = {ONSET_KEY, OFFSET_KEY, INSET_KEY}

hed/tools/analysis/event_manager.py

@@ -1,5 +1,6 @@
 """ Manager of events of temporal extent. """
 import pandas as pd
+import bisect
 
 from hed.errors import HedFileError
 from hed.models import HedString
@@ -28,7 +29,6 @@ def __init__(self, input_data, hed_schema, extra_defs=None):
         are separated from the rest of the annotations, which are contained in self.hed_strings.
 
         """
-
         self.event_list = [[] for _ in range(len(input_data.dataframe))]
         self.hed_schema = hed_schema
         self.input_data = input_data
@@ -56,11 +56,26 @@ def _create_event_list(self, input_data):
         onset_dict = {}  # Temporary dictionary keeping track of temporal events that haven't ended yet.
         for event_index, hed in enumerate(hed_strings):
             self._extract_temporal_events(hed, event_index, onset_dict)
+            self._extract_duration_events(hed, event_index)
         # Now handle the events that extend to end of list
         for item in onset_dict.values():
             item.set_end(len(self.onsets), None)
         self.hed_strings = hed_strings
 
+    def _extract_duration_events(self, hed, event_index):
+        groups = hed.find_top_level_tags(anchor_tags={DefTagNames.DURATION_KEY})
+        to_remove = []
+        for duration_tag, group in groups:
+            start_time = self.onsets[event_index]
+            new_event = TemporalEvent(group, event_index, start_time)
+            end_time = new_event.end_time
+            # Todo: This may need updating.  end_index==len(self.onsets) in the edge case
+            end_index = bisect.bisect_right(self.onsets, end_time)
+            new_event.set_end(end_index, end_time)
+            self.event_list[event_index].append(new_event)
+            to_remove.append(group)
+        hed.remove(to_remove)
+
     def _extract_temporal_events(self, hed, event_index, onset_dict):
         """ Extract the temporal events and remove them from the other HED strings.
 
@@ -77,18 +92,19 @@ def _extract_temporal_events(self, hed, event_index, onset_dict):
             return
         group_tuples = hed.find_top_level_tags(anchor_tags={DefTagNames.ONSET_KEY, DefTagNames.OFFSET_KEY},
                                                include_groups=2)
+
         to_remove = []
-        for tup in group_tuples:
-            anchor_tag = tup[1].find_def_tags(recursive=False, include_groups=0)[0]
+        for def_tag, group in group_tuples:
+            anchor_tag = group.find_def_tags(recursive=False, include_groups=0)[0]
             anchor = anchor_tag.extension.lower()
-            if anchor in onset_dict or tup[0].short_base_tag.lower() == DefTagNames.OFFSET_KEY:
+            if anchor in onset_dict or group.short_base_tag.lower() == DefTagNames.OFFSET_KEY:
                 temporal_event = onset_dict.pop(anchor)
                 temporal_event.set_end(event_index, self.onsets[event_index])
-            if tup[0] == DefTagNames.ONSET_KEY:
-                new_event = TemporalEvent(tup[1], event_index, self.onsets[event_index])
+            if def_tag == DefTagNames.ONSET_KEY:
+                new_event = TemporalEvent(group, event_index, self.onsets[event_index])
                 self.event_list[event_index].append(new_event)
                 onset_dict[anchor] = new_event
-            to_remove.append(tup[1])
+            to_remove.append(group)
         hed.remove(to_remove)
 
     def unfold_context(self, remove_types=[]):

hed/tools/analysis/temporal_event.py

@@ -40,7 +40,9 @@ def _split_group(self, contents):
                 to_remove.append(item)
             elif item.short_base_tag.lower() == "duration":
                 to_remove.append(item)
-                self.end_time = self.start_time + float(item.extension.lower())  # Will need to be fixed for units
+                self.end_time = self.start_time + item.value_as_default_unit()
+                # We don't want the empty group
+                contents = contents.get_first_group()
             elif item.short_base_tag.lower() == "def":
                 self.anchor = item.short_tag
         contents.remove(to_remove)

tests/tools/analysis/test_event_manager.py

@@ -107,11 +107,19 @@ def test_onset_ordering_bad(self):
 
     def test_duration_context(self):
         df = pd.DataFrame({'onset': [1, 2, 3, 4, 5],
-                           'HED': ['(Duration/4.0 s, Black)', '(Duration/2 s, Red)', 'Blue', 'n/a', 'Label/1']})
+                           'HED': ['(Duration/4.0 s, (Black))', '(Duration/2 s, (Red))', 'Blue', 'n/a', 'Label/1']})
         manager = EventManager(TabularInput(df), self.schema)
         hed, base, context = manager.unfold_context()
         pass
 
+    def test_duration_context2(self):
+        df = pd.DataFrame({'onset': [1, 2, 3, 4, 5],
+                           'HED': ['(Duration/5.0 s, (Black))', '(Duration/2 s, (Red))', 'Blue', 'n/a', 'Label/1']})
+        manager = EventManager(TabularInput(df), self.schema)
+        hed, base, context = manager.unfold_context()
+        pass
+
+
 
 if __name__ == '__main__':
     unittest.main()