including usage of decoupler results from anndata directly

Author: vckraemer

src/intratalkerpy/tf/IntraTalker_analysis.py

@@ -8,6 +8,7 @@
 from statsmodels.stats.multitest import multipletests
 
 from .utils import validate_input_arguments, AverageExpression
+from .utils import load_tf_activities
 from .get_significant_TFs import get_significant_tfs
 from .generate_intracellular_network import generate_intracellular_network, generate_CrossTalkeR_input
 from .plot import plot_condition_tf_activities
@@ -60,27 +61,8 @@ def IntraTalker_analysis(anndataobject, tf_activities = None, arguments_list = N
     condition = anndataobject.obs[arguments_list["condition"]]
 
     #load tf activities(csv input)
-    if isinstance(tf_activities, str):
-         if arguments_list["decoupler_matrix_format"] == None or arguments_list["decoupler_matrix_format"] == "Python":
-            tf_activities = ad.io.read_csv(tf_activities)
-            tf_activities.obs = anndataobject.obs.reindex(tf_activities.obs.index)
-            tf_activities.obsm = anndataobject.obsm
-            tf_activities.uns = anndataobject.uns
-
-        #if the R version of decoupler was used, the tf activity matrix needs to be transposed
-         elif arguments_list["decoupler_matrix_format"] == "R":
-            tf_activities = pd.read_csv(tf_activities, index_col=0)
-            tf_activities = tf_activities.T
-            tf_activities = ad.AnnData(tf_activities)
-            tf_activities.obs = anndataobject.obs.reindex(tf_activities.obs.index)
-            tf_activities.obsm = anndataobject.obsm
-            tf_activities.uns = anndataobject.uns
-
-
-    elif tf_activities is None:
-         raise NameError("Please attach a csv file with the tf activity values. (For further clarification view the 'Decoupler' section of the vignette.)")
-    
-    
+    tf_activities = load_tf_activities(anndataobject, tf_activities, arguments_list)
+
     sc.pp.scale(tf_activities)
 
     #sets the stage for decision if single condition or comparison analysis is done

src/intratalkerpy/tf/utils.py

@@ -1,3 +1,4 @@
+import anndata as ad
 import pandas as pd
 import numpy as np
 import seaborn as sns
@@ -8,6 +9,52 @@
 from .plot import plot_top_variable_tfs
 
 
+def load_tf_activities(anndataobject, tf_activities, arguments_list):
+  """
+  Load TF activity values from an explicit input or fall back to
+  anndataobject.obsm['score_ulm'] when available.
+  """
+  if isinstance(tf_activities, str):
+    if arguments_list["decoupler_matrix_format"] == None or arguments_list["decoupler_matrix_format"] == "Python":
+      tf_activities = ad.io.read_csv(tf_activities)
+      tf_activities.obs = anndataobject.obs.reindex(tf_activities.obs.index)
+      tf_activities.obsm = anndataobject.obsm
+      tf_activities.uns = anndataobject.uns
+
+    #if the R version of decoupler was used, the tf activity matrix needs to be transposed
+    elif arguments_list["decoupler_matrix_format"] == "R":
+      tf_activities = pd.read_csv(tf_activities, index_col=0)
+      tf_activities = tf_activities.T
+      tf_activities = ad.AnnData(tf_activities)
+      tf_activities.obs = anndataobject.obs.reindex(tf_activities.obs.index)
+      tf_activities.obsm = anndataobject.obsm
+      tf_activities.uns = anndataobject.uns
+
+  elif tf_activities is None:
+    if "score_ulm" in anndataobject.obsm:
+      score_ulm = anndataobject.obsm["score_ulm"]
+      tf_activities = score_ulm
+      if not isinstance(tf_activities, ad.AnnData):
+        tf_activities = ad.AnnData(tf_activities)
+        if hasattr(score_ulm, "index"):
+          tf_activities.obs_names = score_ulm.index
+        else:
+          tf_activities.obs_names = anndataobject.obs_names
+
+        if hasattr(score_ulm, "columns"):
+          tf_activities.var_names = score_ulm.columns
+
+      tf_activities.obs = anndataobject.obs.reindex(tf_activities.obs.index)
+      tf_activities.obsm = anndataobject.obsm
+      tf_activities.uns = anndataobject.uns
+    else:
+      raise NameError(
+        "Please attach a csv file with the tf activity values. (For further clarification view the 'Decoupler' section of the vignette.)"
+      )
+
+  return tf_activities
+
+
 def load_csv(filename):
     """
     Description:

tests/test_package_smoke.py

@@ -1,6 +1,9 @@
 import csv
 from importlib import resources
 
+import anndata as ad
+import pandas as pd
+
 
 def test_public_modules_import_and_expose_expected_api():
     import intratalkerpy
@@ -22,6 +25,29 @@ def test_external_dependency_functions_are_callable():
     assert callable(colDeltaCorpartial)
 
 
+def test_score_ulm_is_used_when_tf_activities_is_missing():
+    from intratalkerpy.tf.utils import load_tf_activities
+
+    anndataobject = ad.AnnData(
+        X=pd.DataFrame([[1, 2], [3, 4]], index=["cell1", "cell2"], columns=["gene1", "gene2"])
+    )
+    anndataobject.obs["celltype"] = ["a", "b"]
+    anndataobject.obs["condition"] = ["x", "y"]
+    anndataobject.obsm["score_ulm"] = pd.DataFrame(
+        [[0.1, 0.2], [0.3, 0.4]], index=anndataobject.obs_names, columns=["tf1", "tf2"]
+    )
+
+    tf_activities = load_tf_activities(
+        anndataobject,
+        None,
+        {"decoupler_matrix_format": None},
+    )
+
+    assert isinstance(tf_activities, ad.AnnData)
+    assert list(tf_activities.obs_names) == ["cell1", "cell2"]
+    assert list(tf_activities.var_names) == ["tf1", "tf2"]
+
+
 def test_transcription_factor_reference_data_is_packaged():
     data_files = [
         "rtf_db_human.csv",