Make replay signal cues audible and strengthen engagement contrast

roadscore/prototype/app.py

@@ -90,7 +90,10 @@ alert_enabled=config.get('alert_accent',{}).get('enabled') is True
 curve_enabled=config.get('curve_reaction',{}).get('enabled') is True
 road_model_valid=False
 if (motion_enabled or presentation_config.enabled or shaker_enabled or apex_enabled or alert_enabled or curve_enabled) and render_mode!='gold-core':raise ValueError('Optional presentation layers require gold-core baseline')
-presentation=EngagementPresentation(rate,block) if presentation_config.enabled else None
+presentation=EngagementPresentation(rate,block,cutoff_hz=presentation_config.cutoff_hz,
+ width=presentation_config.width,gain=presentation_config.gain) if presentation_config.enabled else None
+shaker_after_containment=config.get('signal_shaker',{}).get('after_containment') is True
+shaker_contained_gain=float(np.clip(config.get('signal_shaker',{}).get('contained_gain',1.),0.,1.))
 motion_state=(0.,False,0,0.,0)
 engagement=(False,False,0,0.,0);signal_state=(False,False,0,0.,0);alert_state=(None,False)
 initial_identity=identity;musical_mode=config.get('musical',False);arrival=Arrival();ending_start=None;ending_audio=None;ending_info={};playing_identity=identity;identity_queue=[];last_requested_identity=identity
@@ -225,7 +228,7 @@ def callback(out,n,ti,status):
   competing=bool((shaker is not None and shaker.active) or (apex is not None and apex.rendered_active) or (presentation is not None and abs(presentation.mix-float(active))>.01))
   rendered=alert_accent.process(rendered,frames-n,alert_key,meaningful,fresh,competing)
   alert_priority=alert_accent.priority_active
- if shaker is not None:
+ if shaker is not None and not shaker_after_containment:
   shaken=shaker.process(rendered,frames-n,signal_on,signal_fresh)
   if alert_priority:shaker.rendered_active=False;shaker.rendered_peak=0.
   else:rendered=shaken
@@ -241,6 +244,13 @@ def callback(out,n,ti,status):
   _,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)
  if presentation is not None:rendered=presentation.process(rendered,active,presentation_config)
+ if shaker is not None and shaker_after_containment:
+  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
+  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
  out[:]=0 if a.mute else rendered
  cue={**presentation_narrative,**{key:event_state[key] for key in PRESENTATION_FIELDS if key in event_state}}
  cue['replay_demo']=controls_snapshot(demo_mode,demo_signal_mode)

roadscore/prototype/engagement_presentation.py

@@ -16,6 +16,9 @@ class PresentationConfig:
   enabled: bool = False
   attack_ms: float = 220.
   release_ms: float = 650.
+  cutoff_hz: float = 4500.
+  width: float = .85
+  gain: float = .8912509
 
   @classmethod
   def read(cls, value):
@@ -27,8 +30,16 @@ class PresentationConfig:
         return min(5000., max(20., n)) if math.isfinite(n) else default
       except (TypeError, ValueError):
         return default
+    def bounded(key, default, low, high):
+      try:
+        n = float(value.get(key, default))
+        return min(high, max(low, n)) if math.isfinite(n) else default
+      except (TypeError, ValueError):
+        return default
     return cls(enabled=value.get('enabled') is True,
-               attack_ms=duration('attack_ms', 220.), release_ms=duration('release_ms', 650.))
+               attack_ms=duration('attack_ms', 220.), release_ms=duration('release_ms', 650.),
+               cutoff_hz=bounded('cutoff_hz', 4500., 300., 12000.),
+               width=bounded('width', .85, 0., 1.), gain=bounded('gain', .8912509, .25, 1.))
 
 
 def engagement_active(valid, active, source_ns, latest_source_ns, received_wall, now_wall):
@@ -47,7 +58,7 @@ class EngagementPresentation:
     self.rate = rate
     self.frames_processed = 0
     self.max_frames = max_frames
-    self.width=float(width);self.gain=float(gain)
+    self.cutoff_hz=float(cutoff_hz);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
@@ -87,7 +98,7 @@ class EngagementPresentation:
       np.add(ramp, self.mix, out=ramp)
       np.clip(ramp, min(self.mix, target), max(self.mix, target), out=ramp)
       self.mix = float(ramp[-1])
