use scirs2_core::ndarray::{Array1, Array2, ArrayView2, Axis};
use sklears_core::{
error::{Result as SklResult, SklearsError},
traits::{Estimator, Fit, Predict, Untrained},
types::Float,
};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread;
#[derive(Debug, Clone)]
pub struct MultiOutputClassifier<S = Untrained> {
state: S,
n_jobs: Option<i32>,
}
impl MultiOutputClassifier<Untrained> {
pub fn new() -> Self {
Self {
state: Untrained,
n_jobs: None,
}
}
pub fn n_jobs(mut self, n_jobs: Option<i32>) -> Self {
self.n_jobs = n_jobs;
self
}
}
impl Default for MultiOutputClassifier<Untrained> {
fn default() -> Self {
Self::new()
}
}
impl Estimator for MultiOutputClassifier<Untrained> {
type Config = ();
type Error = SklearsError;
type Float = Float;
fn config(&self) -> &Self::Config {
&()
}
}
impl Fit<ArrayView2<'_, Float>, Array2<i32>> for MultiOutputClassifier<Untrained> {
type Fitted = MultiOutputClassifier<MultiOutputClassifierTrained>;
#[allow(non_snake_case)]
fn fit(self, X: &ArrayView2<'_, Float>, y: &Array2<i32>) -> SklResult<Self::Fitted> {
let X = X.to_owned();
let (n_samples, n_features) = X.dim();
if n_samples != y.nrows() {
return Err(SklearsError::InvalidInput(
"X and y must have the same number of samples".to_string(),
));
}
let n_targets = y.ncols();
if n_targets == 0 {
return Err(SklearsError::InvalidInput(
"y must have at least one target".to_string(),
));
}
let mut classes_per_target = Vec::new();
let mut target_models = HashMap::new();
for target_idx in 0..n_targets {
let y_target = y.column(target_idx);
let mut target_classes: Vec<i32> = y_target
.iter()
.cloned()
.collect::<std::collections::HashSet<_>>()
.into_iter()
.collect();
target_classes.sort();
let mut class_centroids = HashMap::new();
for &class_label in &target_classes {
let mut centroid = Array1::<Float>::zeros(n_features);
let mut count = 0;
for (sample_idx, &sample_class) in y_target.iter().enumerate() {
if sample_class == class_label {
for feature_idx in 0..n_features {
centroid[feature_idx] += X[[sample_idx, feature_idx]];
}
count += 1;
}
}
if count > 0 {
centroid /= count as f64;
}
class_centroids.insert(class_label, centroid);
}
target_models.insert(target_idx, class_centroids);
classes_per_target.push(target_classes);
}
if let Some(n_jobs) = self.n_jobs {
if n_jobs > 1 && n_targets > 1 {
return self.fit_parallel(X, y, n_jobs as usize);
}
}
Ok(MultiOutputClassifier {
state: MultiOutputClassifierTrained {
classes_per_target,
target_models,
n_targets,
n_features,
},
n_jobs: self.n_jobs,
})
}
}
impl MultiOutputClassifier<Untrained> {
#[allow(non_snake_case)]
fn fit_parallel(
self,
X: Array2<Float>,
y: &Array2<i32>,
n_jobs: usize,
) -> SklResult<MultiOutputClassifier<MultiOutputClassifierTrained>> {
let (_n_samples, n_features) = X.dim();
let n_targets = y.ncols();
let X_arc = Arc::new(X);
let y_arc = Arc::new(y.clone());
let classes_per_target = Arc::new(Mutex::new(Vec::with_capacity(n_targets)));
let target_models = Arc::new(Mutex::new(HashMap::new()));
let chunk_size = n_targets.div_ceil(n_jobs);
let mut handles = vec![];
for worker_id in 0..n_jobs {
let start_target = worker_id * chunk_size;
let end_target = std::cmp::min(start_target + chunk_size, n_targets);
if start_target >= n_targets {
break; }
let X_thread = Arc::clone(&X_arc);
let y_thread = Arc::clone(&y_arc);
let classes_thread = Arc::clone(&classes_per_target);
let models_thread = Arc::clone(&target_models);
let handle = thread::spawn(move || -> SklResult<()> {
let mut local_classes = Vec::new();
let mut local_models = HashMap::new();
for target_idx in start_target..end_target {
let y_target = y_thread.column(target_idx);
let mut target_classes: Vec<i32> = y_target
.iter()
.cloned()
.collect::<std::collections::HashSet<_>>()
.into_iter()
.collect();
target_classes.sort();
let mut class_centroids = HashMap::new();
for &class_label in &target_classes {
let mut centroid = Array1::<Float>::zeros(n_features);
let mut count = 0;
for (sample_idx, &sample_class) in y_target.iter().enumerate() {
if sample_class == class_label {
for feature_idx in 0..n_features {
centroid[feature_idx] += X_thread[[sample_idx, feature_idx]];
}
count += 1;
}
}
if count > 0 {
centroid /= count as f64;
}
class_centroids.insert(class_label, centroid);
}
local_models.insert(target_idx, class_centroids);
local_classes.push((target_idx, target_classes));
}
{
let mut classes_guard =
classes_thread.lock().expect("lock should not be poisoned");
let mut models_guard =
models_thread.lock().expect("lock should not be poisoned");
local_classes.sort_by_key(|(idx, _)| *idx);
for (target_idx, target_classes) in local_classes {
while classes_guard.len() <= target_idx {
classes_guard.push(vec![]);
}
classes_guard[target_idx] = target_classes;
}
for (target_idx, class_centroids) in local_models {
models_guard.insert(target_idx, class_centroids);
}
}
Ok(())
});
handles.push(handle);
}
for handle in handles {
handle.join().map_err(|_| {
SklearsError::InvalidInput("Thread panicked during parallel training".to_string())
})??;
}
let final_classes = Arc::try_unwrap(classes_per_target)
.map_err(|_| SklearsError::InvalidInput("Failed to extract classes".to_string()))?
.into_inner()
.expect("operation should succeed");
let final_models = Arc::try_unwrap(target_models)
.map_err(|_| SklearsError::InvalidInput("Failed to extract models".to_string()))?
.into_inner()
.expect("operation should succeed");
Ok(MultiOutputClassifier {
state: MultiOutputClassifierTrained {
classes_per_target: final_classes,
target_models: final_models,
n_targets,
n_features,
},
n_jobs: Some(n_jobs as i32),
})
}
}
impl MultiOutputClassifier<MultiOutputClassifierTrained> {
pub fn classes(&self) -> &[Vec<i32>] {
&self.state.classes_per_target
}
pub fn n_targets(&self) -> usize {
self.state.n_targets
}
}
impl Predict<ArrayView2<'_, Float>, Array2<i32>>
for MultiOutputClassifier<MultiOutputClassifierTrained>
{
#[allow(non_snake_case)]
fn predict(&self, X: &ArrayView2<'_, Float>) -> SklResult<Array2<i32>> {
let X = X.to_owned();
let (n_samples, n_features) = X.dim();
if n_features != self.state.n_features {
return Err(SklearsError::InvalidInput(
"Number of features doesn't match training data".to_string(),
));
}
let mut predictions = Array2::<i32>::zeros((n_samples, self.state.n_targets));
for target_idx in 0..self.state.n_targets {
if let Some(class_centroids) = self.state.target_models.get(&target_idx) {
for (sample_idx, sample) in X.axis_iter(Axis(0)).enumerate() {
let mut min_distance = f64::INFINITY;
let mut best_class = 0;
for (&class_label, centroid) in class_centroids {
let mut distance = 0.0;
for feature_idx in 0..n_features {
let diff = sample[feature_idx] - centroid[feature_idx];
distance += diff * diff;
}
distance = distance.sqrt();
if distance < min_distance {
min_distance = distance;
best_class = class_label;
}
}
predictions[[sample_idx, target_idx]] = best_class;
}
}
}
Ok(predictions)
}
}
#[derive(Debug, Clone)]
pub struct MultiOutputRegressor<S = Untrained> {
state: S,
n_jobs: Option<i32>,
}
impl MultiOutputRegressor<Untrained> {
pub fn new() -> Self {
Self {
state: Untrained,
n_jobs: None,
}
}
pub fn n_jobs(mut self, n_jobs: Option<i32>) -> Self {
self.n_jobs = n_jobs;
self
}
}
impl Default for MultiOutputRegressor<Untrained> {
fn default() -> Self {
Self::new()
}
}
impl Estimator for MultiOutputRegressor<Untrained> {
type Config = ();
type Error = SklearsError;
type Float = Float;
fn config(&self) -> &Self::Config {
&()
}
}
impl Fit<ArrayView2<'_, Float>, Array2<f64>> for MultiOutputRegressor<Untrained> {
type Fitted = MultiOutputRegressor<MultiOutputRegressorTrained>;
#[allow(non_snake_case)]
fn fit(self, X: &ArrayView2<'_, Float>, y: &Array2<f64>) -> SklResult<Self::Fitted> {
let X = X.to_owned();
let (n_samples, n_features) = X.dim();
if n_samples != y.nrows() {
return Err(SklearsError::InvalidInput(
"X and y must have the same number of samples".to_string(),
));
}
let n_targets = y.ncols();
if n_targets == 0 {
return Err(SklearsError::InvalidInput(
"y must have at least one target".to_string(),
));
}
let mut target_models = HashMap::new();
for target_idx in 0..n_targets {
let y_target = y.column(target_idx);
let mut weights = Array1::<Float>::zeros(n_features);
let y_mean = y_target
.mean()
.expect("array should have elements for mean computation");
let bias = y_mean;
for feature_idx in 0..n_features {
let mut correlation = 0.0;
let mut x_mean = 0.0;
for sample_idx in 0..n_samples {
x_mean += X[[sample_idx, feature_idx]];
}
x_mean /= n_samples as f64;
let mut numerator = 0.0;
let mut x_var = 0.0;
let mut y_var = 0.0;
for sample_idx in 0..n_samples {
let x_diff = X[[sample_idx, feature_idx]] - x_mean;
let y_diff = y_target[sample_idx] - y_mean;
numerator += x_diff * y_diff;
x_var += x_diff * x_diff;
y_var += y_diff * y_diff;
}
if x_var > 1e-10 && y_var > 1e-10 {
correlation = numerator / (x_var.sqrt() * y_var.sqrt());
}
weights[feature_idx] = correlation * 0.1; }
target_models.insert(target_idx, (weights, bias));
}
if let Some(n_jobs) = self.n_jobs {
if n_jobs > 1 && n_targets > 1 {
return self.fit_parallel(X, y, n_jobs as usize);
}
}
Ok(MultiOutputRegressor {
state: MultiOutputRegressorTrained {
target_models,
n_targets,
n_features,
},
n_jobs: self.n_jobs,
})
}
}
impl MultiOutputRegressor<Untrained> {
#[allow(non_snake_case)]
fn fit_parallel(
self,
X: Array2<Float>,
y: &Array2<f64>,
n_jobs: usize,
) -> SklResult<MultiOutputRegressor<MultiOutputRegressorTrained>> {
let (n_samples, n_features) = X.dim();
let n_targets = y.ncols();
let X_arc = Arc::new(X);
let y_arc = Arc::new(y.clone());
let target_models = Arc::new(Mutex::new(HashMap::new()));
let chunk_size = n_targets.div_ceil(n_jobs);
let mut handles = vec![];
for worker_id in 0..n_jobs {
let start_target = worker_id * chunk_size;
let end_target = std::cmp::min(start_target + chunk_size, n_targets);
if start_target >= n_targets {
break; }
let X_thread = Arc::clone(&X_arc);
let y_thread = Arc::clone(&y_arc);
let models_thread = Arc::clone(&target_models);
let handle = thread::spawn(move || -> SklResult<()> {
let mut local_models = HashMap::new();
for target_idx in start_target..end_target {
let y_target = y_thread.column(target_idx);
let mut weights = Array1::<f64>::zeros(n_features);
let y_mean = y_target
.mean()
.expect("array should have elements for mean computation");
let bias: f64 = y_mean;
for feature_idx in 0..n_features {
let mut correlation = 0.0;
let mut x_mean = 0.0;
for sample_idx in 0..n_samples {
x_mean += X_thread[[sample_idx, feature_idx]];
}
x_mean /= n_samples as f64;
let mut numerator = 0.0;
let mut x_var = 0.0;
let mut y_var = 0.0;
for sample_idx in 0..n_samples {
let x_diff = X_thread[[sample_idx, feature_idx]] - x_mean;
let y_diff = y_target[sample_idx] - y_mean;
numerator += x_diff * y_diff;
x_var += x_diff * x_diff;
y_var += y_diff * y_diff;
}
if x_var > 1e-10 && y_var > 1e-10 {
correlation = numerator / (x_var.sqrt() * y_var.sqrt());
}
weights[feature_idx] = correlation * 0.1; }
local_models.insert(target_idx, (weights, bias));
}
{
let mut models_guard =
models_thread.lock().expect("lock should not be poisoned");
for (target_idx, model) in local_models {
models_guard.insert(target_idx, model);
}
}
Ok(())
});
handles.push(handle);
}
for handle in handles {
handle.join().map_err(|_| {
SklearsError::InvalidInput("Thread panicked during parallel training".to_string())
})??;
}
let final_models = Arc::try_unwrap(target_models)
.map_err(|_| SklearsError::InvalidInput("Failed to extract models".to_string()))?
.into_inner()
.expect("operation should succeed");
Ok(MultiOutputRegressor {
state: MultiOutputRegressorTrained {
target_models: final_models,
n_targets,
n_features,
},
n_jobs: Some(n_jobs as i32),
})
}
}
impl MultiOutputRegressor<MultiOutputRegressorTrained> {
pub fn n_targets(&self) -> usize {
self.state.n_targets
}
}
impl Predict<ArrayView2<'_, Float>, Array2<f64>>
for MultiOutputRegressor<MultiOutputRegressorTrained>
{
#[allow(non_snake_case)]
fn predict(&self, X: &ArrayView2<'_, Float>) -> SklResult<Array2<f64>> {
let X = X.to_owned();
let (n_samples, n_features) = X.dim();
if n_features != self.state.n_features {
return Err(SklearsError::InvalidInput(
"Number of features doesn't match training data".to_string(),
));
}
let mut predictions = Array2::<Float>::zeros((n_samples, self.state.n_targets));
for target_idx in 0..self.state.n_targets {
if let Some((weights, bias)) = self.state.target_models.get(&target_idx) {
for (sample_idx, sample) in X.axis_iter(Axis(0)).enumerate() {
let prediction: f64 = sample
.iter()
.zip(weights.iter())
.map(|(&x, &w)| x * w)
.sum::<f64>()
+ bias;
predictions[[sample_idx, target_idx]] = prediction;
}
}
}
Ok(predictions)
}
}
#[derive(Debug, Clone)]
pub struct MultiOutputClassifierTrained {
pub classes_per_target: Vec<Vec<i32>>,
pub target_models: HashMap<usize, HashMap<i32, Array1<f64>>>,
pub n_targets: usize,
pub n_features: usize,
}
#[derive(Debug, Clone)]
pub struct MultiOutputRegressorTrained {
pub target_models: HashMap<usize, (Array1<f64>, f64)>,
pub n_targets: usize,
pub n_features: usize,
}
#[allow(non_snake_case)]
#[cfg(test)]
mod tests {
use super::*;
use approx::assert_abs_diff_eq;
use scirs2_core::ndarray::array;
use std::time::Instant;
#[test]
#[allow(non_snake_case)]
fn test_parallel_multi_output_classifier() {
let X = array![
[1.0, 2.0, 3.0],
[2.0, 3.0, 4.0],
[3.0, 4.0, 5.0],
[4.0, 5.0, 6.0],
[5.0, 6.0, 7.0],
[6.0, 7.0, 8.0]
];
let y = array![
[0, 1, 0],
[1, 0, 1],
[0, 1, 0],
[1, 0, 1],
[0, 1, 0],
[1, 0, 1]
];
let classifier_parallel = MultiOutputClassifier::new().n_jobs(Some(2));
let trained_parallel = classifier_parallel
.fit(&X.view(), &y)
.expect("model fitting should succeed");
let classifier_sequential = MultiOutputClassifier::new().n_jobs(Some(1));
let trained_sequential = classifier_sequential
.fit(&X.view(), &y)
.expect("model fitting should succeed");
assert_eq!(trained_parallel.n_targets(), trained_sequential.n_targets());
assert_eq!(
trained_parallel.classes().len(),
trained_sequential.classes().len()
);
let pred_parallel = trained_parallel
.predict(&X.view())
.expect("prediction should succeed");
let pred_sequential = trained_sequential
.predict(&X.view())
.expect("prediction should succeed");
assert_eq!(pred_parallel.shape(), pred_sequential.shape());
assert_eq!(pred_parallel.shape(), &[6, 3]);
}
#[test]
#[allow(non_snake_case)]
fn test_parallel_multi_output_regressor() {
let X = array![
[1.0, 2.0, 3.0],
[2.0, 3.0, 4.0],
[3.0, 4.0, 5.0],
[4.0, 5.0, 6.0],
[5.0, 6.0, 7.0],
[6.0, 7.0, 8.0]
];
let y = array![
[1.5, 2.5, 3.5],
[2.5, 3.5, 4.5],
[3.5, 4.5, 5.5],
[4.5, 5.5, 6.5],
[5.5, 6.5, 7.5],
[6.5, 7.5, 8.5]
];
let regressor_parallel = MultiOutputRegressor::new().n_jobs(Some(2));
let trained_parallel = regressor_parallel
.fit(&X.view(), &y)
.expect("model fitting should succeed");
let regressor_sequential = MultiOutputRegressor::new().n_jobs(Some(1));
let trained_sequential = regressor_sequential
.fit(&X.view(), &y)
.expect("model fitting should succeed");
assert_eq!(trained_parallel.n_targets(), trained_sequential.n_targets());
let pred_parallel = trained_parallel
.predict(&X.view())
.expect("prediction should succeed");
let pred_sequential = trained_sequential
.predict(&X.view())
.expect("prediction should succeed");
assert_eq!(pred_parallel.shape(), pred_sequential.shape());
assert_eq!(pred_parallel.shape(), &[6, 3]);
for i in 0..pred_parallel.nrows() {
for j in 0..pred_parallel.ncols() {
assert_abs_diff_eq!(
pred_parallel[[i, j]],
pred_sequential[[i, j]],
epsilon = 1e-10
);
}
}
}
#[test]
fn test_parallel_training_performance_classifier() {
let n_samples = 1000;
let n_features = 50;
let n_targets = 20;
let mut X = Array2::<Float>::zeros((n_samples, n_features));
let mut y = Array2::<i32>::zeros((n_samples, n_targets));
for i in 0..n_samples {
for j in 0..n_features {
X[[i, j]] = (i * j) as Float * 0.01;
}
for j in 0..n_targets {
y[[i, j]] = ((i + j) % 2) as i32;
}
}
let start_sequential = Instant::now();
let classifier_sequential = MultiOutputClassifier::new().n_jobs(Some(1));
let trained_sequential = classifier_sequential
.fit(&X.view(), &y)
.expect("model fitting should succeed");
let sequential_time = start_sequential.elapsed();
let start_parallel = Instant::now();
let classifier_parallel = MultiOutputClassifier::new().n_jobs(Some(4));
let trained_parallel = classifier_parallel
.fit(&X.view(), &y)
.expect("model fitting should succeed");
let parallel_time = start_parallel.elapsed();
assert_eq!(trained_parallel.n_targets(), n_targets);
assert_eq!(trained_sequential.n_targets(), n_targets);
let pred_parallel = trained_parallel
.predict(&X.view())
.expect("prediction should succeed");
let pred_sequential = trained_sequential
.predict(&X.view())
.expect("prediction should succeed");
assert_eq!(pred_parallel.shape(), pred_sequential.shape());
println!(
"Sequential time: {:?}, Parallel time: {:?}",
sequential_time, parallel_time
);
}
#[test]
fn test_parallel_training_performance_regressor() {
let n_samples = 1000;
let n_features = 50;
let n_targets = 20;
let mut X = Array2::<Float>::zeros((n_samples, n_features));
let mut y = Array2::<f64>::zeros((n_samples, n_targets));
for i in 0..n_samples {
for j in 0..n_features {
X[[i, j]] = (i * j) as Float * 0.01;
}
for j in 0..n_targets {
y[[i, j]] = (i + j) as f64 * 0.1;
}
}
let start_sequential = Instant::now();
let regressor_sequential = MultiOutputRegressor::new().n_jobs(Some(1));
let trained_sequential = regressor_sequential
.fit(&X.view(), &y)
.expect("model fitting should succeed");
let sequential_time = start_sequential.elapsed();
let start_parallel = Instant::now();
let regressor_parallel = MultiOutputRegressor::new().n_jobs(Some(4));
let trained_parallel = regressor_parallel
.fit(&X.view(), &y)
.expect("model fitting should succeed");
let parallel_time = start_parallel.elapsed();
assert_eq!(trained_parallel.n_targets(), n_targets);
assert_eq!(trained_sequential.n_targets(), n_targets);
let pred_parallel = trained_parallel
.predict(&X.view())
.expect("prediction should succeed");
let pred_sequential = trained_sequential
.predict(&X.view())
.expect("prediction should succeed");
assert_eq!(pred_parallel.shape(), pred_sequential.shape());
println!(
"Sequential time: {:?}, Parallel time: {:?}",
sequential_time, parallel_time
);
}
#[test]
#[allow(non_snake_case)]
fn test_parallel_training_thread_safety() {
let X = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0], [4.0, 5.0]];
let y_class = array![[0, 1], [1, 0], [0, 1], [1, 0]];
let y_reg = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0], [4.0, 5.0]];
for _ in 0..10 {
let classifier = MultiOutputClassifier::new().n_jobs(Some(2));
let trained = classifier
.fit(&X.view(), &y_class)
.expect("model fitting should succeed");
let predictions = trained
.predict(&X.view())
.expect("prediction should succeed");
assert_eq!(predictions.shape(), &[4, 2]);
let regressor = MultiOutputRegressor::new().n_jobs(Some(2));
let trained = regressor
.fit(&X.view(), &y_reg)
.expect("model fitting should succeed");
let predictions = trained
.predict(&X.view())
.expect("prediction should succeed");
assert_eq!(predictions.shape(), &[4, 2]);
}
}
#[test]
#[allow(non_snake_case)]
fn test_parallel_training_edge_cases() {
let X = array![[1.0, 2.0], [2.0, 3.0]];
let y_class = array![[0, 1], [1, 0]];
let y_reg = array![[1.0, 2.0], [2.0, 3.0]];
let classifier = MultiOutputClassifier::new().n_jobs(Some(10));
let trained = classifier
.fit(&X.view(), &y_class)
.expect("model fitting should succeed");
assert_eq!(trained.n_targets(), 2);
let regressor = MultiOutputRegressor::new().n_jobs(Some(10));
let trained = regressor
.fit(&X.view(), &y_reg)
.expect("model fitting should succeed");
assert_eq!(trained.n_targets(), 2);
let y_single = array![[0], [1]];
let classifier_single = MultiOutputClassifier::new().n_jobs(Some(4));
let trained_single = classifier_single
.fit(&X.view(), &y_single)
.expect("model fitting should succeed");
assert_eq!(trained_single.n_targets(), 1);
}
#[test]
#[allow(non_snake_case)]
fn test_parallel_training_error_handling() {
let X = array![[1.0, 2.0], [2.0, 3.0]];
let y_mismatch = array![[0, 1, 0], [1, 0, 1], [0, 1, 0]];
let classifier = MultiOutputClassifier::new().n_jobs(Some(2));
let result = classifier.fit(&X.view(), &y_mismatch);
assert!(result.is_err());
let regressor = MultiOutputRegressor::new().n_jobs(Some(2));
let y_reg_mismatch = array![[1.0, 2.0, 3.0], [2.0, 3.0, 4.0], [3.0, 4.0, 5.0]];
let result = regressor.fit(&X.view(), &y_reg_mismatch);
assert!(result.is_err());
}
}