Changed to streamlit-webrtc object detection

app.py

@@ -1,146 +1,163 @@
-import os
-# os.environ['OPENCV_AVFOUNDATION_SKIP_AUTH'] = '1'
+"""Object detection demo with MobileNet SSD.
+This model and code are based on
+https://github.com/robmarkcole/object-detection-app
+"""
 
-import streamlit as st
+import logging
+import queue
+from pathlib import Path
+from typing import List, NamedTuple
+
+import av
 import cv2
 import numpy as np
-from transformers import pipeline
-from PIL import Image, ImageDraw
-from mtcnn import MTCNN
-
-# Initialize the Hugging Face pipeline for facial emotion detection
-emotion_pipeline = pipeline("image-classification", model="trpakov/vit-face-expression")
-
-# Initialize MTCNN for face detection
-mtcnn = MTCNN()
-
-# Function to analyze sentiment
-def analyze_sentiment(face):
-    # Convert face to RGB
-    rgb_face = cv2.cvtColor(face, cv2.COLOR_BGR2RGB)
-    # Convert the face to a PIL image
-    pil_image = Image.fromarray(rgb_face)
-    # Analyze sentiment using the Hugging Face pipeline
-    results = emotion_pipeline(pil_image)
-    # Get the dominant emotion
-    dominant_emotion = max(results, key=lambda x: x['score'])['label']
-    return dominant_emotion
-
-TEXT_SIZE = 3
-
-# Function to detect faces, analyze sentiment, and draw a red box around them
-def detect_and_draw_faces(frame):
-    # Detect faces using MTCNN
-    results = mtcnn.detect_faces(frame)
-    
-    # Draw on the frame
-    for result in results:
-        x, y, w, h = result['box']
-        face = frame[y:y+h, x:x+w]
-        sentiment = analyze_sentiment(face)
-        cv2.rectangle(frame, (x, y), (x+w, y+h), (0, 0, 255), 10)  # Thicker red box
-        
-        # Calculate position for the text background and the text itself
-        text_size = cv2.getTextSize(sentiment, cv2.FONT_HERSHEY_SIMPLEX, TEXT_SIZE, 2)[0]
-        text_x = x
-        text_y = y - 10
-        background_tl = (text_x, text_y - text_size[1])
-        background_br = (text_x + text_size[0], text_y + 5)
-        
-        # Draw black rectangle as background
-        cv2.rectangle(frame, background_tl, background_br, (0, 0, 0), cv2.FILLED)
-        # Draw white text on top
-        cv2.putText(frame, sentiment, (text_x, text_y), cv2.FONT_HERSHEY_SIMPLEX, TEXT_SIZE, (255, 255, 255), 2)
-    
-    return frame
-
-# Function to capture video from webcam
-def video_stream():
-    video_capture = cv2.VideoCapture(0)
-    if not video_capture.isOpened():
-        st.error("Error: Could not open video capture device.")
-        return
-
-    while True:
-        ret, frame = video_capture.read()
-        if not ret:
-            st.error("Error: Failed to read frame from video capture device.")
-            break
-        yield frame
-
-    video_capture.release()
-
-# Streamlit UI
-st.markdown(
-    """
-    <style>
-        .main {
-            background-color: #FFFFFF;
-        }
-        .reportview-container .main .block-container{
-            padding-top: 2rem;
-        }
-        h1 {
-            color: #E60012;
-            font-family: 'Arial Black', Gadget, sans-serif;
-        }
-        h2 {
-            color: #E60012;
-            font-family: 'Arial', sans-serif;
-        }
-        h3 {
-            color: #333333;
-            font-family: 'Arial', sans-serif;
-        }
-        .stButton button {
-            background-color: #E60012;
-            color: white;
-            border-radius: 5px;
-            font-size: 16px;
-        }
-    </style>
-    """,
-    unsafe_allow_html=True
-)
+import streamlit as st
+from streamlit_webrtc import WebRtcMode, webrtc_streamer
+
+from utils.download import download_file 
+from utils.turn import get_ice_servers
+
+HERE = Path(__file__).parent
+ROOT = HERE.parent
+
+logger = logging.getLogger(__name__)
+
+
+MODEL_URL = "https://github.com/robmarkcole/object-detection-app/raw/master/model/MobileNetSSD_deploy.caffemodel"  # noqa: E501
+MODEL_LOCAL_PATH = ROOT / "./models/MobileNetSSD_deploy.caffemodel"
+PROTOTXT_URL = "https://github.com/robmarkcole/object-detection-app/raw/master/model/MobileNetSSD_deploy.prototxt.txt"  # noqa: E501
+PROTOTXT_LOCAL_PATH = ROOT / "./models/MobileNetSSD_deploy.prototxt.txt"
 
