Give meaningful warnings bounded beat-locked fill priority

roadscore/prototype/alert_accent.py

@@ -1,47 +1,60 @@
-"""Sparse native-alert accents using the signal shaker's percussion vocabulary."""
+"""Sparse beat-locked warning fills; bounded priority over routine musical cues."""
 import math
 import numpy as np
-from signal_shaker import SignalShaker
 
 
 class AlertAccent:
  def __init__(self, grid, rate=48000, enabled=False):
   self.grid=grid;self.rate=rate;self.enabled=enabled
-  self.grain=SignalShaker(grid,rate,peak=.016).grain
+  n=round(.11*rate);t=np.arange(n)/rate;rng=np.random.default_rng(1701)
+  frequency=np.fft.rfftfreq(n,1/rate);spectrum=np.fft.rfft(rng.standard_normal(n))
+  spectrum*=np.clip((frequency-650)/700,0,1)*np.clip((6500-frequency)/2000,0,1)
+  grain=np.fft.irfft(spectrum,n);envelope=(1-np.exp(-t/.004))*np.exp(-t/.032)
+  grain*=envelope;grain/=max(float(np.max(np.abs(grain))),1e-9)
+  self.grain=np.column_stack((grain,grain)).astype(np.float32)*.045
+  self.duck=(np.sin(np.pi*np.arange(n)/(n-1))**2).astype(np.float32)
   self.key=None;self.since=0;self.handled=False;self.pending=[]
-  self.last=-60*rate;self.events=[];self.rendered_active=False
+  self.last=-60*rate;self.events=[];self.rendered_active=False;self.priority_active=False
 
  def process(self, pcm, start, key, meaningful, fresh, competing=False):
-  self.rendered_active=False
+  self.rendered_active=False;self.priority_active=False
   if not self.enabled:return pcm
   current=key if meaningful and fresh else None
   if current!=self.key:
-   self.key=current;self.since=start;self.handled=False;self.pending=[]
-  if not fresh or competing:
+   self.key=current;self.since=start;self.handled=False
+  if not fresh:
    self.pending=[]
-   if current:self.handled=True
    return pcm
-  if self.grid.usable and current and not self.handled and start-self.since>=round(.2*self.rate):
-   self.handled=True
-   if start-self.last>=12*self.rate and sum(start-e['frame']<60*self.rate for e in self.events)<3:
-    beat=self.rate*60/self.grid.bpm;origin=self.grid.beat_phase*self.rate
-    frame=round(origin+math.ceil((start-origin)/beat-1e-10)*beat)
-    self.pending=[(frame,1.),(frame+round(beat/2),.55)]
-    self.last=start;self.events.append({'kind':'native_alert_accent','frame':start,'scheduled_frame':frame,'alert':current})
-  end=start+len(pcm);overlay=None;remaining=[]
+  elapsed=(start-self.since)/self.rate
+  if self.grid.usable and current and not self.handled and elapsed>=.2:
+   # Ordinary cues can defer a warning briefly, but cannot permanently erase it.
+   if not competing or elapsed>=.7:
+    self.handled=True
+    if start-self.last>=12*self.rate and sum(start-e['frame']<60*self.rate for e in self.events)<3:
+     beat=self.rate*60/self.grid.bpm;origin=self.grid.beat_phase*self.rate
+     frame=round(origin+math.ceil((start-origin)/beat-1e-10)*beat)
+     self.pending=[(frame,1.),(frame+round(beat/2),.55),(frame+round(beat),.8)]
+     self.last=start;self.events.append({'kind':'native_alert_fill','frame':start,'scheduled_frame':frame,'alert':current,
+                                       'competition_deferred':bool(competing),'priority':'warning over routine cues'})
+  self.priority_active=bool(self.pending)
+  end=start+len(pcm);overlay=None;duck=None;remaining=[]
   for frame,gain in self.pending:
    left=max(start,frame);right=min(end,frame+len(self.grain))
    if right>left:
-    if overlay is None:overlay=np.zeros_like(pcm)
+    if overlay is None:overlay=np.zeros_like(pcm);duck=np.zeros(len(pcm),np.float32)
     overlay[left-start:right-start]+=self.grain[left-frame:right-frame]*gain
+    duck[left-start:right-start]=np.maximum(duck[left-start:right-start],self.duck[left-frame:right-frame]*gain)
    if frame+len(self.grain)>end:remaining.append((frame,gain))
   self.pending=remaining
-  # Never create an isolated notification sound in a musical rest.
+  # Do not fill an intentional musical rest with an isolated notification.
   if overlay is None or np.max(np.abs(pcm),initial=0)<.002:return pcm
   self.rendered_active=bool(np.any(overlay))
