diff --git a/README.md b/README.md index ee9ae4f..9af1832 100644 --- a/README.md +++ b/README.md @@ -94,7 +94,7 @@ Every panel is a tab you can drag, split, or stack. | Scenes | Pick the scene in preview, and add, rename (double-click), reorder, or remove one. A scene on program, or still leaving it in a transition, cannot be removed, even by an undo, until program cuts away from it. | | History | See every edit, and undo or redo back to any of them. | | Items | Place, show, hide, and order what the preview scene or the top layer draws, back to front. Select an item to filter or copy it. | -| Sources | Add a source by hand, see whether each and its own track are live and where it is shown, clone one, and remove one. Select a source to filter it. A camera or broadcast without a new picture, or a microphone or system audio without new sound, for a second reads as stalled. A browser source shows whether its page is copied on the GPU or, as a fallback, uploaded from memory (CPU), in place of a track, and **Reload** restarts its page. | +| Sources | Add a source by hand, see whether each and its own track are live and where it is shown, clone one, and remove one. Select a source to filter it. A camera or broadcast without a new picture, or a microphone or system audio without new sound, for a second reads as stalled. A browser source shows whether its page is copied on the GPU or, as a fallback, uploaded from memory (CPU), in place of a track, and **Reload** restarts its page. A media file reads as ended once it plays to its end without looping, **Play** plays it from its start, and **Reload** reads the file again. | | Devices | List this machine's cameras, displays, windows, and audio inputs, and add one as a source. | | Guests | Invite a guest, show their join link as a QR code, see who joined, and take a guest out of every scene. | | Filters | Crop the selected source or item, add, order, and set its filters, set an item's blend mode, and copy or paste filters. | @@ -264,6 +264,43 @@ costs nothing until it is shown again, where its animation has carried on. Overlays publish no track of their own, so viewers see them only in the program. +### Media files + +A `media` source plays a video or audio file, such as an intro clip or a music +bed, on the program clock, and its sound joins the mixes like any other +source's. It reads fragmented MP4, MKV or WebM, MPEG-TS, and FLV, as moq's +importers do, and fails to start on a codec this machine cannot decode. A +regular MP4 fails to start with a remux to fragmented MP4 that makes it readable: +`ffmpeg -i in.mp4 -c copy -movflags frag_keyframe+empty_moov out.mp4`. The +whole file is read into memory, compressed, when the source starts, so it +plays from its start at once and loops without reading the disk again; keep it +to clips and music, not hours of video. + +```toml +[sources.intro] +kind = "media" +path = "intro.mkv" # relative to the scene file, like an image's +restart = true # play from the start each time it goes on program + +[sources.music] +kind = "media" +path = "music.webm" +loop = true # start over at the end +``` + +Without `loop`, it plays once, then holds its last picture, silent, and reads +as ended in Sources and, while the program shows or plays it, in Health. +**Play** in Sources plays it from its start again, as does going on program +with `restart`, in a scene on air or the top layer, when it was not on program +before. Either clears its picture at once, so the end of the last play never +shows on program. A file that cannot be read or decoded fails, with the reason. +Its pictures and its sound each have a track of their own (see [Source +tracks](#source-tracks)), its pictures on one timeline across plays. Each is +used apart, as a camera's and a microphone's are: the pictures while a scene on +air shows it, the sound while the mix on program plays it, and either while +read. So a music bed that no scene shows encodes its sound alone. A file with +no sound has no strip in the Audio panel. + ## Guests Guests and phones join a room the studio owns on its relay, from a web page, @@ -467,7 +504,8 @@ rect = { x = 1600, y = 40, width = 280, height = 120 } Source kinds are `camera`, `display`, and `microphone` (each with an optional `id`, the default device without one), `window` (with an `id`), `system_audio` -(macOS), `image` (PNG or JPEG), `color` (`#rrggbb` or `#rrggbbaa`), +(macOS), `image` (PNG or JPEG), `media` (a video or audio file; see +[Media files](#media-files)), `color` (`#rrggbb` or `#rrggbbaa`), `broadcast`, `browser` (a `url`, and a `width` and `height` up to 4096), `overlay` (its `html`, and a `width` and `height` up to 4096), `clone` (with the `source` it shows), `lut` (a 3D Adobe or Resolve `.cube` file, which no item places, but which grades what a `lut` filter names it in), @@ -584,9 +622,9 @@ stream. ### Source tracks -Each camera, display, window, image, color, microphone, and system audio also -has a track of its own, so the studio can replay or record it on its own. These -tracks stay in the studio: the program is all it publishes to a relay. They are listed in a local +Each camera, display, window, image, media file, color, microphone, and system +audio also has a track of its own, so the studio can replay or record it on its +own. These tracks stay in the studio: the program is all it publishes to a relay. They are listed in a local catalog under the `dev.moq.studio` section's `sources`, by source name, each as a hang catalog of its own on the program's clock. @@ -602,8 +640,8 @@ plain capture, before any filter, and a clone has none, since it shows its source's capture. A browser source or an overlay has no track of its own, and a broadcast source is already its publisher's. -A microphone or system audio encodes its track as Opus, from the same device -open the mixes hear, only while it is used: while the mix on program plays it +A microphone, system audio, or media file's sound encodes its track as Opus, +from the same sound the mixes hear, only while it is used: while the mix on program plays it (unmuted and not left out of it), so a replay of it has its recent past, or while a recording reads its track. The track carries the sound as captured, before its gain, mute, and delay, which stay the mixes'. A buffer the capture @@ -620,7 +658,7 @@ encode engine for them all. The Output panel counts the sessions open: | The program's canvas size, live | 1, or none without `--connect` | | Its half size | 1 while someone watches it | | The recording's own canvas size | 1 while recording | -| Each captured source | 1 while on program or recorded | +| Each captured source or media file | 1 while on program or recorded | So a live scene showing a camera, a display, and a window holds 4 sessions, 5 with the half size watched, and 6 while recording too. When the program @@ -702,8 +740,8 @@ it is written, shortly after the click. Nothing records in the background for this. The program's tracks keep their last 30 seconds for late viewers anyway, and each source's track while it is on -program (shown in a scene on air, or for a microphone or system audio, played by -the mix on program) or recorded, and a replay copies from those, so a replay +program (a picture shown in a scene on air, or a sound played by the mix on +program) or recorded, and a replay copies from those, so a replay goes back at most that far. A replay of the program needs `--connect`; a local source keeps its track without a relay. With `--on-demand`, the program has no past from before someone watched it, and a source has none from before it @@ -723,7 +761,8 @@ with its cause in plain words: a relay is unreachable, the uplink is congested or losing packets, the relay is far away, the encoder is behind or stuck, a source stalled (a camera or broadcast the program draws, or a microphone or system audio the mix on program plays: unmuted and not left -out of it), or viewers are late, stalling, or cannot +out of it), a source the program shows or plays failed or, as a media file +that does not loop, played to its end, or viewers are late, stalling, or cannot decode a track. A relay lost while the other serves is degraded, since viewers have nothing left to fall back to; with none serving it is unhealthy. Below that it shows: diff --git a/quest/m0/README.md b/quest/m0/README.md index 0786410..716236c 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -91,7 +91,6 @@ repository. - [Studio on macOS with capture granted](/quest/m0/app-macos.md) - camera, screen, and audio capture run on a Mac, not only Linux - [Studio on Windows](/quest/m0/app-windows.md) - the studio runs on Windows, not only Linux - [Wayland display and window sources](/quest/m0/wayland-sources.md) - portal-picked sources that reopen without the picker -- [Media file source](/quest/m0/media-file.md) - a local video or audio file plays as a source, looping or from the start on demand - [Stinger transitions](/quest/m0/stinger.md) - a clip played over the program that switches scenes at its cut point - [ML video effects](/quest/m0/effects.md) - background removal and other ML effects, baked into their source's track - [Destinations](/quest/m0/destinations.md) - RTMP, RTMPS, and WHIP push with per-destination codecs and shared encoders, on a UI-free egress library a hosted service can reuse diff --git a/quest/m0/media-file.md b/quest/m0/media-file.md deleted file mode 100644 index 8f5008d..0000000 --- a/quest/m0/media-file.md +++ /dev/null @@ -1,29 +0,0 @@ -# [M] Media file source - -## Goal - -A source in `moq-studio` plays a local video or audio file, such as an intro -clip or background music: looping, or from the start each time something asks, -with its sound joining the mixes like any other source. - -## Plan - -Decided 2026-11-04 (maintainer): planned so stinger transitions have a clip -that starts on demand. Demux and decode through the moq crates the studio -already uses (moq-mux import, moq-video and moq-audio decode) rather than a -new media stack; anything they lack lands in moq first, with a moq-side test, -then a pin. Start with the container and codecs those crates already read, and -refuse anything else at setup with the reason. The file path is a local path -in the scene file, like an image source's. The web studio has no local paths: -there a file comes from a browser file picker for the session, and resolving -scene-file paths in the browser belongs to the [web studio](/quest/m0/web.md), -as it does for image sources. - -A file is played on the program clock. Settings: loop, or play once and hold -the last frame (an audio-only file goes silent after its last sample); and whether going on program restarts it. A restart on demand -(the stinger's need) is an API on the source, not only a setting. A file that -ends or fails reads as ended or failed in the Sources panel and Health. - -## Related - -- [Stinger transitions](/quest/m0/stinger.md) - the first consumer of a clip that starts on demand diff --git a/quest/m0/stinger.md b/quest/m0/stinger.md index 3efb4f2..4c4fdf3 100644 --- a/quest/m0/stinger.md +++ b/quest/m0/stinger.md @@ -11,9 +11,6 @@ Split from the keyframes quest on 2026-11-04 (maintainer). A stinger is a transition kind beside fade, slide, wipe, and move ([moq-dev/studio#33](https://github.com/moq-dev/studio/pull/33)), drawn on the program clock, whose clip is a media file source. It must start its clip as -the transition starts, or its cut point lands at random. Its settings (clip -and cut point) change through `Edit`, so they undo. - -## Required - -- [Media file source](/quest/m0/media-file.md) - a clip that starts on demand +the transition starts, or its cut point lands at random: `Feed::play` in +`rs/moq-scene/src/source.rs` starts a media source from its start at once. +Its settings (clip and cut point) change through `Edit`, so they undo. diff --git a/quest/m0/web.md b/quest/m0/web.md index 6fe0701..d490116 100644 --- a/quest/m0/web.md +++ b/quest/m0/web.md @@ -31,6 +31,10 @@ same PR as this page, and the page follows each pin bump. The page was planned for moq's demo/web; it now ships from this repository, and where it is hosted is settled in this quest. +Decided 2026-11-04 (maintainer): a page has no local paths, so an image or +media source's file comes from a browser file picker for the session, and +resolving the scene file's paths in the browser is settled here. + ## Required - [moq-video's browser backends are pinned](/quest/m0/pin-browser-video.md) - the video slice's capture, codecs, and GPU import diff --git a/rs/moq-scene/src/collection.rs b/rs/moq-scene/src/collection.rs index 7cf2a8c..8094c5e 100644 --- a/rs/moq-scene/src/collection.rs +++ b/rs/moq-scene/src/collection.rs @@ -1228,6 +1228,20 @@ pub enum Source { /// The file, relative to the scene file's directory unless absolute. path: PathBuf, }, + /// A video or audio file, such as an intro clip or a music bed, played on + /// the program clock from its start, and again whenever + /// [`Feed::play`](crate::source::Feed::play) asks. + Media { + /// The file, relative to the scene file's directory unless absolute. + path: PathBuf, + /// Whether it starts over at its end, rather than holding its last + /// picture and going silent. + #[serde(default, rename = "loop", skip_serializing_if = "std::ops::Not::not")] + looping: bool, + /// Whether it plays from its start each time it goes on program. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + restart: bool, + }, /// A solid color. Color { /// The color, as `#rrggbb` or `#rrggbbaa`. @@ -1288,10 +1302,10 @@ impl Source { !matches!(self, Self::Microphone { .. } | Self::SystemAudio | Self::Lut { .. }) } - /// The file an image or LUT source reads. + /// The file an image, media, or LUT source reads. pub(crate) fn path(&self) -> Option<&std::path::Path> { match self { - Self::Image { path } | Self::Lut { path } => Some(path), + Self::Image { path } | Self::Media { path, .. } | Self::Lut { path } => Some(path), _ => None, } } @@ -1299,7 +1313,7 @@ impl Source { /// The file the source reads, to resolve or relativize it. pub(crate) fn path_mut(&mut self) -> Option<&mut PathBuf> { match self { - Self::Image { path } | Self::Lut { path } => Some(path), + Self::Image { path } | Self::Media { path, .. } | Self::Lut { path } => Some(path), _ => None, } } @@ -1314,7 +1328,12 @@ impl Source { pub fn has_audio(&self) -> bool { matches!( self, - Self::Microphone { .. } | Self::SystemAudio | Self::Broadcast { .. } | Self::Browser { .. } | Self::Replay + Self::Microphone { .. } + | Self::SystemAudio + | Self::Media { .. } + | Self::Broadcast { .. } + | Self::Browser { .. } + | Self::Replay ) } @@ -1498,6 +1517,14 @@ mod tests { path: "studio/guest".into(), }, ); + collection.sources.insert( + "intro".into(), + Source::Media { + path: "intro.mkv".into(), + looping: false, + restart: true, + }, + ); collection.sources.insert( "chat".into(), Source::Browser { @@ -1833,7 +1860,16 @@ rect = { x = 0, y = 680, width = 1280, height = 40 } collection.validate().unwrap(); collection.top.push(Item::new("grade", Rect::fill(&collection.canvas))); assert!(matches!(collection.validate(), Err(Error::Invalid(_)))); - for source in [Source::Lut { path: "".into() }, Source::Image { path: "".into() }] { + let media = Source::Media { + path: "".into(), + looping: true, + restart: false, + }; + for source in [ + Source::Lut { path: "".into() }, + Source::Image { path: "".into() }, + media, + ] { let mut collection = example(); collection.sources.insert("blank".into(), source.clone()); assert!(matches!(collection.validate(), Err(Error::Invalid(_))), "{source:?}"); diff --git a/rs/moq-scene/src/lib.rs b/rs/moq-scene/src/lib.rs index 33cc911..41d5e51 100644 --- a/rs/moq-scene/src/lib.rs +++ b/rs/moq-scene/src/lib.rs @@ -12,7 +12,7 @@ //! scenes and collections as text. //! - [`source::Sources`] runs each source: a camera, display, window, //! microphone, or system-audio capture (with the `capture` feature), a still -//! image, a solid color, a broadcast subscribed over MoQ, or a web page +//! image, a video or audio file, a solid color, a broadcast subscribed over MoQ, or a web page //! rendered by a browser engine (the [`browser`] module). Pictures go to the //! compositor and sound goes to a `moq_audio::bus::Bus`, routed into its //! program mix. @@ -37,6 +37,7 @@ mod compositor; mod edit; mod effect; mod lut; +mod media; mod overlay; mod playout; pub mod publish; @@ -88,6 +89,11 @@ pub enum Error { #[error("image: {0}")] Image(String), + /// A media source's file could not be read, or holds nothing the studio + /// plays. + #[error("media: {0}")] + Media(String), + /// A LUT source could not be read or parsed. #[error("LUT: {0}")] Lut(String), diff --git a/rs/moq-scene/src/media.rs b/rs/moq-scene/src/media.rs new file mode 100644 index 0000000..2de44ce --- /dev/null +++ b/rs/moq-scene/src/media.rs @@ -0,0 +1,193 @@ +//! Media files, such as an intro clip or a music bed, demuxed by moq-mux into a +//! broadcast of their own for a media source to decode. +//! +//! The whole file is imported when the source opens, still compressed, so each +//! play decodes from its first frame at once, as a stinger needs, and a loop +//! never reads the disk again. The source holds the file's size in memory for +//! as long as it runs. + +use std::path::Path; + +use hang::catalog::{AudioConfig, VideoConfig}; +use moq_mux::import::ContainerFormat; + +use crate::Error; + +/// A media file's video and audio, held in a broadcast of their own. +pub(crate) struct Clip { + broadcast: moq_net::broadcast::Producer, + /// Keeps the imported tracks readable. + _import: moq_mux::import::Container, + _catalog: moq_mux::catalog::Producer, + /// The video rendition, by catalog name. + pub video: Option<(String, VideoConfig)>, + /// The audio rendition, by catalog name. + pub audio: Option<(String, AudioConfig)>, +} + +impl Clip { + /// Read and demux the file at `path`. Refuses a container moq-mux does not + /// read, saying which it does, and a file with neither video nor audio in + /// it that moq-mux reads. + pub fn open(path: &Path) -> Result { + let failed = |err: String| Error::Media(format!("{}: {err}", path.display())); + let bytes = std::fs::read(path).map_err(|err| failed(err.to_string()))?; + let format = container(&bytes).map_err(failed)?; + + let mut broadcast = moq_net::broadcast::Info::new().produce(); + // The file is written whole before anything reads it, so nothing in it + // may go stale behind the newest frame. + let config = moq_mux::catalog::Config::default().with_max_age(None); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config)?; + let init = moq_mux::import::ContainerInit::new(format, bytes); + let mut import = moq_mux::import::Container::new(broadcast.clone(), catalog.reserve(), &init) + .map_err(|err| failed(err.to_string()))?; + import.finish().map_err(|err| failed(err.to_string()))?; + + let listed = catalog.snapshot(); + let video = crate::source::pick(listed.video.enabled())?; + let audio = crate::source::pick(listed.audio.enabled())?; + if video.is_none() && audio.is_none() { + return Err(failed("it holds no video or audio the studio plays".into())); + } + Ok(Self { + broadcast, + _import: import, + _catalog: catalog, + video, + audio, + }) + } + + /// The clip's broadcast, to decode from its start. + pub fn consume(&self) -> moq_net::broadcast::Consumer { + self.broadcast.consume() + } +} + +/// The containers a media file may be in, as moq-mux reads them. +const READ: &str = "fragmented MP4, MKV or WebM, MPEG-TS, or FLV"; + +/// Which container `bytes` are in, by their first bytes. +/// +/// A regular MP4, with its samples indexed in a `moov` rather than in +/// fragments, is refused: moq-mux demuxes only fragmented MP4. +fn container(bytes: &[u8]) -> Result { + const TS_PACKET: usize = 188; + if bytes.starts_with(&[0x1a, 0x45, 0xdf, 0xa3]) { + return Ok(ContainerFormat::Mkv); + } + if bytes.starts_with(b"FLV") { + return Ok(ContainerFormat::Flv); + } + if bytes.first() == Some(&0x47) && bytes.get(TS_PACKET).is_none_or(|sync| *sync == 0x47) { + return Ok(ContainerFormat::Ts); + } + if bytes.get(4..8) == Some(b"ftyp") { + return match fragmented(bytes) { + true => Ok(ContainerFormat::Fmp4), + false => Err(format!( + "a regular MP4 is not read yet; remux it into one of {READ}, such as with \ + `ffmpeg -i in.mp4 -c copy -movflags frag_keyframe+empty_moov out.mp4`" + )), + }; + } + Err(format!("not a container the studio reads; it reads {READ}")) +} + +/// Whether the MP4 in `bytes` holds a fragment (`moof`) among its top-level boxes. +fn fragmented(mut bytes: &[u8]) -> bool { + while let Some(header) = bytes.get(..8) { + if &header[4..] == b"moof" { + return true; + } + let size = u32::from_be_bytes(header[..4].try_into().expect("four bytes")); + let size = match size { + // To the end of the file. + 0 => return false, + // A 64-bit size follows the type. + 1 => match bytes.get(8..16) { + Some(large) => u64::from_be_bytes(large.try_into().expect("eight bytes")), + None => return false, + }, + size => u64::from(size), + }; + match usize::try_from(size).ok().filter(|size| *size >= 8) { + Some(size) if size <= bytes.len() => bytes = &bytes[size..], + _ => return false, + } + } + false +} + +#[cfg(test)] +pub(crate) mod tests { + use super::*; + + /// One second of 64x64 H.264 at 10 fps with a keyframe every half second, + /// and stereo Opus, in MKV: + /// + /// ```sh + /// ffmpeg -f lavfi -i testsrc2=size=64x64:rate=10:duration=1 \ + /// -f lavfi -i sine=frequency=440:sample_rate=48000:duration=1 \ + /// -c:v libx264 -profile:v baseline -g 5 -bf 0 -c:a libopus -b:a 24k -ac 2 clip.mkv + /// ``` + pub(crate) const CLIP: &[u8] = include_bytes!("test_data/clip.mkv"); + + /// The same second, remuxed with `-c copy -movflags frag_keyframe+empty_moov`. + const FRAGMENTED: &[u8] = include_bytes!("test_data/clip.mp4"); + + /// Half a second of stereo Opus alone, in WebM. + pub(crate) const TONE: &[u8] = include_bytes!("test_data/tone.webm"); + + /// The first half second of [`CLIP`]'s pictures alone. + pub(crate) const SILENT: &[u8] = include_bytes!("test_data/silent.mkv"); + + /// [`CLIP`]'s second of pictures with [`TONE`]'s half second of sound. + pub(crate) const UNEVEN: &[u8] = include_bytes!("test_data/uneven.mkv"); + + #[test] + fn reads_by_the_first_bytes() { + assert_eq!(container(CLIP), Ok(ContainerFormat::Mkv)); + assert_eq!(container(FRAGMENTED), Ok(ContainerFormat::Fmp4)); + assert_eq!(container(b"FLV\x01"), Ok(ContainerFormat::Flv)); + let mut ts = vec![0; 2 * 188]; + (ts[0], ts[188]) = (0x47, 0x47); + assert_eq!(container(&ts), Ok(ContainerFormat::Ts)); + ts[188] = 0; + assert!(container(&ts).is_err(), "one sync byte is not a stream"); + assert!(container(b"\x89PNG\r\n\x1a\n").unwrap_err().contains(READ)); + } + + /// A regular MP4 says how to make one moq-mux reads. + #[test] + fn a_regular_mp4_is_refused() { + let mut mp4 = Vec::new(); + for (kind, body) in [(b"ftyp", 8), (b"moov", 32), (b"mdat", 64)] { + mp4.extend_from_slice(&(8 + body as u32).to_be_bytes()); + mp4.extend_from_slice(kind); + mp4.resize(mp4.len() + body, 0); + } + let err = container(&mp4).unwrap_err(); + assert!(err.contains("regular MP4") && err.contains("frag_keyframe"), "{err}"); + // Even when a box says it is larger than the file. + mp4.truncate(mp4.len() - 1); + assert!(container(&mp4).is_err()); + } + + #[test] + fn opens_both_renditions() { + let dir = tempfile::tempdir().unwrap(); + for (name, bytes) in [("clip.mkv", CLIP), ("clip.mp4", FRAGMENTED)] { + let path = dir.path().join(name); + std::fs::write(&path, bytes).unwrap(); + let clip = Clip::open(&path).unwrap(); + assert!(clip.video.is_some(), "{name} has no video"); + assert!(clip.audio.is_some(), "{name} has no audio"); + } + + let missing = dir.path().join("missing.mkv"); + let err = Clip::open(&missing).err().unwrap().to_string(); + assert!(err.contains("missing.mkv"), "{err}"); + } +} diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 665c388..2eb6a92 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -6,10 +6,10 @@ //! sessions a GPU allows. Its renditions are listed in a hang catalog of its //! own, nested by source name under the local catalog's [`SECTION`] section. //! -//! A source encodes only while it is used: while it is on program -//! ([`Publish::set_on_program`]), shown in a scene on air or heard in the -//! program mix, so a replay of it has its recent past, or while something in -//! the studio subscribes to its track, such as a recording. +//! A track encodes only while it is used: while it is on program +//! ([`Publish::set_on_program`]), a picture shown in a scene on air or a sound +//! heard in the program mix, so a replay of it has its recent past, or while +//! something in the studio subscribes to it, such as a recording. use std::collections::{BTreeMap, BTreeSet}; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -88,7 +88,18 @@ pub struct Publish { sessions: Sessions, owners: Arc>, /// The sources on program, which each source's track follows. - on_program: Arc>>, + on_program: Arc>, +} + +/// The sources on program, by name, apart for pictures and sound: a media +/// file heard in the program mix but shown in no scene on air encodes its +/// sound and not its pictures. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct OnProgram { + /// Those whose picture a scene on air shows. + pub shown: BTreeSet, + /// Those whose sound the program mix plays. + pub heard: BTreeSet, } /// Which [`Claim`] owns each source name's entry in the local catalog, and @@ -134,13 +145,14 @@ impl Publish { self.catalog.snapshot() } - /// Encode the sources named in `names`, those on program, whether or not - /// anything reads their tracks; any other encodes only while read. - pub fn set_on_program(&self, names: BTreeSet) { + /// Encode the picture of each source `on_program` shows and the sound of + /// each it hears, whether or not anything reads their tracks; any other + /// track encodes only while read. + pub fn set_on_program(&self, on_program: OnProgram) { // Only a change wakes the sources, since this runs every frame. self.on_program.send_if_modified(|current| { - let changed = *current != names; - *current = names; + let changed = *current != on_program; + *current = on_program; changed }); } @@ -205,7 +217,7 @@ pub(crate) struct Listing<'a> { claim: &'a Claim, kind: moq_video::encode::Kind, sessions: Sessions, - on_program: tokio::sync::watch::Receiver>, + on_program: tokio::sync::watch::Receiver, broadcast: moq_net::broadcast::Producer, studio: Catalog, catalog: moq_mux::catalog::Producer, @@ -471,7 +483,7 @@ pub(crate) struct Video<'a> { track: Encoding, claim: &'a Claim, /// The sources on program. - on_program: tokio::sync::watch::Receiver>, + on_program: tokio::sync::watch::Receiver, config: moq_video::encode::Config, clock: moq_mux::Clock, /// Where the capture's private timeline starts on the studio clock, in @@ -546,20 +558,17 @@ impl<'a> Video<'a> { /// Whether the source is used: shown on program, or its track read. fn used(&self) -> bool { - self.on_program.borrow().contains(&self.claim.name) || self.track.watched() + self.on_program.borrow().shown.contains(&self.claim.name) || self.track.watched() } /// Wait until whether the source is used disagrees with whether its - /// encoder is open, so a capture that delivers nothing new, such as a - /// still screen, can act without waiting on a picture: [`release`] the - /// encoder when it is open and nothing uses the source, or bring a picture - /// when it is closed and the source goes into use. + /// encoder is open, so a source that delivers nothing new, such as a + /// still screen or a media file that ended, can act without waiting on a + /// picture: [`release`] the encoder when it is open and nothing uses the + /// source, or bring a picture when it is closed and the source goes into + /// use. /// /// [`release`]: Self::release - #[cfg_attr( - not(any(feature = "capture", test)), - expect(dead_code, reason = "only a capture waits on it") - )] pub(crate) async fn settle(&self) { let mut demand = self.track.demand(); let mut on_program = self.on_program.clone(); @@ -574,7 +583,6 @@ impl<'a> Video<'a> { } /// Whether the encoder is open. - #[cfg_attr(not(feature = "capture"), expect(dead_code, reason = "only a capture asks"))] pub(crate) fn is_open(&self) -> bool { self.track.is_open() } @@ -598,6 +606,16 @@ impl<'a> Video<'a> { Ok(()) } + /// Start a group with the next picture, as a keyframe, such as where a + /// media file plays again from its start. A closed encoder starts with one + /// anyway. + pub(crate) fn cut(&mut self) -> Result<(), Error> { + if self.track.is_open() { + self.track.cut()?; + } + Ok(()) + } + /// Mark a break, such as a gap in a replay: what follows does not continue /// what came before, and a viewer arriving during it waits for what follows /// rather than being served the last picture as if it were live. @@ -617,28 +635,21 @@ fn encodable(size: moq_video::Size) -> moq_video::Size { moq_video::Size::new((size.width & !1).max(2), (size.height & !1).max(2)) } -/// A captured source's sound, encoded as Opus while the source is used, as -/// captured: before the gain and mute the program mix applies to it. -#[cfg_attr( - not(any(feature = "capture", test)), - expect(dead_code, reason = "only a capture publishes sound") -)] +/// A captured or played source's sound, encoded as Opus while the source is +/// used, as captured or decoded: before the gain and mute the program mix +/// applies to it. pub(crate) struct Audio<'a> { /// Taken only as the track ends. producer: Option, claim: &'a Claim, /// The sources on program. - on_program: tokio::sync::watch::Receiver>, + on_program: tokio::sync::watch::Receiver, clock: moq_mux::Clock, /// Whether sound went to the encoder since the source last went out of /// use, so leaving use marks one gap. encoding: bool, } -#[cfg_attr( - not(any(feature = "capture", test)), - expect(dead_code, reason = "only a capture publishes sound") -)] impl<'a> Audio<'a> { /// Advertise an Opus rendition of interleaved `f32` sound in `input`'s /// sample rate and layout. @@ -662,7 +673,7 @@ impl<'a> Audio<'a> { /// sound after one is stamped afresh, so the track's timeline holds it. pub(crate) fn write(&mut self, samples: &[f32], captured: kio::clock::Instant, gap: bool) -> Result<(), Error> { let producer = self.producer.as_mut().expect("open until it ends"); - let used = self.on_program.borrow().contains(&self.claim.name) || producer.demand().wanted().is_some(); + let used = self.on_program.borrow().heard.contains(&self.claim.name) || producer.demand().wanted().is_some(); if !used { if std::mem::take(&mut self.encoding) { producer.discontinuity()?; @@ -682,7 +693,6 @@ impl<'a> Audio<'a> { } /// End the track, failed with `err` so viewers see why. - #[cfg_attr(not(feature = "capture"), expect(dead_code, reason = "only a capture fails"))] pub(crate) fn abort(mut self, err: &Error) { if let Some(producer) = self.producer.take() { producer.abort(moq_net::Error::Transport(err.to_string())); @@ -763,6 +773,35 @@ mod tests { (publish, catalog, consumer) } + /// On program with only `name`'s picture shown. + fn shown(name: &str) -> OnProgram { + OnProgram { + shown: [name.to_string()].into(), + ..Default::default() + } + } + + /// On program with only `name`'s sound heard. + fn heard(name: &str) -> OnProgram { + OnProgram { + heard: [name.to_string()].into(), + ..Default::default() + } + } + + /// Walk a media source playing a 10 fps clip through `count` pictures, the + /// first due at `first`. The clock moves to each picture's moment and is + /// held while it decodes, encodes, and shows, so no wait counts on work in + /// threads the runtime cannot see, such as a decoder's or an encoder's, + /// finishing within any span of the clock. + async fn pictures(feed: &crate::source::Feed, first: kio::clock::Instant, count: u32) { + let shown = || matches!(feed.take(), Some(crate::source::Picture::Frame(_))).then_some(()); + for index in 0..count { + until_time(first + std::time::Duration::from_millis(100) * index).await; + until(shown).await; + } + } + /// Yield until `f` holds of the local catalog's source listings. async fn listed(catalog: &Catalog, f: impl Fn(&BTreeMap) -> bool) -> hang::Catalog { for _ in 0..1000 { @@ -900,7 +939,7 @@ mod tests { let snapshot = listed(&catalog, |sources| width(sources) == Some(64)).await; assert!(snapshot.video.renditions.is_empty(), "no source is a program rendition"); - publish.set_on_program(["camera".to_string()].into()); + publish.set_on_program(shown("camera")); for millis in [66, 100, 133] { video.encode(&mut gray(small, millis)).unwrap(); } @@ -962,7 +1001,7 @@ mod tests { kio::clock::timeout(wait, video.settle()).await.is_err(), "unused and closed" ); - publish.set_on_program(["display".to_string()].into()); + publish.set_on_program(shown("display")); kio::clock::timeout(wait, video.settle()) .await .expect("on program without a picture, so the capture brings one"); @@ -980,7 +1019,7 @@ mod tests { // lands on the program's clock now rather than a minute back. kio::clock::sleep(std::time::Duration::from_secs(60)).await; video.reopened(); - publish.set_on_program(["display".to_string()].into()); + publish.set_on_program(shown("display")); let now = publish.catalog.clock().now(); let mut picture = gray(size, 0); video.encode(&mut picture).unwrap(); @@ -1128,7 +1167,7 @@ mod tests { let refused = crate::replay::Take::new(&target(), std::time::Duration::from_secs(1)); assert!(refused.is_err(), "nothing to replay"); - publish.set_on_program(["mic".to_string()].into()); + publish.set_on_program(heard("mic")); let heard = clock.now(); capture(&mut audio, false).await; assert!(written() > 0, "on program"); @@ -1353,7 +1392,7 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let (publish, catalog, consumer) = studio(); // On program, so its track encodes. - publish.set_on_program(["replay".to_string()].into()); + publish.set_on_program(shown("replay")); let (mut context, _origin) = context(dir.path()); context.publish = Some(publish); let mut collection = Collection::new(); @@ -1408,4 +1447,227 @@ mod tests { }) .await; } + + /// A media source on program encodes its pictures onto its own track as + /// they show, on one timeline when it plays again from its start. + #[tokio::test(start_paused = true)] + async fn a_media_file_is_published_on_one_timeline() { + local(async { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("clip.mkv"), crate::media::tests::CLIP).unwrap(); + let (publish, catalog, consumer) = studio(); + publish.set_on_program(shown("clip")); + let sessions = publish.sessions.clone(); + let (mut context, _origin) = context(dir.path()); + context.publish = Some(publish.clone()); + let mut collection = Collection::new(); + let clip = Source::Media { + path: "clip.mkv".into(), + looping: false, + restart: false, + }; + collection.sources.insert("clip".into(), clip); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + + // The first picture is due at once, so the clock stays put while + // the encoder opens. + let listed = until(|| { + let listed = catalog.snapshot().ext.studio.sources.remove("clip"); + listed.filter(|listed| !listed.video.renditions.is_empty()) + }) + .await; + let feed = sources.get("clip").unwrap(); + assert_eq!(feed.track(), Some(Status::Live)); + let (name, config) = listed.video.renditions.iter().next().unwrap(); + let track = consumer + .track(hang::catalog::Reference::track(config, name)) + .unwrap() + .subscribe( + moq_net::track::Subscription::default().with_delay(moq_net::track::Delay::default().with_max(None)), + ) + .await + .unwrap(); + let container = moq_mux::catalog::hang::Container::try_from(config).unwrap(); + let mut track = moq_mux::container::Consumer::new(track, container); + + // Half way, five pictures in, it starts over, and plays to its end. + let first = kio::clock::now(); + pictures(feed, first, 5).await; + until_time(first + std::time::Duration::from_millis(450)).await; + let cut = publish.catalog.clock().now(); + feed.play().unwrap(); + let restarted = kio::clock::now(); + pictures(feed, restarted, 10).await; + until_time(restarted + std::time::Duration::from_millis(1_100)).await; + until(|| (feed.status() == Status::Ended).then_some(())).await; + + // How many pictures land depends on the encoder, which may skip + // some to hold its rate, as openh264 does at this size; the + // keyframe each play opens with always does. + let now = kio::Waiter::noop(); + let mut stamps = Vec::new(); + until(|| { + while let std::task::Poll::Ready(read) = track.poll_read(&now) { + if let moq_mux::container::Read::Frame { frame, .. } = read.unwrap().expect("open") { + stamps.push(frame.timestamp); + } + } + stamps.iter().any(|stamp| *stamp >= cut).then_some(()) + }) + .await; + assert!(stamps[0] < cut, "the first play's keyframe: {stamps:?}"); + assert!(stamps.windows(2).all(|pair| pair[0] < pair[1]), "{stamps:?}"); + + // Ended, it shows nothing new, yet gives its session back once off + // program and unread. + assert_eq!(sessions.open(), 1, "on program"); + drop(track); + publish.set_on_program(Default::default()); + until(|| (sessions.open() == 0).then_some(())).await; + }) + .await; + } + + /// A music bed the program mix plays encodes its sound, while its + /// pictures, which no scene on air shows, encode nothing until one does. + #[tokio::test(start_paused = true)] + async fn a_heard_media_file_encodes_only_its_sound() { + local(async { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("bed.mkv"), crate::media::tests::CLIP).unwrap(); + let (publish, catalog, _consumer) = studio(); + publish.set_on_program(heard("bed")); + let sessions = publish.sessions.clone(); + let (mut context, _origin) = context(dir.path()); + context.publish = Some(publish.clone()); + let mut collection = Collection::new(); + let bed = Source::Media { + path: "bed.mkv".into(), + looping: true, + restart: false, + }; + collection.sources.insert("bed".into(), bed); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + + // Both listed, the pictures' at the first picture. + let listed = until(|| { + let listed = catalog.snapshot().ext.studio.sources.remove("bed"); + listed.filter(|listed| !listed.video.renditions.is_empty() && !listed.audio.renditions.is_empty()) + }) + .await; + let audio = listed.audio.renditions.keys().next().unwrap().clone(); + let video = listed.video.renditions.keys().next().unwrap().clone(); + let written = |id: &str| publish.stats().get(id).map_or(0, |rendition| rendition.frames); + + let feed = sources.get("bed").unwrap(); + let first = kio::clock::now(); + pictures(feed, first, 4).await; + until(|| (written(&audio) > 0).then_some(())).await; + assert_eq!(written(&video), 0, "heard, not shown"); + assert_eq!(sessions.open(), 0, "no encoder for the pictures"); + + // Shown too, its pictures encode as they play, past what a + // pipelined encoder holds back. + publish.set_on_program(OnProgram { + shown: ["bed".to_string()].into(), + heard: ["bed".to_string()].into(), + }); + pictures(feed, first + std::time::Duration::from_millis(400), 6).await; + until(|| (written(&video) > 0).then_some(())).await; + }) + .await; + } + + /// A looping file whose sound is shorter than its pictures keeps its sound + /// track on time: each loop's sound starts where the loop does, the + /// silence between kept, rather than right after the last loop's sound. + #[tokio::test(start_paused = true)] + async fn a_looping_media_file_keeps_its_sound_on_time() { + local(async { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("bed.mkv"), crate::media::tests::UNEVEN).unwrap(); + let (publish, catalog, consumer) = studio(); + publish.set_on_program(heard("bed")); + let (mut context, _origin) = context(dir.path()); + context.publish = Some(publish.clone()); + let mut collection = Collection::new(); + let bed = Source::Media { + path: "bed.mkv".into(), + looping: true, + restart: false, + }; + collection.sources.insert("bed".into(), bed); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + + let listed = until(|| { + let listed = catalog.snapshot().ext.studio.sources.remove("bed"); + listed.filter(|listed| !listed.audio.renditions.is_empty()) + }) + .await; + let (name, config) = listed.audio.renditions.iter().next().unwrap(); + let track = consumer + .track(hang::catalog::Reference::track(config, name)) + .unwrap() + .subscribe( + moq_net::track::Subscription::default().with_delay(moq_net::track::Delay::default().with_max(None)), + ) + .await + .unwrap(); + let container = moq_mux::catalog::hang::Container::try_from(config).unwrap(); + let mut track = moq_mux::container::Consumer::new(track, container); + // Two and a half loops of a second, each with half a second of sound. + pictures(sources.get("bed").unwrap(), kio::clock::now(), 25).await; + let now = kio::Waiter::noop(); + let mut stamps = Vec::new(); + let silences = until(|| { + while let std::task::Poll::Ready(read) = track.poll_read(&now) { + if let moq_mux::container::Read::Frame { frame, .. } = read.unwrap().expect("open") { + stamps.push(std::time::Duration::from(frame.timestamp)); + } + } + let gaps = stamps.windows(2).map(|pair| pair[1].saturating_sub(pair[0])); + let silences = gaps.filter(|gap| *gap > std::time::Duration::from_millis(300)).count(); + (silences >= 2).then_some(silences) + }) + .await; + assert_eq!(silences, 2, "each loop's silence is kept: {stamps:?}"); + }) + .await; + } + + /// A file of sound alone lists its sound's track, live, and no pictures. + #[tokio::test(start_paused = true)] + async fn an_audio_file_lists_its_sound() { + local(async { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("tone.webm"), crate::media::tests::TONE).unwrap(); + let (publish, catalog, _consumer) = studio(); + let (mut context, _origin) = context(dir.path()); + context.publish = Some(publish); + let mut collection = Collection::new(); + let tone = Source::Media { + path: "tone.webm".into(), + looping: false, + restart: false, + }; + collection.sources.insert("tone".into(), tone); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let listed = until(|| { + let listed = catalog.snapshot().ext.studio.sources.remove("tone"); + listed.filter(|listed| !listed.audio.renditions.is_empty()) + }) + .await; + assert!(listed.video.renditions.is_empty()); + assert_eq!(sources.get("tone").unwrap().track(), Some(Status::Live)); + }) + .await; + } } diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 8841e29..dea576d 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -11,10 +11,11 @@ //! [`crate::publish`]. use std::cell::{Cell, RefCell}; -use std::collections::{BTreeMap, VecDeque}; +use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::future::Future; use std::path::{Path, PathBuf}; use std::pin::Pin; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -29,6 +30,7 @@ use crate::Error; use crate::browser::Page; use crate::collection::{Collection, Source}; use crate::lut::Cube; +use crate::media::Clip; use crate::playout::{MAX_DELAY, Place, Speaker, VIDEO_DELAY, Writer}; use crate::publish::{Claim, Listing, Publish, Still}; use crate::replay::{MIN_SPEED, Take, Target}; @@ -146,6 +148,9 @@ pub enum Status { /// broadcast to come back; every other source stays stopped until edited /// or [restarted](Sources::restart). Failed(String), + /// Played to its end: a media file that does not loop, holding its last + /// picture, silent, until [played](Feed::play) again. + Ended, } /// What every source shares. @@ -172,6 +177,8 @@ pub struct Sources { context: Context, feeds: BTreeMap, mixes: BTreeMap, + /// The sources on program, as last set. + on_program: BTreeSet, } impl Sources { @@ -181,9 +188,28 @@ impl Sources { context, feeds: BTreeMap::new(), mixes: BTreeMap::new(), + on_program: BTreeSet::new(), } } + /// Note the sources on program now, by name, such as those a scene on air + /// shows. A media source set to restart plays from its start as it goes on + /// program. + pub fn set_on_program(&mut self, on_program: &BTreeSet) { + if *on_program == self.on_program { + return; + } + for name in on_program.difference(&self.on_program) { + let Some(feed) = self.feeds.get(name) else { continue }; + if matches!(feed.source, Source::Media { restart: true, .. }) + && let Err(err) = feed.play() + { + tracing::warn!(source = %name, %err, "cannot restart the source"); + } + } + self.on_program.clone_from(on_program); + } + /// What every source shares. pub fn context(&self) -> &Context { &self.context @@ -305,6 +331,10 @@ pub struct Feed { played: Playback, /// A replay source's next replay. plays: Option>, + /// Marked changed to play a media source from its start. + starts: Option>, + /// Set once a media source's file turns out to hold no sound. + silent: Arc, /// A broadcast source's playout clock. clock: Option, } @@ -347,11 +377,32 @@ impl Feed { } let play = Play { take, speed }; // Fails once the driver is gone, which leaves the reason in the status. - plays.send(Arc::new(Mutex::new(Some(play)))).map_err(|_| { - Error::Stopped(match self.status() { - Status::Failed(reason) => reason, - status => format!("{status:?}"), - }) + plays.send(Arc::new(Mutex::new(Some(play)))).map_err(|_| self.stopped()) + } + + /// Play a media source from its start, now, cutting off wherever it is. + /// One that ended plays again, and one that loops loops from there. + /// + /// Refused by any other kind of source, and by a media source that stopped. + pub fn play(&self) -> Result<(), Error> { + let Some(starts) = &self.starts else { + return Err(Error::Unsupported("only a media source plays from its start".into())); + }; + if starts.is_closed() { + return Err(self.stopped()); + } + // Cleared before the driver hears of it, so the next frame drawn, such + // as the first of a scene going on program, shows nothing rather than + // where the last play was, and the new first picture replaces it. + *self.picture.lock().unwrap() = Some(clear()); + starts.send(()).map_err(|_| self.stopped()) + } + + /// Why the driver is gone, from the reason it left in the status. + fn stopped(&self) -> Error { + Error::Stopped(match self.status() { + Status::Failed(reason) => reason, + status => format!("{status:?}"), }) } @@ -405,8 +456,9 @@ impl Feed { } /// The source's gain, mute, delay, and meter in every mix, if it has sound. + /// A media file without any has none once it opens. pub fn strip(&self) -> Option<&Strip> { - self.strip.as_ref() + self.strip.as_ref().filter(|_| !self.silent.load(Ordering::Relaxed)) } /// A browser source's page: how it paints, and interaction with it. @@ -439,6 +491,8 @@ pub struct Driver { track: Option, played: Playback, plays: Option>, + starts: Option>, + silent: Arc, clock: Option, } @@ -552,6 +606,11 @@ impl Driver { self.show(Picture::Overlay); std::future::pending().await } + Source::Media { path, looping, .. } => { + let starts = self.starts.clone().expect("a media source has starts"); + self.media(&self.context.dir.join(path), *looping, starts, listing) + .await + } Source::Broadcast { path } => self.broadcast(path).await, Source::Browser { url, width, height } => self.browser(url, *width, *height).await, Source::Replay => { @@ -599,6 +658,7 @@ impl Driver { Source::SystemAudio => return self.listen(Audio::System, listing).await, Source::Color { .. } | Source::Image { .. } + | Source::Media { .. } | Source::Broadcast { .. } | Source::Browser { .. } | Source::Overlay { .. } @@ -681,7 +741,6 @@ impl Driver { /// Fail the source's own track with `err`, ending it, while the source /// keeps drawing. - #[cfg(feature = "capture")] fn fail_encoding(&self, track: &mut Option>, err: Error) { tracing::warn!(source = %self.name, %err, "cannot encode the source"); self.fail_track(Status::Failed(err.to_string())); @@ -964,7 +1023,7 @@ impl Driver { let Some(strip) = self.strip.clone() else { return Ok(()); }; - let bus = self.context.bus.config().clone(); + let bus = self.context.bus.config(); let mut options = moq_audio::decode::Options::new(); options.start = moq_audio::decode::Start::Latest; options.delay = moq_audio::decode::Delay { @@ -976,9 +1035,7 @@ impl Driver { options.output.layout = Some(bus.layout); let mut consumer = moq_audio::decode::Consumer::new(broadcast, &config, name, options).await?; - let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); - input.format = moq_audio::Format::F32; - let channel = self.context.bus.channel(input, strip)?; + let channel = self.channel(strip)?; let mut writer = Writer::new(channel, bus.sample_rate, bus.layout.channels()); let speaker = RefCell::new(Speaker::new(clock.timeline.clone(), &clock.changed)); let stride = bus.layout.channels() as usize * size_of::(); @@ -1096,13 +1153,8 @@ impl Driver { let Some((name, config)) = &take.video else { return Ok(()); }; - let mut options = moq_video::decode::Options::new(); - options.decoder.output = self.context.output; - options.delay = REPLAY; - let mut consumer = moq_video::decode::Consumer::new(&broadcast, config, name, options).await?; - // Opened at the first picture, which says its size. - let mut track: Option = None; - let mut opened = false; + let mut consumer = moq_video::decode::Consumer::new(&broadcast, config, name, self.pictures()).await?; + let mut own = Own::new(listing, begin); while let Some(read) = consumer.read().await? { match read { // The group in progress at the start decodes from its @@ -1112,26 +1164,7 @@ impl Driver { moq_video::decode::Read::Frame(frame) if frame.timestamp >= take.end => break, moq_video::decode::Read::Frame(mut frame) => { kio::clock::sleep_until(at(frame.timestamp)).await; - if let Some(listing) = listing.filter(|_| !std::mem::replace(&mut opened, true)) { - let framerate = config - .framerate - .filter(|rate| rate.is_finite() && *rate >= 1.0) - .and_then(|rate| moq_video::FrameRate::new(rate.round() as u32, 1).ok()) - .unwrap_or_else(|| moq_video::FrameRate::new(30, 1).expect("a valid frame rate")); - let video = - crate::publish::Video::new(listing, frame.size(), framerate, frame.surface.color()); - track = self.publish(video.await); - } - if let Some(video) = &mut track { - // Stamped when it shows, on a timeline of the replay's own. - frame.timestamp = moq_net::Timestamp::from_micros(begin.elapsed().as_micros() as u64) - .map_err(moq_video::Error::from)?; - if let Err(err) = video.encode(&mut frame) { - tracing::warn!(source = %self.name, %err, "cannot encode the replay"); - self.fail_track(Status::Failed(err.to_string())); - track.take().expect("encoding").abort(&err); - } - } + self.encode(&mut own, &mut frame, config).await?; self.show(Picture::Frame(frame)); } // A stretch the broadcast was offline, paused, or lost @@ -1141,12 +1174,10 @@ impl Driver { | moq_video::decode::Read::End { end: timestamp } => { kio::clock::sleep_until(at(timestamp.clamp(take.start, take.end))).await; *self.picture.lock().unwrap() = Some(clear()); - if let Some(video) = &mut track + if let Some(video) = &mut own.track && let Err(err) = video.discontinuity() { - tracing::warn!(source = %self.name, %err, "cannot mark a gap in the replay"); - self.fail_track(Status::Failed(err.to_string())); - track.take().expect("encoding").abort(&err); + self.fail_encoding(&mut own.track, err); } } } @@ -1159,16 +1190,9 @@ impl Driver { let (Some((name, config)), Some(strip), true) = (&take.audio, &self.strip, play.speed == 1.0) else { return Ok(()); }; - let bus = self.context.bus.config().clone(); - let mut options = moq_audio::decode::Options::new(); - options.delay = REPLAY; - options.output.format = moq_audio::Format::F32; - options.output.sample_rate = Some(bus.sample_rate); - options.output.layout = Some(bus.layout); - let mut consumer = moq_audio::decode::Consumer::new(&broadcast, config, name.clone(), options).await?; - let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); - input.format = moq_audio::Format::F32; - let mut channel = self.context.bus.channel(input, strip.clone())?; + let mut consumer = + moq_audio::decode::Consumer::new(&broadcast, config, name.clone(), self.sounds()).await?; + let mut channel = self.channel(strip.clone())?; while let Some(read) = consumer.read().await? { match read { moq_audio::decode::Read::Frame(frame) if frame.timestamp < take.start => {} @@ -1198,10 +1222,365 @@ impl Driver { () = kio::clock::sleep_until(at(take.end)) => Ok(()), } } + + /// Play the media file at `path` from its start, and again each time + /// `starts` changes, cutting off the play under way. At its end it starts + /// over when `looping`, or holds its last picture, silent, until the next. + /// + /// With `listing`, the pictures are also encoded onto the source's own + /// track as they show, while it is used, on one timeline across plays. + async fn media( + &self, + path: &Path, + looping: bool, + mut starts: watch::Receiver<()>, + listing: Option<&Listing<'_>>, + ) -> Result<(), Error> { + let clip = Clip::open(path)?; + // Without sound it has no strip to mix or meter. + self.silent.store(clip.audio.is_none(), Ordering::Relaxed); + let mut sound = match (&clip.audio, &self.strip) { + (Some(_), Some(strip)) => { + // Its track opens now, the pictures' at the first picture. + let track = listing.and_then(|listing| { + let bus = self.context.bus.config(); + let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); + input.format = moq_audio::Format::F32; + self.publish(crate::publish::Audio::new(listing, input)) + }); + let channel = Some(self.channel(strip.clone())?); + Some(Sound { + channel, + track, + ended: None, + }) + } + _ => None, + }; + let mut own = Own::new(listing, kio::clock::now()); + // The last channel on a strip to close drops what it has not played + // yet, so a loop keeps its channel, sounding on, while a play after a + // restart or an end takes a new channel, cutting off what the play + // before still had queued. The old channel goes first: one left on the + // strip would keep the strip's queue, delay included, for the new. + // Its pictures start a group there too, a cut to the clip's start. + let restart = |sound: &mut Option>, own: &mut Own<'_, '_>| -> Result<(), Error> { + if let (Some(sound), Some(strip)) = (sound, &self.strip) { + drop(sound.channel.take()); + sound.channel = Some(self.channel(strip.clone())?); + sound.ended = None; + } + if let Some(video) = &mut own.track + && let Err(err) = video.cut() + { + self.fail_encoding(&mut own.track, err); + } + Ok(()) + }; + // Where a loop's next play starts: where the last one ended, so loops + // keep exact time whatever opening the decoders takes. + let mut next = None; + loop { + let restarted = tokio::select! { + result = self.play_clip(&clip, sound.as_mut(), &mut own, next) => { + next = Some(result?); + false + } + changed = starts.changed() => { + // The feed is gone, which stops this driver anyway. + if changed.is_err() { + return std::future::pending().await; + } + true + } + }; + if restarted { + restart(&mut sound, &mut own)?; + next = None; + continue; + } + if looping { + continue; + } + set(&self.status, Status::Ended); + self.ended(&mut starts, &mut own).await; + restart(&mut sound, &mut own)?; + next = None; + } + } + + /// Wait out a media source that ended until `starts` changes. Its own + /// track gets no more pictures, so its encoder is released here once + /// nothing uses the source, giving the encode session back. + async fn ended(&self, starts: &mut watch::Receiver<()>, own: &mut Own<'_, '_>) { + loop { + let settled = async { + match &own.track { + Some(video) if video.is_open() => video.settle().await, + _ => std::future::pending().await, + } + }; + tokio::select! { + changed = starts.changed() => { + // The feed is gone, which stops this driver anyway. + if changed.is_err() { + std::future::pending::<()>().await; + } + return; + } + () = settled => { + if let Some(video) = &mut own.track + && let Err(err) = video.release() + { + self.fail_encoding(&mut own.track, err); + } + } + } + } + } + + /// Play `clip` once, each picture and sound when its moment comes, the + /// first at `begin`, or at once. Returns where it ends, once the last has + /// played for as long as it lasts: a sound for its samples, and a picture + /// for as long as the one before it. A clip that decodes to nothing fails. + async fn play_clip( + &self, + clip: &Clip, + mut sound: Option<&mut Sound<'_>>, + own: &mut Own<'_, '_>, + begin: Option, + ) -> Result { + set(&self.status, Status::Live); + let broadcast = clip.consume(); + let mut video = match &clip.video { + Some((name, config)) => { + let consumer = + moq_video::decode::Consumer::new(&broadcast, config, name.clone(), self.pictures()).await?; + Some((consumer, config)) + } + None => None, + }; + let mut audio = match (&clip.audio, &sound) { + (Some((name, config)), Some(_)) => { + Some(moq_audio::decode::Consumer::new(&broadcast, config, name.clone(), self.sounds()).await?) + } + _ => None, + }; + + // The first of each is decoded before the clock starts, so the earlier + // of them is due at once. + let mut picture = match &mut video { + Some((consumer, _)) => next_picture(consumer).await?, + None => None, + }; + let mut samples = match &mut audio { + Some(consumer) => next_samples(consumer).await?, + None => None, + }; + let first = [ + picture.as_ref().map(|frame| Duration::from(frame.timestamp)), + samples.as_ref().map(|frame| Duration::from(frame.timestamp)), + ]; + let Some(start) = first.into_iter().flatten().min() else { + // Its tracks hold no frame that decodes, which a loop would only + // play again at once, forever. + return Err(Error::Media("the file decodes to no picture or sound".into())); + }; + // The pictures' track opens before the clock starts, so the encoder's + // probe holds back neither the first picture nor the sound beside it. + if let (Some(frame), Some((_, config))) = (&picture, &video) { + self.open(own, frame, config).await; + } + let begin = begin.unwrap_or_else(kio::clock::now); + let at = |timestamp: moq_net::Timestamp| begin + Duration::from(timestamp).saturating_sub(start); + let end = std::cell::Cell::new(begin); + let last = |until: Instant| end.set(end.get().max(until)); + + let pictures = async { + let Some((consumer, config)) = &mut video else { + return Ok(()); + }; + let mut previous = None; + while let Some(mut frame) = picture.take() { + let due = at(frame.timestamp); + kio::clock::sleep_until(due).await; + last(due + previous.map_or(Duration::ZERO, |previous| due - previous)); + previous = Some(due); + self.encode(own, &mut frame, config).await?; + self.show(Picture::Frame(frame)); + picture = next_picture(consumer).await?; + } + Ok::<_, Error>(()) + }; + let sounds = async { + let (Some(consumer), Some(sound)) = (&mut audio, &mut sound) else { + return Ok(()); + }; + let bus = self.context.bus.config(); + let bytes = std::mem::size_of::() * bus.layout.channels() as usize; + // The track's timeline runs on by its samples only from a sound + // that starts right where the last one ended, as a loop's does + // when its sound is its longest track: any other start, such as + // after a loop's silence while its pictures play on, is a gap. + let mut first = true; + while let Some(frame) = samples.take() { + let due = at(frame.timestamp); + kio::clock::sleep_until(due).await; + let length = (frame.data.len() / bytes) as u64 * 1_000_000_000 / u64::from(bus.sample_rate); + last(due + Duration::from_nanos(length)); + let gap = std::mem::take(&mut first) && sound.ended != Some(due); + sound.ended = Some(due + Duration::from_nanos(length)); + sound + .channel + .as_mut() + .expect("a channel while playing") + .write(&frame.data)?; + if let Some(track) = &mut sound.track { + let samples: Vec = bytemuck::pod_collect_to_vec(&frame.data); + if let Err(err) = track.write(&samples, due, gap) { + tracing::warn!(source = %self.name, %err, "cannot encode the source"); + self.fail_track(Status::Failed(err.to_string())); + sound.track.take().expect("encoding").abort(&err); + } + } + samples = next_samples(consumer).await?; + } + Ok::<_, Error>(()) + }; + tokio::try_join!(pictures, sounds)?; + kio::clock::sleep_until(end.get()).await; + Ok(end.get()) + } + + /// Open the source's own track for pictures like `frame`, which says their + /// size, of the rendition `config` describes, unless it was tried already. + async fn open(&self, own: &mut Own<'_, '_>, frame: &moq_video::Frame, config: &VideoConfig) { + if let Some(listing) = own.listing.filter(|_| !own.opened) { + let video = crate::publish::Video::new(listing, frame.size(), framerate(config), frame.surface.color()); + own.track = self.publish(video.await); + // Only once it has, so a play cut off while it opens leaves the + // next picture to open it. + own.opened = true; + } + } + + /// Encode a played picture onto the source's own track, opened at the + /// first picture if not before. Stamped when it shows, on the timeline + /// `own` keeps. + async fn encode( + &self, + own: &mut Own<'_, '_>, + frame: &mut moq_video::Frame, + config: &VideoConfig, + ) -> Result<(), Error> { + self.open(own, frame, config).await; + if let Some(video) = &mut own.track { + frame.timestamp = moq_net::Timestamp::from_micros(own.origin.elapsed().as_micros() as u64) + .map_err(moq_video::Error::from)?; + if let Err(err) = video.encode(frame) { + self.fail_encoding(&mut own.track, err); + } + } + Ok(()) + } + + /// How a replay or media source decodes pictures: from the first group, + /// for the compositor's device. + fn pictures(&self) -> moq_video::decode::Options { + let mut options = moq_video::decode::Options::new(); + options.decoder.output = self.context.output; + options.delay = REPLAY; + options + } + + /// How a replay or media source decodes sound: from the first group, as + /// the bus mixes it. + fn sounds(&self) -> moq_audio::decode::Options { + let bus = self.context.bus.config(); + let mut options = moq_audio::decode::Options::new(); + options.delay = REPLAY; + options.output.format = moq_audio::Format::F32; + options.output.sample_rate = Some(bus.sample_rate); + options.output.layout = Some(bus.layout); + options + } + + /// A channel into the bus for decoded sound, as [`sounds`](Self::sounds) + /// and broadcasts decode it, through `strip`. + fn channel(&self, strip: Strip) -> Result { + let bus = self.context.bus.config(); + let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); + input.format = moq_audio::Format::F32; + Ok(self.context.bus.channel(input, strip)?) + } +} + +/// A media source's sound: the bus channel it plays into, taken anew on a +/// restart, and its own track. +struct Sound<'a> { + channel: Option, + track: Option>, + /// When the last sound written ends, or `None` after a break, such as a + /// restart, so the next sound's track starts its timeline afresh. + ended: Option, +} + +/// A played source's own track, opened at its first picture, and the timeline +/// its pictures are stamped on: from `origin`, when they show. +struct Own<'a, 'b> { + listing: Option<&'a Listing<'b>>, + track: Option>, + /// Whether the track was opened, or failed to, so it is tried once. + opened: bool, + origin: Instant, +} + +impl<'a, 'b> Own<'a, 'b> { + fn new(listing: Option<&'a Listing<'b>>, origin: Instant) -> Self { + Self { + listing, + track: None, + opened: false, + origin, + } + } } -/// How a replay reads its take: from the first group, where the take's tracks -/// declare they start, skipping nothing as late, since a take is history. +/// The next picture `consumer` decodes, passing over gaps, or `None` at the +/// end of its track. +async fn next_picture(consumer: &mut moq_video::decode::Consumer) -> Result, Error> { + while let Some(read) = consumer.read().await? { + if let moq_video::decode::Read::Frame(frame) = read { + return Ok(Some(frame)); + } + } + Ok(None) +} + +/// The next sound `consumer` decodes, passing over gaps, or `None` at the end +/// of its track. +async fn next_samples(consumer: &mut moq_audio::decode::Consumer) -> Result, Error> { + while let Some(read) = consumer.read().await? { + if let moq_audio::decode::Read::Frame(frame) = read { + return Ok(Some(frame)); + } + } + Ok(None) +} + +/// The frame rate a rendition's own track is encoded at: its catalog's, or 30 +/// fps when it names none. +fn framerate(config: &VideoConfig) -> moq_video::FrameRate { + config + .framerate + .filter(|rate| rate.is_finite() && *rate >= 1.0) + .and_then(|rate| moq_video::FrameRate::new(rate.round() as u32, 1).ok()) + .unwrap_or_else(|| moq_video::FrameRate::new(30, 1).expect("a valid frame rate")) +} + +/// How a replay or media source reads its broadcast: from the first group, +/// where its tracks declare they start, skipping nothing as late, since all of +/// it is history. const REPLAY: moq_net::track::Delay = moq_net::track::Delay { min: Some(Duration::ZERO), max: None, @@ -1213,7 +1592,7 @@ const REPLAY: moq_net::track::Delay = moq_net::track::Delay { /// One served from another broadcast would need that broadcast's own /// subscription, which a source does not open, so a catalog offering only /// those is refused rather than shown as black or played as silence. -fn pick<'a, C: Reference + Clone + 'a>( +pub(crate) fn pick<'a, C: Reference + Clone + 'a>( renditions: impl Iterator, ) -> Result, Error> { let mut foreign = None; @@ -1259,6 +1638,14 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { } _ => (None, None), }; + let (starts, started) = match source { + Source::Media { .. } => { + let (starts, started) = watch::channel(()); + (Some(starts), Some(started)) + } + _ => (None, None), + }; + let silent = Arc::::default(); let feed = Feed { source: source.clone(), @@ -1270,6 +1657,8 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { track: track.as_ref().map(|track| track.status.consume()), played: played.clone(), plays, + starts, + silent: silent.clone(), clock: clock.as_ref().map(timeline::Producer::consume), }; let driver = Driver { @@ -1284,14 +1673,17 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { track, played, plays: replays, + starts: started, + silent, clock, }; (feed, driver) } /// Whether `source` publishes its own track: a picture the studio captures or -/// loads itself, before the scene's filters, or the sound of a microphone or -/// system audio as captured, before its gain and mute. +/// loads itself, before the scene's filters, or the sound of a microphone, +/// system audio, or media file as captured or decoded, before its gain and +/// mute. /// /// A broadcast is already published by its own publisher, and a clone shows /// its source's track, its filters carried by the layout rather than encoded @@ -1307,6 +1699,7 @@ fn publishes(source: &Source) -> bool { | Source::SystemAudio | Source::Color { .. } | Source::Image { .. } + | Source::Media { .. } | Source::Replay => true, Source::Broadcast { .. } | Source::Browser { .. } @@ -2239,3 +2632,219 @@ mod tests { .await; } } + +#[cfg(test)] +mod media_tests { + use super::test::{context, local, start, until, until_time}; + use super::*; + use crate::media::tests::{CLIP, SILENT, TONE}; + + /// A collection of media sources, each named for its file in `dir`, which + /// is written with its bytes, and whether it loops and restarts. + fn media(dir: &Path, files: &[(&str, &[u8], bool, bool)]) -> Collection { + let mut collection = Collection::new(); + for (name, bytes, looping, restart) in files { + std::fs::write(dir.join(name), bytes).unwrap(); + let source = Source::Media { + path: name.into(), + looping: *looping, + restart: *restart, + }; + collection.sources.insert(name.to_string(), source); + } + collection + } + + /// The next picture `feed` shows, passing over the clear a restart puts up, + /// and when it showed. + async fn shown(feed: &Feed) -> (moq_net::Timestamp, Instant) { + let timestamp = until(|| match feed.take() { + Some(Picture::Frame(frame)) => Some(frame.timestamp), + Some(Picture::Rgba { pixels, .. }) if pixels == [0; 4] => None, + Some(_) => panic!("not a decoded picture"), + None => None, + }) + .await; + (timestamp, kio::clock::now()) + } + + /// Wait until `name` reads as `status`, without moving the clock. + async fn until_status(sources: &Sources, name: &str, status: Status) { + until(|| (sources.get(name).unwrap().status() == status).then_some(())).await; + } + + /// A file shows each picture at its moment, the first at once. Past its + /// end it holds the last picture, reads as ended, and plays again from + /// its start when asked. + #[tokio::test(start_paused = true)] + async fn a_media_file_plays_once_and_holds_its_end() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, _origin) = context(dir.path()); + let mut sources = Sources::new(context); + let collection = media(dir.path(), &[("clip.mkv", CLIP, false, false)]); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let feed = sources.get("clip.mkv").unwrap(); + + let (first, begin) = shown(feed).await; + assert_eq!(feed.status(), Status::Live); + for index in 1..10u64 { + let into = Duration::from_millis(index * 100); + // Within the timer wheel's millisecond of when the test saw the first. + until_time(begin + into - Duration::from_millis(2)).await; + assert!(feed.take().is_none(), "picture {index} before its moment"); + until_time(begin + into).await; + let (timestamp, _) = shown(feed).await; + assert_eq!(Duration::from(timestamp) - Duration::from(first), into); + } + + // The last picture lasts as long as the one before it, and the + // sound runs a little past it. + until_time(begin + Duration::from_millis(990)).await; + assert_eq!(feed.status(), Status::Live); + until_time(begin + Duration::from_millis(1_100)).await; + until_status(&sources, "clip.mkv", Status::Ended).await; + assert!(feed.take().is_none(), "the last picture holds"); + + feed.play().unwrap(); + let cleared = feed.take(); + assert!( + matches!(cleared, Some(Picture::Rgba { pixels, .. }) if pixels == [0; 4]), + "the held picture is cleared at once" + ); + let (again, _) = shown(feed).await; + assert_eq!(again, first, "from the start"); + assert_eq!(feed.status(), Status::Live); + }) + .await; + } + + /// A looping file starts over at its end, never reading as ended. + #[tokio::test(start_paused = true)] + async fn a_looping_media_file_starts_over() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, _origin) = context(dir.path()); + let mut sources = Sources::new(context); + let collection = media(dir.path(), &[("clip.mkv", CLIP, true, false)]); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let feed = sources.get("clip.mkv").unwrap(); + + let (first, begin) = shown(feed).await; + until_time(begin + Duration::from_millis(950)).await; + let (last, _) = shown(feed).await; + assert_eq!(Duration::from(last) - Duration::from(first), Duration::from_millis(900)); + until_time(begin + Duration::from_millis(998)).await; + assert!(feed.take().is_none(), "the start again before the end"); + until_time(begin + Duration::from_millis(1_100)).await; + let (again, _) = shown(feed).await; + assert_eq!(again, first); + assert_eq!(feed.status(), Status::Live); + }) + .await; + } + + /// A file set to restart plays from its start as it goes on program, not + /// again while it stays on, and one not set to restart never does. + #[tokio::test(start_paused = true)] + async fn a_media_file_restarts_going_on_program() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, _origin) = context(dir.path()); + let mut sources = Sources::new(context); + let collection = media( + dir.path(), + &[("intro.mkv", CLIP, false, true), ("still.mkv", CLIP, false, false)], + ); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let (first, begin) = shown(sources.get("intro.mkv").unwrap()).await; + until_time(begin + Duration::from_millis(1_100)).await; + until_status(&sources, "intro.mkv", Status::Ended).await; + until_status(&sources, "still.mkv", Status::Ended).await; + for name in ["intro.mkv", "still.mkv"] { + sources.get(name).unwrap().take(); + } + + let both: BTreeSet = ["intro.mkv".into(), "still.mkv".into()].into(); + sources.set_on_program(&both); + let (again, begin) = shown(sources.get("intro.mkv").unwrap()).await; + assert_eq!(again, first); + assert_eq!(sources.get("still.mkv").unwrap().status(), Status::Ended); + + until_time(begin + Duration::from_millis(1_100)).await; + until_status(&sources, "intro.mkv", Status::Ended).await; + sources.set_on_program(&both); + assert_eq!( + sources.get("intro.mkv").unwrap().status(), + Status::Ended, + "still on program" + ); + + sources.set_on_program(&BTreeSet::new()); + sources.set_on_program(&both); + until_status(&sources, "intro.mkv", Status::Live).await; + assert!(sources.get("still.mkv").unwrap().take().is_none()); + }) + .await; + } + + /// A file of sound alone shows nothing, and goes silent at its end. One of + /// pictures alone has no sound for a mix to play. + #[tokio::test(start_paused = true)] + async fn an_audio_file_ends_silent() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, _origin) = context(dir.path()); + let mut sources = Sources::new(context); + let files = [("tone.webm", TONE, false, false), ("silent.mkv", SILENT, false, false)]; + let collection = media(dir.path(), &files); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let program = crate::Mixes::DEFAULT; + assert!(sources.plays(program, "tone.webm")); + assert!(!sources.plays(program, "silent.mkv")); + assert!(sources.get("silent.mkv").unwrap().strip().is_none()); + assert_eq!(sources.get("tone.webm").unwrap().status(), Status::Live); + until_time(kio::clock::now() + Duration::from_millis(600)).await; + until_status(&sources, "tone.webm", Status::Ended).await; + assert!(sources.get("tone.webm").unwrap().take().is_none()); + }) + .await; + } + + /// A file the studio cannot read fails its source, saying why, and only a + /// media source plays from its start. + #[tokio::test] + async fn a_media_file_it_cannot_read_fails() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, _origin) = context(dir.path()); + let mut sources = Sources::new(context); + let mut collection = media(dir.path(), &[("notes.txt", b"not a video", false, false)]); + let red = Source::Color { + color: crate::Rgba([255, 0, 0, 255]), + }; + collection.sources.insert("red".into(), red); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + + let feed = sources.get("notes.txt").unwrap(); + let reason = until(|| match feed.status() { + Status::Failed(reason) => Some(reason), + _ => None, + }) + .await; + assert!( + reason.contains("notes.txt") && reason.contains("not a container"), + "{reason}" + ); + assert!(matches!(feed.play(), Err(Error::Stopped(_)))); + let red = sources.get("red").unwrap(); + assert!(matches!(red.play(), Err(Error::Unsupported(_)))); + }) + .await; + } +} diff --git a/rs/moq-scene/src/test_data/clip.mkv b/rs/moq-scene/src/test_data/clip.mkv new file mode 100644 index 0000000..3e30bd4 Binary files /dev/null and b/rs/moq-scene/src/test_data/clip.mkv differ diff --git a/rs/moq-scene/src/test_data/clip.mp4 b/rs/moq-scene/src/test_data/clip.mp4 new file mode 100644 index 0000000..05d12d6 Binary files /dev/null and b/rs/moq-scene/src/test_data/clip.mp4 differ diff --git a/rs/moq-scene/src/test_data/silent.mkv b/rs/moq-scene/src/test_data/silent.mkv new file mode 100644 index 0000000..49c4846 Binary files /dev/null and b/rs/moq-scene/src/test_data/silent.mkv differ diff --git a/rs/moq-scene/src/test_data/tone.webm b/rs/moq-scene/src/test_data/tone.webm new file mode 100644 index 0000000..779704a Binary files /dev/null and b/rs/moq-scene/src/test_data/tone.webm differ diff --git a/rs/moq-scene/src/test_data/uneven.mkv b/rs/moq-scene/src/test_data/uneven.mkv new file mode 100644 index 0000000..471ec38 Binary files /dev/null and b/rs/moq-scene/src/test_data/uneven.mkv differ diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index a4858c1..3a6ff0b 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -23,7 +23,7 @@ //! edits to them settle, and takes or holds an outside edit to the scene file. //! - The UI runs on the main thread and edits [`State`] under its lock. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex, mpsc}; @@ -34,7 +34,7 @@ use bytes::Bytes; use kio::clock::Instant; use moq_net::bandwidth; use moq_scene::publish::{Encoding, Sessions}; -use moq_scene::source::{self, Driver, Sources}; +use moq_scene::source::{self, Driver, Sources, Status}; use moq_scene::{Canvas, Collection, Compositor, Draft, Edit, History, Scaler, wgpu}; use moq_video::render::export::{Exporter, Pending, Route}; @@ -203,6 +203,29 @@ impl State { Ok(()) } + /// The sources the program shows: placed or masking in a scene on air, + /// including one leaving in a transition, or in the top layer over them, a + /// clone by the source it shows. + fn shown(&self) -> BTreeSet { + let collection = &self.collection; + let scenes = self.compositor.on_air().filter_map(|scene| collection.scene(scene)); + let items = scenes + .flat_map(|scene| &scene.items) + .chain(&collection.top) + .filter(|item| item.visible); + // A mask's picture shapes what the program draws too. + let masks = |item| { + collection.filters(item).filter_map(|filter| match filter { + moq_scene::Filter::Mask { source, .. } => Some(source), + _ => None, + }) + }; + items + .flat_map(|item| std::iter::once(&item.source).chain(masks(item))) + .map(|source| collection.capture(source).to_string()) + .collect() + } + /// Restart the source called `name`, as an edit would, such as to reload a /// web page. pub fn restart(&mut self, name: &str) { @@ -512,30 +535,20 @@ impl Program { } } - /// Encode the local sources on program, from `state`, whether or not - /// anything reads their tracks, so a replay of one has its past: those a - /// scene on air shows, and the microphones and system audio the program - /// mix plays. A replay source's track is its pictures, so being heard - /// alone does not put it on program. - fn on_program(&self, state: &State) { - let collection = &state.collection; - let scenes = state.compositor.on_air().filter_map(|scene| collection.scene(scene)); - let shown = scenes - .flat_map(|scene| &scene.items) - .chain(&collection.top) - .filter(|item| item.visible) - .map(|item| collection.capture(&item.source).to_string()); + /// Encode the local sources' tracks on program, whether or not anything + /// reads them, so a replay of one has its past: the pictures `shown` in a + /// scene on air, and the sound the program mix plays. A media file heard + /// and not shown encodes only its sound. + fn on_program(&self, state: &State, shown: BTreeSet) { + let program = &state.collection.mixes.program; let heard = state .sources .iter() - .filter(|(name, feed)| { - matches!( - feed.source(), - moq_scene::Source::Microphone { .. } | moq_scene::Source::SystemAudio - ) && state.sources.plays(&collection.mixes.program, name) - }) - .map(|(name, _)| name.to_string()); - self.sources.set_on_program(shown.chain(heard).collect()); + .filter(|(name, _)| state.sources.plays(program, name)) + .map(|(name, _)| name.to_string()) + .collect(); + self.sources + .set_on_program(moq_scene::publish::OnProgram { shown, heard }); } } @@ -898,30 +911,37 @@ impl Engine { // A stalled picture matters where the program draws it: a scene on air, // including one leaving in a transition, or the top layer over them. // Stalled sound matters only if the mix on program plays it, not one - // only listened to. - let stalled = { + // only listened to. A source that stopped matters either way. + let (stalled, stopped) = { let state = self.state.lock().unwrap(); - let shown: Vec<&str> = state - .compositor - .on_air() - .filter_map(|name| state.collection.scene(name)) - .flat_map(|scene| &scene.items) - .chain(&state.collection.top) - .filter(|item| item.visible) - .map(|item| state.collection.capture(&item.source)) - .collect(); - state + let shown = state.shown(); + let heard = |name: &str| state.sources.plays(&state.collection.mixes.program, name); + let stalled = state .sources .iter() .filter_map(|(name, feed)| { let stall = feed.stalled(now)?; let matters = match stall { - source::Stall::Picture(_) => shown.contains(&name), - source::Stall::Sound(_) => state.sources.plays(&state.collection.mixes.program, name), + source::Stall::Picture(_) => shown.contains(name), + source::Stall::Sound(_) => heard(name), }; matters.then(|| (name.to_string(), stall)) }) - .collect() + .collect(); + let stopped = state + .sources + .iter() + .filter(|(name, _)| shown.contains(*name) || heard(name)) + .filter_map(|(name, feed)| { + // A picture that fails to draw fails the source as the Sources panel shows it. + let status = match state.compositor.error(name) { + Some(err) => Status::Failed(err.to_string()), + None => feed.status(), + }; + matches!(status, Status::Ended | Status::Failed(_)).then(|| (name.to_string(), status)) + }) + .collect(); + (stalled, stopped) }; let canvas = &stats.renditions[0]; health::Local { @@ -934,6 +954,7 @@ impl Engine { encoder: stats.encoder(), failed: stats.error(), stalled, + stopped, } } @@ -992,11 +1013,15 @@ impl Engine { } self.pick_mixes(state); } + // Before drawing, so a media file that restarts going on program + // is cleared in this frame rather than showing where it ended. + let shown = state.shown(); + state.sources.set_on_program(&shown); if let Err(err) = state.compositor.render(&state.collection, &state.sources, tick) { tracing::error!(%err, "cannot draw the canvas"); continue; } - self.program.on_program(state); + self.program.on_program(state, shown); (state.compositor.program().clone(), state.compositor.canvas()) }; self.stats.rendered.fetch_add(1, Ordering::Relaxed); diff --git a/rs/moq-studio/src/health.rs b/rs/moq-studio/src/health.rs index 5cb30a5..410d283 100644 --- a/rs/moq-studio/src/health.rs +++ b/rs/moq-studio/src/health.rs @@ -21,7 +21,7 @@ pub use hang::health::Verdict; use hang::health::{self, Health, Input, Observer, Thresholds}; use kio::clock::Instant; use moq_net::bandwidth::Rate; -use moq_scene::source::Stall; +use moq_scene::source::{Stall, Status}; /// How often the health is graded: as often as reporters write. pub const INTERVAL: Duration = Duration::from_secs(1); @@ -214,6 +214,8 @@ pub struct Local { pub failed: Option, /// Sources the program shows or plays that stopped delivering, and what. pub stalled: Vec<(String, Stall)>, + /// Sources the program shows or plays that ended or failed. + pub stopped: Vec<(String, Status)>, } /// What the studio publishes, for grading its delivery. @@ -629,6 +631,18 @@ fn problems(facts: &Facts<'_>) -> Vec { ), ); } + for (name, status) in &local.stopped { + match status { + Status::Ended => add( + Verdict::Degraded, + format!( + "Source `{name}` played to its end, so the program holds its last picture and plays no sound for it." + ), + ), + Status::Failed(reason) => add(Verdict::Unhealthy, format!("Source `{name}` failed: {reason}.")), + _ => {} + } + } if let Some(drops) = facts.drops && local.encoding @@ -1295,6 +1309,10 @@ mod tests { ("camera".into(), Stall::Picture(Duration::from_secs(3))), ("mic".into(), Stall::Sound(Duration::from_secs(2))), ], + stopped: vec![ + ("intro".into(), Status::Ended), + ("music".into(), Status::Failed("the file is gone".into())), + ], ..Default::default() }; let uplink = Uplink { @@ -1311,6 +1329,8 @@ mod tests { "The relay https://a is unreachable (timed out), so nothing reaches viewers until a relay reconnects.", "Source `camera` stalled: no new picture for 3 s, so the program repeats its last one.", "Source `mic` stalled: no new sound for 2 s, so the program plays silence for it.", + "Source `music` failed: the file is gone.", + "Source `intro` played to its end, so the program holds its last picture and plays no sound for it.", "The relay is far away: a round trip to it takes 400 ms.", ] ); diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index a8802b4..eff996d 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1310,6 +1310,7 @@ impl Panels { let mut remove = None; let mut clone = None; let mut reload = None; + let mut play = None; let now = kio::clock::now(); egui::ScrollArea::vertical().show(ui, |ui| { @@ -1393,11 +1394,23 @@ impl Panels { { clone = Some(name.clone()); } - if matches!(source, Source::Browser { .. }) + if matches!(source, Source::Media { .. }) && ui - .small_button("Reload") - .on_hover_text("Restart the page, keeping the cookies and logins every page shares") + .small_button("Play") + .on_hover_text("Play the file from its start") .clicked() + { + play = Some(name.clone()); + } + let reloads = match source { + Source::Browser { .. } => { + Some("Restart the page, keeping the cookies and logins every page shares") + } + Source::Media { .. } => Some("Read the file again, such as after it changed"), + _ => None, + }; + if let Some(hover) = reloads + && ui.small_button("Reload").on_hover_text(hover).clicked() { reload = Some(name.clone()); } @@ -1424,6 +1437,12 @@ impl Panels { if let Some(name) = reload { state.restart(&name); } + if let Some(name) = play + && let Some(feed) = state.sources.get(&name) + { + let played = feed.play().with_context(|| format!("cannot play `{name}`")); + self.report(played.map(|()| format!("playing `{name}` from its start"))); + } let result = if let Some(name) = remove { // Removing the items that show it moves the ones after them. self.deselect_item(); @@ -3261,6 +3280,10 @@ fn describe(source: &Source) -> String { Source::Microphone { id } => format!("microphone: {}", or_default(id)), Source::SystemAudio => "system audio".into(), Source::Image { path } => format!("image: {}", path.display()), + Source::Media { path, looping, .. } => { + let ending = if *looping { "looping" } else { "once" }; + format!("media: {} ({ending})", path.display()) + } Source::Color { color } => format!("color: {color}"), Source::Broadcast { path } => format!("broadcast: {path}"), Source::Browser { url, width, height } => format!("browser: {url} ({width}x{height})"), @@ -3276,6 +3299,7 @@ fn status(ui: &mut Ui, status: &Status) { Status::Starting => ui.weak("starting"), Status::Live => ui.colored_label(HEALTHY, "live"), Status::Offline => ui.colored_label(DEGRADED, "offline"), + Status::Ended => ui.colored_label(DEGRADED, "ended"), Status::Failed(reason) => ui .colored_label(ui.visuals().error_fg_color, "failed") .on_hover_text(reason), @@ -3323,6 +3347,10 @@ struct AddSource { /// A browser source's page size, or an overlay's. width: u32, height: u32, + /// Whether a media file loops, and plays from its start each time it goes + /// on program. + looping: bool, + restart: bool, } impl Default for AddSource { @@ -3335,6 +3363,8 @@ impl Default for AddSource { color: Color32::WHITE, width: canvas.width, height: canvas.height, + looping: false, + restart: true, } } } @@ -3344,6 +3374,7 @@ enum Kind { #[default] Color, Image, + Media, Broadcast, Camera, Display, @@ -3357,9 +3388,10 @@ enum Kind { } impl Kind { - const ALL: [Self; 12] = [ + const ALL: [Self; 13] = [ Self::Color, Self::Image, + Self::Media, Self::Broadcast, Self::Camera, Self::Display, @@ -3376,6 +3408,7 @@ impl Kind { match self { Self::Color => "Color", Self::Image => "Image", + Self::Media => "Media file", Self::Broadcast => "Broadcast", Self::Camera => "Camera", Self::Display => "Display", @@ -3393,7 +3426,7 @@ impl Kind { fn value(self) -> Option<&'static str> { match self { Self::Color | Self::SystemAudio | Self::Replay => None, - Self::Image => Some("file"), + Self::Image | Self::Media => Some("file"), Self::Lut => Some(".cube file"), Self::Broadcast => Some("path"), Self::Window => Some("window id"), @@ -3444,6 +3477,16 @@ impl AddSource { egui::color_picker::color_edit_button_srgba(ui, &mut self.color, egui::color_picker::Alpha::OnlyBlend); ui.end_row(); } + if self.kind == Kind::Media { + ui.label("Playback"); + ui.horizontal(|ui| { + ui.checkbox(&mut self.looping, "Loop") + .on_hover_text("Start over at the end, rather than hold the last picture"); + ui.checkbox(&mut self.restart, "Restart on program") + .on_hover_text("Play from the start each time it goes on program"); + }); + ui.end_row(); + } add = ui.button("Add source").clicked(); ui.end_row(); }); @@ -3458,6 +3501,11 @@ impl AddSource { color: moq_scene::Rgba(self.color.to_srgba_unmultiplied()), }, Kind::Image => Source::Image { path: value.into() }, + Kind::Media => Source::Media { + path: value.into(), + looping: self.looping, + restart: self.restart, + }, Kind::Lut => Source::Lut { path: value.into() }, Kind::Broadcast => Source::Broadcast { path: value }, Kind::Camera => Source::Camera { id },