Add optional causal stopped-to-moving low-pass lift

roadscore/prototype/engagement_presentation.py

@@ -43,11 +43,12 @@ def engagement_active(valid, active, source_ns, latest_source_ns, received_wall,
 
 
 class EngagementPresentation:
-  def __init__(self, rate=48000, max_frames=4800):
+  def __init__(self, rate=48000, max_frames=4800, *, cutoff_hz=4500., width=.85, gain=.8912509):
     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.width=float(width);self.gain=float(gain)
+    self.sos = butter(2, cutoff_hz, fs=rate, output='sos').astype(np.float32)
     self.zi = np.zeros((len(self.sos), 2, 2), np.float32)
     self.mix = 1.  # startup bypass; opting in ramps into contained presentation
     self.indices = np.arange(1, max_frames + 1, dtype=np.float32)
@@ -58,7 +59,7 @@ class EngagementPresentation:
     # Exercise scipy dispatch before audio callbacks.
     sosfilt(self.sos, np.zeros((1, 2), np.float32), axis=0, zi=self.zi)
 
-  def process(self, pcm, active, config):
+  def process(self, pcm, active, config, *, target_mix=None):
     """Stereo float32 in/out, same frames, never mutates caller PCM.
 
     Fully bypassed output is original PCM (read-only use by downstream). Split
@@ -69,6 +70,9 @@ class EngagementPresentation:
     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.
+    if config.enabled and target_mix is not None:
+      if not math.isfinite(target_mix) or not 0<=target_mix<=1:raise ValueError("Invalid presentation mix")
+      target=float(target_mix)
     output = None if target == 1. and self.mix == 1. else np.empty_like(pcm)
     for start in range(0, len(pcm), self.max_frames):
       x = pcm[start:start + self.max_frames]
@@ -89,9 +93,9 @@ class EngagementPresentation:
       mid *= .5
       wet = self.scratch[:n]
       np.subtract(low, mid[:, None], out=wet)
-      wet *= .85
+      wet *= self.width
       wet += mid[:, None]
-      wet *= .8912509
+      wet *= self.gain
       delta = self.delta[:n]
       np.subtract(x, wet, out=delta)
       np.multiply(delta, ramp[:, None], out=delta)

roadscore/prototype/motion_presentation.py

@@ -0,0 +1,34 @@
+"""Causal stopped-to-moving brightness lift; output only, no musical replanning."""
+import math
+from engagement_presentation import EngagementPresentation, PresentationConfig
+
+
+class MotionPresentation:
+  def __init__(self, rate=48000, enabled=False, lift_ms=500., contain_ms=700.):
+    self.rate=rate;self.enabled=enabled;self.stopped=False;self.stop_seconds=0.;self.move_seconds=0.
+    self.dsp=EngagementPresentation(rate,cutoff_hz=900.,width=1.,gain=1.)
+    self.config=PresentationConfig.read(dict(enabled=True,attack_ms=lift_ms,release_ms=contain_ms))
+    self.target=1.;self.fresh=False
+
+  def process(self, pcm, *, speed, source_fresh, engagement_open_mix=1.):
+    dt=len(pcm)/self.rate
+    self.fresh=bool(source_fresh and math.isfinite(speed) and speed>=0.)
+    if self.fresh:
+      self.stop_seconds=self.stop_seconds+dt if speed<=.25 else 0.
+      self.move_seconds=self.move_seconds+dt if speed>=.8 else 0.
+      if self.stop_seconds>=.4:self.stopped=True
+      if self.move_seconds>=.15:self.stopped=False
+    else:
+      self.stopped=False;self.stop_seconds=0.;self.move_seconds=0.
+    # Moving bypass restores the downstream engagement state, never bypasses it.
+    strength=1. if self.fresh and self.stopped and self.enabled else 0.
+    self.target=1.-strength
+    return self.dsp.process(pcm,False,self.config,target_mix=self.target)
+
+  def snapshot(self):
+    return {'motion_presentation':dict(enabled=self.enabled,stopped=self.stopped,input_fresh=self.fresh,
+                target_open_mix=self.target,rendered_open_mix=self.dsp.mix,cutoff_hz=900.,
+                lift_ms=self.config.attack_ms,contain_ms=self.config.release_ms,
+                stop_threshold_mps=.25,move_threshold_mps=.8,stop_dwell_seconds=.4,move_dwell_seconds=.15,
+                gain=1.,width=1.,added_delay_samples=0,unknown_policy='smooth bypass',
+                source='fresh current carState.vEgo; no future data')}

roadscore/prototype/review_motion_presentation.py

@@ -0,0 +1,59 @@
+"""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 motion_presentation import MotionPresentation
+
+
+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=MotionPresentation(rate,enabled=True);index=-1;times=[];states=[]
+  with sf.SoundFile(source) as src,sf.SoundFile(out/'full_motion.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(block.get('signal_fresh') and 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.
+      before=time.perf_counter()
+      y=dsp.process(pcm,speed=row.get('speed',0.),source_fresh=fresh,engagement_open_mix=open_mix)
+      times.append(time.perf_counter()-before);dst.write(y)
+      states.append(dict(audio_s=start/rate,steering=row.get('steering'),speed=row.get('speed'),fresh=fresh,**dsp.snapshot()['motion_presentation']))
+  transitions=[state for previous,state in zip(states,states[1:]) if previous['stopped'] and not state['stopped'] and state['fresh'] and state.get('speed',0)>=.8]
+  event=transitions[0] if transitions else 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_motion.wav',out/'full_motion.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 first causal stopped-to-moving transition; 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; recorded carState freshness; no interpolation/future rows',
+              output='Post-process exact archived heard.wav; preserve existing engagement/cues, motion filter has unity gain/width; 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/'motion_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 motion 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 presentation A/B</h1><p>Same accepted recording and same 45-second excerpt. B uses a strong 900 Hz low-pass while stopped and opens as the car starts moving. No regeneration, new melody, gain normalization or event samples.</p><h2>A · Locked baseline</h2><audio controls src="A_baseline.wav"></audio><h2>B · Optional motion containment</h2><audio controls src="B_motion.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_motion_presentation.py

@@ -0,0 +1,53 @@
+import unittest
+import numpy as np
+from motion_presentation import MotionPresentation
+from engagement_presentation import EngagementPresentation,PresentationConfig
+
+
+class MotionTests(unittest.TestCase):
+  def setUp(self):
+    self.rate=48000;t=np.arange(4800)/self.rate
+    self.pcm=np.column_stack([.15*np.sin(2*np.pi*6000*t)]*2).astype('float32')
+
+  def run_blocks(self,dsp,n=12,**kw):
+    for _ in range(n):out=dsp.process(self.pcm,**kw)
+    return out
+
+  def test_disabled_unity_and_stopped_strong_lift(self):
+    dsp=MotionPresentation(enabled=False);original=self.pcm.copy()
+    self.assertTrue(np.array_equal(self.run_blocks(dsp,speed=0.,source_fresh=True),self.pcm))
+    dsp.enabled=True;contained=self.run_blocks(dsp,speed=0.,source_fresh=True)
+    self.assertLess(np.std(contained)/np.std(self.pcm),.05)
+    lifted=self.run_blocks(dsp,speed=1.,source_fresh=True)
+    self.assertTrue(np.array_equal(lifted,self.pcm));self.assertTrue(np.array_equal(original,self.pcm))
+
+  def test_hysteresis_dwell_and_stale_input(self):
+    dsp=MotionPresentation(enabled=True)
+    self.run_blocks(dsp,3,speed=0.,source_fresh=True);self.assertFalse(dsp.stopped)
+    self.run_blocks(dsp,1,speed=0.,source_fresh=True);self.assertTrue(dsp.stopped)
+    for speed in (.3,.7,.4,.6):self.run_blocks(dsp,2,speed=speed,source_fresh=True);self.assertTrue(dsp.stopped)
+    self.run_blocks(dsp,1,speed=1.,source_fresh=True);self.assertTrue(dsp.stopped)
+    self.run_blocks(dsp,1,speed=1.,source_fresh=True);self.assertFalse(dsp.stopped)
+    self.run_blocks(dsp,speed=0.,source_fresh=True)
+    y=self.run_blocks(dsp,speed=float('nan'),source_fresh=False)
+    self.assertTrue(np.array_equal(y,self.pcm));self.assertFalse(dsp.stopped)
+
+  def test_engagement_owns_inactive_presentation(self):
+    dsp=MotionPresentation(enabled=True);eng=EngagementPresentation();config=PresentationConfig(enabled=True)
+    y=self.run_blocks(dsp,speed=0.,source_fresh=True,engagement_open_mix=0.)
+    self.assertLess(np.std(y),np.std(self.pcm)*.05)
+    y=self.run_blocks(dsp,speed=1.,source_fresh=True,engagement_open_mix=0.)
+    self.assertTrue(np.array_equal(y,self.pcm))
+    # Moving bypass cannot undo the existing engagement lowpass.
+    for _ in range(10):out=eng.process(y,False,config)
+    self.assertLess(np.std(out),np.std(y)*.7)
+
+  def test_ramps_have_no_step_on_dc_and_disable_releases(self):
+    dsp=MotionPresentation(enabled=True);x=np.full((4800,2),.1,np.float32)
+    blocks=[]
+    for i in range(35):blocks.append(dsp.process(x,speed=0. if i<18 else 1.,source_fresh=True))
+    self.assertLess(np.max(np.abs(np.diff(np.concatenate(blocks)[:,0]))),.0001)
+    self.run_blocks(dsp,speed=0.,source_fresh=True);dsp.enabled=False
+    self.assertTrue(np.array_equal(self.run_blocks(dsp,speed=0.,source_fresh=True),self.pcm))
+
+if __name__=='__main__':unittest.main()