diff --git a/src/auth/middleware.rs b/src/auth/middleware.rs index 40d75cdf..62c4f807 100644 --- a/src/auth/middleware.rs +++ b/src/auth/middleware.rs @@ -93,6 +93,6 @@ fn is_public_endpoint(path: &str) -> bool { fn is_setup_public_endpoint(path: &str) -> bool { matches!( path, - "/setup" | "/setup/init" | "/devices" | "/stream/codecs" + "/setup" | "/setup/init" | "/devices" | "/video/input-status" | "/stream/codecs" ) } diff --git a/src/main.rs b/src/main.rs index df93abc5..e407bb39 100644 --- a/src/main.rs +++ b/src/main.rs @@ -426,25 +426,9 @@ async fn main() -> anyhow::Result<()> { .update_video_config(actual_resolution, actual_format, actual_fps) .await; if let Some(device_path) = device_path { - let (subdev_path, bridge_kind, v4l2_driver) = streamer - .current_device() - .await - .map(|d| { - ( - d.subdev_path.clone(), - d.bridge_kind.clone(), - Some(d.driver.clone()), - ) - }) - .unwrap_or((None, None, None)); + let device_info = streamer.current_device().await; webrtc_streamer - .set_capture_device( - device_path, - jpeg_quality, - subdev_path, - bridge_kind, - v4l2_driver, - ) + .set_capture_device(device_path, jpeg_quality, device_info) .await; tracing::debug!("WebRTC streamer configured for direct capture"); } else { diff --git a/src/video/capture/linux.rs b/src/video/capture/linux.rs index 61913715..3c1a5161 100644 --- a/src/video/capture/linux.rs +++ b/src/video/capture/linux.rs @@ -23,6 +23,7 @@ use v4l2r::{Format as V4l2rFormat, PixelFormat as V4l2rPixelFormat, QueueType}; use crate::error::{AppError, Result}; use crate::video::device::bridge::{self as csi_bridge, CsiBridgeKind, ProbeResult}; +use crate::video::device::VideoControlMode; use crate::video::format::{PixelFormat, Resolution}; use crate::video::signal::SignalStatus; @@ -65,6 +66,7 @@ pub struct CaptureStream { queue: QueueType, resolution: Resolution, format: PixelFormat, + source_fps: Option, stride: u32, timeout: Duration, mappings: Vec>, @@ -92,6 +94,7 @@ impl CaptureStream { buffer_count, timeout, BridgeContext::default(), + VideoControlMode::Configurable, ) } @@ -104,6 +107,7 @@ impl CaptureStream { buffer_count: u32, timeout: Duration, bridge: BridgeContext, + control_mode: VideoControlMode, ) -> Result { // Probe subdev before video open (RK628: no-signal must not reach capture STREAMON). let mut subdev_fd_opt: Option = None; @@ -154,9 +158,8 @@ impl CaptureStream { let caps: V4l2rCapability = ioctl::querycap(&fd) .map_err(|e| AppError::VideoError(format!("Failed to query capabilities: {}", e)))?; let caps_flags = caps.device_caps(); - let driver_name = caps.driver.to_string(); - let is_csi_bridge = is_csi_bridge_driver(&driver_name); - let is_native_hdmirx = is_native_hdmirx_driver(&driver_name); + let is_source_following = control_mode == VideoControlMode::SourceFollowing; + let is_native_hdmirx = bridge.kind == Some(CsiBridgeKind::RkHdmirx); // Prefer multi-planar capture when available, as it is required for some // devices/pixel formats (e.g. NV12 via VIDEO_CAPTURE_MPLANE). @@ -181,7 +184,7 @@ impl CaptureStream { fps: mode.fps, signature: None, }) - } else if is_csi_bridge { + } else if is_source_following { // The native RK3588 HDMI RX driver already latches detected // timings while locking the input. S_DV_TIMINGS is unnecessary // there and rejects some otherwise valid sources whose measured @@ -196,7 +199,7 @@ impl CaptureStream { // `v4l2-ctl --set-fmt-video=width=…,height=…`). let mut fmt: V4l2rFormat = match ( ioctl::g_fmt::(&fd, queue), - is_csi_bridge, + is_source_following, dv_mode.as_ref(), ) { (Ok(f), _, _) if f.width > 0 && f.height > 0 => f, @@ -323,6 +326,7 @@ impl CaptureStream { queue, resolution: actual_resolution, format: actual_format, + source_fps: dv_mode.as_ref().and_then(|mode| mode.fps), stride, timeout, mappings, @@ -366,6 +370,10 @@ impl CaptureStream { self.format } + pub fn source_fps(&self) -> Option { + self.source_fps + } + pub fn stride(&self) -> u32 { self.stride } @@ -696,19 +704,6 @@ impl Drop for CaptureStream { } } -/// Driver-name check for CSI/HDMI bridge devices (rk_hdmirx, rkcif, tc358743, -/// …) that expose DV timings. Kept in sync with `video::device::is_csi_hdmi_bridge` -/// but queries the raw V4L2 driver string so we don't need a full -/// `VideoDeviceInfo` at `CaptureStream::open` time. -fn is_csi_bridge_driver(driver: &str) -> bool { - let d = driver.to_ascii_lowercase(); - is_native_hdmirx_driver(&d) || d == "rkcif" || d == "tc358743" || d.starts_with("rkcif") -} - -fn is_native_hdmirx_driver(driver: &str) -> bool { - driver.eq_ignore_ascii_case("rk_hdmirx") || driver.eq_ignore_ascii_case("snps_hdmirx") -} - /// Drain any pending `V4L2_EVENT_*` events on `fd`. Used after POLLPRI to /// clear the queue so the next poll doesn't immediately wake up on stale /// state. Capped at 16 events per call. @@ -765,34 +760,14 @@ impl NativeHdmirxState { struct DvTimingsSignature { width: u32, height: u32, - total_width: u32, - total_height: u32, - pixelclock: u64, interlaced: bool, } impl DvTimingsSignature { fn matches(self, other: Self) -> bool { - let self_total = u128::from(self.total_width) * u128::from(self.total_height); - let other_total = u128::from(other.total_width) * u128::from(other.total_height); - // RK3588 BSPs can describe the same active mode with different - // blanking/pixel-clock pairs (observed for 1080p60: 2200×1125 at - // 148.5 MHz and 2752×1125 at 185.448 MHz). Compare the resulting - // frame rates by cross multiplication instead of requiring identical - // totals. A 0.5% tolerance absorbs measurement jitter and 59.94/60, - // while still distinguishing normal 50/60 transitions. - let self_rate = u128::from(self.pixelclock) * other_total; - let other_rate = u128::from(other.pixelclock) * self_total; - let rate_delta = self_rate.abs_diff(other_rate); - let rate_tolerance = (self_rate.max(other_rate) / 200).max(1); self.width == other.width && self.height == other.height && self.interlaced == other.interlaced - && self.pixelclock != 0 - && other.pixelclock != 0 - && self_total != 0 - && other_total != 0 - && rate_delta <= rate_tolerance } } @@ -804,20 +779,9 @@ fn dv_timings_signature(timings: &v4l2_dv_timings) -> Option let bt = unsafe { timings.__bindgen_anon_1.bt }; let width = bt.width; let height = bt.height; - let total_width = width - .checked_add(bt.hfrontporch)? - .checked_add(bt.hsync)? - .checked_add(bt.hbackporch)?; - let total_height = height - .checked_add(bt.vfrontporch)? - .checked_add(bt.vsync)? - .checked_add(bt.vbackporch)?; Some(DvTimingsSignature { width, height, - total_width, - total_height, - pixelclock: bt.pixelclock, interlaced: bt.interlaced != 0, }) } @@ -856,7 +820,7 @@ fn probe_dv_timings(fd: &File, apply: bool) -> Result { | QueryDvTimingsError::IoctlError(Errno::ETIMEDOUT) => SignalStatus::NoSync, QueryDvTimingsError::IoctlError(_) => SignalStatus::NoSignal, }; - info!( + debug!( "VIDIOC_QUERY_DV_TIMINGS failed: {} -> SignalStatus::{:?}", err, status ); @@ -973,43 +937,28 @@ fn set_fps(fd: &File, queue: QueueType, fps: u32) -> std::result::Result<(), ioc #[cfg(test)] mod tests { - use super::{is_native_hdmirx_driver, DvTimingsSignature, NativeHdmirxState}; + use super::{DvTimingsSignature, NativeHdmirxState}; use crate::video::format::PixelFormat; - fn timing(pixelclock: u64) -> DvTimingsSignature { + fn timing() -> DvTimingsSignature { DvTimingsSignature { width: 1920, height: 1080, - total_width: 2200, - total_height: 1125, - pixelclock, interlaced: false, } } #[test] - fn recognizes_vendor_and_upstream_native_hdmirx_drivers() { - assert!(is_native_hdmirx_driver("rk_hdmirx")); - assert!(is_native_hdmirx_driver("SNPS_HDMIRX")); - assert!(!is_native_hdmirx_driver("rkcif")); - } + fn timing_match_uses_only_active_geometry_and_scan_mode() { + assert!(timing().matches(timing())); - #[test] - fn timing_match_tolerates_measurement_jitter_but_not_mode_changes() { - assert!(timing(148_500_000).matches(timing(148_000_000))); - assert!(!timing(148_500_000).matches(timing(120_000_000))); + let mut different_width = timing(); + different_width.width = 1280; + assert!(!timing().matches(different_width)); - let mut equivalent_blanking = timing(185_448_000); - equivalent_blanking.total_width = 2752; - assert!(timing(148_500_000).matches(equivalent_blanking)); - - let mut different_refresh = timing(148_500_000); - different_refresh.total_width = 2640; - assert!(!timing(148_500_000).matches(different_refresh)); - - let mut interlaced = timing(148_500_000); + let mut interlaced = timing(); interlaced.interlaced = true; - assert!(!timing(148_500_000).matches(interlaced)); + assert!(!timing().matches(interlaced)); } #[test] @@ -1019,14 +968,16 @@ mod tests { width: 1920, height: 1080, pixelformat: bgr24, - timings: Some(timing(148_500_000)), + timings: Some(timing()), }; assert!(state.format_matches(1920, 1080, bgr24)); assert!(!state.format_matches(1280, 720, bgr24)); assert!(!state.format_matches(1920, 1080, PixelFormat::Nv12.to_v4l2r())); - assert_eq!(state.timings_match(Some(timing(148_000_000))), Some(true)); - assert_eq!(state.timings_match(Some(timing(120_000_000))), Some(false)); + assert_eq!(state.timings_match(Some(timing())), Some(true)); + let mut interlaced = timing(); + interlaced.interlaced = true; + assert_eq!(state.timings_match(Some(interlaced)), Some(false)); assert_eq!(state.timings_match(None), None); let no_timing_state = NativeHdmirxState { diff --git a/src/video/capture/runtime.rs b/src/video/capture/runtime.rs index a09fcdde..fd65a16c 100644 --- a/src/video/capture/runtime.rs +++ b/src/video/capture/runtime.rs @@ -3,6 +3,7 @@ use std::time::Duration; use crate::error::AppError; use crate::video::capture::status::signal_status_from_capture_kind; +use crate::video::device::VideoControlMode; use crate::video::format::{PixelFormat, Resolution}; use crate::video::signal::SignalStatus; @@ -23,6 +24,7 @@ pub fn open_capture_stream( buffer_count: u32, timeout: Duration, bridge_ctx: BridgeContext, + control_mode: VideoControlMode, ) -> Result { CaptureStream::open_with_bridge( device_path, @@ -32,6 +34,7 @@ pub fn open_capture_stream( buffer_count.max(1), timeout, bridge_ctx, + control_mode, ) } @@ -43,6 +46,7 @@ pub fn open_capture_stream_for_retry( buffer_count: u32, timeout: Duration, bridge_ctx: BridgeContext, + control_mode: VideoControlMode, is_device_lost_message: impl FnOnce(&str) -> bool, ) -> CaptureOpenResult { match open_capture_stream( @@ -53,6 +57,7 @@ pub fn open_capture_stream_for_retry( buffer_count, timeout, bridge_ctx, + control_mode, ) { Ok(stream) => CaptureOpenResult::Opened(stream), Err(AppError::CaptureNoSignal { kind }) => { diff --git a/src/video/capture/windows.rs b/src/video/capture/windows.rs index e3d2cbf5..1e8df15e 100644 --- a/src/video/capture/windows.rs +++ b/src/video/capture/windows.rs @@ -4,7 +4,9 @@ use std::time::Duration; use crate::error::{AppError, Result}; use crate::video::device::bridge::{CsiBridgeKind, ProbeResult}; -use crate::video::device::{directshow_display_name_from_path, normalize_windows_device_path}; +use crate::video::device::{ + directshow_display_name_from_path, normalize_windows_device_path, VideoControlMode, +}; use crate::video::format::{PixelFormat, Resolution}; pub const SOURCE_CHANGED_MARKER: &str = "dshow_source_changed"; @@ -95,8 +97,9 @@ impl CaptureStream { buffer_count: u32, timeout: Duration, bridge: BridgeContext, + control_mode: VideoControlMode, ) -> Result { - let _ = bridge; + let _ = (bridge, control_mode); Self::open(device_path, resolution, format, fps, buffer_count, timeout) } @@ -108,6 +111,10 @@ impl CaptureStream { self.format } + pub fn source_fps(&self) -> Option { + None + } + pub fn stride(&self) -> u32 { self.stride } diff --git a/src/video/device/bridge.rs b/src/video/device/bridge.rs index 174688c8..96665fcb 100644 --- a/src/video/device/bridge.rs +++ b/src/video/device/bridge.rs @@ -53,6 +53,7 @@ pub enum ProbeResult { NoSync, OutOfRange, NoSignal, + Unavailable, } impl ProbeResult { @@ -63,6 +64,7 @@ impl ProbeResult { ProbeResult::NoSync => Some(SignalStatus::NoSync), ProbeResult::OutOfRange => Some(SignalStatus::OutOfRange), ProbeResult::NoSignal => Some(SignalStatus::NoSignal), + ProbeResult::Unavailable => None, } } @@ -138,13 +140,23 @@ pub fn open_subdev(path: &Path) -> io::Result { pub fn probe_signal(subdev_fd: &impl AsRawFd, kind: CsiBridgeKind) -> ProbeResult { match ioctl::query_dv_timings::(subdev_fd) { Ok(timings) => classify_timings(timings, kind), - Err(QueryDvTimingsError::NoLink) => ProbeResult::NoCable, - Err(QueryDvTimingsError::UnstableSignal) => ProbeResult::NoSync, - Err(QueryDvTimingsError::IoctlError(Errno::ERANGE)) => ProbeResult::OutOfRange, - Err(QueryDvTimingsError::IoctlError(Errno::EIO | Errno::EREMOTEIO | Errno::ETIMEDOUT)) => { + Err(error) => classify_query_error(&error, kind), + } +} + +fn classify_query_error(error: &QueryDvTimingsError, kind: CsiBridgeKind) -> ProbeResult { + match error { + QueryDvTimingsError::NoLink => ProbeResult::NoCable, + QueryDvTimingsError::UnstableSignal => ProbeResult::NoSync, + QueryDvTimingsError::IoctlError(Errno::ERANGE) => ProbeResult::OutOfRange, + QueryDvTimingsError::IoctlError(Errno::EIO | Errno::EREMOTEIO | Errno::ETIMEDOUT) => { ProbeResult::NoSync } - Err(QueryDvTimingsError::Unsupported) | Err(QueryDvTimingsError::IoctlError(_)) => { + QueryDvTimingsError::Unsupported + | QueryDvTimingsError::IoctlError( + Errno::ENOTTY | Errno::EINVAL | Errno::ENOSYS | Errno::EOPNOTSUPP, + ) if kind == CsiBridgeKind::Unknown => ProbeResult::Unavailable, + QueryDvTimingsError::Unsupported | QueryDvTimingsError::IoctlError(_) => { ProbeResult::NoSignal } } @@ -179,7 +191,7 @@ pub fn probe_signal_thread_timeout( Some(r) } Err(mpsc::RecvTimeoutError::Timeout) => { - warn!( + debug!( "QUERY_DV_TIMINGS exceeded {:?} (RK628 HDMI mode change?) — abandoning probe thread", limit ); @@ -352,4 +364,27 @@ mod tests { ); assert_eq!(CsiBridgeKind::from_subdev_name("mystery"), None); } + + #[test] + fn query_errno_mapping_distinguishes_signal_loss_from_unsupported_nodes() { + assert!(matches!( + classify_query_error(&QueryDvTimingsError::NoLink, CsiBridgeKind::RkHdmirx), + ProbeResult::NoCable + )); + assert!(matches!( + classify_query_error( + &QueryDvTimingsError::UnstableSignal, + CsiBridgeKind::RkHdmirx + ), + ProbeResult::NoSync + )); + assert!(matches!( + classify_query_error(&QueryDvTimingsError::Unsupported, CsiBridgeKind::RkHdmirx), + ProbeResult::NoSignal + )); + assert!(matches!( + classify_query_error(&QueryDvTimingsError::Unsupported, CsiBridgeKind::Unknown), + ProbeResult::Unavailable + )); + } } diff --git a/src/video/device/disabled_bridge.rs b/src/video/device/disabled_bridge.rs index 6da93963..0059c767 100644 --- a/src/video/device/disabled_bridge.rs +++ b/src/video/device/disabled_bridge.rs @@ -22,6 +22,7 @@ pub enum ProbeResult { NoSync, OutOfRange, NoSignal, + Unavailable, } impl ProbeResult { @@ -32,6 +33,7 @@ impl ProbeResult { ProbeResult::NoSync => Some(SignalStatus::NoSync), ProbeResult::OutOfRange => Some(SignalStatus::OutOfRange), ProbeResult::NoSignal => Some(SignalStatus::NoSignal), + ProbeResult::Unavailable => None, } } diff --git a/src/video/device/linux.rs b/src/video/device/linux.rs index dc7a0fcf..8c2551ce 100644 --- a/src/video/device/linux.rs +++ b/src/video/device/linux.rs @@ -17,7 +17,9 @@ use v4l2r::nix::errno::Errno; use v4l2r::{Format as V4l2rFormat, QueueType}; use super::bridge as csi_bridge; -use super::{is_rk_hdmirx_driver, is_rkcif_driver}; +use super::{ + control_mode, is_rk_hdmirx_driver, is_rkcif_driver, VideoControlMode, VideoInputStatus, +}; use crate::error::{AppError, Result}; use crate::video::format::{PixelFormat, Resolution}; @@ -48,6 +50,8 @@ pub struct VideoDeviceInfo { /// Whether an HDMI signal is currently detected (CSI/HDMI bridge devices only; /// always `true` for USB capture cards). pub has_signal: bool, + pub control_mode: VideoControlMode, + pub input_status: VideoInputStatus, /// Path of the bridge subdev (`/dev/v4l-subdevN`) paired with this /// capture node, if any. On Rockchip boards that wire an RK628 / /// TC358746 / RK-HDMIRX through `rkcif`, `QUERY_DV_TIMINGS`, @@ -129,6 +133,16 @@ pub struct VideoDevice { fd: File, } +struct LiveInputProbe { + control_mode: VideoControlMode, + input_status: VideoInputStatus, + has_signal: bool, + hdmi_mode: Option<(u32, u32, Option)>, + hdmi_fps: Option, + subdev_path: Option, + bridge_kind: Option, +} + impl VideoDevice { /// Open a video device by path pub fn open(path: impl AsRef) -> Result { @@ -173,6 +187,106 @@ impl VideoDevice { }) } + pub fn input_status(&self) -> Result { + let caps: V4l2rCapability = ioctl::querycap(&self.fd) + .map_err(|e| AppError::VideoError(format!("Failed to query capabilities: {}", e)))?; + Ok(self.probe_live_input(&caps).input_status) + } + + fn probe_live_input(&self, caps: &V4l2rCapability) -> LiveInputProbe { + let control_mode = control_mode(&caps.driver, &caps.card); + if control_mode == VideoControlMode::Configurable { + let input_status = self + .get_format() + .ok() + .and_then(|fmt| { + PixelFormat::from_v4l2r(fmt.pixelformat) + .map(|format| (format, fmt.width, fmt.height)) + }) + .map(|(format, width, height)| { + VideoInputStatus::locked( + format, + width, + height, + self.current_parm_fps().unwrap_or(0.0), + ) + }) + .unwrap_or_else(VideoInputStatus::unavailable); + return LiveInputProbe { + control_mode, + input_status, + has_signal: true, + hdmi_mode: None, + hdmi_fps: None, + subdev_path: None, + bridge_kind: None, + }; + } + + let (subdev_path, bridge_kind) = match csi_bridge::discover_subdev_for_video(&self.path) { + Some((path, kind)) => (Some(path), Some(format!("{:?}", kind).to_lowercase())), + None if is_rk_hdmirx_driver(&caps.driver, &caps.card) => { + (None, Some("rkhdmirx".to_string())) + } + None => (None, None), + }; + + let probe = if let Some(path) = subdev_path.as_ref() { + match csi_bridge::open_subdev(path) { + Ok(fd) => { + let kind = parse_bridge_kind(bridge_kind.as_deref()) + .unwrap_or(csi_bridge::CsiBridgeKind::Unknown); + csi_bridge::probe_signal_thread_timeout( + &fd, + kind, + csi_bridge::RK628_SUBDEV_PROBE_TIMEOUT, + ) + } + Err(error) => { + warn!("Failed to open subdev {:?}: {}", path, error); + None + } + } + } else { + let kind = if is_rk_hdmirx_driver(&caps.driver, &caps.card) { + csi_bridge::CsiBridgeKind::RkHdmirx + } else { + csi_bridge::CsiBridgeKind::Unknown + }; + Some(csi_bridge::probe_signal(&self.fd, kind)) + }; + + let (input_status, hdmi_mode, hdmi_fps, has_signal) = match probe { + Some(csi_bridge::ProbeResult::Locked(mode)) if mode.width > 64 && mode.height > 64 => { + let fps = mode.fps.or_else(|| self.current_parm_fps()); + let hdmi_mode = Some((mode.width, mode.height, fps)); + let status = VideoInputStatus::locked_with_optional_fps( + self.current_active_format(), + mode.width, + mode.height, + fps, + ); + (status, hdmi_mode, fps, true) + } + Some(csi_bridge::ProbeResult::Unavailable) => { + (VideoInputStatus::unavailable(), None, None, false) + } + Some(_) => (VideoInputStatus::no_signal(), None, None, false), + None if subdev_path.is_some() => (VideoInputStatus::unavailable(), None, None, false), + None => (VideoInputStatus::unavailable(), None, None, false), + }; + + LiveInputProbe { + control_mode, + input_status, + has_signal, + hdmi_mode, + hdmi_fps, + subdev_path, + bridge_kind, + } + } + /// Get detailed device information pub fn info(&self) -> Result { let caps: V4l2rCapability = ioctl::querycap(&self.fd) @@ -186,86 +300,10 @@ impl VideoDevice { read_write: flags.contains(Capabilities::READWRITE), }; - // For CSI/HDMI bridges, try to locate the paired subdev *before* - // the signal check: RK628 + rkcif places QUERY_DV_TIMINGS on the - // subdev (the video node returns ENOTTY). Tc358743 and rk_hdmirx - // typically expose DV ioctls on the video node itself, but having - // the subdev handle for EDID/event subscription doesn't hurt. - let (subdev_path, bridge_kind) = - if is_rkcif_driver(&caps.driver) || is_rk_hdmirx_driver(&caps.driver, &caps.card) { - match csi_bridge::discover_subdev_for_video(&self.path) { - Some((path, kind)) => (Some(path), Some(format!("{:?}", kind).to_lowercase())), - None => (None, None), - } - } else { - (None, None) - }; - - // Probe the HDMI source for both signal presence *and* the live - // frame-rate. rkcif's `VIDIOC_ENUM_FRAMEINTERVALS` returns a - // meaningless `1.0..30.0` StepWise range, so the only trustworthy - // fps for rkcif + RK628 / rk_hdmirx boards comes from the bridge - // subdev's DV timings (pixelclock / total_width / total_height). - // - // Preference order: - // 1. Bridge subdev — on rkcif boards this is the *only* node - // where QUERY_DV_TIMINGS works, and it lets the RK628 - // fingerprint filter kick in before we return has_signal=true. - // 2. Video node fallback — for rk_hdmirx / tc358743 where DV - // timings are exposed on the capture node directly. - // 3. USB UVC — always true (no signal concept), no hdmi_fps. - // Subdev-reported HDMI source mode (width, height, fps). On rkcif + - // RK628 boards this is the *only* place DV timings work; the video - // node itself returns ENOTTY for QUERY/G_DV_TIMINGS, so without - // threading this through to `enumerate_bridge_formats` the format - // list ends up with zero resolutions and `select_resolution` falls - // back to the user's preferred value (e.g. 4K) even when the real - // source is 1080p. - let mut subdev_hdmi_mode: Option<(u32, u32, Option)> = None; - - let (has_signal, hdmi_fps) = if let Some(subdev_path) = subdev_path.as_ref() { - match csi_bridge::open_subdev(subdev_path) { - Ok(subdev_fd) => { - let kind = parse_bridge_kind(bridge_kind.as_deref()) - .unwrap_or(csi_bridge::CsiBridgeKind::Unknown); - let probe = csi_bridge::probe_signal(&subdev_fd, kind); - debug!( - "has_signal via subdev {:?} ({:?}): {:?}", - subdev_path, kind, probe - ); - let fps = match &probe { - csi_bridge::ProbeResult::Locked(mode) => { - subdev_hdmi_mode = Some((mode.width, mode.height, mode.fps)); - mode.fps - } - _ => None, - }; - (probe.is_locked(), fps) - } - Err(e) => { - warn!("Failed to open subdev {:?}: {}", subdev_path, e); - (false, None) - } - } - } else if is_rk_hdmirx_driver(&caps.driver, &caps.card) || is_rkcif_driver(&caps.driver) { - let dv = self.current_dv_timings_mode(); - debug!( - "has_signal via video node {:?} (driver={}): dv_timings={:?}", - self.path, caps.driver, dv - ); - let has_signal = dv - .as_ref() - .map(|(w, h, _)| *w > 64 && *h > 64) - .unwrap_or(false); - let fps = if has_signal { - dv.and_then(|(_, _, f)| f) - } else { - None - }; - (has_signal, fps) - } else { - (true, None) - }; + let live = self.probe_live_input(&caps); + let subdev_hdmi_mode = live.hdmi_mode; + let hdmi_fps = live.hdmi_fps; + let has_signal = live.has_signal; let native_hdmirx = is_rk_hdmirx_driver(&caps.driver, &caps.card); let mut formats = if native_hdmirx || is_rkcif_driver(&caps.driver) { @@ -299,7 +337,7 @@ impl VideoDevice { debug!( "Device {:?}: {} formats, priority={}, has_signal={}, hdmi_fps={:?}, is_capture_card={}, subdev={:?}", - self.path, formats.len(), priority, has_signal, hdmi_fps, is_capture_card, subdev_path + self.path, formats.len(), priority, has_signal, hdmi_fps, is_capture_card, live.subdev_path ); Ok(VideoDeviceInfo { @@ -313,8 +351,10 @@ impl VideoDevice { is_capture_card, priority, has_signal, - subdev_path, - bridge_kind, + control_mode: live.control_mode, + input_status: live.input_status, + subdev_path: live.subdev_path, + bridge_kind: live.bridge_kind, }) } @@ -810,9 +850,7 @@ impl VideoDevice { } fn current_dv_timings_mode(&self) -> Option<(u32, u32, Option)> { - let timings = ioctl::query_dv_timings::(&self.fd) - .or_else(|_| ioctl::g_dv_timings::(&self.fd)) - .ok()?; + let timings = ioctl::query_dv_timings::(&self.fd).ok()?; if timings.type_ != V4L2_DV_BT_656_1120 { return None; @@ -1372,6 +1410,8 @@ mod tests { is_capture_card, priority, has_signal: true, + control_mode: control_mode(driver, card), + input_status: VideoInputStatus::unavailable(), subdev_path: None, bridge_kind: None, } diff --git a/src/video/device/mod.rs b/src/video/device/mod.rs index 83d5b3d1..10b10fa3 100644 --- a/src/video/device/mod.rs +++ b/src/video/device/mod.rs @@ -13,6 +13,82 @@ pub use linux::{ #[cfg(windows)] pub use windows::*; +use serde::{Deserialize, Serialize}; + +use crate::video::format::{PixelFormat, Resolution}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum VideoControlMode { + Configurable, + SourceFollowing, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum VideoInputState { + Locked, + NoSignal, + Unavailable, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct VideoInputStatus { + pub state: VideoInputState, + pub format: Option, + pub width: Option, + pub height: Option, + pub fps: Option, +} + +impl VideoInputStatus { + pub fn locked(format: PixelFormat, width: u32, height: u32, fps: f64) -> Self { + Self::locked_with_optional_fps(Some(format), width, height, Some(fps)) + } + + pub fn locked_with_optional_fps( + format: Option, + width: u32, + height: u32, + fps: Option, + ) -> Self { + Self { + state: VideoInputState::Locked, + format: format.map(|format| format.to_string()), + width: Some(width), + height: Some(height), + fps, + } + } + + pub const fn no_signal() -> Self { + Self { + state: VideoInputState::NoSignal, + format: None, + width: None, + height: None, + fps: None, + } + } + + pub const fn unavailable() -> Self { + Self { + state: VideoInputState::Unavailable, + format: None, + width: None, + height: None, + fps: None, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct ResolvedVideoInputConfig { + pub format: PixelFormat, + pub resolution: Resolution, + pub fps: u32, +} + #[cfg(unix)] pub mod bridge; #[cfg(windows)] @@ -25,23 +101,79 @@ pub(crate) fn is_rk_hdmirx_driver(driver: &str, card: &str) -> bool { }) } -pub(crate) fn is_rk_hdmirx_device(device: &VideoDeviceInfo) -> bool { - is_rk_hdmirx_driver(&device.driver, &device.card) +pub(crate) fn is_rkcif_driver(driver: &str) -> bool { + driver.to_ascii_lowercase().starts_with("rkcif") } -pub(crate) fn is_rkcif_driver(driver: &str) -> bool { - driver.eq_ignore_ascii_case("rkcif") +pub fn control_mode(driver: &str, card: &str) -> VideoControlMode { + if is_rkcif_driver(driver) || is_rk_hdmirx_driver(driver, card) { + VideoControlMode::SourceFollowing + } else { + VideoControlMode::Configurable + } } /// Unified check for CSI/HDMI bridge devices (rk_hdmirx, rkcif, etc.) /// that require special enumeration and format-selection logic. pub(crate) fn is_csi_hdmi_bridge(device: &VideoDeviceInfo) -> bool { - is_rk_hdmirx_device(device) || is_rkcif_driver(&device.driver) + device.control_mode == VideoControlMode::SourceFollowing +} + +pub fn resolve_video_input_config( + device: &VideoDeviceInfo, + requested_format: PixelFormat, + requested_resolution: Resolution, + requested_fps: u32, +) -> ResolvedVideoInputConfig { + if device.control_mode == VideoControlMode::SourceFollowing { + if let VideoInputStatus { + state: VideoInputState::Locked, + format: Some(format), + width: Some(width), + height: Some(height), + fps: Some(fps), + } = &device.input_status + { + if let Ok(format) = format.parse::() { + return ResolvedVideoInputConfig { + format, + resolution: Resolution::new(*width, *height), + fps: fps.round().clamp(1.0, 120.0) as u32, + }; + } + } + } + + ResolvedVideoInputConfig { + format: requested_format, + resolution: requested_resolution, + fps: requested_fps, + } } #[cfg(test)] mod tests { - use super::is_rk_hdmirx_driver; + use super::*; + + #[cfg(unix)] + fn device(control_mode: VideoControlMode, input_status: VideoInputStatus) -> VideoDeviceInfo { + VideoDeviceInfo { + path: "/dev/video0".into(), + name: "test".into(), + driver: "test".into(), + bus_info: "test".into(), + card: "test".into(), + formats: Vec::new(), + capabilities: Default::default(), + is_capture_card: true, + priority: 0, + has_signal: input_status.state == VideoInputState::Locked, + control_mode, + input_status, + subdev_path: None, + bridge_kind: None, + } + } #[test] fn recognizes_vendor_and_upstream_native_hdmirx_names() { @@ -50,6 +182,82 @@ mod tests { assert!(is_rk_hdmirx_driver("other", "SNPS_HDMIRX")); assert!(!is_rk_hdmirx_driver("rkcif", "stream_cif_mipi_id0")); } + + #[test] + fn classifies_source_following_drivers_in_one_place() { + assert_eq!( + control_mode("rkcif", "stream_cif_mipi_id0"), + VideoControlMode::SourceFollowing + ); + assert_eq!( + control_mode("rkcif-mipi", "capture"), + VideoControlMode::SourceFollowing + ); + assert_eq!( + control_mode("rk_hdmirx", "capture"), + VideoControlMode::SourceFollowing + ); + assert_eq!( + control_mode("uvcvideo", "USB Capture"), + VideoControlMode::Configurable + ); + } + + #[cfg(unix)] + #[test] + fn source_following_uses_locked_hardware_mode_and_exact_fps_rounding() { + let device = device( + VideoControlMode::SourceFollowing, + VideoInputStatus::locked(PixelFormat::Nv12, 1920, 1080, 59.94), + ); + let resolved = resolve_video_input_config( + &device, + PixelFormat::Mjpeg, + Resolution::new(3840, 2160), + 15, + ); + assert_eq!(resolved.format, PixelFormat::Nv12); + assert_eq!(resolved.resolution, Resolution::new(1920, 1080)); + assert_eq!(resolved.fps, 60); + } + + #[cfg(unix)] + #[test] + fn no_signal_keeps_fallback_and_configurable_keeps_request() { + for (mode, status) in [ + ( + VideoControlMode::SourceFollowing, + VideoInputStatus::no_signal(), + ), + ( + VideoControlMode::Configurable, + VideoInputStatus::unavailable(), + ), + ] { + let resolved = resolve_video_input_config( + &device(mode, status), + PixelFormat::Yuyv, + Resolution::new(1280, 720), + 30, + ); + assert_eq!(resolved.format, PixelFormat::Yuyv); + assert_eq!(resolved.resolution, Resolution::new(1280, 720)); + assert_eq!(resolved.fps, 30); + } + } + + #[test] + fn no_signal_and_unavailable_never_expose_stale_mode_fields() { + for status in [ + VideoInputStatus::no_signal(), + VideoInputStatus::unavailable(), + ] { + assert!(status.format.is_none()); + assert!(status.width.is_none()); + assert!(status.height.is_none()); + assert!(status.fps.is_none()); + } + } } #[cfg(unix)] diff --git a/src/video/device/windows.rs b/src/video/device/windows.rs index 60d67210..b01d19e9 100644 --- a/src/video/device/windows.rs +++ b/src/video/device/windows.rs @@ -1,6 +1,7 @@ use serde::{Deserialize, Serialize}; use std::path::{Path, PathBuf}; +use super::{VideoControlMode, VideoInputStatus}; use crate::error::{AppError, Result}; use crate::video::format::{PixelFormat, Resolution}; @@ -16,6 +17,8 @@ pub struct VideoDeviceInfo { pub is_capture_card: bool, pub priority: u32, pub has_signal: bool, + pub control_mode: VideoControlMode, + pub input_status: VideoInputStatus, pub subdev_path: Option, pub bridge_kind: Option, } @@ -113,6 +116,10 @@ impl VideoDevice { )) }) } + + pub fn input_status(&self) -> Result { + Ok(self.info()?.input_status) + } } pub(crate) fn normalize_windows_device_path(path: impl AsRef) -> PathBuf { @@ -198,6 +205,8 @@ fn directshow_device_from_name(index: usize, name: String) -> VideoDeviceInfo { is_capture_card: true, priority, has_signal: true, + control_mode: VideoControlMode::Configurable, + input_status: VideoInputStatus::unavailable(), subdev_path: None, bridge_kind: None, } diff --git a/src/video/mod.rs b/src/video/mod.rs index acdd38f4..67d435b9 100644 --- a/src/video/mod.rs +++ b/src/video/mod.rs @@ -10,6 +10,7 @@ pub mod format; pub mod frame; #[cfg(feature = "desktop")] pub mod pipeline; +pub mod recovery; pub mod signal; #[cfg(feature = "desktop")] pub mod stream_manager; diff --git a/src/video/pipeline/mod.rs b/src/video/pipeline/mod.rs index 9a43bbd4..2349b672 100644 --- a/src/video/pipeline/mod.rs +++ b/src/video/pipeline/mod.rs @@ -4,6 +4,6 @@ mod encoder_state; mod shared; pub use shared::{ - EncodedVideoFrame, PipelineStateNotification, SharedVideoPipeline, SharedVideoPipelineConfig, - SharedVideoPipelineStats, + EncodedVideoFrame, PipelineAppliedConfig, PipelineStateNotification, SharedVideoPipeline, + SharedVideoPipelineConfig, SharedVideoPipelineStats, }; diff --git a/src/video/pipeline/shared.rs b/src/video/pipeline/shared.rs index 77300af8..26c9bd1d 100644 --- a/src/video/pipeline/shared.rs +++ b/src/video/pipeline/shared.rs @@ -32,10 +32,7 @@ use super::encoder_state::{build_encoder_state, EncoderThreadState}; const AUTO_STOP_GRACE_PERIOD_SECS: u64 = 3; /// After this many consecutive timeouts, log a prominent warning. const CAPTURE_TIMEOUT_RESTART_THRESHOLD: u32 = 5; -const CAPTURE_TIMEOUT_STOP_THRESHOLD: u32 = 60; const CAPTURE_TIMEOUT_SOFT_RESTART_THRESHOLD: u32 = 3; -const CSI_BRIDGE_NOSIGNAL_INTERVAL_MS: u64 = 500; -const NOSIGNAL_POLL_MAX: Duration = Duration::from_secs(20); /// Throttle repeated encoding errors to avoid log flooding const ENCODE_ERROR_THROTTLE_SECS: u64 = 5; @@ -53,10 +50,11 @@ use crate::video::capture::status::{ use crate::video::capture::{is_source_changed_error, BridgeContext, CaptureStream}; use crate::video::codec::h264_bitstream; use crate::video::codec::registry::{EncoderBackend, VideoEncoderType}; -use crate::video::device::bridge::{self as csi_bridge, ProbeResult}; use crate::video::device::parse_bridge_kind; +use crate::video::device::VideoControlMode; use crate::video::format::{PixelFormat, Resolution}; use crate::video::frame::{FrameBuffer, FrameBufferPool, VideoFrame}; +use crate::video::recovery::{wait_for_source_change, CaptureRecoveryPolicy}; use crate::video::signal::SignalStatus; const MIN_CAPTURE_FRAME_SIZE: usize = 128; @@ -90,14 +88,27 @@ pub struct PipelineStateNotification { pub state: &'static str, pub reason: Option<&'static str>, pub next_retry_ms: Option, + pub applied_config: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PipelineAppliedConfig { + pub resolution: Resolution, + pub format: PixelFormat, + pub fps: u32, } impl PipelineStateNotification { - fn streaming() -> Self { + fn streaming(resolution: Resolution, format: PixelFormat, fps: u32) -> Self { Self { state: "streaming", reason: None, next_retry_ms: None, + applied_config: Some(PipelineAppliedConfig { + resolution, + format, + fps, + }), } } @@ -106,14 +117,7 @@ impl PipelineStateNotification { state: "no_signal", reason: Some(status.as_str()), next_retry_ms, - } - } - - fn device_busy(reason: &'static str) -> Self { - Self { - state: "device_busy", - reason: Some(reason), - next_retry_ms: None, + applied_config: None, } } } @@ -121,6 +125,8 @@ impl PipelineStateNotification { /// Shared video pipeline configuration #[derive(Debug, Clone)] pub struct SharedVideoPipelineConfig { + /// Whether the capture mode is configured by the client or follows HDMI. + pub control_mode: VideoControlMode, /// Input resolution pub resolution: Resolution, /// Input pixel format @@ -138,6 +144,7 @@ pub struct SharedVideoPipelineConfig { impl Default for SharedVideoPipelineConfig { fn default() -> Self { Self { + control_mode: VideoControlMode::Configurable, resolution: Resolution::HD720, input_format: PixelFormat::Yuyv, output_codec: VideoEncoderType::H264, @@ -285,81 +292,6 @@ pub struct SharedVideoPipeline { last_state_notification: ParkingMutex>, } -fn poll_bridge_subdev_after_no_signal(bridge_ctx: &BridgeContext, pipeline: &SharedVideoPipeline) { - let Some(subdev_path) = bridge_ctx.subdev_path.as_ref() else { - return; - }; - let kind = bridge_ctx - .kind - .unwrap_or(csi_bridge::CsiBridgeKind::Unknown); - let deadline = Instant::now() + NOSIGNAL_POLL_MAX; - let mut poll_count: u32 = 0; - info!( - "No-signal poll: scanning subdev {:?} every {} ms (max {:?})", - subdev_path, CSI_BRIDGE_NOSIGNAL_INTERVAL_MS, NOSIGNAL_POLL_MAX - ); - loop { - if !pipeline.running_flag.load(Ordering::Acquire) { - return; - } - if Instant::now() >= deadline { - info!( - "No-signal poll: stopped after {:?} ({} attempts)", - NOSIGNAL_POLL_MAX, poll_count - ); - return; - } - let fd = match csi_bridge::open_subdev(subdev_path) { - Ok(f) => f, - Err(e) => { - debug!( - "No-signal poll: open subdev {:?} failed: {}", - subdev_path, e - ); - std::thread::sleep(Duration::from_millis(CSI_BRIDGE_NOSIGNAL_INTERVAL_MS)); - continue; - } - }; - match csi_bridge::probe_signal_thread_timeout( - &fd, - kind, - csi_bridge::RK628_SUBDEV_PROBE_TIMEOUT, - ) { - Some(ProbeResult::Locked(mode)) => { - info!( - "No-signal poll: locked {}x{} @ {} Hz — proceeding to capture re-open", - mode.width, mode.height, mode.pixelclock - ); - return; - } - Some(other) => { - poll_count = poll_count.saturating_add(1); - if poll_count == 1 || poll_count.is_multiple_of(8) { - debug!( - "No-signal poll: attempt {} — still {:?}", - poll_count, - other.as_status() - ); - } - if let Some(st) = other.as_status() { - pipeline.notify_state(PipelineStateNotification::no_signal( - st, - Some(CSI_BRIDGE_NOSIGNAL_INTERVAL_MS.saturating_add(50)), - )); - } - } - None => { - poll_count = poll_count.saturating_add(1); - debug!( - "No-signal poll: attempt {} — probe ioctl timed out", - poll_count - ); - } - } - std::thread::sleep(Duration::from_millis(CSI_BRIDGE_NOSIGNAL_INTERVAL_MS)); - } -} - impl SharedVideoPipeline { /// Create a new shared video pipeline pub fn new(config: SharedVideoPipelineConfig) -> Result> { @@ -575,7 +507,6 @@ impl SharedVideoPipeline { _jpeg_quality: u8, subdev_path: Option, bridge_kind: Option, - _v4l2_driver: Option, ) -> Result<()> { if *self.running_rx.borrow() { warn!("Pipeline already running"); @@ -601,24 +532,33 @@ impl SharedVideoPipeline { buffer_count.max(1), Duration::from_secs(2), bridge_ctx_probe, + config.control_mode, ) { Ok(s) => { let negotiated_res = s.resolution(); let negotiated_fmt = s.format(); - if negotiated_res != config.resolution || negotiated_fmt != config.input_format { + let previous = (config.resolution, config.input_format, config.fps); + if config.control_mode == VideoControlMode::SourceFollowing { + if let Some(source_fps) = s.source_fps() { + config.fps = source_fps.round().clamp(1.0, 120.0) as u32; + } + } + config.resolution = negotiated_res; + config.input_format = negotiated_fmt; + if previous != (config.resolution, config.input_format, config.fps) { info!( - "Negotiated capture {}x{} {:?} (configured {}x{} {:?}) — aligning encoder to source", + "Negotiated capture {}x{} {:?} @ {} fps (configured {}x{} {:?} @ {} fps) — aligning encoder to source", negotiated_res.width, negotiated_res.height, negotiated_fmt, - config.resolution.width, - config.resolution.height, - config.input_format + config.fps, + previous.0.width, + previous.0.height, + previous.1, + previous.2, ); - config.resolution = negotiated_res; - config.input_format = negotiated_fmt; - *self.config.write().await = config.clone(); } + *self.config.write().await = config.clone(); Some(s) } Err(AppError::CaptureNoSignal { kind }) => { @@ -628,7 +568,11 @@ impl SharedVideoPipeline { let status = signal_status_from_capture_kind(&kind); self.notify_state(PipelineStateNotification::no_signal( status, - Some(Duration::from_secs(2).as_millis() as u64), + Some( + CaptureRecoveryPolicy::new(config.control_mode) + .retry_delay(1) + .as_millis() as u64, + ), )); None } @@ -754,74 +698,19 @@ impl SharedVideoPipeline { let mut initial_geometry: Option<(Resolution, PixelFormat)> = None; let mut resolution = config.resolution; let mut pixel_format = config.input_format; + let mut active_fps = config.fps; let mut stride: u32 = 0; - match preopened { - Some(s) => { - resolution = s.resolution(); - pixel_format = s.format(); - stride = s.stride(); - initial_geometry = Some((resolution, pixel_format)); - stream = Some(s); - } - None => { - match open_capture_stream( - &device_path, - config.resolution, - config.input_format, - config.fps, - buffer_count.max(1), - Duration::from_secs(2), - bridge_ctx.clone(), - ) { - Ok(s) => { - resolution = s.resolution(); - pixel_format = s.format(); - stride = s.stride(); - if resolution != config.resolution - || pixel_format != config.input_format - { - info!( - "First capture open negotiated {}x{} {:?} but encoder expects {}x{} {:?} — stopping for dimension resync", - resolution.width, - resolution.height, - pixel_format, - config.resolution.width, - config.resolution.height, - config.input_format - ); - pipeline.notify_state(PipelineStateNotification::device_busy( - "config_changing", - )); - *pipeline.pending_sync_geometry.lock() = - Some((resolution, pixel_format)); - let _ = pipeline.running.send(false); - pipeline.running_flag.store(false, Ordering::Release); - let _ = frame_seq_tx.send(1); - return; - } - initial_geometry = Some((resolution, pixel_format)); - stream = Some(s); - } - Err(AppError::CaptureNoSignal { kind }) => { - warn!( - "Capture stream open reports no signal ({}) — pipeline will retry", - kind - ); - pipeline.notify_state(PipelineStateNotification::no_signal( - signal_status_from_capture_kind(&kind), - Some(CSI_BRIDGE_NOSIGNAL_INTERVAL_MS), - )); - } - Err(e) => { - error!("Failed to open capture stream: {}", e); - let _ = pipeline.running.send(false); - pipeline.running_flag.store(false, Ordering::Release); - let _ = frame_seq_tx.send(1); - return; - } - } - } + if let Some(s) = preopened { + resolution = s.resolution(); + pixel_format = s.format(); + active_fps = s + .source_fps() + .map(|fps| fps.round().clamp(1.0, 120.0) as u32) + .unwrap_or(config.fps); + stride = s.stride(); + initial_geometry = Some((resolution, pixel_format)); + stream = Some(s); } fn open_or_retry( @@ -838,6 +727,7 @@ impl SharedVideoPipeline { buffer_count.max(1), Duration::from_secs(2), bridge_ctx, + config.control_mode, is_device_lost_message, ) { CaptureOpenResult::NoSignal(status) => { @@ -860,6 +750,7 @@ impl SharedVideoPipeline { let grace_period = Duration::from_secs(AUTO_STOP_GRACE_PERIOD_SECS); let mut sequence: u64 = 0; let mut consecutive_timeouts: u32 = 0; + let recovery_policy = CaptureRecoveryPolicy::new(config.control_mode); let capture_error_throttler = LogThrottler::with_secs(5); let mut suppressed_capture_errors: HashMap = HashMap::new(); @@ -877,9 +768,7 @@ impl SharedVideoPipeline { "No subscribers for {}s, auto-stopping video pipeline", grace_period.as_secs() ); - let _ = pipeline.running.send(false); pipeline.running_flag.store(false, Ordering::Release); - let _ = frame_seq_tx.send(sequence.wrapping_add(1)); break; } } @@ -899,6 +788,10 @@ impl SharedVideoPipeline { let new_res = new_stream.resolution(); let new_fmt = new_stream.format(); let new_stride = new_stream.stride(); + let new_fps = new_stream + .source_fps() + .map(|fps| fps.round().clamp(1.0, 120.0) as u32) + .unwrap_or(config.fps); // Pre-probe was skipped (no signal at pipeline start) but the // encoder was sized to saved settings — if DV timings now @@ -916,14 +809,13 @@ impl SharedVideoPipeline { config.resolution.height, config.input_format ); - pipeline.notify_state(PipelineStateNotification::device_busy( - "config_changing", + pipeline.notify_state(PipelineStateNotification::no_signal( + SignalStatus::NoSignal, + Some(recovery_policy.retry_delay(1).as_millis() as u64), )); *pipeline.pending_sync_geometry.lock() = Some((new_res, new_fmt)); - let _ = pipeline.running.send(false); pipeline.running_flag.store(false, Ordering::Release); - let _ = frame_seq_tx.send(sequence.wrapping_add(1)); break; } @@ -944,15 +836,15 @@ impl SharedVideoPipeline { orig_res, orig_fmt, new_res, new_fmt ); pipeline.notify_state( - PipelineStateNotification::device_busy( - "config_changing", + PipelineStateNotification::no_signal( + SignalStatus::NoSignal, + Some(recovery_policy.retry_delay(1).as_millis() + as u64), ), ); *pipeline.pending_sync_geometry.lock() = Some((new_res, new_fmt)); - let _ = pipeline.running.send(false); pipeline.running_flag.store(false, Ordering::Release); - let _ = frame_seq_tx.send(sequence.wrapping_add(1)); break; } _ => {} @@ -963,6 +855,7 @@ impl SharedVideoPipeline { } resolution = new_res; pixel_format = new_fmt; + active_fps = new_fps; stride = new_stride; stream = Some(new_stream); consecutive_timeouts = 0; @@ -973,36 +866,34 @@ impl SharedVideoPipeline { } CaptureOpenResult::NoSignal(status) => { consecutive_timeouts = consecutive_timeouts.saturating_add(1); - if consecutive_timeouts >= CAPTURE_TIMEOUT_STOP_THRESHOLD { + if !recovery_policy.should_retry(consecutive_timeouts) { warn!( "Capture soft-restart gave up after {} attempts, \ stopping pipeline", consecutive_timeouts ); - let _ = pipeline.running.send(false); pipeline.running_flag.store(false, Ordering::Release); - let _ = frame_seq_tx.send(sequence.wrapping_add(1)); break; } - let wait_ms = CSI_BRIDGE_NOSIGNAL_INTERVAL_MS; + let delay = recovery_policy.retry_delay(consecutive_timeouts); pipeline.notify_state(PipelineStateNotification::no_signal( status, - Some(wait_ms), + Some(delay.as_millis() as u64), )); - std::thread::sleep(Duration::from_millis(wait_ms)); + if wait_for_source_change(&bridge_ctx, delay, || { + pipeline.running_flag.load(Ordering::Acquire) + }) { + info!("SOURCE_CHANGE woke capture retry"); + } continue; } CaptureOpenResult::DeviceLost(reason) => { pipeline.mark_device_lost(reason); - let _ = pipeline.running.send(false); pipeline.running_flag.store(false, Ordering::Release); - let _ = frame_seq_tx.send(sequence.wrapping_add(1)); break; } CaptureOpenResult::Fatal => { - let _ = pipeline.running.send(false); pipeline.running_flag.store(false, Ordering::Release); - let _ = frame_seq_tx.send(sequence.wrapping_add(1)); break; } } @@ -1024,100 +915,37 @@ impl SharedVideoPipeline { // drop the stream so the next iteration re-opens // via a fresh DV_TIMINGS probe. This is the fast // path for source-side resolution switches on - // RK628 / rkcif — sub-second recovery vs. the ~8 s - // timeout fallback. + // RK628 / rkcif; the retry policy is only a fallback + // when a driver does not provide usable events. if is_source_changed_error(&e) { info!( "Capture reported SOURCE_CHANGE — \ dropping stream for immediate re-open" ); + if recovery_policy.control_mode() + == VideoControlMode::SourceFollowing + { + pipeline.notify_state(PipelineStateNotification::no_signal( + SignalStatus::NoSignal, + Some(recovery_policy.retry_delay(1).as_millis() as u64), + )); + } consecutive_timeouts = 0; stream = None; continue; } if e.kind() == std::io::ErrorKind::TimedOut { consecutive_timeouts = consecutive_timeouts.saturating_add(1); - let probe_result = { - let sr = stream.as_mut().expect("stream is Some above"); - sr.probe_bridge_signal_with_timeout( - csi_bridge::RK628_SUBDEV_PROBE_TIMEOUT, - ) - }; - match probe_result { - Some(ProbeResult::Locked(mode)) => { - let probed_resolution = - Resolution::new(mode.width, mode.height); - if probed_resolution == resolution { - info!( - "Capture timeout but bridge is locked at {}x{} — soft-restarting capture without encoder rebuild", - probed_resolution.width, - probed_resolution.height - ); - } else { - info!( - "Capture timeout probe detected geometry change {}x{} -> {}x{} — soft-restarting capture for encoder rebuild", - resolution.width, - resolution.height, - probed_resolution.width, - probed_resolution.height - ); - pipeline.notify_state( - PipelineStateNotification::device_busy( - "config_changing", - ), - ); - } - consecutive_timeouts = 0; - stream = None; - continue; - } - Some(other) => { - let status = - other.as_status().unwrap_or(SignalStatus::NoSignal); - warn!( - "Capture timeout probe reports no signal ({})", - status.as_str() - ); - pipeline.notify_state( - PipelineStateNotification::no_signal( - status, - Some(Duration::from_secs(2).as_millis() as u64), - ), - ); - // Drop capture so RK628 / rkcif can release the queue, - // then poll subdev on a fresh fd until timings lock (or - // timeout). Avoids sitting on DQBUF 2s × N with a dead - // stream while `v4l2-ctl --query-dv-timings` already shows - // a real mode. - stream = None; - consecutive_timeouts = 0; - if bridge_ctx.has_subdev() - && matches!( - other, - ProbeResult::NoSignal - | ProbeResult::NoSync - | ProbeResult::OutOfRange - ) - { - poll_bridge_subdev_after_no_signal( - &bridge_ctx, - &pipeline, - ); - } - continue; - } - None if bridge_ctx.has_subdev() => { - warn!( - "DV-timings probe timed out or failed — forcing stream re-open (RK628 / rkcif)" - ); - consecutive_timeouts = 0; - stream = None; - poll_bridge_subdev_after_no_signal(&bridge_ctx, &pipeline); - continue; - } - None => { - warn!("Capture timeout - no signal?"); - } + if recovery_policy.control_mode() + == VideoControlMode::SourceFollowing + { + let delay = recovery_policy.retry_delay(consecutive_timeouts); + pipeline.notify_state(PipelineStateNotification::no_signal( + SignalStatus::NoSignal, + Some(delay.as_millis() as u64), + )); + stream = None; + continue; } if consecutive_timeouts >= CAPTURE_TIMEOUT_SOFT_RESTART_THRESHOLD { @@ -1145,17 +973,6 @@ impl SharedVideoPipeline { consecutive_timeouts ); } - - if consecutive_timeouts >= CAPTURE_TIMEOUT_STOP_THRESHOLD { - warn!( - "Capture timed out {} consecutive times, stopping video pipeline", - consecutive_timeouts - ); - let _ = pipeline.running.send(false); - pipeline.running_flag.store(false, Ordering::Release); - let _ = frame_seq_tx.send(sequence.wrapping_add(1)); - break; - } } else { consecutive_timeouts = 0; // EIO (5) / EPIPE (32) / EPROTO (71) in next_into generally @@ -1187,9 +1004,7 @@ impl SharedVideoPipeline { CaptureIoErrorKind::DeviceLost => { error!("Capture device lost: {}", e); pipeline.mark_device_lost(e.to_string()); - let _ = pipeline.running.send(false); pipeline.running_flag.store(false, Ordering::Release); - let _ = frame_seq_tx.send(sequence.wrapping_add(1)); break; } CaptureIoErrorKind::Other => {} @@ -1223,7 +1038,11 @@ impl SharedVideoPipeline { owned.truncate(frame_size); // Notify streaming only after the short-frame guard passes. - pipeline.notify_state(PipelineStateNotification::streaming()); + pipeline.notify_state(PipelineStateNotification::streaming( + resolution, + pixel_format, + active_fps, + )); let frame = Arc::new(VideoFrame::from_pooled( Arc::new(FrameBuffer::new(owned, Some(buffer_pool.clone()))), resolution, @@ -1240,10 +1059,14 @@ impl SharedVideoPipeline { let _ = frame_seq_tx.send(sequence); } + // `running` represents completed lifecycle state, not a stop request. + // Drop the V4L2 stream first so STREAMOFF, buffer teardown and FD close + // have all completed before another consumer is told the device is free. + drop(stream); pipeline.running_flag.store(false, Ordering::Release); - let _ = pipeline.running.send(false); let _ = frame_seq_tx.send(sequence.wrapping_add(1)); - info!("Video pipeline stopped"); + let _ = pipeline.running.send(false); + info!("Video pipeline stopped and capture device released"); }); } @@ -1458,9 +1281,7 @@ impl SharedVideoPipeline { /// Stop the pipeline (non-blocking, does not wait for capture thread to exit) pub fn stop(&self) { - if *self.running_rx.borrow() { - let _ = self.running.send(false); - self.running_flag.store(false, Ordering::Release); + if self.running_flag.swap(false, Ordering::AcqRel) { self.clear_cmd_tx(); info!("Stopping video pipeline"); } @@ -1471,32 +1292,38 @@ impl SharedVideoPipeline { /// This ensures the V4L2 device is released before returning, which is /// necessary when another consumer (e.g. MJPEG streamer) needs to open /// the same device immediately after. - pub async fn stop_and_wait(&self, timeout: std::time::Duration) { + pub async fn stop_and_wait(&self, timeout: std::time::Duration) -> Result<()> { self.stop(); let mut rx = self.running_watch(); - if !*rx.borrow() { - // Capture thread may still be running from a previous `stop()` call. - // Wait for the "Video pipeline stopped" log (thread sets running=false - // at exit), unless it already happened. - } let deadline = tokio::time::Instant::now() + timeout; - loop { - if !self.running_flag.load(Ordering::Acquire) { - // Flag is cleared, but the capture thread may still be unwinding - // (dropping the V4L2 stream). Give it a brief moment. - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - break; - } + + while *rx.borrow() { let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); if remaining.is_zero() { - warn!( - "Timed out waiting for video pipeline to stop after {:?}", + return Err(AppError::VideoError(format!( + "Timed out waiting {:?} for video pipeline to release capture device", timeout - ); - break; + ))); + } + match tokio::time::timeout(remaining, rx.changed()).await { + Ok(Ok(())) => {} + Ok(Err(_)) if !*rx.borrow() => break, + Ok(Err(_)) => { + return Err(AppError::VideoError( + "Video pipeline lifecycle channel closed before capture device release" + .to_string(), + )); + } + Err(_) => { + return Err(AppError::VideoError(format!( + "Timed out waiting {:?} for video pipeline to release capture device", + timeout + ))); + } } - let _ = tokio::time::timeout(remaining, rx.changed()).await; } + + Ok(()) } /// Set bitrate using preset @@ -1696,7 +1523,7 @@ fn copy_rows( impl Drop for SharedVideoPipeline { fn drop(&mut self) { - let _ = self.running.send(false); + self.running_flag.store(false, Ordering::Release); } } @@ -1765,4 +1592,49 @@ mod tests { let h265 = SharedVideoPipelineConfig::h265(Resolution::HD720, BitratePreset::Speed); assert_eq!(h265.output_codec, VideoEncoderType::H265); } + + #[test] + fn stop_request_does_not_publish_worker_exit() { + let pipeline = SharedVideoPipeline::new(SharedVideoPipelineConfig::h264( + Resolution::HD720, + BitratePreset::Balanced, + )) + .unwrap(); + let _ = pipeline.running.send(true); + pipeline.running_flag.store(true, Ordering::Release); + + pipeline.stop(); + + assert!(!pipeline.running_flag.load(Ordering::Acquire)); + assert!(pipeline.is_running()); + + // Simulate the capture thread's common cleanup tail. + let _ = pipeline.running.send(false); + assert!(!pipeline.is_running()); + } + + #[tokio::test] + async fn stop_and_wait_observes_completed_worker_cleanup() { + let pipeline = SharedVideoPipeline::new(SharedVideoPipelineConfig::h264( + Resolution::HD720, + BitratePreset::Balanced, + )) + .unwrap(); + let _ = pipeline.running.send(true); + pipeline.running_flag.store(true, Ordering::Release); + + let worker = pipeline.clone(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(30)).await; + let _ = worker.running.send(false); + }); + + let started = Instant::now(); + pipeline + .stop_and_wait(Duration::from_secs(1)) + .await + .unwrap(); + assert!(started.elapsed() >= Duration::from_millis(20)); + assert!(!pipeline.is_running()); + } } diff --git a/src/video/recovery.rs b/src/video/recovery.rs new file mode 100644 index 00000000..f72f4889 --- /dev/null +++ b/src/video/recovery.rs @@ -0,0 +1,134 @@ +//! Shared capture recovery policy. +//! +//! Device discovery decides whether an input follows an external source. The +//! capture layers consume that decision; they must not infer it again from a +//! driver name because doing so makes MJPEG and WebRTC recover differently. + +use std::time::Duration; + +use super::capture::BridgeContext; +use super::device::VideoControlMode; + +const SOURCE_FOLLOWING_RETRY_DELAYS: [Duration; 3] = [ + Duration::from_millis(500), + Duration::from_secs(1), + Duration::from_secs(2), +]; +const CONFIGURABLE_RETRY_DELAY: Duration = Duration::from_millis(500); +const CONFIGURABLE_RETRY_LIMIT: u32 = 60; + +#[derive(Debug, Clone, Copy)] +pub struct CaptureRecoveryPolicy { + control_mode: VideoControlMode, +} + +/// Wait for a source-change edge, falling back to the policy delay when the +/// driver does not expose events. The short slices keep shutdown responsive. +#[cfg(unix)] +pub fn wait_for_source_change( + bridge: &BridgeContext, + delay: Duration, + should_continue: impl Fn() -> bool, +) -> bool { + use std::time::Instant; + + use super::device::bridge; + + let Some(path) = bridge.subdev_path.as_ref() else { + return interruptible_sleep(delay, should_continue); + }; + let Ok(fd) = bridge::open_subdev(path) else { + return interruptible_sleep(delay, should_continue); + }; + if bridge::subscribe_source_change(&fd).is_err() { + return interruptible_sleep(delay, should_continue); + } + + let deadline = Instant::now() + delay; + while should_continue() && Instant::now() < deadline { + let remaining = deadline.saturating_duration_since(Instant::now()); + match bridge::wait_source_change(&fd, remaining.min(Duration::from_millis(250))) { + Ok(true) => return true, + Ok(false) => {} + Err(_) => return false, + } + } + false +} + +#[cfg(windows)] +pub fn wait_for_source_change( + _bridge: &BridgeContext, + delay: Duration, + should_continue: impl Fn() -> bool, +) -> bool { + interruptible_sleep(delay, should_continue) +} + +fn interruptible_sleep(delay: Duration, should_continue: impl Fn() -> bool) -> bool { + use std::time::Instant; + + let deadline = Instant::now() + delay; + while should_continue() && Instant::now() < deadline { + std::thread::sleep( + deadline + .saturating_duration_since(Instant::now()) + .min(Duration::from_millis(100)), + ); + } + false +} + +impl CaptureRecoveryPolicy { + pub const fn new(control_mode: VideoControlMode) -> Self { + Self { control_mode } + } + + pub const fn control_mode(self) -> VideoControlMode { + self.control_mode + } + + /// Delay after `failed_attempts` consecutive attempts (one-based). + pub fn retry_delay(self, failed_attempts: u32) -> Duration { + match self.control_mode { + VideoControlMode::SourceFollowing => { + let index = failed_attempts.saturating_sub(1).min(2) as usize; + SOURCE_FOLLOWING_RETRY_DELAYS[index] + } + VideoControlMode::Configurable => CONFIGURABLE_RETRY_DELAY, + } + } + + /// Source-following inputs keep probing for as long as they have a + /// consumer. Configurable/UVC inputs retain the pre-existing finite policy. + pub const fn should_retry(self, failed_attempts: u32) -> bool { + match self.control_mode { + VideoControlMode::SourceFollowing => true, + VideoControlMode::Configurable => failed_attempts < CONFIGURABLE_RETRY_LIMIT, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn source_following_uses_capped_backoff_and_never_expires() { + let policy = CaptureRecoveryPolicy::new(VideoControlMode::SourceFollowing); + assert_eq!(policy.retry_delay(1), Duration::from_millis(500)); + assert_eq!(policy.retry_delay(2), Duration::from_secs(1)); + assert_eq!(policy.retry_delay(3), Duration::from_secs(2)); + assert_eq!(policy.retry_delay(10_000), Duration::from_secs(2)); + assert!(policy.should_retry(61)); + assert!(policy.should_retry(9_000)); // five hours at the capped delay + } + + #[test] + fn configurable_inputs_keep_the_finite_retry_policy() { + let policy = CaptureRecoveryPolicy::new(VideoControlMode::Configurable); + assert_eq!(policy.retry_delay(1), Duration::from_millis(500)); + assert!(policy.should_retry(59)); + assert!(!policy.should_retry(60)); + } +} diff --git a/src/video/stream_manager.rs b/src/video/stream_manager.rs index d60e4b7d..57343c94 100644 --- a/src/video/stream_manager.rs +++ b/src/video/stream_manager.rs @@ -358,29 +358,9 @@ impl VideoStreamManager { .update_video_config(resolution, format, fps) .await; if let Some(device_path) = device_path { - // Resolve the paired subdev so the WebRTC pipeline can run the - // RK628 STREAMON gate + SOURCE_CHANGE polling identically to the - // MJPEG path. See `csi_bridge::discover_subdev_for_video`. - let (subdev_path, bridge_kind, v4l2_driver) = self - .streamer - .current_device() - .await - .map(|d| { - ( - d.subdev_path.clone(), - d.bridge_kind.clone(), - Some(d.driver.clone()), - ) - }) - .unwrap_or((None, None, None)); + let device_info = self.streamer.current_device().await; self.webrtc_streamer - .set_capture_device( - device_path, - jpeg_quality, - subdev_path, - bridge_kind, - v4l2_driver, - ) + .set_capture_device(device_path, jpeg_quality, device_info) .await; } else { warn!("No capture device configured while syncing WebRTC capture source"); @@ -434,7 +414,7 @@ impl VideoStreamManager { let closed = self .webrtc_streamer .close_all_sessions_and_release_device() - .await; + .await?; if closed > 0 { info!("Closed {} WebRTC sessions", closed); } @@ -549,26 +529,9 @@ impl VideoStreamManager { } if let Some(device_path) = device_path { info!("Configuring direct capture for WebRTC after config change"); - let (subdev_path, bridge_kind, v4l2_driver) = self - .streamer - .current_device() - .await - .map(|d| { - ( - d.subdev_path.clone(), - d.bridge_kind.clone(), - Some(d.driver.clone()), - ) - }) - .unwrap_or((None, None, None)); + let device_info = self.streamer.current_device().await; self.webrtc_streamer - .set_capture_device( - device_path, - jpeg_quality, - subdev_path, - bridge_kind, - v4l2_driver, - ) + .set_capture_device(device_path, jpeg_quality, device_info) .await; } else { warn!("No capture device configured for WebRTC after config change"); diff --git a/src/video/streamer.rs b/src/video/streamer.rs index d7cda550..b7d352bc 100644 --- a/src/video/streamer.rs +++ b/src/video/streamer.rs @@ -13,8 +13,8 @@ use tracing::{debug, error, info, trace, warn}; use super::device::{ bridge as csi_bridge, enumerate_devices, find_best_device, is_csi_hdmi_bridge, - parse_bridge_kind, select_recovery_device, VideoDevice, VideoDeviceInfo, - VideoDeviceRecoveryHint, + parse_bridge_kind, resolve_video_input_config, select_recovery_device, VideoControlMode, + VideoDevice, VideoDeviceInfo, VideoDeviceRecoveryHint, }; use super::format::{PixelFormat, Resolution}; use super::frame::{FrameBuffer, FrameBufferPool, VideoFrame}; @@ -30,6 +30,7 @@ use crate::video::capture::status::{ use crate::video::capture::{ is_source_changed_error, BridgeContext, CaptureStream, DEFAULT_CAPTURE_BUFFER_COUNT, }; +use crate::video::recovery::{wait_for_source_change, CaptureRecoveryPolicy}; const MIN_CAPTURE_FRAME_SIZE: usize = 128; @@ -373,7 +374,10 @@ impl Streamer { .ok_or_else(|| AppError::VideoError("Video device not found".to_string()))? }; - let (format, resolution) = self.resolve_capture_config(&device, format, resolution)?; + let resolved = self.resolve_capture_config(&device, format, resolution, fps)?; + let format = resolved.format; + let resolution = resolved.resolution; + let fps = resolved.fps; // IMPORTANT: Disconnect all MJPEG clients FIRST before stopping capture // This prevents race conditions where clients try to reconnect and reopen the device @@ -442,19 +446,18 @@ impl Streamer { device.path.display() ); - // Determine best format for this device let config = self.config.read().await; - let format = self.select_format(&device, config.format)?; - let resolution = self.select_resolution(&device, &format, config.resolution)?; - + let resolved = + self.resolve_capture_config(&device, config.format, config.resolution, config.fps)?; drop(config); // Update config with actual values { let mut config = self.config.write().await; config.device_path = Some(device.path.clone()); - config.format = format; - config.resolution = resolution; + config.format = resolved.format; + config.resolution = resolved.resolution; + config.fps = resolved.fps; } // Store device info @@ -462,7 +465,10 @@ impl Streamer { *self.state.write().await = StreamerState::Ready; - info!("Streamer initialized: {} @ {}", format, resolution); + info!( + "Streamer initialized: {} @ {} {} fps", + resolved.format, resolved.resolution, resolved.fps + ); Ok(()) } @@ -574,10 +580,20 @@ impl Streamer { device: &VideoDeviceInfo, requested_format: PixelFormat, requested_resolution: Resolution, - ) -> Result<(PixelFormat, Resolution)> { - let format = self.select_format(device, requested_format)?; - let resolution = self.select_resolution(device, &format, requested_resolution)?; - Ok((format, resolution)) + requested_fps: u32, + ) -> Result { + let mut resolved = resolve_video_input_config( + device, + requested_format, + requested_resolution, + requested_fps, + ); + if device.control_mode == VideoControlMode::Configurable { + resolved.format = self.select_format(device, resolved.format)?; + resolved.resolution = + self.select_resolution(device, &resolved.format, resolved.resolution)?; + } + Ok(resolved) } /// Restart capture for recovery (direct capture path) @@ -783,26 +799,9 @@ impl Streamer { const RETRY_DELAY_MS: u64 = 200; const IDLE_STOP_DELAY_SECS: u64 = 5; const BUFFER_COUNT: u32 = DEFAULT_CAPTURE_BUFFER_COUNT; - /// Initial back-off after signal loss before the first soft restart. - /// - /// PiKVM/ustreamer drops to sub-second recovery because it subscribes to - /// `V4L2_EVENT_SOURCE_CHANGE`; lacking that (for now), we bound how long - /// the user has to stare at a placeholder after a source-side resolution - /// change by driving a soft-restart at 1 s, then 2 s, 4 s, …, 8 s. - const NOSIGNAL_SOFT_RESTART_INITIAL_SECS: u64 = 1; - const NOSIGNAL_SOFT_RESTART_MAX_SECS: u64 = 8; - let handle = tokio::runtime::Handle::current(); let mut last_state = StreamerState::Streaming; - // Compute the current soft-restart back-off window (in seconds) - // for the exponential ladder 1 s → 2 s → 4 s → 8 s (capped). - let backoff_secs = |count: u32| -> u64 { - NOSIGNAL_SOFT_RESTART_INITIAL_SECS - .saturating_mul(2u64.pow(count.min(3))) - .min(NOSIGNAL_SOFT_RESTART_MAX_SECS) - }; - let mut set_state = |new_state: StreamerState| { if new_state != last_state { handle.block_on(async { @@ -829,20 +828,23 @@ impl Streamer { self.next_retry_ms.store(ms, Ordering::Relaxed); }; - // How many soft-restart cycles have been attempted (for exponential back-off). + // Consecutive recovery attempts, shared with the common retry policy. let mut no_signal_restart_count: u32 = 0; - - // Last (resolution, format, fps) combination for which we emitted a - // `StreamConfigApplied` event. Used to de-duplicate the event across - // soft-restarts that produce the exact same geometry (e.g. a spurious - // single-frame timeout on a stable source) — the frontend would - // otherwise re-layout the `` on every glitch. - let mut last_applied: Option<(u32, u32, PixelFormat, u32)> = None; + let mut no_consumers_since: Option = None; 'session: loop { if self.direct_stop.load(Ordering::Relaxed) { break 'session; } + if self.mjpeg_handler.client_count() == 0 { + let since = no_consumers_since.get_or_insert_with(std::time::Instant::now); + if since.elapsed() >= Duration::from_secs(IDLE_STOP_DELAY_SECS) { + info!("No MJPEG consumers during recovery; stopping capture"); + break 'session; + } + } else { + no_consumers_since = None; + } // Re-read config at the start of each session so that a re_init_device() // call (from a previous soft-restart or recovery) is reflected here. @@ -855,52 +857,44 @@ impl Streamer { // `VideoDeviceInfo` during enumeration; we re-read it here // rather than caching on Streamer so a hot-plug recovery picks // up a possibly-different subdev path. - let bridge_ctx = handle.block_on(async { + let (bridge_ctx, control_mode) = handle.block_on(async { self.current_device .read() .await .as_ref() .map(|info| { - BridgeContext::from_parts( - info.subdev_path.clone(), - parse_bridge_kind(info.bridge_kind.as_deref()), + ( + BridgeContext::from_parts( + info.subdev_path.clone(), + parse_bridge_kind(info.bridge_kind.as_deref()), + ), + info.control_mode, ) }) - .unwrap_or_default() + .unwrap_or((BridgeContext::default(), VideoControlMode::Configurable)) }); + let recovery_policy = CaptureRecoveryPolicy::new(control_mode); // ── STREAMON gate: for CSI bridges with a subdev, refuse to // open the video node when the subdev reports no signal. // On RK628 this prevents a kernel null-pointer deref. if let Some(subdev_path) = bridge_ctx.subdev_path.as_ref() { - match probe_subdev_signal(subdev_path, bridge_ctx.kind) { - Some(crate::video::signal::SignalStatus::NoCable) - | Some(crate::video::signal::SignalStatus::NoSync) - | Some(crate::video::signal::SignalStatus::NoSignal) - | Some(crate::video::signal::SignalStatus::OutOfRange) => { - let status = probe_subdev_signal(subdev_path, bridge_ctx.kind) - .unwrap_or(crate::video::signal::SignalStatus::NoSignal); - let wait_secs = backoff_secs(no_signal_restart_count); - debug!( - "Pre-STREAMON gate: subdev {:?} reports {:?} — \ - waiting for SOURCE_CHANGE (<= {}s) before opening {:?}", - subdev_path, status, wait_secs, device_path - ); - set_retry(wait_secs.saturating_mul(1000)); - go_offline(); - set_state(status.into()); - // Wait for SOURCE_CHANGE or timeout before retrying. - // Opens the subdev just for the poll — cheap and - // does NOT touch the video node. - wait_subdev_for_source_change( - subdev_path, - &self.direct_stop, - Duration::from_secs(wait_secs), - ); - no_signal_restart_count = no_signal_restart_count.saturating_add(1); - continue 'session; - } - _ => {} // Locked (None from as_status) or unknown — proceed + if let Some(status) = probe_subdev_signal(subdev_path, bridge_ctx.kind) { + let delay = + recovery_policy.retry_delay(no_signal_restart_count.saturating_add(1)); + debug!( + "Pre-STREAMON gate: subdev {:?} reports {:?} — \ + waiting for SOURCE_CHANGE (<= {:?}) before opening {:?}", + subdev_path, status, delay, device_path + ); + set_retry(delay.as_millis() as u64); + go_offline(); + set_state(status.into()); + wait_for_source_change(&bridge_ctx, delay, || { + !self.direct_stop.load(Ordering::Relaxed) + }); + no_signal_restart_count = no_signal_restart_count.saturating_add(1); + continue 'session; } } @@ -922,6 +916,7 @@ impl Streamer { BUFFER_COUNT, Duration::from_secs(2), bridge_ctx.clone(), + control_mode, ) { Ok(stream) => { stream_opt = Some(stream); @@ -938,7 +933,9 @@ impl Streamer { "CSI open probe reports no signal ({:?}), will soft-restart", status ); - set_retry(backoff_secs(no_signal_restart_count).saturating_mul(1000)); + let delay = + recovery_policy.retry_delay(no_signal_restart_count.saturating_add(1)); + set_retry(delay.as_millis() as u64); go_offline(); set_state(status.into()); last_error = Some(format!("CaptureNoSignal({})", kind)); @@ -987,9 +984,16 @@ impl Streamer { } debug!("Open failed in NoSignal-like state, backing off before soft-restart"); - let wait = backoff_secs(no_signal_restart_count); - set_retry(wait.saturating_mul(1000)); - std::thread::sleep(Duration::from_secs(wait)); + if !recovery_policy.should_retry(no_signal_restart_count.saturating_add(1)) { + set_state(StreamerState::Error); + break 'session; + } + let delay = + recovery_policy.retry_delay(no_signal_restart_count.saturating_add(1)); + set_retry(delay.as_millis() as u64); + wait_for_source_change(&bridge_ctx, delay, || { + !self.direct_stop.load(Ordering::Relaxed) + }); no_signal_restart_count = no_signal_restart_count.saturating_add(1); continue 'session; } @@ -997,8 +1001,21 @@ impl Streamer { let resolution = stream.resolution(); let pixel_format = stream.format(); + let source_fps = stream + .source_fps() + .map(|fps| fps.round().clamp(1.0, 120.0) as u32) + .unwrap_or(config.fps); let stride = stream.stride(); + if control_mode == VideoControlMode::SourceFollowing { + handle.block_on(async { + let mut current = self.config.write().await; + current.resolution = resolution; + current.format = pixel_format; + current.fps = source_fps; + }); + } + info!( "Capture format: {}x{} {:?} stride={}", resolution.width, resolution.height, pixel_format, stride @@ -1053,7 +1070,9 @@ impl Streamer { Err(e) => { if is_source_changed_error(&e) { info!("Capture SOURCE_CHANGE — soft-restart for DV re-probe"); - set_retry(backoff_secs(no_signal_restart_count).saturating_mul(1000)); + let delay = recovery_policy + .retry_delay(no_signal_restart_count.saturating_add(1)); + set_retry(delay.as_millis() as u64); go_offline(); set_state(StreamerState::NoSignal); need_soft_restart = true; @@ -1062,8 +1081,9 @@ impl Streamer { if e.kind() == std::io::ErrorKind::TimedOut { if signal_present { signal_present = false; - let wait = backoff_secs(no_signal_restart_count); - set_retry(wait.saturating_mul(1000)); + let delay = recovery_policy + .retry_delay(no_signal_restart_count.saturating_add(1)); + set_retry(delay.as_millis() as u64); go_offline(); set_state(StreamerState::NoSignal); no_signal_since = Some(std::time::Instant::now()); @@ -1071,11 +1091,12 @@ impl Streamer { fps_frame_count = 0; last_fps_time = std::time::Instant::now(); } else if let Some(since) = no_signal_since { - let wait = backoff_secs(no_signal_restart_count); - if since.elapsed().as_secs() >= wait { + let delay = recovery_policy + .retry_delay(no_signal_restart_count.saturating_add(1)); + if since.elapsed() >= delay { info!( - "NoSignal for {}s, attempting soft restart (attempt {})", - wait, + "NoSignal for {:?}, attempting soft restart (attempt {})", + delay, no_signal_restart_count + 1 ); need_soft_restart = true; @@ -1122,12 +1143,7 @@ impl Streamer { "Capture transient error (EPROTO/-71, often UVC USB): {}", e ); - let is_uvc = handle.block_on(async { - self.current_device.read().await.as_ref().is_some_and(|d| { - d.driver.eq_ignore_ascii_case("uvcvideo") - }) - }); - if is_uvc { + if control_mode == VideoControlMode::Configurable { go_offline(); set_state(StreamerState::UvcUsbError); need_soft_restart = true; @@ -1139,9 +1155,9 @@ impl Streamer { e ); } - set_retry( - backoff_secs(no_signal_restart_count).saturating_mul(1000), - ); + let delay = recovery_policy + .retry_delay(no_signal_restart_count.saturating_add(1)); + set_retry(delay.as_millis() as u64); go_offline(); set_state(StreamerState::NoSignal); need_soft_restart = true; @@ -1192,25 +1208,28 @@ impl Streamer { self.mjpeg_handler.set_online(); set_state(StreamerState::Streaming); - let fps_val = config.fps; - let current = (resolution.width, resolution.height, pixel_format, fps_val); - if last_applied != Some(current) { - last_applied = Some(current); - let dp = device_path.display().to_string(); - let fmt = format!("{:?}", pixel_format); - let w = resolution.width; - let h = resolution.height; - handle.block_on(async { - self.publish_event(SystemEvent::StreamConfigApplied { - transition_id: None, - device: dp, - resolution: (w, h), - format: fmt, - fps: fps_val, - }) - .await; - }); - } + let fps_val = source_fps; + let recovered_device = device_path.display().to_string(); + handle.block_on(async { + self.publish_event(SystemEvent::StreamRecovered { + device: recovered_device, + }) + .await; + }); + let dp = device_path.display().to_string(); + let fmt = pixel_format.to_string(); + let w = resolution.width; + let h = resolution.height; + handle.block_on(async { + self.publish_event(SystemEvent::StreamConfigApplied { + transition_id: None, + device: dp, + resolution: (w, h), + format: fmt, + fps: fps_val, + }) + .await; + }); } self.mjpeg_handler.update_frame(frame); @@ -1240,65 +1259,8 @@ impl Streamer { } no_signal_restart_count = no_signal_restart_count.saturating_add(1); - - match VideoDevice::open_readonly(&device_path).and_then(|d| d.info()) { - Ok(device_info) => { - // Skip re-open while rkcif still reports placeholder (≤64²) geometry. - let probed_res = device_info - .formats - .first() - .and_then(|f| f.resolutions.first()) - .map(|r| (r.width, r.height)); - - if matches!(probed_res, Some((w, h)) if w <= 64 || h <= 64) - || probed_res.is_none() - { - warn!( - "Soft restart: probed resolution too small ({:?}), still no signal", - probed_res - ); - set_retry(2_000); - go_offline(); - std::thread::sleep(Duration::from_secs(2)); - continue 'session; - } - - handle.block_on(async { - let fmt; - let res; - { - let cfg = self.config.read().await; - fmt = self - .select_format(&device_info, cfg.format) - .unwrap_or(cfg.format); - res = self - .select_resolution(&device_info, &fmt, cfg.resolution) - .unwrap_or(cfg.resolution); - } - { - let mut cfg = self.config.write().await; - cfg.format = fmt; - cfg.resolution = res; - } - *self.current_device.write().await = Some(device_info); - info!( - "Soft restart: re-probed device → {}x{} {:?}", - res.width, res.height, fmt - ); - }); - } - Err(e) => { - warn!("Soft restart: failed to re-probe device: {}", e); - // Brief wait before retrying to avoid spinning. - let wait = 2u64.pow(no_signal_restart_count.min(3)); - std::thread::sleep(Duration::from_secs(wait)); - } - } - - // Reset no_signal_since so the back-off timer is fresh for the new session. - // no_signal_since will be re-set if the new session immediately times out. - - // Continue 'session → re-open CaptureStream with updated config. + // Continue 'session: the single open path performs QUERY_DV_TIMINGS, + // applies the source mode, and owns the retry delay. } // 'session self.direct_active.store(false, Ordering::SeqCst); @@ -1316,28 +1278,28 @@ impl Streamer { .map_err(|e| AppError::VideoError(format!("Cannot open device for re-init: {}", e)))?; let device_info = device.info()?; - let (format, resolution) = { + let resolved = { let config = self.config.read().await; - let fmt = self - .select_format(&device_info, config.format) - .unwrap_or(config.format); - let res = self - .select_resolution(&device_info, &fmt, config.resolution) - .unwrap_or(config.resolution); - (fmt, res) + self.resolve_capture_config(&device_info, config.format, config.resolution, config.fps) + .unwrap_or(super::device::ResolvedVideoInputConfig { + format: config.format, + resolution: config.resolution, + fps: config.fps, + }) }; { let mut cfg = self.config.write().await; cfg.device_path = Some(device_info.path.clone()); - cfg.format = format; - cfg.resolution = resolution; + cfg.format = resolved.format; + cfg.resolution = resolved.resolution; + cfg.fps = resolved.fps; } *self.current_device.write().await = Some(device_info); info!( "Device re-initialized: {}x{} {:?}", - resolution.width, resolution.height, format + resolved.resolution.width, resolved.resolution.height, resolved.format ); Ok(()) } @@ -1619,50 +1581,6 @@ fn probe_subdev_signal( probe.as_status() } -fn wait_subdev_for_source_change( - subdev_path: &std::path::Path, - direct_stop: &AtomicBool, - max_wait: Duration, -) { - let fd = match csi_bridge::open_subdev(subdev_path) { - Ok(f) => f, - Err(e) => { - debug!( - "wait_subdev_for_source_change: failed to open {:?}: {}", - subdev_path, e - ); - std::thread::sleep(max_wait.min(Duration::from_secs(1))); - return; - } - }; - if let Err(e) = csi_bridge::subscribe_source_change(&fd) { - debug!( - "wait_subdev_for_source_change: subscribe failed on {:?}: {}", - subdev_path, e - ); - } - let slice = Duration::from_millis(250); - let deadline = std::time::Instant::now() + max_wait; - while std::time::Instant::now() < deadline { - if direct_stop.load(Ordering::Relaxed) { - return; - } - let remaining = deadline.saturating_duration_since(std::time::Instant::now()); - let wait = remaining.min(slice); - match csi_bridge::wait_source_change(&fd, wait) { - Ok(true) => { - info!("Subdev SOURCE_CHANGE during no-signal wait, retrying open immediately"); - return; - } - Ok(false) => continue, - Err(e) => { - debug!("wait_source_change error on {:?}: {}", subdev_path, e); - return; - } - } - } -} - impl serde::Serialize for StreamerState { fn serialize(&self, serializer: S) -> std::result::Result where diff --git a/src/video/traits.rs b/src/video/traits.rs index 715c5dc4..3ab62e9c 100644 --- a/src/video/traits.rs +++ b/src/video/traits.rs @@ -10,6 +10,7 @@ use super::types::{ use crate::error::Result; use crate::events::EventBus; use crate::hid::HidController; +use crate::video::device::VideoDeviceInfo; /// Trait for video output consumers that receive encoded video frames. /// @@ -24,14 +25,12 @@ pub trait VideoOutput: Send + Sync { &self, device_path: PathBuf, jpeg_quality: u8, - subdev_path: Option, - bridge_kind: Option, - v4l2_driver: Option, + device_info: Option, ); async fn current_video_codec(&self) -> VideoCodecType; async fn is_hardware_encoding(&self) -> bool; async fn close_all_sessions(&self); - async fn close_all_sessions_and_release_device(&self) -> usize; + async fn close_all_sessions_and_release_device(&self) -> Result; async fn session_count(&self) -> usize; async fn set_hid_controller(&self, hid: Arc); async fn set_audio_enabled(&self, enabled: bool) -> Result<()>; diff --git a/src/video/types.rs b/src/video/types.rs index 56ad9f9f..02845759 100644 --- a/src/video/types.rs +++ b/src/video/types.rs @@ -17,6 +17,6 @@ pub use super::codec::registry::{EncoderBackend, VideoEncoderType}; // From video::pipeline pub use super::pipeline::{ - EncodedVideoFrame, PipelineStateNotification, SharedVideoPipeline, SharedVideoPipelineConfig, - SharedVideoPipelineStats, + EncodedVideoFrame, PipelineAppliedConfig, PipelineStateNotification, SharedVideoPipeline, + SharedVideoPipelineConfig, SharedVideoPipelineStats, }; diff --git a/src/web/handlers/config/types.rs b/src/web/handlers/config/types.rs index 3ec9299f..83f0d487 100644 --- a/src/web/handlers/config/types.rs +++ b/src/web/handlers/config/types.rs @@ -54,6 +54,22 @@ pub struct VideoConfigUpdate { } impl VideoConfigUpdate { + pub fn ignore_source_following_parameters(&mut self) { + if self.format.is_some() + || self.width.is_some() + || self.height.is_some() + || self.fps.is_some() + { + tracing::debug!( + "Ignoring client-supplied format, resolution, and FPS for source-following video input" + ); + } + self.format = None; + self.width = None; + self.height = None; + self.fps = None; + } + pub fn validate(&self) -> crate::error::Result<()> { if let Some(width) = self.width { if !(320..=7680).contains(&width) { @@ -106,6 +122,32 @@ impl VideoConfigUpdate { } } +#[cfg(test)] +mod video_config_update_tests { + use super::VideoConfigUpdate; + + #[test] + fn source_following_parameters_are_silently_discarded() { + let mut update = VideoConfigUpdate { + device: Some("/dev/video0".to_string()), + format: Some("MJPEG".to_string()), + width: Some(7680), + height: Some(4320), + fps: Some(120), + quality: Some(90), + }; + update.ignore_source_following_parameters(); + + assert_eq!(update.device.as_deref(), Some("/dev/video0")); + assert!(update.format.is_none()); + assert!(update.width.is_none()); + assert!(update.height.is_none()); + assert!(update.fps.is_none()); + assert_eq!(update.quality, Some(90)); + assert!(update.validate().is_ok()); + } +} + /// Stream configuration response #[typeshare] #[derive(Debug, serde::Serialize)] diff --git a/src/web/handlers/config/video.rs b/src/web/handlers/config/video.rs index 865da9d5..d1e3b039 100644 --- a/src/web/handlers/config/video.rs +++ b/src/web/handlers/config/video.rs @@ -14,8 +14,30 @@ pub async fn get_video_config(State(state): State>) -> Json>, - Json(req): Json, + Json(mut req): Json, ) -> Result> { + let selected_path = req + .device + .clone() + .or_else(|| state.config.get().video.device.clone()); + if let Some(path) = selected_path { + let source_following = state + .stream_manager + .list_devices() + .await + .ok() + .and_then(|devices| { + devices + .into_iter() + .find(|device| device.path.to_string_lossy() == path) + }) + .is_some_and(|device| { + device.control_mode == crate::video::device::VideoControlMode::SourceFollowing + }); + if source_following { + req.ignore_source_following_parameters(); + } + } req.validate()?; let _apply_guard = try_apply_lock(&state.config_apply_locks.video, "video")?; diff --git a/src/web/handlers/inventory.rs b/src/web/handlers/inventory.rs index 5908d1ed..25082a19 100644 --- a/src/web/handlers/inventory.rs +++ b/src/web/handlers/inventory.rs @@ -23,6 +23,13 @@ pub struct VideoDevice { pub formats: Vec, pub usb_bus: Option, pub has_signal: bool, + pub control_mode: crate::video::device::VideoControlMode, + pub input_status: crate::video::device::VideoInputStatus, +} + +#[derive(Deserialize)] +pub struct VideoInputStatusQuery { + pub device: String, } #[derive(Serialize)] @@ -121,6 +128,8 @@ pub async fn list_devices(State(state): State>) -> Json>) -> Json Option { + let path = std::path::Path::new(path); + let name = path.file_name()?.to_str()?; + if path.parent() != Some(std::path::Path::new("/dev")) + || !name.starts_with("video") + || name.len() == "video".len() + || !name["video".len()..].chars().all(|c| c.is_ascii_digit()) + || !sysfs_root.join(name).exists() + { + return None; + } + Some(path.to_path_buf()) +} + +pub async fn video_input_status( + Query(query): Query, +) -> Result> { + #[cfg(unix)] + let path = validated_video_node( + &query.device, + std::path::Path::new("/sys/class/video4linux"), + ) + .ok_or_else(|| AppError::BadRequest("Invalid video device".to_string()))?; + + #[cfg(windows)] + let path = crate::video::device::enumerate_devices()? + .into_iter() + .find(|device| device.path.to_string_lossy() == query.device) + .map(|device| device.path) + .ok_or_else(|| AppError::BadRequest("Invalid video device".to_string()))?; + + let probe_path = path.clone(); + let status = tokio::task::spawn_blocking(move || { + crate::video::device::VideoDevice::open_readonly(&probe_path) + .and_then(|device| device.input_status()) + }) + .await + .ok() + .and_then(|result| result.ok()) + .unwrap_or_else(|| { + debug!(device = %path.display(), "Unable to read video input status"); + crate::video::device::VideoInputStatus::unavailable() + }); + + Ok(Json(status)) +} + +#[cfg(all(test, unix))] +mod tests { + use super::validated_video_node; + + #[test] + fn only_accepts_dev_video_nodes_present_in_sysfs() { + let root = tempfile::tempdir().unwrap(); + std::fs::create_dir(root.path().join("video7")).unwrap(); + + assert_eq!( + validated_video_node("/dev/video7", root.path()).unwrap(), + std::path::PathBuf::from("/dev/video7") + ); + assert!(validated_video_node("/dev/video8", root.path()).is_none()); + assert!(validated_video_node("/tmp/video7", root.path()).is_none()); + assert!(validated_video_node("/dev/video7/../mem", root.path()).is_none()); + assert!(validated_video_node("/dev/video", root.path()).is_none()); + } +} diff --git a/src/web/handlers/mod.rs b/src/web/handlers/mod.rs index 521861b8..b34927c7 100644 --- a/src/web/handlers/mod.rs +++ b/src/web/handlers/mod.rs @@ -40,7 +40,7 @@ use axum::{ use axum_extra::extract::cookie::{Cookie, CookieJar, SameSite}; use serde::{Deserialize, Serialize}; use std::sync::Arc; -use tracing::{info, warn}; +use tracing::{debug, info, warn}; use self::config::apply::ConfigApplyOptions; use crate::auth::{Session, SESSION_COOKIE}; diff --git a/src/web/handlers/setup.rs b/src/web/handlers/setup.rs index 0179cf83..80392462 100644 --- a/src/web/handlers/setup.rs +++ b/src/web/handlers/setup.rs @@ -44,7 +44,7 @@ pub struct SetupRequest { pub async fn setup_init( State(state): State>, - Json(req): Json, + Json(mut req): Json, ) -> Result> { // Check if already initialized if state.config.is_initialized() { @@ -65,6 +65,37 @@ pub async fn setup_init( )); } + if let Some(path) = req.video_device.as_deref() { + let source_following = state + .stream_manager + .list_devices() + .await + .ok() + .and_then(|devices| { + devices + .into_iter() + .find(|device| device.path.to_string_lossy() == path) + }) + .is_some_and(|device| { + device.control_mode == crate::video::device::VideoControlMode::SourceFollowing + }); + if source_following { + if req.video_format.is_some() + || req.video_width.is_some() + || req.video_height.is_some() + || req.video_fps.is_some() + { + tracing::debug!( + "Ignoring setup-supplied format, resolution, and FPS for source-following video input" + ); + } + req.video_format = None; + req.video_width = None; + req.video_height = None; + req.video_fps = None; + } + } + let old_config = state.config.get(); // Create single system user diff --git a/src/web/routes.rs b/src/web/routes.rs index 21e74e4e..e2a6b30c 100644 --- a/src/web/routes.rs +++ b/src/web/routes.rs @@ -62,6 +62,7 @@ pub fn create_router(state: Arc) -> Router { ) .route("/auth/totp/disable", post(handlers::disable_totp)) .route("/devices", get(handlers::list_devices)) + .route("/video/input-status", get(handlers::video_input_status)) // WebSocket endpoint for real-time events .route("/ws", any(ws_handler)) // Stream control endpoints diff --git a/src/webrtc/webrtc_streamer.rs b/src/webrtc/webrtc_streamer.rs index 8a1fd2c6..3a8588db 100644 --- a/src/webrtc/webrtc_streamer.rs +++ b/src/webrtc/webrtc_streamer.rs @@ -15,7 +15,8 @@ use crate::hid::HidController; use crate::video::capture::DEFAULT_CAPTURE_BUFFER_COUNT; use crate::video::codec::h264_bitstream; use crate::video::device::{ - enumerate_devices, select_recovery_device, VideoDevice, VideoDeviceRecoveryHint, + enumerate_devices, select_recovery_device, VideoControlMode, VideoDevice, VideoDeviceInfo, + VideoDeviceRecoveryHint, }; use crate::video::types::{ BitratePreset, EncoderBackend, PipelineStateNotification, PixelFormat, Resolution, @@ -28,6 +29,18 @@ use super::signaling::{ConnectionState, IceCandidate, SdpAnswer, SdpOffer}; use super::universal_session::{UniversalSession, UniversalSessionConfig}; const H264_PROFILE_DETECT_TIMEOUT: Duration = Duration::from_millis(500); +const PIPELINE_RELEASE_TIMEOUT: Duration = Duration::from_secs(5); + +fn update_signal_recovery_edge(pending: &AtomicBool, state: &str) -> bool { + match state { + "no_signal" => { + pending.store(true, Ordering::Release); + false + } + "streaming" => pending.swap(false, Ordering::AcqRel), + _ => false, + } +} #[derive(Debug, Clone)] pub struct WebRtcStreamerConfig { @@ -63,8 +76,7 @@ pub struct CaptureDeviceConfig { pub jpeg_quality: u8, pub subdev_path: Option, pub bridge_kind: Option, - /// V4L2 driver name (e.g. `uvcvideo`) for UVC-specific recovery hints. - pub v4l2_driver: Option, + pub control_mode: VideoControlMode, pub recovery_hint: VideoDeviceRecoveryHint, } @@ -99,6 +111,7 @@ pub struct WebRtcStreamer { hid_controller: RwLock>>, events: RwLock>>, recovery_in_progress: AtomicBool, + signal_recovery_pending: Arc, self_weak: StdRwLock>>, } @@ -119,6 +132,7 @@ impl WebRtcStreamer { hid_controller: RwLock::new(None), events: RwLock::new(None), recovery_in_progress: AtomicBool::new(false), + signal_recovery_pending: Arc::new(AtomicBool::new(false)), self_weak: StdRwLock::new(None), }); let weak = Arc::downgrade(&streamer); @@ -154,11 +168,7 @@ impl WebRtcStreamer { // Close all existing sessions self.close_all_sessions().await; - // Stop current pipeline - if let Some(ref pipeline) = *self.video_pipeline.read().await { - pipeline.stop(); - } - *self.video_pipeline.write().await = None; + self.stop_video_pipeline_and_release().await?; // Update codec *self.video_codec.write().await = codec; @@ -231,18 +241,48 @@ impl WebRtcStreamer { } } + /// Serialize pipeline teardown with creation and return only after V4L2 + /// STREAMOFF, buffer teardown and FD close have completed. + async fn stop_video_pipeline_and_release(&self) -> Result<()> { + let mut pipeline_guard = self.video_pipeline.write().await; + let Some(pipeline) = pipeline_guard.as_ref().cloned() else { + return Ok(()); + }; + + pipeline.stop_and_wait(PIPELINE_RELEASE_TIMEOUT).await?; + *pipeline_guard = None; + Ok(()) + } + fn build_pipeline_state_notifier( device: String, events: Option>, + recovery_pending: Arc, ) -> Option> { events.map(|events| { Arc::new(move |notification: PipelineStateNotification| { + let recovered = update_signal_recovery_edge(&recovery_pending, notification.state); events.publish(SystemEvent::StreamStateChanged { state: notification.state.to_string(), device: Some(device.clone()), reason: notification.reason.map(|reason| reason.to_string()), next_retry_ms: notification.next_retry_ms, }); + if recovered { + events.publish(SystemEvent::StreamRecovered { + device: device.clone(), + }); + if let Some(applied) = notification.applied_config { + events.publish(SystemEvent::StreamConfigApplied { + transition_id: None, + device: device.clone(), + resolution: (applied.resolution.width, applied.resolution.height), + format: applied.format.to_string(), + fps: applied.fps, + }); + } + events.mark_device_info_dirty(); + } }) as Arc }) } @@ -330,7 +370,7 @@ impl WebRtcStreamer { jpeg_quality, subdev_path: device.subdev_path.clone(), bridge_kind: device.bridge_kind.clone(), - v4l2_driver: Some(device.driver.clone()), + control_mode: device.control_mode, recovery_hint: VideoDeviceRecoveryHint::from(&device), }; @@ -347,6 +387,7 @@ impl WebRtcStreamer { debug!("WebRTC video recovery already in progress"); return; } + self.signal_recovery_pending.store(true, Ordering::Release); let streamer = self.clone(); tokio::spawn(async move { @@ -412,24 +453,11 @@ impl WebRtcStreamer { { Ok(reconnected) => { info!( - "WebRTC video recovered with {} after {} attempts, reconnected {} sessions", + "WebRTC capture reopened with {} after {} attempts; reconnected {} sessions and waiting for first frame", device.path.display(), attempt, reconnected ); - streamer - .publish_stream_event(SystemEvent::StreamRecovered { - device: device.path.display().to_string(), - }) - .await; - streamer - .publish_stream_event(SystemEvent::StreamStateChanged { - state: "streaming".to_string(), - device: Some(device.path.display().to_string()), - reason: None, - next_retry_ms: None, - }) - .await; streamer.recovery_in_progress.store(false, Ordering::SeqCst); return; } @@ -460,6 +488,13 @@ impl WebRtcStreamer { let pipeline_config = { let config = self.config.read().await; SharedVideoPipelineConfig { + control_mode: self + .capture_device + .read() + .await + .as_ref() + .map(|capture| capture.control_mode) + .unwrap_or(VideoControlMode::Configurable), resolution: config.resolution, input_format: config.input_format, output_codec: Self::codec_type_to_encoder_type(codec), @@ -476,6 +511,7 @@ impl WebRtcStreamer { pipeline.set_state_notifier(Self::build_pipeline_state_notifier( device.device_path.display().to_string(), self.events.read().await.clone(), + self.signal_recovery_pending.clone(), )); pipeline .start_with_device( @@ -484,7 +520,6 @@ impl WebRtcStreamer { device.jpeg_quality, device.subdev_path, device.bridge_kind, - device.v4l2_driver, ) .await?; } else { @@ -532,7 +567,8 @@ impl WebRtcStreamer { let should_reconnect = pending_geometry.is_some(); if let Some((r, f)) = pending_geometry { - streamer.sync_video_geometry_from_negotiated(r, f).await; + let fps = streamer.config.read().await.fps; + streamer.sync_video_input_from_negotiated(r, f, fps).await; } if should_reconnect { let streamer_for_reconnect = streamer.clone(); @@ -570,9 +606,10 @@ impl WebRtcStreamer { }); let pipeline_cfg = pipeline.config().await; - self.sync_video_geometry_from_negotiated( + self.sync_video_input_from_negotiated( pipeline_cfg.resolution, pipeline_cfg.input_format, + pipeline_cfg.fps, ) .await; @@ -725,32 +762,41 @@ impl WebRtcStreamer { &self, device_path: PathBuf, jpeg_quality: u8, - subdev_path: Option, - bridge_kind: Option, - v4l2_driver: Option, + device_info: Option, ) { + let (subdev_path, bridge_kind, control_mode, recovery_hint) = match device_info { + Some(info) => ( + info.subdev_path.clone(), + info.bridge_kind.clone(), + info.control_mode, + VideoDeviceRecoveryHint::from(&info), + ), + None => { + let recovery_hint = VideoDevice::open_readonly(&device_path) + .and_then(|device| device.info()) + .map(|info| VideoDeviceRecoveryHint::from(&info)) + .unwrap_or_else(|_| VideoDeviceRecoveryHint { + path: device_path.clone(), + name: String::new(), + driver: String::new(), + bus_info: String::new(), + card: String::new(), + is_capture_card: true, + }); + (None, None, VideoControlMode::Configurable, recovery_hint) + } + }; debug!( - "Setting direct capture device for WebRTC: {:?} (subdev={:?}, kind={:?}, driver={:?})", - device_path, subdev_path, bridge_kind, v4l2_driver + "Setting direct capture device for WebRTC: {:?} (subdev={:?}, kind={:?}, mode={:?})", + device_path, subdev_path, bridge_kind, control_mode ); - let recovery_hint = VideoDevice::open_readonly(&device_path) - .and_then(|device| device.info()) - .map(|info| VideoDeviceRecoveryHint::from(&info)) - .unwrap_or_else(|_| VideoDeviceRecoveryHint { - path: device_path.clone(), - name: String::new(), - driver: v4l2_driver.clone().unwrap_or_default(), - bus_info: String::new(), - card: String::new(), - is_capture_card: true, - }); *self.capture_device.write().await = Some(CaptureDeviceConfig { device_path, buffer_count: DEFAULT_CAPTURE_BUFFER_COUNT, jpeg_quality, subdev_path, bridge_kind, - v4l2_driver, + control_mode, recovery_hint, }); } @@ -764,12 +810,13 @@ impl WebRtcStreamer { /// /// This stops the encoding pipeline and closes all sessions. pub async fn prepare_for_config_change(&self) { - // Stop pipeline and close sessions - will be recreated on next session - if let Some(ref pipeline) = *self.video_pipeline.read().await { - pipeline.stop(); - } - *self.video_pipeline.write().await = None; self.close_all_sessions().await; + if let Err(error) = self.stop_video_pipeline_and_release().await { + warn!( + "Failed to release video pipeline for config change: {}", + error + ); + } } // === Configuration === @@ -804,12 +851,6 @@ impl WebRtcStreamer { resolution.width, resolution.height, format, fps ); - // Stop existing pipeline - if let Some(ref pipeline) = *self.video_pipeline.read().await { - pipeline.stop(); - } - *self.video_pipeline.write().await = None; - // Close all existing sessions - they need to reconnect let session_count = self.close_all_sessions().await; if session_count > 0 { @@ -818,6 +859,13 @@ impl WebRtcStreamer { session_count ); } + if let Err(error) = self.stop_video_pipeline_and_release().await { + warn!( + "Failed to release video pipeline for config change: {}", + error + ); + return; + } // Update config (preserve user-configured bitrate) { @@ -836,31 +884,36 @@ impl WebRtcStreamer { self.notify_device_info_dirty().await; } - /// Update resolution/format to match DV-negotiated capture without stopping + /// Update the input mode to match DV-negotiated capture without stopping /// the pipeline or closing sessions. Used when hardware timing differs from /// saved settings (e.g. RK628 `S_FMT` follows source while SQLite still has /// a user-chosen preset). - pub async fn sync_video_geometry_from_negotiated( + pub async fn sync_video_input_from_negotiated( &self, resolution: Resolution, format: PixelFormat, + fps: u32, ) { { let mut config = self.config.write().await; - if config.resolution == resolution && config.input_format == format { + if config.resolution == resolution && config.input_format == format && config.fps == fps + { return; } info!( - "WebRTC geometry aligned to negotiated capture: {}x{} {:?} (was {}x{} {:?})", + "WebRTC input aligned to negotiated capture: {}x{} {:?} @ {} fps (was {}x{} {:?} @ {} fps)", resolution.width, resolution.height, format, + fps, config.resolution.width, config.resolution.height, - config.input_format + config.input_format, + config.fps, ); config.resolution = resolution; config.input_format = format; + config.fps = fps; } self.notify_device_info_dirty().await; @@ -868,12 +921,6 @@ impl WebRtcStreamer { /// Update encoder backend (software/hardware selection) pub async fn update_encoder_backend(&self, encoder_backend: Option) { - // Stop existing pipeline - if let Some(ref pipeline) = *self.video_pipeline.read().await { - pipeline.stop(); - } - *self.video_pipeline.write().await = None; - // Close all existing sessions - they need to reconnect with new encoder let session_count = self.close_all_sessions().await; if session_count > 0 { @@ -882,6 +929,13 @@ impl WebRtcStreamer { session_count ); } + if let Err(error) = self.stop_video_pipeline_and_release().await { + warn!( + "Failed to release video pipeline for encoder backend change: {}", + error + ); + return; + } // Update config let mut config = self.config.write().await; @@ -1123,17 +1177,12 @@ impl WebRtcStreamer { /// Close all sessions and wait for the video pipeline to fully release the /// capture device. Use this when the caller needs the V4L2 device immediately /// afterwards (e.g. switching to MJPEG mode). - pub async fn close_all_sessions_and_release_device(&self) -> usize { + pub async fn close_all_sessions_and_release_device(&self) -> Result { let count = self.close_all_sessions().await; - if let Some(ref pipeline) = *self.video_pipeline.read().await { - pipeline - .stop_and_wait(std::time::Duration::from_secs(3)) - .await; - } - *self.video_pipeline.write().await = None; + self.stop_video_pipeline_and_release().await?; - count + Ok(count) } /// Get session count @@ -1256,16 +1305,7 @@ impl WebRtcStreamer { if pipeline_running { info!("Restarting video pipeline to apply new bitrate: {}", preset); - // Stop existing pipeline - if let Some(ref pipeline) = *self.video_pipeline.read().await { - pipeline.stop(); - } - - // Wait for pipeline to stop - tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; - - // Clear pipeline reference - will be recreated - *self.video_pipeline.write().await = None; + self.stop_video_pipeline_and_release().await?; let has_source = self.capture_device.read().await.is_some(); if !has_source { @@ -1306,18 +1346,10 @@ impl crate::video::traits::VideoOutput for WebRtcStreamer { &self, device_path: PathBuf, jpeg_quality: u8, - subdev_path: Option, - bridge_kind: Option, - v4l2_driver: Option, + device_info: Option, ) { - self.set_capture_device( - device_path, - jpeg_quality, - subdev_path, - bridge_kind, - v4l2_driver, - ) - .await; + self.set_capture_device(device_path, jpeg_quality, device_info) + .await; } async fn current_video_codec(&self) -> VideoCodecType { @@ -1332,7 +1364,7 @@ impl crate::video::traits::VideoOutput for WebRtcStreamer { self.close_all_sessions().await; } - async fn close_all_sessions_and_release_device(&self) -> usize { + async fn close_all_sessions_and_release_device(&self) -> Result { self.close_all_sessions_and_release_device().await } @@ -1398,6 +1430,7 @@ impl Default for WebRtcStreamer { hid_controller: RwLock::new(None), events: RwLock::new(None), recovery_in_progress: AtomicBool::new(false), + signal_recovery_pending: Arc::new(AtomicBool::new(false)), self_weak: StdRwLock::new(None), } } @@ -1431,4 +1464,14 @@ mod tests { assert!(!WebRtcStreamer::should_stop_pipeline(0, 1)); assert!(!WebRtcStreamer::should_stop_pipeline(2, 3)); } + + #[test] + fn recovery_edge_is_emitted_once_after_first_streaming_frame() { + let pending = AtomicBool::new(false); + assert!(!update_signal_recovery_edge(&pending, "streaming")); + assert!(!update_signal_recovery_edge(&pending, "no_signal")); + assert!(!update_signal_recovery_edge(&pending, "no_signal")); + assert!(update_signal_recovery_edge(&pending, "streaming")); + assert!(!update_signal_recovery_edge(&pending, "streaming")); + } } diff --git a/web/src/api/index.ts b/web/src/api/index.ts index e449ccaa..39319c9b 100644 --- a/web/src/api/index.ts +++ b/web/src/api/index.ts @@ -374,11 +374,12 @@ export const webrtcApi = { createSession: () => request<{ session_id: string }>('/webrtc/session', { method: 'POST' }), - offer: (sdp: string) => + offer: (sdp: string, signal?: AbortSignal) => request<{ sdp: string; session_id: string; ice_candidates: IceCandidate[] }>('/webrtc/offer', { method: 'POST', body: JSON.stringify({ sdp }), - }), + signal, + }, { toastOnError: false }), addIceCandidate: (sessionId: string, candidate: IceCandidate) => request<{ success: boolean }>('/webrtc/ice', { @@ -742,6 +743,56 @@ interface SerialDeviceOption { name: string } +export type VideoControlMode = 'configurable' | 'source_following' +export type VideoInputState = 'locked' | 'no_signal' | 'unavailable' + +export interface VideoInputStatus { + state: VideoInputState + format: string | null + width: number | null + height: number | null + fps: number | null +} + +export interface VideoResolution { + width: number + height: number + fps: number[] +} + +export interface VideoFormat { + format: string + description: string + resolutions: VideoResolution[] +} + +export interface VideoDevice { + path: string + name: string + driver: string + formats: VideoFormat[] + usb_bus: string | null + has_signal: boolean + control_mode: VideoControlMode + input_status: VideoInputStatus +} + +export interface DeviceList { + video: VideoDevice[] + serial: Array<{ path: string; name: string }> + audio: Array<{ + name: string + description: string + is_hdmi: boolean + usb_bus: string | null + }> + udc: Array<{ name: string }> + extensions: { + ttyd_available: boolean + rustdesk_available: boolean + } +} + function encodeDrivePath(path: string): string { if (path === '' || path === '/') { return '/' @@ -774,42 +825,20 @@ function sortSerialDevices(serialDevices: SerialDeviceOption[]): SerialDeviceOpt export const configApi = { listDevices: async () => { - const result = await request<{ - video: Array<{ - path: string - name: string - driver: string - formats: Array<{ - format: string - description: string - resolutions: Array<{ - width: number - height: number - fps: number[] - }> - }> - usb_bus: string | null - has_signal: boolean - }> - serial: Array<{ path: string; name: string }> - audio: Array<{ - name: string - description: string - is_hdmi: boolean - usb_bus: string | null - }> - udc: Array<{ name: string }> - extensions: { - ttyd_available: boolean - rustdesk_available: boolean - } - }>('/devices') + const result = await request('/devices') return { ...result, serial: sortSerialDevices(result.serial), } }, + + getVideoInputStatus: (device: string) => + request( + `/video/input-status?device=${encodeURIComponent(device)}`, + {}, + { toastOnError: false }, + ), } export { diff --git a/web/src/components/VideoConfigPopover.vue b/web/src/components/VideoConfigPopover.vue index 4045de09..9f1b3409 100644 --- a/web/src/components/VideoConfigPopover.vue +++ b/web/src/components/VideoConfigPopover.vue @@ -25,30 +25,17 @@ import { type EncoderBackendInfo, type BitratePreset, type StreamConstraintsResponse, + type VideoDevice, } from '@/api' import { getVideoFormatState, isVideoFormatSelectable } from '@/lib/video-format-support' -import { formatFpsLabel, toConfigFps } from '@/lib/fps' +import { toConfigFps } from '@/lib/fps' import { formatVideoDeviceLabel } from '@/lib/video-device-label' import { useConfigStore } from '@/stores/config' +import { useVideoDeviceConfiguration } from '@/composables/useVideoDeviceConfiguration' +import VideoInputFields from '@/components/VideoInputFields.vue' export type VideoMode = 'mjpeg' | 'h264' | 'h265' | 'vp8' | 'vp9' -interface VideoDevice { - path: string - name: string - driver: string - formats: { - format: string - description: string - resolutions: { - width: number - height: number - fps: number[] - }[] - }[] - has_signal?: boolean -} - const props = defineProps<{ open: boolean videoMode: VideoMode @@ -177,63 +164,36 @@ const translateBackendName = (backend: string | undefined): string => { return backend } -const hasHighFps = (format: { resolutions: { fps: number[] }[] }): boolean => { - return format.resolutions.some(res => res.fps.some(fps => fps >= 30)) -} - -const isFormatRecommended = (formatName: string): boolean => { - if (!isVideoFormatSelectable(formatName, props.videoMode, currentEncoderBackend.value)) { - return false - } - - const formats = availableFormats.value - const upperFormat = formatName.toUpperCase() - - // MJPEG/HTTP mode: recommend MJPEG - if (props.videoMode === 'mjpeg') { - return upperFormat === 'MJPEG' - } - - // WebRTC mode: check NV12 first, then YUYV - const currentFormat = formats.find(f => f.format.toUpperCase() === upperFormat) - if (!currentFormat) return false - - const nv12Format = formats.find(f => f.format.toUpperCase() === 'NV12') - const nv12HasHighFps = nv12Format && hasHighFps(nv12Format) - - const yuyvFormat = formats.find(f => f.format.toUpperCase() === 'YUYV') - const yuyvHasHighFps = yuyvFormat && hasHighFps(yuyvFormat) - - if (nv12HasHighFps) { - return upperFormat === 'NV12' - } - - if (yuyvHasHighFps) { - return upperFormat === 'YUYV' - } - - return false -} - -// In WebRTC mode, compressed formats (MJPEG/JPEG) are not recommended -const isFormatNotRecommended = (formatName: string): boolean => { - return getFormatState(formatName) === 'not_recommended' -} - const selectedDevice = ref('') const selectedFormat = ref('') const selectedResolution = ref('') -const selectedFps = ref(30) +const selectedFps = ref(30) const selectedBitratePreset = ref<'Speed' | 'Balanced' | 'Quality'>('Balanced') const isDirty = ref(false) -const selectedFormatStatus = computed<'recommended' | 'not_recommended' | 'unsupported' | null>(() => { - if (!selectedFormat.value) return null - if (isFormatUnsupported(selectedFormat.value)) return 'unsupported' - if (isFormatRecommended(selectedFormat.value)) return 'recommended' - if (isFormatNotRecommended(selectedFormat.value)) return 'not_recommended' - return null +const videoConfiguration = useVideoDeviceConfiguration({ + devices, + selection: { + device: selectedDevice, + format: selectedFormat, + resolution: selectedResolution, + fps: selectedFps, + }, + active: computed(() => props.open), + listenForStreamEvents: true, + preferredFormat: device => device.formats.find(format => + isVideoFormatSelectable(format.format, props.videoMode, currentEncoderBackend.value), + )?.format, }) +const { + selectedDevice: selectedDeviceInfo, + isSourceFollowing, + availableFormats, + availableResolutions, + availableFps, + refreshInputStatus, + refreshingInputStatus, +} = videoConfiguration const applying = ref(false) const applyingBitrate = ref(false) @@ -288,11 +248,6 @@ const availableCodecs = computed(() => { return backendFiltered.filter(codec => allowed.includes(codec.id)) }) -const availableFormats = computed(() => { - const device = devices.value.find(d => d.path === selectedDevice.value) - return device?.formats || [] -}) - const availableFormatOptions = computed(() => { return availableFormats.value.map(format => ({ ...format, @@ -301,32 +256,6 @@ const availableFormatOptions = computed(() => { })) }) -const availableResolutions = computed(() => { - const format = availableFormats.value.find(f => f.format === selectedFormat.value) - return format?.resolutions || [] -}) - -const availableFps = computed(() => { - const resolution = availableResolutions.value.find( - r => `${r.width}x${r.height}` === selectedResolution.value - ) - return resolution?.fps || [] -}) - -const selectedFormatInfo = computed(() => - availableFormatOptions.value.find(format => format.format === selectedFormat.value) ?? null -) - -const selectedDeviceInfo = computed(() => - devices.value.find(device => device.path === selectedDevice.value) ?? null -) - -const selectedResolutionInfo = computed(() => - availableResolutions.value.find( - resolution => `${resolution.width}x${resolution.height}` === selectedResolution.value, - ) ?? null -) - const selectedCodecInfo = computed(() => { const codec = availableCodecs.value.find(c => c.id === props.videoMode) return codec || null @@ -458,6 +387,10 @@ function handleDeviceChange(devicePath: unknown) { isDirty.value = true const device = devices.value.find(d => d.path === devicePath) + if (device?.control_mode === 'source_following') { + clearFormatSelection() + return + } const format = device ? findFirstSelectableFormat(device.formats) : undefined if (!format) { clearFormatSelection() @@ -519,13 +452,15 @@ async function applyVideoConfig() { applying.value = true try { - await configStore.updateVideo({ - device: selectedDevice.value, - format: selectedFormat.value, - width, - height, - fps: toConfigFps(selectedFps.value), - }) + await configStore.updateVideo(isSourceFollowing.value + ? { device: selectedDevice.value } + : { + device: selectedDevice.value, + format: selectedFormat.value, + width, + height, + fps: toConfigFps(selectedFps.value ?? 30), + }) isDirty.value = false // Stream state will be updated via WebSocket system.device_info event @@ -781,124 +716,28 @@ watch( - -
- - -
- - -
- - -
- - -
- - -
+ - - - -
-
- - - - - - - {{ t('setup.videoFormatHelp') }} - - -
- -
- -
-
- - -
- -
- - -
-
+

{{ t('setup.noVideoDevices') }}