-
Notifications
You must be signed in to change notification settings - Fork 1
/
hifisr.py
136 lines (115 loc) · 4.46 KB
/
hifisr.py
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
#!/usr/bin/env python3
"""
Author : zouyi <[email protected]>
Date : 2022-02-15
Purpose: The Python user interface of the HiFi-SR pipeline
"""
import argparse
from typing import NamedTuple
import os
import subprocess
class Args(NamedTuple):
""" Command-line arguments """
mode: str # single or merge
sample_name: str
num_of_thread: int
input_type: str
type_number: int
samples_txt_prefix: str
# --------------------------------------------------
def get_args() -> Args:
""" Get command-line arguments """
parser = argparse.ArgumentParser(
description='The Python user interface of the HiFi-SR pipeline',
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument('mode',
metavar='mode',
type=str,
help='"single" to run hifisr for each sample; "merge" to merge results of multiple samples')
parser.add_argument('-s',
'--sample_name',
help='In "single" mode, sample name, for example sample_1',
metavar='str',
type=str,
default='')
parser.add_argument('-t',
'--num_of_thread',
help='In "single" mode, number of threads to run the pipeline',
metavar='int',
type=int,
default=8)
parser.add_argument('-i',
'--input_type',
help='In "single" mode, input file type fastq or fasta',
metavar='str',
type=str,
default='fastq')
parser.add_argument('-n',
'--type_number',
help='In "single" mode, analyze reads up to four rearrangements in a single read',
metavar='int',
type=int,
default=5)
parser.add_argument('-m',
'--samples_txt_prefix',
help='In "merge" mode, prefix of txt file containing multiple sample names, for example "merge_1"',
metavar='str',
type=str,
default='')
args = parser.parse_args()
return Args(args.mode, args.sample_name, args.num_of_thread, args.input_type, args.type_number, args.samples_txt_prefix)
# --------------------------------------------------
def main() -> None:
"""Run HiFi-SP pipeline in for a single sample, OR merge reports for multiple samples."""
args = get_args()
mode_arg = args.mode
sample_arg = args.sample_name
thread_arg = args.num_of_thread
input_type_arg = args.input_type
# type_number_arg = args.type_number
samples_txt_prefix_arg = args.samples_txt_prefix
# for "single" mode
if mode_arg == "single" and sample_arg == "":
print()
print("Missing sample name")
print()
exit()
elif mode_arg == "single":
print()
print(f"Running HiFi-SR pipeline for {sample_arg}")
print()
ret = subprocess.call(f"bash ../../scripts/run_stage_all.sh {sample_arg} {thread_arg} {input_type_arg}", shell=True)
print()
print(f"Finished running HiFi-SR for {sample_arg}!")
print()
# for "wp2" mode
if mode_arg == "wp2" and sample_arg == "":
print()
print("Missing sample name")
print()
exit()
elif mode_arg == "wp2":
print()
print(f"Running HiFi-SR pipeline for {sample_arg}, mapping reads by winnowmap2")
print()
ret = subprocess.call(f"bash ../../scripts/run_stage_wp2.sh {sample_arg} {thread_arg} {input_type_arg}", shell=True)
print()
print(f"Finished running HiFi-SR for {sample_arg}!")
print()
# for "merge" mode
if mode_arg == "merge" and samples_txt_prefix_arg == "":
print()
print("Missing txt file prefix")
print()
exit()
elif mode_arg == "merge":
print()
print(f"Merge HiFi-SR reports for multiple samples in {samples_txt_prefix_arg}")
print()
ret = subprocess.call(f"bash ../scripts/merge_samples.sh {samples_txt_prefix_arg}", shell=True)
print()
print(f"Finished merging {samples_txt_prefix_arg}!")
print()
# --------------------------------------------------
if __name__ == '__main__':
main()