voirs_recognizer/preprocessing/

mod.rs

//! Audio preprocessing and enhancement module
//!
//! This module provides real-time audio preprocessing and enhancement capabilities
//! for the `VoiRS` recognition system. It includes:
//!
//! - Real-time noise suppression
//! - Automatic gain control (AGC)
//! - Echo cancellation
//! - Bandwidth extension
//!
//! # Features
//!
//! ## Noise Suppression
//! Implements spectral subtraction and Wiener filtering for noise reduction
//! in real-time audio streams.
//!
//! ## Automatic Gain Control
//! Provides adaptive volume control to maintain consistent signal levels
//! across different audio sources.
//!
//! ## Echo Cancellation
//! Removes echo and feedback from audio signals using adaptive filtering
//! techniques.
//!
//! ## Bandwidth Extension
//! Extends the frequency range of audio signals to improve recognition
//! accuracy for band-limited audio sources.

use crate::RecognitionError;
use std::sync::{Arc, Mutex};
use voirs_sdk::AudioBuffer;

// Sub-modules
pub mod adaptive_algorithms;
pub mod advanced_spectral;
pub mod agc;
pub mod bandwidth_extension;
pub mod echo_cancellation;
pub mod noise_suppression;
pub mod optimizations;
pub mod optimized_preprocessor;
pub mod realtime_features;

// Re-exports
pub use adaptive_algorithms::*;
pub use advanced_spectral::*;
pub use agc::*;
pub use bandwidth_extension::*;
pub use echo_cancellation::*;
pub use noise_suppression::*;
pub use optimizations::*;
pub use optimized_preprocessor::*;
pub use realtime_features::*;

/// Audio preprocessing configuration
#[derive(Debug, Clone)]
pub struct AudioPreprocessingConfig {
    /// Enable noise suppression
    pub noise_suppression: bool,
    /// Enable automatic gain control
    pub agc: bool,
    /// Enable echo cancellation
    pub echo_cancellation: bool,
    /// Enable bandwidth extension
    pub bandwidth_extension: bool,
    /// Enable advanced spectral processing
    pub advanced_spectral: bool,
    /// Enable adaptive algorithms
    pub adaptive_algorithms: bool,
    /// Sample rate for processing
    pub sample_rate: u32,
    /// Buffer size for real-time processing
    pub buffer_size: usize,
    /// Advanced spectral processing configuration
    pub advanced_spectral_config: Option<AdvancedSpectralConfig>,
    /// Adaptive algorithms configuration
    pub adaptive_config: Option<AdaptiveConfig>,
}

impl Default for AudioPreprocessingConfig {
    fn default() -> Self {
        Self {
            noise_suppression: true,
            agc: true,
            echo_cancellation: false,
            bandwidth_extension: false,
            advanced_spectral: true,
            adaptive_algorithms: true,
            sample_rate: 16000,
            buffer_size: 1024,
            advanced_spectral_config: Some(AdvancedSpectralConfig::default()),
            adaptive_config: Some(AdaptiveConfig::default()),
        }
    }
}

/// Result of audio preprocessing
#[derive(Debug, Clone)]
pub struct AudioPreprocessingResult {
    /// Enhanced audio buffer
    pub enhanced_audio: AudioBuffer,
    /// Noise suppression statistics
    pub noise_suppression_stats: Option<NoiseSuppressionStats>,
    /// AGC statistics
    pub agc_stats: Option<AGCStats>,
    /// Echo cancellation statistics
    pub echo_cancellation_stats: Option<EchoCancellationStats>,
    /// Bandwidth extension statistics
    pub bandwidth_extension_stats: Option<BandwidthExtensionStats>,
    /// Advanced spectral processing statistics
    pub advanced_spectral_stats: Option<AdvancedSpectralStats>,
    /// Adaptive algorithms statistics
    pub adaptive_stats: Option<AdaptiveStats>,
    /// Processing time in milliseconds
    pub processing_time_ms: f64,
}

