fix(subtitle): place fragment-relative ISM TTML cues on the track timeline

Smooth Streaming TTML times cues from each fragment's start; a whole track stacked into the first fragment. Also stop rebase_ttml failing on a track with no cues.

docs/guide/subtitles.md

@@ -37,7 +37,8 @@ shown in track listings. The **Extension** column is the file extension used on
     unwrapped into their plain counterparts (`fTTML` becomes TTML and `fVTT` becomes
     WebVTT), so you always end up with an editable text subtitle. For a DASH track, the
     unwrap also subtracts the manifest's `presentationTimeOffset` from the cue times, so the
-    subtitles stay in sync with the video. Note that this is a
+    subtitles stay in sync with the video. For an ISM track whose cues count from the start of
+    each fragment, the unwrap moves every cue onto the timeline of the track. Note that this is a
     one-way, source-only relationship: you can convert *from* these boxed formats, but no
     backend can write them as an output target, so they can never be a conversion
     destination.

tests/tracks/test_subtitle_convert.py

@@ -22,6 +22,7 @@ import pytest
 from unshackle.core import binaries
 from unshackle.core.tracks import subtitle_convert as sc
 from unshackle.core.tracks.subtitle import Subtitle
+from unshackle.core.tracks.track import Track
 
 Codec = Subtitle.Codec
 
@@ -374,32 +375,29 @@ def test_segmented_fttml_parses_and_converts(tmp_path, monkeypatch):
     assert cue_count(out) == 2
 
 
-def dash_subtitle(tmp_path, monkeypatch, codec, data: bytes, template: str):
-    """A DASH Subtitle whose base download is a no-op, so download() runs only the post-processing on data."""
-    from lxml import etree
-
-    from unshackle.core.tracks.track import Track
-
+def downloaded_subtitle(tmp_path, monkeypatch, codec, data: bytes, descriptor: Track.Descriptor, track_data: dict):
+    """A Subtitle whose base download is a no-op, so download() runs only the post-processing on data."""
     monkeypatch.setattr("unshackle.core.config.config.subtitle", {}, raising=False)
     monkeypatch.setattr(Track, "download", lambda *args, **kwargs: None)
-    sub = Subtitle(
-        url="https://example.test/x",
-        language="en",
-        codec=codec,
-        descriptor=Track.Descriptor.DASH,
-        data={
-            "dash": {
-                "adaptation_set": etree.fromstring(f"<AdaptationSet>{template}</AdaptationSet>"),
-                "representation": etree.fromstring("<Representation/>"),
-                "segment_durations": [],
-            }
-        },
-    )
+    sub = Subtitle(url="https://example.test/x", language="en", codec=codec, descriptor=descriptor, data=track_data)
     sub.path = tmp_path / f"seg.{codec.extension}"
     sub.path.write_bytes(data)
     return sub
 
 
