cml/piccal
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Many changes and additions.

Browse Source 
First working version of calibrate.py
Replaced analysis parameter dbRefWndMs with dbLinRef
rms_analysis.py : Added samples_to_linear_residual()
plot_seq_1.py : initial commit.
This commit is contained in:
kpl 2019-12-09 17:37:24 -05:00
parent fc1a0d8a61
commit ca9580cd50
12 changed files with 649 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
AudioDevice.py
View File

@ -126,6 +126,13 @@ class AudioDevice(object):
return res return res
def is_recording_enabled( self ):
if self.inStream is None:
return False
return self.inStream.active == True
def record_enable( self, enableFl ): def record_enable( self, enableFl ):
# if the input stream has not already been configured # if the input stream has not already been configured

95
calibrate.py
View File

@ -1,6 +1,6 @@
import os,types,wave,json,array import os,types,wave,json,array
import numpy as np import numpy as np
from rms_analysis import rms_analyze_one_note from rms_analysis import rms_analyze_one_rt_note
class Calibrate: class Calibrate:
def __init__( self, cfg, audio, midi, api ): def __init__( self, cfg, audio, midi, api ):
@ -65,9 +65,9 @@ class Calibrate:
if self.midi is not None: if self.midi is not None:
self.midi.send_all_notes_off() self.midi.send_all_notes_off()
if not self.playOnlyFl: self.audio.record_enable(False)
self.audio.record_enable(False)
if not self.playOnlyFl:
self._save_results() self._save_results()
def play(self,ms): def play(self,ms):
@ -75,11 +75,15 @@ class Calibrate:
if self.measD is None or len(self.measD) == 0: if self.measD is None or len(self.measD) == 0:
print("Nothing to play.") print("Nothing to play.")
else: else:
self.startMs = ms
self.state = 'started' self.state = 'started'
self.playOnlyFl = True self.playOnlyFl = True
self.nextStateChangeMs = ms + 500 self.nextStateChangeMs = ms + 500
self.curPitchIdx = -1 self.curPitchIdx = -1
self.curTargetDbIdx = 0 self.curTargetDbIdx = 0
self.audio.record_enable(True)
self._do_play_update() self._do_play_update()
def tick(self,ms): def tick(self,ms):
@ -102,6 +106,7 @@ class Calibrate:
elif self.state == 'note_off': elif self.state == 'note_off':
if self.playOnlyFl: if self.playOnlyFl:
if not self._do_play_update(): if not self._do_play_update():
self.stop(ms)
self.state = 'stopped' self.state = 'stopped'
else: else:
if self._do_analysis(ms): if self._do_analysis(ms):
@ -113,10 +118,29 @@ class Calibrate:
# if the state was not changed to 'stopped' # if the state was not changed to 'stopped'
if self.state == 'note_off': if self.state == 'note_off':
self.state = 'started' self.state = 'started'
def _calc_play_pulse_us( self, pitch, targetDb ):
pulseDbL = []
for d in self.measD[ pitch ]:
if d['targetDb'] == targetDb and d['matchFl']==True:
pulseDbL.append( ( d['pulse_us'], d[self.cfg.dbSrcLabel]['db']) )
if len(pulseDbL) == 0:
return -1
pulseL,dbL = zip(*pulseDbL)
# TODO: make a weighted average based on db error
return np.mean(pulseL)
def _do_play_update( self ): def _do_play_update( self ):
if self.curPitchIdx >= 0:
self._meas_note( self.cfg.pitchL[self.curPitchIdx], self.curPulseUs )
self.curPitchIdx +=1 self.curPitchIdx +=1
if self.curPitchIdx >= len(self.cfg.pitchL): if self.curPitchIdx >= len(self.cfg.pitchL):
self.curPitchIdx = 0 self.curPitchIdx = 0
@ -124,13 +148,10 @@ class Calibrate:
if self.curTargetDbIdx >= len(self.cfg.targetDbL): if self.curTargetDbIdx >= len(self.cfg.targetDbL):
return False return False
pitch = self.cfg.pitchL[ self.curPitchIdx ] pitch = self.cfg.pitchL[ self.curPitchIdx ]
targetDb = self.cfg.targetDbL[ self.curTargetDbIdx ] targetDb = self.cfg.targetDbL[ self.curTargetDbIdx ]
self.curPulseUs = -1 self.curPulseUs = self._calc_play_pulse_us( pitch, targetDb )
for d in self.measD[ pitch ]: self.curTargetDb = targetDb
if d['targetDb'] == targetDb and d['matchFl']==True:
self.curPulseUs = d['pulse_us']
break
if self.curPulseUs == -1: if self.curPulseUs == -1:
print("Pitch:%i TargetDb:%f not found." % (pitch,targetDb)) print("Pitch:%i TargetDb:%f not found." % (pitch,targetDb))
@ -195,10 +216,47 @@ class Calibrate:
#print("note-off: ",self.cfg.pitchL[ self.curPitchIdx]) #print("note-off: ",self.cfg.pitchL[ self.curPitchIdx])
def _proportional_step( self, targetDb, dbL, pulseL ):
curPulse,curDb = self.pulseDbL[-1]
# get the point closest to the target db
i = np.argmin( np.array(dbL) - targetDb )
# find the percentage difference to the target - based on the closest point
pd = abs(curDb-targetDb) / abs(curDb - dbL[i])
#
delta_pulse = pd * abs(curPulse - pulseL[i])
print("prop:",pd,"delta_pulse:",delta_pulse)
return int(round(curPulse + np.sign(targetDb - curDb) * delta_pulse))
def _step( self, targetDb, dbL, pulseL ):
pulse0,db0 = self.pulseDbL[-2]
pulse1,db1 = self.pulseDbL[-1]
# microseconds per decibel for the last two points
us_per_db = abs(pulse0-pulse1) / abs(db0-db1)
if us_per_db == 0:
us_per_db = 10 # ************************************** CONSTANT ***********************
# calcuate the decibels we need to move from the last point
error_db = targetDb - db1
print("us_per_db:",us_per_db," error db:", error_db )
return pulse1 + us_per_db * error_db
def _calc_next_pulse_us( self, targetDb ): def _calc_next_pulse_us( self, targetDb ):
# sort pulseDb ascending on db # sort pulseDb ascending on db
self.pulseDbL = sorted( self.pulseDbL, key=lambda x: x[1] ) #self.pulseDbL = sorted( self.pulseDbL, key=lambda x: x[1] )
pulseL,dbL = zip(*self.pulseDbL) pulseL,dbL = zip(*self.pulseDbL)
@ -223,6 +281,10 @@ class Calibrate:
self.deltaDnMult = 1 self.deltaDnMult = 1
pu = np.interp([targetDb],dbL,pulseL) pu = np.interp([targetDb],dbL,pulseL)
if int(pu) in pulseL:
pu = self._step(targetDb, dbL, pulseL )
return max(min(pu,self.cfg.maxPulseUs),self.cfg.minPulseUs) return max(min(pu,self.cfg.maxPulseUs),self.cfg.minPulseUs)
def _do_analysis(self,ms): def _do_analysis(self,ms):
@ -309,16 +371,18 @@ class Calibrate:
sigV = buf_result.value sigV = buf_result.value
# get the annotated begin and end of the note as sample indexes into sigV # get the annotated begin and end of the note as sample indexes into sigV
bi = int(round(annD['beg_ms'] * self.audio.srate / 1000)) bi = int(round(annD['beg_ms'] * self.audio.srate / 1000))
ei = int(round(annD['end_ms'] * self.audio.srate / 1000)) ei = int(round(annD['end_ms'] * self.audio.srate / 1000))
# calculate half the length of the note-off duration in samples # calculate half the length of the note-off duration in samples
noteOffSmp_o_2 = int(round(self.cfg.noteOffDurMs/2 * self.audio.srate / 1000)) noteOffSmp_o_2 = int(round( (self.cfg.noteOffDurMs/2) * self.audio.srate / 1000))
# widen the note analysis space noteOffSmp_o_2 samples pre/post the annotated begin/end of the note # widen the note analysis space noteOffSmp_o_2 samples pre/post the annotated begin/end of the note
bi = max(0,bi - noteOffSmp_o_2) bi = max(0,bi - noteOffSmp_o_2)
ei = min(noteOffSmp_o_2,sigV.shape[0]-1) ei = min(ei+noteOffSmp_o_2,sigV.shape[0]-1)
ar = types.SimpleNamespace(**self.cfg.analysisD) ar = types.SimpleNamespace(**self.cfg.analysisD)
@ -327,8 +391,11 @@ class Calibrate:
begMs = noteOffSmp_o_2 * 1000 / self.audio.srate begMs = noteOffSmp_o_2 * 1000 / self.audio.srate
endMs = begMs + (annD['end_ms'] - annD['beg_ms']) endMs = begMs + (annD['end_ms'] - annD['beg_ms'])
#print("MEAS:",begMs,endMs,bi,ei,sigV.shape,self.audio.is_recording_enabled(),ar)
# analyze the note # analyze the note
resD = rms_analyze_rt_one_note( sigV[bi:ei], self.audio.srate, begMs, endMs, midi_pitch, rmsWndMs=ar.rmsWndMs, rmsHopMs=ar.rmsHopMs, dbRefWndMs=ar.dbRefWndMs, harmCandN=ar.harmCandN, harmN=ar.harmN, durDecayPct=ar.durDecayPct ) resD = rms_analyze_one_rt_note( sigV[bi:ei], self.audio.srate, begMs, endMs, midi_pitch, rmsWndMs=ar.rmsWndMs, rmsHopMs=ar.rmsHopMs, dbRefWndMs=ar.dbRefWndMs, harmCandN=ar.harmCandN, harmN=ar.harmN, durDecayPct=ar.durDecayPct )
resD["pulse_us"] = pulse_us resD["pulse_us"] = pulse_us
resD["midi_pitch"] = midi_pitch resD["midi_pitch"] = midi_pitch

