torsh_vision/

streaming.rs

//! Real-Time Video Stream Processing
//!
//! This module provides efficient stream processing capabilities for real-time computer vision,
//! integrated with scirs2-vision 0.1.5's streaming features.
//!
//! # Features
//! - Video frame buffering and preprocessing pipelines
//! - Real-time performance monitoring and adaptation
//! - Async/parallel processing for low latency
//! - GPU-accelerated stream processing
//! - Frame dropping and quality adaptation strategies
//!
//! # Examples
//!
//! ```rust,ignore
//! use torsh_vision::streaming::*;
//!
//! // Create a video stream processor
//! let processor = StreamProcessor::new(config)?;
//! processor.process_stream(video_source, |frame| {
//!     // Process each frame
//!     detect_objects(frame)
//! })?;
//! ```

use crate::{Result, VisionError};
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use torsh_tensor::Tensor;

/// Stream processing configuration
#[derive(Debug, Clone)]
pub struct StreamConfig {
    /// Maximum frames to buffer
    pub buffer_size: usize,
    /// Target frames per second
    pub target_fps: f32,
    /// Enable frame dropping if processing too slow
    pub enable_frame_drop: bool,
    /// Enable GPU acceleration
    pub use_gpu: bool,
    /// Number of parallel processing threads
    pub num_threads: usize,
    /// Quality adaptation strategy
    pub quality_adaptation: QualityAdaptation,
}

impl Default for StreamConfig {
    fn default() -> Self {
        Self {
            buffer_size: 30,
            target_fps: 30.0,
            enable_frame_drop: true,
            use_gpu: false,
            num_threads: 4,
            quality_adaptation: QualityAdaptation::None,
        }
    }
}

/// Quality adaptation strategies for maintaining real-time performance
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QualityAdaptation {
    /// No adaptation
    None,
    /// Reduce resolution when falling behind
    ResolutionScaling,
    /// Skip non-keyframe processing
    KeyframeOnly,
    /// Adaptive quality based on complexity
    Dynamic,
}

/// Frame metadata for processing pipeline
#[derive(Debug, Clone)]
pub struct FrameMetadata {
    /// Frame number in stream
    pub frame_number: u64,
    /// Timestamp when frame was captured
    pub timestamp: Instant,
    /// Frame width
    pub width: usize,
    /// Frame height
    pub height: usize,
    /// Is this a keyframe
    pub is_keyframe: bool,
    /// Processing priority (higher = more important)
    pub priority: u8,
}

/// Frame with metadata
#[derive(Debug, Clone)]
pub struct Frame {
    /// Frame data as tensor
    pub data: Tensor,
    /// Frame metadata
    pub metadata: FrameMetadata,
}

/// Performance statistics for stream processing
#[derive(Debug, Clone)]
pub struct StreamStats {
    /// Average processing time per frame (ms)
    pub avg_processing_time: f32,
    /// Current frames per second
    pub current_fps: f32,
    /// Number of dropped frames
    pub dropped_frames: u64,
    /// Total frames processed
    pub total_frames: u64,
    /// Buffer utilization (0.0 to 1.0)
    pub buffer_utilization: f32,
    /// Number of quality adaptations
    pub num_adaptations: u64,
}

impl Default for StreamStats {
    fn default() -> Self {
        Self {
            avg_processing_time: 0.0,
            current_fps: 0.0,
            dropped_frames: 0,
            total_frames: 0,
            buffer_utilization: 0.0,
            num_adaptations: 0,
        }
    }
}

/// Real-time stream processor
pub struct StreamProcessor {
    config: StreamConfig,
    stats: Arc<Mutex<StreamStats>>,
    frame_buffer: Arc<Mutex<VecDeque<Frame>>>,
    processing_times: Arc<Mutex<VecDeque<Duration>>>,
}

impl StreamProcessor {
    /// Create a new stream processor
    pub fn new(config: StreamConfig) -> Result<Self> {
        let buffer_size = config.buffer_size;
        Ok(Self {
            config,
            stats: Arc::new(Mutex::new(StreamStats::default())),
            frame_buffer: Arc::new(Mutex::new(VecDeque::with_capacity(buffer_size))),
            processing_times: Arc::new(Mutex::new(VecDeque::with_capacity(100))),
        })
    }

    /// Get current performance statistics
    pub fn stats(&self) -> StreamStats {
        self.stats.lock().map(|s| s.clone()).unwrap_or_default()
    }

