I have the dash app below that pulls data from a DB and shows a graph/dataTable. This works. What isn’t working is trying to stream images from a webcam with openCV to the app. My most recent attempt is to use threading for the image acquisition from the live stream and then the Queue module to allow me to access the images in the app. I’m able to display static images stored on my computer, but I just can’t figure out the best approach to grab frames from a live stream and display them in the dash app.
My intention is to implement this with a clientside callback so the refresh rate is better, but I wanted to try server side callbacks first because I was able to get that working for the graph/table.
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():
if not q.empty():
frame = q.get()
_, buffer = cv2.imencode('.jpg', frame)
jpg_as_text = base64.b64encode(buffer.tobytes()).decode('ascii')
dataURI = 'data:image/jpeg;base64,' + jpg_as_text
return dataURI
# 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(), 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'))
def refresh_img(n_clicks):
return update_img()
def runApp():
app.run(debug=True)
# Run the App
if __name__ == '__main__':
p1 = threading.Thread(target=Receive)
p1.start()
p2 = threading.Thread(target=runApp())
p2.start()