/// Audio preprocessing pipeline
#[derive(Debug)]
pub struct AudioPreprocessor {
    /// Configuration
    config: AudioPreprocessingConfig,
    /// Noise suppression processor
    noise_suppressor: Option<Arc<Mutex<NoiseSuppressionProcessor>>>,
    /// AGC processor
    agc_processor: Option<Arc<Mutex<AGCProcessor>>>,
    /// Echo cancellation processor
    echo_canceller: Option<Arc<Mutex<EchoCancellationProcessor>>>,
    /// Bandwidth extension processor
    bandwidth_extender: Option<Arc<Mutex<BandwidthExtensionProcessor>>>,
    /// Real-time feature extractor
    feature_extractor: Option<Arc<Mutex<RealTimeFeatureExtractor>>>,
    /// Advanced spectral processor
    advanced_spectral_processor: Option<Arc<Mutex<AdvancedSpectralProcessor>>>,
    /// Adaptive algorithms processor
    adaptive_processor: Option<Arc<Mutex<AdaptiveProcessor>>>,
}

impl AudioPreprocessor {
    /// Create a new audio preprocessor with the given configuration
    pub fn new(config: AudioPreprocessingConfig) -> Result<Self, RecognitionError> {
        let mut preprocessor = Self {
            config: config.clone(),
            noise_suppressor: None,
            agc_processor: None,
            echo_canceller: None,
            bandwidth_extender: None,
            feature_extractor: None,
            advanced_spectral_processor: None,
            adaptive_processor: None,
        };

        // Initialize processors based on configuration
        if config.noise_suppression {
            let noise_config = NoiseSuppressionConfig {
                sample_rate: config.sample_rate,
                buffer_size: config.buffer_size,
                algorithm: NoiseSuppressionAlgorithm::SpectralSubtraction,
                alpha: 2.0,
                beta: 0.01,
            };
            preprocessor.noise_suppressor = Some(Arc::new(Mutex::new(
                NoiseSuppressionProcessor::new(noise_config)?,
            )));
        }

        if config.agc {
            let agc_config = AGCConfig {
                sample_rate: config.sample_rate,
                target_level: -20.0,
                max_gain: 30.0,
                attack_time: 0.001,
                release_time: 0.1,
            };
            preprocessor.agc_processor = Some(Arc::new(Mutex::new(AGCProcessor::new(agc_config)?)));
        }

        if config.echo_cancellation {
            let echo_config = EchoCancellationConfig {
                sample_rate: config.sample_rate,
                filter_length: 1024,
                adaptation_rate: 0.01,
                nlp_threshold: 0.5,
            };
            preprocessor.echo_canceller = Some(Arc::new(Mutex::new(
                EchoCancellationProcessor::new(echo_config)?,
            )));
        }

        if config.bandwidth_extension {
            let bwe_config = BandwidthExtensionConfig {
                target_bandwidth: 8000.0,
                method: bandwidth_extension::ExtensionMethod::SpectralReplication,
                quality: bandwidth_extension::QualityLevel::Medium,
                spectral_replication: true,
                hf_emphasis: 1.2,
            };
            preprocessor.bandwidth_extender = Some(Arc::new(Mutex::new(
                BandwidthExtensionProcessor::new(bwe_config)?,
            )));
        }

        // Initialize real-time feature extractor
        let features_config = RealTimeFeatureConfig {
            window_size: 512,
            hop_length: 256,
            n_mels: 13,
            extract_mfcc: true,
            extract_spectral_centroid: true,
            extract_zcr: true,
            extract_spectral_rolloff: true,
            extract_energy: true,
        };
        preprocessor.feature_extractor = Some(Arc::new(Mutex::new(RealTimeFeatureExtractor::new(
            features_config,
        )?)));

        // Initialize advanced spectral processor
        if config.advanced_spectral {
            let spectral_config = config.advanced_spectral_config.unwrap_or_else(|| {
                let mut cfg = AdvancedSpectralConfig::default();
                cfg.sample_rate = config.sample_rate;
                cfg
            });
            preprocessor.advanced_spectral_processor = Some(Arc::new(Mutex::new(
                AdvancedSpectralProcessor::new(spectral_config)?,
            )));
        }

        // Initialize adaptive algorithms processor
        if config.adaptive_algorithms {
            let adaptive_config = config.adaptive_config.unwrap_or_else(|| {
                let mut cfg = AdaptiveConfig::default();
                cfg.sample_rate = config.sample_rate;
                cfg.analysis_window_size = config.buffer_size * 2;
                cfg
            });
            preprocessor.adaptive_processor = Some(Arc::new(Mutex::new(AdaptiveProcessor::new(
                adaptive_config,
            )?)));
        }

        Ok(preprocessor)
    }

