Merge branch 'main' into Adding-mouse-immune-dictionary-ANOVA

nbedanova · web-flow · commit 57d7f5f263b8 · 2025-06-30T18:01:17.000-07:00
diff --git a/pf2rnaseq/factorization.py b/pf2rnaseq/factorization.py
@@ -1,12 +1,16 @@
 import anndata
 import cupy
 import numpy as np
+import pandas as pd
+import scanpy as sc
 import scipy.sparse as sps
 from pacmap import PaCMAP
 from parafac2.parafac2 import parafac2_nd, store_pf2
 from scipy.stats import gmean
 from sklearn.decomposition import PCA
 from sklearn.linear_model import LinearRegression
+from tensorly.cp_tensor import CPTensor
+from tlviz.factor_tools import factor_match_score as fms
 from tqdm import tqdm
 
 
@@ -41,7 +45,13 @@ def pf2(
 ):
     cupy.cuda.Device(1).use()
     pf_out, R2X = parafac2_nd(
-        X, rank=rank, random_state=random_state, tol=tolerance, n_iter_max=500
+
+        X,
+        rank=rank,
+        random_state=random_state,
+        tol=tolerance,
+        n_iter_max=500,
+
     )
 
     X = store_pf2(X, pf_out)
@@ -70,3 +80,115 @@ def pf2_pca_r2x(X: anndata.AnnData, ranks):
     r2x_pca = np.cumsum(pca.explained_variance_ratio_)
 
     return r2x_pf2, r2x_pca[np.array(ranks) - 1]
+
+
+def calculateFMS(A: anndata.AnnData, B: anndata.AnnData):
+    """Calculates FMS between 2 factors"""
+    factors = [A.uns["Pf2_A"], A.uns["Pf2_B"], A.varm["Pf2_C"]]
+    A_CP = CPTensor(
+        (
+            A.uns["Pf2_weights"],
+            factors,
+        )
+    )
+
+    factors = [B.uns["Pf2_A"], B.uns["Pf2_B"], B.varm["Pf2_C"]]
+    B_CP = CPTensor(
+        (
+            B.uns["Pf2_weights"],
+            factors,
+        )
+    )
+
+    return fms(A_CP, B_CP, consider_weights=False, skip_mode=1)  # type: ignore
+
+
+def fms_percent_drop(
+    X: anndata.AnnData,
+    percentList: np.ndarray,
+    runs: int,
+    rank: int = 30,
+):
+    # Plots FMS score when percentage is removed from data
+    dataX = pf2(X, rank, doEmbedding=False)
+
+    fmsLists = []
+
+    for j in range(0, runs, 1):
+        scores = [1.0]
+
+        for i in percentList[1:]:
+            sampled_data: anndata.AnnData = sc.pp.subsample(
+                X, fraction=1 - (i / 100), random_state=j, copy=True
+            )  # type: ignore
+            sampledX = pf2(sampled_data, rank, random_state=j + 2, doEmbedding=False)
+
+            fmsScore = calculateFMS(dataX, sampledX)
+            scores.append(fmsScore)
+
+        fmsLists.append(scores)
+
+    runsList_df = []
+    for i in range(0, runs):
+        for _j in range(0, len(percentList)):
+            runsList_df.append(i)
+    percentList_df = []
+    for _i in range(0, runs):
+        for j in range(0, len(percentList)):
+            percentList_df.append(percentList[j])
+    fmsList_df = []
+    for sublist in fmsLists:
+        fmsList_df += sublist
+    df = pd.DataFrame(
+        {
+            "Run": runsList_df,
+            "Percentage of Data Dropped": percentList_df,
+            "FMS": fmsList_df,
+        }
+    )
+
+    return df
+
+
+def resample(data: anndata.AnnData) -> anndata.AnnData:
+    """Bootstrapping dataset"""
+    indices = np.random.randint(0, data.shape[0], size=(data.shape[0],))
+    data = data[indices].copy()
+    return data
+
+
+def fms_diff_ranks(
+    X: anndata.AnnData,
+    ranksList: list[int],
+    runs: int,
+):
+    # Plots FMS when using different Pf2 components
+    fmsLists = []
+
+    for j in range(0, runs, 1):
+        scores = []
+        for i in ranksList:
+            dataX = pf2(X, rank=i, random_state=j, doEmbedding=False)
+
+            sampledX = pf2(resample(X), rank=i, random_state=j, doEmbedding=False)
+
+            fmsScore = calculateFMS(dataX, sampledX)
+            scores.append(fmsScore)
+        fmsLists.append(scores)
+
+    runsList_df = []
+    for i in range(0, runs):
+        for _j in range(0, len(ranksList)):
+            runsList_df.append(i)
+    ranksList_df = []
+    for _i in range(0, runs):
+        for j in range(0, len(ranksList)):
+            ranksList_df.append(ranksList[j])
+    fmsList_df = []
+    for sublist in fmsLists:
+        fmsList_df += sublist
+    df = pd.DataFrame(
+        {"Run": runsList_df, "Component": ranksList_df, "FMS": fmsList_df}
+    )
+
+    return df
diff --git a/pf2rnaseq/figures/commonFuncs/plotGeneral.py b/pf2rnaseq/figures/commonFuncs/plotGeneral.py
@@ -6,7 +6,7 @@
 import seaborn as sns
 from matplotlib.axes import Axes
 
