import os import numpy as np import pandas as pd import subprocess import os import random import re import pandas as pd import numpy as np import seaborn as sb import matplotlib.pyplot as plt import matplotlib.colors as mplc import subprocess from scipy import signal import plotly.figure_factory as ff import plotly import plotly.graph_objs as go from plotly.offline import download_plotlyjs, init_notebook_mode, plot, iplot # This function takes in a dataframe, changes the names # of the column in various ways, and returns the dataframe. # For best accuracy and generalizability, the code uses # regular expressions (regex) to find strings for replacement. def apply_header_changes(df): # remove lowercase x at beginning of name df.columns = df.columns.str.replace("^x","") # remove space at beginning of name df.columns = df.columns.str.replace("^ ","") # replace space with underscore df.columns = df.columns.str.replace(" ","_") # fix typos df.columns = df.columns.str.replace("AF_AF","AF") # change "Cell Id" into "ID" df.columns = df.columns.str.replace("Cell Id","ID") # if the ID is the index, change "Cell Id" into "ID" df.index.name = "ID" # df.columns = df.columns.str.replace("","") return df def apply_df_changes(df): # Remove "@1" after the ID in the index df.index = df.index.str.replace(r'@1$', '') return df def compare_headers(expected, actual, name): missing_actual = np.setdiff1d(expected, actual) extra_actual = np.setdiff1d(actual, expected) if len(missing_actual) > 0: #print("WARNING: File '" + name + "' lacks the following expected header(s) after import header reformatting: \n" # + str(missing_actual)) print("WARNING: File '" + name + "' lacks the following expected item(s): \n" + str(missing_actual)) if len(extra_actual) > 0: #print("WARNING: '" + name + "' has the following unexpected header(s) after import header reformatting: \n" # + str(extra_actual)) print("WARNING: '" + name + "' has the following unexpected item(s): \n" + str(extra_actual)) return None def add_metadata_location(row): fc = row['full_column'].lower() if 'cytoplasm' in fc and 'cell' not in fc and 'nucleus' not in fc: return 'cytoplasm' elif 'cell' in fc and 'cytoplasm' not in fc and 'nucleus' not in fc: return 'cell' elif 'nucleus' in fc and 'cell' not in fc and 'cytoplasm' not in fc: return 'nucleus' else: return 'unknown' def get_perc(row, cell_type): total = row['stroma'] + row['immune'] + row['cancer']+row['endothelial'] return round(row[cell_type]/total *100,1) # Divide each marker (and its localisation) by the right exposure setting for each group of samples def divide_exp_time(col, exp_col, metadata): exp_time = metadata.loc[metadata['full_column'] == col.name, exp_col].values[0] return col/exp_time def do_background_sub(col, df, metadata): #print(col.name) location = metadata.loc[metadata['full_column'] == col.name, 'localisation'].values[0] #print('location = ' + location) channel = metadata.loc[metadata['full_column'] == col.name, 'Channel'].values[0] #print('channel = ' + channel) af_target = metadata.loc[ (metadata['Channel']==channel) \ & (metadata['localisation']==location) \ & (metadata['target_lower'].str.contains(r'^af\d{3}$')),\ 'full_column'].values[0] return col - df.loc[:,af_target] """ This function plots distributions. It takes in a string title (title), a list of dataframes from which to plot (dfs), a list of dataframe names for the legend (names), a list of the desired colors for the plotted samples (colors), a string for the x-axis label (x_label), ```a float binwidth for histrogram (bin_size)```, a boolean to show the legend or not (legend), and the names of the marker(s) to plot (input_labels). If not specified, the function will plot all markers in one plot. input_labels can either be a single string, e.g., 'my_marker', or a list, e.g., ['my_marker1','my_marker2']. The function will create a distribution plot and save it to png. It requires a list of items not to be considered as markers when evaluating column names (not_markers) to be in memory. It also requires a desired output location of the files (output_dir) to already be in memory. """ def make_distr_plot_per_sample(title, location, dfs, df_names, colors, x_label, legend, xlims = None, markers = ['all'],not_intensities = None): ### GET LIST OF MARKERS TO PLOT ### # Get list of markers to plot if not specified by user, using columns in first df # Writing function(parameter = FILLER) makes that parameter optional when user calls function, # since it is given a default value! if markers == ["all"]: markers = [c for c in dfs[0].columns.values if c not in not_intensities] elif not isinstance(markers, list): markers = [markers] # Make input labels a set to get only unique values, then put back into list markers = list(set(markers)) ### GET XLIMS ### if xlims == None: mins = [df.loc[:,markers].min().min() for df in dfs] maxes = [df.loc[:,markers].max().max() for df in dfs] xlims = [min(mins), max(maxes)] if not isinstance(xlims, list): print("Problem - xlmis not list. Exiting method...") return None ### CHECK DATA CAN BE PLOTTED ### # Check for data with only 1 unique value - this will cause error if plotted group_labels = [] hist_data = [] # Iterate through all dataframes (dfs) for i in range(len(dfs)): # Iterate through all marker labels for f in markers: # If there is only one unique value in the marker data for this dataframe, # you cannot plot a distribution plot. It gives you a linear algebra # singular value matrix error if dfs[i][f].nunique() != 1: # Add df name and marker name to labels list # If we have >1 df, we want to make clear # which legend label is associated with which df if len(df_names) > 1: group_labels.append(df_names[i]+"_"+f) else: group_labels.append(f) # add the data to the data list hist_data.append(dfs[i][f]) # if no data had >1 unique values, there is nothing to plot if len(group_labels) < 1: print("No markers plotted - all were singular value. Names and markers were " + str(df_names) + ", " + str(markers)) return None ### TRANSFORM COLOR ITEMS TO CORRECT TYPE ### if isinstance(colors[0], tuple): colors = ['rgb' + str(color) for color in colors] ### PLOT DATA ### # Create plot fig = ff.create_distplot(hist_data, group_labels, bin_size=0.1, #colors=colors, bin_size=bin_size, show_rug=False)#show_hist=False, colors=colors, show_rug=False) # Adjust title, font, background color, legend... fig.update_layout(title_text=title, font=dict(size=18), plot_bgcolor = 'white', showlegend = legend)#, legend_x = 3) # Adjust opacity fig.update_traces(opacity=0.6) # Adjust x-axis parameters fig.update_xaxes(title_text = x_label, showline=True, linewidth=2, linecolor='black', tickfont=dict(size=18), range = xlims) # x lims was here # Adjust y-axis parameters fig.update_yaxes(title_text = "Kernel density estimate",showline=True, linewidth=1, linecolor='black', tickfont=dict(size=18)) ### SAVE/DISPLAY PLOT ### # Save plot to HTML # plotly.io.write_html(fig, file = output_dir + "/" + title + ".html") # Plot in new tab #plot(fig) # Save to png filename = os.path.join(location, title.replace(" ","_") + ".png") fig.write_image(filename) return None # this could be changed to use recursion and make it 'smarter' def shorten_feature_names(long_names): name_dict = dict(zip(long_names,[n.split('_')[0] for n in long_names])) names_lts, long_names, iteration = shorten_feature_names_helper(name_dict, long_names, 1) # names_lts = names long-to-short # names_stl = names stl names_stl = {} for n in names_lts.items(): names_stl[n[1]] = n[0] return names_lts, names_stl def shorten_feature_names_helper(name_dict, long_names, iteration): #print("\nThis is iteration #"+str(iteration)) #print("name_dict is: " + str(name_dict)) #print("long_names is: " + str(long_names)) ## If the number of unique nicknames == number of long names ## then the work here is done #print('\nCompare lengths: ' + str(len(set(name_dict.values()))) + ", " + str(len(long_names))) #print('set(name_dict.values()): ' + str(set(name_dict.values()))) #print('long_names: ' + str(long_names)) if len(set(name_dict.values())) == len(long_names): #print('All done!') return name_dict, long_names, iteration ## otherwise, if the number of unique nicknames is not ## equal to the number of long names (must be shorter than), ## then we need to find more unique names iteration += 1 nicknames_set = set() non_unique_nicknames = set() # construct set of current nicknames for long_name in long_names: #print('long_name is ' + long_name + ' and non_unique_nicknames set is ' + str(non_unique_nicknames)) short_name = name_dict[long_name] if short_name in nicknames_set: non_unique_nicknames.add(short_name) else: nicknames_set.add(short_name) #print('non_unique_nicknames are: ' + str(non_unique_nicknames)) # figure out all long names associated # with the non-unique short names trouble_long_names = set() for long_name in long_names: short_name = name_dict[long_name] if short_name in non_unique_nicknames: trouble_long_names.add(long_name) #print('troublesome long names are: ' + str(trouble_long_names)) #print('name_dict: ' + str(name_dict)) # operate on all names that are associated with # the non-unique short nicknames for long_name in trouble_long_names: #print('trouble long name is: ' + long_name) #print('old nickname is: ' + name_dict[long_name]) name_dict[long_name] = '_'.join(long_name.split('_')[0:iteration]) #print('new nickname is: ' + name_dict[long_name]) shorten_feature_names_helper(name_dict, long_names, iteration) return name_dict, long_names, iteration def heatmap_function2(title, data, method, metric, cmap, cbar_kws, xticklabels, save_loc, row_cluster, col_cluster, annotations = {'rows':[],'cols':[]}): sb.set(font_scale= 6.0) # Extract row and column mappings row_mappings = [] col_mappings = [] for ann in annotations['rows']: row_mappings.append(ann['mapping']) for ann in annotations['cols']: col_mappings.append(ann['mapping']) # If empty lists, convert to None so seaborn accepts # as the row_colors or col_colors objects if len(row_mappings) == 0: row_mappings = None if len(col_mappings) == 0: col_mappings = None def heatmap_function(title, data, method, metric, cmap, cbar_kws, xticklabels, save_loc, row_cluster, col_cluster, annotations = {'rows':[],'cols':[]}): sb.set(font_scale= 2.0) # Extract row and column mappings row_mappings = [] col_mappings = [] for ann in annotations['rows']: row_mappings.append(ann['mapping']) for ann in annotations['cols']: col_mappings.append(ann['mapping']) # If empty lists, convert to None so seaborn accepts # as the row_colors or col_colors objects if len(row_mappings) == 0: row_mappings = None if len(col_mappings) == 0: col_mappings = None # Create clustermap g = sb.clustermap(data = data, robust = True, method = method, metric = metric, cmap = cmap, row_cluster = row_cluster, col_cluster = col_cluster, figsize = (40,30), row_colors=row_mappings, col_colors=col_mappings, yticklabels = False, cbar_kws = cbar_kws, xticklabels = xticklabels) # To rotate slightly the x labels plt.setp(g.ax_heatmap.xaxis.get_majorticklabels(), rotation=45) # Add title g.fig.suptitle(title, fontsize = 60.0) #And now for the legends: # iterate through 'rows', 'cols' for ann_type in annotations.keys(): # iterate through each individual annotation feature for ann in annotations[ann_type]: color_dict = ann['dict'] handles = [] for item in color_dict.keys(): h = g.ax_col_dendrogram.bar(0,0, color = color_dict[item], label = item, linewidth = 0) handles.append(h) legend = plt.legend(handles = handles, loc = ann['location'], title = ann['label'], bbox_to_anchor=ann['bbox_to_anchor'], bbox_transform=plt.gcf().transFigure) ax = plt.gca().add_artist(legend) # Save image filename = os.path.join(save_loc, title.lower().replace(" ","_") + ".png") g.savefig(filename) return None # sources - #https://stackoverflow.com/questions/27988846/how-to-express-classes-on-the-axis-of-a-heatmap-in-seaborn # https://matplotlib.org/3.1.1/tutorials/intermediate/legend_guide.html def verify_line_no(filename, lines_read): # Use Linux "wc -l" command to get the number of lines in the unopened file wc = subprocess.check_output(['wc', '-l', filename]).decode("utf-8") # Take that string, turn it into a list, extract the first item, # and make that an int - this is the number of lines in the file wc = int(wc.split()[0]) if lines_read != wc: print("WARNING: '" + filename + "' has " + str(wc) + " lines, but imported dataframe has " + str(lines_read) + " (including header).") return None def rgb_tuple_from_str(rgb_str): rgb_str = rgb_str.replace("(","").replace(")","").replace(" ","") rgb = list(map(float,rgb_str.split(","))) return tuple(rgb) def color_dict_to_df(cd, column_name): df = pd.DataFrame.from_dict(cd, orient = 'index') df['rgb'] = df.apply(lambda row: (np.float64(row[0]), np.float64(row[1]), np.float64(row[2])), axis = 1) df = df.drop(columns = [0,1,2]) df['hex'] = df.apply(lambda row: mplc.to_hex(row['rgb']), axis = 1) df[column_name] = df.index return df # p-values that are less than or equal to 0.05 def p_add_star(row): m = [str('{:0.3e}'.format(m)) + "*" if m <= 0.05 \ else str('{:0.3e}'.format(m)) for m in row ] return pd.Series(m) # assigns a specific number of asterisks based on the thresholds def p_to_star(row): output = [] for item in row: if item <= 0.001: stars = 3 elif item <= 0.01: stars = 2 elif item <= 0.05: stars = 1 else: stars = 0 value = '' for i in range(stars): value += '*' output.append(value) return pd.Series(output) def plot_gaussian_distributions(df): # Initialize thresholds list to store all calculated thresholds all_thresholds = [] # Iterate over all columns except the first one (assuming the first one is non-numeric or an index) for column in df.columns: # Extract the marker data marker_data = df[column] # Calculating mean and standard deviation for each marker m_mean, m_std = np.mean(marker_data), np.std(marker_data) # Generating x values for the Gaussian curve x_vals = np.linspace(marker_data.min(), marker_data.max(), 100) # Calculating Gaussian distribution curve gaussian_curve = (1 / (m_std * np.sqrt(2 * np.pi))) * np.exp(-(x_vals - m_mean) ** 2 / (2 * m_std ** 2)) # Creating figure for Gaussian distribution for each marker fig = go.Figure() fig.add_trace(go.Scatter(x=x_vals, y=gaussian_curve, mode='lines', name=f'{column} Gaussian Distribution')) fig.update_layout(title=f'Gaussian Distribution for {column} Marker') # Calculating thresholds based on each marker's distribution seuil_1sigma = m_mean + m_std seuil_2sigma = m_mean + 2 * m_std seuil_3sigma = m_mean + 3 * m_std # Display the figures with thresholds fig.add_shape(type='line', x0=seuil_1sigma, y0=0, x1=seuil_1sigma, y1=np.max(gaussian_curve), line=dict(color='red', dash='dash'), name=f'Seuil 1σ: {seuil_1sigma:.2f}') fig.add_shape(type='line', x0=seuil_2sigma, y0=0, x1=seuil_2sigma, y1=np.max(gaussian_curve), line=dict(color='green', dash='dash'), name=f'Seuil 2σ: {seuil_2sigma:.2f}') fig.add_shape(type='line', x0=seuil_3sigma, y0=0, x1=seuil_3sigma, y1=np.max(gaussian_curve), line=dict(color='blue', dash='dash'), name=f'Seuil 3σ: {seuil_3sigma:.2f}') # Add markers and values to the plot fig.add_trace(go.Scatter(x=[seuil_1sigma, seuil_2sigma, seuil_3sigma], y=[0, 0, 0], mode='markers+text', text=[f'{seuil_1sigma:.2f}', f'{seuil_2sigma:.2f}', f'{seuil_3sigma:.2f}'], textposition="top center", marker=dict(size=10, color=['red', 'green', 'blue']), name='Threshold Values')) fig.show() # Append thresholds for each marker to the list all_thresholds.append((column, seuil_1sigma, seuil_2sigma, seuil_3sigma)) # Include the column name # Return thresholds for all markers return all_thresholds