leptos-helios 0.8.1

//! Enhanced data processing implementation for TDD REFACTOR phase
//! This provides robust, production-ready data processing capabilities
//!
//! ## Module Structure
//!
//! - `errors`: Data processing error types and handling
//! - `types`: Core data processing types and data structures
//! - `strategies`: Processing strategy selection and management
//! - `processors`: Data processing implementations
//! - `validators`: Data validation and schema checking
//! - `optimizers`: Performance optimization utilities

pub mod errors;
pub mod strategies;
pub mod types;

// Re-export main types for convenience
pub use errors::*;
pub use strategies::*;
pub use types::*;

// use crate::performance::{PerformanceConfig, PerformanceManager}; // Temporarily disabled
use polars::prelude::*;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

/// Main data processing engine
pub struct DataProcessingEngine {
    strategy_selector: StrategySelector,
    // performance_manager: Arc<PerformanceManager>, // Temporarily disabled
    config: ProcessingConfig,
}

/// Processing configuration
#[derive(Debug, Clone)]
pub struct ProcessingConfig {
    pub default_strategy: ProcessingStrategy,
    pub enable_adaptive_selection: bool,
    pub performance_threshold: f64,
    pub memory_limit: usize,
    pub timeout: Duration,
}

impl Default for ProcessingConfig {
    fn default() -> Self {
        Self {
            default_strategy: ProcessingStrategy::CPU(RayonConfig::new()),
            enable_adaptive_selection: true,
            performance_threshold: 0.8,
            memory_limit: 1024 * 1024 * 1024,  // 1GB
            timeout: Duration::from_secs(300), // 5 minutes
        }
    }
}

impl DataProcessingEngine {
    /// Create a new data processing engine
    pub fn new() -> Self {
        Self {
            strategy_selector: StrategySelector::new(),
            // performance_manager: Arc::new(PerformanceManager::new(PerformanceConfig::default())), // Temporarily disabled
            config: ProcessingConfig::default(),
        }
    }

    /// Create a data processing engine with custom configuration
    pub fn with_config(config: ProcessingConfig) -> Self {
        Self {
            strategy_selector: StrategySelector::new(),
            // performance_manager: Arc::new(PerformanceManager::new(PerformanceConfig::default())), // Temporarily disabled
            config,
        }
    }

    /// Process data using the optimal strategy
    pub async fn process_data(
        &mut self,
        data: DataFrame,
        spec: &DataSpec,
    ) -> DataResult<DataFrame> {
        let start_time = std::time::Instant::now();

        // Select optimal strategy
        let strategy = if self.config.enable_adaptive_selection {
            self.strategy_selector.select_strategy(spec)?
        } else {
            self.config.default_strategy.clone()
        };

        // Process data with selected strategy
        let result = match strategy {
            ProcessingStrategy::CPU(ref config) => self.process_with_cpu(data, config).await,
            ProcessingStrategy::GPU(ref config) => self.process_with_gpu(data, config).await,
            ProcessingStrategy::Streaming(ref config) => {
                self.process_with_streaming(data, config).await
            }
            ProcessingStrategy::Hybrid(ref config) => self.process_with_hybrid(data, config).await,
        }?;

        // Record performance metrics
        let duration = start_time.elapsed();
        let metric =
            PerformanceMetric::new(strategy.name(), spec.row_count, duration.as_secs_f64())
                .with_memory_usage(spec.memory_size);

        self.strategy_selector.record_performance(metric);

        Ok(result)
    }

    /// Process data using CPU strategy
    async fn process_with_cpu(
        &self,
        data: DataFrame,
        _config: &RayonConfig,
    ) -> DataResult<DataFrame> {
        // Mock CPU processing implementation
        // In a real implementation, this would use Rayon for parallel processing
        Ok(data)
    }

    /// Process data using GPU strategy
    async fn process_with_gpu(
        &self,
        data: DataFrame,
        _config: &ComputeConfig,
    ) -> DataResult<DataFrame> {
        // Mock GPU processing implementation
        // In a real implementation, this would use WebGPU or CUDA
        Ok(data)
    }

    /// Process data using streaming strategy
    async fn process_with_streaming(
        &self,
        data: DataFrame,
        _config: &StreamConfig,
    ) -> DataResult<DataFrame> {
        // Mock streaming processing implementation
        // In a real implementation, this would process data in chunks
        Ok(data)
    }

    /// Process data using hybrid strategy
    async fn process_with_hybrid(
        &self,
        data: DataFrame,
        _config: &HybridConfig,
    ) -> DataResult<DataFrame> {
        // Mock hybrid processing implementation
        // In a real implementation, this would combine CPU and GPU processing
        Ok(data)
    }

    /// Get processing statistics
    pub fn get_statistics(&self) -> ProcessingStatistics {
        ProcessingStatistics {
            total_operations: self.strategy_selector.performance_history().len(),
            average_processing_time: self.calculate_average_processing_time(),
            strategy_usage: self.calculate_strategy_usage(),
            performance_trends: self.calculate_performance_trends(),
        }
    }

    /// Calculate average processing time
    fn calculate_average_processing_time(&self) -> Duration {
        let history = self.strategy_selector.performance_history();
        if history.is_empty() {
            return Duration::from_secs(0);
        }

        let total_duration: f64 = history.iter().map(|m| m.duration).sum();
        let average_duration = total_duration / history.len() as f64;
        Duration::from_secs_f64(average_duration)
    }

    /// Calculate strategy usage statistics
    fn calculate_strategy_usage(&self) -> HashMap<String, f64> {
        let history = self.strategy_selector.performance_history();
        let mut usage = HashMap::new();

        for metric in history {
            let count = usage.entry(metric.strategy.clone()).or_insert(0.0);
            *count += 1.0;
        }

        let total = usage.values().sum::<f64>();
        for count in usage.values_mut() {
            *count /= total;
        }

        usage
    }

