22. Performance Optimization

# Optimized for single-cell data
class SmartSparseStrategy(ConcatenationStrategyBase):
    def concatenate(self, sample_adatas: List[ad.AnnData], **kwargs) -> ConcatenationResult:
        """Memory-efficient sparse matrix concatenation."""
        # 1. Preserve sparsity throughout operation
        # 2. Batch-aware processing for technical replicates
        # 3. Automatic memory estimation and chunking
        # 4. Zero-copy operations where possible

        memory_info = self._estimate_memory_usage(sample_adatas)
        if not memory_info.can_proceed:
            raise MemoryLimitError(f"Insufficient memory: need {memory_info.required_gb:.1f}GB")

        return self._smart_sparse_concatenation(sample_adatas, **kwargs)

# For datasets exceeding memory limits
class MemoryEfficientStrategy(ConcatenationStrategyBase):
    def concatenate(self, sample_adatas: List[ad.AnnData], **kwargs) -> ConcatenationResult:
        """Chunked processing for large datasets."""
        # 1. Intelligent chunking based on available memory
        # 2. Progressive concatenation with garbage collection
        # 3. Disk-backed intermediate storage
        # 4. Real-time memory monitoring

        chunk_size = self._calculate_optimal_chunk_size()
        return self._chunked_concatenation(sample_adatas, chunk_size, **kwargs)

@dataclass
class MemoryInfo:
    """Memory information for concatenation planning."""
    available_gb: float
    required_gb: float
    can_proceed: bool
    recommended_strategy: Optional[ConcatenationStrategy] = None

def estimate_memory_usage(sample_adatas: List[ad.AnnData]) -> MemoryInfo:
    """Intelligent memory estimation for concatenation operations."""
    # Calculate memory requirements
    total_memory_needed = 0
    for adata in sample_adatas:
        if hasattr(adata.X, 'nnz'):  # Sparse matrix
            memory_estimate = adata.X.data.nbytes + adata.X.indices.nbytes + adata.X.indptr.nbytes
        else:  # Dense matrix
            memory_estimate = adata.X.nbytes

        # Account for intermediate processing overhead (2x factor)
        total_memory_needed += memory_estimate * 2

    # Get available system memory
    available_memory = psutil.virtual_memory().available

    return MemoryInfo(
        available_gb=available_memory / (1024**3),
        required_gb=total_memory_needed / (1024**3),
        can_proceed=total_memory_needed < available_memory * 0.8  # 80% safety margin
    )

class PreprocessingService:
    """Optimized preprocessing with memory management."""

    def filter_and_normalize_cells(
        self,
        adata: ad.AnnData,
        **params
    ) -> Tuple[ad.AnnData, Dict[str, Any]]:
        """Memory-efficient cell filtering and normalization."""

        # 1. Copy-on-write operations
        adata_processed = adata.copy()

        # 2. In-place operations where safe
        with self._memory_monitor():
            # Filter cells based on QC metrics
            sc.pp.filter_cells(adata_processed, min_genes=params.get('min_genes', 200))

            # Normalize with sparse-aware methods
            if issparse(adata_processed.X):
                sc.pp.normalize_total(adata_processed, target_sum=1e4, inplace=True)
                sc.pp.log1p(adata_processed, base=2)
            else:
                # Dense normalization pathway
                self._normalize_dense_matrix(adata_processed, params)

        # 3. Garbage collection after major operations
        gc.collect()

        return adata_processed, self._generate_processing_stats(adata, adata_processed)

class QualityService:
    """Optimized quality assessment with parallel processing."""

    def assess_data_quality(
        self,
        adata: ad.AnnData,
        **params
    ) -> Tuple[ad.AnnData, Dict[str, Any]]:
        """Parallel quality metric calculation."""

        adata_qc = adata.copy()

        # Parallel QC metric calculation
        with ThreadPoolExecutor(max_workers=os.cpu_count()) as executor:
            futures = {
                executor.submit(self._calculate_mitochondrial_metrics, adata_qc): 'mito',
                executor.submit(self._calculate_ribosomal_metrics, adata_qc): 'ribo',
                executor.submit(self._calculate_complexity_metrics, adata_qc): 'complexity',
                executor.submit(self._detect_outliers, adata_qc): 'outliers'
            }

            qc_results = {}
            for future in as_completed(futures):
                metric_type = futures[future]
                qc_results[metric_type] = future.result()

        # Combine results efficiently
        self._integrate_qc_results(adata_qc, qc_results)

        return adata_qc, qc_results

