cristhian
3c0e0f327d
validate / validate (push) Successful in 59m15s
Captures frames from a V4L2 camera and publishes them as std/media/v1/Image: an Arrow struct of width, height, encoding and data, so a receiver needs nothing from the message metadata. For a compressed encoding the recorded dimensions are the image's, not the length of the payload. Based on opencv-video-capture from dora-hub, with three changes. The output is the typed image struct rather than a flat UInt8 buffer plus metadata. Camera selection by id reads /dev/v4l/by-id/ only; the upstream macOS system_profiler and Windows PowerShell enumeration is not carried. CAPTURE_FRAMES bounds a run so a test ends on its own, in place of upstream's ten-second CI cutoff -- a timeout would have made the CI step exit non-zero. CI=true keeps a failed capture from raising, publishing a labelled black frame instead, which is what lets dataflow.yml run on a machine with no camera. Two dataflows, run one at a time, as dora has no cross-dataflow wiring: dataflow.yml is the camera alone and the CI gate; dataflow_view.yml wires it into dora-image-view and is not run by CI, since it fetches that node over the network and needs a tag to exist there.
148 lines
3.6 KiB
Python
148 lines
3.6 KiB
Python
"""Opening a capture device and reading frames from it."""
|
|
|
|
import os
|
|
import platform
|
|
|
|
import cv2
|
|
import numpy as np
|
|
|
|
BY_ID = "/dev/v4l/by-id/"
|
|
|
|
# Size of the frame substituted when a read fails under CI.
|
|
PLACEHOLDER = (480, 640)
|
|
|
|
|
|
def resolve(camera_id: str | None, capture_path: str | int) -> 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
|
|
Device path such as ``/dev/video1``, or an index. A numeric string is
|
|
read as an index.
|
|
|
|
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 isinstance(capture_path, str) and capture_path.isnumeric():
|
|
return int(capture_path)
|
|
|
|
return capture_path
|
|
|
|
|
|
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."
|