|
| 1 | +# A Quick Example: Live Stream Alert |
| 2 | + |
| 3 | +A quick example to get used to setting up detectors and asking good questions: set up a monitor on a live stream. |
| 4 | + |
| 5 | +## Requirements |
| 6 | + |
| 7 | +- [Groundlight SDK](/docs/installation/) with Python 3.7 or higher |
| 8 | +- The video ID of a YouTube live stream you'd like to monitor |
| 9 | + |
| 10 | +## Installation |
| 11 | + |
| 12 | +Ensure you have Python 3.7 or higher installed, and then install the Groundlight SDK and OpenCV library: |
| 13 | + |
| 14 | +```bash |
| 15 | +pip install groundlight pillow ffmpeg yt-dlp typer |
| 16 | +``` |
| 17 | + |
| 18 | +## Creating the Application |
| 19 | + |
| 20 | +1. Save this command as a shell script `get_latest_frame.sh`: |
| 21 | + |
| 22 | +``` |
| 23 | +#!/bin/bash |
| 24 | +
|
| 25 | +ffmpeg -i "$(yt-dlp -g $1 | head -n 1)" -vframes 1 last.jpg -y |
| 26 | +``` |
| 27 | + |
| 28 | +This will download the most recent frame from a YouTube live stream and save it to a local file `last.jpg`. |
| 29 | + |
| 30 | +2. Log in to the [Groundlight application](https://app.groundlight.ai) and get an [API Token](api-tokens). |
| 31 | + |
| 32 | +3. Next, we'll write the Python script for the application. |
| 33 | + |
| 34 | +```python notest |
| 35 | +import os |
| 36 | +import subprocess |
| 37 | +import typer |
| 38 | +from groundlight import Groundlight |
| 39 | +from PIL import Image |
| 40 | + |
| 41 | + |
| 42 | +def main(*, video_id: str = None, detector_name: str = None, query: str = None, confidence: float = 0.75, wait: int = 60): |
| 43 | + """ |
| 44 | + Run the script to get the stream's last frame as a subprocess, and submit result as an image query to a Groundlight detector |
| 45 | + :param video_id: Video ID of the YouTube live stream (the URLs have the form https://www.youtube.com/watch?v=<VIDEO_ID>) |
| 46 | + :param detector_name: Name for your Groundlight detector |
| 47 | + :param query: Question you want to ask of the stream (we will alert on the answer of NO) |
| 48 | + """ |
| 49 | + gl = Groundlight() |
| 50 | + detector = gl.create_detector(name=detector_name, query=query, confidence_threshold=confidence) |
| 51 | + |
| 52 | + while True: |
| 53 | + p = subprocess.run(["./get_latest_frame.sh", video_id]) |
| 54 | + if p.returncode != 0: |
| 55 | + raise RuntimeError(f"Could not get image from video ID: {video_id}. Process exited with return code {p.returncode}.") |
| 56 | + |
| 57 | + image = Image.open("last.jpg").convert("RGB") |
| 58 | + response = gl.submit_image_query(detector=detector, image=image, wait=wait) |
| 59 | + |
| 60 | + if response.result.label == "NO": |
| 61 | + os.system("say 'Alert!'") # this may not work on all operating systems |
| 62 | + |
| 63 | + |
| 64 | +if __name__ == "__main__": |
| 65 | + typer.run(main) |
| 66 | + |
| 67 | +``` |
| 68 | + |
| 69 | +4. Save the script as `streaming_alert.py` in the same directory as `get_latest_frame.sh` above and run it: |
| 70 | + |
| 71 | +```bash |
| 72 | +python streaming_alert.py <VIDEO_ID> --detector_name <DETECTOR_NAME> --query <QUERY IN QUOTATION MARKS> |
| 73 | +``` |
| 74 | + |
0 commit comments