Connect prepared Mac showcase and scheduled curve cut-drop treatment

roadscore/onroad

@@ -1,5 +1,10 @@
 #!/usr/bin/env bash
 set -euo pipefail
+for argument in "$@"; do
+  if [[ "$argument" == "--prepared-showcase" ]]; then
+    exec python3 "$(dirname "$0")/prototype/mac_showcase.py" "$@"
+  fi
+done
 if [ -f /TICI ]; then
   # A plain device shell does not activate openpilot's Python environment.
   export PYTHONPATH="/data/roadscore-feasibility/venv/lib/python3.12/site-packages${PYTHONPATH:+:$PYTHONPATH}"

roadscore/prototype/app.py

@@ -242,7 +242,12 @@ def callback(out,n,ti,status):
  if curve_reaction is not None:
   curve_fresh=road_model_valid and 0<=callback_wall-command_wall<=.5
   motion_blocked=bool(motion is not None and motion.stopped and motion.fresh)
-  rendered=curve_reaction.process(rendered,frames-n,event_state.get('demo_curve_state',event_state),source_fresh=curve_fresh,blocked=alert_priority or motion_blocked)
+  curve_state=event_state.get('demo_curve_state',event_state)
+  curve_grid=None
+  if curve_state.get('demo_build_drop') and rhythm_timeline is not None:
+   target_frame=frames-n+round((curve_state['predicted_peak']-source_time)*rate)
+   curve_grid=rhythm_timeline.at(target_frame)
+  rendered=curve_reaction.process(rendered,frames-n,curve_state,source_fresh=curve_fresh,blocked=alert_priority or motion_blocked,route_time=source_time,payoff_grid=curve_grid)
  if motion is not None:
   _,motion_fresh=engagement_active(motion_state[1],True,motion_state[2],motion_state[4],motion_state[3],callback_wall)
   rendered=motion.process(rendered,speed=motion_state[0],source_fresh=motion_fresh)
@@ -251,6 +256,8 @@ def callback(out,n,ti,status):
   open_mix=presentation.mix if presentation is not None else 1.
   if motion is not None:open_mix=min(open_mix,motion.dsp.mix)
   cue_gain=shaker_contained_gain+(1.-shaker_contained_gain)*open_mix
+  curve_priority=bool(curve_reaction is not None and curve_reaction.impact is not None and curve_reaction.impact.snapshot().get('rendered_active'))
+  if curve_priority:cue_gain=0.
   shaken=shaker.process(rendered,frames-n,signal_on,signal_fresh,sequence_key=demo_signal_mode,presentation_gain=cue_gain)
   if alert_priority:shaker.rendered_active=False;shaker.rendered_peak=0.;shaker.suppression_reason='meaningful alert priority'
   else:rendered=shaken

roadscore/prototype/curve_reaction.py

@@ -15,10 +15,46 @@ class CurveReaction:
    # blends toward this high-pass, then restores the track's own bass at apex.
    self.dsp.sos=butter(2,self.cutoff,btype='highpass',fs=rate,output='sos').astype(np.float32)
   self.config=PresentationConfig(enabled=True,attack_ms=80. if self.bass_build else 130.,release_ms=180.)
+  self.normal_config=self.config
+  self.impact_config=PresentationConfig(enabled=True,attack_ms=25.,release_ms=180.)
+  self.impact=None;self.impact_activation=None;self.impact_payoff=None;self.impact_start=None;self.impact_grid=None
+  if self.bass_build:
+   from curve_build_drop import CurveBuildDrop
+   self.impact=CurveBuildDrop(rate)
   self.staged=False
   self.activation=None;self.payoff=None;self.target=1.;self.phase='neutral';self.fresh=False;self.end_frame=0;self.valid=False;self.armed=False
 
