Agnos 19.6.10

system/hardware/tici/agnos.json

@@ -56,7 +56,7 @@
   },
   {
     "name": "boot",
-    "url": "https://www.dropbox.com/scl/fi/9l9io42qfx2shr9er5jqx/boot9.img.xz?rlkey=lbtxz862kxbvn3jn98ejitj8p&st=vr35tgz7&dl=1",
+    "url": "https://files.firestar.link/x/ugiq4cqx08q7/boot9.img.xz",
     "hash": "ab2eba0f96b2f48efa376330c3eb509158361adf3ad9c20f269ec92457aa841f",
     "hash_raw": "ab2eba0f96b2f48efa376330c3eb509158361adf3ad9c20f269ec92457aa841f",
     "size": 48343040,
@@ -67,13 +67,13 @@
   },
   {
     "name": "system",
-    "url": "https://www.dropbox.com/scl/fi/pewhzpqzi3aewuiaffc6m/system10.img.xz?rlkey=olzrzulhs93zzghnjrskmdwxt&st=exnfk2oz&dl=1",
-    "hash": "ab395d4c963a908ab86709f1a6580a62dd24cb34ee71cf5fd5ec29d7d48d0e10",
-    "hash_raw": "ab395d4c963a908ab86709f1a6580a62dd24cb34ee71cf5fd5ec29d7d48d0e10",
+    "url": "https://files.firestar.link/x/07530yj1jd6a/system18.img.xz",
+    "hash": "01c84930849f9be2bdbad5e9a8dda3a6fd2be95e81d1b3556574b18346934f49",
+    "hash_raw": "01c84930849f9be2bdbad5e9a8dda3a6fd2be95e81d1b3556574b18346934f49",
     "size": 4718592000,
     "sparse": false,
     "full_check": false,
     "has_ab": true,
-    "ondevice_hash": "ab395d4c963a908ab86709f1a6580a62dd24cb34ee71cf5fd5ec29d7d48d0e10"
+    "ondevice_hash": "01c84930849f9be2bdbad5e9a8dda3a6fd2be95e81d1b3556574b18346934f49"
   }
 ]

system/loggerd/tests/test_uploader.py

@@ -7,7 +7,7 @@ from pathlib import Path
 from openpilot.system.hardware.hw import Paths
 
 from openpilot.common.swaglog import cloudlog
-from openpilot.system.loggerd.uploader import main, UPLOAD_ATTR_NAME, UPLOAD_ATTR_VALUE
+from openpilot.system.loggerd.uploader import clear_locks, main, UPLOAD_ATTR_NAME, UPLOAD_ATTR_VALUE
 from openpilot.system.loggerd.xattr_cache import getxattr
 
 from openpilot.system.loggerd.tests.loggerd_tests_common import UploaderTestCase
@@ -36,6 +36,10 @@ log_handler = FakeLogHandler()
 cloudlog.addHandler(log_handler)
 
 
+def test_clear_locks_missing_root(tmp_path):
+  clear_locks(str(tmp_path / "missing"))
+
+
 class TestUploader(UploaderTestCase):
   def setup_method(self):
     super().setup_method()

system/loggerd/uploader.py

@@ -63,6 +63,9 @@ def listdir_by_creation(d: str) -> list[str]:
     return []
 
 def clear_locks(root: str) -> None:
+  if not os.path.isdir(root):
+    return
+
   for logdir in os.listdir(root):
     path = os.path.join(root, logdir)
     try:

system/manager/manager.py

@@ -1094,11 +1094,6 @@ def manager_init() -> None:
                        device=HARDWARE.get_device_type())
   last_timing = _log_boot_timing("manager_init", "logging_ready", manager_init_start, last_timing)
 
-  # preimport all processes
-  for p in managed_processes.values():
-    p.prepare()
-  last_timing = _log_boot_timing("manager_init", "preimport_processes", manager_init_start, last_timing)
-
   # StarPilot variables
   install_starpilot(build_metadata, params)
   last_timing = _log_boot_timing("manager_init", "install_starpilot", manager_init_start, last_timing)

system/manager/process.py

@@ -634,19 +634,7 @@ class PythonProcess(ManagerProcess):
     self.launcher = launcher
 
   def prepare(self) -> None:
