pixelflux/webcam/
pipewire.rs

1//! PipeWire sink: the camera as a `Video/Source` node.
2//!
3//! Where a PipeWire daemon is reachable, the same frames that feed the interposer ring are also
4//! offered as a PipeWire camera node, which PipeWire-native consumers (GStreamer `pipewiresrc`,
5//! portal-aware applications) link to directly and which `pipewire-v4l2` exposes as a V4L2 device
6//! to plain libc consumers. `libpipewire-0.3` is loaded at run time and the SPA pods are built by
7//! hand, so pixelflux keeps no build-time or load-time dependency on PipeWire and the sink simply
8//! stays off where the library or the daemon is absent.
9//!
10//! The stream is the graph driver: every published frame is copied into the node's latest-frame
11//! slot and one graph cycle is triggered; the process callback then fills whatever buffer the
12//! consumers negotiated (memfd-backed, mapped for us) from that slot. The library binding and the
13//! pod encoding live in `crate::pipewire`, shared with the portal screen capture.
14
15use std::ffi::{CStr, CString, c_char, c_int, c_void};
16use std::ptr;
17use std::sync::atomic::{AtomicI32, AtomicPtr, AtomicU64, Ordering};
18use std::sync::{Arc, Mutex};
19use std::time::{Duration, Instant};
20
21use super::ring::{
22    RingFormat, V4L2_PIX_FMT_MJPEG, V4L2_PIX_FMT_NV12, V4L2_PIX_FMT_YUV420, V4L2_PIX_FMT_YUYV,
23};
24use crate::pipewire::*;
25
26fn spa_video_format(fourcc: u32) -> Option<u32> {
27    match fourcc {
28        V4L2_PIX_FMT_YUV420 => Some(SPA_VIDEO_FORMAT_I420),
29        V4L2_PIX_FMT_NV12 => Some(SPA_VIDEO_FORMAT_NV12),
30        V4L2_PIX_FMT_YUYV => Some(SPA_VIDEO_FORMAT_YUY2),
31        _ => None,
32    }
33}
34
35/// `SPA_PARAM_EnumFormat`: one video format at the device size and rate — raw with the given
36/// SPA pixel format, or `video/mjpg` for the MJPEG device (`spa_format` None).
37fn format_pod(fmt: &RingFormat, spa_format: Option<u32>) -> Vec<u8> {
38    object(SPA_TYPE_OBJECT_FORMAT, SPA_PARAM_ENUM_FORMAT, |p| {
39        prop(p, SPA_FORMAT_MEDIA_TYPE, |v| {
40            pod_id(v, SPA_MEDIA_TYPE_VIDEO)
41        });
42        match spa_format {
43            Some(raw) => {
44                prop(p, SPA_FORMAT_MEDIA_SUBTYPE, |v| {
45                    pod_id(v, SPA_MEDIA_SUBTYPE_RAW)
46                });
47                prop(p, SPA_FORMAT_VIDEO_FORMAT, |v| pod_id(v, raw));
48            }
49            None => prop(p, SPA_FORMAT_MEDIA_SUBTYPE, |v| {
50                pod_id(v, SPA_MEDIA_SUBTYPE_MJPG)
51            }),
52        }
53        prop(p, SPA_FORMAT_VIDEO_SIZE, |v| {
54            pod_rect(v, fmt.width, fmt.height)
55        });
56        prop(p, SPA_FORMAT_VIDEO_FRAMERATE, |v| {
57            pod_frac(v, fmt.fps_num, fmt.fps_den)
58        });
59    })
60}
61
62/// `SPA_PARAM_Buffers` answered once the format is chosen: memfd (or plain memory) buffers of
63/// exactly one frame.
64fn buffers_pod(fmt: &RingFormat) -> Vec<u8> {
65    object(SPA_TYPE_OBJECT_PARAM_BUFFERS, SPA_PARAM_BUFFERS, |p| {
66        prop(p, SPA_PARAM_BUFFERS_BUFFERS, |v| {
67            pod_choice_int(v, SPA_CHOICE_RANGE, &[4, 2, 8])
68        });
69        prop(p, SPA_PARAM_BUFFERS_BLOCKS, |v| pod_int(v, 1));
70        prop(p, SPA_PARAM_BUFFERS_SIZE, |v| {
71            pod_int(v, fmt.sizeimage as i32)
72        });
73        prop(p, SPA_PARAM_BUFFERS_STRIDE, |v| {
74            pod_int(v, fmt.bytesperline as i32)
75        });
76        prop(p, SPA_PARAM_BUFFERS_ALIGN, |v| pod_int(v, 16));
77        prop(p, SPA_PARAM_BUFFERS_DATATYPE, |v| {
78            pod_choice_int(
79                v,
80                SPA_CHOICE_FLAGS,
81                &[(1 << SPA_DATA_MEMFD) | (1 << SPA_DATA_MEMPTR)],
82            )
83        });
84    })
85}
86
87fn meta_pod() -> Vec<u8> {
88    object(SPA_TYPE_OBJECT_PARAM_META, SPA_PARAM_META, |p| {
89        prop(p, SPA_PARAM_META_TYPE, |v| pod_id(v, SPA_META_HEADER));
90        prop(p, SPA_PARAM_META_SIZE, |v| {
91            pod_int(v, SPA_META_HEADER_SIZE as i32)
92        });
93    })
94}
95
96// --- the sink ---------------------------------------------------------------------------------
97
98/// State shared with the stream callbacks (handed to PipeWire as the listener's user data).
99struct Shared {
100    api: Api,
101    stream: AtomicPtr<c_void>,
102    frame: Mutex<Vec<u8>>,
103    frame_seq: AtomicU64,
104    frame_ts_ns: AtomicU64,
105    state: AtomicI32,
106    stride: i32,
107    buffers_pod: Vec<u8>,
108    meta_pod: Vec<u8>,
109}
110
111unsafe extern "C" fn on_state_changed(
112    data: *mut c_void,
113    _old: c_int,
114    new: c_int,
115    error: *const c_char,
116) {
117    unsafe {
118        let shared = &*(data as *const Shared);
119        shared.state.store(new, Ordering::Release);
120        if new == PW_STREAM_STATE_ERROR {
121            let msg = if error.is_null() {
122                String::new()
123            } else {
124                CStr::from_ptr(error).to_string_lossy().into_owned()
125            };
126            eprintln!("[webcam] PipeWire stream error: {}", msg);
127        }
128    }
129}
130
131unsafe extern "C" fn on_param_changed(data: *mut c_void, id: u32, param: *const c_void) {
132    if id != SPA_PARAM_FORMAT || param.is_null() {
133        return;
134    }
135    unsafe {
136        let shared = &*(data as *const Shared);
137        let stream = shared.stream.load(Ordering::Acquire);
138        if stream.is_null() {
139            return;
140        }
141        let mut params: [*const c_void; 2] = [
142            shared.buffers_pod.as_ptr() as *const c_void,
143            shared.meta_pod.as_ptr() as *const c_void,
144        ];
145        (shared.api.stream_update_params)(stream, params.as_mut_ptr(), 2);
146    }
147}
148
149unsafe extern "C" fn on_process(data: *mut c_void) {
150    unsafe {
151        let shared = &*(data as *const Shared);
152        let stream = shared.stream.load(Ordering::Acquire);
153        if stream.is_null() {
154            return;
155        }
156        let frame = shared.frame.lock().unwrap_or_else(|e| e.into_inner());
157        if frame.is_empty() {
158            return;
159        }
160        let b = (shared.api.stream_dequeue_buffer)(stream);
161        if b.is_null() {
162            return;
163        }
164        let spa_buf = (*b).buffer;
165        if !spa_buf.is_null() && (*spa_buf).n_datas >= 1 {
166            let d = &mut *(*spa_buf).datas;
167            let mut written = 0usize;
168            if !d.data.is_null() && !d.chunk.is_null() {
169                written = frame.len().min(d.maxsize as usize);
170                ptr::copy_nonoverlapping(frame.as_ptr(), d.data as *mut u8, written);
171                (*d.chunk).offset = 0;
172                (*d.chunk).size = written as u32;
173                (*d.chunk).stride = shared.stride;
174                (*d.chunk).flags = 0;
175            }
176            for i in 0..(*spa_buf).n_metas as usize {
177                let m = &*(*spa_buf).metas.add(i);
178                if m.type_ == SPA_META_HEADER && m.size >= SPA_META_HEADER_SIZE && !m.data.is_null()
179                {
180                    let h = &mut *(m.data as *mut SpaMetaHeader);
181                    h.flags = 0;
182                    h.offset = 0;
183                    h.pts = shared.frame_ts_ns.load(Ordering::Relaxed) as i64;
184                    h.dts_offset = 0;
185                    h.seq = shared.frame_seq.load(Ordering::Relaxed);
186                }
187            }
188            (*b).size = written as u64;
189        }
190        (shared.api.stream_queue_buffer)(stream, b);
191    }
192}
193
194pub struct PipeWireSink {
195    api: Api,
196    thread_loop: *mut c_void,
197    context: *mut c_void,
198    core: *mut c_void,
199    stream: *mut c_void,
200    _hook: Box<[u64; 8]>,
201    _events: Box<PwStreamEvents>,
202    _format_pod: Vec<u8>,
203    shared: Arc<Shared>,
204    failed: bool,
205}
206
207unsafe impl Send for PipeWireSink {}
208
209impl PipeWireSink {
210    /// Connect to the PipeWire daemon and publish the camera node; fails (and the sink stays off)
211    /// when the library or the daemon is missing or the format has no PipeWire equivalent.
212    pub fn connect(node_name: &str, description: &str, fmt: &RingFormat) -> Result<Self, String> {
213        let api = *api()?;
214        // The node name reaches here from Python, and a C string is what PipeWire takes.
215        let node_name = CString::new(node_name)
216            .map_err(|_| "node.name must not contain NUL bytes".to_string())?;
217        let description = CString::new(description)
218            .map_err(|_| "node.description must not contain NUL bytes".to_string())?;
219        let spa_format = if fmt.fourcc == V4L2_PIX_FMT_MJPEG {
220            None
221        } else {
222            Some(
223                spa_video_format(fmt.fourcc)
224                    .ok_or_else(|| "pixel format has no PipeWire equivalent".to_string())?,
225            )
226        };
227        unsafe {
228            let thread_loop = (api.thread_loop_new)(c"pixelflux-webcam-pw".as_ptr(), ptr::null());
229            if thread_loop.is_null() {
230                return Err("pw_thread_loop_new failed".into());
231            }
232            let mut sink = PipeWireSink {
233                api,
234                thread_loop,
235                context: ptr::null_mut(),
236                core: ptr::null_mut(),
237                stream: ptr::null_mut(),
238                _hook: Box::new([0u64; 8]),
239                _events: Box::new(PwStreamEvents {
240                    state_changed: Some(on_state_changed),
241                    param_changed: Some(on_param_changed),
242                    process: Some(on_process),
243                    ..PwStreamEvents::empty()
244                }),
245                _format_pod: format_pod(fmt, spa_format),
246                shared: Arc::new(Shared {
247                    api,
248                    stream: AtomicPtr::new(ptr::null_mut()),
249                    frame: Mutex::new(Vec::new()),
250                    frame_seq: AtomicU64::new(0),
251                    frame_ts_ns: AtomicU64::new(0),
252                    state: AtomicI32::new(0),
253                    stride: fmt.bytesperline as i32,
254                    buffers_pod: buffers_pod(fmt),
255                    meta_pod: meta_pod(),
256                }),
257                failed: false,
258            };
259            let pw_loop = (api.thread_loop_get_loop)(thread_loop);
260            sink.context = (api.context_new)(pw_loop, ptr::null_mut(), 0);
261            if sink.context.is_null() {
262                return Err("pw_context_new failed".into());
263            }
264            if (api.thread_loop_start)(thread_loop) < 0 {
265                return Err("pw_thread_loop_start failed".into());
266            }
267            (api.thread_loop_lock)(thread_loop);
268            let result = (|| -> Result<(), String> {
269                sink.core = (api.context_connect)(sink.context, ptr::null_mut(), 0);
270                if sink.core.is_null() {
271                    return Err("no PipeWire daemon reachable".into());
272                }
273                let props = (api.properties_new)(ptr::null::<c_char>());
274                if props.is_null() {
275                    return Err("pw_properties_new failed".into());
276                }
277                for (k, v) in [
278                    (c"media.class", c"Video/Source" as &CStr),
279                    (c"media.role", c"Camera"),
280                    (c"node.name", node_name.as_c_str()),
281                    (c"node.description", description.as_c_str()),
282                    (c"node.virtual", c"true"),
283                ] {
284                    (api.properties_set)(props, k.as_ptr(), v.as_ptr());
285                }
286                sink.stream = (api.stream_new)(sink.core, description.as_ptr(), props);
287                if sink.stream.is_null() {
288                    return Err("pw_stream_new failed".into());
289                }
290                sink.shared.stream.store(sink.stream, Ordering::Release);
291                let user_data = Arc::as_ptr(&sink.shared) as *mut c_void;
292                (api.stream_add_listener)(
293                    sink.stream,
294                    sink._hook.as_mut_ptr() as *mut c_void,
295                    &*sink._events,
296                    user_data,
297                );
298                let mut params: [*const c_void; 1] = [sink._format_pod.as_ptr() as *const c_void];
299                let rc = (api.stream_connect)(
300                    sink.stream,
301                    PW_DIRECTION_OUTPUT,
302                    PW_ID_ANY,
303                    PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_DRIVER,
304                    params.as_mut_ptr(),
305                    1,
306                );
307                if rc < 0 {
308                    return Err(format!("pw_stream_connect failed ({})", rc));
309                }
310                Ok(())
311            })();
312            (api.thread_loop_unlock)(thread_loop);
313            result?;
314            let deadline = Instant::now() + Duration::from_secs(3);
315            loop {
316                let st = sink.shared.state.load(Ordering::Acquire);
317                if st == PW_STREAM_STATE_ERROR {
318                    return Err("PipeWire stream entered the error state".into());
319                }
320                if st == PW_STREAM_STATE_PAUSED
321                    || st == PW_STREAM_STATE_STREAMING
322                    || Instant::now() >= deadline
323                {
324                    break;
325                }
326                std::thread::sleep(Duration::from_millis(10));
327            }
328            Ok(sink)
329        }
330    }
331
332    pub fn is_failed(&self) -> bool {
333        self.failed || self.shared.state.load(Ordering::Acquire) == PW_STREAM_STATE_ERROR
334    }
335
336    /// Whether a consumer is linked (the node is running); the latest frame is always kept so a
337    /// consumer that links later starts from it.
338    pub fn is_streaming(&self) -> bool {
339        self.shared.state.load(Ordering::Acquire) == PW_STREAM_STATE_STREAMING
340    }
341
342    /// Store one device frame and schedule a graph cycle so linked consumers receive it. With no
343    /// consumer linked nothing is copied; the first frame after a link fills the slot.
344    pub fn publish(&mut self, frame: &[u8], ts_ns: u64) {
345        if self.is_failed() || !self.is_streaming() {
346            return;
347        }
348        {
349            let mut slot = self.shared.frame.lock().unwrap_or_else(|e| e.into_inner());
350            slot.clear();
351            slot.extend_from_slice(frame);
352        }
353        self.shared.frame_seq.fetch_add(1, Ordering::Relaxed);
354        self.shared.frame_ts_ns.store(ts_ns, Ordering::Relaxed);
355        unsafe {
356            (self.api.thread_loop_lock)(self.thread_loop);
357            (self.api.stream_trigger_process)(self.stream);
358            (self.api.thread_loop_unlock)(self.thread_loop);
359        }
360    }
361}
362
363impl Drop for PipeWireSink {
364    fn drop(&mut self) {
365        unsafe {
366            (self.api.thread_loop_lock)(self.thread_loop);
367            self.shared.stream.store(ptr::null_mut(), Ordering::Release);
368            if !self.stream.is_null() {
369                (self.api.stream_disconnect)(self.stream);
370                (self.api.stream_destroy)(self.stream);
371            }
372            if !self.core.is_null() {
373                (self.api.core_disconnect)(self.core);
374            }
375            (self.api.thread_loop_unlock)(self.thread_loop);
376            (self.api.thread_loop_stop)(self.thread_loop);
377            if !self.context.is_null() {
378                (self.api.context_destroy)(self.context);
379            }
380            (self.api.thread_loop_destroy)(self.thread_loop);
381        }
382    }
383}
384
385#[cfg(test)]
386mod tests {
387    use super::*;
388
389    fn u32_at(v: &[u8], off: usize) -> u32 {
390        u32::from_ne_bytes(v[off..off + 4].try_into().unwrap())
391    }
392
393    /// One key of an object pod: its id, value type, and value payload.
394    type PodKey = (u32, u32, Vec<u8>);
395
396    /// Walk an object pod: (object type, param id, [keys]).
397    fn parse_object(pod: &[u8]) -> (u32, u32, Vec<PodKey>) {
398        assert_eq!(pod.len() % 8, 0);
399        let (ty, id, props) = object_props(pod).unwrap();
400        (
401            ty,
402            id,
403            props
404                .into_iter()
405                .map(|p| (p.key, p.ty, p.payload.to_vec()))
406                .collect(),
407        )
408    }
409
410    #[test]
411    fn format_pod_is_a_well_formed_object() {
412        let fmt = RingFormat::raw(V4L2_PIX_FMT_YUV420, 1280, 720, 30, 1).unwrap();
413        let (ty, id, props) = parse_object(&format_pod(&fmt, Some(SPA_VIDEO_FORMAT_I420)));
414        assert_eq!((ty, id), (SPA_TYPE_OBJECT_FORMAT, SPA_PARAM_ENUM_FORMAT));
415        let get = |k: u32| props.iter().find(|p| p.0 == k).cloned().unwrap();
416        assert_eq!(get(SPA_FORMAT_MEDIA_TYPE).1, SPA_TYPE_ID);
417        assert_eq!(
418            u32_at(&get(SPA_FORMAT_MEDIA_TYPE).2, 0),
419            SPA_MEDIA_TYPE_VIDEO
420        );
421        assert_eq!(
422            u32_at(&get(SPA_FORMAT_MEDIA_SUBTYPE).2, 0),
423            SPA_MEDIA_SUBTYPE_RAW
424        );
425        assert_eq!(
426            u32_at(&get(SPA_FORMAT_VIDEO_FORMAT).2, 0),
427            SPA_VIDEO_FORMAT_I420
428        );
429        let size = get(SPA_FORMAT_VIDEO_SIZE);
430        assert_eq!(size.1, SPA_TYPE_RECTANGLE);
431        assert_eq!((u32_at(&size.2, 0), u32_at(&size.2, 4)), (1280, 720));
432        let rate = get(SPA_FORMAT_VIDEO_FRAMERATE);
433        assert_eq!(rate.1, SPA_TYPE_FRACTION);
434        assert_eq!((u32_at(&rate.2, 0), u32_at(&rate.2, 4)), (30, 1));
435    }
436
437    #[test]
438    fn mjpg_format_pod_names_no_pixel_format() {
439        let fmt = RingFormat::compressed(V4L2_PIX_FMT_MJPEG, 640, 480, 30, 1).unwrap();
440        let (ty, id, props) = parse_object(&format_pod(&fmt, None));
441        assert_eq!((ty, id), (SPA_TYPE_OBJECT_FORMAT, SPA_PARAM_ENUM_FORMAT));
442        let get = |k: u32| props.iter().find(|p| p.0 == k).cloned();
443        assert_eq!(
444            u32_at(&get(SPA_FORMAT_MEDIA_SUBTYPE).unwrap().2, 0),
445            SPA_MEDIA_SUBTYPE_MJPG
446        );
447        assert!(get(SPA_FORMAT_VIDEO_FORMAT).is_none());
448        let size = get(SPA_FORMAT_VIDEO_SIZE).unwrap();
449        assert_eq!((u32_at(&size.2, 0), u32_at(&size.2, 4)), (640, 480));
450        let (_, _, bprops) = parse_object(&buffers_pod(&fmt));
451        let bget = |k: u32| bprops.iter().find(|p| p.0 == k).cloned().unwrap();
452        assert_eq!(u32_at(&bget(SPA_PARAM_BUFFERS_SIZE).2, 0), 640 * 480 * 2);
453        assert_eq!(u32_at(&bget(SPA_PARAM_BUFFERS_STRIDE).2, 0), 0);
454    }
455
456    #[test]
457    fn buffers_pod_carries_frame_geometry() {
458        let fmt = RingFormat::raw(V4L2_PIX_FMT_YUYV, 640, 480, 30, 1).unwrap();
459        let (ty, id, props) = parse_object(&buffers_pod(&fmt));
460        assert_eq!((ty, id), (SPA_TYPE_OBJECT_PARAM_BUFFERS, SPA_PARAM_BUFFERS));
461        let get = |k: u32| props.iter().find(|p| p.0 == k).cloned().unwrap();
462        assert_eq!(
463            (
464                get(SPA_PARAM_BUFFERS_SIZE).1,
465                u32_at(&get(SPA_PARAM_BUFFERS_SIZE).2, 0)
466            ),
467            (SPA_TYPE_INT, 640 * 480 * 2)
468        );
469        assert_eq!(u32_at(&get(SPA_PARAM_BUFFERS_STRIDE).2, 0), 1280);
470        assert_eq!(u32_at(&get(SPA_PARAM_BUFFERS_BLOCKS).2, 0), 1);
471        let buffers = get(SPA_PARAM_BUFFERS_BUFFERS);
472        assert_eq!(buffers.1, SPA_TYPE_CHOICE);
473        assert_eq!(u32_at(&buffers.2, 0), SPA_CHOICE_RANGE);
474        assert_eq!(
475            (u32_at(&buffers.2, 8), u32_at(&buffers.2, 12)),
476            (4, SPA_TYPE_INT)
477        );
478        assert_eq!(&buffers.2[16..28], &[4u8, 0, 0, 0, 2, 0, 0, 0, 8, 0, 0, 0]);
479        let dt = get(SPA_PARAM_BUFFERS_DATATYPE);
480        assert_eq!(u32_at(&dt.2, 0), SPA_CHOICE_FLAGS);
481        assert_eq!(
482            u32_at(&dt.2, 16),
483            (1 << SPA_DATA_MEMFD) | (1 << SPA_DATA_MEMPTR)
484        );
485        let (mty, mid, mprops) = parse_object(&meta_pod());
486        assert_eq!((mty, mid), (SPA_TYPE_OBJECT_PARAM_META, SPA_PARAM_META));
487        assert_eq!(u32_at(&mprops[0].2, 0), SPA_META_HEADER);
488        assert_eq!(u32_at(&mprops[1].2, 0), SPA_META_HEADER_SIZE);
489    }
490}
pixelflux/webcam/pipewire.rs

pixelflux/webcam/
pipewire.rs
