Files

166 lines
6.0 KiB
Python

"""Standard IP-camera / live-stream protocol catalog.
This is the single source of truth for which streaming standards Cetana
supports. Settings schema entries and Camera.protocol choices are derived
from this registry, so adding a new standard only requires one entry here.
"""
from __future__ import annotations
from typing import Any
# Each protocol describes how streams are addressed and whether a browser can
# play them natively (without a media gateway / WebRTC stack).
STREAM_PROTOCOLS: dict[str, dict[str, Any]] = {
"rtsp": {
"label": "RTSP (Real Time Streaming Protocol)",
"schemes": ("rtsp", "rtsps"),
"default_port": 554,
"default_path": "stream1",
"browser_native": False,
"mime_type": "",
"transports": ("tcp", "udp"),
"auth_methods": ("none", "basic", "digest"),
"url_template": "{scheme}://{host}:{port}/{path}",
"notes": "Most IP cameras. Needs a media gateway (e.g. MediaMTX) for browser playback.",
},
"rtmp": {
"label": "RTMP (Real-Time Messaging Protocol)",
"schemes": ("rtmp", "rtmps"),
"default_port": 1935,
"default_path": "live/stream",
"browser_native": False,
"mime_type": "",
"transports": (),
"auth_methods": ("none", "basic"),
"url_template": "{scheme}://{host}:{port}/{path}",
"notes": "Common for encoders/NVR publish. Needs HLS/WebRTC remux for browsers.",
},
"hls": {
"label": "HLS (HTTP Live Streaming)",
"schemes": ("https", "http"),
"default_port": 443,
"default_path": "live/index.m3u8",
"browser_native": True,
"mime_type": "application/vnd.apple.mpegurl",
"transports": (),
"auth_methods": ("none", "basic", "bearer", "query_token"),
"url_template": "{scheme}://{host}:{port}/{path}",
"notes": "Browser-friendly (.m3u8). Preferred default for web monitoring.",
},
"dash": {
"label": "MPEG-DASH",
"schemes": ("https", "http"),
"default_port": 443,
"default_path": "live/manifest.mpd",
"browser_native": True,
"mime_type": "application/dash+xml",
"transports": (),
"auth_methods": ("none", "basic", "bearer", "query_token"),
"url_template": "{scheme}://{host}:{port}/{path}",
"notes": "Adaptive HTTP streaming via MPD manifest.",
},
"webrtc": {
"label": "WebRTC",
"schemes": ("https", "http", "wss", "ws"),
"default_port": 443,
"default_path": "webrtc/offer",
"browser_native": True,
"mime_type": "",
"transports": ("udp", "tcp"),
"auth_methods": ("none", "bearer", "query_token"),
"url_template": "{scheme}://{host}:{port}/{path}",
"notes": "Low-latency browser playback via a signaling endpoint.",
},
"mjpeg": {
"label": "MJPEG over HTTP",
"schemes": ("https", "http"),
"default_port": 80,
"default_path": "video.mjpg",
"browser_native": True,
"mime_type": "multipart/x-mixed-replace",
"transports": (),
"auth_methods": ("none", "basic", "digest"),
"url_template": "{scheme}://{host}:{port}/{path}",
"notes": "Simple motion-JPEG snapshot stream; plays in <img> or <video> depending on camera.",
},
"http": {
"label": "HTTP Progressive (MP4/WebM)",
"schemes": ("https", "http"),
"default_port": 443,
"default_path": "live.mp4",
"browser_native": True,
"mime_type": "video/mp4",
"transports": (),
"auth_methods": ("none", "basic", "bearer", "query_token"),
"url_template": "{scheme}://{host}:{port}/{path}",
"notes": "Progressive download / pseudo-live file URL.",
},
"onvif": {
"label": "ONVIF Media Profile",
"schemes": ("http", "https"),
"default_port": 80,
"default_path": "onvif/device_service",
"browser_native": False,
"mime_type": "",
"transports": (),
"auth_methods": ("none", "digest", "ws_username_token"),
"url_template": "{scheme}://{host}:{port}/{path}",
"notes": "Device discovery/control. Stream URI is usually resolved to RTSP by the NVR.",
},
"srt": {
"label": "SRT (Secure Reliable Transport)",
"schemes": ("srt",),
"default_port": 9000,
"default_path": "",
"browser_native": False,
"mime_type": "",
"transports": ("udp",),
"auth_methods": ("none", "passphrase"),
"url_template": "{scheme}://{host}:{port}",
"notes": "Low-latency contribution protocol. Needs a gateway for browsers.",
},
}
PROTOCOL_CHOICES = [(key, meta["label"]) for key, meta in STREAM_PROTOCOLS.items()]
PROTOCOL_KEYS = tuple(STREAM_PROTOCOLS.keys())
DEFAULT_PROTOCOL = "hls"
def get_protocol(key: str | None) -> dict[str, Any]:
if key and key in STREAM_PROTOCOLS:
return STREAM_PROTOCOLS[key]
return STREAM_PROTOCOLS[DEFAULT_PROTOCOL]
def protocol_choice_options() -> list[dict[str, str]]:
"""Settings-panel friendly {value, label} options."""
return [{"value": key, "label": meta["label"]} for key, meta in STREAM_PROTOCOLS.items()]
def detect_protocol_from_url(url: str) -> str | None:
"""Best-effort guess from URL scheme / path."""
lower = (url or "").strip().lower()
if not lower:
return None
if lower.startswith("rtsps://") or lower.startswith("rtsp://"):
return "rtsp"
if lower.startswith("rtmps://") or lower.startswith("rtmp://"):
return "rtmp"
if lower.startswith("srt://"):
return "srt"
if ".m3u8" in lower:
return "hls"
if ".mpd" in lower:
return "dash"
if any(token in lower for token in (".mjpg", ".mjpeg", "multipart")):
return "mjpeg"
if "webrtc" in lower or lower.startswith("wss://") or lower.startswith("ws://"):
return "webrtc"
if "onvif" in lower:
return "onvif"
if lower.startswith("http://") or lower.startswith("https://"):
return "http"
return None