dora-webcam/dora_webcam/camera.py

"""Opening a capture device and reading frames from it."""

import glob
import os
import platform
import re

import cv2
import numpy as np

BY_ID = "/dev/v4l/by-id/"
DEVICES = "/dev/video*"

# Where discovery gives up, so the failure is reported by the first read with
# the same message an explicitly chosen device would produce.
FALLBACK = 0

# Size of the frame substituted when a read fails under CI.
PLACEHOLDER = (480, 640)


def resolve(camera_id: str | None, capture_path: str | int | None) -> str | int:
    """Decide which device to open.

    Parameters
    ----------
    camera_id : str or None
        Substring of an entry in ``/dev/v4l/by-id/``. A by-id name is stable
        across reboots and across which USB port a camera is in, unlike an
        index, so it takes precedence when set.
    capture_path : str or int or None
        Device path such as ``/dev/video1``, or an index. A numeric string is
        read as an index. When this is unset too, a camera is discovered.

    Returns
    -------
    str or int
        The argument to hand to ``cv2.VideoCapture``.

    Raises
    ------
    RuntimeError
        If ``camera_id`` is set and matches no device.

    """
    if camera_id:
        index = _by_id(camera_id)

        if index is None:
            raise RuntimeError(
                f"no camera matching id `{camera_id}`. "
                f"{_hint()}",
            )

        return index

    if capture_path is not None and capture_path != "":
        if isinstance(capture_path, str) and capture_path.isnumeric():
            return int(capture_path)

        return capture_path

    return discover() or FALLBACK


def discover() -> str | None:
    """Find the first video device that actually delivers a frame.

    Opening is not enough to go on. A UVC camera exposes a second node for its
    metadata stream -- /dev/video1 beside /dev/video0 on a one-camera machine --
    which opens happily and never yields an image, so taking the lowest device
    number picks the wrong one as often as not.

    Returns
    -------
    str or None
        Device path, or None when nothing on this machine captures.

    """
    for path in _devices():
        if _yields_frame(path):
            return path

    return None


def _devices() -> list[str]:
    """List /dev/video* in numeric order, so video2 does not sort before video10."""
    return sorted(glob.glob(DEVICES), key=_device_number)


def _device_number(path: str) -> int:
    match = re.search(r"(\d+)$", path)
    return int(match.group(1)) if match else 0


def _yields_frame(path: str) -> bool:
    """Whether a device both opens and reads one frame."""
    capture = cv2.VideoCapture(path)

    try:
        return bool(capture.isOpened() and capture.read()[0])
    finally:
        capture.release()


def read(capture: cv2.VideoCapture, device: str | int, ci: bool) -> np.ndarray:
    """Read one frame, or a placeholder when running without a camera.

    Parameters
    ----------
    capture : cv2.VideoCapture
        An opened capture device.
    device : str or int
        What was opened, used only in messages.
    ci : bool
        When true, a failed read yields a labelled black frame instead of
        raising. This is what lets the dataflow run on a machine with no
        camera, which is every CI runner.

    Returns
    -------
    np.ndarray
        A BGR frame.

    Raises
    ------
    RuntimeError
        If the read fails and ``ci`` is false.

    """
    ok, frame = capture.read()

    if ok:
        return frame

    if not ci:
        raise RuntimeError(
            f"could not read a frame from `{device}`. To keep a dataflow "
            f"running across a camera disconnect, set `restart_policy: "
            f"on-failure` on this node.",
        )

    return placeholder(f"no frame from {device}")


def placeholder(message: str) -> np.ndarray:
    """Build the black frame that stands in for a camera that did not read.

    Parameters
    ----------
    message : str
        Text drawn on the frame, so a recording shows why it is black.

    Returns
    -------
    np.ndarray
        A BGR frame of size ``PLACEHOLDER``.

    """
    frame = np.zeros((*PLACEHOLDER, 3), dtype=np.uint8)
    cv2.putText(
        frame,
        message,
        (30, 30),
        cv2.FONT_HERSHEY_SIMPLEX,
        0.5,
        (255, 255, 255),
        1,
    )
    return frame


def _by_id(camera_id: str) -> int | None:
    """Find the /dev/videoN index whose by-id name contains `camera_id`."""
    if not os.path.isdir(BY_ID):
        return None

    for entry in sorted(os.listdir(BY_ID)):
        if camera_id.lower() not in entry.lower():
            continue

        target = os.path.realpath(os.path.join(BY_ID, entry))

        if target.startswith("/dev/video"):
            return int(target.removeprefix("/dev/video"))

    return None


def _hint() -> str:
    if platform.system() != "Linux":
        return (
            "Selecting a camera by id is implemented for Linux only; "
            "use CAPTURE_PATH with an index or device path instead."
        )

    return f"Check {BY_ID} for the ids this machine offers."