diff --git a/mmif/serialize/mmif.py b/mmif/serialize/mmif.py index 632e1c61..1db3f821 100644 --- a/mmif/serialize/mmif.py +++ b/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) diff --git a/mmif/utils/text_document_helper.py b/mmif/utils/text_document_helper.py index 5598b913..aeb98d74 100644 --- a/mmif/utils/text_document_helper.py +++ b/mmif/utils/text_document_helper.py @@ -1,9 +1,17 @@ +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 @@ -11,10 +19,9 @@ def slice_text(mmif_obj, start: int, end: int, unit: str = "milliseconds") -> st :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) diff --git a/tests/test_serialize.py b/tests/test_serialize.py index 77bb1bf0..0e6037f4 100644 --- a/tests/test_serialize.py +++ b/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) diff --git a/tests/test_utils.py b/tests/test_utils.py index f7ccdcb5..6f7617d5 100644 --- a/tests/test_utils.py +++ b/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: