streaming-patterns

What this skill does

# Live Streaming Patterns

Best practices for live streaming to YouTube, Twitch, and other platforms.

## Platform Configuration

### YouTube Live

```python
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials

def create_youtube_broadcast(
    credentials: Credentials,
    title: str,
    description: str,
    scheduled_start: str,
    privacy: str = "unlisted"
):
    """Create a YouTube live broadcast."""
    youtube = build('youtube', 'v3', credentials=credentials)

    # Create broadcast
    broadcast = youtube.liveBroadcasts().insert(
        part="snippet,status,contentDetails",
        body={
            "snippet": {
                "title": title,
                "description": description,
                "scheduledStartTime": scheduled_start
            },
            "status": {
                "privacyStatus": privacy,
                "selfDeclaredMadeForKids": False
            },
            "contentDetails": {
                "enableAutoStart": True,
                "enableAutoStop": True,
                "enableDvr": True,
                "recordFromStart": True
            }
        }
    ).execute()

    # Create stream
    stream = youtube.liveStreams().insert(
        part="snippet,cdn",
        body={
            "snippet": {
                "title": f"Stream for {title}"
            },
            "cdn": {
                "frameRate": "60fps",
                "ingestionType": "rtmp",
                "resolution": "1080p"
            }
        }
    ).execute()

    # Bind stream to broadcast
    youtube.liveBroadcasts().bind(
        part="id,contentDetails",
        id=broadcast['id'],
        streamId=stream['id']
    ).execute()

    return {
        "broadcast_id": broadcast['id'],
        "stream_key": stream['cdn']['ingestionInfo']['streamName'],
        "rtmp_url": stream['cdn']['ingestionInfo']['ingestionAddress'],
        "watch_url": f"https://youtube.com/watch?v={broadcast['id']}"
    }


def transition_broadcast(credentials: Credentials, broadcast_id: str, status: str):
    """Transition broadcast status: testing, live, complete."""
    youtube = build('youtube', 'v3', credentials=credentials)

    return youtube.liveBroadcasts().transition(
        broadcastStatus=status,
        id=broadcast_id,
        part="status"
    ).execute()
```

### Twitch

```python
import requests

class TwitchAPI:
    def __init__(self, client_id: str, access_token: str):
        self.client_id = client_id
        self.access_token = access_token
        self.base_url = "https://api.twitch.tv/helix"
        self.headers = {
            "Client-ID": client_id,
            "Authorization": f"Bearer {access_token}"
        }

    def get_stream_key(self, broadcaster_id: str) -> str:
        """Get stream key for broadcaster."""
        response = requests.get(
            f"{self.base_url}/streams/key",
            headers=self.headers,
            params={"broadcaster_id": broadcaster_id}
        )
        return response.json()['data'][0]['stream_key']

    def update_stream_info(
        self,
        broadcaster_id: str,
        title: str,
        game_id: str = None,
        language: str = "en"
    ):
        """Update stream title and category."""
        data = {
            "broadcaster_id": broadcaster_id,
            "title": title,
            "broadcaster_language": language
        }
        if game_id:
            data["game_id"] = game_id

        return requests.patch(
            f"{self.base_url}/channels",
            headers=self.headers,
            json=data
        )

    def get_stream_status(self, user_login: str) -> dict:
        """Check if channel is live."""
        response = requests.get(
            f"{self.base_url}/streams",
            headers=self.headers,
            params={"user_login": user_login}
        )
        data = response.json()['data']
        return data[0] if data else None

    def create_clip(self, broadcaster_id: str) -> dict:
        """Create clip from live stream."""
        response = requests.post(
            f"{self.base_url}/clips",
            headers=self.headers,
            params={"broadcaster_id": broadcaster_id}
        )
        return response.json()['data'][0]
```

## RTMP Streaming

### FFmpeg RTMP Push

