Я разработал модель глубокого обучения с использованием керасов и успешно развернул ее во флэш-приложении.In содержит конечную точку / Предикат, которую другое приложение может запросить с данными, а затем получить прогноз, возвращенный ему.Это колбовое приложение просто запускается как консольное приложение, но в дополнение к этому я также хотел открыть веб-страницу, которая показывает график разброса входных данных в реальном времени.Я смог сделать это, используя библиотеку bokeh и разместив это в своем приложении для колб.Теперь, когда приложение запускается, веб-страница графика открывается, и он настроен на периодическое попадание на новую конечную точку «/ data» в приложении «колба», чтобы получать последние данные и обновлять 2-мерный точечный график.Кажется, это работает надежно.
Теперь по причине моего поста.Я действительно хотел бы улучшить это, изменив свой двухмерный график рассеяния на трехмерный график рассеяния.К сожалению, боке не предлагает трехмерный график рассеяния из коробки.Это привело меня к сюжету и тире.Я знаю, что заговор может сделать 3D-график рассеяния.Я ищу пример кода или совет о том, как мне поступить.
Должен ли я сохранить существующую структуру приложения фляги, но каким-то образом объединить это с заговором, чтобы заменить боке?Это означает, что в заговоре нужно будет создать веб-страницу и периодически указывать на точку / data моего приложения для фляги.
Должен ли я отказаться от фляги и перейти в приложение для тире?Если так, то мне нужно будет развернуть модель keras в тире.и настройте обе точки: / Предсказание и / Данные.
Ниже мой существующий код с колбой и боке.Есть ли у кого-нибудь какие-либо предложения о том, как мне действовать, и даже лучше, кто-нибудь может показать мне пример кода, который будет применим к тому, что я пытаюсь сделать.
import numpy as np
import tensorflow as tf
import keras
from keras.models import model_from_json
from flask import Flask, jsonify, make_response, request
from bokeh.plotting import figure, show
from bokeh.models import AjaxDataSource, CustomJS, Range1d
# Bokeh related code
adapter = CustomJS(code="""
const result = {x: [], y: []}
const pts = cb_data.response.points
for (i=0; i<pts.length; i++) {
result.x.push(pts[i][0])
result.y.push(pts[i][1])
}
return result
""")
source = AjaxDataSource(data_url='http://10.61.226.215:443/data',
polling_interval=200, adapter=adapter)
p = figure(plot_height=700, plot_width=1400, x_axis_label='Frequency', y_axis_label='Phase', background_fill_color="lightgrey",
title="Scatter Plot of TOI")
p.x_range = Range1d(0, 49)
p.y_range = Range1d(-2**15, 2**15)
p.circle('x', 'y', source=source, color='red', size=10)
# Flask related code
app = Flask(__name__)
def get_model():
global model
global g
json_path = 'RNN_LSTM_128_Drop0p2.json'
h5_path = "RNN-010-0.983732-0.997119.h5"
g = tf.Graph()
with g.as_default():
# Pull in the model we want to test
json_file = open(json_path, 'r')
loaded_model_json = json_file.read()
json_file.close()
model = model_from_json(loaded_model_json)
# load weights into new model
model.load_weights(h5_path)
print("Loaded model from disk")
# Compile the loaded model
model.compile(loss=keras.losses.categorical_crossentropy,
optimizer=keras.optimizers.Adadelta(),
metrics=['accuracy'])
print("Compiled Model")
print(" * Recurrent Neural Network Trained Model Loaded Successfully")
def RNN_scale_input(X):
N_samp = X.shape[0]
N_t_samp = X.shape[1]
n_feat = X.shape[2]
X_scale = np.zeros(shape=(N_samp, N_t_samp, n_feat), dtype=np.float32)
for h in range(N_samp):
# Scale the channel index to be between -1 and 1
X_scale[h][:, 0] = (X[h][:, 0] - 25) / 25
# Scale the phase value to be between -1 and 1
X_scale[h][:, 1] = X[h][:, 1] / 2**15
return X_scale
# Define function to pre-process phase data
def pre_process_data(data, num_rows, num_cols):
# Strip the opening and closing brackets from the input data string
data = data.strip('[]')
# Convert string into a list of floating point numbers
try:
data = [float(h) for h in data.split(',')]
except:
print("***ERROR***")
print(data)
data_np = np.asarray(data).reshape(1, num_rows, num_cols)
rnn_structure = RNN_scale_input(data_np)
return rnn_structure[0]
def pre_process_plot_data(data, num_rows, num_cols):
# Strip the opening and closing brackets from the input data string
data = data.strip('[]')
# Convert string into a list of floating point numbers
try:
list_data = [int(k) for k in data.split(',')]
except:
print("***ERROR***")
print(data)
plot_data_np = np.asarray(list_data).reshape(num_rows, num_cols)
return plot_data_np
N_t, n_features = 64, 2
print(" * Loading Neural Network Trained Model...")
get_model()
# Define Tag of Interest to plot
TOI = "307401320416C2054B6E99D7"
def crossdomain(f):
def wrapped_function(*args, **kwargs):
resp = make_response(f(*args, **kwargs))
h = resp.headers
h['Access-Control-Allow-Origin'] = '*'
h['Access-Control-Allow-Methods'] = "GET, OPTIONS, POST"
h['Access-Control-Max-Age'] = str(21600)
requested_headers = request.headers.get('Access-Control-Request-Headers')
if requested_headers:
h['Access-Control-Allow-Headers'] = requested_headers
return resp
return wrapped_function
x = [0]*N_t
y = [0]*N_t
@app.route("/predict", methods=['POST'])
def predict():
global x
global y
message = request.get_json(force=True)
TagId = message['TagId']
PhaseData = message['PhaseInput']
model_input = np.zeros(shape=(1, N_t, n_features), dtype='float32')
model_input[0] = pre_process_data(PhaseData, N_t, n_features)
# If TagId == TOI, plot the data
if TagId == TOI:
# Update data in plot
plot_data = pre_process_plot_data(PhaseData, N_t, n_features)
x1 = list(plot_data[:, 0])
x = [int(x1[i]) for i in range(N_t)]
y1 = list(plot_data[:, 1])
y = [int(y1[i]) for i in range(N_t)]
# Make prediction with model
with g.as_default():
model_output = model.predict(model_input)
# Take the second element (location at index (0, 1)) as the float prediction
pred_val = model_output[0, 1]
prediction = pred_val.tolist()
# Construct response to send back to client app
response = {
'tag': TagId,
'prediction': prediction
}
return jsonify(response)
@app.route('/data', methods=['GET', 'OPTIONS', 'POST'])
@crossdomain
def data():
global x
global y
return jsonify(points=list(zip(x, y)))
# show and run
show(p)
# app.run(port=443)