refactor: use new algorithm system

Author: tristan-f-r

spras/tiedie.py

@@ -1,6 +1,8 @@
 from pathlib import Path
+from typing import Optional
 
 import pandas as pd
+from pydantic import BaseModel, ConfigDict
 
 from spras.config.container_schema import ProcessedContainerSettings
 from spras.containers import prepare_volume, run_container_and_log
@@ -11,9 +13,35 @@
 from spras.prm import PRM
 from spras.util import add_rank_column, duplicate_edges, raw_pathway_df
 
-__all__ = ["TieDIE"]
+__all__ = ["TieDIE", "TieDIEParams"]
 
-class TieDIE(PRM):
+class TieDIEParams(BaseModel):
+    s: float = 1
+    """Network size control factor"""
+
+    d_expr: Optional[str] = None
+    """List of significantly differentially expressed genes,
+    along with log-FC or FC values (i.e. by edgeR for RNA-Seq or SAM for microarray data.)
+    Generated by a sample-dichotomy of interest."""
+
+    a: Optional[float] = None
+    """Linker Cutoff (overrides the Size factor)"""
+
+    c: int = 3
+    """Search depth for causal paths"""
+
+    p: int = 1000
+    """Number of random permutations performed for significance analysis"""
+
+    pagerank: bool = False
+    """Use Personalized PageRank to Diffuse """
+
+    all_paths: bool = False
+    """Use all paths instead of only causal paths"""
+
+    model_config = ConfigDict(extra='forbid', use_attribute_docstrings=True)
+
+class TieDIE(PRM[TieDIEParams]):
     # we need edges (weighted), source set (with prizes), and target set (with prizes).
     required_inputs = ["edges", "sources", "targets"]
     dois = ["10.1093/bioinformatics/btt471"]
@@ -24,6 +52,9 @@ def generate_inputs(data, filename_map):
         Access fields from the dataset and write the required input files
         @param data: dataset
         @param filename_map: a dict mapping file types in the required_inputs to the filename for that type
