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.
71 lines
1.7 KiB
Python
71 lines
1.7 KiB
Python
"""Turning a captured BGR frame into the bytes of a chosen encoding."""
|
|
|
|
import cv2
|
|
import numpy as np
|
|
|
|
# Compressed encodings, and the OpenCV file suffix each is written through.
|
|
COMPRESSED = {
|
|
"jpeg": ".jpeg",
|
|
"jpg": ".jpg",
|
|
"jpe": ".jpe",
|
|
"png": ".png",
|
|
"bmp": ".bmp",
|
|
"webp": ".webp",
|
|
}
|
|
|
|
RAW = ("bgr8", "rgb8", "yuv420")
|
|
|
|
SUPPORTED = (*RAW, *COMPRESSED)
|
|
|
|
|
|
def encode(frame: np.ndarray, encoding: str, jpeg_quality: int) -> np.ndarray:
|
|
"""Convert a BGR frame into the bytes of ``encoding``.
|
|
|
|
Parameters
|
|
----------
|
|
frame : np.ndarray
|
|
Captured frame, BGR, as OpenCV reads it.
|
|
encoding : str
|
|
One of ``SUPPORTED``.
|
|
jpeg_quality : int
|
|
Quality 0-100, consulted only for the JPEG encodings.
|
|
|
|
Returns
|
|
-------
|
|
np.ndarray
|
|
The bytes to publish. Its shape is not the image shape for a
|
|
compressed encoding, which is why the dimensions are recorded
|
|
separately.
|
|
|
|
Raises
|
|
------
|
|
RuntimeError
|
|
If ``encoding`` is unsupported, or the encoder fails.
|
|
|
|
"""
|
|
if encoding == "bgr8":
|
|
return frame
|
|
|
|
if encoding == "rgb8":
|
|
return cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
|
|
|
|
if encoding == "yuv420":
|
|
return cv2.cvtColor(frame, cv2.COLOR_BGR2YUV_I420)
|
|
|
|
if encoding in COMPRESSED:
|
|
params = (
|
|
[cv2.IMWRITE_JPEG_QUALITY, jpeg_quality]
|
|
if encoding in ("jpeg", "jpg", "jpe")
|
|
else []
|
|
)
|
|
ok, encoded = cv2.imencode(COMPRESSED[encoding], frame, params)
|
|
|
|
if not ok:
|
|
raise RuntimeError(f"could not encode the frame as {encoding}")
|
|
|
|
return encoded
|
|
|
|
raise RuntimeError(
|
|
f"unsupported encoding: {encoding}; expected one of {', '.join(SUPPORTED)}",
|
|
)
|