refractor capture to a module
This commit is contained in:
ykcheungab
@@ -0,0 +1,334 @@
|
||||
"""Discover cameras with optional platform metadata and filters."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Iterable, Optional, Sequence, Tuple
|
||||
|
||||
import cv2
|
||||
|
||||
from .config import MAX_SIMULTANEOUS_CAPTURES
|
||||
from .source import _backend_candidates
|
||||
|
||||
DEFAULT_MAX_CAMERA_INDEX = MAX_SIMULTANEOUS_CAPTURES
|
||||
|
||||
# Common built-in laptop webcam name fragments (case-insensitive).
|
||||
DEFAULT_LAPTOP_WEBCAM_PATTERNS: Tuple[str, ...] = (
|
||||
r"integrated",
|
||||
r"built[- ]?in",
|
||||
r"laptop",
|
||||
r"facetime",
|
||||
r"ir\s*camera",
|
||||
r"rgb\s*camera",
|
||||
r"hd\s*webcam",
|
||||
r"hd\s*camera",
|
||||
r"user[- ]facing",
|
||||
r"front\s*camera",
|
||||
r"dell\s*webcam",
|
||||
r"lenovo\s*easycamera",
|
||||
r"hp\s*(truevision|webcam|hd)",
|
||||
r"microsoft\s*life(cam|studio)",
|
||||
r"realtek",
|
||||
r"chicony",
|
||||
r"sonix",
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CameraInfo:
|
||||
"""Description of a probed camera device."""
|
||||
|
||||
index: int
|
||||
backend_name: str
|
||||
backend_api: int
|
||||
width: int
|
||||
height: int
|
||||
name: str = ""
|
||||
path: str = ""
|
||||
vendor_id: Optional[int] = None
|
||||
product_id: Optional[int] = None
|
||||
bus: str = ""
|
||||
|
||||
@property
|
||||
def source(self) -> str:
|
||||
return str(self.index)
|
||||
|
||||
def label(self) -> str:
|
||||
name = self.name.strip()
|
||||
if name:
|
||||
return f"{self.index}: {name} ({self.width}x{self.height}) [{self.backend_name}]"
|
||||
return f"{self.index} ({self.width}x{self.height}) [{self.backend_name}]"
|
||||
|
||||
@property
|
||||
def vid_pid(self) -> Optional[Tuple[int, int]]:
|
||||
if self.vendor_id is None or self.product_id is None:
|
||||
return None
|
||||
return (int(self.vendor_id), int(self.product_id))
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CameraFilter:
|
||||
"""Optional filters for ``list_cameras`` / ``pick_default_camera``."""
|
||||
|
||||
name_include: Optional[Sequence[str]] = None
|
||||
name_exclude: Optional[Sequence[str]] = None
|
||||
exclude_laptop_webcams: bool = False
|
||||
laptop_patterns: Sequence[str] = field(default_factory=lambda: DEFAULT_LAPTOP_WEBCAM_PATTERNS)
|
||||
vid_pid_allowlist: Optional[Sequence[Tuple[int, int]]] = None
|
||||
min_width: Optional[int] = None
|
||||
min_height: Optional[int] = None
|
||||
prefer_named: bool = True
|
||||
|
||||
def matches(self, cam: CameraInfo) -> bool:
|
||||
name = (cam.name or "").lower()
|
||||
if self.min_width is not None and cam.width < int(self.min_width):
|
||||
return False
|
||||
if self.min_height is not None and cam.height < int(self.min_height):
|
||||
return False
|
||||
if self.vid_pid_allowlist:
|
||||
allowed = {(int(v), int(p)) for v, p in self.vid_pid_allowlist}
|
||||
if cam.vid_pid is None or cam.vid_pid not in allowed:
|
||||
return False
|
||||
if self.name_include:
|
||||
if not any(re.search(pat, name, re.I) for pat in self.name_include):
|
||||
# If name unknown, keep the device unless prefer_named forces drop.
|
||||
if name:
|
||||
return False
|
||||
if self.prefer_named:
|
||||
return False
|
||||
if self.name_exclude:
|
||||
if name and any(re.search(pat, name, re.I) for pat in self.name_exclude):
|
||||
return False
|
||||
if self.exclude_laptop_webcams and name:
|
||||
if any(re.search(pat, name, re.I) for pat in self.laptop_patterns):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _suppress_opencv_noise():
|
||||
"""Hide OpenCV VideoCapture probe warnings (stderr + OpenCV log)."""
|
||||
prev_level = None
|
||||
set_level = None
|
||||
try:
|
||||
logging = getattr(getattr(cv2, "utils", None), "logging", None)
|
||||
if logging is not None:
|
||||
set_level = getattr(logging, "setLogLevel", None)
|
||||
get_level = getattr(logging, "getLogLevel", None)
|
||||
silent = getattr(logging, "LOG_LEVEL_SILENT", 0)
|
||||
if set_level is not None:
|
||||
if get_level is not None:
|
||||
prev_level = get_level()
|
||||
set_level(silent)
|
||||
except Exception:
|
||||
prev_level = None
|
||||
|
||||
devnull = open(os.devnull, "w")
|
||||
try:
|
||||
with contextlib.redirect_stderr(devnull):
|
||||
stderr_fd = sys.stderr.fileno()
|
||||
saved_fd = os.dup(stderr_fd)
|
||||
try:
|
||||
os.dup2(devnull.fileno(), stderr_fd)
|
||||
yield
|
||||
finally:
|
||||
os.dup2(saved_fd, stderr_fd)
|
||||
os.close(saved_fd)
|
||||
finally:
|
||||
devnull.close()
|
||||
if set_level is not None and prev_level is not None:
|
||||
try:
|
||||
set_level(prev_level)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _linux_device_metadata(index: int) -> dict:
|
||||
"""Best-effort V4L2 name / USB VID:PID from sysfs."""
|
||||
meta = {"name": "", "path": f"/dev/video{index}", "vendor_id": None, "product_id": None, "bus": ""}
|
||||
name_path = f"/sys/class/video4linux/video{index}/name"
|
||||
try:
|
||||
with open(name_path, "r", encoding="utf-8", errors="ignore") as fh:
|
||||
meta["name"] = fh.read().strip()
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
# Walk up to a USB device directory containing idVendor/idProduct.
|
||||
device_link = f"/sys/class/video4linux/video{index}/device"
|
||||
try:
|
||||
cur = os.path.realpath(device_link)
|
||||
for _ in range(8):
|
||||
vend = os.path.join(cur, "idVendor")
|
||||
prod = os.path.join(cur, "idProduct")
|
||||
if os.path.isfile(vend) and os.path.isfile(prod):
|
||||
with open(vend, "r", encoding="utf-8") as fh:
|
||||
meta["vendor_id"] = int(fh.read().strip(), 16)
|
||||
with open(prod, "r", encoding="utf-8") as fh:
|
||||
meta["product_id"] = int(fh.read().strip(), 16)
|
||||
meta["bus"] = "usb"
|
||||
break
|
||||
parent = os.path.dirname(cur)
|
||||
if parent == cur:
|
||||
break
|
||||
cur = parent
|
||||
except (OSError, ValueError):
|
||||
pass
|
||||
return meta
|
||||
|
||||
|
||||
def _windows_device_names() -> dict[int, dict]:
|
||||
"""Best-effort mapping of camera index → name / hardware id via PowerShell.
|
||||
|
||||
Index order is not guaranteed to match OpenCV; names are matched later by
|
||||
probing order best-effort. Returns ``{ordinal: meta}`` in enumeration order.
|
||||
"""
|
||||
if sys.platform != "win32":
|
||||
return {}
|
||||
try:
|
||||
import json
|
||||
import subprocess
|
||||
|
||||
# Camera devices from PnP; keep this soft-fail for machines without cameras.
|
||||
cmd = [
|
||||
"powershell",
|
||||
"-NoProfile",
|
||||
"-Command",
|
||||
(
|
||||
"Get-PnpDevice -Class Camera,Image -Status OK -ErrorAction SilentlyContinue | "
|
||||
"Select-Object FriendlyName, InstanceId | ConvertTo-Json -Compress"
|
||||
),
|
||||
]
|
||||
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=5, check=False)
|
||||
if proc.returncode != 0 or not proc.stdout.strip():
|
||||
return {}
|
||||
data = json.loads(proc.stdout)
|
||||
if isinstance(data, dict):
|
||||
data = [data]
|
||||
out: dict[int, dict] = {}
|
||||
for i, item in enumerate(data):
|
||||
name = str(item.get("FriendlyName") or "").strip()
|
||||
instance = str(item.get("InstanceId") or "")
|
||||
vid = pid = None
|
||||
m = re.search(r"VID_([0-9A-Fa-f]{4}).*PID_([0-9A-Fa-f]{4})", instance)
|
||||
if m:
|
||||
vid = int(m.group(1), 16)
|
||||
pid = int(m.group(2), 16)
|
||||
out[i] = {
|
||||
"name": name,
|
||||
"path": "",
|
||||
"vendor_id": vid,
|
||||
"product_id": pid,
|
||||
"bus": "usb" if vid is not None else "",
|
||||
}
|
||||
return out
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
def _device_metadata(index: int, windows_names: Optional[dict[int, dict]] = None) -> dict:
|
||||
if sys.platform.startswith("linux"):
|
||||
return _linux_device_metadata(index)
|
||||
if sys.platform == "win32" and windows_names is not None:
|
||||
# Best-effort: OpenCV index often matches enumeration order, but not always.
|
||||
return windows_names.get(index, {"name": "", "path": "", "vendor_id": None, "product_id": None, "bus": ""})
|
||||
return {"name": "", "path": "", "vendor_id": None, "product_id": None, "bus": ""}
|
||||
|
||||
|
||||
def _try_open(index: int, backend_name: str, backend_api: int, meta: dict) -> Optional[CameraInfo]:
|
||||
cap = cv2.VideoCapture(index, backend_api)
|
||||
if not cap.isOpened():
|
||||
cap.release()
|
||||
return None
|
||||
ok, frame = cap.read()
|
||||
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0)
|
||||
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)
|
||||
if not ok or frame is None:
|
||||
if width <= 0 or height <= 0:
|
||||
cap.release()
|
||||
return None
|
||||
else:
|
||||
h, w = frame.shape[:2]
|
||||
width = width or int(w)
|
||||
height = height or int(h)
|
||||
cap.release()
|
||||
return CameraInfo(
|
||||
index=index,
|
||||
backend_name=backend_name,
|
||||
backend_api=backend_api,
|
||||
width=width,
|
||||
height=height,
|
||||
name=str(meta.get("name") or ""),
|
||||
path=str(meta.get("path") or ""),
|
||||
vendor_id=meta.get("vendor_id"),
|
||||
product_id=meta.get("product_id"),
|
||||
bus=str(meta.get("bus") or ""),
|
||||
)
|
||||
|
||||
|
||||
def list_cameras(
|
||||
max_index: int = DEFAULT_MAX_CAMERA_INDEX,
|
||||
*,
|
||||
camera_filter: Optional[CameraFilter] = None,
|
||||
exclude_laptop_webcams: bool = False,
|
||||
name_include: Optional[Sequence[str]] = None,
|
||||
name_exclude: Optional[Sequence[str]] = None,
|
||||
vid_pid_allowlist: Optional[Sequence[Tuple[int, int]]] = None,
|
||||
min_width: Optional[int] = None,
|
||||
min_height: Optional[int] = None,
|
||||
backend: str = "auto",
|
||||
) -> list[CameraInfo]:
|
||||
"""Probe indices ``0 .. max_index`` and return working cameras.
|
||||
|
||||
Laptop webcam filtering is best-effort and requires device names from the
|
||||
OS. When names are unavailable, devices are kept unless other filters reject
|
||||
them. Pass ``exclude_laptop_webcams=True`` or a ``CameraFilter`` to opt in.
|
||||
"""
|
||||
filt = camera_filter or CameraFilter(
|
||||
name_include=name_include,
|
||||
name_exclude=name_exclude,
|
||||
exclude_laptop_webcams=exclude_laptop_webcams,
|
||||
vid_pid_allowlist=vid_pid_allowlist,
|
||||
min_width=min_width,
|
||||
min_height=min_height,
|
||||
)
|
||||
|
||||
found: list[CameraInfo] = []
|
||||
seen_indices: set[int] = set()
|
||||
backends = _backend_candidates(backend)
|
||||
windows_names = _windows_device_names() if sys.platform == "win32" else None
|
||||
|
||||
with _suppress_opencv_noise():
|
||||
for index in range(0, max_index + 1):
|
||||
meta = _device_metadata(index, windows_names)
|
||||
for backend_name, backend_api in backends:
|
||||
info = _try_open(index, backend_name, backend_api, meta)
|
||||
if info is None:
|
||||
continue
|
||||
if index in seen_indices:
|
||||
break
|
||||
if not filt.matches(info):
|
||||
seen_indices.add(index)
|
||||
break
|
||||
found.append(info)
|
||||
seen_indices.add(index)
|
||||
break
|
||||
return found
|
||||
|
||||
|
||||
def pick_default_camera(
|
||||
cameras: Iterable[CameraInfo],
|
||||
preferred: str | None = None,
|
||||
) -> str | None:
|
||||
cameras = list(cameras)
|
||||
if not cameras:
|
||||
return None
|
||||
preferred = (preferred or "").strip()
|
||||
if preferred:
|
||||
for cam in cameras:
|
||||
if cam.source == preferred:
|
||||
return cam.source
|
||||
return cameras[0].source
|
||||
Reference in New Issue
Block a user