<jupyter_start><jupyter_text>Two Phase Predictions Design PatternThe Two Phased Prediction design pattern provides a way to address the problem of keeping models for specific use cases sophisticated and performant when they have to be deployed onto distributed devices. We'll use this [Kaggle environmental sound dataset](https://www.kaggle.com/mmoreaux/environmental-sound-classification-50?) to build a two phase model:* Phase 1 (on-device): is it an instrument?* Phase 2 (cloud): which instrument is it?To build this solution, both of our models will be trained on audio [spectrograms](https://en.wikipedia.org/wiki/Spectrogram) -- image representations of audio.<jupyter_code>import numpy as np
import pandas as pd
import tensorflow_hub as hub
import tensorflow as tf

import os
import pathlib
import matplotlib.pyplot as plt
from scipy import signal
from scipy.io import wavfile<jupyter_output><empty_output><jupyter_text>Data pre-processingFirst we'll take a look at a CSV file with the audio filenames and their respective labels. Then we'll add a column for the label we'll be using in the first model to determine whether the sound is an instrument or not.All of the data we'll be using to train the models has been made publicly available in a GCS bucket. Inspect the labels<jupyter_code># Copy the label lookup file from GCS
!gsutil cp gs://ml-design-patterns/audio-train.csv .
label_data = pd.read_csv('audio-train.csv')
label_data.head()<jupyter_output><empty_output><jupyter_text>Add a column for our first model: whether the sound is an instrument or not<jupyter_code>instrument_labels = ['Cello', 'Clarinet', 'Double_bass', 'Saxophone', 'Violin_or_fiddle', 'Snare_drum', 'Hi-hat', 'Flute', 'Bass_drum', 'Trumpet', 'Acoustic_guitar', 'Oboe', 'Gong', 'Tambourine', 'Cowbell', 'Harmonica', 'Electric_piano', 'Glockenspiel']
def check_instrument(row):
  if row['label'] in instrument_labels:
    return 1
  else:
    return 0
label_data['is_instrument'] = label_data.apply(check_instrument, axis=1)
label_data.head()
label_data['is_instrument'].value_counts()<jupyter_output><empty_output><jupyter_text>To ensure quality, we'll only use manually verified samples from the dataset.<jupyter_code>verified = label_data.loc[label_data['manually_verified'] == 1]
verified['is_instrument'].value_counts()
verified.head()<jupyter_output><empty_output><jupyter_text>Preview the spectrogram for a sample trumpet sound from our training dataset<jupyter_code>!gsutil cp gs://ml-design-patterns/audio_train/001ca53d.wav .
sample_rate, samples = wavfile.read('001ca53d.wav')
freq, times, spectro = signal.spectrogram(samples, sample_rate)
plt.figure()
fig = plt.gcf()
plt.axis('off')
plt.pcolormesh(times, freq, np.log(spectro))
plt.show()<jupyter_output><empty_output><jupyter_text>Download all the spectrogram data<jupyter_code># This might take a few minutes
!gsutil -m cp -r gs://ml-design-patterns/audio_train_spectro .<jupyter_output><empty_output><jupyter_text>Phase 1: build the offline-optimized binary classification modelFirst we'll move the images into labeled directories since the Keras `ImageDataGenerator` class uses this format (directories with names corresponding to labels) to read our data. Then we'll create our training and validation batches and train a model using the MobileNet V2 architecture with frozen weights from ImageNet.<jupyter_code>!mkdir audio_spectros
!mkdir audio_spectros/not_instrument
!mkdir audio_spectros/instrument
keys = verified['fname'].values
vals = verified['is_instrument'].values
label_lookup = dict(zip(keys, vals))
for i in os.listdir(os.getcwd() + '/audio_train_spectro'):
  id = i.split('.')[0] + '.wav'
  is_instra = label_lookup[id]
  im_path = os.getcwd() + '/audio_train_spectro/' + i
  if is_instra == 0:
    !mv $im_path audio_spectros/not_instrument/ 
  else:
    !mv $im_path audio_spectros/instrument/<jupyter_output><empty_output><jupyter_text>Load the images as a tf dataset<jupyter_code>data_dir = pathlib.Path(os.getcwd() + '/audio')
