cesar
43cb1ff01c
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.
85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
"""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()
|