IQ.Pilot Release Commit @ f82ff4d

system/hardware/hardwared.py

@@ -21,7 +21,6 @@ from openpilot.common.realtime import DT_HW
 from openpilot.selfdrive.selfdrived.alertmanager import set_offroad_alert
 from openpilot.system.hardware import HARDWARE, TICI, AGNOS
 from openpilot.system.loggerd.config import get_available_percent
-from openpilot.system.statsd import statlog
 from openpilot.common.swaglog import cloudlog
 from openpilot.system.hardware.power_monitoring import PowerMonitoring, VBATT_LOW_POWER_EXIT
 from openpilot.system.hardware.fan_controller import FanController
@@ -479,11 +478,9 @@ def hardware_thread(end_event, hw_queue) -> None:
     msg.deviceState.offroadPowerUsageUwh = power_monitor.get_power_used()
     msg.deviceState.carBatteryCapacityUwh = max(0, power_monitor.get_car_battery_capacity())
     current_power_draw = HARDWARE.get_current_power_draw()
-    statlog.sample("power_draw", current_power_draw)
     msg.deviceState.powerDrawW = current_power_draw
 
     som_power_draw = HARDWARE.get_som_power_draw()
-    statlog.sample("som_power_draw", som_power_draw)
     msg.deviceState.somPowerDrawW = som_power_draw
 
     # FastSleep deep standby: shed heavy processes at low battery instead of shutting down,
@@ -523,23 +520,6 @@ def hardware_thread(end_event, hw_queue) -> None:
     msg.deviceState.thermalStatus = thermal_status
     pm.send("deviceState", msg)
 
-    # Log to statsd
-    statlog.gauge("free_space_percent", msg.deviceState.freeSpacePercent)
-    statlog.gauge("gpu_usage_percent", msg.deviceState.gpuUsagePercent)
-    statlog.gauge("memory_usage_percent", msg.deviceState.memoryUsagePercent)
-    for i, usage in enumerate(msg.deviceState.cpuUsagePercent):
-      statlog.gauge(f"cpu{i}_usage_percent", usage)
-    for i, temp in enumerate(msg.deviceState.cpuTempC):
-      statlog.gauge(f"cpu{i}_temperature", temp)
-    for i, temp in enumerate(msg.deviceState.gpuTempC):
-      statlog.gauge(f"gpu{i}_temperature", temp)
-    statlog.gauge("memory_temperature", msg.deviceState.memoryTempC)
-    for i, temp in enumerate(msg.deviceState.pmicTempC):
-      statlog.gauge(f"pmic{i}_temperature", temp)
-    for i, temp in enumerate(last_hw_state.modem_temps):
-      statlog.gauge(f"modem_temperature{i}", temp)
-    statlog.gauge("fan_speed_percent_desired", msg.deviceState.fanSpeedPercentDesired)
-    statlog.gauge("screen_brightness_percent", msg.deviceState.screenBrightnessPercent)
 
     # report to server once every 10 minutes
     rising_edge_started = should_start and not should_start_prev

system/hardware/power_monitoring.py

@@ -4,7 +4,6 @@ import threading
 from openpilot.common.params import Params
 from openpilot.system.hardware import HARDWARE
 from openpilot.common.swaglog import cloudlog
-from openpilot.system.statsd import statlog
 
 CAR_VOLTAGE_LOW_PASS_K = 0.011 # LPF gain for 45s tau (dt/tau / (dt/tau + 1))
 
@@ -56,7 +55,6 @@ class PowerMonitoring:
       # Low-pass battery voltage
       self.car_voltage_instant_mV = voltage
       self.car_voltage_mV = ((voltage * CAR_VOLTAGE_LOW_PASS_K) + (self.car_voltage_mV * (1 - CAR_VOLTAGE_LOW_PASS_K)))
-      statlog.gauge("car_voltage", self.car_voltage_mV / 1e3)
 
       # Cap the car battery power and save it in a param every 10-ish seconds
       self.car_battery_capacity_uWh = max(self.car_battery_capacity_uWh, 0)

system/hardware/tici/tests/test_power_draw.py

@@ -7,7 +7,7 @@ from tabulate import tabulate
 
 import cereal.messaging as messaging
 from cereal.services import SERVICE_LIST
-from opendbc.car.car_helpers import get_demo_car_params
+from iqdbc.car.car_helpers import get_demo_car_params
 from openpilot.common.mock import mock_messages
 from openpilot.common.params import Params
 from openpilot.system.hardware.tici.power_monitor import get_power

system/loggerd/config.py

@@ -5,9 +5,6 @@ from openpilot.system.hardware.hw import Paths
 CAMERA_FPS = 20
 SEGMENT_LENGTH = 60
 
