try libdatachannel

openpilot/system/webrtc/device/video.py

@@ -4,7 +4,6 @@
 
 import av
 from teleoprtc.tracks import TiciVideoStreamTrack
-from aiortc import MediaStreamError
 
 from openpilot.cereal import messaging
 from openpilot.common.realtime import DT_MDL
@@ -55,6 +54,9 @@ def enable(self, enabled: bool):
     if not enabled:
       self._seen_keyframe = False
 
+  def request_keyframe(self) -> None:
+    self.params.put("LivestreamRequestKeyframe", True, block=False)
+
   def _build_frame_data(self, msg) -> bytes:
     encode_data = getattr(msg, msg.which())
     if not self.timing_sei_enabled:
@@ -71,9 +73,6 @@ def _build_frame_data(self, msg) -> bytes:
 
   async def recv(self):
     while True:
-      if self.readyState != "live":
-        raise MediaStreamError
-
       # while video is disabled, pause here without returning
       if not self.video_enabled:
         await asyncio.sleep(0.005)
@@ -95,6 +94,3 @@ async def recv(self):
     self.log_debug("track sending frame %d", self._pts)
 
     return packet
-
-  def codec_preference(self) -> str | None:
-    return "H264"

openpilot/system/webrtc/tests/test_stream_session.py

@@ -1,15 +1,10 @@
 import asyncio
 import json
 import time
-# for aiortc and its dependencies
-import warnings
-warnings.filterwarnings("ignore", category=DeprecationWarning)
-warnings.filterwarnings("ignore", category=RuntimeWarning) # TODO: remove this when google-crc32c publish a python3.12 wheel
 
-from aiortc import RTCDataChannel
-from aiortc.mediastreams import VIDEO_CLOCK_RATE, VIDEO_TIME_BASE
 import capnp
 from openpilot.cereal import messaging, log
+from teleoprtc.tracks import VIDEO_CLOCK_RATE, VIDEO_TIME_BASE
 
 from openpilot.system.webrtc.webrtcd import CerealOutgoingMessageProxy, CerealIncomingMessageProxy
 from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack
@@ -31,7 +26,8 @@ def test_outgoing_proxy(self, mocker):
     expected_dict = {"type": "customReservedRawData0", "logMonoTime": 123, "valid": True, "data": "test"}
     expected_json = json.dumps(expected_dict).encode()
 
-    channel = mocker.Mock(spec=RTCDataChannel)
+    channel = mocker.Mock()
+    channel.is_open.return_value = True
     proxy = CerealOutgoingMessageProxy(["customReservedRawData0"])
     def mocked_update(t):
       proxy.sm.update_msgs(0, [test_msg])
@@ -73,7 +69,6 @@ def test_livestream_track(self, mocker):
     track = LiveStreamVideoStreamTrack("driver")
 
     assert track.id.startswith("driver")
-    assert track.codec_preference() == "H264"
 
     for i in range(5):
       packet = self.loop.run_until_complete(track.recv())
