mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-08-06 08:46:04 +08:00
firestar5683
512 lines
17 KiB
Python
512 lines
17 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from openpilot.selfdrive.ui.mici.onroad import cameraview as mici_cameraview
|
|
from openpilot.selfdrive.ui.onroad import cameraview as big_cameraview
|
|
|
|
|
|
class FakeFrame:
|
|
def __init__(self, frame_id: int, idx: int):
|
|
self.frame_id = frame_id
|
|
self.idx = idx
|
|
|
|
|
|
def _camera_view():
|
|
view = big_cameraview.CameraView.__new__(big_cameraview.CameraView)
|
|
view._name = "camerad"
|
|
view._stream_type = big_cameraview.VisionStreamType.VISION_STREAM_ROAD
|
|
view.frame = None
|
|
view._last_frame_id = -1
|
|
view._regressive_frame_count = 0
|
|
view._texture_needs_update = False
|
|
view._external_texture_id = 0
|
|
view._closed = True
|
|
return view
|
|
|
|
|
|
def test_mici_uses_shared_camera_view():
|
|
assert issubclass(mici_cameraview.CameraView, big_cameraview.CameraView)
|
|
assert mici_cameraview.CameraView._use_upstream_engaged_color
|
|
assert not big_cameraview.CameraView._use_upstream_engaged_color
|
|
|
|
|
|
def test_pending_switch_is_cancelled_when_requested_stream_is_current():
|
|
view = _camera_view()
|
|
view._stream_type = big_cameraview.VisionStreamType.VISION_STREAM_ROAD
|
|
view._target_stream_type = big_cameraview.VisionStreamType.VISION_STREAM_DRIVER
|
|
view._target_client = object()
|
|
view._switching = True
|
|
|
|
view.switch_stream(big_cameraview.VisionStreamType.VISION_STREAM_ROAD)
|
|
|
|
assert view._target_client is None
|
|
assert view._target_stream_type is None
|
|
assert not view._switching
|
|
|
|
|
|
def test_reentry_switch_request_keeps_matching_candidate():
|
|
view = _camera_view()
|
|
candidate = object()
|
|
view._onroad_reentry_pending = True
|
|
view._reentry_stream_selected = True
|
|
view._target_stream_type = view._stream_type
|
|
view._target_client = candidate
|
|
view._switching = True
|
|
|
|
view.switch_stream(view._stream_type)
|
|
|
|
assert view._target_client is candidate
|
|
assert view._target_stream_type == view._stream_type
|
|
assert view._switching
|
|
|
|
|
|
def test_onroad_reentry_selects_requested_stream_before_rendering(monkeypatch):
|
|
view = _camera_view()
|
|
view._name = "camerad"
|
|
view._stream_type = big_cameraview.VisionStreamType.VISION_STREAM_WIDE_ROAD
|
|
view.client = object()
|
|
view._target_client = object()
|
|
view._target_stream_type = big_cameraview.VisionStreamType.VISION_STREAM_ROAD
|
|
view._switching = True
|
|
view._onroad_reentry_pending = True
|
|
view._reentry_stream_selected = False
|
|
view._clear_textures = lambda: None
|
|
clients = []
|
|
|
|
class FakeClient:
|
|
def __init__(self, name, stream_type, conflate):
|
|
self.name = name
|
|
self.stream_type = stream_type
|
|
self.conflate = conflate
|
|
clients.append(self)
|
|
|
|
monkeypatch.setattr(big_cameraview, "VisionIpcClient", FakeClient)
|
|
view.switch_stream(big_cameraview.VisionStreamType.VISION_STREAM_ROAD)
|
|
|
|
assert clients == []
|
|
assert view.client is None
|
|
assert view.stream_type == big_cameraview.VisionStreamType.VISION_STREAM_ROAD
|
|
assert view._target_client is None
|
|
assert view._target_stream_type is None
|
|
assert not view._switching
|
|
assert view._reentry_stream_selected
|
|
|
|
|
|
def test_onroad_reentry_guard_clears_on_first_fresh_frame():
|
|
view = _camera_view()
|
|
view._onroad_reentry_pending = True
|
|
view._reentry_stream_selected = True
|
|
|
|
assert view._accept_frame(FakeFrame(frame_id=1, idx=0), packet_frame_id=1)
|
|
assert not view._onroad_reentry_pending
|
|
assert not view._reentry_stream_selected
|
|
|
|
|
|
def test_standalone_camera_reentry_selects_configured_stream():
|
|
view = _camera_view()
|
|
view._switching = False
|
|
view._onroad_reentry_pending = True
|
|
view._reentry_stream_selected = False
|
|
selected = []
|
|
placeholders = []
|
|
view._select_reentry_stream = lambda stream_type: (
|
|
selected.append(stream_type), setattr(view, "_reentry_stream_selected", True)
|
|
)
|
|
view._draw_placeholder = lambda rect: placeholders.append(rect)
|
|
view._ensure_connection = lambda: False
|
|
|
|
view._render(object())
|
|
|
|
assert selected == [view._stream_type]
|
|
assert len(placeholders) == 1
|
|
|
|
|
|
def test_reused_egl_slot_cannot_move_camera_backwards(monkeypatch):
|
|
monkeypatch.setattr(big_cameraview.cloudlog, "warning", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(big_cameraview, "PC", False)
|
|
view = _camera_view()
|
|
|
|
displayed = FakeFrame(frame_id=10, idx=0)
|
|
assert view._accept_frame(displayed, packet_frame_id=10)
|
|
|
|
# camerad cycles back to this shared slot while it remains displayed.
|
|
displayed.frame_id = 30
|
|
view._observe_displayed_frame()
|
|
|
|
delayed = FakeFrame(frame_id=20, idx=1)
|
|
assert not view._accept_frame(delayed, packet_frame_id=20)
|
|
assert view.frame is displayed
|
|
assert view._last_frame_id == 30
|
|
assert view._regressive_frame_count == 1
|
|
|
|
|
|
def test_texture_camera_accepts_regressive_replay_frame(monkeypatch):
|
|
monkeypatch.setattr(big_cameraview, "PC", True)
|
|
view = _camera_view()
|
|
|
|
assert view._accept_frame(FakeFrame(frame_id=1140, idx=0), packet_frame_id=1140)
|
|
rewind = FakeFrame(frame_id=660, idx=1)
|
|
|
|
assert view._accept_frame(rewind, packet_frame_id=660)
|
|
assert view.frame is rewind
|
|
assert view._last_frame_id == 660
|
|
assert view._regressive_frame_count == 0
|
|
|
|
|
|
def test_newer_camera_frame_is_accepted():
|
|
view = _camera_view()
|
|
view._last_frame_id = 30
|
|
view._regressive_frame_count = 2
|
|
newer = FakeFrame(frame_id=31, idx=2)
|
|
|
|
assert view._accept_frame(newer, packet_frame_id=31)
|
|
assert view.frame is newer
|
|
assert view._last_frame_id == 31
|
|
assert view._regressive_frame_count == 0
|
|
assert view._texture_needs_update
|
|
|
|
|
|
def test_shared_camera_has_upstream_shaders_and_driver_enhancement():
|
|
assert "samplerExternalOES" in big_cameraview.FRAME_FRAGMENT_SHADER_EXTERNAL
|
|
assert "pow(color.rgb, vec3(1.0/1.28))" in big_cameraview.FRAME_FRAGMENT_SHADER_EXTERNAL
|
|
assert "uniform sampler2D texture0" in big_cameraview.FRAME_FRAGMENT_SHADER_YUV
|
|
assert "uniform sampler2D texture1" in big_cameraview.FRAME_FRAGMENT_SHADER_YUV
|
|
assert "uniform int enhance_driver" in big_cameraview.FRAME_FRAGMENT_SHADER_EXTERNAL
|
|
assert "uniform int enhance_driver" in big_cameraview.FRAME_FRAGMENT_SHADER_YUV
|
|
assert "uniform int engaged" not in big_cameraview.FRAME_FRAGMENT_SHADER_EXTERNAL
|
|
assert "uniform int engaged" not in big_cameraview.FRAME_FRAGMENT_SHADER_YUV
|
|
assert hasattr(big_cameraview.CameraView, "_render_egl")
|
|
assert hasattr(big_cameraview.CameraView, "_fallback_to_textures")
|
|
|
|
|
|
def test_shared_camera_falls_back_after_repeated_regressive_frames(monkeypatch):
|
|
monkeypatch.setattr(big_cameraview.cloudlog, "warning", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(big_cameraview, "PC", False)
|
|
view = _camera_view()
|
|
view._use_egl = True
|
|
view.frame = FakeFrame(frame_id=30, idx=0)
|
|
view._last_frame_id = 30
|
|
fallback_reasons = []
|
|
view._fallback_to_textures = fallback_reasons.append
|
|
|
|
for frame_id in (20, 19, 18):
|
|
assert not view._accept_frame(FakeFrame(frame_id=frame_id, idx=1), packet_frame_id=frame_id)
|
|
|
|
assert fallback_reasons == ["repeated regressive frames"]
|
|
assert view.frame.frame_id == 30
|
|
|
|
|
|
def test_shared_camera_fallback_reloads_texture_backend(monkeypatch):
|
|
view = _camera_view()
|
|
view._use_egl = True
|
|
view.shader = SimpleNamespace(id=1)
|
|
events = []
|
|
view._clear_textures = lambda: events.append("clear")
|
|
view._load_frame_shader = lambda: events.append(("shader", view._use_egl))
|
|
view._initialize_textures = lambda: events.append("textures")
|
|
monkeypatch.setattr(big_cameraview.cloudlog, "error", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(big_cameraview.rl, "unload_shader", lambda _shader: events.append("unload_shader"))
|
|
|
|
view._fallback_to_textures("test")
|
|
|
|
assert events == ["clear", "unload_shader", ("shader", False), "textures"]
|
|
assert not view._use_egl
|
|
|
|
|
|
def test_connection_retry_does_not_wait_for_advertisement_and_uses_fresh_candidate(monkeypatch):
|
|
view = _camera_view()
|
|
view._name = "camerad"
|
|
view._clear_textures = lambda: None
|
|
view.client = SimpleNamespace(is_connected=lambda: False)
|
|
view._target_client = None
|
|
view._target_stream_type = None
|
|
view._switching = False
|
|
view.available_streams = []
|
|
view.last_connection_attempt = 0.0
|
|
|
|
candidates = []
|
|
|
|
class FakeClient:
|
|
@staticmethod
|
|
def available_streams(_name, block=False):
|
|
pytest.fail("startup connection waited for stream advertisement")
|
|
|
|
def __init__(self, *_args, **_kwargs):
|
|
candidates.append(self)
|
|
self.connected = False
|
|
self.num_buffers = 0
|
|
|
|
def is_connected(self):
|
|
return self.connected
|
|
|
|
def connect(self, _block):
|
|
return False
|
|
|
|
monkeypatch.setattr(big_cameraview, "VisionIpcClient", FakeClient)
|
|
monkeypatch.setattr(big_cameraview.rl, "get_time", lambda: 1.0)
|
|
|
|
assert not view._ensure_connection()
|
|
assert view.client is None
|
|
assert len(candidates) == 1
|
|
|
|
monkeypatch.setattr(big_cameraview.rl, "get_time", lambda: 1.3)
|
|
assert not view._ensure_connection()
|
|
assert len(candidates) == 2
|
|
assert candidates[0] is not candidates[1]
|
|
|
|
|
|
def test_candidate_is_not_active_until_first_consistent_frame(monkeypatch):
|
|
view = _camera_view()
|
|
view._name = "camerad"
|
|
view._clear_textures = lambda: None
|
|
view._initialize_textures = lambda: None
|
|
view.client = None
|
|
view._target_client = None
|
|
view._target_stream_type = None
|
|
view._switching = False
|
|
view.available_streams = []
|
|
view.last_connection_attempt = 0.0
|
|
|
|
class FakeClient:
|
|
@staticmethod
|
|
def available_streams(_name, block=False):
|
|
return [view._stream_type]
|
|
|
|
def __init__(self, *_args, **_kwargs):
|
|
self.connected = False
|
|
self.num_buffers = 1
|
|
self.frame_id = -1
|
|
self.frames = [None, FakeFrame(frame_id=42, idx=0)]
|
|
|
|
def is_connected(self):
|
|
return self.connected
|
|
|
|
def connect(self, _block):
|
|
self.connected = True
|
|
return True
|
|
|
|
def recv(self, timeout_ms=0):
|
|
frame = self.frames.pop(0)
|
|
if frame is not None:
|
|
self.frame_id = frame.frame_id
|
|
return frame
|
|
|
|
monkeypatch.setattr(big_cameraview, "VisionIpcClient", FakeClient)
|
|
monkeypatch.setattr(big_cameraview.rl, "get_time", lambda: 1.0)
|
|
|
|
assert not view._ensure_connection()
|
|
assert view.client is None
|
|
candidate = view._target_client
|
|
assert candidate is not None
|
|
|
|
assert view._ensure_connection()
|
|
assert view.client is candidate
|
|
assert view.frame.frame_id == 42
|
|
assert view._target_client is None
|
|
assert not view._switching
|
|
|
|
|
|
def test_inconsistent_candidate_frame_is_discarded(monkeypatch):
|
|
view = _camera_view()
|
|
view._name = "camerad"
|
|
view._clear_textures = lambda: None
|
|
view.client = None
|
|
view._target_client = None
|
|
view._target_stream_type = None
|
|
view._switching = False
|
|
view.available_streams = []
|
|
view.last_connection_attempt = 0.0
|
|
|
|
class FakeClient:
|
|
@staticmethod
|
|
def available_streams(_name, block=False):
|
|
return [view._stream_type]
|
|
|
|
def __init__(self, *_args, **_kwargs):
|
|
self.connected = False
|
|
self.num_buffers = 1
|
|
self.frame_id = 10
|
|
|
|
def is_connected(self):
|
|
return self.connected
|
|
|
|
def connect(self, _block):
|
|
self.connected = True
|
|
return True
|
|
|
|
def recv(self, timeout_ms=0):
|
|
return FakeFrame(frame_id=9, idx=0)
|
|
|
|
monkeypatch.setattr(big_cameraview, "VisionIpcClient", FakeClient)
|
|
monkeypatch.setattr(big_cameraview.rl, "get_time", lambda: 1.0)
|
|
|
|
assert not view._ensure_connection()
|
|
assert view.client is None
|
|
assert view._target_client is None
|
|
assert view._target_stream_type == view._stream_type
|
|
assert not view._switching
|
|
|
|
|
|
def test_disconnected_candidate_is_discarded_without_reconnect():
|
|
view = _camera_view()
|
|
|
|
class Candidate:
|
|
num_buffers = 1
|
|
|
|
def __init__(self):
|
|
self.connected = True
|
|
|
|
def is_connected(self):
|
|
return self.connected
|
|
|
|
def connect(self, _block):
|
|
pytest.fail("discarded candidate was reconnected")
|
|
|
|
def recv(self, timeout_ms=0):
|
|
self.connected = False
|
|
return None
|
|
|
|
candidate = Candidate()
|
|
view._target_client = candidate
|
|
view._target_stream_type = view._stream_type
|
|
view._switching = True
|
|
|
|
view._handle_switch()
|
|
|
|
assert view._target_client is None
|
|
assert not view._switching
|
|
assert view._target_stream_type == view._stream_type
|
|
|
|
|
|
def test_steady_state_packet_content_mismatch_is_rejected():
|
|
view = _camera_view()
|
|
displayed = FakeFrame(frame_id=10, idx=0)
|
|
assert view._accept_frame(displayed, packet_frame_id=10)
|
|
|
|
delayed = FakeFrame(frame_id=12, idx=1)
|
|
assert not view._accept_frame(delayed, packet_frame_id=11)
|
|
assert view.frame is displayed
|
|
assert view._last_frame_id == 10
|
|
|
|
|
|
def test_egl_image_creation_failure_is_reported(monkeypatch):
|
|
view = _camera_view()
|
|
view.frame = SimpleNamespace(idx=0, width=1928, height=1208, stride=2048, fd=7, uv_offset=2473984)
|
|
view.egl_texture = SimpleNamespace(id=1)
|
|
view._external_texture_id = 11
|
|
view.egl_images = {}
|
|
monkeypatch.setattr(big_cameraview, "create_egl_image", lambda *_args: None)
|
|
|
|
assert not view._render_egl(None, None)
|
|
assert view.egl_images == {}
|
|
|
|
|
|
def test_invalid_egl_texture_is_reported_without_binding(monkeypatch):
|
|
view = _camera_view()
|
|
view.frame = SimpleNamespace(idx=0)
|
|
view.egl_texture = SimpleNamespace(id=0)
|
|
view._external_texture_id = 11
|
|
view.egl_images = {0: object()}
|
|
monkeypatch.setattr(big_cameraview, "bind_egl_image_to_texture",
|
|
lambda *_args: pytest.fail("invalid EGL texture was bound"))
|
|
|
|
assert not view._render_egl(None, None)
|
|
|
|
|
|
def test_invalid_external_texture_is_reported_without_binding(monkeypatch):
|
|
view = _camera_view()
|
|
view.frame = SimpleNamespace(idx=0)
|
|
view.egl_texture = SimpleNamespace(id=7)
|
|
view.egl_images = {0: object()}
|
|
monkeypatch.setattr(big_cameraview, "bind_egl_image_to_texture",
|
|
lambda *_args: pytest.fail("invalid external texture was bound"))
|
|
|
|
assert not view._render_egl(None, None)
|
|
|
|
|
|
def test_egl_render_always_ends_shader_mode(monkeypatch):
|
|
view = _camera_view()
|
|
view.frame = SimpleNamespace(idx=0, width=1928, height=1208)
|
|
view.egl_texture = SimpleNamespace(id=1, width=0, height=0)
|
|
view._external_texture_id = 11
|
|
view.egl_images = {0: object()}
|
|
view.shader = SimpleNamespace(id=1)
|
|
view._update_shader_state = lambda: None
|
|
events = []
|
|
monkeypatch.setattr(big_cameraview, "bind_egl_image_to_texture", lambda *_args: None)
|
|
monkeypatch.setattr(big_cameraview.rl, "begin_shader_mode", lambda *_args: events.append("begin"))
|
|
def fail_draw(*_args):
|
|
raise RuntimeError("draw failed")
|
|
|
|
monkeypatch.setattr(big_cameraview.rl, "draw_texture_pro", fail_draw)
|
|
monkeypatch.setattr(big_cameraview.rl, "end_shader_mode", lambda: events.append("end"))
|
|
|
|
with pytest.raises(RuntimeError, match="draw failed"):
|
|
view._render_egl(None, None)
|
|
|
|
assert events == ["begin", "end"]
|
|
|
|
|
|
def test_egl_render_keeps_external_and_raylib_texture_targets_separate(monkeypatch):
|
|
view = _camera_view()
|
|
image = object()
|
|
view.frame = SimpleNamespace(idx=3, width=1928, height=1208)
|
|
view.egl_texture = SimpleNamespace(id=7, width=1, height=1)
|
|
view._external_texture_id = 11
|
|
view.egl_images = {3: image}
|
|
view.shader = object()
|
|
view._update_shader_state = lambda: None
|
|
bound = []
|
|
drawn = []
|
|
monkeypatch.setattr(big_cameraview, "bind_egl_image_to_texture",
|
|
lambda texture_id, egl_image: bound.append((texture_id, egl_image)))
|
|
monkeypatch.setattr(big_cameraview.rl, "begin_shader_mode", lambda _shader: None)
|
|
monkeypatch.setattr(big_cameraview.rl, "end_shader_mode", lambda: None)
|
|
monkeypatch.setattr(big_cameraview.rl, "draw_texture_pro", lambda texture, *_args: drawn.append(texture.id))
|
|
|
|
assert view._render_egl(None, None)
|
|
assert bound == [(11, image)]
|
|
assert drawn == [7]
|
|
|
|
|
|
def test_driver_enhancement_tracks_active_stream(monkeypatch):
|
|
view = _camera_view()
|
|
view.shader = SimpleNamespace(id=1)
|
|
view._enhance_driver_loc = 2
|
|
view._enhance_driver_val = [0]
|
|
values = []
|
|
monkeypatch.setattr(big_cameraview.rl, "set_shader_value",
|
|
lambda _shader, _loc, value, _type: values.append(value[0]))
|
|
|
|
view._stream_type = big_cameraview.VisionStreamType.VISION_STREAM_ROAD
|
|
view._update_shader_state()
|
|
view._stream_type = big_cameraview.VisionStreamType.VISION_STREAM_DRIVER
|
|
view._update_shader_state()
|
|
view._stream_type = big_cameraview.VisionStreamType.VISION_STREAM_WIDE_ROAD
|
|
view._update_shader_state()
|
|
|
|
assert values == [0, 1, 0]
|
|
|
|
|
|
def test_texture_fallback_survives_egl_cleanup_failure(monkeypatch):
|
|
view = _camera_view()
|
|
view._use_egl = True
|
|
view.shader = SimpleNamespace(id=1)
|
|
events = []
|
|
|
|
def fail_cleanup():
|
|
raise RuntimeError("cleanup failed")
|
|
|
|
view._clear_textures = fail_cleanup
|
|
view._load_frame_shader = lambda: events.append(("shader", view._use_egl))
|
|
view._initialize_textures = lambda: events.append("textures")
|
|
monkeypatch.setattr(big_cameraview.cloudlog, "error", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(big_cameraview.cloudlog, "exception", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(big_cameraview.rl, "unload_shader", lambda _shader: events.append("unload_shader"))
|
|
|
|
view._fallback_to_textures("test")
|
|
|
|
assert not view._use_egl
|
|
assert events == ["unload_shader", ("shader", False), "textures"]
|