diff --git a/roadscore/DEMO.md b/roadscore/DEMO.md index 5c7493f257..e472377d38 100644 --- a/roadscore/DEMO.md +++ b/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. diff --git a/roadscore/prototype/overlay.py b/roadscore/prototype/overlay.py index de35570a36..8f4392e756 100644 --- a/roadscore/prototype/overlay.py +++ b/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=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): diff --git a/roadscore/prototype/overlay_view.py b/roadscore/prototype/overlay_view.py index 4c6c62d121..adcfd6500b 100644 --- a/roadscore/prototype/overlay_view.py +++ b/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)): diff --git a/roadscore/tests/test_overlay_view.py b/roadscore/tests/test_overlay_view.py index f1049be3b0..0a0c8f69f3 100644 --- a/roadscore/tests/test_overlay_view.py +++ b/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'])