- def process(self,pcm,start_frame,state,source_fresh=True,blocked=False):
+ def process(self,pcm,start_frame,state,source_fresh=True,blocked=False,route_time=None,payoff_grid=None):
+  requested=self.impact is not None and state.get('demo_build_drop') is True
+  valid=requested and self.enabled and source_fresh and not blocked
+  valid=valid and type(route_time) in (int,float) and math.isfinite(route_time)
+  valid=valid and all(type(state.get(k)) in (int,float) and math.isfinite(state[k]) for k in ('activation','predicted_peak'))
+  if valid and state['activation']!=self.impact_activation:
+   self.impact_activation=state['activation'];self.impact_payoff=None;self.impact_start=None;self.impact_grid=None
+  if valid and self.impact_payoff is None:
+   grid=payoff_grid if payoff_grid is not None else self.grid
+   if grid.usable:
+    desired=start_frame+round((state['predicted_peak']-route_time)*self.rate)
+    step=self.rate*60/grid.bpm/2;origin=grid.beat_phase*self.rate
+    scheduled=round(origin+round((desired-origin)/step)*step)
+    # Arm only before the silence window, never introduce a late fake drop.
+    if scheduled-start_frame>=round(.14*self.rate):
+     self.impact_payoff=scheduled;self.impact_grid=grid
+     self.impact_start=max(0,start_frame+round((state['activation']-route_time)*self.rate))
+  armed=bool(valid and self.impact_payoff is not None)
+  self.config=self.impact_config if armed else self.normal_config
+  filtered_state=state
+  if armed and start_frame+len(pcm)>self.impact_payoff:
+   filtered_state={**state,'phase':'event'}
+  result=self._filter(pcm,start_frame,filtered_state,source_fresh,blocked,self.impact_payoff if armed else None)
+  if self.impact is not None:
+   result=self.impact.process(result,start_frame,self.grid,self.impact_start,self.impact_payoff,
+                              enabled=armed,source_fresh=source_fresh,blocked=blocked)
+  if not valid:
+   self.impact_activation=None;self.impact_payoff=None;self.impact_start=None;self.impact_grid=None
+  return result
+
+ def _filter(self,pcm,start_frame,state,source_fresh=True,blocked=False,scheduled_payoff=None):
   self.end_frame=start_frame+len(pcm)
   amount=state.get('amount',0.)
   self.fresh=bool(source_fresh and isinstance(amount,(int,float)) and math.isfinite(amount))
@@ -31,9 +67,11 @@ class CurveReaction:
    self.activation=activation;self.payoff=None;self.armed=False
   if not valid:
    self.payoff=None;self.armed=False
+  elif scheduled_payoff is not None:self.payoff=scheduled_payoff
   target=1.
   if valid and phase=='anticipation':
-   target=1.-(.98 if self.bass_build else .8)*min(1.,max(0.,amount));self.payoff=None
+   target=1.-(.98 if self.bass_build else .8)*min(1.,max(0.,amount))
+   if scheduled_payoff is None:self.payoff=None
   elif valid and phase=='event' and self.armed:
    if self.payoff is None:
     step=self.rate*60/self.grid.bpm/2 if self.grid.usable else 0.
@@ -61,5 +99,6 @@ class CurveReaction:
           rendered_phase=('apex' if apex else 'build' if build else 'neutral'),
           rendered_active=apex or build,rendered_open_mix=self.dsp.mix,target_open_mix=self.target,cutoff_hz=self.cutoff,maximum_wet_mix=.98 if self.bass_build else .8,
           treatment='bass-return' if self.bass_build else 'filter-opening',filter_type='highpass' if self.bass_build else 'lowpass',
+          build_drop=None if self.impact is None else self.impact.snapshot(),
           payoff_audio_s=None if self.payoff is None else self.payoff/self.rate,attack_ms=self.config.attack_ms,contain_ms=180.,
           added_delay_samples=0,source='known replay route event' if self.staged else 'current causal conductor state',demo_staged=self.staged,unknown_policy='smooth bypass')}

roadscore/prototype/demo_curve_plan.py

@@ -15,6 +15,8 @@ def validate(value):
   raise ValueError('Curve plan requires its exact route identity')
  if type(value.get('replay_start')) is not int or value['replay_start']<0:
   raise ValueError('Curve plan requires its exact replay start')