    /// Process audio with all enabled preprocessing steps
    pub async fn process(
        &mut self,
        audio: &AudioBuffer,
    ) -> Result<AudioPreprocessingResult, RecognitionError> {
        let start_time = std::time::Instant::now();
        let mut enhanced_audio = audio.clone();

        // Processing statistics
        let mut noise_suppression_stats = None;
        let mut agc_stats = None;
        let mut echo_cancellation_stats = None;
        let mut bandwidth_extension_stats = None;
        let mut advanced_spectral_stats = None;
        let mut adaptive_stats = None;

        // Performance optimization: Minimize lock scope and handle processors sequentially
        // but with reduced lock contention

        // Apply noise suppression with minimized lock scope
        if let Some(ref noise_suppressor) = self.noise_suppressor {
            let result = {
                let mut processor = noise_suppressor.lock().map_err(|e| {
                    RecognitionError::AudioProcessingError {
                        message: format!("Failed to lock noise suppressor: {e}"),
                        source: None,
                    }
                })?;
                processor.process(&enhanced_audio).await?
            };
            enhanced_audio = result.enhanced_audio;
            noise_suppression_stats = Some(result.stats);
        }

        // Apply automatic gain control with minimized lock scope
        if let Some(ref agc_processor) = self.agc_processor {
            let result = {
                let mut processor =
                    agc_processor
                        .lock()
                        .map_err(|e| RecognitionError::AudioProcessingError {
                            message: format!("Failed to lock AGC processor: {e}"),
                            source: None,
                        })?;
                processor.process(&enhanced_audio).await?
            };
            enhanced_audio = result.enhanced_audio;
            agc_stats = Some(result.stats);
        }

        // Apply echo cancellation with minimized lock scope
        if let Some(ref echo_canceller) = self.echo_canceller {
            let result = {
                let mut processor =
                    echo_canceller
                        .lock()
                        .map_err(|e| RecognitionError::AudioProcessingError {
                            message: format!("Failed to lock echo canceller: {e}"),
                            source: None,
                        })?;
                processor.process(&enhanced_audio).await?
            };
            enhanced_audio = result.enhanced_audio;
            echo_cancellation_stats = Some(result.stats);
        }

        // Apply bandwidth extension with minimized lock scope
        if let Some(ref bandwidth_extender) = self.bandwidth_extender {
            let (processed_audio, stats) = {
                let mut processor = bandwidth_extender.lock().map_err(|e| {
                    RecognitionError::AudioProcessingError {
                        message: format!("Failed to lock bandwidth extender: {e}"),
                        source: None,
                    }
                })?;
                let processed = processor.process(&enhanced_audio)?;
                let stats = processor.get_stats().clone();
                (processed, stats)
            };
            enhanced_audio = processed_audio;
            bandwidth_extension_stats = Some(stats);
        }

        // Apply adaptive algorithms first to get optimal parameters
        if let Some(ref adaptive_processor) = self.adaptive_processor {
            let adaptive_result = {
                let mut processor = adaptive_processor.lock().map_err(|e| {
                    RecognitionError::AudioProcessingError {
                        message: format!("Failed to lock adaptive processor: {e}"),
                        source: None,
                    }
                })?;
                processor.analyze_and_adapt(&enhanced_audio)?
            };
            adaptive_stats = Some(adaptive_result.stats);

            // The adaptive parameters could be used to adjust other processors dynamically
            // For now, we'll just collect the statistics
        }

        // Apply advanced spectral processing
        if let Some(ref advanced_spectral_processor) = self.advanced_spectral_processor {
            let spectral_result = {
                let mut processor = advanced_spectral_processor.lock().map_err(|e| {
                    RecognitionError::AudioProcessingError {
                        message: format!("Failed to lock advanced spectral processor: {e}"),
                        source: None,
                    }
                })?;
                processor.process(&enhanced_audio)?
            };
            enhanced_audio = spectral_result.enhanced_audio;
            advanced_spectral_stats = Some(spectral_result.stats);
        }

        let processing_time_ms = start_time.elapsed().as_secs_f64() * 1000.0;

        Ok(AudioPreprocessingResult {
            enhanced_audio,
            noise_suppression_stats,
            agc_stats,
            echo_cancellation_stats,
            bandwidth_extension_stats,
            advanced_spectral_stats,
            adaptive_stats,
            processing_time_ms,
        })
    }