-st.title("Computer Vision Test Lab")
-st.subheader("Facial Sentiment")
+CLASSES = [
+    "background",
+    "aeroplane",
+    "bicycle",
+    "bird",
+    "boat",
+    "bottle",
+    "bus",
+    "car",
+    "cat",
+    "chair",
+    "cow",
+    "diningtable",
+    "dog",
+    "horse",
+    "motorbike",
+    "person",
+    "pottedplant",
+    "sheep",
+    "sofa",
+    "train",
+    "tvmonitor",
+]
 
-# Columns for input and output streams
-col1, col2 = st.columns(2)
 
-with col1:
-    st.header("Input Stream")
-    st.subheader("Webcam")
-    video_placeholder = st.empty()
+class Detection(NamedTuple):
+    class_id: int
+    label: str
+    score: float
+    box: np.ndarray
 
-with col2:
-    st.header("Output Stream")
-    st.subheader("Analysis")
-    output_placeholder = st.empty()
 
-sentiment_placeholder = st.empty()
+@st.cache_resource  # type: ignore
+def generate_label_colors():
+    return np.random.uniform(0, 255, size=(len(CLASSES), 3))
 
-# Start video stream
-video_capture = cv2.VideoCapture(0)
-if not video_capture.isOpened():
-    st.error("Error: Could not open video capture device.")
+
+COLORS = generate_label_colors()
+
+download_file(MODEL_URL, MODEL_LOCAL_PATH, expected_size=23147564)
+download_file(PROTOTXT_URL, PROTOTXT_LOCAL_PATH, expected_size=29353)
+
+
+# Session-specific caching
+cache_key = "object_detection_dnn"
+if cache_key in st.session_state:
+    net = st.session_state[cache_key]
 else:
