text-slicer app feature now promoted to SDK

mmif/serialize/mmif.py

@@ -1071,54 +1071,99 @@ def get_annotations_of_labels(
                     continue
                 yield ann
 
+    def _get_ms_anchor(self, ann: Annotation) -> tuple[int, int]:
+        """
+        Start and end of a ``TimeFrame`` or ``TimePoint`` in integer
+        milliseconds. A missing ``timeUnit`` means milliseconds.
+
+        :raises ValueError: if the time unit is not milliseconds, or a time
+            point is not an integer
+        """
+        from mmif.utils.timeunit_helper import UNIT_NORMALIZATION
+
+        declared = ann.get_property('timeUnit')
+        unit = 'millisecond' if declared is None \
+            else UNIT_NORMALIZATION.get(str(declared).lower(), declared)
+        if unit != 'millisecond':
+            raise ValueError(f"{ann.id} declares timeUnit={declared!r}; "
+                             f"time anchors must be integer milliseconds")
+        points = []
+        for point in (self.get_start(ann), self.get_end(ann)):
+            # integral floats (1500.0) are accepted, as some writers emit them
+            if isinstance(point, bool) or not isinstance(point, (int, float)) \
+                    or (isinstance(point, float) and not point.is_integer()):
+                raise ValueError(f"{ann.id} has the time point {point!r}, "
+                                 f"which is not an integer number of milliseconds")
+            points.append(int(point))
+        return points[0], points[1]
+
     def _is_in_time_range(self, ann: Annotation, range_s: int | float, range_e: int | float) -> bool:
         """
-        Checks if the annotation is anchored within the given time range. Any overlap is considered included. 
-
-        :param ann: the Annotation object to check, must be time-based itself or anchored to time-based annotations
-        :param range_s: the start time point of the range (in milliseconds)
-        :param range_e: the end time point of the range (in milliseconds)
-
-        :return: True if the annotation is anchored within the time range, False otherwise
+        True if the annotation overlaps ``[range_s, range_e)``, all in integer
+        milliseconds. A zero-length annotation counts when it falls inside, so
+        ranges that meet end to end share no annotation.
         """
-        ann_s, ann_e = self.get_start(ann), self.get_end(ann)
-        return (ann_s < range_s < ann_e) or (ann_s < range_e < ann_e) or (ann_s > range_s and ann_e < range_e)
+        ann_s, ann_e = self._get_ms_anchor(ann)
+        if ann_s == ann_e:
+            return range_s <= ann_s < range_e
+        return ann_s < range_e and ann_e > range_s
 
     def get_annotations_between_time(self, start: int | float, end: int | float, time_unit: str = "ms",
                                      at_types: list[ThingTypesBase | str] = []) -> Iterator[Annotation]:
         """
-        Finds annotations that are anchored between the given time points.
+        Finds the ``TimeFrame`` and ``TimePoint`` annotations that overlap the
+        time range, in time order, each followed by the annotations aligned to
+        it. Each annotation is returned once. All time anchors in the MMIF
+        must be in integer milliseconds.
 
-        :param start: the start time point
-        :param end: the end time point
-        :param time_unit: the unit of the input time points. Default is `ms`.
+        :param start: the start time point, in milliseconds
+        :param end: the end time point, in milliseconds
+        :param time_unit: the unit of ``start`` and ``end``. Default is `ms`.
+            (deprecated since 1.8.0, will be removed in 2.0.0)
         :param at_types: a list of annotation types to filter with. Any type in this list will be included in the return.
         :return: an iterator of Annotation objects that are anchored between the given time points
-        """
+        :raises ValueError: if a time anchor in the MMIF is not in integer milliseconds
+        """
+        from mmif.utils.timeunit_helper import UNIT_NORMALIZATION, convert
+
+        unit = UNIT_NORMALIZATION.get(str(time_unit).lower())
+        if unit is None:
+            raise ValueError(f"Not supported time unit: {time_unit}")
+        if unit == 'frame':
+            raise ValueError("time points in frames cannot be converted to "
+                             "milliseconds without a frame rate")
+        if unit != 'millisecond':
+            warnings.warn('The time_unit argument of '
+                          'get_annotations_between_time() is deprecated and '
+                          'will be removed in 2.0.0. Use integer milliseconds '
+                          'instead.',
+                          DeprecationWarning, stacklevel=2)
+            start = round(cast(int | float, convert(start, time_unit, 'milliseconds', 1)))
+            end = round(cast(int | float, convert(end, time_unit, 'milliseconds', 1)))
         assert start < end, f"Start time point must be smaller than the end time point, given {start} and {end}"
         assert start >= 0, f"Start time point must be non-negative, given {start}"
         assert end >= 0, f"End time point must be non-negative, given {end}"
-
-        from mmif.utils.timeunit_helper import convert
 
         time_anchors_in_range = []
         uniq_types = set(ThingTypesBase.from_str(t) if isinstance(t, str) else t for t in at_types)
 
-        for view in self.get_all_views_contain(AnnotationTypes.TimeFrame) + self.get_all_views_contain(AnnotationTypes.TimePoint):
-            time_unit_in_view = view.metadata.contains.get(AnnotationTypes.TimeFrame)["timeUnit"]
-            
-            start_time = cast(int | float, convert(start, time_unit, time_unit_in_view, 1))
-            end_time = cast(int | float, convert(end, time_unit, time_unit_in_view, 1))
+        anchor_types = (AnnotationTypes.TimeFrame, AnnotationTypes.TimePoint)
+        # a view that contains both anchor types is visited once
+        views = {view.id: view for anchor_type in anchor_types
+                 for view in self.get_all_views_contain(anchor_type)}
+        for view in views.values():
             for ann in view.get_annotations():
-                if ann.at_type in (AnnotationTypes.TimeFrame, AnnotationTypes.TimePoint) and self._is_in_time_range(ann, start_time, end_time):
+                if ann.at_type in anchor_types and self._is_in_time_range(ann, start, end):
                     time_anchors_in_range.append(ann)
         time_anchors_in_range.sort(key=lambda x: self.get_start(x))
+        yielded = set()
         for time_anchor in time_anchors_in_range:
-            if not uniq_types or time_anchor.at_type in uniq_types:
-                yield time_anchor
-            for aligned in time_anchor.get_all_aligned():
-                if not uniq_types or aligned.at_type in uniq_types:
-                    yield aligned
+            for ann in (time_anchor, *time_anchor.get_all_aligned()):
+                if ann.id in yielded:
+                    continue
+                if not uniq_types or ann.at_type in uniq_types:
+                    yielded.add(ann.id)
+                    yield ann
 
     def _get_linear_anchor_point(self, ann: Annotation, targets_sorted=False, start: bool = True) -> int | float:
         # TODO (krim @ 2/5/24): Update the return type once timeunits are unified to `ms` as integers (https://github.com/clamsproject/mmif/issues/192)

mmif/utils/text_document_helper.py

@@ -1,20 +1,27 @@
+import warnings
+
 from mmif.vocabulary import AnnotationTypes
 
 
 def slice_text(mmif_obj, start: int, end: int, unit: str = "milliseconds") -> str:
     """
-    Extracts text from tokens within a specified time range.
+    .. deprecated:: 1.8.0
+       Will be removed in 2.0.0. Use
+       :meth:`~mmif.serialize.mmif.Mmif.get_annotations_between_time` with
+       ``at_types=[AnnotationTypes.Token]`` instead.
+
+    Extracts text from tokens within a specified time range, joining the
+    tokens of every text document.
 
     :param mmif_obj: MMIF object to search for tokens
     :param start: start time point
     :param end: end time point
     :param unit: time unit for start and end parameters (default: "milliseconds")
     :return: space-separated string of token words found in the time range
     """
-    token_type = AnnotationTypes.Token
-    anns_found = mmif_obj.get_annotations_between_time(start, end, unit)
-    tokens_sliced = []
-    for ann in anns_found:
-        if ann.is_type(token_type):
-            tokens_sliced.append(ann.get_property('word'))
-    return ' '.join(tokens_sliced)
+    warnings.warn('slice_text() is deprecated and will be removed in 2.0.0. '
+                  'Use Mmif.get_annotations_between_time() instead.',
+                  DeprecationWarning, stacklevel=2)
+    tokens = mmif_obj.get_annotations_between_time(
+        start, end, unit, at_types=[AnnotationTypes.Token])
+    return ' '.join(token.get_property('text') for token in tokens)

tests/test_serialize.py

@@ -774,15 +774,19 @@ def test_get_annotations_between_time(self):
         for i, ann in enumerate(selected_token_anns):
             self.assertEqual(tokens_in_order[i], ann.get_property("word"))
 
-        # Test case 2: No token annotation are selected
-        selected_token_anns = list(mmif_obj.get_annotations_between_time(0, 5, time_unit="seconds"))
-        self.assertEqual(4, len(list(selected_token_anns))) 
+        # Test case 2: No token annotation are selected; v1:s1 starts at 0,
+        # the range start, so it is among the five
+        # (query times in seconds are deprecated, so they warn)
+        with self.assertWarns(DeprecationWarning):
+            selected_token_anns = list(mmif_obj.get_annotations_between_time(0, 5, time_unit="seconds"))
+        self.assertEqual(5, len(list(selected_token_anns)))
         for ann in selected_token_anns:
             self.assertFalse(ann.is_type(token_type))
 
         # Test case 3(a): Partial tokens are selected (involve partial overlap)
-        selected_token_anns = mmif_obj.get_annotations_between_time(7, 10, time_unit="seconds", 
-                                                                    at_types=['http://vocab.lappsgrid.org/Token'])
+        with self.assertWarns(DeprecationWarning):
+            selected_token_anns = list(mmif_obj.get_annotations_between_time(7, 10, time_unit="seconds",
+                                                                             at_types=['http://vocab.lappsgrid.org/Token']))
         self.assertEqual(tokens_in_order[3:9], [ann.get_property("word") for ann in selected_token_anns])
 
         # Test case 3(b): Partial tokens are selected (only full overlap)

tests/test_utils.py

@@ -1053,6 +1053,148 @@ def test_slice_text(self):
             full_sliced_text)
 
 
+class TestTokensBetweenTime(unittest.TestCase):
+    """Transcript text of a time range, through
+    ``Mmif.get_annotations_between_time``, one text per document."""
+    # (word, start, end) in milliseconds; "delta" is zero-length at 900
+    WORDS = [('alpha', 0, 300), ('beta', 300, 600), ('gamma', 600, 900),
+             ('delta', 900, 900), ('epsilon', 900, 1200)]
+
+    def setUp(self):
+        mmif_obj = Mmif(validate=False)
+        video = Document()
+        video.at_type = DocumentTypes.VideoDocument
+        video.id = 'd1'
+        video.location = 'file:///video.mp4'
+        mmif_obj.add_document(video)
+        self.mmif_obj = mmif_obj
+        self.td_id, self.asr_tfs = self.add_asr('test-asr', self.WORDS)
+        # a view of TimePoints only, as a frame classifier produces
+        points = mmif_obj.new_view()
+        points.metadata.app = 'http://apps.clams.ai/test-points/v1'
+        points.new_contain(AnnotationTypes.TimePoint,
+                           timeUnit='milliseconds', document='d1')
+        points.new_annotation(AnnotationTypes.TimePoint, timePoint=450)
+        self.segments = mmif_obj.new_view()
+        self.segments.metadata.app = 'http://apps.clams.ai/test-seg/v1'
+        self.segments.new_contain(AnnotationTypes.TimeFrame,
+                                  timeUnit='milliseconds', document='d1')
+
+    def add_asr(self, app_name, words):
+        """A whisper-shaped view: one Token per word, aligned to a word-level
+        TimeFrame. Returns the TextDocument id and the TimeFrames."""
+        asr = self.mmif_obj.new_view()
+        asr.metadata.app = f'http://apps.clams.ai/{app_name}/v1'
+        td = asr.new_textdocument(' '.join(w for w, _, _ in words))
+        asr.new_contain(AnnotationTypes.TimeFrame, timeUnit='milliseconds',
+                        document='d1')
+        tfs = []
+        for word, start, end in words:
+            tok = asr.new_annotation(AnnotationTypes.Token, text=word,
+                                     document=td.id)
+            tf = asr.new_annotation(AnnotationTypes.TimeFrame,
+                                    start=start, end=end)
+            asr.new_annotation(AnnotationTypes.Alignment,
+                               source=tf.id, target=tok.id)
+            tfs.append(tf)
+        return td.id, tfs
+
+    def segment(self, start, end, **props):
+        return self.segments.new_annotation(AnnotationTypes.TimeFrame,
+                                            start=start, end=end, **props)
+
+    def texts(self, start, end):
+        texts = {}
+        for token in self.mmif_obj.get_annotations_between_time(
+                start, end, at_types=[AnnotationTypes.Token]):
+            texts.setdefault(token.get_property('document'), []).append(
+                token.get_property('text'))
+        return {doc: ' '.join(words) for doc, words in texts.items()}
+
+    def test_tokens_on_range_boundaries_are_included(self):
+        self.assertEqual({self.td_id: 'alpha beta gamma'},
+                         self.texts(0, 900))
+
+    def test_adjacent_ranges_share_no_token(self):
+        self.assertEqual({self.td_id: 'alpha beta'}, self.texts(0, 600))
+        self.assertEqual({self.td_id: 'gamma'}, self.texts(600, 900))
+
+    def test_zero_length_token_belongs_to_the_range_it_starts(self):
+        self.assertEqual({self.td_id: 'delta epsilon'},
+                         self.texts(900, 1200))
+
+    def test_partial_overlap_is_included(self):
+        self.assertEqual({self.td_id: 'alpha beta'}, self.texts(100, 400))
+
+    def test_no_token_in_range(self):
+        self.assertEqual({}, self.texts(2000, 3000))
+
+    def test_transcripts_are_kept_per_document(self):
+        other_id, _ = self.add_asr(
+            'other-asr', [('uno', 0, 450), ('dos', 450, 900)])
+        self.assertEqual({self.td_id: 'alpha beta gamma',
+                          other_id: 'uno dos'},
+                         self.texts(0, 900))
+
+    def test_token_aligned_twice_is_returned_once(self):
+        # a second view (as a forced aligner makes) aligns "beta" again
+        aligner = self.mmif_obj.new_view()
+        aligner.metadata.app = 'http://apps.clams.ai/test-aligner/v1'
+        aligner.new_contain(AnnotationTypes.TimeFrame,
+                            timeUnit='milliseconds', document='d1')
+        tf = aligner.new_annotation(AnnotationTypes.TimeFrame,
+                                    start=320, end=580)
+        beta = self.asr_tfs[1].get_all_aligned()
+        token = next(a for a in beta if a.is_type(AnnotationTypes.Token))
+        aligner.new_annotation(AnnotationTypes.Alignment,
+                               source=tf.id, target=token.id)
+        self.assertEqual({self.td_id: 'alpha beta'}, self.texts(0, 600))
+
+    def test_integral_float_anchor_times_are_accepted(self):
+        self.segment(0.0, 600.0)
+        self.assertEqual({self.td_id: 'alpha beta'}, self.texts(0, 600))
+
+    def test_non_integer_anchor_time_raises(self):
+        self.segment(0.5, 600)
+        with self.assertRaises(ValueError):
+            self.texts(0, 600)
+
+    def test_non_millisecond_unit_on_an_anchor_raises(self):
+        self.segment(0, 1, timeUnit='seconds')
+        with self.assertRaises(ValueError):
+            self.texts(0, 600)
+
+    def test_non_millisecond_view_anywhere_raises(self):
+        legacy = self.mmif_obj.new_view()
+        legacy.metadata.app = 'http://apps.clams.ai/test-legacy/v1'
+        legacy.new_contain(AnnotationTypes.TimeFrame, timeUnit='frames',
+                           document='d1')
+        legacy.new_annotation(AnnotationTypes.TimeFrame, start=0, end=30)
+        with self.assertRaises(ValueError):
+            self.texts(0, 600)
+
+    def test_query_in_seconds_is_converted_with_warning(self):
+        with self.assertWarns(DeprecationWarning):
+            tokens = list(self.mmif_obj.get_annotations_between_time(
+                0, 0.6, time_unit='seconds',
+                at_types=[AnnotationTypes.Token]))
+        self.assertEqual(['alpha', 'beta'],
+                         [t.get_property('text') for t in tokens])
+
+    def test_query_in_frames_raises(self):
+        with self.assertRaises(ValueError):
+            list(self.mmif_obj.get_annotations_between_time(
+                0, 10, time_unit='frames'))
+
+    def test_slice_text_is_deprecated(self):
+        other_id, _ = self.add_asr(
+            'other-asr', [('uno', 0, 450), ('dos', 450, 900)])
+        with self.assertWarns(DeprecationWarning):
+            text = tdh.slice_text(self.mmif_obj, 0, 900)
+        # it joins every document
+        self.assertEqual('alpha uno beta dos gamma', text)
+
+
 class TestWorkflowHelper(unittest.TestCase):
 
     def setUp(self) -> None: