car livestream (#38193)

* athenad and webrtcd updates

* remove feature stream services from webrtcd split

* stream encoder thread

* reduce diff

* wire webrtc to livestream camera encoder

* request livestream camera switch service

* remove camera list in favour of init camera field

* remove cors

* clean

* remove unused

* remove extra try except

* add back exception trace

* add stream road camera info to stream cameras

* fix

* clean diff

* clean diff

* add testJoystick only on body

* fix camera list

* remove reference to future service

* encode all cameras and swap in video track in webrtc

* clean

* explicitly gate bridge send

* clean leftover

* initial idea

* add a watchdog to kill the process after disconnect or onroad

* start camerad and stream encoderd as well

* add message handler even without bridge

* remove carState when offroad

* transition to onroad works for body, kill for car not working

* turn off stream processes on started car

* cereal messaging sub field allow list, carparamspersistent, ping webrtcd to see when awake

* fix imports

* fix and increase max retries

* reduce max retries, increase timeout

* timeout on wait for connection

* remove cereal sub message and update process config to run on body ignition

* debounce 5s on teardown

* update error messages

* fix tear down crash

* clean

* clean

* fix lint

common/params_keys.h

@@ -58,6 +58,7 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
     {"IsDriverViewEnabled", {CLEAR_ON_MANAGER_START, BOOL}},
     {"IsEngaged", {PERSISTENT, BOOL}},
     {"IsLdwEnabled", {PERSISTENT, BOOL}},
+    {"IsLiveStreaming", {CLEAR_ON_MANAGER_START, BOOL}},
     {"IsMetric", {PERSISTENT, BOOL}},
     {"IsOffroad", {CLEAR_ON_MANAGER_START, BOOL}},
     {"IsOnroad", {PERSISTENT, BOOL}},

system/athena/athenad.py

@@ -45,7 +45,6 @@ from openpilot.system.hardware.hw import Paths
 ATHENA_HOST = os.getenv('ATHENA_HOST', 'wss://athena.comma.ai')
 HANDLER_THREADS = int(os.getenv('HANDLER_THREADS', "4"))
 LOCAL_PORT_WHITELIST = {22, }  # SSH
-WEBRTCD_PORT = 5001
 
 LOG_ATTR_NAME = 'user.upload'
 LOG_ATTR_VALUE_MAX_UNIX_TIME = int.to_bytes(2147483647, 4, sys.byteorder)
@@ -84,9 +83,6 @@ UPLOAD_SESS = requests.Session()
 UPLOAD_SESS.mount("http://", UploadTOSAdapter())
 UPLOAD_SESS.mount("https://", UploadTOSAdapter())
 
-WEBRTCD_SESS = requests.Session()
-WEBRTCD_SESS.mount("http://", HTTPAdapter(max_retries=0))
-
 
 @dataclass
 class UploadFile:
@@ -581,28 +577,28 @@ def getNetworks():
 
 @dispatcher.add_method
 def startStream(sdp: str, enabled: bool) -> dict:
-  from openpilot.system.webrtc.models import StreamRequestBody
+  from openpilot.system.webrtc.helpers import StreamRequestBody, post_stream_request, wait_for_webrtcd
+  params = Params()
   bridge_services_in = []
 
-  # get live car params to avoid stale notCar edge case
-  cp_bytes = Params().get("CarParams")
+  # stale car params case taken care of by webrtcd being shut off on ignition
+  cp_bytes = Params().get("CarParamsPersistent")
   if cp_bytes is not None:
     with car.CarParams.from_bytes(cp_bytes) as CP:
       if CP.notCar:
         bridge_services_in.append("testJoystick")
+  else:
+      raise Exception("failed to get CarParamsPersistent")
+
+  if not params.get_bool("IsOnroad"):
+    # manager owns camerad/stream_encoderd/webrtcd; flip the param and let it bring them up.
+    # webrtcd clears IsLiveStreaming when the session ends
+    params.put_bool("IsLiveStreaming", True)
+    # wait for webrtcd end points to wake up
+    wait_for_webrtcd()
+
+  return post_stream_request(StreamRequestBody(sdp, "wideRoad", enabled, bridge_services_in, ["carState", "deviceState"]))
 
-  t_start = time.monotonic()
-  body = StreamRequestBody(sdp=sdp, init_camera="wideRoad", bridge_services_in=bridge_services_in, bridge_services_out=["carState"], enabled=enabled)
-  try:
-    resp = WEBRTCD_SESS.post(f"http://localhost:{WEBRTCD_PORT}/stream", json=asdict(body), timeout=10)
-    t_end = time.monotonic()
-    ret = resp.json()
-    ret["time"] = (t_end - t_start) * 1000
-    return ret
-  except requests.ConnectTimeout as e:
-    raise Exception("webrtc took too long to respond. is the comma body on?") from e
-  except requests.ConnectionError as e:
-    raise Exception("webrtc is not running. turn on comma body ignition.") from e
 
 @dispatcher.add_method
 def takeSnapshot() -> str | dict[str, str] | None:

system/manager/process_config.py

@@ -58,21 +58,27 @@ def only_onroad(started: bool, params: Params, CP: car.CarParams) -> bool:
 def only_offroad(started: bool, params: Params, CP: car.CarParams) -> bool:
   return not started
 
+def livestream(started: bool, params: Params, CP: car.CarParams) -> bool:
+  return params.get_bool("IsLiveStreaming")
+
 def or_(*fns):
   return lambda *args: operator.or_(*(fn(*args) for fn in fns))
 
 def and_(*fns):
   return lambda *args: operator.and_(*(fn(*args) for fn in fns))
 
+def not_(*fns):
+  return lambda *args: operator.not_(*(fn(*args) for fn in fns))
+
 procs = [
   DaemonProcess("manage_athenad", "system.athena.manage_athenad", "AthenadPid"),
 
   NativeProcess("loggerd", "system/loggerd", ["./loggerd"], logging),
   NativeProcess("encoderd", "system/loggerd", ["./encoderd"], only_onroad),
-  NativeProcess("stream_encoderd", "system/loggerd", ["./encoderd", "--stream"], notcar),
+  NativeProcess("stream_encoderd", "system/loggerd", ["./encoderd", "--stream"], or_(and_(livestream, not_(iscar)), notcar)),
   PythonProcess("logmessaged", "system.logmessaged", always_run),
 
-  NativeProcess("camerad", "system/camerad", ["./camerad"], driverview, enabled=not WEBCAM),
+  NativeProcess("camerad", "system/camerad", ["./camerad"], or_(driverview, livestream), enabled=not WEBCAM),
   PythonProcess("webcamerad", "tools.webcam.camerad", driverview, enabled=WEBCAM),
   PythonProcess("proclogd", "system.proclogd", only_onroad, enabled=platform.system() != "Darwin"),
   PythonProcess("journald", "system.journald", only_onroad, platform.system() != "Darwin"),
@@ -115,7 +121,7 @@ procs = [
 
   # debug procs
   NativeProcess("bridge", "cereal/messaging", ["./bridge"], notcar),
-  PythonProcess("webrtcd", "system.webrtc.webrtcd", notcar),
+  PythonProcess("webrtcd", "system.webrtc.webrtcd", or_(and_(livestream, not_(iscar)), notcar)),
   PythonProcess("webjoystick", "tools.bodyteleop.web", notcar),
   PythonProcess("joystick", "tools.joystick.joystick_control", and_(joystick, iscar)),
 ]

system/webrtc/helpers.py

@@ -0,0 +1,40 @@
+import time
+import requests
+from dataclasses import asdict, dataclass, field
+
+
+WEBRTCD_PORT = 5001
+
+@dataclass
+class StreamRequestBody:
+  sdp: str
+  init_camera: str
+  enabled: bool
+  bridge_services_in: list[str] = field(default_factory=list)
+  bridge_services_out: list[str] = field(default_factory=list)
+
+
+def post_stream_request(body: StreamRequestBody) -> dict:
+  t_start = time.monotonic()
+  try:
+    resp = requests.post(f"http://localhost:{WEBRTCD_PORT}/stream", json=asdict(body), timeout=10)
+    t_end = time.monotonic()
+    ret = resp.json()
+    ret["time"] = (t_end - t_start) * 1000
+    return ret
+  except requests.ConnectTimeout as e:
+    raise Exception("webrtc took too long to respond.") from e
+  except requests.ConnectionError as e:
+    raise Exception("webrtc server on device is not running.") from e
+
+
+def wait_for_webrtcd(max_retries: float = 10) -> None:
+  attempts = 0
+  while attempts < max_retries:
+    try:
+      if requests.get(f"http://localhost:{WEBRTCD_PORT}/schema", timeout=1).ok:
+        return
+    except requests.ConnectionError:
+      attempts += 1
+      time.sleep(0.5)
+  raise TimeoutError("webrtcd did not initialize in time.")

system/webrtc/models.py

@@ -1,9 +0,0 @@
-from dataclasses import dataclass, field
-
-@dataclass
-class StreamRequestBody:
-  sdp: str
-  init_camera: str
-  enabled: bool
-  bridge_services_in: list[str] = field(default_factory=list)
-  bridge_services_out: list[str] = field(default_factory=list)

system/webrtc/webrtcd.py

@@ -23,7 +23,7 @@ if TYPE_CHECKING:
   from aiortc.rtcdatachannel import RTCDataChannel
 import aioice.ice
 
-from openpilot.system.webrtc.models import StreamRequestBody
+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
@@ -329,11 +329,11 @@ class StreamSession:
   async def run(self):
     try:
       self.params.put("LivestreamRequestKeyframe", True)
-      await self.stream.wait_for_connection()
+      await asyncio.wait_for(self.stream.wait_for_connection(), timeout=15)
       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)
-          self.stream.set_message_handler(self.message_handler)
         if self.outgoing_bridge is not None:
           channel = self.stream.get_messaging_channel()
           self.outgoing_bridge.add_channel(channel)
@@ -363,6 +363,17 @@ class StreamSession:
       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()
+  def clear():
+      if not app['streams']:
+          Params().put_bool("IsLiveStreaming", False)
+  app['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()
@@ -397,13 +408,14 @@ async def get_stream(request: 'web.Request'):
 
     def remove_finished_session(_: asyncio.Task) -> None:
       stream_dict.pop(session.identifier, None)
+      schedule_teardown(request.app)
     session.run_task.add_done_callback(remove_finished_session)
 
   return web.json_response({"sdp": answer.sdp, "type": answer.type})
 
 
 async def get_schema(request: 'web.Request'):
-  services = request.query["services"].split(",")
+  services = request.query.get("services", "").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}
@@ -427,10 +439,10 @@ async def post_notify(request: 'web.Request'):
 
 
 async def on_shutdown(app: 'web.Application'):
-  for session in app['streams'].values():
+  for session in list(app['streams'].values()):
     try:
       ch = session.stream.get_messaging_channel()
-      ch.send(json.dumps({"type": "disconnect", "data": "device stream has been stopped."}))
+      ch.send(json.dumps({"type": "disconnect", "data": "device streaming has been stopped."}))
     except Exception:
       pass
     await session.stop()
@@ -458,9 +470,6 @@ def prewarm_stream_session_imports(debug_mode: bool = False) -> None:
 
 def webrtcd_thread(host: str, port: int, debug: bool):
   logging.basicConfig(level=logging.CRITICAL, handlers=[logging.StreamHandler()])
-  logging_level = logging.DEBUG if debug else logging.INFO
-  logging.getLogger("WebRTCStream").setLevel(logging_level)
-  logging.getLogger("webrtcd").setLevel(logging_level)
   prewarm_start = time.monotonic()
   prewarm_stream_session_imports(debug)
   prewarm_end = time.monotonic()