robokudo.vis.visualizer

Base visualization interface for RoboKudo pipelines.

This module provides the base visualization interface and shared state management for RoboKudo pipeline visualization. It implements:

  • Observer pattern for state updates

  • Shared visualization state

  • Visualizer lifecycle management

  • Pipeline data access

  • Common visualization utilities

Classes

Visualizer

Base class for RoboKudo pipeline visualizers.

Module Contents

class robokudo.vis.visualizer.Visualizer(pipeline: robokudo.pipeline.Pipeline, shared_visualizer_state: typing_extensions.Optional[Visualizer] = None)

Bases: object

Base class for RoboKudo pipeline visualizers.

This class provides the foundation for implementing pipeline visualizers. It includes:

  • Observer pattern for state updates

  • Shared visualization state

  • Visualizer lifecycle management

  • Pipeline data access

  • Common visualization utilities

Do not instantiate this class directly. Use new_visualizer_instance() to properly register visualizers.

class Observable

Base class for objects that can be observed.

This class implements the Observable part of the Observer pattern.

_observers: typing_extensions.List[Visualizer] = []
register_observer(observer: Visualizer) → None

Register an observer to receive notifications.

Parameters:

observer – The observer to register

notify_observers(*args: typing_extensions.Any, **kwargs: typing_extensions.Any) → None

Notify all registered observers.

class Observer

Base class for objects that observe state changes.

This class implements the Observer part of the Observer pattern.

notify(observable: Visualizer, *args: typing_extensions.Any, **kwargs: typing_extensions.Any) → None

Handle notification of state changes.

Parameters:

observable – The object that sent the notification

Raises:

Exception – If not implemented by subclass

class SharedState

Bases: Observable

Shared state for single-view visualizers.

This class manages state shared between visualizers, particularly for single-view visualizers that switch between annotators. It uses the Observer pattern to notify visualizers of state changes.

The Observer pattern is used for state changes, not for new data.

active_annotator: typing_extensions.Optional[robokudo.annotators.core.BaseAnnotator] = None

Currently active annotator.

active_annotator_i: int = 0

Index of currently active annotator

instances: typing_extensions.List[Visualizer] = []

List of all active visualizer instances

pipeline: robokudo.pipeline.Pipeline

Pipeline being visualized

indicate_termination_var: bool = False

Flag indicating if visualization should terminate

shared_visualizer_state: typing_extensions.Optional[Visualizer] = None

Shared state object for coordinating between visualizers

update_output: bool = True

Indicate that the output of this Visualizer needs to be renewed/redrawn

new_data: bool = False

Flag indicating if new data is available

rk_logger: logging.Logger = None

Logger instance

identifier() → str

Get a unique identifier for this visualizer.

pre_tick() → None

Prepare for visualization update.

Called before tick(). Override to implement pre-update logic.

tick() → None

Update the visualization display.

This is the main method for visualizers. Override to implement visualization logic.

post_tick() → None

Clean up after visualization update.

Called after tick(). Override to implement post-update logic.

static static_post_tick() → None

Perform static post-update operations.

This method is called once per visualizer type, regardless of how many instances exist. Override to implement static post-update logic.

classmethod new_visualizer_instance(pipeline: robokudo.pipeline.Pipeline, shared_visualizer_state: typing_extensions.Optional[Visualizer] = None) → Visualizer

Create and register a new visualizer instance.

Parameters:

  • pipeline – Pipeline to visualize

  • shared_visualizer_state – Shared state object for coordinating between visualizers

Returns:

New visualizer instance

static clear_visualizer_instances() → None

Remove all registered visualizer instances.

insert_input() → None

Handle input insertion.

Override to implement input handling logic.

activate_update_output() → None

Mark visualizer for update.

Sets the update flag to trigger a redraw.

new_data_available() → None

Signal that new data is available.

Called by external code to indicate new data is ready for visualization.

indicate_termination() → bool

Check if visualization should terminate.

Returns:

True if visualization should terminate, False otherwise

static get_unique_types_of_visualizer_instances() → typing_extensions.Set[typing_extensions.Type]

Get set of unique visualizer types.

Returns:

Set of visualizer classes that have instances

update_output_flag_for_new_data() → None

Update flags when new data arrives.

Sets update flag and clears new data flag when new data is available.

get_visualized_annotator_outputs_for_pipeline() → robokudo.annotators.outputs.AnnotatorOutputs

Get annotator outputs for visualization.

Returns:

Annotator outputs for the current pipeline

Raises:

AssertionError – If outputs are not of the expected type