用Maixcam制作的HTTP推流

1. 整体架构摄像头采集camera.Camera持续读取图像帧。工作线程Server.stream_worker负责不断从摄像头获取最新图像并压缩为JPEG通过事件通知主线程或HTTP请求。Flask HTTP服务提供/stream路由返回一个multipart/x-mixed-replace响应将JPEG帧逐个发送。线程同步使用threading.Eventimage_event协调工作线程和HTTP生成器之间的数据传递。2. 关键组件说明2.1Server类__init__初始化事件、照片保存路径并绑定Flask路由。stream_worker工作线程函数循环执行。get_stream生成器函数用于构建HTTP流。http_streamFlask路由处理器返回一个Response对象将get_stream生成器作为响应体并设置MIME类型为multipart/x-mixed-replace; boundaryframe。start_server启动工作线程和Flask开发服务器通过werkzeug的make_server在独立线程中运行。stop_server停止服务。2.2 Flask路由客户端访问http://设备IP:8080/stream即可开始接收JPEG图像流。3. 推流工作流程3.1 服务器启动阶段server Server()创建实例。server.start_server()执行启动stream_worker线程守护线程。启动Flask服务器线程守护线程。3.2 视频流处理阶段工作线程持续运行读取摄像头帧 → 编码JPEG → 设置image_event。无论是否有HTTP客户端该线程始终在运行保证最新图像随时可用。HTTP请求处理当有客户端请求/stream时Flask调用http_stream它返回一个Response其迭代器是get_stream生成器。生成器在每次迭代中等待image_event一旦被设置立即生成一个完整的multipart帧并yield。浏览器解析该流每个帧作为一张图片显示实现实时视频效果。3.3 同步机制image_event是threading.Event对象用于两个线程间的信号传递生产者工作线程每次捕获并编码完一帧后调用set()。消费者HTTP生成器调用wait()阻塞直到生产者set()后继续然后调用clear()重置准备等待下一帧。这种方式确保了生成器每次返回的都是最新的一帧图像且不会重复或丢失帧当没有客户端时工作线程依然运行但事件可能多次设置生成器只在有请求时才消费。注代码中有部分代码为单目测距的代码以及串口打印的代码可以自行裁剪。github链接https://github.com/T-T-sys/Distance.git# 导入必要的库 import socket,struct from typing import Text import threading as th import os from maix import camera, display, app, image, network,nn,time,touchscreen from flask import Flask, Response, jsonify, send_from_directory from flask_cors import CORS from werkzeug.serving import make_server from maix.v1.machine import UART # 初始化YOLOv5目标检测模型指定模型路径和启用双缓冲以提高性能 detector nn.YOLOv5(model/root/models/yolov5s.mud, dual_buffTrue) #串口 seUART(/dev/ttyS0, 115200) # 初始化摄像头、显示屏和触摸屏 disp display.Display() # 初始化摄像头设置分辨率为 640x480格式为 YVU420SP一种 YUV 格式 cam camera.Camera(640, 480, image.Format.FMT_YVU420SP) # 为摄像头添加一个通道将其输出调整为显示屏的宽度和高度用于显示 cam2 cam.add_channel(disp.width(), disp.height()) ts touchscreen.TouchScreen() # 创建Flask应用实例并启用CORS允许跨域请求 Flask_app Flask(__name__) CORS(Flask_app) #创建退出通道 img_exit image.load(./assets/exit.jpg).resize(50, 50) img_exit_touch image.load(./assets/exit_touch.jpg).resize(50, 50) # 记录上一次眼睛图标状态改变的时间毫秒用于触摸消抖 img_eye_last_change time.ticks_ms() # 定义触摸检测函数判断触摸点是否在指定矩形区域内 # t: 触摸信息t[0] 为 x, t[1] 为 y, t[2] 为触摸状态 # box: 矩形区域 [左上角 x, 左上角 y, 宽度, 高度] # oft: 额外偏移量用于调整检测区域 def touch_box(t, box, oft 0): if t[2] and t[0] oft box[0] and t[0] box[0] box[2] oft and t[1] oft box[1] and t[1] box[1] box[3] oft: return True else: return False # 标志变量是否需要退出程序 need_exit False #测距函数 def measure_distance(obj_size, err_distance, max_distance): #obj_size对象在图像中所占的像素位置 #err_distance摄像头倾斜导致的盲区距离 #max_distance摄像头能拍摄的最大范围不包括盲区距离 #actual_distance最后测得的距离 #obj_height摄像头的最大像素高度 obj_height detector.input_height() #计算距离 actual_distance max_distance - obj_size * (max_distance / obj_height) err_distance end_distance round(actual_distance,2) return end_distance #串口传输 def send_data(temp): data str(distance) \r \n se.write(data.encode(utf-8)) print(data) class Server(): def __init__(self): # 初始化事件和计数器用于控制图片捕获和存储 self.image_event th.Event() self.image_counter 1 self.save_path /root/photos # 确保照片保存目录存在 if not os.path.exists(self.save_path): os.makedirs(self.save_path) # 为Flask应用添加路由 Flask_app.route(/stream)(self.http_stream) Flask_app.route(/take_photo)(self.take_photo) Flask_app.route(/photos/path:filename)(self.send_photo) def stream_worker(self): # 摄像头视频流的工作线程 while self.running: self.img cam.read() self.jpg self.img.to_jpeg(90) if self.jpg is None: print(编码错误) os._exit(-1) self.image_event.set() def get_stream(self): # 生成视频流的生成器函数 while self.running: self.image_event.wait() self.image_event.clear() if self.jpg is None: continue yield ( bContent-Type: data/jpeg\r\n\r\n self.jpg.to_bytes(False) b\r\n\r\n--frame\r\n ) def http_stream(self): # 返回视频流的HTTP响应 return Response( self.get_stream(), mimetypemultipart/x-mixed-replace; boundaryframe ) def take_photo(self): # 拍摄照片并返回照片信息 if not hasattr(self, img): return jsonify({error: No image captured}), 400 filename f{self.save_path}/{self.image_counter}.jpg self.img.save(filename) self.image_counter 1 return jsonify({message: f照片已保存为 {filename}., image_url: f/photos/{self.image_counter - 1}.jpg}) def send_photo(self, filename): # 返回指定照片 return send_from_directory(self.save_path, filename) def start_server(self): # 启动服务器和视频流工作线程 self.running True self.stream_thread th.Thread(targetself.stream_worker, daemonTrue) self.stream_thread.start() self.server make_server(0.0.0.0, 8080, Flask_app, threadedTrue, processes1) self.server.log_startup() self.server_thread th.Thread(targetself.server.serve_forever, daemonTrue) self.server_thread.start() def stop_server(self): # 停止服务器和工作线程 self.running False self.server.shutdown() self.server_thread.join() self.stream_thread.join() # 初始化WiFi wifi network.wifi.Wifi() # 创建服务器实例并启动服务器 server Server() server.start_server() urlsserver.get_stream() disp_switch False while not app.need_exit(): # 从摄像头读取一帧图像 show_img cam2.read() # 读取触摸屏当前触摸状态 t ts.read() # 定义退出按钮的位置和大小在屏幕左上角距边缘 20 像素 box [20, 15, img_exit.width(), img_exit.height()] # 检测触摸是否落在退出按钮区域内 if touch_box(t, box, 0): # 如果触摸到退出按钮绘制按下状态的图片 show_img.draw_image(box[0], box[1], img_exit_touch) need_exit True else: # 否则绘制普通状态的图片 show_img.draw_image(box[0], box[1], img_exit) # 使用检测模型对图像进行目标检测设置置信度阈值和IOU阈值 objs detector.detect(show_img, conf_th0.45, iou_th0.45) # 遍历检测到的所有对象 for obj in objs: # 在图像上绘制矩形框表示检测到的对象 show_img.draw_rect(obj.x, obj.y, obj.w, obj.h, colorimage.COLOR_RED) #测量物体的距离 position obj.y obj.h distance measure_distance(position , 0.3, 5) send_data(distance) # 构造包含类别和置信度的字符串并在图像上绘制 msg f{detector.labels[obj.class_id]}: {distance:.2f} show_img.draw_string(obj.x, obj.y, msg, colorimage.COLOR_RED) disp.show(show_img) # 如果检测到退出按钮被按下跳出循环结束程序 if need_exit: break