scmkl.data_processing
1import numpy as np 2import scipy 3from sklearn.decomposition import TruncatedSVD, PCA 4import anndata as ad 5 6 7def sparse_var(X: scipy.sparse._csc.csc_matrix | np.ndarray, axis: int | None=None): 8 """ 9 Function to calculate variance on a scipy sparse matrix. 10 11 Parameters 12 ---------- 13 X : scipy.sparse._csc.csc_matrix | np.ndarray 14 A scipy sparse or numpy array 15 16 axis : int | None 17 Determines which axis variance is calculated on. Same usage 18 as Numpy. 19 20 Returns 21 ------- 22 var : np.ndarray | float 23 Variance values calculated over the given axis. 24 """ 25 # E[X^2] - E[X]^2 26 if scipy.sparse.issparse(X): 27 exp_mean = np.asarray(X.power(2).mean(axis = axis)).flatten() 28 sq_mean = np.asarray(np.square(X.mean(axis = axis))).flatten() 29 var = np.array(exp_mean - sq_mean) 30 else: 31 var = np.asarray(np.var(X, axis = axis)).flatten() 32 33 return var.ravel() 34 35 36def add_dummy_features(X_train: np.ndarray | scipy.sparse._csc.csc_matrix, 37 X_test: np.ndarray | scipy.sparse._csc.csc_matrix 38 | None=None, orig_test: str | None=None): 39 """ 40 Will add a column of ones to the matrices. If X_test not provided, 41 col will only be added to X_train. 42 """ 43 X_train = np.hstack((X_train, np.ones((X_train.shape[0], 1), 44 dtype=np.float16))) 45 46 if not isinstance(orig_test, type(None)): 47 X_test = np.hstack((X_test, np.ones((X_test.shape[0], 1), 48 dtype=np.float16))) 49 50 return X_train, X_test 51 52 53def process_data(X_train: np.ndarray | scipy.sparse._csc.csc_matrix, 54 X_test: np.ndarray | scipy.sparse._csc.csc_matrix | None=None, 55 scale_data: bool=True, transform_data: bool=False, 56 return_dense: bool=True, add_ones: bool=False): 57 """ 58 Function to preprocess data matrix according to type of data 59 (e.g. counts/rna, or binary/atac). Will process test data 60 according to parameters calculated from test data. 61 62 Parameters 63 ---------- 64 X_train : np.ndarray | scipy.sparse._csc.csc_matrix 65 A scipy sparse or numpy array of cells x features in the 66 training data. 67 68 X_test : np.ndarray | scipy.sparse._csc.csc_matrix 69 A scipy sparse or numpy array of cells x features in the 70 testing data. 71 72 scale_data : bool 73 If `True`, data will be logarithmized then z-score 74 transformed. 75 76 transform_data : bool 77 If `True`, data will be log1p transformed (recommended for 78 counts data). Default is `False`. 79 80 return_dense: bool 81 If `True`, a np.ndarray will be returned as opposed to a 82 scipy.sparse object. 83 84 Returns 85 ------- 86 X_train, X_test : np.ndarray, np.ndarray 87 Numpy arrays with the process train/test data 88 respectively. If X_test is `None`, only X_train is returned. 89 """ 90 if X_test is None: 91 # Creates dummy matrix to for the sake of calculation without 92 # increasing computational time 93 X_test = X_train[:1,:] 94 orig_test = None 95 else: 96 orig_test = 'given' 97 98 # Remove features that have low variance in the training data 99 var = sparse_var(X_train, axis = 0) 100 variable_features = np.where(var > 1e-5)[0] 101 102 X_train = X_train[:,variable_features] 103 X_test = X_test[:, variable_features] 104 105 # Data processing according to data type 106 if transform_data: 107 108 if scipy.sparse.issparse(X_train): 109 X_train = X_train.log1p() 110 X_test = X_test.log1p() 111 else: 112 X_train = np.log1p(X_train) 113 X_test = np.log1p(X_test) 114 115 if scale_data: 116 #Center and scale count data 117 train_means = np.mean(X_train, 0) 118 train_sds = np.sqrt(var[variable_features]) 119 120 # Perform transformation on test data according to parameters 121 # of the training data 122 X_train = (X_train - train_means) / train_sds 123 X_test = (X_test - train_means) / train_sds 124 125 126 if return_dense and scipy.sparse.issparse(X_train): 127 X_train = X_train.toarray() 128 X_test = X_test.toarray() 129 130 if add_ones: 131 X_train, X_test = add_dummy_features(X_train, X_test, orig_test) 132 133 if orig_test is None: 134 return X_train 135 else: 136 return X_train, X_test 137 138 139def svd_transformation(X_train: scipy.sparse._csc.csc_matrix | np.ndarray, 140 X_test: scipy.sparse._csc.csc_matrix | 141 np.ndarray | None=None): 142 """ 143 Returns matrices with SVD reduction. If `X_test is None`, only 144 X_train is returned. 145 146 Parameters 147 ---------- 148 X_train : np.ndarray 149 A 2D array of cells x features filtered to desired features 150 for training data. 151 152 X_test : np.ndarray | None 153 A 2D array of cells x features filtered to desired features 154 for testing data. 155 156 Returns 157 ------- 158 X_train, X_test : np.ndarray, np.ndarray 159 Transformed matrices. Only X_train is returned if 160 `X_test is None`. 161 """ 162 n_components = np.min([50, X_train.shape[1]]) 163 SVD_func = TruncatedSVD(n_components = n_components, random_state = 1) 164 165 # Remove first component as it corresponds with sequencing depth 166 # We convert to a csr_array because the SVD function is faster on this 167 # matrix type 168 X_train = SVD_func.fit_transform(scipy.sparse.csr_array(X_train))[:, 1:] 169 170 if X_test is not None: 171 X_test = SVD_func.transform(scipy.sparse.csr_array(X_test))[:, 1:] 172 173 return X_train, X_test 174 175 176def sample_cells(train_indices: np.ndarray, 177 sample_size: int, 178 seed_obj: np.random._generator.Generator): 179 """ 180 Samples cells indices from training indices for calculations. 181 182 Parameters 183 ---------- 184 train_indices : np.ndarray 185 An array of indices to sample from. 186 187 sample_size : int 188 Number of samples to take from `train_indices`. Must be 189 smaller than length of `train_indices`. 190 191 Returns 192 ------- 193 indices : np.ndarray 194 The sampled indices from `train_indices`. 195 """ 196 n_samples = np.min((train_indices.shape[0], sample_size)) 197 indices = seed_obj.choice(train_indices, n_samples, replace = False) 198 199 return indices 200 201 202def pca_transformation(X_train: scipy.sparse._csc.csc_matrix | np.ndarray, 203 X_test: scipy.sparse._csc.csc_matrix | np.ndarray | None=None): 204 """ 205 Returns matrices with PCA reduction. If `X_test is None`, only 206 X_train is returned. 207 208 Parameters 209 ---------- 210 X_train : scipy.sparse._csc.csc_matrix | np.ndarray 211 A 2D array of cells x features filtered to desired features 212 for training data. 213 214 X_test : scipy.sparse._csc.csc_matrix | np.ndarray | None 215 A 2D array of cells x features filtered to desired features 216 for testing data. 217 218 Returns 219 ------- 220 X_train, X_test : np.ndarray, np.ndarray 221 Transformed matrices. Only X_train is returned if 222 `X_test is None`. 223 """ 224 n_components = np.min([50, X_train.shape[1]]) 225 PCA_func = PCA(n_components = n_components, random_state = 1) 226 227 X_train = PCA_func.fit_transform(np.asarray(X_train)) 228 229 if X_test is not None: 230 X_test = PCA_func.transform(np.asarray(X_test)) 231 232 return X_train, X_test 233 234 235def _no_transformation(X_train: scipy.sparse._csc.csc_matrix | np.ndarray, 236 X_test: scipy.sparse._csc.csc_matrix | np.ndarray | None=None): 237 """ 238 Dummy function used to return mat inputs. 239 """ 240 return X_train, X_test 241 242 243def get_reduction(reduction: str): 244 """ 245 Function used to identify reduction type and return function to 246 apply to data matrices. 247 248 Parameters 249 ---------- 250 reduction : str 251 The reduction for data transformation. Options are `['pca', 252 'svd', 'None']`. 253 254 Returns 255 ------- 256 red_func : function 257 The function to reduce the data. 258 """ 259 match reduction: 260 case 'pca': 261 red_func = pca_transformation 262 case 'svd': 263 red_func = svd_transformation 264 case 'None': 265 red_func = _no_transformation 266 267 return red_func 268 269 270def get_group_mat(adata: ad.AnnData, n_features: int, 271 group_features: np.ndarray, 272 n_group_features: int, 273 process_test: bool=False) -> np.ndarray: 274 """ 275 Filters to only features in group. Will sample features if 276 `n_features < n_group_features`. 277 278 Parameters 279 ---------- 280 adata : anndata.AnnData 281 anndata object with `'seed_obj'`, `'train_indices'`, and 282 `'test_indices'` in `.uns`. 283 284 n_features : int 285 Maximum number of features to keep in matrix. Only 286 impacts mat if `n_features < n_group_features`. 287 288 group_features : list | tuple | np.ndarray 289 Feature names in group to filter matrices to. 290 291 n_group_features : int 292 Number of features in group. 293 294 n_samples : int 295 Number of samples to filter X_train to. 296 297 Returns 298 ------- 299 X_train, X_test : np.ndarray, np.ndarray 300 Filtered matrices. If `n_samples` is provided, only `X_train` 301 is returned. If `adata.uns['reduction']` is `'pca'` or 302 `'svd'` the matrices are transformed before being returned. 303 """ 304 # Getting reduction function 305 reduction_func = get_reduction(adata.uns['reduction']) 306 307 # Sample up to n_features features- important for scalability if 308 # using large groupings 309 # Will use all features if the grouping contains fewer than n_features 310 number_features = np.min([n_features, n_group_features]) 311 312 group_array = np.array(list(group_features)) 313 group_features = adata.uns['seed_obj'].choice(group_array, 314 number_features, 315 replace=False) 316 317 # Create data arrays containing only features within this group 318 if process_test: 319 X_train = adata[adata.uns['train_indices'],:][:, group_features].X 320 X_test = adata[adata.uns['test_indices'],:][:, group_features].X 321 X_train, X_test = reduction_func(X_train, X_test) 322 return X_train, X_test 323 324 else: 325 X_train = adata[:, group_features].X 326 return X_train
def
sparse_var( X: scipy.sparse._csc.csc_matrix | numpy.ndarray, axis: int | None = None):
8def sparse_var(X: scipy.sparse._csc.csc_matrix | np.ndarray, axis: int | None=None): 9 """ 10 Function to calculate variance on a scipy sparse matrix. 11 12 Parameters 13 ---------- 14 X : scipy.sparse._csc.csc_matrix | np.ndarray 15 A scipy sparse or numpy array 16 17 axis : int | None 18 Determines which axis variance is calculated on. Same usage 19 as Numpy. 20 21 Returns 22 ------- 23 var : np.ndarray | float 24 Variance values calculated over the given axis. 25 """ 26 # E[X^2] - E[X]^2 27 if scipy.sparse.issparse(X): 28 exp_mean = np.asarray(X.power(2).mean(axis = axis)).flatten() 29 sq_mean = np.asarray(np.square(X.mean(axis = axis))).flatten() 30 var = np.array(exp_mean - sq_mean) 31 else: 32 var = np.asarray(np.var(X, axis = axis)).flatten() 33 34 return var.ravel()
Function to calculate variance on a scipy sparse matrix.
Parameters
- X (scipy.sparse._csc.csc_matrix | np.ndarray): A scipy sparse or numpy array
- axis (int | None): Determines which axis variance is calculated on. Same usage as Numpy.
Returns
- var (np.ndarray | float): Variance values calculated over the given axis.
def
add_dummy_features( X_train: numpy.ndarray | scipy.sparse._csc.csc_matrix, X_test: numpy.ndarray | scipy.sparse._csc.csc_matrix | None = None, orig_test: str | None = None):
37def add_dummy_features(X_train: np.ndarray | scipy.sparse._csc.csc_matrix, 38 X_test: np.ndarray | scipy.sparse._csc.csc_matrix 39 | None=None, orig_test: str | None=None): 40 """ 41 Will add a column of ones to the matrices. If X_test not provided, 42 col will only be added to X_train. 43 """ 44 X_train = np.hstack((X_train, np.ones((X_train.shape[0], 1), 45 dtype=np.float16))) 46 47 if not isinstance(orig_test, type(None)): 48 X_test = np.hstack((X_test, np.ones((X_test.shape[0], 1), 49 dtype=np.float16))) 50 51 return X_train, X_test
Will add a column of ones to the matrices. If X_test not provided, col will only be added to X_train.
def
process_data( X_train: numpy.ndarray | scipy.sparse._csc.csc_matrix, X_test: numpy.ndarray | scipy.sparse._csc.csc_matrix | None = None, scale_data: bool = True, transform_data: bool = False, return_dense: bool = True, add_ones: bool = False):
54def process_data(X_train: np.ndarray | scipy.sparse._csc.csc_matrix, 55 X_test: np.ndarray | scipy.sparse._csc.csc_matrix | None=None, 56 scale_data: bool=True, transform_data: bool=False, 57 return_dense: bool=True, add_ones: bool=False): 58 """ 59 Function to preprocess data matrix according to type of data 60 (e.g. counts/rna, or binary/atac). Will process test data 61 according to parameters calculated from test data. 62 63 Parameters 64 ---------- 65 X_train : np.ndarray | scipy.sparse._csc.csc_matrix 66 A scipy sparse or numpy array of cells x features in the 67 training data. 68 69 X_test : np.ndarray | scipy.sparse._csc.csc_matrix 70 A scipy sparse or numpy array of cells x features in the 71 testing data. 72 73 scale_data : bool 74 If `True`, data will be logarithmized then z-score 75 transformed. 76 77 transform_data : bool 78 If `True`, data will be log1p transformed (recommended for 79 counts data). Default is `False`. 80 81 return_dense: bool 82 If `True`, a np.ndarray will be returned as opposed to a 83 scipy.sparse object. 84 85 Returns 86 ------- 87 X_train, X_test : np.ndarray, np.ndarray 88 Numpy arrays with the process train/test data 89 respectively. If X_test is `None`, only X_train is returned. 90 """ 91 if X_test is None: 92 # Creates dummy matrix to for the sake of calculation without 93 # increasing computational time 94 X_test = X_train[:1,:] 95 orig_test = None 96 else: 97 orig_test = 'given' 98 99 # Remove features that have low variance in the training data 100 var = sparse_var(X_train, axis = 0) 101 variable_features = np.where(var > 1e-5)[0] 102 103 X_train = X_train[:,variable_features] 104 X_test = X_test[:, variable_features] 105 106 # Data processing according to data type 107 if transform_data: 108 109 if scipy.sparse.issparse(X_train): 110 X_train = X_train.log1p() 111 X_test = X_test.log1p() 112 else: 113 X_train = np.log1p(X_train) 114 X_test = np.log1p(X_test) 115 116 if scale_data: 117 #Center and scale count data 118 train_means = np.mean(X_train, 0) 119 train_sds = np.sqrt(var[variable_features]) 120 121 # Perform transformation on test data according to parameters 122 # of the training data 123 X_train = (X_train - train_means) / train_sds 124 X_test = (X_test - train_means) / train_sds 125 126 127 if return_dense and scipy.sparse.issparse(X_train): 128 X_train = X_train.toarray() 129 X_test = X_test.toarray() 130 131 if add_ones: 132 X_train, X_test = add_dummy_features(X_train, X_test, orig_test) 133 134 if orig_test is None: 135 return X_train 136 else: 137 return X_train, X_test
Function to preprocess data matrix according to type of data (e.g. counts/rna, or binary/atac). Will process test data according to parameters calculated from test data.
Parameters
- X_train (np.ndarray | scipy.sparse._csc.csc_matrix): A scipy sparse or numpy array of cells x features in the training data.
- X_test (np.ndarray | scipy.sparse._csc.csc_matrix): A scipy sparse or numpy array of cells x features in the testing data.
- scale_data (bool):
If
True, data will be logarithmized then z-score transformed. - transform_data (bool):
If
True, data will be log1p transformed (recommended for counts data). Default isFalse. - return_dense (bool):
If
True, a np.ndarray will be returned as opposed to a scipy.sparse object.
Returns
- X_train, X_test (np.ndarray, np.ndarray):
Numpy arrays with the process train/test data
respectively. If X_test is
None, only X_train is returned.
def
svd_transformation( X_train: scipy.sparse._csc.csc_matrix | numpy.ndarray, X_test: scipy.sparse._csc.csc_matrix | numpy.ndarray | None = None):
140def svd_transformation(X_train: scipy.sparse._csc.csc_matrix | np.ndarray, 141 X_test: scipy.sparse._csc.csc_matrix | 142 np.ndarray | None=None): 143 """ 144 Returns matrices with SVD reduction. If `X_test is None`, only 145 X_train is returned. 146 147 Parameters 148 ---------- 149 X_train : np.ndarray 150 A 2D array of cells x features filtered to desired features 151 for training data. 152 153 X_test : np.ndarray | None 154 A 2D array of cells x features filtered to desired features 155 for testing data. 156 157 Returns 158 ------- 159 X_train, X_test : np.ndarray, np.ndarray 160 Transformed matrices. Only X_train is returned if 161 `X_test is None`. 162 """ 163 n_components = np.min([50, X_train.shape[1]]) 164 SVD_func = TruncatedSVD(n_components = n_components, random_state = 1) 165 166 # Remove first component as it corresponds with sequencing depth 167 # We convert to a csr_array because the SVD function is faster on this 168 # matrix type 169 X_train = SVD_func.fit_transform(scipy.sparse.csr_array(X_train))[:, 1:] 170 171 if X_test is not None: 172 X_test = SVD_func.transform(scipy.sparse.csr_array(X_test))[:, 1:] 173 174 return X_train, X_test
Returns matrices with SVD reduction. If X_test is None, only
X_train is returned.
Parameters
- X_train (np.ndarray): A 2D array of cells x features filtered to desired features for training data.
- X_test (np.ndarray | None): A 2D array of cells x features filtered to desired features for testing data.
Returns
- X_train, X_test (np.ndarray, np.ndarray):
Transformed matrices. Only X_train is returned if
X_test is None.
def
sample_cells( train_indices: numpy.ndarray, sample_size: int, seed_obj: numpy.random._generator.Generator):
177def sample_cells(train_indices: np.ndarray, 178 sample_size: int, 179 seed_obj: np.random._generator.Generator): 180 """ 181 Samples cells indices from training indices for calculations. 182 183 Parameters 184 ---------- 185 train_indices : np.ndarray 186 An array of indices to sample from. 187 188 sample_size : int 189 Number of samples to take from `train_indices`. Must be 190 smaller than length of `train_indices`. 191 192 Returns 193 ------- 194 indices : np.ndarray 195 The sampled indices from `train_indices`. 196 """ 197 n_samples = np.min((train_indices.shape[0], sample_size)) 198 indices = seed_obj.choice(train_indices, n_samples, replace = False) 199 200 return indices
Samples cells indices from training indices for calculations.
Parameters
- train_indices (np.ndarray): An array of indices to sample from.
- sample_size (int):
Number of samples to take from
train_indices. Must be smaller than length oftrain_indices.
Returns
- indices (np.ndarray):
The sampled indices from
train_indices.
def
pca_transformation( X_train: scipy.sparse._csc.csc_matrix | numpy.ndarray, X_test: scipy.sparse._csc.csc_matrix | numpy.ndarray | None = None):
203def pca_transformation(X_train: scipy.sparse._csc.csc_matrix | np.ndarray, 204 X_test: scipy.sparse._csc.csc_matrix | np.ndarray | None=None): 205 """ 206 Returns matrices with PCA reduction. If `X_test is None`, only 207 X_train is returned. 208 209 Parameters 210 ---------- 211 X_train : scipy.sparse._csc.csc_matrix | np.ndarray 212 A 2D array of cells x features filtered to desired features 213 for training data. 214 215 X_test : scipy.sparse._csc.csc_matrix | np.ndarray | None 216 A 2D array of cells x features filtered to desired features 217 for testing data. 218 219 Returns 220 ------- 221 X_train, X_test : np.ndarray, np.ndarray 222 Transformed matrices. Only X_train is returned if 223 `X_test is None`. 224 """ 225 n_components = np.min([50, X_train.shape[1]]) 226 PCA_func = PCA(n_components = n_components, random_state = 1) 227 228 X_train = PCA_func.fit_transform(np.asarray(X_train)) 229 230 if X_test is not None: 231 X_test = PCA_func.transform(np.asarray(X_test)) 232 233 return X_train, X_test
Returns matrices with PCA reduction. If X_test is None, only
X_train is returned.
Parameters
- X_train (scipy.sparse._csc.csc_matrix | np.ndarray): A 2D array of cells x features filtered to desired features for training data.
- X_test (scipy.sparse._csc.csc_matrix | np.ndarray | None): A 2D array of cells x features filtered to desired features for testing data.
Returns
- X_train, X_test (np.ndarray, np.ndarray):
Transformed matrices. Only X_train is returned if
X_test is None.
def
get_reduction(reduction: str):
244def get_reduction(reduction: str): 245 """ 246 Function used to identify reduction type and return function to 247 apply to data matrices. 248 249 Parameters 250 ---------- 251 reduction : str 252 The reduction for data transformation. Options are `['pca', 253 'svd', 'None']`. 254 255 Returns 256 ------- 257 red_func : function 258 The function to reduce the data. 259 """ 260 match reduction: 261 case 'pca': 262 red_func = pca_transformation 263 case 'svd': 264 red_func = svd_transformation 265 case 'None': 266 red_func = _no_transformation 267 268 return red_func
Function used to identify reduction type and return function to apply to data matrices.
Parameters
- reduction (str):
The reduction for data transformation. Options are
['pca', 'svd', 'None'].
Returns
- red_func (function): The function to reduce the data.
def
get_group_mat( adata: anndata._core.anndata.AnnData, n_features: int, group_features: numpy.ndarray, n_group_features: int, process_test: bool = False) -> numpy.ndarray:
271def get_group_mat(adata: ad.AnnData, n_features: int, 272 group_features: np.ndarray, 273 n_group_features: int, 274 process_test: bool=False) -> np.ndarray: 275 """ 276 Filters to only features in group. Will sample features if 277 `n_features < n_group_features`. 278 279 Parameters 280 ---------- 281 adata : anndata.AnnData 282 anndata object with `'seed_obj'`, `'train_indices'`, and 283 `'test_indices'` in `.uns`. 284 285 n_features : int 286 Maximum number of features to keep in matrix. Only 287 impacts mat if `n_features < n_group_features`. 288 289 group_features : list | tuple | np.ndarray 290 Feature names in group to filter matrices to. 291 292 n_group_features : int 293 Number of features in group. 294 295 n_samples : int 296 Number of samples to filter X_train to. 297 298 Returns 299 ------- 300 X_train, X_test : np.ndarray, np.ndarray 301 Filtered matrices. If `n_samples` is provided, only `X_train` 302 is returned. If `adata.uns['reduction']` is `'pca'` or 303 `'svd'` the matrices are transformed before being returned. 304 """ 305 # Getting reduction function 306 reduction_func = get_reduction(adata.uns['reduction']) 307 308 # Sample up to n_features features- important for scalability if 309 # using large groupings 310 # Will use all features if the grouping contains fewer than n_features 311 number_features = np.min([n_features, n_group_features]) 312 313 group_array = np.array(list(group_features)) 314 group_features = adata.uns['seed_obj'].choice(group_array, 315 number_features, 316 replace=False) 317 318 # Create data arrays containing only features within this group 319 if process_test: 320 X_train = adata[adata.uns['train_indices'],:][:, group_features].X 321 X_test = adata[adata.uns['test_indices'],:][:, group_features].X 322 X_train, X_test = reduction_func(X_train, X_test) 323 return X_train, X_test 324 325 else: 326 X_train = adata[:, group_features].X 327 return X_train
Filters to only features in group. Will sample features if
n_features < n_group_features.
Parameters
- adata (anndata.AnnData):
anndata object with
'seed_obj','train_indices', and'test_indices'in.uns. - n_features (int):
Maximum number of features to keep in matrix. Only
impacts mat if
n_features < n_group_features. - group_features (list | tuple | np.ndarray): Feature names in group to filter matrices to.
- n_group_features (int): Number of features in group.
- n_samples (int): Number of samples to filter X_train to.
Returns
- X_train, X_test (np.ndarray, np.ndarray):
Filtered matrices. If
n_samplesis provided, onlyX_trainis returned. Ifadata.uns['reduction']is'pca'or'svd'the matrices are transformed before being returned.