From 4fa26af61b5aaa719834070fbaf12a9ee0209ebb Mon Sep 17 00:00:00 2001 From: Jason Wen Date: Fri, 26 Jun 2026 21:13:21 -0400 Subject: [PATCH] Revert "remote teleop multi-stream on 1 video track (#38013)" This reverts commit 9844075bb27fee142a8e123a9e03bdf1afd1e008. --- system/athena/athenad.py | 2 +- system/loggerd/loggerd.h | 4 ++-- system/webrtc/device/video.py | 8 +------- system/webrtc/webrtcd.py | 24 ++++++++++-------------- tools/bodyteleop/web.py | 2 +- 5 files changed, 15 insertions(+), 25 deletions(-) diff --git a/system/athena/athenad.py b/system/athena/athenad.py index 9b3262ad99..52d009c341 100755 --- a/system/athena/athenad.py +++ b/system/athena/athenad.py @@ -611,7 +611,7 @@ def startStream(sdp: str) -> dict: if CP.notCar: bridge_services_in.append("testJoystick") - body = StreamRequestBody(sdp, "wideRoad", bridge_services_in, ["carState"]) + body = StreamRequestBody(sdp, ["driver"], bridge_services_in, ["carState"]) try: resp = requests.post(f"http://localhost:{WEBRTCD_PORT}/stream", json=asdict(body), timeout=10) diff --git a/system/loggerd/loggerd.h b/system/loggerd/loggerd.h index 01bce2c9ea..6aa0c8be40 100644 --- a/system/loggerd/loggerd.h +++ b/system/loggerd/loggerd.h @@ -47,8 +47,8 @@ struct EncoderSettings { } static EncoderSettings StreamEncoderSettings() { - int _stream_bitrate = getenv("STREAM_BITRATE") ? atoi(getenv("STREAM_BITRATE")) : 4'000'000; - return EncoderSettings{.encode_type = cereal::EncodeIndex::Type::QCAMERA_H264, .bitrate = _stream_bitrate , .gop_size = 5}; + int _stream_bitrate = getenv("STREAM_BITRATE") ? atoi(getenv("STREAM_BITRATE")) : 1'000'000; + return EncoderSettings{.encode_type = cereal::EncodeIndex::Type::QCAMERA_H264, .bitrate = _stream_bitrate , .gop_size = 15}; } }; diff --git a/system/webrtc/device/video.py b/system/webrtc/device/video.py index b46d022203..50feab4f4a 100644 --- a/system/webrtc/device/video.py +++ b/system/webrtc/device/video.py @@ -19,16 +19,10 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack): dt = DT_DMON if camera_type == "driver" else DT_MDL super().__init__(camera_type, dt) - self._sock = self._make_sock(camera_type) + self._sock = messaging.sub_sock(self.camera_to_sock_mapping[camera_type], conflate=True) self._pts = 0 self._t0_ns = time.monotonic_ns() - def _make_sock(self, camera_type: str) -> messaging.SubSocket: - return messaging.sub_sock(self.camera_to_sock_mapping[camera_type], conflate=True) - - def switch_camera(self, camera_type: str) -> None: - self._sock = self._make_sock(camera_type) - async def recv(self): while True: msg = messaging.recv_one_or_none(self._sock) diff --git a/system/webrtc/webrtcd.py b/system/webrtc/webrtcd.py index 1239a1759c..7a51d45034 100755 --- a/system/webrtc/webrtcd.py +++ b/system/webrtc/webrtcd.py @@ -124,15 +124,18 @@ class DynamicPubMaster(messaging.PubMaster): class StreamSession: shared_pub_master = DynamicPubMaster([]) - def __init__(self, sdp: str, init_camera: str, incoming_services: list[str], outgoing_services: list[str], debug_mode: bool = False): + def __init__(self, sdp: str, cameras: list[str], incoming_services: list[str], outgoing_services: list[str], debug_mode: bool = False): from aiortc.mediastreams import VideoStreamTrack from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack from teleoprtc import WebRTCAnswerBuilder + from teleoprtc.info import parse_info_from_offer + config = parse_info_from_offer(sdp) builder = WebRTCAnswerBuilder(sdp) - self.video_track = LiveStreamVideoStreamTrack(init_camera) if not debug_mode else VideoStreamTrack() - builder.add_video_stream(init_camera, self.video_track) + assert len(cameras) == config.n_expected_camera_tracks, "Incoming stream has misconfigured number of video tracks" + for cam in cameras: + builder.add_video_stream(cam, LiveStreamVideoStreamTrack(cam) if not debug_mode else VideoStreamTrack()) self.stream = builder.stream() self.identifier = str(uuid.uuid4()) @@ -152,8 +155,8 @@ class StreamSession: self._cleanup_done = False self.logger = logging.getLogger("webrtcd") self.logger.info( - "New stream session (%s), init camera %s, incoming services %s, outgoing services %s", - self.identifier, init_camera, incoming_services, outgoing_services, + "New stream session (%s), cameras %s, incoming services %s, outgoing services %s", + self.identifier, cameras, incoming_services, outgoing_services, ) def start(self): @@ -173,13 +176,6 @@ class StreamSession: def message_handler(self, message: bytes): assert self.incoming_bridge is not None try: - msg_json = json.loads(message) - if msg_json.get("type") == "livestreamCameraSwitch" and hasattr(self.video_track, "switch_camera"): - self.video_track.switch_camera(msg_json["data"]["camera"]) - return - - if msg_json.get("type") not in self.incoming_bridge_services: - return self.incoming_bridge.send(message) except Exception: self.logger.exception("Cereal incoming proxy failure") @@ -218,7 +214,7 @@ class StreamSession: @dataclass class StreamRequestBody: sdp: str - initCamera: str + cameras: list[str] bridge_services_in: list[str] = field(default_factory=list) bridge_services_out: list[str] = field(default_factory=list) @@ -240,7 +236,7 @@ async def get_stream(request: 'web.Request'): await s.stop() del stream_dict[sid] - session = StreamSession(body.sdp, body.initCamera, body.bridge_services_in, body.bridge_services_out, debug_mode) + session = StreamSession(body.sdp, body.cameras, body.bridge_services_in, body.bridge_services_out, debug_mode) try: answer = await session.get_answer() except ValueError as e: diff --git a/tools/bodyteleop/web.py b/tools/bodyteleop/web.py index d357561b22..29d23184bd 100644 --- a/tools/bodyteleop/web.py +++ b/tools/bodyteleop/web.py @@ -56,7 +56,7 @@ async def ping(request: 'web.Request'): async def offer(request: 'web.Request'): params = await request.json() - body = StreamRequestBody(params["sdp"], "driver", ["testJoystick"], ["carState"]) + body = StreamRequestBody(params["sdp"], ["driver"], ["testJoystick"], ["carState"]) body_json = json.dumps(dataclasses.asdict(body)) logger.info("Sending offer to webrtcd...")