-  return pcm+overlay
+  base=pcm*(1-(1-10**(-1.5/20))*duck[:,None])
+  headroom=np.maximum(1-np.abs(base),0)
+  return base+np.clip(overlay,-headroom,headroom)
 
  def snapshot(self):
   return {'enabled':self.enabled,'rhythm_enabled':self.enabled and self.grid.usable,
-          'rendered_active':self.rendered_active,'minimum_spacing_seconds':12,
-          'maximum_per_minute':3,'source':'selfdriveState.alertStatus'}
+          'rendered_active':self.rendered_active,'priority_active':self.priority_active,
+          'minimum_spacing_seconds':12,'maximum_per_minute':3,'maximum_competition_deferral_seconds':.5,
+          'peak_limit':.045,'duck_db':1.5,'fill':'beat, eighth, next beat','source':'selfdriveState.alertStatus'}

roadscore/prototype/review_alert_accent.py

@@ -0,0 +1,54 @@
+"""Offline warning A/B with recorded event timing and exactly the archived music."""
+import argparse,hashlib,json,time
+from pathlib import Path
+import numpy as np
+import soundfile as sf
+from alert_accent import AlertAccent
+from signal_shaker import ShakerGrid
+from engagement_presentation import EngagementPresentation,PresentationConfig
+
+
+def sha(path):return hashlib.sha256(path.read_bytes()).hexdigest()
+
+def render(locked,out):
+ source=locked/'render/heard.wav';trace=locked/'render/trace.jsonl';timing=locked/'render/audio_blocks.jsonl';audit=locked/'source_alert_audit.private.json'
+ paths=(source,trace,timing,audit);hashes={str(p):sha(p) for p in paths}
+ rows=[json.loads(x) for x in trace.read_text().splitlines()];blocks=[json.loads(x) for x in timing.read_text().splitlines()]
+ events=[e for e in json.loads(audit.read_text())['intervals'] if e['eligible_status_and_size']]
+ if not events:raise ValueError('No recorded meaningful alert for comparison')
+ out.mkdir(parents=True,exist_ok=False);info=sf.info(source);rate=info.samplerate
+ accent=AlertAccent(ShakerGrid(128,0,0,0,False,'not yet observed'),rate,True)
+ delta_filter=EngagementPresentation(rate);config=PresentationConfig(enabled=True)
+ index=-1;observations=[];times=[]
+ with sf.SoundFile(source) as src,sf.SoundFile(out/'full_alert.wav','w',samplerate=rate,channels=2,subtype='FLOAT') as dst:
+  for i,block in enumerate(blocks):
+   start=round(block['audio_s']*rate);end=round(blocks[i+1]['audio_s']*rate) if i+1<len(blocks) else info.frames
+   pcm=src.read(end-start,dtype='float32',always_2d=True)
+   while index+1<len(rows) and rows[index+1]['command_wall']<=block['callback_wall']:index+=1
+   row=rows[index] if index>=0 else {};route_t=row.get('route_t',-1)
+   grid=row.get('signal_shaker',{}).get('grid')
+   if grid:accent.grid=ShakerGrid(**grid)
+   event=next((e for e in events if e['start_demo_s']<=route_t<=e['start_demo_s']+e['duration_observed_s']),None)
+   fresh=bool(block.get('engagement_fresh') and row and 0<=block['callback_wall']-row['command_wall']<=1.)
+   competing=bool(row.get('signal_shaker',{}).get('sequence_active') or row.get('core_apex',{}).get('rendered_active'))
+   before=time.perf_counter();wet=accent.process(pcm,start,str(event['key']) if event else None,event is not None,fresh,competing)
+   # Baseline already contains engagement processing. Filter only the new delta;
+   # this approximates placement before that stage without filtering music twice.
+   delta=delta_filter.process(wet-pcm,bool(block.get('engagement_active')),config)
+   y=pcm+delta;times.append(time.perf_counter()-before);dst.write(y)
+   observations.append(dict(audio_s=block['audio_s'],route_t=route_t,meaningful=event is not None,competing=competing,**accent.snapshot()))
+ first=next(x for x in observations if x['rendered_active']);start=max(0.,first['audio_s']-8);duration=min(24.,info.duration-start)
+ for name,path in [('A_baseline.wav',source),('B_alert.wav',out/'full_alert.wav')]:
+  with sf.SoundFile(path) as handle:
+   handle.seek(round(start*rate));wave=handle.read(round(duration*rate),dtype='float32',always_2d=True)
+  sf.write(out/name,wave,rate,subtype='FLOAT')
+ if hashes!={str(p):sha(p) for p in paths}:raise RuntimeError('Protected input changed')
+ report=dict(source_sha256=hashes,source_unchanged=True,excerpt_start_audio_s=start,duration_seconds=duration,events=accent.events,
+             approximation='New alert delta passed through reconstructed engagement filter; archived prior shaker/apex remains. Native priority ordering can differ. Same generated PCM, no normalization.',
+             runtime_cost='Mac offline only',median_ms=float(np.median(times)*1000),p95_ms=float(np.quantile(times,.95)*1000))
+ (out/'report.json').write_text(json.dumps(report,indent=2));(out/'alert_states.json').write_text(json.dumps(observations))
+ (out/'index.html').write_text('<!doctype html><meta name="viewport" content="width=device-width,initial-scale=1"><title>Warning fill A/B</title><style>body{background:#101016;color:#eee;font:18px system-ui;max-width:760px;margin:40px auto;padding:20px}audio{width:100%}</style><h1>Warning fill A/B</h1><p>Same archived music and recorded warning. B adds one beat-locked percussion fill with a brief gentle duck. Existing archived cues remain; live priority will make them yield. This is an offline approximation, not a new hardware recording.</p><h2>A · Baseline</h2><audio controls src="A_baseline.wav"></audio><h2>B · Warning fill</h2><audio controls src="B_alert.wav"></audio>')
+ return report
+
+if __name__=='__main__':
+ p=argparse.ArgumentParser();p.add_argument('locked',type=Path);p.add_argument('output',type=Path);a=p.parse_args();print(json.dumps(render(a.locked,a.output),indent=2))