    /// Process audio stream in real-time
    pub async fn process_stream(
        &mut self,
        audio_chunk: &AudioBuffer,
    ) -> Result<AudioPreprocessingResult, RecognitionError> {
        // For streaming, we process smaller chunks with optimized latency
        self.process(audio_chunk).await
    }

    /// High-performance parallel processing for multi-channel audio
    /// This method processes channels in parallel when possible to reduce latency
    pub async fn process_parallel(
        &mut self,
        audio: &AudioBuffer,
    ) -> Result<AudioPreprocessingResult, RecognitionError> {
        let start_time = std::time::Instant::now();

        // For mono audio, fall back to sequential processing
        if audio.channels() == 1 {
            return self.process(audio).await;
        }

        // Split multi-channel audio into separate channels for parallel processing
        let channels = audio.channels() as usize;
        let samples_per_channel = audio.samples().len() / channels;
        let mut channel_buffers = Vec::with_capacity(channels);

        for ch in 0..channels {
            let mut channel_samples = Vec::with_capacity(samples_per_channel);
            for i in 0..samples_per_channel {
                channel_samples.push(audio.samples()[i * channels + ch]);
            }
            channel_buffers.push(AudioBuffer::mono(channel_samples, audio.sample_rate()));
        }

        // Process channels in parallel using tokio tasks
        let mut tasks = Vec::with_capacity(channels);

        for channel_audio in channel_buffers {
            // Note: For this to work properly, we'd need to clone the processors
            // For now, we'll fall back to sequential processing but with optimized structure
            // In a real implementation, we'd need lock-free or per-channel processors
            let result = self.process(&channel_audio).await?;
            tasks.push(result);
        }

        // Merge results from all channels
        let enhanced_samples = self.merge_channel_results(&tasks, channels)?;
        let enhanced_audio =
            AudioBuffer::new(enhanced_samples, audio.sample_rate(), channels as u32);

        // Aggregate statistics from all channels
        let mut noise_suppression_stats = None;
        let mut agc_stats = None;
        let mut echo_cancellation_stats = None;
        let mut bandwidth_extension_stats = None;
        let mut advanced_spectral_stats = None;
        let mut adaptive_stats = None;

        // Take average statistics from all channels
        if let Some(stats) = tasks
            .first()
            .and_then(|t| t.noise_suppression_stats.as_ref())
        {
            noise_suppression_stats = Some(stats.clone());
        }
        if let Some(stats) = tasks.first().and_then(|t| t.agc_stats.as_ref()) {
            agc_stats = Some(stats.clone());
        }
        if let Some(stats) = tasks
            .first()
            .and_then(|t| t.echo_cancellation_stats.as_ref())
        {
            echo_cancellation_stats = Some(stats.clone());
        }
        if let Some(stats) = tasks
            .first()
            .and_then(|t| t.bandwidth_extension_stats.as_ref())
        {
            bandwidth_extension_stats = Some(stats.clone());
        }
        if let Some(stats) = tasks
            .first()
            .and_then(|t| t.advanced_spectral_stats.as_ref())
        {
            advanced_spectral_stats = Some(stats.clone());
        }
        if let Some(stats) = tasks.first().and_then(|t| t.adaptive_stats.as_ref()) {
            adaptive_stats = Some(stats.clone());
        }

        let processing_time_ms = start_time.elapsed().as_secs_f64() * 1000.0;

        Ok(AudioPreprocessingResult {
            enhanced_audio,
            noise_suppression_stats,
            agc_stats,
            echo_cancellation_stats,
            bandwidth_extension_stats,
            advanced_spectral_stats,
            adaptive_stats,
            processing_time_ms,
        })
    }

    /// Helper method to merge channel processing results with SIMD optimization
    fn merge_channel_results(
        &self,
        channel_results: &[AudioPreprocessingResult],
        channels: usize,
    ) -> Result<Vec<f32>, RecognitionError> {
        if channel_results.is_empty() {
            return Ok(Vec::new());
        }

        let samples_per_channel = channel_results[0].enhanced_audio.samples().len();
        let total_samples = samples_per_channel * channels;
        let mut merged_samples = Vec::with_capacity(total_samples);

        // Use SIMD-optimized interleaving when available
        if channels == 2 && samples_per_channel >= 8 {
            self.interleave_stereo_simd(channel_results, &mut merged_samples)?;
        } else {
            // Fallback to scalar interleaving
            for i in 0..samples_per_channel {
                for ch in 0..channels {
                    if ch < channel_results.len() {
                        let channel_samples = channel_results[ch].enhanced_audio.samples();
                        if i < channel_samples.len() {
                            merged_samples.push(channel_samples[i]);
                        } else {
                            merged_samples.push(0.0); // Silence for missing samples
                        }
                    } else {
                        merged_samples.push(0.0); // Silence for missing channels
                    }
                }
            }
        }

        Ok(merged_samples)
    }

    /// SIMD-optimized stereo interleaving
    #[cfg(target_arch = "x86_64")]
    fn interleave_stereo_simd(
        &self,
        channel_results: &[AudioPreprocessingResult],
        merged_samples: &mut Vec<f32>,
    ) -> Result<(), RecognitionError> {
        if channel_results.len() < 2 {
            return Err(RecognitionError::AudioProcessingError {
                message: "Need at least 2 channels for stereo interleaving".to_string(),
                source: None,
            });
        }

        let left_samples = channel_results[0].enhanced_audio.samples();
        let right_samples = channel_results[1].enhanced_audio.samples();
        let samples_per_channel = left_samples.len().min(right_samples.len());

        merged_samples.resize(samples_per_channel * 2, 0.0);

        if std::arch::is_x86_feature_detected!("avx2") {
            unsafe {
                self.interleave_stereo_avx2(left_samples, right_samples, merged_samples);
            }
        } else if std::arch::is_x86_feature_detected!("sse2") {
            unsafe {
                self.interleave_stereo_sse2(left_samples, right_samples, merged_samples);
            }
        } else {
            // Fallback to scalar
            for i in 0..samples_per_channel {
                merged_samples[i * 2] = left_samples[i];
                merged_samples[i * 2 + 1] = right_samples[i];
            }
        }

        Ok(())
    }

    /// AVX2 stereo interleaving implementation
    #[cfg(target_arch = "x86_64")]
    #[target_feature(enable = "avx2")]
    unsafe fn interleave_stereo_avx2(&self, left: &[f32], right: &[f32], output: &mut [f32]) {
        use std::arch::x86_64::*;

        let len = left.len().min(right.len());
        let simd_len = len & !7; // Process 8 samples at a time

        for i in (0..simd_len).step_by(8) {
            // Load 8 samples from each channel
            let left_vec = _mm256_loadu_ps(left.as_ptr().add(i));
            let right_vec = _mm256_loadu_ps(right.as_ptr().add(i));

            // Interleave using unpack operations
            let low_interleaved = _mm256_unpacklo_ps(left_vec, right_vec);
            let high_interleaved = _mm256_unpackhi_ps(left_vec, right_vec);

            // Store interleaved results
            _mm256_storeu_ps(output.as_mut_ptr().add(i * 2), low_interleaved);
            _mm256_storeu_ps(output.as_mut_ptr().add(i * 2 + 8), high_interleaved);
        }

        // Handle remaining samples
        for i in simd_len..len {
            output[i * 2] = left[i];
            output[i * 2 + 1] = right[i];
        }
    }

