orisys-sdk/src/orisys/capture/discovery.py

"""Discover cameras with optional platform metadata and filters."""

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

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, ...] = (
    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: int | None = None
    product_id: int | None = 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) -> 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))


@dataclass(frozen=True)
class CameraFilter:
    """Optional filters for ``list_cameras`` / ``pick_default_camera``."""

    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: 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:
        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 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:
    """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, 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, encoding="utf-8") as fh:
                    meta["vendor_id"] = int(fh.read().strip(), 16)
                with open(prod, 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: 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:
        # 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) -> CameraInfo | None:
    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: CameraFilter | None = None,
    exclude_laptop_webcams: bool = False,
    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.

    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(max_index + 1):
            meta = _device_metadata(index, windows_names)
            for backend_name, backend_api in backends:
                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:
                    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