use crate::*;
use scirs2_core::ndarray::{Array1, ArrayView1, ArrayView2, Axis};
use rayon::prelude::*;
use sklears_core::prelude::{SklearsError, Transform};
use std::collections::HashMap;
#[derive(Debug, Clone)]
pub struct ParallelFeatureExtractor {
chunk_size: usize,
num_threads: Option<usize>,
include_distance_matrix: bool,
include_statistical_features: bool,
include_correlation_features: bool,
distance_metric: String,
}
impl ParallelFeatureExtractor {
pub fn new() -> Self {
Self {
chunk_size: 1000,
num_threads: None,
include_distance_matrix: false,
include_statistical_features: true,
include_correlation_features: false,
distance_metric: "euclidean".to_string(),
}
}
pub fn chunk_size(mut self, size: usize) -> Self {
self.chunk_size = size.max(1);
self
}
pub fn num_threads(mut self, threads: usize) -> Self {
self.num_threads = Some(threads.max(1));
self
}
pub fn include_distance_matrix(mut self, include: bool) -> Self {
self.include_distance_matrix = include;
self
}
pub fn include_statistical_features(mut self, include: bool) -> Self {
self.include_statistical_features = include;
self
}
pub fn include_correlation_features(mut self, include: bool) -> Self {
self.include_correlation_features = include;
self
}
pub fn distance_metric(mut self, metric: &str) -> Self {
self.distance_metric = metric.to_string();
self
}
pub fn extract_features(&self, data: &ArrayView2<Float>) -> SklResult<Array1<Float>> {
if data.nrows() == 0 || data.ncols() == 0 {
return Err(SklearsError::InvalidInput(
"Input data cannot be empty".to_string(),
));
}
let result = if let Some(threads) = self.num_threads {
rayon::ThreadPoolBuilder::new()
.num_threads(threads)
.build()
.map_err(|e| {
SklearsError::InvalidInput(format!("Failed to create thread pool: {}", e))
})?
.install(|| self.extract_features_impl(data))
} else {
self.extract_features_impl(data)
};
result
}
fn extract_features_impl(&self, data: &ArrayView2<Float>) -> SklResult<Array1<Float>> {
let mut features = Vec::new();
if self.include_statistical_features {
features.extend(self.compute_parallel_statistical_features(data)?);
}
if self.include_distance_matrix {
features.extend(self.compute_parallel_distance_matrix_features(data)?);
}
if self.include_correlation_features {
features.extend(self.compute_parallel_correlation_features(data)?);
}
if features.is_empty() {
return Err(SklearsError::InvalidInput(
"No features selected for extraction".to_string(),
));
}
for &feature in &features {
if !feature.is_finite() {
return Err(SklearsError::InvalidInput(
"Non-finite feature detected".to_string(),
));
}
}
Ok(Array1::from_vec(features))
}
fn compute_parallel_statistical_features(
&self,
data: &ArrayView2<Float>,
) -> SklResult<Vec<Float>> {
let n_samples = data.nrows();
let n_features = data.ncols();
let column_stats: Vec<Vec<Float>> = (0..n_features)
.into_par_iter()
.map(|col_idx| {
let column = data.column(col_idx);
self.compute_column_statistics(&column)
})
.collect();
let mut features = Vec::new();
let column_means: Vec<Float> = column_stats.iter().map(|stats| stats[0]).collect();
features.push(column_means.iter().sum::<Float>() / n_features as Float);
let column_stds: Vec<Float> = column_stats.iter().map(|stats| stats[1]).collect();
features.push(column_stds.iter().sum::<Float>() / n_features as Float);
let column_mins: Vec<Float> = column_stats.iter().map(|stats| stats[2]).collect();
let column_maxs: Vec<Float> = column_stats.iter().map(|stats| stats[3]).collect();
features.push(column_mins.iter().sum::<Float>() / n_features as Float);
features.push(column_maxs.iter().sum::<Float>() / n_features as Float);
features.push(
*column_mins
.iter()
.min_by(|a, b| a.partial_cmp(b).expect("operation should succeed"))
.expect("operation should succeed"),
);
features.push(
*column_maxs
.iter()
.max_by(|a, b| a.partial_cmp(b).expect("operation should succeed"))
.expect("operation should succeed"),
);
let row_stats: Vec<Vec<Float>> = (0..n_samples)
.into_par_iter()
.step_by(self.chunk_size)
.map(|start_idx| {
let end_idx = (start_idx + self.chunk_size).min(n_samples);
let mut chunk_stats = Vec::new();
for row_idx in start_idx..end_idx {
let row = data.row(row_idx);
let row_mean = row.iter().sum::<Float>() / n_features as Float;
let row_var = row.iter().map(|&x| (x - row_mean).powi(2)).sum::<Float>()
/ n_features as Float;
chunk_stats.push(row_mean);
chunk_stats.push(row_var.sqrt());
}
chunk_stats
})
.collect();
let all_row_stats: Vec<Float> = row_stats.into_iter().flatten().collect();
if !all_row_stats.is_empty() {
features.push(all_row_stats.iter().sum::<Float>() / all_row_stats.len() as Float);
features.push(all_row_stats.iter().cloned().fold(0.0, Float::max));
features.push(
all_row_stats
.iter()
.cloned()
.fold(Float::INFINITY, Float::min),
);
}
Ok(features)
}
fn compute_column_statistics(&self, column: &ArrayView1<Float>) -> Vec<Float> {
let n = column.len() as Float;
let mean = column.iter().sum::<Float>() / n;
let variance = column.iter().map(|&x| (x - mean).powi(2)).sum::<Float>() / n;
let std_dev = variance.sqrt();
let min_val = column.iter().cloned().fold(Float::INFINITY, Float::min);
let max_val = column.iter().cloned().fold(Float::NEG_INFINITY, Float::max);
vec![mean, std_dev, min_val, max_val]
}
fn compute_parallel_distance_matrix_features(
&self,
data: &ArrayView2<Float>,
) -> SklResult<Vec<Float>> {
let n_samples = data.nrows();
if n_samples < 2 {
return Ok(vec![0.0, 0.0, 0.0, 0.0]); }
let distance_chunks: Result<Vec<Vec<Float>>, SklearsError> = (0..n_samples)
.into_par_iter()
.step_by(self.chunk_size)
.map(|start_i| -> SklResult<Vec<Float>> {
let end_i = (start_i + self.chunk_size).min(n_samples);
let mut chunk_distances = Vec::new();
for i in start_i..end_i {
for j in (i + 1)..n_samples {
let dist = self.compute_distance(&data.row(i), &data.row(j))?;
chunk_distances.push(dist);
}
}
Ok(chunk_distances)
})
.collect();
let distance_chunks = distance_chunks?;
let all_distances: Vec<Float> = distance_chunks.into_iter().flatten().collect();
if all_distances.is_empty() {
return Ok(vec![0.0, 0.0, 0.0, 0.0]);
}
let n_distances = all_distances.len() as Float;
let mean_distance = all_distances.iter().sum::<Float>() / n_distances;
let variance = all_distances
.iter()
.map(|&d| (d - mean_distance).powi(2))
.sum::<Float>()
/ n_distances;
let std_distance = variance.sqrt();
let min_distance = all_distances
.iter()
.cloned()
.fold(Float::INFINITY, Float::min);
let max_distance = all_distances
.iter()
.cloned()
.fold(Float::NEG_INFINITY, Float::max);
Ok(vec![
mean_distance,
std_distance,
min_distance,
max_distance,
])
}
fn compute_parallel_correlation_features(
&self,
data: &ArrayView2<Float>,
) -> SklResult<Vec<Float>> {
let n_features = data.ncols();
if n_features < 2 {
return Ok(vec![0.0, 0.0, 0.0]); }
let column_means: Vec<Float> = (0..n_features)
.into_par_iter()
.map(|col_idx| {
let column = data.column(col_idx);
column.iter().sum::<Float>() / data.nrows() as Float
})
.collect();
let correlation_chunks: Vec<Vec<Float>> = (0..n_features)
.into_par_iter()
.step_by(self.chunk_size)
.map(|start_i| {
let end_i = (start_i + self.chunk_size).min(n_features);
let mut chunk_correlations = Vec::new();
for i in start_i..end_i {
for j in (i + 1)..n_features {
let corr = self.compute_correlation(
&data.column(i),
&data.column(j),
column_means[i],
column_means[j],
);
chunk_correlations.push(corr);
}
}
chunk_correlations
})
.collect();
let all_correlations: Vec<Float> = correlation_chunks.into_iter().flatten().collect();
if all_correlations.is_empty() {
return Ok(vec![0.0, 0.0, 0.0]);
}
let n_correlations = all_correlations.len() as Float;
let mean_correlation = all_correlations.iter().sum::<Float>() / n_correlations;
let variance = all_correlations
.iter()
.map(|&c| (c - mean_correlation).powi(2))
.sum::<Float>()
/ n_correlations;
let std_correlation = variance.sqrt();
let max_abs_correlation = all_correlations
.iter()
.map(|&c| c.abs())
.fold(0.0, Float::max);
Ok(vec![mean_correlation, std_correlation, max_abs_correlation])
}
fn compute_distance(&self, p1: &ArrayView1<Float>, p2: &ArrayView1<Float>) -> SklResult<Float> {
match self.distance_metric.as_str() {
"euclidean" => {
let dist = p1
.iter()
.zip(p2.iter())
.map(|(a, b)| (a - b).powi(2))
.sum::<Float>()
.sqrt();
Ok(dist)
}
"manhattan" => {
let dist = p1
.iter()
.zip(p2.iter())
.map(|(a, b)| (a - b).abs())
.sum::<Float>();
Ok(dist)
}
"cosine" => {
let dot_product = p1.iter().zip(p2.iter()).map(|(a, b)| a * b).sum::<Float>();
let norm1 = p1.iter().map(|&a| a.powi(2)).sum::<Float>().sqrt();
let norm2 = p2.iter().map(|&a| a.powi(2)).sum::<Float>().sqrt();
if norm1 == 0.0 || norm2 == 0.0 {
Ok(1.0) } else {
Ok(1.0 - dot_product / (norm1 * norm2))
}
}
_ => Err(SklearsError::InvalidInput(format!(
"Unsupported distance metric: {}",
self.distance_metric
))),
}
}
fn compute_correlation(
&self,
col1: &ArrayView1<Float>,
col2: &ArrayView1<Float>,
mean1: Float,
mean2: Float,
) -> Float {
let n = col1.len() as Float;
let covariance = col1
.iter()
.zip(col2.iter())
.map(|(&x, &y)| (x - mean1) * (y - mean2))
.sum::<Float>()
/ n;
let var1 = col1.iter().map(|&x| (x - mean1).powi(2)).sum::<Float>() / n;
let var2 = col2.iter().map(|&x| (x - mean2).powi(2)).sum::<Float>() / n;
let std1 = var1.sqrt();
let std2 = var2.sqrt();
if std1 == 0.0 || std2 == 0.0 {
0.0
} else {
covariance / (std1 * std2)
}
}
}
impl Default for ParallelFeatureExtractor {
fn default() -> Self {
Self::new()
}
}