🚀 Gen 5 of the leading AI app deployment platform launches October 6. Click for the livestream.

Does Dash support opencv video from webcam?

Does Dash support opencv video from webcam?
video= cv2.VideoCapture(0)
html.Img(src=‘video’.format(encoded_image01.decode()),style={
‘width’: 320,
‘height’: 240
}),

Hey @Swartedraak

You can try something like below. Also, for examples on opencv in Dash have you seen Plotly’s Object Detection App

import dash
import dash_core_components as dcc
import dash_html_components as html

from flask import Flask, Response
import cv2

class VideoCamera(object):
    def __init__(self):
        self.video = cv2.VideoCapture(0)

    def __del__(self):
        self.video.release()

    def get_frame(self):
        success, image = self.video.read()
        ret, jpeg = cv2.imencode('.jpg', image)
        return jpeg.tobytes()


def gen(camera):
    while True:
        frame = camera.get_frame()
        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n\r\n')

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

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

app.layout = html.Div([
    html.H1("Webcam Test"),
    html.Img(src="/video_feed")
])

if __name__ == '__main__':
    app.run_server(debug=True)
3 Likes

Is there a way to stream video from a network streaming link? such as “rtsp://10.47.103.73:XXXX/Name/XXXXX.stream”

Figured it out. Replace the cv2.VideoCapture(0) with your link. Should work.

1 Like

Is it possible to do the same thing without having the video_feed shared for all users ?

I’m looking for the same behavior but with a video streaming but one for each user. Is it possible with Flask/Plotly Dash ?

thank you a lot that does hlep me out

Hi cmh,
A bit late, but would you care you copy your code here? I would love to embed a video from a streaming link but I am a novice and don’t even know where to begin.
Thank you!

Hey Guys,

I am building a data collection platform with Dash Plotly where I have to turn on and record 10 min of data from the Webcam while an IQ test is being displayed and performed through the dashboard. Do you guys have any idea of how I might be able to do this?

Currently, I am using open-cv and running a VideoCapture while loop in a separate thread and have callbacks that start and stop the recording loop, but the frame rate drops really quickly.

class recordThread(Thread):
    def __init__(self):
        Thread.__init__(self)
        self.video_nr = 0

    def run(self):
        self.running = True
    
    def record(self):
        self.running = True
        cap = cv2.VideoCapture(0)                                       # start camera stream
        video = np.zeros((720,1280,1))                                  # create initial frame being all black
        times = round(time.time() * 1000000)

        while self.running:
            # Capture frame-by-frame
            times = np.append(times, round(time.time() * 1000000))      # capture times stamp
            ret, frame = cap.read()                                     # capture a frame
            frameGray = np.expand_dims(cv2.cvtColor(frame,cv2.COLOR_BGR2GRAY), axis=2) # convert frame to grayscale (converts from 3D to 2D)
            video = np.append(video,frameGray,axis=2)                   # write current grayscale frame to video object

        print("Thread stopped")                             
        np.save(f'video_{self.video_nr}.npy',video)                     # save the video 
        np.save(f'time_stamps_{self.video_nr}.npy',times)               # save time stamps
        cap.release()                                                   # stop camera stream
        self.video_nr += 1                                              # add one to the video itterator                                                        

    def terminate(self):
        self.running = False
        print(self.running)

app = dash.Dash(__name__)
app.layout = html.Div(
    children=[
        html.H1(children="Video Recording",),
        html.P(
            children="Analyze the behavior of avocado prices"
            " and the number of avocados sold in the US"
            " between 2015 and 2018", 
        ),
        html.Div([
            html.P(children="Not recording",id="rec-string"),
            html.Button('Start Video', id="rec-button",n_clicks=0, className = "graphButtons")
        ]),dcc.Store(id="rec-store"),dcc.Store(id="recording"),
    ]
)


@app.callback([
    Output("rec-button","children"),
    Output("rec-store","value")],
    [Input("rec-button", 'n_clicks')]
)
def rec_button(n_clicks):
    click = n_clicks%2
    if click == 0:
        data = False
        return f"Start recording", data
    elif click == 1:
        data = True
        return f"Stop recording", data

@app.callback([
    Output("rec-string","children")],
    [Input("rec-store","value"),Input("rec-button", 'n_clicks')]
    )
def recordVideo(rec,n_clicks):
    if rec:
        print("video start")
        recording.record()
        print("running")
        return "Now Recording"

    elif n_clicks > 1 and n_clicks%2 == 0:
        print("terminating")
        recording.terminate()
        return "Not Recording"

if __name__ == "__main__":
    recording = recordThread()
    recording.start()
    app.run_server(debug=True)

hope you guys can help!

UPDATE
I got it to work. If anyone wants to see the code then just contact me

Hi Boe,

I’d be quite keen to have a look at your solution since I am in the process of trying to do something that has some similarities!
Would really appreciate if you could share
Thanks heaps!

I’ll be sure to keep an eye on this thread. Looking for the same information. Thanks for sharing. Bumped into your thread.

great people me

Here is a small example, where video is streamed to an Img element via a Websocket component. It enables a smooth video steam, at least on my laptop :slight_smile:

import asyncio
import base64
import dash, cv2
import dash_html_components as html
import threading

from dash.dependencies import Output, Input
from quart import Quart, websocket
from dash_extensions import WebSocket


class VideoCamera(object):
    def __init__(self, video_path):
        self.video = cv2.VideoCapture(video_path)

    def __del__(self):
        self.video.release()

    def get_frame(self):
        success, image = self.video.read()
        _, jpeg = cv2.imencode('.jpg', image)
        return jpeg.tobytes()


# Setup small Quart server for streaming via websocket.
server = Quart(__name__)
delay_between_frames = 0.05  # add delay (in seconds) if CPU usage is too high


@server.websocket("/stream")
async def stream():
    camera = VideoCamera(0)  # zero means webcam
    while True:
        if delay_between_frames is not None:
            await asyncio.sleep(delay_between_frames)  # add delay if CPU usage is too high
        frame = camera.get_frame()
        await websocket.send(f"data:image/jpeg;base64, {base64.b64encode(frame).decode()}")


# Create small Dash application for UI.
app = dash.Dash(__name__)
app.layout = html.Div([
    html.Img(style={'width': '40%', 'padding': 10}, id="video"),
    WebSocket(url=f"ws://127.0.0.1:5000/stream", id="ws")
])
# Copy data from websocket to Img element.
app.clientside_callback("function(m){return m? m.data : '';}", Output(f"video", "src"), Input(f"ws", "message"))

if __name__ == '__main__':
    threading.Thread(target=app.run_server).start()
    server.run()