-STATS_DIR_FILE_LIMIT = 10000
-STATS_SOCKET = "ipc:///tmp/stats"
-STATS_FLUSH_TIME_S = 60
 
 PATH_DICT = {
   "internal": Paths.log_root(),

system/manager/process_config.py

@@ -174,7 +174,6 @@ procs = [
   PythonProcess("tombstoned", "system.tombstoned", always_run, enabled=not PC),
   PythonProcess("updated", "system.updated.updated", and_(only_offroad, not_low_power), enabled=not PC),
   BundleProcess("iquploaderd", "iqpilot_hephaestusd_private", "iqpilot_private.konn3kt.uploaderd.iquploaderd", and_(iquploaderd_ready, not_low_power), restart_if_crash=True),
-  PythonProcess("statsd", "system.statsd", always_run),
   PythonProcess("feedbackd", "selfdrive.ui.feedback.feedbackd", and_(only_onroad, not_lat_maneuver)),
 
   # debug procs

system/statsd.py

@@ -1,234 +0,0 @@
-#!/usr/bin/env python3
-import base64
-import json
-import os
-from decimal import Decimal
-
-import zmq
-import time
-import uuid
-from pathlib import Path
-from collections import defaultdict
-from datetime import datetime, UTC, date
-from typing import NoReturn
-
-from openpilot.common.params import Params
-from cereal.messaging import SubMaster
-from openpilot.system.hardware.hw import Paths
-from openpilot.common.swaglog import cloudlog
-from openpilot.system.hardware import HARDWARE
-from openpilot.common.utils import atomic_write
-from openpilot.system.version import get_build_metadata
-from openpilot.system.loggerd.config import STATS_DIR_FILE_LIMIT, STATS_SOCKET, STATS_FLUSH_TIME_S
-
-
-class METRIC_TYPE:
-  GAUGE = 'g'
-  SAMPLE = 'sa'
-  RAW = 'r'
-
-
-class StatLog:
-  def __init__(self):
-    self.pid = None
-    self.zctx = None
-    self.sock = None
-    self.stats_socket = STATS_SOCKET
-
-  def connect(self) -> None:
-    self.zctx = zmq.Context.instance() or zmq.Context()
-    self.sock = self.zctx.socket(zmq.PUSH)
-    self.sock.setsockopt(zmq.LINGER, 10)
-    self.sock.connect(self.stats_socket)
-    self.pid = os.getpid()
-
-  def __del__(self):
-    if self.sock is not None:
-      self.sock.close()
-    if self.zctx is not None:
-      self.zctx.term()
-
-  def _send(self, metric: str) -> None:
-    if os.getpid() != self.pid:
-      self.connect()
-
-    try:
-      self.sock.send_string(metric, zmq.NOBLOCK)
-    except zmq.error.Again:
-      # drop :/
-      pass
-
-  def gauge(self, name: str, value: float) -> None:
-    self._send(f"{name}:{value}|{METRIC_TYPE.GAUGE}")
-
-  # Samples will be recorded in a buffer and at aggregation time,
-  # statistical properties will be logged (mean, count, percentiles, ...)
-  def sample(self, name: str, value: float):
-    self._send(f"{name}:{value}|{METRIC_TYPE.SAMPLE}")
-
-
-class StatLogIQ(StatLog):
-  def __init__(self, intercept=True):
-    """
-    Initializes the class instance with an optional parameter to determine
-    if statistical logging should be configured or not.
-
-    :param intercept: A boolean flag that indicates whether to initialize
-        the `comma_statlog`. If True, the `comma_statlog` attribute is
-        instantiated as a `StatLog` object. Defaults to True.
-    """
-    super().__init__()
-    self.comma_statlog = StatLog() if intercept else None
-    self.stats_socket = f"{STATS_SOCKET}_iq"
-
-  def connect(self) -> None:
-    super().connect()
-    if self.comma_statlog:
-      self.comma_statlog.connect()
-
-  def __del__(self):
-    super().__del__()
-    if self.comma_statlog:
-      self.comma_statlog.__del__()
-
-  def _send(self, metric: str) -> None:
-    super()._send(metric)
-    if self.comma_statlog:
-      self.comma_statlog._send(metric)
-
-  @staticmethod
-  def default_converter(obj):
-    if isinstance(obj, (datetime, date)):
-      return obj.isoformat()
-    if isinstance(obj, set):
-      return list(obj)
-    if isinstance(obj, Decimal):
-      return float(obj)
-    return str(obj)  # fallback for unknown types
-
-  def raw(self, name: str, value: dict) -> None:
-    encoded_dict = base64.b64encode(json.dumps(value, default=self.default_converter).encode("utf-8")).decode("utf-8")
-    self._send(f"{name}:{encoded_dict}|{METRIC_TYPE.RAW}")
-
-
-def main() -> NoReturn:
-  dongle_id = Params().get("DongleId")
-  def get_influxdb_line(measurement: str, value: float | dict[str, float],  timestamp: datetime, tags: dict) -> str:
-    res = f"{measurement}"
-    for k, v in tags.items():
-      res += f",{k}={str(v)}"
-    res += " "
-
-    if isinstance(value, float):
-      value = {'value': value}
-
-    for k, v in value.items():
-      res += f"{k}={v},"
-
-    res += f"dongle_id=\"{dongle_id}\" {int(timestamp.timestamp() * 1e9)}\n"
-    return res
-
-  # open statistics socket
-  ctx = zmq.Context.instance()
-  sock = ctx.socket(zmq.PULL)
-  sock.bind(STATS_SOCKET)
-
-  STATS_DIR = Paths.stats_root()
-
-  # initialize stats directory
-  Path(STATS_DIR).mkdir(parents=True, exist_ok=True)
-
-  build_metadata = get_build_metadata()
-
-  # initialize tags
-  tags = {
-    'started': False,
-    'version': build_metadata.openpilot.version,
-    'branch': build_metadata.channel,
-    'dirty': build_metadata.openpilot.is_dirty,
-    'origin': build_metadata.openpilot.git_normalized_origin,
-    'deviceType': HARDWARE.get_device_type(),
-  }
-
-  # subscribe to deviceState for started state
-  sm = SubMaster(['deviceState'])
-
-  idx = 0
-  boot_uid = str(uuid.uuid4())[:8]
-  last_flush_time = time.monotonic()
-  gauges = {}
-  samples: dict[str, list[float]] = defaultdict(list)
-  try:
-    while True:
-      started_prev = sm['deviceState'].started
-      sm.update()
-
-      # Update metrics
-      while True:
-        try:
-          metric = sock.recv_string(zmq.NOBLOCK)
-          try:
-            metric_type = metric.split('|')[1]
-            metric_name = metric.split(':')[0]
-            metric_value = float(metric.split('|')[0].split(':')[1])
-
-            if metric_type == METRIC_TYPE.GAUGE:
-              gauges[metric_name] = metric_value
-            elif metric_type == METRIC_TYPE.SAMPLE:
-              samples[metric_name].append(metric_value)
-            else:
-              cloudlog.event("unknown metric type", metric_type=metric_type)
-          except Exception:
-            cloudlog.event("malformed metric", metric=metric)
-        except zmq.error.Again:
-          break
-
-      # flush when started state changes or after FLUSH_TIME_S
-      if (time.monotonic() > last_flush_time + STATS_FLUSH_TIME_S) or (sm['deviceState'].started != started_prev):
-        result = ""
-        current_time = datetime.now(UTC)
-        tags['started'] = sm['deviceState'].started
-
-        for key, value in gauges.items():
-          result += get_influxdb_line(f"gauge.{key}", value, current_time, tags)
-
-        for key, values in samples.items():
-          values.sort()
-          sample_count = len(values)
-          sample_sum = sum(values)
-
-          stats = {
-            'count': sample_count,
-            'min': values[0],
-            'max': values[-1],
-            'mean': sample_sum / sample_count,
-          }
-          for percentile in [0.05, 0.5, 0.95]:
-            value = values[int(round(percentile * (sample_count - 1)))]
-            stats[f"p{int(percentile * 100)}"] = value
-
-          result += get_influxdb_line(f"sample.{key}", stats, current_time, tags)
-
-        # clear intermediate data
-        gauges.clear()
-        samples.clear()
-        last_flush_time = time.monotonic()
-
-        # check that we aren't filling up the drive
-        if len(os.listdir(STATS_DIR)) < STATS_DIR_FILE_LIMIT:
-          if len(result) > 0:
-            stats_path = os.path.join(STATS_DIR, f"{boot_uid}_{idx}")
-            with atomic_write(stats_path) as f:
-              f.write(result)
-            idx += 1
-        else:
-          cloudlog.error("stats dir full")
-  finally:
-    sock.close()
-    ctx.term()
-
-
-if __name__ == "__main__":
-  main()
-else:
-  statlog = StatLogIQ(intercept=True)

system/ui/lib/scroll_panel.py

@@ -41,12 +41,10 @@ class GuiScrollPanel:
     if DEBUG:
       rl.draw_rectangle_lines(0, 0, abs(int(self._velocity_filter_y.x)), 10, rl.RED)
 
-    # Handle mouse wheel only when the mouse cursor is over this panel
-    mouse_wheel = rl.get_mouse_wheel_move()
-    if mouse_wheel != 0:
-      mouse_pos = rl.get_mouse_position()
-      if rl.check_collision_point_rec(mouse_pos, bounds):
-        self._offset_filter_y.x += mouse_wheel * MOUSE_WHEEL_SCROLL_SPEED
+    # wheel scrolls this panel only while the cursor sits inside its bounds
+    wheel = rl.get_mouse_wheel_move()
+    if wheel and rl.check_collision_point_rec(rl.get_mouse_position(), bounds):
+      self._offset_filter_y.x += wheel * MOUSE_WHEEL_SCROLL_SPEED
 
     max_scroll_distance = max(0, content.height - bounds.height)
     if self._scroll_state == ScrollState.IDLE:

system/webrtc/device/audio_ldc.py

@@ -0,0 +1,124 @@
+import asyncio
+import contextlib
+from collections import deque
+from fractions import Fraction
+
+import av
+
+from cereal import messaging
+from openpilot.selfdrive.ui.soundd import SAMPLE_RATE as SOUND_SAMPLE_RATE
+from openpilot.system.webrtc.device.audio import WEBRTC_AUDIO_PTIME, WEBRTC_AUDIO_SERVICE
+
+
+class AudioInputOpusProducer:
+  """Micd PCM -> 48 kHz Opus payloads for libdatachannel's RTP packetizer."""
+
+  def __init__(self) -> None:
+    self._sock = messaging.sub_sock("rawAudioData", conflate=False)
+    self._pcm = bytearray()
+    self._source_rate = 16_000
+    self._next_pts = 0
+    self._pending: deque[tuple[bytes, int]] = deque()
+    self._enabled = True
+    self._resampler = av.AudioResampler(format="fltp", layout="mono", rate=48_000)
+    self._encoder = av.CodecContext.create("libopus", "w")
+    self._encoder.sample_rate = 48_000
+    self._encoder.layout = "mono"
+    self._encoder.format = "fltp"
+    self._encoder.open()
+
+  def enable(self, enabled: bool) -> None:
+    self._enabled = enabled
+
+  async def _read_pcm_frame(self) -> av.AudioFrame:
+    while True:
+      samples = max(1, int(WEBRTC_AUDIO_PTIME * self._source_rate))
+      target_bytes = samples * 2
+      while len(self._pcm) < target_bytes:
+        msg = messaging.recv_one_or_none(self._sock)
+        if msg is None:
+          await asyncio.sleep(0.002)
+          continue
+        audio = msg.rawAudioData
+        rate = int(audio.sampleRate) or self._source_rate
+        if rate != self._source_rate:
+          self._source_rate = rate
+          self._pcm.clear()
+          continue
+        self._pcm.extend(bytes(audio.data))
+
+      data = bytes(self._pcm[:target_bytes])
+      del self._pcm[:target_bytes]
+      frame = av.AudioFrame(format="s16", layout="mono", samples=samples)
+      frame.planes[0].update(data)
+      frame.sample_rate = self._source_rate
+      return frame
+
+  async def recv(self) -> tuple[bytes, int] | None:
+    if not self._enabled:
+      await asyncio.sleep(WEBRTC_AUDIO_PTIME)
+      return None
+    while not self._pending:
+      source_frame = await self._read_pcm_frame()
+      for frame in self._resampler.resample(source_frame):
+        frame.pts = self._next_pts
+        frame.time_base = Fraction(1, 48_000)
+        self._next_pts += frame.samples
+        for packet in self._encoder.encode(frame):
+          self._pending.append((bytes(packet), int(packet.pts or 0)))
+    return self._pending.popleft()
+
+
+class IncomingOpusCerealProxy:
+  """libdatachannel Opus payloads -> soundd-compatible PCM cereal messages."""
+
+  def __init__(self, track) -> None:
+    self._loop = asyncio.get_running_loop()
+    self._queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=32)
+    self._pm = messaging.PubMaster([WEBRTC_AUDIO_SERVICE])
+    self._decoder = av.CodecContext.create("opus", "r")
+    self._resampler = av.AudioResampler(format="s16", layout="mono", rate=SOUND_SAMPLE_RATE)
+    self._task: asyncio.Task | None = None
+    track.on_frame(self._on_frame)
+
+  def _on_frame(self, payload: bytes, _info) -> None:
+    def enqueue() -> None:
+      if self._queue.full():
+        with contextlib.suppress(asyncio.QueueEmpty):
+          self._queue.get_nowait()
+      self._queue.put_nowait(bytes(payload))
+    self._loop.call_soon_threadsafe(enqueue)
+
+  def start(self) -> None:
+    if self._task is None:
+      self._task = asyncio.create_task(self.run())
+
+  async def stop(self) -> None:
+    if self._task is None:
+      return
+    self._task.cancel()
+    try:
+      await self._task
+    except asyncio.CancelledError:
+      pass
+    self._task = None
+
+  def _publish(self, frame: av.AudioFrame) -> None:
+    data = frame.to_ndarray().tobytes()
+    if not data:
+      return
+    msg = messaging.new_message(WEBRTC_AUDIO_SERVICE, valid=True)
+    msg.webrtcAudioData.data = data
+    msg.webrtcAudioData.sampleRate = frame.sample_rate
+    self._pm.send(WEBRTC_AUDIO_SERVICE, msg)
+
+  async def run(self) -> None:
+    while True:
+      payload = await self._queue.get()
+      try:
+        for decoded in self._decoder.decode(av.Packet(payload)):
+          for frame in self._resampler.resample(decoded):
+            self._publish(frame)
+      except Exception:
+        # A malformed or stale packet must not end the video/control session.
+        continue

system/webrtc/device/video.py

@@ -86,6 +86,9 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
       self._kf_requested = True
     self._request_keyframe(True)
 
+  def request_keyframe(self) -> None:
+    self._mark_keyframe_needed()
+
   def _mark_keyframe_received(self) -> None:
     """This track got its keyframe; only clear the global request once no track needs one."""
     if self._kf_requested:

system/webrtc/device/video_ldc.py

@@ -0,0 +1,274 @@
+import asyncio
+import logging
+import struct
+import time
+
+import av
+from openpilot.system.webrtc.teleoprtc_ldc.tracks import TiciVideoStreamTrack
+
+from cereal import messaging
+from openpilot.common.params import Params
+from openpilot.common.realtime import DT_MDL, DT_DMON
+
+# Arbitrary 16-byte UUID identifying konn3kt frame-timing SEI messages. When timing
+# telemetry is enabled, each frame carries a user_data_unregistered SEI NAL with four
+# big-endian doubles (ms): encode duration, IPC/queue delay, host transit, and the
+# device wall clock. The client decodes these to compute true glass-to-glass latency.
+TIMING_SEI_UUID = bytes([
+  0xa5, 0xe0, 0xc4, 0xa4, 0x5b, 0x6e, 0x4e, 0x1e,
+  0x9c, 0x7e, 0x12, 0x34, 0x56, 0x78, 0x9a, 0xbc,
+])
+# Annex-B start code + SEI NAL (type 6) + user_data_unregistered (type 5) + payload size
+# (0x30 = 48 bytes = 16 UUID + 32 data). Trailing 0x80 is the RBSP stop bit.
+_SEI_PREFIX = b'\x00\x00\x00\x01\x06\x05\x30' + TIMING_SEI_UUID
+
+
+class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
+  livestream_camera_to_sock_mapping = {
+    "driver": "livestreamDriverEncodeData",
+    "wideRoad": "livestreamWideRoadEncodeData",
+    "road": "livestreamRoadEncodeData",
+  }
+  main_camera_to_sock_mapping = {
+    "driver": "driverEncodeData",
+    "wideRoad": "wideRoadEncodeData",
+    "road": "roadEncodeData",
+  }
+
+  # Number of live tracks still waiting for their first keyframe. The on-demand
+  # keyframe request (LivestreamRequestKeyframe) is a single global param honored by
+  # every encoder, so with multiple concurrent tracks (dual-camera PiP) we must not
+  # clear it until *all* tracks have received an IDR — otherwise the first track to
+  # get its keyframe clears the request and starves the others (black feed).
+  _kf_pending_count = 0
+
+  def __init__(self, camera_type: str):
+    dt = DT_DMON if camera_type == "driver" else DT_MDL
+    super().__init__(camera_type, dt)
+
+    self._params = Params()
+    self._camera_type = camera_type
+    self._candidate_topics = [
+      self.main_camera_to_sock_mapping[camera_type],
+      self.livestream_camera_to_sock_mapping[camera_type],
+    ]
+    self._socks = {topic: messaging.sub_sock(topic, conflate=True) for topic in self._candidate_topics}
+    self._active_topic = self._preferred_topics()[0]
+    self._pts = 0
+    self._t0_ns = time.monotonic_ns()
+    self._cached_header: bytes = b""
+    self._sent_keyframe = False
+    self._kf_requested = False  # whether this track counts toward _kf_pending_count
+    self._frame_count = 0
+    self._last_frame_time = 0.0
+    self._last_preference_refresh = 0.0
+    # Tracks how long the H264 livestream feed has been silent, to gate the last-resort main-feed
+    # fallback (see recv) without flapping between sources frame-by-frame.
+    self._live_silent_since: float | None = None
+    # Opt-in glass-to-glass latency telemetry (toggled by the client over the data channel).
+    self.timing_sei_enabled = False
+    self._logger = logging.getLogger("LiveStreamVideoStreamTrack")
+
+    # Ask the encoder for an immediate IDR so the stream starts fast instead of waiting up to a full
+    # GOP for the next periodic keyframe (encoderd honors LivestreamRequestKeyframe per-frame).
+    self._mark_keyframe_needed()
+
+  def _request_keyframe(self, enabled: bool) -> None:
+    try:
+      self._params.put_bool("LivestreamRequestKeyframe", enabled)
+    except Exception:
+      self._logger.exception("failed to set LivestreamRequestKeyframe")
+
+  def _mark_keyframe_needed(self) -> None:
+    """This track needs (another) keyframe: keep the global request asserted."""
+    if not self._kf_requested:
+      LiveStreamVideoStreamTrack._kf_pending_count += 1
+      self._kf_requested = True
+    self._request_keyframe(True)
+
+  def request_keyframe(self) -> None:
+    """RTCP PLI hook used by libdatachannel's native H.264 packetizer."""
+    self._mark_keyframe_needed()
+
+  def _mark_keyframe_received(self) -> None:
+    """This track got its keyframe; only clear the global request once no track needs one."""
+    if self._kf_requested:
+      self._kf_requested = False
+      LiveStreamVideoStreamTrack._kf_pending_count = max(0, LiveStreamVideoStreamTrack._kf_pending_count - 1)
+    if LiveStreamVideoStreamTrack._kf_pending_count == 0:
+      self._request_keyframe(False)
+
+  def stop(self):
+    # Release our pending-keyframe hold so a torn-down track that never received an
+    # IDR doesn't pin LivestreamRequestKeyframe True forever (continuous keyframes).
+    if getattr(self, "_kf_requested", False):
+      self._kf_requested = False
+      LiveStreamVideoStreamTrack._kf_pending_count = max(0, LiveStreamVideoStreamTrack._kf_pending_count - 1)
+    try:
+      super().stop()
+    except Exception:
+      pass
+
+  def switch_camera(self, camera_type: str) -> None:
+    """Repoint this track at a different camera without renegotiating the peer connection.
+
+    Lets a single video track back the whole Live View — the client flips cameras over the
+    data channel and we swap the source here, instead of uplinking every camera at once."""
+    if camera_type not in self.livestream_camera_to_sock_mapping:
+      self._logger.warning("[%s] ignoring switch to unknown camera %s", self._id, camera_type)
+      return
+    if camera_type == self._camera_type:
+      return
+    self._logger.info("[%s] switching camera %s -> %s", self._id, self._camera_type, camera_type)
+    self._camera_type = camera_type
+    self._candidate_topics = [
+      self.main_camera_to_sock_mapping[camera_type],
+      self.livestream_camera_to_sock_mapping[camera_type],
+    ]
+    self._socks = {topic: messaging.sub_sock(topic, conflate=True) for topic in self._candidate_topics}
+    self._active_topic = self._preferred_topics()[0]
+    # Force a fresh keyframe/header before emitting frames from the new source, and ask the encoder
+    # for an immediate IDR so the camera switch isn't stalled waiting for the next periodic keyframe.
+    self._cached_header = b""
+    self._sent_keyframe = False
+    self._last_preference_refresh = 0.0
+    self._live_silent_since = None
+    self._mark_keyframe_needed()
+
+  def _preferred_topics(self) -> list[str]:
+    # WebRTC currently forces H.264. The dedicated livestream topics are the H.264 feeds,
+    # while the main encode topics are the full-resolution HEVC recordings. Prefer the
+    # livestream feeds both onroad and offroad, and keep the main topics only as fallback.
+    return [
+      self.livestream_camera_to_sock_mapping[self._camera_type],
+      self.main_camera_to_sock_mapping[self._camera_type],
+    ]
+
+  def _reset_decoder_state(self, topic: str) -> None:
+    if topic == self._active_topic:
+      return
+    self._logger.info("[%s] switching video source from %s to %s", self._id, self._active_topic, topic)
+    self._active_topic = topic
+    self._cached_header = b""
+    self._sent_keyframe = False
+
+  def _timing_sei(self, evta, log_mono_time: int) -> bytes:
+    """Build a timing SEI NAL from encode metadata, or empty bytes when disabled."""
+    if not self.timing_sei_enabled:
+      return b""
+    idx = evta.idx
+    return _SEI_PREFIX + struct.pack(
+      '>4d',
+      (idx.timestampEof - idx.timestampSof) / 1e6,   # encode duration (ms)
+      (log_mono_time - idx.timestampEof) / 1e6,       # IPC/queue delay (ms)
+      (time.monotonic_ns() - log_mono_time) / 1e6,    # host transit so far (ms)
+      time.time() * 1000,                             # device wall clock (ms)  # noqa: TID251
+    ) + b'\x80'
+
+  def _is_keyframe(self, data: bytes) -> bool:
+    """Check if H.264 NAL unit contains an IDR keyframe (NAL type 5)."""
+    i = 0
+    while i < len(data) - 4:
+      # Look for Annex B start codes: 0x000001 or 0x00000001
+      if data[i:i+3] == b'\x00\x00\x01':
+        nal_type = data[i+3] & 0x1f
+        if nal_type == 5:  # IDR slice
+          return True
+        i += 3
+      elif data[i:i+4] == b'\x00\x00\x00\x01':
+        nal_type = data[i+4] & 0x1f
+        if nal_type == 5:  # IDR slice
+          return True
+        i += 4
+      else:
+        i += 1
+    return False
+
+  async def recv(self):
+    while True:
+      now = time.monotonic()
+      # Resolve topics each iteration: a camera switch (different async task) can rebuild self._socks
+      # across the await below, so a value cached before the loop would index a stale key (KeyError).
+      live_topic = self.livestream_camera_to_sock_mapping[self._camera_type]
+      main_topic = self.main_camera_to_sock_mapping[self._camera_type]
+      # Lock onto the dedicated H264 livestream feed. Onroad the HEVC main feed also publishes at
+      # 20fps; eagerly preferring whichever socket had a frame ready raced frame-by-frame, reset the
+      # decoder every frame, and (the track is negotiated H264) shoved HEVC garbage into the stream —
+      # the onroad choppiness. Only fall back to the main feed as a last resort after a long
+      # livestream silence (e.g. stream_encoderd still spinning up), and snap back when it returns.
+      msg = messaging.recv_one_or_none(self._socks[live_topic])
+      if msg is not None:
+        self._reset_decoder_state(live_topic)
+        self._last_frame_time = now
+        self._live_silent_since = None
+        break
+
+      if self._live_silent_since is None:
+        self._live_silent_since = now
+      elif now - self._live_silent_since > 3.0:
+        maybe_msg = messaging.recv_one_or_none(self._socks[main_topic])
+        if maybe_msg is not None:
+          self._reset_decoder_state(main_topic)
+          self._last_frame_time = now
+          msg = maybe_msg
+          break
+
+      await asyncio.sleep(0.005)
+
+    evta = getattr(msg, msg.which())
+
+    header = bytes(evta.header)
+    data = bytes(evta.data)
+    self._frame_count += 1
+
+    # Cache SPS/PPS header when it arrives
+    if header:
+      self._cached_header = header
+      self._logger.debug(f"[{self._id}] cached SPS/PPS header ({len(header)} bytes)")
+
+    # CRITICAL: Cannot decode without SPS/PPS. Wait for it.
+    if not self._cached_header:
+      self._logger.debug(f"[{self._id}] frame {self._frame_count}: no SPS/PPS yet, skipping")
+      return await self.recv()
+
+    is_keyframe = self._is_keyframe(data)
+
+    # Wait for first keyframe before sending any frames
+    # Browser decoder needs IDR to initialize properly
+    if not self._sent_keyframe:
+      if not is_keyframe:
+        self._logger.debug(f"[{self._id}] frame {self._frame_count}: waiting for keyframe")
+        return await self.recv()
+      self._sent_keyframe = True
+      # Got the IDR we asked for — stop nagging the encoder, but only once every
+      # concurrent track has its keyframe (multi-track PiP shares the global param).
+      self._mark_keyframe_received()
+      self._logger.info(f"[{self._id}] first keyframe received, starting stream")
+
+    # Optional timing SEI NAL, inserted before the slice data (and after SPS/PPS on keyframes).
+    sei_nal = self._timing_sei(evta, msg.logMonoTime)
+
+    # Prepend SPS/PPS header to keyframes (required by some decoders)
+    # For non-keyframes, header is optional but safe to include
+    if is_keyframe:
+      payload = self._cached_header + sei_nal + data
+    else:
+      payload = sei_nal + data
+
+    self._pts = ((time.monotonic_ns() - self._t0_ns) * self._clock_rate) // 1_000_000_000
+
+    packet = av.Packet(payload)
+    packet.time_base = self._time_base
+    packet.pts = int(self._pts)
+    packet.dts = int(self._pts)
+    packet.duration = int(self._dt * self._clock_rate)
+
+    if is_keyframe:
+      packet.is_keyframe = True
+
+    self.log_debug("track sending frame %s (keyframe=%s, size=%d)", self._pts, is_keyframe, len(payload))
+
+    return packet
+
+  def codec_preference(self) -> str | None:
+    return "H264"

system/webrtc/teleoprtc_ldc/__init__.py

@@ -0,0 +1,2 @@
+from .builder import WebRTCOfferBuilder, WebRTCAnswerBuilder  # noqa
+from .stream import WebRTCBaseStream, StreamingOffer, ConnectionProvider, MessageHandler  # noqa

system/webrtc/teleoprtc_ldc/builder.py

@@ -0,0 +1,89 @@
+# ruff: noqa: TID251, UP006, UP035
+
+import abc
+from typing import Dict, List, Optional
+
+from .stream import RTCSessionDescription, WebRTCBaseStream, WebRTCOfferStream, WebRTCAnswerStream, ConnectionProvider
+from .tracks import TiciVideoStreamTrack, TiciTrackWrapper
+
+
+class WebRTCStreamBuilder(abc.ABC):
+  @abc.abstractmethod
+  def stream(self) -> WebRTCBaseStream:
+    raise NotImplementedError
+
+
+class WebRTCOfferBuilder(WebRTCStreamBuilder):
+  def __init__(self, connection_provider: ConnectionProvider, bind_address: Optional[str] = None,
+               ice_servers: Optional[List[dict]] = None):
+    self.connection_provider = connection_provider
+    self.bind_address = bind_address
+    self.ice_servers = ice_servers
+    self.requested_camera_types: List[str] = []
+    self.requested_audio = False
+    self.audio_tracks: List[object] = []
+    self.messaging_enabled = False
+
+  def offer_to_receive_video_stream(self, camera_type: str):
+    assert camera_type in ["driver", "wideRoad", "road"]
+    self.requested_camera_types.append(camera_type)
+
+  def offer_to_receive_audio_stream(self):
+    self.requested_audio = True
+
+  def add_audio_stream(self, track: object):
+    assert len(self.audio_tracks) == 0
+    self.audio_tracks = [track]
+
+  def add_messaging(self):
+    self.messaging_enabled = True
+
+  def stream(self) -> WebRTCBaseStream:
+    return WebRTCOfferStream(
+      self.connection_provider,
+      consumed_camera_types=self.requested_camera_types,
+      consume_audio=self.requested_audio,
+      video_producer_tracks=[],
+      audio_producer_tracks=self.audio_tracks,
+      should_add_data_channel=self.messaging_enabled,
+      bind_address=self.bind_address,
+      ice_servers=self.ice_servers,
+    )
+
+
+class WebRTCAnswerBuilder(WebRTCStreamBuilder):
+  def __init__(self, offer_sdp: str, bind_address: Optional[str] = None,
+               ice_servers: Optional[List[dict]] = None):
+    self.offer_sdp = offer_sdp
+    self.bind_address = bind_address
+    self.ice_servers = ice_servers
+    self.video_tracks: Dict[str, TiciVideoStreamTrack] = {}
+    self.requested_audio = False
+    self.audio_tracks: List[object] = []
+
+  def offer_to_receive_audio_stream(self):
+    self.requested_audio = True
+
+  def add_video_stream(self, camera_type: str, track: object):
+    assert camera_type not in self.video_tracks
+    assert camera_type in ["driver", "wideRoad", "road"]
+    if not isinstance(track, TiciVideoStreamTrack):
+      track = TiciTrackWrapper(camera_type, track)
+    self.video_tracks[camera_type] = track
+
+  def add_audio_stream(self, track: object):
+    assert len(self.audio_tracks) == 0
+    self.audio_tracks = [track]
+
+  def stream(self) -> WebRTCBaseStream:
+    description = RTCSessionDescription(sdp=self.offer_sdp, type="offer")
+    return WebRTCAnswerStream(
+      description,
+      consumed_camera_types=[],
+      consume_audio=self.requested_audio,
+      video_producer_tracks=list(self.video_tracks.values()),
+      audio_producer_tracks=self.audio_tracks,
+      should_add_data_channel=False,
+      bind_address=self.bind_address,
+      ice_servers=self.ice_servers,
+    )

system/webrtc/teleoprtc_ldc/decoder.py

@@ -0,0 +1,52 @@
+# ruff: noqa: UP006, UP035
+
+import dataclasses
+import struct
+from typing import List
+
+
+@dataclasses.dataclass(frozen=True)
+class RtcpReceiverReport:
+  ssrc: int
+  fraction_lost: int
+  packets_lost: int
+  highest_seq_no: int
+  jitter: int
+  lsr: int
+  dlsr: int
+
+
+def _decode_receiver_reports(message: bytes) -> List[RtcpReceiverReport]:
+  reports: List[RtcpReceiverReport] = []
+  offset = 0
+
+  while offset + 4 <= len(message):
+    flags, packet_type, length_words = struct.unpack_from("!BBH", message, offset)
+    packet_end = offset + (length_words + 1) * 4
+    if flags >> 6 != 2 or packet_end > len(message):
+      break
+
+    report_count = flags & 0x1F
+    if packet_type == 200:  # Sender Report
+      report_offset = offset + 28
+    elif packet_type == 201:  # Receiver Report
+      report_offset = offset + 8
+    else:
+      offset = packet_end
+      continue
+
+    if report_offset + report_count * 24 > packet_end:
+      break
+
+    for i in range(report_count):
+      block_offset = report_offset + i * 24
+      ssrc, loss, highest_seq_no, jitter, lsr, dlsr = struct.unpack_from("!IIIIII", message, block_offset)
+      fraction_lost = loss >> 24
+      packets_lost = loss & 0xFFFFFF
+      if packets_lost & 0x800000:
+        packets_lost -= 1 << 24
+      reports.append(RtcpReceiverReport(ssrc, fraction_lost, packets_lost, highest_seq_no, jitter, lsr, dlsr))
+
+    offset = packet_end
+
+  return reports

system/webrtc/teleoprtc_ldc/info.py

@@ -0,0 +1,37 @@
+import dataclasses
+
+from libdatachannel import Description
+
+
+@dataclasses.dataclass
+class StreamingMediaInfo:
+  n_expected_camera_tracks: int
+  expected_audio_track: bool
+  incoming_audio_track: bool
+  incoming_datachannel: bool
+
+
+def parse_info_from_offer(sdp: str) -> StreamingMediaInfo:
+  """
+  helper function to parse info about outgoing and incoming streams from an offer sdp
+  """
+  desc = Description(sdp, Description.Type.Offer)
+  n_video = 0
+  expected_audio_track = False
+  incoming_audio_track = False
+  incoming_datachannel = desc.has_application()
+
+  for i in range(desc.media_count()):
+    media = desc.media(i)
+    if media is None:
+      continue
+    direction = media.direction()
+    if media.type() == "video" and direction in (Description.Direction.RecvOnly, Description.Direction.SendRecv):
+      n_video += 1
+    elif media.type() == "audio":
+      if direction in (Description.Direction.RecvOnly, Description.Direction.SendRecv):
+        expected_audio_track = True
+      if direction in (Description.Direction.SendOnly, Description.Direction.SendRecv):
+        incoming_audio_track = True
+
+  return StreamingMediaInfo(n_video, expected_audio_track, incoming_audio_track, incoming_datachannel)

system/webrtc/teleoprtc_ldc/stream.py

@@ -0,0 +1,530 @@
+# ruff: noqa: TID251, UP006, UP035
+
+import abc
+import asyncio
+import contextlib
+import dataclasses
+import logging
+import random
+from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple, Union
+
+from libdatachannel import (
+  Configuration,
+  DataChannel,
+  Candidate,
+  Description,
+  FrameInfo,
+  H264RtpPacketizer,
+  IceServer,
+  NalUnit,
+  OpusRtpDepacketizer,
+  OpusRtpPacketizer,
+  PeerConnection,
+  PliHandler,
+  RtcpNackResponder,
+  RtcpSrReporter,
+  RtpPacketizationConfig,
+  Track,
+)
+
+from .decoder import RtcpReceiverReport, _decode_receiver_reports
+from .tracks import TiciVideoStreamTrack, parse_video_track_id
+
+
+@dataclasses.dataclass
+class StreamingOffer:
+  sdp: str
+  video: List[str]
+
+
+@dataclasses.dataclass
+class RTCSessionDescription:
+  sdp: str
+  type: str
+
+
+ConnectionProvider = Callable[[StreamingOffer], Awaitable[RTCSessionDescription]]
+MessageHandler = Callable[[Union[bytes, str]], None]
+
+
+class WebRTCBaseStream(abc.ABC):
+  def __init__(self,
+               consumed_camera_types: List[str],
+               consume_audio: bool,
+               video_producer_tracks: List[TiciVideoStreamTrack],
+               audio_producer_tracks: List[Any],
+               should_add_data_channel: bool,
+               bind_address: Optional[str] = None,
+               ice_servers: Optional[List[dict]] = None):
+    config = Configuration()
+    config.force_media_transport = True
+    config.disable_auto_negotiation = True
+    config.ice_servers = self._make_ice_servers(ice_servers)
+    if bind_address is not None:
+      config.bind_address = bind_address
+
+    self.peer_connection = PeerConnection(config)
+    self.expected_incoming_camera_types = consumed_camera_types
+    self.expected_incoming_audio = consume_audio
+    self.expected_number_of_incoming_media: Optional[int] = None
+
+    self.incoming_camera_tracks: Dict[str, Any] = {}
+    self.incoming_audio_tracks: List[Any] = []
+    self.outgoing_video_tracks = video_producer_tracks
+    self.outgoing_audio_tracks = audio_producer_tracks
+
+    self.should_add_data_channel = should_add_data_channel
+    self.messaging_channel: Optional[DataChannel] = None
+    self.incoming_message_handlers: List[MessageHandler] = []
+    self._consumer_tracks: List[Track] = []
+    self._sender_tasks: List[asyncio.Task] = []
+    self._track_state: List[Tuple[Track, TiciVideoStreamTrack, RtpPacketizationConfig]] = []
+    self._audio_track_state: List[Tuple[Track, Any, RtpPacketizationConfig]] = []
+    self._receiver_reports: Dict[str, RtcpReceiverReport] = {}
+    self._receiver_report_tracks: Dict[str, Tuple[Track, int]] = {}
+
+    self.incoming_media_ready_event = asyncio.Event()
+    self.messaging_channel_ready_event = asyncio.Event()
+    self.connection_attempted_event = asyncio.Event()
+    self.connection_stopped_event = asyncio.Event()
+    self.gathering_complete_event = asyncio.Event()
+    self._loop: Optional[asyncio.AbstractEventLoop] = None
+
+    self.peer_connection.on_state_change(self._on_connectionstatechange)
+    self.peer_connection.on_gathering_state_change(self._on_gatheringstatechange)
+    self.peer_connection.on_data_channel(self._on_incoming_datachannel)
+    if self.expected_incoming_camera_types or self.expected_incoming_audio:
+      self.peer_connection.on_track(self._on_incoming_track)
+
+    self.logger = logging.getLogger("WebRTCStream")
+
+  @staticmethod
+  def _make_ice_servers(servers: Optional[List[dict]]) -> List[IceServer]:
+    """Preserve Konn3kt's authenticated STUN/TURN configuration in libdatachannel."""
+    parsed: List[IceServer] = []
+    for server in servers or []:
+      urls = server.get("urls", []) if isinstance(server, dict) else []
+      if isinstance(urls, str):
+        urls = [urls]
+      username = str(server.get("username") or "")
+      credential = str(server.get("credential") or "")
+      for url in urls:
+        if not isinstance(url, str) or not url:
+          continue
+        try:
+          ice_server = IceServer(url)
+          ice_server.username = username
+          ice_server.password = credential
+          parsed.append(ice_server)
+        except Exception:
+          logging.getLogger("WebRTCStream").warning("Ignoring invalid ICE server %r", url, exc_info=True)
+    # A supplied list is authoritative, including LAN-only sessions where relay is deliberately filtered.
+    return parsed or [IceServer("stun:stun.l.google.com:19302")]
+
+  def _log_debug(self, msg: Any, *args):
+    self.logger.debug(f"{type(self)}() {msg}", *args)
+
+  def _call_soon_threadsafe(self, fn: Callable, *args) -> None:
+    if self._loop is not None and self._loop.is_running():
+      self._loop.call_soon_threadsafe(fn, *args)
+    else:
+      fn(*args)
+
+  def _set_event(self, event: asyncio.Event) -> None:
+    self._call_soon_threadsafe(event.set)
+
+  @property
+  def _number_of_incoming_media(self) -> int:
+    media = len(self.incoming_camera_tracks) + len(self.incoming_audio_tracks)
+    media += int(self.messaging_channel is not None) if not self.should_add_data_channel else 0
+    return media
+
+  def _add_consumer_transceivers(self):
+    for camera_type in self.expected_incoming_camera_types:
+      media = Description.Video(camera_type, Description.Direction.RecvOnly)
+      media.add_h264_codec(96)
+      track = self.peer_connection.add_track(media)
+      track.set_media_handler(OpusRtpDepacketizer())
+      self._consumer_tracks.append(track)
+      self.incoming_camera_tracks[camera_type] = track
+    if self.expected_incoming_audio:
+      media = Description.Audio("audio", Description.Direction.RecvOnly)
+      media.add_opus_codec(111)
+      track = self.peer_connection.add_track(media)
+      self._consumer_tracks.append(track)
+      self.incoming_audio_tracks.append(track)
+
+  def _find_offer_video(self, remote_sdp: str) -> Tuple[str, int]:
+    desc = Description(remote_sdp, Description.Type.Offer)
+    for i in range(desc.media_count()):
+      media = desc.media(i)
+      if media is None or media.type() != "video":
+        continue
+      for payload_type in media.payload_types():
+        with contextlib.suppress(ValueError):
+          rtp_map = media.rtp_map(payload_type)
+          if rtp_map is not None and rtp_map.format.upper() == "H264":
+            return media.mid(), payload_type
+    raise ValueError("Remote SDP does not offer H264 video")
+
+  def _make_video_media(self, track: TiciVideoStreamTrack, remote_sdp: str) -> Tuple[Description.Video, int, int, str]:
+    mid, payload_type = self._find_offer_video(remote_sdp)
+    ssrc = random.randint(1, 0xFFFFFFFF)
+    cname = f"teleoprtc-{random.getrandbits(32):08x}"
+    stream_id = f"stream-{random.getrandbits(32):08x}"
+    media = Description.Video(mid, Description.Direction.SendOnly)
+    media.add_h264_codec(payload_type)
+    media.add_ssrc(ssrc, cname, stream_id, track.id)
+    return media, ssrc, payload_type, cname
+
+  def _find_offer_audio(self, remote_sdp: str) -> Tuple[str, int]:
+    desc = Description(remote_sdp, Description.Type.Offer)
+    for i in range(desc.media_count()):
+      media = desc.media(i)
+      if media is None or media.type() != "audio":
+        continue
+      for payload_type in media.payload_types():
+        with contextlib.suppress(ValueError):
+          rtp_map = media.rtp_map(payload_type)
+          if rtp_map is not None and rtp_map.format.upper() == "OPUS":
+            return media.mid(), payload_type
+    raise ValueError("Remote SDP does not offer Opus audio")
+
+  def _make_audio_media(self, remote_sdp: str) -> Tuple[Description.Audio, int, int, str]:
+    mid, payload_type = self._find_offer_audio(remote_sdp)
+    ssrc = random.randint(1, 0xFFFFFFFF)
+    cname = f"teleoprtc-audio-{random.getrandbits(32):08x}"
+    direction = Description.Direction.SendRecv if self.expected_incoming_audio else Description.Direction.SendOnly
+    media = Description.Audio(mid, direction)
+    media.add_opus_codec(payload_type)
+    media.add_ssrc(ssrc, cname, "audio", "audio")
+    return media, ssrc, payload_type, cname
+
+  def _add_producer_tracks(self, remote_sdp: Optional[str] = None):
+    for track in self.outgoing_video_tracks:
+      media, ssrc, payload_type, cname = self._make_video_media(track, remote_sdp or "")
+      rtc_track = self.peer_connection.add_track(media)
+
+      rtp_config = RtpPacketizationConfig(ssrc, cname, payload_type, H264RtpPacketizer.CLOCK_RATE)
+      rtp_config.start_timestamp = random.randint(0, 0xFFFFFFFF)
+      rtp_config.timestamp = rtp_config.start_timestamp
+      rtp_config.sequence_number = random.randint(0, 0xFFFF)
+
+      packetizer = H264RtpPacketizer(NalUnit.Separator.LongStartSequence, rtp_config, 1200)
+      packetizer.add_to_chain(RtcpSrReporter(rtp_config))
+      packetizer.add_to_chain(PliHandler(track.request_keyframe))
+      packetizer.add_to_chain(RtcpNackResponder())
+      rtc_track.set_media_handler(packetizer)
+
+      camera_type, _ = parse_video_track_id(track.id)
+      rtc_track.reset_callbacks()
+      self._receiver_report_tracks[camera_type] = (rtc_track, ssrc)
+      self._track_state.append((rtc_track, track, rtp_config))
+
+    for producer in self.outgoing_audio_tracks:
+      media, ssrc, payload_type, cname = self._make_audio_media(remote_sdp or "")
+      if self.expected_incoming_audio and self.incoming_audio_tracks:
+        # set_remote_description() creates the browser's sendrecv audio track. Reuse that
+        # negotiated m-line for our outbound Opus instead of adding a second track with the
+        # same MID, which leaves libdatachannel stuck in signalling/ICE negotiation.
+        rtc_track = self.incoming_audio_tracks[0]
+        negotiated_media = rtc_track.description()
+        negotiated_media.add_ssrc(ssrc, cname, "audio", "audio")
+        rtc_track.set_description(negotiated_media)
+      else:
+        rtc_track = self.peer_connection.add_track(media)
+      rtp_config = RtpPacketizationConfig(ssrc, cname, payload_type, OpusRtpPacketizer.DEFAULT_CLOCK_RATE)
+      rtp_config.start_timestamp = random.randint(0, 0xFFFFFFFF)
+      rtp_config.timestamp = rtp_config.start_timestamp
+      rtp_config.sequence_number = random.randint(0, 0xFFFF)
+      packetizer = OpusRtpPacketizer(rtp_config)
+      packetizer.add_to_chain(RtcpSrReporter(rtp_config))
+      packetizer.add_to_chain(RtcpNackResponder())
+      rtc_track.set_media_handler(packetizer)
+      self._audio_track_state.append((rtc_track, producer, rtp_config))
+
+  def _add_messaging_channel(self, channel: Optional[DataChannel] = None):
+    if channel is None:
+      channel = self.peer_connection.create_data_channel("data")
+    self.messaging_channel = channel
+
+    def on_message(message: Union[bytes, str]):
+      for handler in list(self.incoming_message_handlers):
+        self._call_soon_threadsafe(handler, message)
+
+    channel.on_message(on_message)
+    channel.on_open(lambda: self._set_event(self.messaging_channel_ready_event))
+    channel.on_closed(lambda: self._set_event(self.connection_stopped_event))
+    if channel.is_open():
+      self._set_event(self.messaging_channel_ready_event)
+    self._on_after_media()
+
+  def _on_connectionstatechange(self, state: PeerConnection.State):
+    self._log_debug("connection state is %s", state)
+    if state in (PeerConnection.State.Connected, PeerConnection.State.Failed):
+      self._set_event(self.connection_attempted_event)
+    if state in (PeerConnection.State.Disconnected, PeerConnection.State.Closed, PeerConnection.State.Failed):
+      self._set_event(self.connection_stopped_event)
+
+  def _on_gatheringstatechange(self, state: PeerConnection.GatheringState):
+    self._log_debug("gathering state is %s", state)
+    if state == PeerConnection.GatheringState.Complete:
+      self._set_event(self.gathering_complete_event)
+
+  def _on_incoming_track(self, track: Track):
+    self._log_debug("got track: %s", track.mid())
+    media_type = track.description().type()
+    if media_type == "audio":
+      if self.expected_incoming_audio:
+        self.incoming_audio_tracks.append(track)
+      self._on_after_media()
+      return
+    if media_type != "video":
+      self._on_after_media()
+      return
+    try:
+      camera_type, _ = parse_video_track_id(track.mid())
+    except ValueError:
+      camera_type = track.mid()
+    if camera_type in self.expected_incoming_camera_types:
+      self.incoming_camera_tracks[camera_type] = track
+    self._on_after_media()
+
+  def _on_incoming_datachannel(self, channel: DataChannel):
+    self._log_debug("got data channel: %s", channel.label())
+    if channel.label() == "data" and self.messaging_channel is None:
+      self._add_messaging_channel(channel)
+
+  def _update_receiver_report(self, camera_type: str, ssrc: int, message: bytes) -> None:
+    for report in _decode_receiver_reports(message):
+      if report.ssrc == ssrc:
+        self._receiver_reports[camera_type] = report
+
+  def _on_after_media(self):
+    if self.expected_number_of_incoming_media is not None and self._number_of_incoming_media >= self.expected_number_of_incoming_media:
+      self._set_event(self.incoming_media_ready_event)
+
+  def _parse_incoming_streams(self, remote_sdp: str):
+    desc = Description(remote_sdp, Description.Type.Offer)
+    media_count = 0
+    for i in range(desc.media_count()):
+      media = desc.media(i)
+      if media is None:
+        continue
+      direction = media.direction()
+      if media.type() == "video" and direction in (Description.Direction.SendOnly, Description.Direction.SendRecv):
+        media_count += 1
+      elif media.type() == "audio" and self.expected_incoming_audio and direction in (Description.Direction.SendOnly, Description.Direction.SendRecv):
+        media_count += 1
+    data_media_count = int(desc.has_application()) if not self.should_add_data_channel else 0
+    self.expected_number_of_incoming_media = media_count + data_media_count
+    if self.expected_number_of_incoming_media == 0:
+      self._set_event(self.incoming_media_ready_event)
+
+  def has_incoming_video_track(self, camera_type: str) -> bool:
+    return camera_type in self.incoming_camera_tracks
+
+  def has_incoming_audio_track(self) -> bool:
+    return len(self.incoming_audio_tracks) > 0
+
+  def has_messaging_channel(self) -> bool:
+    return self.messaging_channel is not None
+
+  def get_incoming_video_track(self, camera_type: str) -> Track:
+    assert camera_type in self.incoming_camera_tracks, "Video tracks are not enabled on this stream"
+    assert self.is_started, "Stream must be started"
+    return self.incoming_camera_tracks[camera_type]
+
+  def get_incoming_audio_track(self) -> Track:
+    assert len(self.incoming_audio_tracks) > 0, "Audio tracks are not enabled on this stream"
+    assert self.is_started, "Stream must be started"
+    return self.incoming_audio_tracks[0]
+
+  def get_messaging_channel(self) -> DataChannel:
+    assert self.messaging_channel is not None, "Messaging channel is not enabled on this stream"
+    assert self.is_started, "Stream must be started"
+    return self.messaging_channel
+
+  def get_receiver_report_stats(self) -> Dict[str, RtcpReceiverReport]:
+    return dict(self._receiver_reports)
+
+  def set_message_handler(self, message_handler: MessageHandler):
+    self.incoming_message_handlers.append(message_handler)
+
+  def add_ice_candidate(self, candidate: dict) -> None:
+    """Accept post-offer browser candidates for Konn3kt's existing trickle endpoint."""
+    candidate_sdp = str(candidate.get("candidate") or "") if isinstance(candidate, dict) else ""
+    if not candidate_sdp:
+      return
+    try:
+      self.peer_connection.add_remote_candidate(Candidate(candidate_sdp, str(candidate.get("sdpMid") or "")))
+    except Exception:
+      self.logger.warning("Ignoring invalid trickle ICE candidate", exc_info=True)
+
+  @property
+  def is_started(self) -> bool:
+    return self.peer_connection is not None and \
+           self.peer_connection.local_description() is not None and \
+           self.peer_connection.remote_description() is not None and \
+           self.peer_connection.state() != PeerConnection.State.Closed
+
+  @property
+  def is_connected_and_ready(self) -> bool:
+    return self.peer_connection is not None and \
+           self.peer_connection.state() == PeerConnection.State.Connected and \
+           (self.expected_number_of_incoming_media == 0 or self.incoming_media_ready_event.is_set())
+
+  async def _wait_for_gathering_complete(self):
+    if self.peer_connection.gathering_state() != PeerConnection.GatheringState.Complete:
+      await self.gathering_complete_event.wait()
+
+  async def _send_track_loop(self, rtc_track: Track, producer_track: TiciVideoStreamTrack, rtp_config: RtpPacketizationConfig):
+    while True:
+      if not rtc_track.is_open():
+        await asyncio.sleep(0.01)
+        continue
+
+      try:
+        packet = await producer_track.recv()
+        data = bytes(packet)
+        if not data:
+          continue
+
+        pts = int(packet.pts or 0)
+        timestamp = (rtp_config.start_timestamp + pts) & 0xFFFFFFFF
+        rtc_track.send_frame(data, FrameInfo(timestamp))
+      except asyncio.CancelledError:
+        raise
+      except Exception:
+        self.logger.exception("Error in send track loop for track %s", producer_track.id)
+        self._set_event(self.connection_stopped_event)
+        break
+
+  async def _receiver_report_loop(self):
+    while True:
+      for camera_type, (rtc_track, ssrc) in self._receiver_report_tracks.items():
+        for _ in range(32):
+          try:
+            message = rtc_track.receive()
+            if message is None: # go until queue empty (bounded to 32)
+              break
+            if isinstance(message, bytes):
+              self._update_receiver_report(camera_type, ssrc, message)
+          except asyncio.CancelledError:
+            raise
+          except Exception:
+            self.logger.exception("Error receiving report for %s", camera_type)
+            break
+      await asyncio.sleep(0.05)
+
+  async def _send_audio_track_loop(self, rtc_track: Track, producer: Any, rtp_config: RtpPacketizationConfig):
+    while True:
+      if not rtc_track.is_open():
+        await asyncio.sleep(0.01)
+        continue
+      try:
+        packet = await producer.recv()
+        if packet is None:
+          continue
+        data, pts = packet
+        if data:
+          rtc_track.send_frame(data, FrameInfo((rtp_config.start_timestamp + int(pts)) & 0xFFFFFFFF))
+      except asyncio.CancelledError:
+        raise
+      except Exception:
+        self.logger.exception("Error in audio send loop")
+        self._set_event(self.connection_stopped_event)
+        break
+
+  def _start_sender_tasks(self):
+    for rtc_track, producer_track, rtp_config in self._track_state:
+      self._sender_tasks.append(asyncio.create_task(self._send_track_loop(rtc_track, producer_track, rtp_config)))
+    for rtc_track, producer, rtp_config in self._audio_track_state:
+      self._sender_tasks.append(asyncio.create_task(self._send_audio_track_loop(rtc_track, producer, rtp_config)))
+    if self._track_state:
+      self._sender_tasks.append(asyncio.create_task(self._receiver_report_loop()))
+
+  async def wait_for_connection(self):
+    assert self.is_started
+    await self.connection_attempted_event.wait()
+    if self.peer_connection.state() != PeerConnection.State.Connected:
+      raise ValueError("Connection failed.")
+    if self.expected_number_of_incoming_media:
+      await self.incoming_media_ready_event.wait()
+    if self.messaging_channel is not None:
+      await self.messaging_channel_ready_event.wait()
+    self._start_sender_tasks()
+
+  async def wait_for_disconnection(self):
+    assert self.is_connected_and_ready, "Stream is not connected/ready yet (make sure wait_for_connection was awaited)"
+    await self.connection_stopped_event.wait()
+
+  async def stop(self):
+    for task in self._sender_tasks:
+      task.cancel()
+    for task in self._sender_tasks:
+      with contextlib.suppress(asyncio.CancelledError):
+        await task
+    self._sender_tasks.clear()
+    self.peer_connection.close()
+    self.peer_connection.reset_callbacks()
+    self.messaging_channel = None
+    self.incoming_camera_tracks.clear()
+    self.incoming_audio_tracks.clear()
+    self._consumer_tracks.clear()
+    self._track_state.clear()
+    self._audio_track_state.clear()
+    self._receiver_reports.clear()
+    self._receiver_report_tracks.clear()
+
+  @abc.abstractmethod
+  async def start(self) -> RTCSessionDescription:
+    raise NotImplementedError
+
+
+class WebRTCOfferStream(WebRTCBaseStream):
+  def __init__(self, session_provider: ConnectionProvider, *args, **kwargs):
+    super().__init__(*args, **kwargs)
+    self.session_provider = session_provider
+
+  async def start(self) -> RTCSessionDescription:
+    self._loop = asyncio.get_running_loop()
+    self._add_consumer_transceivers()
+    if self.should_add_data_channel:
+      self._add_messaging_channel()
+
+    self.peer_connection.set_local_description(Description.Type.Offer)
+    await self._wait_for_gathering_complete()
+    actual_offer = self.peer_connection.local_description()
+
+    streaming_offer = StreamingOffer(
+      sdp=str(actual_offer),
+      video=list(self.expected_incoming_camera_types),
+    )
+    remote_answer = await self.session_provider(streaming_offer)
+    self._parse_incoming_streams(remote_sdp=remote_answer.sdp)
+    self.peer_connection.set_remote_description(Description(remote_answer.sdp, Description.Type.Answer))
+    self._on_after_media()
+    actual_answer = self.peer_connection.remote_description()
+
+    return RTCSessionDescription(str(actual_answer), actual_answer.type_string())
+
+
+class WebRTCAnswerStream(WebRTCBaseStream):
+  def __init__(self, session: RTCSessionDescription, *args, **kwargs):
+    super().__init__(*args, **kwargs)
+    self.session = session
+
+  async def start(self) -> RTCSessionDescription:
+    self._loop = asyncio.get_running_loop()
+    assert self.peer_connection.remote_description() is None, "Connection already established"
+
+    self._parse_incoming_streams(remote_sdp=self.session.sdp)
+    self.peer_connection.set_remote_description(Description(self.session.sdp, Description.Type.Offer))
+    self._add_producer_tracks(self.session.sdp)
+
+    self.peer_connection.set_local_description(Description.Type.Answer)
+    await self._wait_for_gathering_complete()
+    actual_answer = self.peer_connection.local_description()
+
+    return RTCSessionDescription(str(actual_answer), actual_answer.type_string())

system/webrtc/teleoprtc_ldc/tracks.py

@@ -0,0 +1,77 @@
+# ruff: noqa: UP006, UP035
+
+import fractions
+import logging
+import uuid
+from typing import Any, Tuple
+
+
+VIDEO_CLOCK_RATE = 90000
+VIDEO_TIME_BASE = fractions.Fraction(1, VIDEO_CLOCK_RATE)
+
+
+def video_track_id(camera_type: str, track_id: str) -> str:
+  return f"{camera_type}:{track_id}"
+
+
+def parse_video_track_id(track_id: str) -> Tuple[str, str]:
+  parts = track_id.split(":")
+  if len(parts) != 2:
+    raise ValueError(f"Invalid video track id: {track_id}")
+
+  camera_type, track_id = parts
+  return camera_type, track_id
+
+
+class TiciVideoStreamTrack:
+  """
+  Abstract video track which associates video track with camera_type.
+  """
+  kind = "video"
+
+  def __init__(self, camera_type: str, dt: float, time_base: fractions.Fraction = VIDEO_TIME_BASE, clock_rate: int = VIDEO_CLOCK_RATE):
+    assert camera_type in ["driver", "wideRoad", "road"]
+    self._id: str = video_track_id(camera_type, str(uuid.uuid4()))
+    self._time_base: fractions.Fraction = time_base
+    self._clock_rate: int = clock_rate
+    self._logger = logging.getLogger("WebRTCStream")
+    self.readyState = "live"
+
+  @property
+  def id(self) -> str:
+    return self._id
+
+  def stop(self) -> None:
+    self.readyState = "ended"
+
+  def log_debug(self, msg: Any, *args):
+    self._logger.debug(f"{type(self)}() {msg}", *args)
+
+  async def recv(self):
+    raise NotImplementedError()
+
+  def request_keyframe(self) -> None:
+    pass
+
+
+class TiciTrackWrapper(TiciVideoStreamTrack):
+  """
+  Associates a generic video track with camera_type.
+  """
+  def __init__(self, camera_type: str, track: Any):
+    assert track.kind == "video"
+    super().__init__(camera_type, getattr(track, "_dt", 0.05))
+    try:
+      parsed_camera, _ = parse_video_track_id(track.id)
+      self._id = track.id if parsed_camera == camera_type else video_track_id(camera_type, track.id)
+    except ValueError:
+      self._id = video_track_id(camera_type, track.id)
+    self._track = track
+
+  async def recv(self):
+    return await self._track.recv()
+
+  def stop(self) -> None:
+    super().stop()
+    if hasattr(self._track, "stop"):
+      self._track.stop()

system/webrtc/webrtcd.py

@@ -542,6 +542,22 @@ class StreamRequestBody:
   ui_stream: bool = False
 
 
+def _new_stream_session(offer_sdp: str, body: StreamRequestBody, debug_mode: bool):
+  if Params().get_bool("Konn3ktLibdatachannelWebRTC"):
+    try:
+      from openpilot.system.webrtc.webrtcd_ldc import StreamSessionLibdatachannel
+      return StreamSessionLibdatachannel(
+        offer_sdp, body.cameras, body.bridge_services_in, body.bridge_services_out, body.iceServers, debug_mode,
+        ui_stream=body.ui_stream,
+      )
+    except Exception:
+      logging.getLogger("webrtcd").exception("libdatachannel unavailable; falling back to aiortc")
+  return StreamSession(
+    offer_sdp, body.cameras, body.bridge_services_in, body.bridge_services_out, body.iceServers, debug_mode,
+    ui_stream=body.ui_stream,
+  )
+
+
 async def get_stream(request: 'web.Request'):
   stream_dict, debug_mode = request.app['streams'], request.app['debug']
   logger = logging.getLogger("webrtcd")
@@ -561,8 +577,7 @@ async def get_stream(request: 'web.Request'):
         logger.exception("Failed to stop previous stream session")
     stream_dict.clear()
 
-    session = StreamSession(offer_sdp, body.cameras, body.bridge_services_in, body.bridge_services_out, body.iceServers, debug_mode,
-                            ui_stream=body.ui_stream)
+    session = _new_stream_session(offer_sdp, body, debug_mode)
     # Creating an answer can occasionally stall (ICE gathering, codec negotiation, etc).
     # Bound it so the HTTP request doesn't hang forever and athena can surface a useful error.
     try:
@@ -579,8 +594,7 @@ async def get_stream(request: 'web.Request'):
       else:
         logger.info("Retrying with fresh session and original SDP (no mDNS host candidates removed)")
 
-      session = StreamSession(retry_offer_sdp, body.cameras, body.bridge_services_in, body.bridge_services_out, body.iceServers, debug_mode,
-                              ui_stream=body.ui_stream)
+      session = _new_stream_session(retry_offer_sdp, body, debug_mode)
       answer = await asyncio.wait_for(session.get_answer(), timeout=15.0)
 
     session.start()

system/webrtc/webrtcd_ldc.py

@@ -0,0 +1,244 @@
+import asyncio
+import contextlib
+import json
+import logging
+import os
+import socket
+import uuid
+from typing import Any
+
+from openpilot.common.params import Params
+from cereal import messaging
+
+from openpilot.system.webrtc.webrtcd import CerealIncomingMessageProxy, CerealOutgoingMessageProxy, CerealProxyRunner, DynamicPubMaster
+
+
+def _default_route_ip() -> str | None:
+  """Use the interface the kernel will actually use for Internet/relay media."""
+  sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+  try:
+    sock.connect(("8.8.8.8", 53))
+    return sock.getsockname()[0]
+  except OSError:
+    return None
+  finally:
+    sock.close()
+
+
+class LibdatachannelBitrateController:
+  """Loss-driven bitrate control using native RTCP receiver reports."""
+
+  bitrates = [500_000, 1_500_000, int(os.environ.get("STREAM_BITRATE", 5_000_000))]
+  label_to_bitrate = {"low": bitrates[0], "med": bitrates[1], "high": bitrates[2]}
+  sample_interval = 0.2
+  high_loss = 0.10
+  sustained_loss = 0.05
+  down_samples = 5
+
+  def __init__(self, get_stats, enabled: bool = True):
+    self._get_stats = get_stats
+    self._enabled = enabled
+    self._auto = True
+    self._level = 1
+    self._counter = 0
+    self._up_samples = 5
+    self._previous: tuple[Any, ...] | None = None
+    self._task: asyncio.Task | None = None
+    self.current_bitrate = self.bitrates[self._level]
+    self._publish(self.current_bitrate)
+
+  def start(self) -> None:
+    if self._task is None:
+      self._task = asyncio.create_task(self.run())
+
+  def stop(self) -> None:
+    if self._task is not None:
+      self._task.cancel()
+    self._task = None
+
+  def enable(self, enabled: bool) -> None:
+    self._enabled = enabled
+
+  def set_quality(self, quality: str) -> None:
+    if quality in self.label_to_bitrate:
+      self._auto = False
+      self._publish(self.label_to_bitrate[quality])
+    elif quality == "auto":
+      self._auto = True
+
+  def _publish(self, bitrate: int) -> None:
+    self.current_bitrate = int(bitrate)
+    Params().put("LivestreamEncoderBitrate", self.current_bitrate)
+
+  def _sample_loss(self) -> float | None:
+    report = next(iter(self._get_stats().values()), None)
+    if report is None:
+      return None
+    current = (report.ssrc, report.fraction_lost, report.packets_lost, report.highest_seq_no, report.jitter, report.lsr, report.dlsr)
+    if current == self._previous:
+      return None
+    self._previous = current
+    return report.fraction_lost / 256.0
+
+  async def run(self) -> None:
+    while True:
+      await asyncio.sleep(self.sample_interval)
+      if not self._enabled or not self._auto:
+        continue
+      loss = self._sample_loss()
+      if loss is None:
+        continue
+      if loss >= self.sustained_loss and self._level > 0:
+        self._counter += 1
+        if loss >= self.high_loss or self._counter >= self.down_samples:
+          self._level -= 1
+          self._up_samples *= 2
+          self._counter = 0
+          self._publish(self.bitrates[self._level])
+      elif loss <= 0 and self._level < len(self.bitrates) - 1:
+        self._counter -= 1
+        if -self._counter >= self._up_samples:
+          self._level += 1
+          self._counter = 0
+          self._publish(self.bitrates[self._level])
+
+
+class StreamSessionLibdatachannel:
+  """IQ.Pilot's libdatachannel stream path. It preserves Konn3kt signalling and controls."""
+
+  shared_pub_master = DynamicPubMaster([])
+
+  def __init__(self, sdp: str, cameras: list[str], incoming_services: list[str], outgoing_services: list[str],
+               ice_servers: list[dict[str, Any]] | None = None, debug_mode: bool = False, ui_stream: bool = False):
+    from openpilot.system.webrtc.device.video_ldc import LiveStreamVideoStreamTrack
+    from openpilot.system.webrtc.teleoprtc_ldc.builder import WebRTCAnswerBuilder
+    from openpilot.system.webrtc.teleoprtc_ldc.info import parse_info_from_offer
+
+    config = parse_info_from_offer(sdp)
+    if len(cameras) != config.n_expected_camera_tracks:
+      raise ValueError("Incoming stream has misconfigured number of video tracks")
+    if debug_mode:
+      raise ValueError("libdatachannel debug tracks are not supported")
+
+    builder = WebRTCAnswerBuilder(sdp, bind_address=_default_route_ip(), ice_servers=ice_servers or [])
+    self.video_tracks = [LiveStreamVideoStreamTrack(camera) for camera in cameras]
+    for camera, track in zip(cameras, self.video_tracks, strict=True):
+      builder.add_video_stream(camera, track)
+
+    # The browser uses a single sendrecv audio m-line. libdatachannel's Python binding
+    # currently cannot negotiate that bidirectional track reliably, so this experimental
+    # transport deliberately remains video/control-only. The default aiortc path keeps
+    # both audio directions until the native duplex path passes the same integration test.
+    self.audio_output = None
+    self.stream = builder.stream()
+
+    self.identifier = str(uuid.uuid4())
+    self.incoming_bridge_services = incoming_services
+    self.incoming_bridge = CerealIncomingMessageProxy(self.shared_pub_master) if incoming_services else None
+    self.outgoing_bridge = CerealOutgoingMessageProxy(messaging.SubMaster(outgoing_services)) if outgoing_services else None
+    self.outgoing_bridge_runner = CerealProxyRunner(self.outgoing_bridge) if self.outgoing_bridge is not None else None
+    self.ui_stream_requested = ui_stream
+    self.ui_stream_runner: CerealProxyRunner | None = None
+    self.audio_input_proxy = None
+    self.audio_recv_requested = False
+    self.bitrate_controller = LibdatachannelBitrateController(self.stream.get_receiver_report_stats)
+    self.run_task: asyncio.Task | None = None
+    self._cleanup_lock = asyncio.Lock()
+    self._cleanup_done = False
+    self.logger = logging.getLogger("webrtcd")
+
+  def start(self) -> None:
+    self.run_task = asyncio.create_task(self.run())
+
+  async def get_answer(self):
+    return await self.stream.start()
+
+  async def stop_async(self) -> None:
+    if self.run_task is not None and not self.run_task.done() and self.run_task is not asyncio.current_task():
+      self.run_task.cancel()
+      with contextlib.suppress(asyncio.CancelledError):
+        await self.run_task
+    self.run_task = None
+    await self.post_run_cleanup()
+
+  def add_ice_candidate(self, candidate: Any) -> None:
+    self.stream.add_ice_candidate(candidate)
+
+  def message_handler(self, message: bytes | str) -> None:
+    try:
+      payload = json.loads(message)
+    except (TypeError, ValueError):
+      return
+    if not isinstance(payload, dict):
+      return
+    message_type = payload.get("type")
+    if message_type == "timingSei":
+      for track in self.video_tracks:
+        track.timing_sei_enabled = bool(payload.get("enabled", False))
+    elif message_type == "setQuality":
+      self.bitrate_controller.set_quality(str(payload.get("quality", "auto")))
+    elif message_type == "setAudioEnabled" and self.audio_output is not None:
+      self.audio_output.enable(bool(payload.get("enabled", True)))
+    elif message_type == "switchCamera":
+      for track in self.video_tracks:
+        track.switch_camera(str(payload.get("camera", "")))
+    elif message_type == "setUiStream":
+      self.set_ui_stream(bool(payload.get("enabled", False)))
+    elif self.incoming_bridge is not None:
+      try:
+        self.incoming_bridge.send(message)
+      except Exception:
+        self.logger.exception("Cereal incoming proxy failure")
+
+  def set_ui_stream(self, enabled: bool) -> None:
+    if enabled:
+      if self.ui_stream_runner is not None or not self.stream.has_messaging_channel():
+        return
+      from openpilot.system.webrtc.ui_stream import UIStreamMessageProxy
+      proxy = UIStreamMessageProxy(bitrate_getter=lambda: self.bitrate_controller.current_bitrate)
+      proxy.add_channel(self.stream.get_messaging_channel())
+      self.ui_stream_runner = CerealProxyRunner(proxy)
+      self.ui_stream_runner.start()
+    elif self.ui_stream_runner is not None:
+      self.ui_stream_runner.stop()
+      self.ui_stream_runner = None
+
+  async def run(self) -> None:
+    try:
+      await self.stream.wait_for_connection()
+      if self.stream.has_messaging_channel():
+        self.stream.set_message_handler(self.message_handler)
+        if self.incoming_bridge is not None:
+          await self.shared_pub_master.add_services_if_needed(self.incoming_bridge_services)
+        if self.outgoing_bridge_runner is not None:
+          self.outgoing_bridge_runner.proxy.add_channel(self.stream.get_messaging_channel())
+          self.outgoing_bridge_runner.start()
+        if self.ui_stream_requested:
+          self.set_ui_stream(True)
+      if self.audio_recv_requested and self.stream.has_incoming_audio_track():
+        from openpilot.system.webrtc.device.audio_ldc import IncomingOpusCerealProxy
+        self.audio_input_proxy = IncomingOpusCerealProxy(self.stream.get_incoming_audio_track())
+        self.audio_input_proxy.start()
+      self.bitrate_controller.start()
+      await self.stream.wait_for_disconnection()
+    except Exception:
+      self.logger.exception("libdatachannel stream session failure")
+    finally:
+      await self.post_run_cleanup()
+
+  async def post_run_cleanup(self) -> None:
+    async with self._cleanup_lock:
+      if self._cleanup_done:
+        return
+      self._cleanup_done = True
+      self.bitrate_controller.stop()
+      if self.ui_stream_runner is not None:
+        self.ui_stream_runner.stop()
+        self.ui_stream_runner = None
+      if self.outgoing_bridge_runner is not None:
+        self.outgoing_bridge_runner.stop()
+      if self.audio_input_proxy is not None:
+        await self.audio_input_proxy.stop()
+      for track in self.video_tracks:
+        track.stop()
+      await self.stream.stop()