-    while True:
-        ret, frame = video_capture.read()
-        if not ret:
-            st.error("Error: Failed to read frame from video capture device.")
-            break
-
-        # Display the input stream with the red box around the face
-        video_placeholder.image(frame, channels="BGR")
-        
-        # Detect faces, analyze sentiment, and draw red boxes with sentiment labels
-        frame_with_boxes = detect_and_draw_faces(frame)
-        
-        # Display the output stream (here it's the same as input, modify as needed)
-        output_placeholder.image(frame_with_boxes, channels="BGR")
-        
-        # Add a short delay to control the frame rate
-        if cv2.waitKey(1) & 0xFF == ord('q'):
-            break
+    net = cv2.dnn.readNetFromCaffe(str(PROTOTXT_LOCAL_PATH), str(MODEL_LOCAL_PATH))
+    st.session_state[cache_key] = net
+
+score_threshold = st.slider("Score threshold", 0.0, 1.0, 0.5, 0.05)
+
+# NOTE: The callback will be called in another thread,
+#       so use a queue here for thread-safety to pass the data
+#       from inside to outside the callback.
+# TODO: A general-purpose shared state object may be more useful.
+result_queue: "queue.Queue[List[Detection]]" = queue.Queue()
+
+
+def video_frame_callback(frame: av.VideoFrame) -> av.VideoFrame:
+    image = frame.to_ndarray(format="bgr24")
+
+    # Run inference
+    blob = cv2.dnn.blobFromImage(
+        cv2.resize(image, (300, 300)), 0.007843, (300, 300), 127.5
+    )
+    net.setInput(blob)
+    output = net.forward()
+
+    h, w = image.shape[:2]
+
+    # Convert the output array into a structured form.
+    output = output.squeeze()  # (1, 1, N, 7) -> (N, 7)
+    output = output[output[:, 2] >= score_threshold]
+    detections = [
+        Detection(
+            class_id=int(detection[1]),
+            label=CLASSES[int(detection[1])],
+            score=float(detection[2]),
+            box=(detection[3:7] * np.array([w, h, w, h])),
+        )
+        for detection in output
+    ]
+
+    # Render bounding boxes and captions
+    for detection in detections:
+        caption = f"{detection.label}: {round(detection.score * 100, 2)}%"
+        color = COLORS[detection.class_id]
+        xmin, ymin, xmax, ymax = detection.box.astype("int")
+
+        cv2.rectangle(image, (xmin, ymin), (xmax, ymax), color, 2)
+        cv2.putText(
+            image,
+            caption,
+            (xmin, ymin - 15 if ymin - 15 > 15 else ymin + 15),
+            cv2.FONT_HERSHEY_SIMPLEX,
+            0.5,
+            color,
+            2,
+        )
+
+    result_queue.put(detections)
+
+    return av.VideoFrame.from_ndarray(image, format="bgr24")
+
+
+webrtc_ctx = webrtc_streamer(
+    key="object-detection",
+    mode=WebRtcMode.SENDRECV,
+    rtc_configuration={"iceServers": get_ice_servers()},
+    video_frame_callback=video_frame_callback,
+    media_stream_constraints={"video": True, "audio": False},
+    async_processing=True,
+)
+
+if st.checkbox("Show the detected labels", value=True):
+    if webrtc_ctx.state.playing:
+        labels_placeholder = st.empty()
+        # NOTE: The video transformation with object detection and
+        # this loop displaying the result labels are running
+        # in different threads asynchronously.
+        # Then the rendered video frames and the labels displayed here
+        # are not strictly synchronized.
+        while True:
+            result = result_queue.get()
+            labels_placeholder.table(result)
+
+st.markdown(
+    "This demo uses a model and code from "
+    "https://github.com/robmarkcole/object-detection-app. "
+    "Many thanks to the project."
+)

app.py.sentiment-one

@@ -1,118 +0,0 @@
-import os
-os.environ['OPENCV_AVFOUNDATION_SKIP_AUTH'] = '1'
-
-import streamlit as st
-import cv2
-from transformers import pipeline
-from PIL import Image
-
-# Initialize the Hugging Face pipeline for facial emotion detection
-emotion_pipeline = pipeline("image-classification", model="dima806/facial_emotions_image_detection")
-
-# Function to analyze sentiment
-def analyze_sentiment(frame):
-    # Convert frame to RGB
-    rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-    # Convert the frame to a PIL image
-    pil_image = Image.fromarray(rgb_frame)
-    # Analyze sentiment using the Hugging Face pipeline
-    results = emotion_pipeline(pil_image)    # Analyze sentiment using the Hugging Face pipeline
-    results = emotion_pipeline(pil_image)
-    # Get the dominant emotion
-    dominant_emotion = max(results, key=lambda x: x['score'])['label']
-    return dominant_emotion
-
-# Function to capture video from webcam
-def video_stream():
-    video_capture = cv2.VideoCapture(0)
-    if not video_capture.isOpened():
-        st.error("Error: Could not open video capture device.")
-        return
-
-    while True:
-        ret, frame = video_capture.read()
-        if not ret:
-            st.error("Error: Failed to read frame from video capture device.")
-            break
-        yield frame
-
-    video_capture.release()
-
-# Streamlit UI
-st.markdown(
-    """
-    <style>
-        .main {
-            background-color: #FFFFFF;
-        }
-        .reportview-container .main .block-container{
-            padding-top: 2rem;
-        }
-        h1 {
-            color: #E60012;
-            font-family: 'Arial Black', Gadget, sans-serif;
-        }
-        h2 {
-            color: #E60012;
-            font-family: 'Arial', sans-serif;
-        }
-        h3 {
-            color: #333333;
-            font-family: 'Arial', sans-serif;
-        }
-        .stButton button {
-            background-color: #E60012;
-            color: white;
-            border-radius: 5px;
-            font-size: 16px;
-        }
-    </style>
-    """,
-    unsafe_allow_html=True
-)
-
-st.title("Computer Vision Test Lab")
-st.subheader("Facial Sentiment")
-
-# Columns for input and output streams
-col1, col2 = st.columns(2)
-
-with col1:
-    st.header("Input Stream")
-    st.subheader("Webcam")
-    video_placeholder = st.empty()
-
-with col2:
-    st.header("Output Stream")
-    st.subheader("Analysis")
-    output_placeholder = st.empty()
-
-sentiment_placeholder = st.empty()
-
-# Start video stream
-video_capture = cv2.VideoCapture(0)
-if not video_capture.isOpened():
-    st.error("Error: Could not open video capture device.")
-else:
-    while True:
-        ret, frame = video_capture.read()
-        if not ret:
-            st.error("Error: Failed to read frame from video capture device.")
-            break
-
-        # Display the input stream
-        video_placeholder.image(frame, channels="BGR")
-        
-        # Analyze sentiment
-        sentiment = analyze_sentiment(frame)
-        
-        # Display the output stream (here it's the same as input, modify as needed)
-        output_placeholder.image(frame, channels="BGR")
-        
-        # Display sentiment
-        sentiment_placeholder.write(f"Sentiment: {sentiment}")
-
-        # Add a short delay to control the frame rate
-        if cv2.waitKey(1) & 0xFF == ord('q'):
-            break
-        