-    if self.enabled:
-      cloudlog.info(f"preimporting {self.module}")
-      start = time.monotonic()
-      try:
-        importlib.import_module(self.module)
-      finally:
-        line = f"SP_BOOT_TIMING preimport {self.name} module={self.module} +{time.monotonic() - start:.3f}s"
-        try:
-          with open(os.environ.get("SP_BOOT_TIMING_LOG", "/tmp/starpilot_boot_timing.log"), "a") as f:
-            f.write(line + "\n")
-        except OSError:
-          pass
-        cloudlog.warning(line)
+    pass
 
   def start(self) -> None:
     # In case we only tried a non blocking stop we need to stop it before restarting

system/webrtc/device/video.py

@@ -1,17 +1,19 @@
 import asyncio
+from dataclasses import dataclass
 import struct
 import time
 
-import av
 from teleoprtc.tracks import TiciVideoStreamTrack
-from aiortc.mediastreams import MediaStreamError
 
 from cereal import messaging
-from openpilot.common.params import Params
 from openpilot.common.realtime import DT_MDL
+from openpilot.common.params import Params
 
 
+# v4l2 buffer flag marking an encoded keyframe (linux/videodev2.h)
 V4L2_BUF_FLAG_KEYFRAME = 0x8
+
+# arbitrary 16-byte UUID identifying openpilot frame-timing SEI messages
 TIMING_SEI_UUID = bytes([
   0xa5, 0xe0, 0xc4, 0xa4, 0x5b, 0x6e, 0x4e, 0x1e,
   0x9c, 0x7e, 0x12, 0x34, 0x56, 0x78, 0x9a, 0xbc,
@@ -19,6 +21,15 @@ TIMING_SEI_UUID = bytes([
 _SEI_PREFIX = b'\x00\x00\x00\x01\x06\x05\x30' + TIMING_SEI_UUID
 
 
+@dataclass(frozen=True)
+class EncodedVideoFrame:
+  data: bytes
+  pts: int
+
+  def __bytes__(self) -> bytes:
+    return self.data
+
+
 class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
   camera_to_sock_mapping = {
     "driver": "livestreamDriverEncodeData",
@@ -52,6 +63,9 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
     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:
@@ -68,9 +82,7 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
 
   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)
         continue
@@ -79,18 +91,11 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
       if msg is not None:
         if not self._seen_keyframe and (getattr(msg, msg.which()).idx.flags & V4L2_BUF_FLAG_KEYFRAME):
           self._seen_keyframe = True
-          self.params.put("LivestreamRequestKeyframe", False)
+          self.params.put("LivestreamRequestKeyframe", False, block=False)
         break
       await asyncio.sleep(0.005)
 
-    packet = av.Packet(self._build_frame_data(msg))
-    packet.time_base = self._time_base
-
     self._pts =  ((time.monotonic_ns() - self._t0_ns) * self._clock_rate) // 1_000_000_000
-    packet.pts = self._pts
     self.log_debug("track sending frame %d", self._pts)
 
-    return packet
-
-  def codec_preference(self) -> str | None:
-    return "H264"
+    return EncodedVideoFrame(self._build_frame_data(msg), self._pts)

system/webrtc/tests/test_stream_session.py

@@ -1,20 +1,17 @@
 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
-import pyaudio
+import pytest
+
+pytest.importorskip("libdatachannel", reason="the upstream WebRTC backend requires Python 3.12")
+
 from cereal import messaging, log
+from teleoprtc.tracks import VIDEO_CLOCK_RATE
 
 from openpilot.system.webrtc.webrtcd import CerealOutgoingMessageProxy, CerealIncomingMessageProxy
 from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack
-from openpilot.system.webrtc.device.audio import AudioInputStreamTrack
 
 
 class TestStreamSession:
@@ -33,40 +30,37 @@ class TestStreamSession:
     expected_dict = {"type": "customReservedRawData0", "logMonoTime": 123, "valid": True, "data": "test"}
     expected_json = json.dumps(expected_dict).encode()
 
-    channel = mocker.Mock(spec=RTCDataChannel)
-    mocked_submaster = messaging.SubMaster(["customReservedRawData0"])
-    def mocked_update(t):
-      mocked_submaster.update_msgs(0, [test_msg])
+    channel = mocker.Mock()
+    channel.is_open.return_value = True
+    proxy = CerealOutgoingMessageProxy(["customReservedRawData0"])
+
+    def mocked_update(_):
+      proxy.sm.update_msgs(0, [test_msg])
 
     mocker.patch.object(messaging.SubMaster, "update", side_effect=mocked_update)
-    proxy = CerealOutgoingMessageProxy(["customReservedRawData0"])
-    proxy.sm = mocked_submaster
     proxy.add_channel(channel)
-
     proxy.update()
 
     channel.send.assert_called_once_with(expected_json)
 
   def test_incoming_proxy(self, mocker):
     tested_msgs = [
-      {"type": "customReservedRawData0", "data": "test"}, # primitive
-      {"type": "can", "data": [{"address": 0, "dat": "", "src": 0}]}, # list
-      {"type": "testJoystick", "data": {"axes": [0, 0], "buttons": [False]}}, # dict
+      {"type": "customReservedRawData0", "data": "test"},
+      {"type": "can", "data": [{"address": 0, "dat": "", "src": 0}]},
+      {"type": "testJoystick", "data": {"axes": [0, 0], "buttons": [False]}},
     ]
 
     mocked_pubmaster = mocker.MagicMock(spec=messaging.PubMaster)
-
     proxy = CerealIncomingMessageProxy(mocked_pubmaster)
 
     for msg in tested_msgs:
       proxy.send(json.dumps(msg).encode())
 
       mocked_pubmaster.send.assert_called_once()
-      mt, md = mocked_pubmaster.send.call_args.args
-      assert mt == msg["type"]
-      assert isinstance(md, capnp._DynamicStructBuilder)
-      assert hasattr(md, msg["type"])
-
+      msg_type, message = mocked_pubmaster.send.call_args.args
+      assert msg_type == msg["type"]
+      assert isinstance(message, capnp._DynamicStructBuilder)
+      assert hasattr(message, msg_type)
       mocked_pubmaster.reset_mock()
 
   def test_livestream_track(self, mocker):
@@ -78,29 +72,11 @@ class TestStreamSession:
     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())
-      assert packet.time_base == VIDEO_TIME_BASE
       if i == 0:
         start_ns = time.monotonic_ns()
         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
-
-  def test_input_audio_track(self, mocker):
-    packet_time, rate = 0.02, 16000
-    sample_count = int(packet_time * rate)
-    mocked_stream = mocker.MagicMock(spec=pyaudio.Stream)
-    mocked_stream.read.return_value = b"\x00" * 2 * sample_count
-
-    config = {"open.side_effect": lambda *args, **kwargs: mocked_stream}
-    mocker.patch("pyaudio.PyAudio", spec=True, **config)
-    track = AudioInputStreamTrack(audio_format=pyaudio.paInt16, packet_time=packet_time, rate=rate)
-
-    for i in range(5):
-      frame = self.loop.run_until_complete(track.recv())
-      assert frame.rate == rate
-      assert frame.samples == sample_count
-      assert frame.pts == i * sample_count
+      assert abs(i + packet.pts - (start_pts + (((time.monotonic_ns() - start_ns) * VIDEO_CLOCK_RATE) // 1_000_000_000))) < 450
+      assert bytes(packet) == b""

system/webrtc/tests/test_webrtcd.py

@@ -1,65 +1,42 @@
-import pytest
-import asyncio
 import json
-# 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 openpilot.system.webrtc.webrtcd import get_stream
+import pytest
 
-import aiortc
-from teleoprtc import WebRTCOfferBuilder
-from parameterized import parameterized_class
+pytest.importorskip("libdatachannel", reason="the upstream WebRTC backend requires Python 3.12")
+
+from openpilot.system.webrtc.webrtcd import ServerState, handle_get_schema, handle_post_notify, on_shutdown
 
 
-@parameterized_class(("in_services", "out_services"), [
-  (["testJoystick"], ["carState"]),
-  ([], ["carState"]),
-  (["testJoystick"], []),
-  ([], []),
-])
 @pytest.mark.asyncio
-class TestWebrtcdProc:
-  async def assertCompletesWithTimeout(self, awaitable, timeout=1):
-    try:
-      async with asyncio.timeout(timeout):
-        await awaitable
-    except TimeoutError:
-      pytest.fail("Timeout while waiting for awaitable to complete")
+async def test_get_schema():
+  status, body, content_type = await handle_get_schema(ServerState(), "carState")
 
-  async def test_webrtcd(self, mocker):
-    mock_request = mocker.MagicMock()
-    async def connect(offer):
-      body = {'sdp': offer.sdp, 'init_camera': offer.video[0], 'enabled': True,
-              'bridge_services_in': self.in_services, 'bridge_services_out': self.out_services}
-      mock_request.json.side_effect = mocker.AsyncMock(return_value=body)
-      response = await get_stream(mock_request)
-      response_json = json.loads(response.text)
-      return aiortc.RTCSessionDescription(**response_json)
+  assert status == 200
+  assert content_type.startswith("application/json")
+  assert "carState" in json.loads(body)
 
-    builder = WebRTCOfferBuilder(connect)
-    builder.offer_to_receive_video_stream("road")
-    builder.offer_to_receive_audio_stream()
-    if len(self.in_services) > 0 or len(self.out_services) > 0:
-      builder.add_messaging()
 
-    stream = builder.stream()
+@pytest.mark.asyncio
+async def test_get_schema_rejects_unknown_service():
+  with pytest.raises(AssertionError, match="Invalid service name"):
+    await handle_get_schema(ServerState(), "notARealService")
 
-    await self.assertCompletesWithTimeout(stream.start())
-    await self.assertCompletesWithTimeout(stream.wait_for_connection())
 
-    assert stream.has_incoming_video_track("road")
-    assert stream.has_incoming_audio_track()
-    assert stream.has_messaging_channel() == (len(self.in_services) > 0 or len(self.out_services) > 0)
+@pytest.mark.asyncio
+async def test_notify_and_shutdown_active_stream(mocker):
+  state = ServerState()
+  session = mocker.MagicMock()
+  session.stop = mocker.AsyncMock()
+  state.streams["test"] = session
 
-    video_track, audio_track = stream.get_incoming_video_track("road"), stream.get_incoming_audio_track()
-    await self.assertCompletesWithTimeout(video_track.recv())
-    await self.assertCompletesWithTimeout(audio_track.recv())
+  status, body, content_type = await handle_post_notify(state, {"type": "ping"})
 
-    await self.assertCompletesWithTimeout(stream.stop())
+  assert (status, body) == (200, b"OK")
+  assert content_type.startswith("text/plain")
+  channel = session.stream.get_messaging_channel.return_value
+  channel.send.assert_called_once_with(json.dumps({"type": "ping"}))
 
-    # cleanup, very implementation specific, test may break if it changes
-    assert mock_request.app["streams"].__setitem__.called, "Implementation changed, please update this test"
-    _, session = mock_request.app["streams"].__setitem__.call_args.args
-    await self.assertCompletesWithTimeout(session.post_run_cleanup())
+  await on_shutdown(state)
+
+  session.stop.assert_awaited_once()
+  assert state.streams == {}

system/webrtc/webrtcd.py

@@ -1,33 +1,29 @@
 #!/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
 import json
 import uuid
 import logging
-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
-from aiohttp import web
-if TYPE_CHECKING:
-  from aiortc.rtcdatachannel import RTCDataChannel
-import aioice.ice
+import signal
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import urlparse, parse_qs
+from typing import Any
 
 from openpilot.system.webrtc.helpers import StreamRequestBody
 from openpilot.system.webrtc.schema import generate_field
 from openpilot.common.params import Params
 from cereal import messaging, log
 
+SESSION_TIMEOUT_SECONDS = 300
 
 # 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
@@ -41,20 +37,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")
 
@@ -83,10 +67,10 @@ class CerealOutgoingMessageProxy(AsyncTaskRunner):
     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):
@@ -115,20 +99,17 @@ class CerealOutgoingMessageProxy(AsyncTaskRunner):
       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)
@@ -169,17 +150,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
@@ -196,7 +177,7 @@ class LivestreamBitrateController(AsyncTaskRunner):
       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:
@@ -213,22 +194,18 @@ class LivestreamBitrateController(AsyncTaskRunner):
           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)
@@ -244,48 +221,41 @@ class LivestreamBitrateController(AsyncTaskRunner):
 class StreamSession:
   shared_pub_master = DynamicPubMaster([])
 
-  def __init__(self, body: StreamRequestBody, debug_mode: bool = False):
-    if debug_mode:
-      from aiortc.mediastreams import AudioStreamTrack, VideoStreamTrack
-      from aiortc.contrib.media import MediaBlackhole
+  def __init__(self, body: StreamRequestBody):
     from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack
-    from openpilot.system.webrtc.device.audio import AudioInputStreamTrack, AudioOutputSpeaker
     from teleoprtc.builder import WebRTCAnswerBuilder
-    from teleoprtc.info import parse_info_from_offer
 
     self.identifier = str(uuid.uuid4())
     self.params = Params()
-    builder = WebRTCAnswerBuilder(body.sdp)
-    config = parse_info_from_offer(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()
-    builder.add_video_stream(body.init_camera, self.video_track)
-    if config.expected_audio_track:
-      builder.add_audio_stream(AudioInputStreamTrack() if not debug_mode else AudioStreamTrack())
-    if config.incoming_audio_track:
-      self.audio_output_cls = AudioOutputSpeaker if not debug_mode else MediaBlackhole
-      builder.offer_to_receive_audio_stream()
+    self.video_tracks = []
+    for camera in body.cameras:
+      track = LiveStreamVideoStreamTrack(camera, self.enabled)
+      self.video_tracks.append(track)
+      builder.add_video_stream(camera, track)
     self.stream = builder.stream()
 
+    self.is_body = "testJoystick" in body.bridge_services_in
+
     self.incoming_bridge: CerealIncomingMessageProxy | None = None
     self.incoming_bridge_services = body.bridge_services_in
     self.outgoing_bridge: CerealOutgoingMessageProxy | None = None
     self.bitrate_controller: LivestreamBitrateController | None = None
-    self.audio_output: AudioOutputSpeaker | MediaBlackhole | None = None
     if len(body.bridge_services_in) > 0:
       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()
     self._cleanup_done = False
     self.logger = logging.getLogger("webrtcd")
     self.logger.info(
-      "New stream session (%s), init camera %s, video enabled %s, incoming services %s, outgoing services %s",
-      self.identifier, body.init_camera, body.enabled, body.bridge_services_in, body.bridge_services_out,
+      "New stream session (%s), video cameras %s, video enabled %s, incoming services %s, outgoing services %s",
+      self.identifier, [t.id for t in self.video_tracks], body.enabled, body.bridge_services_in, body.bridge_services_out,
     )
 
   def start(self):
@@ -310,16 +280,21 @@ class StreamSession:
 
         match msg_type:
           case "livestreamCameraSwitch":
-            self.video_track.switch_camera(payload["data"]["camera"])
+            # only needed for 1 track stream
+            if len(self.video_tracks) == 1:
+              self.video_tracks[0].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)
+            for track in self.video_tracks:
+              track.enable(enabled)
             if self.outgoing_bridge is not None:
               self.outgoing_bridge.enable(enabled)
-            self.bitrate_controller.enable(enabled)
+            if self.bitrate_controller is not None:
+              self.bitrate_controller.enable(enabled)
             if not enabled:
               self.params.put("LivestreamRequestKeyframe", True)
           case "clockSync":
@@ -328,15 +303,29 @@ class StreamSession:
             }})
             self.stream.get_messaging_channel().send(pong)
           case "enableTimingSei":
