"""Ask each camera what it can actually deliver, without the app in the way. The studio's sensor configs pin the capture at 1920x1080 @ 30 fps (the tile grid is 5x3 tiles of 384x360, so the size is structural), and nothing downstream can raise a frame rate the driver never offered. When a session runs at ~15-17 fps, there are three questions to separate, and this script answers all three: 1. what does the driver *report* for a given request? 2. what does it actually **deliver**? (Measured from 60 reads. On DSHOW the reported figure is a liar: ``CAP_PROP_FPS`` echoes back whatever was set, so a camera that advertises 30 and delivers 15 reports "30.0".) 3. does the delivered rate follow the request? The sweep at the end asks for 60/30/20/15/10 fps at one format. A rate that *halves* with the request means the request path is wrong; a rate that sticks at the same number whatever is asked means that number is the hardware's ceiling. Run it with the venv interpreter, with the app closed (two processes cannot own the same camera):: python apps/studio/tools/camera_probe.py # every camera python apps/studio/tools/camera_probe.py --index 0 # just one python apps/studio/tools/camera_probe.py --samples 120 # longer measurement """ from __future__ import annotations import argparse import sys import time import cv2 #: What the sensor bundles ask for, and the alternatives worth trying when the #: delivered rate falls short of the requested one. REQUESTS: tuple[tuple[str, int, int, float], ...] = ( ("MJPG", 1920, 1080, 30.0), ("MJPG", 1920, 1080, 0.0), # "whatever the driver prefers at this size" ("MJPG", 1280, 720, 30.0), ("YUYV", 1920, 1080, 30.0), ("YUYV", 1280, 720, 30.0), ) BACKENDS: tuple[tuple[str, int], ...] = ( ("DSHOW", cv2.CAP_DSHOW), ("MSMF", cv2.CAP_MSMF), ) #: Frame rates asked for at one fixed format, to see whether the delivered rate #: follows the request or ignores it. SWEEP_RATES: tuple[float, ...] = (60.0, 30.0, 20.0, 15.0, 10.0) #: Sizes asked for at one fixed rate. This is what separates the two remaining #: explanations for a low rate: "the link cannot carry the pixels" (smaller #: frames go faster) from "this camera is a 15 fps device" (the rate does not #: move at all). SWEEP_SIZES: tuple[tuple[int, int], ...] = ( (1920, 1080), (1280, 720), (960, 540), (848, 480), (640, 480), ) SAMPLES = 60 #: A request is "missed" when the measured rate is below this fraction of it. TOLERANCE = 0.9 def _fourcc_code(text: str) -> int: return cv2.VideoWriter_fourcc(*text) def _decode(code: float) -> str: value = int(code) if value <= 0: return "-" letters = "".join(chr((value >> (8 * index)) & 0xFF) for index in range(4)) return letters if letters.isprintable() else f"0x{value:08x}" def probe( index: int, backend_name: str, backend: int, request: tuple[str, int, int, float], *, samples: int = SAMPLES, ) -> dict | None: """Open, ask for one format, and measure what actually comes out.""" fourcc, width, height, fps = request cap = cv2.VideoCapture(index, backend) if not cap.isOpened(): cap.release() return None try: cap.set(cv2.CAP_PROP_FOURCC, _fourcc_code(fourcc)) cap.set(cv2.CAP_PROP_FRAME_WIDTH, float(width)) cap.set(cv2.CAP_PROP_FRAME_HEIGHT, float(height)) if fps > 0: cap.set(cv2.CAP_PROP_FPS, float(fps)) cap.set(cv2.CAP_PROP_FOURCC, _fourcc_code(fourcc)) for _ in range(5): # let the format settle before believing anything cap.grab() started = time.perf_counter() taken = 0 for _ in range(samples): ok, frame = cap.read() if not ok or frame is None: break taken += 1 span = time.perf_counter() - started got_w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) got_h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) reported = cap.get(cv2.CAP_PROP_FPS) got_cc = _decode(cap.get(cv2.CAP_PROP_FOURCC)) finally: cap.release() return { "asked": (fourcc, width, height, fps), "backend": backend_name, "fourcc": got_cc, "width": got_w, "height": got_h, "reported_fps": reported, "measured_fps": (taken / span) if span > 0 else 0.0, "taken": taken, "samples": samples, } def format_row(result: dict) -> str: fourcc, width, height, fps = result["asked"] asked = f"{fourcc} {width}x{height}" + (f"@{fps:g}" if fps else "@auto") measured = result["measured_fps"] missed = "" if fps > 0 and measured < fps * TOLERANCE: missed = " <-- misses the request" return ( f" {asked:<20} -> {result['fourcc']} {result['width']}x{result['height']} " f"reported {result['reported_fps']:5.1f} measured {measured:5.1f} fps " f"({result['taken']}/{result['samples']}) [{result['backend']}]{missed}" ) def sweep(index: int, backend_name: str, backend: int, size: tuple[int, int], *, samples: int = SAMPLES) -> list[dict]: """Ask for a range of rates at one format: does the delivery follow?""" width, height = size rows = [] for rate in SWEEP_RATES: result = probe( index, backend_name, backend, ("MJPG", width, height, rate), samples=samples, ) if result is not None: rows.append(result) return rows def size_sweep(index: int, backend_name: str, backend: int, fps: float = 30.0, *, samples: int = SAMPLES) -> list[dict]: """Ask for one rate at shrinking sizes: bandwidth, or a fixed ceiling? A link that cannot carry the pixels gets faster as the frames get smaller. A camera whose own rate is 15 fps delivers 15 whatever it is asked for. """ rows = [] for width, height in SWEEP_SIZES: result = probe( index, backend_name, backend, ("MJPG", width, height, fps), samples=samples, ) if result is not None: rows.append(result) return rows def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--index", type=int, default=None, help="camera index to probe") parser.add_argument("--max-index", type=int, default=4, help="highest index to try") parser.add_argument("--samples", type=int, default=SAMPLES, help="frames per row") args = parser.parse_args(argv) indices = [args.index] if args.index is not None else list(range(args.max_index + 1)) found = 0 for index in indices: header = False first_size: tuple[int, int] | None = None first_backend: tuple[str, int] | None = None for backend_name, backend in BACKENDS: for request in REQUESTS: result = probe(index, backend_name, backend, request, samples=args.samples) if result is None: continue if not header: print(f"\ncamera {index}:") header = True print(format_row(result)) if first_size is None and request[0] == "MJPG" and request[3] > 0: first_size = (request[1], request[2]) first_backend = (backend_name, backend) if header and first_size is not None and first_backend is not None: print( f"\n sweep at MJPG {first_size[0]}x{first_size[1]} " f"[{first_backend[0]}] -- does the delivered rate follow the request?" ) rows = sweep(index, first_backend[0], first_backend[1], first_size, samples=args.samples) for result in rows: asked = result["asked"][3] measured = result["measured_fps"] verdict = "" if asked > 0 and measured > 0: ratio = measured / asked if ratio < 0.35: verdict = f" -> {ratio:.2f}x the request" elif ratio > 0.9: verdict = " -> follows the request" print( f" asked {asked:5.1f} fps -> measured {measured:5.1f} fps" f"{verdict}" ) rates = [r["measured_fps"] for r in rows if r["measured_fps"] > 0] if rates: print( f" ceiling seen here: {max(rates):.1f} fps " f"(a ceiling that does not move with the request is the " f"hardware's, not the request's)" ) print( f"\n sizes at MJPG@{30:g} fps [{first_backend[0]}] -- bandwidth, " f"or a fixed ceiling?" ) for result in size_sweep( index, first_backend[0], first_backend[1], samples=args.samples ): width, height = result["asked"][1], result["asked"][2] print( f" {width}x{height:<5} -> measured {result['measured_fps']:5.1f} fps" ) print( " smaller frames going faster means the link is the limit; the " "same number at every size means the camera is" ) if header: found += 1 if not found: print("No camera answered on indices 0..%d." % args.max_index) print( "That is expected on a machine with no camera attached. On the machine " "with the sensor, run it with the app closed: two processes cannot own " "the same camera." ) return 1 return 0 if __name__ == "__main__": sys.exit(main())