+def dash_subtitle(tmp_path, monkeypatch, codec, data: bytes, template: str):
+    from lxml import etree
+
+    track_data = {
+        "dash": {
+            "adaptation_set": etree.fromstring(f"<AdaptationSet>{template}</AdaptationSet>"),
+            "representation": etree.fromstring("<Representation/>"),
+            "segment_durations": [],
+        }
+    }
+    return downloaded_subtitle(tmp_path, monkeypatch, codec, data, Track.Descriptor.DASH, track_data)
+
+
 STPP_CUE = (
     '<?xml version="1.0" encoding="utf-8"?>'
     '<tt xmlns="http://www.w3.org/ns/ttml" xmlns:ttp="http://www.w3.org/ns/ttml#parameter"'
@@ -450,6 +448,66 @@ def test_dash_fvtt_cues_lose_the_presentation_time_offset(tmp_path, monkeypatch)
     assert "00:00:02.000 --> 00:00:03.000" in text
 
 
+def ttml_fragment(*cues: tuple[str, str]) -> str:
+    body = "".join(f'<p begin="{begin}" end="{end}">Line</p>' for begin, end in cues)
+    return f'<tt xmlns="http://www.w3.org/ns/ttml" xml:lang="en"><body><div>{body}</div></body></tt>'
+
+
+def test_dash_fttml_with_no_cues_converts(tmp_path, monkeypatch):
+    template = '<SegmentTemplate timescale="1000" presentationTimeOffset="2000" duration="6000"/>'
+    sub = dash_subtitle(tmp_path, monkeypatch, Codec.fTTML, build_stpp_mp4(ttml_fragment()), template)
+
+    sub.download(None, None)  # type: ignore[arg-type]
+
+    assert sub.codec == Codec.TimedTextMarkupLang
+
+
+@pytest.mark.parametrize(
+    "fragments, expected",
+    [
+        # timed from each fragment's start, as Smooth Streaming packagers write it
+        (
+            [
+                ttml_fragment(),
+                ttml_fragment(("00:00:06.000", "00:00:07.000")),
+                ttml_fragment(("00:00:00.500", "00:00:01.000")),
+            ],
+            [(14.0, 15.0), (16.5, 17.0)],
+        ),
+        # already on the stream timeline
+        (
+            [
+                ttml_fragment(),
+                ttml_fragment(("00:00:14.000", "00:00:15.000")),
+                ttml_fragment(("00:00:16.500", "00:00:17.000")),
+            ],
+            [(14.0, 15.0), (16.5, 17.0)],
+        ),
+        # one cue carried across a fragment boundary comes out whole
+        (
+            [
+                ttml_fragment(),
+                ttml_fragment(("00:00:07.000", "00:00:08.000")),
+                ttml_fragment(("00:00:00.000", "00:00:01.000")),
+            ],
+            [(15.0, 17.0)],
+        ),
+    ],
+)
+def test_ism_fttml_cues_land_on_the_stream_timeline(tmp_path, monkeypatch, fragments, expected):
+    times = [0, 80_000_000, 160_000_000]
+    track_data = {"ism": {"segment_times": times, "fragment_times": times, "timescale": 10_000_000}}
+    sub = downloaded_subtitle(
+        tmp_path, monkeypatch, Codec.fTTML, build_stpp_mp4(*fragments), Track.Descriptor.ISM, track_data
+    )
+
+    sub.download(None, None)  # type: ignore[arg-type]
+
+    assert sub.codec == Codec.TimedTextMarkupLang
+    captions = Subtitle.parse(sub.path.read_bytes(), Codec.TimedTextMarkupLang).get_captions("en")
+    assert [(c.start / 1e6, c.end / 1e6) for c in captions] == expected
+
+
 def test_fttml_with_viewport_units_converts(tmp_path, monkeypatch):
     """TTML2 ``vw``/``vh`` and ``rw``/``rh`` sizes, which pycaption rejects, still parse and convert.
 

unshackle/core/manifests/ism.py

@@ -416,7 +416,9 @@ class ISM:
         segments: list[str] = track.data["ism"]["segments"]
         segment_times: list[int] = track.data["ism"].get("segment_times") or []
         if segment_times:
-            segments = segments[track.sample_slice(segment_times, track.data["ism"]["timescale"], config.sample)]
+            kept = track.sample_slice(segment_times, track.data["ism"]["timescale"], config.sample)
+            segments = segments[kept]
+            track.data["ism"]["fragment_times"] = segment_times[kept]
 
         session_drm = None
         if track.drm:

unshackle/core/tracks/subtitle.py

@@ -269,7 +269,9 @@ class Subtitle(Track):
         if not self.path:
             return
 
-        if self.codec == Subtitle.Codec.fTTML:
+        if self.codec == Subtitle.Codec.fTTML and self.descriptor == Track.Descriptor.ISM:
+            self.unwrap_ism_ttml()
+        elif self.codec == Subtitle.Codec.fTTML:
             self.convert(Subtitle.Codec.TimedTextMarkupLang)
             self.rebase_ttml(*self.dash_offset_ms())
         elif self.codec == Subtitle.Codec.fVTT:
@@ -353,11 +355,11 @@ class Subtitle(Track):
         if offset_ms <= 0 or not self.path:
             return
         caption_set = Subtitle.parse(self.path.read_bytes(), Subtitle.Codec.TimedTextMarkupLang)
-        languages = caption_set.get_languages()
         offset_us = offset_ms * 1000
-        if not languages or caption_set.get_captions(languages[0])[0].start < offset_us - segment_ms * 1000:
+        first = Subtitle.first_cue_start(caption_set)
+        if first is None or first < offset_us - segment_ms * 1000:
             return
-        for language in languages:
+        for language in caption_set.get_languages():
             kept = CaptionList()
             for caption in caption_set.get_captions(language):
                 caption.start = max(caption.start - offset_us, 0)
@@ -367,6 +369,65 @@ class Subtitle(Track):
             caption_set.set_captions(language, kept)
         self.path.write_text(pycaption.DFXPWriter().write(caption_set), encoding="utf8")
 
+    @staticmethod
+    def mdat_payloads(data: bytes) -> list[bytes]:
+        """The payload of every mdat box in a fragmented MP4 file, in file order."""
+        with PYMP4_LOCK:
+            boxes = MP4.parse_stream(BytesIO(data))
+        return [box.data for box in boxes if box.type == b"mdat"]
+
+    @staticmethod
+    def first_cue_start(caption_set: pycaption.CaptionSet) -> Optional[int]:
+        """The start of the first cue, in microseconds, looking past empty languages; None with no cues."""
+        return next(
+            (c.start for language in caption_set.get_languages() for c in caption_set.get_captions(language)[:1]),
+            None,
+        )
+
+    def unwrap_ism_ttml(self) -> None:
+        """
+        Convert ISM fTTML to TTML with every cue on the stream timeline, counted from its first fragment.
+        Smooth Streaming packagers often time the cues of a fragment from that fragment's start. A cue
+        that starts before the previous fragment does marks the stream as timed that way.
+        """
+        from unshackle.core.tracks.subtitle_convert import finalize
+
+        if not self.path:
+            return
+        ism = self.data["ism"]
+        times = ism.get("fragment_times") or []
+        payloads = Subtitle.mdat_payloads(self.path.read_bytes()) if times else []
+        captions: dict[str, CaptionList] = defaultdict(CaptionList)
+        if times and len(payloads) == len(times):
+            fragments = [Subtitle.parse(payload, Subtitle.Codec.TimedTextMarkupLang) for payload in payloads]
+            starts = [t * 1_000_000 // ism["timescale"] for t in times]
+            origin = ism["segment_times"][0] * 1_000_000 // ism["timescale"]
+            relative = any(
+                (first := Subtitle.first_cue_start(fragment)) is not None and first < previous
+                for previous, fragment in zip(starts, fragments[1:])
+            )
+            for start, fragment in zip(starts, fragments):
+                shift = (start if relative else 0) - origin
+                for language in fragment.get_languages():
+                    kept = captions[language]
+                    for caption in fragment.get_captions(language):
+                        caption.start = max(caption.start + shift, 0)
+                        caption.end += shift
+                        if kept and kept[-1].end == caption.start and kept[-1].get_text() == caption.get_text():
+                            kept[-1].end = caption.end  # a cue carried into the next fragment
+                        elif caption.end > 0:
+                            kept.append(caption)
+        if not any(captions.values()):
+            logging.getLogger("Subtitle").debug(f"ISM text: {len(payloads)} fragments, {len(times)} fragment times")
+            self.convert(Subtitle.Codec.TimedTextMarkupLang)
+            return
+
+        caption_set = pycaption.CaptionSet(captions)
+        Subtitle.merge_same_cues(caption_set)
+        out = self.path.with_suffix(f".{Subtitle.Codec.TimedTextMarkupLang.value.lower()}")
+        out.write_text(pycaption.DFXPWriter().write(caption_set), encoding="utf8")
+        finalize(self, Subtitle.Codec.TimedTextMarkupLang, out)
+
     @staticmethod
     def strip_webvtt_cue_classes(text: str) -> str:
         """
@@ -771,10 +832,9 @@ class Subtitle(Track):
                 caption_set = pycaption.SRTReader().read(text)
             elif codec == Subtitle.Codec.fTTML:
                 caption_lists: dict[Optional[str], pycaption.CaptionList] = defaultdict(pycaption.CaptionList)
-                with PYMP4_LOCK:
-                    boxes = MP4.parse_stream(BytesIO(data))
                 for segment in (
-                    Subtitle.parse(box.data, Subtitle.Codec.TimedTextMarkupLang) for box in boxes if box.type == b"mdat"
+                    Subtitle.parse(payload, Subtitle.Codec.TimedTextMarkupLang)
+                    for payload in Subtitle.mdat_payloads(data)
                 ):
                     for lang in segment.get_languages():
                         caption_lists[lang].extend(segment.get_captions(lang))