-
Notifications
You must be signed in to change notification settings - Fork 2
/
app.py
145 lines (112 loc) · 5.58 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
from multiprocessing.connection import wait
from time import sleep
from flask import Flask, request, redirect, url_for, send_file, send_from_directory, abort
from flask.helpers import make_response
import execute as ex
import os, tempfile
import sys
import lib
app = Flask(__name__)
#Set up logfile
os.system('touch pylog.txt')
# Start API
@app.route('/', methods=["POST, GET"])
def default():
return redirect(url_for('status')) #redirect empty url to status check
@app.route('/analyze', methods=['POST'])
def analyze():
print('Running analysis!', file=open('pylog.txt', 'a'))
with tempfile.TemporaryDirectory() as tempDir:
output = ex.exec(request, 'analysis', tempDir)
if(output == -1):
return make_response('An error occured during analysis', 202)
return send_file(output, as_attachment=True, attachment_filename='sim_output.zip')
@app.route('/convert', methods=['POST'])
def convert():
print('Running conversion!', file=open('pylog.txt', 'a'))
with tempfile.TemporaryDirectory() as tempDir:
output = ex.exec(request, 'conversion', tempDir)
if(output == -1):
return make_response('An error occured during conversion', 202)
return send_file(output, as_attachment=True, attachment_filename='conv_output.zip')
@app.route('/convert_and_simulate', methods=['POST'])
def conv_and_sim():
print('Running conversion and analysis!', file=open('pylog.txt', 'a'))
with tempfile.TemporaryDirectory() as tempDir:
c_output = ex.exec(request, 'both', tempDir)
if c_output == '':
return make_response('An error occured during conversion', 202)
# if conversion returns a zip file, send the archive straight to analysis
print("Run analysis...", file=open('pylog.txt', 'a'))
if c_output.endswith('.zip'):
c_omex = os.path.join(tempDir,'c.omex')
os.system('mv ' + c_output + ' ' + c_omex)
output = ex.analysis(tempDir, ex.args.getArgs(), c_omex)
return send_file(output, as_attachment=True, attachment_filename='sim_output.zip')
# otherwise, simulate topModule with default/given parameters
# copy topModule file to new working tempDir
with tempfile.TemporaryDirectory() as aTempDir:
os.system('cp ' + c_output + ' ' + aTempDir)
topMod = os.listdir(aTempDir)[0]
print("topMod: " + topMod, file=open('pylog.txt', 'a'))
conv_output = os.path.join(aTempDir, topMod)
print("conversion output: " + conv_output, file=open('pylog.txt', 'a'))
output = ex.analysis(aTempDir, ex.args.getArgs(), conv_output)
# sleep(100)
return send_file(output, as_attachment=True, attachment_filename='sim_output.zip')
@app.route('/status', methods=['GET', 'POST'])
# Status endpoint to communicate that the plug-in is up and running
def status():
return("The iBioSim Server V1.0.1 is up and running")
# Follow SBH plug-in structure
@app.route('/evaluate', methods=['GET', 'POST'])
# Evaluate endpoint to check if file can be handeled by plug-in
# Plug-in works on archive so the accepted type is Collection
def evaluate():
data = request.get_json(force=True)
rdf_type = data['type']
# ~~~~~~~~~~~~~~~~~ REPLACE THIS SECTION WITH OWN RUN CODE ~~~~~~~~~~~~~~
# uses rdf types
accepted_types = {'Collection'}
acceptable = rdf_type in accepted_types
# # to ensure it shows up on all pages
# acceptable = True
# ~~~~~~~~~~~~~~~~~~~~~~~~~ END SECTION ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
if acceptable:
return f'The type sent ({rdf_type}) is an accepted type', 200
else:
return f'The type sent ({rdf_type}) is NOT an accepted type', 415
# Expected: Combine archive file, arguments given in request parameters
@app.route('/run', methods=['POST'])
def run():
#temporary directory to write intermediate files to
temp_dir = tempfile.TemporaryDirectory()
data = request.get_json(force=True)
# Set variables
top_level_url = data['top_level']
complete_sbol = data['complete_sbol']
instance_url = data['instanceUrl']
genbank_url = data['genbank']
size = data['size']
rdf_type = data['type']
shallow_sbol = data['shallow_sbol']
url = complete_sbol.replace('/sbol', '')
try:
# ~~~~~~~~~~~~~ REPLACE THIS SECTION WITH OWN RUN CODE ~~~~~~~~~~~~~~~
archive_url = top_level_url + '/archive'
# calls the function for analysis
image = lib.call(archive_url)
out_name = "iBioSim_Results.png"
file_out_name = os.path.join(temp_dir.name, out_name)
# Write file in temporary directory
with open(file_out_name, 'wb') as out_file:
out_file.write(image)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~ END SECTION ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
return send_from_directory(temp_dir.name, out_name,
as_attachment=True,
attachment_filename=out_name)
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
lnum = exc_tb.tb_lineno
abort(400, f'Exception is: {e}, exc_type: {exc_type}, exc_obj: {exc_obj}, fname: {fname}, line_number: {lnum}, traceback: {traceback.format_exc()}')