@@ -83,4 +78,3 @@ def test_livestream_track(self, mocker):
         start_pts = packet.pts
       assert abs(i + packet.pts - (start_pts + (((time.monotonic_ns() - start_ns) * VIDEO_CLOCK_RATE) // 1_000_000_000))) < 450 #5ms
       assert packet.size == 0
-

openpilot/system/webrtc/webrtcd.py

@@ -1,9 +1,11 @@
 #!/usr/bin/env python3
 
 from abc import abstractmethod
+from collections.abc import Callable
 import os
 import socket
 import time
+import capnp
 import argparse
 import asyncio
 import contextlib
@@ -14,17 +16,7 @@
 import threading
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from urllib.parse import urlparse, parse_qs
-from typing import Any, TYPE_CHECKING
-
-# aiortc and its dependencies have lots of internal warnings :(
-import warnings
-warnings.filterwarnings("ignore", category=DeprecationWarning)
-warnings.filterwarnings("ignore", category=RuntimeWarning) # TODO: remove this when google-crc32c publish a python3.12 wheel
-
-import capnp
-if TYPE_CHECKING:
-  from aiortc.rtcdatachannel import RTCDataChannel
-import aioice.ice
+from typing import Any
 
 from openpilot.system.webrtc.helpers import StreamRequestBody
 from openpilot.system.webrtc.schema import generate_field
@@ -44,20 +36,8 @@ def _default_route_ip() -> str | 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 [primary, ]
-aioice.ice.get_host_addresses = _primary_host_addresses
-
-
 class AsyncTaskRunner:
   def __init__(self):
-    self.is_running = False
     self.task = None
     self.logger = logging.getLogger("webrtcd")
 
@@ -86,10 +66,10 @@ def __init__(self, services: list[str], enabled: bool = True):
     super().__init__()
     self.services = list(services)
     self.sm = messaging.SubMaster(self.services)
-    self.channels: list[RTCDataChannel] = []
+    self.channels = []
     self._enabled = enabled
 
-  def add_channel(self, channel: 'RTCDataChannel'):
+  def add_channel(self, channel):
     self.channels.append(channel)
 
   def enable(self, enable: bool):
@@ -118,20 +98,17 @@ def update(self):
       outgoing_msg = {"type": service, "logMonoTime": mono_time, "valid": valid, "data": msg_dict}
       encoded_msg = json.dumps(outgoing_msg).encode()
       for channel in self.channels:
+        if not channel.is_open():
+          continue
         channel.send(encoded_msg)
 
   async def run(self):
-    from aiortc.exceptions import InvalidStateError
-
     while True:
       if not self._enabled:
         await asyncio.sleep(0.01)
         continue
       try:
         self.update()
-      except InvalidStateError:
-        self.logger.warning("Cereal outgoing proxy invalid state (connection closed)")
-        break
       except Exception:
         self.logger.exception("Cereal outgoing proxy failure")
       await asyncio.sleep(0.01)
@@ -172,17 +149,17 @@ class LivestreamBitrateController(AsyncTaskRunner):
   high_level = 0.1 # drop immediately
   med_level = 0.05 # drop after # of samples
   low_level = 0 # raise after # of samples
-  down_samples = 5 # 1s
+  down_samples = 5
   param_name = "LivestreamEncoderBitrate"
 
-  def __init__(self, peer_connection: Any, params: Params, enabled: bool = True):
+  def __init__(self, get_stats: Callable[[], dict[str, Any]], params: Params, enabled: bool = True):
     super().__init__()
-    self.pc = peer_connection
+    self.get_stats = get_stats
     self.params = params
 
     self.level = 2
     self._publish(self.bitrates[self.level])
-    self.prev_lost, self.prev_sent = None, None
+    self.prev_stats: tuple[Any, ...] | None = None
     self.counter = 0
     self.up_samples = 5 # 1s
     self._auto = True
@@ -199,7 +176,7 @@ async def run(self):
       if not self._auto:
         continue
 
-      loss_rate = await self._sample()
+      loss_rate = self._sample()
       if loss_rate is None:
         continue
       if loss_rate >= self.med_level and self.level > 0:
@@ -216,22 +193,18 @@ async def run(self):
           self.counter = 0
           self._publish(self.bitrates[self.level])
 
-  async def _sample(self) -> float | None:
-    report = await self.pc.getStats()
-    packets_lost = packets_sent = 0
-    for s in report.values():
-      if s.type == "remote-inbound-rtp":
-        packets_lost += s.packetsLost
-      elif s.type == "outbound-rtp":
-        packets_sent += s.packetsSent
-
-    if self.prev_lost is None:
-      self.prev_lost, self.prev_sent = packets_lost, packets_sent
+  def _sample(self) -> float | None:
+    report = next(iter(self.get_stats().values()), None)
+    if report is None:
       return None
-    lost_delta = max(0, packets_lost - self.prev_lost)
-    sent_delta = max(0, packets_sent - self.prev_sent)
-    self.prev_lost, self.prev_sent = packets_lost, packets_sent
-    return lost_delta / sent_delta if sent_delta else 0.0
+
+    current = (report.ssrc, report.fraction_lost, report.packets_lost, report.highest_seq_no, report.jitter, report.lsr, report.dlsr)
+    if self.prev_stats == current:
+      return None
+    self.prev_stats = current
+
+    loss_rate = report.fraction_lost / 256
+    return loss_rate
 
   def _publish(self, bitrate: float):
     self.params.put(self.param_name, bitrate)
@@ -248,17 +221,15 @@ class StreamSession:
   shared_pub_master = DynamicPubMaster([])
 
   def __init__(self, body: StreamRequestBody, debug_mode: bool = False):
-    if debug_mode:
-      from aiortc.mediastreams import VideoStreamTrack
     from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack
     from teleoprtc.builder import WebRTCAnswerBuilder
 
     self.identifier = str(uuid.uuid4())
     self.params = Params()
-    builder = WebRTCAnswerBuilder(body.sdp)
+    builder = WebRTCAnswerBuilder(body.sdp, bind_address=_default_route_ip())
 
     self.enabled = body.enabled
-    self.video_track = LiveStreamVideoStreamTrack(body.init_camera, self.enabled) if not debug_mode else VideoStreamTrack()
+    self.video_track = LiveStreamVideoStreamTrack(body.init_camera, self.enabled)
     builder.add_video_stream(body.init_camera, self.video_track)
     self.stream = builder.stream()
 
@@ -270,7 +241,7 @@ def __init__(self, body: StreamRequestBody, debug_mode: bool = False):
       self.incoming_bridge = CerealIncomingMessageProxy(self.shared_pub_master)
     if len(body.bridge_services_out) > 0:
       self.outgoing_bridge = CerealOutgoingMessageProxy(body.bridge_services_out, self.enabled)
-    self.bitrate_controller = LivestreamBitrateController(self.stream.peer_connection, self.params, self.enabled)
+    self.bitrate_controller = LivestreamBitrateController(self.stream.get_receiver_report_stats, self.params, self.enabled)
 
     self.run_task: asyncio.Task | None = None
     self._cleanup_lock = asyncio.Lock()
@@ -305,13 +276,16 @@ def message_handler(self, message: bytes):
           case "livestreamCameraSwitch":
             self.video_track.switch_camera(payload["data"]["camera"])
           case "livestreamSettings":
-            self.bitrate_controller.set_quality(payload["data"]["quality"])
+            if self.bitrate_controller is not None:
+              self.bitrate_controller.set_quality(payload["data"]["quality"])
           case "livestreamVideoEnable":
             enabled = payload["data"]["enabled"]
             self.enabled = enabled
             self.video_track.enable(enabled)
-            self.outgoing_bridge.enable(enabled)
-            self.bitrate_controller.enable(enabled)
+            if self.outgoing_bridge is not None:
+              self.outgoing_bridge.enable(enabled)
+            if self.bitrate_controller is not None:
+              self.bitrate_controller.enable(enabled)
             if not enabled:
               self.params.put("LivestreamRequestKeyframe", True)
           case "clockSync":
@@ -325,7 +299,8 @@ def message_handler(self, message: bytes):
           case _:
             if payload.get("type") not in self.incoming_bridge_services:
               return
-            self.incoming_bridge.send(message)
+            if self.incoming_bridge is not None:
+              self.incoming_bridge.send(message)
     except Exception:
       self.logger.exception("Cereal incoming proxy failure")
 
@@ -341,7 +316,8 @@ async def run(self):
           channel = self.stream.get_messaging_channel()
           self.outgoing_bridge.add_channel(channel)
           self.outgoing_bridge.start()
-      self.bitrate_controller.start()
+      if self.bitrate_controller is not None:
+        self.bitrate_controller.start()
 
       self.logger.info("Stream session (%s) connected", self.identifier)
       await self.stream.wait_for_disconnection()
@@ -357,7 +333,8 @@ async def post_run_cleanup(self):
         return
       self._cleanup_done = True
       self.params.put("LivestreamRequestKeyframe", False)
-      await self.bitrate_controller.stop()
+      if self.bitrate_controller is not None:
+        await self.bitrate_controller.stop()
       if self.outgoing_bridge is not None:
         await self.outgoing_bridge.stop()
       if self.video_track is not None:
@@ -417,7 +394,12 @@ async def handle_get_stream(state: ServerState, raw_body: bytes) -> tuple[int, b
     session = StreamSession(body, debug_mode)
     stream_dict[session.identifier] = session
     try:
-      answer = await session.get_answer()
+      answer = await asyncio.wait_for(session.get_answer(), timeout=30)
+    except TimeoutError:
+      await session.stop()
+      stream_dict.pop(session.identifier, None)
+      logging.getLogger("webrtcd").exception("Timed out creating stream answer")
+      raise
     except Exception:
       await session.stop()
       stream_dict.pop(session.identifier, None)
@@ -558,9 +540,6 @@ async def _shutdown(server: WebrtcdHTTPServer, state: ServerState, loop: asyncio
 
 
 def prewarm_stream_session_imports(debug_mode: bool = False) -> None:
-  if debug_mode:
-    from aiortc.mediastreams import VideoStreamTrack
-    assert VideoStreamTrack
   from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack
   from teleoprtc.builder import WebRTCAnswerBuilder
   assert LiveStreamVideoStreamTrack

pyproject.toml

@@ -39,7 +39,7 @@ dependencies = [
 
   # body / webrtcd
   "av",
-  "aiortc",
+  "libdatachannel-py>=2026.1.0.dev2",
 
   # logging
   "pyzmq",