diff --git a/cereal/log.capnp b/cereal/log.capnp index cf7fc540c6..0fc2635afc 100644 --- a/cereal/log.capnp +++ b/cereal/log.capnp @@ -2604,6 +2604,8 @@ struct Event { livestreamRoadEncodeData @120 :EncodeData; livestreamWideRoadEncodeData @121 :EncodeData; livestreamDriverEncodeData @122 :EncodeData; + youtubeRoadEncodeData @152 :EncodeData; + youtubeRoadEncodeIdx @153 :EncodeIndex; # *********** Custom: reserved for forks *********** diff --git a/cereal/services.py b/cereal/services.py index f1e8c1b861..27f77cf256 100644 --- a/cereal/services.py +++ b/cereal/services.py @@ -107,6 +107,7 @@ def __init__(self, should_log: bool, frequency: float, decimation: Optional[int] "livestreamWideRoadEncodeData": (False, 20., None, QueueSize.MEDIUM), "livestreamRoadEncodeData": (False, 20., None, QueueSize.MEDIUM), "livestreamDriverEncodeData": (False, 20., None, QueueSize.MEDIUM), + "youtubeRoadEncodeData": (False, 20., None, QueueSize.BIG), "customReservedRawData0": (True, 0.), "customReservedRawData1": (True, 0.), "customReservedRawData2": (True, 0.), diff --git a/common/params_keys.h b/common/params_keys.h index df8737acca..65edf6f6e9 100644 --- a/common/params_keys.h +++ b/common/params_keys.h @@ -184,6 +184,9 @@ inline static std::unordered_map keys = { {"ClusterHudRadarDisplay", {PERSISTENT, INT, "0"}}, {"ClusterHudRadarSourceColor", {PERSISTENT, INT, "0"}}, {"RecordRoadCam", {PERSISTENT, INT, "0"}}, + {"CarrotYouTubeLive", {PERSISTENT, INT, "0"}}, + {"CarrotYouTubeQuality", {PERSISTENT, INT, "0"}}, + {"CarrotYouTubeTimestamp", {PERSISTENT, INT, "0"}}, {"HDPuse", {PERSISTENT, INT, "0"}}, {"AutoCruiseControl", {PERSISTENT, INT, "0"}}, diff --git a/selfdrive/carrot/server/config.py b/selfdrive/carrot/server/config.py index 0271a54291..4c4e28acb8 100644 --- a/selfdrive/carrot/server/config.py +++ b/selfdrive/carrot/server/config.py @@ -26,6 +26,8 @@ CARROT_WEB_SETTINGS_PATH = os.path.join(CARROT_STATE_DIR, "web_settings.json") CARROT_SETTING_FAVORITES_PATH = os.path.join(CARROT_STATE_DIR, "setting_favorites.json") CARROT_SETTING_PROFILES_PATH = os.path.join(CARROT_STATE_DIR, "setting_profiles.json") +CARROT_YOUTUBE_LIVE_STATE_PATH = os.path.join(CARROT_STATE_DIR, "youtube_live.json") +CARROT_YOUTUBE_LIVE_SECRET_PATH = os.path.join(CARROT_STATE_DIR, "youtube_live_secret.json") # Dashcam DASHCAM_ROOT = "/data/media/0/realdata" diff --git a/selfdrive/carrot/server/features/__init__.py b/selfdrive/carrot/server/features/__init__.py index 0f83b705f4..3c1cec5ddd 100644 --- a/selfdrive/carrot/server/features/__init__.py +++ b/selfdrive/carrot/server/features/__init__.py @@ -19,6 +19,7 @@ vision_test, web_settings, ws, + youtube_live, ) @@ -39,5 +40,6 @@ def register_all(app: web.Application) -> None: dashcam.register(app) screenrecord.register(app) tools.register(app) + youtube_live.register(app) vision_test.register(app) vision_diag.register(app) diff --git a/selfdrive/carrot/server/features/youtube_live.py b/selfdrive/carrot/server/features/youtube_live.py new file mode 100644 index 0000000000..3842a4d5f9 --- /dev/null +++ b/selfdrive/carrot/server/features/youtube_live.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import asyncio + +from aiohttp import web + +from ..services.youtube_live import YouTubeLiveService + + +YOUTUBE_LIVE_APP_KEY = "youtube_live_service" + + +def _service(request: web.Request) -> YouTubeLiveService: + service = request.app.get(YOUTUBE_LIVE_APP_KEY) + if not isinstance(service, YouTubeLiveService): + raise web.HTTPServiceUnavailable(text="youtube live service unavailable") + return service + + +async def youtube_live_context(app: web.Application): + service = YouTubeLiveService() + app[YOUTUBE_LIVE_APP_KEY] = service + await service.start() + try: + yield + finally: + await service.stop() + app.pop(YOUTUBE_LIVE_APP_KEY, None) + + +async def api_status(request: web.Request) -> web.Response: + return web.json_response({"ok": True, **_service(request).status()}) + + +async def api_set_stream_key(request: web.Request) -> web.Response: + try: + body = await request.json() + except Exception: + return web.json_response({"ok": False, "error": "invalid json"}, status=400) + key = body.get("stream_key", body.get("key", "")) + try: + status = _service(request).set_stream_key(str(key or "")) + except ValueError as exc: + return web.json_response({"ok": False, "error": str(exc)}, status=400) + return web.json_response({"ok": True, **status}) + + +async def api_get_stream_key(request: web.Request) -> web.Response: + stream_key = _service(request).get_stream_key() + return web.json_response({"ok": True, "configured": bool(stream_key), "stream_key": stream_key}) + + +async def api_clear_stream_key(request: web.Request) -> web.Response: + return web.json_response({"ok": True, **_service(request).clear_stream_key()}) + + +async def api_test(request: web.Request) -> web.Response: + result = await asyncio.to_thread(_service(request).test_config) + return web.json_response(result, status=200 if result.get("ok") else 409) + + +async def api_validate_stream_key(request: web.Request) -> web.Response: + body = {} + if request.can_read_body: + try: + body = await request.json() + except Exception: + return web.json_response({"ok": False, "error": "invalid json"}, status=400) + key = body.get("stream_key", body.get("key")) if isinstance(body, dict) else None + result = await asyncio.to_thread(_service(request).validate_stream_key, key) + return web.json_response(result, status=200 if result.get("ok") else 409) + + +async def api_diagnostics(request: web.Request) -> web.Response: + return web.json_response({"ok": True, "diagnostics": _service(request).diagnostics()}) + + +def register(app: web.Application) -> None: + app.cleanup_ctx.append(youtube_live_context) + app.router.add_get("/api/youtube_live/status", api_status) + app.router.add_get("/api/youtube_live/diagnostics", api_diagnostics) + app.router.add_get("/api/youtube_live/stream_key", api_get_stream_key) + app.router.add_post("/api/youtube_live/stream_key", api_set_stream_key) + app.router.add_post("/api/youtube_live/stream_key/validate", api_validate_stream_key) + app.router.add_delete("/api/youtube_live/stream_key", api_clear_stream_key) + app.router.add_post("/api/youtube_live/test", api_test) diff --git a/selfdrive/carrot/server/services/youtube_live.py b/selfdrive/carrot/server/services/youtube_live.py new file mode 100644 index 0000000000..eca92b7634 --- /dev/null +++ b/selfdrive/carrot/server/services/youtube_live.py @@ -0,0 +1,782 @@ +from __future__ import annotations + +import asyncio +import json +import os +import re +import socket +import ssl +import time +from collections import deque +from pathlib import Path +from typing import Any + +from ..config import CARROT_YOUTUBE_LIVE_SECRET_PATH, CARROT_YOUTUBE_LIVE_STATE_PATH +from .youtube_live_captions import Cea608TimestampInjector +from .youtube_live_muxer import H264FlvMuxer, pyav_capabilities +from .youtube_live_transport import LibrtmpClient, RtmpSink, librtmp_capabilities + + +YOUTUBE_LIVE_PARAM = "CarrotYouTubeLive" +YOUTUBE_QUALITY_PARAM = "CarrotYouTubeQuality" +YOUTUBE_TIMESTAMP_PARAM = "CarrotYouTubeTimestamp" +# The high-quality source has a dedicated encoder so Carrot Vision's shared +# livestream resolution and bitrate remain unchanged. +SOURCE_BY_QUALITY = { + 0: "qRoadEncodeData", + 1: "livestreamRoadEncodeData", + 2: "youtubeRoadEncodeData", + 3: "livestreamWideRoadEncodeData", +} +QUALITY_LABELS = {0: "low", 1: "medium", 2: "high", 3: "wide"} +YOUTUBE_RTMPS_BASE = "rtmps://a.rtmps.youtube.com:443/live2" +YOUTUBE_RTMPS_HOST = "a.rtmps.youtube.com" +YOUTUBE_RTMPS_PORT = 443 +NO_FRAME_STOP_SECONDS = 8.0 +START_BACKOFF_SECONDS = 5.0 +STATUS_WRITE_MIN_INTERVAL = 1.0 +BACKOFF_BASE_SECONDS = 3.0 +BACKOFF_MAX_SECONDS = 60.0 +STREAM_STABLE_SECONDS = 10.0 +MIN_RECONNECT_INTERVAL_SECONDS = 3.0 +EVENT_LOG_MAX = 50 +PROC_CACHE_SECONDS = 5.0 + +_PROCESS_MATCHES = { + "carrot_cluster": "selfdrive.carrot.cluster_autorun", + "webrtcd": "system.webrtc.webrtcd", + "stream_encoderd": "encoderd\x00--stream", + "youtube_encoderd": "encoderd\x00--youtube", +} + + +def _now() -> float: + return time.time() + + +def _mono() -> float: + # Monotonic clock for durations/intervals — immune to wall-clock jumps + # (comma syncs system time from GPS/NTP after boot, which can step the clock). + return time.monotonic() + + +def _read_json(path: Path) -> dict[str, Any]: + try: + raw = json.loads(path.read_text(encoding="utf-8")) + return raw if isinstance(raw, dict) else {} + except Exception: + return {} + + +def _write_json_atomic(path: Path, payload: dict[str, Any], *, mode: int | None = None) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temp_path = path.with_suffix(path.suffix + ".tmp") + temp_path.write_text(json.dumps(payload, sort_keys=True), encoding="utf-8") + if mode is not None: + try: + os.chmod(temp_path, mode) + except OSError: + pass + temp_path.replace(path) + if mode is not None: + try: + os.chmod(path, mode) + except OSError: + pass + + +def _mask_stream_key(stream_key: str) -> str: + key = stream_key.strip() + if not key: + return "" + if len(key) <= 8: + return "*" * len(key) + return f"{key[:4]}...{key[-4:]}" + + +def _extract_stream_key(value: str) -> str: + raw = str(value or "").strip() + if not raw: + return "" + if raw.startswith("rtmp://") or raw.startswith("rtmps://"): + return raw.rstrip("/").rsplit("/", 1)[-1].strip() + return raw + + +def _validate_stream_key_format(stream_key: str) -> tuple[bool, str]: + key = stream_key.strip() + if not key: + return False, "stream key is required" + if len(key) < 8: + return False, "stream key is too short" + if len(key) > 256: + return False, "stream key is too long" + if re.search(r"\s", key): + return False, "stream key must not contain spaces" + if not re.fullmatch(r"[A-Za-z0-9._/-]+", key): + return False, "stream key contains unsupported characters" + return True, "format looks valid" + + +def _check_rtmps_reachable(timeout: float = 2.5) -> tuple[bool, str]: + try: + context = ssl.create_default_context() + with socket.create_connection((YOUTUBE_RTMPS_HOST, YOUTUBE_RTMPS_PORT), timeout=timeout) as sock: + with context.wrap_socket(sock, server_hostname=YOUTUBE_RTMPS_HOST): + return True, "YouTube RTMPS ingest is reachable" + except Exception as exc: + return False, f"YouTube RTMPS ingest unreachable: {exc}" + + +class YouTubeLiveService: + def __init__( + self, + *, + state_path: str = CARROT_YOUTUBE_LIVE_STATE_PATH, + secret_path: str = CARROT_YOUTUBE_LIVE_SECRET_PATH, + ) -> None: + self.state_path = Path(state_path) + self.secret_path = Path(secret_path) + self._task: asyncio.Task | None = None + self._stop_event = asyncio.Event() + self._transport: LibrtmpClient | None = None + self._transport_connected = False + self._muxer: H264FlvMuxer | None = None + self._caption_injector = Cea608TimestampInjector() + self._messaging: Any | None = None + self._socket: Any | None = None + self._socket_source = "" + self._active_source = "" + self._params: Any | None = None + self._last_status_write = 0.0 + self._started_at = 0.0 + self._started_mono = 0.0 + self._last_frame_at = 0.0 + self._last_frame_mono = 0.0 + self._last_frame_id: int | None = None + self._frame_width = 526 + self._frame_height = 330 + self._frame_fps = 20 + self._bytes_sent = 0 + self._session_started_bytes = 0 + self._restart_count = 0 + self._consecutive_failures = 0 + self._next_retry_mono = 0.0 + self._last_error = "" + self._state = "disabled" + self._events: deque[dict[str, Any]] = deque(maxlen=EVENT_LOG_MAX) + self._last_start_mono = 0.0 + self._proc_cache: dict[str, Any] | None = None + self._proc_cache_mono = 0.0 + self._muxer_capabilities = pyav_capabilities() + self._transport_capabilities = librtmp_capabilities() + persisted = _read_json(self.state_path) + try: + self._bytes_sent = int(persisted.get("bytes_sent") or 0) + self._restart_count = int(persisted.get("restart_count") or 0) + except Exception: + pass + + async def start(self) -> None: + if self._task is not None and not self._task.done(): + return + self._stop_event.clear() + self._task = asyncio.create_task(self._run(), name="carrot-youtube-live") + + async def stop(self) -> None: + self._stop_event.set() + task = self._task + if task is not None: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + await self._stop_stream() + self._task = None + self._write_status(force=True) + + def status(self) -> dict[str, Any]: + stream_key = self.get_stream_key() + running = self._transport_connected + now = _now() + mono = _mono() + uptime = int(mono - self._started_mono) if self._started_mono and running else 0 + elapsed = max(1.0, mono - self._started_mono) if self._started_mono and running else 1.0 + session_bytes = max(0, self._bytes_sent - self._session_started_bytes) if running else 0 + last_frame_age_ms = int((mono - self._last_frame_mono) * 1000) if self._last_frame_mono else None + retry_in_sec = max(0, int(self._next_retry_mono - mono)) if self._next_retry_mono else 0 + resource_status = self._resource_status() + warnings = self._warnings(resource_status) + return { + "state": self._state, + "enabled": self._param_enabled(), + "configured": bool(stream_key), + "masked_key": _mask_stream_key(stream_key), + "source": self._current_source(), + "muxer": "pyav-flv-aac", + "muxer_available": bool( + self._muxer_capabilities.get("available") + and self._muxer_capabilities.get("flv") + and self._muxer_capabilities.get("h264") + and self._muxer_capabilities.get("aac") + ), + "transport": "librtmp-rtmps", + "transport_available": bool(self._transport_capabilities.get("available")), + "quality": self._quality_label(), + "requested_quality": self._param_int(YOUTUBE_QUALITY_PARAM, 0), + "timestamp_caption_enabled": self._param_bool(YOUTUBE_TIMESTAMP_PARAM), + "timestamp_caption_mode": "cea608-sei", + "timestamp_caption_packets": self._caption_injector.packets_injected, + "phase": 1, + "running": running, + "pid": None, + "started_at": self._started_at if running else 0, + "uptime_sec": uptime, + "bytes_sent": self._bytes_sent, + "total_mb": round(max(0, self._bytes_sent) / (1024 * 1024), 2), + "session_bytes": session_bytes, + "session_mb": round(session_bytes / (1024 * 1024), 2), + "estimated_kbps": int((session_bytes * 8 / 1000) / elapsed) if running else 0, + "restart_count": self._restart_count, + "consecutive_failures": self._consecutive_failures, + "next_retry_at": (now + retry_in_sec) if retry_in_sec else 0.0, + "retry_in_sec": retry_in_sec, + "last_error": self._last_error, + "last_frame_at": self._last_frame_at, + "last_frame_age_ms": last_frame_age_ms, + "last_frame_id": self._last_frame_id, + "frame_width": self._frame_width, + "frame_height": self._frame_height, + "frame_fps": self._frame_fps, + "log_tail": [f"{event['level']}: {event['message']}" for event in list(self._events)[-12:]], + "resource_status": resource_status, + "warnings": warnings, + } + + def test_config(self) -> dict[str, Any]: + stream_key = self.get_stream_key() + rtmps_ok, rtmps_message = _check_rtmps_reachable() + muxer_ok = bool( + self._muxer_capabilities.get("available") + and self._muxer_capabilities.get("flv") + and self._muxer_capabilities.get("h264") + and self._muxer_capabilities.get("aac") + ) + transport_ok = bool(self._transport_capabilities.get("available")) + ok = bool(stream_key and muxer_ok and transport_ok and rtmps_ok) + status = self.status() + return { + "ok": ok, + "configured": bool(stream_key), + "transport_available": transport_ok, + "transport": dict(self._transport_capabilities), + "rtmps_reachable": rtmps_ok, + "rtmps_message": rtmps_message, + "muxer_available": muxer_ok, + "muxer": dict(self._muxer_capabilities), + "source": self._current_source(), + "quality": self._quality_label(), + "resource_status": status["resource_status"], + "warnings": status["warnings"], + "log_tail": [f"{event['level']}: {event['message']}" for event in list(self._events)[-12:]], + "message": "ready" if ok else "stream key, RTMPS network, librtmp, or PyAV FLV/AAC is unavailable", + } + + def validate_stream_key(self, value: str | None = None) -> dict[str, Any]: + stream_key = _extract_stream_key(value) if value is not None else self.get_stream_key() + format_ok, format_message = _validate_stream_key_format(stream_key) + rtmps_ok, rtmps_message = _check_rtmps_reachable() + transport_ok = bool(self._transport_capabilities.get("available")) + muxer_ok = bool( + self._muxer_capabilities.get("available") + and self._muxer_capabilities.get("flv") + and self._muxer_capabilities.get("h264") + and self._muxer_capabilities.get("aac") + ) + ok = bool(format_ok and rtmps_ok and transport_ok and muxer_ok) + return { + "ok": ok, + "configured": bool(self.get_stream_key()), + "format_ok": format_ok, + "format_message": format_message, + "rtmps_reachable": rtmps_ok, + "rtmps_message": rtmps_message, + "transport_available": transport_ok, + "muxer_available": muxer_ok, + "masked_key": _mask_stream_key(stream_key), + "note": "YouTube only confirms whether the key is accepted when an encoder starts streaming.", + } + + def diagnostics(self) -> dict[str, Any]: + status = self.status() + stream_key = self.get_stream_key() + return { + "generated_at": _now(), + "status": status, + "config": { + "stream_key_configured": bool(stream_key), + "stream_key_masked": _mask_stream_key(stream_key), + "source": self._current_source(), + "quality": self._quality_label(), + "rtmps_ingest": f"{YOUTUBE_RTMPS_BASE}/{_mask_stream_key(stream_key)}" if stream_key else "", + }, + "params": { + YOUTUBE_LIVE_PARAM: self._param_enabled(), + YOUTUBE_QUALITY_PARAM: self._param_int(YOUTUBE_QUALITY_PARAM, 0), + YOUTUBE_TIMESTAMP_PARAM: self._param_bool(YOUTUBE_TIMESTAMP_PARAM), + "ClusterHud": self._param_int("ClusterHud", 0), + "DisableDM": self._param_int("DisableDM", 0), + "IsOnroad": self._param_bool("IsOnroad", False), + }, + "transport": { + "name": "librtmp-rtmps", + **self._transport_capabilities, + "connected": self._transport_connected, + "log_tail": [f"{event['level']}: {event['message']}" for event in list(self._events)[-12:]], + }, + "muxer": { + "name": "pyav-flv-aac", + **self._muxer_capabilities, + }, + "processes": self._process_status(), + "state_path": str(self.state_path), + } + + def get_stream_key(self) -> str: + payload = _read_json(self.secret_path) + return str(payload.get("stream_key") or "").strip() + + def set_stream_key(self, value: str) -> dict[str, Any]: + stream_key = _extract_stream_key(value) + if not stream_key: + raise ValueError("stream key is required") + _write_json_atomic(self.secret_path, {"stream_key": stream_key, "updated_at": _now()}, mode=0o600) + self._last_error = "" + self._write_status(force=True) + return self.status() + + def clear_stream_key(self) -> dict[str, Any]: + try: + self.secret_path.unlink() + except FileNotFoundError: + pass + self._write_status(force=True) + return self.status() + + async def _run(self) -> None: + while not self._stop_event.is_set(): + try: + await self._tick() + except asyncio.CancelledError: + raise + except Exception as exc: + self._last_error = str(exc) + self._set_state("error") + await self._stop_stream() + await asyncio.sleep(START_BACKOFF_SECONDS) + await asyncio.sleep(0.02 if self._transport else 0.25) + + async def _tick(self) -> None: + if not self._param_enabled(): + if self._transport is not None: + await self._stop_stream() + self._last_error = "" + self._consecutive_failures = 0 + self._next_retry_mono = 0.0 + self._set_state("disabled") + return + + stream_key = self.get_stream_key() + if not stream_key: + await self._stop_stream() + self._set_state("needs_setup") + return + + if not self._transport_capabilities.get("available"): + await self._stop_stream() + self._last_error = self._transport_capabilities.get("error") or "librtmp is unavailable" + self._set_state("error") + return + + if not ( + self._muxer_capabilities.get("available") + and self._muxer_capabilities.get("flv") + and self._muxer_capabilities.get("h264") + and self._muxer_capabilities.get("aac") + ): + await self._stop_stream() + self._last_error = "PyAV FLV/AAC muxer is unavailable" + self._set_state("error") + return + + if self._transport is not None and self._active_source and self._active_source != self._current_source(): + self._log(f"source changed to {self._current_source()}; restarting") + await self._stop_stream() + + if self._next_retry_mono and _mono() < self._next_retry_mono: + self._set_state("backoff") + return + + if self._transport is not None and not await asyncio.to_thread(self._transport.is_connected): + self._transport_connected = False + self._last_error = "YouTube RTMPS connection closed" + await self._stop_stream() + self._schedule_backoff("RTMPS connection closed") + self._set_state("backoff") + return + + header, data, frame_id, keyframe, width, height = self._recv_frame() + if not data: + if self._transport is not None and self._last_frame_mono and _mono() - self._last_frame_mono > NO_FRAME_STOP_SECONDS: + self._last_error = f"no {self._current_source()} frames" + await self._stop_stream() + self._schedule_backoff("frame timeout") + self._set_state("waiting_frame" if self._transport is None else "live") + return + + self._last_frame_at = _now() + self._last_frame_mono = _mono() + self._last_frame_id = frame_id + self._frame_width = width + self._frame_height = height + + if self._transport is None: + if not keyframe or not header: + self._set_state("waiting_keyframe") + return + if self._last_start_mono and (_mono() - self._last_start_mono) < MIN_RECONNECT_INTERVAL_SECONDS: + # Avoid hammering YouTube with rapid reconnects (it rejects them and it + # spins the state machine); hold briefly between start attempts. + self._set_state("backoff") + return + try: + await self._start_stream(stream_key, codec_header=header, width=width, height=height) + except Exception as exc: + self._last_error = f"YouTube RTMPS start failed: {exc}" + self._schedule_backoff("RTMPS start failed") + self._set_state("backoff") + return + + data = self._caption_injector.inject( + data, + enabled=self._param_bool(YOUTUBE_TIMESTAMP_PARAM), + ) + if await self._write_frame(data, keyframe=keyframe): + self._set_state("live") + + def _set_state(self, state: str) -> None: + self._state = state + self._write_status() + + def _write_status(self, *, force: bool = False) -> None: + mono = _mono() + if not force and mono - self._last_status_write < STATUS_WRITE_MIN_INTERVAL: + return + self._last_status_write = mono + try: + _write_json_atomic(self.state_path, {"updated_at": _now(), **self.status()}) + except Exception: + pass + + def _log(self, message: str, level: str = "info") -> None: + # Rolling event history that survives restarts (unlike _last_error, which is + # cleared on each start attempt). Surfaced via status()/diagnostics log_tail. + text = str(message or "").strip() + if not text: + return + self._events.append({"t": round(_now(), 3), "level": level, "message": text}) + + def _param_enabled(self) -> bool: + try: + params = self._get_params() + return bool(params and params.get_bool(YOUTUBE_LIVE_PARAM)) + except Exception: + return False + + def _param_bool(self, name: str, default: bool = False) -> bool: + try: + params = self._get_params() + if params is None: + return default + if hasattr(params, "get_bool"): + return bool(params.get_bool(name)) + raw = params.get(name) + if isinstance(raw, bytes): + raw = raw.decode("utf-8", errors="replace") + return str(raw).strip() in ("1", "true", "True") + except Exception: + return default + + def _param_int(self, name: str, default: int = 0) -> int: + try: + params = self._get_params() + if params is None: + return default + if hasattr(params, "get_int"): + return int(params.get_int(name)) + raw = params.get(name) + if isinstance(raw, bytes): + raw = raw.decode("utf-8", errors="replace") + return int(str(raw).strip()) + except Exception: + return default + + def _get_params(self) -> Any | None: + if self._params is not None: + return self._params + try: + from openpilot.common.params import Params + self._params = Params() + except Exception: + self._params = None + return self._params + + def _get_messaging(self) -> Any | None: + if self._messaging is not None: + return self._messaging + try: + from cereal import messaging + self._messaging = messaging + except Exception as exc: + self._last_error = f"messaging unavailable: {exc}" + self._messaging = None + return self._messaging + + def _current_source(self) -> str: + quality = self._param_int(YOUTUBE_QUALITY_PARAM, 0) + return SOURCE_BY_QUALITY.get(quality, SOURCE_BY_QUALITY[0]) + + def _quality_label(self) -> str: + return QUALITY_LABELS.get(self._param_int(YOUTUBE_QUALITY_PARAM, 0), "standard") + + def _get_socket(self) -> Any | None: + source = self._current_source() + if self._socket is not None and self._socket_source == source: + return self._socket + # first subscription or the selected source (quality/camera) changed + self._socket = None + messaging = self._get_messaging() + if messaging is None: + return None + try: + self._socket = messaging.sub_sock(source, conflate=True) + self._socket_source = source + except Exception as exc: + self._last_error = f"{source} socket failed: {exc}" + self._socket = None + return self._socket + + def _recv_frame(self) -> tuple[bytes, bytes, int | None, bool, int, int]: + messaging = self._get_messaging() + sock = self._get_socket() + if messaging is None or sock is None: + return b"", b"", None, False, 526, 330 + try: + msg = messaging.recv_one_or_none(sock) + if msg is None: + return b"", b"", None, False, 526, 330 + which = msg.which() + frame = getattr(msg, which, None) + if frame is None: + return b"", b"", None, False, 526, 330 + header = bytes(getattr(frame, "header", b"") or b"") + data = bytes(getattr(frame, "data", b"") or b"") + frame_id = None + flags = 0 + idx = getattr(frame, "idx", None) + if idx is not None: + try: + frame_id = int(getattr(idx, "frameId", 0) or 0) or None + except Exception: + frame_id = None + try: + flags = int(getattr(idx, "flags", 0) or 0) + except Exception: + flags = 0 + width = int(getattr(frame, "width", 0) or 526) + height = int(getattr(frame, "height", 0) or 330) + keyframe = bool(header) or bool(flags & 0x8) + return header, data, frame_id, keyframe, width, height + except Exception as exc: + self._last_error = f"{self._current_source()} recv failed: {exc}" + return b"", b"", None, False, 526, 330 + + async def _start_stream(self, stream_key: str, *, codec_header: bytes, width: int, height: int) -> None: + await self._stop_stream() + self._last_error = "" + self._restart_count += 1 + self._last_start_mono = _mono() + self._set_state("starting") + self._log("connecting to YouTube RTMPS") + rtmp_url = f"{YOUTUBE_RTMPS_BASE}/{stream_key}" + transport = LibrtmpClient(rtmp_url) + try: + await asyncio.to_thread(transport.connect) + muxer = await asyncio.to_thread( + H264FlvMuxer, + RtmpSink(transport), + codec_header=codec_header, + fps=self._frame_fps, + width=width, + height=height, + ) + except Exception: + await asyncio.to_thread(transport.close) + raise + self._transport = transport + self._transport_connected = True + self._muxer = muxer + self._started_at = _now() + self._started_mono = _mono() + self._session_started_bytes = self._bytes_sent + self._next_retry_mono = 0.0 + self._active_source = self._current_source() + self._caption_injector.reset() + self._log(f"stream started ({width}x{height})") + + async def _write_frame(self, payload: bytes, *, keyframe: bool) -> bool: + transport = self._transport + muxer = self._muxer + if transport is None or muxer is None: + return False + if not await asyncio.to_thread(transport.is_connected): + self._transport_connected = False + self._last_error = "YouTube RTMPS connection closed" + self._schedule_backoff("RTMPS connection closed") + await self._stop_stream() + return False + try: + await asyncio.to_thread(muxer.mux, payload, keyframe=keyframe) + self._bytes_sent = self._session_started_bytes + transport.bytes_written + if self._started_mono and _mono() - self._started_mono >= STREAM_STABLE_SECONDS: + self._consecutive_failures = 0 + return True + except Exception as exc: + self._transport_connected = False + self._last_error = f"YouTube RTMPS publish failed: {exc}" + self._schedule_backoff("RTMPS publish failed") + await self._stop_stream() + return False + + async def _stop_stream(self) -> None: + transport = self._transport + self._transport = None + self._transport_connected = False + muxer = self._muxer + self._muxer = None + if transport is None and muxer is None: + self._started_at = 0.0 + self._started_mono = 0.0 + return + self._set_state("stopping") + if muxer is not None: + try: + await asyncio.to_thread(muxer.close) + except Exception: + pass + if transport is not None: + self._bytes_sent = self._session_started_bytes + transport.bytes_written + try: + await asyncio.to_thread(transport.close) + except Exception: + pass + self._started_at = 0.0 + self._started_mono = 0.0 + + def _schedule_backoff(self, reason: str = "") -> None: + self._consecutive_failures += 1 + delay = min(BACKOFF_MAX_SECONDS, BACKOFF_BASE_SECONDS * (2 ** max(0, self._consecutive_failures - 1))) + self._next_retry_mono = _mono() + delay + if reason and not self._last_error: + self._last_error = reason + self._log(f"retry in {delay:.0f}s: {reason or self._last_error or 'reconnect'}", "warn") + + def _process_status(self) -> dict[str, Any]: + mono = _mono() + if self._proc_cache is not None and (mono - self._proc_cache_mono) < PROC_CACHE_SECONDS: + return self._proc_cache + result = {} + for name, match in _PROCESS_MATCHES.items(): + pids = _find_matching_pids(match) + result[name] = { + "running": bool(pids), + "pids": pids[:8], + } + self._proc_cache = result + self._proc_cache_mono = mono + return result + + def _resource_status(self) -> dict[str, Any]: + processes = self._process_status() + cluster_param = self._param_int("ClusterHud", 0) + disable_dm = self._param_int("DisableDM", 0) + return { + "cluster": { + "enabled": cluster_param in (1, 2), + "param": cluster_param, + **processes.get("carrot_cluster", {"running": False, "pids": []}), + }, + "carrot_vision": { + "enabled": disable_dm == 2, + "disable_dm": disable_dm, + "webrtcd_running": bool(processes.get("webrtcd", {}).get("running")), + "stream_encoderd_running": bool(processes.get("stream_encoderd", {}).get("running")), + "webrtcd_pids": list(processes.get("webrtcd", {}).get("pids") or []), + "stream_encoderd_pids": list(processes.get("stream_encoderd", {}).get("pids") or []), + }, + "youtube_encoder": { + **processes.get("youtube_encoderd", {"running": False, "pids": []}), + }, + } + + def _warnings(self, resource_status: dict[str, Any]) -> list[str]: + warnings = [] + if not self._param_enabled(): + return warnings + cluster = resource_status.get("cluster") if isinstance(resource_status, dict) else {} + vision = resource_status.get("carrot_vision") if isinstance(resource_status, dict) else {} + if cluster and cluster.get("enabled"): + warnings.append("Cluster HUD is enabled; monitor overall load and temperature during simultaneous use.") + if vision and vision.get("enabled"): + warnings.append("Carrot Vision is enabled; simultaneous streaming increases network and memory bandwidth use.") + quality = self._param_int(YOUTUBE_QUALITY_PARAM, 0) + if quality in (1, 3): + if not (vision and vision.get("stream_encoderd_running")): + warnings.append("The selected video mode is waiting for the shared livestream encoder to start onroad.") + elif quality == 2: + youtube_encoder = resource_status.get("youtube_encoder") if isinstance(resource_status, dict) else {} + if not (youtube_encoder and youtube_encoder.get("running")): + warnings.append("High quality is waiting for the dedicated YouTube encoder to start onroad.") + return warnings + + +def _pid_cmdline(pid: int) -> str: + try: + return Path(f"/proc/{int(pid)}/cmdline").read_bytes().decode(errors="replace") + except Exception: + return "" + + +def _pid_alive(pid: int, match: str = "") -> bool: + if pid <= 0: + return False + try: + os.kill(pid, 0) + except OSError: + return False + return not match or match in _pid_cmdline(pid) + + +def _find_matching_pids(match: str) -> list[int]: + if not Path("/proc").exists(): + return [] + matches = [] + for proc_path in Path("/proc").glob("[0-9]*"): + try: + pid = int(proc_path.name) + except ValueError: + continue + if _pid_alive(pid, match): + matches.append(pid) + return sorted(matches) diff --git a/selfdrive/carrot/server/services/youtube_live_captions.py b/selfdrive/carrot/server/services/youtube_live_captions.py new file mode 100644 index 0000000000..f6c1207b16 --- /dev/null +++ b/selfdrive/carrot/server/services/youtube_live_captions.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Iterable + + +_ANNEXB_START_CODE = b"\x00\x00\x00\x01" +_CEA608_RCL = (0x14, 0x20) +_CEA608_EDM = (0x14, 0x2C) +_CEA608_ENM = (0x14, 0x2E) +_CEA608_EOC = (0x14, 0x2F) +# Row 1, white, regular, indent 4. This is the closest stable PAC position to +# center for a 19-character timestamp without cumulative tab-offset commands. +_CEA608_TOP_CENTER_PAC = (0x11, 0x52) + + +def _with_odd_parity(value: int) -> int: + value &= 0x7F + return value | (0x80 if value.bit_count() % 2 == 0 else 0) + + +def _rbsp_escape(payload: bytes) -> bytes: + escaped = bytearray() + zero_count = 0 + for value in payload: + if zero_count >= 2 and value <= 0x03: + escaped.append(0x03) + zero_count = 0 + escaped.append(value) + zero_count = zero_count + 1 if value == 0 else 0 + return bytes(escaped) + + +def build_cea608_sei(pairs: tuple[int, int] | Iterable[tuple[int, int]]) -> bytes: + if isinstance(pairs, tuple) and len(pairs) == 2 and all(isinstance(value, int) for value in pairs): + caption_pairs = [pairs] + else: + caption_pairs = list(pairs) + if not caption_pairs or len(caption_pairs) > 31: + raise ValueError("CEA-608 SEI requires 1 to 31 caption pairs") + + cc_data = b"".join( + bytes((0xFC, _with_odd_parity(first), _with_odd_parity(second))) + for first, second in caption_pairs + ) + itu_t_t35 = ( + b"\xB5\x00\x31GA94\x03" + + bytes((0x40 | len(caption_pairs), 0xFF)) + + cc_data + + b"\xFF" + ) + sei_rbsp = bytes((0x04, len(itu_t_t35))) + itu_t_t35 + b"\x80" + return _ANNEXB_START_CODE + b"\x06" + _rbsp_escape(sei_rbsp) + + +def _timestamp_pairs(text: str) -> list[tuple[int, int]]: + clean = "".join(ch if 0x20 <= ord(ch) <= 0x7E else "?" for ch in text)[:32] + if len(clean) % 2: + clean += " " + text_pairs = [(ord(clean[index]), ord(clean[index + 1])) for index in range(0, len(clean), 2)] + return [ + _CEA608_RCL, _CEA608_RCL, + _CEA608_ENM, _CEA608_ENM, + _CEA608_TOP_CENTER_PAC, _CEA608_TOP_CENTER_PAC, + *text_pairs, + _CEA608_EOC, + ] + + +class Cea608TimestampInjector: + def __init__(self) -> None: + self._enabled = False + self._last_text = "" + self.packets_injected = 0 + + def reset(self) -> None: + self._enabled = False + self._last_text = "" + + def inject(self, payload: bytes, *, enabled: bool, now: datetime | None = None) -> bytes: + if not payload: + return payload + + if enabled != self._enabled: + self._enabled = enabled + self._last_text = "" + if not enabled: + self.packets_injected += 1 + return build_cea608_sei((_CEA608_EDM, _CEA608_EDM)) + payload + + if enabled: + local_now = now if now is not None else datetime.now().astimezone() + text = local_now.strftime("%Y-%m-%d %H:%M:%S") + if text != self._last_text: + self._last_text = text + self.packets_injected += 1 + return build_cea608_sei(_timestamp_pairs(text)) + payload + + return payload diff --git a/selfdrive/carrot/server/services/youtube_live_muxer.py b/selfdrive/carrot/server/services/youtube_live_muxer.py new file mode 100644 index 0000000000..f79c349b39 --- /dev/null +++ b/selfdrive/carrot/server/services/youtube_live_muxer.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +import threading +from fractions import Fraction +from typing import Any, BinaryIO + + +H264_CODEC = "h264" +AAC_CODEC = "aac" +AUDIO_RATE = 44_100 +AUDIO_BITRATE = 128_000 +AUDIO_SAMPLES = 1_024 + + +def _annexb_nalus(payload: bytes) -> list[bytes]: + starts: list[tuple[int, int]] = [] + index = 0 + size = len(payload) + while index + 3 <= size: + if index + 4 <= size and payload[index:index + 4] == b"\x00\x00\x00\x01": + starts.append((index, 4)) + index += 4 + elif payload[index:index + 3] == b"\x00\x00\x01": + starts.append((index, 3)) + index += 3 + else: + index += 1 + + nalus = [] + for position, (offset, start_size) in enumerate(starts): + start = offset + start_size + end = starts[position + 1][0] if position + 1 < len(starts) else size + while end > start and payload[end - 1] == 0: + end -= 1 + if end > start: + nalus.append(payload[start:end]) + return nalus + + +def _avc_decoder_configuration(codec_header: bytes) -> bytes: + if codec_header[:1] == b"\x01" and not codec_header.startswith((b"\x00\x00\x01", b"\x00\x00\x00\x01")): + return bytes(codec_header) + nalus = _annexb_nalus(codec_header) + sps_units = [nalu for nalu in nalus if nalu and nalu[0] & 0x1F == 7] + pps_units = [nalu for nalu in nalus if nalu and nalu[0] & 0x1F == 8] + if not sps_units or not pps_units or len(sps_units[0]) < 4: + raise ValueError("H.264 codec header has no SPS/PPS") + if len(sps_units) > 31 or len(pps_units) > 255: + raise ValueError("H.264 codec header has too many parameter sets") + + first_sps = sps_units[0] + result = bytearray((1, first_sps[1], first_sps[2], first_sps[3], 0xFF, 0xE0 | len(sps_units))) + for sps in sps_units: + result.extend(len(sps).to_bytes(2, "big")) + result.extend(sps) + result.append(len(pps_units)) + for pps in pps_units: + result.extend(len(pps).to_bytes(2, "big")) + result.extend(pps) + return bytes(result) + + +def _annexb_to_avcc(payload: bytes) -> bytes: + nalus = _annexb_nalus(payload) + if not nalus: + return bytes(payload) + return b"".join(len(nalu).to_bytes(4, "big") + nalu for nalu in nalus) + + +def pyav_capabilities() -> dict[str, Any]: + try: + import av + except Exception as exc: + return { + "available": False, + "version": "", + "flv": False, + "h264": False, + "aac": False, + "error": str(exc), + } + codecs = getattr(av, "codecs_available", set()) + return { + "available": True, + "version": str(getattr(av, "__version__", "")), + "flv": True, + "h264": H264_CODEC in codecs, + "aac": AAC_CODEC in codecs, + "error": "", + } + + +class H264FlvMuxer: + def __init__( + self, + output: BinaryIO, + *, + codec_header: bytes, + fps: int = 20, + width: int = 526, + height: int = 330, + ) -> None: + if not codec_header: + raise ValueError("H.264 codec header is required") + + import av + + self._av = av + self._output = output + self._fps = max(1, int(fps)) + self._video_time_base = Fraction(1, self._fps) + self._audio_time_base = Fraction(1, AUDIO_RATE) + self._video_config = _avc_decoder_configuration(codec_header) + self._audio_codec = av.CodecContext.create(AAC_CODEC, "w") + self._audio_codec.sample_rate = AUDIO_RATE + self._audio_codec.layout = "stereo" + self._audio_codec.format = "fltp" + self._audio_codec.bit_rate = AUDIO_BITRATE + self._audio_codec.time_base = self._audio_time_base + self._audio_codec.open() + self._packet_index = 0 + self._audio_pts = 0 + self._last_video_ms = 0 + self._closed = False + self._lock = threading.RLock() + self._output.write(b"FLV\x01\x05\x00\x00\x00\x09\x00\x00\x00\x00") + self._write_tag(9, 0, b"\x17\x00\x00\x00\x00" + self._video_config) + audio_config = bytes(self._audio_codec.extradata or b"\x12\x10") + self._write_tag(8, 0, b"\xAF\x00" + audio_config) + + def mux(self, payload: bytes, *, keyframe: bool = False, timestamp_ms: int | None = None) -> None: + with self._lock: + if self._closed: + raise RuntimeError("FLV muxer is closed") + if not payload: + return + + if timestamp_ms is None: + video_ms = int(self._packet_index * 1000 / self._fps) + else: + video_ms = max(0, int(timestamp_ms)) + # FLV/RTMP timestamps must be non-decreasing + if video_ms < self._last_video_ms: + video_ms = self._last_video_ms + self._last_video_ms = video_ms + + # keep the silent audio track filled up to the current video time so a + # dropped-frame gap stays A/V aligned + self._mux_silence_until(int(video_ms * AUDIO_RATE / 1000)) + + frame_header = b"\x17" if keyframe else b"\x27" + self._write_tag(9, video_ms, frame_header + b"\x01\x00\x00\x00" + _annexb_to_avcc(payload)) + self._packet_index += 1 + + def close(self) -> None: + with self._lock: + if self._closed: + return + self._closed = True + try: + self._mux_silence_until(int(self._last_video_ms * AUDIO_RATE / 1000)) + for packet in self._audio_codec.encode(None): + self._write_audio_packet(packet) + finally: + self._output.flush() + + def _mux_silence_until(self, target_pts: int) -> None: + while self._audio_pts <= target_pts: + frame = self._av.AudioFrame(format="fltp", layout="stereo", samples=AUDIO_SAMPLES) + frame.sample_rate = AUDIO_RATE + frame.pts = self._audio_pts + frame.time_base = self._audio_time_base + for plane in frame.planes: + plane.update(bytes(plane.buffer_size)) + for packet in self._audio_codec.encode(frame): + self._write_audio_packet(packet) + self._audio_pts += AUDIO_SAMPLES + + def _write_audio_packet(self, packet: Any) -> None: + packet_pts = packet.pts if packet.pts is not None else self._audio_pts + time_base = packet.time_base or self._audio_time_base + timestamp_ms = max(0, int(packet_pts * time_base * 1000)) + self._write_tag(8, timestamp_ms, b"\xAF\x01" + bytes(packet)) + + def _write_tag(self, tag_type: int, timestamp_ms: int, payload: bytes) -> None: + timestamp = max(0, int(timestamp_ms)) & 0xFFFFFFFF + header = ( + bytes((tag_type,)) + + len(payload).to_bytes(3, "big") + + (timestamp & 0xFFFFFF).to_bytes(3, "big") + + bytes(((timestamp >> 24) & 0xFF,)) + + b"\x00\x00\x00" + ) + self._output.write(header + payload + (len(payload) + 11).to_bytes(4, "big")) diff --git a/selfdrive/carrot/server/services/youtube_live_transport.py b/selfdrive/carrot/server/services/youtube_live_transport.py new file mode 100644 index 0000000000..a52abf49db --- /dev/null +++ b/selfdrive/carrot/server/services/youtube_live_transport.py @@ -0,0 +1,308 @@ +from __future__ import annotations + +import ctypes +import ctypes.util +import io +import select +import socket +import ssl +import threading +from typing import Any +from urllib.parse import urlsplit, urlunsplit + + +REQUIRED_RTMP_SYMBOLS = ( + "RTMP_Alloc", + "RTMP_Init", + "RTMP_SetupURL", + "RTMP_SetOpt", + "RTMP_EnableWrite", + "RTMP_Connect", + "RTMP_ConnectStream", + "RTMP_Write", + "RTMP_IsConnected", + "RTMP_Close", + "RTMP_Free", +) + + +class LibrtmpError(RuntimeError): + pass + + +class _AVal(ctypes.Structure): + _fields_ = [("value", ctypes.c_char_p), ("length", ctypes.c_int)] + + +class _TlsTunnel: + def __init__(self, host: str, port: int) -> None: + self._host = host + self._port = port + self._stop_event = threading.Event() + self._listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._listener.bind(("127.0.0.1", 0)) + self._listener.listen(1) + self.local_port = int(self._listener.getsockname()[1]) + self.error = "" + self._local: socket.socket | None = None + self._remote: ssl.SSLSocket | None = None + self._thread = threading.Thread(target=self._run, name="youtube-rtmps-tunnel", daemon=True) + + def start(self) -> None: + self._thread.start() + + def close(self) -> None: + self._stop_event.set() + for sock in (self._local, self._remote, self._listener): + if sock is not None: + try: + sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass + try: + sock.close() + except OSError: + pass + if self._thread.is_alive() and threading.current_thread() is not self._thread: + self._thread.join(timeout=2.0) + + def _run(self) -> None: + raw_remote: socket.socket | None = None + try: + self._listener.settimeout(10.0) + self._local, _ = self._listener.accept() + raw_remote = socket.create_connection((self._host, self._port), timeout=8.0) + self._remote = ssl.create_default_context().wrap_socket(raw_remote, server_hostname=self._host) + raw_remote = None + self._local.settimeout(None) + self._remote.settimeout(None) + sockets = (self._local, self._remote) + while not self._stop_event.is_set(): + readable, _, _ = select.select(sockets, [], [], 0.5) + for source in readable: + payload = source.recv(64 * 1024) + if not payload: + return + target = self._remote if source is self._local else self._local + target.sendall(payload) + except Exception as exc: + if not self._stop_event.is_set(): + self.error = str(exc) + finally: + if raw_remote is not None: + raw_remote.close() + for sock in (self._local, self._remote): + if sock is not None: + try: + sock.close() + except OSError: + pass + + +def librtmp_capabilities() -> dict[str, Any]: + path = ctypes.util.find_library("rtmp") or "" + if not path: + return {"available": False, "path": "", "missing_symbols": [], "error": "librtmp not found"} + try: + library = ctypes.CDLL(path) + missing = [name for name in REQUIRED_RTMP_SYMBOLS if not hasattr(library, name)] + return { + "available": not missing, + "path": path, + "rtmps_mode": "python-tls-tunnel", + "missing_symbols": missing, + "error": f"missing librtmp symbols: {', '.join(missing)}" if missing else "", + } + except Exception as exc: + return {"available": False, "path": path, "missing_symbols": [], "error": str(exc)} + + +class LibrtmpClient: + def __init__(self, url: str) -> None: + path = ctypes.util.find_library("rtmp") + if not path: + raise LibrtmpError("librtmp not found") + self._library = ctypes.CDLL(path) + self._configure_library() + self._url = url + self._url_buffer: ctypes.Array[ctypes.c_char] | None = None + self._tc_url_buffer: ctypes.Array[ctypes.c_char] | None = None + self._tunnel: _TlsTunnel | None = None + self._handle: int | None = None + self._lock = threading.RLock() + self._bytes_written = 0 + + @property + def bytes_written(self) -> int: + with self._lock: + return self._bytes_written + + def connect(self) -> None: + with self._lock: + if self._handle is not None: + return + handle = self._library.RTMP_Alloc() + if not handle: + raise LibrtmpError("librtmp allocation failed") + self._handle = handle + try: + self._library.RTMP_Init(handle) + setup_url, tc_url = self._prepare_url() + self._url_buffer = ctypes.create_string_buffer(setup_url.encode("utf-8")) + if not self._library.RTMP_SetupURL(handle, self._url_buffer): + raise LibrtmpError("librtmp URL setup failed") + if tc_url: + self._set_string_option(handle, "tcUrl", tc_url) + self._library.RTMP_EnableWrite(handle) + if not self._library.RTMP_Connect(handle, None): + detail = f": {self._tunnel.error}" if self._tunnel and self._tunnel.error else "" + raise LibrtmpError(f"YouTube RTMPS connection failed{detail}") + if not self._library.RTMP_ConnectStream(handle, 0): + raise LibrtmpError("YouTube rejected the publish connection") + except Exception: + self._close_locked() + raise + + def write(self, data: bytes | bytearray | memoryview) -> int: + # Returns bytes CONSUMED by librtmp (may be < len). librtmp's RTMP_Write + # parses FLV tags and reads look-ahead beyond the current tag; feeding it a + # small single-tag buffer makes it read out of bounds and segfault, so the + # caller must batch and carry the unconsumed remainder forward. + payload = bytes(data) + if not payload: + return 0 + with self._lock: + handle = self._handle + if handle is None or not self._library.RTMP_IsConnected(handle): + raise LibrtmpError("YouTube RTMPS connection is closed") + buffer = ctypes.create_string_buffer(payload) + written = int(self._library.RTMP_Write(handle, buffer, len(payload))) + if written < 0: + raise LibrtmpError("YouTube RTMPS write failed") + self._bytes_written += written + return written + + def is_connected(self) -> bool: + with self._lock: + return bool(self._handle and self._library.RTMP_IsConnected(self._handle)) + + def close(self) -> None: + with self._lock: + self._close_locked() + + def _close_locked(self) -> None: + handle = self._handle + self._handle = None + if handle is None: + self._url_buffer = None + self._tc_url_buffer = None + if self._tunnel is not None: + self._tunnel.close() + self._tunnel = None + return + try: + self._library.RTMP_Close(handle) + finally: + self._library.RTMP_Free(handle) + self._url_buffer = None + self._tc_url_buffer = None + if self._tunnel is not None: + self._tunnel.close() + self._tunnel = None + + def _prepare_url(self) -> tuple[str, str]: + parsed = urlsplit(self._url) + if parsed.scheme.lower() != "rtmps": + return self._url, "" + host = parsed.hostname or "" + if not host: + raise LibrtmpError("RTMPS host is missing") + port = parsed.port or 443 + self._tunnel = _TlsTunnel(host, port) + self._tunnel.start() + setup_url = urlunsplit(("rtmp", f"127.0.0.1:{self._tunnel.local_port}", parsed.path, parsed.query, "")) + path_parts = [part for part in parsed.path.split("/") if part] + app_path = f"/{path_parts[0]}" if path_parts else "/" + tc_url = urlunsplit(("rtmps", parsed.netloc, app_path, "", "")) + return setup_url, tc_url + + def _set_string_option(self, handle: int, name: str, value: str) -> None: + name_buffer = ctypes.create_string_buffer(name.encode("utf-8")) + self._tc_url_buffer = ctypes.create_string_buffer(value.encode("utf-8")) + option = _AVal(ctypes.cast(name_buffer, ctypes.c_char_p), len(name_buffer.value)) + argument = _AVal(ctypes.cast(self._tc_url_buffer, ctypes.c_char_p), len(self._tc_url_buffer.value)) + if not self._library.RTMP_SetOpt(handle, ctypes.byref(option), ctypes.byref(argument)): + raise LibrtmpError(f"librtmp option failed: {name}") + + def _configure_library(self) -> None: + missing = [name for name in REQUIRED_RTMP_SYMBOLS if not hasattr(self._library, name)] + if missing: + raise LibrtmpError(f"missing librtmp symbols: {', '.join(missing)}") + + self._library.RTMP_Alloc.restype = ctypes.c_void_p + self._library.RTMP_Init.argtypes = [ctypes.c_void_p] + self._library.RTMP_SetupURL.argtypes = [ctypes.c_void_p, ctypes.c_char_p] + self._library.RTMP_SetupURL.restype = ctypes.c_int + self._library.RTMP_SetOpt.argtypes = [ctypes.c_void_p, ctypes.POINTER(_AVal), ctypes.POINTER(_AVal)] + self._library.RTMP_SetOpt.restype = ctypes.c_int + self._library.RTMP_EnableWrite.argtypes = [ctypes.c_void_p] + self._library.RTMP_Connect.argtypes = [ctypes.c_void_p, ctypes.c_void_p] + self._library.RTMP_Connect.restype = ctypes.c_int + self._library.RTMP_ConnectStream.argtypes = [ctypes.c_void_p, ctypes.c_int] + self._library.RTMP_ConnectStream.restype = ctypes.c_int + self._library.RTMP_Write.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_int] + self._library.RTMP_Write.restype = ctypes.c_int + self._library.RTMP_IsConnected.argtypes = [ctypes.c_void_p] + self._library.RTMP_IsConnected.restype = ctypes.c_int + self._library.RTMP_Close.argtypes = [ctypes.c_void_p] + self._library.RTMP_Free.argtypes = [ctypes.c_void_p] + if hasattr(self._library, "RTMP_LogSetLevel"): + self._library.RTMP_LogSetLevel.argtypes = [ctypes.c_int] + self._library.RTMP_LogSetLevel(1) + + +class RtmpSink(io.RawIOBase): + # Buffer muxer output and hand librtmp large, multi-tag chunks. Feeding + # librtmp one small FLV tag at a time makes RTMP_Write read past the buffer + # (look-ahead) and segfault. We keep whatever librtmp does not consume and + # prepend it to the next batch so it always has trailing context. + FLUSH_THRESHOLD = 16384 + + def __init__(self, client: LibrtmpClient) -> None: + super().__init__() + self._client = client + self._position = 0 + self._pending = bytearray() + + def writable(self) -> bool: + return True + + def seekable(self) -> bool: + return False + + def write(self, data: bytes | bytearray | memoryview) -> int: + chunk = bytes(data) + self._pending.extend(chunk) + self._position += len(chunk) + if len(self._pending) >= self.FLUSH_THRESHOLD: + self._drain(self.FLUSH_THRESHOLD) + return len(chunk) + + def _drain(self, floor: int) -> None: + # Keep at least `floor` bytes buffered so librtmp always has look-ahead. + while len(self._pending) >= floor: + written = self._client.write(bytes(self._pending)) + if written <= 0: + break + del self._pending[:written] + + def tell(self) -> int: + return self._position + + def flush(self) -> None: + while self._pending: + written = self._client.write(bytes(self._pending)) + if written <= 0: + break + del self._pending[:written] diff --git a/selfdrive/carrot/server/tests/test_youtube_live_captions.py b/selfdrive/carrot/server/tests/test_youtube_live_captions.py new file mode 100644 index 0000000000..e25ba17dee --- /dev/null +++ b/selfdrive/carrot/server/tests/test_youtube_live_captions.py @@ -0,0 +1,87 @@ +from datetime import datetime, timezone + +from selfdrive.carrot.server.services.youtube_live_captions import Cea608TimestampInjector, build_cea608_sei +from selfdrive.carrot.server.services.youtube_live_muxer import _annexb_to_avcc + + +def _unescape_rbsp(payload: bytes) -> bytes: + result = bytearray() + index = 0 + while index < len(payload): + if payload[index:index + 3] == b"\x00\x00\x03": + result.extend(b"\x00\x00") + index += 3 + else: + result.append(payload[index]) + index += 1 + return bytes(result) + + +def _caption_pairs(sei: bytes) -> list[tuple[int, int]]: + assert sei.startswith(b"\x00\x00\x00\x01\x06") + rbsp = _unescape_rbsp(sei[5:]) + assert rbsp[0] == 0x04 + assert rbsp[2:10] == b"\xB5\x00\x31GA94\x03" + count = rbsp[10] & 0x1F + assert rbsp[10] & 0x40 + assert rbsp[11] == 0xFF + result = [] + for index in range(count): + offset = 12 + index * 3 + assert rbsp[offset] == 0xFC + result.append((rbsp[offset + 1] & 0x7F, rbsp[offset + 2] & 0x7F)) + return result + + +def _caption_pair(sei: bytes) -> tuple[int, int]: + pairs = _caption_pairs(sei) + assert len(pairs) == 1 + return pairs[0] + + +def test_build_cea608_sei_has_atsc_payload_and_odd_parity() -> None: + sei = build_cea608_sei((ord("2"), ord("0"))) + rbsp = _unescape_rbsp(sei[5:]) + assert _caption_pair(sei) == (ord("2"), ord("0")) + assert rbsp[13].bit_count() % 2 == 1 + assert rbsp[14].bit_count() % 2 == 1 + assert rbsp[-1] == 0x80 + + +def test_flv_avcc_conversion_keeps_caption_sei_before_video() -> None: + video = b"\x00\x00\x00\x01\x65\x88\x84" + avcc = _annexb_to_avcc(build_cea608_sei((ord("2"), ord("0"))) + video) + first_size = int.from_bytes(avcc[:4], "big") + first_nalu = avcc[4:4 + first_size] + second_offset = 4 + first_size + second_size = int.from_bytes(avcc[second_offset:second_offset + 4], "big") + second_nalu = avcc[second_offset + 4:second_offset + 4 + second_size] + assert first_nalu[0] & 0x1F == 6 + assert second_nalu[0] & 0x1F == 5 + + +def test_timestamp_sequence_is_injected_without_changing_video_payload() -> None: + injector = Cea608TimestampInjector() + video = b"\x00\x00\x00\x01\x65\x88\x84" + now = datetime(2026, 7, 1, 8, 32, 9, tzinfo=timezone.utc) + injected = injector.inject(video, enabled=True, now=now) + assert injected.endswith(video) + pairs = _caption_pairs(injected[:-len(video)]) + + text = "".join(chr(value) for pair in pairs[6:-1] for value in pair).rstrip() + assert text == "2026-07-01 08:32:09" + assert pairs[:2] == [(0x14, 0x20)] * 2 + assert pairs[2:4] == [(0x14, 0x2E)] * 2 + assert pairs[4:6] == [(0x11, 0x52)] * 2 + assert pairs[-1] == (0x14, 0x2F) + assert injector.inject(video, enabled=True, now=now) == video + assert injector.packets_injected == 1 + + +def test_disabling_captions_sends_erase_display_memory() -> None: + injector = Cea608TimestampInjector() + video = b"\x00\x00\x00\x01\x41\x01" + injector.inject(video, enabled=True, now=datetime(2026, 7, 1, tzinfo=timezone.utc)) + first = injector.inject(video, enabled=False) + assert _caption_pairs(first[:-len(video)]) == [(0x14, 0x2C)] * 2 + assert injector.inject(video, enabled=False) == video diff --git a/selfdrive/carrot/web/css/components/nav_status_badges.css b/selfdrive/carrot/web/css/components/nav_status_badges.css new file mode 100644 index 0000000000..f9b1d07597 --- /dev/null +++ b/selfdrive/carrot/web/css/components/nav_status_badges.css @@ -0,0 +1,80 @@ +.topbar #btnHome .nav-status-badges { + position: absolute; + top: 5px; + right: 6px; + z-index: 1; + display: none; + max-width: calc(100% - 12px); + align-items: center; + justify-content: flex-end; + gap: 2px; + pointer-events: none; +} + +.topbar #btnHome .nav-status-badges.is-visible { + display: flex; +} + +.topbar #btnHome.youtube-live, +.topbar #btnHome.youtube-live.active { + color: #ff8f84; +} + +.topbar #btnHome.youtube-live.active { + background: color-mix(in srgb, #ff6b6b 18%, transparent); +} + +.topbar #btnHome.youtube-live.active > span:last-child::after { + background: currentColor; +} + +.topbar #btnHome.recording, +.topbar #btnHome.youtube-live { + animation: nav-status-button-fade 2.4s ease-in-out infinite; +} + +.topbar #btnHome .nav-status-badge { + min-width: 24px; + padding: 2px 4px; + border-radius: 999px; + color: var(--md-on-error, #ffffff); + font-size: 8px; + font-weight: 900; + letter-spacing: 0; + line-height: 1; + text-align: center; + white-space: nowrap; + box-shadow: 0 0 0 1px color-mix(in srgb, var(--md-surface) 72%, transparent); +} + +.topbar #btnHome .nav-status-badge--record { + background: rgba(118, 18, 20, 0.92); + color: #ffd7d3; +} + +.topbar #btnHome .nav-status-badge--live { + background: var(--md-error, #ba1a1a); +} + +/* The shared badge container replaces the legacy REC pseudo-element. */ +.topbar #btnHome.recording::before { + display: none; + animation: none; +} + +@keyframes nav-status-button-fade { + 0%, 100% { + background: color-mix(in srgb, #c9827b 5%, transparent); + } + 50% { + background: color-mix(in srgb, #c9827b 14%, transparent); + } +} + +@media (prefers-reduced-motion: reduce) { + .topbar #btnHome.recording, + .topbar #btnHome.youtube-live { + animation: none; + background: color-mix(in srgb, #c9827b 10%, transparent); + } +} diff --git a/selfdrive/carrot/web/css/pages/settings/youtube_live.css b/selfdrive/carrot/web/css/pages/settings/youtube_live.css new file mode 100644 index 0000000000..644bbdedf5 --- /dev/null +++ b/selfdrive/carrot/web/css/pages/settings/youtube_live.css @@ -0,0 +1,145 @@ +.youtube-live-section { + display: grid; + gap: var(--sp-md); + margin: 0; +} + +.youtube-live-key-card, +.youtube-live-card, +.youtube-live-help-card { + display: grid; + gap: var(--sp-md); + padding-top: var(--sp-lg); + padding-bottom: var(--sp-lg); +} + +.youtube-live-card__summary { + min-width: 0; +} + +.youtube-live-card__title { + color: var(--md-on-surface); + font-size: var(--fs-title-sm); + font-weight: 800; + letter-spacing: 0; +} + +.youtube-live-card__desc, +.youtube-live-metric__label { + color: var(--md-on-surface-var); + font-size: var(--fs-body-sm); + letter-spacing: 0; + overflow-wrap: anywhere; +} + +.youtube-live-help-list { + display: grid; + gap: var(--sp-sm); + margin: 0; + padding-left: 1.35em; + color: var(--md-on-surface-var); + font-size: var(--fs-body-sm); + letter-spacing: 0; +} + +.youtube-live-help-list > li { + min-width: 0; + overflow-wrap: anywhere; +} + +.youtube-live-key-row { + display: grid; + grid-template-columns: minmax(0, 1fr) auto auto auto; + gap: 8px; + align-items: center; +} + +.youtube-live-key-input { + min-width: 0; + min-height: 40px; + border: 1px solid color-mix(in srgb, var(--md-outline-var) 42%, transparent); + border-radius: var(--r-md); + padding: 0 12px; + color: var(--md-on-surface); + background: color-mix(in srgb, var(--md-surface) 72%, transparent); + font: inherit; + font-size: var(--fs-body); + letter-spacing: 0; +} + +.youtube-live-key-input:focus { + outline: none; + border-color: color-mix(in srgb, var(--md-primary) 62%, var(--md-outline-var)); + box-shadow: 0 0 0 2px color-mix(in srgb, var(--md-primary) 16%, transparent); +} + +.youtube-live-metrics { + display: grid; + grid-template-columns: repeat(3, minmax(0, 1fr)); + gap: 8px; +} + +.youtube-live-metric { + min-width: 0; + border: 1px solid color-mix(in srgb, var(--md-outline-var) 22%, transparent); + border-radius: var(--r-md); + padding: 8px 10px; + background: color-mix(in srgb, var(--md-surface) 55%, transparent); +} + +.youtube-live-metric__value { + margin-top: 2px; + color: var(--md-on-surface); + font-size: var(--fs-body); + font-weight: 800; + letter-spacing: 0; + overflow-wrap: anywhere; +} + +.youtube-live-actions { + --ui-action-min: 136px; +} + +.page--setting #settingScreenItems .youtube-timestamp-disabled .setting-switch { + cursor: not-allowed; +} + +.page--setting #settingScreenItems .youtube-timestamp-disabled .setting-switch__track, +.page--setting #settingScreenItems .youtube-timestamp-disabled .setting-switch__input:checked + .setting-switch__track { + border-color: color-mix(in srgb, #000 84%, var(--md-outline-var)); + background: #000; +} + +.page--setting #settingScreenItems .youtube-timestamp-disabled .setting-switch__track::after, +.page--setting #settingScreenItems .youtube-timestamp-disabled .setting-switch__input:checked + .setting-switch__track::after { + transform: none; + background: color-mix(in srgb, var(--md-on-surface-var) 42%, #000); + box-shadow: none; +} + +.youtube-live-key-row .smallBtn { + display: inline-flex; + align-items: center; + justify-content: center; + width: auto; + min-width: 0; + min-height: 36px; + flex: 0 1 auto; + white-space: nowrap; + padding: 0 12px; +} + +@media (max-width: 430px) { + .youtube-live-key-row { + display: grid; + grid-template-columns: 1fr; + } + + .youtube-live-key-row .smallBtn { + width: 100%; + } + + .youtube-live-metrics { + grid-template-columns: 1fr; + } +} diff --git a/selfdrive/carrot/web/index.html b/selfdrive/carrot/web/index.html index ed811bb590..85c6bec7bf 100644 --- a/selfdrive/carrot/web/index.html +++ b/selfdrive/carrot/web/index.html @@ -95,11 +95,13 @@ + + @@ -620,6 +622,10 @@

Home