Add causal curve contrast and shorten warning quantization delay

roadscore/prototype/alert_accent.py

@@ -26,16 +26,15 @@ class AlertAccent:
    self.pending=[]
    return pcm
   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'})
+  if self.grid.usable and current and not self.handled and elapsed>=.05:
+   # Warning priority replaces routine-cue deferral; qualify one stable input first.
+   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/2)-1e-10)*(beat/2))
+    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:
@@ -56,5 +55,5 @@ class AlertAccent:
  def snapshot(self):
   return {'enabled':self.enabled,'rhythm_enabled':self.enabled and self.grid.usable,
           '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'}
+          'minimum_spacing_seconds':12,'maximum_per_minute':3,'maximum_competition_deferral_seconds':0.,'qualification_seconds':.05,
+          'peak_limit':.045,'duck_db':1.5,'fill':'next eighth, plus eighth, plus beat','source':'selfdriveState.alertStatus'}

roadscore/prototype/curve_reaction.py

@@ -0,0 +1,46 @@
+"""Live conductor-driven curve contrast, using only the current music samples."""
+import math
+from engagement_presentation import EngagementPresentation,PresentationConfig
+
+
+class CurveReaction:
+ def __init__(self,grid,rate=48000,enabled=False):
+  self.grid=grid;self.rate=rate;self.enabled=enabled
+  self.dsp=EngagementPresentation(rate,cutoff_hz=1800.,width=.92,gain=1.)
+  self.config=PresentationConfig(enabled=True,attack_ms=130.,release_ms=180.)
+  self.activation=None;self.payoff=None;self.target=1.;self.phase='neutral';self.fresh=False;self.end_frame=0
+
+ def process(self,pcm,start_frame,state,source_fresh=True,blocked=False):
+  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))
+  phase=state.get('phase','neutral');activation=state.get('activation')
+  valid=self.enabled and self.fresh and not blocked and state.get('kind')=='curve'
+  if activation!=self.activation:
+   self.activation=activation;self.payoff=None
+  target=1.
+  if valid and phase=='anticipation':
+   target=1.-.8*min(1.,max(0.,amount));self.payoff=None
+  elif valid and phase=='event':
+   if self.payoff is None:
+    step=self.rate*60/self.grid.bpm/2 if self.grid.usable else 0.
+    origin=self.grid.beat_phase*self.rate if self.grid.usable else 0.
+    self.payoff=round(origin+math.ceil((start_frame-origin)/step-1e-10)*step) if step else start_frame
+   if start_frame<self.payoff:
+    target=self.target
+    split=min(len(pcm),self.payoff-start_frame)
+    first=self.dsp.process(pcm[:split],False,self.config,target_mix=target)
+    if split<len(pcm):
+     import numpy as np
+     second=self.dsp.process(pcm[split:],False,self.config,target_mix=1.)
+     self.target=1.;self.phase=phase
+     return np.concatenate((first,second))
+  self.target=target;self.phase=phase
+  return self.dsp.process(pcm,False,self.config,target_mix=target)
+
+ def snapshot(self):
+  return {'curve_reaction':dict(enabled=self.enabled,input_fresh=self.fresh,phase=self.phase,
+          rendered_phase=('apex' if self.payoff is not None and self.payoff<=self.end_frame<self.payoff+round(.3*self.rate) else 'build' if self.dsp.mix<.999 else 'neutral'),
+          rendered_active=self.dsp.mix<.999,rendered_open_mix=self.dsp.mix,target_open_mix=self.target,cutoff_hz=1800.,maximum_wet_mix=.8,
+          payoff_audio_s=None if self.payoff is None else self.payoff/self.rate,attack_ms=130.,contain_ms=180.,
+          added_delay_samples=0,source='current causal conductor state',unknown_policy='smooth bypass')}

roadscore/prototype/review_curve_reaction.py