-            if hasattr(self.video_track, 'timing_sei_enabled'):
-              self.video_track.timing_sei_enabled = bool(payload["data"]["enabled"])
+            for track in self.video_tracks:
+              track.timing_sei_enabled = bool(payload["data"]["enabled"])
           case _:
-            if payload.get("type") not in self.incoming_bridge_services:
+            if msg_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")
 
+  async def run_normal_session(self):
+    try:
+      await asyncio.wait_for(self.stream.wait_for_disconnection(), timeout=SESSION_TIMEOUT_SECONDS)
+    except TimeoutError:
+      self.logger.warning("Stream session (%s) timed out after %d s", self.identifier, SESSION_TIMEOUT_SECONDS)
+      try:
+        self.stream.get_messaging_channel().send(json.dumps({"type": "disconnect", "data": "Session timed out"}))
+      except Exception:
+        pass
+
+  async def run_body_session(self):
+    await self.stream.wait_for_disconnection()
+
   async def run(self):
     try:
       self.params.put("LivestreamRequestKeyframe", True)
@@ -349,15 +338,14 @@ class StreamSession:
           channel = self.stream.get_messaging_channel()
           self.outgoing_bridge.add_channel(channel)
           self.outgoing_bridge.start()
-      if self.stream.has_incoming_audio_track():
-        track = self.stream.get_incoming_audio_track(buffered=False)
-        self.audio_output = self.audio_output_cls()
-        self.audio_output.addTrack(track)
-        self.audio_output.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()
+      if self.is_body:
+        await self.run_body_session()
+      else:
+        await self.run_normal_session()
       self.logger.info("Stream session (%s) ended", self.identifier)
     except Exception:
       self.logger.exception("Stream session failure")
