New Component: streamlit-webrtc, a new way to deal with real-time media streams

Custom Components streamlit-webrtc
, ,
103

Hi everyone (and whitphx!)

First of all I am super amazed by your contribution to the streamlit community. I still haven’t come across a better alternative for manipulation video and classification! :partying_face:

I am currently using streamlit-webrtc for reading barcode and it totally works! :grin:

image
image1073×730 78.8 KB

However as soon as I have detected a barcode I wish to have print the barcode like so st.subheader((“Detected barcode:” detected_barcode)), and then use barcode to fetch more information in a database etc. However I read that it’s hard to extract information from the webrtc_streamer object to global :smiling_face_with_tear:. Any ideas for a work around?

I could only come up with a complicated attempt, but I feel like it doesn’t need to be this complicated :sweat_smile::

  1. In the beginning of the script I would generate a random string
  2. Upon detecting a barcode, store this value in a new SQL entry WHERE id = random_string
  3. Then after initiating live_detection(), I would start a “while true” loop that would constantly look for new SQL entries WHERE id = random_string (this should work this it’s the same session). If found, then do st.subheader((“Detected barcode:” sql_barcode())) and del live_detection to freeze / collapse the videostream.
  4. At the same time I spawn a button that when triggered would reload the page/session and start live_detection() again.

Note: For barcodes, it is extremely painful to only use single images as the detection rate is super low and you often have to take 4-5 images before anything is detected. Video solves this problem beautifully.

Here is my current code (without attempting to write to a SQL database)

import av
import cv2
import streamlit as st 
from pyzbar.pyzbar import decode
from streamlit_webrtc import (webrtc_streamer, VideoProcessorBase,WebRtcMode)

def live_detection(play_state):

   class BarcodeProcessor(VideoProcessorBase):

      def __init__(self) -> None:
         self.barcode_val = False
      
      def BarcodeReader(self, image):
         detectedBarcodes = decode(image)
         if not detectedBarcodes:
            print("\n No barcode! \n")
            return image, False

         else:
            for barcode in detectedBarcodes: 
               (x, y, w, h) = barcode.rect
               cv2.rectangle(image, (x-10, y-10),
                              (x + w+10, y + h+10),
                              (0, 255, 0), 2)

            if detectedBarcodes[0] != "":
               return image, detectedBarcodes[0]


      def recv(self, frame: av.VideoFrame) -> av.VideoFrame:
         image = frame.to_ndarray(format="bgr24")

         annotated_image, result = self.BarcodeReader(image)
        
         if result == False:
            return av.VideoFrame.from_ndarray(image, format="bgr24")

         else:

            self.barcode_val = result[0]
            play_state = False
            return av.VideoFrame.from_ndarray(annotated_image, format="bgr24")

   stream = webrtc_streamer(
         key="barcode-detection",
         mode=WebRtcMode.SENDRECV,
         desired_playing_state=play_state,
         video_processor_factory=BarcodeProcessor,
         media_stream_constraints={"video": True, "audio": False},
         async_processing=True,
      )

play_state = True
detected_barcode = live_detection(play_state)
1 Like