Skip to Content
DocumentationML Model Development Guide

ML Model Development Guide

This comprehensive guide covers developing, training, and deploying machine learning models for brain-computer interface applications using NeuraScale’s ML platform.

Overview

NeuraScale provides a complete machine learning development environment specifically designed for neural data analysis, featuring:

  • Pre-trained Models: Ready-to-use models for common BCI tasks 🚀 Coming Soon
  • Custom Model Training: Build domain-specific models 🔧 Beta
  • Real-time Inference: Deploy models for live neural data processing 📅 Planned
  • Model Management: Version control, A/B testing, and deployment pipelines 📅 Planned
  • Performance Monitoring: Track model performance in production 📅 Planned

ML Platform Status

FeatureStatusDescription
Model Framework✓ AvailablePyTorch-based architecture implementations
EEGNet Model🔧 BetaTraining on motor imagery datasets
Pre-trained Weights🚀 Coming SoonValidated models for common tasks
Real-time Inference📅 PlannedSub-100ms latency prediction engine
Vertex AI Integration📅 PlannedCloud-based training and deployment
AutoML Features📅 PlannedAutomated hyperparameter tuning

The ML platform is under active development. Model architectures are implemented and being trained on standard BCI datasets. Pre-trained models will be available soon.

Supported ML Models

Pre-trained Models Available

NeuraScale provides several state-of-the-art neural network architectures optimized for BCI applications:

1. EEGNet (Lawhern et al., 2018)

  • Purpose: Compact CNN for EEG classification
  • Applications: Motor imagery, P300 detection, error-related potentials
  • Parameters: ~2,600 (extremely lightweight)
  • Accuracy: 75-85% on motor imagery tasks
  • Channels: 8-64 channels supported

2. ShallowConvNet (Schirrmeister et al., 2017)

  • Purpose: Shallow CNN optimized for oscillatory features
  • Applications: Motor imagery, sleep stage classification
  • Parameters: ~47,000
  • Accuracy: 80-90% on motor imagery tasks
  • Channels: 16-128 channels supported

3. DeepConvNet (Schirrmeister et al., 2017)

  • Purpose: Deep CNN for complex pattern recognition
  • Applications: Seizure detection, emotion recognition
  • Parameters: ~287,000
  • Accuracy: 85-92% on complex tasks
  • Channels: 32-256 channels supported

4. Hybrid Models

  • CNN-LSTM: For temporal sequence analysis
  • Transformer-based: For long-range dependencies
  • Graph Neural Networks: For spatial relationships

Task-Specific Models

Motor Imagery Classification

  • 2-class (left/right hand)
  • 4-class (left/right hand, feet, tongue)
  • Multi-class (up to 10 movements)

Clinical Applications

  • Seizure Detection: Real-time epileptic seizure detection
  • Sleep Stage Classification: 5-stage sleep analysis
  • Cognitive Load Assessment: Mental workload estimation
  • Emotion Recognition: Valence/arousal classification

Communication & Control

  • P300 Speller: Character selection for communication
  • SSVEP Decoder: Steady-state visual evoked potentials
  • ERD/ERS Detection: Event-related (de)synchronization

Getting Started with ML

Setting up the ML Environment

from neurascale.ml import MLClient, ModelRegistry
from neurascale.ml.preprocessing import NeuraPreprocessor
from neurascale.ml.models import BCIClassifier
from neurascale.ml.training import TrainingPipeline
 
# Initialize ML client
ml_client = MLClient(
    api_key="your-api-key",
    compute_backend="gpu",  # cpu, gpu, tpu
    distributed=False
)
 
# Configure ML environment
ml_config = {
    "environment": {
        "framework": "pytorch",  # pytorch, tensorflow, sklearn
        "accelerator": "cuda",   # cuda, cpu, mps (Apple Silicon)
        "precision": "mixed",    # full, mixed, half
        "memory_optimization": True
    },
    "data": {
        "batch_size": 32,
        "num_workers": 4,
        "prefetch_factor": 2,
        "pin_memory": True
    },
    "training": {
        "checkpoint_frequency": 100,
        "early_stopping": True,
        "patience": 10,
        "metric": "accuracy"
    }
}
 
await ml_client.configure(ml_config)

Model Registry and Pre-trained Models

# Access model registry
registry = ModelRegistry(ml_client)
 
# List available pre-trained models
available_models = await registry.list_models()
print("Available pre-trained models:")
for model in available_models:
    print(f"  {model.name}: {model.description}")
    print(f"    Task: {model.task_type}")
    print(f"    Accuracy: {model.validation_accuracy:.2%}")
    print(f"    Channels: {model.channel_requirements}")
 
