Какие знаешь протоколы для обмена видеопотоком?

Протоколы для обмена видеопотоком

Видеоструминг — это сложная область, требующая понимания различных протоколов, кодеков и архитектур. Поделюсь своим опытом.

RTMP (Real-Time Messaging Protocol)

RTMP — один из первых и самых распространённых протоколов для live-трансляций.

import subprocess

# Для захвата видео с веб-камеры и отправки на RTMP сервер
command = [
    'ffmpeg',
    '-f', 'dshow',  # DirectShow для Windows
    '-i', 'video="Webcam"',
    '-c:v', 'libx264',
    '-preset', 'veryfast',
    '-b:v', '2500k',
    '-f', 'flv',
    'rtmp://localhost/live/stream'
]

process = subprocess.Popen(command)

Преимущества:

• Низкая задержка (1-2 сек)
• Надёжная доставка
• Поддержка на большинстве платформ

Недостатки:

• TCP-based (медленнее в неидеальных сетях)
• Требует особого оборудования
• Устаревает (Adobe перестал поддерживать Flash)

HLS (HTTP Live Streaming)

HLS — протокол на основе HTTP, стандарт Apple.

from pathlib import Path
import subprocess

# Создание HLS плейлиста из видеофайла
def create_hls_stream(input_file: str, output_dir: str):
    command = [
        'ffmpeg',
        '-i', input_file,
        '-c:v', 'h264',
        '-c:a', 'aac',
        '-hls_time', '10',  # 10 секунд на сегмент
        '-hls_playlist_type', 'event',  # или 'vod' для VOD
        '-hls_segment_filename', f'{output_dir}/segment_%03d.ts',
        f'{output_dir}/stream.m3u8'
    ]
    
    subprocess.run(command)

# Структура HLS плейлиста:
# stream.m3u8:
# #EXTM3U
# #EXT-X-VERSION:3
# #EXT-X-TARGETDURATION:10
# #EXTINF:10.0,
# segment_000.ts
# #EXTINF:10.0,
# segment_001.ts

Преимущества:

• HTTP — везде работает
• Адаптивный битрейт (ABR)
• Кроссплатформенность
• Хорошая поддержка на mobile

Недостатки:

• Задержка 20-30 секунд (не для live)
• Требует нарезки на сегменты

DASH (Dynamic Adaptive Streaming over HTTP)

DASH — открытый стандарт, аналог HLS.

# DASH генерирует MPD (Media Presentation Description) — XML файл
# Структура:
# <MPD>
#   <Period>
#     <AdaptationSet mimeType="video/mp4">
#       <Representation id="480p" width="854" height="480" bandwidth="1000000">
#         <BaseURL>stream_480p.mp4</BaseURL>
#       </Representation>
#       <Representation id="720p" width="1280" height="720" bandwidth="2500000">
#         <BaseURL>stream_720p.mp4</BaseURL>
#       </Representation>
#     </AdaptationSet>
#   </Period>
# </MPD>

from lxml import etree

def create_dash_mpd(video_files: dict) -> str:
    """video_files: {resolution: bandwidth}"""
    mpd = etree.Element("MPD")
    mpd.set("xmlns", "urn:mpeg:dash:schema:mpd:2011")
    
    period = etree.SubElement(mpd, "Period")
    adaptation_set = etree.SubElement(period, "AdaptationSet")
    adaptation_set.set("mimeType", "video/mp4")
    
    for resolution, bandwidth in video_files.items():
        representation = etree.SubElement(adaptation_set, "Representation")
        representation.set("id", resolution)
        representation.set("bandwidth", str(bandwidth))
        
        base_url = etree.SubElement(representation, "BaseURL")
        base_url.text = f"stream_{resolution}.mp4"
    
    return etree.tostring(mpd, pretty_print=True, encoding='utf-8').decode()

WebRTC (Web Real-Time Communication)

WebRTC — для low-latency live streaming (1-2 sec).

import asyncio
from aiortc import RTCPeerConnection, RTCSessionDescription, VideoStreamTrack
from aiortc.contrib.media import MediaPlayer, MediaRecorder

# Простой WebRTC сервер на Python
from aiohttp import web
import json

pc_list = []

async def handle_offer(request):
    params = await request.json()
    offer = RTCSessionDescription(sdp=params["sdp"], type=params["type"])
    
    pc = RTCPeerConnection()
    pc_list.append(pc)
    
    # Добавляем видеотрек
    player = MediaPlayer('video.mp4')
    pc.addTrack(player.video)
    
    # Обрабатываем offer
    await pc.setRemoteDescription(offer)
    
    # Создаём answer
    answer = await pc.createAnswer()
    await pc.setLocalDescription(answer)
    
    return web.json_response({
        "sdp": pc.localDescription.sdp,
        "type": pc.localDescription.type
    })

