import cmsisdsp as dsp
import cmsisdsp.fixedpoint as fix
import numpy as np
import os.path
import glob
import pathlib
import random
import soundfile as sf
import matplotlib.pyplot as plt
from IPython.display import display,Audio,HTML
import scipy.signal
from numpy.lib.stride_tricks import sliding_window_view
from scipy.signal.windows import hann
from sklearn import svm
from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay
from sklearn.model_selection import GridSearchCV
from sklearn.model_selection import RandomizedSearchCV
from scipy.stats import uniform
from sklearn.linear_model import LogisticRegression
import pickle

"""
Explanation: Simple keyword spotting with CMSIS-DSP Python wrapper and Arduino
The goal of this notebook is to demonstrate how to use the CMSIS-DSP Python wrapper on an example which is complex enough.
It is not a state of the art keyword recognition system. The feature used for the machine learning is very simple and just able to recognize the "Yes" keyword.
But it is a good start and enough to demonstrate lots of features of the Python wrapper like:

Testing the CMSIS-DSP algorithm directly in Python
Test of fixed point implementation
Implementation of the compute graph and streaming computation from the CMSIS-DSP Synchronous Data Flow framework
C++ code generation for the compute graph
Final implementation for Arduino Nano 33 BLE

Several Python packages are required. If they are not already installed on you system, you can install them from the notebook by using:
!pip install packagename
The machine learning is using scikit-learn. For scientific computations, we are using SciPy, NumPy and Matplotlib.
For reading wav files, the soundfile package is used.
Other packages will be used below in the notebook and will have to be installed.
End of explanation
"""


MINISPEECH="mini_speech_commands"
commands=np.array([os.path.basename(f) for f in glob.glob(os.path.join(MINISPEECH,"mini_speech_commands","*"))])
commands=commands[commands != "README.md"]
# Any other word will be recognized as unknown
to_keep=['yes']

"""
Explanation: The speech commands
We are using the simplified speech commands from the TensorFlow Lite tutorial.
Those commands can be downloaded from this link: "http://storage.googleapis.com/download.tensorflow.org/data/mini_speech_commands.zip"
Once the zip has been uncompressed, you'll need to change the path of the folder below.
The below code is loading the list of commands available in the mini_speech_commands folder and it is describing the words we want to detect. Here we only want to detect the Yes keyword.
You can add other keywords but the CMSIS-DSP implementation in this notebook is only supporting one keyword.
Nevertheless, if you'd like to experiment with the training of the ML model and different features, then you can work with more commands.
End of explanation
"""


UNKNOWN_CLASS = -1
def get_label(name):
    return(pathlib.PurePath(name).parts[-2])
def get_label_id(name):
    label=get_label(name)
    if label in to_keep:
        return(to_keep.index(label))
    else:
        return(UNKNOWN_CLASS)

"""
Explanation: The below code is generating a label ID for a command. The ID will be -1 for any command not in the to_keep list. Other ID will be the index of the keyword in this list.
End of explanation
"""


def zcr(w):
    w = w-np.mean(w)
    f=w[:-1]
    g=w[1:]
    k=np.count_nonzero(np.logical_and(f*g<0, g>f))
    return(1.0*k/len(f))
        

"""
Explanation: The feature
The feature is based on a simple zero crossing rate (zcr). We choose to only keep the increasing crossing. I don't think it is making a lot of differences for the final performance of the keyword recognition.
The zcr function is computing the zcr for a window of samples.
End of explanation
"""


def feature(data):
    samplerate=16000
    input_len = 16000
    
    # The speech pattern is padded to ensure it has a duration of 1 second
    
    waveform = data[:input_len]
    
    zero_padding = np.zeros(
      16000 - waveform.shape[0],
      dtype=np.float32)
   
   
    signal = np.hstack([waveform, zero_padding])
    
    
    # We decompose the intput signal into overlapping window. And the signal in each window
    # is premultiplied by a Hann window of the right size.
    # Warning : if you change the window duration and audio offset, you'll need to change the value
    # in the scripts used for the scheduling of the compute graph later.
    winDuration=25e-3
    audioOffsetDuration=10e-3
    winLength=int(np.floor(samplerate*winDuration))
    audioOffset=int(np.floor(samplerate*audioOffsetDuration))
    overlap=winLength-audioOffset
    window=hann(winLength,sym=False)
    reta=[zcr(x*window) for x in sliding_window_view(signal,winLength)[::audioOffset,:]]
    
    # The final signal is filtered. We have tested several variations on the feature. This filtering is
    # improving the recognition
    reta=scipy.signal.lfilter(np.ones(10)/10.0,[1],reta)
    return(np.array(reta))

