blob: fd508faa85e1d05c9f4c36b05bed0d209dabfa42 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
|
"""module contains code to preprocess and call wgcna script"""
import os
import json
import uuid
from gn3.settings import TMPDIR
from gn3.commands import run_cmd
def dump_wgcna_data(request_data: dict):
"""function to dump request data to json file"""
filename = f"{str(uuid.uuid4())}.json"
temp_file_path = os.path.join(TMPDIR, filename)
with open(temp_file_path, "w") as output_file:
json.dump(request_data, output_file)
return temp_file_path
def compose_wgcna_cmd(rscript_path: str, temp_file_path: str):
"""function to componse wgcna cmd"""
# (todo):issue relative paths to abs paths
cmd = f"Rscript ./scripts/{rscript_path} {temp_file_path}"
return cmd
def call_wgcna_script(rscript_path: str, request_data: dict):
"""function to call wgcna script"""
generated_file = dump_wgcna_data(request_data)
cmd = compose_wgcna_cmd(rscript_path, generated_file)
try:
run_cmd_results = run_cmd(cmd)
with open(generated_file, "r") as outputfile:
if run_cmd_results["code"] != 0:
return run_cmd_results
return {
"data": json.load(outputfile),
**run_cmd_results
}
except FileNotFoundError:
# relook at handling errors gn3
return {
"output": "output file not found"
}
|
