Athena webrtc streamer

system/athena/athenad.py

@@ -14,6 +14,7 @@
 import tempfile
 import threading
 import time
+import asyncio
 import zstandard as zstd
 from dataclasses import asdict, dataclass, replace
 from datetime import datetime
@@ -40,6 +41,7 @@
 from openpilot.common.swaglog import cloudlog
 from openpilot.system.version import get_build_metadata
 from openpilot.system.hardware.hw import Paths
+from openpilot.system.athena.streamer import Streamer
 
 
 ATHENA_HOST = os.getenv('ATHENA_HOST', 'wss://athena.comma.ai')
@@ -101,6 +103,9 @@ def from_dict(cls, d: dict) -> UploadItem:
 low_priority_send_queue: Queue[str] = queue.Queue()
 log_recv_queue: Queue[str] = queue.Queue()
 cancelled_uploads: set[str] = set()
+sdp_recv_queue: Queue[str] = queue.Queue()
+sdp_send_queue: Queue[str] = queue.Queue()
+ice_send_queue: Queue[str] = queue.Queue()
 
 cur_upload_items: dict[int, UploadItem | None] = {}
 
@@ -147,6 +152,7 @@ def handle_long_poll(ws: WebSocket, exit_event: threading.Event | None) -> None:
     threading.Thread(target=upload_handler, args=(end_event,), name='upload_handler'),
     threading.Thread(target=log_handler, args=(end_event,), name='log_handler'),
     threading.Thread(target=stat_handler, args=(end_event,), name='stat_handler'),
+    threading.Thread(target=rtc_handler, args=(end_event, sdp_send_queue, sdp_recv_queue, ice_send_queue), name='rtc_handler')
   ] + [
     threading.Thread(target=jsonrpc_handler, args=(end_event,), name=f'worker_{x}')
     for x in range(HANDLER_THREADS)
@@ -167,6 +173,43 @@ def handle_long_poll(ws: WebSocket, exit_event: threading.Event | None) -> None:
       thread.join()
 
 
+
+def rtc_handler(end_event: threading.Event, sdp_send_queue: queue.Queue, sdp_recv_queue: queue.Queue, ice_recv_queue: queue.Queue) -> None:
+  loop = asyncio.new_event_loop()
+  asyncio.set_event_loop(loop)
+
+  try:
+    streamer = Streamer(sdp_send_queue, sdp_recv_queue, ice_recv_queue)
+    loop.run_until_complete(streamer.event_loop(end_event))
+  finally:
+    loop.close()
+
+@dispatcher.add_method
+def setSdpAnswer(answer):
+  sdp_recv_queue.put_nowait(answer)
+
+@dispatcher.add_method
+def getSdp():
+  start_time = time.time()
+  timeout = 10
+  while time.time() - start_time < timeout:
+    try:
+      sdp = json.loads(sdp_send_queue.get(timeout=0.1))
+      if sdp:
+        return sdp
+    except queue.Empty:
+      pass
+
+@dispatcher.add_method
+def getIce():
+  if not ice_send_queue.empty():
+    return json.loads(ice_send_queue.get_nowait())
+  else:
+    return {
+      'error': True
+    }
+
+
 def jsonrpc_handler(end_event: threading.Event) -> None:
   dispatcher["startLocalProxy"] = partial(startLocalProxy, end_event)
   while not end_event.is_set():