"""
Explanation: The final feature is the zcr computed on a segment of 1 second and filtered. We are using a sliding window and using a Hann window.
End of explanation
"""


class Pattern:
    def __init__(self,p):
        global UNKNOWN_CLASS
        if isinstance(p, str):
            self._isFile=True 
            self._filename=p
            self._label=get_label_id(p)
    
            data, samplerate = sf.read(self._filename)
            self._feature = feature(data)
        else:
            self._isFile=False
            self._noiseLevel=p
            self._label=UNKNOWN_CLASS
            
            noise=np.random.randn(16000)*p
            self._feature=feature(noise)
            
    @property
    def label(self):
        return(self._label)
            
    @property
    def feature(self):
        return(self._feature)
        
        
    # Only useful for plotting
    # The random pattern will be different each time
    @property
    def signal(self):
        if not self._isFile:
            return(np.random.randn(16000)*self._noiseLevel)
        else:
            data, samplerate = sf.read(self._filename)
            return(data)
        

"""
Explanation: The patterns
The below class is representing a Pattern. A pattern can either be a sound file from the TensorFlow Lite examples and in this case we compute the feature and the label id.
The pattern can also be a random white noise. In that case, we also compute the feature and the class id is -1.
Note that when you use the signal property, the speech patterns will return the content of the file but noise patterns will generate a random noise which will thus be different each time.
End of explanation
"""


files_per_command=len(glob.glob(os.path.join(MINISPEECH,"mini_speech_commands",commands[0],"*")))
files_per_command

"""
Explanation: Following code is giving the number of speech samples for each keyword.
It is assuming that all keywords contain the same number of samples.
End of explanation
"""


# Add patterns we want to detect
filenames=[]
for f in to_keep:
    filenames+=glob.glob(os.path.join(MINISPEECH,"mini_speech_commands",f,"*"))

    
random.shuffle(filenames)
    
# Add remaining patterns
remaining_words=list(set(commands)-set(to_keep))
nb_noise=0

remaining=[]
for f in remaining_words:
    remaining+=glob.glob(os.path.join(MINISPEECH,"mini_speech_commands",f,"*"))
    
random.shuffle(remaining)


filenames += remaining[0:files_per_command-nb_noise]

patterns=[Pattern(x) for x in filenames]

for i in range(nb_noise):
    patterns.append(Pattern(np.abs(np.random.rand(1)*0.05)[0]))
    
random.shuffle(patterns)

"""
Explanation: The following code is generating the patterns used for the training of the ML model,.
It is reading patterns for all the words we want to keep (from to_keep list) and it is aggregating all other keywords in the unknown class.
It is also generating some random noise patterns.For the unknown class, the number of patterns will always be files_per_command but some patterns may be noise rather than sound files.
There is some randomization of file names. So each time this code is executed, you'll get patterns in a different order and for the unknown class, which is containing more than files_per_command, you'll get a different subset of those patterns (thr subset will have the right length files_per_command).
Finally the patterns are also randomized so that the split between training and test patterns will select different patterns each time this code is executed.
End of explanation
"""


print(len(patterns))
patterns=np.array(patterns)

nb_patterns = len(patterns)
nb_train= int(np.floor(0.8 * nb_patterns))
nb_tests=nb_patterns-nb_train

train_patterns = patterns[:nb_train]
test_patterns = patterns[-nb_tests:]

"""
Explanation: Below code is extracting the training and test patterns.
This will be used later to generate the array used by scikit learn to train the model.
End of explanation
"""


nbpat=50
data = patterns[nbpat].signal
samplerate=16000
plt.plot(data)
plt.show()

audio=Audio(data=data,rate=samplerate,autoplay=False)
audio

"""
Explanation: Testing on a signal
The following code is displaying a pattern as example.
End of explanation
"""


def get_spectrogram(waveform,fs):
    # Zero-padding for an audio waveform with less than 16,000 samples.
    input_len = 16000
    waveform = waveform[:input_len]
    zero_padding = np.zeros(
      16000 - waveform.shape[0],
      dtype=np.float32)
    mmax=np.max(np.abs(waveform))
    
    equal_length = np.hstack([waveform, zero_padding])
    f, t, Zxx = scipy.signal.stft(equal_length, fs, nperseg=1000)
    plt.pcolormesh(t, f, np.abs(Zxx), vmin=0, vmax=mmax/100, shading='gouraud')
    plt.title('STFT Magnitude')
    plt.ylabel('Frequency [Hz]')
    plt.xlabel('Time [sec]')
    plt.show()

get_spectrogram(data,16000)

"""
Explanation: Simple function to display a spectrogram. It is adapted from a SciPy example.
End of explanation
"""


feat=feature(data)
plt.plot(feat)
plt.show()

"""
Explanation: Display of the feature to compare with the spectrogram.
End of explanation
"""


X=np.array([x.feature for x in train_patterns])
X.shape

y=np.array([x.label for x in train_patterns])
y.shape

y_test = [x.label for x in test_patterns]
X_test = [x.feature for x in test_patterns]

"""
Explanation: Patterns for training
Now we generate the arrays needed to train and test the model.
We have an array of feature : X_array.
An array of label ID : y
and similar arrays for the tests.
End of explanation
"""


distributionsb = dict(C=uniform(loc=1, scale=1000)
                    )
reg = LogisticRegression(penalty="l1", solver="saga", tol=0.1)
clfb=RandomizedSearchCV(reg, distributionsb,random_state=0,n_iter=50).fit(X, y)

"""
Explanation: Logistic Regression
We have chosen to use a simple logistic regression. We are doing a randomized search on the hyperparameter space.
End of explanation
"""


clfb.best_estimator_

"""
Explanation: We are using the best estimator found during the randomized search:
End of explanation
"""


y_pred = clfb.predict(X_test)
labels=["Unknown"] + to_keep
ConfusionMatrixDisplay.from_predictions(y_test, y_pred,display_labels=labels)

"""
Explanation: The confusion matrix is generated from the test patterns to check the behavior of the classifier:
End of explanation
"""


clfb.score(X_test, y_test)

"""
Explanation: We compute the final score. 0.8 is really the minimum acceptable value for this kind of demo.
With the zcr feature, if you try to detect Yes, No, Unknown, you'll get a score of around 0.6 which is very bad.
End of explanation
"""


with open("logistic.pickle","wb") as f:
    s = pickle.dump(clfb,f)

"""
Explanation: We can now save the model so that next time we want to play with the notebook and test the CMSIS-DSP implementation we do not have to retrain the model:
End of explanation
"""


with open("logistic.pickle","rb") as f:
    clfb=pickle.load(f)

"""
Explanation: And we can reload the saved model:
End of explanation
"""


def predict(feat):
    coef=clfb.best_estimator_.coef_
    intercept=clfb.best_estimator_.intercept_
    
    res=np.dot(coef,feat) + intercept
    
    if res<0:
        return(-1)
    else:
        return(0)

"""
Explanation: Reference implementation with Matrix
This is the reference implementation which will be used to build the CMSIS-DSP implementation. We are no more using the scikit-learn predict function instead we are using an implementation of predict using linear algebra. It should give the same results.
End of explanation
"""


y_pred_ref = [predict(x) for x in X_test]
labels=["Unknown"] + to_keep
ConfusionMatrixDisplay.from_predictions(y_test, y_pred_ref,display_labels=labels)

np.count_nonzero(np.equal(y_test,y_pred_ref))/len(y_test)

"""
Explanation: And like in the code above with scikit-learn, we are checking the result with the confusion matrix and the score. It should give the same results:
End of explanation
"""


coef_f32=clfb.best_estimator_.coef_
intercept_f32=clfb.best_estimator_.intercept_

def dsp_zcr(w):
    m = dsp.arm_mean_f32(w)
    m = -m
    w = dsp.arm_offset_f32(w,m)
    f=w[:-1]
    g=w[1:]
    k=np.count_nonzero(np.logical_and(f*g<0, g>f))
    return(1.0*k/len(f))

"""
Explanation: CMSIS-DSP implementation
Now we are ready to implement the code using CMSIS-DSP API. Once we have a running implementation in Python, writing the C code will be easy since the API is the same.
We are testing 3 implementations here : F32, Q31 and Q15.
At the end, we will check that Q15 is giving good enough results and thus can be implemented in C for the Arduino.
F32 Implementation
It will be very similar to the implemenattion above with matrix but will instead use the CMSIS-DSP API.
End of explanation
"""


firf32 = dsp.arm_fir_instance_f32()

def dsp_feature(data):
    samplerate=16000
    input_len = 16000
    
    waveform = data[:input_len]
    
    zero_padding = np.zeros(
      16000 - waveform.shape[0],
      dtype=np.float32)
   
   
    signal = np.hstack([waveform, zero_padding])
    
    
    winDuration=25e-3
    audioOffsetDuration=10e-3
    winLength=int(np.floor(samplerate*winDuration))
    audioOffset=int(np.floor(samplerate*audioOffsetDuration))
    overlap=winLength -audioOffset
    window=hann(winLength,sym=False)
    reta=[dsp_zcr(dsp.arm_mult_f32(x,window)) for x in sliding_window_view(signal,winLength)[::audioOffset,:]]
    
    # Reset state and filter
    # We want to start with a clean filter each time we filter a new feature.
    # So the filter state is reset each time.
    blockSize=98
    numTaps=10
    stateLength = numTaps + blockSize - 1
    dsp.arm_fir_init_f32(firf32,10,np.ones(10)/10.0,np.zeros(stateLength))
    reta=dsp.arm_fir_f32(firf32,reta)
    return(np.array(reta))

