Expose actual rendered cue and engagement state for honest UI

roadscore/prototype/MUSICAL_PRESENTATION.md

@@ -84,3 +84,21 @@ The normal launcher must use gold-core when this fragment is eventually selected
 copying the fragment is not itself an instruction to run hardware. Defaults in
 existing runtime configurations remain unchanged. Full archive source hashes and
 runtime configuration record the selected presentation policy version.
+
+## Honest UI contract
+
+`status.signal_shaker.rendered_active` is true only when the most recent audio
+block actually contains nonzero added shaker samples (including the release
+tail). `rendered_peak` is that addition's peak; block start/end seconds use the
+same audio timeline as `status.elapsed`. It resets false every callback, even
+when a signal remains active after its32-pulse budget. `sequence_active` and
+`rhythm_enabled` alone do not establish audible output. UI should expire an
+indicator if elapsed minus rendered_block_end_seconds exceeds0.25s, or if the
+status itself is stale. Never display a queued prediction as a played cue.
+
+`status.core_apex.rendered_active` similarly means a source-bearing block was
+actually attenuated; it is false at the unity apex. The same block-end timestamp
+expires its indicator. Engagement exposes `rendered_open_mix` and
+`rendered_state` (`open`, `transition`, `contained`) after actual processing,
+plus `rendered_block_end_seconds`. Display it only when enabled, current and
+actually applied; `active` is input provenance, not proof the ramp has finished.

roadscore/prototype/app.py

@@ -349,6 +349,7 @@ try:
      engagement_on,engagement_fresh=engagement_active(engagement[1],engagement[0],engagement[2],engagement[4],engagement[3],time.monotonic())
      snapshot.update(presentation.snapshot(presentation_config,engagement_on,engagement_fresh))
     if shaker is not None:snapshot['signal_shaker']=shaker.snapshot()
+    if apex is not None:snapshot['core_apex']=apex.snapshot()
     if songform:snapshot.update(songform.snapshot())
     if composition:snapshot.update(composition.snapshot(frames/rate))
     if gestures:snapshot.update(gestures.status())

roadscore/prototype/core_apex.py

@@ -4,8 +4,9 @@ import numpy as np
 
 class CoreApex:
  def __init__(self, grid, rate=48000, enabled=False):
-  self.grid=grid;self.rate=rate;self.enabled=enabled;self.last_activation=None;self.peak=None;self.events=[]
+  self.grid=grid;self.rate=rate;self.enabled=enabled;self.last_activation=None;self.peak=None;self.events=[];self.rendered_active=False;self.rendered_end=0.
  def process(self, pcm, start_frame, state):
