piccal/p_ac.py

##| Copyright: (C) 2019-2020 Kevin Larke <contact AT larke DOT org> 
##| License: GNU GPL version 3.0 or above. See the accompanying LICENSE file.
import sys,os,argparse,types,logging,select,time,json
from datetime import datetime

import multiprocessing
from multiprocessing import Process, Pipe

from picadae_api  import Picadae
from AudioDevice  import AudioDevice
from MidiDevice   import MidiDevice
from result       import Result
from common       import parse_yaml_cfg
from plot_seq     import form_resample_pulse_time_list
from plot_seq_1   import get_resample_points_wrap
from plot_seq     import form_final_pulse_list
from rt_note_analysis import RT_Analyzer
from keyboard         import Keyboard
from calibrate        import Calibrate
from rms_analysis import rms_analyze_one_rt_note_wrap
from MidiFilePlayer import MidiFilePlayer

class AttackPulseSeq:
    """ Sequence a fixed pitch over a list of attack pulse lengths."""
    
    def __init__(self, cfg, audio, api, noteDurMs=1000, pauseDurMs=1000 ):
        self.cfg         = cfg
        self.audio       = audio
        self.api         = api
        self.outDir      = None           # directory to write audio file and results
        self.pitch       = None           # pitch to paly
        self.pulseUsL    = []             # one onset pulse length in microseconds per sequence element
        self.noteDurMs   = noteDurMs      # duration of each chord in milliseconds
        self.pauseDurMs  = pauseDurMs     # duration between end of previous note and start of next
        self.holdDutyPctL= None           # hold voltage duty cycle table [ (minPulseSeqUsec,dutyCyclePct) ]
        self.holdDutyPctD= None          # { us:dutyPct } for each us in self.pulseUsL
        self.silentNoteN = None 
        self.pulse_idx            = 0     # Index of next pulse 
        self.state                = None  # 'note_on','note_off'
        self.prevHoldDutyPct      = None
        self.next_ms              = 0     # Time of next event (note-on or note_off)
        self.eventTimeL           = []    # Onset/offset time of each note [ [onset_ms,offset_ms] ] (used to locate the note in the audio file)
        self.beginMs              = 0
        self.playOnlyFl           = False
        self.rtAnalyzer           = RT_Analyzer()

    def start( self, ms, outDir, pitch, pulseUsL, holdDutyPctL, holdDutyPctD, playOnlyFl=False ):
        self.outDir     = outDir         # directory to write audio file and results
        self.pitch     = pitch           # note to play
        self.pulseUsL   = pulseUsL       # one onset pulse length in microseconds per sequence element
        self.holdDutyPctL = holdDutyPctL
        self.holdDutyPctD = holdDutyPctD
        self.silentNoteN = 0 
        self.pulse_idx  = 0
        self.state      = 'note_on'
        self.prevHoldDutyPct = None
        self.next_ms    = ms + 500       # wait for 500ms to play the first note (this will guarantee that there is some empty space in the audio file before the first note)
        self.eventTimeL = [[0,0]  for _ in range(len(pulseUsL))] # initialize the event time         
        self.beginMs    = ms
        self.playOnlyFl = playOnlyFl


        # kpl if not playOnlyFl:
        self.audio.record_enable(True)   # start recording audio
        
        self.tick(ms)                    # play the first note
        
    def stop(self, ms):
        self._send_note_off() # be sure that all notes are actually turn-off
        
        # kpl if not self.playOnlyFl:
        self.audio.record_enable(False)  # stop recording audio
            
        self._disable()          # disable this sequencer
        
        if not self.playOnlyFl:
            self._write()            # write the results

    def is_enabled(self):
        return self.state is not None
    
    def tick(self, ms):

        # if next event time has arrived
        if self.is_enabled() and  ms >= self.next_ms:

            # if waiting to turn note on
            if self.state == 'note_on':
                self._note_on(ms)
        
            # if waiting to turn a note off
            elif self.state == 'note_off':
                self._note_off(ms)
                self._count_silent_notes()
                    
                    
                self.pulse_idx += 1
                
                # if all notes have been played
                if self.pulse_idx >= len(self.pulseUsL): # or self.silentNoteN >= self.cfg.maxSilentNoteCount:
                    self.stop(ms)
                    
            else:                
                assert(0)
                    
    def _count_silent_notes( self ):
        annBegMs = self.eventTimeL[ self.pulse_idx ][0]
        annEndMs = self.eventTimeL[ self.pulse_idx ][1]
        minDurMs = self.cfg.silentNoteMinDurMs
        maxPulseUs = self.cfg.silentNoteMaxPulseUs
        
        resD = rms_analyze_one_rt_note_wrap( self.audio, annBegMs, annEndMs, self.pitch, self.pauseDurMs, self.cfg.analysisArgs['rmsAnalysisArgs'] )

        print( "         %4.1f db  %4i ms %i" %  (resD['hm']['db'], resD['hm']['durMs'], self.pulse_idx))

        if resD is not None and resD['hm']['durMs'] < minDurMs and self.pulseUsL[ self.pulse_idx ] < maxPulseUs:
            self.silentNoteN += 1
            print("SILENT", self.silentNoteN)
        else:
            self.silentNoteN = 0

        return self.silentNoteN
        
    def _get_duty_cycle( self, pulseUsec ):
        return self.holdDutyPctD[ pulseUsec ]
    
        dutyPct = self.holdDutyPctL[0][1]
        for refUsec,refDuty in self.holdDutyPctL:
            if pulseUsec < refUsec:
                break
            dutyPct = refDuty

        return dutyPct
            
    def _set_duty_cycle( self, pitch, pulseUsec ):

        dutyPct = self._get_duty_cycle( pulseUsec )

        if dutyPct != self.prevHoldDutyPct:
            self.api.set_pwm_duty( pitch, dutyPct )
            print("Hold Duty:",dutyPct)
            
        self.prevHoldDutyPct = dutyPct
    
    def _note_on( self, ms ):

        self.eventTimeL[ self.pulse_idx ][0] = self.audio.buffer_sample_ms().value
        self.next_ms = ms + self.noteDurMs
        self.state = 'note_off'

        pulse_usec = int(self.pulseUsL[ self.pulse_idx ])
        self._set_duty_cycle( self.pitch, pulse_usec )
        self.api.note_on_us( self.pitch, pulse_usec )
        print("note-on:",self.pitch, self.pulse_idx, pulse_usec)

    def _note_off( self, ms ):
        self.eventTimeL[ self.pulse_idx ][1] = self.audio.buffer_sample_ms().value
        self.next_ms = ms + self.pauseDurMs
        self.state   = 'note_on'

        if self.playOnlyFl:
            begTimeMs = self.eventTimeL[ self.pulse_idx ][0]
            endTimeMs = self.eventTimeL[ self.pulse_idx ][1]
            self.rtAnalyzer.analyze_note( self.audio, self.pitch, begTimeMs, endTimeMs, self.cfg.analysisArgs['rmsAnalysisArgs'] )
        
        self._send_note_off()        

    
    def _send_note_off( self ):
        self.api.note_off( self.pitch )
        #print("note-off:",self.pitch,self.pulse_idx)

    def _disable(self):
        self.state = None
        
            
    def _write( self ):

        d = {
            "pulseUsL":self.pulseUsL,
            "pitch":self.pitch,
            "noteDurMs":self.noteDurMs,
            "pauseDurMs":self.pauseDurMs,
            "holdDutyPctL":self.holdDutyPctL,
            "eventTimeL":self.eventTimeL,
            "beginMs":self.beginMs
            }

        print("Writing: ", self.outDir )

        outDir = os.path.expanduser(self.outDir)

        if  not os.path.isdir(outDir):
            os.mkdir(outDir)
        
        with open(os.path.join( outDir, "seq.json" ),"w") as f:
            f.write(json.dumps( d ))

        self.audio.write_buffer( os.path.join( outDir, "audio.wav" ) )
        
            
