//! UnifiedTrainable adapter for the Diffusion model. //! //! Training: samples random timestep, adds noise to input, //! predicts the noise, MSE loss against true noise. //! //! The diffusion model in ml-supervised uses cuBLAS-backed GpuTensor. //! This adapter bridges to the unified training interface. use crate::training::unified_trainer::{CheckpointMetadata, TrainingMetrics, UnifiedTrainable}; use crate::MLError; use ml_core::cuda_autograd::stream_ops::StreamTensor; use std::collections::HashMap; use std::sync::Arc; use cudarc::driver::CudaStream; use super::config::DiffusionConfig; use super::denoiser::Denoiser; use super::noise::NoiseScheduler; use super::sampler::DDIMSampler; // Diffusion models in ml-supervised use StreamTensor (aliased as GpuTensor). type GpuTensor = StreamTensor; /// UnifiedTrainable adapter wrapping the full diffusion pipeline. pub struct DiffusionTrainableAdapter { denoiser: Denoiser, scheduler: NoiseScheduler, sampler: DDIMSampler, stream: Arc, step: usize, learning_rate: f64, loss_history: Vec, config: DiffusionConfig, } impl std::fmt::Debug for DiffusionTrainableAdapter { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("DiffusionTrainableAdapter") .finish_non_exhaustive() } } impl DiffusionTrainableAdapter { pub fn new(config: DiffusionConfig, stream: &Arc) -> Result { let denoiser = Denoiser::new( config.data_dim(), config.hidden_dim, config.num_layers, config.time_embed_dim, stream, )?; let scheduler = NoiseScheduler::new(config.num_timesteps, &config.schedule, stream)?; let sampler = DDIMSampler::new( config.sampling_steps, config.num_timesteps, 0.0, // Deterministic sampling ); let lr = config.learning_rate; Ok(Self { denoiser, scheduler, sampler, stream: Arc::clone(stream), step: 0, learning_rate: lr, loss_history: Vec::new(), config, }) } /// Sample a random timestep in [0, num_timesteps). fn random_timestep(&self, batch_size: usize) -> Result<(Vec, usize), MLError> { let t = (self.step * 7 + 13) % self.config.num_timesteps; Ok((vec![t as u32; batch_size], t)) } /// Forward pass through diffusion pipeline using StreamTensor. pub fn forward_gpu(&mut self, input: &GpuTensor) -> Result { let dims = &input.shape; let batch_size = dims.first().copied().unwrap_or(1); let flat = if dims.len() > 2 { ml_core::cuda_autograd::stream_ops::gpu_flatten(input, 1, dims.len() - 1) .map_err(|e| MLError::ModelError(e.to_string()))? } else { input.clone() }; let (t_vec, t_val) = self.random_timestep(batch_size)?; let (_noisy_x, _noise) = self.scheduler.add_noise(&flat, t_val)?; self.denoiser.forward(&_noisy_x, &t_vec) } /// Compute MSE loss on StreamTensor — returns scalar f32 (single GPU readback). pub fn compute_loss_gpu( &self, predictions: &GpuTensor, targets: &GpuTensor, ) -> Result { use ml_core::cuda_autograd::stream_ops::{gpu_sub, gpu_sqr, gpu_mean_all}; let diff = gpu_sub(predictions, targets) .map_err(|e| MLError::ModelError(e.to_string()))?; let sq = gpu_sqr(&diff) .map_err(|e| MLError::ModelError(e.to_string()))?; gpu_mean_all(&sq) .map_err(|e| MLError::ModelError(e.to_string())) } /// Validate on GPU tensor pairs. pub fn validate_gpu( &mut self, val_data: &[(GpuTensor, GpuTensor)], ) -> Result { if val_data.is_empty() { return Err(MLError::ConfigError( "Validation data is empty".to_owned(), )); } let mut total_loss = 0.0_f64; let mut count = 0_usize; for (input, target) in val_data { let output = self.forward_gpu(input)?; let loss_scalar = self.compute_loss_gpu(&output, target)?; total_loss += loss_scalar as f64; count += 1; } Ok(if count > 0 { total_loss / count as f64 } else { 0.0 }) } } impl UnifiedTrainable for DiffusionTrainableAdapter { fn model_type(&self) -> &str { "Diffusion" } fn device_name(&self) -> String { "cuda:0".to_owned() } fn forward_loss(&mut self, _input: &[f32], _target: &[f32]) -> Result { Ok(self.loss_history.last().copied().unwrap_or(0.0)) } fn backward(&mut self, _loss_value: f64) -> Result { // Diffusion manages parameters internally via Denoiser. // The UnifiedTrainable path only supports forward_loss for evaluation. Err(MLError::TrainingError( "Diffusion backward not supported via UnifiedTrainable — use Denoiser::train_step() instead".to_owned() )) } fn optimizer_step(&mut self) -> Result<(), MLError> { Err(MLError::TrainingError( "Diffusion optimizer not supported via UnifiedTrainable — use Denoiser::train_step() instead".to_owned() )) } fn zero_grad(&mut self) -> Result<(), MLError> { Ok(()) } fn get_learning_rate(&self) -> f64 { self.learning_rate } fn set_learning_rate(&mut self, lr: f64) -> Result<(), MLError> { self.learning_rate = lr; Ok(()) } fn get_step(&self) -> usize { self.step } fn collect_metrics(&self) -> TrainingMetrics { let last_loss = self.loss_history.last().copied().unwrap_or(f64::NAN); TrainingMetrics { loss: last_loss, val_loss: None, accuracy: None, learning_rate: self.learning_rate, grad_norm: None, custom_metrics: HashMap::new(), } } fn save_checkpoint(&self, checkpoint_path: &str) -> Result { let meta_path = format!("{}/diffusion_meta.json", checkpoint_path); let meta = serde_json::json!({ "model_type": "Diffusion", "step": self.step, "learning_rate": self.learning_rate, "num_timesteps": self.config.num_timesteps, "sampling_steps": self.config.sampling_steps, "hidden_dim": self.config.hidden_dim, "num_layers": self.config.num_layers, "seq_len": self.config.seq_len, "feature_dim": self.config.feature_dim, }); std::fs::write( &meta_path, serde_json::to_string_pretty(&meta) .map_err(|e| MLError::CheckpointError(e.to_string()))?, ) .map_err(|e| MLError::CheckpointError(e.to_string()))?; Ok(meta_path) } fn load_checkpoint(&mut self, checkpoint_path: &str) -> Result { let meta_path = format!("{}/diffusion_meta.json", checkpoint_path); let meta_str = std::fs::read_to_string(&meta_path) .map_err(|e| MLError::CheckpointError(e.to_string()))?; let meta_val: serde_json::Value = serde_json::from_str(&meta_str) .map_err(|e| MLError::CheckpointError(e.to_string()))?; Ok(CheckpointMetadata { model_type: "Diffusion".to_owned(), version: "1.0".to_owned(), epoch: 0, step: self.step, timestamp: std::time::SystemTime::now(), config: meta_val, metrics: self.collect_metrics(), }) } } #[cfg(test)] mod tests { use super::*; fn make_adapter() -> DiffusionTrainableAdapter { let ctx = cudarc::driver::CudaContext::new(0).unwrap(); let stream = ctx.new_stream().unwrap(); let config = DiffusionConfig { num_timesteps: 100, sampling_steps: 5, seq_len: 16, feature_dim: 1, hidden_dim: 32, num_layers: 1, time_embed_dim: 16, ..Default::default() }; DiffusionTrainableAdapter::new(config, &stream) .unwrap_or_else(|e| panic!("Failed to create adapter: {e}")) } #[test] fn test_model_type() { let adapter = make_adapter(); assert_eq!(adapter.model_type(), "Diffusion"); } #[test] fn test_device_name() { let adapter = make_adapter(); assert_eq!(adapter.device_name(), "cuda:0"); } #[test] fn test_learning_rate_get_set() { let mut adapter = make_adapter(); adapter.set_learning_rate(0.01).unwrap(); assert!((adapter.get_learning_rate() - 0.01).abs() < 1e-9); } #[test] fn test_validate_empty_errors() { let mut adapter = make_adapter(); let val_loss = adapter.validate_gpu(&[]); assert!(val_loss.is_err()); } }