app.py

from streamlit_webrtc import webrtc_streamer
import numpy as np
import streamlit as st

import time
import cv2
import numpy as np
import av
import threading

lock = threading.Lock()

class FrameRate:
    def __init__(self) -> None:
        self.c: int = 0
        self.start_time: float = None
        self.NO_FRAMES = 100
        self.fps: float = -1

    def reset(self) -> None:
        self.start_time = time.time()
        self.c = 0
        self.fps = -1

    def count(self) -> None:
        self.c += 1
        if self.c % self.NO_FRAMES == 0:
            self.c = 0
            end_time = time.time()
            self.fps = self.NO_FRAMES / (end_time - self.start_time)
            self.start_time = end_time

    def show_fps(self, image: np.ndarray) -> np.ndarray:
        if self.fps != -1:
            return cv2.putText(
                image, 
                f'FPS {self.fps:.0f}', 
                (50, 50), 
                cv2.FONT_HERSHEY_SIMPLEX, 
                fontScale=1, 
                color=(255, 0, 0), 
                thickness=2
            )
        else:
            return image


rtc_configuration = {
    "iceServers": [{"urls": [
        "stun:stun1.l.google.com:19302",
        "stun:stun2.l.google.com:19302",
        "stun:stun3.l.google.com:19302",
        "stun:stun4.l.google.com:19302",
    ]}],
}

class ImgContainer:
    img: np.ndarray = None # raw image
    frame_rate: FrameRate = FrameRate()

def video_frame_callback(frame: av.VideoFrame) -> av.VideoFrame:
    img = frame.to_ndarray(format="bgr24")
    with lock:
        img_container.img = img
        img_container.frame_rate.count()
        img = img_container.frame_rate.show_fps(img)
    return av.VideoFrame.from_ndarray(img, format="bgr24")

img_container = ImgContainer()
img_container.frame_rate.reset()
ctx = st.session_state.ctx = webrtc_streamer(
    key="snapshot", 
    video_frame_callback=video_frame_callback,
    rtc_configuration=rtc_configuration
)