    /// Calculate performance trends
    fn calculate_performance_trends(&self) -> HashMap<String, f64> {
        let history = self.strategy_selector.performance_history();
        let mut trends = HashMap::new();

        // Group by strategy and calculate trend
        let mut strategy_metrics: HashMap<String, Vec<&PerformanceMetric>> = HashMap::new();
        for metric in history {
            strategy_metrics
                .entry(metric.strategy.clone())
                .or_default()
                .push(metric);
        }

        for (strategy, metrics) in strategy_metrics {
            if metrics.len() < 2 {
                continue;
            }

            // Calculate simple linear trend
            let first_throughput = metrics[0].throughput();
            let last_throughput = metrics[metrics.len() - 1].throughput();
            let trend = (last_throughput - first_throughput) / first_throughput;

            trends.insert(strategy, trend);
        }

        trends
    }

    /// Get strategy selector
    pub fn strategy_selector(&self) -> &StrategySelector {
        &self.strategy_selector
    }

    /// Get mutable strategy selector
    pub fn strategy_selector_mut(&mut self) -> &mut StrategySelector {
        &mut self.strategy_selector
    }

    /// Update configuration
    pub fn update_config(&mut self, config: ProcessingConfig) {
        self.config = config;
    }

    /// Get current configuration
    pub fn config(&self) -> &ProcessingConfig {
        &self.config
    }
}

impl Default for DataProcessingEngine {
    fn default() -> Self {
        Self::new()
    }
}

/// Processing statistics
#[derive(Debug, Clone)]
pub struct ProcessingStatistics {
    pub total_operations: usize,
    pub average_processing_time: Duration,
    pub strategy_usage: HashMap<String, f64>,
    pub performance_trends: HashMap<String, f64>,
}

impl ProcessingStatistics {
    /// Create new processing statistics
    pub fn new() -> Self {
        Self {
            total_operations: 0,
            average_processing_time: Duration::from_secs(0),
            strategy_usage: HashMap::new(),
            performance_trends: HashMap::new(),
        }
    }

    /// Get the most used strategy
    pub fn most_used_strategy(&self) -> Option<&String> {
        self.strategy_usage
            .iter()
            .max_by(|a, b| a.1.partial_cmp(b.1).unwrap())
            .map(|(strategy, _)| strategy)
    }

    /// Get the best performing strategy
    pub fn best_performing_strategy(&self) -> Option<&String> {
        self.performance_trends
            .iter()
            .max_by(|a, b| a.1.partial_cmp(b.1).unwrap())
            .map(|(strategy, _)| strategy)
    }

    /// Get overall performance score
    pub fn overall_performance_score(&self) -> f64 {
        if self.performance_trends.is_empty() {
            return 0.0;
        }

        let average_trend: f64 =
            self.performance_trends.values().sum::<f64>() / self.performance_trends.len() as f64;
        average_trend
    }
}

/// Data processing utilities
pub mod utils {
    use super::*;

    /// Create a data specification from a DataFrame
    pub fn create_data_spec(df: &DataFrame) -> DataResult<DataSpec> {
        let mut spec = DataSpec::new()
            .with_row_count(df.height())
            .with_memory_size(df.estimated_size());

        for (_i, column) in df.get_columns().iter().enumerate() {
            let column_spec =
                ColumnSpec::new(column.name().to_string(), column.dtype().to_string())
                    .with_nullable(column.null_count() > 0);

            spec.add_column(column_spec);
            spec.add_data_type(column.name().to_string(), column.dtype().to_string());
        }

        Ok(spec)
    }

    /// Estimate processing time for a given strategy and data size
    pub fn estimate_processing_time(strategy: &ProcessingStrategy, data_size: usize) -> Duration {
        let base_time = data_size as u64 * 100; // 100ns per row

        let multiplier = match strategy {
            ProcessingStrategy::CPU(config) => {
                let threads = config.num_threads.unwrap_or(num_cpus::get());
                1.0 / threads as f64
            }
            ProcessingStrategy::GPU(_) => 0.1, // GPU is typically 10x faster
            ProcessingStrategy::Streaming(_) => 1.5, // Streaming has overhead
            ProcessingStrategy::Hybrid(_) => 0.5, // Hybrid is in between
        };

        Duration::from_nanos((base_time as f64 * multiplier) as u64)
    }

    /// Check if data is suitable for a given strategy
    pub fn is_strategy_suitable(strategy: &ProcessingStrategy, data_spec: &DataSpec) -> bool {
        match strategy {
            ProcessingStrategy::CPU(_) => data_spec.row_count < 100000,
            ProcessingStrategy::GPU(_) => {
                data_spec.row_count > 10000 && data_spec.memory_size < 1024 * 1024 * 1024
            }
            ProcessingStrategy::Streaming(_) => data_spec.row_count > 1000000,
            ProcessingStrategy::Hybrid(_) => {
                data_spec.row_count > 1000 && data_spec.row_count < 1000000
            }
        }
    }

    /// Get recommended strategies for given data
    pub fn get_recommended_strategies(data_spec: &DataSpec) -> Vec<ProcessingStrategy> {
        let mut strategies = Vec::new();

        if data_spec.is_small() {
            strategies.push(ProcessingStrategy::CPU(RayonConfig::new()));
        } else if data_spec.is_large() {
            strategies.push(ProcessingStrategy::GPU(ComputeConfig::new()));
            strategies.push(ProcessingStrategy::Streaming(StreamConfig::new()));
        } else {
            strategies.push(ProcessingStrategy::Hybrid(HybridConfig::new()));
            strategies.push(ProcessingStrategy::CPU(RayonConfig::new()));
        }

        strategies
    }
}