-from ...factorization import pf2_pca_r2x
+from ...factorization import fms_percent_drop, pf2_pca_r2x, fms_diff_ranks
 
 
 def plot_r2x(data, rank_vec, ax: Axes):
@@ -439,3 +439,24 @@ def plot_boxplot_gene_celltype(
     ax.set(title=gene)
     ax.set_xticks(ax.get_xticks())
     ax.set_xticklabels(labels=ax.get_xticklabels(), rotation=45)
+
+
+def plot_fms_diff_ranks(
+    X: anndata.AnnData,
+    ax: Axes,
+    ranksList: list[int],
+    runs=3,
+):
+    """Plots FMS when using different Pf2 components"""
+    df = fms_diff_ranks(X, ranksList, runs)
+    sns.lineplot(data=df, x="Component", y="FMS", ax=ax)
+    ax.set_ylim(0, 1)
+
+
+def plot_fms_percent_drop(
+    X: anndata.AnnData, ax: Axes, percentList: np.ndarray, runs=3, rank: int = 30
+):
+    """Plots FMS when dropping different percentages of data"""
+    df = fms_percent_drop(X, percentList, runs, rank)
+    sns.lineplot(data=df, x="Percentage of Data Dropped", y="FMS", ax=ax)
+    ax.set_ylim(0, 1)
diff --git a/pf2rnaseq/figures/figureHeiserFMS.py b/pf2rnaseq/figures/figureHeiserFMS.py
@@ -1,19 +1,16 @@
 """
 factorization score
+
 """
 
-import anndata
 import numpy as np
-import pandas as pd
-import scanpy as sc
-import seaborn as sns
-from matplotlib.axes import Axes
-from tensorly.cp_tensor import CPTensor
-from tlviz.factor_tools import factor_match_score as fms
 
-from ..factorization import pf2
 from ..imports import import_Heiser
 from .common import getSetup, subplotLabel
+from .commonFuncs.plotGeneral import (
+    plot_fms_diff_ranks,
+    plot_fms_percent_drop,
+)
 
 
 def makeFigure():
@@ -22,125 +19,9 @@ def makeFigure():
 
     X = import_Heiser()
     percentList = np.arange(0.0, 55.0, 5.0)
-    # plot_fms_percent_drop(X, ax[0], percentList=percentList, runs=2)
+    plot_fms_percent_drop(X, ax[0], percentList=percentList, runs=2, rank=30)
 
-    ranks = list(range(30, 51))
+    ranks = list(range(1, 31))
     plot_fms_diff_ranks(X, ax[1], ranksList=ranks, runs=2)
 
     return f
-
-
-def calculateFMS(A: anndata.AnnData, B: anndata.AnnData):
-    """Calculates FMS between 2 factors"""
-    factors = [A.uns["Pf2_A"], A.uns["Pf2_B"], A.varm["Pf2_C"]]
-    A_CP = CPTensor(
-        (
-            A.uns["Pf2_weights"],
-            factors,
-        )
-    )
-
-    factors = [B.uns["Pf2_A"], B.uns["Pf2_B"], B.varm["Pf2_C"]]
-    B_CP = CPTensor(
-        (
-            B.uns["Pf2_weights"],
-            factors,
-        )
-    )
-
-    return fms(A_CP, B_CP, consider_weights=False, skip_mode=1)  # type: ignore
-
-
-def plot_fms_percent_drop(
-    X: anndata.AnnData,
-    ax: Axes,
-    percentList: np.ndarray,
-    runs: int,
-    rank: int = 30,
-):
-    # Plots FMS score when percentage is removed from data
-    dataX = pf2(X, rank, doEmbedding=False)
-
-    fmsLists = []
-
-    for j in range(0, runs, 1):
-        scores = [1.0]
-
-        for i in percentList[1:]:
-            sampled_data: anndata.AnnData = sc.pp.subsample(
-                X, fraction=1 - (i / 100), random_state=j, copy=True
-            )  # type: ignore
-            sampledX = pf2(sampled_data, rank, random_state=j + 2, doEmbedding=False)
-
-            fmsScore = calculateFMS(dataX, sampledX)
-            scores.append(fmsScore)
-
-        fmsLists.append(scores)
-
-    runsList_df = []
-    for i in range(0, runs):
-        for j in range(0, len(percentList)):
-            runsList_df.append(i)
-    percentList_df = []
-    for i in range(0, runs):
-        for j in range(0, len(percentList)):
-            percentList_df.append(percentList[j])
-    fmsList_df = []
-    for sublist in fmsLists:
-        fmsList_df += sublist
-    df = pd.DataFrame(
-        {
-            "Run": runsList_df,
-            "Percentage of Data Dropped": percentList_df,
-            "FMS": fmsList_df,
-        }
-    )
-
-    sns.lineplot(data=df, x="Percentage of Data Dropped", y="FMS", ax=ax)
-    ax.set_ylim(0, 1)
-
-
-def resample(data: anndata.AnnData) -> anndata.AnnData:
-    """Bootstrapping dataset"""
-    indices = np.random.randint(0, data.shape[0], size=(data.shape[0],))
-    data = data[indices].copy()
-    return data
-
-
-def plot_fms_diff_ranks(
-    X: anndata.AnnData,
-    ax: Axes,
-    ranksList: list[int],
-    runs: int,
-):
-    # Plots FMS when using different Pf2 components
-    fmsLists = []
-
-    for j in range(0, runs, 1):
-        scores = []
-        for i in ranksList:
-            dataX = pf2(X, rank=i, random_state=j, doEmbedding=False)
-
-            sampledX = pf2(resample(X), rank=i, random_state=j, doEmbedding=False)
-
-            fmsScore = calculateFMS(dataX, sampledX)
-            scores.append(fmsScore)
-        fmsLists.append(scores)
-
-    runsList_df = []
-    for i in range(0, runs):
-        for j in range(0, len(ranksList)):
-            runsList_df.append(i)
-    ranksList_df = []
-    for i in range(0, runs):
-        for j in range(0, len(ranksList)):
-            ranksList_df.append(ranksList[j])
-    fmsList_df = []
-    for sublist in fmsLists:
-        fmsList_df += sublist
-    df = pd.DataFrame(
-        {"Run": runsList_df, "Component": ranksList_df, "FMS": fmsList_df}
-    )
-
-    sns.lineplot(data=df, x="Component", y="FMS", ax=ax)
-    ax.set_ylim(0, 1)
