tes-demo/main.py at master · genxnetwork/tes-demo · GitHub

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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
import requests
import json
from requests.auth import HTTPBasicAuth

# Define AWS S3 credentials and region
AWS_ACCESS_KEY_ID = ""
AWS_SECRET_ACCESS_KEY = ""
AWS_REGION = "us-east-1"

DOCKER_REGISTRY = "docker.io"
DOCKER_LOGIN = ""
DOCKER_PASS = ""

# Your GENXT credentials
username = ''
password = ''

# Initialize JSON Data
task_data = {
    # Task name
    "name": "GRAPE run",

    # Persistent volume limit
    "resources": {"disk_gb": 200},

    # Persistent volume
    "volumes": ["/vol/a/"],

    "executors": []
}

# Add Executors - steps of the workflow

# Downloading the input files from AWS S3 to persistent volume using S3 credentials
task_data["executors"].append({
    "image": "amazon/aws-cli",
    "command": ["aws", "s3", "cp", "s3://grapetestbucket/input.vcf.gz", "/vol/a/input.vcf.gz"],
    "env": {
        "AWS_ACCESS_KEY_ID": AWS_ACCESS_KEY_ID,
        "AWS_SECRET_ACCESS_KEY": AWS_SECRET_ACCESS_KEY,
        "AWS_REGION": AWS_REGION
    }
})

# Running GRAPE software. Step 1: reference panel download
task_data["executors"].append({
    "docker_login": DOCKER_LOGIN,
    "docker_pass": DOCKER_PASS,
    "docker_registry": DOCKER_REGISTRY,

    "image": "genxnetwork/grape",
    "command": [
        "python",
        "launcher.py",
        "reference",
        "--use-bundle",
        "--ref-directory",
        "/vol/a/media/ref",
        "--real-run"
    ]
})

# Running GRAPE software. Step 2: preprocessing of the input file
task_data["executors"].append({
    "docker_login": DOCKER_LOGIN,
    "docker_pass": DOCKER_PASS,
    "docker_registry": DOCKER_REGISTRY,

    "image": "genxnetwork/grape",
    "command": [
        "python",
        "launcher.py",
        "preprocess",
        "--ref-directory",
        "/vol/a/media/ref",
        "--vcf-file",
        "/vol/a/input.vcf.gz",
        "--directory",
        "/vol/a/media/data",
        "--assembly",
        "hg37",
        "--real-run"
    ]
})

# Running GRAPE software. Step 3: relatives search using IBIS flow
task_data["executors"].append({
    "docker_login": DOCKER_LOGIN,
    "docker_pass": DOCKER_PASS,
    "docker_registry": DOCKER_REGISTRY,

    "image": "genxnetwork/grape",
    "command": [
        "python",
        "launcher.py",
        "find",
        "--flow",
        "ibis",
        "--ref-directory",
        "/vol/a/media/ref",
        "--directory",
        "/vol/a/media/data",
        "--real-run"
    ]
})

# Uploading the result files to AWS S3 bucket from persistent volume
task_data["executors"].append({
    "image": "amazon/aws-cli",
    "command": ["aws", "s3", "cp", "/vol/a/media/data/results/relatives.tsv", "s3://grapetestbucket/relatives_output.tsv"],
    "env": {
        "AWS_ACCESS_KEY_ID": AWS_ACCESS_KEY_ID,
        "AWS_SECRET_ACCESS_KEY": AWS_SECRET_ACCESS_KEY,
        "AWS_REGION": AWS_REGION
    }
})

# Print the whole JSON structure
# print(json.dumps(task_data, indent=4))

# Requesting TES API - task creation
url = "http://tesktest.genxt.cloud/ga4gh/tes/v1/tasks"
headers = {
    "accept": "application/json",
    "Content-Type": "application/json"
}
response = requests.post(url, headers=headers, data=json.dumps(task_data), auth=HTTPBasicAuth(username, password))
print("POST request response:", response.json())
task_id = response.json().get("id")

# Requesting TES API - task status
url = f"http://tesktest.genxt.cloud/ga4gh/tes/v1/tasks/{task_id}"
headers = {
    "accept": "application/json",
    "Content-Type": "application/json"
}
response = requests.get(url, headers=headers, auth=HTTPBasicAuth(username, password))
print("GET request response:", response.json())

# Requesting TES API - task list
url = f"http://tesktest.genxt.cloud/ga4gh/tes/v1/tasks"
headers = {
    "accept": "application/json",
    "Content-Type": "application/json"
}
response = requests.get(url, headers=headers, auth=HTTPBasicAuth(username, password))
all_tasks = response.json().get("tasks")
for task in all_tasks:
    print(task)