diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 6217cd3d..ea153b75 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -1,8 +1,11 @@ name: Nightly -# The gates too slow or too specialized for a pull request: compositing and -# publishing through wgpu on Mesa's software Vulkan driver, and `just ci` on -# macOS, whose capture backends Check's Linux runner never compiles. +# The gates too slow or too specialized for a pull request: compositing, +# publishing, and the virtual camera through wgpu on Mesa's software Vulkan +# driver, the camera in a private PipeWire session; the virtual camera's +# v4l2loopback fallback, a module load no pull request job should make; and +# `just ci` on macOS, whose capture backends Check's Linux runner never +# compiles. permissions: contents: read @@ -36,6 +39,35 @@ jobs: - name: Render run: nix develop --command just render-sw + v4l2loopback: + name: v4l2loopback camera + runs-on: ubuntu-24.04 + timeout-minutes: 30 + env: + NEXTEST_PROFILE: ci + + steps: + - name: Checkout + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + submodules: true + + - uses: ./.github/actions/nix + + - name: Load v4l2loopback + # Out of tree, so DKMS builds it for the runner's kernel. + run: | + sudo apt-get update + sudo apt-get install -y "linux-headers-$(uname -r)" v4l2loopback-dkms + sudo modprobe v4l2loopback exclusive_caps=1 + sudo udevadm settle + # The runner user is not in the video group. + sudo chmod a+rw /dev/video* + + - name: Test + run: nix develop --command just v4l2loopback + macos: name: macOS runs-on: macos-latest diff --git a/Cargo.lock b/Cargo.lock index 9047d301..e87b2ec4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4370,8 +4370,10 @@ dependencies = [ "moq-scene", "moq-stats", "moq-tokio", + "moq-v4l", "moq-video", "mp4-atom", + "pipewire", "pollster 1.0.1", "qrcodegen", "raw-window-handle", diff --git a/Cargo.toml b/Cargo.toml index b4418c67..90774c0a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,6 +35,7 @@ moq-room = { path = "moq/rs/moq-room" } moq-scene = { path = "rs/moq-scene" } moq-stats = { path = "moq/rs/moq-stats" } moq-tokio = { path = "moq/rs/moq-tokio", default-features = false } +moq-v4l = { path = "moq/rs/moq-v4l" } # Default features off, as in moq: each consumer opts into native codecs and # rendering explicitly. moq-video = { path = "moq/rs/moq-video", default-features = false } diff --git a/README.md b/README.md index 2e080328..79fa973e 100644 --- a/README.md +++ b/README.md @@ -35,8 +35,10 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ a second, so viewers and recordings on the relay see the broadcast end rather than cut off. - `capture` adds camera, display, window, microphone, and system-audio - sources, and listening to a mix. On Linux it links ALSA; add `pipewire` for PipeWire cameras. `openh264` - is the software H.264 fallback for a machine without a hardware encoder. + sources, and listening to a mix. On Linux it links ALSA; add `pipewire` for + PipeWire cameras and the [virtual camera](#virtual-camera)'s PipeWire node. + `openh264` is the software H.264 fallback for a machine without a hardware + encoder. - `cef` adds browser sources, on Linux only for now. The build takes CEF from `CEF_PATH`, which the dev shell sets, or else downloads it, and the binary loads `libcef.so` from beside itself or from that `CEF_PATH`. @@ -104,7 +106,7 @@ Every panel is a tab you can drag, split, or stack. | Filters | Remove the selected source's background, crop the selected source or item, add, order, and set its filters, set an item's blend mode, and copy or paste filters. | | Keyframes | Animate the selected item: add, time, ease, and remove its keyframes, and set how far each moves, resizes, crops, and filters it (see [Keyframes](#keyframes)). **Play** plays the preview scene's keyframes again. | | Audio | Add, rename (double-click), and remove mixes, pick the one on program and the one to listen to and on which output, and set each audio source's gain, mute, delay, and the mixes it feeds, with a meter on every mix and source. | -| Output | Start and stop recording, pick the mix it records, set the canvas size and frame rate, add, resize, and remove further canvases, and see the broadcast path, whether the program is encoding, the encoder sessions open, and each canvas's renditions' encoder work. | +| Output | Start and stop recording, pick the mix it records, set the canvas size and frame rate, add, resize, and remove further canvases, see the broadcast path, whether the program is encoding, the encoder sessions open, and each canvas's renditions' encoder work, and turn the virtual camera on. | | Health | See how the stream reaches viewers, relay by relay and track by track, and the cause of each problem, and which recordings a crash left unfinished were repaired at launch. | | Replay | Replay the last moments of the program or a source in a replay source, slowed down if you like, or save them as a clip. | | Hotkeys | Bind keys to cut, transition, preview a scene, replay, save a clip, and start or stop recording, and clear them (see [Hotkeys](#hotkeys)). | @@ -997,6 +999,25 @@ The catalog asks viewers to report under `/viewers/`: a viewer publishes `/viewers/.echo`, as `moq export --echo ` does. The relay decides who may publish there, as for any broadcast. +## Virtual camera + +The **Virtual camera** toggle in the Output panel shows the program as a +camera that video-call and conferencing apps can pick, at the canvas size and +frame rate. It stays on until you turn it off or quit; it is not saved in the +scene file. The program is copied for the camera only while an app reads it. +Changing the canvas size or frame rate reopens the camera at the new one, so +an app reading it opens it again. The Devices panel leaves the camera out +while it is on, so it cannot be added as a source of the program it shows. + +- **Linux:** a PipeWire camera named "MoQ Studio" (needs the `pipewire` + feature), which browsers and other PipeWire-aware apps list beside your + other cameras. Without PipeWire, the studio writes to the first + [v4l2loopback](https://github.com/v4l2loopback/v4l2loopback) device that + takes the canvas size, for apps that open `/dev/video*` directly. The + studio never loads that module, since it needs root: + `sudo modprobe v4l2loopback exclusive_caps=1 card_label="MoQ Studio"`. +- **macOS and Windows:** not available yet. + ## Development Use the Nix dev shell (`direnv allow`, or `nix develop --command ...`) so diff --git a/justfile b/justfile index 8f69f8fb..3aa284d6 100644 --- a/justfile +++ b/justfile @@ -2,7 +2,7 @@ set shell := ["bash", "-euo", "pipefail", "-c"] # The device, codec, and browser features off by default, which every check # covers too. -features := "moq-scene/capture,moq-studio/capture,moq-studio/openh264,moq-studio/cef" +features := "moq-scene/capture,moq-studio/capture,moq-studio/openh264,moq-studio/pipewire,moq-studio/cef" # List the available development commands. default: @@ -45,8 +45,9 @@ fix: bun biome check --write . cargo fmt --all -# Composite scenes, render browser sources, and publish a program through wgpu -# on Mesa's software Vulkan driver, so no GPU is needed. +# Composite scenes, render browser sources, publish a program, and show it as a +# PipeWire camera through wgpu on Mesa's software Vulkan driver, so no GPU is +# needed. render-sw: #!/usr/bin/env bash set -euo pipefail @@ -57,6 +58,12 @@ render-sw: cargo nextest run --locked -p moq-scene --lib --run-ignored only -E 'test(/^compositor::tests::/) | test(/^browser::layers::tests::/)' cargo nextest run --locked -p moq-studio --features openh264 --run-ignored only -E 'test(/^engine::tests::the_program_/) | test(=setup::tests::times_every_canvas)' cargo nextest run --locked -p moq-studio --features cef --run-ignored only -E 'binary(browser)' + sh/pipewire-session.sh cargo nextest run --locked -p moq-studio --features pipewire --run-ignored only -E 'test(=engine::tests::the_virtual_camera_reaches_a_pipewire_app)' + +# Show the program through a v4l2loopback device, which needs the module loaded +# (`sudo modprobe v4l2loopback exclusive_caps=1`) and the device writable. +v4l2loopback: + cargo nextest run --locked -p moq-studio --run-ignored only -E 'test(=vcam::v4l2::tests::a_v4l2_app_reads_the_program)' # Move the moq submodule to moq's latest main (or a given commit) and refresh Cargo.lock and bun.lock. upgrade-moq rev="origin/main": diff --git a/quest/m0/vcam/README.md b/quest/m0/vcam/README.md index a4425146..a6d6aa48 100644 --- a/quest/m0/vcam/README.md +++ b/quest/m0/vcam/README.md @@ -18,8 +18,15 @@ Decided 2026-10-22: - **macOS:** a CoreMediaIO Camera Extension, which needs signing. - **Order:** the first backend to land adds the studio toggle. +The Linux backend landed the toggle: **Virtual camera** in the Output panel, +for the program, not saved in the scene file. Each OS backend is a +`vcam::backend` in `rs/moq-studio`, fed tightly packed RGBA program frames on +the `vcam` thread only while an app reads the camera. Each published scene +gets a toggle of its own once [published scenes](/quest/m0/scenes/published.md) +land. + ## Required -- [Linux virtual camera](/quest/m0/vcam/linux.md) - a PipeWire video node carrying the program +- [Linux virtual camera for each published scene](/quest/m0/vcam/linux.md) - a PipeWire camera per published scene, beside the program's - [Windows virtual camera](/quest/m0/vcam/windows.md) - a Media Foundation virtual camera on Windows 11 - [macOS virtual camera](/quest/m0/vcam/macos.md) - a CoreMediaIO Camera Extension inside the app bundle diff --git a/quest/m0/vcam/linux.md b/quest/m0/vcam/linux.md index 73151e61..6071e394 100644 --- a/quest/m0/vcam/linux.md +++ b/quest/m0/vcam/linux.md @@ -1,9 +1,23 @@ -# [M] Linux virtual camera +# [S] Linux virtual camera for each published scene ## Goal -The studio publishes the program or a published scene as a PipeWire video node that -PipeWire-aware apps (browsers and conferencing apps) list as a camera. It -falls back to an existing v4l2loopback device when one is present. -The studio's Devices panel leaves out its own virtual camera, so picking it -cannot feed the program back into itself. +Each published scene has a **Virtual camera** toggle of its own, as the +program does, and shows as its own PipeWire camera named after the scene, or +else its own v4l2loopback device. An app can then pick the program or any +published scene. The Devices panel leaves out every camera the studio +publishes, so picking one cannot feed it back into itself. + +## Plan + +The program's camera landed in +[moq-dev/studio#9](https://github.com/moq-dev/studio/pull/9): `vcam::Camera` +in `rs/moq-studio` reads the main canvas back through moq-video's +`render::readback` and reopens when the canvas changes size or frame rate. +Its PipeWire node name (`moq-studio`) and description ("MoQ Studio") are +fixed, so a camera per scene needs a name of its own, such as +`moq-studio-`. + +## Required + +- [Published scenes](/quest/m0/scenes/published.md) - any scene publishes as its own broadcast at its own size, replacing canvases diff --git a/rs/moq-studio/Cargo.toml b/rs/moq-studio/Cargo.toml index 1bca7df2..6736a4f7 100644 --- a/rs/moq-studio/Cargo.toml +++ b/rs/moq-studio/Cargo.toml @@ -29,10 +29,10 @@ nvidia = ["moq-video/nvidia"] openh264 = ["moq-video/openh264"] # Intel/AMD VAAPI hardware codecs (Linux); its bindgen needs libclang. vaapi = ["moq-video/vaapi"] -# PipeWire cameras, drawn without a copy, and audio through the sound server. -# Wayland screens and windows are not reachable yet: display and window sources -# capture X11 only. -pipewire = ["capture", "moq-scene/pipewire"] +# PipeWire cameras, drawn without a copy, audio through the sound server, and +# the virtual camera as a PipeWire node. Wayland screens and windows are not +# reachable yet: display and window sources capture X11 only. +pipewire = ["capture", "moq-scene/pipewire", "dep:pipewire"] # Browser sources, rendered by CEF in a helper process (Linux). Off by default: # the build downloads the CEF runtime (or takes it from `CEF_PATH`) and copies # it beside the binary, where the binary looks for it when it starts. @@ -92,6 +92,12 @@ winit = { version = "0.30", default-features = false } # PipeWire capture uses, so the two share one D-Bus stack. ashpd = { version = "0.13", default-features = false, features = ["global_shortcuts", "tokio"] } futures-util = "0.3" +# The virtual camera's v4l2loopback fallback. Pure Rust with its kernel bindings +# checked in, and already in the graph through moq-video's capture. +moq-v4l = { workspace = true } +# The virtual camera's PipeWire node, behind `pipewire`. moq-video links the +# same release; `v0_3_34` adds the trigger a driving stream paces frames with. +pipewire = { version = "0.10", optional = true, features = ["v0_3_34"] } # Whether the window is on Wayland, where only the portal hears keys globally. raw-window-handle = "0.6" # Which keysyms the X11 layout puts on each key, so a hotkey grabs the key diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 51017e43..72486123 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -18,6 +18,8 @@ //! - **encode**: one per program rendition, which waits for each program copy //! and encodes it. A new canvas size or frame rate starts new renditions. //! - Recordings write their files on the tokio runtime. +//! - **vcam** waits for each program copy the virtual camera takes, while it is +//! on, and hands it to the camera's backend. //! - **source-***: one per running source, since a capture stream may not //! leave the thread that opened it. //! - **autosave** writes the collection and its history to their files once @@ -96,6 +98,8 @@ pub struct State { pub reloads: u64, pub sources: Sources, pub compositor: Compositor, + /// The main canvas's virtual camera, while it is on. + camera: Option, /// Edited since the sources were last synced. stale: bool, /// For the autosave, which writes the collection once edits settle. @@ -108,6 +112,11 @@ pub struct State { } impl State { + /// How the virtual camera is doing, or `None` while it is off. + pub fn camera(&self) -> Option { + self.camera.as_ref().map(crate::vcam::Camera::status) + } + pub fn collection(&self) -> &Collection { &self.collection } @@ -669,6 +678,8 @@ impl Program { /// The running studio. pub struct Engine { + /// For opening the virtual camera, which reads the program back from it. + gpu: Gpu, pub state: Mutex, pub autosave: Autosave, pub stats: Arc, @@ -828,6 +839,7 @@ impl Engine { compositor.cut(&collection, &sources)?; let engine = Arc::new(Self { + gpu, state: Mutex::new(State { collection, history, @@ -835,6 +847,7 @@ impl Engine { reloads: 0, sources, compositor, + camera: None, stale: true, edits: autosave.edits(), renamed: Vec::new(), @@ -1163,6 +1176,21 @@ impl Engine { } } + /// Turn the main canvas's virtual camera on or off. Turning it on fails, + /// saying why, when this machine has no way to publish one. + pub fn set_camera(&self, on: bool) -> anyhow::Result<()> { + let canvas = self.state.lock().unwrap().compositor.canvas(); + // Opened and closed outside the lock, so the program clock does not + // wait on the camera's backend. + let camera = match on { + true => Some(crate::vcam::Camera::open(&self.gpu.device, &self.gpu.queue, canvas)?), + false => None, + }; + let previous = std::mem::replace(&mut self.state.lock().unwrap().camera, camera); + drop(previous); + Ok(()) + } + /// Draw every canvas once per tick of the program clock, and hand each /// canvas's program to its renditions' encoders, stamped `epoch` past /// `start`, while their gates let them encode: the canvas size every frame @@ -1174,7 +1202,8 @@ impl Engine { /// studio started with. Each further canvas added since gets a [`Line`] of /// its own from `extras` once drawn, in a broadcast of its own, and each /// ends once the collection drops its canvas. Drawing outlives the encoders: a failed - /// output stops copying its program, and the canvases keep moving. + /// output stops copying its program, and the canvases keep moving. The + /// virtual camera, while on, takes every main canvas frame an app reads. /// /// Runs until the studio [stops](Self::stop), and returns once every /// rendition's encode thread has finished its track. @@ -1225,6 +1254,9 @@ impl Engine { continue; } self.program.on_program(state, shown); + if let Some(camera) = &mut state.camera { + camera.offer(state.compositor.program(), state.compositor.canvas()); + } state .compositor .outputs() @@ -3769,4 +3801,88 @@ mod tests { } } } + + /// The virtual camera reaches an app that opens PipeWire cameras as the + /// picture the studio composed, and follows the canvas to a new size. + #[cfg(all(target_os = "linux", feature = "pipewire"))] + #[tokio::test(flavor = "multi_thread")] + #[ignore = "needs a PipeWire session and a Vulkan device; run by just render-sw"] + async fn the_virtual_camera_reaches_a_pipewire_app() { + /// Open the camera the way an app opens a PipeWire camera, by role and + /// class, and wait for the red picture the studio composes. + async fn red(engine: &Engine, width: u32, height: u32) -> moq_video::capture::Stream { + let mut config = moq_video::capture::Config::default(); + config.source = moq_video::capture::Source::Camera(Some("pipewire:moq-studio".into())); + config.width = Some(width); + config.height = Some(height); + let mut stream = kio::clock::timeout(TIMEOUT, moq_video::capture::open(&config)) + .await + .expect("the camera opens") + .expect("a camera stream"); + assert_eq!((stream.width(), stream.height()), (width, height)); + // The canvas is black until the color source delivers its first + // picture, so wait for the composed one rather than take the first. + kio::clock::timeout(TIMEOUT, async { + loop { + let frame = stream.read().await.expect("capture").expect("the camera ended"); + let rgba = frame.surface.to_rgba(&Default::default()).expect("rgba"); + let center = ((height / 2 * width + width / 2) * 4) as usize; + let [r, g, b, _] = rgba.data()[center..center + 4] else { + unreachable!() + }; + // The reader converts to I420, which shifts a primary a little. + if r > 200 && g < 60 && b < 60 { + return; + } + } + }) + .await + .expect("the composed picture arrives"); + let status = engine.state.lock().unwrap().camera().expect("a camera"); + assert!(status.reading, "the camera does not know an app reads it"); + assert!(status.sent > 0); + assert_eq!(status.error, None); + stream + } + + let dir = tempfile::tempdir().unwrap(); + let gpu = Gpu::new().await.expect("a GPU"); + let (autosave, _) = Autosave::open(dir.path().join("show.toml")).unwrap(); + let engine = Engine::start(launch(gpu, Vec::new(), dir.path()), autosave, red_canvas()) + .await + .expect("the studio starts"); + engine.set_camera(true).expect("the camera turns on"); + let status = engine.state.lock().unwrap().camera().expect("a camera"); + assert_eq!(status.backend, "PipeWire"); + assert_eq!(status.id, "pipewire:moq-studio"); + assert!(!status.reading, "nothing reads the camera yet"); + + let stream = red(&engine, 640, 360).await; + drop(stream); + + let canvas = moq_scene::Canvas { + width: 320, + height: 180, + framerate: 30, + }; + { + let mut state = engine.state.lock().unwrap(); + let mut draft = state.collection().draft("Resize the canvas"); + draft.set_canvas(canvas).unwrap(); + state.commit(draft, None).unwrap(); + } + kio::clock::timeout(TIMEOUT, async { + while engine.state.lock().unwrap().camera().expect("a camera").canvas != canvas { + kio::clock::sleep(Duration::from_millis(100)).await; + } + }) + .await + .expect("the camera reopens at the new size"); + let stream = red(&engine, 320, 180).await; + + drop(stream); + engine.set_camera(false).expect("the camera turns off"); + assert!(engine.state.lock().unwrap().camera().is_none()); + engine.stop(); + } } diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index c445521e..6e19103d 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -29,6 +29,7 @@ mod setup; mod snap; mod ui; mod uplink; +mod vcam; #[derive(usage_rs::Cli, Clone)] #[usage(unknown_flags = "error", args_override_self = false)] diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 8f209ef4..84cfc4b0 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1830,7 +1830,10 @@ impl Panels { let refreshed = self.devices.refresh(&self.runtime).context("cannot list devices"); self.check(refreshed); } - let Some(picked) = self.devices.ui(ui) else { return }; + let camera = self.engine.state.lock().unwrap().camera(); + let Some(picked) = self.devices.ui(ui, camera.as_ref().map(|camera| camera.id.as_str())) else { + return; + }; let mut state = self.engine.state.lock().unwrap(); // Two cameras of the same model share a name; number the later ones. let name = state.collection().unused_name(&picked.name); @@ -2668,6 +2671,46 @@ impl Panels { ui.colored_label(ui.visuals().error_fg_color, error); } } + + ui.separator(); + self.camera(ui); + } + + /// The main canvas's virtual camera: a toggle, and how it is doing while on. + fn camera(&mut self, ui: &mut Ui) { + let camera = self.engine.state.lock().unwrap().camera(); + let mut on = camera.is_some(); + if ui + .checkbox(&mut on, "Virtual camera") + .on_hover_text("Show the program as a camera that video-call apps can pick") + .changed() + { + let result = self.engine.set_camera(on); + let message = if on { "virtual camera on" } else { "virtual camera off" }; + self.report(result.map(|()| message.to_string())); + } + let Some(camera) = camera else { return }; + egui::Grid::new("camera").show(ui, |ui| { + ui.label("Camera"); + ui.label(&camera.backend); + ui.end_row(); + ui.label("Picture"); + let canvas = camera.canvas; + ui.label(format!( + "{}x{} at {} fps", + canvas.width, canvas.height, canvas.framerate + )); + ui.end_row(); + ui.label("Apps"); + ui.label(if camera.reading { "reading" } else { "none; not copying" }); + ui.end_row(); + ui.label("Sent"); + ui.label(format!("{} frames, {} dropped", camera.sent, camera.dropped)); + ui.end_row(); + }); + if let Some(error) = &camera.error { + ui.colored_label(ui.visuals().error_fg_color, error); + } } /// Replay the recent past of the program or a source in a replay source, @@ -4357,8 +4400,10 @@ impl Devices { } } - /// Draw the listing; the device to add once the user picks one. - fn ui(&mut self, ui: &mut Ui) -> Option { + /// Draw the listing, leaving out the camera `own` is the id of, the + /// studio's own virtual camera, so the program is not fed back into + /// itself; the device to add once the user picks one. + fn ui(&mut self, ui: &mut Ui, own: Option<&str>) -> Option { if self.pending.is_some() { ui.spinner(); } @@ -4377,6 +4422,11 @@ impl Devices { } Ok(devices) => { for (index, device) in devices.iter().enumerate() { + if let Source::Camera { id: Some(id) } = &device.source + && Some(id.as_str()) == own + { + continue; + } ui.horizontal(|ui| { if ui.small_button("Add").clicked() { picked = Some(index); diff --git a/rs/moq-studio/src/vcam.rs b/rs/moq-studio/src/vcam.rs new file mode 100644 index 00000000..553b3f37 --- /dev/null +++ b/rs/moq-studio/src/vcam.rs @@ -0,0 +1,283 @@ +//! The program as a system camera, for video-call and conferencing apps that +//! offer no ingest. +//! +//! On Linux the camera is a PipeWire video node (with the `pipewire` feature), +//! which PipeWire-aware apps such as browsers list as a camera. Without +//! PipeWire it is an existing v4l2loopback device, since that kernel module +//! needs root to load and so is never the first choice. +//! +//! The program clock offers every frame of the main canvas; [`Camera::offer`] +//! has moq-video read it back only while an app reads the camera, and the +//! `vcam` thread waits for the copy and hands the pixels to the backend. + +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, mpsc}; + +use anyhow::Context; +use moq_scene::{Canvas, wgpu}; +use moq_video::render::readback::{Pending, Readback}; + +#[cfg(all(target_os = "linux", feature = "pipewire"))] +mod pipewire; +#[cfg(target_os = "linux")] +mod v4l2; + +/// The main canvas's program, published as a camera other apps can open. +/// +/// Dropping it removes the camera. +pub struct Camera { + device: wgpu::Device, + queue: wgpu::Queue, + /// The canvas the backend was opened for. + canvas: Canvas, + readback: Readback, + /// `None` once dropping. + work: Option>, + worker: Option>, + shared: Arc, +} + +/// What the program clock hands the `vcam` thread. +enum Work { + /// A program copy on its way back from the GPU. Unbounded in the channel + /// only in name: the readback lends out at most [`Readback::DEPTH`] at once. + Picture(Pending), + /// The canvas changed size or frame rate: close the backend and open it + /// again for this one. + Reopen(Canvas), +} + +/// What the camera's threads report to the program clock and the UI. +#[derive(Default)] +struct Shared { + /// Where the camera lives now. + opened: Mutex, + /// Whether an app reads the camera, so frames are worth copying. + reading: AtomicBool, + /// Frames handed to the backend. + sent: AtomicU64, + /// Frames not copied because every staging buffer was still in flight. + dropped: AtomicU64, + /// Set with `error`, so the program clock checks it without a lock. + failed: AtomicBool, + /// Why the camera stopped, if it did. + error: Mutex>, +} + +impl Shared { + fn fail(&self, err: impl std::fmt::Display) { + tracing::error!(%err, "virtual camera stopped"); + self.error.lock().unwrap().get_or_insert_with(|| err.to_string()); + self.failed.store(true, Ordering::Relaxed); + } + + fn failed(&self) -> bool { + self.failed.load(Ordering::Relaxed) + } + + /// Whether a frame is worth copying: an app reads the camera, which has + /// not stopped. + fn wanted(&self) -> bool { + self.reading.load(Ordering::Relaxed) && !self.failed() + } +} + +/// Where an open camera lives. +#[derive(Clone, Default)] +struct Opened { + /// For the Output panel: `PipeWire`, or the v4l2loopback device's path. + backend: String, + /// The id moq-video's camera capture lists it by. + id: String, + /// The size and frame rate it shows. + canvas: Canvas, +} + +/// How the camera is doing, for the Output panel. +pub struct Status { + /// Where the camera lives, such as a PipeWire node or a `/dev/videoN` path. + pub backend: String, + /// The id the Devices panel would list the camera by, so it can leave it out. + pub id: String, + /// The size and frame rate it shows. + pub canvas: Canvas, + /// Whether an app reads the camera. + pub reading: bool, + pub sent: u64, + pub dropped: u64, + /// Why the camera stopped, if it did. + pub error: Option, +} + +/// Where the camera's frames go, on the `vcam` thread. +trait Sink: Send { + /// Show one picture: tightly packed RGBA rows at the canvas size. + fn write(&mut self, rgba: &[u8]) -> anyhow::Result<()>; +} + +impl Camera { + /// Publish the `canvas` programs drawn on `device` as a camera. + /// + /// Fails, naming why each backend refused, when this machine has no way to + /// publish one. + pub fn open(device: &wgpu::Device, queue: &wgpu::Queue, canvas: Canvas) -> anyhow::Result { + let readback = Readback::new(device, queue, size(canvas))?; + let shared = Arc::new(Shared::default()); + let sink = open(canvas, &shared)?; + + let (work, pending) = mpsc::channel(); + let worker = { + let shared = shared.clone(); + std::thread::Builder::new() + .name("vcam".into()) + .spawn(move || run(pending, sink, &shared)) + .context("cannot start the virtual camera thread")? + }; + Ok(Self { + device: device.clone(), + queue: queue.clone(), + canvas, + readback, + work: Some(work), + worker: Some(worker), + shared, + }) + } + + /// Copy `program`, drawn at `canvas`, to the camera if an app reads it. + /// + /// A canvas of a new size or frame rate reopens the camera for it, so an + /// app reading it opens it again. + /// + /// The copy is submitted before this returns, so the caller draws into + /// `program` again freely. Never waits on the GPU. + pub fn offer(&mut self, program: &wgpu::Texture, canvas: Canvas) { + let Some(work) = &self.work else { return }; + if canvas != self.canvas { + self.canvas = canvas; + match Readback::new(&self.device, &self.queue, size(canvas)) { + Ok(readback) => self.readback = readback, + Err(err) => return self.shared.fail(format!("{err:#}")), + } + // A send fails only once the thread stopped, having said why. + let _ = work.send(Work::Reopen(canvas)); + } + if !self.shared.wanted() { + return; + } + match self.readback.copy(program) { + Ok(Some(pending)) => { + let _ = work.send(Work::Picture(pending)); + } + Ok(None) => { + self.shared.dropped.fetch_add(1, Ordering::Relaxed); + } + Err(err) => self.shared.fail(format!("{err:#}")), + } + } + + pub fn status(&self) -> Status { + let Opened { backend, id, canvas } = self.shared.opened.lock().unwrap().clone(); + Status { + backend, + id, + canvas, + reading: self.shared.wanted(), + sent: self.shared.sent.load(Ordering::Relaxed), + dropped: self.shared.dropped.load(Ordering::Relaxed), + error: self.shared.error.lock().unwrap().clone(), + } + } +} + +impl Drop for Camera { + fn drop(&mut self) { + // Ends the thread, which removes the camera before this returns, so + // turning it straight back on finds the device free. + self.work.take(); + if let Some(worker) = self.worker.take() { + let _ = worker.join(); + } + } +} + +fn size(canvas: Canvas) -> moq_video::Size { + moq_video::Size::new(canvas.width, canvas.height) +} + +/// Open the first backend that takes `canvas`, and say where it lives in +/// `shared`. +fn open(canvas: Canvas, shared: &Arc) -> anyhow::Result> { + let (sink, backend, id) = backend(size(canvas), canvas.framerate, shared)?; + tracing::info!(%backend, "virtual camera on"); + *shared.opened.lock().unwrap() = Opened { backend, id, canvas }; + Ok(sink) +} + +/// The first backend that takes a `size` camera at `framerate`, with where it +/// lives and its capture id. +#[cfg(target_os = "linux")] +fn backend( + size: moq_video::Size, + framerate: u32, + shared: &Arc, +) -> anyhow::Result<(Box, String, String)> { + #[cfg(feature = "pipewire")] + let pipewire = match pipewire::Node::open(size, framerate, shared.clone()) { + Ok(node) => return Ok((Box::new(node), "PipeWire".to_string(), pipewire::ID.to_string())), + Err(err) => { + tracing::warn!(err = %format!("{err:#}"), "no PipeWire camera, trying v4l2loopback"); + format!("{err:#}") + } + }; + #[cfg(not(feature = "pipewire"))] + let pipewire = "this build has no `pipewire` feature".to_string(); + + match v4l2::Loopback::open(size, framerate) { + Ok(loopback) => { + let path = loopback.name(); + // The kernel keeps no count of readers to wait for. + shared.reading.store(true, Ordering::Relaxed); + Ok((Box::new(loopback), path.clone(), path)) + } + Err(err) => anyhow::bail!("no virtual camera. PipeWire: {pipewire}. v4l2loopback: {err:#}"), + } +} + +#[cfg(not(target_os = "linux"))] +fn backend( + size: moq_video::Size, + framerate: u32, + shared: &Arc, +) -> anyhow::Result<(Box, String, String)> { + let _ = (size, framerate, shared); + anyhow::bail!("the virtual camera is not available on this platform yet") +} + +/// Hand each copy to `sink` as it lands, until the camera is dropped or fails. +fn run(pending: mpsc::Receiver, mut sink: Box, shared: &Arc) { + while let Ok(work) = pending.recv() { + let result = match work { + Work::Picture(pending) => pending + .wait() + .context("reading the program back") + .and_then(|rgba| sink.write(rgba.data())), + Work::Reopen(canvas) => { + // Closed first, so the new one finds the device free. + drop(sink); + shared.reading.store(false, Ordering::Relaxed); + match open(canvas, shared) { + Ok(opened) => { + sink = opened; + continue; + } + Err(err) => return shared.fail(format!("{err:#}")), + } + } + }; + if let Err(err) = result { + return shared.fail(format!("{err:#}")); + } + shared.sent.fetch_add(1, Ordering::Relaxed); + } +} diff --git a/rs/moq-studio/src/vcam/pipewire.rs b/rs/moq-studio/src/vcam/pipewire.rs new file mode 100644 index 00000000..47b75cf2 --- /dev/null +++ b/rs/moq-studio/src/vcam/pipewire.rs @@ -0,0 +1,364 @@ +//! The camera as a PipeWire video node. +//! +//! A `Video/Source` node with the `Camera` role is what PipeWire-aware apps +//! (browsers through the camera portal, conferencing apps) list as a camera. +//! It offers one format, the canvas's RGBA at its size and frame rate, so the +//! program reaches an app without a conversion here. +//! +//! PipeWire objects stay on the thread that made them, so the node runs its +//! own main loop. The `vcam` thread leaves the newest picture in a [`Slot`] +//! and wakes the loop; the loop copies it into a PipeWire buffer. + +use std::cell::RefCell; +use std::rc::Rc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, mpsc}; + +use anyhow::Context; +use moq_video::Size; +use pipewire as pw; +use pw::spa; +use spa::param::format::{FormatProperties, MediaSubtype, MediaType}; +use spa::param::video::{VideoFormat, VideoInfoRaw}; + +use super::{Shared, Sink}; + +/// The node's name, which `pw-cli` and pickers that show ids print. +const NAME: &str = "moq-studio"; + +/// The id moq-video's camera capture lists the node by: `pipewire:` and its name. +pub const ID: &str = "pipewire:moq-studio"; + +/// What apps list the camera as. +const DESCRIPTION: &str = "MoQ Studio"; + +pub struct Node { + slot: Arc, + wake: pw::channel::Sender, + thread: Option>, + shared: Arc, +} + +/// The newest picture, handed from the `vcam` thread to the node's loop. +#[derive(Default)] +struct Slot { + picture: Mutex>, + /// A wake is on its way to the loop, so another is not sent: the channel + /// never holds more than one. + woken: AtomicBool, +} + +enum Message { + /// A new picture is in the slot. + Picture, + Quit, +} + +impl Node { + /// Add a `size` camera node at `framerate` to the session's PipeWire. + /// + /// Fails when there is no PipeWire session to add it to. + pub fn open(size: Size, framerate: u32, shared: Arc) -> anyhow::Result { + let slot = Arc::new(Slot::default()); + let (wake, messages) = pw::channel::channel(); + let (ready, opened) = mpsc::channel(); + let thread = { + let slot = slot.clone(); + let shared = shared.clone(); + std::thread::Builder::new() + .name("vcam-pipewire".into()) + .spawn(move || { + let ready = Rc::new(RefCell::new(Some(ready))); + if let Err(err) = run(size, framerate, slot, shared, messages, ready.clone()) + && let Some(ready) = ready.take() + { + let _ = ready.send(Err(format!("{err:#}"))); + } + }) + .context("cannot start the PipeWire thread")? + }; + match opened.recv() { + Ok(Ok(())) => Ok(Self { + slot, + wake, + thread: Some(thread), + shared, + }), + Ok(Err(err)) => { + let _ = thread.join(); + Err(anyhow::anyhow!(err)) + } + Err(_) => { + let _ = thread.join(); + anyhow::bail!("the PipeWire thread stopped") + } + } + } +} + +impl Sink for Node { + fn write(&mut self, rgba: &[u8]) -> anyhow::Result<()> { + if self.shared.failed() { + anyhow::bail!("the PipeWire node stopped"); + } + { + let mut picture = self.slot.picture.lock().unwrap(); + picture.clear(); + picture.extend_from_slice(rgba); + } + if !self.slot.woken.swap(true, Ordering::AcqRel) && self.wake.send(Message::Picture).is_err() { + anyhow::bail!("the PipeWire node stopped"); + } + Ok(()) + } +} + +impl Drop for Node { + fn drop(&mut self) { + let _ = self.wake.send(Message::Quit); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + +/// Whoever waits for the node to be added, until it is or PipeWire refuses it. +type Ready = Rc>>>>; + +/// Add the node and run it until it is dropped or PipeWire ends it, saying on +/// `ready` whether PipeWire added it. A failure before then goes to `ready` +/// rather than `shared`, so the caller can still fall back. +fn run( + size: Size, + framerate: u32, + slot: Arc, + shared: Arc, + messages: pw::channel::Receiver, + ready: Ready, +) -> anyhow::Result<()> { + pw::init(); + let mainloop = pw::main_loop::MainLoopRc::new(None).context("PipeWire main loop")?; + let context = pw::context::ContextRc::new(&mainloop, None).context("PipeWire context")?; + let core = context.connect_rc(None).context("connecting to PipeWire")?; + let stream = pw::stream::StreamRc::new( + core, + NAME, + pw::properties::properties! { + *pw::keys::MEDIA_TYPE => "Video", + *pw::keys::MEDIA_CATEGORY => "Capture", + *pw::keys::MEDIA_CLASS => "Video/Source", + *pw::keys::MEDIA_ROLE => "Camera", + *pw::keys::NODE_NAME => NAME, + *pw::keys::NODE_DESCRIPTION => DESCRIPTION, + }, + ) + .context("PipeWire stream")?; + + let stride = size.width * 4; + let _listener = stream + .add_local_listener::<()>() + .state_changed({ + let mainloop = mainloop.downgrade(); + let shared = shared.clone(); + move |_, _, _, state| { + tracing::debug!(?state, "virtual camera node"); + // Streaming once an app links to the node, paused when it leaves. + let reading = matches!(state, pw::stream::StreamState::Streaming); + shared.reading.store(reading, Ordering::Relaxed); + // Our own teardown quits the loop before anything disconnects, so + // either of these is PipeWire ending the node. + let ended = match state { + pw::stream::StreamState::Error(err) => Some(format!("PipeWire: {err}")), + pw::stream::StreamState::Unconnected => Some("the PipeWire session ended".to_string()), + // Paused first once PipeWire has added the node. + pw::stream::StreamState::Paused => { + if let Some(ready) = ready.take() { + let _ = ready.send(Ok(())); + } + None + } + _ => None, + }; + if let Some(ended) = ended { + match ready.take() { + Some(ready) => { + let _ = ready.send(Err(ended)); + } + None => shared.fail(ended), + } + if let Some(mainloop) = mainloop.upgrade() { + mainloop.quit(); + } + } + } + }) + .param_changed({ + let shared = shared.clone(); + move |stream, _, id, param| { + let Some(param) = param else { return }; + if id != spa::param::ParamType::Format.as_raw() { + return; + } + // The offer is fixed, so the format an app picks can only be it. + let mut info = VideoInfoRaw::new(); + if info.parse(param).is_err() || info.format() != VideoFormat::RGBA { + shared.fail(format!("an app negotiated {:?}, not RGBA", info.format())); + return; + } + let buffers = buffers(size, stride); + let Some(buffers) = spa::pod::Pod::from_bytes(&buffers) else { + return; + }; + if let Err(err) = stream.update_params(&mut [buffers]) { + shared.fail(format!("PipeWire buffers: {err}")); + } + } + }) + .process({ + let slot = slot.clone(); + let shared = shared.clone(); + move |stream, _| { + let picture = slot.picture.lock().unwrap(); + // Nothing to show before the first picture lands. + if picture.is_empty() { + return; + } + // Every buffer is with the app; it gets the next picture instead. + let Some(mut buffer) = stream.dequeue_buffer() else { + return; + }; + let Some(data) = buffer.datas_mut().first_mut() else { + return; + }; + let Some(target) = data.data().and_then(|target| target.get_mut(..picture.len())) else { + // Sized by our own buffers param, so this is PipeWire + // handing over something else. + shared.fail("PipeWire handed over a buffer smaller than a picture"); + return; + }; + target.copy_from_slice(&picture); + let chunk = data.chunk_mut(); + *chunk.offset_mut() = 0; + *chunk.size_mut() = picture.len() as u32; + *chunk.stride_mut() = stride as i32; + } + }) + .register() + .context("PipeWire stream listener")?; + + let format = format(size, framerate); + let format = spa::pod::Pod::from_bytes(&format).context("the PipeWire format offer")?; + // Driving, so the node runs on the program clock: each picture triggers a + // graph cycle that hands it to whichever app reads the camera. No + // autoconnect: an app links to the node when it opens the camera. + stream + .connect( + spa::utils::Direction::Output, + None, + pw::stream::StreamFlags::DRIVER | pw::stream::StreamFlags::MAP_BUFFERS, + &mut [format], + ) + .context("adding the PipeWire node")?; + + let _messages = messages.attach(mainloop.loop_(), { + let mainloop = mainloop.downgrade(); + let slot = slot.clone(); + let stream = stream.downgrade(); + move |message| match message { + Message::Picture => { + slot.woken.store(false, Ordering::Release); + let Some(stream) = stream.upgrade() else { return }; + // Another driver runs the graph and takes the newest picture on + // its own cycle; only a driving node triggers one. + if stream.is_driving() + && let Err(err) = stream.trigger_process() + { + tracing::debug!(%err, "cannot trigger the virtual camera"); + } + } + Message::Quit => { + if let Some(mainloop) = mainloop.upgrade() { + mainloop.quit(); + } + } + } + }); + + mainloop.run(); + Ok(()) +} + +/// The one format the node offers: `size` RGBA at `framerate`. +fn format(size: Size, framerate: u32) -> Vec { + let object = spa::pod::object!( + spa::utils::SpaTypes::ObjectParamFormat, + spa::param::ParamType::EnumFormat, + spa::pod::property!(FormatProperties::MediaType, Id, MediaType::Video), + spa::pod::property!(FormatProperties::MediaSubtype, Id, MediaSubtype::Raw), + spa::pod::property!(FormatProperties::VideoFormat, Id, VideoFormat::RGBA), + spa::pod::property!( + FormatProperties::VideoSize, + Rectangle, + spa::utils::Rectangle { + width: size.width, + height: size.height, + } + ), + spa::pod::property!( + FormatProperties::VideoFramerate, + Fraction, + spa::utils::Fraction { + num: framerate, + denom: 1, + } + ), + ); + serialize(spa::pod::Value::Object(object)) +} + +/// Buffers sized for one `size` RGBA picture in shared memory. +fn buffers(size: Size, stride: u32) -> Vec { + let mem_ptr = 1 << spa::buffer::DataType::MemPtr.as_raw(); + let mem_fd = 1 << spa::buffer::DataType::MemFd.as_raw(); + let object = spa::pod::Object { + type_: spa::utils::SpaTypes::ObjectParamBuffers.as_raw(), + id: spa::param::ParamType::Buffers.as_raw(), + properties: vec![ + spa::pod::Property::new( + spa::sys::SPA_PARAM_BUFFERS_buffers, + spa::pod::Value::Choice(spa::pod::ChoiceValue::Int(spa::utils::Choice( + spa::utils::ChoiceFlags::empty(), + spa::utils::ChoiceEnum::Range { + default: 4, + min: 2, + max: 8, + }, + ))), + ), + spa::pod::Property::new(spa::sys::SPA_PARAM_BUFFERS_blocks, spa::pod::Value::Int(1)), + spa::pod::Property::new( + spa::sys::SPA_PARAM_BUFFERS_size, + spa::pod::Value::Int((stride * size.height) as i32), + ), + spa::pod::Property::new(spa::sys::SPA_PARAM_BUFFERS_stride, spa::pod::Value::Int(stride as i32)), + spa::pod::Property::new( + spa::sys::SPA_PARAM_BUFFERS_dataType, + spa::pod::Value::Choice(spa::pod::ChoiceValue::Int(spa::utils::Choice( + spa::utils::ChoiceFlags::empty(), + spa::utils::ChoiceEnum::Flags { + default: mem_fd | mem_ptr, + flags: Vec::new(), + }, + ))), + ), + ], + }; + serialize(spa::pod::Value::Object(object)) +} + +fn serialize(value: spa::pod::Value) -> Vec { + spa::pod::serialize::PodSerializer::serialize(std::io::Cursor::new(Vec::new()), &value) + .expect("serializing a pod into memory cannot fail") + .0 + .into_inner() +} diff --git a/rs/moq-studio/src/vcam/v4l2.rs b/rs/moq-studio/src/vcam/v4l2.rs new file mode 100644 index 00000000..6d80dc21 --- /dev/null +++ b/rs/moq-studio/src/vcam/v4l2.rs @@ -0,0 +1,150 @@ +//! The fallback camera: an existing v4l2loopback device, written to as I420. +//! +//! v4l2loopback is a kernel module, so the studio never creates a device; it +//! takes the first one it finds that accepts the program's size. + +use std::io::Write; +use std::path::PathBuf; + +use anyhow::Context; +use moq_v4l::video::Output; +use moq_video::{I420, Size}; + +use super::Sink; + +/// The driver name v4l2loopback devices report. +const DRIVER: &str = "v4l2 loopback"; + +/// I420, which every V4L2 camera reader takes. +const FOURCC: &[u8; 4] = b"YU12"; + +pub struct Loopback { + device: moq_v4l::Device, + path: PathBuf, + size: Size, +} + +impl Loopback { + /// Open the first v4l2loopback device that takes `size` I420 frames. + pub fn open(size: Size, framerate: u32) -> anyhow::Result { + let mut refused = Vec::new(); + for path in devices()? { + let Ok(device) = moq_v4l::Device::with_path(&path) else { + continue; + }; + match device.query_caps() { + Ok(caps) if caps.driver == DRIVER => {} + _ => continue, + } + match configure(&device, size, framerate) { + Ok(()) => return Ok(Self { device, path, size }), + Err(err) => refused.push(format!("{}: {err:#}", path.display())), + } + } + match refused.is_empty() { + true => anyhow::bail!("no device; load the v4l2loopback module to create one"), + false => anyhow::bail!("no device takes the program ({})", refused.join("; ")), + } + } + + pub fn name(&self) -> String { + self.path.display().to_string() + } +} + +impl Sink for Loopback { + fn write(&mut self, rgba: &[u8]) -> anyhow::Result<()> { + let i420 = moq_video::Surface::rgba(rgba, self.size)?.into_i420()?; + // v4l2loopback takes one frame per write, so a short one would split it. + let written = self + .device + .write(i420.data()) + .with_context(|| format!("writing to {}", self.path.display()))?; + anyhow::ensure!( + written == i420.data().len(), + "{} took {written} of a frame's {} bytes", + self.path.display(), + i420.data().len() + ); + Ok(()) + } +} + +/// Every `/dev/videoN`, in number order. +fn devices() -> anyhow::Result> { + let mut devices: Vec<(u32, PathBuf)> = std::fs::read_dir("/dev") + .context("listing /dev")? + .filter_map(|entry| { + let path = entry.ok()?.path(); + let number = path.file_name()?.to_str()?.strip_prefix("video")?.parse().ok()?; + Some((number, path)) + }) + .collect(); + devices.sort(); + Ok(devices.into_iter().map(|(_, path)| path).collect()) +} + +/// Set `device` to take `size` I420 frames at `framerate`, refusing any other +/// layout it answers with. +fn configure(device: &moq_v4l::Device, size: Size, framerate: u32) -> anyhow::Result<()> { + let fourcc = moq_v4l::FourCC::new(FOURCC); + let mut want = moq_v4l::Format::new(size.width, size.height, fourcc); + want.stride = size.width; + let len = I420::len(size)?; + want.size = len as u32; + let set = device.set_format(&want).context("setting the format")?; + anyhow::ensure!( + set.width == size.width && set.height == size.height && set.fourcc == fourcc, + "it took {}x{} {} rather than {size} {}", + set.width, + set.height, + set.fourcc, + fourcc + ); + anyhow::ensure!( + set.stride == size.width && set.size as usize == len, + "it pads I420 rows to {} bytes", + set.stride + ); + device + .set_params(&moq_v4l::video::output::Parameters::with_fps(framerate)) + .context("setting the frame rate")?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::io::Read; + + use super::*; + + /// A V4L2 app reading the loopback device gets the picture written to it. + #[test] + #[ignore = "needs a v4l2loopback device; run by just v4l2loopback"] + fn a_v4l2_app_reads_the_program() { + let size = Size::new(320, 180); + let mut loopback = Loopback::open(size, 30).expect("a v4l2loopback device"); + let red = [255, 0, 0, 255].repeat((size.width * size.height) as usize); + // A reader opens the device once a writer has set it up, as an app does. + loopback.write(&red).expect("write"); + let mut reader = moq_v4l::Device::with_path(&loopback.path).expect("the reader opens"); + let mut frame = vec![0; I420::len(size).unwrap()]; + loopback.write(&red).expect("write"); + let read = reader.read(&mut frame).expect("read"); + assert_eq!(read, frame.len()); + + let i420 = I420::new(size, frame).expect("an I420 frame"); + let center = |plane: &[u8], width: u32, height: u32| plane[(height / 2 * width + width / 2) as usize]; + let (y, u, v) = ( + center(i420.y(), size.width, size.height), + center(i420.u(), size.width / 2, size.height / 2), + center(i420.v(), size.width / 2, size.height / 2), + ); + // Red, in any matrix: dim luma, blue-difference below gray, and + // red-difference well above it. + assert!( + y < 100 && u < 128 && v > 200, + "the reader got y={y} u={u} v={v}, not red" + ); + } +} diff --git a/sh/pipewire-session.sh b/sh/pipewire-session.sh new file mode 100755 index 00000000..c235a82e --- /dev/null +++ b/sh/pipewire-session.sh @@ -0,0 +1,46 @@ +#!/usr/bin/env bash +# Run a command inside a private PipeWire session: a PipeWire daemon and the +# WirePlumber session manager that links streams, on a runtime directory of +# their own, so a test reaches neither the desktop's session nor needs one. +set -euo pipefail + +# The dev shell carries PipeWire for linking; the session manager comes from the +# same nixpkgs. +wireplumber=$(nix build --inputs-from . --no-link --print-out-paths 'nixpkgs#wireplumber^out') + +# Only the packages' own configuration, never the host's, which can name +# components these builds do not ship. +pipewire_share=$(dirname "$(command -v pipewire)")/../share +export PIPEWIRE_CONFIG_DIR=$pipewire_share/pipewire +export WIREPLUMBER_CONFIG_DIR=$wireplumber/share/wireplumber +export WIREPLUMBER_DATA_DIR=$wireplumber/share/wireplumber + +runtime=$(mktemp -d) +export XDG_RUNTIME_DIR=$runtime +unset PIPEWIRE_REMOTE DBUS_SESSION_BUS_ADDRESS +pids=() +cleanup() { + if ((${#pids[@]})); then + kill "${pids[@]}" 2>/dev/null || true + wait "${pids[@]}" 2>/dev/null || true + fi + rm -rf "$runtime" +} +trap cleanup EXIT + +pipewire & +pids+=("$!") +# The socket appears once the daemon listens; a client started before it fails. +until [[ -S $runtime/pipewire-0 ]]; do + kill -0 "${pids[0]}" + sleep 0.1 +done +"$wireplumber/bin/wireplumber" & +pids+=("$!") +# WirePlumber connects as the session manager once it has loaded its policy. +until pw-cli ls Client 2>/dev/null | grep -q 'application.name = "WirePlumber"'; do + kill -0 "${pids[1]}" + sleep 0.1 +done + +"$@"