Show actual musical road cues and yield score ribbon to native alerts

roadscore/DEMO.md

@@ -43,3 +43,7 @@ The preview renders six full 536 × 240 synthetic canvases with the repository's
 Road gestures take priority over job timing in the detail row; degraded explanations take priority over both. The startup label uses the selected Prism/Aurora profile; stored playback is labeled as such from preparation onward.
 
 The active native overlay is `prototype/overlay.py`; the pure presentation helper is `prototype/overlay_view.py`. `prototype/index.html` and `prototype/native_display.py` are older excerpt-specific bench interfaces, not the normal onroad demo. The historical style selector is not an ACE profile selector. Keep those older entrypoints distinct when presenting the current demo.
+
+The contextual line names reported musical cues: Turn signal / percussion, Curve ahead / build, Curve apex / impact, Navigation turn / accent, and arrival/stop/resume cues. Active cues take priority over queued cues; queued cues are explicitly prefixed Next. A brief curve apex takes priority over ongoing signal percussion. Unknown kinds are labeled Music cue without guessing a road cause. The archived player supplies these fields from the original scheduler timing.
+
+The score ribbon yields completely to native selfdrive/StarPilot alerts and their fade-out. Optional `ROADSCORE_CAPTURE_EVENTS=1` captures the first displayed active cue of each kind into the ignored replay output for UI review; it does not alter status or music.

roadscore/prototype/overlay.py

@@ -7,15 +7,21 @@ def install():
  if os.environ.get('ROADSCORE_OVERLAY')!='1':return
  import pyray as rl
  from openpilot.system.ui.lib.application import gui_app,FontWeight
- original=gui_app.render;last=0.;state={};frames=0;captured=False;capture_ready_since=None
+ original=gui_app.render;last=0.;state={};frames=0;captured=False;capture_ready_since=None;captured_events=set();alert_clear_after=0.
  path=Path(os.environ['ROADSCORE_STATUS_FILE'])
  def draw():
-  nonlocal last,state,frames,captured,capture_ready_since
+  nonlocal last,state,frames,captured,capture_ready_since,alert_clear_after
   now=time.monotonic()
   if now-last>.2:
    try:state=json.loads(path.read_text())
    except (OSError,ValueError):pass
    last=now
+  from openpilot.selfdrive.ui.ui_state import ui_state
+  # The native alert owns the display. Leave room for its existing fade-out too.
+  for service in ('selfdriveState','starpilotSelfdriveState'):
+   size=ui_state.sm[service].alertSize
+   if int(size.raw)>0:alert_clear_after=now+1.
+  if now<alert_clear_after:return
   view=draw_panel(rl,gui_app.font(FontWeight.NORMAL),state,gui_app.width,gui_app.height)
   ready=view['ready']
   frames+=1
@@ -25,8 +31,15 @@ def install():
   if frames%60==1:path.with_name('overlay_status.json').write_text(json.dumps({'frames':frames,'native_gpu_icon':False,'presentation':view,'state':state}))
   from openpilot.selfdrive.ui.ui_state import ui_state
   if ready and ui_state.started and capture_ready_since is None:capture_ready_since=now
-  if capture_ready_since is not None and now-capture_ready_since>=5 and ui_state.started and not captured and os.environ.get("ROADSCORE_OVERLAY_CAPTURE"):
-   captured=True
+  capture_target=None
+  capture_path=os.environ.get("ROADSCORE_OVERLAY_CAPTURE")
+  if capture_path and ui_state.started:
+   if capture_ready_since is not None and now-capture_ready_since>=5 and not captured:
+    captured=True;capture_target=Path(capture_path)
+   kind=view.get('event_kind')
+   if os.environ.get('ROADSCORE_CAPTURE_EVENTS')=='1' and view.get('event_state')=='active' and kind and kind not in captured_events:
+    captured_events.add(kind);capture_target=Path(capture_path).with_name('overlay-event-'+kind+'.png')
+  if capture_target is not None:
    from PIL import Image
    rl.rl_draw_render_batch_active()
    image=rl.load_image_from_texture(gui_app._render_texture.texture) if gui_app._render_texture else rl.load_image_from_screen()
