Start late apex breaths smoothly at unity gain

roadscore/prototype/core_apex.py

@@ -4,23 +4,24 @@ import numpy as np
 
 class CoreApex:
  def __init__(self, grid, rate=48000, enabled=False,dip_db=-1.):
-  self.dip_db=max(-3.,min(0.,float(dip_db)));self.grid=grid;self.rate=rate;self.enabled=enabled;self.last_activation=None;self.peak=None;self.events=[];self.rendered_active=False;self.rendered_end=0.
+  self.dip_db=max(-3.,min(0.,float(dip_db)));self.grid=grid;self.rate=rate;self.enabled=enabled;self.last_activation=None;self.peak=None;self.breath_start=None;self.events=[];self.rendered_active=False;self.rendered_end=0.
  def process(self, pcm, start_frame, state):
   self.rendered_active=False;self.rendered_end=(start_frame+len(pcm))/self.rate
-  if not self.enabled or not self.grid.usable:return pcm
+  if not self.enabled:return pcm
   activation=state.get('activation');lead=state.get('lead');now=start_frame/self.rate
-  if (state.get('kind')=='curve' and state.get('phase')=='anticipation' and activation is not None
+  if (self.grid.usable and state.get('kind')=='curve' and state.get('phase')=='anticipation' and activation is not None
       and activation!=self.last_activation and lead is not None and math.isfinite(lead) and .25<=lead<=1.5):
    self.last_activation=activation
    if (not self.events or now-self.events[-1]['trigger_seconds']>=8) and sum(now-e['trigger_seconds']<60 for e in self.events)<3:
     subdivision=60/self.grid.bpm/2
     target=self.grid.beat_phase+round((now+lead-self.grid.beat_phase)/subdivision)*subdivision
-    self.peak=round(target*self.rate);self.events.append({'trigger_seconds':now,'peak_seconds':target,'activation':activation,'dip_db':self.dip_db,'peak_gain':1.})
+    self.peak=round(target*self.rate);self.breath_start=max(start_frame,self.peak-round(.7*self.rate));self.events.append({'trigger_seconds':now,'peak_seconds':target,'activation':activation,'dip_db':self.dip_db,'peak_gain':1.})
   if self.peak is None:return pcm
-  offset=(np.arange(start_frame,start_frame+len(pcm))-self.peak)/self.rate
-  # Half-second breath, then original unity source at the estimated apex. Never boosts/clips the hook.
-  gain=np.ones(len(pcm),np.float32);inside=(offset>-.7)&(offset<0)
-  gain[inside]=1-(1-10**(self.dip_db/20))*np.sin(np.pi*(offset[inside]+.7)/.7)**2
+  frames=np.arange(start_frame,start_frame+len(pcm))
+  # A late prediction shortens the breath instead of jumping into a partly completed dip.
+  gain=np.ones(len(pcm),np.float32);inside=(frames>self.breath_start)&(frames<self.peak)
+  span=max(1,self.peak-self.breath_start)
+  gain[inside]=1-(1-10**(self.dip_db/20))*np.sin(np.pi*(frames[inside]-self.breath_start)/span)**2
   self.rendered_active=bool(inside.any() and np.any(pcm[inside]))
   return pcm if not self.rendered_active else pcm*gain[:,None]
  def snapshot(self):

roadscore/prototype/test_core_apex.py

@@ -0,0 +1,44 @@
+import unittest
+import numpy as np
+from core_apex import CoreApex
+from signal_shaker import ShakerGrid
+
+class ApexEnvelopeTests(unittest.TestCase):
+ def test_late_predictions_start_and_end_at_unity(self):
+  rate=8000;start=rate*10
+  grid=ShakerGrid(128,0,1,1,True,'test',True)
+  for lead in [.25,.35,.5,1.]:
+   with self.subTest(lead=lead):
+    c=CoreApex(grid,rate,True,-3.)
+    source=np.full((rate*2,2),.3,np.float32)
+    result=c.process(source,start,{'kind':'curve','phase':'anticipation','activation':1,'lead':lead})
+    self.assertEqual(result[0,0],source[0,0])
+    np.testing.assert_array_equal(result[c.peak-start:],source[c.peak-start:])
+    self.assertLess(float(result.min()),.3*.72)
+    self.assertGreaterEqual(float(result.min()),.3*10**(-3/20)-1e-6)
+    self.assertLess(float(np.max(np.abs(np.diff(result[:,0])))),.0002)
+    np.testing.assert_array_equal(source,np.full_like(source,.3))
+ def test_uncertain_grid_finishes_existing_breath_without_new_triggers(self):
+  rate=8000;good=ShakerGrid(128,0,1,1,True,'test',True)
+  c=CoreApex(good,rate,True,-3)
+  state={'kind':'curve','phase':'anticipation','activation':1,'lead':.25}
+  x=np.full((80,2),.3,np.float32)
+  c.process(x,rate*10,state)
+  c.grid=ShakerGrid(128,0,0,0,False,'uncertain')
+  heard=False
+  for i in range(1,100):
+   result=c.process(x,rate*10+i*80,dict(state,activation=2))
+   heard |= bool(np.any(result<x))
+  self.assertTrue(heard)
+  self.assertEqual(len(c.events),1)
+  np.testing.assert_array_equal(result,x)
+ def test_callback_partition_preserves_envelope(self):
+  rate=8000;grid=ShakerGrid(128,0,1,1,True,'test',True)
+  state={'kind':'curve','phase':'anticipation','activation':1,'lead':.25}
+  source=np.full((rate,2),.3,np.float32)
+  whole=CoreApex(grid,rate,True,-3).process(source,10*rate,state)
+  c=CoreApex(grid,rate,True,-3)
+  blocks=[c.process(source[i:i+80],10*rate+i,state) for i in range(0,rate,80)]
+  np.testing.assert_array_equal(np.concatenate(blocks),whole)
+
+if __name__=='__main__':unittest.main()