diff --git a/docs/guide/subtitles.md b/docs/guide/subtitles.md index 4df5b0be..fe734a73 100644 --- a/docs/guide/subtitles.md +++ b/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. diff --git a/tests/tracks/test_subtitle_convert.py b/tests/tracks/test_subtitle_convert.py index 44c3d96d..e361ee1e 100644 --- a/tests/tracks/test_subtitle_convert.py +++ b/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"{template}"), - "representation": etree.fromstring(""), - "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"{template}"), + "representation": etree.fromstring(""), + "segment_durations": [], + } + } + return downloaded_subtitle(tmp_path, monkeypatch, codec, data, Track.Descriptor.DASH, track_data) + + STPP_CUE = ( '' ' 00:00:03.000" in text +def ttml_fragment(*cues: tuple[str, str]) -> str: + body = "".join(f'

Line

' for begin, end in cues) + return f'
{body}
' + + +def test_dash_fttml_with_no_cues_converts(tmp_path, monkeypatch): + template = '' + 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. diff --git a/unshackle/core/manifests/ism.py b/unshackle/core/manifests/ism.py index 1647ef7e..20b97baa 100644 --- a/unshackle/core/manifests/ism.py +++ b/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: diff --git a/unshackle/core/tracks/subtitle.py b/unshackle/core/tracks/subtitle.py index 74b65f74..b031b5d5 100644 --- a/unshackle/core/tracks/subtitle.py +++ b/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))