python - 用于提供任何预测/对象检测的 Flask 应用程序
问题描述
尝试构建一个 Flask 应用程序进行实时预测。使用 SocketIO 访问客户端网络摄像头。该模型是一个暗网 Yolov3 对象检测模型。这是我的代码:
Payload.max_decode_packets = 50
clients = []
current_source = ""
# load classifier
with open(r".\backup\classes.names") as f:
# Getting labels reading every line
# and putting them into the list
labels = [line.strip() for line in f]
network = cv2.dnn.readNetFromDarknet(r".\backup\yolovhafiz.cfg",r".\backup\backup\yolovhafiz_best.weights")
layers_names_all = network.getLayerNames()
layers_names_output = [layers_names_all[i[0] - 1] for i in network.getUnconnectedOutLayers()]
probability_minimum = 0.60
threshold = 0.60
colours = np.array([[0, 255, 0], [0, 0, 255], [255, 0, 0]], np.uint8)
# end load classifier
app = Flask(__name__)
app.logger.addHandler(logging.StreamHandler(stdout))
app.config['SECRET_KEY'] = 'secret!'
app.config['DEBUG'] = True
app.config['cors_allowed_origins'] = "*"
socketio = SocketIO(app)
camera = Camera(Makeup_artist())
@socketio.on('input image')
def test_message(input):
global network, layers_names_all, layers_names_output, probability_minimum, threshold, colours
# if current_source == request.sid:
input = input.split(",")[1]
# camera.enqueue_input(input)
image_data = input
img = base64_to_pil_image(image_data)
open_cv_image = np.array(img)
img = open_cv_image[:, :, ::-1].copy()
# DO IMAGE PROCESSING HERE
try:
frame = img
if(len(frame) < 1):
return False
h, w = frame.shape[:2]
scale_width = 680
h = int(round(scale_width * frame.shape[1] / frame.shape[0]))
dim = (h, scale_width)
frame = cv2.resize(frame, dim, interpolation=cv2.INTER_AREA)
blob = cv2.dnn.blobFromImage(frame, 1 / 255.0, (416, 416),
swapRB=True, crop=False)
network.setInput(blob)
output_from_network = network.forward(layers_names_output)
bounding_boxes = []
confidences = []
class_numbers = []
for result in output_from_network:
for detected_objects in result:
scores = detected_objects[5:]
class_current = np.argmax(scores)
confidence_current = scores[class_current]
if confidence_current > probability_minimum:
box_current = detected_objects[0:4] * \
np.array([w, h, w, h])
x_center, y_center, box_width, box_height = box_current
x_min = int(x_center - (box_width / 2))
y_min = int(y_center - (box_height / 2))
bounding_boxes.append([x_min, y_min,
int(box_width), int(box_height)])
confidences.append(float(confidence_current))
class_numbers.append(class_current)
results = cv2.dnn.NMSBoxes(bounding_boxes, confidences,
probability_minimum, threshold)
if len(results) > 0:
for i in results.flatten():
x_min, y_min = bounding_boxes[i][0], bounding_boxes[i][1]
box_width, box_height = bounding_boxes[i][2], bounding_boxes[i][3]
colour_box_current = colours[class_numbers[i]].tolist()
cv2.rectangle(frame, (x_min, y_min),
(x_min + box_width, y_min + box_height),
colour_box_current, 2)
# Putting text with label and confidence on the original image
# cv2.putText(frame, text_box_current, (x_min, y_min - 5),
# cv2.FONT_HERSHEY_SIMPLEX, 0.5, colour_box_current, 1)
img = frame
# END IMAGE PROCESSING
retval, buffer = cv2.imencode('.jpg', img)
img = base64.b64encode(buffer)
img = img.decode('utf-8')
image_data = "data:image/jpeg;base64," + img
broadcasting(image_data)
except:
print('exepct')
return False
# camera.enqueue_input(base64_to_pil_image(input))
def broadcasting(image_data):
emit('out-image-event', {'image_data': image_data})
@socketio.on('set_source')
def set_source(data):
global current_source
current_source = data['sid']
print(data)
@socketio.on('connect')
def test_connect():
clients.append(request.sid)
send_user_list()
@socketio.on('disconnect')
def disconnect_user():
try:
clients.pop(request.sid)
send_user_list()
except:
return False
@app.route('/')
def index():
"""Video streaming home page."""
return render_template('index.html')
@app.route('/source')
def to_source():
return render_template('source.html')
@app.route('/client')
def to_client():
return render_template('client.html')
def gen():
"""Video streaming generator function."""
app.logger.info("starting to generate frames!")
while True:
frame = camera.get_frame() # pil_image_to_base64(camera.get_frame())
yield (b'--frame\r\n'
b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')
@app.route('/video_feed')
def video_feed():
"""Video streaming route. Put this in the src attribute of an img tag."""
return Response(gen(), mimetype='multipart/x-mixed-replace; boundary=frame')
def send_user_list():
print(json.dumps(clients, indent=4))
emit('list_users', json.dumps(clients, indent=4))
if __name__ == '__main__':
socketio.run(app, host="localhost")
但不幸的是,它只是显示没有预测/对象检测的相机流。为什么应用程序无法检测到任何物体?我将非常感谢您的帮助。
解决方案
推荐阅读
- php - Wordpress 4.9 session_start 中的主题功能
- spring - 如何配置一个可以访问spring security中所有url的管理员角色
- php - PHP 7.2 升级代码 - REGEX 未定义常量的使用
- express - SocketIO 阻塞服务器
- python - 最有效的字符串缓冲
- dji-sdk - 这里有人让 DJI Windows SDK 工作了吗?
- sql - 添加 SQL EXIT HANDLER FOR SQLEXCEPTION 会导致语法错误和意外文本
- javascript - 严重:PWC6117:使用 jQuery 加载页面时找不到文件“null”
- javascript - JavaScript 没有兑现承诺
- c - 如何在文件 IO 中成功使用 Delete 函数——C