"""
Explanation: For the FIR, CMSIS-DSP is using a FIR instance structure and thus we need to define it
End of explanation
"""


feat=dsp_feature(data)
plt.plot(feat)
plt.show()

"""
Explanation: Let's check that the feature is giving the same result as the reference implemenattion using linear algebra.
End of explanation
"""


def dsp_predict(feat):
    
    res=dsp.arm_dot_prod_f32(coef_f32,feat)
    res = res + intercept_f32
    
    if res[0]<0:
        return(-1)
    else:
        return(0)

"""
Explanation: The feature code is working, so now we can implement the predict:
End of explanation
"""


y_pred_ref = [dsp_predict(dsp_feature(x.signal)) for x in test_patterns]
labels=["Unknown"] + to_keep
ConfusionMatrixDisplay.from_predictions(y_test, y_pred_ref,display_labels=labels)

np.count_nonzero(np.equal(y_test,y_pred_ref))/len(y_test)

"""
Explanation: And finally we can check the CMSIS-DSP behavior of the test patterns:
End of explanation
"""


scaled_coef=clfb.best_estimator_.coef_ 
coef_shift=0
while np.max(np.abs(scaled_coef)) > 1:
    scaled_coef = scaled_coef / 2.0 
    coef_shift = coef_shift + 1

coef_q31=fix.toQ31(scaled_coef)

scaled_intercept = clfb.best_estimator_.intercept_ 
intercept_shift = 0
while np.abs(scaled_intercept) > 1:
    scaled_intercept = scaled_intercept / 2.0 
    intercept_shift = intercept_shift + 1
    
intercept_q31=fix.toQ31(scaled_intercept)

"""
Explanation: We are getting very similar results to the reference implementation. Now let's explore fixed point.
Q31 implementation
First thing to do is to convert the F32 values of the ML mode into Q31.
But we need values in [-1,1]. So we rescale those values and keep track of the shift required to restore the original value.
Then, we convert those rescaled values to Q31.
End of explanation
"""


def dsp_zcr_q31(w):
    m = dsp.arm_mean_q31(w)
    # Negate can saturate so we use CMSIS-DSP function which is working on array (and we have a scalar)
    m = dsp.arm_negate_q31(np.array([m]))[0]
    w = dsp.arm_offset_q31(w,m)
    
    f=w[:-1]
    g=w[1:]
    k=np.count_nonzero(np.logical_and(np.logical_or(np.logical_and(f>0,g<0), np.logical_and(f<0,g>0)),g>f))
    
    # k < len(f) so shift should be 0 except when k == len(f)
    # When k==len(f) normally quotient is 0x40000000 and shift 1 and we convert
    # this to 0x7FFFFFF and shift 0
    status,quotient,shift_val=dsp.arm_divide_q31(k,len(f))
    if shift_val==1:
        return(dsp.arm_shift_q31(np.array([quotient]),shift)[0])
    else:
        return(quotient)

firq31 = dsp.arm_fir_instance_q31()

def dsp_feature_q31(data):
    samplerate=16000
    input_len = 16000
    
    waveform = data[:input_len]
    
    zero_padding = np.zeros(
      16000 - waveform.shape[0],
      dtype=np.int32)
   
   
    signal = np.hstack([waveform, zero_padding])
    
    
    winDuration=25e-3
    audioOffsetDuration=10e-3
    winLength=int(np.floor(samplerate*winDuration))
    audioOffset=int(np.floor(samplerate*audioOffsetDuration))
    overlap=winLength-audioOffset
    
    window=fix.toQ31(hann(winLength,sym=False))
    reta=[dsp_zcr_q31(dsp.arm_mult_q31(x,window)) for x in sliding_window_view(signal,winLength)[::audioOffset,:]]
    
    # Reset state and filter
    blockSize=98
    numTaps=10
    stateLength = numTaps + blockSize - 1
    dsp.arm_fir_init_q31(firq31,10,fix.toQ31(np.ones(10)/10.0),np.zeros(stateLength,dtype=np.int32))
    reta=dsp.arm_fir_q31(firq31,reta)
    return(np.array(reta))