class CalibrateKeys:
    def __init__(self, cfg, audioDev, api):
        self.cfg      = cfg
        self.seq      = AttackPulseSeq(  cfg, audioDev, api, noteDurMs=cfg.noteDurMs, pauseDurMs=cfg.pauseDurMs )
        
        self.pulseUsL  = None
        self.pitchL   = None
        self.pitch_idx = -1

        
    def start( self, ms, pitchL, pulseUsL, playOnlyFl=False ):
        if len(pitchL) > 0:
            self.pulseUsL  = pulseUsL
            self.pitchL   = pitchL
            self.pitch_idx = -1
            self._start_next_note( ms, playOnlyFl )
        
        
    def stop( self, ms ):
        self.pitch_idx = -1
        self.seq.stop(ms)

    def is_enabled( self ):
        return self.pitch_idx >= 0

    def tick( self, ms ):
        if self.is_enabled():
            self.seq.tick(ms)

            # if the sequencer is done playing 
            if not self.seq.is_enabled():
                self._start_next_note( ms, self.seq.playOnlyFl ) # ... else start the next sequence

        return None

    def _start_next_note( self, ms, playOnlyFl ):
        
        self.pitch_idx += 1

        # if the last note in pitchL has been played ...
        if self.pitch_idx >= len(self.pitchL):
            self.stop(ms)  # ... then we are done
        else:

            pitch = self.pitchL[ self.pitch_idx ]
            
            # be sure that the base directory exists
            baseDir = os.path.expanduser( cfg.outDir )
            if not os.path.isdir( baseDir ):
                os.mkdir( baseDir )

            outDir = os.path.join(baseDir, str(pitch) )

            if not os.path.isdir(outDir):
                os.mkdir(outDir)
            
            # get the next available output directory id
            outDir_id = self._calc_next_out_dir_id( outDir )

            print(outDir_id,outDir)
            
            # if this is not the first time this note has been sampled then get the resample locations
            if (outDir_id == 0) or self.cfg.useFullPulseListFl:
                self.pulseUsL = self.cfg.full_pulseL
            else:
                #self.pulseUsL,_,_ = form_resample_pulse_time_list( outDir, self.cfg.analysisArgs )
                self.pulseUsL = get_resample_points_wrap( baseDir, pitch, self.cfg.analysisArgs )

            holdDutyPctL = self.cfg.calibrateArgs['holdDutyPctD'][pitch]
            
            if playOnlyFl:
                self.pulseUsL,_,holdDutyPctL = form_final_pulse_list( outDir,  pitch,  self.cfg.analysisArgs, take_id=None )

                noteN = cfg.analysisArgs['auditionNoteN']
                self.pulseUsL   = [ self.pulseUsL[ int(round(i*126.0/(noteN-1)))] for i in range(noteN) ]

            else:
                outDir = os.path.join( outDir, str(outDir_id) )

                if not os.path.isdir(outDir):
                    os.mkdir(outDir)

            #------------------------
            j = 0
            holdDutyPctD = {}
            for us in self.pulseUsL:

                if j+1<len(holdDutyPctL) and us >= holdDutyPctL[j+1][0]:
                    j += 1
                    
                    
                holdDutyPctD[ us ] = holdDutyPctL[j][1]                
            #------------------------

            if self.cfg.reversePulseListFl:
                self.pulseUsL = [ us for us in reversed(self.pulseUsL) ]
                    
            # start the sequencer
            self.seq.start( ms, outDir, pitch, self.pulseUsL, holdDutyPctL, holdDutyPctD, playOnlyFl )
        

    def _calc_next_out_dir_id( self, outDir ):

        id = 0
        while os.path.isdir( os.path.join(outDir,"%i" % id)):
            id += 1

        return id
                    
        
