Invalidate stale curve payoff labels and process waiting blocks once

roadscore/prototype/curve_reaction.py

@@ -8,20 +8,24 @@ class CurveReaction:
   self.grid=grid;self.rate=rate;self.enabled=enabled
   self.dsp=EngagementPresentation(rate,cutoff_hz=1800.,width=.92,gain=1.)
   self.config=PresentationConfig(enabled=True,attack_ms=130.,release_ms=180.)
-  self.activation=None;self.payoff=None;self.target=1.;self.phase='neutral';self.fresh=False;self.end_frame=0
+  self.activation=None;self.payoff=None;self.target=1.;self.phase='neutral';self.fresh=False;self.end_frame=0;self.valid=False;self.armed=False
 
  def process(self,pcm,start_frame,state,source_fresh=True,blocked=False):
   self.end_frame=start_frame+len(pcm)
   amount=state.get('amount',0.)
   self.fresh=bool(source_fresh and isinstance(amount,(int,float)) and math.isfinite(amount))
   phase=state.get('phase','neutral');activation=state.get('activation')
-  valid=self.enabled and self.fresh and not blocked and state.get('kind')=='curve'
+  identified=isinstance(activation,(int,float)) and not isinstance(activation,bool) and math.isfinite(activation)
+  valid=self.enabled and self.fresh and not blocked and state.get('kind')=='curve' and identified
+  self.valid=bool(valid)
   if activation!=self.activation:
-   self.activation=activation;self.payoff=None
+   self.activation=activation;self.payoff=None;self.armed=False
+  if not valid:
+   self.payoff=None;self.armed=False
   target=1.
   if valid and phase=='anticipation':
    target=1.-.8*min(1.,max(0.,amount));self.payoff=None
-  elif valid and phase=='event':
+  elif valid and phase=='event' and self.armed:
    if self.payoff is None:
     step=self.rate*60/self.grid.bpm/2 if self.grid.usable else 0.
     origin=self.grid.beat_phase*self.rate if self.grid.usable else 0.
@@ -35,12 +39,18 @@ class CurveReaction:
      second=self.dsp.process(pcm[split:],False,self.config,target_mix=1.)
      self.target=1.;self.phase=phase
      return np.concatenate((first,second))
+    self.target=target;self.phase=phase
+    return first
   self.target=target;self.phase=phase
-  return self.dsp.process(pcm,False,self.config,target_mix=target)
+  result=self.dsp.process(pcm,False,self.config,target_mix=target)
+  if valid and phase=='anticipation' and self.dsp.mix<.999:self.armed=True
+  return result
 
  def snapshot(self):
+  apex=bool(self.valid and self.phase=='event' and self.armed and self.payoff is not None and self.payoff<self.end_frame<self.payoff+round(.3*self.rate))
+  build=bool(self.valid and self.armed and self.phase in ('anticipation','event') and self.dsp.mix<.999)
   return {'curve_reaction':dict(enabled=self.enabled,input_fresh=self.fresh,phase=self.phase,
-          rendered_phase=('apex' if self.payoff is not None and self.payoff<=self.end_frame<self.payoff+round(.3*self.rate) else 'build' if self.dsp.mix<.999 else 'neutral'),
-          rendered_active=self.dsp.mix<.999,rendered_open_mix=self.dsp.mix,target_open_mix=self.target,cutoff_hz=1800.,maximum_wet_mix=.8,
+          rendered_phase=('apex' if apex else 'build' if build else 'neutral'),
+          rendered_active=apex or build,rendered_open_mix=self.dsp.mix,target_open_mix=self.target,cutoff_hz=1800.,maximum_wet_mix=.8,
           payoff_audio_s=None if self.payoff is None else self.payoff/self.rate,attack_ms=130.,contain_ms=180.,
           added_delay_samples=0,source='current causal conductor state',unknown_policy='smooth bypass')}

roadscore/prototype/test_curve_reaction.py

@@ -26,6 +26,35 @@ class CurveTests(unittest.TestCase):
   self.assertTrue(np.array_equal(y,self.pcm))
   for i in range(15,20):y=dsp.process(self.pcm,i*4800,self.state,source_fresh=False)
   self.assertTrue(np.array_equal(y,self.pcm))
+ def test_stale_or_blocked_payoff_cannot_show_apex(self):
+  for kwargs in ({'blocked':True},{'source_fresh':False}):
+   dsp=CurveReaction(GRID,enabled=True)
+   for i in range(10):dsp.process(self.pcm,i*4800,self.state)
+   event={**self.state,'phase':'event'}
+   dsp.process(self.pcm,48000,event)
+   dsp.process(self.pcm,52800,event,**kwargs)
+   view=dsp.snapshot()['curve_reaction']
+   self.assertEqual(view['rendered_phase'],'neutral');self.assertFalse(view['rendered_active'])
+   self.assertIsNone(view['payoff_audio_s'])
+   dsp.process(self.pcm,57600,event)
+   self.assertEqual(dsp.snapshot()['curve_reaction']['rendered_phase'],'neutral')
+ def test_missing_activation_and_unbuilt_event_never_claim_payoff(self):
+  for activation in (None,float('nan'),True):
+   dsp=CurveReaction(GRID,enabled=True)
+   for i in range(5):y=dsp.process(self.pcm,i*4800,{**self.state,'activation':activation})
+   self.assertTrue(np.array_equal(y,self.pcm))
+   self.assertEqual(dsp.snapshot()['curve_reaction']['rendered_phase'],'neutral')
+  dsp=CurveReaction(GRID,enabled=True)
+  dsp.process(self.pcm,0,{**self.state,'phase':'event'})
+  self.assertIsNone(dsp.payoff)
+ def test_quantized_wait_processes_each_sample_once(self):
+  dsp=CurveReaction(GRID,enabled=True)
+  dsp.process(self.pcm,0,self.state)
+  before=dsp.dsp.frames_processed
+  # Next eighth is11250; this entire100ms block precedes that boundary.
+  dsp.process(self.pcm,4800,{**self.state,'phase':'event'})
+  self.assertEqual(dsp.dsp.frames_processed-before,len(self.pcm))
+  self.assertGreater(dsp.payoff,9600)
  def test_uncertain_grid_has_filter_release_without_added_notes(self):
   grid=ShakerGrid(128,0,0,0,False,'uncertain');dsp=CurveReaction(grid,enabled=True)
   silence=np.zeros_like(self.pcm)