From 85cd88056b09e9889b7fee31b4b61b19cd0aecc7 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Sun, 20 Sep 2026 10:01:23 -0700 Subject: [PATCH] Make replay signal cues audible and strengthen engagement contrast --- roadscore/prototype/app.py | 14 +++- .../prototype/engagement_presentation.py | 18 ++++- roadscore/prototype/presentation_policy.py | 17 +++-- .../prototype/review_signal_engagement.py | 76 +++++++++++++++++++ roadscore/prototype/signal_shaker.py | 26 +++++-- .../prototype/test_engagement_presentation.py | 16 ++++ .../prototype/test_presentation_policy.py | 15 +++- roadscore/prototype/test_signal_shaker.py | 30 ++++++++ 8 files changed, 194 insertions(+), 18 deletions(-) create mode 100644 roadscore/prototype/review_signal_engagement.py diff --git a/roadscore/prototype/app.py b/roadscore/prototype/app.py index 254254f542..76bdcfd390 100644 --- a/roadscore/prototype/app.py +++ b/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) diff --git a/roadscore/prototype/engagement_presentation.py b/roadscore/prototype/engagement_presentation.py index 0b26680499..cb849a48aa 100644 --- a/roadscore/prototype/engagement_presentation.py +++ b/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}} diff --git a/roadscore/prototype/presentation_policy.py b/roadscore/prototype/presentation_policy.py index ec067dd942..fb43459e9b 100644 --- a/roadscore/prototype/presentation_policy.py +++ b/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}) diff --git a/roadscore/prototype/review_signal_engagement.py b/roadscore/prototype/review_signal_engagement.py new file mode 100644 index 0000000000..47827d0148 --- /dev/null +++ b/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() diff --git a/roadscore/prototype/signal_shaker.py b/roadscore/prototype/signal_shaker.py index 1b2329e9bf..c0baf57713 100644 --- a/roadscore/prototype/signal_shaker.py +++ b/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} diff --git a/roadscore/prototype/test_engagement_presentation.py b/roadscore/prototype/test_engagement_presentation.py index bba3336f64..659f184a6a 100644 --- a/roadscore/prototype/test_engagement_presentation.py +++ b/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() diff --git a/roadscore/prototype/test_presentation_policy.py b/roadscore/prototype/test_presentation_policy.py index 51f27b5777..9797ec8436 100644 --- a/roadscore/prototype/test_presentation_policy.py +++ b/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() diff --git a/roadscore/prototype/test_signal_shaker.py b/roadscore/prototype/test_signal_shaker.py index ac75fd1626..f1b7c0b715 100644 --- a/roadscore/prototype/test_signal_shaker.py +++ b/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()