object_detection.py

@@ -0,0 +1,163 @@
+"""Object detection demo with MobileNet SSD.
+This model and code are based on
+https://github.com/robmarkcole/object-detection-app
+"""
+
+import logging
+import queue
+from pathlib import Path
+from typing import List, NamedTuple
+
+import av
+import cv2
+import numpy as np
+import streamlit as st
+from streamlit_webrtc import WebRtcMode, webrtc_streamer
+
+from utils.download import download_file 
+from utils.turn import get_ice_servers
+
+HERE = Path(__file__).parent
+ROOT = HERE.parent
+
+logger = logging.getLogger(__name__)
+
+
+MODEL_URL = "https://github.com/robmarkcole/object-detection-app/raw/master/model/MobileNetSSD_deploy.caffemodel"  # noqa: E501
+MODEL_LOCAL_PATH = ROOT / "./models/MobileNetSSD_deploy.caffemodel"
+PROTOTXT_URL = "https://github.com/robmarkcole/object-detection-app/raw/master/model/MobileNetSSD_deploy.prototxt.txt"  # noqa: E501
+PROTOTXT_LOCAL_PATH = ROOT / "./models/MobileNetSSD_deploy.prototxt.txt"
+
+CLASSES = [
+    "background",
+    "aeroplane",
+    "bicycle",
+    "bird",
+    "boat",
+    "bottle",
+    "bus",
+    "car",
+    "cat",
+    "chair",
+    "cow",
+    "diningtable",
+    "dog",
+    "horse",
+    "motorbike",
+    "person",
+    "pottedplant",
+    "sheep",
+    "sofa",
+    "train",
+    "tvmonitor",
+]
+
+
+class Detection(NamedTuple):
+    class_id: int
+    label: str
+    score: float
+    box: np.ndarray
+
+
+@st.cache_resource  # type: ignore
+def generate_label_colors():
+    return np.random.uniform(0, 255, size=(len(CLASSES), 3))
+
+
+COLORS = generate_label_colors()
+
+download_file(MODEL_URL, MODEL_LOCAL_PATH, expected_size=23147564)
+download_file(PROTOTXT_URL, PROTOTXT_LOCAL_PATH, expected_size=29353)
+
+
+# Session-specific caching
+cache_key = "object_detection_dnn"
+if cache_key in st.session_state:
+    net = st.session_state[cache_key]
+else:
+    net = cv2.dnn.readNetFromCaffe(str(PROTOTXT_LOCAL_PATH), str(MODEL_LOCAL_PATH))
+    st.session_state[cache_key] = net
+
+score_threshold = st.slider("Score threshold", 0.0, 1.0, 0.5, 0.05)
+
+# NOTE: The callback will be called in another thread,
+#       so use a queue here for thread-safety to pass the data
+#       from inside to outside the callback.
+# TODO: A general-purpose shared state object may be more useful.
+result_queue: "queue.Queue[List[Detection]]" = queue.Queue()
+
+
+def video_frame_callback(frame: av.VideoFrame) -> av.VideoFrame:
+    image = frame.to_ndarray(format="bgr24")
+
+    # Run inference
+    blob = cv2.dnn.blobFromImage(
+        cv2.resize(image, (300, 300)), 0.007843, (300, 300), 127.5
+    )
+    net.setInput(blob)
+    output = net.forward()
+
+    h, w = image.shape[:2]
+
+    # Convert the output array into a structured form.
+    output = output.squeeze()  # (1, 1, N, 7) -> (N, 7)
+    output = output[output[:, 2] >= score_threshold]
+    detections = [
+        Detection(
+            class_id=int(detection[1]),
+            label=CLASSES[int(detection[1])],
+            score=float(detection[2]),
+            box=(detection[3:7] * np.array([w, h, w, h])),
+        )
+        for detection in output
+    ]
+
+    # Render bounding boxes and captions
+    for detection in detections:
+        caption = f"{detection.label}: {round(detection.score * 100, 2)}%"
+        color = COLORS[detection.class_id]
+        xmin, ymin, xmax, ymax = detection.box.astype("int")
+
+        cv2.rectangle(image, (xmin, ymin), (xmax, ymax), color, 2)
+        cv2.putText(
+            image,
+            caption,
+            (xmin, ymin - 15 if ymin - 15 > 15 else ymin + 15),
+            cv2.FONT_HERSHEY_SIMPLEX,
+            0.5,
+            color,
+            2,
+        )
+
+    result_queue.put(detections)
+
+    return av.VideoFrame.from_ndarray(image, format="bgr24")
+
+
+webrtc_ctx = webrtc_streamer(
+    key="object-detection",
+    mode=WebRtcMode.SENDRECV,
+    rtc_configuration={"iceServers": get_ice_servers()},
+    video_frame_callback=video_frame_callback,
+    media_stream_constraints={"video": True, "audio": False},
+    async_processing=True,
+)
+
+if st.checkbox("Show the detected labels", value=True):
+    if webrtc_ctx.state.playing:
+        labels_placeholder = st.empty()
+        # NOTE: The video transformation with object detection and
+        # this loop displaying the result labels are running
+        # in different threads asynchronously.
+        # Then the rendered video frames and the labels displayed here
+        # are not strictly synchronized.
+        while True:
+            result = result_queue.get()
+            labels_placeholder.table(result)
+
+st.markdown(
+    "This demo uses a model and code from "
+    "https://github.com/robmarkcole/object-detection-app. "
+    "Many thanks to the project."
+)