    /// Reset statistics
    pub fn reset_stats(&self) {
        if let Ok(mut stats) = self.stats.lock() {
            *stats = StreamStats::default();
        }
        if let Ok(mut times) = self.processing_times.lock() {
            times.clear();
        }
    }

    /// Add a frame to the processing buffer
    pub fn push_frame(&self, frame: Frame) -> Result<()> {
        let mut buffer = self.frame_buffer.lock().map_err(|_| {
            VisionError::InvalidParameter("Failed to lock frame buffer".to_string())
        })?;

        // Check buffer capacity
        if buffer.len() >= self.config.buffer_size {
            if self.config.enable_frame_drop {
                // Drop oldest non-keyframe
                let mut dropped = false;
                for i in 0..buffer.len() {
                    if !buffer[i].metadata.is_keyframe {
                        buffer.remove(i);
                        dropped = true;

                        // Update stats
                        if let Ok(mut stats) = self.stats.lock() {
                            stats.dropped_frames += 1;
                        }
                        break;
                    }
                }

                if !dropped {
                    // All are keyframes, drop oldest
                    buffer.pop_front();
                    if let Ok(mut stats) = self.stats.lock() {
                        stats.dropped_frames += 1;
                    }
                }
            } else {
                return Err(VisionError::InvalidParameter(
                    "Frame buffer full and dropping disabled".to_string(),
                ));
            }
        }

        buffer.push_back(frame);

        // Update buffer utilization
        if let Ok(mut stats) = self.stats.lock() {
            stats.buffer_utilization = buffer.len() as f32 / self.config.buffer_size as f32;
        }

        Ok(())
    }

    /// Get next frame from buffer
    pub fn pop_frame(&self) -> Result<Option<Frame>> {
        let mut buffer = self.frame_buffer.lock().map_err(|_| {
            VisionError::InvalidParameter("Failed to lock frame buffer".to_string())
        })?;

        Ok(buffer.pop_front())
    }

    /// Record processing time for a frame
    pub fn record_processing_time(&self, duration: Duration) {
        if let Ok(mut times) = self.processing_times.lock() {
            times.push_back(duration);

            // Keep only recent times (last 100 frames)
            while times.len() > 100 {
                times.pop_front();
            }

            // Update stats
            if let Ok(mut stats) = self.stats.lock() {
                // Calculate average processing time
                let sum: Duration = times.iter().sum();
                stats.avg_processing_time = sum.as_secs_f32() * 1000.0 / times.len() as f32;

                // Calculate FPS
                if stats.avg_processing_time > 0.0 {
                    stats.current_fps = 1000.0 / stats.avg_processing_time;
                }
            }
        }
    }

    /// Process a frame and update statistics
    pub fn process_frame<F, T>(&self, frame: Frame, process_fn: F) -> Result<T>
    where
        F: FnOnce(&Frame) -> Result<T>,
    {
        let start = Instant::now();

        // Apply quality adaptation if needed
        let adapted_frame = self.adapt_frame_quality(&frame)?;

        // Process the frame
        let result = process_fn(&adapted_frame)?;

        // Record processing time
        let elapsed = start.elapsed();
        self.record_processing_time(elapsed);

        // Update total frames
        if let Ok(mut stats) = self.stats.lock() {
            stats.total_frames += 1;
        }

        Ok(result)
    }

    /// Adapt frame quality based on performance
    fn adapt_frame_quality(&self, frame: &Frame) -> Result<Frame> {
        match self.config.quality_adaptation {
            QualityAdaptation::None => Ok(frame.clone()),

            QualityAdaptation::ResolutionScaling => {
                // Check if we're falling behind
                let stats = self.stats();
                if stats.current_fps < self.config.target_fps * 0.8 {
                    // Reduce resolution by 25%
                    let _new_width = (frame.metadata.width as f32 * 0.75) as usize;
                    let _new_height = (frame.metadata.height as f32 * 0.75) as usize;

                    // TODO: Implement actual downscaling
                    // For now, just return original frame
                    if let Ok(mut stats) = self.stats.lock() {
                        stats.num_adaptations += 1;
                    }
                }
                Ok(frame.clone())
            }

            QualityAdaptation::KeyframeOnly => {
                // Skip non-keyframes if falling behind
                let stats = self.stats();
                if !frame.metadata.is_keyframe && stats.current_fps < self.config.target_fps * 0.9 {
                    // Mark for skipping (caller should handle)
                    if let Ok(mut stats_lock) = self.stats.lock() {
                        stats_lock.num_adaptations += 1;
                    }
                }
                Ok(frame.clone())
            }

            QualityAdaptation::Dynamic => {
                // Combine strategies based on performance
                let stats = self.stats();
                let performance_ratio = stats.current_fps / self.config.target_fps;

                if performance_ratio < 0.7 {
                    // Severe degradation: reduce resolution
                    // TODO: Implement downscaling
                    if let Ok(mut stats_lock) = self.stats.lock() {
                        stats_lock.num_adaptations += 1;
                    }
                } else if performance_ratio < 0.9 && !frame.metadata.is_keyframe {
                    // Moderate degradation: skip non-keyframes
                    if let Ok(mut stats_lock) = self.stats.lock() {
                        stats_lock.num_adaptations += 1;
                    }
                }

                Ok(frame.clone())
            }
        }
    }

