diff --git a/system/athena/athenad.py b/system/athena/athenad.py index 27a9afe535..a03a150330 100755 --- a/system/athena/athenad.py +++ b/system/athena/athenad.py @@ -40,6 +40,7 @@ from openpilot.system.loggerd.xattr_cache import getxattr, setxattr from openpilot.common.swaglog import cloudlog from openpilot.system.version import get_build_metadata from openpilot.system.hardware.hw import Paths +from openpilot.system.webrtc.webrtcd import StreamRequestBody ATHENA_HOST = os.getenv('ATHENA_HOST', 'wss://athena.comma.ai') @@ -84,6 +85,9 @@ UPLOAD_SESS = requests.Session() UPLOAD_SESS.mount("http://", UploadTOSAdapter()) UPLOAD_SESS.mount("https://", UploadTOSAdapter()) +WEBRTCD_SESS = requests.Session() +WEBRTCD_SESS.mount("http://", HTTPAdapter(max_retries=0)) + @dataclass class UploadFile: @@ -578,7 +582,6 @@ def getNetworks(): @dispatcher.add_method def startStream(sdp: str) -> dict: - from openpilot.system.webrtc.webrtcd import StreamRequestBody bridge_services_in = [] # get live car params to avoid stale notCar edge case @@ -590,7 +593,7 @@ def startStream(sdp: str) -> dict: body = StreamRequestBody(sdp, "wideRoad", bridge_services_in, ["carState"]) try: - resp = requests.post(f"http://localhost:{WEBRTCD_PORT}/stream", + resp = WEBRTCD_SESS.post(f"http://localhost:{WEBRTCD_PORT}/stream", json=asdict(body), timeout=10) if not resp.ok: try: @@ -600,7 +603,7 @@ def startStream(sdp: str) -> dict: resp.raise_for_status() return resp.json() except requests.ConnectTimeout as e: - raise Exception("webrtc took too long to respond. is it on?") from e + raise Exception("webrtc took too long to respond. is the comma body on?") from e except requests.ConnectionError as e: raise Exception("webrtc is not running. turn on comma body ignition.") from e @@ -608,8 +611,8 @@ def startStream(sdp: str) -> dict: def addIceCandidate(session_id: str, candidate: dict | None) -> dict: if session_id is None: return Exception("cannot add ice candidate without session_id") - resp = requests.post(f"http://localhost:{WEBRTCD_PORT}/candidate", - json={"session_id": session_id, "candidate": candidate}, timeout=10) + resp = WEBRTCD_SESS.post(f"http://localhost:{WEBRTCD_PORT}/candidate", + json={"session_id": session_id, "candidate": candidate}, timeout=10) return resp.json() @dispatcher.add_method diff --git a/system/webrtc/device/video.py b/system/webrtc/device/video.py index 8a07c5b9be..2083233dc9 100644 --- a/system/webrtc/device/video.py +++ b/system/webrtc/device/video.py @@ -4,6 +4,7 @@ import time import av from teleoprtc.tracks import TiciVideoStreamTrack +from aiortc import MediaStreamError from cereal import messaging from openpilot.common.realtime import DT_MDL, DT_DMON @@ -32,6 +33,10 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack): self._t0_ns = time.monotonic_ns() self.timing_sei_enabled = False + def stop(self) -> None: + super().stop() + self._sock = None + def _make_sock(self, camera_type: str) -> messaging.SubSocket: return messaging.sub_sock(self.camera_to_sock_mapping[camera_type], conflate=True) @@ -54,6 +59,8 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack): async def recv(self): while True: + if self.readyState != "live": + raise MediaStreamError msg = messaging.recv_one_or_none(self._sock) if msg is not None: break diff --git a/system/webrtc/webrtcd.py b/system/webrtc/webrtcd.py index e86b8f17d5..14c39fcd37 100755 --- a/system/webrtc/webrtcd.py +++ b/system/webrtc/webrtcd.py @@ -2,6 +2,7 @@ from abc import abstractmethod import os +import socket import time import argparse import asyncio @@ -26,6 +27,36 @@ from openpilot.system.webrtc.schema import generate_field from openpilot.common.params import Params from cereal import messaging, log +from aiortc import RTCBundlePolicy, RTCConfiguration, RTCPeerConnection +from aiortc.mediastreams import VideoStreamTrack +from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack +from teleoprtc import WebRTCAnswerBuilder + +import aioice.ice + + +# socket trick: route lookup for 8.8.8.8 (nothing is sent or actually connected to) +# return the source interfaces IP which is the default interface of the device +def _default_route_ip() -> str | None: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + s.connect(("8.8.8.8", 53)) # selects a route, sends nothing + return s.getsockname()[0] + except OSError: + return None + finally: + s.close() + +# aioice patch: gather ICE candidates only on the default-route interface +_get_host_addresses = aioice.ice.get_host_addresses +def _primary_host_addresses(use_ipv4: bool, use_ipv6: bool) -> list[str]: + addresses = _get_host_addresses(use_ipv4, use_ipv6) + primary = _default_route_ip() + if primary not in addresses: + return addresses + return [a for a in addresses if a == primary] +aioice.ice.get_host_addresses = _primary_host_addresses + class AsyncTaskRunner: def __init__(self): @@ -206,10 +237,6 @@ class StreamSession: shared_pub_master = DynamicPubMaster([]) def __init__(self, sdp: str, init_camera: str, incoming_services: list[str], outgoing_services: list[str], debug_mode: bool = False): - from aiortc.mediastreams import VideoStreamTrack - from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack - from teleoprtc import WebRTCAnswerBuilder - builder = WebRTCAnswerBuilder(sdp) self.video_track = LiveStreamVideoStreamTrack(init_camera) if not debug_mode else VideoStreamTrack() @@ -327,6 +354,8 @@ class StreamSession: await self.bitrate_controller.stop() if self.outgoing_bridge is not None: await self.outgoing_bridge.stop() + if self.video_track is not None: + self.video_track.stop() await self.stream.stop() @@ -353,7 +382,7 @@ async def get_stream(request: 'web.Request'): except Exception: pass await s.stop() - del stream_dict[sid] + stream_dict.pop(sid, None) session = StreamSession(body.sdp, body.initCamera, body.bridge_services_in, body.bridge_services_out, debug_mode) try: @@ -367,9 +396,14 @@ async def get_stream(request: 'web.Request'): except Exception: await session.stop() raise + stream_dict[session.identifier] = session session.start() - stream_dict[session.identifier] = session + session_id = session.identifier + + def remove_finished_session(_: asyncio.Task) -> None: + stream_dict.pop(session_id, None) + session.run_task.add_done_callback(remove_finished_session) return web.json_response({"sdp": answer.sdp, "type": answer.type, "session_id": session.identifier}) @@ -409,6 +443,26 @@ async def post_notify(request: 'web.Request'): return web.Response(status=200, text="OK") +async def on_startup(app: 'web.Application'): + logger = logging.getLogger("webrtcd") + start_time = time.monotonic() + pc = None + + # warmup imports for webrtc stack + try: + pc = RTCPeerConnection(RTCConfiguration(bundlePolicy=RTCBundlePolicy.MAX_BUNDLE)) + pc.addTransceiver("video", direction="recvonly") + pc.createDataChannel("data", ordered=True) + offer = await pc.createOffer() + await asyncio.wait_for(pc.setLocalDescription(offer), timeout=1.5) + logger.info("Warmed WebRTC stack in %.1f ms", (time.monotonic() - start_time) * 1000) + except Exception: + logger.exception("WebRTC stack warmup failed") + finally: + if pc is not None: + await pc.close() + + async def on_shutdown(app: 'web.Application'): for session in app['streams'].values(): await session.stop() @@ -426,6 +480,7 @@ def webrtcd_thread(host: str, port: int, debug: bool): app['streams'] = dict() app['stream_lock'] = asyncio.Lock() app['debug'] = debug + app.on_startup.append(on_startup) app.on_shutdown.append(on_shutdown) app.router.add_post("/stream", get_stream) app.router.add_post("/candidate", post_candidate)