From c4732e7f6a6416a9babeccb9686f73f57617565f Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Sun, 20 Sep 2026 12:20:03 -0700 Subject: [PATCH] Restore recorded turn icon after replay signal override --- roadscore/prototype/normal_ui_audit.py | 3 +- roadscore/prototype/replay_ui_controls.py | 28 +++++++++++ .../prototype/test_replay_ui_controls.py | 50 ++++++++++++++++++- 3 files changed, 78 insertions(+), 3 deletions(-) diff --git a/roadscore/prototype/normal_ui_audit.py b/roadscore/prototype/normal_ui_audit.py index afc08cc2b5..deaec7d89f 100644 --- a/roadscore/prototype/normal_ui_audit.py +++ b/roadscore/prototype/normal_ui_audit.py @@ -4,7 +4,7 @@ from pathlib import Path from openpilot.selfdrive.ui.ui_state import UIState, device from preparing_awake import PreparationWake from replay_display_hold import ReplayDisplayHold -from replay_ui_controls import ReplayUIControls, ReplayStateView, isolated_replay, apply_turn_intent, replay_turn_alert +from replay_ui_controls import ReplayUIControls, ReplayStateView, isolated_replay, apply_turn_intent, replay_turn_alert, restore_recorded_turn_icon replay_controls=ReplayUIControls(os.environ['ROADSCORE_STATUS_FILE'],enabled=isolated_replay(os.environ)) if replay_controls.enabled: from openpilot.selfdrive.ui.mici.onroad.hud_renderer import TurnIntent @@ -17,6 +17,7 @@ if replay_controls.enabled: global replay_arrow_mode native_alert=original_get_alert(self,sm) mode=sm.signal_mode if isinstance(sm,ReplayStateView) and ui_state.started else 'recorded' + if isinstance(sm,ReplayStateView):restore_recorded_turn_icon(self,mode,sm.recorded_turn_side()) result=replay_turn_alert(self,mode,native_alert,lambda **fields:Alert(size=AlertSize.mid,status=AlertStatus.normal,**fields)) ui_state.roadscore_replay_prompt_active=result is not None and result is getattr(self,'_roadscore_demo_alert',None) native_priority=not ui_state.roadscore_replay_prompt_active and (result is not None or self._prev_alert is not None) diff --git a/roadscore/prototype/replay_ui_controls.py b/roadscore/prototype/replay_ui_controls.py index 60fc7ffaf5..f9e155e32c 100644 --- a/roadscore/prototype/replay_ui_controls.py +++ b/roadscore/prototype/replay_ui_controls.py @@ -91,6 +91,20 @@ def replay_turn_alert(widget, signal_mode, native_alert, alert_factory): return previous_demo +def restore_recorded_turn_icon(widget, signal_mode, recorded_side): + """Keep native laneChange's cached side separate from simulated steer icons.""" + previous = getattr(widget, '_roadscore_icon_mode', 'recorded') + if signal_mode != 'recorded': + if previous == 'recorded': + widget._roadscore_recorded_icon_side = getattr(widget, '_last_icon_side', None) + if recorded_side in ('left', 'right'): + widget._roadscore_recorded_icon_side = recorded_side + elif previous != 'recorded': + widget._last_icon_side = (recorded_side if recorded_side in ('left', 'right') + else getattr(widget, '_roadscore_recorded_icon_side', None)) + widget._roadscore_icon_mode = signal_mode + + class ReplayUIControls: def __init__(self, status_path, *, enabled=False, clock=time.monotonic): self.path = Path(status_path) @@ -188,6 +202,20 @@ class ReplayStateView: state.always_on_lateral_active = self.mode == 'disengaged' state.switchback_mode_enabled = False + def recorded_turn_side(self): + """Read original fresh direction, never the UI's manual blinker copies.""" + sm = self.subscriber + if sm.valid.get('modelV2', False) and sm.alive.get('modelV2', False): + meta = sm['modelV2'].meta + side = str(meta.laneChangeDirection) + if str(meta.laneChangeState) != 'off' and side in ('left', 'right'): + return side + if sm.valid.get('carState', False) and sm.alive.get('carState', False): + car = sm['carState'] + if bool(car.leftBlinker) != bool(car.rightBlinker): + return 'left' if car.leftBlinker else 'right' + return None + def snapshot(self): return {'mode': self.mode, 'signal_mode': self.signal_mode, 'session_id': self.controls.session, 'scope': 'isolated-replay-ui'} diff --git a/roadscore/prototype/test_replay_ui_controls.py b/roadscore/prototype/test_replay_ui_controls.py index 9b29baa45a..3e5e83827a 100644 --- a/roadscore/prototype/test_replay_ui_controls.py +++ b/roadscore/prototype/test_replay_ui_controls.py @@ -7,7 +7,7 @@ from types import SimpleNamespace from cereal import log, car, custom from cue_timing import REFERENCE -from replay_ui_controls import ReplayUIControls, ReplayStateView, isolated_replay, apply_turn_intent, replay_turn_alert +from replay_ui_controls import ReplayUIControls, ReplayStateView, isolated_replay, apply_turn_intent, replay_turn_alert, restore_recorded_turn_icon class Subscriber: @@ -17,6 +17,7 @@ class Subscriber: 'starpilotSelfdriveState': custom.StarPilotSelfdriveState.new_message(), 'starpilotCarState': custom.StarPilotCarState.new_message(alwaysOnLateralEnabled=False, pauseLateral=True), 'carState': car.CarState.new_message(leftBlinker=False, rightBlinker=True, vEgo=12), + 'modelV2': log.ModelDataV2.new_message(), } self.updated = {key: False for key in self.builders} self.valid = {key: True for key in self.builders} @@ -66,6 +67,31 @@ class Tests(unittest.TestCase): self.sm.builders['selfdriveState'].enabled=False self.state('recorded','recorded');self.view.update() self.assertFalse(self.view['selfdriveState'].enabled);self.assertTrue(self.view['carState'].rightBlinker) + def test_recorded_icon_direction_uses_original_fresh_model_and_blinkers(self): + self.state('recorded','left');self.view.update() + self.assertTrue(self.view['carState'].leftBlinker) + self.assertEqual(self.view.recorded_turn_side(),'right') + model=self.sm.builders['modelV2'].meta + model.laneChangeState='laneChangeStarting';model.laneChangeDirection='left' + self.assertEqual(self.view.recorded_turn_side(),'left') + self.sm.alive['modelV2']=False + self.assertEqual(self.view.recorded_turn_side(),'right') + self.sm.builders['carState'].leftBlinker=True + self.assertIsNone(self.view.recorded_turn_side()) # Both blinkers are ambiguous. + self.sm.builders['carState'].leftBlinker=False;self.sm.valid['carState']=False + self.assertIsNone(self.view.recorded_turn_side()) + def test_recorded_icon_cache_survives_manual_off_and_unknown_direction(self): + widget=SimpleNamespace(_last_icon_side='right') + restore_recorded_turn_icon(widget,'left','right') + widget._last_icon_side='left' # Native renderer draws the simulated prompt. + restore_recorded_turn_icon(widget,'off',None) + self.assertEqual(widget._last_icon_side,'left') # Do not alter its normal fade. + restore_recorded_turn_icon(widget,'recorded',None) + self.assertEqual(widget._last_icon_side,'right') + widget=SimpleNamespace(_last_icon_side=None) + restore_recorded_turn_icon(widget,'left',None);widget._last_icon_side='left' + restore_recorded_turn_icon(widget,'recorded',None) + self.assertIsNone(widget._last_icon_side) # Never promote a manual side to recorded. def test_no_override_without_fresh_same_replay_session(self): self.state();self.view.update() for patch in ({'command_wall':97},{'command_wall':101},{'command_wall':float('nan')}, @@ -258,7 +284,8 @@ class Tests(unittest.TestCase): root=Path(__file__).resolve().parents[2] ui=SimpleNamespace(started=True) rectangle=lambda *values:SimpleNamespace(x=values[0],y=values[1],width=values[2],height=values[3]) - namespace=dict(ReplayStateView=ReplayStateView,replay_turn_alert=replay_turn_alert,ui_state=ui, + namespace=dict(ReplayStateView=ReplayStateView,replay_turn_alert=replay_turn_alert, + restore_recorded_turn_icon=restore_recorded_turn_icon,ui_state=ui, original_get_alert=lambda widget,sm:getattr(widget,'native_alert',None), Alert=lambda **fields:SimpleNamespace(**fields),AlertSize=SimpleNamespace(mid=2), AlertStatus=SimpleNamespace(normal=0),replay_arrow_mode='recorded', @@ -284,6 +311,25 @@ class Tests(unittest.TestCase): self.assertIs(icon.texture,getattr(widget,'_txt_turn_signal_'+side)) self.assertTrue(ui.roadscore_replay_prompt_active) self.assertEqual(namespace['replay_arrow_mode'],side) + # Route2 at64.4s: manual Left has populated the native cache, while the + # recording is already changing lanes right with a directionless alert. + self.state('recorded','left');self.view.update() + prompt=namespace['replay_get_alert'](widget,self.view) + namespace['_icon_helper'](widget,prompt) + self.assertEqual(widget._last_icon_side,'left') + model=self.sm.builders['modelV2'].meta + model.laneChangeState='laneChangeStarting';model.laneChangeDirection='right' + self.state('recorded','recorded');self.view.update() + native=SimpleNamespace(text1='Changing Lanes',alert_type='laneChange/warning',status=0) + widget.native_alert=widget._prev_alert=native + restored=namespace['replay_get_alert'](widget,self.view) + self.assertIs(restored,native) + icon=namespace['_icon_helper'](widget,restored).icon + self.assertEqual(icon.side,'right');self.assertIs(icon.texture,widget._txt_turn_signal_right) + self.assertFalse(ui.roadscore_replay_prompt_active) + self.assertEqual(namespace['replay_arrow_mode'],'recorded') + # Important native alerts continue to preempt manual simulated arrows. + self.state('recorded','left');self.view.update() native=SimpleNamespace(text1='Recorded critical alert') widget.native_alert=widget._prev_alert=native self.assertIs(namespace['replay_get_alert'](widget,self.view),native)