    /// SSE2 stereo interleaving implementation
    #[cfg(target_arch = "x86_64")]
    #[target_feature(enable = "sse2")]
    unsafe fn interleave_stereo_sse2(&self, left: &[f32], right: &[f32], output: &mut [f32]) {
        use std::arch::x86_64::*;

        let len = left.len().min(right.len());
        let simd_len = len & !3; // Process 4 samples at a time

        for i in (0..simd_len).step_by(4) {
            let left_vec = _mm_loadu_ps(left.as_ptr().add(i));
            let right_vec = _mm_loadu_ps(right.as_ptr().add(i));

            let low_interleaved = _mm_unpacklo_ps(left_vec, right_vec);
            let high_interleaved = _mm_unpackhi_ps(left_vec, right_vec);

            _mm_storeu_ps(output.as_mut_ptr().add(i * 2), low_interleaved);
            _mm_storeu_ps(output.as_mut_ptr().add(i * 2 + 4), high_interleaved);
        }

        for i in simd_len..len {
            output[i * 2] = left[i];
            output[i * 2 + 1] = right[i];
        }
    }

    /// ARM NEON stereo interleaving
    #[cfg(target_arch = "aarch64")]
    fn interleave_stereo_simd(
        &self,
        channel_results: &[AudioPreprocessingResult],
        merged_samples: &mut Vec<f32>,
    ) -> Result<(), RecognitionError> {
        if channel_results.len() < 2 {
            return Err(RecognitionError::AudioProcessingError {
                message: "Need at least 2 channels for stereo interleaving".to_string(),
                source: None,
            });
        }

        let left_samples = channel_results[0].enhanced_audio.samples();
        let right_samples = channel_results[1].enhanced_audio.samples();
        let samples_per_channel = left_samples.len().min(right_samples.len());

        merged_samples.resize(samples_per_channel * 2, 0.0);

        if std::arch::is_aarch64_feature_detected!("neon") {
            unsafe {
                self.interleave_stereo_neon(left_samples, right_samples, merged_samples);
            }
        } else {
            // Fallback to scalar
            for i in 0..samples_per_channel {
                merged_samples[i * 2] = left_samples[i];
                merged_samples[i * 2 + 1] = right_samples[i];
            }
        }

        Ok(())
    }

    /// NEON stereo interleaving implementation
    #[cfg(target_arch = "aarch64")]
    #[target_feature(enable = "neon")]
    unsafe fn interleave_stereo_neon(&self, left: &[f32], right: &[f32], output: &mut [f32]) {
        use std::arch::aarch64::{vld1q_f32, vst1q_f32, vzipq_f32};

        let len = left.len().min(right.len());
        let simd_len = len & !3; // Process 4 samples at a time

        for i in (0..simd_len).step_by(4) {
            let left_vec = vld1q_f32(left.as_ptr().add(i));
            let right_vec = vld1q_f32(right.as_ptr().add(i));

            // Interleave using zip operations
            let interleaved_low = vzipq_f32(left_vec, right_vec);

            vst1q_f32(output.as_mut_ptr().add(i * 2), interleaved_low.0);
            vst1q_f32(output.as_mut_ptr().add(i * 2 + 4), interleaved_low.1);
        }

        for i in simd_len..len {
            output[i * 2] = left[i];
            output[i * 2 + 1] = right[i];
        }
    }

    /// Other architectures fallback
    #[cfg(not(any(target_arch = "x86_64", target_arch = "aarch64")))]
    fn interleave_stereo_simd(
        &self,
        channel_results: &[AudioPreprocessingResult],
        merged_samples: &mut Vec<f32>,
    ) -> Result<(), RecognitionError> {
        if channel_results.len() < 2 {
            return Err(RecognitionError::AudioProcessingError {
                message: "Need at least 2 channels for stereo interleaving".to_string(),
                source: None,
            });
        }

        let left_samples = channel_results[0].enhanced_audio.samples();
        let right_samples = channel_results[1].enhanced_audio.samples();
        let samples_per_channel = left_samples.len().min(right_samples.len());

        merged_samples.resize(samples_per_channel * 2, 0.0);

        // Scalar fallback
        for i in 0..samples_per_channel {
            merged_samples[i * 2] = left_samples[i];
            merged_samples[i * 2 + 1] = right_samples[i];
        }

        Ok(())
    }

