added ruff code reivew

This commit is contained in:
ykcheungab
2026-09-27 19:21:55 +08:00
parent 07da372204
commit e2ac36269c
110 changed files with 890 additions and 426 deletions
+47 -25
View File
@@ -3,21 +3,24 @@
from __future__ import annotations
import contextlib
import logging
import os
import re
import sys
from collections.abc import Iterable, Sequence
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
logger = logging.getLogger(__name__)
DEFAULT_MAX_CAMERA_INDEX = MAX_SIMULTANEOUS_CAPTURES
# Common built-in laptop webcam name fragments (case-insensitive).
DEFAULT_LAPTOP_WEBCAM_PATTERNS: Tuple[str, ...] = (
DEFAULT_LAPTOP_WEBCAM_PATTERNS: tuple[str, ...] = (
r"integrated",
r"built[- ]?in",
r"laptop",
@@ -49,8 +52,8 @@ class CameraInfo:
height: int
name: str = ""
path: str = ""
vendor_id: Optional[int] = None
product_id: Optional[int] = None
vendor_id: int | None = None
product_id: int | None = None
bus: str = ""
@property
@@ -64,7 +67,7 @@ class CameraInfo:
return f"{self.index} ({self.width}x{self.height}) [{self.backend_name}]"
@property
def vid_pid(self) -> Optional[Tuple[int, int]]:
def vid_pid(self) -> tuple[int, int] | None:
if self.vendor_id is None or self.product_id is None:
return None
return (int(self.vendor_id), int(self.product_id))
@@ -74,13 +77,13 @@ class CameraInfo:
class CameraFilter:
"""Optional filters for ``list_cameras`` / ``pick_default_camera``."""
name_include: Optional[Sequence[str]] = None
name_exclude: Optional[Sequence[str]] = None
name_include: Sequence[str] | None = None
name_exclude: Sequence[str] | None = 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
vid_pid_allowlist: Sequence[tuple[int, int]] | None = None
min_width: int | None = None
min_height: int | None = None
prefer_named: bool = True
def matches(self, cam: CameraInfo) -> bool:
@@ -143,8 +146,15 @@ def _suppress_opencv_noise():
if set_level is not None and prev_level is not None:
try:
set_level(prev_level)
except Exception:
pass
except Exception as exc:
# Restoring OpenCV's log level is best-effort, but a silent failure
# here leaves probing noise suppressed for the rest of the process,
# which hides exactly the diagnostics a failed enumeration needs.
logger.warning(
"capture discovery: could not restore OpenCV log level %r: %s",
prev_level,
exc,
)
def _linux_device_metadata(index: int) -> dict:
@@ -152,7 +162,7 @@ def _linux_device_metadata(index: int) -> dict:
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:
with open(name_path, encoding="utf-8", errors="ignore") as fh:
meta["name"] = fh.read().strip()
except OSError:
pass
@@ -165,9 +175,9 @@ def _linux_device_metadata(index: int) -> dict:
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:
with open(vend, encoding="utf-8") as fh:
meta["vendor_id"] = int(fh.read().strip(), 16)
with open(prod, "r", encoding="utf-8") as fh:
with open(prod, encoding="utf-8") as fh:
meta["product_id"] = int(fh.read().strip(), 16)
meta["bus"] = "usb"
break
@@ -229,7 +239,7 @@ def _windows_device_names() -> dict[int, dict]:
return {}
def _device_metadata(index: int, windows_names: Optional[dict[int, dict]] = None) -> dict:
def _device_metadata(index: int, windows_names: dict[int, dict] | None = None) -> dict:
if sys.platform.startswith("linux"):
return _linux_device_metadata(index)
if sys.platform == "win32" and windows_names is not None:
@@ -238,7 +248,7 @@ def _device_metadata(index: int, windows_names: Optional[dict[int, dict]] = None
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]:
def _try_open(index: int, backend_name: str, backend_api: int, meta: dict) -> CameraInfo | None:
cap = cv2.VideoCapture(index, backend_api)
if not cap.isOpened():
cap.release()
@@ -272,13 +282,13 @@ def _try_open(index: int, backend_name: str, backend_api: int, meta: dict) -> Op
def list_cameras(
max_index: int = DEFAULT_MAX_CAMERA_INDEX,
*,
camera_filter: Optional[CameraFilter] = None,
camera_filter: CameraFilter | None = 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,
name_include: Sequence[str] | None = None,
name_exclude: Sequence[str] | None = None,
vid_pid_allowlist: Sequence[tuple[int, int]] | None = None,
min_width: int | None = None,
min_height: int | None = None,
backend: str = "auto",
) -> list[CameraInfo]:
"""Probe indices ``0 .. max_index`` and return working cameras.
@@ -302,10 +312,22 @@ def list_cameras(
windows_names = _windows_device_names() if sys.platform == "win32" else None
with _suppress_opencv_noise():
for index in range(0, max_index + 1):
for index in range(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)
try:
info = _try_open(index, backend_name, backend_api, meta)
except Exception as exc:
# One bad backend/device must not abort the whole probe: an
# escaping error here made every *other* camera invisible, not
# just this index. Skip the candidate and keep enumerating.
logger.warning(
"capture discovery: probing index %d via %s failed: %s",
index,
backend_name,
exc,
)
continue
if info is None:
continue
if index in seen_indices: