From 45d8bcd7f3f2da7e03d95f45218b49e00fb06963 Mon Sep 17 00:00:00 2001 From: stef <19478336+stefpi@users.noreply.github.com> Date: Fri, 19 Jun 2026 21:20:55 -0400 Subject: [PATCH] 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 | 1 + system/athena/athenad.py | 34 ++++++++++++--------------- system/manager/process_config.py | 12 +++++++--- system/webrtc/helpers.py | 40 ++++++++++++++++++++++++++++++++ system/webrtc/models.py | 9 ------- system/webrtc/webrtcd.py | 27 ++++++++++++++------- 6 files changed, 83 insertions(+), 40 deletions(-) create mode 100644 system/webrtc/helpers.py delete mode 100644 system/webrtc/models.py diff --git a/common/params_keys.h b/common/params_keys.h index 546a53c76c..01b051545a 100644 --- a/common/params_keys.h +++ b/common/params_keys.h @@ -58,6 +58,7 @@ inline static std::unordered_map 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}}, diff --git a/system/athena/athenad.py b/system/athena/athenad.py index d72745a389..8cadfb857a 100755 --- a/system/athena/athenad.py +++ b/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: diff --git a/system/manager/process_config.py b/system/manager/process_config.py index 7ed4a2dacb..84f0313714 100644 --- a/system/manager/process_config.py +++ b/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)), ] diff --git a/system/webrtc/helpers.py b/system/webrtc/helpers.py new file mode 100644 index 0000000000..776b64573e --- /dev/null +++ b/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.") diff --git a/system/webrtc/models.py b/system/webrtc/models.py deleted file mode 100644 index 5a80576f7d..0000000000 --- a/system/webrtc/models.py +++ /dev/null @@ -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) diff --git a/system/webrtc/webrtcd.py b/system/webrtc/webrtcd.py index c2bceb5d90..da10855a59 100755 --- a/system/webrtc/webrtcd.py +++ b/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()