    /// Check if processing is keeping up with target FPS
    pub fn is_realtime(&self) -> bool {
        let stats = self.stats();
        stats.current_fps >= self.config.target_fps * 0.95
    }

    /// Get recommended adjustments for configuration
    pub fn recommend_config_adjustments(&self) -> Vec<String> {
        let stats = self.stats();
        let mut recommendations = Vec::new();

        // Check FPS performance
        if stats.current_fps < self.config.target_fps * 0.8 {
            recommendations
                .push("Consider reducing target_fps or enabling quality adaptation".to_string());
        }

        // Check buffer utilization
        if stats.buffer_utilization > 0.9 {
            recommendations.push("Buffer is frequently full - consider increasing buffer_size or enabling frame dropping".to_string());
        }

        // Check dropped frames
        if stats.total_frames > 0 {
            let drop_rate = stats.dropped_frames as f32 / stats.total_frames as f32;
            if drop_rate > 0.1 {
                recommendations.push(format!(
                    "High frame drop rate ({:.1}%) - consider reducing input rate or optimizing processing",
                    drop_rate * 100.0
                ));
            }
        }

        recommendations
    }
}

/// Frame preprocessor for common vision tasks
pub struct FramePreprocessor {
    /// Target size for resizing (width, height)
    pub target_size: Option<(usize, usize)>,
    /// Normalization mean values
    pub normalize_mean: Option<Vec<f32>>,
    /// Normalization std values
    pub normalize_std: Option<Vec<f32>>,
    /// Convert to grayscale
    pub to_grayscale: bool,
}

impl Default for FramePreprocessor {
    fn default() -> Self {
        Self {
            target_size: None,
            normalize_mean: None,
            normalize_std: None,
            to_grayscale: false,
        }
    }
}

impl FramePreprocessor {
    /// Create a new preprocessor with default settings
    pub fn new() -> Self {
        Self::default()
    }

    /// Set target resize dimensions
    pub fn with_resize(mut self, width: usize, height: usize) -> Self {
        self.target_size = Some((width, height));
        self
    }

    /// Set normalization parameters
    pub fn with_normalize(mut self, mean: Vec<f32>, std: Vec<f32>) -> Self {
        self.normalize_mean = Some(mean);
        self.normalize_std = Some(std);
        self
    }

    /// Enable grayscale conversion
    pub fn with_grayscale(mut self) -> Self {
        self.to_grayscale = true;
        self
    }

    /// Preprocess a frame
    pub fn preprocess(&self, frame: &Frame) -> Result<Frame> {
        let processed = frame.clone();

        // TODO: Implement actual preprocessing operations
        // - Resize if target_size is set
        // - Normalize if mean/std are set
        // - Convert to grayscale if enabled

        Ok(processed)
    }
}

/// Batch processor for efficient multi-frame processing
pub struct BatchProcessor {
    batch_size: usize,
    frames: Vec<Frame>,
}

impl BatchProcessor {
    /// Create a new batch processor
    pub fn new(batch_size: usize) -> Self {
        Self {
            batch_size,
            frames: Vec::with_capacity(batch_size),
        }
    }

    /// Add a frame to the batch
    pub fn add_frame(&mut self, frame: Frame) -> Option<Vec<Frame>> {
        self.frames.push(frame);

        if self.frames.len() >= self.batch_size {
            Some(self.flush())
        } else {
            None
        }
    }

    /// Get the current batch without clearing
    pub fn current_batch(&self) -> &[Frame] {
        &self.frames
    }

