text_plain_output_1.png
from collections import Counter

from keras.callbacks import History, EarlyStopping

from keras.utils import Sequence

from tensorflow.python.client import device_lib

import keras

import numpy as np

import os

import pywt

import soundfile as sf

import tensorflow as tf

import os
from time import perf_counter
import numpy as np
import soundfile as sf
from collections import Counter
import matplotlib.pyplot as plt
from tensorflow.python.client import device_lib
import tensorflow as tf
import keras
from keras.layers import Conv1D, BatchNormalization, Dense, Flatten, Activation, Dropout
from tensorflow.keras.layers.experimental import preprocessing
from keras.models import Sequential
from keras.callbacks import History, EarlyStopping
history = History()
generator_init = False
model_saved = False
checkpoint_saved = False
SAVED_MODEL_PATH = './'
MODEL_NAME = SAVED_MODEL_PATH + 'langid_model'
CHECKPOINT_FILEPATH = MODEL_NAME + '_CP'
USE_OVERFITTING_ORIGINAL_NW = 1
USE_DROPOUT_NO_REGULARIZATION_NW = 2
USE_DROPOUT_REGULARIZATION_NW = 3
nn_choice = USE_DROPOUT_REGULARIZATION_NW
WAVELET = 'bior6.8'
train_path = '../input/spoken-language-identification/train/train/'
test_path = '../input/spoken-language-identification/test/test/'

from keras.utils import Sequence
import pywt
import pdb
class langidDataGenerator(Sequence):
    """Generates data for Keras"""
    def __init__(self, list_IDs, labels, wavelet='rbio3.1', drop_levels=None, batch_size=32, n_channels=1, n_classes=6, shuffle=True):
        """Initialization"""
        self.wvlt = wavelet
        self.drop_lvls = slice(0, -drop_levels)
        self.batch_size = batch_size
        self.labels = labels
        self.list_IDs = list_IDs
        self.n_channels = n_channels
        self.n_classes = n_classes
        self.shuffle = shuffle
        self.on_epoch_end()
        X, y = self.__data_generation(self.list_IDs[:2])
        X = np.expand_dims(X, 2)
        self.dim = X.shape[1:]
    def __len__(self):
        """Denotes the number of batches per epoch"""
        return int(np.floor(len(self.list_IDs) / self.batch_size))
    def __getitem__(self, index):
        """Generate one batch of data"""
        indexes = self.indexes[index * self.batch_size:(index + 1) * self.batch_size]
        list_IDs_temp = [self.list_IDs[k] for k in indexes]
        X, y = self.__data_generation(list_IDs_temp)
        X = tf.expand_dims(X, 2)
        return (X, y)
    def on_epoch_end(self):
        """Updates indexes after each epoch"""
        self.indexes = tf.range(len(self.list_IDs))
        if self.shuffle == True:
            tf.random.shuffle(self.indexes)
    def wavelet_features(self, list_IDs_temp):
        features = []
        y = []
        for ID in list_IDs_temp:
            signal, fs = sf.read(ID)
            list_coeff = pywt.wavedec(signal, self.wvlt, mode='per')
            dwt_local_coeff = []
            end_flag = 0
            for coeff in list_coeff[self.drop_lvls]:
                dwt_local_coeff.extend(coeff)
            features.append(dwt_local_coeff)
            y.append(self.labels[ID])
        X = tf.convert_to_tensor(features)
        return (X, y)
    def __data_generation(self, list_IDs_temp):
        """Generates data containing batch_size samples"""
        X, y = self.wavelet_features(list_IDs_temp)
        return (X, keras.utils.to_categorical(y, num_classes=self.n_classes))

"""
DATA_FOLDER = '../datasets/langid/'
train_path = DATA_FOLDER+'test/'
test_path = DATA_FOLDER+'test/'
"""
train_labels = []
for filename in os.listdir(train_path):
    train_labels.append(filename[:4])
test_labels = []
for filename in os.listdir(test_path):
    test_labels.append(filename[:4])
lb = 0
labeld = {}
for k in Counter(train_labels).keys():
    labeld[k] = lb
    lb = lb + 1
num_classes = lb
train_files = []
train_labels2 = {}
for filename in os.listdir(train_path):
    train_files.append(train_path + filename)
    train_labels2[train_path + filename] = labeld[filename[:4]]
test_files = []
test_labels2 = {}
for filename in os.listdir(test_path):
    test_files.append(test_path + filename)
    test_labels2[test_path + filename] = labeld[filename[:4]]

sig, f = sf.read(train_files[0])
siglen = len(sig)
print('Signal length is {}, sampling frequency {}'.format(siglen, f))