# Load a pre-trained model
motor_imagery_model = await registry.load_model(
    model_name="motor_imagery_4class_v2.1",
    version="latest"
)
 
print(f"Model loaded: {motor_imagery_model.name}")
print(f"Classes: {motor_imagery_model.classes}")
print(f"Input shape: {motor_imagery_model.input_shape}")

Data Preparation and Preprocessing

Neural Data Preprocessing Pipeline

Basic Neural Data Preprocessing

from neurascale.ml.preprocessing import (
    NeuraPreprocessor,
    SignalProcessor,
    EpochExtractor
)
 
# Initialize preprocessor
preprocessor = NeuraPreprocessor(
    sample_rate=250,
    channels=["C3", "C4", "Cz", "FC1", "FC2", "CP1", "CP2"],
    target_length=2.0  # 2 seconds
)
 
# Configure preprocessing pipeline
preprocessing_steps = [
    {
        "name": "resampling",
        "type": "resample",
        "parameters": {"target_rate": 250}
    },
    {
        "name": "filtering",
        "type": "bandpass",
        "parameters": {
            "low_freq": 8,
            "high_freq": 30,
            "filter_type": "butterworth",
            "order": 4
        }
    },
    {
        "name": "artifact_removal",
        "type": "ica",
        "parameters": {
            "n_components": 7,
            "method": "fastica",
            "max_iter": 200
        }
    },
    {
        "name": "normalization",
        "type": "z_score",
        "parameters": {"axis": "time"}
    }
]
 
# Apply preprocessing
async def preprocess_data(raw_data):
    """Preprocess raw neural data for ML training"""
 
    processed_data = raw_data.copy()
 
    for step in preprocessing_steps:
        print(f"Applying {step['name']}...")
 
        if step["type"] == "resample":
            processed_data = await preprocessor.resample(
                processed_data,
                step["parameters"]["target_rate"]
            )
 
        elif step["type"] == "bandpass":
            processed_data = await preprocessor.bandpass_filter(
                processed_data,
                step["parameters"]["low_freq"],
                step["parameters"]["high_freq"],
                order=step["parameters"]["order"]
            )
 
        elif step["type"] == "ica":
            processed_data = await preprocessor.apply_ica(
                processed_data,
                n_components=step["parameters"]["n_components"]
            )
 
        elif step["type"] == "z_score":
            processed_data = await preprocessor.z_score_normalize(
                processed_data,
                axis=step["parameters"]["axis"]
            )
 
    return processed_data
 
# Example usage
raw_eeg_data = await ml_client.data.load_session("session_123")
processed_data = await preprocess_data(raw_eeg_data)
 
print(f"Raw data shape: {raw_eeg_data.shape}")
print(f"Processed data shape: {processed_data.shape}")

Model Development

Building Custom BCI Models

Deep Learning Models for BCI

import torch
import torch.nn as nn
import torch.nn.functional as F
from neurascale.ml.models import BaseBCIModel
 
class EEGNet(BaseBCIModel):
    """EEGNet architecture for EEG classification"""
 
    def __init__(self, n_classes=4, n_channels=64, n_timepoints=128):
        super(EEGNet, self).__init__()
 
        self.n_classes = n_classes
        self.n_channels = n_channels
        self.n_timepoints = n_timepoints
 
        # Temporal convolution
        self.temporal_conv = nn.Conv2d(
            1, 16,
            kernel_size=(1, 64),
            padding=(0, 32),
            bias=False
        )
        self.temporal_bn = nn.BatchNorm2d(16)
 
        # Spatial convolution (depthwise)
        self.spatial_conv = nn.Conv2d(
            16, 32,
            kernel_size=(n_channels, 1),
            groups=16,
            bias=False
        )
        self.spatial_bn = nn.BatchNorm2d(32)
 
        # Separable convolution
        self.separable_conv1 = nn.Conv2d(
            32, 32,
            kernel_size=(1, 16),
            padding=(0, 8),
            groups=32,
            bias=False
        )
        self.separable_conv2 = nn.Conv2d(32, 16, kernel_size=1, bias=False)
        self.separable_bn = nn.BatchNorm2d(16)
 
        # Calculate output size after convolutions
        self.feature_size = self._get_conv_output_size()
 
        # Classification head
        self.classifier = nn.Linear(self.feature_size, n_classes)
        self.dropout = nn.Dropout(0.25)
 
    def _get_conv_output_size(self):
        """Calculate the output size after convolutions"""
        with torch.no_grad():
            x = torch.zeros(1, 1, self.n_channels, self.n_timepoints)
            x = self._forward_features(x)
            return x.numel()
 
    def _forward_features(self, x):
        """Forward pass through feature extraction layers"""
 
        # Temporal convolution
        x = self.temporal_conv(x)
        x = self.temporal_bn(x)
 
        # Spatial convolution
        x = self.spatial_conv(x)
        x = self.spatial_bn(x)
        x = F.elu(x)
        x = F.avg_pool2d(x, kernel_size=(1, 4))
        x = self.dropout(x)
 
        # Separable convolution
        x = self.separable_conv1(x)
        x = self.separable_conv2(x)
        x = self.separable_bn(x)
        x = F.elu(x)
        x = F.avg_pool2d(x, kernel_size=(1, 8))
        x = self.dropout(x)
 
        return x
 
    def forward(self, x):
        # Input shape: (batch_size, n_channels, n_timepoints)
        # Add channel dimension for conv2d
        x = x.unsqueeze(1)  # (batch_size, 1, n_channels, n_timepoints)
 
        # Feature extraction
        x = self._forward_features(x)
 
        # Flatten for classification
        x = x.view(x.size(0), -1)
 
        # Classification
        x = self.classifier(x)
 
        return x
 
class ShallowConvNet(BaseBCIModel):
    """Shallow ConvNet for motor imagery classification"""
 
    def __init__(self, n_classes=4, n_channels=64, n_timepoints=1000):
        super(ShallowConvNet, self).__init__()
 
        self.n_classes = n_classes
        self.n_channels = n_channels
        self.n_timepoints = n_timepoints
 
        # Temporal convolution
        self.temporal_conv = nn.Conv2d(
            1, 40,
            kernel_size=(1, 25),
            bias=False
        )
 
        # Spatial convolution
        self.spatial_conv = nn.Conv2d(
            40, 40,
            kernel_size=(n_channels, 1),
            bias=False
        )
 
        self.bn = nn.BatchNorm2d(40)
        self.dropout = nn.Dropout(0.5)
 
        # Calculate feature size
        self.feature_size = self._get_conv_output_size()
 
        # Classification head
        self.classifier = nn.Linear(self.feature_size, n_classes)
 
    def _get_conv_output_size(self):
        with torch.no_grad():
            x = torch.zeros(1, 1, self.n_channels, self.n_timepoints)
            x = self._forward_features(x)
            return x.numel()
 
    def _forward_features(self, x):
        # Temporal convolution
        x = self.temporal_conv(x)
 
        # Spatial convolution
        x = self.spatial_conv(x)
 
        # Batch normalization
        x = self.bn(x)
 
        # Square activation
        x = x ** 2
 
        # Average pooling
        x = F.avg_pool2d(x, kernel_size=(1, 75), stride=(1, 15))
 
        # Log activation
        x = torch.log(torch.clamp(x, min=1e-6))
 
        # Dropout
        x = self.dropout(x)
 
        return x
 
    def forward(self, x):
        x = x.unsqueeze(1)  # Add channel dimension
        x = self._forward_features(x)
        x = x.view(x.size(0), -1)  # Flatten
        x = self.classifier(x)
        return x
 
class DeepConvNet(BaseBCIModel):
    """Deep ConvNet for EEG classification"""
 
    def __init__(self, n_classes=4, n_channels=64, n_timepoints=1000):
        super(DeepConvNet, self).__init__()
 
        self.n_classes = n_classes
        self.n_channels = n_channels
 
        # Block 1
        self.conv1 = nn.Conv2d(1, 25, kernel_size=(1, 10), bias=False)
        self.conv2 = nn.Conv2d(25, 25, kernel_size=(n_channels, 1), bias=False)
        self.bn1 = nn.BatchNorm2d(25)
 
        # Block 2
        self.conv3 = nn.Conv2d(25, 50, kernel_size=(1, 10), bias=False)
        self.bn2 = nn.BatchNorm2d(50)
 
        # Block 3
        self.conv4 = nn.Conv2d(50, 100, kernel_size=(1, 10), bias=False)
        self.bn3 = nn.BatchNorm2d(100)
 
        # Block 4
        self.conv5 = nn.Conv2d(100, 200, kernel_size=(1, 10), bias=False)
        self.bn4 = nn.BatchNorm2d(200)
 
        self.dropout = nn.Dropout(0.5)
 
        # Calculate feature size
        self.feature_size = self._get_conv_output_size()
 
        # Classification
        self.classifier = nn.Linear(self.feature_size, n_classes)
 
    def _conv_block(self, x, conv_layer, bn_layer, pool_size=(1, 3)):
        x = conv_layer(x)
        x = bn_layer(x)
        x = F.elu(x)
        x = F.max_pool2d(x, kernel_size=pool_size)
        x = self.dropout(x)
        return x
 
    def _get_conv_output_size(self):
        with torch.no_grad():
            x = torch.zeros(1, 1, self.n_channels, 1000)  # Use fixed size for calculation
            x = self._forward_features(x)
            return x.numel()
 
    def _forward_features(self, x):
        # Block 1
        x = self.conv1(x)
        x = self.conv2(x)
        x = self.bn1(x)
        x = F.elu(x)
        x = F.max_pool2d(x, kernel_size=(1, 3))
        x = self.dropout(x)
 
        # Block 2
        x = self._conv_block(x, self.conv3, self.bn2)
 
        # Block 3
        x = self._conv_block(x, self.conv4, self.bn3)
 
        # Block 4
        x = self._conv_block(x, self.conv5, self.bn4)
 
        return x
 
    def forward(self, x):
        x = x.unsqueeze(1)
        x = self._forward_features(x)
        x = x.view(x.size(0), -1)
        x = self.classifier(x)
        return x