# This is the main application API it is running in a child process.
class App:
    def __init__(self ):
        self.cfg            = None
        self.audioDev       = None
        self.api            = None
        self.cal_keys       = None
        self.keyboard       = None
        self.calibrate      = None
        self.midiFilePlayer = None
        
    def setup( self, cfg ):
        self.cfg = cfg

        self.audioDev = AudioDevice()
        self.midiDev = MidiDevice()

        res = None
        
        #
        # TODO: unify the result error handling
        # (the API and the audio device return two diferent 'Result' types
        #
        if hasattr(cfg,'audio'):
            res = self.audioDev.setup(**cfg.audio)

            if not res:
                self.audio_dev_list(0)
            
        else:
            self.audioDev = None

        if True:
            if hasattr(cfg,'midi'):
                res = self.midiDev.setup(**cfg.midi)

                if not res:
                    self.midi_dev_list(0)
            else:
                self.midiDev = None

            self.api = Picadae( key_mapL=cfg.key_mapL)

            # wait for the letter 'a' to come back from the serial port
            api_res = self.api.wait_for_serial_sync(timeoutMs=cfg.serial_sync_timeout_ms)

            # did the serial port sync fail?
            if not api_res:
                res.set_error("Serial port sync failed.")
            else:
                print("Serial port sync'ed")

                self.cal_keys = CalibrateKeys( cfg, self.audioDev, self.api )

                self.keyboard  = Keyboard( cfg, self.audioDev, self.api )

                self.calibrate = None #Calibrate( cfg.calibrateArgs, self.audioDev, self.midiDev, self.api )

                self.midiFilePlayer = MidiFilePlayer( cfg, self.api, self.midiDev, cfg.midiFileFn )
    
        return res

    def tick( self, ms ):

        if self.audioDev is not None:
            self.audioDev.tick(ms)
        
        if self.cal_keys:
            self.cal_keys.tick(ms)
            
        if self.keyboard:
            self.keyboard.tick(ms)
            
        if self.calibrate:
            self.calibrate.tick(ms)

        if self.midiFilePlayer:
            self.midiFilePlayer.tick(ms)

    def audio_dev_list( self, ms ):
        portL = self.audioDev.get_port_list( True )

        for port in portL:
            print("chs:%4i label:%s" % (port['chN'],port['label']))

    def midi_dev_list( self, ms ):
        d = self.midiDev.get_port_list( True )

        for port in d['listL']:
            print("IN:",port)
        
        d = self.midiDev.get_port_list( False )
          
        for port in d['listL']:
            print("OUT:",port)
        

    def calibrate_keys_start( self, ms, pitchRangeL ):
        pitchL = [ pitch  for pitch in range(pitchRangeL[0], pitchRangeL[1]+1)]
        self.cal_keys.start(  ms, pitchL, cfg.full_pulseL )

    def play_keys_start( self, ms, pitchRangeL ):
        chordL = [ [pitch]  for pitch in range(pitchRangeL[0], pitchRangeL[1]+1)]
        self.cal_keys.start(  ms, chordL, cfg.full_pulseL, playOnlyFl=True )

    def keyboard_start_pulse_idx( self, ms, argL ):
        pitchL = [ pitch  for pitch in range(argL[0], argL[1]+1)]        
        self.keyboard.start(  ms, pitchL, argL[2], None )

    def keyboard_repeat_pulse_idx( self, ms, argL ):
        self.keyboard.repeat(  ms, argL[0], None )
        
    def keyboard_start_target_db( self, ms, argL ):
        pitchL = [ pitch  for pitch in range(argL[0], argL[1]+1)]        
        self.keyboard.start(  ms, pitchL, None, argL[2] )

    def keyboard_repeat_target_db( self, ms, argL ):
        self.keyboard.repeat(  ms, None, argL[0] )

    def calibrate_start( self, ms, argL ):
        self.calibrate.start(ms)
        
    def calibrate_play( self, ms, argL ):
        self.calibrate.play(ms)
        
    def calibrate_keys_stop( self, ms ):
        self.cal_keys.stop(ms)
        self.keyboard.stop(ms)
        self.calibrate.stop(ms)

    def midi_file_player_start( self, ms ):
        self.midiFilePlayer.start(ms)

    def midi_file_player_stop( self, ms ):
        self.midiFilePlayer.stop(ms)

    def pedal_down( self, ms ):
        print("pedal_down")
        self.midiDev.send_controller(64, 100 )

    def pedal_up( self, ms ):
        print("pedal_up");
        self.midiDev.send_controller(64, 0 )
        
    def quit( self, ms ):
        if self.api:
            self.api.close()
    

