diff --git a/requirements.txt b/requirements.txt index dbfbc3d..7cb161e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,4 +7,5 @@ numpy>=1.24.4 opencv_python>=4.10.0.84 scipy>=1.10.1 setuptools>=45.2.0 +av diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index 5f4d0b3..5f65bd8 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -146,7 +146,11 @@ class Sensor: self._av_stream = None self._av_decoder = None self.cap = None - self._is_live_camera = isinstance(self.vid_src, int) + # Treat both camera index (0) and V4L2 path ("/dev/video0") as live camera for threading/low-latency path. + self._is_live_camera = ( + isinstance(self.vid_src, int) + or (isinstance(self.vid_src, str) and self.vid_src.startswith("/dev/video")) + ) self._capture_thread = None self._capture_running = False self._latest_frame = None @@ -468,14 +472,18 @@ class Sensor: def _open_video_source(self, override, box): """Initialize video capture and calibration.""" + _is_camera_index = isinstance(self.vid_src, int) + _is_v4l2_path = isinstance(self.vid_src, str) and self.vid_src.startswith("/dev/video") use_linux_camera_pyav = ( platform.system() == "Linux" - and isinstance(self.vid_src, int) + and (_is_camera_index or _is_v4l2_path) and av is not None ) if use_linux_camera_pyav: - device = f"/dev/video{self.vid_src}" + device = self.vid_src if _is_v4l2_path else f"/dev/video{self.vid_src}" + if self.verbose: + print(f"Using PyAV Linux camera backend: {device}") try: self._av_container = av.open( device, @@ -489,8 +497,6 @@ class Sensor: self._av_stream = self._av_container.streams.video[0] self._av_decoder = self._av_container.decode(self._av_stream) self._use_pyav = True - if self.verbose: - print(f"Using PyAV Linux camera backend: {device}") except Exception as exc: if self.verbose: print(f"PyAV camera open failed ({exc}), fallback to OpenCV backend.") diff --git a/test_usb_fps.py b/test_usb_fps.py new file mode 100644 index 0000000..4560170 --- /dev/null +++ b/test_usb_fps.py @@ -0,0 +1,213 @@ +#!/usr/bin/env python3 +""" +USB camera format and FPS tester for Jetson. + +- Lists available formats/resolutions via v4l2-ctl. +- For selected modes, opens the camera with OpenCV and measures per-frame + capture time (ms) to show real delivered FPS/latency. +""" + +import argparse +import subprocess +import sys +import time +import statistics +from typing import List, Tuple + +import cv2 + + +def run_v4l2_list_formats(device: str) -> str: + """Return the raw output of v4l2-ctl --list-formats-ext.""" + try: + out = subprocess.check_output( + ["v4l2-ctl", f"--device={device}", "--list-formats-ext"], + stderr=subprocess.STDOUT, + text=True, + ) + return out + except (subprocess.CalledProcessError, FileNotFoundError) as e: + print(f"[WARN] Failed to run v4l2-ctl: {e}") + return "" + + +def parse_formats(output: str) -> List[Tuple[str, int, int, List[float]]]: + """ + Parse v4l2-ctl --list-formats-ext output. + + Returns list of (fourcc, width, height, fps_list). + """ + lines = output.splitlines() + formats: List[Tuple[str, int, int, List[float]]] = [] + + current_fourcc = None + for line in lines: + line = line.strip() + if line.startswith("[") and "):" in line and "'" in line: + # Example: [0]: 'MJPG' (Motion-JPEG, compressed) + try: + fourcc = line.split("'")[1] + current_fourcc = fourcc + except Exception: + current_fourcc = None + elif line.startswith("Size: Discrete") and current_fourcc: + # Example: Size: Discrete 1280x720 + try: + parts = line.split() + wh = parts[-1] + w_str, h_str = wh.split("x") + width = int(w_str) + height = int(h_str) + formats.append((current_fourcc, width, height, [])) + except Exception: + pass + elif "Interval:" in line and "fps" in line and formats: + # Example: Interval: Discrete 0.033s (30.000 fps) + try: + fps_part = line.split("(")[-1].split("fps")[0] + fps = float(fps_part) + fourcc, w, h, fps_list = formats[-1] + fps_list.append(fps) + formats[-1] = (fourcc, w, h, fps_list) + except Exception: + pass + + return formats + + +def fourcc_to_int(code: str) -> int: + return cv2.VideoWriter_fourcc(*code) + + +def benchmark_mode( + device_index: int, + fourcc: str, + width: int, + height: int, + target_fps: float, + warmup: int = 10, + measure_frames: int = 60, +) -> None: + """Open a specific mode and measure per-frame capture time.""" + print(f"\n=== Testing {fourcc} {width}x{height} @ {target_fps:.1f} fps ===") + + cap = cv2.VideoCapture(device_index, cv2.CAP_V4L2) + if not cap.isOpened(): + print(" [ERROR] Failed to open device with V4L2 backend.") + return + + cap.set(cv2.CAP_PROP_FOURCC, fourcc_to_int(fourcc)) + cap.set(cv2.CAP_PROP_FRAME_WIDTH, width) + cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height) + cap.set(cv2.CAP_PROP_FPS, target_fps) + + # Query negotiated parameters + negotiated_w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + negotiated_h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + negotiated_fps = cap.get(cv2.CAP_PROP_FPS) + negotiated_fourcc_val = int(cap.get(cv2.CAP_PROP_FOURCC)) + negotiated_fourcc = "".join( + chr((negotiated_fourcc_val >> 8 * i) & 0xFF) for i in range(4) + ) + print( + f" Negotiated: {negotiated_w}x{negotiated_h} {negotiated_fourcc} " + f"@ {negotiated_fps:.1f} fps (backend={cap.getBackendName()})" + ) + + # Warm-up frames (to let auto-exposure, buffers, etc. settle) + for i in range(warmup): + ret, _ = cap.read() + if not ret: + print(f" [WARN] Warmup frame {i} failed.") + time.sleep(0.05) + + times_ms: List[float] = [] + for i in range(measure_frames): + t0 = time.perf_counter() + ret, _ = cap.read() + t1 = time.perf_counter() + if not ret: + print(f" [WARN] Capture failed at frame {i}.") + break + times_ms.append((t1 - t0) * 1000.0) + + cap.release() + + if not times_ms: + print(" [ERROR] No successful frames captured.") + return + + mean_ms = statistics.mean(times_ms) + med_ms = statistics.median(times_ms) + min_ms = min(times_ms) + max_ms = max(times_ms) + eff_fps = 1000.0 / mean_ms + + print( + f" Frames measured: {len(times_ms)}\n" + f" Capture time (ms): mean={mean_ms:.2f}, median={med_ms:.2f}, " + f"min={min_ms:.2f}, max={max_ms:.2f}\n" + f" Effective FPS from capture timing: {eff_fps:.2f}" + ) + + +def main() -> None: + parser = argparse.ArgumentParser( + description="List USB camera formats and benchmark capture times." + ) + parser.add_argument( + "--device", + default="/dev/video4", + help="V4L2 device path (default: /dev/video0)", + ) + parser.add_argument( + "--index", + type=int, + default=0, + help="OpenCV device index (default: 0)", + ) + parser.add_argument( + "--test-modes", + nargs="*", + default=["MJPG:640x480:30", "MJPG:1280x720:30"], + help=( + "Modes to benchmark as FOURCC:WxH:FPS. " + "Example: MJPG:640x480:30 YUYV:640x480:30" + ), + ) + args = parser.parse_args() + + print(f"== v4l2-ctl formats for {args.device} ==") + raw = run_v4l2_list_formats(args.device) + if not raw: + print("No v4l2-ctl output; skipping format list.") + else: + print(raw) + parsed = parse_formats(raw) + if parsed: + print("\n== Parsed formats (FOURCC WxH [fps...]) ==") + for fourcc, w, h, fps_list in parsed: + fps_str = ", ".join(f"{f:.1f}" for f in fps_list) if fps_list else "?" + print(f" {fourcc} {w}x{h} [{fps_str}]") + + print("\n== Benchmarking capture time with OpenCV ==") + for mode_str in args.test_modes: + try: + fourcc_s, wh_s, fps_s = mode_str.split(":") + w_s, h_s = wh_s.lower().split("x") + w = int(w_s) + h = int(h_s) + fps = float(fps_s) + except ValueError: + print(f"[WARN] Cannot parse test mode '{mode_str}', expected FOURCC:WxH:FPS") + continue + + benchmark_mode(args.index, fourcc_s, w, h, fps) + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + print("\nInterrupted by user.", file=sys.stderr) +