ykcheungab
217 lines
8.0 KiB
Python
217 lines
8.0 KiB
Python
"""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)
|
|
|
|
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 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)"
|
|
)
|
|
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())
|