Hi all,
Does anyone know how to have access to softmax values at the output of NDP101 on the tinyML board to adjust post paramters?
Thanks,
Saeid
Hi Safavi40
It is better to look at neural network outputs before the softmax as softmax hides some of the information. I will write some of the python scripts to further analyze data.
Atul
Converting csv file data into wav file
import numpy
import scipy.io.wavfile
define fileName in your directory
constructMatrix = numpy.loadtxt(fileName + '.csv', dtype=numpy.int16, delimiter=',')
OutputwaveFilename = fileName + '.wav'
print('length of the data ',int(len(constructMatrix)), ' data snippet ', constructMatrix[0:int(len(constructMatrix))])
sample_rate = 16000
scipy.io.wavfile.write(OutputwaveFilename,sample_rate,constructMatrix[0:int(len(constructMatrix))])
Mixing Audio files
import numpy
import scipy.io.wavfile
import glob
import random
#dataPath = ‘define per your directory’
noiseScalingFactor = 2
transientWindow = 1600
startingPointNoise = 4560 # For randamizing
directoryList = [‘stop’]
#directoryList = [‘go’]
def extractingFileName(inputText): # From complete address, finding the name of the file
lastSlash = 0
for charPosition in range(0,len(inputText)):
if ((inputText[charPosition]==“/”) or (inputText[charPosition]==“\”)):
lastSlash = charPosition
return inputText[(lastSlash+1):(len(inputText))] # returning a portion of the input text
if name == ‘main’:
sampleRate, noiseArray = scipy.io.wavfile.read(‘C:/noise_1.wav’)
counter = 0
for files in glob.glob(dataPath+directoryList[0]+"/" + "*.wav"):
randomStart=random.randint(1600,noiseArray.shape[0]-32000) # Running loop for all files in a directory
print("file counter",counter," ", "File name ", files," random start" ,randomStart)
counter += 1
sampleRate, samplesRawArray = scipy.io.wavfile.read(files)
mixedData = numpy.zeros(samplesRawArray.shape[0],dtype = numpy.int16)
mixedData = samplesRawArray + noiseScalingFactor*noiseArray[randomStart:randomStart+samplesRawArray.shape[0]]
scipy.io.wavfile.write(dataPath+directoryList[0] + '_noise_2x/n_2x_'+extractingFileName(files),16000,mixedData)
import numpy as np
import scipy.io.wavfile
import glob
import matplotlib.pyplot as plt
import shutil
enablePlotting = 0 # set it to zero for disable plotting
copyOrMoveFlag = 1 # set it to 1 to copy in bins and and move into Quarantine
analysisRun = 0 # Running Analysis run after this program has already been run
classifierList = [‘marvin_go_1st_set’]
inputDataPath = ‘"path’
outputDataPath = ‘analysis/’
inputDataPathAnalysis = outputDataPath + classifierList[0] + “/quarantine/”
counter = 0
sample_rate = 16000
def copyOrMove(inputFileName,outputFileName,flag): # flag = 0, do nothing, flag = 1, copy files during development, flag = 2 Move files for final sorting
if (flag==1):
shutil.copy(inputFileName, outputFileName)
if (flag==2):
shutil.move(inputFileName, outputFileName)
def plotCurve(inputMatrix): # Plots an array
if (enablePlotting>0):
fig, ax = plt.subplots()
ax.plot(inputMatrix)
plt.show()
def firstRisingEdge(inputMatrix,threshold): # Finding index of first edge crossing the threshold of a linear matrix
lengthInputMatrix = inputMatrix.shape[0]
indexRisingEdge = 0
if (inputMatrix[0] > threshold): # if very first value is larger than the threshold
return 0
for i in range (lengthInputMatrix-2): # Finding index for which threshold is between two consecuting values
if (inputMatrix[i] <= threshold) and (threshold < inputMatrix[i+1]):
indexRisingEdge = i
break # breaking the loop to prevent same sitruation later in the curve
return indexRisingEdge
def lastFallingEdge(inputMatrix,threshold): # Finding index for which the data cross the threshold towards the end
lengthInputMatrix = inputMatrix.shape[0]
indexFallingEdge = lengthInputMatrix-1
if (inputMatrix[lengthInputMatrix-1] > threshold): # If the last value is larger than the threshold
return lengthInputMatrix
for i in range (1,lengthInputMatrix-1):
if (inputMatrix[lengthInputMatrix-i] <= threshold) and (threshold < inputMatrix[lengthInputMatrix-i-1]):
indexFallingEdge = lengthInputMatrix-i
#print("falling edge matched ", i, lengthInputMatrix-i, inputMatrix[lengthInputMatrix-i], inputMatrix[lengthInputMatrix-i-1] )
break
return indexFallingEdge
def extractingFileName(inputText): # From complete address, finding the name of the file
lastSlash = 0
for charPosition in range(0,len(inputText)):
if ((inputText[charPosition]==“/”) or (inputText[charPosition]==“\”)):
lastSlash = charPosition
return inputText[(lastSlash+1):(len(inputText))] # returning a portion of the input text
def runningWindowAnalysis(inputMatrix): # Running analysis of the wave files
dcOffset = np.average(inputMatrix)
plotCurve(inputMatrix) # Average is same as DC offset
inputMatrix = inputMatrix - dcOffset # Removing DC offset
windowSize = 320 # Window for averaging
plotCurve(inputMatrix)
peakValue = np.max(np.abs(inputMatrix)) # Finding max on input if greater than 32000 then pretty close to saturation
lenAvgMatrix = len(inputMatrix)-windowSize # Length of the remaining matrix
envenlopRunningAverageMatrix = np.zeros(lenAvgMatrix)
for i in range(lenAvgMatrix):
envenlopRunningAverageMatrix[i] = np.average(np.abs(inputMatrix[i:i+windowSize])) # Taking abs and then pass it through running average
noiseFloor = np.min(envenlopRunningAverageMatrix)
plotCurve(envenlopRunningAverageMatrix)
envenlopRunningAverageMatrixWONoiseFloor = np.zeros(lenAvgMatrix)
for i in range(lenAvgMatrix):
if (envenlopRunningAverageMatrix[i]>noiseFloor ):
envenlopRunningAverageMatrixWONoiseFloor[i] = envenlopRunningAverageMatrix[i]-noiseFloor # Baseline is noise floor (WO = With Out)
plotCurve(envenlopRunningAverageMatrixWONoiseFloor)
amplitdue = np.max(envenlopRunningAverageMatrixWONoiseFloor) # Amplitude is average peak value with respect to noise floor
normalizedData = envenlopRunningAverageMatrixWONoiseFloor / amplitdue
plotCurve(normalizedData)
firstRisingEdgeValue = firstRisingEdge(normalizedData,0.35) # Position of the rising edge
lastFallingEdgeValue = lastFallingEdge(normalizedData,0.35) # Position of the falling edge
lengthOfWord = lastFallingEdgeValue - firstRisingEdgeValue # Length of the worde
centerOfWord = (lastFallingEdgeValue + firstRisingEdgeValue)/2 # Center point of the word
flagFail = np.zeros(5, dtype= int)
if (noiseFloor>2048): # Too Noisy background
flagFail[0] = 1
if (amplitdue<256): # Too low volume
flagFail[1] = 1
if (firstRisingEdgeValue < 1): # Left truncation
flagFail[2] = 1
if (lastFallingEdgeValue > lenAvgMatrix-1): # Right truncation
flagFail[3] = 1
if (lengthOfWord < 2000): # Too small word
flagFail[4] = 1
return normalizedData, flagFail, noiseFloor, amplitdue, firstRisingEdgeValue, lastFallingEdgeValue, lengthOfWord, centerOfWord
for classifier in range(len(classifierList)): # loop to process multiple classifiers
print(classifierList[classifier])
counter = 0
if (analysisRun): # Reset values for the analysis run
inputDataPathDirectory = inputDataPathAnalysis + "*.wav"
enablePlotting = 1
copyOrMoveFlag = 0
analysisFileName = "subsequent_analysis.csv"
else:
inputDataPathDirectory = inputDataPath + classifierList[classifier] +"/" + "*.wav"
analysisFileName = "analysis.csv"
f = open(outputDataPath + classifierList[classifier] + "/" + analysisFileName , 'w')
f.write('wave_file,noiseFloor, amplitdue, firstRisingEdgeValue, lastFallingEdgeValue, lengthOfWord, centerOfWord, NoiseFloor, Amplitude, RisingEdge, FallingEdge, WordLength \n')
f.close()
numberOfFiles = 50
lengthOfFiles = 3 # seconds
totalLengthOfAudioFile = numberOfFiles*lengthOfFiles*16000
concatenateData = np.zeros(totalLengthOfAudioFile, dtype=np.int16)
for files in glob.glob(inputDataPathDirectory): # Running loop for all files in a directory
print("file counter",counter," ", "File name ", files)
sampleRate, samplesRawArray = scipy.io.wavfile.read(files)
normalizedData, flagFail, noiseFloor, amplitdue, firstRisingEdgeValue, lastFallingEdgeValue, lengthOfWord, centerOfWord= runningWindowAnalysis(samplesRawArray)
print("flagFail ",flagFail, "noise Floor ", noiseFloor, "amplitude ", amplitdue, "First rising edge ", firstRisingEdgeValue )
print( " First rising edge ", firstRisingEdgeValue, "last rising edge ", lastFallingEdgeValue, " Length of the word ", lengthOfWord, "Center of word ", centerOfWord )
startOfData = int(centerOfWord - (lengthOfFiles*16000)/2)
if (startOfData<0):
startOfData=0
if ((startOfData + (lengthOfFiles*16000) ) > samplesRawArray.shape[0] ):
startOfData = samplesRawArray.shape[0] - (lengthOfFiles*16000)
concatenateData[counter*lengthOfFiles*16000:(counter+1)*lengthOfFiles*16000 ] = samplesRawArray[startOfData:startOfData+lengthOfFiles*16000]
f = open(outputDataPath + classifierList[classifier] + "/" + analysisFileName , 'a')
f.write('{},{},{},{},{},{},{}'.format(files, noiseFloor, amplitdue, firstRisingEdgeValue, lastFallingEdgeValue, lengthOfWord, centerOfWord ))
for flags in range(flagFail.shape[0]):
f.write(',{}'.format(flagFail[flags]))
f.write('\n')
f.close()
counter += 1
if (copyOrMoveFlag): # Duplication files in the error bin files and then moving all the error files into the quarantine directory
if (flagFail[0]):
copyOrMove(files, outputDataPath + classifierList[classifier] + "/highNoiseFloor/" + extractingFileName(files),1 )
if (flagFail[1]):
copyOrMove(files, outputDataPath + classifierList[classifier] + "/tooLowAmplitude/" + extractingFileName(files),1 )
if (flagFail[2]):
copyOrMove(files, outputDataPath + classifierList[classifier] + "/startTruncation/" + extractingFileName(files),1 )
if (flagFail[3]):
copyOrMove(files, outputDataPath + classifierList[classifier] + "/endTrunctation/" + extractingFileName(files),1 )
if (flagFail[4]):
copyOrMove(files, outputDataPath + classifierList[classifier] + "/tooSmallWord/" + extractingFileName(files),1 )
if (np.sum(flagFail)>0):
copyOrMove(files, outputDataPath + classifierList[classifier] + "/quarantine/" + extractingFileName(files),2 )
contanateFileName = outputDataPath + classifierList[classifier] + "/concatenate.wav"
scipy.io.wavfile.write(contanateFileName,16000,concatenateData)
import numpy as np
import glob
import matplotlib.pyplot as plt
Setting for pyth
458on script
enablePlotting = 0 # set it to zero for disable plotting
analysisRun = 0 # set it to zero for disable plotting
softMaxClassifiersIncludingOpenSet = 4
windowSize = [39, 10, 29]
#windowSizeRunningMax = [40, 40, 40]
#thresholdSetting = [0.3, 0.3, 0.3]
thresholdSetting = [0.5, 0.5, 0.5]
lowerThresholdSettings = [0.1, 0.1, 0.1]
MarvinGoDelayMin = 31 # 31
MarvinGoDelayMax = 68 # 68
MarvinStopDelayMin = 28 # 28
MarvinStopDelayMax = 55 # 55
GoClassifier = 0
MarvinClassifier = 1
StopClassifier = 2
deadZone = 60
#inputDataPath = ‘marvin_go_fifty_no_noise/’
#inputDataPath = ‘Sept9_2022_pbs_testing_baseline/’
inputDataPath = ‘/marvin_stop_fifty_no_noise/’
outputDataPath = inputDataPath
counter = 0
def plotCurve(inputMatrix): # Plots an array
if (enablePlotting>0):
fig, ax = plt.subplots()
plt.ylim( [ 0, 70000 ] )
#ax.plot(inputMatrix[0:700,:])
ax.plot(inputMatrix)
plt.show()
def centerOfGravity(inputMatrix): # Finding Center of Gravity
lengthInputMatrix = inputMatrix.shape[0]
weightedSum = 0
for i in range (lengthInputMatrix): #
weightedSum = i*inputMatrix[i] + weightedSum
#print(i,inputMatrix[i],weightedSum )
#print(“Weighted sum and grand sum”, weightedSum, np.sum(inputMatrix))
centerOfGravity = 0
if (np.sum(inputMatrix)>0):
centerOfGravity = weightedSum/np.sum(inputMatrix)
return int(centerOfGravity)
def extractingFileName(inputText): # From complete address, finding the name of the file
lastSlash = 0
for charPosition in range(0,len(inputText)):
if ((inputText[charPosition]==“/”) or (inputText[charPosition]==“\”)):
lastSlash = charPosition
return inputText[(lastSlash+1):(len(inputText))] # returning a portion of the input text
def runningAverage(inputMatrix,windowSize):
maxWindowSize = np.max(windowSize)
lenAvgMatrix = inputMatrix.shape[0]-maxWindowSize
runningAverageMatrix = np.zeros( (inputMatrix.shape[0],softMaxClassifiersIncludingOpenSet-1) )
for classifier in range(softMaxClassifiersIncludingOpenSet-1):
for i in range(maxWindowSize,inputMatrix.shape[0]):
runningAverageMatrix[i,classifier] = np.average(inputMatrix[i-windowSize[classifier]:i,classifier])
if (enablePlotting>0):
plotCurve(inputMatrix)
plotCurve(runningAverageMatrix)
return runningAverageMatrix
def countLumps(inputMatrix, expectedLumps, upperThreshold, lowerThreshold):
countMatrix = np.zeros(softMaxClassifiersIncludingOpenSet-1,dtype=int)
foundLump = np.zeros(softMaxClassifiersIncludingOpenSet-1,dtype=int)
lumpStatistics = np.zeros( (expectedLumps, softMaxClassifiersIncludingOpenSet-1,4),dtype=int) # start, stop and centerOfGravity
for i in range(inputMatrix.shape[0]):
for classifier in range(inputMatrix.shape[1]):
if (foundLump[classifier]>0):
if (inputMatrix[i,classifier]<lowerThreshold[classifier]*65536):
foundLump[classifier] = 0
lumpStatistics[ (countMatrix[classifier]-1 ), classifier, 1] = i
startOfLump = lumpStatistics[ (countMatrix[classifier]-1 ), classifier, 0]
lumpStatistics[ (countMatrix[classifier]-1 ), classifier, 2] = startOfLump + centerOfGravity( inputMatrix[startOfLump: i, classifier])
lumpStatistics[ (countMatrix[classifier]-1 ), classifier, 3] = np.max( inputMatrix[startOfLump: i, classifier])
else:
if (inputMatrix[i,classifier]>upperThreshold[classifier]*65536):
foundLump[classifier] = 1
countMatrix[classifier] += 1
lumpStatistics[ (countMatrix[classifier]-1 ), classifier, 0] = i
return countMatrix, lumpStatistics
def countPhraseOccurance(inputMatrix, upperThresholdInput):
phraseResultMatrix = np.zeros((inputMatrix.shape[0],2),dtype = int)
martinGoCounter = 0
martinStopCounter = 0
print(len(upperThresholdInput))
upperThreshold = np.zeros(len(upperThresholdInput))
for i in range(len(upperThresholdInput)):
upperThreshold[i] = 65536*upperThresholdInput[i]
print(upperThreshold)
counterAfterFoundMarvin = 0
lastClassifierTriggered = 0
for i in range(inputMatrix.shape[0]):
counterAfterFoundMarvin = counterAfterFoundMarvin + 1
lastClassifierTriggered = lastClassifierTriggered - 1
if (inputMatrix[i,MarvinClassifier]>upperThreshold[MarvinClassifier]):
counterAfterFoundMarvin = 0
#print(“Marvin triggered”)
if (inputMatrix[i,GoClassifier]>upperThreshold[GoClassifier]):
#if (martinGoCounter<3):
#print(“go triggered”, inputMatrix[i,GoClassifier])
if (counterAfterFoundMarvin >= MarvinGoDelayMin ) and (counterAfterFoundMarvin < MarvinGoDelayMax ):
#print(" Marvin go found", counterAfterFoundMarvin, “at”, i)
if (lastClassifierTriggered<0):
#print(“Marvin Go triggered”)
martinGoCounter += 1
phraseResultMatrix[i,0] = 65000
lastClassifierTriggered = deadZone
#f.write(‘,MarvinGo,{}’.format(i))
if (inputMatrix[i,StopClassifier]>upperThreshold[StopClassifier]):
#print(“stop triggered”)
if (counterAfterFoundMarvin >= MarvinStopDelayMin ) and (counterAfterFoundMarvin < MarvinStopDelayMax ):
if (lastClassifierTriggered<0):
#print(“Marvin stop triggered”)
martinStopCounter += 1
phraseResultMatrix[i,1] = 65000
lastClassifierTriggered = deadZone
return martinGoCounter, martinStopCounter, phraseResultMatrix
def createAnalysisFile():
f = open(outputDataPath+‘softmax_analysis_results.csv’ , ‘w’)
f.write(‘fileName’)
for index in range(softMaxClassifiersIncludingOpenSet-1):
f.write(‘,unaveraged_{}’.format(index))
for index in range(softMaxClassifiersIncludingOpenSet-1):
f.write(‘,averaged_{}’.format(index))
f.write(‘\n’)
f.close()
def createRunningWindowSweepFile():
f = open(outputDataPath+‘runningWindow_sweep.csv’ , ‘w’)
f.write(‘running_window’)
for index in range(softMaxClassifiersIncludingOpenSet-1):
f.write(‘,averaged_{}’.format(index))
f.write(‘\n’)
f.close()
def createDelaySavingFile():
f = open(outputDataPath+‘delaySaving.csv’ , ‘w’)
f.write(‘running_window’)
for index in range(softMaxClassifiersIncludingOpenSet-1):
f.write(‘,classifier{}_start,classifier{}_end,classifier{}_centerGtavity,classifier{}_peak’.format(index,index,index,index))
f.write(‘\n’)
f.close()
def populateOneRow(fileName, unaveragedResults, avergedResults):
f = open(outputDataPath+‘softmax_analysis_results.csv’, ‘a’)
f.write(‘{}’.format(fileName))
for index in range(unaveragedResults.shape[0]):
f.write(‘,{}’.format(unaveragedResults[index]))
for index in range( avergedResults.shape[0]):
f.write(‘,{}’.format( avergedResults[index]))
f.write(‘\n’)
f.close()
def populateOneRowRunningWindow(windowSizeIndex, avergedResults):
f = open(outputDataPath+‘runningWindow_sweep.csv’, ‘a’)
f.write(‘{}’.format(windowSizeIndex))
for index in range( avergedResults.shape[0]):
f.write(‘,{}’.format( avergedResults[index]))
f.write(‘\n’)
f.close()
def populateDelaySavingFile(lumpStatistics):
f = open(outputDataPath+‘delaySaving.csv’, ‘a’)
for i in range(lumpStatistics.shape[0]):
for classifiers in range(lumpStatistics.shape[1]):
for index in range(lumpStatistics.shape[2]):
f.write(‘,{}’.format( lumpStatistics[i,classifiers,index]))
f.write(‘\n’)
f.close()
inputDataPathDirectory = inputDataPath +“/” + “softmax_NN_*.csv”
counter = 0
#createAnalysisFile()
createRunningWindowSweepFile()
#sweepLimit = 80
sweepLimit = 1
for windowSizeIndex in range(sweepLimit ):
print(“window size”, windowSizeIndex)
windowSizeSweepResults = np.zeros(softMaxClassifiersIncludingOpenSet-1, dtype = int)
#windowSize = [windowSizeIndex+1, windowSizeIndex+1, windowSizeIndex+1]
grandMarvinGoCounter = 0
grandMarvinStopCounter = 0
for files in glob.glob(inputDataPathDirectory): # Running loop for all files in a directory
print(“file counter”,counter," ", "File name ", files)
samplesRawArray = np.loadtxt(files,delimiter=',') # removing first column of pointer
samplesArray = samplesRawArray[:,1:softMaxClassifiersIncludingOpenSet]
unaveragedAboveThresold = np.zeros(softMaxClassifiersIncludingOpenSet-1, dtype=int)
for i in range(samplesArray.shape[0]):
for classifier in range(softMaxClassifiersIncludingOpenSet-1):
if (samplesArray[i,classifier]>(thresholdSetting[classifier]*65536)):
#print(samplesArray[i,classifier],(thresholdSetting[classifier]*65536))
unaveragedAboveThresold[classifier] += 1
print("unaveraged stats")
print("total inferences", samplesArray.shape[0], "classifier above threshol", thresholdSetting, unaveragedAboveThresold )
runningAverageMatrix = runningAverage(samplesArray,windowSize)
np.savetxt(outputDataPath + "runnivAverage_" + extractingFileName(files), runningAverageMatrix, delimiter=',' )
windowSizeSweepResultsNew, lumpStatistics= countLumps(runningAverageMatrix ,200,thresholdSetting,lowerThresholdSettings)
#print(lumpStatistics)
createDelaySavingFile()
populateDelaySavingFile(lumpStatistics)
print("count lumps matrix :", windowSizeSweepResultsNew )
windowSizeSweepResults = windowSizeSweepResults + windowSizeSweepResultsNew
#populateOneRow(files, unaveragedAboveThresold, windowSizeSweepResults)
print(thresholdSetting)
martinGoCounter, martinStopCounter, phraseResultMatrix = countPhraseOccurance(runningAverageMatrix, thresholdSetting)
np.savetxt(outputDataPath + "phraseResults_" + extractingFileName(files), phraseResultMatrix, delimiter=',' )
grandMarvinGoCounter = grandMarvinGoCounter + martinGoCounter
grandMarvinStopCounter = grandMarvinStopCounter + martinStopCounter
print("Martin go counter ", martinGoCounter, "martin stop counter ", martinStopCounter)
populateOneRowRunningWindow(windowSizeIndex, windowSizeSweepResults)
print("Martin go counter ", grandMarvinGoCounter, "martin stop counter ", grandMarvinStopCounter)
enablePlotting = 1 # set it to zero for disable plotting
softMaxClassifiersIncludingOpenSet = 4
samplesPerInference = 384
inputDataPath = ‘/marvin_stop_fifty_no_noise’
outputDataPath = inputDataPath
counter = 0
def plotCurve(inputMatrix): # Plots an array
if (enablePlotting>0):
fig, ax = plt.subplots()
plt.ylim( [ 0, 65000 ] )
ax.plot(inputMatrix)
plt.show()
def extractingFileName(inputText): # From complete address, finding the name of the file
lastSlash = 0
for charPosition in range(0,len(inputText)):
if ((inputText[charPosition]==“/”) or (inputText[charPosition]==“\”)):
lastSlash = charPosition
return inputText[(lastSlash+1):(len(inputText))] # returning a portion of the input text
counter = 0
inputDataPathDirectory = inputDataPath +“/” + “.csv"
for files in glob.glob(inputDataPathDirectory): # Running loop for all files in a directory
print(“file counter”,counter," ", "File name ", files)
counter += 1
samplesRawArray = np.loadtxt(files,delimiter=‘,’)
print(samplesRawArray.shape)
totalSamples = samplesRawArray[samplesRawArray.shape[0]-1,0]
print(int(totalSamples/samplesPerInference))
outputVector = np.zeros( (int(totalSamples/samplesPerInference),softMaxClassifiersIncludingOpenSet+1),dtype = int)
counter = 0
for index in range(samplesRawArray.shape[0]):
if (countersamplesPerInference < samplesRawArray[index,0] ):
if (int(totalSamples/samplesPerInference)>counter):
outputVector[counter,0] = countersamplesPerInference
for column in range(softMaxClassifiersIncludingOpenSet):
value = 65536np.exp(samplesRawArray[index,1+column]/32)/np.sum(np.exp(samplesRawArray[index,1:1+softMaxClassifiersIncludingOpenSet]/32))
outputVector[counter,1+column] = value
counter += 1
np.savetxt(outputDataPath +”/softmax_" + extractingFileName(files) ,outputVector,delimiter=‘,’ )