-      # Contained: 85% stereo width and -1 dB. Convex mixing avoids gain boosts.
+      # Contained settings are fixed at startup. Convex mixing avoids gain boosts.
       mid = self.mid[:n]
       np.add(low[:, 0], low[:, 1], out=mid)
       mid *= .5
@@ -110,4 +121,5 @@ class EngagementPresentation:
             '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',
+            'cutoff_hz': self.cutoff_hz, 'width': self.width, 'gain': self.gain,
             'unknown_policy': 'contained', 'added_delay_samples': 0}}

roadscore/prototype/presentation_policy.py

@@ -2,7 +2,7 @@
 import copy
 import os
 
-POLICIES = ('conservative-v1', 'conservative-v2', 'conservative-v3', 'off', 'frozen')
+POLICIES = ('conservative-v1', 'conservative-v2', 'conservative-v3', 'conservative-v4', 'off', 'frozen')
 CONSERVATIVE = {
  'presentation_policy_version': 'conservative-v1',
  'signal_shaker': {'enabled': True},
@@ -14,6 +14,13 @@ CONSERVATIVE_V2 = {**copy.deepcopy(CONSERVATIVE), 'presentation_policy_version':
 
 CONSERVATIVE_V3 = {**copy.deepcopy(CONSERVATIVE_V2), 'presentation_policy_version':'conservative-v3', 'stopped_motion':{'enabled':True}, 'curve_reaction':{'enabled':True}, 'signal_shaker':{'enabled':True,'peak':.018}}
 
+CONSERVATIVE_V4 = {**copy.deepcopy(CONSERVATIVE_V3),
+ 'presentation_policy_version':'conservative-v4',
+ 'signal_shaker':{'enabled':True,'peak':.055,'after_containment':True,'contained_gain':.45},
+ 'engagement_presentation':{'version':2,'enabled':True,'attack_ms':220,'release_ms':650,
+                          'cutoff_hz':1300.,'width':.65,'gain':.7079458},
+}
+
 def select_launch(composer, profile, replay=False, judging=False, render_mode=None, policy=None):
  if policy is not None and policy not in POLICIES:raise ValueError('Unknown presentation policy')
  if render_mode is not None and render_mode not in ('current','gold-core'):raise ValueError('Unknown rendering mode')
@@ -29,8 +36,8 @@ def select_launch(composer, profile, replay=False, judging=False, render_mode=No
   return {'render_mode':'current','policy':'frozen'}
  normal_prism=composer=='ace' and profile=='prism'
  mode=render_mode or ('gold-core' if normal_prism else 'current')
- chosen=policy or ('conservative-v3' if normal_prism and mode=='gold-core' else 'off')
- if chosen in ('conservative-v1','conservative-v2','conservative-v3') and (composer!='ace' or mode!='gold-core'):
+ chosen=policy or ('conservative-v4' if normal_prism and mode=='gold-core' else 'off')
+ if chosen in ('conservative-v1','conservative-v2','conservative-v3','conservative-v4') and (composer!='ace' or mode!='gold-core'):
   raise ValueError('Conservative presentation requires ACE gold-core rendering')
  return {'render_mode':mode,'policy':chosen}
 
@@ -43,8 +50,8 @@ def effective_config(config,environ=None):
  """Return a copy; never mutate the global runtime file or an archived policy."""
  result=copy.deepcopy(config);policy=selected(environ)
  if policy=='frozen':return result
- if policy in ('conservative-v1','conservative-v2','conservative-v3'):
-  result.update(copy.deepcopy(CONSERVATIVE_V3 if policy=='conservative-v3' else CONSERVATIVE_V2 if policy=='conservative-v2' else CONSERVATIVE))
+ if policy in ('conservative-v1','conservative-v2','conservative-v3','conservative-v4'):
+  result.update(copy.deepcopy(CONSERVATIVE_V4 if policy=='conservative-v4' else CONSERVATIVE_V3 if policy=='conservative-v3' else CONSERVATIVE_V2 if policy=='conservative-v2' else CONSERVATIVE))
   if policy=='conservative-v1':result['alert_accent']={'enabled':False}
  else:
   result.update(presentation_policy_version='off',stopped_motion={'enabled':False},curve_reaction={'enabled':False},alert_accent={'enabled':False},signal_shaker={'enabled':False},core_apex={'enabled':False},engagement_presentation={'version':2,'enabled':False})

roadscore/prototype/review_signal_engagement.py

@@ -0,0 +1,76 @@
+"""Render the same PCM through two presentation policies; no playback or inference."""
+import argparse
+from dataclasses import asdict
+import hashlib
+import json
+from pathlib import Path
+import time
+
+import numpy as np
+import soundfile as sf
+
+from engagement_presentation import EngagementPresentation, PresentationConfig
+from presentation_policy import effective_config
+from rhythm_timeline import RhythmTimeline
+from signal_shaker import SignalShaker
+
+
+def render(wave, rate, policy):
+  config=effective_config({}, {'ROADSCORE_PRESENTATION_POLICY':policy})
+  settings=PresentationConfig.read(config['engagement_presentation'])
+  dsp=EngagementPresentation(rate,cutoff_hz=settings.cutoff_hz,width=settings.width,gain=settings.gain)
+  timeline=RhythmTimeline(rate,128.);timeline.add(wave,0)
+  shaker=SignalShaker(timeline.at(0),rate,True,peak=config['signal_shaker']['peak'])
+  after=config['signal_shaker'].get('after_containment',False)
+  blocks=[];timings=[];snapshots=[]
+  for frame in range(0,len(wave),4800):
+    t=frame/rate;x=wave[frame:frame+4800]
+    active=not (4<=t<8 or 16<=t<20)
+    direction='left' if 10<=t<14 else 'right' if 14<=t<18 else 'off'
+    started=time.perf_counter()
+    shaker.set_grid(timeline.at(frame),frame)
+    if not after:x=shaker.process(x,frame,direction!='off',True)
+    output=dsp.process(x,active,settings)
+    if after:
+      contained=config['signal_shaker']['contained_gain']
+      output=shaker.process(output,frame,direction!='off',True,sequence_key=direction,
+                            presentation_gain=contained+(1-contained)*dsp.mix)
+    timings.append(time.perf_counter()-started)
+    blocks.append(output)
+    snapshots.append(dict(seconds=t,active=active,direction=direction,shaker=shaker.snapshot()))
+  return np.concatenate(blocks),dict(policy=policy,config=config,grid=timeline.snapshot(),
+    sequence_events=shaker.events,pulse_frames=shaker.pulse_frames,
+    block_work_mean_ms=float(np.mean(timings)*1000),block_work_max_ms=max(timings)*1000,
+    frames_unchanged=True,added_timing_delay_samples=0,snapshots=snapshots)
+
+
+def main():
+  parser=argparse.ArgumentParser(description=__doc__)
+  parser.add_argument('source',type=Path);parser.add_argument('output',type=Path)
+  args=parser.parse_args();args.output.mkdir(parents=True,exist_ok=False)
+  wave,rate=sf.read(args.source,dtype='float32',always_2d=True)
+  if rate!=48000 or wave.shape[1]!=2 or len(wave)!=26*rate:
+    raise ValueError('Use the exact 26-second 48kHz stereo native baseline')
+  report=dict(source=str(args.source),source_sha256=hashlib.sha256(args.source.read_bytes()).hexdigest(),
+    source_preserved=True,common_gain='Exact source gain; no normalization, resampling or limiter',
+    event_source='Synthetic manual-controls rehearsal, not route telemetry; moving throughout',
+    timeline=[dict(seconds=0,event='Engaged/open'),dict(seconds=4,event='Simulate disengage'),
+      dict(seconds=8,event='Simulate engage'),dict(seconds=10,event='Left signal'),
+      dict(seconds=14,event='Right signal'),dict(seconds=16,event='Simulate disengage'),
+      dict(seconds=18,event='Signals off; natural release'),dict(seconds=20,event='Simulate engage')],
+    scope='Signal and engagement comparison only; curves/alerts/motion not active in this fixture',
+    hardware_validated=False,files={})
+  for filename,policy in [('before-v3.wav','conservative-v3'),('after-v4.wav','conservative-v4')]:
+    audio,details=render(wave,rate,policy)
+    target=args.output/filename;sf.write(target,audio,rate,subtype='FLOAT')
+    report['files'][filename]=dict(sha256=hashlib.sha256(target.read_bytes()).hexdigest(),
+      seconds=len(audio)/rate,peak=float(abs(audio).max()),rms=float(np.sqrt(np.mean(audio*audio))),
+      samples_above_full_scale=int(np.count_nonzero(abs(audio)>1)),**details)
+  sf.write(args.output/'unchanged-core.wav',wave,rate,subtype='FLOAT')
+  report['core_sample_identical']=bool(np.array_equal(sf.read(args.output/'unchanged-core.wav',dtype='float32')[0],wave))
+  (args.output/'comparison.json').write_text(json.dumps(report,indent=2)+'\n')
+  summary={name:{key:value[key] for key in ('seconds','peak','rms','samples_above_full_scale','block_work_mean_ms','block_work_max_ms','sequence_events')} for name,value in report['files'].items()}
+  print(json.dumps(summary,indent=2))
+
+
+if __name__=='__main__':main()

roadscore/prototype/signal_shaker.py

@@ -40,10 +40,11 @@ def assess_grid(wave, rate, bpm_prior=None):
 
 class SignalShaker:
  def __init__(self,grid,rate=48000,enabled=False,peak=.012,debounce_seconds=1.2):
-  self.grid=grid;self.rate=rate;self.enabled=enabled;self.peak=min(.02,max(0.,float(peak)))
+  self.grid=grid;self.rate=rate;self.enabled=enabled;self.peak=min(.08,max(0.,float(peak)))
   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.sequence_pulses=0;self.sequence_start=0;self.sequence_key=None
+  self.presentation_gain=1.;self.suppression_reason='not rendered'
   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.
@@ -60,16 +61,23 @@ class SignalShaker:
   if self.active and grid.usable:
    earliest=max(start_frame,(self.pulse_frames[-1]+self.step*.5) if self.pulse_frames else start_frame)
    self.next_tick=math.ceil((earliest-self.origin)/self.step-1e-10)
- def process(self,pcm,start_frame,signal_on,signal_fresh):
+ def process(self,pcm,start_frame,signal_on,signal_fresh,*,sequence_key=None,presentation_gain=1.):
   self.rendered_active=False;self.rendered_peak=0.;self.rendered_start=start_frame;self.rendered_end=start_frame+len(pcm)
+  self.presentation_gain=float(np.clip(presentation_gain,0.,1.)) if math.isfinite(presentation_gain) else 0.
+  self.suppression_reason='disabled'
   if not self.enabled:return pcm
+  key=sequence_key if sequence_key in ('left','right') else None
+  changed=key is not None and self.sequence_key is not None and key!=self.sequence_key
+  self.sequence_key=key
   end=start_frame+len(pcm)
   if self.grid.usable and signal_fresh and signal_on:
    self.last_on=start_frame
-   if not self.active:
+   # Only a deliberate opposite-direction command may restart an active motif.
+   # One bar cooldown prevents rapid button presses from adding percussion spam.
+   if not self.active or (changed and start_frame-self.sequence_start>=8*self.step):
     self.active=True;self.stop_frame=None;self.sequence_pulses=0;self.sequence_start=start_frame
     tick=math.ceil((start_frame-self.origin)/self.step-1e-10);self.next_tick=tick
-    self.sequence_starts.append(start_frame);self.events.append({'kind':'sequence_start','frame':start_frame,'next_pulse_frame':round(self.origin+tick*self.step)})
+    self.sequence_starts.append(start_frame);self.events.append({'kind':'sequence_start','frame':start_frame,'direction':key,'next_pulse_frame':round(self.origin+tick*self.step)})
   if self.active and (not self.grid.usable or not signal_fresh or (self.last_on is not None and start_frame-self.last_on>self.debounce)):
    self.active=False;self.stop_frame=start_frame;self.events.append({'kind':'release','frame':start_frame})
   overlay=np.zeros((len(pcm),2),np.float32)
@@ -85,10 +93,16 @@ 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]
+  overlay*=self.presentation_gain
   # Add percussion only within available sample headroom; preserve the source.
   np.clip(overlay,-np.maximum(0.,1.+pcm),np.maximum(0.,1.-pcm),out=overlay)
   self.rendered_peak=float(np.max(np.abs(overlay),initial=0));self.rendered_active=self.rendered_peak>0
+  self.suppression_reason=('rendered' if self.rendered_active else
+    'uncertain beat grid' if not self.grid.usable else
+    'stale signal source' if not signal_fresh else
+    'sequence budget exhausted' if self.active and self.sequence_pulses>=32 else
+    'waiting for next eighth note' if self.active else 'no signal sequence')
   if not self.rendered_active:return pcm
   return pcm+overlay
  def snapshot(self):
-  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}
+  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,'sequence_pulses':self.sequence_pulses,'remaining_pulses':max(0,32-self.sequence_pulses),'suppression_reason':self.suppression_reason,'presentation_gain':self.presentation_gain,'direction':self.sequence_key}

