Create your own Annotator

As described in the Introduction, RoboKudo is a multi-expert approach which employs multiple algorithms to annotate parts of the incoming sensor data. We usually call these components Annotators, since their main purpose is reasoning and annotating new features or information.

Create the Annotator

We will create a very simple Annotator which will simply look for a PointCloud in the CAS and report the size of it.

First, please go to your rk_tutorial package and open the following path:

ROS 2

cd rk_tutorial/annotators

ROS 1

cd src/rk_tutorial/annotators

Here, you can simply create a new file called my_first_annotator.py with the following content:

from timeit import default_timer

from py_trees.common import Status
from robokudo.annotators.core import BaseAnnotator


class MyFirstAnnotator(BaseAnnotator):

    def __init__(self, name: str = "MyFirstAnnotator") -> None:
        """Default construction. Minimal one-time init!"""
        super(MyFirstAnnotator, self).__init__(name)

    def update(self) -> Status:
        start_timer = default_timer()

        cloud = self.get_cas().cloud
        if cloud is None:
            self.feedback_message = "No point cloud available"
            return Status.FAILURE

        self.rk_logger.info(f"Cloud size is: {len(cloud.points)}")

        end_timer = default_timer()
        self.feedback_message = f"Processing took {(end_timer - start_timer):.4f}s"
        return Status.SUCCESS

Let us have a closer look at some key aspects of the Annotator. Every Annotator will inherit from the BaseAnnotator class, which is a child of a normal py_trees Behaviour. The main method for every Behaviour is the update method, which is called everytime the Behaviour is invoked. Please note, that code running in your update method should not exceed a runtime of a couple of milliseconds, to keep the overall Behaviour Tree reactive.

Note

In computer vision, we often need to run methods that exceed this time constraint. For this purpose, we have developed the ThreadedAnnotator which you can use in that case. It is an Annotator which keeps your workload running in a thread. Make sure to put your code into the compute method instead of update to make proper use of that functionality when using the ThreadedAnnotator.

Another key element is access to the CAS, the common data structure for all Annotators in a pipeline. It contains a views dictionary for data such as images, point clouds, and queries, plus an annotations list. Standard view names are defined in CASViews; common views also have properties such as cas.cloud and cas.query. These properties can return None when the data is unavailable.

Now we need to include your new annotator into your pipeline that you’ve defined in the previous tutorial.

Integrate your Annotator into the Pipeline

Open rk_tutorial/rk_tutorial/descriptors/analysis_engines/my_demo.py relative to your ROS 2 workspace’s source directory. On ROS 1, the corresponding path is rk_tutorial/src/rk_tutorial/descriptors/analysis_engines/my_demo.py. Replace the file with the following ROS 2 example:

from rk_tutorial.annotators.my_first_annotator import MyFirstAnnotator
from robokudo.descriptors.factories.cr_descriptor_factory import (
    CollectionReaderDescriptorFactory,
)

from robokudo.analysis_engine import AnalysisEngineInterface
from robokudo.annotators.collection_reader import CollectionReaderAnnotator
from robokudo.annotators.image_preprocessor import ImagePreprocessorAnnotator
from robokudo.annotators.plane import PlaneAnnotator
from robokudo.annotators.pointcloud_cluster_extractor import PointCloudClusterExtractor
from robokudo.annotators.pointcloud_crop import PointcloudCropAnnotator
from robokudo.idioms import pipeline_init
from robokudo.pipeline import Pipeline


class AnalysisEngine(AnalysisEngineInterface):
    def name(self) -> str:
        return "my_demo"

    def implementation(self) -> Pipeline:
        """
        Create a basic pipeline that does tabletop segmentation
        """
        kinect_config = CollectionReaderDescriptorFactory.create_descriptor("kinect_wo_tf")

        seq = Pipeline("RWPipeline")
        seq.add_children(
            [
                pipeline_init(),
                CollectionReaderAnnotator(descriptor=kinect_config),
                ImagePreprocessorAnnotator("ImagePreprocessor"),
                PointcloudCropAnnotator(),
                PlaneAnnotator(),
                PointCloudClusterExtractor(),
                MyFirstAnnotator(),
            ])
        return seq

Restart the my_demo analysis engine using the command from the previous tutorial, with the test bag playing. If you are using a regular ROS 2 colcon install, rebuild and source your workspace after changing Python files. Observe the console where RoboKudo is running. You should see an output like the following:

robokudo  INFO  2022-06-30 19:08:54,052      my_first_annotator.py   in 24 MyFirstAnnotator.update    Cloud size is: 108262

Note: You might have to observe this directly after starting up RoboKudo in the case that tf Messages are spamming the output.