+        - source: input node types with sources (required)
+        - target: input node types with targets (required)
+        - edges: input edges file (required)
         """
         # ensures the required input are within the filename_map
         for input_type in TieDIE.required_inputs:
@@ -62,39 +93,22 @@ def generate_inputs(data, filename_map):
 
     # Skips parameter validation step
     @staticmethod
-    def run(edges=None, sources=None, targets=None, output_file=None, s: float = 1.0 , c: int = 3 , p: int = 1000, pagerank: bool = False, all_paths: bool = False, container_settings=None):
-        """
-        Run TieDIE with Docker
-        @param source: input node types with sources (required)
-        @param target: input node types with targets (required)
-        @param edges: input edges file (required)
-        @param output_file: path to the output pathway file (required)
-        @param s: Network size control factor (optional) (default 1)
-        @param d_expr: List of significantly differentially expressed genes, along with log-FC or FC values (i.e. by edgeR for RNA-Seq or SAM for microarray data.) Generated by a sample-dichotomy of interest. (optional)
-        @param a: Linker Cutoff (overrides the Size factor) (optional)
-        @param c: Search depth for causal paths (optional) (default 3)
-        @param p: Number of random permutations performed for significance analysis (optional) (default 1000)
-        @param pagerank: Use Personalized PageRank to Diffuse (optional)
-        @param all_paths: Use all paths instead of only causal paths (optional) (default False)
-        @param singularity: if True, run using the Singularity container instead of the Docker container
-        """
-
+    def run(inputs, output_file, args=None, container_settings=None):
+        if not args: args = TieDIEParams()
         if not container_settings: container_settings = ProcessedContainerSettings()
-        if not edges or not sources or not targets or not output_file:
-            raise ValueError("Required TieDIE arguments are missing")
 
         work_dir = "/spras"
 
         # Each volume is a tuple (src, dest) - data generated by Docker
         volumes = list()
 
-        bind_path, edges_file = prepare_volume(edges, work_dir, container_settings)
+        bind_path, edges_file = prepare_volume(inputs["edges"], work_dir, container_settings)
         volumes.append(bind_path)
 
-        bind_path, sources_file = prepare_volume(sources, work_dir, container_settings)
+        bind_path, sources_file = prepare_volume(inputs["sources"], work_dir, container_settings)
         volumes.append(bind_path)
 
-        bind_path, targets_file = prepare_volume(targets, work_dir, container_settings)
+        bind_path, targets_file = prepare_volume(inputs["targets"], work_dir, container_settings)
         volumes.append(bind_path)
 
         out_dir = Path(output_file).parent
@@ -110,11 +124,11 @@ def run(edges=None, sources=None, targets=None, output_file=None, s: float = 1.0
             "--up_heats", sources_file,
             "--down_heats", targets_file,
             "--network", edges_file,
-            "--size", str(s),
-            "--depth", str(c),
-            "--permute", str(p),
-            "--pagerank", "True" if pagerank else "False",
-            "--all_paths", "False" if all_paths else "False",
+            "--size", str(args.s),
+            "--depth", str(args.c),
+            "--permute", str(args.p),
+            "--pagerank", "True" if args.pagerank else "False",
+            "--all_paths", "False" if args.all_paths else "False",
             "--output_folder", mapped_out_dir,
         ]
 

test/TieDIE/test_tiedie.py

@@ -4,7 +4,7 @@
 import pytest
 
 from spras.config.container_schema import ContainerFramework, ProcessedContainerSettings
-from spras.tiedie import TieDIE
+from spras.tiedie import TieDIE, TieDIEParams
 
 TEST_DIR = Path('test', 'TieDIE')
 OUT_FILES = TEST_DIR / 'output' / 'output1' / 'tiedie_pathway.txt'
@@ -19,68 +19,70 @@ def test_tiedie_required(self):
         out_path = Path(OUT_FILES)
         out_path.unlink(missing_ok=True)
         # Only include required arguments
-        TieDIE.run(sources=TEST_DIR / 'input' / 'source1.txt',
-                    targets=TEST_DIR / 'input' / 'target1.txt',
-                    edges=TEST_DIR / 'input' / 'pathway1.txt',
+        TieDIE.run({"sources": TEST_DIR / 'input' / 'source1.txt',
+                    "targets": TEST_DIR / 'input' / 'target1.txt',
+                    "edges": TEST_DIR / 'input' / 'pathway1.txt'},
                     output_file=OUT_FILES)
         assert out_path.exists()
 
     def test_tiedie_alternative_graph(self):
         out_path = Path(OUT_FILES_1)
         out_path.unlink(missing_ok=True)
-        TieDIE.run(sources=TEST_DIR / 'input' / 'source2.txt',
-                        targets=TEST_DIR / 'input' / 'target2.txt',
-                        edges=TEST_DIR / 'input' / 'pathway2.txt',
-                        output_file=OUT_FILES_1)
+        TieDIE.run({"sources": TEST_DIR / 'input' / 'source2.txt',
+                    "targets": TEST_DIR / 'input' / 'target2.txt',
+                    "edges": TEST_DIR / 'input' / 'pathway2.txt'},
+                    output_file=OUT_FILES_1)
         assert out_path.exists()
 
     def test_tiedie_some_optional(self):
         out_path = Path(OUT_FILES)
         out_path.unlink(missing_ok=True)
         # Include optional argument
-        TieDIE.run(sources=TEST_DIR / 'input' / 'source1.txt',
-                        targets=TEST_DIR / 'input' / 'target1.txt',
-                        edges=TEST_DIR / 'input' / 'pathway1.txt',
-                        output_file=OUT_FILES,
+        TieDIE.run({"sources": TEST_DIR / 'input' / 'source1.txt',
+                    "targets": TEST_DIR / 'input' / 'target1.txt',
+                    "edges": TEST_DIR / 'input' / 'pathway1.txt'},
+                    output_file=OUT_FILES,
+                    args=TieDIEParams(
                         s=1.1,
                         p=2000,
-                        pagerank = True)
+                        pagerank=True))
         assert out_path.exists()
 
     def test_tiedie_all_optional(self):
         out_path = Path(OUT_FILES)
         out_path.unlink(missing_ok=True)
         # Include optional argument
-        TieDIE.run(sources=TEST_DIR / 'input' / 'source1.txt',
-                        targets=TEST_DIR / 'input' / 'target1.txt',
-                        edges=TEST_DIR / 'input' / 'pathway1.txt',
-                        output_file=OUT_FILES,
-                        s=1.1,
+        TieDIE.run({"sources": TEST_DIR / 'input' / 'source1.txt',
+                    "targets": TEST_DIR / 'input' / 'target1.txt',
+                    "edges": TEST_DIR / 'input' / 'pathway1.txt'},
+                    output_file=OUT_FILES,
+                    args=TieDIEParams(s=1.1,
                         c=4,
                         p=2000,
                         pagerank=True,
-                        all_paths=True)
+                        all_paths=True))
         assert out_path.exists()
 
     def test_tiedie_missing(self):
         # Test the expected error is raised when required arguments are missing
         with pytest.raises(ValueError):
             # No edges file
-            TieDIE.run(sources=TEST_DIR / 'input' / '/source1.txt',
-                            targets=TEST_DIR / 'input' / '/target1.txt',
-                            output_file=OUT_FILES)
+            TieDIE.run({"sources": TEST_DIR / 'input' / '/source1.txt',
+                        "targets": TEST_DIR / 'input' / '/target1.txt'},
+                        output_file=OUT_FILES)
 
     @pytest.mark.skipif(not shutil.which('singularity'), reason='Singularity not found on system')
     def test_tiedie_singularity(self):
         out_path = Path(OUT_FILES)
         out_path.unlink(missing_ok=True)
         # Only include required arguments and run with Singularity
-        TieDIE.run(sources=TEST_DIR / 'input' / 'source1.txt',
-                        targets=TEST_DIR / 'input' / 'target1.txt',
-                        edges=TEST_DIR / 'input' / 'pathway1.txt',
-                        output_file=OUT_FILES,
+        TieDIE.run({"sources": TEST_DIR / 'input' / 'source1.txt',
+                    "targets": TEST_DIR / 'input' / 'target1.txt',
+                    "edges": TEST_DIR / 'input' / 'pathway1.txt'},
+                    output_file=OUT_FILES,
+                    args=TieDIEParams(
                         s=1.1,
                         p=2000,
-                        pagerank=True,
-                        container_settings=ProcessedContainerSettings(framework=ContainerFramework.singularity))
+                        pagerank=True),
+                    container_settings=ProcessedContainerSettings(framework=ContainerFramework.singularity))
         assert out_path.exists()