class_names = ['not_instrument', 'instrument']
BATCH_SIZE = 64
IMG_HEIGHT = 128
IMG_WIDTH = 128
STEPS_PER_EPOCH = np.ceil(3700/BATCH_SIZE)
image_generator = tf.keras.preprocessing.image.ImageDataGenerator(rescale=1./255, validation_split=0.1)
train_data_gen = image_generator.flow_from_directory(directory=data_dir,
                                                     batch_size=BATCH_SIZE,
                                                     shuffle=True,
                                                     target_size=(IMG_HEIGHT, IMG_WIDTH),
                                                     classes = class_names,
                                                     class_mode='binary')
val_data_gen = image_generator.flow_from_directory(directory=data_dir,
                                                     batch_size=BATCH_SIZE,
                                                     shuffle=True,
                                                     target_size=(IMG_HEIGHT, IMG_WIDTH),
                                                     classes = class_names,
                                                     class_mode='binary',
                                                    subset='validation')
image_count = len(list(data_dir.glob('*/*.png')))
print(image_count)
instrument_modellabel_batch = next(train_data_gen)
val_image, val_label = next(val_data_gen)<jupyter_output><empty_output><jupyter_text>Build a model for binary classification: is it an instrument or not? We'll use **transfer learning** for this, by loading the MobileNet V2 model architecture trained on the ImageNet dataset and then adding a few additional layers on top specific to our prediction task.<jupyter_code>mobilenet = tf.keras.applications.MobileNetV2(
    input_shape=((128,128,3)), 
    include_top=False,
    weights='imagenet'
)
mobilenet.trainable = False
feature_batch = mobilenet(image_batch)
global_avg_layer = tf.keras.layers.GlobalAveragePooling2D()
feature_batch_avg = global_avg_layer(feature_batch)
print(feature_batch_avg.shape)
prediction_layer = tf.keras.layers.Dense(1, activation='sigmoid')
prediction_batch = prediction_layer(feature_batch_avg)
print(prediction_batch.shape)
model = tf.keras.Sequential([
  mobilenet,
  global_avg_layer,
  prediction_layer                            
])
model.summary()
model.compile(optimizer='SGD',
              loss='binary_crossentropy',
              metrics=['accuracy'])<jupyter_output><empty_output><jupyter_text>*Note: we could make changes to our model architecture, perform progressive fine-tuning to find the optimal number of layers to fine-tune, or employ hyperparameter tuning to improve model accuracy. Here the focus is on tooling and process rather than accuracy.*<jupyter_code>model.fit_generator(train_data_gen, 
                    validation_data=val_data_gen,
                    steps_per_epoch=STEPS_PER_EPOCH, epochs=10)<jupyter_output><empty_output><jupyter_text>Convert model to TFLite and quantizeSave the TF Lite model file and get predictions on it using Python.<jupyter_code>converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_model = converter.convert()
open('converted_model.tflite', 'wb').write(tflite_model)<jupyter_output><empty_output><jupyter_text>Get a prediction on one spectrogram from our validation dataset, print the model's prediction (sigmoid probability from 0 to 1) along with ground truth label. For more details, check out the TF Lite [inference docs](https://www.tensorflow.org/lite/guide/inference).<jupyter_code># Load TFLite model and allocate tensors.
interpreter = tf.lite.Interpreter(model_path="converted_model.tflite")
interpreter.allocate_tensors()

# Get input and output tensors.
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()

# Test model on random input data.
input_shape = input_details[0]['shape']
input_data = np.array([val_image[0]], dtype=np.float32)
interpreter.set_tensor(input_details[0]['index'], input_data)

interpreter.invoke()