app = web.Application()
app.router.post('/offer', handle_offer)
web.run_app(app, port=8080)

Преимущества:

• Очень низкая задержка (P2P)
• Безопасность (зашифрованный)
• Не требует плагинов

Недостатки:

• Сложнее в реализации
• NAT traversal проблемы
• Требует сигнализации (signaling server)

RTSP (Real Time Streaming Protocol)

RTSP — старый протокол для камер и сетевых устройств.

import subprocess

# Захват RTSP потока с IP камеры и преобразование в HLS
def convert_rtsp_to_hls(rtsp_url: str, output_dir: str):
    command = [
        'ffmpeg',
        '-i', rtsp_url,
        '-c:v', 'h264',
        '-c:a', 'aac',
        '-hls_time', '10',
        '-hls_list_size', '3',  # Только 3 последних сегмента (для live)
        '-hls_flags', 'delete_segments',  # Удаляем старые сегменты
        f'{output_dir}/stream.m3u8'
    ]
    
    subprocess.Popen(command)

Где используется:

• IP камеры видеонаблюдения
• Системы видеоконференций

Adaptive Bitrate Streaming (ABR)

Динамическое изменение качества в зависимости от интернета.

from enum import Enum

class BitRate(Enum):
    P240 = 500_000
    P360 = 1_000_000
    P480 = 2_500_000
    P720 = 5_000_000
    P1080 = 8_000_000

class AdaptiveBitrateController:
    def __init__(self):
        self.current_bitrate = BitRate.P720
        self.buffered_segments = 3
    
    def adjust_bitrate(self, network_bandwidth: int, buffer_health: float):
        """
        network_bandwidth: измеренная пропускная способность сети
        buffer_health: количество буферизированных сегментов (0-10)
        """
        # Если буфер пуст — срочно снижаем качество
        if buffer_health < 1:
            self.current_bitrate = BitRate.P360
            return
        
        # Если буфер полный и пропускная способность хорошая — повышаем
        if buffer_health > 5 and network_bandwidth > BitRate.P1080.value:
            self.current_bitrate = BitRate.P1080
            return
        
        # Адаптируемся к сети
        for bitrate in sorted(BitRate, key=lambda x: x.value):
            if network_bandwidth > bitrate.value * 1.5:  # +50% на буфер
                self.current_bitrate = bitrate
        
        return self.current_bitrate

Сравнение протоколов

Практический пример: Live streaming сервис

from fastapi import FastAPI, WebSocket
from typing import Set
import asyncio
import subprocess

app = FastAPI()

class StreamManager:
    def __init__(self):
        self.active_streams: dict = {}
        self.viewers: dict[str, Set[WebSocket]] = {}
    
    async def start_stream(self, stream_id: str, rtmp_url: str):
        """Начинаем захват потока и конвертируем в HLS"""
        command = [
            'ffmpeg',
            '-i', rtmp_url,
            '-c:v', 'h264',
            '-preset', 'veryfast',
            '-c:a', 'aac',
            '-hls_time', '2',
            '-hls_list_size', '5',
            f'/tmp/streams/{stream_id}/stream.m3u8'
        ]
        
        process = await asyncio.create_subprocess_exec(*command)
        self.active_streams[stream_id] = process
    
    async def stop_stream(self, stream_id: str):
        """Останавливаем трансляцию"""
        process = self.active_streams.pop(stream_id, None)
        if process:
            process.terminate()
            await process.wait()

manager = StreamManager()

@app.post("/stream/{stream_id}/start")
async def start_stream(stream_id: str, rtmp_url: str):
    await manager.start_stream(stream_id, rtmp_url)
    return {"status": "streaming", "stream_id": stream_id}

@app.get("/stream/{stream_id}")
async def get_stream(stream_id: str):
    """Возвращаем HLS плейлист"""
    with open(f'/tmp/streams/{stream_id}/stream.m3u8', 'r') as f:
        return f.read()

Выбор протокола

• Live с низкой задержкой: WebRTC или RTMP
• Live с приемлемой задержкой: HLS или DASH
• VOD (фильмы, YouTube-style): HLS или DASH
• IP камеры: RTSP
• Кроссплатформа: HLS (везде работает)

В современной разработке чаще всего используется комбинация: захватываем через RTMP, конвертируем в HLS/DASH для клиентов.