Spaces:
Sleeping
Sleeping
File size: 1,615 Bytes
2c0045a 78b6d28 2c0045a 34474e7 2c0045a 78b6d28 34474e7 78b6d28 ed3c8a2 6e17fd0 78b6d28 2c0045a 78b6d28 2c0045a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 |
import os
import streamlit as st
from pathlib import Path
from landingai.predict import Predictor
from landingai.vision_pipeline import NetworkedCamera, FrameSet
from landingai.st_utils import render_svg
VIDEO_CACHE_PATH = Path("cached_data")
VIDEO_CACHE_PATH.mkdir(exist_ok=True, parents=True)
VIDEO_CACHE_PATH = VIDEO_CACHE_PATH / "latest.mp4"
VIDEO_LEN_SEC = 10
FPS = 2
PLAYLIST_URL = (
"https://live.hdontap.com/hls/hosb1/topanga_swellmagnet.stream/playlist.m3u8"
)
API_KEY = os.environ["API_KEY"]
ENDPOINT_ID = os.environ["ENDPOINT_ID"]
render_svg(Path("./assets/landing-logo.svg").read_text())
st.title("Topanga Beach Surfer Counter")
st.write(
"This application will grab the latest 10s clip of surfers from the Topanga Beach surf cam "
"and count the number of surfers there. It uses a model built with LandingLens to detect "
"the surfers. You can find out more at landing.ai"
)
def get_latest_surfer_count():
vid_src = NetworkedCamera(PLAYLIST_URL, fps=FPS)
surfer_model = Predictor(ENDPOINT_ID, api_key=API_KEY)
frs = FrameSet()
for i, frame in enumerate(vid_src):
if i >= VIDEO_LEN_SEC * FPS:
break
frs.extend(frame.run_predict(predictor=surfer_model).overlay_predictions())
frs.save_video(str(VIDEO_CACHE_PATH), video_fps=FPS, image_src="overlay")
surfers = frs.get_class_counts()["surfer"] / (VIDEO_LEN_SEC * FPS)
st.video(open(VIDEO_CACHE_PATH, "rb").read())
st.write(f"Surfer count: **{surfers}**")
st.title("Surfer Counter")
button = st.button("Get Topanga Beach Surfer Count", on_click=get_latest_surfer_count)
|