"""
Explanation: Now we can implement the zcr and feature in Q31.
End of explanation
"""


feat=fix.Q31toF32(dsp_feature_q31(fix.toQ31(data)))
plt.plot(feat)
plt.show()

"""
Explanation: Let's check the feature on the data to compare with the F32 version and check it is working:
End of explanation
"""


def dsp_predict_q31(feat):
    
    res=dsp.arm_dot_prod_q31(coef_q31,feat)
    
    # Before adding the res and the intercept we need to ensure they are in the same Qx.x format
    # The scaling applied to the coefs and to the intercept is different so we need to scale
    # the intercept to take this into account
    scaled=dsp.arm_shift_q31(np.array([intercept_q31]),intercept_shift-coef_shift)[0]
    # Because dot prod output is in Q16.48
    # and ret is on 64 bits
    scaled = np.int64(scaled) << 17 
    
    res = res + scaled
    
   
    
    if res<0:
        return(-1)
    else:
        return(0)

"""
Explanation: The Q31 feature is very similar to the F32 one so now we can implement the predict:
End of explanation
"""


y_pred_ref = [dsp_predict_q31(dsp_feature_q31(fix.toQ31(x.signal))) for x in test_patterns]
labels=["Unknown"] + to_keep
ConfusionMatrixDisplay.from_predictions(y_test, y_pred_ref,display_labels=labels)

np.count_nonzero(np.equal(y_test,y_pred_ref))/len(y_test)

"""
Explanation: Now we can check the Q31 implementation on the test patterns:
End of explanation
"""


scaled_coef=clfb.best_estimator_.coef_ 
coef_shift=0
while np.max(np.abs(scaled_coef)) > 1:
    scaled_coef = scaled_coef / 2.0 
    coef_shift = coef_shift + 1

coef_q15=fix.toQ15(scaled_coef)

scaled_intercept = clfb.best_estimator_.intercept_ 
intercept_shift = 0
while np.abs(scaled_intercept) > 1:
    scaled_intercept = scaled_intercept / 2.0 
    intercept_shift = intercept_shift + 1
    
intercept_q15=fix.toQ15(scaled_intercept)

def dsp_zcr_q15(w):
    m = dsp.arm_mean_q15(w)
    # Negate can saturate so we use CMSIS-DSP function which is working on array (and we have a scalar)
    m = dsp.arm_negate_q15(np.array([m]))[0]
    w = dsp.arm_offset_q15(w,m)
    
    f=w[:-1]
    g=w[1:]
    k=np.count_nonzero(np.logical_and(np.logical_or(np.logical_and(f>0,g<0), np.logical_and(f<0,g>0)),g>f))
    
    # k < len(f) so shift should be 0 except when k == len(f)
    # When k==len(f) normally quotient is 0x4000 and shift 1 and we convert
    # this to 0x7FFF and shift 0
    status,quotient,shift_val=dsp.arm_divide_q15(k,len(f))
    if shift_val==1:
        return(dsp.arm_shift_q15(np.array([quotient]),shift)[0])
    else:
        return(quotient)

firq15 = dsp.arm_fir_instance_q15()

def dsp_feature_q15(data):
    samplerate=16000
    input_len = 16000
    
    waveform = data[:input_len]
    
    zero_padding = np.zeros(
      16000 - waveform.shape[0],
      dtype=np.int16)
   
   
    signal = np.hstack([waveform, zero_padding])
    
    
    winDuration=25e-3
    audioOffsetDuration=10e-3
    winLength=int(np.floor(samplerate*winDuration))
    audioOffset=int(np.floor(samplerate*audioOffsetDuration))
    overlap=winLength - audioOffset
    
    window=fix.toQ15(hann(winLength,sym=False))
    reta=[dsp_zcr_q15(dsp.arm_mult_q15(x,window)) for x in sliding_window_view(signal,winLength)[::audioOffset,:]]
    
    # Reset state and filter
    blockSize=98
    numTaps=10
    stateLength = numTaps + blockSize - 1
    dsp.arm_fir_init_q15(firq15,10,fix.toQ15(np.ones(10)/10.0),np.zeros(stateLength,dtype=np.int16))
    reta=dsp.arm_fir_q15(firq15,reta)
    return(np.array(reta))

feat=fix.Q15toF32(dsp_feature_q15(fix.toQ15(data)))
plt.plot(feat)
plt.show()

def dsp_predict_q15(feat):
    
    res=dsp.arm_dot_prod_q15(coef_q15,feat)
    
    scaled=dsp.arm_shift_q15(np.array([intercept_q15]),intercept_shift-coef_shift)[0]
    # Because dot prod output is in Q34.30
    # and ret is on 64 bits
    scaled = np.int64(scaled) << 15 
    
    res = res + scaled
    
   
    
    if res<0:
        return(-1)
    else:
        return(0)

