Visual Polish + Suppress "LAST" in UI

This commit is contained in:
firestarsdog
2026-11-08 03:09:44 -05:00
parent 54f3897174
commit 4e4d258ad3
12 changed files with 369 additions and 88 deletions
@@ -21,7 +21,7 @@ from openpilot.starpilot.ui.onroad_customization import (CAMERA_WIDGETS, PROFILE
validate_document, widget_size)
from openpilot.starpilot.ui.onroad_widget_style import draw_widget_frame
from openpilot.starpilot.curve_speed.status import CurveObservation
from openpilot.starpilot.ui.onroad_state import ConditionalLead
from openpilot.starpilot.ui.onroad_state import ConditionalLead, speed_limit_bounds
from openpilot.starpilot.ui.onroad_state import AlertSize, ObservationKind, OnroadAlert, OnroadState, SourceReading, SpeedLimitObservation
from openpilot.starpilot.ui.appearance_preferences import CameraViewChoice, OnroadAppearance
from openpilot.starpilot.ui.onroad_dm import draw_monitor_art, monitor_bounds, monitor_visible
@@ -81,7 +81,7 @@ def sample_state(scene: str, document: dict) -> OnroadState:
speed_limit=SpeedLimitObservation(kind=ObservationKind.VALID, source="dashboard",
speed_limit_mps=22.0 if pending else 20.0,
accepted_speed_limit_mps=20.0, accepted_source='dashboard',
accepted_adjusted_limit_mps=22.0, effective_cluster_target_mps=22.0,
accepted_adjusted_limit_mps=22.0, effective_cluster_target_mps=22.0, effective_cap_mps=22.0,
offset_mps=2.0, limiting_max_set=True,
pending_adjusted_limit_mps=24.0 if pending else None,
presentation_adjusted_limit_mps=24.0 if pending else 22.0,
@@ -341,10 +341,13 @@ class _Canvas:
crop_y = max(0, math.floor(position['y'] - widget.get('visualInsetTop', 0)))
offset_x, offset_y = crop_x - position['x'], crop_y - position['y']
width, height = math.ceil(logical_width - offset_x), math.ceil(logical_height - offset_y)
if self.profile == Profile.COMPACT and key == 'speed_limit':
crop_x = max(0, math.floor(position['x'] - 2))
offset_x = crop_x - position['x']
width = math.ceil(logical_width + 2 - offset_x)
if (key == ('speed_limit' if self.profile == Profile.COMPACT else 'cruise_limits') and
state.speed_limit.pending_speed_limit_mps is None and (area := speed_limit_bounds(self.profile, state)) is not None):
# The existing padded touch area includes the raised and SET-limited outlines.
crop_x, crop_y = max(0, math.floor(min(crop_x, area[0]))), max(0, math.floor(min(crop_y, area[1])))
width = math.ceil(max(position['x'] + logical_width, area[2]) - crop_x)
height = math.ceil(max(position['y'] + logical_height, area[3]) - crop_y)
offset_x, offset_y = crop_x - position['x'], crop_y - position['y']
if state.speed_limit.pending_speed_limit_mps is not None and (area := widget.get('confirmation')) is not None:
crop_x = math.floor(min(crop_x, position['x'] + area['x']))
crop_y = math.floor(min(crop_y, position['y'] + area['y']))
@@ -137,7 +137,8 @@ class CompactHudRenderer:
def render_max_speed(self, state: OnroadState) -> None:
shown = resolve_unified_speed(state)
status = set_speed_status(state, Profile.COMPACT, shown)
persistent = shown.slc_target_text is not None or shown.override_text is not None or speed_status(state, Profile.COMPACT, shown) is not None
persistent = (shown.set_limited or shown.slc_target_text is not None or shown.override_text is not None or
speed_status(state, Profile.COMPACT, shown) is not None)
set_speed_alpha = self._set_speed_opacity(state, (state.observed_ns or time.monotonic_ns()), slc_active=persistent)
mx, my = widget_offset(state.customization, "compact", "max_speed")
center = widget_size(state.customization, 'compact', 'max_speed')[0] / 2 + mx
+2 -2
View File
@@ -67,8 +67,8 @@ class SpeedLimitShapes:
self._draw(name, x + dx, y + dy, rl.Color(6, 14, 10, 230), scale=scale)
self._draw(name, x, y, color, scale=scale)
def outline(self, x, y, color):
self._outlined('outline', x, y, color)
def outline(self, x, y, color, *, scale=None):
self._outlined('outline', x, y, color, scale=scale)
def button(self, x, y, color, *, profile, scale, pressed=False):
self._draw(f'button_{profile}' + ('_pressed' if pressed else ''), x, y, color, scale=scale)
+19 -17
View File
@@ -22,7 +22,8 @@ OVERRIDE = rl.Color(214, 188, 240, 255)
REJECT = rl.Color(242, 139, 139, 255)
SOURCE_LABELS = {'dashboard': 'DASH', 'map': 'OSM', 'vision': 'VISION', 'mapbox': 'NAV'}
NUMBER_SIZE, HEADING_SIZE = 70, 26
HEADING_ROW, NUMBER_ROW, SOURCE_ROW = 25, 68, 120
# Ink-centered rows, shared by SET, the portrait sign and the confirmation notice.
HEADING_ROW, NUMBER_ROW, SOURCE_ROW = 16.75, 61, 114.5
def draw_text(fonts, text, role, size, x, y, color=INK):
@@ -58,7 +59,7 @@ def draw_set_speed(fonts, shown, center, top, color):
scale = LARGE_SIGN_SCALE if fonts.profile == Profile.LARGE else 1.
text_scale = scale * Profile.COMPACT.font_scale / fonts.profile.font_scale
draw_speed_row(fonts, 'SET', FontRole.SEMI_BOLD, HEADING_SIZE * text_scale, center,
top + HEADING_ROW * scale, color, 'H')
top + HEADING_ROW * scale, color, 'SET')
if shown.max_text == '–':
draw_line(fonts.profile, rl.Vector2(center - 20 * scale, top + NUMBER_ROW * scale),
rl.Vector2(center + 20 * scale, top + NUMBER_ROW * scale), 5 * scale, color)
@@ -206,7 +207,7 @@ class SpeedLimitWidget:
direction = (-1 if target > state.speed_mps else 1) if target is not None and state.speed_mps is not None and target != state.speed_mps else 0
start = (width - label_width - (19.2 if direction else 0)) / 2
draw_speed_row(self.fonts, source, FontRole.SEMI_BOLD, text_size(label_size),
x + (start + label_width / 2) * scale, content_top + HEADING_ROW * scale, source_color, 'H')
x + (start + label_width / 2) * scale, content_top + HEADING_ROW * scale, source_color, source)
if direction:
self.shapes.arrow('compact', x + (start + label_width + 12.6) * scale, content_top + (HEADING_ROW + .5) * scale,
direction, source_color, scale=1.1 * scale)
@@ -245,9 +246,8 @@ class SpeedLimitWidget:
def _sign(self, state: OnroadState, bounds, now: float) -> None:
shown, obs = resolve_unified_speed(state), state.speed_limit
last = shown.posted_retained or obs.sign_recovering
heading = SOURCE_LABELS.get(shown.source, 'SPEED')
if last:
if shown.show_last:
heading = f'LAST {heading}' if heading != 'SPEED' else 'LAST'
x, top = bounds[0] + 12, bounds[1] + 12
scale = self.scale
@@ -273,7 +273,7 @@ class SpeedLimitWidget:
return origin + left * scale, y + a * scale, origin + right * scale, y + b * scale
vienna_center_y = 67
mutcd_center_y = SIGN_HEIGHT / 2
mutcd_center_y = SIGN_HEIGHT / 2 - 7
def circle_fits(left, right, a, b):
return all((px - 59) ** 2 + (py - vienna_center_y) ** 2 <= 46 ** 2
@@ -281,14 +281,17 @@ class SpeedLimitWidget:
vienna = state.appearance.use_vienna_sign
sign_center_y = vienna_center_y if vienna else mutcd_center_y
active_text = shown.override_text
active_text = shown.override_text or (shown.limit_text if shown.set_limited else None)
if shown.slc_target_text is not None and (not shown.limit_in_use or shown.slc_target_text != shown.posted_text):
active_text = active_text or shown.slc_target_text
pedal = speed_status(state, self.profile, shown) == 'PEDAL'
if (active_text is not None or pedal or shown.limit_in_use and not vienna) and not shown.pending:
# Driver targets are rectangular readouts, never regulatory red-ring signs.
self.shapes.outline(x + center_x * scale, top + mutcd_center_y * scale, IN_USE if shown.limit_in_use else white)
label = 'Override' if shown.override_text is not None else 'Limit'
if shown.set_limited:
self.shapes.outline(x + center_x * scale, top + 70 * scale, white, scale=scale * 13 / 12)
else:
self.shapes.outline(x + center_x * scale, top + mutcd_center_y * scale, IN_USE if shown.limit_in_use else white)
label = 'Override' if shown.override_text is not None else 'LIMIT'
active_text = active_text or shown.slc_target_text or shown.posted_text
target_rows = (('Pedal', FontRole.SEMI_BOLD, 44, 68, OVERRIDE),) if pedal else (
(label, FontRole.SEMI_BOLD, HEADING_SIZE, HEADING_ROW, OVERRIDE if label == 'Override' else white),
@@ -297,16 +300,15 @@ class SpeedLimitWidget:
detail = f'{heading} {posted}' if pedal or shown.override_text is not None or active_text != shown.posted_text else heading
left, right, _, _ = ink(detail, FontRole.MEDIUM, 22)
source_rows = ((detail, FontRole.MEDIUM, 22, SOURCE_ROW, secondary),) if detail == heading or right - left <= 108 else (
(heading, FontRole.SEMI_BOLD, 18, 106, secondary), (posted, FontRole.MEDIUM, 20, 125, secondary))
(heading, FontRole.SEMI_BOLD, 18, 99, secondary), (posted, FontRole.MEDIUM, 20, 118, secondary))
if shown.set_limited:
source_rows = (('SET IS LOW', FontRole.SEMI_BOLD, 20, 103.5, white),
('RAISE SET', FontRole.SEMI_BOLD, 22, 124.5, white))
for value, role, size, cy, color in (*target_rows, *source_rows):
left, right, _, _ = ink(value, role, size)
size *= min(1, 108 / max(1, right - left))
_, _, a, b = ink(value, role, size)
if role == FontRole.SET_SPEED or value in ('Limit', 'Override'):
draw_speed_row(self.fonts, value, role, size * self.text_scale, x + center_x * scale, top + cy * scale,
color, '0123456789' if role == FontRole.SET_SPEED else 'H')
else:
centered(value, role, size, cy - (b - a) / 2, color)
size *= min(1, (118 if shown.set_limited else 108) / max(1, right - left))
draw_speed_row(self.fonts, value, role, size * self.text_scale, x + center_x * scale, top + cy * scale,
color, '0123456789' if role == FontRole.SET_SPEED else value)
return
color = white
@@ -46,7 +46,7 @@ def test_both_sign_styles_keep_their_height_with_or_without_offset(profile, offs
else:
historical_outline.assert_not_called()
widget.shapes.outline.assert_called_once()
assert widget.shapes.outline.call_args.args[:2] == (bounds[0]+12+SIGN_WIDTH/2*scale, bounds[1]+12+SIGN_HEIGHT/2*scale)
assert widget.shapes.outline.call_args.args[:2] == (bounds[0]+12+SIGN_WIDTH/2*scale, bounds[1]+12+(SIGN_HEIGHT/2-7)*scale)
assert widget.shapes.outline.call_args.kwargs == {}
assert ('+5' in [call.args[0] for call in foreground_calls(widget.fonts.draw)]) == offset
+105 -27
View File
@@ -14,7 +14,7 @@ import pytest
from openpilot.starpilot.ui.presentation import BitmapFonts, FontRole, Profile, default_font_directory, font_filename, font_path, validate_bitmap_font
from openpilot.starpilot.ui.onroad_customization import default_document, validate_document
from openpilot.starpilot.ui.large_speed_geometry import LARGE_SIGN_SCALE, SIGN_WIDTH, SIGN_HEIGHT, LIMIT_RECTS, confirmation_bounds
from openpilot.starpilot.ui.large_speed_geometry import LARGE_SIGN_SCALE, SIGN_WIDTH, SIGN_HEIGHT, confirmation_bounds
from openpilot.starpilot.ui.onroad_state import ObservationKind, slc_controls, speed_limit_bounds
from openpilot.starpilot.ui.tests.test_speed_stack import road
from openpilot.starpilot.system.starport.projection_layout import default_layout_for_viewport, validate_layout_for_viewport
@@ -100,8 +100,8 @@ def test_large_set_and_limit_share_typography_at_the_saved_position():
assert abs(posted[3] + (left + right) / 2 - (418 + SIGN_WIDTH/2*LARGE_SIGN_SCALE)) <= .5
a, b = fonts.vertical_ink('130', FontRole.SET_SPEED, posted[2])
assert 60 <= b - a <= 70
assert abs(posted[4]+(a+b)/2-(95+68*LARGE_SIGN_SCALE)) <= .5
assert {'Limit', '120 +10'} <= {call[0] for call in calls}
assert abs(posted[4]+(a+b)/2-(95+61*LARGE_SIGN_SCALE)) <= .5
assert {'LIMIT', '120 +10'} <= {call[0] for call in calls}
set_calls = [call for call in calls if call[1] == FontRole.SET_SPEED and call[3] < 388]
assert bool(set_calls) == (show_set and value is not None)
if set_calls:
@@ -113,14 +113,16 @@ def test_large_set_and_limit_share_typography_at_the_saved_position():
top, bottom = fonts.vertical_ink(text, role, size)
assert (role, size, y) == (posted[1], posted[2], posted[4])
heading = next(call for call in calls if call[0] == 'SET')
limit = next(call for call in calls if call[0] == 'Limit')
assert (heading[1:3], heading[4]) == (limit[1:3], limit[4])
limit = next(call for call in calls if call[0] == 'LIMIT')
assert heading[1:3] == limit[1:3]
assert heading[4] + sum(fonts.vertical_ink(*heading[:3])) / 2 == pytest.approx(
limit[4] + sum(fonts.vertical_ink(*limit[:3])) / 2)
assert not any('mph' in call[0] or 'km/h' in call[0] for call in calls)
if show_set and value is None:
start, end, stroke, color = lines.call_args.args
assert (start.x, start.y, end.x, end.y, stroke) == pytest.approx(
(280 - 20 * LARGE_SIGN_SCALE, 95 + 68 * LARGE_SIGN_SCALE,
280 + 20 * LARGE_SIGN_SCALE, 95 + 68 * LARGE_SIGN_SCALE, 5 * LARGE_SIGN_SCALE))
(280 - 20 * LARGE_SIGN_SCALE, 95 + 61 * LARGE_SIGN_SCALE,
280 + 20 * LARGE_SIGN_SCALE, 95 + 61 * LARGE_SIGN_SCALE, 5 * LARGE_SIGN_SCALE))
assert (color.r, color.g, color.b, color.a) == (255, 255, 255, 255)
assert all(call[0] not in ('–', '?') for call in calls)
else:
@@ -163,10 +165,14 @@ def test_set_and_slc_match_font_and_baseline_on_both_displays(pending):
sign = widget.sign if isinstance(widget, UnifiedSpeedWidget) else widget._speed_limit_widget
sign.render_confirmation(current)
rows = {row[0]: row for row in calls if row[-1].a == 255}
for a, b in (('SET', 'OSM' if pending else 'Limit'), ('140', str(value))):
for a, b in (('SET', 'OSM' if pending else 'LIMIT'), ('140', str(value))):
assert rows[a][1] == rows[b][1]
assert rows[a][2] == pytest.approx(rows[b][2])
assert rows[a][4] == pytest.approx(rows[b][4])
if rows[a][1] == FontRole.SET_SPEED:
assert rows[a][4] == pytest.approx(rows[b][4])
else:
assert rows[a][4] + sum(fonts.vertical_ink(*rows[a][:3])) / 2 == pytest.approx(
rows[b][4] + sum(fonts.vertical_ink(*rows[b][:3])) / 2)
assert not any('mph' in text or 'km/h' in text for text in rows)
finally:
widget.close()
@@ -282,7 +288,7 @@ def test_sign_status_rows_keep_balanced_insets_with_real_fonts():
with patch.object(fonts, 'draw', lambda *args, sink=calls: sink.append(args)):
widget.render(current)
if variant == 'retained':
assert [call[0] for call in calls if call[-1].a == 255] == ['Limit', '100', 'LAST VISION']
assert [call[0] for call in calls if call[-1].a == 255] == ['LIMIT', '100', 'LAST VISION']
bounds = speed_limit_bounds(profile, current)
assert bounds is not None
x, y = bounds[0] + 12, bounds[1] + 12
@@ -296,7 +302,7 @@ def test_sign_status_rows_keep_balanced_insets_with_real_fonts():
box = left+a, top+c, left+b, top+d
# Keep real glyph ink clear of the portrait outline, including both footer rows.
assert x+9.5*scale <= box[0] < box[2] <= x+118.5*scale, (profile, variant, text)
assert y+10.5*scale <= box[1] < box[3] <= y+133.5*scale, (profile, variant, text)
assert y+3.5*scale <= box[1] < box[3] <= y+126.5*scale, (profile, variant, text)
assert (box[0]+box[2])/2 == pytest.approx(x+SIGN_WIDTH/2*scale, abs=.5)
boxes.append(box)
assert len(boxes) >= 2
@@ -307,6 +313,73 @@ def test_sign_status_rows_keep_balanced_insets_with_real_fonts():
rl.close_window()
def test_set_limited_sign_matches_approved_spacing_and_existing_touch_area():
from unittest.mock import patch
from openpilot.starpilot.ui.speed_limit_widget import SpeedLimitWidget, draw_set_speed, IN_USE
from openpilot.starpilot.ui.tests.test_unified_speed_native import state
from openpilot.starpilot.ui.unified_speed_presentation import resolve_unified_speed
owns_context = not rl.is_window_ready()
if owns_context:
rl.init_window(128, 128, 'SET-limited sign margins')
try:
for profile in Profile:
with BitmapFonts(profile, default_font_directory()) as fonts:
widget = SpeedLimitWidget(fonts, profile)
widget.shapes = Mock()
scale = widget.scale
for metric, limited, vienna in product((False, True), (False, True), (False, True)):
if vienna and not limited:
continue # Regulatory Vienna geometry has its own existing checks.
factor = 3.6 if metric else 2.2369362921
target, selected = (120, 100 if limited else 130) if metric else (55, 45 if limited else 70)
current = state(metric=metric, cruise_kph=selected / factor * 3.6)
current = replace(current, appearance=replace(current.appearance, use_vienna_sign=vienna),
speed_limit=replace(current.speed_limit, source='dashboard', speed_limit_mps=target / factor,
accepted_source='dashboard', accepted_speed_limit_mps=target / factor,
accepted_adjusted_limit_mps=target / factor, presentation_adjusted_limit_mps=target / factor,
effective_cluster_target_mps=target / factor, effective_cap_mps=target / factor,
pending_speed_limit_mps=None, limiting_max_set=not limited))
shown = resolve_unified_speed(current)
bounds = speed_limit_bounds(profile, current)
assert bounds is not None
x, y = bounds[0] + 12, bounds[1] + 12
cx = x + SIGN_WIDTH / 2 * scale
outline_scale, cy = (13 / 12, 70) if limited else (1, 65)
calls = []
with patch.object(fonts, 'draw', lambda *args, sink=calls: sink.append(args)):
draw_set_speed(fonts, shown, cx - 200 * scale, y, rl.WHITE)
widget.render(current)
rows = [row for row in calls if row[-1].a == 255]
assert [row[0] for row in rows] == ['SET', str(selected), 'LIMIT', str(target), *(
['SET IS LOW', 'RAISE SET'] if limited else ['DASH'])]
boxes = []
for text, role, size, left, top, _ in rows:
a, b = fonts.horizontal_ink(text, role, size)
c, d = fonts.vertical_ink(text, role, size)
boxes.append((left+a, top+c, left+b, top+d))
assert rows[1][1:3] == rows[3][1:3]
for first, second in ((0, 2), (1, 3)):
assert sum(boxes[first][1::2]) == pytest.approx(sum(boxes[second][1::2]), abs=.01)
half_width, half_height = (SIGN_WIDTH / 2 - 5) * outline_scale * scale, (SIGN_HEIGHT / 2 - 5) * outline_scale * scale
radius = 1.15 if profile == Profile.LARGE else 1.
for left, top, right, bottom in boxes[2:]:
assert (left+right)/2 == pytest.approx(cx, abs=.5)
assert min(left-(cx-half_width), cx+half_width-right,
top-(y+cy*scale-half_height), y+cy*scale+half_height-bottom)-radius >= (8 if limited else 7.5)*scale
assert boxes[-2][3] + 5*scale <= boxes[-1][1]
assert abs((boxes[2][1]-(y+cy*scale-half_height)) - (y+cy*scale+half_height-boxes[-1][3])) <= scale
# The painted stroke plus its contour remains inside the existing tap target.
outer_width, outer_height = (SIGN_WIDTH/2-1)*outline_scale*scale, (SIGN_HEIGHT/2-1)*outline_scale*scale
assert bounds[0] <= cx-outer_width < cx+outer_width <= bounds[2]
assert bounds[1] <= y+cy*scale-outer_height < y+cy*scale+outer_height <= bounds[3]
accent = widget.shapes.outline.call_args.args[2]
assert (accent.r == IN_USE.r) == (not limited)
widget.close()
finally:
if owns_context:
rl.close_window()
def test_galaxy_card_pixels_align_without_reading_font_textures():
from unittest.mock import patch
from openpilot.starpilot.ui.layout_preview_renderer import _Canvas, sample_state
@@ -424,40 +497,45 @@ def test_native_signs_offsets_and_captions_fit_their_touch_targets():
rl.close_window()
def test_compact_preview_sprite_keeps_the_full_confirmation_and_override_sign():
@pytest.mark.parametrize('profile', list(Profile))
def test_preview_sprite_keeps_the_full_confirmation_override_and_set_limited_sign(profile):
import json
from unittest.mock import patch
from openpilot.starpilot.ui.layout_preview_renderer import _Canvas, _straight_rgba, sample_state
owns_context = not rl.is_window_ready()
if owns_context:
rl.init_window(128, 128, 'SLC preview crop fixture')
canvas = _Canvas(Profile.COMPACT)
canvas = _Canvas(profile)
owner = 'speed_limit' if profile == Profile.COMPACT else 'cruise_limits'
positions = (12, 16) if profile == Profile.COMPACT else (75, 95)
width, height = profile.size
try:
for y, scene, vienna in product((12, 16), ('slc_notice', 'slc_pending', 'slc_retained', 'slc_display'), (False, True)):
for y, scene, vienna in product(positions, ('slc_notice', 'slc_pending', 'slc_retained', 'slc_display', 'slc_max'), (False, True)):
document = default_document()
document['layouts']['compact']['speed_limit']['y'] = y
assert validate_document(document)['layouts']['compact']['speed_limit']['y'] == y
document['layouts'][profile][owner]['y'] = y
assert validate_document(document)['layouts'][profile][owner]['y'] == y
state = sample_state(scene, document)
state = replace(state, appearance=replace(state.appearance, use_vienna_sign=vienna))
with patch.dict('openpilot.starpilot.ui.layout_preview_renderer.RUNTIME_PREVIEW_WIDGETS', compact=('speed_limit',)):
png = canvas.render_widget_art(state, document['layouts']['compact'])
with patch.dict('openpilot.starpilot.ui.layout_preview_renderer.RUNTIME_PREVIEW_WIDGETS', {str(profile): (owner,)}):
png = canvas.render_widget_art(state, document['layouts'][profile])
with Image.open(io.BytesIO(png)) as sheet:
slot = json.loads(sheet.info['starpilot-widget-art'])['sprites']['speed_limit']
slot = json.loads(sheet.info['starpilot-widget-art'])['sprites'][owner]
sprite = sheet.crop((slot['x'], slot['y'], slot['x']+slot['width'], slot['y']+slot['height']))
rebuilt = Image.new('RGBA', Profile.COMPACT.size)
rebuilt.paste(sprite, (round(LIMIT_RECTS['compact'][0]+slot['offsetX']), round(y+slot['offsetY'])))
rebuilt = Image.new('RGBA', profile.size)
rebuilt.paste(sprite, (round(document['layouts'][profile][owner]['x']+slot['offsetX']), round(y+slot['offsetY'])))
image = rl.load_image_from_texture(canvas.target.texture)
try:
rl.image_flip_vertical(image)
direct = np.frombuffer(_straight_rgba(bytes(image.data[:536*240*4])), np.uint8).reshape(240, 536, 4)
direct = np.frombuffer(_straight_rgba(bytes(image.data[:width*height*4])), np.uint8).reshape(height, width, 4)
finally:
rl.unload_image(image)
assert np.array_equal(direct[..., 3], np.asarray(rebuilt)[..., 3])
if scene in ('slc_notice', 'slc_pending'):
assert (slot['offsetX'], slot['offsetY'], slot['width'], slot['height']) == (-172, -8, 328, 159)
else:
assert (slot['offsetX'], slot['offsetY'], slot['width'], slot['height']) == (-2, 0, SIGN_WIDTH+4, SIGN_HEIGHT)
assert np.any(direct[..., 3]), (y, scene)
if profile == Profile.COMPACT:
if scene in ('slc_notice', 'slc_pending'):
assert (slot['offsetX'], slot['offsetY'], slot['width'], slot['height']) == (-172, -8, 328, 159)
else:
assert (slot['offsetX'], slot['offsetY'], slot['width'], slot['height']) == (-12, -12, SIGN_WIDTH+24, SIGN_HEIGHT+20)
assert np.any(direct[..., 3]), (profile, y, scene)
finally:
canvas.close()
if owns_context:
@@ -560,7 +560,7 @@ def test_large_draw_origins_and_wheel_input_move_together():
with patch("openpilot.starpilot.ui.onroad_large_widgets.draw_control_card") as card, patch.object(rl, "draw_line_ex"):
UnifiedSpeedWidget(Mock(spec=BitmapFonts, **vars(fonts))).render(content, state)
card.assert_not_called()
assert foreground_calls(fonts.draw)[0].args[3:5] == pytest.approx((275, 117.14), abs=.01)
assert foreground_calls(fonts.draw)[0].args[3:5] == pytest.approx((275, 106.07), abs=.01)
CurrentSpeedHud(Mock(spec=BitmapFonts, **vars(fonts))).render(content, state)
assert foreground_calls(fonts.draw)[-2].args[3:5] == (965, 112)
wheel = SteeringWheelWidget(Path('/unused'))
@@ -593,7 +593,7 @@ def test_frozen_default_draw_geometry_and_neutral_colors():
UnifiedSpeedWidget(Mock(spec=BitmapFonts, **vars(fonts))).render(content, state)
card.assert_not_called()
assert fonts.draw.call_args_list[0].args[0] == 'SET'
assert foreground_calls(fonts.draw)[0].args[3:5] == pytest.approx((175, 97.14), abs=.01)
assert foreground_calls(fonts.draw)[0].args[3:5] == pytest.approx((175, 86.07), abs=.01)
CurrentSpeedHud(Mock(spec=BitmapFonts, **vars(fonts))).render(content, state)
assert foreground_calls(fonts.draw)[-2].args[3:5] == (925, 72)
assert foreground_calls(fonts.draw)[-1].args[3:5] == pytest.approx((925, 215.2))
@@ -619,9 +619,9 @@ def test_compact_us_sign_keeps_camera_visible_at_saved_position(speed):
hud._speed_limit_sign(state)
fill.assert_not_called()
outline.assert_called_once()
assert outline.call_args.args[:2] == (264, 102)
assert outline.call_args.args[:2] == (264, 95)
assert [call.args[0] for call in foreground_calls(fonts.draw)] == ['SPEED', str(speed)]
assert foreground_calls(fonts.draw)[0].args[3:5] == (259, 46)
assert foreground_calls(fonts.draw)[0].args[3:5] == (259, 38)
def test_compact_drag_keeps_animation_state_and_separate_origins():
@@ -637,7 +637,7 @@ def test_compact_drag_keeps_animation_state_and_separate_origins():
with patch.object(hud, "_speed_limit_sign"), patch.object(rl, "draw_circle_gradient"), patch.object(rl, "draw_texture_pro") as wheel:
hud.render(state)
hud.render(state)
assert foreground_calls(fonts.draw)[-1].args[3:5] == (138, 69.5)
assert foreground_calls(fonts.draw)[-1].args[3:5] == (138, 62.5)
rect = wheel.call_args.args[2]
assert (rect.x, rect.y) == (115, 175)
alpha = hud._set_speed_alpha.x
@@ -23,6 +23,31 @@ from openpilot.starpilot.ui.tests.test_unified_speed_native import drawing as dr
from openpilot.starpilot.ui.unified_speed_presentation import resolve_unified_speed
@pytest.mark.parametrize('metric', [False, True])
@pytest.mark.parametrize('offset', [0, 5])
@pytest.mark.parametrize('cluster_delta', [0, 2])
def test_set_limited_tracks_controller_target_without_changing_set(metric, offset, cluster_delta):
fixture = replay.RuntimeReplayTests()
fixture.setUp()
sm, cp = fixture.sm, fixture.cp
factor = 3.6 if metric else 2.2369362921
runtime = controller(offset / factor)
sm['slcDashboardObservation'].speedMps = 55 / factor
sm['carState'].vEgo = (40 - cluster_delta) / factor
sm['carState'].vEgoCluster = 40 / factor
for step, selected in enumerate((45, 55 + offset, 70, 45)):
sm['carState'].vCruise = (selected - cluster_delta) / factor * 3.6
sm['carState'].vCruiseCluster = selected / factor * 3.6
message = tick(runtime, sm, cp, step).message.slcState
current = OnroadState(True, False, 40 / factor, float(sm['carState'].vCruiseCluster),
speed_limit_from_message(message), metric=metric, longitudinal_active=True)
shown = resolve_unified_speed(current)
assert shown.set_limited == (selected == 45)
assert shown.limit_text == str(55 + offset)
assert shown.max_text == str(selected)
assert sm['carState'].vCruise == pytest.approx((selected - cluster_delta) / factor * 3.6)
@pytest.mark.parametrize('mode', [settings.PedalOverride.SPEED_REACHED, settings.PedalOverride.SET_SPEED])
def test_controller_messages_keep_set_stable_through_pedal_release_stop_and_resume(mode):
fixture = replay.RuntimeReplayTests()
@@ -385,7 +385,8 @@ def test_retired_fallback_cannot_reset_session_or_change_set_speed_sign_and_reco
_, view, target = drive.step(1000, status='noCoverage')
shown = resolve_unified_speed(view)
assert target == pytest.approx(45, abs=1e-4)
assert (shown.max_text, shown.posted_text, shown.source, shown.status) == ('50', '45', 'map', 'Unavailable')
assert (shown.max_text, shown.posted_text, shown.source, shown.status) == ('50', '45', 'map', 'Using speed limit')
assert shown.limit_in_use and not shown.show_last
assert not view.speed_limit.driver_override_active
recovered, view, target = drive.step(1100)
assert not view.speed_limit.driver_override_active
@@ -395,7 +396,7 @@ def test_retired_fallback_cannot_reset_session_or_change_set_speed_sign_and_reco
@pytest.mark.parametrize('loss', ['noCoverage', 'noMatch', 'noGps', 'matchedNoLimit', 'missingTile',
'lookahead', 'stopped', 'deadTransport', 'externalGpsStale'])
def test_maps_loss_keeps_latest_display_independent_of_control_fallback(tmp_path, loss):
def test_maps_loss_keeps_accepted_target_display_and_control(tmp_path, loss):
drive = RetentionDrive(tmp_path)
drive.step(0)
changes = {'status': loss}
@@ -415,7 +416,8 @@ def test_maps_loss_keeps_latest_display_independent_of_control_fallback(tmp_path
assert output.message.slcState.acceptedAdjustedLimit == pytest.approx(45 * MPH)
assert view.speed_limit.kind != ObservationKind.VALID
shown = resolve_unified_speed(view)
assert (shown.max_text, shown.posted_text, shown.status) == ('70', '45', 'Unavailable')
assert (shown.max_text, shown.posted_text, shown.status) == ('70', '45', 'Using speed limit')
assert shown.limit_in_use and not shown.show_last
assert all(slc_controls(profile, view) == () for profile in Profile)
assert not output.message.slcState.hasPending
@@ -431,7 +433,7 @@ def test_long_gap_confirmation_and_recovery_keep_the_accepted_limit(tmp_path, an
assert target == pytest.approx(45, rel=0, abs=1e-4)
_, view, target = drive.step(35_000, received=False)
assert target == pytest.approx(45, rel=0, abs=1e-4)
assert resolve_unified_speed(view).status == 'Unavailable'
assert resolve_unified_speed(view).status == 'Using speed limit'
pending, _, target = drive.step(40_000, limit=limit, wayId=124)
assert target == pytest.approx(45, rel=0, abs=1e-4) and pending.message.slcState.hasPending
action = None if answer == 'timeout' else Action(drive.runtime.session_id, 2, pending.message.slcState.decisionId,
@@ -440,7 +442,10 @@ def test_long_gap_confirmation_and_recovery_keep_the_accepted_limit(tmp_path, an
output, view, target = drive.step(300_000, received=False)
assert target == pytest.approx(limit if answer == 'accept' else 45, rel=0, abs=1e-4)
assert not output.message.slcState.hasPending
assert resolve_unified_speed(view).posted_text == str(limit)
shown = resolve_unified_speed(view)
assert shown.posted_text == str(limit if answer == 'accept' else 45)
assert shown.limit_in_use and not shown.show_last
assert view.speed_limit.last_received_speed_limit_mps == pytest.approx(limit * MPH)
# A dead SLC publisher cannot make the widget claim retained control indefinitely.
drive.ui.sm.alive['slcState'] = False
expired = drive.adapter.build(ShellMode.ONROAD, now_ns=NOW + 300_001_000_000).onroad
@@ -482,17 +487,31 @@ def test_returned_map_limit_still_requires_confirmation_when_set_is_unaligned(tm
assert target == pytest.approx(70, rel=0, abs=1e-4)
def test_retention_respects_lower_max_and_reset(tmp_path):
@pytest.mark.parametrize('offset', [0, 5, 10])
def test_retention_respects_lower_max_and_reset(tmp_path, offset):
drive = RetentionDrive(tmp_path / 'max')
params = Params(str(tmp_path / 'max'))
document = od.OffsetDocument(od.IMPERIAL_BOUNDS, (0, 0, 0, offset * MPH, 0, 0, 0))
params.put('SLCOffsetSchedule', od.to_value(document), block=True)
drive.runtime = Runtime(read_params(params))
drive.step(0)
drive.sm['carState'].vCruise = drive.sm['carState'].vCruiseCluster = 40 * MPH * 3.6
_, view, target = drive.step(300, received=False)
_, before, _ = drive.step(50)
_, view, target = drive.step(350, received=False)
assert target == pytest.approx(40, rel=0, abs=1e-4)
shown = resolve_unified_speed(view)
assert view.speed_limit.kind != ObservationKind.VALID and shown.posted_retained
assert shown.status == 'Using set speed'
assert shown.set_limited and not shown.show_last
assert replace(shown, posted_retained=False) == resolve_unified_speed(before)
_, recovered, target = drive.step(400)
assert recovered.speed_limit.sign_recovering
assert resolve_unified_speed(recovered) == resolve_unified_speed(before)
assert target == pytest.approx(40, rel=0, abs=1e-4)
assert resolve_unified_speed(view).status == 'Unavailable'
drive.runtime.reset()
output, _, target = drive.step(400, received=False)
output, _, target = drive.step(450, received=False)
assert output.message.slcState.hasAccepted
assert output.result.ceiling.speed_mps == pytest.approx(45 * MPH)
assert output.result.ceiling.speed_mps == pytest.approx((45 + offset) * MPH)
assert target == pytest.approx(40, rel=0, abs=1e-4)
@@ -89,7 +89,7 @@ def test_sign_keeps_distinct_slc_target_when_set_widget_is_hidden(profile, vienn
effective_cluster_target_mps=45 / 3.6))
font = fonts(profile)
render_sign(SpeedLimitWidget(font, profile), current)
assert [call.args[0] for call in foreground_calls(font.draw)] == ['Limit', '45', 'OSM 50 +5']
assert [call.args[0] for call in foreground_calls(font.draw)] == ['LIMIT', '45', 'OSM 50 +5']
drawing['outline'].assert_called_once()
drawing['circle'].assert_not_called()
@@ -124,12 +124,66 @@ def test_set_and_binding_slc_stay_visible_beside_sign_or_prompt(profile, stage):
assert 'SET' in texts and '35' in texts and not any(text.startswith('SLC ') for text in texts)
assert ('LIMIT 30' in texts) == (stage == 'prompt')
assert ('Now: LIMIT 30' in texts) == (stage == 'expanded')
assert ('Limit' in texts) == (stage in ('live', 'retained'))
assert ('LIMIT' in texts) == (stage in ('live', 'retained'))
if pending:
assert '25' in texts and not any(text.startswith('LAST') for text in texts)
assert ('OSM' in texts) == (stage != 'expanded')
else:
assert ('LAST DASH' if gap else 'DASH') in texts and '30' in texts
assert 'DASH' in texts and '30' in texts and not any(text.startswith('LAST') for text in texts)
@pytest.mark.parametrize('gap', [False, True])
def test_compact_set_stays_visible_while_the_accepted_limit_is_restricted(gap, drawing):
current = state(cruise_kph=45, drive_frame=1)
current = replace(current, speed_limit=replace(current.speed_limit, source='dashboard', speed_limit_mps=55 / 3.6,
accepted_source='dashboard', accepted_speed_limit_mps=55 / 3.6, accepted_adjusted_limit_mps=55 / 3.6,
presentation_adjusted_limit_mps=55 / 3.6, effective_cluster_target_mps=55 / 3.6, effective_cap_mps=55 / 3.6,
pending_speed_limit_mps=None, limiting_max_set=False))
if gap:
current = replace(current, speed_limit=replace(current.speed_limit, kind=ObservationKind.STALE,
source='unknown', speed_limit_mps=None))
font = fonts(Profile.COMPACT)
hud = CompactHudRenderer(font, Mock())
for elapsed in range(0, 5_000_000_000, 100_000_000):
hud.render_max_speed(replace(current, observed_ns=current.observed_ns + elapsed))
assert foreground_calls(font.draw)[-1].args[-1].a >= 250
assert {call.args[0] for call in foreground_calls(font.draw)} == {'SET', '45'}
font.draw.reset_mock()
hud._speed_limit_sign(current)
assert [call.args[0] for call in foreground_calls(font.draw)] == ['LIMIT', '55', 'SET IS LOW', 'RAISE SET']
@pytest.mark.parametrize('profile', list(Profile))
@pytest.mark.parametrize('vienna', [False, True])
@pytest.mark.parametrize('offset', [0, 5, 10])
@pytest.mark.parametrize('selected', ['below', 'equal', 'above'])
def test_accepted_sign_keeps_text_geometry_and_styling_through_source_gaps(profile, vienna, offset, selected, drawing):
current = state(cruise_kph={'below': 45, 'equal': 55 + offset, 'above': 70}[selected])
obs = replace(current.speed_limit, source='dashboard', speed_limit_mps=55 / 3.6,
accepted_source='dashboard', accepted_speed_limit_mps=55 / 3.6, accepted_adjusted_limit_mps=(55 + offset) / 3.6,
presentation_adjusted_limit_mps=(55 + offset) / 3.6, effective_cluster_target_mps=(55 + offset) / 3.6,
effective_cap_mps=(55 + offset) / 3.6, pending_speed_limit_mps=None, limiting_max_set=selected == 'above')
current = replace(current, appearance=replace(current.appearance, use_vienna_sign=vienna))
font = fonts(profile)
sign = SpeedLimitWidget(font, profile)
gap = replace(obs, kind=ObservationKind.STALE, source='unknown', speed_limit_mps=None,
last_received_source='vision', last_received_speed_limit_mps=99 / 3.6,
last_received_adjusted_limit_mps=109 / 3.6)
rendered = []
for observation in (obs, gap, replace(obs, sign_recovering=True)):
font.draw.reset_mock()
for shape in ('outline', 'circle'):
drawing[shape].reset_mock()
render_sign(sign, replace(current, speed_limit=observation))
text = [(*call.args[:-1], rgba_tuple(call.args[-1])) for call in foreground_calls(font.draw)]
shapes = [(shape, *call.args[:2], rgba_tuple(call.args[2]), call.kwargs)
for shape in ('outline', 'circle') for call in drawing[shape].call_args_list]
rendered.append((text, shapes))
assert rendered[0] == rendered[1] == rendered[2]
texts = [row[0] for row in rendered[0][0]]
assert not any(text.startswith('LAST') for text in texts)
assert ('SET IS LOW' in texts) == (selected == 'below')
assert ('RAISE SET' in texts) == (selected == 'below')
@pytest.mark.parametrize('profile', list(Profile))
@@ -407,7 +461,7 @@ def test_compact_set_keeps_dinish_number_and_has_no_backplate(drawing):
assert font_filename(Profile.COMPACT, FontRole.SET_SPEED) == 'DINish-Speed.fnt'
number = foreground_calls(font.draw)[1].args
top, bottom = font.vertical_ink(*number[:3])
assert number[4] + (top + bottom) / 2 == 84
assert number[4] + (top + bottom) / 2 == 77
drawing['draw_circle_gradient'].assert_not_called()
@@ -120,10 +120,10 @@ def test_confirmation_proposal_uses_the_pending_target_and_its_offset(pending, a
@pytest.mark.parametrize('change', ['none', 'retained', 'hold', 'recovering', 'pending', 'source', 'rounded-posted',
'rounded-target', 'pedal', 'override', 'display-only', 'not-limiting', 'disengaged'])
def test_in_use_requires_fresh_accepted_source_and_actual_control(change):
def test_in_use_requires_accepted_target_and_actual_control(change):
current = state()
obs = replace(current.speed_limit, accepted_speed_limit_mps=30., accepted_adjusted_limit_mps=32.,
accepted_source='vision', effective_cluster_target_mps=32., limiting_max_set=True)
accepted_source='vision', effective_cluster_target_mps=32., effective_cap_mps=32., limiting_max_set=True)
changes = {
'retained': {'kind': ObservationKind.STALE, 'speed_limit_mps': None, 'last_received_speed_limit_mps': 30.},
'hold': {'status': 'display_hold'}, 'recovering': {'sign_recovering': True},
@@ -135,7 +135,7 @@ def test_in_use_requires_fresh_accepted_source_and_actual_control(change):
}
current = replace(current, speed_limit=replace(obs, **changes.get(change, {})),
longitudinal_overridden=change == 'pedal', longitudinal_active=change != 'disengaged')
assert resolve_unified_speed(current).limit_in_use == (change == 'none')
assert resolve_unified_speed(current).limit_in_use == (change in ('none', 'retained', 'recovering'))
def test_pedal_basis_cannot_be_presented_as_a_held_override_between_message_updates():
@@ -145,3 +145,84 @@ def test_pedal_basis_cannot_be_presented_as_a_held_override_between_message_upda
assert resolve_unified_speed(current).override_text is None
released = replace(current, speed_limit=replace(current.speed_limit, override_basis='persistent'))
assert resolve_unified_speed(released).override_text == '94'
@pytest.mark.parametrize('metric', [False, True])
@pytest.mark.parametrize('selected,limited', [(45, True), (55, True), (59.6, False), (60, False), (70, False)])
def test_set_limited_uses_the_adjusted_cluster_target_and_visible_units(metric, selected, limited):
factor = 3.6 if metric else 2.2369362921
current = state(metric=metric, cruise_kph=selected / factor * 3.6)
obs = replace(current.speed_limit, source='dashboard', speed_limit_mps=55 / factor,
accepted_speed_limit_mps=55 / factor, accepted_adjusted_limit_mps=60 / factor,
presentation_adjusted_limit_mps=60 / factor, effective_cluster_target_mps=60 / factor,
effective_cap_mps=48 / factor)
shown = resolve_unified_speed(replace(current, speed_limit=obs))
assert shown.set_limited == limited
assert (shown.posted_text, shown.limit_text, shown.active_side) == ('55', '60', 'max')
assert not shown.limit_in_use and shown.slc_target_text is None
@pytest.mark.parametrize('change', ['none', 'retained', 'hold', 'recovering', 'pending', 'automatic', 'source',
'posted', 'target', 'pedal', 'override', 'display-only', 'binding', 'disengaged',
'no-cap', 'no-target', 'no-set'])
def test_set_limited_only_instructs_for_a_qualified_accepted_target(change):
current = state(cruise_kph=60., longitudinal_active=change != 'disengaged', longitudinal_overridden=change == 'pedal')
obs = replace(current.speed_limit, source='dashboard', speed_limit_mps=20.,
presentation_adjusted_limit_mps=22., effective_cluster_target_mps=22., effective_cap_mps=21.)
changes = {
'retained': {'kind': ObservationKind.STALE, 'speed_limit_mps': None, 'last_received_speed_limit_mps': 20.},
'hold': {'status': 'display_hold'}, 'recovering': {'sign_recovering': True},
'pending': {'pending_speed_limit_mps': 25.},
'automatic': {'pending_speed_limit_mps': 25., 'automatic_adoption_pending': True},
'source': {'accepted_source': 'vision'}, 'posted': {'accepted_speed_limit_mps': 20.01},
'target': {'effective_cluster_target_mps': 22.01}, 'override': {'driver_override_active': True},
'display-only': {'action_enabled': False}, 'binding': {'limiting_max_set': True},
'no-cap': {'effective_cap_mps': None}, 'no-target': {'effective_cluster_target_mps': None},
}
current = replace(current, speed_limit=replace(obs, **changes.get(change, {})),
cruise_kph=None if change == 'no-set' else current.cruise_kph)
assert resolve_unified_speed(current).set_limited == (change in ('none', 'retained', 'recovering'))
@pytest.mark.parametrize('metric', [False, True])
@pytest.mark.parametrize('selected', [45, 60, 70])
@pytest.mark.parametrize('kind', [ObservationKind.ABSENT, ObservationKind.UNKNOWN, ObservationKind.STALE])
def test_accepted_target_survives_source_gaps_without_presenting_rejected_history(metric, selected, kind):
factor = 3.6 if metric else 2.2369362921
current = state(metric=metric, cruise_kph=selected / factor * 3.6)
obs = replace(current.speed_limit, source='dashboard', speed_limit_mps=55 / factor,
accepted_speed_limit_mps=55 / factor, accepted_adjusted_limit_mps=60 / factor,
presentation_adjusted_limit_mps=60 / factor, effective_cluster_target_mps=60 / factor,
effective_cap_mps=58 / factor, limiting_max_set=selected > 60)
before = resolve_unified_speed(replace(current, speed_limit=obs))
gap = replace(obs, kind=kind, source='unknown', speed_limit_mps=None,
last_received_speed_limit_mps=99 / factor, last_received_adjusted_limit_mps=109 / factor,
last_received_source='vision')
held = resolve_unified_speed(replace(current, speed_limit=gap))
assert held.posted_retained
assert not held.show_last
assert (held.posted_text, held.limit_text, held.source, held.offset_text) == ('55', '60', 'dashboard', '+5')
assert (held.limit_in_use, held.set_limited, held.active_side, held.status) == (
before.limit_in_use, before.set_limited, before.active_side, before.status)
recovered = resolve_unified_speed(replace(current, speed_limit=replace(obs, sign_recovering=True)))
assert recovered == before
@pytest.mark.parametrize('change', ['no-cap', 'zero-cap', 'no-target', 'different-target', 'no-accepted', 'no-adjusted',
'no-session', 'no-source', 'off', 'pedal', 'override', 'display-only', 'display-hold', 'pending'])
def test_source_history_stays_last_without_a_usable_accepted_controller_target(change):
current = state(cruise_kph=60., longitudinal_active=change != 'off', longitudinal_overridden=change == 'pedal')
obs = replace(current.speed_limit, kind=ObservationKind.STALE, speed_limit_mps=None, source='none',
last_received_speed_limit_mps=20., last_received_source='dashboard',
last_received_adjusted_limit_mps=22., effective_cluster_target_mps=22., effective_cap_mps=21.)
changes = {
'no-cap': {'effective_cap_mps': None}, 'zero-cap': {'effective_cap_mps': 0.},
'no-target': {'effective_cluster_target_mps': None}, 'different-target': {'effective_cluster_target_mps': 23.},
'no-accepted': {'accepted_speed_limit_mps': None}, 'no-adjusted': {'accepted_adjusted_limit_mps': None},
'no-session': {'session_id': None}, 'no-source': {'accepted_source': 'none'},
'override': {'driver_override_active': True}, 'display-only': {'action_enabled': False},
'display-hold': {'status': 'display_hold'}, 'pending': {'pending_speed_limit_mps': 25.},
}
shown = resolve_unified_speed(replace(current, speed_limit=replace(obs, **changes.get(change, {}))))
assert shown.show_last and shown.posted_retained
assert not shown.limit_in_use and not shown.set_limited
@@ -1,4 +1,4 @@
"""The selected set speed and latest received limit, shared by both layouts."""
"""Set speed and source limit, retaining an accepted controller target through source gaps."""
from dataclasses import dataclass
import math
@@ -21,6 +21,8 @@ class UnifiedSpeedPresentation:
slc_target_text: str | None = None
posted_retained: bool = False
limit_in_use: bool = False
set_limited: bool = False
show_last: bool = False
@property
def max_status(self) -> str:
@@ -79,10 +81,23 @@ def resolve_unified_speed(state, *, confirmation=False) -> UnifiedSpeedPresentat
return str(round(value * factor)) if positive(value) else '–'
posted = displayed_limit_mps(obs)
target = displayed_adjusted_limit_mps(obs)
source = displayed_limit_source(obs)
active_target = obs.effective_cluster_target_mps
pending = positive(obs.pending_speed_limit_mps)
slc_limiting = (state.longitudinal_active and not state.longitudinal_overridden and obs.action_enabled and
obs.limiting_max_set and obs.source != 'unknown' and not obs.driver_override_active)
slc_enabled = (state.longitudinal_active and not state.longitudinal_overridden and obs.action_enabled and
not obs.driver_override_active)
posted_retained = positive(posted) and (str(obs.kind) != 'valid' or obs.status == 'display_hold')
show_last = posted_retained or obs.sign_recovering
# A live controller may keep its accepted target after the source stops reporting.
retained_target = bool(show_last and slc_enabled and not pending and obs.status != 'display_hold' and obs.session_id and
obs.accepted_source not in ('', 'none', 'unknown') and positive(obs.effective_cap_mps) and
positive(obs.accepted_speed_limit_mps) and positive(obs.accepted_adjusted_limit_mps) and
positive(active_target) and math.isclose(active_target, obs.accepted_adjusted_limit_mps, abs_tol=.001))
if retained_target:
posted, target, source = obs.accepted_speed_limit_mps, active_target, obs.accepted_source
show_last = False
slc_enabled = slc_enabled and (obs.source != 'unknown' or retained_target)
slc_limiting = slc_enabled and obs.limiting_max_set
if confirmation and pending:
posted = obs.pending_speed_limit_mps
posted_retained = False
@@ -102,7 +117,7 @@ def resolve_unified_speed(state, *, confirmation=False) -> UnifiedSpeedPresentat
obs.action_enabled and obs.presentation_id and obs.presentation_id > 0):
status = 'Confirm speed limit'
elif (getattr(obs, 'status', '') == 'display_hold' or pending and obs.presentation_id == 0 or
retaining_limit(obs)):
retaining_limit(obs) and not retained_target):
status = 'Unavailable'
elif not obs.action_enabled:
status = 'Display only'
@@ -117,20 +132,23 @@ def resolve_unified_speed(state, *, confirmation=False) -> UnifiedSpeedPresentat
elif state.cruise_active:
status = 'Using set speed'
max_text = speed(state.cruise_kph / 3.6) if positive(state.cruise_kph) else '–'
source = displayed_limit_source(obs)
if confirmation and pending and obs.pending_source not in ('', 'none'):
source = obs.pending_source
proposal = None
if pending:
proposal = obs.pending_adjusted_limit_mps or (
target if obs.pending_speed_limit_mps == posted else obs.pending_speed_limit_mps + (obs.offset_mps or 0))
active_target = obs.effective_cluster_target_mps
# A matching displayed number alone does not establish that this source is controlling.
limit_in_use = bool(slc_limiting and not pending and not posted_retained and not obs.sign_recovering and
positive(posted) and positive(target) and positive(active_target) and
positive(obs.accepted_speed_limit_mps) and source == obs.accepted_source and
math.isclose(posted, obs.accepted_speed_limit_mps, abs_tol=.001) and
math.isclose(target, active_target, abs_tol=.001))
accepted_display = bool(not pending and not show_last and
positive(posted) and positive(target) and positive(active_target) and
positive(obs.accepted_speed_limit_mps) and source == obs.accepted_source and
math.isclose(posted, obs.accepted_speed_limit_mps, abs_tol=.001) and
math.isclose(target, active_target, abs_tol=.001))
limit_in_use = slc_limiting and accepted_display
# Compare cluster values in displayed units; equal-looking speeds need no instruction.
set_limited = bool(slc_enabled and accepted_display and not obs.limiting_max_set and
positive(obs.effective_cap_mps) and positive(state.cruise_kph) and
round(active_target * factor) > round(state.cruise_kph / 3.6 * factor))
return UnifiedSpeedPresentation(mode, max_text, speed(posted), offset, 'km/h' if state.metric else 'mph',
source, pending,
('limit' if slc_limiting else 'max') if state.cruise_active else 'none',
@@ -141,4 +159,4 @@ def resolve_unified_speed(state, *, confirmation=False) -> UnifiedSpeedPresentat
obs.override_basis != 'pedal' and
positive(obs.effective_cluster_target_mps) else None,
speed(active_target) if slc_limiting and positive(active_target) else None,
posted_retained, limit_in_use)
posted_retained, limit_in_use, set_limited, show_last)