+ if value.get('treatment','bass-return') not in ('bass-return','build-cut-drop'):
+  raise ValueError('Unknown demo curve treatment')
  events=value.get('curves')
  if not isinstance(events,list) or not 1<=len(events)<=3:
   raise ValueError('Curve plan requires one to three measured events')
@@ -61,5 +63,5 @@ class ReplayCurvePlan:
     amount=progress*progress*(3.-2.*progress)
     return {**recorded,'kind':'curve','activation':start,'predicted_peak':apex,
             'phase':'anticipation' if now<apex else 'event','amount':amount,
-            'demo_staged_curve':True}
+            'demo_staged_curve':True,'demo_build_drop':self.value.get('treatment')=='build-cut-drop'}
   return recorded

roadscore/prototype/mac_showcase.py

@@ -86,7 +86,7 @@ def audio_worker(a):
       raise ValueError('Staged curve plan does not match prepared route/start')
     staged = ReplayCurvePlan(value)
     write_json(a.out/'demo_curve_plan.json',value)
-  state = dict(active=False,signal_on=False,fresh=False,car_fresh=False,model_fresh=False,speed=0.,alert_key='',alert_meaningful=False,curve={'kind':'curve','phase':'neutral','amount':0.,'activation':None})
+  state = dict(active=False,signal_on=False,fresh=False,car_fresh=False,model_fresh=False,speed=0.,alert_key='',alert_meaningful=False,route_t=None,curve={'kind':'curve','phase':'neutral','amount':0.,'activation':None})
   received = {name:float('-inf') for name in sm.services}
   anchor = None
   position = None
@@ -132,11 +132,13 @@ def audio_worker(a):
         def fresh(name):return bool(sm.valid[name] and 0<=now-received[name]<=.6 and 0<=(latest-sm.logMonoTime[name])/1e9<=1.)
         c=sm['carState'];s=sm['selfdriveState']
         update=dict(active=bool(s.active) and fresh('selfdriveState'),signal_on=bool(c.leftBlinker or c.rightBlinker),fresh=fresh('selfdriveState'),car_fresh=fresh('carState'),model_fresh=fresh('modelV2'),speed=float(c.vEgo),alert_key=str(s.alertType),alert_meaningful=int(s.alertStatus.raw)>0 and int(s.alertSize.raw)>0,curve=state['curve'])
+        update['route_t']=state['route_t']
         if sm.updated['modelV2']:
           mono=sm.logMonoTime['modelV2']
           if anchor is None:anchor=(mono,now);started=now
           if abs((mono-anchor[0])/1e9-(now-anchor[1]))>.75:raise RuntimeError('Prepared showcase left its 1x replay clock')
           route_t=(mono-meta['first_model_ns'])/1e9
+          update['route_t']=route_t
           m=sm['modelV2']
           curve=conductor.update(route_t,{'mono':mono,'eof':m.timestampEof,'t':list(m.orientationRate.t),'yaw':list(m.orientationRate.z),'v':list(m.velocity.x)},float(c.vEgo)) if fresh('modelV2') and len(m.position.t)==33 else conductor.state(route_t)
           update['curve']=staged.state(route_t,a.route,curve) if staged else curve
@@ -152,6 +154,7 @@ def audio_worker(a):
         if started and now-started>=a.duration:break
         if started and now-received['modelV2']>2 and (position or 0)/rate < len(audio)/rate-2:raise RuntimeError('Replay model stream stopped before prepared audio ended')
   finally:
+    write_json(a.out/'audio_drained.json',dict(wall=time.monotonic(),drained=not errors))
     server.shutdown();trace.close()
     write_json(a.out/'prepared_summary.json',dict(generation_invoked=False,source=str(a.score_archive),first_source_frame=first_frame,last_source_frame=position,sample_rate=rate,portaudio_flags=flags,max_clock_error_seconds=max_drift,callback_errors=errors,muted=a.muted,session_id=session,manual_scope='isolated replay display and presentation only'))
   if errors:raise RuntimeError(errors[0])
