-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathgenerate_feature_file.py
165 lines (129 loc) · 4.91 KB
/
generate_feature_file.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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
#MIT Open License
#Manisha
import pickle
import sys
import os
import subprocess
import shutil
from optparse import OptionParser
from shutil import copy2
def feature_generation(curr_dir1,output_file, feature_dir):
#copy list.txt to features/ and features/kanalyze-2.0.0/code
feature_destpath = os.path.join(curr_dir1, feature_dir)
#print(f'The feature dest path : {feature_destpath}')
kanalyzer_destpath = feature_destpath + "/kanalyze-2.0.0/code/"
kanalyzer_input_destpath = feature_destpath + "/kanalyze-2.0.0/input_data/"
kanalyzer_output_destpath = feature_destpath + "/kanalyze-2.0.0/output_data/"
mer2_dir = kanalyzer_output_destpath + '2mer/'
mer3_dir = kanalyzer_output_destpath + '3mer/'
mer4_dir = kanalyzer_output_destpath + '4mer/'
if os.path.isdir(mer2_dir):
subprocess.run(['rm','-R',kanalyzer_output_destpath + '2mer'])
os.mkdir(mer2_dir)
else:
os.mkdir(mer2_dir)
if os.path.isdir(mer3_dir):
subprocess.run(['rm','-R',kanalyzer_output_destpath + '3mer'])
os.mkdir(mer3_dir)
else:
os.mkdir(mer3_dir)
if os.path.isdir(mer4_dir):
subprocess.run(['rm','-R',kanalyzer_output_destpath + '4mer'])
os.mkdir(mer4_dir)
else:
os.mkdir(mer4_dir)
#chmod 775 runKanalyzer_generate_all_features and run this script
curr_dir = os.getcwd()
kanalyzer_dir = os.chdir(kanalyzer_destpath)
curr_dir = os.getcwd() + "/"
subprocess.run(['chmod', '775', 'runKanalyzer_generate_all_features'])
subprocess.run(['./runKanalyzer_generate_all_features'])
change_dir = os.chdir(feature_destpath)
subprocess.run(['javac','KmersFeaturesCollector.java'])
subprocess.run(['javac','BufferReaderAndWriter.java'])
subprocess.run(['java', 'KmersFeaturesCollector'])
if not output_file == "feature_file.csv":
subprocess.run(['mv', 'feature_file.csv', output_file])
curr_dir2 = os.getcwd()
change_dir = os.chdir(curr_dir1)
data_dir = 'data/'
shutil.copy2(curr_dir2+'/' + output_file, data_dir+output_file)
change_dir = os.chdir(feature_destpath)
subprocess.run(['rm', output_file])
change_dir = os.chdir(curr_dir1)
def get_data(fasta_file, feature_dir):
curr_dir1 = os.getcwd()
data_filepath = curr_dir1 + "/data/"
feature_destpath = os.path.join(curr_dir1,feature_dir)
kanalyzer_destpath = feature_destpath + "/kanalyze-2.0.0/code/"
kanalyzer_input_destpath = feature_destpath + "/kanalyze-2.0.0/input_data/"
kanalyzer_output_destpath = feature_destpath + "/kanalyze-2.0.0/output_data/"
if os.path.isdir(kanalyzer_input_destpath):
subprocess.run(['rm','-R',kanalyzer_input_destpath])
os.mkdir(kanalyzer_input_destpath)
else:
os.mkdir(kanalyzer_input_destpath)
if os.path.isdir(kanalyzer_output_destpath):
subprocess.run(['rm','-R',kanalyzer_output_destpath])
os.mkdir(kanalyzer_output_destpath)
else:
os.mkdir(kanalyzer_output_destpath)
sequence = ""
with open(data_filepath + fasta_file, 'rt') as fp:
content = fp.read()
data = content.split(">")
#print(data)
i=0
for line in data:
meta = line
if meta != "":
i+=1
of = open(kanalyzer_input_destpath + "seq" + str(i)+".fasta" ,"w")
of.write(">" + meta)
of.close()
fp.close()
curr_dir2 = os.path.join(os.getcwd(),feature_dir) + "/kanalyze-2.0.0/input_data/"
change_dir = os.chdir(curr_dir2)
#files = [f for f in os.listdir('.') if os.path.isfile(f)]
_, _, files = next(os.walk(curr_dir2))
files = sorted(files)
with open("list.txt", "w") as ff:
for f in files:
ff.write(f)
ff.write('\n')
ff.close()
shutil.copy2('./list.txt', feature_destpath+"list.txt")
shutil.copy2('./list.txt', kanalyzer_destpath+"list.txt")
subprocess.run(['rm', 'list.txt'])
def main():
curr_dir1 = os.getcwd()
parser = OptionParser()
parser.add_option("-f", "--filename", dest="filename", help="Name of the fasta file.")
parser.add_option("-o", "--output", dest="output_filename", help="Name of feature file", default="feature_file.csv")
parser.add_option("-d", "--featuredir", dest="feature_dir", help="feature directory.", default="features")
(options, args) = parser.parse_args()
fasta_file = options.filename
output_file = options.output_filename
feature_dir = os.path.join(os.getcwd(),options.feature_dir) + '/'
src = os.getcwd() + "/features/"
if not options.feature_dir == "features":
destination = shutil.copytree(src, feature_dir)
get_data(fasta_file, feature_dir)
#if we want to provide sequence in the command line, uncomment following
# te_id = sys.argv[1]
# fasta = sys.argv[2]
# fasta_filepath = "fasta"
# if not os.path.isdir(fasta_filepath):
# os.mkdir(fasta_filepath)
# fasta_filename = str(te_id) + ".fasta"
# with open(os.path.join(fasta_filepath, fasta_filename), "w") as fd:
# fd.write(">" + str(te_id) + "\n")
# fd.write(fasta)
# fd.close()
# with open("list.txt", "w") as f:
# f.write(str(te_id)+ ".fasta")
# f.close()
feature_generation(curr_dir1, output_file, feature_dir)
#subprocess.run(['python', 'evaluate.py','-f','feature_file.csv','-n','node.txt', 'm', ''])
if __name__ == '__main__':
main()