```bash
# Stream to YouTube
ffmpeg -re -i input.mp4 \
    -c:v libx264 -preset veryfast -maxrate 4500k -bufsize 9000k \
    -pix_fmt yuv420p -g 60 \
    -c:a aac -b:a 160k -ar 44100 \
    -f flv "rtmp://a.rtmp.youtube.com/live2/YOUR_STREAM_KEY"

# Stream to Twitch
ffmpeg -re -i input.mp4 \
    -c:v libx264 -preset veryfast -maxrate 6000k -bufsize 12000k \
    -pix_fmt yuv420p -g 60 \
    -c:a aac -b:a 160k -ar 44100 \
    -f flv "rtmp://live.twitch.tv/app/YOUR_STREAM_KEY"

# Stream desktop (macOS)
ffmpeg -f avfoundation -framerate 30 -i "1:0" \
    -c:v libx264 -preset ultrafast -tune zerolatency \
    -c:a aac -b:a 128k \
    -f flv "rtmp://destination/stream_key"

# Stream desktop (Linux)
ffmpeg -f x11grab -framerate 30 -video_size 1920x1080 -i :0.0 \
    -f pulse -i default \
    -c:v libx264 -preset ultrafast -tune zerolatency \
    -c:a aac -b:a 128k \
    -f flv "rtmp://destination/stream_key"
```

### Multi-Platform Streaming

```bash
# Using tee muxer to stream to multiple platforms
ffmpeg -re -i input.mp4 \
    -c:v libx264 -preset veryfast -b:v 4500k \
    -c:a aac -b:a 160k \
    -f tee -map 0:v -map 0:a \
    "[f=flv]rtmp://a.rtmp.youtube.com/live2/YT_KEY|\
     [f=flv]rtmp://live.twitch.tv/app/TWITCH_KEY|\
     [f=flv]rtmp://live-api-s.facebook.com:443/rtmp/FB_KEY"
```

### Python RTMP Handler

```python
import subprocess
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class StreamDestination:
    name: str
    rtmp_url: str
    stream_key: str

    @property
    def full_url(self) -> str:
        return f"{self.rtmp_url}/{self.stream_key}"


class MultiStreamer:
    def __init__(
        self,
        input_source: str,
        destinations: List[StreamDestination],
        video_bitrate: str = "4500k",
        audio_bitrate: str = "160k"
    ):
        self.input_source = input_source
        self.destinations = destinations
        self.video_bitrate = video_bitrate
        self.audio_bitrate = audio_bitrate
        self.process: Optional[subprocess.Popen] = None

    def build_command(self) -> List[str]:
        """Build FFmpeg command for multi-platform streaming."""
        cmd = [
            "ffmpeg",
            "-re", "-i", self.input_source,
            "-c:v", "libx264",
            "-preset", "veryfast",
            "-b:v", self.video_bitrate,
            "-maxrate", self.video_bitrate,
            "-bufsize", str(int(self.video_bitrate[:-1]) * 2) + "k",
            "-pix_fmt", "yuv420p",
            "-g", "60",
            "-c:a", "aac",
            "-b:a", self.audio_bitrate,
            "-ar", "44100"
        ]

        if len(self.destinations) == 1:
            cmd.extend(["-f", "flv", self.destinations[0].full_url])
        else:
            # Use tee muxer for multiple destinations
            tee_outputs = "|".join(
                f"[f=flv]{dest.full_url}" for dest in self.destinations
            )
            cmd.extend([
                "-f", "tee",
                "-map", "0:v", "-map", "0:a",
                tee_outputs
            ])

        return cmd

    def start(self):
        """Start streaming."""
        cmd = self.build_command()
        self.process = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

    def stop(self):
        """Stop streaming."""
        if self.process:
            self.process.terminate()
            self.process.wait()
```

## OBS WebSocket Integration

```python
import obswebsocket
from obswebsocket import obsws, requests as obs_requests

class OBSController:
    def __init__(self, host: str = "localhost", port: int = 4455, password: str = ""):