bob_moondream.moondream_node

Classes

`ImageCache`
`MoondreamNode`

Functions

main([args])

Module Contents

class bob_moondream.moondream_node.ImageCache(node)

node

latest_msg = None

latest_image = None

latest_cv_image = None

latest_image_time = 0

lock

update(image_msg)

cv_image()

pil_image()

get()

class bob_moondream.moondream_node.MoondreamNode

Bases: rclpy.node.Node

callback_group

model_name

model_revision

cache_dir

frame_id

device

model

bridge

image_cache

subscription

detection_publisher

point_publisher

annotated_image_publisher

visual_query_result_publisher

object_detection_result_publisher

pointing_result_publisher

visual_query_prompt_subscriber

object_detection_prompt_subscriber

pointing_prompt_subscriber

caption_service

visual_query_service

object_detection_service

pointing_service

image_callback(msg)

Callback function for the image subscriber.

Args:: msg (sensor_msgs.msg.Image): The ROS Image message received from the image topic.

call_service(service, prompt): Wrapper to call one of the VisualQuery services.

visual_query_prompt_callback(msg): Callback for topic-based visual query requests.

object_detection_prompt_callback(msg): Callback for topic-based object detection requests.

pointing_prompt_callback(msg): Callback for topic-based pointing requests.

detect()

Performs object detection and publishes results if subscribers are present.

This method is intended to be called periodically. It checks for active subscribers on the detection topics. If any exist, it calls the model to find all objects in the provided image and then uses publish_detections to process and publish the results.

publish_detections(objects, image)

Processes and publishes object detection results.

This function converts the raw detection data from the model into a vision_msgs/Detection2DArray message and publishes it. It also draws the bounding boxes on a copy of the input image.

Args:: objects (list): A list of dictionaries, where each dictionary represents a detected object. image (numpy.ndarray): The OpenCV image on which to draw the detections.
Returns:: bool: True if publishing was successful, False otherwise.

point()

Performs pointing and publishes results if subscribers are present.

This method is intended to be called periodically. It checks for active subscribers on the pointing topics. If any exist, it calls the model to find all points in the provided image based on the ‘prompt_point’ parameter and then uses publish_points to process and publish the results.

publish_points(points, image)

Processes and publishes pointing results.

This function converts the raw point data from the model into a vision_msgs/Detection2DArray message (where each point is a zero-size bbox) and publishes it. It also draws markers on a copy of the input image.

Args:: points (list): A list of dictionaries, where each dict represents a point (e.g., {‘x’: 0.5, ‘y’: 0.5}). image (numpy.ndarray): The OpenCV image on which to draw the points.
Returns:: bool: True if publishing was successful, False otherwise.

check_image_availability()

Verifies that a recent image is available for processing.

Checks if an image has been received and if it is not too old (older than 2 seconds).

Returns:: bool: True if a recent image is available, False otherwise.

caption_callback(request, response)

ROS service callback to generate a caption for the latest image.

Args:: request: The service request object (prompt is ignored). response: The service response object to be populated.
Returns:: The populated service response object.

visual_query_callback(request, response)

ROS service callback to answer a question about the latest image.

Args:: request: The service request object containing the text prompt (question). response: The service response object to be populated.
Returns:: The populated service response object.

object_detection_callback(request, response)

ROS service callback to detect objects in the latest image.

The results are returned as a JSON string in the service response. If there are subscribers to the detection topics, it also calls publish_detections.

Args:: request: The service request object containing the prompt describing the object. response: The service response object to be populated.
Returns:: The populated service response object.

pointing_callback(request, response)

ROS service callback to find specific points in the latest image.

Args:: request: The service request object containing the prompt describing the point. response: The service response object to be populated with a JSON string of points.
Returns:: The populated service response object.

bob_moondream.moondream_node.main(args=None)