@@ -370,39 +358,52 @@ class StreamSession:
         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:
-        self.video_track.stop()
-        self.video_track = None
-      if self.audio_output is not None:
-        self.audio_output.stop()
-        self.audio_output = None
+      for track in self.video_tracks:
+        track.stop()
+      self.video_tracks.clear()
       await self.stream.stop()
 
 
-def schedule_teardown(app):
-  # if nothing connects for 5 seconds, tear down livestreaming processes
-  h = app.get('teardown')
-  if h:
-      h.cancel()
+class ServerState:
+  def __init__(self):
+    self.streams: dict[str, StreamSession] = {}
+    self.stream_lock = asyncio.Lock()
+    self.teardown: asyncio.TimerHandle | None = None
+
+
+# if nothing connects for 5 seconds, tear down livestreaming processes
+def schedule_teardown(state: ServerState):
+  if state.teardown is not None:
+    state.teardown.cancel()
+
   def clear():
-      if not app['streams']:
-          Params().put_bool("IsLiveStreaming", False)
-  app['teardown'] = asyncio.get_running_loop().call_later(5.0, clear)
+    if not state.streams:
+      Params().put_bool("IsLiveStreaming", False)
+
+  state.teardown = asyncio.get_running_loop().call_later(5.0, clear)
 
 
-async def get_stream(request: 'web.Request'):
-  stream_dict, debug_mode = request.app['streams'], request.app['debug']
-  raw_body = await request.json()
-  body = StreamRequestBody(**raw_body)
+def _json_response(obj: Any, status: int = 200) -> tuple[int, bytes, str]:
+  return (status, json.dumps(obj).encode(), "application/json; charset=utf-8")
 
