robokudo.annotators.testing =========================== .. py:module:: robokudo.annotators.testing .. autoapi-nested-parse:: 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 ------- .. autoapisummary:: robokudo.annotators.testing.SlowAnnotator robokudo.annotators.testing.EmptyAnnotator robokudo.annotators.testing.FailingAnnotator robokudo.annotators.testing.FakeCollectionReaderAnnotator robokudo.annotators.testing.ScopedAnnotator Module Contents --------------- .. py:class:: SlowAnnotator(name='SlowAnnotator', sleep_in_s=1) Bases: :py:obj:`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 :ivar sleep_in_s: Duration to sleep in seconds :type sleep_in_s: float .. py:attribute:: sleep_in_s :value: 1 .. py:method:: 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 :return: Success status after completion :rtype: py_trees.Status .. py:class:: EmptyAnnotator(name='EmptyAnnotator', sleep_in_s=1) Bases: :py:obj:`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. :ivar sleep_in_s: Sleep duration in seconds :type sleep_in_s: float .. py:method:: update() Generate empty visual output. Creates a black image and sets it as the annotator output. :return: SUCCESS status :rtype: py_trees.Status .. py:class:: FailingAnnotator(name='FailingAnnotator') Bases: :py:obj:`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. :ivar counter: Counter for alternating between success and failure :type counter: int .. py:attribute:: counter :value: 0 .. py:method:: initialise() Initialize the annotator state. Called on first tick and whenever status changes from non-running. .. py:method:: compute() Simulate processing with alternating success/failure. Sleeps for a fixed duration and alternates between success and failure states based on an internal counter. :return: SUCCESS or FAILURE status :rtype: py_trees.Status .. py:class:: FakeCollectionReaderAnnotator(name='FakeCollectionReader') Bases: :py:obj:`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 :ivar counter: Internal counter for tracking iterations :type counter: int :ivar collection_readers: List of collection reader descriptors :type collection_readers: list .. py:attribute:: collection_readers .. py:method:: setup(timeout) Set up the collection reader. :param timeout: Maximum time to wait for setup completion :type timeout: float :return: True if setup successful :rtype: bool .. py:method:: 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. .. py:method:: 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 :return: Current processing status (RUNNING or SUCCESS) :rtype: py_trees.common.Status .. py:method:: terminate(new_status) Clean up when transitioning to non-running state. :param new_status: New status being transitioned to :type new_status: py_trees.common.Status .. py:class:: ScopedAnnotator(name='ScopedAnnotator', descriptor=Descriptor()) Bases: :py:obj:`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. .. py:class:: Descriptor Bases: :py:obj:`robokudo.annotators.core.BaseAnnotator.Descriptor` Descriptor class defining the annotator's parameters. :ivar parameters: Configuration parameters for the annotator :type parameters: ScopedAnnotator.Descriptor.Parameters .. py:class:: Parameters Parameter class for the ScopedAnnotator. :ivar analysis_scope: List of data types to analyze :type analysis_scope: list .. py:attribute:: analysis_scope .. py:attribute:: parameters .. py:method:: 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 :return: Processing status :rtype: py_trees.Status :raises: Any exceptions are caught and raised to the blackboard