Model Training and Evaluation

Training Pipeline

from neurascale.ml.training import BCITrainingPipeline
from neurascale.ml.evaluation import BCIEvaluator
 
class ComprehensiveTrainingPipeline:
    def __init__(self, config):
        self.config = config
        self.pipeline = BCITrainingPipeline(config)
        self.evaluator = BCIEvaluator()
 
    async def train_model(self, model, train_data, val_data, test_data):
        """Comprehensive model training with monitoring"""
 
        # Setup training configuration
        training_config = {
            "optimizer": {
                "type": "adamw",
                "lr": 1e-3,
                "weight_decay": 1e-5,
                "betas": [0.9, 0.999]
            },
            "scheduler": {
                "type": "cosine_annealing",
                "T_max": 100,
                "eta_min": 1e-6
            },
            "training": {
                "epochs": 100,
                "batch_size": 32,
                "early_stopping": {
                    "patience": 15,
                    "monitor": "val_accuracy",
                    "min_delta": 0.001
                },
                "gradient_clipping": 1.0
            },
            "regularization": {
                "dropout": 0.25,
                "label_smoothing": 0.1,
                "mixup_alpha": 0.2
            }
        }
 
        # Initialize training components
        trainer = await self.pipeline.create_trainer(model, training_config)
 
        # Setup monitoring
        monitor = await self.setup_training_monitor()
 
        # Training loop with comprehensive monitoring
        best_model = None
        best_accuracy = 0
        patience_counter = 0
 
        for epoch in range(training_config["training"]["epochs"]):
            print(f"\nEpoch {epoch + 1}/{training_config['training']['epochs']}")
 
            # Training phase
            train_metrics = await trainer.train_epoch(train_data)
 
            # Validation phase
            val_metrics = await trainer.validate_epoch(val_data)
 
            # Update learning rate
            trainer.scheduler.step()
 
            # Log metrics
            await monitor.log_metrics(epoch, train_metrics, val_metrics)
 
            # Check for improvement
            if val_metrics["accuracy"] > best_accuracy:
                best_accuracy = val_metrics["accuracy"]
                best_model = copy.deepcopy(model)
                patience_counter = 0
 
                # Save checkpoint
                await trainer.save_checkpoint(
                    epoch, model,
                    f"best_model_epoch_{epoch}.pth"
                )
            else:
                patience_counter += 1
 
            # Early stopping
            if patience_counter >= training_config["training"]["early_stopping"]["patience"]:
                print(f"Early stopping at epoch {epoch + 1}")
                break
 
            # Print progress
            print(f"Train Loss: {train_metrics['loss']:.4f}, "
                  f"Train Acc: {train_metrics['accuracy']:.4f}")
            print(f"Val Loss: {val_metrics['loss']:.4f}, "
                  f"Val Acc: {val_metrics['accuracy']:.4f}")
 
        # Final evaluation
        test_results = await self.evaluator.comprehensive_evaluation(
            best_model, test_data
        )
 
        return {
            "model": best_model,
            "training_history": monitor.get_history(),
            "test_results": test_results,
            "best_accuracy": best_accuracy
        }
 
