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: str = 'SlowAnnotator', sleep_in_s: float = 1.0)

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

sleep_in_s = 1.0

Sleep duration in seconds

compute() → py_trees.common.Status

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

class robokudo.annotators.testing.EmptyAnnotator(name: str = 'EmptyAnnotator', sleep_in_s: float = 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.

update() → py_trees.common.Status

Generate empty visual output.

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

Returns:

SUCCESS status

class robokudo.annotators.testing.FailingAnnotator(name: str = '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.

counter = 0

Counter for alternating between success and failure states

initialise() → None

Initialize the annotator state.

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

compute() → py_trees.common.Status

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

class robokudo.annotators.testing.FakeCollectionReaderAnnotator(name: str = '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

collection_readers

List of collection reader descriptors

setup(timeout: float) → bool

Set up the collection reader.

Parameters:

timeout – Maximum time to wait for setup completion

Returns:

True if setup successful

initialise() → None

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() → py_trees.common.Status

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)

terminate(new_status: py_trees.common.Status) → None

Clean up when transitioning to non-running state.

Parameters:

new_status – New status being transitioned to

class robokudo.annotators.testing.ScopedAnnotator(name: str = 'ScopedAnnotator', descriptor: ScopedAnnotator = 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.

class Parameters

Parameter class for the ScopedAnnotator.

analysis_scope

List of data types to analyze

parameters
update() → py_trees.common.Status

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

Raises:

Any exceptions are caught and raised to the blackboard