-  async with request.app['stream_lock']:
+
+def _text_response(text: str, status: int = 200) -> tuple[int, bytes, str]:
+  return (status, text.encode(), "text/plain; charset=utf-8")
+
+
+async def handle_get_stream(state: ServerState, raw_body: bytes) -> tuple[int, bytes, str]:
+  stream_dict = state.streams
+  body = StreamRequestBody(**json.loads(raw_body))
+
+  async with state.stream_lock:
     # don't remove existing connection on prewarm request
     enabled = any(s.run_task and not s.run_task.done() and s.enabled for s in stream_dict.values())
     if enabled and not body.enabled:
-      return web.json_response({"error": "busy", "message": "someone else is connected."})
+      return _json_response({"error": "busy", "message": "someone else is connected."})
 
     for sid, s in list(stream_dict.items()):
       if s.run_task and not s.run_task.done():
@@ -414,10 +415,15 @@ async def get_stream(request: 'web.Request'):
       await s.stop()
       stream_dict.pop(sid, None)
 
-    session = StreamSession(body, debug_mode)
+    session = StreamSession(body)
     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)
@@ -427,94 +433,189 @@ async def get_stream(request: 'web.Request'):
 
     def remove_finished_session(_: asyncio.Task) -> None:
       stream_dict.pop(session.identifier, None)