# Model evaluation and validation
async def comprehensive_model_evaluation(model, test_data, test_labels):
    """Comprehensive evaluation of BCI model"""
 
    evaluator = BCIEvaluator()
 
    # Basic metrics
    predictions = await model.predict(test_data)
    basic_metrics = await evaluator.calculate_basic_metrics(
        test_labels, predictions
    )
 
    # Advanced metrics
    advanced_metrics = await evaluator.calculate_advanced_metrics(
        test_labels, predictions, test_data
    )
 
    # Cross-validation
    cv_results = await evaluator.cross_validation(
        model, test_data, test_labels, cv_folds=5
    )
 
    # Statistical significance tests
    significance_tests = await evaluator.statistical_tests(
        test_labels, predictions
    )
 
    # Generate comprehensive report
    evaluation_report = {
        "basic_metrics": basic_metrics,
        "advanced_metrics": advanced_metrics,
        "cross_validation": cv_results,
        "significance_tests": significance_tests,
        "confusion_matrix": await evaluator.plot_confusion_matrix(
            test_labels, predictions
        ),
        "roc_curves": await evaluator.plot_roc_curves(
            test_labels, predictions
        )
    }
 
    return evaluation_report

Model Deployment

Real-time Inference

from neurascale.ml.deployment import RealTimeInference
from neurascale.ml.serving import ModelServer
 
class BCIModelServer:
    def __init__(self, model_path, config):
        self.model_path = model_path
        self.config = config
        self.model = None
        self.preprocessor = None
 
    async def initialize(self):
        """Initialize model server"""
 
        # Load trained model
        self.model = await self.load_model(self.model_path)
 
        # Setup preprocessing pipeline
        self.preprocessor = await self.setup_preprocessing()
 
        # Initialize inference engine
        self.inference_engine = RealTimeInference(
            model=self.model,
            preprocessor=self.preprocessor,
            config=self.config
        )
 
        print("Model server initialized successfully")
 
    async def process_real_time_data(self, data_stream):
        """Process real-time neural data stream"""
 
        async for data_packet in data_stream:
            try:
                # Preprocess data
                processed_data = await self.preprocessor.process(data_packet)
 
                # Make prediction
                prediction = await self.inference_engine.predict(processed_data)
 
                # Post-process prediction
                result = await self.post_process_prediction(prediction)
 
                # Send result
                await self.send_prediction_result(result)
 
            except Exception as e:
                print(f"Error processing data packet: {e}")
                await self.handle_inference_error(e)
 
    async def batch_inference(self, batch_data):
        """Process batch of data for offline analysis"""
 
        results = []
 
        for data in batch_data:
            processed_data = await self.preprocessor.process(data)
            prediction = await self.inference_engine.predict(processed_data)
            results.append(prediction)
 
        return results
 
# Deploy model with NeuraScale
async def deploy_bci_model():
    """Deploy trained BCI model for production use"""
 
    deployment_config = {
        "model": {
            "path": "models/motor_imagery_model.pth",
            "type": "pytorch",
            "version": "1.0.0"
        },
        "serving": {
            "batch_size": 1,
            "max_latency": 50,  # milliseconds
            "device": "cuda",
            "precision": "fp16"
        },
        "monitoring": {
            "enable_metrics": True,
            "log_predictions": True,
            "alert_on_degradation": True
        }
    }
 
    # Deploy to NeuraScale platform
    deployment = await ml_client.deploy_model(
        model_name="motor_imagery_v1",
        config=deployment_config
    )
 
    print(f"Model deployed: {deployment.endpoint_url}")
 
    return deployment

This ML development guide provides a comprehensive framework for building, training, and deploying machine learning models for BCI applications. For specific implementation details and advanced techniques, refer to the neurascale.ml module documentation.

Best Practices

Model Development Guidelines

  1. Data Quality: Always validate data quality before training
  2. Cross-Subject Validation: Test models across different subjects
  3. Robust Preprocessing: Implement comprehensive artifact removal
  4. Feature Engineering: Combine multiple feature domains for better performance
  5. Model Validation: Use proper cross-validation and statistical testing
  6. Real-time Constraints: Consider latency requirements for online applications
  7. Transfer Learning: Leverage pre-trained models for faster development
  8. Ensemble Methods: Combine multiple models for improved reliability

Performance Optimization

# Optimization tips for production deployment
optimization_config = {
    "model_optimization": {
        "quantization": "int8",
        "pruning": 0.3,  # 30% pruning
        "knowledge_distillation": True
    },
    "inference_optimization": {
        "batch_size": 8,
        "use_tensorrt": True,
        "enable_amp": True,  # Automatic Mixed Precision
        "memory_optimization": True
    },
    "caching": {
        "model_cache": True,
        "feature_cache": True,
        "prediction_cache": False  # For real-time applications
    }
}

This comprehensive ML development guide provides the foundation for building sophisticated brain-computer interface applications with NeuraScale’s machine learning platform.

Last updated on
Clinical WorkflowsPerformance Tuning