class CloudAwareCache:
    """Smart caching that adapts to client type."""

    def __init__(self, client):
        self.is_cloud = hasattr(client, 'list_workspace_files') and hasattr(client, 'session')
        self.cache = {}
        self.timeouts = {
            'commands': float('inf'),  # Commands never change
            'files': 60 if self.is_cloud else 10,  # Longer cache for cloud
            'workspace': 30 if self.is_cloud else 5
        }

    def get_or_fetch(self, key: str, fetch_func, category: str = 'default'):
        """Get cached value or fetch if expired with intelligent fallback."""
        current_time = time.time()
        timeout = self.timeouts.get(category, 10)

        # Check if cache is valid
        if (key not in self.cache or
            current_time - self.cache[key]['timestamp'] > timeout):
            try:
                # Fetch fresh data
                self.cache[key] = {
                    'data': fetch_func(),
                    'timestamp': current_time
                }
            except Exception as e:
                if self.is_cloud and self._is_network_error(e):
                    # Use stale cache for network issues
                    if key in self.cache:
                        logger.warning(f"Using stale cache due to network issue: {e}")
                        return self.cache[key]['data']
                raise e

        return self.cache[key]['data']

    def _is_network_error(self, exception: Exception) -> bool:
        """Detect network-related errors for cache fallback."""
        error_str = str(exception).lower()
        return any(keyword in error_str for keyword in
                  ['connection', 'timeout', 'network', 'unreachable'])

class LobsterContextualCompleter(Completer):
    """High-performance contextual completer."""

    def __init__(self, client):
        self.client = client
        self.adapter = LobsterClientAdapter(client)
        self.command_completer = LobsterCommandCompleter()
        self.file_completer = LobsterFileCompleter(client)

        # Performance optimizations
        self.completion_cache = {}
        self.last_context = None
        self.cache_timeout = 30  # seconds

    def get_completions(self, document: Document, complete_event: CompleteEvent):
        """Generate context-aware completions with performance optimization."""
        text = document.text_before_cursor.strip()

        # Cache hit optimization
        if text == self.last_context and self._cache_valid():
            return self.completion_cache.get(text, [])

        # Context-specific completion with caching
        if text.startswith('/') and ' ' not in text:
            # Command completion (cached indefinitely)
            completions = list(self.command_completer.get_completions(document, complete_event))
        elif any(text.startswith(cmd + ' ') for cmd in self.file_commands):
            # File completion (cached with timeout)
            completions = list(self._get_cached_file_completions(document, complete_event))
        else:
            # Other completions
            completions = []

        # Update cache
        self.completion_cache[text] = completions
        self.last_context = text
        self.cache_timestamp = time.time()

        return completions

def optimize_sparse_operations(adata: ad.AnnData) -> ad.AnnData:
    """Optimize operations for sparse matrices."""

    if not issparse(adata.X):
        # Convert to sparse if beneficial
        sparsity = 1.0 - np.count_nonzero(adata.X) / adata.X.size
        if sparsity > 0.7:  # 70% zeros threshold
            adata.X = sparse.csr_matrix(adata.X)
            logger.info(f"Converted to sparse matrix (sparsity: {sparsity:.1%})")

    # Optimize sparse matrix format
    if issparse(adata.X) and not isinstance(adata.X, sparse.csr_matrix):
        adata.X = adata.X.tocsr()  # CSR format for row operations

    return adata

def vectorized_quality_metrics(adata: ad.AnnData) -> Dict[str, np.ndarray]:
    """Vectorized calculation of quality metrics."""

    X = adata.X
    if issparse(X):
        # Sparse-optimized calculations
        n_genes = np.array((X > 0).sum(axis=1)).flatten()
        total_counts = np.array(X.sum(axis=1)).flatten()
    else:
        # Dense calculations with broadcasting
        n_genes = (X > 0).sum(axis=1)
        total_counts = X.sum(axis=1)

    # Vectorized mitochondrial percentage
    mito_genes = adata.var_names.str.startswith('MT-')
    if mito_genes.any():
        if issparse(X):
            mito_counts = np.array(X[:, mito_genes].sum(axis=1)).flatten()
        else:
            mito_counts = X[:, mito_genes].sum(axis=1)
        pct_mito = (mito_counts / total_counts) * 100
    else:
        pct_mito = np.zeros(adata.n_obs)

    return {
        'n_genes': n_genes,
        'total_counts': total_counts,
        'pct_mito': pct_mito
    }

class ParallelProcessingMixin:
    """Mixin for parallel processing capabilities."""

    def __init__(self, max_workers: int = None):
        self.max_workers = max_workers or min(os.cpu_count(), 8)
        self.executor = ThreadPoolExecutor(max_workers=self.max_workers)

    def parallel_apply(self, func, data_chunks: List, **kwargs) -> List:
        """Apply function to data chunks in parallel."""

        futures = []
        for chunk in data_chunks:
            future = self.executor.submit(func, chunk, **kwargs)
            futures.append(future)

        results = []
        for future in as_completed(futures):
            try:
                result = future.result(timeout=300)  # 5 minute timeout
                results.append(result)
            except Exception as e:
                logger.error(f"Parallel processing error: {e}")
                raise

        return results

