import time import cv2 import numpy as np import streamlit as st from streamlit_webrtc import webrtc_streamer, WebRtcMode, VideoTransformerBase # Set wide layout and page title st.set_page_config(page_title="Live Stream Broadcast", layout="wide") # ----------------------------------------------------------------------------- # Custom CSS: Hide the timeline parts of the video controls (works in WebKit‑based browsers) st.markdown( """ """, unsafe_allow_html=True, ) # ----------------------------------------------------------------------------- st.title("Live Streaming Space") # ----------------------------------------------------------------------------- # A simple video transformer that “buffers” frames for 3 seconds. # (In a more complex app you might encode these frames into a video segment file.) class BufferingTransformer(VideoTransformerBase): def __init__(self): self.buffer = [] # List to hold frames self.buffer_duration = 3.0 # Buffer duration in seconds self.last_flush_time = time.time() def transform(self, frame): # Convert the frame (a VideoFrame object) into a NumPy array (BGR format) img = frame.to_ndarray(format="bgr24") self.buffer.append(img) now = time.time() if now - self.last_flush_time >= self.buffer_duration: # Here you could (for example) encode/store the buffered frames. # For this demo we just clear the buffer every 3 seconds. self.buffer = [] self.last_flush_time = now # Return the image unchanged for display return img # ----------------------------------------------------------------------------- # Sidebar: Password input for broadcasting. Only if you enter the correct password # ("test123") do you get access to the broadcast controls. password = st.sidebar.text_input("Enter broadcasting password", type="password") if password == "test123": st.sidebar.success("Authenticated for broadcasting!") broadcast_mode = True else: broadcast_mode = False # ----------------------------------------------------------------------------- if broadcast_mode: st.sidebar.header("Broadcast Settings") # (In a real app you might enumerate the actual connected cameras. # Here we simply provide a dummy list of choices.) camera_options = ["Camera 0", "Camera 1", "Camera 2"] camera_choice = st.sidebar.selectbox("Select Camera", camera_options) camera_index = int(camera_choice.split(" ")[-1]) st.write("### You are in **BROADCASTING** mode") st.write("Your live stream is being sent to all viewers.") # Start the WebRTC streamer in SENDONLY mode (broadcasting). # The same key ("live_stream") is used so that viewers join the same “room.” webrtc_ctx = webrtc_streamer( key="live_stream", mode=WebRtcMode.SENDONLY, video_device_index=camera_index, media_stream_constraints={"video": True, "audio": False}, video_transformer_factory=BufferingTransformer, video_html_attrs={ "controls": True, "style": { "width": "100%", "border": "2px solid #ccc", "border-radius": "10px", }, }, rtc_configuration={"iceServers": [{"urls": ["stun:stun.l.google.com:19302"]}]}, ) else: st.write("### Viewing Broadcast") st.info("If you have the broadcasting password, enter it in the sidebar to broadcast your own stream.") # In viewing mode, join the same “room” in RECVONLY mode so you can see the active stream. webrtc_ctx = webrtc_streamer( key="live_stream", mode=WebRtcMode.RECVONLY, media_stream_constraints={"video": True, "audio": False}, video_transformer_factory=BufferingTransformer, video_html_attrs={ "controls": True, "style": { "width": "100%", "border": "2px solid #ccc", "border-radius": "10px", }, }, rtc_configuration={"iceServers": [{"urls": ["stun:stun.l.google.com:19302"]}]}, )