roadscore/prototype/test_engagement_presentation.py

@@ -74,4 +74,20 @@ class PresentationTests(unittest.TestCase):
     self.assertEqual(PresentationConfig.read({'attack_ms':0}).attack_ms,20)
     self.assertEqual(PresentationConfig.read({'release_ms':float('nan')}).release_ms,650)
 
+  def test_v4_containment_is_stronger_and_still_recovers_exact_core(self):
+    from presentation_policy import CONSERVATIVE_V4
+    config=PresentationConfig.read(CONSERVATIVE_V4['engagement_presentation'])
+    old=EngagementPresentation();old.mix=0
+    new=EngagementPresentation(cutoff_hz=config.cutoff_hz,width=config.width,gain=config.gain);new.mix=0
+    t=np.arange(48000)/48000
+    x=np.column_stack([np.sin(t*2*np.pi*6000),-np.sin(t*2*np.pi*6000)]).astype('float32')*.4
+    before=old.process(x,False,ON);after=new.process(x,False,config)
+    self.assertLess(np.sqrt(np.mean(after[4800:]**2)),np.sqrt(np.mean(before[4800:]**2))*.12)
+    resumed=new.process(x,True,config)
+    np.testing.assert_array_equal(resumed[12000:],x[12000:])
+    self.assertEqual(new.snapshot(config,True,True)['engagement_presentation']['cutoff_hz'],1300.)
+    self.assertEqual(PresentationConfig.read({'width':float('nan'),'gain':9,'cutoff_hz':0}).width,.85)
+    self.assertEqual(PresentationConfig.read({'gain':9}).gain,1.)
+    self.assertEqual(PresentationConfig.read({'cutoff_hz':0}).cutoff_hz,300.)
+
 if __name__=='__main__':unittest.main()

