Во время телефонных или видеозвонков качество связи может резко ухудшаться: звук пропадает, речь прерывается, собеседники перестают друг друга слышать. Один из способов…

Краткий ответ

Нужно обучить бинарный классификатор:

\[ y_i= \begin{cases} 1,& \text{id}_i\in word\_bounds.json,\\ 0,& \text{иначе}. \end{cases} \]

Для положительных файлов можно использовать отмеченный фрагмент аудио, а при проверке тестового файла — просматривать его скользящими окнами. Если хотя бы одно окно похоже на фразу «не слышу» или «не слышно», всему файлу присваивается метка 1.

По одному условию нельзя получить конкретные строки submition.csv: для этого нужны сами архивы. Ниже — рабочая схема базового решения.

1. Как понимать метрику

\[ 1-FRR=\frac{TP}{TP+FN} \]

— полнота положительного класса, а

\[ 1-FAR=\frac{TN}{TN+FP} \]

— доля правильно распознанных отрицательных файлов.

Поэтому итоговая метрика равна:

\[ score= \frac{2\cdot TPR\cdot TNR}{TPR+TNR}, \]

где:

• \(TPR=1-FRR\);
• \(TNR=1-FAR\).

Важно: проверять метрику нужно на уровне аудиофайлов, а не отдельных окон.

2. Рекомендуемый алгоритм

1. Распаковать архивы.
2. Для каждого обучающего файла определить метку по наличию ID в word_bounds.json.
3. Перевести звук в моно и частоту дискретизации 16 кГц.
4. Для положительных файлов вырезать участок [start, end] с небольшим запасом.
5. Для отрицательных файлов взять несколько окон такой же примерной длины.
6. Из аудио получить log-mel-признаки и их динамику.
7. Обучить классификатор.
8. Для тестового файла пройтись окнами по всей записи.
9. Взять максимальную вероятность среди окон.
10. Порог выбрать на валидации по максимальному score, а не слепо использовать 0.5.

Разделять обучающую и проверочную выборки нужно до создания окон, иначе разные окна одного файла попадут и в обучение, и в проверку.

3. Пример базовой реализации

После распаковки должны существовать папки train_data и test_data.

import json
from pathlib import Path

import librosa
import numpy as np
import pandas as pd

from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import confusion_matrix


SR = 16_000


def find_audio(root):
    return {
        p.stem: p
        for p in Path(root).rglob("*")
        if p.suffix.lower() == ".opus"
    }


def load_audio(path):
    # Если librosa не читает Opus, сначала декодируйте файлы через ffmpeg.
    y, _ = librosa.load(str(path), sr=SR, mono=True)
    return y.astype(np.float32)


def features(y):
    """Статистики log-mel и delta-признаков."""
    min_length = int(0.8 * SR)

    if len(y) < min_length:
        y = np.pad(y, (0, min_length - len(y)))

    mel = librosa.feature.melspectrogram(
        y=y,
        sr=SR,
        n_fft=512,
        win_length=400,
        hop_length=160,
        n_mels=64,
        fmin=50,
        fmax=7600,
        power=2.0,
    )

    mel_db = librosa.power_to_db(mel, ref=np.max)
    delta = librosa.feature.delta(mel_db)

    result = np.concatenate([
        mel_db.mean(axis=1),
        mel_db.std(axis=1),
        delta.mean(axis=1),
        delta.std(axis=1),
    ])

    return result.astype(np.float32)


def crop_with_context(y, start, end, context=0.2):
    left = max(0, int((start - context) * SR))
    right = min(len(y), int((end + context) * SR))
    return y[left:right]


def fixed_windows(y, length_sec, count=3):
    """Несколько окон для отрицательного примера."""
    window = max(1, int(length_sec * SR))

    if len(y) <= window:
        return [y]

    last_start = len(y) - window
    starts = np.linspace(
        0,
        last_start,
        num=min(count, last_start + 1),
        dtype=int,
    )

    return [y[start:start + window] for start in starts]


def scan_windows(y, length_sec, hop_sec=0.25):
    """Перекрывающиеся окна для проверки одного файла."""
    window = max(1, int(length_sec * SR))
    hop = max(1, int(hop_sec * SR))

    if len(y) <= window:
        return [y]

    last_start = len(y) - window
    starts = list(range(0, last_start + 1, hop))

    if starts[-1] != last_start:
        starts.append(last_start)

    return [y[start:start + window] for start in starts]


train_files = find_audio("train_data")
test_files = find_audio("test_data")

bounds_path = next(Path("train_data").rglob("word_bounds.json"))

with open(bounds_path, encoding="utf-8") as file:
    bounds = json.load(file)

bounds = {
    str(file_id): (float(value[0]), float(value[1]))
    for file_id, value in bounds.items()
}

train_ids = sorted(train_files)