roadscore/prototype/test_alert_accent.py

@@ -12,15 +12,18 @@ class AccentTests(unittest.TestCase):
   for i,key in enumerate(alerts):
    wet=accent.process(source,i*800,key,bool(key),fresh,competing)
    heard |= bool(np.any(wet!=source))
-   self.assertLessEqual(float(np.max(np.abs(wet-source))),.016001)
+   self.assertLessEqual(float(np.max(np.abs(wet-source))),.061)
   return accent,heard
  def test_persistent_alert_once_and_grid_aligned(self):
   accent,heard=self.render(['takeover']*200)
   self.assertTrue(heard);self.assertEqual(len(accent.events),1)
   self.assertEqual(accent.events[0]['scheduled_frame'],5000)
- def test_tiny_changes_and_competing_cues_are_omitted(self):
+ def test_tiny_changes_omitted_but_competing_cues_do_not_erase_warning(self):
   self.assertFalse(self.render(['a','b']*100)[1])
-  self.assertFalse(self.render(['a']*100,competing=True)[1])
+  accent,heard=self.render(['a']*100,competing=True)
+  self.assertTrue(heard);self.assertEqual(len(accent.events),1)
+  self.assertGreaterEqual(accent.events[0]['frame'],round(.7*8000))
+  self.assertLess(accent.events[0]['scheduled_frame'],round(1.3*8000))
   self.assertFalse(self.render(['a']*100,fresh=False)[1])
  def test_density_limit(self):
   accent,_=self.render([str(i//30) for i in range(600)])
@@ -33,6 +36,17 @@ class AccentTests(unittest.TestCase):
   for i in range(3):a.process(x,i*800,'a',True,True)
   self.assertTrue(a.pending)
   self.assertIs(a.process(x,2400,'a',True,False),x);self.assertFalse(a.pending)
+ def test_headroom_and_no_notification_in_rest(self):
+  a=AlertAccent(GRID,rate=8000,enabled=True);x=np.full((800,2),.999,np.float32)
+  for i in range(20):
+   y=a.process(x,i*800,'warning',True,True,True)
+   self.assertLessEqual(np.max(np.abs(y)),1.)
+  b=AlertAccent(GRID,rate=8000,enabled=True);silence=np.zeros_like(x)
+  for i in range(20):self.assertIs(b.process(silence,i*800,'warning',True,True,True),silence)
+ def test_disabling_and_stale_input_remain_bypass(self):
+  a=AlertAccent(GRID,rate=8000,enabled=False);x=np.full((800,2),.1,np.float32)
+  self.assertIs(a.process(x,8000,'warning',True,True),x)
+  self.assertFalse(a.priority_active)
  def test_short_opening_assessed_without_assuming_confidence(self):
   # Real estimator must accept a 23s opening; silence must still veto cues.
   grid,_=assess_grid(np.zeros((23*8000,2),np.float32),8000,128)