diff --git a/roadscore/prototype/signal_shaker.py b/roadscore/prototype/signal_shaker.py index c0baf57713..234852cb6f 100644 --- a/roadscore/prototype/signal_shaker.py +++ b/roadscore/prototype/signal_shaker.py @@ -39,8 +39,8 @@ def assess_grid(wave, rate, bpm_prior=None): return ShakerGrid(g['bpm'],g['beat_phase'],g['pulse_confidence'],phase,coherent,reason),{**g,'acoustic_agreement':agreement,'phase_drift_seconds':drift,'phase_confidence':phase} 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(.08,max(0.,float(peak))) + def __init__(self,grid,rate=48000,enabled=False,peak=.012,debounce_seconds=1.2,accented=False): + self.grid=grid;self.rate=rate;self.enabled=enabled;self.peak=min(.16,max(0.,float(peak)));self.accented=bool(accented) 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_key=None @@ -48,10 +48,11 @@ class SignalShaker: 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. - n=round(.085*rate);t=np.arange(n)/rate;rng=np.random.default_rng(1701) + n=round((.115 if self.accented else .085)*rate);t=np.arange(n)/rate;rng=np.random.default_rng(1701) noise=rng.standard_normal(n);freq=np.fft.rfftfreq(n,1/rate);spectrum=np.fft.rfft(noise) - spectrum*=np.clip((freq-2300)/1000,0,1)*np.clip((11500-freq)/3000,0,1) - grain=np.fft.irfft(spectrum,n);env=(1-np.exp(-t/.004))*np.exp(-t/.022) + low,high,decay=(1400,10500,.035) if self.accented else (2300,11500,.022) + spectrum*=np.clip((freq-low)/1000,0,1)*np.clip((high-freq)/3000,0,1) + grain=np.fft.irfft(spectrum,n);env=(1-np.exp(-t/.004))*np.exp(-t/decay) grain=grain*env;grain/=max(abs(grain).max(),1e-9) self.grain=np.column_stack((grain,grain)).astype(np.float32)*self.peak self.step=rate*60/grid.bpm/2 if grid.usable else 1.;self.origin=grid.beat_phase*rate @@ -85,7 +86,15 @@ class SignalShaker: while self.active and self.sequence_pulses<32 and round(self.origin+self.next_tick*self.step)=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,'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} + 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,'accented':self.accented,'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_signal_shaker.py b/roadscore/prototype/test_signal_shaker.py index f1b7c0b715..5ef8d59e13 100644 --- a/roadscore/prototype/test_signal_shaker.py +++ b/roadscore/prototype/test_signal_shaker.py @@ -76,4 +76,35 @@ class ShakerTests(unittest.TestCase): for i in range(100):s.process(x[:4800],i*4800,True,True) self.assertEqual(s.snapshot()['suppression_reason'],'sequence budget exhausted') + def test_accented_signal_is_directional_and_first_bar_is_strongest(self): + x=np.zeros((4800,2),np.float32) + samples=[];s=SignalShaker(GRID,enabled=True,peak=.14,accented=True) + for i in range(90):samples.append(s.process(x,i*4800,True,True,sequence_key='left')) + y=np.concatenate(samples) + self.assertAlmostEqual(float(abs(y).max()),.14,places=6) + np.testing.assert_allclose(y[:,1],y[:,0]*.65,atol=1e-8) + first=s.pulse_frames[0];n=len(s.grain) + later=s.pulse_frames[8] + np.testing.assert_allclose(y[later:later+n],y[first:first+n]*.65,atol=1e-8) + self.assertEqual(len(s.sequence_starts),1) + self.assertEqual(len(s.pulse_frames),32) + + def test_accented_signal_keeps_unknown_phase_and_headroom_guards(self): + unknown=ShakerGrid(128,0,.8,0,True,'unverified phase') + x=np.full((48000,2),.98,np.float32) + self.assertIs(SignalShaker(unknown,enabled=True,peak=.14,accented=True).process(x,0,True,True),x) + s=SignalShaker(GRID,enabled=True,peak=9,accented=True) + y=s.process(x,0,True,True,sequence_key='right') + self.assertEqual(s.peak,.16);self.assertLessEqual(float(y.max()),1.) + np.testing.assert_array_equal(x,np.full_like(x,.98)) + + def test_silenced_priority_is_reported_and_does_not_restart_sequence(self): + s=SignalShaker(GRID,enabled=True,peak=.14,accented=True) + x=np.zeros((4800,2),np.float32) + for i in range(40): + np.testing.assert_array_equal(s.process(x,i*4800,True,True,sequence_key='left',presentation_gain=0),x) + self.assertEqual(s.snapshot()['suppression_reason'],'presentation priority') + self.assertEqual(len(s.sequence_starts),1) + self.assertGreater(s.sequence_pulses,0) + if __name__=='__main__':unittest.main()