+  self.rendered_active=False;self.rendered_end=(start_frame+len(pcm))/self.rate
   if not self.enabled or not self.grid.usable:return pcm
   activation=state.get('activation');lead=state.get('lead');now=start_frame/self.rate
   if (state.get('kind')=='curve' and state.get('phase')=='anticipation' and activation is not None
@@ -20,4 +21,7 @@ class CoreApex:
   # Half-second breath, then original unity source at the estimated apex. Never boosts/clips the hook.
   gain=np.ones(len(pcm),np.float32);inside=(offset>-.7)&(offset<0)
   gain[inside]=1-(1-10**(-1/20))*np.sin(np.pi*(offset[inside]+.7)/.7)**2
-  return pcm if not inside.any() else pcm*gain[:,None]
+  self.rendered_active=bool(inside.any() and np.any(pcm[inside]))
+  return pcm if not self.rendered_active else pcm*gain[:,None]
+ def snapshot(self):
+  return {'enabled':self.enabled,'rhythm_enabled':self.enabled and self.grid.usable,'rendered_active':self.rendered_active,'rendered_block_end_seconds':self.rendered_end}

roadscore/prototype/engagement_presentation.py

@@ -45,6 +45,7 @@ def engagement_active(valid, active, source_ns, latest_source_ns, received_wall,
 class EngagementPresentation:
   def __init__(self, rate=48000, max_frames=4800):
     self.rate = rate
+    self.frames_processed = 0
     self.max_frames = max_frames
     self.sos = butter(2, 4500, fs=rate, output='sos').astype(np.float32)
     self.zi = np.zeros((len(self.sos), 2, 2), np.float32)
@@ -65,6 +66,7 @@ class EngagementPresentation:
     """
     if not len(pcm):
       return pcm
+    self.frames_processed += len(pcm)
     # Keep filter warm even during bypass; allocated size bounded by max_frames.
     target = 1. if active or not config.enabled else 0.
     output = None if target == 1. and self.mix == 1. else np.empty_like(pcm)
@@ -101,5 +103,7 @@ class EngagementPresentation:
 
   def snapshot(self, config, active, fresh):
     return {'engagement_presentation': {**asdict(config), 'active': active,
-            'input_fresh': fresh, 'source': 'selfdriveState.active',
+            'input_fresh': fresh, 'rendered_open_mix': self.mix,
+            'rendered_state': ('open' if self.mix >= 1. else 'contained' if self.mix <= 0. else 'transition'),
+            'rendered_block_end_seconds': self.frames_processed/self.rate, 'source': 'selfdriveState.active',
             'unknown_policy': 'contained', 'added_delay_samples': 0}}

roadscore/prototype/signal_shaker.py

@@ -43,6 +43,7 @@ class SignalShaker:
   self.debounce=round(debounce_seconds*rate);self.release=round(.16*rate)
   self.last_on=None;self.active=False;self.next_tick=None;self.stop_frame=None;self.tail=np.zeros((0,2),np.float32)
   self.sequence_pulses=0;self.sequence_start=0
+  self.rendered_active=False;self.rendered_peak=0.;self.rendered_start=0;self.rendered_end=0
   self.sequence_starts=[];self.pulse_frames=[];self.events=[]
   # Fixed filtered grains, prepared before audio. No callback RNG or file/FFT operations.
   n=round(.085*rate);t=np.arange(n)/rate;rng=np.random.default_rng(1701)
@@ -53,6 +54,7 @@ class SignalShaker:
   self.grain=np.column_stack((grain,grain)).astype(np.float32)*self.peak
   self.step=rate*60/grid.bpm/2 if grid.usable else 1.;self.origin=grid.beat_phase*rate
  def process(self,pcm,start_frame,signal_on,signal_fresh):
+  self.rendered_active=False;self.rendered_peak=0.;self.rendered_start=start_frame;self.rendered_end=start_frame+len(pcm)
   if not self.enabled or not self.grid.usable:return pcm
   end=start_frame+len(pcm)
   if signal_fresh and signal_on:
@@ -76,7 +78,8 @@ class SignalShaker:
   if self.stop_frame is not None:
    envelope=np.clip(1-(np.arange(start_frame,end)-self.stop_frame)/self.release,0,1).astype(np.float32)
    overlay*=envelope[:,None]
-  if not np.any(overlay):return pcm
+  self.rendered_peak=float(np.max(np.abs(overlay),initial=0));self.rendered_active=self.rendered_peak>0
+  if not self.rendered_active:return pcm
   return pcm+overlay
  def snapshot(self):
-  return {'enabled':self.enabled,'grid':asdict(self.grid),'rhythm_enabled':self.enabled and self.grid.usable,'uncertain_policy':'no added pulses','sequence_active':self.active,'peak_limit':self.peak,'subdivision':'eighth notes','debounce_seconds':self.debounce/self.rate,'maximum_pulses_per_sequence':32}
+  return {'rendered_active':self.rendered_active,'rendered_peak':self.rendered_peak,'rendered_block_start_seconds':self.rendered_start/self.rate,'rendered_block_end_seconds':self.rendered_end/self.rate,'enabled':self.enabled,'grid':asdict(self.grid),'rhythm_enabled':self.enabled and self.grid.usable,'uncertain_policy':'no added pulses','sequence_active':self.active,'peak_limit':self.peak,'subdivision':'eighth notes','debounce_seconds':self.debounce/self.rate,'maximum_pulses_per_sequence':32}

roadscore/prototype/test_engagement_presentation.py

@@ -60,6 +60,14 @@ class PresentationTests(unittest.TestCase):
       self.assertEqual(engagement_active(valid,True,stamp,latest,10.,wall),(False,False))
     self.assertEqual(engagement_active(True,False,2e9,2.1e9,10.,10.1),(False,True))
 
+  def test_status_reports_applied_mix_not_requested_state(self):
+    dsp=EngagementPresentation();x=np.ones((4800,2),np.float32)*.1
+    dsp.process(x,False,ON);status=dsp.snapshot(ON,False,True)['engagement_presentation']
+    self.assertEqual(status['rendered_state'],'transition');self.assertGreater(status['rendered_open_mix'],0)
+    self.assertEqual(status['rendered_block_end_seconds'],.1)
+    for _ in range(8):dsp.process(x,False,ON)
+    self.assertEqual(dsp.snapshot(ON,False,True)['engagement_presentation']['rendered_state'],'contained')
+
   def test_configuration_fail_closed_and_clamps(self):
     self.assertFalse(PresentationConfig.read({'version':3,'enabled':True}).enabled)
     self.assertFalse(PresentationConfig.read({'enabled':'true'}).enabled)

roadscore/prototype/test_signal_shaker.py

@@ -30,6 +30,12 @@ class ShakerTests(unittest.TestCase):
   for i in range(20):
    y=s.process(x,i*4800,True,True);self.assertLessEqual(abs(y-x).max(),.012001)
   np.testing.assert_array_equal(x,before)
+ def test_audible_flag_expires_after_budget_despite_active_sequence(self):
+  s=SignalShaker(GRID,enabled=True);x=np.zeros((4800,2),np.float32);heard=False
+  for i in range(100):
+   s.process(x,i*4800,True,True);heard |= s.snapshot()['rendered_active']
+  self.assertTrue(heard);self.assertTrue(s.active);self.assertFalse(s.snapshot()['rendered_active'])
+  self.assertEqual(s.snapshot()['rendered_block_end_seconds'],10.)
  def test_new_sequence_after_quiet_gap(self):
   s=SignalShaker(GRID,enabled=True);x=np.zeros((4800,2),np.float32)
   for i in range(50):s.process(x,i*4800,i<5 or i>=35,True)