Restore recorded turn icon after replay signal override

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)

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'}

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)