-
Notifications
You must be signed in to change notification settings - Fork 5
Expand file tree
/
Copy pathtestMergedb
More file actions
executable file
·208 lines (184 loc) · 8.35 KB
/
Copy pathtestMergedb
File metadata and controls
executable file
·208 lines (184 loc) · 8.35 KB
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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
#!/usr/bin/env python
'''
The testMergedb program runs the Mergedb program repeatedly on a set
of UniProtKB/SwissProt and UniProt-GOA annoation file pairs.
How to run this program:
python testMergedb -I=dataset-testmg-yeast.txt -G=559292 -O=output-testmg-yeast.txt
dataset-testmg-yeast.txt lists the pairs of UniProtKB/SwissProt and
UniProt-GOA annotatation file names.
-G=559292 tells testMergedb to execute the Mergedb Program on for the
organsim with taxon id 559292.
output-testmg-yeast.txt will have all the messages coming from running
the testMergedb program.
'''
import os
import sys
import subprocess
from os.path import basename
import urllib2
import ArgParser_testMergedb as ap
import Config
config_filename = '.cafarc' # Default configuration file name
class testMergedb:
def __init__(self):
self.parsed_dict = ap.parse_args()
self.ConfigParam = Config.read_config(config_filename)
self.work_dir = (self.ConfigParam['workdir'].rstrip('/'))
if not os.path.exists(self.work_dir):
os.makedirs(self.work_dir)
# Create work direcoty, if it does not exist
self.testData_filename = self.work_dir + '/' + basename(self.parsed_dict['input1'])
self.output_filename = self.work_dir + '/' + basename(self.parsed_dict['output1'])
self.taxon_id = self.parsed_dict['taxon'] # The benchmark verion number that will be verified
self.goa_arc = 'ftp://ftp.ebi.ac.uk/pub/databases/GO/goa/old'
self.taxon_dict = OrderedDict()
self.taxon_dict['arabidopsis'] = 3702
self.taxon_dict['chicken'] = 9031
self.taxon_dict['cow'] = 9913
self.taxon_dict['dicty'] = 44689
self.taxon_dict['dog'] = 9615
self.taxon_dict['fly'] = 7227
self.taxon_dict['human'] = 9606
self.taxon_dict['mouse'] = 10090
self.taxon_dict['pig'] = 9823
self.taxon_dict['rat'] = 10116
self.taxon_dict['worm'] = 6239
self.taxon_dict['yeast'] = 559292
self.taxon_dict['zebrafish'] = 7955
def download_testDataset(self, testDataset_fh):
'''
This method iterates through the list of the testdata file names.
If it does not find any of the testdata files on workspace, it
downloads it from the UniProtKB/SwissProt archive.
'''
for line in testDataset_fh:
sprot_fname = line.strip().lower()
goa_fname = (next(testDataset_fh)).strip().lower()
print sprot_fname
print goa_fname
break
def download_goa_file(self, goa_fname):
'''
This method returns True if the goa_fname is already in
the workspace or it successfully downloaded the file
onto the workspace.
Otherwise, it returns False.
'''
if (not os.path.isfile(self.work_dir + '/' + goa_fname)):
# Organism specific folder name at UniProt-GOA archive:
folder_name = line.strip().split('.')[1].split('_')[-1]
# Organism specific archived file name at UniProt-GOA archive:
fname = line.strip() + '.gz'
# Organism specific URL at UniProt-GOA archive:
url = self.goa_arc + '/' + folder_name.upper()
# Download the file from the UniProt-GOA archive:
print('Downloading ' + fname + ' ... ')
if (not self.download(url, fname)):
print('Downloading failed for ' + fname)
return False
os.system('gzip -d ' + self.work_dir + '/' + fname)
return True
def download_sprot_file(self, sprot_fname):
'''
This method returns True if the sprot_fname is already in
the workspace or it successfully downloaded the file
from the UniProtKB/SwissProt archive onto the workspace.
Otherwise, it returns False.
'''
testdata_fname = sprot-fname
release_name = 'release-' + testdata_fname.split('.dat.')[1]
folder_name = release_name + '/' + \
'knowledgebase'
url = self.sprot_arc_base + '/' + folder_name
download_fname = testdata_fname.split('.dat.')[0] + '-only' + \
testdata_fname.split('.dat.')[1] + '.tar.gz'
tar_fname = download_fname.rstrip('.gz')
zip_fname = 'uniprot_sprot.dat.gz'
if (os.path.isfile(self.work_dir + '/' + testdata_fname)):
pass
elif (os.path.isfile(self.work_dir + '/' + tar_fname)):
os.system('tar -xC ' + self.work_dir + ' -f ' + \
self.work_dir + '/' + tar_fname + \
' ' + zip_fname)
os.system('gzip -d ' + self.work_dir + '/' + zip_fname)
os.system('mv ' + self.work_dir + '/' + \
zip_fname.rstrip('.gz') + ' ' + \
self.work_dir + '/' + testdata_fname)
#os.system('rm ' + self.work_dir + '/' + tar_fname)
elif (os.path.isfile(self.work_dir + '/' + download_fname)):
os.system('gzip -d ' + self.work_dir + '/' + download_fname)
os.system('tar -xC ' + self.work_dir + ' -f ' + \
self.work_dir + '/' + tar_fname + \
' ' + zip_fname)
os.system('gzip -d ' + self.work_dir + '/' + zip_fname)
os.system('mv ' + self.work_dir + '/' + \
zip_fname.rstrip('.gz') + ' ' + \
self.work_dir + '/' + testdata_fname)
#os.system('rm ' + self.work_dir + '/' + tar_fname)
else:
print('Downloading ' + download_fname + ' ...')
if (not self.download(url, download_fname)):
print('Downloading failed for ' + download_fname)
return False
os.system('gzip -d ' + self.work_dir + '/' + download_fname)
os.system('tar -xC ' + self.work_dir + ' -f ' + \
self.work_dir + '/' + tar_fname + \
' ' + zip_fname)
os.system('tar -xC ' + self.work_dir + ' -f ' + \
self.work_dir + '/' + tar_fname + \
' ' + zip_fname)
os.system('gzip -d ' + self.work_dir + '/' + zip_fname)
os.system('mv ' + self.work_dir + '/' + \
zip_fname.rstrip('.gz') + ' ' + \
self.work_dir + '/' + testdata_fname)
os.system('rm ' + self.work_dir + '/' + tar_fname)
return True
def download(self, url, fname):
try:
response = urllib2.urlopen(url + '/' + fname)
except urllib2.HTTPError, err:
return False
except urllib2.URLError, err:
return False
out_fh = open(self.work_dir + '/' + fname, 'w')
out_fh.write(response.read())
out_fh.close()
return True
def exec_Mergedb(self, input1, input2):
# Create benchmark files:
cmd_mergedb = 'python' + ' ' + 'Mergedb' + ' -I1=' + \
input1 + ' -I2=' + input2 + \
' -G ' + self.taxon_id + ' >> ' + \
self.output_filename
fh = open(self.output_filename, 'a')
fh.write('\n' + cmd_mergedb + '\n')
fh.close()
subprocess.call(cmd_mergedb, shell=True)
return None
def run_test(self, testDataset_fh):
for line in testDataset_fh:
input1 = self.work_dir + '/' + line.strip()
input2 = self.work_dir + '/' + (next(testDataset_fh)).strip()
self.exec_Mergedb(input1, input2)
break
return None
def process_test(self):
self.download_testDataset(open(self.testData_filename, 'r'))
# raise SystemExit
subprocess.call('echo ' + 'This is the output from running ' + \
'testMergedb program' + ' ' + \
'>' + self.output_filename, shell=True)
self.run_test(open(self.testData_filename, 'r'))
subprocess.call('echo ' + 'End of running testMergedb ' + \
'program' + ' ' + '>>' + \
self.output_filename, shell=True)
return None
if __name__ == '__main__':
if len(sys.argv) == 1:
print (sys.argv[0] + ':')
print(__doc__)
sys.exit(0)
else:
testObject = testMergedb()
testObject.process_test()
sys.exit(0)