aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--gn3/computations/gemma.py26
-rw-r--r--tests/unit/computations/test_gemma.py20
2 files changed, 46 insertions, 0 deletions
diff --git a/gn3/computations/gemma.py b/gn3/computations/gemma.py
index ae04c40..5741692 100644
--- a/gn3/computations/gemma.py
+++ b/gn3/computations/gemma.py
@@ -5,8 +5,11 @@ import string
from base64 import b64encode
from hashlib import md5
+from typing import Dict
from typing import List
+from typing import Optional
from typing import ValuesView
+from gn3.commands import compose_gemma_cmd
def generate_random_n_string(n_length: int) -> str:
"""Generate a random string that is N chars long"""
@@ -50,3 +53,26 @@ def do_paths_exist(paths: ValuesView) -> bool:
if not os.path.isfile(path):
return False
return True
+
+
+def generate_gemma_computation_cmd(gemma_cmd: str,
+ gemma_kwargs: Dict[str, str],
+ output_file: str) -> Optional[str]:
+ """Create a higher order function that generates a command"""
+ geno_filename = gemma_kwargs.get("geno_filename", "")
+ trait_filename = gemma_kwargs.get("trait_filename")
+ ext, snps_filename = geno_filename.partition(".")[-1], ""
+ if geno_filename:
+ snps_filename = geno_filename.replace(f".{ext}", "")
+ snps_filename += f"_snps.{ext}"
+ _kwargs = {"g": geno_filename, "p": trait_filename}
+ if gemma_kwargs.get("covar_filename"):
+ _kwargs["a"] = gemma_kwargs.get("covar_filename")
+ if not do_paths_exist(_kwargs.values()): # Prevents injection!
+ return None
+ if _kwargs.get("lmm"):
+ _kwargs["lmm"] = gemma_kwargs.get("lmm")
+ return compose_gemma_cmd(gemma_wrapper_cmd=gemma_cmd,
+ gemma_kwargs=_kwargs,
+ gemma_args=["-gk", ">",
+ output_file])
diff --git a/tests/unit/computations/test_gemma.py b/tests/unit/computations/test_gemma.py
index 23707ef..a7884ea 100644
--- a/tests/unit/computations/test_gemma.py
+++ b/tests/unit/computations/test_gemma.py
@@ -4,6 +4,7 @@ import unittest
from unittest import mock
from gn3.computations.gemma import generate_hash_of_string
from gn3.computations.gemma import generate_pheno_txt_file
+from gn3.computations.gemma import generate_gemma_computation_cmd
class TestGemma(unittest.TestCase):
@@ -29,3 +30,22 @@ class TestGemma(unittest.TestCase):
"""Test that a string is hashed correctly"""
self.assertEqual(generate_hash_of_string("I^iQP&TlSR^z"),
"hMVRw8kbEp49rOmoIkhMjA")
+
+ @mock.patch("gn3.computations.gemma.do_paths_exist")
+ def test_compose_k_computation_cmd(self, mock_pathsp):
+ """Test that a K computation cmd is constructed properly"""
+ mock_pathsp.return_value = True
+ self.assertEqual(
+ generate_gemma_computation_cmd(
+ gemma_cmd="gemma-wrapper",
+ gemma_kwargs={
+ "geno_filename": "genofile.txt",
+ "trait_filename": "test.txt",
+ "covar_filename": "genofile_snps.txt"},
+ output_file="/tmp/gn2/k_output_gUFhGu4rLG7k+CXLPk1OUg.txt",
+ ),
+ ("gemma-wrapper --json -- "
+ "-g genofile.txt -p "
+ "test.txt -a genofile_snps.txt "
+ "-gk > /tmp/gn2/"
+ "k_output_gUFhGu4rLG7k+CXLPk1OUg.txt"))