import sklearn.model_selection
import itertools
import pandas as pd
import numpy as np
import requests
import io
[docs]def group_organelles(df_, mapping):
'''Group together organelles based on a defined mapping pattern.
Args:
df_ (pd.DataFrame): Pandas dataframe properly formatted for TRANSPIRE analysis. Must contain a "localization" index level.
mapping (dict): dict of values defining how to combine organelles.
Returns:
df: Copy of the inuput dataframe with the "localization" index level mapped according to mapping
Note that organelles not found in mapping, but defined in the "localization" index level will retain their original value.
'''
df = df_.copy().reset_index()
df['localization'] = df['localization'].where(~df['localization'].isin(list(mapping.keys())), df['localization'].map(mapping))
return df.set_index(df_.index.names)
[docs]def sample_balanced(X, n_limit, n_folds, random_state = 17):
'''Take class-balanced subsets of data from an input dataframe (based on the "label" index level)
Args:
X (pd.DataFrame): Pandas dataframe properly formatted for TRANSPIRE analysis. Must contain a "label" index level. Most commonly, this would be a dataframe containing synthetic translocations generated by TRANSPIRE
n_limit (int): The maximum number of samples to take from any one class
n_folds (int): The number of different balanced folds to generate
random_state (int, optional): random state to be passed to sklearn.model_selection.StratifiedKFold and pd.DataFrame.sample
Returns:
X_train_dfs (pd.DataFrame): Dataframe with an additional "fold" level appended to the index. This dataframe consists of n_folds of the input data, stratified such that any one fold does not have more than n_limit members of a given "label".
Note that the returned folds of data will overlap for any classes (ie. labels) where the number of members is < n_folds*n_limit.
'''
def parse_data(x):
if x.shape[0]>= n_limit:
return x.sample(n_limit, random_state=17).loc[x.name[0], :].reset_index('label', drop=True)
else:
to_samp = n_limit-x.shape[0]
poss = X[(X.index.get_level_values('label')==x.name[1])&(~X.index.isin(x.loc[x.name[0], :].index))]
poss_sample = poss.sample(min(poss.shape[0], to_samp), random_state=x.name[0])
return pd.concat([x.loc[x.name[0]], poss_sample]).reset_index('label', drop=True)
X_train_dfs = []
skf = sklearn.model_selection.StratifiedKFold(n_splits=n_folds, random_state=17)
print('Creating balanced training partitions (this may take a while)....', end=' ')
for _, test_idx in skf.split(X, X.index.get_level_values('label')):
X_train_dfs.append(X.iloc[test_idx, :])
temp = pd.concat(X_train_dfs, keys=range(1, n_folds+1), names=['fold'])
X_train_dfs = temp.groupby(['fold', 'label']).apply(parse_data)
print('done')
return X_train_dfs
[docs]def train_test_validate_split(X, groupby_levels, f_train = 0.5, f_validate = 0.25, f_test = 0.25, random_state=17):
'''Split input dataframe into training, validation, and testing partitions, stratified by a set of index levels.
Args:
X (pd.DataFrame): data to be split into partitions as defined by f_train, f_validate, and f_test
groupby_levels (list): which levels to groupby when taking stratified samples
f_train (float, optional): fraction of the data to be used for training
f_validate (float, optional): fraction of the data to be used for validation
f_test (float, optional): fraction of the data to be used for testing (i.e. to evaluate predictive performance)
random_state (int, optional): random state to be passed pd.DataFrame.sample
Returns:
X_train_df (pd.DataFrame): stratified data to be used for model training
X_validate_df (pd.DataFrame): stratified data to be used for validation
X_test_df (pd.DataFrame): stratified data for testing
Raises:
ValueError: If the sum of f_train, f_validate, and f_test does not equal 1
'''
if not abs(sum([f_train, f_validate, f_test])-1) < 0.05:
raise ValueError ('Sum of f_train, f_validate, and f_test must equal 1.')
print('Splitting data into training, validation, and testing folds (this may take a while) . . . ', end='')
X_train_validate = X.groupby(groupby_levels, group_keys=False).apply(lambda x: x.sample(frac=f_train, random_state=random_state))
X_train_df = X_train_validate.groupby(groupby_levels, group_keys=False).apply(lambda x: x.sample(frac=(2*f_validate)/(f_train+f_validate), random_state=random_state))
X_validate_df = X_train_validate[~X_train_validate.index.isin(X_train_df.index)].copy()
X_test_df = X[~X.index.isin(X_train_validate.index)].copy()
print('done')
return X_train_df, X_validate_df, X_test_df
[docs]def map_binary(x, mapping):
'''Map scores to their binary translocation score representation
Args:
x (pd.DataFrame): scores to be mapped via mapping
mapping (pd.Series): mapping that defines how to map encoded columns in x back to their translocation labels as strings
Returns:
df (pd.DataFrame): Dataframe with the same index as x, with the mapped translocation scores. If the input dataframe also has a "label" index level, the returned dataframe will also have a "true label" column corresponding to the label mapped to its binary representation.
'''
cols = np.array([np.array(arr) for arr in x[x.columns[~x.isnull().all()]].columns.astype(int).map(mapping).str.split(' to ', expand=True).values])
one = x[x[x.columns[~x.isnull().all()]].columns[(cols[:, 0]!=cols[:, 1])&(cols[:, 0]!='No translocation')]]
if 'label' in x.index.names:
df = pd.concat([one.sum(axis=1), pd.Series(x.index.get_level_values('localization_A')!=x.index.get_level_values('localization_B'), index=x.index)*1], keys = ['translocation score', 'true label'], axis=1)
else:
df = pd.concat([one.sum(axis=1)], keys = ['translocation score'], axis=1)
return df
[docs]def lookup(string, df, level):
'''Find rows in dataframe that correspond to a search term.
Args:
string (str): String to search for
df (pd.DataFrame): Dataframe to search within
level (str): level of index to search for string (e.g. 'accession', 'gene name', etc)
Returns:
df (pd.DataFrame): rows of the dataframe containing the given search term in their index
'''
return df[~df.index.get_level_values(level).isnull()][df.index.get_level_values(level).dropna().str.lower().str.contains(string.lower())]
[docs]def get_mapping(df):
'''Get label encodings for translocation classes based on the "localization" level of the dataframe
Args:
df (pd.DataFrame): dataframe with a "localization" index level
Returns:
mapping (pd.Series): Translocation labels mapped to integers
mapping_r(pd.Series): Integers mapped back to translocation labels
'''
labels = np.unique([' to '.join(tup) for tup in list(itertools.product(np.unique(df.index.get_level_values('localization').dropna()), np.unique(df.index.get_level_values('localization').dropna())))])
mapping = pd.Series(range(0, len(labels)), index = labels)
mapping_r = pd.Series(mapping.index, index=mapping)
return mapping, mapping_r
[docs]def uniprot_mapping_service(proteins, to):
'''Use the Uniprot REST ID mapping service.
Args:
proteins (Union[list, np.array]): values to be mapped (e.g. Uniprot accession numbers)
to (str): type of mapping to be accomplished as defined by Uniprot standards
Returns:
df (pd.DataFrame): Dataframe with accession numbers (as index) mapped to their corresponding values
'''
url = 'https://www.uniprot.org/uploadlists/'
_from = 'ACC+ID'
# check that there are no NaNs in proteins
if pd.isnull(proteins).any():
raise ValueError('remove NaN values before mapping')
p = ' '.join([protein for protein in proteins])
if 'string_db' in to.lower():
to = 'STRING_ID'
name = 'StringID'
elif 'gene' in to.lower() and 'id' in to.lower():
to = 'P_ENTREZGENEID'
name ='GeneID'
else:
raise ValueError('{} is not a valid mapping type'.format(to))
params = {'from': _from,
'to':to,
'format':'tab',
'query': p}
r = requests.post(url, data = params)
df = pd.read_csv(io.StringIO(r.text), sep = '\t')
df.columns = ['accession number', name]
df = df.set_index(['accession number'])
df = df[~df.index.duplicated()]
return df