roadscore/prototype/test_presentation_policy.py

@@ -7,7 +7,7 @@ class PolicyTests(unittest.TestCase):
   fragment=Path(__file__).resolve().parents[1]/'tools/presentation_conservative_v1.json'
   self.assertEqual(json.loads(fragment.read_text()),CONSERVATIVE)
  def test_normal_prism_gets_integrated_policy_without_extra_flags(self):
-  self.assertEqual(select_launch('ace','prism'),{'render_mode':'gold-core','policy':'conservative-v3'})
+  self.assertEqual(select_launch('ace','prism'),{'render_mode':'gold-core','policy':'conservative-v4'})
  def test_explicit_optouts(self):
   self.assertEqual(select_launch('ace','prism',policy='off'),{'render_mode':'gold-core','policy':'off'})
   self.assertEqual(select_launch('ace','prism',render_mode='current'),{'render_mode':'current','policy':'off'})
@@ -35,7 +35,7 @@ class PolicyTests(unittest.TestCase):
   self.assertTrue(latest['alert_accent']['enabled'])
   self.assertEqual(latest['core_apex']['dip_db'],-3.)
  def test_stopped_motion_candidate_preserves_default_and_optout(self):
-  self.assertEqual(select_launch('ace','prism')['policy'],'conservative-v3')
+  self.assertEqual(select_launch('ace','prism')['policy'],'conservative-v4')
   candidate=effective_config({}, {'ROADSCORE_PRESENTATION_POLICY':'conservative-v3'})
   self.assertTrue(candidate['stopped_motion']['enabled'])
   self.assertTrue(candidate['engagement_presentation']['enabled'])