@@ -0,0 +1,61 @@
+"""Offline same-generation motion A/B from an immutable captured demo."""
+import argparse,hashlib,json,time
+from pathlib import Path
+import numpy as np
+import soundfile as sf
+from curve_reaction import CurveReaction
+from signal_shaker import ShakerGrid
+
+
+def sha(path):
+  h=hashlib.sha256()
+  with path.open('rb') as handle:
+    for chunk in iter(lambda:handle.read(1024*1024),b''):h.update(chunk)
+  return h.hexdigest()
+
+
+def render(locked,out):
+  source=locked/'render/heard.wav';tracepath=locked/'render/trace.jsonl';blockpath=locked/'render/audio_blocks.jsonl'
+  protected={str(p):sha(p) for p in (source,tracepath,blockpath)}
+  rows=[json.loads(line) for line in tracepath.read_text().splitlines()]
+  blocks=[json.loads(line) for line in blockpath.read_text().splitlines()]
+  out.mkdir(parents=True,exist_ok=False)
+  info=sf.info(source);rate=info.samplerate
+  if rate!=48000 or info.channels!=2:raise ValueError('Expected archived48k stereo PCM')
+  dsp=CurveReaction(ShakerGrid(128,0,0,0,False,"not observed"),rate,enabled=True);index=-1;times=[];states=[]
+  with sf.SoundFile(source) as src,sf.SoundFile(out/'full_curve.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
+      if src.tell()!=start:raise ValueError('Captured audio blocks are not contiguous')
+      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 {}
+      fresh=bool(row and 0<=block['callback_wall']-row['command_wall']<=.5)
+      open_mix=row.get('engagement_presentation',{}).get('rendered_open_mix',0.) if block.get('engagement_active') and block.get('engagement_fresh') else 0.
+      grid=row.get('signal_shaker',{}).get('grid')
+      if grid:dsp.grid=ShakerGrid(**grid)
+      before=time.perf_counter()
+      y=dsp.process(pcm,start,row,source_fresh=fresh)
+      times.append(time.perf_counter()-before);dst.write(y)
+      states.append(dict(audio_s=start/rate,route_t=row.get('route_t'),source_age_seconds=block['callback_wall']-row.get('command_wall',block['callback_wall']),signal_fresh=block.get('signal_fresh'),steering=row.get('steering'),speed=row.get('speed'),fresh=fresh,**dsp.snapshot()['curve_reaction']))
+  event=min(states,key=lambda x:x['rendered_open_mix'])
+  start=max(0.,min(info.duration-45,event['audio_s']-15));end=min(info.duration,start+45)
+  for name,path in [('A_baseline.wav',source),('B_curve.wav',out/'full_curve.wav')]:
+    with sf.SoundFile(path) as audio:
+      audio.seek(round(start*rate));pcm=audio.read(round((end-start)*rate),dtype='float32',always_2d=True)
+    sf.write(out/name,pcm,rate,subtype='FLOAT')
+  after={str(p):sha(p) for p in (source,tracepath,blockpath)}
+  if protected!=after:raise RuntimeError('Protected source changed during review')
+  report=dict(source_sha256=protected,source_unchanged=True,excerpt_start_audio_s=start,excerpt_end_audio_s=end,
+              selection='45 seconds around strongest causal curve containment; no manual event triggers',
+              cpu_host='Mac offline; not a device benchmark',dsp_median_ms=float(np.median(times)*1000),dsp_p95_ms=float(np.quantile(times,.95)*1000),dsp_max_ms=max(times)*1000,dsp_rtf=sum(times)/info.duration,
+              input_alignment='Latest recorded trace command_wall at or before callback_wall; bounded recorded conductor receipt age; no interpolation/future rows',
+              output='Post-process exact archived heard.wav; preserve existing engagement/cues, curve filter unity gain; existing engagement remains present',
+              physical_bluetooth_latency='Not measured; these files compare musical processing, not speaker/video synchronization',
+              settings=dsp.snapshot(),sample_frames=info.frames)
+  (out/'report.json').write_text(json.dumps(report,indent=2));(out/'curve_states.json').write_text(json.dumps(states))
+  (out/'index.html').write_text('''<!doctype html><meta name="viewport" content="width=device-width,initial-scale=1"><title>RoadScore curve A/B</title><style>body{background:#101016;color:#eee;font:18px system-ui;max-width:760px;margin:40px auto;padding:20px}audio{width:100%;margin:12px 0}p{line-height:1.5;color:#bbb}</style><h1>Curve build / payoff A/B</h1><p>Same accepted recording and same 45-second excerpt. B progressively contains brightness during the recorded curve build, then opens on an eighth-note boundary at the payoff. No regeneration, new melody, gain normalization or event samples.</p><h2>A · Locked baseline</h2><audio controls src="A_baseline.wav"></audio><h2>B · Curve build and payoff</h2><audio controls src="B_curve.wav"></audio><p>Default remains off. This is a musical comparison, not Bluetooth/video synchronization. Existing engagement processing remains intact. This cannot isolate a lead instrument from a stereo mix.</p>''')
+  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

@@ -17,13 +17,13 @@ class AccentTests(unittest.TestCase):
  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)