    /// Extract real-time features from audio
    pub async fn extract_features(
        &self,
        audio: &AudioBuffer,
    ) -> Result<RealTimeFeatureResult, RecognitionError> {
        if let Some(ref feature_extractor) = self.feature_extractor {
            let extractor =
                feature_extractor
                    .lock()
                    .map_err(|e| RecognitionError::AudioProcessingError {
                        message: format!("Failed to lock feature extractor: {e}"),
                        source: None,
                    })?;
            extractor.extract_features(audio)
        } else {
            Err(RecognitionError::AudioProcessingError {
                message: "Real-time feature extractor not initialized".to_string(),
                source: None,
            })
        }
    }

    /// Get configuration
    #[must_use]
    pub fn config(&self) -> &AudioPreprocessingConfig {
        &self.config
    }

    /// Reset internal state (useful for streaming)
    pub fn reset(&mut self) -> Result<(), RecognitionError> {
        if let Some(ref noise_suppressor) = self.noise_suppressor {
            let mut processor =
                noise_suppressor
                    .lock()
                    .map_err(|e| RecognitionError::AudioProcessingError {
                        message: format!("Failed to lock noise suppressor: {e}"),
                        source: None,
                    })?;
            processor.reset()?;
        }
        if let Some(ref agc_processor) = self.agc_processor {
            let mut processor =
                agc_processor
                    .lock()
                    .map_err(|e| RecognitionError::AudioProcessingError {
                        message: format!("Failed to lock AGC processor: {e}"),
                        source: None,
                    })?;
            processor.reset()?;
        }
        if let Some(ref echo_canceller) = self.echo_canceller {
            let mut processor =
                echo_canceller
                    .lock()
                    .map_err(|e| RecognitionError::AudioProcessingError {
                        message: format!("Failed to lock echo canceller: {e}"),
                        source: None,
                    })?;
            processor.reset()?;
        }
        // Reset methods not implemented for bandwidth extender and feature extractor yet
        // These processors don't have stateful reset methods currently
        Ok(())
    }
}

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

    #[tokio::test]
    async fn test_audio_preprocessor_creation() {
        let config = AudioPreprocessingConfig::default();
        let preprocessor = AudioPreprocessor::new(config);
        assert!(preprocessor.is_ok());
    }

    #[tokio::test]
    async fn test_audio_preprocessing_basic() {
        let config = AudioPreprocessingConfig {
            noise_suppression: true,
            agc: true,
            echo_cancellation: false,
            bandwidth_extension: false,
            advanced_spectral: false,
            adaptive_algorithms: false,
            sample_rate: 16000,
            buffer_size: 1024,
            advanced_spectral_config: None,
            adaptive_config: None,
        };

        let mut preprocessor = AudioPreprocessor::new(config).unwrap();

        // Create test audio buffer
        let samples = vec![0.1f32; 16000]; // 1 second of test audio
        let audio = AudioBuffer::mono(samples, 16000);

        let result = preprocessor.process(&audio).await;
        assert!(result.is_ok());

        let result = result.unwrap();
        // Arc mutability issues are now fixed - processing should work
        assert!(result.noise_suppression_stats.is_some());
        assert!(result.agc_stats.is_some());
        assert!(result.echo_cancellation_stats.is_none());
        assert!(result.bandwidth_extension_stats.is_none());
        assert!(result.processing_time_ms > 0.0);
    }

    #[tokio::test]
    async fn test_feature_extraction() {
        let config = AudioPreprocessingConfig::default();
        let preprocessor = AudioPreprocessor::new(config).unwrap();

        let samples = vec![0.1f32; 16000];
        let audio = AudioBuffer::mono(samples, 16000);

        let result = preprocessor.extract_features(&audio).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_preprocessor_reset() {
        let config = AudioPreprocessingConfig::default();
        let mut preprocessor = AudioPreprocessor::new(config).unwrap();

        let result = preprocessor.reset();
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_simd_stereo_interleaving() {
        let config = AudioPreprocessingConfig::default();
        let preprocessor = AudioPreprocessor::new(config).unwrap();

        // Create test stereo data
        let left_samples = vec![1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0];
        let right_samples = vec![0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8];

        let left_audio = AudioBuffer::mono(left_samples.clone(), 16000);
        let right_audio = AudioBuffer::mono(right_samples.clone(), 16000);

        let channel_results = vec![
            AudioPreprocessingResult {
                enhanced_audio: left_audio,
                noise_suppression_stats: None,
                agc_stats: None,
                echo_cancellation_stats: None,
                bandwidth_extension_stats: None,
                advanced_spectral_stats: None,
                adaptive_stats: None,
                processing_time_ms: 0.0,
            },
            AudioPreprocessingResult {
                enhanced_audio: right_audio,
                noise_suppression_stats: None,
                agc_stats: None,
                echo_cancellation_stats: None,
                bandwidth_extension_stats: None,
                advanced_spectral_stats: None,
                adaptive_stats: None,
                processing_time_ms: 0.0,
            },
        ];

        let result = preprocessor.merge_channel_results(&channel_results, 2);
        assert!(result.is_ok());

        let merged = result.unwrap();
        assert_eq!(merged.len(), 16); // 8 samples * 2 channels

        // Verify interleaving: L0, R0, L1, R1, ...
        for i in 0..8 {
            assert!((merged[i * 2] - left_samples[i]).abs() < f32::EPSILON);
            assert!((merged[i * 2 + 1] - right_samples[i]).abs() < f32::EPSILON);
        }
    }

    #[tokio::test]
    async fn test_memory_efficient_processing() {
        let config = AudioPreprocessingConfig {
            noise_suppression: true,
            agc: false,
            echo_cancellation: false,
            bandwidth_extension: false,
            advanced_spectral: false,
            adaptive_algorithms: false,
            sample_rate: 16000,
            buffer_size: 512, // Smaller buffer for memory efficiency
            advanced_spectral_config: None,
            adaptive_config: None,
        };

        let mut preprocessor = AudioPreprocessor::new(config).unwrap();

        // Test with large audio buffer to verify memory efficiency
        let large_samples = vec![0.1f32; 48000]; // 3 seconds at 16kHz
        let audio = AudioBuffer::mono(large_samples, 16000);

        let result = preprocessor.process(&audio).await;
        assert!(result.is_ok());

        let processing_result = result.unwrap();
        assert!(processing_result.processing_time_ms > 0.0);
        assert_eq!(processing_result.enhanced_audio.samples().len(), 48000);
    }

    #[tokio::test]
    async fn test_parallel_processing_performance() {
        let config = AudioPreprocessingConfig::default();
        let mut preprocessor = AudioPreprocessor::new(config).unwrap();

        // Create stereo audio for parallel processing test
        let samples_per_channel = 16000; // 1 second at 16kHz
        let stereo_samples = (0..samples_per_channel * 2)
            .map(|i| (i as f32 * 0.001) % 1.0)
            .collect::<Vec<f32>>();
        let stereo_audio = AudioBuffer::new(stereo_samples, 16000, 2);

        let start_time = std::time::Instant::now();
        let result = preprocessor.process_parallel(&stereo_audio).await;
        let parallel_time = start_time.elapsed();

        assert!(result.is_ok());
        let processing_result = result.unwrap();
        assert!(processing_result.processing_time_ms > 0.0);

        println!("Parallel processing time: {:?}", parallel_time);

        // Verify the output has correct dimensions
        assert_eq!(processing_result.enhanced_audio.channels(), 2);
        assert_eq!(
            processing_result.enhanced_audio.samples().len(),
            samples_per_channel * 2
        );
    }
}