Flask接收OpenCV无人机视频流返回405错误问题咨询
问题修复方案
核心错误点
- 405错误直接原因:
/video路由的允许方法列表未添加PUT,无人机端的PUT请求被框架直接拦截 - 逻辑设计错误:将「无人机上传帧」和「网页拉取视频流」两个完全不同的请求逻辑耦合在同一个路由分支,浏览器访问img标签时发送的是GET请求,无法触发帧处理逻辑
- 代码逻辑错误:
request.json在Flask中已经是自动解析完成的Python字典,不需要额外调用json.loads()二次解析,会直接抛出类型错误 - 响应格式错误:返回图片数据时未指定正确的
Content-Type,也没有遵循MJPEG流的标准分段格式,浏览器无法将连续帧识别为可播放的视频流
修正后代码
1. Flask服务端代码
用全局变量缓存无人机最新上传的帧,将上传接口和流推送接口拆分,返回符合MJPEG标准的流式响应:
import base64 import json import time from threading import Lock from flask import Flask, make_response, render_template, request, Response app = Flask(__name__) # 全局缓存最新帧,加锁保证线程安全 latest_frame = None frame_lock = Lock() @app.route('/video/upload', methods=['PUT']) def upload_frame(): global latest_frame # 直接取解析好的json数据,不需要二次loads data = request.json # 解码base64得到jpg二进制数据 img_bytes = base64.b64decode(data['image']) # 加锁更新最新帧 with frame_lock: latest_frame = img_bytes return make_response({"code": 0, "msg": "ok"}) def generate_stream(): """生成MJPEG流的生成器""" global latest_frame while True: # 没有帧的时候等待10ms重试 with frame_lock: frame = latest_frame if frame is None: time.sleep(0.01) continue # 按照MJPEG格式拼接帧数据 yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n') # 控制帧率,避免CPU占用过高 time.sleep(0.03) @app.route('/video/stream') def video_stream(): # 返回MJPEG流,设置正确的Content-Type return Response(generate_stream(), mimetype='multipart/x-mixed-replace; boundary=frame') @app.route('/') def index(): return render_template('index.html') @app.route('/cmd') def cmd(): # 后续控制指令逻辑写在这里 pass if __name__ == "__main__": # 关闭reloader避免多进程导致全局帧缓存异常 app.run(debug=True, use_reloader=False, host='0.0.0.0')
2. 前端index.html代码
将img标签的src指向流接口即可:
<!DOCTYPE html> <html> <head> <title>Drone Live Stream</title> </head> <body> <h1>Drone Live Stream</h1> <div> <img src="{{ url_for('video_stream') }}" width="50%"> </div> </body> </html>
3. 无人机端上传代码
补充请求头指定JSON格式,将上传地址改为新的上传接口:
import base64 import json import requests import cv2 cap = cv2.VideoCapture(1) # 替换为你的Flask服务地址,本地测试填http://127.0.0.1:5000 ip = 'http://127.0.0.1:5000' # 补充请求头,指定JSON格式 headers = {"Content-Type": "application/json"} while True: success, img = cap.read() if not success: break cv2.imshow("Drone Capture", img) _, imdata = cv2.imencode('.JPG', img) # base64编码后转字符串 img_b64 = base64.b64encode(imdata).decode('ascii') jStr = json.dumps({"image": img_b64}) try: # 上传到专门的上传接口 requests.put(url=f"{ip}/video/upload", data=jStr, headers=headers, timeout=1) except Exception as e: print(f"upload error: {e}") if cv2.waitKey(1) == 27: break cap.release() cv2.destroyAllWindows()
注意事项
- 跨设备测试时,无人机端ip填笔记本的局域网IP即可,注意提前关闭笔记本的防火墙规则
- 代码中
time.sleep(0.03)对应约30帧的上限,可根据网络情况调整,网络差可以调大间隔降低帧率 - 后续添加视频处理逻辑时,直接在
upload_frame接口里对解码后的img_bytes做处理即可,处理完再更新到latest_frame缓存,网页端会自动拉取处理后的帧 - 如果需要更高的实时性,可以后续把base64编码改成直接传二进制jpg数据,减少编解码开销
内容的提问来源于stack exchange,提问作者Aryan
相关产品推荐
相关产品推荐