requirements.txt

@@ -7,3 +7,4 @@ mtcnn
 setuptools
 tensorflow
 tf-keras
+streamlit_webrtc

run_streamlist.sh

@@ -1,5 +0,0 @@
-#!/bin/bash
-# Set Chrome as the default browser for this session
-export BROWSER="/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
-# Run Streamlit with the provided arguments
-streamlit run "$@"

app.py.safe → sentiment.py

@@ -1,15 +1,27 @@
-import os
-os.environ['OPENCV_AVFOUNDATION_SKIP_AUTH'] = '1'
+import threading
 
 import streamlit as st
 import cv2
 import numpy as np
 from transformers import pipeline
 from PIL import Image, ImageDraw
+from mtcnn import MTCNN
+from streamlit_webrtc import webrtc_streamer
+import logging
 
-# Initialize the Hugging Face pipeline for facial emotion detection using the "trpakov/vit-face-expression" model
+# Suppress transformers progress bars
+logging.getLogger("transformers").setLevel(logging.ERROR)
+
+lock = threading.Lock()
+img_container = {"webcam": None,
+                 "analzyed": None}
+
+# Initialize the Hugging Face pipeline for facial emotion detection
 emotion_pipeline = pipeline("image-classification", model="trpakov/vit-face-expression")
 
+# Initialize MTCNN for face detection
+mtcnn = MTCNN()
+
 # Function to analyze sentiment
 def analyze_sentiment(face):
     # Convert face to RGB