def detect_gpu_acceleration() -> Dict[str, Any]:
    """Detect available GPU acceleration options."""

    gpu_info = {
        'cuda_available': False,
        'mps_available': False,  # Apple Silicon
        'devices': [],
        'recommended_backend': 'cpu'
    }

    # Check for CUDA
    try:
        import torch
        if torch.cuda.is_available():
            gpu_info['cuda_available'] = True
            gpu_info['devices'].extend([f"cuda:{i}" for i in range(torch.cuda.device_count())])
            gpu_info['recommended_backend'] = 'cuda'
    except ImportError:
        pass

    # Check for Apple Silicon MPS
    try:
        import torch
        if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
            gpu_info['mps_available'] = True
            gpu_info['devices'].append('mps')
            gpu_info['recommended_backend'] = 'mps'
    except (ImportError, AttributeError):
        pass

    return gpu_info

class OptimizedH5ADBackend:
    """H5AD backend with performance optimizations."""

    def load(self, path: Path, backed: bool = None, **kwargs) -> ad.AnnData:
        """Load H5AD with intelligent backing strategy."""

        # Automatic backing decision based on file size
        if backed is None:
            file_size_mb = path.stat().st_size / (1024**2)
            backed = file_size_mb > 500  # Auto-back files > 500MB

        if backed:
            # Memory-efficient backed loading
            adata = sc.read_h5ad(path, backed='r')
            logger.info(f"Loaded {path.name} in backed mode (size: {file_size_mb:.1f}MB)")
        else:
            # Full memory loading with progress tracking
            adata = sc.read_h5ad(path)
            logger.info(f"Loaded {path.name} into memory (size: {file_size_mb:.1f}MB)")

        return adata

    def save(self, adata: ad.AnnData, path: Path, compression: str = 'gzip', **kwargs):
        """Save with optimal compression settings."""

        # Adaptive compression based on data characteristics
        if issparse(adata.X):
            # Sparse data compresses well
            compression_opts = 9
        else:
            # Dense data - balance speed vs size
            compression_opts = 6

        adata.write_h5ad(
            path,
            compression=compression,
            compression_opts=compression_opts,
            **kwargs
        )

        saved_size_mb = path.stat().st_size / (1024**2)
        logger.info(f"Saved {path.name} (size: {saved_size_mb:.1f}MB, compression: {compression})")

def optimized_workspace_scan(workspace_path: Path) -> Dict[str, Any]:
    """Optimized scanning of workspace directories."""

    datasets = {}

    # Parallel directory scanning
    def scan_h5ad_file(file_path: Path) -> Optional[Dict[str, Any]]:
        """Scan individual H5AD file for metadata."""
        try:
            # Quick metadata extraction without full loading
            with h5py.File(file_path, 'r') as f:
                # Extract basic info from HDF5 structure
                if 'X' in f:
                    shape = f['X'].shape
                    file_info = {
                        'path': str(file_path),
                        'size_mb': file_path.stat().st_size / 1e6,
                        'shape': shape,
                        'modified': datetime.fromtimestamp(file_path.stat().st_mtime).isoformat(),
                        'type': 'h5ad'
                    }
                    return file_path.stem, file_info
        except Exception as e:
            logger.warning(f"Could not scan {file_path}: {e}")

        return None

    # Find all H5AD files
    h5ad_files = list(workspace_path.glob("**/*.h5ad"))

    # Parallel processing for large numbers of files
    if len(h5ad_files) > 10:
        with ThreadPoolExecutor(max_workers=4) as executor:
            futures = [executor.submit(scan_h5ad_file, f) for f in h5ad_files]

            for future in as_completed(futures):
                result = future.result()
                if result:
                    name, info = result
                    datasets[name] = info
    else:
        # Sequential processing for small numbers
        for file_path in h5ad_files:
            result = scan_h5ad_file(file_path)
            if result:
                name, info = result
                datasets[name] = info

    return datasets

@contextmanager
def performance_monitor(operation_name: str):
    """Context manager for performance monitoring."""

    import psutil
    import time

    # Initial measurements
    process = psutil.Process()
    start_time = time.time()
    start_memory = process.memory_info().rss / (1024**2)  # MB
    start_cpu_percent = process.cpu_percent()

    try:
        yield
    finally:
        # Final measurements
        end_time = time.time()
        end_memory = process.memory_info().rss / (1024**2)  # MB
        duration = end_time - start_time

        # Log performance metrics
        logger.info(f"Performance - {operation_name}:")
        logger.info(f"  Duration: {duration:.2f}s")
        logger.info(f"  Memory change: {end_memory - start_memory:+.1f}MB")
        logger.info(f"  Peak memory: {end_memory:.1f}MB")

# Usage in services
def optimized_clustering(adata: ad.AnnData, **params) -> Tuple[ad.AnnData, Dict]:
    """Clustering with performance monitoring."""

    with performance_monitor("leiden_clustering"):
        # Perform clustering operations
        sc.tl.leiden(adata, resolution=params.get('resolution', 0.5))

    return adata, {'clusters_found': len(adata.obs['leiden'].unique())}

HTTPS (99%) → FTP fallback (1%) → Pipeline fallback → Error