piccal/PolyNoteTester.py

import types
import time
from random import randrange

class PolyNoteTester:
    def __init__( self, cfg, api  ):
        self.api = api

        r = types.SimpleNamespace(**cfg.PolyNoteTester)
        self.cfg = r

        if r.mode == "simple":
            print("mode:simple")
            self.schedL = self._gen_simple_sched(r)
        else:
            print("mode:poly")
            self.schedL = self._gen_sched(r)

            
        self.schedL = sorted( self.schedL, key=lambda x: x[0] )    
        self.nextMs   = 0        # next transition time
        self.schedIdx = 0        # next event to play
        self.isStartedFl   = False

        #self._report()

    def _report( self ):
        for t,cmd,pitch,atkUs in self.schedL:
            print("%s %6i %3i %5i" % (cmd,t,pitch,atkUs))

    def _gen_simple_sched( self, r ):
        """ Play each note sequentially from lowest to highest. """
        durMs  = int(r.minNoteDurMs + (r.maxNoteDurMs - r.minNoteDurMs)/2)
        ioMs   = int(r.minInterOnsetMs + (r.maxInterOnsetMs - r.minInterOnsetMs)/2)
        atkUs  = int(r.minAttackUsec + (r.maxAttackUsec - r.minAttackUsec)/2)
        schedL = []

        t0 = 0
        for pitch in range(r.minPitch,r.maxPitch+1):
            schedL.append((t0,'on',pitch,atkUs))
            schedL.append((t0+durMs,'off',pitch,0))
            t0 += durMs + ioMs

        return schedL

        
    def _does_note_overlap( self, beg0Ms, end0Ms, beg1Ms, end1Ms ):
        """ if note 0 is entirely before or after note 1 """
        return not (beg0Ms > end1Ms or end0Ms < beg1Ms)

    def _do_any_notes_overlap( self, begMs, endMs, begEndL ):
        for beg1,end1 in begEndL:
            if self._does_note_overlap(begMs,endMs,beg1,end1):
                return True
        return False

    def _get_last_end_time( self, begEndL ):
        end0 = 0
        for beg,end in begEndL:
            if end > end0:
                end0 = end

        return end0
            
                    
    def _gen_sched( self, r ):

        pitchL = [ randrange(r.minPitch,r.maxPitch) for _ in range(r.noteCount) ]
        durMsL = [ randrange(r.minNoteDurMs, r.maxNoteDurMs) for _ in range(r.noteCount) ]
        ioMsL  = [ randrange(r.minInterOnsetMs, r.maxInterOnsetMs) for _ in range(r.noteCount) ]
        atkUsL = [ randrange(r.minAttackUsec,   r.maxAttackUsec)   for _ in range(r.noteCount) ]
        schedL = []
        pitchD = {} # pitch: [ (begMs,endMs) ]

        t0 = 0
        # for each pitch,dur,ioi,atkUs tuple
        for pitch,durMs,interOnsetMs,atkUs in zip(pitchL,durMsL,ioMsL,atkUsL):

            # calc note begin and end time
            begMs = t0
            endMs = t0 + durMs

            # if this pitch hasn't yet been added to pitchD
            if pitch not in pitchD:
                pitchD[ pitch ] = [ (begMs,endMs) ]
            else:

                # if the proposed note overlaps with other notes for this pitch
                if self._do_any_notes_overlap( begMs, endMs, pitchD[pitch] ):
                    # move this pitch past the last note
                    begMs = self._get_last_end_time( pitchD[pitch] ) + interOnsetMs
                    endMs = begMs + durMs

                # add the new note to pitchD
                pitchD[ pitch ].append( (begMs,endMs) )

            
            # update the schedule
            schedL.append( (begMs, 'on', pitch, atkUs)) 
            schedL.append( (endMs, 'off', pitch, 0 ))
            
            t0 += interOnsetMs

        
        return schedL
        
        
    def start( self ):
        self.schedIdx = 0
        self.nextMs   = 0
        self.api.set_hold_duty_all(self.cfg.holdDutyPct)
        self.isStartedFl = True
        

    def stop( self ):
        self.isStartedFl = False
        self.api.all_notes_off()
        

    def tick( self, ms ):

        while self.isStartedFl and ms >= self.nextMs and self.schedIdx < len(self.schedL):

            t0,cmd,pitch,usec = self.schedL[self.schedIdx]

            if cmd == 'on':
                if pitch not in self.cfg.skipPitchL:
                    decay_level = self.api.calc_decay_level( usec )
                    self.api.note_on_us( pitch, usec, decay_level )
                    print("on  %i %i %i" % (pitch,usec,decay_level))
                
            elif cmd == 'off':
                if pitch not in self.cfg.skipPitchL:
                    self.api.note_off( pitch )
                    print("off %i" % pitch)
            
            self.schedIdx += 1
            if self.schedIdx < len(self.schedL):
                self.nextMs = ms + (self.schedL[self.schedIdx][0] - t0)
            else:
                self.isStartedFl = False
                print("Done.")