# The function `get_tensor()` returns a copy of the tensor data.
# Use `tensor()` in order to get a pointer to the tensor.
output_data = interpreter.get_tensor(output_details[0]['index'])
print(output_data)
print(val_label[0])
input_details<jupyter_output><empty_output><jupyter_text>Phase 2: identifying instrument soundsTrain a multilabel classification model to predict the instrument associated with a given instrument sound.<jupyter_code>instrument_data = verified.loc[verified['is_instrument'] == 1]
instrument_data.head()
instrument_data['label'].value_counts()
inst_keys = instrument_data['fname'].values
inst_vals = instrument_data['label'].values
instrument_label_lookup = dict(zip(inst_keys, inst_vals))
!mkdir instruments
for i in instrument_labels:
    path = os.getcwd() + '/instruments/' + i
    !mkdir $path<jupyter_output><empty_output><jupyter_text>Create directories for each instrument label. We'll use this later when we load our images with Keras's `ImageDataGenerator` class.<jupyter_code>for i in os.listdir(os.getcwd() + '/audio_train_spectro'):
    id = i.split('.')[0] + '.wav'
    try:
        instrument_name = instrument_label_lookup[id]
        im_path = os.getcwd() + '/audio_train_spectro/' + i
        new_path = os.getcwd() + '/instruments/' + instrument_name + '/' + i
        !mv $im_path $new_path
    except:
        pass
instrument_image_generator = tf.keras.preprocessing.image.ImageDataGenerator(rescale=1./255, validation_split=0.1)
BATCH_SIZE = 256
IMG_HEIGHT = 128
IMG_WIDTH = 128
STEPS_PER_EPOCH = np.ceil(2002/BATCH_SIZE)
train_data_instrument = instrument_image_generator.flow_from_directory(directory=os.getcwd() + '/instruments',
                                                     batch_size=BATCH_SIZE,
                                                     shuffle=True,
                                                     target_size=(IMG_HEIGHT, IMG_WIDTH),
                                                     classes = instrument_labels)
val_data_instrument = instrument_image_generator.flow_from_directory(directory=os.getcwd() + '/instruments',
                                                     batch_size=BATCH_SIZE,
                                                     shuffle=False,
                                                     target_size=(IMG_HEIGHT, IMG_WIDTH),
                                                     classes = instrument_labels,
                                                     subset='validation')
image_instrument_train, label_instrument_train = next(train_data_instrument)
image_instrument_val, label_instrument_val = next(val_data_instrument)
vgg_model = tf.keras.applications.VGG19(
    include_top=False, 
    weights='imagenet', 
    input_shape=((128,128,3))
)
vgg_model.trainable = False
feature_batch = vgg_model(image_batch)
global_avg_layer = tf.keras.layers.GlobalAveragePooling2D()
feature_batch_avg = global_avg_layer(feature_batch)
print(feature_batch_avg.shape)
prediction_layer = tf.keras.layers.Dense(len(instrument_labels), activation='softmax')
prediction_batch = prediction_layer(feature_batch_avg)
print(prediction_batch.shape)
instrument_model = tf.keras.Sequential([
    vgg_model,
    global_avg_layer,
    prediction_layer
])
instrument_model.summary()
instrument_model.compile(optimizer='adam',
              loss=tf.keras.losses.CategoricalCrossentropy(from_logits=True),
              metrics=['accuracy'])<jupyter_output><empty_output><jupyter_text>*Note: we could make changes to our model architecture, perform progressive fine-tuning to find the optimal number of layers to fine-tune, or employ hyperparameter tuning to improve model accuracy. Here the focus is on tooling and process rather than accuracy.*<jupyter_code>instrument_model.fit_generator(
    train_data_instrument, 
    validation_data=val_data_instrument,
    steps_per_epoch=STEPS_PER_EPOCH, epochs=10)<jupyter_output><empty_output><jupyter_text>Get a test prediction on the instrument model<jupyter_code>test_pred = instrument_model.predict(np.array([image_instrument_val[0]]))
predicted_index = np.argmax(test_pred)
confidence = test_pred[0][predicted_index]
test_pred[0]
print('Predicted instrument: ', instrument_labels[predicted_index], round(confidence * 100), '% confidence')
print('Actual instrument: ', instrument_labels[np.argmax(label_instrument_val[0])])<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.0925332478769428, 'fraction_numerical': 0.00825188271110399, 'mean_word_length': 3.8780773739742087, 'pattern_counts': {'":': 0, '<': 65, '<?xml version=': 0, '>': 65, 'https://': 3, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 41, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '16', 'dir': 'jupyter-structured-clean-dedup', 'id': '440673', 'license': 'Apache-2.0', 'n_tokens_mistral': 3763, 'n_tokens_neox': 3633, 'n_words': 877, 'path': '05_resilience/two_phase_predictions.ipynb', 'repo_name': 'modoai/ml-design-patterns', 'size': '12482'}