forked from evallen/querycap
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathserver.py
More file actions
98 lines (74 loc) · 2.56 KB
/
server.py
File metadata and controls
98 lines (74 loc) · 2.56 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
import os
import pathlib
from flask import Flask, request
from werkzeug.utils import secure_filename
from datetime import datetime
from gtts import gTTS
import requests
import subprocess
IMAGE_FOLDER = "./images"
SOUND_FOLDER = "./sounds"
app = Flask(__name__)
app.config['IMAGE_FOLDER'] = IMAGE_FOLDER
app.config['SOUND_FOLDER'] = SOUND_FOLDER
pathlib.Path(IMAGE_FOLDER).mkdir(parents=True, exist_ok=True)
pathlib.Path(SOUND_FOLDER).mkdir(parents=True, exist_ok=True)
@app.after_request
def after(response):
print(response.status)
print(response.headers)
print(response.get_data())
return response
def check_for_file(name):
if name not in request.files:
print(f"didn't find file {name}")
return {
"success": False,
"message": f"'{name}' file not found in request.",
}, 400
def check_for_form(name):
if name not in request.form:
print(f"didn't find form field {name}")
return {
"success": False,
"message": f"'{name}' field not found in form.",
}, 400
def generate_sound(answer):
print("Saving audio...")
tts = gTTS(answer)
tts.save("answer.mp3")
subprocess.call(["afplay", "answer.mp3"])
def send_image_sound(img_path, sound_path):
'''Send image and sound file to AWS server through SSH tunnel.'''
# ssh -o "ServerAliveInterval 60" -L 5001:ec2-18-190-153-119.us-east-2.compute.amazonaws.com:5000
files = {
"image": open(img_path, "rb"),
"sound": open(sound_path, "rb"),
}
URL = 'http://127.0.0.1:5001/send_image'
response = requests.post(URL, files=files)
response_json = response.json()
SUCCESS = 200
if response.status_code == SUCCESS:
print("Data sent to model.")
generate_sound(response_json['answer'])
else:
print(f"Error sending POST to {URL}")
@app.route("/send_image", methods=['POST'])
def send_image():
check_for_file('image')
check_for_file('sound')
image = request.files['image']
sound = request.files['sound']
now = datetime.now()
now_str = now.strftime("%Y-%m-%d_%H-%M-%S")
image_filename = now_str + secure_filename(image.filename)
image_filepath = os.path.join(app.config['IMAGE_FOLDER'], image_filename)
image.save(image_filepath)
sound_filename = now_str + secure_filename(sound.filename)
sound_filepath = os.path.join(app.config['SOUND_FOLDER'], sound_filename)
sound.save(sound_filepath)
send_image_sound(image_filepath, sound_filepath)
return {
"success": True,
}, 200