diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index 814518cff1..9388626c0b 100644 --- a/doc/lib/rs/moq-mux.md +++ b/doc/lib/rs/moq-mux.md @@ -13,10 +13,10 @@ back. This is what `moq import` / `export` and the gateways are built on. | Format | Import | Export | Notes | | --- | --- | --- | --- | -| fMP4 / CMAF | yes | yes | Passthrough as `cmaf` or repackaged as `legacy`. | +| fMP4 / CMAF | yes | yes | Audio/video passthrough as `cmaf` or repackaged as `legacy`; `wvtt` subtitles import as timed WebVTT text, preserving source language and label when present. | | MPEG-TS | yes | yes | H.264/H.265; AAC, MP2, AC-3, E-AC-3, Opus up to 7.1; SCTE-35 and subtitle PIDs as tracks; service tables round-trip. | | FLV / RTMP | yes | yes | Legacy H.264 + AAC + MP3, plus enhanced-RTMP HEVC, AV1, VP9, Opus, AC-3, E-AC-3, and multitrack. | -| Matroska / WebM | yes | yes | | +| Matroska / WebM | yes | yes | Timed SRT and WebVTT subtitles import as WebVTT text renditions, preserving source language and label when present; other subtitle codecs are skipped. | | Annex-B (H.264, H.265) | yes | yes | Parameter sets extracted to the catalog or re-injected per keyframe. | Importers fill the catalog from the bitstream, split groups at keyframes, and diff --git a/quest/m1/README.md b/quest/m1/README.md index cfd91c7596..49b08b2e0f 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -154,7 +154,8 @@ session outcomes, and embedder refusals inside session outcomes. - [Android capture](/quest/m1/mobile-capture-android.md) - Rust captures through NDK/JNI on Android, reusing the existing codecs and setting the catalog rotation - [Time stretch](/quest/m1/watch-audio-time-stretch.md) - js/watch: the audio ring converges by time-stretching instead of skipping or going silent - [Native audio quality](/quest/m1/audio-quality-native.md) - the browser lane's profiles, budgets, and metric schema run against `moq play` on a dummy device -- [Caption import](/quest/m1/captions-import.md) - fMP4 and MKV subtitle tracks import as text renditions instead of erroring or being dropped +- [TTML caption import](/quest/m1/captions-ttml.md) - fMP4 `stpp` tracks within a supported TTML/IMSC profile import as `ttml` renditions on the shared clock instead of being skipped +- [fMP4 WebVTT ctim and vtta](/quest/m1/wvtt-ctim-vtta.md) - `wvtt` samples with `ctim` cue times or `vtta` comments import instead of failing - [Encoder colour](/quest/m1/color-model.md) - every moq-video encode path signals the colour its output actually has, or refuses instead of mislabelling - [T-STD TS export](/quest/m1/tstd/README.md) - `moq export ts` is a proper remux that passes the T-STD buffer model, starting with a fixed `--delay` - [ATSC AC-3](/quest/m1/ts-atsc-ac3.md) - `export ts` carries 48 kHz ATSC AC-3 at every A/52 rate up to 640 kb/s instead of aborting from 384 kb/s up diff --git a/quest/m1/captions-import.md b/quest/m1/captions-import.md deleted file mode 100644 index 87c1f72765..0000000000 --- a/quest/m1/captions-import.md +++ /dev/null @@ -1,31 +0,0 @@ -# [M] Caption import - -## Goal - -Importing a file with subtitles yields a text rendition instead of an error or -a dropped track. A broadcast with captions currently loses them at every -container boundary. - -## Plan - -Two importers refuse timed text today: - -- fMP4 returns `Error::UnsupportedSubtitle` for an `sbtl` handler, so a file - with a subtitle track fails the whole import rather than importing the media - it does understand. -- MKV logs and drops them, alongside unsupported codecs. - -Import both as `text` renditions. Map each source track to a `TextFormat` -(`wvtt` sample entries and MKV's `S_TEXT/WEBVTT` are WebVTT; `S_TEXT/UTF8` is -`utf8`; TTML is `ttml`) and carry language and track name onto `lang` and -`label`. A format nothing maps to should degrade to skipping that one track -with a warning, never to failing the import, which is the behavior an `sbtl` -handler gets wrong today in the other direction. - -Cue timing comes from the sample's own presentation time on the shared media -clock, which is what a text frame's timestamp means. Text is sparse, so a -subtitle track must not gate the import's progress on an open group waiting -for a cue that may be minutes away. - -Cover a file with subtitles and no supported text format, a file with several -subtitle languages, and a file whose subtitle track is empty. diff --git a/quest/m1/captions-ttml.md b/quest/m1/captions-ttml.md new file mode 100644 index 0000000000..ebdbec4848 --- /dev/null +++ b/quest/m1/captions-ttml.md @@ -0,0 +1,24 @@ +# [M] TTML caption import + +## Goal + +Importing an fMP4 `stpp` track within a supported TTML/IMSC profile yields a `ttml` text +rendition whose cues stay on the shared media clock. Today those XML tracks are +skipped with a warning while SRT and WebVTT import as `vtt`. + +## Plan + +Copying XML bytes is not enough. MP4 `stpp` names an XML namespace, which is not +necessarily TTML, and the importer shifts the source clock, so every TTML time +expression must be rewritten onto the catalog clock. + +- Support a bounded TTML/IMSC timing profile: validate the namespace and refuse + anything outside the profile instead of relabelling arbitrary XML as TTML. +- Convert every time expression, including nested `begin`/`end`/`dur` and + `ttp:timeBase` variants the profile admits, after the clock shift. +- Keep skipping XML tracks the profile does not cover, one track at a time. + +Matroska has no registered TTML codec ID (the old `S_TEXT/TTML` claim was +removed), so confirm a real mapping before adding one. + +Cover a namespace mismatch, nested timing, and a shifted source clock. diff --git a/quest/m1/wvtt-ctim-vtta.md b/quest/m1/wvtt-ctim-vtta.md new file mode 100644 index 0000000000..70929b635e --- /dev/null +++ b/quest/m1/wvtt-ctim-vtta.md @@ -0,0 +1,23 @@ +# [S] fMP4 WebVTT ctim and vtta + +## Goal + +An fMP4 `wvtt` sample whose cue carries a `ctim` CueTimeBox, or which +interleaves `vtta` additional-text boxes, imports as WebVTT text instead of +failing the import. Today both are refused with `InvalidSubtitle`, as is any +inline timestamp in a cue payload. + +## Plan + +Both boxes are defined by ISO/IEC 14496-30 ยง6.6. + +- `ctim` records the cue's start on the original WebVTT timeline. Rebase each + inline `` timestamp in the payload by `inline - ctim` onto the + sample start, reusing the Matroska rebase in `rs/moq-mux/src/container/text.rs`. + Refuse inline timestamps that arrive without `ctim`, since their origin is + unknown. +- `vtta` holds text between cues (such as `NOTE` blocks). Serialize it in sample + order, validated like the Matroska cue-comment path. + +Cover an inline timestamp with `ctim` on a shifted clock, one without `ctim`, +and a sample mixing `vtta` and `vttc`. diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index 3d765ecd02..a0f07f49e6 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -26,6 +26,8 @@ use crate::catalog::Estimator; /// - AAC (MP4A) /// - Opus /// - FLAC +/// +/// **Text:** WebVTT (`wvtt`), reconstructed as timed text; unsupported subtitle codecs are skipped. pub struct Import { /// The broadcast being produced broadcast: moq_net::broadcast::Producer, @@ -47,6 +49,7 @@ pub struct Import { // A lookup to tracks in the broadcast tracks: HashMap>, + text: HashMap, // Track ids skipped by `select`, so their moof fragments are ignored rather // than treated as referencing an unknown track. @@ -152,6 +155,7 @@ impl Import { initial_reservation: Some(reserved), select: None, tracks: HashMap::default(), + text: HashMap::default(), skipped: HashSet::default(), moov: None, moof: None, @@ -286,6 +290,15 @@ impl Import { let handler = &trak.mdia.hdlr.handler; let suffix = ".m4s"; + if matches!(handler.as_ref(), b"sbtl" | b"subt" | b"text") { + if self.init_text(trak, &moov, &reserved)? { + continue; + } + tracing::warn!(track_id, handler = %handler, "unsupported fMP4 subtitle encoding, skipping"); + self.skipped.insert(track_id); + continue; + } + let kind = Kind::from_handler(*handler)?; // Drop tracks whose role isn't selected before minting or publishing them; their @@ -347,7 +360,7 @@ impl Import { // The reservation stays held until the first fragment anchors the clock, unless no track // was declared: then no fragment ever will. - if self.tracks.is_empty() { + if self.tracks.is_empty() && self.text.is_empty() { self.initial_reservation = None; } @@ -356,6 +369,39 @@ impl Import { Ok(()) } + fn init_text(&mut self, trak: &Trak, moov: &Moov, reserved: &crate::catalog::Reserved) -> Result { + let [mp4_atom::Codec::Wvtt(codec)] = trak.mdia.minf.stbl.stsd.codecs.as_slice() else { + return Ok(false); + }; + if codec.plaintext.data_reference_index != 1 { + return Err(Error::InvalidSubtitle("external subtitle data reference").into()); + } + let header = crate::container::text::header(&codec.config.config).map_err(Error::InvalidSubtitle)?; + let mut config = hang::catalog::TextConfig::new(hang::catalog::TextFormat::Vtt); + config.role = crate::container::text::role(&header); + config.lang = Some(trak.mdia.mdhd.language.clone()); + // The picker label is the HandlerBox name; `vlab` is an opaque source identifier. + config.label = (!trak.mdia.hdlr.name.is_empty()).then(|| trak.mdia.hdlr.name.clone()); + let wire = crate::catalog::hang::Container::new(&config.container, crate::container::Kind::Data)?; + let track = self.broadcast.create_track( + self.broadcast.unique_name(".mp4-t"), + self.catalog.track_info(hang::catalog::PRIORITY.text), + )?; + let producer = reserved.text(track, wire, config)?; + self.text.insert( + trak.tkhd.track_id, + super::text::Track { + producer, + header, + defaults: sample::Defaults::from_moov(moov, trak.tkhd.track_id), + timescale: moq_net::Timescale::new(trak.mdia.mdhd.timescale as u64)?, + end_decode_time: None, + discontinuity: false, + }, + ); + Ok(true) + } + fn container(&self, trak: &Trak, moov: &Moov) -> Result { // Build a single-track init segment (ftyp+moov) for this track. { @@ -674,6 +720,19 @@ impl Import { // Loop over all of the traf boxes in the moof. for traf in &moof.traf { let track_id = traf.tfhd.track_id; + if let Some(track) = self.text.get_mut(&track_id) { + track.extract( + super::text::Fragment { + traf, + data: &mdat.data, + moof_size, + header_size, + }, + &mut self.timebase, + )?; + self.initial_reservation = None; + continue; + } let track = match self.tracks.get_mut(&track_id) { Some(track) => track, // A fragment for a track `select` dropped: ignore it. @@ -960,6 +1019,9 @@ impl Import { impl Import { /// Finish all tracks, flushing current groups. pub fn finish(&mut self) -> Result<()> { + for track in self.text.values_mut() { + track.producer.finish()?; + } // No frame follows to anchor the clock, so publish the declared track set now. self.initial_reservation = None; for track in self.tracks.values_mut() { @@ -979,6 +1041,9 @@ impl Import { /// Abort all tracks with `err` instead of finishing, so subscribers see the real /// cause rather than [`moq_net::Error::Dropped`]. Consumes the importer. pub fn abort(mut self, err: moq_net::Error) { + for track in std::mem::take(&mut self.text).into_values() { + track.producer.abort(err.clone()); + } // Dropping each track's rendition retires its catalog entry. for mut track in std::mem::take(&mut self.tracks).into_values() { if let Some(g) = track.group.take() { @@ -993,8 +1058,13 @@ impl Import { /// Broadcast-wide: every track inside this fMP4 import advances together; per-track /// control is intentionally not exposed. Skipping sequences is not a new timeline, so the /// next fragment must still advance past the last decode time; see - /// [`discontinuity`](Self::discontinuity). + /// [`discontinuity`](Self::discontinuity). Text spends a group per cue, so a text track that + /// is already past `sequence` continues from its own next group instead. pub fn seek(&mut self, sequence: u64) -> Result<()> { + for track in self.text.values_mut() { + let next = track.producer.track().latest().map_or(0, |latest| latest + 1); + track.producer.seek(sequence.max(next))?; + } for track in self.tracks.values_mut() { track.estimator.cut(None); track.publish_estimate()?; @@ -1012,6 +1082,10 @@ impl Import { /// The source signalled a new timeline (an HLS `EXT-X-DISCONTINUITY`), so the next /// fragment's decode time is not compared with the last one on each track. pub fn discontinuity(&mut self) { + for track in self.text.values_mut() { + track.end_decode_time = None; + track.discontinuity = true; + } for track in self.tracks.values_mut() { track.last_decode_time = None; } diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index f553823f9b..e1a6b0b13f 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -1689,6 +1689,244 @@ async fn imports_the_sample_defaults_fixtures() { } } +#[test] +fn an_unknown_subtitle_track_does_not_refuse_supported_media() { + use mp4_atom::{DecodeMaybe, Encode}; + let mut cursor = std::io::Cursor::new(include_bytes!("test_data/bbb.mp4")); + let mut input = Vec::new(); + while let Some(atom) = mp4_atom::Any::decode_maybe(&mut cursor).unwrap() { + if let mp4_atom::Any::Moov(mut moov) = atom { + let mut subtitle = moov.trak[0].clone(); + subtitle.tkhd.track_id = 99; + subtitle.mdia.hdlr.handler = b"sbtl".into(); + moov.trak.push(subtitle); + moov.encode(&mut input).unwrap(); + break; + } else { + atom.encode(&mut input).unwrap(); + } + } + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + importer.decode(&input).unwrap(); + importer.finish().unwrap(); + assert!(!catalog.snapshot().video.renditions.is_empty()); + assert!(catalog.snapshot().text.renditions.is_empty()); +} + +fn webvtt_init(header: &str) -> Vec { + let (_, moov) = decode_init(include_bytes!("test_data/bbb.mp4")); + let mut trak = moov.trak[0].clone(); + trak.tkhd.track_id = 1; + trak.mdia.mdhd.timescale = 1000; + trak.mdia.mdhd.language = "eng".into(); + trak.mdia.hdlr.handler = b"text".into(); + trak.mdia.hdlr.name = "English captions".into(); + trak.mdia.minf.stbl.stsd.codecs = vec![mp4_atom::Codec::Wvtt(mp4_atom::Wvtt { + plaintext: mp4_atom::PlainText { + data_reference_index: 1, + }, + config: mp4_atom::VttC { config: header.into() }, + label: Some(mp4_atom::Vlab { + source_label: "urn:example:source".into(), + }), + btrt: None, + })]; + audio_init(trak) +} + +/// One fragment carrying `samples` back to back, each lasting `duration` ms from `decode_time`. +fn webvtt_fragment(decode_time: u64, samples: &[Vec], duration: u32) -> Vec { + let moof = mp4_atom::Moof { + mfhd: mp4_atom::Mfhd { sequence_number: 1 }, + traf: vec![mp4_atom::Traf { + tfhd: mp4_atom::Tfhd { + track_id: 1, + default_sample_duration: Some(duration), + ..Default::default() + }, + tfdt: Some(mp4_atom::Tfdt { + base_media_decode_time: decode_time, + }), + trun: vec![mp4_atom::Trun { + data_offset: None, + entries: samples + .iter() + .map(|sample| mp4_atom::TrunEntry { + size: Some(sample.len() as u32), + ..Default::default() + }) + .collect(), + }], + ..Default::default() + }], + }; + let mut output = Vec::new(); + moof.encode(&mut output).unwrap(); + mp4_atom::Mdat { data: samples.concat() }.encode(&mut output).unwrap(); + output +} + +fn webvtt_input(payload: Vec, duration: u32) -> Vec { + let mut input = webvtt_init("WEBVTT"); + input.extend(webvtt_fragment(250, &[payload], duration)); + input +} + +fn subtitle_box(kind: &[u8; 4], payload: &[u8]) -> Vec { + let mut bytes = Vec::new(); + bytes.extend_from_slice(&((8 + payload.len()) as u32).to_be_bytes()); + bytes.extend_from_slice(kind); + bytes.extend_from_slice(payload); + bytes +} + +#[tokio::test(start_paused = true)] +async fn webvtt_samples_become_timed_text_with_source_metadata() { + let mut cue = subtitle_box(b"iden", b"cue-1"); + cue.extend(subtitle_box(b"sttg", b"align:start")); + cue.extend(subtitle_box(b"payl", b"Hello")); + let payload = subtitle_box(b"vttc", &cue); + let input = webvtt_input(payload, 1250); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + importer.decode(&input).unwrap(); + importer.finish().unwrap(); + let snapshot = catalog.snapshot(); + assert_eq!(snapshot.text.renditions.len(), 1); + let (name, config) = snapshot.text.renditions.iter().next().unwrap(); + assert_eq!(config.lang.as_deref(), Some("eng")); + assert_eq!( + config.label.as_deref(), + Some("English captions"), + "hdlr name, not the vlab source" + ); + assert_eq!(config.role, hang::catalog::TextRole::Subtitle); + let track = consumer + .track(name) + .unwrap() + .subscribe(moq_net::track::Subscription::default()) + .await + .unwrap(); + let mut reader = crate::container::Consumer::new( + track, + crate::catalog::hang::Container::Legacy(crate::container::Kind::Data), + ); + let frame = reader.read().await.unwrap().unwrap(); + assert_eq!(frame.timestamp.as_micros(), 250_000); + assert_eq!( + &frame.payload[..], + b"WEBVTT\n\ncue-1\n00:00:00.250 --> 00:00:01.500 align:start\nHello\n\n" + ); +} + +#[test] +fn malformed_webvtt_samples_are_refused() { + let payl = subtitle_box(b"payl", b"hello"); + let mut duplicate = payl.clone(); + duplicate.extend_from_slice(&payl); + for (payload, duration) in [ + (Vec::new(), 1250), + (vec![0, 0, 0, 4], 1250), + (subtitle_box(b"vttc", &[]), 1250), + (subtitle_box(b"vttc", &duplicate), 1250), + (subtitle_box(b"vttc", &subtitle_box(b"payl", b"\xff")), 1250), + (subtitle_box(b"vttc", &payl), 0), + // Inline timestamps need `ctim` to anchor them. + (subtitle_box(b"vttc", &subtitle_box(b"payl", b"a <00:01.000>b")), 1250), + ] { + let input = webvtt_input(payload, duration); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + assert!(importer.decode(&input).is_err()); + } +} + +#[test] +fn empty_and_multiple_webvtt_cues_are_supported() { + let cue = subtitle_box(b"vttc", &subtitle_box(b"payl", b"hello")); + let mut multiple = cue.clone(); + multiple.extend(cue); + for payload in [subtitle_box(b"vtte", &[]), multiple] { + let input = webvtt_input(payload, 1250); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + importer.decode(&input).unwrap(); + importer.finish().unwrap(); + assert_eq!(catalog.snapshot().text.renditions.len(), 1); + } +} + +/// The WebVTT `Kind` header sets the role, and lone CR line terminators are accepted. +#[tokio::test(start_paused = true)] +async fn webvtt_kind_and_lone_cr_terminators() { + let mut input = webvtt_init("WEBVTT\rKind: captions\r"); + input.extend(webvtt_fragment( + 250, + &[subtitle_box(b"vttc", &subtitle_box(b"payl", b"Hello\rworld"))], + 1250, + )); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + importer.decode(&input).unwrap(); + importer.finish().unwrap(); + let snapshot = catalog.snapshot(); + let (name, config) = snapshot.text.renditions.iter().next().unwrap(); + assert_eq!(config.role, hang::catalog::TextRole::Caption); + let track = consumer.track(name).unwrap().subscribe(None).await.unwrap(); + let mut reader = crate::container::Consumer::new( + track, + crate::catalog::hang::Container::Legacy(crate::container::Kind::Data), + ); + let frame = reader.read().await.unwrap().unwrap(); + assert_eq!( + &frame.payload[..], + b"WEBVTT\nKind: captions\n\n00:00:00.250 --> 00:00:01.500\nHello\nworld\n\n" + ); +} + +/// Text closes a group per cue, so a segment seek must not rewind onto a cue group already written. +#[test] +fn webvtt_seek_after_several_cues_does_not_reuse_a_group() { + let cue = subtitle_box(b"vttc", &subtitle_box(b"payl", b"hello")); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + importer.decode(&webvtt_init("WEBVTT")).unwrap(); + importer.seek(100).unwrap(); + importer + .decode(&webvtt_fragment(0, &[cue.clone(), cue.clone(), cue.clone()], 1000)) + .unwrap(); + // Segment 101 was missed; groups 100..=102 already hold the three cues. + importer.seek(102).unwrap(); + importer.decode(&webvtt_fragment(4000, &[cue], 1000)).unwrap(); + importer.finish().unwrap(); +} + +/// A text fragment may not start inside the decode span of the previous fragment's samples. +#[test] +fn webvtt_fragment_overlapping_previous_samples_is_refused() { + let cue = subtitle_box(b"vttc", &subtitle_box(b"payl", b"hello")); + for (start, valid) in [(1500, false), (2000, true)] { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + importer.decode(&webvtt_init("WEBVTT")).unwrap(); + importer + .decode(&webvtt_fragment(0, &[cue.clone(), cue.clone()], 1000)) + .unwrap(); + let result = importer.decode(&webvtt_fragment(start, std::slice::from_ref(&cue), 1000)); + assert_eq!(result.is_ok(), valid, "start {start}: {result:?}"); + } +} + #[test] #[tracing_test::traced_test] fn emsg_metadata_drops_warn_once() { @@ -1730,3 +1968,34 @@ fn emsg_metadata_drops_warn_once() { } }); } + +/// A cue shorter than a millisecond keeps a positive WebVTT duration. +#[tokio::test(start_paused = true)] +async fn webvtt_sub_millisecond_cue_stays_positive() { + // A 1 MHz track, so the cue can last 400 us. + let (ftyp, mut moov) = decode_init(&webvtt_init("WEBVTT")); + moov.trak[0].mdia.mdhd.timescale = 1_000_000; + let mut init = Vec::new(); + ftyp.encode(&mut init).unwrap(); + moov.encode(&mut init).unwrap(); + init.extend(webvtt_fragment( + 250_200, + &[subtitle_box(b"vttc", &subtitle_box(b"payl", b"hi"))], + 400, + )); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + importer.decode(&init).unwrap(); + importer.finish().unwrap(); + let snapshot = catalog.snapshot(); + let name = snapshot.text.renditions.keys().next().unwrap(); + let track = consumer.track(name).unwrap().subscribe(None).await.unwrap(); + let mut reader = crate::container::Consumer::new( + track, + crate::catalog::hang::Container::Legacy(crate::container::Kind::Data), + ); + let frame = reader.read().await.unwrap().unwrap(); + assert_eq!(&frame.payload[..], b"WEBVTT\n\n00:00:00.250 --> 00:00:00.251\nhi\n\n"); +} diff --git a/rs/moq-mux/src/container/fmp4/mod.rs b/rs/moq-mux/src/container/fmp4/mod.rs index 35ed31917e..e28e27ced9 100644 --- a/rs/moq-mux/src/container/fmp4/mod.rs +++ b/rs/moq-mux/src/container/fmp4/mod.rs @@ -12,6 +12,7 @@ mod fragmenter; mod import; mod muxer; mod sample; +mod text; pub use export::*; pub use fragmenter::*; @@ -40,6 +41,10 @@ use crate::container::{Container, Frame}; #[derive(Debug, Clone, thiserror::Error)] #[non_exhaustive] pub enum Error { + /// A subtitle sample cannot be reconstructed with valid text and timing. + #[error("invalid subtitle: {0}")] + InvalidSubtitle(&'static str), + #[error("mp4: {0}")] Mp4(std::sync::Arc), diff --git a/rs/moq-mux/src/container/fmp4/text.rs b/rs/moq-mux/src/container/fmp4/text.rs new file mode 100644 index 0000000000..262ea2830f --- /dev/null +++ b/rs/moq-mux/src/container/fmp4/text.rs @@ -0,0 +1,147 @@ +use moq_net::{Timescale, Timestamp}; +use mp4_atom::{Any, DecodeMaybe}; + +use super::{Error, sample}; +use crate::container::text::{self, Cue}; + +pub(super) struct Track { + pub producer: crate::container::Producer, + pub header: String, + pub defaults: sample::Defaults, + pub timescale: Timescale, + /// The exclusive decode end of the previous fragment; the next one may not start before it. + pub end_decode_time: Option, + pub discontinuity: bool, +} + +pub(super) struct Fragment<'a> { + pub traf: &'a mp4_atom::Traf, + pub data: &'a [u8], + pub moof_size: usize, + pub header_size: usize, +} + +impl Track { + pub fn extract( + &mut self, + input: Fragment<'_>, + timebase: &mut crate::catalog::Timebase, + ) -> crate::Result<()> { + let traf = input.traf; + let mut dts = traf.tfdt.as_ref().ok_or(Error::MissingTfdt)?.base_media_decode_time; + if self.end_decode_time.is_some_and(|end| dts < end) { + return Err(Error::InvalidSubtitle("WebVTT fragment overlaps the previous one").into()); + } + if traf.trun.is_empty() { + return Err(Error::MissingTrun.into()); + } + if self.discontinuity { + self.producer.discontinuity()?; + self.discontinuity = false; + } + let base: usize = traf + .tfhd + .base_data_offset + .unwrap_or_default() + .try_into() + .map_err(|_| Error::InvalidDataOffset)?; + let mut offset = base; + for trun in &traf.trun { + if let Some(value) = trun.data_offset { + let value: usize = value.try_into().map_err(|_| Error::InvalidDataOffset)?; + offset = value + .checked_sub(input.moof_size) + .and_then(|value| value.checked_sub(input.header_size)) + .and_then(|value| value.checked_add(base)) + .ok_or(Error::InvalidDataOffset)?; + } + for entry in &trun.entries { + let sample = self.defaults.resolve(&traf.tfhd, entry); + let duration = sample.duration.ok_or(Error::MissingSampleDuration)?; + let pts = dts + .checked_add_signed(entry.cts.unwrap_or_default() as i64) + .ok_or(Error::PtsOverflow)?; + let timestamp = timebase.shift(Timestamp::new(pts, self.timescale)?)?; + let duration = Timestamp::new(duration as u64, self.timescale)?; + let end = timestamp.checked_add(duration)?; + let next = offset + .checked_add(sample.size as usize) + .ok_or(Error::InvalidDataOffset)?; + let data = input.data.get(offset..next).ok_or(Error::InvalidDataOffset)?; + let payload = self.sample(data, timestamp, end)?; + self.producer.write(crate::container::Frame { + timestamp, + duration: Some(duration), + keyframe: true, + payload, + })?; + self.producer.cut(None)?; + offset = next; + dts = dts.checked_add(duration.value()).ok_or(Error::PtsOverflow)?; + } + } + self.end_decode_time = Some(dts); + Ok(()) + } + + fn sample(&self, mut data: &[u8], start: Timestamp, end: Timestamp) -> crate::Result { + if data.is_empty() { + return Err(Error::InvalidSubtitle("empty WebVTT sample").into()); + } + let mut output = self.header.clone(); + while !data.is_empty() { + let atom = Any::decode_maybe(&mut data) + .map_err(Error::from)? + .ok_or(Error::InvalidSubtitle("truncated WebVTT sample box"))?; + let Any::Unknown(kind, body) = atom else { + return Err(Error::InvalidSubtitle("unsupported WebVTT sample box").into()); + }; + match kind.as_ref() { + b"vtte" if body.is_empty() => {} + b"vttc" => { + let mut body = body.as_slice(); + let mut payload = None; + let mut id = None; + let mut settings = None; + while !body.is_empty() { + let atom = Any::decode_maybe(&mut body) + .map_err(Error::from)? + .ok_or(Error::InvalidSubtitle("truncated WebVTT cue box"))?; + let Any::Unknown(kind, value) = atom else { + return Err(Error::InvalidSubtitle("unsupported WebVTT cue box").into()); + }; + let field = match kind.as_ref() { + b"payl" => &mut payload, + b"iden" => &mut id, + b"sttg" => &mut settings, + _ => return Err(Error::InvalidSubtitle("unsupported WebVTT cue box").into()), + }; + if field.is_some() { + return Err(Error::InvalidSubtitle("duplicate WebVTT cue box").into()); + } + *field = + Some(String::from_utf8(value).map_err(|_| Error::InvalidSubtitle("invalid WebVTT UTF-8"))?); + } + let payload = payload.ok_or(Error::InvalidSubtitle("missing WebVTT cue payload"))?; + // Inline timestamps sit on the source timeline, which only `ctim` anchors. + if text::has_timestamp(&payload) { + return Err(Error::InvalidSubtitle("inline WebVTT timestamps are unsupported").into()); + } + text::append( + &mut output, + Cue { + start, + end, + body: &payload, + id: id.as_deref().unwrap_or_default(), + settings: settings.as_deref().unwrap_or_default(), + }, + ) + .map_err(Error::InvalidSubtitle)?; + } + _ => return Err(Error::InvalidSubtitle("unsupported WebVTT sample box").into()), + } + } + Ok(output.into()) + } +} diff --git a/rs/moq-mux/src/container/mkv/import.rs b/rs/moq-mux/src/container/mkv/import.rs index 5c0333c8e3..bdf1094660 100644 --- a/rs/moq-mux/src/container/mkv/import.rs +++ b/rs/moq-mux/src/container/mkv/import.rs @@ -37,9 +37,11 @@ const DEFAULT_TIMESTAMP_SCALE_NS: u64 = 1_000_000; /// - FLAC (`A_FLAC`) /// - MP3 (`A_MPEG/L3`) /// +/// **Text:** timed SRT (`S_TEXT/UTF8`) and WebVTT (`S_TEXT/WEBVTT`), emitted as WebVTT. +/// /// Laced blocks require a positive TrackEntry DefaultDuration to timestamp each frame. /// -/// Unsupported codecs (e.g. Vorbis, AC3, subtitles) are logged and dropped. +/// Unsupported track roles and subtitle encodings are logged and dropped. pub struct Import { broadcast: moq_net::broadcast::Producer, catalog: crate::catalog::Producer, @@ -77,6 +79,7 @@ pub struct Import { enum TrackKind { Video, Audio, + Text, } impl TrackKind { @@ -86,6 +89,7 @@ impl TrackKind { match self { Self::Video => hang::catalog::PRIORITY.video, Self::Audio => hang::catalog::PRIORITY.audio, + Self::Text => hang::catalog::PRIORITY.text, } } } @@ -93,13 +97,26 @@ impl TrackKind { struct MkvTrack { default_duration: Option, kind: TrackKind, + text_header: Option, track: Media, group: Option, } +/// BlockGroup fields that only subtitle tracks consume. +#[derive(Default)] +struct Cue<'a> { + /// BlockDuration, in Matroska ticks. + duration: Option, + /// WebVTT settings, identifier and comments (BlockAddID 1). + addition: Option<&'a [u8]>, + /// Whether the BlockGroup replaces the codec initialization (CodecState). + codec_state: bool, +} + enum Media { Video(crate::container::Producer), Audio(crate::container::Producer), + Text(crate::container::Producer), } impl Media { @@ -107,6 +124,7 @@ impl Media { match self { Self::Video(track) => track.write(frame), Self::Audio(track) => track.write(frame), + Self::Text(track) => track.write(frame), } } @@ -114,6 +132,7 @@ impl Media { match self { Self::Video(track) => track.cut(timestamp), Self::Audio(track) => track.cut(timestamp), + Self::Text(track) => track.cut(timestamp), } } @@ -121,6 +140,7 @@ impl Media { match self { Self::Video(track) => track.seek(sequence), Self::Audio(track) => track.seek(sequence), + Self::Text(track) => track.seek(sequence), } } @@ -128,6 +148,7 @@ impl Media { match self { Self::Video(track) => track.finish(), Self::Audio(track) => track.finish(), + Self::Text(track) => track.finish(), } } @@ -135,6 +156,7 @@ impl Media { match self { Self::Video(track) => track.abort(err), Self::Audio(track) => track.abort(err), + Self::Text(track) => track.abort(err), } } } @@ -289,7 +311,7 @@ impl Import { } MatroskaSpec::SimpleBlock(ref data) => { let sb = SimpleBlock::try_from(data.as_slice()).map_err(|_| Error::InvalidSimpleBlock)?; - self.handle_block(&sb, sb.keyframe)?; + self.handle_block(&sb, sb.keyframe, Cue::default())?; } MatroskaSpec::BlockGroup(Master::Full(children)) => { self.handle_block_group(&children)?; @@ -317,6 +339,9 @@ impl Import { if let MatroskaSpec::TrackEntry(Master::Full(children)) = entry && let Err(e) = self.add_track(reserved, children) { + if matches!(e, crate::Error::Mkv(Error::InvalidSubtitle(_))) { + return Err(e); + } tracing::warn!(error = ?e, "skipping MKV track"); } } @@ -324,13 +349,16 @@ impl Import { } fn add_track(&mut self, reserved: &crate::catalog::Reserved, children: Vec) -> Result<()> { - let mut default_duration = None; let mut track_number: Option = None; let mut track_type: Option = None; let mut codec_id: Option = None; let mut codec_private: Option = None; let mut audio_children: Option> = None; let mut video_children: Option> = None; + let mut language = None; + let mut language_ietf = None; + let mut name = None; + let mut default_duration = None; for c in children { match c { @@ -340,6 +368,9 @@ impl Import { MatroskaSpec::TrackNumber(v) => track_number = Some(v), MatroskaSpec::TrackType(v) => track_type = Some(v), MatroskaSpec::CodecID(v) => codec_id = Some(v), + MatroskaSpec::Language(v) => language = Some(v), + MatroskaSpec::LanguageIETF(v) => language_ietf = Some(v), + MatroskaSpec::Name(v) => name = Some(v), MatroskaSpec::CodecPrivate(v) => codec_private = Some(Bytes::from(v)), MatroskaSpec::Audio(Master::Full(v)) => audio_children = Some(v), MatroskaSpec::Video(Master::Full(v)) => video_children = Some(v), @@ -351,10 +382,15 @@ impl Import { let track_type = track_type.ok_or(Error::MissingTrackType)?; let codec_id = codec_id.ok_or(Error::MissingCodecId)?; - // Matroska TrackType: 1 = video, 2 = audio. + // Matroska TrackType: 1 = video, 2 = audio, 17 = subtitles. let (kind, suffix) = match track_type { 1 => (TrackKind::Video, ".mkv-v"), 2 => (TrackKind::Audio, ".mkv-a"), + 17 if matches!(codec_id.as_str(), "S_TEXT/UTF8" | "S_TEXT/WEBVTT") => (TrackKind::Text, ".mkv-t"), + 17 => { + tracing::warn!(codec_id, "unsupported MKV subtitle encoding, skipping"); + return Ok(()); + } other => { tracing::warn!(track_type = other, codec_id, "unsupported MKV track type, skipping"); return Ok(()); @@ -373,9 +409,30 @@ impl Import { match kind { TrackKind::Video => crate::container::Kind::Video, TrackKind::Audio => crate::container::Kind::Audio, + TrackKind::Text => crate::container::Kind::Data, }, )?; + let text_header = if codec_id == "S_TEXT/WEBVTT" { + let private = codec_private + .as_deref() + .ok_or(Error::InvalidSubtitle("missing WebVTT codec private"))?; + let header = + std::str::from_utf8(private).map_err(|_| Error::InvalidSubtitle("invalid WebVTT header UTF-8"))?; + Some(crate::container::text::header(header).map_err(Error::InvalidSubtitle)?) + } else { + None + }; let media = match kind { + TrackKind::Text => { + let mut config = hang::catalog::TextConfig::new(hang::catalog::TextFormat::Vtt); + if let Some(header) = &text_header { + config.role = crate::container::text::role(header); + } + config.lang = language_ietf.or(language); + config.label = name; + config.container = self.container.clone(); + Media::Text(reserved.text(track, wire, config)?) + } TrackKind::Video => { let mut config = build_video_config(&codec_id, codec_private.as_ref(), video_children.as_deref())?; config.container = self.container.clone(); @@ -393,6 +450,7 @@ impl Import { MkvTrack { default_duration, kind, + text_header, track: media, group: None, }, @@ -404,11 +462,37 @@ impl Import { fn handle_block_group(&mut self, children: &[MatroskaSpec]) -> Result<()> { let mut block_data: Option<&[u8]> = None; let mut has_reference = false; + let mut duration = None; + let mut addition = None; + let mut codec_state = false; for c in children { match c { MatroskaSpec::Block(data) => block_data = Some(data.as_slice()), MatroskaSpec::ReferenceBlock(_) => has_reference = true, + MatroskaSpec::BlockAdditions(Master::Full(entries)) => { + for entry in entries { + if let MatroskaSpec::BlockMore(Master::Full(fields)) = entry { + let mut id = 1; + let mut data = None; + for field in fields { + match field { + MatroskaSpec::BlockAddID(value) => id = *value, + MatroskaSpec::BlockAdditional(value) => data = Some(value.as_slice()), + _ => {} + } + } + if id == 1 { + if addition.is_some() { + return Err(Error::InvalidSubtitle("duplicate cue additions").into()); + } + addition = data; + } + } + } + } + MatroskaSpec::BlockDuration(value) => duration = Some(*value), + MatroskaSpec::CodecState(_) => codec_state = true, _ => {} } } @@ -423,10 +507,18 @@ impl Import { let parsed = SimpleBlock::try_from(data).map_err(|_| Error::InvalidBlock)?; let keyframe = !has_reference; - self.handle_block(&parsed, keyframe) + self.handle_block( + &parsed, + keyframe, + Cue { + duration, + addition, + codec_state, + }, + ) } - fn handle_block(&mut self, block: &SimpleBlock<'_>, keyframe: bool) -> Result<()> { + fn handle_block(&mut self, block: &SimpleBlock<'_>, keyframe: bool, cue: Cue<'_>) -> Result<()> { let Some(track) = self.tracks.get_mut(&block.track) else { // Unknown or skipped track. return Ok(()); @@ -442,6 +534,45 @@ impl Import { let pts_ns = (block_ticks as u64) .checked_mul(self.timestamp_scale_ns) .ok_or(Error::TimestampOverflow)?; + if track.kind == TrackKind::Text { + if block.lacing.is_some() { + return Err(Error::InvalidSubtitle("laced subtitles are unsupported").into()); + } + // A replaced WebVTT header would otherwise leave stale global blocks on every later cue. + if cue.codec_state { + return Err(Error::InvalidSubtitle("subtitle CodecState updates are unsupported").into()); + } + let duration = match cue.duration { + Some(ticks) => ticks + .checked_mul(self.timestamp_scale_ns) + .ok_or(Error::TimestampOverflow)?, + None => { + let duration = track + .default_duration + .ok_or(Error::InvalidSubtitle("missing cue duration"))?; + u64::try_from(duration.as_nanos()).map_err(|_| Error::TimestampOverflow)? + } + }; + let duration = Timestamp::from_nanos(duration)?; + let timestamp = self.timebase.shift(Timestamp::from_nanos(pts_ns)?)?; + self.initial_reservation = None; + let payload = cue_payload( + track.text_header.as_deref(), + cue.addition, + block.raw_frame_data(), + timestamp, + duration, + )?; + track.track.write(crate::container::Frame { + timestamp, + payload, + keyframe: true, + duration: Some(duration), + })?; + // Each cue is self-contained, so close its group immediately. + return track.track.cut(None); + } + let frames = split_lacing(block).map_err(anyhow::Error::new)?; let duration = if block.lacing.is_some() { Some( @@ -469,7 +600,7 @@ impl Import { payload: Bytes::copy_from_slice(payload), keyframe: match track.kind { TrackKind::Audio => index == 0, - TrackKind::Video => keyframe, + TrackKind::Video | TrackKind::Text => keyframe, }, duration, }; @@ -485,9 +616,14 @@ impl Import { /// Close the current group on every track and open the next one at `sequence`. /// /// Broadcast-wide: every track inside this MKV import advances together; per-track - /// control is intentionally not exposed. + /// control is intentionally not exposed. Text spends a group per cue, so a text track that is + /// already past `sequence` continues from its own next group instead. pub fn seek(&mut self, sequence: u64) -> Result<()> { for track in self.tracks.values_mut() { + let sequence = match &track.track { + Media::Text(text) => sequence.max(text.track().latest().map_or(0, |latest| latest + 1)), + _ => sequence, + }; track.track.seek(sequence)?; } Ok(()) @@ -731,6 +867,54 @@ fn build_av1_config(codec_private: Option<&Bytes>) -> Result { Ok(config) } +/// Render one Matroska subtitle block as a self-contained WebVTT document. +/// +/// `header` is the WebVTT CodecPrivate; SRT (`S_TEXT/UTF8`) tracks have none and get a bare header. +fn cue_payload( + header: Option<&str>, + addition: Option<&[u8]>, + body: &[u8], + start: Timestamp, + duration: Timestamp, +) -> Result { + let end = start.checked_add(duration)?; + let body = std::str::from_utf8(body).map_err(|_| Error::InvalidSubtitle("invalid UTF-8"))?; + let mut output = header.unwrap_or("WEBVTT\n\n").to_string(); + let mut id = ""; + let mut settings = ""; + if let Some(addition) = addition.filter(|_| header.is_some()) { + let addition = + std::str::from_utf8(addition).map_err(|_| Error::InvalidSubtitle("invalid cue additions UTF-8"))?; + let mut lines = addition.splitn(3, '\n'); + settings = lines.next().unwrap_or_default(); + id = lines.next().unwrap_or_default(); + if let Some(comments) = lines.next().filter(|value| !value.is_empty()) { + if !comments.starts_with("NOTE") || comments.contains("-->") { + return Err(Error::InvalidSubtitle("invalid cue comments").into()); + } + output.push_str(comments.trim_end_matches('\n')); + output.push_str("\n\n"); + } + } + let body = match header { + Some(_) => crate::container::text::rebase(body, start, end).map_err(Error::InvalidSubtitle)?, + // SRT is plain text, where a literal `-->` is valid but would end a WebVTT cue. + None => body.replace("-->", "-->"), + }; + crate::container::text::append( + &mut output, + crate::container::text::Cue { + start, + end, + body: &body, + id, + settings, + }, + ) + .map_err(Error::InvalidSubtitle)?; + Ok(Bytes::from(output)) +} + #[derive(Debug, thiserror::Error)] enum LacingError { #[error("invalid MKV lacing")] diff --git a/rs/moq-mux/src/container/mkv/import_test.rs b/rs/moq-mux/src/container/mkv/import_test.rs index aae2b00bd0..f931a97ff0 100644 --- a/rs/moq-mux/src/container/mkv/import_test.rs +++ b/rs/moq-mux/src/container/mkv/import_test.rs @@ -624,6 +624,238 @@ fn tracks_after_finish_are_refused() { assert!(catalog.snapshot().audio.renditions.is_empty(), "no track was declared"); } +fn text_entry(number: u64, codec: &str, language: &str, name: &str) -> MatroskaSpec { + MatroskaSpec::TrackEntry(Master::Full(vec![ + MatroskaSpec::TrackNumber(number), + MatroskaSpec::TrackUID(number), + MatroskaSpec::TrackType(17), + MatroskaSpec::CodecID(codec.into()), + MatroskaSpec::Language(language.into()), + MatroskaSpec::Name(name.into()), + ])) +} + +#[tokio::test] +async fn subtitle_languages_and_empty_tracks_are_published() { + let data = MkvBuilder::new() + .header("matroska") + .segment_start() + .info(1_000_000) + .tracks(vec![ + text_entry(1, "S_TEXT/UTF8", "en", "English"), + text_entry(2, "S_TEXT/UTF8", "es", "Spanish"), + text_entry(3, "S_TEXT/UTF8", "fr", "French"), + text_entry(4, "S_TEXT/ASS", "de", "Unsupported"), + ]) + .cluster(0, || { + let MatroskaSpec::SimpleBlock(block) = simple_block(1, 250, true, b"Hello") else { + unreachable!() + }; + vec![MatroskaSpec::BlockGroup(Master::Full(vec![ + MatroskaSpec::Block(block), + MatroskaSpec::BlockDuration(1250), + ]))] + }) + .segment_end() + .build(); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + importer.decode(&data).unwrap(); + let snapshot = catalog.snapshot(); + assert_eq!(snapshot.text.renditions.len(), 3); + let (name, english) = snapshot + .text + .renditions + .iter() + .find(|(_, config)| config.lang.as_deref() == Some("en")) + .unwrap(); + assert_eq!(english.label.as_deref(), Some("English")); + assert_eq!(english.format, hang::catalog::TextFormat::Vtt); + let track = consumer + .track(name) + .unwrap() + .subscribe(moq_net::track::Subscription::default()) + .await + .unwrap(); + let mut reader = crate::container::Consumer::new( + track, + crate::catalog::hang::Container::Legacy(crate::container::Kind::Data), + ); + let frame = reader.read().await.unwrap().unwrap(); + assert_eq!( + &frame.payload[..], + b"WEBVTT\n\n00:00:00.250 --> 00:00:01.500\nHello\n\n" + ); + assert_eq!(frame.timestamp.as_micros(), 250_000); + importer.finish().unwrap(); +} + +/// An SRT cue keeps its explicit end, and a plain-text `-->` is escaped for WebVTT. +#[tokio::test(start_paused = true)] +async fn subtitle_duration_survives_as_a_timed_cue() { + let MatroskaSpec::SimpleBlock(block) = simple_block(1, 250, true, b"Go --> there") else { + unreachable!() + }; + let data = MkvBuilder::new() + .header("matroska") + .segment_start() + .info(1_000_000) + .tracks(vec![text_entry(1, "S_TEXT/UTF8", "en", "English")]) + .cluster(0, || { + vec![MatroskaSpec::BlockGroup(Master::Full(vec![ + MatroskaSpec::Block(block), + MatroskaSpec::BlockDuration(1250), + ]))] + }) + .segment_end() + .build(); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + importer.decode(&data).unwrap(); + importer.finish().unwrap(); + let snapshot = catalog.snapshot(); + let (name, config) = snapshot.text.renditions.iter().next().unwrap(); + assert_eq!(config.format, hang::catalog::TextFormat::Vtt); + let track = consumer + .track(name) + .unwrap() + .subscribe(moq_net::track::Subscription::default()) + .await + .unwrap(); + let mut reader = crate::container::Consumer::new( + track, + crate::catalog::hang::Container::Legacy(crate::container::Kind::Data), + ); + let frame = reader.read().await.unwrap().unwrap(); + assert_eq!(frame.timestamp.as_micros(), 250_000); + assert_eq!( + &frame.payload[..], + b"WEBVTT\n\n00:00:00.250 --> 00:00:01.500\nGo --> there\n\n" + ); +} + +#[tokio::test(start_paused = true)] +async fn webvtt_preserves_cue_metadata_and_relative_timestamps() { + let MatroskaSpec::TrackEntry(Master::Full(mut fields)) = text_entry(1, "S_TEXT/WEBVTT", "en", "English") else { + unreachable!() + }; + fields.push(MatroskaSpec::CodecPrivate( + b"WEBVTT\n\nSTYLE\n::cue { color: lime; }\n".to_vec(), + )); + fields.push(MatroskaSpec::LanguageIETF("en-GB".into())); + let MatroskaSpec::SimpleBlock(block) = simple_block(1, 250, true, b"Hello <00:00.500>world") else { + unreachable!() + }; + let data = MkvBuilder::new() + .header("matroska") + .segment_start() + .info(1_000_000) + .tracks(vec![MatroskaSpec::TrackEntry(Master::Full(fields))]) + .cluster(0, || { + vec![MatroskaSpec::BlockGroup(Master::Full(vec![ + MatroskaSpec::Block(block), + MatroskaSpec::BlockDuration(1250), + MatroskaSpec::BlockAdditions(Master::Full(vec![MatroskaSpec::BlockMore(Master::Full(vec![ + MatroskaSpec::BlockAddID(1), + MatroskaSpec::BlockAdditional(b"align:start\ncue-1\nNOTE source".to_vec()), + ]))])), + ]))] + }) + .segment_end() + .build(); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + importer.decode(&data).unwrap(); + importer.finish().unwrap(); + let snapshot = catalog.snapshot(); + let (name, config) = snapshot.text.renditions.iter().next().unwrap(); + assert_eq!(config.lang.as_deref(), Some("en-GB")); + let track = consumer + .track(name) + .unwrap() + .subscribe(moq_net::track::Subscription::default()) + .await + .unwrap(); + let mut reader = crate::container::Consumer::new( + track, + crate::catalog::hang::Container::Legacy(crate::container::Kind::Data), + ); + let frame = reader.read().await.unwrap().unwrap(); + assert_eq!(&frame.payload[..], b"WEBVTT\n\nSTYLE\n::cue { color: lime; }\n\nNOTE source\n\ncue-1\n00:00:00.250 --> 00:00:01.500 align:start\nHello <00:00:00.750>world\n\n"); +} + +#[test] +fn text_requires_valid_utf8_and_explicit_or_default_duration() { + for (default_duration, payload, valid) in [ + (None, b"text".as_slice(), false), + (Some(0), b"text".as_slice(), false), + (Some(1_000_000_000), b"text".as_slice(), true), + (Some(1_000_000_000), b"\xff".as_slice(), false), + ] { + let MatroskaSpec::TrackEntry(Master::Full(mut fields)) = text_entry(1, "S_TEXT/UTF8", "en", "English") else { + unreachable!() + }; + if let Some(duration) = default_duration { + fields.push(MatroskaSpec::DefaultDuration(duration)); + } + let data = MkvBuilder::new() + .header("matroska") + .segment_start() + .info(1_000_000) + .tracks(vec![MatroskaSpec::TrackEntry(Master::Full(fields))]) + .cluster(0, || vec![simple_block(1, 0, true, payload)]) + .segment_end() + .build(); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + let result = importer.decode(&data); + if valid { + result.unwrap(); + } else { + assert!( + matches!(result, Err(crate::Error::Mkv(super::Error::InvalidSubtitle(_)))), + "{result:?}" + ); + } + } +} + +#[test] +fn laced_subtitle_blocks_are_refused() { + let MatroskaSpec::TrackEntry(Master::Full(mut fields)) = text_entry(1, "S_TEXT/UTF8", "en", "English") else { + unreachable!() + }; + fields.push(MatroskaSpec::DefaultDuration(1_000_000_000)); + let MatroskaSpec::SimpleBlock(mut block) = simple_block(1, 0, true, b"\x01ab") else { + unreachable!() + }; + block[3] |= 0x04; // Fixed-size lacing, two one-byte frames. + let data = MkvBuilder::new() + .header("matroska") + .segment_start() + .info(1_000_000) + .tracks(vec![MatroskaSpec::TrackEntry(Master::Full(fields))]) + .cluster(0, || vec![MatroskaSpec::SimpleBlock(block)]) + .segment_end() + .build(); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + assert!(matches!( + importer.decode(&data), + Err(crate::Error::Mkv(super::Error::InvalidSubtitle( + "laced subtitles are unsupported" + ))) + )); +} + /// Opus packets packed like mkvmerge's laced audio blocks retain their boundaries and times. #[tokio::test] async fn laced_opus_frames() { @@ -790,3 +1022,77 @@ async fn unlaced_block_duration_does_not_use_track_default() { } assert_eq!(last, Some((80_000, 0))); } + +/// Text closes a group per cue, so a segment seek must not rewind onto a cue group already written. +#[test] +fn subtitle_seek_after_several_cues_does_not_reuse_a_group() { + let MatroskaSpec::TrackEntry(Master::Full(mut fields)) = text_entry(1, "S_TEXT/UTF8", "en", "English") else { + unreachable!() + }; + fields.push(MatroskaSpec::DefaultDuration(500_000_000)); + let data = MkvBuilder::new() + .header("matroska") + .segment_start() + .info(1_000_000) + .tracks(vec![MatroskaSpec::TrackEntry(Master::Full(fields))]) + .cluster(0, || { + (0..3).map(|index| simple_block(1, index * 1000, true, b"hi")).collect() + }) + .cluster(4000, || vec![simple_block(1, 0, true, b"hi")]) + .segment_end() + .build(); + // Split just before each Cluster's ID. + const CLUSTER: [u8; 4] = [0x1f, 0x43, 0xb6, 0x75]; + let clusters: Vec = data + .windows(4) + .enumerate() + .filter(|(_, window)| *window == CLUSTER) + .map(|(index, _)| index) + .collect(); + let (head, rest) = data.split_at(clusters[0]); + let (first, second) = rest.split_at(clusters[1] - clusters[0]); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + importer.decode(head).unwrap(); + importer.seek(100).unwrap(); + importer.decode(first).unwrap(); + // The next segment is 101, but the cues already spent groups past it. + importer.seek(101).unwrap(); + importer.decode(second).unwrap(); + importer.finish().unwrap(); +} + +/// A subtitle BlockGroup that replaces the codec initialization is refused, not silently ignored. +#[test] +fn subtitle_codec_state_is_refused() { + let MatroskaSpec::TrackEntry(Master::Full(mut fields)) = text_entry(1, "S_TEXT/WEBVTT", "en", "English") else { + unreachable!() + }; + fields.push(MatroskaSpec::CodecPrivate(b"WEBVTT\n".to_vec())); + let MatroskaSpec::SimpleBlock(block) = simple_block(1, 0, true, b"Hello") else { + unreachable!() + }; + let data = MkvBuilder::new() + .header("matroska") + .segment_start() + .info(1_000_000) + .tracks(vec![MatroskaSpec::TrackEntry(Master::Full(fields))]) + .cluster(0, || { + vec![MatroskaSpec::BlockGroup(Master::Full(vec![ + MatroskaSpec::Block(block), + MatroskaSpec::BlockDuration(1000), + MatroskaSpec::CodecState(b"WEBVTT\n\nSTYLE\n::cue { color: red; }\n".to_vec()), + ]))] + }) + .segment_end() + .build(); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + let result = importer.decode(&data); + assert!( + matches!(result, Err(crate::Error::Mkv(super::Error::InvalidSubtitle(_)))), + "{result:?}" + ); +} diff --git a/rs/moq-mux/src/container/mkv/mod.rs b/rs/moq-mux/src/container/mkv/mod.rs index 78d2f28ec4..77c7bb7621 100644 --- a/rs/moq-mux/src/container/mkv/mod.rs +++ b/rs/moq-mux/src/container/mkv/mod.rs @@ -19,6 +19,10 @@ mod import_test; #[derive(Debug, Clone, thiserror::Error)] #[non_exhaustive] pub enum Error { + /// A subtitle cue cannot be represented with valid text and timing. + #[error("invalid subtitle: {0}")] + InvalidSubtitle(&'static str), + #[error("unsupported EBML DocType: {0}")] UnsupportedDocType(String), diff --git a/rs/moq-mux/src/container/mod.rs b/rs/moq-mux/src/container/mod.rs index 0cc44b14c7..440859d861 100644 --- a/rs/moq-mux/src/container/mod.rs +++ b/rs/moq-mux/src/container/mod.rs @@ -20,6 +20,7 @@ mod producer; mod source; #[cfg(test)] pub(crate) mod test_util; +mod text; pub mod flv; pub mod fmp4; diff --git a/rs/moq-mux/src/container/text.rs b/rs/moq-mux/src/container/text.rs new file mode 100644 index 0000000000..214ea3b651 --- /dev/null +++ b/rs/moq-mux/src/container/text.rs @@ -0,0 +1,214 @@ +//! Timed text reconstructed from container sample metadata. + +use moq_net::Timestamp; + +pub(super) struct Cue<'a> { + pub start: Timestamp, + pub end: Timestamp, + pub body: &'a str, + pub id: &'a str, + pub settings: &'a str, +} + +pub(super) fn stamp(value: Timestamp) -> String { + format_millis(value.as_micros() / 1000) +} + +fn format_millis(millis: u128) -> String { + format!( + "{:02}:{:02}:{:02}.{:03}", + millis / 3_600_000, + millis / 60_000 % 60, + millis / 1000 % 60, + millis % 1000 + ) +} + +pub(super) fn header(value: &str) -> Result { + let value = newlines(value.trim_start_matches('\u{feff}')); + let first = value.lines().next().ok_or("missing WEBVTT header")?; + if first != "WEBVTT" && !first.starts_with("WEBVTT ") && !first.starts_with("WEBVTT\t") { + return Err("missing WEBVTT header"); + } + if value.contains("-->") || value.contains("X-TIMESTAMP-MAP") || value.contains('\0') { + return Err("unsupported WebVTT header"); + } + Ok(format!("{}\n\n", value.trim_end_matches('\n'))) +} + +/// The role a normalized WebVTT header's `Kind` metadata declares; subtitles when absent. +pub(super) fn role(header: &str) -> hang::catalog::TextRole { + let kind = header + .lines() + .skip(1) + .take_while(|line| !line.is_empty()) + .find_map(|line| line.strip_prefix("Kind:")) + .map(str::trim); + match kind { + None | Some("subtitles") => hang::catalog::TextRole::Subtitle, + Some("captions") => hang::catalog::TextRole::Caption, + Some(other) => hang::catalog::TextRole::Unknown(other.into()), + } +} + +/// WebVTT accepts CRLF, LF, and lone CR line terminators. +fn newlines(value: &str) -> String { + value.replace("\r\n", "\n").replace('\r', "\n") +} + +pub(super) fn append(output: &mut String, cue: Cue<'_>) -> Result<(), &'static str> { + if cue.end <= cue.start { + return Err("cue duration must be positive"); + } + for field in [cue.id, cue.settings] { + if field.contains(['\r', '\n', '\0']) || field.contains("-->") { + return Err("invalid cue identifier or settings"); + } + } + let body = newlines(cue.body); + if body.contains('\0') || body.contains("\n\n") || body.contains("-->") { + return Err("invalid WebVTT cue text"); + } + if !cue.id.is_empty() { + output.push_str(cue.id); + output.push('\n'); + } + // WebVTT has millisecond precision: round the end up so a sub-millisecond cue stays positive. + let start = cue.start.as_micros() / 1000; + let end = cue.end.as_micros().div_ceil(1000).max(start + 1); + output.push_str(&format_millis(start)); + output.push_str(" --> "); + output.push_str(&format_millis(end)); + if !cue.settings.is_empty() { + output.push(' '); + output.push_str(cue.settings); + } + output.push('\n'); + output.push_str(body.trim_end_matches('\n')); + output.push_str("\n\n"); + Ok(()) +} + +/// Whether cue text contains an inline `` tag. +pub(super) fn has_timestamp(body: &str) -> bool { + body.split('<') + .skip(1) + .any(|tag| tag.starts_with(|c: char| c.is_ascii_digit())) +} + +/// Matroska stores inline WebVTT timestamps relative to the Block timestamp. +pub(super) fn rebase(body: &str, start: Timestamp, end: Timestamp) -> Result { + let mut output = String::new(); + let mut remaining = body; + let mut previous = start; + while let Some(open) = remaining.find('<') { + output.push_str(&remaining[..open + 1]); + remaining = &remaining[open + 1..]; + if remaining.as_bytes().first().is_some_and(u8::is_ascii_digit) { + let close = remaining.find('>').ok_or("unterminated inline timestamp")?; + let value = &remaining[..close]; + let mut parts: Vec<&str> = value.split(':').collect(); + if !(2..=3).contains(&parts.len()) { + return Err("invalid inline timestamp"); + } + let seconds = parts.pop().unwrap(); + let (seconds, millis) = seconds.split_once('.').ok_or("invalid inline timestamp")?; + let number = |value: &str| -> Result { + if value.len() < 2 || !value.bytes().all(|value| value.is_ascii_digit()) { + return Err("invalid inline timestamp"); + } + value.parse().map_err(|_| "inline timestamp overflow") + }; + if millis.len() != 3 || seconds.len() != 2 { + return Err("invalid inline timestamp"); + } + let seconds = number(seconds)?; + let minutes = parts.pop().unwrap(); + if minutes.len() != 2 { + return Err("invalid inline timestamp"); + } + let minutes = number(minutes)?; + let hours = parts.pop().map(number).transpose()?.unwrap_or(0); + if seconds >= 60 || minutes >= 60 { + return Err("invalid inline timestamp"); + } + let millis = hours + .checked_mul(3600) + .and_then(|value| value.checked_add(minutes * 60 + seconds)) + .and_then(|value| value.checked_mul(1000)) + .and_then(|value| value.checked_add(number(millis).ok()?)) + .ok_or("inline timestamp overflow")?; + let relative = Timestamp::from_micros(millis.checked_mul(1000).ok_or("inline timestamp overflow")?) + .map_err(|_| "inline timestamp overflow")?; + let relative = relative + .convert(start.scale()) + .map_err(|_| "inline timestamp overflow")?; + let absolute = start.checked_add(relative).map_err(|_| "inline timestamp overflow")?; + if absolute <= previous || absolute >= end { + return Err("inline timestamp outside cue"); + } + previous = absolute; + output.push_str(&stamp(absolute)); + remaining = &remaining[close..]; + } + } + output.push_str(remaining); + Ok(output) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn inline_timestamps_must_advance_inside_the_cue() { + let start = Timestamp::from_nanos(250_000_000).unwrap(); + let end = Timestamp::from_nanos(1_500_000_000).unwrap(); + for body in [ + "<00:00.000>", + "<00:01.250>", + "<00:00.500><00:00.400>", + "<000:00.500>", + "<00:60.000>", + "<00:00.50>", + ] { + assert!(rebase(body, start, end).is_err(), "{body}"); + } + assert_eq!( + rebase("text<00:00.500>next", start, end).unwrap(), + "text<00:00:00.750>next" + ); + } + + #[test] + fn headers_and_cue_fields_cannot_inject_timing() { + for value in [ + "bad", + "WEBVTT\nX-TIMESTAMP-MAP=LOCAL:00:00:00.000,MPEGTS:0", + "WEBVTT\n\n00:00:00.000 --> 00:00:01.000", + ] { + assert!(header(value).is_err()); + } + let start = Timestamp::from_micros(0).unwrap(); + let end = Timestamp::from_micros(1_000_000).unwrap(); + for (id, settings, body) in [ + ("id\ninjected", "", "body"), + ("", "align:start\n", "body"), + ("", "", "a\n\nb"), + ] { + assert!( + append( + &mut String::new(), + Cue { + start, + end, + id, + settings, + body + } + ) + .is_err() + ); + } + } +}