@@ -206,7 +209,8 @@ def main():
   out.mkdir(parents=True,exist_ok=False)
   env=os.environ.copy()
   env.update(PYTHONDONTWRITEBYTECODE='1',ZMQ='1',OPENPILOT_ZMQ_NAMESPACE='roadscore-showcase-'+session,ROADSCORE_SHOWCASE_SESSION=session,ROADSCORE_PREPARED_SHOWCASE='1',ROADSCORE_REPLAY_UI_CONTROLS='1',PARAMS_ROOT=str(out/'params'),BASEDIR=str(rt),NOBOARD='1',SIMULATION='1',SKIP_FW_QUERY='1',BIG='0',SP_ALLOW_DESKTOP_FAKE_WIFI='0',SP_ALLOW_DESKTOP_FAKE_BLUETOOTH='0',SP_ONROAD_NAV_DEMO='0',SP_ONROAD_CEM_DEMO='0',ROADSCORE_CLEAN_DEMO_UI='1',ROADSCORE_OVERLAY='1',ROADSCORE_STATUS_FILE=str(out/'status.json'),ROADSCORE_UI_AUDIT=str(out/'ui_audit.jsonl'),ROADSCORE_OVERLAY_CAPTURE=str(out/'overlay.png'),ROADSCORE_PRESENTATION_POLICY='conservative-v4')
-  env.pop('OPENPILOT_PREFIX',None)
+  env['OPENPILOT_PREFIX']='roadscore-showcase-'+session
+  env['ROADSCORE_AUDIO_DRAIN_FILE']=str(out/'audio_drained.json')
   env['PWD']=str(rt)
   env['PYTHONPATH']=':'.join(map(str,[HERE,rt,rt/'starpilot/third_party',*rt.glob('*_repo'),project/'roadscore/.analysis-venv/lib/python3.12/site-packages']))
   args=launch['native_replay_args'][:]

roadscore/prototype/normal_ui_audit.py

@@ -28,7 +28,7 @@ if replay_controls.enabled:
   if isinstance(sm,ReplayStateView) and apply_turn_intent(self,replay_arrow_mode):return
   return original_turn_intent(self)
  TurnIntent._update_state=replay_turn_intent
-display_hold=ReplayDisplayHold(os.environ.get("ROADSCORE_AUDIO_DRAIN_FILE"), enabled=os.environ.get("OPENPILOT_PREFIX")=="roadscore_replay")
+display_hold=ReplayDisplayHold(os.environ.get("ROADSCORE_AUDIO_DRAIN_FILE"), enabled=os.environ.get("OPENPILOT_PREFIX")=="roadscore_replay" or replay_controls.enabled)
 original_state=UIState._update_state
 def replay_state(self,*args,**kwargs):
  result=original_state(self,*args,**kwargs)

roadscore/prototype/overlay_view.py

@@ -83,7 +83,7 @@ def overlay_view(state):
     activity = 'PREPARING'
   stored = state.get('compute') == 'none'
   composer = display_text(state.get('composer')).upper()
-  backend = ('STORED SCORE', 'NO COMPUTE') if stored else ((composer, 'CHESTNUT') if composer in ('ACE', 'SA3') else ('LOCAL SCORE', ''))
+  backend = ('PREPARED', 'LOCAL') if state.get('compute')=='prepared-core' else (('STORED SCORE', 'NO COMPUTE') if stored else ((composer, 'CHESTNUT') if composer in ('ACE', 'SA3') else ('LOCAL SCORE', '')))
   profile = state.get('style') or str(state.get('profile') or 'Preparing').title()
   section = display_text(state.get('section') or 'Waiting for score').upper()
   section = section.removesuffix(' / CONTINUOUS')

roadscore/prototype/prepared_core.py

@@ -69,10 +69,14 @@ class PreparedPresentation:
     self.curve.grid = self.alert.grid = grid
     competing = self.shaker.active or abs(self.engagement.mix-float(active)) > .01
     result = self.alert.process(pcm, frame, state['alert_key'], state['alert_meaningful'], state['fresh'], competing)
