#![forbid(unsafe_code)]
use std::collections::VecDeque;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::SyncSender;
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::Duration;
use crate::camera::{CameraCapture, CameraCaptureConfig, CaptureOutputPreference};
use crate::{CaptureError, Select};
use mediaway_common::{
Bytes, CodecKind, PixelFormat, Rational, StreamInfo, VideoFrame, VideoFrameStorage,
VideoGeometry,
};
use v4l::buffer::Type as V4lBufferType;
use v4l::capability::Flags as V4lCapabilityFlags;
use v4l::io::mmap::Stream as MmapStream;
use v4l::io::traits::CaptureStream as _;
use v4l::video::Capture as _;
use v4l::{Device as V4lDevice, Format as V4lFormat, FourCC};
const FRAME_QUEUE_CAP: usize = 4;
const CAPTURE_BUFFER_COUNT: u32 = 4;
const STREAM_POLL_TIMEOUT: Duration = Duration::from_millis(200);
const FALLBACK_WIDTH: u32 = 640;
const FALLBACK_HEIGHT: u32 = 480;
struct FrameQueue {
frames: Mutex<VecDeque<VideoFrame>>,
}
struct CameraSession {
stream_info: StreamInfo,
queue: Arc<FrameQueue>,
stop: Arc<AtomicBool>,
worker: Option<JoinHandle<()>>,
}
pub struct LinuxCameraCapture {
inner: Option<CameraSession>,
}
impl LinuxCameraCapture {
pub fn open(config: &CameraCaptureConfig) -> Result<Self, CaptureError> {
if config.select != Select::Default {
return Err(CaptureError::Unsupported);
}
let device = 0usize;
if config.output != CaptureOutputPreference::CpuFramesOk {
return Err(CaptureError::Unsupported);
}
let queue = Arc::new(FrameQueue {
frames: Mutex::new(VecDeque::new()),
});
let stop = Arc::new(AtomicBool::new(false));
let queue_worker = Arc::clone(&queue);
let stop_worker = Arc::clone(&stop);
let time_base = config.time_base;
let (tx_info, rx_info) = std::sync::mpsc::sync_channel(1);
let worker = thread::Builder::new()
.name("mediaway-v4l2-camera".into())
.spawn(move || {
run_camera_worker(device, time_base, &queue_worker, &stop_worker, &tx_info);
})
.map_err(|_| CaptureError::Backend)?;
let stream_info = rx_info.recv().map_err(|_| CaptureError::Backend)??;
Ok(Self {
inner: Some(CameraSession {
stream_info,
queue,
stop,
worker: Some(worker),
}),
})
}
}
impl CameraCapture for LinuxCameraCapture {
fn stream_info(&self) -> &StreamInfo {
#[allow(
clippy::option_if_let_else,
reason = "map_or_else forces 'static vs 'self lifetime clash"
)]
if let Some(s) = self.inner.as_ref() {
&s.stream_info
} else {
closed_video_info()
}
}
fn poll_frame(&mut self) -> Result<Option<VideoFrame>, CaptureError> {
let Some(session) = self.inner.as_ref() else {
return Err(CaptureError::Closed);
};
let mut q = session
.queue
.frames
.lock()
.map_err(|_| CaptureError::Backend)?;
Ok(q.pop_front())
}
fn release_frame(&mut self) -> Result<(), CaptureError> {
if self.inner.is_none() {
return Err(CaptureError::Closed);
}
Ok(())
}
fn close(&mut self) -> Result<(), CaptureError> {
let Some(mut session) = self.inner.take() else {
return Ok(());
};
session.stop.store(true, Ordering::SeqCst);
if let Some(h) = session.worker.take() {
let _ = h.join();
}
Ok(())
}
}
impl Drop for LinuxCameraCapture {
fn drop(&mut self) {
let _ = self.close();
}
}
fn closed_video_info() -> &'static StreamInfo {
use std::sync::OnceLock;
static INFO: OnceLock<StreamInfo> = OnceLock::new();
INFO.get_or_init(|| StreamInfo::Video {
id: 0,
codec: CodecKind::RawVideo,
time_base: Rational::new(1, 30),
geometry: VideoGeometry {
width: 0,
height: 0,
},
extra_data: Bytes::new(),
})
}
fn run_camera_worker(
device_index: usize,
time_base: Rational,
queue: &FrameQueue,
stop: &AtomicBool,
tx_info: &SyncSender<Result<StreamInfo, CaptureError>>,
) {
let (device, stream_info, format, width, height, stride) =
match open_and_negotiate(device_index, time_base) {
Ok(v) => v,
Err(e) => {
let _ = tx_info.send(Err(e));
return;
}
};
let Ok(mut stream) =
MmapStream::with_buffers(&device, V4lBufferType::VideoCapture, CAPTURE_BUFFER_COUNT)
else {
let _ = tx_info.send(Err(CaptureError::Backend));
return;
};
stream.set_timeout(STREAM_POLL_TIMEOUT);
let _ = tx_info.send(Ok(stream_info));
let mut pts: i64 = 0;
while !stop.load(Ordering::Relaxed) {
match stream.next() {
Ok((bytes, _meta)) => {
if let Some(data) = pack_frame_bytes(bytes, format, width, height, stride) {
push_frame(queue, format, width, height, pts, data);
pts = pts.saturating_add(1);
}
}
Err(e) if e.kind() == std::io::ErrorKind::TimedOut => {}
Err(_) => break,
}
}
}
fn open_and_negotiate(
device_index: usize,
time_base: Rational,
) -> Result<(V4lDevice, StreamInfo, PixelFormat, u32, u32, u32), CaptureError> {
let nodes = enumerate_capture_nodes();
let path = nodes.get(device_index).ok_or(CaptureError::InvalidInput)?;
let device = V4lDevice::with_path(path).map_err(|e| map_io_error(&e))?;
let available: Vec<[u8; 4]> = device
.enum_formats()
.map_err(|e| map_io_error(&e))?
.into_iter()
.map(|d| d.fourcc.repr)
.collect();
let (format, fourcc) = pick_capture_format(&available).ok_or(CaptureError::Unsupported)?;
let current = device.format().map_err(|e| map_io_error(&e))?;
let (width, height) = if current.width > 0 && current.height > 0 {
(current.width, current.height)
} else {
(FALLBACK_WIDTH, FALLBACK_HEIGHT)
};
let requested = V4lFormat::new(width, height, FourCC::new(&fourcc));
let negotiated = device
.set_format(&requested)
.map_err(|e| map_io_error(&e))?;
if negotiated.fourcc != requested.fourcc {
return Err(CaptureError::Unsupported);
}
if negotiated.width == 0 || negotiated.height == 0 {
return Err(CaptureError::Backend);
}
let stride = if negotiated.stride > 0 {
negotiated.stride
} else {
min_stride(format, negotiated.width)
};
let info = StreamInfo::Video {
id: 0,
codec: CodecKind::RawVideo,
time_base,
geometry: VideoGeometry {
width: negotiated.width,
height: negotiated.height,
},
extra_data: Bytes::new(),
};
Ok((
device,
info,
format,
negotiated.width,
negotiated.height,
stride,
))
}
fn enumerate_capture_nodes() -> Vec<PathBuf> {
let mut nodes = v4l::context::enum_devices();
nodes.sort_by_key(v4l::context::Node::index);
nodes
.into_iter()
.filter_map(|node| {
let device = V4lDevice::with_path(node.path()).ok()?;
let caps = device.query_caps().ok()?;
caps.capabilities
.contains(V4lCapabilityFlags::VIDEO_CAPTURE)
.then(|| node.path().to_path_buf())
})
.collect()
}
fn map_io_error(e: &std::io::Error) -> CaptureError {
match e.kind() {
std::io::ErrorKind::PermissionDenied => CaptureError::AccessDenied,
std::io::ErrorKind::NotFound => CaptureError::InvalidInput,
_ => CaptureError::Backend,
}
}
const PREFERRED_FOURCCS: [(PixelFormat, [u8; 4]); 3] = [
(PixelFormat::Yuyv, *b"YUYV"),
(PixelFormat::Nv12, *b"NV12"),
(PixelFormat::I420, *b"YU12"),
];
fn pick_capture_format(available: &[[u8; 4]]) -> Option<(PixelFormat, [u8; 4])> {
PREFERRED_FOURCCS
.into_iter()
.find(|(_, fourcc)| available.contains(fourcc))
}
const fn min_stride(format: PixelFormat, width: u32) -> u32 {
match format {
PixelFormat::Yuyv => width * 2,
_ => width,
}
}
fn copy_rows(
src: &[u8],
out: &mut Vec<u8>,
row_bytes: usize,
rows: usize,
stride: usize,
) -> Option<()> {
for row in 0..rows {
let start = row.checked_mul(stride)?;
let end = start.checked_add(row_bytes)?;
out.extend_from_slice(src.get(start..end)?);
}
Some(())
}
fn pack_frame_bytes(
src: &[u8],
format: PixelFormat,
width: u32,
height: u32,
stride: u32,
) -> Option<Bytes> {
let (w, h, stride) = (width as usize, height as usize, stride as usize);
if w == 0 || h == 0 || stride == 0 {
return None;
}
let mut out = Vec::new();
match format {
PixelFormat::Yuyv if stride >= w * 2 => {
copy_rows(src, &mut out, w * 2, h, stride)?;
}
PixelFormat::Nv12 if stride >= w => {
copy_rows(src, &mut out, w, h, stride)?;
let chroma = src.get(stride.checked_mul(h)?..)?;
copy_rows(chroma, &mut out, w, h / 2, stride)?;
}
PixelFormat::I420 if stride >= w => {
copy_rows(src, &mut out, w, h, stride)?;
let chroma_stride = stride / 2;
let chroma_w = w / 2;
let chroma_rows = h / 2;
let u_plane = src.get(stride.checked_mul(h)?..)?;
copy_rows(u_plane, &mut out, chroma_w, chroma_rows, chroma_stride)?;
let v_plane = u_plane.get(chroma_stride.checked_mul(chroma_rows)?..)?;
copy_rows(v_plane, &mut out, chroma_w, chroma_rows, chroma_stride)?;
}
_ => return None,
}
Some(Bytes::from(out))
}
fn push_frame(
queue: &FrameQueue,
format: PixelFormat,
width: u32,
height: u32,
pts: i64,
data: Bytes,
) {
let frame = VideoFrame {
pts,
duration: 1,
width,
height,
format,
storage: VideoFrameStorage::Cpu { data },
};
if let Ok(mut q) = queue.frames.lock() {
if q.len() >= FRAME_QUEUE_CAP {
let _ = q.pop_front();
}
q.push_back(frame);
}
}
pub(crate) fn enumerate_camera_paths() -> Vec<PathBuf> {
enumerate_capture_nodes()
}
#[cfg(test)]
#[path = "camera_tests.rs"]
mod tests;