releasing 1.8.0

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/audio_document_helper.py

@@ -0,0 +1,217 @@
+import warnings
+from collections.abc import Iterator
+from typing import TYPE_CHECKING
+
+from mmif import Document
+from mmif.utils.av_helper import (
+    _check_av_dep,
+    _check_document,
+    _ffmpeg_op,
+    _media_check_hint,
+    _open_container_and_stream,
+)
+from mmif.utils.timeunit_helper import iter_time_windows
+from mmif.vocabulary import DocumentTypes
+
+if TYPE_CHECKING:
+    import numpy
+
+AUDIO_DOCUMENT_TYPES = (DocumentTypes.AudioDocument, DocumentTypes.VideoDocument)
+# encoder padding can leave the decoded audio a few tens of ms short of the
+# stated duration; anything shorter than this is not reported as truncation
+TRUNCATION_TOLERANCE_MS = 100
+
+
+def iter_audio_windows(
+    document: Document,
+    window_ms: int,
+    stride_ms: int | None = None,
+    sample_rate: int = 16000,
+    min_last_window_ms: int = 0,
+) -> Iterator[tuple["numpy.ndarray", int, int]]:
+    """
+    Decodes an audio document, or the audio track of a video, into windows.
+
+    Samples are resampled to mono ``float32`` at ``sample_rate`` and decoded
+    incrementally, so memory use is bounded by one window plus its overlap.
+    The windows are those of :func:`~mmif.utils.timeunit_helper.iter_time_windows`
+    from the start of the audio stream, which can be later than zero. After
+    the last full window, one window runs from the next window start to the
+    end of the stream, unless it is shorter than ``min_last_window_ms``.
+
+    :param document: audio or video document with an audio stream
+    :param window_ms: window length in milliseconds
+    :param stride_ms: distance between window starts; defaults to ``window_ms``
+    :param sample_rate: output sample rate in Hz
+    :param min_last_window_ms: shortest final window to keep
+    :returns: ``(samples, start_ms, end_ms)`` triples
+    :raises ValueError: if a size is not positive, ``min_last_window_ms`` is
+        outside ``[0, window_ms]``, or the document has an unsupported type
+    :raises FileNotFoundError: if there is no document or no file
+    :raises RuntimeError: if the file cannot be opened, has no audio stream,
+        or fails to decode
+    :warns UserWarning: if the decoded audio ends more than
+        ``TRUNCATION_TOLERANCE_MS`` before the duration the file states
+    """
+    if window_ms <= 0:
+        raise ValueError('window_ms must be positive')
+    if stride_ms is None:
+        stride_ms = window_ms
+    elif stride_ms <= 0:
+        raise ValueError('stride_ms must be positive')
+    if sample_rate <= 0:
+        raise ValueError('sample_rate must be positive')
+    if not 0 <= min_last_window_ms <= window_ms:
+        raise ValueError('min_last_window_ms must be between 0 and window_ms')
+    _check_document(document, AUDIO_DOCUMENT_TYPES)
+    return _iter_audio_windows(document, window_ms, stride_ms, sample_rate, min_last_window_ms)
+
+
+def _iter_audio_windows(document, window_ms, stride_ms, sample_rate, min_last_window_ms):
+    av = _check_av_dep()
+    numpy = _check_av_dep('numpy')
+    container, stream, path = _open_container_and_stream(document, 'audio', AUDIO_DOCUMENT_TYPES)
+    # channels are averaged here, as librosa does; swresample's own downmix
+    # scales a stereo pair by the square root of two
+    resampler = None
+    stream_start_s = 0.0
+    if stream.start_time is not None and stream.time_base is not None:
+        stream_start_s = float(stream.start_time * stream.time_base)
+
+    # sample indices count from the first decoded sample; `origin_ms` is its
+    # time on the stream timeline and the start of the window grid
+    origin_ms = None
+    origin_s = 0.0
+    windows: Iterator[tuple[int, int]] = iter(())
+    window = None
+    chunks = []  # decoded samples from `buffer_start` on, in arrival order
+    buffer_start = 0
+    decoded_end = 0
+    window_samples = round(window_ms * sample_rate / 1000)
+    min_last_samples = max(1, round(min_last_window_ms * sample_rate / 1000))
+
+    def to_sample(ms):
+        return round((ms - origin_ms) * sample_rate / 1000)
+
+    def to_ms(sample):
+        return origin_ms + round(sample * 1000 / sample_rate)
+
+    def append(samples):
+        nonlocal decoded_end
+        # a stride longer than the window leaves samples that no window needs
+        skipped = buffer_start - decoded_end
+        decoded_end += len(samples)
+        if skipped > 0:
+            samples = samples[skipped:]
+        if len(samples):
+            chunks.append(samples)
+
+    def buffered():
+        if len(chunks) != 1:
+            chunks[:] = [numpy.concatenate(chunks)] if chunks else [numpy.empty(0, numpy.float32)]
+        return chunks[0]
+
+    def emit(final=False):
+        nonlocal window, buffer_start
+        while True:
+            if window is None:
+                window = next(windows)
+            start_ms, end_ms = window
+            start, end = to_sample(start_ms), to_sample(end_ms)
+            if end > decoded_end:
+                if not final or decoded_end - start < min_last_samples:
+                    return
+                end, end_ms = decoded_end, to_ms(decoded_end)
+            yield buffered()[start - buffer_start:end - buffer_start].copy(), start_ms, end_ms
+            if final:
+                return
+            window = None
+            keep_from = to_sample(start_ms + stride_ms)
+            chunks[:] = [buffered()[keep_from - buffer_start:]]
+            buffer_start = keep_from
+
+    def resampled_frames():
+        nonlocal resampler
+        for decoded in container.decode(stream):
+            if resampler is None:
+                resampler = av.AudioResampler(format='fltp', layout=decoded.layout, rate=sample_rate)
+            yield from resampler.resample(decoded)
+        if resampler is not None:
+            yield from resampler.resample(None)
+
+    # contiguous resampled frames wait here, so that numpy runs once per
+    # second of audio and not once per (often 1024-sample) frame
+    fifo = av.AudioFifo()
+
+    def flush():
+        if fifo.samples:
+            append(fifo.read().to_ndarray().mean(axis=0, dtype=numpy.float32))
+            yield from emit()
+
+    stated_end_s = _stated_end_s(av, container, stream, stream_start_s)
+
+    try:
+        with _ffmpeg_op('decode', document, path, 'audio'):
+            for frame in resampled_frames():
+                frame_s = None
+                if frame.pts is not None and frame.time_base is not None:
+                    frame_s = float(frame.pts * frame.time_base)
+                if origin_ms is None:
+                    origin_s = frame_s if frame_s is not None else stream_start_s
+                    origin_ms = round(origin_s * 1000)
+                    windows = iter_time_windows(origin_ms, None, window_ms, stride_ms)
+                if frame_s is not None:
+                    # a timestamp jump is a gap (filled with silence) or an
+                    # overlap (dropped) on the sample timeline
+                    frame_start = round((frame_s - origin_s) * sample_rate)
+                    gap = frame_start - decoded_end - fifo.samples
+                    if gap:
+                        yield from flush()
+                    while gap > 0:
+                        append(numpy.zeros(min(gap, window_samples), numpy.float32))
+                        yield from emit()
+                        gap = frame_start - decoded_end
+                    if gap < 0:
+                        samples = frame.to_ndarray().mean(axis=0, dtype=numpy.float32)
+                        append(samples[-gap:])
+                        yield from emit()
+                        continue
+                # the fifo checks pts against its own count from zero
+                frame.pts = None
+                fifo.write(frame)
+                if fifo.samples >= sample_rate:
+                    yield from flush()
+            yield from flush()
+            if origin_ms is not None:
+                yield from emit(final=True)
+    finally:
+        container.close()
+    # a cut that ends on a packet boundary decodes cleanly, so only the
+    # duration the file states shows that audio is missing
+    decoded_end_ms = to_ms(decoded_end) if origin_ms is not None else round(stream_start_s * 1000)
+    if stated_end_s is not None:
+        stated_end_ms = round(stated_end_s * 1000)
+        if stated_end_ms - decoded_end_ms > TRUNCATION_TOLERANCE_MS:
+            warnings.warn(
+                f'The audio of {document.id} ends at {decoded_end_ms} ms, '
+                f'{stated_end_ms - decoded_end_ms} ms before the end its file '
+                f'states ({stated_end_ms} ms); the windows stop there. '
+                f'{_media_check_hint(path, "audio")}',
+                stacklevel=2)
+
+
+def _stated_end_s(av, container, stream, stream_start_s):
+    """
+    The end of the audio stream according to the file's metadata, in
+    seconds, or ``None`` when the file does not state it. A container's own
+    duration counts only when it holds no other stream, since it covers the
+    longest stream (a video's audio can end before its picture). A WAV file
+    cut short states its shorter length, since FFmpeg takes it from the file
+    size, so its truncation cannot be seen here.
+    """
+    if stream.duration is not None and stream.time_base is not None:
+        return stream_start_s + float(stream.duration * stream.time_base)
+    if container.duration and len(container.streams) == 1:
+        start_s = (container.start_time or 0) / av.time_base
+        return start_s + container.duration / av.time_base
+    return None