def _send_error( pipe, res ):
    if res is None:
        return
    
    if res.msg:
        pipe.send( [{"type":"error", 'value':res.msg}] )

def _send_error_msg( pipe, msg ):
    _send_error( pipe, Result(None,msg))

def _send_quit( pipe ):
    pipe.send( [{ 'type':'quit' }] )
            
# This is the application engine async. process loop
def app_event_loop_func( pipe, cfg ):

    multiprocessing.get_logger().info("App Proc Started.")

    # create the asynchronous application object
    app = App()
    
    res = app.setup(cfg)

    # if the app did not initialize successfully
    if not res:
        _send_error( pipe, res )
        _send_quit(pipe)
        return

    dt0 = datetime.now()

    ms = 0

    while True:
        
        # have any message arrived from the parent process?
        if pipe.poll():

            msg = None
            try:
                msg = pipe.recv()
            except EOFError:
                return

            if not hasattr(app,msg.type):
                _send_error_msg( pipe, "Unknown message type:'%s'." % (msg.type) )
            else:

                # get the command handler function in 'app'
                func = getattr(app,msg.type)

                ms  = int(round( (datetime.now() - dt0).total_seconds() * 1000.0) )

                # call the command handler
                if msg.value:
                    res = func( ms, msg.value )
                else:
                    res = func( ms )

                # handle any errors returned from the commands
                _send_error( pipe, res )

            # if a 'quit' msg was recived then break out of the loop
            if msg.type == 'quit':
                _send_quit(pipe)
                break


        # give some time to the system
        time.sleep(0.05)
        
        # calc the tick() time stamp
        ms  = int(round( (datetime.now() - dt0).total_seconds() * 1000.0) )
        
        # tick the app
        app.tick( ms )
        

