cristhian
7628a8515b
validate / validate (push) Successful in 59m14s
CAPTURE_PATH defaulted to index 0, which is a coin flip on a machine with one camera. A UVC camera exposes a second device node for its metadata stream -- /dev/video1 beside /dev/video0 -- and that node opens happily and never yields an image, so the lowest device number is wrong about as often as it is right. With neither CAMERA_ID nor CAPTURE_PATH set, the node now opens each /dev/video* in numeric order and keeps the first that actually delivers a frame. Reading one is the test, because opening is not evidence. Discovery finding nothing falls back to index 0 rather than raising, which keeps the failure in one place: the first read, where CI publishes a placeholder and a normal run raises with a message naming the device. CAMERA_ID is still worth preferring on a robot. Neither an index nor discovery order is stable across reboots or across which port a camera is in; a by-id name is.
124 lines
3.7 KiB
Python
124 lines
3.7 KiB
Python
"""A dora webcam source: captures frames and publishes them as std/media/v1/Image.
|
|
|
|
Each `tick` reads one frame, optionally flips and resizes it, encodes it, and
|
|
sends it. The node stops when the tick input closes, after CAPTURE_FRAMES
|
|
frames if that is set, or when `q` is pressed in the preview window.
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
|
|
import cv2
|
|
from dora import Node
|
|
|
|
from dora_webcam import camera, encode, image, preview
|
|
|
|
FLIPS = {
|
|
"VERTICAL": 0,
|
|
"HORIZONTAL": 1,
|
|
"BOTH": -1,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
"""Open the camera and publish one frame per tick."""
|
|
args = _parse_args()
|
|
|
|
device = camera.resolve(os.getenv("CAMERA_ID"), os.getenv("CAPTURE_PATH"))
|
|
capture = cv2.VideoCapture(device)
|
|
|
|
width = _int("IMAGE_WIDTH")
|
|
height = _int("IMAGE_HEIGHT")
|
|
|
|
# Ask the camera for the size rather than only resizing after the fact: a
|
|
# camera that can deliver it natively avoids a scale per frame.
|
|
if width and height:
|
|
capture.set(cv2.CAP_PROP_FRAME_WIDTH, width)
|
|
capture.set(cv2.CAP_PROP_FRAME_HEIGHT, height)
|
|
|
|
encoding = os.getenv("ENCODING", "bgr8").lower()
|
|
quality = _int("JPEG_QUALITY") or 95
|
|
flip = FLIPS.get(os.getenv("FLIP", "").upper())
|
|
ci = os.getenv("CI") == "true"
|
|
# 0 runs until the tick closes; a positive count is what lets CI run the
|
|
# dataflow to completion instead of killing it on a timeout.
|
|
limit = _int("CAPTURE_FRAMES") or 0
|
|
# Never under CI: there is no display there, and imshow would fail.
|
|
previewing = os.getenv("PREVIEW", "").lower() in ("1", "true") and not ci
|
|
|
|
node = Node(args.name)
|
|
sent = 0
|
|
|
|
try:
|
|
for event in node:
|
|
if event["type"] == "ERROR":
|
|
raise RuntimeError(event["error"])
|
|
|
|
if event["type"] == "INPUT_CLOSED" and event.get("id") == "tick":
|
|
break
|
|
|
|
if event["type"] != "INPUT" or event["id"] != "tick":
|
|
continue
|
|
|
|
frame = camera.read(capture, device, ci)
|
|
|
|
if flip is not None:
|
|
frame = cv2.flip(frame, flip)
|
|
|
|
if width and height and (frame.shape[1], frame.shape[0]) != (width, height):
|
|
frame = cv2.resize(frame, (width, height))
|
|
|
|
# After the flip and the resize, so what is on screen is what is
|
|
# published rather than what came off the sensor.
|
|
if previewing and preview.show(frame):
|
|
break
|
|
|
|
# Taken before encoding: a compressed payload's byte count is not
|
|
# its pixel dimensions.
|
|
frame_height, frame_width = frame.shape[:2]
|
|
|
|
node.send_output(
|
|
"image",
|
|
image.pack(
|
|
encode.encode(frame, encoding, quality),
|
|
encoding,
|
|
frame_width,
|
|
frame_height,
|
|
),
|
|
{},
|
|
)
|
|
|
|
sent += 1
|
|
if limit and sent >= limit:
|
|
break
|
|
finally:
|
|
capture.release()
|
|
|
|
if previewing:
|
|
preview.close()
|
|
|
|
|
|
def _parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Dora webcam source: publishes captured frames as std/media/v1/Image.",
|
|
)
|
|
parser.add_argument(
|
|
"--name",
|
|
type=str,
|
|
default="dora-webcam",
|
|
# Passing the name lets the node also run standalone as a dynamic node;
|
|
# when dora spawns it, the id from the dataflow wins.
|
|
help="The name of the node in the dataflow.",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def _int(name: str) -> int | None:
|
|
"""Read an integer environment variable, treating unset and empty as absent."""
|
|
value = os.getenv(name)
|
|
return int(value) if value else None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|