streamsim.src.core.simulator module

Multi-Threaded Streaming Simulator

This module provides a general-purpose framework for real-time visualization and analysis of streaming time-series data, featuring a producer-consumer architecture with pluggable components for feature extraction and change point detection.

Architecture:

The simulator employs a dual-thread design: a background worker thread continuously pulls data from a source, derives features, and runs change point detection, pushing results to a thread-safe queue. The main thread consumes queued updates and renders via an animated visualization, ensuring UI responsiveness independent of processing load.

Use Cases:

  • Real-time sensor monitoring with live anomaly visualization

  • Educational demonstrations of signal processing pipelines

Important Dependencies:

  • threading: Background processing thread

  • queue.Queue: Thread-safe communication between producer and consumer

  • collections.deque: Efficient circular buffers for history management

  • matplotlib.animation.FuncAnimation: Animated rendering engine

  • streamsim.src.core.interfaces: Pluggable component interfaces

  • streamsim.src.core.config.PlottingSetup: Visualization configuration

Example

>>> sim = StreamingSimulator(
...    plotting_setup=setup,
...    feature_deriver=deriver,
...    change_point_detector=detector,
...    renderer=renderer,
...    data_source=sinus_source,
...    window_duration_sec=5.0,
...    max_history=5000,
...    interval_ms=100
)
>>> sim.start()
# [Matplotlib window opens with live streaming visualization]

Author: F.Feenstra

class streamsim.src.core.simulator.StreamingSimulator(plotting_setup: PlottingSetup, feature_deriver: StreamingFeatureDeriver, change_point_detector: StreamingChangePointDetector, renderer: StreamingRenderer = None, data_source: Callable[[], Tuple[float, float] | None] = None, window_duration_sec: float = 30.0, interval_ms: int = 50, max_history: int = 2000)[source]

Bases: object

General framework for streaming time-series visualization.

Provides multi-threaded architecture with pluggable components for feature extraction and change point detection.

start() → None[source]

Launches the streaming simulation and enters the main event loop.

This method initializes and starts both the background processing thread and the matplotlib animation engine, coordinating the full simulation pipeline from data ingestion to real-time visualization.

Initialization Steps:

  1. Sets the is_running flag to True, enabling the worker thread.

  2. Spawns a daemon thread executing _processing_loop for continuous data acquisition, feature derivation, and change point detection.

  3. Configures the FuncAnimation engine with the plotting setup, binding _update_plot as the frame callback and _init_plot for initial render state.

  4. Enters the matplotlib main loop via plt.show(), blocking until the window is closed or stop() is called.

Threading Model:

  • Worker thread: Handles all data processing and queue production

  • Main thread: Manages GUI rendering and queue consumption

  • Both threads communicate via a thread-safe Queue instance

Lifecycle:

The simulation runs indefinitely until: - The user closes the plot window - stop() is called programmatically - The data source signals completion (returns None)

stop() → None[source]

Terminates the streaming simulation and releases resources.

This method initiates a clean shutdown sequence for both the background processing thread and the matplotlib animation engine, ensuring all resources are properly released and no orphaned threads remain.

Shutdown Sequence:

  1. Sets is_running to False, signaling the worker thread to exit its processing loop on the next iteration.

  2. Waits for the processing thread to join with a 1-second timeout, allowing it to finish any in-flight operations.

  3. Stops the FuncAnimation event source to halt frame callbacks.

  4. Invokes the renderer’s cleanup routine to release any allocated graphics resources or figure handles.

Thread Safety:

This method is thread-safe and can be called from any thread. It uses the shared is_running flag to coordinate shutdown with the background worker, avoiding race conditions or deadlocks.

Post-Execution State:

After stop() completes: - The processing thread is terminated (or timed out) - The animation engine is halted - Internal buffers retain their last state (call reset() if needed) - The plot window remains open until manually closed

Note

Always call stop() before closing the application to prevent resource leaks or hanging threads. For a fresh start, call reset() on components before invoking start() again.