system/ui: add stream switching capability to CameraView (#35414)

add stream switching capability to CameraView

system/ui/onroad/augmented_road_view.py

@@ -30,9 +30,6 @@ class AugmentedRoadView(CameraView):
     super().__init__("camerad", stream_type)
 
     self.sm = sm
-    self.stream_type = stream_type
-    self.is_wide_camera = stream_type == VisionStreamType.VISION_STREAM_WIDE_ROAD
-
     self.device_camera: DeviceCameraConfig | None = None
     self.view_from_calib = view_frame_from_device_frame.copy()
     self.view_from_wide_calib = view_frame_from_device_frame.copy()
@@ -130,9 +127,10 @@ class AugmentedRoadView(CameraView):
 
     # Get camera configuration
     device_camera = self.device_camera or DEFAULT_DEVICE_CAMERA
-    intrinsic = device_camera.ecam.intrinsics if self.is_wide_camera else device_camera.fcam.intrinsics
-    calibration = self.view_from_wide_calib if self.is_wide_camera else self.view_from_calib
-    zoom = 2.0 if self.is_wide_camera else 1.1
+    is_wide_camera = self.stream_type == VisionStreamType.VISION_STREAM_WIDE_ROAD
+    intrinsic = device_camera.ecam.intrinsics if is_wide_camera else device_camera.fcam.intrinsics
+    calibration = self.view_from_wide_calib if is_wide_camera else self.view_from_calib
+    zoom = 2.0 if is_wide_camera else 1.1
 
     # Calculate transforms for vanishing point
     inf_point = np.array([1000.0, 0.0, 0.0])
@@ -184,9 +182,13 @@ if __name__ == "__main__":
     "pandaStates", "carParams", "driverMonitoringState", "carState", "driverStateV2",
     "roadCameraState", "wideRoadCameraState", "managerState", "selfdriveState", "longitudinalPlan"])
   road_camera_view = AugmentedRoadView(sm, VisionStreamType.VISION_STREAM_ROAD)
+  print("***press space to switch camera view***")
   try:
     for _ in gui_app.render():
       sm.update(0)
+      if rl.is_key_released(rl.KeyboardKey.KEY_SPACE):
+        is_wide = road_camera_view.stream_type == VisionStreamType.VISION_STREAM_WIDE_ROAD
+        road_camera_view.switch_stream(VisionStreamType.VISION_STREAM_ROAD if is_wide else VisionStreamType.VISION_STREAM_WIDE_ROAD)
       road_camera_view.render(rl.Rectangle(0, 0, gui_app.width, gui_app.height))
   finally:
     road_camera_view.close()

system/ui/widgets/cameraview.py

@@ -3,6 +3,7 @@ import pyray as rl
 
 from openpilot.system.hardware import TICI
 from msgq.visionipc import VisionIpcClient, VisionStreamType, VisionBuf
+from openpilot.common.swaglog import cloudlog
 from openpilot.system.ui.lib.application import gui_app
 from openpilot.system.ui.lib.egl import init_egl, create_egl_image, destroy_egl_image, bind_egl_image_to_texture, EGLImage
 
@@ -57,6 +58,9 @@ else:
 class CameraView:
   def __init__(self, name: str, stream_type: VisionStreamType):
     self.client = VisionIpcClient(name, stream_type, False)
+    self._name = name
+    self._stream_type = stream_type
+
     self._texture_needs_update = True
     self.last_connection_attempt: float = 0.0
     self.shader = rl.load_shader_from_memory(VERTEX_SHADER, FRAME_FRAGMENT_SHADER)
@@ -80,6 +84,18 @@ class CameraView:
       self.egl_texture = rl.load_texture_from_image(temp_image)
       rl.unload_image(temp_image)
 
+  def switch_stream(self, stream_type: VisionStreamType) -> None:
+    if self._stream_type != stream_type:
+      cloudlog.debug(f'switching stream from {self._stream_type} to {stream_type}')
+      self._clear_textures()
+      self.frame = None
+      self._stream_type = stream_type
+      self.client = VisionIpcClient(self._name, stream_type, False)
+
+  @property
+  def stream_type(self) -> VisionStreamType:
+    return self._stream_type
+
   def close(self) -> None:
     self._clear_textures()
 
@@ -92,6 +108,8 @@ class CameraView:
     if self.shader and self.shader.id:
       rl.unload_shader(self.shader)
 
+    self.client = None
+
   def _calc_frame_matrix(self, rect: rl.Rectangle) -> np.ndarray:
     if not self.frame:
       return np.eye(3)
@@ -201,12 +219,12 @@ class CameraView:
       current_time = rl.get_time()
       if current_time - self.last_connection_attempt < CONNECTION_RETRY_INTERVAL:
         return False
-
       self.last_connection_attempt = current_time
 
       if not self.client.connect(False) or not self.client.num_buffers:
         return False
 
+      cloudlog.debug(f"Connected to {self._name} stream: {self._stream_type}, buffers: {self.client.num_buffers}")
       self._clear_textures()
 
       if not TICI: