-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
373 lines (298 loc) · 13.6 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
362
363
364
365
366
367
368
369
370
371
372
373
from flask import Flask, render_template, request, redirect, url_for, session, jsonify, send_file
from functools import wraps
import numpy as np
import datetime
import os
import io
import torch
import torch.nn.functional as F
from PIL import Image
from transformers import pipeline
from pathlib import Path
import uuid
import pymongo
from scipy.stats import skew, kurtosis
import pandas as pd
import gridfs
from preprocessing import preprocess_gait_data
from io import BytesIO
from bson import ObjectId
from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas
os.environ['TF_ENABLE_ONEDNN_OPTS'] = '0'
os.environ['PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION'] = 'python'
from tensorflow.keras.models import load_model
from tensorflow.keras.optimizers import RMSprop
from sklearn.preprocessing import LabelEncoder, StandardScaler
# Load model directly
from transformers import AutoImageProcessor, AutoModelForImageClassification
scaler = StandardScaler()
app = Flask(__name__)
app.secret_key = os.urandom(24).hex()
print(app.secret_key)
# model = load_model('model.h5')
# pipe = pipeline("image-classification", "gianlab/swin-tiny-patch4-window7-224-finetuned-parkinson-classification")
# target_img = os.path.join(os.getcwd() , 'static/images')
processor = AutoImageProcessor.from_pretrained("gianlab/swin-tiny-patch4-window7-224-finetuned-parkinson-classification")
model = AutoModelForImageClassification.from_pretrained("gianlab/swin-tiny-patch4-window7-224-finetuned-parkinson-classification")
lstm_model = load_model('lstm_model3.h5', compile=False)
# Connection string
connection_string = "mongodb+srv://sem6:ssn@cluster0.q0hpe8v.mongodb.net/?retryWrites=true&w=majority&appName=Cluster0"
# Connect to MongoDB
client = pymongo.MongoClient(connection_string)
# Access a specific database
db = client['ndd_prediction']
# Access a specific collection within the database
collection = db['Credentials']
storage = db['storage']
fs = gridfs.GridFS(db)
curr_username = ''
for document in collection.find():
print(document)
# Helper function for login required
def login_required(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if 'username' not in session:
return redirect(url_for('login'))
return f(*args, **kwargs)
return decorated_function
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
password = int(password)
curr_username = username
if username and password:
# Query MongoDB for the username and password
user = collection.find_one({"Username": username, "Password": password})
if user:
# If user exists, store the user ID in the session and redirect to the index page
print("success")
session['username'] = username
session['user_id'] = str(user['_id'])
print(f"User ID: {session['user_id']}")
return redirect(url_for('index'))
# If user does not exist or credentials are incorrect, redirect back to the login page
return render_template('login.html')
@app.route('/signup', methods=['GET', 'POST'])
def signup():
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
if username and password:
password = int(password)
# Check if the user already exists
existing_user = collection.find_one({"Username": username})
if existing_user is None:
# Add the new user to the MongoDB collection
new_user = {"Username": username, "Password": password}
result = collection.insert_one(new_user)
user_id = str(result.inserted_id)
print(f"New user ID: {user_id}")
return redirect(url_for('login'))
else:
error_message = "User already exists. Please try logging in."
return render_template('signup.html', error=error_message)
return render_template('signup.html')
@app.route('/index') # Define the route URL path
@login_required
def index():
# You can render the index.html template here
return render_template('index.html')
@app.route('/')
def home():
return render_template('login.html')
@app.route('/main', methods=['POST'])
@login_required
def main():
return render_template('main.html')
# Allow files with IMGension png, jpg, and jpeg
ALLOWED_IMG = {'jpg', 'jpeg', 'png'}
ALLOWED_FILE = {'csv'}
def allowed_img(filename):
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_IMG
def allowed_file(filename):
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_FILE
def calculate_statistics(df):
stats = {}
for column in df.columns:
stats[f'{column}Min'] = df[column].min()
stats[f'{column}Max'] = df[column].max()
stats[f'{column}Std'] = df[column].std()
stats[f'{column}Med'] = df[column].median()
stats[f'{column}Avg'] = df[column].mean()
stats[f'{column}Skewness'] = skew(df[column])
stats[f'{column}Kurtosis'] = kurtosis(df[column])
return stats
@app.route('/predict', methods=['POST'])
@login_required
def predict():
file = request.files['file']
acc = request.files['accelerometer']
if (file and allowed_img(file.filename)) or (acc and allowed_file(acc.filename)): # Checking file format
image_path = Path('temp.png')
file.save(image_path)
# Open and resize the image
image = Image.open(image_path).resize((200, 200))
temp_image_filename = str(uuid.uuid4()) + '.jpg'
temp_image_path = os.path.join('static/temp_images', temp_image_filename)
# Save the resized image
image = image.convert('RGB')
image.save(temp_image_path)
proc = processor(image, return_tensors="pt")
with torch.no_grad():
outputs = model(**proc)
logits = outputs.logits
# Apply softmax to get probabilities
probabilities = F.softmax(logits, dim=-1)
# Get the predicted class (optional)
predicted_class = logits.argmax(-1).item()
# Convert probabilities to a list or other desired format
probabilities_list = probabilities.squeeze().tolist()
# Get predicted class
#predictions = outputs.logits.argmax(-1).item()
# Delete the temporary image file
image_path.unlink()
# Format the predictions
if predicted_class == 1:
predicted_class = "Parkinson"
else:
predicted_class = "Healthy"
""" acc_data = acc.read()
acc_buffer = io.BytesIO(acc_data)
preprocessed_data = preprocess_gait_data(acc_buffer)
df = preprocessed_data
df = df.drop(df.columns[0], axis=1)
test_df = calculate_statistics(df)
test_df = pd.DataFrame.from_dict(test_df, orient='index').T
testt_X = test_df.values
testX_scaled = scaler.fit_transform(testt_X)
X_test_reshaped = testX_scaled.reshape((testX_scaled.shape[0], 1, testX_scaled.shape[1]))
# Predict with the loaded model (example)
categories = ['healthy', 'parkinson', 'huntington', 'als']
label_encoder = LabelEncoder()
label_encoder.fit(categories)
label_names = label_encoder.classes_
# Create a dictionary of label name to their probabilities
# predicted_probabilities = np.array([0.2, 0.5, 0.1, 0.2]) # Example predicted probabilities
predicted_probablities = lstm_model.predict(X_test_reshaped)
lstmpred = np.argmax(predicted_probabilities,axis=1)
label_to_probability = {label: prob for label, prob in zip(label_names, predicted_probabilities)}
"""
acc_data = acc.read()
acc_buffer = io.BytesIO(acc_data)
df = preprocess_gait_data(acc_buffer)
# df = preprocessed_data.drop(preprocessed_data.columns[0], axis=1)
print(df)
test_df = calculate_statistics(df)
test_df = pd.DataFrame.from_dict(test_df, orient='index').T
print(test_df)
testX_scaled = StandardScaler().fit_transform(test_df.values)
if testX_scaled.shape[1] != 91:
raise ValueError(f"Expected 91 features, but got {testX_scaled.shape[1]} features.")
# Reshape the data for LSTM input
X_test_reshaped = testX_scaled.reshape((testX_scaled.shape[0], 1, testX_scaled.shape[1]))
categories = ['healthy', 'parkinson', 'huntington', 'als']
label_encoder = LabelEncoder()
label_encoder.fit(categories)
label_names = label_encoder.classes_
predicted_probabilities = lstm_model.predict(X_test_reshaped )
lstmpred = np.argmax(predicted_probabilities, axis=1)
label_to_probability = {label: prob for label, prob in zip(label_names, predicted_probabilities[0])}
gait_probabilities = {label: prob for label, prob in label_to_probability.items()}
swapped_probabilities = {
'als': gait_probabilities['als'],
'parkinson': gait_probabilities['healthy'],
'huntington': gait_probabilities['huntington'],
'healthy': gait_probabilities['parkinson']
}
max_label = max(label_to_probability, key=label_to_probability.get)
max_probability = label_to_probability[max_label]
value = np.float32(max_probability)
max_probability = float(value)
print(gait_probabilities)
""" max_label = max(label_to_probability, key=label_to_probability.get)
max_probability = label_to_probability[max_label] """
# Read the file content into memory before uploading to GridFS
file_content = file.read()
acc_content = acc.read()
file_id = fs.put(file_content, filename=file.filename)
acc_id = fs.put(acc_content, filename=acc.filename)
# Get the username from the session
username = session['username']
store = {
'Name': username,
'Date': datetime.datetime.now(),
'Drawing': file_id,
'IMU Data': acc_id,
'Img Detected': predicted_class,
'Img probability': max(probabilities_list),
'Gait Detected': max_label,
'Gait probability': max_probability}
storage.insert_one(store)
# Generate PDF report
pdf_path = f'static/reports/report_{uuid.uuid4()}.pdf'
generate_pdf(pdf_path, username, predicted_class, max(probabilities_list), max_label, max_probability, temp_image_path)
return render_template('result.html', disease=predicted_class, prob=max(probabilities_list),
user_image=temp_image_path, img_name=file.filename, acc_name=acc.filename,
gait=swapped_probabilities, pdf_report=pdf_path)
else:
return "Unable to read the file. Please check file IMGension"
@app.route('/logout')
def logout():
return render_template('login.html')
@app.route('/history', methods=['GET'])
@login_required
def history():
username = session.get('username')
# Access the user's collection
user_collection = db['storage']
# Retrieve all records for the user
user_data = list(user_collection.find({"Name": username}))
return render_template('history.html', data=user_data)
@app.route('/download/<file_id>')
@login_required
def download_file(file_id):
try:
# Convert file_id to ObjectId
file_id = ObjectId(file_id)
# Retrieve the file from GridFS using the ObjectId
file_data = fs.get(file_id)
# Debug: Check file data size
file_content = file_data.read()
print(f"File size: {len(file_content)} bytes")
# Create a BytesIO object to send the file data
file_io = BytesIO(file_content)
# Set the file's name and content type
filename = file_data.filename
content_type = file_data.content_type if file_data.content_type else 'application/octet-stream'
# Debug: Log the filename and content type
print(f"Downloading file: {filename}, Content Type: {content_type}")
# Send the file to the client
return send_file(file_io, as_attachment=True, download_name=filename, mimetype=content_type)
except Exception as e:
print(f"Error: {str(e)}")
return jsonify({"error": str(e)}), 404
def generate_pdf(pdf_path, username, predicted_class, img_prob, gait_class, gait_prob, image_path):
c = canvas.Canvas(pdf_path, pagesize=letter)
width, height = letter
c.setFont("Helvetica", 12)
c.drawString(100, height - 50, f"Prediction Report for {username}")
c.drawString(100, height - 70, f"Date: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
c.drawString(100, height - 90, f"Image Detected: {predicted_class}")
c.drawString(100, height - 110, f"Image Probability: {img_prob:.2f}")
c.drawString(100, height - 130, f"Gait Detected: {gait_class}")
c.drawString(100, height - 150, f"Gait Probability: {gait_prob:.2f}")
# Add the image to the PDF
if os.path.exists(image_path):
c.drawImage(image_path, 100, height - 400, width=200, height=200)
c.showPage()
c.save()
@app.route('/download_report/<path:filename>', methods=['GET'])
@login_required
def download_report(filename):
return send_file(filename, as_attachment=True)
if __name__ == '__main__':
app.run(debug=True, use_reloader=True, port=8000)