@@ -34,7 +47,7 @@ def install():
     data=bytes(rl.ffi.buffer(image.data,image.width*image.height*4))
     picture=Image.frombytes('RGBA',(image.width,image.height),data)
     if gui_app._render_texture:picture=picture.transpose(Image.Transpose.FLIP_TOP_BOTTOM)
-    picture.save(os.environ['ROADSCORE_OVERLAY_CAPTURE'])
+    picture.save(capture_target)
    finally:rl.unload_image(image)
  def render(*args,**kwargs):
   for should_render in original(*args,**kwargs):

roadscore/prototype/overlay_view.py

@@ -15,6 +15,47 @@ def seconds(value):
   return None
 
 
+# Explicit presentation of scheduler kinds, ordered by brief event salience.
+# These names describe existing gesture-bank actions, never inferred road input.
+GESTURE_LABELS = {
+  'arrival': ('Arrival', 'Arrival cue'),
+  'curve_apex': ('Curve apex', 'Curve apex / impact'),
+  'curve_prepare': ('Curve ahead', 'Curve ahead / build'),
+  'navigation_turn': ('Navigation turn', 'Navigation turn / accent'),
+  'arrival_prepare': ('Arrival ahead', 'Arrival ahead / build'),
+  'lane_change': ('Lane change', 'Lane change / sweep'),
+  'stop': ('Stop', 'Stop cue'),
+  'resume': ('Resume', 'Resume cue'),
+  'turn_signal': ('Turn signal', 'Turn signal / percussion'),
+  'turn_signal_sustain': ('Turn signal', 'Turn signal / percussion'),
+  'turn_signal_off': ('Signal ended', 'Signal ended / release'),
+}
+
+
+def gesture_view(state):
+  active = state.get('gesture_active')
+  active = [kind for kind in active if isinstance(kind, str)] if isinstance(active, (list, tuple)) else []
+  queued = state.get('gesture_queued')
+  queued = [item.get('kind') for item in queued if isinstance(item, dict) and isinstance(item.get('kind'), str)] if isinstance(queued, (list, tuple)) else []
+  for kind, (_, label) in GESTURE_LABELS.items():
+    if kind in active:
+      return label, 'active', kind
+  if active:
+    return 'Music cue', 'active', None
+  for kind in queued:
+    if kind in GESTURE_LABELS:
+      return 'Next: ' + GESTURE_LABELS[kind][0], 'queued', kind
+  if queued:
+    return 'Next: music cue', 'queued', None
+  # This legacy flag can include queued cues; it does not prove audible playback.
+  if state.get('turn_signal_music'):
+    return 'Turn signal cue', 'reported', 'turn_signal'
+  lead = seconds(state.get('lead'))
+  if state.get('kind') == 'curve' and state.get('phase') == 'anticipation' and lead is not None and lead > 0:
+    return f'Curve ahead / {lead:.1f}s', 'anticipated', 'curve_prepare'
+  return '', '', None
+
+
 def overlay_view(state):
   readiness = str(state.get('readiness') or 'PREPARING').upper()
   degraded = readiness == 'DEGRADED' or any(state.get(k) for k in ('worker_failed', 'holding_accepted_music', 'quality_failures'))
@@ -41,17 +82,9 @@ def overlay_view(state):
     note = 'Quality check / reserve in use'
   elif state.get('job_inflight'):
     note = f'Job {elapsed:.1f}s elapsed' if elapsed is not None else 'New passage in progress'
-  event = ''
-  if state.get('turn_signal_music'):
-    event = 'Signal percussion'
-  elif state.get('gesture_active'):
-    event = display_text(state['gesture_active'][0]).replace('_', ' ')
-  elif state.get('gesture_queued'):
-    event = 'Queued ' + display_text(state['gesture_queued'][0].get('kind')).replace('_', ' ')
-  elif (lead := seconds(state.get('lead'))) is not None and lead > 0:
-    event = f'Curve in {lead:.1f}s'
+  event, event_state, event_kind = gesture_view(state)
   return dict(profile=display_text(profile), section=section, backend=backend, activity=activity,
-              buffered=seconds(state.get('buffered')), note=note, event=event,
+              buffered=seconds(state.get('buffered')), note=note, event=event, event_state=event_state, event_kind=event_kind,
               ready=readiness == 'READY' and not degraded, stored=stored)
 
 
