add opencv-plot node for dora dataflows
validate / validate (push) Successful in 58m52s
validate / test (push) Failing after 58m59s
validate / validate (pull_request) Successful in 58m50s
validate / test (pull_request) Failing after 58m53s

A display sink that overlays bounding boxes and text on an image input and
shows it in an OpenCV window. Based on opencv-plot from dora-hub, brought up
to the dora 1.0 node contract and packaged with uv.

Python rather than Rust: the opencv Rust crate generates its bindings from the
OpenCV headers on the build machine, so it needs libopencv-dev and libclang-dev
per machine and per architecture and does not cross-compile. Every dependency
here ships prebuilt wheels for linux-x86_64, linux-aarch64 and macos-arm64.

Covers every encoding the upstream node did — bgr8, rgb8, jpeg/jpg/jpe/bmp/
webp/png, yuv420 and avif — plus both bbox formats, the text overlay,
PLOT_WIDTH/PLOT_HEIGHT and headless operation under CI=true.
This commit is contained in:
cesar
2026-09-10 18:55:28 -03:00
parent f043beec21
commit 43cb1ff01c
16 changed files with 1412 additions and 0 deletions
+84
View File
@@ -0,0 +1,84 @@
"""A dora display sink: overlays bounding boxes and text on an image input.
An `image` event decodes the frame and immediately redraws. `bbox` and `text`
events only update stored state, which the next `image` event draws.
"""
import argparse
import os
from dora import Node
from opencv_plot import bbox, decode
from opencv_plot.plot import Plot
def main() -> None:
"""Connect to the dataflow and display every frame that arrives."""
args = _parse_args()
plot = Plot(
width=_size("PLOT_WIDTH", args.plot_width),
height=_size("PLOT_HEIGHT", args.plot_height),
headless=os.getenv("CI") == "true",
)
# Passing the name lets the node also run standalone as a dynamic node; when
# dora spawns it, the id from the dataflow wins.
node = Node(args.name)
for event in node:
if event["type"] == "ERROR":
raise RuntimeError(event["error"])
if event["type"] != "INPUT":
continue
event_id = event["id"]
metadata = event["metadata"]
if event_id == "image":
frame = decode.to_bgr(
event["value"],
metadata.get("encoding", "bgr8").lower(),
metadata.get("width"),
metadata.get("height"),
)
if plot.show(frame):
break
elif event_id == "bbox":
plot.bboxes = bbox.parse(event["value"], metadata.get("format", "xyxy").lower())
elif event_id == "text":
plot.text = event["value"][0].as_py()
else:
print(f"ignoring unexpected input `{event_id}`")
def _parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="OpenCV plotter: draws text and bounding boxes on an image.",
)
parser.add_argument(
"--name",
default="opencv-plot",
help="The name of the node in the dataflow, used when running as a dynamic node.",
)
parser.add_argument(
"--plot-width",
type=int,
help="Resize the displayed frame to this width. Overridden by PLOT_WIDTH.",
)
parser.add_argument(
"--plot-height",
type=int,
help="Resize the displayed frame to this height. Overridden by PLOT_HEIGHT.",
)
return parser.parse_args()
def _size(name: str, fallback: int | None) -> int | None:
value = os.getenv(name)
return int(value) if value and value.isnumeric() else fallback
if __name__ == "__main__":
main()