y_pred_ref = [dsp_predict_q15(dsp_feature_q15(fix.toQ15(x.signal))) for x in test_patterns]
labels=["Unknown"] + to_keep
ConfusionMatrixDisplay.from_predictions(y_test, y_pred_ref,display_labels=labels)

np.count_nonzero(np.equal(y_test,y_pred_ref))/len(y_test)

"""
Explanation: The score is as good as the F32 implementation.
Q15 Implementation
It is the same as Q31 but using Q15 functions.
End of explanation
"""


from cmsisdsp.sdf.scheduler import *

"""
Explanation: Q15 version is as good as other versions so we are selecting this implementation to run on the Arduino (once it has been converted to C).
Synchronous Data Flow
We are receiving stream of samples but our functions are using buffers.
We have sliding windows which make it more difficult to connect those functions together : we need FIFOs.
So we are going to use the CMSIS-DSP Synchronous Data Flow framework to describe the compute graph, compute the FIFO lengths and generate a static schedule implementing the streaming computation.
End of explanation
"""


class Source(GenericSource):
    def __init__(self,name,inLength):
        GenericSource.__init__(self,name)
        q15Type=CType(Q15)
        self.addOutput("o",q15Type,inLength)

    @property
    def typeName(self):
        return "Source"
    
class Sink(GenericSink):
    def __init__(self,name,outLength):
        GenericSink.__init__(self,name)
        q15Type=CType(Q15)
        self.addInput("i",q15Type,outLength)

    @property
    def typeName(self):
        return "Sink"
    
class Feature(GenericNode):
    def __init__(self,name,inLength):
        GenericNode.__init__(self,name)

        q15Type=CType(Q15)
        self.addInput("i",q15Type,inLength)
        self.addOutput("o",q15Type,1)

    @property
    def typeName(self):
        return "Feature"
    
class FIR(GenericNode):
    def __init__(self,name,inLength,outLength):
        GenericNode.__init__(self,name)

        q15Type=CType(Q15)
        self.addInput("i",q15Type,inLength)
        self.addOutput("o",q15Type,outLength)

    @property
    def typeName(self):
        return "FIR"
    
class KWS(GenericNode):
    def __init__(self,name,inLength):
        GenericNode.__init__(self,name)

        q15Type=CType(Q15)
        self.addInput("i",q15Type,inLength)
        self.addOutput("o",q15Type,1)

    @property
    def typeName(self):
        return "KWS"

"""
Explanation: To describe our compute graph, we need to describe the nodes which are used in this graph.
Each node is described by its inputs and outputs. For each IO, we define the data type and the number of samples read or written on the IO.
We need the following nodes in our system:
* A source to get audio samples from the Arduino PDM driver
* A Sink to generate the messages on the Arduino serial port
* A feature node to compute the feature for a given window (and pre-multipliying with the Hann window)
* A FIR node which is filtering all the features for one second of signal
* A KWS node which is doing the logistic regression
In addition to that we need sliding windows:
* Like in the Python code above we need a sliding window for audio
* After each recognition attempt on a segment of 1 second, we want to slide the recognition window by 0.5 seconds. It can be implemented with a sliding window on the features before the FIR
End of explanation
"""


q15Type=CType(Q15)
FS=16000
winDuration=25e-3
audioOffsetDuration=10e-3
winLength=int(np.floor(FS*winDuration))
audio_input_length=int(np.floor(FS*audioOffsetDuration))
AUDIO_INTERRUPT_LENGTH = audio_input_length

"""
Explanation: We need some parameters. Those parameters need to be coherent with the values defined in the features in the above code.
AUDIO_INTERRUPT_LENGTH is the audio length generated by the source. But it is not the audio length generated by th PDM driver on Arduino. The Arduino implementation of the Source is doing the adaptation as we will see below.
End of explanation
"""