# Примерная длина окна
durations = np.array([
    end - start
    for start, end in bounds.values()
])

window_sec = float(np.clip(np.median(durations) + 0.4, 1.0, 3.0))


def build_dataset(ids):
    X = []
    y = []

    for file_id in ids:
        audio = load_audio(train_files[file_id])

        if file_id in bounds:
            start, end = bounds[file_id]
            segments = [
                crop_with_context(audio, start, end)
            ]
            label = 1
        else:
            segments = fixed_windows(
                audio,
                length_sec=window_sec,
                count=3,
            )
            label = 0

        for segment in segments:
            X.append(features(segment))
            y.append(label)

    return np.asarray(X), np.asarray(y)


# Разделяем именно файлы, а не отдельные аудиоокна
file_labels = np.array([
    int(file_id in bounds)
    for file_id in train_ids
])

learn_ids, valid_ids = train_test_split(
    train_ids,
    test_size=0.2,
    random_state=42,
    stratify=file_labels,
)

X_learn, y_learn = build_dataset(learn_ids)

model = make_pipeline(
    StandardScaler(),
    LogisticRegression(
        max_iter=2000,
        class_weight="balanced",
    ),
)

model.fit(X_learn, y_learn)


def file_probability(path):
    audio = load_audio(path)

    segments = scan_windows(
        audio,
        length_sec=window_sec,
        hop_sec=0.25,
    )

    X = np.vstack([
        features(segment)
        for segment in segments
    ])

    probabilities = model.predict_proba(X)[:, 1]

    # Фраза может находиться в любом месте записи
    return float(probabilities.max())


# Вероятности для валидационных файлов
valid_probabilities = np.array([
    file_probability(train_files[file_id])
    for file_id in valid_ids
])

valid_labels = np.array([
    int(file_id in bounds)
    for file_id in valid_ids
])


def competition_score(y_true, probabilities, threshold):
    predictions = (probabilities >= threshold).astype(int)

    tn, fp, fn, tp = confusion_matrix(
        y_true,
        predictions,
        labels=[0, 1],
    ).ravel()

    tpr = tp / (tp + fn) if tp + fn else 0.0
    tnr = tn / (tn + fp) if tn + fp else 0.0

    if tpr + tnr == 0:
        score = 0.0
    else:
        score = 2 * tpr * tnr / (tpr + tnr)

    return score, tpr, tnr


# Подбираем порог по метрике соревнования
thresholds = np.unique(np.concatenate([
    np.linspace(0, 1, 1001),
    valid_probabilities,
]))

best = max(
    (
        (
            threshold,
            competition_score(
                valid_labels,
                valid_probabilities,
                threshold,
            ),
        )
        for threshold in thresholds
    ),
    key=lambda item: item[1][0],
)

best_threshold = best[0]
best_score, best_tpr, best_tnr = best[1]

print(f"Validation score: {best_score:.4f}")
print(f"TPR = {best_tpr:.4f}")
print(f"TNR = {best_tnr:.4f}")
print(f"Threshold = {best_threshold:.4f}")


# После выбора порога обучаем модель на всех обучающих данных
X_all, y_all = build_dataset(train_ids)
model.fit(X_all, y_all)


# Предсказания для теста
test_ids = sorted(test_files)

test_probabilities = np.array([
    file_probability(test_files[file_id])
    for file_id in test_ids
])

test_labels = (
    test_probabilities >= best_threshold
).astype(int)

submission = pd.DataFrame({
    "id": test_ids,
    "label": test_labels,
})

submission.to_csv("submition.csv", index=False)

assert len(submission) == len(test_files)
assert submission["id"].is_unique
assert set(submission["label"]).issubset({0, 1})

print("Файл submition.csv создан.")

4. Почему используется максимум по окнам

Целевая метка положительна, если в записи есть хотя бы одна нужная фраза. Поэтому при вероятностях окон \(p_1,p_2,\ldots,p_n\) естественно использовать:

\[ p_{\text{file}}=\max_i p_i. \]

Если использовать среднее значение, короткая фраза может потеряться среди длинной записи.

5. Как улучшить качество

Для более сильного решения можно:

• обучить CNN или CRNN непосредственно на log-mel-спектрограммах;
• добавить шум, изменение громкости и скорости воспроизведения;
• после первой модели собрать отрицательные окна с наибольшими вероятностями и добавить их как «трудные отрицательные примеры»;
• использовать несколько разбиений и получить out-of-fold-предсказания для более надёжного подбора порога;
• объединить аудиоклассификатор с русскоязычным ASR и искать нормализованные варианты фраз;
• попробовать не только max, но и среднее трёх самых вероятных окон.

Главная проверка результата: на валидации отдельно посчитать FRR, FAR, TPR, TNR и убедиться, что итоговый файл содержит ровно столько строк, сколько тестовых аудиофайлов.