<jupyter_start><jupyter_text>Практика №3Теперь в качестве акустической модели мы будем использовать простой DNN классификатор. На каждом кадре записи мы будем подавать его вектор признаков на DNN, которая будет выдавать "распределение вероятностей" по фонемам. Далее это распределение мы будем использовать в нашем алгоритме распознавания по словарю. Схема работы такого классификатора представлена ниже: ![](https://drive.google.com/uc?export=view&id=1dv0VuE8DIoKXETbvnZq2PMixkWMCJOxr)  Bootstrap<jupyter_code>!gdown --id '1meikF0XqBUnVZKoY3160MQ0R25lq-EMu'
!gdown --id '1f0LkW6I3Q5CnOThuNFAJ9umgnb2eU4Ox'

!rm -rf lab3
!unzip -q lab2.zip
!unzip -q train_librispeech_10h.zip
!mv lab2 lab3
!mv train_librispeech_10h lab3/data
!rm -rf lab2.zip sample_data train_librispeech_10h.zip
%cd lab3
import os
import numpy as np
import time
from tqdm.notebook import tqdm
import matplotlib.pyplot as plt
%matplotlib inline
import IPython.display as ipd
import librosa, librosa.display
import torch
import torch.nn as nn
import torch.functional as F
plt.rcParams["figure.figsize"] = (15.0, 5.0)
plt.rcParams["image.interpolation"] = "nearest"
plt.rcParams["image.cmap"] = "gray"
# загрузка списка допустимых фонем:
def load_phones_dict(phones_file):
    phones_dict = {}
    with open(phones_file, encoding="utf-8") as fn:
        for index, line in enumerate(fn):
            phones_dict[line.strip()] = index
    return phones_dict


# загрузка покадровой фонемной разметки в словарь:
def load_phone_aligment(aligment_file):
    ali_dict = {}
    with open(aligment_file, encoding="utf-8") as fn:
        for line in fn:
            line = line.strip().split()
            ali_dict[line[0]] = line[1:]
    return ali_dict


# подсчет акустических признаков для записей из папки:
def load_data_dict(dir_path, count=None):
    data_dict = {}
    step = 0
    for wav_name in tqdm(os.listdir(dir_path)):
        x, sr = librosa.load(os.path.join(dir_path, wav_name), sr=None)
        mfcc = librosa.feature.mfcc(
            x, sr=sr, n_mfcc=13, n_fft=int(sr * 0.025), hop_length=int(sr * 0.01)
        )
        step += 1
        if count and step == count:
            print(f"Early stop at {step}th file..")
            break
        data_dict[wav_name.split(".")[0]] = mfcc.T
    return data_dict


phones_file = "data/train_librispeech_10h/phones"
phones_dict = load_phones_dict(phones_file)

ali_file = "data/train_librispeech_10h/ali_phones.txt"
ali_dict = load_phone_aligment(ali_file)

train_records_dir = "data/train_librispeech_10h/wav_dir"
feats_dict = load_data_dict(train_records_dir, count=1000)<jupyter_output><empty_output><jupyter_text>Априорная вероятность:![](https://drive.google.com/uc?export=view&id=1WhjclhXm2POYsCgvxp3iAIOvT9F_k8If)<jupyter_code># объединение кадров признаков с соответствующими им фонемами:
def combine_data(ali_dict, feats_dict, phones_dict, side_context=None):
    data_list = []
    priors_list = [0] * len(phones_dict)
    for file_name in feats_dict.keys():
        file_feats = feats_dict[file_name]
        file_ali = ali_dict[file_name]
        for frame in range(min(len(file_feats), len(file_ali))):
            data_list.append(np.append(file_feats[frame], phones_dict[file_ali[frame]]))
            priors_list[phones_dict[file_ali[frame]]] += 1
    return data_list, np.array(priors_list) / sum(priors_list)


side_context = None
data_list, priors_list = combine_data(ali_dict, feats_dict, phones_dict, side_context)
data_list = np.random.permutation(data_list)
class CommonDataSet:
    def __init__(self, data_list, device, valid_proportion=0.1):
        self.train = data_list[: int((1 - valid_proportion) * len(data_list))]
        self.valid = data_list[int((1 - valid_proportion) * len(data_list)) :]
        self.device = device

    def num_batches(self, batch_size):
        return len(self.train) // batch_size

    def next_batch(self, batch_size):
        for i in range(self.num_batches(batch_size)):
            batch_xs = torch.tensor(
                self.train[i * batch_size : (i + 1) * batch_size, :-1],
                dtype=torch.float32,
            )
            batch_ys = torch.tensor(
                self.train[i * batch_size : (i + 1) * batch_size, -1].ravel(),
                dtype=torch.int64,
            )

            yield batch_xs.to(device), batch_ys.to(device)
class MyNet(torch.nn.Module):
    def __init__(self, input_dim, output_dim):
        super(MyNet, self).__init__()
        # params:
        input_dim = input_dim  # размерность входных признаков
        n_nodes_hl1 = 256
        n_nodes_hl2 = 256
        n_nodes_hl3 = 256
        n_nodes_hl4 = 256
        output_dim = output_dim  # количество выходных классов (фонем)
        # FC layers:
        self.fc1 = nn.Linear(input_dim, n_nodes_hl1)
        self.fc2 = nn.Linear(n_nodes_hl1, n_nodes_hl2)
        self.fc3 = nn.Linear(n_nodes_hl2, n_nodes_hl3)
        self.fc4 = nn.Linear(n_nodes_hl3, n_nodes_hl4)
        self.fc5 = nn.Linear(n_nodes_hl4, output_dim)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = torch.relu(self.fc3(x))
        x = torch.relu(self.fc4(x))
        x = self.fc5(x)
        return x
def train(model, dataset, batch_size, criterion, optimizer, loss_list, verbose=True):
    if verbose:
        start = time.time()
        print("TRAINING HAS STARTED...")

    for epoch in range(10):
        running_loss = 0
        counter = 0
        for batch_xs, batch_ys in tqdm(
            dataset.next_batch(batch_size),
            total=dataset.num_batches(batch_size),
            desc="Epoch {}".format(epoch),
            disable=not verbose,
        ):
            optimizer.zero_grad()
            output = model(batch_xs)

            loss = criterion(output, batch_ys)
            loss.backward()
            optimizer.step()

            running_loss += loss.item()
            counter += 1

        if verbose:
            print("Loss: {:.4f}".format((running_loss / counter)))
        loss_list.append(running_loss / counter)

    if verbose:
        end = time.time()
        minutes = (end - start) // 60
        seconds = np.mod((end - start), 60)
        print("Training time: {} min {} sec".format(int(minutes), int(seconds)))
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

batch_size = 256

# load data:
dataset = CommonDataSet(data_list, device)

net = MyNet(
    input_dim=(data_list[0].shape[0] - 1), output_dim=len(phones_dict.keys())
).to(device)

criterion = nn.CrossEntropyLoss()

optimizer = torch.optim.Adam(net.parameters(), lr=0.0001)

loss_list = []
# training network model:
train(net, dataset, batch_size, criterion, optimizer, loss_list)
def plot_losses(loss_list):
    plt.plot(loss_list)
    plt.title("Loss per epoch")
    plt.ylabel("Loss")
    plt.xlabel("Epoch")
    plt.grid()
    plt.show()
plot_losses(loss_list)
def evaluate(model, dataset, verbose=True):
    model.cpu()
    with torch.no_grad():
        correct_cnt = 0
        for ftr, true_label in tqdm(
            zip(dataset.valid[:, :-1], dataset.valid[:, -1]),
            total=dataset.valid.shape[0],
            disable=not verbose,
        ):
            ftr = torch.from_numpy(ftr).float()
            output = model(ftr)
            correct_cnt += output.argmax().item() == true_label

    total_frame_accuracy = correct_cnt / dataset.valid.shape[0]
    print("Total frame accuracy: {:.3f}.".format(total_frame_accuracy))
evaluate(net, dataset, verbose=False)
def phonemes_accuracy(model, dataset, phones_dict):
    class_correct = [0 for x in range(len(phones_dict.keys()))]
    class_total = [0 for x in range(len(phones_dict.keys()))]
    model.cpu()
    with torch.no_grad():
        for ftr, true_label in tqdm(
            zip(dataset.valid[:, :-1], dataset.valid[:, -1]),
            total=dataset.valid.shape[0],
        ):
            ftr = torch.from_numpy(ftr).float()
            output = model(ftr)
            predict_label = output.argmax().item()

            class_correct[predict_label] += predict_label == int(true_label)
            class_total[int(true_label)] += 1

    for idx, phone in enumerate(phones_dict.keys()):
        if class_total[idx]:
            print(
                "{:3s}: {:.2f} : {}/{}".format(
                    phone,
                    class_correct[idx] / class_total[idx],
                    class_correct[idx],
                    class_total[idx],
                )
            )
phonemes_accuracy(net, dataset, phones_dict)<jupyter_output><empty_output><jupyter_text>Распознавание по словарю с помощью нейронки<jupyter_code># теперь будем распознавать по словарю, состоящему из конечных слов
# для этого нужно загрузить файл лексикона, который определяет из каких фонем состоят наши слова
def load_lexicon_dict(lexicon_file):
    lexicon_dict = {}
    with open(lexicon_file, encoding="utf-8") as fn:
        for line in fn:
            line = line.strip().split()
            lexicon_dict[line[0]] = line[1:]
    return lexicon_dict
class AcousticModel:
    def __init__(self, phone_name, phone_index, priors_list):
        self.phone_name = phone_name
        self.phone_index = phone_index
        self.phone_prior = priors_list[phone_index]

    def distance(self, input_feats, nnet):
        with torch.no_grad():
            predicts = nnet.forward(torch.tensor(input_feats, dtype=torch.float32))
            softmax = torch.nn.functional.softmax(predicts, dim=0)
        return -np.log(softmax[self.phone_index].item() / self.phone_prior)


class State:
    def __init__(self, model, index):
        self.model = model  # вектор признаков узла
        self.is_final = False  # является ли этот узел финальнвм в слове
        self.word = None  # слово эталона (назначается только для финального узла)
        self.best_token = None  # лучший токен (по минимуму дистанции) в узле
        self.current_word = None  # текущее слово эталона
        self.next_states = []  # список следующих узлов
        self.index = index  # индекс узла


def load_graph(lexicon_dict, phones_dict, priors_list):
    start_state = State(None, 0)
    graph = [
        start_state,
    ]
    state_index = 1
    for word in lexicon_dict.keys():
        previous_state = start_state
        for phone in lexicon_dict[word]:
            state = State(
                AcousticModel(phone, phones_dict[phone], priors_list), state_index
            )
            state.current_word = (
                word  # слово эталона теперь будет храниться в каждом узле
            )
            state.next_states.append(state)  # add loop
            previous_state.next_states.append(state)
            previous_state = state
            graph.append(state)
            state_index += 1
        if state:
            state.word = word
            state.is_final = True
    return graph
class Token:
    def __init__(self, state, dist=0.0, word=""):
        self.state = state  # стейт графа, в котором токен находится в текущий момент
        self.dist = dist  # общая накопленная дистанция пути, пройденного токеном
        self.word = word  # слово, которое распозналось токеном
        self.alive = True


def beam_pruning(next_tokens, threshold):
    alive_tokens = [token for token in next_tokens if token.alive]
    best_token = alive_tokens[
        np.argmin([token.dist for token in alive_tokens if token.alive])
    ]
    for token in next_tokens:
        if token.dist > best_token.dist + threshold:
            token.alive = False
    return next_tokens


def state_pruning(next_tokens):
    for i in range(len(next_tokens)):
        state_index = next_tokens[i].state.index
        if not graph[state_index].best_token:
            graph[state_index].best_token = next_tokens[i]
        else:
            if next_tokens[i].dist <= graph[state_index].best_token.dist:
                graph[state_index].best_token.alive = False
                graph[state_index].best_token = next_tokens[i]
            else:
                next_tokens[i].alive = False
    for token in next_tokens:
        index = token.state.index
        if graph[index].best_token:
            graph[index].best_token = None
    return next_tokens


def recognize(filename, features, graph, nnet, recognition_results, beam_threshold):
    start_state = graph[0]
    active_tokens = [
        Token(start_state),
    ]
    next_tokens = []

    for frame in range(len(features)):
        ftr_frame = features[frame]
        for token in active_tokens:
            if token.alive:
                for transition_state in token.state.next_states:
                    new_token = Token(transition_state, token.dist, token.word)
                    new_token.dist += transition_state.model.distance(ftr_frame, nnet)
                    next_tokens.append(new_token)
        # state and beam prunings:
        next_tokens = state_pruning(next_tokens)
        next_tokens = beam_pruning(next_tokens, beam_threshold)

        active_tokens = next_tokens
        next_tokens = []

    # поиск финальных токенов:
    final_tokens = []
    for token in active_tokens:
        if token.state.is_final and token.alive:
            final_tokens.append(token)

    # если нет финальных, то берем лучший из выживших:
    if len(final_tokens) != 0:
        win_token = final_tokens[np.argmin([token.dist for token in final_tokens])]
    else:
        alive_tokens = [token for token in active_tokens if token.alive]
        win_token = alive_tokens[np.argmin([token.dist for token in alive_tokens])]
        win_token.state.word = win_token.state.current_word

    # вывод результата
    print(f"result: {filename:<27} ==> {win_token.state.word}")

    # совпадает ли запись с полученным эталоном:
    record_word = filename.split("_")[0]
    etalon_word = win_token.state.word.split("_")[0]
    recognition_results.append(etalon_word.lower() == record_word)

    return recognition_results
def run_recognizer(records_data_dict, graph, nnet, beam_threshold=10):
    start_time = time.time()
    recognition_results = []

    for filename in records_data_dict.keys():
        recognition_results = recognize(
            filename,
            records_data_dict[filename],
            graph,
            nnet,
            recognition_results,
            beam_threshold,
        )

    wer = (1 - sum(recognition_results) / len(recognition_results)) * 100
    total_time = time.time() - start_time
    print("-" * 60)
    print(f"WER: {wer:.2f}%")
    print(f"Total time: {total_time:.2f} sec")
    print("-" * 60)
    return wer, total_time<jupyter_output><empty_output><jupyter_text>Запустим распознавание на базе YES_NO<jupyter_code>lexicon_file = "data/lexicon_yes_no.txt"
lexicon_dict = load_lexicon_dict(lexicon_file)
graph = load_graph(lexicon_dict, phones_dict, priors_list)
records_data_dict = load_data_dict("data/yes_no/records")

wer_yn_no_context, time_yn_no_context = run_recognizer(
    records_data_dict, graph, net.cpu(), beam_threshold=100
)<jupyter_output><empty_output><jupyter_text>Запустим распознавание на базе Digits<jupyter_code>lexicon_file = "data/lexicon_digits.txt"
lexicon_dict = load_lexicon_dict(lexicon_file)
graph = load_graph(lexicon_dict, phones_dict, priors_list)
records_data_dict = load_data_dict("data/digits/records")

wer_digits_no_context, time_digits_no_context = run_recognizer(
    records_data_dict, graph, net.cpu(), beam_threshold=50
)<jupyter_output><empty_output><jupyter_text>Задание №1 (5 баллов):Реализовать контексты для входных признаков путем конкатенации кадров слева и справа от текущего. Не забудьте поддержать это и в функции распознавания. Насколько улучшилось качество на датасетах?![](https://drive.google.com/uc?export=view&id=1ft0z6CPhX20cJnA373uLoYwHPEagaKyo)<jupyter_code>def make_windowed(data, side_context=0):
    """
    Expands each array value to the window of size (2 * side_context + 1).
    Uses edge padding.

    Example:
        >>> arr = np.asarray([1, 2, 3]).reshape(-1, 1)
        >>> print(arr)
        [[1]
         [2]
         [3]]
        >>> print(make_windowed(arr, side_context=1))
        [[1 1 2]
         [1 2 3]
         [2 3 3]]
    """
    # The data must be numpy array:
    data = np.pad(data, pad_width=((side_context, side_context), (0, 0)), mode="edge")

    data_list = []
    for frame in range(side_context, len(data) - side_context):
        frames = list(range(frame - side_context, frame + side_context + 1))
        data_list.append(np.concatenate([data[x] for x in frames]))

    return np.asarray(data_list)
print("Original data_list shape:", data_list.shape)
# объединение кадров признаков с соответствующими им фонемами:
def combine_data(ali_dict, feats_dict, phones_dict, side_context=None):
    data_list = []
    priors_list = [0] * len(phones_dict)
    for file_name in feats_dict.keys():
        file_feats = feats_dict[file_name]  # MFCC features for file frames
        file_feats = make_windowed(file_feats, side_context=side_context)
        file_ali = ali_dict[file_name]  # phonemes for file frames

        for frame in range(min(len(file_feats), len(file_ali))):
            data_list.append(np.append(file_feats[frame], phones_dict[file_ali[frame]]))
            priors_list[phones_dict[file_ali[frame]]] += 1

    return data_list, np.array(priors_list) / max(sum(priors_list), 1)


side_context = 2
data_list, priors_list = combine_data(ali_dict, feats_dict, phones_dict, side_context)
data_list = np.random.permutation(data_list)
print("Windowed data_list shape:", data_list.shape)
# load data:
dataset = CommonDataSet(data_list, device)

net = MyNet(
    input_dim=(data_list[0].shape[0] - 1), output_dim=len(phones_dict.keys())
).to(device)

criterion = nn.CrossEntropyLoss()

optimizer = torch.optim.Adam(net.parameters(), lr=0.0001)

loss_list = []
train(net, dataset, batch_size, criterion, optimizer, loss_list)
plot_losses(loss_list)
evaluate(net, dataset)
for side_context in [1, 2, 3, 4, 5]:
    data_list, priors_list = combine_data(
        ali_dict, feats_dict, phones_dict, side_context
    )
    data_list = np.random.permutation(data_list)

    dataset = CommonDataSet(data_list, device)

    net = MyNet(
        input_dim=(data_list[0].shape[0] - 1), output_dim=len(phones_dict.keys())
    ).to(device)

    criterion = nn.CrossEntropyLoss()

    optimizer = torch.optim.Adam(net.parameters(), lr=0.0001)

    loss_list = []

    train(net, dataset, batch_size, criterion, optimizer, loss_list, verbose=False)

    print(f"Side context size: {side_context}.")
    evaluate(net, dataset, verbose=False)
phonemes_accuracy(net, dataset, phones_dict)
lexicon_file = "data/lexicon_yes_no.txt"
lexicon_dict = load_lexicon_dict(lexicon_file)
graph = load_graph(lexicon_dict, phones_dict, priors_list)
records_data_dict = load_data_dict("data/yes_no/records")

records_data_dict = {
    key: make_windowed(value, side_context=side_context)
    for key, value in records_data_dict.items()
}

wer_yn_context, time_yn_context = run_recognizer(
    records_data_dict, graph, net.cpu(), beam_threshold=100
)
lexicon_file = "data/lexicon_digits.txt"
lexicon_dict = load_lexicon_dict(lexicon_file)
graph = load_graph(lexicon_dict, phones_dict, priors_list)
records_data_dict = load_data_dict("data/digits/records")

records_data_dict = {
    key: make_windowed(value, side_context=side_context)
    for key, value in records_data_dict.items()
}

wer_digits_context, time_digits_context = run_recognizer(
    records_data_dict, graph, net.cpu(), beam_threshold=50
)
print(f"WER for YES_NO improved by {wer_yn_no_context - wer_yn_context:.2f}%.")
print(
    f"Time for YES_NO increased by "
    f"{time_yn_context / time_yn_no_context:.1f} times."
)

print(f"WER for DIGITS improved by {wer_digits_no_context - wer_digits_context:.2f}%.")
print(
    f"Time for DIGITS increased by "
    f"{time_digits_context / time_digits_no_context:.1f} times."
)<jupyter_output>WER for YES_NO improved by 0.00%.
Time for YES_NO increased by 0.9 times.
WER for DIGITS improved by 10.00%.
Time for DIGITS increased by 0.9 times.<jupyter_text>Также можно отметить, что по мере увеличения размера окна контекста от 1 до 5 росла и точность классификации фонем.  Задание №2 (5 баллов):Повысить качество классификации фонем за счет улучшения архитектуры нейронной сети, оптимизатора, других речевых признаков и т.п. Составить небольшой отчет на основе результатов перебора использованных способов.<jupyter_code>class ListenerNet(torch.nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim):
        super(ListenerNet, self).__init__()
        # Layer sizes:
        n_nodes_hl1 = hidden_dim
        n_nodes_hl2 = hidden_dim
        n_nodes_hl3 = hidden_dim
        n_nodes_hl4 = hidden_dim
        # FC layers with ELU activations:
        self.fc1 = nn.Linear(input_dim, n_nodes_hl1)
        self.a1 = nn.ELU()
        self.fc2 = nn.Linear(n_nodes_hl1, n_nodes_hl2)
        self.a2 = nn.ELU()
        self.fc3 = nn.Linear(n_nodes_hl2, n_nodes_hl3)
        self.a3 = nn.ELU()
        self.fc4 = nn.Linear(n_nodes_hl3, n_nodes_hl4)
        self.a4 = nn.ELU()
        self.fc5 = nn.Linear(n_nodes_hl4, output_dim)

    def forward(self, x):
        x = self.fc1(x)
        x = self.a1(x)

        x = self.fc2(x)
        x = self.a2(x)

        x = self.fc3(x)
        x = self.a3(x)

        x = self.fc4(x)
        x = self.a4(x)

        x = self.fc5(x)
        return x
data_list, priors_list = combine_data(ali_dict, feats_dict, phones_dict, side_context=5)
data_list = np.random.permutation(data_list)

dataset = CommonDataSet(data_list, device)

criterion = nn.CrossEntropyLoss()
net = ListenerNet(
    input_dim=(data_list[0].shape[0] - 1),
    hidden_dim=512,
    output_dim=len(phones_dict.keys()),
).to(device)

optimizer = torch.optim.Adam(net.parameters(), lr=1e-4, weight_decay=5e-4)

loss_list = []

train(net, dataset, batch_size, criterion, optimizer, loss_list, verbose=False)
plot_losses(loss_list)
evaluate(net, dataset)
net = ListenerNet(
    input_dim=(data_list[0].shape[0] - 1),
    hidden_dim=1024,
    output_dim=len(phones_dict.keys()),
).to(device)

optimizer = torch.optim.Adam(net.parameters(), lr=1e-4, weight_decay=5e-4)

loss_list = []

train(net, dataset, batch_size, criterion, optimizer, loss_list, verbose=False)
plot_losses(loss_list)
evaluate(net, dataset)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10852678571428571, 'fraction_numerical': 0.014419642857142857, 'mean_word_length': 3.194943820224719, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 39, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '179640', 'license': 'MIT', 'n_tokens_mistral': 8071, 'n_tokens_neox': 8048, 'n_words': 1600, 'path': '08-dnn-classifier/dnn_classifier.ipynb', 'repo_name': 'Illumaria/made-speech-signal-processing', 'size': '22400'}