class AppProcess(Process):
    def __init__(self,cfg):
        self.parent_end, child_end = Pipe()
        super(AppProcess, self).__init__(target=app_event_loop_func,name="AppProcess",args=(child_end,cfg))
        self.doneFl    = False
        

    def send(self, d):
        # This function is called by the parent process to send an arbitrary msg to the App process
        self.parent_end.send( types.SimpleNamespace(**d) )
        return None
        
    def recv(self):
        # This function is called by the parent process to receive lists of child messages.
        
        msgL = None
        if not self.doneFl and self.parent_end.poll():
            
            msgL = self.parent_end.recv()

            for msg in msgL:
                if msg['type'] == 'quit':
                    self.doneFl = True
            
        return msgL
    
    def isdone(self):
        return self.doneFl
        

class Shell:
    def __init__( self, cfg ):
        self.appProc = None
        self.parseD  = {
            'q':{ "func":'quit',                      "minN":0,  "maxN":0, "help":"quit"},
            '?':{ "func":"_help",                     "minN":0,  "maxN":0, "help":"Print usage text."},
            'a':{ "func":"audio_dev_list",            "minN":0,  "maxN":0, "help":"List the audio devices."},
            'm':{ "func":"midi_dev_list",             "minN":0,  "maxN":0, "help":"List the MIDI devices."},
            'c':{ "func":"calibrate_keys_start",      "minN":1,  "maxN":2, "help":"Calibrate a range of keys. "},
            'd':{ "func":"calibrate_start",           "minN":1,  "maxN":1, "help":"Calibrate based on fixed db levels. "},
            'D':{ "func":"calibrate_play",            "minN":1,  "maxN":1, "help":"Play back last calibration."},
            's':{ "func":"calibrate_keys_stop",       "minN":0,  "maxN":0, "help":"Stop key calibration"},
            'p':{ "func":"play_keys_start",           "minN":1,  "maxN":2, "help":"Play current calibration"},
            'k':{ "func":"keyboard_start_pulse_idx",  "minN":3,  "maxN":3, "help":"Play pulse index across keyboard"},
            'r':{ "func":"keyboard_repeat_pulse_idx", "minN":1,  "maxN":1, "help":"Repeat pulse index across keyboard with new pulse_idx"},
            'K':{ "func":"keyboard_start_target_db",  "minN":3,  "maxN":3, "help":"Play db across keyboard"},
            'R':{ "func":"keyboard_repeat_target_db", "minN":1,  "maxN":1, "help":"Repeat db across keyboard with new pulse_idx"},
            'F':{ "func":"midi_file_player_start",    "minN":0,  "maxN":0, "help":"Play the MIDI file."},
            'f':{ "func":"midi_file_player_stop",     "minN":0,  "maxN":0, "help":"Stop the MIDI file."},
            'P':{ "func":"pedal_down",                "minN":0,  "maxN":0, "help":"Pedal down."},
            'U':{ "func":"pedal_up",                  "minN":0,  "maxN":0, "help":"Pedal up."},
            }

    def _help( self, _=None ):
        for k,d in self.parseD.items():
            s = "{} = {}".format( k, d['help'] )
            print(s)
        return None


    def _syntaxError( self, msg ):
        return Result(None,"Syntax Error: " + msg )
    
    def _exec_cmd( self, tokL ):

        if len(tokL) <= 0:
            return None

        opcode = tokL[0]
        
        if opcode not in self.parseD:
            return self._syntaxError("Unknown opcode: '{}'.".format(opcode))

        d = self.parseD[ opcode ]

        func_name = d['func']
        func      = None

        # find the function associated with this command
        if hasattr(self, func_name ):
            func = getattr(self, func_name )

        try:
            # convert the parameter list into integers
            argL = [ int(tokL[i]) for i in range(1,len(tokL)) ]
        except:
            return self._syntaxError("Unable to create integer arguments.")

        # validate the count of command args
        if  d['minN'] != -1 and (d['minN'] > len(argL) or len(argL) > d['maxN']):                
            return self._syntaxError("Argument count mismatch. {} is out of range:{} to {}".format(len(argL),d['minN'],d['maxN']))

        # call the command function
        if func:
            result = func(*argL)
        else:
            result = self.appProc.send( { 'type':func_name, 'value':argL } )

        return result


    
    def run( self ):

        # create the API object
        self.appProc = AppProcess(cfg)

        self.appProc.start()
            
        print("'q'=quit '?'=help")
        time_out_secs = 1

        # this is the shell main loop
        while True:

            # wait for keyboard activity
            i, o, e = select.select( [sys.stdin], [], [], time_out_secs )

            if i:
                # read the command
                s = sys.stdin.readline().strip() 

                # tokenize the command
                tokL = s.split(' ')

                # execute the command
                result = self._exec_cmd( tokL )
                        
                # if this is the 'quit' command
                if tokL[0] == 'q':
                    break

            # check for msg's from the async application process
            if self._handle_app_msgs( self.appProc.recv() ):
                break


        # wait for the appProc to complete                    
        while not self.appProc.isdone():
            self.appProc.recv()  # drain the AppProc() as it shutdown
            time.sleep(0.1)
       
 
    def _handle_app_msgs( self, msgL ):
        quitAppFl = False
        if msgL:
            for msg in msgL:
                if msg:
                    if msg['type'] == 'error':
                        print("Error: {}".format(msg['value']))

                    elif msg['type'] == 'quit':
                        quitAppFl = True
                    else:
                       print(msg) 

        return quitAppFl

def parse_args():
    """Parse the command line arguments."""
    
    descStr  = """Picadae auto-calibrate."""
    logL     = ['debug','info','warning','error','critical']
    
    ap = argparse.ArgumentParser(description=descStr)


    ap.add_argument("-c","--config",                     default="p_ac.yml",  help="YAML configuration file.")
    ap.add_argument("-l","--log_level",choices=logL,     default="warning",             help="Set logging level: debug,info,warning,error,critical. Default:warning")

    return ap.parse_args()
    
            
    
if __name__ == "__main__":
    
    logging.basicConfig()

    #mplog = multiprocessing.log_to_stderr()
    #mplog.setLevel(logging.INFO)
    

    args = parse_args()

    cfg = parse_yaml_cfg(args.config)

    shell = Shell(cfg)

    shell.run()