@@ -26,58 +38,29 @@ TEXT_SIZE = 3
 
 # Function to detect faces, analyze sentiment, and draw a red box around them
 def detect_and_draw_faces(frame):
-    # Convert frame to RGB
-    rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-    # Convert the frame to a PIL image
-    pil_image = Image.fromarray(rgb_frame)
-    # Analyze sentiment using the Hugging Face pipeline
-    results = emotion_pipeline(pil_image)
-
-    # Print the results to understand the structure
-    print(results)
+    # Detect faces using MTCNN
+    results = mtcnn.detect_faces(frame)
     
-    # Draw on the PIL image
-    draw = ImageDraw.Draw(pil_image)
-    
-    # Iterate through detected faces
+    # Draw on the frame
     for result in results:
-        box = result['box']
-        sentiment = result['label']
-        
-        # Draw rectangle and text
-        x, y, w, h = box['left'], box['top'], box['width'], box['height']
-        draw.rectangle(((x, y), (x+w, y+h)), outline="red", width=3)
+        x, y, w, h = result['box']
+        face = frame[y:y+h, x:x+w]
+        sentiment = analyze_sentiment(face)
+        cv2.rectangle(frame, (x, y), (x+w, y+h), (0, 0, 255), 10)  # Thicker red box
         
         # Calculate position for the text background and the text itself
-        text_size = draw.textsize(sentiment)
-        background_tl = (x, y - text_size[1] - 5)
-        background_br = (x + text_size[0], y)
+        text_size = cv2.getTextSize(sentiment, cv2.FONT_HERSHEY_SIMPLEX, TEXT_SIZE, 2)[0]
+        text_x = x
+        text_y = y - 10
+        background_tl = (text_x, text_y - text_size[1])
+        background_br = (text_x + text_size[0], text_y + 5)
         
         # Draw black rectangle as background
-        draw.rectangle([background_tl, background_br], fill="black")
+        cv2.rectangle(frame, background_tl, background_br, (0, 0, 0), cv2.FILLED)
         # Draw white text on top
-        draw.text((x, y - text_size[1]), sentiment, fill="white")
-    
-    # Convert back to OpenCV format
-    frame_with_boxes = cv2.cvtColor(np.array(pil_image), cv2.COLOR_RGB2BGR)
+        cv2.putText(frame, sentiment, (text_x, text_y), cv2.FONT_HERSHEY_SIMPLEX, TEXT_SIZE, (255, 255, 255), 2)
     
