mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-09-30 19:33:45 +08:00
You & I
This commit is contained in:
@@ -231,8 +231,12 @@ class ModelRenderer(Widget):
|
||||
"""Update model visualization data based on model message"""
|
||||
model_ui_enabled = self._params.get_bool('ModelUI', default=True)
|
||||
custom_path_width, pw = self._param_float_changed('PathWidth', DEFAULT_PATH_WIDTH) if model_ui_enabled else (False, DEFAULT_PATH_WIDTH)
|
||||
custom_lane_line_width, llw = self._param_float_changed('LaneLinesWidth', DEFAULT_LANE_LINES_WIDTH) if model_ui_enabled else (False, DEFAULT_LANE_LINES_WIDTH)
|
||||
custom_road_edge_width, rew = self._param_float_changed('RoadEdgesWidth', DEFAULT_ROAD_EDGES_WIDTH) if model_ui_enabled else (False, DEFAULT_ROAD_EDGES_WIDTH)
|
||||
custom_lane_line_width, llw = (
|
||||
self._param_float_changed('LaneLinesWidth', DEFAULT_LANE_LINES_WIDTH) if model_ui_enabled else (False, DEFAULT_LANE_LINES_WIDTH)
|
||||
)
|
||||
custom_road_edge_width, rew = (
|
||||
self._param_float_changed('RoadEdgesWidth', DEFAULT_ROAD_EDGES_WIDTH) if model_ui_enabled else (False, DEFAULT_ROAD_EDGES_WIDTH)
|
||||
)
|
||||
custom_path_edge_width, pew = self._param_float_changed('PathEdgeWidth', DEFAULT_PATH_EDGE_WIDTH) if model_ui_enabled else (False, DEFAULT_PATH_EDGE_WIDTH)
|
||||
|
||||
path_width = self._path_width_to_half_m(pw) if custom_path_width else 0.9
|
||||
@@ -253,15 +257,14 @@ class ModelRenderer(Widget):
|
||||
unclipped_max_distance = np.clip(path_x_array[-1], MIN_DRAW_DISTANCE, MAX_DRAW_DISTANCE)
|
||||
unclipped_max_idx = self._get_path_length_idx(self._lane_lines[0].raw_points[:, 0], unclipped_max_distance)
|
||||
|
||||
# Update lane lines using raw points
|
||||
for i, lane_line in enumerate(self._lane_lines):
|
||||
lane_line.projected_points = self._map_line_to_polygon(
|
||||
lane_line.raw_points, lane_line_width_m * self._lane_line_probs[i], 0.0, unclipped_max_idx, unclipped_max_distance, clip_by_lead=True
|
||||
)
|
||||
|
||||
# Update road edges using raw points
|
||||
for road_edge in self._road_edges:
|
||||
road_edge.projected_points = self._map_line_to_polygon(road_edge.raw_points, road_edge_width_m, 0.0, unclipped_max_idx, unclipped_max_distance, clip_by_lead=True)
|
||||
# Update lane lines and road edges using batched projection
|
||||
lines = [*self._lane_lines, *self._road_edges]
|
||||
widths = [lane_line_width_m * prob for prob in self._lane_line_probs] + [road_edge_width_m] * len(self._road_edges)
|
||||
polygons = self._map_lines_to_polygons(
|
||||
[line.raw_points for line in lines], widths, 0.0, unclipped_max_idx, unclipped_max_distance, clip_by_lead=True
|
||||
)
|
||||
for line, polygon in zip(lines, polygons, strict=True):
|
||||
line.projected_points = polygon
|
||||
|
||||
# Update path using raw points
|
||||
max_distance = unclipped_max_distance
|
||||
@@ -581,7 +584,7 @@ class ModelRenderer(Widget):
|
||||
text_lines.append(f"{distance_string} {lead_distance_unit} (Desired: {desired_distance})")
|
||||
else:
|
||||
text_lines.append(f"{distance_string} {lead_distance_unit}")
|
||||
|
||||
|
||||
text_lines.append(f"{speed_string}{lead_speed_unit}")
|
||||
|
||||
v_ego = max(ui_state.sm["carState"].vEgo, 0.0)
|
||||
@@ -839,15 +842,72 @@ class ModelRenderer(Widget):
|
||||
|
||||
return (x, y)
|
||||
|
||||
def _map_line_to_polygon(self, line: np.ndarray, y_off: float, z_off: float, max_idx: int, max_distance: float, allow_invert: bool = True, clip_by_lead: bool = False) -> np.ndarray:
|
||||
"""Convert 3D line to 2D polygon for rendering."""
|
||||
def _get_active_leads(self) -> list[tuple[str, object]]:
|
||||
active_leads = []
|
||||
sm = ui_state.sm
|
||||
if sm.valid.get("radarState", False):
|
||||
rs = sm["radarState"]
|
||||
if rs.leadOne and rs.leadOne.status:
|
||||
active_leads.append(("ego", rs.leadOne))
|
||||
if rs.leadTwo and rs.leadTwo.status:
|
||||
active_leads.append(("ego", rs.leadTwo))
|
||||
if sm.valid.get("starpilotRadarState", False):
|
||||
srs = sm["starpilotRadarState"]
|
||||
if srs.leadLeft and srs.leadLeft.status:
|
||||
active_leads.append(("left", srs.leadLeft))
|
||||
if srs.leadRight and srs.leadRight.status:
|
||||
active_leads.append(("right", srs.leadRight))
|
||||
return active_leads
|
||||
|
||||
def _clip_line_by_lead(self, points: np.ndarray, active_leads: list[tuple[str, object]]) -> np.ndarray:
|
||||
if not active_leads or points.shape[0] == 0:
|
||||
return points
|
||||
|
||||
x = points[:, 0]
|
||||
y = points[:, 1]
|
||||
clipped_mask = np.zeros(len(points), dtype=bool)
|
||||
|
||||
for lead_type, lead in active_leads:
|
||||
lead_x = lead.dRel
|
||||
lead_y = -lead.yRel
|
||||
y_limit = 5.0 if lead_type == "ego" else 1.8
|
||||
in_box = (x >= lead_x - 1.5) & (x <= lead_x + 5.0) & (y >= lead_y - y_limit) & (y <= lead_y + y_limit)
|
||||
clipped_mask |= in_box
|
||||
|
||||
indices = np.where(clipped_mask)[0]
|
||||
if indices.size > 0:
|
||||
first_clip_idx = indices[0]
|
||||
if first_clip_idx > 0:
|
||||
p_prev = points[first_clip_idx - 1]
|
||||
p_curr = points[first_clip_idx]
|
||||
|
||||
best_lead_x = None
|
||||
for lead_type, lead in active_leads:
|
||||
lead_x = lead.dRel
|
||||
lead_y = -lead.yRel
|
||||
y_limit = 5.0 if lead_type == "ego" else 1.8
|
||||
if (lead_x - 1.5) <= p_curr[0] <= (lead_x + 5.0) and (lead_y - y_limit) <= p_curr[1] <= (lead_y + y_limit):
|
||||
best_lead_x = lead_x
|
||||
break
|
||||
|
||||
if best_lead_x is not None:
|
||||
x_clip = best_lead_x - 1.5
|
||||
x0, x1 = p_prev[0], p_curr[0]
|
||||
if x1 > x0:
|
||||
t = (x_clip - x0) / (x1 - x0)
|
||||
y_clip = p_prev[1] + t * (p_curr[1] - p_prev[1])
|
||||
z_clip = p_prev[2] + t * (p_curr[2] - p_prev[2])
|
||||
interp_pt = np.array([x_clip, y_clip, z_clip], dtype=points.dtype)
|
||||
return np.concatenate((points[:first_clip_idx], interp_pt[None, :]), axis=0)
|
||||
return points[:first_clip_idx]
|
||||
return np.empty((0, 3), dtype=points.dtype)
|
||||
return points
|
||||
|
||||
def _prepare_line_points(self, line: np.ndarray, max_idx: int, max_distance: float, active_leads: list[tuple[str, object]] | None = None) -> np.ndarray:
|
||||
if line.shape[0] == 0:
|
||||
return np.empty((0, 2), dtype=np.float32)
|
||||
return np.empty((0, 3), dtype=np.float32)
|
||||
|
||||
# Slice points and filter non-negative x-coordinates
|
||||
points = line[:max_idx + 1]
|
||||
|
||||
# Interpolate around max_idx so path end is smooth (max_distance is always >= p0.x)
|
||||
if 0 < max_idx < line.shape[0] - 1:
|
||||
p0 = line[max_idx]
|
||||
p1 = line[max_idx + 1]
|
||||
@@ -858,130 +918,99 @@ class ModelRenderer(Widget):
|
||||
points = np.concatenate((points, interp_point[None, :]), axis=0)
|
||||
|
||||
points = points[points[:, 0] >= 0]
|
||||
if points.shape[0] == 0:
|
||||
return np.empty((0, 2), dtype=np.float32)
|
||||
if active_leads:
|
||||
points = self._clip_line_by_lead(points, active_leads)
|
||||
return points
|
||||
|
||||
# Lead vehicle clipping to prevent drawing through/on top of lead vehicles
|
||||
if clip_by_lead:
|
||||
active_leads = []
|
||||
sm = ui_state.sm
|
||||
if sm.valid.get("radarState", False):
|
||||
rs = sm["radarState"]
|
||||
if rs.leadOne and rs.leadOne.status:
|
||||
active_leads.append(("ego", rs.leadOne))
|
||||
if rs.leadTwo and rs.leadTwo.status:
|
||||
active_leads.append(("ego", rs.leadTwo))
|
||||
if sm.valid.get("starpilotRadarState", False):
|
||||
srs = sm["starpilotRadarState"]
|
||||
if srs.leadLeft and srs.leadLeft.status:
|
||||
active_leads.append(("left", srs.leadLeft))
|
||||
if srs.leadRight and srs.leadRight.status:
|
||||
active_leads.append(("right", srs.leadRight))
|
||||
def _map_lines_to_polygons(
|
||||
self,
|
||||
lines: list[np.ndarray],
|
||||
y_offs: list[float],
|
||||
z_off: float = 0.0,
|
||||
max_idx: int = 100,
|
||||
max_distance: float = MAX_DRAW_DISTANCE,
|
||||
allow_invert: bool = True,
|
||||
clip_by_lead: bool = False,
|
||||
) -> list[np.ndarray]:
|
||||
if not lines:
|
||||
return []
|
||||
|
||||
if active_leads:
|
||||
x = points[:, 0]
|
||||
y = points[:, 1]
|
||||
clipped_mask = np.zeros(len(points), dtype=bool)
|
||||
active_leads = self._get_active_leads() if clip_by_lead else None
|
||||
prepared_lines = [self._prepare_line_points(line, max_idx, max_distance, active_leads) for line in lines]
|
||||
counts = [len(p) for p in prepared_lines]
|
||||
total = sum(counts)
|
||||
if total == 0:
|
||||
return [np.empty((0, 2), dtype=np.float32) for _ in lines]
|
||||
|
||||
for lead_type, lead in active_leads:
|
||||
lead_x = lead.dRel
|
||||
lead_y = -lead.yRel
|
||||
y_limit = 5.0 if lead_type == "ego" else 1.8
|
||||
# Collision box for lead car: x ∈ [lead_x - 1.5, lead_x + 5.0], y ∈ [lead_y - y_limit, lead_y + y_limit]
|
||||
in_box = (x >= lead_x - 1.5) & (x <= lead_x + 5.0) & (y >= lead_y - y_limit) & (y <= lead_y + y_limit)
|
||||
clipped_mask |= in_box
|
||||
offsets = np.zeros((2, total, 3), dtype=np.float32)
|
||||
offsets[1, :, 1] = np.repeat(y_offs, counts)
|
||||
offsets[0, :, 1] = -offsets[1, :, 1]
|
||||
if z_off != 0.0:
|
||||
offsets[:, :, 2] = z_off
|
||||
|
||||
indices = np.where(clipped_mask)[0]
|
||||
if indices.size > 0:
|
||||
first_clip_idx = indices[0]
|
||||
if first_clip_idx > 0:
|
||||
# Interpolate to the exact entry boundary (lead_x - 1.5)
|
||||
p_prev = points[first_clip_idx - 1]
|
||||
p_curr = points[first_clip_idx]
|
||||
|
||||
# Find the lead vehicle that triggered the clip
|
||||
best_lead_x = None
|
||||
for lead_type, lead in active_leads:
|
||||
lead_x = lead.dRel
|
||||
lead_y = -lead.yRel
|
||||
y_limit = 5.0 if lead_type == "ego" else 1.8
|
||||
if (lead_x - 1.5) <= p_curr[0] <= (lead_x + 5.0) and (lead_y - y_limit) <= p_curr[1] <= (lead_y + y_limit):
|
||||
best_lead_x = lead_x
|
||||
break
|
||||
concat_pts = np.concatenate(prepared_lines)
|
||||
points_3d = concat_pts[None, :, :] + offsets
|
||||
proj = (self._car_space_transform @ points_3d.reshape(2 * total, 3).T).reshape(3, 2, total)
|
||||
|
||||
if best_lead_x is not None:
|
||||
x_clip = best_lead_x - 1.5
|
||||
x0, x1 = p_prev[0], p_curr[0]
|
||||
if x1 > x0:
|
||||
t = (x_clip - x0) / (x1 - x0)
|
||||
y_clip = p_prev[1] + t * (p_curr[1] - p_prev[1])
|
||||
z_clip = p_prev[2] + t * (p_curr[2] - p_prev[2])
|
||||
interp_pt = np.array([x_clip, y_clip, z_clip], dtype=points.dtype)
|
||||
points = np.concatenate((points[:first_clip_idx], interp_pt[None, :]), axis=0)
|
||||
else:
|
||||
points = points[:first_clip_idx]
|
||||
else:
|
||||
points = points[:first_clip_idx]
|
||||
else:
|
||||
points = np.empty((0, 3), dtype=points.dtype)
|
||||
valid_proj = (np.abs(proj[2, 0]) >= 1e-6) & (np.abs(proj[2, 1]) >= 1e-6)
|
||||
|
||||
if points.shape[0] == 0:
|
||||
return np.empty((0, 2), dtype=np.float32)
|
||||
left_screen = np.zeros((2, total), dtype=np.float32)
|
||||
right_screen = np.zeros((2, total), dtype=np.float32)
|
||||
np.divide(proj[:2, 0], proj[2, 0], out=left_screen, where=valid_proj)
|
||||
np.divide(proj[:2, 1], proj[2, 1], out=right_screen, where=valid_proj)
|
||||
|
||||
N = points.shape[0]
|
||||
# Generate left and right 3D points in one array using broadcasting
|
||||
offsets = np.array([[0, -y_off, z_off], [0, y_off, z_off]], dtype=np.float32)
|
||||
points_3d = points[None, :, :] + offsets[:, None, :] # Shape: 2xNx3
|
||||
points_3d = points_3d.reshape(2 * N, 3) # Shape: (2*N)x3
|
||||
|
||||
# Transform all points to projected space in one operation
|
||||
proj = self._car_space_transform @ points_3d.T # Shape: 3x(2*N)
|
||||
proj = proj.reshape(3, 2, N)
|
||||
left_proj = proj[:, 0, :]
|
||||
right_proj = proj[:, 1, :]
|
||||
|
||||
# Filter points where z is sufficiently large
|
||||
valid_proj = (np.abs(left_proj[2]) >= 1e-6) & (np.abs(right_proj[2]) >= 1e-6)
|
||||
if not np.any(valid_proj):
|
||||
return np.empty((0, 2), dtype=np.float32)
|
||||
|
||||
# Compute screen coordinates
|
||||
left_screen = left_proj[:2, valid_proj] / left_proj[2, valid_proj][None, :]
|
||||
right_screen = right_proj[:2, valid_proj] / right_proj[2, valid_proj][None, :]
|
||||
|
||||
# Define clip region bounds
|
||||
clip = self._clip_region
|
||||
x_min, x_max = clip.x, clip.x + clip.width
|
||||
y_min, y_max = clip.y, clip.y + clip.height
|
||||
if clip is not None:
|
||||
x_min, x_max = clip.x, clip.x + clip.width
|
||||
y_min, y_max = clip.y, clip.y + clip.height
|
||||
in_clip = (
|
||||
valid_proj &
|
||||
(left_screen[0] >= x_min) & (left_screen[0] <= x_max) &
|
||||
(left_screen[1] >= y_min) & (left_screen[1] <= y_max) &
|
||||
(right_screen[0] >= x_min) & (right_screen[0] <= x_max) &
|
||||
(right_screen[1] >= y_min) & (right_screen[1] <= y_max)
|
||||
)
|
||||
else:
|
||||
in_clip = valid_proj
|
||||
|
||||
# Filter points within clip region
|
||||
left_in_clip = (
|
||||
(left_screen[0] >= x_min) & (left_screen[0] <= x_max) &
|
||||
(left_screen[1] >= y_min) & (left_screen[1] <= y_max)
|
||||
)
|
||||
right_in_clip = (
|
||||
(right_screen[0] >= x_min) & (right_screen[0] <= x_max) &
|
||||
(right_screen[1] >= y_min) & (right_screen[1] <= y_max)
|
||||
)
|
||||
both_in_clip = left_in_clip & right_in_clip
|
||||
polygons = []
|
||||
start = 0
|
||||
for count in counts:
|
||||
end = start + count
|
||||
if count == 0:
|
||||
polygons.append(np.empty((0, 2), dtype=np.float32))
|
||||
start = end
|
||||
continue
|
||||
|
||||
if not np.any(both_in_clip):
|
||||
return np.empty((0, 2), dtype=np.float32)
|
||||
mask = in_clip[start:end]
|
||||
if not np.any(mask):
|
||||
polygons.append(np.empty((0, 2), dtype=np.float32))
|
||||
else:
|
||||
l_scr = left_screen[:, start:end][:, mask]
|
||||
r_scr = right_screen[:, start:end][:, mask]
|
||||
if not allow_invert and l_scr.shape[1] > 1:
|
||||
y = l_scr[1, :]
|
||||
keep = y == np.minimum.accumulate(y)
|
||||
if not np.any(keep):
|
||||
polygons.append(np.empty((0, 2), dtype=np.float32))
|
||||
start = end
|
||||
continue
|
||||
l_scr = l_scr[:, keep]
|
||||
r_scr = r_scr[:, keep]
|
||||
poly = np.vstack((l_scr.T, r_scr[:, ::-1].T)).astype(np.float32)
|
||||
polygons.append(poly)
|
||||
start = end
|
||||
|
||||
# Select valid and clipped points
|
||||
left_screen = left_screen[:, both_in_clip]
|
||||
right_screen = right_screen[:, both_in_clip]
|
||||
return polygons
|
||||
|
||||
# Handle Y-coordinate inversion on hills
|
||||
if not allow_invert and left_screen.shape[1] > 1:
|
||||
y = left_screen[1, :] # y-coordinates
|
||||
keep = y == np.minimum.accumulate(y)
|
||||
if not np.any(keep):
|
||||
return np.empty((0, 2), dtype=np.float32)
|
||||
left_screen = left_screen[:, keep]
|
||||
right_screen = right_screen[:, keep]
|
||||
|
||||
return np.vstack((left_screen.T, right_screen[:, ::-1].T)).astype(np.float32)
|
||||
def _map_line_to_polygon(
|
||||
self, line: np.ndarray, y_off: float, z_off: float, max_idx: int, max_distance: float,
|
||||
allow_invert: bool = True, clip_by_lead: bool = False,
|
||||
) -> np.ndarray:
|
||||
"""Convert 3D line to 2D polygon for rendering."""
|
||||
return self._map_lines_to_polygons(
|
||||
[line], [y_off], z_off, max_idx, max_distance, allow_invert=allow_invert, clip_by_lead=clip_by_lead,
|
||||
)[0]
|
||||
|
||||
@staticmethod
|
||||
def _hsla_to_color(h, s, l, a):
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
import sys
|
||||
import types
|
||||
from types import SimpleNamespace
|
||||
|
||||
# Mock C-extensions and modules not available in x86 dev environment
|
||||
if "cereal.messaging" not in sys.modules:
|
||||
try:
|
||||
import cereal.messaging # noqa: F401
|
||||
except ImportError:
|
||||
class DummyMaster:
|
||||
pass
|
||||
|
||||
msg_mod = types.ModuleType("cereal.messaging")
|
||||
msg_mod.SubMaster = DummyMaster
|
||||
msg_mod.PubMaster = DummyMaster
|
||||
sys.modules["cereal.messaging"] = msg_mod
|
||||
|
||||
if "openpilot.selfdrive.locationd.calibrationd" not in sys.modules:
|
||||
try:
|
||||
import openpilot.selfdrive.locationd.calibrationd
|
||||
except ImportError:
|
||||
calib_mod = types.ModuleType("openpilot.selfdrive.locationd.calibrationd")
|
||||
calib_mod.HEIGHT_INIT = [1.22]
|
||||
sys.modules["openpilot.selfdrive.locationd.calibrationd"] = calib_mod
|
||||
|
||||
if "openpilot.selfdrive.ui.ui_state" not in sys.modules:
|
||||
try:
|
||||
import openpilot.selfdrive.ui.ui_state # noqa: F401
|
||||
except ImportError:
|
||||
ui_state_mod = types.ModuleType("openpilot.selfdrive.ui.ui_state")
|
||||
ui_state_mod.ui_state = SimpleNamespace(
|
||||
sm=SimpleNamespace(valid={}),
|
||||
status=0,
|
||||
always_on_lateral_active=False,
|
||||
is_metric=False,
|
||||
starpilot_toggles={},
|
||||
started_frame=0,
|
||||
)
|
||||
ui_state_mod.UIStatus = SimpleNamespace(DISENGAGED=0, ENGAGED=1, OVERRIDE=2)
|
||||
sys.modules["openpilot.selfdrive.ui.ui_state"] = ui_state_mod
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from openpilot.selfdrive.ui.onroad.model_renderer import ModelRenderer
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def renderer():
|
||||
r = object.__new__(ModelRenderer)
|
||||
r._car_space_transform = np.array([
|
||||
[580.0, -480.0, 0.0],
|
||||
[400.0, 0.0, -480.0],
|
||||
[1.0, 0.0, 0.0],
|
||||
], dtype=np.float32)
|
||||
r._transform_dirty = False
|
||||
r._clip_region = SimpleNamespace(x=-500, y=-500, width=3160, height=2080)
|
||||
r._get_active_leads = list
|
||||
return r
|
||||
|
||||
|
||||
def test_batched_projection_matches_single_line_projection(renderer):
|
||||
rng = np.random.default_rng(42)
|
||||
lines = [
|
||||
np.column_stack((np.linspace(0, 100, 33), rng.normal(0, 3, 33), rng.normal(0, 0.5, 33))).astype(np.float32)
|
||||
for _ in range(6)
|
||||
]
|
||||
widths = [0.1, 0.15, 0.15, 0.1, 0.05, 0.05]
|
||||
max_idx = 30
|
||||
max_distance = 90.0
|
||||
|
||||
# Batched call
|
||||
batched_polys = renderer._map_lines_to_polygons(lines, widths, 0.0, max_idx, max_distance)
|
||||
|
||||
# Individual calls via _map_line_to_polygon
|
||||
individual_polys = [
|
||||
renderer._map_line_to_polygon(line, w, 0.0, max_idx, max_distance)
|
||||
for line, w in zip(lines, widths, strict=True)
|
||||
]
|
||||
|
||||
assert len(batched_polys) == len(lines)
|
||||
for b_poly, i_poly in zip(batched_polys, individual_polys, strict=True):
|
||||
assert b_poly.dtype == np.float32
|
||||
assert b_poly.shape == i_poly.shape
|
||||
np.testing.assert_allclose(b_poly, i_poly, rtol=1e-5, atol=1e-3)
|
||||
|
||||
|
||||
def test_empty_lines_and_zero_lengths(renderer):
|
||||
lines = [
|
||||
np.empty((0, 3), dtype=np.float32),
|
||||
np.array([[10, 1, 0], [20, 2, 0]], dtype=np.float32),
|
||||
np.empty((0, 3), dtype=np.float32),
|
||||
]
|
||||
widths = [0.1, 0.2, 0.1]
|
||||
polys = renderer._map_lines_to_polygons(lines, widths, 0.0, 10, 100.0)
|
||||
|
||||
assert len(polys) == 3
|
||||
assert polys[0].shape == (0, 2)
|
||||
assert polys[1].shape[0] > 0
|
||||
assert polys[2].shape == (0, 2)
|
||||
|
||||
|
||||
def test_lead_vehicle_clipping_parity(renderer):
|
||||
lead_mock = SimpleNamespace(dRel=30.0, yRel=0.0, status=True)
|
||||
renderer._get_active_leads = lambda: [("ego", lead_mock)]
|
||||
|
||||
rng = np.random.default_rng(123)
|
||||
lines = [
|
||||
np.column_stack((np.linspace(0, 80, 50), rng.normal(0, 1, 50), np.zeros(50))).astype(np.float32)
|
||||
for _ in range(4)
|
||||
]
|
||||
widths = [0.1] * 4
|
||||
|
||||
batched_polys = renderer._map_lines_to_polygons(lines, widths, 0.0, 45, 80.0, clip_by_lead=True)
|
||||
unclipped_polys = renderer._map_lines_to_polygons(lines, widths, 0.0, 45, 80.0, clip_by_lead=False)
|
||||
individual_polys = [
|
||||
renderer._map_line_to_polygon(l, w, 0.0, 45, 80.0, clip_by_lead=True)
|
||||
for l, w in zip(lines, widths, strict=True)
|
||||
]
|
||||
|
||||
# Parity: batched output must match individual output
|
||||
for b_poly, i_poly in zip(batched_polys, individual_polys, strict=True):
|
||||
assert b_poly.shape == i_poly.shape
|
||||
np.testing.assert_allclose(b_poly, i_poly, rtol=1e-5, atol=1e-3)
|
||||
|
||||
# Functional guarantee: lines MUST be truncated before the lead vehicle at 30.0m - 1.5m
|
||||
for b_poly, u_poly in zip(batched_polys, unclipped_polys, strict=True):
|
||||
assert b_poly.shape[0] < u_poly.shape[0]
|
||||
|
||||
|
||||
def test_allow_invert_hill_geometry(renderer):
|
||||
x = np.linspace(5, 100, 40, dtype=np.float32)
|
||||
y = np.zeros(40, dtype=np.float32)
|
||||
z = np.sin(np.linspace(0, np.pi, 40)).astype(np.float32) * 5.0
|
||||
line = np.column_stack((x, y, z))
|
||||
|
||||
poly_no_invert = renderer._map_lines_to_polygons([line], [0.2], 0.0, 39, 100.0, allow_invert=False)[0]
|
||||
poly_invert = renderer._map_lines_to_polygons([line], [0.2], 0.0, 39, 100.0, allow_invert=True)[0]
|
||||
individual_no_invert = renderer._map_line_to_polygon(line, 0.2, 0.0, 39, 100.0, allow_invert=False)
|
||||
|
||||
# Parity check
|
||||
assert poly_no_invert.shape == individual_no_invert.shape
|
||||
np.testing.assert_allclose(poly_no_invert, individual_no_invert, rtol=1e-5, atol=1e-3)
|
||||
|
||||
# Functional guarantee: allow_invert=False MUST discard downward reverse-slope points
|
||||
assert poly_no_invert.shape[0] < poly_invert.shape[0]
|
||||
|
||||
|
||||
def test_clipping_region_bounds_parity(renderer):
|
||||
renderer._clip_region = SimpleNamespace(x=100, y=100, width=500, height=500)
|
||||
rng = np.random.default_rng(999)
|
||||
lines = [
|
||||
np.column_stack((np.linspace(1, 150, 60), rng.uniform(-10, 10, 60), np.zeros(60))).astype(np.float32)
|
||||
for _ in range(3)
|
||||
]
|
||||
widths = [0.1, 0.2, 0.3]
|
||||
|
||||
batched = renderer._map_lines_to_polygons(lines, widths, 0.0, 55, 150.0)
|
||||
individual = [renderer._map_line_to_polygon(l, w, 0.0, 55, 150.0) for l, w in zip(lines, widths, strict=True)]
|
||||
|
||||
for b, i in zip(batched, individual, strict=True):
|
||||
assert b.shape == i.shape
|
||||
np.testing.assert_allclose(b, i, rtol=1e-5, atol=1e-3)
|
||||
@@ -23,7 +23,7 @@ from openpilot.system.ui.lib.multilang import multilang
|
||||
from openpilot.common.realtime import Ratekeeper
|
||||
|
||||
DEVICE_TYPE = HARDWARE.get_device_type()
|
||||
_DEFAULT_FPS = int(os.getenv("FPS", {'tizi': 20}.get(DEVICE_TYPE, 60)))
|
||||
_DEFAULT_FPS = int(os.getenv("FPS", "60"))
|
||||
FPS_LOG_INTERVAL = 5 # Seconds between logging FPS drops
|
||||
FPS_DROP_THRESHOLD = 0.9 # FPS drop threshold for triggering a warning
|
||||
FPS_CRITICAL_THRESHOLD = 0.5 # Critical threshold for triggering strict actions
|
||||
@@ -39,6 +39,7 @@ BIG_UI = os.getenv("BIG", "0") == "1"
|
||||
MACOS = platform.system() == "Darwin"
|
||||
ENABLE_VSYNC = os.getenv("ENABLE_VSYNC", "0") == "1"
|
||||
MICI_FORCE_RENDER_TEXTURE = os.getenv("MICI_FORCE_RENDER_TEXTURE", "0") == "1"
|
||||
TICI_FORCE_RENDER_TEXTURE = os.getenv("TICI_FORCE_RENDER_TEXTURE", "0") == "1"
|
||||
BURN_IN_PREVENTION = os.getenv("BURN_IN_PREVENTION", "0" if PC else "1") == "1"
|
||||
BURN_IN_SHIFT_INTERVAL = max(1.0, float(os.getenv("BURN_IN_SHIFT_INTERVAL", "180")))
|
||||
BURN_IN_SHIFT_PIXELS = max(0, int(os.getenv("BURN_IN_SHIFT_PIXELS", "2")))
|
||||
@@ -619,6 +620,16 @@ class GuiApplication:
|
||||
def request_close(self):
|
||||
self._window_close_requested = True
|
||||
|
||||
def _needs_render_texture(self) -> bool:
|
||||
return bool(
|
||||
(self._scale != 1.0 and not PC)
|
||||
or BURN_IN_MODE
|
||||
or RECORD
|
||||
or MICI_FORCE_RENDER_TEXTURE
|
||||
or TICI_FORCE_RENDER_TEXTURE
|
||||
or WHITE_LUMINANCE_CAP < 1.0
|
||||
)
|
||||
|
||||
def init_window(self, title: str, fps: int = _DEFAULT_FPS):
|
||||
with self._startup_profile_context():
|
||||
def _request_close(sig, frame):
|
||||
@@ -639,12 +650,7 @@ class GuiApplication:
|
||||
self._render_texture_width = max(1, int(round(self._scaled_width * self._pixel_scale_x)))
|
||||
self._render_texture_height = max(1, int(round(self._scaled_height * self._pixel_scale_y)))
|
||||
|
||||
# Keep big-UI burn-in movement in final-frame composition. Translating the live EGL
|
||||
# camera/widget pass can corrupt the camera presentation instead of shifting the UI.
|
||||
needs_render_texture = ((self._scale != 1.0 and not PC) or BURN_IN_MODE or RECORD or
|
||||
MICI_FORCE_RENDER_TEXTURE or
|
||||
(BURN_IN_PREVENTION and DEVICE_TYPE != "mici") or
|
||||
WHITE_LUMINANCE_CAP < 1.0)
|
||||
needs_render_texture = self._needs_render_texture()
|
||||
if PC and self._scale != 1.0:
|
||||
rl.set_mouse_scale(1 / self._scale, 1 / self._scale)
|
||||
if PC:
|
||||
@@ -652,6 +658,8 @@ class GuiApplication:
|
||||
if needs_render_texture:
|
||||
if MICI_FORCE_RENDER_TEXTURE:
|
||||
cloudlog.warning("Forcing render texture path for mici UI")
|
||||
elif TICI_FORCE_RENDER_TEXTURE:
|
||||
cloudlog.warning("Forcing render texture path for tici UI")
|
||||
self._render_texture = rl.load_render_texture(self._render_texture_width, self._render_texture_height)
|
||||
rl.set_texture_filter(self._render_texture.texture, rl.TextureFilter.TEXTURE_FILTER_BILINEAR)
|
||||
|
||||
@@ -1017,6 +1025,9 @@ class GuiApplication:
|
||||
rl.unload_shader(self._white_luminance_shader)
|
||||
self._white_luminance_shader = None
|
||||
|
||||
if hasattr(rl, "_orig_begin_scissor_mode"):
|
||||
rl.begin_scissor_mode = rl._orig_begin_scissor_mode
|
||||
|
||||
self._mouse.stop()
|
||||
|
||||
self.close_ffmpeg()
|
||||
@@ -1244,18 +1255,18 @@ class GuiApplication:
|
||||
if not hasattr(rl, "_orig_begin_scissor_mode"):
|
||||
rl._orig_begin_scissor_mode = rl.begin_scissor_mode
|
||||
|
||||
scale_x = self._scale * (self._pixel_scale_x if self._render_texture else 1.0)
|
||||
scale_y = self._scale * (self._pixel_scale_y if self._render_texture else 1.0)
|
||||
if scale_x == 1.0 and scale_y == 1.0:
|
||||
rl.begin_scissor_mode = rl._orig_begin_scissor_mode
|
||||
return
|
||||
|
||||
def _begin_scissor_mode_scaled(x, y, width, height):
|
||||
def _begin_scissor_mode(x, y, width, height):
|
||||
scale_x = self._scale * (self._pixel_scale_x if self._render_texture else 1.0)
|
||||
scale_y = self._scale * (self._pixel_scale_y if self._render_texture else 1.0)
|
||||
shift_x, shift_y = self._burn_in_shift() if self._render_texture is None else (0.0, 0.0)
|
||||
return rl._orig_begin_scissor_mode(
|
||||
int(x * scale_x), int(y * scale_y),
|
||||
int(math.ceil(width * scale_x)), int(math.ceil(height * scale_y)))
|
||||
int(round((x + shift_x) * scale_x)),
|
||||
int(round((y + shift_y) * scale_y)),
|
||||
int(math.ceil(width * scale_x)),
|
||||
int(math.ceil(height * scale_y)),
|
||||
)
|
||||
|
||||
rl.begin_scissor_mode = _begin_scissor_mode_scaled
|
||||
rl.begin_scissor_mode = _begin_scissor_mode
|
||||
|
||||
def _set_log_callback(self):
|
||||
ffi_libc = cffi.FFI()
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import os
|
||||
from importlib.resources import as_file
|
||||
from types import SimpleNamespace
|
||||
|
||||
@@ -6,6 +7,34 @@ import pytest
|
||||
from openpilot.system.ui.lib import application
|
||||
|
||||
|
||||
def test_default_fps_uses_shared_target_or_fps_override():
|
||||
assert application._DEFAULT_FPS == int(os.getenv("FPS", "60"))
|
||||
|
||||
|
||||
def test_big_ui_adaptive_fps_uses_60_active_15_idle(monkeypatch):
|
||||
monkeypatch.setattr(application, "OFFSCREEN", False)
|
||||
monkeypatch.setattr(application, "RECORD", False)
|
||||
now = [100.0]
|
||||
monkeypatch.setattr(application.time, "monotonic", lambda: now[0])
|
||||
|
||||
app = object.__new__(application.GuiApplication)
|
||||
app._full_target_fps = 60
|
||||
app._target_fps = 60
|
||||
applied_targets = []
|
||||
app._set_target_fps = applied_targets.append
|
||||
|
||||
app.configure_adaptive_rendering(True)
|
||||
assert app._idle_target_fps == 15
|
||||
assert applied_targets[-1] == 60
|
||||
|
||||
now[0] += application.UI_INTERACTION_FPS_DURATION + 0.01
|
||||
app._apply_render_mode()
|
||||
assert applied_targets[-1] == 15
|
||||
|
||||
app.set_render_mode(True)
|
||||
assert applied_targets[-1] == 60
|
||||
|
||||
|
||||
def test_raylib_target_fps_uses_mici_display_refresh(monkeypatch):
|
||||
monkeypatch.setattr(application, "OFFSCREEN", False)
|
||||
monkeypatch.setattr(application, "DEVICE_TYPE", "mici")
|
||||
@@ -134,3 +163,58 @@ def test_brand_font_is_not_replaced_by_language_fallback(monkeypatch):
|
||||
|
||||
assert application.font_fallback(brand_font) is brand_font
|
||||
assert application.font_fallback(SimpleNamespace(texture=SimpleNamespace(id=3))) is unifont
|
||||
|
||||
|
||||
def test_scissor_mode_shifted_for_direct_framebuffer(monkeypatch):
|
||||
orig_scissor_calls = []
|
||||
monkeypatch.setattr(application.rl, "begin_scissor_mode", lambda x, y, w, h: orig_scissor_calls.append((x, y, w, h)))
|
||||
if hasattr(application.rl, "_orig_begin_scissor_mode"):
|
||||
delattr(application.rl, "_orig_begin_scissor_mode")
|
||||
|
||||
app = object.__new__(application.GuiApplication)
|
||||
app._scale = 1.0
|
||||
app._pixel_scale_x = 1.0
|
||||
app._pixel_scale_y = 1.0
|
||||
app._render_texture = None
|
||||
app._burn_in_shift = lambda: (2.0, -1.0)
|
||||
|
||||
app._patch_scissor_mode()
|
||||
application.rl.begin_scissor_mode(100, 200, 300, 400)
|
||||
|
||||
assert orig_scissor_calls[-1] == (102, 199, 300, 400)
|
||||
|
||||
# Inside an offscreen render texture, scissor must remain unshifted
|
||||
app._render_texture = SimpleNamespace()
|
||||
application.rl.begin_scissor_mode(100, 200, 300, 400)
|
||||
assert orig_scissor_calls[-1] == (100, 200, 300, 400)
|
||||
|
||||
# Clean up patched function
|
||||
if hasattr(application.rl, "_orig_begin_scissor_mode"):
|
||||
application.rl.begin_scissor_mode = application.rl._orig_begin_scissor_mode
|
||||
|
||||
|
||||
def test_needs_render_texture_bypassed_on_tici_by_default(monkeypatch):
|
||||
monkeypatch.setattr(application, "PC", False)
|
||||
monkeypatch.setattr(application, "DEVICE_TYPE", "tici")
|
||||
monkeypatch.setattr(application, "BURN_IN_MODE", False)
|
||||
monkeypatch.setattr(application, "RECORD", False)
|
||||
monkeypatch.setattr(application, "MICI_FORCE_RENDER_TEXTURE", False)
|
||||
monkeypatch.setattr(application, "TICI_FORCE_RENDER_TEXTURE", False)
|
||||
monkeypatch.setattr(application, "WHITE_LUMINANCE_CAP", 1.0)
|
||||
|
||||
app = object.__new__(application.GuiApplication)
|
||||
app._scale = 1.0
|
||||
|
||||
# On TICI by default, render texture MUST be False (saving 21.8 ms)
|
||||
assert app._needs_render_texture() is False
|
||||
|
||||
# Explicit override forces render texture
|
||||
monkeypatch.setattr(application, "TICI_FORCE_RENDER_TEXTURE", True)
|
||||
assert app._needs_render_texture() is True
|
||||
|
||||
# Recording mode forces render texture
|
||||
monkeypatch.setattr(application, "TICI_FORCE_RENDER_TEXTURE", False)
|
||||
monkeypatch.setattr(application, "RECORD", True)
|
||||
assert app._needs_render_texture() is True
|
||||
|
||||
|
||||
|
||||
@@ -7,7 +7,7 @@ from enum import IntEnum
|
||||
|
||||
import pyray as rl
|
||||
|
||||
from openpilot.system.hardware import PC
|
||||
from openpilot.system.hardware import HARDWARE, PC
|
||||
from openpilot.system.ui.lib.application import gui_app, FontWeight, FONT_SCALE
|
||||
from openpilot.system.ui.widgets import Widget
|
||||
from openpilot.system.ui.widgets.button import Button, ButtonStyle
|
||||
@@ -113,7 +113,10 @@ def main():
|
||||
if sys.argv[1] == '--recover':
|
||||
mode = ResetMode.RECOVER
|
||||
|
||||
gui_app.init_window("System Reset", 20)
|
||||
if HARDWARE.get_device_type() in ("tici", "tizi"):
|
||||
gui_app.init_window("System Reset")
|
||||
else:
|
||||
gui_app.init_window("System Reset", 20)
|
||||
reset = Reset(mode)
|
||||
|
||||
gui_app.push_widget(reset)
|
||||
|
||||
@@ -408,7 +408,10 @@ class Setup(Widget):
|
||||
|
||||
def main():
|
||||
try:
|
||||
gui_app.init_window("Setup", 20)
|
||||
if HARDWARE.get_device_type() in ("tici", "tizi"):
|
||||
gui_app.init_window("Setup")
|
||||
else:
|
||||
gui_app.init_window("Setup", 20)
|
||||
setup = Setup()
|
||||
gui_app.push_widget(setup)
|
||||
for _ in gui_app.render():
|
||||
|
||||
Reference in New Issue
Block a user