-      schedule_teardown(request.app)
+      schedule_teardown(state)
+
     session.run_task.add_done_callback(remove_finished_session)
 
-  return web.json_response({"sdp": answer.sdp, "type": answer.type})
+  return _json_response({"sdp": answer.sdp, "type": answer.type})
 
 
-async def get_schema(request: 'web.Request'):
-  services = request.query.get("services", "").split(",")
+async def handle_get_schema(state: ServerState, services_param: str) -> tuple[int, bytes, str]:
+  services = services_param.split(",")
   services = [s for s in services if s]
   assert all(s in log.Event.schema.fields and not s.endswith("DEPRECATED") for s in services), "Invalid service name"
   schema_dict = {s: generate_field(log.Event.schema.fields[s]) for s in services}
-  return web.json_response(schema_dict)
+  return _json_response(schema_dict)
 
 
-async def post_notify(request: 'web.Request'):
-  try:
-    payload = await request.json()
-  except Exception as e:
-    raise web.HTTPBadRequest(text="Invalid JSON") from e
-
-  for session in list(request.app.get('streams', {}).values()):
+async def handle_post_notify(state: ServerState, payload: Any) -> tuple[int, bytes, str]:
+  for session in list(state.streams.values()):
     try:
       ch = session.stream.get_messaging_channel()
       ch.send(json.dumps(payload))
     except Exception:
       continue
 