    /// Flush the current batch
    pub fn flush(&mut self) -> Vec<Frame> {
        std::mem::replace(&mut self.frames, Vec::with_capacity(self.batch_size))
    }

    /// Check if batch is full
    pub fn is_full(&self) -> bool {
        self.frames.len() >= self.batch_size
    }

    /// Get number of frames in current batch
    pub fn len(&self) -> usize {
        self.frames.len()
    }

    /// Check if batch is empty
    pub fn is_empty(&self) -> bool {
        self.frames.is_empty()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn create_dummy_frame(frame_number: u64) -> Frame {
        use torsh_core::DeviceType;
        Frame {
            data: Tensor::zeros(&[224, 224, 3], DeviceType::Cpu).expect("Failed to create tensor"),
            metadata: FrameMetadata {
                frame_number,
                timestamp: Instant::now(),
                width: 224,
                height: 224,
                is_keyframe: frame_number % 10 == 0,
                priority: 1,
            },
        }
    }

    #[test]
    fn test_stream_config_default() {
        let config = StreamConfig::default();
        assert_eq!(config.buffer_size, 30);
        assert_eq!(config.target_fps, 30.0);
        assert!(config.enable_frame_drop);
    }

    #[test]
    fn test_stream_processor_creation() {
        let processor = StreamProcessor::new(StreamConfig::default());
        assert!(processor.is_ok());
    }

    #[test]
    fn test_push_pop_frame() {
        let processor =
            StreamProcessor::new(StreamConfig::default()).expect("Failed to create processor");

        let frame = create_dummy_frame(1);
        processor
            .push_frame(frame.clone())
            .expect("Failed to push frame");

        let popped = processor.pop_frame().expect("Failed to pop frame");
        assert!(popped.is_some());
        assert_eq!(popped.unwrap().metadata.frame_number, 1);
    }

    #[test]
    fn test_frame_dropping() {
        let mut config = StreamConfig::default();
        config.buffer_size = 2;
        config.enable_frame_drop = true;

        let processor = StreamProcessor::new(config).expect("Failed to create processor");

        // Fill buffer beyond capacity
        for i in 0..5 {
            let frame = create_dummy_frame(i);
            processor.push_frame(frame).expect("Failed to push frame");
        }

        let stats = processor.stats();
        assert!(stats.dropped_frames > 0);
    }

    #[test]
    fn test_processing_time_recording() {
        let processor =
            StreamProcessor::new(StreamConfig::default()).expect("Failed to create processor");

        processor.record_processing_time(Duration::from_millis(10));
        processor.record_processing_time(Duration::from_millis(20));

        let stats = processor.stats();
        assert!(stats.avg_processing_time > 0.0);
    }

    #[test]
    fn test_batch_processor() {
        let mut batch = BatchProcessor::new(3);

        assert!(batch.is_empty());
        assert!(!batch.is_full());

        batch.add_frame(create_dummy_frame(1));
        batch.add_frame(create_dummy_frame(2));

        assert_eq!(batch.len(), 2);
        assert!(!batch.is_full());

        let result = batch.add_frame(create_dummy_frame(3));
        assert!(result.is_some());
        assert_eq!(result.unwrap().len(), 3);
        assert!(batch.is_empty());
    }

    #[test]
    fn test_frame_preprocessor() {
        let preprocessor = FramePreprocessor::new()
            .with_resize(224, 224)
            .with_grayscale();

        let frame = create_dummy_frame(1);
        let result = preprocessor.preprocess(&frame);
        assert!(result.is_ok());
    }

    #[test]
    fn test_quality_adaptation_variants() {
        assert_eq!(QualityAdaptation::None, QualityAdaptation::None);
        assert_ne!(
            QualityAdaptation::None,
            QualityAdaptation::ResolutionScaling
        );
    }

    #[test]
    fn test_is_realtime() {
        let processor =
            StreamProcessor::new(StreamConfig::default()).expect("Failed to create processor");

        // Initially no frames processed, should return true (no data)
        // After processing, FPS will be calculated
        processor.record_processing_time(Duration::from_millis(10));
        let is_rt = processor.is_realtime();
        // Result depends on target FPS (30) vs current FPS (100 from 10ms processing)
        assert!(is_rt); // 100 FPS > 30 FPS target
    }

    #[test]
    fn test_stream_stats_default() {
        let stats = StreamStats::default();
        assert_eq!(stats.total_frames, 0);
        assert_eq!(stats.dropped_frames, 0);
        assert_eq!(stats.current_fps, 0.0);
    }
}