Upload my_modules.py

my_modules.py

@@ -0,0 +1,468 @@
+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
+
+
+