live_digest.py

"""
Trio API - Stream Summarization

This example shows how to generate a narrative summary of what's
happening on a live stream over a time window.

Usage:
    export TRIO_API_KEY="your_api_key"
    python live_digest.py

The API captures frames at intervals and generates a cohesive
summary of the activity.

Note:
    This example uses SSE (Server-Sent Events) streaming to receive
    summaries in real-time. For webhook-based delivery, provide a
    webhook_url parameter.
"""

import os
import requests

TRIO_API = "https://trio.machinefi.com/api"
API_KEY = os.environ.get("TRIO_API_KEY", "YOUR_API_KEY")

# Example: Summarize activity on a traffic cam
STREAM_URL = "https://www.youtube.com/watch?v=jfKfPfyJRdk"


def start_live_digest_sse(
    stream_url: str,
    window_minutes: int = 5,
    capture_interval_seconds: int = 30
):
    """
    Start a summarization job and stream results via SSE.

    Args:
        stream_url: YouTube Live URL to summarize
        window_minutes: Duration of summary window (1-60)
        capture_interval_seconds: How often to capture frames (10-300)

    Yields:
        SSE events with progress and summary data
    """
    response = requests.post(
        f"{TRIO_API}/live-digest",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
            "Accept": "text/event-stream"
        },
        json={
            "stream_url": stream_url,
            "window_minutes": window_minutes,
            "capture_interval_seconds": capture_interval_seconds
        },
        stream=True
    )
    response.raise_for_status()

    # Parse SSE events
    for line in response.iter_lines():
        if line:
            line = line.decode('utf-8')
            if line.startswith('data: '):
                yield line[6:]  # Remove 'data: ' prefix


def start_live_digest_webhook(
    stream_url: str,
    webhook_url: str,
    window_minutes: int = 5,
    capture_interval_seconds: int = 30
) -> dict:
    """
    Start a summarization job with webhook delivery.

    Args:
        stream_url: YouTube Live URL to summarize
        webhook_url: URL to receive the summary
        window_minutes: Duration of summary window (1-60)
        capture_interval_seconds: How often to capture frames (10-300)

    Returns:
        API response with job_id
    """
    response = requests.post(
        f"{TRIO_API}/live-digest",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "stream_url": stream_url,
            "webhook_url": webhook_url,
            "window_minutes": window_minutes,
            "capture_interval_seconds": capture_interval_seconds
        }
    )
    response.raise_for_status()
    return response.json()


def get_job_status(job_id: str) -> dict:
    """Check the status of a job."""
    response = requests.get(
        f"{TRIO_API}/jobs/{job_id}",
        headers={"Authorization": f"Bearer {API_KEY}"}
    )
    response.raise_for_status()
    return response.json()


if __name__ == "__main__":
    import json

    print(f"Starting stream digest (SSE mode)...")
    print(f"Stream: {STREAM_URL}")
    print(f"Window: 5 minutes")
    print(f"Capture interval: 30 seconds")
    print()

    print("Waiting for summary events...")
    print("-" * 50)

    for event_data in start_live_digest_sse(
        stream_url=STREAM_URL,
        window_minutes=5,
        capture_interval_seconds=30
    ):
        try:
            event = json.loads(event_data)
            if "summary" in event:
                print(f"\nSummary: {event['summary']}")
            elif "frame_count" in event:
                print(f"Progress: {event['frame_count']} frames captured")
            elif "job_id" in event:
                print(f"Job started: {event['job_id']}")
        except json.JSONDecodeError:
            print(f"Event: {event_data}")