-
Notifications
You must be signed in to change notification settings - Fork 1
/
app.py
executable file
·361 lines (303 loc) · 15.3 KB
/
app.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
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
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
# -*- coding: utf-8 -*-
import os
import yaml
from flask import (Flask, flash, request, redirect, url_for, render_template, send_from_directory, session)
from werkzeug.utils import secure_filename
import do_process
import marge_bart
from utils import model_logger as logger
PROJECT_DIR = os.path.dirname(__file__)
# related to flask app
app = Flask(__name__)
app.secret_key = os.urandom(24)
@app.route('/', methods=['GET', 'POST'])
def index():
if request.method == 'POST':
# submit job button
if 'submit_button' in request.form:
# fetch user name and generate unique project path
username = request.form['username']
jobname = request.form['jobname']
user_key = do_process.generate_user_key(username, jobname)
# docker user path
user_path = do_process.init_project_path(user_key)
# record user data
user_data = {}
user_data['user_email'] = username
user_data['user_job'] = jobname
user_data['user_key'] = user_key
user_data['user_path'] = user_path
user_data['dataType'] = request.form['dataType']
user_data['assembly'] = request.form['species']
user_data['files'] = []
# process input data
if user_data['dataType'] == "ChIP-seq":
allowed_extensions = set(['bam', 'bed'])
# execute bart only
user_data['marge'] = False
user_data['bart'] = True
# process input data
if user_data['dataType'] == "Geneset":
allowed_extensions = set(['txt'])
# execute marge+bart
user_data['marge'] = True
user_data['bart'] = True
# if using genelist, save to file
if request.form.get('uploadList', None):
gene_list = request.form['uploadList']
gene_list_file = 'Geneset.txt'
gene_list_file_path = os.path.join(user_path, 'upload/' + gene_list_file)
with open(gene_list_file_path, 'w') as fopen:
for gene in gene_list:
fopen.write(gene)
user_data['files'].append(gene_list_file)
if request.form['dataType'] == 'ChIP-seq' or \
(request.form['dataType'] == 'Geneset' and request.form['geneType'] == 'geneFile'):
# process what user has uploaded
if 'uploadFiles' not in request.files:
flash('Please choose a file')
return redirect(request.url)
files = request.files.getlist('uploadFiles')
# if user does not select file, browser also
# submit an empty part without filename
for file in files:
if file.filename == '':
flash('One of the files does not have a legal file name.')
return redirect(request.url)
# make sure the suffix of filename in [.txt, .bam, .bed]
if file and allowed_file(file.filename, allowed_extensions):
filename = secure_filename(file.filename)
upload_path = os.path.join(user_path, 'upload')
ext = filename.split('.')[-1]
new_filename = ""
if request.form['dataType'] == 'ChIP-seq':
new_filename = "ChIP-seq."+ext
filename_abs_path = os.path.join(upload_path, new_filename)
else:
new_filename = "Geneset."+ext
filename_abs_path = os.path.join(upload_path, new_filename)
file.save(filename_abs_path)
user_data['files'].append(new_filename) # only save file name, since the uploaded path is always the same
do_process.init_user_config(user_path, user_data)
marge_bart.do_marge_bart(user_data)
# post key
return redirect(url_for('show_key', key=user_key))
# get result button
if 'result_button' in request.form:
user_key = request.form['result_button']
# when key is null, refresh the website
if user_key == "":
return render_template('index.html')
logger.info("Retrieve result: for " + user_key)
if do_process.is_user_key_exists(user_key):
logger.info("Retrieve result: user exists.")
return redirect(url_for('get_result', user_key=user_key))
else:
logger.error("Retrieve result: did not find the result.")
err_msg = "Job does not exist, make sure you enter the right key."
return redirect(url_for('error_page', msg=err_msg))
# navbar result button
if 'navbar_button' in request.form:
logger.info("Retrieve result...")
user_key = request.form['navbar_button']
if do_process.is_user_key_exists(user_key):
logger.info("Retrieve result: user exists.")
return redirect(url_for('get_result', user_key=user_key))
else:
logger.error("Retrieve result: did not find the result.")
err_msg = "Job does not exist, make sure you enter the right key."
return redirect(url_for('error_page', msg=err_msg))
return render_template('index.html')
@app.route('/contact', methods=['GET', 'POST'])
def contact():
if request.method == 'POST':
# navbar result button
if 'navbar_button' in request.form:
logger.info("Retrieve result...")
user_key = request.form['navbar_button']
if do_process.is_user_key_exists(user_key):
logger.info("Retrieve result: user exists.")
return redirect(url_for('get_result', user_key=user_key))
else:
logger.error("Retrieve result: did not find the result.")
err_msg = "Job does not exist, make sure you enter the right key."
return redirect(url_for('error_page', msg=err_msg))
return render_template('contact.html')
@app.route('/help', methods=['GET', 'POST'])
def help():
if request.method == 'POST':
# navbar result button
if 'navbar_button' in request.form:
logger.info("Retrieve result...")
user_key = request.form['navbar_button']
if do_process.is_user_key_exists(user_key):
logger.info("Retrieve result: user exists.")
return redirect(url_for('get_result', user_key=user_key))
else:
logger.error("Retrieve result: did not find the result.")
err_msg = "Job does not exist, make sure you enter the right key."
return redirect(url_for('error_page', msg=err_msg))
return render_template('help.html')
@app.route('/result', methods=['GET', 'POST'])
def get_result():
if request.method == 'POST':
# navbar result button
if 'navbar_button' in request.form:
logger.info("Retrieve result...")
user_key = request.form['navbar_button']
if do_process.is_user_key_exists(user_key):
logger.info("Retrieve result: user exists.")
return redirect(url_for('get_result', user_key=user_key))
else:
logger.error("Retrieve result: did not find the result.")
err_msg = "Job does not exist, make sure you enter the right key."
return redirect(url_for('error_page', msg=err_msg))
else:
user_key = request.args['user_key']
user_data = do_process.get_user_data(user_key)
logger.info('Get result: for ' + user_key)
logger.info(user_data)
results = do_process.generate_results(user_data)
results['sample'] = False
return render_template('result_demonstration.html', results=results)
@app.route('/key', methods=['GET', 'POST'])
def show_key():
user_key = request.args['key']
if request.method == 'POST':
if 'result_button' in request.form:
user_key = request.form['result_button']
# when key is null, refresh the website
if user_key == "":
return render_template('key_demonstration.html', key=request.args['key'])
logger.info("Retrieve result: for " + user_key)
if do_process.is_user_key_exists(user_key):
logger.info("Retrieve result: user exists.")
return redirect(url_for('get_result', user_key=user_key))
else:
logger.error("Retrieve result: did not find the result.")
err_msg = "Job does not exist, make sure you enter the right key."
return render_template('key_demonstration.html', key=request.args['key'])
# navbar result button
if 'navbar_button' in request.form:
logger.info("Retrieve result...")
user_key = request.form['navbar_button']
if do_process.is_user_key_exists(user_key):
logger.info("Retrieve result: user exists.")
return redirect(url_for('get_result', user_key=user_key))
else:
logger.error("Retrieve result: did not find the result.")
err_msg = "Job does not exist, make sure you enter the right key."
return redirect(url_for('error_page', msg=err_msg))
return render_template('key_demonstration.html', key=user_key)
@app.route('/error', methods=['GET', 'POST'])
def error_page():
err_msg = request.args['msg']
return render_template('error.html', msg=err_msg)
@app.route('/plot/<userkey_tfname>')
def bart_plot_result(userkey_tfname):
user_key, tf_name = userkey_tfname.split('___')
# =========test d3.js below=============
# use user_key to retrieve plot related results
user_path = os.path.join(PROJECT_DIR, 'usercase/' + user_key)
bart_output_dir = os.path.join(user_path, 'download/bart_output')
plot_results = do_process.generate_plot_results(bart_output_dir, tf_name)
return render_template('plot_result.html', plotResults=plot_results)
#============test end========================
# where plots locate
# plot_path = os.path.join(PROJECT_DIR, 'usercase/' + user_key + '/download/bart_output/plot')
# distribution_plot = '/download/bart_output/plot/' + user_key + '___' + tf_name + '_ranked_dot.png'
# auc_plot = '/download/bart_output/plot/' + user_key + '___' + tf_name + '_cumulative_distribution.png'
# plot_results = {}
# plot_results['user_key'] = user_key
# plot_results['tf_name'] = tf_name
# plot_results['auc_plot'] = auc_plot
# plot_results['dist_plot'] = distribution_plot
# return render_template('plot_result.html', plotResults=plot_results)
@app.route('/log/<userkey_filename>')
def download_log_file(userkey_filename):
user_key, filename = userkey_filename.split('___')
user_path = os.path.join(PROJECT_DIR, 'usercase/' + user_key)
log_path = os.path.join(user_path, 'log')
return send_from_directory(log_path, filename)
@app.route('/download/<userkey_filename>')
def download_marge_file(userkey_filename):
user_key, filename = userkey_filename.split('___')
user_path = os.path.join(PROJECT_DIR, 'usercase/' + user_key)
download_path = os.path.join(user_path, 'download')
return send_from_directory(download_path, filename)
@app.route('/download/bart_output/<userkey_filename>')
def download_bart_res_file(userkey_filename):
user_key, filename = userkey_filename.split('___')
user_path = os.path.join(PROJECT_DIR, 'usercase/' + user_key)
download_path = os.path.join(user_path, 'download/bart_output')
return send_from_directory(download_path, filename)
# @app.route('/download/bart_output/plot/<userkey_filename>')
# def download_bart_chart_file(userkey_filename):
# user_key, filename = userkey_filename.split('___')
# user_path = os.path.join(PROJECT_DIR, 'usercase/' + user_key)
# download_path = os.path.join(user_path, 'download/bart_output/plot')
# return send_from_directory(download_path, filename)
# ===== for genelist/ChIPdata sample =====
# download sample data
@app.route('/sample/<sample_type>')
def download_sample_file(sample_type):
sample_path = ""
sample_name = ""
if sample_type == 'genelist':
sample_name = "genelist.txt"
sample_path = os.path.join(PROJECT_DIR, 'sample/genelist/upload')
elif sample_type == 'ChIPdata':
sample_name = "ChIPseq.txt"
sample_path = os.path.join(PROJECT_DIR, 'sample/ChIPdata/upload')
return send_from_directory(sample_path, sample_name)
# sample result
@app.route('/sample_result/<sample_type>')
def sample_result(sample_type):
config_file = os.path.join(PROJECT_DIR, 'sample/' + sample_type + '/user.config')
if not os.path.exists(config_file):
return None
user_data = {}
with open(config_file, 'r') as fopen:
user_data = yaml.load(fopen)
if user_data:
results = do_process.generate_results(user_data)
if 'marge_result_files' in results:
marge_result_list = []
for marge_res_file in results['marge_result_files']:
filename, file_url = marge_res_file
marge_result_list.append((filename, file_url.replace('download', 'sample_download')))
results['marge_result_files'] = marge_result_list
if 'bart_result_files' in results:
bart_result_list = []
for bart_res_file in results['bart_result_files']:
filename, file_url = bart_res_file
bart_result_list.append((filename, file_url.replace('download', 'sample_download')))
results['bart_result_files'] = bart_result_list
results['sample'] = True
results['sample_type'] = sample_type
return render_template('result_demonstration.html', results=results)
# show sample plot result
@app.route('/sample_plot/<sample_type>/<tf_name>')
def bart_sample_plot_result(sample_type, tf_name):
user_path = os.path.join(PROJECT_DIR, 'sample/' + sample_type)
bart_output_dir = os.path.join(user_path, 'download/bart_output')
plot_results = do_process.generate_plot_results(bart_output_dir, tf_name)
return render_template('plot_result.html', plotResults=plot_results)
# download sample result files
@app.route('/sample_download/<userkey_filename>')
def download_sample_marge_file(userkey_filename):
user_key, filename = userkey_filename.split('___')
user_path = os.path.join(PROJECT_DIR, 'sample/' + user_key)
download_path = os.path.join(user_path, 'download')
return send_from_directory(download_path, filename)
@app.route('/sample_download/bart_output/<userkey_filename>')
def download_sample_bart_res_file(userkey_filename):
user_key, filename = userkey_filename.split('___')
user_path = os.path.join(PROJECT_DIR, 'sample/' + user_key)
download_path = os.path.join(user_path, 'download/bart_output')
return send_from_directory(download_path, filename)
def allowed_file(filename, allowed_extensions):
return '.' in filename and \
filename.rsplit('.', 1)[1].lower() in allowed_extensions
# if __name__ == '__main__':
# app.run(debug=True,host='0.0.0.0')