def gen_sched(python_code=True):
    src=Source("src",AUDIO_INTERRUPT_LENGTH)
    # For Python code, the input is a numpy array which is passed
    # as argument of the node
    if python_code:
        src.addVariableArg("input_array")
    sink=Sink("sink",1)

    feature=Feature("feature",winLength)
    feature.addVariableArg("window")

    sliding_audio=SlidingBuffer("audioWin",q15Type,winLength,winLength-audio_input_length)


    FEATURE_LENGTH=98 # for one second
    FEATURE_OVERLAP = 49 # We slide feature by 0.5 second
    sliding_feature=SlidingBuffer("featureWin",q15Type,FEATURE_LENGTH,FEATURE_OVERLAP)
    kws=KWS("kws",FEATURE_LENGTH)
    # Parameters of the ML model used by the node.
    kws.addVariableArg("coef_q15")
    kws.addVariableArg("coef_shift")
    kws.addVariableArg("intercept_q15")
    kws.addVariableArg("intercept_shift")

    fir=FIR("fir",FEATURE_LENGTH,FEATURE_LENGTH)



    # Description of the compute graph
    g = Graph()

    g.connect(src.o, sliding_audio.i)
    g.connect(sliding_audio.o, feature.i)
    g.connect(feature.o, sliding_feature.i)
    g.connect(sliding_feature.o, fir.i)
    g.connect(fir.o, kws.i)
    g.connect(kws.o, sink.i)

  
    # For Python we run for only around 13 seconds of input signal.
    # Without this, it would run forever.
    conf=Configuration()
    if python_code:
        conf.debugLimit=13
    
    # We compute the scheduling
    sched = g.computeSchedule(conf)

    print("Schedule length = %d" % sched.scheduleLength)
    print("Memory usage %d bytes" % sched.memory)

    # We generate the scheduling code for a Python and C++ implementations
    if python_code:
        conf.pyOptionalArgs="input_array,window,coef_q15,coef_shift,intercept_q15,intercept_shift"
        sched.pythoncode(".",config=conf)
        with open("test.dot","w") as f:
             sched.graphviz(f)
    else:
        conf.cOptionalArgs="""const q15_t *window,
        const q15_t *coef_q15,
        const int coef_shift,
        const q15_t intercept_q15,
        const int intercept_shift"""
        conf.memoryOptimization=True
        # When schedule is long
        conf.codeArray=True
        sched.ccode("kws",config=conf)
        with open("kws/test.dot","w") as f:
             sched.graphviz(f)

"""
Explanation: Below function is :
* Defining the compute graph by connecting all the nodes
* Generating a Python implementation of the compute graph and its scheduling
* Generating a C++ implementation of the compute graph and its scheduling
* Generating a graphviz description of the graph
The feature length is hardcoded. So if the sliding window parameters are changed, you'll need to change the values for the FEATURE_LENGTH and FEATURE_OVERLAP.
End of explanation
"""


gen_sched(True)

"""
Explanation: Next line is generating sched.py which is the Python implementation of the compute graph and its static scheduling. This file is describing the FIFOs connecting the nodes and describing how the nodes are scheduled.
You still need to provide an implementation of the nodes. It is available in appnodes.py and it is nearly a copy/paste of the Q15 implementation above.
But it is simpler because the sliding window and the static schedule ensure that each node is run only when enough data is available. So a big part of the control logic has been removed from the nodes.
sched.py is long because the static schedule is long and there are lots of function calls.. When we generate the C++ implementation, we are using an option which is using an array to describe the static schedule. It makes the C++ code much shorter. But having the sequence of function calls can be useful for debugging.
End of explanation
"""


gen_sched(False)

"""
Explanation: Next line is generating the C++ schedule that we will need for the Arduino implementation : kws/scheduler.cpp
End of explanation
"""


from urllib.request import urlopen
import io
import soundfile as sf

test_pattern_url="https://github.com/ARM-software/VHT-SystemModeling/blob/main/EchoCanceller/sounds/yesno.wav?raw=true"
f = urlopen(test_pattern_url)
filedata = f.read()
data, samplerate = sf.read(io.BytesIO(filedata))
if len(data.shape)>1:
    data=data[:,0]

"""
Explanation: Now we'd like to test the Q15 classifier and the static schedule on a real patterns.
We are using the Yes/No pattern from our VHT-SystemModeling example.
Below code is loading the pattern into a NumPy array.
End of explanation
"""


plt.plot(data)
plt.show()

"""
Explanation: Let's plot the signal to check we have the right one:
End of explanation
"""


import sched as s 
from importlib import reload
import appnodes 

appnodes= reload(appnodes)

s = reload(s)

dataQ15=fix.toQ15(data)
windowQ15=fix.toQ15(hann(winLength,sym=False))

nb,error = s.scheduler(dataQ15,windowQ15,coef_q15,coef_shift,intercept_q15,intercept_shift)

"""
Explanation: Now we can run our static schedule on this file.
The reload function are needed when debugging (or implementing) the appnodes.py. Without this, the package would not be reloaded in the notebook.
This code needs some variables evaluated in the Q15 estimator code above.
End of explanation
"""


with open("logistic.pickle","rb") as f:
    clfb=pickle.load(f)