@@ -83,18 +116,18 @@ def draw_panel(rl, font, state, screen_width, screen_height):
   if view['activity'] == 'DEGRADED':
     detail = view['note']
   elif view['event']:
-    detail = view['event'].capitalize()
+    detail = view['event']
   elif view['activity'] == 'GENERATING':
     detail = view['note']
   else:
     detail = 'Archived playback' if view['stored'] else ' / '.join(filter(None, view['backend']))
-  secondary = 'RoadScore' + (' / ' + detail if detail else '')
+  secondary = detail if view['event'] or view['activity'] == 'DEGRADED' else 'RoadScore' + (' / ' + detail if detail else '')
   buffer = view['buffered']
   reserve = '--' if buffer is None else f'{buffer:.0f}s'
   activity_width = measure(view['activity'], 9)
   max_width = min(300, screen_width * .64, screen_width - 24)
   width = min(max_width, max(210, measure(primary, 11) + activity_width + 57,
-                             measure(secondary, 9) + measure(reserve, 9) + 60))
+                             measure(secondary, 10) + measure(reserve, 9) + 60))
   camera_width = screen_width - 64  # Native mici control rail stays unobstructed.
   x, y, height = max(12, (camera_width - width) / 2), 8, 34
   # One quiet translucent surface, with no enclosing badge or oversized title.
@@ -112,7 +145,7 @@ def draw_panel(rl, font, state, screen_width, screen_height):
   rl.draw_circle(int(x + activity_left - 7), int(y + 10), 2, accent)
   text(view['activity'], activity_left, 6, 9, accent, activity_width + 1)
   reserve_width = measure(reserve, 9)
-  text(secondary, 30, 21, 9, accent if view['activity'] == 'DEGRADED' else muted,
+  text(secondary, 30, 21, 10, accent if view['activity'] == 'DEGRADED' or view['event_state'] == 'active' else muted,
        width - reserve_width - 62)
   # Three quiet queue bars distinguish buffered seconds from job elapsed time.
   for i, bar_height in enumerate((3, 5, 7)):

roadscore/tests/test_overlay_view.py

@@ -25,9 +25,27 @@ class OverlayTests(unittest.TestCase):
   def test_gesture_remains_separate_from_generation_timing(self):
     view = overlay_view(dict(readiness='READY', job_inflight=True, generation_elapsed_seconds=123.4,
                              turn_signal_music=True))
-    self.assertEqual(view['event'], 'Signal percussion')
+    self.assertEqual(view['event'], 'Turn signal cue')
     self.assertIn('123.4s', view['note'])
 
+  def test_active_cue_precedes_queued_and_persistent_signal(self):
+    view = overlay_view(dict(turn_signal_music=True, gesture_active=['turn_signal_sustain', 'curve_apex'],
+                             gesture_queued=[{'kind': 'arrival'}]))
+    self.assertEqual(view['event'], 'Curve apex / impact')
+    self.assertEqual(view['event_state'], 'active')
+
+  def test_queued_cue_does_not_claim_current_music(self):
+    view = overlay_view(dict(turn_signal_music=True, gesture_queued=[{'kind': 'turn_signal'}]))
+    self.assertEqual(view['event'], 'Next: Turn signal')
+    self.assertEqual(view['event_state'], 'queued')
+
+  def test_unknown_and_malformed_cues_do_not_invent_a_road_reason(self):
+    self.assertEqual(overlay_view({'gesture_active': ['unknown_future_kind']})['event'], 'Music cue')
+    self.assertEqual(overlay_view({'gesture_active': 'curve_apex', 'gesture_queued': [None, {}, 3]})['event'], '')
+    self.assertEqual(overlay_view({'lead': 4})['event'], '')
+    self.assertEqual(overlay_view(dict(kind='navigation', phase='anticipation', lead=4))['event'], '')
+    self.assertEqual(overlay_view(dict(kind='curve', phase='anticipation', lead=4))['event'], 'Curve ahead / 4.0s')
+
   def test_missing_buffer_is_not_zero(self):
     for value in (None, float('nan'), float('inf'), '12', True):
       self.assertIsNone(overlay_view({'buffered': value})['buffered'])