-    return frame_with_boxes
-
-# Function to capture video from webcam
-def video_stream():
-    video_capture = cv2.VideoCapture(0)
-    if not video_capture.isOpened():
-        st.error("Error: Could not open video capture device.")
-        return
-
-    while True:
-        ret, frame = video_capture.read()
-        if not ret:
-            st.error("Error: Failed to read frame from video capture device.")
-            break
-        yield frame
-
-    video_capture.release()
+    return frame
 
 # Streamlit UI
 st.markdown(
@@ -130,26 +113,29 @@ with col2:
 
 sentiment_placeholder = st.empty()
 
-# Start video stream
-video_capture = cv2.VideoCapture(0)
-if not video_capture.isOpened():
-    st.error("Error: Could not open video capture device.")
-else:
-    while True:
-        ret, frame = video_capture.read()
-        if not ret:
-            st.error("Error: Failed to read frame from video capture device.")
-            break
-
-        # Detect faces, analyze sentiment, and draw red boxes with sentiment labels
-        frame_with_boxes = detect_and_draw_faces(frame)
-        
-        # Display the input stream with the red box around the face
-        video_placeholder.image(frame_with_boxes, channels="BGR")
-        
-        # Display the output stream (here it's the same as input, modify as needed)
-        output_placeholder.image(frame_with_boxes, channels="BGR")
-        
-        # Add a short delay to control the frame rate
-        if cv2.waitKey(1) & 0xFF == ord('q'):
-            break
+def video_frame_callback(frame):
+    try:
+        with lock:
+            img = frame.to_ndarray(format="bgr24")
+            img_container["webcam"] = img
+            frame_with_boxes = detect_and_draw_faces(img)
+            img_container["analyzed"] = frame_with_boxes
+
+    except Exception as e:
+        st.error(f"Error processing frame: {e}")
+
+    return frame
+
+ctx = webrtc_streamer(key="webcam", video_frame_callback=video_frame_callback)
+
+while ctx.state.playing:
+    with lock:
+        print(img_container)
+        img = img_container["webcam"]
+        frame_with_boxes = img_container["analyzed"]
+
+    if img is None:
+        continue
+
+    video_placeholder.image(img, channels="BGR")
+    output_placeholder.image(frame_with_boxes, channels="BGR")

utils/__init__.py

@@ -0,0 +1 @@
+

utils/download.py

@@ -0,0 +1,50 @@
+import urllib.request
+from pathlib import Path
+
+import streamlit as st
+
+
+# This code is based on https://github.com/streamlit/demo-self-driving/blob/230245391f2dda0cb464008195a470751c01770b/streamlit_app.py#L48  # noqa: E501
+def download_file(url, download_to: Path, expected_size=None):
+    # Don't download the file twice.
+    # (If possible, verify the download using the file length.)
+    if download_to.exists():
+        if expected_size:
+            if download_to.stat().st_size == expected_size:
+                return
+        else:
+            st.info(f"{url} is already downloaded.")
+            if not st.button("Download again?"):
+                return
+
+    download_to.parent.mkdir(parents=True, exist_ok=True)
+
+    # These are handles to two visual elements to animate.
+    weights_warning, progress_bar = None, None
+    try:
+        weights_warning = st.warning("Downloading %s..." % url)
+        progress_bar = st.progress(0)
+        with open(download_to, "wb") as output_file:
+            with urllib.request.urlopen(url) as response:
+                length = int(response.info()["Content-Length"])
+                counter = 0.0
+                MEGABYTES = 2.0**20.0
+                while True:
+                    data = response.read(8192)
+                    if not data:
+                        break
+                    counter += len(data)
+                    output_file.write(data)
+
+                    # We perform animation by overwriting the elements.
+                    weights_warning.warning(
+                        "Downloading %s... (%6.2f/%6.2f MB)"
+                        % (url, counter / MEGABYTES, length / MEGABYTES)
+                    )
+                    progress_bar.progress(min(counter / length, 1.0))
+    # Finally, we remove these visual elements by calling .empty().
+    finally:
+        if weights_warning is not None:
+            weights_warning.empty()
+        if progress_bar is not None:
+            progress_bar.empty()

utils/turn.py

@@ -0,0 +1,39 @@
+import logging 
+import os
+
+import streamlit as st
+from twilio.base.exceptions import TwilioRestException
+from twilio.rest import Client
+
+logger = logging.getLogger(__name__)
+
+
+def get_ice_servers():
+    """Use Twilio's TURN server because Streamlit Community Cloud has changed
+    its infrastructure and WebRTC connection cannot be established without TURN server now.  # noqa: E501
+    We considered Open Relay Project (https://www.metered.ca/tools/openrelay/) too,
+    but it is not stable and hardly works as some people reported like https://github.com/aiortc/aiortc/issues/832#issuecomment-1482420656  # noqa: E501
+    See https://github.com/whitphx/streamlit-webrtc/issues/1213
+    """
+
+    # Ref: https://www.twilio.com/docs/stun-turn/api
+    try:
+        account_sid = os.environ["TWILIO_ACCOUNT_SID"]
+        auth_token = os.environ["TWILIO_AUTH_TOKEN"]
+    except KeyError:
+        logger.warning(
+            "Twilio credentials are not set. Fallback to a free STUN server from Google."  # noqa: E501
+        )
+        return [{"urls": ["stun:stun.l.google.com:19302"]}]
+
+    client = Client(account_sid, auth_token)
+
+    try:
+        token = client.tokens.create()
+    except TwilioRestException as e:
+        st.warning(
+            f"Error occurred while accessing Twilio API. Fallback to a free STUN server from Google. ({e})"  # noqa: E501
+        )
+        return [{"urls": ["stun:stun.l.google.com:19302"]}]
+
+    return token.ice_servers