如何用Python结合OpenCV实现摄像头画面向Dash应用的流式传输

我有一个从数据库拉取数据并展示图表/数据表的Dash应用，这部分功能正常。但无法实现用OpenCV将摄像头画面流式传输到该应用中。最近尝试使用线程获取直播流画面，再通过Queue模块让应用访问这些画面，能显示本地存储的静态图片，但始终找不到获取直播帧并在Dash应用中显示的最佳方案。

原本打算使用客户端回调来提升刷新率，但因为之前用服务器端回调实现了图表/数据表的功能，所以先尝试服务器端回调的方式。

import base64
import threading
import queue
from dash import Dash, html, dash_table, dcc, callback, Output, Input, dash, State
import plotly.express as px
import sqlite3
import pandas as pd
import dash_mantine_components as dmc
import cv2

dfSettings = pd.DataFrame


# Initialize the app - incorporate a Dash Mantine theme
#external_stylesheets = [dmc.theme.DEFAULT_COLORS]
app = Dash(__name__)#, external_stylesheets=external_stylesheets)


q = queue.Queue()


def Receive():
    cap = cv2.VideoCapture(0)
    ret, frame = cap.read()
    q.put(frame)
    while ret:
        ret, frame = cap.read()
        q.put(frame)

def update_graph():
    global dfSettings
    conn = sqlite3.connect('test.db')
    dfSettings = pd.read_sql_query("SELECT * FROM settings", conn)
    conn.commit()
    df = pd.read_sql_query("SELECT * FROM test ORDER BY datetime DESC", conn) #limit row # to db setting
    conn.commit()
    conn.close()

    fig = px.line(df, x = 'datetime', y = df.columns[1:5], template='plotly_dark')
    fig.update_layout(title={'text': 'Stove Room'}, title_x=0.5, xaxis_title="Date-Time", yaxis_title="Temperature F")
    fig.update_layout(legend_title_text="", legend_y=0.5)
    return fig


def update_metrics():
    conn = sqlite3.connect('test.db')
    df = pd.read_sql_query("SELECT * FROM test ORDER BY datetime", conn) #need to do more work here
    conn.commit()

    conn.close()

    metrics = df.to_dict('records')
    return metrics


def update_img(current_src):
    if not q.empty():
        frame = q.get()
        _, buffer = cv2.imencode('.jpg', frame)
        jpg_as_text = base64.b64encode(buffer.tobytes()).decode('ascii')
        return 'data:image/jpeg;base64,' + jpg_as_text
    return current_src



# App layout
app.layout = dmc.Container([
    dmc.Title('My First App with Data and Graph', color="blue", size="h3", align="center"),
    dmc.Grid([
        dmc.Col([
            dash_table.DataTable(data=update_metrics(), id='metrics', page_size=6, style_table={'overflowX': 'auto',
                                                                                                'height': '50vh'},
                style_header={'backgroundColor': 'rgb(30, 30, 30)', 'color': 'white'},
                style_data={'backgroundColor': 'rgb(50, 50, 50)', 'color': 'white'},
                style_cell={'text_align': 'center'})#font size add in the % string format thing to scale font to some percentage of total screen height
            #dcc.Interval(id='interval-component', interval=5*1000, n_intervals=0),
        ], span=6),
        dmc.Col([
            dcc.Graph(figure=update_graph(), id='graph-placeholder'),
            dcc.Interval(id='interval-component', interval=5*1000, n_intervals=0),
        ], span=6),
        dmc.Col([
            dmc.Image(src=update_img(None), id='img'),
            dcc.Interval(id='interval-component-2', interval=250, n_intervals=0)
        ], span=6),
    ], justify='center', align='stretch'),

], fluid=True, style={'backgroundColor': 'rgb(50, 50, 50)', 'color': 'white', 'width': '95vw', 'height': '95vh'})


@app.callback(
    dash.Output('graph-placeholder', 'figure'),
    dash.Input('interval-component', 'n_intervals'))
def refresh_graph(n_clicks):
    return update_graph()



@app.callback(
    dash.Output('metrics', 'data'),
    dash.Input('interval-component', 'n_intervals'))
def refresh_metrics(n_clicks):
    return update_metrics()


@app.callback(
    dash.Output('img', 'src'),
    dash.Input('interval-component-2', 'n_intervals'),
    dash.State('img', 'src'))
def refresh_img(n_clicks, current_src):
    return update_img(current_src)


def runApp():
    app.run(debug=True, use_reloader=False)


# Run the App
if __name__ == '__main__':
    p1 = threading.Thread(target=Receive, daemon=True)
    p1.start()
    runApp()

• 线程启动修复：原代码中threading.Thread(target=runApp())是直接调用函数而非传入引用，导致线程逻辑错误。改为直接调用runApp()，同时将摄像头线程设为daemon=True，确保主进程退出时线程自动终止；在app.run()中添加use_reloader=False，避免Debug模式下重复启动线程。
• 队列积压控制：优化Receive函数，添加队列大小判断，避免无限制存入帧导致内存占用过高：

  def Receive():
      cap = cv2.VideoCapture(0)
      try:
          while True:
              ret, frame = cap.read()
              if not ret:
                  break
              # 限制队列大小，避免帧积压
              if q.qsize() < 10:
                  q.put(frame)
      finally:
          cap.release()
  

• 空队列处理：修改update_img函数，传入当前图片的src状态，当队列为空时返回原图片链接，避免画面空白。
• 摄像头资源释放：在Receive线程中添加try...finally块，确保线程退出时释放摄像头硬件资源。

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