-  return web.Response(status=200, text="OK")
+  return _text_response("OK")
 
 
-async def on_shutdown(app: 'web.Application'):
-  for session in list(app['streams'].values()):
+async def on_shutdown(state: ServerState):
+  for session in list(state.streams.values()):
     try:
       ch = session.stream.get_messaging_channel()
       ch.send(json.dumps({"type": "disconnect", "data": "device streaming has been stopped."}))
     except Exception:
       pass
     await session.stop()
-  del app['streams']
+  state.streams.clear()
 
 
-@web.middleware
-async def error_middleware(request: 'web.Request', handler):
-  try:
-    return await handler(request)
-  except Exception as e:
-    logging.getLogger("webrtcd").exception("Unhandled error handling %s", request.path)
-    return web.json_response({"error": "exception", "message": f"{type(e).__name__}: {e}"}, status=500)
+class WebrtcdHandler(BaseHTTPRequestHandler):
+  protocol_version = "HTTP/1.1"
+
+  # path -> allowed methods (aiohttp registered POST /stream, POST /notify, GET /schema + its auto HEAD)
+  _routes = {
+    "/schema": ("GET", "HEAD"),
+    "/stream": ("POST",),
+    "/notify": ("POST",),
+  }
+
+  def _send(self, status: int, body: bytes, content_type: str) -> None:
+    self.send_response(status)
+    self.send_header("Content-Type", content_type)
+    self.send_header("Content-Length", str(len(body)))
+    self.end_headers()
+    if self.command != "HEAD":
+      self.wfile.write(body)
+
+  def _read_body(self) -> bytes:
+    length = int(self.headers.get("Content-Length", 0))
+    return self.rfile.read(length) if length else b""
+
+  def _run(self, coro) -> tuple[int, bytes, str]:
+    return asyncio.run_coroutine_threadsafe(coro, self.server.loop).result()
+
+  def _dispatch_request(self) -> None:
+    parsed = urlparse(self.path)
+    allowed = self._routes.get(parsed.path)
+
+    try:
+      if allowed is None:
+        result = _json_response({"error": "not found"}, status=404)
+      elif self.command not in allowed:
+        result = _json_response({"error": "method not allowed"}, status=405)
+      elif parsed.path == "/schema":
+        services = parse_qs(parsed.query).get("services", [""])[0]
+        result = self._run(handle_get_schema(self.server.state, services))
+      elif parsed.path == "/stream":
+        result = self._run(handle_get_stream(self.server.state, self._read_body()))
+      else:  # /notify
+        try:
+          payload = json.loads(self._read_body())
+        except Exception:
+          result = _json_response({"error": "bad request"}, status=400)
+        else:
+          result = self._run(handle_post_notify(self.server.state, payload))
+    except Exception as e:
+      logging.getLogger("webrtcd").exception("Unhandled error handling %s", self.path)
+      result = _json_response({"error": "exception", "message": f"{type(e).__name__}: {e}"}, status=500)
+
+    self._send(*result)
+
+  def do_GET(self) -> None:
+    self._dispatch_request()
+
+  def do_HEAD(self) -> None:
+    self._dispatch_request()
+
+  def do_POST(self) -> None:
+    self._dispatch_request()
+
+  def do_PUT(self) -> None:
+    self._dispatch_request()
+
+  def do_DELETE(self) -> None:
+    self._dispatch_request()
+
+  def do_PATCH(self) -> None:
+    self._dispatch_request()
+
+  def do_OPTIONS(self) -> None:
+    self._dispatch_request()
+
+  def log_message(self, format: str, *args: object) -> None:  # noqa: A002  # stdlib override
+    # silence default access logging; errors are logged explicitly in _dispatch_request
+    pass
 
 
-def prewarm_stream_session_imports(debug_mode: bool = False) -> None:
-  if debug_mode:
-    from aiortc.mediastreams import VideoStreamTrack
-    assert VideoStreamTrack
+class WebrtcdHTTPServer(ThreadingHTTPServer):
+  daemon_threads = True
+  allow_reuse_address = True
+  state: ServerState
+  loop: asyncio.AbstractEventLoop
+
+
+async def _shutdown(server: WebrtcdHTTPServer, state: ServerState, loop: asyncio.AbstractEventLoop) -> None:
+  # stop accepting new HTTP connections (blocks until serve_forever returns, so
+  # run it off the loop) then tear down active stream sessions.
+  await loop.run_in_executor(None, server.shutdown)
+  await on_shutdown(state)
+  loop.stop()
+
+
+def prewarm_stream_session_imports() -> None:
   from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack
   from teleoprtc.builder import WebRTCAnswerBuilder
   assert LiveStreamVideoStreamTrack
   assert WebRTCAnswerBuilder
 
 
-def webrtcd_thread(host: str, port: int, debug: bool):
-  logging.basicConfig(level=logging.CRITICAL, handlers=[logging.StreamHandler()])
+def webrtcd_thread(host: str, port: int):
+  logging.basicConfig(level=logging.INFO, handlers=[logging.StreamHandler()])
   prewarm_start = time.monotonic()
-  prewarm_stream_session_imports(debug)
+  prewarm_stream_session_imports()
   prewarm_end = time.monotonic()
   logging.getLogger("webrtcd").info(f"webrtc prewarm finished in {(prewarm_end - prewarm_start) * 1000} ms")
 
-  app = web.Application(middlewares=[error_middleware])
+  loop = asyncio.new_event_loop()
+  asyncio.set_event_loop(loop)
+  state = ServerState()
 
-  app['streams'] = dict()
-  app['stream_lock'] = asyncio.Lock()
-  app['debug'] = debug
-  app.on_shutdown.append(on_shutdown)
-  app.router.add_post("/stream", get_stream)
-  app.router.add_post("/notify", post_notify)
-  app.router.add_get("/schema", get_schema)
+  server = WebrtcdHTTPServer((host, port), WebrtcdHandler)
+  server.state = state
+  server.loop = loop
 
-  web.run_app(app, host=host, port=port)
+  # serve HTTP on a daemon thread so the asyncio loop can own the main thread
+  http_thread = threading.Thread(target=server.serve_forever, name="webrtcd-http", daemon=True)
+  http_thread.start()
+
+  shutting_down = False
+  shutdown_task = None
+
+  def request_shutdown() -> None:
+    nonlocal shutting_down, shutdown_task
+    if shutting_down:
+      return
+    shutting_down = True
+    shutdown_task = loop.create_task(_shutdown(server, state, loop))
+
+  for sig in (signal.SIGINT, signal.SIGTERM):
+    loop.add_signal_handler(sig, request_shutdown)
+
+  try:
+    loop.run_forever()
+  finally:
+    server.server_close()
+    loop.close()
 
 
 def main():
   parser = argparse.ArgumentParser(description="WebRTC daemon")
   parser.add_argument("--host", type=str, default="0.0.0.0", help="Host to listen on")
   parser.add_argument("--port", type=int, default=5001, help="Port to listen on")
-  parser.add_argument("--debug", action="store_true", help="Enable debug mode")
   args = parser.parse_args()
 
-  webrtcd_thread(args.host, args.port, args.debug)
+  webrtcd_thread(args.host, args.port)
 
 
 if __name__=="__main__":