"""
Explanation: The code is working. We are getting more printed Yes than Yes in the pattern because we are sliding by 0.5 second between each recognition and the same word can be recognized several time.
Now we are ready to implement the same on an Arduino.
First we need to generate the parameters of the model. If you have saved the model, you can reload it with the code below:
End of explanation
"""


scaled_coef=clfb.best_estimator_.coef_ 
coef_shift=0
while np.max(np.abs(scaled_coef)) > 1:
    scaled_coef = scaled_coef / 2.0 
    coef_shift = coef_shift + 1

coef_q15=fix.toQ15(scaled_coef)

scaled_intercept = clfb.best_estimator_.intercept_ 
intercept_shift = 0
while np.abs(scaled_intercept) > 1:
    scaled_intercept = scaled_intercept / 2.0 
    intercept_shift = intercept_shift + 1
    
intercept_q15=fix.toQ15(scaled_intercept)

"""
Explanation: Once the model is loaded, we extract the values and convert them to Q15:
End of explanation
"""


def carray(a):
    s="{"
    k=0 
    for x in a:
        s = s + ("%d," % (x,))
        k = k + 1 
        if k == 10:
            k=0;
            s = s + "\n"
    s = s + "}"
    return(s)

ccode="""#include "arm_math.h"
#include "coef.h"


const q15_t fir_coefs[NUMTAPS]=%s;


const q15_t coef_q15[%d]=%s;

const q15_t intercept_q15 = %d;
const int coef_shift=%d;
const int intercept_shift=%d;
 
const q15_t window[%d]=%s;
"""

def gen_coef_code():
    fir_coef = carray(fix.toQ15(np.ones(10)/10.0))
    winq15=carray(fix.toQ15(hann(winLength,sym=False)))
    res = ccode % (fir_coef,
                   len(coef_q15[0]),
                   carray(coef_q15[0]),
                   intercept_q15,
                   coef_shift,
                   intercept_shift,
                   winLength,
                   winq15
                  )
    
    with open(os.path.join("kws","coef.cpp"),"w") as f:
        print(res,file=f)
        

"""
Explanation: Now we need to generate C arrays for the ML model parameters. Those parameters are generated into kws/coef.cpp
End of explanation
"""


gen_coef_code()

"""
Explanation: Generation of the coef code:
End of explanation
"""


!arduino-cli board list

!arduino-cli config init

!arduino-cli lib install Arduino_CMSIS-DSP

"""
Explanation: The implementation of the nodes is in kws/AppNodes.h. It is very similar to the appnodes.py but using the CMSIS-DSP C API.
The C++ template are used only to minimize the overhead at runtime.
Arduino
You need to have the arduino command line tools installed. And they need to be in your PATH so that the notebook can find the tool.
We are using the Arduino Nano 33 BLE
Building and upload
End of explanation
"""


!arduino-cli compile -b arduino:mbed_nano:nano33ble kws

!arduino-cli upload -b arduino:mbed_nano:nano33ble -p COM5 kws

"""
Explanation: The first time the below command is executed, it will take a very long time. The full CMSIS-DSP library has to be rebuilt for the Arduino.
End of explanation
"""


import serial
import ipywidgets as widgets
import time
import threading

STOPSERIAL=False   
def stop_action(btn):
    global STOPSERIAL
    STOPSERIAL=True
    
out = widgets.Output(layout={'border': '1px solid black','height':'40px'})
button = widgets.Button(
    description='Stop',
    disabled=False,
    button_style='', # 'success', 'info', 'warning', 'danger' or ''
    tooltip='Click me'
)
button.on_click(stop_action)

out.clear_output()
display(widgets.VBox([out,button]))

STOPSERIAL = False

def get_serial():
    try:
        with serial.Serial('COM6', 115200, timeout=1) as ser:  
            ser.reset_input_buffer()
            global STOPSERIAL
            while not STOPSERIAL:
                data=ser.readline()
                if (len(data)>0):
                    with out:
                        out.clear_output()
                        res=data.decode('ascii').rstrip()
                        if res=="Yes":
                            display(HTML("<p style='color:#00AA00';>YES</p>"))
                        else:
                            print(res)
            with out:
                out.clear_output()
                print("Communication closed")
    except Exception as inst:
        with out:
            out.clear_output()
            print(inst)
            
t = threading.Thread(target=get_serial)
t.start()


"""
Explanation: Testing
Below code is connecting to the Arduino board an displaying the output in a cell.
If you say Yes loudly enough (so not too far from the board) and if you don't have too much background noise in your room, then it should work.
You need to install the pyserial python package
End of explanation
"""