79
p_ac.py
View File

@ -10,20 +10,21 @@ from MidiDevice import MidiDevice
from result import Result from result import Result
from common import parse_yaml_cfg from common import parse_yaml_cfg
from plot_seq import form_resample_pulse_time_list from plot_seq import form_resample_pulse_time_list
from plot_seq import get_resample_points_wrap
from plot_seq import form_final_pulse_list from plot_seq import form_final_pulse_list
from rt_note_analysis import RT_Analyzer from rt_note_analysis import RT_Analyzer
from keyboard import Keyboard from keyboard import Keyboard
from calibrate import Calibrate from calibrate import Calibrate
class AttackPulseSeq: class AttackPulseSeq:
""" Sequence a fixed chord over a list of attack pulse lengths.""" """ Sequence a fixed pitch over a list of attack pulse lengths."""
def __init__(self, cfg, audio, api, noteDurMs=1000, pauseDurMs=1000 ): def __init__(self, cfg, audio, api, noteDurMs=1000, pauseDurMs=1000 ):
self.cfg = cfg self.cfg = cfg
self.audio = audio self.audio = audio
self.api = api self.api = api
self.outDir = None # directory to write audio file and results self.outDir = None # directory to write audio file and results
self.pitchL = None # chord to play self.pitch = None # pitch to paly
self.pulseUsL = [] # one onset pulse length in microseconds per sequence element self.pulseUsL = [] # one onset pulse length in microseconds per sequence element
self.noteDurMs = noteDurMs # duration of each chord in milliseconds self.noteDurMs = noteDurMs # duration of each chord in milliseconds
self.pauseDurMs = pauseDurMs # duration between end of previous note and start of next self.pauseDurMs = pauseDurMs # duration between end of previous note and start of next
@ -38,9 +39,9 @@ class AttackPulseSeq:
self.playOnlyFl = False self.playOnlyFl = False
self.rtAnalyzer = RT_Analyzer() self.rtAnalyzer = RT_Analyzer()
def start( self, ms, outDir, pitchL, pulseUsL, holdDutyPctL, playOnlyFl=False ): def start( self, ms, outDir, pitch, pulseUsL, holdDutyPctL, playOnlyFl=False ):
self.outDir = outDir # directory to write audio file and results self.outDir = outDir # directory to write audio file and results
self.pitchL = pitchL # chord to play self.pitch = pitch # note to play
self.pulseUsL = pulseUsL # one onset pulse length in microseconds per sequence element self.pulseUsL = pulseUsL # one onset pulse length in microseconds per sequence element
self.holdDutyPctL = holdDutyPctL self.holdDutyPctL = holdDutyPctL
self.pulse_idx = 0 self.pulse_idx = 0
@ -51,9 +52,6 @@ class AttackPulseSeq:
self.beginMs = ms self.beginMs = ms
self.playOnlyFl = playOnlyFl self.playOnlyFl = playOnlyFl
#for pitch in pitchL:
# self.api.set_pwm_duty( pitch, self.holdDutyPct )
# print("set PWM:%i"%(self.holdDutyPct))
# kpl if not playOnlyFl: # kpl if not playOnlyFl:
self.audio.record_enable(True) # start recording audio self.audio.record_enable(True) # start recording audio
@ -121,11 +119,10 @@ class AttackPulseSeq:
self.next_ms = ms + self.noteDurMs self.next_ms = ms + self.noteDurMs
self.state = 'note_off' self.state = 'note_off'
for pitch in self.pitchL: pulse_usec = int(self.pulseUsL[ self.pulse_idx ])
pulse_usec = int(self.pulseUsL[ self.pulse_idx ]) self._set_duty_cycle( self.pitch, pulse_usec )
self._set_duty_cycle( pitch, pulse_usec ) self.api.note_on_us( self.pitch, pulse_usec )
self.api.note_on_us( pitch, pulse_usec ) print("note-on:",self.pitch, self.pulse_idx, pulse_usec)
print("note-on:",pitch, self.pulse_idx, pulse_usec)
def _note_off( self, ms ): def _note_off( self, ms ):
self.eventTimeL[ self.pulse_idx ][1] = self.audio.buffer_sample_ms().value self.eventTimeL[ self.pulse_idx ][1] = self.audio.buffer_sample_ms().value
@ -135,15 +132,14 @@ class AttackPulseSeq:
if self.playOnlyFl: if self.playOnlyFl:
begTimeMs = self.eventTimeL[ self.pulse_idx ][0] begTimeMs = self.eventTimeL[ self.pulse_idx ][0]
endTimeMs = self.eventTimeL[ self.pulse_idx ][1] endTimeMs = self.eventTimeL[ self.pulse_idx ][1]
self.rtAnalyzer.analyze_note( self.audio, self.pitchL[0], begTimeMs, endTimeMs, self.cfg.analysisArgs['rmsAnalysisArgs'] ) self.rtAnalyzer.analyze_note( self.audio, self.pitch, begTimeMs, endTimeMs, self.cfg.analysisArgs['rmsAnalysisArgs'] )
self._send_note_off() self._send_note_off()
def _send_note_off( self ): def _send_note_off( self ):
for pitch in self.pitchL: self.api.note_off( self.pitch )
self.api.note_off( pitch ) #print("note-off:",self.pitch,self.pulse_idx)
#print("note-off:",pitch,self.pulse_idx)
def _disable(self): def _disable(self):
self.state = None self.state = None
@ -153,7 +149,7 @@ class AttackPulseSeq:
d = { d = {
"pulseUsL":self.pulseUsL, "pulseUsL":self.pulseUsL,
"pitchL":self.pitchL, "pitch":self.pitch,
"noteDurMs":self.noteDurMs, "noteDurMs":self.noteDurMs,
"pauseDurMs":self.pauseDurMs, "pauseDurMs":self.pauseDurMs,
"holdDutyPctL":self.holdDutyPctL, "holdDutyPctL":self.holdDutyPctL,
@ -180,16 +176,16 @@ class CalibrateKeys:
self.seq = AttackPulseSeq( cfg, audioDev, api, noteDurMs=cfg.noteDurMs, pauseDurMs=cfg.pauseDurMs ) self.seq = AttackPulseSeq( cfg, audioDev, api, noteDurMs=cfg.noteDurMs, pauseDurMs=cfg.pauseDurMs )
self.pulseUsL = None self.pulseUsL = None
self.chordL = None self.pitchL = None
self.pitch_idx = -1 self.pitch_idx = -1
def start( self, ms, chordL, pulseUsL, playOnlyFl=False ): def start( self, ms, pitchL, pulseUsL, playOnlyFl=False ):
if len(chordL) > 0: if len(pitchL) > 0:
self.pulseUsL = pulseUsL self.pulseUsL = pulseUsL
self.chordL = chordL self.pitchL = pitchL
self.pitch_idx = -1 self.pitch_idx = -1
self._start_next_chord( ms, playOnlyFl ) self._start_next_note( ms, playOnlyFl )
def stop( self, ms ): def stop( self, ms ):
@ -205,31 +201,27 @@ class CalibrateKeys:
# if the sequencer is done playing # if the sequencer is done playing
if not self.seq.is_enabled(): if not self.seq.is_enabled():
self._start_next_chord( ms, self.seq.playOnlyFl ) # ... else start the next sequence self._start_next_note( ms, self.seq.playOnlyFl ) # ... else start the next sequence
return None return None
def _start_next_chord( self, ms, playOnlyFl ): def _start_next_note( self, ms, playOnlyFl ):
self.pitch_idx += 1 self.pitch_idx += 1
# if the last chord in chordL has been played ... # if the last note in pitchL has been played ...
if self.pitch_idx >= len(self.chordL): if self.pitch_idx >= len(self.pitchL):
self.stop(ms) # ... then we are done self.stop(ms) # ... then we are done
else: else:
pitchL = self.chordL[ self.pitch_idx ] pitch = self.pitchL[ self.pitch_idx ]
# be sure that the base directory exists # be sure that the base directory exists
outDir = os.path.expanduser( cfg.outDir ) baseDir = os.path.expanduser( cfg.outDir )
if not os.path.isdir( outDir ): if not os.path.isdir( baseDir ):
os.mkdir( outDir ) os.mkdir( baseDir )
# form the output directory as "<label>_<pitch0>_<pitch1> ... " outDir = os.path.join(baseDir, str(pitch) )
dirStr = "_".join([ str(pitch) for pitch in pitchL ])
outDir = os.path.join(outDir, dirStr )
if not os.path.isdir(outDir): if not os.path.isdir(outDir):
os.mkdir(outDir) os.mkdir(outDir)
@ -240,13 +232,16 @@ class CalibrateKeys:
print(outDir_id,outDir) print(outDir_id,outDir)
# if this is not the first time this note has been sampled then get the resample locations # if this is not the first time this note has been sampled then get the resample locations
if outDir_id != 0: if outDir_id == 0:
self.pulseUsL,_,_ = form_resample_pulse_time_list( outDir, self.cfg.analysisArgs ) 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.holdDutyPctL holdDutyPctL = self.cfg.calibrateArgs['holdDutyPctD'][pitch]
if playOnlyFl: if playOnlyFl:
self.pulseUsL,_,holdDutyPctL = form_final_pulse_list( outDir, pitchL[0], self.cfg.analysisArgs, take_id=None ) self.pulseUsL,_,holdDutyPctL = form_final_pulse_list( outDir, pitch, self.cfg.analysisArgs, take_id=None )
noteN = cfg.analysisArgs['auditionNoteN'] noteN = cfg.analysisArgs['auditionNoteN']
self.pulseUsL = [ self.pulseUsL[ int(round(i*126.0/(noteN-1)))] for i in range(noteN) ] self.pulseUsL = [ self.pulseUsL[ int(round(i*126.0/(noteN-1)))] for i in range(noteN) ]
@ -258,7 +253,7 @@ class CalibrateKeys:
os.mkdir(outDir) os.mkdir(outDir)
# start the sequencer # start the sequencer
self.seq.start( ms, outDir, pitchL, self.pulseUsL, holdDutyPctL, playOnlyFl ) self.seq.start( ms, outDir, pitch, self.pulseUsL, holdDutyPctL, playOnlyFl )
def _calc_next_out_dir_id( self, outDir ): def _calc_next_out_dir_id( self, outDir ):
@ -356,8 +351,8 @@ class App:
def calibrate_keys_start( self, ms, pitchRangeL ): def calibrate_keys_start( self, ms, pitchRangeL ):
chordL = [ [pitch] for pitch in range(pitchRangeL[0], pitchRangeL[1]+1)] pitchL = [ pitch for pitch in range(pitchRangeL[0], pitchRangeL[1]+1)]
self.cal_keys.start( ms, chordL, cfg.full_pulseL ) self.cal_keys.start( ms, pitchL, cfg.full_pulseL )
def play_keys_start( self, ms, pitchRangeL ): def play_keys_start( self, ms, pitchRangeL ):
chordL = [ [pitch] for pitch in range(pitchRangeL[0], pitchRangeL[1]+1)] chordL = [ [pitch] for pitch in range(pitchRangeL[0], pitchRangeL[1]+1)]

21
p_ac.yml
View File

@ -27,10 +27,10 @@
# MeasureSeq args # MeasureSeq args
outDir: "~/temp/p_ac_3c", outDir: "~/temp/p_ac_3e",
noteDurMs: 1000, noteDurMs: 1000,
pauseDurMs: 1000, pauseDurMs: 1000,
holdDutyPctL: [ [0,50], [22000,55] ], #holdDutyPctL: [ [0,50], [22000,55] ],
full_pulse0L: [ 500, 1000, 1500, 2000, 2500, 3000, 3500, 4000, 4500, 5000, 5500, 6000, 6500, 7000, 8000, 9000, 10000, 12000, 14000, 18000, 22000, 26000, 30000, 34000, 40000], full_pulse0L: [ 500, 1000, 1500, 2000, 2500, 3000, 3500, 4000, 4500, 5000, 5500, 6000, 6500, 7000, 8000, 9000, 10000, 12000, 14000, 18000, 22000, 26000, 30000, 34000, 40000],
full_pulse1L: [ 10000, 11000, 12000, 13000, 14000, 15000, 16000, 17000, 18000, 20000, 22000, 24000, 26000, 30000, 32000, 34000, 36000, 40000], full_pulse1L: [ 10000, 11000, 12000, 13000, 14000, 15000, 16000, 17000, 18000, 20000, 22000, 24000, 26000, 30000, 32000, 34000, 36000, 40000],
@ -55,10 +55,15 @@
rmsAnalysisArgs: { rmsAnalysisArgs: {
rmsWndMs: 300, # length of the RMS measurment window rmsWndMs: 300, # length of the RMS measurment window
rmsHopMs: 30, # RMS measurement inter window distance rmsHopMs: 30, # RMS measurement inter window distance
dbRefWndMs: 500, # length of initial portion of signal to use to calculate the dB reference level dbLinRef: 0.01, # length of initial portion of signal to use to calculate the dB reference level
harmCandN: 5, # count of harmonic candidates to locate during harmonic based RMS analysis harmCandN: 5, # count of harmonic candidates to locate during harmonic based RMS analysis
harmN: 3, # count of harmonics to use to calculate harmonic based RMS analysis harmN: 3, # count of harmonics to use to calculate harmonic based RMS analysis
durDecayPct: 40, # percent drop in RMS to indicate the end of a note
}, },
resampleMinDb: 10.0, # note's less than this will be skipped
resampleNoiseLimitPct: 1.0, #
resampleMinDurMs: 800, # notes's whose duration is less than this will be skipped
minAttkDb: 7.0, # threshold of silence level minAttkDb: 7.0, # threshold of silence level
maxDbOffset: 0.25, # travel down the from the max. note level by at most this amount to locate the max. peak maxDbOffset: 0.25, # travel down the from the max. note level by at most this amount to locate the max. peak
@ -75,12 +80,12 @@
calibrateArgs: { calibrateArgs: {
outDir: "~/temp/calib0", outDir: "~/temp/calib0",
outLabel: "test", outLabel: "test_1",
analysisD: { analysisD: {
rmsWndMs: 300, # length of the RMS measurment window rmsWndMs: 300, # length of the RMS measurment window
rmsHopMs: 30, # RMS measurement inter window distance rmsHopMs: 30, # RMS measurement inter window distance
dbRefWndMs: 500, # length of initial portion of signal to use to calculate the dB reference level dbLinRef: 0.01, # length of initial portion of signal to use to calculate the dB reference level
harmCandN: 5, # count of harmonic candidates to locate during harmonic based RMS analysis harmCandN: 5, # count of harmonic candidates to locate during harmonic based RMS analysis
harmN: 3, # count of harmonics to use to calculate harmonic based RMS analysis harmN: 3, # count of harmonics to use to calculate harmonic based RMS analysis
durDecayPct: 40 # percent drop in RMS to indicate the end of a note durDecayPct: 40 # percent drop in RMS to indicate the end of a note
@ -90,11 +95,11 @@
noteOffDurMs: 1000, noteOffDurMs: 1000,
pitchL: [ 50, 51, 52 ], # list of pitches pitchL: [ 44, 45, 46, 47, 48, 49, 50, 51 ], # list of pitches
targetDbL: [ 16, 20, 23 ], # list of target db targetDbL: [ 16, 17, 18, 19, 20, 21, 22, 23, 24 ], # list of target db
minMeasDurMs: 800, # minimum candidate note duration minMeasDurMs: 800, # minimum candidate note duration
tolDbPct: 5.0, # tolerance as a percent of targetDb above/below used to form match db window tolDbPct: 2.0, # tolerance as a percent of targetDb above/below used to form match db window
maxPulseUs: 45000, # max. allowable pulse us maxPulseUs: 45000, # max. allowable pulse us
minPulseUs: 8000, # min. allowable pulse us minPulseUs: 8000, # min. allowable pulse us
initPulseUs: 15000, # pulseUs for first note initPulseUs: 15000, # pulseUs for first note

13
plot_calibrate.ipynb
View File

File diff suppressed because one or more lines are too long

11
plot_calibrate.py
View File

@ -5,8 +5,9 @@ import matplotlib._color_data as mcd
from matplotlib.pyplot import figure from matplotlib.pyplot import figure
from rms_analysis import calibrate_recording_analysis from rms_analysis import calibrate_recording_analysis
from rms_analysis import key_info_dictionary
def plot_by_pitch( inDir, pitch=None ): def plot_by_pitch( inDir, keyInfoD, pitch=None ):
anlD = calibrate_recording_analysis( inDir ) anlD = calibrate_recording_analysis( inDir )
jsonFn = os.path.join(inDir, "meas.json" ) jsonFn = os.path.join(inDir, "meas.json" )
@ -97,7 +98,7 @@ def plot_by_pitch( inDir, pitch=None ):
axL[axi].set_title("pitch:%i " % (midi_pitch)) axL[axi].set_title("pitch:%i %s" % (midi_pitch,keyInfoD[midi_pitch].type))
plt.legend() plt.legend()
plt.show() plt.show()
@ -140,11 +141,13 @@ if __name__ == "__main__":
pitch = None pitch = None
inDir = sys.argv[1] inDir = sys.argv[1]
if len(sys.argv) > 2: yamlFn = sys.argv[2]
if len(sys.argv) > 3:
pitch = int(sys.argv[2]) pitch = int(sys.argv[2])
keyInfoD = key_info_dictionary( yamlCfgFn=yamlFn)
#plot_all_notes( inDir ) #plot_all_notes( inDir )
plot_by_pitch(inDir,pitch) plot_by_pitch(inDir,keyInfoD,pitch)
#calibrate_recording_analysis( inDir ) #calibrate_recording_analysis( inDir )

21
plot_note_analysis.py
View File

@ -50,6 +50,8 @@ def select_min_note( rr, statsL, minDurMs=600, minDb=8, contextSecs=10 ):
#print(rr.tdRmsDbV.shape,rr.rmsDbV.shape) #print(rr.tdRmsDbV.shape,rr.rmsDbV.shape)
sL = []
if sel_note_r is None: if sel_note_r is None:
print("ERROR: No min note found.") print("ERROR: No min note found.")
@ -65,7 +67,6 @@ def select_min_note( rr, statsL, minDurMs=600, minDb=8, contextSecs=10 ):
rr.rmsDbV = rr.rmsDbV[bi:ei] rr.rmsDbV = rr.rmsDbV[bi:ei]
offsSec = bi / rr.rms_srate offsSec = bi / rr.rms_srate
sL = []
for r in statsL: for r in statsL:
begSmpIdx = int(round(r.begSmpSec * rr.rms_srate)) begSmpIdx = int(round(r.begSmpSec * rr.rms_srate))
endSmpIdx = int(round(r.endSmpSec * rr.rms_srate)) endSmpIdx = int(round(r.endSmpSec * rr.rms_srate))
@ -84,11 +85,11 @@ def select_min_note( rr, statsL, minDurMs=600, minDb=8, contextSecs=10 ):
return rr,sL return rr,sL
def plot_note_analysis( inDir ): def plot_note_audio_reanalysis( inDir ):
rmsWndMs=300 rmsWndMs=300
rmsHopMs=30 rmsHopMs=30
dbRefWndMs=500 dbLinRef=0.001
harmCandN=5 harmCandN=5
harmN=3 harmN=3
durDecayPct = 50 durDecayPct = 50
@ -99,21 +100,21 @@ def plot_note_analysis( inDir ):
take_id = int(pathL[-1]) take_id = int(pathL[-1])
midi_pitch = int(pathL[-2]) midi_pitch = int(pathL[-2])
r = rms_analysis_main( inDir, midi_pitch, rmsWndMs=rmsWndMs, rmsHopMs=rmsHopMs, dbRefWndMs=dbRefWndMs, harmCandN=harmCandN, harmN=harmN, durDecayPct=durDecayPct ) r = rms_analysis_main( inDir, midi_pitch, rmsWndMs=rmsWndMs, rmsHopMs=rmsHopMs, dbLinRef=dbLinRef, harmCandN=harmCandN, harmN=harmN, durDecayPct=durDecayPct )
r,statsL = select_min_note(r,r.statsL) r,statsL = select_min_note(r,r.statsL)
do_plot(r,statsL) do_plot(r,statsL)
def plot_note_analysis_dir( inDir, dirL ): def plot_note_audio_reanalysis_dir( inDir, dirL ):
for folder in dirL: for folder in dirL:
path = os.path.join(inDir,str(folder),"0") path = os.path.join(inDir,str(folder),"0")
plot_note_analysis( path ) plot_note_audio_reanalysis( path )
@ -127,7 +128,7 @@ def get_all_note_durations( inDir, cacheFn ):
takeId = 0 takeId = 0
rmsWndMs=300 rmsWndMs=300
rmsHopMs=30 rmsHopMs=30
dbRefWndMs=500 dbLinRef=0.001
harmCandN=5 harmCandN=5
harmN=3 harmN=3
durDecayPct = 40 durDecayPct = 40
@ -139,7 +140,7 @@ def get_all_note_durations( inDir, cacheFn ):
if os.path.isfile(os.path.join(takePath,'seq.json')): if os.path.isfile(os.path.join(takePath,'seq.json')):
print(midi_pitch) print(midi_pitch)
r = rms_analysis_main( takePath, midi_pitch, rmsWndMs=rmsWndMs, rmsHopMs=rmsHopMs, dbRefWndMs=dbRefWndMs, harmCandN=harmCandN, harmN=harmN, durDecayPct=durDecayPct ) r = rms_analysis_main( takePath, midi_pitch, rmsWndMs=rmsWndMs, rmsHopMs=rmsHopMs, dbLinRef=dbLinRef, harmCandN=harmCandN, harmN=harmN, durDecayPct=durDecayPct )
xL = [] xL = []
for i,sr in enumerate(r.statsL): for i,sr in enumerate(r.statsL):
@ -319,7 +320,7 @@ if __name__ == "__main__":
pitchL = [ 30,31,32,33,34,35 ] pitchL = [ 30,31,32,33,34,35 ]
pitchL = [ 70,71,72,73,74,75 ] pitchL = [ 70,71,72,73,74,75 ]
#plot_note_analysis_dir( "/home/kevin/temp/p_ac_3c",pitchL) plot_note_audio_reanalysis_dir( "/home/kevin/temp/p_ac_3c",pitchL)
durFn = "/home/kevin/temp/cache_note_dur.pickle" durFn = "/home/kevin/temp/cache_note_dur.pickle"
#get_all_note_durations("/home/kevin/temp/p_ac_3c",durFn) #get_all_note_durations("/home/kevin/temp/p_ac_3c",durFn)
@ -327,4 +328,4 @@ if __name__ == "__main__":
#plot_quiet_note_db(durFn,"p_ac.yml") #plot_quiet_note_db(durFn,"p_ac.yml")
dump_hold_duty_pct( "/home/kevin/temp/p_ac_3c" ) #dump_hold_duty_pct( "/home/kevin/temp/p_ac_3c" )

52
plot_seq.py
View File

@ -5,6 +5,7 @@ from common import parse_yaml_cfg
from rms_analysis import rms_analysis_main from rms_analysis import rms_analysis_main
from rms_analysis import select_first_stable_note_by_delta_db from rms_analysis import select_first_stable_note_by_delta_db
from rms_analysis import select_first_stable_note_by_dur from rms_analysis import select_first_stable_note_by_dur
from rms_analysis import samples_to_linear_residual
def is_nanV( xV ): def is_nanV( xV ):
@ -24,7 +25,7 @@ def _find_max_take_id( inDir ):
id -= 1 id -= 1
return id return id
def form_final_pulse_list( inDir, midi_pitch, analysisArgsD, take_id=None ): def form_final_pulse_list( inDir, midi_pitch, analysisArgsD, take_id=None ):
@ -40,6 +41,10 @@ def form_final_pulse_list( inDir, midi_pitch, analysisArgsD, take_id=None ):
take_number = int(idir) take_number = int(idir)
if not os.path.isfile(os.path.join( inDir,idir, "seq.json")):
continue
# analyze this takes audio and locate the note peaks # analyze this takes audio and locate the note peaks
r = rms_analysis_main( os.path.join(inDir,idir), midi_pitch, **analysisArgsD['rmsAnalysisArgs'] ) r = rms_analysis_main( os.path.join(inDir,idir), midi_pitch, **analysisArgsD['rmsAnalysisArgs'] )
@ -234,9 +239,14 @@ def form_resample_pulse_time_list( inDir, analysisArgsD ):
return resampleUsL, pkDbL, pkUsL return resampleUsL, pkDbL, pkUsL
def plot_curve( ax, pulseUsL, rmsDbV ):
coeff = np.polyfit(pulseUsL,rmsDbV,5)
func = np.poly1d(coeff)
def plot_resample_pulse_times( inDir, analysisArgsD ): ax.plot( pulseUsL, func(pulseUsL), color='red')
def plot_resample_pulse_times_0( inDir, analysisArgsD ):
newPulseUsL, rmsDbV, pulseUsL = form_resample_pulse_time_list( inDir, analysisArgsD ) newPulseUsL, rmsDbV, pulseUsL = form_resample_pulse_time_list( inDir, analysisArgsD )
@ -245,15 +255,43 @@ def plot_resample_pulse_times( inDir, analysisArgsD ):
fig,ax = plt.subplots() fig,ax = plt.subplots()
ax.plot(pulseUsL,rmsDbV ) ax.plot(pulseUsL,rmsDbV,marker='.' )
for us in newPulseUsL: for us in newPulseUsL:
ax.axvline( x = us ) ax.axvline( x = us )
ax.plot(velTblUsL,velTblDbL,marker='.',linestyle='None') ax.plot(velTblUsL,velTblDbL,marker='.',linestyle='None')
plt.show() plt.show()
def plot_resample_pulse_times( inDir, analysisArgsD ):
newPulseUsL, rmsDbV, pulseUsL = form_resample_pulse_time_list( inDir, analysisArgsD )
midi_pitch = int( inDir.split("/")[-1] )
velTblUsL,velTblDbL,_ = form_final_pulse_list( inDir, midi_pitch, analysisArgsD, take_id=None )
fig,axL = plt.subplots(2,1,gridspec_kw={'height_ratios': [2, 1]})
axL[0].plot(pulseUsL,rmsDbV,marker='.' )
#plot_curve( ax, velTblUsL,velTblDbL)
scoreV = samples_to_linear_residual( pulseUsL, rmsDbV)
axL[0].plot(pulseUsL,rmsDbV + scoreV)
axL[0].plot(pulseUsL,rmsDbV + np.power(scoreV,2.0))
axL[0].plot(pulseUsL,rmsDbV - np.power(scoreV,2.0))
axL[1].axhline(0.0,color='black')
axL[1].axhline(1.0,color='black')
axL[1].plot(pulseUsL,np.abs(scoreV * 100.0 / rmsDbV))
axL[1].set_ylim((0.0,50))
plt.show()
def find_min_max_peak_index( pkDbL, minDb, maxDbOffs ): def find_min_max_peak_index( pkDbL, minDb, maxDbOffs ):
""" """
Find the min db and max db peak. Find the min db and max db peak.
@ -348,7 +386,7 @@ def plot_spectrum( ax, srate, binHz, specV, midiPitch, harmN ):
ax.set_ylabel(str(midiPitch)) ax.set_ylabel(str(midiPitch))
def plot_spectral_ranges( inDir, pitchL, rmsWndMs=300, rmsHopMs=30, harmN=5, dbRefWndMs=500 ): def plot_spectral_ranges( inDir, pitchL, rmsWndMs=300, rmsHopMs=30, harmN=5, dbLinRef=0.001 ):
""" Plot the spectrum from one note (7th from last) in each attack pulse length sequence referred to by pitchL.""" """ Plot the spectrum from one note (7th from last) in each attack pulse length sequence referred to by pitchL."""
plotN = len(pitchL) plotN = len(pitchL)
@ -369,7 +407,7 @@ def plot_spectral_ranges( inDir, pitchL, rmsWndMs=300, rmsHopMs=30, harmN=5, dbR
sigV = signalM / float(0x7fff) sigV = signalM / float(0x7fff)
# calc. the RMS envelope in the time domain # calc. the RMS envelope in the time domain
rms0DbV, rms0_srate = audio_rms( srate, sigV, rmsWndMs, rmsHopMs, dbRefWndMs ) rms0DbV, rms0_srate = audio_rms( srate, sigV, rmsWndMs, rmsHopMs, dbLinRef )
# locate the sample index of the peak of each note attack # locate the sample index of the peak of each note attack
pkIdx0L = locate_peak_indexes( rms0DbV, rms0_srate, r['eventTimeL'] ) pkIdx0L = locate_peak_indexes( rms0DbV, rms0_srate, r['eventTimeL'] )
@ -383,7 +421,7 @@ def plot_spectral_ranges( inDir, pitchL, rmsWndMs=300, rmsHopMs=30, harmN=5, dbR
# calc. the RMS envelope by taking the max spectral peak in each STFT window # calc. the RMS envelope by taking the max spectral peak in each STFT window
rmsDbV, rms_srate, specV, specHopIdx, binHz = audio_stft_rms( srate, sigV, rmsWndMs, rmsHopMs, dbRefWndMs, spectrumSmpIdx) rmsDbV, rms_srate, specV, specHopIdx, binHz = audio_stft_rms( srate, sigV, rmsWndMs, rmsHopMs, dbLinRef, spectrumSmpIdx)
# specV[] is the spectrum of the note at spectrumSmpIdx # specV[] is the spectrum of the note at spectrumSmpIdx

248
plot_seq_1.py Normal file
View File

@ -0,0 +1,248 @@
import os, sys
import matplotlib.pyplot as plt
import numpy as np
from common import parse_yaml_cfg
import rms_analysis
def get_merged_pulse_db_measurements( inDir, midi_pitch, analysisArgsD ):
inDir = os.path.join(inDir,"%i" % (midi_pitch))
dirL = os.listdir(inDir)
pkL = []
# for each take in this directory
for idir in dirL:
take_number = int(idir)
# analyze this takes audio and locate the note peaks
r = rms_analysis.rms_analysis_main( os.path.join(inDir,idir), midi_pitch, **analysisArgsD )
# store the peaks in pkL[ (db,us) ]
for db,us,stats in zip(r.pkDbL,r.pkUsL,r.statsL):
pkL.append( (db,us,stats.durMs,take_number) )
# sort the peaks on increasing attack pulse microseconds
pkL = sorted( pkL, key= lambda x: x[1] )
# merge sample points that separated by less than 'minSampleDistUs' milliseconds
#pkL = merge_close_sample_points( pkL, analysisArgsD['minSampleDistUs'] )
# split pkL
pkDbL,pkUsL,durMsL,takeIdL = tuple(zip(*pkL))
return pkUsL,pkDbL,durMsL,takeIdL,r.holdDutyPctL
def select_resample_reference_indexes( noiseIdxL ):
resampleIdxS = set()
for i in noiseIdxL:
resampleIdxS.add( i )
resampleIdxS.add( i+1 )
resampleIdxS.add( i-1 )
resampleIdxL = list(resampleIdxS)
# if a single sample point is left out of a region of
# contiguous sample points then include this as a resample point
for i in resampleIdxL:
if i + 1 not in resampleIdxL and i + 2 in resampleIdxL: # BUG BUG BUG: Hardcoded constant
resampleIdxL.append(i+1)
return resampleIdxL
def locate_resample_regions( usL, dbL, resampleIdxL ):
# locate regions of points to resample
regionL = [] # (bi,ei)
inRegionFl = False
bi = None
for i in range(len(usL)):
if inRegionFl:
if i not in resampleIdxL:
regionL.append((bi,i-1))
inRegionFl = False
bi = None
else:
if i in resampleIdxL:
inRegionFl = True
bi = i
if bi is not None:
regionL.append((bi,len(usL)-1))
# select points around and within the resample regions
# to resample
reUsL = []
reDbL = []
for bi,ei in regionL:
for i in range(bi,ei+2):
if i == 0:
us = usL[i]
db = dbL[i]
elif i >= len(usL):
us = usL[i-1]
db = dbL[i-1]
else:
us = usL[i-1] + (usL[i]-usL[i-1])/2
db = dbL[i-1] + (dbL[i]-dbL[i-1])/2
reUsL.append(us)
reDbL.append(db)
return reUsL,reDbL
def get_dur_skip_indexes( durMsL, dbL, takeIdL, minDurMs, minDb ):
firstAudibleIdx = None
firstNonSkipIdx = None
skipIdxL = [ i for i,(ms,db) in enumerate(zip(durMsL,dbL)) if ms < minDurMs or db < minDb ]
# if a single sample point is left out of a region of
# contiguous skipped points then skip this point also
for i in range(len(durMsL)):
if i not in skipIdxL and i-1 in skipIdxL and i+1 in skipIdxL:
skipIdxL.append(i)
# find the first set of 3 contiguous samples that
# are greater than minDurMs - all samples prior
# to these will be skipped
xL = []
for i in range(len(durMsL)):
if i in skipIdxL:
xL = []
else:
xL.append(i)
if len(xL) == 3: # BUG BUG BUG: Hardcoded constant
firstAudibleIdx = xL[0]
break
# decrease by one decibel to locate the first non-skip
# TODO: what if no note exists that is one decibel less
# The recordings of very quiet notes do not give reliabel decibel measures
# so this may not be the best backup criteria
if firstAudibleIdx is not None:
i = firstAudibleIdx-1
while abs(dbL[i] - dbL[firstAudibleIdx]) < 1.0: # BUG BUG BUG: Hardcoded constant
i -= 1
firstNonSkipIdx = i
return skipIdxL, firstAudibleIdx, firstNonSkipIdx
def get_resample_points( usL, dbL, durMsL, takeIdL, minDurMs, minDb, noiseLimitPct ):
skipIdxL, firstAudibleIdx, firstNonSkipIdx = get_dur_skip_indexes( durMsL, dbL, takeIdL, minDurMs, minDb )
durL = [ (usL[i],dbL[i]) for i in skipIdxL ]
scoreV = np.abs( rms_analysis.samples_to_linear_residual( usL, dbL) * 100.0 / dbL )
noiseIdxL = [ i for i in range(scoreV.shape[0]) if scoreV[i] > noiseLimitPct ]
noiseL = [ (usL[i],dbL[i]) for i in noiseIdxL ]
resampleIdxL = select_resample_reference_indexes( noiseIdxL )
resampleIdxL = [ i for i in resampleIdxL if i >= firstNonSkipIdx ]
resampleL = [ (usL[i],dbL[i]) for i in resampleIdxL ]
reUsL,reDbL = locate_resample_regions( usL, dbL, resampleIdxL )
return reUsL, reDbL, noiseL, resampleL, durL, firstAudibleIdx, firstNonSkipIdx
def get_resample_points_wrap( inDir, midi_pitch, analysisArgsD ):
usL, dbL, durMsL,_,_ = get_merged_pulse_db_measurements( inDir, midi_pitch, analysisArgsD['rmsAnalysisArgs'] )
reUsL,_,_,_,_,_,_ = get_resample_points( usL, dbL, durMsL, analysisArgsD['resampleMinDurMs'], analysisArgsD['resampleMinDb'], analysisArgsD['resampleNoiseLimitPct'] )
return reUsL
def plot_noise_region( ax, inDir, keyMapD, midi_pitch, analysisArgsD ):
plotResampleFl = False
plotTakesFl = True
usL, dbL, durMsL, takeIdL, holdDutyPctL = get_merged_pulse_db_measurements( inDir, midi_pitch, analysisArgsD['rmsAnalysisArgs'] )
reUsL, reDbL, noiseL, resampleL, durL, firstAudibleIdx, firstNonSkipIdx = get_resample_points( usL, dbL, durMsL, takeIdL, analysisArgsD['resampleMinDurMs'], analysisArgsD['resampleMinDb'], analysisArgsD['resampleNoiseLimitPct'] )
# plot first audible and non-skip position
ax.plot( usL[firstNonSkipIdx], dbL[firstNonSkipIdx], markersize=15, marker='+', linestyle='None', color='red')
ax.plot( usL[firstNonSkipIdx], dbL[firstAudibleIdx], markersize=15, marker='*', linestyle='None', color='red')
# plot the resample points
if plotResampleFl:
ax.plot( reUsL, reDbL, markersize=10, marker='x', linestyle='None', color='green')
# plot the noisy sample positions
if noiseL:
nUsL,nDbL = zip(*noiseL)
ax.plot( nUsL, nDbL, marker='o', linestyle='None', color='black')
# plot the noisy sample positions and the neighbors included in the noisy region
if resampleL:
nUsL,nDbL = zip(*resampleL)
ax.plot( nUsL, nDbL, marker='*', linestyle='None', color='red')
# plot actual sample points
if plotTakesFl:
for takeId in list(set(takeIdL)):
xL,yL = zip(*[(usL[i],dbL[i]) for i in range(len(usL)) if takeIdL[i]==takeId ])
ax.plot(xL,yL, marker='.')
for i,(x,y) in enumerate(zip(xL,yL)):
ax.text(x,y,str(i))
else:
ax.plot(usL, dbL, marker='.')
# plot the duration skip points
if durL:
nUsL,nDbL = zip(*durL)
ax.plot( nUsL, nDbL, marker='.', linestyle='None', color='yellow')
# plot the locations where the hold duty cycle changes with vertical black lines
for us_duty in holdDutyPctL:
us,duty = tuple(us_duty)
if us > 0:
ax.axvline(us,color='black')
# plot the 'minDb' reference line
ax.axhline(analysisArgsD['resampleMinDb'] ,color='black')
ax.set_ylabel( "%i %s %s" % (midi_pitch, keyMapD[midi_pitch]['type'],keyMapD[midi_pitch]['class']))
def plot_noise_regions_main( inDir, cfg, pitchL ):
analysisArgsD = cfg.analysisArgs
keyMapD = { d['midi']:d for d in cfg.key_mapL }
axN = len(pitchL)
fig,axL = plt.subplots(axN,1)
if axN == 1:
axL = [axL]
fig.set_size_inches(18.5, 10.5*axN)
for ax,midi_pitch in zip(axL,pitchL):
plot_noise_region( ax,inDir, cfg.key_mapL, midi_pitch, analysisArgsD )
plt.show()
if __name__ == "__main__":
inDir = sys.argv[1]
cfgFn = sys.argv[2]
pitch = int(sys.argv[3])
cfg = parse_yaml_cfg( cfgFn )
pitchL = [pitch]
plot_noise_regions_main( inDir, cfg, pitchL )
#rms_analysis.write_audacity_label_files( inDir, cfg.analysisArgs['rmsAnalysisArgs'] )

76
plot_us_db_range.ipynb Normal file
View File

File diff suppressed because one or more lines are too long

148
rms_analysis.py
View File

@ -26,15 +26,16 @@ def calc_harm_bins( srate, binHz, midiPitch, harmN ):
return fund_l_binL, fund_m_binL, fund_u_binL return fund_l_binL, fund_m_binL, fund_u_binL
def rms_to_db( xV, rms_srate, refWndMs ): def rms_to_db( xV, rms_srate, dbLinRef ):
#dbWndN = int(round(refWndMs * rms_srate / 1000.0)) #dbWndN = int(round(refWndMs * rms_srate / 1000.0))
#dbRef = ref = np.mean(xV[0:dbWndN]) #dbRef = ref = np.mean(xV[0:dbWndN])
dbRef = refWndMs ######################################################### HACK HACK HACK HACK HACK
rmsDbV = 20.0 * np.log10( xV / dbRef ) #print("DB REF:",dbLinRef)
rmsDbV = 20.0 * np.log10( xV / dbLinRef )
return rmsDbV return rmsDbV
def audio_rms( srate, xV, rmsWndMs, hopMs, refWndMs ): def audio_rms( srate, xV, rmsWndMs, hopMs, dbLinRef ):
wndSmpN = int(round( rmsWndMs * srate / 1000.0)) wndSmpN = int(round( rmsWndMs * srate / 1000.0))
hopSmpN = int(round( hopMs * srate / 1000.0)) hopSmpN = int(round( hopMs * srate / 1000.0))
@ -61,10 +62,10 @@ def audio_rms( srate, xV, rmsWndMs, hopMs, refWndMs ):
j += 1 j += 1
rms_srate = srate / hopSmpN rms_srate = srate / hopSmpN
return rms_to_db( yV, rms_srate, refWndMs ), rms_srate return rms_to_db( yV, rms_srate, dbLinRef ), rms_srate
def audio_stft_rms( srate, xV, rmsWndMs, hopMs, refWndMs, spectrumIdx ): def audio_stft_rms( srate, xV, rmsWndMs, hopMs, dbLinRef, spectrumIdx ):
wndSmpN = int(round( rmsWndMs * srate / 1000.0)) wndSmpN = int(round( rmsWndMs * srate / 1000.0))
hopSmpN = int(round( hopMs * srate / 1000.0)) hopSmpN = int(round( hopMs * srate / 1000.0))
@ -82,12 +83,12 @@ def audio_stft_rms( srate, xV, rmsWndMs, hopMs, refWndMs, spectrumIdx ):
rms_srate = srate / hopSmpN rms_srate = srate / hopSmpN
mV = rms_to_db( mV, rms_srate, refWndMs ) mV = rms_to_db( mV, rms_srate, dbLinRef )
return mV, rms_srate, specV, specHopIdx, binHz return mV, rms_srate, specV, specHopIdx, binHz
def audio_harm_rms( srate, xV, rmsWndMs, hopMs, dbRefWndMs, midiPitch, harmCandN, harmN ): def audio_harm_rms( srate, xV, rmsWndMs, hopMs, dbLinRef, midiPitch, harmCandN, harmN ):
wndSmpN = int(round( rmsWndMs * srate / 1000.0)) wndSmpN = int(round( rmsWndMs * srate / 1000.0))
hopSmpN = int(round( hopMs * srate / 1000.0)) hopSmpN = int(round( hopMs * srate / 1000.0))
@ -116,7 +117,7 @@ def audio_harm_rms( srate, xV, rmsWndMs, hopMs, dbRefWndMs, midiPitch, harmCandN
rms_srate = srate / hopSmpN rms_srate = srate / hopSmpN
rmsV = rms_to_db( rmsV, rms_srate, dbRefWndMs ) rmsV = rms_to_db( rmsV, rms_srate, dbLinRef )
return rmsV, rms_srate, binHz return rmsV, rms_srate, binHz
def measure_duration_ms( rmsV, rms_srate, peak_idx, end_idx, decay_pct ): def measure_duration_ms( rmsV, rms_srate, peak_idx, end_idx, decay_pct ):
@ -257,26 +258,20 @@ def key_info_dictionary( keyMapL=None, yamlCfgFn=None):
return kmD return kmD
def rms_analyze_one_rt_note( sigV, srate, begMs, endMs, midi_pitch, rmsWndMs=300, rmsHopMs=30, dbRefWndMs=500, harmCandN=5, harmN=3, durDecayPct=40 ): def rms_analyze_one_rt_note( sigV, srate, begMs, endMs, midi_pitch, rmsWndMs=300, rmsHopMs=30, dbLinRef=0.001, harmCandN=5, harmN=3, durDecayPct=40 ):
sigV = np.squeeze(sigV) sigV = np.squeeze(sigV)
# HACK HACK HACK HACK td_rmsDbV, td_srate = audio_rms( srate, sigV, rmsWndMs, rmsHopMs, dbLinRef )
dbRefWndMs = 0.002 # HACK HACK HACK HACK
# HACK HACK HACK HACK
td_rmsDbV, td_srate = audio_rms( srate, sigV, rmsWndMs, rmsHopMs, dbRefWndMs )
begSmpIdx = int(round(begMs * td_srate/1000)) begSmpIdx = int(round(begMs * td_srate/1000))
endSmpIdx = int(round(endMs * td_srate/1000)) endSmpIdx = int(round(endMs * td_srate/1000))
td_pk_idx = begSmpIdx + np.argmax(td_rmsDbV[begSmpIdx:endSmpIdx]) td_pk_idx = begSmpIdx + np.argmax(td_rmsDbV[begSmpIdx:endSmpIdx])
td_durMs = measure_duration_ms( td_rmsDbV, td_srate, td_pk_idx, len(sigV)-1, durDecayPct ) td_durMs = measure_duration_ms( td_rmsDbV, td_srate, td_pk_idx, len(sigV)-1, durDecayPct )
# HACK HACK HACK HACK hm_rmsDbV, hm_srate, binHz = audio_harm_rms( srate, sigV, rmsWndMs, rmsHopMs, dbLinRef, midi_pitch, harmCandN, harmN )
dbRefWndMs = 0.01 # HACK HACK HACK HACK
# HACK HACK HACK HACK
hm_rmsDbV, hm_srate, binHz = audio_harm_rms( srate, sigV, rmsWndMs, rmsHopMs, dbRefWndMs, midi_pitch, harmCandN, harmN )
begSmpIdx = int(round(begMs * hm_srate/1000)) begSmpIdx = int(round(begMs * hm_srate/1000))
endSmpIdx = int(round(endMs * hm_srate/1000)) endSmpIdx = int(round(endMs * hm_srate/1000))
@ -289,6 +284,7 @@ def rms_analyze_one_rt_note( sigV, srate, begMs, endMs, midi_pitch, rmsWndMs=300
return { "td":tdD, "hm":hmD } return { "td":tdD, "hm":hmD }
def calibrate_rms( sigV, srate, beg_ms, end_ms ): def calibrate_rms( sigV, srate, beg_ms, end_ms ):
bi = int(round(beg_ms * srate / 1000)) bi = int(round(beg_ms * srate / 1000))
@ -321,11 +317,7 @@ def calibrate_recording_analysis( inDir ):
anlr = types.SimpleNamespace(**cfg.analysisD) anlr = types.SimpleNamespace(**cfg.analysisD)
# HACK HACK HACK HACK tdRmsDbV, td_srate = audio_rms( srate, sigV, anlr.rmsWndMs, anlr.rmsHopMs, anlr.dbLinRef )
dbRefWndMs = 0.002 # HACK HACK HACK HACK
# HACK HACK HACK HACK
tdRmsDbV, td_srate = audio_rms( srate, sigV, anlr.rmsWndMs, anlr.rmsHopMs, dbRefWndMs )
# for each measured pitch # for each measured pitch
for midi_pitch,measL in measD.items(): for midi_pitch,measL in measD.items():
@ -358,7 +350,7 @@ def calibrate_recording_analysis( inDir ):
def rms_analysis_main( inDir, midi_pitch, rmsWndMs=300, rmsHopMs=30, dbRefWndMs=500, harmCandN=5, harmN=3, durDecayPct=40 ): def rms_analysis_main( inDir, midi_pitch, rmsWndMs=300, rmsHopMs=30, dbLinRef=0.001, harmCandN=5, harmN=3, durDecayPct=40 ):
seqFn = os.path.join( inDir, "seq.json") seqFn = os.path.join( inDir, "seq.json")
audioFn = os.path.join( inDir, "audio.wav") audioFn = os.path.join( inDir, "audio.wav")
@ -371,15 +363,13 @@ def rms_analysis_main( inDir, midi_pitch, rmsWndMs=300, rmsHopMs=30, dbRefWndMs=
srate, signalM = wavfile.read(audioFn) srate, signalM = wavfile.read(audioFn)
sigV = signalM / float(0x7fff) sigV = signalM / float(0x7fff)
tdRmsDbV, rms0_srate = audio_rms( srate, sigV, rmsWndMs, rmsHopMs, dbRefWndMs ) tdRmsDbV, rms0_srate = audio_rms( srate, sigV, rmsWndMs, rmsHopMs, dbLinRef )
tdPkIdxL = locate_peak_indexes( tdRmsDbV, rms0_srate, r['eventTimeL']) tdPkIdxL = locate_peak_indexes( tdRmsDbV, rms0_srate, r['eventTimeL'])
rmsDbV, rms_srate, binHz = audio_harm_rms( srate, sigV, rmsWndMs, rmsHopMs, dbRefWndMs, midi_pitch, harmCandN, harmN ) rmsDbV, rms_srate, binHz = audio_harm_rms( srate, sigV, rmsWndMs, rmsHopMs, dbLinRef, midi_pitch, harmCandN, harmN )
pkIdxL = locate_peak_indexes( rmsDbV, rms_srate, r['eventTimeL'] )
pkIdxL = locate_peak_indexes( rmsDbV, rms_srate, r['eventTimeL'] )
holdDutyPctL = None holdDutyPctL = None
if 'holdDutyPct' in r: if 'holdDutyPct' in r:
@ -390,6 +380,7 @@ def rms_analysis_main( inDir, midi_pitch, rmsWndMs=300, rmsHopMs=30, dbRefWndMs=
r = types.SimpleNamespace(**{ r = types.SimpleNamespace(**{
"audio_srate":srate, "audio_srate":srate,
"eventTimeMsL":r['eventTimeL'],
"tdRmsDbV": tdRmsDbV, "tdRmsDbV": tdRmsDbV,
"tdPkIdxL": tdPkIdxL, "tdPkIdxL": tdPkIdxL,
"tdPkDbL": [ tdRmsDbV[i] for i in tdPkIdxL ], "tdPkDbL": [ tdRmsDbV[i] for i in tdPkIdxL ],
@ -397,8 +388,6 @@ def rms_analysis_main( inDir, midi_pitch, rmsWndMs=300, rmsHopMs=30, dbRefWndMs=
"rmsDbV":rmsDbV, "rmsDbV":rmsDbV,
"rms_srate":rms_srate, "rms_srate":rms_srate,
"pkIdxL":pkIdxL, # pkIdxL[ len(pulsUsL) ] - indexes into rmsDbV[] of peaks "pkIdxL":pkIdxL, # pkIdxL[ len(pulsUsL) ] - indexes into rmsDbV[] of peaks
#"min_pk_idx":min_pk_idx,
#"max_pk_idx":max_pk_idx,
"eventTimeL":r['eventTimeL'], "eventTimeL":r['eventTimeL'],
"holdDutyPctL":holdDutyPctL, "holdDutyPctL":holdDutyPctL,
'pkDbL': [ rmsDbV[ i ] for i in pkIdxL ], 'pkDbL': [ rmsDbV[ i ] for i in pkIdxL ],
@ -412,7 +401,7 @@ def rms_analysis_main( inDir, midi_pitch, rmsWndMs=300, rmsHopMs=30, dbRefWndMs=
return r return r
def rms_analysis_main_all( inDir, cacheFn, rmsWndMs=300, rmsHopMs=30, dbRefWndMs=500, harmCandN=5, harmN=3, durDecayPct=40 ): def rms_analysis_main_all( inDir, cacheFn, rmsWndMs=300, rmsHopMs=30, dbLinRef=0.001, harmCandN=5, harmN=3, durDecayPct=40 ):
if os.path.isfile(cacheFn): if os.path.isfile(cacheFn):
print("READING analysis cache file: %s" % (cacheFn)) print("READING analysis cache file: %s" % (cacheFn))
@ -436,7 +425,7 @@ def rms_analysis_main_all( inDir, cacheFn, rmsWndMs=300, rmsHopMs=30, dbRefWndMs
path = os.path.join(inDir,folder,'0') path = os.path.join(inDir,folder,'0')
if os.path.isdir(path) and os.path.isfile(os.path.join(os.path.join(path,"seq.json"))): if os.path.isdir(path) and os.path.isfile(os.path.join(os.path.join(path,"seq.json"))):
r = rms_analysis_main( path, midi_pitch, rmsWndMs=rmsWndMs, rmsHopMs=rmsHopMs, dbRefWndMs=dbRefWndMs, harmCandN=harmCandN, harmN=harmN, durDecayPct=durDecayPct ) r = rms_analysis_main( path, midi_pitch, rmsWndMs=rmsWndMs, rmsHopMs=rmsHopMs, dbLinRef=dbLinRef, harmCandN=harmCandN, harmN=harmN, durDecayPct=durDecayPct )
rD[ midi_pitch ] = r rD[ midi_pitch ] = r
@ -445,3 +434,94 @@ def rms_analysis_main_all( inDir, cacheFn, rmsWndMs=300, rmsHopMs=30, dbRefWndMs
pickle.dump(rD,f) pickle.dump(rD,f)
return rD return rD
def samples_to_linear_residual( usL, dbL, pointsPerLine=5 ):
# Score the quality of each sampled point by measuring the
# quality of fit to a local line.
scoreD = { us:[] for us in usL }
pointsPerLine = 5
i = pointsPerLine
# for each sampled point
while i < len(usL):
# beginning with sample at index 'pointsPerLine'
if i >= pointsPerLine:
k = i - pointsPerLine
# get the x (us) and y (db) sample values
xL,yL = zip(*[ ((usL[k+j],1.0), dbL[k+j]) for j in range(pointsPerLine)])
xV = np.array(xL)
yV = np.array(yL)
# fit the sampled point to a line
m,c = np.linalg.lstsq(xV,yV,rcond=None)[0]
# calc the residual of the fit at each point
resV = (m*xV+c)[:,0] - yV
# assign the residual to the associated point in scoreD[x]
for j in range(pointsPerLine):
scoreD[usL[k+j]].append(resV[j])
i += 1
scoreL = []
# calc the mean of the residuals for each point
# (most points were used in 'pointsPerLine' line estimations
# and so they will have 'pointsPerLine' residual values)
for us in usL:
resL = scoreD[us]
if len(resL) == 0:
scoreL.append(0.0)
elif len(resL) == 1:
scoreL.append(resL[0])
else:
scoreL.append(np.mean(resL))
# their should be one mean resid. value for each sampled point
assert( len(scoreL) == len(usL) )
return np.array(scoreL)
def write_audacity_label_files( inDir, analysisArgsD ):
pitchDirL = os.listdir(inDir)
for pitchDir in pitchDirL:
folderL = pitchDir.split(os.sep)
midi_pitch = int(folderL[-1])
pitchDir = os.path.join(inDir,pitchDir)
takeDirL = os.listdir(pitchDir)
for takeFolder in takeDirL:
takeDir = os.path.join(pitchDir,takeFolder)
r = rms_analysis_main( takeDir, midi_pitch, **analysisArgsD )
labelFn = os.path.join(takeDir,"audacity.txt")
print("Writing:",labelFn)
with open(labelFn,"w") as f:
for i,s in enumerate(r.statsL):
label = "%i %4.1f %6.1f" % (i, s.pkDb, s.durMs )
f.write("%f\t%f\t%s\n" % ( s.begSmpSec, s.endSmpSec, label ))

4
rt_note_analysis.py
View File

@ -33,7 +33,7 @@ class RT_Analyzer:
anlArgs = types.SimpleNamespace(**anlArgD) anlArgs = types.SimpleNamespace(**anlArgD)
rmsDbV, rms_srate, binHz = audio_harm_rms( audioDev.srate, np.squeeze(sigV), anlArgs.rmsWndMs, anlArgs.rmsHopMs, anlArgs.dbRefWndMs, midi_pitch, anlArgs.harmCandN, anlArgs.harmN ) rmsDbV, rms_srate, binHz = audio_harm_rms( audioDev.srate, np.squeeze(sigV), anlArgs.rmsWndMs, anlArgs.rmsHopMs, anlArgs.dbLinRef, midi_pitch, anlArgs.harmCandN, anlArgs.harmN )
pkIdxL = locate_peak_indexes( rmsDbV, rms_srate, [( begTimeMs, endTimeMs)] ) pkIdxL = locate_peak_indexes( rmsDbV, rms_srate, [( begTimeMs, endTimeMs)] )
@ -46,7 +46,7 @@ class RT_Analyzer:
hm_db = rmsDbV[ pkIdxL[0] ] hm_db = rmsDbV[ pkIdxL[0] ]
tdRmsDbV, rms0_srate = audio_rms( audioDev.srate, np.squeeze(sigV), anlArgs.rmsWndMs, anlArgs.rmsHopMs, anlArgs.dbRefWndMs ) tdRmsDbV, rms0_srate = audio_rms( audioDev.srate, np.squeeze(sigV), anlArgs.rmsWndMs, anlArgs.rmsHopMs, anlArgs.dbLinRef )
tdPkIdxL = locate_peak_indexes( tdRmsDbV, rms0_srate, [( begTimeMs, endTimeMs)] ) tdPkIdxL = locate_peak_indexes( tdRmsDbV, rms0_srate, [( begTimeMs, endTimeMs)] )