@@ -45,4 +45,15 @@ class PolicyTests(unittest.TestCase):
   self.assertEqual(select_launch('sa3','prism'),{'render_mode':'current','policy':'off'})
   with self.assertRaises(ValueError):select_launch('ace','prism',render_mode='current',policy='conservative-v1')
   with self.assertRaises(ValueError):effective_config({}, {'ROADSCORE_PRESENTATION_POLICY':'typo'})
+ def test_v4_strengthens_only_presentation_and_preserves_previous_policy(self):
+  previous=effective_config({}, {'ROADSCORE_PRESENTATION_POLICY':'conservative-v3'})
+  current=effective_config({}, {'ROADSCORE_PRESENTATION_POLICY':'conservative-v4'})
+  self.assertEqual(previous['signal_shaker']['peak'],.018)
+  self.assertNotIn('cutoff_hz',previous['engagement_presentation'])
+  self.assertEqual(current['signal_shaker']['peak'],.055)
+  self.assertTrue(current['signal_shaker']['after_containment'])
+  self.assertEqual(current['engagement_presentation']['cutoff_hz'],1300.)
+  self.assertEqual(select_launch('ace','prism',policy='conservative-v3')['policy'],'conservative-v3')
+  with self.assertRaises(ValueError):select_launch('ace','prism',judging=True,policy='conservative-v4')
+
 if __name__=='__main__':unittest.main()

