robokudo.annotators.testing

Testing annotators for RoboKudo.

This module provides annotators for testing and debugging purposes. It supports:

• Empty annotator for baseline testing

• Failing annotator for error handling testing

• Simulated processing delays

• Visual output generation

• Status feedback testing

The module is used for:

• System testing

• Error handling verification

• Performance testing

• Debug visualization

Classes

SlowAnnotator	A slow annotator that demonstrates long-running processing using ThreadedAnnotator.
EmptyAnnotator	Basic annotator that generates empty visual output.
FailingAnnotator	Annotator that simulates failures for testing with a long-running annotator using the ThreadedAnnotator class.
FakeCollectionReaderAnnotator	A simulated collection reader for testing pipeline behavior.
ScopedAnnotator	Demonstrates the usage of analysis scopes in annotators.

Module Contents

class robokudo.annotators.testing.SlowAnnotator(name='SlowAnnotator', sleep_in_s=1)

Bases: robokudo.annotators.core.ThreadedAnnotator

A slow annotator that demonstrates long-running processing using ThreadedAnnotator.

This annotator simulates a time-consuming process by adding a configurable delay and generating visual output. It is useful for:

• Testing thread handling

• Performance monitoring

• Timeout behavior verification

• Visual feedback testing

Variables:

sleep_in_s – Duration to sleep in seconds

sleep_in_s = 1

compute()

Perform the main computation with artificial delay.

This method:

• Retrieves the color image from CAS

• Adds timestamp and visual markers

• Sleeps for configured duration

• Sets feedback message

Returns:

Success status after completion

Return type:

py_trees.Status

class robokudo.annotators.testing.EmptyAnnotator(name='EmptyAnnotator', sleep_in_s=1)

Bases: robokudo.annotators.core.BaseAnnotator

Basic annotator that generates empty visual output.

This annotator creates a black image and sets it as output. Used for testing visualization and output handling.

Variables:

sleep_in_s – Sleep duration in seconds

update()

Generate empty visual output.

Creates a black image and sets it as the annotator output.

Returns:

SUCCESS status

Return type:

py_trees.Status

class robokudo.annotators.testing.FailingAnnotator(name='FailingAnnotator')

Bases: robokudo.annotators.core.ThreadedAnnotator

Annotator that simulates failures for testing with a long-running annotator using the ThreadedAnnotator class.

This annotator alternates between success and failure states, with configurable delays. Used for testing error handling.

Variables:

counter – Counter for alternating between success and failure

counter = 0

initialise()

Initialize the annotator state.

Called on first tick and whenever status changes from non-running.

compute()

Simulate processing with alternating success/failure.

Sleeps for a fixed duration and alternates between success and failure states based on an internal counter.

Returns:

SUCCESS or FAILURE status

Return type:

py_trees.Status

class robokudo.annotators.testing.FakeCollectionReaderAnnotator(name='FakeCollectionReader')

Bases: robokudo.annotators.core.BaseAnnotator

A simulated collection reader for testing pipeline behavior.

This annotator simulates a collection reader by:

• Waiting for a configurable number of iterations

• Creating synthetic CAS data

• Managing feedback messages

• Testing pipeline flow control

Used for:

• Pipeline integration testing

• Flow control verification

• Feedback message handling

• CAS creation testing

Variables:

• counter – Internal counter for tracking iterations

• collection_readers – List of collection reader descriptors

collection_readers

setup(timeout)

Set up the collection reader.

Parameters:

timeout (float) – Maximum time to wait for setup completion

Returns:

True if setup successful

Return type:

bool

initialise()

Initialize the reader state.

Called on first tick and whenever status changes from non-running. Resets counter and clears feedback messages for all children in sequence.

update()

Update the reader state and generate synthetic data.

This method:

• Increments internal counter

• Creates new CAS data after 3 iterations

• Sets appropriate feedback messages

• Updates processing status

Returns:

Current processing status (RUNNING or SUCCESS)

Return type:

py_trees.common.Status

terminate(new_status)

Clean up when transitioning to non-running state.

Parameters:

new_status (py_trees.common.Status) – New status being transitioned to

class robokudo.annotators.testing.ScopedAnnotator(name='ScopedAnnotator', descriptor=Descriptor())

Bases: robokudo.annotators.core.BaseAnnotator

Demonstrates the usage of analysis scopes in annotators.

This annotator shows how to:

• Define and use analysis scopes

• Handle different data types in the scope

• Process object hypotheses and scene data

• Manage scope parameters

The annotator can operate in two modes:

• Object Hypothesis Analysis - processes individual object hypotheses

• Scene Analysis - processes scene-level data

Note

This is primarily a demonstration annotator for educational purposes.

class Descriptor

Bases: robokudo.annotators.core.BaseAnnotator.Descriptor

Descriptor class defining the annotator’s parameters.

Variables:

parameters – Configuration parameters for the annotator

class Parameters

Parameter class for the ScopedAnnotator.

Variables:

analysis_scope – List of data types to analyze

analysis_scope

parameters

update()

Process data based on the current analysis scope.

This method:

• Retrieves data based on the analysis scope

• Determines the processing mode (OH or Scene analysis)

• Processes object hypotheses if present

• Handles scene-level analysis otherwise

Returns:

Processing status

Return type:

py_trees.Status

Raises:

Any exceptions are caught and raised to the blackboard