Add opt-in beat-locked curve build cut and drop presentation

roadscore/prototype/curve_build_drop.py

@@ -0,0 +1,119 @@
+"""Opt-in presentation for one explicitly scheduled replay curve; no route inference."""
+import math
+
+import numpy as np
+from scipy.signal import butter, sosfilt
+
+from signal_shaker import ShakerGrid, SignalShaker
+
+
+class CurveBuildDrop:
+  def __init__(self, rate=48000, max_frames=4800, *, cut_ms=120., ramp_ms=10., low_boost_db=1.5):
+    self.rate=rate;self.max_frames=max_frames
+    self.cut_frames=round(cut_ms*rate/1000);self.ramp_frames=max(1,round(ramp_ms*rate/1000))
+    self.cancel_frames=max(1,round(.03*rate));self.boost_frames=round(.22*rate)
+    self.low_boost=10**(min(3.,max(0.,low_boost_db))/20)-1.
+    # Reuse the signal shaker's unpitched voice; RNG/FFT/filter design happen once.
+    prepared=SignalShaker(ShakerGrid(128.,0.,1.,1.,True,'prepared voice only'),rate,peak=.08)
+    self.grain=prepared.grain/.08
+    self.sos=butter(2,160.,fs=rate,output='sos').astype(np.float32)
+    self.zi=np.zeros((len(self.sos),2,2),np.float32)
+    sosfilt(self.sos,np.zeros((1,2),np.float32),axis=0,zi=self.zi)
+    self.indices=np.arange(max_frames,dtype=np.float64)
+    self.schedule=None;self.last_pulse=None;self.tail=np.zeros((0,2),np.float32)
+    self.strength=0.;self.actual_payoff=None;self.pulse_frames=[]
+    self.status=dict(rendered_stage='bypass',rendered_active=False,added_delay_samples=0)
+
+  def process(self, pcm, start_frame, grid, build_start_frame, payoff_frame, *, enabled=False,
+              source_fresh=True, blocked=False):
+    """Same frames, no input mutation. Caller supplies and authorizes the schedule.
+
+    Disabling a fully bypassed instance is bit exact; cancelling an active effect
+    returns smoothly to the original PCM within 30ms. Uncertain beat grids never
+    start a roll. Nothing in this class reads future route state or calls I/O.
+    """
+    if not len(pcm):return pcm
+    finite=lambda n:isinstance(n,(int,float)) and not isinstance(n,bool) and math.isfinite(n)
+    valid=bool(enabled and source_fresh and not blocked and grid.usable
+               and finite(build_start_frame) and finite(payoff_frame)
+               and payoff_frame-build_start_frame>self.cut_frames+self.ramp_frames)
+    candidate=(round(build_start_frame),round(payoff_frame)) if valid else None
+    # Keep the old schedule only long enough to finish a cancellation ramp.
+    if candidate is not None and candidate!=self.schedule:
+      self.schedule=candidate;self.last_pulse=None;self.actual_payoff=None;self.tail=np.zeros((0,2),np.float32)
+    outputs=[];peaks=[];cut_min=1.;any_active=False;stage='bypass'
+    for offset in range(0,len(pcm),self.max_frames):
+      x=pcm[offset:offset+self.max_frames];n=len(x);first=start_frame+offset;end=first+n
+      low,self.zi=sosfilt(self.sos,x,axis=0,zi=self.zi)
+      if self.schedule is None:
+        outputs.append(x);continue
+      build,payoff=self.schedule;cut_start=payoff-self.cut_frames
+      requested=valid and end>build and first<payoff+self.boost_frames
+      target=1. if requested else 0.
+      strength=np.clip(self.strength+(self.indices[:n]+1)*(1. if target>self.strength else -1.)/self.cancel_frames,
+                       min(self.strength,target),max(self.strength,target)).astype(np.float32)
+      self.strength=float(strength[-1])
+      if not np.any(strength):
+        self.tail=np.zeros((0,2),np.float32);outputs.append(x);continue
+      frames=self.indices[:n]+first
+      roll=np.zeros_like(x);take=min(n,len(self.tail));roll[:take]+=self.tail[:take];self.tail=self.tail[take:].copy()
+      if requested and first<cut_start and end>build:
+        beat=self.rate*60/grid.bpm;origin=grid.beat_phase*self.rate
+        # Eighth notes establish the motif; the final four beats use sixteenths.
+        boundary=max(build,payoff-4*beat)
+        for lo,hi,step in ((build,boundary,beat/2),(boundary,cut_start,beat/4)):
+          lo=max(first,lo);hi=min(end,hi)
+          if hi<=lo:continue
+          tick=math.ceil((lo-origin)/step-1e-10)
+          while round(origin+tick*step)<hi:
+            frame=round(origin+tick*step);tick+=1
+            if frame<first or (self.last_pulse is not None and frame-self.last_pulse<beat/8):continue
+            progress=np.clip((frame-build)/max(1.,cut_start-build),0.,1.)
+            gain=.024+.056*progress**1.3
+            grain=self.grain*gain;index=frame-first;count=min(len(grain),n-index)
+            roll[index:index+count]+=grain[:count]
+            if count<len(grain):
+              rest=grain[count:];tail=np.zeros((max(len(rest),len(self.tail)),2),np.float32)
+              tail[:len(self.tail)]+=self.tail;tail[:len(rest)]+=rest;self.tail=tail
+            self.last_pulse=frame;self.pulse_frames.append(frame)
+      cut=np.ones(n,np.float32)
+      falling=(frames>=cut_start)&(frames<payoff)
+      cut[falling]=np.clip(1.-(frames[falling]-cut_start+1)/self.ramp_frames,0.,1.)
+      rising=(frames>=payoff)&(frames<payoff+self.ramp_frames)
+      cut[rising]=np.clip((frames[rising]-payoff+1)/self.ramp_frames,0.,1.)
+      elapsed=frames-payoff
+      boost=np.where((elapsed>=0)&(elapsed<self.boost_frames),
+                     np.clip(elapsed/self.ramp_frames,0.,1.)*np.exp(-np.maximum(elapsed,0)/(self.rate*.075)),0.).astype(np.float32)
+      delta=roll+low*(boost*self.low_boost)[:,None]
+      np.clip(delta,-np.maximum(0.,1.+x),np.maximum(0.,1.-x),out=delta)
+      wet=(x+delta)*cut[:,None]
+      out=x+(wet-x)*strength[:,None]
+      # Exact neutral endpoint avoids floating-point cancellation after the drop.
+      neutral=(cut==1)&(boost==0)&(np.max(abs(roll),axis=1)==0)
+      out[neutral]=x[neutral]
+      effective_cut=1.-strength+strength*cut
+      rendered_roll=roll*(strength*cut)[:,None]
+      outputs.append(out);peaks.append(float(np.max(abs(rendered_roll),initial=0)))
+      cut_min=min(cut_min,float(np.min(effective_cut)))
+      rendered=bool(np.any(effective_cut<1.) or np.any(rendered_roll) or np.any(boost*strength*self.low_boost))
+      any_active |= rendered
+      if valid and first<=payoff<end and strength[int(payoff-first)]>0:self.actual_payoff=payoff
+      if not rendered:continue
+      if not valid:stage='cancelling'
+      elif end<=cut_start:stage='build'
+      elif first<payoff:stage='cut' if end<=payoff else 'drop'
+      else:stage='drop'
+    if not any_active:
+      if self.strength==0 and not valid:self.schedule=None
+      result=pcm
+    else:result=np.concatenate(outputs) if len(outputs)>1 else outputs[0]
+    self.status=dict(enabled=bool(enabled),input_fresh=bool(source_fresh),blocked=bool(blocked),grid_usable=bool(grid.usable),
+      rendered_stage=stage,rendered_active=any_active,rendered_strength=self.strength,rendered_roll_peak=max(peaks,default=0.),
+      rendered_cut_gain_min=cut_min,planned_build_frame=None if self.schedule is None else self.schedule[0],
+      planned_payoff_frame=None if self.schedule is None else self.schedule[1],actual_payoff_frame=self.actual_payoff,
+      pulse_count=len(self.pulse_frames),cut_ms=1000*self.cut_frames/self.rate,ramp_ms=1000*self.ramp_frames/self.rate,
+      low_boost_db=20*math.log10(1+self.low_boost),added_delay_samples=0,
+      source='explicit replay presentation schedule; no composition or route lookup')
+    return result
+
+  def snapshot(self):return dict(self.status)

roadscore/prototype/review_curve_build_drop.py

@@ -0,0 +1,56 @@
+"""Offline same-PCM build/cut/drop audition from an archived rhythm assessment."""
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import time
+
+import numpy as np
+import soundfile as sf
+
+from curve_build_drop import CurveBuildDrop
+from signal_shaker import ShakerGrid
+
+
+def main():
+  p=argparse.ArgumentParser(description=__doc__)
+  p.add_argument('source',type=Path);p.add_argument('output',type=Path)
+  p.add_argument('--timeline',type=Path)
+  p.add_argument('--build-start',type=float,required=True);p.add_argument('--apex',type=float,required=True)
+  p.add_argument('--start',type=float);p.add_argument('--end',type=float)
+  a=p.parse_args();a.output.mkdir(parents=True,exist_ok=False)
+  wave,rate=sf.read(a.source,dtype='float32',always_2d=True)
+  timeline_path=a.timeline or a.source.with_name('rhythm_timeline.json')
+  timeline=json.loads(timeline_path.read_text());apex=round(a.apex*rate)
+  entry=next(e for e in reversed(timeline) if e['start_frame']<=apex<e['end_frame'])
+  grid=ShakerGrid(**entry['grid'])
+  if not grid.usable:raise ValueError('The archived apex grid is not usable; no guessed pulse')
+  step=rate*60/grid.bpm/2;origin=grid.beat_phase*rate
+  payoff=round(origin+round((apex-origin)/step)*step)
+  build=round(a.build_start*rate)
+  first=max(0,round((a.start if a.start is not None else a.build_start-2)*rate))
+  end=min(len(wave),round((a.end if a.end is not None else a.apex+3)*rate))
+  fx=CurveBuildDrop(rate);blocks=[];timings=[]
+  for frame in range(0,end,4800):
+    x=wave[frame:min(frame+4800,end)];started=time.perf_counter()
+    blocks.append(fx.process(x,frame,grid,build,payoff,enabled=True))
+    timings.append(time.perf_counter()-started)
+  after=np.concatenate(blocks)[first:end];before=wave[first:end]
+  report=dict(source=str(a.source),source_sha256=hashlib.sha256(a.source.read_bytes()).hexdigest(),
+    rhythm_source=str(timeline_path),rhythm_sha256=hashlib.sha256(timeline_path.read_bytes()).hexdigest(),
+    grid=entry['grid'],build_audio_s=build/rate,requested_apex_audio_s=a.apex,actual_payoff_audio_s=payoff/rate,
+    apex_quantization_offset_ms=1000*(payoff/rate-a.apex),clip_start_audio_s=first/rate,
+    clip_end_audio_s=end/rate,clip_build_s=(build-first)/rate,clip_drop_s=(payoff-first)/rate,
+    scope='Explicit staged presentation from supplied real-event timing; no generation or runtime changes. Component only; root bass-removal build is not included.',
+    common_gain='Source unchanged; no normalization, resampling or source attenuation',
+    hardware_validated=False,work_mean_ms=float(np.mean(timings)*1000),work_max_ms=max(timings)*1000,
+    pulse_frames=fx.pulse_frames,status=fx.snapshot(),files={})
+  for name,audio in (('before-core.wav',before),('after-build-drop.wav',after)):
+    target=a.output/name;sf.write(target,audio,rate,subtype='FLOAT')
+    report['files'][name]=dict(seconds=len(audio)/rate,peak=float(abs(audio).max()),
+      samples_above_full_scale=int(np.count_nonzero(abs(audio)>1)),sha256=hashlib.sha256(target.read_bytes()).hexdigest())
+  (a.output/'comparison.json').write_text(json.dumps(report,indent=2)+'\n')
+  print(json.dumps({key:report[key] for key in ('clip_build_s','clip_drop_s','apex_quantization_offset_ms','work_mean_ms','work_max_ms','files')},indent=2))
+
+
+if __name__=='__main__':main()

roadscore/prototype/test_curve_build_drop.py

@@ -0,0 +1,67 @@
+import unittest
+
+import numpy as np
+
+from curve_build_drop import CurveBuildDrop
+from signal_shaker import ShakerGrid
+
+GRID=ShakerGrid(120.,0.,.9,.9,True,'test grid')
+
+
+class BuildDropTests(unittest.TestCase):
+  def test_disabled_uncertain_and_blocked_are_bit_exact(self):
+    x=np.random.default_rng(2).normal(0,.1,(4800,2)).astype(np.float32)
+    unknown=ShakerGrid(120.,0.,0.,0.,False,'uncertain')
+    for enabled,fresh,blocked,grid in ((False,True,False,GRID),(True,False,False,GRID),
+                                       (True,True,True,GRID),(True,True,False,unknown)):
+      before=x.copy();fx=CurveBuildDrop()
+      y=fx.process(x,0,grid,0,48000,enabled=enabled,source_fresh=fresh,blocked=blocked)
+      self.assertIs(y,x);np.testing.assert_array_equal(x,before)
+
+  def test_grid_aligned_crescendo_and_exact_cut_payoff(self):
+    fx=CurveBuildDrop(low_boost_db=0);x=np.full((4*48000,2),.2,np.float32)
+    y=fx.process(x,0,GRID,0,3*48000,enabled=True)
+    self.assertEqual(y.shape,x.shape);self.assertTrue(np.isfinite(y).all())
+    self.assertTrue(np.all(y[round(2.89*48000):3*48000]==0))
+    self.assertGreater(y[3*48000,0],0)
+    np.testing.assert_array_equal(y[3*48000+480:],x[3*48000+480:])
+    self.assertEqual(fx.snapshot()['actual_payoff_frame'],3*48000)
+    self.assertEqual(fx.snapshot()['added_delay_samples'],0)
+    for frame in fx.pulse_frames:
+      self.assertEqual(frame%6000,0)
+    self.assertGreater(max(abs(y[:2*48000]-x[:2*48000]).ravel()),.02)
+    # The final mute and return use 480-sample ramps, not a discontinuity.
+    smooth=CurveBuildDrop(low_boost_db=0);smooth.grain.fill(0)
+    ramped=smooth.process(x,0,GRID,0,3*48000,enabled=True)
+    self.assertLess(np.max(abs(np.diff(ramped[round(2.88*48000):round(3.02*48000),0]))),.001)
+
+  def test_cancellation_during_cut_restores_core_without_hanging_mute(self):
+    for cancel in ({'source_fresh':False},{'blocked':True},{'enabled':False}):
+      fx=CurveBuildDrop(low_boost_db=0)
+      fx.process(np.full((round(2.95*48000),2),.2,np.float32),0,GRID,0,3*48000,enabled=True)
+      x=np.full((4800,2),.2,np.float32)
+      y=fx.process(x,round(2.95*48000),GRID,0,3*48000,**({'enabled':True}|cancel))
+      self.assertGreater(y[0,0],0);self.assertLess(y[0,0],.001)
+      self.assertLess(np.max(abs(np.diff(y[:,0]))),.001)
+      np.testing.assert_array_equal(y[round(.031*48000):],x[round(.031*48000):])
+      self.assertIsNone(fx.snapshot()['actual_payoff_frame'])
+      self.assertEqual(fx.strength,0.)
+
+  def test_headroom_and_source_ownership(self):
+    fx=CurveBuildDrop();t=np.arange(4*48000)/48000
+    x=(.97*np.sin(2*np.pi*80*t))[:,None]*np.ones((1,2),np.float32);x=x.astype(np.float32)
+    before=x.copy();y=fx.process(x,0,GRID,0,3*48000,enabled=True)
+    self.assertLessEqual(float(abs(y).max()),1.)
+    np.testing.assert_array_equal(x,before)
+
+  def test_irregular_callback_blocks_match_whole_render(self):
+    x=np.random.default_rng(7).normal(0,.1,(4*48000,2)).astype(np.float32)
+    whole=CurveBuildDrop().process(x,0,GRID,0,3*48000,enabled=True)
+    fx=CurveBuildDrop();blocks=[];start=0;sizes=[37,800,13,9701,4800]
+    while start<len(x):
+      n=min(sizes[len(blocks)%len(sizes)],len(x)-start)
+      blocks.append(fx.process(x[start:start+n],start,GRID,0,3*48000,enabled=True));start+=n
+    np.testing.assert_allclose(np.concatenate(blocks),whole,atol=3e-7)
+
+
+if __name__=='__main__':unittest.main()