robokudo.vis.ros_visualizer

ROS-based visualization for RoboKudo pipelines.

This module provides ROS-based visualization capabilities for RoboKudo pipelines. It handles:

• Image publishing to ROS topics

• Single and multi-view visualization

• Pipeline state visualization

• Dynamic topic management

• Image format conversion

Classes

SharedROSVisualizer	A single-view ROS Image Publisher. It publishes the active annotator from the SharedState.
AllAnnotatorROSVisualizer	A ROS Image Publisher that publishes the output of all images in the given Pipeline.

Module Contents

class robokudo.vis.ros_visualizer.SharedROSVisualizer(*args: typing_extensions.Any, **kwargs: typing_extensions.Any)

Bases: robokudo.vis.visualizer.Visualizer, robokudo.vis.visualizer.Visualizer.Observer, rclpy.node.Node

A single-view ROS Image Publisher. It publishes the active annotator from the SharedState.

ros_image_publisher: rclpy.publisher.Publisher

Publisher for the image topic

ros_image_cv_bridge: cv_bridge.CvBridge

Bridge for converting between ROS and OpenCV image formats

tick() → None

Update the visualization display.

This method:

• Gets current annotator outputs

• Updates display if needed

• Renders annotator name overlay

• Publishes image to ROS topic

notify(observable: robokudo.vis.visualizer.Visualizer.Observable, *args: typing_extensions.Any, **kwargs: typing_extensions.Any) → None

Handle notification of state changes.

Parameters:

observable – The object that sent the notification

class robokudo.vis.ros_visualizer.AllAnnotatorROSVisualizer(*args: typing_extensions.Any, **kwargs: typing_extensions.Any)

Bases: robokudo.vis.visualizer.Visualizer, rclpy.node.Node

A ROS Image Publisher that publishes the output of all images in the given Pipeline.

This class provides visualization of all annotator outputs in a pipeline through separate ROS image topics. It supports:

• Multiple image topic publishing

• Dynamic topic creation

• Image format conversion

• Per-annotator output streams

Note

This Visualizer works with a shared state and needs notifications

ros_image_publishers: typing_extensions.Dict[str, rclpy.publisher.Publisher]

Mapping of annotator names to ROS publishers

ros_image_cv_bridge: cv_bridge.CvBridge

Bridge for converting between ROS and OpenCV image formats

update_ros_image_publishers() → None

Update ROS publishers for all annotators.

This method:

• Gets current list of annotators

• Creates publishers for new annotators

• Maintains existing publishers

TODO: Consider removing publishers for non-existing annotators. May not be

worth the cost if re-instantiation is frequent due to changing annotator lists.

tick() → None

Update all visualization displays.

This method:

• Gets current annotator outputs

• Updates publishers if needed

• Publishes all annotator images to ROS topics

notify(observable: robokudo.vis.visualizer.Visualizer.Observable, *args: typing_extensions.Any, **kwargs: typing_extensions.Any) → None

Handle notification of state changes.

Parameters:

observable – The object that sent the notification