Dash Python中如何在@app.callback回调函数内获取视频流单帧

OpenCV 的 cv2.VideoCapture 硬件操作实例不支持多线程并发访问，你当前的代码中，视频流生成器和dcc.Interval触发的回调函数两个执行路径会同时调用camera.read()抢占摄像头资源，间隔小于200ms时抢资源的频率升高，长时间运行会积累句柄冲突、读帧异常，最终导致崩溃。另外Dash默认debug模式会启动重载进程，重复初始化摄像头，也会加剧不稳定问题。

核心思路是统一摄像头读取入口，避免多路径直接访问硬件：

• 启动独立后台线程专门负责读取摄像头帧，加线程锁保护更新全局最新帧缓存
• 视频流输出、回调快照两个场景都直接从缓存读取帧，不再直接调用camera.read()
• 禁用debug模式的自动重载，避免重复初始化摄像头占用硬件

import dash
from dash.dependencies import Output, Input
import dash_core_components as dcc
import dash_html_components as html
from flask import Flask, Response
import threading
import time
import cv2
import base64


colors = {
    'background': '#111111',
    'text': '#7FDBFF'
}

external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css']

server = Flask(__name__)
app = dash.Dash(__name__, server=server, external_stylesheets=external_stylesheets)

# 全局帧缓存 + 线程锁
latest_frame = None
frame_lock = threading.Lock()

# 初始化摄像头
camera = cv2.VideoCapture(0)

# 后台专属读帧线程，唯一直接访问摄像头的入口
def camera_read_task():
    global latest_frame
    while True:
        success, frame = camera.read()
        if success:
            with frame_lock:
                latest_frame = frame.copy()
        # 匹配摄像头默认帧率，30帧对应约33ms间隔，可根据实际调整
        time.sleep(0.03)

# 启动后台线程，daemon=True随主进程退出自动销毁
threading.Thread(target=camera_read_task, daemon=True).start()

def video_generator():
    while True:
        with frame_lock:
            if latest_frame is None:
                time.sleep(0.01)
                continue
            frame = latest_frame.copy()
        ret, jpeg = cv2.imencode('.jpg', frame)
        frame_bytes = jpeg.tobytes()
        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + frame_bytes + b'\r\n\r\n')

def photo_generator():
    with frame_lock:
        if latest_frame is None:
            return None
        return latest_frame.copy()


@server.route('/video_feed')
def video_feed():
    return Response(video_generator(),
                    mimetype='multipart/x-mixed-replace; boundary=frame')


app.layout = html.Div([
    html.Div([
        html.H4("Webcam"),
        html.Img(src="/video_feed")
    ], style={'width': '49%', 'display': 'inline-block', 'padding-left': '25px', 'vertical-align': 'top'}),
    html.Div([
        html.H4("Snapshot"),
        dcc.Interval( 
            id = 'graph-update', 
            interval = 100, # 可下调到10ms都不会崩溃
            n_intervals = 0
        ),
        html.Img(id="image", src=app.get_asset_url('test_0.png'))
    ], style={'display': 'inline-block', 'width': '49%', 'padding-right': '0px', 'vertical-align': 'top'})
], style={'backgroundColor': colors['background']})


@app.callback(
               Output('image', 'src'),
               Input('graph-update', 'n_intervals')
             )
def update_snapshot(n):
    frame = photo_generator()
    if frame is None:
        # 帧未初始化时返回原有内容不报错
        return dash.no_update
    _, buffer = cv2.imencode('.png', frame)
    source_image = base64.b64encode(buffer).decode('utf-8')
    return 'data:image/png;base64,{}'.format(source_image)


if __name__ == '__main__':
    # 加use_reloader=False避免debug模式重复启动进程占用摄像头
    app.run_server(debug=True, use_reloader=False)

• 可在缓存中同时存储编码好的JPG/PNG字节数据，避免视频流和快照场景重复编码，进一步降低CPU占用
• 若需要部署到多进程生产环境，可改用Redis等共享内存组件存储帧缓存，替换全局变量实现跨进程共享
• 可增加异常捕获逻辑，摄像头断开时自动尝试重连，进一步提升长时间运行的稳定性

内容的提问来源于stack exchange，提问作者Christian