roadscore/prototype/test_signal_shaker.py

@@ -46,4 +46,34 @@ class ShakerTests(unittest.TestCase):
    state={'kind':'curve','phase':'anticipation','lead':.5,'activation':i//10}
    y=c.process(x,i*4800,state);self.assertTrue(np.all(y<=x));self.assertTrue(np.all(y>=x*10**(-1/20)-1e-6))
   self.assertLessEqual(len(c.events),3);np.testing.assert_array_equal(x,before)
+ def test_deliberate_opposite_direction_restarts_once_not_each_callback(self):
+  s=SignalShaker(GRID,enabled=True,peak=.055);x=np.zeros((4800,2),np.float32)
+  for i in range(140):
+   direction='left' if i<90 else 'right'
+   s.process(x,i*4800,True,True,sequence_key=direction)
+  self.assertEqual(len(s.sequence_starts),2)
+  self.assertEqual([e['direction'] for e in s.events if e['kind']=='sequence_start'],['left','right'])
+  self.assertGreater(len(s.pulse_frames),32)
+  self.assertEqual(s.snapshot()['remaining_pulses'],32-s.sequence_pulses)
+ def test_rapid_direction_changes_do_not_restart_and_stale_input_stays_silent(self):
+  s=SignalShaker(GRID,enabled=True);x=np.zeros((4800,2),np.float32)
+  for i in range(18):s.process(x,i*4800,True,True,sequence_key='left' if i%2 else 'right')
+  self.assertEqual(len(s.sequence_starts),1)
+  for i in range(18,28):s.process(x,i*4800,True,False,sequence_key='left' if i%2 else 'right')
+  self.assertFalse(s.rendered_active);self.assertFalse(s.active)
+  self.assertEqual(s.snapshot()['suppression_reason'],'stale signal source')
+ def test_stronger_cue_scales_and_respects_final_output_headroom(self):
+  old=SignalShaker(GRID,enabled=True,peak=.018);new=SignalShaker(GRID,enabled=True,peak=.055)
+  x=np.zeros((48000,2),np.float32)
+  before=old.process(x,0,True,True);after=new.process(x,0,True,True)
+  np.testing.assert_allclose(after,before*(.055/.018),atol=1e-8)
+  contained=SignalShaker(GRID,enabled=True,peak=.055).process(x,0,True,True,presentation_gain=.45)
+  np.testing.assert_allclose(contained,after*.45,atol=1e-8)
+  for value in (.99,-.99):
+   y=SignalShaker(GRID,enabled=True,peak=.055).process(np.full_like(x,value),0,True,True)
+   self.assertLessEqual(float(abs(y).max()),1.)
+  s=SignalShaker(GRID,enabled=True)
+  for i in range(100):s.process(x[:4800],i*4800,True,True)
+  self.assertEqual(s.snapshot()['suppression_reason'],'sequence budget exhausted')
+
 if __name__=='__main__':unittest.main()