app.py

import os
import streamlit as st

from pathlib import Path
from landingai.predict import Predictor
from landingai.pipeline.image_source import NetworkedCamera, FrameSet
from landingai.pipeline.postprocessing import get_class_counts
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 build your own model at [landing.ai](https://landing.ai/) or run the "
    "code yourself by getting it from our [github page](https://github.com/landing-ai/landingai-python/tree/main/examples/apps/surfer-count)."
)


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)
    frs.run_predict(predictor=surfer_model).overlay_predictions()
    frs.save_video(str(VIDEO_CACHE_PATH), video_fps=FPS, image_src="overlay")
    counts = get_class_counts(frs)
    if "surfer" in counts:
        surfers = counts["surfer"] / (VIDEO_LEN_SEC * FPS)
    else:
        surfers = 0
    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)