-    result = self.curve.process(result, frame, state['curve'], source_fresh=state['model_fresh'], blocked=self.alert.priority_active or (self.motion.stopped and self.motion.fresh))
+    curve_grid=None
+    if state['curve'].get('demo_build_drop') and state.get('route_t') is not None:
+      curve_grid=self.rhythm.at(frame+round((state['curve']['predicted_peak']-state['route_t'])*self.rhythm.rate))
+    result = self.curve.process(result, frame, state['curve'], source_fresh=state['model_fresh'], blocked=self.alert.priority_active or (self.motion.stopped and self.motion.fresh),route_time=state.get('route_t'),payoff_grid=curve_grid)
     result = self.motion.process(result, speed=state['speed'], source_fresh=state['car_fresh'])
     result = self.engagement.process(result, active, self.config)
     gain = self.contained_gain+(1-self.contained_gain)*min(self.engagement.mix, self.motion.dsp.mix)
+    if self.curve.impact is not None and self.curve.impact.snapshot().get('rendered_active'):gain=0.
     shaken = self.shaker.process(result, frame, signal_on, state['car_fresh'], sequence_key=signal_mode, presentation_gain=gain)
     if not self.alert.priority_active:
       result = shaken

roadscore/prototype/review_curve_build.py

@@ -37,7 +37,8 @@ def main():
    state={key:row.get(key) for key in ('kind','phase','amount','activation','predicted_peak')}
    if staged:state=plan.state(row['route_t'],plan.value['route'],state)
    dsp.grid=timeline.at(frame)
-   began=time.perf_counter();output.append(dsp.process(wave[offset:offset+4800],frame,state));timings.append(time.perf_counter()-began)
+   target=frame+round((state.get('predicted_peak',row['route_t'])-row['route_t'])*rate) if state.get('demo_build_drop') else frame
+   began=time.perf_counter();output.append(dsp.process(wave[offset:offset+4800],frame,state,route_time=row['route_t'],payoff_grid=timeline.at(target)));timings.append(time.perf_counter()-began)
    events.append({'audio_s':frame/rate,'route_s':row['route_t'],**dsp.snapshot()['curve_reaction']})
   audio=np.concatenate(output);sf.write(args.output/name,audio,rate,subtype='FLOAT')
   report['files'][name]={'peak':float(abs(audio).max()),'over_full_scale':int(np.count_nonzero(abs(audio)>1)),

roadscore/prototype/test_curve_reaction.py

@@ -79,6 +79,20 @@ class CurveTests(unittest.TestCase):
   self.assertTrue(np.array_equal(source,original));self.assertTrue(np.isfinite(wave).all())
   self.assertLess(np.abs(wave).max(),1.)
   self.assertEqual(dsp.snapshot()['curve_reaction']['source'],'known replay route event')
+ def test_explicit_cut_drop_schedules_before_apex_and_returns_to_core(self):
+  rate=48000;t=np.arange(rate*4)/rate
+  pcm=np.column_stack([.2*np.sin(2*np.pi*80*t)+.1*np.sin(2*np.pi*1000*t)]*2).astype('float32')
+  grid=ShakerGrid(120,0,1,1,True,'fixture');dsp=CurveReaction(grid,enabled=True,bass_build=True)
+  output=[]
+  for frame in range(0,len(pcm),4800):
+   now=frame/rate
+   state=dict(kind='curve',phase='anticipation' if now<3 else 'event',amount=min(1.,now/3),activation=0.,predicted_peak=3.,demo_staged_curve=True,demo_build_drop=True)
+   output.append(dsp.process(pcm[frame:frame+4800],frame,state,route_time=now,payoff_grid=grid))
+  result=np.concatenate(output)
+  self.assertEqual(dsp.impact_payoff,3*rate)
+  self.assertEqual(dsp.impact.snapshot()['actual_payoff_frame'],3*rate)
+  self.assertLess(np.max(abs(result[round(2.9*rate):3*rate])),1e-6)
+  np.testing.assert_array_equal(result[round(3.6*rate):],pcm[round(3.6*rate):])
 
 
 if __name__=='__main__':unittest.main()