Unify Big UI speed limit card

This commit is contained in:
firestarsdog
2026-09-29 02:39:36 -04:00
parent 452dc42868
commit a45ecf73ab
21 changed files with 651 additions and 471 deletions
+2
View File
@@ -237,6 +237,8 @@ struct StarPilotPlan @0xf98d843bfd7004a3 {
cscLearnedLatAccel @40 :Float32; # learned comfort at the current curvature, before margin
cscBindingDistance @41 :Float32; # distance to the horizon point setting the target, m
approachStopLength @42 :Float32; # pre-commit distance to a detected stop, m; 0 when off
slcPresentedSpeedLimitSource @43 :Text; # source of the shown accepted or pending posted limit
slcIsLimitingMaxSet @44 :Bool; # SLC target is below the configured Max Set
}
struct StarPilotRadarState @0xb86e6369214c01c8 {
@@ -8,7 +8,7 @@ from cereal import custom
from openpilot.common.constants import CV
from openpilot.common.realtime import DT_MDL
from openpilot.starpilot.controls.lib.speed_limit_controller import (
SOURCE_DASHBOARD, SOURCE_MAP, SOURCE_MAPBOX, SOURCE_NONE, SOURCE_VISION, SpeedLimitController,
SOURCE_DASHBOARD, SOURCE_MAP, SOURCE_MAPBOX, SOURCE_NONE, SOURCE_PREVIOUS_LIMIT, SOURCE_VISION, SpeedLimitController,
)
from openpilot.starpilot.controls.lib.mapbox_speed_limit import MapboxSpeedLimit
@@ -262,6 +262,14 @@ def test_previous_fallback_startup_has_unknown_source(controller_factory):
assert controller.target == pytest.approx(mph(45))
assert controller.source == SOURCE_NONE
assert controller.last_valid_source == SOURCE_NONE
assert controller.presented_source == SOURCE_PREVIOUS_LIMIT
def test_set_speed_fallback_does_not_present_a_posted_limit(controller_factory):
controller = controller_factory(persisted=mph(45), slc_fallback_set_speed=True)
step(controller, cruise=mph(60))
assert controller.target == pytest.approx(mph(60))
assert controller.presented_source == SOURCE_NONE
@pytest.mark.parametrize("fallback", ["set", "experimental"])
@@ -397,6 +405,7 @@ def test_rejection_and_timeout_do_not_change_history(controller_factory):
step(controller, dashboard=mph(45))
step(controller, dashboard=mph(45), decel=True)
assert controller.denied_limit == pytest.approx(mph(45))
assert controller.presented_source == SOURCE_DASHBOARD
assert controller.last_valid_limit == pytest.approx(mph(55))
assert controller.starpilot_planner.params.writes == writes
step(controller, dashboard=mph(45))
@@ -408,6 +417,20 @@ def test_rejection_and_timeout_do_not_change_history(controller_factory):
assert controller.last_valid_limit == pytest.approx(mph(55))
def test_rejected_limit_does_not_label_set_speed_fallback_as_posted(controller_factory):
controller = controller_factory(
speed_limit_confirmation_lower=True,
slc_fallback_set_speed=True,
)
step(controller, dashboard=mph(55))
step(controller, dashboard=mph(45))
step(controller, dashboard=mph(45), decel=True)
assert controller.presented_source == SOURCE_DASHBOARD
step(controller, cruise=mph(60))
assert controller.target == pytest.approx(mph(60))
assert controller.presented_source == SOURCE_NONE
def test_disabling_confirmation_accepts_a_previously_denied_limit(controller_factory):
controller = controller_factory(speed_limit_confirmation_lower=True)
step(controller, dashboard=mph(55))
@@ -582,6 +605,24 @@ def test_fully_disengaged_auto_accept_takes_precedence_over_decel(controller_fac
assert controller.denied_limit == 0
def test_presented_source_tracks_pending_candidate_and_rejected_accepted_limit(controller_factory):
controller = controller_factory(speed_limit_confirmation_lower=True)
step(controller, dashboard=mph(55))
assert controller.presented_source == SOURCE_DASHBOARD
step(controller, map_limit=mph(45), way=custom.WaySelectionType.current)
assert controller.confirmation_pending
assert controller.source == SOURCE_NONE
assert controller.presented_source == SOURCE_MAP
step(controller, map_limit=mph(45), way=custom.WaySelectionType.current, decel=True)
assert not controller.confirmation_pending
assert controller.presented_source == SOURCE_DASHBOARD
step(controller)
assert controller.presented_source == SOURCE_NONE
def test_explicit_accept_takes_precedence_over_simultaneous_reject(controller_factory):
controller = controller_factory(speed_limit_confirmation_lower=True)
step(controller, dashboard=mph(55))
@@ -194,6 +194,7 @@ def test_active_slc_target_constrains_vcruise_below_csc_minimum(slc_target_mph,
)
assert result == pytest.approx(expected_v_cruise_mph * CV.MPH_TO_MS)
assert vcruise.slc_is_limiting_max_set == (expected_v_cruise_mph < 35.0)
def test_elantra_gets_lead_veto_margin_before_force_stop():
@@ -628,7 +628,7 @@ class StarPilotLongitudinalLayout(_SettingsPage):
set_state=lambda s: self._params.put_bool("SLCMapboxFiller", s),
visible=self._mapbox_available),
SettingRow("ShowSLCOffset", "toggle", tr_noop("Show SLC Offset"),
subtitle="",
subtitle=tr_noop("Compact display only; the unified card always shows nonzero offsets."),
get_state=lambda: self._params.get_bool("ShowSLCOffset"),
set_state=lambda s: self._params.put_bool("ShowSLCOffset", s)),
SettingRow("SpeedLimitSources", "toggle", tr_noop("Show Sources"),
+23 -274
View File
@@ -1,16 +1,13 @@
import math
from typing import Optional
import pyray as rl
from openpilot.common.constants import CV
from openpilot.selfdrive.ui.ui_state import ui_state, UIStatus
from openpilot.selfdrive.ui.onroad.hud_renderer import COLORS
from openpilot.selfdrive.ui.ui_state import ui_state
from openpilot.system.ui.lib.application import gui_app, FontWeight
from openpilot.system.ui.lib.multilang import tr
from openpilot.system.ui.lib.text_measure import measure_text_cached
from openpilot.selfdrive.ui.onroad.starpilot.widget_style import (
CONTROL_BG, CONTROL_BORDER, CONTROL_BORDER_WIDTH, CONTROL_ROUNDNESS, CONTROL_SEGMENTS, SLC_HEIGHT,
draw_control_card, roundness_for,
CONTROL_BORDER, CONTROL_ROUNDNESS, CONTROL_SEGMENTS,
)
from openpilot.selfdrive.ui.onroad.starpilot.source_bubble_layout import (
enabled_source_titles, fit_source_label, source_abbreviated_value_text,
@@ -23,14 +20,6 @@ _WHITE = rl.Color(255, 255, 255, 255)
# ── Constants ─────────────────────────────────────────────────────────
# EU Vienna sign
EU_SIGN_SIZE = 176
EU_SIGN_WIDTH = 176
RED_RING_WIDTH = 20
# Pending sign blink cadence — 1s period, 50% duty cycle.
PENDING_BLINK_MS = 500
# Source display metadata: source name, main label, value key, bubble label, icon.
SOURCE_DEFS = [
("Dashboard", "Dash", "dashboard_sl", "Dashboard", "dashboard"),
@@ -39,16 +28,12 @@ SOURCE_DEFS = [
("Mapbox", "MBOX", "mapbox_sl", "Mapbox", "map"),
("Upcoming", "NEXT", "next_sl", "Next", "next"),
]
_SOURCE_ICON_KEYS = {source: icon for source, _, _, _, icon in SOURCE_DEFS}
# Fonts
FONT_LABEL = 30
FONT_SOURCE = 40 # Set Speed MAX label size.
FONT_SPEED = 90 # Set Speed value size.
FONT_OFFSET = 29 # Compact offset text.
OFFSET_CHIP_SEGMENTS = 8 # Capsule curve segments.
FONT_EU_LARGE = 70
FONT_EU_SMALL = 60
FONT_EU_OFFSET = 40
def source_icon_key(source: str) -> str | None:
"""Use the same source glyph as the detailed source diagnostics."""
return _SOURCE_ICON_KEYS.get(source)
# Vision speed-limit pulse — one-shot purple highlight when the active source
# is "Vision" and the resolved value just changed.
@@ -109,8 +94,6 @@ def _get_slc_state():
return None
speed_conversion = CV.MS_TO_KPH if ui_state.is_metric else CV.MS_TO_MPH
show_offset = params.get_bool("ShowSLCOffset")
dashboard_sl = sm["starpilotCarState"].dashboardSpeedLimit if sm.valid.get("starpilotCarState", False) else 0.0
vision_enabled = params.get_bool("VisionSpeedLimitDetection")
vision_sl = ui_state.params_memory.get_float("VisionSpeedLimit") if vision_enabled else 0.0
@@ -120,41 +103,25 @@ def _get_slc_state():
params.get("MapboxSecretKey", encoding="utf-8")
)
slc_overridden_speed = plan.slcOverriddenSpeed
# Keep the source limit visible when overridden.
speed_limit = plan.slcSpeedLimit
# Resolved limit in m/s (pre-conversion, pre-offset) — feeds the vision pulse
# change detector so the comparison is unit-stable across km/h ↔ mph flips.
resolved_ms = speed_limit
# Add the per-limit offset to the displayed value only when NOT overridden
# AND ShowSLCOffset is off (when the offset toggle is on, it's rendered as
# a separate field below the speed number instead).
if slc_overridden_speed == 0 and not show_offset:
speed_limit += plan.slcSpeedLimitOffset
speed_limit *= speed_conversion
speed_limit_offset = plan.slcSpeedLimitOffset * speed_conversion
offset_str = f"{'+' if speed_limit_offset > 0 else '-'}{abs(int(round(speed_limit_offset)))}" if speed_limit_offset != 0 else "\u2013"
# Update the vision-source pulse once per frame, after resolved_ms is known
# and before any sign colors are computed downstream.
_tick_pulse(plan.slcSpeedLimitSource, resolved_ms)
# The pulse uses the accepted raw limit, so unit changes cannot retrigger it.
_tick_pulse(plan.slcSpeedLimitSource, plan.slcSpeedLimit)
toggles = getattr(ui_state, 'starpilot_toggles', {})
slc_enabled = toggles.get('speed_limit_controller') if 'speed_limit_controller' in toggles else params.get_bool('SpeedLimitController')
return {
'speed_limit': speed_limit,
'speed_limit_str': "\u2013" if speed_limit <= 1 else str(int(round(speed_limit))),
'slc_overridden_speed': slc_overridden_speed,
'accepted_speed_limit_ms': plan.slcSpeedLimit,
# Match the control target's non-negative base before cluster compensation.
'effective_target_ms': max(0.0, plan.slcSpeedLimit + plan.slcSpeedLimitOffset),
'offset_ms': plan.slcSpeedLimitOffset,
'slc_overridden_speed': plan.slcOverriddenSpeed,
'speed_limit_source': plan.slcSpeedLimitSource,
'presented_source': getattr(plan, 'slcPresentedSpeedLimitSource', plan.slcSpeedLimitSource),
'slc_enabled': bool(slc_enabled),
'slc_is_limiting_max_set': bool(getattr(plan, 'slcIsLimitingMaxSet', False)),
'unconfirmed_speed_limit': max(0.0, plan.unconfirmedSlcSpeedLimit * speed_conversion),
'unconfirmed_valid': unconfirmed_valid,
'speed_limit_changed': speed_limit_changed,
'show_offset': show_offset,
'use_vienna': params.get_bool("UseVienna"),
'offset_str': offset_str,
'speed_conversion': speed_conversion,
'speed_unit': " km/h" if ui_state.is_metric else " mph",
'slc_abbreviated_sources': params.get_bool("SLCAbbreviatedSources"),
'slc_active_sources_only': params.get_bool("SLCActiveSourcesOnly"),
'slc_enabled_sources': enabled_source_titles(
@@ -191,203 +158,6 @@ def _get_semi_bold():
return _font_semi_bold
_ACTIVE_SOURCE_LABELS = {title: abbrev.upper() for title, abbrev, *_ in SOURCE_DEFS}
def _active_source_label(state: dict) -> str:
source = state.get("speed_limit_source")
if not source or source == "None":
return tr("LIMIT")
return _ACTIVE_SOURCE_LABELS.get(source, source.upper())
def _source_label_color(alpha: int, is_overridden: bool = False) -> rl.Color:
"""Match Set Speed's MAX label color."""
if is_overridden or ui_state.status in (UIStatus.DISENGAGED, UIStatus.OVERRIDE):
base = COLORS.DISENGAGED
elif ui_state.status == UIStatus.ENGAGED:
base = COLORS.ENGAGED
else:
base = COLORS.GREY
return _speed_limit_pulse_color(base, alpha)
# ── US MUTCD Sign ─────────────────────────────────────────────────────
def _draw_offset_chip(rect: rl.Rectangle, offset_str: str, color: rl.Color) -> None:
"""Draw the optional SLC offset as a compact accent chip."""
font = _get_semi_bold()
text_size = measure_text_cached(font, offset_str, FONT_OFFSET)
chip_w = max(64.0, text_size.x + 24.0)
chip_h = 36.0
chip_rect = rl.Rectangle(
rect.x + (rect.width - chip_w) / 2,
rect.y + rect.height - chip_h - 10,
chip_w,
chip_h,
)
chip_fill = rl.Color(0, 0, 0, min(120, color.a))
roundness = roundness_for(chip_rect, 18)
rl.draw_rectangle_rounded(chip_rect, roundness, OFFSET_CHIP_SEGMENTS, chip_fill)
rl.draw_rectangle_rounded_lines_ex(chip_rect, roundness, OFFSET_CHIP_SEGMENTS, 2, color)
rl.draw_text_ex(
font,
offset_str,
rl.Vector2(chip_rect.x + (chip_w - text_size.x) / 2, chip_rect.y + (chip_h - text_size.y) / 2),
FONT_OFFSET,
0,
color,
)
def _draw_us_sign(x: float, y: float, sign_width: float, sign_height: float,
speed_text: str, offset_str: str,
source_label: str, alpha: int, show_offset: bool, *,
pending: bool = False, is_overridden: bool = False):
"""Draw the NA control card at (x, y).
The card keeps the SLC's label/value hierarchy while sharing the exact
visible frame geometry with Set Speed. Border and text colors continue to
use the existing Vision pulse and pending blink behavior.
"""
# Pending: blink white/red. Active: shared blue-grey.
if pending:
blink_on = int(rl.get_time() * 1000) % 1000 < PENDING_BLINK_MS
base_border = rl.Color(255, 255, 255, alpha) if blink_on else rl.Color(201, 34, 49, alpha)
else:
base_border = rl.Color(CONTROL_BORDER.r, CONTROL_BORDER.g, CONTROL_BORDER.b,
min(alpha, CONTROL_BORDER.a))
# Compose the blink base with the active vision pulse (no-op outside window).
border_color = _speed_limit_pulse_color(base_border, base_border.a)
# White value text reads on the translucent road background.
text_color = _speed_limit_pulse_color(rl.Color(255, 255, 255, 255), alpha)
card_rect = rl.Rectangle(x, y, sign_width, sign_height)
card_fill = rl.Color(CONTROL_BG.r, CONTROL_BG.g, CONTROL_BG.b, min(CONTROL_BG.a, alpha))
draw_control_card(card_rect, fill=card_fill, border=border_color,
border_width=CONTROL_BORDER_WIDTH)
font_bold = _get_bold()
font_semi = _get_semi_bold()
cx = x + sign_width / 2
# Pending layout: "PENDING" + "LIMIT" + speed (no offset shown when pending).
if pending:
pending_size = measure_text_cached(font_semi, tr("PENDING"), FONT_LABEL - 2)
rl.draw_text_ex(font_semi, tr("PENDING"), rl.Vector2(cx - pending_size.x / 2, y + 20), FONT_LABEL - 2, 0, text_color)
limit_size = measure_text_cached(font_semi, tr("LIMIT"), FONT_LABEL)
rl.draw_text_ex(font_semi, tr("LIMIT"), rl.Vector2(cx - limit_size.x / 2, y + 48), FONT_LABEL, 0, text_color)
speed_size = measure_text_cached(font_bold, speed_text, FONT_SPEED - 6)
rl.draw_text_ex(font_bold, speed_text, rl.Vector2(cx - speed_size.x / 2, y + 85), FONT_SPEED - 6, 0, text_color)
elif show_offset:
# Offset ON: source at the top, speed below it, and the offset in a chip.
source_size = measure_text_cached(font_semi, source_label, FONT_SOURCE)
source_color = _source_label_color(alpha, is_overridden=is_overridden)
rl.draw_text_ex(font_semi, source_label, rl.Vector2(cx - source_size.x / 2, y + 8), FONT_SOURCE, 0, source_color)
speed_size = measure_text_cached(font_bold, speed_text, FONT_SPEED)
rl.draw_text_ex(font_bold, speed_text, rl.Vector2(cx - speed_size.x / 2, y + 44), FONT_SPEED, 0, text_color)
_draw_offset_chip(card_rect, offset_str, text_color)
else:
# Offset OFF: match Set Speed typography.
source_size = measure_text_cached(font_semi, source_label, FONT_SOURCE)
source_color = _source_label_color(alpha, is_overridden=is_overridden)
rl.draw_text_ex(font_semi, source_label, rl.Vector2(cx - source_size.x / 2, y + 27), FONT_SOURCE, 0, source_color)
speed_size = measure_text_cached(font_bold, speed_text, FONT_SPEED)
rl.draw_text_ex(font_bold, speed_text, rl.Vector2(cx - speed_size.x / 2, y + 77), FONT_SPEED, 0, text_color)
# ── EU Vienna Sign ────────────────────────────────────────────────────
def _draw_eu_sign(x: float, y: float, speed_text: str, offset_str: str,
source_label: str, text_alpha: int, show_offset: bool, *, pending: bool = False):
"""Draw EU-style (Vienna) speed limit sign at (x, y).
White disk with a pulsable red ring and pulsable black text. The pre-existing
pending-text blink (black <-> red) composes with the vision pulse: outside the
pulse window the blink is unchanged, inside it both colors are eased toward
VISION_SPEED_LIMIT_PULSE_COLOR.
"""
center_x = x + EU_SIGN_SIZE / 2
center_y = y + EU_SIGN_SIZE / 2
radius = EU_SIGN_SIZE / 2
# White disk fill.
rl.draw_circle(int(center_x), int(center_y), radius, rl.Color(255, 255, 255, text_alpha))
# Red ring; eased toward VISION_SPEED_LIMIT_PULSE_COLOR when a Vision-sourced
# limit just changed.
ring_color = _speed_limit_pulse_color(rl.Color(201, 34, 49, 255), text_alpha)
rl.draw_ring(rl.Vector2(center_x, center_y), radius - RED_RING_WIDTH, radius,
0, 360, 64, ring_color)
font_bold = _get_bold()
eu_font = FONT_EU_LARGE if len(speed_text) <= 2 else FONT_EU_SMALL
# EU pending: text blinks black/red, composed with the vision pulse.
if pending:
blink_on = int(rl.get_time() * 1000) % 1000 < PENDING_BLINK_MS
base_text = rl.Color(0, 0, 0, 255) if blink_on else rl.Color(201, 34, 49, 255)
else:
base_text = rl.Color(0, 0, 0, 255)
text_color = _speed_limit_pulse_color(base_text, text_alpha)
# Pending: text centered (no offset display)
if pending:
speed_size = measure_text_cached(font_bold, speed_text, eu_font)
speed_pos = rl.Vector2(center_x - speed_size.x / 2, center_y - speed_size.y / 2)
rl.draw_text_ex(font_bold, speed_text, speed_pos, eu_font, 0, text_color)
elif not show_offset:
font_semi = _get_semi_bold()
source_size = measure_text_cached(font_semi, source_label, FONT_LABEL - 4)
source_pos = rl.Vector2(center_x - source_size.x / 2, y + 16)
rl.draw_text_ex(font_semi, source_label, source_pos, FONT_LABEL - 4, 0, text_color)
speed_size = measure_text_cached(font_bold, speed_text, eu_font)
speed_pos = rl.Vector2(center_x - speed_size.x / 2, center_y - speed_size.y / 2)
rl.draw_text_ex(font_bold, speed_text, speed_pos, eu_font, 0, text_color)
else:
# Offset ON: source at the top, speed below it, offset at the bottom.
font_semi = _get_semi_bold()
source_size = measure_text_cached(font_semi, source_label, FONT_LABEL - 4)
source_pos = rl.Vector2(center_x - source_size.x / 2, y + 16)
rl.draw_text_ex(font_semi, source_label, source_pos, FONT_LABEL - 4, 0, text_color)
speed_size = measure_text_cached(font_bold, speed_text, eu_font)
speed_pos = rl.Vector2(center_x - speed_size.x / 2, center_y - speed_size.y / 2 - 5)
rl.draw_text_ex(font_bold, speed_text, speed_pos, eu_font, 0, text_color)
offset_size = measure_text_cached(font_semi, offset_str, FONT_EU_OFFSET)
offset_pos = rl.Vector2(center_x - offset_size.x / 2, y + 122)
rl.draw_text_ex(font_semi, offset_str, offset_pos, FONT_EU_OFFSET, 0, text_color)
# ── Dispatcher (pending and active sign share the same rect) ─────────
def _draw_sign(state: dict, rect: rl.Rectangle, *, pending: bool = False):
"""Draw either the pending or active sign in the given rect."""
if pending:
# Pending shows the unconfirmed value, full opacity
speed_text = ("\u2013" if state['unconfirmed_speed_limit'] <= 1
else str(int(round(state['unconfirmed_speed_limit']))))
else:
speed_text = state['speed_limit_str']
text_alpha = 255
is_overridden = not pending and state['slc_overridden_speed'] != 0
source_label = _active_source_label(state)
if state['use_vienna']:
_draw_eu_sign(rect.x, rect.y, speed_text, state['offset_str'], source_label, text_alpha,
state['show_offset'], pending=pending)
else:
_draw_us_sign(rect.x, rect.y, rect.width, rect.height, speed_text, state['offset_str'],
source_label, text_alpha, state['show_offset'], pending=pending,
is_overridden=is_overridden)
# ── Sources Bubble (expandable overlay) ────────────────────────────────
# Fixed outer footprint; the content scale adapts to the visible row count.
@@ -417,7 +187,7 @@ _SOURCE_COMPACT_LABELS = {
def _draw_source_icon(icon_key: str, x: float, y: float, size: float, color: rl.Color) -> None:
"""Draw the small, intentionally simple source glyphs used by the panel."""
"""Draw the existing source glyph for both the header and diagnostics."""
cx = x + size / 2
cy = y + size / 2
stroke = max(2.5, size / 12.0)
@@ -480,7 +250,7 @@ def _draw_source_icon(icon_key: str, x: float, y: float, size: float, color: rl.
color,
)
rl.draw_circle_v(pin_center, size * 0.09, _SOURCE_PANEL_BG)
else: # Dashboard / fallback
elif icon_key == "dashboard":
dashboard_scale = 1.22
pivot = rl.Vector2(cx, cy + size * 0.17)
inner_radius = size * 0.27 * dashboard_scale
@@ -523,7 +293,7 @@ def _draw_sources_bubble_empty_state(panel_rect: rl.Rectangle) -> None:
total_h = sum(sz.y for sz in line_sizes) + line_gap * (len(lines) - 1)
curr_y = round(panel_rect.y + (panel_rect.height - total_h) / 2)
for line, sz in zip(lines, line_sizes):
for line, sz in zip(lines, line_sizes, strict=True):
pos_x = round(panel_rect.x + (panel_rect.width - sz.x) / 2)
rl.draw_text_ex(font, line, rl.Vector2(pos_x, curr_y), font_size, 0, _WHITE)
curr_y += round(sz.y + line_gap)
@@ -608,7 +378,7 @@ def _draw_sources_bubble(state: dict, sign_rect: rl.Rectangle):
f"{tr(compact_label)}-{source_abbreviated_value_text(value)}",
"",
content_right - label_left,
lambda text: measure_text_cached(text_font, text, font_size).x,
lambda text, font=text_font: measure_text_cached(font, text, font_size).x,
)
label_size = measure_text_cached(text_font, label_text, font_size)
text_y = round(row_y + (row_h - label_size.y) / 2)
@@ -647,24 +417,3 @@ def _draw_sources_bubble(state: dict, sign_rect: rl.Rectangle):
value_pos = rl.Vector2(round(content_right - value_size.x), text_y)
rl.draw_text_ex(font_semi, label_text, label_pos, font_size, 0, text_color)
rl.draw_text_ex(font_bold, value_text, value_pos, font_size, 0, text_color)
# ── Public API ────────────────────────────────────────────────────────
def render_speed_limit_at(state: dict, rect: rl.Rectangle, expanded: bool = False) -> Optional[rl.Rectangle]:
"""Render the SLC sign and optional source bubble at a layout rect."""
flashing_pending = state['speed_limit_changed'] and state['unconfirmed_valid']
if flashing_pending:
_draw_sign(state, rect, pending=True)
return None
_draw_sign(state, rect, pending=False)
use_vienna = state['use_vienna']
visual_rect = rl.Rectangle(rect.x, rect.y, EU_SIGN_SIZE, EU_SIGN_SIZE) if use_vienna else rect
if expanded:
_draw_sources_bubble(state, visual_rect)
return visual_rect
@@ -8,7 +8,7 @@ from openpilot.selfdrive.ui.onroad.starpilot.torque_bar import TorqueBar
from openpilot.selfdrive.ui.onroad.starpilot.rivian_lateral_mode import rivian_lateral_mode
from openpilot.selfdrive.ui.onroad.starpilot.widget_layout_manager import WidgetLayoutManager
from openpilot.selfdrive.ui.onroad.starpilot.widgets import (
SetSpeedWidget, SpeedLimitWidget, PedalIconsWidget,
UnifiedSpeedWidget, PedalIconsWidget,
AetherGaugeWidget, PersonalityButtonWidget, DriverMonitorWidget,
SteeringWheelWidget, StoppedTimerWidget, ModelSourceWidget
)
@@ -25,7 +25,6 @@ from openpilot.starpilot.common.favorite_slots import (
build_favorite_slot_options,
filter_favorite_slot_options,
favorite_key_is_valid,
is_bool_param,
)
from openpilot.system.ui.lib.application import MousePos, gui_app, FontWeight
@@ -64,8 +63,7 @@ class StarPilotOnroadView(AugmentedRoadView):
self._hud_renderer.draw_exp_button = False
# Initialize layout widgets
self._set_speed_widget = SetSpeedWidget(self._hud_renderer)
self._speed_limit_widget = SpeedLimitWidget()
self._unified_speed_widget = UnifiedSpeedWidget(self._hud_renderer)
self._aethergauge_widget = AetherGaugeWidget(self._hud_renderer)
self._steering_wheel_widget = SteeringWheelWidget(self._hud_renderer._exp_button)
self._pedals_widget = PedalIconsWidget()
@@ -75,8 +73,7 @@ class StarPilotOnroadView(AugmentedRoadView):
self._stopped_timer_widget = StoppedTimerWidget(self.is_in_reverse)
# Register to layout zones
self.layout_manager.register_widget("left", self._set_speed_widget)
self.layout_manager.register_widget("left", self._speed_limit_widget)
self.layout_manager.register_widget("left", self._unified_speed_widget)
self.layout_manager.register_widget("left", self._aethergauge_widget)
self.layout_manager.register_widget("right", self._steering_wheel_widget)
self.layout_manager.register_widget("right", self._pedals_widget)
@@ -85,8 +82,7 @@ class StarPilotOnroadView(AugmentedRoadView):
self.layout_manager.register_widget("bottom", self._driver_monitor_widget)
# Register as child widgets for click propagation
self._child(self._set_speed_widget)
self._child(self._speed_limit_widget)
self._child(self._unified_speed_widget)
self._child(self._aethergauge_widget)
self._child(self._steering_wheel_widget)
self._child(self._pedals_widget)
@@ -133,7 +129,7 @@ class StarPilotOnroadView(AugmentedRoadView):
if self._draw_hud_controls:
dm = self.driver_state_renderer
self.layout_manager.update_layout(self._content_rect, is_rhd=dm.is_rhd if dm else False)
self._render_slc()
self._render_speed_card()
self._render_overlays()
self._render_road_name()
@@ -167,13 +163,11 @@ class StarPilotOnroadView(AugmentedRoadView):
render_background_effects(rect, border_width)
render_overlay(border_rect, border_width)
def _render_slc(self):
def _render_speed_card(self):
if self._full_alert_showing():
return
if self._speed_limit_widget.is_visible:
self._speed_limit_widget.render(self._speed_limit_widget.rect)
if self._set_speed_widget.is_visible:
self._set_speed_widget.render(self._set_speed_widget.rect)
if self._unified_speed_widget.is_visible:
self._unified_speed_widget.render(self._unified_speed_widget.rect)
def _render_overlays(self):
alert_showing, _ = self.alert_renderer.will_render()
@@ -186,7 +180,7 @@ class StarPilotOnroadView(AugmentedRoadView):
self._render_developer_metrics()
self.layout_manager.render_widgets(exclude={"speed_limit", "set_speed"})
self.layout_manager.render_widgets(exclude={"unified_speed"})
self._render_torque_bar()
self._render_bottom_row_widgets()
@@ -0,0 +1,46 @@
"""Displayed Max Set and posted-limit values for the Big UI speed card."""
from dataclasses import dataclass
@dataclass(frozen=True)
class UnifiedSpeedPresentation:
mode: str
max_speed_text: str
posted_speed_text: str
effective_speed_text: str
offset_text: str | None
unit_text: str
source: str
confirmation_pending: bool
active_side: str
def resolve_unified_speed(show_max: bool, cruise_set: bool, max_speed: float,
slc_state: dict | None, is_metric: bool) -> UnifiedSpeedPresentation:
"""Compare the rounded values the driver sees; ignore override speed for layout."""
unit = "km/h" if is_metric else "mph"
max_text = str(round(max_speed)) if cruise_set else "–"
if slc_state is None:
return UnifiedSpeedPresentation("max_only", max_text, "–", "–", None, unit, "None", False, "max" if cruise_set else "none")
conversion = slc_state['speed_conversion']
accepted = slc_state['accepted_speed_limit_ms']
effective = slc_state['effective_target_ms']
candidate = slc_state['unconfirmed_speed_limit']
pending = bool(slc_state['speed_limit_changed'] and slc_state['unconfirmed_valid'])
source = slc_state['presented_source']
has_limit = (source not in ("", "None") and accepted > 1) or pending
if not has_limit:
return UnifiedSpeedPresentation("max_only", max_text, "–", "–", None, unit, "None", False, "max" if cruise_set else "none")
posted_text = str(round(candidate)) if pending else str(round(accepted * conversion))
effective_text = str(round(effective * conversion)) if effective > 0 else "–"
offset_display = round(slc_state['offset_ms'] * conversion)
offset_text = f"{offset_display:+d}" if offset_display else None
merged = bool(show_max and cruise_set and slc_state['slc_enabled'] and not pending and max_text == effective_text)
mode = "split" if pending else "merged" if merged else "split" if show_max else "limit_only"
active_side = "none" if slc_state['slc_overridden_speed'] else "shared" if merged else (
"slc" if slc_state['slc_is_limiting_max_set'] else "max" if (show_max or pending) and cruise_set else "none"
)
return UnifiedSpeedPresentation(mode, max_text, posted_text, effective_text, offset_text, unit, source, pending, active_side)
@@ -30,12 +30,12 @@ class WidgetLayoutManager:
active_widgets = [w for w in self.zones["left"] if w.is_visible]
# Left zone stacks vertically from the top-left offset
# X anchor is the shared left-control center (content x + 146).
center_x = self.content_rect.x + WIDGET_ANCHOR_OFFSET
# Keep wide cards inside the content rect without moving compact widgets.
current_y = self.content_rect.y + 45
for widget in active_widgets:
w, h = widget.get_size()
center_x = self.content_rect.x + max(float(WIDGET_ANCHOR_OFFSET), w / 2 + 30)
widget.set_rect(rl.Rectangle(center_x - w / 2, current_y, w, h))
current_y += h + self.spacing
@@ -1,6 +1,5 @@
from openpilot.selfdrive.ui.onroad.starpilot.widgets.base import LayoutWidget
from openpilot.selfdrive.ui.onroad.starpilot.widgets.set_speed import SetSpeedWidget
from openpilot.selfdrive.ui.onroad.starpilot.widgets.speed_limit import SpeedLimitWidget
from openpilot.selfdrive.ui.onroad.starpilot.widgets.unified_speed import UnifiedSpeedWidget
from openpilot.selfdrive.ui.onroad.starpilot.widgets.pedal_icons import PedalIconsWidget
from openpilot.selfdrive.ui.onroad.starpilot.widgets.aethergauge import AetherGaugeWidget
from openpilot.selfdrive.ui.onroad.starpilot.widgets.personality_button import PersonalityButtonWidget
@@ -11,8 +10,7 @@ from openpilot.selfdrive.ui.onroad.starpilot.widgets.model_source import ModelSo
__all__ = [
"LayoutWidget",
"SetSpeedWidget",
"SpeedLimitWidget",
"UnifiedSpeedWidget",
"PedalIconsWidget",
"AetherGaugeWidget",
"PersonalityButtonWidget",
@@ -1,72 +0,0 @@
import pyray as rl
from openpilot.selfdrive.ui.ui_state import ui_state, UIStatus
from openpilot.system.ui.lib.application import gui_app, FontWeight
from openpilot.system.ui.lib.multilang import tr
from openpilot.system.ui.lib.text_measure import measure_text_cached
from openpilot.selfdrive.ui.onroad.starpilot.widgets.base import LayoutWidget
from openpilot.selfdrive.ui.onroad.hud_renderer import (
UI_CONFIG, FONT_SIZES, COLORS, CRUISE_DISABLED_CHAR
)
from openpilot.selfdrive.ui.onroad.starpilot.widget_style import draw_control_card
class SetSpeedWidget(LayoutWidget):
def __init__(self, hud_renderer):
super().__init__("set_speed", priority=1)
self.hud_renderer = hud_renderer
self._font_semi_bold = gui_app.font(FontWeight.SEMI_BOLD)
self._font_bold = gui_app.font(FontWeight.BOLD)
@property
def is_visible(self) -> bool:
return (
self.hud_renderer.is_cruise_available
and not ui_state.starpilot_toggles.get("hide_max_speed", False)
)
def get_size(self) -> tuple[float, float]:
set_speed_width = (
UI_CONFIG.set_speed_width_metric
if ui_state.is_metric
else UI_CONFIG.set_speed_width_imperial
)
return float(set_speed_width), float(UI_CONFIG.set_speed_height)
def _render(self, rect: rl.Rectangle) -> None:
draw_control_card(rect)
max_color = COLORS.GREY
set_speed_color = COLORS.DARK_GREY
if self.hud_renderer.is_cruise_set:
set_speed_color = COLORS.WHITE
if ui_state.status == UIStatus.ENGAGED:
max_color = COLORS.ENGAGED
elif ui_state.status == UIStatus.DISENGAGED:
max_color = COLORS.DISENGAGED
elif ui_state.status == UIStatus.OVERRIDE:
max_color = COLORS.OVERRIDE
max_text = tr("MAX")
max_text_width = measure_text_cached(self._font_semi_bold, max_text, FONT_SIZES.max_speed).x
rl.draw_text_ex(
self._font_semi_bold,
max_text,
rl.Vector2(rect.x + (rect.width - max_text_width) / 2, rect.y + 27),
FONT_SIZES.max_speed,
0,
max_color,
)
set_speed_text = (
CRUISE_DISABLED_CHAR
if not self.hud_renderer.is_cruise_set
else str(round(self.hud_renderer.set_speed))
)
speed_text_width = measure_text_cached(self._font_bold, set_speed_text, FONT_SIZES.set_speed).x
rl.draw_text_ex(
self._font_bold,
set_speed_text,
rl.Vector2(rect.x + (rect.width - speed_text_width) / 2, rect.y + 77),
FONT_SIZES.set_speed,
0,
set_speed_color,
)
@@ -1,67 +0,0 @@
import pyray as rl
from typing import Optional
from openpilot.common.params import Params
from openpilot.selfdrive.ui.ui_state import ui_state
from openpilot.selfdrive.ui.onroad.starpilot.widgets.base import LayoutWidget
from openpilot.selfdrive.ui.onroad.starpilot.slc_speed_limit import (
_get_slc_state, render_speed_limit_at, EU_SIGN_SIZE,
)
from openpilot.selfdrive.ui.onroad.starpilot.widget_style import CONTROL_WIDTH, SLC_HEIGHT
class SpeedLimitWidget(LayoutWidget):
TOUCH_SLOP = 20
def __init__(self):
super().__init__("speed_limit", priority=2)
self._slc_state: dict | None = None
self._sign_rect: Optional[rl.Rectangle] = None
@property
def _hit_rect(self) -> rl.Rectangle:
rect = self._sign_rect or self.rect
slop = self.TOUCH_SLOP
return rl.Rectangle(
rect.x - slop,
rect.y - slop,
rect.width + 2 * slop,
rect.height + 2 * slop,
)
@property
def is_visible(self) -> bool:
self._slc_state = _get_slc_state()
if self._slc_state is None:
self._sign_rect = None
return False
return True
def get_size(self) -> tuple[float, float]:
if self._slc_state is None:
return 0.0, 0.0
use_vienna = self._slc_state['use_vienna']
w = float(EU_SIGN_SIZE if use_vienna else CONTROL_WIDTH)
h = float(EU_SIGN_SIZE if use_vienna else SLC_HEIGHT)
return w, h
def _render(self, rect: rl.Rectangle) -> None:
if self._slc_state is None:
return
params = ui_state.ui_params
expanded = params.get_bool("SpeedLimitSources")
self._sign_rect = render_speed_limit_at(self._slc_state, rect, expanded)
def _handle_mouse_press(self, mouse_pos) -> None:
state = self._slc_state
if state is None or not rl.check_collision_point_rec(mouse_pos, self._hit_rect):
return
if state['speed_limit_changed'] and state['unconfirmed_valid']:
Params(memory=True).put_bool("SpeedLimitAccepted", True)
return
params = ui_state.ui_params
current = params.get_bool("SpeedLimitSources")
params.put_bool("SpeedLimitSources", not current)
@@ -0,0 +1,236 @@
"""One Big UI card for Max Set and the accepted speed limit."""
from __future__ import annotations
import math
import pyray as rl
from openpilot.common.params import Params
from openpilot.selfdrive.ui.ui_state import ui_state, UIStatus
from openpilot.selfdrive.ui.onroad.hud_renderer import COLORS
from openpilot.selfdrive.ui.onroad.starpilot.slc_speed_limit import (
_draw_source_icon, _draw_sources_bubble, _get_slc_state, _speed_limit_pulse_color, source_icon_key,
)
from openpilot.selfdrive.ui.onroad.starpilot.unified_speed_presentation import (
UnifiedSpeedPresentation, resolve_unified_speed,
)
from openpilot.selfdrive.ui.onroad.starpilot.widget_style import (
CONTROL_BG, CONTROL_BORDER, CONTROL_BORDER_WIDTH, CONTROL_ROUNDNESS,
CONTROL_SEGMENTS, draw_control_card, roundness_for,
)
from openpilot.selfdrive.ui.onroad.starpilot.widgets.base import LayoutWidget
from openpilot.system.ui.lib.application import gui_app, FontWeight
from openpilot.system.ui.lib.multilang import tr
from openpilot.system.ui.lib.text_measure import measure_text_cached
UNIFIED_WIDTH = 520
UNIFIED_HEIGHT = 250
SINGLE_WIDTH = 250
HEADER_ICON_SIZE = 34
HEADER_FONT_SIZE = 28
VALUE_FONT_SIZE = 96
UNIT_FONT_SIZE = 28
OFFSET_FONT_SIZE = 25
CONFIRMATION_COLOR = rl.Color(188, 132, 255, 255)
OFFSET_COLOR = rl.Color(188, 156, 226, 220)
class UnifiedSpeedWidget(LayoutWidget):
TOUCH_SLOP = 20
def __init__(self, hud_renderer):
super().__init__("unified_speed", priority=1)
self.hud_renderer = hud_renderer
self._font_semi_bold = gui_app.font(FontWeight.SEMI_BOLD)
self._font_bold = gui_app.font(FontWeight.BOLD)
self._slc_state: dict | None = None
self._presentation: UnifiedSpeedPresentation | None = None
self._show_max = False
self._snapshot_frame: int | None = None
def _refresh_snapshot(self) -> None:
frame = getattr(ui_state.sm, "frame", None)
if frame is not None and frame == self._snapshot_frame:
return
self._snapshot_frame = frame
self._slc_state = _get_slc_state()
self._show_max = (
self.hud_renderer.is_cruise_available and
not ui_state.starpilot_toggles.get("hide_max_speed", False)
)
self._presentation = resolve_unified_speed(
self._show_max, self.hud_renderer.is_cruise_set, self.hud_renderer.set_speed,
self._slc_state, ui_state.is_metric,
)
@property
def is_visible(self) -> bool:
self._refresh_snapshot()
return self._show_max or self._presentation.mode != "max_only"
def get_size(self) -> tuple[float, float]:
self._refresh_snapshot()
width = UNIFIED_WIDTH if self._presentation.mode in ("split", "merged") else SINGLE_WIDTH
return float(width), float(UNIFIED_HEIGHT)
@property
def _hit_rect(self) -> rl.Rectangle:
rect = self.rect
return rl.Rectangle(
rect.x, rect.y - self.TOUCH_SLOP,
rect.width + self.TOUCH_SLOP, rect.height + 2 * self.TOUCH_SLOP,
)
def _speed_limit_bounds(self, rect: rl.Rectangle) -> rl.Rectangle | None:
mode = self._presentation.mode
if mode in ("split", "merged"):
return rl.Rectangle(rect.x + rect.width / 2, rect.y, rect.width / 2, rect.height)
if mode == "limit_only":
return rect
return None
def _draw_centered_text(self, text: str, bounds: rl.Rectangle, y: float,
font_size: int, color: rl.Color, *, bold: bool = False) -> None:
font = self._font_bold if bold else self._font_semi_bold
text_size = measure_text_cached(font, text, font_size)
text_x = bounds.x + (bounds.width - text_size.x) / 2
rl.draw_text_ex(font, text, rl.Vector2(text_x, y), font_size, 0, color)
def _draw_header(self, bounds: rl.Rectangle, text: str, icon_key: str | None, label_color: rl.Color) -> None:
text = tr(text)
text_size = measure_text_cached(self._font_semi_bold, text, HEADER_FONT_SIZE)
slot_gap = 9
group_width = HEADER_ICON_SIZE + slot_gap + text_size.x
group_x = bounds.x + (bounds.width - group_width) / 2
icon_y = bounds.y + 20
if icon_key:
_draw_source_icon(icon_key, group_x, icon_y, HEADER_ICON_SIZE, rl.WHITE)
rl.draw_text_ex(
self._font_semi_bold, text,
rl.Vector2(group_x + HEADER_ICON_SIZE + slot_gap, icon_y + (HEADER_ICON_SIZE - text_size.y) / 2),
HEADER_FONT_SIZE, 0, label_color,
)
def _draw_offset_pill(self, bounds: rl.Rectangle, text: str, y: float) -> None:
text_size = measure_text_cached(self._font_semi_bold, text, OFFSET_FONT_SIZE)
width = max(56.0, text_size.x + 20.0)
pill = rl.Rectangle(bounds.x + (bounds.width - width) / 2, y, width, 34)
rl.draw_rectangle_rounded(pill, roundness_for(pill, 17), 8, rl.Color(32, 20, 45, 170))
rl.draw_rectangle_rounded_lines_ex(pill, roundness_for(pill, 17), 8, 2, OFFSET_COLOR)
self._draw_centered_text(text, pill, y + (pill.height - text_size.y) / 2, OFFSET_FONT_SIZE, OFFSET_COLOR)
@staticmethod
def _max_header_color(active_side: str, cruise_set: bool) -> rl.Color:
if cruise_set and ui_state.status == UIStatus.ENGAGED and active_side in ("max", "shared"):
return COLORS.ENGAGED
if cruise_set and ui_state.status in (UIStatus.DISENGAGED, UIStatus.OVERRIDE):
return COLORS.DISENGAGED
return COLORS.GREY
@staticmethod
def _limit_header_color(active_side: str, overridden: bool) -> rl.Color:
if overridden or ui_state.status in (UIStatus.DISENGAGED, UIStatus.OVERRIDE):
return COLORS.DISENGAGED
if ui_state.status == UIStatus.ENGAGED and active_side in ("slc", "shared"):
return COLORS.ENGAGED
return COLORS.GREY
def _draw_active_emphasis(self, rect: rl.Rectangle) -> None:
presentation = self._presentation
if ui_state.status != UIStatus.ENGAGED or presentation.active_side == "none":
return
if presentation.mode in ("max_only", "limit_only"):
bounds = rect
elif presentation.active_side == "slc":
bounds = self._speed_limit_bounds(rect)
elif presentation.active_side == "max":
bounds = rl.Rectangle(rect.x, rect.y, rect.width / 2, rect.height)
else:
bounds = rect
rl.draw_line_ex(
rl.Vector2(bounds.x + 18, rect.y + 65),
rl.Vector2(bounds.x + bounds.width - 18, rect.y + 65),
3, COLORS.ENGAGED,
)
def _draw_speed_limit_border(self, rect: rl.Rectangle, right: rl.Rectangle, color: rl.Color) -> None:
# Clip the shared rounded outline so only the Speed Limit side changes.
rl.begin_scissor_mode(int(right.x), int(rect.y), int(right.width + 1), int(rect.height + 1))
try:
rl.draw_rectangle_rounded_lines_ex(rect, CONTROL_ROUNDNESS, CONTROL_SEGMENTS, CONTROL_BORDER_WIDTH, color)
finally:
rl.end_scissor_mode()
if right.x > rect.x:
rl.draw_line_ex(rl.Vector2(right.x, rect.y + 8), rl.Vector2(right.x, rect.y + rect.height - 8), 3, color)
def _render(self, rect: rl.Rectangle) -> None:
presentation = self._presentation
state = self._slc_state
draw_control_card(rect, fill=CONTROL_BG, border=CONTROL_BORDER)
if presentation.mode == "split":
divider_x = rect.x + rect.width / 2
rl.draw_line_ex(
rl.Vector2(divider_x, rect.y + 8), rl.Vector2(divider_x, rect.y + rect.height - 8),
2, CONTROL_BORDER,
)
self._draw_active_emphasis(rect)
max_bounds = rl.Rectangle(rect.x, rect.y, rect.width / 2, rect.height) if presentation.mode in ("split", "merged") else rect
limit_bounds = self._speed_limit_bounds(rect)
if self._show_max or presentation.confirmation_pending:
max_color = COLORS.DARK_GREY if not self.hud_renderer.is_cruise_set else COLORS.WHITE
max_label_color = self._max_header_color(presentation.active_side, self.hud_renderer.is_cruise_set)
self._draw_header(max_bounds, "MAX SET", "dashboard", max_label_color)
if presentation.mode != "merged":
self._draw_centered_text(presentation.max_speed_text, max_bounds, rect.y + 79, VALUE_FONT_SIZE, max_color, bold=True)
self._draw_centered_text(tr(presentation.unit_text), max_bounds, rect.y + 210, UNIT_FONT_SIZE, COLORS.WHITE_TRANSLUCENT)
if limit_bounds is not None:
icon_key = source_icon_key(presentation.source)
overridden = bool(state and state['slc_overridden_speed'])
label_color = self._limit_header_color(presentation.active_side, overridden)
self._draw_header(limit_bounds, "SPEED LIMIT", icon_key, label_color)
if presentation.mode != "merged":
self._draw_centered_text(presentation.posted_speed_text, limit_bounds, rect.y + 79, VALUE_FONT_SIZE, COLORS.WHITE, bold=True)
if presentation.confirmation_pending:
self._draw_centered_text(tr("PENDING"), limit_bounds, rect.y + 175, 25, CONFIRMATION_COLOR)
elif presentation.offset_text is not None:
self._draw_offset_pill(limit_bounds, presentation.offset_text, rect.y + 171)
self._draw_centered_text(tr(presentation.unit_text), limit_bounds, rect.y + 210, UNIT_FONT_SIZE, COLORS.WHITE_TRANSLUCENT)
if presentation.mode == "merged":
self._draw_centered_text(presentation.effective_speed_text, rect, rect.y + 84, VALUE_FONT_SIZE, COLORS.WHITE, bold=True)
self._draw_centered_text(tr(presentation.unit_text), rect, rect.y + 210, UNIT_FONT_SIZE, COLORS.WHITE_TRANSLUCENT)
if presentation.offset_text is not None:
self._draw_offset_pill(limit_bounds, presentation.offset_text, rect.y + 70)
if presentation.confirmation_pending and limit_bounds is not None:
intensity = (1.0 + math.sin(2.0 * math.pi * rl.get_time())) / 2.0
alpha = round(100 + 155 * intensity)
pulse = rl.Color(CONFIRMATION_COLOR.r, CONFIRMATION_COLOR.g, CONFIRMATION_COLOR.b, alpha)
self._draw_speed_limit_border(rect, limit_bounds, pulse)
else:
if limit_bounds is not None and state is not None:
vision_color = _speed_limit_pulse_color(CONTROL_BORDER, CONTROL_BORDER.a)
if (vision_color.r, vision_color.g, vision_color.b) != (CONTROL_BORDER.r, CONTROL_BORDER.g, CONTROL_BORDER.b):
self._draw_speed_limit_border(rect, limit_bounds, vision_color)
if state is not None and ui_state.ui_params.get_bool("SpeedLimitSources"):
_draw_sources_bubble(state, rect)
def _handle_mouse_press(self, mouse_pos) -> None:
right = self._speed_limit_bounds(self.rect)
if right is None and self._slc_state is not None:
# The detailed source panel remains dismissible when no limit is valid.
right = self.rect
if right is None:
return
target = rl.Rectangle(right.x, right.y - self.TOUCH_SLOP,
right.width + self.TOUCH_SLOP, right.height + 2 * self.TOUCH_SLOP)
if not rl.check_collision_point_rec(mouse_pos, target):
return
if self._presentation.confirmation_pending:
Params(memory=True).put_bool("SpeedLimitAccepted", True)
return
params = ui_state.ui_params
params.put_bool("SpeedLimitSources", not params.get_bool("SpeedLimitSources"))
@@ -69,8 +69,7 @@ def _load_starpilot_onroad_view(monkeypatch):
stub_module("openpilot.selfdrive.ui.onroad.starpilot.widget_layout_manager", WidgetLayoutManager=dummy_widget)
stub_module(
"openpilot.selfdrive.ui.onroad.starpilot.widgets",
SetSpeedWidget=dummy_widget,
SpeedLimitWidget=dummy_widget,
UnifiedSpeedWidget=dummy_widget,
PedalIconsWidget=dummy_widget,
AetherGaugeWidget=dummy_widget,
PersonalityButtonWidget=dummy_widget,
+9 -29
View File
@@ -80,39 +80,19 @@ def test_visible_source_rows_honor_active_only_and_source_order():
]
# When no sources have a valid speed reading (> 0), returns empty list (triggers empty state)
assert visible_source_rows(
source_defs, {key: 0.0 for key in values}, "Map Data", ("Map Data",),
source_defs, dict.fromkeys(values, 0.0), "Map Data", ("Map Data",),
) == []
def test_source_label_color_override_and_engagement_states():
from openpilot.selfdrive.ui.onroad.starpilot.slc_speed_limit import _source_label_color
from openpilot.selfdrive.ui.onroad.hud_renderer import COLORS
from openpilot.selfdrive.ui.ui_state import ui_state, UIStatus
def test_header_reuses_diagnostic_source_icons_without_unknown_fallback():
from openpilot.selfdrive.ui.onroad.starpilot.slc_speed_limit import source_icon_key
# Engaged and not overridden -> Active green
ui_state.status = UIStatus.ENGAGED
color = _source_label_color(255, is_overridden=False)
assert (color.r, color.g, color.b, color.a) == (COLORS.ENGAGED.r, COLORS.ENGAGED.g, COLORS.ENGAGED.b, 255)
# Engaged but overridden -> Disengaged/override gray
color_overridden = _source_label_color(255, is_overridden=True)
assert (color_overridden.r, color_overridden.g, color_overridden.b, color_overridden.a) == (
COLORS.DISENGAGED.r, COLORS.DISENGAGED.g, COLORS.DISENGAGED.b, 255
)
# Disengaged -> Disengaged/override gray
ui_state.status = UIStatus.DISENGAGED
color_disengaged = _source_label_color(255, is_overridden=False)
assert (color_disengaged.r, color_disengaged.g, color_disengaged.b, color_disengaged.a) == (
COLORS.DISENGAGED.r, COLORS.DISENGAGED.g, COLORS.DISENGAGED.b, 255
)
# Override UI status -> Disengaged/override gray
ui_state.status = UIStatus.OVERRIDE
color_ui_override = _source_label_color(255, is_overridden=False)
assert (color_ui_override.r, color_ui_override.g, color_ui_override.b, color_ui_override.a) == (
COLORS.OVERRIDE.r, COLORS.OVERRIDE.g, COLORS.OVERRIDE.b, 255
)
assert source_icon_key("Vision") == "camera"
assert source_icon_key("Dashboard") == "dashboard"
assert source_icon_key("Map Data") == "map"
assert source_icon_key("Mapbox") == "map"
assert source_icon_key("None") is None
assert source_icon_key("Unexpected") is None
def test_vision_pulse_ignores_same_limit_source_flapping(monkeypatch):
@@ -0,0 +1,142 @@
import pytest
from openpilot.common.constants import CV
from openpilot.selfdrive.ui.onroad.starpilot.unified_speed_presentation import resolve_unified_speed
def slc_state(posted_mph=65, offset_mph=0, source="Map Data", *, pending_mph=0,
enabled=True, limiting=False, overridden=False, metric=False):
conversion = CV.MS_TO_KPH if metric else CV.MS_TO_MPH
return {
"accepted_speed_limit_ms": posted_mph / conversion,
"effective_target_ms": max(0, posted_mph + offset_mph) / conversion,
"offset_ms": offset_mph / conversion,
"speed_conversion": conversion,
"unconfirmed_speed_limit": pending_mph,
"unconfirmed_valid": pending_mph > 0,
"speed_limit_changed": pending_mph > 0,
"presented_source": source,
"slc_enabled": enabled,
"slc_is_limiting_max_set": limiting,
"slc_overridden_speed": 1.0 if overridden else 0.0,
}
@pytest.mark.parametrize("max_speed,posted,offset,expected_mode", [
(80, 65, 5, "split"),
(70, 70, 0, "merged"),
(70, 65, 5, "merged"),
(65, 70, -5, "merged"),
])
def test_split_and_merge_use_effective_accepted_limit(max_speed, posted, offset, expected_mode):
result = resolve_unified_speed(True, True, max_speed, slc_state(posted, offset), False)
assert result.mode == expected_mode
assert result.posted_speed_text == str(posted)
assert result.effective_speed_text == str(posted + offset)
assert result.offset_text == (f"{offset:+d}" if offset else None)
def test_pending_candidate_forces_split_without_replacing_accepted_target():
state = slc_state(65, 5, source="Vision", pending_mph=75)
result = resolve_unified_speed(True, True, 70, state, False)
assert result.mode == "split"
assert result.confirmation_pending
assert result.posted_speed_text == "75"
assert result.effective_speed_text == "70"
assert result.source == "Vision"
def test_resolution_after_confirmation_uses_same_equality_rule():
state = slc_state(65, 5, pending_mph=75)
assert resolve_unified_speed(True, True, 70, state, False).mode == "split"
state["speed_limit_changed"] = state["unconfirmed_valid"] = False
assert resolve_unified_speed(True, True, 70, state, False).mode == "merged"
assert resolve_unified_speed(True, True, 80, state, False).mode == "split"
def test_source_change_and_override_do_not_change_layout():
state = slc_state(65, 5)
assert resolve_unified_speed(True, True, 70, state, False).mode == "merged"
state["presented_source"] = "Dashboard"
state["slc_overridden_speed"] = 40.0
result = resolve_unified_speed(True, True, 70, state, False)
assert result.mode == "merged"
assert result.source == "Dashboard"
def test_source_target_change_recomputes_layout_independently():
state = slc_state(65, 5, source="Map Data")
assert resolve_unified_speed(True, True, 70, state, False).mode == "merged"
state.update(slc_state(55, 5, source="Vision"))
result = resolve_unified_speed(True, True, 70, state, False)
assert result.mode == "split"
assert result.source == "Vision"
def test_active_side_uses_published_control_semantic():
state = slc_state(65, 5, limiting=True)
assert resolve_unified_speed(True, True, 80, state, False).active_side == "slc"
state["slc_is_limiting_max_set"] = False
assert resolve_unified_speed(True, True, 80, state, False).active_side == "max"
state["slc_overridden_speed"] = 40.0
assert resolve_unified_speed(True, True, 80, state, False).active_side == "none"
def test_display_only_speed_limit_stays_split():
result = resolve_unified_speed(True, True, 70, slc_state(70, enabled=False), False)
assert result.mode == "split"
def test_disabled_confirmation_does_not_force_split():
state = slc_state(65, 5, pending_mph=75)
state["speed_limit_changed"] = False
assert resolve_unified_speed(True, True, 70, state, False).mode == "merged"
def test_missing_limit_never_renders_zero_or_a_stale_source():
result = resolve_unified_speed(True, True, 70, slc_state(0, source="None"), False)
assert result.mode == "max_only"
assert result.posted_speed_text == "–"
assert result.source == "None"
def test_persisted_previous_limit_without_source_remains_visible():
result = resolve_unified_speed(True, True, 70, slc_state(45, source="Previous Limit"), False)
assert result.mode == "split"
assert result.posted_speed_text == "45"
assert result.source == "Previous Limit"
def test_low_limit_with_large_negative_offset_preserves_configured_offset():
result = resolve_unified_speed(True, True, 70, slc_state(5, -99), False)
assert result.mode == "split"
assert result.posted_speed_text == "5"
assert result.effective_speed_text == "–"
assert result.offset_text == "-99"
def test_metric_and_rounding_follow_the_displayed_value():
state = slc_state(65.4, 4.4, metric=True)
result = resolve_unified_speed(True, True, 70, state, True)
assert result.mode == "merged"
assert result.posted_speed_text == "65"
assert result.offset_text == "+4"
assert result.unit_text == "km/h"
def test_invisible_fraction_does_not_keep_card_split():
state = slc_state(65.1, 5.2)
result = resolve_unified_speed(True, True, 70.4, state, False)
assert result.mode == "merged"
def test_hidden_max_still_shows_posted_limit():
result = resolve_unified_speed(False, True, 70, slc_state(65), False)
assert result.mode == "limit_only"
def test_confirmation_forces_split_even_when_max_is_hidden():
result = resolve_unified_speed(False, True, 70, slc_state(65, pending_mph=75), False)
assert result.mode == "split"
assert result.max_speed_text == "70"
assert result.confirmation_pending
@@ -0,0 +1,94 @@
from types import SimpleNamespace
import pyray as rl
from openpilot.selfdrive.ui.onroad.starpilot.unified_speed_presentation import UnifiedSpeedPresentation
from openpilot.selfdrive.ui.onroad.starpilot.widgets import unified_speed
def make_widget(mode="split", pending=False):
widget = object.__new__(unified_speed.UnifiedSpeedWidget)
widget._rect = rl.Rectangle(30, 75, 520, 250)
widget._presentation = UnifiedSpeedPresentation(mode, "70", "65", "70", "+5", "mph", "Map Data", pending, "slc")
widget._show_max = True
widget._slc_state = None
widget.hud_renderer = SimpleNamespace(is_cruise_set=True)
return widget
def test_speed_limit_hit_target_is_right_half_in_both_layouts():
for mode in ("split", "merged"):
right = make_widget(mode)._speed_limit_bounds(rl.Rectangle(30, 75, 520, 250))
assert (right.x, right.width) == (290, 260)
def test_confirmation_touch_only_accepts_on_speed_limit_side(monkeypatch):
widget = make_widget(pending=True)
writes = []
monkeypatch.setattr(unified_speed, "Params", lambda memory: SimpleNamespace(put_bool=lambda key, value: writes.append((key, value))))
widget._handle_mouse_press(rl.Vector2(100, 150))
assert writes == []
widget._handle_mouse_press(rl.Vector2(400, 150))
assert writes == [("SpeedLimitAccepted", True)]
def test_diagnostic_sources_can_be_dismissed_from_max_only_card(monkeypatch):
widget = make_widget("max_only")
widget._slc_state = {}
params = SimpleNamespace(get_bool=lambda _key: True, put_bool=lambda key, value: writes.append((key, value)))
writes = []
monkeypatch.setattr(unified_speed, "ui_state", SimpleNamespace(ui_params=params))
widget._handle_mouse_press(rl.Vector2(100, 150))
assert writes == [("SpeedLimitSources", False)]
def test_right_border_overlay_is_clipped_to_speed_limit_side(monkeypatch):
widget = make_widget()
events = []
monkeypatch.setattr(unified_speed.rl, "begin_scissor_mode", lambda *args: events.append(("begin", args)))
monkeypatch.setattr(unified_speed.rl, "draw_rectangle_rounded_lines_ex", lambda *args: events.append(("outline", args)))
monkeypatch.setattr(unified_speed.rl, "draw_line_ex", lambda *args: events.append(("divider", args)))
monkeypatch.setattr(unified_speed.rl, "end_scissor_mode", lambda: events.append(("end",)))
rect = widget.rect
right = widget._speed_limit_bounds(rect)
widget._draw_speed_limit_border(rect, right, rl.Color(188, 132, 255, 200))
assert events[0] == ("begin", (290, 75, 261, 251))
assert [event[0] for event in events] == ["begin", "outline", "end", "divider"]
def test_split_and_merged_draw_one_card_with_both_headers(monkeypatch):
cards = []
monkeypatch.setattr(unified_speed, "draw_control_card", lambda *args, **kwargs: cards.append(args[0]))
monkeypatch.setattr(unified_speed, "ui_state", SimpleNamespace(status=unified_speed.UIStatus.DISENGAGED,
ui_params=SimpleNamespace(get_bool=lambda _key: False)))
monkeypatch.setattr(unified_speed.rl, "draw_line_ex", lambda *args: None)
for mode in ("split", "merged"):
widget = make_widget(mode)
headers = []
monkeypatch.setattr(widget, "_draw_header", lambda _bounds, text, icon, _color, rows=headers: rows.append((text, icon)))
monkeypatch.setattr(widget, "_draw_centered_text", lambda *args, **kwargs: None)
monkeypatch.setattr(widget, "_draw_offset_pill", lambda *args: None)
monkeypatch.setattr(widget, "_draw_active_emphasis", lambda *args: None)
widget._render(widget.rect)
assert headers == [("MAX SET", "dashboard"), ("SPEED LIMIT", "map")]
assert len(cards) == 2
def test_header_colors_preserve_engaged_disengaged_and_override_semantics(monkeypatch):
widget = make_widget()
colors = unified_speed.COLORS
monkeypatch.setattr(unified_speed, "ui_state", SimpleNamespace(status=unified_speed.UIStatus.ENGAGED))
assert widget._max_header_color("max", True) == colors.ENGAGED
assert widget._max_header_color("slc", True) == colors.GREY
assert widget._limit_header_color("slc", False) == colors.ENGAGED
monkeypatch.setattr(unified_speed, "ui_state", SimpleNamespace(status=unified_speed.UIStatus.DISENGAGED))
assert widget._max_header_color("max", True) == colors.DISENGAGED
assert widget._limit_header_color("slc", False) == colors.DISENGAGED
monkeypatch.setattr(unified_speed, "ui_state", SimpleNamespace(status=unified_speed.UIStatus.OVERRIDE))
assert widget._max_header_color("max", True) == colors.DISENGAGED
assert widget._limit_header_color("slc", False) == colors.DISENGAGED
monkeypatch.setattr(unified_speed, "ui_state", SimpleNamespace(status=unified_speed.UIStatus.ENGAGED))
assert widget._limit_header_color("none", True) == colors.DISENGAGED
@@ -137,6 +137,19 @@ class TestWidgetLayoutManager(unittest.TestCase):
# w3: should stack directly below w1: y = 75 + 100 + 15 = 190
self.assertEqual(w3.rect.y, 190)
def test_wide_unified_card_stays_inside_the_left_edge(self):
card = DummyLayoutWidget("unified_speed", priority=1, width=520, height=250)
gauge = DummyLayoutWidget("aethergauge", priority=3, width=176, height=260)
self.layout_manager.register_widget("left", card)
self.layout_manager.register_widget("left", gauge)
self.layout_manager.update_layout(self.content_rect)
self.assertEqual(card.rect.x, self.content_rect.x + 30)
self.assertEqual(card.rect.y, self.content_rect.y + 45)
self.assertEqual(gauge.rect.x + gauge.rect.width / 2, self.content_rect.x + 146)
self.assertEqual(gauge.rect.y, card.rect.y + card.rect.height + self.layout_manager.spacing)
def test_dynamic_repositioning_on_rect_change(self):
# Register a widget
w1 = DummyLayoutWidget("w1", priority=1, width=100, height=100)
@@ -2370,8 +2370,8 @@
{
"key": "ShowSLCOffset",
"label": "Show Speed Limit Offset",
"description": "Show the current offset from the posted limit on the driving screen.",
"picker_description": "Shows the current offset from the posted limit.",
"description": "Show the current offset on the compact driving display. The unified Max Set / Speed Limit card always shows nonzero offsets.",
"picker_description": "Shows the offset on the compact display; the unified card always shows nonzero offsets.",
"data_type": "bool",
"ui_type": "toggle",
"parent_key": "SpeedLimitController",
@@ -2990,8 +2990,8 @@
{
"key": "UseVienna",
"label": "Use Vienna-Style Speed Signs",
"description": "Show Vienna-style (EU) speed-limit signs instead of MUTCD (US).",
"picker_description": "Uses Vienna-style speed-limit signs.",
"description": "Use Vienna-style (EU) speed-limit signs on the compact driving display. The unified Max Set / Speed Limit card uses its own layout.",
"picker_description": "Uses Vienna-style signs on the compact display.",
"data_type": "bool",
"ui_type": "toggle",
"parent_key": "NavigationUI",
@@ -13,6 +13,7 @@ SOURCE_DASHBOARD = "Dashboard"
SOURCE_MAP = "Map Data"
SOURCE_VISION = "Vision"
SOURCE_MAPBOX = "Mapbox"
SOURCE_PREVIOUS_LIMIT = "Previous Limit"
REAL_SOURCES = (SOURCE_DASHBOARD, SOURCE_MAP, SOURCE_VISION, SOURCE_MAPBOX)
OFFSET_MAP_IMPERIAL = [
@@ -73,6 +74,7 @@ class SpeedLimitController:
self.confirmation_button_consumed = False
self._active_control = False
self._using_experimental_fallback = False
self._using_previous_limit_fallback = False
self._mode = "off"
def shutdown(self):
@@ -90,6 +92,19 @@ class SpeedLimitController:
def unconfirmed_speed_limit(self):
return self.pending_limit
@property
def presented_source(self):
if self.confirmation_pending:
return self.pending_source
if self.source in REAL_SOURCES:
return self.source
if self._using_previous_limit_fallback and self.target >= 1:
return self.last_valid_source if self.last_valid_source in REAL_SOURCES else SOURCE_PREVIOUS_LIMIT
if (self.denied_limit > 0 and self.last_valid_limit > 0 and self.target >= 1 and
abs(self.target - self.last_valid_limit) < SAME_LIMIT_TOLERANCE):
return self.last_valid_source if self.last_valid_source in REAL_SOURCES else SOURCE_PREVIOUS_LIMIT
return SOURCE_NONE
@property
def experimental_mode(self):
return self._active_control and self._using_experimental_fallback
@@ -205,10 +220,12 @@ class SpeedLimitController:
def _apply_fallback(self, v_cruise, enabled):
self._using_experimental_fallback = False
self._using_previous_limit_fallback = False
previous_vision_filtered = self.last_valid_source == SOURCE_VISION and self.low_vision_limit_filtered(self.last_valid_limit)
if self.starpilot_toggles.slc_fallback_previous_speed_limit and self.last_valid_limit > 0 and not previous_vision_filtered:
self.source = self.last_valid_source
self.target = self.last_valid_limit
self._using_previous_limit_fallback = True
elif enabled and self.starpilot_toggles.slc_fallback_set_speed:
self.source = SOURCE_NONE
self.target = v_cruise
@@ -385,6 +402,7 @@ class SpeedLimitController:
self.limit_change_started = False
self.confirmation_button_consumed = False
self._using_experimental_fallback = False
self._using_previous_limit_fallback = False
mode = "display" if display_only else "active" if active else "off"
if mode != self._mode:
self.mapbox.reset()
@@ -208,6 +208,7 @@ class StarPilotVCruise:
self._nav_instruction_state_raw = None
self._nav_instruction_state = {}
self._applied_slc_control_target = 0.0
self.slc_is_limiting_max_set = False
self.csc_controlling_speed = False
self.csc_glow_release_timer = 0.0
self.csc_override = False
@@ -342,6 +343,7 @@ class StarPilotVCruise:
# ===== Main update =====
def update(self, controls_enabled, now, time_validated, v_cruise, v_ego, sm, starpilot_toggles):
self.slc_is_limiting_max_set = False
if not controls_enabled or not getattr(starpilot_toggles, "speed_limit_controller", False):
self._applied_slc_control_target = 0.0
@@ -759,6 +761,8 @@ class StarPilotVCruise:
self.slc.overridden_speed > 0.0,
getattr(self.slc, "source", "None"),
)
# Publish the semantic used by the UI after the lead-drop adjustment.
self.slc_is_limiting_max_set = bool(controls_enabled and 0 < slc_control_target < v_cruise)
self._applied_slc_control_target = slc_control_target if slc_control_target > 0.0 else 0.0
if slc_control_target > 0.0:
targets.append(slc_control_target)
+2
View File
@@ -382,6 +382,8 @@ class StarPilotPlanner:
starpilotPlan.slcSpeedLimit = self.starpilot_vcruise.slc_target
starpilotPlan.slcSpeedLimitOffset = self.starpilot_vcruise.slc_offset
starpilotPlan.slcSpeedLimitSource = self.starpilot_vcruise.slc.source
starpilotPlan.slcPresentedSpeedLimitSource = self.starpilot_vcruise.slc.presented_source
starpilotPlan.slcIsLimitingMaxSet = self.starpilot_vcruise.slc_is_limiting_max_set
starpilotPlan.speedLimitChanged = self.starpilot_vcruise.slc.confirmation_pending
starpilotPlan.unconfirmedSlcSpeedLimit = self.starpilot_vcruise.slc.unconfirmed_speed_limit