feat(webrtc): addIceCandidate dispatcher (#38161)

* addIceCandidate dispatcher

* clean

* fix lint

system/athena/athenad.py

@@ -604,6 +604,14 @@ def startStream(sdp: str) -> dict:
   except requests.ConnectionError as e:
     raise Exception("webrtc is not running. turn on comma body ignition.") from e
 
+@dispatcher.add_method
+def addIceCandidate(session_id: str, candidate: dict | None) -> dict:
+  if session_id is None:
+    return Exception("cannot add ice candidate without session_id")
+  resp = requests.post(f"http://localhost:{WEBRTCD_PORT}/candidate",
+                       json={"session_id": session_id, "candidate": candidate}, timeout=10)
+  return resp.json()
+
 @dispatcher.add_method
 def takeSnapshot() -> str | dict[str, str] | None:
   from openpilot.system.camerad.snapshot import jpeg_write, snapshot

system/webrtc/webrtcd.py

@@ -251,6 +251,23 @@ class StreamSession:
   async def get_answer(self):
     return await self.stream.start()
 
+  async def add_ice_candidate(self, candidate_init: dict | None):
+    from aiortc.sdp import candidate_from_sdp
+
+    pc = self.stream.peer_connection
+    if pc.iceConnectionState not in ("new", "checking"):
+      return
+
+    # a null/empty candidate signals end-of-candidates per the WebRTC convention
+    if not candidate_init or not candidate_init.get("candidate"):
+      await pc.addIceCandidate(None)
+      return
+
+    candidate = candidate_from_sdp(candidate_init["candidate"].split(":", 1)[1])
+    candidate.sdpMid = candidate_init.get("sdpMid")
+    candidate.sdpMLineIndex = candidate_init.get("sdpMLineIndex")
+    await pc.addIceCandidate(candidate)
+
   def message_handler(self, message: bytes):
     assert self.incoming_bridge is not None
     try:
@@ -354,7 +371,18 @@ async def get_stream(request: 'web.Request'):
 
     stream_dict[session.identifier] = session
 
-  return web.json_response({"sdp": answer.sdp, "type": answer.type})
+  return web.json_response({"sdp": answer.sdp, "type": answer.type, "session_id": session.identifier})
+
+
+async def post_candidate(request: 'web.Request'):
+  body = await request.json()
+  session = request.app.get('streams', {}).get(body.get("session_id"))
+
+  try:
+    await session.add_ice_candidate(body.get("candidate"))
+  except Exception as e:
+    raise web.HTTPBadRequest(text=json.dumps({"error": "invalid_candidate", "message": str(e)}), content_type="application/json") from e
+  return web.Response(status=200, text="OK")
 
 
 async def get_schema(request: 'web.Request'):
@@ -400,6 +428,7 @@ def webrtcd_thread(host: str, port: int, debug: bool):
   app['debug'] = debug
   app.on_shutdown.append(on_shutdown)
   app.router.add_post("/stream", get_stream)
+  app.router.add_post("/candidate", post_candidate)
   app.router.add_post("/notify", post_notify)
   app.router.add_get("/schema", get_schema)