+  self.assertEqual(accent.events[0]['scheduled_frame'],1000)
  def test_tiny_changes_omitted_but_competing_cues_do_not_erase_warning(self):
   self.assertFalse(self.render(['a','b']*100)[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.assertGreaterEqual(accent.events[0]['frame'],round(.05*8000))
+  self.assertLess(accent.events[0]['scheduled_frame'],round(.4*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)])
@@ -36,6 +36,14 @@ 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_128bpm_next_eighth_has_no_competition_delay(self):
+  grid=ShakerGrid(128,.099,.8,.9,True,'test')
+  a=AlertAccent(grid,rate=48000,enabled=True);x=np.full((4800,2),.1,np.float32)
+  a.process(x,0,'warning',True,True,True)
+  a.process(x,4800,'warning',True,True,True)
+  self.assertEqual(len(a.events),1)
+  self.assertLessEqual(a.events[0]['scheduled_frame']/48000,.1+60/128/2+.0001)
+  self.assertTrue(a.priority_active)
  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):

roadscore/prototype/test_curve_reaction.py

@@ -0,0 +1,36 @@
+import unittest
+import numpy as np
+from curve_reaction import CurveReaction
+from signal_shaker import ShakerGrid
+GRID=ShakerGrid(128,0,.8,.9,True,'test')
+
+class CurveTests(unittest.TestCase):
+ def setUp(self):
+  t=np.arange(4800)/48000;self.pcm=np.column_stack([.1*np.sin(2*np.pi*6000*t)]*2).astype('float32')
+  self.state=dict(kind='curve',phase='anticipation',activation=1.,amount=.9)
+ def test_build_then_quantized_payoff_preserves_source(self):
+  dsp=CurveReaction(GRID,enabled=True);original=self.pcm.copy()
+  for i in range(15):y=dsp.process(self.pcm,i*4800,self.state)
+  self.assertLess(np.std(y),np.std(self.pcm)*.4)
+  state={**self.state,'phase':'event'}
+  dsp.process(self.pcm,72000,state)
+  self.assertGreaterEqual(dsp.payoff,72000);self.assertLessEqual(dsp.payoff-72000,48000*60/128/2)
+  for i in range(16,23):y=dsp.process(self.pcm,i*4800,state)
+  self.assertTrue(np.array_equal(y,self.pcm));self.assertTrue(np.array_equal(original,self.pcm))
+ def test_bypass_stale_and_priority(self):
+  dsp=CurveReaction(GRID,enabled=False)
+  self.assertIs(dsp.process(self.pcm,0,self.state),self.pcm)
+  dsp.enabled=True
+  for i in range(10):dsp.process(self.pcm,i*4800,self.state)
+  for i in range(10,15):y=dsp.process(self.pcm,i*4800,self.state,blocked=True)
+  self.assertTrue(np.array_equal(y,self.pcm))
+  for i in range(15,20):y=dsp.process(self.pcm,i*4800,self.state,source_fresh=False)
+  self.assertTrue(np.array_equal(y,self.pcm))
+ def test_uncertain_grid_has_filter_release_without_added_notes(self):
+  grid=ShakerGrid(128,0,0,0,False,'uncertain');dsp=CurveReaction(grid,enabled=True)
+  silence=np.zeros_like(self.pcm)
+  for i in range(10):self.assertFalse(np.any(dsp.process(silence,i*4800,self.state)))
+  dsp.process(silence,48000,{**self.state,'phase':'event'})
+  self.assertEqual(dsp.payoff,48000)
+
+if __name__=='__main__':unittest.main()